From 03663f413bb76d57e1dad1e8eea36e693edb0c05 Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 28 Aug 2026 23:05:09 +0400 Subject: [PATCH] Abandon a release transaction another lease has superseded The late-handoff path fences on the epoch that expired. Once the task is leased again under a different transaction, that epoch is two owners old and the commit can never be accepted. The worker kept asking anyway, every five seconds, holding the pane and pinning both ActiveTask and the single last_error slot. Run 10's task did that for seven hours. Proven live at 19:01:30Z: the lease expired while the anchor was pushing, a successor took the task 13ms later, and the coordinator refused the late commit with 409 lease not owned. The refusal is right. The retry loop behind it was not. TaskLeased now abandons a release transaction whose id the lease does not carry, and quarantines its session. A successor pickup carries the predecessor's own transaction id, so the recoverable predecessor F30 protects is left alone. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1 --- cmd/orchestra-worker/main.go | 26 ++++++++--- cmd/orchestra-worker/main_test.go | 75 +++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 5 deletions(-) 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") + } +}