runner.go
1// Package runner is the CI host's daemon (DESIGN.md ยง10): spindle-compatible
2// on the outside, a host for the engine on the inside.
3//
4// It follows the bridge's sh.tangled.git.refUpdate stream, reads the pushed
5// commit's .tangled/workflows from a mirror, matches `when:`, records a
6// sh.tangled.pipeline, prepares a run directory per matching workflow, and
7// starts the engine (`ci runner-exec`) in a sandboxed unit per segment. It
8// tails each run's events.ndjson to publish sh.tangled.pipeline.status,
9// run.rtw.ci.gate and run.rtw.ci.transition records on its own /events, and
10// serves logs on /logs in upstream's format. Gates are answered through its
11// API (the appview), which starts the next segment.
12package runner
13
14import (
15 "context"
16 "database/sql"
17 "encoding/json"
18 "errors"
19 "fmt"
20 "log/slog"
21 "os"
22 "path"
23 "path/filepath"
24 "slices"
25 "strings"
26 "sync"
27 "time"
28
29 "git.kilimanjaro.io/ci/internal/checkout"
30 "git.kilimanjaro.io/ci/internal/engine"
31 "git.kilimanjaro.io/ci/internal/eventlog"
32 "git.kilimanjaro.io/ci/internal/ident"
33 "git.kilimanjaro.io/ci/internal/lex"
34 "git.kilimanjaro.io/ci/internal/manifest"
35 "git.kilimanjaro.io/ci/internal/tid"
36)
37
38// Config configures a Runner.
39type Config struct {
40 RunsDir string // run directories
41 CacheDir string // mirrors and per-repo HOMEs
42 LocksDir string // concurrency locks
43 // Knot is the forge host: pipelines are at://did:web:<Knot>/sh.tangled.pipeline/<rkey>.
44 Knot string
45 // OwnerDid owns the repos (sh.tangled.pipeline#triggerRepo.did).
46 OwnerDid string
47 // CloneURL is a template with {repo}, e.g. [email protected]:{repo}.
48 CloneURL string
49 // MaxConcurrent limits segments running at once (default 2).
50 MaxConcurrent int
51 // SlimAfter and DeleteAfter are the retention of finished runs (see
52 // prune.go): after SlimAfter a run keeps only its record, after DeleteAfter
53 // it is deleted. Zero means DefaultSlimAfter / DefaultDeleteAfter; negative
54 // turns that stage off.
55 SlimAfter, DeleteAfter time.Duration
56 // Notify is called for gates and finished runs (ntfy), may be nil.
57 Notify func(Notification)
58}
59
60// Runner is the daemon.
61type Runner struct {
62 Cfg Config
63 Log *eventlog.Log
64 Launcher Launcher
65 Logger *slog.Logger
66
67 src *checkout.Source
68 slots chan struct{}
69 mu sync.Mutex
70 // tailing marks runs whose events are being followed.
71 tailing map[string]bool
72 leases *leases
73}
74
75// New prepares a runner and its tables.
76func New(cfg Config, log *eventlog.Log, l Launcher, logger *slog.Logger) (*Runner, error) {
77 if cfg.MaxConcurrent <= 0 {
78 cfg.MaxConcurrent = 2
79 }
80 if cfg.LocksDir == "" {
81 cfg.LocksDir = filepath.Join(cfg.RunsDir, ".locks")
82 }
83 if logger == nil {
84 logger = slog.Default()
85 }
86 for _, d := range []string{cfg.RunsDir, cfg.CacheDir, cfg.LocksDir} {
87 if err := os.MkdirAll(d, 0o755); err != nil {
88 return nil, err
89 }
90 }
91 r := &Runner{
92 Cfg: cfg, Log: log, Launcher: l, Logger: logger,
93 src: &checkout.Source{MirrorsDir: filepath.Join(cfg.CacheDir, "mirrors"), URL: checkout.URLFunc(cfg.CloneURL, "")},
94 slots: make(chan struct{}, cfg.MaxConcurrent),
95 tailing: map[string]bool{},
96 }
97 if _, err := log.DB().Exec(`
98 create table if not exists cursors (source text primary key, cursor integer not null);
99 create table if not exists runs (
100 id text primary key,
101 pipeline text not null,
102 rkey text not null,
103 workflow text not null,
104 repo text not null,
105 dir text not null,
106 parent text not null default '',
107 status text not null,
108 offset integer not null default 0,
109 segment integer not null default 0,
110 handle text not null default '',
111 created text not null,
112 updated text not null
113 );
114 create index if not exists runs_rkey on runs (rkey, workflow);`); err != nil {
115 return nil, err
116 }
117 return r, nil
118}
119
120// Run is a row of the runs table.
121type Run struct {
122 ID string `json:"id"`
123 Pipeline string `json:"pipeline"`
124 Rkey string `json:"rkey"`
125 Workflow string `json:"workflow"`
126 Repo string `json:"repo"`
127 Dir string `json:"dir"`
128 Parent string `json:"parent,omitempty"`
129 Status string `json:"status"`
130 Created string `json:"created"`
131 Updated string `json:"updated"`
132 offset int64
133 segment int
134 handle string
135}
136
137func (r *Runner) db() *sql.DB { return r.Log.DB() }
138
139func (r *Runner) cursor(source string) int64 {
140 var c int64
141 _ = r.db().QueryRow(`select cursor from cursors where source = ?`, source).Scan(&c)
142 return c
143}
144
145func (r *Runner) setCursor(source string, c int64) error {
146 _, err := r.db().Exec(`insert into cursors (source, cursor) values (?, ?) on conflict(source) do update set cursor = excluded.cursor`, source, c)
147 return err
148}
149
150const runCols = `id, pipeline, rkey, workflow, repo, dir, parent, status, offset, segment, handle, created, updated`
151
152func scanRun(sc interface{ Scan(...any) error }) (Run, error) {
153 var x Run
154 err := sc.Scan(&x.ID, &x.Pipeline, &x.Rkey, &x.Workflow, &x.Repo, &x.Dir, &x.Parent, &x.Status, &x.offset, &x.segment, &x.handle, &x.Created, &x.Updated)
155 return x, err
156}
157
158// GetRun returns one run.
159func (r *Runner) GetRun(id string) (Run, error) {
160 return scanRun(r.db().QueryRow(`select `+runCols+` from runs where id = ?`, id))
161}
162
163// RunByWorkflow finds a pipeline's run of a workflow (for /logs).
164func (r *Runner) RunByWorkflow(rkey, workflow string) (Run, error) {
165 return scanRun(r.db().QueryRow(`select `+runCols+` from runs where rkey = ? and workflow = ? and parent = '' order by created limit 1`, rkey, workflow))
166}
167
168// ListRuns returns the most recent runs, newest first, optionally of a repo.
169func (r *Runner) ListRuns(repo string, limit int) ([]Run, error) {
170 q := `select ` + runCols + ` from runs`
171 args := []any{}
172 if repo != "" {
173 q += ` where repo = ?`
174 args = append(args, repo)
175 }
176 q += ` order by created desc limit ?`
177 args = append(args, limit)
178 rows, err := r.db().Query(q, args...)
179 if err != nil {
180 return nil, err
181 }
182 defer rows.Close()
183 var out []Run
184 for rows.Next() {
185 x, err := scanRun(rows)
186 if err != nil {
187 return nil, err
188 }
189 out = append(out, x)
190 }
191 return out, rows.Err()
192}
193
194func (r *Runner) insertRun(x Run) error {
195 _, err := r.db().Exec(`insert into runs (`+runCols+`) values (?, ?, ?, ?, ?, ?, ?, ?, 0, 0, '', ?, ?)`,
196 x.ID, x.Pipeline, x.Rkey, x.Workflow, x.Repo, x.Dir, x.Parent, x.Status, x.Created, x.Updated)
197 return err
198}
199
200func (r *Runner) updateRun(id string, set string, args ...any) error {
201 args = append(args, now(), id)
202 _, err := r.db().Exec(`update runs set `+set+`, updated = ? where id = ?`, args...)
203 return err
204}
205
206func now() string { return time.Now().UTC().Format(time.RFC3339Nano) }
207
208// pipelineURI is a pipeline's AT-URI, as the spindle forms it.
209func (r *Runner) pipelineURI(rkey string) string {
210 return fmt.Sprintf("at://%s/%s/%s", ident.ServiceDID(r.Cfg.Knot), lex.PipelineNSID, rkey)
211}
212
213// Follow consumes the bridge's stream until ctx ends.
214func (r *Runner) Follow(ctx context.Context, url, token string) error {
215 source := "bridge:" + url
216 return eventlog.Subscribe(ctx, url, r.cursor(source), token, func(ev lex.Event) error {
217 if ev.Nsid == lex.GitRefUpdateNSID {
218 var u lex.GitRefUpdate
219 if err := json.Unmarshal(ev.Event, &u); err != nil {
220 r.Logger.Error("bad ref update", "rkey", ev.Rkey, "err", err)
221 } else if err := r.OnRefUpdate(ctx, u); err != nil {
222 // a push we can't process is logged, not retried forever
223 r.Logger.Error("ref update", "repo", u.Repo, "ref", u.Ref, "err", err)
224 }
225 }
226 return r.setCursor(source, ev.Created)
227 })
228}
229
230// Trigger says what started a pipeline.
231type Trigger struct {
232 Kind string // push | manual
233 RepoDid string
234 Ref string
235 OldSha string
236 Sha string
237 ChangedFiles []string
238 Inputs []*lex.PipelinePair
239 Workflows []string // manual: only these (default all)
240}
241
242// OnRefUpdate starts the workflows a push matches.
243func (r *Runner) OnRefUpdate(ctx context.Context, u lex.GitRefUpdate) error {
244 if u.NewSha == "" || strings.Trim(u.NewSha, "0") == "" {
245 return nil // a deleted ref
246 }
247 for _, o := range u.PushOptions {
248 if o == "skip-ci" || o == "ci-skip" {
249 r.Logger.Info("push asked to skip ci", "repo", u.Repo, "ref", u.Ref)
250 return nil
251 }
252 }
253 _, err := r.StartPipeline(ctx, Trigger{Kind: "push", RepoDid: u.Repo, Ref: u.Ref, OldSha: u.OldSha, Sha: u.NewSha, ChangedFiles: u.ChangedFiles})
254 return err
255}
256
257// StartPipeline reads the workflows at the trigger's commit, records a
258// pipeline for those that match, and starts a run of each. It returns the
259// pipeline's AT-URI, or "" if no workflow matched.
260func (r *Runner) StartPipeline(ctx context.Context, t Trigger) (string, error) {
261 host, repo, err := ident.ParseRepoDID(t.RepoDid)
262 if err != nil {
263 return "", err
264 }
265 if host != r.Cfg.Knot {
266 return "", fmt.Errorf("repo %s is not on this runner's forge %s", t.RepoDid, r.Cfg.Knot)
267 }
268 mirror, err := r.src.Mirror(ctx, repo)
269 if err != nil {
270 return "", err
271 }
272 if t.Sha, err = checkout.Resolve(ctx, mirror, t.Sha); err != nil {
273 return "", err
274 }
275 defaultBranch := "main"
276 if out, err := gitOut(ctx, mirror, "symbolic-ref", "--short", "HEAD"); err == nil {
277 defaultBranch = strings.TrimSpace(out)
278 }
279 files, err := workflowFiles(ctx, mirror, t.Sha)
280 if err != nil {
281 return "", err
282 }
283 var matched []*manifest.Manifest
284 var invalid []invalidWorkflow
285 for _, f := range files {
286 raw, err := gitOut(ctx, mirror, "show", t.Sha+":"+f)
287 if err != nil {
288 return "", err
289 }
290 m, err := manifest.Parse(manifest.NameOf(f), []byte(raw))
291 if errors.Is(err, manifest.ErrOtherEngine) {
292 continue
293 }
294 if err != nil {
295 // an invalid workflow doesn't stop the others; it is logged and
296 // listed as failed in the pipeline so its author sees why
297 r.Logger.Warn("invalid workflow", "repo", repo, "file", f, "err", err)
298 invalid = append(invalid, invalidWorkflow{manifest.NameOf(f), raw, err})
299 continue
300 }
301 if len(t.Workflows) > 0 && !slices.Contains(t.Workflows, m.Name) {
302 continue
303 }
304 if t.Kind == "manual" || m.Matches(manifest.Event{Kind: t.Kind, Ref: t.Ref, ChangedFiles: t.ChangedFiles}) {
305 matched = append(matched, m)
306 }
307 }
308 if len(matched) == 0 && len(invalid) == 0 {
309 r.Logger.Info("no workflow matched", "repo", repo, "ref", t.Ref, "sha", t.Sha)
310 return "", nil
311 }
312
313 rkey := tid.New()
314 uri := r.pipelineURI(rkey)
315 repoName := repo
316 tm := &lex.PipelineTriggerMetadata{
317 Kind: t.Kind,
318 Repo: &lex.PipelineTriggerRepo{Knot: r.Cfg.Knot, Did: r.Cfg.OwnerDid, RepoDid: &t.RepoDid, Repo: &repoName, DefaultBranch: defaultBranch},
319 }
320 switch t.Kind {
321 case "push":
322 tm.Push = &lex.PipelinePushTriggerData{Ref: t.Ref, NewSha: t.Sha, OldSha: t.OldSha}
323 default:
324 ref := t.Ref
325 tm.Manual = &lex.PipelineManualTriggerData{Sha: t.Sha, Inputs: t.Inputs}
326 if ref != "" {
327 tm.Manual.Ref = &ref
328 }
329 }
330 p := lex.Pipeline{LexiconTypeID: lex.PipelineNSID, TriggerMetadata: tm}
331 for _, m := range matched {
332 p.Workflows = append(p.Workflows, &lex.PipelineWorkflow{
333 Name: m.Name, Engine: lex.EngineDevenv, Clone: &lex.PipelineCloneOpts{Depth: 0}, Raw: string(m.Raw),
334 })
335 }
336 for _, w := range invalid {
337 p.Workflows = append(p.Workflows, &lex.PipelineWorkflow{
338 Name: w.name, Engine: lex.EngineDevenv, Clone: &lex.PipelineCloneOpts{}, Raw: w.raw,
339 })
340 }
341 if _, err := r.Log.Append(lex.PipelineNSID, rkey, p); err != nil {
342 return "", err
343 }
344 for _, w := range invalid {
345 r.status(uri, w.name, lex.StatusFailed, "invalid workflow: "+w.err.Error())
346 }
347 r.Logger.Info("pipeline", "uri", uri, "repo", repo, "sha", t.Sha, "workflows", len(matched))
348 for _, m := range matched {
349 meta := engine.Meta{
350 Run: rkey + "-" + m.Name, Workflow: m.Name, Repo: repo, RepoDid: t.RepoDid, Knot: r.Cfg.Knot,
351 Sha: t.Sha, Ref: t.Ref, DefaultBranch: defaultBranch, Kind: t.Kind, Pipeline: uri, Host: "runner",
352 }
353 if err := r.prepare(ctx, m, meta, mirror, rkey); err != nil {
354 r.Logger.Error("prepare run", "run", meta.Run, "err", err)
355 r.status(uri, m.Name, lex.StatusFailed, "could not prepare the run: "+err.Error())
356 }
357 }
358 return uri, nil
359}
360
361// prepare lays out a run directory and creates the run, then starts it.
362func (r *Runner) prepare(ctx context.Context, m *manifest.Manifest, meta engine.Meta, mirror, rkey string) error {
363 dir := engine.Dir(filepath.Join(r.Cfg.RunsDir, meta.Run))
364 meta.Root = filepath.Join(dir.Path(), meta.Repo)
365 if err := dir.Create(); err != nil {
366 return err
367 }
368 origin, _ := r.src.URL(meta.Repo)
369 checked, err := checkout.Layout(ctx, r.src, dir.Path(), checkout.Main{Repo: meta.Repo, From: mirror, Sha: meta.Sha, Origin: origin}, m.Repos, nil)
370 if err != nil {
371 return err
372 }
373 for _, c := range checked {
374 meta.Checkouts = append(meta.Checkouts, engine.Checkout{Repo: c.Repo, Path: c.Path, Rev: c.Rev, Env: c.Env})
375 }
376 e, err := engine.Create(m, dir, meta, nil, engine.Options{})
377 if err != nil {
378 return err
379 }
380 e.Log.Close()
381 t := now()
382 if err := r.insertRun(Run{ID: meta.Run, Pipeline: meta.Pipeline, Rkey: rkey, Workflow: m.Name, Repo: meta.Repo,
383 Dir: dir.Path(), Status: lex.StatusPending, Created: t, Updated: t}); err != nil {
384 return err
385 }
386 r.status(meta.Pipeline, m.Name, lex.StatusPending, "")
387 return r.launch(ctx, meta.Run, Segment{})
388}
389
390// status appends a sh.tangled.pipeline.status record.
391func (r *Runner) status(pipeline, workflow, status, errMsg string) {
392 s := lex.PipelineStatus{LexiconTypeID: lex.PipelineStatusNSID, Pipeline: pipeline, Workflow: workflow, Status: status, CreatedAt: now()}
393 if errMsg != "" {
394 s.Error = &errMsg
395 }
396 if _, err := r.Log.Append(lex.PipelineStatusNSID, "", s); err != nil {
397 r.Logger.Error("append status", "err", err)
398 }
399}
400
401type invalidWorkflow struct {
402 name, raw string
403 err error
404}
405
406func workflowFiles(ctx context.Context, gitDir, sha string) ([]string, error) {
407 out, err := gitOut(ctx, gitDir, "ls-tree", "--name-only", sha+":"+manifest.Dir)
408 if err != nil {
409 return nil, nil // no .tangled/workflows at that commit
410 }
411 var files []string
412 for _, f := range strings.Fields(out) {
413 if ext := path.Ext(f); ext == ".yml" || ext == ".yaml" {
414 files = append(files, manifest.Dir+"/"+f)
415 }
416 }
417 slices.Sort(files)
418 return files, nil
419}
