diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index 052c8369..aa8f92bc 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/src/api/router/snapshot_content_tests.rs @@ -131,6 +131,22 @@ async fn await_read_open_hold(hold: Option) { release.notified().await; } } + +struct NoRootedReuse; + +#[async_trait::async_trait] +impl crate::ceres::snapshot::rooted_metadata_projection::RootedReuseLookup for NoRootedReuse { + async fn lookup_reuse( + &self, + _tree_oid: &str, + _identity: &crate::ceres::snapshot::metadata_install::MetadataInstallIdentity, + ) -> Result< + Option, + crate::ceres::snapshot::error::SnapshotError, + > { + Ok(None) + } +} #[path = "snapshot_rooted_metadata_tests.rs"] mod rooted_metadata; @@ -184,6 +200,14 @@ struct CountingStorage { #[async_trait::async_trait] impl MegaObjectStorage for CountingStorage { + fn supports_chunk_map_retention(&self) -> bool { + !self + .counts + .receipt_retention_unsupported + .load(Ordering::SeqCst) + && self.inner.inner.supports_chunk_map_retention() + } + async fn chunk_map_receipt_inventory( &self, ) -> OrbitResult { @@ -1018,6 +1042,242 @@ async fn error(response: Response, status: u16, code: &str, retryable: bool) { assert_eq!(value["error"]["request_id"], request_id); } +#[tokio::test] +async fn mst2_rooted_streamed_fact_persists_one_source_pass_and_reuses_warm_alias_facts() { + use crate::ceres::snapshot::rooted_metadata_projection::prepare_rooted_native_metadata; + + let fixture = Fixture::new().await; + let handler = MonoApiService::from(&fixture.state); + let mut raw = vec![0xff; crate::orbit_api::object_storage::OBJECT_STREAM_ITEM_BYTES + 113]; + let prefix = format!("blob 3\0abc\0{}", uuid::Uuid::new_v4()); + raw[..prefix.len()].copy_from_slice(prefix.as_bytes()); + let oid = fixture + .state + .storage + .git_service + .save_object_from_raw(Bytes::copy_from_slice(&raw)) + .await + .unwrap(); + let source = ObjectHash::from_hex_for_kind(HashKind::Sha1, &oid).unwrap(); + let root = tree(vec![ + item(TreeItemMode::Blob, source, "alias"), + item(TreeItemMode::Blob, source, "file"), + ]); + let mono = fixture.state.storage.mono_storage(); + assert!( + mono.get_verified_blobs(vec![oid.clone()]) + .await + .unwrap() + .is_empty() + ); + fixture.counts.reset(); + let first = prepare_rooted_native_metadata(&handler, &root, "/", &NoRootedReuse) + .await + .unwrap(); + fixture.counts.assert(1, raw.len()); + assert_eq!(first.work.blob_fetches, 1); + assert_eq!(first.work.raw_bytes_fetched, raw.len() as u64); + assert_eq!(first.work.raw_bytes_hashed, raw.len() as u64); + assert_eq!(first.work.verified_blob_persistence_batches, 1); + let stored = mono.get_verified_blobs(vec![oid.clone()]).await.unwrap(); + assert_eq!(stored[&oid].size, raw.len() as i64); + assert_eq!(stored[&oid].raw_sha256, Sha256::digest(&raw).to_vec()); + fixture.counts.reset(); + let warm = prepare_rooted_native_metadata(&handler, &root, "/", &NoRootedReuse) + .await + .unwrap(); + fixture.counts.assert(0, 0); + assert_eq!(warm.plan.identity, first.plan.identity); + assert_eq!(warm.plan.root, first.plan.root); + assert_eq!(warm.work.blob_fetches, 0); + assert_eq!(warm.work.raw_bytes_fetched, 0); + assert_eq!(warm.work.raw_bytes_hashed, 0); + assert_eq!(warm.work.verified_blob_persistence_batches, 0); + raw.push(b'2'); + let next_oid = fixture + .state + .storage + .git_service + .save_object_from_raw(Bytes::copy_from_slice(&raw)) + .await + .unwrap(); + let next_source = ObjectHash::from_hex_for_kind(HashKind::Sha1, &next_oid).unwrap(); + let next_root = tree(vec![ + item(TreeItemMode::Blob, source, "alias"), + item(TreeItemMode::Blob, next_source, "file"), + ]); + fixture.counts.reset(); + let next = prepare_rooted_native_metadata(&handler, &next_root, "/", &NoRootedReuse) + .await + .unwrap(); + fixture.counts.assert(1, raw.len()); + assert_ne!( + next.plan.identity.tagged_root_tree_oid, + first.plan.identity.tagged_root_tree_oid + ); + assert_ne!(next.plan.root, first.plan.root); + assert_eq!(next.work.blob_fetches, 1); + assert_eq!(next.work.raw_bytes_fetched, raw.len() as u64); + assert_eq!(next.work.raw_bytes_hashed, raw.len() as u64); + assert_eq!(next.work.verified_blob_persistence_batches, 1); + assert_eq!(next.work.verified_blob_facts_loaded, 2); +} + +#[tokio::test] +async fn mst2_rooted_streamed_fact_never_persists_truncated_or_late_failed_sources() { + use crate::ceres::snapshot::rooted_metadata_projection::prepare_rooted_native_metadata; + + let fixture = Fixture::new().await; + let handler = MonoApiService::from(&fixture.state); + for late_error in [false, true] { + let raw = Bytes::from(format!("raw\0{}", uuid::Uuid::new_v4())); + let oid = fixture + .state + .storage + .git_service + .save_object_from_raw(raw.clone()) + .await + .unwrap(); + let source = ObjectHash::from_hex_for_kind(HashKind::Sha1, &oid).unwrap(); + let root = tree(vec![item(TreeItemMode::Blob, source, "file")]); + *fixture.counts.object_fault.lock().unwrap() = Some(bounded_objects::StreamFault { + oid: oid.clone(), + kind: if late_error { + bounded_objects::FaultKind::LateError(raw.clone()) + } else { + bounded_objects::FaultKind::Parts(vec![raw.slice(..raw.len() - 1)]) + }, + }); + fixture.counts.reset(); + let error = prepare_rooted_native_metadata(&handler, &root, "/", &NoRootedReuse) + .await + .unwrap_err(); + assert_eq!( + error.code, + if late_error { + SnapshotErrorCode::ObjectUnavailable + } else { + SnapshotErrorCode::IntegrityError + } + ); + fixture + .counts + .assert(1, raw.len() - usize::from(!late_error)); + assert!( + fixture + .state + .storage + .mono_storage() + .get_verified_blobs(vec![oid]) + .await + .unwrap() + .is_empty() + ); + } +} + +#[tokio::test] +async fn mst2_capabilities_use_actual_backend_without_inventory_and_preserve_metadata_and_objects() +{ + let fixture = Fixture::new().await; + let inventory_before = fixture + .counts + .receipt_inventory_calls + .load(Ordering::SeqCst); + let get_capabilities = || async { + success_json( + fixture + .app + .clone() + .oneshot( + Request::builder() + .uri("/api/v2/snapshots/capabilities") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(), + ) + .await + }; + let supported = get_capabilities().await; + for feature in ["raw_blob", "small_objects", "chunk_reads", "full_hydration"] { + assert_eq!(supported["features"][feature], true); + } + fixture + .counts + .receipt_retention_unsupported + .store(true, Ordering::SeqCst); + let mut unsupported = supported.clone(); + for feature in ["raw_blob", "chunk_reads", "full_hydration"] { + unsupported["features"][feature] = json!(false); + } + for _ in 0..3 { + assert_eq!(get_capabilities().await, unsupported); + } + fixture.counts.assert(0, 0); + assert_eq!( + fixture + .counts + .receipt_inventory_calls + .load(Ordering::SeqCst), + inventory_before + ); + assert_eq!(fixture.counts.receipt_reads.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + let head = fixture.send("HEAD", "blob?path=/file", Body::empty()).await; + assert_eq!(head.status(), 200); + assert_eq!( + head.headers()["content-length"], + fixture.raw.len().to_string() + ); + assert_eq!( + head.headers()["etag"], + format!("\"{}\"", fixture.digest_string()) + ); + assert!(to_bytes(head.into_body(), 1024).await.unwrap().is_empty()); + success_json(fixture.send("GET", "directory?path=/", Body::empty()).await).await; + fixture.counts.assert(0, 0); + for path in ["blob?path=/file", "chunk-map?path=/file"] { + error( + fixture.send("GET", path, Body::empty()).await, + 503, + "TEMPORARY_UNAVAILABLE", + true, + ) + .await; + } + fixture.counts.assert(0, 0); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + let link_digest = digest(b"file"); + let request = json!({"items":[{"path":"/link", "expected_digest":format!("sha256:{}", hex_of(&link_digest))}], "encoding":"identity"}).to_string(); + let response = fixture + .send("POST", "objects", Body::from(request.clone())) + .await; + 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::Object(object), Frame::End(end)] = frames.as_slice() else { + panic!("expected OBJECT and terminal END"); + }; + assert_eq!(object.objects, vec![(link_digest, b"file".to_vec())]); + assert_eq!(end.request_item_count, 1); + assert_eq!(end.logical_bytes, 4); + assert_eq!( + end.request_body_sha256, + <[u8; 32]>::from(Sha256::digest(request.as_bytes())) + ); + fixture.counts.assert(1, 4); + fixture + .counts + .receipt_retention_unsupported + .store(false, Ordering::SeqCst); + assert_eq!(get_capabilities().await, supported); + fixture.counts.assert(1, 4); +} + #[tokio::test] async fn mst2_fixed_head_uses_verified_facts_without_body_reads_and_preserves_raw_bytes() { let fixture = Fixture::new().await; diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 4b568aeb..cb593e72 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -35,6 +35,7 @@ use crate::{ runtime::{now_unix, runtime}, view::{SnapshotView, validate_scope_relative_path}, }, + orbit_api::factory::MegaObjectStorageWrapper, }; pub fn routers(api_state: MonoApiServiceState) -> Router { @@ -534,11 +535,14 @@ fn internal(e: E) -> SnapshotError { SnapshotError::new(SnapshotErrorCode::Internal, e.to_string()) } -async fn capabilities() -> Json { - // Keep this document on the canonical discovery contract consumed by the - // v3 reader. The full delivery path serves the complete metadata/object - // closure, so advertising `full_hydration` as false would make a typed - // resolve reject before it reaches this router. +async fn capabilities(State(state): State) -> Json { + capabilities_document(&state.storage.git_service.obj_storage) +} + +fn capabilities_document(backend: &MegaObjectStorageWrapper) -> Json { + // Full delivery must work from a cold source, including receipt admission. + // Existing warm receipts do not establish this backend-wide contract. + let full_delivery = backend.supports_chunk_map_retention(); Json(json!({ "protocol_versions": [2], "metadata_codecs": [1], @@ -548,10 +552,10 @@ async fn capabilities() -> Json { "directory": true, "lookup": true, "metadata_pages": true, - "raw_blob": true, + "raw_blob": full_delivery, "small_objects": true, - "chunk_reads": true, - "full_hydration": true, + "chunk_reads": full_delivery, + "full_hydration": full_delivery, "region_hints": false, "offline_export": false, }, @@ -1640,9 +1644,10 @@ async fn metadata_pages( mod tests { use super::*; - #[tokio::test] - async fn capabilities_advertise_the_canonical_full_delivery_contract() { - let Json(value) = capabilities().await; + #[test] + fn capabilities_advertise_the_canonical_full_delivery_contract() { + let backend = crate::jupiter::storage::object_storage::mock_object_storage(); + let Json(value) = capabilities_document(&backend); assert_eq!( value, json!({ diff --git a/src/ceres/snapshot/rooted_metadata_projection.rs b/src/ceres/snapshot/rooted_metadata_projection.rs index 086db637..b4b1a517 100644 --- a/src/ceres/snapshot/rooted_metadata_projection.rs +++ b/src/ceres/snapshot/rooted_metadata_projection.rs @@ -4,9 +4,12 @@ use std::{ borrow::Cow, collections::{BTreeMap, BTreeSet, VecDeque}, + sync::Arc, + time::Duration, }; use async_trait::async_trait; +use futures::StreamExt; use git_internal::{hash::ObjectHash, internal::object::tree::Tree}; use mst2_codec::{ descriptor::{ @@ -19,6 +22,7 @@ use sea_orm::ActiveValue::Set; use sha2::{Digest, Sha256}; use super::{ + content_budget::{MemoryBudget, RANGE_WORK_BYTES, projection_budget}, error::{SnapshotError, SnapshotErrorCode}, metadata_install::MetadataInstallIdentity, projection_observation::NATIVE_PROJECTION_REVISION, @@ -31,8 +35,12 @@ use crate::{ ceres::api_service::ApiHandler, common::errors::MegaError, jupiter::storage::{Storage, mono_storage::MST2_VERIFICATION_VERSION}, + orbit_api::object_storage::{ObjectByteStream, ObjectMeta}, }; +const MAX_VERIFIED_FILE_BYTES: u64 = 8_796_093_022_208; +const SOURCE_PROGRESS_TIMEOUT: Duration = Duration::from_secs(30); + #[async_trait] pub(crate) trait RootedReuseLookup: Send + Sync { async fn lookup_reuse( @@ -732,28 +740,12 @@ async fn project_directory 8_796_093_022_208 { - return Err(integrity( - "raw file exceeds the native verified-object size profile", - )); - } - let digest: [u8; 32] = Sha256::digest(&raw).into(); - state.work.raw_bytes_hashed = state - .work - .raw_bytes_hashed - .checked_add(size) - .ok_or_else(budget_limit)?; - drop(raw); - let fact = BlobFact { size, digest }; + let fact = stream_blob_fact_with_resources( + || handler.get_raw_blob_stream_with_meta(&raw_oid), + projection_budget(), + &mut state.work, + ) + .await?; state.blob_facts.insert(raw_oid.clone(), fact); pending_facts.push((raw_oid.clone(), fact)); if pending_facts.len() == 64 { @@ -795,12 +787,107 @@ async fn project_directory( + open: F, + budget: &Arc, + work: &mut RootedProjectionWork, +) -> Result +where + F: FnOnce() -> Fut, + Fut: std::future::Future>, +{ + // Credit bounds consumer-visible processing, not the backend's backing + // allocation or work surviving cancellation. No source bytes accumulate. + let _lease = budget.reserve(RANGE_WORK_BYTES)?; + work.blob_fetches = work.blob_fetches.checked_add(1).ok_or_else(budget_limit)?; + let (mut input, meta) = tokio::time::timeout(SOURCE_PROGRESS_TIMEOUT, open()) + .await + .map_err(|_| source_progress_timeout())? + .map_err(source_error)?; + let size = u64::try_from(meta.size) + .ok() + .filter(|size| *size <= MAX_VERIFIED_FILE_BYTES) + .ok_or_else(|| integrity("raw file exceeds the native verified-object size profile"))?; + let mut digest = Sha256::new(); + let mut received = 0u64; + let mut since_yield = 0usize; + let mut empty_parts = 0usize; + let mut progress_deadline = tokio::time::Instant::now() + SOURCE_PROGRESS_TIMEOUT; + loop { + if tokio::time::Instant::now() >= progress_deadline { + return Err(source_progress_timeout()); + } + let part = tokio::time::timeout_at(progress_deadline, input.next()) + .await + .map_err(|_| source_progress_timeout())?; + let Some(part) = part else { break }; + let bytes = part.map_err(|error| { + tracing::warn!(error = %error, "rooted metadata raw source stream failed"); + SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "raw source stream failed", + ) + })?; + if bytes.len() as u64 > size - received { + return Err(integrity( + "raw source length disagrees with its physical size", + )); + } + if bytes.len() > RANGE_WORK_BYTES { + return Err(SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "raw source producer item exceeds the consumer processing limit", + )); + } + digest.update(&bytes); + received += bytes.len() as u64; + since_yield += bytes.len(); + if bytes.is_empty() { + empty_parts += 1; + } else { + empty_parts = 0; + progress_deadline = tokio::time::Instant::now() + SOURCE_PROGRESS_TIMEOUT; + } + if since_yield >= RANGE_WORK_BYTES || empty_parts == 32 { + tokio::task::yield_now().await; + since_yield = 0; + empty_parts = 0; + } + } + if received != size { + return Err(integrity( + "raw source length disagrees with its physical size", + )); + } + let fetched = work + .raw_bytes_fetched + .checked_add(size) + .ok_or_else(budget_limit)?; + let hashed = work + .raw_bytes_hashed + .checked_add(size) + .ok_or_else(budget_limit)?; + work.raw_bytes_fetched = fetched; + work.raw_bytes_hashed = hashed; + Ok(BlobFact { + size, + digest: digest.finalize().into(), + }) +} + +fn source_progress_timeout() -> SnapshotError { + SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "rooted metadata raw source made no progress", + ) +} + fn verified_fact( fact: &crate::callisto::mst2_verified_object::Model, ) -> Result { if fact.state != "VERIFIED" || fact.verification_version != MST2_VERIFICATION_VERSION - || !(0..=8_796_093_022_208).contains(&fact.size) + || !(0..=MAX_VERIFIED_FILE_BYTES as i64).contains(&fact.size) { return Err(integrity( "native projection received an invalid current verified blob fact", @@ -894,11 +981,293 @@ fn path_limit() -> SnapshotError { #[cfg(test)] mod tests { + use std::{ + pin::Pin, + sync::atomic::{AtomicUsize, Ordering}, + task::{Context, Poll}, + }; + + use bytes::Bytes; use git_internal::hash::HashKind; + use tokio::sync::Notify; use uuid::Uuid; use super::*; + struct FactInput { + parts: VecDeque>, + polls: Arc, + drops: Arc, + hold_eof: Option>, + } + + impl futures::Stream for FactInput { + type Item = std::io::Result; + + fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + let input = self.get_mut(); + input.polls.fetch_add(1, Ordering::SeqCst); + if let Some(part) = input.parts.pop_front() { + Poll::Ready(Some(part)) + } else if let Some(entered) = &input.hold_eof { + entered.notify_one(); + Poll::Pending + } else { + Poll::Ready(None) + } + } + } + + impl Drop for FactInput { + fn drop(&mut self) { + self.drops.fetch_add(1, Ordering::SeqCst); + } + } + + fn fact_input( + parts: Vec>, + hold_eof: Option>, + ) -> (ObjectByteStream, Arc, Arc) { + let polls = Arc::new(AtomicUsize::new(0)); + let drops = Arc::new(AtomicUsize::new(0)); + ( + Box::pin(FactInput { + parts: parts.into(), + polls: polls.clone(), + drops: drops.clone(), + hold_eof, + }), + polls, + drops, + ) + } + + #[tokio::test] + async fn streamed_cold_fact_keeps_large_memory_source_and_empty_raw_costs_exact() { + use crate::orbit_api::object_storage::{ObjectKey, ObjectNamespace}; + + for size in [0, RANGE_WORK_BYTES + 113] { + let backend = crate::jupiter::storage::object_storage::mock_object_storage(); + let key = ObjectKey { + namespace: ObjectNamespace::Git, + key: "ab".repeat(20), + }; + let mut raw = vec![0xff; size]; + if !raw.is_empty() { + raw[..11].copy_from_slice(b"blob 3\0abc\0"); + } + let expected: [u8; 32] = Sha256::digest(&raw).into(); + backend + .inner + .put_stream( + &key, + Box::pin(futures::stream::iter([Ok(Bytes::from(raw))])), + ObjectMeta::default(), + ) + .await + .unwrap(); + let budget = MemoryBudget::new(RANGE_WORK_BYTES); + let opens = AtomicUsize::new(0); + let mut work = RootedProjectionWork::default(); + let fact = stream_blob_fact_with_resources( + || async { + opens.fetch_add(1, Ordering::SeqCst); + backend + .inner + .get_stream(&key) + .await + .map_err(MegaError::from) + }, + &budget, + &mut work, + ) + .await + .unwrap(); + assert_eq!( + fact, + BlobFact { + size: size as u64, + digest: expected + } + ); + assert_eq!(opens.load(Ordering::SeqCst), 1); + assert_eq!(work.blob_fetches, 1); + assert_eq!(work.raw_bytes_fetched, size as u64); + assert_eq!(work.raw_bytes_hashed, size as u64); + assert_eq!(work.verified_blob_persistence_batches, 0); + assert_eq!(budget.used(), 0); + } + } + + #[tokio::test] + async fn streamed_cold_fact_rejects_physical_mismatch_and_late_error_before_fact_return() { + let raw = Bytes::from_static(b"raw"); + for (size, parts, code, polls_expected) in [ + ( + -1, + vec![Ok(raw.clone())], + SnapshotErrorCode::IntegrityError, + 0, + ), + ( + MAX_VERIFIED_FILE_BYTES as i64 + 1, + vec![Ok(raw.clone())], + SnapshotErrorCode::IntegrityError, + 0, + ), + ( + MAX_VERIFIED_FILE_BYTES as i64, + vec![], + SnapshotErrorCode::IntegrityError, + 1, + ), + ( + RANGE_WORK_BYTES as i64 + 1, + vec![Ok(Bytes::from(vec![7; RANGE_WORK_BYTES + 1]))], + SnapshotErrorCode::TemporaryUnavailable, + 1, + ), + ( + 2, + vec![Ok(raw.clone())], + SnapshotErrorCode::IntegrityError, + 1, + ), + ( + 4, + vec![Ok(raw.clone())], + SnapshotErrorCode::IntegrityError, + 2, + ), + ( + 3, + vec![Ok(raw.clone()), Ok(Bytes::from_static(b"x"))], + SnapshotErrorCode::IntegrityError, + 2, + ), + ( + 3, + vec![Ok(raw), Err(std::io::Error::other("late source error"))], + SnapshotErrorCode::ObjectUnavailable, + 2, + ), + ( + 0, + vec![Err(std::io::Error::other("empty source error"))], + SnapshotErrorCode::ObjectUnavailable, + 1, + ), + ] { + let (input, polls, drops) = fact_input(parts, None); + let budget = MemoryBudget::new(RANGE_WORK_BYTES); + let mut work = RootedProjectionWork::default(); + let error = stream_blob_fact_with_resources( + || async move { + Ok(( + input, + ObjectMeta { + size, + ..Default::default() + }, + )) + }, + &budget, + &mut work, + ) + .await + .unwrap_err(); + assert_eq!(error.code, code); + assert_eq!(polls.load(Ordering::SeqCst), polls_expected); + assert_eq!(drops.load(Ordering::SeqCst), 1); + assert_eq!(work.blob_fetches, 1); + assert_eq!(work.raw_bytes_fetched, 0); + assert_eq!(work.raw_bytes_hashed, 0); + assert_eq!(budget.used(), 0); + } + } + + #[tokio::test] + async fn streamed_cold_fact_cancellation_at_exact_size_drops_source_and_refunds_credit() { + let entered = Arc::new(Notify::new()); + let (input, polls, drops) = + fact_input(vec![Ok(Bytes::from_static(b"raw"))], Some(entered.clone())); + let budget = MemoryBudget::new(RANGE_WORK_BYTES); + let task_budget = budget.clone(); + let task = tokio::spawn(async move { + let mut work = RootedProjectionWork::default(); + stream_blob_fact_with_resources( + || async move { + Ok(( + input, + ObjectMeta { + size: 3, + ..Default::default() + }, + )) + }, + &task_budget, + &mut work, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(2), entered.notified()) + .await + .unwrap(); + assert_eq!(polls.load(Ordering::SeqCst), 2); + assert_eq!(budget.used(), RANGE_WORK_BYTES); + assert_eq!(drops.load(Ordering::SeqCst), 0); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + assert_eq!(drops.load(Ordering::SeqCst), 1); + assert_eq!(budget.used(), 0); + } + + #[tokio::test] + async fn streamed_cold_fact_reserves_fixed_credit_before_open_and_preserves_open_errors() { + let budget = MemoryBudget::new(RANGE_WORK_BYTES); + let occupied = budget.reserve(RANGE_WORK_BYTES).unwrap(); + let mut work = RootedProjectionWork::default(); + let opens = AtomicUsize::new(0); + let error = stream_blob_fact_with_resources( + || async { + opens.fetch_add(1, Ordering::SeqCst); + Err(MegaError::Other("unexpected open".into())) + }, + &budget, + &mut work, + ) + .await + .unwrap_err(); + assert_eq!(error.code, SnapshotErrorCode::TemporaryUnavailable); + assert_eq!(opens.load(Ordering::SeqCst), 0); + assert_eq!(work, RootedProjectionWork::default()); + drop(occupied); + for (error, code) in [ + ( + MegaError::ObjStorageNotFound("missing".into()), + SnapshotErrorCode::ObjectUnavailable, + ), + ( + MegaError::ObjStorageInconsistent("inconsistent".into()), + SnapshotErrorCode::IntegrityError, + ), + ( + MegaError::Other("transport".into()), + SnapshotErrorCode::Internal, + ), + ] { + let mut work = RootedProjectionWork::default(); + let got = stream_blob_fact_with_resources(|| async { Err(error) }, &budget, &mut work) + .await + .unwrap_err(); + assert_eq!(got.code, code); + assert_eq!(work.blob_fetches, 1); + assert_eq!(work.raw_bytes_fetched, 0); + assert_eq!(work.raw_bytes_hashed, 0); + assert_eq!(budget.used(), 0); + } + } + fn state() -> ProjectionState { ProjectionState::new(MetadataInstallIdentity { source_domain: "native-git".into(), diff --git a/src/jupiter/storage/object_storage.rs b/src/jupiter/storage/object_storage.rs index 3fbe45d8..891232cf 100644 --- a/src/jupiter/storage/object_storage.rs +++ b/src/jupiter/storage/object_storage.rs @@ -199,7 +199,9 @@ impl InMemoryObjectStorage { } fn stream_bytes(bytes: Bytes) -> ObjectByteStream { - Box::pin(futures::stream::once(async move { Ok(bytes) })) + crate::orbit_api::object_storage::fragment_object_stream(Box::pin(futures::stream::once( + async move { Ok(bytes) }, + ))) } } @@ -209,6 +211,10 @@ pub fn mock_object_storage() -> MegaObjectStorageWrapper { #[async_trait::async_trait] impl MegaObjectStorage for InMemoryObjectStorage { + fn supports_chunk_map_retention(&self) -> bool { + true + } + async fn put_stream( &self, key: &ObjectKey, diff --git a/src/orbit/adapter/object.rs b/src/orbit/adapter/object.rs index 4ae923b0..14c3d175 100644 --- a/src/orbit/adapter/object.rs +++ b/src/orbit/adapter/object.rs @@ -6,6 +6,10 @@ impl MegaObjectStorage for ObjectStoreAdapter { matches!(&self.store, BackendStore::S3(_) | BackendStore::Gcs(_)) } + fn supports_chunk_map_retention(&self) -> bool { + matches!(&self.store, BackendStore::Local(_)) + } + async fn put_stream( &self, key: &ObjectKey, @@ -101,7 +105,10 @@ impl MegaObjectStorage for ObjectStoreAdapter { let meta = build_object_meta(&res.meta); let stream = res.into_stream().map_err(std::io::Error::other); - Ok((Box::pin(stream), meta)) + Ok(( + crate::orbit_api::object_storage::fragment_object_stream(Box::pin(stream)), + meta, + )) } async fn chunk_map_receipt_inventory( @@ -112,7 +119,7 @@ impl MegaObjectStorage for ObjectStoreAdapter { }; // object_store 0.14 cannot inventory/delete retained cloud versions. // A current-object listing is not a physical backing-byte quota. - if !matches!(&self.store, BackendStore::Local(_)) { + if !self.supports_chunk_map_retention() { return Err(IoOrbitError::ChunkMapRetentionUnsupported); } let prefix = object_store::path::Path::from("chunk-map-receipt"); @@ -161,7 +168,7 @@ impl MegaObjectStorage for ObjectStoreAdapter { &self, authority: &crate::orbit_api::object_storage::ChunkMapReceiptDeletion, ) -> OrbitResult { - if !matches!(&self.store, BackendStore::Local(_)) { + if !self.supports_chunk_map_retention() { return Err(IoOrbitError::ChunkMapRetentionUnsupported); } let key = authority.key(); @@ -341,6 +348,91 @@ pub(super) fn reject_receipt_mutation(key: &ObjectKey) -> OrbitResult<()> { mod exact_range_tests { use super::*; + #[tokio::test] + async fn backend_chunk_retention_capabilities_match_cold_admission_without_cloud_io() { + let s3 = object_store::aws::AmazonS3Builder::new() + .with_bucket_name("capabilities-test") + .with_region("us-east-1") + .with_access_key_id("fixture-access") + .with_secret_access_key("fixture-secret") + .with_skip_signature(true) + .build() + .unwrap(); + let gcs = object_store::gcp::GoogleCloudStorageBuilder::new() + .with_bucket_name("capabilities-test") + .with_credentials(Arc::new(object_store::StaticCredentialProvider::new( + object_store::gcp::GcpCredential { + bearer: "fixture-bearer".into(), + }, + ))) + .with_skip_signature(true) + .build() + .unwrap(); + for store in [ + BackendStore::S3(Arc::new(s3)), + BackendStore::Gcs(Arc::new(gcs)), + ] { + let backend = crate::orbit_api::factory::MegaObjectStorageWrapper::new(Arc::new( + ObjectStoreAdapter { + store, + upload_strategy: UploadStrategy::SinglePut, + presign_store: None, + }, + )); + assert!(!backend.supports_chunk_map_retention()); + assert!(matches!( + backend.inner.chunk_map_receipt_inventory().await, + Err(IoOrbitError::ChunkMapRetentionUnsupported) + )); + } + } + + #[tokio::test] + async fn local_whole_stream_keeps_large_file_size_and_digest_with_bounded_fragments() { + use sha2::{Digest, Sha256}; + + 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: "ab".repeat(20), + }; + let raw = Bytes::from(vec![ + 0xff; + crate::orbit_api::object_storage::OBJECT_STREAM_ITEM_BYTES + + 113 + ]); + let expected: [u8; 32] = Sha256::digest(&raw).into(); + let size = raw.len(); + adapter + .put_stream( + &key, + Box::pin(stream::iter([Ok(raw)])), + ObjectMeta::default(), + ) + .await + .unwrap(); + let (mut input, meta) = adapter.get_stream(&key).await.unwrap(); + assert_eq!(meta.size, size as i64); + assert!(adapter.supports_chunk_map_retention()); + let mut received = 0; + let mut hashed = Sha256::new(); + while let Some(part) = input.next().await { + let part = part.unwrap(); + assert!(part.len() <= crate::orbit_api::object_storage::OBJECT_STREAM_ITEM_BYTES); + received += part.len(); + hashed.update(&part); + } + assert_eq!(received, size); + assert_eq!(<[u8; 32]>::from(hashed.finalize()), expected); + } + #[tokio::test] async fn local_chunk_receipts_reject_all_mutation_routes() { let directory = tempfile::tempdir().unwrap(); diff --git a/src/orbit_api/factory.rs b/src/orbit_api/factory.rs index c1f8e86e..f6891a18 100644 --- a/src/orbit_api/factory.rs +++ b/src/orbit_api/factory.rs @@ -135,6 +135,10 @@ impl MegaObjectStorageWrapper { pub fn supports_presigned_urls(&self) -> bool { MegaObjectStorage::supports_presigned_urls(&*self.inner) } + + pub fn supports_chunk_map_retention(&self) -> bool { + MegaObjectStorage::supports_chunk_map_retention(&*self.inner) + } } #[cfg(test)] diff --git a/src/orbit_api/object_storage.rs b/src/orbit_api/object_storage.rs index 9264d944..e3898424 100644 --- a/src/orbit_api/object_storage.rs +++ b/src/orbit_api/object_storage.rs @@ -168,6 +168,30 @@ pub struct ObjectMeta { /// - The stream must be fully consumed by the caller. pub type ObjectByteStream = Pin> + Send>>; +pub(crate) const OBJECT_STREAM_ITEM_BYTES: usize = 8 * 1024 * 1024; + +/// Split visible items without copying or prebuilding a fragment list. Bytes +/// retain their original owner; this does not bound backend backing buffers. +pub(crate) fn fragment_object_stream(input: ObjectByteStream) -> ObjectByteStream { + Box::pin(futures::stream::unfold( + (input, Bytes::new()), + |(mut input, pending)| async move { + let next = if pending.is_empty() { + input.next().await? + } else { + Ok(pending) + }; + match next { + Ok(mut bytes) => { + let part = bytes.split_to(bytes.len().min(OBJECT_STREAM_ITEM_BYTES)); + Some((Ok(part), (input, bytes))) + } + Err(error) => Some((Err(error), (input, Bytes::new()))), + } + }, + )) +} + /// Upper bound for [`MegaObjectStorage::put_metadata_atomic`] (ADR-MF-05). pub const MAX_METADATA_ATOMIC_BYTES: usize = 1024 * 1024; @@ -232,6 +256,13 @@ pub trait MegaObjectStorage: Send + Sync { false } + /// Complete receipt inventory and sealed deletion are available without + /// uncounted retained versions. This is a static backend contract, not an + /// I/O health or quota check; cold requests still perform real admission. + fn supports_chunk_map_retention(&self) -> bool { + false + } + /// Upload a single object to the storage backend. /// /// # Parameters @@ -676,6 +707,80 @@ mod tests { struct UnsupportedBoundedStore; + struct SharedStreamOwner { + bytes: Vec, + drops: std::sync::Arc, + } + + impl AsRef<[u8]> for SharedStreamOwner { + fn as_ref(&self) -> &[u8] { + &self.bytes + } + } + + impl Drop for SharedStreamOwner { + fn drop(&mut self) { + self.drops.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + } + + #[tokio::test] + async fn fragmented_source_is_lazy_zero_copy_and_keeps_original_owner_until_last_clone() { + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + + let drops = Arc::new(AtomicUsize::new(0)); + let input_polls = Arc::new(AtomicUsize::new(0)); + let bytes = Bytes::from_owner(SharedStreamOwner { + bytes: vec![0xa5; 2 * OBJECT_STREAM_ITEM_BYTES + 17], + drops: drops.clone(), + }); + let pointer = bytes.as_ptr(); + let polls = input_polls.clone(); + let input = + futures::stream::iter([Ok(bytes), Err(std::io::Error::other("late backend error"))]) + .inspect(move |_| { + polls.fetch_add(1, Ordering::SeqCst); + }); + let mut fragmented = fragment_object_stream(Box::pin(input)); + assert_eq!(input_polls.load(Ordering::SeqCst), 0); + let first = fragmented.next().await.unwrap().unwrap(); + assert_eq!(first.len(), OBJECT_STREAM_ITEM_BYTES); + assert_eq!(first.as_ptr(), pointer); + let transport = first.clone(); + drop(first); + let second = fragmented.next().await.unwrap().unwrap(); + assert_eq!(second.len(), OBJECT_STREAM_ITEM_BYTES); + assert_eq!( + second.as_ptr() as usize, + pointer as usize + OBJECT_STREAM_ITEM_BYTES + ); + drop(second); + let tail = fragmented.next().await.unwrap().unwrap(); + assert_eq!(tail.as_ref(), &[0xa5; 17]); + drop(tail); + assert_eq!(input_polls.load(Ordering::SeqCst), 1); + assert!(fragmented.next().await.unwrap().is_err()); + assert_eq!(input_polls.load(Ordering::SeqCst), 2); + assert!(fragmented.next().await.is_none()); + drop(fragmented); + assert_eq!(drops.load(Ordering::SeqCst), 0); + drop(transport); + assert_eq!(drops.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn default_backend_does_not_advertise_or_infer_chunk_retention() { + let backend = UnsupportedBoundedStore; + assert!(!backend.supports_chunk_map_retention()); + assert!(matches!( + backend.chunk_map_receipt_inventory().await, + Err(IoOrbitError::ChunkMapRetentionUnsupported) + )); + } + #[async_trait::async_trait] impl MegaObjectStorage for UnsupportedBoundedStore { async fn put_stream(