Wire the initial snapshot into startup

[lukashes]

Closes the #49 series (PR3 of 3): drives the snapshot reader from #145 at startup and makes an interrupted snapshot safe.

Problem

Streaming only emits changes made after the slot was created, so a new consumer never sees pre-existing rows. The reader and the READ event landed in #145 but nothing called them.

Solution

Wiring:

• main.zig: connect and ensure the slot without streaming, run the snapshot, then START_REPLICATION. Order matters: START_REPLICATION invalidates the slot's exported snapshot, so the reads happen first on a second regular connection bound to that snapshot.
• Gate: the snapshot runs only when the slot is created this run, [snapshot].mode is initial (default; no-snapshot disables), and at least one stream lists read. Only the distinct read-opted resources are snapshotted.
• processor.runInitialSnapshot produces READ events through the same match/serialize/partition/produce path as streaming (factored into produceEvent). READ rows carry the slot's start LSN, so snapshot and stream share one boundary. pending_lsn stays 0, so the slot advances only once the stream is confirmed.
• replication_protocol.zig captures the slot's snapshot_name alongside consistent_point.

Interrupted-snapshot recovery:

• The streaming publication stays before the slot, so changes made during the snapshot still decode: pgoutput resolves the publication by name in the historical catalog per change, so a change that predates the publication cannot be decoded (verified on PG17).
• A transient marker publication <pub>_snapshotting flags an in-progress snapshot: created before the slot, dropped once the snapshot is flushed. On restart a slot with the marker present is dropped and the whole bootstrap is redone from a fresh consistent point; a slot without it resumes. The marker is never passed to START_REPLICATION, so steady state keeps a single publication.
• A signal-interrupted snapshot leaves the marker in place and exits before streaming, so the next start redoes it. No rows are missed, because the redo re-reads current state. Consumers must treat READ as an upsert.

Tests

• Unit: snapshot.mode validation, wantsInitialSnapshot, Stream.hasReadOperation.
• Integration: reconciliation (interrupted snapshot recreates the slot from a fresh LSN with the marker still present; a completed one resumes without re-snapshotting and leaves one publication).
• E2E: seed rows before slot creation, run the snapshot, insert a live row after streaming starts, and assert READ (with the start LSN) for the seeded rows and INSERT for the live one, with no gap or overlap.

Unit, integration, and e2e pass locally.

[github-actions]

📊 Benchmark Results

Current run is the minimum over 3 passes, compared against the base branch (main @ 2a225de), built on the same runner.

Benchmark	Baseline	Current	Δ Time	Allocs	Status
Converter DELETE	141.85μs	142.11μs	+0.2%	13 → 13	➡️
Converter INSERT	142.12μs	142.24μs	+0.1%	13 → 13	➡️
Converter UPDATE	171.57μs	172.38μs	+0.5%	22 → 22	➡️
JsonSerializer	49.27μs	49.12μs	-0.3%	3 → 3	➡️
KafkaProducer produce	230.20μs	201.08μs	-12.6%	1 → 1	🟢 faster
KafkaProducer send	247.30μs	249.58μs	+0.9%	0 → 0	➡️
PgOutputDecoder	94.99μs	95.24μs	+0.3%	6 → 6	➡️
getPartitionKeyValue boolean	15.41μs	15.42μs	+0.0%	1 → 1	➡️
getPartitionKeyValue integer	0.04μs	0.04μs	+0.0%	0 → 0	⚪ noise
getPartitionKeyValue not found	0.03μs	0.03μs	+3.1%	0 → 0	⚪ noise
getPartitionKeyValue string	17.19μs	17.23μs	+0.3%	1 → 1	➡️
matchStreams found	17.12μs	17.21μs	+0.5%	1 → 1	➡️
matchStreams not found	0.05μs	0.05μs	+0.0%	0 → 0	⚪ noise

Summary: 🟢 1 faster · ➡️ 9 neutral · ⚪ 3 ignored (sub-μs)

_{Thresholds: <1μs ignore · 1–20μs 15% · 20–50μs 10% · ≥50μs 5%. Measured on a shared CI runner — treat small deltas as noise. Informational only; this check never fails the build.}

[lukashes]

/bench