Historical HTTP retries, backpressured foreach_record, start_dt/end_dt rename

.gitignore

@@ -21,3 +21,6 @@ config.toml
 # Benchmark artifacts
 benchmark/results/
 benchmark/data/
+
+# Live capture output dirs
+live_capture_*/

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

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.
 

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"])

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,
     )

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"]
 

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)
 

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)

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.

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)
 

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")
 ```
 """

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

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),

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) =

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,17 +101,17 @@ 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((
         dataset  = String(dataset),
         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,17 +133,17 @@ 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((
         dataset  = String(dataset),
         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,
     ))