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..f652bb9 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,52 @@ 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)
+        # 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
+                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
+                consumed[] = true
+                result_ref[] = f(io)
+                return nothing
             end
-            while !eof(http)
-                write(body_stream, readavailable(http))
+        catch e
+            # 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
+            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..2485e3b 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)
@@ -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())
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