From e29e983fe4857bdca01aaff91cc696492d6e39aa Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Tue, 9 Jun 2026 21:25:40 -0500 Subject: [PATCH 1/5] Docs: fix foreach_record example (no positional record type) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Historical guide showed `foreach_record(client, DBN.TradeMsg; ...)`, a method that does not exist — the call raises a MethodError. The record type is inferred from `schema` (overridable via the `record_type` kwarg). Corrected the example and the surrounding prose. Fixes #32 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 7 +++++++ docs/src/guide/historical.md | 12 +++++++----- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0946ec3..3fc184d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed +- **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`. + The record type is inferred from `schema` and overridable via the + `record_type` keyword (#32). + ## [0.2.0] - 2026-06-04 ### Added diff --git a/docs/src/guide/historical.md b/docs/src/guide/historical.md index da2f33a..dc4bb63 100644 --- a/docs/src/guide/historical.md +++ b/docs/src/guide/historical.md @@ -54,7 +54,7 @@ the hot path allocation-free: ```julia n_trades = Ref(0) -foreach_record(client, DBN.TradeMsg; +foreach_record(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], start_dt = DateTime(2024, 1, 2, 14, 30), @@ -64,10 +64,12 @@ end @show n_trades[] ``` -The first argument after `client` is the concrete record type for the -schema you've requested (`DBN.TradeMsg` for `Schema.TRADES`, -`DBN.MBP1Msg` for `Schema.MBP_1`, etc.). DBN.jl's `record_type_for_dbn_schema` -maps schemas → types. +The concrete record type is inferred from `schema` (`DBN.TradeMsg` for +`Schema.TRADES`, `DBN.MBP1Msg` for `Schema.MBP_1`, etc.), so the callback +receives concrete records on the allocation-free typed path by default. Pass +`record_type = DBN.DBNRecord` to force the generic Union-typed path, or +`record_type = SomeT` to override the inferred type. The function returns the +response's `DBN.Metadata`. ## Metadata: cheap, free, useful From 90a784d0244733a41a648e16f579d061ed148ff9 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Tue, 9 Jun 2026 21:32:01 -0500 Subject: [PATCH 2/5] Raise default read timeout to 600s, map timeouts to BentoTimeoutError Long-range get_range queries spend well over 100s in server-side assembly before the first byte streams, so the old 100s default killed every multi-year pull. Read timeouts are deterministic for a given query shape, so they are no longer retried; HTTP.jl TimeoutError is mapped to a new BentoTimeoutError whose message names the remedy. Fixes #31 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 18 ++++++++++++ docs/src/api/errors.md | 1 + docs/src/troubleshooting.md | 15 ++++++++++ src/DatabentoAPI.jl | 3 +- src/errors.jl | 23 ++++++++++++++++ src/historical/client.jl | 9 +++++- src/historical/timeseries.jl | 7 +++++ src/http.jl | 26 +++++++++++++++--- test/test_historical_timeseries.jl | 15 ++++++++++ test/test_http.jl | 44 +++++++++++++++++++++++++++++- 10 files changed, 154 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3fc184d..e554928 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **`BentoTimeoutError`.** HTTP read timeouts are now mapped to a `BentoError` + subtype whose message names the remedy (raise `Historical(timeout = ...)` or + reduce the range) instead of surfacing HTTP.jl's bare + `TimeoutError: Connection closed after N seconds` (#31). + +### Changed +- **Default HTTP read timeout raised from 100s to 600s.** Long-range + `get_range` queries (e.g. multi-year continuous-symbol pulls) can spend well + over 100s in server-side assembly before the first byte streams, so every + such request used to die on the client's own timeout (#31). Note the + timeout is inactivity-based, so a genuinely hung connection now takes up to + 10 minutes to surface. +- **Read timeouts are no longer retried.** Server-side assembly time is + deterministic for a given query shape, so retrying a timed-out request burns + the same timeout again with zero success probability. Connect errors and + transient statuses (429/5xx) are retried as before (#31). + ### Fixed - **Docs: corrected the `foreach_record` example in the Historical guide.** It showed a nonexistent positional record-type method diff --git a/docs/src/api/errors.md b/docs/src/api/errors.md index 0fdd63d..0bd967e 100644 --- a/docs/src/api/errors.md +++ b/docs/src/api/errors.md @@ -10,4 +10,5 @@ BentoAuthError BentoHttpError BentoClientError BentoServerError +BentoTimeoutError ``` diff --git a/docs/src/troubleshooting.md b/docs/src/troubleshooting.md index 2f8d155..8028a12 100644 --- a/docs/src/troubleshooting.md +++ b/docs/src/troubleshooting.md @@ -28,6 +28,21 @@ support. ## Historical +### `BentoTimeoutError` on long-range `get_range` + +The gateway can spend minutes on server-side assembly (continuous-symbol +resolution, day-partitioned scans) before streaming the first byte — for +multi-year pulls this can exceed the client's read timeout (default 600s). +The fix is a bigger client budget: + +```julia +client = Historical(timeout = 1800) # seconds of read inactivity tolerated +``` + +or a smaller range per request. Read timeouts are not retried: the assembly +time is deterministic for a given query shape, so a retry would burn the same +timeout again. + ### Query too large to materialize If `get_range` exhausts memory or returns extremely slowly, switch to diff --git a/src/DatabentoAPI.jl b/src/DatabentoAPI.jl index f621064..d9b4119 100644 --- a/src/DatabentoAPI.jl +++ b/src/DatabentoAPI.jl @@ -85,7 +85,8 @@ export SplitDuration, Packaging, Delivery, SymbologyResolution, RollRule export SlowReaderBehavior # Errors -export BentoError, BentoAuthError, BentoHttpError, BentoClientError, BentoServerError +export BentoError, BentoAuthError, BentoHttpError, BentoClientError, BentoServerError, + BentoTimeoutError # Auth export load_api_key, default_config_path diff --git a/src/errors.jl b/src/errors.jl index 2931e9f..f3460ed 100644 --- a/src/errors.jl +++ b/src/errors.jl @@ -64,6 +64,29 @@ struct BentoServerError <: BentoHttpError body::String end +""" + BentoTimeoutError(timeout_s) <: BentoError + +Raised when the gateway produces no response bytes for `timeout_s` seconds and +the client gives up (HTTP read/inactivity timeout). Long-range +`timeseries.get_range` queries — e.g. multi-year continuous-symbol pulls — can +spend minutes in server-side assembly before the first byte arrives, so this +usually means the request needs a larger client timeout, not that the server +failed. Construct the client with a bigger budget (`Historical(timeout = ...)`) +or reduce the requested range. + +Read timeouts are **not** retried: for a given query shape the server-side +assembly time is deterministic, so a retry would burn the same timeout again. +""" +struct BentoTimeoutError <: BentoError + timeout_s::Int +end + +Base.showerror(io::IO, e::BentoTimeoutError) = + print(io, "BentoTimeoutError: response exceeded client read timeout of ", + e.timeout_s, "s; construct Historical(timeout = ...) with a larger ", + "value or reduce the requested range") + function _parse_http_error_body(body::AbstractString) case = "" message = String(body) diff --git a/src/historical/client.jl b/src/historical/client.jl index e8d0830..712f91b 100644 --- a/src/historical/client.jl +++ b/src/historical/client.jl @@ -2,13 +2,20 @@ const DEFAULT_HIST_GATEWAY = "https://hist.databento.com" const HIST_API_VERSION = "v0" """ - Historical([key]; gateway=DEFAULT_HIST_GATEWAY, timeout=100, user_agent=...) + Historical([key]; gateway=DEFAULT_HIST_GATEWAY, timeout=600, user_agent=...) Client for the Databento Historical (HTTP) API. `key` falls back to `~/.databento/config.toml` then `ENV["DATABENTO_API_KEY"]` (see [`load_api_key`](@ref)). +`timeout` is the HTTP read (inactivity) timeout in seconds — how long the +client waits without receiving any response bytes before giving up with a +[`BentoTimeoutError`](@ref). Long-range `get_range` queries (e.g. multi-year +continuous-symbol pulls) can spend minutes in server-side assembly before the +first byte streams, so set this generously for such workloads; read timeouts +are deterministic for a given query shape and are therefore not retried. + ```julia client = Historical() store = get_range(client; dataset="XNAS.ITCH", schema=Schema.TRADES, diff --git a/src/historical/timeseries.jl b/src/historical/timeseries.jl index b761b9d..55bca33 100644 --- a/src/historical/timeseries.jl +++ b/src/historical/timeseries.jl @@ -18,6 +18,13 @@ return for callers that rely on the Union element type. `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 exact hint avoids that waste. + +!!! note "Long ranges and the read timeout" + The gateway can spend minutes on server-side assembly (continuous-symbol + resolution, day-partitioned scans) before streaming the first byte. If + that exceeds the client's read timeout (default 600s) the call raises + [`BentoTimeoutError`](@ref) — construct `Historical(timeout = ...)` with a + larger budget or reduce the range. """ function get_range(c::Historical; dataset::AbstractString, diff --git a/src/http.jl b/src/http.jl index b81856b..e8c43e1 100644 --- a/src/http.jl +++ b/src/http.jl @@ -1,5 +1,9 @@ const USER_AGENT = "DatabentoAPI.jl/0.2.0" -const DEFAULT_TIMEOUT = 100 +# Read (inactivity) timeout. Long-range `timeseries.get_range` queries can +# spend minutes in server-side assembly (continuous-symbol resolution + +# day-partitioned scans) before the first byte streams, so this needs to be +# generous — 100s provably kills multi-year pulls (#31). +const DEFAULT_TIMEOUT = 600 const DEFAULT_CONNECT_TIMEOUT = 30 # Retry policy. Transient failures (HTTP 429 / 5xx and connection/timeout @@ -73,9 +77,15 @@ HTTPClient(api_key::AbstractString, base_url::AbstractString; # 501 Not Implemented (a permanent "this server can't do that"). _is_retryable_status(status::Integer) = status == 429 || (500 <= status < 600 && status != 501) -# Transient transport-layer failures from HTTP.jl (connect/timeout/request -# errors). Status errors don't appear here because we pass status_exception=false. -_is_retryable_exception(e) = e isa HTTP.Exceptions.HTTPError +# Transient transport-layer failures from HTTP.jl (connect/request errors). +# Status errors don't appear here because we pass status_exception=false. +# Read timeouts are excluded: for long-range queries the server-side assembly +# time is deterministic, so a request that exceeded the read timeout once will +# exceed it on every retry — retrying just multiplies the wall-clock cost by +# (1 + max_retries) with zero success probability. Connect timeouts surface as +# the distinct `HTTP.ConnectError` and remain retryable. +_is_retryable_exception(e) = + e isa HTTP.Exceptions.HTTPError && !(e isa HTTP.Exceptions.TimeoutError) # Parse a `Retry-After` header. The delta-seconds form is honored (capped); # the rarer HTTP-date form falls back to `nothing` so the caller uses backoff. @@ -191,6 +201,10 @@ function request(c::HTTPClient, method::Symbol, path::AbstractString; _retry_wait(c, attempt, nothing) continue end + # Read timeout: not retryable (see _is_retryable_exception). Map to + # a BentoError naming the actual remedy — HTTP.jl's raw TimeoutError + # gives no hint that it's the *client* giving up. + e isa HTTP.Exceptions.TimeoutError && throw(BentoTimeoutError(c.timeout)) rethrow() end @@ -322,6 +336,10 @@ function open_stream(f, c::HTTPClient, path::AbstractString; _retry_wait(c, attempt, nothing) continue end + # Read timeout → BentoTimeoutError whether or not `f` started: a + # mid-body stall deserves the same actionable message as a + # pre-first-byte one. + e isa HTTP.Exceptions.TimeoutError && throw(BentoTimeoutError(c.timeout)) rethrow() end diff --git a/test/test_historical_timeseries.jl b/test/test_historical_timeseries.jl index dac2089..d56acc2 100644 --- a/test/test_historical_timeseries.jl +++ b/test/test_historical_timeseries.jl @@ -265,6 +265,21 @@ end end end +@testset "foreach_record maps read timeout → BentoTimeoutError" begin + calls = Ref(0) + opener = (consume, c, method, url, headers, qpairs) -> begin + calls[] += 1 + throw(HTTP.Exceptions.TimeoutError(100)) + end + c = Historical("test-key"; gateway = "https://hist.test", + stream_opener = opener, retry_sleep = _NOSLEEP_TS, timeout = 100) + @test_throws BentoTimeoutError 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 + end + @test calls[] == 1 # read timeouts are not retried +end + @testset "foreach_record propagates a consumer exception" begin bytes, n = _build_sample_dbn_zstd() opener = _seq_opener([(200, Pair{String,String}[], bytes)]) diff --git a/test/test_http.jl b/test/test_http.jl index 1156c12..d254512 100644 --- a/test/test_http.jl +++ b/test/test_http.jl @@ -3,7 +3,7 @@ using Base64 using DatabentoAPI using DatabentoAPI: HTTPClient, basic_auth_header, _clean_params, request, get_json, post_json using DatabentoAPI: _is_retryable_status, _retry_after_seconds, _retry_delay, RETRY_CAP_S, - RETRY_AFTER_CAP_S, DEFAULT_MAX_RETRIES + RETRY_AFTER_CAP_S, DEFAULT_MAX_RETRIES, DEFAULT_TIMEOUT using HTTP # No-op sleep so retry tests don't actually wait on the wall clock. @@ -134,6 +134,48 @@ const _NOSLEEP = _ -> nothing @test calls[] == 2 end + @testset "read timeout → BentoTimeoutError, no retry" begin + calls = Ref(0) + function mock(method, url, headers, body; kwargs...) + calls[] += 1 + throw(HTTP.Exceptions.TimeoutError(100)) + end + c = HTTPClient("k", "https://example.test"; + dispatcher = mock, retry_sleep = _NOSLEEP, timeout = 100) + @test_throws BentoTimeoutError request(c, :GET, "/v0/foo") + @test calls[] == 1 # deterministic failure: no retry budget burned + + # The mapped error carries the configured timeout and an actionable hint. + err = try + request(c, :GET, "/v0/foo") + catch e + e + end + @test err isa BentoTimeoutError + @test err.timeout_s == 100 + msg = sprint(showerror, err) + @test occursin("100s", msg) + @test occursin("Historical(timeout = ...)", msg) + end + + @testset "connect errors remain retryable after timeout exclusion" begin + calls = Ref(0) + function mock(method, url, headers, body; kwargs...) + calls[] += 1 + calls[] < 3 && throw(HTTP.Exceptions.ConnectError(url, ErrorException("refused"))) + HTTP.Response(200; body = "{}") + end + c = HTTPClient("k", "https://example.test"; + dispatcher = mock, retry_sleep = _NOSLEEP) + resp = request(c, :GET, "/v0/foo") + @test resp.status == 200 + @test calls[] == 3 + end + + @testset "DEFAULT_TIMEOUT raised for long-range queries" begin + @test DEFAULT_TIMEOUT == 600 + end + @testset "request does not retry a non-transient exception" begin calls = Ref(0) function mock(method, url, headers, body; kwargs...) From 5ff32d146e9420701619cbbfb780c7f6111c2563 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Tue, 9 Jun 2026 21:37:14 -0500 Subject: [PATCH 3/5] Tolerant typed decode: skip interleaved non-schema records Historical responses can legitimately interleave gateway ErrorMsg, SystemMsg, and SymbolMappingMsg records with the data; the typed decode path threw on the first one, forcing callers back to typed=false. The new _foreach_typed_tolerant loop keeps the zero-allocation typed hot path, @warns ErrorMsg contents (they explain missing data), quietly skips other control records, and skips mismatched/unknown rtypes with one summary warning. Fixes #30 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 11 +++ src/historical/timeseries.jl | 15 +++- src/store.jl | 73 ++++++++++++++++- test/test_historical_timeseries.jl | 127 +++++++++++++++++++++++++++++ 4 files changed, 224 insertions(+), 2 deletions(-) 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 d56acc2..a3d05f5 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 From 82ea5f23c06b4c61ec4a73be7340df0631ec4340 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Wed, 10 Jun 2026 07:13:25 -0500 Subject: [PATCH 4/5] Docs: show the DBN alias import before using DBN types The guide only imports DatabentoAPI and Dates, so the record_type override examples referenced an undefined DBN name (review feedback). Co-Authored-By: Claude Fable 5 --- docs/src/guide/historical.md | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/docs/src/guide/historical.md b/docs/src/guide/historical.md index dc4bb63..8000f29 100644 --- a/docs/src/guide/historical.md +++ b/docs/src/guide/historical.md @@ -66,10 +66,18 @@ end The concrete record type is inferred from `schema` (`DBN.TradeMsg` for `Schema.TRADES`, `DBN.MBP1Msg` for `Schema.MBP_1`, etc.), so the callback -receives concrete records on the allocation-free typed path by default. Pass -`record_type = DBN.DBNRecord` to force the generic Union-typed path, or -`record_type = SomeT` to override the inferred type. The function returns the -response's `DBN.Metadata`. +receives concrete records on the allocation-free typed path by default. To +name the record types yourself — e.g. to override the inferred type — bring +the conventional `DBN` alias into scope first: + +```julia +import DatabentoBinaryEncoding as DBN + +foreach_record(client; record_type = DBN.DBNRecord, ...) # generic Union path +foreach_record(client; record_type = DBN.TradeMsg, ...) # explicit override +``` + +The function returns the response's `DBN.Metadata`. ## Metadata: cheap, free, useful From ce6748b5ab3d4f2729cf511875aa2eff79947991 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Wed, 10 Jun 2026 07:16:27 -0500 Subject: [PATCH 5/5] Do not map consumer-raised TimeoutError to BentoTimeoutError If the foreach_record callback itself throws HTTP.Exceptions.TimeoutError (e.g. it makes its own HTTP request), open_stream unconditionally mapped it to BentoTimeoutError, blaming the Databento stream and reporting the wrong timeout value (review feedback). Track the exact exception object that escapes the consumer and map only stream-originated timeouts; a mid-body stall of our own stream still maps correctly because HTTP.jl raises the TimeoutError from the HTTP.open frame, a different object than whatever escaped the consumer. Co-Authored-By: Claude Fable 5 --- src/http.jl | 22 +++++++++++++++++++--- test/test_historical_timeseries.jl | 15 +++++++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/src/http.jl b/src/http.jl index e8c43e1..09f1ba8 100644 --- a/src/http.jl +++ b/src/http.jl @@ -310,6 +310,11 @@ function open_stream(f, c::HTTPClient, path::AbstractString; # records, so a mid-body transport error must NOT be retried (that # would replay the partial stream). This flips true just before `f`. consumed = Ref(false) + # The exact exception object that escaped `f`, if any. Needed to + # distinguish "our stream timed out" from "the consumer's own code + # threw an HTTP TimeoutError" (e.g. `f` makes its own HTTP request): + # only the former should be mapped to BentoTimeoutError below. + f_err = Ref{Any}(nothing) try c.stream_opener(c, "GET", url, headers, qpairs) do status, hdrs, io @@ -326,7 +331,12 @@ function open_stream(f, c::HTTPClient, path::AbstractString; return nothing end consumed[] = true - result_ref[] = f(io) + result_ref[] = try + f(io) + catch fe + f_err[] = fe + rethrow() + end return nothing end catch e @@ -338,8 +348,14 @@ function open_stream(f, c::HTTPClient, path::AbstractString; end # Read timeout → BentoTimeoutError whether or not `f` started: a # mid-body stall deserves the same actionable message as a - # pre-first-byte one. - e isa HTTP.Exceptions.TimeoutError && throw(BentoTimeoutError(c.timeout)) + # pre-first-byte one. (When our stream stalls mid-body, HTTP.jl's + # timeout layer closes the connection — `f`'s read fails with an + # IO error, and the TimeoutError is raised by the HTTP.open frame + # itself, so it is a *different* object than f_err[]. An identical + # object means the consumer's own code threw it: propagate that + # untouched rather than blaming the Databento stream.) + e isa HTTP.Exceptions.TimeoutError && e !== f_err[] && + throw(BentoTimeoutError(c.timeout)) rethrow() end diff --git a/test/test_historical_timeseries.jl b/test/test_historical_timeseries.jl index d56acc2..c933f23 100644 --- a/test/test_historical_timeseries.jl +++ b/test/test_historical_timeseries.jl @@ -280,6 +280,21 @@ end @test calls[] == 1 # read timeouts are not retried end +@testset "consumer-raised HTTP TimeoutError is not blamed on the stream" begin + # If the callback's own code throws HTTP.Exceptions.TimeoutError (e.g. it + # makes its own HTTP request), that must propagate untouched — mapping it + # to BentoTimeoutError would misreport the Databento stream as the culprit. + bytes, _ = _build_sample_dbn_zstd() + opener = _seq_opener([(200, Pair{String,String}[], bytes)]) + c = Historical("test-key"; gateway = "https://hist.test", + stream_opener = opener, retry_sleep = _NOSLEEP_TS, timeout = 100) + @test_throws HTTP.Exceptions.TimeoutError 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 + throw(HTTP.Exceptions.TimeoutError(42)) # consumer's own timeout + end +end + @testset "foreach_record propagates a consumer exception" begin bytes, n = _build_sample_dbn_zstd() opener = _seq_opener([(200, Pair{String,String}[], bytes)])