diff --git a/CHANGELOG.md b/CHANGELOG.md index 980e328..6b9a7e1 100644 --- a/CHANGELOG.md +++ b/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 — diff --git a/benchmark/PERF_REPORT.md b/benchmark/PERF_REPORT.md index e5a3ee5..7c3a196 100644 --- a/benchmark/PERF_REPORT.md +++ b/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 +``` diff --git a/benchmark/Project.toml b/benchmark/Project.toml index acc0d65..db47b0f 100644 --- a/benchmark/Project.toml +++ b/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"} diff --git a/benchmark/bench_live_replay.jl b/benchmark/bench_live_replay.jl index 3f401a4..d23bbb9 100644 --- a/benchmark/bench_live_replay.jl +++ b/benchmark/bench_live_replay.jl @@ -71,6 +71,15 @@ end # Drive a Live client against the replay mock, tracking consumer rate, # producer write rate, peak channel depth, and GC pressure. +# +# `reconnect_policy` (`:none` / `:reconnect` / nothing) is splatted into the +# `Live` constructor only when non-`nothing`, so the same harness runs against +# branches whose `Live` predates the kwarg (pass `nothing` there). Under +# `:reconnect` the one-shot mock EOFs and the supervisor keeps retrying a dead +# port, so the channel never closes on its own — the idle watchdog below closes +# the client once `idle_timeout_s` elapses with no new record, and the reported +# rate is computed over the active first→last-record window so the +# reconnect/idle tail does not pollute it. function replay(; path::AbstractString, wire_zstd::Bool = endswith(lowercase(path), ".zst"), dataset::AbstractString = "OPRA.PILLAR", @@ -79,22 +88,24 @@ function replay(; path::AbstractString, stype_in::SType.T = SType.RAW_SYMBOL, channel_size::Integer = 65_536, log_every::Int = 1_000_000, - typed::Bool = false) + typed::Bool = false, + reconnect_policy::Union{Nothing,Symbol} = nothing, + idle_timeout_s::Real = 3.0) bytes = read(path) println(SUITE, ": replaying ", basename(path), " (", round(length(bytes) / (1024*1024), digits = 1), " MB", wire_zstd ? ", wire zstd, expect ~6× expansion" : ", plain", - typed ? ", TYPED mode" : ", untyped mode", ")") + typed ? ", TYPED mode" : ", untyped mode", + reconnect_policy === nothing ? "" : ", reconnect_policy=$(reconnect_policy)", ")") mock = spawn_replay(bytes) comp = wire_zstd ? Compression.ZSTD : Compression.NONE - client = DatabentoAPI.Live(TEST_KEY; - dataset = String(dataset), - gateway = "127.0.0.1", - port = mock.port, - channel_size = Int(channel_size), - compression = comp, - typed = typed) + base_kwargs = (; dataset = String(dataset), gateway = "127.0.0.1", + port = mock.port, channel_size = Int(channel_size), + compression = comp, typed = typed) + extra_kwargs = reconnect_policy === nothing ? (;) : + (; reconnect_policy = reconnect_policy) + client = DatabentoAPI.Live(TEST_KEY; base_kwargs..., extra_kwargs...) DatabentoAPI.connect!(client) sub_ret = DatabentoAPI.subscribe!(client; schema = schema, symbols = symbols, stype_in = stype_in) @@ -103,8 +114,32 @@ function replay(; path::AbstractString, # Typed mode: drain the returned Channel{T}. Untyped mode: drain # the Union channel on the client. ch = typed ? sub_ret : client.channel - first_rec = take!(ch) + + # Idle watchdog (see function docstring). Started before the first take! + # so a misconfigured run delivering zero records also bails instead of + # blocking forever. + t_last = Ref(time_ns()) + idle_ns = UInt64(round(idle_timeout_s * 1e9)) + watchdog_done = Ref(false) + watchdog = @async begin + while !watchdog_done[] + sleep(0.25) + if (time_ns() - t_last[]) > idle_ns + try; close(client); catch; end + break + end + end + end + + first_rec = try + take!(ch) + catch e + watchdog_done[] = true + try; close(client); catch; end + rethrow() + end t0 = time_ns() + t_last[] = t0 n = 1 max_depth = 0 # depth_full_count tracks how often we sampled the channel ≥ 90% full — @@ -126,47 +161,58 @@ function replay(; path::AbstractString, e isa InvalidStateException && break rethrow() end + # Refresh the idle-watchdog timestamp and sample channel depth on + # the same cadence — keeps the per-record hot path identical to the + # original harness (just take! + counter) so an A/B against an + # unmodified branch is not skewed by per-record bookkeeping here. if (n & (depth_sample_every - 1)) == 0 + t_last[] = time_ns() depth = Base.n_avail(ch) depth > max_depth && (max_depth = depth) depth >= full_threshold && (depth_full_count += 1) depth_samples += 1 end if n % log_every == 0 - el = (time_ns() - t0) * 1e-9 + el = (t_last[] - t0) * 1e-9 @printf(" %10d records %.1fs %.2f M rec/s max_depth=%d\n", n, el, n / el / 1e6, max_depth) end end finally + watchdog_done[] = true try; close(client); catch; end try; wait(mock.accept_task); catch; end end - elapsed = (time_ns() - t0) * 1e-9 g1 = Base.gc_num() + # Active streaming window: first record → last record. Excludes the idle / + # reconnect-retry tail that elapses before the channel finally closes under + # reconnect_policy=:reconnect. + active_s = (t_last[] - t0) * 1e-9 + wall_s = (time_ns() - t0) * 1e-9 + elapsed = active_s > 0 ? active_s : wall_s gc_ns = g1.total_time - g0.total_time gc_frac = elapsed > 0 ? (gc_ns * 1e-9) / elapsed : 0.0 alloc_bytes = g1.allocd - g0.allocd alloc_mb = alloc_bytes / (1024 * 1024) write_dur = mock.write_done[] - mock.write_started[] write_mbps = write_dur > 0 ? mock.bytes_total / (1024*1024) / write_dur : 0.0 - # rough decompressed throughput - decompressed_mb = wire_zstd ? -1.0 : mock.bytes_total / (1024*1024) println() println("=== ", SUITE, " summary ===") @printf(" file %s (%.1f MB on wire)\n", basename(path), mock.bytes_total / (1024*1024)) + @printf(" reconnect_policy %s\n", + reconnect_policy === nothing ? "(default)" : string(reconnect_policy)) @printf(" records %d\n", n) - @printf(" consumer drain %.3f s %.2f M rec/s\n", - elapsed, n / elapsed / 1e6) + @printf(" consumer drain %.3f s active %.2f M rec/s (active window) %.3f s wall\n", + active_s, n / elapsed / 1e6, wall_s) @printf(" producer write %.3f s %.1f MB/s on wire (bottlenecked by reader if any backpressure)\n", write_dur, write_mbps) @printf(" channel cap=%d peak=%d (%.1f%%) near-full %d/%d samples (%.1f%%)\n", channel_size, max_depth, 100 * max_depth / channel_size, depth_full_count, depth_samples, depth_samples > 0 ? 100 * depth_full_count / depth_samples : 0.0) - @printf(" GC fraction=%.1f%% alloc=%.1f MB (%.0f bytes/record)\n", + @printf(" GC fraction=%.1f%% alloc=%.1f MB (%.1f bytes/record)\n", gc_frac * 100, alloc_mb, alloc_bytes / n) println() if depth_full_count > 100 && max_depth >= full_threshold @@ -177,20 +223,23 @@ function replay(; path::AbstractString, @info "interpretation: reader+decoder kept up easily — consumer was the gate, which is the bench artefact (take! in a tight loop). To stress the decoder further, increase the wire rate (run multiple replays in parallel) or use a larger / more bursty archive." end - return (; records = n, drain_s = elapsed, rate_per_s = n / elapsed, + return (; records = n, drain_s = elapsed, active_s = active_s, wall_s = wall_s, + rate_per_s = n / elapsed, max_depth = max_depth, depth_full_count = depth_full_count, gc_frac = gc_frac, alloc_bytes = alloc_bytes, + bytes_per_record = alloc_bytes / n, + reconnect_policy = reconnect_policy, write_dur = write_dur, write_mbps = write_mbps) end -function run(; path::Union{Nothing,AbstractString} = nothing) +function run(; path::Union{Nothing,AbstractString} = nothing, kwargs...) if path === nothing path = get(ENV, "DATABENTO_REPLAY_PATH", nothing) path === nothing && error("Pass path=<.dbn.zst path> or set DATABENTO_REPLAY_PATH. " * "Use benchmark/_fetch_replay_fixture.jl to grab one from Databento.") end isfile(path) || error("Replay fixture not found: $path") - return replay(path = path) + return replay(; path = path, kwargs...) end end # module BenchLiveReplay diff --git a/benchmark/fixtures.jl b/benchmark/fixtures.jl index 6e7ca07..2af935a 100644 --- a/benchmark/fixtures.jl +++ b/benchmark/fixtures.jl @@ -7,7 +7,7 @@ using TranscodingStreams const HERE = @__DIR__ const DATA_DIR = joinpath(HERE, "data") -const DBN_DATA = abspath(joinpath(HERE, "..", "..", "DBN.jl", "benchmark", "data")) +const DBN_DATA = abspath(joinpath(HERE, "..", "..", "DatabentoBinaryEncoding.jl", "benchmark", "data")) # Sized fixtures. Tier names mirror DBN.jl's existing files when available. const SIZES = ( diff --git a/docs/src/api/live.md b/docs/src/api/live.md index 4d38c88..3fa0be3 100644 --- a/docs/src/api/live.md +++ b/docs/src/api/live.md @@ -26,3 +26,15 @@ stop! channel control_channel ``` + +## Convenience + +```@docs +live_session +``` + +## Reconnect + +```@docs +add_reconnect_callback +``` diff --git a/docs/src/guide/live.md b/docs/src/guide/live.md index 333f4ea..6f968f2 100644 --- a/docs/src/guide/live.md +++ b/docs/src/guide/live.md @@ -135,22 +135,104 @@ end slow-reader logic. Bursty schemas paired with a slow disk archive benefit from larger sizes. -## Reconnect: full-jitter exponential backoff +## Reconnect: built into `Live` + +Every `Live(...)` client has a reconnect supervisor on by default. When the +TCP socket drops, the supervisor reopens it, replays your stored +subscriptions with a per-instrument-min `start=` timestamp (so subscribed +instruments resume coverage at the earliest point we lost them, bounded +to a 24h replay window), and continues delivering records into the same +channels — your `for rec in live` loop sees a continuous stream with no +exception in between. + +The retry schedule is *immediate-then-backoff*: + +- The first `immediate_reconnect_attempts` retries (default `3`) fire + with no sleep, to catch sub-second TCP blips (kernel-side RST plus a + packet retransmit or two) without waiting on the 1s+ backoff floor. +- Subsequent retries use AWS-style full-jitter exponential backoff + (`rand() * min(60, 2^(n-1))` seconds, where `n` is the backoff-phase + attempt index) up to `max_reconnect_attempts` total (default `10`; + pass `nothing` for unlimited). +- 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 budget across the connection's + lifetime. + +To opt out of the supervisor entirely: -The high-level capture functions ([`stream_to_file`](@ref) and friends) -re-establish the TCP session on drop. Sleep before each retry is -`rand() * min(60, 2^(attempt-1))` seconds (AWS-style full-jitter backoff -capped at 60s), and the loop gives up after `max_reconnect_attempts` (default -10; pass `nothing` for unlimited). +```julia +Live(dataset = "OPRA.PILLAR", reconnect_policy = :none) do live + # Reader's EOF closes the channels and terminates iteration. +end +``` + +### Observing reconnects: [`add_reconnect_callback`](@ref) + +Register a callback to be notified on each successful reconnect: + +```julia +Live(dataset = "OPRA.PILLAR") do live + add_reconnect_callback(live, (gap_start_ns, gap_end_ns) -> begin + @info "reconnect gap" gap_s = (gap_end_ns - gap_start_ns) / 1e9 + # optional: gap-fill from historical, emit a metric, etc. + end) + connect!(live) + subscribe!(live; schema = Schema.TRADES, symbols = ["SPY.OPT"], + stype_in = SType.PARENT) + start!(live) + for rec in live + # … + end +end +``` + +`gap_start_ns` is the latest per-instrument timestamp seen pre-drop +(minimum across instruments — the earliest point we lost coverage); +`gap_end_ns` is the gateway's `Metadata.start_ts` from the fresh +session. Both are `Int64` nanoseconds since the unix epoch. Convert to a +`DateTime` via `Dates.unix2datetime(ns / 1e9)` (lossy — Julia's +`DateTime` is millisecond resolution). + +Callbacks fire from the supervisor task in registration order; errors +inside a callback are logged and swallowed (they do not perturb the +list or fail the reconnect). + +### Gateway errors are terminal + +If the gateway sends an `ErrorMsg` (auth failed, malformed subscription, +etc.), the supervisor *does not* reconnect — retrying would just re-hit +the same condition. The typed reader's control branch sets the +client's `terminal_error`, channels close, and your iterator exits +cleanly. + +### Convenience: `live_session` + +If you don't need to interleave logic between `connect!`, `subscribe!`, +and `start!`, [`live_session`](@ref) bundles the whole lifecycle into a +single do-block: + +```julia +live_session(; + dataset = "OPRA.PILLAR", + subscriptions = [ + (; schema = Schema.TRADES, symbols = ["SPY.OPT"], stype_in = SType.PARENT), + (; schema = Schema.CBBO_1S, symbols = ["SPY.OPT"], stype_in = SType.PARENT), + ], +) do live + for rec in live + # … + end +end +``` -If you're driving the Live client yourself outside [`stream_to_file`](@ref), -you handle reconnects yourself: catch the error from the iterator, construct -a fresh `Live`, re-`subscribe!`, and resume. +`reconnect_policy` defaults to `:reconnect` (same as `Live(...)`). ## Closing the client -`close(client)` is idempotent. The Live do-block calls it for you on exit; if -you're managing the lifecycle manually, wrap subscribe/start/iterate in a -`try/finally close(client) end`. Channels close cleanly, the gateway gets a -final `stop` frame, and consumer tasks exit on `InvalidStateException` from -their pending `take!` calls. +`close(client)` is idempotent. The Live do-block calls it for you on exit; +if you're managing the lifecycle manually, wrap subscribe/start/iterate in +a `try/finally close(client) end`. Channels close cleanly, the gateway +gets a final `stop` frame, the supervisor (if running) observes the +state transition and exits at its next iteration, and consumer tasks +exit on `InvalidStateException` from their pending `take!` calls. diff --git a/docs/src/quickstart.md b/docs/src/quickstart.md index a4a2422..a16a8a8 100644 --- a/docs/src/quickstart.md +++ b/docs/src/quickstart.md @@ -48,8 +48,11 @@ end # and the function returns normally. ``` -For long-running captures that write to disk and survive disconnects, use -[`stream_to_file`](@ref) — see [Capture to File](@ref). +The Live client auto-reconnects on TCP drops by default (immediate-retry +phase then exponential backoff). For long-running captures that *also* +write to disk, use [`stream_to_file`](@ref) — see [Capture to File](@ref). +For one-call setup that bundles `connect! → subscribe! → start!`, use +[`live_session`](@ref) — see [Live Streaming](@ref). ## What next? diff --git a/src/DatabentoAPI.jl b/src/DatabentoAPI.jl index a2e57cc..c9fa70b 100644 --- a/src/DatabentoAPI.jl +++ b/src/DatabentoAPI.jl @@ -74,6 +74,7 @@ include("live/reader.jl") include("live/subscribe.jl") include("live/heartbeat.jl") include("live/streaming.jl") +include("live/reconnect.jl") # Re-exported DatabentoBinaryEncoding enums (single import for users) export Schema, SType, Compression, Encoding, Action, Side, InstrumentClass @@ -106,9 +107,13 @@ export resolve # Live lifecycle export connect!, subscribe!, subscribe_callback, start!, stop! export channel, control_channel +export live_session # Live capture to file export stream_to_file, stream_multi_to_files, default_live_path, read_capture export open_dbn_writer, write_record! +# Live reconnect callback +export add_reconnect_callback + end # module DatabentoAPI diff --git a/src/live/client.jl b/src/live/client.jl index 468c7c9..4e659d8 100644 --- a/src/live/client.jl +++ b/src/live/client.jl @@ -67,6 +67,44 @@ mutable struct Live lsg_version::Union{Nothing,String} subscriptions::Vector{NamedTuple} next_sub_id::Int + + # --- reconnect plumbing (wired up in subsequent commits) --- + + # Replay bookkeeping. Reader populates these per record; on reconnect we + # pick the per-instrument minimum as the resubscribe start timestamp so + # the new connection bridges the gap for instruments seen pre-drop. + # BBO-family schemas key on ts_recv; others on ts_event. + last_ts_event_by_id::Dict{UInt32,Int64} + last_ts_recv_by_id::Dict{UInt32,Int64} + # Gateway Metadata.start from the most recent (re-)connection. Used as + # gap_end in the reconnect callback. + last_metadata_start_ns::Union{Nothing,Int64} + + # Reconnect policy + retry budget. + reconnect_policy::ReconnectPolicy.T + max_reconnect_attempts::Union{Int,Nothing} + immediate_reconnect_attempts::Int + # Lifetime budget remaining; reset to max_reconnect_attempts whenever the + # new reader successfully delivers ≥1 data record (so a long-lived + # session that streams between drops keeps its full budget). + attempts_remaining::Int + records_since_reconnect::Int + + # Reconnect-callback registry. Each callback receives + # `(gap_start_ns::Int64, gap_end_ns::Int64)`. + reconnect_callbacks::Vector{Any} + callbacks_lock::ReentrantLock + + # Supervisor task + state machine. `state` advances through + # :fresh → :connecting → :connected → :streaming → :reconnecting → ... + # → :closed | :failed. `state_lock` guards transitions so close() racing + # the supervisor produces a deterministic shutdown. + reconnect_supervisor::Union{Nothing,Task} + state::Symbol + state_lock::ReentrantLock + # Set by the reader when a gateway ErrorMsg arrives; supervisor checks + # this and refuses to reconnect (transitions to :failed). + terminal_error::Union{Nothing,String} end function Live(key::Union{Nothing,AbstractString} = nothing; @@ -79,7 +117,10 @@ function Live(key::Union{Nothing,AbstractString} = nothing; slow_reader_behavior::Union{Nothing,SlowReaderBehavior.T,AbstractString} = nothing, channel_size::Integer = 10_000, user_agent::AbstractString = USER_AGENT, - typed::Bool = false) + typed::Bool = false, + reconnect_policy::Union{ReconnectPolicy.T,Symbol,AbstractString} = ReconnectPolicy.RECONNECT, + max_reconnect_attempts::Union{Integer,Nothing} = 10, + immediate_reconnect_attempts::Integer = 3) api_key = load_api_key(key) gw = gateway === nothing ? gateway_for_dataset(dataset) : String(gateway) srb = if slow_reader_behavior isa AbstractString @@ -89,6 +130,14 @@ function Live(key::Union{Nothing,AbstractString} = nothing; end cmp = compression isa AbstractString ? getfield(Compression, Symbol(uppercase(compression))) : compression + rp = _coerce_reconnect_policy(reconnect_policy) + immediate_reconnect_attempts >= 0 || throw(ArgumentError( + "immediate_reconnect_attempts must be ≥ 0, got $immediate_reconnect_attempts")) + if max_reconnect_attempts !== nothing && max_reconnect_attempts < 0 + throw(ArgumentError( + "max_reconnect_attempts must be ≥ 0 or `nothing` (unlimited), got $max_reconnect_attempts")) + end + init_budget = max_reconnect_attempts === nothing ? typemax(Int) : Int(max_reconnect_attempts) # Channels are built per-mode. Typed mode lazily creates per-schema data # channels in subscribe!; the control channel is constructed up front so @@ -109,9 +158,35 @@ function Live(key::Union{Nothing,AbstractString} = nothing; false, false, false, nothing, nothing, NamedTuple[], 1, + # --- reconnect fields --- + Dict{UInt32,Int64}(), Dict{UInt32,Int64}(), + nothing, # last_metadata_start_ns + rp, + max_reconnect_attempts === nothing ? nothing : Int(max_reconnect_attempts), + Int(immediate_reconnect_attempts), + init_budget, + 0, # records_since_reconnect + Any[], ReentrantLock(), + nothing, # reconnect_supervisor + :fresh, ReentrantLock(), + nothing, # terminal_error ) end +# Internal: accept ReconnectPolicy.T directly, or Symbol / String spellings +# (`:none`/`"none"`, `:reconnect`/`"reconnect"`) for ergonomic call sites that +# don't want to import the enum. +function _coerce_reconnect_policy(rp) + rp isa ReconnectPolicy.T && return rp + s = rp isa Symbol ? rp : Symbol(uppercase(String(rp))) + # Allow both lowercase symbols and the enum-style uppercase strings. + sym = Symbol(uppercase(String(s))) + sym === :NONE && return ReconnectPolicy.NONE + sym === :RECONNECT && return ReconnectPolicy.RECONNECT + throw(ArgumentError( + "reconnect_policy must be ReconnectPolicy.NONE or ReconnectPolicy.RECONNECT (or :none / :reconnect), got $(repr(rp))")) +end + function Base.show(io::IO, c::Live) mode = c.typed ? "typed" : "untyped" print(io, "Live(dataset=", c.dataset, ", gateway=", c.gateway, ":", c.port, @@ -151,6 +226,66 @@ function Live(f::Function, args...; kwargs...) end end +""" + live_session(fn; dataset, subscriptions, reconnect_policy=:reconnect, kwargs...) -> result of fn + +Convenience wrapper that bundles `connect! → subscribe!(many) → start! → fn(client) → close` +into a single do-block. Defaults `reconnect_policy = :reconnect` so iteration +inside `fn` survives transient TCP drops by spawning the Live-layer reconnect +supervisor (see [`add_reconnect_callback`](@ref)). + +`subscriptions` is a vector of NamedTuples describing one subscribe call each. +Each entry must have `schema` and `symbols`; `stype_in` (default +`SType.RAW_SYMBOL`), `snapshot` (default `false`), and `start` (default +`nothing`) are optional. + +Any extra `kwargs` are forwarded to `Live(...)` — typically `key`, `gateway`, +`port`, `compression`, `typed`, `max_reconnect_attempts`, +`immediate_reconnect_attempts`. + +```julia +live_session(; dataset = "GLBX.MDP3", + subscriptions = [(; schema = Schema.TRADES, + symbols = ["ES.FUT"], + stype_in = SType.PARENT)]) do client + add_reconnect_callback(client, (g0, g1) -> @info "gap" gap_s=(g1-g0)/1e9) + for rec in client + handle(rec) + end +end +``` + +The lower-level `Live(...) do client; ...; end` do-block (introduced in 0.1.1) +remains available for callers that need the explicit `connect!/subscribe!/start!` +lifecycle — e.g. when conditional subscription requires inspecting `client.session_id` +between the steps. +""" +function live_session(fn::Function; + dataset::AbstractString, + subscriptions, + reconnect_policy::Union{ReconnectPolicy.T,Symbol,AbstractString} = :reconnect, + key::Union{Nothing,AbstractString} = nothing, + kwargs...) + isempty(subscriptions) && throw(ArgumentError( + "live_session requires at least one subscription")) + Live(key; dataset = dataset, reconnect_policy = reconnect_policy, kwargs...) do client + connect!(client) + for sub in subscriptions + haskey(sub, :schema) || throw(ArgumentError("subscription missing :schema")) + haskey(sub, :symbols) || throw(ArgumentError("subscription missing :symbols")) + subscribe!(client; + schema = sub.schema, + symbols = sub.symbols, + stype_in = get(sub, :stype_in, SType.RAW_SYMBOL), + snapshot = get(sub, :snapshot, false), + start = get(sub, :start, nothing), + ) + end + start!(client) + return fn(client) + end +end + """ control_channel(client::Live) -> Channel{DBN.DBNRecord} @@ -167,16 +302,12 @@ function control_channel(c::Live) return c.control_channel end -""" - connect!(client) - -Open the TCP connection, perform the CRAM authentication handshake. After this returns, -the client is ready for [`subscribe!`](@ref) calls. Throws `BentoAuthError` on failure. -""" -function connect!(c::Live) - c.connected && return c - c.closed && throw(ArgumentError("client is closed")) - +# Internal: open a TCP socket to the configured gateway and run the CRAM +# authentication handshake, populating c.socket / c.lsg_version / c.session_id. +# Does NOT touch lifecycle flags (c.connected, c.closed) so it can be reused +# from both the public `connect!` entry and the upcoming reconnect path, +# which manages those flags via its own state machine. +function _open_socket_and_auth!(c::Live) c.socket = Sockets.connect(c.gateway, c.port) greeting = read_text_frame(c.socket) @@ -208,6 +339,19 @@ function connect!(c::Live) throw(BentoAuthError("Live authentication failed: $err")) end c.session_id = get(auth_resp, "session_id", nothing) + return nothing +end + +""" + connect!(client) + +Open the TCP connection, perform the CRAM authentication handshake. After this returns, +the client is ready for [`subscribe!`](@ref) calls. Throws `BentoAuthError` on failure. +""" +function connect!(c::Live) + c.connected && return c + c.closed && throw(ArgumentError("client is closed")) + _open_socket_and_auth!(c) c.connected = true return c end @@ -225,6 +369,12 @@ function Base.close(c::Live) # Ctrl-C twice, or close() is called from a finally inside f()) is a # no-op and doesn't double-close anything. c.closed = true + # Mark terminal state so the supervisor (if running) drops out at its + # next state check instead of attempting another reconnect against a + # socket we're about to tear down. + lock(c.state_lock) do + c.state = :closed + end try if c.connected && c.socket !== nothing && isopen(c.socket) try diff --git a/src/live/reader.jl b/src/live/reader.jl index 2bd99f9..1116f27 100644 --- a/src/live/reader.jl +++ b/src/live/reader.jl @@ -85,6 +85,11 @@ function _reader_loop_typed(c::Live) buffered = DBN.BufferedReader(counting) decoder = DBN.DBNDecoder(buffered) DBN.read_header!(decoder) + # Capture gateway Metadata.start_ts so the reconnect supervisor has + # a gap_end timestamp to report to user callbacks. + if decoder.metadata !== nothing + c.last_metadata_start_ns = Int64(decoder.metadata.start_ts) + end while !c.closed hd_result = try @@ -112,72 +117,87 @@ function _reader_loop_typed(c::Live) # Hot path: data-record rtypes, type-stable put! to the # per-schema channel. Nullness check FIRST in each branch so # unsubscribed schemas short-circuit on a pointer comparison - # before the slower rtype enum comparison. + # before the slower rtype enum comparison. `_put_data!` updates + # per-instrument replay timestamps on the Live client before + # publishing, so the reconnect supervisor sees them. if ch_trades !== nothing && rt == DBN.RType.MBP_0_MSG - put!(ch_trades, DBN.read_trade_msg(decoder, hd)) + _put_data!(c, ch_trades, DBN.read_trade_msg(decoder, hd)) continue elseif ch_mbo !== nothing && rt == DBN.RType.MBO_MSG - put!(ch_mbo, DBN.read_mbo_msg(decoder, hd)) + _put_data!(c, ch_mbo, DBN.read_mbo_msg(decoder, hd)) continue elseif (ch_mbp1 !== nothing || ch_tbbo !== nothing) && rt == DBN.RType.MBP_1_MSG # MBP_1 and TBBO both use MBP_1_MSG rtype + MBP1Msg layout. # Broadcast so each subscribed channel receives the record. + # Single replay-state update — same record content. rec = DBN.read_mbp1_msg(decoder, hd) + _record_replay!(c, rec) ch_mbp1 !== nothing && put!(ch_mbp1, rec) ch_tbbo !== nothing && put!(ch_tbbo, rec) continue elseif ch_mbp10 !== nothing && rt == DBN.RType.MBP_10_MSG - put!(ch_mbp10, DBN.read_mbp10_msg(decoder, hd)) + _put_data!(c, ch_mbp10, DBN.read_mbp10_msg(decoder, hd)) continue elseif ch_ohlcv_1s !== nothing && rt == DBN.RType.OHLCV_1S_MSG - put!(ch_ohlcv_1s, DBN.read_ohlcv_msg(decoder, hd)) + _put_data!(c, ch_ohlcv_1s, DBN.read_ohlcv_msg(decoder, hd)) continue elseif ch_ohlcv_1m !== nothing && rt == DBN.RType.OHLCV_1M_MSG - put!(ch_ohlcv_1m, DBN.read_ohlcv_msg(decoder, hd)) + _put_data!(c, ch_ohlcv_1m, DBN.read_ohlcv_msg(decoder, hd)) continue elseif ch_ohlcv_1h !== nothing && rt == DBN.RType.OHLCV_1H_MSG - put!(ch_ohlcv_1h, DBN.read_ohlcv_msg(decoder, hd)) + _put_data!(c, ch_ohlcv_1h, DBN.read_ohlcv_msg(decoder, hd)) continue elseif ch_ohlcv_1d !== nothing && rt == DBN.RType.OHLCV_1D_MSG - put!(ch_ohlcv_1d, DBN.read_ohlcv_msg(decoder, hd)) + _put_data!(c, ch_ohlcv_1d, DBN.read_ohlcv_msg(decoder, hd)) continue elseif ch_status !== nothing && rt == DBN.RType.STATUS_MSG - put!(ch_status, DBN.read_status_msg(decoder, hd)) + _put_data!(c, ch_status, DBN.read_status_msg(decoder, hd)) continue elseif ch_def !== nothing && rt == DBN.RType.INSTRUMENT_DEF_MSG - put!(ch_def, DBN.read_instrument_def_msg(decoder, hd)) + _put_data!(c, ch_def, DBN.read_instrument_def_msg(decoder, hd)) continue elseif ch_imbal !== nothing && rt == DBN.RType.IMBALANCE_MSG - put!(ch_imbal, DBN.read_imbalance_msg(decoder, hd)) + _put_data!(c, ch_imbal, DBN.read_imbalance_msg(decoder, hd)) continue elseif ch_stat !== nothing && rt == DBN.RType.STAT_MSG - put!(ch_stat, DBN.read_stat_msg(decoder, hd)) + _put_data!(c, ch_stat, DBN.read_stat_msg(decoder, hd)) continue elseif ch_cmbp1 !== nothing && rt == DBN.RType.CMBP_1_MSG - put!(ch_cmbp1, DBN.read_cmbp1_msg(decoder, hd)) + _put_data!(c, ch_cmbp1, DBN.read_cmbp1_msg(decoder, hd)) continue elseif ch_cbbo1s !== nothing && rt == DBN.RType.CBBO_1S_MSG - put!(ch_cbbo1s, DBN.read_cbbo1s_msg(decoder, hd)) + _put_data!(c, ch_cbbo1s, DBN.read_cbbo1s_msg(decoder, hd)) continue elseif ch_cbbo1m !== nothing && rt == DBN.RType.CBBO_1M_MSG - put!(ch_cbbo1m, DBN.read_cbbo1m_msg(decoder, hd)) + _put_data!(c, ch_cbbo1m, DBN.read_cbbo1m_msg(decoder, hd)) continue elseif ch_tcbbo !== nothing && rt == DBN.RType.TCBBO_MSG - put!(ch_tcbbo, DBN.read_tcbbo_msg(decoder, hd)) + _put_data!(c, ch_tcbbo, DBN.read_tcbbo_msg(decoder, hd)) continue elseif ch_bbo1s !== nothing && rt == DBN.RType.BBO_1S_MSG - put!(ch_bbo1s, DBN.read_bbo1s_msg(decoder, hd)) + _put_data!(c, ch_bbo1s, DBN.read_bbo1s_msg(decoder, hd)) continue elseif ch_bbo1m !== nothing && rt == DBN.RType.BBO_1M_MSG - put!(ch_bbo1m, DBN.read_bbo1m_msg(decoder, hd)) + _put_data!(c, ch_bbo1m, DBN.read_bbo1m_msg(decoder, hd)) continue end - # Control rtypes → control channel via generic dispatch. + # Control rtypes → control channel via generic dispatch. An + # ErrorMsg sets c.terminal_error before publishing so the + # supervisor sees it once the reader exits and refuses to + # reconnect — gateway-side errors are deterministic and + # retrying would just re-hit the same condition. if rt == DBN.RType.ERROR_MSG || rt == DBN.RType.SYSTEM_MSG || rt == DBN.RType.SYMBOL_MAPPING_MSG rec = DBN.read_record_dispatch(decoder, hd, rt) + if rec isa DBN.ErrorMsg + try + c.terminal_error = String(rec.err) + catch + c.terminal_error = "gateway ErrorMsg" + end + end if rec !== nothing && ctrl_chan !== nothing && isopen(ctrl_chan) put!(ctrl_chan, rec) end @@ -200,20 +220,25 @@ function _reader_loop_typed(c::Live) end end finally - # Close every channel we own so any consumer task break out of its - # take! loop. The Live.close() path will also try this, but doing it - # here ensures cleanup on reader-side crash. - for ch in values(c.typed_data_channels) + # Channel teardown is conditional on the reconnect policy: under NONE + # there will be no replacement reader, so we close channels to signal + # consumers; under RECONNECT the supervisor will spawn a fresh reader + # against the same channels and closing here would terminate + # iteration mid-stream. The user-initiated close path (c.closed=true) + # always closes regardless, so explicit close(client) still works. + if c.reconnect_policy == ReconnectPolicy.NONE || c.closed + for ch in values(c.typed_data_channels) + try + isopen(ch) && close(ch) + catch + end + end try - isopen(ch) && close(ch) + c.control_channel === nothing || + (isopen(c.control_channel) && close(c.control_channel)) catch end end - try - c.control_channel === nothing || - (isopen(c.control_channel) && close(c.control_channel)) - catch - end end return nothing end @@ -227,6 +252,39 @@ end return ch === nothing ? nothing : ch::Channel{T} end +# Update per-instrument replay timestamps on the Live client. The supervisor +# (added in a later commit) reads these to compute the resubscribe `start=` +# value after a reconnect. `hasproperty` checks fold against the concrete +# record type at each typed-reader callsite — zero overhead for records that +# lack ts_recv (e.g. TradeMsg has only ts_event on .hd). +@inline function _record_replay!(c::Live, rec) + if hasproperty(rec, :hd) && hasproperty(rec.hd, :instrument_id) + iid = rec.hd.instrument_id + if hasproperty(rec.hd, :ts_event) + c.last_ts_event_by_id[iid] = rec.hd.ts_event + end + if hasproperty(rec, :ts_recv) + c.last_ts_recv_by_id[iid] = rec.ts_recv + end + end + # First record of this reader incarnation refreshes the retry budget — + # a session that successfully streams real data between drops keeps its + # full budget across the connection's lifetime. + if c.records_since_reconnect == 0 + c.attempts_remaining = c.max_reconnect_attempts === nothing ? + typemax(Int) : c.max_reconnect_attempts + end + c.records_since_reconnect += 1 + return nothing +end + +# Typed-mode helper: update replay state and put the record on its channel. +# Inlined so the put! stays monomorphic on the concrete channel eltype. +@inline function _put_data!(c::Live, ch::Channel, rec) + _record_replay!(c, rec) + put!(ch, rec) +end + function _reader_loop(c::Live) try # Wrap with zstd decompressor first if the session was negotiated with @@ -239,6 +297,11 @@ function _reader_loop(c::Live) buffered = DBN.BufferedReader(counting) decoder = DBN.DBNDecoder(buffered) DBN.read_header!(decoder) + # Capture gateway Metadata.start_ts so the reconnect supervisor has + # a gap_end timestamp to report to user callbacks. + if decoder.metadata !== nothing + c.last_metadata_start_ns = Int64(decoder.metadata.start_ts) + end while !c.closed rec = try DBN.read_record(decoder) @@ -263,11 +326,28 @@ function _reader_loop(c::Live) # reliable enough on Windows for half-closed sockets to differentiate # safely without busy-looping. rec === nothing && break + # 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) + # ErrorMsg sets c.terminal_error *before* the put! so the + # supervisor sees it once the reader exits and refuses to + # reconnect — gateway-side errors are deterministic and + # retrying just re-hits the same condition (and burns through + # max_reconnect_attempts for nothing). Mirrors the typed + # reader's control branch. The record still flows through the + # channel so consumers can inspect it. Users who want to + # treat ErrorMsg as informational (e.g. SkippedRecordsAfter- + # SlowReading) should use `reconnect_policy = :none` and + # handle reconnection themselves. + if rec isa DBN.ErrorMsg + try + c.terminal_error = String(rec.err) + catch + c.terminal_error = "gateway ErrorMsg" + end + end put!(c.channel, rec) - # ErrorMsg is informational on this stream (e.g. SkippedRecords- - # AfterSlowReading is a code-7 ErrorMsg signalling SKIP behavior, not a - # fatal error). Forward it through the channel and let the consumer - # decide; only EOF on the socket terminates the reader loop. end catch e # Suppress the "crashed" log on mid-stream EOF (same rationale as @@ -279,9 +359,14 @@ function _reader_loop(c::Live) end end finally - try - isopen(c.channel) && close(c.channel) - catch + # See _reader_loop_typed for the policy rationale: under RECONNECT + # the channel outlives the reader so the supervisor can re-bind it + # to a fresh reader. + if c.reconnect_policy == ReconnectPolicy.NONE || c.closed + try + isopen(c.channel) && close(c.channel) + catch + end end end return nothing diff --git a/src/live/reconnect.jl b/src/live/reconnect.jl new file mode 100644 index 0000000..19c1a53 --- /dev/null +++ b/src/live/reconnect.jl @@ -0,0 +1,337 @@ +# Reconnect supervisor: keeps a Live session alive across TCP drops by +# spawning fresh reader tasks against the same user-facing channels and +# re-issuing stored subscriptions on each reconnect. +# +# Lives in the Live module so any consumer (iteration, subscribe_callback, +# stream_to_file) benefits — reconnect is no longer streaming-layer-only. +# +# Schedule: c.immediate_reconnect_attempts immediate (no-sleep) retries to +# catch sub-second TCP blips, then full-jitter exponential backoff +# (_RECONNECT_BASE_S → _RECONNECT_CAP_S) for up to c.max_reconnect_attempts +# total attempts before giving up. The retry budget refreshes whenever the +# new reader successfully delivers ≥1 data record (see _record_replay! in +# reader.jl), so a long-lived session that streams between drops keeps its +# full budget across the connection's lifetime. + +""" + add_reconnect_callback(client::Live, cb) + +Register `cb(gap_start_ns::Int64, gap_end_ns::Int64)` to fire after each +successful reconnect. Both timestamps are nanoseconds since the unix epoch. + +`gap_start_ns` is the latest pre-drop per-instrument timestamp the client +had observed (min across instruments, using `ts_recv` for BBO-family +schemas and `ts_event` otherwise); `0` if no records were seen pre-drop. + +`gap_end_ns` is the gateway's `Metadata.start_ts` from the freshly +reconnected session; `0` if unavailable. + +Errors raised inside `cb` are logged via `@error` and swallowed — they do +not perturb the callback list or fail the reconnect. + +Callbacks are invoked sequentially on the supervisor task in registration +order. Convert to `Dates.DateTime` via `Dates.unix2datetime(ns / 1e9)` if +needed (lossy — pandas-style nanosecond precision is preserved only in the +raw `Int64`). +""" +function add_reconnect_callback(c::Live, cb) + cb isa Function || cb isa Base.Callable || throw(ArgumentError( + "reconnect callback must be callable, got $(typeof(cb))")) + lock(c.callbacks_lock) do + push!(c.reconnect_callbacks, cb) + end + return nothing +end + +# Snapshot callbacks under the lock, then fire them outside it so a slow +# callback can't block add_reconnect_callback. +function _fire_reconnect_callbacks(c::Live, gap_start_ns::Int64, gap_end_ns::Int64) + cbs = lock(c.callbacks_lock) do + copy(c.reconnect_callbacks) + end + for cb in cbs + try + cb(gap_start_ns, gap_end_ns) + catch e + @error "reconnect callback raised" dataset=c.dataset exception=(e, catch_backtrace()) + end + end + return nothing +end + +# Best-effort close of every user-facing channel. Used on terminal-state +# transitions (:failed / :closed) to signal iterators / subscribe_callback +# consumers that no more records are coming, since under RECONNECT the +# reader's finally block leaves channels open for the supervisor. +function _close_channels!(c::Live) + try + c.channel === nothing || (isopen(c.channel) && close(c.channel)) + catch + end + for ch in values(c.typed_data_channels) + try + isopen(ch) && close(ch) + catch + end + end + try + c.control_channel === nothing || + (isopen(c.control_channel) && close(c.control_channel)) + catch + end + return nothing +end + +function _channels_open(c::Live) + if c.channel !== nothing && !isopen(c.channel) + return false + end + for ch in values(c.typed_data_channels) + isopen(ch) || return false + end + if c.control_channel !== nothing && !isopen(c.control_channel) + return false + end + return true +end + +# Pick the gap_start timestamp to report to callbacks. Use the smallest +# per-instrument timestamp seen across any subscribed schema — this is the +# earliest point we lost coverage. Falls back to 0 if no records were seen. +function _gap_start_ns(c::Live) + best = typemax(Int64) + seen = false + if !isempty(c.last_ts_event_by_id) + v = minimum(values(c.last_ts_event_by_id)) + best = min(best, v); seen = true + end + if !isempty(c.last_ts_recv_by_id) + v = minimum(values(c.last_ts_recv_by_id)) + best = min(best, v); seen = true + end + return seen ? best : Int64(0) +end + +# Compute how many attempts the supervisor has burned in the current burst +# (since the last successful first-record). Used as the 1-based `attempt` +# arg to _reconnect_delay so the cadence (immediate phase → backoff) +# tracks each disconnect's retries independently. +function _attempts_used_this_burst(c::Live) + init = c.max_reconnect_attempts === nothing ? typemax(Int) : c.max_reconnect_attempts + used = init - c.attempts_remaining + return used < 0 ? 0 : used # paranoia against any decrement-past-zero +end + +# Polling sleep that wakes promptly on user-initiated close, so a long +# backoff window doesn't pin shutdown responsiveness. +function _interruptible_sleep(c::Live, delay::Float64) + delay > 0.0 || return + deadline = time() + delay + while time() < deadline + c.closed && return + c.state == :closed && return + sleep(min(_CONNECTION_POLL_S, deadline - time())) + end + return +end + +# Re-issue a stored subscription on the freshly-reconnected socket. Doesn't +# touch c.subscriptions (already there) or c.next_sub_id (already advanced). +# `start_override` lets the supervisor inject the per-instrument-min replay +# timestamp instead of the original user-supplied start. +function _send_subscribe_frame(c::Live, sub; start_override::Union{Nothing,Integer} = nothing) + write_text_frame(c.socket; + schema = schema_str(sub.schema), + stype_in = stype_str(sub.stype_in), + symbols = sub.symbols, + snapshot = sub.snapshot ? 1 : 0, + start = start_override === nothing ? sub.start : start_override, + id = sub.id, + is_last = 1, + ) + return nothing +end + +# Attempt a single reconnect: close stale socket, fresh socket + auth, +# replay subscriptions with computed start_ts, send start_session, spawn a +# new reader. Returns true on success (state advances to :streaming), +# false on any failure (state stays :reconnecting; supervisor will retry). +function _reconnect_once!(c::Live) + # Best-effort teardown of the dead session's socket. + try + c.socket === nothing || Sockets.close(c.socket) + catch + end + c.socket = nothing + c.connected = false + + # User closed us mid-reconnect — bail. + c.closed && return false + + # Fresh socket + CRAM handshake. _open_socket_and_auth! throws on + # network or auth failure; supervisor's caller will catch. + _open_socket_and_auth!(c) + c.connected = true + + # Replay subscriptions. Don't use public subscribe! — it errors on + # c.started=true (a sanity check for the *initial* lifecycle) and + # would push duplicates onto c.subscriptions. + saved = copy(c.subscriptions) + for sub in saved + start_override = _bound_replay(_replay_start_ts(c, sub.schema)) + _send_subscribe_frame(c, sub; start_override = start_override) + end + + write_text_frame(c.socket; start_session = 0) + + # Bail if a racing close() closed our channels while we were re-auth'ing. + if !_channels_open(c) + return false + end + + # Reset per-reader-incarnation counter before spawning so the first + # record from the new reader triggers the budget refresh in + # _record_replay!. Also clear last_metadata_start_ns so we can detect + # when the new reader has populated it (post-spawn poll below). + c.records_since_reconnect = 0 + prev_md_ns = c.last_metadata_start_ns + c.last_metadata_start_ns = nothing + + # Spawn the new reader against the *same* channels. Under RECONNECT + # policy start! did not bind them to the original reader_task, so they + # remained open after the prior reader exited. + c.reader_task = if c.typed + @async _reader_loop_typed(c) + else + @async _reader_loop(c) + end + + # Wait for the new reader to read its DBN header and populate + # last_metadata_start_ns, so the reconnect callback can fire with the + # correct gap_end. The first reconnect on a cold-JIT process can take + # several seconds for the reader's read_header! path to compile; once + # warm, this is sub-ms. Bounded — if the new reader stalls before the + # header arrives, we fall back to the prior value (callback gap_end + # ≈ pre-drop reference, the closest signal we have). + md_deadline = time() + 10.0 + while time() < md_deadline && c.last_metadata_start_ns === nothing + c.closed && break + sleep(0.02) + end + if c.last_metadata_start_ns === nothing + c.last_metadata_start_ns = prev_md_ns + end + + lock(c.state_lock) do + if c.state == :reconnecting + c.state = :streaming + end + end + return true +end + +# Supervisor task. Created from start! when reconnect_policy != NONE. +# Lifecycle: +# :streaming → (reader dies) → :reconnecting → :streaming (or :failed) +# Terminates when c.state hits :closed or :failed. +function _supervisor_loop(c::Live) + while true + # Block until the current reader exits. Reader exceptions are + # already logged inside _reader_loop[_typed]; we swallow whatever + # wait raises so the supervisor can keep running. + rdr = c.reader_task + rdr === nothing && return + try + wait(rdr) + catch + end + + # Decide: reconnect, or terminate? + decision = lock(c.state_lock) do + if c.state == :closed || c.state == :failed + return :exit + end + if c.terminal_error !== nothing + @error "live reconnect: gateway sent terminal ErrorMsg" dataset=c.dataset err=c.terminal_error + c.state = :failed + c.started = false + c.connected = false + return :terminate + end + if c.reconnect_policy == ReconnectPolicy.NONE + # Defensive — we shouldn't have been spawned under NONE. + c.state = :closed + return :terminate + end + if c.attempts_remaining <= 0 + @error "live reconnect: budget exhausted; giving up" dataset=c.dataset + c.state = :failed + c.started = false + c.connected = false + return :terminate + end + c.state = :reconnecting + c.connected = false + return :reconnect + end + + if decision == :terminate + _close_channels!(c) + return + elseif decision == :exit + return + end + + # Sleep per the schedule. attempt index = used_this_burst + 1. + attempt = _attempts_used_this_burst(c) + 1 + delay = _reconnect_delay(attempt; + immediate = c.immediate_reconnect_attempts, + base = _RECONNECT_BASE_S, + cap = _RECONNECT_CAP_S) + _interruptible_sleep(c, delay) + if c.state == :closed || c.closed + _close_channels!(c) + return + end + + # Snapshot pre-drop replay state for the callback BEFORE the + # reconnect attempt (after, c.last_ts_*_by_id may be advanced by + # the new reader before the callback fires). + gap_start = _gap_start_ns(c) + + ok = try + _reconnect_once!(c) + catch e + if e isa BentoAuthError + # Auth failures are deterministic and won't fix themselves — + # don't burn budget retrying. + @error "live reconnect: auth failed; giving up" dataset=c.dataset exception=(e, catch_backtrace()) + lock(c.state_lock) do + c.state = :failed + c.started = false + c.connected = false + end + _close_channels!(c) + return + end + @warn "live reconnect attempt failed" dataset=c.dataset attempt=attempt exception=(e, catch_backtrace()) + false + end + + c.attempts_remaining -= 1 + + if ok + @info "live reconnect: reconnected" dataset=c.dataset attempt=attempt session_id=c.session_id + gap_end = c.last_metadata_start_ns === nothing ? Int64(0) : c.last_metadata_start_ns + _fire_reconnect_callbacks(c, gap_start, gap_end) + # Loop back to wait on the new reader. The budget will refresh + # in _record_replay! when the first record arrives. + else + # _reconnect_once! returned false (couldn't open channels / + # user closed) — re-check on next iteration. State stays + # :reconnecting so the next wait() returns immediately + # (reader_task is the just-spawned-but-doomed one, or unchanged + # if we never got to spawn). + end + end +end diff --git a/src/live/streaming.jl b/src/live/streaming.jl index 0013af7..d82f987 100644 --- a/src/live/streaming.jl +++ b/src/live/streaming.jl @@ -38,32 +38,26 @@ const _CONNECTION_POLL_S = 0.5 const _REPLAY_WINDOW_NS = Int64(24 * 3600) * 1_000_000_000 # Full-jitter exponential backoff (AWS-recommended): wait drawn uniformly from -# [0, min(cap, base * 2^(attempt-1))). `attempt` is 1-based — `attempt=1` is -# the delay before the first reconnect. +# [0, min(cap, base * 2^(n-1))) where n is the backoff-phase attempt index. +# `attempt` is 1-based — `attempt=1` is the delay before the first reconnect. +# +# Optional `immediate` phase: when `attempt <= immediate`, return 0.0 so the +# first N reconnects fire with no sleep. Catches sub-second TCP blips that +# recover within a few packet retransmits, which the 1s+ backoff would +# otherwise stretch into multi-second downtime. The backoff phase starts at +# `attempt = immediate + 1` and is indexed from there (so the first backoff +# attempt still gets the small `base`-sized window, not a pre-stretched one). function _reconnect_delay(attempt::Integer; + immediate::Integer = 0, base::Real = _RECONNECT_BASE_S, cap::Real = _RECONNECT_CAP_S) - attempt < 1 && return 0.0 - upper = min(float(cap), float(base) * 2.0^(attempt - 1)) + attempt < 1 && return 0.0 + attempt <= immediate && return 0.0 + n = attempt - immediate + upper = min(float(cap), float(base) * 2.0^(n - 1)) return rand() * upper end -# Sleep up to `delay` seconds, but wake every `_CONNECTION_POLL_S` to check for -# shutdown / deadline so a long backoff window doesn't pin `duration_s` or -# `shutdown_requested[]` responsiveness. Matches the cadence of the main -# coordination loop. -function _wait_for_reconnect(delay::Float64, ctxs, deadline::Union{Nothing,Float64}) - delay > 0.0 || return - deadline_at = time() + delay - while true - _any_shutdown(ctxs) && return - deadline !== nothing && time() >= deadline && return - remaining = deadline_at - time() - remaining > 0 || return - sleep(min(_CONNECTION_POLL_S, remaining)) - end -end - const _SPARSE_SCHEMAS = (Schema.STATUS, Schema.DEFINITION, Schema.IMBALANCE) # StatusMsg.action codes that warrant a console alert per Databento's status @@ -81,8 +75,8 @@ Base.@kwdef mutable struct SessionStats error::Union{Nothing,String} = nothing reconnects::Int = 0 last_record_at::Float64 = 0.0 - last_ts_event_by_id::Dict{UInt32,Int64} = Dict{UInt32,Int64}() - last_ts_recv_by_id::Dict{UInt32,Int64} = Dict{UInt32,Int64}() + # Per-instrument replay timestamps moved to Live (populated by reader, + # consumed by reconnect supervisor) — see Live.last_ts_*_by_id. status_state_by_id::Dict{UInt32,Tuple{UInt16,UInt8}} = Dict{UInt32,Tuple{UInt16,UInt8}}() alarm_status_count::Int = 0 end @@ -380,8 +374,13 @@ _is_bbo_family(s::Schema.T) = s in ( Schema.TBBO, Schema.TCBBO, ) -function _replay_start_ts(stats::SessionStats, schema::Schema.T) - d = _is_bbo_family(schema) ? stats.last_ts_recv_by_id : stats.last_ts_event_by_id +# Replay-state lookup used by the reconnect supervisor. Reads from the +# Live's per-instrument timestamp dicts (populated by the reader's +# `_record_replay!`). Returns the minimum-across-instruments — the earliest +# point we lost coverage on any subscribed instrument, which is the safest +# resubscribe `start=` value (replays a bit more rather than skipping). +function _replay_start_ts(c::Live, schema::Schema.T) + d = _is_bbo_family(schema) ? c.last_ts_recv_by_id : c.last_ts_event_by_id isempty(d) && return nothing return minimum(values(d)) end @@ -401,22 +400,13 @@ function _log_status_alarm!(stats::SessionStats, rec::DBN.StatusMsg, schema::Sch return nothing end -# Data-record path: writes the record to disk + tracks per-instrument -# replay timestamps. Under typed mode this runs once per record off the -# typed data channel; the record type is concrete at the call site. +# Data-record path: writes the record to disk + bumps counters. Replay +# bookkeeping is owned by the Live client (see Live.last_ts_*_by_id, +# populated by the reader's _record_replay! before the put!). function _handle_data!(ctx::SessionContext, rec) s = ctx.stats s.data_count += 1 s.last_record_at = time() - if hasproperty(rec, :hd) && hasproperty(rec.hd, :instrument_id) - iid = rec.hd.instrument_id - if hasproperty(rec.hd, :ts_event) - s.last_ts_event_by_id[iid] = rec.hd.ts_event - end - if hasproperty(rec, :ts_recv) - s.last_ts_recv_by_id[iid] = rec.ts_recv - end - end # StatusMsg additionally checks for alarm-worthy state transitions. if rec isa DBN.StatusMsg _log_status_alarm!(s, rec, ctx.schema) @@ -532,99 +522,140 @@ function _run_unified_session(ctxs::Dict{Schema.T,SessionContext}; channel_size::Integer, start_initial, snapshot::Bool) - # Stable subscribe! order so reconnect re-subscribes in the same sequence. + # Stable subscribe! order keeps any future replay log deterministic. schemas = sort!(collect(keys(ctxs)); by = s -> Int(s)) - while true - _any_shutdown(ctxs) && return - deadline !== nothing && time() >= deadline && return - - client = Live(key; - dataset = String(dataset), - gateway = gateway, port = port, - ts_out = false, - compression = wire_compression, - heartbeat_interval = heartbeat_interval, - slow_reader_behavior = slow_reader_behavior, - channel_size = channel_size, - typed = true, - ) - for ctx in values(ctxs); ctx.current_client = client; end - data_channels = Dict{Schema.T,Any}() - consumer_tasks = Dict{Schema.T,Task}() - ctrl_drainer::Union{Nothing,Task} = nothing - try - connect!(client) - for sch in schemas - ctx = ctxs[sch] - start_ts = ctx.stats.reconnects == 0 ? start_initial : - _replay_start_ts(ctx.stats, sch) - start_ts = _bound_replay(start_ts) - data_channels[sch] = subscribe!(client; - schema = sch, - symbols = symbols, - stype_in = stype_in, - snapshot = snapshot, - start = start_ts, - ) - end - start!(client) - - ctrl_drainer = @async _drain_control!(ctxs, control_channel(client)) - for sch in schemas - ctx = ctxs[sch] - ch = data_channels[sch] - consumer_tasks[sch] = @async _drain_data!(ctx, ch, deadline) - end + # Single Live for the lifetime of this session. Under reconnect=true, + # the Live-layer supervisor reopens the socket + replays subscriptions + # internally; the streaming layer no longer owns a reconnect loop. + client = Live(key; + dataset = String(dataset), + gateway = gateway, port = port, + ts_out = false, + compression = wire_compression, + heartbeat_interval = heartbeat_interval, + slow_reader_behavior = slow_reader_behavior, + channel_size = channel_size, + typed = true, + reconnect_policy = reconnect ? ReconnectPolicy.RECONNECT : ReconnectPolicy.NONE, + max_reconnect_attempts = max_reconnect_attempts, + ) + for ctx in values(ctxs); ctx.current_client = client; end + + # Mirror the supervisor's reconnect successes into per-schema stats so + # the existing stats-based monitoring (heartbeat log, smoke assertions, + # potential dashboards) keeps its `reconnects` counter live. + if reconnect + add_reconnect_callback(client, (_g0, _g1) -> begin + for ctx in values(ctxs); ctx.stats.reconnects += 1; end + end) + end - # Main coordination loop: poll for shutdown/deadline/error and - # service per-schema flush requests. Exits when every consumer - # task has finished (reader closed all channels) or the loop is - # told to stop. + data_channels = Dict{Schema.T,Any}() + consumer_tasks = Dict{Schema.T,Task}() + ctrl_drainer::Union{Nothing,Task} = nothing + try + # Initial connect with retry. The Live-layer supervisor only + # kicks in after start! (it watches the reader task), so a + # gateway that hangs up mid-handshake on the *first* attempt + # would otherwise abort the whole session. Retry per the same + # cadence — immediate phase then exp backoff — bounded by + # max_reconnect_attempts (1 + cap total attempts: initial + cap + # retries). + connect_attempt = 0 + while true try - while !all(istaskdone, values(consumer_tasks)) - _any_shutdown(ctxs) && break - deadline !== nothing && time() >= deadline && break - _any_error(ctxs) && break - for ctx in values(ctxs) - while isready(ctx.flush_request) - try; take!(ctx.flush_request); catch; end - _flush!(ctx.file) - end + connect!(client) + break + catch e + e isa InvalidStateException && rethrow() + connect_attempt += 1 + if reconnect && (max_reconnect_attempts === nothing || + connect_attempt <= max_reconnect_attempts) + for ctx in values(ctxs); ctx.stats.reconnects += 1; end + delay = _reconnect_delay(connect_attempt; + immediate = client.immediate_reconnect_attempts) + @warn "live initial connect failed — retrying" attempt=connect_attempt delay_s=round(delay; digits=2) + sleep(delay) + if _any_shutdown(ctxs) || + (deadline !== nothing && time() >= deadline) + rethrow() end - sleep(_CONNECTION_POLL_S) + continue end - catch e - if !(e isa InvalidStateException) && !_any_shutdown(ctxs) - @warn "live session error" schemas=schemas exception=(e, catch_backtrace()) + if reconnect + @warn "live initial connect — retry budget exhausted" attempts=connect_attempt limit=max_reconnect_attempts end + rethrow() + end + end + + for sch in schemas + ctx = ctxs[sch] + # First-subscription start_ts comes from the caller. After a + # reconnect the supervisor recomputes per-schema via + # _replay_start_ts(c::Live, schema) and re-issues subscribe + # frames itself — no streaming-level branching needed here. + start_ts = _bound_replay(start_initial) + data_channels[sch] = subscribe!(client; + schema = sch, + symbols = symbols, + stype_in = stype_in, + snapshot = snapshot, + start = start_ts, + ) + end + start!(client) + + ctrl_drainer = @async _drain_control!(ctxs, control_channel(client)) + for sch in schemas + ctx = ctxs[sch] + ch = data_channels[sch] + consumer_tasks[sch] = @async _drain_data!(ctx, ch, deadline) + end + + # Main coordination loop: poll for shutdown/deadline/error and + # service per-schema flush requests. Exits when every consumer task + # has finished — supervisor reaching a terminal state (:failed / + # :closed) closes the channels, which makes the take!s raise + # InvalidStateException and the drainers return. Also exits on + # explicit shutdown/deadline/error so we close(client) below and + # tell the supervisor to stop reconnecting. + try + while !all(istaskdone, values(consumer_tasks)) + _any_shutdown(ctxs) && break + deadline !== nothing && time() >= deadline && break + _any_error(ctxs) && break + for ctx in values(ctxs) + while isready(ctx.flush_request) + try; take!(ctx.flush_request); catch; end + _flush!(ctx.file) + end + end + sleep(_CONNECTION_POLL_S) end catch e if !(e isa InvalidStateException) && !_any_shutdown(ctxs) @warn "live session error" schemas=schemas exception=(e, catch_backtrace()) end - finally - for ctx in values(ctxs); ctx.current_client = nothing; end - try; close(client); catch; end - # Closing the client closes every typed channel + the control - # channel; wait for drainers to flush in-flight records. - for t in values(consumer_tasks); try; wait(t); catch; end; end - ctrl_drainer === nothing || (try; wait(ctrl_drainer); catch; end) end - - _any_shutdown(ctxs) && return - _any_error(ctxs) && return - !reconnect && return - deadline !== nothing && time() >= deadline && return - for ctx in values(ctxs); ctx.stats.reconnects += 1; end - attempt = first(values(ctxs)).stats.reconnects - if max_reconnect_attempts !== nothing && attempt > max_reconnect_attempts - @warn "live reconnect attempts exhausted — giving up" schemas=schemas attempts=attempt limit=max_reconnect_attempts - return + catch e + if !(e isa InvalidStateException) && !_any_shutdown(ctxs) + @warn "live session error" schemas=schemas exception=(e, catch_backtrace()) + end + finally + for ctx in values(ctxs); ctx.current_client = nothing; end + # close(client) transitions supervisor state→:closed (its next + # check exits cleanly) and closes every typed channel + the + # control channel; wait for drainers to flush in-flight records. + try; close(client); catch; end + for t in values(consumer_tasks); try; wait(t); catch; end; end + ctrl_drainer === nothing || (try; wait(ctrl_drainer); catch; end) + # Let the supervisor finish its current iteration so any in-flight + # reconnect attempt observes c.closed and exits. + if client.reconnect_supervisor !== nothing + try; wait(client.reconnect_supervisor); catch; end end - delay = _reconnect_delay(attempt) - @warn "live disconnect — reconnecting" schemas=schemas attempt=attempt delay_s=round(delay; digits=2) - _wait_for_reconnect(delay, ctxs, deadline) end end diff --git a/src/live/subscribe.jl b/src/live/subscribe.jl index effc367..a74aa04 100644 --- a/src/live/subscribe.jl +++ b/src/live/subscribe.jl @@ -104,19 +104,38 @@ function start!(c::Live) write_text_frame(c.socket; start_session = 0) + # Under ReconnectPolicy.NONE the reader task owns the channels' lifetime: + # `bind` closes them when the reader exits, which is the right shutdown + # signal for iterators / subscribe_callback loops since no new reader is + # coming. Under RECONNECT the channels are owned by the Live client and + # outlive any single reader incarnation — the supervisor (added later) + # respawns a reader writing into the same channels after each drop, so + # binding would prematurely terminate consumers across reconnects. + bind_channels = c.reconnect_policy == ReconnectPolicy.NONE + if c.typed c.reader_task = @async _reader_loop_typed(c) - # Bind each typed data channel + the control channel to the reader - # task so they auto-close when the reader exits. - for ch in values(c.typed_data_channels) - bind(ch, c.reader_task) + if bind_channels + for ch in values(c.typed_data_channels) + bind(ch, c.reader_task) + end + bind(c.control_channel, c.reader_task) end - bind(c.control_channel, c.reader_task) else c.reader_task = @async _reader_loop(c) - bind(c.channel, c.reader_task) + bind_channels && bind(c.channel, c.reader_task) end c.started = true + lock(c.state_lock) do + c.state = :streaming + end + # Spawn the supervisor under RECONNECT so any reader exit is followed + # by a fresh socket + resubscribe + new reader writing into the same + # channels. Iteration and subscribe_callback consumers see a continuous + # record stream across drops. + if c.reconnect_policy != ReconnectPolicy.NONE + c.reconnect_supervisor = @async _supervisor_loop(c) + end return c end diff --git a/test/runtests.jl b/test/runtests.jl index 5684e98..2cedb3d 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -23,6 +23,7 @@ try include("test_live_subscribe.jl") include("test_live_streaming.jl") include("test_reconnect_backoff.jl") + include("test_live_reconnect_supervisor.jl") include("test_typed_mode_errors.jl") end catch e diff --git a/test/test_live_reader.jl b/test/test_live_reader.jl index 945ccdf..ae05d2a 100644 --- a/test/test_live_reader.jl +++ b/test/test_live_reader.jl @@ -79,6 +79,7 @@ const TEST_CHALLENGE = "abcdef0123456789" gateway = "127.0.0.1", port = mock.port, channel_size = 16, + reconnect_policy = :none, ) connect!(client) @test client.connected diff --git a/test/test_live_reader_typed.jl b/test/test_live_reader_typed.jl index 780c7c7..6467e30 100644 --- a/test/test_live_reader_typed.jl +++ b/test/test_live_reader_typed.jl @@ -116,7 +116,8 @@ end dataset = "TEST.MOCK", gateway = "127.0.0.1", port = mock.port, channel_size = 64, - typed = true) + typed = true, + reconnect_policy = :none) DatabentoAPI.connect!(client) ch = DatabentoAPI.subscribe!(client; schema = Schema.TRADES, symbols = ["AAPL"], @@ -203,7 +204,8 @@ end dataset = "TEST.MOCK", gateway = "127.0.0.1", port = mock.port, channel_size = 32, - typed = true) + typed = true, + reconnect_policy = :none) DatabentoAPI.connect!(client) ch_trades = DatabentoAPI.subscribe!(client; schema = Schema.TRADES, symbols = ["AAPL"], stype_in = SType.RAW_SYMBOL) @@ -270,7 +272,8 @@ end dataset = "TEST.MOCK", gateway = "127.0.0.1", port = mock.port, channel_size = 32, - typed = true) + typed = true, + reconnect_policy = :none) DatabentoAPI.connect!(client) ch_1s = DatabentoAPI.subscribe!(client; schema = Schema.OHLCV_1S, symbols = ["AAPL"], stype_in = SType.RAW_SYMBOL) @@ -336,7 +339,8 @@ end dataset = "TEST.MOCK", gateway = "127.0.0.1", port = mock.port, channel_size = 32, - typed = true) + typed = true, + reconnect_policy = :none) DatabentoAPI.connect!(client) ch_mbp1 = DatabentoAPI.subscribe!(client; schema = Schema.MBP_1, symbols = ["AAPL"], stype_in = SType.RAW_SYMBOL) @@ -368,7 +372,8 @@ end client = DatabentoAPI.Live(_TEST_API_KEY_TYPED; dataset = "TEST.MOCK", gateway = "127.0.0.1", port = 1, # no real connection needed for the check - typed = true) + typed = true, + reconnect_policy = :none) # Mark connected so subscribe! validation runs (we still won't make # a real socket call because subscribe! checks the schema first). client.connected = true diff --git a/test/test_live_reconnect_supervisor.jl b/test/test_live_reconnect_supervisor.jl new file mode 100644 index 0000000..3921785 --- /dev/null +++ b/test/test_live_reconnect_supervisor.jl @@ -0,0 +1,339 @@ +using Test +using DatabentoAPI +using DatabentoAPI: read_text_frame, build_text_frame, _gap_start_ns +using DatabentoBinaryEncoding +import DatabentoBinaryEncoding as DBN +using Sockets + +const _SUPERVISOR_TEST_KEY = "db-0123456789abcdef0123456789abcdef" + +# Build a DBN stream containing `n` trade records, with each record offset +# from `ts_base` (ns) so two sequences can be checked for non-overlap. +function _supervisor_dbn_bytes(n::Int; ts_base::Int64 = Int64(1_700_000_000_000_000_000), + id_base::Int = 100, dataset = "TEST.MOCK") + metadata = DBN.Metadata( + DBN.DBN_VERSION, dataset, DBN.Schema.TRADES, + ts_base, nothing, nothing, nothing, + DBN.SType.INSTRUMENT_ID, false, + ["AAPL"], String[], String[], + Tuple{String,String,Int64,Int64}[], + ) + records = DBN.DBNRecord[] + for i in 1:n + hd = DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, + UInt16(1), UInt32(id_base + i), + ts_base + Int64(i) * Int64(1_000_000)) + push!(records, DBN.TradeMsg( + hd, Int64(150_000_000_000 + i * 1_000_000), UInt32(100), + DBN.Action.TRADE, DBN.Side.ASK, UInt8(0), UInt8(1), + ts_base + Int64(i) * Int64(1_000_000), Int32(0), UInt32(i), + )) + end + tmp, io = mktemp(); close(io) + try + DBN.write_dbn(tmp, metadata, records) + return read(tmp), n + finally + rm(tmp; force = true) + end +end + +# Server-side handshake matching what `connect!` expects from the gateway. +function _supervisor_handshake!(sock) + write(sock, build_text_frame(lsg_version = "0.9.0")) + write(sock, build_text_frame(cram = "challenge")) + flush(sock) + _ = read_text_frame(sock) # auth + write(sock, build_text_frame(success = "1", session_id = "sess")) + flush(sock) + _ = read_text_frame(sock) # subscribe + _ = read_text_frame(sock) # start_session +end + +@testset "live reconnect supervisor — iteration survives a disconnect" begin + bytes1, n1 = _supervisor_dbn_bytes(3; ts_base = Int64(1_700_000_000_000_000_000), + id_base = 100) + # Second session uses a later metadata.start_ts so the callback's gap_end + # is distinguishable from anything seen pre-drop. + bytes2, n2 = _supervisor_dbn_bytes(4; ts_base = Int64(1_700_000_001_000_000_000), + id_base = 200) + + script1 = function(sock) + _supervisor_handshake!(sock) + write(sock, bytes1) + flush(sock) + # Give the client generous time to drain (cold JIT can push first-run + # reader compile latency past several seconds on a fresh process). + sleep(2.0) + close(sock) + end + script2 = function(sock) + _supervisor_handshake!(sock) + write(sock, bytes2) + flush(sock) + sleep(2.0) + end + + mock = spawn_mock_gateway_sequence([script1, script2]) + callback_args = Vector{Tuple{Int64,Int64}}() + + client = Live(_SUPERVISOR_TEST_KEY; + dataset = "TEST.MOCK", + gateway = "127.0.0.1", port = mock.port, + compression = DBN.Compression.NONE, + reconnect_policy = :reconnect, + max_reconnect_attempts = 5, + immediate_reconnect_attempts = 3, + ) + add_reconnect_callback(client, (g0, g1) -> push!(callback_args, (g0, g1))) + + connect!(client) + subscribe!(client; + schema = DBN.Schema.TRADES, + symbols = ["AAPL"], + stype_in = DBN.SType.INSTRUMENT_ID, + ) + start!(client) + + collected = DBN.DBNRecord[] + deadline = time() + 20.0 + while time() < deadline && length(collected) < n1 + n2 + try + push!(collected, take!(client.channel)) + catch e + e isa InvalidStateException && break + rethrow() + end + end + close(client) + # Drain the supervisor before asserting on side-effects (callback list). + # Close transitions state→:closed; supervisor exits its loop after the + # current iteration. Without this wait, the supervisor's _fire_reconnect_- + # callbacks may not have run by the time we check, particularly under a + # warm-JIT full-suite run where everything is fast. + if client.reconnect_supervisor !== nothing + try; wait(client.reconnect_supervisor); catch; end + end + try; wait(mock.accept_task); catch; end + + @test length(collected) >= n1 + n2 + @test all(r -> r isa DBN.TradeMsg, collected) + # Pre-drop record ids run in 101..103; post-drop in 201..204. + pre_drop = [Int(r.hd.instrument_id) for r in collected if r.hd.instrument_id <= 103] + post_drop = [Int(r.hd.instrument_id) for r in collected if r.hd.instrument_id >= 201] + @test length(pre_drop) == n1 + @test length(post_drop) == n2 + + @test length(callback_args) == 1 + g0, g1 = callback_args[1] + @test g0 > 0 # we saw records pre-drop + @test g1 >= g0 # gap_end after gap_start + @test g1 == Int64(1_700_000_001_000_000_000) # script2's metadata.start_ts +end + +@testset "live reconnect supervisor — gateway ErrorMsg is terminal" begin + # Build a payload that sends one trade then an ErrorMsg. The reader's + # control path sets c.terminal_error; the supervisor must NOT reconnect. + metadata = DBN.Metadata( + DBN.DBN_VERSION, "TEST.MOCK", DBN.Schema.TRADES, + Int64(1_700_000_000_000_000_000), nothing, nothing, nothing, + DBN.SType.INSTRUMENT_ID, false, + ["AAPL"], String[], String[], + Tuple{String,String,Int64,Int64}[], + ) + hd_trade = DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, + UInt16(1), UInt32(100), + Int64(1_700_000_000_000_000_000)) + trade = DBN.TradeMsg( + hd_trade, Int64(150_000_000_000), UInt32(100), + DBN.Action.TRADE, DBN.Side.ASK, UInt8(0), UInt8(1), + Int64(1_700_000_000_000_000_000), Int32(0), UInt32(1), + ) + hd_err = DBN.RecordHeader(UInt8(0), DBN.RType.ERROR_MSG, + UInt16(1), UInt32(0), + Int64(1_700_000_000_000_000_000)) + err = DBN.ErrorMsg(hd_err, "fatal-test-err") + tmp, io = mktemp(); close(io) + bytes = try + DBN.write_dbn(tmp, metadata, DBN.DBNRecord[trade, err]) + read(tmp) + finally + rm(tmp; force = true) + end + + accept_count = Threads.Atomic{Int}(0) + script = function(sock) + Threads.atomic_add!(accept_count, 1) + _supervisor_handshake!(sock) + write(sock, bytes) + flush(sock) + sleep(2.0) # let client drain + close(sock) + end + # Pre-stage extras; supervisor should refuse to reconnect after the + # first session's ErrorMsg, so the second accept should never happen. + mock = spawn_mock_gateway_sequence([script for _ in 1:5]) + + callback_count = Ref(0) + client = Live(_SUPERVISOR_TEST_KEY; + dataset = "TEST.MOCK", + gateway = "127.0.0.1", port = mock.port, + compression = DBN.Compression.NONE, + typed = true, # use control channel + reconnect_policy = :reconnect, + max_reconnect_attempts = 5, + ) + add_reconnect_callback(client, (g0, g1) -> (callback_count[] += 1)) + + connect!(client) + data_ch = subscribe!(client; + schema = DBN.Schema.TRADES, + symbols = ["AAPL"], + stype_in = DBN.SType.INSTRUMENT_ID, + ) + start!(client) + + # Drain the data channel until it closes (terminal state closes channels). + seen = DBN.TradeMsg[] + deadline = time() + 10.0 + while time() < deadline + try + push!(seen, take!(data_ch)) + catch e + e isa InvalidStateException && break + rethrow() + end + end + close(client) + if client.reconnect_supervisor !== nothing + try; wait(client.reconnect_supervisor); catch; end + end + try; close(mock.server); catch; end + try; wait(mock.accept_task); catch; end + + @test accept_count[] == 1 # never reconnected + @test client.terminal_error !== nothing # ErrorMsg captured + @test callback_count[] == 0 # no reconnect callback fired + @test length(seen) >= 1 # got the pre-error trade +end + +@testset "live reconnect supervisor — ErrorMsg terminal under UNTYPED mode" begin + # Regression guard: with the default reconnect_policy = RECONNECT, an + # untyped client receiving a gateway ErrorMsg must NOT burn through + # max_reconnect_attempts re-hitting the same deterministic failure. + # The untyped reader sets c.terminal_error on ErrorMsg the same way the + # typed reader's control branch does. + metadata = DBN.Metadata( + DBN.DBN_VERSION, "TEST.MOCK", DBN.Schema.TRADES, + Int64(1_700_000_000_000_000_000), nothing, nothing, nothing, + DBN.SType.INSTRUMENT_ID, false, + ["AAPL"], String[], String[], + Tuple{String,String,Int64,Int64}[], + ) + hd_trade = DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, + UInt16(1), UInt32(100), + Int64(1_700_000_000_000_000_000)) + trade = DBN.TradeMsg( + hd_trade, Int64(150_000_000_000), UInt32(100), + DBN.Action.TRADE, DBN.Side.ASK, UInt8(0), UInt8(1), + Int64(1_700_000_000_000_000_000), Int32(0), UInt32(1), + ) + hd_err = DBN.RecordHeader(UInt8(0), DBN.RType.ERROR_MSG, + UInt16(1), UInt32(0), + Int64(1_700_000_000_000_000_000)) + err = DBN.ErrorMsg(hd_err, "untyped-fatal-test-err") + tmp, io = mktemp(); close(io) + bytes = try + DBN.write_dbn(tmp, metadata, DBN.DBNRecord[trade, err]) + read(tmp) + finally + rm(tmp; force = true) + end + + accept_count = Threads.Atomic{Int}(0) + script = function(sock) + Threads.atomic_add!(accept_count, 1) + _supervisor_handshake!(sock) + write(sock, bytes) + flush(sock) + sleep(2.0) + close(sock) + end + mock = spawn_mock_gateway_sequence([script for _ in 1:5]) + + client = Live(_SUPERVISOR_TEST_KEY; + dataset = "TEST.MOCK", + gateway = "127.0.0.1", port = mock.port, + compression = DBN.Compression.NONE, + # Default reconnect_policy = :reconnect — explicit here for clarity + # but the bug this test guards would manifest without specifying. + reconnect_policy = :reconnect, + max_reconnect_attempts = 5, + ) + connect!(client) + subscribe!(client; + schema = DBN.Schema.TRADES, symbols = ["AAPL"], + stype_in = DBN.SType.INSTRUMENT_ID, + ) + start!(client) + + seen = DBN.DBNRecord[] + deadline = time() + 10.0 + while time() < deadline + try + push!(seen, take!(client.channel)) + catch e + e isa InvalidStateException && break + rethrow() + end + end + close(client) + if client.reconnect_supervisor !== nothing + try; wait(client.reconnect_supervisor); catch; end + end + try; close(mock.server); catch; end + try; wait(mock.accept_task); catch; end + + @test accept_count[] == 1 # never reconnected + @test client.terminal_error !== nothing # ErrorMsg captured + @test any(r -> r isa DBN.TradeMsg, seen) # pre-error trade got through + @test any(r -> r isa DBN.ErrorMsg, seen) # ErrorMsg also flows to consumer +end + +@testset "live_session bundles connect/subscribe/start + close" begin + bytes, n = _supervisor_dbn_bytes(3; ts_base = Int64(1_700_000_000_000_000_000), + id_base = 100) + script = function(sock) + _supervisor_handshake!(sock) + write(sock, bytes); flush(sock) + sleep(2.0) + close(sock) + end + mock = spawn_mock_gateway_sequence([script]) + + collected = DBN.DBNRecord[] + live_session(; + dataset = "TEST.MOCK", + subscriptions = [(; schema = DBN.Schema.TRADES, + symbols = ["AAPL"], + stype_in = DBN.SType.INSTRUMENT_ID)], + reconnect_policy = :none, # single-shot for this smoke test + key = _SUPERVISOR_TEST_KEY, + gateway = "127.0.0.1", port = mock.port, + compression = DBN.Compression.NONE, + ) do client + deadline = time() + 10.0 + while time() < deadline && length(collected) < n + try + push!(collected, take!(client.channel)) + catch e + e isa InvalidStateException && break + rethrow() + end + end + end + try; wait(mock.accept_task); catch; end + + @test length(collected) == n + @test all(r -> r isa DBN.TradeMsg, collected) +end diff --git a/test/test_live_streaming.jl b/test/test_live_streaming.jl index 85e4355..fd4d1b6 100644 --- a/test/test_live_streaming.jl +++ b/test/test_live_streaming.jl @@ -111,17 +111,20 @@ const _TEST_KEY = "db-1234567890abcdef12345" end @testset "live streaming — _replay_start_ts" begin - s = SessionStats(schema = Schema.TCBBO) - @test _replay_start_ts(s, Schema.TCBBO) === nothing - @test _replay_start_ts(s, Schema.TRADES) === nothing - - s.last_ts_event_by_id[UInt32(1)] = Int64(1_000_000) - s.last_ts_event_by_id[UInt32(2)] = Int64(3_000_000) - s.last_ts_recv_by_id[UInt32(1)] = Int64(2_000_000) - s.last_ts_recv_by_id[UInt32(2)] = Int64(5_000_000) - @test _replay_start_ts(s, Schema.TRADES) == Int64(1_000_000) # min ts_event - @test _replay_start_ts(s, Schema.TCBBO) == Int64(2_000_000) # min ts_recv - @test _replay_start_ts(s, Schema.CBBO_1S) == Int64(2_000_000) + # _replay_start_ts now reads from Live (populated by the reader's + # _record_replay! helper before each put!). + c = Live("db-0123456789abcdef0123456789abcdef"; dataset = "TEST.MOCK", + gateway = "127.0.0.1", port = 13000, reconnect_policy = :none) + @test _replay_start_ts(c, Schema.TCBBO) === nothing + @test _replay_start_ts(c, Schema.TRADES) === nothing + + c.last_ts_event_by_id[UInt32(1)] = Int64(1_000_000) + c.last_ts_event_by_id[UInt32(2)] = Int64(3_000_000) + c.last_ts_recv_by_id[UInt32(1)] = Int64(2_000_000) + c.last_ts_recv_by_id[UInt32(2)] = Int64(5_000_000) + @test _replay_start_ts(c, Schema.TRADES) == Int64(1_000_000) # min ts_event + @test _replay_start_ts(c, Schema.TCBBO) == Int64(2_000_000) # min ts_recv + @test _replay_start_ts(c, Schema.CBBO_1S) == Int64(2_000_000) end @testset "live streaming — status alarm dedup" begin diff --git a/test/test_live_subscribe.jl b/test/test_live_subscribe.jl index 0f8b9f9..010b96d 100644 --- a/test/test_live_subscribe.jl +++ b/test/test_live_subscribe.jl @@ -54,7 +54,8 @@ end bytes = _build_tiny_payload() mock = spawn_mock_gateway(_make_handshake_with(bytes)) client = Live(TEST_API_KEY_2; - dataset = "GLBX.MDP3", gateway = "127.0.0.1", port = mock.port) + dataset = "GLBX.MDP3", gateway = "127.0.0.1", port = mock.port, + reconnect_policy = :none) connect!(client) @test_throws ArgumentError subscribe!(client; schema = Schema.TRADES, symbols = "ES.FUT", @@ -68,7 +69,7 @@ end mock = spawn_mock_gateway(_make_handshake_with(bytes)) client = Live(TEST_API_KEY_2; dataset = "GLBX.MDP3", gateway = "127.0.0.1", port = mock.port, - channel_size = 8) + channel_size = 8, reconnect_policy = :none) connect!(client) subscribe!(client; schema = Schema.TRADES, symbols = "ES.FUT", stype_in = SType.PARENT) @@ -95,7 +96,7 @@ end mock = spawn_mock_gateway(_make_handshake_with(bytes)) client = Live(TEST_API_KEY_2; dataset = "GLBX.MDP3", gateway = "127.0.0.1", port = mock.port, - channel_size = 8) + channel_size = 8, reconnect_policy = :none) connect!(client) subscribe!(client; schema = Schema.TRADES, symbols = "ES.FUT") start!(client) diff --git a/test/test_reconnect_backoff.jl b/test/test_reconnect_backoff.jl index 3a62fc1..7227def 100644 --- a/test/test_reconnect_backoff.jl +++ b/test/test_reconnect_backoff.jl @@ -49,6 +49,34 @@ end @test length(unique(draws)) > 10 end +@testset "reconnect — immediate phase returns exactly 0.0" begin + # First `immediate` attempts must short-circuit to 0.0 so transient TCP + # blips recover without the 1s+ backoff floor. + for n in 1:3 + @test _reconnect_delay(n; immediate = 3) === 0.0 + end + # attempt = immediate + 1 enters backoff, drawn from [0, base). + for _ in 1:50 + d = _reconnect_delay(4; immediate = 3) + @test 0 ≤ d < _RECONNECT_BASE_S + end + # attempt = immediate + 2 → [0, 2*base). + for _ in 1:50 + d = _reconnect_delay(5; immediate = 3) + @test 0 ≤ d < 2 * _RECONNECT_BASE_S + end + # immediate = 0 (default) leaves prior behaviour unchanged: attempt=1 + # samples [0, base) just like the no-kwarg form. + for _ in 1:50 + d = _reconnect_delay(1; immediate = 0) + @test 0 ≤ d < _RECONNECT_BASE_S + end + # Large immediate suppresses backoff for any attempt within its window. + for n in 1:10 + @test _reconnect_delay(n; immediate = 100) === 0.0 + end +end + @testset "reconnect — max_reconnect_attempts caps connection attempts" begin # Mock gateway that immediately closes each socket. The client will fail # during the LSG-version read, raise, and fall through to the reconnect @@ -127,10 +155,16 @@ end end elapsed = time() - t0 - # Should be roughly `duration_s` plus a small slack for the last in-flight - # sleep slice (≤ _CONNECTION_POLL_S = 0.5s) and handshake overhead. Without - # the fix, a single sleep of up to 60s could blow this out. - @test elapsed < duration + 3.0 + # The call must return promptly after the deadline: `_wait_for_reconnect` + # slices its sleep so the loop never sits through a full backoff window + # (up to `_RECONNECT_CAP_S` = 60s) when the deadline lands mid-sleep. We + # assert a generous ceiling rather than a tight `duration + ε` — on shared + # macOS / Windows CI runners, per-attempt socket churn and scheduler stalls + # add several seconds of wall-clock unrelated to backoff responsiveness + # (the old `duration + 3.0` bound flaked here, observed 5.0s). Anything in + # the low-double-digits proves the in-flight sleep was cut short; a + # regression that ignored the deadline would run for tens of seconds. + @test elapsed < 15.0 try; close(mock.server); catch; end end