carlos / platform Public

moves: ready when a round's delta is small, not zero (#175)

runMoveRound promoted preparing -> ready only on a round that changed
nothing. An awake instance never produces one: litestream lands new L0
files between any two listings and compaction rewrites level files, so
kpmg.bdn.oncarlos.com sat in preparing for 80 minutes re-staging ~400 MB
a round until it was aborted. An idle instance moved fine.

Zero was never what correctness needed. The cut quiesces the route,
fences wakes, takes a final checkpoint and runs a byte-exact final
mirror round before it flips the record. The readiness bar only bounds
how much that final round, and the downtime around it, has left to copy.

A preparing round now promotes when it copied at most ReadyDelta bytes:
16 MiB by default, or CARLOS_MOVE_READY_DELTA_BYTES (a whole number of
bytes; 0 restores the old rule). An unparsable or negative value keeps
the default and is logged, the way the CARLOS_ALERT_*_PCT overrides
are. The ready log line now carries the bytes copied, the change count
and the bar.

The cut, abort and landing paths are unchanged. A new test reaches ready
with a non-zero delta, keeps writing, compacting and pruning after
ready, and checks the cut still lands staging byte-exact (mutation
check: dropping the cut's final round fails it).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 3 weeks ago fe3d31b8f4e8195523b2c7d03b2fb7da2c9cc4a1 parent 2f5469f

2 files changed, +267 −11

internal/host/movespass.go +55 −5
internal/host/movespass_test.go +212 −6

CI passed — run details

diff --git a/internal/host/movespass.go b/internal/host/movespass.go

index 6f9951983..e43a2d48b 100644

--- a/internal/host/movespass.go

+++ b/internal/host/movespass.go

@@ -9,6 +9,7 @@ import (

"log"

"os"

"path/filepath"

+ "strconv"

"strings"

"sync"

"sync/atomic"

@@ -30,7 +31,11 @@ import (

// 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.

+// every step derives from the record and the buckets. "Converged" is a

+// round that copied at most moveCfg.ReadyDelta bytes (16 MiB by default),

+// not a round that copied nothing: an awake instance never produces the

+// latter (#175). Ready therefore means "small gap", and the cut's final

+// round below is what makes staging exact.

// The cut ordering is the never-two-writers argument, mechanically:

@@ -89,7 +94,10 @@ func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.Mirro

// the service that answers whether a child is down. Called once, before

// svc.Start.

func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service) {

+ readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv)

+ if !ok {

+ log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta)

+ }

cfg := moveCfg{

Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")),

OwnBucket: hib.Bucket,

@@ -128,7 +136,8 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,

defer rep.Close(ctx)

return rep.SyncAndWait(ctx)

- Now: time.Now,

+ Now: time.Now,

+ ReadyDelta: readyDelta,

}

armedMovesWorker.Store(&movesWorker{cfg: cfg, mirrors: map[string]lease.MirrorStore{}})

}

