amadan

carlos / platform Public

zdd plan 3 task 6: the hold swap, a pinned wake installed by finishStop

A hold stops the old generation through the barrier and finishStop
installs a waking entry pinned to the attempt's staged build in the same
critical section, before e.stopped closes, so held requests are served by
that build. Rows that cannot run a staged build (unit-backed, nested
before Task 9, fenced or conflicting layouts, claims on without Pins) are
refused before anything stops. The caller waits for the stop at most
wakeWaitBound, its own context or the agent's life; each withdraws the
hold under a.mu while the stop is under way. An overlap attempt on a
foreign row runs as a hold.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1e1271161455efea5d3b5782d55c9b0df4f4032b parent e0daa4e
7 files changed, +967 −5
  • internal/activator/activator.go +41 −0
  • internal/activator/claimwake.go +28 −1
  • internal/activator/swap.go +17 −1
  • internal/activator/swap_fixture_test.go +4 −0
  • internal/activator/swap_hold_test.go +571 −0
  • internal/activator/swap_overlap_test.go +0 −3
  • internal/activator/swaphold.go +306 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 9192a5447..304b6c2b2 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -433,6 +433,23 @@ type actEntry struct {
// health-checking or draining. Written under a.mu. Empty unless a swap
// has run on this entry, and swaps need plan 1's generations switch.
ops opSet
+ // pin is a hold swap's chosen build (zero-downtime spec §7.8). The wake
+ // that follows the hold's stop execs exactly this staged artifact rather
+ // than the live link. It is set at construction (holdWakeLocked), before
+ // the wake goroutine starts, and never written again, like the worker
+ // fields.
+ pin *Gen
+ // pinCheck is a hold's attempt checks, re-run under a.mu at the pinned
+ // exec. Set at construction (holdWakeLocked), like pin.
+ pinCheck func() error
+ // holdNext is a hold swap waiting for this entry's stop to finish.
+ // finishStop installs its pinned wake. Written under a.mu.
+ holdNext *holdSpec
+ // settleCancel, on a hold canary's pinned entry, ends its settle
+ // observation. beginStopLocked calls it while the process has not
+ // exited, so an intentional stop is never read as a crash (spec §6.4).
+ // Written under a.mu.
+ settleCancel context.CancelFunc
// genReaping is true while reapCurrentGen is running for this pending
// stop, so finishPendingStops does not start a second one. Written
// under a.mu.
@@ -1433,6 +1450,10 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
a.mu.Unlock()
_ = a.persistGenerations()
}
+ } else if e.pin != nil { // swapsim:M3
+ var g Gen
+ proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin)
+ w.gen = g
} else {
var g Gen
proc, g, err = a.startGen(ctx, rt)
@@ -1532,6 +1553,12 @@ const actStopTimeout = 20 * time.Second
// Bounce (config changed — the instance may be mid-request, but the drain
// still gives it actStopTimeout to finish before SIGKILL).
func (a *Service) beginStopLocked(e *actEntry) {
+ // A hold canary's settle ends with an intentional stop, before anything
+ // is stopped (zero-downtime spec §6.4). A process that already exited
+ // died of its build, and its stop must not hide that.
+ if e.settleCancel != nil && (e.proc == nil || !e.proc.Exited()) {
+ e.settleCancel()
+ }
e.state = actStopping
e.stopped = make(chan struct{})
go a.stop(e)
@@ -2596,6 +2623,17 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
log.Printf("activator: %s: remove link %s: %v", e.host, e.route.Socket, err)
}
}
+ // A hold swap (zero-downtime spec §7.8): its pinned wake is installed in
+ // the same critical section, before e.stopped closes, so a request
+ // waiting on this stop finds the pinned entry, never an empty slot it
+ // would fill with a wake of the live link.
+ var next *actEntry
+ if h := e.holdNext; h != nil {
+ e.holdNext = nil
+ var herr error
+ next, herr = a.holdWakeLocked(e, h)
+ h.woke <- holdWoke{next: next, err: herr}
+ }
a.mu.Unlock()
if e.worker {
// Its child is gone, so its build is no longer referenced by it.
@@ -2605,6 +2643,9 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
// its idle connections close now, not at the transport's idle timeout).
a.genRetired(e.host, retired)
close(e.stopped)
+ if next != nil {
+ go a.wake(next)
+ }
log.Printf("activator: %s hibernated", e.host)
}
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index c6e5ac36e..80434b07f 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -25,6 +25,11 @@ type StartPolicy struct {
Probe bool
Prepared *moves.InstallManifest
Inputs *provision.MoveInputs
+ // Pin is a hold swap's chosen build (zero-downtime spec §7.8). Prepare
+ // must build a start that execs exactly Pin.StagedPath after checking
+ // Pin.Digest. startAdmitted refuses a ClaimStart whose Version is not
+ // Pin.Build.
+ Pin *Gen
}
type startPolicy = StartPolicy
@@ -39,6 +44,10 @@ type ClaimRuntime struct {
Prerequisites func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error
Admit func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error)
Prepare func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error)
+ // Pins reports that Prepare honours StartPolicy.Pin. A hold swap is
+ // refused on a claims-on box whose runtime does not, before anything
+ // stops.
+ Pins bool
}
// ClaimStart is one admitted session. Its closures retain the admission tuple.
@@ -119,7 +128,19 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
}
rt = *fresh
}
- policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second}
+ policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Pin: e.pin}
+ if e.pin != nil && e.pinCheck != nil {
+ // A claims-on hold's pinned wake: the attempt gate before anything
+ // is admitted. Task 10 runs it again inside the work at every exec,
+ // after the admission and Restore (spec §6.3, §7.4), as
+ // startPinnedChecked does for a claims-off one.
+ a.mu.Lock()
+ err := e.pinCheck()
+ a.mu.Unlock()
+ if err != nil {
+ return nil, err
+ }
+ }
if err := r.Prerequisites(ctx, rt, observed, policy); err != nil {
return nil, err
}
@@ -171,6 +192,12 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
if !s.valid() || s.Host != e.host || (policy.Probe && s.OpenProbe == nil) {
return hostwrite.ErrUnavailable
}
+ if policy.Pin != nil && s.Version != policy.Pin.Build {
+ // A hold swap's wake must run the build its attempt chose
+ // (zero-downtime spec §7.8). A start prepared for anything else is
+ // refused before Restore touches the database.
+ return hostwrite.ErrUnavailable
+ }
if s.Live != nil {
if err := s.Live.Err(); err != nil {
return err
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
index e5ddde3b7..433752f9a 100644
--- a/internal/activator/swap.go
+++ b/internal/activator/swap.go
@@ -232,7 +232,23 @@ func (a *Service) Swap(ctx context.Context, at Attempt, newest func(host string)
}
switch at.Mode {
case ModeOverlap:
- return a.swapOverlap(ctx, gb, at, newest)
+ if a.foreignRow(at.Host) {
+ // A foreign row keeps its canonical socket (plan 1), so two
+ // generations of it cannot coexist: its replacement is a hold
+ // (spec §2.9, rev 17). Its result reads as a hold's, SwapDown
+ // when the new build fails, and Mode tells the scheduler so: a
+ // hold build that ran is held down (spec §6.4, §8.3).
+ res := a.swapHold(ctx, at, newest)
+ res.Mode = ModeHold
+ return res
+ }
+ res := a.swapOverlap(ctx, gb, at, newest)
+ res.Mode = ModeOverlap
+ return res
+ case ModeHold:
+ res := a.swapHold(ctx, at, newest)
+ res.Mode = ModeHold
+ return res
}
return abandon(errSwapMode)
}
diff --git a/internal/activator/swap_fixture_test.go b/internal/activator/swap_fixture_test.go
index 2fdb1c807..619ed5040 100644
--- a/internal/activator/swap_fixture_test.go
+++ b/internal/activator/swap_fixture_test.go
@@ -192,6 +192,10 @@ func newSwapFixture(t *testing.T) *swapFixture {
rt := testRoute(f.host)
rdir := t.TempDir()
rt.Socket, rt.DBPath = filepath.Join(rdir, "app.sock"), filepath.Join(rdir, "app.db")
+ // Generation sockets too: a hold's pinned wake runs wake()'s awake
+ // transition, which points the canonical path at the generation's socket
+ // for real (swapCanonicalLink).
+ b.genSocketDir = t.TempDir()
arrangeAwake(t, a, rt)
f.incumbent = Gen{N: 1 << 40, Socket: "/run/carlos/g/incumbent.sock", Control: "/run/carlos/g/incumbent.sock.control",
Build: "v1", StagedPath: "/opt/carlos/versions/v1/app"}
diff --git a/internal/activator/swap_hold_test.go b/internal/activator/swap_hold_test.go
new file mode 100644
index 000000000..ab342ae2d
--- /dev/null
+++ b/internal/activator/swap_hold_test.go
@@ -0,0 +1,571 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os/exec"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// A hold is stop-then-start through the barrier (spec §7.8). A request that
+// arrives during the stop is held, not refused, and is served by the new
+// generation, which execs the attempt's staged build while the live link
+// names something else. This is the witness for mutation M3.
+func TestSwapWitnessHoldExecsTheStagedBuild(t *testing.T) {
+ f := newSwapFixture(t)
+ f.b.execVerFn = func() string { return "v3" }
+ gate := make(chan struct{})
+ f.b.lastProc().setStopGate(gate)
+ results := make(chan SwapResult, 1)
+ go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) }()
+ waitFor(t, "the hold's stop", func() bool {
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ e := f.a.entries[f.host]
+ return e != nil && e.state == actStopping
+ })
+ served := make(chan error, 1)
+ go func() {
+ release, err := f.a.beginRequest(context.Background(), testRoute(f.host))
+ if err == nil {
+ release()
+ }
+ served <- err
+ }()
+ time.Sleep(50 * time.Millisecond)
+ close(gate)
+ var res SwapResult
+ select {
+ case res = <-results:
+ case <-time.After(10 * time.Second):
+ t.Fatal("the hold never finished")
+ }
+ if res.Outcome != SwapDone || res.Gen.Build != "v2" || res.Gen.StagedPath != f.staged {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if res.Drain < 40*time.Millisecond {
+ t.Fatalf("Drain = %v, want the old generation's gated stop (spec §9.4)", res.Drain)
+ }
+ if err := <-served; err != nil {
+ t.Fatalf("the held request was refused: %v", err)
+ }
+ // #231: one live-link Start (the original wake). The hold's wake went
+ // through StartGen alone.
+ if got := f.b.startCount(); got != 1 {
+ t.Fatalf("Start (live link) calls = %d, want 1", got)
+ }
+ if gens, _, _ := f.b.calls(); gens != 1 {
+ t.Fatalf("StartGen calls = %d, want 1", gens)
+ }
+ if got := f.appSocket(t); got != res.Gen.Socket {
+ t.Fatalf("AppSocket = %q, want the new generation's %q", got, res.Gen.Socket)
+ }
+}
+
+// A hold canary is not done when it is healthy: it must survive the 10 s
+// settle (spec §6.2, §8.5). Without the settle, swapHold answers SwapDone
+// the moment the pinned wake is ready, and a crash a second later neither
+// fails the attempt nor halts the rollout.
+func TestSwapHoldCanarySettles(t *testing.T) {
+ old := swapSettle
+ t.Cleanup(func() { swapSettle = old })
+ start := func(t *testing.T, f *swapFixture) (chan SwapResult, Gen, *fakeProc) {
+ t.Helper()
+ at := f.attempt(ModeHold)
+ at.Canary = true
+ results := make(chan SwapResult, 1)
+ go func() { results <- f.a.Swap(context.Background(), at, f.newest) }()
+ var g Gen
+ var p *fakeProc
+ waitFor(t, "the pinned generation awake", func() bool {
+ g, p = f.b.lastGen()
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ e := f.a.entries[f.host]
+ return p != nil && e != nil && e.state == actAwake && e.gen.N == g.N
+ })
+ return results, g, p
+ }
+ t.Run("an exit in the window is down, with no fallback", func(t *testing.T) {
+ swapSettle = 5 * time.Second
+ f := newSwapFixture(t)
+ results, _, p := start(t, f)
+ select {
+ case res := <-results:
+ t.Fatalf("Swap answered %+v before the settle ended", res)
+ case <-time.After(50 * time.Millisecond):
+ }
+ p.exit()
+ res := <-results
+ if res.Outcome != SwapDown || !strings.Contains(res.Reason, errSettleExit.Error()) {
+ t.Fatalf("Swap = %+v, want down: %v", res, errSettleExit)
+ }
+ if gens, _, _ := f.b.calls(); gens != 1 || f.b.startCount() != 1 {
+ t.Fatalf("a failed hold canary fell back: StartGen %d, Start %d", gens, f.b.startCount())
+ }
+ })
+ t.Run("surviving the window is done", func(t *testing.T) {
+ swapSettle = 100 * time.Millisecond
+ f := newSwapFixture(t)
+ began := time.Now()
+ results, g, _ := start(t, f)
+ res := <-results
+ if res.Outcome != SwapDone || res.Gen.N != g.N {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if took := time.Since(began); took < swapSettle {
+ t.Fatalf("done after %s, inside the %s settle", took, swapSettle)
+ }
+ })
+ t.Run("an intentional stop in the window is not a failure", func(t *testing.T) {
+ swapSettle = 5 * time.Second
+ f := newSwapFixture(t)
+ results, _, _ := start(t, f)
+ f.a.Bounce(f.host)
+ res := <-results
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapStopped.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned: %v", res, errSwapStopped)
+ }
+ })
+}
+
+// A hold build that fails its health check leaves nothing serving and does
+// not fall back (spec §8.3).
+func TestSwapHoldFailureIsDown(t *testing.T) {
+ f := newSwapFixture(t)
+ f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapDown {
+ t.Fatalf("Swap = %+v, want down", res)
+ }
+ if f.a.entryCount() != 0 {
+ t.Fatal("a failed hold left an entry behind")
+ }
+ if _, p := f.b.lastGen(); p == nil || !p.isStopped() {
+ t.Fatal("the failed hold generation was left running")
+ }
+ if got := f.b.startCount(); got != 1 {
+ t.Fatalf("Start (live link) calls = %d: a failed hold fell back", got)
+ }
+}
+
+// acquireHookBackend runs hook inside Acquire: the window between a hold's
+// re-check at its stop and its pinned wake's exec.
+type acquireHookBackend struct {
+ *fakeSwapBackend
+ hook func()
+}
+
+func (b *acquireHookBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
+ b.hook()
+ return b.fakeSwapBackend.Acquire(ctx, rt)
+}
+
+// Spec §6.3, §7.4, §8.1: a hold's attempt checks run again at the pinned
+// wake's exec and once more before its result counts. Checked only when
+// the stop finished, a newer attempt (or a halted cohort, which bumps the
+// revision) landing during the wake still exec'd the superseded build and
+// answered SwapDone, which records it good.
+func TestSwapHoldReChecksAtTheExecAndBeforeTheResult(t *testing.T) {
+ t.Run("superseded before the exec", func(t *testing.T) {
+ f := newSwapFixture(t)
+ var once sync.Once
+ f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }}
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned: superseded", res)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("the superseded hold build was exec'd (%d)", gens)
+ }
+ })
+ t.Run("superseded before the result", func(t *testing.T) {
+ f := newSwapFixture(t)
+ f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil }
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned: a superseded result is never done", res)
+ }
+ })
+}
+
+// killingSwapBackend is fakeSwapBackend plus plan 2's KillGen. fail decides,
+// per generation, whether the child is still occupied.
+type killingSwapBackend struct {
+ *fakeSwapBackend
+ kmu sync.Mutex
+ fail func(n uint64) bool
+}
+
+func (k *killingSwapBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error {
+ k.kmu.Lock()
+ defer k.kmu.Unlock()
+ if k.fail != nil && k.fail(n) {
+ return errors.New("the child still holds a process")
+ }
+ return nil
+}
+
+func (k *killingSwapBackend) setFail(f func(uint64) bool) {
+ k.kmu.Lock()
+ k.fail = f
+ k.kmu.Unlock()
+}
+
+// A failed wake gives its lease back only once its generation's child is
+// proven empty (spec §7.4). Until then the host takes no new wake, and the
+// sweep retries the proof. cleanupFailedWake that accepts the leader's exit
+// releases the lease over a surviving descendant and lets a new wake start
+// beside it.
+func TestSwapHoldFailedWakeKeepsTheLeaseUntilItsChildIsEmpty(t *testing.T) {
+ f := newSwapFixture(t)
+ kb := &killingSwapBackend{fakeSwapBackend: f.b}
+ kb.setFail(func(n uint64) bool { return n != f.incumbent.N }) // the hold's new generation only
+ f.a.backend = kb
+ f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
+ before := f.b.releaseCount()
+ if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
+ t.Fatalf("Swap = %+v, want down", res)
+ }
+ if got := f.b.releaseCount(); got != before+1 {
+ t.Fatalf("lease releases %d -> %d: want the old generation's only, the failed wake's kept", before, got)
+ }
+ if _, err := f.a.beginRequest(context.Background(), testRoute(f.host)); err == nil || !strings.Contains(err.Error(), errWakeUnproven.Error()) {
+ t.Fatalf("a new wake = %v, want refused while the failed generation is unproven", err)
+ }
+ kb.setFail(nil)
+ f.a.sweepOnce()
+ waitFor(t, "the failed wake's lease given back", func() bool { return f.b.releaseCount() == before+2 })
+}
+
+// The same when the failed wake's leader itself never stops: an unproven
+// stop (ErrStillRunning) is no termination, so the lease is kept until the
+// sweep's retried stop and proof hold. Retaining only when the leader had
+// exited released the lease over a live process.
+func TestSwapHoldFailedWakeWhoseLeaderNeverStopsKeepsTheLease(t *testing.T) {
+ f := newSwapFixture(t)
+ f.b.gmu.Lock()
+ f.b.genStuck = true
+ f.b.gmu.Unlock()
+ f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
+ before := f.b.releaseCount()
+ if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
+ t.Fatalf("Swap = %+v, want down", res)
+ }
+ if got := f.b.releaseCount(); got != before+1 {
+ t.Fatalf("lease releases %d -> %d: the failed wake's lease was given back over a live leader", before, got)
+ }
+ _, p := f.b.lastGen()
+ p.exit()
+ f.a.sweepOnce()
+ waitFor(t, "the failed wake's lease given back", func() bool { return f.b.releaseCount() == before+2 })
+}
+
+// The re-check before the spawn happens when the stop finishes. A fence, a
+// newer attempt or the agent's shutdown abandons the hold, and nothing is
+// spawned.
+func TestSwapHoldRefusedAtTheStop(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ act func(t *testing.T, f *swapFixture)
+ want error
+ }{
+ {"fenced", func(_ *testing.T, f *swapFixture) {
+ f.a.mu.Lock()
+ if f.a.fences == nil {
+ f.a.fences = map[string]uint64{}
+ }
+ f.a.fences[f.host] = 5
+ f.a.mu.Unlock()
+ }, errSwapFenced},
+ {"superseded", func(_ *testing.T, f *swapFixture) { f.rev.Add(1) }, errSwapSuperseded},
+ {"shutdown", func(t *testing.T, f *swapFixture) {
+ go f.a.Shutdown(context.Background())
+ // The shutdown is announced before the old generation's stop is
+ // let go, so the outcome does not hang on goroutine scheduling.
+ waitFor(t, "the shutdown announced", func() bool {
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ return f.a.shuttingDown
+ })
+ }, errSwapShutdown},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newSwapFixture(t)
+ gate := make(chan struct{})
+ f.b.lastProc().setStopGate(gate)
+ results := make(chan SwapResult, 1)
+ go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) }()
+ waitFor(t, "the hold's stop", func() bool {
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ e := f.a.entries[f.host]
+ return e != nil && e.state == actStopping
+ })
+ tc.act(t, f)
+ close(gate)
+ res := <-results
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned: %v", res, tc.want)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatal("a refused hold spawned a generation")
+ }
+ })
+ }
+}
+
+// Claims on: a hold needs a claim runtime that can pin a wake to a staged
+// build. Without one it is refused BEFORE anything stops.
+func TestSwapHoldClaimsOnNeedsPins(t *testing.T) {
+ f := newSwapFixture(t)
+ f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}}
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errHoldNeedsPins.Error()) {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if f.b.lastProc().isStopped() {
+ t.Fatal("a refused hold stopped the incumbent")
+ }
+}
+
+// A row that cannot run a staged generation is refused BEFORE anything
+// stops (spec §7.8 is stop-then-start: a start that cannot happen must not
+// cost the instance its running process). Refused after the stop, the
+// pinned wake failed and the hold answered SwapDown for a build that never
+// ran. Swaps off refuses a hold as it refuses an overlap.
+func TestSwapHoldRefusesARowThatCannotRunAStagedBuild(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ arrange func(f *swapFixture)
+ want error
+ }{
+ {"swap off", func(f *swapFixture) { f.a.gens = false }, errSwapOff},
+ {"unit-backed", func(f *swapFixture) {
+ f.a.mu.Lock()
+ f.a.entries[f.host].route.Backing = registry.BackingUnit
+ f.a.mu.Unlock()
+ }, errSwapKind},
+ {"nested", func(f *swapFixture) {
+ f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
+ }, errSwapLayout},
+ {"fenced layout", func(f *swapFixture) {
+ f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutFenced, "a migration journal" }
+ }, errSwapLayout},
+ {"worker", func(f *swapFixture) {
+ f.a.mu.Lock()
+ f.a.entries[f.host].worker = true
+ f.a.mu.Unlock()
+ }, errSwapKind},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newSwapFixture(t)
+ tc.arrange(f)
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned: %v", res, tc.want)
+ }
+ if f.b.lastProc().isStopped() {
+ t.Fatal("a refused hold stopped the incumbent")
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 || f.b.startCount() != 1 {
+ t.Fatalf("a refused hold spawned: StartGen %d, Start %d", gens, f.b.startCount())
+ }
+ f.a.mu.Lock()
+ e := f.a.entries[f.host]
+ held := e != nil && e.holdNext != nil
+ f.a.mu.Unlock()
+ if held {
+ t.Fatal("a refused hold left its spec on the entry")
+ }
+ })
+ }
+}
+
+// The caller is released when the old generation's stop outlasts
+// wakeWaitBound, the bound §7.8 holds requests for, or when the caller's own
+// context ends. The hold is withdrawn under the entry lock, so the stop's
+// eventual finish installs no pinned wake and the next request wakes the
+// ordinary way. Waiting on the stop alone held the caller for as long as an
+// unproven stop stayed pending.
+func TestSwapHoldReleasesItsCallerWhileTheStopIsPending(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ run func(f *swapFixture) SwapResult
+ want error
+ }{
+ {"the stop outlasts wakeWaitBound", func(f *swapFixture) SwapResult {
+ return f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ }, errHoldStopSlow},
+ {"the caller's context ends", func(f *swapFixture) SwapResult {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ go func() {
+ // Cancel once the hold's stop has begun (bounded: the test's
+ // own assertion reports a hold that never got there).
+ for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) {
+ f.a.mu.Lock()
+ e := f.a.entries[f.host]
+ stopping := e != nil && e.state == actStopping
+ f.a.mu.Unlock()
+ if stopping {
+ break
+ }
+ }
+ cancel()
+ }()
+ return f.a.Swap(ctx, f.attempt(ModeHold), f.newest)
+ }, errSwapStopped},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ old := wakeWaitBound
+ t.Cleanup(func() { wakeWaitBound = old })
+ f := newSwapFixture(t)
+ if errors.Is(tc.want, errHoldStopSlow) {
+ wakeWaitBound = 100 * time.Millisecond
+ }
+ gate := make(chan struct{})
+ incumbent := f.b.lastProc()
+ incumbent.setStopGate(gate)
+ res := tc.run(f)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want.Error()) {
+ close(gate)
+ t.Fatalf("Swap = %+v, want abandoned: %v", res, tc.want)
+ }
+ f.a.mu.Lock()
+ prev := f.a.entries[f.host]
+ if prev == nil || prev.state != actStopping || prev.holdNext != nil {
+ f.a.mu.Unlock()
+ close(gate)
+ t.Fatal("want the old entry still stopping, with its hold withdrawn")
+ }
+ stopped := prev.stopped
+ f.a.mu.Unlock()
+ close(gate)
+ <-stopped
+ // finishStop would install the pinned wake in the critical section
+ // that closes stopped, so its absence now is final.
+ f.a.mu.Lock()
+ next := f.a.entries[f.host]
+ f.a.mu.Unlock()
+ if next != nil {
+ t.Fatalf("a withdrawn hold installed an entry (state %v)", next.state)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("a withdrawn hold exec'd its build (%d)", gens)
+ }
+ })
+ }
+}
+
+// A claims-on pinned wake runs the hold's attempt gate before anything is
+// admitted: a superseded hold reaches neither the prerequisites nor the
+// admission.
+func TestDoClaimWakeRunsThePinGateFirst(t *testing.T) {
+ obs := hostclaim.Observed{Host: "x.example", ETag: "one", Claim: &hostclaim.Claim{
+ Holder: &hostclaim.Holder{Location: "instances/x.example/inc", Session: 1, Generation: "one", WriterBox: "box"},
+ Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"},
+ }}
+ admitted := false
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Pins: true,
+ Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil },
+ StopForeign: func(context.Context, string) error { return nil },
+ FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil },
+ Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error {
+ admitted = true
+ return nil
+ },
+ Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
+ admitted = true
+ return obs, nil
+ },
+ Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ admitted = true
+ return ClaimStart{}, nil
+ }}}}
+ e := &actEntry{host: obs.Host, route: registry.Route{Host: obs.Host, Kind: "instance"}, pin: &Gen{Build: "v2"},
+ pinCheck: func() error { return errSwapSuperseded }}
+ if _, err := a.doClaimWake(context.Background(), e); !errors.Is(err, errSwapSuperseded) || admitted {
+ t.Fatalf("doClaimWake = %v (admitted %v), want the pin gate's refusal before admission", err, admitted)
+ }
+}
+
+// pinTestClaimStart is a ClaimStart that passes valid(): main's claims-on
+// launch needs a RoleLauncher and CurrentAttempt (claimwindow_test.go). On
+// this base RoleLaunch.Start always refuses, so no test here reaches it.
+func pinTestClaimStart(obs hostclaim.Observed, work hostwrite.Work, version string, restore func(context.Context) error) ClaimStart {
+ return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} },
+ Host: obs.Host, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Version: version,
+ Restore: restore,
+ Materialize: func(context.Context) error { return nil }, Build: func() *exec.Cmd { return nil },
+ Ready: func(context.Context, *exec.Cmd) error { return nil }, Reserve: func(context.Context, int) error { return nil },
+ Failed: func(context.Context, int) error { return nil }, Healthy: func(context.Context, int, *exec.Cmd) error { return nil },
+ Replicate: func(context.Context, string) (Rep, error) { return nil, nil },
+ Publish: func(context.Context, hostclaim.Observed) (hostclaim.Observed, error) { return obs, nil },
+ Retain: func(Proc, Rep, error) {}}
+}
+
+// A claim runtime that prepared a different build from the pinned one is
+// refused before Restore touches the database.
+func TestStartAdmittedRefusesAPinMismatch(t *testing.T) {
+ obs := hostclaim.Observed{Host: "x.example", ETag: "one", Claim: &hostclaim.Claim{
+ Holder: &hostclaim.Holder{Location: "instances/x.example/inc", Session: 1, Generation: "one"},
+ Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"},
+ }}
+ work, _ := (hostwrite.Config{}).Select(context.Background(), obs.Host, "spawn")
+ stop := errors.New("stop at restore")
+ for _, tc := range []struct {
+ version string
+ restored bool
+ }{{"v1", false}, {"v2", true}} {
+ restored := false
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
+ Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil },
+ Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ return pinTestClaimStart(obs, work, tc.version, func(context.Context) error { restored = true; return stop }), nil
+ }}}}
+ policy := startPolicy{MaxStarts: 1, Window: time.Second, Pin: &Gen{Build: "v2"}}
+ err := a.startAdmitted(context.Background(), &actEntry{host: obs.Host}, obs, policy)
+ if restored != tc.restored {
+ t.Fatalf("prepared %s: restored = %v, want %v (err %v)", tc.version, restored, tc.restored, err)
+ }
+ if !tc.restored && !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("prepared %s: %v, want hostwrite.ErrUnavailable", tc.version, err)
+ }
+ if tc.restored && !errors.Is(err, stop) {
+ t.Fatalf("prepared %s: %v, want the restore's own error", tc.version, err)
+ }
+ }
+}
+
+// A foreign row keeps its canonical socket, so an overlap attempt on it
+// runs as a hold (plan 1's reading): the old process stops, then the staged
+// build starts on the same socket. Refusing it (errSwapLayout) left a
+// foreign row unable to take any release with the flag on.
+func TestSwapOverlapOnAForeignRowRunsAsAHold(t *testing.T) {
+ f := newSwapFixture(t)
+ f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutForeign, "hand-set socket" }
+ incumbent := f.b.lastProc()
+ res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
+ if res.Outcome != SwapDone || res.Gen.StagedPath != f.staged || res.Mode != ModeHold {
+ t.Fatalf("Swap = %+v, want done through the hold, reported as a hold (Mode)", res)
+ }
+ ev := f.b.events()
+ if i, j := slices.Index(ev, "stop:instance"), indexPrefix(ev, "startGen:"+f.staged); i < 0 || j < 0 || i > j {
+ t.Fatalf("order %v: a foreign row stops before its replacement starts", ev)
+ }
+ if indexPrefix(ev, "raise") >= 0 {
+ t.Fatalf("events %v: a foreign row never raises an overlap parent", ev)
+ }
+ if !incumbent.isStopped() {
+ t.Fatal("the old process outlived the hold")
+ }
+}
diff --git a/internal/activator/swap_overlap_test.go b/internal/activator/swap_overlap_test.go
index a7ed526f2..07a35cb85 100644
--- a/internal/activator/swap_overlap_test.go
+++ b/internal/activator/swap_overlap_test.go
@@ -182,9 +182,6 @@ func TestSwapRefusals(t *testing.T) {
f.a.mu.Unlock()
}, errSwapFenced},
{"lease lost", func(_ *testing.T, f *swapFixture, _ *Attempt) { f.b.lastRep().setErr(errors.New("renewal failed")) }, errSwapLeaseLost},
- {"foreign layout", func(_ *testing.T, f *swapFixture, _ *Attempt) {
- f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutForeign, "hand-set socket" }
- }, errSwapLayout},
} {
t.Run(tc.name, func(t *testing.T) {
f := newSwapFixture(t)
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
new file mode 100644
index 000000000..298f0efa1
--- /dev/null
+++ b/internal/activator/swaphold.go
@@ -0,0 +1,306 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+var (
+ errHoldNeedsPins = errors.New("swap: claims are on and the claim runtime cannot pin a wake to a staged build")
+ // errHoldStopSlow: the old generation's stop outlasted wakeWaitBound, the
+ // bound a hold holds requests for (spec §7.8). The hold is withdrawn
+ // before its stop finishes, so nothing of the attempt's build runs.
+ errHoldStopSlow = errors.New("swap: the old generation's stop outlasted the hold's bound; the hold is withdrawn")
+)
+
+// holdSpec is a hold swap waiting for its entry's stop (spec §7.8).
+// finishStop answers on woke exactly once, unless swapHold withdraws the
+// hold first (under a.mu, by clearing the entry's holdNext).
+type holdSpec struct {
+ gen Gen // the attempt's build, staged path, digest and demand; N is drawn at spawn
+ at Attempt
+ newest func(string) uint64
+ woke chan holdWoke // buffered 1
+ // settleCancel ends a canary's settle observation. holdWakeLocked puts
+ // it on the pinned entry, so an intentional stop of that entry cancels
+ // the observation before anything is stopped (spec §6.4).
+ settleCancel context.CancelFunc
+}
+
+// holdWoke is finishStop's answer: the pinned wake it installed, or why it
+// installed none.
+type holdWoke struct {
+ next *actEntry
+ err error
+}
+
+// lifeDone is the agent lifetime's Done channel. It is nil (blocks forever
+// in a select) for a Service built without New.
+func (a *Service) lifeDone() <-chan struct{} {
+ if a.lifeCtx == nil {
+ return nil
+ }
+ return a.lifeCtx.Done()
+}
+
+// swapHold is §7.8: stop-then-start on the existing path, through the
+// barrier. The old generation stops while begin() holds requests, up to
+// wakeWaitBound. When the stop finishes, finishStop installs a waking
+// entry pinned to the attempt's staged build before it releases anyone
+// waiting on the stop. Held requests are served once that generation is
+// healthy.
+//
+// A row that cannot run the staged build at all is refused before
+// anything stops: a stop-then-start whose start cannot happen would only
+// cost the instance its running process.
+//
+// The caller waits for the stop for at most wakeWaitBound, and no longer
+// than its own context or the agent's life. Any of the three withdraws the
+// hold under a.mu while the stop is still under way, and the attempt is
+// abandoned: nothing of its build has run. Once finishStop has taken the
+// hold, the pinned wake may exec the build against the database, so the
+// result follows that wake to its end (bounded by the wake's own deadlines)
+// and is never abandoned on the caller's account.
+func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string) uint64) SwapResult {
+ if err := verifyStaged(at.StagedPath, at.Digest); err != nil {
+ return abandon(fmt.Errorf("staging: %w", err))
+ }
+ work, err := a.selectWork(ctx, at.Host)
+ if err != nil {
+ return abandon(fmt.Errorf("admission: %w", err))
+ }
+ if !hostwrite.Legacy(work) && (a.cfg.Claims == nil || !a.cfg.Claims.Pins) {
+ return abandon(errHoldNeedsPins)
+ }
+ sctx, scancel := context.WithCancel(ctx)
+ defer scancel()
+ h := &holdSpec{
+ gen: Gen{Build: at.Build, StagedPath: at.StagedPath, Digest: at.Digest, Demand: at.Demand},
+ at: at,
+ newest: newest,
+ woke: make(chan holdWoke, 1),
+ }
+ if at.Canary {
+ h.settleCancel = scancel
+ }
+ var (
+ held *actEntry // the entry whose stop the hold waits on
+ stopAt time.Time // wall clock: the old generation's stop, for the drain time (spec §9.4)
+ )
+ err = a.admit(ctx, work, func() error {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[at.Host]
+ if err := a.swapChecksLocked(e, at, newest); err != nil {
+ return err
+ }
+ if e.worker || e.route.Kind != "instance" || e.route.Backing == registry.BackingUnit {
+ return errSwapKind
+ }
+ // The layouts that can run a staged generation: flat, and foreign on
+ // its canonical socket. Nested waits for Task 9's executable mount.
+ if c, ok := a.backend.(Classifier); ok {
+ if l, _ := c.Layout(e.route); l != LayoutFlat && l != LayoutForeign {
+ return errSwapLayout
+ }
+ }
+ if len(e.ops.members) > 0 {
+ return errSwapBusy
+ }
+ e.holdNext = h
+ held, stopAt = e, time.Now()
+ a.beginStopLocked(e)
+ return nil
+ })
+ if err != nil {
+ if held != nil {
+ // The work refused after its body began the stop. The stop
+ // stands; the hold does not.
+ if !a.withdrawHold(held, h) {
+ return a.holdResult(sctx, at, newest, <-h.woke, time.Since(stopAt))
+ }
+ }
+ return abandon(err)
+ }
+ bound := time.NewTimer(wakeWaitBound)
+ defer bound.Stop()
+ var hw holdWoke
+ select {
+ case hw = <-h.woke:
+ case <-bound.C:
+ hw, err = a.withdrawOrTake(held, h, errHoldStopSlow)
+ case <-ctx.Done():
+ hw, err = a.withdrawOrTake(held, h, errSwapStopped)
+ case <-a.lifeDone():
+ hw, err = a.withdrawOrTake(held, h, errSwapShutdown)
+ }
+ if err != nil {
+ return abandon(err)
+ }
+ return a.holdResult(sctx, at, newest, hw, time.Since(stopAt)) // finishStop answers once the old generation is proven gone
+}
+
+// withdrawHold clears h from e while the stop has not taken it, and reports
+// whether it did. False means finishStop already took the hold and answered
+// on h.woke in the same critical section.
+func (a *Service) withdrawHold(e *actEntry, h *holdSpec) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if e.holdNext != h {
+ return false
+ }
+ e.holdNext = nil
+ return true
+}
+
+// withdrawOrTake withdraws the hold for why, or, when finishStop has taken
+// it already, returns its answer.
+func (a *Service) withdrawOrTake(e *actEntry, h *holdSpec, why error) (holdWoke, error) {
+ if a.withdrawHold(e, h) {
+ return holdWoke{}, why
+ }
+ return <-h.woke, nil
+}
+
+// holdResult follows a hold's pinned wake to its outcome.
+func (a *Service) holdResult(sctx context.Context, at Attempt, newest func(string) uint64, hw holdWoke, drain time.Duration) SwapResult {
+ if hw.err != nil {
+ return abandon(hw.err)
+ }
+ <-hw.next.ready
+ a.mu.Lock()
+ g, werr, proc := hw.next.gen, hw.next.err, hw.next.proc
+ a.mu.Unlock()
+ switch {
+ case werr == nil:
+ if at.Canary {
+ // A canary must survive the settle before it counts (spec §6.2,
+ // §8.5). Plan 5 does not observe a pinned wake itself.
+ r := observeSettle(sctx, proc, swapSettle)
+ a.mu.Lock()
+ hw.next.settleCancel = nil
+ a.mu.Unlock()
+ switch r {
+ case settleExited:
+ // Nothing falls back from a hold (spec §8.3): the instance is
+ // down until the sweep notices the exit and a later wake runs.
+ return SwapResult{Outcome: SwapDown, Gen: g, Reason: errSettleExit.Error(), Drain: drain}
+ case settleCancelled:
+ return abandon(errSwapStopped)
+ }
+ }
+ // The checks once more before the result counts (spec §6.3, §8.1):
+ // a newer attempt or a halted cohort since the wake began means this
+ // build is not recorded good. It keeps serving until the newer
+ // attempt replaces it. The same gate as every other exec point.
+ a.mu.Lock()
+ cerr := a.attemptCurrentLocked(hw.next, at, newest, false)
+ a.mu.Unlock()
+ if cerr != nil {
+ return abandon(cerr)
+ }
+ return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain}
+ case classifyStep(nil, werr) == stepStale:
+ // The pinned wake's own gate refused the exec (pinCheck): nothing
+ // of the build ran, and this is no failure of it.
+ return abandon(werr)
+ case classifyStep(nil, werr) == stepCancelled:
+ // An intentional stop ended the pinned wake (spec §6.4).
+ return abandon(errSwapStopped)
+ }
+ return SwapResult{Outcome: SwapDown, Gen: g, Reason: werr.Error(), Drain: drain}
+}
+
+// holdWakeLocked is a hold's re-check before the spawn, made when its stop
+// has finished and under the same lock. If the check passes, it installs
+// the pinned waking entry. The caller holds a.mu and starts the wake.
+func (a *Service) holdWakeLocked(e *actEntry, h *holdSpec) (*actEntry, error) {
+ pin := h.gen
+ next := &actEntry{host: e.host, route: e.route, state: actWaking, ready: make(chan struct{}), pin: &pin,
+ settleCancel: h.settleCancel}
+ // The same gate here, at the stop's end, and again at the pinned wake's
+ // exec (spec §6.3, §7.4): the lease and the restore come between.
+ next.pinCheck = func() error { return a.attemptCurrentLocked(next, h.at, h.newest, true) }
+ a.entries[e.host] = next
+ if err := next.pinCheck(); err != nil {
+ delete(a.entries, e.host)
+ return nil, err
+ }
+ return next, nil
+}
+
+// startPinnedChecked is a hold's pinned spawn behind its attempt's checks,
+// run under a.mu at the exec (pinCheck, set by holdWakeLocked). A check
+// that fails spawns nothing.
+func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt registry.Route, pin Gen) (Proc, string, Gen, error) {
+ a.mu.Lock()
+ var err error
+ if e.pinCheck != nil {
+ err = e.pinCheck()
+ }
+ a.mu.Unlock()
+ if err != nil {
+ return nil, "", pin, err
+ }
+ return a.startPinned(ctx, rt, pin)
+}
+
+// startPinned is a hold's spawn: the attempt's staged build, verified and
+// exec'd by path at its own tier, then the health check through the
+// generation handle. StartStaged returns as soon as the process runs (Task
+// 1), so this health check, and its deadline, is the only one. The proc
+// comes back even on failure, so doWake's cleanupFailedWake stops it and
+// records the outcome, and a caller that reads a nil proc as "nothing ran"
+// is right to.
+func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) (Proc, string, Gen, error) {
+ gb, ok := a.backend.(SwapBackend)
+ if !ok {
+ return nil, "", pin, errSwapNoGens
+ }
+ if err := verifyStaged(pin.StagedPath, pin.Digest); err != nil {
+ return nil, "", pin, fmt.Errorf("staging: %w", err)
+ }
+ pin.N = nextGen()
+ if pin.Tier == (cgroup.Limit{}) {
+ pin.Tier = gb.GenTier(rt, pin.Build) // the pinned artifact's own tier (spec §7.3)
+ }
+ g, proc, err := gb.StartStaged(ctx, rt, pin)
+ if err != nil {
+ return proc, "", pin, err
+ }
+ if g.N != pin.N || g.Build != pin.Build || g.StagedPath != pin.StagedPath {
+ return proc, "", pin, fmt.Errorf("backend ran generation %d of %s from %q; the hold chose %s from %q", g.N, g.Build, g.StagedPath, pin.Build, pin.StagedPath)
+ }
+ if err := a.genHealthy(ctx, g); err != nil {
+ return proc, "", g, err
+ }
+ gb.GenReady(rt, g, proc)
+ return proc, g.Build, g, nil
+}
+
+// foreignRow reports whether host's row has the foreign layout (a
+// hand-set socket path), which keeps its canonical socket.
+func (a *Service) foreignRow(host string) bool {
+ c, ok := a.backend.(Classifier)
+ if !ok {
+ return false
+ }
+ a.mu.Lock()
+ e := a.entries[host]
+ var rt registry.Route
+ if e != nil {
+ rt = e.route
+ }
+ a.mu.Unlock()
+ if e == nil {
+ return false
+ }
+ l, _ := c.Layout(rt)
+ return l == LayoutForeign
+}