Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 595a1d3533 | |||
| 0797432d6f | |||
| 3c7cf95d8c | |||
| 0f83559ecc |
@@ -2274,3 +2274,253 @@ tmux -L orchestra kill-session -t <session>
|
||||
|
||||
Task `06G4JX6MSQEP7N0D5JWW9EP5X4` is the other run 12 task and is `in_review` on
|
||||
a pull request. It is real work and should be reviewed or failed, not cleaned.
|
||||
|
||||
### F18 closed: a bounded observation ring
|
||||
|
||||
`d6ee10f`. `WorkerHealth` now carries up to sixteen distinct observations, each
|
||||
with a repeat count and first/last times. Collapsing is by message, not by
|
||||
position, because a loop interleaved with other failures would otherwise still
|
||||
flush the ring. Eviction drops the least recently seen, so a loop keeps its
|
||||
slot but carries the count that says what it is. `last_error` and `error_at`
|
||||
keep their wire names and still report only the newest failure.
|
||||
|
||||
The ring lives in memory beside `last_error` and is not persisted, which is
|
||||
what `last_error` already did across a restart.
|
||||
|
||||
Deployed and verified on both halves. It reports nothing yet, because
|
||||
`omitempty` hides an empty ring and nothing has failed since the deploy. No UI
|
||||
renders it: `internal/ui` never showed `last_error` either, so
|
||||
`GET /v1/federation/workers` is the only reader today.
|
||||
|
||||
### Run 12 closed out
|
||||
|
||||
PR 17 merged, task `06G4JX6MSQEP7N0D5JWW9EP5X4` completed at v32, pane closed
|
||||
by the normal completion path. The quality gate was re-run independently
|
||||
against the submitted commit `419e17fc` before the merge, rather than trusted
|
||||
from the worker's own report:
|
||||
|
||||
```text
|
||||
bash -n scripts/*.sh syntax ok
|
||||
bash scripts/orchestra_e2e_healthcheck.sh OK - all healthchecks passed, exit 0
|
||||
bash scripts/test_healthcheck.sh all checks passed, exit 0
|
||||
```
|
||||
|
||||
Worker state is empty: no sessions, no leases, no release transactions, and
|
||||
`tmux -L orchestra ls` lists nothing. The two panes the previous section left
|
||||
for an operator were already gone by then.
|
||||
|
||||
| Half | Revision |
|
||||
|---|---|
|
||||
| Coordinator, homesrv container | `d6ee10f` |
|
||||
| Worker, workpc systemd | `d6ee10f` |
|
||||
|
||||
```text
|
||||
commit d6ee10f02843e2267c859b03d17db00995941e64
|
||||
coordinator sha256 f8a847477ae0ac03fe78338775f52ed1f167f0e67eab557feffe4d67da3ad99e
|
||||
worker sha256 fe7d2188b12369fb7682d3424fa11e2db47abed9c55313291a4b467b5972d02d
|
||||
```
|
||||
|
||||
### Still open after run 12
|
||||
|
||||
- **29 blocked `test-e2e` tasks**, mostly burn-in debris. Operator hygiene, not
|
||||
lifecycle code.
|
||||
- **Three queued `correx` tasks** that cannot be scheduled, because `correx` has
|
||||
no entry in the coordinator's `config.jsonc`. Either give it one or block them
|
||||
explicitly.
|
||||
- **Gitea returns 422 for a review on your own pull request**, so
|
||||
`REQUEST_CHANGES` needs a separate bot account for `ORCHESTRA_GITEA_TOKEN`.
|
||||
- **A blocked task that is never reopened can still loop** on a dead release
|
||||
transaction. F59 covers failed, not blocked.
|
||||
|
||||
## Run 13, 2026-08-28: F60, and the ring proven under a real failure
|
||||
|
||||
### F60: a transaction must settle deterministically, in every case
|
||||
|
||||
`3c7cf95`. F58 and F59 each fixed one case of the same rule. F60 states the
|
||||
rule and covers the rest:
|
||||
|
||||
> a transaction that can no longer be legitimately committed must be abandoned
|
||||
> deterministically, and must never spin on an answer that cannot change.
|
||||
|
||||
- **Terminal is failed or completed.** Both drop the transaction and free the
|
||||
session. Completed was the gap F59 left.
|
||||
- **Blocked keeps the transaction**, because a reopen returns the task to the
|
||||
queue and that exact owner can still commit.
|
||||
- **`TaskBlocked` now retains the ending epoch**, as `TaskReleased` already did.
|
||||
Without it a reopened task has no epoch for the late-handoff path to fence
|
||||
against, so keeping the transaction would be a lie.
|
||||
- **A refusal parks the commit**, 30s backing off to a 5 minute cap. Any event
|
||||
about the task un-parks it. A reopen arrives as `TaskCorrected`, so the rule
|
||||
cannot be a list of event types.
|
||||
- **A transport failure is not an answer** and retries at once. The park keys on
|
||||
a 4xx `StatusError`, never on any error.
|
||||
|
||||
### Proven live, and the contrast with run 10
|
||||
|
||||
Task `06G4KVSHAK9B8M9K8HENAF23CG`, same race-guard rig: force the expiry inside
|
||||
the push, then hand the task to `race-guard-probe`.
|
||||
|
||||
```text
|
||||
19:54:10.888 transaction 06G4KVYT91GQH5ZYY6CNG6FWZ8 opens at prepared
|
||||
19:54:10.907 forced expiry accepted
|
||||
19:54:10.916 successor lease to race-guard-probe accepted
|
||||
19:54:11.776 commit refused once, 409 lease not owned, parked
|
||||
19:54:15.741 superseded by lease 06G4KVYSXJH053K3DAG6FKHD20, transaction abandoned
|
||||
```
|
||||
|
||||
Run 10's equivalent ran roughly 5,000 retries over seven hours and needed an
|
||||
operator to clear the state file by hand. This one asked once and settled
|
||||
itself in four seconds. Worker state afterwards: no sessions, no leases, no
|
||||
transactions, and `tmux -L orchestra ls` empty.
|
||||
|
||||
### F18 populated under a real failure
|
||||
|
||||
The gap the previous section recorded is closed. The same run produced this
|
||||
health payload, two distinct observations rather than one overwritten slot:
|
||||
|
||||
```json
|
||||
"observations": [
|
||||
{"message": "release 06G4KVSHAK9B8M9K8HENAF23CG commit: federation: 409 Conflict: lease not owned",
|
||||
"count": 1, "first": "2026-08-28T19:54:11.776815524Z", "last": "2026-08-28T19:54:11.776815524Z"},
|
||||
{"message": "release 06G4KVSHAK9B8M9K8HENAF23CG superseded by lease 06G4KVYSXJH053K3DAG6FKHD20: abandoning transaction 06G4KVYT91GQH5ZYY6CNG6FWZ8",
|
||||
"count": 1, "first": "2026-08-28T19:54:15.74116014Z", "last": "2026-08-28T19:54:15.74116014Z"}
|
||||
]
|
||||
```
|
||||
|
||||
Under the old single slot the abandonment message would have erased the 409
|
||||
that caused it, and the causal chain would have been unreadable. Both counts
|
||||
are 1, which is itself the F60 evidence: nothing looped.
|
||||
|
||||
### Deployed state
|
||||
|
||||
| Half | Revision |
|
||||
|---|---|
|
||||
| Coordinator, homesrv container | `3c7cf95` |
|
||||
| Worker, workpc systemd | `3c7cf95` |
|
||||
|
||||
```text
|
||||
commit 3c7cf95d8cdaaff5dcec460c5d7874f7f4f0dbea
|
||||
coordinator sha256 3fec1e99a638da851f5c970abaf80a006dc87425936f71fe33f10830f4bac573
|
||||
worker sha256 a373445f167f0a321dfc3866a1d2ed703b670b0fb3c188a8d00ad8099586dfdf
|
||||
```
|
||||
|
||||
`orchestra-f18-baseline` tags `d6ee10f`, the post-F18 deployed baseline, and is
|
||||
pushed. The branch is pushed too: 91 commits, because nothing had gone up since
|
||||
`97a9c65` on 2026-07-31.
|
||||
|
||||
### Still open
|
||||
|
||||
Unchanged and all operational rather than runtime: 29 blocked `test-e2e` tasks,
|
||||
three unschedulable `correx` tasks, and the Gitea 422 on reviewing your own
|
||||
pull request. The rig task above will requeue when the probe lease expires and
|
||||
start fresh, which is correct: its anchor was abandoned, not committed.
|
||||
|
||||
## Run 14, 2026-08-28: plan-phase execution proven on `3c7cf95`
|
||||
|
||||
Four tasks. Two carried a full lifecycle to a pull request, one failed for a
|
||||
cause outside Orchestra, and the F60 rig task failed on its retry budget.
|
||||
|
||||
### The rung: the worker runs the sealed plan's commands
|
||||
|
||||
Task `06G4M8WHGQ4P3GQMPEEH0RJRHM`, plan `789ed6a8477b`, both phases verified
|
||||
against the plan document rather than the request.
|
||||
|
||||
```text
|
||||
plan phase-1 - run: ["bash", "-n", "scripts/orchestra_e2e_healthcheck.sh"]
|
||||
- run: ["bash", "scripts/orchestra_e2e_healthcheck.sh"]
|
||||
event v19 commands identical, exit_codes [0,0], at_sha 3b66b2b6a669,
|
||||
evidence_ref a5dbc4ecc9ce, status awaiting_manual_verification
|
||||
|
||||
plan phase-2 - run: ["bash", "-n", "scripts/orchestra_e2e_healthcheck.sh"]
|
||||
- run: ["bash", "-n", "scripts/test_healthcheck.sh"]
|
||||
- run: ["bash", "scripts/orchestra_e2e_healthcheck.sh"]
|
||||
event v20 commands identical, exit_codes [0,0,0], same at_sha and plan_ref
|
||||
```
|
||||
|
||||
Every record binds `plan_ref`, `phase_id`, `at_sha`, `evidence_ref`,
|
||||
`lease_epoch` and `harness_id`. Rotation ran between every work phase with
|
||||
`TaskPickupValidated` each time.
|
||||
|
||||
**Not yet proven:** a successor picking up mid-implement. Both phases verified
|
||||
under one lease epoch, so no rotation separated phase 1 from phase 2.
|
||||
|
||||
**The planner found the legal route to a forbidden command.**
|
||||
`scripts/test_healthcheck.sh` may not be executed, but `["bash", "-n", ...]`
|
||||
covers any path, so it syntax-checked the file it could not run.
|
||||
|
||||
### The refusal loop is actionable, not merely observable
|
||||
|
||||
Task `06G4M6HF1Z3EREX1X3NEKSHP24` sealed a plan naming
|
||||
`bash scripts/test_healthcheck.sh`, which the policy forbids.
|
||||
|
||||
```text
|
||||
20:45:02 phase request refused: "bash scripts/test_healthcheck.sh" is not in the policy
|
||||
20:45:33 implement <- the planner rewrote and resealed, 31 seconds later
|
||||
20:48:03 in_review
|
||||
```
|
||||
|
||||
F52 refused on the planner, F55's answer path delivered the reason, and the
|
||||
agent corrected. Count stayed at 1 for both tasks that hit it. A brief that
|
||||
states the policy avoids the round trip entirely: the `--list-checks` task was
|
||||
told the allowed argv and sealed on the first attempt.
|
||||
|
||||
### F61, recorded and deliberately not built
|
||||
|
||||
The verification allow-list is enforced and never stated to the planner. Every
|
||||
plan that guesses pays one refused round trip. It is not a one-liner:
|
||||
`agentctx.Input.Policy` is filled from the worker's `SafeOperations`, while the
|
||||
verification policy is coordinator-side by design, so this needs a new channel.
|
||||
|
||||
The reason to promote it later is local models. A strong planner consumed the
|
||||
refusal and repaired the plan in 31 seconds. A weaker one may propose forbidden
|
||||
commands repeatedly, because it cannot infer the allowed substitute. Then F61
|
||||
stops being latency and becomes model compatibility.
|
||||
|
||||
### opencode is not running, and the reason was only in the pane
|
||||
|
||||
Task `06G4M6HH5BAM235AC4PWYXF1HM` failed at attempt 3 having never left
|
||||
`frame`. Its worktree held only `launch.md`. The pane said why:
|
||||
|
||||
```text
|
||||
This model is not available in your country.
|
||||
▣ Build · Muse Spark 1.2 Free OpenCode Zen
|
||||
```
|
||||
|
||||
Orchestra behaved correctly throughout: lease, no progress, expire, requeue,
|
||||
fail. Two things are worth keeping.
|
||||
|
||||
- **The adapter cannot resolve a session file for opencode**, so activity is
|
||||
`unknown` and the worker falls back to agent status. The failure therefore
|
||||
reads as "done and pane unchanged" rather than "dead". Orchestra cannot tell
|
||||
finished from never-started on this harness.
|
||||
- **No observation carried the cause.** Four of them named the lease, the
|
||||
rotation and the activity. The fatal line existed only in the pane capture,
|
||||
which Orchestra publishes but nothing summarises. Parsing harness chrome is
|
||||
not Orchestra's job; surfacing the capture is.
|
||||
|
||||
**Do not route to `workpc-opencode` until its model is reachable.** Every task
|
||||
sent there burns three attempts and then fails.
|
||||
|
||||
### The retry budget counts idleness, not just failure
|
||||
|
||||
The F60 rig task `06G4KVSHAK9B8M9K8HENAF23CG` failed at attempt 3 mid-review.
|
||||
Two attempts were rig-induced expiries and the third was a natural idle
|
||||
expiry. Every lease expiry increments `attempt`, and an agent going quiet at a
|
||||
phase boundary is a routine expiry here. A task making real progress can spend
|
||||
its retry budget on idleness. The two clean tasks never expired at all, so this
|
||||
is a dynamic to watch rather than a defect to fix.
|
||||
|
||||
F59 was confirmed live twice more: both failed tasks left no release
|
||||
transaction and no session behind, on either worker.
|
||||
|
||||
### Result
|
||||
|
||||
| Task | Harness | Outcome |
|
||||
|---|---|---|
|
||||
| `06G4M6HF1Z3EREX1X3NEKSHP24` | workpc-claude | in_review, recovered from a policy refusal |
|
||||
| `06G4M8WHGQ4P3GQMPEEH0RJRHM` | workpc-claude | in_review, plan-phase execution proven |
|
||||
| `06G4M6HH5BAM235AC4PWYXF1HM` | workpc-opencode | failed, model unavailable |
|
||||
| `06G4KVSHAK9B8M9K8HENAF23CG` | workpc-claude | failed, retry budget |
|
||||
|
||||
Both workers hold no sessions and no release transactions. Deployed pair is
|
||||
still `3c7cf95` on both halves.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user