diff --git a/CHANGELOG.md b/CHANGELOG.md index e554928..4f520ad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 transient statuses (429/5xx) are retried as before (#31). ### Fixed +- **Typed `get_range`/`foreach_record` no longer throw on interleaved + non-schema records** (#30). Historical responses can legitimately carry + gateway `ErrorMsg` (e.g. partial continuous-symbol resolution), `SystemMsg`, + and `SymbolMappingMsg` records among the data; the typed decode path used to + die on the first one (`Expected ... but got rtype=ERROR_MSG`), forcing + `typed=false`. Now `ErrorMsg` records are surfaced via `@warn` (they explain + why data is missing), `SystemMsg`/`SymbolMappingMsg` are skipped quietly, + and any other mismatched or unknown record types are skipped with a single + summary warning. Behavior note: an explicitly wrong `record_type` override + on `foreach_record` now yields zero callbacks plus the summary warning + instead of throwing. - **Docs: corrected the `foreach_record` example in the Historical guide.** It showed a nonexistent positional record-type method (`foreach_record(client, DBN.TradeMsg; ...)`), which raises a `MethodError`. diff --git a/src/historical/timeseries.jl b/src/historical/timeseries.jl index 55bca33..1d17d6e 100644 --- a/src/historical/timeseries.jl +++ b/src/historical/timeseries.jl @@ -14,6 +14,12 @@ concrete record type — roughly 10× faster decode and 60% less allocation than the generic path. Pass `typed=false` to force the legacy `Vector{DBN.DBNRecord}` return for callers that rely on the Union element type. +The typed path tolerates records the gateway legitimately interleaves with the +data: `ErrorMsg` records are logged with `@warn` (their text usually explains +why data is missing), `SystemMsg`/`SymbolMappingMsg` are skipped quietly, and +any other mismatched or unknown record types are skipped with one summary +warning rather than throwing. + `size_hint` lets callers who know the record-count bound (e.g. from a prior `get_record_count` call) pre-size the records vector exactly. The default heuristic over-allocates by 10–20% to avoid realloc churn under growth; an @@ -113,6 +119,13 @@ By default, the concrete record type is inferred from `schema` (e.g. is used (~2× faster than generic dispatch). Pass `record_type = DBN.DBNRecord` to force the generic Union-typed path, or `record_type = SomeT` to override. +Like [`get_range`](@ref), the typed path tolerates interleaved non-schema +records: gateway `ErrorMsg` records are `@warn`-logged, +`SystemMsg`/`SymbolMappingMsg` skipped quietly, and other mismatched rtypes +skipped with a single summary warning — `f` only ever sees the schema's record +type. (Consequently an explicitly wrong `record_type` override yields zero +callbacks plus the summary warning rather than an error.) + Returns the `DBN.Metadata` from the response header. """ function foreach_record(f, c::Historical; @@ -155,7 +168,7 @@ function foreach_record(f, c::Historical; f(rec) end else - DBN._foreach_record_impl(f, decoder, T) + _foreach_typed_tolerant(f, decoder, T) end decoder.metadata end diff --git a/src/store.jl b/src/store.jl index 1e8652e..0ddffef 100644 --- a/src/store.jl +++ b/src/store.jl @@ -30,6 +30,72 @@ function Base.show(io::IO, s::DBNStore{T}) where {T} ", n=", length(s.records), ")") end +# Typed record loop that tolerates interleaved non-schema records. +# +# Historical responses are *mostly* type-pure, but the gateway can legitimately +# interleave control records — ErrorMsg (e.g. partial continuous-symbol +# resolution), SystemMsg, SymbolMappingMsg — with the data (#30). DBN's +# `_foreach_record_impl` throws on the first rtype mismatch, which forced +# callers back to `typed=false` (10× slower) and silently discarded the +# ErrorMsg text explaining *why* data was missing. This loop keeps the +# zero-allocation typed hot path (peek header → read straight into a reused +# Ref{T} buffer) and handles everything else by category: +# +# - ErrorMsg → per-record @warn (rare, and the content is the +# whole point: it explains gaps in the response) +# - SystemMsg / SymbolMappingMsg → decode + @debug (routine noise) +# - other known rtypes → skip by header length, count, one summary @warn +# - unknown raw rtypes → skip by header length, count, one summary @warn +# +# Mismatched data records get a single post-loop summary, never a per-record +# warn — a wrong `record_type` override against a million-record stream must +# not emit a million log lines. The summary (rather than silence) matters for +# data-loss visibility: a skipped data rtype means the caller's type and the +# stream disagree. +# +# Unlike DBN's loop, there is deliberately no `finally` that closes +# `decoder.io` — our decoders wrap HTTP bodies / TranscodingStreams whose +# lifecycle is owned by `open_stream` / the caller, not by this function. +function _foreach_typed_tolerant(f, decoder::DBN.DBNDecoder, ::Type{T}) where {T} + expected = DBN._type_to_rtype_stream(T) + buffer = Ref{T}() # reused per record: the typed read is allocation-free + skipped = 0 + while !eof(decoder.io) + hd_result = DBN.read_record_header(decoder.io) + # Unknown raw rtype byte: read_record_header returns a tuple after + # consuming only the 2 (length, rtype) bytes; length is in 4-byte units. + if hd_result isa Tuple + _, _, len_units = hd_result + skip(decoder.io, Int(len_units) * DBN.LENGTH_MULTIPLIER - 2) + skipped += 1 + continue + end + hd = hd_result + rt = hd.rtype + if rt == expected || + (T === DBN.OHLCVMsg && (rt == DBN.RType.OHLCV_1S_MSG || rt == DBN.RType.OHLCV_1M_MSG || + rt == DBN.RType.OHLCV_1H_MSG || rt == DBN.RType.OHLCV_1D_MSG)) + buffer[] = DBN._read_typed_record_stream(decoder, T, hd) + f(buffer[]) + elseif rt == DBN.RType.ERROR_MSG + rec = DBN.read_error_msg(decoder, hd) + @warn "Databento gateway error record interleaved in response stream" err = rec.err + elseif rt == DBN.RType.SYSTEM_MSG || rt == DBN.RType.SYMBOL_MAPPING_MSG + rec = DBN.read_record_dispatch(decoder, hd, rt) + @debug "Skipping control record in typed decode" rtype = rt + else + # Known data rtype that doesn't match the schema's record type. + # The full 16-byte header is already consumed; hd.length is in + # 4-byte units and covers the whole record. + skip(decoder.io, Int(hd.length) * DBN.LENGTH_MULTIPLIER - 16) + skipped += 1 + end + end + skipped > 0 && + @warn "Typed decode skipped records whose rtype did not match the schema's record type" expected_type = T skipped + return nothing +end + """ decode_dbn_stream(io) -> DBNStore{DBN.DBNRecord} decode_dbn_stream(io, ::Type{T}) -> DBNStore{T} @@ -42,6 +108,11 @@ The single-argument form decodes generically into a `Vector{DBN.DBNRecord}` (slow GC-bound path). The two-argument form decodes directly into a `Vector{T}` via DBN.jl's type-specific reader, which is ~10× faster and has near-zero per-record allocation. + +The typed form tolerates interleaved non-`T` records instead of throwing: +gateway `ErrorMsg` records are logged with `@warn` (their text explains why +data is missing), `SystemMsg`/`SymbolMappingMsg` are skipped quietly, and any +other mismatched or unknown rtypes are skipped with a single summary warning. """ function decode_dbn_stream(io::IO)::DBNStore{DBN.DBNRecord} decoder = DBN.DBNDecoder(io) @@ -68,7 +139,7 @@ function decode_dbn_stream(io::IO, ::Type{T}; elseif md_limit !== nothing && md_limit > 0 sizehint!(records, Int(md_limit)) end - DBN._foreach_record_impl(decoder, T) do rec + _foreach_typed_tolerant(decoder, T) do rec push!(records, rec) end return DBNStore(decoder.metadata, records) diff --git a/test/test_historical_timeseries.jl b/test/test_historical_timeseries.jl index c933f23..cc2a8d2 100644 --- a/test/test_historical_timeseries.jl +++ b/test/test_historical_timeseries.jl @@ -2,6 +2,7 @@ using Test using DatabentoAPI using HTTP using CodecZstd +using Logging using TranscodingStreams using DatabentoBinaryEncoding import DatabentoBinaryEncoding as DBN @@ -52,6 +53,63 @@ function _build_sample_dbn_zstd() end end +# Like _build_sample_dbn_zstd but for an arbitrary record vector, and with +# *correct* hd.length values (the fixture above writes 0, which typed readers +# ignore but the tolerant decode loop relies on to skip mismatched records by +# header length — real gateway streams always carry correct lengths). +function _build_dbn_zstd(records::Vector{DBN.DBNRecord}) + metadata = DBN.Metadata( + DBN.DBN_VERSION, + "XNAS.ITCH", + DBN.Schema.TRADES, + Int64(0), + nothing, nothing, nothing, + DBN.SType.INSTRUMENT_ID, + false, + ["AAPL"], String[], String[], + Tuple{String,String,Int64,Int64}[], + ) + tmp_dbn, tmp_io = mktemp() + close(tmp_io) + try + DBN.write_dbn(tmp_dbn, metadata, records) + raw = read(tmp_dbn) + return raw, transcode(ZstdCompressor, raw) + finally + rm(tmp_dbn; force = true) + end +end + +# TradeMsg is 48 bytes on the wire → hd.length = 12 four-byte units. +function _trade_rec(i) + hd = DBN.RecordHeader( + UInt8(12), DBN.RType.MBP_0_MSG, + UInt16(1), UInt32(100 + i), Int64(1_700_000_000_000_000_000 + i * 1_000_000), + ) + DBN.TradeMsg(hd, + Int64(150_000_000_000 + i * 1_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)) +end + +# Trades with a gateway ErrorMsg and SystemMsg interleaved — the shape that +# made typed decode throw before #30. ErrorMsg payload is padded to +# hd.length*4-16, so hd.length must cover the message text (+ null). +function _build_mixed_dbn_zstd() + err_hd = DBN.RecordHeader(UInt8(16), DBN.RType.ERROR_MSG, UInt16(1), UInt32(0), Int64(0)) + sys_hd = DBN.RecordHeader(UInt8(8), DBN.RType.SYSTEM_MSG, UInt16(1), UInt32(0), Int64(0)) + records = DBN.DBNRecord[ + _trade_rec(1), + _trade_rec(2), + DBN.ErrorMsg(err_hd, "partial symbol resolution"), + _trade_rec(3), + DBN.SystemMsg(sys_hd, "Heartbeat", ""), + _trade_rec(4), + ] + raw, compressed = _build_dbn_zstd(records) + return raw, compressed, 4 # 4 trades +end + @testset "historical get_range" begin bytes, n = _build_sample_dbn_zstd() @@ -265,6 +323,75 @@ end end end +@testset "typed get_range tolerates interleaved control records (#30)" begin + _, compressed, n_trades = _build_mixed_dbn_zstd() + mock = (method, url, headers, body; kwargs...) -> HTTP.Response(200; body = compressed) + c = Historical("test-key"; gateway = "https://hist.test", dispatcher = mock) + store = @test_logs (:warn, r"gateway error record") match_mode=:any get_range(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = "2024-01-02T14:30:00", end_dt = "2024-01-02T14:31:00") + @test store.records isa Vector{DBN.TradeMsg} + @test length(store) == n_trades + @test [r.sequence for r in store] == UInt32[1, 2, 3, 4] +end + +@testset "typed foreach_record tolerates interleaved control records (#30)" begin + _, compressed, n_trades = _build_mixed_dbn_zstd() + opener = _seq_opener([(200, Pair{String,String}[], compressed)]) + c = Historical("test-key"; gateway = "https://hist.test", stream_opener = opener) + seen = Ref(0) + md = @test_logs (:warn, r"gateway error record") match_mode=:any begin + DatabentoAPI.foreach_record(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = "2024-01-02T14:30:00", end_dt = "2024-01-02T14:31:00") do rec + @test rec isa DBN.TradeMsg + seen[] += 1 + end + end + @test seen[] == n_trades + @test md isa DBN.Metadata +end + +@testset "typed decode skips unknown rtypes with a summary warn (#30)" begin + raw, _, n_trades = _build_mixed_dbn_zstd() + # Splice a fake 8-byte record (length = 2 four-byte units, rtype 0xFE) + # onto the end of the uncompressed stream — records are just concatenated. + fake = vcat(UInt8[0x02, 0xFE], zeros(UInt8, 6)) + compressed = transcode(ZstdCompressor, vcat(raw, fake)) + mock = (method, url, headers, body; kwargs...) -> HTTP.Response(200; body = compressed) + c = Historical("test-key"; gateway = "https://hist.test", dispatcher = mock) + store = @test_logs (:warn, r"gateway error record") (:warn, r"skipped records") get_range(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = "2024-01-02T14:30:00", end_dt = "2024-01-02T14:31:00") + @test length(store) == n_trades +end + +@testset "wrong record_type override skips with summary warn instead of throwing (#30)" begin + _, compressed, n_trades = _build_mixed_dbn_zstd() + opener = _seq_opener([(200, Pair{String,String}[], compressed)]) + c = Historical("test-key"; gateway = "https://hist.test", stream_opener = opener) + seen = Ref(0) + @test_logs (:warn, r"gateway error record") (:warn, r"skipped records") begin + DatabentoAPI.foreach_record(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = "2024-01-02T14:30:00", end_dt = "2024-01-02T14:31:00", + record_type = DBN.MBOMsg) do rec + seen[] += 1 + end + end + @test seen[] == 0 # every trade skipped: stream rtype ≠ MBOMsg +end + +@testset "pure typed stream decodes with no warnings (#30 regression)" begin + bytes, n = _build_sample_dbn_zstd() + mock = (method, url, headers, body; kwargs...) -> HTTP.Response(200; body = bytes) + c = Historical("test-key"; gateway = "https://hist.test", dispatcher = mock) + store = @test_logs min_level=Logging.Warn get_range(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = "2024-01-02T14:30:00", end_dt = "2024-01-02T14:31:00") + @test length(store) == n +end + @testset "foreach_record maps read timeout → BentoTimeoutError" begin calls = Ref(0) opener = (consume, c, method, url, headers, qpairs) -> begin