feat(stack): pgid-record v2 kind-tagged union for container teardown (RIG-2760, T8a)

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 <name>`) — 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 <seconds> <name>`); Remove is the
+// SIGKILL-tier escalation that force-removes it (`podman rm -f <name>`).
+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

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 <budget>` 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)
+}

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)
+	}
+}