Tolerant typed decode: skip interleaved non-schema records (#30)

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`.

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

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)

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