amadan

carlos / platform Public

provision, console, moves: tests for the queued state, receipts, stamps, the bound write and the open record

TestOfflineQueueThenStaging, TestOfflineQueueRefusesARecreatedLifetime,
TestOfflineQueuedRefusesEveryOtherVerb, TestOfflinePrecheckMovesOnlyTheStamp,
TestOfflineAbortAtClearsAQueuedMove, TestOfflineSetEstimateRefusesWithoutSelfServe,
TestReturnOfCarriedByStageAndStart, TestReceiptsOnEveryTransition,
TestProposeStampIsAfterEveryStamp, TestStampsAreMonotonicUnderABackwardClock,
TestSetRecordMoveAtIsBoundToTheObservedRecord and
TestOfflineSelfServeFieldsOnTheWire; the queued rows of
TestOfflineTransitionsAdvanceMoveSeq and the self-serve allowlist
(TestOfflineSelfServeRecordDiffersOnlyWhereTheMoveSaysSo); the operator
doors on a queued move (TestOfflineQueuedMoveAtTheOperatorDoors); the open
record's key and wire (TestOpenRecordKeyAndWire).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 774596f18e52805eec7687d5f3312f2687201004 parent d25d66f
4 files changed, +921 −3
  • internal/console/apiinstancemove_offline_queued_test.go +51 −0
  • internal/moves/offline_openrecord_test.go +56 −0
  • internal/provision/offline_move_test.go +93 −3
  • internal/provision/offline_queue_test.go +721 −0
