Drive the S2 storage sink from its controller

[archandatta]

Stacked on #401, which adds events.S2StorageController. This layer replaces the ad-hoc atomic.Pointer[events.S2StorageWriter], sync.WaitGroup, and startS2Writer lifecycle in cmd/api/main.go with that controller, leaving main.go net −6 lines. The sink still starts at boot when the instance is not waiting for a fork identity or has already applied one, and otherwise from the fork-identity hook, off the handoff critical path per the contract documented on forkIdentityHandler. api.New also takes the controller behind a new S2Storage interface for the later telemetry-side integration; nothing reads it yet.

Stream selection keeps both existing paths, and keeps the hook's payload authoritative by construction rather than by disk state. Boot and API-process restarts resolve the persisted applied payload through appliedS2Stream. A successful fork hook stores its own payload's S2_STREAM as an in-process override before starting the controller, so resolution no longer depends on appliedS2Stream re-reading ReadyFile, the applied marker, and the payload file. That matters because the controller binds one stream for the life of the instance and never retries: had that re-read fallen through to the boot config.S2Stream, a fork would have written its telemetry into the stream of the instance it forked from, permanently and silently. The wrapper writes ReadyFile before starting the API whenever fork-identity wait is armed, so this closes a latent failure mode rather than an observed production failure. The override is atomic because resolution and the asynchronous start run on different goroutines. A missing S2_STREAM still falls back to the boot configuration.

Shutdown still drains the HTTP servers first, then S2, then OTLP. The controller waits for an in-flight start and bounds the whole stop with the existing 10-second context. The old 2-second abandonment guard is gone: in s2-sdk-go v0.22.1, AppendSession (s2/append_session.go:54) only starts a pump goroutine and returns, and the transport session is created from processInflightQueue after the first record submit, so S2StorageWriter.Start never holds w.mu across a network dial. That guard was skipping the drain — losing the shutdown window's events — to avoid a block that cannot occur.

Existing behavior is otherwise unchanged: missing credentials or a missing stream keep the sink closed, concurrent starts open at most one writer, a failed start stays retryable, a successful start emits exactly one S2 storage enabled line, and a boot failure in the optional sink does not crashloop the browser. S2StorageWriter, StorageWriter, the telemetry handlers, and storage configuration are untouched. Deferring startup to PUT /telemetry remains a later change.

Tests cover boot resolution, applied-identity restarts, the missing-ready-file hook path, and the surrounding API packages. go build ./... and go vet ./... are clean at 0fe7b21. go test $(go list ./... | grep -v /e2e$) -count=1 -race passes all 38 non-e2e packages; an earlier run hit TestUpstreamManagerDetectsChromiumAndRestart in lib/devtoolsproxy on its own t.TempDir() teardown racing a Chromium profile directory — a package this PR does not touch — and it has not recurred at -count=3. TestS2StreamResolverUsesHookPayloadWithoutReadyFile fails under the pre-fix mutation, returning seed-stream instead of fork-stream. A headless image built from 0fe7b21 was booted in three modes, reading the API log at /var/log/supervisord/kernel-images-api rather than docker logs: with S2 unset, 0 S2 storage enabled lines; with fake credentials, exactly 1; and with KERNEL_FORK_IDENTITY_WAIT=true plus a seed S2_STREAM=a2-seed-stream, 0 at boot and exactly 1 after POST /internal/fork-identity returned 204, bound to the payload's a2-fork-stream rather than the seed. In each mode GET /spec.yaml and GET /telemetry returned 200, PUT /telemetry 201, POST /telemetry/events 200 with the envelope arriving on GET /telemetry/stream, and docker stop -t 30 exited 0 with no drain warning. The full e2e suite passes against locally built headless and headful images: go test ./e2e/ -count=1 -timeout 110m with E2E_CHROMIUM_HEADLESS_IMAGE and E2E_CHROMIUM_HEADFUL_IMAGE set gives 52 passed, 0 failed, 2 skipped (34 subtests, all passing). TestOTLPExportForkIdentityRefresh is the closest analogue to this change: it drives the real /internal/fork-identity endpoint and confirms the sibling sink retargets after the handoff. Of the two skips, TestReplayRecordingZombocomArchiveAudio needs a network fixture, and TestS2StorageWriter (server/e2e/e2e_s2_storage_test.go) self-skips unless S2_BASIN, S2_ACCESS_TOKEN, and S2_STREAM are set — no S2 credentials are available here, so nothing has verified a record landing in a real S2 stream.

That last gap was closed against a local mock instead. The SDK's append session is a bidirectional protobuf stream, so a stand-in basin serving h2c was pointed at via S2_BASIN_ENDPOINT, which required a throwaway local patch (newS2Storage passes nil where s2.LoadConfigFromEnv() would be needed) that was not committed. Against it: three published events arrived as one batched append; three more published immediately before SIGTERM all arrived during the drain, with the API log showing shutdown signal received and no stop failed, drain incomplete, or ack failed; and a fork container seeded with S2_STREAM=seed-stream opened its session on /v1/streams/fork-stream/records after applying a payload naming fork-stream. The mock also timestamps the dial: S2 storage enabled logged at 14:18:35 and the append session opened only on first record submit, confirming the lazy-dial premise for removing the 2-second guard.

---

Note

Medium Risk
Changes telemetry persistence lifecycle and fork-time S2 stream selection; a wrong bind is permanent, though behavior is covered by new tests and matches prior boot/defer semantics.

Overview
Replaces the ad-hoc S2 writer lifecycle in cmd/api/main.go (atomic.Pointer, WaitGroup, startS2Writer) with events.S2StorageController, wired through boot, fork-identity hook, and shutdown (HTTP drain first, then bounded Stop).

Adds s2StreamResolver so stream name comes from persisted appliedS2Stream on boot/restart, and from an in-process atomic override when the fork hook runs—so the controller binds the payload’s S2_STREAM even before ReadyFile/disk state would make appliedS2Stream trustworthy (avoids silently appending to the parent’s stream).

Plumbs the controller into api.New via a new S2Storage interface on ApiService (parallel to OTLP); handlers do not use it yet. Tests gain a hook-path case (TestS2StreamResolverUsesHookPayloadWithoutReadyFile) and updated New(..., nil, nil) call sites.

^{Reviewed by Cursor Bugbot for commit 68dabd8. Bugbot is set up for automated code reviews on this repo. Configure here.}

[Sayan-]

yeet