amadan

carlos / platform Public

provision: the offline-queued state, the self-serve fields, receipts and caller-proposed stamps

MoveState gains self_serve, participants, queued_at, precheck_at and
return_of; last_move gains request. OfflineQueue creates a self-serve move
in offline-queued (move_seq unchanged, move.seq = move_seq, refused for a
recreated lifetime); OfflineQueuedToStaging, OfflinePrecheck,
OfflineSetEstimate and OfflineAbortAt are the queued steps. Every
transition the job writes takes a Receipt as its final parameter: a
non-empty nonce lands in self_serve.last_request (or last_move.request on a
clearing write) and is never a no-op; the operator doors pass Receipt{}.
OfflineStage and OfflineStart take returnOf.

Stamps are the caller's: the in-mutation clamp of OfflineStartStaged and
OfflineRestage is retired for ProposeStamp, max(now, the record's latest
stamp + 1 ns). A staged start not after staged_at is refused rather than
moved. The operator doors propose from the record their write reads until
Task 3 binds them to the record they observed; the abort door uses
OfflineAbortAt so a queued move clears. Every existing caller is updated
mechanically.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 7ceefeabf4448025e5a2724d42dc3ed3c81adcfd parent 5aad903
15 files changed, +611 −206
  • internal/console/apiinstancemove_offline.go +49 −13
  • internal/console/apiinstancemove_offline_test.go +7 −7
  • internal/console/claim_invariant_fixtures_test.go +4 −4
  • internal/console/lifetimes_e2e_test.go +7 −7
  • internal/host/movesoffline_e2e_test.go +1 −1
  • internal/host/movesoffline_lifetime_test.go +7 −7
  • internal/host/movesoffline_seal_test.go +1 −1
  • internal/host/movesoffline_stage_test.go +15 −9
  • internal/host/movesoffline_test.go +19 −7
  • internal/provision/offline_move.go +333 −51
  • internal/provision/offline_move_copy_test.go +4 −4
  • internal/provision/offline_move_test.go +52 −52
  • internal/provision/offline_stage_test.go +50 −34
  • internal/provision/provision.go +53 −0
  • internal/provision/reconcile_lifetime_test.go +9 −9
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index 103ac3927..3b2c34b09 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -628,11 +628,15 @@ func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Re
writeRefusal(w, ref)
return
}
- s.offlineAdvance(w, r, d, provision.OfflineStartStaged(mv.ID, d.addr, time.Now()))
+ s.offlineAdvance(w, r, d, offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStartStaged(mv.ID, d.addr, at, provision.Receipt{})
+ }))
return
}
s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error {
- return provision.OfflineStart(id, d.rec.Region, to, d.addr, time.Now())
+ return offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStart(id, d.rec.Region, to, d.addr, at, "")
+ })
})
}
@@ -660,7 +664,9 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re
writeRefusal(w, ref)
return
}
- s.offlineAdvance(w, r, d, provision.OfflineRestage(mv.ID, time.Now()))
+ s.offlineAdvance(w, r, d, offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineRestage(mv.ID, at, provision.Receipt{})
+ }))
return
}
for _, b := range append(fleet.boxesAt(d.rec.Region), fleet.boxesAt(to)...) {
@@ -691,7 +697,9 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re
}
st := provision.StageTarget{Stager: moves.OfflineStager(names), Bucket: tgt.DataBucket, Region: tgt.DataBucketRegion}
s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error {
- return provision.OfflineStage(id, d.rec.Region, to, d.addr, st, time.Now())
+ return offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStage(id, d.rec.Region, to, d.addr, st, at, "")
+ })
})
}
@@ -879,6 +887,9 @@ const (
// wait for a box, not for the operator.
func offlineVerbOf(state string) string {
switch state {
+ case provision.OfflineQueued:
+ // The self-serve job moves a queued move on; no operator verb does.
+ return ""
case provision.OfflineStaging:
return offlineVerbStart
case provision.OfflineStopping:
@@ -900,7 +911,19 @@ type offlineStep struct {
// offlineOpenMut is the open transition. The one place the door names it.
func offlineOpenMut(id string, now time.Time) func(*provision.Record) error {
- return provision.OfflineOpenTransition(id, now)
+ return offlineStamped(now, func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineOpenTransition(id, at, provision.Receipt{})
+ })
+}
+
+// offlineStamped is a stamped operator transition whose stamp is proposed
+// from the record the write reads: provision.ProposeStamp, max(now, the
+// record's latest stamp + 1 ns), so a clock that went back never writes an
+// earlier stamp. The operator doors pass the empty receipt.
+func offlineStamped(now time.Time, mut func(at time.Time) func(*provision.Record) error) func(*provision.Record) error {
+ return func(rec *provision.Record) error {
+ return mut(provision.ProposeStamp(rec, now))(rec)
+ }
}
// offlineStepFor runs the checks of the verb the move's state waits for,
@@ -925,7 +948,9 @@ func (s *Server) offlineStepFor(ctx context.Context, d offlineDoor, mv *provisio
if ref := s.offlineStageCoords(fleet, mv, host); ref != nil {
return offlineStep{}, ref
}
- return offlineStep{mut: provision.OfflineStartStaged(mv.ID, d.addr, now)}, nil
+ return offlineStep{mut: offlineStamped(now, func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStartStaged(mv.ID, d.addr, at, provision.Receipt{})
+ })}, nil
case provision.OfflineStopping:
fleet, ok := s.readOfflineFleet(ctx)
if !ok {
@@ -935,12 +960,12 @@ func (s *Server) offlineStepFor(ctx context.Context, d offlineDoor, mv *provisio
if ref != nil {
return offlineStep{}, ref
}
- return offlineStep{mut: provision.OfflineSeal(mv.ID, sealer, boxes)}, nil
+ return offlineStep{mut: provision.OfflineSeal(mv.ID, sealer, boxes, provision.Receipt{})}, nil
case provision.OfflineSealed:
if ref := s.offlineFlipCheck(ev, mv, host, from, now); ref != nil {
return offlineStep{}, ref
}
- return offlineStep{mut: provision.OfflineFlip(mv.ID)}, nil
+ return offlineStep{mut: provision.OfflineFlip(mv.ID, provision.Receipt{})}, nil
case provision.OfflineLanded:
if ref := s.offlineOpenCheck(ev, mv, host, from, now); ref != nil {
return offlineStep{}, ref
@@ -950,12 +975,12 @@ func (s *Server) offlineStepFor(ctx context.Context, d offlineDoor, mv *provisio
if ref := offlineFinishCheck(ev, mv, host, moves.OfflinePhaseOpened, "", moves.OfflinePhaseParked, mv.Boxes); ref != nil {
return offlineStep{}, ref
}
- return offlineStep{mut: provision.OfflineFinishingTransition(mv.ID)}, nil
+ return offlineStep{mut: provision.OfflineFinishingTransition(mv.ID, provision.Receipt{})}, nil
case provision.OfflineFinishing:
if ref := offlineFinishCheck(ev, mv, host, "", moves.OfflinePhaseLifted, moves.OfflinePhaseGone, mv.Boxes); ref != nil {
return offlineStep{}, ref
}
- return offlineStep{mut: provision.OfflineFinish(mv.ID, now)}, nil
+ return offlineStep{mut: offlineFinishMut(mv.ID, now)}, nil
case provision.OfflineReverting:
// Every box the seal named, and the sealer. A move aborted before
// the seal named neither: the boxes are the source region's as the
@@ -983,11 +1008,18 @@ func (s *Server) offlineStepFor(ctx context.Context, d offlineDoor, mv *provisio
return offlineStep{}, offlineStateRefusal(mv.State, host)
}
}
- return offlineStep{mut: provision.OfflineFinish(mv.ID, now)}, nil
+ return offlineStep{mut: offlineFinishMut(mv.ID, now)}, nil
}
return offlineStep{}, offlineStateRefusal(mv.State, host)
}
+// offlineFinishMut is the finish transition, its stamp proposed.
+func offlineFinishMut(id string, now time.Time) func(*provision.Record) error {
+ return offlineStamped(now, func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineFinish(id, at, provision.Receipt{})
+ })
+}
+
// offlineSealCheck: every source box reports this move held, fenced and
// quiesced, and its edges file no longer lists the host. It names the sealer
// and the boxes the later doors wait for.
@@ -1227,7 +1259,11 @@ func (s *Server) handleAPIInstanceOfflineAbort(w http.ResponseWriter, r *http.Re
writeJSON(w, http.StatusAccepted, s.offlineViewOf(r.Context(), d.rec))
return
}
- s.offlineAdvance(w, r, d, provision.OfflineAbort(mv.ID))
+ // OfflineAbortAt: a queued move is cleared at once (it holds nothing),
+ // every other state is OfflineAbort's.
+ s.offlineAdvance(w, r, d, offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineAbortAt(mv.ID, at, provision.Receipt{})
+ }))
}
// ---- status ----
@@ -1360,7 +1396,7 @@ func (s *Server) offlineViewOf(ctx context.Context, rec *provision.Record) offli
v.OpenedAt = mv.OpenedAt.Unix()
}
switch mv.State {
- case provision.OfflineStaging, provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded:
+ case provision.OfflineQueued, provision.OfflineStaging, provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded:
v.CanAbort = true
}
ev, err := s.readOfflineEvidence(ctx, rec.Host, mv.ID)
diff --git a/internal/console/apiinstancemove_offline_test.go b/internal/console/apiinstancemove_offline_test.go
index 7b6d0c144..ba7162c75 100644
--- a/internal/console/apiinstancemove_offline_test.go
+++ b/internal/console/apiinstancemove_offline_test.go
@@ -1069,7 +1069,7 @@ func TestOfflineAbortDoor(t *testing.T) {
w3.toLanded()
w3.ok("/open", "", provision.OfflineOpen)
w3.refused(http.MethodPost, "/abort", "", http.StatusConflict, offlineHost+" is already served from uk. move it back with a new move")
- w3.edit(provision.OfflineFinishingTransition(w3.id))
+ w3.edit(provision.OfflineFinishingTransition(w3.id, provision.Receipt{}))
w3.refused(http.MethodPost, "/abort", "", http.StatusConflict, offlineHost+" is already served from uk. move it back with a new move")
}
@@ -1180,11 +1180,11 @@ func TestOfflineConflictMapsOntoRecordedSentences(t *testing.T) {
mut func(*provision.Record) error
want string
}{
- {"wrong state", provision.OfflineFlip(w.id), "this offline move is offline-stopping. this step follows offline-sealed"},
- {"another move's id", provision.OfflineSeal(other, "ie-box", []string{"ie-box"}), "the record carries a different offline move"},
- {"a seal without its sealer among the boxes", provision.OfflineSeal(w.id, "ie-x", []string{"ie-box"}),
+ {"wrong state", provision.OfflineFlip(w.id, provision.Receipt{}), "this offline move is offline-stopping. this step follows offline-sealed"},
+ {"another move's id", provision.OfflineSeal(other, "ie-box", []string{"ie-box"}, provision.Receipt{}), "the record carries a different offline move"},
+ {"a seal without its sealer among the boxes", provision.OfflineSeal(w.id, "ie-x", []string{"ie-box"}, provision.Receipt{}),
"a seal names the box that seals and the boxes that stopped the instance. the box that seals is one of them"},
- {"an abort already past", provision.OfflineFinish(w.id, time.Now()), "this offline move is offline-stopping. it finishes from finishing or reverting"},
+ {"an abort already past", provision.OfflineFinish(w.id, time.Now(), provision.Receipt{}), "this offline move is offline-stopping. it finishes from finishing or reverting"},
}
for _, c := range cases {
err := apply(c.mut)
@@ -1206,8 +1206,8 @@ func TestOfflineConflictMapsOntoRecordedSentences(t *testing.T) {
w.stopped(nil, "ie-box")
w.ok("/flip", "", provision.OfflineLanding)
w.landed("uk-box")
- w.edit(provision.OfflineOpenTransition(w.id, time.Now()))
- err := apply(provision.OfflineAbort(w.id))
+ w.edit(provision.OfflineOpenTransition(w.id, time.Now(), provision.Receipt{}))
+ err := apply(provision.OfflineAbort(w.id, provision.Receipt{}))
if got := offlineConflict(err); got != "this offline move is offline-open. it cannot be aborted once it is open" {
t.Errorf("abort after open = %q", got)
}
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go
index 02ea1539d..97d08a452 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -565,18 +565,18 @@ func (cx *claimCase) offlineMove(t *testing.T, state string) {
t.Helper()
cx.movableInstance(t)
now := time.Now()
- steps := []func(*provision.Record) error{provision.OfflineStart(claimOfflineMoveID, "ie", "uk", claimFxOwner, now)}
+ steps := []func(*provision.Record) error{provision.OfflineStart(claimOfflineMoveID, "ie", "uk", claimFxOwner, now, "")}
switch state {
case provision.OfflineStopping:
case provision.OfflineSealed, provision.OfflineLanded, provision.OfflineOpen:
steps = append(steps,
- provision.OfflineSeal(claimOfflineMoveID, "ie-box", []string{"ie-box"}),
+ provision.OfflineSeal(claimOfflineMoveID, "ie-box", []string{"ie-box"}, provision.Receipt{}),
provision.OfflineSealedBy(claimOfflineMoveID, "ie-box", provision.MoveSource{}))
if state != provision.OfflineSealed {
- steps = append(steps, provision.OfflineFlip(claimOfflineMoveID), provision.OfflineLandedBy(claimOfflineMoveID, "uk"))
+ steps = append(steps, provision.OfflineFlip(claimOfflineMoveID, provision.Receipt{}), provision.OfflineLandedBy(claimOfflineMoveID, "uk"))
}
if state == provision.OfflineOpen {
- steps = append(steps, provision.OfflineOpenTransition(claimOfflineMoveID, now))
+ steps = append(steps, provision.OfflineOpenTransition(claimOfflineMoveID, now, provision.Receipt{}))
}
default:
t.Fatalf("offlineMove: no staging for %s", state)
diff --git a/internal/console/lifetimes_e2e_test.go b/internal/console/lifetimes_e2e_test.go
index 37599927a..ce134e1d0 100644
--- a/internal/console/lifetimes_e2e_test.go
+++ b/internal/console/lifetimes_e2e_test.go
@@ -41,14 +41,14 @@ func (l ltOn) moveThrough(t *testing.T, srv *Server, host, id, to string) provis
now := l.clock.Now()
var rec *provision.Record
for _, mut := range []func(*provision.Record) error{
- provision.OfflineStart(id, from, to, "paul@keymail.dev", now),
- provision.OfflineSeal(id, "ie-a", []string{"ie-a"}),
+ provision.OfflineStart(id, from, to, "paul@keymail.dev", now, ""),
+ provision.OfflineSeal(id, "ie-a", []string{"ie-a"}, provision.Receipt{}),
provision.OfflineSealedBy(id, "ie-a", provision.MoveSource{}),
- provision.OfflineFlip(id),
+ provision.OfflineFlip(id, provision.Receipt{}),
provision.OfflineLandedBy(id, to),
- provision.OfflineOpenTransition(id, now),
- provision.OfflineFinishingTransition(id),
- provision.OfflineFinish(id, now),
+ provision.OfflineOpenTransition(id, now, provision.Receipt{}),
+ provision.OfflineFinishingTransition(id, provision.Receipt{}),
+ provision.OfflineFinish(id, now, provision.Receipt{}),
} {
var err error
if rec, err = srv.offlineApply(ctx, d, mut); err != nil {
@@ -239,7 +239,7 @@ func TestLateLegacyCreateIsInert(t *testing.T) {
t.Errorf("UpdateRecord = %v, want not found", err)
}
if _, err := srv.offlineApply(ctx, offlineDoor{acct: l.acct(t), app: ltApp, host: ltOne},
- provision.OfflineStart("mv_"+strings.Repeat("b", 26), "", "uk", "paul@keymail.dev", l.clock.Now())); !errors.Is(err, os.ErrNotExist) {
+ provision.OfflineStart("mv_"+strings.Repeat("b", 26), "", "uk", "paul@keymail.dev", l.clock.Now(), "")); !errors.Is(err, os.ErrNotExist) {
t.Errorf("the offline start's write = %v, want not found", err)
}
srv.cfg.LifetimesMigrate = true
diff --git a/internal/host/movesoffline_e2e_test.go b/internal/host/movesoffline_e2e_test.go
index acd1dec79..10f2327ab 100644
--- a/internal/host/movesoffline_e2e_test.go
+++ b/internal/host/movesoffline_e2e_test.go
@@ -1670,7 +1670,7 @@ func offlineReleaseSkipsAMovedRecord(t *testing.T) {
return
}
moved++
- w.wd.move(provision.OfflineFinishingTransition(w.id))
+ w.wd.move(provision.OfflineFinishingTransition(w.id, provision.Receipt{}))
}
w.tick(w.uk)
if moved != 1 {
diff --git a/internal/host/movesoffline_lifetime_test.go b/internal/host/movesoffline_lifetime_test.go
index 9aa37e7a5..25ff02a12 100644
--- a/internal/host/movesoffline_lifetime_test.go
+++ b/internal/host/movesoffline_lifetime_test.go
@@ -252,14 +252,14 @@ func TestReconstructedSourceAfterAbortWins(t *testing.T) {
// was down throughout finds afterwards. Each move advances move_seq by 3.
func (wd *offWorld) moveOnTheRecord(id, from, to string) {
wd.t.Helper()
- wd.move(provision.OfflineStart(id, from, to, e2eAddr, wd.clock()))
- wd.move(provision.OfflineSeal(id, "sealer", []string{"sealer"}))
+ wd.move(provision.OfflineStart(id, from, to, e2eAddr, wd.clock(), ""))
+ wd.move(provision.OfflineSeal(id, "sealer", []string{"sealer"}, provision.Receipt{}))
wd.move(provision.OfflineSealedBy(id, "sealer", provision.MoveSource{}))
- wd.move(provision.OfflineFlip(id))
+ wd.move(provision.OfflineFlip(id, provision.Receipt{}))
wd.move(provision.OfflineLandedBy(id, to))
- wd.move(provision.OfflineOpenTransition(id, wd.clock()))
- wd.move(provision.OfflineFinishingTransition(id))
- wd.move(provision.OfflineFinish(id, wd.clock()))
+ wd.move(provision.OfflineOpenTransition(id, wd.clock(), provision.Receipt{}))
+ wd.move(provision.OfflineFinishingTransition(id, provision.Receipt{}))
+ wd.move(provision.OfflineFinish(id, wd.clock(), provision.Receipt{}))
}
// moveID is a well-formed offline move id, one per letter.
@@ -424,7 +424,7 @@ func TestRecreatedHostWithAStartedMoveIsHeldAtOnce(t *testing.T) {
w.wd.retireThroughTombstone(offHost)
w.wd.recreateLifetime(offHost, 2, "uk")
id := moveID("e")
- w.wd.move(provision.OfflineStart(id, "uk", "", e2eAddr, w.wd.clock()))
+ w.wd.move(provision.OfflineStart(id, "uk", "", e2eAddr, w.wd.clock(), ""))
if rec := w.wd.record(); rec.Incarnation != 2 || rec.MoveSeq != 1 || rec.Move == nil || rec.Move.Seq != 1 {
t.Fatalf("fixture: the record is at (%d, %d), move %+v", rec.Incarnation, rec.MoveSeq, rec.Move)
}
diff --git a/internal/host/movesoffline_seal_test.go b/internal/host/movesoffline_seal_test.go
index 15e7d04ba..4868f8588 100644
--- a/internal/host/movesoffline_seal_test.go
+++ b/internal/host/movesoffline_seal_test.go
@@ -270,7 +270,7 @@ func (sw *sealWorld) startAndSeal(sealer string, boxes ...*offBox) {
b.tick()
names = append(names, b.name)
}
- sw.move(provision.OfflineSeal(offID, sealer, names))
+ sw.move(provision.OfflineSeal(offID, sealer, names, provision.Receipt{}))
}
func (sw *sealWorld) mirror() lease.MirrorStore {
diff --git a/internal/host/movesoffline_stage_test.go b/internal/host/movesoffline_stage_test.go
index 047efc554..e685ad4a5 100644
--- a/internal/host/movesoffline_stage_test.go
+++ b/internal/host/movesoffline_stage_test.go
@@ -54,13 +54,13 @@ func (sw *sealWorld) stage() { sw.stageBy("ie-a") }
// stageBy stages with the stager the door pinned.
func (sw *sealWorld) stageBy(stager string) {
- sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", provision.StageTarget{Stager: stager, Bucket: sealUK, Region: sealUKRegion}, sw.clock()))
+ sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", provision.StageTarget{Stager: stager, Bucket: sealUK, Region: sealUKRegion}, sw.clock(), ""))
}
// restage is a second carlos move offline stage: one more round.
func (sw *sealWorld) restage() {
sw.advance(time.Second)
- sw.move(provision.OfflineRestage(offID, sw.clock()))
+ sw.move(provision.OfflineRestage(offID, sw.clock(), provision.Receipt{}))
}
// copySpy records the source key of every server-side copy the fake S3
@@ -222,10 +222,12 @@ func TestOfflineStageThenStartThenSealSkipsThePreCopy(t *testing.T) {
// only what changed since the last stage round.
preStaged := sw.stagedRel()
preReplica := sw.replicaRel()
- sw.move(provision.OfflineStartStaged(offID, "op@example.com", sw.clock()))
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStartStaged(offID, "op@example.com", at, provision.Receipt{})
+ }))
ie.tick()
ieB.tick()
- sw.move(provision.OfflineSeal(offID, "ie-a", []string{"ie-a", "ie-b"}))
+ sw.move(provision.OfflineSeal(offID, "ie-a", []string{"ie-a", "ie-b"}, provision.Receipt{}))
spy.take()
sw.mustPass(ie)
if st := sw.record().Move.State; st != provision.OfflineSealed {
@@ -320,7 +322,7 @@ func TestOfflineAbortFromStagingRemovesTheStagedCopy(t *testing.T) {
if sw.fake.Snapshot() != snap {
t.Fatal("a second revert pass wrote to the buckets")
}
- rec := sw.move(provision.OfflineFinish(offID, sw.clock()))
+ rec := sw.move(provision.OfflineFinish(offID, sw.clock(), provision.Receipt{}))
if rec.Move != nil || rec.LastMove == nil || rec.LastMove.Outcome != provision.OfflineOutcomeAborted || rec.LastMove.OfflineS != 0 {
t.Fatalf("finish: %+v", rec.LastMove)
}
@@ -339,7 +341,7 @@ func TestOfflineStageRefusesTheSourcesOwnBucket(t *testing.T) {
ie := sw.ie
sw.instance(ie)
sw.publishTarget(sealIE, sealIERegion)
- sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", provision.StageTarget{Stager: "ie-a", Bucket: sealIE, Region: sealIERegion}, sw.clock()))
+ sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", provision.StageTarget{Stager: "ie-a", Bucket: sealIE, Region: sealIERegion}, sw.clock(), ""))
if err := sw.stagePass(ie); err == nil {
t.Fatal("staged into the source's own bucket")
}
@@ -454,7 +456,9 @@ func TestOfflineStageNoRoundAfterTheMoveLeftStaging(t *testing.T) {
sw.mustStagePass(ie)
sw.restage()
stale := sw.record() // a pass's listing, read while still staging
- sw.move(provision.OfflineStartStaged(offID, "op@example.com", sw.clock()))
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStartStaged(offID, "op@example.com", at, provision.Receipt{})
+ }))
sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/scratch/new.bin", []byte("new"))
spy := sw.spyCopies()
if err := ie.w.offlineRound(context.Background(), sw.s, stale); err != nil {
@@ -491,7 +495,7 @@ func TestOfflineStageRoundDeletesNothingOnceTheMoveLeftStaging(t *testing.T) {
once.Do(func() {
go func() { // the operator's start, while the copy is in flight
_, _ = provision.SetRecordMove(context.Background(), sw.s, sw.acct, "app", offHost,
- provision.OfflineStartStaged(offID, "op@example.com", time.Now()))
+ provision.OfflineStartStaged(offID, "op@example.com", time.Now(), provision.Receipt{}))
}()
time.Sleep(200 * time.Millisecond)
})
@@ -516,7 +520,9 @@ func TestOfflineStagerAcknowledgesItHasStopped(t *testing.T) {
sw.instance(ie)
sw.stage()
sw.mustStagePass(ie)
- sw.move(provision.OfflineStartStaged(offID, "op@example.com", sw.clock()))
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStartStaged(offID, "op@example.com", at, provision.Receipt{})
+ }))
ie.tick()
r := sw.report(moves.OfflineRoleStager, "ie-a")
if r.Phase != moves.OfflinePhaseStopped || !r.At.Equal(sw.clock()) || r.StagedObjects == 0 {
diff --git a/internal/host/movesoffline_test.go b/internal/host/movesoffline_test.go
index 45cbea708..f6c443814 100644
--- a/internal/host/movesoffline_test.go
+++ b/internal/host/movesoffline_test.go
@@ -92,6 +92,14 @@ func (wd *offWorld) create(host string) {
}
}
+// proposed is a stamped transition whose stamp is proposed from the record
+// the write reads at the world's clock (provision.ProposeStamp), as every
+// writer proposes it: the transitions store the stamp they are given.
+func (wd *offWorld) proposed(mut func(at time.Time) func(*provision.Record) error) func(*provision.Record) error {
+ now := wd.clock()
+ return func(rec *provision.Record) error { return mut(provision.ProposeStamp(rec, now))(rec) }
+}
+
// move applies one offline transition to offHost's record.
func (wd *offWorld) move(mut func(*provision.Record) error) provision.Record {
wd.t.Helper()
@@ -118,19 +126,23 @@ func (wd *offWorld) record() provision.Record {
// The legal path to a state, from the record at rest.
func (wd *offWorld) start() {
- wd.move(provision.OfflineStart(offID, "", "uk", "op@example.com", wd.clock()))
+ wd.move(provision.OfflineStart(offID, "", "uk", "op@example.com", wd.clock(), ""))
+}
+func (wd *offWorld) seal() {
+ wd.move(provision.OfflineSeal(offID, "ie-a", []string{"ie-a", "ie-b"}, provision.Receipt{}))
}
-func (wd *offWorld) seal() { wd.move(provision.OfflineSeal(offID, "ie-a", []string{"ie-a", "ie-b"})) }
func (wd *offWorld) sealed() {
wd.move(provision.OfflineSealedBy(offID, "ie-a", provision.MoveSource{}))
}
-func (wd *offWorld) flip() { wd.move(provision.OfflineFlip(offID)) }
+func (wd *offWorld) flip() { wd.move(provision.OfflineFlip(offID, provision.Receipt{})) }
func (wd *offWorld) landed() { wd.move(provision.OfflineLandedBy(offID, "uk")) }
-func (wd *offWorld) open() { wd.move(provision.OfflineOpenTransition(offID, wd.clock())) }
+func (wd *offWorld) open() {
+ wd.move(provision.OfflineOpenTransition(offID, wd.clock(), provision.Receipt{}))
+}
func (wd *offWorld) finishing() {
- wd.move(provision.OfflineFinishingTransition(offID))
+ wd.move(provision.OfflineFinishingTransition(offID, provision.Receipt{}))
}
-func (wd *offWorld) abort() { wd.move(provision.OfflineAbort(offID)) }
+func (wd *offWorld) abort() { wd.move(provision.OfflineAbort(offID, provision.Receipt{})) }
// report reads one box's report for offID, or fails.
func (wd *offWorld) report(role, box string) moves.OfflineReport {
@@ -548,7 +560,7 @@ func TestOfflineRevertReleasesEverySourceRow(t *testing.T) {
}
}
// The move finishes; nothing holds the rows again.
- wd.move(provision.OfflineFinish(offID, wd.clock()))
+ wd.move(provision.OfflineFinish(offID, wd.clock(), provision.Receipt{}))
for _, bx := range []*offBox{a, b} {
bx.tick()
if rt := bx.row(offHost); rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
diff --git a/internal/provision/offline_move.go b/internal/provision/offline_move.go
index 2ddc0dcbe..5cc11fa99 100644
--- a/internal/provision/offline_move.go
+++ b/internal/provision/offline_move.go
@@ -33,6 +33,24 @@ import (
// 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.
+//
+// The self-serve move (self-serve moves plan 2 Task 2) adds offline-queued
+// before staging: OfflineQueue creates it, OfflineQueuedToStaging begins the
+// pre-copy, OfflinePrecheck and OfflineSetEstimate move one stamp or one
+// number, and OfflineAbortAt clears a queued move. None of them advances
+// move_seq: a queued move holds nothing.
+//
+// Stamps are the caller's. Every transition that takes `at` stores it as
+// given (in UTC) and reads no clock; every caller proposes it with
+// ProposeStamp, max(now, the latest stamp on the record it observed + 1 ns),
+// so a clock that went back never writes an earlier stamp.
+//
+// Receipts. Every transition the self-serve job writes takes a Receipt as
+// its last parameter. A non-empty nonce is written to move.self_serve.
+// last_request, or to last_move.request by a write that clears the move,
+// and such a write is never a no-op, so the job can always confirm it. The
+// operator doors pass Receipt{}, which leaves the receipt fields alone, so
+// an operator transition never confirms a job request.
// MoveKindOffline is MoveState.Kind for an offline operator move. Every other
// move leaves Kind empty.
@@ -40,7 +58,7 @@ const MoveKindOffline = "offline"
// The offline move's states, in the order a move takes them:
//
-// [offline-staging ->] offline-stopping -> offline-sealing -> offline-sealed
+// [[offline-queued ->] offline-staging ->] offline-stopping -> offline-sealing -> offline-sealed
// -> offline-landing -> offline-landed -> offline-open -> offline-finishing
// -> (cleared)
//
@@ -51,7 +69,11 @@ const MoveKindOffline = "offline"
// in staging. OfflineStartStaged then takes it to stopping, and the seal's
// own round finds the pre-copied objects and skips them by their receipts.
// offline-reverting is abort's state, reached from staging through landed.
+// offline-queued is the self-serve move's first state (OfflineQueue): the
+// job has admitted it and nothing has been copied; OfflineQueuedToStaging
+// takes it to staging and OfflineAbortAt clears it.
const (
+ OfflineQueued = "offline-queued"
OfflineStaging = "offline-staging"
OfflineStopping = "offline-stopping"
OfflineSealing = "offline-sealing"
@@ -87,12 +109,19 @@ const (
// refusedOfflineSeal: a seal that names no sealer, no boxes, or a sealer
// that is not one of them (MV-O37).
refusedOfflineSeal = "a seal names the box that seals and the boxes that stopped the instance. the box that seals is one of them"
+ // refusedNotSelfServe: a self-serve step (the estimate, or a receipt)
+ // met a move the self-serve job did not queue.
+ refusedNotSelfServe = "not a self-serve move"
+ // refusedRecreated: the queue met a record of another lifetime than
+ // the one the move was requested for.
+ refusedRecreated = "the instance was recreated since the move was requested"
)
// errRecordUnchanged is what a transition returns to say "the record already
// says this": updateRecord then writes nothing and returns the record as read
-// (no record_seq advance, no updated_at stamp). Only OfflineAbort from
-// reverting uses it. Applied outside updateRecord the mut returns it as an
+// (no record_seq advance, no updated_at stamp). OfflineAbort from reverting
+// and OfflineSetEstimate with the estimate already set use it, and only
+// under an empty receipt. Applied outside updateRecord the mut returns it as an
// error, so a transition that can return it is called only through
// SetRecordMove.
var errRecordUnchanged = errors.New("provision: the record already says this")
@@ -113,6 +142,71 @@ func advanceMoveSeq(rec *Record) error {
return nil
}
+// Receipt is the request a self-serve job write carries: Nonce is persisted
+// with the request before the write, and the record carrying it is the
+// confirmation that the write landed. The zero Receipt is the operator's.
+type Receipt struct{ Nonce string }
+
+// onMove writes the receipt into the move the record still carries. An
+// empty nonce changes nothing; a non-empty one on a move the job did not
+// queue is refused. Called after the transition's own checks.
+func (r Receipt) onMove(rec *Record) error {
+ if r.Nonce == "" {
+ return nil
+ }
+ if err := r.check(rec); err != nil {
+ return err
+ }
+ ss := *rec.Move.SelfServe
+ ss.LastRequest = r.Nonce
+ rec.Move.SelfServe = &ss
+ return nil
+}
+
+// check refuses a non-empty receipt for a move without SelfServe.
+func (r Receipt) check(rec *Record) error {
+ if r.Nonce != "" && (rec.Move == nil || rec.Move.SelfServe == nil) {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, refusedNotSelfServe)
+ }
+ return nil
+}
+
+// ProposeStamp is the stamp a writer proposes for a transition decided on
+// rec: now, or one nanosecond after the latest stamp rec carries when the
+// clock is not past it. The latest stamp is the largest of the move's
+// requested_at, queued_at, staged_at, precheck_at, opened_at and
+// self_serve.offline_at, and last_move.at. A nil record proposes now. The
+// result is in UTC.
+func ProposeStamp(rec *Record, now time.Time) time.Time {
+ t := now.UTC()
+ if rec == nil {
+ return t
+ }
+ var latest time.Time
+ later := func(p *time.Time) {
+ if p != nil && p.After(latest) {
+ latest = *p
+ }
+ }
+ if m := rec.Move; m != nil {
+ later(&m.RequestedAt)
+ later(m.QueuedAt)
+ later(m.StagedAt)
+ later(m.PrecheckAt)
+ later(m.OpenedAt)
+ if m.SelfServe != nil {
+ later(m.SelfServe.OfflineAt)
+ }
+ }
+ if rec.LastMove != nil {
+ later(&rec.LastMove.At)
+ }
+ if !latest.IsZero() && !t.After(latest) {
+ t = latest.Add(time.Nanosecond).UTC()
+ }
+ return t
+}
+
// 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 }
@@ -149,9 +243,10 @@ func offlineIs(rec *Record, id string) error {
// a seeded record, an open run, a move already in flight, the same region and
// a replica. It also refuses a `from` that is not the record's region, and a
// malformed id. It stamps kind, id, from, to, state, requested_by and
-// requested_at; it does not touch the region.
-func OfflineStart(id, from, to, by string, now time.Time) func(*Record) error {
- return offlineBegin(id, from, to, by, now, OfflineStopping)
+// requested_at; it does not touch the region. returnOf is the move this one
+// returns ("" for any other), recorded as move.return_of.
+func OfflineStart(id, from, to, by string, now time.Time, returnOf string) func(*Record) error {
+ return offlineBegin(id, from, to, by, now, OfflineStopping, returnOf)
}
// StageTarget is what a stage fixes for the whole move: the one box that
@@ -162,9 +257,9 @@ type StageTarget struct{ Stager, Bucket, Region string }
// OfflineStage is OfflineStart into offline-staging: the same refusals and
// the same stamps, plus staged_at, the stager and the staging destination
// (a stage that names any of them empty is malformed). The instance keeps
-// serving.
-func OfflineStage(id, from, to, by string, st StageTarget, now time.Time) func(*Record) error {
- begin := offlineBegin(id, from, to, by, now, OfflineStaging)
+// serving. returnOf is as OfflineStart's.
+func OfflineStage(id, from, to, by string, st StageTarget, now time.Time, returnOf string) func(*Record) error {
+ begin := offlineBegin(id, from, to, by, now, OfflineStaging, returnOf)
return func(rec *Record) error {
if st.Stager == "" || st.Bucket == "" {
return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
@@ -177,7 +272,7 @@ func OfflineStage(id, from, to, by string, st StageTarget, now time.Time) func(*
}
}
-func offlineBegin(id, from, to, by string, now time.Time, state string) func(*Record) error {
+func offlineBegin(id, from, to, by string, now time.Time, state, returnOf string) func(*Record) error {
return func(rec *Record) error {
if err := seedMoveRefusal(rec); err != nil {
return err
@@ -208,22 +303,150 @@ func offlineBegin(id, from, to, by string, now time.Time, state string) func(*Re
State: state,
RequestedBy: by,
RequestedAt: now.UTC(),
+ ReturnOf: returnOf,
}
- if state == OfflineStaging {
+ switch state {
+ case OfflineStaging:
t := now.UTC()
rec.Move.StagedAt = &t
return nil // a stage holds nothing
+ case OfflineQueued:
+ t := now.UTC()
+ q, st, pc := t, t, t
+ rec.Move.QueuedAt, rec.Move.StagedAt, rec.Move.PrecheckAt = &q, &st, &pc
+ rec.Move.Seq = rec.MoveSeq
+ return nil // a queued move holds nothing
}
return advanceMoveSeq(rec) // the direct start is the hold
}
}
-// OfflineStartStaged is staging to stopping: the operator takes a staged
-// instance offline. The record's region must still be the move's source.
-// It stamps requested_by and requested_at anew, since the instance is
-// offline from now, and requested_at is always after staged_at, which is
-// how OfflineFinish tells a move that started from one that never did.
-func OfflineStartStaged(id, by string, now time.Time) func(*Record) error {
+// OfflineQueue creates a self-serve move in offline-queued. It carries
+// OfflineStage's refusals (a seeded record, an open run, a move in flight,
+// the same region, a from that is not the record's region, a replica, a
+// malformed id, stamp or stage target) and refuses a record whose
+// incarnation is not the one the move was requested for. The move is
+// offline from `from` to `to`, staged by the first source participant into
+// st's bucket and region, and carries p and ss. requested_at, queued_at,
+// staged_at and precheck_at are all `at`, the caller's proposed stamp, so
+// StartedOffline is false until the start; move.seq is the record's
+// move_seq, which the queue does not advance. A non-empty receipt is ss's
+// last_request.
+func OfflineQueue(id, from, to, by string, st StageTarget, p Participants, ss SelfServeMove, at time.Time, req Receipt) func(*Record) error {
+ begin := offlineBegin(id, from, to, by, at, OfflineQueued, "")
+ return func(rec *Record) error {
+ if st.Bucket == "" || len(p.Source) == 0 || p.Source[0].Label == "" {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
+ }
+ if rec.Incarnation != ss.Incarnation {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, refusedRecreated)
+ }
+ if err := begin(rec); err != nil {
+ return err
+ }
+ pin := Participants{Source: append([]Participant(nil), p.Source...), Target: p.Target}
+ self := ss
+ if ss.OfflineAt != nil {
+ t := *ss.OfflineAt
+ self.OfflineAt = &t
+ }
+ if req.Nonce != "" {
+ self.LastRequest = req.Nonce
+ }
+ m := rec.Move
+ m.Stager, m.StageBucket, m.StageRegion = p.Source[0].Label, st.Bucket, st.Region
+ m.Participants, m.SelfServe = &pin, &self
+ return nil
+ }
+}
+
+// OfflineQueuedToStaging is queued to staging: the job begins the pre-copy.
+// staged_at becomes `at`, the stage request the stager answers with one
+// round; requested_at stays, so the move still reads as never started. The
+// record's region must still be the move's source.
+func OfflineQueuedToStaging(id string, at time.Time, req Receipt) func(*Record) error {
+ return func(rec *Record) error {
+ if err := offlineAt(rec, id, OfflineQueued); err != nil {
+ return err
+ }
+ if rec.Region != rec.Move.From {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
+ }
+ if at.IsZero() {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
+ }
+ if err := req.onMove(rec); err != nil {
+ return err
+ }
+ t := at.UTC()
+ rec.Move.State, rec.Move.StagedAt = OfflineStaging, &t
+ return nil
+ }
+}
+
+// offlineBeforeStart checks what the steps a self-serve move takes before
+// its start check: the record carries this offline move, queued or staging.
+func offlineBeforeStart(rec *Record, id string) error {
+ if err := offlineIs(rec, id); err != nil {
+ return err
+ }
+ if s := rec.Move.State; s != OfflineQueued && s != OfflineStaging {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, offlineWrongState(s, OfflineStaging))
+ }
+ return nil
+}
+
+// OfflinePrecheck asks the participants for a precheck: precheck_at
+// becomes `at`, in queued or staging, and nothing else of the move changes
+// (the receipt aside).
+func OfflinePrecheck(id string, at time.Time, req Receipt) func(*Record) error {
+ return func(rec *Record) error {
+ if err := offlineBeforeStart(rec, id); err != nil {
+ return err
+ }
+ if at.IsZero() {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
+ }
+ if err := req.onMove(rec); err != nil {
+ return err
+ }
+ t := at.UTC()
+ rec.Move.PrecheckAt = &t
+ return nil
+ }
+}
+
+// OfflineSetEstimate records the estimate the owner is shown, in seconds,
+// on a self-serve move in queued or staging. A move without SelfServe is
+// refused. The estimate already set writes nothing under an empty receipt
+// (errRecordUnchanged); a non-empty receipt is written all the same.
+func OfflineSetEstimate(id string, seconds int, req Receipt) func(*Record) error {
+ return func(rec *Record) error {
+ if err := offlineBeforeStart(rec, id); err != nil {
+ return err
+ }
+ if rec.Move.SelfServe == nil {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, refusedNotSelfServe)
+ }
+ if rec.Move.SelfServe.EstimateSeconds == seconds && req.Nonce == "" {
+ return errRecordUnchanged
+ }
+ ss := *rec.Move.SelfServe
+ ss.EstimateSeconds = seconds
+ rec.Move.SelfServe = &ss
+ return req.onMove(rec)
+ }
+}
+
+// OfflineStartStaged is staging to stopping: the operator or the job takes
+// a staged instance offline. The record's region must still be the move's
+// source. It stamps requested_by and requested_at anew (and
+// self_serve.offline_at on a self-serve move), since the instance is
+// offline from `at`. requested_at must be after staged_at, which is how
+// OfflineFinish tells a move that started from one that never did: the
+// caller's ProposeStamp always is, and a stamp that is not is refused as
+// malformed rather than moved.
+func OfflineStartStaged(id, by string, at time.Time, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineAt(rec, id, OfflineStaging); err != nil {
return err
@@ -231,28 +454,35 @@ func OfflineStartStaged(id, by string, now time.Time) func(*Record) error {
if rec.Region != rec.Move.From {
return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
}
- if now.IsZero() {
+ if at.IsZero() || (rec.Move.StagedAt != nil && !at.After(*rec.Move.StagedAt)) {
return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
}
+ if err := req.check(rec); err != nil {
+ return err
+ }
// 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)
- }
+ t := at.UTC()
rec.Move.State, rec.Move.RequestedBy, rec.Move.RequestedAt = OfflineStopping, by, t
- return nil
+ if rec.Move.SelfServe != nil {
+ ss := *rec.Move.SelfServe
+ off := t
+ ss.OfflineAt = &off
+ rec.Move.SelfServe = &ss
+ }
+ return req.onMove(rec)
}
}
-// OfflineRestage is staging to staging: the operator asked for another
-// pre-copy round. It moves staged_at to now (always forward), which is the
-// request the stager answers with exactly one round (the stager report's
-// stage_request names the staged_at it served). requested_at stays, so the
-// move still reads as never started.
-func OfflineRestage(id string, now time.Time) func(*Record) error {
+// OfflineRestage is staging to staging: the operator or the job asked for
+// another pre-copy round. It moves staged_at to `at`, the caller's proposed
+// stamp (ProposeStamp keeps it moving forward), which is the request the
+// stager answers with exactly one round (the stager report's stage_request
+// names the staged_at it served). requested_at stays, so the move still
+// reads as never started.
+func OfflineRestage(id string, at time.Time, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineAt(rec, id, OfflineStaging); err != nil {
return err
@@ -260,13 +490,13 @@ func OfflineRestage(id string, now time.Time) func(*Record) error {
if rec.Region != rec.Move.From {
return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
}
- if now.IsZero() {
+ if at.IsZero() {
return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
}
- t := now.UTC()
- if s := rec.Move.StagedAt; s != nil && !t.After(*s) {
- t = s.Add(time.Nanosecond)
+ if err := req.onMove(rec); err != nil {
+ return err
}
+ t := at.UTC()
rec.Move.StagedAt = &t
return nil
}
@@ -290,7 +520,7 @@ func regionWord(r string) string {
// OfflineSeal is stopping to sealing: every source box has confirmed the stop
// and the operator asked for the copy. It names the sealer (the one box that
// copies) and the source boxes whose reports the later doors wait for.
-func OfflineSeal(id, sealer string, boxes []string) func(*Record) error {
+func OfflineSeal(id, sealer string, boxes []string, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineAt(rec, id, OfflineStopping); err != nil {
return err
@@ -298,6 +528,9 @@ func OfflineSeal(id, sealer string, boxes []string) func(*Record) error {
if sealer == "" || len(boxes) == 0 || !slices.Contains(boxes, sealer) {
return fmt.Errorf("%w: %s", ErrMoveConflict, refusedOfflineSeal)
}
+ if err := req.onMove(rec); err != nil {
+ return err
+ }
rec.Move.State = OfflineSealing
rec.Move.Sealer = sealer
rec.Move.Boxes = append([]string(nil), boxes...)
@@ -330,7 +563,7 @@ func OfflineSealedBy(id, sealer string, src MoveSource) func(*Record) error {
// OfflineFlip is sealed to landing, and the one record write that moves the
// region: region = to. The record's region must still be the move's source,
// so a record that was changed under the move is refused rather than flipped.
-func OfflineFlip(id string) func(*Record) error {
+func OfflineFlip(id string, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineAt(rec, id, OfflineSealed); err != nil {
return err
@@ -338,6 +571,9 @@ func OfflineFlip(id string) func(*Record) error {
if rec.Region != rec.Move.From {
return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region))
}
+ if err := req.onMove(rec); err != nil {
+ return err
+ }
rec.Region = rec.Move.To
rec.Move.State = OfflineLanding
return nil
@@ -360,15 +596,15 @@ func OfflineLandedBy(id, to string) func(*Record) error {
}
}
-// OfflineOpenTransition is landed to open: the operator opens the target. The
-// region must be the target's. It stamps opened_at from now, which is when the
-// instance came back and so what the finished move's offline time is counted
-// to.
+// OfflineOpenTransition is landed to open: the operator or the job opens the
+// target. The region must be the target's. It stamps opened_at with `at`, the
+// caller's proposed stamp, which is when the instance came back and so what
+// the finished move's offline time is counted to.
//
// The brief named this OfflineOpen, which is also the name of the state
// constant; Go has one namespace, so the two transitions that clash with a
// state (this one and OfflineFinishingTransition) carry a suffix.
-func OfflineOpenTransition(id string, now time.Time) func(*Record) error {
+func OfflineOpenTransition(id string, at time.Time, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineAt(rec, id, OfflineLanded); err != nil {
return err
@@ -376,23 +612,29 @@ 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))
}
+ if err := req.check(rec); err != nil {
+ return err
+ }
// The target serves.
if err := advanceMoveSeq(rec); err != nil {
return err
}
- t := now.UTC()
+ t := at.UTC()
rec.Move.State, rec.Move.OpenedAt = OfflineOpen, &t
- return nil
+ return req.onMove(rec)
}
}
// OfflineFinishingTransition is open to finishing: the operator asked the
// source to clean up. After open the move only goes forward.
-func OfflineFinishingTransition(id string) func(*Record) error {
+func OfflineFinishingTransition(id string, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineAt(rec, id, OfflineOpen); err != nil {
return err
}
+ if err := req.onMove(rec); err != nil {
+ return err
+ }
rec.Move.State = OfflineFinishing
return nil
}
@@ -405,15 +647,25 @@ func OfflineFinishingTransition(id string) func(*Record) error {
// the target already serves. From reverting it changes nothing and writes
// nothing (updateRecord sees errRecordUnchanged and returns the record as
// read), so a second abort is a success that leaves record_seq and updated_at
-// alone. Call it through SetRecordMove.
-func OfflineAbort(id string) func(*Record) error {
+// alone; under a non-empty receipt it writes the receipt and nothing else. A
+// queued move is refused: OfflineAbortAt clears it. Call it through
+// SetRecordMove or SetRecordMoveAt.
+func OfflineAbort(id string, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineIs(rec, id); err != nil {
return err
}
+ if err := req.check(rec); err != nil {
+ return err
+ }
switch rec.Move.State {
case OfflineReverting:
- return errRecordUnchanged
+ if req.Nonce == "" {
+ return errRecordUnchanged
+ }
+ return req.onMove(rec)
+ case OfflineQueued:
+ return fmt.Errorf("%w: %s", ErrMoveConflict, offlineWrongState(OfflineQueued, OfflineStaging))
case OfflineStaging:
// Never offline: nothing reopens, move_seq stays.
case OfflineStopping, OfflineSealing, OfflineSealed, OfflineLanding, OfflineLanded:
@@ -428,6 +680,32 @@ func OfflineAbort(id string) func(*Record) error {
rec.Region = rec.Move.From
}
rec.Move.State = OfflineReverting
+ return req.onMove(rec)
+ }
+}
+
+// OfflineAbortAt is the abort that also clears a queued move: from
+// offline-queued it clears the move and leaves last_move {id, aborted, at,
+// no offline time}, the receipt in last_move.request, move_seq unchanged
+// (nothing was held or copied). From every other state it is OfflineAbort,
+// a repeated abort on a reverting record included.
+func OfflineAbortAt(id string, at time.Time, req Receipt) func(*Record) error {
+ abort := OfflineAbort(id, req)
+ return func(rec *Record) error {
+ if err := offlineIs(rec, id); err != nil {
+ return err
+ }
+ if rec.Move.State != OfflineQueued {
+ return abort(rec)
+ }
+ if err := req.check(rec); err != nil {
+ return err
+ }
+ if at.IsZero() {
+ return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
+ }
+ rec.LastMove = &MoveOutcome{ID: id, Outcome: OfflineOutcomeAborted, At: at.UTC(), Request: req.Nonce}
+ rec.Move = nil
return nil
}
}
@@ -436,16 +714,20 @@ func OfflineAbort(id string) func(*Record) error {
// legal from finishing (outcome done, region must be the target's) and from
// reverting (outcome aborted, region must be the source's); anything else is
// refused. The outcome's offline time runs from the request to the open for a
-// done move and to now for an aborted one; a move aborted while it was still
-// staging was never offline, and records none.
-func OfflineFinish(id string, now time.Time) func(*Record) error {
+// done move and to `at` (the caller's proposed stamp) for an aborted one; a
+// move aborted while it was still staging was never offline, and records
+// none. The receipt goes to last_move.request.
+func OfflineFinish(id string, at time.Time, req Receipt) func(*Record) error {
return func(rec *Record) error {
if err := offlineIs(rec, id); err != nil {
return err
}
+ if err := req.check(rec); err != nil {
+ return err
+ }
m := rec.Move
- out := &MoveOutcome{ID: id, At: now.UTC()}
- end := now
+ out := &MoveOutcome{ID: id, At: at.UTC(), Request: req.Nonce}
+ end := at
switch m.State {
case OfflineFinishing:
if rec.Region != m.To {
diff --git a/internal/provision/offline_move_copy_test.go b/internal/provision/offline_move_copy_test.go
index a277457ee..e8473bf11 100644
--- a/internal/provision/offline_move_copy_test.go
+++ b/internal/provision/offline_move_copy_test.go
@@ -30,11 +30,11 @@ func TestOfflineHelperRefusalsMatchTheCopyRecord(t *testing.T) {
mut func(*Record) error
fills []string // in the order the placeholders appear
}{
- {"MV-O36", 1, OfflineFlip(offID), []string{OfflineStopping, OfflineSealed}},
+ {"MV-O36", 1, OfflineFlip(offID, Receipt{}), []string{OfflineStopping, OfflineSealed}},
{"MV-O38", 2, OfflineSealedBy(offID, "box-b", MoveSource{}), []string{offSealer}},
- {"MV-O40", 6, OfflineAbort(offID), []string{OfflineOpen}},
- {"MV-O41", 1, OfflineFinish(offID, now), []string{OfflineStopping}},
- {"MV-O43", 0, OfflineStart(offID, "us", "uk", "ops@example.com", now), []string{"home", "us"}},
+ {"MV-O40", 6, OfflineAbort(offID, Receipt{}), []string{OfflineOpen}},
+ {"MV-O41", 1, OfflineFinish(offID, now, Receipt{}), []string{OfflineStopping}},
+ {"MV-O43", 0, OfflineStart(offID, "us", "uk", "ops@example.com", now, ""), []string{"home", "us"}},
} {
t.Run(c.id, func(t *testing.T) {
text, ok := approved[c.id]
diff --git a/internal/provision/offline_move_test.go b/internal/provision/offline_move_test.go
index 1e52c760b..673c6c719 100644
--- a/internal/provision/offline_move_test.go
+++ b/internal/provision/offline_move_test.go
@@ -52,14 +52,14 @@ type offlineStep struct {
func offlineSteps(now time.Time) []offlineStep {
src := MoveSource{Replication: "cold", NextDue: 99, FinalTXID: 42, Release: "v0.28.4", EnvDigest: "d1", Files: &MoveFilesDigest{Plain: "p", Sealed: "s", Bytes: 7, Entries: 2}}
return []offlineStep{
- {"start", func() func(*Record) error { return OfflineStart(offID, "", "uk", "ops@example.com", now) }, ""},
- {"seal", func() func(*Record) error { return OfflineSeal(offID, offSealer, offBoxes) }, OfflineStopping},
+ {"start", func() func(*Record) error { return OfflineStart(offID, "", "uk", "ops@example.com", now, "") }, ""},
+ {"seal", func() func(*Record) error { return OfflineSeal(offID, offSealer, offBoxes, Receipt{}) }, OfflineStopping},
{"sealed", func() func(*Record) error { return OfflineSealedBy(offID, offSealer, src) }, OfflineSealing},
- {"flip", func() func(*Record) error { return OfflineFlip(offID) }, OfflineSealed},
+ {"flip", func() func(*Record) error { return OfflineFlip(offID, Receipt{}) }, OfflineSealed},
{"landed", func() func(*Record) error { return OfflineLandedBy(offID, "uk") }, OfflineLanding},
- {"open", func() func(*Record) error { return OfflineOpenTransition(offID, now.Add(30*time.Minute)) }, OfflineLanded},
- {"finishing", func() func(*Record) error { return OfflineFinishingTransition(offID) }, OfflineOpen},
- {"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour)) }, OfflineFinishing},
+ {"open", func() func(*Record) error { return OfflineOpenTransition(offID, now.Add(30*time.Minute), Receipt{}) }, OfflineLanded},
+ {"finishing", func() func(*Record) error { return OfflineFinishingTransition(offID, Receipt{}) }, OfflineOpen},
+ {"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour), Receipt{}) }, OfflineFinishing},
}
}
@@ -189,10 +189,10 @@ func TestOfflineRecordDiffersOnlyWhereTheMoveSaysSo(t *testing.T) {
seqs = append(seqs, got.RecordSeq)
}
if via != "done" {
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{})); err != nil {
t.Fatal(err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Hour))); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Hour), Receipt{})); err != nil {
t.Fatal(err)
}
}
@@ -275,7 +275,7 @@ func TestOfflineEveryIllegalPairIsRefused(t *testing.T) {
rec := offlineRecord(t, s)
offlineWalk(t, s, rec, at.walk)
if at.abort {
- if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{})); err != nil {
t.Fatal(err)
}
}
@@ -304,14 +304,14 @@ func TestOfflineWrongIDIsRefusedEverywhere(t *testing.T) {
now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
steps := offlineSteps(now)
wrong := map[string]func() func(*Record) error{
- "seal": func() func(*Record) error { return OfflineSeal(otherOff, offSealer, offBoxes) },
+ "seal": func() func(*Record) error { return OfflineSeal(otherOff, offSealer, offBoxes, Receipt{}) },
"sealed": func() func(*Record) error { return OfflineSealedBy(otherOff, offSealer, MoveSource{}) },
- "flip": func() func(*Record) error { return OfflineFlip(otherOff) },
+ "flip": func() func(*Record) error { return OfflineFlip(otherOff, Receipt{}) },
"landed": func() func(*Record) error { return OfflineLandedBy(otherOff, "uk") },
- "open": func() func(*Record) error { return OfflineOpenTransition(otherOff, now) },
- "finishing": func() func(*Record) error { return OfflineFinishingTransition(otherOff) },
- "finish": func() func(*Record) error { return OfflineFinish(otherOff, now) },
- "abort": func() func(*Record) error { return OfflineAbort(otherOff) },
+ "open": func() func(*Record) error { return OfflineOpenTransition(otherOff, now, Receipt{}) },
+ "finishing": func() func(*Record) error { return OfflineFinishingTransition(otherOff, Receipt{}) },
+ "finish": func() func(*Record) error { return OfflineFinish(otherOff, now, Receipt{}) },
+ "abort": func() func(*Record) error { return OfflineAbort(otherOff, Receipt{}) },
}
// Each wrong-id step is tried in the state where its right-id twin is legal.
for i, st := range steps[1:] {
@@ -349,7 +349,7 @@ func TestOfflineStartRefusals(t *testing.T) {
start := func(t *testing.T, s release.Store, rec Record, id, from, to string) (*Record, error) {
t.Helper()
- return SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStart(id, from, to, "ops@example.com", now))
+ return SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStart(id, from, to, "ops@example.com", now, ""))
}
t.Run("stamps the head and leaves the region", func(t *testing.T) {
s := release.NewDirStore(t.TempDir())
@@ -429,7 +429,7 @@ func TestOfflineStartRefusals(t *testing.T) {
// before any journal exists, as MoveRequest does.
unplaced := rec
unplaced.Seed = &SeedSource{ID: "sd_1", Kind: seeds.KindReplica, State: seeds.StateReady, By: "ops@example.com", At: now}
- err := OfflineStart(offID, "", "uk", "ops@example.com", now)(&unplaced)
+ err := OfflineStart(offID, "", "uk", "ops@example.com", now, "")(&unplaced)
var refusal *SeedMoveError
if !errors.Is(err, ErrMoveConflict) || !errors.As(err, &refusal) {
t.Fatalf("seed: %v", err)
@@ -443,7 +443,7 @@ func TestOfflineStartRefusals(t *testing.T) {
rec := offlineRecord(t, s)
running := rec
running.ActiveRun = &RunState{ID: "run_1", State: RunPicked, Deadline: now.Add(time.Hour)}
- err := OfflineStart(offID, "", "uk", "ops@example.com", now)(&running)
+ err := OfflineStart(offID, "", "uk", "ops@example.com", now, "")(&running)
var open *MoveRunOpenError
if !errors.As(err, &open) {
t.Fatalf("open run: %v", err)
@@ -463,20 +463,20 @@ func TestOfflineSealAndSealedByChecks(t *testing.T) {
rec := offlineRecord(t, s)
offlineWalk(t, s, rec, steps[:1])
for name, mut := range map[string]func(*Record) error{
- "no sealer": OfflineSeal(offID, "", offBoxes),
- "no boxes": OfflineSeal(offID, offSealer, nil),
+ "no sealer": OfflineSeal(offID, "", offBoxes, Receipt{}),
+ "no boxes": OfflineSeal(offID, offSealer, nil, Receipt{}),
} {
if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); !errors.Is(err, ErrMoveConflict) {
t.Errorf("%s: %v", name, err)
}
}
// The sealer is one of the boxes.
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSeal(offID, "box-z", offBoxes)); !errors.Is(err, ErrMoveConflict) || !strings.Contains(err.Error(), refusedOfflineSeal) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSeal(offID, "box-z", offBoxes, Receipt{})); !errors.Is(err, ErrMoveConflict) || !strings.Contains(err.Error(), refusedOfflineSeal) {
t.Errorf("a sealer outside the boxes: %v", err)
}
// The seal copies the caller's slice.
boxes := []string{"box-a", "box-b"}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSeal(offID, offSealer, boxes)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSeal(offID, offSealer, boxes, Receipt{})); err != nil {
t.Fatal(err)
}
boxes[0] = "changed"
@@ -513,7 +513,7 @@ func TestOfflineFlipAndLandedGuardTheRegion(t *testing.T) {
if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil {
t.Fatal(err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFlip(offID)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFlip(offID, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("flip over a changed region: %v", err)
}
@@ -527,7 +527,7 @@ func TestOfflineFlipAndLandedGuardTheRegion(t *testing.T) {
}
}
// An abort puts the region back; a landing that raced it is refused.
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{})); err != nil {
t.Fatal(err)
}
if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineLandedBy(offID, "uk")); !errors.Is(err, ErrMoveConflict) {
@@ -555,7 +555,7 @@ func TestOfflineAbort(t *testing.T) {
if got.Move.State != c.state || got.Region != c.region {
t.Fatalf("setup %s: %s %q", c.state, got.Move.State, got.Region)
}
- got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID))
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{}))
if err != nil {
t.Fatalf("abort from %s: %v", c.state, err)
}
@@ -569,7 +569,7 @@ func TestOfflineAbort(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- again, err := SetRecordMove(WithRecordSequence(ctx), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID))
+ again, err := SetRecordMove(WithRecordSequence(ctx), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{}))
rawAfter, etagAfter, gerr := s.Get(ctx, key)
if gerr != nil || !bytes.Equal(rawBefore, rawAfter) || etagBefore != etagAfter {
t.Fatalf("second abort from %s wrote the record: %v", c.state, gerr)
@@ -578,7 +578,7 @@ func TestOfflineAbort(t *testing.T) {
t.Fatalf("second abort from %s: %v %+v", c.state, err, again)
}
// The way back to rest is finish.
- done, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Minute)))
+ done, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Minute), Receipt{}))
if err != nil || done.Move != nil || done.LastMove == nil || done.LastMove.Outcome != "aborted" || done.LastMove.ID != offID || done.Region != "" {
t.Fatalf("finish after abort from %s: %v %+v", c.state, err, done)
}
@@ -590,7 +590,7 @@ func TestOfflineAbort(t *testing.T) {
rec := offlineRecord(t, s)
offlineWalk(t, s, rec, steps[:walked])
before := offlineGet(t, s, rec)
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("abort from %s: %v", state, err)
}
if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
@@ -622,19 +622,19 @@ func TestOfflineFinishFromBothAndTheRegionGuard(t *testing.T) {
if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil {
t.Fatal(err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("done finish with the wrong region: %v", err)
}
s = release.NewDirStore(t.TempDir())
rec = offlineRecord(t, s)
offlineWalk(t, s, rec, steps[:5])
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{})); err != nil {
t.Fatal(err)
}
if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "uk"; return nil }); err != nil {
t.Fatal(err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("aborted finish with the target's region: %v", err)
}
}
@@ -753,7 +753,7 @@ func TestEveryLegacyAndPhase1TransitionRefusesAnOfflineMove(t *testing.T) {
s = release.NewDirStore(t.TempDir())
rec = offlineRecord(t, s)
offlineWalk(t, s, rec, steps[:3])
- if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{})); err != nil {
t.Fatal(err)
}
check("reverting")
@@ -777,7 +777,7 @@ func TestOfflineTransitionsRefuseOtherKinds(t *testing.T) {
t.Errorf("%s move: offline %s: %v", name, st.name, err)
}
}
- if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Errorf("%s move: offline abort: %v", name, err)
}
if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
@@ -824,28 +824,28 @@ func TestOfflineRefusalsAreTheRecordedSentences(t *testing.T) {
}
s, rec := fresh(t, 0)
- refuse(t, s, rec, OfflineSeal(offID, offSealer, offBoxes), refusedNoMove)
- refuse(t, s, rec, OfflineStart(offID, "us", "uk", "ops", now), "the instance is in home, not us")
- refuse(t, s, rec, OfflineStart(offID, "", "", "ops", now), "the instance is already in home")
- refuse(t, s, rec, OfflineStart("mv_short", "", "uk", "ops", now), refusedMalformedStart)
+ refuse(t, s, rec, OfflineSeal(offID, offSealer, offBoxes, Receipt{}), refusedNoMove)
+ refuse(t, s, rec, OfflineStart(offID, "us", "uk", "ops", now, ""), "the instance is in home, not us")
+ refuse(t, s, rec, OfflineStart(offID, "", "", "ops", now, ""), "the instance is already in home")
+ refuse(t, s, rec, OfflineStart("mv_short", "", "uk", "ops", now, ""), refusedMalformedStart)
s, rec = fresh(t, 1)
- refuse(t, s, rec, OfflineSeal(otherOff, offSealer, offBoxes), refusedOtherOfflineMove)
- refuse(t, s, rec, OfflineFlip(offID), "this offline move is offline-stopping. this step follows offline-sealed")
- refuse(t, s, rec, OfflineSeal(offID, "", offBoxes), refusedOfflineSeal)
- refuse(t, s, rec, OfflineFinish(offID, now), "this offline move is offline-stopping. it finishes from finishing or reverting")
+ refuse(t, s, rec, OfflineSeal(otherOff, offSealer, offBoxes, Receipt{}), refusedOtherOfflineMove)
+ refuse(t, s, rec, OfflineFlip(offID, Receipt{}), "this offline move is offline-stopping. this step follows offline-sealed")
+ refuse(t, s, rec, OfflineSeal(offID, "", offBoxes, Receipt{}), refusedOfflineSeal)
+ refuse(t, s, rec, OfflineFinish(offID, now, Receipt{}), "this offline move is offline-stopping. it finishes from finishing or reverting")
s, rec = fresh(t, 2)
refuse(t, s, rec, OfflineSealedBy(offID, "box-b", MoveSource{}), "this offline move is sealed by box-a")
s, rec = fresh(t, 6)
- refuse(t, s, rec, OfflineAbort(offID), "this offline move is offline-open. it cannot be aborted once it is open")
+ refuse(t, s, rec, OfflineAbort(offID, Receipt{}), "this offline move is offline-open. it cannot be aborted once it is open")
s, rec = fresh(t, 3)
if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil {
t.Fatal(err)
}
- refuse(t, s, rec, OfflineFlip(offID), "the record's region is us, which this step does not expect. the instance changed under the move")
+ refuse(t, s, rec, OfflineFlip(offID, Receipt{}), "the record's region is us, which this step does not expect. the instance changed under the move")
// A legacy move meets an offline step.
s = release.NewDirStore(t.TempDir())
@@ -853,7 +853,7 @@ func TestOfflineRefusalsAreTheRecordedSentences(t *testing.T) {
if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, MoveRequest("uk", now)); err != nil {
t.Fatal(err)
}
- refuse(t, s, rec, OfflineSeal(offID, offSealer, offBoxes), refusedNotOffline)
+ refuse(t, s, rec, OfflineSeal(offID, offSealer, offBoxes, Receipt{}), refusedNotOffline)
// And the other way: the Phase 1 guard on an offline move.
s, rec = fresh(t, 1)
@@ -878,10 +878,10 @@ func moveSeqAt(t *testing.T, s release.Store, rec Record, n uint64) {
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},
+ {"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), Receipt{}) }, OfflineStaging},
{"start staged", func() func(*Record) error {
- return OfflineStartStaged(offID, "ops@example.com", now.Add(2*time.Minute))
+ return OfflineStartStaged(offID, "ops@example.com", now.Add(2*time.Minute), Receipt{})
}, OfflineStaging},
}
}
@@ -895,8 +895,8 @@ 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)) }, ""}
+ abort := offlineStep{"abort", func() func(*Record) error { return OfflineAbort(offID, Receipt{}) }, ""}
+ finish := offlineStep{"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour), Receipt{}) }, ""}
cat := func(parts ...[]offlineStep) []offlineStep {
var out []offlineStep
for _, p := range parts {
@@ -974,7 +974,7 @@ func TestOfflineTransitionsAdvanceMoveSeq(t *testing.T) {
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))
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{}))
if err != nil {
t.Fatal(err)
}
@@ -1016,7 +1016,7 @@ 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) }, ""}
+ abort := offlineStep{"abort", func() func(*Record) error { return OfflineAbort(offID, Receipt{}) }, ""}
for _, c := range []struct {
name string
walk []offlineStep
@@ -1028,7 +1028,7 @@ func TestMoveSeqExhaustionAtEveryTransition(t *testing.T) {
{"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)) }, ""}},
+ {"finish aborted", append(append([]offlineStep(nil), steps[:3]...), abort), offlineStep{"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour), Receipt{}) }, ""}},
} {
t.Run(c.name, func(t *testing.T) {
s := release.NewDirStore(t.TempDir())
diff --git a/internal/provision/offline_stage_test.go b/internal/provision/offline_stage_test.go
index 9da62fd55..577d84afc 100644
--- a/internal/provision/offline_stage_test.go
+++ b/internal/provision/offline_stage_test.go
@@ -21,7 +21,7 @@ func TestOfflineStageThenStartThenTheWholeMove(t *testing.T) {
s := release.NewDirStore(t.TempDir())
rec := offlineRecord(t, s)
- got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now))
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now, ""))
if err != nil {
t.Fatal(err)
}
@@ -31,15 +31,15 @@ func TestOfflineStageThenStartThenTheWholeMove(t *testing.T) {
t.Fatalf("staged record: %+v region %q", m, got.Region)
}
// Staging is a move in flight: no second move starts beside it.
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStart(otherOff, "", "uk", "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStart(otherOff, "", "uk", "ops@example.com", now, "")); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("a start beside a staged move: %v", err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(otherOff, "", "us", "ops@example.com", stageAt, now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(otherOff, "", "us", "ops@example.com", stageAt, now, "")); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("a stage beside a staged move: %v", err)
}
start := now.Add(10 * time.Minute)
- got, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "other@example.com", start))
+ got, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "other@example.com", start, Receipt{}))
if err != nil {
t.Fatal(err)
}
@@ -65,7 +65,7 @@ func TestOfflineStartStagedRefusals(t *testing.T) {
rec := offlineRecord(t, s)
offlineWalk(t, s, rec, steps[:walked])
before := offlineGet(t, s, rec)
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("start-staged from %s: %v", before.Move.State, err)
}
if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
@@ -74,24 +74,24 @@ func TestOfflineStartStagedRefusals(t *testing.T) {
}
s := release.NewDirStore(t.TempDir())
rec := offlineRecord(t, s)
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now, "")); err != nil {
t.Fatal(err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(otherOff, "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(otherOff, "ops@example.com", now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("start-staged with another id: %v", err)
}
// A region written under the staged move.
if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil {
t.Fatal(err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("start-staged over a moved region: %v", err)
}
// No other step leaves staging: seal through finish all refuse.
s = release.NewDirStore(t.TempDir())
rec = offlineRecord(t, s)
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now, "")); err != nil {
t.Fatal(err)
}
before := offlineGet(t, s, rec)
@@ -105,22 +105,34 @@ func TestOfflineStartStagedRefusals(t *testing.T) {
}
}
-// A start in the same instant as the stage still lands after it, so the
-// finish can tell the two apart.
+// A start proposed in the same instant as the stage, or with a clock behind
+// it, still lands after it (ProposeStamp), so the finish can tell the two
+// apart. The transition stores the caller's stamp and moves nothing: a stamp
+// that is not after staged_at is refused and nothing is written.
func TestOfflineStartStagedIsAlwaysAfterTheStage(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
s := release.NewDirStore(t.TempDir())
rec := offlineRecord(t, s)
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now, "")); err != nil {
t.Fatal(err)
}
- got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now.Add(-time.Second)))
+ before := offlineGet(t, s, rec)
+ for _, at := range []time.Time{now, now.Add(-time.Second)} {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", at, Receipt{})); !errors.Is(err, ErrMoveConflict) {
+ t.Fatalf("a start stamped %v, not after the stage: %v", at, err)
+ }
+ if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
+ t.Fatal("a refused start changed the record")
+ }
+ }
+ at := ProposeStamp(&before, now.Add(-time.Second))
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", at, Receipt{}))
if err != nil {
t.Fatal(err)
}
- if !got.Move.RequestedAt.After(*got.Move.StagedAt) || !got.Move.StartedOffline() {
- t.Fatalf("start %v is not after the stage %v", got.Move.RequestedAt, got.Move.StagedAt)
+ if !got.Move.RequestedAt.After(*got.Move.StagedAt) || !got.Move.StartedOffline() || !got.Move.RequestedAt.Equal(at) {
+ t.Fatalf("start %v is not after the stage %v, or not the proposed %v", got.Move.RequestedAt, got.Move.StagedAt, at)
}
}
@@ -132,14 +144,14 @@ func TestOfflineAbortFromStaging(t *testing.T) {
now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
s := release.NewDirStore(t.TempDir())
rec := offlineRecord(t, s)
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now, "")); err != nil {
t.Fatal(err)
}
- got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID))
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID, Receipt{}))
if err != nil || got.Move.State != OfflineReverting || got.Region != "" {
t.Fatalf("abort from staging: %v %+v", err, got)
}
- done, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Hour)))
+ done, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Hour), Receipt{}))
if err != nil || done.Move != nil || done.LastMove.Outcome != OfflineOutcomeAborted || done.LastMove.OfflineS != 0 {
t.Fatalf("finish after abort from staging: %v %+v", err, done.LastMove)
}
@@ -147,15 +159,15 @@ func TestOfflineAbortFromStaging(t *testing.T) {
s = release.NewDirStore(t.TempDir())
rec = offlineRecord(t, s)
for _, mut := range []func(*Record) error{
- OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now),
- OfflineStartStaged(offID, "ops@example.com", now.Add(20*time.Minute)),
- OfflineAbort(offID),
+ OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now, ""),
+ OfflineStartStaged(offID, "ops@example.com", now.Add(20*time.Minute), Receipt{}),
+ OfflineAbort(offID, Receipt{}),
} {
if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); err != nil {
t.Fatal(err)
}
}
- done, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(30*time.Minute)))
+ done, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(30*time.Minute), Receipt{}))
if err != nil || done.LastMove.Outcome != OfflineOutcomeAborted || done.LastMove.OfflineS != 600 {
t.Fatalf("finish after a started staged move: %v %+v", err, done.LastMove)
}
@@ -185,30 +197,34 @@ func TestOfflineRestage(t *testing.T) {
now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
s := release.NewDirStore(t.TempDir())
rec := offlineRecord(t, s)
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("restage at rest: %v", err)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now, "")); err != nil {
t.Fatal(err)
}
- got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now.Add(5*time.Minute)))
+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now.Add(5*time.Minute), Receipt{}))
if err != nil || !got.Move.StagedAt.Equal(now.Add(5*time.Minute)) || !got.Move.RequestedAt.Equal(now) || got.Move.StartedOffline() || got.Move.State != OfflineStaging || got.Move.Stager != "box-a" {
t.Fatalf("restage: %v %+v", err, got.Move)
}
- // A clock behind the last request still moves it forward.
- got, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now))
- if err != nil || !got.Move.StagedAt.After(now.Add(5*time.Minute)) {
+ // A clock behind the last request still moves it forward: the caller
+ // proposes the stamp from the record it observed, and the transition
+ // stores it as given.
+ at := ProposeStamp(got, now)
+ got, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, at, Receipt{}))
+ if err != nil || !got.Move.StagedAt.After(now.Add(5*time.Minute)) || !got.Move.StagedAt.Equal(at) {
t.Fatalf("restage with an earlier clock: %v %+v", err, got.Move)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(otherOff, now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(otherOff, now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("restage with another id: %v", err)
}
- // The start after a restage is still after the last request.
- got, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now.Add(time.Minute)))
+ // The start after a restage is still after the last request, its
+ // stamp proposed from the record with a clock behind it.
+ got, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", ProposeStamp(got, now.Add(time.Minute)), Receipt{}))
if err != nil || !got.Move.StartedOffline() || got.Move.State != OfflineStopping {
t.Fatalf("start after restage: %v %+v", err, got.Move)
}
- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now)); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now, Receipt{})); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("restage after the start: %v", err)
}
}
@@ -222,10 +238,10 @@ func TestOfflineStageNamesItsStager(t *testing.T) {
s := release.NewDirStore(t.TempDir())
rec := offlineRecord(t, s)
before := offlineGet(t, s, rec)
- if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", StageTarget{Bucket: "data-uk"}, time.Now())); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", StageTarget{Bucket: "data-uk"}, time.Now(), "")); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("a stage with no stager: %v", err)
}
- if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", StageTarget{Stager: "box-a"}, time.Now())); !errors.Is(err, ErrMoveConflict) {
+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", StageTarget{Stager: "box-a"}, time.Now(), "")); !errors.Is(err, ErrMoveConflict) {
t.Fatalf("a stage with no staging bucket: %v", err)
}
if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index 35b0faaf8..9f687a486 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -1174,6 +1174,55 @@ type MoveState struct {
// 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"`
+
+ // The self-serve move's fields (self-serve moves plan 2 Task 2,
+ // offline_move.go). All are additive and omitempty: a journal written
+ // before them parses and re-encodes byte for byte. SelfServe is set
+ // only on a move the self-serve job queued (OfflineQueue); its
+ // LastRequest is the receipt of the job's latest write.
+ SelfServe *SelfServeMove `json:"self_serve,omitempty"`
+ // Participants are the boxes the move admitted, pinned at the queue.
+ Participants *Participants `json:"participants,omitempty"`
+ // QueuedAt is when the move was queued; PrecheckAt is the latest
+ // precheck request (OfflinePrecheck), the stamp the boxes' precheck
+ // reports answer.
+ QueuedAt *time.Time `json:"queued_at,omitempty"`
+ PrecheckAt *time.Time `json:"precheck_at,omitempty"`
+ // ReturnOf names the move this one returns, "" for any other move
+ // (OfflineStage, OfflineStart).
+ ReturnOf string `json:"return_of,omitempty"`
+}
+
+// SelfServeMove is the self-serve part of an offline move: the row the job
+// drives (ID), the lifetime it was requested for, who asked, the estimate
+// the owner was shown, when the instance went offline, and LastRequest, the
+// receipt: every self-serve mutation sets it to the request's nonce, and a
+// request is confirmed by the record carrying it.
+type SelfServeMove struct {
+ ID string `json:"id"`
+ Incarnation uint64 `json:"incarnation"`
+ DisplayName string `json:"display_name,omitempty"`
+ RequestedBy string `json:"requested_by,omitempty"`
+ EstimateSeconds int `json:"estimate_seconds,omitempty"`
+ OfflineAt *time.Time `json:"offline_at,omitempty"`
+ LastRequest string `json:"last_request,omitempty"`
+}
+
+// Participant is one box a move admitted: its box id and label from the
+// edges file, the instance id its status report named (empty when no fresh
+// report did), and whether that id was verified.
+type Participant struct {
+ BoxID string `json:"box_id"`
+ Label string `json:"label"`
+ InstanceID string `json:"instance_id,omitempty"`
+ Verified bool `json:"verified,omitempty"`
+}
+
+// Participants are a move's pinned boxes: the source boxes (the first is
+// the stager) and the one target.
+type Participants struct {
+ Source []Participant `json:"source"`
+ Target Participant `json:"target"`
}
// MoveInputs (what the target was checked against, §7.3 T3) is NOT
@@ -1219,6 +1268,10 @@ type MoveOutcome struct {
Outcome string `json:"outcome"` // done | rolled_back | aborted
At time.Time `json:"at"`
OfflineS float64 `json:"offline_s,omitempty"`
+ // Request is the receipt of the write that cleared the move: the
+ // self-serve request's nonce (OfflineFinish, OfflineAbortAt), "" for
+ // any other write. Additive and omitempty.
+ Request string `json:"request,omitempty"`
}
// MoveRollback is §7.8's journal.
diff --git a/internal/provision/reconcile_lifetime_test.go b/internal/provision/reconcile_lifetime_test.go
index 2dce21778..e33611d21 100644
--- a/internal/provision/reconcile_lifetime_test.go
+++ b/internal/provision/reconcile_lifetime_test.go
@@ -759,11 +759,11 @@ func (x *lifeScene) move(t *testing.T, muts ...func(*Record) error) Record {
var pairNow = time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
func startAs(id string) func(*Record) error {
- return OfflineStart(id, "ie", "uk", "ops@example.com", pairNow)
+ return OfflineStart(id, "ie", "uk", "ops@example.com", pairNow, "")
}
func throughLanded(id string) []func(*Record) error {
- return []func(*Record) error{OfflineSeal(id, "box-a", []string{"box-a"}),
- OfflineSealedBy(id, "box-a", MoveSource{Release: "v1"}), OfflineFlip(id), OfflineLandedBy(id, "uk")}
+ return []func(*Record) error{OfflineSeal(id, "box-a", []string{"box-a"}, Receipt{}),
+ OfflineSealedBy(id, "box-a", MoveSource{Release: "v1"}), OfflineFlip(id, Receipt{}), OfflineLandedBy(id, "uk")}
}
func (x *lifeScene) pairIs(t *testing.T, want registry.Pair, status, op string) {
@@ -857,8 +857,8 @@ func TestReconcileInstallsTheRecordsPairOnAMintedRow(t *testing.T) {
muts []func(*Record) error
}{
{"after finish", "uk", append(append([]func(*Record) error{startAs(pairMove)}, throughLanded(pairMove)...),
- OfflineOpenTransition(pairMove, pairNow.Add(time.Hour)), OfflineFinishingTransition(pairMove), OfflineFinish(pairMove, pairNow.Add(2*time.Hour)))},
- {"after abort then finish", "ie", []func(*Record) error{startAs(pairMove), OfflineAbort(pairMove), OfflineFinish(pairMove, pairNow.Add(time.Hour))}},
+ OfflineOpenTransition(pairMove, pairNow.Add(time.Hour), Receipt{}), OfflineFinishingTransition(pairMove, Receipt{}), OfflineFinish(pairMove, pairNow.Add(2*time.Hour), Receipt{}))},
+ {"after abort then finish", "ie", []func(*Record) error{startAs(pairMove), OfflineAbort(pairMove, Receipt{}), OfflineFinish(pairMove, pairNow.Add(time.Hour), Receipt{})}},
} {
t.Run(c.name, func(t *testing.T) {
x := newLifeScene(t)
@@ -887,13 +887,13 @@ func TestAnExistingRowKeepsOpenSequenceThroughFinish(t *testing.T) {
x.pass(t)
x.pairIs(t, registry.Pair{Incarnation: 1, ServingSeq: 1}, "landing", pairMove)
x.move(t, OfflineLandedBy(pairMove, "uk"))
- rec := x.move(t, OfflineOpenTransition(pairMove, pairNow.Add(time.Hour)))
+ rec := x.move(t, OfflineOpenTransition(pairMove, pairNow.Add(time.Hour), Receipt{}))
// The target's open releases with the pair of the record it observed.
if ok, err := x.reg.ReleaseHoldSeq(lifeHost, pairMove, movePair(rec)); err != nil || !ok {
t.Fatalf("open's release: %v %v", ok, err)
}
x.pairIs(t, registry.Pair{Incarnation: 1, ServingSeq: 2}, "ready", "")
- for _, mut := range []func(*Record) error{OfflineFinishingTransition(pairMove), OfflineFinish(pairMove, pairNow.Add(2*time.Hour))} {
+ for _, mut := range []func(*Record) error{OfflineFinishingTransition(pairMove, Receipt{}), OfflineFinish(pairMove, pairNow.Add(2*time.Hour), Receipt{})} {
x.pass(t)
x.move(t, mut)
x.pass(t)
@@ -990,9 +990,9 @@ func TestExistingRowsAcrossLifetimes(t *testing.T) {
var rec Record
if end == "open" {
x.move(t, throughLanded(pairMove)...)
- rec = x.move(t, OfflineOpenTransition(pairMove, pairNow.Add(time.Hour)))
+ rec = x.move(t, OfflineOpenTransition(pairMove, pairNow.Add(time.Hour), Receipt{}))
} else {
- rec = x.move(t, OfflineAbort(pairMove))
+ rec = x.move(t, OfflineAbort(pairMove, Receipt{}))
}
if ok, err := x.reg.ReleaseHoldSeq(lifeHost, pairMove, movePair(rec)); err != nil || !ok {
t.Fatalf("%s's release: %v %v", end, ok, err)