fix(ci): outbox supersede requeue, lost+cancel recovery, policy hook home store (D-1120, D-1121, D-1124)

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 {

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
 }

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 {

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
 			}

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")
+			}
+		})
+	}
+}

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)
+	}
+}

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<? AND state=? AND claimed_at>?", revision, statusContext, seq, OutboxPosting, cutoff).Scan(&older); err != nil {

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)
+	}
+}