diff --git a/src/api/router/snapshot_chunks_bounded_tests.rs b/src/api/router/snapshot_chunks_bounded_tests.rs new file mode 100644 index 00000000..a32add5c --- /dev/null +++ b/src/api/router/snapshot_chunks_bounded_tests.rs @@ -0,0 +1,472 @@ +use std::io; + +use tokio::{sync::Notify, time::timeout}; + +use super::*; +use crate::orbit_api::error::IoOrbitError; + +#[derive(Clone)] +pub(super) struct ChunkFault { + pub(super) oid: String, + pub(super) size: u64, + pattern: Bytes, + mode: RangeMode, + requests: Arc>>, +} + +#[derive(Clone)] +enum RangeMode { + Good, + WrongMeta, + WrongDigest, + Truncated, + TooLong, + LateError, + Unsupported, + WrongOffset, + Missing, + Held { + entered: Arc, + release: Arc, + drops: Arc, + }, +} + +struct DropCount(Arc); +impl Drop for DropCount { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +impl ChunkFault { + pub(super) fn full_stream(self) -> ObjectByteStream { + Box::pin(futures::stream::unfold( + (self.pattern, 0u64, self.size), + |(pattern, offset, size)| async move { + if offset == size { + return None; + } + let len = (size - offset).min(CHUNK_SIZE as u64) as usize; + let bytes = pattern.slice(..len); + Some((Ok(bytes), (pattern, offset + len as u64, size))) + }, + )) + } + + pub(super) fn range_stream( + self, + start: u64, + end: u64, + ) -> OrbitResult> { + self.requests + .lock() + .unwrap() + .push((self.oid.clone(), start, end)); + if matches!(self.mode, RangeMode::Unsupported) { + return Ok(None); + } + if matches!(self.mode, RangeMode::WrongOffset) { + return Err(io::Error::new(io::ErrorKind::InvalidData, "wrong backend range").into()); + } + if matches!(self.mode, RangeMode::Missing) { + return Err(IoOrbitError::object_store_not_found("fixed raw missing")); + } + assert!(start < end && end <= self.size && end - start <= CHUNK_SIZE as u64); + assert_eq!(start % CHUNK_SIZE as u64, 0); + let len = (end - start) as usize; + let raw = self.pattern.slice(..len); + let meta = ObjectMeta { + size: if matches!(self.mode, RangeMode::WrongMeta) { + self.size as i64 + 1 + } else { + self.size as i64 + }, + ..Default::default() + }; + let stream: ObjectByteStream = match self.mode { + RangeMode::WrongDigest => { + Box::pin(futures::stream::iter([Ok(Bytes::from(vec![0; len]))])) + } + RangeMode::Truncated => Box::pin(futures::stream::iter([Ok(raw.slice(..len - 1))])), + RangeMode::TooLong => Box::pin(futures::stream::iter([ + Ok(raw), + Ok(Bytes::from_static(b"!")), + ])), + RangeMode::LateError => Box::pin(futures::stream::iter([ + Ok(raw), + Err(io::Error::other("late range error")), + ])), + RangeMode::Held { + entered, + release, + drops, + } => Box::pin(futures::stream::unfold( + (Some(raw), entered, release, DropCount(drops)), + |(raw, entered, release, owner)| async move { + let raw = raw?; + entered.notify_one(); + release.notified().await; + Some((Ok(raw), (None, entered, release, owner))) + }, + )), + _ => Box::pin(futures::stream::iter([Ok(raw)])), + }; + Ok(Some((stream, meta))) + } +} + +async fn oid_for(fixture: &Fixture, path: &str) -> String { + let handler = MonoApiService::from(&fixture.state); + let main = fixture + .state + .storage + .mono_storage() + .get_main_ref("/project") + .await + .unwrap() + .unwrap(); + let tree = handler.get_tree_by_hash(&main.ref_tree_hash).await.unwrap(); + match resolve_abs_metadata(&handler, &tree, path).await.unwrap() { + MetadataWalkOutcome::FoundFile { oid, .. } => oid, + other => panic!("fixture did not resolve: {other:?}"), + } +} + +async fn set_fact(fixture: &Fixture, oid: &str, size: u64, digest: [u8; 32]) { + let db = fixture.state.storage.mono_storage().get_connection(); + let fact = mst2_verified_object::Entity::find() + .filter(mst2_verified_object::Column::GitOid.eq(oid)) + .one(db) + .await + .unwrap() + .unwrap(); + let mut fact = fact.into_active_model(); + fact.size = Set(size as i64); + fact.raw_sha256 = Set(digest.to_vec()); + fact.update(db).await.unwrap(); +} + +fn request(path: &str, digest: [u8; 32], map_id: &str, index: u64) -> Value { + json!({"items":[{"path":path,"expected_digest":format!("sha256:{}",hex_of(&digest)),"map_id":map_id,"chunk_index":index.to_string()}],"encoding":"identity"}) +} + +async fn assert_chunk(response: Response, body: &Value, raw: &[u8], index: u64) { + assert_eq!(response.status(), 200); + let wire = to_bytes(response.into_body(), 2 * 1024 * 1024) + .await + .unwrap(); + let frames = parse_stream(&wire).unwrap(); + let [Frame::Chunk(chunk), Frame::End(end)] = frames.as_slice() else { + panic!("one CHUNK and END required"); + }; + assert_eq!(chunk.chunk_bytes, raw); + assert_eq!(chunk.chunk_index, index); + assert_eq!(end.request_item_count, 1); + assert_eq!(end.unique_unit_count, 1); + assert_eq!(end.logical_bytes, raw.len() as u64); + assert_eq!( + end.request_body_sha256, + <[u8; 32]>::from(Sha256::digest(body.to_string().as_bytes())) + ); +} + +#[tokio::test] +async fn mst2_large_chunk_uses_current_oid_strict_range_faults_cancel_retry_and_lease() { + let fixture = Fixture::new_with_pg_config_directories_and_objects( + false, + 0, + &[("other".to_string(), vec![19; 4096])], + ) + .await; + let mut pattern = vec![51; CHUNK_SIZE as usize]; + let seed = uuid::Uuid::new_v4(); + pattern[..16].copy_from_slice(seed.as_bytes()); + let pattern = Bytes::from(pattern); + let size = 512 * CHUNK_SIZE as u64 + 7; + let mut hash = Sha256::new(); + for _ in 0..512 { + hash.update(&pattern); + } + hash.update(&pattern[..7]); + let digest: [u8; 32] = hash.finalize().into(); + let other = oid_for(&fixture, "/other").await; + let requests = Arc::new(std::sync::Mutex::new(Vec::new())); + for oid in [&fixture.oid, &other] { + set_fact(&fixture, oid, size, digest).await; + fixture + .counts + .chunk_faults + .lock() + .unwrap() + .push(ChunkFault { + oid: oid.clone(), + size, + pattern: pattern.clone(), + mode: RangeMode::Good, + requests: requests.clone(), + }); + } + fixture.counts.reset(); + let map = fixture.map("/file").await; + assert_eq!(map["map"]["file_size"], size.to_string()); + fixture.counts.assert(1, size as usize); + let map_id = map["map"]["map_id"].as_str().unwrap(); + fixture.counts.reset(); + // Equal content map sharing never carries the first source's OID. + let body = request("/other", digest, map_id, 512); + assert_chunk( + fixture + .send("POST", "chunks", Body::from(body.to_string())) + .await, + &body, + &pattern[..7], + 512, + ) + .await; + assert_eq!( + requests.lock().unwrap().as_slice(), + &[(other.clone(), 512 * CHUNK_SIZE as u64, size)] + ); + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.range.load(Ordering::SeqCst), 1); + assert_eq!(fixture.counts.bytes.load(Ordering::SeqCst), 7); + for (index, encoding) in [(0, "identity"), (511, "zstd")] { + let mut full = request("/other", digest, map_id, index); + full["encoding"] = json!(encoding); + fixture.counts.reset(); + assert_chunk( + fixture + .send("POST", "chunks", Body::from(full.to_string())) + .await, + &full, + &pattern, + index, + ) + .await; + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.range.load(Ordering::SeqCst), 1); + assert_eq!( + fixture.counts.bytes.load(Ordering::SeqCst), + CHUNK_SIZE as usize + ); + assert_eq!( + requests.lock().unwrap().last().unwrap(), + &( + other.clone(), + index * CHUNK_SIZE as u64, + (index + 1) * CHUNK_SIZE as u64 + ) + ); + } + for (mode, status, code, bytes) in [ + (RangeMode::WrongMeta, 502, "INTEGRITY_ERROR", 0), + (RangeMode::WrongDigest, 502, "INTEGRITY_ERROR", 7), + (RangeMode::Truncated, 502, "INTEGRITY_ERROR", 6), + (RangeMode::TooLong, 502, "INTEGRITY_ERROR", 8), + (RangeMode::LateError, 503, "OBJECT_UNAVAILABLE", 7), + (RangeMode::Unsupported, 400, "RANGE_NOT_SUPPORTED", 0), + (RangeMode::WrongOffset, 502, "INTEGRITY_ERROR", 0), + (RangeMode::Missing, 503, "OBJECT_UNAVAILABLE", 0), + ] { + fixture.counts.chunk_faults.lock().unwrap()[1].mode = mode; + fixture.counts.reset(); + error( + fixture + .send("POST", "chunks", Body::from(body.to_string())) + .await, + status, + code, + false, + ) + .await; + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.range.load(Ordering::SeqCst), 1); + assert_eq!(fixture.counts.bytes.load(Ordering::SeqCst), bytes); + } + fixture.counts.chunk_faults.lock().unwrap()[1].mode = RangeMode::LateError; + let first = request("/file", digest, map_id, 0); + let batch = json!({"items":[first["items"][0],body["items"][0]],"encoding":"identity"}); + fixture.counts.reset(); + error( + fixture + .send("POST", "chunks", Body::from(batch.to_string())) + .await, + 503, + "OBJECT_UNAVAILABLE", + false, + ) + .await; + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.range.load(Ordering::SeqCst), 2); + assert_eq!( + fixture.counts.bytes.load(Ordering::SeqCst), + CHUNK_SIZE as usize + 7 + ); + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let drops = Arc::new(AtomicUsize::new(0)); + fixture.counts.chunk_faults.lock().unwrap()[1].mode = RangeMode::Held { + entered: entered.clone(), + release: release.clone(), + drops: drops.clone(), + }; + let task = tokio::spawn(fixture.app.clone().oneshot(fixture.request( + "POST", + "chunks", + Body::from(body.to_string()), + ))); + timeout(Duration::from_secs(10), entered.notified()) + .await + .unwrap(); + task.abort(); + assert!(task.await.err().unwrap().is_cancelled()); + assert_eq!(drops.load(Ordering::SeqCst), 1); + fixture.counts.chunk_faults.lock().unwrap()[1].mode = RangeMode::Good; + assert_chunk( + fixture + .send("POST", "chunks", Body::from(body.to_string())) + .await, + &body, + &pattern[..7], + 512, + ) + .await; + // Revocation while the range is held is a pre-header 410 JSON response. + fixture.counts.chunk_faults.lock().unwrap()[1].mode = RangeMode::Held { + entered: entered.clone(), + release: release.clone(), + drops: drops.clone(), + }; + let task = tokio::spawn(fixture.app.clone().oneshot(fixture.request( + "POST", + "chunks", + Body::from(body.to_string()), + ))); + timeout(Duration::from_secs(10), entered.notified()) + .await + .unwrap(); + let revoked = fixture + .app + .clone() + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/api/v2/snapshots/leases/{}", fixture.lease)) + .header("authorization", format!("Bearer {TOKEN}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(revoked.status(), 200); + release.notify_one(); + error( + timeout(Duration::from_secs(10), task) + .await + .unwrap() + .unwrap() + .unwrap(), + 410, + "LEASE_EXPIRED", + false, + ) + .await; + assert_eq!(drops.load(Ordering::SeqCst), 2); +} + +#[tokio::test] +async fn mst2_chunk_batch_live_budget_and_invalid_later_path_reject_before_body_io() { + let fixture = Fixture::new_with_pg_config_directories_and_objects( + false, + 0, + &[ + ("one".to_string(), vec![11; 4096]), + ("two".to_string(), vec![12; 4096]), + ], + ) + .await; + let mut items = Vec::new(); + for (index, path) in ["/file", "/one", "/two"].iter().enumerate() { + let oid = oid_for(&fixture, path).await; + let digest = [index as u8 + 1; 32]; + set_fact(&fixture, &oid, 512 * CHUNK_SIZE as u64, digest).await; + items.push( + request( + path, + digest, + &format!("sha256:{}", hex_of(&[index as u8 + 1; 32])), + 0, + )["items"][0] + .clone(), + ); + } + fixture.counts.reset(); + error( + fixture + .send( + "POST", + "chunks", + Body::from(json!({"items":items,"encoding":"identity"}).to_string()), + ) + .await, + 413, + "LIMIT_EXCEEDED", + false, + ) + .await; + fixture.counts.assert(0, 0); + let fixture = Fixture::new().await; + let mut body = fixture.chunk_body("/file", &format!("sha256:{}", hex_of(&[1; 32])), "0"); + let mut invalid = body["items"][0].clone(); + invalid["path"] = json!("/absent"); + body["items"].as_array_mut().unwrap().push(invalid); + error( + fixture + .send("POST", "chunks", Body::from(body.to_string())) + .await, + 404, + "PATH_NOT_FOUND", + false, + ) + .await; + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_budgeted_chunk_body_preserves_per_frame_revocation_and_no_end() { + let fixture = Fixture::new().await; + let map = fixture.map("/file").await; + let map_id = map["map"]["map_id"].as_str().unwrap(); + let mut body = fixture.chunk_body("/file", map_id, "0"); + body["items"] + .as_array_mut() + .unwrap() + .push(fixture.chunk_body("/file", map_id, "1")["items"][0].clone()); + let response = fixture + .send("POST", "chunks", Body::from(body.to_string())) + .await; + assert_eq!(response.status(), 200); + let mut data = response.into_body().into_data_stream(); + let first = data.next().await.unwrap().unwrap(); + assert!(matches!( + parse_stream(&first).unwrap().as_slice(), + [Frame::Chunk(_)] + )); + let revoked = fixture + .app + .clone() + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/api/v2/snapshots/leases/{}", fixture.lease)) + .header("authorization", format!("Bearer {TOKEN}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(revoked.status(), 200); + assert!(data.next().await.unwrap().is_err()); + assert!(data.next().await.is_none()); +} diff --git a/src/api/router/snapshot_content.rs b/src/api/router/snapshot_content.rs index a6d9efc4..ea7d8938 100644 --- a/src/api/router/snapshot_content.rs +++ b/src/api/router/snapshot_content.rs @@ -15,12 +15,14 @@ use serde_json::json; use sha2::{Digest, Sha256}; use super::{ - abs_view_path, guarded_treeframe_response, internal, mst2_error_response, request::Mst2Bytes, + abs_view_path, guarded_treeframe_response, guarded_treeframe_response_with_budget, internal, + mst2_error_response, request::Mst2Bytes, }; use crate::ceres::snapshot::{ - chunks::{ChunkProjection, get_or_project}, + chunks::{ChunkProjection, get_or_project_stream, projection_reservation_bytes}, + content_budget::{PROJECTION_LIVE_BYTES, reserve_range_work, reserve_response}, error::{SnapshotError, SnapshotErrorCode}, - pages::{MetadataWalkOutcome, base64_of, fetch_raw_blob, hex_of, resolve_abs_metadata}, + pages::{MetadataWalkOutcome, base64_of, hex_of, resolve_abs_metadata}, resolver::FsKind, view::validate_scope_relative_path, }; @@ -475,20 +477,24 @@ async fn project_for( expected_digest: Option<&str>, ) -> Result, Response> { let f = resolve_file_metadata(handler, root_tree, scope, path, expected_digest).await?; + project_resolved(handler, &f).await +} + +#[allow(clippy::result_large_err)] +async fn project_resolved( + handler: &T, + f: &ResolvedFileMetadata, +) -> Result, Response> { // The first request for a digest builds the projection from the fixed // Git object; later requests slice the cached representation. A miss // rebuilds, never errors with "missing chunk". let digest = f.digest; let size = f.size; - let projection = get_or_project(digest, || async move { - let raw = fetch_raw_blob(handler, &f.oid).await?; - if raw.len() as u64 != size { - return Err(SnapshotError::new( - SnapshotErrorCode::IntegrityError, - "fixed blob length disagrees with its verified size fact", - )); - } - Ok(raw) + let projection = get_or_project_stream(digest, size, || async move { + handler + .get_raw_blob_stream_by_hash(&f.oid) + .await + .map_err(content_read_error) }) .await .map_err(|error| { @@ -658,9 +664,91 @@ const CHUNKS_TOTAL_MAX: u64 = 128 * 1024 * 1024; struct Planned { projection: std::sync::Arc, + oid: String, index: u64, } +struct ResolvedChunk { + file: ResolvedFileMetadata, + index: u64, + map_id: String, +} + +fn content_read_error(error: crate::common::errors::MegaError) -> SnapshotError { + use crate::common::errors::MegaError; + let code = match &error { + MegaError::ObjStorageNotFound(_) => SnapshotErrorCode::ObjectUnavailable, + MegaError::ObjStorageInconsistent(_) => SnapshotErrorCode::IntegrityError, + MegaError::Io(error) if error.kind() == std::io::ErrorKind::InvalidData => { + SnapshotErrorCode::IntegrityError + } + _ => SnapshotErrorCode::Internal, + }; + tracing::warn!(error = %error, "fixed-view content read failed"); + SnapshotError::new(code, "fixed-view content could not be read") +} + +#[allow(clippy::result_large_err)] +async fn read_chunk_range( + handler: &T, + planned: &Planned, +) -> Result, Response> { + let projection = &planned.projection; + let len = projection.map.chunk_len(planned.index).map_err(|error| { + mst2_error_response(internal(format!("invalid admitted chunk: {error}"))) + })?; + let start = planned + .index + .checked_mul(mst2_codec::chunkmap::CHUNK_SIZE as u64) + .ok_or_else(|| mst2_error_response(internal("chunk offset overflow")))?; + let end = start + .checked_add(len) + .ok_or_else(|| mst2_error_response(internal("chunk end overflow")))?; + let (mut input, meta) = handler + .get_raw_blob_range_stream_exact(&planned.oid, start, end) + .await + .map_err(|error| mst2_error_response(content_read_error(error)))? + .ok_or_else(|| { + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::RangeNotSupported, + "fixed source does not support exact raw ranges", + )) + })?; + if u64::try_from(meta.size).ok() != Some(projection.map.file_size) { + return Err(mst2_error_response(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "range source size disagrees with the fixed verified fact", + ))); + } + let mut raw = Vec::new(); + raw.try_reserve_exact(len as usize).map_err(|_| { + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "range allocation could not be admitted", + )) + })?; + while let Some(part) = input.next().await { + let bytes = part.map_err(|error| { + tracing::warn!(error = %error, "fixed-view range stream failed"); + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "fixed-view range stream failed", + )) + })?; + if bytes.len() > len as usize - raw.len() { + return Err(mst2_error_response(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "range response exceeds its exact requested length", + ))); + } + raw.extend_from_slice(&bytes); + } + projection + .verify_chunk(planned.index, &raw) + .map_err(mst2_error_response)?; + Ok(raw) +} + #[allow(clippy::result_large_err)] pub(super) async fn chunks( state: State, @@ -693,13 +781,16 @@ pub(super) async fn chunks( let root_tree = fixed_root_tree(handler.as_ref(), &ctx.root_tree_oid).await?; let scope = ctx.built.descriptor.scope.clone(); let mut planned: Vec = Vec::new(); + let mut resolved: Vec = Vec::new(); let mut units: Vec<(String, u64)> = Vec::new(); + let mut distinct: std::collections::HashMap<[u8; 32], u64> = std::collections::HashMap::new(); + let mut projection_bytes = 0usize; let mut logical_bytes = 0u64; for item in &req.items { validate_scope_relative_path(&item.path).map_err(mst2_error_response)?; let index = parse_decimal_count(&item.chunk_index, "chunk_index").map_err(mst2_error_response)?; - let proj = project_for( + let file = resolve_file_metadata( handler.as_ref(), &root_tree, &scope, @@ -707,19 +798,13 @@ pub(super) async fn chunks( Some(&item.expected_digest), ) .await?; - let want_map = format!("sha256:{}", hex_of(&proj.map_id)); - if want_map != item.map_id { - return Err(mst2_error_response(SnapshotError::new( - SnapshotErrorCode::DigestMismatch, - format!("{}: map_id does not bind to this file", item.path), - ))); - } - if index >= proj.map.chunk_count { + let chunk_count = file.size.div_ceil(mst2_codec::chunkmap::CHUNK_SIZE as u64); + if index >= chunk_count { return Err(mst2_error_response(SnapshotError::new( SnapshotErrorCode::ScopeInvalid, format!( "{}: chunk_index {index} >= chunk_count {}", - item.path, proj.map.chunk_count + item.path, chunk_count ), ))); } @@ -731,10 +816,8 @@ pub(super) async fn chunks( format!("duplicate chunk unit map_id={} index={index}", item.map_id), ))); } - let len = proj - .map - .chunk_len(index) - .map_err(|e| mst2_error_response(internal(e.to_string())))?; + let start = index * mst2_codec::chunkmap::CHUNK_SIZE as u64; + let len = (file.size - start).min(mst2_codec::chunkmap::CHUNK_SIZE as u64); if logical_bytes + len > CHUNKS_TOTAL_MAX { return Err(mst2_error_response(SnapshotError::new( SnapshotErrorCode::ScopeInvalid, @@ -743,9 +826,53 @@ pub(super) async fn chunks( } logical_bytes += len; units.push(unit); + match distinct.entry(file.digest) { + std::collections::hash_map::Entry::Occupied(entry) => { + if *entry.get() != file.size { + return Err(mst2_error_response(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "fixed facts disagree for the same content digest", + ))); + } + } + std::collections::hash_map::Entry::Vacant(entry) => { + let bytes = projection_reservation_bytes(file.size).map_err(mst2_error_response)?; + projection_bytes = projection_bytes.checked_add(bytes).ok_or_else(|| { + mst2_error_response(internal("projection batch memory overflow")) + })?; + if projection_bytes > PROJECTION_LIVE_BYTES { + return Err(mst2_error_response(SnapshotError::new( + SnapshotErrorCode::LimitExceeded, + "chunk batch exceeds its live projection memory budget", + ))); + } + entry.insert(file.size); + } + } + resolved.push(ResolvedChunk { + file, + index, + map_id: item.map_id.clone(), + }); + } + let response_bytes = usize::try_from(logical_bytes) + .ok() + .and_then(|bytes| bytes.checked_add(req.items.len() * 1024 + 1024)) + .ok_or_else(|| mst2_error_response(internal("chunk response memory overflow")))?; + let response_memory = reserve_response(response_bytes).map_err(mst2_error_response)?; + for item in resolved { + let proj = project_resolved(handler.as_ref(), &item.file).await?; + let want_map = format!("sha256:{}", hex_of(&proj.map_id)); + if want_map != item.map_id { + return Err(mst2_error_response(SnapshotError::new( + SnapshotErrorCode::DigestMismatch, + "map_id does not bind to the fixed file", + ))); + } planned.push(Planned { projection: proj, - index, + oid: item.file.oid, + index: item.index, }); } @@ -753,17 +880,23 @@ pub(super) async fn chunks( let mut stream = FrameStream::new(1, encoding); let mut out: Vec> = Vec::new(); for p in planned.iter() { - // Re-verified slice (digest + length) from the staged projection. - let bytes = p - .projection - .chunk_bytes(p.index) - .map_err(mst2_error_response)?; + let _work_memory = reserve_range_work().map_err(mst2_error_response)?; + // The source is the current request's fixed OID, never a cached + // handler/backend/credential from a different scope. + let bytes = if p.projection.has_inline_bytes() { + p.projection + .chunk_bytes(p.index) + .map_err(mst2_error_response)? + .to_vec() + } else { + read_chunk_range(handler.as_ref(), p).await? + }; let frame = stream .chunk( p.projection.map_id, p.projection.map.file_content_id, p.index, - bytes.to_vec(), + bytes, ) .map_err(mst2_error_response)?; out.push(frame); @@ -776,7 +909,15 @@ pub(super) async fn chunks( ); out.push(end); - guarded_treeframe_response(&state, &ctx, &snapshot_id, &body, out).map_err(mst2_error_response) + guarded_treeframe_response_with_budget( + &state, + &ctx, + &snapshot_id, + &body, + out, + Some(response_memory), + ) + .map_err(mst2_error_response) } /// Strict decimal-string parse for unsigned counts (spec 04 §1: no leading diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index add40d69..bb58d788 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/src/api/router/snapshot_content_tests.rs @@ -82,6 +82,9 @@ const TOKEN: &str = "mst2-fixed-content-test"; #[path = "snapshot_objects_bounded_tests.rs"] mod bounded_objects; +#[path = "snapshot_chunks_bounded_tests.rs"] +mod bounded_chunks; + #[path = "snapshot_session_tests.rs"] mod durable_sessions; @@ -103,6 +106,7 @@ struct ReadCounts { range: AtomicUsize, bytes: AtomicUsize, object_fault: std::sync::Mutex>, + chunk_faults: std::sync::Mutex>, } impl ReadCounts { @@ -137,6 +141,31 @@ impl MegaObjectStorage for CountingStorage { async fn get_stream(&self, key: &ObjectKey) -> OrbitResult<(ObjectByteStream, ObjectMeta)> { self.counts.whole.fetch_add(1, Ordering::SeqCst); + let chunk_fault = self + .counts + .chunk_faults + .lock() + .unwrap() + .iter() + .find(|fault| fault.oid == key.key) + .cloned(); + if let Some(fault) = chunk_fault { + let counts = self.counts.clone(); + let size = fault.size; + let stream = fault.full_stream().map(move |part| { + if let Ok(bytes) = &part { + counts.bytes.fetch_add(bytes.len(), Ordering::SeqCst); + } + part + }); + return Ok(( + Box::pin(stream), + ObjectMeta { + size: size as i64, + ..Default::default() + }, + )); + } let (stream, meta) = self.inner.inner.get_stream(key).await?; let fault = self.counts.object_fault.lock().unwrap().clone(); let stream = match fault { @@ -163,6 +192,40 @@ impl MegaObjectStorage for CountingStorage { self.inner.inner.get_range_stream(key, start, end).await } + async fn get_range_stream_exact( + &self, + key: &ObjectKey, + start: u64, + end: u64, + ) -> OrbitResult> { + self.counts.range.fetch_add(1, Ordering::SeqCst); + let fault = self + .counts + .chunk_faults + .lock() + .unwrap() + .iter() + .find(|fault| fault.oid == key.key) + .cloned(); + if let Some(fault) = fault { + let range = fault.range_stream(start, end)?; + return Ok(range.map(|(stream, meta)| { + let counts = self.counts.clone(); + let stream = stream.map(move |part| { + if let Ok(bytes) = &part { + counts.bytes.fetch_add(bytes.len(), Ordering::SeqCst); + } + part + }); + (Box::pin(stream) as ObjectByteStream, meta) + })); + } + self.inner + .inner + .get_range_stream_exact(key, start, end) + .await + } + async fn exists(&self, key: &ObjectKey) -> OrbitResult { self.inner.inner.exists(key).await } diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 40876942..eed58484 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -131,15 +131,51 @@ fn guarded_treeframe_response( request_body: &[u8], frames: Vec>, ) -> Result { + guarded_treeframe_response_with_budget(state, context, snapshot_id, request_body, frames, None) +} + +fn guarded_treeframe_response_with_budget( + state: &MonoApiServiceState, + context: &crate::ceres::snapshot::runtime::SnapshotContext, + snapshot_id: &str, + request_body: &[u8], + frames: Vec>, + memory: Option, +) -> Result { + // Field order also keeps admission credits until frame allocations drop + // on failures before ownership has moved into individual Bytes. + struct FrameAllocation { + frames: Vec>, + memory: Option, + } + let mut allocation = FrameAllocation { frames, memory }; + if let Some(lease) = &allocation.memory { + let allocated = allocation + .frames + .iter() + .try_fold(0usize, |total, frame| total.checked_add(frame.capacity())); + if allocated.is_none_or(|allocated| allocated > lease.bytes) { + return Err(internal("encoded frames exceed their memory reservation")); + } + } let headers = REQUEST_HEADERS.try_with(Clone::clone).map_err(|_| { SnapshotError::new( SnapshotErrorCode::Unauthenticated, "request authentication context missing", ) })?; - let units = frames + let memory = allocation.memory.take().map(std::sync::Arc::new); + let units = std::mem::take(&mut allocation.frames) .into_iter() - .map(Bytes::from) + .map(|bytes| match &memory { + Some(lease) => { + Bytes::from_owner(crate::ceres::snapshot::content_budget::BudgetedFrame { + bytes, + lease: lease.clone(), + }) + } + None => Bytes::from(bytes), + }) .collect::>(); let stream = futures::stream::unfold( (units, state.clone(), context.clone(), headers), diff --git a/src/ceres/api_service/mod.rs b/src/ceres/api_service/mod.rs index 32bbf780..29ec5d74 100644 --- a/src/ceres/api_service/mod.rs +++ b/src/ceres/api_service/mod.rs @@ -191,6 +191,31 @@ pub trait ApiHandler: Send + Sync { } } + async fn get_raw_blob_range_stream_exact( + &self, + hash: &str, + start: u64, + end: u64, + ) -> Result< + Option<( + crate::orbit_api::object_storage::ObjectByteStream, + crate::orbit_api::object_storage::ObjectMeta, + )>, + MegaError, + > { + let storage = self.get_context(); + match storage + .git_service + .get_object_range_stream_exact(hash, start, end) + .await + { + Ok(range) => Ok(range), + Err(error) => Err(storage + .classify_blob_objstorage_not_found(hash, error) + .await), + } + } + /// Preview unified diff for a single file change async fn preview_file_diff( &self, diff --git a/src/ceres/snapshot/chunk_singleflight_tests.rs b/src/ceres/snapshot/chunk_singleflight_tests.rs index 2534acf9..78daff83 100644 --- a/src/ceres/snapshot/chunk_singleflight_tests.rs +++ b/src/ceres/snapshot/chunk_singleflight_tests.rs @@ -70,7 +70,7 @@ fn assert_uncached(content_id: [u8; 32]) { fn evict(content_id: [u8; 32]) { let mut cache = STAGED.get().unwrap().lock().unwrap(); let projection = cache.entries.remove(&content_id).unwrap(); - cache.total_bytes -= projection.raw.len(); + cache.total_bytes -= projection.retained_bytes(); let position = cache.order.iter().position(|id| *id == content_id).unwrap(); cache.order.remove(position); } diff --git a/src/ceres/snapshot/chunks.rs b/src/ceres/snapshot/chunks.rs index e4182c01..8999a45e 100644 --- a/src/ceres/snapshot/chunks.rs +++ b/src/ceres/snapshot/chunks.rs @@ -2,9 +2,9 @@ //! //! Persistent segment/locator storage is T04/T12 work; this identity slice //! builds the MCM2 map and MCL2 leaves from a fixed view's verified blob and -//! stages the raw bytes in a bounded in-process cache, so a CHUNK request -//! slices its range out of an already-verified representation rather than -//! reconstructing the whole Git object per chunk (spec 07 §8). +//! retains inline bytes through 512 MiB and only verified map metadata for +//! larger files. Large CHUNK reads use the current request's strict raw range +//! source. This is a reproducible process cache, not a durable locator. //! //! The cache is an optimization, never the authority: every projected file //! is re-hashed against the `content_id` the fixed view advertised, and a @@ -19,21 +19,33 @@ use std::{ use mst2_codec::chunkmap::{CHUNK_SIZE, CHUNKS_PER_PAGE, ChunkLeaf, ChunkMap}; use sha2::{Digest, Sha256}; +use super::content_budget::{MemoryLease, projection_budget}; use crate::ceres::snapshot::error::{SnapshotError, SnapshotErrorCode}; +#[path = "chunks_stream.rs"] +mod streaming; + +pub use streaming::get_or_project_stream; + +pub(crate) fn projection_reservation_bytes(size: u64) -> Result { + streaming::reserved_bytes(size) +} + /// One file's verified range-readable projection. pub struct ChunkProjection { pub map: ChunkMap, pub map_id: [u8; 32], leaves: Vec, leaf_hashes: Vec<[u8; 32]>, - /// Full verified content. Indexed by fixed 1 MiB chunk boundaries. + /// Empty for range-only projections; empty files have no projection. raw: Vec, + memory: Option, } impl ChunkProjection { /// Build from bytes that the caller already resolved at a fixed path. /// `content_id` is the digest the fixed view advertises. + #[cfg(test)] pub fn build(content_id: [u8; 32], raw: Vec) -> Result { let mut hasher = Sha256::new(); hasher.update(&raw); @@ -95,9 +107,52 @@ impl ChunkProjection { leaves, leaf_hashes, raw, + memory: None, }) } + pub fn has_inline_bytes(&self) -> bool { + !self.raw.is_empty() + } + + fn retained_bytes(&self) -> usize { + self.memory + .as_ref() + .map_or_else(|| self.allocated_bytes(), |lease| lease.bytes) + } + + fn allocated_bytes(&self) -> usize { + std::mem::size_of::() + + self.raw.capacity() + + self.leaves.capacity() * std::mem::size_of::() + + self.leaf_hashes.capacity() * 32 + + self + .leaves + .iter() + .map(|leaf| leaf.chunk_sha256.capacity() * 32) + .sum::() + } + + pub fn verify_chunk(&self, index: u64, bytes: &[u8]) -> Result<(), SnapshotError> { + let want_len = self.map.chunk_len(index).map_err(codec_err)?; + if bytes.len() as u64 != want_len { + return Err(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "range length disagrees with the verified chunk map", + )); + } + let page = (index / CHUNKS_PER_PAGE as u64) as usize; + let slot = (index % CHUNKS_PER_PAGE as u64) as usize; + let got: [u8; 32] = Sha256::digest(bytes).into(); + if got != self.leaves[page].chunk_sha256[slot] { + return Err(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "range digest disagrees with the verified chunk map", + )); + } + Ok(()) + } + pub fn page_count(&self) -> u64 { self.map.page_count } @@ -175,6 +230,7 @@ static STAGED: OnceLock> = OnceLock::new(); /// 512 MiB cap for staged file bytes in this identity slice. The real /// persistent RAW_OBJECT locator layout replaces this (spec 10 §4). const STAGED_CAP_BYTES: usize = 512 * 1024 * 1024; +const STAGED_MAX_ENTRIES: usize = 4096; struct ProjectionCache { entries: HashMap<[u8; 32], std::sync::Arc>, @@ -196,26 +252,39 @@ impl ProjectionCache { self.entries.get(&id).cloned() } - fn put(&mut self, proj: std::sync::Arc) { + fn evict_oldest(&mut self) -> Option> { + while let Some(victim) = self.order.pop_front() { + if let Some(projection) = self.entries.remove(&victim) { + self.total_bytes -= projection.retained_bytes(); + return Some(projection); + } + } + None + } + + fn put(&mut self, proj: std::sync::Arc) -> Vec> { + let mut evicted = Vec::new(); let id = proj.map.file_content_id; if self.entries.contains_key(&id) { - return; + return evicted; } // Reclaim oldest entries until the new one fits. A file larger than // the cap alone is not cached between flights but still // served correctly. - while self.total_bytes + proj.raw.len() > STAGED_CAP_BYTES - && let Some(victim) = self.order.pop_front() - { - if let Some(v) = self.entries.remove(&victim) { - self.total_bytes = self.total_bytes.saturating_sub(v.raw.len()); - } + let bytes = proj.retained_bytes(); + if bytes > STAGED_CAP_BYTES { + return evicted; } - if proj.raw.len() <= STAGED_CAP_BYTES { - self.total_bytes += proj.raw.len(); - self.order.push_back(id); - self.entries.insert(id, proj); + while (self.total_bytes + bytes > STAGED_CAP_BYTES + || self.entries.len() >= STAGED_MAX_ENTRIES) + && let Some(victim) = self.evict_oldest() + { + evicted.push(victim); } + self.total_bytes += bytes; + self.order.push_back(id); + self.entries.insert(id, proj); + evicted } } @@ -301,6 +370,7 @@ impl Drop for FlightClaim<'_> { /// Return the cached projection, or build one via `load` (which must resolve /// the file in the fixed view and return its verified bytes). Concurrent /// misses for the same digest share one successful load and projection. +#[cfg(test)] pub async fn get_or_project( content_id: [u8; 32], load: F, @@ -308,6 +378,20 @@ pub async fn get_or_project( where F: FnOnce() -> Fut, Fut: std::future::Future, SnapshotError>>, +{ + get_or_project_with(content_id, || async { + ChunkProjection::build(content_id, load().await?) + }) + .await +} + +async fn get_or_project_with( + content_id: [u8; 32], + build: F, +) -> Result, SnapshotError> +where + F: FnOnce() -> Fut, + Fut: std::future::Future>, { let cache = STAGED.get_or_init(|| Mutex::new(ProjectionCache::new())); if let Some(p) = cache @@ -336,12 +420,14 @@ where if let Some(p) = result.as_ref() { return Ok(p.clone()); } - let raw = load().await?; - let proj = std::sync::Arc::new(ChunkProjection::build(content_id, raw)?); - cache + let proj = Arc::new(build().await?); + let evicted = cache .lock() .map_err(|_| internal("chunk projection cache lock poisoned"))? .put(proj.clone()); + // Final projection/credit drops can be expensive and must not hold the + // process cache mutex. Eviction does not refund other Arc owners. + drop(evicted); *result = Some(proj.clone()); Ok(proj) } diff --git a/src/ceres/snapshot/chunks_stream.rs b/src/ceres/snapshot/chunks_stream.rs new file mode 100644 index 00000000..56f007fd --- /dev/null +++ b/src/ceres/snapshot/chunks_stream.rs @@ -0,0 +1,274 @@ +use futures::StreamExt; +use tokio::sync::Semaphore; + +use super::*; +use crate::orbit_api::object_storage::ObjectByteStream; + +const STREAM_ITEM_MAX_BYTES: usize = 8 * 1024 * 1024; +const MAX_FILE_BYTES: u64 = 8 * 1024 * 1024 * 1024 * 1024; +const MAX_BUILDERS: usize = 4; +const CONSTRUCTION_ALLOWANCE: usize = 64 * 1024; + +pub(super) fn reserved_bytes(size: u64) -> Result { + if size == 0 || size > MAX_FILE_BYTES { + return Err(SnapshotError::new( + if size == 0 { + SnapshotErrorCode::ScopeInvalid + } else { + SnapshotErrorCode::LimitExceeded + }, + "file is outside the chunk projection profile", + )); + } + let chunks = size.div_ceil(CHUNK_SIZE as u64); + let pages = chunks.div_ceil(CHUNKS_PER_PAGE as u64); + let inline = if size <= STAGED_CAP_BYTES as u64 { + size + } else { + 0 + }; + let bytes = chunks + .checked_mul(32) + .and_then(|n| { + pages + .checked_mul((std::mem::size_of::() + 32) as u64) + .and_then(|pages| n.checked_add(pages)) + }) + .and_then(|n| n.checked_add(inline)) + .and_then(|n| n.checked_add(CONSTRUCTION_ALLOWANCE as u64)) + .and_then(|n| usize::try_from(n).ok()) + .ok_or_else(|| internal("chunk projection memory arithmetic overflow"))?; + Ok(bytes) +} + +fn reserve_projection( + size: u64, + budget: &Arc, + cache: &Mutex, +) -> Result { + let bytes = reserved_bytes(size)?; + loop { + match budget.reserve(bytes) { + Ok(lease) => return Ok(lease), + Err(error) => { + if error.code != SnapshotErrorCode::TemporaryUnavailable { + return Err(error); + } + let victim = cache + .lock() + .map_err(|_| internal("chunk projection cache lock poisoned"))? + .evict_oldest(); + let Some(victim) = victim else { + return Err(error); + }; + drop(victim); + } + } + } +} + +pub async fn get_or_project_stream( + content_id: [u8; 32], + size: u64, + open: F, +) -> Result, SnapshotError> +where + F: FnOnce() -> Fut, + Fut: std::future::Future>, +{ + // Profile arithmetic also runs on hits; fixed facts remain per-request. + reserved_bytes(size)?; + get_or_project_with(content_id, || async { + static BUILDERS: OnceLock> = OnceLock::new(); + build_with_resources( + content_id, + size, + open, + projection_budget(), + BUILDERS.get_or_init(|| Arc::new(Semaphore::new(MAX_BUILDERS))), + STAGED.get_or_init(|| Mutex::new(ProjectionCache::new())), + ) + .await + }) + .await +} + +async fn build_with_resources( + content_id: [u8; 32], + size: u64, + open: F, + budget: &Arc, + builders: &Arc, + cache: &Mutex, +) -> Result +where + F: FnOnce() -> Fut, + Fut: std::future::Future>, +{ + let _builder = builders.clone().try_acquire_owned().map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "chunk projection builders are occupied", + ) + })?; + let lease = reserve_projection(size, budget, cache)?; + // Source body I/O starts only after builder and retained-memory credit. + let input = open().await?; + build_stream(content_id, size, input, lease).await +} + +async fn build_stream( + content_id: [u8; 32], + size: u64, + mut input: ObjectByteStream, + lease: MemoryLease, +) -> Result { + let chunk_count = size.div_ceil(CHUNK_SIZE as u64); + let page_count = chunk_count.div_ceil(CHUNKS_PER_PAGE as u64); + let mut raw = Vec::new(); + let mut leaves = Vec::new(); + let mut leaf_hashes = Vec::new(); + let mut current = Vec::new(); + let inline = size <= STAGED_CAP_BYTES as u64; + if inline { + raw.try_reserve_exact(size as usize) + .map_err(allocation_error)?; + } + leaves + .try_reserve_exact(page_count as usize) + .map_err(allocation_error)?; + leaf_hashes + .try_reserve_exact(page_count as usize) + .map_err(allocation_error)?; + current + .try_reserve_exact(chunk_count.min(CHUNKS_PER_PAGE as u64) as usize) + .map_err(allocation_error)?; + let mut full_hash = Sha256::new(); + let mut chunk_hash = Sha256::new(); + let mut received = 0u64; + let mut chunk_bytes = 0usize; + let mut digests = 0u64; + let mut since_yield = 0usize; + while let Some(part) = input.next().await { + let bytes = part.map_err(|error| { + tracing::warn!(error = %error, "fixed-view chunk projection stream failed"); + SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "content stream failed", + ) + })?; + if bytes.len() as u64 > size - received { + return Err(length_error()); + } + // A visible-item admission limit cannot bound a producer's backing + // allocation, buffers or work surviving cancellation. + if bytes.len() > STREAM_ITEM_MAX_BYTES { + return Err(SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "content producer item exceeds the consumer processing limit", + )); + } + full_hash.update(&bytes); + if inline { + raw.extend_from_slice(&bytes); + } + received += bytes.len() as u64; + let mut rest = bytes.as_ref(); + while !rest.is_empty() { + let len = rest.len().min(CHUNK_SIZE as usize - chunk_bytes); + chunk_hash.update(&rest[..len]); + chunk_bytes += len; + rest = &rest[len..]; + if chunk_bytes == CHUNK_SIZE as usize { + current.push(chunk_hash.finalize_reset().into()); + digests += 1; + chunk_bytes = 0; + if current.len() == CHUNKS_PER_PAGE { + append_leaf(&mut leaves, &mut leaf_hashes, std::mem::take(&mut current))?; + if digests < chunk_count { + current + .try_reserve_exact( + (chunk_count - digests).min(CHUNKS_PER_PAGE as u64) as usize + ) + .map_err(allocation_error)?; + } + } + } + } + since_yield += bytes.len(); + if since_yield >= STREAM_ITEM_MAX_BYTES { + // A ready stream still cooperates with cancellation and the + // executor while hashing a large file. No detached CPU worker. + tokio::task::yield_now().await; + since_yield = 0; + } + } + if received != size { + return Err(length_error()); + } + let got: [u8; 32] = full_hash.finalize().into(); + if got != content_id { + return Err(SnapshotError::new( + SnapshotErrorCode::DigestMismatch, + "chunk projection input does not hash to content_id", + )); + } + if chunk_bytes != 0 { + current.push(chunk_hash.finalize().into()); + digests += 1; + } + if !current.is_empty() { + append_leaf(&mut leaves, &mut leaf_hashes, current)?; + } + if digests != chunk_count || leaves.len() as u64 != page_count { + return Err(internal( + "streamed chunk boundaries disagree with file size", + )); + } + let pages_root = mst2_codec::chunkmap::merkle_root(&leaf_hashes).map_err(codec_err)?; + let map = ChunkMap::new(content_id, size, pages_root).map_err(codec_err)?; + let projection = ChunkProjection { + map_id: map.map_id(), + map, + leaves, + leaf_hashes, + raw, + memory: Some(lease), + }; + if projection.allocated_bytes() > projection.retained_bytes() { + return Err(internal("projection allocation exceeds its reservation")); + } + Ok(projection) +} + +fn append_leaf( + leaves: &mut Vec, + hashes: &mut Vec<[u8; 32]>, + chunk_sha256: Vec<[u8; 32]>, +) -> Result<(), SnapshotError> { + let leaf = ChunkLeaf { + page_index: leaves.len() as u64, + chunk_sha256, + }; + hashes.push(leaf.leaf_hash().map_err(codec_err)?); + leaves.push(leaf); + Ok(()) +} + +fn allocation_error(_: std::collections::TryReserveError) -> SnapshotError { + SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "chunk projection allocation could not be admitted", + ) +} + +fn length_error() -> SnapshotError { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "fixed blob length disagrees with its verified size fact", + ) +} + +#[cfg(test)] +#[path = "chunks_stream_tests.rs"] +mod tests; diff --git a/src/ceres/snapshot/chunks_stream_tests.rs b/src/ceres/snapshot/chunks_stream_tests.rs new file mode 100644 index 00000000..2ccebae8 --- /dev/null +++ b/src/ceres/snapshot/chunks_stream_tests.rs @@ -0,0 +1,391 @@ +use std::{ + io, + sync::atomic::{AtomicUsize, Ordering}, + time::Duration, +}; + +use bytes::Bytes; +use tokio::{sync::Notify, time::timeout}; + +use super::*; +use crate::ceres::snapshot::content_budget::MemoryBudget; + +fn stream(parts: Vec>) -> ObjectByteStream { + Box::pin(futures::stream::iter(parts)) +} + +async fn project(raw: &[u8], parts: Vec>) -> ChunkProjection { + let budget = MemoryBudget::new(reserved_bytes(raw.len() as u64).unwrap()); + let lease = budget + .reserve(reserved_bytes(raw.len() as u64).unwrap()) + .unwrap(); + build_stream( + Sha256::digest(raw).into(), + raw.len() as u64, + stream(parts), + lease, + ) + .await + .unwrap() +} + +#[tokio::test] +async fn arbitrary_fragmentation_preserves_canonical_map_leaves_and_full_final_chunk() { + for size in [ + 1, + CHUNK_SIZE as usize - 1, + CHUNK_SIZE as usize, + CHUNK_SIZE as usize + 7, + 2 * CHUNK_SIZE as usize, + ] { + let raw: Vec = (0..size).map(|i| (i % 251) as u8).collect(); + let parts = raw + .chunks(997) + .map(|part| Ok(Bytes::copy_from_slice(part))) + .collect(); + let projection = project(&raw, parts).await; + let oracle = ChunkProjection::build(Sha256::digest(&raw).into(), raw.clone()).unwrap(); + assert_eq!(projection.map, oracle.map); + assert_eq!(projection.map_id, oracle.map_id); + assert_eq!(projection.leaves, oracle.leaves); + assert_eq!(projection.leaf_hashes, oracle.leaf_hashes); + assert!(projection.has_inline_bytes()); + for index in 0..projection.map.chunk_count { + assert_eq!( + projection.chunk_bytes(index).unwrap(), + oracle.chunk_bytes(index).unwrap() + ); + let (leaf, proof) = projection + .leaf_and_proof(index / CHUNKS_PER_PAGE as u64) + .unwrap(); + mst2_codec::chunkmap::verify_leaf( + projection.page_count(), + leaf.page_index, + leaf.leaf_hash().unwrap(), + &proof, + projection.map.pages_root, + ) + .unwrap(); + } + } +} + +#[tokio::test] +async fn large_production_threshold_retains_only_map_and_verifies_every_page() { + // Reuse one producer allocation; never create the >512 MiB raw Vec. + let block = Bytes::from(vec![37; CHUNK_SIZE as usize]); + let size = STAGED_CAP_BYTES as u64 + 7; + let mut full = Sha256::new(); + for _ in 0..512 { + full.update(&block); + } + full.update(&block[..7]); + let content_id: [u8; 32] = full.finalize().into(); + let input = Box::pin(futures::stream::iter((0..513).map(move |index| { + Ok(if index == 512 { + block.slice(..7) + } else { + block.clone() + }) + }))); + let weight = reserved_bytes(size).unwrap(); + let budget = MemoryBudget::new(weight); + let lease = budget.reserve(weight).unwrap(); + let projection = build_stream(content_id, size, input, lease).await.unwrap(); + assert!(!projection.has_inline_bytes()); + assert_eq!(projection.raw.capacity(), 0); + assert_eq!(projection.map.chunk_count, 513); + assert_eq!(projection.page_count(), 3); + assert!(projection.allocated_bytes() < 128 * 1024); + assert_eq!(budget.used(), weight); + let full_chunk: [u8; 32] = Sha256::digest(vec![37; CHUNK_SIZE as usize]).into(); + let last_chunk: [u8; 32] = Sha256::digest([37; 7]).into(); + assert_eq!(projection.leaves[0].chunk_sha256, vec![full_chunk; 256]); + assert_eq!(projection.leaves[1].chunk_sha256, vec![full_chunk; 256]); + assert_eq!(projection.leaves[2].chunk_sha256, vec![last_chunk]); + for page in 0..3 { + let (leaf, proof) = projection.leaf_and_proof(page).unwrap(); + mst2_codec::chunkmap::verify_leaf( + 3, + page, + leaf.leaf_hash().unwrap(), + &proof, + projection.map.pages_root, + ) + .unwrap(); + } + projection.verify_chunk(512, &[37; 7]).unwrap(); + assert_eq!( + projection.verify_chunk(512, &[38; 7]).unwrap_err().code, + SnapshotErrorCode::IntegrityError + ); + drop(projection); + assert_eq!(budget.used(), 0); +} + +#[tokio::test] +async fn exact_eof_wrong_hash_growth_and_late_error_never_publish_a_projection() { + let cases = [ + ( + vec![Ok(Bytes::from_static(b"ab"))], + SnapshotErrorCode::IntegrityError, + ), + ( + vec![Ok(Bytes::from_static(b"abc")), Ok(Bytes::from_static(b"d"))], + SnapshotErrorCode::IntegrityError, + ), + ( + vec![Ok(Bytes::from_static(b"abd"))], + SnapshotErrorCode::DigestMismatch, + ), + ( + vec![ + Ok(Bytes::from_static(b"abc")), + Err(io::Error::other("late")), + ], + SnapshotErrorCode::ObjectUnavailable, + ), + ]; + for (parts, code) in cases { + let budget = MemoryBudget::new(reserved_bytes(3).unwrap()); + let lease = budget.reserve(reserved_bytes(3).unwrap()).unwrap(); + let failure = build_stream(Sha256::digest(b"abc").into(), 3, stream(parts), lease) + .await + .err() + .unwrap(); + assert_eq!(failure.code, code); + assert_eq!(budget.used(), 0); + } +} + +#[tokio::test] +async fn overlong_producer_is_rejected_before_hash_copy_and_tail_poll() { + let tail = Arc::new(AtomicUsize::new(0)); + let input = Box::pin(futures::stream::unfold( + (false, tail.clone()), + |(sent, tail)| async move { + if sent { + tail.fetch_add(1, Ordering::SeqCst); + Some((Err(io::Error::other("must not poll tail")), (true, tail))) + } else { + Some((Ok(Bytes::from_static(b"abcd")), (true, tail))) + } + }, + )); + let budget = MemoryBudget::new(reserved_bytes(3).unwrap()); + let lease = budget.reserve(reserved_bytes(3).unwrap()).unwrap(); + assert_eq!( + build_stream(Sha256::digest(b"abc").into(), 3, input, lease) + .await + .err() + .unwrap() + .code, + SnapshotErrorCode::IntegrityError + ); + assert_eq!(tail.load(Ordering::SeqCst), 0); + assert_eq!(budget.used(), 0); +} + +#[tokio::test] +async fn visible_producer_item_limit_does_not_collect_a_large_item() { + let size = STREAM_ITEM_MAX_BYTES + 1; + let budget = MemoryBudget::new(reserved_bytes(size as u64).unwrap()); + let lease = budget + .reserve(reserved_bytes(size as u64).unwrap()) + .unwrap(); + let input = stream(vec![Ok(Bytes::from(vec![1; size]))]); + assert_eq!( + build_stream([0; 32], size as u64, input, lease) + .await + .err() + .unwrap() + .code, + SnapshotErrorCode::TemporaryUnavailable + ); + assert_eq!(budget.used(), 0); +} + +#[tokio::test] +async fn evicted_projection_remains_charged_until_last_reader_drops() { + let raw = b"owned credits"; + let budget = MemoryBudget::new(reserved_bytes(raw.len() as u64).unwrap()); + let lease = budget + .reserve(reserved_bytes(raw.len() as u64).unwrap()) + .unwrap(); + let projection = Arc::new( + build_stream( + Sha256::digest(raw).into(), + raw.len() as u64, + stream(vec![Ok(Bytes::from_static(raw))]), + lease, + ) + .await + .unwrap(), + ); + let weight = projection.retained_bytes(); + let mut cache = ProjectionCache::new(); + assert!(cache.put(projection.clone()).is_empty()); + let reader = projection.clone(); + drop(projection); + drop(cache.evict_oldest()); + assert_eq!(cache.total_bytes, 0); + assert_eq!(budget.used(), weight); + assert!(budget.reserve(1).is_err()); + drop(reader); + assert_eq!(budget.used(), 0); +} + +struct DropCount(Arc); +impl Drop for DropCount { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +#[tokio::test] +async fn production_admission_rejects_live_credit_and_builder_overload_before_open() { + let weight = reserved_bytes(3).unwrap(); + let cache = Mutex::new(ProjectionCache::new()); + let budget = MemoryBudget::new(weight); + let builders = Arc::new(Semaphore::new(MAX_BUILDERS)); + let calls = AtomicUsize::new(0); + let held = budget.reserve(weight).unwrap(); + let error = build_with_resources( + Sha256::digest(b"abc").into(), + 3, + || async { + calls.fetch_add(1, Ordering::SeqCst); + Ok(stream(vec![Ok(Bytes::from_static(b"abc"))])) + }, + &budget, + &builders, + &cache, + ) + .await + .err() + .unwrap(); + assert_eq!(error.code, SnapshotErrorCode::TemporaryUnavailable); + assert_eq!(calls.load(Ordering::SeqCst), 0); + assert_eq!(builders.available_permits(), MAX_BUILDERS); + drop(held); + let mut workers = Vec::new(); + for _ in 0..MAX_BUILDERS { + workers.push(builders.clone().try_acquire_owned().unwrap()); + } + let error = build_with_resources( + Sha256::digest(b"abc").into(), + 3, + || async { + calls.fetch_add(1, Ordering::SeqCst); + Ok(stream(vec![Ok(Bytes::from_static(b"abc"))])) + }, + &budget, + &builders, + &cache, + ) + .await + .err() + .unwrap(); + assert_eq!(error.code, SnapshotErrorCode::TemporaryUnavailable); + assert_eq!(calls.load(Ordering::SeqCst), 0); + assert_eq!(budget.used(), 0); + drop(workers); + let projection = build_with_resources( + Sha256::digest(b"abc").into(), + 3, + || async { + calls.fetch_add(1, Ordering::SeqCst); + Ok(stream(vec![Ok(Bytes::from_static(b"abc"))])) + }, + &budget, + &builders, + &cache, + ) + .await + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(budget.used(), weight); + assert_eq!(builders.available_permits(), MAX_BUILDERS); + drop(projection); + assert_eq!(budget.used(), 0); +} + +#[tokio::test] +async fn cancelled_stream_owner_refunds_after_stream_drop_and_same_flight_waiter_takes_over() { + let raw = Bytes::from(uuid::Uuid::new_v4().as_bytes().to_vec()); + let id: [u8; 32] = Sha256::digest(&raw).into(); + let entered = Arc::new(Notify::new()); + let drops = Arc::new(AtomicUsize::new(0)); + let leader = tokio::spawn({ + let entered = entered.clone(); + let drops = drops.clone(); + async move { + get_or_project_stream(id, 16, || async { + let input: ObjectByteStream = Box::pin(futures::stream::unfold( + (entered, DropCount(drops)), + |(entered, owner)| async move { + entered.notify_one(); + std::future::pending::<()>().await; + Some((Ok(Bytes::new()), (entered, owner))) + }, + )); + Ok(input) + }) + .await + } + }); + timeout(Duration::from_secs(5), entered.notified()) + .await + .unwrap(); + let waiter = tokio::spawn(async move { + get_or_project_stream(id, 16, || async { + let input: ObjectByteStream = stream(vec![Ok(raw)]); + Ok(input) + }) + .await + }); + timeout(Duration::from_secs(5), async { + loop { + if FLIGHTS + .get() + .unwrap() + .lock() + .unwrap() + .entries + .get(&id) + .map_or(0, Weak::strong_count) + == 2 + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + leader.abort(); + assert!(leader.await.err().unwrap().is_cancelled()); + assert_eq!(drops.load(Ordering::SeqCst), 1); + let projection = timeout(Duration::from_secs(5), waiter) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(projection.map.file_content_id, id); + assert_eq!(projection.chunk_bytes(0).unwrap().len(), 16); +} + +#[test] +fn maximum_file_metadata_fits_and_protocol_or_quota_rejection_needs_no_source() { + assert!(reserved_bytes(MAX_FILE_BYTES).unwrap() < 300 * 1024 * 1024); + assert_eq!( + reserved_bytes(0).unwrap_err().code, + SnapshotErrorCode::ScopeInvalid + ); + assert_eq!( + reserved_bytes(MAX_FILE_BYTES + 1).unwrap_err().code, + SnapshotErrorCode::LimitExceeded + ); + assert!(reserved_bytes(STAGED_CAP_BYTES as u64).unwrap() > STAGED_CAP_BYTES); + assert!(reserved_bytes(STAGED_CAP_BYTES as u64 + 1).unwrap() < 128 * 1024); +} diff --git a/src/ceres/snapshot/content_budget.rs b/src/ceres/snapshot/content_budget.rs new file mode 100644 index 00000000..79db0289 --- /dev/null +++ b/src/ceres/snapshot/content_budget.rs @@ -0,0 +1,147 @@ +//! Ownership-based consumer memory admission. Backend allocations and allocator +//! overhead are separate; these credits are not a process RSS guarantee. + +use std::sync::{ + Arc, OnceLock, + atomic::{AtomicUsize, Ordering}, +}; + +use super::error::{SnapshotError, SnapshotErrorCode}; + +pub(crate) const PROJECTION_LIVE_BYTES: usize = 1024 * 1024 * 1024; +const RESPONSE_LIVE_BYTES: usize = 512 * 1024 * 1024; +const RANGE_SCRATCH_BYTES: usize = 128 * 1024 * 1024; +pub(crate) const RANGE_WORK_BYTES: usize = 8 * 1024 * 1024; + +pub(crate) struct MemoryBudget { + limit: usize, + used: AtomicUsize, +} + +impl MemoryBudget { + pub(crate) fn new(limit: usize) -> Arc { + Arc::new(Self { + limit, + used: AtomicUsize::new(0), + }) + } + + pub(crate) fn reserve(self: &Arc, bytes: usize) -> Result { + if bytes > self.limit { + return Err(SnapshotError::new( + SnapshotErrorCode::LimitExceeded, + "content operation exceeds its consumer memory budget", + )); + } + let mut used = self.used.load(Ordering::Acquire); + loop { + if bytes > self.limit - used { + return Err(SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "content consumer memory budget is occupied", + )); + } + match self.used.compare_exchange_weak( + used, + used + bytes, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => { + return Ok(MemoryLease { + budget: self.clone(), + bytes, + }); + } + Err(actual) => used = actual, + } + } + } + + #[cfg(test)] + pub(crate) fn used(&self) -> usize { + self.used.load(Ordering::Acquire) + } +} + +pub(crate) struct MemoryLease { + budget: Arc, + pub(crate) bytes: usize, +} + +impl Drop for MemoryLease { + fn drop(&mut self) { + self.budget.used.fetch_sub(self.bytes, Ordering::AcqRel); + } +} + +pub(crate) fn projection_budget() -> &'static Arc { + static BUDGET: OnceLock> = OnceLock::new(); + BUDGET.get_or_init(|| MemoryBudget::new(PROJECTION_LIVE_BYTES)) +} + +pub(crate) fn reserve_response(bytes: usize) -> Result { + static BUDGET: OnceLock> = OnceLock::new(); + BUDGET + .get_or_init(|| MemoryBudget::new(RESPONSE_LIVE_BYTES)) + .reserve(bytes) +} + +pub(crate) fn reserve_range_work() -> Result { + static BUDGET: OnceLock> = OnceLock::new(); + BUDGET + .get_or_init(|| MemoryBudget::new(RANGE_SCRATCH_BYTES)) + .reserve(RANGE_WORK_BYTES) +} + +/// Every encoded frame owns shared credit, including after the body passes a +/// frame to a transport that retains or clones its Bytes. +pub(crate) struct BudgetedFrame { + pub(crate) bytes: Vec, + pub(crate) lease: Arc, +} + +impl AsRef<[u8]> for BudgetedFrame { + fn as_ref(&self) -> &[u8] { + &self.bytes + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn frames_retain_credit_until_last_transport_clone_drops() { + let budget = MemoryBudget::new(64); + let lease = Arc::new(budget.reserve(64).unwrap()); + let frame = bytes::Bytes::from_owner(BudgetedFrame { + bytes: vec![7; 32], + lease: lease.clone(), + }); + let transport = frame.clone(); + drop(lease); + drop(frame); + assert_eq!(budget.used(), 64); + assert!(budget.reserve(1).is_err()); + drop(transport); + assert_eq!(budget.used(), 0); + assert!(budget.reserve(64).is_ok()); + } + + #[test] + fn quota_rejects_without_waiting_for_a_callers_own_credits() { + let budget = MemoryBudget::new(12); + let first = budget.reserve(8).unwrap(); + assert_eq!( + budget.reserve(5).err().unwrap().code, + SnapshotErrorCode::TemporaryUnavailable + ); + assert_eq!( + budget.reserve(13).err().unwrap().code, + SnapshotErrorCode::LimitExceeded + ); + drop(first); + assert_eq!(budget.used(), 0); + } +} diff --git a/src/ceres/snapshot/mod.rs b/src/ceres/snapshot/mod.rs index 97a1a50b..532d1da6 100644 --- a/src/ceres/snapshot/mod.rs +++ b/src/ceres/snapshot/mod.rs @@ -4,6 +4,7 @@ //! composition seam; identity, attestation and publication integration remain //! separate gates. Fixed-source readers must never look up current refs. pub mod chunks; +pub(crate) mod content_budget; pub mod descriptor; pub mod error; pub mod frame_stream; diff --git a/src/jupiter/service/git_service.rs b/src/jupiter/service/git_service.rs index bf42b5ee..6569e9a0 100644 --- a/src/jupiter/service/git_service.rs +++ b/src/jupiter/service/git_service.rs @@ -119,6 +119,26 @@ impl GitService { Ok(stream) } + pub async fn get_object_range_stream_exact( + &self, + hash: &str, + start: u64, + end: u64, + ) -> Result, MegaError> { + if !is_full_hex_object_id(hash) { + return Err(MegaError::Other("Invalid object ID format".to_string())); + } + let key = ObjectKey { + namespace: ObjectNamespace::Git, + key: hash.to_string(), + }; + Ok(self + .obj_storage + .inner + .get_range_stream_exact(&key, start, end) + .await?) + } + pub fn get_objects_stream(&self, hashes: Vec) -> MultiObjectByteStream<'_> { // Filter out obviously invalid object ids early to avoid spurious backend requests. // Callers that need strict validation should validate up-front and return 4xx. diff --git a/src/jupiter/storage/object_storage.rs b/src/jupiter/storage/object_storage.rs index f77fbc86..506f2974 100644 --- a/src/jupiter/storage/object_storage.rs +++ b/src/jupiter/storage/object_storage.rs @@ -35,6 +35,51 @@ pub async fn build_object_storage( }) } +#[cfg(test)] +mod exact_range_tests { + use super::*; + use crate::orbit_api::object_storage::ObjectNamespace; + + #[tokio::test] + async fn memory_exact_range_has_no_clipped_or_empty_success() { + let storage = mock_object_storage(); + let key = ObjectKey { + namespace: ObjectNamespace::Git, + key: "abcdef1234567890".to_string(), + }; + storage + .inner + .put_stream( + &key, + Box::pin(futures::stream::iter([Ok(Bytes::from_static(b"abcdef"))])), + ObjectMeta { + size: 6, + ..Default::default() + }, + ) + .await + .unwrap(); + let (mut stream, meta) = storage + .inner + .get_range_stream_exact(&key, 2, 5) + .await + .unwrap() + .unwrap(); + assert_eq!(meta.size, 6); + assert_eq!(stream.next().await.unwrap().unwrap(), b"cde".as_slice()); + assert!(stream.next().await.is_none()); + for (start, end) in [(5, 7), (6, 7), (4, 4), (u64::MAX, u64::MAX)] { + assert!( + storage + .inner + .get_range_stream_exact(&key, start, end) + .await + .is_err() + ); + } + } +} + #[derive(Default)] struct InMemoryObjectStorage { objects: Mutex>, @@ -127,6 +172,32 @@ impl MegaObjectStorage for InMemoryObjectStorage { Ok((Self::stream_bytes(bytes.slice(start..end)), meta)) } + async fn get_range_stream_exact( + &self, + key: &ObjectKey, + start: u64, + end: u64, + ) -> OrbitResult> { + let (bytes, meta) = self + .objects + .lock() + .map_err(|_| IoOrbitError::Other("object storage lock poisoned".to_string()))? + .get(key) + .cloned() + .ok_or_else(|| IoOrbitError::object_store_not_found(key.default_sharding()))?; + let start = + usize::try_from(start).map_err(|_| IoOrbitError::object_store("range overflow"))?; + let end = usize::try_from(end).map_err(|_| IoOrbitError::object_store("range overflow"))?; + if start >= end || end > bytes.len() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "exact object range is outside stored bytes", + ) + .into()); + } + Ok(Some((Self::stream_bytes(bytes.slice(start..end)), meta))) + } + async fn exists(&self, key: &ObjectKey) -> OrbitResult { Ok(self .objects diff --git a/src/orbit/adapter/object.rs b/src/orbit/adapter/object.rs index ca1751b3..58ccacc0 100644 --- a/src/orbit/adapter/object.rs +++ b/src/orbit/adapter/object.rs @@ -106,6 +106,43 @@ impl MegaObjectStorage for ObjectStoreAdapter { Ok((Box::pin(stream), meta)) } + async fn get_range_stream_exact( + &self, + key: &ObjectKey, + start: u64, + end: u64, + ) -> OrbitResult> { + if start >= end { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "invalid exact object range", + ) + .into()); + } + let path = Self::checked_path(key)?; + let res = self + .to_store() + .get_opts( + &path, + object_store::GetOptions { + range: Some(object_store::GetRange::Bounded(start..end)), + ..Default::default() + }, + ) + .await + .map_err(IoOrbitError::from)?; + if res.range != (start..end) || end > res.meta.size { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "backend returned a different object range", + ) + .into()); + } + let meta = build_object_meta(&res.meta); + let stream = res.into_stream().map_err(std::io::Error::other); + Ok(Some((Box::pin(stream), meta))) + } + async fn signed_url( &self, key: &ObjectKey, @@ -149,3 +186,55 @@ impl MegaObjectStorage for ObjectStoreAdapter { Ok(()) } } + +#[cfg(test)] +mod exact_range_tests { + use super::*; + + #[tokio::test] + async fn local_exact_range_returns_selected_bytes_and_full_object_size_without_clipping() { + let directory = tempfile::tempdir().unwrap(); + let adapter = ObjectStoreAdapter { + store: BackendStore::Local(Arc::new( + LocalFileSystem::new_with_prefix(directory.path()).unwrap(), + )), + upload_strategy: UploadStrategy::SinglePut, + presign_store: None, + }; + let key = ObjectKey { + namespace: ObjectNamespace::Git, + key: "abcdef1234567890".to_string(), + }; + adapter + .put_stream( + &key, + Box::pin(stream::iter([Ok(Bytes::from_static(b"0123456789"))])), + ObjectMeta::default(), + ) + .await + .unwrap(); + let (input, meta) = adapter + .get_range_stream_exact(&key, 3, 7) + .await + .unwrap() + .unwrap(); + assert_eq!(meta.size, 10); + assert_eq!( + ObjectStoreAdapter::buffer_stream(input, 4).await.unwrap(), + b"3456".as_slice() + ); + let (input, meta) = adapter + .get_range_stream_exact(&key, 9, 10) + .await + .unwrap() + .unwrap(); + assert_eq!(meta.size, 10); + assert_eq!( + ObjectStoreAdapter::buffer_stream(input, 1).await.unwrap(), + b"9".as_slice() + ); + assert!(adapter.get_range_stream_exact(&key, 9, 11).await.is_err()); + assert!(adapter.get_range_stream_exact(&key, 5, 5).await.is_err()); + assert!(adapter.get_range_stream_exact(&key, 10, 11).await.is_err()); + } +} diff --git a/src/orbit_api/object_storage.rs b/src/orbit_api/object_storage.rs index 00e47dda..7fe6f47a 100644 --- a/src/orbit_api/object_storage.rs +++ b/src/orbit_api/object_storage.rs @@ -306,6 +306,20 @@ pub trait MegaObjectStorage: Send + Sync { end: Option, ) -> OrbitResult<(ObjectByteStream, ObjectMeta)>; + /// Exact raw byte range, with no full-download fallback. `None` means + /// unsupported and must be returned without source I/O. Implementations + /// must validate the backend's actual start/end before exposing the stream; + /// metadata describes the complete object. Consumers still verify EOF, + /// length and content hashes. This does not promise bounded producer RSS. + async fn get_range_stream_exact( + &self, + _key: &ObjectKey, + _start: u64, + _end: u64, + ) -> OrbitResult> { + Ok(None) + } + /// Check whether an object exists. async fn exists(&self, key: &ObjectKey) -> OrbitResult; @@ -682,4 +696,22 @@ mod tests { .contains("atomic metadata put is not supported") ); } + + #[tokio::test] + async fn exact_range_default_is_unsupported_without_calling_fallback() { + let store = UnsupportedBoundedStore; + let key = ObjectKey { + namespace: ObjectNamespace::Git, + key: "abcdef".to_string(), + }; + // Its general range getter returns an error; the new default must + // return typed absence without invoking that method or full get. + assert!( + store + .get_range_stream_exact(&key, 1, 2) + .await + .unwrap() + .is_none() + ); + } }