diff --git a/go/cmd/compass-stack/container_postgres_podman_test.go b/go/cmd/compass-stack/container_postgres_podman_test.go index 7c2175c80..a54e023cf 100644 --- a/go/cmd/compass-stack/container_postgres_podman_test.go +++ b/go/cmd/compass-stack/container_postgres_podman_test.go @@ -261,12 +261,13 @@ func newContainerFixture(t *testing.T, root string) containerFixture { RuntimeDir: runtimeDir, } deps, err := buildDeps(stack.Config{ - StateDir: cfg.StateDir, - SocketPath: cfg.SocketPath, - ListenAddr: cfg.ListenAddr, - DatabaseDSN: cfg.DatabaseDSN, - AgentImage: cfg.AgentImage, - RuntimeDir: cfg.RuntimeDir, + StateDir: cfg.StateDir, + SocketPath: cfg.SocketPath, + ListenAddr: cfg.ListenAddr, + DatabaseDSN: cfg.DatabaseDSN, + AgentImage: cfg.AgentImage, + RuntimeDir: cfg.RuntimeDir, + ExternalNatsURL: "nats://127.0.0.1:4222", }) if err != nil { t.Fatalf("buildDeps: %v", err) diff --git a/go/cmd/compass-stack/cross_process_podman_test.go b/go/cmd/compass-stack/cross_process_podman_test.go index dc1a34d72..15d0bb9fd 100644 --- a/go/cmd/compass-stack/cross_process_podman_test.go +++ b/go/cmd/compass-stack/cross_process_podman_test.go @@ -75,9 +75,12 @@ const pgidRecordName = "stack.pgids" const upBudget = 3 * time.Minute // downBudget bounds each `compass-stack down` subprocess. DownDetached's absolute -// worst case (every component escalating to SIGKILL) is ~65s; the SIGTERM-succeeds -// path is far quicker. Bounded so a wedged down fails fast. -const downBudget = 90 * time.Second +// worst case (every component escalating to SIGKILL) is ~105s now that the nats +// drain budget joins the collector/postgres teardown; the SIGTERM-succeeds path +// is ~85s. Bounded generously above the escalation worst case so a `down` that +// legitimately escalates is never killed mid-teardown, while a genuinely wedged +// down still fails fast. +const downBudget = 150 * time.Second // answerPollInterval is the gap between server-socket readiness probes. Small // enough to make the post-up live assertion prompt, an explicit event-gate diff --git a/go/cmd/compass-stack/integration_podman_test.go b/go/cmd/compass-stack/integration_podman_test.go index 204a0333e..42155dec1 100644 --- a/go/cmd/compass-stack/integration_podman_test.go +++ b/go/cmd/compass-stack/integration_podman_test.go @@ -179,6 +179,11 @@ func newFixture(t *testing.T, shortRoot string) (stackFixture, stack.Deps) { // CollectorImage and ExternalOTLPEndpoint empty -> an empty-image // `podman run` deep in the adapter. otelExternal: "127.0.0.1:4317", + // A struct-literal configFlags bypasses newFlagSet's NatsImage default + // exactly as it does the collector's; this headless stack connects to + // no broker, so opt out rather than bundle a NATS these subtests never + // exercise. The bundled-nats path deserves its own podman-gated test. + natsExternal: "nats://127.0.0.1:4222", }) if err != nil { t.Fatalf("resolveConfig: %v", err) diff --git a/go/cmd/compass-stack/main.go b/go/cmd/compass-stack/main.go index 9619a0bc7..de100410a 100644 --- a/go/cmd/compass-stack/main.go +++ b/go/cmd/compass-stack/main.go @@ -118,6 +118,11 @@ type configFlags struct { // way --database-external+empty-DSN is rejected — a bare string flag cannot // tell "set to empty" from "unset" on the value alone. otelExternalSet bool + natsImage string + natsExternal string + // natsExternalSet records whether --nats-external was explicitly passed, for + // the same explicit-empty reject otelExternalSet drives. + natsExternalSet bool linger bool } @@ -161,6 +166,12 @@ func newFlagSet(name string, lingerable bool) (*flag.FlagSet, *configFlags) { "Do not start the bundled OTel Collector; point compass surfaces at this "+ "OTLP endpoint instead (the --otel-external opt-out, D3). The managed "+ "plane supplies its own collector.") + fs.StringVar(&f.natsImage, "nats-image", stack.DefaultNatsImage, + "Container image for the bundled NATS message broker. Defaults to the "+ + "pinned upstream nats alpine digest. Ignored with --nats-external.") + fs.StringVar(&f.natsExternal, "nats-external", "", + "Do not start the bundled NATS; point compass surfaces at this nats:// URL "+ + "instead. The managed plane supplies its own broker.") if lingerable { fs.BoolVar(&f.linger, "linger", false, "Leave the stack running after this process exits (records Config.Linger).") @@ -170,14 +181,17 @@ func newFlagSet(name string, lingerable bool) (*flag.FlagSet, *configFlags) { // markExplicitFlags records which string flags were explicitly passed (as // opposed to left at their default), for the flags whose "set to empty" must be -// told apart from "unset". Today that is only --otel-external: an explicit empty -// endpoint is a misuse resolveConfig rejects, while an unset flag is the D3 -// default (bundle the collector). Called after fs.Parse on every subcommand. +// told apart from "unset": --otel-external and --nats-external. For each, an +// explicit empty value is a misuse resolveConfig rejects, while an unset flag is +// the default (bundle the component). Called after fs.Parse on every subcommand. func markExplicitFlags(fs *flag.FlagSet, f *configFlags) { fs.Visit(func(fl *flag.Flag) { if fl.Name == "otel-external" { f.otelExternalSet = true } + if fl.Name == "nats-external" { + f.natsExternalSet = true + } }) } @@ -244,6 +258,13 @@ func resolveConfig(f configFlags) (stack.Config, error) { return stack.Config{}, errors.New("--otel-external requires an explicit OTLP endpoint: point compass surfaces at your own collector (omit the flag to bundle one)") } + // The --nats-external opt-out is rejected on an explicit empty value for the + // same reason: an empty ExternalNatsURL reads as "bundle NATS", the opposite + // of the operator's intent. + if f.natsExternalSet && f.natsExternal == "" { + return stack.Config{}, errors.New("--nats-external requires an explicit nats:// URL: point compass surfaces at your own broker (omit the flag to bundle one)") + } + cfg := stack.Config{ StateDir: f.stateDir, SocketPath: socketPath, @@ -255,6 +276,8 @@ func resolveConfig(f configFlags) (stack.Config, error) { ExternalDatabase: f.databaseExternal, CollectorImage: f.collectorImage, ExternalOTLPEndpoint: f.otelExternal, + NatsImage: f.natsImage, + ExternalNatsURL: f.natsExternal, Linger: f.linger, } if err := cfg.Validate(); err != nil { @@ -344,6 +367,24 @@ func buildDeps(cfg stack.Config) (stack.Deps, error) { deps.Containers = cc } } + // The bundled nats seams (start + readiness probe) are wired whenever a + // bundled NATS could be in play: not on the --nats-external opt-out. Its + // teardown reuses the same name-agnostic ContainerController contract, so the + // one Containers seam already set above tears down every container component + // by its recorded name; set it from the nats adapter only when neither block + // above did (external DB + external OTLP + bundled nats), so a cross-process + // down that reads a nats container entry always has a teardown seam. + if cfg.ExternalNatsURL == "" { + nc, err := adapters.NewNatsContainer() + if err != nil { + return stack.Deps{}, err + } + deps.NatsContainer = nc + deps.NatsProber = nc + if deps.Containers == nil { + deps.Containers = nc + } + } return deps, nil } diff --git a/go/cmd/compass-stack/main_test.go b/go/cmd/compass-stack/main_test.go index 3b2151613..500a0e15e 100644 --- a/go/cmd/compass-stack/main_test.go +++ b/go/cmd/compass-stack/main_test.go @@ -253,6 +253,73 @@ func TestResolveConfigCollectorFlags(t *testing.T) { }) } +// TestResolveConfigNatsFlags covers the bundled NATS / --nats-external flags. +func TestResolveConfigNatsFlags(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", "/run/user/1000") + t.Setenv("COMPASS_DATABASE_DSN", "") + + t.Run("nats-external threads into config", func(t *testing.T) { + f := baseFlags(t.TempDir()) + f.natsExternal = "nats.example.com:4222" + f.natsExternalSet = true + cfg, err := resolveConfig(f) + if err != nil { + t.Fatalf("resolveConfig: %v", err) + } + if cfg.ExternalNatsURL != f.natsExternal { + t.Errorf("ExternalNatsURL = %q, want %q", cfg.ExternalNatsURL, f.natsExternal) + } + }) + t.Run("nats-external explicit empty is rejected", func(t *testing.T) { + f := baseFlags(t.TempDir()) + f.natsExternalSet = true + _, err := resolveConfig(f) + if err == nil { + t.Fatal("resolveConfig(--nats-external \"\") = nil error, want a rejection") + } + if !strings.Contains(err.Error(), "--nats-external requires an explicit nats:// URL") { + t.Errorf("error = %v, want it to name the missing URL", err) + } + }) + t.Run("nats-external unset leaves endpoint empty", func(t *testing.T) { + cfg, err := resolveConfig(baseFlags(t.TempDir())) + if err != nil { + t.Fatalf("resolveConfig: %v", err) + } + if cfg.ExternalNatsURL != "" { + t.Errorf("ExternalNatsURL = %q, want empty (bundle NATS)", cfg.ExternalNatsURL) + } + }) + t.Run("nats-image defaults to the pinned digest via newFlagSet", func(t *testing.T) { + _, f := newFlagSet("up", true) + f.stateDir = t.TempDir() + f.image = "example.com/agent:latest" + cfg, err := resolveConfig(*f) + if err != nil { + t.Fatalf("resolveConfig: %v", err) + } + if cfg.NatsImage != stack.DefaultNatsImage { + t.Errorf("NatsImage = %q, want the pinned default %q", cfg.NatsImage, stack.DefaultNatsImage) + } + }) + t.Run("nats-external derives natsExternalSet through markExplicitFlags", func(t *testing.T) { + fs, f := newFlagSet("up", true) + f.stateDir = t.TempDir() + f.image = "example.com/agent:latest" + // Drive the real flag set end to end so the fs.Visit derivation runs. + if err := fs.Parse([]string{"--nats-external", "", "--state-dir", f.stateDir, "--image", f.image}); err != nil { + t.Fatalf("Parse: %v", err) + } + markExplicitFlags(fs, f) + if !f.natsExternalSet { + t.Fatalf("natsExternalSet = false after --nats-external \"\" parsed; want true") + } + if _, err := resolveConfig(*f); err == nil { + t.Fatal("resolveConfig(--nats-external \"\") = nil error, want the naming rejection") + } + }) +} + func TestRunDispatch(t *testing.T) { t.Run("unknown subcommand names the three", func(t *testing.T) { err := run([]string{"bogus"}) diff --git a/go/cmd/compass-stack/nats_podman_test.go b/go/cmd/compass-stack/nats_podman_test.go new file mode 100644 index 000000000..4850461b5 --- /dev/null +++ b/go/cmd/compass-stack/nats_podman_test.go @@ -0,0 +1,254 @@ +//go:build podman + +package main + +// T3 (RIG-3107) bundled NATS integration proof, mirroring the cross-process +// harness used by the bundled OTel Collector leg: it drives the REAL +// compass-stack binary as separate up/down processes against REAL rootless +// podman containers and proves the supervised NATS path end to end: +// +// up -> NATS container exists + health :8222 answers 200 -> JetStream posture +// matches the rendered store directory and sync interval -> fresh-process +// down -> NATS container gone -> pgid record removed. +// +// Build-tagged `podman` (out of the hermetic unit lane) and podmanUsable()-guarded +// so a container-less sandbox skips rather than fails. +// +// PROCESS SAFETY (rule://process-safety): the NATS container is torn down only +// by `compass-stack down` (reads the stack's OWN v2 pgid record, drives podman +// stop/rm by the recorded name) or by an explicit podman-rm of a container THIS +// test created by a unique derived name — never a pattern/name scan. +// +// DETERMINISM: no blind sleeps for readiness — up returns only after Ready, so +// its exit 0 is the gate; every post-condition polls an event under a bounded +// budget. + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/RigelBuild/compass/go/internal/stack" +) + +// natsHealthEndpoint is the host loopback endpoint on the NATS monitor port +// used by the bundled server's /healthz readiness gate. +const natsHealthEndpoint = "http://127.0.0.1:8222/healthz" + +// natsVarzEndpoint is the host loopback NATS /varz endpoint used to inspect +// the rendered JetStream configuration. +const natsVarzEndpoint = "http://127.0.0.1:8222/varz" + +// natsHealthyBudget bounds the post-up health confirmation. up already gated on +// readiness, so this is a fast confirmation, not the readiness wait. +const natsHealthyBudget = 30 * time.Second + +// TestNatsUpDown drives the bundled-NATS path end to end. +func TestNatsUpDown(t *testing.T) { + if !podmanUsable() { + t.Skip("rootless podman not usable in this environment") + } + ctx := context.Background() // test root context (rule://go-thread-context exemption for a _test.go root) + + binDir := buildBinariesFromModuleRoot(t) + stackBin := buildStackBinary(t, binDir) + env := stackEnv(binDir) + + fx := newContainerFixture(t, shortRoot(t, "-nats")) + cfg := fx.cfg + recordPath := filepath.Join(cfg.StateDir, pgidRecordName) + pgName := derivedContainerName(cfg.StateDir) + natsName := derivedNatsName(cfg.StateDir) + + // Cleanup guard: a fresh down (tears both containers down by recorded name), + // then a belt-and-suspenders explicit rm of the two containers THIS test + // would have created, by their exact unique names (never a scan). Both + // best-effort: a guard failure must not fail an already-finished test. + t.Cleanup(func() { + cctx, cancel := context.WithTimeout(context.Background(), downBudget) + defer cancel() + if out, runErr, infraErr := runStack(cctx, t, stackBin, env, cfg.args("down")...); infraErr != nil { + t.Logf("cleanup guard: down harness error (ignored): %v", infraErr) + } else if runErr != nil { + t.Logf("cleanup guard: down exited non-zero (ignored): %v\n%s", runErr, out) + } + for _, n := range []string{natsName, pgName} { + if out, err := exec.Command("podman", "rm", "--force", "--volumes", n).CombinedOutput(); err != nil { + t.Logf("cleanup guard: podman rm %s (ignored): %v\n%s", n, err, out) + } + } + }) + + upCtx, upCancel := context.WithTimeout(ctx, upBudget) + defer upCancel() + out, err := mustRunStack(upCtx, t, stackBin, env, + cfg.args("up", "--postgres-image", pgImagePinned, "--otel-external", "127.0.0.1:4317", "--linger")...) + if err != nil { + t.Fatalf("compass-stack up (bundled NATS): %v\n%s", err, out) + } + + waitServerAnswering(t, fx.deps, cfg.SocketPath) + if !containerExists(t, natsName) { + t.Fatalf("NATS container %q not present after up", natsName) + } + waitNatsHealthy(t, natsHealthEndpoint, natsHealthyBudget) + assertNatsJetStreamPosture(t, natsVarzEndpoint) + if _, err := os.Stat(recordPath); err != nil { + t.Fatalf("stack.pgids record %q missing after up: %v", recordPath, err) + } + assertRecordHasNatsEntry(t, recordPath, natsName) + + downCtx, downCancel := context.WithTimeout(ctx, downBudget) + defer downCancel() + out, err = mustRunStack(downCtx, t, stackBin, env, cfg.args("down")...) + if err != nil { + t.Fatalf("compass-stack down (bundled NATS): %v\n%s", err, out) + } + + waitContainerGone(t, natsName, containerGoneBudget) + assertServerGone(t, fx.deps, cfg.SocketPath) + if _, err := os.Stat(recordPath); !os.IsNotExist(err) { + t.Fatalf("stack.pgids record %q still present after a full down: stat err = %v", recordPath, err) + } +} + +// TestExternalNatsUpDown drives the --nats-external opt-out against REAL podman: +// the stack comes up WITHOUT a bundled NATS, and no NATS container is ever +// created. The hermetic seam (internal/stack TestExternalNatsSkipsNats) already +// gates the spawn-chain decision; this is its real-container counterpart, the +// negative sibling of TestNatsUpDown, mirroring the collector's +// TestExternalOTLPUpDown. +func TestExternalNatsUpDown(t *testing.T) { + if !podmanUsable() { + t.Skip("rootless podman not usable in this environment") + } + ctx := context.Background() // test root context (rule://go-thread-context exemption for a _test.go root) + + binDir := buildBinariesFromModuleRoot(t) + stackBin := buildStackBinary(t, binDir) + env := stackEnv(binDir) + + fx := newContainerFixture(t, shortRoot(t, "-extnats")) + cfg := fx.cfg + pgName := derivedContainerName(cfg.StateDir) + natsName := derivedNatsName(cfg.StateDir) + + t.Cleanup(func() { + cctx, cancel := context.WithTimeout(context.Background(), downBudget) + defer cancel() + if out, runErr, infraErr := runStack(cctx, t, stackBin, env, cfg.args("down")...); infraErr != nil { + t.Logf("cleanup guard: down harness error (ignored): %v", infraErr) + } else if runErr != nil { + t.Logf("cleanup guard: down exited non-zero (ignored): %v\n%s", runErr, out) + } + for _, n := range []string{natsName, pgName} { + if out, err := exec.Command("podman", "rm", "--force", "--volumes", n).CombinedOutput(); err != nil { + t.Logf("cleanup guard: podman rm %s (ignored): %v\n%s", n, err, out) + } + } + }) + + // up with --nats-external: postgres still bundled, NATS NOT. --otel-external + // too so the collector (not the subject here) also stays unbundled. + upCtx, upCancel := context.WithTimeout(ctx, upBudget) + defer upCancel() + out, err := mustRunStack(upCtx, t, stackBin, env, + cfg.args("up", "--postgres-image", pgImagePinned, "--otel-external", "127.0.0.1:4317", "--nats-external", "nats://127.0.0.1:4222", "--linger")...) + if err != nil { + t.Fatalf("compass-stack up (--nats-external): %v\n%s", err, out) + } + + waitServerAnswering(t, fx.deps, cfg.SocketPath) + // No bundled NATS container was created on the opt-out path. + if containerExists(t, natsName) { + t.Fatalf("NATS container %q present on the --nats-external path; want none", natsName) + } + + downCtx, downCancel := context.WithTimeout(ctx, downBudget) + defer downCancel() + out, err = mustRunStack(downCtx, t, stackBin, env, cfg.args("down")...) + if err != nil { + t.Fatalf("compass-stack down (--nats-external): %v\n%s", err, out) + } + assertServerGone(t, fx.deps, cfg.SocketPath) +} + +// derivedNatsName reproduces stack.natsContainerName (package-internal): the +// cleaned state directory's sha256 prefix determines the unique name. Kept in +// lockstep with production so fresh-process down and this test agree. +func derivedNatsName(stateDir string) string { + sum := sha256.Sum256([]byte(filepath.Clean(stateDir))) + return "compass-nats-" + hex.EncodeToString(sum[:6]) +} + +// waitNatsHealthy polls until the NATS health endpoint answers 200 or the +// budget elapses — the event gate for the running bundled server. +func waitNatsHealthy(t *testing.T, url string, budget time.Duration) { + t.Helper() + deadline := time.Now().Add(budget) + client := &http.Client{Timeout: 3 * time.Second} + for { + if collectorHealthOK(client, url) { + return + } + if time.Now().After(deadline) { + t.Fatalf("NATS health %q not 200 within %s", url, budget) + } + time.Sleep(answerPollInterval) //nolint:forbidigo // bounded poll tick, event-gated by deadline + } +} + +// assertNatsJetStreamPosture verifies the rendered NATS JetStream settings. +func assertNatsJetStreamPosture(t *testing.T, varzURL string) { + t.Helper() + client := &http.Client{Timeout: 3 * time.Second} + resp, err := client.Get(varzURL) + if err != nil { + t.Fatalf("GET NATS varz %q: %v", varzURL, err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET NATS varz %q: status = %s, want 200", varzURL, resp.Status) + } + var v struct { + JetStream struct { + Config struct { + StoreDir string `json:"store_dir"` + SyncInterval int64 `json:"sync_interval"` + } `json:"config"` + } `json:"jetstream"` + } + if err := json.NewDecoder(resp.Body).Decode(&v); err != nil { + t.Fatalf("decode NATS varz: %v", err) + } + if !strings.Contains(v.JetStream.Config.StoreDir, stack.NatsStoreDir) { + t.Fatalf("NATS JetStream store_dir = %q, want substring %q", v.JetStream.Config.StoreDir, stack.NatsStoreDir) + } + want := int64(100 * time.Millisecond) + if v.JetStream.Config.SyncInterval != want { + t.Fatalf("NATS JetStream sync_interval = %s (%d), want %s (%d)", time.Duration(v.JetStream.Config.SyncInterval), v.JetStream.Config.SyncInterval, time.Duration(want), want) + } +} + +// assertRecordHasNatsEntry requires the v2 pgid record's NATS container entry +// to use the exact derived name. +func assertRecordHasNatsEntry(t *testing.T, recordPath, name string) { + t.Helper() + for _, f := range recordLines(t, recordPath) { + if len(f) >= 3 && f[0] == "ctr" && f[1] == "nats" { + if f[2] != name { + t.Fatalf("NATS entry name = %q, want %q", f[2], name) + } + return + } + } + t.Fatalf("no `ctr nats %s` entry in %q", name, recordPath) +} diff --git a/go/e2e/fixture.go b/go/e2e/fixture.go index e3ac6cf20..a51c78de5 100644 --- a/go/e2e/fixture.go +++ b/go/e2e/fixture.go @@ -263,6 +263,13 @@ func NewFixture(ctx context.Context, t *testing.T, opts ...fixtureOption) *Fixtu // light-postgres choice, so spawnChain skips startCollector entirely rather // than dereferencing the (deliberately unwired) CollectorContainer seam. ExternalOTLPEndpoint: "127.0.0.1:4317", + // This headless stack connects to no broker — the server/runner NATS + // cutover is a later slice — so a bundled NATS would be pure CI cost + // (and the e2e deps below wire no NatsContainer/NatsProber, so the + // bundle path would hit the nil-dep error next). Opt out via the + // --nats-external switch, mirroring the collector choice above, so + // spawnChain skips startNats entirely. + ExternalNatsURL: "nats://127.0.0.1:4222", } // Canned-model mode (RIG-1787 H3): stand up the deterministic stub, write a diff --git a/go/internal/stack/adapters/collector_container.go b/go/internal/stack/adapters/collector_container.go index 9c8a8e6f2..6191daa91 100644 --- a/go/internal/stack/adapters/collector_container.go +++ b/go/internal/stack/adapters/collector_container.go @@ -152,11 +152,11 @@ func (c *CollectorContainer) ProbeCollector(ctx context.Context, healthEndpoint // pointing at the bind-mounted config. func collectorRunArgs(spec stack.CollectorContainerSpec, configFile string) []string { return []string{ - "run", "--detach", - "--rm", - "--replace", - "--name", spec.Name, - "--stop-timeout", strconv.FormatInt(stopSeconds(spec.StopTimeout), 10), + cmdPodmanRun, flagPodmanDetach, + flagPodmanRM, + flagPodmanReplace, + flagPodmanName, spec.Name, + flagPodmanStopTimeout, strconv.FormatInt(stopSeconds(spec.StopTimeout), 10), "-p", spec.GRPCEndpoint + ":4317", "-p", spec.HTTPEndpoint + ":4318", "-p", spec.HealthEndpoint + ":13133", diff --git a/go/internal/stack/adapters/nats_container.go b/go/internal/stack/adapters/nats_container.go new file mode 100644 index 000000000..d06c900b7 --- /dev/null +++ b/go/internal/stack/adapters/nats_container.go @@ -0,0 +1,189 @@ +//go:build unix + +package adapters + +import ( + "context" + "fmt" + "net/http" + "os" + "path/filepath" + "strconv" + "time" + + "github.com/RigelBuild/compass/go/internal/stack" +) + +// natsConfigDir is the in-container directory the generated nats-server config +// is bind-mounted into, and natsConfigPath the file within it the server is +// pointed at via its `-c` argument. A dedicated compass-owned path (rather than +// the image's own /etc/nats) keeps our generated config from colliding with the +// image's default nats-server.conf, which the entrypoint's default CMD reads — +// the explicit `-c` overrides that CMD entirely. +const ( + natsConfigDir = "/etc/compass-nats" + natsConfigFile = "nats-server.conf" + natsConfigPath = natsConfigDir + "/" + natsConfigFile +) + +// natsHealthTimeout bounds a single readiness-probe HTTP GET against the NATS +// monitoring endpoint. The probe is a fast liveness check the core polls; a +// per-request timeout keeps one wedged request from stalling a poll tick. +const natsHealthTimeout = 5 * time.Second + +// NatsContainer is the real stack.NatsContainer AND stack.ContainerController +// AND stack.NatsProber: it starts the bundled NATS server via `podman run`, +// tears it down by name, and probes its HTTP monitoring endpoint. The three +// seams share one podman surface and one container-naming contract, so they live +// in one adapter — the start side writes the config and records the name the +// teardown side later signals, the probe side reads the monitoring endpoint the +// same spec fixes. This is the collector adapter's shape verbatim. +type NatsContainer struct { + cli containerCLI + health healthGetter +} + +// Compile-time proof the adapter satisfies all three seams it fills. +var ( + _ stack.NatsContainer = (*NatsContainer)(nil) + _ stack.ContainerController = (*NatsContainer)(nil) + _ stack.NatsProber = (*NatsContainer)(nil) +) + +// NewNatsContainer builds a NatsContainer over the real podman CLI and a real +// HTTP client. Like NewCollectorContainer it resolves no OS user — nats-server +// runs as the image's own user against a host-owned bind-mount — but it returns +// an error for signature parity with the sibling container constructors and so a +// future construction-time dependency has a place to surface. +func NewNatsContainer() (*NatsContainer, error) { + return &NatsContainer{ + cli: newPodmanExec(), + health: &httpHealthGetter{client: &http.Client{Timeout: natsHealthTimeout}}, + }, nil +} + +// Start writes the generated nats-server config to disk, creates the JetStream +// data dir, and runs the NATS container detached, returning a Process handle for +// the in-process lifecycle. Both bind-mount sources are created before the run +// so podman does not fail on a missing source. Start returns at launch, not at +// readiness — the core's waitNats poll is the health gate. +func (c *NatsContainer) Start(ctx context.Context, spec stack.NatsContainerSpec) (stack.Process, error) { + // Both bind-mount SOURCES must pre-exist or `podman run` fails with a statfs + // error. Both dirs are 0700 (app-private): the config dir holds only the + // generated server config, and the data dir holds the fabric's JetStream + // store, which no other user has any business reading. + if err := os.MkdirAll(spec.ConfigDir, 0o700); err != nil { + return nil, fmt.Errorf("creating nats config dir %q: %w", spec.ConfigDir, err) + } + if err := os.MkdirAll(spec.DataDir, 0o700); err != nil { + return nil, fmt.Errorf("creating nats jetstream data dir %q: %w", spec.DataDir, err) + } + configFile := filepath.Join(spec.ConfigDir, natsConfigFile) + // The pinned image runs as root and 0600 is verified to work; retain 0644 + // defensively for a future image that runs non-root, as the collector does. + // The config is non-secret and only contains ports, store dir, and interval. + if err := os.WriteFile(configFile, []byte(spec.ConfigYAML), 0o644); err != nil { //nolint:gosec // G306: the nats server config is non-secret and must be readable by the container's user over the read-only bind-mount + return nil, fmt.Errorf("writing nats config %q: %w", configFile, err) + } + if err := c.cli.run(ctx, natsRunArgs(spec, configFile)); err != nil { + return nil, fmt.Errorf("podman run nats %q: %w", spec.Name, err) + } + return &containerProcess{cli: c.cli, name: spec.Name, stopTimeout: spec.StopTimeout}, nil +} + +// Exists reports whether the named nats container is present +// (stack.ContainerController). Like the postgres and collector adapters, a +// genuine podman engine error (neither exit-0 present nor exit-1 absent — a +// wedged daemon) is treated as PRESENT, not absent: a false "absent" would drop +// the teardown target after the pgid record is consumed, stranding a live +// container holding the JetStream store's file locks. Stop/Remove are +// idempotent, so assuming-present is safe. +// +// The ContainerController seam takes no ctx (stack/deps.go), so there is no +// caller context to thread here — the podman call is bounded by the CLI's own +// per-command timeout. +func (c *NatsContainer) Exists(name string) bool { + present, err := c.cli.exists(context.Background(), name) + if err != nil { + return true // cannot confirm absence → assume present and drive teardown + } + return present +} + +// Stop requests a graceful `podman stop -t ` (stack.ContainerController), +// which SIGTERMs nats-server and lets it flush the JetStream store. +func (c *NatsContainer) Stop(name string, timeout time.Duration) error { + return c.cli.stop(context.Background(), name, timeout) +} + +// Remove force-removes the container, the SIGKILL-tier escalation +// (stack.ContainerController): `podman rm -f`. This kills nats-server mid-flush, +// so the store recovers on next boot — the escalation is for a server that +// ignored the graceful stop, never the first resort. +func (c *NatsContainer) Remove(name string) error { + return c.cli.remove(context.Background(), name) +} + +// ProbeNats issues an HTTP GET against the NATS server's monitoring /healthz +// endpoint (stack.NatsProber). A 200 means the server is up and JetStream has +// finished enabling; any non-200 status or a dial/request error means +// not-yet-ready, which the core's readiness poll retries. The endpoint is +// spec.MonitorEndpoint (host:port). +// +// It is deliberately an HTTP probe, not a nats:// client connect: the readiness +// gate must not pull a NATS client library into the supervisor, and /healthz is +// the server's own readiness verdict, strictly better than "a TCP dial +// succeeded". +func (c *NatsContainer) ProbeNats(ctx context.Context, monitorEndpoint string) error { + url := "http://" + monitorEndpoint + "/healthz" + code, err := c.health.get(ctx, url) + if err != nil { + return fmt.Errorf("nats health GET %q: %w", url, err) + } + if code != http.StatusOK { + return fmt.Errorf("nats health %q returned status %d, want 200", url, code) + } + return nil +} + +// natsRunArgs assembles the NATS `podman run` argv (detached). Split out as a +// pure function so the full flag/publish/mount set is unit-tested without +// spawning podman, mirroring collectorRunArgs. The contract (verified against +// docker.io/library/nats:2.14.6-alpine): +// +// - --rm: the container auto-removes when it exits, so a graceful `podman stop` +// (the SIGTERM teardown tier) both stops AND removes it. The +// ContainerController.Remove (`podman rm -f`) escalation tier stays for the +// stop-ignored case and is idempotent on an already-removed container. The +// JetStream store survives regardless: it lives on the host bind-mount, not +// the container layer. +// - --replace: idempotent name reuse — a survivor of this name (a crash before +// --rm fired, an escalation race) is cleared so a fresh up never collides on +// the stable per-state-dir name. +// - --stop-timeout: the safe default for any `podman stop` that names no -t, +// sized to let nats-server flush its JetStream store (natsStopTimeout). +// - -p host:port:container-port for the client and monitoring ports, published +// on the loopback only (the spec endpoints carry the 127.0.0.1 host) — a +// trusted-tier local broker, never network-exposed. +// - -v config file -> the in-container config path, read-only: the server reads +// exactly the core-rendered config. +// - -v data dir -> the in-container JetStream store dir, read-WRITE: unlike the +// collector, NATS holds durable state that must survive a container replace. +// - server args: an explicit `-c `, which overrides the image's default +// CMD (`nats-server --config /etc/nats/nats-server.conf`) so the server reads +// our generated config rather than the image's stock one. +func natsRunArgs(spec stack.NatsContainerSpec, configFile string) []string { + return []string{ + cmdPodmanRun, flagPodmanDetach, + flagPodmanRM, + flagPodmanReplace, + flagPodmanName, spec.Name, + flagPodmanStopTimeout, strconv.FormatInt(stopSeconds(spec.StopTimeout), 10), + "-p", spec.ClientEndpoint + ":4222", + "-p", spec.MonitorEndpoint + ":8222", + "-v", configFile + ":" + natsConfigPath + ":ro,Z", + "-v", spec.DataDir + ":" + stack.NatsStoreDir + ":Z", + spec.Image, + "-c", natsConfigPath, + } +} diff --git a/go/internal/stack/adapters/nats_container_test.go b/go/internal/stack/adapters/nats_container_test.go new file mode 100644 index 000000000..c2655d6f2 --- /dev/null +++ b/go/internal/stack/adapters/nats_container_test.go @@ -0,0 +1,230 @@ +//go:build unix + +package adapters + +import ( + "context" + "errors" + "net/http" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "github.com/RigelBuild/compass/go/internal/stack" +) + +// natsTestSpec is a representative resolved nats spec. The config and data dirs +// are real temp paths so Start's MkdirAll + WriteFile succeed. +func natsTestSpec(t *testing.T) stack.NatsContainerSpec { + t.Helper() + root := t.TempDir() + return stack.NatsContainerSpec{ + Name: "compass-nats-deadbeef", + Image: "docker.io/library/nats@sha256:abc", + ConfigDir: filepath.Join(root, "nats-config"), + ConfigYAML: "port: 4222\n", + DataDir: filepath.Join(root, "nats"), + ClientEndpoint: "127.0.0.1:4222", + MonitorEndpoint: "127.0.0.1:8222", + StopTimeout: 20 * time.Second, + } +} + +// TestNatsRunArgsContract pins the exact `podman run` argv the nats contract +// requires: the auto-remove/replace/name/stop-timeout flags, both ports +// published on the loopback and mapped to the server's fixed client + monitoring +// ports, the read-only config bind-mount, the read-WRITE JetStream data mount, +// and the explicit `-c` server arg that overrides the image's default CMD. +func TestNatsRunArgsContract(t *testing.T) { + spec := natsTestSpec(t) + configFile := filepath.Join(spec.ConfigDir, "nats-server.conf") + got := natsRunArgs(spec, configFile) + want := []string{ + "run", "--detach", + "--rm", + "--replace", + "--name", "compass-nats-deadbeef", + "--stop-timeout", "20", + "-p", "127.0.0.1:4222:4222", + "-p", "127.0.0.1:8222:8222", + "-v", configFile + ":/etc/compass-nats/nats-server.conf:ro,Z", + "-v", spec.DataDir + ":/var/lib/nats:Z", + "docker.io/library/nats@sha256:abc", + "-c", "/etc/compass-nats/nats-server.conf", + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("natsRunArgs mismatch:\n got %v\n want %v", got, want) + } +} + +// TestNatsRunArgsMountModes pins the load-bearing asymmetry between the two +// bind-mounts, which a plain argv comparison would let a future edit invert: the +// config is read-ONLY (the container must never rewrite the core-rendered +// config), while the JetStream data mount is read-WRITE (durable stream state +// that must survive a container replace — a `:ro` there would make nats-server +// fail to open its store, and an in-container-layer store would silently lose +// every stream on restart). +func TestNatsRunArgsMountModes(t *testing.T) { + spec := natsTestSpec(t) + configFile := filepath.Join(spec.ConfigDir, "nats-server.conf") + args := natsRunArgs(spec, configFile) + + var mounts []string + for i, a := range args { + if a == "-v" && i+1 < len(args) { + mounts = append(mounts, args[i+1]) + } + } + if len(mounts) != 2 { + t.Fatalf("natsRunArgs has %d bind mounts, want exactly 2 (config + JetStream data): %v", len(mounts), mounts) + } + + var configMount, dataMount string + for _, m := range mounts { + switch { + case strings.HasPrefix(m, configFile+":"): + configMount = m + case strings.HasPrefix(m, spec.DataDir+":"): + dataMount = m + } + } + if configMount == "" || dataMount == "" { + t.Fatalf("could not identify both mounts in %v", mounts) + } + if !strings.HasSuffix(configMount, ":ro,Z") { + t.Errorf("config mount %q is not read-only; want a :ro,Z suffix", configMount) + } + if strings.Contains(dataMount, ":ro") { + t.Errorf("JetStream data mount %q is read-only; it must be read-WRITE (durable stream state)", dataMount) + } + // The data mount target must be exactly the store_dir the core renders into + // the config, or the server writes its store into the container layer. + if !strings.HasPrefix(dataMount, spec.DataDir+":"+stack.NatsStoreDir) { + t.Errorf("JetStream data mount %q does not target the rendered store_dir %q", dataMount, stack.NatsStoreDir) + } +} + +// TestNatsStartWritesConfigAndRuns pins Start's side effects: it creates both +// bind-mount source dirs, writes the generated config to +// /nats-server.conf, and issues exactly the run argv, returning a +// Process handle. +func TestNatsStartWritesConfigAndRuns(t *testing.T) { + spec := natsTestSpec(t) + cli := &fakeContainerCLI{} + nc := &NatsContainer{cli: cli, health: &fakeHealthGetter{code: 200}} + + p, err := nc.Start(context.Background(), spec) + if err != nil { + t.Fatalf("Start() = %v, want nil", err) + } + if p == nil { + t.Fatal("Start returned a nil Process") + } + configFile := filepath.Join(spec.ConfigDir, "nats-server.conf") + body, err := os.ReadFile(configFile) + if err != nil { + t.Fatalf("config file not written: %v", err) + } + if string(body) != spec.ConfigYAML { + t.Errorf("config file body = %q, want the spec ConfigYAML %q", body, spec.ConfigYAML) + } + // The JetStream data dir is the other bind-mount SOURCE: podman fails with a + // statfs error if it does not pre-exist, so Start must create it too. + info, err := os.Stat(spec.DataDir) + if err != nil { + t.Fatalf("JetStream data dir not created: %v", err) + } + if !info.IsDir() { + t.Fatalf("JetStream data dir %q is not a directory", spec.DataDir) + } + if !reflect.DeepEqual(cli.runArgs, natsRunArgs(spec, configFile)) { + t.Errorf("Start ran %v, want the nats argv", cli.runArgs) + } +} + +// TestNatsStartRunFailurePropagates pins that a failed `podman run` surfaces as +// an error, not a phantom Process handle. +func TestNatsStartRunFailurePropagates(t *testing.T) { + spec := natsTestSpec(t) + cli := &fakeContainerCLI{runErr: errors.New("podman: pull denied")} + nc := &NatsContainer{cli: cli, health: &fakeHealthGetter{code: 200}} + + if _, err := nc.Start(context.Background(), spec); err == nil { + t.Fatal("Start() = nil error on a failed run, want the run error") + } +} + +// TestNatsProbeHealthy pins the NatsProber seam: a 200 is healthy (nil error) +// and the probe GETs the monitoring endpoint's /healthz path — the server's own +// readiness verdict, not a bare TCP dial. +func TestNatsProbeHealthy(t *testing.T) { + hg := &fakeHealthGetter{code: http.StatusOK} + nc := &NatsContainer{cli: &fakeContainerCLI{}, health: hg} + + if err := nc.ProbeNats(context.Background(), "127.0.0.1:8222"); err != nil { + t.Fatalf("ProbeNats(200) = %v, want nil", err) + } + if hg.lastURL != "http://127.0.0.1:8222/healthz" { + t.Errorf("probed %q, want http://127.0.0.1:8222/healthz", hg.lastURL) + } +} + +// TestNatsProbeUnhealthy pins that a non-200 status and a transport error (a +// refused dial while the container is still booting) are both not-yet-ready: a +// non-nil error the core's poll retries. +func TestNatsProbeUnhealthy(t *testing.T) { + t.Run("non-200 status", func(t *testing.T) { + nc := &NatsContainer{cli: &fakeContainerCLI{}, health: &fakeHealthGetter{code: 503}} + if err := nc.ProbeNats(context.Background(), "127.0.0.1:8222"); err == nil { + t.Fatal("ProbeNats(503) = nil, want a not-ready error") + } + }) + t.Run("transport error", func(t *testing.T) { + nc := &NatsContainer{cli: &fakeContainerCLI{}, health: &fakeHealthGetter{err: errors.New("dial refused")}} + if err := nc.ProbeNats(context.Background(), "127.0.0.1:8222"); err == nil { + t.Fatal("ProbeNats(dial error) = nil, want a not-ready error") + } + }) +} + +// TestNatsControllerDispatch pins the ContainerController seam this adapter also +// fills: Exists reads the fake's existence map, Stop and Remove drive the +// respective podman calls by name. +func TestNatsControllerDispatch(t *testing.T) { + cli := &fakeContainerCLI{existsResp: map[string]bool{"compass-nats-x": true}} + nc := &NatsContainer{cli: cli, health: &fakeHealthGetter{}} + + if !nc.Exists("compass-nats-x") { + t.Error("Exists(present) = false, want true") + } + if nc.Exists("absent") { + t.Error("Exists(absent) = true, want false") + } + if err := nc.Stop("compass-nats-x", 20*time.Second); err != nil { + t.Fatalf("Stop() = %v", err) + } + if !reflect.DeepEqual(cli.stopped, []string{"compass-nats-x"}) { + t.Errorf("stop calls = %v, want one stop", cli.stopped) + } + if err := nc.Remove("compass-nats-x"); err != nil { + t.Fatalf("Remove() = %v", err) + } + if !reflect.DeepEqual(cli.removed, []string{"compass-nats-x"}) { + t.Errorf("remove calls = %v, want one remove", cli.removed) + } +} + +// TestNatsExistsAssumesPresentOnEngineError pins the stranded-container guard: a +// genuine podman engine error makes Exists report PRESENT, so entryAlive still +// builds a teardown target instead of dropping a live container that still holds +// the JetStream store. +func TestNatsExistsAssumesPresentOnEngineError(t *testing.T) { + cli := &fakeContainerCLI{existsErr: errors.New("podman daemon wedged")} + nc := &NatsContainer{cli: cli, health: &fakeHealthGetter{}} + if !nc.Exists("compass-nats-x") { + t.Fatal("Exists on engine error = false, want true (assume present, drive teardown)") + } +} diff --git a/go/internal/stack/adapters/postgres_container.go b/go/internal/stack/adapters/postgres_container.go index f2055ed5d..406a32a72 100644 --- a/go/internal/stack/adapters/postgres_container.go +++ b/go/internal/stack/adapters/postgres_container.go @@ -161,15 +161,20 @@ func (c *PostgresContainer) Remove(name string) error { // - server args: unix_socket_directories lists BOTH the image's compiled // default (for the entrypoint bootstrap) and our bind-mounted DSN dir; // listen_addresses=” is socket-only (no TCP); -p is the DSN port. +// +// The flags every container adapter shares (detach/rm/replace/name/stop-timeout) +// come from the flagPodman* consts below rather than repeated literals: three +// adapters now assemble a `podman run`, and a typo in one copy would surface as +// an opaque podman usage error at start time rather than a compile failure. func runArgs(spec stack.PostgresContainerSpec, superuser string) []string { pgdata := "/pgdata" return []string{ - "run", "--detach", - "--rm", - "--replace", - "--name", spec.Name, + cmdPodmanRun, flagPodmanDetach, + flagPodmanRM, + flagPodmanReplace, + flagPodmanName, spec.Name, "--userns=keep-id", - "--stop-timeout", strconv.FormatInt(stopSeconds(spec.StopTimeout), 10), + flagPodmanStopTimeout, strconv.FormatInt(stopSeconds(spec.StopTimeout), 10), "-e", "POSTGRES_DB=" + postgresDB, "-e", "POSTGRES_HOST_AUTH_METHOD=trust", "-e", "POSTGRES_USER=" + superuser, @@ -183,6 +188,29 @@ func runArgs(spec stack.PostgresContainerSpec, superuser string) []string { } } +// The `podman run` subcommand and the flags every container adapter (postgres, +// collector, nats) emits identically. They live here beside the shared +// containerCLI/podmanExec surface rather than in any one component's file, since +// all three argv builders consume them. +const ( + // cmdPodmanRun is the podman subcommand every argv builder leads with. + cmdPodmanRun = "run" + // flagPodmanDetach runs the container in the background, returning at launch. + flagPodmanDetach = "--detach" + // flagPodmanRM auto-removes the container on exit, so a graceful stop both + // stops AND removes it. + flagPodmanRM = "--rm" + // flagPodmanReplace clears a survivor of the same name, so a fresh up never + // collides on the stable per-state-dir name. + flagPodmanReplace = "--replace" + // flagPodmanName pins the stable per-state-dir container name that is the + // component's durable teardown identity. + flagPodmanName = "--name" + // flagPodmanStopTimeout is the safe default grace for any `podman stop` that + // passes no explicit -t. + flagPodmanStopTimeout = "--stop-timeout" +) + // postgresDB is the single database the private store holds, created by the // image entrypoint from POSTGRES_DB. It matches the dbname compass-server's DSN // opens; kept here beside the argv builder rather than imported from the core so diff --git a/go/internal/stack/config.go b/go/internal/stack/config.go index 878d6c9cd..4a761ffc0 100644 --- a/go/internal/stack/config.go +++ b/go/internal/stack/config.go @@ -59,6 +59,20 @@ type Config struct { // endpoint. Empty is the D3 default posture: the bundled collector is // provisioned per CollectorImage (present and receiving, exporting nowhere). ExternalOTLPEndpoint string + // NatsImage selects the image the bundled NATS component runs, mirroring + // CollectorImage. Non-empty is the installed-stack default: a + // container-backed nats-server run from this image ref (the pinned + // DefaultNatsImage) via the Deps.NatsContainer seam. The CLI slice resolves + // the default; the core applies none. Ignored when ExternalNatsURL is set + // (no nats component starts at all). + NatsImage string + // ExternalNatsURL opts the stack out of starting the bundled NATS entirely + // (the --nats-external opt-out, the ExternalOTLPEndpoint template). When + // set, Up skips the nats component and consumers point at this + // operator/managed-plane-supplied nats:// URL instead. Empty is the default + // posture: NATS is provisioned as a bundled stack service, reachable on the + // fixed loopback client endpoint. + ExternalNatsURL string // AgentImage is the container image ref every agent workstream runs; the // runner refuses to boot without it present in the local store. AgentImage string diff --git a/go/internal/stack/deps.go b/go/internal/stack/deps.go index 3ac54ea5b..224b87fed 100644 --- a/go/internal/stack/deps.go +++ b/go/internal/stack/deps.go @@ -70,6 +70,23 @@ type Deps struct { // where no collector component starts; the core dereferences it only on the // collector-readiness gate. CollectorProber CollectorProber + // NatsContainer starts the container-backed NATS child: the default when + // Config.ExternalNatsURL is empty. Start runs `podman run` (detached) and + // returns a Process handle whose Pid the caller does NOT persist as a pgid + // (a rootless container runs beneath conmon, outside the client's group) — + // the container's teardown identity is its stable name, recorded as a v2 + // container entry and torn down via Containers on a fresh down, exactly like + // CollectorContainer. Nil on the --nats-external path, where no nats + // container starts; the core dereferences it only when it dispatches to the + // nats start. + NatsContainer NatsContainer + // NatsProber probes the bundled NATS server's HTTP monitoring endpoint for + // readiness between launching it and the components that connect to it — + // the nats analogue of CollectorProber, since NatsContainer.Start returns at + // launch, not at readiness. Nil on the --nats-external path, where no nats + // component starts; the core dereferences it only on the nats-readiness + // gate. + NatsProber NatsProber // Now is the clock the cert-expiry math reads. Nil defaults to time.Now. Now func() time.Time @@ -109,6 +126,14 @@ const ( // child, so it uses no componentBinary; the enum + String case are its log // label and the component key its v2 container pgid entry records. ComponentCollector + // ComponentNats is the bundled NATS child (the fabric's message broker). + // Like ComponentCollector it is a container child, so it uses no + // componentBinary; the enum + String case are its log label and the + // component key its v2 container pgid entry records. Appended after + // ComponentCollector as a new iota value — the existing values must not + // renumber, since a persisted pgid record round-trips components by their + // String() name and a reorder would silently retag entries. + ComponentNats ) // String renders the component for logs and errors. @@ -122,6 +147,8 @@ func (c Component) String() string { return "compass-runner" case ComponentCollector: return "otel-collector" + case ComponentNats: + return "nats" default: return "unknown-component" } @@ -230,6 +257,23 @@ type CollectorContainer interface { Start(ctx context.Context, spec CollectorContainerSpec) (Process, error) } +// NatsContainer starts the container-backed NATS child, the nats analogue of +// CollectorContainer. It is the START seam; ContainerController tears NATS down +// on a fresh cross-process down by the persisted name (the same reusable +// teardown contract postgres and the collector use). +// +// Start writes the generated nats-server config to disk, creates the JetStream +// data dir, runs `podman run` (detached) from spec, and returns a Process +// handle for the in-process lifecycle: Signal(SignalTerm) maps to `podman stop` +// and Wait blocks until the container exits, so an in-process Down drains it the +// same way it drains a process child. The handle's Pid is NOT a process-group id +// (a rootless container runs beneath conmon) and is never persisted as a pgid; +// the container's durable teardown identity is spec.Name, recorded as a v2 +// container entry. A non-nil error means the container could not be launched. +type NatsContainer interface { + Start(ctx context.Context, spec NatsContainerSpec) (Process, 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 @@ -290,6 +334,18 @@ type CollectorProber interface { ProbeCollector(ctx context.Context, healthEndpoint string) error } +// NatsProber probes the bundled NATS server's HTTP monitoring endpoint for +// readiness — the nats analogue of CollectorProber. NatsContainer.Start returns +// at launch, not at readiness, so spawnChain polls this between launching NATS +// and the components that connect to it. A nil error means the server answered +// healthy on monitorEndpoint; a non-nil error means not yet (still starting, or +// JetStream still recovering its store). Deliberately an HTTP probe, not a +// nats:// client connect: the readiness gate must not pull a NATS client +// dependency into the supervisor. +type NatsProber interface { + ProbeNats(ctx context.Context, monitorEndpoint string) error +} + // ServerInfo is the subset of GetServerInfo the core consumes. type ServerInfo struct { Version string diff --git a/go/internal/stack/downdetached.go b/go/internal/stack/downdetached.go index ce9aa106b..7768d4065 100644 --- a/go/internal/stack/downdetached.go +++ b/go/internal/stack/downdetached.go @@ -14,10 +14,10 @@ import ( // Per-component drain budgets: after SIGTERM, how long DownDetached waits for a // component's confirmation channel to go quiet before escalating to a group -// SIGKILL. Reverse start order (runner → server → collector → postgres); the +// SIGKILL. Reverse start order (runner → server → nats → collector → postgres); the // server's graceful drain is the long pole. On the SIGTERM-succeeds path they -// sum to 65s; on the escalation path the three non-runner components each add a -// postKillGrace, so the true worst case is 15 + (30+5) + (10+5) + (10+5) = 80s. +// sum to 85s; on the escalation path the four non-runner components each add a +// postKillGrace, so the true worst case is 15 + (30+5) + (20+5) + (10+5) + (10+5) = 105s. // That can exceed the app's 60s stackDownTimeout (lifecycle.go:35) — and is // bounded by its ctx cancellation, not by this arithmetic: on ctx.Done waitDead // returns the current dead() verdict, so an overrun becomes a partial-failure @@ -34,8 +34,14 @@ var ( // state to drain (D3 drops rather than buffering), so it stops fast; the // budget matches postgres's container-drain tier for parity. collectorDrainBudget = 10 * time.Second + // natsDrainBudget bounds the nats container's graceful `podman stop` before + // the `podman rm -f` escalation. Unlike the collector, NATS flushes its + // JetStream file store on SIGTERM, so it gets the wider budget its in-process + // natsStopTimeout also reserves — a `rm -f` mid-flush is exactly the + // unclean-shutdown case the store has to recover from on next boot. + natsDrainBudget = 20 * time.Second // postKillGrace bounds the confirm after the hard kill for the non-runner - // components (server, collector, postgres): the kill is unblockable + // components (server, collector, nats, postgres): the kill is unblockable // (group SIGKILL for the socket children, `podman rm -f` for the container), // so a component still confirming alive past this grace — a socket still // answering, or the container still existing — is a genuine survivor, not a @@ -167,7 +173,7 @@ type target struct { } // liveTargets returns the identity-matched live groups in reverse start order -// (runner → server → collector → postgres). Each recorded group is checked with +// (runner → server → nats → collector → postgres). Each recorded group is checked with // GroupSignaller.Alive (existence AND start-time identity); a gone or recycled // group is omitted — never signaled. func liveTargets(ctx context.Context, cfg Config, deps Deps, rec pgidRecord) []target { @@ -192,6 +198,14 @@ func liveTargets(ctx context.Context, cfg Config, deps Deps, rec pgidRecord) []t // Socket quiescence: the UDS stops answering GetServerInfo. return func() bool { _, err := deps.Prober.Probe(ctx, cfg.SocketPath); return err != nil } }}, + {ComponentNats, natsDrainBudget, func(e pgidEntry) func() bool { + // Container existence: nats is a container child torn down by name, + // confirmed gone when `podman container exists` reports absent. + // Reverse start order places it after the server and runner (its + // future consumers, PR3/PR4) so no live consumer outlives the broker + // it publishes to. + return func() bool { return !deps.Containers.Exists(e.ContainerName) } + }}, {ComponentCollector, collectorDrainBudget, func(e pgidEntry) func() bool { // Container existence: the collector is a container child (like the // container-backed postgres), torn down by name; it is confirmed gone diff --git a/go/internal/stack/downdetached_test.go b/go/internal/stack/downdetached_test.go index e932badbe..937ecfb81 100644 --- a/go/internal/stack/downdetached_test.go +++ b/go/internal/stack/downdetached_test.go @@ -45,15 +45,16 @@ func downTestDeps(t *testing.T, h *harness) Deps { // test, restoring them after. Small but nonzero so the deadline math is real. func shrinkBudgets(t *testing.T) { t.Helper() - pr, ps, pp, pc, pk, pi := runnerDrainBudget, serverDrainBudget, postgresDrainBudget, collectorDrainBudget, postKillGrace, downPollInterval + pr, ps, pp, pc, pn, pk, pi := runnerDrainBudget, serverDrainBudget, postgresDrainBudget, collectorDrainBudget, natsDrainBudget, postKillGrace, downPollInterval runnerDrainBudget = 20 * time.Millisecond serverDrainBudget = 20 * time.Millisecond postgresDrainBudget = 20 * time.Millisecond collectorDrainBudget = 20 * time.Millisecond + natsDrainBudget = 20 * time.Millisecond postKillGrace = 20 * time.Millisecond downPollInterval = time.Millisecond t.Cleanup(func() { - runnerDrainBudget, serverDrainBudget, postgresDrainBudget, collectorDrainBudget, postKillGrace, downPollInterval = pr, ps, pp, pc, pk, pi + runnerDrainBudget, serverDrainBudget, postgresDrainBudget, collectorDrainBudget, natsDrainBudget, postKillGrace, downPollInterval = pr, ps, pp, pc, pn, pk, pi }) } @@ -636,10 +637,11 @@ func seedCollectorRecord(t *testing.T, cfg Config, h *harness) { h.groupSig.set(runnerPgid, pgToken(runnerPgid), true) } -// collectorDownDeps points the collector confirm at container existence while -// the server/postgres confirms track their group liveness (they are processes -// in this record). -func collectorDownDeps(t *testing.T, h *harness) Deps { +// sidecarContainerDownDeps points a container component's confirm at container +// existence while the server/postgres confirms track their group liveness (they +// are processes in these records). Shared by the collector and nats teardown +// tests — the Containers seam is name-agnostic, so one wiring serves both. +func sidecarContainerDownDeps(t *testing.T, h *harness) Deps { t.Helper() deps := downTestDeps(t, h) deps.Containers = h.containers @@ -665,7 +667,7 @@ func indexOf(events []string, want string) int { func TestDownDetachedCollectorContainerTornDownByName(t *testing.T) { cfg, h := newHarness(t) seedCollectorRecord(t, cfg, h) - deps := collectorDownDeps(t, h) + deps := sidecarContainerDownDeps(t, h) for _, pgid := range []int{pgPgid, serverPgid, runnerPgid} { h.groupSig.onTerm[pgid] = func() { h.groupSig.set(pgid, pgToken(pgid), false) } @@ -701,7 +703,7 @@ func TestDownDetachedCollectorContainerTornDownByName(t *testing.T) { func TestDownDetachedCollectorEscalatesToRemove(t *testing.T) { cfg, h := newHarness(t) seedCollectorRecord(t, cfg, h) - deps := collectorDownDeps(t, h) + deps := sidecarContainerDownDeps(t, h) for _, pgid := range []int{pgPgid, serverPgid, runnerPgid} { h.groupSig.onTerm[pgid] = func() { h.groupSig.set(pgid, pgToken(pgid), false) } @@ -722,6 +724,98 @@ func TestDownDetachedCollectorEscalatesToRemove(t *testing.T) { assertPgidFileGone(t, cfg.StateDir) } +// The stable name a v2 nats container entry carries in these tests. +const natsContainerNameTest = "compass-nats-test01" + +// seedNatsRecord writes a v2 record whose nats entry is a container (ctr) +// sitting between the server and postgres process entries — the start-order +// shape a real up records — and marks all four children live. It is the +// hermetic guard for the cross-process nats teardown: a liveTargets that does +// not know ComponentNats silently skips the container and leaks it, which no +// podman-gated integration test would catch on a default-CI run. +func seedNatsRecord(t *testing.T, cfg Config, h *harness) { + t.Helper() + rec := pgidRecord{ + WriterPid: 4242, + Version: pgidFileVersion, + Entries: []pgidEntry{ + {Kind: entryProc, Component: ComponentPostgres, Pgid: pgPgid, StartTime: pgToken(pgPgid)}, + {Kind: entryContainer, Component: ComponentNats, ContainerName: natsContainerNameTest}, + {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 nats record = %v", err) + } + h.containers.setExistsName(natsContainerNameTest, true) + h.groupSig.set(pgPgid, pgToken(pgPgid), true) + h.groupSig.set(serverPgid, pgToken(serverPgid), true) + h.groupSig.set(runnerPgid, pgToken(runnerPgid), true) +} + +// TestDownDetachedNatsContainerTornDownByName proves the cross-process teardown +// of the bundled nats: a detached down reads the v2 record, stops the nats +// container BY NAME (graceful, no rm -f) and in reverse start order — after the +// server is signaled (its future consumer, PR3/PR4), before postgres. Graceful +// matters more here than for the collector: `podman stop` SIGTERMs nats-server +// so it flushes the JetStream store, while an rm -f would leave the store to +// recover on next boot. +func TestDownDetachedNatsContainerTornDownByName(t *testing.T) { + cfg, h := newHarness(t) + seedNatsRecord(t, cfg, h) + deps := sidecarContainerDownDeps(t, h) + + for _, pgid := range []int{pgPgid, serverPgid, runnerPgid} { + h.groupSig.onTerm[pgid] = func() { h.groupSig.set(pgid, pgToken(pgid), false) } + } + h.containers.onStop[natsContainerNameTest] = func() { + h.containers.setExistsName(natsContainerNameTest, false) + } + + if err := DownDetached(context.Background(), cfg, deps); err != nil { + t.Fatalf("DownDetached = %v, want nil", err) + } + + if got := ctrEvents(h.rec.snapshot()); !reflect.DeepEqual(got, []string{"ctr-stop " + natsContainerNameTest}) { + t.Fatalf("nats teardown:\n got %v\n want [ctr-stop %s] (graceful, by name)", got, natsContainerNameTest) + } + ev := h.rec.snapshot() + iServer := indexOf(ev, "group-term "+strconv.Itoa(serverPgid)) + iNats := indexOf(ev, "ctr-stop "+natsContainerNameTest) + iPg := indexOf(ev, "group-term "+strconv.Itoa(pgPgid)) + if !(iServer >= 0 && iNats >= 0 && iPg >= 0 && iServer < iNats && iNats < iPg) { + t.Fatalf("teardown order wrong: server@%d nats@%d postgres@%d; want server/nats-config). Kept DISTINCT from DataDir so the read-only + // config mount never overlaps the read-write JetStream store. + ConfigDir string + // ConfigYAML is the fully-rendered nats-server config (JetStream on with the + // record's bounded-fsync sync_interval, the monitoring endpoint, the client + // port). The core renders it so the posture is a pure, unit-tested value; + // the adapter only writes it to disk. Named ConfigYAML for parity with + // CollectorContainerSpec — the nats-server config grammar is its own + // JSON-superset dialect, not literally YAML. + ConfigYAML string + // DataDir is the host directory JetStream's file store is bind-mounted from, + // fixed under the state dir (/nats), mirroring how + // PostgresContainerSpec.DataDir fixes PGDATA. It is mounted read-WRITE at + // NatsStoreDir: this is the fabric's durable state and must survive a + // container replace. + DataDir string + // ClientEndpoint is the host loopback endpoint the client port is published + // on (host:port). It is what a future nats:// URL is formed from; nothing + // in-tree connects to it yet. + ClientEndpoint string + // MonitorEndpoint is the host loopback endpoint the HTTP monitoring port is + // published on (host:port); the readiness probe issues an HTTP GET against + // its /healthz. + MonitorEndpoint string + // StopTimeout is the `--stop-timeout` pinned into the run (natsStopTimeout). + StopTimeout time.Duration +} + +// natsContainerSpec builds the NATS run spec from the resolved config: it +// derives the stable container name and the config + data dirs from the state +// dir, renders the JetStream-on config, and fixes the published loopback +// endpoints. It is pure (no I/O) so the config and endpoint set it encodes is +// unit-tested directly, and it errors on a config missing the state dir the +// bind-mounts and name derivation need rather than running podman against a +// half-formed spec. +func natsContainerSpec(cfg Config) (NatsContainerSpec, error) { + if cfg.StateDir == "" { + return NatsContainerSpec{}, errors.New("stack config: StateDir is required for the nats container (config + JetStream data bind-mounts + name derivation)") + } + if cfg.NatsImage == "" { + return NatsContainerSpec{}, errors.New("stack config: NatsImage is required to bundle nats (set --nats-image or use --nats-external to opt out)") + } + spec := NatsContainerSpec{ + Name: natsContainerName(cfg.StateDir), + Image: cfg.NatsImage, + ConfigDir: filepath.Join(cfg.StateDir, "nats-config"), + DataDir: filepath.Join(cfg.StateDir, "nats"), + ClientEndpoint: natsListenHost + ":" + natsClientPort, + MonitorEndpoint: natsListenHost + ":" + natsMonitorPort, + StopTimeout: natsStopTimeout, + } + spec.ConfigYAML = natsConfigYAML() + return spec, nil +} + +// natsContainerName derives the stable per-state-dir NATS container name. Like +// containerName (postgres) and collectorContainerName it is a deterministic +// function of the state dir alone so a fresh `down` with no in-memory handle +// reconstructs the same name, and the hash keeps concurrent stacks on different +// state dirs from colliding in podman's flat container namespace. A distinct +// prefix from the postgres and collector names keeps the three components' +// containers legible apart in `podman ps`. +func natsContainerName(stateDir string) string { + sum := sha256.Sum256([]byte(filepath.Clean(stateDir))) + return "compass-nats-" + hex.EncodeToString(sum[:6]) +} + +// natsConfigYAML renders the nats-server config realizing the record's fabric +// posture. It is a pure function (every value is a fixed container-internal +// port or path) so the posture is unit-tested directly: +// +// - host/port: the client listener, bound 0.0.0.0 INSIDE the container (the +// run spec publishes it on the host loopback only, so the 0.0.0.0 bind is +// scoped to the container's own netns, not the host's). +// - http: the HTTP monitoring endpoint. It is the readiness probe target +// (/healthz) — the reason the monitoring subsystem is enabled at all, since +// the probe is plain net/http and pulls no NATS client library into the +// stack package. +// - jetstream.store_dir: NatsStoreDir, the read-write bind-mount of the host +// DataDir, so streams survive a container replace. Single-node R1 file +// storage; clustering is out of scope here. +// - jetstream.sync_interval: 100ms, the record's Jepsen-driven bounded-fsync +// value (design.md:363). It is a SERVER setting, not a per-stream +// jetstream.StreamConfig field, which is exactly why it lives in this +// container's config and not in any producer's stream declaration: it caps +// how much acknowledged-but-unflushed data a hard power loss can lose, +// uniformly for every stream the server holds. +func natsConfigYAML() string { + return `host: "0.0.0.0" +port: ` + natsClientPort + ` + +http: "0.0.0.0:` + natsMonitorPort + `" + +jetstream { + store_dir: "` + NatsStoreDir + `" + sync_interval: "100ms" +} +` +} diff --git a/go/internal/stack/nats_container_test.go b/go/internal/stack/nats_container_test.go new file mode 100644 index 000000000..2407f7010 --- /dev/null +++ b/go/internal/stack/nats_container_test.go @@ -0,0 +1,288 @@ +//go:build unix + +package stack + +import ( + "context" + "strings" + "testing" + "time" +) + +// TestNatsContainerSpecBuildsFromConfig pins the nats spec builder: image +// passthrough, the two DISTINCT dirs under the state dir (a read-only config dir +// and a read-write JetStream data dir), the fixed loopback client + monitor +// endpoints, the stop timeout, and a stable derived name. +func TestNatsContainerSpecBuildsFromConfig(t *testing.T) { + cfg := Config{ + StateDir: "/state", + NatsImage: "docker.io/library/nats@sha256:abc", + } + spec, err := natsContainerSpec(cfg) + if err != nil { + t.Fatalf("natsContainerSpec() = %v, want nil", err) + } + if spec.Image != cfg.NatsImage { + t.Errorf("spec.Image = %q, want %q", spec.Image, cfg.NatsImage) + } + if spec.ConfigDir != "/state/nats-config" { + t.Errorf("spec.ConfigDir = %q, want /state/nats-config", spec.ConfigDir) + } + if spec.DataDir != "/state/nats" { + t.Errorf("spec.DataDir = %q, want /state/nats", spec.DataDir) + } + // The read-only config mount and the read-write JetStream store must never + // be the same host dir, or the config bind would land inside the store (or + // vice versa) and one of the two mounts would shadow the other. + if spec.ConfigDir == spec.DataDir { + t.Errorf("spec.ConfigDir and spec.DataDir are the same path %q; the ro config mount must not overlap the rw JetStream store", spec.ConfigDir) + } + if spec.ClientEndpoint != "127.0.0.1:4222" { + t.Errorf("spec.ClientEndpoint = %q, want 127.0.0.1:4222", spec.ClientEndpoint) + } + if spec.MonitorEndpoint != "127.0.0.1:8222" { + t.Errorf("spec.MonitorEndpoint = %q, want 127.0.0.1:8222", spec.MonitorEndpoint) + } + if spec.StopTimeout != natsStopTimeout { + t.Errorf("spec.StopTimeout = %v, want %v", spec.StopTimeout, natsStopTimeout) + } + if spec.Name != natsContainerName(cfg.StateDir) { + t.Errorf("spec.Name = %q, want the derived name %q", spec.Name, natsContainerName(cfg.StateDir)) + } + if spec.ConfigYAML == "" { + t.Error("spec.ConfigYAML is empty; want the rendered nats-server config") + } +} + +// TestNatsContainerSpecRejectsMissingStateDir pins that a config with no state +// dir is a hard error, not a run against a half-formed spec (the state dir is +// the root of both bind-mounts and the name-derivation input). +func TestNatsContainerSpecRejectsMissingStateDir(t *testing.T) { + _, err := natsContainerSpec(Config{NatsImage: "img:pinned"}) + if err == nil { + t.Fatal("natsContainerSpec(no StateDir) = nil error, want a rejection") + } + if !strings.Contains(err.Error(), "StateDir") { + t.Fatalf("error %q does not mention StateDir", err.Error()) + } +} + +// TestNatsContainerSpecRejectsMissingImage pins that a bundle-path config with +// no nats image is a hard error at spec time, not an opaque `podman run ""` deep +// in the adapter. NATS is container-only (no process fallback like postgres), so +// an empty image is always invalid; a struct-literal Config that leaves NatsImage +// empty without opting out via ExternalNatsURL must be rejected here. +func TestNatsContainerSpecRejectsMissingImage(t *testing.T) { + _, err := natsContainerSpec(Config{StateDir: "/state"}) + if err == nil { + t.Fatal("natsContainerSpec(no NatsImage) = nil error, want a rejection") + } + if !strings.Contains(err.Error(), "NatsImage") { + t.Fatalf("error %q does not mention NatsImage", err.Error()) + } +} + +// TestNatsContainerNameDeterministicPerStateDir pins the stable-name contract: +// the name is a pure function of the state dir (so a fresh down reconstructs +// it), distinct across state dirs (so concurrent stacks never collide), +// clean-normalized, and carries the nats-specific prefix (so it is legible apart +// from the postgres and collector containers in `podman ps`). +func TestNatsContainerNameDeterministicPerStateDir(t *testing.T) { + a1 := natsContainerName("/state/a") + a2 := natsContainerName("/state/a") + b := natsContainerName("/state/b") + if a1 != a2 { + t.Fatalf("natsContainerName not deterministic: %q vs %q", a1, a2) + } + if a1 == b { + t.Fatalf("natsContainerName collides across state dirs: both %q", a1) + } + if !strings.HasPrefix(a1, "compass-nats-") { + t.Fatalf("natsContainerName %q missing the compass-nats- prefix", a1) + } + if natsContainerName("/state/a/") != a1 { + t.Fatalf("natsContainerName not clean-normalized: %q vs %q", natsContainerName("/state/a/"), a1) + } + // Distinct from the other two container components' names for the same state + // dir, so the three never collide in podman's flat namespace. + if a1 == containerName("/state/a") || a1 == collectorContainerName("/state/a") { + t.Fatalf("nats container name collides with postgres/collector for the same state dir: %q", a1) + } +} + +// TestNatsConfigYAMLRealizesJetStreamPosture is the record-compliance gate: the +// generated nats-server config MUST enable JetStream with a store_dir on the +// read-write bind-mount and the record's bounded-fsync sync_interval of 100ms +// (design.md:363). sync_interval is a SERVER setting, not a per-stream +// jetstream.StreamConfig field, so this config is the only place the value can +// live — a regression dropping it here silently reverts every stream to the +// server default fsync cadence, with no error anywhere. The config must also +// carry the client port and the HTTP monitoring endpoint the readiness probe +// hits. +func TestNatsConfigYAMLRealizesJetStreamPosture(t *testing.T) { + conf := natsConfigYAML() + + // JetStream on, storing into the read-write bind-mount target. + if !strings.Contains(conf, "jetstream {") { + t.Errorf("config missing the jetstream block:\n%s", conf) + } + if !strings.Contains(conf, `store_dir: "`+NatsStoreDir+`"`) { + t.Errorf("config missing the JetStream store_dir at the mount target %q:\n%s", NatsStoreDir, conf) + } + // The Jepsen-driven bounded-fsync value. Asserted as the literal setting, not + // just the substring "100ms", so a stray 100ms elsewhere could not satisfy it. + if !strings.Contains(conf, `sync_interval: "100ms"`) { + t.Errorf("config missing sync_interval: \"100ms\" (design.md:363, the bounded-fsync value):\n%s", conf) + } + + // The client listener and the monitoring endpoint the probe GETs /healthz on. + if !strings.Contains(conf, "port: "+natsClientPort) { + t.Errorf("config missing the client port %s:\n%s", natsClientPort, conf) + } + if !strings.Contains(conf, `http: "0.0.0.0:`+natsMonitorPort+`"`) { + t.Errorf("config missing the http monitoring endpoint on %s:\n%s", natsMonitorPort, conf) + } + + // Out of scope for this shape, and each would be a silent posture change: + // clustering (T5) and any authorization/credentials block (the RIG-2861 auth + // seam). Their absence is the contract, so assert it directly. + for _, forbidden := range []string{"cluster", "authorization", "accounts", "operator"} { + if strings.Contains(conf, forbidden) { + t.Errorf("config contains forbidden %q — single-node, no-auth is this shape's posture:\n%s", forbidden, conf) + } + } +} + +// TestExternalNatsSkipsNats pins the --nats-external opt-out at the spawn-chain +// gate: with ExternalNatsURL set, Up starts NO nats component (the NatsContainer +// seam is never touched) and nothing nats-shaped is recorded for teardown. The +// rest of the cold chain is unchanged. +func TestExternalNatsSkipsNats(t *testing.T) { + cfg, h := newHarness(t) + cfg.ExternalNatsURL = "nats://nats.example.com:4222" + + s, err := Up(context.Background(), cfg, h.deps) + if err != nil { + t.Fatalf("Up() = %v, want nil", err) + } + if s.attached { + t.Fatal("cold Up should not be attached") + } + if h.nats.started != 0 { + t.Fatalf("NatsContainer.Start called %d times on the --nats-external path, want 0", h.nats.started) + } + rec, rerr := readPgidFile(cfg.StateDir) + if rerr != nil { + t.Fatalf("readPgidFile = %v", rerr) + } + for _, e := range rec.Entries { + if e.Component == ComponentNats { + t.Fatalf("nats entry recorded on the --nats-external path: %+v", e) + } + } +} + +// TestNatsStartRecordsContainerEntry pins the bundled-nats start path: with +// ExternalNatsURL empty (the default), Up starts nats via the NatsContainer +// seam, probes its monitoring endpoint, and records a v2 container-kind pgid +// entry keyed by the stable per-state-dir name — the teardown identity a fresh +// down reconstructs. An in-process Down then drains it. +func TestNatsStartRecordsContainerEntry(t *testing.T) { + cfg, h := newHarness(t) + cfg.NatsImage = "docker.io/library/nats@sha256:abc" + + s, err := Up(context.Background(), cfg, h.deps) + if err != nil { + t.Fatalf("Up() = %v, want nil", err) + } + if h.nats.started != 1 { + t.Fatalf("NatsContainer.Start called %d times, want 1", h.nats.started) + } + // The readiness probe ran against the spec's monitor endpoint — not the + // client endpoint, which speaks no HTTP. + spec := h.nats.spec() + if got := h.natsProber.lastEndpoint(); got != spec.MonitorEndpoint { + t.Fatalf("NatsProber probed %q, want the spec monitor endpoint %q", got, spec.MonitorEndpoint) + } + + rec, rerr := readPgidFile(cfg.StateDir) + if rerr != nil { + t.Fatalf("readPgidFile = %v", rerr) + } + var entry *pgidEntry + for i := range rec.Entries { + if rec.Entries[i].Component == ComponentNats { + entry = &rec.Entries[i] + } + } + if entry == nil { + t.Fatal("no nats entry recorded on the bundled path") + } + if entry.Kind != entryContainer { + t.Errorf("nats entry kind = %v, want entryContainer", entry.Kind) + } + if entry.ContainerName != natsContainerName(cfg.StateDir) { + t.Errorf("recorded nats name = %q, want %q", entry.ContainerName, natsContainerName(cfg.StateDir)) + } + if entry.Pgid != 0 || entry.StartTime != 0 { + t.Errorf("nats entry carries pgid/starttime %d/%d, want zero (torn down by name)", entry.Pgid, entry.StartTime) + } + + if err := s.Down(context.Background()); err != nil { + t.Fatalf("Down() = %v", err) + } + assertPgidFileGone(t, cfg.StateDir) +} + +// TestNatsStartFailureDrains pins the failure surface on the nats path: a nats +// container that fails to launch surfaces the error and leaves no half-started +// stack (compass-server never started after it). +func TestNatsStartFailureDrains(t *testing.T) { + cfg, h := newHarness(t) + h.nats.startErr = errNotAnswering + + if _, err := Up(context.Background(), cfg, h.deps); err == nil { + t.Fatal("Up() = nil, want the nats start error") + } + for _, e := range h.rec.snapshot() { + if e == "start compass-server" { + t.Fatal("compass-server started after a failed nats start") + } + } +} + +// TestNatsNeverReady pins the nats-readiness budget failure: nats launches but +// its health probe never answers, so waitNats times out with a legible error and +// the children started so far are drained — and crucially compass-server never +// starts against a not-ready broker. +func TestNatsNeverReady(t *testing.T) { + cfg, h := newHarness(t) + h.natsProber.never = true + var ticks int + start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + h.deps.Now = func() time.Time { + ticks++ + if ticks > 4 { + return start.Add(natsReadyPollBudget + time.Second) + } + return start + } + + s, err := Up(context.Background(), cfg, h.deps) + if err == nil { + t.Fatal("Up() = nil, want nats-readiness timeout") + } + if s != nil { + t.Fatal("Up() returned a Stack; want nil") + } + if n := countEvent(h.rec.snapshot(), "start compass-server"); n != 0 { + t.Fatalf("compass-server started %d times before nats was ready; want 0", n) + } + // The launched-but-never-ready nats must be drained on the failure path + // (drainChildren owns s.nats): a regression dropping it from the drain list + // would leak the container here. + if n := countEvent(h.rec.snapshot(), "signal nats"); n != 1 { + t.Fatalf("nats signalled %d times on the never-ready drain; want 1", n) + } + assertLockFree(t, cfg.StateDir) +} diff --git a/go/internal/stack/nats_image.go b/go/internal/stack/nats_image.go new file mode 100644 index 000000000..2eea9bbc7 --- /dev/null +++ b/go/internal/stack/nats_image.go @@ -0,0 +1,36 @@ +//go:build unix + +package stack + +// DefaultNatsImage is the container image the bundled NATS component runs by +// default: the official `nats` image (which ships nats-server), the -alpine +// variant, pinned BY DIGEST. The alpine variant is chosen over the scratch-based +// default tag because the stack bind-mounts a config file and a JetStream data +// dir into it, and a shell-bearing base keeps a live `podman exec` diagnosis of +// a wedged store possible without swapping the image. +// +// The pin is by DIGEST, not a mutable tag, for the same reason +// DefaultCollectorImage and DefaultPostgresImage pin theirs: NATS is a +// supervised component of the installed stack holding the fabric's durable +// JetStream state, and both its config grammar and its on-disk store layout are +// version-sensitive, so a mutable tag would ship an unreviewed server — and an +// unreviewed store-format change — under the installed stack. This is a Go const +// Renovate cannot see, so it moves only via a reviewed manual PR. +// +// Bump procedure: advance the digest below, then re-run the NATS component +// bring-up against the new digest before landing (up -> GET /healthz on the +// monitor port answers 200 -> the JetStream block in /varz reports the rendered +// store_dir and a sync_interval of 100ms -> fresh-process down -> container +// gone), and re-validate the generated config against the new image +// (`nats-server -t -c `), since the config grammar and JetStream +// defaults can drift between server versions. +// The current image also boots with a 0600 config; 0644 is retained only for +// future non-root image variants, so the mode is not load-bearing today. +// +// Resolution provenance: this is the multi-arch image index digest for +// nats:2.14.6-alpine, resolved from the Docker Hub registry +// (docker-content-digest for the OCI image-index media type) and verified to +// pull and boot in-environment on 2026-09-04 — the booted server reported +// version 2.14.6 with JetStream enabled at sync_interval 100ms and answered +// /healthz 200. +const DefaultNatsImage = "docker.io/library/nats@sha256:ad7a43eb7e3337c3c38ce5d784d1461791f95f730f252d2b25eee699752a0ca3" diff --git a/go/internal/stack/pgidfile.go b/go/internal/stack/pgidfile.go index c86f35f21..db65e00ad 100644 --- a/go/internal/stack/pgidfile.go +++ b/go/internal/stack/pgidfile.go @@ -299,8 +299,8 @@ func parseContainerEntry(line string, f []string) (pgidEntry, error) { return pgidEntry{Kind: entryContainer, Component: comp, ContainerName: name}, nil } -// componentFromString is the inverse of Component.String for the three -// supervised children. It is defined here beside the parser (its only consumer) +// componentFromString is the inverse of Component.String for every supervised +// child. It is defined here beside the parser (its only consumer) // rather than on Component, keeping the pgid file format self-contained. func componentFromString(s string) (Component, bool) { switch s { @@ -312,6 +312,8 @@ func componentFromString(s string) (Component, bool) { return ComponentRunner, true case ComponentCollector.String(): return ComponentCollector, true + case ComponentNats.String(): + return ComponentNats, true default: return 0, false } diff --git a/go/internal/stack/postgres_container_test.go b/go/internal/stack/postgres_container_test.go index 519e7e340..0ede13fa1 100644 --- a/go/internal/stack/postgres_container_test.go +++ b/go/internal/stack/postgres_container_test.go @@ -38,6 +38,7 @@ func TestExternalDatabaseSkipsPostgres(t *testing.T) { // postgres step. want := []string{ "start otel-collector", + "start nats", "ensure-cert", "start compass-server", "ensure-token", @@ -92,6 +93,7 @@ func TestContainerPathBuildsSpecAndRecordsContainerEntry(t *testing.T) { want := []string{ "start postgres-container", "start otel-collector", + "start nats", "ensure-cert", "start compass-server", "ensure-token", @@ -163,6 +165,7 @@ func TestContainerPathBuildsSpecAndRecordsContainerEntry(t *testing.T) { wantStops := []string{ "signal compass-runner", "wait compass-runner", "signal compass-server", "wait compass-server", + "signal nats", "wait nats", "signal otel-collector", "wait otel-collector", "signal postgres", "wait postgres", } diff --git a/go/internal/stack/stack.go b/go/internal/stack/stack.go index b7f802d76..56aa89a23 100644 --- a/go/internal/stack/stack.go +++ b/go/internal/stack/stack.go @@ -43,6 +43,16 @@ const ( // collector legibly rather than hanging. collectorReadyPollInterval = 100 * time.Millisecond collectorReadyPollBudget = 30 * time.Second + // natsReadyPollInterval/natsReadyPollBudget bound the nats-readiness poll + // between launching the bundled NATS and the components that connect to it. + // NATS boots fast — no cold init like postgres's initdb — and a single-node + // R1 JetStream store recovers in well under a second at the scales this + // stack runs, so the budget is the same readyPollBudget tier as the + // collector: ample for a cold `podman run` of a present image plus store + // recovery, while still failing a genuinely wedged server legibly rather + // than hanging. + natsReadyPollInterval = 100 * time.Millisecond + natsReadyPollBudget = 30 * time.Second ) // Stack is a supervised embedded stack: the resolved config plus the child @@ -56,6 +66,7 @@ type Stack struct { runner Process pg Process collector Process + nats Process // collectorContainerName is the stable name of the bundled collector // container when it ran (T4); empty on the --otel-external opt-out path. It // is the in-process Down's teardown identity for the collector (the same @@ -66,6 +77,16 @@ type Stack struct { // startCollector captures it off the spec it already builds so waitCollector // reads the readiness target without rebuilding the spec. collectorHealthEndpoint string + // natsContainerName is the stable name of the bundled nats container when it + // ran; empty on the --nats-external opt-out path. Down does not read this + // write-only field: durable teardown identity is the persisted v2 pgid entry. + // Retained for sibling parity and debuggability. + natsContainerName string + // natsMonitorEndpoint is the host loopback HTTP monitoring endpoint the + // bundled nats published when it ran; empty on the opt-out path. startNats + // captures it off the spec it already builds so waitNats reads the readiness + // target without rebuilding the spec. + natsMonitorEndpoint string // pgContainerName is the stable name of the container-backed postgres child // when the container path ran (S4); empty on the process and external paths. // It is the in-process Down's teardown identity for the container (the same @@ -246,6 +267,22 @@ func (s *Stack) spawnChain(ctx context.Context) error { return err } + // 1d. Bundled NATS (the fabric's message broker). Grouped with the other + // infra preconditions — after postgres and the collector, before the TLS + // anchor and the server/runner — because server and runner are the surfaces + // that will CONNECT to it (that cutover is PR3/PR4; nothing in-tree connects + // yet), so the broker must be accepting before a consumer comes up, exactly + // the collector's ordering rationale. On the --nats-external opt-out + // (ExternalNatsURL set) startNats is a no-op and no readiness gate runs: + // consumers point straight at the external URL, so nothing bundled starts. + // Start returns at launch; waitNats is the readiness gate. + if err := s.startNats(ctx); err != nil { + return err + } + if err := s.waitNats(ctx); err != nil { + return err + } + // 2. TLS anchor, expiry-aware (rotates when NotAfter is within the window). cert, err := s.deps.Certs.EnsureCert(ctx, s.cfg.StateDir, s.deps.now()) if err != nil { @@ -374,6 +411,36 @@ func (s *Stack) startCollector(ctx context.Context) error { return s.appendEntry(ComponentCollector, pgidEntry{Kind: entryContainer, Component: ComponentCollector, ContainerName: spec.Name}) } +// startNats brings up the bundled NATS container and records it as a v2 +// container entry (torn down by name, never by pgid — a rootless container runs +// beneath conmon, outside the client's process group), exactly like +// startCollector. The Process handle is held on the Stack so the in-process Down +// drains it (Signal → podman stop); the persisted name is what a fresh +// cross-process down reconstructs and signals. On the --nats-external opt-out +// (ExternalNatsURL set) it is a no-op: no bundled nats starts, nothing is +// recorded, and a down tears down only the other children — the nats analogue of +// startCollector's ExternalOTLPEndpoint early return. +func (s *Stack) startNats(ctx context.Context) error { + if s.cfg.ExternalNatsURL != "" { + return nil + } + spec, err := natsContainerSpec(s.cfg) + if err != nil { + return err + } + if s.deps.NatsContainer == nil { + return errors.New("start nats: NatsContainer dep is nil on the bundle path (ExternalNatsURL unset but no nats adapter wired) — a legible failure, not a nil-deref panic") + } + n, err := s.deps.NatsContainer.Start(ctx, spec) + if err != nil { + return fmt.Errorf("start nats container: %w", err) + } + s.nats = n + s.natsContainerName = spec.Name + s.natsMonitorEndpoint = spec.MonitorEndpoint + return s.appendEntry(ComponentNats, pgidEntry{Kind: entryContainer, Component: ComponentNats, ContainerName: spec.Name}) +} + // 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 @@ -481,8 +548,40 @@ func (s *Stack) waitCollector(ctx context.Context) error { } } +// waitNats polls NatsProber.ProbeNats until the bundled NATS answers healthy on +// its HTTP monitoring endpoint or the budget elapses — a direct mirror of +// waitCollector for the nats precondition, since NatsContainer.Start returns at +// launch. On the --nats-external opt-out no nats was started (deps.NatsProber is +// nil), so the gate is skipped entirely — mirroring how startNats no-ops on that +// path. A budget timeout is a legible error the caller renders as Failed. The +// poll respects ctx cancellation. +func (s *Stack) waitNats(ctx context.Context) error { + if s.cfg.ExternalNatsURL != "" { + return nil + } + if s.deps.NatsProber == nil { + return errors.New("wait nats: NatsProber dep is nil on the bundle path (ExternalNatsURL unset but no nats adapter wired) — a legible failure, not a nil-deref panic") + } + deadline := s.deps.now().Add(natsReadyPollBudget) + ticker := time.NewTicker(natsReadyPollInterval) + defer ticker.Stop() + for { + if err := s.deps.NatsProber.ProbeNats(ctx, s.natsMonitorEndpoint); err == nil { + return nil + } + if !s.deps.now().Before(deadline) { + return fmt.Errorf("nats did not answer healthy within %s", natsReadyPollBudget) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + // drainChildren signals and waits each owned child in reverse start order -// (runner → server → collector → postgres). It is safe to call with nil handles +// (runner → server → nats → collector → postgres). It is safe to call with nil handles // (a mid-sequence failure) and on an attached stack (all nil). func (s *Stack) drainChildren(ctx context.Context) error { var errs error @@ -492,6 +591,7 @@ func (s *Stack) drainChildren(ctx context.Context) error { }{ {"compass-runner", s.runner}, {"compass-server", s.server}, + {"nats", s.nats}, {"otel-collector", s.collector}, {"postgres", s.pg}, } { @@ -506,6 +606,6 @@ func (s *Stack) drainChildren(ctx context.Context) error { errs = errors.Join(errs, fmt.Errorf("wait %s: %w", c.name, err)) } } - s.runner, s.server, s.collector, s.pg = nil, nil, nil, nil + s.runner, s.server, s.nats, s.collector, s.pg = nil, nil, nil, nil, nil return errs } diff --git a/go/internal/stack/stack_test.go b/go/internal/stack/stack_test.go index 479ed7851..cb1975367 100644 --- a/go/internal/stack/stack_test.go +++ b/go/internal/stack/stack_test.go @@ -15,6 +15,7 @@ import ( var coldStartSequence = []string{ "start postgres", "start otel-collector", + "start nats", "ensure-cert", "start compass-server", "ensure-token", @@ -220,10 +221,11 @@ func TestUpServerNeverReady(t *testing.T) { h.deps.Now = func() time.Time { ticks++ // First few reads are the deadline bases (postgres gate, collector gate, - // cert, server-readiness deadline) + polls within budget; after enough - // reads jump past the budget so waitReady gives up deterministically. The - // collector-readiness gate adds one now() read ahead of waitReady. - if ticks > 4 { + // nats gate, cert, server-readiness deadline) + polls within budget; + // after enough reads jump past the budget so waitReady gives up + // deterministically. Each container-readiness gate adds one now() read + // ahead of waitReady, so this threshold tracks their count. + if ticks > 5 { return start.Add(readyPollBudget + time.Second) } return start @@ -326,11 +328,12 @@ func TestDownDrainsReverseAndReleasesLock(t *testing.T) { // stack.pgids is left for a later cross-process down to act on. assertPgidFileGone(t, cfg.StateDir) - // Children stopped in reverse start order: runner → server → collector → - // postgres. + // Children stopped in reverse start order: runner → server → nats → + // collector → postgres. wantStops := []string{ "signal compass-runner", "wait compass-runner", "signal compass-server", "wait compass-server", + "signal nats", "wait nats", "signal otel-collector", "wait otel-collector", "signal postgres", "wait postgres", }