amadan
host: a revert's probe has nine attempts, and its window's end survives a crash (Task 6 fix round 1)
The revert's probe gets the recovery probe's nine attempts as well as its fifteen minutes; an exhausted probe stops the revert with the recovery probe's sentence (MoveMarkStuck now takes it for a revert). The window's end is journalled (rv-probe-exhausted, with the record's updated_at) before the stuck mark, so a crash or a refused record write between them marks the move stuck on the next pass without probing again; only a retry, the record back from stuck, starts a new window.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+245
−30
internal/host/movesrevert.go+88 −21internal/host/movesrevert_test.go+145 −1internal/provision/move_test.go+4 −4internal/provision/provision.go+8 −4
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go| index d4b70c874..d57869f32 100644 |
| --- a/internal/host/movesrevert.go |
| +++ b/internal/host/movesrevert.go |
| @@ -55,10 +55,13 @@ import ( |
| // never changed). It records the inputs it renders before it renders them |
| // (MoveRecordRevertInputs). Never verify again: the probe's process may |
| // have written, and its writes are the authoritative state (INV-2). The |
| -// first attempt of a round of probing is journalled; fifteen minutes of |
| -// failed probes stop for an operator (StuckRevertProbeBudget), the |
| -// journalled start cleared first so a retry probes for fifteen minutes |
| -// again. Answered, the claim steps to probed. |
| +// probe has the recovery probe's nine attempts (§7.7), and the first |
| +// attempt of a window is journalled: nine failed attempts |
| +// (StuckRecoveryProbeAttempts) or fifteen minutes of failed probes |
| +// (StuckRevertProbeBudget) stop for an operator. The window's end is |
| +// journalled before the stuck mark, so a crash or a refused write between |
| +// them still ends in the mark, never in a new window; only a retry starts |
| +// one. Answered, the claim steps to probed. |
| // - probed: reopened_at on the record (MoveRevertReopened, J26), then the |
| // conditional open on hold_op = move id under the host's hold lock with |
| // the record read fresh (ReleaseHeld, a plain registry call, J12), then |
| @@ -108,10 +111,16 @@ const ( |
| // begins (revertJournalMark), so the sweep can tell it from a cut's that |
| // is still running, and knows the incarnation to park it under. |
| rvStepSource = "rv-source" |
| - // rvStepProbeSince is the first probe attempt of this round of probing |
| - // (RFC 3339), the start of the fifteen minutes. Cleared before a stuck |
| - // mark, so a retry starts again. |
| + // rvStepProbeSince is the first probe attempt of this window of probing |
| + // (RFC 3339), the start of the fifteen minutes. Cleared, with |
| + // rvStepProbeExhausted, only once a retry has returned the record from |
| + // stuck, so a retry starts a new window and nothing else does. |
| rvStepProbeSince = "rv-probe-since" |
| + // rvStepProbeExhausted is the window's end, journalled before the stuck |
| + // mark (probeExhaustion): the probe ran out of its nine attempts, or of |
| + // its fifteen minutes. A replay finds it and marks the move stuck without |
| + // probing again (Task 6 fix round 1). |
| + rvStepProbeExhausted = "rv-probe-exhausted" |
| // rvStepClosed is "done" once the record closed. |
| rvStepClosed = "rv-source-closed" |
| // rvStepJournalPark is parkLocalArtifacts' step: the retirement the |
| @@ -119,6 +128,19 @@ const ( |
| rvStepJournalPark = "rv-source-journal-park" |
| ) |
| +// probeExhaustion is rvStepProbeExhausted's value: the recorded reason the |
| +// window ended with, and the record's updated_at when it ended. SetRecordMove |
| +// stamps updated_at on every transition, and once the window has ended the |
| +// only transitions a reverting record takes are the stuck mark and then the |
| +// operator's retry (this box records no probe inputs and no reopen after it, |
| +// the door's abort refuses a reverting record, and MoveRevertClose needs the |
| +// reopen). So a record still reverting at that updated_at has not been marked |
| +// stuck, and one reverting at a later one has been retried. |
| +type probeExhaustion struct { |
| + Reason string `json:"reason"` |
| + UpdatedAt time.Time `json:"updated_at"` |
| +} |
| + |
| // revertJournalMark is rvStepSource's value. |
| type revertJournalMark struct { |
| Move string `json:"move"` |
| @@ -499,8 +521,12 @@ func (r *revertRun) verify(ctx context.Context, obs hostclaim.Observed) (bool, e |
| return true, nil |
| } |
| -// probing (§7.4 probing): the probe only, against a fifteen-minute budget |
| -// from its first attempt. Nothing earlier runs again from here. |
| +// probing (§7.4 probing): the probe only, bounded twice (§7.7): its nine |
| +// attempts and fifteen minutes from its first attempt. Nothing earlier runs |
| +// again from here. The window's end is journalled before the stuck mark, so |
| +// a crash or a refused record write between the two marks the move stuck on |
| +// the next pass without probing again; a new window starts only once a retry |
| +// has returned the record from stuck. |
| func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| cfg := r.cfg() |
| host, id := r.rec.Host, r.g.ID |
| @@ -511,6 +537,23 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, |
| if err != nil { |
| return false, err |
| } |
| + if raw, ok := j.Steps[rvStepProbeExhausted]; ok { |
| + var ex probeExhaustion |
| + if err := json.Unmarshal([]byte(raw), &ex); err != nil { |
| + return false, fmt.Errorf("move %s: the journalled end of the probe's window does not read: %w", id, err) |
| + } |
| + if r.rec.Move.UpdatedAt.Equal(ex.UpdatedAt) { |
| + // The window ended and the stuck mark never landed. |
| + return r.stuck(ctx, ex.Reason, errors.New("the probe's window ended before this pass")) |
| + } |
| + // Marked stuck and retried since: a new window. |
| + delete(j.Steps, rvStepProbeExhausted) |
| + delete(j.Steps, rvStepProbeSince) |
| + if err := j.Write(cfg.DataDir); err != nil { |
| + return false, err |
| + } |
| + log.Printf("moves: %s: the revert of %s was retried; its probe has a new window", host, id) |
| + } |
| since, perr := time.Parse(time.RFC3339Nano, j.Steps[rvStepProbeSince]) |
| if perr != nil { |
| since = r.now().UTC() |
| @@ -521,9 +564,9 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, |
| } |
| // A revert has no rollback: it probes the source's own generation, warm |
| // or cold as for serving (§7.14 reverting row), with no recovery |
| - // coverage and no attempt count (its budget is the fifteen minutes, |
| - // counted here). The probe records what it renders before it renders it. |
| - req := recoveryProbeRequest{Clean: true, SourceGeneration: obs.Claim.Holder.Generation, |
| + // coverage, and the recovery probe's nine attempts. The probe records |
| + // what it renders before it renders it. |
| + req := recoveryProbeRequest{Clean: true, SourceGeneration: obs.Claim.Holder.Generation, Attempts: recoveryProbeAttempts, |
| Record: func(ctx context.Context, in provision.MoveInputs) error { |
| _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRecordRevertInputs(r.g, in)) |
| return err |
| @@ -543,16 +586,40 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, |
| } |
| return true, nil |
| } |
| - if elapsed := r.now().Sub(since); elapsed >= revertProbeBudget { |
| - // The start goes first: a retry probes for fifteen minutes again. |
| - delete(j.Steps, rvStepProbeSince) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| - return false, err |
| - } |
| - return r.stuck(ctx, provision.StuckRevertProbeBudget, fmt.Errorf("no answer for %s", elapsed.Round(time.Second))) |
| + var reason string |
| + var why error |
| + elapsed := r.now().Sub(since) |
| + switch { |
| + case out == recoveryProbeExhausted: |
| + reason, why = provision.StuckRecoveryProbeAttempts, fmt.Errorf("%d attempts", recoveryProbeAttempts) |
| + case elapsed >= revertProbeBudget: |
| + reason, why = provision.StuckRevertProbeBudget, fmt.Errorf("no answer for %s", elapsed.Round(time.Second)) |
| + default: |
| + r.w.notice(host+"#rv-probe", "moves: %s: the revert's probe of %s has not answered yet", host, id) |
| + return false, nil |
| + } |
| + // The window's end first, against the record as it stands now (the |
| + // probe may have recorded its inputs this pass), then the stuck mark. |
| + fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host) |
| + if err != nil { |
| + return false, fmt.Errorf("re-read the record: %w", err) |
| + } |
| + if !r.same(fresh) { |
| + return false, nil |
| + } |
| + r.rec = *fresh |
| + body, err := json.Marshal(probeExhaustion{Reason: reason, UpdatedAt: fresh.Move.UpdatedAt}) |
| + if err != nil { |
| + return false, err |
| + } |
| + j.Mark(rvStepProbeExhausted, string(body)) |
| + if err := j.Write(cfg.DataDir); err != nil { |
| + return false, err |
| + } |
| + if err := r.fault("after-probe-exhausted"); err != nil { |
| + return false, err |
| } |
| - r.w.notice(host+"#rv-probe", "moves: %s: the revert's probe of %s has not answered yet", host, id) |
| - return false, nil |
| + return r.stuck(ctx, reason, why) |
| } |
| // probed (§7.4 probed): reopened_at on the record (J26), the conditional open |
diff --git a/internal/host/movesrevert_test.go b/internal/host/movesrevert_test.go| index 088158c84..dd8dfd456 100644 |
| --- a/internal/host/movesrevert_test.go |
| +++ b/internal/host/movesrevert_test.go |
| @@ -363,7 +363,7 @@ func TestRevertFromStoppingComputesEvidenceFirst(t *testing.T) { |
| t.Fatalf("%d probes, want 1", len(h.probes)) |
| } |
| req := h.probes[0] |
| - if req.RollbackID != "" || !req.Clean || req.Generation != "" || req.SourceGeneration != "cur" || req.Coverage != "" { |
| + if req.RollbackID != "" || !req.Clean || req.Generation != "" || req.SourceGeneration != "cur" || req.Coverage != "" || req.Attempts != recoveryProbeAttempts { |
| t.Fatalf("the probe request: %+v", req) |
| } |
| @@ -926,3 +926,147 @@ func TestRevertFromStoppingWithUnarchivableFiles(t *testing.T) { |
| t.Fatalf("%d verifications, %d probes", h.verifications(), len(h.probes)) |
| } |
| } |
| + |
| +// TestRevertProbeAttemptsExhaustedIsStuck (§7.7; Task 6 fix round 1): a |
| +// revert's probe has the recovery probe's nine attempts as well as the |
| +// fifteen minutes. A probe that uses them all stops the revert for an |
| +// operator with the recovery probe's sentence, long before fifteen minutes; |
| +// rev_phase stays probing and the row held. A retry probes again with nine |
| +// attempts and never verifies again. |
| +func TestRevertProbeAttemptsExhaustedIsStuck(t *testing.T) { |
| + h := newRevertHarness(t) |
| + h.fromStopped(t) |
| + h.arm(t) |
| + h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeExhausted, nil } |
| + h.pass(t) |
| + m := h.move(t) |
| + if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts || m.Reason != provision.RevertCutDeadline { |
| + t.Fatalf("not stuck on the exhausted probe: %+v", m) |
| + } |
| + if len(h.probes) != 1 || h.probes[0].Attempts != recoveryProbeAttempts { |
| + t.Fatalf("probes %+v, want one with %d attempts", h.probes, recoveryProbeAttempts) |
| + } |
| + if h.revPhase(t) != hostclaim.PhaseProbing || h.route.Status != "restoring" || h.route.HoldOp != p1ID { |
| + t.Fatalf("rev_phase %q row %+v", h.revPhase(t), h.route) |
| + } |
| + h.retry(t) |
| + h.probe = nil |
| + h.pass(t) |
| + h.reopened(t) |
| + if len(h.probes) != 2 || h.probes[1].Attempts != recoveryProbeAttempts || h.verifications() != 1 { |
| + t.Fatalf("after the retry: probes %+v, %d verifications", h.probes, h.verifications()) |
| + } |
| +} |
| + |
| +// failStuckStore fails the record write that would mark the move stuck, as a |
| +// store that refuses or loses the request does. |
| +type failStuckStore struct { |
| + release.Store |
| + fail bool |
| +} |
| + |
| +func (f *failStuckStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if f.fail && key == provision.RecordKey("bab", "app", p1Host) && strings.Contains(string(body), `"state":"stuck"`) { |
| + return "", errors.New("injected: the record write failed") |
| + } |
| + return f.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// TestRevertProbeExhaustionSurvivesACrash (Task 6 fix round 1): once the |
| +// revert's probe is exhausted (its attempts, or its fifteen minutes) the |
| +// exhaustion is journalled before the stuck mark. A crash between the two, or |
| +// a stuck mark the store refused, leaves the record reverting; the next pass |
| +// marks it stuck without probing again and never grants a fresh window. Only |
| +// a retry (the record back from stuck through MoveRetry) starts one. |
| +func TestRevertProbeExhaustionSurvivesACrash(t *testing.T) { |
| + for _, reason := range []string{provision.StuckRecoveryProbeAttempts, provision.StuckRevertProbeBudget} { |
| + for _, how := range []string{"crash", "failed record write"} { |
| + t.Run(reason+"/"+how, func(t *testing.T) { |
| + h := newRevertHarness(t) |
| + h.fromStopped(t) |
| + h.arm(t) |
| + exhausted := reason == provision.StuckRecoveryProbeAttempts |
| + // The probe records what it renders before each attempt, so |
| + // the record moves in the very pass the window ends. |
| + h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| + if err := req.Record(context.Background(), revertRendered()); err != nil { |
| + return recoveryProbeWaiting, err |
| + } |
| + if exhausted { |
| + return recoveryProbeExhausted, nil |
| + } |
| + return recoveryProbeWaiting, nil |
| + } |
| + if !exhausted { |
| + h.pass(t) // the window opens |
| + h.clock.Advance(16 * time.Minute) |
| + } |
| + probes := len(h.probes) |
| + fs := &failStuckStore{Store: h.s} |
| + if how == "crash" { |
| + h.rvAt = "after-probe-exhausted" |
| + } else { |
| + fs.fail = true |
| + } |
| + if err := movesPass(context.Background(), fs, h.w()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if how == "crash" && !h.rvDone { |
| + t.Fatalf("never reached after-probe-exhausted: %v", h.rvHits) |
| + } |
| + if m := h.move(t); m.State != provision.MoveReverting { |
| + t.Fatalf("the fixture's stuck mark landed: %+v", m) |
| + } |
| + if len(h.probes) != probes+1 { |
| + t.Fatalf("%d probes in the exhausting pass, want 1", len(h.probes)-probes) |
| + } |
| + // The next pass, on a restarted agent with the clock where it |
| + // was: stuck, with no probe and no fresh window. |
| + h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeAnswered, nil } |
| + h.freshWorker() |
| + h.pass(t) |
| + m := h.move(t) |
| + if m.State != provision.MoveStuck || m.StuckReason != reason { |
| + t.Fatalf("the next pass did not mark it stuck: %+v", m) |
| + } |
| + if len(h.probes) != probes+1 { |
| + t.Fatalf("the next pass probed again (%d probes)", len(h.probes)-probes) |
| + } |
| + // A retry starts a new window, and its probe answers. |
| + h.retry(t) |
| + h.pass(t) |
| + h.reopened(t) |
| + if h.verifications() != 1 { |
| + t.Fatalf("verified %d times, want once", h.verifications()) |
| + } |
| + }) |
| + } |
| + } |
| + // A retry's new window is a whole fifteen minutes, even with the earlier |
| + // exhaustion still journalled when the stuck mark landed. |
| + t.Run("the retry's window", func(t *testing.T) { |
| + h := newRevertHarness(t) |
| + h.fromStopped(t) |
| + h.arm(t) |
| + h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil } |
| + h.pass(t) |
| + h.clock.Advance(16 * time.Minute) |
| + h.pass(t) |
| + if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget { |
| + t.Fatalf("not stuck: %+v", m) |
| + } |
| + h.retry(t) |
| + h.freshWorker() |
| + h.pass(t) |
| + h.clock.Advance(14 * time.Minute) |
| + h.pass(t) |
| + if m := h.move(t); m.State != provision.MoveReverting { |
| + t.Fatalf("the retry's window was not fresh: %+v", m) |
| + } |
| + h.clock.Advance(2 * time.Minute) |
| + h.pass(t) |
| + if m := h.move(t); m.State != provision.MoveStuck { |
| + t.Fatalf("the retry's window never ended: %+v", m) |
| + } |
| + }) |
| +} |
diff --git a/internal/provision/move_test.go b/internal/provision/move_test.go| index 53038d40a..4a692bd50 100644 |
| --- a/internal/provision/move_test.go |
| +++ b/internal/provision/move_test.go |
| @@ -1215,7 +1215,10 @@ func TestStuckAndRetryReturnToThePhase(t *testing.T) { |
| // A revert sticks with no rollback journal and retries to reverting; its |
| // own cause survives both. |
| rec := &Record{Region: "ie", Move: &MoveState{ID: g.ID, Incarnation: g.Incarnation, From: "ie", To: "uk", State: MoveReverting, CutPhase: CutStaged, Reason: RevertCutDeadline}} |
| - for _, reason := range []string{StuckRevertProbeBudget, StuckBaselineMismatch} { |
| + // A revert's probe has the recovery probe's nine attempts as well as its |
| + // fifteen minutes (§7.7; Task 6 fix round 1), so both reasons are a |
| + // revert's. |
| + for _, reason := range []string{StuckRevertProbeBudget, StuckBaselineMismatch, StuckRecoveryProbeAttempts} { |
| if err := MoveMarkStuck(g, reason)(rec); err != nil || rec.Move.State != MoveStuck || rec.Move.StuckReason != reason || rec.Move.Reason != RevertCutDeadline || rec.Move.Rollback != nil { |
| t.Fatalf("revert %q: %v %+v", reason, err, rec.Move) |
| } |
| @@ -1223,9 +1226,6 @@ func TestStuckAndRetryReturnToThePhase(t *testing.T) { |
| t.Fatalf("revert retry: %v %+v", err, rec.Move) |
| } |
| } |
| - if err := MoveMarkStuck(g, StuckRecoveryProbeAttempts)(rec); !errors.Is(err, ErrMoveConflict) || rec.Move.State != MoveReverting { |
| - t.Fatalf("a revert with the recovery probe's reason: %v", err) |
| - } |
| // Neither a landing nor a committed move sticks or retries. |
| for _, state := range []string{MoveLanding, MoveCommitted, MoveCutting} { |
| rec := rollbackTestLanding(g, now, LandProbing) |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index ae41b2a1c..cbc74a44f 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -1795,8 +1795,9 @@ const ( |
| // what it recorded when it stopped (INV-4). A rollback's clean return |
| // at restoring or verified, or a revert's verify. |
| StuckBaselineMismatch = "the source no longer matches what it recorded when it stopped" |
| - // StuckRecoveryProbeAttempts: a rollback's recovery probe used every |
| - // attempt it was given without the app answering (§7.7, nine). |
| + // StuckRecoveryProbeAttempts: a recovery probe, a rollback's or a |
| + // revert's, used every attempt it was given without the app answering |
| + // (§7.7, nine). |
| StuckRecoveryProbeAttempts = "the recovery probe ran out of attempts" |
| // StuckRevertProbeBudget: a revert's probes failed for 15 minutes. |
| StuckRevertProbeBudget = "the source did not answer within 15 minutes of moving back" |
| @@ -2071,7 +2072,10 @@ func MoveRecordRecoveryInputs(g MoveGuard, rbID string, in MoveInputs) func(*Rec |
| // StuckReturnTargetFiles: a dirty rollback whose target is at returning |
| // (the source still waiting for it); the target phase is kept, so a |
| // retry runs the return again; |
| -// - StuckRevertProbeBudget: a revert. |
| +// - StuckRevertProbeBudget: a revert; |
| +// - StuckRecoveryProbeAttempts: a revert too, whose probe has the recovery |
| +// probe's nine attempts as well as its fifteen minutes (§7.7; Plan 4 |
| +// Task 6 fix round 1). |
| // |
| // A rollback records the source phase in StuckAt, which MoveRetry returns |
| // to; a revert has no Rollback journal and resumes from the claim's |
| @@ -2128,7 +2132,7 @@ func MoveMarkStuck(g MoveGuard, reason string) func(*Record) error { |
| } |
| } |
| case MoveReverting: |
| - ok = why == StuckBaselineMismatch || why == StuckRevertProbeBudget |
| + ok = why == StuckBaselineMismatch || why == StuckRevertProbeBudget || why == StuckRecoveryProbeAttempts |
| } |
| if !ok { |
| return fmt.Errorf("%w: the move cannot stop here for that reason", ErrMoveConflict) |