From c28d8d0e48f845571dec809c28074d18ca713b71 Mon Sep 17 00:00:00 2001 From: admin Date: Sun, 25 Oct 2026 10:54:47 -0400 Subject: [PATCH 1/3] fix(ci): requeue the newest status when superseding an expired outbox claim --- free/ci/internal/store/outbox_claim.go | 19 +++++++++-- free/ci/internal/store/report_outbox_test.go | 36 ++++++++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/free/ci/internal/store/outbox_claim.go b/free/ci/internal/store/outbox_claim.go index 155200e9c..8b9bc74e4 100644 --- a/free/ci/internal/store/outbox_claim.go +++ b/free/ci/internal/store/outbox_claim.go @@ -34,7 +34,9 @@ const claimTTL = 60 * time.Second // OutboxClaim takes the right to deliver seq so statuses of one key reach the // host in seq order across processes: refused when a newer seq is posted or // posting (seq is marked superseded), told to wait while an older seq is -// mid-post. On ClaimOK the returned token fences release and posted. +// mid-post. Superseding a seq whose claim had expired (its poster died, its +// post may have landed) queues the newest status again. On ClaimOK the +// returned token fences release and posted. func (s *Store) OutboxClaim(ctx context.Context, revision, statusContext string, seq int64) (int, int64, error) { result, token := ClaimOK, int64(0) err := s.write(ctx, func(tx *sql.Tx) error { @@ -46,7 +48,20 @@ func (s *Store) OutboxClaim(ctx context.Context, revision, statusContext string, } if newer > 0 { result = ClaimSuperseded - _, err := tx.ExecContext(ctx, "UPDATE outbox SET state=? WHERE revision=? AND context=? AND seq=? AND state IN (?,?,?)", OutboxSuperseded, revision, statusContext, seq, OutboxReserved, OutboxPending, OutboxPosting) + // an expired claim of seq was a poster that died mid-delivery: its + // post may have landed after the newer status, so the newest one is + // queued again, as OutboxPosted does for a poster that comes back + var expired int + if err := tx.QueryRowContext(ctx, "SELECT count(*) FROM outbox WHERE revision=? AND context=? AND seq=? AND state=?", revision, statusContext, seq, OutboxPosting).Scan(&expired); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, "UPDATE outbox SET state=? WHERE revision=? AND context=? AND seq=? AND state IN (?,?,?)", OutboxSuperseded, revision, statusContext, seq, OutboxReserved, OutboxPending, OutboxPosting); err != nil { + return err + } + if expired == 0 { + return nil + } + _, err := requeueNewest(ctx, tx, revision, statusContext, seq) return err } if err := tx.QueryRowContext(ctx, "SELECT count(*) FROM outbox WHERE revision=? AND context=? AND seq?", revision, statusContext, seq, OutboxPosting, cutoff).Scan(&older); err != nil { diff --git a/free/ci/internal/store/report_outbox_test.go b/free/ci/internal/store/report_outbox_test.go index 3e6fa0f47..69e4ac35b 100644 --- a/free/ci/internal/store/report_outbox_test.go +++ b/free/ci/internal/store/report_outbox_test.go @@ -246,3 +246,39 @@ func TestOutboxReorderedAcknowledgementIsHealed(t *testing.T) { t.Fatalf("the newest status must be pending: %+v", left) } } + +// A poster that died after its HTTP post landed leaves an expired claim on the +// older seq. Superseding it must queue the newest status again, or the host +// stays on the older (possibly green) status. +func TestOutboxSupersedingAnExpiredClaimRequeuesTheNewest(t *testing.T) { + now := time.Now() + s, err := Open(filepath.Join(t.TempDir(), "state.db"), Options{Clock: func() time.Time { return now }}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.Close() }() + ctx := context.Background() + a, _ := s.OutboxReserve(ctx, "rev", "nself-ci") + b, _ := s.OutboxReserve(ctx, "rev", "nself-ci") + for _, seq := range []int64{a, b} { + _ = s.OutboxReady(ctx, "rev", "nself-ci", seq, []byte(`{"seq":`+strconv.FormatInt(seq, 10)+`}`)) + } + if r, _ := claim(s, a); r != ClaimOK { // A claims, posts, then dies: never acknowledged + t.Fatalf("a: %d", r) + } + now = now.Add(claimTTL + time.Second) + r, tb := claim(s, b) + if r != ClaimOK { + t.Fatalf("b after expiry: %d", r) + } + if err = s.OutboxPosted(ctx, "rev", "nself-ci", b, tb); err != nil { + t.Fatal(err) + } + if r, _ = claim(s, a); r != ClaimSuperseded { + t.Fatalf("a behind a posted b: %d", r) + } + left, _ := s.OutboxUnposted(ctx) + if len(left) != 1 || left[0].Seq != b { + t.Fatalf("the newest status must be queued again: %+v", left) + } +} From 98225936ff27c55f71b2a4b418e5200533d63c3a Mon Sep 17 00:00:00 2001 From: admin Date: Sun, 25 Oct 2026 10:56:10 -0400 Subject: [PATCH 2/3] fix(ci): recovery ends a lost attempt that has a cancel intent as cancelled --- free/ci/internal/run/recovery.go | 10 +++- free/ci/internal/run/recovery_cancel_test.go | 56 ++++++++++++++++++++ 2 files changed, 65 insertions(+), 1 deletion(-) create mode 100644 free/ci/internal/run/recovery_cancel_test.go diff --git a/free/ci/internal/run/recovery.go b/free/ci/internal/run/recovery.go index 059ebeff4..80343bbef 100644 --- a/free/ci/internal/run/recovery.go +++ b/free/ci/internal/run/recovery.go @@ -105,8 +105,16 @@ func Recover(ctx context.Context, st *store.Store, current, home string) Recover switch { case err == nil: rep.Retryable++ + case CodeOf(err) == codeStoreRef: // a cancel intent won: the operator's cancel ends the lost attempt + if err = st.Transition(ctx, a.ID, "lost", "cancelled", store.TransitionDetail{Epoch: a.Epoch, FailureClass: "cancelled", Reason: "coordinator lost"}); err != nil { + rep.Errors = append(rep.Errors, err) + } case errors.As(err, &se) && (se.Code == "E604" || se.Code == "E609"): - if err = st.Transition(ctx, a.ID, "lost", "needs-operator", d); err != nil { + err = st.Transition(ctx, a.ID, "lost", "needs-operator", d) + if CodeOf(err) == codeStoreRef { // same, for a job that must not run twice + err = st.Transition(ctx, a.ID, "lost", "cancelled", store.TransitionDetail{Epoch: a.Epoch, FailureClass: "cancelled", Reason: "coordinator lost"}) + } + if err != nil { rep.Errors = append(rep.Errors, err) continue } diff --git a/free/ci/internal/run/recovery_cancel_test.go b/free/ci/internal/run/recovery_cancel_test.go new file mode 100644 index 000000000..57f35e347 --- /dev/null +++ b/free/ci/internal/run/recovery_cancel_test.go @@ -0,0 +1,56 @@ +package run + +import ( + "context" + "path/filepath" + "strconv" + "testing" + + "github.com/nself-org/plugins/free/ci/internal/store" + "github.com/nself-org/plugins/free/ci/internal/storetest" +) + +// A lost attempt that got a cancel intent is ended cancelled by recovery, for +// a retryable job and for one that must not run twice, and its run closes. +func TestRecoveryCancelsALostAttemptWithACancelIntent(t *testing.T) { + for _, idempotent := range []bool{true, false} { + t.Run("idempotent="+strconv.FormatBool(idempotent), func(t *testing.T) { + home := t.TempDir() + t.Setenv("NSELF_CI_HOME", home) + st, err := store.Open(filepath.Join(home, "state.db"), store.Options{}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = st.Close() }() + ctx := context.Background() + dead, err := st.Register(ctx, 2147483000, strconv.Itoa(1), bootID()) + if err != nil { + t.Fatal(err) + } + must := func(e error) { + t.Helper() + if e != nil { + t.Fatal(e) + } + } + must(st.CreatePipeline(ctx, store.PipelineRow{ID: "p1", Project: "x", Status: "running"})) + must(st.CreateJob(ctx, store.JobRow{ID: "p1.j", PipelineID: "p1", Name: "j", Kind: "test", Idempotent: idempotent, Required: true, InfraMax: 2})) + must(st.CreateAttempt(ctx, store.AttemptRow{ID: "p1.j.a1", JobID: "p1.j", CoordinatorID: dead.ID, N: 1}, "")) + must(storetest.Admit(ctx, st, "p1.j.a1")) + _, epoch, err := st.AttemptStateEpoch(ctx, "p1.j.a1") + must(err) + must(st.Transition(ctx, "p1.j.a1", "leased", "lost", store.TransitionDetail{Epoch: epoch, FailureClass: "lost"})) + must(st.RequestCancel(ctx, "p1.j.a1", "operator", epoch+1)) // the cancel arrives after the attempt was lost + rep := Recover(ctx, st, "", home) + if len(rep.Errors) != 0 { + t.Fatalf("report %+v", rep) + } + if state, _, _ := st.AttemptStateEpoch(ctx, "p1.j.a1"); state != "cancelled" { + t.Fatalf("attempt state %s", state) + } + if row, _, _ := st.PipelineByID(ctx, "p1"); row.Status == "running" { + t.Fatalf("run still running") + } + }) + } +} From 980f3ad132780f93e2897b2de4e3463caf3e4bb5 Mon Sep 17 00:00:00 2001 From: admin Date: Sun, 25 Oct 2026 10:57:03 -0400 Subject: [PATCH 3/3] fix(ci): policy hook reads the coordinator home store instead of the default one --- free/ci/internal/run/coordinator.go | 2 +- free/ci/internal/run/hooks.go | 1 + free/ci/internal/run/policy_hook.go | 14 +++++++------- free/ci/internal/run/runscope_test.go | 21 +++++++++++++++++++++ 4 files changed, 30 insertions(+), 8 deletions(-) diff --git a/free/ci/internal/run/coordinator.go b/free/ci/internal/run/coordinator.go index 3445fd79d..aea8ea9d5 100644 --- a/free/ci/internal/run/coordinator.go +++ b/free/ci/internal/run/coordinator.go @@ -106,7 +106,7 @@ func (c *Coordinator) policy(ctx context.Context, h Hooks) (PolicyResult, error) if h.Policy != nil { r = h.Policy } - return r.Resolve(ctx, PolicyRequest{Root: c.Opts.Root, Pipeline: c.Plan.Pipeline.Name, Run: c.Opts.PolicyRun}) + return r.Resolve(ctx, PolicyRequest{Root: c.Opts.Root, Store: c.storePath(), Pipeline: c.Plan.Pipeline.Name, Run: c.Opts.PolicyRun}) } func (c *Coordinator) storePath() string { diff --git a/free/ci/internal/run/hooks.go b/free/ci/internal/run/hooks.go index 3ee65f8b2..d75d4f036 100644 --- a/free/ci/internal/run/hooks.go +++ b/free/ci/internal/run/hooks.go @@ -101,6 +101,7 @@ type TrustClassifier interface { // PolicyRequest is the input of the policy resolver, called once per pipeline. type PolicyRequest struct { Root, Pipeline string + Store string // state.db of the coordinator's home; "" = the default store Config map[string]any Run RunValues // per-run values (the flags of one invocation): only ever tighten, except the preset; nil = none } diff --git a/free/ci/internal/run/policy_hook.go b/free/ci/internal/run/policy_hook.go index 7839bb431..56fdf44fd 100644 --- a/free/ci/internal/run/policy_hook.go +++ b/free/ci/internal/run/policy_hook.go @@ -6,10 +6,9 @@ package run // Inputs: a PolicyRequest (the repository root). Outputs: a PolicyResult. // Constraints: the digest is the policy in force without the per-run values (they // steer placement only; evidence verification re-resolves without them); registered from init(); git runs through the exec funnel -// (report.Capture); the operator rows come from the default state.db, because -// PolicyRequest carries no store path (handoff: Coordinator.policy should pass -// Opts.Home); the local run is classed owner and the repository visibility -// is unknown, which keeps hosted execution off. +// (report.Capture); the operator rows come from PolicyRequest.Store (the +// coordinator's state.db; "" = the default state.db); the local run is classed +// owner and the repository visibility is unknown, which keeps hosted execution off. import ( "context" @@ -20,8 +19,9 @@ import ( "github.com/nself-org/plugins/free/ci/internal/store" ) -// PolicyHook resolves the policy for a run. The zero value uses the default -// store, the process-wide key set and report.Capture. +// PolicyHook resolves the policy for a run. The zero value uses the store of +// the request (the default one when it names none), the process-wide key set +// and report.Capture. type PolicyHook struct { OpenStore func() (*store.Store, error) Capture Capture @@ -34,7 +34,7 @@ func init() { RegisterPolicy(PolicyHook{}) } func (h PolicyHook) Resolve(ctx context.Context, r PolicyRequest) (PolicyResult, error) { open := h.OpenStore if open == nil { - open = func() (*store.Store, error) { return OpenStore("") } + open = func() (*store.Store, error) { return OpenStore(r.Store) } } capture := h.Capture if capture == nil { diff --git a/free/ci/internal/run/runscope_test.go b/free/ci/internal/run/runscope_test.go index a7ce9fd8f..32398126a 100644 --- a/free/ci/internal/run/runscope_test.go +++ b/free/ci/internal/run/runscope_test.go @@ -163,3 +163,24 @@ func TestIgnoredRunValuesReachSummary(t *testing.T) { t.Fatalf("notes = %q", res.Summary.Notes) } } + +// TestPolicyHookReadsTheRequestStore: the hook reads the operator rows from the +// store the request names (the coordinator's home), not the default one. +func TestPolicyHookReadsTheRequestStore(t *testing.T) { + ctx := context.Background() + t.Setenv("NSELF_CI_HOME", t.TempDir()) // the default store: empty + root, st, db := runScopeRepo(t) + project := policy.ProjectIDGitCached(ctx, policy.Runner(report.Capture), root) + err := st.ApplyWrite(ctx, store.PolicyWrite{Op: store.PolicySet, Scope: "project", Subject: project, Key: schedpolicy.KeyRunner, Value: []byte(`"ms6"`), Actor: "test"}, nil) + if err != nil { + t.Fatal(err) + } + run := RunValues{schedpolicy.KeyRunner: json.RawMessage(`"ms5"`)} + got, err := PolicyHook{}.Resolve(ctx, PolicyRequest{Root: root, Store: db, Run: run}) + if err != nil { + t.Fatal(err) + } + if len(got.IgnoredRun) != 1 { // only the pin in the request's store makes --runner ms5 refused + t.Fatalf("the project pin of the request store was not read: ignored %v", got.IgnoredRun) + } +}