amadan

carlos / platform Public

console: tests for the self-serve journal, the lease, recovery mode and the pass

The store's round trip, atomic admission, the request lifecycle's
predicates, the request deadline by kind, the cut and expiry races,
per-box FIFO, sequenced events delivered once and transactionally, the
open-row order; the lease excluding a second console and bounding every
write attempt; the pass starting on a failed read and retrying, held on by
each kind of evidence, draining events and leaving rows it cannot
advance; and the move-capable gate's matrix in recovery mode.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 0d237b37b5d7deca43e036f29f62c3a8f85b7688 parent f618773
3 files changed, +1444 −0
  • internal/console/selfserve_gate_test.go +30 −0
  • internal/console/selfserve_pass_test.go +398 −0
  • internal/console/selfserve_store_test.go +1016 −0
diff --git a/internal/console/selfserve_gate_test.go b/internal/console/selfserve_gate_test.go
index 128071277..65098c846 100644
--- a/internal/console/selfserve_gate_test.go
+++ b/internal/console/selfserve_gate_test.go
@@ -408,6 +408,17 @@ func TestMoveCapableGateAtTheRoutes(t *testing.T) {
if got := s.selfServeMode(); got != c.want {
t.Errorf("selfServeMode(%q, claims %v, phase 1 %v) = %v, want %v", c.sw, c.hostClaim, c.phase, got, c.want)
}
+ // Admission is the switch alone.
+ if got := s.selfServeAdmits(); got != c.want {
+ t.Errorf("selfServeAdmits(%q, claims %v, phase 1 %v) = %v, want %v", c.sw, c.hostClaim, c.phase, got, c.want)
+ }
+ // Recovery mode is self-serve mode whatever the switch, and
+ // admits nothing new.
+ s.selfServeRecovery.Store(true)
+ if !s.selfServeMode() || s.selfServeAdmits() != c.want {
+ t.Errorf("recovery mode (%q, claims %v, phase 1 %v): mode %v admits %v, want true %v",
+ c.sw, c.hostClaim, c.phase, s.selfServeMode(), s.selfServeAdmits(), c.want)
+ }
}
})
@@ -473,6 +484,25 @@ func TestMoveCapableGateAtTheRoutes(t *testing.T) {
requireUnauthorisedCostNothing(t, before, c, call)
})
+ // The same in recovery mode: the switch off, self-serve mode on through
+ // an open self-serve row the console found at start.
+ t.Run("unauthorised callers in recovery mode", func(t *testing.T) {
+ rec := newMoveScene(t)
+ srv := rec.l.fx.srv
+ srv.cfg.SelfServeMoves = ""
+ ok, _, err := rec.l.fx.st.AdmitSelfServeMove(SelfServeMove{ID: "mv_recoveryrecoveryrecovery", Account: "bab", App: ltApp,
+ Host: ltOne, From: "ie", To: "uk", Participants: ssParticipants("ie-1", "uk-1"), CreatedAt: time.Now()})
+ if err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ srv.selfServeScanAtStart(context.Background())
+ if !srv.selfServeMode() || srv.selfServeAdmits() {
+ t.Fatalf("mode %v admits %v; want recovery mode without admission", srv.selfServeMode(), srv.selfServeAdmits())
+ }
+ c, call := rec.counting(t)
+ requireUnauthorisedCostNothing(t, rec, c, call)
+ })
+
// The owner's move-capable token passes the gate, then meets the
// instance gate's rails exactly as the instance gate gives them.
t.Run("the owner's token meets the rails", func(t *testing.T) {
diff --git a/internal/console/selfserve_pass_test.go b/internal/console/selfserve_pass_test.go
new file mode 100644
index 000000000..214ed41bb
--- /dev/null
+++ b/internal/console/selfserve_pass_test.go
@@ -0,0 +1,398 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "reflect"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The console lease, recovery mode and the pass skeleton (self-serve moves
+// plan 2 Task 4).
+
+// ssFaultStore fails Gets of one key while failGet is set, and counts the
+// record Puts that reached the bucket.
+type ssFaultStore struct {
+ release.Store
+ key string
+ failGet atomic.Bool
+ recordPuts atomic.Int32
+}
+
+func (f *ssFaultStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if f.failGet.Load() && key == f.key {
+ return nil, "", errors.New("injected read failure")
+ }
+ return f.Store.Get(ctx, key)
+}
+
+func (f *ssFaultStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if provision.IsRecordKey(key) {
+ f.recordPuts.Add(1)
+ }
+ return f.Store.Put(ctx, key, body, ifMatch)
+}
+
+// ssClock is a settable clock for the lease.
+type ssClock struct {
+ mu sync.Mutex
+ t time.Time
+}
+
+func (c *ssClock) now() time.Time { c.mu.Lock(); defer c.mu.Unlock(); return c.t }
+func (c *ssClock) advance(d time.Duration) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.t = c.t.Add(d)
+}
+
+func ssLeaseBody(t *testing.T, rel release.Store) (consoleLease, bool) {
+ t.Helper()
+ raw, _, err := rel.Get(context.Background(), selfServeLeaseKey)
+ if errors.Is(err, os.ErrNotExist) {
+ return consoleLease{}, false
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l consoleLease
+ if err := json.Unmarshal(raw, &l); err != nil {
+ t.Fatal(err)
+ }
+ return l, true
+}
+
+func TestConsoleLeaseExcludesASecondConsole(t *testing.T) {
+ if selfServeLeaseKey != "control/offline-moves/console.json" {
+ t.Fatalf("the lease key is %q", selfServeLeaseKey)
+ }
+ rel := release.NewDirStore(t.TempDir())
+ clock := &ssClock{t: time.Now()}
+ a := &Server{cfg: Config{Releases: rel}, selfServeClock: clock.now}
+ b := &Server{cfg: Config{Releases: rel}, selfServeClock: clock.now}
+ ctx := context.Background()
+
+ ta, ok, err := a.acquireSelfServeLease(ctx, clock.now())
+ if err != nil || !ok || ta == nil {
+ t.Fatalf("first console = %v, %v", ok, err)
+ }
+ if !ta.valid(clock.now()) || !ta.deadline().Equal(clock.now().Add(selfServeLeaseTTL)) {
+ t.Fatalf("lease valid %v until %v, want %v", ta.valid(clock.now()), ta.deadline(), clock.now().Add(selfServeLeaseTTL))
+ }
+ if l, ok := ssLeaseBody(t, rel); !ok || l.Console == "" || !l.Until.Equal(ta.deadline()) {
+ t.Fatalf("the lease object = %+v (%v)", l, ok)
+ }
+ if tb, ok, err := b.acquireSelfServeLease(ctx, clock.now()); err != nil || ok || tb != nil {
+ t.Fatalf("second console while held = %v, %v", ok, err)
+ }
+ // Renewed every 20 s: still the first console's.
+ clock.advance(selfServeLeaseRenewEvery)
+ if err := ta.renew(ctx, clock.now()); err != nil {
+ t.Fatal(err)
+ }
+ clock.advance(50 * time.Second) // 70 s after the acquire, 50 s after the renewal
+ if !ta.valid(clock.now()) {
+ t.Fatal("a renewed lease ran out early")
+ }
+ if _, ok, _ := b.acquireSelfServeLease(ctx, clock.now()); ok {
+ t.Fatal("the second console took a renewed lease")
+ }
+ // The first console stops renewing: past its until, the second takes
+ // over and the first's renewal loses.
+ clock.advance(11 * time.Second)
+ if ta.valid(clock.now()) {
+ t.Fatal("an expired lease is still valid")
+ }
+ tb, ok, err := b.acquireSelfServeLease(ctx, clock.now())
+ if err != nil || !ok {
+ t.Fatalf("second console after expiry = %v, %v", ok, err)
+ }
+ if err := ta.renew(ctx, clock.now()); !errors.Is(err, errSelfServeLeaseLost) {
+ t.Fatalf("the first console's renewal = %v, want lost", err)
+ }
+ if ta.valid(clock.now()) || !tb.valid(clock.now()) {
+ t.Fatal("after the takeover the first lease is valid or the second is not")
+ }
+ if _, ok, _ := a.acquireSelfServeLease(ctx, clock.now()); ok {
+ t.Fatal("the first console took the lease back while the second holds it")
+ }
+ // A malformed lease is never taken over blind.
+ _, etag, err := rel.Get(ctx, selfServeLeaseKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := rel.Put(ctx, selfServeLeaseKey, []byte(`not json`), etag); err != nil {
+ t.Fatal(err)
+ }
+ clock.advance(2 * selfServeLeaseTTL)
+ if _, ok, err := a.acquireSelfServeLease(ctx, clock.now()); ok || err == nil {
+ t.Fatalf("a malformed lease = %v, %v; want an error and no lease", ok, err)
+ }
+}
+
+func TestLeaseBoundsEveryWriteAttempt(t *testing.T) {
+ rel := &ssFaultStore{Store: release.NewDirStore(t.TempDir())}
+ clock := &ssClock{t: time.Now()}
+ s := &Server{cfg: Config{Releases: rel}, selfServeClock: clock.now}
+ ctx := context.Background()
+ tok, ok, err := s.acquireSelfServeLease(ctx, clock.now())
+ if err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ wctx, cancel := s.selfServeWriteCtx(ctx, tok)
+ defer cancel()
+ if dl, ok := wctx.Deadline(); !ok || !dl.Equal(tok.deadline()) {
+ t.Fatalf("the write context's deadline = %v (%v), want the lease's %v", dl, ok, tok.deadline())
+ }
+ // A record to write, created outside the pass.
+ rec := provision.Record{Schema: 1, Host: ssHost, Slug: "shop", App: "shop", Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}
+ if err := provision.CreateRecord(ctx, rel, rec); err != nil {
+ t.Fatal(err)
+ }
+ rel.recordPuts.Store(0)
+ touch := func(r *provision.Record) error { r.Channel = "beta"; return nil }
+ if _, err := provision.SetRecordMoveAt(wctx, rel, "bab", "shop", ssHost, provision.Expect{}, touch); err != nil || rel.recordPuts.Load() != 1 {
+ t.Fatalf("a bound write under a valid lease = %v, %d puts", err, rel.recordPuts.Load())
+ }
+ // Each attempt checks the lease with a fresh clock: past its until, the
+ // next attempt never reaches the bucket, whatever the context says.
+ clock.advance(selfServeLeaseTTL + time.Second)
+ if wctx.Err() != nil {
+ t.Fatal("the test's real-time context ran out; the guard is what must refuse")
+ }
+ if _, err := provision.SetRecordMoveAt(wctx, rel, "bab", "shop", ssHost, provision.Expect{Seq: 1}, touch); !errors.Is(err, errSelfServeLeaseLost) {
+ t.Fatalf("a bound write after the lease ran out = %v, want lost", err)
+ }
+ key := provision.RecordKey("bab", "shop", ssHost)
+ before, etag, err := rel.Get(ctx, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := provision.WriteRecord(wctx, rel, key, before, before, etag); !errors.Is(err, errSelfServeLeaseLost) {
+ t.Fatalf("a raw record write after the lease ran out = %v, want lost", err)
+ }
+ if n := rel.recordPuts.Load(); n != 1 {
+ t.Fatalf("%d record puts, want only the first", n)
+ }
+}
+
+// ssPassFixture is a claims-off console with the switch off and a bucket
+// whose active.json reads can be failed.
+func ssPassFixture(t *testing.T) (*consoleFixture, *ssFaultStore) {
+ t.Helper()
+ fx := newConsoleFixture(t)
+ rel := &ssFaultStore{Store: fx.rel, key: moves.OfflineActiveKey}
+ fx.srv.cfg.Releases = rel
+ if fx.srv.selfServeMode() {
+ t.Fatal("a fresh console with the switch off is in self-serve mode")
+ }
+ return fx, rel
+}
+
+func TestPassStartsAndRetriesOnReadFailure(t *testing.T) {
+ ctx := context.Background()
+ runs := func(t *testing.T, s *Server) {
+ t.Helper()
+ done := make(chan struct{})
+ go func() { s.StartSelfServeMovesPass(ctx); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("the pass did not stop with nothing left to recover")
+ }
+ }
+
+ t.Run("an unreadable active.json at start runs the pass until the reads succeed", func(t *testing.T) {
+ fx, rel := ssPassFixture(t)
+ rel.failGet.Store(true)
+ var logs strings.Builder
+ restore := captureLogTo(&logs)
+ fx.srv.selfServeScanAtStart(ctx)
+ if !fx.srv.selfServeMode() || fx.srv.selfServeAdmits() {
+ restore()
+ t.Fatalf("mode %v admits %v after a failed start read; want recovery mode without admission", fx.srv.selfServeMode(), fx.srv.selfServeAdmits())
+ }
+ if !fx.srv.selfServeIteration(ctx) {
+ restore()
+ t.Fatal("the pass did not run")
+ }
+ restore()
+ if _, ok := ssLeaseBody(t, rel); !ok {
+ t.Fatal("the pass ran without taking the lease")
+ }
+ if !fx.srv.selfServeMode() {
+ t.Fatal("a tick whose read failed turned recovery mode off")
+ }
+ if !strings.Contains(logs.String(), "injected read failure") {
+ t.Fatalf("the read failure was not logged: %q", logs.String())
+ }
+ rel.failGet.Store(false)
+ fx.srv.selfServeIteration(ctx)
+ if fx.srv.selfServeMode() {
+ t.Fatal("recovery mode stayed on with nothing to recover")
+ }
+ runs(t, fx.srv)
+ })
+
+ t.Run("an unreadable journal at start runs the pass", func(t *testing.T) {
+ fx, _ := ssPassFixture(t)
+ if _, err := fx.st.db.Exec(`ALTER TABLE self_serve_events RENAME TO self_serve_events_away`); err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.selfServeScanAtStart(ctx)
+ if !fx.srv.selfServeMode() {
+ t.Fatal("a failed journal read at start did not put the console in recovery mode")
+ }
+ fx.srv.selfServeIteration(ctx)
+ if !fx.srv.selfServeMode() {
+ t.Fatal("a tick whose journal read failed turned recovery mode off")
+ }
+ if _, err := fx.st.db.Exec(`ALTER TABLE self_serve_events_away RENAME TO self_serve_events`); err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.selfServeIteration(ctx)
+ if fx.srv.selfServeMode() {
+ t.Fatal("recovery mode stayed on after the journal read back empty")
+ }
+ })
+
+ t.Run("each kind of evidence holds recovery mode on", func(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ stage func(t *testing.T, fx *consoleFixture)
+ }{
+ {"an open row", func(t *testing.T, fx *consoleFixture) {
+ ssAdmit(t, fx.st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ }},
+ {"a pending release", func(t *testing.T, fx *consoleFixture) {
+ ssAdmit(t, fx.st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ if err := fx.st.FinishSelfServeMove(ssID, "done", "moved", "", ssT0); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.st.db.Exec(`UPDATE self_serve_moves SET effects_done = 1 WHERE id = ?`, ssID); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.st.db.Exec(`UPDATE self_serve_events SET emitted = 1`); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ {"an undelivered event", func(t *testing.T, fx *consoleFixture) {
+ ssAdmit(t, fx.st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ if err := fx.st.FinishSelfServeMove(ssID, "done", "moved", "", ssT0); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.st.db.Exec(`UPDATE self_serve_moves SET effects_done = 1, release_pending = 0 WHERE id = ?`, ssID); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ {"a reservation", func(t *testing.T, fx *consoleFixture) {
+ body, err := json.Marshal(moves.OfflineActiveBody(map[string]moves.OfflineReservation{
+ "ie-1": mres(ssID, ssHost, time.Minute)}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.rel.Put(context.Background(), moves.OfflineActiveKey, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ fx, _ := ssPassFixture(t)
+ c.stage(t, fx)
+ fx.srv.selfServeScanAtStart(ctx)
+ if !fx.srv.selfServeMode() || fx.srv.selfServeAdmits() {
+ t.Fatalf("mode %v admits %v; want recovery mode without admission", fx.srv.selfServeMode(), fx.srv.selfServeAdmits())
+ }
+ fx.srv.selfServeIteration(ctx)
+ if !fx.srv.selfServeMode() {
+ t.Fatal("a tick turned recovery mode off while its evidence exists")
+ }
+ })
+ }
+ })
+
+ t.Run("the pass drains events and leaves rows it cannot advance alone", func(t *testing.T) {
+ fx, _ := ssPassFixture(t)
+ ssAdmit(t, fx.st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ before := ssLoad(t, fx.st, ssID)
+ fx.srv.selfServeScanAtStart(ctx)
+ prev := selfServeTickInterval
+ selfServeTickInterval = 10 * time.Millisecond
+ defer func() { selfServeTickInterval = prev }()
+ pctx, cancel := context.WithCancel(ctx)
+ done := make(chan struct{})
+ go func() { fx.srv.StartSelfServeMovesPass(pctx); close(done) }()
+ deadline := time.Now().Add(5 * time.Second)
+ for auditCount(t, fx.st) == 0 && time.Now().Before(deadline) {
+ time.Sleep(10 * time.Millisecond)
+ }
+ time.Sleep(50 * time.Millisecond) // a few more ticks
+ cancel()
+ <-done
+ if n := auditCount(t, fx.st); n != 1 {
+ t.Fatalf("%d audit rows, want the admission event once", n)
+ }
+ if after := ssLoad(t, fx.st, ssID); !reflect.DeepEqual(after, before) {
+ t.Fatalf("the pass changed a row it cannot advance:\nbefore %+v\n after %+v", before, after)
+ }
+ if !fx.srv.selfServeMode() {
+ t.Fatal("recovery mode ended with the row open")
+ }
+ })
+
+ t.Run("the pass is a unit of plan 1's admission", func(t *testing.T) {
+ fx, rel := ssPassFixture(t)
+ fx.srv.cfg.SelfServeMoves = "offline"
+ var names []string
+ lifetimesIterationHook = func(name string, _ context.Context, _ *admittedUnit) { names = append(names, name) }
+ defer func() { lifetimesIterationHook = nil }()
+ if !fx.srv.selfServeIteration(ctx) || !reflect.DeepEqual(names, []string{"self-serve moves"}) {
+ t.Fatalf("iterations %v", names)
+ }
+ // Unknown admits no unit: the tick does not run and takes no lease.
+ fx2, rel2 := ssPassFixture(t)
+ fx2.srv.cfg.SelfServeMoves = "offline"
+ fx2.srv.cutover = newCutoverGate(time.Now)
+ if fx2.srv.selfServeIteration(ctx) {
+ t.Fatal("the pass ran while the admission gate is unknown")
+ }
+ if _, ok := ssLeaseBody(t, rel2); ok {
+ t.Fatal("a refused iteration took the lease")
+ }
+ _ = rel
+ })
+
+ t.Run("the switch on runs the pass with an empty journal", func(t *testing.T) {
+ fx, rel := ssPassFixture(t)
+ fx.srv.cfg.SelfServeMoves = "offline"
+ if !fx.srv.selfServeMode() || !fx.srv.selfServeAdmits() {
+ t.Fatal("the switch on is not self-serve mode with admission")
+ }
+ fx.srv.selfServeIteration(ctx)
+ if _, ok := ssLeaseBody(t, rel); !ok {
+ t.Fatal("the pass took no lease")
+ }
+ if !fx.srv.selfServeMode() {
+ t.Fatal("an empty journal turned the switch's mode off")
+ }
+ })
+
+ t.Run("with nothing to recover and the switch off the pass does not run", func(t *testing.T) {
+ fx, rel := ssPassFixture(t)
+ fx.srv.selfServeScanAtStart(ctx)
+ runs(t, fx.srv)
+ if _, ok := ssLeaseBody(t, rel); ok {
+ t.Fatal("a pass with nothing to do took the lease")
+ }
+ })
+}
diff --git a/internal/console/selfserve_store_test.go b/internal/console/selfserve_store_test.go
new file mode 100644
index 000000000..218b67134
--- /dev/null
+++ b/internal/console/selfserve_store_test.go
@@ -0,0 +1,1016 @@
+package console
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "reflect"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The self-serve journal (self-serve moves plan 2 Task 4): the rows, the
+// request lifecycle and its predicates, the per-row event sequence, the
+// transactional delivery and the open-row order the pass steps in.
+
+const (
+ ssHost = "shop.example.com"
+ ssHost2 = "blog.shop.example.com"
+ ssID = "mv_ssssssssssssssssssssssssss"
+)
+
+// ssT0 is a fixed instant with nanoseconds, so a round trip through the
+// Unix-nanosecond columns is exact.
+var ssT0 = time.Date(2026, 10, 22, 9, 30, 0, 123456789, time.UTC)
+
+// ssParticipants pins boxes: every box but the last is a source, the last
+// the target, each with an instance id.
+func ssParticipants(boxes ...string) provision.Participants {
+ var p provision.Participants
+ for i, b := range boxes {
+ x := provision.Participant{BoxID: b, Label: b + "-label", InstanceID: "i-" + b, Verified: true}
+ if i == len(boxes)-1 {
+ p.Target = x
+ } else {
+ p.Source = append(p.Source, x)
+ }
+ }
+ return p
+}
+
+// ssMove is a freshly admitted row for host on boxes, created at.
+func ssMove(id, host string, at time.Time, boxes ...string) SelfServeMove {
+ return SelfServeMove{
+ ID: id, Account: "bab", App: "shop", Host: host, Incarnation: 3,
+ From: "ie", To: "uk", Participants: ssParticipants(boxes...),
+ DisplayName: "Shop's account", RequestedBy: "tok_shop",
+ Phase: "requested", PhaseStartedAt: at,
+ StepDeadline: at.Add(2 * time.Minute), WindowDeadline: at.Add(30 * time.Minute),
+ CreatedAt: at, UpdatedAt: at,
+ }
+}
+
+// ssStore is a console store with the ops account (bab), as NewServer
+// leaves it, so an event's audit row has its account.
+func ssStore(t *testing.T) *Store {
+ t.Helper()
+ st := openTestStore(t)
+ if _, err := st.BackfillOpsMemberships(time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ return st
+}
+
+func ssAdmit(t *testing.T, st *Store, m SelfServeMove) {
+ t.Helper()
+ ok, open, err := st.AdmitSelfServeMove(m)
+ if err != nil || !ok {
+ t.Fatalf("admit %s = %v, open %+v, %v", m.ID, ok, open, err)
+ }
+}
+
+func ssLoad(t *testing.T, st *Store, id string) *SelfServeMove {
+ t.Helper()
+ m, err := st.SelfServeMove(id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if m == nil {
+ t.Fatalf("no row %s", id)
+ }
+ return m
+}
+
+// ssPhase moves a row straight to phase with the given step deadline, for
+// a test that starts mid-journey.
+func ssPhase(t *testing.T, st *Store, id, phase string, step time.Time) {
+ t.Helper()
+ if _, err := st.db.Exec(`UPDATE self_serve_moves SET phase = ?, step_deadline = ? WHERE id = ?`, phase, step.UnixNano(), id); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func ssIssue(t *testing.T, st *Store, id, kind, phase, nonce string, now time.Time) {
+ t.Helper()
+ ok, existing, err := st.IssueSelfServeRequest(id, kind, phase, persistedRequest{Nonce: nonce, ExpectInc: 3, ExpectSeq: 7, Proposed: now.Add(time.Nanosecond)}, now)
+ if err != nil || !ok {
+ t.Fatalf("issue %s %s = %v, existing %+v, %v", kind, nonce, ok, existing, err)
+ }
+}
+
+func ssEvents(t *testing.T, st *Store, id string) []SelfServeEvent {
+ t.Helper()
+ rows, err := st.db.Query(`SELECT id, seq, phase, account, app, host, at, reason, detail, emitted FROM self_serve_events WHERE id = ? ORDER BY seq`, id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer rows.Close()
+ var out []SelfServeEvent
+ for rows.Next() {
+ var e SelfServeEvent
+ var at, emitted int64
+ if err := rows.Scan(&e.ID, &e.Seq, &e.Phase, &e.Account, &e.App, &e.Host, &at, &e.Reason, &e.Detail, &emitted); err != nil {
+ t.Fatal(err)
+ }
+ e.At, e.Emitted = time.Unix(0, at).UTC(), emitted != 0
+ out = append(out, e)
+ }
+ return out
+}
+
+func auditCount(t *testing.T, st *Store) int {
+ t.Helper()
+ var n int
+ if err := st.db.QueryRow(`SELECT COUNT(*) FROM audit WHERE action = 'instance-move'`).Scan(&n); err != nil {
+ t.Fatal(err)
+ }
+ return n
+}
+
+func TestSelfServeStoreRoundTrip(t *testing.T) {
+ st := ssStore(t)
+ if m, err := st.SelfServeMove(ssID); err != nil || m != nil {
+ t.Fatalf("an absent row = %+v, %v; want nil, nil", m, err)
+ }
+ want := ssMove(ssID, ssHost, ssT0, "ie-1", "ie-2", "uk-1")
+ want.Reason = "reserving the boxes"
+ ssAdmit(t, st, want)
+ got := ssLoad(t, st, ssID)
+ want.EventSeq = 1 // the admission's own phase event
+ if !reflect.DeepEqual(*got, want) {
+ t.Fatalf("round trip:\n got %+v\nwant %+v", *got, want)
+ }
+ // Timestamps are Unix nanoseconds.
+ var created, step int64
+ if err := st.db.QueryRow(`SELECT created_at, step_deadline FROM self_serve_moves WHERE id = ?`, ssID).Scan(&created, &step); err != nil {
+ t.Fatal(err)
+ }
+ if created != ssT0.UnixNano() || step != ssT0.Add(2*time.Minute).UnixNano() {
+ t.Fatalf("created_at %d, step_deadline %d: want Unix nanoseconds %d, %d", created, step, ssT0.UnixNano(), ssT0.Add(2*time.Minute).UnixNano())
+ }
+
+ open, terminal, err := st.SelfServeMoveForHost("bab", "shop", ssHost)
+ if err != nil || open == nil || open.ID != ssID || terminal != nil {
+ t.Fatalf("for host = %+v, %+v, %v; want the open row and no terminal", open, terminal, err)
+ }
+
+ // The setters each write their own columns.
+ t1 := ssT0.Add(time.Minute)
+ steps := []func() error{
+ func() error { return st.SetSelfServeEstimate(ssID, 240, `{"copy":60}`) },
+ func() error { return st.RecordSelfServeHealth(ssID, t1, true, "") },
+ func() error { return st.RecordSelfServeHealth(ssID, t1.Add(time.Second), true, "") },
+ func() error {
+ return st.RecordSelfServeHealth(ssID, t1.Add(2*time.Second), false, "502 from the target")
+ },
+ func() error { return st.RecordSelfServeHealth(ssID, t1.Add(3*time.Second), true, "") },
+ func() error { return st.SetSelfServeFinishGate(ssID, "gate1") },
+ func() error { return st.SetSelfServeWarning(ssID, "the health checks did not pass in time") },
+ func() error { return st.SetSelfServeNotifiedSeq(ssID, 41) },
+ func() error { return st.SetSelfServeReopened(ssID, t1.Add(4*time.Second)) },
+ func() error { return st.SetSelfServeAbortFinished(ssID, t1.Add(5*time.Second)) },
+ func() error { return st.SetSelfServeOpenRecordPending(ssID, true) },
+ }
+ for i, f := range steps {
+ if err := f(); err != nil {
+ t.Fatalf("setter %d: %v", i, err)
+ }
+ }
+ got = ssLoad(t, st, ssID)
+ if got.EstimateSeconds != 240 || got.EstimateTerms != `{"copy":60}` {
+ t.Errorf("estimate = %d %q", got.EstimateSeconds, got.EstimateTerms)
+ }
+ if got.HealthAttempts != 4 || got.HealthConsecutiveOK != 1 || !got.HealthFirstOKAt.Equal(t1) ||
+ !got.HealthLastProbeAt.Equal(t1.Add(3*time.Second)) || got.HealthLastError != "" {
+ t.Errorf("health = %d attempts, %d consecutive, first ok %v, last %v, error %q; want 4, 1, the first ok kept",
+ got.HealthAttempts, got.HealthConsecutiveOK, got.HealthFirstOKAt, got.HealthLastProbeAt, got.HealthLastError)
+ }
+ if got.FinishGate != "gate1" || got.Warning != "the health checks did not pass in time" || got.NotifiedSeq != 41 ||
+ !got.SourceReopenedAt.Equal(t1.Add(4*time.Second)) || !got.AbortFinishedAt.Equal(t1.Add(5*time.Second)) || !got.OpenRecordPending {
+ t.Errorf("setters = %+v", got)
+ }
+
+ // Finishing: the outcome, a pending release, effects not done, one
+ // terminal event.
+ if err := st.FinishSelfServeMove(ssID, "done", "moved", "", t1.Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ got = ssLoad(t, st, ssID)
+ if got.Outcome != "done" || got.OutcomeReason != "moved" || !got.ReleasePending || got.EffectsDone ||
+ got.Warning != "the health checks did not pass in time" {
+ t.Fatalf("finished = outcome %q reason %q release %v effects %v warning %q", got.Outcome, got.OutcomeReason, got.ReleasePending, got.EffectsDone, got.Warning)
+ }
+ open, terminal, err = st.SelfServeMoveForHost("bab", "shop", ssHost)
+ if err != nil || open != nil || terminal == nil || terminal.ID != ssID {
+ t.Fatalf("for host after finish = %+v, %+v, %v; want only the terminal row", open, terminal, err)
+ }
+ if err := st.SetSelfServeEffectsDone(ssID); err != nil {
+ t.Fatal(err)
+ }
+ if err := st.ClearSelfServeRelease(ssID); err != nil {
+ t.Fatal(err)
+ }
+ if err := st.SetSelfServeOpenRecordPending(ssID, false); err != nil {
+ t.Fatal(err)
+ }
+ got = ssLoad(t, st, ssID)
+ if !got.EffectsDone || got.ReleasePending || got.OpenRecordPending {
+ t.Fatalf("settled = effects %v release %v open record %v", got.EffectsDone, got.ReleasePending, got.OpenRecordPending)
+ }
+ // A setter on an absent row is an error, not a silent no-op.
+ if err := st.SetSelfServeWarning("mv_absent", "x"); err == nil {
+ t.Error("a setter on an absent row succeeded")
+ }
+}
+
+// The reclaim rule's row evidence (Task 3's selfServeRowEvidence) is the
+// store, set by NewServer.
+func TestSelfServeRowEvidence(t *testing.T) {
+ fx := newConsoleFixture(t)
+ if fx.srv.selfServeRows != selfServeRowEvidence(fx.st) {
+ t.Fatalf("NewServer's selfServeRows = %T, want the store", fx.srv.selfServeRows)
+ }
+ st := fx.st
+ done := func(id string, want bool) {
+ t.Helper()
+ if got, err := st.MoveTerminalOrAbsent(id); err != nil || got != want {
+ t.Errorf("MoveTerminalOrAbsent(%s) = %v, %v; want %v", id, got, err, want)
+ }
+ }
+ issued := func(host string, want bool) {
+ t.Helper()
+ if got, err := st.HostHasIssuedRequest("bab", "shop", host); err != nil || got != want {
+ t.Errorf("HostHasIssuedRequest(%s) = %v, %v; want %v", host, got, err, want)
+ }
+ }
+ done(ssID, true)
+ issued(ssHost, false)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ done(ssID, false)
+ issued(ssHost, false)
+ ssPhase(t, st, ssID, "reserving", ssT0.Add(time.Minute))
+ ssIssue(t, st, ssID, "queue", "reserving", "n-queue", ssT0)
+ issued(ssHost, true)
+ issued(ssHost2, false)
+ if err := st.FinishSelfServeMove(ssID, "aborted", "cancelled", "", ssT0.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ done(ssID, false) // terminal, release pending
+ // Still issued: the queue request was never confirmed or consumed.
+ issued(ssHost, true)
+ if err := st.ClearSelfServeRelease(ssID); err != nil {
+ t.Fatal(err)
+ }
+ done(ssID, true)
+ if _, err := st.db.Exec(`UPDATE self_serve_moves SET request_consumed = 1 WHERE id = ?`, ssID); err != nil {
+ t.Fatal(err)
+ }
+ // A terminal row whose request was consumed issues nothing more.
+ issued(ssHost, false)
+}
+
+func TestAdmitIsAtomic(t *testing.T) {
+ st := ssStore(t)
+ const n = 8
+ var (
+ wg sync.WaitGroup
+ mu sync.Mutex
+ admitted []string
+ opens = map[string]bool{}
+ )
+ for i := 0; i < n; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ id := fmt.Sprintf("mv_%026d", i)
+ ok, open, err := st.AdmitSelfServeMove(ssMove(id, ssHost, ssT0.Add(time.Duration(i)), "ie-1", "uk-1"))
+ mu.Lock()
+ defer mu.Unlock()
+ switch {
+ case err != nil:
+ t.Errorf("admit %s: %v", id, err)
+ case ok:
+ admitted = append(admitted, id)
+ case open == nil:
+ t.Errorf("admit %s refused with no open row", id)
+ default:
+ opens[open.ID] = true
+ }
+ }(i)
+ }
+ wg.Wait()
+ if len(admitted) != 1 {
+ t.Fatalf("admitted %v, want exactly one", admitted)
+ }
+ for id := range opens {
+ if id != admitted[0] {
+ t.Errorf("a refused admission named %s as the open row, want %s", id, admitted[0])
+ }
+ }
+ // Another host is independent; the same id again is the open row.
+ ssAdmit(t, st, ssMove("mv_other", ssHost2, ssT0, "ie-1", "uk-1"))
+ if ok, open, err := st.AdmitSelfServeMove(ssMove(admitted[0], ssHost, ssT0, "ie-1", "uk-1")); err != nil || ok || open == nil || open.ID != admitted[0] {
+ t.Fatalf("re-admit = %v, %+v, %v; want refused with the open row", ok, open, err)
+ }
+ // Once terminal, the host admits again.
+ if err := st.FinishSelfServeMove(admitted[0], "done", "moved", "", ssT0.Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ ssAdmit(t, st, ssMove("mv_next", ssHost, ssT0.Add(2*time.Hour), "ie-1", "uk-1"))
+}
+
+func TestRequestLifecycle(t *testing.T) {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ step := ssT0.Add(10 * time.Minute)
+ ssPhase(t, st, ssID, "staging", step)
+ now := ssT0.Add(time.Second)
+
+ req := func() persistedRequest {
+ t.Helper()
+ pr, err := st.selfServeRequest(ssID)
+ if err != nil || pr == nil {
+ t.Fatalf("request = %+v, %v", pr, err)
+ }
+ return *pr
+ }
+
+ // Issue only from the owning phase.
+ if ok, existing, err := st.IssueSelfServeRequest(ssID, "stage", "ready", persistedRequest{Nonce: "n-wrong"}, now); err != nil || ok || existing != nil {
+ t.Fatalf("issue from another phase = %v, %+v, %v; want refused, nothing outstanding", ok, existing, err)
+ }
+ ssIssue(t, st, ssID, "stage", "staging", "n1", now)
+ pr := req()
+ if pr.Kind != "stage" || pr.Phase != "staging" || pr.Nonce != "n1" || pr.ExpectInc != 3 || pr.ExpectSeq != 7 ||
+ !pr.Proposed.Equal(now.Add(time.Nanosecond)) || !pr.Deadline.Equal(step) || pr.Confirmed || pr.Consumed || pr.Voided || pr.Expired {
+ t.Fatalf("issued = %+v; want stage/staging n1 with the phase's step deadline", pr)
+ }
+
+ // No second issue while unconsumed: the outstanding request comes back,
+ // whatever the caller asked for.
+ ok, existing, err := st.IssueSelfServeRequest(ssID, "precheck", "staging", persistedRequest{Nonce: "n-second"}, now)
+ if err != nil || ok || existing == nil || existing.Nonce != "n1" || existing.Kind != "stage" {
+ t.Fatalf("second issue = %v, %+v, %v; want refused with n1", ok, existing, err)
+ }
+
+ // Confirm only with the nonce, and only once; a second confirm is a
+ // resume the caller reads back.
+ if ok, err := st.ConfirmSelfServeRequest(ssID, "n-other", now); err != nil || ok {
+ t.Fatalf("confirm with another nonce = %v, %v", ok, err)
+ }
+ if ok, err := st.ConfirmSelfServeRequest(ssID, "n1", now); err != nil || !ok {
+ t.Fatalf("confirm = %v, %v", ok, err)
+ }
+ if ok, err := st.ConfirmSelfServeRequest(ssID, "n1", now.Add(time.Second)); err != nil || ok {
+ t.Fatalf("second confirm = %v, %v; want false", ok, err)
+ }
+ if pr := req(); !pr.Confirmed || pr.Consumed {
+ t.Fatalf("after the second confirm = %+v; want confirmed, not consumed (resume)", pr)
+ }
+ m := ssLoad(t, st, ssID)
+ if !m.StageRequest.Equal(now.Add(time.Nanosecond)) || !m.RequestConfirmedAt.Equal(now) {
+ t.Fatalf("stage_request %v, confirmed at %v; want the proposed stamp and the first confirmation", m.StageRequest, m.RequestConfirmedAt)
+ }
+
+ // A confirmed request is never voided.
+ if err := st.VoidSelfServeRequest(ssID, "n1"); err != nil {
+ t.Fatal(err)
+ }
+ if pr := req(); pr.Voided || pr.Consumed {
+ t.Fatalf("void of a confirmed request = %+v", pr)
+ }
+
+ // The standalone consume: the substep and the next wait's deadline in
+ // the same statement, once.
+ next := ssT0.Add(20 * time.Minute)
+ if ok, err := st.ConsumeSelfServeRequest(ssID, "n-other", "round", next); err != nil || ok {
+ t.Fatalf("consume with another nonce = %v, %v", ok, err)
+ }
+ if ok, err := st.ConsumeSelfServeRequest(ssID, "n1", "round", next); err != nil || !ok {
+ t.Fatalf("consume = %v, %v", ok, err)
+ }
+ if ok, err := st.ConsumeSelfServeRequest(ssID, "n1", "round", next.Add(time.Minute)); err != nil || ok {
+ t.Fatalf("second consume = %v, %v", ok, err)
+ }
+ m = ssLoad(t, st, ssID)
+ if !m.Request.Consumed || m.PhaseStep != "round" || !m.StepDeadline.Equal(next) {
+ t.Fatalf("consumed = %v, step %q, deadline %v", m.Request.Consumed, m.PhaseStep, m.StepDeadline)
+ }
+
+ // A page-and-retain kind carries no request deadline.
+ ssIssue(t, st, ssID, "estimate", "staging", "n2", now)
+ if pr := req(); !pr.Deadline.IsZero() {
+ t.Fatalf("estimate request deadline = %v, want none", pr.Deadline)
+ }
+ // A voided request cannot be confirmed, consumed or advanced with, even
+ // with its own nonce.
+ if err := st.VoidSelfServeRequest(ssID, "n2"); err != nil {
+ t.Fatal(err)
+ }
+ if pr := req(); !pr.Voided || !pr.Consumed {
+ t.Fatalf("voided = %+v", pr)
+ }
+ if ok, err := st.ConfirmSelfServeRequest(ssID, "n2", now); err != nil || ok {
+ t.Fatalf("confirm after void = %v, %v", ok, err)
+ }
+ if ok, err := st.ConsumeSelfServeRequest(ssID, "n2", "round", next); err != nil || ok {
+ t.Fatalf("consume after void = %v, %v", ok, err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "staging", "prechecks", "", next, time.Time{}, "n2", now); err != nil || ok {
+ t.Fatalf("advance with a voided nonce = %v, %v", ok, err)
+ }
+ if ok, err := st.AbandonSelfServeRequest(ssID, "n2", next); err != nil || ok {
+ t.Fatalf("abandon after void = %v, %v", ok, err)
+ }
+
+ // Advance: the wrong nonce is refused; an unconfirmed nonce is refused;
+ // the right confirmed nonce consumes in the same statement.
+ ssIssue(t, st, ssID, "stage", "staging", "n3", now)
+ if ok, err := st.AdvanceSelfServe(ssID, "staging", "prechecks", "", next, time.Time{}, "n3", now); err != nil || ok {
+ t.Fatalf("advance with an unconfirmed nonce = %v, %v", ok, err)
+ }
+ if ok, err := st.ConfirmSelfServeRequest(ssID, "n3", now); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "staging", "prechecks", "", next, time.Time{}, "n-wrong", now); err != nil || ok {
+ t.Fatalf("advance with the wrong nonce = %v, %v", ok, err)
+ }
+ if m := ssLoad(t, st, ssID); m.Phase != "staging" || m.Request.Consumed {
+ t.Fatalf("a refused advance changed the row: %+v", m)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "staging", "prechecks", "", next, time.Time{}, "n3", now); err != nil || !ok {
+ t.Fatalf("advance with the confirmed nonce = %v, %v", ok, err)
+ }
+ m = ssLoad(t, st, ssID)
+ if m.Phase != "prechecks" || !m.Request.Consumed || m.PhaseStep != "precheck" || !m.StepDeadline.Equal(next) || !m.PhaseStartedAt.Equal(now) {
+ t.Fatalf("advanced = phase %q consumed %v step %q deadline %v started %v", m.Phase, m.Request.Consumed, m.PhaseStep, m.StepDeadline, m.PhaseStartedAt)
+ }
+
+ // A precheck's confirmation records its stamp; an abandon needs a
+ // confirmed request and gives the next wait its own allowance.
+ ssIssue(t, st, ssID, "precheck", "prechecks", "n4", now)
+ if ok, err := st.AbandonSelfServeRequest(ssID, "n4", next); err != nil || ok {
+ t.Fatalf("abandon of an unconfirmed request = %v, %v", ok, err)
+ }
+ if ok, err := st.ConfirmSelfServeRequest(ssID, "n4", now); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if m := ssLoad(t, st, ssID); !m.PrecheckRequest.Equal(now.Add(time.Nanosecond)) {
+ t.Fatalf("precheck_request = %v", m.PrecheckRequest)
+ }
+ later := next.Add(10 * time.Minute)
+ if ok, err := st.AbandonSelfServeRequest(ssID, "n4", later); err != nil || !ok {
+ t.Fatalf("abandon = %v, %v", ok, err)
+ }
+ m = ssLoad(t, st, ssID)
+ if !m.Request.Consumed || !m.RequestAbandoned || !m.StepDeadline.Equal(later) {
+ t.Fatalf("abandoned = consumed %v abandoned %v deadline %v", m.Request.Consumed, m.RequestAbandoned, m.StepDeadline)
+ }
+ // The next issue starts a fresh request: not abandoned, not confirmed.
+ ssIssue(t, st, ssID, "precheck", "prechecks", "n5", now)
+ if m := ssLoad(t, st, ssID); m.RequestAbandoned || m.Request.Confirmed || !m.RequestConfirmedAt.IsZero() {
+ t.Fatalf("a fresh issue kept the last request's flags: %+v", m)
+ }
+
+ // An ended row issues nothing.
+ if _, err := st.db.Exec(`UPDATE self_serve_moves SET request_consumed = 1 WHERE id = ?`, ssID); err != nil {
+ t.Fatal(err)
+ }
+ if err := st.FinishSelfServeMove(ssID, "aborted", "cancelled", "", now); err != nil {
+ t.Fatal(err)
+ }
+ if ok, existing, err := st.IssueSelfServeRequest(ssID, "abort", "prechecks", persistedRequest{Nonce: "n6"}, now); err != nil || ok || existing != nil {
+ t.Fatalf("issue on an ended row = %v, %+v, %v", ok, existing, err)
+ }
+}
+
+// The request deadline is the issuing phase's step deadline for every kind
+// whose expiry action is abort, and none for the kinds that page and
+// retain.
+func TestSelfServeRequestDeadlineByKind(t *testing.T) {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ step := ssT0.Add(10 * time.Minute)
+ for _, c := range []struct {
+ kind string
+ abort bool
+ }{
+ {"queue", true}, {"stage", true}, {"maintain", true}, {"precheck", true}, {"start", true},
+ {"seal", true}, {"flip", true}, {"open", true},
+ {"gate1", false}, {"gate2", false}, {"finish", false}, {"abort", false}, {"abort-finish", false}, {"estimate", false},
+ } {
+ ssPhase(t, st, ssID, "staging", step)
+ if _, err := st.db.Exec(`UPDATE self_serve_moves SET request_consumed = 1 WHERE id = ?`, ssID); err != nil {
+ t.Fatal(err)
+ }
+ ssIssue(t, st, ssID, c.kind, "staging", "n-"+c.kind, ssT0)
+ pr, err := st.selfServeRequest(ssID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := !pr.Deadline.IsZero(); got != c.abort || (c.abort && !pr.Deadline.Equal(step)) {
+ t.Errorf("%s: request deadline %v, want the step deadline: %v", c.kind, pr.Deadline, c.abort)
+ }
+ }
+}
+
+// The cut and the expiry decisions race on one row; each UPDATE's
+// predicate decides who won.
+func TestSelfServeCutAndExpiryPredicates(t *testing.T) {
+ window := ssT0.Add(30 * time.Minute)
+ setup := func(t *testing.T) *Store {
+ st := ssStore(t)
+ m := ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1")
+ m.Phase, m.WindowDeadline = "ready", window
+ ssAdmit(t, st, m)
+ return st
+ }
+ copyDeadline := ssT0.Add(40 * time.Minute)
+
+ t.Run("a stale window expiry loses to the cut", func(t *testing.T) {
+ st := setup(t)
+ if ok, err := st.MarkSelfServeCut(ssID, ssT0.Add(29*time.Minute), copyDeadline); err != nil || !ok {
+ t.Fatalf("cut = %v, %v", ok, err)
+ }
+ m := ssLoad(t, st, ssID)
+ if !m.WindowDeadline.IsZero() || !m.StepDeadline.Equal(copyDeadline) || m.CutRequestedAt.IsZero() {
+ t.Fatalf("cut row = window %v step %v cut %v", m.WindowDeadline, m.StepDeadline, m.CutRequestedAt)
+ }
+ // A worker whose snapshot predates the cut saw the old window.
+ if ok, err := st.SetSelfServeExpiry(ssID, "ready", "window", window, true); err != nil || ok {
+ t.Fatalf("the obsolete window expiry = %v, %v; want refused", ok, err)
+ }
+ // The copy's step expiry after the cut is accepted.
+ if ok, err := st.SetSelfServeExpiry(ssID, "ready", "step", copyDeadline, true); err != nil || !ok {
+ t.Fatalf("the copy's step expiry = %v, %v", ok, err)
+ }
+ m = ssLoad(t, st, ssID)
+ if !m.ExpiryPending || m.ExpiryPhase != "ready" || m.ExpiryKind != "step" || !m.ExpiryDeadline.Equal(copyDeadline) {
+ t.Fatalf("expiry = %+v", m)
+ }
+ // A second cut is refused.
+ if ok, err := st.MarkSelfServeCut(ssID, ssT0.Add(31*time.Minute), copyDeadline); err != nil || ok {
+ t.Fatalf("a second cut = %v, %v", ok, err)
+ }
+ })
+ t.Run("expiry first: the cut is refused", func(t *testing.T) {
+ st := setup(t)
+ if ok, err := st.SetSelfServeExpiry(ssID, "ready", "window", window, true); err != nil || !ok {
+ t.Fatalf("window expiry = %v, %v", ok, err)
+ }
+ if ok, err := st.MarkSelfServeCut(ssID, ssT0.Add(31*time.Minute), copyDeadline); err != nil || ok {
+ t.Fatalf("cut after a committed expiry = %v, %v; want refused", ok, err)
+ }
+ // An expiry whose deadline moved is refused too.
+ if ok, err := st.SetSelfServeExpiry(ssID, "ready", "step", ssT0, true); err != nil || ok {
+ t.Fatalf("an expiry for a deadline the row no longer has = %v, %v", ok, err)
+ }
+ if _, err := st.SetSelfServeExpiry(ssID, "ready", "later", window, true); err == nil {
+ t.Fatal("an unknown expiry kind was accepted")
+ }
+ })
+ t.Run("a cut adopts the outstanding maintenance round's deadline", func(t *testing.T) {
+ st := setup(t)
+ ssIssue(t, st, ssID, "maintain", "ready", "n-m", ssT0)
+ if ok, err := st.MarkSelfServeCut(ssID, ssT0.Add(time.Minute), copyDeadline); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if pr, _ := st.selfServeRequest(ssID); !pr.Deadline.Equal(copyDeadline) || !pr.CutRequestedAt.Equal(ssT0.Add(time.Minute)) {
+ t.Fatalf("maintenance request = %+v; want the cut's copy deadline and the cut time", pr)
+ }
+ })
+ t.Run("a late receipt retires the expiry and the cut is accepted", func(t *testing.T) {
+ st := setup(t)
+ ssIssue(t, st, ssID, "maintain", "ready", "n-m", ssT0)
+ // Task 7's expiry decision won (request_expired = 1).
+ if _, err := st.db.Exec(`UPDATE self_serve_moves SET request_expired = 1 WHERE id = ?`, ssID); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := st.MarkSelfServeCut(ssID, ssT0.Add(time.Minute), copyDeadline); err != nil || ok {
+ t.Fatalf("cut while the request is expired = %v, %v", ok, err)
+ }
+ if ok, err := st.ConfirmSelfServeRequest(ssID, "n-m", ssT0.Add(time.Minute)); err != nil || !ok {
+ t.Fatalf("the late receipt's confirmation = %v, %v", ok, err)
+ }
+ if ok, err := st.ConsumeSelfServeRequest(ssID, "n-m", "round", copyDeadline); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if ok, err := st.MarkSelfServeCut(ssID, ssT0.Add(2*time.Minute), copyDeadline); err != nil || !ok {
+ t.Fatalf("cut after the receipt retired the expiry = %v, %v", ok, err)
+ }
+ })
+ t.Run("abort", func(t *testing.T) {
+ st := setup(t)
+ if ok, err := st.RequestSelfServeAbort(ssID, []string{"staging"}, ssT0); err != nil || ok {
+ t.Fatalf("abort from another phase = %v, %v", ok, err)
+ }
+ if ok, err := st.RequestSelfServeAbort(ssID, []string{"staging", "ready"}, ssT0); err != nil || !ok {
+ t.Fatalf("abort = %v, %v", ok, err)
+ }
+ if ok, err := st.RequestSelfServeAbort(ssID, []string{"ready"}, ssT0.Add(time.Second)); err != nil || ok {
+ t.Fatalf("a second abort = %v, %v", ok, err)
+ }
+ if m := ssLoad(t, st, ssID); !m.AbortRequestedAt.Equal(ssT0) {
+ t.Fatalf("abort_requested_at = %v, want the first", m.AbortRequestedAt)
+ }
+ if ok, err := st.MarkSelfServeCut(ssID, ssT0.Add(time.Minute), copyDeadline); err != nil || ok {
+ t.Fatalf("cut after an abort = %v, %v", ok, err)
+ }
+ })
+}
+
+// Per-box FIFO: a reserving row waits for an older open row in requested
+// or reserving that shares a box; disjoint rows never wait on each other.
+func TestPerBoxFIFO(t *testing.T) {
+ st := ssStore(t)
+ a := ssMove("mv_a", "a.shop.example.com", ssT0, "X", "Y")
+ b := ssMove("mv_b", "b.shop.example.com", ssT0.Add(time.Second), "X", "Z")
+ c := ssMove("mv_c", "c.shop.example.com", ssT0.Add(2*time.Second), "V", "W")
+ for _, m := range []SelfServeMove{a, b, c} {
+ m.Phase = "reserving"
+ ssAdmit(t, st, m)
+ }
+ waits := func(id string) bool {
+ t.Helper()
+ open, err := st.OpenSelfServeMoves()
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i := range open {
+ if open[i].ID == id {
+ return selfServeWaitsForEarlier(open, &open[i])
+ }
+ }
+ t.Fatalf("%s is not open", id)
+ return false
+ }
+ // A cannot reserve yet (Y is held elsewhere): B shares X and waits.
+ if waits("mv_a") || !waits("mv_b") || waits("mv_c") {
+ t.Fatalf("waits a %v b %v c %v; want only b", waits("mv_a"), waits("mv_b"), waits("mv_c"))
+ }
+ if msgSelfServeWaitingEarlier != "waiting for an earlier move on this box" {
+ t.Fatalf("the reason is %q", msgSelfServeWaitingEarlier)
+ }
+ // A reserved and moved on: B proceeds.
+ if ok, err := st.AdvanceSelfServe("mv_a", "reserving", "queued", "", ssT0.Add(time.Hour), time.Time{}, "", ssT0.Add(time.Minute)); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if waits("mv_b") {
+ t.Fatal("b still waits after a reserved")
+ }
+ // An older row that ended no longer holds anyone back.
+ d := ssMove("mv_d", "d.shop.example.com", ssT0.Add(3*time.Second), "V", "U")
+ d.Phase = "reserving"
+ ssAdmit(t, st, d)
+ if !waits("mv_d") {
+ t.Fatal("d does not wait for c on V")
+ }
+ if err := st.FinishSelfServeMove("mv_c", "aborted", "cancelled", "", ssT0.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if waits("mv_d") {
+ t.Fatal("d waits for an ended row")
+ }
+ // A participant pinned without an instance id is the sentinel: it
+ // shares every box.
+ e := ssMove("mv_e", "e.shop.example.com", ssT0.Add(4*time.Second), "Q", "R")
+ e.Participants.Source[0].InstanceID = ""
+ e.Phase = "requested"
+ ssAdmit(t, st, e)
+ f := ssMove("mv_f", "f.shop.example.com", ssT0.Add(5*time.Second), "S", "T")
+ f.Phase = "reserving"
+ ssAdmit(t, st, f)
+ if !waits("mv_f") {
+ t.Fatal("f does not wait for the older sentinel row")
+ }
+}
+
+func TestEventDeliveryOnce(t *testing.T) {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ snapA, err := st.UnemittedSelfServeEvents(10)
+ if err != nil {
+ t.Fatal(err)
+ }
+ snapB, err := st.UnemittedSelfServeEvents(10)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(snapA) != 1 || len(snapB) != 1 || snapA[0].Seq != 1 || snapA[0].Phase != "requested" {
+ t.Fatalf("snapshots = %+v, %+v; want the admission event in both", snapA, snapB)
+ }
+ var mu sync.Mutex
+ delivered := 0
+ deliver := func(e SelfServeEvent) func(tx *sql.Tx) error {
+ return func(tx *sql.Tx) error {
+ mu.Lock()
+ delivered++
+ mu.Unlock()
+ return auditActivityTx(tx, e.Account, e.App, "self-serve move", Activity{Action: "instance-move", TargetType: "instance", Target: e.Host, Message: e.Phase, Detail: e.Reason})
+ }
+ }
+ var wg sync.WaitGroup
+ for _, snap := range [][]SelfServeEvent{snapA, snapB} {
+ wg.Add(1)
+ go func(e SelfServeEvent) {
+ defer wg.Done()
+ if err := st.DeliverSelfServeEvent(context.Background(), e, deliver(e)); err != nil {
+ t.Error(err)
+ }
+ }(snap[0])
+ }
+ wg.Wait()
+ if delivered != 1 || auditCount(t, st) != 1 {
+ t.Fatalf("delivered %d times, %d audit rows; want once", delivered, auditCount(t, st))
+ }
+ if left, _ := st.UnemittedSelfServeEvents(10); len(left) != 0 {
+ t.Fatalf("undelivered after delivery: %+v", left)
+ }
+
+ // A crash between finalisation steps: the finaliser runs again and
+ // allocates no second terminal event.
+ for i := 0; i < 2; i++ {
+ if err := st.FinishSelfServeMove(ssID, "done", "moved", "", ssT0.Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ terminal := 0
+ for _, e := range ssEvents(t, st, ssID) {
+ if e.Phase == "done" {
+ terminal++
+ }
+ }
+ if terminal != 1 {
+ t.Fatalf("%d terminal events, want one", terminal)
+ }
+}
+
+func TestAdvanceIsPredicatedOnPhaseAndNoAbort(t *testing.T) {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ now := ssT0.Add(time.Minute)
+ if ok, err := st.AdvanceSelfServe(ssID, "staging", "prechecks", "", now, time.Time{}, "", now); err != nil || ok {
+ t.Fatalf("advance from a phase the row is not in = %v, %v", ok, err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "requested", "reserving", "", now, time.Time{}, "", now); err != nil || !ok {
+ t.Fatalf("advance = %v, %v", ok, err)
+ }
+ if ok, err := st.RequestSelfServeAbort(ssID, []string{"reserving"}, now); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "reserving", "queued", "", now, time.Time{}, "", now); err != nil || ok {
+ t.Fatalf("advance past a requested abort = %v, %v", ok, err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "reserving", "reverting", "cancelled", now, time.Time{}, "", now); err != nil || !ok {
+ t.Fatalf("advance into reverting under an abort = %v, %v", ok, err)
+ }
+ if m := ssLoad(t, st, ssID); m.Phase != "reverting" || m.PhaseStep != "abort" || m.Reason != "cancelled" {
+ t.Fatalf("row = phase %q step %q reason %q", m.Phase, m.PhaseStep, m.Reason)
+ }
+}
+
+func TestAdvanceIntoOpenedIgnoresAnObsoleteAbortFlag(t *testing.T) {
+ for _, to := range []string{"opened", "health", "finishing", "done", "reverting"} {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ ssPhase(t, st, ssID, "landed", ssT0.Add(time.Hour))
+ if ok, err := st.RequestSelfServeAbort(ssID, []string{"landed"}, ssT0); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "landed", to, "", ssT0.Add(time.Hour), time.Time{}, "", ssT0.Add(time.Minute)); err != nil || !ok {
+ t.Errorf("landed -> %s under an obsolete abort flag = %v, %v; want advanced", to, ok, err)
+ }
+ }
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ ssPhase(t, st, ssID, "landed", ssT0.Add(time.Hour))
+ if ok, err := st.RequestSelfServeAbort(ssID, []string{"landed"}, ssT0); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "landed", "open-uncertain", "", ssT0.Add(time.Hour), time.Time{}, "", ssT0.Add(time.Minute)); err != nil || ok {
+ t.Fatalf("landed -> open-uncertain under an abort = %v, %v; want refused", ok, err)
+ }
+}
+
+func TestAdvanceClearsAForeignExpiry(t *testing.T) {
+ st := ssStore(t)
+ m := ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1")
+ m.Phase = "ready"
+ ssAdmit(t, st, m)
+ step := m.StepDeadline
+ if ok, err := st.SetSelfServeExpiry(ssID, "ready", "step", step, true); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ // Re-entering the expiry's own phase keeps it.
+ if ok, err := st.AdvanceSelfServe(ssID, "ready", "ready", "", step, m.WindowDeadline, "", ssT0.Add(time.Second)); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if got := ssLoad(t, st, ssID); !got.ExpiryPending || got.ExpiryPhase != "ready" || got.ExpiryKind != "step" || !got.ExpiryDeadline.Equal(step) {
+ t.Fatalf("an expiry of the phase entered was cleared: %+v", got)
+ }
+ // Leaving it clears it: it is foreign to the phase entered.
+ if ok, err := st.AdvanceSelfServe(ssID, "ready", "restage", "", step.Add(time.Minute), time.Time{}, "", ssT0.Add(2*time.Second)); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if got := ssLoad(t, st, ssID); got.ExpiryPending || got.ExpiryPhase != "" || got.ExpiryKind != "" || !got.ExpiryDeadline.IsZero() {
+ t.Fatalf("a foreign expiry survived the advance: %+v", got)
+ }
+}
+
+func TestEventAndAdvanceAreOneTransaction(t *testing.T) {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ now := ssT0.Add(time.Minute)
+ if ok, err := st.AdvanceSelfServe(ssID, "requested", "reserving", "reserving the boxes", now.Add(time.Minute), time.Time{}, "", now); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ evs := ssEvents(t, st, ssID)
+ if len(evs) != 2 || evs[1].Seq != 2 || evs[1].Phase != "reserving" || evs[1].Reason != "reserving the boxes" ||
+ evs[1].Account != "bab" || evs[1].App != "shop" || evs[1].Host != ssHost || !evs[1].At.Equal(now) || evs[1].Emitted {
+ t.Fatalf("events = %+v", evs)
+ }
+ // The next event's slot is taken: the advance must fail whole.
+ if _, err := st.db.Exec(`INSERT INTO self_serve_events (id, seq, phase, account, app, host, at, reason, detail, emitted)
+VALUES (?, 3, 'squatter', 'bab', 'shop', ?, 0, '', '', 1)`, ssID, ssHost); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := st.AdvanceSelfServe(ssID, "reserving", "queued", "", now.Add(time.Hour), time.Time{}, "", now.Add(time.Second)); err == nil || ok {
+ t.Fatalf("advance whose event cannot be inserted = %v, %v; want an error", ok, err)
+ }
+ if m := ssLoad(t, st, ssID); m.Phase != "reserving" || m.EventSeq != 2 {
+ t.Fatalf("a failed advance left phase %q event_seq %d; want reserving 2", m.Phase, m.EventSeq)
+ }
+}
+
+func TestEventDeliveryIsTransactional(t *testing.T) {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ evs, err := st.UnemittedSelfServeEvents(10)
+ if err != nil || len(evs) != 1 {
+ t.Fatal(evs, err)
+ }
+ e := evs[0]
+ boom := errors.New("the audit sink failed")
+ err = st.DeliverSelfServeEvent(context.Background(), e, func(tx *sql.Tx) error {
+ if err := auditActivityTx(tx, e.Account, e.App, "self-serve move", Activity{Action: "instance-move", Target: e.Host, Message: e.Phase}); err != nil {
+ return err
+ }
+ return boom
+ })
+ if !errors.Is(err, boom) {
+ t.Fatalf("delivery = %v, want the sink's error", err)
+ }
+ if n := auditCount(t, st); n != 0 {
+ t.Fatalf("a failed delivery left %d audit rows", n)
+ }
+ if left, _ := st.UnemittedSelfServeEvents(10); len(left) != 1 {
+ t.Fatalf("a failed delivery marked the event emitted: %+v", left)
+ }
+ // A cancelled context delivers nothing either.
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ if err := st.DeliverSelfServeEvent(ctx, e, func(tx *sql.Tx) error { return nil }); err == nil {
+ t.Fatal("a delivery under a cancelled context succeeded")
+ }
+ if left, _ := st.UnemittedSelfServeEvents(10); len(left) != 1 {
+ t.Fatal("a cancelled delivery marked the event emitted")
+ }
+}
+
+// Repeated prechecks are distinct events; each row has its own sequence.
+func TestEventsHaveAPerRowSequence(t *testing.T) {
+ st := ssStore(t)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ ssAdmit(t, st, ssMove("mv_other", ssHost2, ssT0, "ie-1", "uk-1"))
+ for i := 0; i < 3; i++ {
+ tx, err := st.db.Begin()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := InsertSelfServeEventTx(tx, SelfServeEvent{ID: ssID, Phase: "prechecks", At: ssT0.Add(time.Duration(i) * time.Second), Reason: "the source precheck passed"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := tx.Commit(); err != nil {
+ t.Fatal(err)
+ }
+ }
+ evs := ssEvents(t, st, ssID)
+ if len(evs) != 4 {
+ t.Fatalf("events = %+v", evs)
+ }
+ for i, e := range evs {
+ if e.Seq != int64(i+1) {
+ t.Errorf("event %d has seq %d", i, e.Seq)
+ }
+ }
+ if other := ssEvents(t, st, "mv_other"); len(other) != 1 || other[0].Seq != 1 {
+ t.Fatalf("the other row's events = %+v", other)
+ }
+ if m := ssLoad(t, st, ssID); m.EventSeq != 4 {
+ t.Fatalf("event_seq = %d", m.EventSeq)
+ }
+ tx, err := st.db.Begin()
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer tx.Rollback()
+ if err := InsertSelfServeEventTx(tx, SelfServeEvent{ID: "mv_absent", Phase: "prechecks"}); err == nil {
+ t.Fatal("an event for an absent row was inserted")
+ }
+}
+
+func TestOpenRowsOrderedAndIncludePending(t *testing.T) {
+ st := ssStore(t)
+ // Admitted out of order; two share a creation time.
+ rows := []SelfServeMove{
+ ssMove("mv_3", "c.shop.example.com", ssT0.Add(3*time.Second), "ie-1", "uk-1"),
+ ssMove("mv_1b", "b.shop.example.com", ssT0.Add(time.Second), "ie-1", "uk-1"),
+ ssMove("mv_1a", "a.shop.example.com", ssT0.Add(time.Second), "ie-1", "uk-1"),
+ ssMove("mv_0", "z.shop.example.com", ssT0, "ie-1", "uk-1"),
+ ssMove("mv_2", "d.shop.example.com", ssT0.Add(2*time.Second), "ie-1", "uk-1"),
+ ssMove("mv_4", "e.shop.example.com", ssT0.Add(4*time.Second), "ie-1", "uk-1"),
+ ssMove("mv_5", "f.shop.example.com", ssT0.Add(5*time.Second), "ie-1", "uk-1"),
+ }
+ for _, m := range rows {
+ ssAdmit(t, st, m)
+ }
+ end := func(id string) {
+ if err := st.FinishSelfServeMove(id, "done", "moved", "", ssT0.Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // mv_0: terminal, release pending. mv_2: released, effects owed.
+ // mv_4: released, effects done, open record owed. mv_5: settled.
+ end("mv_0")
+ end("mv_2")
+ if err := st.ClearSelfServeRelease("mv_2"); err != nil {
+ t.Fatal(err)
+ }
+ end("mv_4")
+ for _, f := range []func() error{
+ func() error { return st.ClearSelfServeRelease("mv_4") },
+ func() error { return st.SetSelfServeEffectsDone("mv_4") },
+ func() error { return st.SetSelfServeOpenRecordPending("mv_4", true) },
+ } {
+ if err := f(); err != nil {
+ t.Fatal(err)
+ }
+ }
+ end("mv_5")
+ for _, f := range []func() error{
+ func() error { return st.ClearSelfServeRelease("mv_5") },
+ func() error { return st.SetSelfServeEffectsDone("mv_5") },
+ } {
+ if err := f(); err != nil {
+ t.Fatal(err)
+ }
+ }
+ open, err := st.OpenSelfServeMoves()
+ if err != nil {
+ t.Fatal(err)
+ }
+ var ids []string
+ for _, m := range open {
+ ids = append(ids, m.ID)
+ }
+ want := []string{"mv_0", "mv_1a", "mv_1b", "mv_2", "mv_3", "mv_4"}
+ if !reflect.DeepEqual(ids, want) {
+ t.Fatalf("open rows = %v, want %v", ids, want)
+ }
+}
+
+// The gated (claims-on) store opens with every table this task adds, each
+// classified as the plan says.
+func TestClaimStoreOpensWithEveryNewTable(t *testing.T) {
+ s := openGatedStore(t)
+ present, err := claimSQLCensus(s.bootDB(), claimSQLTables)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, name := range []string{"self_serve_moves", "self_serve_events", "self_serve_timings"} {
+ sp, ok := present[name]
+ if !ok {
+ t.Errorf("%s is missing from the claim store", name)
+ continue
+ }
+ if sp.NonApp != "" || sp.Sqid != "account" || sp.App != "app" || sp.Account != "" || sp.AppID != "" {
+ t.Errorf("%s classified as %+v, want {Sqid: account, App: app}", name, sp)
+ }
+ }
+ sp, ok := present["move_timings"]
+ if !ok || sp.NonApp == "" {
+ t.Errorf("move_timings = %+v (present %v), want a NonApp reason", sp, ok)
+ }
+ for _, c := range []struct{ table, cols string }{
+ {"self_serve_timings", "id, mark, account, app, host, at, bytes"},
+ {"move_timings", "id, phase, from_region, to_region, bytes, seconds, at"},
+ } {
+ if _, err := s.bootDB().Exec(`SELECT ` + c.cols + ` FROM ` + c.table); err != nil {
+ t.Errorf("%s: %v", c.table, err)
+ }
+ }
+}