amadan
Retain process leases for root-replicated sidecars
13 files changed,
+551
−46
internal/activator/activator.go+67 −17internal/activator/activator_test.go+43 −23internal/activator/backend_exec.go+2 −0internal/activator/backend_unit.go+3 −0internal/activator/maintenance.go+7 −0internal/activator/maintenance_test.go+3 −3internal/activator/replication_owner_test.go+340 −0internal/host/schedule_wake_test.go+2 −0internal/ops/hibernate_verify.go+1 −1internal/ops/litestream.go+2 −2internal/ops/ops_test.go+63 −0internal/registry/registry.go+2 −0internal/registry/replication_owner_test.go+16 −0
CI passed — run details
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index b8889e66a..54c354fed 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -240,6 +240,7 @@ type RouteLookup interface { |
| // Lease is a held-but-not-yet-replicating lease (wake in progress; Eleven's |
| // actLease). |
| type Lease interface { |
| + Err() error |
| // Replicate trades the lease for running replication of the (now present) |
| // local DB; the lease handle is spent on success. syncInterval is the |
| // route's replication-class tick (replicationSyncInterval; 0 = stream). |
| @@ -360,8 +361,10 @@ type actEntry struct { |
| inflight int |
| lastActive time.Time |
| - rep Rep |
| - proc Proc |
| + replicates bool |
| + lease Lease |
| + rep Rep |
| + proc Proc |
| // Convergence: the `live/<host>` symlink target this child was spawned |
| // against. A running process never re-reads the symlink, so once a ship |
| @@ -447,6 +450,17 @@ type actEntry struct { |
| respawnDone chan struct{} |
| } |
| +func entryHealthErr(e *actEntry) error { |
| + if e.replicates { |
| + if e.rep != nil { |
| + return e.rep.Err() |
| + } |
| + } else if e.lease != nil { |
| + return e.lease.Err() |
| + } |
| + return os.ErrInvalid |
| +} |
| + |
| // Sidecar restart pacing — Restart=always semantics without a supervisor: the |
| // sweeper redoes a dead sidecar with exponential backoff, capped, and a run |
| // that survived sidecarHealthyRun earns a fresh backoff (so one crash a day |
| @@ -723,7 +737,7 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel |
| // lost lease found at request time turns into a stop, and the |
| // request waits for it then re-wakes — recovery is the routine |
| // wake path. |
| - if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() { |
| + if repErr := entryHealthErr(e); repErr != nil || e.proc.Exited() { |
| if repErr != nil { |
| log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| } else { |
| @@ -808,7 +822,7 @@ func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) er |
| // the recovery stops at the stop: the loop then finds a stopping |
| // (or absent) entry and turns this request away instead of |
| // chaining into a re-wake. |
| - if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() { |
| + if repErr := entryHealthErr(e); repErr != nil || e.proc.Exited() { |
| if repErr != nil { |
| log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| } else { |
| @@ -939,10 +953,17 @@ func (a *Service) wake(e *actEntry) { |
| outcome = w.stopOutcome |
| } |
| a.rememberStopLocked(e.host, outcome) |
| - delete(a.entries, e.host) // next request retries from scratch |
| + if w != nil && w.pendingStop { |
| + e.state, e.stopped = actStopping, make(chan struct{}) |
| + e.proc, e.lease, e.replicates = w.proc, w.lease, w.replicates |
| + e.pendingStop, e.stopOutcome, e.stopRetriedAt = true, outcome, a.now() |
| + } else { |
| + delete(a.entries, e.host) // next request retries from scratch |
| + } |
| } else { |
| e.state = actAwake |
| e.rep, e.proc = w.rep, w.proc |
| + e.replicates, e.lease = w.replicates, w.lease |
| e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted |
| e.sidecarUncertain = w.sidecarUncertain |
| e.sidecarStarted = a.now() |
| @@ -1029,6 +1050,9 @@ func wakeEventMessage(w *wokenInstance, worker bool) string { |
| type wokenInstance struct { |
| stopOutcome *stopObservation |
| sidecarUncertain bool |
| + replicates bool |
| + lease Lease |
| + pendingStop bool |
| rep Rep |
| proc Proc |
| sidecar Proc |
| @@ -1102,6 +1126,11 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| } |
| } |
| + w.replicates = !e.worker || rt.ActivatorOwnsReplication() |
| + if !w.replicates && rt.Seed != "" { |
| + return w, os.ErrInvalid |
| + } |
| + |
| // 1. The lease first: everything after this line is done as the prefix's |
| // single writer. Contention is not retried here — on one box it means a |
| // previous incarnation hasn't released; tell the client to come back after |
| @@ -1120,6 +1149,9 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| return w, err |
| } |
| + if !w.replicates { |
| + w.lease = lease |
| + } |
| if ctx.Err() != nil || !a.wakeAllowed(host) { |
| w.stopOutcome.leaseReleased = lease.Release(ctx) == nil |
| return w, ErrFenced |
| @@ -1129,7 +1161,10 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // S3 state written by the true holder). When it's still on disk it is the |
| // warm cache: on one box, with the lease held, the local copy can only be |
| // what we ourselves checkpointed at the last sleep. |
| - fresh, err := a.backend.Restore(ctx, rt) |
| + fresh := false |
| + if w.replicates { |
| + fresh, err = a.backend.Restore(ctx, rt) |
| + } |
| if err != nil { |
| w.stopOutcome.leaseReleased = lease.Release(ctx) == nil |
| return w, fmt.Errorf("restore %s: %w", host, err) |
| @@ -1202,8 +1237,8 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // is a wake failure like any other: release and report. |
| token, err = newWorkerToken() |
| if err != nil { |
| - _ = lease.Release(ctx) |
| - return nil, fmt.Errorf("start %s: %w", host, err) |
| + w.stopOutcome.leaseReleased = lease.Release(ctx) == nil |
| + return w, fmt.Errorf("start %s: %w", host, err) |
| } |
| // Published onto the entry NOW, not at the awake transition: the |
| // child's first act is to call /v1/ready over the agent socket, which |
| @@ -1270,7 +1305,10 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // The DB path rides the route; when empty the backend defaults it to |
| // <DataDir>/<host>.db internally (Eleven defaulted here via a package |
| // global — base dirs are backend fields now). |
| - rep, err := lease.Replicate(ctx, rt.DBPath, replicationSyncInterval(rt.Replication)) |
| + var rep Rep |
| + if w.replicates { |
| + rep, err = lease.Replicate(ctx, rt.DBPath, replicationSyncInterval(rt.Replication)) |
| + } |
| w.rep = rep |
| if err != nil { |
| a.cleanupFailedWake(ctx, w, lease) |
| @@ -1535,7 +1573,7 @@ func (a *Service) WakeWorker(rt registry.Route) error { |
| // makes a broken sidecar look healthy. A respawn in flight is not a |
| // dead child: its proc is mid-replacement, so leave it be. |
| if !e.workerRespawning { |
| - if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() { |
| + if repErr := entryHealthErr(e); repErr != nil || e.proc.Exited() { |
| if repErr != nil { |
| log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| } else { |
| @@ -2040,7 +2078,7 @@ func (a *Service) stop(e *actEntry) { |
| var err error |
| if e.proc != nil { |
| err = e.proc.Stop(procCtx) |
| - if errors.Is(err, ErrStillRunning) { |
| + if errors.Is(err, ErrStillRunning) || (!e.replicates && !e.proc.Exited()) { |
| // The lease is never released over an unproven termination |
| // (design §6.3, codex R2-5): stop() bails out here rather than |
| // falling through to the checkpoint and lease release below, |
| @@ -2064,6 +2102,12 @@ func (a *Service) stop(e *actEntry) { |
| outcome.mainStop, outcome.mainExited = err == nil, e.proc.Exited() |
| } else { |
| err = ErrStopUnproven |
| + if !e.replicates { |
| + a.mu.Lock() |
| + e.pendingStop, e.stopOutcome, e.stopRetriedAt = true, outcome, a.now() |
| + a.mu.Unlock() |
| + return |
| + } |
| } |
| if err != nil { |
| log.Printf("activator: stop %s: %v", e.host, err) |
| @@ -2162,11 +2206,15 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) { |
| ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| defer cancel() |
| repErr := ErrStopUnproven |
| - if e.rep != nil { |
| - repErr = e.rep.Close(ctx) |
| + if e.replicates { |
| + if e.rep != nil { |
| + repErr = e.rep.Close(ctx) |
| + } |
| + outcome.replicaClosed = repErr == nil |
| + } else if outcome.mainExited && e.lease != nil { |
| + outcome.leaseReleased = e.lease.Release(ctx) == nil |
| } |
| - outcome.replicaClosed = repErr == nil |
| - if repErr != nil { |
| + if e.replicates && repErr != nil { |
| log.Printf("activator: final checkpoint %s: %v", e.host, repErr) |
| } |
| a.mu.Lock() |
| @@ -2176,7 +2224,9 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) { |
| delete(a.entries, e.host) |
| a.mu.Unlock() |
| close(e.stopped) |
| - log.Printf("activator: %s hibernated", e.host) |
| + if e.replicates { |
| + log.Printf("activator: %s hibernated", e.host) |
| + } |
| } |
| // Start launches the background sweeper that puts idle instances to sleep and |
| @@ -2263,7 +2313,7 @@ func (a *Service) sweepOnce() { |
| if e.workerRespawning { |
| continue |
| } |
| - if err := e.rep.Err(); err != nil { |
| + if err := entryHealthErr(e); err != nil { |
| log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err) |
| a.beginStopLocked(e) |
| continue |
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go| index a16d90892..cf7354bbb 100644 |
| --- a/internal/activator/activator_test.go |
| +++ b/internal/activator/activator_test.go |
| @@ -3380,6 +3380,36 @@ func alwaysOnWorkerRoute(host string) registry.Route { |
| return rt |
| } |
| +// An always-on native sidecar's database is owned by the box-wide replicator |
| +// (sidecar design §15.3), while its single-runner lease remains held across |
| +// child recycles (§7.2 and §13). Starting it must therefore acquire and retain |
| +// the lease without restoring from the activator prefix or turning that lease |
| +// into an embedded replicator. This desired regression intentionally fails on |
| +// the pre-fix implementation. |
| +func TestAlwaysOnWorkerDoesNotEnterHibernationReplicationPath(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| + t.Fatalf("WakeWorker: %v", err) |
| + } |
| + acquires, restores, _ := b.counts() |
| + if starts := b.workerStarts(); starts != 1 { |
| + t.Fatalf("worker starts = %d, want 1", starts) |
| + } |
| + if acquires != 1 { |
| + t.Fatalf("lease acquires = %d, want 1 single-runner lease", acquires) |
| + } |
| + b.mu.Lock() |
| + replicates := len(b.syncIntervals) |
| + releases := b.releases |
| + b.mu.Unlock() |
| + if restores != 0 || replicates != 0 { |
| + t.Fatalf("always-on worker entered hibernation data path: restores=%d replicates=%d", restores, replicates) |
| + } |
| + if releases != 0 { |
| + t.Fatalf("single-runner lease released while worker is awake: releases=%d", releases) |
| + } |
| +} |
| + |
| // An always-on worker's exit is a CRASH, not a finished job: it is restarted |
| // in place, keeping its lease and replication — and the backoff GROWS, so a |
| // crash-looping bot paces itself instead of hot-looping the box. |
| @@ -3601,7 +3631,7 @@ func TestAlwaysOnWorkerMaxRunRecyclesItsChild(t *testing.T) { |
| if !first.isStopped() { |
| t.Fatal("the wedged child was left running beside its replacement") |
| } |
| - if b.lastRep().isClosed() { |
| + if b.releaseCount() != 0 { |
| t.Fatal("the lease was released by a max_run recycle — the entry keeps its lease and replication") |
| } |
| @@ -3768,7 +3798,6 @@ func TestRespawnShedsARacedChildBeforeTheLeaseIsReleased(t *testing.T) { |
| if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| t.Fatalf("WakeWorker: %v", err) |
| } |
| - rep := b.lastRep() |
| // The gate applies to children spawned from now on — i.e. the respawn's, |
| // the one that will have to be shed. Its drain is held open so the |
| // ordering is observable rather than a microsecond wide. |
| @@ -3794,7 +3823,7 @@ func TestRespawnShedsARacedChildBeforeTheLeaseIsReleased(t *testing.T) { |
| // It is now draining, held by the gate. The lease must still be held. |
| deadline := time.Now().Add(300 * time.Millisecond) |
| for time.Now().Before(deadline) { |
| - if rep.isClosed() { |
| + if b.releaseCount() != 0 { |
| t.Fatal("the lease was released while the respawn's raced child was still draining — a live process is pointed at a database we no longer hold") |
| } |
| time.Sleep(10 * time.Millisecond) |
| @@ -3802,7 +3831,7 @@ func TestRespawnShedsARacedChildBeforeTheLeaseIsReleased(t *testing.T) { |
| close(gate) // let the shed finish |
| waitFor(t, "the stop to complete", func() bool { return a.entryCount() == 0 }) |
| - if n := rep.closeCount(); n != 1 { |
| + if n := b.releaseCount(); n != 1 { |
| t.Fatalf("lease closes = %d, want exactly 1", n) |
| } |
| for i, p := range b.workers { |
| @@ -3831,7 +3860,6 @@ func TestStopScheduledAfterTheInterlockClearStillWaitsForTheShed(t *testing.T) { |
| if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| t.Fatalf("WakeWorker: %v", err) |
| } |
| - rep := b.lastRep() |
| gate := make(chan struct{}) |
| b.mu.Lock() |
| b.workerStartDelay = 150 * time.Millisecond |
| @@ -3866,7 +3894,7 @@ func TestStopScheduledAfterTheInterlockClearStillWaitsForTheShed(t *testing.T) { |
| // releasing it here is releasing it under a live process. |
| deadline := time.Now().Add(300 * time.Millisecond) |
| for time.Now().Before(deadline) { |
| - if rep.isClosed() { |
| + if b.releaseCount() != 0 { |
| t.Fatal("the lease was released while the respawn's raced child was still draining — the stop skipped its wait because the interlock had already been cleared") |
| } |
| time.Sleep(10 * time.Millisecond) |
| @@ -3874,7 +3902,7 @@ func TestStopScheduledAfterTheInterlockClearStillWaitsForTheShed(t *testing.T) { |
| close(gate) |
| waitFor(t, "the stop to complete", func() bool { return a.entryCount() == 0 }) |
| - if n := rep.closeCount(); n != 1 { |
| + if n := b.releaseCount(); n != 1 { |
| t.Fatalf("lease closes = %d, want exactly 1", n) |
| } |
| for i, p := range b.workers { |
| @@ -3894,7 +3922,6 @@ func TestShutdownCancelsAnInFlightRespawnInsteadOfWaiting(t *testing.T) { |
| if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| t.Fatalf("WakeWorker: %v", err) |
| } |
| - rep := b.lastRep() |
| b.mu.Lock() |
| b.workerStartDelay = 5 * time.Second // a materialize nobody should wait for |
| b.mu.Unlock() |
| @@ -3927,20 +3954,11 @@ func TestShutdownCancelsAnInFlightRespawnInsteadOfWaiting(t *testing.T) { |
| t.Fatalf("worker child %d was left running by Shutdown", i) |
| } |
| } |
| - // The lease: released exactly once, by stop()'s rep.Close — the respawn |
| - // never owned it, so cancelling one can neither skip nor double the |
| - // release. |
| - if n := rep.closeCount(); n != 1 { |
| - t.Fatalf("lease closes = %d, want exactly 1 on the cancelled-respawn shutdown path", n) |
| - } |
| - // releaseCount counts the lease coming FREE by either route — fakeRep.Close |
| - // after a wake got past Replicate, or a raw Lease.Release before it (see |
| - // fakeRep.b). The two are mutually exclusive per wake, so exactly 1 here |
| - // says the same release rep.closeCount() just counted and no second one |
| - // beside it: cancelling a respawn that never owned the lease can neither |
| - // skip the release nor double it. |
| if n := b.releaseCount(); n != 1 { |
| - t.Fatalf("lease releases = %d, want exactly 1 — the lease was traded for replication at wake, so it comes free through rep.Close and nowhere else", n) |
| + t.Fatalf("lease releases = %d, want exactly 1", n) |
| + } |
| + if n := b.lastRep().closeCount(); n != 0 { |
| + t.Fatalf("replica closes = %d, want 0", n) |
| } |
| // And the abandoned goroutine really is gone: nothing appears later. |
| time.Sleep(300 * time.Millisecond) |
| @@ -3981,7 +3999,7 @@ func TestStopWaitsOutAnInFlightRespawn(t *testing.T) { |
| // leaving nothing to ask. |
| deadline := time.Now().Add(300 * time.Millisecond) |
| for time.Now().Before(deadline) { |
| - if b.lastRep().isClosed() && b.workerStarts() < 2 { |
| + if b.releaseCount() != 0 && b.workerStarts() < 2 { |
| t.Fatal("the lease was released while a respawn's exec was still running — the child appearing behind it would have no single-writer claim") |
| } |
| time.Sleep(10 * time.Millisecond) |
| @@ -4063,7 +4081,7 @@ func TestAlwaysOnWorkerRespawnsOntoANewBuild(t *testing.T) { |
| if !a.entryAwake("bot.x.internal") { |
| t.Fatal("the worker is not awake after converging — the respawn must land on the SAME entry") |
| } |
| - if b.lastRep().isClosed() || b.releaseCount() != 0 { |
| + if b.releaseCount() != 0 { |
| t.Fatal("converging respawned through a stop: the lease must not churn for an always-on worker") |
| } |
| // The new child holds the live token, so the agent socket still knows it. |
| @@ -6404,3 +6422,5 @@ func TestStaleSidecarRestartNeverInstallsOntoAFreshWake(t *testing.T) { |
| t.Fatal("the stale restart installed over the fresh wake's own sidecar") |
| } |
| } |
| + |
| +func (l *fakeLease) Err() error { return l.rep.Err() } |
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 68b5657c7..d3dfeddc9 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -1652,3 +1652,5 @@ func (b *ExecBackend) dropStagedReplicaWithMirror(ctx context.Context, staged ma |
| _ = mirror.Delete(ctx, b.Cfg.Bucket, k) |
| } |
| } |
| + |
| +func (l *execLease) Err() error { return l.held.Err() } |
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go| index 2f35a7bf6..3fa616e9a 100644 |
| --- a/internal/activator/backend_unit.go |
| +++ b/internal/activator/backend_unit.go |
| @@ -51,6 +51,7 @@ import ( |
| "errors" |
| "fmt" |
| "net/http" |
| + "os" |
| "os/exec" |
| "strings" |
| "sync" |
| @@ -393,3 +394,5 @@ func (b *SplitBackend) LiveVersion(host string) (string, error) { |
| } |
| return "", nil |
| } |
| + |
| +func (unitLease) Err() error { return os.ErrInvalid } |
diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go| index 7717dd27e..6d9542418 100644 |
| --- a/internal/activator/maintenance.go |
| +++ b/internal/activator/maintenance.go |
| @@ -174,6 +174,13 @@ func (a *Service) cleanupFailedWake(ctx context.Context, w *wokenInstance, lease |
| if w.proc != nil { |
| err := w.proc.Stop(stopCtx) |
| w.stopOutcome.mainStop, w.stopOutcome.mainExited = err == nil, w.proc.Exited() |
| + if !w.replicates && errors.Is(err, ErrStillRunning) { |
| + w.pendingStop = true |
| + } |
| + } |
| + if !w.replicates && w.proc != nil && (!w.stopOutcome.mainExited || w.pendingStop) { |
| + w.pendingStop = true |
| + return |
| } |
| // A backend failure may conceal resources beyond the returned handle. |
| // Successful cleanup of that handle does not erase the uncertainty. |
diff --git a/internal/activator/maintenance_test.go b/internal/activator/maintenance_test.go| index 8a78ed15f..8b6ae5ffa 100644 |
| --- a/internal/activator/maintenance_test.go |
| +++ b/internal/activator/maintenance_test.go |
| @@ -72,7 +72,7 @@ func maintenanceAwake(t *testing.T) (*Service, *actEntry, *maintenanceProc, *mai |
| a, _, _, reg := testActivator(t, time.Hour) |
| t.Cleanup(func() { reg.Close() }) |
| main, side, rep := &maintenanceProc{}, &maintenanceProc{}, &maintenanceRep{} |
| - e := &actEntry{host: "fence.example", route: testRoute("fence.example"), state: actAwake, proc: main, sidecar: side, rep: rep, lastActive: a.now()} |
| + e := &actEntry{replicates: true, host: "fence.example", route: testRoute("fence.example"), state: actAwake, proc: main, sidecar: side, rep: rep, lastActive: a.now()} |
| a.entries[e.host] = e |
| return a, e, main, side, rep |
| } |
| @@ -232,7 +232,7 @@ func TestMaintenanceLaterOrdinaryStopCannotEraseUnresolvedCleanup(t *testing.T) |
| done := e.stopped |
| a.mu.Unlock() |
| <-done |
| - replacement := &actEntry{host: e.host, route: e.route, state: actAwake, proc: &maintenanceProc{}, sidecar: &maintenanceProc{}, rep: &maintenanceRep{}} |
| + replacement := &actEntry{replicates: true, host: e.host, route: e.route, state: actAwake, proc: &maintenanceProc{}, sidecar: &maintenanceProc{}, rep: &maintenanceRep{}} |
| a.mu.Lock() |
| a.entries[e.host] = replacement |
| a.beginStopLocked(replacement) |
| @@ -527,7 +527,7 @@ func TestMaintenanceFailedSidecarHandlesRemainOwned(t *testing.T) { |
| } else { |
| old := &maintenanceProc{} |
| old.exited.Store(true) |
| - a.entries[rt.Host] = &actEntry{host: rt.Host, route: rt, state: actAwake, proc: main, rep: rep, sidecar: old, sidecarWanted: true, lastActive: a.now()} |
| + a.entries[rt.Host] = &actEntry{replicates: true, host: rt.Host, route: rt, state: actAwake, proc: main, rep: rep, sidecar: old, sidecarWanted: true, lastActive: a.now()} |
| a.sweepOnce() |
| // The restart call runs off a.mu (design §6.7); give it a |
| // moment to land before inspecting what it installed. |
diff --git a/internal/activator/replication_owner_test.go b/internal/activator/replication_owner_test.go| new file mode 100644 |
| index 000000000..6ca51b94e |
| --- /dev/null |
| +++ b/internal/activator/replication_owner_test.go |
| @@ -0,0 +1,340 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +type ownerLease struct { |
| + Lease |
| + health atomic.Int32 |
| +} |
| + |
| +func (l *ownerLease) Err() error { l.health.Add(1); return os.ErrInvalid } |
| + |
| +type ownerBackend struct { |
| + *fakeBackend |
| + child Proc |
| + failure error |
| + acquired chan struct{} |
| + proceed <-chan struct{} |
| +} |
| + |
| +func (b *ownerBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) { |
| + l, err := b.fakeBackend.Acquire(ctx, rt) |
| + if b.acquired != nil { |
| + close(b.acquired) |
| + <-b.proceed |
| + } |
| + return l, err |
| +} |
| +func (b *ownerBackend) StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error) { |
| + if b.child != nil { |
| + return b.child, "", b.failure |
| + } |
| + return b.fakeBackend.StartWorker(ctx, rt, env) |
| +} |
| + |
| +func TestReplicationOwnerUsesRefreshedRoute(t *testing.T) { |
| + for _, hibernate := range []bool{false, true} { |
| + t.Run(map[bool]string{false: "root", true: "embedded"}[hibernate], func(t *testing.T) { |
| + a, b, _, reg := testActivator(t, time.Hour) |
| + old := testWorkerRoute("owner.example") |
| + old.Hibernate = !hibernate |
| + fresh := old |
| + fresh.Hibernate = hibernate |
| + if err := reg.Upsert(fresh); err != nil { |
| + t.Fatal(err) |
| + } |
| + a.Routes = reg |
| + if err := a.WakeWorker(old); err != nil { |
| + t.Fatal(err) |
| + } |
| + b.mu.Lock() |
| + acq, restore, repl := b.acquires, b.restores, len(b.syncIntervals) |
| + b.mu.Unlock() |
| + want := 0 |
| + if hibernate { |
| + want = 1 |
| + } |
| + if acq != 1 || restore != want || repl != want { |
| + t.Fatalf("acquire/restore/replicate=%d/%d/%d", acq, restore, repl) |
| + } |
| + a.mu.Lock() |
| + e := a.entries[old.Host] |
| + mode, held := e.replicates, e.lease |
| + a.mu.Unlock() |
| + if mode != hibernate || (held != nil) == hibernate { |
| + t.Fatal("wrong health owner") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestReplicationOwnerRefusesSeedBeforeAcquire(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + rt := alwaysOnWorkerRoute("seed-owner.example") |
| + rt.Seed = "[\"seed\"]" |
| + if err := a.WakeWorker(rt); !errors.Is(err, os.ErrInvalid) { |
| + t.Fatalf("wake=%v", err) |
| + } |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + if b.acquires != 0 || b.seeds != 0 || b.workerStartN != 0 { |
| + t.Fatalf("side effects: acquire=%d seed=%d start=%d", b.acquires, b.seeds, b.workerStartN) |
| + } |
| +} |
| + |
| +func TestReplicationOwnerHealthHasNoFallback(t *testing.T) { |
| + l := &ownerLease{} |
| + for _, e := range []*actEntry{{replicates: true, lease: l}, {replicates: false, rep: &maintenanceRep{}}, {replicates: true}, {}} { |
| + if !errors.Is(entryHealthErr(e), os.ErrInvalid) { |
| + t.Fatal("missing selected health accepted") |
| + } |
| + } |
| + if l.health.Load() != 0 { |
| + t.Fatal("fallback read spent lease") |
| + } |
| + if err := entryHealthErr(&actEntry{replicates: true, rep: &maintenanceRep{}, lease: l}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if l.health.Load() != 0 { |
| + t.Fatal("embedded health read spent lease") |
| + } |
| + if !errors.Is((unitLease{}).Err(), os.ErrInvalid) { |
| + t.Fatal("unit lease pretends healthy") |
| + } |
| +} |
| + |
| +func TestReplicationOwnerLeaseLossStopsBeforeRelease(t *testing.T) { |
| + for _, wake := range []bool{false, true} { |
| + t.Run(map[bool]string{false: "sweep", true: "wake"}[wake], func(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + rt := alwaysOnWorkerRoute("lost-owner.example") |
| + if err := a.WakeWorker(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + gate := make(chan struct{}) |
| + p := &maintenanceProc{gate: gate} |
| + a.mu.Lock() |
| + a.entries[rt.Host].proc = p |
| + a.mu.Unlock() |
| + b.lastRep().setErr(os.ErrInvalid) |
| + done := make(chan error, 1) |
| + if wake { |
| + go func() { done <- a.WakeWorker(rt) }() |
| + } else { |
| + a.sweepOnce() |
| + } |
| + waitFor(t, "stop entered", func() bool { return p.calls.Load() > 0 }) |
| + if b.releaseCount() != 0 { |
| + t.Fatal("released before stop") |
| + } |
| + close(gate) |
| + if wake { |
| + select { |
| + case err := <-done: |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + case <-time.After(3 * time.Second): |
| + t.Fatal("wake blocked") |
| + } |
| + } else { |
| + waitFor(t, "entry stopped", func() bool { return a.entryCount() == 0 }) |
| + } |
| + if b.releaseCount() != 1 || b.lastRep().closeCount() != 0 { |
| + t.Fatal("wrong release path") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestReplicationOwnerUnprovenStopRetainsLease(t *testing.T) { |
| + for _, stopErr := range []error{nil, os.ErrPermission, ErrStillRunning} { |
| + t.Run(stringErr(stopErr), func(t *testing.T) { |
| + a, b, clk, _ := testActivator(t, time.Hour) |
| + rt := alwaysOnWorkerRoute("stuck-owner.example") |
| + if err := a.WakeWorker(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + p := &maintenanceProc{remains: true, err: stopErr} |
| + a.mu.Lock() |
| + e := a.entries[rt.Host] |
| + e.proc = p |
| + a.mu.Unlock() |
| + a.Bounce(rt.Host) |
| + waitFor(t, "pending stop", func() bool { a.mu.Lock(); defer a.mu.Unlock(); return e.pendingStop }) |
| + if b.releaseCount() != 0 || a.entryCount() != 1 { |
| + t.Fatal("unproven child lost ownership") |
| + } |
| + clk.advance(actStopTimeout + time.Second) |
| + a.sweepOnce() |
| + waitFor(t, "stop retry", func() bool { return p.calls.Load() >= 2 }) |
| + p.exited.Store(true) |
| + waitFor(t, "release", func() bool { a.sweepOnce(); return a.entryCount() == 0 }) |
| + if b.releaseCount() != 1 || b.lastRep().closeCount() != 0 { |
| + t.Fatal("wrong terminal owner") |
| + } |
| + if _, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); !errors.Is(err, ErrStopUnproven) { |
| + t.Fatalf("invented replica proof: %v", err) |
| + } |
| + }) |
| + } |
| +} |
| +func stringErr(err error) string { |
| + if err == nil { |
| + return "nil" |
| + } |
| + return err.Error() |
| +} |
| + |
| +func TestReplicationOwnerFailedWakeRetainsReturnedChild(t *testing.T) { |
| + a, b, clk, _ := testActivator(t, time.Hour) |
| + rt := alwaysOnWorkerRoute("failed-owner.example") |
| + p := &maintenanceProc{remains: true, err: os.ErrPermission} |
| + a.backend = &ownerBackend{fakeBackend: b, child: p, failure: os.ErrInvalid} |
| + if err := a.WakeWorker(rt); !errors.Is(err, os.ErrInvalid) { |
| + t.Fatalf("wake=%v", err) |
| + } |
| + a.mu.Lock() |
| + e := a.entries[rt.Host] |
| + pending := e != nil && e.pendingStop && e.state == actStopping |
| + a.mu.Unlock() |
| + if !pending || b.releaseCount() != 0 { |
| + t.Fatal("failed wake orphaned child") |
| + } |
| + done := make(chan error, 1) |
| + go func() { done <- a.WakeWorker(rt) }() |
| + select { |
| + case err := <-done: |
| + t.Fatalf("retry passed stopping child: %v", err) |
| + case <-time.After(30 * time.Millisecond): |
| + } |
| + clk.advance(actStopTimeout + time.Second) |
| + a.sweepOnce() |
| + waitFor(t, "cleanup retry", func() bool { return p.calls.Load() >= 2 }) |
| + a.mu.Lock() |
| + a.shuttingDown = true |
| + a.mu.Unlock() |
| + p.exited.Store(true) |
| + waitFor(t, "release", func() bool { a.sweepOnce(); return a.entryCount() == 0 }) |
| + select { |
| + case <-done: |
| + case <-time.After(time.Second): |
| + t.Fatal("retry did not join stop") |
| + } |
| + if b.releaseCount() != 1 || b.lastRep().closeCount() != 0 { |
| + t.Fatal("wrong cleanup owner") |
| + } |
| +} |
| + |
| +func TestReplicationOwnerPreservesInstancePath(t *testing.T) { |
| + for _, backing := range []string{"", registry.BackingUnit} { |
| + for _, hibernate := range []bool{false, true} { |
| + t.Run(backing+map[bool]string{false: "always", true: "hibernate"}[hibernate], func(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + rt := testRoute("instance-owner.example") |
| + rt.Backing = backing |
| + rt.Hibernate = hibernate |
| + arrangeAwake(t, a, rt) |
| + b.mu.Lock() |
| + acq, restore, repl := b.acquires, b.restores, len(b.syncIntervals) |
| + b.mu.Unlock() |
| + if acq != 1 || restore != 1 || repl != 1 { |
| + t.Fatalf("instance path changed: %d/%d/%d", acq, restore, repl) |
| + } |
| + if !a.Bounce(rt.Host) { |
| + t.Fatal("missing instance") |
| + } |
| + waitFor(t, "instance stopped", func() bool { return a.entryCount() == 0 }) |
| + if b.lastRep().closeCount() != 1 { |
| + t.Fatal("instance replica not closed") |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +func TestReplicationOwnerPreSpawnFailureReleasesLease(t *testing.T) { |
| + for _, failure := range []string{"start", "token", "fence"} { |
| + t.Run(failure, func(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + rt := alwaysOnWorkerRoute("pre-owner.example") |
| + switch failure { |
| + case "start": |
| + b.workerErr = os.ErrInvalid |
| + case "token": |
| + old := randRead |
| + randRead = func([]byte) (int, error) { return 0, os.ErrInvalid } |
| + defer func() { randRead = old }() |
| + case "fence": |
| + acquired, proceed := make(chan struct{}), make(chan struct{}) |
| + a.backend = &ownerBackend{fakeBackend: b, acquired: acquired, proceed: proceed} |
| + go func() { |
| + <-acquired |
| + a.mu.Lock() |
| + if a.fences == nil { |
| + a.fences = map[string]uint64{} |
| + } |
| + a.fences[rt.Host] = 1 |
| + a.mu.Unlock() |
| + close(proceed) |
| + }() |
| + } |
| + if err := a.WakeWorker(rt); err == nil { |
| + t.Fatal("failure became success") |
| + } |
| + if b.releaseCount() != 1 || b.lastRep().closeCount() != 0 || a.entryCount() != 0 { |
| + t.Fatal("failed pre-spawn owner not released") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestReplicationOwnerDeadDuringSettleReleasesLease(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + p := &maintenanceProc{} |
| + p.exited.Store(true) |
| + a.backend = &ownerBackend{fakeBackend: b, child: p} |
| + if err := a.WakeWorker(alwaysOnWorkerRoute("settle-owner.example")); err == nil { |
| + t.Fatal("dead child became awake") |
| + } |
| + if b.releaseCount() != 1 || b.lastRep().closeCount() != 0 || a.entryCount() != 0 { |
| + t.Fatal("dead child retained or lost wrong owner") |
| + } |
| +} |
| + |
| +func TestReplicationOwnerMissingProcessCannotRelease(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + rt := alwaysOnWorkerRoute("missing-owner.example") |
| + if err := a.WakeWorker(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + a.mu.Lock() |
| + e := a.entries[rt.Host] |
| + p := e.proc |
| + e.proc = nil |
| + a.mu.Unlock() |
| + a.Bounce(rt.Host) |
| + waitFor(t, "pending missing child", func() bool { a.mu.Lock(); defer a.mu.Unlock(); return e.pendingStop }) |
| + if b.releaseCount() != 0 || a.entryCount() != 1 { |
| + t.Fatal("missing child became stop proof") |
| + } |
| + a.mu.Lock() |
| + e.proc = p |
| + a.mu.Unlock() |
| + if err := p.Stop(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + waitFor(t, "retained owner cleanup", func() bool { a.sweepOnce(); return a.entryCount() == 0 }) |
| + if b.releaseCount() != 1 { |
| + t.Fatal("retained owner not released") |
| + } |
| +} |
diff --git a/internal/host/schedule_wake_test.go b/internal/host/schedule_wake_test.go| index 4cd0d4109..059e022db 100644 |
| --- a/internal/host/schedule_wake_test.go |
| +++ b/internal/host/schedule_wake_test.go |
| @@ -612,3 +612,5 @@ func TestDeliveryStillCarriesScheduleAtHeader(t *testing.T) { |
| t.Fatal("the handler never ran") |
| } |
| } |
| + |
| +func (nopLease) Err() error { return nil } |
diff --git a/internal/ops/hibernate_verify.go b/internal/ops/hibernate_verify.go| index 7073f3260..fdf386408 100644 |
| --- a/internal/ops/hibernate_verify.go |
| +++ b/internal/ops/hibernate_verify.go |
| @@ -55,7 +55,7 @@ func HibernatingHosts(reg *registry.Registry) ([]HibernatingRoute, error) { |
| if r.Backing == registry.BackingUnit { |
| continue |
| } |
| - if r.Hibernate && r.DBPath != "" { |
| + if r.ActivatorOwnsReplication() && r.DBPath != "" { |
| hosts = append(hosts, HibernatingRoute{Host: r.Host, DBPath: r.DBPath}) |
| } |
| } |
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go| index 6311e9847..30892662e 100644 |
| --- a/internal/ops/litestream.go |
| +++ b/internal/ops/litestream.go |
| @@ -524,7 +524,7 @@ func fleetDBsFromRoutes(routes []registry.Route, registryPath, dataDir string, t |
| } |
| continue |
| } |
| - if r.Hibernate && r.DBPath != "" { |
| + if r.ActivatorOwnsReplication() && r.DBPath != "" { |
| hibernating[filepath.Clean(r.DBPath)] = true |
| continue |
| } |
| @@ -872,7 +872,7 @@ func liveClaim(routes []registry.Route, p string) string { |
| // hibernates (the activator replicates exec routes' hibernating |
| // databases, not units'), so only a hibernating exec route is no |
| // live claim. |
| - if r.DBPath == "" || (r.Hibernate && r.Backing != registry.BackingUnit) { |
| + if r.DBPath == "" || r.ActivatorOwnsReplication() { |
| continue |
| } |
| claim := filepath.Clean(r.DBPath) |
diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go| index ce80eda71..7d56a36ba 100644 |
| --- a/internal/ops/ops_test.go |
| +++ b/internal/ops/ops_test.go |
| @@ -85,6 +85,69 @@ func TestConfigGenSkipsHibernatingRoutes(t *testing.T) { |
| } |
| } |
| +// An explicitly always-on native sidecar is started by the activator, but its |
| +// database belongs to the box-wide replicator (sidecar design §15.3). This |
| +// pins both selectors against the same real registry row and database file: |
| +// the overlap is intentional for process ownership, but must not become an |
| +// overlap in replication ownership inside the activator wake path. |
| +func TestAlwaysOnSidecarProcessAndFleetSelectors(t *testing.T) { |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { _ = reg.Close() }) |
| + |
| + hibernate := false |
| + manifest := provision.SidecarManifest{Hibernate: &hibernate} |
| + if manifest.HibernateOrDefault() { |
| + t.Fatal("an explicit hibernate:false sidecar was treated as hibernating") |
| + } |
| + |
| + host := "worker.app.bab.sidecar.internal" |
| + dbPath := provision.DefaultDBPath(dir, host) |
| + db, err := sql.Open("sqlite", dbPath) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := db.Exec(`CREATE TABLE state (value TEXT)`); err != nil { |
| + _ = db.Close() |
| + t.Fatal(err) |
| + } |
| + if err := db.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + rt := registry.Route{ |
| + Host: host, Kind: provision.SidecarKind, Account: "bab", App: "app", |
| + DBPath: dbPath, Hibernate: manifest.HibernateOrDefault(), |
| + Provisioned: true, Status: "ready", |
| + } |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + workers, err := reg.AlwaysOnWorkers() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(workers) != 1 || workers[0].Host != host { |
| + t.Fatalf("AlwaysOnWorkers = %+v, want %s", workers, host) |
| + } |
| + dbs, err := FleetDBs(reg, dir) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Contains(dbs, dbPath) { |
| + t.Fatalf("FleetDBs = %v, want always-on sidecar database %s", dbs, dbPath) |
| + } |
| + if got := liveClaim([]registry.Route{rt}, dbPath); got != host { |
| + t.Fatalf("liveClaim = %q, want %q", got, host) |
| + } |
| + if got := hostsOf(mustHibernating(t, reg)); slices.Contains(got, host) { |
| + t.Fatalf("HibernatingHosts claimed box-wide-owned always-on sidecar: %v", got) |
| + } |
| +} |
| + |
| // TestFleetDBsDiscoversNestedDatabases pins the isolated-exec companion to |
| // the flat glob above (design §7): a route's DBPath being "claimed" only |
| // FILTERS what the glob already found. For a non-unit route it never adds |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index 29dfd932b..0c5713fa7 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -2491,3 +2491,5 @@ func (r *Registry) SnapshotTo(path string) error { |
| _, err = db.Exec("VACUUM INTO ?", path) |
| return err |
| } |
| + |
| +func (t Route) ActivatorOwnsReplication() bool { return t.Hibernate && t.Backing != BackingUnit } |
diff --git a/internal/registry/replication_owner_test.go b/internal/registry/replication_owner_test.go| new file mode 100644 |
| index 000000000..dbc364222 |
| --- /dev/null |
| +++ b/internal/registry/replication_owner_test.go |
| @@ -0,0 +1,16 @@ |
| +package registry |
| + |
| +import "testing" |
| + |
| +func TestActivatorReplicationOwnerIgnoresInventoryMetadata(t *testing.T) { |
| + for _, hibernate := range []bool{false, true} { |
| + for _, backing := range []string{"", BackingUnit, "exec"} { |
| + for _, metadata := range []Route{{}, {Kind: "sidecar", Status: "ready", Account: "bab", App: "worker", DBPath: "/missing.db"}, {Kind: "static", Status: "pending", DBPath: ""}} { |
| + metadata.Hibernate, metadata.Backing = hibernate, backing |
| + if got, want := metadata.ActivatorOwnsReplication(), hibernate && backing != BackingUnit; got != want { |
| + t.Fatalf("route=%+v owner=%v want=%v", metadata, got, want) |
| + } |
| + } |
| + } |
| + } |
| +} |