diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 0ba32b2..adddae1 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -157,6 +157,22 @@ type releaseTransaction struct { AgentReleased bool `json:"agent_released,omitempty"` LastError string `json:"last_error,omitempty"` UpdatedAt time.Time `json:"updated_at"` + // NextAttemptAt parks a commit the coordinator has refused. A refusal is + // an answer about the task, not a transport failure, so it stays true + // until something about the task changes. + NextAttemptAt time.Time `json:"next_attempt_at,omitempty"` + Attempts int `json:"attempts,omitempty"` +} + +// releaseBackoff spaces out refused commits. The first wait is long enough +// that a parked transaction stops filling the observation ring, and the cap +// keeps a reopen from waiting more than five minutes to be noticed. +func releaseBackoff(attempts int) time.Duration { + d := 30 * time.Second << (attempts - 1) + if attempts < 1 || d > 5*time.Minute { + return 5 * time.Minute + } + return d } type projectConfig struct { Repo string `json:"repo"` @@ -894,18 +910,31 @@ func (w *worker) advanceRelease(ctx context.Context, id string, s herdr.Session) w.releases[id] = tx _ = w.save() } + if tx.Phase == "anchor_pushed" && time.Now().Before(tx.NextAttemptAt) { + return + } if tx.Phase == "anchor_pushed" { // 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() + // A refusal is the coordinator's answer about who owns the task. + // It cannot change until an event about that task does, so asking + // again every five seconds only burns the observation ring. A + // transport failure is the opposite and must retry at once. + var refused *federation.StatusError + if errors.As(err, &refused) && refused.Code >= 400 && refused.Code < 500 { + tx.Attempts++ + tx.NextAttemptAt = time.Now().UTC().Add(releaseBackoff(tx.Attempts)) + } w.releases[id] = tx _ = w.save() w.recordError(fmt.Errorf("release %s commit: %w", id, err)) return } tx.Phase, tx.LastError, tx.UpdatedAt = "event_committed", "", time.Now().UTC() + tx.NextAttemptAt, tx.Attempts = time.Time{}, 0 w.releases[id] = tx _ = w.save() } @@ -1313,6 +1342,13 @@ func (w *worker) once(ctx context.Context) error { if t, ok := created(e); ok { w.tasks[t.ID] = t } + // Any event about this task is the change a parked commit was waiting + // for. A reopen arrives as TaskCorrected, so this cannot be a list of + // specific types without going stale. + if tx, parked := w.releases[e.TaskID]; parked && !tx.NextAttemptAt.IsZero() { + tx.NextAttemptAt, tx.Attempts = time.Time{}, 0 + w.releases[e.TaskID] = tx + } if e.Type == "TaskLeased" { var p struct { HarnessID string `json:"harness_id"` @@ -1403,7 +1439,7 @@ func (w *worker) once(ctx context.Context) error { w.leases[e.TaskID] = l } } - if e.Type == "TaskReleased" || e.Type == "TaskFailed" || e.Type == "TaskBlocked" { + if e.Type == "TaskReleased" || e.Type == "TaskFailed" || e.Type == "TaskBlocked" || e.Type == "TaskCompleted" { if e.Type == "TaskReleased" { var p struct { TransactionID string `json:"transaction_id"` @@ -1425,12 +1461,13 @@ func (w *worker) once(ctx context.Context) error { // mapping protects nothing. F30: a transaction stuck at "prepared" // held the session forever once its pane was gone, health() kept // reporting ActiveTask, and the harness never leased again. - // A failed task is terminal: no successor will ever lease it, so - // its anchor protects nothing and its transaction can only retry - // a refusal forever. Blocked is different, because a reopen still - // produces a successor. + // Failed and completed are terminal: no successor will ever lease + // the task, so the anchor protects nothing and the transaction can + // only retry a refusal forever. Blocked is different, because a + // reopen returns the task to the queue and the epoch that ended is + // still on record, so that exact commit can still be accepted. tx, releasing := w.releases[e.TaskID] - if !releasing || tx.Ref == "" || e.Type == "TaskFailed" { + if !releasing || tx.Ref == "" || e.Type == "TaskFailed" || e.Type == "TaskCompleted" { if releasing { delete(w.releases, e.TaskID) } diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index 4315093..60fadb0 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -1327,3 +1327,120 @@ func TestObservationRingEvictsLeastRecentlySeen(t *testing.T) { t.Fatal("least recently seen entry survived") } } + +// F60. A refusal is an answer about the task, not a transport failure, and it +// stays true until something about that task changes. Run 10's blocked task +// asked 5,000 times over seven hours and got the same 409 every time. +func TestRefusedCommitParksUntilSomethingChanges(t *testing.T) { + var commits int + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/handoff"): + commits++ + http.Error(w, "lease not owned", http.StatusConflict) + case strings.HasSuffix(r.URL.Path, "/events"): + _, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"c","type":"TaskCorrected","task_id":"t","version":9,"payload":{"state":"queued"},"surface":"web"}]}`)) + default: + w.WriteHeader(http.StatusNoContent) + } + })) + defer s.Close() + tx := releaseTransaction{ID: "tx", Phase: "anchor_pushed", Ref: "sha256:abc", AnchorSHA: "abc", LeaseEpoch: "e1"} + 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": tx}, + quarantined: map[string]bool{}, + statePath: t.TempDir() + "/state.json", + hard: .75, + } + w.advanceRelease(context.Background(), "t", w.sessions["t"]) + w.advanceRelease(context.Background(), "t", w.sessions["t"]) + if commits != 1 { + t.Fatalf("refused commit retried %d times without waiting", commits) + } + if w.releases["t"].NextAttemptAt.IsZero() { + t.Fatal("refused commit was not parked") + } + // A reopen arrives as TaskCorrected. Any event about the task is the + // change the parked commit was waiting for, so the same tick retries it + // and, still refused, parks it again. + if err := w.once(context.Background()); err != nil { + t.Fatal(err) + } + if commits != 2 { + t.Fatalf("an event about the task did not un-park its commit, commits=%d", commits) + } + if w.releases["t"].NextAttemptAt.IsZero() { + t.Fatal("the second refusal did not park it again") + } +} + +// The opposite case, and the one a backoff must not break: the coordinator is +// unreachable or broken rather than answering. That says nothing about who +// owns the task, so it has to retry at once. +func TestTransientCommitFailureKeepsRetryingAtOnce(t *testing.T) { + var commits int + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/handoff") { + commits++ + http.Error(w, "upstream unavailable", http.StatusBadGateway) + 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{}, + 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"]) + w.advanceRelease(context.Background(), "t", w.sessions["t"]) + if commits != 2 { + t.Fatalf("transport failure was parked like a refusal, commits=%d", commits) + } + if !w.releases["t"].NextAttemptAt.IsZero() { + t.Fatal("transport failure must not park the transaction") + } +} + +// Completion is terminal for a release transaction just as failure is. The +// task is done; nothing will ever lease it again to pick the anchor up. +func TestCompletedTaskDropsItsReleaseTransaction(t *testing.T) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/events") { + _, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"c","type":"TaskCompleted","task_id":"t","version":9,"payload":{"report_ref":"sha256:r"},"surface":"system"}]}`)) + return + } + 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.releases) != 0 || len(w.sessions) != 0 { + t.Fatalf("completed task kept its release: releases=%v sessions=%v", w.releases, w.sessions) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index dd1b415..a5cad53 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -424,6 +424,13 @@ func (s *Store) apply(e domain.Event) error { t.Lease = nil case "TaskBlocked", "TaskNeedsAttention": if e.Type == "TaskBlocked" { + if t.Lease != nil { + // Same reason as TaskReleased: a worker may hold a pushed + // anchor whose commit was refused. A reopen returns the task + // to the queue, and the late-handoff path can only accept it + // if the epoch that ended is still on record. + t.LastLeaseEpoch = t.Lease.Epoch + } t.State = domain.StateBlocked t.Lease = nil } else { diff --git a/internal/store/store_test.go b/internal/store/store_test.go index b5c1fa0..21aad52 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -693,3 +693,33 @@ func TestExpiryRetainsLeaseEpoch(t *testing.T) { t.Fatalf("last lease epoch %q, want %q", after.LastLeaseEpoch, epoch) } } + +// A worker can hold a pushed anchor whose commit was refused when an operator +// blocks the task. A reopen returns it to the queue, and the late-handoff path +// can only accept that exact owner if the epoch that ended is still recorded. +func TestBlockRetainsLeaseEpochForALaterReopen(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.Minute); err != nil { + t.Fatal(err) + } + leased, _ := s.Task(id) + epoch := leased.Lease.Epoch + p, _ := json.Marshal(map[string]any{"blocker": "parked by the operator", "harness_id": "h1", "lease_epoch": epoch}) + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskBlocked", TaskID: id, Version: leased.Version + 1, Payload: p, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + after, _ := s.Task(id) + if after.State != domain.StateBlocked || after.Lease != nil { + t.Fatalf("expected a blocked unleased task, got %s lease=%v", after.State, after.Lease) + } + if after.LastLeaseEpoch != epoch || epoch == "" { + t.Fatalf("last lease epoch %q, want %q", after.LastLeaseEpoch, epoch) + } +}