diff --git a/internal/console/apiinstancemove_offline_queued_test.go b/internal/console/apiinstancemove_offline_queued_test.go
new file mode 100644
index 000000000..18dc0a230
--- /dev/null
+++ b/internal/console/apiinstancemove_offline_queued_test.go
@@ -0,0 +1,51 @@
+package console
+
+import (
+ "net/http"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// A queued self-serve move at the operator's doors (self-serve moves plan 2
+// Task 2): no operator verb moves it on, so every verb door answers MV-O25
+// and writes nothing; the status view says it can be aborted and was never
+// offline; and the abort door clears it at once, with nothing reopened and
+// no receipt written.
+func TestOfflineQueuedMoveAtTheOperatorDoors(t *testing.T) {
+ w := newOfflineWorld(t, oneEach...)
+ rec := w.record()
+ const id = "mv_abcdefghijklmnopqrstuvwxyz"
+ pin := provision.Participants{
+ Source: []provision.Participant{{BoxID: "ie-box", Label: "ie-box"}},
+ Target: provision.Participant{BoxID: "uk-box", Label: "uk-box"},
+ }
+ w.edit(provision.OfflineQueue(id, "ie", "uk", "owner@example.com", provision.StageTarget{Bucket: "data-uk"}, pin,
+ provision.SelfServeMove{ID: "ss_1", Incarnation: rec.Incarnation}, provision.ProposeStamp(rec, time.Now()), provision.Receipt{Nonce: "n1"}))
+ w.id = id
+ queued := w.record()
+
+ if offlineVerbOf(provision.OfflineQueued) != "" {
+ t.Fatalf("offline-queued waits on the %q door", offlineVerbOf(provision.OfflineQueued))
+ }
+ v := w.status()
+ if v.State != provision.OfflineQueued || v.ID != id || !v.CanAbort || !v.NeverOffline || v.Next != "" {
+ t.Fatalf("status of a queued move: %+v", v)
+ }
+ for _, verb := range []string{"/seal", "/flip", "/open", "/finish"} {
+ w.refused(http.MethodPost, verb, "", http.StatusConflict, msgO25(provision.OfflineQueued))
+ }
+
+ got := w.ok("/abort", "", "none")
+ if got.Last == nil || got.Last.ID != id || got.Last.Outcome != provision.OfflineOutcomeAborted || got.Last.OfflineS != 0 {
+ t.Fatalf("abort of a queued move: %+v", got.Last)
+ }
+ after := w.record()
+ if after.Move != nil || after.MoveSeq != queued.MoveSeq || after.Region != "ie" || after.LastMove.Request != "" {
+ t.Fatalf("after the abort: move %+v move_seq %d -> %d region %q last %+v", after.Move, queued.MoveSeq, after.MoveSeq, after.Region, after.LastMove)
+ }
+ if !after.LastMove.At.After(queued.Move.RequestedAt) {
+ t.Fatalf("the abort's stamp %v is not after the queue's %v", after.LastMove.At, queued.Move.RequestedAt)
+ }
+}
diff --git a/internal/moves/offline_openrecord_test.go b/internal/moves/offline_openrecord_test.go
new file mode 100644
index 000000000..70af28219
--- /dev/null
+++ b/internal/moves/offline_openrecord_test.go
@@ -0,0 +1,56 @@
+package moves
+
+import (
+ "encoding/json"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The open record (self-serve moves plan 2 Task 2): its key sits in the
+// move's prefix beside the seal, is never a report key, and refuses a host
+// or an id that could escape it; its body carries exactly these names.
+func TestOpenRecordKeyAndWire(t *testing.T) {
+ const host, id = "shop.example.com", "mv_abcdefghijklmnopqrstuvwxyz"
+ key := OpenRecordKey(host, id)
+ if key != "control/offline-moves/"+host+"/"+id+"/open.json" {
+ t.Fatalf("open record key = %q", key)
+ }
+ if !strings.HasPrefix(key, OfflineReportPrefix(host, id)) || key == OfflineSealKey(host, id) {
+ t.Fatalf("%s is not beside the seal in the move's prefix", key)
+ }
+ for _, role := range []string{OfflineRoleSource, OfflineRoleSealer, OfflineRoleTarget, OfflineRoleStager} {
+ if strings.HasPrefix(key, OfflineReportPrefix(host, id)+role+"/") {
+ t.Fatalf("%s is under the %s reports", key, role)
+ }
+ }
+ for _, bad := range [][2]string{{"", id}, {"not a host", id}, {host, ""}, {host, "../x"}, {host, "a/b"}, {host, ".."}} {
+ if OpenRecordKey(bad[0], bad[1]) != "" {
+ t.Errorf("open record key for %q accepted", bad)
+ }
+ }
+
+ at := time.Date(2026, 10, 22, 11, 0, 0, 5, time.UTC)
+ rec := OpenRecord{MoveID: id, Host: host, Account: "bab", App: "shop", Incarnation: 2, From: "ie", To: "uk",
+ StageBucket: "data-uk", StageRegion: "eu-west-2", Layout: "nested", OpenedAt: at,
+ Participants: &provision.Participants{Source: []provision.Participant{{BoxID: "i-0a", Label: "box-a"}}, Target: provision.Participant{BoxID: "i-0c", Label: "box-c"}}}
+ raw, err := json.Marshal(rec)
+ if err != nil {
+ t.Fatal(err)
+ }
+ const want = `{"move_id":"mv_abcdefghijklmnopqrstuvwxyz","host":"shop.example.com","account":"bab","app":"shop","incarnation":2,"from":"ie","to":"uk","stage_bucket":"data-uk","stage_region":"eu-west-2","participants":{"source":[{"box_id":"i-0a","label":"box-a"}],"target":{"box_id":"i-0c","label":"box-c"}},"layout":"nested","opened_at":"2026-10-22T11:00:00.000000005Z"}`
+ if string(raw) != want {
+ t.Fatalf("open record:\n got %s\nwant %s", raw, want)
+ }
+ var back OpenRecord
+ if err := json.Unmarshal(raw, &back); err != nil || !back.OpenedAt.Equal(at) || back.Participants == nil || back.Participants.Target.Label != "box-c" {
+ t.Fatalf("round trip: %v %+v", err, back)
+ }
+ // An open record of a move pinned to no participants leaves them off.
+ rec.Participants = nil
+ if raw, _ := json.Marshal(rec); strings.Contains(string(raw), "participants") {
+ t.Fatalf("nil participants on the wire: %s", raw)
+ }
+}
diff --git a/internal/provision/offline_move_test.go b/internal/provision/offline_move_test.go
index 673c6c719..61f44c1bc 100644
--- a/internal/provision/offline_move_test.go
+++ b/internal/provision/offline_move_test.go
@@ -250,6 +250,57 @@ func TestOfflineRecordDiffersOnlyWhereTheMoveSaysSo(t *testing.T) {
}
}
+// The allowlist holds for a self-serve move too: queued, the pre-copy, a
+// precheck and an estimate, then either the whole move or an abort while
+// queued. Across it the record differs only in region, move, last_move,
+// record_seq and move_seq; the self-serve fields live in the move and the
+// receipt of the clearing write in last_move.
+func TestOfflineSelfServeRecordDiffersOnlyWhereTheMoveSaysSo(t *testing.T) {
+ now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
+ tail := offlineSteps(now.Add(10 * time.Minute))[1:] // seal through finish
+ for _, c := range []struct {
+ name string
+ walk []offlineStep
+ advance uint64
+ region string
+ receipt string // last_move.request: the operator's finish carries none
+ }{
+ {"done", append(offlineQueueSteps(now), tail...), 3, "uk", ""},
+ {"aborted while queued", append(offlineQueueSteps(now)[:1], offlineStep{"abort at", func() func(*Record) error {
+ return OfflineAbortAt(offID, now.Add(time.Minute), Receipt{Nonce: "n-abort"})
+ }, ""}), 0, "", "n-abort"},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ rec := offlineRecord(t, s)
+ if _, err := updateRecord(WithRecordSequence(ctx), s, rec.Account, rec.App, rec.Host, func(*Record) error { return nil }); err != nil {
+ t.Fatal(err)
+ }
+ before := offlineGet(t, s, rec)
+ after := *offlineWalk(t, s, rec, c.walk)
+ if after.Move != nil || after.LastMove == nil || after.LastMove.Request != c.receipt {
+ t.Fatalf("the move did not end with receipt %q: %+v %+v", c.receipt, after.Move, after.LastMove)
+ }
+ if after.MoveSeq != before.MoveSeq+c.advance || after.Region != c.region || after.RecordSeq <= before.RecordSeq {
+ t.Fatalf("move_seq %d -> %d, region %q, record_seq %d -> %d", before.MoveSeq, after.MoveSeq, after.Region, before.RecordSeq, after.RecordSeq)
+ }
+ allowed := map[string]bool{"region": true, "move": true, "last_move": true, "record_seq": true, "move_seq": true}
+ b, a := asMap(t, before), asMap(t, offlineGet(t, s, rec))
+ for k := range a {
+ if _, ok := b[k]; !ok {
+ b[k] = nil
+ }
+ }
+ for k := range b {
+ if !reflect.DeepEqual(b[k], a[k]) && !allowed[k] {
+ t.Errorf("field %q changed: %v -> %v", k, b[k], a[k])
+ }
+ }
+ })
+ }
+}
+
// Every transition, tried from every state but its own, is refused with
// ErrMoveConflict and leaves the record as it was.
func TestOfflineEveryIllegalPairIsRefused(t *testing.T) {
@@ -886,16 +937,43 @@ func offlineStageSteps(now time.Time) []offlineStep {
}
}
+// offlineQueueSteps is the self-serve start: queue, the pre-copy, a
+// precheck and an estimate, then the job takes the instance offline.
+func offlineQueueSteps(now time.Time) []offlineStep {
+ st := StageTarget{Bucket: "data-uk", Region: "eu-west-2"}
+ p := Participants{Source: []Participant{{BoxID: "i-0a", Label: offSealer}}, Target: Participant{BoxID: "i-0c", Label: "box-c"}}
+ return []offlineStep{
+ {"queue", func() func(*Record) error {
+ return OfflineQueue(offID, "", "uk", "owner@example.com", st, p, SelfServeMove{ID: "ss_1"}, now, Receipt{Nonce: "n1"})
+ }, ""},
+ {"queued to staging", func() func(*Record) error {
+ return OfflineQueuedToStaging(offID, now.Add(time.Minute), Receipt{Nonce: "n2"})
+ }, OfflineQueued},
+ {"precheck", func() func(*Record) error {
+ return OfflinePrecheck(offID, now.Add(2*time.Minute), Receipt{Nonce: "n3"})
+ }, OfflineStaging},
+ {"estimate", func() func(*Record) error { return OfflineSetEstimate(offID, 600, Receipt{Nonce: "n4"}) }, OfflineStaging},
+ {"start staged", func() func(*Record) error {
+ return OfflineStartStaged(offID, "owner@example.com", now.Add(3*time.Minute), Receipt{Nonce: "n5"})
+ }, 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.
+// move_seq 4, a host moved before. Plan 2's queued steps (the queue, the
+// pre-copy's start, the precheck, the estimate and the queued abort) advance
+// nothing; the queue sets move.seq to the record's move_seq.
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)
+ steps, staged, queued := offlineSteps(now), offlineStageSteps(now), offlineQueueSteps(now)
abort := offlineStep{"abort", func() func(*Record) error { return OfflineAbort(offID, Receipt{}) }, ""}
+ abortAt := offlineStep{"abort at", func() func(*Record) error {
+ return OfflineAbortAt(offID, now.Add(time.Hour), Receipt{Nonce: "n-abort"})
+ }, ""}
finish := offlineStep{"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour), Receipt{}) }, ""}
cat := func(parts ...[]offlineStep) []offlineStep {
var out []offlineStep
@@ -932,6 +1010,16 @@ func TestOfflineTransitionsAdvanceMoveSeq(t *testing.T) {
{"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},
+ {"queue", nil, queued[0], false},
+ {"queued to staging", queued[:1], queued[1], false},
+ {"precheck", queued[:2], queued[2], false},
+ {"estimate", queued[:3], queued[3], false},
+ {"start staged after a queue", queued[:4], queued[4], true},
+ {"abort at from queued", queued[:1], abortAt, false},
+ {"abort at from a queued move's staging", queued[:2], abortAt, false},
+ {"abort at from a queued move's start", queued, abortAt, true},
+ {"finish after an abort from a queued move's staging", cat(queued[:2], one(abort)), finish, false},
+ {"finish after an abort from a queued move's start", cat(queued, one(abort)), finish, true},
} {
t.Run(c.name, func(t *testing.T) {
s := release.NewDirStore(t.TempDir())
@@ -961,7 +1049,9 @@ func TestOfflineTransitionsAdvanceMoveSeq(t *testing.T) {
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:
+ case !c.advance && before.Move == nil && stored.Move.State == OfflineQueued && stored.Move.Seq != before.MoveSeq:
+ t.Fatalf("a queued move carries move.seq %d, want the record's %d", stored.Move.Seq, before.MoveSeq)
+ case !c.advance && before.Move == nil && stored.Move.State != OfflineQueued && stored.Move.Seq != 0:
t.Fatalf("a move created without the hold carries move.seq %d", stored.Move.Seq)
}
})
diff --git a/internal/provision/offline_queue_test.go b/internal/provision/offline_queue_test.go
new file mode 100644
index 000000000..ce89be607
--- /dev/null
+++ b/internal/provision/offline_queue_test.go
@@ -0,0 +1,721 @@
+package provision
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The self-serve move's record side (self-serve moves plan 2 Task 2): the
+// offline-queued state, the receipts, the caller's stamps and the bound
+// write.
+
+var (
+ ssStage = StageTarget{Bucket: "data-uk", Region: "eu-west-2"}
+ ssPin = Participants{
+ Source: []Participant{{BoxID: "i-0a", Label: "box-a", InstanceID: "i-0a", Verified: true}, {BoxID: "i-0b", Label: "box-b"}},
+ Target: Participant{BoxID: "i-0c", Label: "box-c", InstanceID: "i-0c", Verified: true},
+ }
+)
+
+// ssRecord is offlineRecord at incarnation 2 and move_seq 4, with record_seq
+// present, as a host moved before carries them.
+func ssRecord(t *testing.T, s release.Store) Record {
+ t.Helper()
+ rec := offlineRecord(t, s)
+ if _, err := updateRecord(WithRecordSequence(context.Background()), s, rec.Account, rec.App, rec.Host, func(r *Record) error {
+ r.Incarnation, r.MoveSeq = 2, 4
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ return offlineGet(t, s, rec)
+}
+
+func ssMove(inc uint64) SelfServeMove {
+ return SelfServeMove{ID: "ss_1", Incarnation: inc, DisplayName: "Shop's account", RequestedBy: "owner@example.com"}
+}
+
+func ssQueue(at time.Time, req Receipt) func(*Record) error {
+ return OfflineQueue(offID, "", "uk", "owner@example.com", ssStage, ssPin, ssMove(2), at, req)
+}
+
+// apply is one transition through the record writer.
+func apply(t *testing.T, s release.Store, rec Record, mut func(*Record) error) *Record {
+ t.Helper()
+ got, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, mut)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return got
+}
+
+// rawRecord is the stored bytes and etag.
+func rawRecord(t *testing.T, s release.Store, rec Record) ([]byte, string) {
+ t.Helper()
+ raw, etag, err := s.Get(context.Background(), RecordKey(rec.Account, rec.App, rec.Host))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return raw, etag
+}
+
+func TestOfflineQueueThenStaging(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ // The caller's stamp, in another zone: stored as the same instant.
+ at := time.Date(2026, 10, 22, 10, 0, 0, 7, time.FixedZone("x", 3600))
+ got := apply(t, s, rec, ssQueue(at, Receipt{Nonce: "n-queue"}))
+ m := got.Move
+ if m == nil || m.State != OfflineQueued || !m.Offline() || m.ID != offID || m.From != "" || m.To != "uk" || m.RequestedBy != "owner@example.com" {
+ t.Fatalf("queued move: %+v", m)
+ }
+ if m.Stager != "box-a" || m.StageBucket != "data-uk" || m.StageRegion != "eu-west-2" || !reflect.DeepEqual(*m.Participants, ssPin) {
+ t.Fatalf("stage and participants: %+v %+v", m, m.Participants)
+ }
+ want := ssMove(2)
+ want.LastRequest = "n-queue"
+ if !reflect.DeepEqual(*m.SelfServe, want) {
+ t.Fatalf("self_serve %+v, want %+v", *m.SelfServe, want)
+ }
+ for name, p := range map[string]*time.Time{"requested_at": &m.RequestedAt, "queued_at": m.QueuedAt, "staged_at": m.StagedAt, "precheck_at": m.PrecheckAt} {
+ if p == nil || !p.Equal(at) {
+ t.Errorf("%s %v, want the caller's %v", name, p, at)
+ }
+ }
+ if m.StartedOffline() {
+ t.Error("a queued move reads as started")
+ }
+ if got.MoveSeq != 4 || m.Seq != 4 || got.Region != "" || got.RecordSeq != rec.RecordSeq+1 {
+ t.Fatalf("move_seq %d move.seq %d region %q record_seq %d -> %d", got.MoveSeq, m.Seq, got.Region, rec.RecordSeq, got.RecordSeq)
+ }
+ // The caller's slices are not the journal's.
+ if &m.Participants.Source[0] == &ssPin.Source[0] {
+ t.Fatal("the journal aliases the caller's participants")
+ }
+
+ // Queued to staging: staged_at is the new stamp, the rest stays.
+ at2 := at.Add(time.Minute)
+ got = apply(t, s, rec, OfflineQueuedToStaging(offID, at2, Receipt{Nonce: "n-stage"}))
+ m = got.Move
+ if m.State != OfflineStaging || !m.StagedAt.Equal(at2) || !m.RequestedAt.Equal(at) || !m.QueuedAt.Equal(at) || !m.PrecheckAt.Equal(at) || m.StartedOffline() {
+ t.Fatalf("queued to staging: %+v", m)
+ }
+ if got.MoveSeq != 4 || m.Seq != 4 || m.SelfServe.LastRequest != "n-stage" {
+ t.Fatalf("queued to staging: move_seq %d move.seq %d receipt %q", got.MoveSeq, m.Seq, m.SelfServe.LastRequest)
+ }
+ // And the staged start is the hold: offline from its stamp.
+ at3 := ProposeStamp(got, at2)
+ got = apply(t, s, rec, OfflineStartStaged(offID, "owner@example.com", at3, Receipt{Nonce: "n-start"}))
+ m = got.Move
+ if m.State != OfflineStopping || !m.StartedOffline() || !m.RequestedAt.Equal(at3) || m.SelfServe.OfflineAt == nil || !m.SelfServe.OfflineAt.Equal(at3) {
+ t.Fatalf("staged start: %+v %+v", m, m.SelfServe)
+ }
+ if got.MoveSeq != 5 || m.Seq != 5 {
+ t.Fatalf("staged start: move_seq %d move.seq %d, want 5", got.MoveSeq, m.Seq)
+ }
+}
+
+func TestOfflineQueueRefusesARecreatedLifetime(t *testing.T) {
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ ctx := context.Background()
+ refused := func(t *testing.T, s release.Store, rec Record, mut func(*Record) error, want string) {
+ t.Helper()
+ before, etag := rawRecord(t, s, rec)
+ _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut)
+ if !errors.Is(err, ErrMoveConflict) || strings.TrimPrefix(err.Error(), ErrMoveConflict.Error()+": ") != want {
+ t.Fatalf("err %v, want %q", err, want)
+ }
+ if after, etag2 := rawRecord(t, s, rec); !bytes.Equal(before, after) || etag != etag2 {
+ t.Fatal("a refused queue wrote the record")
+ }
+ }
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ for _, inc := range []uint64{0, 1, 3} {
+ refused(t, s, rec, OfflineQueue(offID, "", "uk", "owner@example.com", ssStage, ssPin, ssMove(inc), now, Receipt{Nonce: "n"}), "the instance was recreated since the move was requested")
+ }
+ // OfflineStage's refusals hold for the queue too.
+ refused(t, s, rec, OfflineQueue(offID, "", "", "o", ssStage, ssPin, ssMove(2), now, Receipt{}), "the instance is already in home")
+ refused(t, s, rec, OfflineQueue(offID, "us", "uk", "o", ssStage, ssPin, ssMove(2), now, Receipt{}), "the instance is in home, not us")
+ refused(t, s, rec, OfflineQueue("mv_short", "", "uk", "o", ssStage, ssPin, ssMove(2), now, Receipt{}), refusedMalformedStart)
+ refused(t, s, rec, OfflineQueue(offID, "", "uk", "o", ssStage, ssPin, ssMove(2), time.Time{}, Receipt{}), refusedMalformedStart)
+ refused(t, s, rec, OfflineQueue(offID, "", "uk", "o", StageTarget{Region: "eu-west-2"}, ssPin, ssMove(2), now, Receipt{}), refusedMalformedStart)
+ refused(t, s, rec, OfflineQueue(offID, "", "uk", "o", ssStage, Participants{Target: ssPin.Target}, ssMove(2), now, Receipt{}), refusedMalformedStart)
+ apply(t, s, rec, ssQueue(now, Receipt{}))
+ refused(t, s, rec, OfflineQueue(otherOff, "", "uk", "o", ssStage, ssPin, ssMove(2), now, Receipt{}), "a move to uk is already offline-queued")
+}
+
+// Nothing but the queued steps admits a queued move; every other transition
+// is refused with its recorded sentence and writes nothing.
+func TestOfflineQueuedRefusesEveryOtherVerb(t *testing.T) {
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{Nonce: "n"}))
+ later := now.Add(time.Hour)
+ for name, c := range map[string]struct {
+ mut func(*Record) error
+ want string
+ }{
+ "start staged": {OfflineStartStaged(offID, "o", later, Receipt{}), "this offline move is offline-queued. this step follows offline-staging"},
+ "restage": {OfflineRestage(offID, later, Receipt{}), "this offline move is offline-queued. this step follows offline-staging"},
+ "seal": {OfflineSeal(offID, offSealer, offBoxes, Receipt{}), "this offline move is offline-queued. this step follows offline-stopping"},
+ "sealed by": {OfflineSealedBy(offID, offSealer, MoveSource{}), "this offline move is offline-queued. this step follows offline-sealing"},
+ "flip": {OfflineFlip(offID, Receipt{}), "this offline move is offline-queued. this step follows offline-sealed"},
+ "landed by": {OfflineLandedBy(offID, "uk"), "this offline move is offline-queued. this step follows offline-landing"},
+ "open": {OfflineOpenTransition(offID, later, Receipt{}), "this offline move is offline-queued. this step follows offline-landed"},
+ "finishing": {OfflineFinishingTransition(offID, Receipt{}), "this offline move is offline-queued. this step follows offline-open"},
+ "finish": {OfflineFinish(offID, later, Receipt{}), "this offline move is offline-queued. it finishes from finishing or reverting"},
+ "abort": {OfflineAbort(offID, Receipt{}), "this offline move is offline-queued. this step follows offline-staging"},
+ "abort, receipted": {OfflineAbort(offID, Receipt{Nonce: "n2"}), "this offline move is offline-queued. this step follows offline-staging"},
+ "start": {OfflineStart(otherOff, "", "uk", "o", later, ""), "a move to uk is already offline-queued"},
+ "stage": {OfflineStage(otherOff, "", "uk", "o", StageTarget{Stager: "box-a", Bucket: "data-uk"}, later, ""), "a move to uk is already offline-queued"},
+ } {
+ before, etag := rawRecord(t, s, rec)
+ _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, c.mut)
+ if !errors.Is(err, ErrMoveConflict) || strings.TrimPrefix(err.Error(), ErrMoveConflict.Error()+": ") != c.want {
+ t.Errorf("%s from queued: %v, want %q", name, err, c.want)
+ }
+ if after, etag2 := rawRecord(t, s, rec); !bytes.Equal(before, after) || etag != etag2 {
+ t.Errorf("%s from queued wrote the record", name)
+ }
+ }
+ // The legacy and Phase 1 guards refuse the kind, as for every offline move.
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, MoveGuardedCut(MoveGuard{ID: offID})); !errors.Is(err, ErrMoveConflict) {
+ t.Errorf("the Phase 1 guard on a queued move: %v", err)
+ }
+}
+
+// The precheck moves precheck_at (and the receipt and updated_at) and nothing
+// else, in queued and in staging; every later state refuses it.
+func TestOfflinePrecheckMovesOnlyTheStamp(t *testing.T) {
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ for _, staged := range []bool{false, true} {
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{Nonce: "n1"}))
+ if staged {
+ apply(t, s, rec, OfflineQueuedToStaging(offID, now.Add(time.Minute), Receipt{Nonce: "n2"}))
+ }
+ before := offlineGet(t, s, rec)
+ at := now.Add(2 * time.Minute)
+ got := apply(t, s, rec, OfflinePrecheck(offID, at, Receipt{}))
+ if !got.Move.PrecheckAt.Equal(at) || got.RecordSeq != before.RecordSeq+1 {
+ t.Fatalf("precheck: %v record_seq %d -> %d", got.Move.PrecheckAt, before.RecordSeq, got.RecordSeq)
+ }
+ a, b := asMap(t, before), asMap(t, *got)
+ am, bm := a["move"].(map[string]any), b["move"].(map[string]any)
+ for _, k := range []string{"precheck_at", "updated_at"} {
+ delete(am, k)
+ delete(bm, k)
+ }
+ delete(a, "record_seq")
+ delete(b, "record_seq")
+ if !reflect.DeepEqual(a, b) {
+ t.Fatalf("staged %v: the precheck changed more than its stamp:\n%v\n%v", staged, a, b)
+ }
+ }
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{}))
+ apply(t, s, rec, OfflineQueuedToStaging(offID, now.Add(time.Minute), Receipt{}))
+ apply(t, s, rec, OfflineStartStaged(offID, "o", now.Add(2*time.Minute), Receipt{}))
+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflinePrecheck(offID, now.Add(3*time.Minute), Receipt{})); !errors.Is(err, ErrMoveConflict) ||
+ !strings.HasSuffix(err.Error(), "this offline move is offline-stopping. this step follows offline-staging") {
+ t.Fatalf("precheck from stopping: %v", err)
+ }
+}
+
+func TestOfflineAbortAtClearsAQueuedMove(t *testing.T) {
+ ctx := context.Background()
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{}))
+ at := now.Add(time.Minute)
+ got := apply(t, s, rec, OfflineAbortAt(offID, at, Receipt{Nonce: "n-abort"}))
+ want := &MoveOutcome{ID: offID, Outcome: OfflineOutcomeAborted, At: at, Request: "n-abort"}
+ if got.Move != nil || !reflect.DeepEqual(got.LastMove, want) || got.MoveSeq != 4 || got.Region != "" {
+ t.Fatalf("abort of a queued move: move %+v last %+v move_seq %d region %q", got.Move, got.LastMove, got.MoveSeq, got.Region)
+ }
+ if stored := offlineGet(t, s, rec); !reflect.DeepEqual(stored.LastMove, want) || stored.Move != nil {
+ t.Fatalf("stored: %+v %+v", stored.Move, stored.LastMove)
+ }
+
+ // From staging it is OfflineAbort: reverting, nothing reopens.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{}))
+ apply(t, s, rec, OfflineQueuedToStaging(offID, now.Add(time.Minute), Receipt{}))
+ got = apply(t, s, rec, OfflineAbortAt(offID, now.Add(2*time.Minute), Receipt{}))
+ if got.Move == nil || got.Move.State != OfflineReverting || got.MoveSeq != 4 || got.LastMove != nil {
+ t.Fatalf("abort at from staging: %+v move_seq %d last %+v", got.Move, got.MoveSeq, got.LastMove)
+ }
+ // A repeated abort on the reverting record is a success that writes
+ // nothing.
+ raw, etag := rawRecord(t, s, rec)
+ if _, err := SetRecordMove(WithRecordSequence(ctx), s, rec.Account, rec.App, rec.Host, OfflineAbortAt(offID, now.Add(3*time.Minute), Receipt{})); err != nil {
+ t.Fatal(err)
+ }
+ if raw2, etag2 := rawRecord(t, s, rec); !bytes.Equal(raw, raw2) || etag != etag2 {
+ t.Fatal("a repeated abort wrote the record")
+ }
+ // And the finish after it was never offline.
+ got = apply(t, s, rec, OfflineFinish(offID, now.Add(4*time.Minute), Receipt{}))
+ if got.LastMove == nil || got.LastMove.Outcome != OfflineOutcomeAborted || got.LastMove.OfflineS != 0 || got.MoveSeq != 4 {
+ t.Fatalf("finish: %+v move_seq %d", got.LastMove, got.MoveSeq)
+ }
+
+ // From stopping it is OfflineAbort too: the source reopens.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, OfflineStart(offID, "", "uk", "o", now, ""))
+ got = apply(t, s, rec, OfflineAbortAt(offID, now.Add(time.Minute), Receipt{}))
+ if got.Move.State != OfflineReverting || got.MoveSeq != 6 {
+ t.Fatalf("abort at from stopping: %+v move_seq %d", got.Move, got.MoveSeq)
+ }
+ // A zero stamp on a queued move is malformed.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{}))
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbortAt(offID, time.Time{}, Receipt{})); !errors.Is(err, ErrMoveConflict) {
+ t.Fatalf("abort at with no stamp: %v", err)
+ }
+}
+
+func TestOfflineSetEstimateRefusesWithoutSelfServe(t *testing.T) {
+ ctx := context.Background()
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ // An operator's staged move has no self_serve.
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ apply(t, s, rec, OfflineStage(offID, "", "uk", "o", StageTarget{Stager: "box-a", Bucket: "data-uk"}, now, ""))
+ raw, _ := rawRecord(t, s, rec)
+ _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSetEstimate(offID, 600, Receipt{}))
+ if !errors.Is(err, ErrMoveConflict) || !strings.HasSuffix(err.Error(), ": not a self-serve move") {
+ t.Fatalf("estimate on an operator move: %v", err)
+ }
+ if raw2, _ := rawRecord(t, s, rec); !bytes.Equal(raw, raw2) {
+ t.Fatal("a refused estimate wrote the record")
+ }
+
+ // On a self-serve move, queued or staging.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{Nonce: "n1"}))
+ got := apply(t, s, rec, OfflineSetEstimate(offID, 600, Receipt{}))
+ if got.Move.SelfServe.EstimateSeconds != 600 || got.Move.SelfServe.LastRequest != "n1" {
+ t.Fatalf("estimate: %+v", got.Move.SelfServe)
+ }
+ // The same estimate under an empty receipt writes nothing.
+ raw, etag := rawRecord(t, s, rec)
+ if _, err := SetRecordMove(WithRecordSequence(ctx), s, rec.Account, rec.App, rec.Host, OfflineSetEstimate(offID, 600, Receipt{})); err != nil {
+ t.Fatal(err)
+ }
+ if raw2, etag2 := rawRecord(t, s, rec); !bytes.Equal(raw, raw2) || etag != etag2 {
+ t.Fatal("an unchanged estimate wrote the record")
+ }
+ apply(t, s, rec, OfflineQueuedToStaging(offID, now.Add(time.Minute), Receipt{}))
+ got = apply(t, s, rec, OfflineSetEstimate(offID, 900, Receipt{}))
+ if got.Move.SelfServe.EstimateSeconds != 900 || got.Move.State != OfflineStaging {
+ t.Fatalf("estimate in staging: %+v", got.Move.SelfServe)
+ }
+ // After the start it is refused.
+ apply(t, s, rec, OfflineStartStaged(offID, "o", now.Add(2*time.Minute), Receipt{}))
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSetEstimate(offID, 1200, Receipt{})); !errors.Is(err, ErrMoveConflict) {
+ t.Fatalf("estimate from stopping: %v", err)
+ }
+}
+
+func TestReturnOfCarriedByStageAndStart(t *testing.T) {
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ const back = "mv_returnreturnreturnreturnr"
+ for name, mut := range map[string]func(string) func(*Record) error{
+ "start": func(r string) func(*Record) error { return OfflineStart(offID, "", "uk", "o", now, r) },
+ "stage": func(r string) func(*Record) error {
+ return OfflineStage(offID, "", "uk", "o", StageTarget{Stager: "box-a", Bucket: "data-uk"}, now, r)
+ },
+ } {
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ if got := apply(t, s, rec, mut(back)); got.Move.ReturnOf != back || offlineGet(t, s, rec).Move.ReturnOf != back {
+ t.Fatalf("%s: return_of %q", name, got.Move.ReturnOf)
+ }
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, mut(""))
+ if m := asMap(t, offlineGet(t, s, rec))["move"].(map[string]any); m["return_of"] != nil {
+ t.Fatalf("%s: an empty return_of is on the wire: %v", name, m["return_of"])
+ }
+ }
+}
+
+// ssSteps is a whole self-serve move, each step with its own nonce, and the
+// receipt field each step must leave.
+type ssStep struct {
+ name string
+ mut func(nonce string) func(*Record) error
+ clear bool // the step clears the move: the receipt is last_move.request
+}
+
+func ssSteps(now time.Time) []ssStep {
+ src := MoveSource{FinalTXID: 42}
+ at := func(m int) time.Time { return now.Add(time.Duration(m) * time.Minute) }
+ r := func(n string) Receipt { return Receipt{Nonce: n} }
+ return []ssStep{
+ {"queue", func(n string) func(*Record) error { return ssQueue(at(0), r(n)) }, false},
+ {"queued to staging", func(n string) func(*Record) error { return OfflineQueuedToStaging(offID, at(1), r(n)) }, false},
+ {"precheck", func(n string) func(*Record) error { return OfflinePrecheck(offID, at(2), r(n)) }, false},
+ {"estimate", func(n string) func(*Record) error { return OfflineSetEstimate(offID, 600, r(n)) }, false},
+ {"estimate again", func(n string) func(*Record) error { return OfflineSetEstimate(offID, 600, r(n)) }, false},
+ {"restage", func(n string) func(*Record) error { return OfflineRestage(offID, at(3), r(n)) }, false},
+ {"start staged", func(n string) func(*Record) error { return OfflineStartStaged(offID, "o", at(4), r(n)) }, false},
+ {"seal", func(n string) func(*Record) error { return OfflineSeal(offID, offSealer, offBoxes, r(n)) }, false},
+ {"sealed by", func(string) func(*Record) error { return OfflineSealedBy(offID, offSealer, src) }, false},
+ {"flip", func(n string) func(*Record) error { return OfflineFlip(offID, r(n)) }, false},
+ {"landed by", func(string) func(*Record) error { return OfflineLandedBy(offID, "uk") }, false},
+ {"open", func(n string) func(*Record) error { return OfflineOpenTransition(offID, at(5), r(n)) }, false},
+ {"finishing", func(n string) func(*Record) error { return OfflineFinishingTransition(offID, r(n)) }, false},
+ {"finish", func(n string) func(*Record) error { return OfflineFinish(offID, at(6), r(n)) }, true},
+ }
+}
+
+// receiptOf is the receipt the record carries: last_request while the move
+// is on it, last_move.request after it cleared.
+func receiptOf(rec *Record) string {
+ if rec.Move != nil {
+ if rec.Move.SelfServe == nil {
+ return ""
+ }
+ return rec.Move.SelfServe.LastRequest
+ }
+ if rec.LastMove != nil {
+ return rec.LastMove.Request
+ }
+ return ""
+}
+
+func TestReceiptsOnEveryTransition(t *testing.T) {
+ ctx := context.Background()
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ boxSteps := map[string]bool{"sealed by": true, "landed by": true}
+
+ // Each listed transition with a nonce writes it; the box's own steps
+ // carry none and leave the last one.
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ last := ""
+ for i, st := range ssSteps(now) {
+ nonce := "n" + string(rune('a'+i))
+ before := offlineGet(t, s, rec)
+ got := apply(t, s, rec, st.mut(nonce))
+ want := nonce
+ if boxSteps[st.name] {
+ want = last
+ }
+ if receiptOf(got) != want || receiptOf(ptr(offlineGet(t, s, rec))) != want {
+ t.Fatalf("%s: receipt %q, want %q", st.name, receiptOf(got), want)
+ }
+ if got.RecordSeq != before.RecordSeq+1 {
+ t.Fatalf("%s with a receipt did not write: record_seq %d -> %d", st.name, before.RecordSeq, got.RecordSeq)
+ }
+ if st.clear != (got.Move == nil) {
+ t.Fatalf("%s: move %+v", st.name, got.Move)
+ }
+ last = want
+ }
+
+ // An empty receipt leaves the receipt as it was, on every listed step.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, ssSteps(now)[0].mut("n-first"))
+ for _, st := range ssSteps(now)[1:] {
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, st.mut(""))
+ if err != nil {
+ t.Fatalf("%s: %v", st.name, err)
+ }
+ if st.clear {
+ if got.LastMove.Request != "" {
+ t.Fatalf("%s: an empty receipt wrote last_move.request %q", st.name, got.LastMove.Request)
+ }
+ continue
+ }
+ if got.Move.SelfServe.LastRequest != "n-first" {
+ t.Fatalf("%s: an empty receipt changed last_request to %q", st.name, got.Move.SelfServe.LastRequest)
+ }
+ }
+
+ // The abort path: the abort, the repeated abort on reverting (which a
+ // receipt makes a write), the finish.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ for _, st := range ssSteps(now)[:7] {
+ apply(t, s, rec, st.mut("n0"))
+ }
+ if got := apply(t, s, rec, OfflineAbort(offID, Receipt{Nonce: "n-abort"})); got.Move.State != OfflineReverting || receiptOf(got) != "n-abort" {
+ t.Fatalf("abort: %+v", got.Move)
+ }
+ before := offlineGet(t, s, rec)
+ got := apply(t, s, rec, OfflineAbort(offID, Receipt{Nonce: "n-abort-2"}))
+ if receiptOf(got) != "n-abort-2" || got.RecordSeq != before.RecordSeq+1 || got.MoveSeq != before.MoveSeq || got.Move.State != OfflineReverting {
+ t.Fatalf("the repeated abort with a receipt: %q record_seq %d -> %d", receiptOf(got), before.RecordSeq, got.RecordSeq)
+ }
+ got = apply(t, s, rec, OfflineAbortAt(offID, now.Add(time.Hour), Receipt{Nonce: "n-abort-3"}))
+ if receiptOf(got) != "n-abort-3" {
+ t.Fatalf("the repeated abort at with a receipt: %q", receiptOf(got))
+ }
+ if got := apply(t, s, rec, OfflineFinish(offID, now.Add(2*time.Hour), Receipt{Nonce: "n-finish"})); got.LastMove.Request != "n-finish" || got.LastMove.Outcome != OfflineOutcomeAborted {
+ t.Fatalf("finish after the abort: %+v", got.LastMove)
+ }
+
+ // The queued abort's receipt is last_move.request.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, ssQueue(now, Receipt{}))
+ if got := apply(t, s, rec, OfflineAbortAt(offID, now.Add(time.Minute), Receipt{Nonce: "n-q"})); got.LastMove.Request != "n-q" {
+ t.Fatalf("queued abort: %+v", got.LastMove)
+ }
+
+ // A receipt on a move the job did not queue is refused and writes
+ // nothing.
+ s = release.NewDirStore(t.TempDir())
+ rec = ssRecord(t, s)
+ apply(t, s, rec, OfflineStart(offID, "", "uk", "o", now, ""))
+ raw, _ := rawRecord(t, s, rec)
+ for name, mut := range map[string]func(*Record) error{
+ "seal": OfflineSeal(offID, offSealer, offBoxes, Receipt{Nonce: "n"}),
+ "abort": OfflineAbort(offID, Receipt{Nonce: "n"}),
+ "finish": OfflineFinish(offID, now, Receipt{Nonce: "n"}),
+ } {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); !errors.Is(err, ErrMoveConflict) {
+ t.Errorf("%s with a receipt on an operator move: %v", name, err)
+ }
+ }
+ if raw2, _ := rawRecord(t, s, rec); !bytes.Equal(raw, raw2) {
+ t.Fatal("a refused receipt wrote the record")
+ }
+}
+
+func ptr(r Record) *Record { return &r }
+
+// ProposeStamp is now, or one nanosecond after the latest stamp the record
+// carries, whichever is later, in UTC.
+func TestProposeStampIsAfterEveryStamp(t *testing.T) {
+ base := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ back := base.Add(-time.Hour)
+ at := func(m int) *time.Time { v := base.Add(time.Duration(m) * time.Minute); return &v }
+ if got := ProposeStamp(nil, back); !got.Equal(back) || got.Location() != time.UTC {
+ t.Fatalf("a nil record: %v", got)
+ }
+ if got := ProposeStamp(&Record{}, back); !got.Equal(back) {
+ t.Fatalf("a record at rest with no last move: %v", got)
+ }
+ for name, rec := range map[string]Record{
+ "last_move.at": {LastMove: &MoveOutcome{At: *at(9)}},
+ "requested_at": {Move: &MoveState{RequestedAt: *at(9), StagedAt: at(1)}},
+ "queued_at": {Move: &MoveState{RequestedAt: *at(1), QueuedAt: at(9)}},
+ "staged_at": {Move: &MoveState{RequestedAt: *at(1), StagedAt: at(9), PrecheckAt: at(2)}},
+ "precheck_at": {Move: &MoveState{RequestedAt: *at(1), StagedAt: at(2), PrecheckAt: at(9)}},
+ "opened_at": {Move: &MoveState{RequestedAt: *at(1), OpenedAt: at(9)}, LastMove: &MoveOutcome{At: *at(3)}},
+ "self_serve.offline_at": {Move: &MoveState{RequestedAt: *at(1), SelfServe: &SelfServeMove{OfflineAt: at(9)}}},
+ } {
+ want := at(9).Add(time.Nanosecond)
+ if got := ProposeStamp(&rec, back); !got.Equal(want) {
+ t.Errorf("%s latest: proposed %v, want %v", name, got, want)
+ }
+ if got := ProposeStamp(&rec, *at(9)); !got.Equal(want) {
+ t.Errorf("%s latest, the clock at it: proposed %v, want %v", name, got, want)
+ }
+ later := base.Add(time.Hour)
+ if got := ProposeStamp(&rec, later.In(time.FixedZone("x", 7200))); !got.Equal(later) || got.Location() != time.UTC {
+ t.Errorf("%s, a clock past it: proposed %v, want %v in UTC", name, got, later)
+ }
+ }
+}
+
+// Every writer proposes from the record it observed, so a clock that moved
+// back never writes an earlier stamp: an operator stage request after a job
+// restage, with the operator's clock an hour behind, still lands later, and
+// so does the start after it.
+func TestStampsAreMonotonicUnderABackwardClock(t *testing.T) {
+ ctx := context.Background()
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ // The job: queue, staging, a restage, each bound to what it observed.
+ job := func(mut func(at time.Time) func(*Record) error, clock time.Time) *Record {
+ t.Helper()
+ obs := offlineGet(t, s, rec)
+ got, err := SetRecordMoveAt(ctx, s, rec.Account, rec.App, rec.Host, Expect{Incarnation: obs.Incarnation, Seq: obs.RecordSeq}, mut(ProposeStamp(&obs, clock)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return got
+ }
+ job(func(at time.Time) func(*Record) error { return ssQueue(at, Receipt{Nonce: "q"}) }, now)
+ job(func(at time.Time) func(*Record) error { return OfflineQueuedToStaging(offID, at, Receipt{Nonce: "s"}) }, now.Add(time.Minute))
+ restaged := job(func(at time.Time) func(*Record) error { return OfflineRestage(offID, at, Receipt{Nonce: "r"}) }, now.Add(2*time.Minute))
+
+ // The operator, clock an hour back, proposes from its observation.
+ behind := now.Add(-time.Hour)
+ obs := offlineGet(t, s, rec)
+ at := ProposeStamp(&obs, behind)
+ got := apply(t, s, rec, OfflineRestage(offID, at, Receipt{}))
+ if !got.Move.StagedAt.After(*restaged.Move.StagedAt) || !got.Move.StagedAt.Equal(at) {
+ t.Fatalf("operator restage %v is not after the job's %v", got.Move.StagedAt, restaged.Move.StagedAt)
+ }
+ if got.Move.SelfServe.LastRequest != "r" {
+ t.Fatalf("an operator write confirmed a job request: %q", got.Move.SelfServe.LastRequest)
+ }
+ obs = offlineGet(t, s, rec)
+ got = apply(t, s, rec, OfflineStartStaged(offID, "ops@example.com", ProposeStamp(&obs, behind), Receipt{}))
+ if !got.Move.RequestedAt.After(*got.Move.StagedAt) || !got.Move.StartedOffline() {
+ t.Fatalf("operator start %v is not after staged_at %v", got.Move.RequestedAt, got.Move.StagedAt)
+ }
+}
+
+// countingStore counts Puts and can lose the race once: before forwarding
+// the first Put it writes the record through the inner store, so the Put's
+// etag is stale.
+type countingStore struct {
+ release.Store
+ puts int
+ race func()
+ raced bool
+}
+
+func (c *countingStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ c.puts++
+ if c.race != nil && !c.raced {
+ c.raced = true
+ c.race()
+ }
+ return c.Store.Put(ctx, key, body, etag)
+}
+
+func TestSetRecordMoveAtIsBoundToTheObservedRecord(t *testing.T) {
+ now := time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC)
+ for name, ctx := range map[string]context.Context{
+ "no judge": context.Background(),
+ "under a judge": WithLifetimes(context.Background(), NewLifetimeJudge()),
+ } {
+ t.Run(name, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec := ssRecord(t, s)
+ if name == "under a judge" {
+ // The judge reads a record of a lifetime it knows; the
+ // legacy lifetime 0 is one.
+ if _, err := updateRecord(context.Background(), s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Incarnation = 0; return nil }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ obs := offlineGet(t, s, rec)
+ exp := Expect{Incarnation: obs.Incarnation, Seq: obs.RecordSeq}
+ calls := 0
+ counted := func(mut func(*Record) error) func(*Record) error {
+ return func(r *Record) error { calls++; return mut(r) }
+ }
+ refused := func(t *testing.T, st release.Store, e Expect, mut func(*Record) error) {
+ t.Helper()
+ raw, etag := rawRecord(t, s, rec)
+ calls = 0
+ _, err := SetRecordMoveAt(ctx, st, rec.Account, rec.App, rec.Host, e, counted(mut))
+ if !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("err %v, want ErrCASConflict", err)
+ }
+ if calls != 0 {
+ t.Fatalf("the mutation ran %d times on a refused expectation", calls)
+ }
+ if raw2, etag2 := rawRecord(t, s, rec); !bytes.Equal(raw, raw2) || etag != etag2 {
+ t.Fatal("a refused bound write changed the record")
+ }
+ }
+ start := OfflineStart(offID, "", "uk", "o", now, "")
+ // A sequence that moved on, a recreated lifetime with the same
+ // sequence, both: refused before the mutation runs.
+ refused(t, s, Expect{Incarnation: exp.Incarnation, Seq: exp.Seq - 1}, start)
+ refused(t, s, Expect{Incarnation: exp.Incarnation, Seq: exp.Seq + 1}, start)
+ refused(t, s, Expect{Incarnation: exp.Incarnation + 1, Seq: exp.Seq}, start)
+
+ // The observed record: one write, record_seq advanced once.
+ got, err := SetRecordMoveAt(ctx, s, rec.Account, rec.App, rec.Host, exp, counted(start))
+ if err != nil || calls != 1 || got.RecordSeq != exp.Seq+1 || got.Move == nil || got.Move.State != OfflineStopping || got.Move.UpdatedAt.IsZero() {
+ t.Fatalf("bound write: %v calls %d %+v", err, calls, got)
+ }
+ if stored := offlineGet(t, s, rec); stored.RecordSeq != got.RecordSeq || stored.Move.State != OfflineStopping {
+ t.Fatalf("stored %+v", stored)
+ }
+ // Replaying the same bound write is refused: the sequence moved.
+ refused(t, s, exp, start)
+
+ // A lost race is ErrCASConflict with no fresh read and no
+ // retry: the mutation ran once, one Put was attempted, and the
+ // record is what the racer wrote.
+ obs = offlineGet(t, s, rec)
+ exp = Expect{Incarnation: obs.Incarnation, Seq: obs.RecordSeq}
+ cs := &countingStore{Store: s}
+ cs.race = func() {
+ if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Steering = "pinned"; return nil }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ calls = 0
+ _, err = SetRecordMoveAt(ctx, cs, rec.Account, rec.App, rec.Host, exp, counted(OfflineSeal(offID, offSealer, offBoxes, Receipt{})))
+ if !errors.Is(err, release.ErrCASConflict) || calls != 1 || cs.puts != 1 {
+ t.Fatalf("lost race: %v calls %d puts %d", err, calls, cs.puts)
+ }
+ if after := offlineGet(t, s, rec); after.Steering != "pinned" || after.Move.State != OfflineStopping {
+ t.Fatalf("after the lost race: steering %q state %s", after.Steering, after.Move.State)
+ }
+
+ // A mutation that changes nothing writes nothing.
+ obs = offlineGet(t, s, rec)
+ exp = Expect{Incarnation: obs.Incarnation, Seq: obs.RecordSeq}
+ apply(t, s, rec, OfflineAbort(offID, Receipt{}))
+ obs = offlineGet(t, s, rec)
+ raw, etag := rawRecord(t, s, rec)
+ if _, err := SetRecordMoveAt(ctx, s, rec.Account, rec.App, rec.Host, Expect{Incarnation: obs.Incarnation, Seq: obs.RecordSeq}, OfflineAbort(offID, Receipt{})); err != nil {
+ t.Fatal(err)
+ }
+ if raw2, etag2 := rawRecord(t, s, rec); !bytes.Equal(raw, raw2) || etag != etag2 {
+ t.Fatal("an unchanged bound write wrote the record")
+ }
+ })
+ }
+}
+
+// The self-serve fields are additive: a journal without them round-trips
+// byte for byte (TestOfflineFieldsAreAdditiveOnTheWire), and one with them
+// carries exactly these names.
+func TestOfflineSelfServeFieldsOnTheWire(t *testing.T) {
+ const queued = `{"record_seq":3,"schema":1,"host":"shop.example.com","slug":"shop","app":"shop","account":"bab","region":"ie","channel":"stable","created_at":"2026-09-01T08:00:00Z","move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","from":"ie","to":"uk","state":"offline-queued","requested_by":"owner@example.com","requested_at":"2026-10-22T10:00:00Z","updated_at":"2026-10-22T10:00:00Z","kind":"offline","staged_at":"2026-10-22T10:00:00Z","stager":"box-a","stage_bucket":"data-uk","stage_region":"eu-west-2","seq":4,"self_serve":{"id":"ss_1","incarnation":2,"display_name":"Shop's account","requested_by":"owner@example.com","estimate_seconds":600,"offline_at":"2026-10-22T10:05:00Z","last_request":"n1"},"participants":{"source":[{"box_id":"i-0a","label":"box-a","instance_id":"i-0a","verified":true}],"target":{"box_id":"i-0c","label":"box-c"}},"queued_at":"2026-10-22T10:00:00Z","precheck_at":"2026-10-22T10:00:00Z","return_of":"mv_zyxwvutsrqponmlkjihgfedcba"},"incarnation":2,"move_seq":4}`
+ rec, err := DecodeRecord([]byte(queued))
+ if err != nil {
+ t.Fatal(err)
+ }
+ m := rec.Move
+ if m.State != OfflineQueued || m.SelfServe == nil || m.SelfServe.LastRequest != "n1" || m.SelfServe.EstimateSeconds != 600 || m.Participants == nil || m.Participants.Target.Label != "box-c" ||
+ m.QueuedAt == nil || m.PrecheckAt == nil || m.ReturnOf != otherOff {
+ t.Fatalf("self-serve journal: %+v", m)
+ }
+ if out, _ := json.Marshal(rec); string(out) != queued {
+ t.Errorf("self-serve journal did not round-trip:\n got %s\nwant %s", out, queued)
+ }
+ const cleared = `{"schema":1,"host":"shop.example.com","slug":"shop","app":"shop","account":"bab","region":"uk","channel":"stable","created_at":"2026-09-01T08:00:00Z","last_move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","outcome":"done","at":"2026-10-22T11:00:00Z","offline_s":60,"request":"n9"}}`
+ rec, err = DecodeRecord([]byte(cleared))
+ if err != nil || rec.LastMove.Request != "n9" {
+ t.Fatalf("cleared: %v %+v", err, rec.LastMove)
+ }
+ if out, _ := json.Marshal(rec); string(out) != cleared {
+ t.Errorf("last_move.request did not round-trip:\n got %s\nwant %s", out, cleared)
+ }
+}