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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1
2026-08-28 23:05:09 +04:00
parent 6565b9fce2
commit 03663f413b
2 changed files with 96 additions and 5 deletions
@@ -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" {
@@ -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")
+	}
+}