From 60f7885c63d596fb043a964b0cd207664b6d01a2 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Wed, 24 Jun 2026 13:18:24 -0500 Subject: [PATCH 1/3] Add timestamp-paced replay (replay_dbn / replay_records) Re-emit DBN records in time order, paced by their timestamps to simulate a live feed (backtesting, demos, driving real-time consumers). - replay_dbn(f, filename; ...) streams from a file (Zstd-aware, skips unknown rtypes like DBNStream) and invokes f(record) paced in real time. - replay_records(f, records; ...) does the same over an in-memory collection. - Wall-clock-anchored pacing absorbs callback execution time instead of accumulating drift; the stream re-synchronizes after slow callbacks. Options: speed (time-compression multiplier; Inf = no waiting), timestamp (:ts_event or :ts_recv with fallback), max_sleep (cap large gaps, re-anchors), and precise (busy-wait sub-millisecond gaps for dense-burst fidelity, since Base.sleep only resolves ~1ms on Unix / ~15ms on Windows). clock/sleep_fn are injectable for deterministic testing. Adds test/test_replay.jl (35 tests, fake-clock deterministic pacing checks + real-clock precise-mode checks) and documents the timing-resolution limit in the docstrings and README. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 46 ++++++ src/DatabentoBinaryEncoding.jl | 2 + src/replay.jl | 292 +++++++++++++++++++++++++++++++++ test/runtests.jl | 1 + test/test_replay.jl | 179 ++++++++++++++++++++ 5 files changed, 520 insertions(+) create mode 100644 src/replay.jl create mode 100644 test/test_replay.jl diff --git a/README.md b/README.md index 3228f87..219c55a 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ For more details, read the [introduction to DBN](https://databento.com/docs/stan - ✅ Complete DBN v3 Format Support - ✅ Efficient streaming support (read and write) +- ✅ Timestamp-paced replay (simulate a live feed at any speed) - ✅ Zstd file compression support (read and write) - ✅ Bidirectional format conversion (DBN ↔ JSON/Parquet/CSV) - ✅ Byte-for-byte compatibility with official implementations @@ -79,6 +80,51 @@ end println("Total: $(total[])") ``` +### Replaying DBN Files + +Re-emit a file's records in time order, paced by their timestamps, to simulate a +live feed (useful for backtesting, demos, or driving consumers that expect +real-time data): + +```julia +# Replay in real time — each record fires after the real gap to the previous one +replay_dbn("trades.dbn") do rec + println(price_to_float(rec.price)) +end + +# Replay 100x faster, paced by receive time, with overnight gaps capped at 1s +replay_dbn("mbo.dbn.zst"; speed = 100, timestamp = :ts_recv, max_sleep = 1.0) do rec + handle(rec) +end + +# Replay an already-loaded collection (e.g. from read_dbn) +records = read_dbn("trades.dbn") +replay_records(records; speed = 5) do rec + handle(rec) +end +``` + +Key options: `speed` (time-compression multiplier; `Inf` = no waiting), +`timestamp` (`:ts_event` or `:ts_recv`), and `max_sleep` (cap on any single +wait, in seconds). Compression and unknown record types are handled exactly as +in `DBNStream`. + +**Timing resolution.** DBN timestamps are nanosecond precision, but pacing +accuracy is bounded by the sleep function. `Base.sleep` resolves to roughly +1 ms on Unix and as coarse as the system timer tick (~15 ms) on Windows, so +records spaced more tightly than that arrive clumped rather than as distinct +waits. Pacing is anchored to absolute wall-clock targets, so this clumping is +local — the stream re-synchronizes and timing error does not accumulate across +the file, and records sharing a timestamp are delivered back-to-back. For +sub-millisecond fidelity (e.g. dense MBO bursts) pass `precise = true`, which +busy-waits small gaps at the cost of pinning a CPU core: + +```julia +replay_dbn("mbo.dbn"; precise = true) do rec + handle(rec) +end +``` + ### Writing DBN Files ```julia diff --git a/src/DatabentoBinaryEncoding.jl b/src/DatabentoBinaryEncoding.jl index 36d50d3..655d9f9 100644 --- a/src/DatabentoBinaryEncoding.jl +++ b/src/DatabentoBinaryEncoding.jl @@ -99,6 +99,7 @@ include("buffered_io.jl") include("decode.jl") include("encode.jl") include("streaming.jl") +include("replay.jl") include("export.jl") include("symbols.jl") include("import.jl") @@ -117,6 +118,7 @@ export foreach_record, foreach_record_with_control, foreach_trade, foreach_mbo, export record_type_for_dbn_schema # Schema -> concrete record type export foreach_ohlcv, foreach_ohlcv_1s, foreach_ohlcv_1m, foreach_ohlcv_1h, foreach_ohlcv_1d # OHLCV streaming export foreach_cmbp1, foreach_cbbo1s, foreach_cbbo1m, foreach_tcbbo, foreach_bbo1s, foreach_bbo1m # Consolidated/BBO streaming +export replay_dbn, replay_records # Timestamp-paced replay export compress_dbn_file, compress_daily_files export Schema, Compression, Encoding, SType, RType, Action, Side, InstrumentClass export price_to_float, float_to_price, ts_to_datetime, datetime_to_ts, ts_to_date_time, date_time_to_ts, to_nanoseconds diff --git a/src/replay.jl b/src/replay.jl new file mode 100644 index 0000000..8516031 --- /dev/null +++ b/src/replay.jl @@ -0,0 +1,292 @@ +# DBN replay functionality +# +# "Replay" re-emits the records of a DBN file (or an in-memory record +# collection) in their original time order, pacing each callback by the gap +# between record timestamps so the stream is delivered as if it were arriving +# live. This is useful for backtesting, demos, and exercising consumers that +# expect a real-time feed. + +""" + _replay_as_ns(x) -> Int64 + +Normalize a timestamp field to `Int64` nanoseconds. `ts_recv` is `UInt64` on a +few record types (`StatusMsg`, `StatMsg`); a value that does not fit in an +`Int64` is treated as the undefined sentinel. +""" +@inline _replay_as_ns(x::Int64) = x +@inline _replay_as_ns(x::UInt64) = x > UInt64(typemax(Int64)) ? UNDEF_TIMESTAMP : Int64(x) + +""" + _replay_timestamp(rec, which::Symbol) -> Int64 + +Extract the pacing timestamp (in nanoseconds) from a record. + +`:ts_event` uses the record header's `ts_event`, which every record type has. +`:ts_recv` uses the record's `ts_recv` field when present and defined, falling +back to `ts_event` otherwise (several record types — `OHLCVMsg`, `ErrorMsg`, +`SystemMsg`, `SymbolMappingMsg` — have no `ts_recv`). +""" +@inline function _replay_timestamp(rec, which::Symbol) + if which === :ts_event + return rec.hd.ts_event + elseif which === :ts_recv + if hasproperty(rec, :ts_recv) + ts = _replay_as_ns(getproperty(rec, :ts_recv)) + return ts == UNDEF_TIMESTAMP ? rec.hd.ts_event : ts + end + return rec.hd.ts_event + else + throw(ArgumentError("`timestamp` must be :ts_event or :ts_recv, got $(which)")) + end +end + +""" + _replay_loop(f, produce; speed, timestamp, max_sleep, clock, sleep_fn) -> Int + +Core replay engine. `produce` is a zero-argument function returning the next +record or `nothing` when the stream is exhausted. Returns the number of records +delivered to `f`. + +Pacing is anchored to wall-clock time: the wait before record `i` is computed +so that record `i` fires at `wall_anchor + (ts_i - data_anchor) / speed` +seconds. Anchoring (rather than sleeping the raw inter-record gap each step) +means time spent inside `f` is absorbed instead of accumulating as drift — a +slow callback simply shortens the next wait rather than pushing every +subsequent record later. +""" +function _replay_loop(f::Function, produce::Function; + speed::Real, + timestamp::Symbol, + max_sleep::Union{Real,Nothing}, + clock::Function, + sleep_fn::Function) + speed > 0 || throw(ArgumentError("`speed` must be positive, got $(speed)")) + max_sleep === nothing || max_sleep >= 0 || + throw(ArgumentError("`max_sleep` must be non-negative, got $(max_sleep)")) + + # `pace` is false when we should deliver records as fast as possible + # (infinite speed compression) — pacing math is skipped entirely. + pace = isfinite(speed) + + count = 0 + have_anchor = false + data_anchor = Int64(0) # ts (ns) of the anchor record + wall_anchor = 0.0 # clock() reading at the anchor + + while true + rec = produce() + rec === nothing && break + count += 1 + + if pace + ts = _replay_timestamp(rec, timestamp) + + if ts == UNDEF_TIMESTAMP + # No usable timestamp — deliver immediately without disturbing + # the anchor. + f(rec) + continue + end + + if !have_anchor + # First timed record establishes the anchor; it fires now. + data_anchor = ts + wall_anchor = clock() + have_anchor = true + f(rec) + continue + end + + elapsed = (ts - data_anchor) / 1.0e9 / speed # target seconds since anchor + sleep_s = wall_anchor + elapsed - clock() + + if sleep_s > 0 + if max_sleep !== nothing && sleep_s > max_sleep + # Cap the wait and re-anchor so we don't try to "catch up" + # the time we deliberately skipped over a large gap. + sleep_fn(float(max_sleep)) + data_anchor = ts + wall_anchor = clock() + else + sleep_fn(sleep_s) + end + end + end + + f(rec) + end + + return count +end + +# Below this gap, `Base.sleep` (≈1 ms granularity on Unix, and as coarse as the +# system timer tick — often ~15 ms — on Windows) overshoots badly. `precise` +# mode busy-waits the residual on the real clock to recover sub-millisecond +# accuracy at the cost of pinning a CPU core. +const _SPIN_THRESHOLD = 2.0e-3 + +""" + _precise_sleep(s::Real) + +Sleep for `s` seconds with sub-millisecond accuracy. Coarse-sleeps everything +beyond `_SPIN_THRESHOLD`, then busy-waits the remainder against `Base.time()`. +Used when `replay_dbn`/`replay_records` are called with `precise = true`. +""" +function _precise_sleep(s::Real) + s <= 0 && return nothing + deadline = time() + s + if s > _SPIN_THRESHOLD + sleep(s - _SPIN_THRESHOLD) # yield the bulk to the scheduler + end + while time() < deadline # spin the residual (≤ _SPIN_THRESHOLD) + end + return nothing +end + +# Resolve the sleep function: an explicit `sleep_fn` always wins (tests inject +# one); otherwise `precise` selects the spin-accurate sleeper or plain `sleep`. +@inline function _resolve_sleep_fn(sleep_fn::Union{Function,Nothing}, precise::Bool) + sleep_fn !== nothing && return sleep_fn + return precise ? _precise_sleep : sleep +end + +""" + replay_dbn(f::Function, filename::AbstractString; + speed::Real = 1.0, + timestamp::Symbol = :ts_event, + max_sleep::Union{Real,Nothing} = nothing, + precise::Bool = false, + clock::Function = time, + sleep_fn::Union{Function,Nothing} = nothing) -> Int + +Replay a DBN file, invoking `f(record)` for each record paced in real time +according to the records' timestamps. Returns the number of records replayed. + +Transparently handles Zstd compression and skips unknown record types, exactly +like [`DBNStream`](@ref). Records are delivered in file order. + +# Keyword arguments +- `speed`: time-compression multiplier. `1.0` replays at the original rate, + `2.0` at twice the speed (half the waits), `10.0` ten times faster. + `Inf` delivers every record as fast as possible (no waiting), equivalent to + plain streaming. +- `timestamp`: which timestamp to pace by — `:ts_event` (default, present on + every record) or `:ts_recv` (the receive timestamp, falling back to + `ts_event` for record types that have none). +- `max_sleep`: optional cap (in seconds) on any single wait. Large gaps (e.g. + an overnight session break) are clamped to this so replay doesn't stall. + After a clamp, pacing re-anchors to the current record. +- `precise`: when `true`, busy-wait sub-millisecond gaps instead of using + `Base.sleep`. See the timing-resolution note below. +- `clock` / `sleep_fn`: injection points for the wall clock and sleep function, + primarily for testing. `clock` defaults to `Base.time`. An explicit `sleep_fn` + overrides `precise`; when left as `nothing` the sleeper is `Base.sleep` + (or the precise busy-wait sleeper when `precise = true`). + +# Timing resolution +DBN timestamps are nanosecond precision, but pacing accuracy is bounded by the +sleep function, not the timestamps. `Base.sleep` resolves to roughly 1 ms on +Unix, and as coarse as the system timer tick (often ~15 ms) on Windows, so +records spaced more tightly than that cannot be reproduced as distinct waits — +they arrive clumped. Because pacing is anchored to absolute wall-clock targets, +this clumping is *local*: the stream re-synchronizes and the error does not +accumulate across the file, and records sharing a timestamp are delivered +back-to-back. For sub-millisecond fidelity (e.g. dense MBO bursts) pass +`precise = true`, which busy-waits small gaps at the cost of pinning a CPU core. + +# Examples +```julia +# Replay trades in real time +replay_dbn("trades.dbn") do rec + println(price_to_float(rec.price)) +end + +# Replay 100x faster, paced by receive time, with overnight gaps capped at 1s +replay_dbn("mbo.dbn.zst"; speed = 100, timestamp = :ts_recv, max_sleep = 1.0) do rec + handle(rec) +end + +# Microsecond-accurate replay of a dense MBO burst (busy-waits; uses a full core) +replay_dbn("mbo.dbn"; precise = true) do rec + handle(rec) +end +``` + +See also [`replay_records`](@ref) to replay an already-loaded record collection, +and [`DBNStream`](@ref) / [`foreach_record`](@ref) for unpaced streaming. +""" +function replay_dbn(f::Function, filename::AbstractString; + speed::Real = 1.0, + timestamp::Symbol = :ts_event, + max_sleep::Union{Real,Nothing} = nothing, + precise::Bool = false, + clock::Function = time, + sleep_fn::Union{Function,Nothing} = nothing) + sleeper = _resolve_sleep_fn(sleep_fn, precise) + decoder = DBNDecoder(String(filename)) + produce = function () + while !eof(decoder.io) + rec = read_record(decoder) + rec === nothing && continue # unknown rtype: skip, keep going + return rec + end + return nothing + end + try + return _replay_loop(f, produce; + speed = speed, timestamp = timestamp, + max_sleep = max_sleep, clock = clock, sleep_fn = sleeper) + finally + if decoder.io !== decoder.base_io + close(decoder.io) + end + if isa(decoder.base_io, IOStream) + close(decoder.base_io) + end + end +end + +""" + replay_records(f::Function, records; + speed::Real = 1.0, + timestamp::Symbol = :ts_event, + max_sleep::Union{Real,Nothing} = nothing, + precise::Bool = false, + clock::Function = time, + sleep_fn::Union{Function,Nothing} = nothing) -> Int + +Replay an in-memory collection of records (e.g. the result of [`read_dbn`](@ref)), +invoking `f(record)` for each one paced in real time by its timestamp. Returns +the number of records replayed. + +Accepts any iterable of DBN record types. All keyword arguments behave exactly +as in [`replay_dbn`](@ref), including the `precise` flag and the timing-resolution +caveats documented there. + +# Example +```julia +records = read_dbn("trades.dbn") +replay_records(records; speed = 5) do rec + handle(rec) +end +``` +""" +function replay_records(f::Function, records; + speed::Real = 1.0, + timestamp::Symbol = :ts_event, + max_sleep::Union{Real,Nothing} = nothing, + precise::Bool = false, + clock::Function = time, + sleep_fn::Union{Function,Nothing} = nothing) + sleeper = _resolve_sleep_fn(sleep_fn, precise) + next = iterate(records) + produce = function () + next === nothing && return nothing + rec, state = next + next = iterate(records, state) + return rec + end + return _replay_loop(f, produce; + speed = speed, timestamp = timestamp, + max_sleep = max_sleep, clock = clock, sleep_fn = sleeper) +end diff --git a/test/runtests.jl b/test/runtests.jl index 4f98981..3c1246d 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -19,6 +19,7 @@ include("test_utils.jl") include("test_phase10_complete.jl") # Integration and performance testing include("test_convenience_functions.jl") # Test all convenience read_*/foreach_* functions include("test_phase11_typed_with_control.jl") # foreach_record_with_control: typed data + Union control split + include("test_replay.jl") # replay_dbn / replay_records: timestamp-paced re-emission include("test_issue23_unset_stype.jl") # regression: 0xFF unset stype in v3 SymbolMappingMsg (issue #23) include("test_issues_32_35.jl") # regressions: pre-v3 StatMsg layout, stat_to_dataframe, mapping intervals, invalid enum bytes (issues #32-#35) include("test_show.jl") # compact one-line Base.show for record types diff --git a/test/test_replay.jl b/test/test_replay.jl new file mode 100644 index 0000000..eadba7e --- /dev/null +++ b/test/test_replay.jl @@ -0,0 +1,179 @@ +# Replay: timestamp-paced re-emission of records. +# +# Pacing is verified deterministically with an injected fake clock + sleep +# function so the tests never actually wait. The fake clock advances by exactly +# the amount each (zero-time) sleep requests, so captured sleep durations equal +# the inter-record gaps that real-time replay would produce. + +using Test +using DatabentoBinaryEncoding +import DatabentoBinaryEncoding as DBN + +@testset "replay" begin + + # A fake clock whose time only advances when sleep_fn is called. Records + # every requested sleep so pacing can be asserted exactly. + function fake_clock() + now = Ref(0.0) + sleeps = Float64[] + clock = () -> now[] + sleep_fn = s -> (push!(sleeps, s); now[] += s; nothing) + return clock, sleep_fn, sleeps + end + + # Build a TradeMsg with a given ts_event (and optional ts_recv). + function trade(ts_event::Int64; ts_recv::Int64 = ts_event, seq::Integer = 0) + hd = DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, + UInt16(1), UInt32(100), ts_event) + return DBN.TradeMsg(hd, Int64(150_000_000_000), UInt32(100), + DBN.Action.TRADE, DBN.Side.ASK, UInt8(0), UInt8(1), + ts_recv, Int32(0), UInt32(seq)) + end + + S = 1_000_000_000 # one second in nanoseconds + + @testset "paces by inter-record gaps at speed 1" begin + recs = [trade(Int64(0)), trade(Int64(1S)), trade(Int64(3S)), trade(Int64(6S))] + clock, sleep_fn, sleeps = fake_clock() + seen = DBN.TradeMsg[] + n = replay_records(r -> push!(seen, r), recs; + clock = clock, sleep_fn = sleep_fn) + @test n == 4 + @test length(seen) == 4 + # Gaps: 1s, 2s, 3s. First record fires immediately (no sleep). + @test sleeps == [1.0, 2.0, 3.0] + end + + @testset "speed multiplier compresses waits" begin + recs = [trade(Int64(0)), trade(Int64(2S)), trade(Int64(4S))] + clock, sleep_fn, sleeps = fake_clock() + n = replay_records(_ -> nothing, recs; + speed = 2.0, clock = clock, sleep_fn = sleep_fn) + @test n == 3 + @test sleeps == [1.0, 1.0] # 2s gaps at 2x → 1s waits + end + + @testset "speed = Inf delivers with no waiting" begin + recs = [trade(Int64(0)), trade(Int64(5S)), trade(Int64(99S))] + clock, sleep_fn, sleeps = fake_clock() + seen = Int[] + n = replay_records(r -> push!(seen, length(seen) + 1), recs; + speed = Inf, clock = clock, sleep_fn = sleep_fn) + @test n == 3 + @test isempty(sleeps) + end + + @testset "out-of-order timestamps do not produce negative waits" begin + recs = [trade(Int64(5S)), trade(Int64(2S)), trade(Int64(7S))] + clock, sleep_fn, sleeps = fake_clock() + n = replay_records(_ -> nothing, recs; + clock = clock, sleep_fn = sleep_fn) + @test n == 3 + # Backwards gap → no sleep; then 7s is 2s after the 5s anchor → 2s wait. + @test sleeps == [2.0] + @test all(>=(0.0), sleeps) + end + + @testset "max_sleep caps large gaps and re-anchors" begin + # 0s, then a 100s gap (capped to 1s), then a 1s gap after. + recs = [trade(Int64(0)), trade(Int64(100S)), trade(Int64(101S))] + clock, sleep_fn, sleeps = fake_clock() + n = replay_records(_ -> nothing, recs; + max_sleep = 1.0, clock = clock, sleep_fn = sleep_fn) + @test n == 3 + # First gap clamped to 1.0; re-anchored, so the following 1s gap waits 1.0. + @test sleeps == [1.0, 1.0] + end + + @testset "timestamp = :ts_recv paces by receive time" begin + # ts_event identical; ts_recv carries the real spacing. + recs = [trade(Int64(0); ts_recv = Int64(0)), + trade(Int64(0); ts_recv = Int64(4S))] + clock, sleep_fn, sleeps = fake_clock() + n = replay_records(_ -> nothing, recs; + timestamp = :ts_recv, clock = clock, sleep_fn = sleep_fn) + @test n == 2 + @test sleeps == [4.0] + end + + @testset "empty collection is a no-op" begin + clock, sleep_fn, sleeps = fake_clock() + n = replay_records(_ -> error("should not be called"), DBN.TradeMsg[]; + clock = clock, sleep_fn = sleep_fn) + @test n == 0 + @test isempty(sleeps) + end + + @testset "invalid arguments throw" begin + recs = [trade(Int64(0)), trade(Int64(1S))] + @test_throws ArgumentError replay_records(_ -> nothing, recs; speed = 0) + @test_throws ArgumentError replay_records(_ -> nothing, recs; speed = -1) + @test_throws ArgumentError replay_records(_ -> nothing, recs; max_sleep = -1.0) + @test_throws ArgumentError replay_records(_ -> nothing, recs; timestamp = :bogus) + end + + @testset "sleep_fn resolution honors precise / explicit override" begin + # Default: plain Base.sleep. + @test DBN._resolve_sleep_fn(nothing, false) === sleep + # precise = true selects the spin-accurate sleeper. + @test DBN._resolve_sleep_fn(nothing, true) === DBN._precise_sleep + # An explicit sleep_fn always wins, even with precise = true. + custom = s -> nothing + @test DBN._resolve_sleep_fn(custom, true) === custom + end + + @testset "_precise_sleep does not return early" begin + # Zero / negative is a no-op. + @test DBN._precise_sleep(0) === nothing + @test DBN._precise_sleep(-1.0) === nothing + # A sub-millisecond request must actually wait at least that long + # (the spin path must not under-sleep). Upper bound left loose to + # avoid flakiness under load. + req = 0.0008 # 0.8 ms, below the sleep() resolution floor + t = @elapsed DBN._precise_sleep(req) + @test t >= req * 0.9 + @test t < req + 0.05 + end + + @testset "precise = true replays sub-millisecond gaps (real clock)" begin + # Five records 300 µs apart — total span 1.2 ms, well below what + # Base.sleep can resolve. Exercises the real _precise_sleep path. + gap = 300_000 # 0.3 ms in ns + recs = [trade(Int64(i * gap); seq = i) for i in 0:4] + seen = DBN.TradeMsg[] + t = @elapsed (n = replay_records(r -> push!(seen, r), recs; precise = true)) + @test n == 5 + @test [Int(r.sequence) for r in seen] == [0, 1, 2, 3, 4] + # Should take at least the real span and not balloon. + @test t >= 4 * gap / 1e9 * 0.9 + @test t < 0.1 + end + + @testset "replay_dbn over a file" begin + metadata = DBN.Metadata( + DBN.DBN_VERSION, "TEST.MOCK", DBN.Schema.TRADES, + Int64(0), nothing, nothing, nothing, + DBN.SType.INSTRUMENT_ID, false, + ["AAPL"], String[], String[], + Tuple{String,String,Int64,Int64}[], + ) + ts0 = Int64(1_700_000_000_000_000_000) + recs = DBN.DBNRecord[trade(ts0 + i * S; seq = i) for i in 0:3] + + tmp, io = mktemp(); close(io) + try + DBN.write_dbn(tmp, metadata, recs) + + clock, sleep_fn, sleeps = fake_clock() + seen = DBN.TradeMsg[] + n = replay_dbn(r -> push!(seen, r), tmp; + clock = clock, sleep_fn = sleep_fn) + @test n == 4 + @test length(seen) == 4 + @test [Int(r.sequence) for r in seen] == [0, 1, 2, 3] + @test sleeps == [1.0, 1.0, 1.0] # 1s spacing + finally + rm(tmp; force = true) + end + end +end From 7c10ea96af804e0172b99454c9589963fa9fd415 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Wed, 24 Jun 2026 13:22:19 -0500 Subject: [PATCH 2/3] Release 0.1.6: bump version, changelog entry for replay Bump version 0.1.5 -> 0.1.6 (additive, non-breaking: new replay_dbn / replay_records exports only). Version the changelog's Unreleased section as 0.1.6 and document the replay feature under Added. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 15 +++++++++++++-- Project.toml | 2 +- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5987d64..0213233 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ All notable changes to DatabentoBinaryEncoding.jl are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [0.1.6] - 2026-06-24 ### Changed @@ -18,6 +18,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Timestamp-paced **replay**: `replay_dbn(f, filename; ...)` re-emits a DBN file's + records in time order, pacing each `f(record)` callback by the gap between record + timestamps to simulate a live feed (backtesting, demos, driving real-time + consumers). `replay_records(f, records; ...)` does the same over an already-loaded + collection. Zstd-aware and skips unknown record types like `DBNStream`. Pacing is + anchored to absolute wall-clock targets so callback execution time is absorbed + instead of accumulating as drift. Options: `speed` (time-compression multiplier; + `Inf` = no waiting), `timestamp` (`:ts_event` or `:ts_recv`, with fallback), + `max_sleep` (cap large gaps; re-anchors after a clamp), and `precise` (busy-wait + sub-millisecond gaps, since `Base.sleep` only resolves ~1 ms on Unix / ~15 ms on + Windows). `clock`/`sleep_fn` are injectable for deterministic testing. - `dbn_to_parquet` now compresses with **ZSTD by default** and accepts a `compression` keyword — one of `"zstd"` (default), `"snappy"`, `"gzip"`, or `"uncompressed"`. - Symbol resolution helpers that join the human-readable `raw_symbol` back onto @@ -130,7 +141,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 conversion to JSON/Parquet/CSV; byte-for-byte compatibility with the official Rust implementation ([#18], [#19]). -[Unreleased]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/compare/v0.1.3...HEAD +[0.1.6]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/compare/v0.1.5...v0.1.6 [0.1.3]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/compare/v0.1.2...v0.1.3 [0.1.2]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/compare/v0.1.1...v0.1.2 [0.1.1]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/compare/v0.1.0...v0.1.1 diff --git a/Project.toml b/Project.toml index a29a645..3d60e3c 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "DatabentoBinaryEncoding" uuid = "90689371-c8cb-40d1-831f-18033db90f74" -version = "0.1.5" +version = "0.1.6" authors = ["Tyler Beason "] [deps] From 2081e454a3fb31129dbbf8f23d98046a530cb7a6 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Wed, 24 Jun 2026 13:33:11 -0500 Subject: [PATCH 3/3] Use monotonic clock for precise-sleep deadline _precise_sleep measured its busy-wait deadline against wall-clock time(); a system-clock adjustment (NTP, manual) mid-spin could make it return early or busy-wait far longer than requested (pinning a core). Measure the deadline with time_ns() (monotonic) instead. Addresses Codex PR review (P2). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/replay.jl | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/replay.jl b/src/replay.jl index 8516031..4bac762 100644 --- a/src/replay.jl +++ b/src/replay.jl @@ -129,16 +129,20 @@ const _SPIN_THRESHOLD = 2.0e-3 _precise_sleep(s::Real) Sleep for `s` seconds with sub-millisecond accuracy. Coarse-sleeps everything -beyond `_SPIN_THRESHOLD`, then busy-waits the remainder against `Base.time()`. -Used when `replay_dbn`/`replay_records` are called with `precise = true`. +beyond `_SPIN_THRESHOLD`, then busy-waits the remainder. Used when +`replay_dbn`/`replay_records` are called with `precise = true`. + +The deadline is measured against `Base.time_ns()` — a monotonic counter — rather +than wall-clock `time()`, so a system-clock adjustment (NTP, manual correction) +mid-spin can't make the busy-wait return early or pin a core indefinitely. """ function _precise_sleep(s::Real) s <= 0 && return nothing - deadline = time() + s + deadline = time_ns() + round(UInt64, s * 1.0e9) if s > _SPIN_THRESHOLD sleep(s - _SPIN_THRESHOLD) # yield the bulk to the scheduler end - while time() < deadline # spin the residual (≤ _SPIN_THRESHOLD) + while time_ns() < deadline # spin the residual (≤ _SPIN_THRESHOLD) end return nothing end