amadan

carlos / platform Public

host: the residence-move worker — prepare rounds, ready promotion, the cut (#331)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PEypt7de2X1G84aQ9rNkCf
Paul Campbell pushed by paul@keymail.dev 13fb387c9c31831e18c484382cf1c78ffc5f9944 parent 296d8ac
5 files changed, +438 −0
  • internal/host/movespass.go +169 −0
  • internal/host/movespass_test.go +218 −0
  • internal/lease/mirror.go +12 −0
  • internal/lease/mirror_test.go +18 −0
  • internal/registry/registry.go +21 −0
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
new file mode 100644
index 000000000..4c2c07786
--- /dev/null
+++ b/internal/host/movespass.go
@@ -0,0 +1,169 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "log"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The residence-move worker (issue #331, instance-move spec): everything a
+// move does to DATA happens here, on the SOURCE box — the box whose
+// CARLOS_REGION matches the record's region for as long as the move is in
+// flight. The console only writes requests; the target box only restores.
+//
+// One pass = one delta round per in-flight move. The caller runs it on the
+// same trigger as the reconcile pass (serial bumps land in seconds via
+// carlos-adopt.path), and prepare's convergence is the round-over-round
+// shrinking of the delta, not a resident goroutine — restart-safe because
+// every step derives from the record and the buckets.
+//
+// The cut ordering is the never-two-writers argument, mechanically:
+//
+// route Status "moving" (edge stops serving; RetireExcept stops the child)
+// → child quiesced (Quiesced(host) — retries next pass until true)
+// → final mirror round (own replica is byte-final; staging matches it)
+// → local purge (the warm-cache trap: a stale local file must
+// never resurrect a moved-away history)
+// → CAS flip+clear (Region=To, Move=nil, ONE write — the last act)
+//
+// Every step is idempotent; a box that dies mid-cut resumes at the top and
+// falls through the already-done steps.
+
+// 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
+}
+
+// moveTarget is one box's coordinates as its edges file publishes them.
+type moveTarget struct {
+ Placement string `json:"placement"`
+ DataBucket string `json:"data_bucket"`
+ DataBucketRegion string `json:"data_bucket_region"`
+ EIP string `json:"eip"`
+ V int `json:"v"`
+ Box string `json:"box"`
+}
+
+// movesPass runs one round over every in-flight move this box owns.
+func movesPass(ctx context.Context, s release.Store, cfg moveCfg) error {
+ recs, err := provision.ListAllRecords(ctx, s)
+ if err != nil {
+ return fmt.Errorf("moves: list records: %w", err)
+ }
+ var targets map[string]moveTarget // lazy: most passes carry no move
+ for _, rec := range recs {
+ if rec.Move == nil || rec.Region != cfg.Region {
+ continue
+ }
+ 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)
+ return nil
+ }
+ }
+ 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)
+ }
+ }
+ return nil
+}
+
+// moveTargets folds the fleet's edges files into placement → coordinates.
+// A duplicate placement code keeps the lexically-first box (deterministic
+// everywhere, same rule as foldHomes) — and a target with no data bucket is
+// simply absent, which runMoveRound reports per-move.
+func moveTargets(ctx context.Context, s release.Store) (map[string]moveTarget, error) {
+ bodies, err := loadEdgesFileBodies(ctx, s)
+ if err != nil {
+ return nil, err
+ }
+ out := map[string]moveTarget{}
+ for _, body := range bodies {
+ var t moveTarget
+ if err := json.Unmarshal(body, &t); err != nil || t.V < 1 || t.Placement == "" || t.DataBucket == "" {
+ continue
+ }
+ if prev, dup := out[t.Placement]; dup && prev.Box <= t.Box {
+ continue
+ }
+ out[t.Placement] = t
+ }
+ return out, nil
+}
+
+func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provision.Record, targets map[string]moveTarget) error {
+ tgt, ok := targets[rec.Move.To]
+ if !ok {
+ return fmt.Errorf("target region %s publishes no data bucket yet — waiting", rec.Move.To)
+ }
+ m, err := cfg.Mirror(ctx, tgt.DataBucketRegion)
+ if err != nil {
+ return fmt.Errorf("mirror client for %s: %w", tgt.DataBucketRegion, err)
+ }
+ srcPrefix := activator.HibernatePrefix(cfg.OwnPrefix, rec.Host)
+ dstPrefix := activator.MovesPrefix(rec.Host)
+
+ switch rec.Move.State {
+ case provision.MovePreparing, provision.MoveReady:
+ changed, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix)
+ if err != nil {
+ return err
+ }
+ if rec.Move.State == provision.MovePreparing && changed == 0 {
+ err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MovePromoteReady(cfg.Now()))
+ if err != nil && !isMoveConflict(err) {
+ return err
+ }
+ log.Printf("moves: %s: staged replica caught up in %s/%s — ready to cut", rec.Host, tgt.DataBucket, dstPrefix)
+ }
+ return nil
+
+ case provision.MoveCutting:
+ // Step 1+2: quiesce. SetStatus is idempotent; a route already
+ // "moving" (a resumed cut) just re-writes the word.
+ if err := cfg.SetStatus(rec.Host, "moving"); err != nil {
+ return fmt.Errorf("set route status: %w", err)
+ }
+ if !cfg.Quiesced(rec.Host) {
+ return nil // the retire is in flight; next pass re-enters here
+ }
+ // Step 3: 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.
+ err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveComplete())
+ if err != nil && !isMoveConflict(err) {
+ return err
+ }
+ log.Printf("moves: %s: cut complete — now homed in %s (route retires on the next reconcile pass)", rec.Host, rec.Move.To)
+ return nil
+ }
+ return nil
+}
+
+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
+ // it. That is completion, not failure.
+ return err != nil && strings.Contains(err.Error(), provision.ErrMoveConflict.Error())
+}
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go
new file mode 100644
index 000000000..353a7568d
--- /dev/null
+++ b/internal/host/movespass_test.go
@@ -0,0 +1,218 @@
+package host
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// fakeMoveMirror: bucket→key→etag plus an op log for ordering assertions.
+type fakeMoveMirror struct {
+ buckets map[string]map[string]string
+ ops []string
+}
+
+func newFakeMoveMirror() *fakeMoveMirror {
+ return &fakeMoveMirror{buckets: map[string]map[string]string{}}
+}
+func (f *fakeMoveMirror) put(bucket, key, etag string) {
+ if f.buckets[bucket] == nil {
+ f.buckets[bucket] = map[string]string{}
+ }
+ f.buckets[bucket][key] = etag
+}
+func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]string, error) {
+ out := map[string]string{}
+ for k, v := range f.buckets[bucket] {
+ if strings.HasPrefix(k, prefix) {
+ out[k] = v
+ }
+ }
+ return out, nil
+}
+func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error {
+ f.ops = append(f.ops, "copy "+dstBucket+"/"+dstKey)
+ f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey])
+ return nil
+}
+func (f *fakeMoveMirror) Delete(_ context.Context, bucket, key string) error {
+ f.ops = append(f.ops, "delete "+bucket+"/"+key)
+ delete(f.buckets[bucket], key)
+ return nil
+}
+
+type moveHarness struct {
+ s release.Store
+ f *fakeMoveMirror
+ cfg moveCfg
+ statuses []string
+ quiet bool
+}
+
+func newMoveHarness(t *testing.T) *moveHarness {
+ t.Helper()
+ h := &moveHarness{s: release.NewDirStore(t.TempDir()), f: newFakeMoveMirror(), quiet: true}
+ h.cfg = moveCfg{
+ Region: "ie",
+ OwnBucket: "data-ie",
+ 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,
+ }
+ // The target's edges file: uk publishes a data bucket.
+ if _, err := h.s.Put(context.Background(), "control/edges/uk-box.json",
+ []byte(`{"v":1,"box":"uk-box","eip":"1.2.3.4","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ return h
+}
+
+func (h *moveHarness) record(t *testing.T, host string, mut ...func(*provision.Record) error) provision.Record {
+ t.Helper()
+ rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab",
+ Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
+ if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
+ t.Fatal(err)
+ }
+ for _, m := range mut {
+ if err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, m); err != nil {
+ t.Fatal(err)
+ }
+ }
+ got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return *got
+}
+
+func (h *moveHarness) read(t *testing.T, host string) provision.Record {
+ t.Helper()
+ got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return *got
+}
+
+func TestMovesPassPrepareAndPromote(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")
+ h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e2")
+
+ // Round 1: copies happen; the SAME pass does not promote (the round
+ // that copied cannot also claim caught-up — a write may have landed
+ // mid-copy; only a round that finds NOTHING to do promotes).
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if got := h.read(t, host); got.Move.State != provision.MovePreparing {
+ t.Fatalf("after the copying round: %s, want still preparing", got.Move.State)
+ }
+ staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
+ if len(staged) != 2 {
+ t.Fatalf("staging = %v", staged)
+ }
+ // Round 2: nothing to do → ready.
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ got := h.read(t, host)
+ if got.Move.State != provision.MoveReady || got.Move.ReadyAt == nil {
+ t.Fatalf("after the quiet round: %+v, want ready", got.Move)
+ }
+ // Ready keeps mirroring: a late write flows on the next round.
+ h.f.put("data-ie", "instances/"+host+"/generations/g/wal/001", "e3")
+ 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) != 3 || h.read(t, host).Move.State != provision.MoveReady {
+ t.Fatalf("late write did not flow: %v", staged)
+ }
+}
+
+func TestMovesPassCutSequence(t *testing.T) {
+ h := newMoveHarness(t)
+ host := "a.go.example.com"
+ h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
+ h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
+ // A local warm cache that must be purged.
+ db := filepath.Join(h.cfg.DataDir, host+".db")
+ if err := writeFileForTest(db, "stale"); err != nil {
+ t.Fatal(err)
+ }
+
+ // Not quiesced yet: the pass sets the status and stops there.
+ h.quiet = false
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if len(h.statuses) != 1 || h.statuses[0] != host+"=moving" {
+ t.Fatalf("statuses = %v", h.statuses)
+ }
+ if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
+ t.Fatalf("cut advanced past an un-quiesced child: %+v", got)
+ }
+ if len(h.f.ops) != 0 {
+ t.Fatalf("mirror ran before quiesce: %v", h.f.ops)
+ }
+
+ // Quiesced: the whole tail runs, in order.
+ h.quiet = true
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ got := h.read(t, host)
+ if got.Region != "uk" || got.Move != nil {
+ t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
+ }
+ if fileExistsForTest(db) {
+ t.Fatal("local DB survived the cut — the warm-cache trap")
+ }
+ staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
+ if len(staged) != 1 {
+ t.Fatalf("staging after final round = %v", staged)
+ }
+ // The pass is now inert for this record (region uk ≠ cfg ie).
+ before := len(h.f.ops)
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if len(h.f.ops) != before {
+ t.Fatal("a completed move kept mirroring")
+ }
+}
+
+func TestMovesPassSkipsWithoutTargetCoords(t *testing.T) {
+ h := newMoveHarness(t)
+ host := "a.go.example.com"
+ h.record(t, host, provision.MoveRequest("se", time.Now())) // se publishes nothing
+ h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
+ if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if got := h.read(t, host); got.Move.State != provision.MovePreparing {
+ t.Fatalf("state advanced with no target coords: %+v", got.Move)
+ }
+ if len(h.f.ops) != 0 {
+ t.Fatalf("mirrored toward an unknown target: %v", h.f.ops)
+ }
+}
+
+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 }
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go
index 15dd29c20..b0cda7f7d 100644
--- a/internal/lease/mirror.go
+++ b/internal/lease/mirror.go
@@ -5,6 +5,7 @@ import (
"fmt"
"net/url"
"sort"
+ "strings"
awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
)
@@ -39,6 +40,17 @@ type MirrorStore interface {
// both sides of a server-side COPY carry the copy's own single-part ETag,
// and litestream's own objects are single PUTs.
func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string) (int, error) {
+ // Both prefixes are DIRECTORY boundaries, normalized here to end in
+ // "/": activator.HibernatePrefix carries none (path.Join), MovesPrefix
+ // carries one, and relative-key math over a slashless prefix would both
+ // mint fused keys (instances/hgenerations/…) and let a listing bleed
+ // into a sibling host whose name merely extends this one's.
+ if !strings.HasSuffix(srcPrefix, "/") {
+ srcPrefix += "/"
+ }
+ if !strings.HasSuffix(dstPrefix, "/") {
+ dstPrefix += "/"
+ }
src, err := m.List(ctx, srcBucket, srcPrefix)
if err != nil {
return 0, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err)
diff --git a/internal/lease/mirror_test.go b/internal/lease/mirror_test.go
index 7ef40543c..3ab8f11b0 100644
--- a/internal/lease/mirror_test.go
+++ b/internal/lease/mirror_test.go
@@ -92,3 +92,21 @@ func TestMirrorPrefix(t *testing.T) {
t.Fatal("injected failure did not surface")
}
}
+
+// Prefix boundaries: a slashless prefix (activator.HibernatePrefix's shape)
+// is normalized to a directory boundary — no fused keys, and a sibling host
+// whose name extends this one's is never swept in.
+func TestMirrorPrefixNormalizesBoundaries(t *testing.T) {
+ ctx := context.Background()
+ f := newFakeMirror()
+ f.put("src", "instances/h/generations/a/snapshot", "e1")
+ f.put("src", "instances/h2/generations/a/snapshot", "eX") // the sibling
+ n, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h")
+ if err != nil || n != 1 {
+ t.Fatalf("round = %d, %v; want exactly the one h object", n, err)
+ }
+ got, _ := f.List(ctx, "dst", "")
+ if _, ok := got["moves/h/generations/a/snapshot"]; !ok || len(got) != 1 {
+ t.Fatalf("staging keys = %v, want the one properly-joined key", got)
+ }
+}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 0c9a5fafa..130626a37 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -832,6 +832,27 @@ func (r *Registry) SetChannel(host, channel string) error {
return nil
}
+// SetStatus writes a route's lifecycle status word. Same contract as
+// SetChannel: a narrow UPDATE that cannot clobber concurrent column writes,
+// and an error when no row matched. The residence-move cut (internal/host's
+// move worker) is its first caller — status "moving" takes a route out of
+// the edge's served set and the activator's managed set in one write, which
+// is what quiesces the child ahead of the final mirror round.
+func (r *Registry) SetStatus(host, status string) error {
+ res, err := r.sql.Exec(`UPDATE routes SET status = ? WHERE host = ?`, status, host)
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n == 0 {
+ return fmt.Errorf("no route for host %q", host)
+ }
+ return nil
+}
+
// SetCaps writes a route's operator-granted capability tokens — the ONLY
// writer of that column, and deliberately the only one: `carlos route
// -grant/-revoke` is the operator's door to a transfer of box authority, and