Let an expired lease still commit the anchor it pushed
A release pushes the anchor first and commits second. When the lease expired in between, the commit could never land: the worker sent the epoch from w.leases, which the expiry replay had already deleted, and the coordinator refused a /handoff without a live owned lease. Run 10 lost a finished task this way, its work sitting in the worktree until retry_limit. The epoch now belongs to the release transaction, so it survives the lease. TaskReleased retains the ending epoch as Task.LastLeaseEpoch, and lateHandoffAccepted lets exactly that owner commit while the task is queued, unleased, and carrying no handoff of its own. A successor that has already re-leased the task holds the lease, so the late handoff loses. Store.Append's version fence settles the race, which is why the late path skips the version check the worker cannot satisfy. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
+18
-2
@@ -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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user