carlos / platform Public

moves worker: an unplaced seed waits before any copy (Plan 8 Task 4)

A record that slipped past an old console with an unplaced --from seed
or an uncommitted import is skipped while its move is preparing: before
the legacy mirror, the Phase 1 source round (lease, claim touch,
staging, report) and the target's report. The source and target log the
seed's sentence once a minute; a bystander box stays quiet.

M8(b): the S1 probe fails with s1NoAnswer for a row with an import gate,
so the moves probe never wakes a gated import, preparing or ready
(ServeScheduled does not read the gate).

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

claude 2 weeks ago 67794a55bfb5faee664b8cdabda8dc31791a05f4 parent 367ad94

4 files changed, +301 −1

internal/host/movespass.go +20 −1
internal/host/movespass_test.go +182 −0
internal/host/movesprobe_test.go +96 −0
internal/host/movesreport.go +3 −0

CI passed — run details

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

index 12f4ba48b..83060b44e 100644

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

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

@@ -353,9 +353,14 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin

release.NewProofGate(release.ProofPolicyPath, dataDir), stageRequester(dataDir))

if svc != nil {

// S1 asks the instance itself, woken through the activator.

+ // An uncommitted import's row (a gate set, or closed by its

+ // rollback) is never woken for S1: only its own verify requests

+ // reach it (seeded-instances spec §5.4), and it moves only once

+ // committed (§5.8). ServeScheduled does not read the gate, so

+ // this does, for every round that probes, preparing or ready.

cfg.ReadyProbe = func(ctx context.Context, host, path string) moves.Check {

rt, err := reg.Get(host)

- if err != nil || rt == nil {

+ if err != nil || rt == nil || rt.ImportGate != "" {

return moves.Check{Reason: s1NoAnswer(host)}

}

return probeInstance(ctx, svc.ServeScheduled, *rt, path)

@@ -621,6 +626,20 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {

}

for _, rec := range recs {

+ // A seed moves only once placed, an import only once committed

+ // (seeded-instances spec §5.8): a record that slipped past an old

+ // console is refused here, before the first copy, in every round

+ // (legacy, Phase 1 source, participant). targetReportPass holds

+ // the same rule for the target. Only the move's source and target

+ // say so: every box lists every record.

+ if rec.Move != nil && rec.Move.State == provision.MovePreparing {

+ if b := rec.Seed.MoveBlock(); b != "" {

+ if rec.Region == cfg.Region || rec.Move.To == cfg.Region {

+ w.notice(rec.Host+"#seed", "moves: %s: %s", rec.Host, provision.SeedMoveSentence(b))

+ }

+ continue

+ }

if rec.Move != nil && !rec.Move.Legacy() {

// A Phase 1 record. With the switch off it is never touched

// (none exists unless the switch was on somewhere).

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

index 7a4161b49..2d61b1d6a 100644

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

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

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

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

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

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

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

)

// fakeMoveMirror keeps S3's ETag rules (lease's fakeMirror explains them):

@@ -1036,3 +1037,184 @@ func TestMovesPassAbortClearsReceipts(t *testing.T) {

t.Fatal("the receipts survived the abort")

}

+func TestMovesPassSkipsAnUnplacedSeed(t *testing.T) {

+ logs := captureLog(t)

+ h := newMoveHarness(t)

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

+ id, _ := seeds.NewID()

+ rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable",

+ CreatedAt: time.Now().UTC(),

+ Move: &provision.MoveState{To: "uk", State: provision.MovePreparing, RequestedAt: time.Now().UTC()},

+ Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "t", At: time.Now().UTC()}}

+ if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {

+ t.Fatal(err)

+ }

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

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

+ t.Fatal(err)

+ }

+ if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 0 {

+ t.Fatalf("the worker mirrored an unplaced seed: %v", staged)

+ }

+ if got := h.read(t, host); got.Move.State != provision.MovePreparing {

+ t.Fatalf("state = %s", got.Move.State)

+ }

+ if len(h.f.ops) != 0 {

+ t.Fatalf("a skipped round wrote: %v", h.f.ops)

+ }

+ if want := "moves: " + host + ": " + provision.SeedMoveSentence(provision.SeedMoveUnplaced); !strings.Contains(logs(), want) {

+ t.Fatalf("log %q lacks %q", logs(), want)

+ }

+ // Placed, the same record moves on the next pass.

+ if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", host, func(r *provision.Record) error {

+ r.Seed.State = seeds.StatePlaced

+ return nil

+ }); err != nil {

+ t.Fatal(err)

+ }

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

+ t.Fatal(err)

+ }

