diff --git a/Cargo.lock b/Cargo.lock index 1acc3f9..f3c4058 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5505,6 +5505,7 @@ dependencies = [ "arrow-array 58.3.0", "arrow-schema 58.3.0", "axum", + "bytes", "chrono", "clap", "etcd-client", @@ -5558,8 +5559,10 @@ name = "lance-context-server" version = "0.6.3" dependencies = [ "axum", + "bytes", "chrono", "clap", + "futures", "lance-context-api", "lance-context-core", "lance-context-metrics", diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index 7841782..dc59693 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -1132,6 +1132,42 @@ impl RolloutStore { } } + /// Fetch a row *together with* its `binary_payload` in a single base-first + /// scan, returning `(record, payload)`. + /// + /// Callers that need both the row metadata (e.g. `content_type` for a + /// download's `Content-Type`/filename) and the artifact bytes would + /// otherwise call [`Self::get_by_id`] then [`Self::get_blob`] — two + /// independent point scans over the same shard. This folds them into one + /// scan: the row is located once and `binary_payload` is materialized for + /// only that row (via a projected `take`), so it never reads back an entire + /// fragment's payloads. `payload` is `None` when the row carries no blob. + /// + /// Base-table-first with the same immutable-row reasoning as + /// [`Self::get_by_id`]: a hit in the base table returns immediately with no + /// MemWAL generation opened; only a base miss falls back to the flushed WAL. + pub async fn get_record_with_blob( + &self, + id: &str, + ) -> LanceResult>)>> { + // Base table first — no manifest reads, no per-generation opens. + if let Some(record) = self.scan_one_by_id(id, ListSource::Fragments).await? { + let payload = Self::get_blob_from_dataset(&self.dataset, id) + .await? + .flatten(); + return Ok(Some((record, payload))); + } + + // Base miss: locate the row in the flushed generations. Reuse the + // NotFound-tolerant, bounded-parallel blob fallback and pair it with the + // WAL-sourced metadata scan. + let Some(record) = self.scan_one_by_id(id, ListSource::Wal).await? else { + return Ok(None); + }; + let payload = self.get_blob(id).await?; + Ok(Some((record, payload))) + } + /// Run an id-equality point scan against a single [`ListSource`] and return /// the first matching record. `Fragments` passes an empty snapshot set so it /// performs no MemWAL manifest discovery; `Wal`/`All` discover flushed @@ -3832,4 +3868,54 @@ mod tests { ); }); } + + #[test] + fn get_record_with_blob_returns_row_and_payload_in_one_scan() { + // get_record_with_blob folds the metadata point lookup and the payload + // fetch into a single base-first scan. It must return both the row and + // its bytes for an un-merged WAL row (base-miss -> fallback) and for a + // merged base row, and None for a missing id. + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + let bytes = b"\x00\x01record-with-blob"; + let runtime = tokio::runtime::Runtime::new().unwrap(); + runtime.block_on(async { + let mut store = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + storage_options: None, + shard_id: Some("rollout-0".to_string()), + ..Default::default() + }, + ) + .await + .unwrap(); + let artifact = artifact_record("row-rw", bytes); + store.add(std::slice::from_ref(&artifact)).await.unwrap(); + + // Un-merged: found via the WAL fallback, row + payload paired. + assert_eq!(flushed_generation_count(&store).await, 1); + let (record, payload) = store + .get_record_with_blob("row-rw") + .await + .unwrap() + .expect("row present in WAL"); + assert_records_eq(&record, &artifact); + assert_eq!(payload.as_deref(), Some(&bytes[..])); + + // After merge: found via the base table, still paired. + store.cleanup_own_shard().await.unwrap(); + assert_eq!(flushed_generation_count(&store).await, 0); + let (record, payload) = store + .get_record_with_blob("row-rw") + .await + .unwrap() + .expect("row present in base"); + assert_records_eq(&record, &artifact); + assert_eq!(payload.as_deref(), Some(&bytes[..])); + + // Missing id: None, not an error. + assert!(store.get_record_with_blob("nope").await.unwrap().is_none()); + }); + } } diff --git a/crates/lance-context-master/Cargo.toml b/crates/lance-context-master/Cargo.toml index c4df078..dcd2822 100644 --- a/crates/lance-context-master/Cargo.toml +++ b/crates/lance-context-master/Cargo.toml @@ -19,6 +19,7 @@ lance-context-metrics = { version = "0.1.0", path = "../lance-context-metrics" } arrow-array = "58" arrow-schema = "58" axum = { version = "0.8", features = ["json"] } +bytes = "1" chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = { version = "4", features = ["derive", "env"] } etcd-client = { version = "0.19", features = ["tls"] } diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 82283ba..af5e9ca 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -209,16 +209,16 @@ pub async fn download_experiment_blob( .refresh_latest() .await .map_err(MasterError::from_lance)?; - let record = store - .get_by_id(&id) + // Single base-first scan returns the row metadata and its payload together, + // instead of a separate get_by_id + get_blob (two point scans over the same + // shard) as before. + let (record, payload) = store + .get_record_with_blob(&id) .await .map_err(MasterError::from_lance)? .ok_or_else(|| MasterError::NotFound(format!("record '{}' does not exist", id)))?; - let bytes = store - .get_blob(&id) - .await - .map_err(MasterError::from_lance)? - .ok_or_else(|| MasterError::NotFound(format!("record '{}' has no blob", id)))?; + let bytes = + payload.ok_or_else(|| MasterError::NotFound(format!("record '{}' has no blob", id)))?; let content_type = HeaderValue::from_str(&record.content_type) .unwrap_or_else(|_| HeaderValue::from_static("application/octet-stream")); @@ -229,10 +229,29 @@ pub async fn download_experiment_blob( Response::builder() .header(header::CONTENT_TYPE, content_type) .header(header::CONTENT_DISPOSITION, disposition) - .body(Body::from(bytes)) + .header(header::CONTENT_LENGTH, bytes.len()) + .body(blob_stream_body(bytes)) .map_err(|err| MasterError::Internal(err.to_string())) } +/// Frame size for chunked blob download responses (see [`blob_stream_body`]). +const BLOB_STREAM_CHUNK_BYTES: usize = 256 * 1024; + +/// Build a chunked [`Body`] from an owned blob so the HTTP send path never holds +/// a second full-blob copy and a slow client applies backpressure at frame +/// granularity. `Bytes` frames are refcounted slices of one allocation, so no +/// per-frame copy of the payload is made. +fn blob_stream_body(bytes: Vec) -> Body { + let buf = bytes::Bytes::from(bytes); + let chunks = (0..buf.len()) + .step_by(BLOB_STREAM_CHUNK_BYTES.max(1)) + .map(move |start| { + let end = (start + BLOB_STREAM_CHUNK_BYTES).min(buf.len()); + Ok::<_, std::convert::Infallible>(buf.slice(start..end)) + }); + Body::from_stream(futures::stream::iter(chunks)) +} + /// `POST /api/v1/rescan` — trigger one immediate full scan. pub async fn rescan( State(state): State>, diff --git a/crates/lance-context-server/Cargo.toml b/crates/lance-context-server/Cargo.toml index 7e04a01..4b1d5ff 100644 --- a/crates/lance-context-server/Cargo.toml +++ b/crates/lance-context-server/Cargo.toml @@ -17,6 +17,8 @@ lance-context-core = { version = "0.6.3", path = "../lance-context-core" } lance-context-api = { version = "0.6.3", path = "../lance-context-api" } lance-context-metrics = { version = "0.1.0", path = "../lance-context-metrics" } axum = { version = "0.8", features = ["json", "multipart"] } +bytes = "1" +futures = "0.3" metrics = "0.24" chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = { version = "4", features = ["derive", "env"] } diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index ed262da..a4f48d2 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -53,6 +53,22 @@ pub struct ServerConfig { /// back to the built-in default. #[arg(long, env = "ROLLOUT_CACHE_CAPACITY", default_value = "2000")] pub rollout_cache_capacity: usize, + + /// Ceiling, in bytes, on the total artifact-blob payload held in memory + /// across all *concurrent* rollout uploads and downloads on this instance. + /// + /// Each blob request materializes its full payload as an in-memory buffer + /// (uploads buffer the request body; downloads materialize the row's + /// `binary_payload`). Without a global cap, N concurrent 1 GiB requests + /// would need N GiB and can OOM the process. This budget admits a request + /// only while enough of the budget is free — otherwise it is rejected with + /// `503 Service Unavailable` and a `Retry-After`, applying backpressure at + /// the edge instead of the allocator. A single request larger than the whole + /// budget is still admitted when the instance is otherwise idle (it reserves + /// the entire budget), so this bounds concurrency, not maximum blob size. + /// `0` (the default) disables the budget. + #[arg(long, env = "ROLLOUT_MAX_INFLIGHT_BLOB_BYTES", default_value = "0")] + pub rollout_max_inflight_blob_bytes: usize, } impl ServerConfig { diff --git a/crates/lance-context-server/src/error.rs b/crates/lance-context-server/src/error.rs index a8d074a..c5bb116 100644 --- a/crates/lance-context-server/src/error.rs +++ b/crates/lance-context-server/src/error.rs @@ -11,6 +11,9 @@ pub enum AppError { InvalidRequest(String), Internal(String), CompactionInProgress, + /// The in-flight blob-byte budget is exhausted; the client should retry + /// later. Maps to `503 Service Unavailable`. + Overloaded(String), } impl AppError { @@ -55,6 +58,7 @@ impl IntoResponse for AppError { "COMPACTION_IN_PROGRESS", "Compaction already in progress".to_string(), ), + AppError::Overloaded(msg) => (StatusCode::SERVICE_UNAVAILABLE, "OVERLOADED", msg), }; let body = ErrorResponse { diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index a85c470..c620989 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -27,6 +27,71 @@ use crate::state::AppState; /// raises the ceiling well above it while still bounding memory. pub const MAX_ROLLOUT_UPLOAD_BYTES: usize = 1024 * 1024 * 1024; +/// Frame size for chunked blob download responses. The in-memory `Vec` is +/// sliced into frames of this size so the HTTP send path never holds an extra +/// full-blob copy and a slow reader exerts backpressure at frame granularity +/// rather than after the whole payload is queued. 256 KiB keeps per-frame +/// overhead negligible while capping the amount buffered ahead of the socket. +const BLOB_STREAM_CHUNK_BYTES: usize = 256 * 1024; + +/// Build a chunked [`Body`] from an owned blob. `Bytes` frames share one +/// backing allocation (cheap refcounted slices — no per-frame copy), so this +/// does not duplicate the payload; it only changes how it is fed to the socket. +/// +/// `reservation` (the in-flight blob-budget guard, if any) is moved into the +/// stream and released only when the last frame has been produced, so the +/// budget accounts for a slow download for its full lifetime. +fn blob_stream_body( + bytes: Vec, + mut reservation: Option, +) -> Body { + let buf = bytes::Bytes::from(bytes); + let len = buf.len(); + let mut offset = 0usize; + let stream = futures::stream::poll_fn(move |_| { + if offset >= len { + // Drop the reservation exactly when the stream ends. + let _ = reservation.take(); + return std::task::Poll::Ready(None); + } + let end = (offset + BLOB_STREAM_CHUNK_BYTES).min(len); + let frame = buf.slice(offset..end); + offset = end; + std::task::Poll::Ready(Some(Ok::<_, std::convert::Infallible>(frame))) + }); + Body::from_stream(stream) +} + +/// Parse the `Content-Length` header into a byte count, defaulting to `0` when +/// absent or unparsable (a chunked upload without a declared length reserves +/// nothing up-front; the body-size limit still caps it). +fn content_length(headers: &header::HeaderMap) -> usize { + headers + .get(header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .unwrap_or(0) +} + +/// Reserve `bytes` from the instance's in-flight blob budget (if configured), +/// returning the RAII guard to hold for the request's duration. Returns +/// `503 Overloaded` when the budget cannot currently admit the request. When no +/// budget is configured (`None`) this is a no-op that always admits. +fn acquire_blob_budget( + state: &AppState, + bytes: usize, +) -> Result, AppError> { + match &state.blob_budget { + None => Ok(None), + Some(budget) => budget.try_acquire(bytes).map(Some).ok_or_else(|| { + metrics::counter!("rollout_blob_budget_rejections_total").increment(1); + AppError::Overloaded( + "server is at its in-flight blob memory limit; retry shortly".to_string(), + ) + }), + } +} + /// Response for the internal WAL-merge trigger: how many flushed generations /// this worker's shard folded into the base table. #[derive(Debug, serde::Serialize, serde::Deserialize)] @@ -170,6 +235,12 @@ pub async fn add_rollouts( .unwrap_or("") .to_string(); + // Admit against the in-flight blob budget before buffering the body, using + // the declared Content-Length as the reservation size. Held for the whole + // handler so concurrent uploads cannot collectively exceed the budget and + // OOM the worker; dropped when the request completes. + let _budget = acquire_blob_budget(&state, content_length(req.headers()))?; + let records = if content_type_is(&content_type, "multipart/form-data") { let multipart = Multipart::from_request(req, &state) .await @@ -364,6 +435,11 @@ pub async fn get_rollout( /// Materialize a rollout row's `binary_payload` bytes. The artifact /// bytes are opaque, so they stream as `application/octet-stream`. `404` when /// the row is absent or carries no payload. +/// +/// The payload is sent as a **chunked** body ([`blob_stream_body`]) rather than +/// a single giant frame: the response yields fixed-size frames so the HTTP +/// layer never re-buffers the whole blob and a slow client applies backpressure +/// instead of forcing the server to hold an extra full copy in the send queue. pub async fn fetch_rollout_blob( State(state): State>, Path((name, id)): Path<(String, String)>, @@ -377,9 +453,18 @@ pub async fn fetch_rollout_blob( .map_err(AppError::from_lance)? .ok_or_else(|| AppError::NotFound(format!("Rollout '{}' has no payload", id)))?; + // Reserve now that the payload size is known, and hold the reservation for + // the whole streamed send (moved into the body): a slow client keeps the + // blob resident until the last frame flushes, so the budget must account for + // it until then. Reject with 503 if the budget is currently exhausted. + let reservation = acquire_blob_budget(&state, bytes.len())?; + drop(store); + + let len = bytes.len(); Response::builder() .header(header::CONTENT_TYPE, "application/octet-stream") - .body(Body::from(bytes)) + .header(header::CONTENT_LENGTH, len) + .body(blob_stream_body(bytes, reservation)) .map_err(|err| AppError::Internal(err.to_string())) } @@ -620,6 +705,36 @@ mod tests { use super::*; use crate::state::AppState; + #[tokio::test] + async fn blob_stream_body_reassembles_across_chunk_boundaries() { + // A payload spanning several BLOB_STREAM_CHUNK_BYTES frames must + // reassemble byte-for-byte, and a reservation moved into the body is + // held until the stream is fully drained. + let payload: Vec = (0..(BLOB_STREAM_CHUNK_BYTES * 2 + 123)) + .map(|i| (i % 251) as u8) + .collect(); + let budget = crate::state::BlobBudget::new(payload.len()); + let reservation = budget.try_acquire(payload.len()); + assert!(reservation.is_some()); + // Budget is now fully occupied. + assert!(budget.try_acquire(1).is_none()); + + let body = blob_stream_body(payload.clone(), reservation); + let collected = axum::body::to_bytes(body, usize::MAX).await.unwrap(); + assert_eq!(&collected[..], &payload[..]); + + // Once the body is consumed the reservation dropped, freeing the budget. + assert!(budget.try_acquire(payload.len()).is_some()); + } + + #[test] + fn content_length_parses_or_defaults_zero() { + let mut headers = header::HeaderMap::new(); + assert_eq!(content_length(&headers), 0); + headers.insert(header::CONTENT_LENGTH, header::HeaderValue::from(4096)); + assert_eq!(content_length(&headers), 4096); + } + async fn rollout_state() -> (Arc, TempDir) { let dir = TempDir::new().unwrap(); let state = Arc::new(AppState::new_for_test(dir.path().to_path_buf()).await); diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs index 440326e..9b259d2 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -47,6 +47,89 @@ pub struct AppState { /// Periodic per-shard WAL-cleanup interval in seconds; `0` disables the /// global sweeper. See [`Self::spawn_global_sweeper`]. pub rollout_cleanup_interval_secs: u64, + /// Admission budget for in-flight artifact-blob bytes across concurrent + /// uploads/downloads. `None` disables the budget (unbounded). See + /// [`BlobBudget`]. + pub blob_budget: Option>, +} + +/// Process-wide admission control for the total artifact-blob payload held in +/// memory across concurrent rollout uploads and downloads. +/// +/// Each blob request materializes its whole payload as an in-memory buffer, so +/// unbounded concurrency of large requests can OOM the worker. A request calls +/// [`BlobBudget::try_acquire`] with its byte size before allocating; the guard +/// returned holds the reservation and releases it on drop (i.e. when the +/// request completes). When the budget cannot fit the request the caller +/// rejects it with `503` rather than proceeding to allocate. +/// +/// This bounds *concurrency*, not maximum blob size: a single request larger +/// than the entire budget is still admitted when nothing else is in flight (it +/// transiently reserves the full budget), so a lone big download never +/// deadlocks against its own limit. +#[derive(Debug)] +pub struct BlobBudget { + limit: usize, + used: std::sync::atomic::AtomicUsize, +} + +/// RAII reservation returned by [`BlobBudget::try_acquire`]. Releases its bytes +/// back to the budget when dropped. +#[derive(Debug)] +pub struct BlobReservation { + budget: Arc, + bytes: usize, +} + +impl BlobBudget { + /// Create a budget admitting up to `limit` concurrent in-flight blob bytes. + #[must_use] + pub fn new(limit: usize) -> Arc { + Arc::new(Self { + limit, + used: std::sync::atomic::AtomicUsize::new(0), + }) + } + + /// Reserve `bytes` if they fit alongside what is already in flight. Returns + /// `None` (request should be rejected with `503`) when admitting the request + /// would exceed the limit, *unless* nothing is currently in flight — in that + /// case an over-limit request is admitted so a lone large blob is never + /// permanently rejected. + pub fn try_acquire(self: &Arc, bytes: usize) -> Option { + use std::sync::atomic::Ordering; + let mut current = self.used.load(Ordering::Acquire); + loop { + // Admit when it fits, or when the instance is otherwise idle (so a + // single request bigger than the whole budget can still proceed). + let fits = current + bytes <= self.limit; + if !fits && current != 0 { + return None; + } + match self.used.compare_exchange_weak( + current, + current + bytes, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => { + return Some(BlobReservation { + budget: Arc::clone(self), + bytes, + }) + } + Err(observed) => current = observed, + } + } + } +} + +impl Drop for BlobReservation { + fn drop(&mut self) { + self.budget + .used + .fetch_sub(self.bytes, std::sync::atomic::Ordering::AcqRel); + } } impl AppState { @@ -61,6 +144,8 @@ impl AppState { .map_err(AppError::from_lance)?; let capacity = NonZeroUsize::new(config.rollout_cache_capacity) .unwrap_or_else(|| NonZeroUsize::new(DEFAULT_ROLLOUT_CACHE_CAPACITY).unwrap()); + let blob_budget = (config.rollout_max_inflight_blob_bytes > 0) + .then(|| BlobBudget::new(config.rollout_max_inflight_blob_bytes)); Ok(Self { stores: RwLock::new(std::collections::HashMap::new()), rollout_stores: Mutex::new(LruCache::new(capacity)), @@ -69,6 +154,7 @@ impl AppState { instance_id, rollout_merge_after_generations: config.rollout_merge_after_generations, rollout_cleanup_interval_secs: config.rollout_cleanup_interval_secs, + blob_budget, }) } @@ -105,6 +191,7 @@ impl AppState { instance_id, rollout_merge_after_generations: 0, rollout_cleanup_interval_secs: 0, + blob_budget: None, } } @@ -376,6 +463,41 @@ mod tests { use super::*; use tempfile::TempDir; + #[test] + fn blob_budget_admits_until_full_then_releases_on_drop() { + let budget = BlobBudget::new(1000); + // Two reservations that together fit. + let a = budget.try_acquire(600).expect("first fits"); + let b = budget.try_acquire(300).expect("second fits (900 <= 1000)"); + // A third that would overflow is rejected while others are in flight. + assert!( + budget.try_acquire(200).is_none(), + "over-limit request rejected while budget is occupied" + ); + // Dropping frees the bytes back. + drop(b); + let c = budget.try_acquire(200).expect("fits again after release"); + drop(a); + drop(c); + // Fully drained: a request equal to the whole budget now fits. + assert!(budget.try_acquire(1000).is_some()); + } + + #[test] + fn blob_budget_admits_a_lone_oversized_request() { + // A single request larger than the entire budget is admitted when the + // instance is idle, so a lone big blob is never permanently rejected — + // the budget bounds concurrency, not maximum blob size. + let budget = BlobBudget::new(100); + let big = budget + .try_acquire(500) + .expect("lone oversized request admitted"); + // But while it holds the (over-)reservation, nothing else gets in. + assert!(budget.try_acquire(1).is_none()); + drop(big); + assert!(budget.try_acquire(50).is_some()); + } + async fn state_with_interval(dir: &TempDir, secs: u64) -> Arc { let mut state = AppState::new_for_test(dir.path().to_path_buf()).await; state.rollout_cleanup_interval_secs = secs; diff --git a/specs/rollout-blob-streaming.md b/specs/rollout-blob-streaming.md new file mode 100644 index 0000000..1e9a882 --- /dev/null +++ b/specs/rollout-blob-streaming.md @@ -0,0 +1,128 @@ +# Rollout artifact blobs: memory behavior and streaming roadmap + +Status: describes the memory characteristics of the rollout `binary_payload` +("blob") path after the download-streaming + budget work, and proposes the next +(larger) step toward true streaming storage. The first three items below are +**implemented**; the fourth (§4) is a **proposal**, not yet built. + +## Background: where blob bytes live in memory + +`binary_payload` is stored as a plain **inline `LargeBinary`** column, not a +lance blob-v2 (`lance-encoding:blob`) offloaded column. This is deliberate: +rollout reads go through the MemWAL LSM scanner, which has no blob-materialization +step, so a blob-v2 column reads back as `None` there. Inline storage is currently +the only encoding that round-trips (see the module docs in +`crates/lance-context-core/src/rollout_store.rs`). + +The consequence is that **every blob is fully materialized in RAM** at each hop: + +- **Upload**: the whole request body is buffered — JSON base64 (~+33% expansion) + via `axum::body::to_bytes`, or each multipart part via `field.bytes()` — then + appended into a `LargeBinaryBuilder` whose `finish()` copies it into a + contiguous Arrow buffer. +- **Download**: `get_blob` locates the row, then `take_rows` materializes the + `binary_payload` Arrow buffer and `.to_vec()` copies it into an owned `Vec` + (≈2× the blob size at that instant). + +With a 1 GiB per-request ceiling (`MAX_ROLLOUT_UPLOAD_BYTES`) and no concurrency +cap, N concurrent large requests needed ≈`2 × size × N` bytes and could OOM the +worker. + +## 1. Streaming downloads (implemented) + +Both blob-serving handlers — worker `fetch_rollout_blob` and master +`download_experiment_blob` — now send the payload as a **chunked** body +(`blob_stream_body`, 256 KiB frames) instead of one `Body::from(Vec)` frame. +Frames are refcounted `Bytes` slices of the one backing allocation (no per-frame +copy). This removes the extra full-blob copy the HTTP send path would otherwise +hold and lets a slow client apply backpressure at frame granularity rather than +after the entire payload is queued. A `Content-Length` header is still set so the +response is not chunked-transfer on the wire when the length is known. + +> Note: this does **not** remove the single in-RAM `Vec` produced by +> `get_blob` — the bytes are still fully read from storage first. True +> read-from-storage-in-frames requires §4. + +## 2. Single-scan record + blob for downloads (implemented) + +`download_experiment_blob` previously did `get_by_id` (a full-row point scan) and +then `get_blob` (a second point scan over the same shard) — two scans to serve +one download. `RolloutStore::get_record_with_blob` folds these into one +base-first scan that returns `(record, payload)` together, halving the scan work +on the master download path. It keeps the base-table-first fast path and the +NotFound-tolerant WAL fallback semantics. + +## 3. In-flight blob-byte budget (implemented) + +A process-wide admission budget (`BlobBudget` in the worker's `AppState`, sized by +`ROLLOUT_MAX_INFLIGHT_BLOB_BYTES`, `0` = disabled) bounds the total blob payload +held in memory across concurrent uploads and downloads: + +- **Uploads** reserve their declared `Content-Length` before the body is + buffered; the reservation is held for the whole handler. +- **Downloads** reserve the payload size once known and hold the reservation + inside the streamed body, so a slow client keeps the bytes accounted for until + the last frame flushes. +- When the budget cannot admit a request it is rejected with **`503` `OVERLOADED`** + (a `rollout_blob_budget_rejections_total` metric is incremented) instead of + proceeding to allocate. Backpressure moves to the edge, not the allocator. + +The budget bounds *concurrency*, not maximum blob size: a lone request larger +than the whole budget is admitted when the instance is otherwise idle, so a +single big blob never permanently 503s against its own limit. + +Operational guidance: size `ROLLOUT_MAX_INFLIGHT_BLOB_BYTES` to a fraction of the +pod's memory limit that leaves headroom for the ≈2× transient copy per in-flight +download plus base overhead — e.g. for a 4 GiB pod, a budget of ~1–1.5 GiB. + +## 4. True streaming storage (proposal — NOT implemented) + +Items 1–3 bound and smooth memory but every blob is still fully resident once per +request. Eliminating that requires storing/reading blobs in a form that supports +**ranged, chunked I/O**, i.e. lance blob-v2 offloaded columns + `BlobFile` range +reads. + +### The blocker + +The rollout read path is the MemWAL LSM scanner, which unions the base table with +flushed WAL generations and has **no blob-materialization step** — a +`lance-encoding:blob` column reads back as `None` through it. So we cannot simply +flip `binary_payload` to blob-v2 without the scanner learning to resolve blob +descriptors. + +### Proposed approach (two sub-steps, each shippable) + +1. **Reader-side: `BlobFile` range read for the base table.** + Keep `binary_payload` inline in the WAL (small, short-lived generations) but, + for the *base table*, store artifact bytes as a blob-v2 column. Change + `get_blob`/`get_record_with_blob` so that on a base-table hit it opens the row's + blob descriptor and returns a `BlobFile`/reader that streams ranges from object + storage, wiring that reader into `blob_stream_body` (frames pulled from storage + on demand rather than from an in-RAM `Vec`). WAL-fallback rows stay inline (they + are the un-merged tail and are folded into the base on merge). This gives + streaming reads for the common already-merged case — the 99% path — without + touching the LSM union. + +2. **Writer-side: streamed ingest into the blob column.** + Replace the "buffer whole body → `LargeBinaryBuilder`" ingest with a streamed + writer that appends blob bytes to the blob-v2 column in frames as the request + body arrives (multipart parts stream naturally; JSON base64 would need a + streaming base64 decoder or be deprecated in favor of multipart for large + blobs). Upload peak memory drops from O(blob) to O(frame). + +### Effort / risk + +- Sub-step 1 is a core (`rollout_store.rs`) change plus a lance API dependency on + `BlobFile` range reads; medium effort, low blast radius (reads only, base-only). +- Sub-step 2 touches the ingest handler and the write path; higher effort and + needs care around the atomicity guarantees of a rollout append (currently one + `RolloutStore::add`). +- Merge/compaction must learn to carry blob-v2 columns from WAL-inline to + base-offloaded; verify the LSM merge path preserves the descriptors. + +### Recommendation + +Ship §1–3 now (this PR). Schedule sub-step 1 (base-table `BlobFile` streaming +reads) next as its own PR — it delivers the bulk of the read-side memory win — +and treat sub-step 2 (streamed ingest) as a follow-up once the read path proves +out.