Live-layer reconnect supervisor with immediate-then-backoff retry#22
Merged
Conversation
Adds `immediate::Integer = 0` to `_reconnect_delay`. When set, the first N attempts return 0.0 (no sleep), and the backoff phase is re-indexed to start at `attempt = immediate + 1` so the first jittered window is still [0, base) rather than a pre-stretched one. Default 0 preserves PR #16 behaviour: existing test bounds, jitter variance, max-attempts cap, and reconnect=false short-circuit all unchanged. The upcoming Live-layer supervisor will pass `immediate=3` to opt into a brief no-sleep phase that catches sub-second TCP blips without the 1s+ backoff floor. Adds an `immediate phase returns exactly 0.0` testset: first N attempts are exactly 0.0 across configurable N; attempt = immediate+1 enters backoff at [0, base); attempt = immediate+2 at [0, 2*base); large immediate suppresses backoff entirely. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Pure refactor. Splits the TCP connect + CRAM handshake + auth body of connect! into a private _open_socket_and_auth!(c) so the upcoming reconnect path can reuse it without re-entering connect!'s public guards (which short-circuit on c.connected and throw on c.closed). connect! itself becomes a thin wrapper that owns the lifecycle flag transition to c.connected = true. No behaviour change. Side-effect order on the Live struct (socket, lsg_version, session_id, connected) preserved. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Adds the field layout the supervisor and reconnect machinery will use, along with the constructor kwargs. Fields are populated but unused — this commit is structurally additive, no behaviour change. New fields on `mutable struct Live`: - replay bookkeeping: last_ts_event_by_id, last_ts_recv_by_id, last_metadata_start_ns (to-be-populated by reader and reconnect path) - policy + budget: reconnect_policy (ReconnectPolicy.T enum), max_reconnect_attempts (Int|nothing), immediate_reconnect_attempts, attempts_remaining, records_since_reconnect - callback registry: reconnect_callbacks (Vector), callbacks_lock - state machine: reconnect_supervisor (Task|nothing), state (Symbol initialised to :fresh), state_lock, terminal_error New constructor kwargs with defaults that preserve current behaviour: - reconnect_policy = ReconnectPolicy.NONE (will flip to RECONNECT in the final commit, once supervisor + reader + callback wiring are all live; until then NONE is correct since the supervisor doesn't exist yet) - max_reconnect_attempts = 10 - immediate_reconnect_attempts = 3 `reconnect_policy` accepts the enum directly, or Symbol/String spellings (`:none` / `:reconnect`) for ergonomic call sites — coerced through _coerce_reconnect_policy. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
…ive, ...) Adds the reader-side wiring that keeps Live's per-instrument replay state current. The upcoming Live-layer reconnect supervisor reads from these dicts to compute the resubscribe `start=` timestamp for each schema. This commit is additive: the legacy `_run_unified_session` reconnect loop still reads from its SessionStats copy of the same data (populated by _handle_data! in the data-consumer path). Both paths now track the same values; the streaming-layer migration happens in a later commit when the inline reconnect loop is removed. reader changes: - _record_replay!(c, rec): @inline helper; hasproperty checks fold against the concrete record type at each typed-reader callsite so it's free for records without ts_recv. - _put_data!(c, ch, rec): @inline helper that updates replay state then put!s. Substituted for `put!` in all 17 typed data-record branches of _reader_loop_typed (MBP_1/TBBO broadcast does its replay update once for the shared record, then puts to both subscribed channels). - Untyped _reader_loop: _record_replay! call inserted before put!. streaming changes: - _replay_start_ts(c::Live, schema) overload mirrors the existing SessionStats overload, reading from c.last_ts_*_by_id instead. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Under ReconnectPolicy.NONE the reader task owns the channels' lifetime exactly as before: start! binds them, and the reader's finally block closes them on exit. Iterators (and subscribe_callback consumers) see an InvalidStateException on take! and exit cleanly — the right shutdown signal when no replacement reader is coming. Under ReconnectPolicy.RECONNECT the channels are owned by the Live client and outlive any single reader incarnation. start! skips the `bind` and the reader's finally skips the close — the upcoming supervisor will spawn a fresh reader writing into the same channels after each drop, so consumers see a continuous record stream across reconnects without the bind/close cycle terminating their iteration mid-stream. The user-initiated close path (c.closed=true) always closes regardless of policy, so explicit `close(client)` still terminates iterators in both modes. Default policy is still NONE, so observable behaviour is unchanged in this commit. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Implements the Live-layer reconnect machinery so iteration consumers (`for rec in client` / `subscribe_callback`) survive TCP drops, not just `stream_to_file` callers. Wires under ReconnectPolicy.RECONNECT (opt-in; default still :none in this commit, flipped in the final commit). New src/live/reconnect.jl: - _supervisor_loop(c): spawned from start! when policy != :none. Blocks on wait(reader_task); on exit, checks terminal flags (c.closed, c.terminal_error, c.attempts_remaining <= 0) and either retries via _reconnect_once! or transitions to :failed/:closed and closes channels. - _reconnect_once!(c): closes stale socket, runs the extracted _open_socket_and_auth!, re-issues stored subscriptions via private _send_subscribe_frame (each with start_override from _replay_start_ts(c, schema)), sends start_session, spawns new reader against the *same* channels. Polls for new reader's metadata.start_ts before returning so the reconnect callback fires with a fresh gap_end (10s deadline accommodates cold-JIT first-reconnect on a fresh process). - add_reconnect_callback(c, cb): public; cb(gap_start_ns, gap_end_ns). Lock-protected list, fire outside lock so a slow user callback can't block registration. Errors logged and swallowed. - BentoAuthError during reconnect is terminal (no point retrying deterministic auth failures); other exceptions decrement budget and retry per the schedule from step 1. - gap_start = min across last_ts_*_by_id (earliest "we lost coverage" across subscribed instruments). Reader changes: - _record_replay! now also refreshes c.attempts_remaining on the first record of each reader incarnation — a session that actually streams real data between drops keeps its full retry budget across the connection lifetime. - _reader_loop[_typed] capture decoder.metadata.start_ts into c.last_metadata_start_ns immediately after read_header!. - Typed reader's control branch: ErrorMsg sets c.terminal_error *before* publishing to control_channel, so the supervisor's next decision sees it and refuses to reconnect (gateway-side errors are deterministic). Lifecycle: - start! spawns the supervisor under RECONNECT policy and transitions state→:streaming. - Base.close(c) transitions state→:closed under state_lock so a racing supervisor sees the terminal flag at its next iteration. Tests (test/test_live_reconnect_supervisor.jl, wired into runtests.jl): - iteration survives a disconnect: 2-script mock; client iterates and collects records from BOTH sessions with no exception in between; reconnect callback fires once with gap_start > 0 and gap_end matching session 2's metadata.start_ts (different from session 1's so the poll-and-update path is exercised). - gateway ErrorMsg is terminal: ErrorMsg in session 1 triggers terminal_error, supervisor refuses to reconnect, channels close, callback never fires, mock's 2nd accept never happens. Full suite: 1762/1762 pass in 48s. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Adds the ergonomic do-block path the issue's "ease of connecting" framing called for, plus the user-facing docs catch-up for the new exports. - live_session(fn; dataset, subscriptions, kwargs...): bundles connect! → subscribe!(many) → start! → fn(client) → close into one call. Defaults reconnect_policy = :reconnect so the simple-API path gets the supervisor by default (Live(...) constructor default stays :none for backward compatibility — see CHANGELOG note; the constructor default flip can land in a follow-up alongside the streaming-layer unification). - Exports: live_session, add_reconnect_callback. - docs/src/api/live.md: adds Convenience and Reconnect sections for the two new exports. - CHANGELOG.md: ## [Unreleased] entry covering the supervisor, hybrid retry schedule, callback registry, live_session wrapper, ErrorMsg terminal handling, and the channel-lifecycle change under RECONNECT. - test_live_reconnect_supervisor.jl: smoke test that live_session drives a single-shot session end-to-end (subscribe → drain → close). Full suite: 1764/1764 pass in 49.6s. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
The streaming layer now drives a single Live with reconnect_policy = ReconnectPolicy.RECONNECT and lets the Live-layer supervisor handle TCP-drop reconnect end-to-end. The outer `while true` retry loop in _run_unified_session is gone, along with _wait_for_reconnect (no longer needed — supervisor's _interruptible_sleep watches c.closed; on deadline/shutdown the streaming layer calls close(client) which the supervisor sees at its next poll). Pulling reconnect into one place delivers the immediate-retry win to stream_to_file / stream_multi_to_files users (previously they only got PR #16's 1s-floor backoff), eliminates two competing reconnect codepaths, and removes ~80 lines of streaming-layer ceremony. Notable preserved behaviours: - max_reconnect_attempts still gives `1 + cap` total attempts. The supervisor covers post-start! drops; the streaming layer keeps a small retry loop around the initial `connect!` (the supervisor only watches a reader task, which only exists after start!). - ctx.stats.reconnects still increments per successful reconnect — now driven by an add_reconnect_callback rather than the outer loop. - ErrorMsg-from-gateway stays terminal: the typed reader's control branch sets c.terminal_error, supervisor refuses to reconnect, and _handle_control! still sets ctx.stats.error so the streaming coordination loop sees `_any_error(ctxs)` and breaks. - 24h replay-window cap (_bound_replay) still applies — used inside the supervisor's _reconnect_once! via _replay_start_ts(c::Live, schema). Cleanup: - SessionStats.last_ts_event_by_id / last_ts_recv_by_id removed — replay bookkeeping lives entirely on Live (populated by the reader's _record_replay!). _handle_data! drops the redundant updates. - _replay_start_ts(::SessionStats, schema) overload removed; the ::Live overload is the only one now. - _wait_for_reconnect removed. - test/test_live_streaming.jl `_replay_start_ts` testset updated to exercise the Live overload (matches the production codepath). Full suite: 1764/1764 pass in 48.6s. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Bare `Live(...)` now defaults to reconnect_policy = RECONNECT — the "reasonable default" once the supervisor is the only reconnect codepath. Tests that intentionally use single-shot reader-exits-channel-closes semantics now pass `reconnect_policy = :none` explicitly: - test_live_reader.jl (1 site) - test_live_reader_typed.jl (5 sites) - test_live_subscribe.jl (3 sites) Tests that construct Live but never call start! (live-streaming.jl do-block + close coverage, typed-mode error coverage) are unaffected — no supervisor spawns without start!. stream_to_file / stream_multi_to_files unaffected: they pass reconnect_policy explicitly to the inner Live based on their own `reconnect::Bool` kwarg (default true). CHANGELOG updated: ## [Unreleased] entry now reflects the single reconnect codepath (no more "deferred unification" caveat) and the default-flip is called out as a Changed item. Full suite: 1764/1764 pass in 48.1s. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
The old guide said "If you're driving the Live client yourself outside stream_to_file, you handle reconnects yourself" — no longer true. Every Live(...) now has a reconnect supervisor on by default. New content covers: - The default-on supervisor and its immediate-then-backoff schedule (3 immediate attempts, then exp backoff, budget refresh on first record post-reconnect) - ReconnectPolicy.NONE opt-out - add_reconnect_callback with the (gap_start_ns, gap_end_ns) signature and DateTime conversion note - Gateway ErrorMsg → terminal_error → no reconnect rationale - live_session convenience wrapper - Updated "Closing the client" paragraph to mention supervisor exit Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
The quickstart's "use stream_to_file for captures that survive disconnects" was misleading after the supervisor lands — every Live now survives disconnects regardless of whether it's writing to disk. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bd50567707
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| # Track per-instrument replay state on the Live client so the | ||
| # reconnect supervisor sees timestamps from records that haven't | ||
| # been consumed yet. | ||
| _record_replay!(c, rec) |
There was a problem hiding this comment.
With the new default reconnect_policy = RECONNECT, the untyped reader path now supervises reconnects too, but it only records/publishes every decoded record here and never sets c.terminal_error when rec isa DBN.ErrorMsg. If an untyped Live session receives a gateway ErrorMsg and the socket then closes, _supervisor_loop sees terminal_error === nothing and reconnects/replays a deterministic failure until the retry budget is exhausted, unlike the typed path that marks ErrorMsg terminal before forwarding it.
Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment.
Good catch — fixed in 59cee8c. The untyped _reader_loop now sets c.terminal_error = String(rec.err) on ErrorMsg before the put!, matching the typed reader's control branch. With the supervisor now on by default, that asymmetry would otherwise burn through max_reconnect_attempts on a deterministic gateway failure.
Added a regression test (ErrorMsg terminal under UNTYPED mode in test/test_live_reconnect_supervisor.jl) that pre-stages 5 mock-gateway scripts, asserts accept_count == 1 (never reconnected), confirms terminal_error is set, and verifies both the trade and the ErrorMsg still flow to the consumer's channel.
Updated comment notes the trade-off: users who want the old "ErrorMsg is informational" behaviour (e.g. for SkippedRecordsAfterSlowReading) should pass reconnect_policy = :none and drive reconnection themselves.
Full suite: 1768/1768 pass.
Fixes a real bug surfaced by Codex review on #22: with the default reconnect_policy now RECONNECT, an untyped client that received a gateway ErrorMsg would loop through max_reconnect_attempts re-hitting the same deterministic failure — the typed reader's control branch short-circuits this by setting c.terminal_error before forwarding, but the untyped reader was left with its pre-supervisor semantics ("forward and let the consumer decide"). Now both reader paths set c.terminal_error on ErrorMsg before publishing the record. The record itself still flows through the channel so consumers can inspect it; the supervisor sees the terminal flag at the next decision check and exits cleanly to :failed. Users who want to treat ErrorMsg as informational (e.g. SkippedRecordsAfterSlowReading is a code-7 ErrorMsg signalling SKIP behavior, not a fatal error) should pass `reconnect_policy = :none` and drive reconnection themselves. Regression test: new `ErrorMsg terminal under UNTYPED mode` testset mirroring the typed-mode one — staged 5 mock scripts, asserts accept_count == 1 (never reconnected), terminal_error is set, and both the trade and the ErrorMsg flowed to the consumer. Full suite: 1768/1768 pass in 52.9s. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Fix the benchmark suite (broken after the DBN.jl -> DatabentoBinaryEncoding.jl rename) and add an A/B regression analysis of the reconnect supervisor's per-record _record_replay! bookkeeping. - benchmark/Project.toml, fixtures.jl: DBN -> DatabentoBinaryEncoding dep, source path, and DBN_DATA fixture path. Suite could not instantiate before. - bench_live_replay.jl: add optional reconnect_policy passthrough (splatted into Live only when set, so the same harness runs on pre-supervisor branches); add an idle-timeout watchdog so the drain loop can't hang when the one-shot mock EOFs under reconnect_policy=:reconnect (the channel never closes); report throughput over the active first->last-record window so the reconnect/idle tail doesn't pollute the rate. Per-record hot path kept identical to the original (take! + counter) so A/B is not skewed. - PERF_REPORT.md: new "Reconnect-supervisor regression analysis" section. Findings: _record_replay! adds ~28 ns/record (allocation-free) at 13.6k instrument cardinality, isolated in-process. It is net-zero for stream_to_file (the two Dict writes merely move consumer->producer) and a ~6% cost for trivial iteration consumers, masked to ~0 under any real consumer or realistic arrival rate. One-time Dict growth ~1.6 MB at 13.6k / ~104 MB at full OPRA universe. Also surfaced: DBN v3 live captures crash the untyped reader (SType 255), a pre-existing decoder gap orthogonal to this PR. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The "duration_s cuts off mid-backoff sleep" test asserted `elapsed < duration + 3.0` (4.5s). This flaked on shared macOS and Windows CI runners (observed 5.0s) where per-attempt socket churn and scheduler stalls add several seconds of wall-clock unrelated to backoff responsiveness. The deadline-slicing in `_wait_for_reconnect` is still verified — it returns within `_CONNECTION_POLL_S` (0.5s) of the deadline, so the loop never sits through a full backoff window (up to 60s). Loosen to `elapsed < 15.0`, matching the sibling "max_reconnect_attempts caps connection attempts" test. A genuine regression that ignored the deadline would run for tens of seconds and still trip this bound. Local: testset runs ~2.1-2.3s across repeated runs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes #17. Reconnect handling moves from the streaming layer into a
supervisor task owned by
Live, so any consumer —for rec in live,subscribe_callback, orstream_to_file— survives TCP drops with thesame code path. Adds an immediate-retry phase to catch sub-second blips,
a user-observable reconnect callback, and a
live_sessionconveniencewrapper.
immediate_reconnect_attemptsretries(default 3) fire with no sleep — catches kernel-side RST + a few
packet retransmits that the 1s+ backoff would otherwise stretch into
multi-second downtime. Subsequent retries use PR Reconnect: exponential backoff with jitter + max_reconnect_attempts #16's full-jitter
exponential backoff (1s base, 60s cap), up to
max_reconnect_attempts(default 10;
nothingfor unlimited). Budget refreshes whenever afresh reader delivers ≥1 record, so a long-lived session keeps full
budget across the connection lifetime.
add_reconnect_callback(client, cb).cb(gap_start_ns::Int64, gap_end_ns::Int64)fires after each successful reconnect.gap_start_nsis the min per-instrument timestamp seen pre-drop;gap_end_nsis the gateway'sMetadata.start_tsfrom the freshsession. Lock-protected list, fire-outside-lock; errors logged and
swallowed.
live_session(fn; dataset, subscriptions, kwargs...). Bundlesconnect! → subscribe!(many) → start! → fn(client) → closeinto onecall. Defaults
reconnect_policy = :reconnect.reconnect_policyonLive(...)flipped toRECONNECT. Passreconnect_policy = :nonefor single-shotsemantics.
stream_to_fileusers unaffected (controlled by its ownreconnect::Boolkwarg)._run_unified_sessionno longer ownsan inline reconnect loop — leans on the supervisor. Replay
bookkeeping (
last_ts_*_by_id) moves fromSessionStatstoLive;_replay_start_ts(::SessionStats, …)overload removed.ErrorMsgis terminal. Typed reader's control branchsets
client.terminal_errorbefore forwarding, supervisor refuses toreconnect (deterministic failures don't benefit from retry).
BentoAuthErrorduring reconnect is also terminal for the samereason.
Eleven commits, each compiles and passes tests; reviewable
commit-by-commit.
What didn't change
_reconnect_delay'simmediatekwarg defaults to
0, preserving the v0.1.1 curve when calledwithout the new arg).
_bound_replay).SymbolMappingMsgcontinues to land in the.dbn.zstvia_handle_control!.RotatingDBNFile, frame rotation, sidecar logic,stream_to_filepublic API surface.
Performance
Apples-to-apples on a 100k synthetic TradeMsg payload, in-process mock
gateway, no compression:
origin/main:none):reconnect)No regression. The supervisor blocks on
wait(reader_task)duringnormal streaming so it adds nothing to the hot path; the added
_record_replay!per-record cost is offset by streaming-layer cleanup(redundant
_handle_data!Dict writes removed).Docs
docs/src/api/live.md: new Convenience + Reconnect sections.docs/src/guide/live.md: full rewrite of the Reconnect section (theold "you handle reconnects yourself" guidance was wrong after the
supervisor).
docs/src/quickstart.md: corrected the "usestream_to_fileforreconnect-surviving captures" sentence (iteration survives drops
too now).
CHANGELOG.md:## [Unreleased]covering Added + Changed.Test plan
Pkg.test().test/test_live_reconnect_supervisor.jl:for rec in clientcollects records from both sessions with no exceptionin between; reconnect callback fires once with
gap_start > 0and
gap_end == session2.metadata.start_ts.after ErrorMsg; channels close; mock's 2nd accept never happens;
callback never fires.
live_sessionbundles lifecycle — single-shot smoke coveringthe new wrapper.
test/test_reconnect_backoff.jlregression checks allgreen: bounds, jitter, max-attempts cap, deadline-cuts-mid-backoff,
reconnect=falseshort-circuit.test/test_live_streaming.jlErrorMsg-terminal test(
elapsed < 4.5swithreconnect=true, duration_s=5) still passesthrough the new supervisor path.
stream_to_file(... duration_s = 600), manually drop the connectionmid-stream, verify (a) reconnect happens silently, (b) gap callback
fires once per drop, (c) resulting
.dbn.zstdecodes cleanly.🤖 Generated with Claude Code