amadan

carlos / platform Public

provision, host: an explicit move retry grants the recovery probe a fresh nine-start budget, once (Plan 10 fix round 2)

MoveRetry advances an optional recovery_retry_epoch in the same record
transition that resumes recovery; the durable attempt key gains the
epoch, so a retry starts a new budget while restarts, replays and new
sessions keep the count. Epoch zero keeps the original key, so existing
journal counters survive an upgrade.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude ff72310fb1aba932906cc5dddd3f7cdb9b6f622a parent d41c412
5 files changed, +225 −51
  • internal/host/movesprobeattempts.go +51 −44
  • internal/host/movesprobeattempts_test.go +133 −4
  • internal/host/plan10_j2_adapter_test.go +1 −1
  • internal/provision/move_test.go +31 −0
  • internal/provision/provision.go +9 −2

CI failed — run details

diff --git a/internal/host/movesprobeattempts.go b/internal/host/movesprobeattempts.go
index 98dc5ae7f..32c793799 100644
--- a/internal/host/movesprobeattempts.go
+++ b/internal/host/movesprobeattempts.go
@@ -107,51 +107,53 @@ func (p *movesProbeAttempts) valid() bool {
// bound returns the landing record attempt, zero for a committed wake,
// or -1 for a recovery probe. Every case validates the admitted writer.
-func (p *movesProbeAttempts) bound(ctx context.Context, s OperationStart) (int, error) {
+// Recovery also returns the budget key from the same record read.
+func (p *movesProbeAttempts) bound(ctx context.Context, s OperationStart) (int, string, error) {
if !p.valid() {
- return 0, hostwrite.ErrUnavailable
+ return 0, "", hostwrite.ErrUnavailable
}
a := s.Admitted
if a.Claim != nil && claimRecoveryKind(a.Claim.Op.Kind) {
- return -1, p.boundRecovery(ctx, s)
+ key, err := p.boundRecovery(ctx, s)
+ return -1, key, err
}
if a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindLanding {
- return 0, hostbox.ErrUnsupported
+ return 0, "", hostbox.ErrUnsupported
}
if !s.Policy.Probe {
- return 0, p.boundCommitted(ctx, s)
+ return 0, "", p.boundCommitted(ctx, s)
}
h := a.Claim.Holder
if s.Owner.Host != a.Host || s.Owner.Session != h.Session || s.Owner.Stamp != h.Stamp || s.Owner.Incarnation != h.Incarnation {
- return 0, hostbox.ErrChanged
+ return 0, "", hostbox.ErrChanged
}
fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
if err != nil {
- return 0, err
+ return 0, "", err
}
if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseLandingProbing {
- return 0, hostbox.ErrChanged
+ return 0, "", hostbox.ErrChanged
}
rec, err := p.Records(ctx, h.Account, h.App, a.Host)
if err != nil {
- return 0, err
+ return 0, "", err
}
if rec == nil || rec.Host != a.Host || rec.Move == nil {
- return 0, hostbox.ErrChanged
+ return 0, "", hostbox.ErrChanged
}
m := rec.Move
if m.ID != a.Claim.Op.ID || m.Incarnation != h.Incarnation || m.State != provision.MoveLanding || m.LandingPhase != provision.LandProbing || m.Reason != "" || m.Landing == nil {
- return 0, hostbox.ErrChanged
+ return 0, "", hostbox.ErrChanged
}
admitted, err := p.Admitted(ctx, rec)
if err != nil {
- return 0, err
+ return 0, "", err
}
k := m.Landing.Attempts
if k < 1 || k > moves.AttemptsFor(moves.ProbeLanding, admitted) {
- return 0, hostbox.ErrChanged
+ return 0, "", hostbox.ErrChanged
}
- return k, nil
+ return k, "", nil
}
// boundCommitted is bound for a committed wake: the claim still names the
@@ -183,9 +185,9 @@ func (p *movesProbeAttempts) boundCommitted(ctx context.Context, s OperationStar
// startKey names a start's journal step: the record attempt k for a probe,
// the admitted session for a committed wake (k 0), or the recovery and
// admitted session for a recovery probe (k -1).
-func startKey(s OperationStart, k int) string {
+func startKey(s OperationStart, k int, recoveryKey string) string {
if k < 0 {
- return recoveryAttemptKey(s) + "-" + strconv.FormatUint(s.Owner.Session, 10)
+ return recoveryKey + "-" + strconv.FormatUint(s.Owner.Session, 10)
}
if k == 0 {
return "committed-" + strconv.FormatUint(s.Owner.Session, 10)
@@ -224,7 +226,7 @@ func bindSession(j *moves.Journal, s OperationStart, k int, write bool) error {
// OpenProbe journals the attempt's first start time, once.
func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, proposed time.Time) (time.Time, error) {
- k, err := p.bound(ctx, s)
+ k, recoveryKey, err := p.bound(ctx, s)
if err != nil {
return time.Time{}, err
}
@@ -243,7 +245,7 @@ func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, pr
if err := bindSession(j, s, k, true); err != nil {
return time.Time{}, err
}
- step := probeStepOpened + startKey(s, k)
+ step := probeStepOpened + startKey(s, k, recoveryKey)
if raw, ok := j.Steps[step]; ok {
if err := j.Write(p.DataDir); err != nil {
return time.Time{}, err
@@ -295,7 +297,7 @@ func sameStartOwner(a hostclaim.WakeRetryAttempt, s OperationStart, digest strin
// reserved by the first Reserve(0) of the admission and validated by every
// later one; start n > 0 only after start n-1 was settled failed.
func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n int) error {
- k, err := p.bound(ctx, s)
+ k, recoveryKey, err := p.bound(ctx, s)
if err != nil {
return err
}
@@ -312,7 +314,7 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
if err := bindSession(j, s, k, n == 0); err != nil {
return err
}
- st, ok, err := readStart(j, startKey(s, k))
+ st, ok, err := readStart(j, startKey(s, k, recoveryKey))
if err != nil {
return err
}
@@ -343,28 +345,28 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
}
if st.Attempt.ID == "" {
if k < 0 {
- count, err := recoveryAttemptCount(j, s)
+ count, err := recoveryAttemptCount(j, recoveryKey)
if err != nil {
return err
}
if count >= recoveryProbeAttempts {
return ErrRecoveryProbeExhausted
}
- j.Mark(recoveryAttemptKey(s), strconv.Itoa(count+1))
+ j.Mark(recoveryKey, strconv.Itoa(count+1))
}
o := s.Owner
st.Attempt = hostclaim.WakeRetryAttempt{ID: p.NewID(), Box: o.Box, Enrollment: o.Enrollment, BootID: o.BootID, ControllerID: o.ControllerID, Session: o.Session, InvocationID: p.NewID(), LaunchInputs: digest}
if st.Attempt.ID == "" || st.Attempt.InvocationID == "" || st.Attempt.ID == st.Attempt.InvocationID {
return hostwrite.ErrUnavailable
}
- if err := writeStart(p.DataDir, j, startKey(s, k), st); err != nil {
+ if err := writeStart(p.DataDir, j, startKey(s, k, recoveryKey), st); err != nil {
return err
}
}
if p.current == nil {
p.current = map[string]boundAttempt{}
}
- p.current[probeKey(s)] = boundAttempt{k: k, key: startKey(s, k), attempt: st.Attempt}
+ p.current[probeKey(s)] = boundAttempt{k: k, key: startKey(s, k, recoveryKey), attempt: st.Attempt}
return nil
}
@@ -385,7 +387,7 @@ func (p *movesProbeAttempts) CurrentAttempt(s OperationStart) hostclaim.WakeRetr
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
- if k, err := p.bound(ctx, s); err != nil || k != cur.k || startKey(s, k) != cur.key {
+ if k, recoveryKey, err := p.bound(ctx, s); err != nil || k != cur.k || startKey(s, k, recoveryKey) != cur.key {
return hostclaim.WakeRetryAttempt{}
}
return cur.attempt
@@ -399,7 +401,7 @@ func (p *movesProbeAttempts) Failed(ctx context.Context, s OperationStart, n int
}
func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int, state string) error {
- k, err := p.bound(ctx, s)
+ k, recoveryKey, err := p.bound(ctx, s)
if err != nil {
return err
}
@@ -412,7 +414,7 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
if err := bindSession(j, s, k, false); err != nil {
return err
}
- st, ok, err := readStart(j, startKey(s, k))
+ st, ok, err := readStart(j, startKey(s, k, recoveryKey))
if err != nil {
return err
}
@@ -421,11 +423,11 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
return hostbox.ErrChanged
}
st.State = state
- if err := writeStart(p.DataDir, j, startKey(s, k), st); err != nil {
+ if err := writeStart(p.DataDir, j, startKey(s, k, recoveryKey), st); err != nil {
return err
}
if k < 0 && state == probeFailed {
- count, err := recoveryAttemptCount(j, s)
+ count, err := recoveryAttemptCount(j, recoveryKey)
if err != nil {
return err
}
@@ -441,7 +443,7 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
// check again (again, the claim's ClaimProbeReadiness against the started
// role), with the process still alive after it (moves §7.7).
func (p *movesProbeAttempts) Healthy(ctx context.Context, s OperationStart, n int, cmd *exec.Cmd, again func(context.Context) error) error {
- k, err := p.bound(ctx, s)
+ k, _, err := p.bound(ctx, s)
if err != nil {
return err
}
@@ -501,54 +503,59 @@ func probeAlive(cmd *exec.Cmd) bool {
// RecoveryProbeAttemptsReady is checked before the adapter records inputs.
func (p *movesProbeAttempts) RecoveryProbeAttemptsReady() bool { return p.valid() }
-func (p *movesProbeAttempts) boundRecovery(ctx context.Context, s OperationStart) error {
+func (p *movesProbeAttempts) boundRecovery(ctx context.Context, s OperationStart) (string, error) {
a := s.Admitted
if a.Claim.Holder == nil || !s.Policy.Probe {
- return hostbox.ErrChanged
+ return "", hostbox.ErrChanged
}
h := a.Claim.Holder
if s.Owner.Host != a.Host || s.Owner.Incarnation != h.Incarnation || s.Owner.Stamp != h.Stamp || s.Owner.Session != h.Session || s.Owner.Box != a.Claim.Op.Exec {
- return hostbox.ErrChanged
+ return "", hostbox.ErrChanged
}
fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
if err != nil {
- return err
+ return "", err
}
if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseProbing || fresh.Claim.Op.Exec != s.Owner.Box {
- return hostbox.ErrChanged
+ return "", hostbox.ErrChanged
}
rec, err := p.Records(ctx, h.Account, h.App, a.Host)
if err != nil {
- return err
+ return "", err
}
if rec == nil || rec.Host != a.Host || rec.Move == nil || rec.Move.ID != a.Claim.Op.ID || rec.Move.Incarnation != h.Incarnation {
- return hostbox.ErrChanged
+ return "", hostbox.ErrChanged
}
m := rec.Move
if a.Claim.Op.Kind == hostclaim.KindReverting {
if m.State != provision.MoveReverting || m.ReopenedAt != nil || fresh.Claim.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing {
- return hostbox.ErrChanged
+ return "", hostbox.ErrChanged
}
} else if m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID == "" || m.Rollback.ID != a.Claim.Op.Intent["rollback"] || m.Rollback.ID != fresh.Claim.Op.Intent["rollback"] || m.Rollback.SourcePhase != provision.RbSourceProbing {
- return hostbox.ErrChanged
+ return "", hostbox.ErrChanged
}
- return nil
+ return recoveryAttemptKey(s, m.RecoveryRetryEpoch), nil
}
// The move has its own journal directory. JSON preserves the rollback and
// incarnation boundaries; sessions share this counter across controller restarts.
-func recoveryAttemptKey(s OperationStart) string {
+func recoveryAttemptKey(s OperationStart, retryEpoch uint64) string {
c := s.Admitted.Claim
rollback := c.Op.ID
if c.Op.Kind == hostclaim.KindRestoring {
rollback = c.Op.Intent["rollback"]
}
- raw, _ := json.Marshal([]string{c.Op.ID, rollback, c.Holder.Incarnation})
+ parts := []string{c.Op.ID, rollback, c.Holder.Incarnation}
+ // Keep epoch zero's key so an upgrade preserves existing counts.
+ if retryEpoch != 0 {
+ parts = append(parts, strconv.FormatUint(retryEpoch, 10))
+ }
+ raw, _ := json.Marshal(parts)
return "recovery_attempts_" + string(raw)
}
-func recoveryAttemptCount(j *moves.Journal, s OperationStart) (int, error) {
- raw, ok := j.Steps[recoveryAttemptKey(s)]
+func recoveryAttemptCount(j *moves.Journal, key string) (int, error) {
+ raw, ok := j.Steps[key]
if !ok {
return 0, nil
}
diff --git a/internal/host/movesprobeattempts_test.go b/internal/host/movesprobeattempts_test.go
index 179cefc81..1861e80fb 100644
--- a/internal/host/movesprobeattempts_test.go
+++ b/internal/host/movesprobeattempts_test.go
@@ -441,6 +441,135 @@ func TestRecoveryProbeAttemptsDurableBudget(t *testing.T) {
}
}
+func TestRecoveryProbeAttemptsOperatorRetryBudget(t *testing.T) {
+ for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
+ t.Run(kind, func(t *testing.T) {
+ ctx := t.Context()
+ f := recoveryPAFixture(t, kind)
+ f.rec.Schema, f.rec.Slug, f.rec.Channel = 1, "shop", "stable"
+ f.rec.CreatedAt = f.start.Admitted.Claim.Holder.CreatedAt
+ if err := provision.CreateRecord(ctx, f.claims, *f.rec); err != nil {
+ t.Fatal(err)
+ }
+ restart := func() {
+ f.ops = f.attempts()
+ f.ops.Records = func(ctx context.Context, account, app, host string) (*provision.Record, error) {
+ return provision.GetRecord(ctx, f.claims, account, app, host)
+ }
+ }
+ restart()
+ g := provision.MoveGuard{ID: f.rec.Move.ID, Incarnation: f.rec.Move.Incarnation}
+ transition := func(mut func(*provision.Record) error) error {
+ _, err := provision.SetRecordMove(ctx, f.claims, f.rec.Account, f.rec.App, paHost, mut)
+ return err
+ }
+ newSession := func() {
+ c := f.read().Claim
+ c.Gen++
+ c.Holder.Session = c.Gen
+ f.putClaim(c)
+ f.start.Admitted = f.read()
+ f.start.Owner.Session = c.Gen
+ restart()
+ }
+ var previous hostclaim.WakeRetryAttempt
+ for budget := 0; budget < 3; budget++ {
+ if budget > 0 {
+ if err := transition(provision.MoveMarkStuck(g, provision.StuckRecoveryProbeAttempts)); err != nil {
+ t.Fatal(err)
+ }
+ if err := transition(provision.MoveRetry(g)); err != nil {
+ t.Fatal(err)
+ }
+ if got := f.ops.CurrentAttempt(f.start); got.ID != "" {
+ t.Fatal("old reservation survived retry", got)
+ }
+ if err := f.ops.Failed(ctx, f.start, 2); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("old failure settled after retry: %v", err)
+ }
+ }
+ rec, err := provision.GetRecord(ctx, f.claims, f.rec.Account, f.rec.App, paHost)
+ if err != nil || rec.Move.RecoveryRetryEpoch != uint64(budget) {
+ t.Fatalf("retry epoch not persisted: record=%+v err=%v", rec, err)
+ }
+ for n := 0; n < 9; n++ {
+ local := n % 3
+ if local == 0 && n > 0 {
+ newSession()
+ }
+ if err := f.ops.Reserve(ctx, f.start, local); err != nil {
+ t.Fatalf("budget %d start %d: %v", budget, n+1, err)
+ }
+ attempt := f.ops.CurrentAttempt(f.start)
+ if attempt.ID == "" || attempt.ID == previous.ID {
+ t.Fatalf("start has no fresh reservation: %+v", attempt)
+ }
+ previous = attempt
+ if local == 0 {
+ restart()
+ if err := f.ops.Reserve(ctx, f.start, 0); err != nil || f.ops.CurrentAttempt(f.start) != attempt {
+ t.Fatalf("reservation replay after restart: %v", err)
+ }
+ }
+ if budget > 0 {
+ if err := transition(provision.MoveRetry(g)); !errors.Is(err, provision.ErrMoveConflict) {
+ t.Fatalf("retry replay: %v", err)
+ }
+ }
+ err := f.ops.Failed(ctx, f.start, local)
+ if n == 8 {
+ if !errors.Is(err, ErrRecoveryProbeExhausted) {
+ t.Fatalf("budget %d ninth failure: %v", budget, err)
+ }
+ } else if err != nil {
+ t.Fatal(err)
+ }
+ }
+ previousProvider := f.ops
+ restart()
+ if err := f.ops.Reserve(ctx, f.start, 3); !errors.Is(err, ErrRecoveryProbeExhausted) {
+ t.Fatalf("budget %d restart lost exhaustion: %v", budget, err)
+ }
+ f.ops = previousProvider
+ }
+ newSession()
+ if err := transition(provision.MoveRetry(g)); !errors.Is(err, provision.ErrMoveConflict) {
+ t.Fatalf("exhausted retry replay: %v", err)
+ }
+ if err := f.ops.Reserve(ctx, f.start, 0); !errors.Is(err, ErrRecoveryProbeExhausted) {
+ t.Fatalf("retry replay or fresh session granted extra starts: %v", err)
+ }
+ })
+ }
+}
+
+func TestRecoveryProbeAttemptsLegacyBudget(t *testing.T) {
+ for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
+ t.Run(kind, func(t *testing.T) {
+ f := recoveryPAFixture(t, kind)
+ rollback := paOp
+ if kind == hostclaim.KindRestoring {
+ rollback = plan10J2Rollback
+ }
+ raw, err := json.Marshal([]string{paOp, rollback, f.rec.Move.Incarnation})
+ if err != nil {
+ t.Fatal(err)
+ }
+ j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
+ if err != nil {
+ t.Fatal(err)
+ }
+ j.Mark("recovery_attempts_"+string(raw), "9")
+ if err := j.Write(f.dataDir); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.ops.Reserve(t.Context(), f.start, 0); !errors.Is(err, ErrRecoveryProbeExhausted) {
+ t.Fatalf("legacy exhaustion lost: %v", err)
+ }
+ })
+ }
+}
+
func TestRecoveryProbeAttemptsReadinessAndBinding(t *testing.T) {
f := recoveryPAFixture(t, hostclaim.KindRestoring)
w := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}, ProbeReady: func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }}
@@ -458,13 +587,13 @@ func TestRecoveryProbeAttemptsReadinessAndBinding(t *testing.T) {
if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil {
t.Fatal(err)
}
- oldKey := recoveryAttemptKey(f.start)
+ oldKey := recoveryAttemptKey(f.start, 0)
c := f.read().Claim
c.Op.Intent["rollback"] = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz"
f.putClaim(c)
f.start.Admitted = f.read()
f.rec.Move.Rollback.ID = c.Op.Intent["rollback"]
- if recoveryAttemptKey(f.start) == oldKey {
+ if recoveryAttemptKey(f.start, 0) == oldKey {
t.Fatal("rollback not bound")
}
if got := f.ops.CurrentAttempt(f.start); got.ID != "" {
@@ -477,7 +606,7 @@ func TestRecoveryProbeAttemptsReadinessAndBinding(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if j.Steps[oldKey] != "1" || j.Steps[recoveryAttemptKey(f.start)] != "1" {
+ if j.Steps[oldKey] != "1" || j.Steps[recoveryAttemptKey(f.start, 0)] != "1" {
t.Fatal("rollback budgets combined")
}
for _, mutate := range []func(*OperationStart){
@@ -490,7 +619,7 @@ func TestRecoveryProbeAttemptsReadinessAndBinding(t *testing.T) {
_ = json.Unmarshal(raw, &claim)
copied.Admitted.Claim = &claim
mutate(&copied)
- if recoveryAttemptKey(copied) == recoveryAttemptKey(f.start) {
+ if recoveryAttemptKey(copied, 0) == recoveryAttemptKey(f.start, 0) {
t.Fatal("budget key omitted identity")
}
}
diff --git a/internal/host/plan10_j2_adapter_test.go b/internal/host/plan10_j2_adapter_test.go
index 159aac957..5f22d9dbf 100644
--- a/internal/host/plan10_j2_adapter_test.go
+++ b/internal/host/plan10_j2_adapter_test.go
@@ -514,7 +514,7 @@ func TestPlan10J2NinthFailedStartSticksAndKeepsJournal(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if count, err := recoveryAttemptCount(j, last); err != nil || count != 9 {
+ if count, err := recoveryAttemptCount(j, recoveryAttemptKey(last, 0)); err != nil || count != 9 {
t.Fatalf("journal count=%d err=%v", count, err)
}
}
diff --git a/internal/provision/move_test.go b/internal/provision/move_test.go
index a30c96eb3..43ab35d84 100644
--- a/internal/provision/move_test.go
+++ b/internal/provision/move_test.go
@@ -1127,6 +1127,37 @@ func TestRollbackSourcePhasesCleanAndDirty(t *testing.T) {
}
}
+func TestMoveRetryEpochRefusals(t *testing.T) {
+ for _, rollback := range []bool{false, true} {
+ for _, epoch := range []uint64{0, 7, ^uint64(0)} {
+ g := cutTestGuard()
+ rec := &Record{Move: &MoveState{ID: g.ID, Incarnation: g.Incarnation,
+ State: MoveStuck, StuckReason: StuckRecoveryProbeAttempts, RecoveryRetryEpoch: epoch}}
+ if rollback {
+ rec.Move.Rollback = &MoveRollback{ID: rbTestID, StuckAt: RbSourceProbing}
+ }
+ before, _ := json.Marshal(rec)
+ err := MoveRetry(g)(rec)
+ if epoch == ^uint64(0) {
+ after, _ := json.Marshal(rec)
+ if !errors.Is(err, ErrMoveConflict) || string(before) != string(after) {
+ t.Fatalf("epoch overflow mutated record: %v %+v", err, rec.Move)
+ }
+ continue
+ }
+ if err != nil || rec.Move.RecoveryRetryEpoch != epoch+1 {
+ t.Fatalf("retry epoch: %v %+v", err, rec.Move)
+ }
+ before, _ = json.Marshal(rec)
+ err = MoveRetry(g)(rec)
+ after, _ := json.Marshal(rec)
+ if !errors.Is(err, ErrMoveConflict) || string(before) != string(after) {
+ t.Fatalf("retry replay mutated record: %v %+v", err, rec.Move)
+ }
+ }
+ }
+}
+
func TestStuckAndRetryReturnToThePhase(t *testing.T) {
g := cutTestGuard()
now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index 47c555ea6..bd81a4d22 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -994,6 +994,9 @@ type MoveState struct {
// revert's, a landing's mark) and Rollback.Reason the rollback's
// through a stuck and a retry.
StuckReason string `json:"stuck_reason,omitempty"`
+ // RecoveryRetryEpoch advances only on an operator retry. Recovery
+ // probe budgets use it so restarts and replays keep their spent starts.
+ RecoveryRetryEpoch uint64 `json:"recovery_retry_epoch,omitempty"`
// Relocation is the durable pre-flip conversion decision
// (move_relocation.go). It never installs anything.
@@ -2182,8 +2185,8 @@ func MoveMarkStuck(g MoveGuard, reason string) func(*Record) error {
// recovery it stopped. A rollback returns to rolling_back with the source
// at the phase StuckAt names, so a mismatch verifies again and a probe
// probes again; a revert (no Rollback) returns to reverting. It clears
-// Move.StuckReason and StuckAt only: the move's cause (Move.Reason) and the
-// rollback's (Rollback.Reason) stay.
+// Move.StuckReason and StuckAt and advances RecoveryRetryEpoch: the move's
+// cause (Move.Reason) and the rollback's (Rollback.Reason) stay.
func MoveRetry(g MoveGuard) func(*Record) error {
return func(rec *Record) error {
if err := g.check(rec); err != nil {
@@ -2193,6 +2196,9 @@ func MoveRetry(g MoveGuard) func(*Record) error {
if m.State != MoveStuck {
return fmt.Errorf("%w: %s", ErrMoveConflict, RefusedNotStuck)
}
+ if m.RecoveryRetryEpoch == ^uint64(0) {
+ return ErrMoveConflict
+ }
if rb := m.Rollback; rb != nil {
if rb.StuckAt == "" {
return fmt.Errorf("%w: the stuck rollback records no phase to return to", ErrMoveConflict)
@@ -2202,6 +2208,7 @@ func MoveRetry(g MoveGuard) func(*Record) error {
m.State = MoveReverting
}
m.StuckReason = ""
+ m.RecoveryRetryEpoch++
return nil
}
}