Skip to content

Fix STATISTICS decode corruption and related decoder bugs (#32-#35) #36

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tbeason merged 2 commits into from
Jun 10, 2026
Merged

Fix STATISTICS decode corruption and related decoder bugs (#32-#35) #36

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c53b846
Fix STATISTICS decode corruption and related decoder bugs
tbeason Jun 10, 2026
aa023c9
Address Codex review: harden StatMsg skip path and header normalization
tbeason Jun 10, 2026
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
27 changes: 27 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Fixed

- `read_stat_msg` now sizes the record body from `hd.length` instead of assuming
the 80-byte DBN v3 layout. Pre-v3 64-byte `StatMsg` records (32-bit `quantity`,
as currently served by the historical gateway for STATISTICS) decode correctly
and are upgraded to the v3 shape, instead of overrunning the record boundary
and desyncing every subsequent record in the stream ([#32]).
- `stat_to_dataframe` no longer references nonexistent `StatMsg` fields
(`stat_value`, `flags`); STATISTICS records export with `price`, `quantity`,
`flags` (from `stat_flags`), `ts_ref`, `stat_type`, `channel_id`, and
`update_action` columns ([#33]).
- The metadata decoder reads **all** symbol-mapping intervals instead of only
the first. `Metadata.mappings` now holds one
`(raw_symbol, mapped_symbol, start_date, end_date)` tuple per interval, so a
continuous contract's full roll history survives decoding; the encoder groups
consecutive same-symbol tuples back into one mapping entry ([#34]).
- Invalid enum bytes are handled uniformly: `Side`, `Action`, and
`InstrumentClass` all warn (once) and fall back to a sentinel
(`Side.NONE`, `Action.NONE`, `InstrumentClass.OTHER`) instead of `Side`/
`Action` throwing and killing the decode stream. The conversions use lookup
tables, so the hot path stays free of `try`/`catch`. The old `safe_action`
fallback of `Action.TRADE` was changed to `Action.NONE` ([#35]).

## [0.1.3] - 2026-06-05

### Added
Expand Down Expand Up @@ -67,3 +90,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
[#22]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/pull/22
[#23]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/23
[#24]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/pull/24
[#32]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/32
[#33]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/33
[#34]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/34
[#35]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/35
130 changes: 64 additions & 66 deletions src/decode.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,26 +15,6 @@ Finds the first null byte and creates string from valid data only.
return len == 0 ? "" : String(bytes[1:len])
end

"""
unsafe_action(raw_val::UInt8) -> Action.T

Fast enum conversion without safety checks. 1000x+ faster than safe_action.
Only use when you know the data is well-formed (e.g., reading from trusted DBN files).
"""
@inline function unsafe_action(raw_val::UInt8)
return raw_val == 0x00 ? Action.NONE : Action.T(raw_val)
end

"""
unsafe_side(raw_val::UInt8) -> Side.T

Fast enum conversion without safety checks. 1000x+ faster than safe_side.
Only use when you know the data is well-formed (e.g., reading from trusted DBN files).
"""
@inline function unsafe_side(raw_val::UInt8)
return raw_val == 0x00 ? Side.NONE : Side.T(raw_val)
end

"""
DBNDecoder

Expand Down Expand Up @@ -339,17 +319,17 @@ function read_header!(decoder::DBNDecoder)
# Intervals count
intervals_count = ltoh(reinterpret(UInt32, metadata_bytes[pos:pos+3])[1])
pos += 4
# For now, just read the first interval (simplified)
if intervals_count > 0

# One mappings entry per interval (roll history for continuous symbols)
for _ in 1:intervals_count
# Start date (4 bytes)
start_date_raw = ltoh(reinterpret(UInt32, metadata_bytes[pos:pos+3])[1])
pos += 4

# End date (4 bytes)
end_date_raw = ltoh(reinterpret(UInt32, metadata_bytes[pos:pos+3])[1])
pos += 4

# Mapped symbol
mapped_symbol_bytes = metadata_bytes[pos:pos+symbol_cstr_len-1]
pos += symbol_cstr_len
Expand All @@ -359,13 +339,8 @@ function read_header!(decoder::DBNDecoder)
else
mapped_symbol = String(mapped_symbol_bytes)
end

push!(mappings, (raw_symbol, mapped_symbol, Int64(start_date_raw), Int64(end_date_raw)))

# Skip remaining intervals for now
for _ in 2:intervals_count
pos += 4 + 4 + symbol_cstr_len # start_date + end_date + symbol
end
end
end

Expand Down Expand Up @@ -533,8 +508,8 @@ end
size = read(decoder.io, UInt32) # 4 bytes (positions 32-35)
flags = read(decoder.io, UInt8) # 1 byte (position 36)
channel_id = read(decoder.io, UInt8) # 1 byte (position 37)
action = unsafe_action(read(decoder.io, UInt8)) # 1 byte (position 38)
side = unsafe_side(read(decoder.io, UInt8)) # 1 byte (position 39)
action = safe_action(read(decoder.io, UInt8)) # 1 byte (position 38)
side = safe_side(read(decoder.io, UInt8)) # 1 byte (position 39)
price = read(decoder.io, Int64) # 8 bytes (positions 40-47)
ts_in_delta = read(decoder.io, Int32) # 4 bytes (positions 48-51)
sequence = read(decoder.io, UInt32) # 4 bytes (positions 52-55)
Expand All @@ -545,8 +520,8 @@ end
@inline function read_trade_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand All @@ -558,8 +533,8 @@ end
@inline function read_mbp1_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand All @@ -580,8 +555,8 @@ end
@inline function read_mbp10_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand Down Expand Up @@ -637,7 +612,7 @@ end
market_imbalance_qty = read(decoder.io, UInt32)
unpaired_qty = read(decoder.io, UInt32)
auction_type = read(decoder.io, UInt8)
side = unsafe_side(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
auction_status = read(decoder.io, UInt8)
freeze_status = read(decoder.io, UInt8)
num_extensions = read(decoder.io, UInt8)
Expand All @@ -648,26 +623,44 @@ end
end

@inline function read_stat_msg(decoder::DBNDecoder, hd::RecordHeader)
# StatMsg layout depends on DBN version. Per the public spec:
# v1/v2: ts_recv(8) ts_ref(8) price(8) quantity(4, Int32) sequence(4)
# ts_in_delta(4) stat_type(2) channel_id(2) update_action(1)
# stat_flags(1) reserved(6) = 48-byte body
# v3: same but quantity(8, Int64) and reserved(18) = 64-byte body
body_size = Int(hd.length) * LENGTH_MULTIPLIER - 16
if body_size != 48 && body_size < 64
@warn "Unexpected StatMsg body size $body_size; skipping record" maxlog = 1
skip(decoder.io, body_size)
return nothing
Comment on lines +633 to +635

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Skip malformed statistics records in the typed reader

When a STATISTICS file contains a record whose body is between 49 and 63 bytes, this returns nothing, but read_dbn selects the Vector{StatMsg} typed path for Schema.STATISTICS; that path assigns or pushes the result without checking for nothing, causing a conversion MethodError rather than skipping the record as intended. The typed loops in read_dbn_typed need to recognize this skip result, or this function must signal it in a way those loops handle.

Useful? React with 👍 / 👎.

@tbeason tbeason Jun 10, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in aa023c9 — all four typed call sites (read_dbn_typed exact/dynamic loops, _foreach_record_impl, _foreach_record_with_control_impl) now skip a nothing result and continue, with a regression test decoding a STATISTICS file containing a 56-byte-body record through the typed read_dbn path.

end
ts_recv = read(decoder.io, UInt64)
ts_ref = read(decoder.io, UInt64)
ts_ref = read(decoder.io, UInt64)
price = read(decoder.io, Int64)
# Handle UNDEF values in quantity field - read as UInt64 first
quantity_raw = read(decoder.io, UInt64)
quantity = if quantity_raw == 0xffffffffffffffff
# UNDEF_STAT_QUANTITY - use a special value or convert safely
typemax(Int64)
quantity = if body_size == 48
# v1/v2: 32-bit quantity, UNDEF_STAT_QUANTITY = typemax(Int32)
quantity_raw = read(decoder.io, Int32)
quantity_raw == typemax(Int32) ? typemax(Int64) : Int64(quantity_raw)
else
# Safe conversion for normal values
quantity_raw <= typemax(Int64) ? Int64(quantity_raw) : typemax(Int64)
# v3: 64-bit quantity, UNDEF_STAT_QUANTITY = typemax(Int64)
quantity_raw = read(decoder.io, UInt64)
quantity_raw >= 0x7fffffffffffffff ? typemax(Int64) : Int64(quantity_raw)
end
sequence = read(decoder.io, UInt32)
ts_in_delta = read(decoder.io, Int32)
stat_type = read(decoder.io, UInt16)
channel_id = read(decoder.io, UInt16)
update_action = read(decoder.io, UInt8)
stat_flags = read(decoder.io, UInt8)
_ = read(decoder.io, 18) # Reserved (adjusted for field size changes)
return StatMsg(hd, ts_recv, ts_ref, price, quantity, sequence, ts_in_delta, stat_type, channel_id, update_action, stat_flags)
# Skip reserved tail: whatever the header says is left of the record
skip(decoder.io, body_size - (body_size == 48 ? 42 : 46))
# Normalize the header to the v3 size whenever the on-disk size differed
# (pre-v3 upgrade, or an extended record whose tail we discarded above),
# so re-encoding the 64-byte v3 body stays self-consistent
v3_length = UInt8(80 ÷ LENGTH_MULTIPLIER)
out_hd = hd.length == v3_length ? hd :
RecordHeader(v3_length, hd.rtype, hd.publisher_id, hd.instrument_id, hd.ts_event)
return StatMsg(out_hd, ts_recv, ts_ref, price, quantity, sequence, ts_in_delta, stat_type, channel_id, update_action, stat_flags)
end

@inline function read_error_msg(decoder::DBNDecoder, hd::RecordHeader)
Expand Down Expand Up @@ -775,8 +768,8 @@ end
@inline function read_cmbp1_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand All @@ -797,8 +790,8 @@ end
@inline function read_cbbo1s_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand All @@ -819,8 +812,8 @@ end
@inline function read_cbbo1m_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand All @@ -841,8 +834,8 @@ end
@inline function read_tcbbo_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand All @@ -863,8 +856,8 @@ end
@inline function read_bbo1s_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand All @@ -885,8 +878,8 @@ end
@inline function read_bbo1m_msg(decoder::DBNDecoder, hd::RecordHeader)
price = read(decoder.io, Int64)
size = read(decoder.io, UInt32)
action = unsafe_action(read(decoder.io, UInt8))
side = unsafe_side(read(decoder.io, UInt8))
action = safe_action(read(decoder.io, UInt8))
side = safe_side(read(decoder.io, UInt8))
flags = read(decoder.io, UInt8)
depth = read(decoder.io, UInt8)
ts_recv = read(decoder.io, Int64)
Expand Down Expand Up @@ -1108,7 +1101,7 @@ end
leg_instrument_class_byte = read(decoder.io, UInt8)
leg_instrument_class = safe_instrument_class(leg_instrument_class_byte)
leg_side_byte = read(decoder.io, UInt8)
leg_side = unsafe_side(leg_side_byte)
leg_side = safe_side(leg_side_byte)

# v3: 17 bytes _reserved
skip(decoder.io, 17)
Expand Down Expand Up @@ -1433,8 +1426,11 @@ function read_dbn_typed(filename::String, ::Type{T}) where T
end
end

# Read record directly - uses same helpers as streaming
@inbounds records[idx] = _read_typed_record_stream(decoder, T, hd)
# Read record directly - uses same helpers as streaming.
# A reader can return nothing for a malformed record it skipped.
rec = _read_typed_record_stream(decoder, T, hd)
rec === nothing && continue
@inbounds records[idx] = rec
idx += 1
end

Expand Down Expand Up @@ -1479,7 +1475,9 @@ function read_dbn_typed(filename::String, ::Type{T}) where T
end
end

push!(records, _read_typed_record_stream(decoder, T, hd))
# A reader can return nothing for a malformed record it skipped.
rec = _read_typed_record_stream(decoder, T, hd)
rec === nothing || push!(records, rec)
end
finally
if decoder.io !== decoder.base_io
Expand Down
Loading
Loading