diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 9214bc6..8be65be 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -119,6 +119,7 @@ type lease struct { } type releaseTransaction struct { ID string `json:"id"` + LeaseEpoch string `json:"lease_epoch,omitempty"` LeaseVersion int `json:"lease_version"` Ref string `json:"handoff_ref,omitempty"` AnchorSHA string `json:"anchor_sha,omitempty"` @@ -836,7 +837,7 @@ func (w *worker) advanceRelease(ctx context.Context, id string, s herdr.Session) w.recordError(fmt.Errorf("release %s: lease missing", id)) return } - tx = releaseTransaction{ID: domain.NewID(), LeaseVersion: l.Version, Phase: "prepared", UpdatedAt: time.Now().UTC()} + tx = releaseTransaction{ID: domain.NewID(), LeaseEpoch: l.Epoch, LeaseVersion: l.Version, Phase: "prepared", UpdatedAt: time.Now().UTC()} w.releases[id] = tx _ = w.save() } @@ -864,8 +865,10 @@ func (w *worker) advanceRelease(ctx context.Context, id string, s herdr.Session) _ = w.save() } if tx.Phase == "anchor_pushed" { - l := w.leases[id] - if err := w.api.Release(ctx, id, tx.Ref, tx.AnchorSHA, tx.ID, l.Epoch, tx.LeaseVersion, w.sessionEvidence(ctx, id, s)); err != nil { + // The epoch comes from the transaction, not from w.leases: an expiry + // replay deletes the lease, and the coordinator needs the epoch of the + // lease this anchor was pushed under to accept the late commit. + if err := w.api.Release(ctx, id, tx.Ref, tx.AnchorSHA, tx.ID, tx.LeaseEpoch, tx.LeaseVersion, w.sessionEvidence(ctx, id, s)); err != nil { tx.LastError, tx.UpdatedAt = err.Error(), time.Now().UTC() w.releases[id] = tx _ = w.save() diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index 7f0db49..d9a7d49 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -1123,3 +1123,40 @@ func TestHoldsLeaseGatesWorkOnATaskTheWorkerLost(t *testing.T) { t.Error("a task this worker never leased is held") } } + +// Run 10 lost a finished task here: the anchor was pushed, the lease expired, +// and every retry sent the epoch from w.leases — which the expiry replay had +// already deleted. An empty epoch can never be accepted, so the work sat in +// the worktree until retry_limit. The epoch belongs to the transaction. +func TestExpiredReleaseStillCommitsWithTheTransactionEpoch(t *testing.T) { + var sent struct { + LeaseEpoch string `json:"lease_epoch"` + } + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/handoff") { + _ = json.NewDecoder(r.Body).Decode(&sent) + w.WriteHeader(http.StatusNoContent) + return + } + w.WriteHeader(http.StatusNoContent) + })) + defer s.Close() + w := &worker{ + api: federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"}, + harnessID: "h", + tasks: map[string]domain.Task{"t": {ID: "t"}}, + sessions: map[string]herdr.Session{"t": {PaneID: "pane"}}, + leases: map[string]lease{}, // expiry deleted it + releases: map[string]releaseTransaction{"t": {ID: "tx", Phase: "anchor_pushed", Ref: "sha256:abc", AnchorSHA: "abc", LeaseEpoch: "e1"}}, + quarantined: map[string]bool{}, + statePath: t.TempDir() + "/state.json", + hard: .75, + } + w.advanceRelease(context.Background(), "t", w.sessions["t"]) + if sent.LeaseEpoch != "e1" { + t.Fatalf("release sent epoch %q, want the transaction's e1", sent.LeaseEpoch) + } + if got := w.releases["t"].Phase; got != "event_committed" { + t.Fatalf("phase %q, want event_committed", got) + } +} diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 9b7c6f0..7d0b730 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -162,6 +162,19 @@ func validateLocalMachine(rr registry.Registry, localMachine string) error { return nil } +// lateHandoffAccepted reports whether a release may still commit after its +// lease expired. A worker pushes the anchor first and commits second; when the +// lease dies in between, the finished work is durable in git and unreachable +// forever (run 10 lost a whole task that way). The expired owner is the only +// caller that can present the ended lease's epoch, and this fires only while +// the task is unleased and its reclaim recorded no handoff, so it can never +// overwrite a successor's work. Store.Append's version fence settles a race +// with a concurrent re-lease. +func lateHandoffAccepted(t domain.Task, leaseEpoch string) bool { + return t.State == domain.StateQueued && t.Lease == nil && t.HandoffRef == "" && + leaseEpoch != "" && leaseEpoch == t.LastLeaseEpoch +} + func main() { dir := os.Getenv("ORCHESTRA_DATA") if dir == "" { @@ -1428,7 +1441,8 @@ func main() { w.WriteHeader(http.StatusNoContent) return } - if !ownedLease { + lateHandoff := !ownedLease && strings.HasSuffix(r.URL.Path, "/handoff") && lateHandoffAccepted(t, b.LeaseEpoch) + if !ownedLease && !lateHandoff { http.Error(w, "lease not owned", 409) return } @@ -1656,7 +1670,9 @@ func main() { http.Error(w, "anchor_sha required", 400) return } - if b.TransactionID == "" || b.ExpectedVersion != t.Version { + // A late handoff cannot know the version: expiry bumped it after the + // worker read it. Its fence is the epoch plus the append. + if b.TransactionID == "" || (!lateHandoff && b.ExpectedVersion != t.Version) { http.Error(w, "release transaction and current lease version required", http.StatusConflict) return } diff --git a/cmd/orchestra/main_test.go b/cmd/orchestra/main_test.go index e64e6a4..28ddbec 100644 --- a/cmd/orchestra/main_test.go +++ b/cmd/orchestra/main_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "orchestra/internal/domain" "orchestra/internal/registry" ) @@ -89,3 +90,22 @@ func TestTmuxRegistryRequiresMachineIdentityEvenOnOneMachine(t *testing.T) { t.Fatal(err) } } + +func TestLateHandoffAcceptedOnlyFromTheExpiredOwner(t *testing.T) { + expired := domain.Task{State: domain.StateQueued, LastLeaseEpoch: "ep1"} + if !lateHandoffAccepted(expired, "ep1") { + t.Fatal("the expired owner must still commit its pushed anchor") + } + for name, tk := range map[string]domain.Task{ + "re-leased": {State: domain.StateLeased, Lease: &domain.Lease{Epoch: "ep2"}, LastLeaseEpoch: "ep1"}, + "already handed off": {State: domain.StateQueued, LastLeaseEpoch: "ep1", HandoffRef: "sha256:x"}, + "blocked": {State: domain.StateBlocked, LastLeaseEpoch: "ep1"}, + } { + if lateHandoffAccepted(tk, "ep1") { + t.Fatalf("%s must refuse a late handoff", name) + } + } + if lateHandoffAccepted(expired, "ep2") || lateHandoffAccepted(domain.Task{State: domain.StateQueued}, "") { + t.Fatal("a foreign or empty epoch must refuse") + } +} diff --git a/internal/domain/domain.go b/internal/domain/domain.go index f88f02b..55bbfb9 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -173,6 +173,12 @@ type Task struct { BlockedAt time.Time `json:"blocked_at,omitempty"` LastPaneID string `json:"last_pane_id,omitempty"` LastHarness string `json:"last_harness_id,omitempty"` + // LastLeaseEpoch is the fencing token of the lease that most recently + // ended. A worker can push its release anchor and only then discover the + // lease expired; the finished work is durable in git but the commit can + // never land. Retaining the epoch lets exactly that owner still commit + // while the task sits unleased. + LastLeaseEpoch string `json:"last_lease_epoch,omitempty"` PaneState string `json:"pane_state,omitempty"` // open, closed, unreachable, unknown LastSession SessionEvidence `json:"last_session,omitempty"` // Recovery state is part of the durable projection, never process-local diff --git a/internal/store/store.go b/internal/store/store.go index ec7efe0..dd1b415 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -387,6 +387,9 @@ func (s *Store) apply(e domain.Event) error { case "TaskLaunchAcknowledged": t.LifecyclePhase = "started" case "TaskReleased": + if t.Lease != nil { + t.LastLeaseEpoch = t.Lease.Epoch + } t.State = domain.StateQueued t.LifecyclePhase = "reclaimed" t.Lease = nil diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 12aca6c..b5c1fa0 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -663,3 +663,33 @@ func TestQuotaSinceReportsEmptyWindowAsKnownZero(t *testing.T) { } const fiveHours = 5 * time.Hour + +// TestExpiryRetainsLeaseEpoch covers the late-handoff fence: a worker that +// pushed its release anchor and then lost the lease can only commit if the +// projection still knows which epoch just ended. +func TestExpiryRetainsLeaseEpoch(t *testing.T) { + s, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := s.Append(created("e1")); err != nil { + t.Fatal(err) + } + id := s.Tasks()[0].ID + if _, err := s.Lease(id, "h1", time.Millisecond); err != nil { + t.Fatal(err) + } + leased, _ := s.Task(id) + epoch := leased.Lease.Epoch + time.Sleep(2 * time.Millisecond) + if _, err := s.ExpireLease(id, time.Now()); err != nil { + t.Fatal(err) + } + after, _ := s.Task(id) + if after.Lease != nil { + t.Fatal("expired task still holds a lease") + } + if after.LastLeaseEpoch != epoch || epoch == "" { + t.Fatalf("last lease epoch %q, want %q", after.LastLeaseEpoch, epoch) + } +}