diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f520ad..573a450 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **Chunked + concurrent `get_range`** (#33). New `chunk` kwarg (a + `Dates.Period`, e.g. `Year(1)`) splits `[start_dt, end_dt)` into half-open + chunks fetched concurrently under a semaphore (`concurrency` kwarg, default + 8), hiding the ~25–30s fixed per-request server-side assembly latency that + made long continuous-symbol pulls take hours sequentially. Records are + concatenated in time order; chunks that fail after retries are warned about + and recorded in the new `DBNStore.failed_ranges` field for targeted retry + (the call throws only if every chunk failed). `chunk` requires an explicit + `end_dt` and calendar endpoints, and is incompatible with `limit`. + `DBNStore` gained the `failed_ranges` field (always empty outside chunked + mode); the 2-argument constructors are unchanged. - **`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 diff --git a/docs/src/guide/historical.md b/docs/src/guide/historical.md index 8000f29..33686ab 100644 --- a/docs/src/guide/historical.md +++ b/docs/src/guide/historical.md @@ -46,6 +46,44 @@ metadata header with a vector of decoded records. Convert it with [`to_dataframe`](@ref), [`to_csv`](@ref), [`to_json`](@ref), [`to_parquet`](@ref), or write it back to disk with [`to_file`](@ref). +## Chunked, concurrent long-range fetches + +Each `get_range` request carries a fixed server-side assembly latency — for +continuous-symbol queries, ~25–30s regardless of payload size. A multi-year +pull as one request risks the read timeout; as sequential per-year requests it +accumulates the fixed latency a hundred times over. Pass `chunk` to split the +range and fetch chunks concurrently (up to `concurrency`, default 8): + +```julia +store = get_range(client; + dataset = "GLBX.MDP3", + schema = Schema.STATISTICS, + symbols = ["ES.n.0", "ZT.n.0", "ZF.n.0", "ZN.n.0"], + stype_in = SType.CONTINUOUS, + start_dt = Date(2010, 1, 1), + end_dt = Date(2025, 1, 1), + chunk = Year(1)) +``` + +Records come back concatenated in time order. Chunks that fail after the +client's usual retries are warned about and recorded in `store.failed_ranges` +rather than sinking the whole call (it only throws if every chunk failed), so +a retry loop is one line per range: + +```julia +for (s, e) in store.failed_ranges + retry_store = get_range(client; dataset = "GLBX.MDP3", schema = Schema.STATISTICS, + symbols = ["ES.n.0"], stype_in = SType.CONTINUOUS, + start_dt = s, end_dt = e) + append!(store.records, retry_store.records) +end +``` + +`chunk` requires an explicit `end_dt` and calendar endpoints +(`DateTime`/`Date`/parseable string), and is incompatible with `limit`. For +multi-GB pulls, [`submit_job`](@ref) → [`batch_download`](@ref) remains the +better tool. + ## Streaming records: `foreach_record` For larger queries (millions of records), don't materialize the whole vector. diff --git a/docs/src/performance.md b/docs/src/performance.md index 845a0d7..df126f1 100644 --- a/docs/src/performance.md +++ b/docs/src/performance.md @@ -54,6 +54,9 @@ bookkeeping regardless of element type — so the win is in dispatch, not GC. - **Historical, small query (≤ 1 GB):** [`get_range`](@ref) → [`DBNStore`](@ref) → conversion helpers. Easy and ergonomic. +- **Historical, long range / many fixed-latency requests:** [`get_range`](@ref) + with `chunk = Year(1)` (or similar) — chunks fetch concurrently, hiding the + ~25–30s per-request server-side assembly latency. - **Historical, large query:** [`foreach_record`](@ref) with a typed callback, or [`submit_job`](@ref) + [`batch_download`](@ref) and decode the resulting files with `DBN.foreach_record` directly. diff --git a/src/historical/timeseries.jl b/src/historical/timeseries.jl index 1d17d6e..a0e7f51 100644 --- a/src/historical/timeseries.jl +++ b/src/historical/timeseries.jl @@ -1,7 +1,8 @@ """ get_range(client; dataset, schema, symbols, start_dt, end_dt=nothing, stype_in=SType.RAW_SYMBOL, stype_out=SType.INSTRUMENT_ID, - limit=nothing, typed=true, size_hint=nothing) + limit=nothing, typed=true, size_hint=nothing, + chunk=nothing, concurrency=8) Fetch a time range of records from the Databento Historical API. Returns a [`DBNStore`](@ref). @@ -31,6 +32,33 @@ exact hint avoids that waste. 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. + +# Chunked, concurrent fetching + +Every `get_range` request carries a fixed server-side assembly latency that for +continuous-symbol queries can run ~25–30s regardless of payload size, so a +long range fetched as one request risks the read timeout, and fetched as +sequential chunks accumulates an hour of fixed latency over a hundred calls. +Pass `chunk` (a `Dates.Period`, e.g. `Year(1)` or `Month(1)`) to split +`[start_dt, end_dt)` into half-open chunks fetched concurrently — at most +`concurrency` (default 8) requests in flight at once: + +```julia +store = get_range(client; dataset = "GLBX.MDP3", schema = Schema.STATISTICS, + symbols = ["ES.n.0"], stype_in = SType.CONTINUOUS, + start_dt = Date(2010, 1, 1), end_dt = Date(2025, 1, 1), + chunk = Year(1)) +``` + +Records are concatenated in time order. A chunk that fails after the client's +usual retries is warned about and recorded in the returned store's +`failed_ranges` field (a `Vector{Tuple{DateTime,DateTime}}`) so you can +re-fetch exactly the missing ranges; the call only throws if *every* chunk +failed. Constraints: `chunk` requires an explicit `end_dt`, `start_dt`/`end_dt` +as `DateTime`/`Date`/parseable strings (not unix-ns integers), and is +incompatible with `limit` (a record cap can't be apportioned across chunks); +`size_hint` is ignored in chunked mode. For multi-GB pulls `submit_job` remains +the better tool. """ function get_range(c::Historical; dataset::AbstractString, @@ -42,7 +70,29 @@ function get_range(c::Historical; stype_out::SType.T = SType.INSTRUMENT_ID, limit::Union{Nothing,Integer} = nothing, typed::Bool = true, - size_hint::Union{Nothing,Integer} = nothing) + size_hint::Union{Nothing,Integer} = nothing, + chunk::Union{Nothing,Dates.Period} = nothing, + concurrency::Integer = 8) + chunk === nothing && + return _get_range_single(c; dataset, schema, symbols, start_dt, end_dt, + stype_in, stype_out, limit, typed, size_hint) + return _get_range_chunked(c; dataset, schema, symbols, start_dt, end_dt, + stype_in, stype_out, limit, typed, chunk, concurrency) +end + +# The original single-request fetch: one HTTP GET, buffer the zstd payload, +# decode in memory. Chunked mode calls this once per chunk. +function _get_range_single(c::Historical; + dataset::AbstractString, + schema::Schema.T, + symbols, + start_dt, + end_dt, + stype_in::SType.T, + stype_out::SType.T, + limit::Union{Nothing,Integer}, + typed::Bool, + size_hint::Union{Nothing,Integer}) query = ( dataset = String(dataset), symbols = symbols_str(symbols), @@ -72,6 +122,132 @@ function get_range(c::Historical; end end +# Normalize a chunk endpoint to DateTime. Chunk boundaries are computed with +# Period arithmetic, so the endpoints must be calendar values — unix-ns +# integers are accepted by the single-request path (ts_str passes them +# through) but can't be split without assuming a timezone, so reject them. +_chunk_datetime(d::DateTime) = d +_chunk_datetime(d::Date) = DateTime(d) +_chunk_datetime(s::AbstractString) = + try + DateTime(s) + catch + DateTime(Date(s)) + end +_chunk_datetime(x) = + throw(ArgumentError("chunked get_range requires start_dt/end_dt as DateTime, " * + "Date, or a parseable string; got $(typeof(x))")) + +# Merge per-chunk response metadata into one header for the combined store. +# Identity fields (version/dataset/schema/stypes/ts_out) are taken from the +# first success; the time range spans all chunks; the symbol-resolution +# vectors are unioned because they legitimately differ per date-chunk under +# continuous/parent symbology (a contract that rolls mid-range maps +# differently in different chunks). +function _merge_chunk_metadata(mds::Vector{DBN.Metadata}) + md1 = first(mds) + end_ts_vals = Int64[md.end_ts for md in mds if md.end_ts !== nothing] + return DBN.Metadata( + md1.version, md1.dataset, md1.schema, + minimum(md.start_ts for md in mds), + isempty(end_ts_vals) ? nothing : maximum(end_ts_vals), + nothing, # a per-chunk limit would be meaningless for the union + md1.stype_in, md1.stype_out, md1.ts_out, + unique(reduce(vcat, (md.symbols for md in mds))), + unique(reduce(vcat, (md.partial for md in mds))), + unique(reduce(vcat, (md.not_found for md in mds))), + unique(reduce(vcat, (md.mappings for md in mds))), + ) +end + +function _get_range_chunked(c::Historical; + dataset::AbstractString, + schema::Schema.T, + symbols, + start_dt, + end_dt, + stype_in::SType.T, + stype_out::SType.T, + limit::Union{Nothing,Integer}, + typed::Bool, + chunk::Dates.Period, + concurrency::Integer) + limit === nothing || + throw(ArgumentError("limit is not supported with chunk: a record cap cannot " * + "be apportioned across chunk requests")) + end_dt === nothing && + throw(ArgumentError("chunked get_range requires an explicit end_dt")) + concurrency >= 1 || throw(ArgumentError("concurrency must be ≥ 1")) + + start = _chunk_datetime(start_dt) + stop = _chunk_datetime(end_dt) + start < stop || throw(ArgumentError("start_dt must be before end_dt")) + + # Half-open [t, min(t+chunk, stop)) chunks compose exactly: Databento + # ranges are themselves [start, end), so no record is duplicated or lost + # at the seams and the concatenation needs no dedup. Positivity is checked + # at *every* boundary, not just the first: calendar arithmetic means a + # period's effective length varies by date, so a single upfront check + # cannot rule out a later non-advancing step (which would otherwise append + # zero-length chunks forever). + chunks = Tuple{DateTime,DateTime}[] + let t = start + while t < stop + nxt = min(t + chunk, stop) + nxt > t || throw(ArgumentError( + "chunk must advance time at every boundary; $chunk does not advance from $t")) + push!(chunks, (t, nxt)) + t = nxt + end + end + + # Fetch concurrently, gated by a semaphore. Each task writes only its own + # index slot, so no lock is needed and chunk order (= time order) is + # preserved for assembly. HTTPClient is safe to share across tasks, and + # its built-in 429 retry-with-backoff absorbs rate-limit brushes from the + # concurrency. Threads.@spawn (vs @async) lets the CPU-bound zstd+decode + # work parallelize when threads are available; on a single-threaded + # runtime it still overlaps the network waits. + sem = Base.Semaphore(concurrency) + results = Vector{Any}(nothing, length(chunks)) + @sync for (i, (cs, ce)) in pairs(chunks) + Threads.@spawn Base.acquire(sem) do + results[i] = try + _get_range_single(c; dataset, schema, symbols, + start_dt = cs, end_dt = ce, + stype_in, stype_out, + limit = nothing, typed, size_hint = nothing) + catch e + e isa InterruptException && rethrow() + hint = e isa BentoTimeoutError ? " (consider a smaller chunk)" : "" + @warn "get_range chunk failed; continuing with remaining chunks$hint" start_dt = cs end_dt = ce exception = e + e + end + end + end + + success_idx = [i for i in eachindex(results) if results[i] isa DBNStore] + if isempty(success_idx) + # An empty store would mask total failure — surface the first error. + throw(first(e for e in results if e isa Exception)) + end + + R = eltype(results[first(success_idx)].records) + records = Vector{R}() + sizehint!(records, sum(i -> length(results[i].records), success_idx)) + for i in success_idx # index order == chunk order == time order + append!(records, results[i].records) + end + + failed = Tuple{DateTime,DateTime}[chunks[i] for i in eachindex(results) + if !(results[i] isa DBNStore)] + isempty(failed) || + @warn "get_range: $(length(failed)) of $(length(chunks)) chunks failed; their ranges are in the returned store's failed_ranges for retry" + + md = _merge_chunk_metadata(DBN.Metadata[results[i].metadata for i in success_idx]) + return DBNStore{R}(md, records, failed) +end + """ record_type_for_schema(schema) diff --git a/src/store.jl b/src/store.jl index 0ddffef..24e8d0c 100644 --- a/src/store.jl +++ b/src/store.jl @@ -8,13 +8,23 @@ When the schema is type-pure (the common case), `T` is the concrete record struct (e.g. `DBN.TradeMsg`) and the records vector is type-stable. For mixed-record streams or when `get_range` is called with `typed=false`, `T` falls back to the `DBN.DBNRecord` Union. + +`failed_ranges` is only populated by chunked `get_range` calls (the `chunk` +kwarg): each entry is the `(start_dt, end_dt)` of a chunk whose request failed +after retries, so the caller can re-fetch exactly the missing ranges. It is +always empty for single-request fetches. """ struct DBNStore{T} metadata::DBN.Metadata records::Vector{T} + failed_ranges::Vector{Tuple{DateTime,DateTime}} end -DBNStore(metadata, records::Vector{T}) where {T} = DBNStore{T}(metadata, records) +# 2-arg forms preserve the pre-failed_ranges construction API. +DBNStore(metadata, records::Vector{T}) where {T} = + DBNStore{T}(metadata, records, Tuple{DateTime,DateTime}[]) +DBNStore{T}(metadata, records::Vector{T}) where {T} = + DBNStore{T}(metadata, records, Tuple{DateTime,DateTime}[]) Base.length(s::DBNStore) = length(s.records) Base.iterate(s::DBNStore, st...) = iterate(s.records, st...) @@ -27,7 +37,75 @@ Base.getindex(s::DBNStore, i) = getindex(s.records, i) function Base.show(io::IO, s::DBNStore{T}) where {T} print(io, "DBNStore{", T, "}(dataset=", s.metadata.dataset, ", schema=", s.metadata.schema, - ", n=", length(s.records), ")") + ", n=", length(s.records)) + isempty(s.failed_ranges) || print(io, ", failed_chunks=", length(s.failed_ranges)) + print(io, ")") +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 # Typed record loop that tolerates interleaved non-schema records. diff --git a/test/test_historical_timeseries.jl b/test/test_historical_timeseries.jl index cc2a8d2..a29b8c5 100644 --- a/test/test_historical_timeseries.jl +++ b/test/test_historical_timeseries.jl @@ -1,5 +1,6 @@ using Test using DatabentoAPI +using Dates using HTTP using CodecZstd using Logging @@ -323,6 +324,163 @@ end end end +# ---- chunked + concurrent get_range (#33) ---- + +# Task-safe mock dispatcher for chunked mode: captures every query under a +# lock and lets the test key the response (or failure) off the chunk's +# `start` query param. +function _chunk_mock(respond) # respond(start_str) -> HTTP.Response (or throw) + lk = ReentrantLock() + queries = Vector{Dict{String,String}}() + dispatcher = (method, url, headers, body; kwargs...) -> begin + q = Dict(get(kwargs, :query, Pair{String,String}[])) + lock(lk) do + push!(queries, q) + end + respond(q["start"]) + end + return dispatcher, queries +end + +# Payload whose single trade is tagged (via sequence) so tests can tell which +# chunk each record came from. +function _tagged_payload(tag::Integer) + _, compressed = _build_dbn_zstd(DBN.DBNRecord[_trade_rec(tag)]) + return compressed +end + +@testset "chunked get_range splits, fetches, and concatenates in order (#33)" begin + # Tag each chunk's record by the day-of-month of the chunk start. + dispatcher, queries = _chunk_mock(start_str -> + HTTP.Response(200; body = _tagged_payload(Dates.day(DateTime(start_str))))) + c = Historical("test-key"; gateway = "https://hist.test", dispatcher = dispatcher) + store = get_range(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 4), + chunk = Day(1)) + @test length(queries) == 3 + starts = sort([q["start"] for q in queries]) + @test starts == ["2024-01-01T00:00:00", "2024-01-02T00:00:00", "2024-01-03T00:00:00"] + ends = sort([q["end"] for q in queries]) + @test ends == ["2024-01-02T00:00:00", "2024-01-03T00:00:00", "2024-01-04T00:00:00"] + # Records concatenated in chunk (= time) order regardless of completion order. + @test store isa DBNStore{DBN.TradeMsg} + @test [Int(r.sequence) for r in store] == [1, 2, 3] + @test isempty(store.failed_ranges) + + # Ragged final chunk is clamped to end_dt. + dispatcher2, queries2 = _chunk_mock(start_str -> + HTTP.Response(200; body = _tagged_payload(1))) + c2 = Historical("test-key"; gateway = "https://hist.test", dispatcher = dispatcher2) + get_range(c2; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 3, 12), + chunk = Day(1)) + @test length(queries2) == 3 + @test maximum(q["end"] for q in queries2) == "2024-01-03T12:00:00" +end + +@testset "chunked get_range warns and records failed chunks (#33)" begin + # Middle chunk (Jan 2) 404s; the others succeed. + dispatcher, _ = _chunk_mock(start_str -> + startswith(start_str, "2024-01-02") ? + HTTP.Response(404; body = """{"detail":{"case":"not_found","message":"nope"}}""") : + HTTP.Response(200; body = _tagged_payload(Dates.day(DateTime(start_str))))) + c = Historical("test-key"; gateway = "https://hist.test", + dispatcher = dispatcher, max_retries = 0) + store = @test_logs (:warn, r"chunk failed") (:warn, r"1 of 3 chunks failed") get_range(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 4), + chunk = Day(1)) + @test [Int(r.sequence) for r in store] == [1, 3] + @test store.failed_ranges == [(DateTime(2024, 1, 2), DateTime(2024, 1, 3))] +end + +@testset "chunked get_range throws when every chunk fails (#33)" begin + dispatcher, _ = _chunk_mock(_ -> + HTTP.Response(404; body = """{"detail":{"case":"not_found","message":"nope"}}""")) + c = Historical("test-key"; gateway = "https://hist.test", + dispatcher = dispatcher, max_retries = 0) + @test_logs (:warn, r"chunk failed") (:warn, r"chunk failed") match_mode=:any begin + @test_throws BentoClientError get_range(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 3), + chunk = Day(1)) + end +end + +@testset "chunked get_range validation (#33)" begin + c = Historical("test-key"; gateway = "https://hist.test", + dispatcher = (args...; kw...) -> error("must not dispatch")) + common = (dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"]) + @test_throws ArgumentError get_range(c; common..., + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 3), + chunk = Day(1), limit = 100) + @test_throws ArgumentError get_range(c; common..., + start_dt = DateTime(2024, 1, 1), chunk = Day(1)) # no end_dt + @test_throws ArgumentError get_range(c; common..., + start_dt = 1_700_000_000_000_000_000, end_dt = DateTime(2024, 1, 3), + chunk = Day(1)) # unix-ns int + @test_throws ArgumentError get_range(c; common..., + start_dt = DateTime(2024, 1, 3), end_dt = DateTime(2024, 1, 1), + chunk = Day(1)) # start ≥ end + @test_throws ArgumentError get_range(c; common..., + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 3), + chunk = Day(1), concurrency = 0) + @test_throws ArgumentError get_range(c; common..., + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 3), + chunk = Day(0)) # non-advancing + @test_throws ArgumentError get_range(c; common..., + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 3), + chunk = Day(-1)) # negative +end + +@testset "chunk=nothing keeps single-request behavior (#33)" begin + bytes, n = _build_sample_dbn_zstd() + calls = Ref(0) + mock = (method, url, headers, body; kwargs...) -> begin + calls[] += 1 + HTTP.Response(200; body = bytes) + end + c = Historical("test-key"; gateway = "https://hist.test", dispatcher = mock) + store = 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 calls[] == 1 + @test length(store) == n + @test isempty(store.failed_ranges) + # 2-arg constructors (pre-failed_ranges API) still work. + s2 = DBNStore(store.metadata, store.records) + @test s2 isa DBNStore{DBN.TradeMsg} && isempty(s2.failed_ranges) + s3 = DBNStore{DBN.TradeMsg}(store.metadata, store.records) + @test isempty(s3.failed_ranges) +end + +@testset "chunked get_range respects the concurrency gate (#33)" begin + in_flight = Threads.Atomic{Int}(0) + max_seen = Threads.Atomic{Int}(0) + payload = _tagged_payload(1) + dispatcher = (method, url, headers, body; kwargs...) -> begin + cur = Threads.atomic_add!(in_flight, 1) + 1 + # Record the high-water mark of simultaneous in-flight requests. + old = max_seen[] + while cur > old + Threads.atomic_cas!(max_seen, old, cur) == old && break + old = max_seen[] + end + sleep(0.05) # hold the slot long enough for overlap to be observable + Threads.atomic_sub!(in_flight, 1) + HTTP.Response(200; body = payload) + end + c = Historical("test-key"; gateway = "https://hist.test", dispatcher = dispatcher) + store = get_range(c; + dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], + start_dt = DateTime(2024, 1, 1), end_dt = DateTime(2024, 1, 5), + chunk = Day(1), concurrency = 2) + @test length(store) == 4 + @test max_seen[] <= 2 +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)