Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
transient statuses (429/5xx) are retried as before (#31).

### Fixed
- **Package precompiles cleanly again.** `src/store.jl` defined the internal
`_foreach_typed_tolerant` helper twice (an identical copy-paste), which made
the module fail precompilation on Julia 1.12+ with "Method overwriting is not
permitted during Module precompilation". `Pkg.test` masked it by falling back
to no-precompile, so tests/CI stayed green while a bare `using DatabentoAPI`
errored. Removed the duplicate definition.
- **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`,
Expand Down
66 changes: 0 additions & 66 deletions src/store.jl
Original file line number Diff line number Diff line change
Expand Up @@ -108,72 +108,6 @@ function _foreach_typed_tolerant(f, decoder::DBN.DBNDecoder, ::Type{T}) where {T
return nothing
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}
Expand Down
Loading