+ if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 1 {

+ t.Fatalf("a placed seed did not mirror: %v", staged)

+ }

+// The skip is for preparing only: a legacy move an old console already

+// took past preparing has had its effects, and the worker carries it on.

+func TestMovesPassCarriesOnAReadyMoveWithAnUnplacedSeed(t *testing.T) {

+ h := newMoveHarness(t)

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

+ id, _ := seeds.NewID()

+ now := time.Now().UTC()

+ rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable",

+ CreatedAt: now,

+ Move: &provision.MoveState{To: "uk", State: provision.MoveReady, RequestedAt: now, ReadyAt: &now},

+ Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "t", At: now}}

+ if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {

+ t.Fatal(err)

+ }

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

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

+ t.Fatal(err)

+ }

+ if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 1 {

+ t.Fatalf("a ready move stopped mirroring: %v", staged)

+ }

+// seedTheMove gives the Phase 1 harness's record a seed after its move

+// started: a record that slipped past an old console, whose doors did not

+// refuse it (spec §5.8). The seed writer is the only writer of Seed.

+func seedTheMove(t *testing.T, h *phase1Harness, kind, state string) {

+ t.Helper()

+ id, err := seeds.NewID()

+ if err != nil {

+ t.Fatal(err)

+ }

+ if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", p1Host, func(r *provision.Record) error {

+ r.Seed = &provision.SeedSource{ID: id, Kind: kind, State: state, By: "t", At: h.clock.Now()}

+ return nil

+ }); err != nil {

+ t.Fatal(err)

+ }

+func setSeedState(t *testing.T, h *phase1Harness, state string) {

+ t.Helper()

+ if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", p1Host, func(r *provision.Record) error {

+ r.Seed.State = state

+ return nil

+ }); err != nil {

+ t.Fatal(err)

+ }

+// The Phase 1 source round refuses an uncommitted import before its first

+// effect: no staging lease, no claim touch, no staging write, no source

+// report. Committed, the same move prepares.

