amadan

carlos / platform Public

review round 1: the data-loss fixes — staging wins, hard-fail fallback, two-phase abort, cut re-checkpoint, ownership + class rails, log throttle (#331)

Six findings from the high-effort review, each with a live failure path:

- Boomerang moves restored a months-stale own lineage: promoteStagedReplica
  now treats a staged copy as the newer truth (it was mirrored from live
  data at cut) and REPLACES the own prefix; abort became two-phase
  (aborting → the source deletes staging → clear) so abandoned staging
  can't linger to poison a later landing.
- A transient S3 error in the staging fallback fell through to a fresh
  boot + seed that permanently shadowed the staged data: the fallback now
  hard-fails the wake, the same rule restoreReplica always had.
- The slashless own-prefix guard could let a sibling host's keys block a
  promotion: gone with the precedence flip.
- Non-hibernating and pooled records passed the gate with Quiesced
  trivially true (purging a live DB / promoting ready over nothing): both
  classes now refused at the door AND skipped by the worker.
- A standby or second same-region box could drive a cut whose child runs
  on its sibling (per-box registries): the worker only acts on routes its
  own registry serves.
- The activator's stop swallows a failed final checkpoint (safe when the
  warm cache stood behind it, fatal once the cut deletes that cache): the
  cut re-checkpoints the local DB with the lease and waits before
  mirroring.

