diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 8be65be..f4477de 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -1291,12 +1291,28 @@ func (w *worker) once(ctx context.Context) error { TransactionID string `json:"transaction_id"` AnchorSHA string `json:"anchor_sha"` } - if json.Unmarshal(e.Payload, &p) == nil && p.HarnessID == w.harnessID { - var until struct { - UntilNS int64 `json:"until_ns"` + if json.Unmarshal(e.Payload, &p) == nil { + if p.HarnessID == w.harnessID { + var until struct { + UntilNS int64 `json:"until_ns"` + } + _ = json.Unmarshal(e.Payload, &until) + w.leases[e.TaskID] = lease{Epoch: p.Epoch, HandoffRef: p.HandoffRef, TransactionID: p.TransactionID, AnchorSHA: p.AnchorSHA, Version: e.Version, Until: time.Unix(0, until.UntilNS)} + } + // A release transaction is superseded the moment the task is + // leased under a different transaction, our own re-lease + // included. Its anchor can never commit: the late-handoff path + // fences on the epoch that expired, and the owner has moved on + // twice since. Retrying forever pins the session, ActiveTask + // and the single last_error slot, which is where run 10's task + // spent seven hours. + if tx, releasing := w.releases[e.TaskID]; releasing && tx.ID != p.TransactionID { + delete(w.releases, e.TaskID) + w.recordError(fmt.Errorf("release %s superseded by lease %s: abandoning transaction %s", e.TaskID, p.Epoch, tx.ID)) + if session, active := w.sessions[e.TaskID]; active { + w.quarantine(ctx, e.TaskID, session) + } } - _ = json.Unmarshal(e.Payload, &until) - w.leases[e.TaskID] = lease{Epoch: p.Epoch, HandoffRef: p.HandoffRef, TransactionID: p.TransactionID, AnchorSHA: p.AnchorSHA, Version: e.Version, Until: time.Unix(0, until.UntilNS)} } } if e.Type == "TaskLeaseRenewed" { diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index d9a7d49..8b9b861 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -1160,3 +1160,78 @@ func TestExpiredReleaseStillCommitsWithTheTransactionEpoch(t *testing.T) { t.Fatalf("phase %q, want event_committed", got) } } + +// Proven live at 19:01:30Z on 2026-08-28: the lease expired while the anchor +// was pushing, a successor took the task, and the coordinator refused the late +// commit. That refusal is correct and permanent, so the worker must stop +// asking. Without this it retried every five seconds forever, holding the pane +// and pinning the single last_error slot. +func TestSupersededReleaseIsAbandonedNotRetriedForever(t *testing.T) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/federation/events": + _, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"l","type":"TaskLeased","task_id":"t","version":9,` + + `"payload":{"harness_id":"other","lease_epoch":"e2"},"surface":"system"}]}`)) + default: + w.WriteHeader(http.StatusNoContent) + } + })) + defer s.Close() + w := &worker{ + api: federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"}, + harnessID: "h", + backend: deadTmuxBackend(t), + tasks: map[string]domain.Task{"t": {ID: "t"}}, + sessions: map[string]herdr.Session{"t": {PaneID: "pane"}}, + leases: map[string]lease{}, + 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, + } + if err := w.once(context.Background()); err != nil { + t.Fatal(err) + } + if len(w.sessions) != 0 { + t.Fatalf("session still pinned: %v", w.sessions) + } + if _, still := w.releases["t"]; still { + t.Fatalf("superseded transaction retained: %v", w.releases) + } + if got := w.health(context.Background()).ActiveTask; got != "" { + t.Fatalf("ActiveTask=%q, worker still advertises the dead release", got) + } +} + +// The successor pickup carries the predecessor's own transaction id. That +// lease is the handoff completing, not a supersession, and dropping it there +// would destroy the recoverable predecessor F30 exists to protect. +func TestOwnPickupLeaseKeepsTheReleaseTransaction(t *testing.T) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/federation/events": + _, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"l","type":"TaskLeased","task_id":"t","version":9,` + + `"payload":{"harness_id":"h","lease_epoch":"e2","transaction_id":"tx","handoff_ref":"sha256:abc"},"surface":"system"}]}`)) + default: + 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{}, + releases: map[string]releaseTransaction{"t": {ID: "tx", Phase: "event_committed", Ref: "sha256:abc", AnchorSHA: "abc", LeaseEpoch: "e1"}}, + quarantined: map[string]bool{}, + statePath: t.TempDir() + "/state.json", + hard: .75, + } + if err := w.once(context.Background()); err != nil { + t.Fatal(err) + } + if _, ok := w.releases["t"]; !ok { + t.Fatal("pickup lease dropped its own release transaction") + } +}