@@ -184,6 +193,41 @@ type moveCfg struct {

// replica really carries the tail. Nil skips (tests without S3).

Checkpoint func(ctx context.Context, dbPath string) error

Now func() time.Time

+ // ReadyDelta is the readiness bar (#175): a preparing round that copied

+ // at most this many bytes promotes the move to ready. Zero means only a

+ // round that copied nothing at all qualifies, which an awake instance

+ // never produces. armMovesPass sets it from moveReadyDeltaFromEnv.

+ ReadyDelta int64

+// defaultMoveReadyDelta is the readiness bar when

+// CARLOS_MOVE_READY_DELTA_BYTES is unset or rejected. A busy instance's

+// per-round churn of new L0 files sits well under 16 MiB, so it reaches

+// ready; a round in which compaction rewrote a large level file (138 MB

+// for the 293 MB database in #175) does not, and waits for a quieter one.

+// Ready keeps mirroring every pass until the cut, so the bar roughly

+// bounds what the cut's final round, run while the instance is down, has

+// left to copy.

+const defaultMoveReadyDelta int64 = 16 << 20

+// moveReadyDeltaEnv overrides defaultMoveReadyDelta with a whole number of

+// bytes. Zero is accepted and restores the old "nothing changed" rule.

+const moveReadyDeltaEnv = "CARLOS_MOVE_READY_DELTA_BYTES"

+// moveReadyDeltaFromEnv reads the readiness bar. An unset, unparsable or

+// negative value keeps the default, so an operator's typo never stops the

+// agent; ok is false only when a value was SET and rejected, so the caller

+// can say so instead of leaving the operator to believe it took.

+func moveReadyDeltaFromEnv(getenv func(string) string) (delta int64, ok bool) {

+ v := strings.TrimSpace(getenv(moveReadyDeltaEnv))

+ if v == "" {

+ return defaultMoveReadyDelta, true

+ }

+ n, err := strconv.ParseInt(v, 10, 64)

+ if err != nil || n < 0 {

+ return defaultMoveReadyDelta, false

+ }

+ return n, true

}

// movesPass runs one round over every in-flight move this box owns.

@@ -326,12 +370,18 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov

if err != nil {

return err

}

- if rec.Move.State == provision.MovePreparing && round.Changed == 0 {

+ // Ready means the gap is small, not that it is closed (#175): an

+ // awake instance writes new litestream files between any two

+ // listings, so a round that copied NOTHING never comes. The cut

+ // closes the gap itself (quiesce, fence, final checkpoint, a final

+ // round), so this bar only bounds how much that last round, and

+ // the downtime around it, has left to copy.

+ if rec.Move.State == provision.MovePreparing && round.BytesCopied <= cfg.ReadyDelta {

if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MovePromoteReady(cfg.Now())); err != nil && !errors.Is(err, provision.ErrMoveConflict) {

return err

}

nudgeSerial(ctx, s)

- log.Printf("moves: %s: staged replica caught up in %s/%s — ready to cut", rec.Host, tgt.DataBucket, dstPrefix)

+ log.Printf("moves: %s: staged replica caught up in %s/%s (last round copied %d bytes in %d changes, bar %d) — ready to cut", rec.Host, tgt.DataBucket, dstPrefix, round.BytesCopied, round.Changed, cfg.ReadyDelta)

}

return nil

diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go

index d57ee7b1e..7ee3701db 100644

--- a/internal/host/movespass_test.go

+++ b/internal/host/movespass_test.go

@@ -3,12 +3,14 @@ package host

import (

"context"

"errors"

+ "fmt"

"os"

"path/filepath"

"strings"

"testing"

"time"

+ "github.com/carlosframework/platform/internal/activator"

"github.com/carlosframework/platform/internal/lease"

"github.com/carlosframework/platform/internal/provision"

"github.com/carlosframework/platform/internal/registry"

@@ -80,7 +82,8 @@ func newMoveHarness(t *testing.T) *moveHarness {

h.f.ops = append(h.f.ops, "checkpoint "+filepath.Base(dbPath))

return nil

- Now: time.Now,

+ Now: time.Now,

+ ReadyDelta: defaultMoveReadyDelta, // what armMovesPass sets with the env unset

}

// The target's edges file: uk publishes a data bucket.

if _, err := h.s.Put(context.Background(), "control/edges/uk-box.json",

@@ -129,12 +132,12 @@ 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.putSized("data-ie", "instances/"+host+"/generations/g/snapshot", "e1", 64<<20)

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).

+ // Round 1: copies happen; the SAME pass does not promote, because the

+ // round copied more than the readiness bar (the 64 MiB snapshot): the

+ // cut's own round could have that much again to do.

if err := movesPass(context.Background(), h.s, h.worker()); err != nil {

t.Fatal(err)

}

@@ -145,7 +148,7 @@ func TestMovesPassPrepareAndPromote(t *testing.T) {

if len(staged) != 2 {

t.Fatalf("staging = %v", staged)

}

- // Round 2: nothing to do → ready.

+ // Round 2: nothing to do → ready (zero bytes is under any bar).

if err := movesPass(context.Background(), h.s, h.worker()); err != nil {

t.Fatal(err)

}

@@ -478,6 +481,209 @@ func TestMovesPassCutStopsWhenTheFenceFails(t *testing.T) {

}

+// moveRound runs one pass and returns the record's move state after it.

+func (h *moveHarness) moveRound(t *testing.T, host string) string {

+ t.Helper()

+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {

+ t.Fatal(err)

+ }

+ got := h.read(t, host)

+ if got.Move == nil {

+ t.Fatalf("move cleared unexpectedly: %+v", got)

+ }

+ return got.Move.State

+// An AWAKE instance never stops writing: litestream lands a new L0 file

+// between every two listings, so no round ever copies nothing. Production

+// (#175, kpmg.bdn.oncarlos.com) sat in preparing for 80 minutes that way.

+// The bar is bytes, not zero: once the initial bulk has moved, a round

+// that copied only the small new files is ready.

+func TestMovesPassBusySourceReachesReady(t *testing.T) {

+ h := newMoveHarness(t)

+ host := "a.go.example.com"

+ src := "instances/" + host + "/generations/g/"

+ h.record(t, host, provision.MoveRequest("uk", time.Now()))

+ h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20)

+ h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-0", 200)

+ // Round 1 moves the bulk: well over the bar, still preparing.

+ if st := h.moveRound(t, host); st != provision.MovePreparing {

+ t.Fatalf("after the bulk round: %s, want preparing", st)

+ }

+ // Every later round finds a fresh L0 file and a lease renewal: never a

+ // zero-change round, always a small one.

+ st := ""

+ for i := 1; i <= 3 && st != provision.MoveReady; i++ {

+ h.f.putSized("data-ie", src+fmt.Sprint("ltx/0/", i), fmt.Sprint("l0-", i), 8<<10)

+ h.f.putSized("data-ie", "instances/"+host+"/lock.json", fmt.Sprint("renewal-", i), 200)

+ st = h.moveRound(t, host)

+ }

+ if st != provision.MoveReady {

+ t.Fatalf("a busy source with small deltas never reached ready (state %s)", st)

+ }

+// A round that copied more than the bar does not promote, however many

+// rounds it takes: compaction rewriting a 138 MiB level file every round

+// (the #175 shape) keeps the move preparing. The bar is inclusive.

+func TestMovesPassLargeDeltaDoesNotPromote(t *testing.T) {

+ h := newMoveHarness(t)

+ host := "a.go.example.com"

+ src := "instances/" + host + "/generations/g/"

+ h.record(t, host, provision.MoveRequest("uk", time.Now()))

+ h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20)

+ for i := 0; i < 4; i++ {

+ // Compaction rewrites the level file under a new ETag each round.

+ h.f.putSized("data-ie", src+"ltx/1/compacted", fmt.Sprint("c", i), 138<<20)

+ if st := h.moveRound(t, host); st != provision.MovePreparing {

+ t.Fatalf("round %d copied 138 MiB and promoted to %s", i, st)

+ }

+ // The boundary, with a small bar: one byte over stays preparing, the

+ // bar exactly promotes.

+ h.cfg.ReadyDelta = 1000

+ h.f.putSized("data-ie", src+"ltx/0/a", "a", 1001)

+ if st := h.moveRound(t, host); st != provision.MovePreparing {

+ t.Fatalf("a 1001-byte round under a 1000-byte bar promoted to %s", st)

+ }

+ h.f.putSized("data-ie", src+"ltx/0/b", "b", 1000)

+ if st := h.moveRound(t, host); st != provision.MoveReady {

+ t.Fatalf("a 1000-byte round under a 1000-byte bar left the move %s", st)

+ }

+// A zero bar is the old rule: only a round that copied nothing promotes.

+func TestMovesPassZeroBarNeedsAQuietRound(t *testing.T) {

+ h := newMoveHarness(t)

+ h.cfg.ReadyDelta = 0

+ host := "a.go.example.com"

+ h.record(t, host, provision.MoveRequest("uk", time.Now()))

+ h.f.putSized("data-ie", "instances/"+host+"/generations/g/wal/000", "e1", 1)

+ if st := h.moveRound(t, host); st != provision.MovePreparing {

+ t.Fatalf("a 1-byte round under a zero bar promoted to %s", st)

+ }

+ if st := h.moveRound(t, host); st != provision.MoveReady {

+ t.Fatalf("a quiet round under a zero bar left the move %s", st)

+ }

+// Ready no longer means "staging matched the source". It means the gap was

+// small. The cut's final round still has to make staging exact: what was

+// written after the last ready round, what compaction rewrote, and what

+// retention pruned all reach the target, and the lease never does.

+func TestMovesPassCutAfterBusyReadyLandsByteExact(t *testing.T) {

+ h := newMoveHarness(t)

+ host := "a.go.example.com"

+ own := "instances/" + host + "/"

+ gen := own + "generations/g/"

+ h.record(t, host, provision.MoveRequest("uk", time.Now()))

+ h.f.putSized("data-ie", gen+"snapshots/0", "s0", 293<<20)

+ h.f.putSized("data-ie", gen+"ltx/1/compacted", "c0", 138<<20)

+ h.f.putSized("data-ie", gen+"ltx/0/0", "l0-0", 4<<10)

+ h.f.putSized("data-ie", own+"lock.json", "renewal-0", 200)

+ if st := h.moveRound(t, host); st != provision.MovePreparing {

+ t.Fatalf("after the bulk round: %s", st)

+ }

+ h.f.putSized("data-ie", gen+"ltx/0/1", "l0-1", 4<<10)

+ if st := h.moveRound(t, host); st != provision.MoveReady {

+ t.Fatalf("a 4 KiB round left the move %s, want ready", st)

+ }

+ staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")

+ if _, ok := staged["moves/"+host+"/generations/g/ltx/0/1"]; !ok {

+ t.Fatalf("the ready round's own delta was not staged: %v", staged)

+ }

+ // After ready and before the cut, the instance keeps writing: a new L0

+ // file, a compaction rewrite, retention pruning an old L0 file, and a

+ // lease renewal.

+ h.f.putSized("data-ie", gen+"ltx/0/2", "l0-2", 4<<10)

+ h.f.putSized("data-ie", gen+"ltx/1/compacted", "c1", 139<<20)

+ delete(h.f.buckets["data-ie"], gen+"ltx/0/0")

+ h.f.putSized("data-ie", own+"lock.json", "renewal-1", 200)

+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveMarkCutting()); err != nil {

+ t.Fatal(err)

+ }

+ h.quiet = true

+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {

+ t.Fatal(err)

+ }

+ if got := h.read(t, host); got.Region != "uk" || got.Move != nil {

+ t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)

+ }

+ src, _ := h.f.List(context.Background(), "data-ie", own)

+ staged, _ = h.f.List(context.Background(), "data-uk", "moves/"+host+"/")

+ want := map[string]lease.MirrorObject{}

+ for k, v := range src {

+ if strings.HasSuffix(k, "/lock.json") {

+ continue

+ }

+ want["moves/"+host+"/"+strings.TrimPrefix(k, own)] = v

+ }

+ if len(staged) != len(want) {

+ t.Fatalf("staged %d objects, want %d:\nstaged %v\nwant %v", len(staged), len(want), staged, want)

+ }

+ for k, v := range want {

+ if staged[k] != v {

+ t.Fatalf("staged[%s] = %+v, want %+v (the cut's final round must make staging exact)", k, staged[k], v)

+ }

+func TestMoveReadyDeltaFromEnv(t *testing.T) {

+ for _, tc := range []struct {

+ raw string

+ want int64

+ ok bool

+ }{

+ {"", defaultMoveReadyDelta, true}, // unset: the default, nothing to report

+ {"0", 0, true}, // the old zero-change rule, on purpose

+ {"1048576", 1 << 20, true},

+ {" 67108864 ", 64 << 20, true},

+ {"-1", defaultMoveReadyDelta, false},

+ {"16MiB", defaultMoveReadyDelta, false},

+ {"1.5", defaultMoveReadyDelta, false},

+ {"99999999999999999999", defaultMoveReadyDelta, false},

+ } {

+ got, ok := moveReadyDeltaFromEnv(func(k string) string {

+ if k != "CARLOS_MOVE_READY_DELTA_BYTES" {

+ t.Fatalf("read %s, want CARLOS_MOVE_READY_DELTA_BYTES", k)

+ }

+ return tc.raw

+ })

+ if got != tc.want || ok != tc.ok {

+ t.Errorf("%q: got %d, %v; want %d, %v", tc.raw, got, ok, tc.want, tc.ok)

+ }

+ if defaultMoveReadyDelta != 16<<20 {

+ t.Fatalf("default bar = %d, want 16 MiB", defaultMoveReadyDelta)

+ }

+// armMovesPass is where the bar reaches the worker: the env override when

+// it parses, the default when it does not.

+func TestArmMovesPassReadsTheReadyDelta(t *testing.T) {

+ prev := armedMovesWorker.Load()

+ t.Cleanup(func() { armedMovesWorker.Store(prev) })

+ for _, tc := range []struct {

+ raw string

+ want int64

+ }{

+ {"", defaultMoveReadyDelta},

+ {"4096", 4096},

+ {"-5", defaultMoveReadyDelta},

+ {"lots", defaultMoveReadyDelta},

+ } {

+ t.Setenv("CARLOS_MOVE_READY_DELTA_BYTES", tc.raw)

+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil)

+ if got := armedMovesWorker.Load().cfg.ReadyDelta; got != tc.want {

+ t.Errorf("CARLOS_MOVE_READY_DELTA_BYTES=%q armed a bar of %d, want %d", tc.raw, got, tc.want)

+ }

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 }