amadan

carlos / platform Public

zdd-5 task 11: socketless attempts on the same entry (ReplaceWorker)

ReplaceWorker runs a worker's attempt through respawnWorkerLocked. The
attempt's checks are ANDed into main's ownership check at all three
points (before the preparation, at the exec, before the install), and
plan 3's termination proof (retired) stays. Only a supersession restores
the replaced build. A worker's serving identity is published through one
incarnation-ordered writer (OnWorkerServing) whatever the flag. A wake
publishes after afterWake, and a fallback's publication is held for
PublishHeldServing.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0d3469a44d39aa2d51a586221541825a51268143 parent 3baca10
3 files changed, +1083 −8
  • internal/activator/activator.go +130 −8
  • internal/activator/workerswap.go +307 −0
  • internal/activator/workerswap_test.go +646 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index ad620c19a..1b3370b6e 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -565,6 +565,37 @@ type actEntry struct {
// not release the lease while a spawn against the same DB is in flight.
workerRespawning bool
respawnDone chan struct{}
+
+ // The socketless attempt's state (zero-downtime plan 5 Task 11,
+ // ReplaceWorker). attemptBuilds names the builds an attempt in flight may
+ // exec (its own and its fallback), so the generations file pins them
+ // before anything execs them (Task 12); startWorker's pending generation
+ // already publishes the exact build about to exec, so this is for the
+ // fallback above all. retireTook is how long the last respawn took to
+ // retire its old child: the "swapped" event's drain time for a worker
+ // (spec §9.4).
+ //
+ // attemptGate is an attempt's §7.4 checks (swapChecksLocked with the
+ // attempt's revision), and holdServing marks a ReplaceWorker fallback
+ // whose publication waits for the scheduler's failure record (spec
+ // §8.1). Both arm the NEXT respawnWorkerLocked only: it takes them when
+ // it is called, so a later respawn (a restore, a fallback, a crash
+ // restart) never inherits them, and a ReplaceWorker that stops waiting
+ // cannot disarm a respawn still running. attemptVeto is that respawn's
+ // answer when the gate refused, read and cleared by ReplaceWorker.
+ // heldServing is the publication holdServing kept back, delivered by
+ // PublishHeldServing. workerInc is the worker child's incarnation,
+ // bumped at every install (a wake, a respawn): an attempt counts only if
+ // it moved, and a publication is delivered only for the latest. All
+ // written under a.mu.
+ attemptBuilds []string
+ retireTook time.Duration
+ attemptGate func() error
+ attemptVeto error
+ holdServing bool
+ heldServing *heldServing
+ workerInc uint64
+
// shed is a respawn child, refused at its installation with generations
// off, whose exit is not proven yet (final review R2-1). With
// generations on the operation set holds such a child as a member. The
@@ -970,6 +1001,19 @@ type Service struct {
// Service. Set before Start, like OnWake. Plan 5's scheduler fills it.
CoalesceRecycle func(host string) bool
+ // OnWorkerServing, when set, hears a worker's captured serving identity
+ // each time a wake or a respawn installs a child (zero-downtime spec
+ // §7.9), an empty version included. It runs with no lock held, through
+ // publishWorkerServing, which delivers only the entry's latest
+ // incarnation. The agent sets it whatever CARLOS_SWAP says, so a
+ // flag-off box keeps `carlos sidecars ls`'s running version right
+ // (registry ServingVersion). Set before Start, like OnWake.
+ OnWorkerServing func(host string, id release.StagedIdentity)
+ // servingMu orders OnWorkerServing deliveries: the incarnation check and
+ // the delivery happen under it, so an older publication can never land
+ // after a newer one. Taken BEFORE a.mu, never while holding it.
+ servingMu sync.Mutex
+
// GenerationsFile, when set, is where the activator publishes every
// worker's captured serving build for root's pin set
// (release.GenerationsPath of the data dir root's passes use as BakeDir;
@@ -1323,6 +1367,11 @@ func (a *Service) wake(e *actEntry) {
emitEvent(a.Logs, e.route, "info", wakeEventMessage(w, e.worker))
}
+ var (
+ servingID release.StagedIdentity
+ servingInc uint64
+ publishServing bool
+ )
a.mu.Lock()
if err != nil {
e.err = err
@@ -1353,6 +1402,12 @@ func (a *Service) wake(e *actEntry) {
// cycle. Fresh wake, so it can't be stale yet.
e.spawnVersion = w.spawnVersion
e.workerServing = w.spawnIdentity
+ if e.worker {
+ // A new worker incarnation (spec §7.9): its identity is published
+ // once the wake's outcome has been reported (below).
+ e.workerInc++
+ servingID, servingInc, publishServing = e.workerServing, e.workerInc, true
+ }
e.staleSince = time.Time{}
// The generation is current from here (spec §7.1). A wake the
// claim path ran (doClaimWake) spawned through its provider, not
@@ -1419,6 +1474,14 @@ func (a *Service) wake(e *actEntry) {
// A planned wake's outcome, for the scheduler: off a.mu, since OnGen
// reaches its goodMu, which calls back into Current.
a.afterWake(e, w, err)
+ if publishServing {
+ // A worker's identity, after afterWake and never before it (spec
+ // §8.1; P5 review finding 7): for a fallback wake, afterWake's
+ // GenFellBack has the scheduler record B's failure synchronously,
+ // so B is held before the fallback's identity is published, and a
+ // publication that blocks cannot hold that record up.
+ a.publishWorkerServing(e, servingID, servingInc)
+ }
if err != nil && e.worker {
// The failed wake's entry is gone, and with it any build its spawn
// published (generations.go).
@@ -2528,6 +2591,26 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.workerRespawning = true
e.respawnDone = make(chan struct{})
done, rt, host := e.respawnDone, e.route, e.host
+ // A socketless attempt's checks and a fallback's held publication arm
+ // this respawn only (ReplaceWorker sets them just before the call): taken
+ // here, under the caller's a.mu, so no later respawn inherits them.
+ attemptGate, hold := e.attemptGate, e.holdServing
+ e.attemptGate, e.holdServing = nil, false
+ // check is the ownership check every exec asks, whatever the flag (spec
+ // §7.4), ANDed with an attempt's own checks when this respawn execs an
+ // attempt's build (spec §6.3, §8.1): a newer release, or a halted
+ // cohort, whose halt bumps the revision, that lands while the old child
+ // drains or the build is prepared means the attempt's build must not
+ // run. The caller holds a.mu.
+ check := func() error {
+ if err := a.ownedLocked(e, false); err != nil {
+ return err
+ }
+ if attemptGate != nil {
+ return attemptGate()
+ }
+ return nil
+ }
// With generations on, the respawn is a member of the operation set
// until it has finished with every child it made: a stop's barrier joins
// it, and cancelling it tears down the exec (zero-downtime plan 3,
@@ -2571,12 +2654,17 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
// lease from a sibling that could run it.
refused error
)
+ // The old child's stop, timed on the wall clock: a socketless
+ // attempt reports it as its drain (spec §9.4).
+ var retireTook time.Duration
if stopOld != nil {
+ retireStart := time.Now()
stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
if serr := stopOld.Stop(stopCtx); serr != nil {
log.Printf("activator: worker %s respawn: retiring the old child: %v", host, serr)
}
cancel()
+ retireTook = time.Since(retireStart)
}
// The retired child must be proven gone, its host cgroup included,
@@ -2605,9 +2693,11 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
// ownership check (spec §7.4): every exec asks it, whatever the
// switch says. It covers the entry still being the host's, awake
// and not shutting down, and adds the lease still held and no
- // fence.
+ // fence. An attempt's own checks are ANDed in (check), and plan 3's
+ // termination proof (retired) is kept beside both: nothing execs
+ // beside an old child that is not proven gone.
a.mu.Lock()
- ownErr := a.ownedLocked(e, false)
+ ownErr := check()
a.mu.Unlock()
owned := ownErr == nil
wanted := owned && retired
@@ -2622,14 +2712,16 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
// build's publication all come between the check above and the
// exec, and the lease can be lost, a fence set or the entry
// stopped in that time. startWorker composes it into the
- // before-exec hook, after the publication.
+ // before-exec hook, after the publication. An attempt's checks
+ // ride on it (check): a supersession in that time refuses the
+ // exec too.
gate := func() error {
if err := ctx.Err(); err != nil {
return err
}
a.mu.Lock()
defer a.mu.Unlock()
- return a.ownedLocked(e, false)
+ return check()
}
if g, ok := a.backend.(configAdmitter); ok {
refused = g.AdmitConfig(ctx, rt)
@@ -2648,11 +2740,26 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
// And once more before the child is installed: ownership lost after
// the exec sheds the child (final review F1). ownedLocked covers what
// the old liveness check did (the entry is the host's, awake, not
- // shutting down) and adds the lease and the fences.
+ // shutting down) and adds the lease and the fences; an attempt's
+ // checks ride on it (check), so an attempt superseded during its exec
+ // sheds its child rather than install it.
if owned {
- ownErr = a.ownedLocked(e, false)
+ ownErr = check()
}
live := ownErr == nil
+ if attemptGate != nil {
+ // The attempt's answer for ReplaceWorker: which gate refused, at
+ // the first check, at the exec, or before the install. Nil when
+ // nothing refused (an unproven old child, a failed exec, a
+ // config refusal or an install), which ReplaceWorker reads from
+ // the incarnation instead.
+ switch {
+ case ownErr != nil:
+ e.attemptVeto = ownErr
+ case refused == nil && verdictOf(err) != gateProceed:
+ e.attemptVeto = err
+ }
+ }
switch {
case owned && !retired && live:
a.mu.Unlock()
@@ -2727,8 +2834,10 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
case verdictOf(err) != gateProceed:
// The exec's own ownership check refused (startWorker's gate),
// and ownership came back before the install check (a fence
- // lifted): nothing exec'd, and this is no failure of the build.
- // The dead child stays, and the sweep's crash restart asks again.
+ // lifted), or an attempt's checks refused there: nothing exec'd,
+ // and this is no failure of the build. The dead child stays, and
+ // the sweep's crash restart asks again (an attempt's restore is
+ // ReplaceWorker's, from attemptVeto).
a.mu.Unlock()
log.Printf("activator: worker %s NOT respawned (%s): refused at its exec: %v", host, why, err)
case err != nil:
@@ -2747,6 +2856,16 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.spawnVersion = ident.Version
e.workerServing = ident
e.workerVersion = "" // the new child has not reported yet
+ e.retireTook = retireTook
+ // A new incarnation, published below (spec §7.9), unless this is
+ // a ReplaceWorker fallback: its publication is held until the
+ // scheduler has recorded the failure it falls back from (spec
+ // §8.1), and PublishHeldServing delivers it.
+ e.workerInc++
+ inc := e.workerInc
+ if hold {
+ e.heldServing = &heldServing{id: ident, inc: inc}
+ }
if m != nil {
e.gen = Gen{N: m.gen.N, Build: ident.Version, Worker: true}
}
@@ -2765,6 +2884,9 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
a.mu.Unlock()
// The captured build may have changed (a converge): publish it.
_ = a.persistGenerations()
+ if !hold {
+ a.publishWorkerServing(e, ident, inc)
+ }
log.Printf("activator: worker %s respawned (%s), now running %s", host, why, ident.Version)
emitEvent(a.Logs, rt, "info", "sidecar restarted")
}
diff --git a/internal/activator/workerswap.go b/internal/activator/workerswap.go
new file mode 100644
index 000000000..2eef2820e
--- /dev/null
+++ b/internal/activator/workerswap.go
@@ -0,0 +1,307 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ReplaceWorker runs one socketless attempt (zero-downtime spec §6.5, §7.9)
+// on an awake worker or native sidecar member: stop-then-start on the SAME
+// entry, through respawnWorkerLocked, so the lease and the replicator stay
+// put. The new child must still be running firstWakeSettle (10 s) after it
+// started; a clean exit inside that window is a one-shot's success. On a
+// failure:
+// - restart or config: the build it replaced is respawned (spec §6.5);
+// - release, install or retry in hold mode: nothing falls back, and the
+// entry stops, so the instance stays down (spec §6.4, §8.3);
+// - release, install or retry in overlap mode: fallback, the host's good
+// build, is respawned in its place; with none, the entry stops, as a
+// failed wake does today (spec §8.2).
+//
+// Every respawn here asks plan 3's ownership check at its exec, and keeps
+// its termination proof (respawnWorkerLocked). The attempt's own respawn
+// also asks the attempt's §7.4 checks there, at all three of its points
+// (before the preparation, at the exec, before the install). A refusal
+// that is a supersession brings back the build the attempt replaced; any
+// other refusal (no ownership) spawns nothing at all (spec §7.4).
+//
+// The scheduler runs the cohort-failure transition from the result, then
+// calls PublishHeldServing for a fallback's identity. newest is called with
+// a.mu held, as Swap's is.
+func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) SwapResult {
+ if newest == nil {
+ return abandon(errSwapSuperseded)
+ }
+ if _, ok := a.backend.(stagedWorkerStarter); !ok {
+ // A backend that cannot exec a named build would run whatever the
+ // live link names: nothing of the attempt's choosing (as a planned
+ // worker wake is refused, spawnPlannedWorker).
+ return abandon(errSwapNoGens)
+ }
+ onto := release.StagedIdentity{Version: at.Build, StagedPath: at.StagedPath, Digest: at.Digest, Demand: at.Demand}
+ a.mu.Lock()
+ e := a.entries[at.Host]
+ if err := a.swapChecksLocked(e, at, newest); err != nil {
+ a.mu.Unlock()
+ return abandon(err)
+ }
+ if !e.worker {
+ a.mu.Unlock()
+ return abandon(errSwapKind)
+ }
+ if e.workerRespawning || e.firstWatch != nil {
+ a.mu.Unlock()
+ return abandon(errSwapBusy)
+ }
+ prev := e.workerServing
+ // The attempt's build and its fallback, for the generations file (Task
+ // 12). startWorker's pending generation already names the build about
+ // to exec at each spawn; listing the attempt's own here too is harmless.
+ e.attemptBuilds = []string{onto.Version}
+ if fallback != nil {
+ e.attemptBuilds = append(e.attemptBuilds, fallback.Version)
+ }
+ // The same checks, carried into the respawn: it runs them again before
+ // its preparation, at the exec and before the install, after the old
+ // child has retired (spec §6.3, §7.4, §8.1).
+ e.attemptGate = func() error { return a.swapChecksLocked(e, at, newest) }
+ e.attemptVeto = nil
+ // Spec §9.4: the bounce event names the attempt's cause.
+ emitEvent(a.Logs, e.route, "info", bounceText(true, at.Cause))
+ // The incarnation before the respawn: the attempt's child counts only if
+ // the respawn installed a NEW one (P5 review finding 5). A failed exec
+ // keeps the old e.proc and its identity, which for a restart or a config
+ // attempt is the same artifact, so the artifact alone passed it.
+ inc0 := e.workerInc
+ a.respawnWorkerLocked(e, e.proc, "attempt ("+string(at.Cause)+")", &onto)
+ done := e.respawnDone
+ a.mu.Unlock()
+ defer func() {
+ a.mu.Lock()
+ e.attemptBuilds = nil
+ a.mu.Unlock()
+ }()
+
+ if !a.awaitRespawn(ctx, done) {
+ return abandon(errSwapStopped)
+ }
+ a.mu.Lock()
+ veto := e.attemptVeto
+ e.attemptVeto = nil
+ if a.entries[at.Host] != e || e.state != actAwake {
+ a.mu.Unlock()
+ return abandon(errSwapStopped)
+ }
+ if veto != nil {
+ // The respawn's re-check refused the attempt's build: nothing of it
+ // runs. Only a supersession brings back the build it replaced (index
+ // P5 #1); without ownership (a lost lease, a fence, shutdown) nothing
+ // spawns, a restore included, and the entry stops through the
+ // sweep's lease-loss path.
+ if verdictOf(veto) == gateSuperseded {
+ a.restoreLocked(e, prev)
+ }
+ a.mu.Unlock()
+ return abandon(fmt.Errorf("re-checked at the spawn: %w", veto))
+ }
+ if e.workerInc == inc0 || !sameArtifact(e.workerServing, onto) {
+ // Nothing of the attempt's build ran: no new incarnation was
+ // installed (the exec failed, its build could not be published
+ // first, the old child is not proven gone), or the backend exec'd
+ // other bytes. Put the build it replaced back, as a failed restart
+ // does (spec §6.5), through a respawn that asks ownership and the
+ // termination proof at its own exec. This is a staging refusal,
+ // never a failure of the build, and never a success: a restart or
+ // config attempt whose child never ran acknowledges nothing (spec
+ // §8.1).
+ a.restoreLocked(e, prev)
+ a.mu.Unlock()
+ return abandon(fmt.Errorf("staging: worker not respawned onto %s", onto.Version))
+ }
+ // The observation. firstWatch hides the worker from Current and lets an
+ // intentional stop (beginStopLocked) cancel it. unexpectedExit is
+ // cleared here, for this child: the arms that set it never clear it, and
+ // a mark left by an earlier crash would make beginStopLocked leave an
+ // intentional stop to be read as this build's exit.
+ obsCtx, cancel := context.WithCancel(ctx)
+ e.firstWatch, e.unexpectedExit = cancel, false
+ proc, drain, inc1 := e.proc, e.retireTook, e.workerInc
+ a.mu.Unlock()
+
+ r := observeSettle(obsCtx, proc, firstWakeSettle)
+ cancel()
+
+ a.mu.Lock()
+ e.firstWatch = nil
+ unexpected := e.unexpectedExit
+ // mine: the entry, awake, still running the attempt's child. With the
+ // scheduler's CoalesceRecycle nothing recycles it meanwhile; without it
+ // (or after a stop), the attempt does not touch what replaced it.
+ mine := a.entries[at.Host] == e && e.state == actAwake && e.workerInc == inc1 && !e.workerRespawning
+ if r == settleOK || (r == settleExited && !procCrashed(proc)) {
+ // The checks once more before the result counts (spec §6.3, §8.1):
+ // a newer release or a halted cohort since the spawn means this
+ // build must not be recorded good. Its child stays until the newer
+ // attempt replaces it. A one-shot that finished may have been parked
+ // meanwhile: no longer awake, but still not superseded.
+ err := a.swapChecksLocked(e, at, newest)
+ if r == settleExited && errors.Is(err, errSwapNotAwake) {
+ err = nil
+ if newest(at.Host) != at.SchedRev {
+ err = errSwapSuperseded
+ }
+ }
+ if err != nil {
+ a.mu.Unlock()
+ return abandon(err)
+ }
+ g := a.workerResultGen(e, onto)
+ a.mu.Unlock()
+ return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain}
+ }
+ if r == settleCancelled && !unexpected {
+ // An intentional stop cancels the observation and records nothing
+ // (spec §6.4).
+ a.mu.Unlock()
+ return abandon(errSwapStopped)
+ }
+ reason := "exited within " + firstWakeSettle.String() + " of starting"
+ adoption := at.Cause == CauseRelease || at.Cause == CauseInstall || at.Cause == CauseRetry
+ switch {
+ case !adoption:
+ if mine {
+ a.restoreLocked(e, prev)
+ }
+ a.mu.Unlock()
+ return SwapResult{Outcome: SwapFailed, Reason: reason}
+ case at.Mode == ModeHold:
+ if mine {
+ a.beginStopLocked(e)
+ }
+ a.mu.Unlock()
+ return SwapResult{Outcome: SwapDown, Reason: reason}
+ case !mine:
+ // The entry stopped, or another child replaced B, while it was
+ // observed: there is nothing here to fall back on.
+ a.mu.Unlock()
+ return SwapResult{Outcome: SwapFailed, Reason: reason}
+ case fallback == nil:
+ a.beginStopLocked(e)
+ a.mu.Unlock()
+ return SwapResult{Outcome: SwapFailed, Reason: reason + "; no good build to fall back to"}
+ }
+ // The fallback's publication waits for the failure transition (spec
+ // §8.1; P5 review finding 7): the respawn stores it (holdServing), and
+ // the scheduler delivers it with PublishHeldServing once done() has
+ // recorded B's failure. A synchronous publication here would sit in
+ // front of that record, and a blocked callback would leave B unheld.
+ // Like every respawn, the fallback asks ownership and the termination
+ // proof at its exec: a lost lease spawns nothing (index P6 new#1).
+ e.holdServing = true
+ inc2 := e.workerInc
+ a.respawnWorkerLocked(e, e.proc, "falling back to "+fallback.Version, fallback)
+ done = e.respawnDone
+ a.mu.Unlock()
+ if !a.awaitRespawn(ctx, done) {
+ return SwapResult{Outcome: SwapFailed, Reason: reason}
+ }
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if a.entries[at.Host] == e && e.workerInc != inc2 && sameArtifact(e.workerServing, *fallback) {
+ return SwapResult{Outcome: SwapFailed, Gen: a.workerResultGen(e, *fallback), Reason: reason}
+ }
+ return SwapResult{Outcome: SwapFailed, Reason: reason + "; the good build did not start either"}
+}
+
+// restoreLocked respawns prev, the build an attempt replaced, on e (spec
+// §6.5), retiring whatever child e holds. Like every respawn it asks
+// ownership and the termination proof at its exec, so a host no longer
+// owned gets nothing. Nothing is done when another respawn is in flight,
+// or when prev was never captured (the sweep's crash restart then
+// respawns from the live link, as a recycle with no captured build does).
+// The caller holds a.mu.
+func (a *Service) restoreLocked(e *actEntry, prev release.StagedIdentity) {
+ if e.workerRespawning || prev.StagedPath == "" {
+ return
+ }
+ a.respawnWorkerLocked(e, e.proc, "restoring "+prev.Version, &prev)
+}
+
+// workerResultGen is the generation a socketless attempt reports: id's
+// artifact, marked Worker so the scheduler compares it by its artifact and
+// leaves its serving version to publishWorkerServing, with the entry's
+// generation number. The caller holds a.mu.
+func (a *Service) workerResultGen(e *actEntry, id release.StagedIdentity) Gen {
+ g := workerGenOf(id)
+ g.N = e.gen.N
+ return g
+}
+
+// PublishHeldServing delivers a worker's publication that a fallback held
+// back (holdServing) until the caller had recorded the failure it falls
+// back from (spec §8.1). Incarnation-ordered like any other: a newer install
+// since makes it a no-op. Called with no lock held.
+func (a *Service) PublishHeldServing(host string) {
+ a.mu.Lock()
+ e := a.entries[host]
+ var held *heldServing
+ if e != nil {
+ held, e.heldServing = e.heldServing, nil
+ }
+ a.mu.Unlock()
+ if held != nil {
+ a.publishWorkerServing(e, held.id, held.inc)
+ }
+}
+
+// heldServing is a worker publication held for PublishHeldServing.
+type heldServing struct {
+ id release.StagedIdentity
+ inc uint64
+}
+
+// publishWorkerServing delivers a worker's captured identity for
+// incarnation inc of entry e (zero-downtime spec §7.9). It is delivered only
+// while e is the host's entry and inc its current incarnation, checked and
+// delivered under servingMu, so an older publication can never land after a
+// newer one. Called with no lock held.
+//
+// An empty version is published too (P5 review finding 8): a child exec'd
+// from an external live-link target has no version to capture, and the
+// row's previous ServingVersion must be cleared, not left naming a build
+// that no longer runs.
+func (a *Service) publishWorkerServing(e *actEntry, id release.StagedIdentity, inc uint64) {
+ if a.OnWorkerServing == nil {
+ return
+ }
+ a.servingMu.Lock()
+ defer a.servingMu.Unlock()
+ a.mu.Lock()
+ current := a.entries[e.host] == e && e.workerInc == inc
+ a.mu.Unlock()
+ if !current {
+ return
+ }
+ a.OnWorkerServing(e.host, id)
+}
+
+// sameArtifact reports whether the backend exec'd exactly the chosen bytes:
+// the staged path and its digest (the release demand is not the backend's
+// to report).
+func sameArtifact(got, want release.StagedIdentity) bool {
+ return got.StagedPath == want.StagedPath && got.Digest == want.Digest
+}
+
+// awaitRespawn waits for a respawn's goroutine to finish with every child it
+// made (respawnDone closes last). False when ctx ended first.
+func (a *Service) awaitRespawn(ctx context.Context, done chan struct{}) bool {
+ select {
+ case <-done:
+ return true
+ case <-ctx.Done():
+ return false
+ }
+}
diff --git a/internal/activator/workerswap_test.go b/internal/activator/workerswap_test.go
new file mode 100644
index 000000000..5c022fde8
--- /dev/null
+++ b/internal/activator/workerswap_test.go
@@ -0,0 +1,646 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "slices"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func workerAttempt(cause Cause, mode SwapMode, onto release.StagedIdentity) Attempt {
+ return Attempt{Host: "bot.x.internal", Cause: cause, Build: onto.Version, StagedPath: onto.StagedPath,
+ Digest: onto.Digest, Demand: onto.Demand, Mode: mode, SchedRev: 1}
+}
+
+func one(string) uint64 { return 1 }
+
+// waitForWorkerStarts waits until the backend has made n worker children:
+// a spawn call is recorded before its exec, so b.calls() alone runs ahead
+// of the child a test wants to act on.
+func waitForWorkerStarts(t *testing.T, b *stagedFake, n int) {
+ t.Helper()
+ waitFor(t, "a worker child", func() bool { return b.workerStarts() == n })
+}
+
+// Spec §7.9: the attempt respawns on the SAME entry. The lease is never
+// released or re-taken, and the new child execs the attempt's staged build.
+func TestReplaceWorkerKeepsTheEntryAndItsLease(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ acquires, _, _ := b.counts()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ if res.Outcome != SwapDone || res.Gen.Build != "v2" || res.Gen.Digest != fakeIdentity("v2").Digest || !res.Gen.Worker {
+ t.Fatalf("ReplaceWorker = %+v", res)
+ }
+ if got, _, _ := b.counts(); got != acquires {
+ t.Fatalf("lease acquires %d -> %d: a socketless attempt must keep the lease (spec §7.9)", acquires, got)
+ }
+ if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v2") {
+ t.Fatalf("spawns %v, want the wake, then v2 by its staged path", calls)
+ }
+ if got := a.entryServing(host); got != fakeIdentity("v2") {
+ t.Fatalf("serving %+v", got)
+ }
+ if _, ok := a.Current(host); !ok {
+ t.Fatal("Current hides the worker after its attempt finished")
+ }
+}
+
+// Spec §6.5: a failed release in overlap mode respawns the good build in
+// its place, on the same entry, and reports the failure.
+func TestReplaceWorkerFallsBackToTheGoodBuild(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ good := fakeIdentity("v1")
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ time.Sleep(50 * time.Millisecond)
+ b.lastWorker().crash()
+ }()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
+ if res.Outcome != SwapFailed || res.Gen.Build != "v1" {
+ t.Fatalf("ReplaceWorker = %+v, want failed with the good v1 serving", res)
+ }
+ if calls := b.calls(); len(calls) != 3 || *calls[2] != good {
+ t.Fatalf("spawns %v, want the good build last", calls)
+ }
+ if got := a.entryServing(host); got != good {
+ t.Fatalf("serving %+v, want the good build", got)
+ }
+}
+
+// Spec §6.4, §8.3: a hold build that ran never falls back. The instance
+// stays down.
+func TestReplaceWorkerInHoldModeStaysDown(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ good := fakeIdentity("v1")
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ time.Sleep(50 * time.Millisecond)
+ b.lastWorker().crash()
+ }()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeHold, fakeIdentity("v2")), &good, one)
+ if res.Outcome != SwapDown {
+ t.Fatalf("ReplaceWorker = %+v, want down", res)
+ }
+ if len(b.calls()) != 2 {
+ t.Fatalf("spawns %v: a hold build fell back", b.calls())
+ }
+ waitFor(t, "the entry to stop", func() bool { return !a.Owns(host) })
+}
+
+// Spec §6.5: a failed restart respawns the build it replaced, and writes
+// nothing against it.
+func TestReplaceWorkerRestartFailureRespawnsTheServingBuild(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ serving := a.entryServing(host)
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ time.Sleep(50 * time.Millisecond)
+ b.lastWorker().crash()
+ }()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRestart, ModeOverlap, serving), nil, one)
+ if res.Outcome != SwapFailed {
+ t.Fatalf("ReplaceWorker = %+v", res)
+ }
+ waitFor(t, "the serving build back", func() bool { return len(b.calls()) == 3 && !a.entryRespawning(host) })
+ if calls := b.calls(); *calls[2] != serving {
+ t.Fatalf("respawned %+v, want the serving build", calls[2])
+ }
+}
+
+// Spec §6.5: a clean exit inside the 10 s is a one-shot's success.
+func TestReplaceWorkerOneShotCleanExitIsReady(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatal(err)
+ }
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ b.lastWorker().exit()
+ }()
+ if res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one); res.Outcome != SwapDone {
+ t.Fatalf("ReplaceWorker = %+v, want done", res)
+ }
+}
+
+// Spec §6.3, §7.4, §8.1: the attempt's checks run again at the actual
+// spawn, after the old child has retired, and again before its result is
+// accepted. A newer release (or a sibling's failure, whose halt bumps the
+// revision) that lands during the retire means the attempt's build never
+// execs: the build it replaced comes back instead. One that lands during
+// the observation means the result is abandoned, never done, so the build
+// is not recorded good. Checked only before the retire, both slip through.
+func TestReplaceWorkerReChecksAtTheSpawnAndBeforeTheResult(t *testing.T) {
+ t.Run("superseded while the old child retires", func(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ gate := make(chan struct{})
+ b.lastWorker().setStopGate(gate)
+ var rev atomic.Uint64
+ rev.Store(1)
+ results := make(chan SwapResult, 1)
+ go func() {
+ results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
+ func(string) uint64 { return rev.Load() })
+ }()
+ waitFor(t, "the retire to begin", func() bool { return a.entryRespawning(host) })
+ rev.Store(2) // a newer release lands while the old child drains
+ close(gate)
+ res := <-results
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded", res)
+ }
+ for _, c := range b.calls() {
+ if c != nil && *c == fakeIdentity("v2") {
+ t.Fatalf("spawns %v: the superseded build was exec'd", b.calls())
+ }
+ }
+ waitFor(t, "a child back on the entry", func() bool { return len(b.calls()) == 2 && !a.entryRespawning(host) })
+ if got := a.entryServing(host); got != fakeIdentity("v1") {
+ t.Fatalf("serving %+v, want the build the attempt replaced", got)
+ }
+ })
+ t.Run("superseded during the observation", func(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ var rev atomic.Uint64
+ rev.Store(1)
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ rev.Store(2)
+ }()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
+ func(string) uint64 { return rev.Load() })
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: a superseded result is never done", res)
+ }
+ })
+}
+
+// PF11: the attempt's checks also run in main's exec-time gate, inside
+// startWorker's before-exec hook, after the preparation (here the config
+// admission) and the publication. A newer release that lands after the
+// respawn's first check but before the exec means the attempt's build never
+// execs, and the build it replaced comes back.
+func TestReplaceWorkerReChecksAtTheExecItself(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ var rev atomic.Uint64
+ rev.Store(1)
+ b.mu.Lock()
+ b.admitHook = func() { rev.Store(2) } // lands during the respawn's preparation
+ b.mu.Unlock()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
+ func(string) uint64 { return rev.Load() })
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded at the exec", res)
+ }
+ waitFor(t, "the restore", func() bool { return b.workerStarts() == 2 && !a.entryRespawning(host) })
+ if got := a.entryServing(host); got != fakeIdentity("v1") {
+ t.Fatalf("serving %+v, want the build the attempt replaced (v2 must never have exec'd)", got)
+ }
+}
+
+// PF11: and once more before the child is installed. A newer release that
+// lands while the attempt's child is being exec'd sheds that child; the
+// build it replaced comes back, and the result is abandoned, never done.
+func TestReplaceWorkerReChecksBeforeInstalling(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ var rev atomic.Uint64
+ rev.Store(1)
+ var hooked atomic.Bool
+ b.mu.Lock()
+ b.workerStartHook = func(registry.Route) {
+ if hooked.CompareAndSwap(false, true) {
+ rev.Store(2) // lands after the exec gate, before the install check
+ }
+ }
+ b.mu.Unlock()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
+ func(string) uint64 { return rev.Load() })
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded before the install", res)
+ }
+ waitFor(t, "the restore", func() bool { return b.workerStarts() == 3 && !a.entryRespawning(host) })
+ b.mu.Lock()
+ shed := b.workers[1]
+ b.mu.Unlock()
+ if !shed.isStopped() {
+ t.Fatal("the superseded attempt's child was left running")
+ }
+ if got := a.entryServing(host); got != fakeIdentity("v1") {
+ t.Fatalf("serving %+v, want the build the attempt replaced", got)
+ }
+}
+
+// PF11 (plan 3's termination proof is never dropped): a respawn whose old
+// child is not proven gone execs nothing, the attempt's build and a restore
+// alike, and the attempt is abandoned.
+func TestReplaceWorkerNeverExecsBesideAnUnprovenOldChild(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ a.gens = true
+ hk := &plannedHostKill{stagedFake: b}
+ a.backend = hk
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ hk.bad.Store(true) // the host cgroup still holds a process
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ if res.Outcome != SwapAbandoned {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned", res)
+ }
+ waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
+ time.Sleep(50 * time.Millisecond)
+ if n := b.workerStarts(); n != 1 {
+ t.Fatalf("%d worker children: something exec'd beside an unproven old child", n)
+ }
+}
+
+// Spec §7.4: a respawn whose gate finds the host no longer owned (here the
+// lease lost while the old child drained) spawns nothing at all, the
+// replaced build included. Turning every refusal into a restore ran the old
+// build against a database the box no longer owned.
+func TestReplaceWorkerWithoutOwnershipSpawnsNothing(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ gate := make(chan struct{})
+ b.lastWorker().setStopGate(gate)
+ results := make(chan SwapResult, 1)
+ go func() {
+ results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ }()
+ waitFor(t, "the retire to begin", func() bool { return a.entryRespawning(host) })
+ b.lastRep().setErr(errors.New("lease renewal failed")) // ownership lost mid-retire
+ close(gate)
+ if res := <-results; res.Outcome != SwapAbandoned {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned", res)
+ }
+ waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(host) })
+ if calls := b.calls(); len(calls) != 1 {
+ t.Fatalf("spawns %v: something exec'd after the lease was lost", calls)
+ }
+}
+
+// A superseded attempt touches nothing.
+func TestReplaceWorkerSupersededSpawnsNothing(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatal(err)
+ }
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, func(string) uint64 { return 2 })
+ if res.Outcome != SwapAbandoned || len(b.calls()) != 1 {
+ t.Fatalf("ReplaceWorker = %+v, spawns %v", res, b.calls())
+ }
+}
+
+// PF9: an attempt whose build could not be published before its exec
+// (ErrBuildNotPublished) ran nothing of the build. It is abandonment, a
+// staging refusal, never a failure of the build.
+func TestReplaceWorkerRefusedForPublicationIsAbandoned(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ blocker := filepath.Join(t.TempDir(), "not-a-dir")
+ if err := os.WriteFile(blocker, nil, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ a.mu.Lock()
+ a.GenerationsFile = filepath.Join(blocker, "generations.json") // unwritable from here on
+ a.mu.Unlock()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ if res.Outcome != SwapAbandoned {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: an unpublished build never ran", res)
+ }
+ waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
+ if n := b.workerStarts(); n != 1 {
+ t.Fatalf("%d worker children: an unpublished build was exec'd", n)
+ }
+}
+
+// Carried from Task 9: e.unexpectedExit is never cleared by the arms that
+// set it. A stale mark from an earlier crash must not turn an intentional
+// stop during the attempt's observation into a verdict: the stop cancels
+// the observation, and the attempt is abandoned, never done.
+func TestReplaceWorkerAnIntentionalStopDuringTheObservationIsNoVerdict(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ a.mu.Lock()
+ a.entries[host].unexpectedExit = true // left by an earlier crash restart
+ a.mu.Unlock()
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ waitFor(t, "the observation", func() bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ return e != nil && e.firstWatch != nil
+ })
+ a.mu.Lock()
+ a.beginStopLocked(a.entries[host]) // an operator's stop
+ a.mu.Unlock()
+ }()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ if res.Outcome != SwapAbandoned {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: an intentional stop records nothing", res)
+ }
+}
+
+// Spec §7.9: with CARLOS_SWAP unset (no generations, no rollout manager), a
+// worker's captured serving identity is still published on its wake and on
+// every respawn, so `carlos sidecars ls` shows what runs. Only the rollout
+// manager wrote it, and a flag-off box showed "unknown", or a stale build
+// after the flag was turned off.
+func TestAWorkersServingIdentityIsPublishedWithTheFlagOff(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ var mu sync.Mutex
+ var got []string
+ a.OnWorkerServing = func(host string, id release.StagedIdentity) {
+ mu.Lock()
+ got = append(got, host+"="+id.Version)
+ mu.Unlock()
+ }
+ published := func() int { mu.Lock(); defer mu.Unlock(); return len(got) }
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ waitFor(t, "the wake's identity", func() bool { return published() == 1 })
+ b.lastWorker().crash()
+ a.sweepOnce() // the always-on worker respawns
+ waitFor(t, "the respawn's identity", func() bool { return published() == 2 })
+ mu.Lock()
+ defer mu.Unlock()
+ if want := host + "=" + a.entryServing(host).Version; got[1] != want {
+ t.Fatalf("published %v, want the respawned child's %q last", got, want)
+ }
+}
+
+// Spec §7.9: a delayed publication of an older worker incarnation never
+// overwrites a newer one. Here the wake's publication (incarnation 1) is
+// replayed after the respawn published incarnation 2: it is dropped. An
+// unconditional write left the older version on display indefinitely.
+func TestAnOlderWorkerPublicationIsDropped(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ var mu sync.Mutex
+ var got []string
+ a.OnWorkerServing = func(_ string, id release.StagedIdentity) {
+ mu.Lock()
+ got = append(got, id.Version)
+ mu.Unlock()
+ }
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ a.mu.Lock()
+ e := a.entries[host]
+ firstID, firstInc := e.workerServing, e.workerInc
+ a.mu.Unlock()
+ b.lastWorker().crash()
+ a.sweepOnce()
+ waitFor(t, "the respawn's publication", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 2 })
+ a.publishWorkerServing(e, firstID, firstInc) // the older one, delayed
+ mu.Lock()
+ defer mu.Unlock()
+ if len(got) != 2 {
+ t.Fatalf("published %v: an older incarnation's publication was delivered after a newer one", got)
+ }
+}
+
+// Spec §7.9 (P5 review finding 8): a child exec'd from an external live-link
+// target has no captured version. Its publication still goes out, empty,
+// so the row stops naming the build the previous child ran. Skipping empty
+// versions left ServingVersion=v1 on display indefinitely.
+func TestAnUnknownVersionClearsTheServingVersion(t *testing.T) {
+ a, _, _, _, _ := stagedActivator(t)
+ var mu sync.Mutex
+ var got []string
+ a.OnWorkerServing = func(_ string, id release.StagedIdentity) {
+ mu.Lock()
+ got = append(got, id.Version)
+ mu.Unlock()
+ }
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ waitFor(t, "the wake's identity", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 1 })
+ a.mu.Lock()
+ e := a.entries[host]
+ e.workerInc++ // a replacement exec'd from an external live-link target
+ inc := e.workerInc
+ a.mu.Unlock()
+ a.publishWorkerServing(e, release.StagedIdentity{StagedPath: "/srv/external/app"}, inc)
+ mu.Lock()
+ defer mu.Unlock()
+ if len(got) != 2 || got[1] != "" {
+ t.Fatalf("published %q, want the empty version last: the old one must be cleared", got)
+ }
+}
+
+// Spec §7.4 (P5 review finding 1): every respawn passes the ownership check
+// at its exec, a fallback included. Here B execs, the lease is lost, and
+// then B crashes: the good build is NOT respawned. Without the check the
+// fallback ran A against a database the box no longer owned.
+func TestAWorkerFallbackWithoutOwnershipSpawnsNothing(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ good := fakeIdentity("v1")
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ b.lastRep().setErr(errors.New("lease renewal failed")) // ownership lost after B's exec
+ time.Sleep(50 * time.Millisecond)
+ b.lastWorker().crash()
+ }()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
+ if res.Outcome != SwapFailed || res.Gen.Build != "" {
+ t.Fatalf("ReplaceWorker = %+v, want failed with nothing serving", res)
+ }
+ if calls := b.calls(); len(calls) != 2 {
+ t.Fatalf("spawns %v: the fallback exec'd after the lease was lost", calls)
+ }
+}
+
+// Spec §6.5, §8.1 (P5 review finding 5): a restart counts only once a NEW
+// child is installed. Here the restart's exec fails after the old child
+// stopped cleanly: the entry still holds that child, with the same
+// artifact, and its clean exit read as a one-shot's success, so the
+// scheduler acknowledged config that never ran.
+func TestAFailedRestartExecIsNeverDone(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ serving := a.entryServing(host)
+ b.mu.Lock()
+ b.workerErr = errors.New("exec: text file busy")
+ b.mu.Unlock()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRestart, ModeOverlap, serving), nil, one)
+ if res.Outcome == SwapDone {
+ t.Fatalf("ReplaceWorker = %+v: a restart whose child never ran was done", res)
+ }
+ if res.Outcome != SwapAbandoned {
+ t.Fatalf("ReplaceWorker = %+v: a failed exec is a staging refusal, never a failure of the build", res)
+ }
+ waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
+}
+
+// Spec §8.1 (P5 review finding 7): a ReplaceWorker fallback holds its
+// publication until the scheduler has recorded B's failure and calls
+// PublishHeldServing. Published from the respawn, a blocked callback sat
+// in front of the SwapFailed result, and so of the failure record.
+func TestAWorkerFallbackPublishesOnlyWhenReleased(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ var mu sync.Mutex
+ var got []string
+ block := make(chan struct{})
+ a.OnWorkerServing = func(_ string, id release.StagedIdentity) {
+ mu.Lock()
+ got = append(got, id.Version)
+ n := len(got)
+ mu.Unlock()
+ if n > 2 {
+ <-block // a slow registry write for the fallback's identity (the wake's v1, then B's v2, come first)
+ }
+ }
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ good := fakeIdentity("v1")
+ go func() {
+ waitForWorkerStarts(t, b, 2)
+ time.Sleep(50 * time.Millisecond)
+ b.lastWorker().crash()
+ }()
+ results := make(chan SwapResult, 1)
+ go func() {
+ results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
+ }()
+ select {
+ case res := <-results:
+ if res.Outcome != SwapFailed || res.Gen.Build != "v1" {
+ t.Fatalf("ReplaceWorker = %+v", res)
+ }
+ case <-time.After(5 * time.Second):
+ close(block)
+ t.Fatal("the failure result waited on the fallback's publication")
+ }
+ mu.Lock()
+ n := len(got)
+ mu.Unlock()
+ if n != 2 {
+ t.Fatalf("published %v before the failure was recorded", got)
+ }
+ close(block)
+ a.PublishHeldServing(host) // the scheduler, after done()
+ mu.Lock()
+ defer mu.Unlock()
+ if len(got) != 3 || got[2] != "v1" {
+ t.Fatalf("published %v, want the fallback's v1 once released", got)
+ }
+}
+
+// Spec §8.1 (P5 review finding 7): on a planned worker wake that falls back,
+// the failure transition (GenFellBack, whose OnGen records B's failure)
+// completes before the fallback's identity is published, and a blocked
+// publication does not hold it up. Publishing before afterWake left B
+// unheld while the callback blocked.
+func TestAFallbackWakeRecordsTheFailureBeforeItPublishes(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ a.gens = true
+ a.backend = &plannedHostKill{stagedFake: b} // the failed child's host cgroup proves empty
+ evs := &genLog{}
+ a.OnGen = evs.add
+ dir := t.TempDir()
+ good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
+ a.Planner = func(context.Context, registry.Route) WakePlan {
+ return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
+ }
+ block := make(chan struct{})
+ var sawFailure, called atomic.Bool
+ a.OnWorkerServing = func(string, release.StagedIdentity) {
+ sawFailure.Store(slices.Contains(evs.kinds(), GenFellBack))
+ called.Store(true)
+ <-block
+ }
+ go func() {
+ waitForWorkerStarts(t, b, 1)
+ time.Sleep(50 * time.Millisecond)
+ b.lastWorker().crash()
+ }()
+ go a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")) //nolint:errcheck
+ evs.wait(t, GenFellBack) // arrives while the publication is still blocked
+ waitFor(t, "the publication", func() bool { return called.Load() })
+ close(block)
+ if !sawFailure.Load() {
+ t.Fatal("the fallback's identity was published before its failure transition")
+ }
+}