From a1a66efbba657ec7f1be242f5e4ec17527f02bc1 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Wed, 3 Jun 2026 16:13:55 -0500 Subject: [PATCH 1/2] Historical HTTP retries, backpressured foreach_record, start_dt/end_dt rename HTTP layer: - Retry transient failures (HTTP 429/5xx + connection/timeout errors) with full-jitter exponential backoff, honoring Retry-After (capped). Configurable via the new max_retries keyword on Historical(...) (default 3); retry budget, backoff, and sleep hook are injectable for testing. Once the budget is spent, the final status still maps to BentoClientError/BentoServerError as before. - Rewrite foreach_record/open_stream to read the response body synchronously on the calling task: real backpressure (no unbounded Base.BufferStream buffering the whole payload) and deterministic teardown on every exit path, including an early return or an exception out of the callback (no leaked producer task). Adds an injectable stream_opener seam so the streaming path is unit-testable. API rename (breaking): - Rename the time-range keyword arguments start/end_ -> start_dt/end_dt across get_range, foreach_record, submit_job, get_record_count, get_billable_size, get_cost, and the Live subscribe! / live_session subscription / stream_to_file / stream_multi_to_files start keyword. The trailing underscore only existed because `end` is a Julia reserved word; the pair is now symmetric. Databento wire parameters (start/end) are unchanged. Adds retry + streaming test coverage; updates docs, README, benchmarks, CHANGELOG. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 3 + CHANGELOG.md | 29 ++++ README.md | 6 +- benchmark/_fetch_replay_fixture.jl | 4 +- benchmark/bench_historical_decode.jl | 4 +- docs/src/guide/historical.md | 16 +- docs/src/index.md | 4 +- docs/src/quickstart.md | 4 +- docs/src/troubleshooting.md | 6 +- src/DatabentoAPI.jl | 4 +- src/conversion.jl | 4 +- src/errors.jl | 4 +- src/historical/batch.jl | 10 +- src/historical/client.jl | 14 +- src/historical/metadata.jl | 32 ++-- src/historical/timeseries.jl | 28 ++-- src/http.jl | 232 ++++++++++++++++++++------- src/live/client.jl | 4 +- src/live/streaming.jl | 8 +- src/live/subscribe.jl | 16 +- test/live/test_historical_smoke.jl | 12 +- test/test_historical_batch.jl | 4 +- test/test_historical_metadata.jl | 4 +- test/test_historical_timeseries.jl | 87 +++++++++- test/test_http.jl | 98 ++++++++++- test/test_live_subscribe.jl | 2 +- 26 files changed, 479 insertions(+), 160 deletions(-) diff --git a/.gitignore b/.gitignore index 8e0ed17..e44b21a 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,6 @@ config.toml # Benchmark artifacts benchmark/results/ benchmark/data/ + +# Live capture output dirs +live_capture_*/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ee4976..aa9c96a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **Automatic HTTP retries on the Historical API.** Transient failures — HTTP + `429` (rate limit) and `5xx`, plus connection/timeout errors — are now retried + with full-jitter exponential backoff. A `Retry-After` response header, when + present, overrides the computed backoff (capped to guard against hostile + values). Configurable via the new `max_retries` keyword on `Historical(...)` + (default 3); the retry budget, backoff, and sleep hook are all injectable for + testing. Once the budget is exhausted the final status is mapped to its + `BentoClientError`/`BentoServerError` as before. + +### Changed +- **Breaking: renamed the time-range keyword arguments to `start_dt` / `end_dt`.** + The old `start` / `end_` pair was asymmetric (the trailing underscore only + existed because `end` is a Julia reserved word). All affected entry points are + updated: `get_range`, `foreach_record`, `submit_job`, `get_record_count`, + `get_billable_size`, `get_cost`, and the Live `subscribe!` / `live_session` + subscription / `stream_to_file` / `stream_multi_to_files` `start` keyword + (now `start_dt`). The Databento wire parameters (`start` / `end`) are + unchanged. Update call sites from `start = …, end_ = …` to + `start_dt = …, end_dt = …`. +- **`foreach_record` now applies real backpressure.** The streaming download + previously drained the connection into an unbounded `Base.BufferStream` on a + background task, which could buffer the entire compressed payload in memory and + left the producer task running (blocking until server EOF/timeout) when the + consumer aborted or finished early. It now reads the response body synchronously + on the calling task, so records are pulled from the socket only as fast as the + consumer processes them, and the connection is torn down deterministically on + every exit path — including an early return or an exception out of the callback. + ## [0.1.2] - 2026-06-01 ### Added diff --git a/README.md b/README.md index abd4eb5..8bab739 100644 --- a/README.md +++ b/README.md @@ -43,8 +43,8 @@ store = get_range(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = DateTime(2024, 1, 2, 14, 30), - end_ = DateTime(2024, 1, 2, 14, 31), + start_dt = DateTime(2024, 1, 2, 14, 30), + end_dt = DateTime(2024, 1, 2, 14, 31), stype_in = SType.RAW_SYMBOL) df = to_dataframe(store) @@ -64,7 +64,7 @@ for details. ## Status -Current version: **0.1.1**. Offline test suite: **1500+ tests**, ~30 s. +Current version: **0.1.2**. Offline test suite: **1500+ tests**, ~30 s. Live-network smoke tests are gated behind `DATABENTO_LIVE_TESTS=1`. See [CHANGELOG.md](CHANGELOG.md) for the full release history. diff --git a/benchmark/_fetch_replay_fixture.jl b/benchmark/_fetch_replay_fixture.jl index ebca1a3..cdde61f 100644 --- a/benchmark/_fetch_replay_fixture.jl +++ b/benchmark/_fetch_replay_fixture.jl @@ -98,8 +98,8 @@ function main(args = ARGS) schema = Schema.CMBP_1, symbols = split(opts["symbols"], ','), stype_in = SType.PARENT, - start = opts["start"], - end_ = opts["end"], + start_dt = opts["start"], + end_dt = opts["end"], encoding = "dbn", compression = "zstd") job_id = String(job["id"]) println(" job_id = ", job_id, " cost_usd = ", job["cost_usd"]) diff --git a/benchmark/bench_historical_decode.jl b/benchmark/bench_historical_decode.jl index 4f1f4a9..7240a86 100644 --- a/benchmark/bench_historical_decode.jl +++ b/benchmark/bench_historical_decode.jl @@ -24,8 +24,8 @@ function _get_range_call(c) store = DatabentoAPI.get_range(c; dataset = "TEST.MOCK", schema = Schema.TRADES, symbols = ["AAPL"], - start = "2024-01-02T14:30:00", - end_ = "2024-01-02T14:31:00", + start_dt = "2024-01-02T14:30:00", + end_dt = "2024-01-02T14:31:00", stype_in = SType.RAW_SYMBOL, stype_out = SType.INSTRUMENT_ID, ) diff --git a/docs/src/guide/historical.md b/docs/src/guide/historical.md index 3031d26..da2f33a 100644 --- a/docs/src/guide/historical.md +++ b/docs/src/guide/historical.md @@ -33,8 +33,8 @@ store = get_range(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL", "MSFT"], - start = DateTime(2024, 1, 2, 14, 30), - end_ = DateTime(2024, 1, 2, 15, 0), + start_dt = DateTime(2024, 1, 2, 14, 30), + end_dt = DateTime(2024, 1, 2, 15, 0), stype_in = SType.RAW_SYMBOL) length(store.records) # Vector{DBN.DBNRecord} @@ -57,8 +57,8 @@ n_trades = Ref(0) foreach_record(client, DBN.TradeMsg; dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = DateTime(2024, 1, 2, 14, 30), - end_ = DateTime(2024, 1, 2, 20, 0)) do trade + start_dt = DateTime(2024, 1, 2, 14, 30), + end_dt = DateTime(2024, 1, 2, 20, 0)) do trade n_trades[] += 1 end @show n_trades[] @@ -75,15 +75,15 @@ Before issuing a large `get_range`, preview the query: ```julia get_record_count(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, - symbols = ["AAPL"], start = "2024-01-02", end_ = "2024-01-03") + symbols = ["AAPL"], start_dt = "2024-01-02", end_dt = "2024-01-03") # → integer record count get_billable_size(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, - symbols = ["AAPL"], start = "2024-01-02", end_ = "2024-01-03") + symbols = ["AAPL"], start_dt = "2024-01-02", end_dt = "2024-01-03") # → integer billable bytes get_cost(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, - symbols = ["AAPL"], start = "2024-01-02", end_ = "2024-01-03", + symbols = ["AAPL"], start_dt = "2024-01-02", end_dt = "2024-01-03", mode = FeedMode.HISTORICAL) # → USD cost ``` @@ -108,7 +108,7 @@ Submit an asynchronous job, poll for completion, then download: job = submit_job(client; dataset = "XNAS.ITCH", schema = Schema.MBO, symbols = ["AAPL"], - start = "2024-01-02", end_ = "2024-01-09") + start_dt = "2024-01-02", end_dt = "2024-01-09") job_id = job["id"] diff --git a/docs/src/index.md b/docs/src/index.md index 5e0bdbf..d00ae55 100644 --- a/docs/src/index.md +++ b/docs/src/index.md @@ -32,8 +32,8 @@ store = get_range(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = DateTime(2024, 1, 2, 14, 30), - end_ = DateTime(2024, 1, 2, 14, 31), + start_dt = DateTime(2024, 1, 2, 14, 30), + end_dt = DateTime(2024, 1, 2, 14, 31), stype_in = SType.RAW_SYMBOL) df = to_dataframe(store) diff --git a/docs/src/quickstart.md b/docs/src/quickstart.md index a16a8a8..197e23b 100644 --- a/docs/src/quickstart.md +++ b/docs/src/quickstart.md @@ -17,8 +17,8 @@ store = get_range(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = DateTime(2024, 1, 2, 14, 30), - end_ = DateTime(2024, 1, 2, 14, 31), + start_dt = DateTime(2024, 1, 2, 14, 30), + end_dt = DateTime(2024, 1, 2, 14, 31), stype_in = SType.RAW_SYMBOL) df = to_dataframe(store) diff --git a/docs/src/troubleshooting.md b/docs/src/troubleshooting.md index 82751b5..2f8d155 100644 --- a/docs/src/troubleshooting.md +++ b/docs/src/troubleshooting.md @@ -39,9 +39,9 @@ If `get_range` exhausts memory or returns extremely slowly, switch to Always preview before issuing a large `get_range`: ```julia -get_record_count(client; dataset, schema, symbols, start, end_) -get_billable_size(client; dataset, schema, symbols, start, end_) -get_cost(client; dataset, schema, symbols, start, end_, mode = FeedMode.HISTORICAL) +get_record_count(client; dataset, schema, symbols, start_dt, end_dt) +get_billable_size(client; dataset, schema, symbols, start_dt, end_dt) +get_cost(client; dataset, schema, symbols, start_dt, end_dt, mode = FeedMode.HISTORICAL) ``` All three are free and run server-side. diff --git a/src/DatabentoAPI.jl b/src/DatabentoAPI.jl index c9fa70b..f621064 100644 --- a/src/DatabentoAPI.jl +++ b/src/DatabentoAPI.jl @@ -18,8 +18,8 @@ store = get_range(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = DateTime(2024, 1, 2, 14, 30), - end_ = DateTime(2024, 1, 2, 14, 31), + start_dt = DateTime(2024, 1, 2, 14, 30), + end_dt = DateTime(2024, 1, 2, 14, 31), stype_in = SType.RAW_SYMBOL) df = to_dataframe(store) diff --git a/src/conversion.jl b/src/conversion.jl index 16bd273..d59191f 100644 --- a/src/conversion.jl +++ b/src/conversion.jl @@ -39,8 +39,8 @@ paths; for large stores prefer running `DBN.dbn_to_csv` directly on a captured ```julia store = get_range(client; dataset = "XNAS.ITCH", schema = Schema.TRADES, - symbols = ["AAPL"], start = DateTime(2024,1,2,14,30), - end_ = DateTime(2024,1,2,14,31), stype_in = SType.RAW_SYMBOL) + symbols = ["AAPL"], start_dt = DateTime(2024,1,2,14,30), + end_dt = DateTime(2024,1,2,14,31), stype_in = SType.RAW_SYMBOL) to_csv(store, "trades.csv") ``` """ diff --git a/src/errors.jl b/src/errors.jl index d640131..2931e9f 100644 --- a/src/errors.jl +++ b/src/errors.jl @@ -51,7 +51,9 @@ end Raised when an HTTP request to the Historical API returns a 5xx status (gateway timeout, internal error, etc.). Same field layout as -[`BentoClientError`](@ref); retrying after backoff is usually appropriate. +[`BentoClientError`](@ref). The client already retries 5xx (and 429) with +backoff up to its `max_retries` budget, so seeing this means the server kept +failing across every retry. """ struct BentoServerError <: BentoHttpError status::Int diff --git a/src/historical/batch.jl b/src/historical/batch.jl index c6fc8d0..0d51134 100644 --- a/src/historical/batch.jl +++ b/src/historical/batch.jl @@ -1,7 +1,7 @@ # Batch endpoints — submit asynchronous jobs and download results. """ - submit_job(client; dataset, symbols, schema, start, end_=nothing, ...) -> Dict + submit_job(client; dataset, symbols, schema, start_dt, end_dt=nothing, ...) -> Dict Submit an asynchronous batch job. The Historical API queues the request and returns a `job_id`; poll [`list_jobs`](@ref) until the [`JobState`](@ref) is @@ -29,8 +29,8 @@ function submit_job(c::Historical; dataset::AbstractString, symbols, schema::Schema.T, - start, - end_ = nothing, + start_dt, + end_dt = nothing, encoding::Union{Encoding.T,AbstractString} = "dbn", compression::Union{Compression.T,AbstractString} = "zstd", stype_in::SType.T = SType.RAW_SYMBOL, @@ -54,8 +54,8 @@ function submit_job(c::Historical; dataset = String(dataset), symbols = symbols_str(symbols), schema = schema_str(schema), - start = ts_str(start), - end_ = ts_str(end_), + start = ts_str(start_dt), + end_ = ts_str(end_dt), encoding = enc_v, compression = cmp_v, stype_in = stype_str(stype_in), diff --git a/src/historical/client.jl b/src/historical/client.jl index c3d590b..e8d0830 100644 --- a/src/historical/client.jl +++ b/src/historical/client.jl @@ -12,8 +12,8 @@ Client for the Databento Historical (HTTP) API. ```julia client = Historical() store = get_range(client; dataset="XNAS.ITCH", schema=Schema.TRADES, - symbols=["AAPL"], start=DateTime(2024,1,2,14,30), - end_=DateTime(2024,1,2,14,31), stype_in=SType.RAW_SYMBOL) + symbols=["AAPL"], start_dt=DateTime(2024,1,2,14,30), + end_dt=DateTime(2024,1,2,14,31), stype_in=SType.RAW_SYMBOL) ``` """ struct Historical @@ -25,13 +25,19 @@ function Historical(key::Union{Nothing,AbstractString} = nothing; timeout::Integer = DEFAULT_TIMEOUT, connect_timeout::Integer = DEFAULT_CONNECT_TIMEOUT, user_agent::AbstractString = USER_AGENT, - dispatcher::Function = _default_http_request) + dispatcher::Function = _default_http_request, + max_retries::Integer = DEFAULT_MAX_RETRIES, + retry_sleep::Function = sleep, + stream_opener::Function = _default_http_stream) api_key = load_api_key(key) return Historical(HTTPClient(api_key, gateway; timeout = timeout, connect_timeout = connect_timeout, user_agent = user_agent, - dispatcher = dispatcher)) + dispatcher = dispatcher, + max_retries = max_retries, + retry_sleep = retry_sleep, + stream_opener = stream_opener)) end Base.show(io::IO, c::Historical) = diff --git a/src/historical/metadata.jl b/src/historical/metadata.jl index 9ec6018..f0479c0 100644 --- a/src/historical/metadata.jl +++ b/src/historical/metadata.jl @@ -65,7 +65,7 @@ list_unit_prices(c::Historical; dataset::AbstractString) = get_dataset_range(client; dataset) Earliest and latest available timestamps for `dataset`. Use the result to -bound `start`/`end_` on subsequent [`get_range`](@ref) or [`submit_job`](@ref) +bound `start_dt`/`end_dt` on subsequent [`get_range`](@ref) or [`submit_job`](@ref) calls. Wraps `GET /v0/metadata.get_dataset_range`. """ get_dataset_range(c::Historical; dataset::AbstractString) = @@ -89,7 +89,7 @@ function get_dataset_condition(c::Historical; end """ - get_record_count(client; dataset, symbols, schema, start, end_=nothing, + get_record_count(client; dataset, symbols, schema, start_dt, end_dt=nothing, stype_in=SType.RAW_SYMBOL, limit=nothing) Count of records the same query would return from [`get_range`](@ref) without @@ -101,8 +101,8 @@ function get_record_count(c::Historical; dataset::AbstractString, symbols, schema::Schema.T, - start, - end_ = nothing, + start_dt, + end_dt = nothing, stype_in::SType.T = SType.RAW_SYMBOL, limit::Union{Nothing,Integer} = nothing) qpairs = _clean_params(( @@ -110,8 +110,8 @@ function get_record_count(c::Historical; symbols = symbols_str(symbols), schema = schema_str(schema), stype_in = stype_str(stype_in), - start = ts_str(start), - end_ = ts_str(end_), + start = ts_str(start_dt), + end_ = ts_str(end_dt), limit = limit, )) for (i, (k, v)) in enumerate(qpairs) @@ -121,7 +121,7 @@ function get_record_count(c::Historical; end """ - get_billable_size(client; dataset, symbols, schema, start, end_=nothing, + get_billable_size(client; dataset, symbols, schema, start_dt, end_dt=nothing, stype_in=SType.RAW_SYMBOL, limit=nothing) Billable size in bytes (uncompressed CSV-equivalent) the same query would @@ -133,8 +133,8 @@ function get_billable_size(c::Historical; dataset::AbstractString, symbols, schema::Schema.T, - start, - end_ = nothing, + start_dt, + end_dt = nothing, stype_in::SType.T = SType.RAW_SYMBOL, limit::Union{Nothing,Integer} = nothing) qpairs = _clean_params(( @@ -142,8 +142,8 @@ function get_billable_size(c::Historical; symbols = symbols_str(symbols), schema = schema_str(schema), stype_in = stype_str(stype_in), - start = ts_str(start), - end_ = ts_str(end_), + start = ts_str(start_dt), + end_ = ts_str(end_dt), limit = limit, )) for (i, (k, v)) in enumerate(qpairs) @@ -153,7 +153,7 @@ function get_billable_size(c::Historical; end """ - get_cost(client; dataset, symbols, schema, start, end_=nothing, + get_cost(client; dataset, symbols, schema, start_dt, end_dt=nothing, mode=nothing, stype_in=SType.RAW_SYMBOL, limit=nothing) Dollar cost preview for the same query as [`get_range`](@ref) under the @@ -165,8 +165,8 @@ function get_cost(c::Historical; dataset::AbstractString, symbols, schema::Schema.T, - start, - end_ = nothing, + start_dt, + end_dt = nothing, mode::Union{Nothing,FeedMode.T,AbstractString} = nothing, stype_in::SType.T = SType.RAW_SYMBOL, limit::Union{Nothing,Integer} = nothing) @@ -177,8 +177,8 @@ function get_cost(c::Historical; symbols = symbols_str(symbols), schema = schema_str(schema), stype_in = stype_str(stype_in), - start = ts_str(start), - end_ = ts_str(end_), + start = ts_str(start_dt), + end_ = ts_str(end_dt), mode = mode_v, limit = limit, )) diff --git a/src/historical/timeseries.jl b/src/historical/timeseries.jl index 2b433b0..b761b9d 100644 --- a/src/historical/timeseries.jl +++ b/src/historical/timeseries.jl @@ -1,5 +1,5 @@ """ - get_range(client; dataset, schema, symbols, start, end_=nothing, + 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) @@ -23,8 +23,8 @@ function get_range(c::Historical; dataset::AbstractString, schema::Schema.T, symbols, - start, - end_ = nothing, + start_dt, + end_dt = nothing, stype_in::SType.T = SType.RAW_SYMBOL, stype_out::SType.T = SType.INSTRUMENT_ID, limit::Union{Nothing,Integer} = nothing, @@ -36,8 +36,8 @@ function get_range(c::Historical; schema = schema_str(schema), stype_in = stype_str(stype_in), stype_out = stype_str(stype_out), - start = ts_str(start), - end_ = ts_str(end_), + start = ts_str(start_dt), + end_ = ts_str(end_dt), limit = limit, encoding = "dbn", compression = "zstd", @@ -90,14 +90,16 @@ function record_type_for_schema(schema::Schema.T) end """ - foreach_record(f, client::Historical; dataset, schema, symbols, start, end_=nothing, + foreach_record(f, client::Historical; dataset, schema, symbols, start_dt, end_dt=nothing, stype_in=SType.RAW_SYMBOL, stype_out=SType.INSTRUMENT_ID, limit=nothing, record_type=nothing) Streaming variant of [`get_range`](@ref). Calls `f(record)` for each `DBN` record as -it arrives off the wire — overlaps HTTP download with decompress + decode and never -materialises the compressed payload or the records vector in memory. Use this for -very large queries where `get_range`'s in-memory buffering would be wasteful. +it arrives off the wire — interleaves HTTP download with decompress + decode and never +materialises the compressed payload or the records vector in memory. Records are +pulled from the socket only as fast as `f` consumes them (backpressure), so a slow +consumer throttles the download rather than buffering it. Use this for very large +queries where `get_range`'s in-memory buffering would be wasteful. By default, the concrete record type is inferred from `schema` (e.g. `Schema.TRADES → DBN.TradeMsg`) and the type-specific zero-allocation decode path @@ -110,8 +112,8 @@ function foreach_record(f, c::Historical; dataset::AbstractString, schema::Schema.T, symbols, - start, - end_ = nothing, + start_dt, + end_dt = nothing, stype_in::SType.T = SType.RAW_SYMBOL, stype_out::SType.T = SType.INSTRUMENT_ID, limit::Union{Nothing,Integer} = nothing, @@ -122,8 +124,8 @@ function foreach_record(f, c::Historical; schema = schema_str(schema), stype_in = stype_str(stype_in), stype_out = stype_str(stype_out), - start = ts_str(start), - end_ = ts_str(end_), + start = ts_str(start_dt), + end_ = ts_str(end_dt), limit = limit, encoding = "dbn", compression = "zstd", diff --git a/src/http.jl b/src/http.jl index 76ba091..b936105 100644 --- a/src/http.jl +++ b/src/http.jl @@ -1,7 +1,16 @@ -const USER_AGENT = "DatabentoAPI.jl/0.1.0" +const USER_AGENT = "DatabentoAPI.jl/0.1.2" const DEFAULT_TIMEOUT = 100 const DEFAULT_CONNECT_TIMEOUT = 30 +# Retry policy. Transient failures (HTTP 429 / 5xx and connection/timeout +# errors) are retried with full-jitter exponential backoff; a `Retry-After` +# header, when present, takes precedence over the computed backoff. +const DEFAULT_MAX_RETRIES = 3 +const RETRY_BASE_S = 0.5 +const RETRY_CAP_S = 30.0 +# Cap an honored `Retry-After` so a misbehaving/hostile header can't pin a call. +const RETRY_AFTER_CAP_S = 60.0 + """ basic_auth_header(api_key) @@ -14,6 +23,29 @@ basic_auth_header(api_key::AbstractString) = "Basic " * base64encode(string(api_ _default_http_request(method, url, headers, body; kwargs...) = HTTP.request(method, url, headers, body; kwargs...) +# Default low-level streaming opener. Opens a connection, reads the response +# head, and invokes `consume(status, headers, io)` with a readable `io` +# positioned at the body start. `consume` drives the IO synchronously, so the +# socket itself provides backpressure (no unbounded buffering), and the +# `HTTP.open` do-block tears the connection down on every exit path — +# including an early return or an exception out of `consume`. Tests substitute +# their own opener to feed canned bodies/statuses without a live socket. +# +# `consume` is the first parameter so callers can use do-block syntax +# (`stream_opener(c, method, url, headers, qpairs) do status, headers, io ...`). +function _default_http_stream(consume, c, method, url, headers, qpairs) + HTTP.open(method, url, headers; + query = qpairs, + status_exception = false, + decompress = false, + retry = false, + readtimeout = c.timeout, + connect_timeout = c.connect_timeout) do http + HTTP.startread(http) + return consume(http.message.status, http.message.headers, http) + end +end + mutable struct HTTPClient api_key::String base_url::String @@ -21,15 +53,51 @@ mutable struct HTTPClient connect_timeout::Int user_agent::String dispatcher::Function + max_retries::Int + retry_sleep::Function + stream_opener::Function end HTTPClient(api_key::AbstractString, base_url::AbstractString; timeout::Integer = DEFAULT_TIMEOUT, connect_timeout::Integer = DEFAULT_CONNECT_TIMEOUT, user_agent::AbstractString = USER_AGENT, - dispatcher::Function = _default_http_request) = + dispatcher::Function = _default_http_request, + max_retries::Integer = DEFAULT_MAX_RETRIES, + retry_sleep::Function = sleep, + stream_opener::Function = _default_http_stream) = HTTPClient(String(api_key), String(base_url), Int(timeout), Int(connect_timeout), - String(user_agent), dispatcher) + String(user_agent), dispatcher, Int(max_retries), retry_sleep, stream_opener) + +# A transient HTTP status worth retrying: 429 (rate limit) and 5xx, except +# 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 + +# 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. +function _retry_after_seconds(headers) + for (k, v) in headers + if lowercase(String(k)) == "retry-after" + n = tryparse(Float64, strip(String(v))) + n === nothing && return nothing + return clamp(n, 0.0, RETRY_AFTER_CAP_S) + end + end + return nothing +end + +# Full-jitter exponential backoff for a 1-based attempt index. +_retry_delay(attempt::Integer; base::Real = RETRY_BASE_S, cap::Real = RETRY_CAP_S) = + rand() * min(cap, base * 2.0^(attempt - 1)) + +# Sleep duration before the next attempt: honor Retry-After when supplied, +# otherwise fall back to jittered backoff. +_retry_wait(c::HTTPClient, attempt::Integer, retry_after::Union{Nothing,Real}) = + c.retry_sleep(retry_after === nothing ? _retry_delay(attempt) : retry_after) # Drop entries whose value is `nothing`; convert remaining values to strings. function _clean_params(params)::Vector{Pair{String,String}} @@ -69,6 +137,12 @@ end Perform an HTTP request against the Databento gateway. Maps 4xx → `BentoClientError` and 5xx → `BentoServerError`. Returns the underlying `HTTP.Messages.Response`. +Transient failures — HTTP `429` (rate limit) and `5xx`, plus connection/timeout +errors — are retried up to `c.max_retries` times with full-jitter exponential +backoff. A `Retry-After` response header, when present, overrides the backoff. +Once the retry budget is exhausted the final response is mapped to its error as +usual, so a persistently rate-limited request still surfaces `BentoClientError(429)`. + Use `body` as a Dict / NamedTuple for form-encoded POST bodies, or as a `String` / `Vector{UInt8}` for a pre-encoded body. """ @@ -98,22 +172,44 @@ function request(c::HTTPClient, method::Symbol, path::AbstractString; Vector{UInt8}(string(body)) end - resp = c.dispatcher(String(uppercase(string(method))), url, headers, body_bytes; - query = qpairs, - status_exception = false, - readtimeout = c.timeout, - connect_timeout = c.connect_timeout, - retry = false, - decompress = false) - - if 400 <= resp.status < 500 - throw(http_error_from_response(BentoClientError, resp.status, - String(copy(resp.body)), _request_id(resp))) - elseif resp.status >= 500 - throw(http_error_from_response(BentoServerError, resp.status, - String(copy(resp.body)), _request_id(resp))) + method_str = String(uppercase(string(method))) + attempt = 0 + while true + attempt += 1 + resp = try + c.dispatcher(method_str, url, headers, body_bytes; + query = qpairs, + status_exception = false, + readtimeout = c.timeout, + connect_timeout = c.connect_timeout, + retry = false, + decompress = false) + catch e + # Transient transport failure: back off and retry until the budget + # is spent, then let the original exception propagate. + if _is_retryable_exception(e) && attempt <= c.max_retries + _retry_wait(c, attempt, nothing) + continue + end + rethrow() + end + + # Retry transient statuses while budget remains; on the final attempt + # fall through to the error mapping below. + if _is_retryable_status(resp.status) && attempt <= c.max_retries + _retry_wait(c, attempt, _retry_after_seconds(resp.headers)) + continue + end + + if 400 <= resp.status < 500 + throw(http_error_from_response(BentoClientError, resp.status, + String(copy(resp.body)), _request_id(resp))) + elseif resp.status >= 500 + throw(http_error_from_response(BentoServerError, resp.status, + String(copy(resp.body)), _request_id(resp))) + end + return resp end - return resp end # Convenience wrappers for the common cases. @@ -151,15 +247,32 @@ function get_bytes(c::HTTPClient, path::AbstractString; return Vector{UInt8}(resp.body) end +# Pull the request-id out of a header collection (case-insensitive). +function _request_id_from_headers(headers)::String + for (k, v) in headers + lowercase(String(k)) == "request-id" && return String(v) + end + return "" +end + """ open_stream(f, c, path; query=nothing, accept="application/octet-stream") -Open a streaming GET against `path` and call `f(io)` with an `IO` that produces the -response body bytes as they arrive. Status-checking happens before `f` is invoked; -errors raise `BentoClientError`/`BentoServerError` like the eager `request` path. +Open a streaming GET against `path` and call `f(io)` with a readable `IO` that +yields the response body bytes as they arrive off the wire. `f` reads `io` +synchronously on the calling task, so the connection provides natural +backpressure — bytes are pulled from the socket only as fast as `f` consumes +them, and the payload is never buffered ahead in memory. The connection is +torn down on every exit path, including `f` returning early (e.g. an iterator +`break`) or throwing, so no socket or task is leaked. + +Status-checking happens before `f` is invoked; errors raise +`BentoClientError`/`BentoServerError` like the eager `request` path. The +pre-body phase (connect + read response head) is retried on transient failures +per `c.max_retries`; once bytes have been handed to `f` the call is not +retried (the consumer may already have observed partial output). -The body `IO` is a `Base.BufferStream`; the HTTP download runs in a background task -that drains the underlying connection into the stream. +Returns whatever `f` returns. """ function open_stream(f, c::HTTPClient, path::AbstractString; query = nothing, @@ -171,47 +284,46 @@ function open_stream(f, c::HTTPClient, path::AbstractString; "Accept" => String(accept), "User-Agent" => c.user_agent, ] - body_stream = Base.BufferStream() - err_ref = Ref{Any}(nothing) - http_task = @async try - HTTP.open(String(uppercase(string("GET"))), url, headers; - query = qpairs, - status_exception = false, - decompress = false, - retry = false, - readtimeout = c.timeout, - connect_timeout = c.connect_timeout) do http - HTTP.startread(http) - status = http.message.status - if status >= 400 - # Drain body to construct a structured error. - buf = IOBuffer() - while !eof(http); write(buf, readavailable(http)); end - rid = "" - for (k, v) in http.message.headers - lowercase(String(k)) == "request-id" && (rid = String(v)) + + attempt = 0 + while true + attempt += 1 + result_ref = Ref{Any}(nothing) + err_ref = Ref{Any}(nothing) + retry_ref = Ref(false) + retry_after = Ref{Union{Nothing,Float64}}(nothing) + + try + c.stream_opener(c, "GET", url, headers, qpairs) do status, hdrs, io + if status >= 400 + body = read(io) # drain the error body for a structured message + if _is_retryable_status(status) && attempt <= c.max_retries + retry_ref[] = true + retry_after[] = _retry_after_seconds(hdrs) + return nothing + end + T = status < 500 ? BentoClientError : BentoServerError + err_ref[] = http_error_from_response(T, status, String(body), + _request_id_from_headers(hdrs)) + return nothing end - T = status < 500 ? BentoClientError : BentoServerError - err_ref[] = http_error_from_response(T, status, String(take!(buf)), rid) - return + result_ref[] = f(io) + return nothing end - while !eof(http) - write(body_stream, readavailable(http)) + catch e + # Transport failure before/while reading the head — safe to retry. + if _is_retryable_exception(e) && attempt <= c.max_retries + _retry_wait(c, attempt, nothing) + continue end + rethrow() + end + + if retry_ref[] + _retry_wait(c, attempt, retry_after[]) + continue end - catch e - err_ref[] = e - finally - close(body_stream) - end - try - result = f(body_stream) - wait(http_task) err_ref[] === nothing || throw(err_ref[]) - return result - catch - # Make sure the HTTP task is allowed to clean up before propagating. - try; wait(http_task); catch; end - rethrow() + return result_ref[] end end diff --git a/src/live/client.jl b/src/live/client.jl index 4e659d8..4b92a8e 100644 --- a/src/live/client.jl +++ b/src/live/client.jl @@ -236,7 +236,7 @@ supervisor (see [`add_reconnect_callback`](@ref)). `subscriptions` is a vector of NamedTuples describing one subscribe call each. Each entry must have `schema` and `symbols`; `stype_in` (default -`SType.RAW_SYMBOL`), `snapshot` (default `false`), and `start` (default +`SType.RAW_SYMBOL`), `snapshot` (default `false`), and `start_dt` (default `nothing`) are optional. Any extra `kwargs` are forwarded to `Live(...)` — typically `key`, `gateway`, @@ -278,7 +278,7 @@ function live_session(fn::Function; symbols = sub.symbols, stype_in = get(sub, :stype_in, SType.RAW_SYMBOL), snapshot = get(sub, :snapshot, false), - start = get(sub, :start, nothing), + start_dt = get(sub, :start_dt, nothing), ) end start!(client) diff --git a/src/live/streaming.jl b/src/live/streaming.jl index d82f987..7b99b43 100644 --- a/src/live/streaming.jl +++ b/src/live/streaming.jl @@ -602,7 +602,7 @@ function _run_unified_session(ctxs::Dict{Schema.T,SessionContext}; symbols = symbols, stype_in = stype_in, snapshot = snapshot, - start = start_ts, + start_dt = start_ts, ) end start!(client) @@ -774,7 +774,7 @@ function stream_to_file(; schema::Schema.T, slow_reader_behavior = nothing, channel_size::Integer = 10_000, heartbeat_log_interval_s::Real = 30.0, - start = nothing, + start_dt = nothing, snapshot::Bool = false)::String base = base_dir === nothing ? joinpath(pwd(), "live") : String(base_dir) return open_dbn_writer( @@ -801,7 +801,7 @@ function stream_to_file(; schema::Schema.T, heartbeat_interval = heartbeat_interval, slow_reader_behavior = slow_reader_behavior, channel_size = channel_size, - start_initial = start, snapshot = snapshot, + start_initial = start_dt, snapshot = snapshot, ) finally monitor_stop[] = true @@ -857,7 +857,7 @@ function stream_multi_to_files(; schemas::AbstractVector, slow_reader_behavior = nothing, channel_size::Integer = 10_000, heartbeat_log_interval_s::Real = 30.0, - start::Union{Nothing,Dates.DateTime,Integer} = nothing, + start_dt::Union{Nothing,Dates.DateTime,Integer} = nothing, snapshot::Bool = false)::Dict{Schema.T,String} isempty(schemas) && throw(ArgumentError("schemas must not be empty")) base = base_dir === nothing ? joinpath(pwd(), "live") : String(base_dir) diff --git a/src/live/subscribe.jl b/src/live/subscribe.jl index a74aa04..f1ffc3f 100644 --- a/src/live/subscribe.jl +++ b/src/live/subscribe.jl @@ -1,6 +1,6 @@ """ subscribe!(client; schema, symbols, stype_in=SType.RAW_SYMBOL, - snapshot=false, start=nothing) -> Int | Channel{T} + snapshot=false, start_dt=nothing) -> Int | Channel{T} Send a subscription request to the Live gateway. May be called multiple times before [`start!`](@ref). @@ -14,18 +14,18 @@ Return type depends on the client's mode: type land together. Subscribing under typed mode to a schema with no concrete record type (e.g. `Schema.MIX`) errors. -`snapshot=true` (request an order book snapshot) is mutually exclusive with `start`. +`snapshot=true` (request an order book snapshot) is mutually exclusive with `start_dt`. """ function subscribe!(c::Live; schema::Schema.T, symbols, stype_in::SType.T = SType.RAW_SYMBOL, snapshot::Bool = false, - start::Union{Nothing,DateTime,Integer} = nothing) + start_dt::Union{Nothing,DateTime,Integer} = nothing) c.connected || throw(ArgumentError("call connect!(client) before subscribing")) c.started && throw(ArgumentError("cannot subscribe after start!")) - snapshot && start !== nothing && - throw(ArgumentError("snapshot=true and start are mutually exclusive")) + snapshot && start_dt !== nothing && + throw(ArgumentError("snapshot=true and start_dt are mutually exclusive")) # Typed mode requires every subscribed schema to have a concrete record # type. Validate up front so the user sees a clear error at subscribe! @@ -43,9 +43,9 @@ function subscribe!(c::Live; c.next_sub_id += 1 syms = symbols_str(symbols) - start_ns = start === nothing ? nothing : - (start isa Integer ? Int64(start) : - Int64(round(Dates.datetime2unix(start) * 1_000_000_000))) + start_ns = start_dt === nothing ? nothing : + (start_dt isa Integer ? Int64(start_dt) : + Int64(round(Dates.datetime2unix(start_dt) * 1_000_000_000))) write_text_frame(c.socket; schema = schema_str(schema), diff --git a/test/live/test_historical_smoke.jl b/test/live/test_historical_smoke.jl index 41df52c..49ba153 100644 --- a/test/live/test_historical_smoke.jl +++ b/test/live/test_historical_smoke.jl @@ -42,8 +42,8 @@ using Dates dataset = dataset, symbols = symbols, schema = Schema.TRADES, - start = start_s, - end_ = end_s, + start_dt = start_s, + end_dt = end_s, stype_in = stype_in) @info "cost estimate" cost @test cost !== nothing @@ -52,8 +52,8 @@ using Dates dataset = dataset, symbols = symbols, schema = Schema.TRADES, - start = start_s, - end_ = end_s, + start_dt = start_s, + end_dt = end_s, stype_in = stype_in) @info "billable size" sz @test sz !== nothing @@ -65,8 +65,8 @@ using Dates dataset = dataset, schema = Schema.TRADES, symbols = symbols, - start = start_s, - end_ = end_s, + start_dt = start_s, + end_dt = end_s, stype_in = stype_in) @test store isa DBNStore @test store.metadata.dataset == dataset diff --git a/test/test_historical_batch.jl b/test/test_historical_batch.jl index 2a324c7..f60a701 100644 --- a/test/test_historical_batch.jl +++ b/test/test_historical_batch.jl @@ -16,8 +16,8 @@ using HTTP dataset = "XNAS.ITCH", symbols = ["AAPL"], schema = Schema.TRADES, - start = "2024-01-02", - end_ = "2024-01-03") + start_dt = "2024-01-02", + end_dt = "2024-01-03") @test result["id"] == "job-123" # Body uses "end" not "end_" @test occursin("end=", captured[]) diff --git a/test/test_historical_metadata.jl b/test/test_historical_metadata.jl index 2536242..17561fc 100644 --- a/test/test_historical_metadata.jl +++ b/test/test_historical_metadata.jl @@ -52,8 +52,8 @@ end dataset = "XNAS.ITCH", symbols = ["AAPL"], schema = Schema.MBP_1, - start = "2024-01-02T14:30:00", - end_ = "2024-01-02T14:31:00") + start_dt = "2024-01-02T14:30:00", + end_dt = "2024-01-02T14:31:00") @test result["cost"] == 1.23 end end diff --git a/test/test_historical_timeseries.jl b/test/test_historical_timeseries.jl index 9a76e56..dac2089 100644 --- a/test/test_historical_timeseries.jl +++ b/test/test_historical_timeseries.jl @@ -66,8 +66,8 @@ end dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = "2024-01-02T14:30:00", - end_ = "2024-01-02T14:31:00", + start_dt = "2024-01-02T14:30:00", + end_dt = "2024-01-02T14:31:00", stype_in = SType.RAW_SYMBOL, stype_out = SType.INSTRUMENT_ID) @@ -102,8 +102,8 @@ end dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = "2024-01-02T14:30:00", - end_ = "2024-01-02T14:31:00", + start_dt = "2024-01-02T14:30:00", + end_dt = "2024-01-02T14:31:00", typed = false, ) @test store isa DBNStore @@ -124,8 +124,8 @@ end dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = "2024-01-02T14:30:00", - end_ = "2024-01-02T14:31:00", + start_dt = "2024-01-02T14:30:00", + end_dt = "2024-01-02T14:31:00", size_hint = n, ) @test length(store) == n @@ -135,8 +135,8 @@ end dataset = "XNAS.ITCH", schema = Schema.TRADES, symbols = ["AAPL"], - start = "2024-01-02T14:30:00", - end_ = "2024-01-02T14:31:00", + start_dt = "2024-01-02T14:30:00", + end_dt = "2024-01-02T14:31:00", typed = false, size_hint = 999_999, ) @test length(store2) == n @@ -206,3 +206,74 @@ end @test t_t === DBN.TradeMsg @test t_g === DBN.TradeMsg end + +# A stream_opener that replays a fixed sequence of (status, headers, body) tuples, +# one per reconnect attempt, feeding each body to the consumer as a readable IO. +# This exercises open_stream/foreach_record end-to-end without a live socket. +function _seq_opener(responses) + i = Ref(0) + # `consume` is first to match the do-block opener contract (see _default_http_stream). + return (consume, c, method, url, headers, qpairs) -> begin + i[] += 1 + status, hdrs, body = responses[i[]] + return consume(status, hdrs, IOBuffer(body)) + end +end + +const _NOSLEEP_TS = _ -> nothing + +@testset "foreach_record streams via injected opener" begin + bytes, n = _build_sample_dbn_zstd() + opener = _seq_opener([(200, Pair{String,String}[], bytes)]) + c = Historical("test-key"; gateway = "https://hist.test", stream_opener = opener) + seen = Ref(0) + md = 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 + seen[] += 1 + @test rec isa DBN.TradeMsg + end + @test seen[] == n + @test md isa DBN.Metadata + @test md.dataset == "XNAS.ITCH" +end + +@testset "foreach_record retries transient status then streams" begin + bytes, n = _build_sample_dbn_zstd() + opener = _seq_opener([ + (503, Pair{String,String}[], Vector{UInt8}("temporarily down")), + (200, Pair{String,String}[], bytes), + ]) + c = Historical("test-key"; gateway = "https://hist.test", + stream_opener = opener, retry_sleep = _NOSLEEP_TS) + seen = Ref(0) + 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 + seen[] += 1 + end + @test seen[] == n +end + +@testset "foreach_record maps 4xx → BentoClientError" begin + body = Vector{UInt8}("""{"detail":{"case":"not_found","message":"nope"}}""") + opener = _seq_opener([(404, ["request-id" => "rid-1"], body)]) + c = Historical("test-key"; gateway = "https://hist.test", stream_opener = opener) + @test_throws BentoClientError 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 +end + +@testset "foreach_record propagates a consumer exception" begin + bytes, n = _build_sample_dbn_zstd() + opener = _seq_opener([(200, Pair{String,String}[], bytes)]) + c = Historical("test-key"; gateway = "https://hist.test", stream_opener = opener) + # An exception from the consumer must propagate (open_stream's do-block tears + # the connection down on the way out — verified here via the error path). + @test_throws ErrorException 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 + error("consumer aborted") + end +end diff --git a/test/test_http.jl b/test/test_http.jl index fc7635b..1156c12 100644 --- a/test/test_http.jl +++ b/test/test_http.jl @@ -2,8 +2,13 @@ using Test 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 using HTTP +# No-op sleep so retry tests don't actually wait on the wall clock. +const _NOSLEEP = _ -> nothing + @testset "http" begin @testset "basic_auth_header" begin h = basic_auth_header("db-abc") @@ -44,14 +49,103 @@ using HTTP @test any(p -> first(p) == "Authorization", captured[].headers) end - @testset "request maps 5xx → BentoServerError" begin + @testset "request maps 5xx → BentoServerError (no retry)" begin function mock(method, url, headers, body; kwargs...) HTTP.Response(503; body = "boom") end - c = HTTPClient("k", "https://example.test"; dispatcher = mock) + c = HTTPClient("k", "https://example.test"; dispatcher = mock, max_retries = 0) @test_throws BentoServerError request(c, :POST, "/v0/foo"; body = (a = 1,)) end + @testset "retry helpers" begin + @test _is_retryable_status(429) + @test _is_retryable_status(500) + @test _is_retryable_status(503) + @test !_is_retryable_status(501) # Not Implemented is permanent + @test !_is_retryable_status(404) + @test !_is_retryable_status(200) + + # Retry-After: integer delta-seconds honored (capped), HTTP-date → nothing. + @test _retry_after_seconds(["Retry-After" => "7"]) == 7.0 + @test _retry_after_seconds(["retry-after" => "2.5"]) == 2.5 + @test _retry_after_seconds(["Retry-After" => string(RETRY_AFTER_CAP_S + 100)]) == RETRY_AFTER_CAP_S + @test _retry_after_seconds(["Retry-After" => "Wed, 21 Oct 2026 07:28:00 GMT"]) === nothing + @test _retry_after_seconds(["X-Other" => "5"]) === nothing + + # Backoff stays within the jitter envelope for every attempt. + for attempt in 1:8 + d = _retry_delay(attempt) + @test 0.0 <= d <= RETRY_CAP_S + end + end + + @testset "request retries transient status then succeeds" begin + calls = Ref(0) + function mock(method, url, headers, body; kwargs...) + calls[] += 1 + calls[] < 3 ? HTTP.Response(503; body = "down") : + HTTP.Response(200; body = """{"ok":true}""") + 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 # two 503s retried, third call succeeds + end + + @testset "request honors Retry-After on 429" begin + slept = Float64[] + calls = Ref(0) + function mock(method, url, headers, body; kwargs...) + calls[] += 1 + calls[] == 1 ? HTTP.Response(429, ["Retry-After" => "4"]; body = "slow down") : + HTTP.Response(200; body = "{}") + end + c = HTTPClient("k", "https://example.test"; + dispatcher = mock, retry_sleep = d -> push!(slept, d)) + resp = request(c, :GET, "/v0/foo") + @test resp.status == 200 + @test slept == [4.0] # Retry-After overrode the jittered backoff + end + + @testset "request exhausts retry budget then maps the final status" begin + calls = Ref(0) + function mock(method, url, headers, body; kwargs...) + calls[] += 1 + HTTP.Response(503; body = "still down") + end + c = HTTPClient("k", "https://example.test"; + dispatcher = mock, retry_sleep = _NOSLEEP, max_retries = 2) + @test_throws BentoServerError request(c, :GET, "/v0/foo") + @test calls[] == 3 # initial attempt + 2 retries + end + + @testset "request retries transient transport exception" begin + calls = Ref(0) + function mock(method, url, headers, body; kwargs...) + calls[] += 1 + calls[] < 2 && 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[] == 2 + end + + @testset "request does not retry a non-transient exception" begin + calls = Ref(0) + function mock(method, url, headers, body; kwargs...) + calls[] += 1 + throw(ArgumentError("boom")) + end + c = HTTPClient("k", "https://example.test"; + dispatcher = mock, retry_sleep = _NOSLEEP) + @test_throws ArgumentError request(c, :GET, "/v0/foo") + @test calls[] == 1 # propagated immediately, not retried + end + @testset "request returns response on 2xx" begin function mock(method, url, headers, body; kwargs...) HTTP.Response(200; body = """{"ok":true}""") diff --git a/test/test_live_subscribe.jl b/test/test_live_subscribe.jl index 010b96d..9a5d76c 100644 --- a/test/test_live_subscribe.jl +++ b/test/test_live_subscribe.jl @@ -59,7 +59,7 @@ end connect!(client) @test_throws ArgumentError subscribe!(client; schema = Schema.TRADES, symbols = "ES.FUT", - snapshot = true, start = 1) + snapshot = true, start_dt = 1) close(client) try wait(mock.accept_task) catch end end From 21aeb03cfeb97fc4c54cc5797118603b89bcb09c Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Wed, 3 Jun 2026 16:32:02 -0500 Subject: [PATCH 2/2] Fix incomplete start_dt rename and open_stream mid-body retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - streaming.jl: `stream_multi_to_files` still passed `start_initial = start` (the 12-space-indented copy my earlier replace_all missed — it only matched the 16-space copy in `stream_to_file`). On the capture path this threw `UndefVarError: start not defined`, crashing the session mid-capture and causing the live-streaming round-trip test to fail/hang. Now `start_dt`. - http.jl (open_stream): only retry transient transport errors in the pre-body phase. Once the consumer callback has started reading the body it may already have observed records, so a mid-body socket error must surface rather than replay the partial stream. Addresses Codex review (P2). Verified: full offline suite passes locally (all 19 files, including test_live_streaming, which previously hung/failed). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/http.jl | 10 ++++++++-- src/live/streaming.jl | 2 +- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/src/http.jl b/src/http.jl index b936105..f652bb9 100644 --- a/src/http.jl +++ b/src/http.jl @@ -292,6 +292,10 @@ function open_stream(f, c::HTTPClient, path::AbstractString; err_ref = Ref{Any}(nothing) retry_ref = Ref(false) retry_after = Ref{Union{Nothing,Float64}}(nothing) + # Once we hand `io` to `f`, the consumer may have already observed + # 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) try c.stream_opener(c, "GET", url, headers, qpairs) do status, hdrs, io @@ -307,12 +311,14 @@ function open_stream(f, c::HTTPClient, path::AbstractString; _request_id_from_headers(hdrs)) return nothing end + consumed[] = true result_ref[] = f(io) return nothing end catch e - # Transport failure before/while reading the head — safe to retry. - if _is_retryable_exception(e) && attempt <= c.max_retries + # Retry only transport failures in the pre-body phase (connect + + # response head). Once `f` has started reading, surface the error. + if _is_retryable_exception(e) && !consumed[] && attempt <= c.max_retries _retry_wait(c, attempt, nothing) continue end diff --git a/src/live/streaming.jl b/src/live/streaming.jl index 7b99b43..2485e3b 100644 --- a/src/live/streaming.jl +++ b/src/live/streaming.jl @@ -896,7 +896,7 @@ function stream_multi_to_files(; schemas::AbstractVector, heartbeat_interval = heartbeat_interval, slow_reader_behavior = slow_reader_behavior, channel_size = channel_size, - start_initial = start, snapshot = snapshot, + start_initial = start_dt, snapshot = snapshot, ) catch e @error "live session crashed" exception=(e, catch_backtrace())