Plus: per-host move-error logging throttled to one line a minute (an
unapplied staging grant would otherwise flood at converge cadence).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PEypt7de2X1G84aQ9rNkCf
Paul Campbell pushed by paul@keymail.dev 38b0b8d89c4615f0744f58ff233179597f654d35 parent 55ff91c
8 files changed, +331 −51
  • internal/activator/backend_exec.go +18 −8
  • internal/activator/backend_exec_test.go +32 −6
  • internal/console/apiinstancemove.go +13 −1
  • internal/console/apiinstancemove_test.go +36 −6
  • internal/host/movespass.go +108 −19
  • internal/host/movespass_test.go +83 −7
  • internal/provision/provision.go +26 −2
  • internal/provision/provision_test.go +15 −2
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 2903a7c35..995870cbf 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -247,7 +247,12 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err
if b.Mirror != nil {
var err error
if promoted, err = b.promoteStagedReplica(ctx, host); err != nil {
- log.Printf("activator: %s: staging fallback: %v (continuing without it)", host, err)
+ // HARD failure, same rule as restoreReplica's own error paths
+ // (review round 1): "continuing without it" here would fall
+ // through to a FRESH boot + seed, and once that empty database
+ // replicates, the staged data is shadowed forever. A failed
+ // wake retries; a seeded ghost does not.
+ return false, fmt.Errorf("staging fallback for %s: %w", host, err)
}
}
@@ -1055,13 +1060,18 @@ func PurgeLocalDBArtifacts(dbPath string, full bool) {
// box's OWN bucket when — and only when — the ordinary prefix is empty. A
// no-op (nil error) when there is nothing staged.
func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (bool, error) {
- own, err := b.Mirror.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
- if err != nil {
- return false, err
- }
- if len(own) > 0 {
- return false, nil // the ordinary lineage exists; staging never wins
- }
+ // STAGING WINS when present (review round 1). A staged replica exists
+ // only while an inbound move is landing or has just landed: it was
+ // mirrored from the then-live data at cut, so it is by construction
+ // newer than any lineage this box kept from a PREVIOUS residence — the
+ // boomerang case (ie→uk→ie) that "own lineage never loses" restored
+ // months-stale data from. An abandoned move's staging can't linger to
+ // poison this: abort is two-phase, and the source deletes staging
+ // before the move clears (MoveAborting).
+ //
+ // The trailing slash on the guard listing matters: hibernatePrefix has
+ // none (path.Join), and a slashless List would let app.example.com's
+ // keys answer for app.example.co.
staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host))
if err != nil {
return false, err
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go
index 3c14ea294..84f291c12 100644
--- a/internal/activator/backend_exec_test.go
+++ b/internal/activator/backend_exec_test.go
@@ -163,22 +163,45 @@ func TestRestoreStagingFallback(t *testing.T) {
t.Fatalf("staging not dropped after a clean restore: %v", staged)
}
- // Precedence: own prefix present → staging never consulted, never
- // dropped.
- f.put(b.Cfg.Bucket, "moves/"+host+"/generations/b/snapshot", "e9")
+ // STAGING WINS (review round 1): a fresh staged copy beside a stale
+ // own lineage is the boomerang landing — the staged bytes replace the
+ // own prefix and are what gets restored.
+ f.put(b.Cfg.Bucket, "moves/"+host+"/generations/b/newer-snapshot", "e9")
_ = os.Remove(filepath.Join(b.DataDir, host+".db"))
restoredFrom = nil
if _, err := b.Restore(ctx, rt); err != nil {
t.Fatal(err)
}
+ own, _ = f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
+ if len(own) != 1 {
+ t.Fatalf("own prefix after boomerang promote = %v, want exactly the staged copy", own)
+ }
+ for k := range own {
+ if !strings.Contains(k, "newer-snapshot") {
+ t.Fatalf("own prefix kept the stale lineage: %v", own)
+ }
+ }
staged, _ = f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
- if len(staged) != 1 {
- t.Fatalf("staging touched despite an existing own lineage: %v", staged)
+ if len(staged) != 0 {
+ t.Fatalf("staging not dropped after the boomerang restore: %v", staged)
}
+
+ // A staging-fallback ERROR fails the wake — never a fresh boot + seed
+ // that would shadow the staged data forever.
+ f.put(b.Cfg.Bucket, "moves/"+host+"/generations/c/snapshot", "eA")
+ f.failList = true
+ _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
+ if _, err := b.Restore(ctx, rt); err == nil {
+ t.Fatal("List failure in the staging fallback did not fail the restore")
+ }
+ f.failList = false
}
// fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→etag.
-type fakeMirrorStore struct{ buckets map[string]map[string]string }
+type fakeMirrorStore struct {
+ buckets map[string]map[string]string
+ failList bool
+}
func newFakeMirrorStore() *fakeMirrorStore {
return &fakeMirrorStore{buckets: map[string]map[string]string{}}
@@ -190,6 +213,9 @@ func (f *fakeMirrorStore) put(bucket, key, etag string) {
f.buckets[bucket][key] = etag
}
func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]string, error) {
+ if f.failList {
+ return nil, context.DeadlineExceeded
+ }
out := map[string]string{}
for k, v := range f.buckets[bucket] {
if strings.HasPrefix(k, prefix) {
diff --git a/internal/console/apiinstancemove.go b/internal/console/apiinstancemove.go
index 2f5e21f17..69cf3f836 100644
--- a/internal/console/apiinstancemove.go
+++ b/internal/console/apiinstancemove.go
@@ -63,6 +63,14 @@ func (s *Server) moveGate(w http.ResponseWriter, r *http.Request) (acct *Account
jsonError(w, http.StatusBadRequest, "fleet-placed instances keep state-does-not-follow semantics — move is for platform regions")
return nil, "", "", nil, false
}
+ if rec.Pool != "" {
+ jsonError(w, http.StatusBadRequest, "pooled instances do not move in v1 — their data lives on the pool's own box")
+ return nil, "", "", nil, false
+ }
+ if !rec.HibernateOrDefault() {
+ jsonError(w, http.StatusBadRequest, "a non-hibernating instance runs outside the activator — nothing manages its data, nothing can move it")
+ return nil, "", "", nil, false
+ }
return acct, app, host, rec, true
}
@@ -135,10 +143,14 @@ func (s *Server) handleAPIInstanceMoveCut(w http.ResponseWriter, r *http.Request
s.moveTransition(w, r, acct, app, host, provision.MoveMarkCutting(), http.StatusAccepted)
}
+// Abort is two-phase (review round 1): this door marks MoveAborting and the
+// SOURCE box deletes the staged copy before clearing the field — a bare
+// clear would leave staging behind to win a later landing on the target
+// (staging-wins). 202, like its siblings: the caller polls the record.
func (s *Server) handleAPIInstanceMoveAbort(w http.ResponseWriter, r *http.Request) {
acct, app, host, _, ok := s.moveGate(w, r)
if !ok {
return
}
- s.moveTransition(w, r, acct, app, host, provision.MoveAbort(), http.StatusOK)
+ s.moveTransition(w, r, acct, app, host, provision.MoveAbort(), http.StatusAccepted)
}
diff --git a/internal/console/apiinstancemove_test.go b/internal/console/apiinstancemove_test.go
index 463f6e4e0..1aa6258ee 100644
--- a/internal/console/apiinstancemove_test.go
+++ b/internal/console/apiinstancemove_test.go
@@ -20,6 +20,7 @@ func TestAPIInstanceMoveVerbs(t *testing.T) {
fx, token := instanceFixture(t, "code-inst-move")
m := titoManifest()
m.Regions = nil
+ m.Pool = "" // pooled records are refused at the move door — separate rail, pinned below
enableInstances(t, fx, "titogo", m)
u := instancesURL(fx, "bab", "titogo")
put := func(key, body string) {
@@ -72,16 +73,27 @@ func TestAPIInstanceMoveVerbs(t *testing.T) {
if code, _, out := rawReq(t, http.MethodPost, moveURL+"/cut", token, ""); code != http.StatusConflict {
t.Fatalf("cut from preparing = %d %s, want 409", code, out)
}
- // Abort from preparing → 200, move cleared; GET shows no move.
- if code, _, out := rawReq(t, http.MethodDelete, moveURL, token, ""); code != http.StatusOK {
- t.Fatalf("abort = %d %s, want 200", code, out)
+ // Abort from preparing → 202 aborting (two-phase: the SOURCE box
+ // deletes staging, then clears the field).
+ if code, _, out := rawReq(t, http.MethodDelete, moveURL, token, ""); code != http.StatusAccepted || !strings.Contains(out, provision.MoveAborting) {
+ t.Fatalf("abort = %d %s, want 202 aborting", code, out)
+ }
+ // A second abort while aborting → 409; a new request while aborting →
+ // 409 (still in flight).
+ if code, _, _ := rawReq(t, http.MethodDelete, moveURL, token, ""); code != http.StatusConflict {
+ t.Fatalf("double abort = %d, want 409", code)
+ }
+ if code, _, _ := rawReq(t, http.MethodPost, moveURL, token, `{"region":"uk"}`); code != http.StatusConflict {
+ t.Fatalf("request while aborting = %d, want 409", code)
+ }
+ // The box's half clears it; then a fresh request proceeds.
+ if err := provision.SetRecordMove(context.Background(), fx.rel, "bab", "titogo", "acme.go.tito.io", provision.MoveAbortComplete()); err != nil {
+ t.Fatal(err)
}
code, _, out = rawReq(t, http.MethodGet, u+"/acme.go.tito.io", token, "")
if code != http.StatusOK || strings.Contains(out, `"move"`) {
- t.Fatalf("after abort: %d %s, want no move key", code, out)
+ t.Fatalf("after abort-complete: %d %s, want no move key", code, out)
}
- // Promote to ready out-of-band (the box's act), then cut succeeds and
- // abort refuses.
if code, _, _ := rawReq(t, http.MethodPost, moveURL, token, `{"region":"uk"}`); code != http.StatusAccepted {
t.Fatal("re-request")
}
@@ -109,6 +121,7 @@ func TestAPIInstanceMoveClassRails(t *testing.T) {
fx, token := instanceFixture(t, "code-inst-move-rails")
m := titoManifest()
m.Regions = nil
+ m.Pool = ""
enableInstances(t, fx, "titogo", m)
u := instancesURL(fx, "bab", "titogo")
if _, err := fx.rel.Put(context.Background(), "control/edges/uk-box.json",
@@ -151,4 +164,21 @@ func TestAPIInstanceMoveClassRails(t *testing.T) {
t.Fatalf("move fleet-placed = %d %s, want 400 mentioning fleet", code, out)
}
_ = fmt.Sprint()
+ _ = provisionPooledMoveRefusal(t, fx, token, u)
+}
+
+// provisionPooledMoveRefusal pins the pool rail: a pooled record's data
+// lives on the pool's box, and v1 refuses the class at the door.
+func provisionPooledMoveRefusal(t *testing.T, fx *consoleFixture, token, u string) bool {
+ t.Helper()
+ rec := provision.Record{Schema: 1, Host: "pooled.go.tito.io", Slug: "pooled", App: "titogo",
+ Account: "bab", Pool: "dedicated-x", Channel: "stable", CreatedAt: time.Now().UTC()}
+ if err := provision.CreateRecord(context.Background(), fx.rel, rec); err != nil {
+ t.Fatal(err)
+ }
+ code, _, out := rawReq(t, http.MethodPost, u+"/pooled.go.tito.io/move", token, `{"region":"uk"}`)
+ if code != http.StatusBadRequest || !strings.Contains(out, "pool") {
+ t.Fatalf("move pooled = %d %s, want 400 mentioning the pool rail", code, out)
+ }
+ return true
}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 37df170a4..1772374c5 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -8,6 +8,7 @@ import (
"os"
"path/filepath"
"strings"
+ "sync"
"sync/atomic"
"time"
@@ -54,16 +55,30 @@ var armedMoveCfg atomic.Pointer[moveCfg]
// svc.Start.
func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service) {
cfg := moveCfg{
- Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")),
- OwnBucket: hib.Bucket,
- OwnPrefix: hib.Prefix,
- DataDir: dataDir,
+ Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")),
+ OwnBucket: hib.Bucket,
+ OwnBucketRegion: hib.Region,
+ OwnPrefix: hib.Prefix,
+ DataDir: dataDir,
Mirror: func(ctx context.Context, dstRegion string) (lease.MirrorStore, error) {
return lease.NewS3Mirror(ctx, hib.Region, dstRegion)
},
Quiesced: svc.Quiesced,
SetStatus: reg.SetStatus,
- Now: time.Now,
+ OwnsRoute: func(host string) bool {
+ rt, err := reg.Get(host)
+ return err == nil && rt != nil && rt.Provisioned
+ },
+ Checkpoint: func(ctx context.Context, dbPath string) error {
+ host := strings.TrimSuffix(filepath.Base(dbPath), ".db")
+ rep, err := lease.ReplicateWithLease(ctx, dbPath, hib.Bucket, activator.HibernatePrefix(hib.Prefix, host), hib.Region)
+ if err != nil {
+ return err
+ }
+ defer rep.Close(ctx)
+ return rep.SyncAndWait(ctx)
+ },
+ Now: time.Now,
}
armedMoveCfg.Store(&cfg)
}
@@ -82,14 +97,27 @@ func runMovesPass(ctx context.Context, s release.Store) {
// moveCfg carries what the worker needs, injectable for tests.
type moveCfg struct {
- Region string // this box's placement code, "" included (the home)
- OwnBucket string // CARLOS_HIBERNATE_BUCKET
- OwnPrefix string // CARLOS_HIBERNATE_PREFIX ("instances" in production)
- DataDir string // /data — where the local DB lives
- Mirror func(ctx context.Context, dstRegion string) (lease.MirrorStore, error)
- Quiesced func(host string) bool
- SetStatus func(host, status string) error
- Now func() time.Time
+ Region string // this box's placement code, "" included (the home)
+ OwnBucket string // CARLOS_HIBERNATE_BUCKET
+ OwnBucketRegion string // CARLOS_HIBERNATE_REGION
+ OwnPrefix string // CARLOS_HIBERNATE_PREFIX ("instances" in production)
+ DataDir string // /data — where the local DB lives
+ Mirror func(ctx context.Context, dstRegion string) (lease.MirrorStore, error)
+ Quiesced func(host string) bool
+ SetStatus func(host, status string) error
+ // OwnsRoute answers whether THIS box's registry serves the host
+ // (review round 1): region equality alone would let a standby or
+ // second same-region box drive a cut whose child runs on its sibling —
+ // each box has its OWN registry, so the sibling's Quiesced is
+ // trivially, wrongly true. The box that owns the route owns the move.
+ OwnsRoute func(host string) bool
+ // Checkpoint re-replicates the LOCAL DB to the own bucket and waits
+ // (review round 1): the activator's stop logs — and swallows — a failed
+ // final checkpoint, because pre-move the local warm cache made that
+ // safe. The cut deletes the warm cache, so it must first prove the own
+ // replica really carries the tail. Nil skips (tests without S3).
+ Checkpoint func(ctx context.Context, dbPath string) error
+ Now func() time.Time
}
// moveTarget is one box's coordinates as its edges file publishes them.
@@ -113,6 +141,18 @@ func movesPass(ctx context.Context, s release.Store, cfg moveCfg) error {
if rec.Move == nil || rec.Region != cfg.Region {
continue
}
+ // The rails the console's moveGate also holds (and this side must
+ // re-hold — records have other writers): only an activator-managed
+ // exec instance's state is this worker's to ship. A pooled record's
+ // data lives on the pool's box, not this one (v1 refuses the whole
+ // class), and a non-hibernating instance runs OUTSIDE the activator
+ // — Quiesced would answer true over a live, writing process.
+ if rec.Pool != "" || !rec.HibernateOrDefault() || (rec.Kind != "" && rec.Kind != "instance") {
+ continue
+ }
+ if cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host) {
+ continue // another box's child; its owner drives the move
+ }
if targets == nil {
if targets, err = moveTargets(ctx, s); err != nil {
log.Printf("moves: %v — every in-flight move waits for the next pass", err)
@@ -120,8 +160,11 @@ func movesPass(ctx context.Context, s release.Store, cfg moveCfg) error {
}
}
if err := runMoveRound(ctx, s, cfg, rec, targets); err != nil {
- // One move's trouble must not starve its neighbours.
- log.Printf("moves: %s: %v", rec.Host, err)
+ // One move's trouble must not starve its neighbours — and a
+ // PERSISTENT one (an unapplied staging grant's AccessDenied,
+ // review round 1) must not flood the journal at the fast
+ // converge loop's cadence: once a minute per host is plenty.
+ logMoveTrouble(rec.Host, err)
}
}
return nil
@@ -177,6 +220,26 @@ func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provisi
}
return nil
+ case provision.MoveAborting:
+ // Abort's box half: delete the staged copy, THEN clear the move —
+ // a lingering staging would win a later landing on the target
+ // (staging-wins, activator.promoteStagedReplica).
+ staged, err := m.List(ctx, tgt.DataBucket, dstPrefix)
+ if err != nil {
+ return err
+ }
+ for k := range staged {
+ if err := m.Delete(ctx, tgt.DataBucket, k); err != nil {
+ return err
+ }
+ }
+ err = provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveAbortComplete())
+ if err != nil && !isMoveConflict(err) {
+ return err
+ }
+ log.Printf("moves: %s: abort complete — staged copy deleted, instance stays in %q", rec.Host, rec.Region)
+ return nil
+
case provision.MoveCutting:
// Step 1+2: quiesce. SetStatus is idempotent; a route already
// "moving" (a resumed cut) just re-writes the word.
@@ -186,13 +249,27 @@ func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provisi
if !cfg.Quiesced(rec.Host) {
return nil // the retire is in flight; next pass re-enters here
}
- // Step 3: the byte-final mirror round.
+ // Step 3: prove the own replica carries the tail. The activator's
+ // stop SWALLOWS a failed final checkpoint (safe pre-move: the warm
+ // cache stood behind it; fatal here: step 5 deletes that cache).
+ // Re-checkpointing an already-synced DB is cheap and idempotent; a
+ // held lease (the child not fully released) or an S3 blip returns
+ // an error and the pass retries.
+ dbPath := filepath.Join(cfg.DataDir, rec.Host+".db")
+ if cfg.Checkpoint != nil {
+ if _, statErr := os.Stat(dbPath); statErr == nil {
+ if err := cfg.Checkpoint(ctx, dbPath); err != nil {
+ return fmt.Errorf("final checkpoint: %w", err)
+ }
+ }
+ }
+ // Step 4: the byte-final mirror round.
if _, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix); err != nil {
return err
}
- // Step 4: the warm-cache purge.
- activator.PurgeLocalDBArtifacts(filepath.Join(cfg.DataDir, rec.Host+".db"), true)
- // Step 5: the last act — Region flips, Move clears, one write.
+ // Step 5: the warm-cache purge.
+ activator.PurgeLocalDBArtifacts(dbPath, true)
+ // Step 6: the last act — Region flips, Move clears, one write.
err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveComplete())
if err != nil && !isMoveConflict(err) {
return err
@@ -203,6 +280,18 @@ func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provisi
return nil
}
+// logMoveTrouble rate-limits per-host move errors to one line a minute.
+var moveTroubleLast sync.Map // host → time.Time
+
+func logMoveTrouble(host string, err error) {
+ now := time.Now()
+ if last, ok := moveTroubleLast.Load(host); ok && now.Sub(last.(time.Time)) < time.Minute {
+ return
+ }
+ moveTroubleLast.Store(host, now)
+ log.Printf("moves: %s: %v", host, err)
+}
+
func isMoveConflict(err error) bool {
// A lost CAS whose re-read finds the transition already made surfaces
// as ErrMoveConflict: someone (this box, last life) already advanced
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go
index 353a7568d..3772d3c9f 100644
--- a/internal/host/movespass_test.go
+++ b/internal/host/movespass_test.go
@@ -54,22 +54,29 @@ type moveHarness struct {
cfg moveCfg
statuses []string
quiet bool
+ owns bool
}
func newMoveHarness(t *testing.T) *moveHarness {
t.Helper()
- h := &moveHarness{s: release.NewDirStore(t.TempDir()), f: newFakeMoveMirror(), quiet: true}
+ h := &moveHarness{s: release.NewDirStore(t.TempDir()), f: newFakeMoveMirror(), quiet: true, owns: true}
h.cfg = moveCfg{
- Region: "ie",
- OwnBucket: "data-ie",
- OwnPrefix: "instances",
- DataDir: t.TempDir(),
+ Region: "ie",
+ OwnBucket: "data-ie",
+ OwnBucketRegion: "eu-west-1",
+ OwnPrefix: "instances",
+ DataDir: t.TempDir(),
Mirror: func(_ context.Context, dstRegion string) (lease.MirrorStore, error) {
return h.f, nil
},
Quiesced: func(string) bool { return h.quiet },
SetStatus: func(host, status string) error { h.statuses = append(h.statuses, host+"="+status); return nil },
- Now: time.Now,
+ OwnsRoute: func(host string) bool { return h.owns },
+ Checkpoint: func(_ context.Context, dbPath string) error {
+ h.f.ops = append(h.f.ops, "checkpoint "+filepath.Base(dbPath))
+ return nil
+ },
+ Now: time.Now,
}
// The target's edges file: uk publishes a data bucket.
if _, err := h.s.Put(context.Background(), "control/edges/uk-box.json",
@@ -172,11 +179,25 @@ func TestMovesPassCutSequence(t *testing.T) {
t.Fatalf("mirror ran before quiesce: %v", h.f.ops)
}
- // Quiesced: the whole tail runs, in order.
+ // Quiesced: the whole tail runs, in order — the final own-bucket
+ // checkpoint strictly BEFORE the mirror round (the mirror of a stale
+ // replica is faithful and useless).
h.quiet = true
if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
t.Fatal(err)
}
+ ckIdx, cpIdx := -1, -1
+ for i, op := range h.f.ops {
+ if strings.HasPrefix(op, "checkpoint ") && ckIdx == -1 {
+ ckIdx = i
+ }
+ if strings.HasPrefix(op, "copy ") && cpIdx == -1 {
+ cpIdx = i
+ }
+ }
+ if ckIdx == -1 || cpIdx == -1 || ckIdx > cpIdx {
+ t.Fatalf("cut op order = %v, want checkpoint before the mirror copy", h.f.ops)
+ }
got := h.read(t, host)
if got.Region != "uk" || got.Move != nil {
t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
@@ -216,3 +237,58 @@ func TestMovesPassSkipsWithoutTargetCoords(t *testing.T) {
func writeFileForTest(path, content string) error { return os.WriteFile(path, []byte(content), 0o600) }
func fileExistsForTest(path string) bool { _, err := os.Stat(path); return err == nil }
+
+// The rails and the ownership gate (review round 1): pooled, non-hibernate,
+// non-instance, and not-my-route records are skipped untouched.
+func TestMovesPassRailsAndOwnership(t *testing.T) {
+ h := newMoveHarness(t)
+ host := "a.go.example.com"
+ h.record(t, host, provision.MoveRequest("uk", time.Now()))
+ h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
+
+ // Not this box's route (a standby sharing the region): untouched.
+ h.owns = false
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if len(h.f.ops) != 0 || h.read(t, host).Move.State != provision.MovePreparing {
+ t.Fatalf("a non-owning box acted: ops=%v", h.f.ops)
+ }
+ h.owns = true
+
+ // A pooled sibling with a move field: skipped (the platform box holds
+ // none of its data — a round here would promote ready over nothing).
+ pooled := "p.go.example.com"
+ rec := provision.Record{Schema: 1, Host: pooled, Slug: "p", App: "app", Account: "bab",
+ Region: "ie", Pool: "dedicated-x", Channel: "stable", CreatedAt: time.Now().UTC()}
+ if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
+ t.Fatal(err)
+ }
+ if err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", pooled, provision.MoveRequest("uk", time.Now())); err != nil {
+ t.Fatal(err)
+ }
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if h.read(t, pooled).Move.State != provision.MovePreparing {
+ t.Fatal("a pooled record's move advanced on the platform box")
+ }
+}
+
+// Abort's box half: staging deleted, then the field cleared.
+func TestMovesPassAbortPhase(t *testing.T) {
+ h := newMoveHarness(t)
+ host := "a.go.example.com"
+ h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MoveAbort())
+ h.f.put("data-uk", "moves/"+host+"/generations/g/snapshot", "e1")
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
+ if len(staged) != 0 {
+ t.Fatalf("staging survived the abort: %v", staged)
+ }
+ if got := h.read(t, host); got.Move != nil || got.Region != "ie" {
+ t.Fatalf("after abort phase: %+v region=%q", got.Move, got.Region)
+ }
+}
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index dc4b1c287..05acf3438 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -695,6 +695,13 @@ const (
MovePreparing = "preparing"
MoveReady = "ready"
MoveCutting = "cutting"
+ // MoveAborting is abort's in-flight half (review round 1): the console
+ // marks it, the SOURCE box deletes the staged copy from the target
+ // bucket and then clears Move — because a staged replica left behind by
+ // a bare clear would win a LATER landing on that box (staging-wins,
+ // backend_exec.go's promoteStagedReplica) with data from an abandoned
+ // move.
+ MoveAborting = "aborting"
)
// ErrMoveConflict is a transition refused because the record's current move
@@ -784,14 +791,31 @@ func MoveMarkCutting() func(*Record) error {
}
}
-// MoveAbort clears a move that has not passed the point of no return.
+// MoveAbort marks a move for abort — legal before the point of no return.
+// The clear itself is the source box's (MoveAbortComplete), after it has
+// deleted the staged copy.
func MoveAbort() func(*Record) error {
return func(rec *Record) error {
if rec.Move == nil {
return fmt.Errorf("%w: no move in flight", ErrMoveConflict)
}
- if rec.Move.State == MoveCutting {
+ switch rec.Move.State {
+ case MoveCutting:
return fmt.Errorf("%w: past the point of no return — the cut completes forward", ErrMoveConflict)
+ case MoveAborting:
+ return fmt.Errorf("%w: already aborting", ErrMoveConflict)
+ }
+ rec.Move.State = MoveAborting
+ return nil
+ }
+}
+
+// MoveAbortComplete is the source box's half of an abort: staging is gone,
+// the move field clears.
+func MoveAbortComplete() func(*Record) error {
+ return func(rec *Record) error {
+ if rec.Move == nil || rec.Move.State != MoveAborting {
+ return fmt.Errorf("%w: abort-complete is only legal from aborting", ErrMoveConflict)
}
rec.Move = nil
return nil
diff --git a/internal/provision/provision_test.go b/internal/provision/provision_test.go
index 69845d5f4..a152520b3 100644
--- a/internal/provision/provision_test.go
+++ b/internal/provision/provision_test.go
@@ -947,7 +947,8 @@ func TestSetRecordMoveTransitions(t *testing.T) {
if got := read(); got.Region != "uk" || got.Move != nil {
t.Fatalf("after complete: region=%q move=%+v, want uk/nil in one write", got.Region, got.Move)
}
- // Abort round-trip from ready on a fresh move.
+ // Abort is two-phase: ready → aborting (cut and promote both refuse
+ // there) → cleared only by MoveAbortComplete.
if err := mv(MoveRequest("ie", now)); err != nil {
t.Fatal(err)
}
@@ -957,8 +958,20 @@ func TestSetRecordMoveTransitions(t *testing.T) {
if err := mv(MoveAbort()); err != nil {
t.Fatal(err)
}
+ if got := read(); got.Move == nil || got.Move.State != MoveAborting {
+ t.Fatalf("after abort: %+v, want aborting", got.Move)
+ }
+ if err := mv(MoveMarkCutting()); !errors.Is(err, ErrMoveConflict) {
+ t.Fatalf("cut from aborting = %v, want ErrMoveConflict", err)
+ }
+ if err := mv(MoveAbort()); !errors.Is(err, ErrMoveConflict) {
+ t.Fatalf("double abort = %v, want ErrMoveConflict", err)
+ }
+ if err := mv(MoveAbortComplete()); err != nil {
+ t.Fatal(err)
+ }
if got := read(); got.Move != nil || got.Region != "uk" {
- t.Fatalf("after abort: %+v region=%q", got.Move, got.Region)
+ t.Fatalf("after abort-complete: %+v region=%q", got.Move, got.Region)
}
}