amadan

carlos / platform Public

Sidecar isolation Task 8 (4/4): every wake source refuses a fenced worker (D8, Review Focus 3)

All four sources (the due-time sweep, the sweep's always-on start, the
boot pass and carlos sidecars wake) reach doWake, which refuses the
in-memory fence at its first step and a journal through the Classifier
before the lease; the always-on arm now skips a fenced host as the due
arm already did, so a worker mid-move is not re-attempted and logged on
every sweep tick. Pinned at the activator (both fences, both sweep arms,
WakeWorker) and at the host (boot pass and wake-verb dispatch over a
journal with no in-memory fence, the agent-restart-mid-move case).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 603f43dc981613fedd27adbf93ed6819d78bfdc4 parent 5954fb4
3 files changed, +334 −0
  • internal/activator/activator.go +10 −0
  • internal/activator/worker_fenced_wake_test.go +178 −0
  • internal/host/worker_fenced_wake_test.go +146 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 4e37ea0af..c4dc469ba 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -5150,6 +5150,16 @@ func (a *Service) sweepOnce() {
if _, live := a.entries[rt.Host]; live {
continue // awake, waking, or draining a stop — none of those is stranded
}
+ // A fenced host is not stranded either: the layout pass (sidecar
+ // isolation D8) or another operation holds it stopped on purpose,
+ // and doWake would refuse the wake at its first step. Skipped here,
+ // as the due arm skips it, so a worker mid-move is not re-attempted
+ // and logged on every tick of its move. The journal fence (an agent
+ // restarted mid-move) is doWake's Classifier check; it has no cheap
+ // read here and the wake's refusal is the record of it.
+ if a.fences[rt.Host] != 0 {
+ continue
+ }
log.Printf("activator: always-on sidecar %s is not running — starting it", rt.Host)
e := a.newWorkerEntry(rt)
a.entries[rt.Host] = e
diff --git a/internal/activator/worker_fenced_wake_test.go b/internal/activator/worker_fenced_wake_test.go
new file mode 100644
index 000000000..01a7adc09
--- /dev/null
+++ b/internal/activator/worker_fenced_wake_test.go
@@ -0,0 +1,178 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// Sidecar isolation D8, Review Focus 3: while a worker is fenced for its
+// move, every wake source refuses it. The four sources (sidecar design
+// §7.2) all reach doWake: the due-time sweep and the sweep's always-on
+// start create the entry inside sweepOnce; `carlos sidecars wake` (host's
+// sidecar wake leg) and the agent's boot pass (host's runWorkerBootPass)
+// both call WakeWorker. Two fences cover two lives of the agent: the
+// in-memory fence the pass installs (FenceAndStop) while it runs, and the
+// journal in <data>/layout, which is all that remains after an agent
+// restart mid-move and which Classify reads as fenced before the lease.
+func TestFencedWorkerRefusesEveryWakeSource(t *testing.T) {
+ const (
+ timed = "bot.shop.abc123" + provision.SidecarSuffix // hibernating, due now
+ always = "cron.shop.abc123" + provision.SidecarSuffix // always-on
+ )
+ type fixture struct {
+ a *Service
+ b *fakeBackend
+ reg *registry.Registry
+ dd string
+ }
+ setup := func(t *testing.T) fixture {
+ t.Helper()
+ a, b, clk, reg := testActivator(t, time.Hour)
+ dd := t.TempDir()
+ // The real classification over a real journal directory, the way
+ // ExecBackend.Layout classifies in production.
+ b.layoutFn = func(rt registry.Route) (Layout, string) {
+ return Classify(rt, LayoutInputs{DataDir: dd, Journal: func(h string) bool {
+ _, err := os.Lstat(provision.LayoutJournalPath(dd, h))
+ return !errors.Is(err, os.ErrNotExist)
+ }})
+ }
+ for _, rt := range []registry.Route{
+ {Host: timed, Kind: "sidecar", Args: `["work"]`, Hibernate: true, Status: "ready", DBPath: provision.DefaultDBPath(dd, timed)},
+ {Host: always, Kind: "sidecar", Args: `["work"]`, Hibernate: false, Status: "ready", DBPath: provision.DefaultDBPath(dd, always)},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := reg.SetNextDue(timed, clk.now().Unix()-10); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ a.Shutdown(ctx)
+ })
+ return fixture{a, b, reg, dd}
+ }
+ row := func(t *testing.T, f fixture, h string) registry.Route {
+ t.Helper()
+ rt, err := f.reg.Get(h)
+ if err != nil || rt == nil {
+ t.Fatalf("row %s: %v %v", h, rt, err)
+ }
+ return *rt
+ }
+ // settle waits for the entries a refused sweep wake created to be
+ // gone: a failed wake deletes its entry on its own goroutine.
+ settle := func(t *testing.T, a *Service) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for a.entryCount() != 0 {
+ if time.Now().After(deadline) {
+ t.Fatalf("entries still live after refused wakes: %d", a.entryCount())
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ }
+ assertNothingStarted := func(t *testing.T, f fixture, when string) {
+ t.Helper()
+ if n := f.b.workerStarts(); n != 0 {
+ t.Fatalf("%s: %d worker start(s), want none", when, n)
+ }
+ if acquires, _, _ := f.b.counts(); acquires != 0 {
+ t.Fatalf("%s: %d lease acquire(s), want none: a refused wake must stop before the lease", when, acquires)
+ }
+ }
+
+ t.Run("in-memory fence while the pass runs", func(t *testing.T) {
+ f := setup(t)
+ for _, h := range []string{timed, always} {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ _, err := f.a.FenceAndStop(ctx, h, 7)
+ cancel()
+ if err != nil && !errors.Is(err, ErrStopUnproven) {
+ t.Fatalf("fence %s: %v", h, err)
+ }
+ }
+ // The two WakeWorker callers: the boot pass and `carlos sidecars wake`.
+ for _, h := range []string{timed, always} {
+ if err := f.a.WakeWorker(row(t, f, h)); !errors.Is(err, ErrFenced) {
+ t.Fatalf("WakeWorker(%s) under the fence = %v, want ErrFenced", h, err)
+ }
+ }
+ // The sweep: the due arm for the hibernating worker whose timer has
+ // fired, the always-on arm for the other. Neither makes an entry.
+ f.a.sweepOnce()
+ if n := f.a.entryCount(); n != 0 {
+ t.Fatalf("the sweep made %d entr(ies) for fenced workers", n)
+ }
+ assertNothingStarted(t, f, "under the in-memory fence")
+ // Lifted, the same sources wake them: the fixture is not inert.
+ for _, h := range []string{timed, always} {
+ f.a.LiftFence(h, 7)
+ }
+ if err := f.a.WakeWorker(row(t, f, timed)); err != nil {
+ t.Fatalf("WakeWorker after the lift: %v", err)
+ }
+ f.a.sweepOnce()
+ waitForStarts(t, f.b, 2)
+ })
+
+ t.Run("journal fence after an agent restart mid-move", func(t *testing.T) {
+ f := setup(t)
+ if err := os.MkdirAll(filepath.Join(f.dd, "layout"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ for _, h := range []string{timed, always} {
+ if err := os.WriteFile(provision.LayoutJournalPath(f.dd, h), []byte(`{"host":"`+h+`"}`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // No in-memory fence: this agent did not install one. The journal
+ // alone must refuse.
+ for _, h := range []string{timed, always} {
+ if f.a.Fenced(h) {
+ t.Fatalf("%s carries an in-memory fence the restarted agent never installed", h)
+ }
+ err := f.a.WakeWorker(row(t, f, h))
+ if err == nil || !strings.Contains(err.Error(), "layout migration in flight") {
+ t.Fatalf("WakeWorker(%s) with a journal = %v, want the fenced refusal", h, err)
+ }
+ }
+ f.a.sweepOnce()
+ settle(t, f.a)
+ assertNothingStarted(t, f, "under the journal fence")
+ // Recovery removed the journals: the same sources wake them.
+ for _, h := range []string{timed, always} {
+ if err := os.Remove(provision.LayoutJournalPath(f.dd, h)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := f.a.WakeWorker(row(t, f, timed)); err != nil {
+ t.Fatalf("WakeWorker after the journal is gone: %v", err)
+ }
+ f.a.sweepOnce()
+ waitForStarts(t, f.b, 2)
+ })
+}
+
+// waitForStarts waits for the backend to have started n workers.
+func waitForStarts(t *testing.T, b *fakeBackend, n int) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for b.workerStarts() < n {
+ if time.Now().After(deadline) {
+ t.Fatalf("worker starts = %d, want %d", b.workerStarts(), n)
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+}
diff --git a/internal/host/worker_fenced_wake_test.go b/internal/host/worker_fenced_wake_test.go
new file mode 100644
index 000000000..942b81ca1
--- /dev/null
+++ b/internal/host/worker_fenced_wake_test.go
@@ -0,0 +1,146 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// fencedWorkerBackend is an activator.Backend with the production
+// classifier wiring (activator.Classify over layout.JournalPresent, the
+// LayoutFacts RunAgent gives the exec backend) and a worker start that
+// records and reports ready at once.
+type fencedWorkerBackend struct {
+ mu sync.Mutex
+ dd string
+ svc *activator.Service
+ acquires int
+ started []registry.Route
+}
+
+type nopWorkerLease struct{}
+
+func (nopWorkerLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) {
+ return nopWorkerRep{}, nil
+}
+func (nopWorkerLease) Release(context.Context) error { return nil }
+
+type nopWorkerRep struct{}
+
+func (nopWorkerRep) Err() error { return nil }
+func (nopWorkerRep) Close(context.Context) error { return nil }
+
+type nopWorkerProc struct{}
+
+func (nopWorkerProc) Stop(context.Context) error { return nil }
+func (nopWorkerProc) Exited() bool { return false }
+
+func (b *fencedWorkerBackend) Layout(rt registry.Route) (activator.Layout, string) {
+ return activator.Classify(rt, activator.LayoutInputs{DataDir: b.dd, Journal: func(h string) bool { return layout.JournalPresent(b.dd, h) }})
+}
+func (b *fencedWorkerBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ b.acquires++
+ return nopWorkerLease{}, nil
+}
+func (b *fencedWorkerBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
+func (b *fencedWorkerBackend) Seed(context.Context, registry.Route) error { return nil }
+func (b *fencedWorkerBackend) Start(context.Context, registry.Route) (activator.Proc, string, error) {
+ return nil, "", errors.New("fencedWorkerBackend: no instance path in this fixture")
+}
+func (b *fencedWorkerBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
+func (b *fencedWorkerBackend) HasSidecar(registry.Route) bool { return false }
+func (b *fencedWorkerBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
+ return nil, errors.New("fencedWorkerBackend: no co-spawn in this fixture")
+}
+func (b *fencedWorkerBackend) StartWorker(_ context.Context, rt registry.Route, _ map[string]string) (activator.Proc, string, error) {
+ b.mu.Lock()
+ b.started = append(b.started, rt)
+ b.mu.Unlock()
+ go b.svc.NoteWorkerReady(rt.Host, "")
+ return nopWorkerProc{}, "", nil
+}
+
+func (b *fencedWorkerBackend) counts() (acquires, starts int) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.acquires, len(b.started)
+}
+
+// Sidecar isolation D8, Review Focus 3, the two wake sources that do not
+// come in through a request: the agent's boot pass (runWorkerBootPass) and
+// `carlos sidecars wake` (the sidecar wake leg's dispatchSidecarWakes),
+// both WakeWorker callers, refuse a worker whose layout journal is on disk
+// with no in-memory fence, which is what an agent restarted mid-move sees.
+// The activator package pins the sweep's two arms and the in-memory fence
+// (TestFencedWorkerRefusesEveryWakeSource).
+func TestWorkerBootPassAndWakeVerbRefuseAFencedWorker(t *testing.T) {
+ dd := t.TempDir()
+ reg := restartTestRegistry(t)
+ b := &fencedWorkerBackend{dd: dd}
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
+ svc.Routes = reg
+ b.svc = svc
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(ctx)
+ })
+ worker := "cron.shop.abc123" + provision.SidecarSuffix
+ rt := registry.Route{Host: worker, Kind: "sidecar", Args: `["work"]`, Status: "ready", DBPath: provision.DefaultDBPath(dd, worker)}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.MkdirAll(filepath.Join(dd, "layout"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(provision.LayoutJournalPath(dd, worker), []byte(`{"host":"`+worker+`"}`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ routes := []registry.Route{rt}
+
+ // The boot pass: synchronous, so its refusal is complete on return.
+ runWorkerBootPass(routes, svc.WakeWorker)
+ if acquires, starts := b.counts(); acquires != 0 || starts != 0 {
+ t.Fatalf("boot pass under the journal fence: %d acquire(s), %d start(s), want none", acquires, starts)
+ }
+ // The wake verb's dispatch: off the leg's return path, joined through
+ // the wake it calls.
+ done := make(chan error, 1)
+ dispatchSidecarWakes(routes, func(rt registry.Route) error {
+ err := svc.WakeWorker(rt)
+ done <- err
+ return err
+ })
+ select {
+ case err := <-done:
+ if err == nil || !strings.Contains(err.Error(), "layout migration in flight") {
+ t.Fatalf("wake verb under the journal fence = %v, want the fenced refusal", err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("the dispatched wake never returned")
+ }
+ if acquires, starts := b.counts(); acquires != 0 || starts != 0 {
+ t.Fatalf("wake verb under the journal fence: %d acquire(s), %d start(s), want none", acquires, starts)
+ }
+
+ // Recovery removed the journal: the boot pass starts it.
+ if err := os.Remove(provision.LayoutJournalPath(dd, worker)); err != nil {
+ t.Fatal(err)
+ }
+ runWorkerBootPass(routes, svc.WakeWorker)
+ if acquires, starts := b.counts(); acquires != 1 || starts != 1 {
+ t.Fatalf("boot pass after the journal is gone: %d acquire(s), %d start(s), want one each", acquires, starts)
+ }
+}