+func TestPhase1SourceSkipsAnUncommittedImport(t *testing.T) {

+ logs := captureLog(t)

+ h := newPhase1Harness(t)

+ h.start(t, "manual")

+ seedTheMove(t, h, seeds.KindImport, seeds.StateSwapped)

+ before := h.claim(t)

+ h.pass(t)

+ if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {

+ t.Fatalf("state = %s, want preparing", got)

+ }

+ if h.lease.acquired != 0 {

+ t.Fatalf("the skipped round took the staging lease %d times", h.lease.acquired)

+ }

+ if calls := h.f.stagingCalls(); len(calls) != 0 {

+ t.Fatalf("the skipped round staged: %+v", calls)

+ }

+ if keys := stagedKeys(h, moves.StagingPrefix(p1Host, p1ID)); len(keys) != 0 {

+ t.Fatalf("the lane holds %v", keys)

+ }

+ if sr := h.sourceReport(t); sr != nil {

+ t.Fatalf("the skipped round published a source report: %+v", sr)

+ }

+ if after := h.claim(t); fmt.Sprint(after.Touched) != fmt.Sprint(before.Touched) {

+ t.Fatalf("the skipped round touched the claim: %+v, was %+v", after.Touched, before.Touched)

+ }

+ if len(h.probePaths) != 0 {

+ t.Fatalf("the skipped round probed the instance: %v", h.probePaths)

+ }

+ if want := "moves: " + p1Host + ": " + provision.SeedMoveSentence(provision.SeedMoveUncommitted); !strings.Contains(logs(), want) {

+ t.Fatalf("log %q lacks %q", logs(), want)

+ }

+ setSeedState(t, h, seeds.StateCommitted)

+ h.pass(t)

+ if keys := stagedKeys(h, moves.StagingPrefix(p1Host, p1ID)); !containsString(keys, "move.json") {

+ t.Fatalf("a committed import did not prepare: %v", keys)

+ }

+ if h.sourceReport(t) == nil {

+ t.Fatal("a committed import published no source report")

+ }

+// The target box writes no report for a move whose seed is unplaced.

+// Placed, it does.

+func TestPhase1TargetSkipsAnUnplacedSeed(t *testing.T) {

+ logs := captureLog(t)

+ h := newPhase1Harness(t)

+ h.start(t, "manual")

+ seedTheMove(t, h, seeds.KindReplica, seeds.StateReady)

+ h.targetPass(t, h.targetCfg())

+ if tr := h.targetReport(t); tr != nil {

+ t.Fatalf("the target reported on an unplaced seed: %+v", tr)

+ }

+ if calls := h.f.stagingCalls(); len(calls) != 0 {

+ t.Fatalf("the target's skipped round staged: %+v", calls)

+ }

+ if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {

+ t.Fatalf("state = %s, want preparing", got)

+ }

+ if want := "moves: " + p1Host + ": " + provision.SeedMoveSentence(provision.SeedMoveUnplaced); !strings.Contains(logs(), want) {

+ t.Fatalf("log %q lacks %q", logs(), want)

+ }

+ setSeedState(t, h, seeds.StatePlaced)

+ h.targetPass(t, h.targetCfg())

+ if h.targetReport(t) == nil {

+ t.Fatal("the target wrote no report for a placed seed")

+ }

+// A box that is neither the move's source nor its target says nothing

+// about the seed: the log line belongs to the boxes the move waits on.

+func TestMovesPassSeedNoticeOnlyOnParticipants(t *testing.T) {

+ logs := captureLog(t)

+ h := newPhase1Harness(t)

+ h.start(t, "manual")

+ seedTheMove(t, h, seeds.KindReplica, seeds.StateReady)

+ c := h.targetCfg()

+ c.Region, c.Box = "us", "us-box"

+ h.targetPass(t, c)

+ if strings.Contains(logs(), provision.SeedMoveSentence(provision.SeedMoveUnplaced)) {

+ t.Fatalf("a bystander logged the seed: %q", logs())

+ }

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

index 0f7f88d97..b9c3c6202 100644

--- a/internal/host/movesprobe_test.go

+++ b/internal/host/movesprobe_test.go

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

"net/http"

"os"

"path/filepath"

+ "strings"

+ "sync"

"testing"

"time"

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

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

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

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

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

)

@@ -144,3 +150,93 @@ func TestS1ProbeSendsTheDeclaredPathVerbatim(t *testing.T) {

}

+// armedProbe is the production S1 (armPhase1's ReadyProbe) over a real

+// activator and one sleeping instance whose row carries gate as its import

+// gate. The socket directory is short, as a unix socket path must be.

+func armedProbe(t *testing.T, gate string) (func(context.Context, string, string) moves.Check, *countingWakeBackend) {

+ t.Helper()

+ dir, err := os.MkdirTemp("", "s1")

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { os.RemoveAll(dir) })

+ reg, err := registry.Open(filepath.Join(dir, "r.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { reg.Close() })

+ inst := &fakeInstance{socket: filepath.Join(dir, "i.sock"), handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {

+ w.WriteHeader(http.StatusOK)

+ })}

+ if err := reg.Upsert(registry.Route{Host: p1Host, Kind: "instance", Account: "bab", App: "app", Status: "ready",

+ Hibernate: true, Socket: inst.socket, DBPath: filepath.Join(dir, "app.db"), ImportGate: gate}); err != nil {

+ t.Fatal(err)

+ }

+ b := &countingWakeBackend{fakeWakeBackend: &fakeWakeBackend{inst: inst}}

+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)

+ svc.Start()

+ t.Cleanup(func() {

+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)

+ defer cancel()

+ svc.Shutdown(ctx)

+ })

+ var cfg moveCfg

+ hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"}

+ armPhase1(&cfg, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, dir, reg, svc, nil, hostwrite.Config{}, func(string) string { return "" })

+ if cfg.ReadyProbe == nil {

+ t.Fatal("armPhase1 wired no S1 probe")

+ }

+ return cfg.ReadyProbe, b

+// countingWakeBackend counts every wake the activator begins (its lease

+// acquire, the wake's first step), whether or not the spawn then succeeds.

+type countingWakeBackend struct {

+ *fakeWakeBackend

+ mu sync.Mutex

+ acquires int

+func (b *countingWakeBackend) Acquire(ctx context.Context, rt registry.Route) (activator.Lease, error) {

+ b.mu.Lock()

+ b.acquires++

+ b.mu.Unlock()

+ return b.fakeWakeBackend.Acquire(ctx, rt)

+func (b *countingWakeBackend) wakes() int {

+ b.mu.Lock()

+ defer b.mu.Unlock()

+ return b.acquires

+// The moves probe never wakes a gated import (seeded-instances spec §5.4:

+// only the app's own verify requests reach it; §5.8: it moves only once

+// committed). ServeScheduled does not read the gate, so the probe must:

+// whatever the move's state, S1 fails without a wake.

+func TestS1ProbeNeverWakesAGatedImport(t *testing.T) {

+ for _, gate := range []string{strings.Repeat("a", 64), registry.ImportGateClosed} {

+ probe, b := armedProbe(t, gate)

+ got := probe(context.Background(), p1Host, "/")

+ if got.OK || got.Reason != s1NoAnswer(p1Host) {

+ t.Fatalf("gate %q: S1 = %+v, want %q", gate, got, s1NoAnswer(p1Host))

+ }

+ if n := b.wakes(); n != 0 {

+ t.Fatalf("gate %q: the probe began %d wakes of a gated import", gate, n)

+ }

+ if reqs := b.inst.served(); len(reqs) != 0 {

+ t.Fatalf("gate %q: the gated instance served %d requests", gate, len(reqs))

+ }

+ // The control: the same row with no gate is woken and answers (this

+ // half listens on a unix socket, so it needs a run outside a sandbox

+ // that forbids one).

+ probe, b := armedProbe(t, "")

+ if got := probe(context.Background(), p1Host, "/"); !got.OK {

+ t.Fatalf("an ungated instance: S1 = %+v", got)

+ }

+ if b.wakes() != 1 || b.inst.startCount() != 1 {

+ t.Fatalf("an ungated instance: %d wakes, %d starts", b.wakes(), b.inst.startCount())

+ }

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

index 147dbe336..b6765fb33 100644

--- a/internal/host/movesreport.go

+++ b/internal/host/movesreport.go

@@ -138,6 +138,9 @@ func (w *movesWorker) targetReportPass(ctx context.Context, s release.Store, rec

if m.State != provision.MovePreparing && m.State != provision.MoveReady {

continue

}

+ if m.State == provision.MovePreparing && rec.Seed.MoveBlock() != "" {

+ continue // movesPass's seed rule (spec §5.8), which logged it

+ }

if err := w.targetReport(ctx, s, rec, fleet); err != nil {

w.logTrouble(rec.Host+" (target)", err)

continue