diff --git a/cmd/gitmote/main.go b/cmd/gitmote/main.go
index 76d4dfe..5c59619 100644
--- a/cmd/gitmote/main.go
+++ b/cmd/gitmote/main.go
@@ -106,7 +106,7 @@ func run(ctx context.Context, logger *slog.Logger, addr string) error {
ctx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
defer stop()
- gitHandler, ui, closeMeta, err := buildGitHandler(ctx, logger)
+ gitHandler, ui, reportAPI, closeMeta, err := buildGitHandler(ctx, logger)
if err != nil {
return err
}
@@ -118,7 +118,7 @@ func run(ctx context.Context, logger *slog.Logger, addr string) error {
srv := &http.Server{
Addr: addr,
- Handler: newHandler(gitHandler, ui),
+ Handler: newHandler(gitHandler, ui, reportAPI),
}
errCh := make(chan error, 1)
@@ -149,7 +149,7 @@ func run(ctx context.Context, logger *slog.Logger, addr string) error {
// management UI under /ui/ and /login (more specific patterns that win over the
// git catch-all). gitHandler, when non-nil, serves the git smart-HTTP endpoints
// at "/"; the exact health/version routes stay more specific.
-func newHandler(gitHandler http.Handler, ui *webui.Handler) http.Handler {
+func newHandler(gitHandler http.Handler, ui *webui.Handler, reportAPI *ci.ReportAPI) http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
fmt.Fprintln(w, "ok")
@@ -160,6 +160,9 @@ func newHandler(gitHandler http.Handler, ui *webui.Handler) http.Handler {
if ui != nil {
ui.Register(mux)
}
+ if reportAPI != nil {
+ reportAPI.Register(mux)
+ }
if gitHandler != nil {
mux.Handle("/", gitHandler)
}
@@ -179,17 +182,17 @@ func newHandler(gitHandler http.Handler, ui *webui.Handler) http.Handler {
// repos cache under GITMOTE_CACHE (default a temp dir). The push path listens on
// GITMOTE_SOCK (default a temp path) and installs the pre-receive hook binary
// at GITMOTE_HOOK (default gitmote-hook alongside this binary).
-func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *webui.Handler, func() error, error) {
+func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *webui.Handler, *ci.ReportAPI, func() error, error) {
noop := func() error { return nil }
if os.Getenv("GITMOTE_S3_BUCKET") == "" {
logger.Warn("GITMOTE_S3_BUCKET unset; git endpoints disabled (health/version only)")
- return nil, nil, noop, nil
+ return nil, nil, nil, noop, nil
}
objs, err := store.NewS3FromEnv(ctx)
if err != nil {
- return nil, nil, noop, fmt.Errorf("object store: %w", err)
+ return nil, nil, nil, noop, fmt.Errorf("object store: %w", err)
}
// RoleAuto: this instance becomes the writer when it can acquire the lease,
@@ -199,7 +202,7 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
// unchanged.
md, err := meta.Open(ctx, metaConfigFromEnv(logger, s3lite.RoleAuto))
if err != nil {
- return nil, nil, noop, fmt.Errorf("metadata: %w", err)
+ return nil, nil, nil, noop, fmt.Errorf("metadata: %w", err)
}
md.OnPromote(func() {
logger.Info("became writer: acquired the lease", "generation", md.Generation())
@@ -212,7 +215,7 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
writer, err := githttp.NewWriter(md, objs, hookBinaryPath(), sockPath, logger)
if err != nil {
_ = md.Close()
- return nil, nil, noop, fmt.Errorf("write path: %w", err)
+ return nil, nil, nil, noop, fmt.Errorf("write path: %w", err)
}
cacheRoot := envOr("GITMOTE_CACHE", filepath.Join(os.TempDir(), "gitmote-repos"))
materializer := repo.New(md, objs, cacheRoot)
@@ -226,7 +229,7 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
if jobDefID != "" && (workerSecret == "" || gitmoteURL == "") {
_ = writer.Close()
_ = md.Close()
- return nil, nil, noop, fmt.Errorf("SCW_CI_JOB_DEFINITION_ID is set but WORKER_SECRET or GITMOTE_URL is missing")
+ return nil, nil, nil, noop, fmt.Errorf("SCW_CI_JOB_DEFINITION_ID is set but WORKER_SECRET or GITMOTE_URL is missing")
}
jobs := scaleway.NewJobsClient(os.Getenv("SCW_SECRET_KEY"), envOr("SCW_REGION", os.Getenv("AWS_REGION")), jobDefID)
// A successful, branch-advancing push discovers its workflows, records a run,
@@ -240,6 +243,10 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
WorkerSecret: workerSecret,
Logger: logger,
})
+ // The runner's authenticated claim/complete API. Only the leader may write
+ // completions (a follower returns a retryable 503), and it authenticates with
+ // the same WORKER_SECRET injected into the runner env.
+ reportAPI := ci.NewReportAPI(md, objs, md.IsLeader, workerSecret, logger)
writer.AfterCommit = func(ctx context.Context, commits []githttp.CommitInfo) {
for _, c := range commits {
dispatcher.Dispatch(ctx, ci.Event{
@@ -275,15 +282,15 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
})
if err != nil {
_ = cleanup()
- return nil, nil, noop, err
+ return nil, nil, nil, noop, err
}
ui, err := buildUI(md, materializer, guard, logger)
if err != nil {
_ = cleanup()
- return nil, nil, noop, err
+ return nil, nil, nil, noop, err
}
- return handler, ui, cleanup, nil
+ return handler, ui, reportAPI, cleanup, nil
}
// buildUI constructs the management UI when GITMOTE_COOKIE_KEY is set; otherwise
diff --git a/cmd/gitmote/main_test.go b/cmd/gitmote/main_test.go
index 141b7d8..d48d4e3 100644
--- a/cmd/gitmote/main_test.go
+++ b/cmd/gitmote/main_test.go
@@ -13,7 +13,7 @@ import (
)
func TestHealthz(t *testing.T) {
- srv := httptest.NewServer(newHandler(nil, nil))
+ srv := httptest.NewServer(newHandler(nil, nil, nil))
defer srv.Close()
resp, err := http.Get(srv.URL + "/healthz")
@@ -28,7 +28,7 @@ func TestHealthz(t *testing.T) {
}
func TestVersion(t *testing.T) {
- srv := httptest.NewServer(newHandler(nil, nil))
+ srv := httptest.NewServer(newHandler(nil, nil, nil))
defer srv.Close()
resp, err := http.Get(srv.URL + "/version")
@@ -43,7 +43,7 @@ func TestVersion(t *testing.T) {
}
func TestUnknownRouteNotFound(t *testing.T) {
- srv := httptest.NewServer(newHandler(nil, nil))
+ srv := httptest.NewServer(newHandler(nil, nil, nil))
defer srv.Close()
resp, err := http.Get(srv.URL + "/nope")
@@ -63,7 +63,7 @@ func TestGitHandlerMountedAtRoot(t *testing.T) {
gitHandler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusTeapot)
})
- srv := httptest.NewServer(newHandler(gitHandler, nil))
+ srv := httptest.NewServer(newHandler(gitHandler, nil, nil))
defer srv.Close()
health, err := http.Get(srv.URL + "/healthz")
diff --git a/internal/ci/report.go b/internal/ci/report.go
new file mode 100644
index 0000000..9b66e60
--- /dev/null
+++ b/internal/ci/report.go
@@ -0,0 +1,302 @@
+package ci
+
+import (
+ "bytes"
+ "context"
+ "crypto/subtle"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net/http"
+ "strconv"
+ "time"
+
+ "github.com/atmin/gitmote/internal/meta"
+ "github.com/atmin/gitmote/internal/store"
+)
+
+const (
+ // logCap bounds a stored job log so a runaway workflow can't fill the store
+ // (epic §Stage 0 #5). Over the cap, the log is truncated with an explicit
+ // marker — never silently.
+ logCap = 10 << 20 // 10 MiB
+ // logTruncationMarker is appended when a log is truncated at the cap.
+ logTruncationMarker = "\n\n[log truncated: exceeded 10 MiB cap]\n"
+)
+
+// ReportMeta is the slice of meta the report API reads and writes. *meta.Metadata
+// satisfies it.
+type ReportMeta interface {
+ GetJob(ctx context.Context, jobID int64) (*meta.Job, error)
+ ClaimJob(ctx context.Context, jobID int64) (*meta.Job, error)
+ SetJobResult(ctx context.Context, jobID int64, status meta.RunStatus, logKey string) error
+ ListJobs(ctx context.Context, runID int64) ([]meta.Job, error)
+ GetRun(ctx context.Context, runID int64) (*meta.Run, error)
+ SetRunStatus(ctx context.Context, runID int64, status meta.RunStatus) error
+ GetRepoByID(ctx context.Context, id int64) (*meta.Repo, error)
+ SweepStuckJobs(ctx context.Context, cutoff time.Time) ([]int64, error)
+}
+
+// ReportAPI serves the runner's authenticated internal endpoints: claim a job
+// and report its completion. Runners never touch s3lite or S3 directly — the
+// parent (leader) writes the log blob and the status rows, matching the push-hook
+// discipline (tasks/16-ci.md). Auth is a constant-time WORKER_SECRET compare;
+// only the leader may write, so a follower returns a retryable 503.
+type ReportAPI struct {
+ meta ReportMeta
+ store store.Store
+ isLeader func() bool
+ secret string
+ logger *slog.Logger
+}
+
+// NewReportAPI returns a report API. isLeader gates writes (nil means always
+// leader, for tests/local). An empty secret rejects every request.
+func NewReportAPI(m ReportMeta, s store.Store, isLeader func() bool, secret string, logger *slog.Logger) *ReportAPI {
+ if isLeader == nil {
+ isLeader = func() bool { return true }
+ }
+ if logger == nil {
+ logger = slog.New(slog.DiscardHandler)
+ }
+ return &ReportAPI{meta: m, store: s, isLeader: isLeader, secret: secret, logger: logger}
+}
+
+// Register mounts the internal CI routes on mux (not under /ui — these are
+// machine-to-machine, WORKER_SECRET-authenticated).
+func (a *ReportAPI) Register(mux *http.ServeMux) {
+ mux.Handle("GET /internal/ci/jobs/{id}", http.HandlerFunc(a.handleClaim))
+ mux.Handle("POST /internal/ci/jobs/{id}/complete", http.HandlerFunc(a.handleComplete))
+}
+
+// jobSpec is what a claim returns: everything the runner needs to check out and
+// run the job.
+type jobSpec struct {
+ JobID int64 `json:"job_id"`
+ RunID int64 `json:"run_id"`
+ Repo string `json:"repo"`
+ SHA string `json:"sha"`
+ Ref string `json:"ref"`
+ WorkflowDir string `json:"workflow_dir"`
+}
+
+// handleClaim atomically claims a queued job (queued→running) and returns its
+// spec. A job that is not claimable (absent, already claimed, terminal) is 404.
+func (a *ReportAPI) handleClaim(w http.ResponseWriter, r *http.Request) {
+ if !a.authed(w, r) {
+ return
+ }
+ id, ok := jobID(w, r)
+ if !ok {
+ return
+ }
+
+ job, err := a.meta.ClaimJob(r.Context(), id)
+ if errors.Is(err, meta.ErrNotFound) {
+ http.Error(w, "job not found or not claimable", http.StatusNotFound)
+ return
+ }
+ if err != nil {
+ a.logger.Error("ci: claim failed", "job", id, "error", err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+
+ run, err := a.meta.GetRun(r.Context(), job.RunID)
+ if err != nil {
+ a.logger.Error("ci: claim run lookup failed", "run", job.RunID, "error", err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+ repo, err := a.meta.GetRepoByID(r.Context(), run.RepoID)
+ if err != nil {
+ a.logger.Error("ci: claim repo lookup failed", "repo_id", run.RepoID, "error", err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+
+ writeJSON(w, jobSpec{
+ JobID: job.ID,
+ RunID: run.ID,
+ Repo: repo.Name,
+ SHA: run.SHA,
+ Ref: run.Ref,
+ WorkflowDir: workflowDir,
+ })
+}
+
+// handleComplete records a job's outcome. Status and error travel as query
+// params; the request body is the combined log. Content-before-pointer: the log
+// blob is stored first, then the job status + log_key, then the run rolls up.
+// Idempotent: completing an already-terminal job is a no-op.
+func (a *ReportAPI) handleComplete(w http.ResponseWriter, r *http.Request) {
+ if !a.authed(w, r) {
+ return
+ }
+ id, ok := jobID(w, r)
+ if !ok {
+ return
+ }
+ status, ok := completionStatus(w, r)
+ if !ok {
+ return
+ }
+
+ job, err := a.meta.GetJob(r.Context(), id)
+ if errors.Is(err, meta.ErrNotFound) {
+ http.Error(w, "job not found", http.StatusNotFound)
+ return
+ }
+ if err != nil {
+ a.logger.Error("ci: complete job lookup failed", "job", id, "error", err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+ if isTerminal(job.Status) {
+ writeJSON(w, map[string]bool{"ok": true}) // idempotent no-op
+ return
+ }
+
+ run, err := a.meta.GetRun(r.Context(), job.RunID)
+ if err != nil {
+ a.logger.Error("ci: complete run lookup failed", "run", job.RunID, "error", err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+
+ logKey := fmt.Sprintf("ci/%d/%d/%d.log", run.RepoID, run.ID, job.ID)
+ if err := a.storeLog(r.Context(), logKey, r.Body); err != nil {
+ a.logger.Error("ci: store log failed", "job", id, "key", logKey, "error", err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+ // Pointer after content: the blob is durable before the row references it.
+ if err := a.meta.SetJobResult(r.Context(), job.ID, status, logKey); err != nil {
+ a.logger.Error("ci: set job result failed", "job", id, "error", err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+ a.rollupRun(r.Context(), run.ID)
+
+ writeJSON(w, map[string]bool{"ok": true})
+}
+
+// storeLog writes the body under key, enforcing the size cap with an explicit
+// truncation marker rather than a silent cut.
+func (a *ReportAPI) storeLog(ctx context.Context, key string, body io.Reader) error {
+ data, err := io.ReadAll(io.LimitReader(body, logCap+1))
+ if err != nil {
+ return err
+ }
+ if len(data) > logCap {
+ data = append(data[:logCap], []byte(logTruncationMarker)...)
+ }
+ return a.store.Put(ctx, key, bytes.NewReader(data))
+}
+
+// rollupRun recomputes a run's status from its jobs: still running while any job
+// is non-terminal, else error > failed > passed by worst outcome. Logged, not
+// returned — a rollup failure must not fail an already-recorded completion.
+func (a *ReportAPI) rollupRun(ctx context.Context, runID int64) {
+ jobs, err := a.meta.ListJobs(ctx, runID)
+ if err != nil {
+ a.logger.Error("ci: rollup list jobs failed", "run", runID, "error", err)
+ return
+ }
+ status := rollup(jobs)
+ if err := a.meta.SetRunStatus(ctx, runID, status); err != nil {
+ a.logger.Error("ci: rollup set run status failed", "run", runID, "error", err)
+ }
+}
+
+// ReconcileStuck marks running jobs older than maxAge as error and rolls up their
+// runs. The leader calls it periodically so a runner that dies without reporting
+// doesn't leave a job running forever.
+func (a *ReportAPI) ReconcileStuck(ctx context.Context, maxAge time.Duration, now time.Time) error {
+ stuck, err := a.meta.SweepStuckJobs(ctx, now.Add(-maxAge))
+ if err != nil {
+ return err
+ }
+ for _, id := range stuck {
+ job, err := a.meta.GetJob(ctx, id)
+ if err != nil {
+ a.logger.Error("ci: reconcile job lookup failed", "job", id, "error", err)
+ continue
+ }
+ a.rollupRun(ctx, job.RunID)
+ }
+ return nil
+}
+
+// authed enforces the constant-time WORKER_SECRET check and the leader gate,
+// writing the response and returning false on failure.
+func (a *ReportAPI) authed(w http.ResponseWriter, r *http.Request) bool {
+ if a.secret == "" || subtle.ConstantTimeCompare([]byte(r.Header.Get("X-Worker-Secret")), []byte(a.secret)) != 1 {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return false
+ }
+ if !a.isLeader() {
+ // A follower can't write s3lite; the runner should retry against the leader.
+ http.Error(w, "not the writer; retry", http.StatusServiceUnavailable)
+ return false
+ }
+ return true
+}
+
+// jobID parses the {id} path value, writing a 400 on failure.
+func jobID(w http.ResponseWriter, r *http.Request) (int64, bool) {
+ id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
+ if err != nil {
+ http.Error(w, "invalid job id", http.StatusBadRequest)
+ return 0, false
+ }
+ return id, true
+}
+
+// completionStatus reads and validates the terminal status query param.
+func completionStatus(w http.ResponseWriter, r *http.Request) (meta.RunStatus, bool) {
+ status := meta.RunStatus(r.URL.Query().Get("status"))
+ switch status {
+ case meta.RunPassed, meta.RunFailed, meta.RunError:
+ return status, true
+ default:
+ http.Error(w, "status must be passed, failed, or error", http.StatusBadRequest)
+ return "", false
+ }
+}
+
+// rollup derives a run's status from its jobs.
+func rollup(jobs []meta.Job) meta.RunStatus {
+ worst := meta.RunPassed
+ for _, j := range jobs {
+ if !isTerminal(j.Status) {
+ return meta.RunRunning
+ }
+ switch j.Status {
+ case meta.RunError:
+ worst = meta.RunError
+ case meta.RunFailed:
+ if worst != meta.RunError {
+ worst = meta.RunFailed
+ }
+ }
+ }
+ return worst
+}
+
+// isTerminal reports whether a status is a finished state.
+func isTerminal(s meta.RunStatus) bool {
+ switch s {
+ case meta.RunPassed, meta.RunFailed, meta.RunError, meta.RunSuperseded:
+ return true
+ default:
+ return false
+ }
+}
+
+func writeJSON(w http.ResponseWriter, v any) {
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(v)
+}
diff --git a/internal/ci/report_test.go b/internal/ci/report_test.go
new file mode 100644
index 0000000..b2349cb
--- /dev/null
+++ b/internal/ci/report_test.go
@@ -0,0 +1,260 @@
+package ci
+
+import (
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/atmin/gitmote/internal/meta"
+ "github.com/atmin/gitmote/internal/store"
+)
+
+const workerSecret = "worker-secret"
+
+// reportFixture wires a report API over a real meta + in-memory store, seeds a
+// repo/run/job, and returns the API's test server, the store, and the ids.
+type reportFixture struct {
+ srv *httptest.Server
+ store store.Store
+ md *meta.Metadata
+ api *ReportAPI
+ repoID int64
+ runID int64
+ jobID int64
+}
+
+func newReportFixture(t *testing.T, isLeader func() bool) reportFixture {
+ t.Helper()
+ ctx := context.Background()
+ md, err := meta.Open(ctx, meta.Config{LocalPath: filepath.Join(t.TempDir(), "meta.sqlite3")})
+ if err != nil {
+ t.Fatalf("meta.Open: %v", err)
+ }
+ t.Cleanup(func() { _ = md.Close() })
+ s := store.NewMem()
+
+ r, err := md.CreateRepo(ctx, "atmin/app", "main")
+ if err != nil {
+ t.Fatalf("CreateRepo: %v", err)
+ }
+ run, err := md.CreateRun(ctx, r.ID, "refs/heads/main", "deadbeef")
+ if err != nil {
+ t.Fatalf("CreateRun: %v", err)
+ }
+ job, err := md.CreateJob(ctx, run.ID, "ci.yml")
+ if err != nil {
+ t.Fatalf("CreateJob: %v", err)
+ }
+
+ api := NewReportAPI(md, s, isLeader, workerSecret, nil)
+ mux := http.NewServeMux()
+ api.Register(mux)
+ srv := httptest.NewServer(mux)
+ t.Cleanup(srv.Close)
+
+ return reportFixture{srv: srv, store: s, md: md, api: api, repoID: r.ID, runID: run.ID, jobID: job.ID}
+}
+
+// do issues an authenticated request unless secret is "" (then it sends none).
+func (f reportFixture) do(t *testing.T, method, path, secret, body string) *http.Response {
+ t.Helper()
+ req, err := http.NewRequest(method, f.srv.URL+path, strings.NewReader(body))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if secret != "" {
+ req.Header.Set("X-Worker-Secret", secret)
+ }
+ resp, err := f.srv.Client().Do(req)
+ if err != nil {
+ t.Fatalf("%s %s: %v", method, path, err)
+ }
+ return resp
+}
+
+func TestClaimTransitionsAndReturnsSpec(t *testing.T) {
+ ctx := context.Background()
+ f := newReportFixture(t, nil)
+
+ resp := f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "")
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("claim status = %d, want 200", resp.StatusCode)
+ }
+ var spec jobSpec
+ if err := json.NewDecoder(resp.Body).Decode(&spec); err != nil {
+ t.Fatalf("decode spec: %v", err)
+ }
+ if spec.JobID != f.jobID || spec.RunID != f.runID || spec.Repo != "atmin/app" ||
+ spec.SHA != "deadbeef" || spec.Ref != "refs/heads/main" || spec.WorkflowDir != workflowDir {
+ t.Errorf("spec = %+v, want the seeded job's coordinates", spec)
+ }
+
+ // The job is now running.
+ job, _ := f.md.GetJob(ctx, f.jobID)
+ if job.Status != meta.RunRunning {
+ t.Errorf("job status = %q, want running", job.Status)
+ }
+
+ // A second claim of the now-running job is not claimable → 404.
+ resp2 := f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "")
+ defer resp2.Body.Close()
+ if resp2.StatusCode != http.StatusNotFound {
+ t.Errorf("re-claim status = %d, want 404", resp2.StatusCode)
+ }
+}
+
+func TestCompleteStoresLogAndRollsUp(t *testing.T) {
+ ctx := context.Background()
+ f := newReportFixture(t, nil)
+ // Claim first, then complete.
+ f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "").Body.Close()
+
+ resp := f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/complete?status=passed", workerSecret, "build output\n")
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("complete status = %d, want 200", resp.StatusCode)
+ }
+
+ // Job is passed, log_key recorded, and the log blob is under the ci/ key.
+ job, _ := f.md.GetJob(ctx, f.jobID)
+ if job.Status != meta.RunPassed {
+ t.Errorf("job status = %q, want passed", job.Status)
+ }
+ wantKey := "ci/" + id(f.repoID) + "/" + id(f.runID) + "/" + id(f.jobID) + ".log"
+ if job.LogKey != wantKey {
+ t.Errorf("log_key = %q, want %q", job.LogKey, wantKey)
+ }
+ rc, err := f.store.Get(ctx, wantKey)
+ if err != nil {
+ t.Fatalf("store.Get(%q): %v", wantKey, err)
+ }
+ data, _ := io.ReadAll(rc)
+ rc.Close()
+ if string(data) != "build output\n" {
+ t.Errorf("stored log = %q, want %q", data, "build output\n")
+ }
+
+ // The single job passed, so the run rolls up to passed.
+ run, _ := f.md.GetRun(ctx, f.runID)
+ if run.Status != meta.RunPassed {
+ t.Errorf("run status = %q, want passed", run.Status)
+ }
+}
+
+func TestCompleteIsIdempotent(t *testing.T) {
+ ctx := context.Background()
+ f := newReportFixture(t, nil)
+ f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "").Body.Close()
+ f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/complete?status=passed", workerSecret, "first\n").Body.Close()
+
+ // A second complete with a different status/log is a no-op — the terminal
+ // job keeps its first outcome and log.
+ resp := f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/complete?status=failed", workerSecret, "second\n")
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("second complete status = %d, want 200", resp.StatusCode)
+ }
+ job, _ := f.md.GetJob(ctx, f.jobID)
+ if job.Status != meta.RunPassed {
+ t.Errorf("job status = %q, want passed (idempotent)", job.Status)
+ }
+ rc, _ := f.store.Get(ctx, job.LogKey)
+ data, _ := io.ReadAll(rc)
+ rc.Close()
+ if string(data) != "first\n" {
+ t.Errorf("log = %q, want the first log unchanged", data)
+ }
+}
+
+func TestReportAPIAuth(t *testing.T) {
+ f := newReportFixture(t, nil)
+ path := "/internal/ci/jobs/" + id(f.jobID)
+
+ // Missing secret → 401.
+ resp := f.do(t, http.MethodGet, path, "", "")
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusUnauthorized {
+ t.Errorf("no secret: status = %d, want 401", resp.StatusCode)
+ }
+ // Wrong secret → 401.
+ resp = f.do(t, http.MethodGet, path, "nope", "")
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusUnauthorized {
+ t.Errorf("wrong secret: status = %d, want 401", resp.StatusCode)
+ }
+}
+
+func TestReportAPIFollowerReturns503(t *testing.T) {
+ f := newReportFixture(t, func() bool { return false }) // a follower
+ resp := f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "")
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusServiceUnavailable {
+ t.Errorf("follower claim: status = %d, want 503", resp.StatusCode)
+ }
+}
+
+func TestCompleteTruncatesOversizeLog(t *testing.T) {
+ ctx := context.Background()
+ f := newReportFixture(t, nil)
+ f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "").Body.Close()
+
+ big := strings.Repeat("x", logCap+1024) // over the cap
+ resp := f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/complete?status=passed", workerSecret, big)
+ resp.Body.Close()
+
+ job, _ := f.md.GetJob(ctx, f.jobID)
+ rc, err := f.store.Get(ctx, job.LogKey)
+ if err != nil {
+ t.Fatalf("store.Get: %v", err)
+ }
+ data, _ := io.ReadAll(rc)
+ rc.Close()
+ if !strings.HasSuffix(string(data), logTruncationMarker) {
+ t.Error("oversize log missing the truncation marker (silent truncation)")
+ }
+ // Body cut to the cap, plus the marker — not the full oversize input.
+ if len(data) != logCap+len(logTruncationMarker) {
+ t.Errorf("stored len = %d, want cap+marker = %d", len(data), logCap+len(logTruncationMarker))
+ }
+}
+
+func TestCompleteRejectsBadStatus(t *testing.T) {
+ f := newReportFixture(t, nil)
+ f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "").Body.Close()
+ resp := f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/complete?status=bogus", workerSecret, "x")
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusBadRequest {
+ t.Errorf("bad status: code = %d, want 400", resp.StatusCode)
+ }
+}
+
+func TestReconcileStuckSweepsRunningJobs(t *testing.T) {
+ ctx := context.Background()
+ f := newReportFixture(t, nil)
+ // Claim to put the job in running.
+ f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "").Body.Close()
+
+ // Sweep with a clock far in the future so the running job is past the bound.
+ future := time.Now().Add(48 * time.Hour)
+ if err := f.api.ReconcileStuck(ctx, time.Hour, future); err != nil {
+ t.Fatalf("ReconcileStuck: %v", err)
+ }
+ job, _ := f.md.GetJob(ctx, f.jobID)
+ if job.Status != meta.RunError {
+ t.Errorf("stuck job status = %q, want error", job.Status)
+ }
+ run, _ := f.md.GetRun(ctx, f.runID)
+ if run.Status != meta.RunError {
+ t.Errorf("run status after sweep = %q, want error", run.Status)
+ }
+}
+
+func id(n int64) string { return strconv.FormatInt(n, 10) }
diff --git a/internal/meta/ci.go b/internal/meta/ci.go
index 7fbe543..ed1a9b9 100644
--- a/internal/meta/ci.go
+++ b/internal/meta/ci.go
@@ -189,6 +189,110 @@ func (m *Metadata) SetJobStatus(ctx context.Context, jobID int64, status RunStat
return nil
}
+// GetJob returns the job with the given id, or ErrNotFound.
+func (m *Metadata) GetJob(ctx context.Context, jobID int64) (*Job, error) {
+ var (
+ j Job
+ status string
+ logKey sql.NullString
+ cts string
+ uts string
+ )
+ err := m.db.QueryRowContext(ctx,
+ `SELECT id, run_id, name, status, log_key, created_at, updated_at
+ FROM ci_jobs WHERE id = ?`, jobID).
+ Scan(&j.ID, &j.RunID, &j.Name, &status, &logKey, &cts, &uts)
+ if isNoRows(err) {
+ return nil, ErrNotFound
+ }
+ if err != nil {
+ return nil, err
+ }
+ j.Status = RunStatus(status)
+ j.LogKey = logKey.String
+ j.CreatedAt = parseTime(cts)
+ j.UpdatedAt = parseTime(uts)
+ return &j, nil
+}
+
+// ClaimJob atomically transitions a queued job to running and returns it. It
+// returns ErrNotFound when no job with the id is queued — already claimed,
+// terminal, or absent — so a concurrent double-claim yields exactly one winner.
+func (m *Metadata) ClaimJob(ctx context.Context, jobID int64) (*Job, error) {
+ res, err := m.db.ExecContext(ctx,
+ `UPDATE ci_jobs SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
+ string(RunRunning), now(), jobID, string(RunQueued))
+ if err != nil {
+ return nil, err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return nil, err
+ }
+ if n == 0 {
+ return nil, ErrNotFound
+ }
+ return m.GetJob(ctx, jobID)
+}
+
+// SetJobResult records a job's terminal status and its log key together. It
+// returns ErrNotFound when no job has the given id.
+func (m *Metadata) SetJobResult(ctx context.Context, jobID int64, status RunStatus, logKey string) error {
+ res, err := m.db.ExecContext(ctx,
+ `UPDATE ci_jobs SET status = ?, log_key = ?, updated_at = ? WHERE id = ?`,
+ string(status), logKey, now(), jobID)
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n == 0 {
+ return ErrNotFound
+ }
+ return nil
+}
+
+// SweepStuckJobs marks every running job last updated before cutoff as error and
+// returns the ids it changed, so the caller can roll up their runs. A runner that
+// dies without completing would otherwise leave a job running forever.
+func (m *Metadata) SweepStuckJobs(ctx context.Context, cutoff time.Time) ([]int64, error) {
+ rows, err := m.db.QueryContext(ctx,
+ `SELECT id, updated_at FROM ci_jobs WHERE status = ?`, string(RunRunning))
+ if err != nil {
+ return nil, err
+ }
+ var stuck []int64
+ for rows.Next() {
+ var (
+ id int64
+ uts string
+ )
+ if err := rows.Scan(&id, &uts); err != nil {
+ _ = rows.Close()
+ return nil, err
+ }
+ if parseTime(uts).Before(cutoff) {
+ stuck = append(stuck, id)
+ }
+ }
+ if err := rows.Err(); err != nil {
+ _ = rows.Close()
+ return nil, err
+ }
+ _ = rows.Close()
+
+ for _, id := range stuck {
+ if _, err := m.db.ExecContext(ctx,
+ `UPDATE ci_jobs SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
+ string(RunError), now(), id, string(RunRunning)); err != nil {
+ return nil, err
+ }
+ }
+ return stuck, nil
+}
+
// ListJobs returns a run's jobs ordered by creation (id).
func (m *Metadata) ListJobs(ctx context.Context, runID int64) ([]Job, error) {
rows, err := m.db.QueryContext(ctx,
diff --git a/internal/meta/repos.go b/internal/meta/repos.go
index bff4de9..236cc73 100644
--- a/internal/meta/repos.go
+++ b/internal/meta/repos.go
@@ -70,6 +70,25 @@ func (m *Metadata) GetRepo(ctx context.Context, name string) (*Repo, error) {
return &r, nil
}
+// GetRepoByID returns the repository with the given id, or ErrNotFound.
+func (m *Metadata) GetRepoByID(ctx context.Context, id int64) (*Repo, error) {
+ var (
+ r Repo
+ ts string
+ )
+ err := m.db.QueryRowContext(ctx,
+ `SELECT id, name, default_branch, created_at FROM repos WHERE id = ?`, id).
+ Scan(&r.ID, &r.Name, &r.DefaultBranch, &ts)
+ if isNoRows(err) {
+ return nil, ErrNotFound
+ }
+ if err != nil {
+ return nil, err
+ }
+ r.CreatedAt = parseTime(ts)
+ return &r, nil
+}
+
// ListRepos returns all repositories ordered by name.
func (m *Metadata) ListRepos(ctx context.Context) ([]Repo, error) {
rows, err := m.db.QueryContext(ctx,
diff --git a/tasks/21-ci-runner.md b/tasks/21-ci-runner.md
index 49c4be1..fbcf230 100644
--- a/tasks/21-ci-runner.md
+++ b/tasks/21-ci-runner.md
@@ -1,90 +1,72 @@
-# CI stage 4 — runner image + internal report API
+# CI stage 4b — runner image + engine execution
-Part of the CI epic ([16-ci.md](16-ci.md)). Depends on **stages 1–3
-([17](17-ci-data-model.md)–[19](19-ci-scaleway-trigger.md))** (runs/jobs +
-trigger with env) and the **scoped-token precursor ([20](20-ci-scoped-tokens.md))**
-(per-run clone credential). This is the stage that actually **executes** a
-workflow. It is the largest stage and may split into 4a (internal API) and 4b
-(runner image) during implementation.
+Part of the CI epic ([16-ci.md](16-ci.md)). **Stage 4a (the internal report API)
+has landed**: gitmote exposes the authenticated claim/complete endpoints, the log
+store, run rollup, and the stuck-job reconciliation sweep. What remains is the
+**runner** that actually executes a workflow and drives those endpoints.
-> **Infra unknown to resolve first (blocks 4b, not 4a):** `act` runs each job in
-> a Docker container, so the runner needs a Docker daemon inside the Scaleway
-> Serverless Job. Confirm Jobs expose a usable daemon (privileged / DinD) or pick
-> the fallback: run steps in a single provided container without full `act`
-> nesting. Spike this before building the image; the internal API (4a) does not
-> depend on it.
+> **Infra spike to resolve first:** `act` runs each job in a Docker container, so
+> the runner needs a Docker daemon inside the Scaleway Serverless Job. Confirm
+> Jobs expose a usable daemon (privileged / DinD) or pick the fallback: run steps
+> in a single provided container without full `act` nesting. Spike this before
+> building the image.
-## Spec
+## What 4a already provides (the contract to build against)
-A runner process, launched as a Scaleway Serverless Job with the stage-3 env,
-**claims** its job from gitmote, **clones** the repo at the SHA using the per-run
-read-only scoped token, **runs `act`** over `.github/workflows`, streams the
-combined log back, and **reports completion**. gitmote exposes an authenticated
-internal API for claim + completion, secured by a constant-time `WORKER_SECRET`
-compare. **Runners never touch s3lite or S3 directly** — logs and status flow
-through the API (the parent writes the `ci/` log blob), matching the push-hook
-discipline.
-
-## Current
-
-- The dispatcher triggers a job with `GITMOTE_CI_JOB_ID`, `GITMOTE_URL`,
- `WORKER_SECRET`, `GITMOTE_REPO/SHA/REF`, `GITMOTE_CI_CLONE_TOKEN` (stages 3+20).
-- The **template** is pan0's `cmd/worker/main.go` +
- `server/handle_internal_jobs.go`: `GET /internal/jobs/{id}` (claim, header
- `X-Worker-Secret`) → run → `POST /internal/jobs/{id}/complete` with
- `{status,error,outputs}`, and an **idempotent** `ApplyCompletion` (no-op if the
- job is already terminal). Steal the shape.
-- gitmote already speaks git-HTTPS; a scoped token (stage 20) is a valid clone
- credential on the normal auth path — no special-casing in the git handler.
-- Log layout + caps are locked (epic §Stage 0 #5): `ci/{repoID}/{runID}/{jobID}.log`,
- 10 MiB cap with an explicit truncation marker, 30-day retention.
+- `GET /internal/ci/jobs/{id}` — constant-time `X-Worker-Secret` auth; atomically
+ claims (`queued`→`running`) and returns the spec
+ `{job_id, run_id, repo, sha, ref, workflow_dir}`. Not claimable → 404; a
+ follower → retryable 503. (`internal/ci/report.go`.)
+- `POST /internal/ci/jobs/{id}/complete?status=passed|failed|error` — the request
+ **body is the combined log**. Content-before-pointer: the blob is stored under
+ `ci/{repoID}/{runID}/{jobID}.log` (10 MiB cap, explicit truncation marker),
+ then `SetJobResult` records status + `log_key`, then the run rolls up
+ (running > error > failed > passed). Idempotent: a terminal job is a no-op.
+- Per-run clone credential: `auth.Guard.MintScoped` (task 20) mints a read-only,
+ repo-scoped, expiring token; the dispatcher already injects
+ `GITMOTE_CI_CLONE_TOKEN` (currently a placeholder — **wire the real mint** here).
+- `ReportAPI.ReconcileStuck(ctx, maxAge, now)` exists and is tested; **wire it to
+ a leader-only periodic ticker** in `main.go` once runners can produce
+ `running` jobs (nothing does until this stage).
## Change
-**4a. Internal report API** (new `internal/ci` handlers, mounted on the main mux,
-**not** under `/ui`):
-- `GET /internal/ci/jobs/{id}` — constant-time `X-Worker-Secret`
- (`crypto/subtle`); atomically claim (`queued`→`running`); return the job spec
- `{repo, sha, ref, workflowDir}`.
-- `POST /internal/ci/jobs/{id}/complete` — `{status, error}` + the log as the
- body (or a multipart field). **Content-before-pointer:** enforce the 10 MiB cap
- (truncate-with-marker), `store.Put` the log under the `ci/` key, **then**
- `SetJobStatus` + `log_key` and roll the run status up. Idempotent: a second
- complete for a terminal job is a no-op.
-- `WORKER_SECRET` from env; reject missing/wrong with 401. Only the **leader**
- can write completions (a follower can't write s3lite) — a follower returns a
- retryable 503, same as receive-pack.
+**Wire the real clone token.** In the dispatcher (`internal/ci/dispatcher.go`),
+replace the `GITMOTE_CI_CLONE_TOKEN` placeholder with a `MintScoped` call
+(read-only, scoped to the pushed repo, TTL = run max-duration + margin). Revoke on
+completion (or lean on expiry). Needs the dispatcher to reach the guard/mint and
+a CI service user to own the token.
-**4b. Runner** (`cmd/gitmote-runner` + `Dockerfile.runner`):
-- Read env; `GET` claim; `git clone https://x-access-token:$TOKEN@$GITMOTE_URL/$REPO`
- and checkout `$SHA` (shallow where possible).
+**Runner** (`cmd/gitmote-runner` + `Dockerfile.runner`):
+- Read env (`GITMOTE_URL`, `GITMOTE_CI_JOB_ID`, `WORKER_SECRET`,
+ `GITMOTE_REPO/SHA/REF`, `GITMOTE_CI_CLONE_TOKEN`); `GET` the claim.
+- `git clone https://x-access-token:$TOKEN@$GITMOTE_URL/$REPO`, checkout `$SHA`
+ (shallow where possible).
- Run `act` over `.github/workflows`, capturing combined stdout/stderr.
-- `POST …/complete` with pass/fail (from `act`'s exit) + the captured log. Exit
- non-zero on internal failure so a crashed runner is distinguishable from a
- failed build.
-- `Dockerfile.runner`: `act` + git + the daemon strategy from the spike above.
- One-time provisioning (`scw jobs definition create name=gitmote-ci-runner
- image=… cpu-limit=… memory-limit=… local-storage-capacity=…`) documented in
+- `POST …/complete?status=…` with the captured log as the body (pass/fail from
+ `act`'s exit). Exit non-zero on internal failure so a crashed runner is
+ distinguishable from a failed build.
+- `Dockerfile.runner`: `act` + git + the daemon strategy from the spike. One-time
+ `scw jobs definition create name=gitmote-ci-runner image=… cpu-limit=…
+ memory-limit=… local-storage-capacity=…` documented in
[ops.md](../docs/ops.md); one definition serves all repos (env injected at
trigger, per stage 3).
-**Reconciliation:** a runner that dies without completing leaves a `running` job
-forever. Add a timeout sweep (the leader marks `running` jobs older than a bound
-`error`); this pairs naturally with the log/run GC in stage 6/retention.
+**Wire the reconciliation ticker.** A leader-only goroutine in `main.go` calling
+`ReportAPI.ReconcileStuck` on an interval (bound ≈ run max-duration), with clean
+shutdown.
## Verify
-- **internal API tests** (httptest, cf. pan0): claim transitions `queued`→
- `running` and returns the spec; **double-complete** is idempotent; missing/wrong
- `X-Worker-Secret` → 401 (constant-time); a follower (`IsLeader()==false`) →
- retryable 503; a log over 10 MiB is stored **truncated with the marker**, never
- silently cut; the log blob lands under `ci/{repoID}/{runID}/{jobID}.log` with
- the recorded `log_key`.
- **runner test** against a stub gitmote server with a **fake engine** (a script
that echoes and exits 0/1): full claim→clone→run→complete flow drives the job
to `passed`/`failed`; the reported log round-trips.
-- **reconciliation:** a `running` job past the bound is swept to `error`.
-- **security:** the runner holds only the scoped token + `WORKER_SECRET`; assert
- it has no S3/DB credentials in its env.
+- **clone-token test:** the minted token is read-only + repo-scoped and clones the
+ target repo but is denied a push and denied another repo (leans on task 20's
+ guard tests).
+- **security:** the runner env holds only the scoped token + `WORKER_SECRET` — no
+ S3/DB credentials.
+- **ticker:** a `running` job past the bound is swept to `error` by the wired
+ reconciler (the sweep itself is already unit-tested in 4a).
- `gofmt`/`golangci-lint`/`go test ./...` clean; `make e2e-local` +
- `make e2e-restore` green (internal CI routes don't touch the git write path).
+ `make e2e-restore` green.
diff --git a/tasks/README.md b/tasks/README.md
index c1a99ff..5710442 100644
--- a/tasks/README.md
+++ b/tasks/README.md
@@ -17,8 +17,8 @@ index [16-ci.md](16-ci.md) (the *why* is in
[../docs/evolution/ci-runner.md](../docs/evolution/ci-runner.md)). Split into
dependency-ordered stages, each implementable from fresh context:
-- [21-ci-runner.md](21-ci-runner.md) — runner image + internal claim/complete API
- (the execution stage; has an infra spike to resolve first).
+- [21-ci-runner.md](21-ci-runner.md) — **4b:** runner image + engine execution
+ (4a, the internal claim/complete API, has landed; has an infra spike first).
- [22-ci-secrets.md](22-ci-secrets.md) — encrypted per-repo secrets store
(AES-256-GCM + HKDF + versioned keys), injected at trigger.
- [23-ci-status-ui.md](23-ci-status-ui.md) — runs list, run detail, log viewer,