From ee692c231520ef6bdf77c020e2de83c09759b663 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Tue, 19 May 2026 19:02:10 -0500 Subject: [PATCH] Live capture durability: sidecar, sentinel, multi-frame MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three related changes to stream_to_file / stream_multi_to_files so captures survive both clean and unexpected shutdowns and the SymbolMappingMsg records the Live gateway sends at subscribe time are persisted alongside each output file. 1. SymbolMappingMsg sidecar (.symbology.jsonl per RotatingDBNFile) -------------------------------------------------------------- The Live gateway emits SymbolMappingMsg in v1 wire layout (~80 bytes, hd.length=20). DBN.jl's encoder, writing into a v3-metadata file, would lay them out as v3 (~180 bytes) while keeping the on-wire hd.length, corrupting the file. So we route mappings to a JSONL sidecar (one JSON object per line) next to each .dbn.zst. _append_symbology! is called from _handle_control!, flushes per-line for crash safety, and the sidecar opens/closes/ rotates in lockstep with its main DBN file. Sample sidecar line: {"ts_event":1779203015494476543,"instrument_id":1191182337, "stype_in":"INSTRUMENT_ID","stype_in_symbol":"SPX.OPT", "stype_out":"INSTRUMENT_ID","stype_out_symbol":"SPX 271217C02800000", "start_ts":-1,"end_ts":-1} 2. Stop-sentinel shutdown ----------------------- New stop_sentinel kwarg (default joinpath(base_dir,"STOP")). _watch_sentinel polls once per second; on detection it calls _request_shutdown! on every ctx so the capture exits through its normal finally → _close_stack! path (writes final zstd frame footer cleanly). Any stale sentinel from a previous run is removed at startup so it doesn't trigger immediate shutdown. To stop a running capture cleanly: New-Item -Path '\STOP' -ItemType File 3. In-file zstd frame rotation (crash safety) ------------------------------------------- New frame_seconds kwarg (default 60s). _rotate_frame_if_needed! uses TranscodingStreams.TOKEN_END + flush to seal the active zstd frame and start a new one on the SAME stream — no encoder or stream re-creation needed. Resulting .dbn.zst is a multi- frame zstd file (standards-compliant; libzstd concatenates frame payloads on read). A hard kill loses at most one open frame instead of corrupting the whole file. Crucially, _rotate_frame_if_needed! is invoked both from _write_record! (the high-traffic path) AND from _session_monitor (the only path that fires for quiet schemas — _write_record! never runs when no records arrive). The lock-free entry point wraps the rotation body in lock(f.write_lock); ReentrantLock allows the writer-path call to re-enter the lock it already holds. The monitor also calls _flush! directly so a quiet schema's buffered metadata header reaches disk. Test coverage: all 236 existing tests still pass. Multi-frame round-trip and clean-shutdown round-trip were validated by hand during a live OPRA.PILLAR capture session (T213349Z files, 109- 110 bytes, parse cleanly with strict DBN.read_dbn_with_metadata after sentinel-triggered shutdown). Co-Authored-By: Claude Opus 4.7 (1M context) --- src/live/streaming.jl | 200 +++++++++++++++++++++++++++++++++++++----- 1 file changed, 180 insertions(+), 20 deletions(-) diff --git a/src/live/streaming.jl b/src/live/streaming.jl index ffbc889..a9229bf 100644 --- a/src/live/streaming.jl +++ b/src/live/streaming.jl @@ -63,6 +63,23 @@ mutable struct RotatingDBNFile rotations::Int all_paths::Vector{String} write_lock::ReentrantLock + # Sidecar JSONL file capturing SymbolMappingMsg events. Rotates alongside + # the main DBN file. We don't write mappings into the .dbn.zst itself + # because the live gateway emits them in v1 layout (80 bytes) while our + # file metadata is v3 — DBN.jl's encoder would write them as v3 (~180 + # bytes) with the wire hd.length=20, corrupting the file. JSONL sidecar + # avoids that mismatch entirely and is human-readable / easy to join. + sidecar_io::Union{Nothing,IO} + sidecar_path::String + sidecar_paths::Vector{String} + # In-file zstd frame rotation for crash safety. Every `frame_seconds`, + # close the current zstd frame (writes footer to raw_io) and open a new + # one on the same file. Multi-frame .dbn.zst is standards-compliant to + # any zstd reader, so a hard kill loses ≤ one frame of records rather + # than corrupting the whole file. `nothing` disables it (single frame). + frame_seconds::Union{Nothing,Float64} + frame_opened_at::Float64 + frame_count::Int end function RotatingDBNFile(; base_dir::AbstractString, @@ -73,7 +90,8 @@ function RotatingDBNFile(; base_dir::AbstractString, stype_in::SType.T, compress::Bool, compress_level::Integer, - rotate_seconds::Union{Nothing,Real}) + rotate_seconds::Union{Nothing,Real}, + frame_seconds::Union{Nothing,Real} = nothing) syms_vec = symbols isa AbstractString ? [String(symbols)] : String.(symbols) f = RotatingDBNFile( String(base_dir), @@ -82,11 +100,23 @@ function RotatingDBNFile(; base_dir::AbstractString, compress, Int(compress_level), rotate_seconds === nothing ? nothing : Float64(rotate_seconds), nothing, nothing, nothing, "", 0.0, 0, String[], ReentrantLock(), + nothing, "", String[], + frame_seconds === nothing ? nothing : Float64(frame_seconds), + 0.0, 0, ) _open!(f) return f end +# Sidecar path mirrors the main DBN file, swapping .dbn[.zst] for +# .symbology.jsonl. Lives in the same per-schema directory. +function _sidecar_path_for(dbn_path::AbstractString) + p = String(dbn_path) + base = endswith(p, ".dbn.zst") ? p[1:end-8] : + endswith(p, ".dbn") ? p[1:end-4] : p + return base * ".symbology.jsonl" +end + function _next_path(f::RotatingDBNFile) f.explicit_path === nothing || return f.explicit_path safe = replace(schema_str(f.schema), '/' => '_') @@ -117,15 +147,54 @@ function _open!(f::RotatingDBNFile) ) enc = DBN.DBNEncoder(top, md) DBN.write_header(enc) - f.raw_io = raw - f.zstd_io = top - f.encoder = enc - f.current_path = path - f.opened_at = time() + f.raw_io = raw + f.zstd_io = top + f.encoder = enc + f.current_path = path + f.opened_at = time() + f.frame_opened_at = time() # first frame opens with the file push!(f.all_paths, path) + # Open the symbology sidecar alongside the DBN file. + sc_path = _sidecar_path_for(path) + f.sidecar_io = open(sc_path, "w") + f.sidecar_path = sc_path + push!(f.sidecar_paths, sc_path) return nothing end +# In-file zstd frame rotation. Writes TranscodingStreams' TOKEN_END through +# the active zstd stream + flushes, which makes the codec emit the zstd +# frame footer to raw_io. Subsequent writes start a fresh frame on the +# same stream — the resulting .dbn.zst is a multi-frame zstd file, fully +# standards-compliant. DBN-level layout is unaffected: the metadata +# header was written once at file open, and the decompressed output is +# concatenated across frames, so readers see "header then records". +# +# Crucially this does NOT close the underlying TranscodingStream (which +# would also close raw_io), and does NOT recreate the DBN encoder — same +# instance keeps working across the frame boundary. +function _rotate_frame_if_needed!(f::RotatingDBNFile) + f.frame_seconds === nothing && return false + f.compress || return false # only meaningful for compressed files + # Acquire write_lock so we can be called either from _write_record! + # (already-held lock; ReentrantLock allows reentrance) or from the + # session monitor (separate task, lock-free entry point). + return lock(f.write_lock) do + (time() - f.frame_opened_at) < f.frame_seconds && return false + f.zstd_io === nothing && return false # file closed concurrently + try + write(f.zstd_io, TranscodingStreams.TOKEN_END) + flush(f.zstd_io) + catch e + @warn "frame rotation: writing frame footer raised" schema=f.schema exception=e + return false + end + f.frame_opened_at = time() + f.frame_count += 1 + return true + end +end + function _close_stack!(f::RotatingDBNFile) # ORDER MATTERS: flush + close the zstd stream so the frame footer is # emitted into raw_io, THEN close raw_io. @@ -143,9 +212,14 @@ function _close_stack!(f::RotatingDBNFile) f.raw_io === nothing || close(f.raw_io) catch end - f.encoder = nothing - f.zstd_io = nothing - f.raw_io = nothing + try + f.sidecar_io === nothing || (flush(f.sidecar_io); close(f.sidecar_io)) + catch + end + f.encoder = nothing + f.zstd_io = nothing + f.raw_io = nothing + f.sidecar_io = nothing return nothing end @@ -162,7 +236,9 @@ end function _write_record!(f::RotatingDBNFile, rec) lock(f.write_lock) do - _rotate_if_needed!(f) + # File rotation (new .dbn.zst) takes precedence over frame rotation: + # if we just opened a fresh file, its zstd frame is brand new anyway. + _rotate_if_needed!(f) || _rotate_frame_if_needed!(f) DBN.write_record(f.encoder, rec) end return nothing @@ -170,14 +246,40 @@ end function _flush!(f::RotatingDBNFile) lock(f.write_lock) do - f.zstd_io === nothing || flush(f.zstd_io) - f.raw_io === nothing || flush(f.raw_io) + f.zstd_io === nothing || flush(f.zstd_io) + f.raw_io === nothing || flush(f.raw_io) + f.sidecar_io === nothing || flush(f.sidecar_io) end return nothing end _close!(f::RotatingDBNFile) = lock(() -> _close_stack!(f), f.write_lock) +# Append one JSON line per SymbolMappingMsg to the sidecar. Held under the +# same write_lock as DBN writes so it can't race with a rotation. +function _append_symbology!(f::RotatingDBNFile, rec::DBN.SymbolMappingMsg) + lock(f.write_lock) do + f.sidecar_io === nothing && return nothing + m = ( + ts_event = rec.hd.ts_event, + instrument_id = rec.hd.instrument_id, + stype_in = string(Symbol(rec.stype_in)), + stype_in_symbol = rec.stype_in_symbol, + stype_out = string(Symbol(rec.stype_out)), + stype_out_symbol = rec.stype_out_symbol, + start_ts = rec.start_ts, + end_ts = rec.end_ts, + ) + JSON3.write(f.sidecar_io, m) + write(f.sidecar_io, '\n') + # Mappings are sparse (one per (schema, instrument) at subscribe time, + # then occasional updates). Flushing per-line is cheap and means a + # crashed capture still has a recoverable, complete sidecar. + flush(f.sidecar_io) + end + return nothing +end + # ---------- session context ---------- mutable struct SessionContext @@ -258,18 +360,15 @@ function _handle_data!(ctx::SessionContext, rec) end # Control-record path: SymbolMappingMsg / SystemMsg / ErrorMsg. -# Counts + special handling per type. Never written to disk (mappings -# break round-trip due to v1/v3 layout drift; system/error are noise). +# Counts + special handling per type. SymbolMappingMsg goes to the per-file +# JSONL sidecar (not the .dbn.zst, because gateway emits v1-layout records +# that DBN.jl's v3 encoder would corrupt the file with). SystemMsg/ErrorMsg +# are just counted/logged. function _handle_control!(ctx::SessionContext, rec) s = ctx.stats if rec isa DBN.SymbolMappingMsg - # Don't write SymbolMappingMsg to disk: live gateway sends v1-layout - # records (80 bytes) but our file metadata is v3, so DBN.jl's encoder - # would write them as v3 (~180 bytes) with the original hd.length=20, - # desyncing the file. Schema-pure files are the simplest fix; if the - # caller needs instrument_id → raw_symbol resolution, use the - # `symbology.resolve` historical endpoint with the same symbols. s.mapping_count += 1 + _append_symbology!(ctx.file, rec) elseif rec isa DBN.SystemMsg s.system_count += 1 is_heartbeat(rec) || @debug "SystemMsg" schema=ctx.schema msg=rec.msg @@ -436,12 +535,44 @@ function _session_monitor(ctx::SessionContext, interval_s::Float64, if isopen(ctx.flush_request) && !isready(ctx.flush_request) try; put!(ctx.flush_request, nothing); catch; end end + # Directly flush + rotate frame from the monitor task. This is the + # only mechanism that fires for QUIET schemas — _write_record! only + # runs when records arrive, and the session loop blocks on take! + # when the data channel is empty. Without this, a quiet schema's + # in-memory zstd buffer (including the DBN metadata header) never + # reaches disk and a hard kill loses the entire file. + try; _flush!(ctx.file); catch; end + try; _rotate_frame_if_needed!(ctx.file); catch; end last_count = cur last_tick = now end return nothing end +# ---------- sentinel watcher ---------- + +# Polls for the existence of `sentinel_path` once per second. On detection, +# calls _request_shutdown! on every context so the capture exits cleanly +# through its finally → _close_stack! path (no truncated zstd frames). +# Removes the sentinel after triggering so a leftover file doesn't cause +# the next launch to shut down immediately. +function _watch_sentinel(contexts, sentinel_path::String, stop::Threads.Atomic{Bool}) + while !stop[] + try + if isfile(sentinel_path) + @info "stop sentinel detected — initiating clean shutdown" path=sentinel_path + for ctx in contexts; _request_shutdown!(ctx); end + try; rm(sentinel_path; force = true); catch; end + return nothing + end + catch e + @warn "sentinel watcher error" exception=e + end + sleep(1.0) + end + return nothing +end + # ---------- public API ---------- """ @@ -491,6 +622,8 @@ function stream_to_file(; schema::Schema.T, compress::Bool = true, compress_level::Integer = 1, rotate_seconds::Union{Nothing,Real} = nothing, + frame_seconds::Union{Nothing,Real} = 60.0, + stop_sentinel::Union{Nothing,AbstractString} = nothing, reconnect::Bool = true, key = nothing, gateway = nothing, @@ -508,6 +641,7 @@ function stream_to_file(; schema::Schema.T, dataset = dataset, schema = schema, symbols = symbols, stype_in = stype_in, compress = compress, compress_level = compress_level, rotate_seconds = path === nothing ? rotate_seconds : nothing, + frame_seconds = frame_seconds, ) ctx = SessionContext(schema, SessionStats(schema = schema), file) @@ -515,6 +649,17 @@ function stream_to_file(; schema::Schema.T, monitor_stop = Threads.Atomic{Bool}(false) monitor_task = @async _session_monitor(ctx, Float64(heartbeat_log_interval_s), monitor_stop) + sentinel = stop_sentinel === nothing ? joinpath(base, "STOP") : String(stop_sentinel) + # Stale sentinel from a previous killed run would trigger immediate + # shutdown — clear it on startup so the watcher only fires on a fresh + # touch from this session onward. + if isfile(sentinel) + @warn "removing leftover stop sentinel" path=sentinel + try; rm(sentinel; force = true); catch; end + end + sentinel_stop = Threads.Atomic{Bool}(false) + sentinel_task = @async _watch_sentinel((ctx,), sentinel, sentinel_stop) + try _run_session(ctx; dataset = dataset, stype_in = stype_in, symbols = symbols, @@ -528,7 +673,9 @@ function stream_to_file(; schema::Schema.T, ) finally monitor_stop[] = true + sentinel_stop[] = true try; istaskdone(monitor_task) || sleep(0.1); catch; end + try; istaskdone(sentinel_task) || sleep(0.1); catch; end _close!(file) end return ctx.file.current_path @@ -561,6 +708,8 @@ function stream_multi_to_files(; schemas::AbstractVector, compress::Bool = true, compress_level::Integer = 1, rotate_seconds::Union{Nothing,Real} = nothing, + frame_seconds::Union{Nothing,Real} = 60.0, + stop_sentinel::Union{Nothing,AbstractString} = nothing, reconnect::Bool = true, key = nothing, gateway = nothing, @@ -585,6 +734,7 @@ function stream_multi_to_files(; schemas::AbstractVector, dataset = dataset, schema = sch, symbols = symbols, stype_in = stype_in, compress = compress, compress_level = compress_level, rotate_seconds = rotate_seconds, + frame_seconds = frame_seconds, ) contexts[sch] = SessionContext(sch, SessionStats(schema = sch), file) end @@ -612,6 +762,14 @@ function stream_multi_to_files(; schemas::AbstractVector, end end + sentinel = stop_sentinel === nothing ? joinpath(base, "STOP") : String(stop_sentinel) + if isfile(sentinel) + @warn "removing leftover stop sentinel" path=sentinel + try; rm(sentinel; force = true); catch; end + end + sentinel_stop = Threads.Atomic{Bool}(false) + sentinel_task = @async _watch_sentinel(values(contexts), sentinel, sentinel_stop) + try while any(!istaskdone(t) for t in values(worker_tasks)) sleep(_CONNECTION_POLL_S) @@ -636,9 +794,11 @@ function stream_multi_to_files(; schemas::AbstractVector, end end for s in keys(monitor_stops); monitor_stops[s][] = true; end + sentinel_stop[] = true for t in values(monitor_tasks) try; istaskdone(t) || sleep(0.1); catch; end end + try; istaskdone(sentinel_task) || sleep(0.1); catch; end out = Dict{Schema.T,String}() for (sch, ctx) in contexts