amadan
hostrelocate: a worker pass that refuses in releasing or probe-failed checks the release deadline first, so a source removed from the roster leaves the operation stuck at the deadline instead of refusing for ever
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+74
−3
internal/hostrelocate/worker.go+27 −3internal/hostrelocate/worker_test.go+47 −0
CI failed — run details
diff --git a/internal/hostrelocate/worker.go b/internal/hostrelocate/worker.go| index 1614e27c8..c79758010 100644 |
| --- a/internal/hostrelocate/worker.go |
| +++ b/internal/hostrelocate/worker.go |
| @@ -306,16 +306,17 @@ func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, erro |
| } |
| id := c.Op.ID |
| // Evidence first: a fenced participant changes what every phase may do. |
| + // A pass that refuses here checks the release deadline first (refused). |
| if src := c.Op.Intent["source_box"]; !slices.Contains(c.Lost, src) { |
| ref, record, err := w.fence(ctx, o, src) |
| if err != nil { |
| - return obs, err |
| + return w.refused(ctx, obs, err) |
| } |
| if ref == nil && c.Op.Journal["fence."+src] != "" { |
| // A converted move bound the source's fence before FenceSource |
| // consumed it. A bound fence that is no longer there is not |
| // "not fenced" (absence is never evidence): refuse the pass. |
| - return obs, fmt.Errorf("%w: %s's fence is bound in the journal but cannot be read", ErrRefused, src) |
| + return w.refused(ctx, obs, fmt.Errorf("%w: %s's fence is bound in the journal but cannot be read", ErrRefused, src)) |
| } |
| if ref != nil { |
| next, err := hostclaim.FenceSource(ctx, w.Claims, obs, id, *ref, w.Discharge(ctx, o.Roster.Boxes[src], record), w.Fences) |
| @@ -328,7 +329,7 @@ func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, erro |
| if dst := c.Op.Intent["target_box"]; dst != "" && !slices.Contains(c.Lost, dst) { |
| ref, record, err := w.fence(ctx, o, dst) |
| if err != nil { |
| - return obs, err |
| + return w.refused(ctx, obs, err) |
| } |
| if ref != nil { |
| return w.replace(ctx, obs, *ref, w.Discharge(ctx, o.Roster.Boxes[dst], record), record.ObservedAt) |
| @@ -466,6 +467,29 @@ func (w Worker) pinUnderLease(ctx context.Context, obs hostclaim.Observed) (host |
| return w.update(ctx, obs, hostclaim.JournalPin(obs.Claim.Op.ID, pin)) |
| } |
| +// refused is a pass that refuses with cause. The release deadline is |
| +// checked before the refusal stands: in releasing or probe-failed, |
| +// ReleaseDeadline after the phase began (ruling 14), the operation is |
| +// stuck instead, whether or not a release was journaled. So a refusal |
| +// that never clears, a source removed from the roster before its pin |
| +// (Task 7's ruling), leaves the operation stuck for an operator, never |
| +// refusing for ever. Only refusals (ErrRefused) count: any other error is |
| +// returned as it is, for a later pass to retry. |
| +func (w Worker) refused(ctx context.Context, obs hostclaim.Observed, cause error) (hostclaim.Observed, error) { |
| + sinceKey := map[string]string{hostclaim.PhaseReleasing: "releasing_since", hostclaim.PhaseProbeFailed: "probe_failed_since"}[obs.Claim.Op.Phase] |
| + if sinceKey == "" || !errors.Is(cause, ErrRefused) { |
| + return obs, cause |
| + } |
| + since, err := time.Parse(time.RFC3339Nano, obs.Claim.Op.Journal[sinceKey]) |
| + if err != nil { |
| + return obs, errors.Join(cause, err) |
| + } |
| + if w.Now().Sub(since) <= ReleaseDeadline { |
| + return obs, cause |
| + } |
| + return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, fmt.Sprintf("%s refused past the release deadline (%s): %v", obs.Claim.Op.Phase, ReleaseDeadline, cause))) |
| +} |
| + |
| // releaseOverdue marks the operation stuck when a live box has not |
| // journaled its release ReleaseDeadline after entering the phase whose |
| // start key names; before that it waits. |
diff --git a/internal/hostrelocate/worker_test.go b/internal/hostrelocate/worker_test.go| index 408aba755..84764ed25 100644 |
| --- a/internal/hostrelocate/worker_test.go |
| +++ b/internal/hostrelocate/worker_test.go |
| @@ -632,6 +632,53 @@ func TestWorkerMarksAnOverdueReleaseStuck(t *testing.T) { |
| } |
| } |
| +// Final review Minor 3 (Task 7's ruling): a source removed from the roster |
| +// makes every pass refuse at its fence read. The release deadline is |
| +// checked before that refusal stands, so the operation goes stuck at the |
| +// deadline, released or not, instead of refusing for ever. |
| +func TestWorkerMarksARosterRemovedSourceStuckAtTheReleaseDeadline(t *testing.T) { |
| + for _, journaled := range []bool{false, true} { |
| + t.Run(fmt.Sprintf("release journaled %v", journaled), func(t *testing.T) { |
| + f := newFixture(t) |
| + if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + f.ack("src") |
| + f.ack("dst") |
| + f.worker(nil) // releasing |
| + if journaled { |
| + h, id := f.claim().Claim.Holder, f.claim().Claim.Op.ID |
| + f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)})) |
| + } |
| + delete(f.roster.Boxes, "src") |
| + f.roster.Revision++ |
| + f.writeRoster() |
| + f.now = f.now.Add(ReleaseDeadline) |
| + if o := f.worker(ErrRefused); o.Claim.Op.Journal["stuck"] != "" { |
| + t.Fatal("stuck before the deadline") |
| + } |
| + f.now = f.now.Add(time.Second) |
| + o := f.worker(nil) |
| + if o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhaseReleasing || o.Claim.Op.Exec != "src" { |
| + t.Fatalf("a roster-removed source must leave the operation stuck at the release deadline: %+v", o.Claim.Op) |
| + } |
| + }) |
| + } |
| + // Outside releasing and probe-failed there is no release deadline: the |
| + // refusal stands. |
| + f := newFixture(t) |
| + if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + delete(f.roster.Boxes, "src") |
| + f.roster.Revision++ |
| + f.writeRoster() |
| + f.now = f.now.Add(ReleaseDeadline + time.Hour) |
| + if o := f.worker(ErrRefused); o.Claim.Op.Journal["stuck"] != "" || o.Claim.Op.Phase != hostclaim.PhaseStarting { |
| + t.Fatalf("starting: %+v", o.Claim.Op) |
| + } |
| +} |
| + |
| // Round 5 finding 1: a spare that owes an older local retirement for the |
| // host is never offered to the chooser; the worker replaces the dead |
| // target with a clean spare (the claim refuses the other in its CAS). |