amadan

carlos / platform Public

activator: record a seed placed after the first upload (Plan 4 Task 4)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude bdcf811bcc64bcbc4098fb1e1d3ed957f695bee0 parent 950e953
5 files changed, +226 −3
  • internal/activator/activator.go +15 −0
  • internal/activator/activator_test.go +45 −0
  • internal/activator/backend_unit.go +9 −0
  • internal/activator/seedplace.go +35 −3
  • internal/activator/seedplace_linux_test.go +122 −0

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 3fd5980d1..9e0107012 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -619,6 +619,12 @@ type configAdmitter interface {
AdmitConfig(ctx context.Context, rt registry.Route) error
}
+// seedPlacer is the optional half of a Backend that records a seed placed
+// after the first upload (seeded-instances spec §3.4).
+type seedPlacer interface {
+ SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error
+}
+
// configRefusedRetryAfter is the Retry-After (seconds) a refused wake sends.
const configRefusedRetryAfter = 30
@@ -1344,6 +1350,15 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
return w, fmt.Errorf("replicate %s: %w", host, err)
}
+ // 5b. A seeded row's first upload, then its record (spec §3.4 step 5).
+ // A failure here does not fail the wake: the instance is serving from
+ // the placed file, and the next wake records it (finishPlacement).
+ if p, ok := a.backend.(seedPlacer); ok {
+ if err := p.SeedPlaced(ctx, rt, rep); err != nil {
+ log.Printf("activator: %s: seed placement not recorded yet: %v", host, err)
+ }
+ }
+
// 6. Co-spawn the sidecar, last — only once the instance is healthy and
// replicating, so the sidecar always boots against a live instance (its
// marked requests pass through the edge while we're awake). A spawn
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index e9823ca7b..ac567e48c 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -35,6 +35,7 @@ import (
"github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/seeds"
_ "modernc.org/sqlite"
)
@@ -6452,3 +6453,47 @@ func TestStaleSidecarRestartNeverInstallsOntoAFreshWake(t *testing.T) {
t.Fatal("the stale restart installed over the fresh wake's own sidecar")
}
}
+
+// seedCountingBackend is fakeBackend plus the optional seed recorder.
+type seedCountingBackend struct {
+ fakeBackend
+ placed []string
+ rep Rep
+ replicated int // Replicate calls seen when SeedPlaced ran
+ err error
+}
+
+func (b *seedCountingBackend) SeedPlaced(_ context.Context, rt registry.Route, rep Rep) error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ b.placed = append(b.placed, rt.Host)
+ b.rep, b.replicated = rep, len(b.syncIntervals)
+ return b.err
+}
+
+func TestDoWakeRecordsSeedPlacementAfterReplication(t *testing.T) {
+ for _, recErr := range []error{nil, errors.New("record unreachable")} {
+ b := &seedCountingBackend{err: recErr}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
+ rt := testRoute("shop.example.com")
+ rt.SeedSource = seeds.RowSeed{ID: "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa", Kind: seeds.KindReplica, State: seeds.StateReady}.Encode()
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ // A failed record never fails the wake: the instance serves.
+ arrangeAwake(t, a, rt)
+ if len(b.placed) != 1 || b.placed[0] != rt.Host {
+ t.Fatalf("SeedPlaced calls = %q, want one, after replication started", b.placed)
+ }
+ if len(b.reps) != 1 || b.rep != Rep(b.reps[0]) || b.replicated != 1 {
+ t.Fatalf("SeedPlaced saw rep %v after %d replications; want the wake's own, after Replicate", b.rep, b.replicated)
+ }
+ if !a.Owns(rt.Host) {
+ t.Fatalf("recorder error %v: the wake did not serve", recErr)
+ }
+ }
+}
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index 19535c0a5..91ef5a1aa 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -308,6 +308,15 @@ func (b *SplitBackend) AdmitConfig(ctx context.Context, rt registry.Route) error
}
return nil
}
+
+// SeedPlaced dispatches on Backing, like every other call. A unit tenant
+// never carries a seed.
+func (b *SplitBackend) SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error {
+ if p, ok := b.pick(rt).(seedPlacer); ok {
+ return p.SeedPlaced(ctx, rt, rep)
+ }
+ return nil
+}
func (b *SplitBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
return b.pick(rt).Restore(ctx, rt)
}
diff --git a/internal/activator/seedplace.go b/internal/activator/seedplace.go
index 5801cba2f..e998ef61b 100644
--- a/internal/activator/seedplace.go
+++ b/internal/activator/seedplace.go
@@ -355,6 +355,38 @@ func (e *readErrs) Read(p []byte) (int, error) {
return n, err
}
-// finishPlacement is Task 4's: the record's CAS to placed and the
-// snapshot's delete, best effort, once the host has a replica of its own.
-func (b *ExecBackend) finishPlacement(context.Context, registry.Route, *seeds.RowSeed) {}
+// SeedPlaced records a seed placed once the host's own replica holds it
+// (spec §3.4 steps 4 and 5): it waits for the first upload, then
+// OnSeedPlaced CASes the record to placed and deletes seeds/<host>/<id>/.
+// The snapshot goes only after the replica exists, so a crash at any point
+// retries from the placed local file or the snapshot. A replicator that
+// cannot be asked for its first upload records nothing.
+func (b *ExecBackend) SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error {
+ src, err := seeds.ParseRowSeed(rt.SeedSource)
+ if err != nil || src == nil || !src.Placeable() || src.State == seeds.StatePlaced {
+ return err
+ }
+ s, ok := rep.(interface{ SyncAndWait(context.Context) error })
+ if !ok {
+ return fmt.Errorf("first upload of %s: the replicator cannot confirm it", rt.Host)
+ }
+ if err := s.SyncAndWait(ctx); err != nil {
+ return fmt.Errorf("first upload of %s: %w", rt.Host, err)
+ }
+ if b.OnSeedPlaced == nil {
+ return errors.New("no seed recorder on this box")
+ }
+ return b.OnSeedPlaced(ctx, rt.Account, rt.App, rt.Host, src.ID)
+}
+
+// finishPlacement is SeedPlaced for a host whose own replica already
+// exists (a placed CAS lost on an earlier wake). Best effort: a failure is
+// logged, never the wake's, and the next wake asks again.
+func (b *ExecBackend) finishPlacement(ctx context.Context, rt registry.Route, src *seeds.RowSeed) {
+ if b.OnSeedPlaced == nil {
+ return
+ }
+ if err := b.OnSeedPlaced(ctx, rt.Account, rt.App, rt.Host, src.ID); err != nil {
+ log.Printf("activator: %s: seed %s not recorded as placed yet: %v", rt.Host, src.ID, err)
+ }
+}
diff --git a/internal/activator/seedplace_linux_test.go b/internal/activator/seedplace_linux_test.go
index 13e2381bc..41282ac8e 100644
--- a/internal/activator/seedplace_linux_test.go
+++ b/internal/activator/seedplace_linux_test.go
@@ -587,3 +587,125 @@ func TestSeededRestorePlacesAWALModeSnapshot(t *testing.T) {
}
}
}
+
+// syncRep is a Rep that can be asked for its first upload.
+type syncRep struct {
+ synced int
+ err error
+}
+
+func (r *syncRep) Err() error { return nil }
+func (r *syncRep) Close(context.Context) error { return nil }
+func (r *syncRep) SyncAndWait(context.Context) error { r.synced++; return r.err }
+
+// plainRep is a Rep that cannot be asked for its first upload.
+type plainRep struct{}
+
+func (plainRep) Err() error { return nil }
+func (plainRep) Close(context.Context) error { return nil }
+
+func withSeedState(t *testing.T, rt registry.Route, state string) registry.Route {
+ t.Helper()
+ src, err := seeds.ParseRowSeed(rt.SeedSource)
+ if err != nil || src == nil {
+ t.Fatalf("seed column %q: %v", rt.SeedSource, err)
+ }
+ src.State = state
+ rt.SeedSource = src.Encode()
+ return rt
+}
+
+func TestSeedPlacedWaitsForTheFirstUpload(t *testing.T) {
+ b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{})
+ rt, _, id := seededRoute(t, b, rt, seeds.StateReady)
+ rt.Account, rt.App = "acct", "shop"
+ var calls []string
+ b.OnSeedPlaced = func(_ context.Context, acct, app, host, got string) error {
+ calls = append(calls, acct+"/"+app+"/"+host+"/"+got)
+ return nil
+ }
+ rep := &syncRep{}
+ if err := b.SeedPlaced(context.Background(), rt, rep); err != nil {
+ t.Fatal(err)
+ }
+ if rep.synced != 1 || len(calls) != 1 || calls[0] != "acct/shop/"+rt.Host+"/"+id {
+ t.Fatalf("synced %d, calls %q", rep.synced, calls)
+ }
+ // A failed first upload records nothing.
+ rep = &syncRep{err: errors.New("upload failed")}
+ calls = nil
+ if err := b.SeedPlaced(context.Background(), rt, rep); err == nil || len(calls) != 0 {
+ t.Fatalf("after a failed sync: err %v calls %q", err, calls)
+ }
+ // A replicator that cannot confirm the upload records nothing either:
+ // the snapshot must outlive every moment the replica might not exist.
+ if err := b.SeedPlaced(context.Background(), rt, plainRep{}); err == nil || len(calls) != 0 {
+ t.Fatalf("unconfirmable upload: err %v calls %q", err, calls)
+ }
+ // A recorder that fails is the caller's to log.
+ b.OnSeedPlaced = func(context.Context, string, string, string, string) error { return errors.New("conflict") }
+ if err := b.SeedPlaced(context.Background(), rt, &syncRep{}); err == nil {
+ t.Fatal("a failed record reported success")
+ }
+ b.OnSeedPlaced = nil
+ if err := b.SeedPlaced(context.Background(), rt, &syncRep{}); err == nil {
+ t.Fatal("a box with no recorder reported the seed recorded")
+ }
+ b.OnSeedPlaced = func(_ context.Context, acct, app, host, got string) error {
+ calls = append(calls, got)
+ return nil
+ }
+ // Placed already, not placeable, or no seed: nothing at all.
+ for name, r := range map[string]registry.Route{
+ "placed": withSeedState(t, rt, seeds.StatePlaced),
+ "aborted": withSeedState(t, rt, seeds.StateAborted),
+ "none": func() registry.Route { r := rt; r.SeedSource = ""; return r }(),
+ } {
+ rep = &syncRep{}
+ if err := b.SeedPlaced(context.Background(), r, rep); err != nil || rep.synced != 0 || len(calls) != 0 {
+ t.Fatalf("%s row: err %v synced %d calls %q", name, err, rep.synced, calls)
+ }
+ }
+}
+
+// Review Focus 5: the placed CAS failed after the first upload, so the row
+// still reads ready. The next wake finds the host's own replica and
+// finishes the placement; a failure there never fails the wake, and the
+// wake after that asks again.
+func TestSeedPlacedIsFinishedOnTheNextWake(t *testing.T) {
+ old := hasOwnReplica
+ hasOwnReplica = func(context.Context, string, string, string) (bool, error) { return true, nil }
+ defer func() { hasOwnReplica = old }()
+ oldRestore := restoreReplica
+ restoreReplica = func(_ context.Context, dbPath, _, _, _ string) (bool, error) { return true, writeMinimalSQLite(dbPath) }
+ defer func() { restoreReplica = oldRestore }()
+ b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{})
+ rt, objs, id := seededRoute(t, b, rt, seeds.StateReady) // the CAS was lost last wake
+ rt.Account, rt.App = "acct", "shop"
+ var got []string
+ var recErr error
+ b.OnSeedPlaced = func(_ context.Context, acct, app, host, seed string) error {
+ got = append(got, acct+"/"+app+"/"+host+"/"+seed)
+ return recErr
+ }
+ want := "acct/shop/" + rt.Host + "/" + id
+ recErr = errors.New("record unreachable")
+ if fresh, err := b.Restore(context.Background(), rt); err != nil || fresh {
+ t.Fatalf("Restore with a failing recorder = fresh %v, %v; want a served wake", fresh, err)
+ }
+ recErr = nil
+ if fresh, err := b.Restore(context.Background(), rt); err != nil || fresh {
+ t.Fatalf("Restore = fresh %v, %v", fresh, err)
+ }
+ if len(got) != 2 || got[0] != want || got[1] != want {
+ t.Fatalf("finishPlacement calls = %q, want two for %s", got, want)
+ }
+ if objs.opens != 0 {
+ t.Fatalf("snapshot opened %d times; the own replica wins", objs.opens)
+ }
+ // No recorder on this box: the wake still serves.
+ b.OnSeedPlaced = nil
+ if _, err := b.Restore(context.Background(), rt); err != nil {
+ t.Fatal(err)
+ }
+}