amadan

carlos / platform Public

zdd plan 3 task 1: swap types, the SwapBackend seam, flat staged spawn without health, each artifact's tier, canonical link at the switch

Contract check (Step 1): every C1-C9 name present on the branch under the plan's name.
spawnInCgroupAtTier keeps the branch's three-result (settle) shape and strict-host gate.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e6227f40a4a377c477507edf8c1ca1f684376dc7 parent 412ca78
10 files changed, +741 −20
  • internal/activator/backend_exec.go +47 −14
  • internal/activator/containment.go +18 −6
  • internal/activator/genbackend.go +209 −0
  • internal/activator/genbackend_test.go +309 −0
  • internal/activator/generations.go +5 −0
  • internal/activator/staged.go +46 −0
  • internal/activator/swap.go +68 −0
  • internal/host/agent.go +5 −0
  • internal/host/memtier.go +10 −0
  • internal/host/memtier_test.go +24 −0
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index fa22b4cf1..99e2f193d 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -119,6 +119,11 @@ type ExecBackend struct {
// layout byte for byte: one cgroup per host and carlos-instance@. Set
// only by plan 5's CARLOS_SWAP wiring.
GenContainment bool
+ // TierOf is one named version's tier, read from that version's staged
+ // manifest (zero-downtime spec §7.3). A generation's limits are its own
+ // artifact's, never whatever the live link names at the moment. Nil
+ // falls back to Tier (tests, and every flag-off spawn).
+ TierOf func(rt registry.Route, version string) cgroup.Limit
// Logs, when non-nil, gets each instance and seed child's stdout/stderr
// teed in line-wise for `carlos logs` (spec §5) — journald keeps every
// byte it got before. Nil (tests, a box without the shipping loop)
@@ -1241,6 +1246,30 @@ func clearOwnGenSocket(sock string) error {
// n; 0 from Start). It only places the spawn's cgroup, and only under
// GenContainment (placeFor).
func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock string, n uint64) (Proc, string, error) {
+ p, spawnVer, err := b.spawnChildFrom(ctx, rt, sock, n, "", cgroup.Limit{})
+ if err != nil {
+ return p, "", err
+ }
+ if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil {
+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ defer cancel()
+ _ = p.Stop(stopCtx)
+ return nil, "", err
+ }
+ b.normalizeSocketMode(rt, sock, p)
+ return p, spawnVer, nil
+}
+
+// spawnChildFrom is the flat spawn with the executable and the tier named,
+// and nothing after the exec: it returns the moment the process runs. Empty
+// exe is the live link, which is every wake (startChildAt adds today's
+// health wait). A staged path is a swap's successor or a pinned wake
+// (zero-downtime spec §7.5 step 3): the bytes the attempt chose, whatever
+// root does to the live link meanwhile, with the caller owning the health
+// deadline and the termination. tier is that artifact's own tier; zero is
+// the live tier (b.tier). n is the generation number plan 2's containment
+// places the child by.
+func (b *ExecBackend) spawnChildFrom(ctx context.Context, rt registry.Route, sock string, n uint64, exe string, tier cgroup.Limit) (Proc, string, error) {
host := rt.Host
paths := b.pathsFor(rt)
dbPath := paths.dbPath
@@ -1252,9 +1281,12 @@ func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock
// Exec exactly what a systemd unit would: the per-instance live symlink
// (<OptDir>/live/<host> → a staged version), same --socket/--db flags,
// same EnvironmentFile.
- bin := b.liveLink(host)
- if err := b.ensureLiveLink(ctx, host); err != nil {
- return nil, "", err
+ bin := exe
+ if bin == "" {
+ bin = b.liveLink(host)
+ if err := b.ensureLiveLink(ctx, host); err != nil {
+ return nil, "", err
+ }
}
// Mint the instance's admin token BEFORE the env is read for the spawn, so
@@ -1354,13 +1386,20 @@ func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock
// the safe direction — we'd record the older version, so the sweep cycles
// once too often rather than never. Empty on a read error: the sweep won't
// cycle.
- spawnVer, _ := b.LiveVersion(host)
+ //
+ // A staged spawn names its build from the attempt; the caller fills
+ // Gen.Build. Reading the live link here would name whatever root
+ // installed, not what this process runs.
+ spawnVer := ""
+ if exe == "" {
+ spawnVer, _ = b.LiveVersion(host)
+ }
// Spawned through the cgroup manager (nil-tolerant: no delegation means
// a plain start, exactly as before) so the child lands in
// carlos-exec@<host> and the sampler can account it. build may run
// twice — the manager's clone3-refusal fallback — so everything about
// the Cmd is configured inside it.
- cmd, settle, err := b.spawnInCgroupAt(rt, child, ownTier, func() *exec.Cmd {
+ cmd, settle, err := b.spawnInCgroupAtTier(rt, child, ownTier, tier, func() *exec.Cmd {
cmd := exec.Command(bin, "--socket", sock, "--db", dbPath)
cmd.Env = childEnv(childVars)
// The child logs through the agent's own stderr/stdout (journald
@@ -1382,7 +1421,9 @@ func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock
}
p := &childProc{cmd: cmd, done: make(chan struct{}), ctl: ctl}
- if b.Cache != nil {
+ // A staged successor spawns while the incumbent still serves, and its
+ // purge happens at the switch (GenSwitched), not here.
+ if b.Cache != nil && exe == "" {
// A fresh process for this host is exactly the moment its cached
// content becomes suspect: a deploy/redeploy means new code and
// potentially new pages, and nothing else would invalidate them until
@@ -1411,14 +1452,6 @@ func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock
b.cleanupHost(host)
close(p.done)
}()
-
- if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil {
- stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
- defer cancel()
- _ = p.Stop(stopCtx)
- return nil, "", err
- }
- b.normalizeSocketMode(rt, sock, p)
return p, spawnVer, nil
}
diff --git a/internal/activator/containment.go b/internal/activator/containment.go
index ec0a824ec..6d0c0569f 100644
--- a/internal/activator/containment.go
+++ b/internal/activator/containment.go
@@ -57,8 +57,17 @@ func (b *ExecBackend) placeFor(kind spawnKind, n uint64) (child string, ownTier
return "", false, nil
}
-// spawnInCgroupAt is spawnInCgroup for a placed child. With no child it IS
-// spawnInCgroup, so a flag-off spawn is byte for byte today's.
+// spawnInCgroupAt is spawnInCgroupAtTier with the live tier: the spawn of a
+// placed child whose artifact is the one the live link names.
+func (b *ExecBackend) spawnInCgroupAt(rt registry.Route, child string, ownTier bool, build func() *exec.Cmd) (*exec.Cmd, func(), error) {
+ return b.spawnInCgroupAtTier(rt, child, ownTier, cgroup.Limit{}, build)
+}
+
+// spawnInCgroupAtTier is spawnInCgroup for a placed child, with the
+// artifact's tier named: the generation child's own limit, and the parent's
+// floor for StartChild (which never lowers a parent a swap has raised). Zero
+// tier is b.tier(rt). With no child it IS spawnInCgroup, so a flag-off spawn
+// is byte for byte today's.
//
// A placed spawn is never strict: strict containment of a generation layout
// is the claim path's (PrepareStrictParent, plan 5). It still honours the
@@ -66,17 +75,20 @@ func (b *ExecBackend) placeFor(kind spawnKind, n uint64) (child string, ownTier
// owner's group, and a host that is strict is refused rather than started
// outside its strict group. The returned settle is spawnInCgroup's: a no-op
// here. A spawn that fails removes the child it may have created.
-func (b *ExecBackend) spawnInCgroupAt(rt registry.Route, child string, ownTier bool, build func() *exec.Cmd) (*exec.Cmd, func(), error) {
+func (b *ExecBackend) spawnInCgroupAtTier(rt registry.Route, child string, ownTier bool, tier cgroup.Limit, build func() *exec.Cmd) (*exec.Cmd, func(), error) {
if child == "" {
return b.spawnInCgroup(rt, build)
}
start := func() (*exec.Cmd, func(), error) {
- tier := b.tier(rt)
+ t := tier
+ if t == (cgroup.Limit{}) {
+ t = b.tier(rt)
+ }
own := cgroup.Limit{}
if ownTier {
- own = tier
+ own = t
}
- cmd, err := b.Cgroups.StartChild(rt.Host, child, tier, own, build)
+ cmd, err := b.Cgroups.StartChild(rt.Host, child, t, own, build)
if err != nil {
b.Cgroups.RemoveChild(rt.Host, child)
return nil, nil, err
diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go
new file mode 100644
index 000000000..cfe75f52b
--- /dev/null
+++ b/internal/activator/genbackend.go
@@ -0,0 +1,209 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+var (
+ errSwapNoGens = errors.New("swap: backend cannot run staged generations")
+ errUnitNoGens = errors.New("swap: unit-backed routes keep stop-then-start")
+ errSwapLayout = errors.New("swap: this row's layout has no generation sockets")
+ errNestedStaged = errors.New("swap: nested staged generations need Task 9")
+)
+
+// SwapBackend is what a swap needs from a backend beyond Backend and plan
+// 1's GenBackend (spec §7.3–§7.5). ExecBackend implements it on top of plan
+// 1's generation spawn and plan 2's containment. SplitBackend forwards it
+// for exec rows and refuses unit rows, which keep stop-then-start (spec §2
+// decision 9).
+type SwapBackend interface {
+ // StartStaged spawns generation g.N of rt from g.StagedPath, never
+ // through the live link, in its own child cgroup at g.Tier, on the
+ // generation's own socket (a foreign row: its canonical socket). It
+ // returns as soon as the process runs, with g's Socket and Control
+ // filled in, and never waits for health: the caller checks health
+ // through that handle under its own deadline, and stops (or kills) a
+ // process that fails. A nil Proc means nothing ran.
+ StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error)
+ // GenTier is build's own tier, from its staged manifest (spec §7.3).
+ // The swap captures it into Gen.Tier at admission.
+ GenTier(rt registry.Route, build string) cgroup.Limit
+ // GenReady runs once a pinned wake's generation is healthy: it widens
+ // the socket's mode (issue #103), as the live-link spawn's health tail
+ // does. A swap's successor gets the same from GenSwitched.
+ GenReady(rt registry.Route, g Gen, p Proc)
+ // RaiseGenParent sets the host's cgroup parent to the incumbent's tier
+ // plus the successor's, before the successor spawns (spec §7.3).
+ RaiseGenParent(ctx context.Context, rt registry.Route, incumbent, successor cgroup.Limit) error
+ // RestoreGenParent sets the parent back to tier, the current
+ // generation's captured tier, once every other generation is empty.
+ RestoreGenParent(ctx context.Context, rt registry.Route, tier cgroup.Limit) error
+ // GenSwitched runs under the entry lock at the switch (spec §7.5 step
+ // 5), before the in-memory switch. It points the canonical path at the
+ // new generation's socket (rev 15), purges the host's cached responses
+ // and widens the new socket's mode. An error refuses the switch.
+ GenSwitched(rt registry.Route, g Gen, p Proc) error
+}
+
+// StartStaged is plan 1's StartGen with perGen true, except for the
+// executable, the tier and the health wait, which is the caller's. A flat
+// child execs g.StagedPath directly. A nested unit's executable mount names
+// it, which is Task 9's leg; until then a nested row is refused. A foreign
+// row keeps its canonical socket (plan 1): it gets one generation at a time
+// (a cold, pinned or hold start), and Swap never overlaps it.
+func (b *ExecBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) {
+ if g.Tier == (cgroup.Limit{}) {
+ g.Tier = b.GenTier(rt, g.Build)
+ }
+ p := b.pathsFor(rt)
+ switch p.layout {
+ case LayoutFlat:
+ sock := b.flatGenSocket(rt.Host, g.N)
+ if err := clearOwnGenSocket(sock); err != nil {
+ return g, nil, fmt.Errorf("%s: generation socket: %w", rt.Host, err)
+ }
+ proc, _, err := b.spawnChildFrom(ctx, rt, sock, g.N, g.StagedPath, g.Tier)
+ if err != nil {
+ return g, proc, err
+ }
+ g.Socket, g.Control = sock, sock+".control"
+ return g, proc, nil
+ case LayoutForeign:
+ // As StartGen does for a canonical-socket spawn: a link left there
+ // by an agent that ran with generations on would fail the app's bind.
+ if err := clearLeftoverCanonicalLink(rt.Socket); err != nil {
+ return g, nil, fmt.Errorf("%s: leftover canonical link: %w", rt.Host, err)
+ }
+ proc, _, err := b.spawnChildFrom(ctx, rt, rt.Socket, g.N, g.StagedPath, g.Tier)
+ if err != nil {
+ return g, proc, err
+ }
+ g.Socket, g.Control = rt.Socket, rt.Socket+".control"
+ return g, proc, nil
+ case LayoutNested:
+ return g, nil, errNestedStaged
+ case LayoutFenced, LayoutConflict:
+ return g, nil, fmt.Errorf("%s: refusing to spawn: %s", rt.Host, p.reason)
+ }
+ return g, nil, errSwapLayout
+}
+
+// GenTier is build's own tier: TierOf when it is wired (the agent reads
+// the build's staged manifest), the live tier otherwise.
+func (b *ExecBackend) GenTier(rt registry.Route, build string) cgroup.Limit {
+ if b.TierOf != nil && build != "" {
+ return b.TierOf(rt, build)
+ }
+ return b.tier(rt)
+}
+
+// GenReady widens a healthy pinned generation's socket mode (issue #103).
+func (b *ExecBackend) GenReady(rt registry.Route, g Gen, p Proc) {
+ b.normalizeSocketMode(rt, g.Socket, p)
+}
+
+// RaiseGenParent sets the parent to the incumbent's tier plus the
+// successor's (spec §7.3, plan 2 D3), each captured from its own artifact
+// at the attempt's admission, through plan 2's ParentLimit. That keeps the
+// parent's expected-limit record in step.
+func (b *ExecBackend) RaiseGenParent(ctx context.Context, rt registry.Route, incumbent, successor cgroup.Limit) error {
+ return b.ParentLimit(ctx, rt, cgroup.OverlapLimit(incumbent, successor))
+}
+
+// RestoreGenParent puts the parent back at the current generation's tier.
+func (b *ExecBackend) RestoreGenParent(ctx context.Context, rt registry.Route, tier cgroup.Limit) error {
+ return b.ParentLimit(ctx, rt, tier)
+}
+
+// GenSwitched points the canonical socket path at g's socket first, with
+// plan 1's atomic helper. A failure refuses the switch, because outside
+// dialers (#103) would otherwise be left on a process that is about to
+// drain. Then it drops the host's cached responses (plan 1's generation tag
+// already makes them miss; this reclaims the space) and widens the new
+// socket to 0660 (issue #103).
+//
+// Lock contract (plan 1, Task 4). The caller holds a.mu, the lock under
+// which plan 1 creates and removes the link, and sets e.gen to g in the
+// same critical section. So finishStop always removes the link that names
+// the generation it retires, and one entry's stop can never delete the
+// next entry's link.
+func (b *ExecBackend) GenSwitched(rt registry.Route, g Gen, p Proc) error {
+ if !linksCanonical(rt, g) {
+ return errSwapLayout
+ }
+ if err := swapCanonicalLink(rt.Socket, g.Socket); err != nil {
+ return fmt.Errorf("%s: canonical socket link: %w", rt.Host, err)
+ }
+ if b.Cache != nil {
+ if n := b.Cache.PurgeHost(rt.Host); n > 0 {
+ log.Printf("activator: purged %d cached response(s) for %s at the switch to generation %d", n, rt.Host, g.N)
+ }
+ }
+ b.normalizeSocketMode(rt, g.Socket, p)
+ return nil
+}
+
+// genExec picks the exec half for a swap operation. Unit rows refuse, and
+// so does an exec half that cannot run staged generations.
+func (b *SplitBackend) genExec(rt registry.Route) (SwapBackend, error) {
+ if rt.Backing == registry.BackingUnit {
+ return nil, errUnitNoGens
+ }
+ g, ok := b.Exec.(SwapBackend)
+ if !ok {
+ return nil, errSwapNoGens
+ }
+ return g, nil
+}
+
+func (b *SplitBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) {
+ gb, err := b.genExec(rt)
+ if err != nil {
+ return g, nil, err
+ }
+ return gb.StartStaged(ctx, rt, g)
+}
+
+func (b *SplitBackend) GenTier(rt registry.Route, build string) cgroup.Limit {
+ gb, err := b.genExec(rt)
+ if err != nil {
+ return cgroup.TierDefault
+ }
+ return gb.GenTier(rt, build)
+}
+
+func (b *SplitBackend) GenReady(rt registry.Route, g Gen, p Proc) {
+ if gb, err := b.genExec(rt); err == nil {
+ gb.GenReady(rt, g, p)
+ }
+}
+
+func (b *SplitBackend) RaiseGenParent(ctx context.Context, rt registry.Route, incumbent, successor cgroup.Limit) error {
+ gb, err := b.genExec(rt)
+ if err != nil {
+ return err
+ }
+ return gb.RaiseGenParent(ctx, rt, incumbent, successor)
+}
+
+func (b *SplitBackend) RestoreGenParent(ctx context.Context, rt registry.Route, tier cgroup.Limit) error {
+ gb, err := b.genExec(rt)
+ if err != nil {
+ return err
+ }
+ return gb.RestoreGenParent(ctx, rt, tier)
+}
+
+func (b *SplitBackend) GenSwitched(rt registry.Route, g Gen, p Proc) error {
+ gb, err := b.genExec(rt)
+ if err != nil {
+ return err
+ }
+ return gb.GenSwitched(rt, g, p)
+}
diff --git a/internal/activator/genbackend_test.go b/internal/activator/genbackend_test.go
new file mode 100644
index 000000000..0601823fd
--- /dev/null
+++ b/internal/activator/genbackend_test.go
@@ -0,0 +1,309 @@
+package activator
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// The attempt's chosen bytes are the bytes that run (spec §7.5 step 3,
+// §4.2): verifyStaged admits a regular file whose sha256 matches, and
+// refuses everything it cannot compare.
+func TestVerifyStaged(t *testing.T) {
+ dir := t.TempDir()
+ bin := filepath.Join(dir, "app")
+ if err := os.WriteFile(bin, []byte("build v2"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ sum := sha256.Sum256([]byte("build v2"))
+ good := hex.EncodeToString(sum[:])
+ if err := verifyStaged(bin, good); err != nil {
+ t.Fatalf("matching digest refused: %v", err)
+ }
+ if err := verifyStaged(bin, strings.ToUpper(good)); err != nil {
+ t.Fatalf("upper-case hex refused: %v", err)
+ }
+ link := filepath.Join(dir, "live")
+ if err := os.Symlink(bin, link); err != nil {
+ t.Fatal(err)
+ }
+ for name, tc := range map[string]struct{ path, digest string }{
+ "mismatch": {bin, strings.Repeat("0", 64)},
+ "empty digest": {bin, ""},
+ "short digest": {bin, "abc"},
+ "relative path": {"app", good},
+ "through a symlink": {link, good},
+ "missing": {filepath.Join(dir, "gone"), good},
+ "directory": {dir, good},
+ } {
+ if err := verifyStaged(tc.path, tc.digest); err == nil {
+ t.Errorf("%s: accepted", name)
+ }
+ }
+}
+
+// recordingGenExec is an exec backend that can run staged generations.
+type recordingGenExec struct {
+ *fakeBackend
+ started int
+}
+
+func (r *recordingGenExec) StartStaged(_ context.Context, _ registry.Route, g Gen) (Gen, Proc, error) {
+ r.started++
+ return g, &fakeProc{}, nil
+}
+func (r *recordingGenExec) GenTier(registry.Route, string) cgroup.Limit { return cgroup.TierDefault }
+func (r *recordingGenExec) GenReady(registry.Route, Gen, Proc) {}
+func (r *recordingGenExec) RaiseGenParent(context.Context, registry.Route, cgroup.Limit, cgroup.Limit) error {
+ return nil
+}
+func (r *recordingGenExec) RestoreGenParent(context.Context, registry.Route, cgroup.Limit) error {
+ return nil
+}
+func (r *recordingGenExec) GenSwitched(registry.Route, Gen, Proc) error { return nil }
+
+// Swaps are for exec rows only (spec §2 decision 9): a unit-backed row is
+// refused, and an exec half that cannot run staged generations refuses
+// rather than falling back to anything.
+func TestSplitBackendSwapsAreExecOnly(t *testing.T) {
+ ctx := context.Background()
+ exec := &recordingGenExec{fakeBackend: &fakeBackend{}}
+ s := &SplitBackend{Exec: exec, Unit: &UnitBackend{}}
+ unitRt := registry.Route{Host: "u.example.com", Backing: registry.BackingUnit}
+ if _, _, err := s.StartStaged(ctx, unitRt, Gen{N: 1}); !errors.Is(err, errUnitNoGens) {
+ t.Fatalf("unit row StartStaged: %v, want errUnitNoGens", err)
+ }
+ if err := s.RaiseGenParent(ctx, unitRt, cgroup.TierDefault, cgroup.TierDefault); !errors.Is(err, errUnitNoGens) {
+ t.Fatalf("unit row RaiseGenParent: %v", err)
+ }
+ execRt := registry.Route{Host: "e.example.com"}
+ g, _, err := s.StartStaged(ctx, execRt, Gen{N: 7, StagedPath: "/opt/carlos/versions/v2/app"})
+ if err != nil || g.N != 7 || exec.started != 1 {
+ t.Fatalf("exec row StartStaged = %+v, %v (started %d)", g, err, exec.started)
+ }
+ plain := &SplitBackend{Exec: &fakeBackend{}, Unit: &UnitBackend{}}
+ if _, _, err := plain.StartStaged(ctx, execRt, Gen{N: 1}); !errors.Is(err, errSwapNoGens) {
+ t.Fatalf("exec half without staged generations: %v, want errSwapNoGens", err)
+ }
+}
+
+// A flat staged generation execs the attempt's staged path, not the live
+// link, on its own generation socket (spec §7.5 step 3). The fake binary
+// never binds. What it recorded as argv[0] is the proof.
+func TestExecBackendStartStagedExecsTheStagedPath(t *testing.T) {
+ b := testExecBackend(t)
+ b.RunDirOverride = t.TempDir()
+ host := "staged.example.com"
+ stageLive(t, b, host, "v1") // the live link names v1 throughout
+ marker := filepath.Join(t.TempDir(), "argv0")
+ v2 := filepath.Join(b.OptDir, "versions", "v2", "app")
+ if err := os.MkdirAll(filepath.Dir(v2), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(v2, []byte("#!/bin/sh\nprintf '%s' \"$0\" > '"+marker+"'\n"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ rt := registry.Route{Host: host, Kind: "instance", DBPath: filepath.Join(b.DataDir, host+".db"),
+ Socket: filepath.Join(b.RunDirOverride, host+".sock")} // flat under the override
+ _, proc, err := b.StartStaged(context.Background(), rt, Gen{N: 5, Build: "v2", StagedPath: v2})
+ if err != nil {
+ t.Fatalf("StartStaged: %v", err)
+ }
+ defer proc.Stop(context.Background()) //nolint:errcheck
+ if got := waitForFile(t, marker, 2*time.Second); got != v2 {
+ t.Fatalf("the generation execd %q, want the staged %q", got, v2)
+ }
+}
+
+// StartStaged is a spawn, not a health check (spec §8.2): it hands back the
+// running process at once, so the caller's own deadline governs and a build
+// that never answers is the caller's to record and kill. This runs the real
+// flat spawn with the real 30 s spawnHealthTimeout. A StartStaged that still
+// waits for health inside blocks for those 30 s and then returns a nil
+// handle, which a planned wake reads as "nothing executed".
+func TestExecBackendStartStagedReturnsTheProcessAtOnce(t *testing.T) {
+ b := testExecBackend(t)
+ b.RunDirOverride = t.TempDir()
+ host := "silent.example.com"
+ stageLive(t, b, host, "v1")
+ v2 := filepath.Join(b.OptDir, "versions", "v2", "app")
+ if err := os.MkdirAll(filepath.Dir(v2), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(v2, []byte("#!/bin/sh\nexec sleep 30\n"), 0o755); err != nil { // never binds
+ t.Fatal(err)
+ }
+ rt := registry.Route{Host: host, Kind: "instance", DBPath: filepath.Join(b.DataDir, host+".db"),
+ Socket: filepath.Join(b.RunDirOverride, host+".sock")}
+ type out struct {
+ g Gen
+ proc Proc
+ err error
+ }
+ done := make(chan out, 1)
+ go func() {
+ g, p, err := b.StartStaged(context.Background(), rt, Gen{N: 6, Build: "v2", StagedPath: v2})
+ done <- out{g, p, err}
+ }()
+ var o out
+ select {
+ case o = <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("StartStaged waited for health: it must return the running process at once")
+ }
+ if o.err != nil || o.proc == nil {
+ t.Fatalf("StartStaged = %v, %v: want the running process", o.proc, o.err)
+ }
+ defer o.proc.Stop(context.Background()) //nolint:errcheck
+ if o.proc.Exited() {
+ t.Fatal("the returned process has already exited")
+ }
+ if o.g.Socket != b.flatGenSocket(host, 6) {
+ t.Fatalf("Socket = %q, want generation 6's", o.g.Socket)
+ }
+}
+
+// A foreign row keeps its canonical socket (plan 1): a cold, pinned or hold
+// start execs the staged build there, one generation at a time. Refusing it
+// would break every planned wake of a foreign row once the flag is on.
+func TestExecBackendStartStagedRunsAForeignRowOnItsCanonicalSocket(t *testing.T) {
+ b := testExecBackend(t)
+ host := "f.example.com"
+ marker := filepath.Join(t.TempDir(), "argv0")
+ v2 := filepath.Join(b.OptDir, "versions", "v2", "app")
+ if err := os.MkdirAll(filepath.Dir(v2), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(v2, []byte("#!/bin/sh\nprintf '%s' \"$0\" > '"+marker+"'\n"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ rt := registry.Route{Host: host, Kind: "instance", DBPath: filepath.Join(b.DataDir, host+".db"),
+ Socket: filepath.Join(t.TempDir(), "odd.sock")} // hand-set: LayoutForeign
+ if l, _ := b.Layout(rt); l != LayoutForeign {
+ t.Fatalf("fixture layout = %v, want LayoutForeign", l)
+ }
+ g, proc, err := b.StartStaged(context.Background(), rt, Gen{N: 1, Build: "v2", StagedPath: v2})
+ if err != nil {
+ t.Fatalf("foreign row: %v", err)
+ }
+ defer proc.Stop(context.Background()) //nolint:errcheck
+ if g.Socket != rt.Socket || g.Control != rt.Socket+".control" {
+ t.Fatalf("foreign generation sockets %q %q, want the canonical %q", g.Socket, g.Control, rt.Socket)
+ }
+ if got := waitForFile(t, marker, 2*time.Second); got != v2 {
+ t.Fatalf("the generation execd %q, want the staged %q", got, v2)
+ }
+}
+
+// A link left at a foreign row's canonical path by an agent that ran with
+// generations on would fail the app's bind, so a staged start there clears
+// it first, as StartGen does.
+func TestExecBackendStartStagedClearsALeftoverCanonicalLink(t *testing.T) {
+ b := testExecBackend(t)
+ host := "l.example.com"
+ v2 := filepath.Join(b.OptDir, "versions", "v2", "app")
+ if err := os.MkdirAll(filepath.Dir(v2), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(v2, []byte("#!/bin/sh\nexec sleep 30\n"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ rt := registry.Route{Host: host, Kind: "instance", DBPath: filepath.Join(b.DataDir, host+".db"),
+ Socket: filepath.Join(t.TempDir(), "odd.sock")}
+ if err := os.Symlink("/nonexistent/g/old.sock", rt.Socket); err != nil {
+ t.Fatal(err)
+ }
+ _, proc, err := b.StartStaged(context.Background(), rt, Gen{N: 2, Build: "v2", StagedPath: v2})
+ if err != nil {
+ t.Fatalf("foreign row: %v", err)
+ }
+ defer proc.Stop(context.Background()) //nolint:errcheck
+ if _, err := os.Lstat(rt.Socket); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the leftover link at the canonical path survived: %v", err)
+ }
+}
+
+// Each artifact's own tier (spec §7.3): a staged child gets g.Tier, not the
+// live link's tier, and the overlap parent is the incumbent's tier plus the
+// successor's. Here the live link (b.Tier) says 512 MiB throughout, while
+// the incumbent declares children (1 GiB) and the successor does not. Using
+// b.tier(rt) for both terms gives a 1 GiB parent, and an incumbent that
+// fills its own 1 GiB is OOM-killed by the successor's first allocation.
+func TestExecBackendUsesEachArtifactsTier(t *testing.T) {
+ const host = "t.example.com"
+ b, parent := delegatedBackend(t, host)
+ b.GenContainment = true
+ b.Tier = func(registry.Route) cgroup.Limit { return cgroup.TierDefault }
+ b.TierOf = func(_ registry.Route, v string) cgroup.Limit {
+ if v == "v1" {
+ return cgroup.TierChildren
+ }
+ return cgroup.TierDefault
+ }
+ rt := registry.Route{Host: host}
+ if got := b.GenTier(rt, "v1"); got != cgroup.TierChildren {
+ t.Fatalf("GenTier(v1) = %v, want the children tier", got)
+ }
+ if err := b.RaiseGenParent(context.Background(), rt, b.GenTier(rt, "v1"), b.GenTier(rt, "v2")); err != nil {
+ t.Fatal(err)
+ }
+ if got, err := os.ReadFile(filepath.Join(parent, "memory.max")); err != nil || string(got) != "1610612736" {
+ t.Fatalf("overlap parent memory.max = %q, %v; want 1 GiB + 512 MiB", got, err)
+ }
+ cmd, settle, err := b.spawnInCgroupAtTier(rt, "g9", true, cgroup.TierChildren, func() *exec.Cmd { return exec.Command("/bin/true") })
+ if err != nil {
+ t.Fatal(err)
+ }
+ _ = cmd.Wait()
+ settle()
+ if got, err := os.ReadFile(filepath.Join(parent, "g9", "memory.max")); err != nil || string(got) != "1073741824" {
+ t.Fatalf("g9 memory.max = %q, %v; want its own artifact's 1 GiB", got, err)
+ }
+ if err := b.RestoreGenParent(context.Background(), rt, cgroup.TierChildren); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := os.ReadFile(filepath.Join(parent, "memory.max")); string(got) != "1073741824" {
+ t.Fatalf("restored parent memory.max = %q, want the current generation's 1 GiB", got)
+ }
+}
+
+// The switch points the canonical path at the new generation's socket
+// (spec §7.1, rev 15), for dialers outside the agent (#103).
+func TestExecBackendGenSwitchedRepointsTheCanonicalLink(t *testing.T) {
+ b := testExecBackend(t)
+ dir := t.TempDir()
+ rt := registry.Route{Host: "c.example.com", Socket: filepath.Join(dir, "c.example.com.sock")}
+ for _, n := range []uint64{1, 2} {
+ g := Gen{N: n, Socket: filepath.Join(dir, "g", "c."+strings.Repeat("x", int(n))+".sock")}
+ if err := b.GenSwitched(rt, g, &fakeProc{}); err != nil {
+ t.Fatal(err)
+ }
+ if got, err := os.Readlink(rt.Socket); err != nil || got != g.Socket {
+ t.Fatalf("after switch %d the canonical path names %q (%v), want %q", n, got, err, g.Socket)
+ }
+ }
+}
+
+// Without plan 2's per-generation containment there is no parent to raise,
+// and the swap is refused before anything spawns.
+func TestExecBackendRaiseNeedsContainment(t *testing.T) {
+ b := testExecBackend(t)
+ if err := b.RaiseGenParent(context.Background(), registry.Route{Host: "h.example.com"}, cgroup.TierDefault, cgroup.TierDefault); !errors.Is(err, errNoGenContainment) {
+ t.Fatalf("RaiseGenParent without GenContainment: %v", err)
+ }
+}
+
+var (
+ _ SwapBackend = (*ExecBackend)(nil)
+ _ SwapBackend = (*SplitBackend)(nil)
+)
diff --git a/internal/activator/generations.go b/internal/activator/generations.go
index 2de681242..ceba2c8dd 100644
--- a/internal/activator/generations.go
+++ b/internal/activator/generations.go
@@ -12,6 +12,7 @@ import (
"strconv"
"sync/atomic"
+ "github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
)
@@ -55,6 +56,10 @@ type Gen struct {
StagedPath string // immutable staged executable it exec'd (spec §7.5 step 3)
Digest string // sha256 of StagedPath at exec, hex
Demand string // the release's own `swap` demand verified for this artifact: "hold" or "" (spec §4.2, §8.5)
+ // Tier is the generation's own artifact's tier (zero-downtime spec
+ // §7.3), captured when an attempt is admitted (plan 3). Zero for a
+ // live-link wake until a swap captures it from Build.
+ Tier cgroup.Limit
}
// GenSocketDir is where flat generation sockets live: /run/carlos/g.
diff --git a/internal/activator/staged.go b/internal/activator/staged.go
new file mode 100644
index 000000000..f1c806ef8
--- /dev/null
+++ b/internal/activator/staged.go
@@ -0,0 +1,46 @@
+package activator
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "strings"
+)
+
+// verifyStaged checks, immediately before anything execs it, that path is
+// an absolute regular file (never a link) whose sha256 is digest (spec §7.5
+// step 3, §4.2). A staged copy is immutable once something refers to it
+// (plan 4 refuses to restage over one), so this is the check that the bytes
+// the attempt chose are the bytes on disk. An artifact that cannot be
+// compared never runs: an empty or malformed digest refuses.
+func verifyStaged(path, digest string) error {
+ if !filepath.IsAbs(path) {
+ return fmt.Errorf("staged path %q is not absolute", path)
+ }
+ if len(digest) != 2*sha256.Size {
+ return fmt.Errorf("staged digest %q is not a sha256", digest)
+ }
+ fi, err := os.Lstat(path)
+ if err != nil {
+ return err
+ }
+ if !fi.Mode().IsRegular() {
+ return fmt.Errorf("staged path %q is not a regular file", path)
+ }
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ h := sha256.New()
+ if _, err := io.Copy(h, f); err != nil {
+ return err
+ }
+ if got := hex.EncodeToString(h.Sum(nil)); got != strings.ToLower(digest) {
+ return fmt.Errorf("staged digest mismatch: have %s, want %s", got, strings.ToLower(digest))
+ }
+ return nil
+}
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
new file mode 100644
index 000000000..beb1a58f5
--- /dev/null
+++ b/internal/activator/swap.go
@@ -0,0 +1,68 @@
+// swap.go is the overlap and hold swap (zero-downtime deploys spec §7.4,
+// §7.5, §7.8): one attempt moves one awake instance to a new generation.
+// Nothing calls Service.Swap until plan 5 wires CARLOS_SWAP. With plan 1's
+// generations switch off, it refuses every attempt and spawns nothing.
+package activator
+
+import "time"
+
+// Cause is why an attempt exists (spec §6.3).
+type Cause string
+
+const (
+ CauseRelease Cause = "release"
+ CauseInstall Cause = "install"
+ CauseRetry Cause = "retry"
+ CauseRestart Cause = "restart"
+ CauseConfig Cause = "config"
+)
+
+// SwapMode is how an awake instance moves to a new generation (spec §4.1).
+type SwapMode string
+
+const (
+ ModeOverlap SwapMode = "overlap"
+ ModeHold SwapMode = "hold"
+)
+
+// Attempt is one try at moving one instance to a new generation (spec §6.3).
+type Attempt struct {
+ Host string
+ Cause Cause
+ Build string
+ StagedPath string
+ Digest string // expected sha256 of StagedPath, verified before exec
+ Demand string // release demand for the artifact: "hold" or ""
+ Mode SwapMode
+ ConfigRev string
+ RestartID string
+ SchedRev uint64 // scheduler revision; the newest-attempt check compares it
+ Canary bool // observe the 10 s settle
+}
+
+// SwapOutcome is how an attempt ended.
+type SwapOutcome string
+
+const (
+ SwapDone SwapOutcome = "done" // switched (overlap) or replaced (hold)
+ SwapFailed SwapOutcome = "failed" // successor failed health or settle; incumbent still serving (overlap)
+ SwapAbandoned SwapOutcome = "abandoned" // superseded or a re-check failed; not a build failure
+ SwapDown SwapOutcome = "down" // hold mode failed: nothing serving
+)
+
+// SwapResult is Swap's answer. Reason is diagnostic text for the
+// scheduler (plan 5); member wording is plan 6's.
+type SwapResult struct {
+ Outcome SwapOutcome
+ Gen Gen // the generation now current (on SwapDone)
+ Reason string
+ // Drain is how long the retiring generation took from SIGTERM to its
+ // proven exit (an overlap), or the old generation's stop (a hold). Zero
+ // when nothing retired. The "swapped" event reports it (spec §9.4).
+ Drain time.Duration
+ // Mode is the mode the attempt actually ran in. It differs from
+ // Attempt.Mode when an overlap attempt on a foreign-layout row ran as a
+ // hold (spec §2.9): the scheduler records a failure under this mode, so
+ // a failed hold is held down (spec §6.4, §8.3). Swap sets it.
+ Mode SwapMode
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index c3afd0949..738c89d71 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -336,6 +336,11 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// staged manifest of the route's live version (memory-pressure
// spec §2.3): 512M, or 1G for a family that declares children.
Tier: memTier(stagedManifestOfLiveVersion(release.OptDir())),
+ // The same read for one named version, for a generation that runs
+ // a build other than the live link's (zero-downtime spec §7.3).
+ TierOf: memTierOf(func(rt registry.Route, v string) (*release.Manifest, error) {
+ return release.ReadStagedManifest(release.OptDir(), release.RouteAccount(rt), v, rt.App)
+ }),
// The scheduled-work verbs on every spawned instance's control
// socket write to the SAME registry this agent's converger and
// runner read (schedule spec §7.2): an app that asks for a
diff --git a/internal/host/memtier.go b/internal/host/memtier.go
index 5c877b777..acc2ffe59 100644
--- a/internal/host/memtier.go
+++ b/internal/host/memtier.go
@@ -24,3 +24,13 @@ func memTier(load func(registry.Route) (*release.Manifest, error)) func(registry
return cgroup.TierChildren
}
}
+
+// memTierOf is memTier for one named version: that version's staged
+// manifest, never the live link's (zero-downtime spec §7.3: a generation's
+// limits are its own artifact's). A staged copy is immutable once anything
+// refers to it (plan 4), so the answer for a version does not move.
+func memTierOf(load func(rt registry.Route, version string) (*release.Manifest, error)) func(registry.Route, string) cgroup.Limit {
+ return func(rt registry.Route, version string) cgroup.Limit {
+ return memTier(func(rt registry.Route) (*release.Manifest, error) { return load(rt, version) })(rt)
+ }
+}
diff --git a/internal/host/memtier_test.go b/internal/host/memtier_test.go
index d193bb7fd..201a4e721 100644
--- a/internal/host/memtier_test.go
+++ b/internal/host/memtier_test.go
@@ -2,6 +2,7 @@ package host
import (
"errors"
+ "strings"
"testing"
"github.com/carlosframework/platform/internal/caps"
@@ -36,3 +37,26 @@ func TestMemTier(t *testing.T) {
}
}
}
+
+// memTierOf reads the NAMED version's manifest, never the live link's: a
+// generation's limits are its own artifact's (zero-downtime spec §7.3).
+func TestMemTierOfReadsTheNamedVersion(t *testing.T) {
+ rt := registry.Route{Host: "h.example.com", App: "app"}
+ var asked []string
+ tier := memTierOf(func(_ registry.Route, v string) (*release.Manifest, error) {
+ asked = append(asked, v)
+ if v == "v1" {
+ return &release.Manifest{Capabilities: &caps.Set{V: caps.Version, Children: true}}, nil
+ }
+ return &release.Manifest{}, nil
+ })
+ if got := tier(rt, "v1"); got != cgroup.TierChildren {
+ t.Errorf("v1 = %v, want the children tier", got)
+ }
+ if got := tier(rt, "v2"); got != cgroup.TierDefault {
+ t.Errorf("v2 = %v, want the default tier", got)
+ }
+ if strings.Join(asked, ",") != "v1,v2" {
+ t.Errorf("read the manifests of %v, want v1 then v2", asked)
+ }
+}