amadan

carlos / platform Public

console: tests for the self-serve doors, pinning and the view

TestRecordEligibilityIsMoveGatesChecks, TestStartValidationSplitPreservesOperatorOrdering,
TestParticipantPinning, TestPrepareRefusesBeforeWriting,
TestPrepareMigratesALegacyRecordBoundThenAdmits, TestPrepareWritesTheRowFirst,
TestMutationsNameTheMoveAndRace, TestViewPrecedence, TestEstimateDoorWritesNothing,
TestStatusStaysOpenToPlainInstanceTokens, TestSwitchOffIsByteIdentical.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 2a9530307bbb4d0ce64a51921f04de3d6f3cc68d parent 24a467d
1 file changed, +1150 −0
  • internal/console/selfserve_door_test.go +1150 −0
diff --git a/internal/console/selfserve_door_test.go b/internal/console/selfserve_door_test.go
new file mode 100644
index 000000000..ba6031c40
--- /dev/null
+++ b/internal/console/selfserve_door_test.go
@@ -0,0 +1,1150 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/boxid"
+ "github.com/carlosframework/platform/internal/boxid/boxidtest"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+// The self-serve door bodies, the split start validation, participant
+// pinning and the view (self-serve moves plan 2 Task 5), driven through the
+// routes on the offline doors' world: the instance acme.shop.example.com of
+// the app shop (account bab) in ie, one box in ie and one in uk, the
+// owner's move-capable token, lifetimes on.
+
+// ssWorld is offlineWorld with the self-serve switch on.
+type ssWorld struct {
+ *offlineWorld
+ clock *lifetimesClock
+ owner string // a move-capable instance token of bab/shop
+ plain string // a plain instance token of bab/shop
+ signer *boxidtest.Signer
+ move string // .../instances/<host>/move
+ boxes []offlineBox
+}
+
+// newSelfServeWorld cuts lifetimes over, then creates the instance (so it
+// has a lifetime; legacy writes a fieldless record instead), publishes the
+// fleet and the boxes' healthy status reports.
+func newSelfServeWorld(t *testing.T, legacy bool, boxes ...offlineBox) *ssWorld {
+ t.Helper()
+ clock := useLifetimesClock(t)
+ fx, token := instanceFixture(t, "code-ss-"+strings.NewReplacer("/", "-", " ", "-").Replace(t.Name()))
+ if err := fx.st.SetOperator("alice@keymail.dev", true); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ m := offlineManifest()
+ m.Regions, m.Pool = nil, ""
+ enableInstances(t, fx, "shop", m)
+ if code, _, body := rawReq(t, http.MethodPost, fx.ts.URL+lifetimesCutoverPath, token, ""); code != http.StatusOK {
+ t.Fatalf("cutover = %d %s", code, body)
+ }
+ clock.Advance(lifetimeSettle)
+ fx.srv.lifetimesWatchOnce(context.Background())
+ if !fx.srv.lifetimesOn() {
+ t.Fatal("lifetimes are not on")
+ }
+ u := instancesURL(fx, "bab", "shop")
+ if legacy {
+ rec := provision.Record{Schema: 1, Host: offlineHost, Slug: "acme", App: "shop", Account: release.OpsAccount,
+ Channel: "stable", Region: "ie", CreatedAt: time.Now().UTC()}
+ if err := provision.CreateRecord(context.Background(), fx.rel, rec); err != nil {
+ t.Fatal(err)
+ }
+ } else if c, _, out := rawReq(t, http.MethodPost, u, token, `{"host":"`+offlineHost+`","region":"ie"}`); c != http.StatusCreated {
+ t.Fatalf("create = %d %s", c, out)
+ }
+ w := &ssWorld{offlineWorld: &offlineWorld{t: t, fx: fx, token: token, base: u + "/" + offlineHost + "/move/offline"},
+ clock: clock, move: u + "/" + offlineHost + "/move"}
+ if len(boxes) == 0 {
+ boxes = oneEach
+ }
+ w.fleet(boxes...)
+ w.boxes = boxes
+ fx.srv.cfg.SelfServeMoves = "offline"
+ fx.srv.cfg.AWSAccount = moveTestAccount
+ w.signer = boxidtest.NewSigner(t)
+ fx.srv.boxCerts = w.signer.Certs()
+ var err error
+ if w.owner, err = fx.st.NewAppToken(OpsAccountID, "shop", "paul@keymail.dev", "shop-mover", true, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if w.plain, err = fx.st.NewAppToken(OpsAccountID, "shop", "paul@keymail.dev", "shop-router", false, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ w.healthy()
+ return w
+}
+
+// report writes box's status report: the box's id and instance, then edit.
+func (w *ssWorld) report(box string, edit func(b *status.Box)) {
+ w.t.Helper()
+ sb := status.Box{Label: box, GeneratedAt: time.Now().UTC(), Metrics: &status.BoxMetrics{BoxID: box, InstanceID: "i-" + box}}
+ if edit != nil {
+ edit(&sb)
+ }
+ if err := status.WriteBox(context.Background(), w.fx.rel, sb); err != nil {
+ w.t.Fatal(err)
+ }
+}
+
+// facts edits the source's facts for the host and rewrites its report.
+func (w *ssWorld) facts(edit func(f *status.HostFacts)) {
+ w.t.Helper()
+ rec := w.record()
+ now := time.Now().Unix()
+ w.report("ie-box", func(b *status.Box) {
+ b.Metrics.Lanes = []string{"data-uk"}
+ f := status.HostFacts{CollectedAt: now, Known: true, Incarnation: rec.Incarnation, ReplicaBytes: 3 << 20,
+ ReplicaObjects: 3, ReplicaCollectedAt: now, FilesBytes: 1 << 20, FilesEntries: 2}
+ if edit != nil {
+ edit(&f)
+ }
+ b.Metrics.Hosts = map[string]status.HostFacts{offlineHost: f}
+ })
+}
+
+// healthy is every box reporting its id and instance, and ie-box, the
+// source, its facts and its lane to data-uk.
+func (w *ssWorld) healthy() {
+ w.t.Helper()
+ for _, b := range w.boxes {
+ if b.Box == "ie-box" {
+ w.facts(nil)
+ } else {
+ w.report(b.Box, nil)
+ }
+ }
+}
+
+// queued writes the record's move as the job's queue write would
+// (provision.OfflineQueue), naming row.
+func (w *ssWorld) queued(row *SelfServeMove) {
+ w.t.Helper()
+ w.edit(provision.OfflineQueue(row.ID, row.From, row.To, row.RequestedBy, provision.StageTarget{Bucket: "data-uk", Region: "eu-west-2"},
+ row.Participants, provision.SelfServeMove{ID: row.ID, Incarnation: row.Incarnation}, time.Now(), provision.Receipt{}))
+}
+
+func (w *ssWorld) do(method, suffix, token, body string) (int, string) {
+ w.t.Helper()
+ code, _, out := rawReq(w.t, method, w.move+suffix, token, body)
+ return code, out
+}
+
+func (w *ssWorld) prepare(body string) (int, string) {
+ w.t.Helper()
+ return w.do(http.MethodPost, "", w.owner, body)
+}
+
+func ssView(t *testing.T, out string) selfServeMoveView {
+ t.Helper()
+ var v selfServeMoveView
+ if err := json.Unmarshal([]byte(out), &v); err != nil {
+ t.Fatalf("%v in %s", err, out)
+ }
+ return v
+}
+
+// prepared runs a prepare that must be admitted and returns the row.
+func (w *ssWorld) prepared() *SelfServeMove {
+ w.t.Helper()
+ code, out := w.prepare(`{"region":"uk","display_name":"Shop's account","requested_by":"shop-admin"}`)
+ if code != http.StatusAccepted {
+ w.t.Fatalf("prepare = %d %s", code, out)
+ }
+ v := ssView(w.t, out)
+ row, err := w.fx.st.SelfServeMove(v.ID)
+ if err != nil || row == nil {
+ w.t.Fatalf("the prepared row %s: %v %v", v.ID, row, err)
+ }
+ return row
+}
+
+// rows is the instance's open and newest terminal row.
+func (w *ssWorld) rows() (open, terminal *SelfServeMove) {
+ w.t.Helper()
+ open, terminal, err := w.fx.st.SelfServeMoveForHost("bab", "shop", offlineHost)
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ return open, terminal
+}
+
+// setPhase moves a row as the job would, for the door tests.
+func (w *ssWorld) setPhase(id, phase string) {
+ w.t.Helper()
+ if _, err := w.fx.st.db.Exec(`UPDATE self_serve_moves SET phase = ? WHERE id = ?`, phase, id); err != nil {
+ w.t.Fatal(err)
+ }
+}
+
+// refusedSS posts a prepare and wants exactly this refusal, with nothing
+// written: the bucket as it was and no row.
+func (w *ssWorld) refusedSS(body string, wantCode int, wantMsg string) {
+ w.t.Helper()
+ before := bucketState(w.t, w.fx.rel)
+ code, out := w.prepare(body)
+ if code != wantCode || errorText(out) != wantMsg {
+ w.t.Fatalf("prepare %s = %d %s\nwant %d %q", body, code, out, wantCode, wantMsg)
+ }
+ after := bucketState(w.t, w.fx.rel)
+ if len(before) != len(after) {
+ w.t.Fatalf("a refusal changed the bucket: %d objects before, %d after", len(before), len(after))
+ }
+ for k, v := range before {
+ if after[k] != v {
+ w.t.Fatalf("a refusal changed %s", k)
+ }
+ }
+ if open, _ := w.rows(); open != nil && wantMsg != "a move to uk is already preparing" {
+ w.t.Fatalf("a refusal admitted a row: %+v", open)
+ }
+}
+
+const ssBody = `{"region":"uk"}`
+
+// ---- record eligibility and the split ----------------------------------------
+
+func TestRecordEligibilityIsMoveGatesChecks(t *testing.T) {
+ c := mainMoveCopy()
+ no := false
+ cases := []struct {
+ name string
+ edit func(r *provision.Record)
+ want string
+ }{
+ {"an eligible record", func(r *provision.Record) {}, ""},
+ {"a service", func(r *provision.Record) { r.Kind = "service" }, c.kind("service")},
+ {"an addr", func(r *provision.Record) { r.Addr = "127.0.0.1:9000" }, c.addr},
+ {"addrs", func(r *provision.Record) { r.Addrs = []string{"127.0.0.1:9000"} }, c.addr},
+ {"a fleet placement", func(r *provision.Record) { r.Region = "fleet/shop" }, c.fleet},
+ {"a pool", func(r *provision.Record) { r.Pool = "shop" }, c.pool},
+ {"not hibernating", func(r *provision.Record) { r.Hibernate = &no }, c.hibernate},
+ // moveGate's order: the first rail a record fails is the answer.
+ {"a service in a pool", func(r *provision.Record) { r.Kind = "service"; r.Pool = "shop" }, c.kind("service")},
+ {"an addr in a pool", func(r *provision.Record) { r.Addr = "127.0.0.1:9000"; r.Pool = "shop" }, c.addr},
+ {"a fleet placement in a pool", func(r *provision.Record) { r.Region = "fleet/shop"; r.Pool = "shop" }, c.fleet},
+ {"a pool, not hibernating", func(r *provision.Record) { r.Pool = "shop"; r.Hibernate = &no }, c.pool},
+ }
+ w := newSelfServeWorld(t, false)
+ base := *w.record()
+ for _, k := range cases {
+ t.Run(k.name, func(t *testing.T) {
+ rec := base
+ k.edit(&rec)
+ got := recordEligibility(&rec, c)
+ if (got == nil) != (k.want == "") || (got != nil && (got.status != http.StatusBadRequest || got.msg != k.want)) {
+ t.Fatalf("recordEligibility = %+v, want 400 %q", got, k.want)
+ }
+ if k.want == "" {
+ return
+ }
+ // The same record at the routes: moveGate (the status door, a
+ // plain instance token) and the self-serve prepare answer it so.
+ key := provision.RecordKey("bab", "shop", offlineHost)
+ raw, etag, err := w.fx.rel.Get(context.Background(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, _ := json.Marshal(rec)
+ if _, err := w.fx.rel.Put(context.Background(), key, body, etag); err != nil {
+ t.Fatal(err)
+ }
+ defer func() {
+ _, etag, _ := w.fx.rel.Get(context.Background(), key)
+ if _, err := w.fx.rel.Put(context.Background(), key, raw, etag); err != nil {
+ t.Fatal(err)
+ }
+ }()
+ for _, door := range []struct {
+ name, method, token, body string
+ }{{"status", http.MethodGet, w.plain, ""}, {"prepare", http.MethodPost, w.owner, ssBody}} {
+ code, out := w.do(door.method, "", door.token, door.body)
+ if code != http.StatusBadRequest || errorText(out) != k.want {
+ t.Errorf("%s = %d %s, want 400 %q", door.name, code, out, k.want)
+ }
+ }
+ })
+ }
+}
+
+// TestStartValidationSplitPreservesOperatorOrdering: offlineRecordChecks,
+// the body, then offlineRegionChecks is the operator start and stage doors'
+// answer, case by case (the golden replay, TestOperatorDoorAnswersAreUnchanged,
+// holds the doors to main).
+func TestStartValidationSplitPreservesOperatorOrdering(t *testing.T) {
+ cases := []struct {
+ name string
+ boxes []offlineBox
+ setup func(w *offlineWorld)
+ body string
+ }{
+ {"a replica with a malformed body", oneEach, func(w *offlineWorld) {
+ w.edit(func(r *provision.Record) error { r.Role = "replica"; return nil })
+ }, `{"region":`},
+ {"a claim with an invalid region", oneEach, func(w *offlineWorld) { offlineClaimFor(t, w.fx, offlineHost, "ie-box", "ie") }, `{"region":"UK!"}`},
+ {"a malformed body", oneEach, nil, `{"region":`},
+ {"the current region", oneEach, nil, `{"region":"ie"}`},
+ {"two target boxes", []offlineBox{ieOff("ie-box"), ukOff("uk-a"), ukOff("uk-b")}, nil, ssBody},
+ {"a target that cannot stage", []offlineBox{ieOff("ie-box"), func() offlineBox { b := ukOff("uk-box"); b.NoStageCap = true; return b }()}, nil, ssBody},
+ }
+ for _, c := range cases {
+ for _, door := range []struct {
+ path string
+ stage bool
+ }{{"", false}, {"/stage", true}} {
+ t.Run(c.name+door.path, func(t *testing.T) {
+ w := newOfflineWorld(t, c.boxes...)
+ if c.setup != nil {
+ c.setup(w)
+ }
+ rec := w.record()
+ ctx := context.Background()
+ want := func() (int, string) {
+ if ref := w.fx.srv.offlineRecordChecks(ctx, "bab", "shop", offlineHost, rec); ref != nil {
+ return ref.status, ref.msg
+ }
+ var req struct {
+ Region string `json:"region"`
+ }
+ if err := json.Unmarshal([]byte(c.body), &req); err != nil {
+ return http.StatusBadRequest, "malformed JSON body"
+ }
+ if _, ref := w.fx.srv.offlineRegionChecks(ctx, "bab", "shop", offlineHost, rec, req.Region, door.stage, false, ""); ref != nil {
+ return ref.status, ref.msg
+ }
+ return http.StatusAccepted, ""
+ }
+ wantCode, wantMsg := want()
+ code, out := w.req(http.MethodPost, door.path, c.body)
+ if code != wantCode || (wantCode != http.StatusAccepted && errorText(out) != wantMsg) {
+ t.Fatalf("door = %d %s, the two halves = %d %q", code, out, wantCode, wantMsg)
+ }
+ })
+ }
+ }
+}
+
+// ---- participant pinning -------------------------------------------------------
+
+// keyReads records every key read through Get, and fails the reads of one.
+type keyReads struct {
+ release.Store
+ mu sync.Mutex
+ keys []string
+ fail string
+ fault error
+}
+
+func (k *keyReads) Get(ctx context.Context, key string) ([]byte, string, error) {
+ k.mu.Lock()
+ k.keys = append(k.keys, key)
+ fail := k.fail == key
+ k.mu.Unlock()
+ if fail {
+ return nil, "", k.fault
+ }
+ return k.Store.Get(ctx, key)
+}
+
+func (k *keyReads) read(prefix string) []string {
+ k.mu.Lock()
+ defer k.mu.Unlock()
+ var out []string
+ for _, key := range k.keys {
+ if strings.HasPrefix(key, prefix) {
+ out = append(out, key)
+ }
+ }
+ return out
+}
+
+func TestParticipantPinning(t *testing.T) {
+ ctx := context.Background()
+ pin := func(t *testing.T, w *ssWorld, requireIDs bool) (offlineSelection, *offlineRefusal, *keyReads) {
+ t.Helper()
+ kr := &keyReads{Store: w.fx.rel}
+ w.fx.srv.cfg.Releases = kr
+ defer func() { w.fx.srv.cfg.Releases = w.fx.rel }()
+ sel, ref := w.fx.srv.offlineRegionChecks(ctx, "bab", "shop", offlineHost, w.record(), "uk", true, requireIDs, "")
+ return sel, ref, kr
+ }
+ noMetrics := func(w *ssWorld) {
+ for _, b := range []string{"ie-box", "uk-box"} {
+ w.report(b, func(sb *status.Box) { sb.Metrics = nil })
+ }
+ }
+
+ t.Run("operator: no metrics pins id-less and reserves the sentinel", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ noMetrics(w)
+ sel, ref, kr := pin(t, w, false)
+ if ref != nil {
+ t.Fatal(ref.msg)
+ }
+ want := provision.Participant{BoxID: "ie-box", Label: "ie-box"}
+ if len(sel.Source) != 1 || sel.Source[0] != want || sel.Target != (provision.Participant{BoxID: "uk-box", Label: "uk-box"}) || !sel.Sentinel {
+ t.Fatalf("selection = %+v / %+v sentinel %v", sel.Source, sel.Target, sel.Sentinel)
+ }
+ if got := sel.reservationKeys(); len(got) != 1 || got[0] != moves.OfflineSentinel {
+ t.Fatalf("an id-less selection reserves %v", got)
+ }
+ if r := kr.read(boxid.Prefix); len(r) != 0 {
+ t.Fatalf("an operator door read identity records: %v", r)
+ }
+ // And the operator start succeeds as today.
+ w.start()
+ })
+ t.Run("operator: a current report pins the instance, unverified", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ sel, ref, kr := pin(t, w, false)
+ if ref != nil {
+ t.Fatal(ref.msg)
+ }
+ if sel.Source[0] != (provision.Participant{BoxID: "ie-box", Label: "ie-box", InstanceID: "i-ie-box"}) ||
+ sel.Target != (provision.Participant{BoxID: "uk-box", Label: "uk-box", InstanceID: "i-uk-box"}) || sel.Sentinel {
+ t.Fatalf("selection = %+v / %+v sentinel %v", sel.Source, sel.Target, sel.Sentinel)
+ }
+ if r := kr.read(boxid.Prefix); len(r) != 0 {
+ t.Fatalf("an operator door read identity records: %v", r)
+ }
+ })
+ for _, c := range []struct {
+ name string
+ edit func(b *status.Box)
+ }{
+ {"older than ten minutes", func(b *status.Box) { b.GeneratedAt = time.Now().Add(-11 * time.Minute) }},
+ {"from the future", func(b *status.Box) { b.GeneratedAt = time.Now().Add(5 * time.Minute) }},
+ {"naming another box id", func(b *status.Box) { b.Metrics.BoxID = "ie-other" }},
+ {"naming no instance", func(b *status.Box) { b.Metrics.InstanceID = "" }},
+ } {
+ t.Run("operator: a report "+c.name+" pins id-less", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ w.report("ie-box", c.edit)
+ sel, ref, _ := pin(t, w, false)
+ if ref != nil {
+ t.Fatal(ref.msg)
+ }
+ if sel.Source[0].InstanceID != "" || !sel.Sentinel {
+ t.Fatalf("source %+v sentinel %v", sel.Source[0], sel.Sentinel)
+ }
+ })
+ t.Run("self-serve: a report "+c.name+" refuses", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ w.report("ie-box", c.edit)
+ if _, ref, _ := pin(t, w, true); ref == nil || ref.status != http.StatusConflict || ref.msg != "the box ie-box has not reported recently" {
+ t.Fatalf("refusal = %+v", ref)
+ }
+ })
+ }
+ t.Run("self-serve: no metrics refuses", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ noMetrics(w)
+ if _, ref, _ := pin(t, w, true); ref == nil || ref.msg != "the box ie-box has not reported recently" {
+ t.Fatalf("refusal = %+v", ref)
+ }
+ })
+ t.Run("self-serve: no identity record is pinned unverified", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ sel, ref, kr := pin(t, w, true)
+ if ref != nil {
+ t.Fatal(ref.msg)
+ }
+ if sel.Source[0] != (provision.Participant{BoxID: "ie-box", Label: "ie-box", InstanceID: "i-ie-box"}) || sel.Sentinel {
+ t.Fatalf("source %+v sentinel %v", sel.Source[0], sel.Sentinel)
+ }
+ if r := kr.read(boxid.Prefix); len(r) != 2 || r[0] != boxid.Key("ie-box") || r[1] != boxid.Key("uk-box") {
+ t.Fatalf("identity records read: %v", r)
+ }
+ })
+ t.Run("self-serve: verified by the record under the box id, not the label", func(t *testing.T) {
+ ie, uk := ieOff("ie-box"), ukOff("uk-box")
+ ie.ID, uk.ID = "ie-0042", "uk-0043"
+ w := newSelfServeWorld(t, false, ie, uk)
+ w.report("ie-box", func(b *status.Box) {
+ b.Metrics.BoxID, b.Metrics.Lanes = "ie-0042", []string{"data-uk"}
+ })
+ w.report("uk-box", func(b *status.Box) { b.Metrics.BoxID = "uk-0043" })
+ _, raw := w.signer.Record(t, "ie-0042", moveTestAccount, "eu-west-1", "i-ie-box")
+ offlinePut(t, w.fx.rel, boxid.Key("ie-0042"), raw)
+ // A record under the label that would not verify is never read.
+ offlinePut(t, w.fx.rel, boxid.Key("ie-box"), []byte(`{"v":1,"box":"ie-box"}`))
+ sel, ref, kr := pin(t, w, true)
+ if ref != nil {
+ t.Fatal(ref.msg)
+ }
+ if sel.Source[0] != (provision.Participant{BoxID: "ie-0042", Label: "ie-box", InstanceID: "i-ie-box", Verified: true}) {
+ t.Fatalf("source %+v", sel.Source[0])
+ }
+ if sel.Target != (provision.Participant{BoxID: "uk-0043", Label: "uk-box", InstanceID: "i-uk-box"}) {
+ t.Fatalf("target %+v", sel.Target)
+ }
+ for _, k := range kr.read(boxid.Prefix) {
+ if k == boxid.Key("ie-box") {
+ t.Fatalf("the label's key was read: %v", kr.read(boxid.Prefix))
+ }
+ }
+ })
+ t.Run("self-serve: a record for another instance does not verify", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ _, raw := w.signer.Record(t, "ie-box", moveTestAccount, "eu-west-1", "i-replaced")
+ offlinePut(t, w.fx.rel, boxid.Key("ie-box"), raw)
+ if _, ref, _ := pin(t, w, true); ref == nil || ref.status != http.StatusConflict || ref.msg != "box ie-box has an identity record that does not verify" {
+ t.Fatalf("refusal = %+v", ref)
+ }
+ })
+ t.Run("self-serve: a record of another account does not verify", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ _, raw := w.signer.Record(t, "ie-box", "999988887777", "eu-west-1", "i-ie-box")
+ offlinePut(t, w.fx.rel, boxid.Key("ie-box"), raw)
+ if _, ref, _ := pin(t, w, true); ref == nil || ref.msg != "box ie-box has an identity record that does not verify" {
+ t.Fatalf("refusal = %+v", ref)
+ }
+ })
+ t.Run("self-serve: a record that cannot be read refuses", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ kr := &keyReads{Store: w.fx.rel, fail: boxid.Key("uk-box"), fault: errors.New("the bucket is down")}
+ w.fx.srv.cfg.Releases = kr
+ defer func() { w.fx.srv.cfg.Releases = w.fx.rel }()
+ _, ref := w.fx.srv.offlineRegionChecks(ctx, "bab", "shop", offlineHost, w.record(), "uk", true, true, "")
+ if ref == nil || ref.status != http.StatusBadGateway || ref.msg != "could not read the identity record for box uk-box" {
+ t.Fatalf("refusal = %+v", ref)
+ }
+ })
+}
+
+// ---- the prepare --------------------------------------------------------------
+
+func TestPrepareRefusesBeforeWriting(t *testing.T) {
+ type tc struct {
+ name string
+ boxes []offlineBox
+ setup func(w *ssWorld)
+ body string
+ code int
+ msg string
+ }
+ rawEdit := func(w *ssWorld, edit func(m map[string]any)) {
+ key := provision.RecordKey("bab", "shop", offlineHost)
+ raw, etag, err := w.fx.rel.Get(context.Background(), key)
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ var m map[string]any
+ if err := json.Unmarshal(raw, &m); err != nil {
+ w.t.Fatal(err)
+ }
+ edit(m)
+ body, _ := json.Marshal(m)
+ if _, err := w.fx.rel.Put(context.Background(), key, body, etag); err != nil {
+ w.t.Fatal(err)
+ }
+ }
+ long := strings.Repeat("é", 81)
+ cases := []tc{
+ {name: "recovery mode admits nothing", setup: func(w *ssWorld) {
+ w.fx.srv.cfg.SelfServeMoves = ""
+ w.fx.srv.selfServeRecovery.Store(true)
+ }, body: ssBody, code: 409, msg: "self-serve moves are not enabled on this console"},
+ {name: "a corrupt record", setup: func(w *ssWorld) {
+ offlinePut(w.t, w.fx.rel, provision.RecordKey("bab", "shop", offlineHost), []byte("{not a record"))
+ }, body: ssBody, code: 502, msg: "could not read the instance"},
+ {name: "a pool", setup: func(w *ssWorld) { rawEdit(w, func(m map[string]any) { m["pool"] = "shop" }) },
+ body: ssBody, code: 400, msg: mainMoveCopy().pool},
+ {name: "an open run", setup: func(w *ssWorld) {
+ rawEdit(w, func(m map[string]any) {
+ m["run"] = map[string]any{"id": "run-1", "state": provision.RunQueued, "deadline": time.Now().Add(time.Hour)}
+ })
+ }, body: ssBody, code: 409, msg: (&provision.MoveRunOpenError{Run: provision.RunState{ID: "run-1"}}).Sentence()},
+ {name: "an operator move", setup: func(w *ssWorld) { w.ok("/stage", ssBody, provision.OfflineStaging) },
+ body: ssBody, code: 409, msg: "a move to uk is already offline-staging"},
+ {name: "an open row", setup: func(w *ssWorld) { w.prepared() }, body: ssBody, code: 409, msg: "a move to uk is already preparing"},
+ {name: "a malformed body", body: `{"region":`, code: 400, msg: "malformed JSON body"},
+ {name: "a scheduled start", body: `{"region":"uk","start_at":"2026-10-24T09:00:00Z"}`, code: 400, msg: "scheduled starts are not available yet"},
+ {name: "a long display name", body: `{"region":"uk","display_name":"` + long + `"}`, code: 400, msg: "display_name must be at most 80 characters of plain text"},
+ {name: "a display name with a newline", body: `{"region":"uk","display_name":"Shop\nAccount"}`, code: 400, msg: "display_name must be at most 80 characters of plain text"},
+ {name: "a replica", setup: func(w *ssWorld) { w.edit(func(r *provision.Record) error { r.Role = "replica"; return nil }) },
+ body: ssBody, code: 400, msg: "a replica does not move. delete it and create one in the new region"},
+ {name: "a host claim", setup: func(w *ssWorld) { offlineClaimFor(t, w.fx, offlineHost, "ie-box", "ie") }, body: ssBody, code: 409, msg: msgO29},
+ {name: "an invalid region", body: `{"region":"UK!"}`, code: 400, msg: "invalid region"},
+ {name: "the current region", body: `{"region":"ie"}`, code: 409, msg: "the instance is already in ie"},
+ {name: "two target boxes", boxes: []offlineBox{ieOff("ie-box"), ukOff("uk-a"), ukOff("uk-b")}, body: ssBody, code: 409,
+ msg: "uk has 2 boxes. an offline move needs exactly one box in the region it moves to"},
+ {name: "two source boxes", boxes: []offlineBox{ieOff("ie-box"), ieOff("ie-b"), ukOff("uk-box")}, setup: func(w *ssWorld) { w.report("ie-b", nil) },
+ body: ssBody, code: 409, msg: "ie has 2 boxes. a self-serve move needs exactly one box in each region"},
+ {name: "a target that cannot stage", boxes: []offlineBox{ieOff("ie-box"), func() offlineBox { b := ukOff("uk-box"); b.NoStageCap = true; return b }()},
+ body: ssBody, code: 409, msg: "uk-box cannot stage an offline move. it runs an older release"},
+ {name: "a source that has not reported", setup: func(w *ssWorld) {
+ w.report("ie-box", func(b *status.Box) { b.GeneratedAt = time.Now().Add(-time.Hour) })
+ }, body: ssBody, code: 409, msg: "the box ie-box has not reported recently"},
+ {name: "a target that has not reported", setup: func(w *ssWorld) {
+ w.report("uk-box", func(b *status.Box) { b.Metrics = nil })
+ }, body: ssBody, code: 409, msg: "the box uk-box has not reported recently"},
+ {name: "an identity record that does not verify", setup: func(w *ssWorld) {
+ _, raw := w.signer.Record(t, "uk-box", moveTestAccount, "eu-west-2", "i-other")
+ offlinePut(t, w.fx.rel, boxid.Key("uk-box"), raw)
+ }, body: ssBody, code: 409, msg: "box uk-box has an identity record that does not verify"},
+ {name: "no facts for the host", setup: func(w *ssWorld) {
+ w.report("ie-box", func(b *status.Box) { b.Metrics.Lanes = []string{"data-uk"} })
+ }, body: ssBody, code: 409, msg: "could not measure the account's files. try again in a minute"},
+ {name: "over the cap", setup: func(w *ssWorld) { w.facts(func(f *status.HostFacts) { f.OverCap = true }) },
+ body: ssBody, code: 409, msg: "this account's files are too large to move this way"},
+ {name: "over the cap on a cut walk", setup: func(w *ssWorld) {
+ w.facts(func(f *status.HostFacts) { f.OverCap, f.Truncated, f.Known = true, true, false })
+ }, body: ssBody, code: 409, msg: "this account's files are too large to move this way"},
+ {name: "facts not known", setup: func(w *ssWorld) { w.facts(func(f *status.HostFacts) { f.Known = false }) },
+ body: ssBody, code: 409, msg: "could not measure the account's files. try again in a minute"},
+ {name: "facts of another lifetime", setup: func(w *ssWorld) { w.facts(func(f *status.HostFacts) { f.Incarnation++ }) },
+ body: ssBody, code: 409, msg: "the source box has not reported recently. try again in a minute"},
+ {name: "facts collected long ago", setup: func(w *ssWorld) {
+ w.facts(func(f *status.HostFacts) { f.CollectedAt = time.Now().Add(-11 * time.Minute).Unix() })
+ }, body: ssBody, code: 409, msg: "the source box has not reported recently. try again in a minute"},
+ {name: "a replica listing from the future", setup: func(w *ssWorld) {
+ w.facts(func(f *status.HostFacts) { f.ReplicaCollectedAt = time.Now().Add(2 * time.Minute).Unix() })
+ }, body: ssBody, code: 409, msg: "the source box has not reported recently. try again in a minute"},
+ {name: "a truncated walk", setup: func(w *ssWorld) { w.facts(func(f *status.HostFacts) { f.Truncated = true }) },
+ body: ssBody, code: 409, msg: "could not measure the account's files. try again in a minute"},
+ {name: "no lane", setup: func(w *ssWorld) {
+ rec := w.record()
+ now := time.Now().Unix()
+ w.report("ie-box", func(b *status.Box) {
+ b.Metrics.Lanes = []string{"data-se"}
+ b.Metrics.Hosts = map[string]status.HostFacts{offlineHost: {CollectedAt: now, Known: true, Incarnation: rec.Incarnation, ReplicaCollectedAt: now}}
+ })
+ }, body: ssBody, code: 409, msg: "no staging lane is configured from ie to uk"},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ w := newSelfServeWorld(t, false, c.boxes...)
+ if c.setup != nil {
+ c.setup(w)
+ }
+ serial := serialBytes(t, w.fx.rel)
+ w.refusedSS(c.body, c.code, c.msg)
+ if serialBytes(t, w.fx.rel) != serial {
+ t.Fatal("a refusal bumped the serial")
+ }
+ })
+ }
+ t.Run("a host with no record", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ code, h, out := rawReq(t, http.MethodPost, strings.Replace(w.move, offlineHost, "other.shop.example.com", 1), w.owner, ssBody)
+ assertNotFound(t, "prepare of a host with no record", canonicalNotFound(t), code, h, out)
+ })
+}
+
+func TestPrepareWritesTheRowFirst(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ rec := w.record()
+ before := bucketState(t, w.fx.rel)
+ start := time.Now()
+ code, out := w.prepare(`{"region":"uk","display_name":"Shop's account","requested_by":"shop-admin"}`)
+ if code != http.StatusAccepted {
+ t.Fatalf("prepare = %d %s", code, out)
+ }
+ v := ssView(t, out)
+ if v.State != "preparing" || v.Kind != "offline" || v.Target != "uk" || v.ID == "" || v.EstimateSeconds != 120 {
+ t.Fatalf("view = %+v", v)
+ }
+ // The one write is the row, and the serial bump announces it: the
+ // record, the reservations and every other object are as they were.
+ after := bucketState(t, w.fx.rel)
+ for k, b := range after {
+ if k != "control/serial" && before[k] != b {
+ t.Errorf("the prepare wrote %s", k)
+ }
+ }
+ if after["control/serial"] == before["control/serial"] {
+ t.Error("the prepare did not bump the serial")
+ }
+ if got := w.record(); got.Move != nil || got.RecordSeq != rec.RecordSeq {
+ t.Fatalf("the prepare wrote the record: %+v", got.Move)
+ }
+ row, err := w.fx.st.SelfServeMove(v.ID)
+ if err != nil || row == nil {
+ t.Fatal(row, err)
+ }
+ want := provision.Participants{
+ Source: []provision.Participant{{BoxID: "ie-box", Label: "ie-box", InstanceID: "i-ie-box"}},
+ Target: provision.Participant{BoxID: "uk-box", Label: "uk-box", InstanceID: "i-uk-box"},
+ }
+ if row.Phase != ssRequested || row.Incarnation != rec.Incarnation || row.Incarnation == 0 || row.From != "ie" || row.To != "uk" ||
+ row.DisplayName != "Shop's account" || row.RequestedBy != "token:"+hashToken(w.owner)+" shop-admin" ||
+ fmt.Sprint(row.Participants) != fmt.Sprint(want) || row.EstimateSeconds != 120 || row.EstimateTerms == "" {
+ t.Fatalf("row = %+v", row)
+ }
+ if d := row.WindowDeadline.Sub(start); d < 24*time.Hour-time.Minute || d > 24*time.Hour+time.Minute {
+ t.Fatalf("window deadline %v after the prepare", d)
+ }
+ if terms := estimateTermsOf(row.EstimateTerms); terms.Copy <= 10*time.Second {
+ t.Fatalf("stored terms %+v", terms)
+ }
+ // The poll shows the row.
+ code, out = w.do(http.MethodGet, "", w.plain, "")
+ if code != http.StatusOK || ssView(t, out).State != "preparing" || ssView(t, out).ID != v.ID {
+ t.Fatalf("status = %d %s", code, out)
+ }
+}
+
+func TestPrepareMigratesALegacyRecordBoundThenAdmits(t *testing.T) {
+ t.Run("migrated, then admitted at the new lifetime", func(t *testing.T) {
+ w := newSelfServeWorld(t, true)
+ if rec := w.record(); rec.Incarnation != 0 {
+ t.Fatalf("the legacy record has incarnation %d", rec.Incarnation)
+ }
+ row := w.prepared()
+ rec := w.record()
+ if rec.Incarnation == 0 || row.Incarnation != rec.Incarnation || rec.Move != nil {
+ t.Fatalf("record incarnation %d move %+v, row incarnation %d", rec.Incarnation, rec.Move, row.Incarnation)
+ }
+ })
+ t.Run("a record that changed since it was observed is not migrated", func(t *testing.T) {
+ w := newSelfServeWorld(t, true)
+ h := func(door string) {
+ if door == "migrate" {
+ w.edit(func(r *provision.Record) error { r.Channel = "edge"; return nil })
+ }
+ }
+ selfServeDoorHook.Store(&h)
+ t.Cleanup(func() { selfServeDoorHook.Store(nil) })
+ code, out := w.prepare(ssBody)
+ if code != http.StatusConflict || errorText(out) != msgLifetimeSettling {
+ t.Fatalf("prepare = %d %s, want 409 %q", code, out, msgLifetimeSettling)
+ }
+ if open, _ := w.rows(); open != nil {
+ t.Fatalf("a row was admitted: %+v", open)
+ }
+ if rec := w.record(); rec.Incarnation != 0 {
+ t.Fatalf("the moved record was migrated to %d", rec.Incarnation)
+ }
+ })
+ t.Run("before the cutover a legacy record is settling", func(t *testing.T) {
+ w := newOfflineWorld(t, oneEach...)
+ w.fx.srv.cfg.SelfServeMoves = "offline"
+ owner, err := w.fx.st.NewAppToken(OpsAccountID, "shop", "paul@keymail.dev", "shop-mover", true, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := w.record()
+ now := time.Now().Unix()
+ for _, b := range []string{"ie-box", "uk-box"} {
+ sb := status.Box{Label: b, GeneratedAt: time.Now(), Metrics: &status.BoxMetrics{BoxID: b, InstanceID: "i-" + b, Lanes: []string{"data-uk"},
+ Hosts: map[string]status.HostFacts{offlineHost: {CollectedAt: now, Known: true, Incarnation: rec.Incarnation, ReplicaCollectedAt: now}}}}
+ if err := status.WriteBox(context.Background(), w.fx.rel, sb); err != nil {
+ t.Fatal(err)
+ }
+ }
+ code, _, out := rawReq(t, http.MethodPost, strings.TrimSuffix(w.base, "/offline"), owner, ssBody)
+ if code != http.StatusConflict || errorText(out) != msgLifetimeSettling {
+ t.Fatalf("prepare before the cutover = %d %s", code, out)
+ }
+ })
+}
+
+// ---- cut and cancel ------------------------------------------------------------
+
+func TestMutationsNameTheMoveAndRace(t *testing.T) {
+ cut := func(w *ssWorld, id string) (int, string) {
+ return w.do(http.MethodPost, "/cut", w.owner, `{"id":"`+id+`"}`)
+ }
+ cancel := func(w *ssWorld, id string) (int, string) {
+ return w.do(http.MethodDelete, "", w.owner, `{"id":"`+id+`"}`)
+ }
+ want := func(t *testing.T, what string, code int, out string, wantCode int, wantMsg string) {
+ t.Helper()
+ if code != wantCode || (wantMsg != "" && errorText(out) != wantMsg) {
+ t.Fatalf("%s = %d %s, want %d %q", what, code, out, wantCode, wantMsg)
+ }
+ }
+
+ t.Run("the move is named", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ w.setPhase(row.ID, ssReady)
+ for _, id := range []string{"", "mv_notthemoveyouconfirmed1", strings.ToUpper(row.ID)} {
+ code, out := cut(w, id)
+ want(t, "cut "+id, code, out, http.StatusConflict, "this is not the move you confirmed")
+ code, out = cancel(w, id)
+ want(t, "cancel "+id, code, out, http.StatusConflict, "this is not the move you confirmed")
+ }
+ if got, _ := w.fx.st.SelfServeMove(row.ID); !got.CutRequestedAt.IsZero() || !got.AbortRequestedAt.IsZero() {
+ t.Fatalf("a refused mutation wrote the row: %+v", got)
+ }
+ })
+ t.Run("a cut before ready", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ w.setPhase(row.ID, ssStaging)
+ code, out := cut(w, row.ID)
+ want(t, "cut in staging", code, out, http.StatusConflict, "the move is not ready to cut yet. it is staging")
+ })
+ t.Run("cut, cut again, then cancel", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ w.queued(row)
+ w.setPhase(row.ID, ssReady)
+ code, out := cut(w, row.ID)
+ want(t, "cut", code, out, http.StatusAccepted, "")
+ if v := ssView(t, out); v.State != "cutting" || v.ID != row.ID {
+ t.Fatalf("cut view = %+v", v)
+ }
+ got, _ := w.fx.st.SelfServeMove(row.ID)
+ if got.CutRequestedAt.IsZero() || !got.WindowDeadline.IsZero() || got.StepDeadline.Sub(got.CutRequestedAt) != selfServeCutCopyFloor {
+ t.Fatalf("the cut's row = cut %v window %v step %v", got.CutRequestedAt, got.WindowDeadline, got.StepDeadline)
+ }
+ code, out = cut(w, row.ID)
+ want(t, "a second cut", code, out, http.StatusAccepted, "")
+ w.setPhase(row.ID, ssSealing)
+ code, out = cut(w, row.ID)
+ want(t, "a cut while cutting", code, out, http.StatusAccepted, "")
+ code, out = cancel(w, row.ID)
+ want(t, "cancel after the cut", code, out, http.StatusConflict, "the move was confirmed and cannot be cancelled. it completes or returns on its own")
+ })
+ t.Run("cancel, cancel again, then cut", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ code, out := cancel(w, row.ID)
+ want(t, "cancel while preparing", code, out, http.StatusAccepted, "")
+ if v := ssView(t, out); v.Abort == nil || v.Abort.RequestedAt == 0 {
+ t.Fatalf("cancel view = %+v", v)
+ }
+ code, out = cancel(w, row.ID)
+ want(t, "a second cancel", code, out, http.StatusAccepted, "")
+ w.setPhase(row.ID, ssReady)
+ code, out = cut(w, row.ID)
+ want(t, "cut after the cancel", code, out, http.StatusConflict, "the move was cancelled")
+ })
+ // The barrier, both orders: one door parked between its read of the row
+ // and its write while the other door goes through.
+ parked := func(t *testing.T, w *ssWorld, door string, run func()) (release func()) {
+ t.Helper()
+ at, resume := make(chan struct{}), make(chan struct{})
+ var once sync.Once
+ h := func(d string) {
+ if d == door {
+ once.Do(func() { close(at); <-resume })
+ }
+ }
+ selfServeDoorHook.Store(&h)
+ t.Cleanup(func() { selfServeDoorHook.Store(nil) })
+ done := make(chan struct{})
+ go func() { defer close(done); run() }()
+ <-at
+ return func() { close(resume); <-done; selfServeDoorHook.Store(nil) }
+ }
+ t.Run("the cut parked, the cancel through", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ w.setPhase(row.ID, ssReady)
+ var code int
+ var out string
+ resume := parked(t, w, "cut", func() { code, out = cut(w, row.ID) })
+ c2, o2 := cancel(w, row.ID)
+ want(t, "the cancel", c2, o2, http.StatusAccepted, "")
+ resume()
+ want(t, "the parked cut", code, out, http.StatusConflict, "the move was cancelled")
+ if got, _ := w.fx.st.SelfServeMove(row.ID); !got.CutRequestedAt.IsZero() {
+ t.Fatal("the cut landed after the cancel")
+ }
+ })
+ t.Run("the cancel parked, the cut through", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ w.setPhase(row.ID, ssReady)
+ var code int
+ var out string
+ resume := parked(t, w, "abort", func() { code, out = cancel(w, row.ID) })
+ c2, o2 := cut(w, row.ID)
+ want(t, "the cut", c2, o2, http.StatusAccepted, "")
+ resume()
+ want(t, "the parked cancel", code, out, http.StatusConflict, "the move was confirmed and cannot be cancelled. it completes or returns on its own")
+ if got, _ := w.fx.st.SelfServeMove(row.ID); !got.AbortRequestedAt.IsZero() {
+ t.Fatal("the cancel landed after the cut")
+ }
+ })
+ t.Run("a stale snapshot: the confirmation expired under the cut", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ w.setPhase(row.ID, ssReady)
+ var code int
+ var out string
+ resume := parked(t, w, "cut", func() { code, out = cut(w, row.ID) })
+ cur, _ := w.fx.st.SelfServeMove(row.ID)
+ if ok, err := w.fx.st.SetSelfServeExpiry(row.ID, ssReady, "window", cur.WindowDeadline, true); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ resume()
+ want(t, "the cut after the expiry", code, out, http.StatusConflict, "the move expired. prepare again")
+ })
+ t.Run("a stale snapshot: the move finished under the cancel", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ var code int
+ var out string
+ resume := parked(t, w, "abort", func() { code, out = cancel(w, row.ID) })
+ w.setPhase(row.ID, ssDone)
+ if err := w.fx.st.FinishSelfServeMove(row.ID, "done", "", "", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ resume()
+ want(t, "the cancel after the finish", code, out, http.StatusConflict, "no move is in flight")
+ })
+ t.Run("a stale nonce cannot confirm or consume", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ st := w.fx.st
+ now := time.Now()
+ issue := func(nonce string) {
+ t.Helper()
+ ok, _, err := st.IssueSelfServeRequest(row.ID, "queue", ssRequested, persistedRequest{Nonce: nonce, Proposed: now}, now)
+ if err != nil || !ok {
+ t.Fatalf("issue %s: %v %v", nonce, ok, err)
+ }
+ }
+ issue("nonce-a")
+ if ok, err := st.ConfirmSelfServeRequest(row.ID, "nonce-a", now); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if ok, err := st.ConsumeSelfServeRequest(row.ID, "nonce-a", "", time.Time{}); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ issue("nonce-b")
+ if ok, _ := st.ConfirmSelfServeRequest(row.ID, "nonce-a", now); ok {
+ t.Fatal("the stale nonce confirmed the new request")
+ }
+ if ok, _ := st.ConsumeSelfServeRequest(row.ID, "nonce-a", "", time.Time{}); ok {
+ t.Fatal("the stale nonce consumed the new request")
+ }
+ if ok, _ := st.ConsumeSelfServeRequest(row.ID, "nonce-b", "", time.Time{}); ok {
+ t.Fatal("an unconfirmed request was consumed")
+ }
+ if ok, err := st.ConfirmSelfServeRequest(row.ID, "nonce-b", now); err != nil || !ok {
+ t.Fatal("the request's own nonce does not confirm it", ok, err)
+ }
+ })
+ t.Run("a cancel after the cut cannot win in the store", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ w.setPhase(row.ID, ssReady)
+ if ok, err := w.fx.st.MarkSelfServeCut(row.ID, time.Now(), time.Now().Add(time.Hour)); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if ok, _ := w.fx.st.RequestSelfServeAbort(row.ID, selfServeCancellable, time.Now()); ok {
+ t.Fatal("RequestSelfServeAbort won after the cut")
+ }
+ })
+}
+
+// ---- the view -------------------------------------------------------------------
+
+func TestViewPrecedence(t *testing.T) {
+ s := &Server{}
+ now := time.Date(2026, 10, 23, 12, 0, 0, 0, time.UTC)
+ offlineAt := now.Add(-time.Minute)
+ ss := func(id string) *provision.MoveState {
+ return &provision.MoveState{Kind: provision.MoveKindOffline, ID: id, To: "uk", State: provision.OfflineStaging, UpdatedAt: now,
+ SelfServe: &provision.SelfServeMove{ID: id, Incarnation: 3, OfflineAt: &offlineAt}}
+ }
+ row := func(id, phase string, inc uint64) *SelfServeMove {
+ return &SelfServeMove{ID: id, Incarnation: inc, From: "ie", To: "uk", Phase: phase, EstimateSeconds: 300, UpdatedAt: now}
+ }
+ term := func(id string, inc uint64, at time.Time) *SelfServeMove {
+ r := row(id, ssDone, inc)
+ r.Outcome, r.OutcomeReason, r.UpdatedAt = "aborted", "cancelled", at
+ return r
+ }
+ rec := func(m *provision.MoveState) *provision.Record {
+ return &provision.Record{Host: offlineHost, Incarnation: 3, Move: m}
+ }
+ js := func(v any) string { b, _ := json.Marshal(v); return string(b) }
+ today := func(r *provision.Record) string { return js(s.moveViewOf(r)) }
+ operator := &provision.MoveState{Kind: provision.MoveKindOffline, ID: "mv_operator", To: "uk", State: provision.OfflineStopping, UpdatedAt: now}
+ cases := []struct {
+ name string
+ rec *provision.Record
+ open, terminal *SelfServeMove
+ want string
+ }{
+ {"1: the record's self-serve move and its row", rec(ss("mv_a")), row("mv_a", ssStaging, 3), nil,
+ js(selfServeMoveView{State: "preparing", Target: "uk", UpdatedAt: now.Unix(), Kind: "offline", ID: "mv_a", EstimateSeconds: 300, OfflineAt: offlineAt.Unix()})},
+ {"1: ready", rec(ss("mv_a")), row("mv_a", ssReady, 3), nil,
+ js(selfServeMoveView{State: "ready", Target: "uk", UpdatedAt: now.Unix(), Kind: "offline", ID: "mv_a", EstimateSeconds: 300, OfflineAt: offlineAt.Unix()})},
+ {"1: cutting from restage", rec(ss("mv_a")), row("mv_a", ssRestage, 3), nil,
+ js(selfServeMoveView{State: "cutting", Target: "uk", UpdatedAt: now.Unix(), Kind: "offline", ID: "mv_a", EstimateSeconds: 300, OfflineAt: offlineAt.Unix()})},
+ {"2: an operator move over an open row", rec(operator), row("mv_a", ssStaging, 3), nil, today(rec(operator))},
+ {"2: a self-serve move whose row is another", rec(ss("mv_b")), row("mv_a", ssStaging, 3), nil, today(rec(ss("mv_b")))},
+ {"2: a self-serve move of another lifetime's row", rec(ss("mv_a")), row("mv_a", ssStaging, 2), nil, today(rec(ss("mv_a")))},
+ {"3: the row before its queue", rec(nil), row("mv_a", ssReady, 3), nil,
+ js(selfServeMoveView{State: "preparing", Target: "uk", UpdatedAt: now.Unix(), Kind: "offline", ID: "mv_a", EstimateSeconds: 300})},
+ {"3 not: a recreation (the row's lifetime is gone)", rec(nil), row("mv_a", ssRequested, 2), nil, today(rec(nil))},
+ {"4: a terminal row within seven days", rec(nil), nil, term("mv_t", 3, now.Add(-6*24*time.Hour)),
+ js(selfServeMoveView{State: "none", Kind: "offline", Last: &selfServeLast{ID: "mv_t", Outcome: "aborted", Reason: "cancelled", From: "ie", To: "uk", At: now.Add(-6 * 24 * time.Hour).Unix()}})},
+ {"4 not: older than seven days", rec(nil), nil, term("mv_t", 3, now.Add(-8*24*time.Hour)), today(rec(nil))},
+ {"4 not: another lifetime's terminal row", rec(nil), nil, term("mv_t", 2, now), today(rec(nil))},
+ {"4 not: a record that carries a move", rec(operator), nil, term("mv_t", 3, now), today(rec(operator))},
+ {"5: nothing", rec(nil), nil, nil, today(rec(nil))},
+ }
+ for _, c := range cases {
+ if got := js(s.selfServeViewOf(c.rec, c.open, c.terminal, now)); got != c.want {
+ t.Errorf("%s:\n got %s\n want %s", c.name, got, c.want)
+ }
+ }
+
+ // Through the status door: a terminal row is rendered after the switch
+ // goes off, after a restart and on a later poll within seven days, and
+ // not once it is older.
+ w := newSelfServeWorld(t, false)
+ r := w.prepared()
+ w.setPhase(r.ID, ssDone)
+ if err := w.fx.st.FinishSelfServeMove(r.ID, "aborted", "cancelled", "", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ // The finaliser's effects and release are done (Task 8), so nothing
+ // keeps the console in recovery mode.
+ if err := w.fx.st.SetSelfServeEffectsDone(r.ID); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.fx.st.ClearSelfServeRelease(r.ID); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := w.fx.st.db.Exec(`UPDATE self_serve_events SET emitted = 1`); err != nil {
+ t.Fatal(err)
+ }
+ cfg := w.fx.srv.cfg
+ cfg.SelfServeMoves = ""
+ srv, err := NewServer(cfg, w.fx.st)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if srv.selfServeMode() {
+ t.Fatal("the restarted console is in self-serve mode")
+ }
+ ts := httptest.NewServer(srv)
+ t.Cleanup(ts.Close)
+ poll := func() selfServeMoveView {
+ t.Helper()
+ code, _, out := rawReq(t, http.MethodGet, ts.URL+strings.TrimPrefix(w.move, w.fx.ts.URL), w.plain, "")
+ if code != http.StatusOK {
+ t.Fatalf("status = %d %s", code, out)
+ }
+ return ssView(t, out)
+ }
+ if v := poll(); v.State != "none" || v.Last == nil || v.Last.ID != r.ID || v.Last.Outcome != "aborted" || v.Last.From != "ie" || v.Last.To != "uk" {
+ t.Fatalf("the poll after switch-off and a restart = %+v", v)
+ }
+ if _, err := w.fx.st.db.Exec(`UPDATE self_serve_moves SET updated_at = ? WHERE id = ?`, time.Now().Add(-8*24*time.Hour).UnixNano(), r.ID); err != nil {
+ t.Fatal(err)
+ }
+ if v := poll(); v.Last != nil || v.Kind != "" {
+ t.Fatalf("an eight-day-old move still shows: %+v", v)
+ }
+}
+
+// ---- the estimate, the poll and the switch -------------------------------------
+
+func TestEstimateDoorWritesNothing(t *testing.T) {
+ w := newSelfServeWorld(t, true)
+ before := bucketState(t, w.fx.rel)
+ for _, tok := range []string{w.owner, w.plain} {
+ code, out := w.do(http.MethodGet, "?estimate=1&region=uk", tok, "")
+ if code != http.StatusOK || strings.TrimSpace(out) != `{"estimate_seconds":120,"files_bytes":1048576,"replica_bytes":3145728}` {
+ t.Fatalf("estimate = %d %q", code, out)
+ }
+ }
+ // A refusal is the prepare's, and also writes nothing.
+ w.facts(func(f *status.HostFacts) { f.OverCap = true })
+ before = bucketState(t, w.fx.rel)
+ if code, out := w.do(http.MethodGet, "?estimate=1&region=uk", w.plain, ""); code != http.StatusConflict || errorText(out) != "this account's files are too large to move this way" {
+ t.Fatalf("estimate over the cap = %d %s", code, out)
+ }
+ after := bucketState(t, w.fx.rel)
+ if len(after) != len(before) {
+ t.Fatalf("the estimate wrote: %d objects, %d before", len(after), len(before))
+ }
+ for k, v := range before {
+ if after[k] != v {
+ t.Fatalf("the estimate wrote %s", k)
+ }
+ }
+ if rec := w.record(); rec.Incarnation != 0 {
+ t.Fatalf("the estimate migrated the legacy record to %d", rec.Incarnation)
+ }
+ if open, _ := w.rows(); open != nil {
+ t.Fatalf("the estimate admitted a row: %+v", open)
+ }
+ // With the switch off the query is ignored: today's view.
+ w.fx.srv.cfg.SelfServeMoves = ""
+ if code, out := w.do(http.MethodGet, "?estimate=1&region=uk", w.plain, ""); code != http.StatusOK || strings.TrimSpace(out) != `{"state":"none"}` {
+ t.Fatalf("estimate with the switch off = %d %s", code, out)
+ }
+}
+
+func TestStatusStaysOpenToPlainInstanceTokens(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ if code, out := w.do(http.MethodGet, "", w.plain, ""); code != http.StatusOK || strings.TrimSpace(out) != `{"state":"none"}` {
+ t.Fatalf("a plain token's poll = %d %s", code, out)
+ }
+ row := w.prepared()
+ code, out := w.do(http.MethodGet, "", w.plain, "")
+ if v := ssView(t, out); code != http.StatusOK || v.State != "preparing" || v.ID != row.ID {
+ t.Fatalf("a plain token's poll of a prepared move = %d %s", code, out)
+ }
+ // The mutations stay closed to it.
+ for _, d := range selfServeDoors {
+ code, h, body := rawReq(t, d.method, w.move+strings.TrimPrefix(d.suffix, "/move"), w.plain, d.body)
+ assertNotFound(t, "a plain token's "+d.name, canonicalNotFound(t), code, h, body)
+ }
+ // A failing row read is the 502, never a silent view.
+ w.fx.st.db.Exec(`ALTER TABLE self_serve_moves RENAME TO self_serve_moves_gone`)
+ if code, out := w.do(http.MethodGet, "", w.plain, ""); code != http.StatusBadGateway || errorText(out) != "could not read the move" {
+ t.Fatalf("a poll with the rows unreadable = %d %s", code, out)
+ }
+}
+
+func TestSwitchOffIsByteIdentical(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ w.fx.srv.cfg.SelfServeMoves = ""
+ if w.fx.srv.selfServeMode() {
+ t.Fatal("self-serve mode with the switch off")
+ }
+ // The poll is the flat view, with or without the estimate's query.
+ want, _ := json.Marshal(w.fx.srv.moveViewOf(w.record()))
+ for _, q := range []string{"", "?estimate=1&region=uk"} {
+ code, out := w.do(http.MethodGet, q, w.plain, "")
+ if code != http.StatusOK || strings.TrimSpace(out) != string(want) {
+ t.Fatalf("poll %q = %d %s, want %s", q, code, out, want)
+ }
+ }
+ // The mutations are the legacy doors: the plain instance token reaches
+ // them, and a move-capable one is answered as any instance token.
+ quiet := provision.MoveMarkCutting()(&provision.Record{Schema: 1, Host: offlineHost})
+ if quiet == nil {
+ t.Fatal("a quiet record accepted the cut")
+ }
+ for _, tok := range []string{w.plain, w.owner} {
+ code, out := w.do(http.MethodPost, "/cut", tok, "")
+ if code != http.StatusConflict || errorText(out) != quiet.Error() {
+ t.Fatalf("legacy cut = %d %s, want 409 %q", code, out, quiet.Error())
+ }
+ }
+ if open, _ := w.rows(); open != nil {
+ t.Fatal("a legacy door admitted a row")
+ }
+}