11 files changed,
+2019,
-9
+34,
-8
1@@ -439,18 +439,44 @@ when reachable, so a laptop deploy and a CI deploy can't overlap.
2 per repo), then starts a segment as a transient unit:
3
4 ```
5-systemd-run --unit=ci-<run>-<seg> --slice=ci.slice -p User=ci \
6- -p ProtectSystem=strict -p ProtectHome=yes -p PrivateTmp=yes \
7- -p ReadWritePaths=<run> <cache> -p LoadCredential=age-identity:<path> \
8- pasta --config-net -- devenv shell -- ci run|resume …
9+systemd-run --unit=ci-<run>-<seg> --slice=ci.slice --uid=ci --gid=ci \
10+ -p ProtectSystem=strict -p ProtectHome=yes -p PrivateTmp=yes -p NoNewPrivileges=yes \
11+ -p "ReadWritePaths=<run> <cache>/home/<repo> <locks> /nix/var/nix/daemon-socket" \
12+ -p LoadCredential=age-identity:<path> -p StandardOutput=append:<run>/segment-<n>.log \
13+ --setenv=HOME=<cache>/home/<repo> \
14+ -- pasta --config-net -t none -u none -T none -U none -- ci runner-exec <run>
15 ```
16
17-- Tails `<run>/events.ndjson` and serves it as `/logs`, plus status, gate and
18- transition records on `/events`. Translates run state into
19- `sh.tangled.pipeline.status`.
20-- On restart, re-attaches to running units by name and resumes tailing.
21+- Run ids are `<pipeline-rkey>-<workflow>`; manual action runs are
22+ `<run>-<action>-<tid>` (child runs, DESIGN §9). The daemon prepares the run
23+ directory and `state.json` (`engine.Create`, Host `runner`, Pipeline
24+ AT-URI), then each segment runs `ci runner-exec <run-dir>` (fresh,
25+ `--from <action>`, or `--choose N | --dismiss | --expire`), always
26+ pausing at gates. A `runs` table (in `runner.db`, next to the event log)
27+ tracks status, segment, unit and the events offset already published.
28+- Tails `<run>/events.ndjson`: gates and transitions become
29+ `run.rtw.ci.gate` / `run.rtw.ci.transition` records on `/events`; status
30+ becomes `sh.tangled.pipeline.status` (`pending` when prepared, `running`
31+ when the first segment starts, then `success` / `failed` / `cancelled`, or
32+ `timeout` for an expired gate). Waiting at a gate stays `running`.
33+ Manual action runs publish no Tangled status for the parent's workflow.
34+- `/logs/{knot}/{rkey}/{workflow}` streams upstream `LogLine`s from the
35+ events (control start/end per action execution, `step_id` = execution
36+ number, data per line) and closes when the run ends. `/api/...` serves runs,
37+ `log.txt` (plain text for agents), attachments, gate answers, manual
38+ actions and cancel, all behind one bearer token held by the appview;
39+ `/xrpc/sh.tangled.ci.triggerPipeline` starts manual pipelines.
40+- Segments are queued behind `--max-concurrent` slots; a segment that can't
41+ start ends its run as failed. Gates past `expiresAt` are expired on a 30 s
42+ timer. Notifications (ntfy) go out for gates and failed runs (optionally
43+ successes), linking to the appview's run page.
44+- On restart, `Reconcile` follows runs whose unit is still active, settles
45+ finished ones, starts never-started ones, and marks runs whose process is
46+ gone without a final state as failed ("ended without finishing").
47 Transient units survive daemon restarts, including `switch-to-configuration`
48 during a self-deploy.
49+- Invalid workflow files are listed in the pipeline with a `failed` status
50+ and the validation error, so their author sees why.
51
52 Isolation (protects the host's config and services, not against attack)
53
+118,
-0
1@@ -0,0 +1,118 @@
2+package main
3+
4+import (
5+ "context"
6+ "flag"
7+ "fmt"
8+ "log/slog"
9+ "os"
10+ "path/filepath"
11+ "strings"
12+
13+ "git.kilimanjaro.io/ci/internal/eventlog"
14+ "git.kilimanjaro.io/ci/internal/runner"
15+)
16+
17+func init() {
18+ commands["runner"] = command{"runner daemon: spindle-compatible host for the engine (serve)", cmdRunner}
19+}
20+
21+type listFlag []string
22+
23+func (l *listFlag) String() string { return strings.Join(*l, ",") }
24+func (l *listFlag) Set(v string) error { *l = append(*l, v); return nil }
25+
26+func cmdRunner(args []string) int {
27+ if len(args) == 0 || args[0] != "serve" {
28+ fmt.Fprintln(os.Stderr, "usage: ci runner serve [flags]")
29+ return exitUsage
30+ }
31+ fs := flag.NewFlagSet("ci runner serve", flag.ContinueOnError)
32+ listen := fs.String("listen", "127.0.0.1:8476", "address to serve on")
33+ state := fs.String("state-dir", "/var/lib/ci", "runs/, cache/, locks/ and runner.db live here")
34+ bridgeURL := fs.String("bridge", "", "the bridge's /events URL, e.g. ws://127.0.0.1:8475/events (required)")
35+ bridgeToken := fs.String("bridge-token-file", "", "bearer token for the bridge")
36+ tokenFile := fs.String("token-file", "", "bearer token required by this runner's API (the appview holds it)")
37+ knot := fs.String("knot", "", "the forge host (repos are did:web:<knot>:<repo>) (required)")
38+ owner := fs.String("owner-did", "", "DID of the repos' owner")
39+ cloneURL := fs.String("clone-url", "", "clone URL template with {repo}, e.g. [email protected]:{repo} (required)")
40+ maxConc := fs.Int("max-concurrent", 2, "segments running at once")
41+ launcher := fs.String("launcher", "systemd", "systemd (sandboxed transient units) or local (child processes; development)")
42+ shell := fs.String("shell", "devenv", "devenv, or bash (no devenv; tests)")
43+ slice := fs.String("slice", "ci.slice", "systemd slice for segments")
44+ user := fs.String("user", "ci", "user segments run as")
45+ identity := fs.String("identity", "", "age identity loaded into each segment (LoadCredential, SECRETS_IDENTITY)")
46+ pasta := fs.String("pasta", "", "pasta binary: give each segment its own network namespace (egress only)")
47+ var props, env listFlag
48+ fs.Var(&props, "property", "extra unit property, e.g. MemoryMax=8G (repeatable)")
49+ fs.Var(&env, "setenv", "extra environment for segments, NAME=value (repeatable)")
50+ ntfy := fs.String("ntfy", "", "ntfy topic URL for gates and failures")
51+ ntfyToken := fs.String("ntfy-token-file", "", "ntfy access token")
52+ ntfySuccess := fs.Bool("ntfy-success", false, "also notify successful runs")
53+ appview := fs.String("appview-url", "", "appview URL linked from notifications")
54+ if err := fs.Parse(args[1:]); err != nil {
55+ return exitUsage
56+ }
57+ if *bridgeURL == "" || *knot == "" || *cloneURL == "" {
58+ fmt.Fprintln(os.Stderr, "ci runner serve: --bridge, --knot and --clone-url are required")
59+ return exitUsage
60+ }
61+ logger := slog.Default()
62+ tokens := map[string]*string{}
63+ for name, path := range map[string]string{"bridge": *bridgeToken, "api": *tokenFile, "ntfy": *ntfyToken} {
64+ t, err := readToken(path)
65+ if err != nil {
66+ fmt.Fprintln(os.Stderr, "ci runner serve:", err)
67+ return exitFailure
68+ }
69+ tokens[name] = &t
70+ }
71+ exe, err := os.Executable()
72+ if err != nil {
73+ fmt.Fprintln(os.Stderr, "ci runner serve:", err)
74+ return exitFailure
75+ }
76+ if err := os.MkdirAll(*state, 0o755); err != nil {
77+ fmt.Fprintln(os.Stderr, "ci runner serve:", err)
78+ return exitFailure
79+ }
80+ cfg := runner.Config{
81+ RunsDir: filepath.Join(*state, "runs"), CacheDir: filepath.Join(*state, "cache"), LocksDir: filepath.Join(*state, "locks"),
82+ Knot: *knot, OwnerDid: *owner, CloneURL: *cloneURL, MaxConcurrent: *maxConc,
83+ }
84+ if *ntfy != "" {
85+ cfg.Notify = runner.Ntfy(*ntfy, *tokens["ntfy"], *appview, *ntfySuccess, logger)
86+ }
87+ segEnv := append([]string{"CI_CLONE_URL=" + *cloneURL}, env...)
88+ var l runner.Launcher
89+ switch *launcher {
90+ case "systemd":
91+ l = &runner.Systemd{Engine: exe, Shell: *shell, LocksDir: cfg.LocksDir, Slice: *slice, User: *user,
92+ CacheDir: cfg.CacheDir, Identity: *identity, Pasta: *pasta, Properties: props, Env: segEnv}
93+ case "local":
94+ l = &runner.Local{Engine: exe, Shell: *shell, LocksDir: cfg.LocksDir, Env: segEnv}
95+ default:
96+ fmt.Fprintf(os.Stderr, "ci runner serve: --launcher %q (want systemd or local)\n", *launcher)
97+ return exitUsage
98+ }
99+ log, err := eventlog.Open(filepath.Join(*state, "runner.db"))
100+ if err != nil {
101+ fmt.Fprintln(os.Stderr, "ci runner serve:", err)
102+ return exitFailure
103+ }
104+ defer log.Close()
105+ r, err := runner.New(cfg, log, l, logger)
106+ if err != nil {
107+ fmt.Fprintln(os.Stderr, "ci runner serve:", err)
108+ return exitFailure
109+ }
110+ return serve(*listen, r.Handler(*tokens["api"]), func(ctx context.Context) {
111+ if err := r.Reconcile(ctx); err != nil {
112+ logger.Error("reconcile", "err", err)
113+ }
114+ go r.Loop(ctx)
115+ if err := r.Follow(ctx, *bridgeURL, *tokens["bridge"]); err != nil && ctx.Err() == nil {
116+ logger.Error("bridge stream", "err", err)
117+ }
118+ })
119+}
+80,
-0
1@@ -0,0 +1,80 @@
2+package main
3+
4+import (
5+ "context"
6+ "flag"
7+ "fmt"
8+ "os"
9+ "os/signal"
10+ "path/filepath"
11+ "syscall"
12+
13+ "git.kilimanjaro.io/ci/internal/engine"
14+)
15+
16+func init() {
17+ commands["runner-exec"] = command{"(runner internal) run or resume one segment of a prepared run", cmdRunnerExec}
18+}
19+
20+// cmdRunnerExec is what the runner starts in a run's sandboxed unit: it runs
21+// a run the runner prepared (from its first step, or --from an action for a
22+// manual run), or answers its gate and continues. It always pauses at gates
23+// (the runner publishes them; a person answers in the appview).
24+func cmdRunnerExec(args []string) int {
25+ fs := flag.NewFlagSet("ci runner-exec", flag.ContinueOnError)
26+ from := fs.String("from", "", "start at this action (a manual run)")
27+ choose := fs.Int("choose", 0, "answer the gate with option N (1-based)")
28+ dismiss := fs.Bool("dismiss", false, "dismiss the gate")
29+ expire := fs.Bool("expire", false, "expire the gate")
30+ actor := fs.String("actor", "", "who answered the gate")
31+ shell := fs.String("shell", "devenv", "devenv or bash")
32+ locks := fs.String("locks-dir", "", "concurrency lock directory")
33+ if err := fs.Parse(args); err != nil || fs.NArg() != 1 {
34+ fmt.Fprintln(os.Stderr, "usage: ci runner-exec [flags] <run-dir>")
35+ return exitUsage
36+ }
37+ f := &runFlags{shell: *shell}
38+ ex, err := f.executor()
39+ if err != nil {
40+ fmt.Fprintln(os.Stderr, "ci:", err)
41+ return exitUsage
42+ }
43+ // systemd LoadCredential= puts the runner's age identity here; the
44+ // secrets tool reads SECRETS_IDENTITY.
45+ if cd := os.Getenv("CREDENTIALS_DIRECTORY"); cd != "" && os.Getenv("SECRETS_IDENTITY") == "" {
46+ if p := filepath.Join(cd, "age-identity"); fileExists(p) {
47+ _ = os.Setenv("SECRETS_IDENTITY", p)
48+ }
49+ }
50+ dir := engine.Dir(fs.Arg(0))
51+ e, err := engine.Open(dir, ex, engine.Options{Detach: true, LocksDir: *locks})
52+ if err != nil {
53+ fmt.Fprintln(os.Stderr, "ci:", err)
54+ return exitFailure
55+ }
56+ defer e.Log.Close()
57+ ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
58+ defer stop()
59+
60+ st := e.State()
61+ switch {
62+ case *choose > 0 || *dismiss || *expire:
63+ st, err = e.Resume(ctx, engine.Choice{Option: *choose - 1, Dismiss: *dismiss, Expire: *expire, Actor: *actor})
64+ case len(st.History) > 0 || st.Status != engine.StatusRunning:
65+ err = fmt.Errorf("run %s already started (%s); answer its gate instead", dir.Base(), st.Status)
66+ case *from != "":
67+ st, err = e.RunFrom(ctx, *from)
68+ default:
69+ st, err = e.Run(ctx)
70+ }
71+ if err != nil {
72+ fmt.Fprintln(os.Stderr, "ci:", err)
73+ return exitFailure
74+ }
75+ return exitCode(st.Status)
76+}
77+
78+func fileExists(p string) bool {
79+ fi, err := os.Stat(p)
80+ return err == nil && fi.Mode().IsRegular()
81+}
+1,
-1
1@@ -34,7 +34,7 @@
2 ./lexicons
3 ];
4 };
5- vendorHash = "sha256-2lb+pmVbboWhcg5vFHPGaP9imrXvFGis+I+eWEIcu0g=";
6+ vendorHash = "sha256-wShjEvQW4KV3hT4SlF5AYJzpbWonuFkg+2cbP325qjY=";
7 subPackages = [ "cmd/ci" ];
8 env.CGO_ENABLED = "0";
9 ldflags = [ "-X main.version=${version}" ];
+15,
-0
1@@ -163,3 +163,18 @@ func (e *Engine) Resume(ctx context.Context, c Choice) (*State, error) {
2 }
3 return e.drive(ctx, d)
4 }
5+
6+// MarkLost ends a run whose process died without finishing it (the runner
7+// found its unit gone): it is recorded as a failure with detail.
8+func MarkLost(dir Dir, detail string) error {
9+ e, err := Open(dir, nil, Options{})
10+ if err != nil {
11+ return err
12+ }
13+ defer e.Log.Close()
14+ if Terminal(e.st.Status) || e.st.Status == StatusPaused {
15+ return nil
16+ }
17+ _, err = e.finish(End{StatusFailure, lex.ReasonError, detail}, e.lastAction())
18+ return err
19+}
+480,
-0
1@@ -0,0 +1,480 @@
2+package runner
3+
4+import (
5+ "bytes"
6+ "context"
7+ "encoding/json"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "net/http"
12+ "os"
13+ "path/filepath"
14+ "strings"
15+ "time"
16+
17+ "github.com/gorilla/websocket"
18+
19+ "git.kilimanjaro.io/ci/internal/engine"
20+ "git.kilimanjaro.io/ci/internal/eventlog"
21+ "git.kilimanjaro.io/ci/internal/lex"
22+ "git.kilimanjaro.io/ci/internal/manifest"
23+ "git.kilimanjaro.io/ci/internal/runlog"
24+ "git.kilimanjaro.io/ci/internal/tid"
25+)
26+
27+// Handler serves the runner's API. Everything but /healthz needs token
28+// (when set); the appview holds it and decides who may do what.
29+//
30+// GET /events records stream (spindle-compatible)
31+// GET /logs/{knot}/{rkey}/{name} a workflow's log, upstream LogLine messages
32+// POST /xrpc/sh.tangled.ci.triggerPipeline manual pipeline
33+// GET /api/runs[?repo=&limit=] runs, newest first
34+// GET /api/runs/{id} run row + state.json
35+// GET /api/runs/{id}/events[?after=N] events.ndjson lines after seq N
36+// GET /api/runs/{id}/log.txt plain-text log
37+// GET /api/runs/{id}/files/{path...} results/ and artifacts/ files
38+// POST /api/runs/{id}/gate {"option":N|"action":"x"|"dismiss":true, "actor":"..."}
39+// POST /api/runs/{id}/actions/{action} start a manual action run {"actor":"..."}
40+// POST /api/runs/{id}/cancel stop a running segment / dismiss a gate
41+// GET /healthz
42+func (r *Runner) Handler(token string) http.Handler {
43+ mux := http.NewServeMux()
44+ mux.Handle("GET /events", r.Log)
45+ mux.HandleFunc("GET /logs/{knot}/{rkey}/{name}", r.serveLogs)
46+ mux.HandleFunc("POST /xrpc/sh.tangled.ci.triggerPipeline", r.serveTrigger)
47+ mux.HandleFunc("GET /api/runs", r.serveRuns)
48+ mux.HandleFunc("GET /api/runs/{id}", r.serveRun)
49+ mux.HandleFunc("GET /api/runs/{id}/events", r.serveEvents)
50+ mux.HandleFunc("GET /api/runs/{id}/log.txt", r.serveText)
51+ mux.HandleFunc("GET /api/runs/{id}/files/{path...}", r.serveFile)
52+ mux.HandleFunc("POST /api/runs/{id}/gate", r.serveGate)
53+ mux.HandleFunc("POST /api/runs/{id}/actions/{action}", r.serveAction)
54+ mux.HandleFunc("POST /api/runs/{id}/cancel", r.serveCancel)
55+ authed := eventlog.RequireToken(token, mux)
56+ top := http.NewServeMux()
57+ top.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
58+ if _, err := r.Log.List(0, 1); err != nil {
59+ http.Error(w, err.Error(), http.StatusServiceUnavailable)
60+ return
61+ }
62+ fmt.Fprintln(w, "ok")
63+ })
64+ top.Handle("/", authed)
65+ return top
66+}
67+
68+func writeJSON(w http.ResponseWriter, code int, v any) {
69+ w.Header().Set("Content-Type", "application/json")
70+ w.WriteHeader(code)
71+ _ = json.NewEncoder(w).Encode(v)
72+}
73+
74+func fail(w http.ResponseWriter, code int, err error) {
75+ writeJSON(w, code, map[string]string{"error": err.Error()})
76+}
77+
78+// RunView is a run as the API returns it.
79+type RunView struct {
80+ Run
81+ State *engine.State `json:"state,omitempty"`
82+}
83+
84+func (r *Runner) view(x Run) RunView {
85+ v := RunView{Run: x}
86+ if st, err := engine.Load(engine.Dir(x.Dir)); err == nil {
87+ v.State = st
88+ }
89+ return v
90+}
91+
92+func (r *Runner) serveRuns(w http.ResponseWriter, req *http.Request) {
93+ limit := 50
94+ if n, err := fmt.Sscan(req.URL.Query().Get("limit"), &limit); n != 1 || err != nil || limit <= 0 || limit > 500 {
95+ limit = 50
96+ }
97+ runs, err := r.ListRuns(req.URL.Query().Get("repo"), limit)
98+ if err != nil {
99+ fail(w, 500, err)
100+ return
101+ }
102+ out := make([]RunView, 0, len(runs))
103+ for _, x := range runs {
104+ out = append(out, r.view(x))
105+ }
106+ writeJSON(w, 200, out)
107+}
108+
109+func (r *Runner) run(w http.ResponseWriter, req *http.Request) (Run, bool) {
110+ x, err := r.GetRun(req.PathValue("id"))
111+ if err != nil {
112+ fail(w, 404, fmt.Errorf("no run %s", req.PathValue("id")))
113+ return Run{}, false
114+ }
115+ return x, true
116+}
117+
118+func (r *Runner) serveRun(w http.ResponseWriter, req *http.Request) {
119+ if x, ok := r.run(w, req); ok {
120+ writeJSON(w, 200, r.view(x))
121+ }
122+}
123+
124+func (r *Runner) serveEvents(w http.ResponseWriter, req *http.Request) {
125+ x, ok := r.run(w, req)
126+ if !ok {
127+ return
128+ }
129+ after := 0
130+ _, _ = fmt.Sscan(req.URL.Query().Get("after"), &after)
131+ f, err := os.Open(engine.Dir(x.Dir).Events())
132+ if err != nil {
133+ fail(w, 500, err)
134+ return
135+ }
136+ defer f.Close()
137+ w.Header().Set("Content-Type", "application/x-ndjson")
138+ _ = runlog.Scan(f, func(ev lex.RunEvent) error {
139+ if ev.Seq > after {
140+ b, _ := json.Marshal(ev)
141+ _, err := w.Write(append(b, '\n'))
142+ return err
143+ }
144+ return nil
145+ })
146+}
147+
148+// serveText renders the log as plain text: what an agent reads (ci wait).
149+func (r *Runner) serveText(w http.ResponseWriter, req *http.Request) {
150+ x, ok := r.run(w, req)
151+ if !ok {
152+ return
153+ }
154+ f, err := os.Open(engine.Dir(x.Dir).Events())
155+ if err != nil {
156+ fail(w, 500, err)
157+ return
158+ }
159+ defer f.Close()
160+ w.Header().Set("Content-Type", "text/plain; charset=utf-8")
161+ _ = runlog.Scan(f, func(ev lex.RunEvent) error {
162+ _, err := io.WriteString(w, TextLine(ev))
163+ return err
164+ })
165+}
166+
167+// TextLine renders an event as plain text ("" for events not worth a line).
168+func TextLine(ev lex.RunEvent) string {
169+ switch ev.Kind {
170+ case lex.KindStart:
171+ return fmt.Sprintf("== %s (#%d): %s\n", ev.Action, ev.Attempt, firstLine(ev.Command))
172+ case lex.KindLog:
173+ return ev.Content + "\n"
174+ case lex.KindEnd:
175+ code := 0
176+ if ev.ExitCode != nil {
177+ code = *ev.ExitCode
178+ }
179+ return fmt.Sprintf("== %s: %s (exit %d)\n", ev.Action, ev.Status, code)
180+ case lex.KindResult:
181+ if ev.Result != nil && ev.Result.Summary != "" {
182+ return " " + strings.ReplaceAll(strings.TrimRight(ev.Result.Summary, "\n"), "\n", "\n ") + "\n"
183+ }
184+ case lex.KindGate:
185+ if g := ev.Gate; g != nil {
186+ return fmt.Sprintf("== gate %s: %s\n", g.State, orDefault(g.Prompt, g.Choice))
187+ }
188+ case lex.KindRun:
189+ if ev.Status != engine.StatusRunning {
190+ return fmt.Sprintf("== run %s %s\n", ev.Status, ev.Content)
191+ }
192+ }
193+ return ""
194+}
195+
196+func firstLine(s string) string {
197+ s = strings.TrimSpace(s)
198+ if i := strings.IndexByte(s, '\n'); i >= 0 {
199+ return s[:i] + " …"
200+ }
201+ return s
202+}
203+
204+func (r *Runner) serveFile(w http.ResponseWriter, req *http.Request) {
205+ x, ok := r.run(w, req)
206+ if !ok {
207+ return
208+ }
209+ p := filepath.Clean(req.PathValue("path"))
210+ if (!strings.HasPrefix(p, "results/") && !strings.HasPrefix(p, "artifacts/")) || strings.Contains(p, "..") {
211+ fail(w, 404, errors.New("not found"))
212+ return
213+ }
214+ full := filepath.Join(x.Dir, p)
215+ real, err := filepath.EvalSymlinks(full)
216+ if err != nil || !strings.HasPrefix(real, filepath.Join(x.Dir, filepath.Dir(p))) {
217+ fail(w, 404, errors.New("not found"))
218+ return
219+ }
220+ w.Header().Set("Content-Type", "text/plain; charset=utf-8")
221+ http.ServeFile(w, req, real)
222+}
223+
224+type gateAnswer struct {
225+ Option int `json:"option"` // 1-based
226+ Action string `json:"action"`
227+ Dismiss bool `json:"dismiss"`
228+ Actor string `json:"actor"`
229+}
230+
231+func (r *Runner) serveGate(w http.ResponseWriter, req *http.Request) {
232+ x, ok := r.run(w, req)
233+ if !ok {
234+ return
235+ }
236+ var a gateAnswer
237+ if err := json.NewDecoder(io.LimitReader(req.Body, 1<<16)).Decode(&a); err != nil {
238+ fail(w, 400, err)
239+ return
240+ }
241+ if err := r.AnswerGate(req.Context(), x, a); err != nil {
242+ fail(w, 409, err)
243+ return
244+ }
245+ writeJSON(w, 202, map[string]string{"run": x.ID})
246+}
247+
248+// AnswerGate starts the segment that answers a paused run's gate.
249+func (r *Runner) AnswerGate(ctx context.Context, x Run, a gateAnswer) error {
250+ st, err := engine.Load(engine.Dir(x.Dir))
251+ if err != nil {
252+ return err
253+ }
254+ if x.Status != engine.StatusPaused || st.Gate == nil {
255+ return fmt.Errorf("run %s is %s, not waiting at a gate", x.ID, x.Status)
256+ }
257+ seg := Segment{Actor: a.Actor, Dismiss: a.Dismiss}
258+ if !a.Dismiss {
259+ switch {
260+ case a.Option >= 1 && a.Option <= len(st.Gate.Options):
261+ seg.Choose = a.Option
262+ case a.Action != "":
263+ for i, o := range st.Gate.Options {
264+ if o.Action == a.Action {
265+ seg.Choose = i + 1
266+ break
267+ }
268+ }
269+ }
270+ if seg.Choose == 0 {
271+ return errors.New("no such option")
272+ }
273+ }
274+ if err := r.updateRun(x.ID, `status = ?`, engine.StatusRunning); err != nil {
275+ return err
276+ }
277+ return r.launch(context.WithoutCancel(ctx), x.ID, seg)
278+}
279+
280+func (r *Runner) serveAction(w http.ResponseWriter, req *http.Request) {
281+ x, ok := r.run(w, req)
282+ if !ok {
283+ return
284+ }
285+ var body struct {
286+ Actor string `json:"actor"`
287+ }
288+ _ = json.NewDecoder(io.LimitReader(req.Body, 1<<16)).Decode(&body)
289+ id, err := r.ManualAction(req.Context(), x, req.PathValue("action"), body.Actor)
290+ if err != nil {
291+ fail(w, 409, err)
292+ return
293+ }
294+ writeJSON(w, 202, map[string]string{"run": id})
295+}
296+
297+// ManualAction starts a child run of x at a manual action (DESIGN.md §9):
298+// it shares x's checkouts and $CI_STATE and uses the manifest x ran with.
299+func (r *Runner) ManualAction(ctx context.Context, x Run, action, actor string) (string, error) {
300+ parent := x
301+ if x.Parent != "" {
302+ if p, err := r.GetRun(x.Parent); err == nil {
303+ parent = p
304+ }
305+ }
306+ pst, err := engine.Load(engine.Dir(parent.Dir))
307+ if err != nil {
308+ return "", err
309+ }
310+ raw, err := os.ReadFile(engine.Dir(parent.Dir).ManifestFile())
311+ if err != nil {
312+ return "", err
313+ }
314+ m, err := manifest.Parse(pst.Meta.Workflow, raw)
315+ if err != nil {
316+ return "", err
317+ }
318+ a, ok := m.Actions[action]
319+ if !ok || !a.Manual {
320+ return "", fmt.Errorf("%s is not a manual action of %s", action, m.Name)
321+ }
322+ id := parent.ID + "-" + action + "-" + tid.New()
323+ meta := pst.Meta
324+ meta.Run, meta.Parent = id, parent.Dir
325+ dir := engine.Dir(filepath.Join(r.Cfg.RunsDir, id))
326+ e, err := engine.Create(m, dir, meta, nil, engine.Options{})
327+ if err != nil {
328+ return "", err
329+ }
330+ e.Log.Close()
331+ t := now()
332+ if err := r.insertRun(Run{ID: id, Pipeline: parent.Pipeline, Rkey: parent.Rkey, Workflow: parent.Workflow, Repo: parent.Repo,
333+ Dir: dir.Path(), Parent: parent.ID, Status: lex.StatusPending, Created: t, Updated: t}); err != nil {
334+ return "", err
335+ }
336+ return id, r.launch(context.WithoutCancel(ctx), id, Segment{From: action, Actor: actor})
337+}
338+
339+func (r *Runner) serveCancel(w http.ResponseWriter, req *http.Request) {
340+ x, ok := r.run(w, req)
341+ if !ok {
342+ return
343+ }
344+ switch x.Status {
345+ case engine.StatusPaused:
346+ if err := r.AnswerGate(req.Context(), x, gateAnswer{Dismiss: true, Actor: "cancel"}); err != nil {
347+ fail(w, 409, err)
348+ return
349+ }
350+ case engine.StatusRunning, lex.StatusPending:
351+ if x.handle == "" || r.Launcher.Stop(x.handle) != nil {
352+ fail(w, 409, errors.New("nothing to stop"))
353+ return
354+ }
355+ default:
356+ fail(w, 409, fmt.Errorf("run %s is %s", x.ID, x.Status))
357+ return
358+ }
359+ writeJSON(w, 202, map[string]string{"run": x.ID})
360+}
361+
362+func (r *Runner) serveTrigger(w http.ResponseWriter, req *http.Request) {
363+ var in lex.TriggerPipelineInput
364+ if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&in); err != nil || in.Trigger == nil {
365+ fail(w, 400, fmt.Errorf("InvalidRequest: %v", err))
366+ return
367+ }
368+ ref := ""
369+ if in.Trigger.Ref != nil {
370+ ref = *in.Trigger.Ref
371+ }
372+ uri, err := r.StartPipeline(req.Context(), Trigger{Kind: "manual", RepoDid: in.Repo, Sha: in.Trigger.Sha, Ref: ref,
373+ Inputs: in.Trigger.Inputs, Workflows: in.Workflows})
374+ if err != nil {
375+ fail(w, 400, err)
376+ return
377+ }
378+ if uri == "" {
379+ fail(w, 400, errors.New("no matching workflows"))
380+ return
381+ }
382+ writeJSON(w, 200, lex.TriggerPipelineOutput{Pipeline: uri})
383+}
384+
385+var upgrader = websocket.Upgrader{ReadBufferSize: 1024, WriteBufferSize: 1024}
386+
387+// serveLogs streams a workflow's log as upstream's /logs does: LogLine
388+// messages (control start/end per action execution, data per line), replay
389+// then tail, closed when the run ends.
390+func (r *Runner) serveLogs(w http.ResponseWriter, req *http.Request) {
391+ x, err := r.RunByWorkflow(req.PathValue("rkey"), req.PathValue("name"))
392+ if err != nil {
393+ http.Error(w, "no such workflow", http.StatusNotFound)
394+ return
395+ }
396+ conn, err := upgrader.Upgrade(w, req, nil)
397+ if err != nil {
398+ return
399+ }
400+ defer conn.Close()
401+ ctx, cancel := context.WithCancel(req.Context())
402+ defer cancel()
403+ go func() {
404+ for {
405+ if _, _, err := conn.NextReader(); err != nil {
406+ cancel()
407+ return
408+ }
409+ }
410+ }()
411+ dir := engine.Dir(x.Dir)
412+ var off int64
413+ lastPing := time.Now()
414+ for {
415+ n, err := sendLogLines(conn, dir.Events(), off)
416+ if err != nil {
417+ return
418+ }
419+ off += n
420+ if st, err := engine.Load(dir); err == nil && engine.Terminal(st.Status) && n == 0 {
421+ _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "log stream complete"), time.Now().Add(time.Second))
422+ return
423+ }
424+ if time.Since(lastPing) > 30*time.Second {
425+ if conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(time.Second)) != nil {
426+ return
427+ }
428+ lastPing = time.Now()
429+ }
430+ select {
431+ case <-ctx.Done():
432+ return
433+ case <-time.After(300 * time.Millisecond):
434+ }
435+ }
436+}
437+
438+func sendLogLines(conn *websocket.Conn, path string, off int64) (int64, error) {
439+ f, err := os.Open(path)
440+ if err != nil {
441+ return 0, err
442+ }
443+ defer f.Close()
444+ if _, err := f.Seek(off, io.SeekStart); err != nil {
445+ return 0, err
446+ }
447+ b, err := io.ReadAll(f)
448+ if err != nil {
449+ return 0, err
450+ }
451+ end := bytes.LastIndexByte(b, '\n')
452+ if end < 0 {
453+ return 0, nil
454+ }
455+ err = runlog.Scan(bytes.NewReader(b[:end+1]), func(ev lex.RunEvent) error {
456+ if l, ok := LogLineOf(ev); ok {
457+ msg, _ := json.Marshal(l)
458+ return conn.WriteMessage(websocket.TextMessage, msg)
459+ }
460+ return nil
461+ })
462+ return int64(end + 1), err
463+}
464+
465+// LogLineOf translates a run event into upstream's log message, if it is one.
466+func LogLineOf(ev lex.RunEvent) (lex.LogLine, bool) {
467+ switch ev.Kind {
468+ case lex.KindStart:
469+ return lex.LogLine{Kind: "control", Content: ev.Action, Time: ev.Time, StepId: ev.Attempt,
470+ StepStatus: "start", StepKind: "user", StepCommand: ev.Command}, true
471+ case lex.KindEnd:
472+ return lex.LogLine{Kind: "control", Content: ev.Action, Time: ev.Time, StepId: ev.Attempt,
473+ StepStatus: "end", StepKind: "user"}, true
474+ case lex.KindLog:
475+ if ev.Attempt == 0 {
476+ return lex.LogLine{}, false // engine messages outside an action
477+ }
478+ return lex.LogLine{Kind: "data", Content: ev.Content, Time: ev.Time, StepId: ev.Attempt, Stream: ev.Stream}, true
479+ }
480+ return lex.LogLine{}, false
481+}
+250,
-0
1@@ -0,0 +1,250 @@
2+package runner
3+
4+import (
5+ "bufio"
6+ "bytes"
7+ "context"
8+ "encoding/json"
9+ "fmt"
10+ "io"
11+ "os"
12+ "os/exec"
13+ "time"
14+
15+ "git.kilimanjaro.io/ci/internal/engine"
16+ "git.kilimanjaro.io/ci/internal/lex"
17+)
18+
19+// follow tails a run's events while its segment runs, publishing records,
20+// then settles the run once the segment's process is gone.
21+func (r *Runner) follow(ctx context.Context, run, handle string) {
22+ r.mu.Lock()
23+ if r.tailing[run] {
24+ r.mu.Unlock()
25+ return
26+ }
27+ r.tailing[run] = true
28+ r.mu.Unlock()
29+ defer func() {
30+ r.mu.Lock()
31+ delete(r.tailing, run)
32+ r.mu.Unlock()
33+ }()
34+
35+ x, err := r.GetRun(run)
36+ if err != nil {
37+ r.Logger.Error("follow", "run", run, "err", err)
38+ return
39+ }
40+ for {
41+ alive := handle != "" && r.Launcher.Alive(handle)
42+ if err := r.translate(&x); err != nil {
43+ r.Logger.Error("translate events", "run", run, "err", err)
44+ }
45+ if !alive {
46+ break
47+ }
48+ select {
49+ case <-ctx.Done():
50+ return
51+ case <-time.After(250 * time.Millisecond):
52+ }
53+ }
54+ r.settle(x)
55+}
56+
57+// translate publishes the run's new events (from its stored offset).
58+func (r *Runner) translate(x *Run) error {
59+ f, err := os.Open(engine.Dir(x.Dir).Events())
60+ if err != nil {
61+ return err
62+ }
63+ defer f.Close()
64+ if _, err := f.Seek(x.offset, io.SeekStart); err != nil {
65+ return err
66+ }
67+ b, err := io.ReadAll(f)
68+ if err != nil {
69+ return err
70+ }
71+ end := bytes.LastIndexByte(b, '\n')
72+ if end < 0 {
73+ return nil
74+ }
75+ sc := bufio.NewScanner(bytes.NewReader(b[:end+1]))
76+ sc.Buffer(make([]byte, 64*1024), 16*1024*1024)
77+ for sc.Scan() {
78+ var ev lex.RunEvent
79+ if err := json.Unmarshal(sc.Bytes(), &ev); err != nil {
80+ continue
81+ }
82+ switch {
83+ case ev.Kind == lex.KindGate && ev.Gate != nil:
84+ if _, err := r.Log.Append(lex.GateNSID, "", ev.Gate); err != nil {
85+ return err
86+ }
87+ case ev.Kind == lex.KindTransition && ev.Transition != nil:
88+ if _, err := r.Log.Append(lex.TransitionNSID, "", ev.Transition); err != nil {
89+ return err
90+ }
91+ }
92+ }
93+ x.offset += int64(end + 1)
94+ return r.updateRun(x.ID, `offset = ?`, x.offset)
95+}
96+
97+// settle records how a segment left its run: paused at a gate, finished, or
98+// lost (the process died without saying).
99+func (r *Runner) settle(x Run) {
100+ defer func() { <-r.slots }()
101+ st, err := engine.Load(engine.Dir(x.Dir))
102+ if err != nil {
103+ r.Logger.Error("settle", "run", x.ID, "err", err)
104+ return
105+ }
106+ if st.Status == engine.StatusRunning {
107+ if err := engine.MarkLost(engine.Dir(x.Dir), "the run's process ended without finishing it (see segment logs)"); err != nil {
108+ r.Logger.Error("mark lost", "run", x.ID, "err", err)
109+ }
110+ if err := r.translate(&x); err != nil {
111+ r.Logger.Error("translate", "run", x.ID, "err", err)
112+ }
113+ if st, err = engine.Load(engine.Dir(x.Dir)); err != nil {
114+ return
115+ }
116+ }
117+ if err := r.updateRun(x.ID, `status = ?`, st.Status); err != nil {
118+ r.Logger.Error("update run", "run", x.ID, "err", err)
119+ }
120+ r.Logger.Info("segment ended", "run", x.ID, "status", st.Status)
121+ switch {
122+ case st.Status == engine.StatusPaused && st.Gate != nil:
123+ r.notify(Notification{Kind: "gate", Run: x, Status: st.Status,
124+ Title: fmt.Sprintf("%s/%s: waiting for you", x.Repo, x.Workflow),
125+ Message: orDefault(st.Gate.Prompt, "Choose the next step") + summaryLine(st.Gate.Summary)})
126+ case engine.Terminal(st.Status):
127+ if x.Parent == "" {
128+ status, msg := tangledStatus(st)
129+ r.status(x.Pipeline, x.Workflow, status, msg)
130+ }
131+ detail := ""
132+ if st.End != nil {
133+ detail = st.End.Detail
134+ }
135+ r.notify(Notification{Kind: "finished", Run: x, Status: st.Status,
136+ Title: fmt.Sprintf("%s/%s: %s", x.Repo, x.Workflow, st.Status), Message: detail})
137+ }
138+}
139+
140+func summaryLine(s string) string {
141+ if s == "" {
142+ return ""
143+ }
144+ line, _, _ := bytes.Cut([]byte(s), []byte("\n"))
145+ return "\n" + string(line)
146+}
147+
148+// tangledStatus maps a finished run onto sh.tangled.pipeline.status.
149+func tangledStatus(st *engine.State) (string, string) {
150+ detail := ""
151+ if st.End != nil {
152+ detail = st.End.Detail
153+ }
154+ switch st.Status {
155+ case engine.StatusSuccess:
156+ return lex.StatusSuccess, ""
157+ case engine.StatusCancelled:
158+ if st.End != nil && st.End.Reason == lex.ReasonExpired {
159+ return lex.StatusTimeout, detail
160+ }
161+ return lex.StatusCancelled, detail
162+ }
163+ return lex.StatusFailed, detail
164+}
165+
166+func (r *Runner) notify(n Notification) {
167+ if r.Cfg.Notify != nil {
168+ r.Cfg.Notify(n)
169+ }
170+}
171+
172+// Reconcile picks up runs after a restart: it follows live segments again,
173+// settles finished ones, starts runs that never started, and expires gates.
174+func (r *Runner) Reconcile(ctx context.Context) error {
175+ rows, err := r.db().Query(`select ` + runCols + ` from runs where status in ('pending', 'running')`)
176+ if err != nil {
177+ return err
178+ }
179+ var runs []Run
180+ for rows.Next() {
181+ x, err := scanRun(rows)
182+ if err != nil {
183+ rows.Close()
184+ return err
185+ }
186+ runs = append(runs, x)
187+ }
188+ rows.Close()
189+ for _, x := range runs {
190+ if x.handle == "" {
191+ if err := r.launch(ctx, x.ID, Segment{}); err != nil {
192+ r.Logger.Error("start pending run", "run", x.ID, "err", err)
193+ }
194+ continue
195+ }
196+ select {
197+ case r.slots <- struct{}{}: // follow releases it in settle
198+ case <-ctx.Done():
199+ return ctx.Err()
200+ }
201+ go r.follow(context.WithoutCancel(ctx), x.ID, x.handle)
202+ }
203+ return nil
204+}
205+
206+// ExpireGates answers expired gates of paused runs (on a timer).
207+func (r *Runner) ExpireGates(ctx context.Context) {
208+ rows, err := r.db().Query(`select ` + runCols + ` from runs where status = 'paused'`)
209+ if err != nil {
210+ return
211+ }
212+ var due []string
213+ for rows.Next() {
214+ x, err := scanRun(rows)
215+ if err != nil {
216+ break
217+ }
218+ if st, err := engine.Load(engine.Dir(x.Dir)); err == nil && st.Gate != nil && st.Gate.Expired(time.Now()) {
219+ due = append(due, x.ID)
220+ }
221+ }
222+ rows.Close()
223+ for _, id := range due {
224+ if err := r.launch(ctx, id, Segment{Expire: true, Actor: "runner"}); err != nil {
225+ r.Logger.Error("expire gate", "run", id, "err", err)
226+ }
227+ }
228+}
229+
230+// Loop runs the gate expiry timer until ctx ends.
231+func (r *Runner) Loop(ctx context.Context) {
232+ for {
233+ select {
234+ case <-ctx.Done():
235+ return
236+ case <-time.After(30 * time.Second):
237+ r.ExpireGates(ctx)
238+ }
239+ }
240+}
241+
242+func gitOut(ctx context.Context, gitDir string, args ...string) (string, error) {
243+ cmd := exec.CommandContext(ctx, "git", append([]string{"--git-dir", gitDir}, args...)...)
244+ var stderr bytes.Buffer
245+ cmd.Stderr = &stderr
246+ out, err := cmd.Output()
247+ if err != nil {
248+ return "", fmt.Errorf("git %v: %v: %s", args, err, bytes.TrimSpace(stderr.Bytes()))
249+ }
250+ return string(out), nil
251+}
+248,
-0
1@@ -0,0 +1,248 @@
2+package runner
3+
4+import (
5+ "context"
6+ "fmt"
7+ "os"
8+ "os/exec"
9+ "path/filepath"
10+ "strconv"
11+ "strings"
12+ "syscall"
13+
14+ "git.kilimanjaro.io/ci/internal/engine"
15+ "git.kilimanjaro.io/ci/internal/lex"
16+)
17+
18+// Segment is one engine process: a fresh run, a manual run starting at an
19+// action, or a gate answer.
20+type Segment struct {
21+ From string // ci runner-exec --from
22+ Choose int // 1-based option, 0 for none
23+ Dismiss bool
24+ Expire bool
25+ Actor string
26+}
27+
28+func (s Segment) args() []string {
29+ var a []string
30+ if s.From != "" {
31+ a = append(a, "--from", s.From)
32+ }
33+ if s.Choose > 0 {
34+ a = append(a, "--choose", strconv.Itoa(s.Choose))
35+ }
36+ if s.Dismiss {
37+ a = append(a, "--dismiss")
38+ }
39+ if s.Expire {
40+ a = append(a, "--expire")
41+ }
42+ if s.Actor != "" {
43+ a = append(a, "--actor", s.Actor)
44+ }
45+ return a
46+}
47+
48+// Job is what a Launcher starts.
49+type Job struct {
50+ Run string // run id
51+ Dir string // run directory
52+ Repo string
53+ Segment int
54+ Args []string // ci runner-exec arguments, without the run dir
55+}
56+
57+// Launcher starts and inspects engine processes.
58+type Launcher interface {
59+ // Start starts the job and returns a handle for Alive and Stop.
60+ Start(ctx context.Context, j Job) (string, error)
61+ // Alive reports whether the process behind handle is still running.
62+ Alive(handle string) bool
63+ // Stop terminates it.
64+ Stop(handle string) error
65+}
66+
67+// Local runs `ci runner-exec` as a child process: for development and tests.
68+type Local struct {
69+ Engine string // path of the ci binary
70+ Shell string // devenv or bash
71+ LocksDir string
72+ Env []string // extra environment
73+}
74+
75+func (l *Local) Start(ctx context.Context, j Job) (string, error) {
76+ args := append([]string{"runner-exec", "--shell", orDefault(l.Shell, "devenv"), "--locks-dir", l.LocksDir}, j.Args...)
77+ cmd := exec.Command(l.Engine, append(args, j.Dir)...)
78+ cmd.Env = append(os.Environ(), l.Env...)
79+ logf, err := os.OpenFile(filepath.Join(j.Dir, fmt.Sprintf("segment-%d.log", j.Segment)), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
80+ if err != nil {
81+ return "", err
82+ }
83+ cmd.Stdout, cmd.Stderr = logf, logf
84+ cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
85+ if err := cmd.Start(); err != nil {
86+ logf.Close()
87+ return "", err
88+ }
89+ go func() { _ = cmd.Wait(); logf.Close() }()
90+ return "pid:" + strconv.Itoa(cmd.Process.Pid), nil
91+}
92+
93+func (l *Local) Alive(handle string) bool {
94+ pid, err := strconv.Atoi(strings.TrimPrefix(handle, "pid:"))
95+ if err != nil {
96+ return false
97+ }
98+ // a zombie (exited, not yet reaped) still answers signal 0
99+ if b, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)); err == nil {
100+ if f := strings.Fields(string(b)); len(f) > 2 && f[2] == "Z" {
101+ return false
102+ }
103+ }
104+ return syscall.Kill(pid, 0) == nil
105+}
106+
107+func (l *Local) Stop(handle string) error {
108+ pid, err := strconv.Atoi(strings.TrimPrefix(handle, "pid:"))
109+ if err != nil {
110+ return err
111+ }
112+ return syscall.Kill(-pid, syscall.SIGTERM)
113+}
114+
115+// Systemd runs each segment as a transient unit in a sandbox (DESIGN.md
116+// §10): its own slice, an unprivileged user, a read-only system, a private
117+// /tmp, and optionally its own network namespace through pasta. Transient
118+// units outlive the runner, so a runner restart (or a self-deploy) doesn't
119+// kill a run.
120+type Systemd struct {
121+ Engine string // path of the ci binary
122+ Shell string
123+ LocksDir string
124+ Slice string // default ci.slice
125+ User string // default ci
126+ // CacheDir holds per-repo HOMEs (shared state between runs of a repo).
127+ CacheDir string
128+ // Identity, if set, is loaded as the credential age-identity.
129+ Identity string
130+ // Pasta, if set, is the pasta binary: each segment gets its own network
131+ // namespace with egress, so dev services can't clash with the host's.
132+ Pasta string
133+ // Properties are extra unit properties (e.g. MemoryMax=8G).
134+ Properties []string
135+ // Env is extra environment, e.g. CI_CLONE_URL.
136+ Env []string
137+}
138+
139+// UnitName is the transient unit of a run's segment.
140+func UnitName(run string, segment int) string {
141+ return fmt.Sprintf("ci-%s-%d", run, segment)
142+}
143+
144+// Command returns the systemd-run command line for j.
145+func (s *Systemd) Command(j Job) []string {
146+ home := filepath.Join(s.CacheDir, "home", j.Repo)
147+ argv := []string{"systemd-run",
148+ "--unit=" + UnitName(j.Run, j.Segment),
149+ "--slice=" + orDefault(s.Slice, "ci.slice"),
150+ "--uid=" + orDefault(s.User, "ci"),
151+ "--gid=" + orDefault(s.User, "ci"),
152+ "--collect", "--quiet", "--no-block",
153+ "--working-directory=" + j.Dir,
154+ "--setenv=HOME=" + home,
155+ "--setenv=XDG_CACHE_HOME=" + filepath.Join(home, ".cache"),
156+ "-p", "ProtectSystem=strict",
157+ "-p", "ProtectHome=yes",
158+ "-p", "PrivateTmp=yes",
159+ "-p", "NoNewPrivileges=yes",
160+ "-p", "ReadWritePaths=" + strings.Join([]string{j.Dir, home, s.LocksDir, "/nix/var/nix/daemon-socket"}, " "),
161+ "-p", "StandardOutput=append:" + filepath.Join(j.Dir, fmt.Sprintf("segment-%d.log", j.Segment)),
162+ "-p", "StandardError=inherit",
163+ }
164+ if s.Identity != "" {
165+ argv = append(argv, "-p", "LoadCredential=age-identity:"+s.Identity)
166+ }
167+ for _, e := range s.Env {
168+ argv = append(argv, "--setenv="+e)
169+ }
170+ for _, p := range s.Properties {
171+ argv = append(argv, "-p", p)
172+ }
173+ argv = append(argv, "--")
174+ if s.Pasta != "" {
175+ // egress through user-mode NAT; no ports forwarded in either direction
176+ argv = append(argv, s.Pasta, "--config-net", "--quiet", "-t", "none", "-u", "none", "-T", "none", "-U", "none", "--")
177+ }
178+ argv = append(argv, s.Engine, "runner-exec", "--shell", orDefault(s.Shell, "devenv"), "--locks-dir", s.LocksDir)
179+ argv = append(argv, j.Args...)
180+ return append(argv, j.Dir)
181+}
182+
183+func (s *Systemd) Start(ctx context.Context, j Job) (string, error) {
184+ if err := os.MkdirAll(filepath.Join(s.CacheDir, "home", j.Repo), 0o755); err != nil {
185+ return "", err
186+ }
187+ argv := s.Command(j)
188+ out, err := exec.CommandContext(ctx, argv[0], argv[1:]...).CombinedOutput()
189+ if err != nil {
190+ return "", fmt.Errorf("systemd-run: %v: %s", err, strings.TrimSpace(string(out)))
191+ }
192+ return "unit:" + UnitName(j.Run, j.Segment), nil
193+}
194+
195+func (s *Systemd) Alive(handle string) bool {
196+ unit := strings.TrimPrefix(handle, "unit:")
197+ out, _ := exec.Command("systemctl", "show", "-p", "ActiveState", "--value", unit+".service").Output()
198+ st := strings.TrimSpace(string(out))
199+ return st == "active" || st == "activating" || st == "deactivating" || st == "reloading"
200+}
201+
202+func (s *Systemd) Stop(handle string) error {
203+ return exec.Command("systemctl", "stop", "--no-block", strings.TrimPrefix(handle, "unit:")+".service").Run()
204+}
205+
206+func orDefault(s, d string) string {
207+ if s == "" {
208+ return d
209+ }
210+ return s
211+}
212+
213+// launch queues a segment of run: it starts as soon as a slot is free, and
214+// its events are followed from then on. It returns at once; a segment that
215+// can't be started ends the run as lost.
216+func (r *Runner) launch(ctx context.Context, run string, seg Segment) error {
217+ if _, err := r.GetRun(run); err != nil {
218+ return err
219+ }
220+ ctx = context.WithoutCancel(ctx)
221+ go func() {
222+ r.slots <- struct{}{}
223+ x, err := r.GetRun(run)
224+ if err == nil {
225+ n := x.segment + 1
226+ var handle string
227+ handle, err = r.Launcher.Start(ctx, Job{Run: run, Dir: x.Dir, Repo: x.Repo, Segment: n, Args: seg.args()})
228+ if err == nil {
229+ if err := r.updateRun(run, `segment = ?, handle = ?, status = ?`, n, handle, lex.StatusRunning); err != nil {
230+ r.Logger.Error("update run", "run", run, "err", err)
231+ }
232+ r.Logger.Info("segment started", "run", run, "segment", n, "handle", handle)
233+ if n == 1 && x.Parent == "" {
234+ r.status(x.Pipeline, x.Workflow, lex.StatusRunning, "")
235+ }
236+ r.follow(ctx, run, handle) // releases the slot when the segment ends
237+ return
238+ }
239+ }
240+ r.Logger.Error("start segment", "run", run, "err", err)
241+ if x.Dir != "" {
242+ _ = engine.MarkLost(engine.Dir(x.Dir), "could not start the run: "+err.Error())
243+ r.follow(ctx, run, "") // settles it as failed and releases the slot
244+ return
245+ }
246+ <-r.slots
247+ }()
248+ return nil
249+}
+65,
-0
1@@ -0,0 +1,65 @@
2+package runner
3+
4+import (
5+ "context"
6+ "log/slog"
7+ "net/http"
8+ "strings"
9+ "time"
10+)
11+
12+// Notification is a gate waiting for a person, or a finished run.
13+type Notification struct {
14+ Kind string // gate | finished
15+ Run Run
16+ Status string
17+ Title string
18+ Message string
19+}
20+
21+// Ntfy returns a notifier posting to an ntfy topic URL (e.g.
22+// https://ntfy.sh/<topic>). Gates and failures are sent with high priority;
23+// successes only if notifySuccess. appview, if set, is linked from each
24+// notification as a view action (no tokens in notifications).
25+func Ntfy(topicURL, token, appview string, notifySuccess bool, logger *slog.Logger) func(Notification) {
26+ return func(n Notification) {
27+ if n.Kind == "finished" && n.Status == "success" && !notifySuccess {
28+ return
29+ }
30+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
31+ defer cancel()
32+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, topicURL, strings.NewReader(n.Message))
33+ if err != nil {
34+ logger.Error("ntfy", "err", err)
35+ return
36+ }
37+ req.Header.Set("Title", n.Title)
38+ switch {
39+ case n.Kind == "gate":
40+ req.Header.Set("Priority", "high")
41+ req.Header.Set("Tags", "raised_hand")
42+ case n.Status == "success":
43+ req.Header.Set("Tags", "white_check_mark")
44+ default:
45+ req.Header.Set("Priority", "high")
46+ req.Header.Set("Tags", "x")
47+ }
48+ if appview != "" {
49+ u := strings.TrimSuffix(appview, "/") + "/runs/" + n.Run.ID
50+ req.Header.Set("Click", u)
51+ req.Header.Set("Actions", "view, Open, "+u)
52+ }
53+ if token != "" {
54+ req.Header.Set("Authorization", "Bearer "+token)
55+ }
56+ resp, err := http.DefaultClient.Do(req)
57+ if err != nil {
58+ logger.Error("ntfy", "err", err)
59+ return
60+ }
61+ resp.Body.Close()
62+ if resp.StatusCode >= 300 {
63+ logger.Error("ntfy", "status", resp.Status)
64+ }
65+ }
66+}
+413,
-0
1@@ -0,0 +1,413 @@
2+// Package runner is the CI host's daemon (DESIGN.md §10): spindle-compatible
3+// on the outside, a host for the engine on the inside.
4+//
5+// It follows the bridge's sh.tangled.git.refUpdate stream, reads the pushed
6+// commit's .tangled/workflows from a mirror, matches `when:`, records a
7+// sh.tangled.pipeline, prepares a run directory per matching workflow, and
8+// starts the engine (`ci runner-exec`) in a sandboxed unit per segment. It
9+// tails each run's events.ndjson to publish sh.tangled.pipeline.status,
10+// run.rtw.ci.gate and run.rtw.ci.transition records on its own /events, and
11+// serves logs on /logs in upstream's format. Gates are answered through its
12+// API (the appview), which starts the next segment.
13+package runner
14+
15+import (
16+ "context"
17+ "database/sql"
18+ "encoding/json"
19+ "errors"
20+ "fmt"
21+ "log/slog"
22+ "os"
23+ "path"
24+ "path/filepath"
25+ "slices"
26+ "strings"
27+ "sync"
28+ "time"
29+
30+ "git.kilimanjaro.io/ci/internal/checkout"
31+ "git.kilimanjaro.io/ci/internal/engine"
32+ "git.kilimanjaro.io/ci/internal/eventlog"
33+ "git.kilimanjaro.io/ci/internal/ident"
34+ "git.kilimanjaro.io/ci/internal/lex"
35+ "git.kilimanjaro.io/ci/internal/manifest"
36+ "git.kilimanjaro.io/ci/internal/tid"
37+)
38+
39+// Config configures a Runner.
40+type Config struct {
41+ RunsDir string // run directories
42+ CacheDir string // mirrors and per-repo HOMEs
43+ LocksDir string // concurrency locks
44+ // Knot is the forge host: pipelines are at://did:web:<Knot>/sh.tangled.pipeline/<rkey>.
45+ Knot string
46+ // OwnerDid owns the repos (sh.tangled.pipeline#triggerRepo.did).
47+ OwnerDid string
48+ // CloneURL is a template with {repo}, e.g. [email protected]:{repo}.
49+ CloneURL string
50+ // MaxConcurrent limits segments running at once (default 2).
51+ MaxConcurrent int
52+ // Notify is called for gates and finished runs (ntfy), may be nil.
53+ Notify func(Notification)
54+}
55+
56+// Runner is the daemon.
57+type Runner struct {
58+ Cfg Config
59+ Log *eventlog.Log
60+ Launcher Launcher
61+ Logger *slog.Logger
62+
63+ src *checkout.Source
64+ slots chan struct{}
65+ mu sync.Mutex
66+ // tailing marks runs whose events are being followed.
67+ tailing map[string]bool
68+}
69+
70+// New prepares a runner and its tables.
71+func New(cfg Config, log *eventlog.Log, l Launcher, logger *slog.Logger) (*Runner, error) {
72+ if cfg.MaxConcurrent <= 0 {
73+ cfg.MaxConcurrent = 2
74+ }
75+ if cfg.LocksDir == "" {
76+ cfg.LocksDir = filepath.Join(cfg.RunsDir, ".locks")
77+ }
78+ if logger == nil {
79+ logger = slog.Default()
80+ }
81+ for _, d := range []string{cfg.RunsDir, cfg.CacheDir, cfg.LocksDir} {
82+ if err := os.MkdirAll(d, 0o755); err != nil {
83+ return nil, err
84+ }
85+ }
86+ r := &Runner{
87+ Cfg: cfg, Log: log, Launcher: l, Logger: logger,
88+ src: &checkout.Source{MirrorsDir: filepath.Join(cfg.CacheDir, "mirrors"), URL: checkout.URLFunc(cfg.CloneURL, "")},
89+ slots: make(chan struct{}, cfg.MaxConcurrent),
90+ tailing: map[string]bool{},
91+ }
92+ if _, err := log.DB().Exec(`
93+ create table if not exists cursors (source text primary key, cursor integer not null);
94+ create table if not exists runs (
95+ id text primary key,
96+ pipeline text not null,
97+ rkey text not null,
98+ workflow text not null,
99+ repo text not null,
100+ dir text not null,
101+ parent text not null default '',
102+ status text not null,
103+ offset integer not null default 0,
104+ segment integer not null default 0,
105+ handle text not null default '',
106+ created text not null,
107+ updated text not null
108+ );
109+ create index if not exists runs_rkey on runs (rkey, workflow);`); err != nil {
110+ return nil, err
111+ }
112+ return r, nil
113+}
114+
115+// Run is a row of the runs table.
116+type Run struct {
117+ ID string `json:"id"`
118+ Pipeline string `json:"pipeline"`
119+ Rkey string `json:"rkey"`
120+ Workflow string `json:"workflow"`
121+ Repo string `json:"repo"`
122+ Dir string `json:"dir"`
123+ Parent string `json:"parent,omitempty"`
124+ Status string `json:"status"`
125+ Created string `json:"created"`
126+ Updated string `json:"updated"`
127+ offset int64
128+ segment int
129+ handle string
130+}
131+
132+func (r *Runner) db() *sql.DB { return r.Log.DB() }
133+
134+func (r *Runner) cursor(source string) int64 {
135+ var c int64
136+ _ = r.db().QueryRow(`select cursor from cursors where source = ?`, source).Scan(&c)
137+ return c
138+}
139+
140+func (r *Runner) setCursor(source string, c int64) error {
141+ _, err := r.db().Exec(`insert into cursors (source, cursor) values (?, ?) on conflict(source) do update set cursor = excluded.cursor`, source, c)
142+ return err
143+}
144+
145+const runCols = `id, pipeline, rkey, workflow, repo, dir, parent, status, offset, segment, handle, created, updated`
146+
147+func scanRun(sc interface{ Scan(...any) error }) (Run, error) {
148+ var x Run
149+ 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)
150+ return x, err
151+}
152+
153+// GetRun returns one run.
154+func (r *Runner) GetRun(id string) (Run, error) {
155+ return scanRun(r.db().QueryRow(`select `+runCols+` from runs where id = ?`, id))
156+}
157+
158+// RunByWorkflow finds a pipeline's run of a workflow (for /logs).
159+func (r *Runner) RunByWorkflow(rkey, workflow string) (Run, error) {
160+ return scanRun(r.db().QueryRow(`select `+runCols+` from runs where rkey = ? and workflow = ? and parent = '' order by created limit 1`, rkey, workflow))
161+}
162+
163+// ListRuns returns the most recent runs, newest first, optionally of a repo.
164+func (r *Runner) ListRuns(repo string, limit int) ([]Run, error) {
165+ q := `select ` + runCols + ` from runs`
166+ args := []any{}
167+ if repo != "" {
168+ q += ` where repo = ?`
169+ args = append(args, repo)
170+ }
171+ q += ` order by created desc limit ?`
172+ args = append(args, limit)
173+ rows, err := r.db().Query(q, args...)
174+ if err != nil {
175+ return nil, err
176+ }
177+ defer rows.Close()
178+ var out []Run
179+ for rows.Next() {
180+ x, err := scanRun(rows)
181+ if err != nil {
182+ return nil, err
183+ }
184+ out = append(out, x)
185+ }
186+ return out, rows.Err()
187+}
188+
189+func (r *Runner) insertRun(x Run) error {
190+ _, err := r.db().Exec(`insert into runs (`+runCols+`) values (?, ?, ?, ?, ?, ?, ?, ?, 0, 0, '', ?, ?)`,
191+ x.ID, x.Pipeline, x.Rkey, x.Workflow, x.Repo, x.Dir, x.Parent, x.Status, x.Created, x.Updated)
192+ return err
193+}
194+
195+func (r *Runner) updateRun(id string, set string, args ...any) error {
196+ args = append(args, now(), id)
197+ _, err := r.db().Exec(`update runs set `+set+`, updated = ? where id = ?`, args...)
198+ return err
199+}
200+
201+func now() string { return time.Now().UTC().Format(time.RFC3339Nano) }
202+
203+// pipelineURI is a pipeline's AT-URI, as the spindle forms it.
204+func (r *Runner) pipelineURI(rkey string) string {
205+ return fmt.Sprintf("at://%s/%s/%s", ident.ServiceDID(r.Cfg.Knot), lex.PipelineNSID, rkey)
206+}
207+
208+// Follow consumes the bridge's stream until ctx ends.
209+func (r *Runner) Follow(ctx context.Context, url, token string) error {
210+ source := "bridge:" + url
211+ return eventlog.Subscribe(ctx, url, r.cursor(source), token, func(ev lex.Event) error {
212+ if ev.Nsid == lex.GitRefUpdateNSID {
213+ var u lex.GitRefUpdate
214+ if err := json.Unmarshal(ev.Event, &u); err != nil {
215+ r.Logger.Error("bad ref update", "rkey", ev.Rkey, "err", err)
216+ } else if err := r.OnRefUpdate(ctx, u); err != nil {
217+ // a push we can't process is logged, not retried forever
218+ r.Logger.Error("ref update", "repo", u.Repo, "ref", u.Ref, "err", err)
219+ }
220+ }
221+ return r.setCursor(source, ev.Created)
222+ })
223+}
224+
225+// Trigger says what started a pipeline.
226+type Trigger struct {
227+ Kind string // push | manual
228+ RepoDid string
229+ Ref string
230+ OldSha string
231+ Sha string
232+ ChangedFiles []string
233+ Inputs []*lex.PipelinePair
234+ Workflows []string // manual: only these (default all)
235+}
236+
237+// OnRefUpdate starts the workflows a push matches.
238+func (r *Runner) OnRefUpdate(ctx context.Context, u lex.GitRefUpdate) error {
239+ if u.NewSha == "" || strings.Trim(u.NewSha, "0") == "" {
240+ return nil // a deleted ref
241+ }
242+ for _, o := range u.PushOptions {
243+ if o == "skip-ci" || o == "ci-skip" {
244+ r.Logger.Info("push asked to skip ci", "repo", u.Repo, "ref", u.Ref)
245+ return nil
246+ }
247+ }
248+ _, err := r.StartPipeline(ctx, Trigger{Kind: "push", RepoDid: u.Repo, Ref: u.Ref, OldSha: u.OldSha, Sha: u.NewSha, ChangedFiles: u.ChangedFiles})
249+ return err
250+}
251+
252+// StartPipeline reads the workflows at the trigger's commit, records a
253+// pipeline for those that match, and starts a run of each. It returns the
254+// pipeline's AT-URI, or "" if no workflow matched.
255+func (r *Runner) StartPipeline(ctx context.Context, t Trigger) (string, error) {
256+ host, repo, err := ident.ParseRepoDID(t.RepoDid)
257+ if err != nil {
258+ return "", err
259+ }
260+ if host != r.Cfg.Knot {
261+ return "", fmt.Errorf("repo %s is not on this runner's forge %s", t.RepoDid, r.Cfg.Knot)
262+ }
263+ mirror, err := r.src.Mirror(ctx, repo)
264+ if err != nil {
265+ return "", err
266+ }
267+ if t.Sha, err = checkout.Resolve(ctx, mirror, t.Sha); err != nil {
268+ return "", err
269+ }
270+ defaultBranch := "main"
271+ if out, err := gitOut(ctx, mirror, "symbolic-ref", "--short", "HEAD"); err == nil {
272+ defaultBranch = strings.TrimSpace(out)
273+ }
274+ files, err := workflowFiles(ctx, mirror, t.Sha)
275+ if err != nil {
276+ return "", err
277+ }
278+ var matched []*manifest.Manifest
279+ var invalid []invalidWorkflow
280+ for _, f := range files {
281+ raw, err := gitOut(ctx, mirror, "show", t.Sha+":"+f)
282+ if err != nil {
283+ return "", err
284+ }
285+ m, err := manifest.Parse(manifest.NameOf(f), []byte(raw))
286+ if errors.Is(err, manifest.ErrOtherEngine) {
287+ continue
288+ }
289+ if err != nil {
290+ // an invalid workflow doesn't stop the others; it is logged and
291+ // listed as failed in the pipeline so its author sees why
292+ r.Logger.Warn("invalid workflow", "repo", repo, "file", f, "err", err)
293+ invalid = append(invalid, invalidWorkflow{manifest.NameOf(f), raw, err})
294+ continue
295+ }
296+ if len(t.Workflows) > 0 && !slices.Contains(t.Workflows, m.Name) {
297+ continue
298+ }
299+ if t.Kind == "manual" || m.Matches(manifest.Event{Kind: t.Kind, Ref: t.Ref, ChangedFiles: t.ChangedFiles}) {
300+ matched = append(matched, m)
301+ }
302+ }
303+ if len(matched) == 0 && len(invalid) == 0 {
304+ r.Logger.Info("no workflow matched", "repo", repo, "ref", t.Ref, "sha", t.Sha)
305+ return "", nil
306+ }
307+
308+ rkey := tid.New()
309+ uri := r.pipelineURI(rkey)
310+ repoName := repo
311+ tm := &lex.PipelineTriggerMetadata{
312+ Kind: t.Kind,
313+ Repo: &lex.PipelineTriggerRepo{Knot: r.Cfg.Knot, Did: r.Cfg.OwnerDid, RepoDid: &t.RepoDid, Repo: &repoName, DefaultBranch: defaultBranch},
314+ }
315+ switch t.Kind {
316+ case "push":
317+ tm.Push = &lex.PipelinePushTriggerData{Ref: t.Ref, NewSha: t.Sha, OldSha: t.OldSha}
318+ default:
319+ ref := t.Ref
320+ tm.Manual = &lex.PipelineManualTriggerData{Sha: t.Sha, Inputs: t.Inputs}
321+ if ref != "" {
322+ tm.Manual.Ref = &ref
323+ }
324+ }
325+ p := lex.Pipeline{LexiconTypeID: lex.PipelineNSID, TriggerMetadata: tm}
326+ for _, m := range matched {
327+ p.Workflows = append(p.Workflows, &lex.PipelineWorkflow{
328+ Name: m.Name, Engine: lex.EngineDevenv, Clone: &lex.PipelineCloneOpts{Depth: 0}, Raw: string(m.Raw),
329+ })
330+ }
331+ for _, w := range invalid {
332+ p.Workflows = append(p.Workflows, &lex.PipelineWorkflow{
333+ Name: w.name, Engine: lex.EngineDevenv, Clone: &lex.PipelineCloneOpts{}, Raw: w.raw,
334+ })
335+ }
336+ if _, err := r.Log.Append(lex.PipelineNSID, rkey, p); err != nil {
337+ return "", err
338+ }
339+ for _, w := range invalid {
340+ r.status(uri, w.name, lex.StatusFailed, "invalid workflow: "+w.err.Error())
341+ }
342+ r.Logger.Info("pipeline", "uri", uri, "repo", repo, "sha", t.Sha, "workflows", len(matched))
343+ for _, m := range matched {
344+ meta := engine.Meta{
345+ Run: rkey + "-" + m.Name, Workflow: m.Name, Repo: repo, RepoDid: t.RepoDid, Knot: r.Cfg.Knot,
346+ Sha: t.Sha, Ref: t.Ref, DefaultBranch: defaultBranch, Kind: t.Kind, Pipeline: uri, Host: "runner",
347+ }
348+ if err := r.prepare(ctx, m, meta, mirror, rkey); err != nil {
349+ r.Logger.Error("prepare run", "run", meta.Run, "err", err)
350+ r.status(uri, m.Name, lex.StatusFailed, "could not prepare the run: "+err.Error())
351+ }
352+ }
353+ return uri, nil
354+}
355+
356+// prepare lays out a run directory and creates the run, then starts it.
357+func (r *Runner) prepare(ctx context.Context, m *manifest.Manifest, meta engine.Meta, mirror, rkey string) error {
358+ dir := engine.Dir(filepath.Join(r.Cfg.RunsDir, meta.Run))
359+ meta.Root = filepath.Join(dir.Path(), meta.Repo)
360+ if err := dir.Create(); err != nil {
361+ return err
362+ }
363+ origin, _ := r.src.URL(meta.Repo)
364+ checked, err := checkout.Layout(ctx, r.src, dir.Path(), checkout.Main{Repo: meta.Repo, From: mirror, Sha: meta.Sha, Origin: origin}, m.Repos, nil)
365+ if err != nil {
366+ return err
367+ }
368+ for _, c := range checked {
369+ meta.Checkouts = append(meta.Checkouts, engine.Checkout{Repo: c.Repo, Path: c.Path, Rev: c.Rev, Env: c.Env})
370+ }
371+ e, err := engine.Create(m, dir, meta, nil, engine.Options{})
372+ if err != nil {
373+ return err
374+ }
375+ e.Log.Close()
376+ t := now()
377+ if err := r.insertRun(Run{ID: meta.Run, Pipeline: meta.Pipeline, Rkey: rkey, Workflow: m.Name, Repo: meta.Repo,
378+ Dir: dir.Path(), Status: lex.StatusPending, Created: t, Updated: t}); err != nil {
379+ return err
380+ }
381+ r.status(meta.Pipeline, m.Name, lex.StatusPending, "")
382+ return r.launch(ctx, meta.Run, Segment{})
383+}
384+
385+// status appends a sh.tangled.pipeline.status record.
386+func (r *Runner) status(pipeline, workflow, status, errMsg string) {
387+ s := lex.PipelineStatus{LexiconTypeID: lex.PipelineStatusNSID, Pipeline: pipeline, Workflow: workflow, Status: status, CreatedAt: now()}
388+ if errMsg != "" {
389+ s.Error = &errMsg
390+ }
391+ if _, err := r.Log.Append(lex.PipelineStatusNSID, "", s); err != nil {
392+ r.Logger.Error("append status", "err", err)
393+ }
394+}
395+
396+type invalidWorkflow struct {
397+ name, raw string
398+ err error
399+}
400+
401+func workflowFiles(ctx context.Context, gitDir, sha string) ([]string, error) {
402+ out, err := gitOut(ctx, gitDir, "ls-tree", "--name-only", sha+":"+manifest.Dir)
403+ if err != nil {
404+ return nil, nil // no .tangled/workflows at that commit
405+ }
406+ var files []string
407+ for _, f := range strings.Fields(out) {
408+ if ext := path.Ext(f); ext == ".yml" || ext == ".yaml" {
409+ files = append(files, manifest.Dir+"/"+f)
410+ }
411+ }
412+ slices.Sort(files)
413+ return files, nil
414+}
+315,
-0
1@@ -0,0 +1,315 @@
2+package runner
3+
4+import (
5+ "context"
6+ "encoding/json"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "os"
11+ "os/exec"
12+ "path/filepath"
13+ "strings"
14+ "testing"
15+ "time"
16+
17+ "github.com/gorilla/websocket"
18+
19+ "git.kilimanjaro.io/ci/internal/engine"
20+ "git.kilimanjaro.io/ci/internal/eventlog"
21+ "git.kilimanjaro.io/ci/internal/ident"
22+ "git.kilimanjaro.io/ci/internal/lex"
23+)
24+
25+var ciBinary string
26+
27+func TestMain(m *testing.M) {
28+ dir, err := os.MkdirTemp("", "ci-runner-test")
29+ if err != nil {
30+ panic(err)
31+ }
32+ ciBinary = filepath.Join(dir, "ci")
33+ if out, err := exec.Command("go", "build", "-o", ciBinary, "../../cmd/ci").CombinedOutput(); err != nil {
34+ panic(string(out))
35+ }
36+ os.Symlink("ci", filepath.Join(dir, "ci-result"))
37+ code := m.Run()
38+ os.RemoveAll(dir)
39+ os.Exit(code)
40+}
41+
42+func sh(t *testing.T, dir string, args ...string) string {
43+ t.Helper()
44+ cmd := exec.Command(args[0], args[1:]...)
45+ cmd.Dir = dir
46+ cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@t", "GIT_COMMITTER_NAME=t",
47+ "GIT_COMMITTER_EMAIL=t@t", "GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_SYSTEM=/dev/null")
48+ out, err := cmd.CombinedOutput()
49+ if err != nil {
50+ t.Fatalf("%v: %v\n%s", args, err, out)
51+ }
52+ return strings.TrimSpace(string(out))
53+}
54+
55+const deployWF = `
56+when: [{ event: push, branch: main }]
57+engine: devenv
58+steps: [build, deploy, check]
59+actions:
60+ build: { command: "echo building; echo out > $CI_STATE/built" }
61+ deploy: { command: "echo deploying", gate: { prompt: "Deploy demo?" } }
62+ check: { command: "echo checking" }
63+ undo: { command: "cat $CI_STATE/built", manual: true }
64+`
65+
66+type fixture struct {
67+ t *testing.T
68+ r *Runner
69+ forge string
70+ sha string
71+}
72+
73+func newFixture(t *testing.T, files map[string]string) *fixture {
74+ t.Helper()
75+ t.Setenv("GIT_CONFIG_GLOBAL", "/dev/null")
76+ forge := t.TempDir()
77+ work := filepath.Join(t.TempDir(), "demo")
78+ os.MkdirAll(work, 0o755)
79+ sh(t, work, "git", "init", "-q", "-b", "main")
80+ for p, c := range files {
81+ os.MkdirAll(filepath.Dir(filepath.Join(work, p)), 0o755)
82+ os.WriteFile(filepath.Join(work, p), []byte(c), 0o644)
83+ }
84+ sh(t, work, "git", "add", "-A")
85+ sh(t, work, "git", "commit", "-qm", "init")
86+ sha := sh(t, work, "git", "rev-parse", "HEAD")
87+ sh(t, "", "git", "clone", "-q", "--bare", work, filepath.Join(forge, "demo.git"))
88+
89+ state := t.TempDir()
90+ log, err := eventlog.Open(filepath.Join(state, "runner.db"))
91+ if err != nil {
92+ t.Fatal(err)
93+ }
94+ t.Cleanup(func() { log.Close() })
95+ cfg := Config{RunsDir: filepath.Join(state, "runs"), CacheDir: filepath.Join(state, "cache"),
96+ Knot: "git.test", OwnerDid: "did:plc:me", CloneURL: filepath.Join(forge, "{repo}.git")}
97+ l := &Local{Engine: ciBinary, Shell: "bash", LocksDir: filepath.Join(state, "locks")}
98+ r, err := New(cfg, log, l, nil)
99+ if err != nil {
100+ t.Fatal(err)
101+ }
102+ return &fixture{t: t, r: r, forge: forge, sha: sha}
103+}
104+
105+func (f *fixture) push(ref string, opts ...string) error {
106+ return f.r.OnRefUpdate(context.Background(), lex.GitRefUpdate{
107+ Repo: ident.RepoDID("git.test", "demo"), Ref: ref, OldSha: strings.Repeat("0", 40), NewSha: f.sha, PushOptions: opts,
108+ })
109+}
110+
111+// waitStatus polls until run reaches status.
112+func (f *fixture) waitStatus(id, status string) Run {
113+ f.t.Helper()
114+ deadline := time.Now().Add(30 * time.Second)
115+ for time.Now().Before(deadline) {
116+ x, err := f.r.GetRun(id)
117+ if err == nil && x.Status == status {
118+ f.r.mu.Lock()
119+ busy := f.r.tailing[id]
120+ f.r.mu.Unlock()
121+ if !busy {
122+ return x
123+ }
124+ }
125+ time.Sleep(50 * time.Millisecond)
126+ }
127+ x, _ := f.r.GetRun(id)
128+ b, _ := os.ReadFile(filepath.Join(x.Dir, "segment-1.log"))
129+ f.t.Fatalf("run %s: status %q, want %q\n%s", id, x.Status, status, b)
130+ return Run{}
131+}
132+
133+func (f *fixture) records() map[string][]json.RawMessage {
134+ evs, _ := f.r.Log.List(0, 1000)
135+ out := map[string][]json.RawMessage{}
136+ for _, e := range evs {
137+ out[e.Nsid] = append(out[e.Nsid], e.Event)
138+ }
139+ return out
140+}
141+
142+func statuses(t *testing.T, recs []json.RawMessage, workflow string) []string {
143+ t.Helper()
144+ var out []string
145+ for _, b := range recs {
146+ var s lex.PipelineStatus
147+ json.Unmarshal(b, &s)
148+ if s.Workflow == workflow {
149+ out = append(out, s.Status)
150+ }
151+ }
152+ return out
153+}
154+
155+func TestPushGateApproveAndLogs(t *testing.T) {
156+ f := newFixture(t, map[string]string{
157+ ".tangled/workflows/deploy.yml": deployWF,
158+ ".tangled/workflows/bad.yml": "engine: devenv\nsteps: [nope]\nactions: {}\n",
159+ ".tangled/workflows/other.yml": "engine: nixery\nsteps: []\n",
160+ })
161+ if err := f.push("refs/heads/main"); err != nil {
162+ t.Fatal(err)
163+ }
164+ runs, _ := f.r.ListRuns("demo", 10)
165+ if len(runs) != 1 {
166+ t.Fatalf("runs %+v", runs)
167+ }
168+ id := runs[0].ID
169+ f.waitStatus(id, engine.StatusPaused)
170+
171+ recs := f.records()
172+ var p lex.Pipeline
173+ json.Unmarshal(recs[lex.PipelineNSID][0], &p)
174+ if len(p.Workflows) != 2 || p.TriggerMetadata.Push.NewSha != f.sha || *p.TriggerMetadata.Repo.RepoDid != "did:web:git.test:demo" {
175+ t.Errorf("pipeline %+v", p)
176+ }
177+ if got := statuses(t, recs[lex.PipelineStatusNSID], "bad"); len(got) != 1 || got[0] != lex.StatusFailed {
178+ t.Errorf("bad workflow statuses %v", got)
179+ }
180+ if len(recs[lex.GateNSID]) != 1 || !strings.Contains(string(recs[lex.GateNSID][0]), "Deploy demo?") {
181+ t.Errorf("gate records %s", recs[lex.GateNSID])
182+ }
183+
184+ x, _ := f.r.GetRun(id)
185+ if err := f.r.AnswerGate(context.Background(), x, gateAnswer{Action: "deploy", Actor: "did:plc:me"}); err != nil {
186+ t.Fatal(err)
187+ }
188+ f.waitStatus(id, engine.StatusSuccess)
189+ if got := statuses(t, f.records()[lex.PipelineStatusNSID], "deploy"); strings.Join(got, ",") != "pending,running,success" {
190+ t.Errorf("deploy statuses %v", got)
191+ }
192+
193+ // API: log.txt and /logs in upstream's LogLine format
194+ srv := httptest.NewServer(f.r.Handler("tok"))
195+ defer srv.Close()
196+ req, _ := http.NewRequest("GET", srv.URL+"/api/runs/"+id+"/log.txt", nil)
197+ req.Header.Set("Authorization", "Bearer tok")
198+ resp, err := http.DefaultClient.Do(req)
199+ if err != nil {
200+ t.Fatal(err)
201+ }
202+ txt, _ := io.ReadAll(resp.Body)
203+ resp.Body.Close()
204+ for _, want := range []string{"== build (#1)", "building", "== gate approved", "checking", "== run success"} {
205+ if !strings.Contains(string(txt), want) {
206+ t.Errorf("log.txt lacks %q:\n%s", want, txt)
207+ }
208+ }
209+ h := http.Header{"Authorization": {"Bearer tok"}}
210+ conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(srv.URL, "http")+"/logs/git.test/"+x.Rkey+"/deploy", h)
211+ if err != nil {
212+ t.Fatal(err)
213+ }
214+ var lines []lex.LogLine
215+ for {
216+ _, msg, err := conn.ReadMessage()
217+ if err != nil {
218+ break
219+ }
220+ var l lex.LogLine
221+ json.Unmarshal(msg, &l)
222+ lines = append(lines, l)
223+ }
224+ conn.Close()
225+ if len(lines) < 9 || lines[0].Kind != "control" || lines[0].StepStatus != "start" || lines[0].Content != "build" ||
226+ lines[1].Kind != "data" || lines[1].Content != "building" || lines[1].StepId != 1 {
227+ t.Errorf("log lines %+v", lines)
228+ }
229+
230+ // unauthenticated API calls are refused
231+ if resp, _ := http.Get(srv.URL + "/api/runs"); resp.StatusCode != 401 {
232+ t.Errorf("unauthenticated: %d", resp.StatusCode)
233+ }
234+
235+ // a manual action run against the finished run sees its $CI_STATE
236+ child, err := f.r.ManualAction(context.Background(), x, "undo", "did:plc:me")
237+ if err != nil {
238+ t.Fatal(err)
239+ }
240+ cx := f.waitStatus(child, engine.StatusSuccess)
241+ b, _ := os.ReadFile(engine.Dir(cx.Dir).Events())
242+ if !strings.Contains(string(b), `"content":"out"`) {
243+ t.Errorf("child did not read the parent's state:\n%s", b)
244+ }
245+ if _, err := f.r.ManualAction(context.Background(), x, "check", ""); err == nil {
246+ t.Error("non-manual action accepted")
247+ }
248+ // the child's run doesn't touch the workflow's Tangled status
249+ if got := statuses(t, f.records()[lex.PipelineStatusNSID], "deploy"); len(got) != 3 {
250+ t.Errorf("child run published statuses: %v", got)
251+ }
252+}
253+
254+func TestPushFilters(t *testing.T) {
255+ f := newFixture(t, map[string]string{".tangled/workflows/deploy.yml": deployWF})
256+ f.push("refs/heads/feature")
257+ f.push("refs/heads/main", "skip-ci")
258+ f.r.OnRefUpdate(context.Background(), lex.GitRefUpdate{Repo: ident.RepoDID("git.test", "demo"), Ref: "refs/heads/main", NewSha: strings.Repeat("0", 40)})
259+ if runs, _ := f.r.ListRuns("", 10); len(runs) != 0 {
260+ t.Errorf("runs started: %+v", runs)
261+ }
262+ if err := f.r.OnRefUpdate(context.Background(), lex.GitRefUpdate{Repo: ident.RepoDID("other.host", "demo"), Ref: "refs/heads/main", NewSha: f.sha}); err == nil {
263+ t.Error("accepted a repo on another forge")
264+ }
265+}
266+
267+func TestReconcile(t *testing.T) {
268+ f := newFixture(t, map[string]string{".tangled/workflows/deploy.yml": "engine: devenv\nsteps: [a]\nactions: { a: { command: 'true' } }\n"})
269+ if err := f.push("refs/heads/main"); err != nil {
270+ t.Fatal(err)
271+ }
272+ runs, _ := f.r.ListRuns("", 10)
273+ done := f.waitStatus(runs[0].ID, engine.StatusSuccess)
274+
275+ // a run the runner lost track of mid-way: its state says running, its
276+ // process is gone. After a restart it is settled as failed.
277+ lost := done
278+ lost.ID, lost.Dir = "lost", filepath.Join(f.r.Cfg.RunsDir, "lost")
279+ if err := os.CopyFS(lost.Dir, os.DirFS(done.Dir)); err != nil {
280+ t.Fatal(err)
281+ }
282+ st, _ := engine.Load(engine.Dir(lost.Dir))
283+ st.Status, st.End = engine.StatusRunning, nil
284+ st.Meta.Run = "lost"
285+ st.Save(engine.Dir(lost.Dir))
286+ f.r.insertRun(Run{ID: "lost", Pipeline: done.Pipeline, Rkey: done.Rkey, Workflow: "lostwf", Repo: "demo", Dir: lost.Dir, Status: "running", Created: now(), Updated: now()})
287+ f.r.updateRun("lost", `handle = ?, segment = 1`, "pid:999999999")
288+
289+ if err := f.r.Reconcile(context.Background()); err != nil {
290+ t.Fatal(err)
291+ }
292+ x := f.waitStatus("lost", engine.StatusFailure)
293+ st, _ = engine.Load(engine.Dir(x.Dir))
294+ if !strings.Contains(st.End.Detail, "ended without finishing") {
295+ t.Errorf("end %+v", st.End)
296+ }
297+ if got := statuses(t, f.records()[lex.PipelineStatusNSID], "lostwf"); len(got) != 1 || got[0] != lex.StatusFailed {
298+ t.Errorf("lost statuses %v", got)
299+ }
300+}
301+
302+func TestSystemdCommand(t *testing.T) {
303+ s := &Systemd{Engine: "/bin/ci", LocksDir: "/var/lib/ci/locks", CacheDir: "/var/lib/ci/cache",
304+ Identity: "/run/keys/ci-age", Pasta: "/bin/pasta", Properties: []string{"MemoryMax=8G"}, Env: []string{"CI_CLONE_URL=git@h:{repo}"}}
305+ argv := strings.Join(s.Command(Job{Run: "r1-deploy", Dir: "/var/lib/ci/runs/r1-deploy", Repo: "sovrn", Segment: 2, Args: []string{"--choose", "1"}}), " ")
306+ for _, want := range []string{
307+ "--unit=ci-r1-deploy-2", "--slice=ci.slice", "--uid=ci", "ProtectSystem=strict", "PrivateTmp=yes",
308+ "ReadWritePaths=/var/lib/ci/runs/r1-deploy /var/lib/ci/cache/home/sovrn /var/lib/ci/locks /nix/var/nix/daemon-socket",
309+ "LoadCredential=age-identity:/run/keys/ci-age", "--setenv=HOME=/var/lib/ci/cache/home/sovrn", "MemoryMax=8G",
310+ "-- /bin/pasta --config-net --quiet -t none -u none -T none -U none -- /bin/ci runner-exec --shell devenv --locks-dir /var/lib/ci/locks --choose 1 /var/lib/ci/runs/r1-deploy",
311+ } {
312+ if !strings.Contains(argv, want) {
313+ t.Errorf("command lacks %q:\n%s", want, argv)
314+ }
315+ }
316+}