amadan

carlos / platform Public

provision: the offline hold, open, abort and finish advance move_seq and stamp move.seq

The transition table of self-serve moves plan 1 Task 7: OfflineStart,
OfflineStartStaged, OfflineOpenTransition, OfflineAbort from stopping
through landed and OfflineFinish of a move that went offline each set
move_seq + 1 and move.seq to it in the same mutation; staging, restage,
seal, sealed, flip, landed, finishing, an abort from staging and the
repeated abort leave both alone. A move_seq at MaxInt64 refuses with
ErrLifetimeExhausted and writes nothing.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude f50d05655745fc5b1a837e01134b1b749d319fde parent 80f5487
3 files changed, +264 −6
  • internal/provision/offline_move.go +51 −5
  • internal/provision/offline_move_test.go +206 −1
  • internal/provision/provision.go +7 −0
diff --git a/internal/provision/offline_move.go b/internal/provision/offline_move.go
index 8d87fb943..2ddc0dcbe 100644
--- a/internal/provision/offline_move.go
+++ b/internal/provision/offline_move.go
@@ -3,6 +3,7 @@ package provision
import (
"errors"
"fmt"
+ "math"
"slices"
"time"
)
@@ -23,7 +24,15 @@ import (
// No code in this file Puts a record. Each transition is a mut for
// SetRecordMove, so the write is provision's own read-CAS-write, which
// advances record_seq. Across a whole move the record changes only in
-// region, move, last_move, record_seq and move.updated_at.
+// region, move, last_move, record_seq, move_seq and move.updated_at.
+//
+// move_seq (self-serve moves plan 1 Task 7) advances on exactly the
+// transitions that change who serves the host: the hold (OfflineStart,
+// OfflineStartStaged), the target's open (OfflineOpenTransition), the
+// source's reopening (OfflineAbort from stopping through landed) and the
+// finish of a move that went offline (OfflineFinish). Each sets move.seq to
+// the new value in the same write, so a box writes the row's serving_seq from
+// the transition it applies. Every other transition leaves both alone.
// MoveKindOffline is MoveState.Kind for an offline operator move. Every other
// move leaves Kind empty.
@@ -88,6 +97,22 @@ const (
// SetRecordMove.
var errRecordUnchanged = errors.New("provision: the record already says this")
+// advanceMoveSeq is one move_seq advance, inside a transition's mutation:
+// move_seq + 1, and move.seq set to it while the record carries a move. A
+// count already at the largest value the console's int64 counters hold is
+// refused with ErrLifetimeExhausted, so the transition fails and writes
+// nothing. Called after the transition's own checks.
+func advanceMoveSeq(rec *Record) error {
+ if rec.MoveSeq >= math.MaxInt64 {
+ return fmt.Errorf("%s: move_seq %d: %w", rec.Host, rec.MoveSeq, ErrLifetimeExhausted)
+ }
+ rec.MoveSeq++
+ if rec.Move != nil {
+ rec.Move.Seq = rec.MoveSeq
+ }
+ return nil
+}
+
// Offline reports whether this is an offline operator move. Nil-safe, so a
// record at rest answers false.
func (m *MoveState) Offline() bool { return m != nil && m.Kind == MoveKindOffline }
@@ -187,8 +212,9 @@ func offlineBegin(id, from, to, by string, now time.Time, state string) func(*Re
if state == OfflineStaging {
t := now.UTC()
rec.Move.StagedAt = &t
+ return nil // a stage holds nothing
}
- return nil
+ return advanceMoveSeq(rec) // the direct start is the hold
}
}
@@ -208,6 +234,10 @@ func OfflineStartStaged(id, by string, now time.Time) func(*Record) error {
if now.IsZero() {
return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
}
+ // The staged start is the hold.
+ if err := advanceMoveSeq(rec); err != nil {
+ return err
+ }
t := now.UTC()
if s := rec.Move.StagedAt; s != nil && !t.After(*s) {
t = s.Add(time.Nanosecond)
@@ -346,6 +376,10 @@ func OfflineOpenTransition(id string, now time.Time) func(*Record) error {
if rec.Region != rec.Move.To {
return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
}
+ // The target serves.
+ if err := advanceMoveSeq(rec); err != nil {
+ return err
+ }
t := now.UTC()
rec.Move.State, rec.Move.OpenedAt = OfflineOpen, &t
return nil
@@ -380,12 +414,19 @@ func OfflineAbort(id string) func(*Record) error {
switch rec.Move.State {
case OfflineReverting:
return errRecordUnchanged
- case OfflineStaging, OfflineStopping, OfflineSealing, OfflineSealed:
- case OfflineLanding, OfflineLanded:
- rec.Region = rec.Move.From
+ case OfflineStaging:
+ // Never offline: nothing reopens, move_seq stays.
+ case OfflineStopping, OfflineSealing, OfflineSealed, OfflineLanding, OfflineLanded:
+ // The source reopens.
+ if err := advanceMoveSeq(rec); err != nil {
+ return err
+ }
default:
return fmt.Errorf("%w: %s", ErrMoveConflict, offlineAfterOpen(rec.Move.State))
}
+ if rec.Move.State == OfflineLanding || rec.Move.State == OfflineLanded {
+ rec.Region = rec.Move.From
+ }
rec.Move.State = OfflineReverting
return nil
}
@@ -422,6 +463,11 @@ func OfflineFinish(id string, now time.Time) func(*Record) error {
default:
return fmt.Errorf("%w: %s", ErrMoveConflict, offlineFinishState(m.State))
}
+ if m.StartedOffline() {
+ if err := advanceMoveSeq(rec); err != nil {
+ return err
+ }
+ }
if s := end.Sub(m.RequestedAt).Seconds(); s > 0 && m.StartedOffline() {
out.OfflineS = s
}
diff --git a/internal/provision/offline_move_test.go b/internal/provision/offline_move_test.go
index 238599cf9..1e52c760b 100644
--- a/internal/provision/offline_move_test.go
+++ b/internal/provision/offline_move_test.go
@@ -5,6 +5,7 @@ import (
"context"
"encoding/json"
"errors"
+ "math"
"reflect"
"strings"
"testing"
@@ -208,8 +209,13 @@ func TestOfflineRecordDiffersOnlyWhereTheMoveSaysSo(t *testing.T) {
t.Fatalf("record_seq %d -> %d", before.RecordSeq, after.RecordSeq)
}
+ // move_seq changes, by the three transitions that changed who
+ // serves the host: the hold, then open or the abort, then finish.
+ if after.MoveSeq != before.MoveSeq+3 {
+ t.Errorf("move_seq %d -> %d, want three advances", before.MoveSeq, after.MoveSeq)
+ }
afterMap := asMap(t, after)
- allowed := map[string]bool{"region": true, "move": true, "last_move": true, "record_seq": true}
+ allowed := map[string]bool{"region": true, "move": true, "last_move": true, "record_seq": true, "move_seq": true}
keys := map[string]bool{}
for k := range beforeMap {
keys[k] = true
@@ -853,3 +859,202 @@ func TestOfflineRefusalsAreTheRecordedSentences(t *testing.T) {
s, rec = fresh(t, 1)
refuse(t, s, rec, MoveGuardedCut(MoveGuard{ID: offID}), refusedOfflineKind)
}
+
+// moveSeqAt sets the record's move_seq to n, through the record writer, as a
+// record that has already been moved n times carries it. The write puts
+// record_seq into presence, so every later write advances it.
+func moveSeqAt(t *testing.T, s release.Store, rec Record, n uint64) {
+ t.Helper()
+ if _, err := updateRecord(WithRecordSequence(context.Background()), s, rec.Account, rec.App, rec.Host, func(r *Record) error {
+ r.MoveSeq = n
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// offlineStageSteps is the staged start: stage, one more round, then the
+// operator takes the instance offline.
+func offlineStageSteps(now time.Time) []offlineStep {
+ st := StageTarget{Stager: offSealer, Bucket: "data-uk", Region: "eu-west-2"}
+ return []offlineStep{
+ {"stage", func() func(*Record) error { return OfflineStage(offID, "", "uk", "ops@example.com", st, now) }, ""},
+ {"restage", func() func(*Record) error { return OfflineRestage(offID, now.Add(time.Minute)) }, OfflineStaging},
+ {"start staged", func() func(*Record) error {
+ return OfflineStartStaged(offID, "ops@example.com", now.Add(2*time.Minute))
+ }, OfflineStaging},
+ }
+}
+
+// The transition table of self-serve moves plan 1 Task 7: the hold, the
+// target's open, the source's reopening at an abort and the finish of a move
+// that went offline each advance move_seq by one and stamp move.seq with the
+// new value; every other transition leaves both alone. The record starts at
+// move_seq 4, a host moved before.
+func TestOfflineTransitionsAdvanceMoveSeq(t *testing.T) {
+ ctx := context.Background()
+ now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
+ steps, staged := offlineSteps(now), offlineStageSteps(now)
+ abort := offlineStep{"abort", func() func(*Record) error { return OfflineAbort(offID) }, ""}
+ finish := offlineStep{"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour)) }, ""}
+ cat := func(parts ...[]offlineStep) []offlineStep {
+ var out []offlineStep
+ for _, p := range parts {
+ out = append(out, p...)
+ }
+ return out
+ }
+ one := func(st offlineStep) []offlineStep { return []offlineStep{st} }
+ for _, c := range []struct {
+ name string
+ walk []offlineStep // from rest, before the transition under test
+ step offlineStep
+ advance bool
+ }{
+ {"start", nil, steps[0], true},
+ {"stage", nil, staged[0], false},
+ {"restage", staged[:1], staged[1], false},
+ {"start staged", staged[:2], staged[2], true},
+ {"seal", steps[:1], steps[1], false},
+ {"sealed by", steps[:2], steps[2], false},
+ {"flip", steps[:3], steps[3], false},
+ {"landed by", steps[:4], steps[4], false},
+ {"open", steps[:5], steps[5], true},
+ {"finishing", steps[:6], steps[6], false},
+ {"abort from staging", staged[:2], abort, false},
+ {"abort from stopping", steps[:1], abort, true},
+ {"abort from sealing", steps[:2], abort, true},
+ {"abort from sealed", steps[:3], abort, true},
+ {"abort from landing", steps[:4], abort, true},
+ {"abort from landed", steps[:5], abort, true},
+ {"abort from a staged start", staged, abort, true},
+ {"finish done", steps[:7], steps[7], true},
+ {"finish after an abort from stopping", cat(steps[:1], one(abort)), finish, true},
+ {"finish after an abort from a staged start", cat(staged, one(abort)), finish, true},
+ {"finish after an abort from staging", cat(staged[:2], one(abort)), finish, false},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec := offlineRecord(t, s)
+ moveSeqAt(t, s, rec, 4)
+ offlineWalk(t, s, rec, c.walk)
+ before := offlineGet(t, s, rec)
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, c.step.mut())
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := before.MoveSeq
+ if c.advance {
+ want++
+ }
+ stored := offlineGet(t, s, rec)
+ if got.MoveSeq != want || stored.MoveSeq != want {
+ t.Fatalf("move_seq %d -> returned %d, stored %d; want %d", before.MoveSeq, got.MoveSeq, stored.MoveSeq, want)
+ }
+ if c.advance && stored.RecordSeq <= before.RecordSeq {
+ t.Fatalf("the advance did not write: record_seq %d -> %d", before.RecordSeq, stored.RecordSeq)
+ }
+ switch {
+ case stored.Move == nil:
+ // finish clears the move; the record keeps the count.
+ case c.advance && stored.Move.Seq != want:
+ t.Fatalf("move.seq %d after an advance to %d", stored.Move.Seq, want)
+ case !c.advance && before.Move != nil && stored.Move.Seq != before.Move.Seq:
+ t.Fatalf("move.seq %d -> %d on a transition that does not advance", before.Move.Seq, stored.Move.Seq)
+ case !c.advance && before.Move == nil && stored.Move.Seq != 0:
+ t.Fatalf("a move created without the hold carries move.seq %d", stored.Move.Seq)
+ }
+ })
+ }
+
+ // The repeated abort is a success that writes nothing, move_seq
+ // included (TestOfflineAbort pins the bytes).
+ t.Run("abort from reverting", func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec := offlineRecord(t, s)
+ offlineWalk(t, s, rec, cat(steps[:5], one(abort)))
+ before := offlineGet(t, s, rec)
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if after := offlineGet(t, s, rec); got.MoveSeq != before.MoveSeq || !reflect.DeepEqual(after, before) {
+ t.Fatalf("a second abort moved the record: move_seq %d -> %d", before.MoveSeq, after.MoveSeq)
+ }
+ })
+
+ // A whole move from a host's first: the hold 1, open 2, finish 3; an
+ // abort from landed: the hold 1, the reopening 2, finish 3.
+ for _, c := range []struct {
+ name string
+ walk []offlineStep
+ seqs []uint64
+ }{
+ {"done", steps, []uint64{1, 1, 1, 1, 1, 2, 2, 3}},
+ {"aborted from landed", cat(steps[:5], one(abort), one(finish)), []uint64{1, 1, 1, 1, 1, 2, 3}},
+ } {
+ t.Run("from a first move, "+c.name, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec := offlineRecord(t, s)
+ for i, st := range c.walk {
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, st.mut())
+ if err != nil {
+ t.Fatalf("%s: %v", st.name, err)
+ }
+ if got.MoveSeq != c.seqs[i] || (got.Move != nil && got.Move.Seq != c.seqs[i]) {
+ t.Fatalf("%s: move_seq %d move %+v, want %d", st.name, got.MoveSeq, got.Move, c.seqs[i])
+ }
+ }
+ })
+ }
+}
+
+// A move_seq already at the largest value the console's counters hold makes
+// every advancing transition refuse with ErrLifetimeExhausted, and nothing is
+// written.
+func TestMoveSeqExhaustionAtEveryTransition(t *testing.T) {
+ ctx := context.Background()
+ now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
+ steps, staged := offlineSteps(now), offlineStageSteps(now)
+ abort := offlineStep{"abort", func() func(*Record) error { return OfflineAbort(offID) }, ""}
+ for _, c := range []struct {
+ name string
+ walk []offlineStep
+ step offlineStep
+ }{
+ {"start", nil, steps[0]},
+ {"start staged", staged[:2], staged[2]},
+ {"open", steps[:5], steps[5]},
+ {"abort from stopping", steps[:1], abort},
+ {"abort from landed", steps[:5], abort},
+ {"finish done", steps[:7], steps[7]},
+ {"finish aborted", append(append([]offlineStep(nil), steps[:3]...), abort), offlineStep{"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour)) }, ""}},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec := offlineRecord(t, s)
+ offlineWalk(t, s, rec, c.walk)
+ moveSeqAt(t, s, rec, math.MaxInt64)
+ key := RecordKey(rec.Account, rec.App, rec.Host)
+ rawBefore, etagBefore, err := s.Get(ctx, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, c.step.mut()); !errors.Is(err, ErrLifetimeExhausted) {
+ t.Fatalf("%s at move_seq MaxInt64 = %v, want ErrLifetimeExhausted", c.step.name, err)
+ }
+ rawAfter, etagAfter, err := s.Get(ctx, key)
+ if err != nil || !bytes.Equal(rawBefore, rawAfter) || etagBefore != etagAfter {
+ t.Fatalf("%s wrote the record on exhaustion: %v", c.step.name, err)
+ }
+ })
+ }
+ // One below the ceiling still advances, to the ceiling.
+ s := release.NewDirStore(t.TempDir())
+ rec := offlineRecord(t, s)
+ moveSeqAt(t, s, rec, math.MaxInt64-1)
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, steps[0].mut())
+ if err != nil || got.MoveSeq != math.MaxInt64 || got.Move.Seq != math.MaxInt64 {
+ t.Fatalf("start one below the ceiling: %v %+v", err, got)
+ }
+}
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index 8d6e46ef0..5b78f7939 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -1162,6 +1162,13 @@ type MoveState struct {
// Additive and omitempty.
StageBucket string `json:"stage_bucket,omitempty"`
StageRegion string `json:"stage_region,omitempty"`
+ // Seq is the record's move_seq as the transition that produced this
+ // state left it (self-serve moves plan 1 Task 7, offline_move.go): the
+ // hold's for a held row, open's for the target, the abort's for the
+ // reopened source. A box writes it on the row with the status, as the
+ // row's serving_seq. Additive and omitempty; 0 on a move whose
+ // transitions so far change nobody's serving (a staged move).
+ Seq uint64 `json:"seq,omitempty"`
}
// MoveInputs (what the target was checked against, §7.3 T3) is NOT