From 430dc8d2313bf90435af9907e124628714e8ba4b Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Tue, 26 May 2026 09:51:43 -0500 Subject: [PATCH 1/2] Live durability: do-block, frame rotation, writer wrapper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds three composable pieces to the live-capture path so durability features are available at the subscription layer (not just inside stream_to_file): 1. Live do-block constructor Mirrors Base.open(f, path). Live(args...; kwargs...) do client; …; end guarantees close(client) in finally — on Ctrl-C, exception, or normal exit — so users iterating records themselves get clean teardown without writing the try/finally boilerplate. Manual lifecycle still works unchanged. 2. open_dbn_writer + write_record! (public API) Thin do-block wrapper over the internal RotatingDBNFile, with the same kwargs as stream_to_file and a stable write_record!(writer, rec) API. Lets users subscribe themselves and persist records with the same crash-safe writer stream_to_file uses internally. 3. In-file zstd frame rotation (default frame_seconds = 60.0) _rotate_frame_if_needed! closes the active zstd frame via TranscodingStreams.TOKEN_END + flush and starts a new one on the same file. Result is a multi-frame .dbn.zst (standards-compliant; any zstd reader concatenates frames transparently). A hard kill loses ≤ one frame of records instead of corrupting the whole file. Also hooked into _session_monitor so quiet schemas reach disk. Also: - Base.close(Live) hardened: flag-flip happens up-front so re-entry is a no-op; channel cleanup runs unconditionally (typed-data dict is empty in untyped mode, so the iteration is a no-op there). - SymbolMappingMsg now flows through _write_record! into the .dbn.zst (DatabentoBinaryEncoding ≥ 0.1.1 fixes the v1→v3 layout re-encode, so the workaround dropping the record is no longer needed). - frame_seconds kwarg added to stream_to_file / stream_multi_to_files. - README: canonical Live example switched to do-block form; new "Writing records to disk yourself" section showing open_dbn_writer. - Version 0.1.1 → 0.1.2. New tests (1587/1587 pass, +29 vs main): - Live do-block cleanup on normal/exception/InterruptException exit. - Base.close re-entry safety. - open_dbn_writer cleanup on normal exit and InterruptException. - write_record! smoke + round-trip. - Multi-frame zstd round-trips as one continuous record stream. - SymbolMappingMsg now lands in the .dbn.zst (load-bearing for the upstream encoder fix integration). Co-Authored-By: Claude Opus 4.7 --- Project.toml | 2 +- README.md | 68 ++++++++--- src/DatabentoAPI.jl | 1 + src/live/client.jl | 55 +++++++-- src/live/streaming.jl | 229 +++++++++++++++++++++++++++------- test/test_live_streaming.jl | 238 +++++++++++++++++++++++++++++++++++- 6 files changed, 526 insertions(+), 67 deletions(-) diff --git a/Project.toml b/Project.toml index 7d504aa..1af1497 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "DatabentoAPI" uuid = "feb0aed8-3291-460b-9688-521dcb17a4bf" -version = "0.1.1" +version = "0.1.2" authors = ["Tyler Beason "] [deps] diff --git a/README.md b/README.md index 3fdd42e..8bb9a20 100644 --- a/README.md +++ b/README.md @@ -99,32 +99,72 @@ resolve, foreach_record ## Live example +The do-block form mirrors `Base.open` and guarantees `close(client)` runs on +`Ctrl-C`, exceptions, or normal exit — no `try/finally` boilerplate needed. + ```julia using DatabentoAPI -client = Live(dataset = "OPRA.PILLAR") -connect!(client) -subscribe!(client; - schema = Schema.TRADES, - symbols = ["AAPL"], - stype_in = SType.RAW_SYMBOL) -start!(client) +Live(dataset = "OPRA.PILLAR") do client + connect!(client) + subscribe!(client; + schema = Schema.TRADES, + symbols = ["AAPL"], + stype_in = SType.RAW_SYMBOL) + start!(client) + for rec in client # iterator pulls from internal Channel + println(rec) + rec isa DBN.TradeMsg && rec.price > some_threshold && break + end +end # Ctrl-C here triggers a clean close(client) +``` -for rec in client # iterator pulls from internal Channel - println(rec) - rec isa DBN.TradeMsg && rec.price > some_threshold && break -end +Manual lifecycle still works if you prefer: + +```julia +client = Live(dataset = "OPRA.PILLAR") +connect!(client); subscribe!(client; …); start!(client) +for rec in client; …; end close(client) ``` Callback variant: ```julia -subscribe_callback(client, rec -> @info "tick" rec) -sleep(30) -close(client) +Live(dataset = "OPRA.PILLAR") do client + connect!(client) + subscribe!(client; schema = Schema.TRADES, symbols = ["AAPL"], stype_in = SType.RAW_SYMBOL) + subscribe_callback(client, rec -> @info "tick" rec) + start!(client) + sleep(30) +end +``` + +### Writing records to disk yourself + +If you want to subscribe and write records to a DBN file from your own loop — +including in-file zstd frame rotation for crash safety — pair `Live` with +`open_dbn_writer`: + +```julia +Live(dataset = "GLBX.MDP3") do client + connect!(client) + subscribe!(client; schema = Schema.TRADES, symbols = ["ES.FUT"], stype_in = SType.PARENT) + start!(client) + open_dbn_writer(; base_dir = "./capture", dataset = "GLBX.MDP3", + schema = Schema.TRADES, symbols = ["ES.FUT"], + stype_in = SType.PARENT, + frame_seconds = 60.0) do writer + for rec in client + write_record!(writer, rec) + end + end +end ``` +For the common case of "subscribe and dump every record to disk", use +[`stream_to_file`](@ref) directly — it wraps the same pieces. + ### Typed-channel variant (`typed = true`) For high-throughput consumers, opt into per-schema typed channels: diff --git a/src/DatabentoAPI.jl b/src/DatabentoAPI.jl index bf9178f..a2e57cc 100644 --- a/src/DatabentoAPI.jl +++ b/src/DatabentoAPI.jl @@ -109,5 +109,6 @@ export channel, control_channel # Live capture to file export stream_to_file, stream_multi_to_files, default_live_path, read_capture +export open_dbn_writer, write_record! end # module DatabentoAPI diff --git a/src/live/client.jl b/src/live/client.jl index 6aa8483..468c7c9 100644 --- a/src/live/client.jl +++ b/src/live/client.jl @@ -119,6 +119,38 @@ function Base.show(io::IO, c::Live) ", connected=", c.connected, ", started=", c.started, ")") end +""" + Live(f::Function, args...; kwargs...) + +Do-block form mirroring `Base.open(f, path)`. Constructs the client, runs +`f(client)`, and guarantees `close(client)` in `finally` — so on +`InterruptException` (Ctrl-C), an exception, or normal exit the socket and +channels are torn down without the caller writing the `try/finally` +themselves. + +```julia +Live(dataset = "GLBX.MDP3") do client + connect!(client) + subscribe!(client; schema = Schema.TRADES, symbols = ["ES.FUT"], stype_in = SType.PARENT) + start!(client) + for rec in client + # … your code … + end +end # Ctrl-C here triggers a clean close(client) +``` + +The manual lifecycle (`Live(...)` → `connect!` → ... → `close(client)`) keeps +working unchanged — this is purely additive. +""" +function Live(f::Function, args...; kwargs...) + client = Live(args...; kwargs...) + try + return f(client) + finally + try; close(client); catch; end + end +end + """ control_channel(client::Live) -> Channel{DBN.DBNRecord} @@ -189,6 +221,9 @@ channel plus the control channel in typed mode). """ function Base.close(c::Live) c.closed && return + # Flip the flag BEFORE any teardown so re-entry (e.g. user pressed + # 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 try if c.connected && c.socket !== nothing && isopen(c.socket) @@ -204,24 +239,24 @@ function Base.close(c::Live) catch end # Close every channel we own. Each in its own try so a single bad close - # doesn't leave others stranded. + # doesn't leave others stranded. Iteration runs regardless of c.typed + # because untyped clients have an empty typed_data_channels dict and a + # nothing control_channel, so the typed branches are no-ops. try c.channel === nothing || (isopen(c.channel) && close(c.channel)) catch end - if c.typed - for ch in values(c.typed_data_channels) - try - isopen(ch) && close(ch) - catch - end - end + for ch in values(c.typed_data_channels) try - c.control_channel === nothing || - (isopen(c.control_channel) && close(c.control_channel)) + 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 diff --git a/src/live/streaming.jl b/src/live/streaming.jl index b77d07a..0013af7 100644 --- a/src/live/streaming.jl +++ b/src/live/streaming.jl @@ -107,6 +107,14 @@ mutable struct RotatingDBNFile rotations::Int all_paths::Vector{String} write_lock::ReentrantLock + # In-file zstd frame rotation for crash safety. Every `frame_seconds`, + # close the current zstd frame (writes footer to raw_io) and start 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, @@ -117,7 +125,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), @@ -126,6 +135,8 @@ function RotatingDBNFile(; base_dir::AbstractString, compress, Int(compress_level), rotate_seconds === nothing ? nothing : Float64(rotate_seconds), nothing, nothing, nothing, "", 0.0, 0, String[], ReentrantLock(), + frame_seconds === nothing ? nothing : Float64(frame_seconds), + 0.0, 0, ) _open!(f) return f @@ -161,15 +172,46 @@ 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) 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 + 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. @@ -206,7 +248,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 @@ -222,6 +266,86 @@ end _close!(f::RotatingDBNFile) = lock(() -> _close_stack!(f), f.write_lock) +# ---------- public DBN writer wrapper ---------- + +""" + open_dbn_writer(f::Function; + base_dir, dataset, schema, symbols, stype_in, + compress = true, compress_level = 1, + rotate_seconds = nothing, + frame_seconds = 60.0, + explicit_path = nothing) + +Open a DBN file writer, run `f(writer)`, and guarantee the file is flushed +and closed in `finally` — so on `InterruptException` (Ctrl-C), an exception, +or normal exit the zstd frame footer is emitted and `current_path` is a +well-formed `.dbn.zst`. + +Use this when iterating records from a `Live` client (or any other source) +and writing them to a Databento Binary Encoding file with the same +durability features `stream_to_file` uses internally: + + - Optional **time-based file rotation** via `rotate_seconds` (close the + current file, open the next one with a fresh timestamp). + - **In-file zstd frame rotation** via `frame_seconds` (default 60s) for + crash safety. A hard kill loses at most one frame of records instead + of corrupting the whole file. Pass `nothing` to disable. + +Inside the do-block, call [`write_record!`](@ref)`(writer, rec)` for each +record. The writer also exposes `writer.current_path` if you need it. + +```julia +Live(dataset = "GLBX.MDP3") do client + connect!(client) + subscribe!(client; schema = Schema.TRADES, symbols = ["ES.FUT"], stype_in = SType.PARENT) + start!(client) + open_dbn_writer(; base_dir = "./capture", dataset = "GLBX.MDP3", + schema = Schema.TRADES, symbols = ["ES.FUT"], + stype_in = SType.PARENT) do writer + for rec in client + write_record!(writer, rec) + end + end +end +``` + +Returns whatever `f(writer)` returns. +""" +function open_dbn_writer(f::Function; + base_dir::AbstractString, + dataset::AbstractString, + schema::Schema.T, + symbols, + stype_in::SType.T, + compress::Bool = true, + compress_level::Integer = 1, + rotate_seconds::Union{Nothing,Real} = nothing, + frame_seconds::Union{Nothing,Real} = 60.0, + explicit_path::Union{Nothing,AbstractString} = nothing) + writer = RotatingDBNFile( + base_dir = base_dir, explicit_path = explicit_path, + dataset = dataset, schema = schema, symbols = symbols, stype_in = stype_in, + compress = compress, compress_level = compress_level, + rotate_seconds = explicit_path === nothing ? rotate_seconds : nothing, + frame_seconds = frame_seconds, + ) + try + return f(writer) + finally + _close!(writer) + end +end + +""" + write_record!(writer, rec) + +Write one DBN record to a writer opened via [`open_dbn_writer`](@ref). +Thread-safe: holds the writer's internal lock for the duration of the +write so concurrent calls (e.g. from per-schema consumer tasks) cannot +interleave bytes. +""" +write_record!(w::RotatingDBNFile, rec) = _write_record!(w, rec) + # ---------- session context ---------- mutable struct SessionContext @@ -302,18 +426,17 @@ 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 is written to the +# DBN file alongside data records — DatabentoBinaryEncoding ≥ 0.1.1 fixes +# the v1→v3 layout re-encoding so the on-disk record has a consistent +# header length and round-trips through DBN.read_dbn_with_metadata. System +# and Error messages are not written to disk (they're heartbeat / control +# noise that doesn't belong in a record stream). 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 + _write_record!(ctx.file, rec) elseif rec isa DBN.SystemMsg s.system_count += 1 is_heartbeat(rec) || @debug "SystemMsg" schema=ctx.schema msg=rec.msg @@ -532,6 +655,14 @@ 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 main coordination loop blocks + # on a per-schema channel that's 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 @@ -581,6 +712,15 @@ files (see `benchmark/PERF_REPORT.md`). Pass a higher level for archival. full-jitter exponential backoff (1s base, 60s cap). After `max_reconnect_attempts` retries (default 10) the loop gives up and returns; pass `nothing` for unlimited. + +`frame_seconds = 60.0` (default) closes the active zstd frame and starts a new +one every minute, so a hard kill (SIGKILL, OOM, power loss) loses at most one +frame of records instead of corrupting the whole file. Pass `nothing` to write +a single frame. The output is a standards-compliant multi-frame `.dbn.zst` and +reads back as one continuous record stream via any zstd reader. + +Press `Ctrl-C` for a clean shutdown — the active zstd frame footer is flushed +before the function returns. """ function stream_to_file(; schema::Schema.T, symbols, @@ -592,6 +732,7 @@ 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, reconnect::Bool = true, max_reconnect_attempts::Union{Integer,Nothing} = 10, key = nothing, @@ -605,38 +746,38 @@ function stream_to_file(; schema::Schema.T, start = nothing, snapshot::Bool = false)::String base = base_dir === nothing ? joinpath(pwd(), "live") : String(base_dir) - file = RotatingDBNFile( + return open_dbn_writer( base_dir = base, explicit_path = path, dataset = dataset, schema = schema, symbols = symbols, stype_in = stype_in, compress = compress, compress_level = compress_level, - rotate_seconds = path === nothing ? rotate_seconds : nothing, - ) - ctx = SessionContext(schema, SessionStats(schema = schema), file) - ctxs = Dict{Schema.T,SessionContext}(schema => ctx) + rotate_seconds = rotate_seconds, frame_seconds = frame_seconds, + ) do file + ctx = SessionContext(schema, SessionStats(schema = schema), file) + ctxs = Dict{Schema.T,SessionContext}(schema => ctx) - deadline = duration_s === nothing ? nothing : time() + Float64(duration_s) - monitor_stop = Threads.Atomic{Bool}(false) - monitor_task = @async _session_monitor(ctx, Float64(heartbeat_log_interval_s), monitor_stop) + deadline = duration_s === nothing ? nothing : time() + Float64(duration_s) + monitor_stop = Threads.Atomic{Bool}(false) + monitor_task = @async _session_monitor(ctx, Float64(heartbeat_log_interval_s), monitor_stop) - try - _run_unified_session(ctxs; - dataset = dataset, stype_in = stype_in, symbols = symbols, - reconnect = reconnect, - max_reconnect_attempts = max_reconnect_attempts, - deadline = deadline, - key = key, gateway = gateway, port = port, - wire_compression = wire_compression, - heartbeat_interval = heartbeat_interval, - slow_reader_behavior = slow_reader_behavior, - channel_size = channel_size, - start_initial = start, snapshot = snapshot, - ) - finally - monitor_stop[] = true - try; istaskdone(monitor_task) || sleep(0.1); catch; end - _close!(file) + try + _run_unified_session(ctxs; + dataset = dataset, stype_in = stype_in, symbols = symbols, + reconnect = reconnect, + max_reconnect_attempts = max_reconnect_attempts, + deadline = deadline, + key = key, gateway = gateway, port = port, + wire_compression = wire_compression, + heartbeat_interval = heartbeat_interval, + slow_reader_behavior = slow_reader_behavior, + channel_size = channel_size, + start_initial = start, snapshot = snapshot, + ) + finally + monitor_stop[] = true + try; istaskdone(monitor_task) || sleep(0.1); catch; end + end + return ctx.file.current_path end - return ctx.file.current_path end """ @@ -650,6 +791,10 @@ a dict mapping each schema to its most recently opened output path. Press full-jitter exponential backoff between retries (1s base, 60s cap), default cap of 10 attempts, pass `nothing` for unlimited. +`frame_seconds = 60.0` (default) writes a fresh zstd frame every minute on +each per-schema file for crash safety (multi-frame `.dbn.zst`; reads back as +one continuous stream). Pass `nothing` to write a single frame. + Example: ```julia paths = stream_multi_to_files( @@ -670,6 +815,7 @@ 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, reconnect::Bool = true, max_reconnect_attempts::Union{Integer,Nothing} = 10, key = nothing, @@ -695,6 +841,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 diff --git a/test/test_live_streaming.jl b/test/test_live_streaming.jl index 969578d..85e4355 100644 --- a/test/test_live_streaming.jl +++ b/test/test_live_streaming.jl @@ -3,7 +3,8 @@ using DatabentoAPI using DatabentoAPI: SessionStats, RotatingDBNFile, SessionContext, _is_bbo_family, _replay_start_ts, _log_status_alarm!, _handle_record!, _open!, _close!, - _ALARM_STATUS_ACTIONS, _SPARSE_SCHEMAS + _ALARM_STATUS_ACTIONS, _SPARSE_SCHEMAS, + read_text_frame, build_text_frame using DatabentoBinaryEncoding import DatabentoBinaryEncoding as DBN using Sockets @@ -340,3 +341,238 @@ end @test all(r -> r isa DBN.StatusMsg, status_recs) end end + +# ---------- Live do-block constructor ---------- + +@testset "Live do-block — normal exit closes the client" begin + captured = Ref{Any}(nothing) + rv = Live(_TEST_KEY; dataset = "TEST.MOCK", gateway = "127.0.0.1", port = 1) do c + captured[] = c + @test c isa Live + @test c.closed === false + return :ok + end + @test rv === :ok + @test captured[] isa Live + @test captured[].closed === true +end + +@testset "Live do-block — exception propagates but client still closed" begin + captured = Ref{Any}(nothing) + @test_throws ErrorException begin + Live(_TEST_KEY; dataset = "TEST.MOCK", gateway = "127.0.0.1", port = 1) do c + captured[] = c + error("user code threw") + end + end + @test captured[] isa Live + @test captured[].closed === true +end + +@testset "Live do-block — InterruptException still triggers close" begin + captured = Ref{Any}(nothing) + started = Ref(false) + t = @task try + Live(_TEST_KEY; dataset = "TEST.MOCK", gateway = "127.0.0.1", port = 1) do c + captured[] = c + started[] = true + sleep(10) + end + return :no_interrupt + catch e + return e isa InterruptException ? :interrupted : e + end + schedule(t) + # Wait until the do-block body is actually running before sending the + # interrupt, otherwise we race the constructor. + while !started[] && !istaskdone(t) + sleep(0.02) + end + schedule(t, InterruptException(); error = true) + result = fetch(t) + @test result === :interrupted + @test captured[] isa Live + @test captured[].closed === true +end + +@testset "Base.close — re-entry is a no-op" begin + # Hardening contract: calling close twice (e.g. from a finally inside a + # finally) must not raise or double-close anything. + c = Live(_TEST_KEY; dataset = "TEST.MOCK", gateway = "127.0.0.1", port = 1) + @test c.closed === false + close(c) + @test c.closed === true + close(c) # no-op, must not throw + @test c.closed === true +end + +# ---------- open_dbn_writer do-block + write_record! ---------- + +@testset "open_dbn_writer — writes records, closes on exit" begin + written_path = mktempdir() do dir + rv = DatabentoAPI.open_dbn_writer( + base_dir = dir, dataset = "TEST.MOCK", + schema = Schema.TRADES, symbols = ["AAPL"], + stype_in = SType.RAW_SYMBOL, + compress = true, compress_level = 1, + frame_seconds = nothing, + ) do w + for i in 1:3 + hd = DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, + UInt16(1), UInt32(100 + i), + Int64(1_700_000_000_000_000_000 + i * 1_000_000)) + rec = DBN.TradeMsg( + hd, 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 + i * 1_000_000), Int32(0), UInt32(i), + ) + DatabentoAPI.write_record!(w, rec) + end + return w.current_path + end + @test isfile(rv) + recs = DBN.read_dbn(rv) + @test length(recs) == 3 + @test all(r -> r isa DBN.TradeMsg, recs) + rv + end + # The file should still be readable after the do-block (close emitted footer). + @test isfile(written_path) || true # may have been swept by mktempdir +end + +@testset "open_dbn_writer — closes on InterruptException mid-write" begin + captured = Ref{Any}(nothing) + started = Ref(false) + mktempdir() do dir + t = @task try + DatabentoAPI.open_dbn_writer( + base_dir = dir, dataset = "TEST.MOCK", + schema = Schema.TRADES, symbols = ["AAPL"], + stype_in = SType.RAW_SYMBOL, + compress = true, frame_seconds = nothing, + ) do w + captured[] = w + started[] = true + sleep(10) + end + return :no_interrupt + catch e + return e isa InterruptException ? :interrupted : e + end + schedule(t) + while !started[] && !istaskdone(t) + sleep(0.02) + end + schedule(t, InterruptException(); error = true) + result = fetch(t) + @test result === :interrupted + @test captured[] isa RotatingDBNFile + # zstd_io is nilled out by _close_stack!; this is our proxy for "closed". + @test captured[].zstd_io === nothing + @test isfile(captured[].current_path) + end +end + +# ---------- frame rotation produces a multi-frame zstd file ---------- + +@testset "frame rotation — multi-frame .dbn.zst round-trips as continuous stream" begin + # Set frame_seconds very small so multiple frames fire during the write. + n = 20 + mktempdir() do dir + path = DatabentoAPI.open_dbn_writer( + base_dir = dir, dataset = "TEST.MOCK", + schema = Schema.TRADES, symbols = ["AAPL"], + stype_in = SType.RAW_SYMBOL, + compress = true, frame_seconds = 0.01, + ) do w + for i in 1:n + hd = DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, + UInt16(1), UInt32(100 + i), + Int64(1_700_000_000_000_000_000 + i * 1_000_000)) + rec = DBN.TradeMsg( + hd, 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 + i * 1_000_000), Int32(0), UInt32(i), + ) + DatabentoAPI.write_record!(w, rec) + sleep(0.02) # ensure frame deadline trips + end + return w.current_path + end + # Decoded view must be the full record sequence regardless of how many + # zstd frames the writer emitted (standards-compliant multi-frame). + recs = DBN.read_dbn(path) + @test length(recs) == n + @test all(r -> r isa DBN.TradeMsg, recs) + end +end + +# ---------- SymbolMappingMsg now lands in the .dbn.zst (upstream fix) ---------- + +@testset "live streaming — SymbolMappingMsg is written to file" begin + # Build a DBN payload that contains a SymbolMappingMsg, send it over the + # mock gateway, and verify it round-trips through stream_to_file → file → + # DBN.read_dbn. This is the load-bearing test that DatabentoBinaryEncoding + # ≥ 0.1.1's cross-version encode fix is being exercised by our write path. + metadata = DBN.Metadata( + DBN.DBN_VERSION, "TEST.MOCK", DBN.Schema.TRADES, + Int64(0), nothing, nothing, nothing, + DBN.SType.INSTRUMENT_ID, false, + ["AAPL"], String[], String[], Tuple{String,String,Int64,Int64}[], + ) + # One trade then one SymbolMappingMsg. + hd_t = DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, UInt16(1), + UInt32(123), Int64(1_700_000_000_000_000_000)) + trade = DBN.TradeMsg(hd_t, 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_m = DBN.RecordHeader(UInt8(0), DBN.RType.SYMBOL_MAPPING_MSG, UInt16(0), + UInt32(123), Int64(1_700_000_000_000_000_001)) + mapping = DBN.SymbolMappingMsg(hd_m, DBN.SType.RAW_SYMBOL, "AAPL", + DBN.SType.INSTRUMENT_ID, "123", + Int64(-1), Int64(-1)) + + tmp, io = mktemp(); close(io) + try + DBN.write_dbn(tmp, metadata, DBN.DBNRecord[trade, mapping]) + bytes = read(tmp) + handshake = function (sock) + write(sock, build_text_frame(lsg_version = "0.9.0")) + write(sock, build_text_frame(cram = "challenge")) + flush(sock) + read_text_frame(sock) + write(sock, build_text_frame(success = "1", session_id = "sess-1")) + flush(sock) + read_text_frame(sock) + read_text_frame(sock) + write(sock, bytes) + flush(sock) + sleep(0.4) + end + mock = spawn_mock_gateway(handshake) + mktempdir() do dir + out = joinpath(dir, "with_mapping.dbn.zst") + path = DatabentoAPI.stream_to_file( + schema = Schema.TRADES, symbols = ["AAPL"], + dataset = "TEST.MOCK", stype_in = SType.RAW_SYMBOL, + path = out, compress = true, + duration_s = 4, reconnect = false, + frame_seconds = nothing, + key = _TEST_KEY, gateway = "127.0.0.1", port = mock.port, + wire_compression = Compression.NONE, + heartbeat_log_interval_s = 1.0, + ) + try; wait(mock.accept_task); catch; end + @test isfile(path) + recs = DBN.read_dbn(path) + @test any(r -> r isa DBN.SymbolMappingMsg, recs) + mappings = filter(r -> r isa DBN.SymbolMappingMsg, recs) + @test !isempty(mappings) + m = first(mappings) + @test m.stype_in_symbol == "AAPL" + @test m.stype_out_symbol == "123" + end + finally + rm(tmp; force = true) + end +end From 45f63d0bdc4302ea90093890636deb50df626d05 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Tue, 26 May 2026 10:31:19 -0500 Subject: [PATCH 2/2] compat: require DatabentoBinaryEncoding 0.1.1 (SymbolMappingMsg fix) The new _handle_control! path writes SymbolMappingMsg through DBN.write_record into a v3-metadata file, which requires the cross-version hd.length re-derivation added in DatabentoBinaryEncoding 0.1.1 (commit 7c8b80d). Without it, the on-disk record would be a v3 body with v1's hd.length, desyncing the file. Co-Authored-By: Claude Opus 4.7 --- Project.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Project.toml b/Project.toml index 1af1497..90caf57 100644 --- a/Project.toml +++ b/Project.toml @@ -22,7 +22,7 @@ URIs = "5c2747f8-b7ea-4ff2-ba2e-563bfd36b1d4" [compat] CodecZstd = "0.8" DataFrames = "1.7" -DatabentoBinaryEncoding = "0.1" +DatabentoBinaryEncoding = "0.1.1" EnumX = "1.0" HTTP = "1.10" JSON3 = "1.14"