From 68a2ebf4b1fc2e7987c3348234aac1890ae8bfa8 Mon Sep 17 00:00:00 2001 From: mintaka Date: Wed, 26 Aug 2026 00:09:11 -0400 Subject: [PATCH] feat(stack): pgid-record v2 kind-tagged union for container teardown (RIG-2760, T8a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Grow the stack.pgids entry line into a kind-tagged discriminated union so a container-backed child (S4's rootless-podman postgres) can be torn down by its stable name, alongside the existing process-group children torn down by identity-checked pgid. - v2 entry grammar: 'proc ' | 'ctr '. A v1 record (untagged 3-field proc lines) still parses, as all-process entries (v2 is a strict superset). - Cross-version safety by two guards (DL-262): forward — a v2 reader hard-errors on an unknown/newer version; reverse — a shipped v1 parser meeting a v2 record hard-errors by the entry grammar (4-field proc / unknown ctr token), never half-parsing a record it cannot fully understand. - ContainerController Deps seam (Exists/Stop/Remove) mirrors GroupSignaller: DownDetached dispatches per entry kind (group signal vs podman stop/rm -f). Nil until T8 wires the real podman adapter; no container entry is written yet, so the seam is never dereferenced. Realizes DL-262 (pgid-record v2). The container write path + podman adapter are T8's slice. Co-authored-by: Matt Wilkinson Ledger-impact: none (realizes frozen DL-262; no ledger row change) --- go/internal/stack/deps.go | 25 +++ go/internal/stack/downdetached.go | 74 ++++++-- go/internal/stack/downdetached_test.go | 243 +++++++++++++++++++++++++ go/internal/stack/harness_test.go | 62 ++++++- go/internal/stack/pgidfile.go | 177 +++++++++++++++--- go/internal/stack/pgidfile_test.go | 123 ++++++++++++- go/internal/stack/stack.go | 21 ++- 7 files changed, 681 insertions(+), 44 deletions(-) diff --git a/go/internal/stack/deps.go b/go/internal/stack/deps.go index 8de4bf52..fb220ca8 100644 --- a/go/internal/stack/deps.go +++ b/go/internal/stack/deps.go @@ -35,6 +35,12 @@ type Deps struct { // no Process handle for); the real adapter targets the negative pgid, the // same primitive the in-process escalation uses. GroupSignaller GroupSignaller + // Containers tears down container children by their stable name for the + // cross-process teardown (DownDetached), the container analogue of + // GroupSignaller. Nil until the container-backed postgres adapter (T8) wires + // the real podman-exec adapter; a record with no container entries never + // dereferences it. + Containers ContainerController // Now is the clock the cert-expiry math reads. Nil defaults to time.Now. Now func() time.Time @@ -133,6 +139,25 @@ type GroupSignaller interface { Alive(pgid int, startTime uint64) bool } +// ContainerController tears down a container child by its stable name for the +// cross-process teardown (DownDetached), the container analogue of +// GroupSignaller: DownDetached reads container names from the state-dir record +// and drives them here, since the tearing process holds no handle for a +// container a prior up ran. It is the only seam that touches containers this +// process did not start. +// +// Exists reports whether a container with this name is present (the real adapter +// runs `podman container exists `) — the liveness channel, the container +// analogue of GroupSignaller.Alive; a container needs no start-time identity +// token because its name is unique per state dir (S4). Stop requests a graceful +// stop bounded by timeout (`podman stop -t `); Remove is the +// SIGKILL-tier escalation that force-removes it (`podman rm -f `). +type ContainerController interface { + Exists(name string) bool + Stop(name string, timeout time.Duration) error + Remove(name string) error +} + // CertEnsurer ensures the TLS anchor (one PEM that is both the server's // --tls-cert and the runner's --ca) exists under stateDir and is valid well past // now. It is expiry-aware, not skip-if-present: when the existing anchor's diff --git a/go/internal/stack/downdetached.go b/go/internal/stack/downdetached.go index 77c53420..a3bbd95c 100644 --- a/go/internal/stack/downdetached.go +++ b/go/internal/stack/downdetached.go @@ -197,7 +197,7 @@ func liveTargets(ctx context.Context, cfg Config, deps Deps, rec pgidRecord) []t if !ok { continue // never recorded (half-spawned prefix) — nothing to tear down } - if !deps.GroupSignaller.Alive(e.Pgid, e.StartTime) { + if !entryAlive(deps, e) { continue // gone or recycled — skip, never signal } targets = append(targets, target{entry: e, budget: o.budget, confirm: o.confirm(e)}) @@ -218,12 +218,7 @@ func drainTargets(ctx context.Context, deps Deps, targets []target) []Component // intentionally not fatal here (an ESRCH means the group vanished in the // irreducible verify→signal gap, which the confirm reads as dead). for _, t := range targets { - if err := deps.GroupSignaller.Signal(t.entry.Pgid, SignalTerm); err != nil { - // Not actionable: delivery is not proof of death, and death is not - // proof of failure; the confirm channel decides. Recorded only for the - // operator's stderr, never used as the teardown verdict. - logSignalMiss("SIGTERM", t.entry, err) - } + signalTerm(deps, t.entry, t.budget) } // Phase B: per-target confirm with bounded SIGKILL escalation. @@ -253,11 +248,10 @@ func drainOne(ctx context.Context, deps Deps, t target) bool { return true // SIGTERM sufficed (or the group was already gone) } - // Escalate: hard-kill the whole group. A delivery error is not the verdict - // (an ESRCH means it died during the drain); the confirm below decides. - if err := deps.GroupSignaller.Signal(t.entry.Pgid, SignalKill); err != nil { - logSignalMiss("SIGKILL", t.entry, err) - } + // Escalate: hard-kill. For a process group this is a group SIGKILL; for a + // container it is `podman rm -f`. A delivery error is not the verdict (an + // ESRCH / already-gone means it died during the drain); the confirm decides. + signalKill(deps, t.entry) if t.entry.Component == ComponentRunner { // Socketless + SIGKILL unblockable → any residual non-ESRCH group is a @@ -334,3 +328,59 @@ func logSignalMiss(sig string, e pgidEntry, err error) { slog.Debug("group signal not delivered (group likely already gone)", "signal", sig, "component", e.Component.String(), "pgid", e.Pgid, "error", err) } + +// entryAlive reports whether a recorded entry's target is still live, dispatched +// on kind: a process group by identity-checked pgid (existence AND leader +// start-time), a container by name via `podman container exists`. A gone target +// reports not-alive so it is skipped, never signaled. +func entryAlive(deps Deps, e pgidEntry) bool { + switch e.Kind { + case entryContainer: + return deps.Containers.Exists(e.ContainerName) + default: + return deps.GroupSignaller.Alive(e.Pgid, e.StartTime) + } +} + +// signalTerm delivers the graceful-stop tier, dispatched on kind: a group +// SIGTERM for a process, `podman stop -t ` for a container (the budget +// is the container's own drain budget, deliberate parity with the process +// model's capped drain). A delivery error is not the teardown verdict — the +// per-component confirm channel is — so it is logged, never fatal. +func signalTerm(deps Deps, e pgidEntry, budget time.Duration) { + switch e.Kind { + case entryContainer: + if err := deps.Containers.Stop(e.ContainerName, budget); err != nil { + logContainerSignalMiss("stop", e, err) + } + default: + if err := deps.GroupSignaller.Signal(e.Pgid, SignalTerm); err != nil { + logSignalMiss("SIGTERM", e, err) + } + } +} + +// signalKill delivers the hard-kill tier, dispatched on kind: a group SIGKILL +// for a process, `podman rm -f` for a container. A delivery error is not the +// teardown verdict — the per-component confirm decides — so it is logged. +func signalKill(deps Deps, e pgidEntry) { + switch e.Kind { + case entryContainer: + if err := deps.Containers.Remove(e.ContainerName); err != nil { + logContainerSignalMiss("rm -f", e, err) + } + default: + if err := deps.GroupSignaller.Signal(e.Pgid, SignalKill); err != nil { + logSignalMiss("SIGKILL", e, err) + } + } +} + +// logContainerSignalMiss is the container analogue of logSignalMiss: a podman +// teardown-delivery error (most often "no such container": it went away in the +// verify→signal gap) is an expected, benign event, logged at debug rather than +// surfaced, since the socket-quiescence confirm is the verdict. +func logContainerSignalMiss(op string, e pgidEntry, err error) { + slog.Debug("container teardown not delivered (container likely already gone)", + "op", op, "component", e.Component.String(), "container", e.ContainerName, "error", err) +} diff --git a/go/internal/stack/downdetached_test.go b/go/internal/stack/downdetached_test.go index 1a6a45b6..c2dd1e3a 100644 --- a/go/internal/stack/downdetached_test.go +++ b/go/internal/stack/downdetached_test.go @@ -321,6 +321,38 @@ func TestDownDetachedHalfSpawnedPrefix(t *testing.T) { assertPgidFileGone(t, cfg.StateDir) } +// TestDownDetachedNilContainersAllProcessRecord pins the nil-safety contract +// documented in deps.go: a record with no container entries never dereferences +// deps.Containers. Production wiring leaves Deps.Containers nil, so an all-process +// record must tear down cleanly without a nil-pointer panic — the container path +// is gated on `case entryContainer` and never fires here. +func TestDownDetachedNilContainersAllProcessRecord(t *testing.T) { + cfg, h := newHarness(t) + deps := downTestDeps(t, h) + // Mirror production wiring: no container controller is provided. + deps.Containers = nil + + rec := pgidRecord{ + WriterPid: 1, Version: pgidFileVersion, + Entries: []pgidEntry{ + {Component: ComponentPostgres, Pgid: pgPgid, StartTime: pgToken(pgPgid)}, + {Component: ComponentServer, Pgid: serverPgid, StartTime: pgToken(serverPgid)}, + }, + } + if err := writePgidFile(cfg.StateDir, rec); err != nil { + t.Fatalf("seed all-process record = %v", err) + } + h.groupSig.set(pgPgid, pgToken(pgPgid), true) + h.groupSig.onTerm[pgPgid] = func() { h.groupSig.set(pgPgid, pgToken(pgPgid), false) } + h.groupSig.set(serverPgid, pgToken(serverPgid), true) + h.groupSig.onTerm[serverPgid] = func() { h.groupSig.set(serverPgid, pgToken(serverPgid), false) } + + if err := DownDetached(context.Background(), cfg, deps); err != nil { + t.Fatalf("DownDetached with nil Containers on an all-process record = %v, want nil (no deref, clean teardown)", err) + } + assertPgidFileGone(t, cfg.StateDir) +} + // TestDownDetachedAbsentFileNoSocketIsNoStack proves the "no stack" branch: // absent pgid file and no answering socket → nil, no error, no signals. func TestDownDetachedAbsentFileNoSocketIsNoStack(t *testing.T) { @@ -402,6 +434,176 @@ func TestDownDetachedConcurrentSerializedByGuard(t *testing.T) { assertPgidFileGone(t, cfg.StateDir) } +// The stable container name a v2 postgres entry carries in these tests. +const pgContainerName = "compass-postgres-test01" + +// seedContainerRecord writes a v2 record whose postgres entry is a container +// (ctr) and whose server/runner entries are processes, and marks all three live +// (the container present, the two groups alive + identity-matched). +func seedContainerRecord(t *testing.T, cfg Config, h *harness) { + t.Helper() + rec := pgidRecord{ + WriterPid: 4242, + Version: pgidFileVersion, + Entries: []pgidEntry{ + {Kind: entryContainer, Component: ComponentPostgres, ContainerName: pgContainerName}, + {Kind: entryProc, Component: ComponentServer, Pgid: serverPgid, StartTime: pgToken(serverPgid)}, + {Kind: entryProc, Component: ComponentRunner, Pgid: runnerPgid, StartTime: pgToken(runnerPgid)}, + }, + } + if err := writePgidFile(cfg.StateDir, rec); err != nil { + t.Fatalf("seed container record = %v", err) + } + h.containers.setExists(true) + h.groupSig.set(serverPgid, pgToken(serverPgid), true) + h.groupSig.set(runnerPgid, pgToken(runnerPgid), true) +} + +// containerBackedDBProber answers the postgres-reachability probe iff the +// container is still present — the socket goes dark when the container stops +// (the socket dir is bind-mounted, so the confirm channel is unchanged; only the +// signal-delivery side differs from the process path). +type containerBackedDBProber struct { + c *fakeContainerController + name string +} + +func (p *containerBackedDBProber) ProbeDB(ctx context.Context, dsn string) error { + if p.c.Exists(p.name) { + return nil // container up → socket answers → reachable + } + return errPostgresNotReady // container gone → socket dark → confirmed dead +} + +// containerDownDeps points the server confirm at the group signaller (as the +// process path) and the postgres confirm at the container's existence. +func containerDownDeps(t *testing.T, h *harness) Deps { + t.Helper() + shrinkBudgets(t) + deps := h.deps + deps.Prober = &groupBackedProber{gs: h.groupSig, pgid: serverPgid, token: pgToken(serverPgid)} + deps.DBProber = &containerBackedDBProber{c: h.containers, name: pgContainerName} + return deps +} + +// ctrEvents keeps only container stop/rm events in order. +func ctrEvents(events []string) []string { + var out []string + for _, e := range events { + if len(e) >= 4 && e[:4] == "ctr-" { + out = append(out, e) + } + } + return out +} + +// TestDownDetachedContainerGracefulStop proves the container teardown path: a +// container postgres entry is torn down by `podman stop` (graceful), confirmed +// by socket quiescence, with no `rm -f` escalation — the container analogue of +// the reverse-order SIGTERM happy path. +func TestDownDetachedContainerGracefulStop(t *testing.T) { + cfg, h := newHarness(t) + seedContainerRecord(t, cfg, h) + deps := containerDownDeps(t, h) + + // SIGTERM tears the two groups down; `podman stop` removes the container. + h.groupSig.onTerm[serverPgid] = func() { h.groupSig.set(serverPgid, pgToken(serverPgid), false) } + h.groupSig.onTerm[runnerPgid] = func() { h.groupSig.set(runnerPgid, pgToken(runnerPgid), false) } + h.containers.onStop[pgContainerName] = func() { h.containers.setExists(false) } + + if err := DownDetached(context.Background(), cfg, deps); err != nil { + t.Fatalf("DownDetached = %v, want nil", err) + } + + got := ctrEvents(h.rec.snapshot()) + want := []string{"ctr-stop " + pgContainerName} + if !reflect.DeepEqual(got, want) { + t.Fatalf("container teardown:\n got %v\n want %v (graceful stop, no rm -f)", got, want) + } + // The two process groups were still signaled by group SIGTERM. + sig := signalEvents(h.rec.snapshot()) + for _, pgid := range []int{serverPgid, runnerPgid} { + if countEvent(sig, "group-term "+strconv.Itoa(pgid)) != 1 { + t.Fatalf("process group %d should be SIGTERMed once: %v", pgid, sig) + } + } + assertPgidFileGone(t, cfg.StateDir) +} + +// TestDownDetachedContainerEscalatesToRemove proves the container SIGKILL tier: a +// container that survives `podman stop` is force-removed by `podman rm -f`, then +// (socket dark) confirmed — the container analogue of the group-SIGKILL +// escalation. +func TestDownDetachedContainerEscalatesToRemove(t *testing.T) { + cfg, h := newHarness(t) + seedContainerRecord(t, cfg, h) + deps := containerDownDeps(t, h) + + h.groupSig.onTerm[serverPgid] = func() { h.groupSig.set(serverPgid, pgToken(serverPgid), false) } + h.groupSig.onTerm[runnerPgid] = func() { h.groupSig.set(runnerPgid, pgToken(runnerPgid), false) } + // The container ignores stop, only dying on rm -f. + h.containers.onRemove[pgContainerName] = func() { h.containers.setExists(false) } + + if err := DownDetached(context.Background(), cfg, deps); err != nil { + t.Fatalf("DownDetached = %v, want nil", err) + } + events := ctrEvents(h.rec.snapshot()) + want := []string{"ctr-stop " + pgContainerName, "ctr-rm " + pgContainerName} + if !reflect.DeepEqual(events, want) { + t.Fatalf("container escalation:\n got %v\n want %v (stop then rm -f)", events, want) + } + assertPgidFileGone(t, cfg.StateDir) +} + +// TestDownDetachedContainerSurvivorRewritesRecord proves the partial-failure +// policy holds for a container: one still present after `podman rm -f` (its +// socket keeps answering) is a genuine survivor — reported, the record NOT +// removed, rewritten to exactly the surviving container entry so a retry can +// finish by name. +func TestDownDetachedContainerSurvivorRewritesRecord(t *testing.T) { + cfg, h := newHarness(t) + seedContainerRecord(t, cfg, h) + deps := containerDownDeps(t, h) + + h.groupSig.onTerm[serverPgid] = func() { h.groupSig.set(serverPgid, pgToken(serverPgid), false) } + h.groupSig.onTerm[runnerPgid] = func() { h.groupSig.set(runnerPgid, pgToken(runnerPgid), false) } + // The container survives even rm -f (existence never flips) → real survivor. + + err := DownDetached(context.Background(), cfg, deps) + if err == nil { + t.Fatal("DownDetached = nil, want a partial-failure error naming the surviving container") + } + rec, rerr := readPgidFile(cfg.StateDir) + if rerr != nil { + t.Fatalf("survivor record read = %v, want the rewritten survivor set", rerr) + } + if len(rec.Entries) != 1 || rec.Entries[0].Kind != entryContainer || rec.Entries[0].ContainerName != pgContainerName { + t.Fatalf("survivor record = %+v, want exactly the postgres container entry", rec.Entries) + } +} + +// TestDownDetachedContainerGoneIsSkipped proves the identity gate for containers: +// a recorded container that no longer exists is skipped — never stopped, never +// removed, never an error — while the live process entries are still torn down. +func TestDownDetachedContainerGoneIsSkipped(t *testing.T) { + cfg, h := newHarness(t) + seedContainerRecord(t, cfg, h) + deps := containerDownDeps(t, h) + + // The container is already gone. + h.containers.setExists(false) + h.groupSig.onTerm[serverPgid] = func() { h.groupSig.set(serverPgid, pgToken(serverPgid), false) } + h.groupSig.onTerm[runnerPgid] = func() { h.groupSig.set(runnerPgid, pgToken(runnerPgid), false) } + + if err := DownDetached(context.Background(), cfg, deps); err != nil { + t.Fatalf("DownDetached = %v, want nil (gone container skipped)", err) + } + if events := ctrEvents(h.rec.snapshot()); len(events) != 0 { + t.Fatalf("a gone container must not be signaled: %v", events) + } + assertPgidFileGone(t, cfg.StateDir) +} + // assertPgidFileGone fails if the pgid record still exists. func assertPgidFileGone(t *testing.T, stateDir string) { t.Helper() @@ -419,3 +621,44 @@ func writeLockHeldBy(t *testing.T, stateDir string, pid int) { t.Fatalf("write lockfile = %v", err) } } + +// TestSurvivorRecordV1RoundTrip is the FIX-1 regression: a v1 record (header +// version "1", untagged proc entries) read from a shipped build, narrowed to +// survivors on the partial-teardown path, then rewritten and reread must survive +// the round-trip. Before the fix writePgidFile stamped the carried "1" header +// over v2 `proc` grammar, so the reread dispatched to the v1 parser and +// hard-errored `malformed proc entry line`, turning a recoverable partial +// teardown into a permanently unreadable record. +func TestSurvivorRecordV1RoundTrip(t *testing.T) { + dir := t.TempDir() + rec := pgidRecord{ + WriterPid: 7, + Version: pgidFileVersionV1, + Entries: []pgidEntry{ + {Kind: entryProc, Component: ComponentPostgres, Pgid: 200, StartTime: 999}, + {Kind: entryProc, Component: ComponentServer, Pgid: 201, StartTime: 1000}, + {Kind: entryProc, Component: ComponentRunner, Pgid: 202, StartTime: 1001}, + }, + } + + survivors := survivorRecord(rec, []Component{ComponentServer, ComponentRunner}) + if err := writePgidFile(dir, survivors); err != nil { + t.Fatalf("writePgidFile survivor = %v", err) + } + + got, err := readPgidFile(dir) + if err != nil { + t.Fatalf("readPgidFile after survivor round-trip = %v; want success", err) + } + want := pgidRecord{ + WriterPid: 7, + Version: pgidFileVersion, + Entries: []pgidEntry{ + {Kind: entryProc, Component: ComponentServer, Pgid: 201, StartTime: 1000}, + {Kind: entryProc, Component: ComponentRunner, Pgid: 202, StartTime: 1001}, + }, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("reread record = %+v; want %+v", got, want) + } +} diff --git a/go/internal/stack/harness_test.go b/go/internal/stack/harness_test.go index 3e72565e..4f07a97f 100644 --- a/go/internal/stack/harness_test.go +++ b/go/internal/stack/harness_test.go @@ -287,6 +287,63 @@ func (f *fakeGroupSignaller) set(pgid int, startTime uint64, alive bool) { f.identity[pgid] = startTime } +// fakeContainerController is the container-teardown seam under test: it records +// each stop/remove in order and models per-container existence as a controllable +// state machine, the container analogue of fakeGroupSignaller. exists maps +// name→presence; a name absent from exists is treated as gone. onStop / onRemove +// hooks flip a container's existence at the right escalation step so a test can +// model a graceful stop, a stop-ignored→rm-f escalation, or a genuine survivor. +type fakeContainerController struct { + rec *recorder + mu sync.Mutex + exists map[string]bool + onStop map[string]func() + onRemove map[string]func() +} + +func newFakeContainerController(rec *recorder) *fakeContainerController { + return &fakeContainerController{ + rec: rec, + exists: map[string]bool{}, + onStop: map[string]func(){}, + onRemove: map[string]func(){}, + } +} + +func (c *fakeContainerController) Exists(name string) bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.exists[name] +} + +func (c *fakeContainerController) Stop(name string, timeout time.Duration) error { + c.mu.Lock() + c.rec.add("ctr-stop " + name) + cb := c.onStop[name] + c.mu.Unlock() + if cb != nil { + cb() // outside the lock: a hook calls setExists, which locks c.mu. + } + return nil +} + +func (c *fakeContainerController) Remove(name string) error { + c.mu.Lock() + c.rec.add("ctr-rm " + name) + cb := c.onRemove[name] + c.mu.Unlock() + if cb != nil { + cb() + } + return nil +} + +func (c *fakeContainerController) setExists(exists bool) { + c.mu.Lock() + defer c.mu.Unlock() + c.exists[pgContainerName] = exists +} + // harness bundles the recorder, the shared serverStarted flag, and the stub // seams so a test can tweak individual fields before calling Up. type harness struct { @@ -299,6 +356,7 @@ type harness struct { prober *stubProber dbProber *stubDBProber groupSig *fakeGroupSignaller + containers *fakeContainerController deps Deps } @@ -321,6 +379,7 @@ func newHarness(t *testing.T) (Config, *harness) { prober := &stubProber{rec: rec, version: testVersion, serverStarted: started} dbProber := &stubDBProber{rec: rec} groupSig := newFakeGroupSignaller(rec) + containers := newFakeContainerController(rec) // Stub the start-time reader so the pgid-capture path never touches /proc: // map each fake pid to a deterministic token (pid*10) and restore the real @@ -330,7 +389,7 @@ func newHarness(t *testing.T) (Config, *harness) { t.Cleanup(func() { readStartTime = prev }) h := &harness{ rec: rec, serverStarted: started, - sup: sup, cert: cert, token: token, image: image, prober: prober, dbProber: dbProber, groupSig: groupSig, + sup: sup, cert: cert, token: token, image: image, prober: prober, dbProber: dbProber, groupSig: groupSig, containers: containers, } h.deps = Deps{ Supervisor: sup, @@ -340,6 +399,7 @@ func newHarness(t *testing.T) (Config, *harness) { Prober: prober, DBProber: dbProber, GroupSignaller: groupSig, + Containers: containers, ExpectedVersion: testVersion, } cfg := Config{ diff --git a/go/internal/stack/pgidfile.go b/go/internal/stack/pgidfile.go index 62966f2b..735059dd 100644 --- a/go/internal/stack/pgidfile.go +++ b/go/internal/stack/pgidfile.go @@ -19,18 +19,71 @@ const pgidFileName = "stack.pgids" // pgidFileVersion is the format/provenance version written in the record header. // It is a format guard only — never a child-liveness signal (see pgidRecord). -const pgidFileVersion = "1" - -// pgidEntry is one supervised child's teardown identity: its component, the -// process-group id (== the child's pid, set via Setpgid at spawn), and the -// group leader's start time as read at spawn. StartTime is the identity token — -// it turns the down-side check from "does a group with this pgid exist" (which a -// recycled pid passes falsely) into "does a group with this pgid AND this leader -// start time exist", closing the pid-recycling window. +// +// v2 (this build) grows the entry line into a kind-tagged discriminated union +// (proc / ctr, see pgidEntry): the container-backed postgres of S4 has no +// process-group teardown identity, so it is recorded and torn down by container +// name instead. v2 is a strict superset — a v1 record (untagged 3-field proc +// lines) still parses, as all-process entries. A shipped v1 binary reading a v2 +// record never half-parses it: not by a header check (v1 never compares the +// header version) but by the entry grammar — a v2 `proc` line is 4 fields where +// v1 demands exactly 3, and a `ctr` line's leading token is not a known +// component — so v1's parser hard-errors under the same +// signal-off-a-half-understood-record discipline. The forward guard (this +// reader refusing an unknown/newer version) protects a v2 reader from a future +// v3; it never gates v1↔v2. +const pgidFileVersion = "2" + +// pgidFileVersionV1 is the prior format this build still reads for back-compat: +// untagged 3-field proc lines (no kind tag), parsed as all-process entries. +const pgidFileVersionV1 = "1" + +// pgidEntryKind tags an entry line's grammar in the v2 discriminated union. +type pgidEntryKind int + +const ( + // entryProc is a process-group child, torn down by group signal + // (identity-checked pgid + leader start time). It is the zero value, so an + // unqualified pgidEntry literal is a process entry — the v1 shape unchanged. + entryProc pgidEntryKind = iota + // entryContainer is a container child (S4's containerized postgres), torn + // down by name via podman (stop, then rm -f). A container has no process + // group of its own under rootless podman (it runs beneath conmon), so its + // teardown identity is its stable per-state-dir name, not a pgid. + entryContainer +) + +// entryKindTag renders a kind as its on-disk line tag. +func (k pgidEntryKind) entryKindTag() string { + switch k { + case entryContainer: + return "ctr" + default: + return "proc" + } +} + +// pgidEntry is one supervised child's teardown identity, a kind-tagged +// discriminated union (Kind): +// +// - entryProc (the v1 shape, unchanged): a process-group child. Component + +// the process-group id (== the child's pid, set via Setpgid at spawn) + +// the group leader's start time as read at spawn. StartTime is the identity +// token — it turns the down-side check from "does a group with this pgid +// exist" (which a recycled pid passes falsely) into "does a group with this +// pgid AND this leader start time exist", closing the pid-recycling window. +// ContainerName is empty. +// - entryContainer: a container child (S4's containerized postgres). +// Component + ContainerName (the stable per-state-dir podman name, the +// authoritative teardown identity). Pgid/StartTime are unused (zero): a +// rootless container runs beneath conmon, outside the client's process +// group, so it has no group to signal — it is torn down by name. type pgidEntry struct { - Component Component - Pgid int - StartTime uint64 + Kind pgidEntryKind + Component Component + Pgid int + StartTime uint64 + ContainerName string } // pgidRecord is the parsed pgid file: a provenance header plus the per-child @@ -55,16 +108,26 @@ type pgidRecord struct { // successful child spawn, so the on-disk file is always a complete earlier // prefix of the start sequence. // -// Format: +// Format (v2): // // -// +// proc (a process-group child) +// ctr (a container child) // ... (one line per entry, in start order) +// +// The leading kind tag makes the entry line a discriminated union; a v1 record +// (untagged 3-field proc lines) is read-only back-compat — this build never +// writes it. func writePgidFile(stateDir string, rec pgidRecord) error { var b strings.Builder - fmt.Fprintf(&b, "%s %d\n", rec.Version, rec.WriterPid) + fmt.Fprintf(&b, "%s %d\n", pgidFileVersion, rec.WriterPid) for _, e := range rec.Entries { - fmt.Fprintf(&b, "%s %d %d\n", e.Component, e.Pgid, e.StartTime) + switch e.Kind { + case entryContainer: + fmt.Fprintf(&b, "%s %s %s\n", e.Kind.entryKindTag(), e.Component, e.ContainerName) + default: + fmt.Fprintf(&b, "%s %s %d %d\n", e.Kind.entryKindTag(), e.Component, e.Pgid, e.StartTime) + } } path := filepath.Join(stateDir, pgidFileName) @@ -101,6 +164,16 @@ func writePgidFile(stateDir string, rec pgidRecord) error { // header or entry line is a hard error rather than a silent partial parse, since // signaling off a half-understood record is exactly the blast radius the design // forbids. +// +// Version dispatch (the S4/DL-262 cross-version rule): +// - "2" — this build's format: kind-tagged entry lines (proc / ctr). +// - "1" — back-compat: untagged 3-field proc lines, read as all-process +// entries (v2 is a strict superset of v1). +// - anything else — refused legibly. This is the FORWARD guard, protecting a +// v2 reader from a future v3 it cannot understand, the same +// never-signal-off-a-half-understood-record discipline. (The reverse +// direction — a v1 binary meeting a v2 record — is guarded by the entry +// grammar in v1's own parser, not here.) func readPgidFile(stateDir string) (pgidRecord, error) { path := filepath.Join(stateDir, pgidFileName) data, err := os.ReadFile(path) //nolint:gosec // G304: path is the stack-owned pgid file in the state dir, not user input @@ -117,17 +190,21 @@ func readPgidFile(stateDir string) (pgidRecord, error) { if len(header) != 2 { return pgidRecord{}, fmt.Errorf("pgid file %q: malformed header %q", path, lines[0]) } + version := header[0] + if version != pgidFileVersion && version != pgidFileVersionV1 { + return pgidRecord{}, fmt.Errorf("pgid file %q: unsupported record version %q (this build reads %q and %q); stop the stack with the build that started it", path, version, pgidFileVersionV1, pgidFileVersion) + } writerPid, err := strconv.Atoi(header[1]) if err != nil { return pgidRecord{}, fmt.Errorf("pgid file %q: unparseable writer pid %q: %w", path, header[1], err) } - rec := pgidRecord{WriterPid: writerPid, Version: header[0]} + rec := pgidRecord{WriterPid: writerPid, Version: version} for _, line := range lines[1:] { if line == "" { continue } - entry, err := parsePgidLine(line) + entry, err := parsePgidLine(version, line) if err != nil { return pgidRecord{}, fmt.Errorf("pgid file %q: %w", path, err) } @@ -136,14 +213,49 @@ func readPgidFile(stateDir string) (pgidRecord, error) { return rec, nil } -// parsePgidLine parses one " " entry line. An -// unknown component name or an unparseable number is a hard error — the record -// must be understood exactly or not signaled off at all. -func parsePgidLine(line string) (pgidEntry, error) { +// parsePgidLine parses one entry line, dispatched on the record version. +// +// - v1 ("1"): an untagged " " line, parsed as a +// process entry — the format DL-183 froze, kept for back-compat. +// - v2 ("2"): a kind-tagged line, "proc " or +// "ctr ", a discriminated union. +// +// An unknown component name, an unparseable number, or an unknown/mismatched +// kind tag is a hard error — the record must be understood exactly or not +// signaled off at all. This is also the guard a shipped v1 binary relies on when +// it meets a v2 record: its v1 parser sees a 4-field "proc …" line (where it +// demands exactly 3) or a "ctr" leading token that is not a known component, and +// hard-errors — refusing to half-parse, by the entry grammar, not a header +// check. +func parsePgidLine(version, line string) (pgidEntry, error) { + if version == pgidFileVersionV1 { + // v1: the whole line is an untagged proc body. + return parseProcEntry(line, strings.Fields(line)) + } + + // v2: the leading token is the kind tag. f := strings.Fields(line) - if len(f) != 3 { + if len(f) == 0 { return pgidEntry{}, fmt.Errorf("malformed entry line %q", line) } + switch f[0] { + case entryProc.entryKindTag(): + return parseProcEntry(line, f[1:]) + case entryContainer.entryKindTag(): + return parseContainerEntry(line, f[1:]) + default: + return pgidEntry{}, fmt.Errorf("unknown entry kind %q in entry line %q", f[0], line) + } +} + +// parseProcEntry parses a process entry's " " body +// (the fields after the kind tag, or the whole v1 line). It is the identity-and- +// safety-checked path: an unknown component, an unparseable or degenerate pgid, +// or an unparseable start time is a hard error. +func parseProcEntry(line string, f []string) (pgidEntry, error) { + if len(f) != 3 { + return pgidEntry{}, fmt.Errorf("malformed proc entry line %q", line) + } comp, ok := componentFromString(f[0]) if !ok { return pgidEntry{}, fmt.Errorf("unknown component %q in entry line %q", f[0], line) @@ -165,7 +277,26 @@ func parsePgidLine(line string) (pgidEntry, error) { if err != nil { return pgidEntry{}, fmt.Errorf("unparseable start time %q in entry line %q: %w", f[2], line, err) } - return pgidEntry{Component: comp, Pgid: pgid, StartTime: startTime}, nil + return pgidEntry{Kind: entryProc, Component: comp, Pgid: pgid, StartTime: startTime}, nil +} + +// parseContainerEntry parses a container entry's " " body (the +// fields after the ctr kind tag). An unknown component or a missing/empty name +// is a hard error — the name is the container's whole teardown identity, so an +// unusable one must not reach the podman sink. +func parseContainerEntry(line string, f []string) (pgidEntry, error) { + if len(f) != 2 { + return pgidEntry{}, fmt.Errorf("malformed ctr entry line %q", line) + } + comp, ok := componentFromString(f[0]) + if !ok { + return pgidEntry{}, fmt.Errorf("unknown component %q in entry line %q", f[0], line) + } + name := f[1] + if name == "" { + return pgidEntry{}, fmt.Errorf("empty container name in entry line %q", line) + } + return pgidEntry{Kind: entryContainer, Component: comp, ContainerName: name}, nil } // componentFromString is the inverse of Component.String for the three diff --git a/go/internal/stack/pgidfile_test.go b/go/internal/stack/pgidfile_test.go index 23166363..7ff05ba5 100644 --- a/go/internal/stack/pgidfile_test.go +++ b/go/internal/stack/pgidfile_test.go @@ -36,6 +36,127 @@ func TestPgidFileRoundTrip(t *testing.T) { } } +// TestPgidFileRoundTripBothKinds proves the v2 discriminated union round-trips +// both entry kinds: a container entry (ctr) interleaved with process entries +// (proc) survives a write→read cycle intact, in order. +func TestPgidFileRoundTripBothKinds(t *testing.T) { + dir := t.TempDir() + rec := pgidRecord{ + WriterPid: 4242, + Version: pgidFileVersion, + Entries: []pgidEntry{ + {Kind: entryContainer, Component: ComponentPostgres, ContainerName: "compass-postgres-abc123"}, + {Kind: entryProc, Component: ComponentServer, Pgid: 1002, StartTime: 10020}, + {Kind: entryProc, Component: ComponentRunner, Pgid: 1003, StartTime: 10030}, + }, + } + if err := writePgidFile(dir, rec); err != nil { + t.Fatalf("writePgidFile = %v", err) + } + got, err := readPgidFile(dir) + if err != nil { + t.Fatalf("readPgidFile = %v", err) + } + if !reflect.DeepEqual(got, rec) { + t.Fatalf("round-trip mismatch:\n got %+v\n want %+v", got, rec) + } +} + +// TestPgidFileV2ContainerLineGrammar pins the exact on-disk ctr line grammar so a +// format drift is caught: "ctr ", no pgid/starttime columns. +func TestPgidFileV2ContainerLineGrammar(t *testing.T) { + dir := t.TempDir() + rec := pgidRecord{ + WriterPid: 7, + Version: pgidFileVersion, + Entries: []pgidEntry{ + {Kind: entryContainer, Component: ComponentPostgres, ContainerName: "compass-postgres-deadbeef"}, + }, + } + if err := writePgidFile(dir, rec); err != nil { + t.Fatalf("writePgidFile = %v", err) + } + data, err := os.ReadFile(filepath.Join(dir, pgidFileName)) + if err != nil { + t.Fatalf("ReadFile = %v", err) + } + want := "2 7\nctr postgres compass-postgres-deadbeef\n" + if string(data) != want { + t.Fatalf("file content = %q, want %q", string(data), want) + } +} + +// TestReadPgidFileV1CompatAllProcess proves the cross-version back-compat rule: a +// v1 record (untagged 3-field proc lines, header version "1") parses under this +// v2 build as all-process entries — v2 is a strict superset of v1. +func TestReadPgidFileV1CompatAllProcess(t *testing.T) { + dir := t.TempDir() + v1 := "1 7\npostgres 200 999\ncompass-server 201 1000\n" + if err := os.WriteFile(filepath.Join(dir, pgidFileName), []byte(v1), 0o600); err != nil { + t.Fatalf("seed v1 file = %v", err) + } + got, err := readPgidFile(dir) + if err != nil { + t.Fatalf("readPgidFile(v1) = %v, want a parsed record", err) + } + want := pgidRecord{ + WriterPid: 7, + Version: pgidFileVersionV1, + Entries: []pgidEntry{ + {Kind: entryProc, Component: ComponentPostgres, Pgid: 200, StartTime: 999}, + {Kind: entryProc, Component: ComponentServer, Pgid: 201, StartTime: 1000}, + }, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("v1-compat parse:\n got %+v\n want %+v", got, want) + } +} + +// TestReadPgidFileUnknownVersionRefuses proves the forward guard: a record whose +// header version is neither "1" nor "2" is refused legibly rather than +// half-parsed — the never-signal-off-a-half-understood-record discipline +// extended to a future format this build cannot understand. +func TestReadPgidFileUnknownVersionRefuses(t *testing.T) { + for _, version := range []string{"3", "0", "v2", "99"} { + t.Run(version, func(t *testing.T) { + dir := t.TempDir() + content := version + " 7\nproc postgres 200 999\n" + if err := os.WriteFile(filepath.Join(dir, pgidFileName), []byte(content), 0o600); err != nil { + t.Fatalf("seed file = %v", err) + } + if _, err := readPgidFile(dir); err == nil { + t.Fatalf("readPgidFile(version %q) = nil, want a refusal error", version) + } + }) + } +} + +// TestReadPgidFileV2Malformed proves the v2 entry grammar is defensive: a garbled +// kind tag or a wrong-arity proc/ctr body is a hard error, never a partial parse. +func TestReadPgidFileV2Malformed(t *testing.T) { + cases := map[string]string{ + "unknown kind tag": "2 7\nbogus postgres 200 999\n", + "proc wrong arity": "2 7\nproc postgres 200\n", + "proc unknown component": "2 7\nproc not-a-component 200 999\n", + "proc bad pgid": "2 7\nproc postgres xx 999\n", + "proc degenerate pgid": "2 7\nproc postgres 1 999\n", + "ctr wrong arity": "2 7\nctr postgres\n", + "ctr unknown component": "2 7\nctr not-a-component name\n", + "v1-style line under v2": "2 7\npostgres 200 999\n", + } + for name, content := range cases { + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, pgidFileName), []byte(content), 0o600); err != nil { + t.Fatalf("seed file = %v", err) + } + if _, err := readPgidFile(dir); err == nil { + t.Fatalf("readPgidFile(%q) = nil, want parse error", content) + } + }) + } +} + // TestPgidFileMode0600 pins the record file's permissions. func TestPgidFileMode0600(t *testing.T) { dir := t.TempDir() @@ -69,7 +190,7 @@ func TestPgidFileTrailingNewline(t *testing.T) { if err != nil { t.Fatalf("ReadFile = %v", err) } - want := "1 7\npostgres 200 999\n" + want := "2 7\nproc postgres 200 999\n" if string(data) != want { t.Fatalf("file content = %q, want %q", string(data), want) } diff --git a/go/internal/stack/stack.go b/go/internal/stack/stack.go index 7c63146d..088bf0b0 100644 --- a/go/internal/stack/stack.go +++ b/go/internal/stack/stack.go @@ -255,18 +255,25 @@ func (s *Stack) spawnChain(ctx context.Context) error { return nil } -// recordChild appends a spawned child's teardown identity (pgid == pid, plus the -// leader start-time token read at spawn) and rewrites the state-dir pgid record -// so it reflects every child started so far. Rewriting after each spawn keeps -// the crash window one child wide: the atomically-renamed file on disk is always -// a complete earlier prefix of the start sequence, so a fresh down never reads a -// torn record and drains exactly the prefix that was started. +// recordChild appends a spawned process child's teardown identity (pgid == pid, +// plus the leader start-time token read at spawn) and rewrites the state-dir +// pgid record so it reflects every child started so far. Rewriting after each +// spawn keeps the crash window one child wide: the atomically-renamed file on +// disk is always a complete earlier prefix of the start sequence, so a fresh +// down never reads a torn record and drains exactly the prefix that was started. func (s *Stack) recordChild(c Component, p Process) error { startTime, err := readStartTime(p.Pid()) if err != nil { return fmt.Errorf("read start time for %s (pid %d): %w", c, p.Pid(), err) } - s.pgids = append(s.pgids, pgidEntry{Component: c, Pgid: p.Pid(), StartTime: startTime}) + return s.appendEntry(c, pgidEntry{Kind: entryProc, Component: c, Pgid: p.Pid(), StartTime: startTime}) +} + +// appendEntry appends one teardown entry and republishes the record, preserving +// the rewrite-after-each-spawn / one-child-wide-crash-window discipline both +// record paths share. +func (s *Stack) appendEntry(c Component, e pgidEntry) error { + s.pgids = append(s.pgids, e) rec := pgidRecord{WriterPid: os.Getpid(), Version: pgidFileVersion, Entries: s.pgids} if err := writePgidFile(s.cfg.StateDir, rec); err != nil { return fmt.Errorf("persist pgid record after starting %s: %w", c, err)