14 files changed,
+769,
-38
+14,
-4
1@@ -423,7 +423,13 @@ in the group and holds it until it ends or pauses at a gate, so a run waiting
2 for a person never blocks others. `supersede: true` cancels older runs of the
3 same repo and workflow waiting at a gate in that group when a newer run
4 reaches one. Gates past `expiresAt` expire on `ci resume` (any answer) or
5-`ci resume --expire` (the runner calls it on a timer). Optional later: ask the runner for the lock
6+`ci resume --expire` (the runner calls it on a timer). Local runs with the
7+appview configured (`CI_APPVIEW_URL`, `CI_TOKEN`) take their groups from the
8+runner instead: a lease on the same file lock the runner's segments use
9+(`POST /api/locks/<group>`, renewed every 30 s, released after 2 minutes
10+without renewal), so a laptop deploy and a CI deploy can't overlap. If the
11+runner can't be reached they warn once and lock locally; `--local-locks`
12+forces that. Optional later: ask the runner for the lock
13 when reachable, so a laptop deploy and a CI deploy can't overlap.
14
15 ## 10. Runner daemon (CI host)
16@@ -548,9 +554,13 @@ is waiting for you).
17 - **ntfy** (sent by the runner, which sees every transition): gate waiting,
18 failure, optionally success; *view* actions link to the appview's run page
19 (no auth tokens inside notifications).
20-- **Local run reporting**: `ci run --report` (automatic when an `audited`
21- action runs) publishes a local run's events, so laptop deploys appear in the
22- history and their rollback buttons work.
23+- **Local run reporting**: `ci run|action|resume --report` (automatic when
24+ the run executed an `audited` action and the appview is configured)
25+ uploads a snapshot of the run (state, events, results, artifacts; no
26+ checkouts) after each segment, so laptop deploys appear in the history.
27+ Their manual actions (Roll back) run on the runner, which checks the
28+ recorded commits out again; a run of a working copy with changes, or with
29+ `--repo` links, has to be acted on locally (`ci action --run`).
30 - Later: ATProto OAuth login, so approvals carry the owner's DID.
31
32 ## 12. Secrets
+7,
-1
1@@ -83,6 +83,8 @@ type runFlags struct {
2 clean bool
3 rev string
4 repos repoFlag
5+ report bool
6+ local bool
7 }
8
9 // repoFlag collects --repo name=path overrides.
10@@ -109,6 +111,8 @@ func newRunFlags(name string) *runFlags {
11 f.fs.BoolVar(&f.clean, "clean", false, "run a fresh checkout of a commit (as the runner does) instead of the working copy")
12 f.fs.StringVar(&f.rev, "rev", "", "with --clean: the `revision` to check out (jj revset or git rev; default @ or HEAD)")
13 f.fs.Var(f.repos, "repo", "link an extra repo to a local directory instead of a clean clone: `name=path` (repeatable)")
14+ f.fs.BoolVar(&f.report, "report", false, "report the run to the appview (automatic when it runs an audited action; needs CI_APPVIEW_URL, CI_TOKEN)")
15+ f.fs.BoolVar(&f.local, "local-locks", false, "take concurrency groups locally only, not from the runner")
16 return f
17 }
18
19@@ -228,7 +232,7 @@ func runCommon(name string, args []string) int {
20 }
21 e, err := engine.Create(m, dir, meta, ex, engine.Options{
22 Stubs: f.stubs, Until: f.until, NoFollow: noFollow, MaxTransitions: f.maxTrans,
23- Gates: terminalGates, Detach: detach,
24+ Gates: terminalGates, Detach: detach, Locker: locker(dir, f.local),
25 })
26 if err != nil {
27 fmt.Fprintln(os.Stderr, "ci:", err)
28@@ -257,6 +261,8 @@ func runCommon(name string, args []string) int {
29 fmt.Fprintln(os.Stderr, "ci:", err)
30 return exitFailure
31 }
32+ e.Log.Close()
33+ maybeReport(dir, m, st, f.report)
34 return finished(st, dir)
35 }
36
+5,
-1
1@@ -126,6 +126,8 @@ func cmdResume(args []string) int {
2 detach := fs.Bool("detach", false, "pause at the next gate instead of prompting")
3 shell := fs.String("shell", "devenv", "how actions run: devenv or bash")
4 quiet := fs.Bool("q", false, "don't render events")
5+ reportRun := fs.Bool("report", false, "report the run to the appview (automatic for audited actions)")
6+ localLocks := fs.Bool("local-locks", false, "take concurrency groups locally only")
7 fs.Usage = func() {
8 fmt.Fprintln(os.Stderr, "usage: ci resume [flags] <run-id|run-dir|last>")
9 fs.PrintDefaults()
10@@ -147,7 +149,7 @@ func cmdResume(args []string) int {
11 fmt.Fprintln(os.Stderr, "ci:", err)
12 return exitFailure
13 }
14- e, err := engine.Open(dir, ex, engine.Options{Gates: terminalGates, Detach: *detach})
15+ e, err := engine.Open(dir, ex, engine.Options{Gates: terminalGates, Detach: *detach, Locker: locker(dir, *localLocks)})
16 if err != nil {
17 fmt.Fprintln(os.Stderr, "ci:", err)
18 return exitFailure
19@@ -195,6 +197,8 @@ func cmdResume(args []string) int {
20 fmt.Fprintln(os.Stderr, "ci:", err)
21 return exitFailure
22 }
23+ e.Log.Close()
24+ maybeReport(dir, e.M, st, *reportRun)
25 return finished(st, dir)
26 }
27
+162,
-0
1@@ -0,0 +1,162 @@
2+package main
3+
4+import (
5+ "bytes"
6+ "context"
7+ "encoding/json"
8+ "fmt"
9+ "io"
10+ "net/http"
11+ "net/url"
12+ "os"
13+ "path/filepath"
14+ "strings"
15+ "sync"
16+ "time"
17+
18+ "git.kilimanjaro.io/ci/internal/engine"
19+ "git.kilimanjaro.io/ci/internal/manifest"
20+)
21+
22+// remoteLocker takes concurrency groups from the runner (through the
23+// appview), so a local deploy and a CI deploy can't overlap. When the
24+// appview can't be reached it falls back to local file locks, once warning.
25+type remoteLocker struct {
26+ local engine.Locker
27+ warned sync.Once
28+}
29+
30+func (l *remoteLocker) TryLock(ctx context.Context, group, holder string) (func(), string, error) {
31+ path := "/api/locks/" + url.PathEscape(group) + "?holder=" + url.QueryEscape(holder)
32+ locked, by, err := leaseCall(ctx, http.MethodPost, path)
33+ if err != nil {
34+ l.warned.Do(func() {
35+ fmt.Fprintf(os.Stderr, "ci: warning: can't reach the runner for lock %s (%v); using a local lock only\n", group, err)
36+ })
37+ return l.local.TryLock(ctx, group, holder)
38+ }
39+ if !locked {
40+ return nil, by, nil
41+ }
42+ stop := make(chan struct{})
43+ go func() { // renew well within the lease's TTL
44+ for {
45+ select {
46+ case <-stop:
47+ return
48+ case <-time.After(30 * time.Second):
49+ _, _, _ = leaseCall(context.Background(), http.MethodPost, path)
50+ }
51+ }
52+ }()
53+ var once sync.Once
54+ return func() {
55+ once.Do(func() {
56+ close(stop)
57+ _, _, _ = leaseCall(context.Background(), http.MethodDelete, path)
58+ })
59+ }, "", nil
60+}
61+
62+func leaseCall(ctx context.Context, method, path string) (bool, string, error) {
63+ base, token, err := appviewClient()
64+ if err != nil {
65+ return false, "", err
66+ }
67+ ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
68+ defer cancel()
69+ req, err := http.NewRequestWithContext(ctx, method, base+path, nil)
70+ if err != nil {
71+ return false, "", err
72+ }
73+ req.Header.Set("Authorization", "Bearer "+token)
74+ resp, err := http.DefaultClient.Do(req)
75+ if err != nil {
76+ return false, "", err
77+ }
78+ defer resp.Body.Close()
79+ var out struct {
80+ Locked bool `json:"locked"`
81+ Holder string `json:"holder"`
82+ Error string `json:"error"`
83+ }
84+ _ = json.NewDecoder(resp.Body).Decode(&out)
85+ switch resp.StatusCode {
86+ case 200:
87+ return out.Locked || method == http.MethodDelete, out.Holder, nil
88+ case 409:
89+ return false, out.Holder, nil
90+ }
91+ return false, "", fmt.Errorf("%s: %s", resp.Status, out.Error)
92+}
93+
94+// locker picks the concurrency locks for a local run: the runner's when the
95+// appview is configured (CI_APPVIEW_URL, CI_TOKEN), unless local is set.
96+func locker(dir engine.Dir, local bool) engine.Locker {
97+ files := engine.FileLocker{Dir: filepath.Join(filepath.Dir(dir.Path()), ".locks")}
98+ if local {
99+ return files
100+ }
101+ if _, _, err := appviewClient(); err != nil {
102+ return files
103+ }
104+ return &remoteLocker{local: files}
105+}
106+
107+// audited reports whether the run executed an action marked audited.
108+func audited(m *manifest.Manifest, st *engine.State) bool {
109+ for _, x := range st.History {
110+ if m.Actions[x.Action].Audited {
111+ return true
112+ }
113+ }
114+ return false
115+}
116+
117+// report uploads a snapshot of a local run to the runner (through the
118+// appview): its state, events, results and artifacts, not its checkouts.
119+func report(dir engine.Dir) error {
120+ base, token, err := appviewClient()
121+ if err != nil {
122+ return err
123+ }
124+ var buf bytes.Buffer
125+ if err := engine.Snapshot(dir, &buf); err != nil {
126+ return err
127+ }
128+ req, err := http.NewRequest(http.MethodPut, base+"/api/runs/"+url.PathEscape(dir.Base())+"/import", &buf)
129+ if err != nil {
130+ return err
131+ }
132+ req.Header.Set("Authorization", "Bearer "+token)
133+ req.Header.Set("Content-Type", "application/gzip")
134+ resp, err := (&http.Client{Timeout: 2 * time.Minute}).Do(req)
135+ if err != nil {
136+ return err
137+ }
138+ defer resp.Body.Close()
139+ if resp.StatusCode >= 300 {
140+ b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
141+ return fmt.Errorf("%s: %s", resp.Status, strings.TrimSpace(string(b)))
142+ }
143+ return nil
144+}
145+
146+// maybeReport reports a run if asked, or if it executed an audited action
147+// and the appview is configured.
148+func maybeReport(dir engine.Dir, m *manifest.Manifest, st *engine.State, asked bool) {
149+ if !asked && !audited(m, st) {
150+ return
151+ }
152+ if _, _, err := appviewClient(); err != nil {
153+ if asked || audited(m, st) {
154+ fmt.Fprintln(os.Stderr, "ci: not reported to the appview:", err)
155+ }
156+ return
157+ }
158+ if err := report(dir); err != nil {
159+ fmt.Fprintln(os.Stderr, "ci: report:", err)
160+ return
161+ }
162+ fmt.Fprintf(os.Stderr, "ci: reported run %s to the appview\n", dir.Base())
163+}
+57,
-0
1@@ -253,3 +253,60 @@ func TestSafeNext(t *testing.T) {
2 }
3 }
4 }
5+
6+// A local `ci run` waits for a concurrency group the runner holds, then
7+// reports itself (its action is audited) so it shows on the runner.
8+func TestLocalRunUsesRunnerLocksAndReports(t *testing.T) {
9+ e := setup(t)
10+ repo := filepath.Join(t.TempDir(), "local")
11+ os.MkdirAll(filepath.Join(repo, ".tangled/workflows"), 0o755)
12+ os.WriteFile(filepath.Join(repo, ".tangled/workflows/deploy.yml"), []byte(`
13+engine: devenv
14+steps: [deploy]
15+actions:
16+ deploy: { command: "echo deployed-locally", concurrency: { group: fleet }, audited: true }
17+`), 0o644)
18+ sh(t, repo, "git", "init", "-q", "-b", "main")
19+ sh(t, repo, "git", "add", "-A")
20+ sh(t, repo, "git", "commit", "-qm", "init")
21+
22+ if ok, _, _ := e.r.Lock(context.Background(), "fleet", "ci-deploy-on-runner"); !ok {
23+ t.Fatal("runner lock")
24+ }
25+ runs := t.TempDir()
26+ cmd := exec.Command(ciBinary, "run", "--shell", "bash", "-q", "deploy")
27+ cmd.Dir = repo
28+ cmd.Env = append(os.Environ(), "CI_RUNS_DIR="+runs, "CI_APPVIEW_URL="+e.av.URL, "CI_TOKEN="+agentToken)
29+ var out strings.Builder
30+ cmd.Stdout, cmd.Stderr = &out, &out
31+ if err := cmd.Start(); err != nil {
32+ t.Fatal(err)
33+ }
34+ time.Sleep(1500 * time.Millisecond)
35+ ents, _ := os.ReadDir(runs)
36+ var id string
37+ for _, en := range ents {
38+ if en.IsDir() && en.Name() != ".locks" {
39+ id = en.Name()
40+ }
41+ }
42+ b, _ := os.ReadFile(filepath.Join(runs, id, "events.ndjson"))
43+ if !strings.Contains(string(b), "waiting for concurrency group fleet (held by run ci-deploy-on-runner)") || strings.Contains(string(b), "deployed-locally") {
44+ t.Fatalf("local run did not wait for the runner's lock:\n%s", b)
45+ }
46+ e.r.Unlock("fleet", "ci-deploy-on-runner")
47+ if err := cmd.Wait(); err != nil {
48+ t.Fatalf("local run: %v\n%s", err, out.String())
49+ }
50+ if !strings.Contains(out.String(), "reported run "+id) {
51+ t.Errorf("not reported:\n%s", out.String())
52+ }
53+ x, err := e.r.GetRun(id)
54+ if err != nil || x.Status != "success" {
55+ t.Fatalf("runner has %+v %v", x, err)
56+ }
57+ // and the lease was released: the runner can take the group again
58+ if ok, _, _ := e.r.Lock(context.Background(), "fleet", "next"); !ok {
59+ t.Error("lease not released")
60+ }
61+}
+30,
-0
1@@ -160,3 +160,33 @@ func isSha(s string) bool {
2 }
3 return true
4 }
5+
6+// Forward sends a request through to the runner unchanged (method, path
7+// with query, body) and copies the answer back: imports and lock leases from
8+// local runs.
9+func (c *Runner) Forward(w http.ResponseWriter, r *http.Request, path string) {
10+ req, err := http.NewRequestWithContext(r.Context(), r.Method, strings.TrimSuffix(c.URL, "/")+path, r.Body)
11+ if err != nil {
12+ http.Error(w, err.Error(), http.StatusBadGateway)
13+ return
14+ }
15+ if c.Token != "" {
16+ req.Header.Set("Authorization", "Bearer "+c.Token)
17+ }
18+ if ct := r.Header.Get("Content-Type"); ct != "" {
19+ req.Header.Set("Content-Type", ct)
20+ }
21+ hc := c.HTTP
22+ if hc == nil {
23+ hc = &http.Client{Timeout: 2 * time.Minute}
24+ }
25+ resp, err := hc.Do(req)
26+ if err != nil {
27+ http.Error(w, err.Error(), http.StatusBadGateway)
28+ return
29+ }
30+ defer resp.Body.Close()
31+ w.Header().Set("Content-Type", resp.Header.Get("Content-Type"))
32+ w.WriteHeader(resp.StatusCode)
33+ _, _ = io.Copy(w, resp.Body)
34+}
+9,
-0
1@@ -68,6 +68,15 @@ func (s *Server) Handler() http.Handler {
2 mux.HandleFunc("GET /api/runs/{id}/log.txt", s.api(s.apiText))
3 mux.HandleFunc("POST /api/trigger", s.api(s.apiTrigger))
4 mux.HandleFunc("POST /api/runs/{id}/actions/{action}", s.api(s.apiAction))
5+ // local runs: reporting (ci run --report) and concurrency leases
6+ mux.HandleFunc("PUT /api/runs/{id}/import", s.api(func(w http.ResponseWriter, r *http.Request, who Who) {
7+ s.Runner.Forward(w, r, "/api/runs/"+url.PathEscape(r.PathValue("id"))+"/import")
8+ }))
9+ lock := s.api(func(w http.ResponseWriter, r *http.Request, who Who) {
10+ s.Runner.Forward(w, r, "/api/locks/"+url.PathEscape(r.PathValue("group"))+"?"+r.URL.RawQuery)
11+ })
12+ mux.HandleFunc("POST /api/locks/{group}", lock)
13+ mux.HandleFunc("DELETE /api/locks/{group}", lock)
14 mux.HandleFunc("POST /api/runs/{id}/gate", s.api(func(w http.ResponseWriter, r *http.Request, who Who) {
15 apiError(w, http.StatusForbidden, "approving gates needs the owner's session in the appview")
16 }))
+4,
-1
1@@ -33,6 +33,9 @@ type Options struct {
2 Detach bool
3 // LocksDir holds concurrency locks; default <runs>/.locks.
4 LocksDir string
5+ // Locker overrides the file locks in LocksDir (the runner's, for local
6+ // runs that must not overlap CI deploys).
7+ Locker Locker
8 }
9
10 // Engine runs one workflow run.
11@@ -46,7 +49,7 @@ type Engine struct {
12
13 router *Router
14 st *State
15- locks map[string]*os.File
16+ locks map[string]func()
17 }
18
19 // Create starts a new run in dir: it writes the manifest copy, state.json and
+57,
-31
1@@ -26,55 +26,81 @@ func (e *Engine) locksDir() string {
2 return filepath.Join(filepath.Dir(e.Dir.Path()), ".locks")
3 }
4
5-// acquire blocks until this run holds group's lock, or ctx ends.
6+// Locker takes concurrency groups. TryLock doesn't wait: it returns a release
7+// function when it got the group, else who holds it.
8+type Locker interface {
9+ TryLock(ctx context.Context, group, holder string) (release func(), heldBy string, err error)
10+}
11+
12+// FileLocker is flock(2) on <Dir>/<group>.lock, holder written inside.
13+type FileLocker struct{ Dir string }
14+
15+func (l FileLocker) TryLock(_ context.Context, group, holder string) (func(), string, error) {
16+ if err := os.MkdirAll(l.Dir, 0o755); err != nil {
17+ return nil, "", err
18+ }
19+ f, err := os.OpenFile(filepath.Join(l.Dir, group+".lock"), os.O_CREATE|os.O_RDWR, 0o644)
20+ if err != nil {
21+ return nil, "", err
22+ }
23+ if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
24+ by, _ := os.ReadFile(f.Name())
25+ f.Close()
26+ if errors.Is(err, syscall.EWOULDBLOCK) {
27+ return nil, strings.TrimSpace(string(by)), nil
28+ }
29+ return nil, "", fmt.Errorf("lock %s: %w", group, err)
30+ }
31+ if err := f.Truncate(0); err == nil {
32+ _, _ = f.WriteAt([]byte(holder+"\n"), 0)
33+ }
34+ return func() {
35+ _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
36+ f.Close()
37+ }, "", nil
38+}
39+
40+func (e *Engine) locker() Locker {
41+ if e.Opts.Locker != nil {
42+ return e.Opts.Locker
43+ }
44+ return FileLocker{Dir: e.locksDir()}
45+}
46+
47+// acquire blocks until this run holds group, or ctx ends.
48 func (e *Engine) acquire(ctx context.Context, group string) error {
49 if _, held := e.locks[group]; held {
50 return nil
51 }
52- if err := os.MkdirAll(e.locksDir(), 0o755); err != nil {
53- return err
54- }
55- f, err := os.OpenFile(filepath.Join(e.locksDir(), group+".lock"), os.O_CREATE|os.O_RDWR, 0o644)
56- if err != nil {
57- return err
58- }
59- told := false
60+ told := ""
61 for {
62- err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
63- if err == nil {
64- break
65+ release, by, err := e.locker().TryLock(ctx, group, e.st.Meta.Run)
66+ if err != nil {
67+ return err
68 }
69- if !errors.Is(err, syscall.EWOULDBLOCK) {
70- f.Close()
71- return fmt.Errorf("lock %s: %w", group, err)
72+ if release != nil {
73+ if e.locks == nil {
74+ e.locks = map[string]func(){}
75+ }
76+ e.locks[group] = release
77+ return nil
78 }
79- if !told {
80- holder, _ := os.ReadFile(f.Name())
81+ if told != by {
82 e.emit(lex.RunEvent{Kind: lex.KindLog, Stream: "stderr",
83- Content: fmt.Sprintf("ci: waiting for concurrency group %s (held by run %s)", group, strings.TrimSpace(string(holder)))})
84- told = true
85+ Content: fmt.Sprintf("ci: waiting for concurrency group %s (held by run %s)", group, by)})
86+ told = by
87 }
88 select {
89 case <-ctx.Done():
90- f.Close()
91 return ctx.Err()
92 case <-time.After(time.Second):
93 }
94 }
95- if err := f.Truncate(0); err == nil {
96- _, _ = f.WriteAt([]byte(e.st.Meta.Run+"\n"), 0)
97- }
98- if e.locks == nil {
99- e.locks = map[string]*os.File{}
100- }
101- e.locks[group] = f
102- return nil
103 }
104
105 func (e *Engine) releaseLocks() {
106- for g, f := range e.locks {
107- _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
108- f.Close()
109+ for g, release := range e.locks {
110+ release()
111 delete(e.locks, g)
112 }
113 }
+57,
-0
1@@ -0,0 +1,57 @@
2+package engine
3+
4+import (
5+ "archive/tar"
6+ "compress/gzip"
7+ "io"
8+ "io/fs"
9+ "os"
10+ "path/filepath"
11+)
12+
13+// Snapshot writes a tar.gz of what describes a run: state.json,
14+// events.ndjson, manifest.yml, state/, results/ and artifacts/ (not its
15+// checkouts). `ci run --report` uploads it; the runner imports it.
16+func Snapshot(d Dir, w io.Writer) error {
17+ zw := gzip.NewWriter(w)
18+ tw := tar.NewWriter(zw)
19+ for _, rel := range []string{"state.json", "events.ndjson", "manifest.yml", "state", "results", "artifacts"} {
20+ root := filepath.Join(d.Path(), rel)
21+ err := filepath.WalkDir(root, func(p string, e fs.DirEntry, err error) error {
22+ if err != nil {
23+ if os.IsNotExist(err) {
24+ return nil
25+ }
26+ return err
27+ }
28+ if !e.Type().IsRegular() {
29+ return nil
30+ }
31+ name, err := filepath.Rel(d.Path(), p)
32+ if err != nil {
33+ return err
34+ }
35+ fi, err := e.Info()
36+ if err != nil {
37+ return err
38+ }
39+ if err := tw.WriteHeader(&tar.Header{Name: filepath.ToSlash(name), Mode: 0o644, Size: fi.Size(), ModTime: fi.ModTime()}); err != nil {
40+ return err
41+ }
42+ f, err := os.Open(p)
43+ if err != nil {
44+ return err
45+ }
46+ defer f.Close()
47+ _, err = io.Copy(tw, f)
48+ return err
49+ })
50+ if err != nil {
51+ return err
52+ }
53+ }
54+ if err := tw.Close(); err != nil {
55+ return err
56+ }
57+ return zw.Close()
58+}
+11,
-0
1@@ -39,6 +39,8 @@ import (
2 // POST /api/runs/{id}/gate {"option":N|"action":"x"|"dismiss":true, "actor":"..."}
3 // POST /api/runs/{id}/actions/{action} start a manual action run {"actor":"..."}
4 // POST /api/runs/{id}/cancel stop a running segment / dismiss a gate
5+// PUT /api/runs/{id}/import store a reported local run (tar.gz)
6+// POST|DELETE /api/locks/{group}?holder= lease a concurrency group (local runs)
7 // GET /healthz
8 func (r *Runner) Handler(token string) http.Handler {
9 mux := http.NewServeMux()
10@@ -53,6 +55,9 @@ func (r *Runner) Handler(token string) http.Handler {
11 mux.HandleFunc("POST /api/runs/{id}/gate", r.serveGate)
12 mux.HandleFunc("POST /api/runs/{id}/actions/{action}", r.serveAction)
13 mux.HandleFunc("POST /api/runs/{id}/cancel", r.serveCancel)
14+ mux.HandleFunc("PUT /api/runs/{id}/import", r.serveImport)
15+ mux.HandleFunc("POST /api/locks/{group}", r.serveLock)
16+ mux.HandleFunc("DELETE /api/locks/{group}", r.serveLock)
17 authed := eventlog.RequireToken(token, mux)
18 top := http.NewServeMux()
19 top.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
20@@ -348,6 +353,12 @@ func (r *Runner) ManualAction(ctx context.Context, x Run, action, actor string)
21 meta := pst.Meta
22 meta.Run, meta.Parent = id, parent.Dir
23 dir := engine.Dir(filepath.Join(r.Cfg.RunsDir, id))
24+ if meta.Host == "local" { // a reported local run: check its commits out here
25+ if err := r.relayout(ctx, &meta, m.Repos, dir); err != nil {
26+ _ = os.RemoveAll(dir.Path())
27+ return "", err
28+ }
29+ }
30 e, err := engine.Create(m, dir, meta, nil, engine.Options{})
31 if err != nil {
32 return "", err
+246,
-0
1@@ -0,0 +1,246 @@
2+package runner
3+
4+import (
5+ "archive/tar"
6+ "compress/gzip"
7+ "context"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "net/http"
12+ "os"
13+ "path/filepath"
14+ "regexp"
15+ "strings"
16+ "sync"
17+ "time"
18+
19+ "git.kilimanjaro.io/ci/internal/checkout"
20+ "git.kilimanjaro.io/ci/internal/engine"
21+ "git.kilimanjaro.io/ci/internal/lex"
22+)
23+
24+// Remote locks: a local run (`ci run` on the workstation) takes a
25+// concurrency group from the runner, as a lease on the same file lock the
26+// runner's segments use, so a laptop deploy and a CI deploy never overlap.
27+// The client renews the lease while it holds the group; a lease not renewed
28+// within its TTL is released.
29+
30+// LeaseTTL is how long a lease lives without renewal.
31+const LeaseTTL = 2 * time.Minute
32+
33+type lease struct {
34+ holder string
35+ release func()
36+ expires time.Time
37+}
38+
39+type leases struct {
40+ mu sync.Mutex
41+ m map[string]*lease
42+}
43+
44+func (r *Runner) leaseTable() *leases {
45+ r.mu.Lock()
46+ defer r.mu.Unlock()
47+ if r.leases == nil {
48+ r.leases = &leases{m: map[string]*lease{}}
49+ }
50+ return r.leases
51+}
52+
53+var groupName = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*$`)
54+
55+// Lock takes or renews group for holder; it returns who holds it if not.
56+func (r *Runner) Lock(ctx context.Context, group, holder string) (bool, string, error) {
57+ if !groupName.MatchString(group) || holder == "" {
58+ return false, "", errors.New("invalid group or holder")
59+ }
60+ t := r.leaseTable()
61+ t.mu.Lock()
62+ defer t.mu.Unlock()
63+ if l, ok := t.m[group]; ok {
64+ if l.holder == holder {
65+ l.expires = time.Now().Add(LeaseTTL)
66+ return true, holder, nil
67+ }
68+ if time.Now().Before(l.expires) {
69+ return false, l.holder, nil
70+ }
71+ l.release()
72+ delete(t.m, group)
73+ }
74+ release, by, err := engine.FileLocker{Dir: r.Cfg.LocksDir}.TryLock(ctx, group, holder)
75+ if err != nil || release == nil {
76+ return false, by, err
77+ }
78+ t.m[group] = &lease{holder: holder, release: release, expires: time.Now().Add(LeaseTTL)}
79+ return true, holder, nil
80+}
81+
82+// Unlock releases holder's lease on group.
83+func (r *Runner) Unlock(group, holder string) {
84+ t := r.leaseTable()
85+ t.mu.Lock()
86+ defer t.mu.Unlock()
87+ if l, ok := t.m[group]; ok && l.holder == holder {
88+ l.release()
89+ delete(t.m, group)
90+ }
91+}
92+
93+func (r *Runner) serveLock(w http.ResponseWriter, req *http.Request) {
94+ holder := req.URL.Query().Get("holder")
95+ if req.Method == http.MethodDelete {
96+ r.Unlock(req.PathValue("group"), holder)
97+ writeJSON(w, 200, map[string]any{"released": true})
98+ return
99+ }
100+ ok, by, err := r.Lock(req.Context(), req.PathValue("group"), holder)
101+ if err != nil {
102+ fail(w, 400, err)
103+ return
104+ }
105+ code := 200
106+ if !ok {
107+ code = 409
108+ }
109+ writeJSON(w, code, map[string]any{"locked": ok, "holder": by, "ttl": LeaseTTL.Seconds()})
110+}
111+
112+// Imported local runs: `ci run --report` uploads a snapshot of a local run
113+// (state, events, results, artifacts; no checkouts) so it shows in the
114+// appview's history, and its manual actions (rollback) can run here.
115+
116+var runID = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{5,80}$`)
117+
118+// importable names what a snapshot may contain.
119+func importable(name string) bool {
120+ switch name {
121+ case "state.json", "events.ndjson", "manifest.yml":
122+ return true
123+ }
124+ return strings.HasPrefix(name, "state/") || strings.HasPrefix(name, "results/") || strings.HasPrefix(name, "artifacts/")
125+}
126+
127+// Import stores a reported local run (a tar.gz of its run dir) as run id.
128+func (r *Runner) Import(id string, body io.Reader) error {
129+ if !runID.MatchString(id) {
130+ return fmt.Errorf("invalid run id %q", id)
131+ }
132+ if x, err := r.GetRun(id); err == nil && x.Pipeline != "" {
133+ return fmt.Errorf("run %s is not an imported run", id)
134+ }
135+ dir := filepath.Join(r.Cfg.RunsDir, id)
136+ tmp := dir + ".import"
137+ _ = os.RemoveAll(tmp)
138+ if err := os.MkdirAll(tmp, 0o755); err != nil {
139+ return err
140+ }
141+ defer func() { _ = os.RemoveAll(tmp) }()
142+ zr, err := gzip.NewReader(io.LimitReader(body, 256<<20))
143+ if err != nil {
144+ return err
145+ }
146+ tr := tar.NewReader(zr)
147+ for {
148+ h, err := tr.Next()
149+ if err == io.EOF {
150+ break
151+ }
152+ if err != nil {
153+ return err
154+ }
155+ name := filepath.Clean(h.Name)
156+ if h.Typeflag != tar.TypeReg || !importable(name) || strings.Contains(name, "..") {
157+ continue // checkouts, links and anything else stay behind
158+ }
159+ p := filepath.Join(tmp, name)
160+ if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
161+ return err
162+ }
163+ f, err := os.Create(p)
164+ if err != nil {
165+ return err
166+ }
167+ _, err = io.Copy(f, io.LimitReader(tr, 64<<20))
168+ f.Close()
169+ if err != nil {
170+ return err
171+ }
172+ }
173+ st, err := engine.Load(engine.Dir(tmp))
174+ if err != nil {
175+ return fmt.Errorf("snapshot: %w", err)
176+ }
177+ if st.Meta.Run != id {
178+ return fmt.Errorf("snapshot is of run %s, not %s", st.Meta.Run, id)
179+ }
180+ if st.Meta.Host != "local" {
181+ return errors.New("only local runs can be imported")
182+ }
183+ _ = os.RemoveAll(dir)
184+ if err := os.Rename(tmp, dir); err != nil {
185+ return err
186+ }
187+ if err := os.MkdirAll(tmp, 0o755); err != nil { // for the deferred RemoveAll
188+ return err
189+ }
190+ if _, err := r.GetRun(id); err == nil {
191+ return r.updateRun(id, `status = ?, dir = ?`, st.Status, dir)
192+ }
193+ return r.insertRun(Run{ID: id, Workflow: st.Meta.Workflow, Repo: st.Meta.Repo, Dir: dir,
194+ Status: st.Status, Created: st.Created, Updated: now()})
195+}
196+
197+func (r *Runner) serveImport(w http.ResponseWriter, req *http.Request) {
198+ if err := r.Import(req.PathValue("id"), req.Body); err != nil {
199+ fail(w, 400, err)
200+ return
201+ }
202+ writeJSON(w, 200, map[string]string{"run": req.PathValue("id")})
203+}
204+
205+// relayout gives a manual run of an imported local run checkouts of its
206+// own: the local paths mean nothing here, so each repo is checked out again
207+// at the commit the local run recorded. A dirty working copy can't be
208+// reproduced and is refused.
209+func (r *Runner) relayout(ctx context.Context, meta *engine.Meta, m []lex.Repo, dir engine.Dir) error {
210+ if meta.Dirty {
211+ return errors.New("the local run used a working copy with changes; run its actions locally (ci action --run)")
212+ }
213+ if meta.Sha == "" {
214+ return errors.New("the local run recorded no commit")
215+ }
216+ revs := map[string]string{}
217+ for _, c := range meta.Checkouts {
218+ if c.Rev == "linked" || c.Rev == "" {
219+ if c.Repo != meta.Repo {
220+ return fmt.Errorf("repo %s was a linked local directory (--repo); run its actions locally", c.Repo)
221+ }
222+ continue
223+ }
224+ revs[c.Repo] = c.Rev
225+ }
226+ mirror, err := r.src.Mirror(ctx, meta.Repo)
227+ if err != nil {
228+ return err
229+ }
230+ origin, _ := r.src.URL(meta.Repo)
231+ extras := make([]lex.Repo, len(m))
232+ for i, e := range m {
233+ e.Rev = revs[e.Repo]
234+ extras[i] = e
235+ }
236+ checked, err := checkout.Layout(ctx, r.src, dir.Path(), checkout.Main{Repo: meta.Repo, From: mirror, Sha: meta.Sha, Origin: origin}, extras, nil)
237+ if err != nil {
238+ return err
239+ }
240+ meta.Root = filepath.Join(dir.Path(), meta.Repo)
241+ meta.Checkouts = nil
242+ for _, c := range checked {
243+ meta.Checkouts = append(meta.Checkouts, engine.Checkout{Repo: c.Repo, Path: c.Path, Rev: c.Rev, Env: c.Env})
244+ }
245+ meta.Host = "runner"
246+ return nil
247+}
+109,
-0
1@@ -0,0 +1,109 @@
2+package runner
3+
4+import (
5+ "bytes"
6+ "context"
7+ "os"
8+ "path/filepath"
9+ "strings"
10+ "testing"
11+ "time"
12+
13+ "git.kilimanjaro.io/ci/internal/engine"
14+ "git.kilimanjaro.io/ci/internal/manifest"
15+)
16+
17+func TestLeases(t *testing.T) {
18+ f := newFixture(t, map[string]string{"README": "x"})
19+ ctx := context.Background()
20+ if ok, _, err := f.r.Lock(ctx, "fleet", "laptop-run"); !ok || err != nil {
21+ t.Fatalf("lock: %v %v", ok, err)
22+ }
23+ if ok, by, _ := f.r.Lock(ctx, "fleet", "other"); ok || by != "laptop-run" {
24+ t.Fatalf("second holder: %v %q", ok, by)
25+ }
26+ if ok, _, _ := f.r.Lock(ctx, "fleet", "laptop-run"); !ok {
27+ t.Fatal("renewal refused")
28+ }
29+ // the lease is the same file lock the runner's segments take
30+ release, by, err := engine.FileLocker{Dir: f.r.Cfg.LocksDir}.TryLock(ctx, "fleet", "segment")
31+ if release != nil || by != "laptop-run" || err != nil {
32+ t.Fatalf("segment got the group: %v %q %v", release != nil, by, err)
33+ }
34+ f.r.Unlock("fleet", "laptop-run")
35+ if ok, _, _ := f.r.Lock(ctx, "fleet", "other"); !ok {
36+ t.Fatal("lock after unlock")
37+ }
38+ // an expired lease is taken over
39+ f.r.leaseTable().m["fleet"].expires = time.Now().Add(-time.Second)
40+ if ok, _, _ := f.r.Lock(ctx, "fleet", "third"); !ok {
41+ t.Fatal("expired lease not released")
42+ }
43+ if _, _, err := f.r.Lock(ctx, "../x", "h"); err == nil {
44+ t.Error("bad group accepted")
45+ }
46+}
47+
48+// A local run's snapshot is imported; its manual rollback runs on the
49+// runner with fresh checkouts at the recorded commit and the local run's
50+// $CI_STATE.
51+func TestImportAndManualActionOfLocalRun(t *testing.T) {
52+ wf := "engine: devenv\nsteps: [deploy]\nactions:\n deploy: { command: 'echo prev-of-local > $CI_STATE/prev; cat marker' }\n rollback: { command: 'cat $CI_STATE/prev; cat marker', manual: true }\n"
53+ f := newFixture(t, map[string]string{".tangled/workflows/deploy.yml": wf, "marker": "from-commit"})
54+ m, err := manifest.Parse("deploy", []byte(wf))
55+ if err != nil {
56+ t.Fatal(err)
57+ }
58+ // the "local" run, somewhere else on disk
59+ local := engine.Dir(filepath.Join(t.TempDir(), "3mxlocalrun22"))
60+ root := filepath.Join(local.Path(), "demo")
61+ os.MkdirAll(root, 0o755)
62+ os.WriteFile(filepath.Join(root, "marker"), []byte("from-laptop"), 0o644)
63+ meta := engine.Meta{Run: "3mxlocalrun22", Workflow: "deploy", Repo: "demo", Sha: f.sha, Kind: "manual", Host: "local", Root: root,
64+ Checkouts: []engine.Checkout{{Repo: "demo", Path: root, Rev: "linked"}}}
65+ e, err := engine.Create(m, local, meta, engine.Bash{}, engine.Options{})
66+ if err != nil {
67+ t.Fatal(err)
68+ }
69+ if st, _ := e.Run(context.Background()); st.Status != engine.StatusSuccess {
70+ t.Fatalf("local run: %+v", st.End)
71+ }
72+ e.Log.Close()
73+
74+ var snap bytes.Buffer
75+ if err := engine.Snapshot(local, &snap); err != nil {
76+ t.Fatal(err)
77+ }
78+ if err := f.r.Import("3mxlocalrun22", &snap); err != nil {
79+ t.Fatal(err)
80+ }
81+ x, err := f.r.GetRun("3mxlocalrun22")
82+ if err != nil || x.Status != engine.StatusSuccess || x.Workflow != "deploy" {
83+ t.Fatalf("imported %+v %v", x, err)
84+ }
85+ if _, err := os.Stat(filepath.Join(x.Dir, "demo")); err == nil {
86+ t.Error("checkouts were imported")
87+ }
88+
89+ child, err := f.r.ManualAction(context.Background(), x, "rollback", "did:plc:me")
90+ if err != nil {
91+ t.Fatal(err)
92+ }
93+ cx := f.waitStatus(child, engine.StatusSuccess)
94+ b, _ := os.ReadFile(engine.Dir(cx.Dir).Events())
95+ if !strings.Contains(string(b), `"content":"prev-of-local"`) || !strings.Contains(string(b), `"content":"from-commit"`) {
96+ t.Errorf("rollback did not use the local state and the recorded commit:\n%s", b)
97+ }
98+
99+ // a dirty local run can't be acted on remotely
100+ st, _ := engine.Load(engine.Dir(x.Dir))
101+ st.Meta.Dirty = true
102+ st.Save(engine.Dir(x.Dir))
103+ if _, err := f.r.ManualAction(context.Background(), x, "rollback", ""); err == nil || !strings.Contains(err.Error(), "working copy with changes") {
104+ t.Errorf("dirty: %v", err)
105+ }
106+ // imports are only for local runs, and ids must be plain
107+ if err := f.r.Import("../../etc", strings.NewReader("")); err == nil {
108+ t.Error("bad id accepted")
109+ }
110+}
+1,
-0
1@@ -64,6 +64,7 @@ type Runner struct {
2 mu sync.Mutex
3 // tailing marks runs whose events are being followed.
4 tailing map[string]bool
5+ leases *leases
6 }
7
8 // New prepares a runner and its tables.