diff --git a/src/api/router/snapshot_content.rs b/src/api/router/snapshot_content.rs index d642434f..6a1f191f 100644 --- a/src/api/router/snapshot_content.rs +++ b/src/api/router/snapshot_content.rs @@ -27,15 +27,15 @@ use crate::ceres::snapshot::{ }; pub(super) struct ResolvedFileMetadata { - fs_kind: FsKind, - oid: String, + pub(super) fs_kind: FsKind, + pub(super) oid: String, pub(super) digest: [u8; 32], pub(super) size: u64, fact: crate::callisto::mst2_verified_object::Model, } #[allow(clippy::result_large_err)] -async fn fixed_root_tree( +pub(super) async fn fixed_root_tree( handler: &T, oid: &str, ) -> Result { @@ -50,7 +50,7 @@ async fn fixed_root_tree( } #[allow(clippy::result_large_err)] -async fn resolve_file_metadata( +pub(super) async fn resolve_file_metadata( handler: &T, root_tree: &git_internal::internal::object::tree::Tree, scope: &str, @@ -479,7 +479,7 @@ async fn project_for( } #[allow(clippy::result_large_err)] -async fn project_resolved( +pub(super) async fn project_resolved( handler: &T, f: &ResolvedFileMetadata, ) -> Result, Response> @@ -789,7 +789,7 @@ struct ResolvedChunk { map_id: String, } -fn content_read_error(error: crate::common::errors::MegaError) -> SnapshotError { +pub(super) fn content_read_error(error: crate::common::errors::MegaError) -> SnapshotError { use crate::common::errors::MegaError; let code = match &error { MegaError::ObjStorageNotFound(_) => SnapshotErrorCode::ObjectUnavailable, diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index 4399e692..617cf2ea 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/src/api/router/snapshot_content_tests.rs @@ -88,6 +88,9 @@ mod bounded_chunks; #[path = "snapshot_persisted_chunk_map_tests.rs"] mod persisted_chunk_maps; +#[path = "snapshot_raw_blob_tests.rs"] +mod raw_blob; + #[path = "snapshot_session_tests.rs"] mod durable_sessions; @@ -111,6 +114,7 @@ struct ReadCounts { range: AtomicUsize, bytes: AtomicUsize, object_fault: std::sync::Mutex>, + object_size_override: std::sync::Mutex>, chunk_faults: std::sync::Mutex>, receipt_reads: AtomicUsize, receipt_writes: AtomicUsize, @@ -244,7 +248,10 @@ impl MegaObjectStorage for CountingStorage { }, )); } - let (stream, meta) = self.inner.inner.get_stream(key).await?; + let (stream, mut meta) = self.inner.inner.get_stream(key).await?; + if let Some(size) = *self.counts.object_size_override.lock().unwrap() { + meta.size = size; + } let fault = self.counts.object_fault.lock().unwrap().clone(); let stream = match fault { Some(fault) if fault.oid == key.key => fault.stream(), @@ -904,7 +911,8 @@ async fn mst2_fixed_head_uses_verified_facts_without_body_reads_and_preserves_ra .await .unwrap(); assert_eq!(bytes.as_ref(), fixture.raw); - fixture.counts.assert(1, fixture.raw.len()); + fixture.counts.assert(2, 2 * fixture.raw.len()); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 1); } #[tokio::test] diff --git a/src/api/router/snapshot_objects_bounded_tests.rs b/src/api/router/snapshot_objects_bounded_tests.rs index b5e0d303..e7eb0bfa 100644 --- a/src/api/router/snapshot_objects_bounded_tests.rs +++ b/src/api/router/snapshot_objects_bounded_tests.rs @@ -26,6 +26,14 @@ pub(super) enum FaultKind { release: Arc, drops: Arc, }, + HeldFragment { + prefix: Bytes, + fragment: Option, + entered: Arc, + release: Arc, + tail_polls: Arc, + drops: Arc, + }, } struct DropCount(Arc); @@ -39,6 +47,46 @@ impl Drop for DropCount { impl StreamFault { pub(super) fn stream(self) -> ObjectByteStream { match self.kind { + FaultKind::HeldFragment { + prefix, + fragment, + entered, + release, + tail_polls, + drops, + } => Box::pin(futures::stream::unfold( + ( + prefix, + fragment, + entered, + release, + tail_polls, + DropCount(drops), + 0u8, + ), + |(prefix, fragment, entered, release, tail_polls, owner, turn)| async move { + match turn { + 0 => Some(( + Ok(prefix.clone()), + (prefix, fragment, entered, release, tail_polls, owner, 1), + )), + 1 => { + entered.notify_one(); + release.notified().await; + fragment.clone().map(|part| { + ( + Ok(part), + (prefix, fragment, entered, release, tail_polls, owner, 2), + ) + }) + } + _ => { + tail_polls.fetch_add(1, Ordering::SeqCst); + std::future::pending().await + } + } + }, + )), FaultKind::HeldThenError { entered, release, diff --git a/src/api/router/snapshot_raw_blob.rs b/src/api/router/snapshot_raw_blob.rs new file mode 100644 index 00000000..efda9649 --- /dev/null +++ b/src/api/router/snapshot_raw_blob.rs @@ -0,0 +1,390 @@ +//! Whole-file raw delivery from one current-OID stream, authenticated in chunks. + +use std::sync::Arc; + +use axum::{ + body::Body, + extract::{Path as AxumPath, Query, State}, + http::HeaderMap, + response::Response, +}; +use bytes::Bytes; +use futures::StreamExt; +use mst2_codec::chunkmap::{CHUNK_SIZE, CHUNKS_PER_PAGE}; +use serde::Deserialize; +use sha2::{Digest, Sha256}; + +use super::{ + REQUEST_HEADERS, content, ensure_enabled, internal, mst2_error_response, revalidate_access, + revalidate_request, +}; +use crate::{ + api::MonoApiServiceState, + ceres::snapshot::{ + content_budget::{ + MemoryBudget, MemoryLease, RANGE_WORK_BYTES, range_budget, response_budget, + }, + error::{SnapshotError, SnapshotErrorCode}, + pages::hex_of, + runtime::SnapshotContext, + view::validate_scope_relative_path, + }, + jupiter::storage::{ + Storage, + native_chunk_map::{AuthenticatedChunkPage, PersistedChunkMap}, + }, + orbit_api::object_storage::ObjectByteStream, +}; + +const PRODUCER_ITEM_MAX: usize = 8 * 1024 * 1024; + +#[derive(Deserialize, Debug)] +pub(super) struct BlobQuery { + path: String, + #[serde(default)] + expected_digest: Option, +} + +pub(super) struct BlobBudgets { + pub(super) scratch: Arc, + pub(super) response: Arc, +} + +impl Default for BlobBudgets { + fn default() -> Self { + Self { + scratch: range_budget().clone(), + response: response_budget().clone(), + } + } +} + +pub(super) async fn blob( + state: State, + path: AxumPath, + query: Query, + headers: HeaderMap, +) -> Response { + match blob_with_budgets(state, path, query, headers, BlobBudgets::default()).await { + Ok(response) | Err(response) => response, + } +} + +#[allow(clippy::result_large_err)] +pub(super) async fn blob_with_budgets( + state: State, + AxumPath(snapshot_id): AxumPath, + Query(query): Query, + headers: HeaderMap, + budgets: BlobBudgets, +) -> Result { + ensure_enabled(&state).map_err(mst2_error_response)?; + let context = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?; + validate_scope_relative_path(&query.path).map_err(mst2_error_response)?; + if headers.contains_key("range") { + return Err(mst2_error_response(SnapshotError::new( + SnapshotErrorCode::RangeNotSupported, + "raw blob reads are whole-file; use chunks for ranges", + ))); + } + let handler = state + .api_handler(std::path::Path::new("/")) + .await + .map_err(internal)?; + let root = content::fixed_root_tree(handler.as_ref(), &context.root_tree_oid).await?; + let file = content::resolve_file_metadata( + handler.as_ref(), + &root, + &context.built.descriptor.scope, + &query.path, + query.expected_digest.as_deref(), + ) + .await?; + let scratch = budgets + .scratch + .reserve(if file.size == 0 { 0 } else { RANGE_WORK_BYTES }) + .map_err(mst2_error_response)?; + let first_credit = budgets + .response + .reserve(file.size.min(CHUNK_SIZE as u64) as usize) + .map_err(mst2_error_response)?; + revalidate_request(&state, &context) + .await + .map_err(mst2_error_response)?; + let map = if file.size == 0 { + if file.digest != <[u8; 32]>::from(Sha256::digest([])) { + return Err(mst2_error_response(integrity( + "empty raw source has a nonempty verified digest", + ))); + } + None + } else { + Some(content::project_resolved(handler.as_ref(), &file).await?) + }; + let storage = handler.get_context(); + let page = if let Some(map) = map.as_ref() { + Some( + storage + .chunk_maps() + .await + .map_err(mst2_error_response)? + .selected_page(map, 0) + .await + .map_err(mst2_error_response)?, + ) + } else { + None + }; + let (input, meta) = handler + .get_raw_blob_stream_with_meta(&file.oid) + .await + .map_err(|error| mst2_error_response(content::content_read_error(error)))?; + if u64::try_from(meta.size).ok() != Some(file.size) { + return Err(mst2_error_response(integrity( + "raw source physical size disagrees with its fixed verified fact", + ))); + } + let headers = REQUEST_HEADERS.try_with(Clone::clone).map_err(|_| { + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::Unauthenticated, + "request authentication context missing", + )) + })?; + revalidate_access(&state, &context, &headers) + .await + .map_err(mst2_error_response)?; + let digest = file.digest; + let size = file.size; + let kind = content::fs_kind_str(file.fs_kind); + let mut reader = RawBlobReader { + state: state.0, + context, + headers, + storage, + map, + page, + input: Some(input), + pending: Bytes::new(), + pending_offset: 0, + received: 0, + index: 0, + size, + digest, + hash: Sha256::new(), + first_credit: Some(first_credit), + response_budget: budgets.response, + _scratch: scratch, + }; + let body = if size == 0 { + reader.require_eof().await.map_err(mst2_error_response)?; + reader.validate().await.map_err(mst2_error_response)?; + Body::empty() + } else { + let stream = futures::stream::try_unfold(reader, |mut reader| async move { + let Some(bytes) = reader.next_chunk().await? else { + return Ok::<_, SnapshotError>(None); + }; + Ok(Some((bytes, reader))) + }); + Body::from_stream(stream) + }; + Response::builder() + .header("etag", format!("\"sha256:{}\"", hex_of(&digest))) + .header("content-length", size.to_string()) + .header("x-mega-content-size", size.to_string()) + .header("x-mega-fs-kind", kind) + .header("cache-control", "private, no-cache, no-transform") + .header("vary", "Authorization, Accept") + .body(body) + .map_err(|_| mst2_error_response(internal("raw blob response build failed"))) +} + +struct RawBlobReader { + state: MonoApiServiceState, + context: SnapshotContext, + headers: HeaderMap, + storage: Storage, + map: Option>, + page: Option>, + input: Option, + pending: Bytes, + pending_offset: usize, + received: u64, + index: u64, + size: u64, + digest: [u8; 32], + hash: Sha256, + first_credit: Option, + response_budget: Arc, + _scratch: MemoryLease, +} + +struct RawBlobChunk { + bytes: Vec, + _credit: MemoryLease, + _map: Arc, +} + +impl AsRef<[u8]> for RawBlobChunk { + fn as_ref(&self) -> &[u8] { + &self.bytes + } +} + +impl RawBlobReader { + async fn validate(&self) -> Result<(), SnapshotError> { + revalidate_access(&self.state, &self.context, &self.headers).await + } + + async fn next_chunk(&mut self) -> Result, SnapshotError> { + self.validate().await?; + let map = self + .map + .as_ref() + .ok_or_else(|| integrity("nonempty raw source has no authenticated map"))?; + if self.index == map.map.chunk_count { + return Ok(None); + } + let length = map + .map + .chunk_len(self.index) + .map_err(|_| integrity("raw source chunk index disagrees with its map"))? + as usize; + let page_index = self.index / CHUNKS_PER_PAGE as u64; + if self.page.as_ref().map(|p| p.leaf.page_index) != Some(page_index) { + drop(self.page.take()); + self.page = Some( + self.storage + .chunk_maps() + .await? + .selected_page(map, page_index) + .await?, + ); + self.validate().await?; + } + let credit = if let Some(credit) = self.first_credit.take() { + credit + } else { + self.response_budget.reserve(length)? + }; + let mut raw = Vec::new(); + raw.try_reserve_exact(length).map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "raw chunk allocation could not be admitted", + ) + })?; + let mut empty_parts = 0; + while raw.len() < length { + if self.pending_offset == self.pending.len() { + self.pending = Bytes::new(); + self.pending_offset = 0; + let Some(part) = self.poll_source().await? else { + return Err(integrity("raw source is truncated before its final chunk")); + }; + if part.len() as u64 > self.size - self.received { + return Err(integrity("raw source exceeds its fixed verified size")); + } + if part.len() > PRODUCER_ITEM_MAX { + return Err(SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "raw producer item exceeds its consumer processing limit", + )); + } + self.received += part.len() as u64; + if part.is_empty() { + empty_parts += 1; + if empty_parts == 32 { + tokio::task::yield_now().await; + self.validate().await?; + empty_parts = 0; + } + continue; + } + self.pending = part; + } + let amount = (length - raw.len()).min(self.pending.len() - self.pending_offset); + let end = self.pending_offset + amount; + let bytes = &self.pending[self.pending_offset..end]; + self.hash.update(bytes); + raw.extend_from_slice(bytes); + self.pending_offset = end; + } + let map = self + .map + .as_ref() + .ok_or_else(|| integrity("raw source authenticated map disappeared"))?; + self.page + .as_ref() + .ok_or_else(|| integrity("raw source authenticated page disappeared"))? + .verify_chunk(&map.map, self.index, &raw)?; + self.index += 1; + if self.index == map.map.chunk_count { + self.require_eof().await?; + let actual: [u8; 32] = self.hash.clone().finalize().into(); + if actual != self.digest { + return Err(integrity( + "raw source whole-file digest disagrees with its fixed fact", + )); + } + } + self.validate().await?; + if raw.capacity() > credit.bytes { + return Err(integrity("raw chunk allocation exceeds its owned credit")); + } + Ok(Some(Bytes::from_owner(RawBlobChunk { + bytes: raw, + _credit: credit, + _map: self + .map + .as_ref() + .cloned() + .ok_or_else(|| integrity("raw source authenticated map disappeared"))?, + }))) + } + + async fn poll_source(&mut self) -> Result, SnapshotError> { + let input = self + .input + .as_mut() + .ok_or_else(|| integrity("raw source stream was already closed"))?; + let result = input.next().await; + self.validate().await?; + result.transpose().map_err(|error| { + tracing::warn!(%error, "fixed-view raw stream failed"); + SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "fixed-view raw stream failed", + ) + }) + } + + async fn require_eof(&mut self) -> Result<(), SnapshotError> { + if self.received != self.size || self.pending_offset != self.pending.len() { + return Err(integrity( + "raw source final length disagrees with its fixed fact", + )); + } + let mut empty_parts = 0; + while let Some(part) = self.poll_source().await? { + if !part.is_empty() { + return Err(integrity( + "raw source has trailing bytes after its fixed size", + )); + } + empty_parts += 1; + if empty_parts == 32 { + tokio::task::yield_now().await; + self.validate().await?; + empty_parts = 0; + } + } + drop(self.input.take()); + self.pending = Bytes::new(); + Ok(()) + } +} + +fn integrity(message: &str) -> SnapshotError { + SnapshotError::new(SnapshotErrorCode::IntegrityError, message) +} diff --git a/src/api/router/snapshot_raw_blob_tests.rs b/src/api/router/snapshot_raw_blob_tests.rs new file mode 100644 index 00000000..1c4e70f9 --- /dev/null +++ b/src/api/router/snapshot_raw_blob_tests.rs @@ -0,0 +1,791 @@ +use axum::{ + extract::{Path as AxumPath, Query, State}, + http::HeaderMap, + routing::get, +}; +use tokio::{sync::Notify, time::timeout}; + +use super::*; +use crate::{ + api::router::snapshot_router::{raw_blob as serving, snapshot_auth_middleware}, + ceres::snapshot::content_budget::{MemoryBudget, RANGE_WORK_BYTES}, +}; + +fn budgeted_app( + fixture: &Fixture, + response_budget: &Arc, + scratch_budget: &Arc, +) -> Router { + let response_budget = response_budget.clone(); + let scratch_budget = scratch_budget.clone(); + Router::new() + .route( + "/api/v2/snapshots/{snapshot_id}/blob", + get( + move |state: State, + path: AxumPath, + query: Query, + headers: HeaderMap| { + serving::blob_with_budgets( + state, + path, + query, + headers, + serving::BlobBudgets { + response: response_budget.clone(), + scratch: scratch_budget.clone(), + }, + ) + }, + ), + ) + .route_layer(axum::middleware::from_fn_with_state( + fixture.state.clone(), + snapshot_auth_middleware, + )) + .with_state(fixture.state.clone()) +} + +fn fault(fixture: &Fixture, kind: bounded_objects::FaultKind) { + *fixture.counts.object_fault.lock().unwrap() = Some(bounded_objects::StreamFault { + oid: fixture.oid.clone(), + kind, + }); +} + +async fn path_oid(fixture: &Fixture, path: &str) -> String { + use crate::ceres::snapshot::pages::{MetadataWalkOutcome, resolve_abs_metadata}; + let handler = fixture + .state + .api_handler(std::path::Path::new("/")) + .await + .unwrap(); + let context = fixture + .state + .storage + .snapshot_context(&fixture.snapshot, &fixture.lease) + .await + .unwrap(); + let root = handler + .get_tree_by_hash(&context.root_tree_oid) + .await + .unwrap(); + match resolve_abs_metadata(handler.as_ref(), &root, &format!("/project{path}")) + .await + .unwrap() + { + MetadataWalkOutcome::FoundFile { oid, .. } => oid, + other => panic!("raw fixture path must be a fixed file: {other:?}"), + } +} + +async fn release_lease(fixture: &Fixture) { + let response = 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!(response.status(), 200); +} + +#[tokio::test] +async fn cold_raw_get_earns_source_proof_then_serves_one_current_stream_with_exact_headers() { + let fixture = Fixture::new().await; + let response = fixture.send("GET", "blob?path=/file", Body::empty()).await; + assert_eq!(response.status(), 200); + assert_eq!( + response.headers()["content-length"], + fixture.raw.len().to_string() + ); + assert_eq!( + response.headers()["x-mega-content-size"], + fixture.raw.len().to_string() + ); + assert_eq!( + response.headers()["etag"], + format!("\"{}\"", fixture.digest_string()) + ); + assert_eq!(response.headers()["x-mega-fs-kind"], "regular"); + assert_eq!(response.headers()["vary"], "Authorization, Accept"); + assert_eq!( + response.headers()["cache-control"], + "private, no-cache, no-transform" + ); + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 2); + assert_eq!( + fixture.counts.bytes.load(Ordering::SeqCst), + fixture.raw.len() + ); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 1); + assert_eq!( + to_bytes(response.into_body(), 2 * CHUNK_SIZE as usize) + .await + .unwrap() + .as_ref(), + fixture.raw + ); + fixture.counts.assert(2, 2 * fixture.raw.len()); + fixture.counts.reset(); + let response = fixture.send("GET", "blob?path=/alias", Body::empty()).await; + assert_eq!(response.status(), 200); + assert_eq!(fixture.counts.bytes.load(Ordering::SeqCst), 0); + assert_eq!( + to_bytes(response.into_body(), 2 * CHUNK_SIZE as usize) + .await + .unwrap() + .as_ref(), + fixture.raw + ); + fixture.counts.assert(1, fixture.raw.len()); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.receipt_reads.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn raw_memory_pressure_rejects_before_cold_proof_or_delivery_source_io() { + let fixture = Fixture::new().await; + for occupy_scratch in [false, true] { + fixture.counts.reset(); + let response_budget = MemoryBudget::new(CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let held = if occupy_scratch { + scratch_budget.reserve(RANGE_WORK_BYTES).unwrap() + } else { + response_budget.reserve(CHUNK_SIZE as usize).unwrap() + }; + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + error(response, 503, "TEMPORARY_UNAVAILABLE", true).await; + fixture.counts.assert(0, 0); + assert_eq!(fixture.counts.receipt_reads.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + assert_eq!( + scratch_budget.used(), + if occupy_scratch { RANGE_WORK_BYTES } else { 0 } + ); + assert_eq!( + response_budget.used(), + if occupy_scratch { + 0 + } else { + CHUNK_SIZE as usize + } + ); + drop(held); + assert_eq!(response_budget.used(), 0); + assert_eq!(scratch_budget.used(), 0); + } +} + +#[tokio::test] +async fn raw_missing_or_forged_receipt_and_real_stored_corruption_fail_without_rebuilding() { + let fixture = Fixture::new().await; + fixture.map("/file").await; + for missing in [true, false] { + fixture.counts.reset(); + fixture + .counts + .receipt_read_failure + .store(missing, Ordering::SeqCst); + *fixture.counts.receipt_read_corruption.lock().unwrap() = + (!missing).then(|| Bytes::from_static(b"forged receipt")); + error( + fixture.send("GET", "blob?path=/file", Body::empty()).await, + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + fixture.counts.assert(0, 0); + assert_eq!(fixture.counts.receipt_reads.load(Ordering::SeqCst), 1); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + } + fixture + .counts + .receipt_read_failure + .store(false, Ordering::SeqCst); + *fixture.counts.receipt_read_corruption.lock().unwrap() = None; + let mut wrong = fixture.raw.clone(); + wrong[0] ^= 1; + fixture.write_raw(wrong).await; + let response = fixture.send("GET", "blob?path=/file", Body::empty()).await; + assert_eq!(response.status(), 200); + let mut stream = response.into_body().into_data_stream(); + assert!(stream.next().await.unwrap().is_err()); + assert!(stream.next().await.is_none()); + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 1); + assert_eq!(fixture.counts.range.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + fixture.write_raw(fixture.raw.clone()).await; + let response = fixture.send("GET", "blob?path=/file", Body::empty()).await; + assert_eq!(response.status(), 200); + assert_eq!( + to_bytes(response.into_body(), 2 * CHUNK_SIZE as usize) + .await + .unwrap() + .as_ref(), + fixture.raw + ); + fixture.counts.assert(1, fixture.raw.len()); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn warm_fragmented_actual_raw_body_polls_one_chunk_and_transport_clones_keep_exact_credit() { + let fixture = Fixture::new().await; + fixture.map("/file").await; + fixture.counts.reset(); + fault( + &fixture, + bounded_objects::FaultKind::Parts(vec![ + Bytes::new(), + Bytes::copy_from_slice(&fixture.raw[..13]), + Bytes::copy_from_slice(&fixture.raw[13..CHUNK_SIZE as usize - 7]), + Bytes::copy_from_slice(&fixture.raw[CHUNK_SIZE as usize - 7..CHUNK_SIZE as usize]), + Bytes::copy_from_slice(&fixture.raw[CHUNK_SIZE as usize..]), + Bytes::new(), + ]), + ); + let response_budget = MemoryBudget::new(fixture.raw.len()); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + fixture.counts.assert(1, 0); + assert_eq!(response_budget.used(), CHUNK_SIZE as usize); + assert_eq!(scratch_budget.used(), RANGE_WORK_BYTES); + let mut stream = response.into_body().into_data_stream(); + let first = stream.next().await.unwrap().unwrap(); + assert_eq!(first.as_ref(), &fixture.raw[..CHUNK_SIZE as usize]); + fixture.counts.assert(1, CHUNK_SIZE as usize); + let transport = first.clone(); + drop(first); + assert_eq!(response_budget.used(), CHUNK_SIZE as usize); + let final_bytes = stream.next().await.unwrap().unwrap(); + assert_eq!(final_bytes.as_ref(), &fixture.raw[CHUNK_SIZE as usize..]); + assert_eq!(response_budget.used(), fixture.raw.len()); + assert!(stream.next().await.is_none()); + assert_eq!(scratch_budget.used(), 0); + assert_eq!(response_budget.used(), fixture.raw.len()); + drop(stream); + drop(final_bytes); + assert_eq!(response_budget.used(), CHUNK_SIZE as usize); + drop(transport); + assert_eq!(response_budget.used(), 0); + fixture.counts.assert(1, fixture.raw.len()); +} + +#[tokio::test] +async fn raw_transport_quota_rejects_next_chunk_before_more_source_io_and_errors_terminally() { + let fixture = Fixture::new().await; + fixture.map("/file").await; + fixture.counts.reset(); + fault( + &fixture, + bounded_objects::FaultKind::Parts(vec![ + Bytes::copy_from_slice(&fixture.raw[..CHUNK_SIZE as usize]), + Bytes::copy_from_slice(&fixture.raw[CHUNK_SIZE as usize..]), + ]), + ); + let response_budget = MemoryBudget::new(CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + let mut stream = response.into_body().into_data_stream(); + let transport = stream.next().await.unwrap().unwrap(); + assert_eq!(transport.as_ref(), &fixture.raw[..CHUNK_SIZE as usize]); + assert!(stream.next().await.unwrap().is_err()); + assert!(stream.next().await.is_none()); + fixture.counts.assert(1, CHUNK_SIZE as usize); + assert_eq!(scratch_budget.used(), 0); + assert_eq!(response_budget.used(), CHUNK_SIZE as usize); + drop(transport); + assert_eq!(response_budget.used(), 0); +} + +#[tokio::test] +async fn warm_raw_corruption_growth_truncation_and_late_error_never_yield_the_last_bytes() { + let fixture = Fixture::new().await; + fixture.map("/file").await; + let mut wrong = fixture.raw.clone(); + wrong[0] ^= 1; + let mut grown = fixture.raw.clone(); + grown.push(0); + let cases = [ + ( + bounded_objects::FaultKind::Parts(vec![Bytes::from(wrong)]), + 0, + ), + ( + bounded_objects::FaultKind::Parts(vec![Bytes::from(grown)]), + 0, + ), + ( + bounded_objects::FaultKind::Parts(vec![Bytes::copy_from_slice( + &fixture.raw[..fixture.raw.len() - 1], + )]), + CHUNK_SIZE as usize, + ), + ( + bounded_objects::FaultKind::LateError(Bytes::copy_from_slice(&fixture.raw)), + CHUNK_SIZE as usize, + ), + ]; + for (kind, delivered) in cases { + fixture.counts.reset(); + fault(&fixture, kind); + let response_budget = MemoryBudget::new(2 * CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + let mut stream = response.into_body().into_data_stream(); + let mut bytes = Vec::new(); + loop { + match stream.next().await { + Some(Ok(part)) => bytes.extend_from_slice(&part), + Some(Err(_)) => break, + None => panic!("corrupt current raw source must fail the actual response body"), + } + } + assert_eq!(bytes.len(), delivered); + assert_eq!(bytes, fixture.raw[..delivered]); + assert!(stream.next().await.is_none()); + assert_eq!(response_budget.used(), 0); + assert_eq!(scratch_budget.used(), 0); + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 1); + assert_eq!(fixture.counts.range.load(Ordering::SeqCst), 0); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + } + *fixture.counts.object_fault.lock().unwrap() = None; + let response = fixture.send("GET", "blob?path=/file", Body::empty()).await; + assert_eq!( + to_bytes(response.into_body(), 2 * CHUNK_SIZE as usize) + .await + .unwrap() + .as_ref(), + fixture.raw + ); +} + +#[tokio::test] +async fn raw_drop_cancel_and_revocation_during_held_io_drop_producer_and_all_owned_credit() { + for mode in 0..3 { + let fixture = Fixture::new().await; + fixture.map("/file").await; + fixture.counts.reset(); + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let drops = Arc::new(AtomicUsize::new(0)); + fault( + &fixture, + bounded_objects::FaultKind::Held { + raw: Bytes::copy_from_slice(&fixture.raw), + entered: entered.clone(), + release: release.clone(), + drops: drops.clone(), + }, + ); + let response_budget = MemoryBudget::new(2 * CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + fixture.counts.assert(1, 0); + if mode == 0 { + drop(response); + } else { + let mut stream = response.into_body().into_data_stream(); + let task = tokio::spawn(async move { + let first = stream.next().await; + (first, stream) + }); + timeout(Duration::from_secs(10), entered.notified()) + .await + .unwrap(); + fixture.counts.assert(1, 1); + if mode == 1 { + task.abort(); + assert!(task.await.err().unwrap().is_cancelled()); + } else { + release_lease(&fixture).await; + release.notify_one(); + let (first, mut stream) = task.await.unwrap(); + assert!(first.unwrap().is_err()); + assert!(stream.next().await.is_none()); + } + } + assert_eq!(drops.load(Ordering::SeqCst), 1); + assert_eq!(response_budget.used(), 0); + assert_eq!(scratch_budget.used(), 0); + } +} + +#[tokio::test] +async fn raw_first_poll_rechecks_released_lease_before_consuming_body_bytes() { + let fixture = Fixture::new().await; + fixture.map("/file").await; + fixture.counts.reset(); + let response_budget = MemoryBudget::new(2 * CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + release_lease(&fixture).await; + let mut stream = response.into_body().into_data_stream(); + assert!(stream.next().await.unwrap().is_err()); + assert!(stream.next().await.is_none()); + fixture.counts.assert(1, 0); + assert_eq!(response_budget.used(), 0); + assert_eq!(scratch_budget.used(), 0); +} + +#[tokio::test] +async fn raw_post_await_revocation_rejects_fragments_and_eof_before_another_backend_poll() { + for case in 0..4 { + let fixture = Fixture::new().await; + fixture.map("/file").await; + fixture.counts.reset(); + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let tail_polls = Arc::new(AtomicUsize::new(0)); + let drops = Arc::new(AtomicUsize::new(0)); + let prefix = if case < 2 { + Bytes::copy_from_slice(&fixture.raw[..1]) + } else { + Bytes::copy_from_slice(&fixture.raw) + }; + let prefix_length = prefix.len(); + let fragment = match case { + 0 => Some(Bytes::copy_from_slice(&fixture.raw[1..2])), + 1 | 2 => Some(Bytes::new()), + _ => None, + }; + fault( + &fixture, + bounded_objects::FaultKind::HeldFragment { + prefix, + fragment, + entered: entered.clone(), + release: release.clone(), + tail_polls: tail_polls.clone(), + drops: drops.clone(), + }, + ); + let response_budget = MemoryBudget::new(2 * CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + let mut stream = response.into_body().into_data_stream(); + let mut task = tokio::spawn(async move { + let mut delivered = Vec::new(); + loop { + match stream.next().await { + Some(Ok(bytes)) => delivered.extend_from_slice(&bytes), + Some(Err(error)) => return (delivered, error, stream), + None => panic!("revoked source must fail the actual body"), + } + } + }); + timeout(Duration::from_secs(10), entered.notified()) + .await + .unwrap(); + fixture.counts.assert(1, prefix_length); + release_lease(&fixture).await; + release.notify_one(); + let finished = timeout(Duration::from_secs(10), &mut task).await; + if finished.is_err() { + task.abort(); + let _ = task.await; + panic!("revoked fragmented source continued into another held backend poll"); + } + let (delivered, _, mut stream) = finished.unwrap().unwrap(); + assert_eq!( + delivered, + fixture.raw[..if case < 2 { 0 } else { CHUNK_SIZE as usize }] + ); + assert!(stream.next().await.is_none()); + assert_eq!(tail_polls.load(Ordering::SeqCst), 0); + assert_eq!(drops.load(Ordering::SeqCst), 1); + assert_eq!(response_budget.used(), 0); + assert_eq!(scratch_budget.used(), 0); + } +} + +#[tokio::test] +async fn empty_raw_post_await_revocation_fails_before_another_eof_poll_or_empty_success() { + for fragment in [Some(Bytes::new()), None] { + let fixture = Fixture::new().await; + let oid = path_oid(&fixture, "/empty").await; + fixture.counts.reset(); + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let tail_polls = Arc::new(AtomicUsize::new(0)); + let drops = Arc::new(AtomicUsize::new(0)); + *fixture.counts.object_fault.lock().unwrap() = Some(bounded_objects::StreamFault { + oid, + kind: bounded_objects::FaultKind::HeldFragment { + prefix: Bytes::new(), + fragment, + entered: entered.clone(), + release: release.clone(), + tail_polls: tail_polls.clone(), + drops: drops.clone(), + }, + }); + let response_budget = MemoryBudget::new(CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let app = budgeted_app(&fixture, &response_budget, &scratch_budget); + let request = fixture.request("GET", "blob?path=/empty", Body::empty()); + let mut task = tokio::spawn(async move { app.oneshot(request).await.unwrap() }); + timeout(Duration::from_secs(10), entered.notified()) + .await + .unwrap(); + release_lease(&fixture).await; + release.notify_one(); + let finished = timeout(Duration::from_secs(10), &mut task).await; + if finished.is_err() { + task.abort(); + let _ = task.await; + panic!("revoked empty source continued into another held backend poll"); + } + error(finished.unwrap().unwrap(), 410, "LEASE_EXPIRED", false).await; + assert_eq!(tail_polls.load(Ordering::SeqCst), 0); + assert_eq!(drops.load(Ordering::SeqCst), 1); + assert_eq!(response_budget.used(), 0); + assert_eq!(scratch_budget.used(), 0); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + } +} + +#[tokio::test] +async fn raw_physical_size_and_current_fact_fail_before_body_and_path_errors_keep_formal_statuses() +{ + let fixture = Fixture::new().await; + fixture.map("/file").await; + fixture.counts.reset(); + for (path, status, code) in [ + ("/absent", 404, "PATH_NOT_FOUND"), + ("/outside", 404, "PATH_NOT_FOUND"), + ("/directory", 409, "NOT_DIRECTORY"), + ("/file/child", 409, "NOT_DIRECTORY"), + ("/link/child", 409, "SYMLINK_TRAVERSAL"), + ("/../outside", 400, "SCOPE_INVALID"), + ] { + error( + fixture + .send("GET", &format!("blob?path={path}"), Body::empty()) + .await, + status, + code, + false, + ) + .await; + fixture.counts.assert(0, 0); + } + let mut request = fixture.request("GET", "blob?path=/file", Body::empty()); + request + .headers_mut() + .insert("range", "bytes=0-9".parse().unwrap()); + error( + fixture.app.clone().oneshot(request).await.unwrap(), + 400, + "RANGE_NOT_SUPPORTED", + false, + ) + .await; + error( + fixture + .send( + "GET", + &format!("blob?path=/file&expected_digest=sha256:{}", "0".repeat(64)), + Body::empty(), + ) + .await, + 409, + "EXPECTED_DIGEST_MISMATCH", + false, + ) + .await; + fixture.counts.assert(0, 0); + *fixture.counts.object_size_override.lock().unwrap() = Some(fixture.raw.len() as i64 + 1); + error( + fixture.send("GET", "blob?path=/file", Body::empty()).await, + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + fixture.counts.assert(1, 0); + *fixture.counts.object_size_override.lock().unwrap() = None; + let original = fixture.fact().await; + let mut changed = original.clone(); + changed.created_at += chrono::Duration::seconds(1); + fixture.replace_fact(changed).await; + fixture.counts.reset(); + error( + fixture.send("GET", "blob?path=/file", Body::empty()).await, + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + fixture.counts.assert(0, 0); + fixture.replace_fact(original).await; + for (path, kind, bytes) in [ + ("/empty", "regular", &b""[..]), + ("/link", "symlink", &b"file"[..]), + ("/executable", "executable", fixture.raw.as_slice()), + ] { + let response = fixture + .send("GET", &format!("blob?path={path}"), Body::empty()) + .await; + assert_eq!(response.status(), 200); + assert_eq!(response.headers()["x-mega-fs-kind"], kind); + assert_eq!( + response.headers()["content-length"], + bytes.len().to_string() + ); + assert_eq!( + to_bytes(response.into_body(), 2 * CHUNK_SIZE as usize) + .await + .unwrap() + .as_ref(), + bytes + ); + } +} + +#[tokio::test] +async fn raw_visible_producer_item_cap_rejects_before_copy_hash_or_tail_poll() { + let raw = vec![7; 8 * 1024 * 1024 + 1]; + let fixture = Fixture::new_with_pg_config_directories_and_objects( + false, + 0, + &[("large".into(), raw.clone())], + ) + .await; + let oid = path_oid(&fixture, "/large").await; + *fixture.counts.object_fault.lock().unwrap() = Some(bounded_objects::StreamFault { + oid: oid.clone(), + kind: bounded_objects::FaultKind::Parts( + raw.chunks(CHUNK_SIZE as usize) + .map(Bytes::copy_from_slice) + .collect(), + ), + }); + fixture.map("/large").await; + let tails = Arc::new(AtomicUsize::new(0)); + *fixture.counts.object_fault.lock().unwrap() = Some(bounded_objects::StreamFault { + oid, + kind: bounded_objects::FaultKind::Oversized(Bytes::from(raw), tails.clone()), + }); + fixture.counts.reset(); + let response_budget = MemoryBudget::new(2 * CHUNK_SIZE as usize); + let scratch_budget = MemoryBudget::new(RANGE_WORK_BYTES); + let response = budgeted_app(&fixture, &response_budget, &scratch_budget) + .oneshot(fixture.request("GET", "blob?path=/large", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + let mut stream = response.into_body().into_data_stream(); + assert!(stream.next().await.unwrap().is_err()); + assert!(stream.next().await.is_none()); + assert_eq!(tails.load(Ordering::SeqCst), 0); + assert_eq!(response_budget.used(), 0); + assert_eq!(scratch_budget.used(), 0); + assert_eq!(fixture.counts.whole.load(Ordering::SeqCst), 1); + assert_eq!(fixture.counts.range.load(Ordering::SeqCst), 0); + assert_eq!( + fixture.counts.bytes.load(Ordering::SeqCst), + 8 * 1024 * 1024 + 1 + ); +} + +#[tokio::test] +async fn empty_raw_source_requires_physical_zero_exact_eof_and_current_empty_digest() { + let fixture = Fixture::new().await; + let oid = path_oid(&fixture, "/empty").await; + for (kind, status, code) in [ + ( + bounded_objects::FaultKind::Parts(vec![Bytes::from_static(b"growth")]), + 502, + "INTEGRITY_ERROR", + ), + ( + bounded_objects::FaultKind::LateError(Bytes::new()), + 503, + "OBJECT_UNAVAILABLE", + ), + ] { + *fixture.counts.object_fault.lock().unwrap() = Some(bounded_objects::StreamFault { + oid: oid.clone(), + kind, + }); + fixture.counts.reset(); + error( + fixture.send("GET", "blob?path=/empty", Body::empty()).await, + status, + code, + false, + ) + .await; + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); + } + *fixture.counts.object_fault.lock().unwrap() = None; + fixture.counts.reset(); + let mono = fixture.state.storage.mono_storage(); + let original = mono + .get_verified_blobs(vec![oid.clone()]) + .await + .unwrap() + .remove(&oid) + .unwrap(); + let mut changed = original.clone().into_active_model(); + changed.raw_sha256 = Set(vec![9; 32]); + changed.update(mono.get_connection()).await.unwrap(); + error( + fixture.send("GET", "blob?path=/empty", Body::empty()).await, + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + fixture.counts.assert(0, 0); + original + .into_active_model() + .reset_all() + .update(mono.get_connection()) + .await + .unwrap(); + let response = fixture.send("GET", "blob?path=/empty", Body::empty()).await; + assert_eq!(response.status(), 200); + assert_eq!(response.headers()["content-length"], "0"); + assert!(to_bytes(response.into_body(), 1).await.unwrap().is_empty()); + fixture.counts.assert(1, 0); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 0); +} diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index f75b7279..a6838aea 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -28,8 +28,8 @@ use crate::{ descriptor::build as build_descriptor, error::{SnapshotError, SnapshotErrorCode}, pages::{ - MetadataWalkOutcome, WalkOutcome, base64_of, build_directory_page, - build_directory_page_with_work, hex_of, proof_pages, resolve_abs, resolve_abs_metadata, + MetadataWalkOutcome, base64_of, build_directory_page, build_directory_page_with_work, + hex_of, proof_pages, resolve_abs_metadata, }, projection_observation::{NativeResolveSource, ResolvedProjection}, runtime::{now_unix, runtime}, @@ -47,7 +47,7 @@ pub fn routers(api_state: MonoApiServiceState) -> Router { .route("/snapshots/{snapshot_id}/directory", get(directory)) .route( "/snapshots/{snapshot_id}/blob", - get(blob).head(content::blob_head), + get(raw_blob::blob).head(content::blob_head), ) .route("/snapshots/{snapshot_id}/lookup", post(lookup)) .route( @@ -79,6 +79,9 @@ pub fn routers(api_state: MonoApiServiceState) -> Router { #[path = "snapshot_content.rs"] mod content; +#[path = "snapshot_raw_blob.rs"] +mod raw_blob; + #[path = "snapshot_request.rs"] mod request; @@ -1165,127 +1168,6 @@ async fn directory( Ok(resp) } -#[derive(Deserialize, Debug)] -struct BlobQuery { - path: String, - #[serde(default)] - expected_digest: Option, -} - -#[allow(clippy::result_large_err, clippy::too_many_lines)] -async fn blob( - state: State, - AxumPath(snapshot_id): AxumPath, - Query(q): Query, - headers: HeaderMap, -) -> Result { - ensure_enabled(&state).map_err(mst2_error_response)?; - let ctx = request_context(&state, &snapshot_id).map_err(mst2_error_response)?; - validate_scope_relative_path(&q.path).map_err(mst2_error_response)?; - if headers.contains_key("range") { - // Spec 04 section 9: raw blob has no Range semantics this profile. - return Err(mst2_error_response(SnapshotError::new( - SnapshotErrorCode::RangeNotSupported, - "raw blob reads are whole-file; use chunks for ranges", - ))); - } - - let handler = state - .api_handler(std::path::Path::new("/")) - .await - .map_err(internal)?; - let root_tree = handler - .get_tree_by_hash(&ctx.root_tree_oid) - .await - .map_err(internal)?; - let abs_path = abs_view_path(&ctx.built.descriptor.scope, &q.path); - - match resolve_abs(handler.as_ref(), &root_tree, &abs_path) - .await - .map_err(mst2_error_response)? - { - WalkOutcome::FoundFile { - fs_kind, - raw, - digest, - .. - } => { - if let Some(expected) = &q.expected_digest - && expected != &format!("sha256:{}", hex_of(&digest)) - { - return Err(mst2_error_response(SnapshotError::new( - SnapshotErrorCode::DigestMismatch, - "content does not match expected_digest", - ))); - } - let fs_kind_str = match fs_kind { - crate::ceres::snapshot::resolver::FsKind::Regular => "regular", - crate::ceres::snapshot::resolver::FsKind::Executable => "executable", - crate::ceres::snapshot::resolver::FsKind::Symlink => "symlink", - crate::ceres::snapshot::resolver::FsKind::Directory => "directory", - }; - revalidate_request(&state, &ctx) - .await - .map_err(mst2_error_response)?; - let headers = REQUEST_HEADERS.try_with(Clone::clone).map_err(|_| { - mst2_error_response(SnapshotError::new( - SnapshotErrorCode::Unauthenticated, - "request authentication context missing", - )) - })?; - let stream_state = state.0.clone(); - let stream_context = ctx.clone(); - let blocks = futures::stream::unfold( - ( - Bytes::from(raw), - 0usize, - stream_state, - stream_context, - headers, - ), - |(raw, offset, state, context, headers)| async move { - if offset >= raw.len() { - return None; - } - if let Err(error) = revalidate_access(&state, &context, &headers).await { - return Some((Err(error), (raw, usize::MAX, state, context, headers))); - } - let end = (offset + 1_048_576).min(raw.len()); - let bytes = raw.slice(offset..end); - Some((Ok(bytes), (raw, end, state, context, headers))) - }, - ); - Response::builder() - .header("etag", format!("\"sha256:{}\"", hex_of(&digest))) - .header("cache-control", "private, no-cache, no-transform") - .header("x-mega-fs-kind", fs_kind_str) - .body(axum::body::Body::from_stream(blocks)) - .map_err(|e| { - mst2_error_response(SnapshotError::new( - SnapshotErrorCode::Internal, - format!("body build failed: {e}"), - )) - }) - } - WalkOutcome::FoundDir => Err(mst2_error_response(SnapshotError::new( - SnapshotErrorCode::NotDirectory, - "path is a directory", - ))), - WalkOutcome::Absent => Err(mst2_error_response(SnapshotError::new( - SnapshotErrorCode::PathNotFound, - "path absent in the fixed view", - ))), - WalkOutcome::NotDirectory { symlink } => Err(mst2_error_response(SnapshotError::new( - if symlink { - SnapshotErrorCode::SymlinkTraversal - } else { - SnapshotErrorCode::NotDirectory - }, - "intermediate component is not a directory", - ))), - } -} - /// Scope-relative request path -> absolute view path (spec 04 section 1). fn abs_view_path(scope: &str, path: &str) -> String { if path == "/" { diff --git a/src/api/router/snapshot_session_tests.rs b/src/api/router/snapshot_session_tests.rs index 822890d0..ae84d3c6 100644 --- a/src/api/router/snapshot_session_tests.rs +++ b/src/api/router/snapshot_session_tests.rs @@ -300,7 +300,8 @@ async fn mst2_durable_http_resolve_installs_complete_dag_and_warm_leases_share_i .as_ref(), fixture.raw.as_slice() ); - fixture.counts.assert(1, fixture.raw.len()); + fixture.counts.assert(2, 2 * fixture.raw.len()); + assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 1); } #[tokio::test] diff --git a/src/ceres/api_service/mod.rs b/src/ceres/api_service/mod.rs index 29ec5d74..7acbd06a 100644 --- a/src/ceres/api_service/mod.rs +++ b/src/ceres/api_service/mod.rs @@ -191,6 +191,26 @@ pub trait ApiHandler: Send + Sync { } } + /// Raw whole-file stream with the physical object's complete size. + async fn get_raw_blob_stream_with_meta( + &self, + hash: &str, + ) -> Result< + ( + crate::orbit_api::object_storage::ObjectByteStream, + crate::orbit_api::object_storage::ObjectMeta, + ), + MegaError, + > { + let storage = self.get_context(); + match storage.git_service.get_object_stream_with_meta(hash).await { + Ok(value) => Ok(value), + Err(error) => Err(storage + .classify_blob_objstorage_not_found(hash, error) + .await), + } + } + async fn get_raw_blob_range_stream_exact( &self, hash: &str, diff --git a/src/ceres/snapshot/content_budget.rs b/src/ceres/snapshot/content_budget.rs index 79db0289..6bb8d32b 100644 --- a/src/ceres/snapshot/content_budget.rs +++ b/src/ceres/snapshot/content_budget.rs @@ -81,17 +81,21 @@ pub(crate) fn projection_budget() -> &'static Arc { } pub(crate) fn reserve_response(bytes: usize) -> Result { + response_budget().reserve(bytes) +} + +pub(crate) fn response_budget() -> &'static Arc { static BUDGET: OnceLock> = OnceLock::new(); - BUDGET - .get_or_init(|| MemoryBudget::new(RESPONSE_LIVE_BYTES)) - .reserve(bytes) + BUDGET.get_or_init(|| MemoryBudget::new(RESPONSE_LIVE_BYTES)) } pub(crate) fn reserve_range_work() -> Result { + range_budget().reserve(RANGE_WORK_BYTES) +} + +pub(crate) fn range_budget() -> &'static Arc { static BUDGET: OnceLock> = OnceLock::new(); - BUDGET - .get_or_init(|| MemoryBudget::new(RANGE_SCRATCH_BYTES)) - .reserve(RANGE_WORK_BYTES) + BUDGET.get_or_init(|| MemoryBudget::new(RANGE_SCRATCH_BYTES)) } /// Every encoded frame owns shared credit, including after the body passes a diff --git a/src/jupiter/service/git_service.rs b/src/jupiter/service/git_service.rs index 6569e9a0..8652cc97 100644 --- a/src/jupiter/service/git_service.rs +++ b/src/jupiter/service/git_service.rs @@ -108,6 +108,13 @@ impl GitService { } pub async fn get_object_stream(&self, hash: &str) -> Result { + Ok(self.get_object_stream_with_meta(hash).await?.0) + } + + pub async fn get_object_stream_with_meta( + &self, + hash: &str, + ) -> Result<(ObjectByteStream, ObjectMeta), MegaError> { if !is_full_hex_object_id(hash) { return Err(MegaError::Other("Invalid object ID format".to_string())); } @@ -115,8 +122,7 @@ impl GitService { namespace: ObjectNamespace::Git, key: hash.to_string(), }; - let (stream, _) = self.obj_storage.inner.get_stream(&key).await?; - Ok(stream) + Ok(self.obj_storage.inner.get_stream(&key).await?) } pub async fn get_object_range_stream_exact(