Live-layer reconnect supervisor with immediate-then-backoff retry

CHANGELOG.md

@@ -7,6 +7,68 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **Live-layer reconnect supervisor.** Reconnect handling moves into the
+  `Live` client itself, so any consumer — `for rec in client`,
+  `subscribe_callback`, or `stream_to_file` — benefits. Previously
+  reconnect only existed inside the streaming layer; iteration consumers
+  silently saw an `InvalidStateException` on a TCP drop. There is now
+  one reconnect codepath: `_run_unified_session` constructs a single
+  Live with `reconnect_policy = ReconnectPolicy.RECONNECT`, the
+  supervisor handles drops, and the streaming layer no longer carries
+  its own outer reconnect loop.
+- **Hybrid immediate-then-backoff retry schedule.** Under
+  `reconnect_policy = :reconnect`, the first
+  `immediate_reconnect_attempts` retries (default 3) fire with no sleep
+  to catch sub-second TCP blips, then fall back to PR #16's full-jitter
+  exponential backoff (1s base, 60s cap). The retry budget refreshes
+  whenever the new reader successfully delivers ≥1 data record, so a
+  long-lived session that streams real data between drops keeps its
+  full `max_reconnect_attempts` budget across the connection lifetime.
+- **`add_reconnect_callback(client, cb)`.** Register
+  `cb(gap_start_ns::Int64, gap_end_ns::Int64)` to observe each successful
+  reconnect. `gap_start_ns` is the min per-instrument timestamp seen
+  pre-drop; `gap_end_ns` is the gateway's `Metadata.start_ts` from the
+  freshly reconnected session. Useful for gap-fill from historical,
+  alerting, or metric emission. Callbacks are lock-protected and
+  errors are logged-and-swallowed.
+- **`live_session(fn; dataset, subscriptions, kwargs...)`.** Convenience
+  do-block bundling `connect! → subscribe!(many) → start! → fn(client)
+  → close`. Defaults `reconnect_policy = :reconnect` so the simple-API
+  path gets the supervisor by default.
+- **`ReconnectPolicy.{NONE, RECONNECT}` enum is now load-bearing.**
+  Previously exposed for parity but unused by code; now drives whether
+  `start!` spawns the supervisor task.
+- Gateway `ErrorMsg` records in typed mode now set
+  `client.terminal_error` *before* being forwarded to `control_channel`,
+  so the supervisor refuses to reconnect (gateway-side errors are
+  deterministic — retrying would just re-hit the same condition).
+
+### Changed
+- **`Live(...)` default `reconnect_policy` is now `RECONNECT`.** Bare
+  `Live(...)` clients automatically reconnect on TCP drops. Pass
+  `reconnect_policy = :none` for the previous single-shot behaviour
+  (the reader's exit closes the channels and terminates iteration).
+- `_run_unified_session` no longer owns a reconnect loop — it leans on
+  the supervisor. `stream_to_file` / `stream_multi_to_files` users now
+  get the same immediate-then-backoff schedule as iteration consumers
+  (previously their first retry waited at least `_RECONNECT_BASE_S = 1s`).
+- `SessionStats.last_ts_event_by_id` / `last_ts_recv_by_id` removed —
+  replay bookkeeping lives entirely on `Live` (populated by the reader,
+  consumed by the supervisor). `_replay_start_ts` now only has the
+  `::Live` overload; the `::SessionStats` overload is gone.
+- `_reconnect_delay(attempt; immediate=0, base, cap)` gains the
+  `immediate` kwarg. Default `0` preserves the v0.1.1 backoff curve
+  exactly; the Live supervisor passes the user's
+  `immediate_reconnect_attempts`.
+- Under `reconnect_policy = :reconnect`, `start!` does NOT bind the
+  user-facing channels to the reader task. The channels are owned by
+  the Live client and outlive any single reader incarnation so the
+  supervisor can respawn a reader writing into the same channels.
+  Channels are still closed explicitly by `Base.close(client)` and on
+  supervisor terminal-state transitions (`:failed` / `:closed`), so
+  iteration consumers still terminate cleanly on user shutdown.
+
 ## [0.1.1] - 2026-05-26
 
 First release after v0.1.0. Bundles all work since the initial registry tag —

benchmark/PERF_REPORT.md

@@ -498,3 +498,134 @@ Per-suite invocations (faster iteration):
 julia --project=benchmark -e 'include("benchmark/bench_dbn_read.jl"); BenchDBNRead.run(tiers = (:small, :medium))'
 julia --project=benchmark -e 'include("benchmark/profile_hotspots.jl"); ProfileHotspots.run(tier = :medium, seconds = 5.0)'
 ```
+
+---
+
+# Reconnect-supervisor regression analysis (PR #22)
+
+**Date:** 2026-06-01 · **Hardware:** Windows 11 (10.0.26200), x86_64, 24 logical
+cores, 64 GB · **Julia:** 1.13.0-beta3 · **Method:** A/B of `origin/main`
+(7a91c6c) vs `feat/live-reconnect-supervisor` (59cee8c) in two git worktrees,
+each devving its own branch's `DatabentoAPI` source through an *identical*
+benchmark harness so only the package source differs.
+
+### The change under test
+
+PR #22 adds `_record_replay!(c::Live, rec)` to the reader hot path
+(`src/live/reader.jl`). It updates two per-instrument `Dict{UInt32,Int64}` maps
+(`last_ts_event_by_id`, `last_ts_recv_by_id`) and refreshes the retry budget on
+every data record, so the reconnect supervisor can compute the resubscribe
+`start=` after a TCP drop. On `main` the *reader* does a bare `put!`; the same
+two Dict writes lived in `_handle_data!` on the **`stream_to_file` consumer
+loop** (`streaming.jl`). The PR **relocates** those writes consumer→producer
+(and removes them from the consumer).
+
+### Why the obvious benchmark hides the cost
+
+The replay/firehose benches are **channel-bound** — the consumer `take!` loop
+is the floor and the channel saturates (see "Replay-stress" above). In that
+regime producer-side work has slack and is invisible to end-to-end throughput.
+Confirmed: the end-to-end SPY.OPT A/B (7.72 M CBBO records, identical harness,
+5 samples) is dominated by Union-channel GC noise — GC fraction swings
+4.8 %↔13.2 % between otherwise-identical configs and `:reconnect` sometimes
+"beats" `:none`, i.e. the ±10–15 % run-to-run noise is larger than the effect:
+
+| config | wire | M rec/s (median [min–max]) |
+|---|---|---|
+| main          | plain | 3.90 [3.42–4.04] |
+| PR `:none`     | plain | 3.18 [3.06–3.42] |
+| PR `:reconnect`| plain | 3.60 [3.02–3.79] |
+| main          | zstd  | 2.16 [2.10–2.36] |
+| PR `:none`     | zstd  | 2.00 [1.94–2.09] |
+| PR `:reconnect`| zstd  | 1.97 [1.91–1.97] |
+
+So the headline number must come from an **isolated** measurement, not the
+saturated pipeline.
+
+### Isolated per-record cost (the load-bearing measurement)
+
+In-process decode of the real SPY.OPT CBBO bytes (13,604 distinct
+instrument_ids — real OPRA cardinality) through the reader's exact pipeline
+(`CountingIO → BufferedReader → DBNDecoder`), **no channel/consumer**, min of 9
+(`benchmark` tmp driver, reproducible):
+
+| arm | ns/record | M rec/s |
+|---|---|---|
+| A — decode only (≈ main reader inner work)        | 40.4 | 24.8 |
+| B — decode + `_record_replay!` (≈ PR reader work) | 68.5 | 14.6 |
+| **Δ added by `_record_replay!`** | **+28 ns/rec** | — |
+
+Microbench of `_record_replay!` alone vs instrument cardinality M (2 M synthetic
+TradeMsg, both Dicts exercised — TradeMsg carries `ts_recv`), **steady state**:
+
+| M (distinct iids) | 1 | 100 | 1 000 | 50 000 | 200 000 |
+|---|---|---|---|---|---|
+| ns/record | 8.0 | 8.5 | 8.8 | 17.6 | 23.2 |
+| bytes/record | 0 | 0 | 0 | 0 | 0 |
+
+Per-record cost is **allocation-free** (writes hit existing Dict capacity) and
+scales mildly with cardinality (cache pressure on larger Dicts). The 28 ns on
+real CBBO data sits above the synthetic 23 ns at 200 k because the real records
+are larger and the two Dict probes miss cache more often.
+
+One-time memory: the two Dicts grow to the subscribed instrument universe —
+**1.6 MB at 13.6 k instruments, 13 MB at 200 k, ~104 MB for a full ~1.5 M
+OPRA-universe subscription** (≈70–125 bytes/instrument). Bounded, not
+per-record.
+
+### Verdict — net effect by consumer
+
+- **`stream_to_file`: no regression.** The two Dict writes merely moved from the
+  consumer thread to the reader thread; total work per record is unchanged. The
+  reader's 14.6 M rec/s decode+bookkeeping ceiling is far above any disk+zstd
+  consumer, so the reader is never the bottleneck → end-to-end unaffected.
+- **`for rec in client` / `subscribe_callback`: small added cost.** These
+  consumers did *no* replay bookkeeping on `main` (and had no reconnect support).
+  PR adds ~28 ns/record — measurable only when the consumer is trivial
+  (synthetic `bench_live_reader` single-symbol: 7.03→6.62 M rec/s plain,
+  **−5.8 %**, identical 8.17 MB alloc); masked to ~0 under any real consumer or
+  realistic arrival rate. For context, the real OPRA gateway delivers these
+  10-name subscriptions at ~288 k rec/s (see "Live-network stress" above) — 50×
+  below the decode+bookkeeping ceiling. No production impact.
+- **Allocation: no regression.** Per-record allocation is unchanged (0
+  bytes/rec added); the only new allocation is the bounded one-time Dict growth
+  above.
+
+### Optimization opportunity (not blocking)
+
+Under `reconnect_policy = :none` the bookkeeping in `_record_replay!` is never
+read (nothing reconnects), yet it still runs on every record. Gating the Dict
+writes on `policy != NONE` would return single-shot consumers to the `main` hot
+path exactly. Low-risk, ~28 ns/record back for `:none` users.
+
+### Two issues surfaced while building this analysis
+
+1. **Benchmark harness was broken after the `DBN.jl → DatabentoBinaryEncoding.jl`
+   rename.** `benchmark/Project.toml` still declared `DBN = {path =
+   "../../DBN.jl"}` and `fixtures.jl` pointed `DBN_DATA` at the old path, while
+   the bench `.jl` files already `import DatabentoBinaryEncoding` — so the suite
+   could not be instantiated. Fixed in this PR (Project.toml dep + source,
+   `fixtures.jl` path). The PR's original perf table could not have come from the
+   committed harness.
+2. **DBN v3 live captures crash the untyped reader** with `ArgumentError:
+   invalid value for Enum SType: 255` on the first symbol-mapping/control record
+   (reproduced on both Julia- and Python-captured `OPRA.PILLAR` cbbo-1s files;
+   the v1 `SPY.OPT` batch fixture decodes fine). This is a pre-existing decoder
+   gap in `DatabentoBinaryEncoding.jl` (255 = unset stype not mapped), **present
+   on both `main` and this PR** — orthogonal to #22, but it blocks using real v3
+   live captures as throughput fixtures and likely affects real v3 live
+   streaming through the generic reader. Worth a separate issue.
+
+### How to reproduce this section
+
+```bash
+# Decompress the clean v1 fixture once
+julia --project=benchmark -e 'using CodecZstd,TranscodingStreams; \
+  open("/tmp/SPY_plain.dbn","w") do o; open(f->write(o,TranscodingStream(ZstdDecompressor(),f)), \
+  "benchmark/data/replay_SPY.OPT_2026-05-13T13_30_00__2026-05-13T13_31_00.dbn.zst"); end'
+
+# End-to-end A/B (run in a main worktree and a PR worktree)
+julia --project=benchmark -e 'include("benchmark/bench_live_replay.jl"); using .BenchLiveReplay; \
+  BenchLiveReplay.replay(path="benchmark/data/replay_SPY.OPT_….dbn.zst", schema=Schema.CBBO_1S, \
+                         reconnect_policy=:none)'   # add reconnect_policy only on the PR branch
+```

benchmark/Project.toml

@@ -1,8 +1,8 @@
 [deps]
 BenchmarkTools = "6e4b80f9-dd63-53aa-95a3-0cdb28fa8baf"
 CodecZstd = "6b39b394-51ab-5f42-8807-6242bab2b4c2"
-DBN = "90689371-c8cb-40d1-831f-18033db90f74"
 DatabentoAPI = "feb0aed8-3291-460b-9688-521dcb17a4bf"
+DatabentoBinaryEncoding = "90689371-c8cb-40d1-831f-18033db90f74"
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
 HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3"
 Printf = "de0858da-6303-5e67-8744-51eddeeeb8d7"
@@ -12,5 +12,5 @@ Statistics = "10745b16-79ce-11e8-11f9-7d13ad32a3b2"
 TranscodingStreams = "3bb67fe8-82b1-5028-8e26-92a6c54297fa"
 
 [sources]
-DBN = {path = "../../DBN.jl"}
 DatabentoAPI = {path = ".."}
+DatabentoBinaryEncoding = {path = "../../DatabentoBinaryEncoding.jl"}