diff --git a/src/snapshot/cas_content.rs b/src/snapshot/cas_content.rs index b2f4e23..add9191 100644 --- a/src/snapshot/cas_content.rs +++ b/src/snapshot/cas_content.rs @@ -12,9 +12,9 @@ use std::{ use super::{ content::{AccountedBuffer, BudgetClass, ContentBudget}, + content_profile::OBJECT_CAP, durable::DurableStore, frames::parse_digest, - range::OBJECT_CAP, secure_fs, SnapshotError, SnapshotErrorCode, }; diff --git a/src/snapshot/cas_range.rs b/src/snapshot/cas_range.rs index 337a43a..c791a64 100644 --- a/src/snapshot/cas_range.rs +++ b/src/snapshot/cas_range.rs @@ -7,9 +7,9 @@ use std::{fmt, mem::size_of, sync::Arc}; use super::{ cas_index::{self, LocalCasRangeMeters, CHUNK_SIZE}, content::{AccountedBuffer, BudgetClass, ContentBudget, Reservation}, + content_profile::{MAX_FILE_SIZE, OBJECT_CAP}, durable::DurableStore, frames::parse_digest, - range::{MAX_FILE_SIZE, OBJECT_CAP}, SnapshotError, SnapshotErrorCode, }; diff --git a/src/snapshot/client.rs b/src/snapshot/client.rs index 879fc2c..bb767f8 100644 --- a/src/snapshot/client.rs +++ b/src/snapshot/client.rs @@ -221,7 +221,7 @@ impl Mst2Client { let limit = self .profile .as_ref() - .map_or(super::range::MAX_FILE_SIZE, |caps| { + .map_or(super::content_profile::MAX_FILE_SIZE, |caps| { caps.limits().max_file_bytes }); if size > limit { diff --git a/src/snapshot/content_profile.rs b/src/snapshot/content_profile.rs new file mode 100644 index 0000000..067470f --- /dev/null +++ b/src/snapshot/content_profile.rs @@ -0,0 +1,15 @@ +//! Fixed MST/2 serving profile, shared by owned HTTP and local CAS readers. + +/// Files at or below this size use OBJECT frames (spec 07 §2). +pub const OBJECT_CAP: u64 = 256 * 1024; +pub(crate) const MAX_FILE_SIZE: u64 = 8 * 1024 * 1024 * 1024 * 1024; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn object_cap_matches_the_profile_split() { + assert_eq!(OBJECT_CAP, 262_144); + } +} diff --git a/src/snapshot/directory.rs b/src/snapshot/directory.rs index 7e875bb..472937e 100644 --- a/src/snapshot/directory.rs +++ b/src/snapshot/directory.rs @@ -2,8 +2,8 @@ //! establish MTP2 membership or prove that a claimed EOF is cryptographic EOF. use super::{ + content_profile::MAX_FILE_SIZE, frames::parse_count, - range::MAX_FILE_SIZE, types::{DirectoryResponse, SnapshotError, SnapshotErrorCode}, }; diff --git a/src/snapshot/durable.rs b/src/snapshot/durable.rs index df70bb8..0328756 100644 --- a/src/snapshot/durable.rs +++ b/src/snapshot/durable.rs @@ -45,6 +45,7 @@ use tokio::io::AsyncWriteExt; pub use super::cas_index::LocalCasRangeMeters; use crate::snapshot::{ closure::{SnapshotDirectory, ValidatedSnapshotClosure}, + online_file::CompletedOnlineManifest, secure_fs, OfflineGrant, SnapshotError, SnapshotErrorCode, SnapshotFile, SnapshotReader, }; @@ -199,6 +200,39 @@ struct SnapshotHydration<'a> { reader: Option<&'a SnapshotReader>, } +/// Distinguish authenticated full-root facts from the completed online walk. +/// Neither a callback manifest nor an arbitrary tuple can select this source. +#[derive(Clone, Copy)] +enum HydrationRangeSource<'a> { + FixedRoot(&'a SnapshotReader), + OnlineManifest(&'a SnapshotReader, &'a CompletedOnlineManifest), +} + +impl<'a> HydrationRangeSource<'a> { + fn reader(self) -> &'a SnapshotReader { + match self { + Self::FixedRoot(reader) | Self::OnlineManifest(reader, _) => reader, + } + } + + async fn open(self, file: &SnapshotFile) -> Result { + match self { + Self::FixedRoot(reader) => { + super::OwnedChunkedFile::open( + reader, + &file.rel_path, + &file.content_digest, + file.size, + ) + .await + } + Self::OnlineManifest(reader, manifest) => { + super::OwnedChunkedFile::open_online_path(reader, manifest.range_file(file)?).await + } + } + } +} + /// Resume hints may lag the durable CAS by one bounded chunk. Losing that /// chunk on cancellation/crash only requires re-hashing the CAS on resume. /// The per-view OS lock protects the transaction; this mutex serializes its @@ -923,30 +957,43 @@ impl DurableStore { scope: reader.descriptor().scope.clone(), lease_id: reader.lease_id().to_string(), }; - let manifest = if reader.capabilities().features.metadata_pages { + let online = if reader.capabilities().features.metadata_pages { + None + } else { + Some(CompletedOnlineManifest::walk(reader).await?) + }; + let proved_files = if online.is_none() { let closure = reader.snapshot_closure().await?; reader.seed_content_membership(&closure)?; closure.files().to_vec() } else { - reader.file_manifest().await? + Vec::new() }; - let legacy = super::FetchCoordinator::in_reader_content_scope(reader.clone(), 1); + let manifest = online + .as_ref() + .map_or(proved_files.as_slice(), CompletedOnlineManifest::files); + let source = online + .as_ref() + .map_or(HydrationRangeSource::FixedRoot(reader), |manifest| { + HydrationRangeSource::OnlineManifest(reader, manifest) + }); + let online_fetch = super::FetchCoordinator::in_reader_content_scope(reader.clone(), 1); self.hydrate_file_closure( &view, - &manifest, + manifest, None, |f| { let file = f.clone(); - let legacy = legacy.clone(); + let online_fetch = online_fetch.clone(); async move { if reader.capabilities().features.metadata_pages { reader.read_content(&file, false).await } else { - legacy.fetch_owned(file, false).await + online_fetch.fetch_owned(file, false).await } } }, - Some(reader), + Some(source), ) .await } @@ -982,7 +1029,7 @@ impl DurableStore { let file = file.clone(); async move { reader.read_content(&file, use_frames).await } }, - Some(reader), + Some(HydrationRangeSource::FixedRoot(reader)), ) .await } @@ -1060,7 +1107,7 @@ impl DurableStore { manifest: &[SnapshotFile], closure: Option<&ValidatedSnapshotClosure>, fetch: F, - stream_reader: Option<&SnapshotReader>, + stream_reader: Option>, ) -> Result where F: Fn(&SnapshotFile) -> Fut, @@ -1111,10 +1158,11 @@ impl DurableStore { Err(e) => return Err(e), } - if let Some(reader) = stream_reader.filter(|reader| { - f.size > crate::snapshot::OBJECT_CAP && reader.capabilities().features.chunk_reads + if let Some(source) = stream_reader.filter(|source| { + f.size > crate::snapshot::OBJECT_CAP + && source.reader().capabilities().features.chunk_reads }) { - write_reader_blob(&self.content, reader, f) + write_reader_blob(&self.content, source, f) .await .map_err(|error| { tag_hydration_error(error, HydrationSubstage::LargeContentFetch) @@ -1173,7 +1221,7 @@ impl DurableStore { view, manifest, Some(closure), - stream_reader.and_then(|reader| reader.offline_grant()), + stream_reader.and_then(|source| source.reader().offline_grant()), (fetched, resumed, repaired), ), None => self.finish_hydration(view, manifest, bytes_total, fetched, resumed, repaired), @@ -1348,7 +1396,12 @@ impl DurableStore { f.size > crate::snapshot::OBJECT_CAP && reader.capabilities().features.chunk_reads }) { - write_reader_blob(&store.content, reader, &f).await?; + write_reader_blob( + &store.content, + HydrationRangeSource::FixedRoot(reader), + &f, + ) + .await?; } else { let owner: std::sync::Arc = fetch(f.clone()).await?; let bytes = owner.as_ref().as_ref(); @@ -1761,7 +1814,12 @@ impl DurableStore { if let Some(reader) = stream_reader .filter(|reader| reader.capabilities().features.chunk_reads) { - write_reader_blob(&store.content, reader, &f).await?; + write_reader_blob( + &store.content, + HydrationRangeSource::FixedRoot(reader), + &f, + ) + .await?; } else { let owner: std::sync::Arc = fetch_large(f.clone()).await?; let bytes = owner.as_ref().as_ref(); @@ -2615,7 +2673,7 @@ impl DurableStore { } fn check_range_profile(expected_size: u64) -> Result<(), SnapshotError> { - if expected_size > crate::snapshot::range::MAX_FILE_SIZE { + if expected_size > crate::snapshot::content_profile::MAX_FILE_SIZE { return Err(SnapshotError::new( SnapshotErrorCode::LimitExceeded, "file size exceeds the 8 TiB serving profile", @@ -2920,48 +2978,27 @@ pub(super) fn write_atomic(dir: &Path, name: &str, data: &[u8]) -> Result<(), Sn /// process may leave an unpublished temp, as with the buffered atomic writer. async fn write_reader_blob( dir: &Path, - reader: &SnapshotReader, + source: HydrationRangeSource<'_>, file: &SnapshotFile, ) -> Result<(), SnapshotError> { - // Keep canonical v3 hydration on the owned range implementation. The - // previous compatibility `ChunkedFile` reader returned a fresh Vec for - // every chunk, bypassing the reader's output budget and retaining no proof - // ownership across the write. Legacy advertisements may expose chunk - // reads without metadata/pages, so retain their established reader until - // that protocol profile is retired. - if reader.capabilities().features.metadata_pages { - let source = crate::snapshot::OwnedChunkedFile::open( - reader, - &file.rel_path, - &file.content_digest, - file.size, - ) + let source = source + .open(file) .await .map_err(|error| tag_hydration_error(error, HydrationSubstage::LargeChunkMap))?; - return write_reader_blob_stream(dir, file, |offset, length| { - source.read_range_owned(offset, length) - }) - .await; - } - let source = - crate::snapshot::ChunkedFile::open(reader, &file.rel_path, &file.content_digest, file.size) - .await - .map_err(|error| tag_hydration_error(error, HydrationSubstage::LargeChunkMap))?; write_reader_blob_stream(dir, file, |offset, length| { - source.read_range(offset, length) + source.read_range_owned(offset, length) }) .await } -async fn write_reader_blob_stream( +async fn write_reader_blob_stream( dir: &Path, file: &SnapshotFile, mut read_range: F, ) -> Result<(), SnapshotError> where F: FnMut(u64, u64) -> Fut, - Fut: Future>, - B: BlobBytes, + Fut: Future, SnapshotError>>, { create_dirs_durable(dir) .map_err(|error| tag_hydration_error(error, HydrationSubstage::LargeCasWrite))?; @@ -2983,13 +3020,13 @@ where let bytes = read_range(offset, length) .await .map_err(|error| tag_hydration_error(error, HydrationSubstage::LargeChunkRead))?; - if bytes.bytes().len() as u64 != length { + if bytes.len() as u64 != length { return Err(tag_hydration_error( integrity_err("streamed chunk does not cover the expected file range"), HydrationSubstage::LargeChunkRead, )); } - let bytes = bytes.bytes(); + let bytes = bytes.as_bytes(); hash.update(bytes); output .write_all(bytes) @@ -3029,26 +3066,6 @@ where sync_dir(dir).map_err(|error| tag_hydration_error(error, HydrationSubstage::LargeCasWrite)) } -/// Borrow the streamed body while keeping an owned range reservation alive. -/// `Arc` cannot implement `AsRef<[u8]>` through the standard -/// library's `AsRef` implementation, so the private adapter keeps the -/// generic writer independent of the two source reader representations. -trait BlobBytes { - fn bytes(&self) -> &[u8]; -} - -impl BlobBytes for Vec { - fn bytes(&self) -> &[u8] { - self.as_slice() - } -} - -impl BlobBytes for Arc { - fn bytes(&self) -> &[u8] { - self.as_bytes() - } -} - struct PendingBlob(PathBuf); impl Drop for PendingBlob { diff --git a/src/snapshot/durable_online_tests.rs b/src/snapshot/durable_online_tests.rs new file mode 100644 index 0000000..0c8360c --- /dev/null +++ b/src/snapshot/durable_online_tests.rs @@ -0,0 +1,950 @@ +//! Public no-pages hydration uses a completed fixed-SID online walk, paid +//! ranges and whole-content verification, without claiming MTP2 membership. + +use std::sync::{atomic::AtomicBool, Mutex}; + +use axum::{ + body::Body, extract::Request, http::StatusCode, middleware::Next, response::IntoResponse, +}; +use futures::StreamExt; + +use super::*; + +const MIB: usize = 1024 * 1024; +const ONLINE_LEASE: &str = "completed-online-manifest-lease"; +const DIRECTORY_OK: usize = 0; +const MISSING_SECOND_PAGE: usize = 1; +const MISSING_CHILD: usize = 2; +const LOOP_CURSOR: usize = 3; +const WRONG_ROOT: usize = 4; +const CHUNK_OK: usize = 0; +const SHORT_CHUNK: usize = 1; +const WRONG_CHUNK_HASH: usize = 2; +const WRONG_CHUNK_MAP: usize = 3; +const WRONG_CHUNK_CONTENT: usize = 4; +const WRONG_END: usize = 5; +const BEFORE_HEADERS: usize = 1; +const AFTER_CHUNK_BODY: usize = 2; + +struct OnlineFixture { + inner: Arc, + descriptor: ServingDescriptor, + empty_root: [u8; 32], + directory_fault: AtomicUsize, + directory_calls: Mutex)>>, + hold_directory: AtomicBool, + directory_entered: Notify, + directory_release: Notify, + wrong_map_size: AtomicBool, + hold_map: AtomicBool, + map_entered: Notify, + map_release: Notify, + chunk_fault: AtomicUsize, + hold_chunk: AtomicU64, + chunk_stage: AtomicUsize, + chunk_entered: Notify, + chunk_release: Notify, + partial_bodies: AtomicUsize, + short_lease: bool, + renewal_status: StatusCode, + renewals: AtomicUsize, +} + +impl OnlineFixture { + fn new(size: u64, wrong_whole: bool) -> Self { + let inner = Arc::new(Fixture::new(size, wrong_whole)); + let empty_root = page_id(&Page::build(&[]).unwrap()); + let root = Page::build(&[ + Entry::dir(b"a", empty_root), + Entry::file(EntryKind::Regular, b"large.bin", size, inner.advertised), + ]) + .unwrap(); + Self { + inner, + descriptor: ServingDescriptor { + instance_uuid: *uuid::Uuid::parse_str(INSTANCE).unwrap().as_bytes(), + namespace_view_id: [0x42; 32], + scope: "/project".into(), + metadata_root: page_id(&root), + }, + empty_root, + directory_fault: AtomicUsize::new(DIRECTORY_OK), + directory_calls: Mutex::new(Vec::new()), + hold_directory: AtomicBool::new(false), + directory_entered: Notify::new(), + directory_release: Notify::new(), + wrong_map_size: AtomicBool::new(false), + hold_map: AtomicBool::new(false), + map_entered: Notify::new(), + map_release: Notify::new(), + chunk_fault: AtomicUsize::new(CHUNK_OK), + hold_chunk: AtomicU64::new(NONE), + chunk_stage: AtomicUsize::new(0), + chunk_entered: Notify::new(), + chunk_release: Notify::new(), + partial_bodies: AtomicUsize::new(0), + short_lease: false, + renewal_status: StatusCode::FORBIDDEN, + renewals: AtomicUsize::new(0), + } + } + + fn sid(&self) -> String { + id(&self.descriptor.snapshot_id().unwrap()) + } + + fn release(&self) { + self.hold_directory.store(false, Ordering::SeqCst); + self.hold_map.store(false, Ordering::SeqCst); + self.hold_chunk.store(NONE, Ordering::SeqCst); + self.directory_release.notify_one(); + self.map_release.notify_one(); + self.chunk_release.notify_one(); + } + + fn assert_no_content(&self) { + assert_eq!(self.inner.maps.load(Ordering::SeqCst), 0); + assert_eq!(self.inner.chunks.load(Ordering::SeqCst), 0); + assert_eq!(self.inner.raw.load(Ordering::SeqCst), 0); + assert_eq!(self.inner.objects.load(Ordering::SeqCst), 0); + } +} + +struct ReleaseGuard(Arc); +impl Drop for ReleaseGuard { + fn drop(&mut self) { + self.0.release(); + } +} + +fn short_expiry() -> String { + let seconds = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() + + 3; + let expiry = time::OffsetDateTime::from_unix_timestamp(seconds as i64).unwrap(); + format!( + "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z", + expiry.year(), + u8::from(expiry.month()), + expiry.day(), + expiry.hour(), + expiry.minute(), + expiry.second() + ) +} + +async fn online_capabilities() -> Json { + Json(json!({ + "protocol_versions": [2], "metadata_codecs": [1], "frame_encodings": ["identity"], + "features": {"resolve": true, "directory": true, "leases": true, + "metadata_pages": false, "objects": true, "chunk_reads": true, "raw_blob": true} + })) +} + +async fn online_resolve(State(f): State>) -> Json { + let d = &f.descriptor; + Json(json!({ + "descriptor": {"schema_version": 2, "metadata_codec": 1, "instance_id": INSTANCE, + "namespace_view_id": id(&d.namespace_view_id), "scope": "/project", "materialization_policy": 1, + "fs_semantics": 1, "access_projection": 0, "metadata_root": id(&d.metadata_root), + "snapshot_id": f.sid()}, + "lease_id": ONLINE_LEASE, + "lease_expires_at": if f.short_lease { short_expiry() } else { "2099-01-01T00:00:00Z".into() }, + "publication_sequence": "1", "authorization_epoch": "1" + })) +} + +async fn lease_header(request: Request, next: Next) -> Response { + if !matches!( + request.uri().path(), + "/api/v2/snapshots/capabilities" | "/api/v2/snapshots/resolve" + ) { + assert_eq!(request.headers()["x-mega-snapshot-lease"], ONLINE_LEASE); + } + next.run(request).await +} + +fn missing(message: &str) -> Response { + ( + StatusCode::NOT_FOUND, + Json(json!({"error":{"code":"PATH_NOT_FOUND","message":message}})), + ) + .into_response() +} + +async fn online_directory( + State(f): State>, + HttpPath(snapshot): HttpPath, + Query(query): Query>, +) -> Response { + assert_eq!(snapshot, f.sid()); + let path = &query["path"]; + let cursor = query.get("cursor").cloned(); + f.directory_calls + .lock() + .unwrap() + .push((path.clone(), cursor.clone())); + let fault = f.directory_fault.load(Ordering::SeqCst); + if path == "/a" && fault == MISSING_CHILD { + return missing("parent-advertised child is missing"); + } + if path == "/" && cursor.is_some() { + assert_eq!(cursor.as_deref(), Some("root-second")); + if f.hold_directory.load(Ordering::SeqCst) { + f.directory_entered.notify_one(); + f.directory_release.notified().await; + } + if fault == MISSING_SECOND_PAGE { + return missing("advertised second root page is missing"); + } + } + let mut metadata_root = id(&f.descriptor.metadata_root); + if fault == WRONG_ROOT { + metadata_root = id(&[0x93; 32]); + } + let (directory_root, entries, count, start, next) = match (path.as_str(), cursor.as_deref()) { + ("/", None) => ( + id(&f.descriptor.metadata_root), + vec![json!({"name":"a","fs_kind":"directory","directory_root":id(&f.empty_root)})], + "2", + None, + Some("root-second"), + ), + ("/", Some("root-second")) => ( + id(&f.descriptor.metadata_root), + vec![ + json!({"name":"large.bin","fs_kind":"regular", "size":f.inner.map.file_size.to_string(),"content_digest":id(&f.inner.advertised)}), + ], + "2", + Some("a"), + if fault == LOOP_CURSOR { + Some("root-second") + } else { + None + }, + ), + ("/a", None) => (id(&f.empty_root), vec![], "0", None, None), + _ => panic!("unexpected directory request: {path:?} {cursor:?}"), + }; + Json(json!({ + "snapshot_id":snapshot,"path":path,"metadata_root":metadata_root, + "directory_root":directory_root,"node_class":"native_tree","lifecycle":"mutable", + "range_start_exclusive":start,"entries":entries,"entry_count":count, + "next_cursor":next,"proof_pages":[] + })) + .into_response() +} + +async fn online_map( + State(f): State>, + HttpPath(snapshot): HttpPath, + Query(query): Query>, +) -> Json { + assert_eq!(snapshot, f.sid()); + assert_eq!(query["path"], "/large.bin"); + assert_eq!(query["expected_digest"], id(&f.inner.advertised)); + let Json(mut value) = + super::map(State(f.inner.clone()), HttpPath(snapshot), Query(query)).await; + if f.wrong_map_size.load(Ordering::SeqCst) { + let wrong = ChunkMap::new( + f.inner.advertised, + f.inner.map.file_size + 1, + f.inner.map.pages_root, + ) + .unwrap(); + value["file_size"] = json!(wrong.file_size.to_string()); + value["map_id"] = json!(id(&wrong.map_id())); + value["chunk_count"] = json!(wrong.chunk_count.to_string()); + value["page_count"] = json!(wrong.page_count.to_string()); + } + if f.hold_map.load(Ordering::SeqCst) { + f.map_entered.notify_one(); + f.map_release.notified().await; + } + Json(value) +} + +async fn online_leaf( + State(f): State>, + HttpPath(snapshot): HttpPath, + Query(query): Query>, +) -> Json { + assert_eq!(snapshot, f.sid()); + super::leaf(State(f.inner.clone()), HttpPath(snapshot), Query(query)).await +} + +async fn online_chunks( + State(f): State>, + HttpPath(snapshot): HttpPath, + body: Bytes, +) -> Response { + assert_eq!(snapshot, f.sid()); + let request: Value = serde_json::from_slice(&body).unwrap(); + let items = request["items"].as_array().unwrap(); + assert_eq!(items.len(), 1); + let item = &items[0]; + let index: u64 = item["chunk_index"].as_str().unwrap().parse().unwrap(); + assert_eq!(item["path"], "/large.bin"); + assert_eq!(item["map_id"], id(&f.inner.map.map_id())); + assert_eq!(item["expected_digest"], id(&f.inner.advertised)); + f.inner.chunks.fetch_add(1, Ordering::SeqCst); + let held = f.hold_chunk.load(Ordering::SeqCst) == index; + if held && f.chunk_stage.load(Ordering::SeqCst) == BEFORE_HEADERS { + f.chunk_entered.notify_one(); + f.chunk_release.notified().await; + } + let length = f.inner.map.chunk_len(index).unwrap(); + let mut payload = ChunkPayload { + map_id: f.inner.map.map_id(), + file_content_id: f.inner.advertised, + chunk_index: index, + chunk_bytes: vec![pattern(index); length as usize], + }; + let fault = if index == 1 { + f.chunk_fault.load(Ordering::SeqCst) + } else { + CHUNK_OK + }; + match fault { + SHORT_CHUNK => { + payload.chunk_bytes.pop(); + } + WRONG_CHUNK_HASH => payload.chunk_bytes[0] ^= 1, + WRONG_CHUNK_MAP => payload.map_id[0] ^= 1, + WRONG_CHUNK_CONTENT => payload.file_content_id[0] ^= 1, + _ => {} + } + let data = payload.encode(21, 0).unwrap(); + let mut end = EndPayload { + request_item_count: 1, + unique_unit_count: 1, + logical_bytes: length, + request_body_sha256: hash(&body), + }; + if fault == WRONG_END { + end.request_body_sha256[0] ^= 1; + } + let end = end.encode(21, 1); + if held && f.chunk_stage.load(Ordering::SeqCst) == AFTER_CHUNK_BODY { + // The second stream poll signals that actual CHUNK bytes have been + // supplied to HTTP; END remains withheld, so no chunk can publish. + let pending = f.clone(); + let stream = futures::stream::once( + async move { Ok::<_, std::io::Error>(Bytes::from(data)) }, + ) + .chain(futures::stream::once(async move { + pending.partial_bodies.fetch_add(1, Ordering::SeqCst); + pending.chunk_entered.notify_one(); + pending.chunk_release.notified().await; + Ok::<_, std::io::Error>(Bytes::from(end)) + })); + return Response::builder() + .header("content-type", "application/vnd.mega.treeframe;version=2") + .header("x-mega-snapshot-id", snapshot) + .header("x-mega-request-digest", id(&hash(&body))) + .body(Body::from_stream(stream)) + .unwrap(); + } + let mut wire = data; + wire.extend(end); + response(&snapshot, &body, wire) +} + +async fn online_raw(State(f): State>) -> Response { + super::raw(State(f.inner.clone())).await +} + +async fn online_objects(State(f): State>) -> StatusCode { + f.inner.objects.fetch_add(1, Ordering::SeqCst); + StatusCode::INTERNAL_SERVER_ERROR +} + +async fn online_renew( + State(f): State>, + HttpPath(lease): HttpPath, + Json(body): Json, +) -> Response { + assert_eq!(lease, ONLINE_LEASE); + assert_eq!(body["lease_seconds"], 600); + f.renewals.fetch_add(1, Ordering::SeqCst); + let code = if f.renewal_status == StatusCode::GONE { + "LEASE_EXPIRED" + } else { + "SCOPE_FORBIDDEN" + }; + ( + f.renewal_status, + Json(json!({"error":{"code":code,"message":"actual online renewal rejected"}})), + ) + .into_response() +} + +async fn open_online( + fixture: OnlineFixture, + temp: &Path, + limits: crate::snapshot::ContentBudgetLimits, +) -> (Server, Arc, SnapshotReader, DurableStore) { + let fixture = Arc::new(fixture); + let app = Router::new() + .route("/api/v2/snapshots/capabilities", get(online_capabilities)) + .route("/api/v2/snapshots/resolve", post(online_resolve)) + .route("/api/v2/snapshots/leases/{lease}/renew", post(online_renew)) + .route( + "/api/v2/snapshots/{snapshot}/directory", + get(online_directory), + ) + .route("/api/v2/snapshots/{snapshot}/chunk-map", get(online_map)) + .route( + "/api/v2/snapshots/{snapshot}/chunk-map/pages", + get(online_leaf), + ) + .route("/api/v2/snapshots/{snapshot}/chunks", post(online_chunks)) + .route("/api/v2/snapshots/{snapshot}/objects", post(online_objects)) + .route("/api/v2/snapshots/{snapshot}/blob", get(online_raw)) + .layer(axum::middleware::from_fn(lease_header)) + .with_state(fixture.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let server = Server(tokio::spawn(async move { + axum::serve(listener, app).await.unwrap() + })); + let reader = SnapshotReader::resolve(crate::snapshot::Mst2Client::new(base), "/project", 600) + .await + .unwrap() + .with_content_limits(limits); + assert!(!reader.capabilities().features.metadata_pages); + assert!(reader.capabilities().features.chunk_reads); + let store = + DurableStore::open_for_reader(temp.join("view"), temp.join("cas"), &reader).unwrap(); + (server, fixture, reader, store) +} + +fn content_limits(output: usize) -> crate::snapshot::ContentBudgetLimits { + crate::snapshot::ContentBudgetLimits::new(output, 8 * MIB).unwrap() +} + +fn assert_walk(fixture: &OnlineFixture, walks: usize) { + let expected: Vec<_> = (0..walks) + .flat_map(|_| { + [ + ("/".to_string(), None), + ("/a".to_string(), None), + ("/".to_string(), Some("root-second".to_string())), + ] + }) + .collect(); + assert_eq!(*fixture.directory_calls.lock().unwrap(), expected); +} + +async fn settled_usage(reader: &SnapshotReader, output: usize) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let usage = reader.content_usage(); + if usage.output_bytes == output && usage.construction_bytes == 0 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("cancelled stream kept managed content credits"); +} + +#[tokio::test] +async fn no_pages_completed_manifest_keeps_tuple_and_reader_authority_boundaries() { + let temp = tempfile::tempdir().unwrap(); + let size = 3 * MIB as u64 + 7; + let (_server, fixture, reader, _store) = open_online( + OnlineFixture::new(size, false), + temp.path(), + content_limits(8 * MIB), + ) + .await; + let manifest = crate::snapshot::online_file::CompletedOnlineManifest::walk(&reader) + .await + .unwrap(); + assert_walk(&fixture, 1); + let file = &manifest.files()[0]; + for altered in [ + SnapshotFile { + rel_path: "another.bin".into(), + ..file.clone() + }, + SnapshotFile { + content_digest: id(&[0x79; 32]), + ..file.clone() + }, + SnapshotFile { + size: file.size + 1, + ..file.clone() + }, + SnapshotFile { + fs_kind: "executable".into(), + ..file.clone() + }, + ] { + assert_eq!( + manifest.range_file(&altered).unwrap_err().code, + SnapshotErrorCode::IntegrityError + ); + } + // Private online authority remains distinct from the strict public open. + assert_eq!( + crate::snapshot::OwnedChunkedFile::open( + &reader, + &file.rel_path, + &file.content_digest, + file.size + ) + .await + .err() + .unwrap() + .code, + SnapshotErrorCode::SnapshotNotReady, + ); + fixture.assert_no_content(); + let token = manifest.range_file(file).unwrap(); + let foreign_temp = tempfile::tempdir().unwrap(); + let (_foreign_server, foreign_fixture, foreign_reader, _foreign_store) = open_online( + OnlineFixture::new(size, false), + foreign_temp.path(), + content_limits(8 * MIB), + ) + .await; + assert_eq!(reader.snapshot_id(), foreign_reader.snapshot_id()); + assert_eq!( + crate::snapshot::OwnedChunkedFile::open_online_path(&foreign_reader, token.clone()) + .await + .err() + .unwrap() + .code, + SnapshotErrorCode::IntegrityError, + ); + foreign_fixture.assert_no_content(); + let range = crate::snapshot::OwnedChunkedFile::open_online_path(&reader, token) + .await + .unwrap(); + let bytes = range.read_range_owned(0, 32).await.unwrap(); + assert_eq!(bytes.as_bytes(), &[pattern(0); 32]); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 1); + assert_eq!(fixture.inner.chunks.load(Ordering::SeqCst), 1); + assert!(reader.content_membership.get().is_none()); + drop(range); + drop(bytes); + settled_usage(&reader, 0).await; + settled_usage(&foreign_reader, 0).await; +} + +#[tokio::test] +async fn no_pages_public_hydrate_streams_large_file_closure_reopens_and_resumes_without_content() { + let temp = tempfile::tempdir().unwrap(); + let size = 65 * MIB as u64 + 7; + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(size, false), + temp.path(), + content_limits(24 * MIB), + ) + .await; + assert!(size > crate::snapshot::client::MAX_BUFFERED_FILE_BYTES); + let report = store.hydrate(&reader).await.unwrap(); + assert_eq!( + ( + report.fetched, + report.resumed, + report.total_files, + report.bytes_total + ), + (1, 0, 1, size) + ); + assert_walk(&fixture, 1); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 1); + assert_eq!( + fixture.inner.chunks.load(Ordering::SeqCst) as u64, + fixture.inner.map.chunk_count + ); + assert_eq!(fixture.inner.raw.load(Ordering::SeqCst), 0); + assert_eq!(fixture.inner.objects.load(Ordering::SeqCst), 0); + assert_eq!( + store.completion_kind().unwrap(), + Some(CompletionKind::FileClosure) + ); + assert!(store.is_complete().unwrap()); + assert!(!store.is_snapshot_complete().unwrap()); + assert_eq!(store.manifest().unwrap()[0].rel_path, "large.bin"); + assert!(store + .verify_blob( + &id(&fixture.inner.whole), + size, + CasVerificationReason::CompletionAudit + ) + .unwrap()); + assert!(reader.content_membership.get().is_none()); + settled_usage(&reader, 0).await; + let reopened = + DurableStore::open_for_reader(store.root(), store.content_dir(), &reader).unwrap(); + assert_eq!( + reopened.completion_kind().unwrap(), + Some(CompletionKind::FileClosure) + ); + assert_eq!( + reopened + .read_verified_blob_range(&id(&fixture.inner.whole), size, size - 4, 20) + .unwrap() + .unwrap(), + vec![pattern(65); 4] + ); + let warm = reopened.hydrate(&reader).await.unwrap(); + assert_eq!( + ( + warm.fetched, + warm.resumed, + warm.total_files, + warm.bytes_total + ), + (0, 1, 1, size) + ); + assert_walk(&fixture, 2); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 1); + assert_eq!( + fixture.inner.chunks.load(Ordering::SeqCst) as u64, + fixture.inner.map.chunk_count + ); + assert_eq!(fixture.inner.raw.load(Ordering::SeqCst), 0); + assert!(reader.content_membership.get().is_none()); +} + +#[tokio::test] +async fn no_pages_missing_second_page_child_loop_and_wrong_root_never_start_content() { + for fault in [MISSING_SECOND_PAGE, MISSING_CHILD, LOOP_CURSOR, WRONG_ROOT] { + let temp = tempfile::tempdir().unwrap(); + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(3 * MIB as u64 + 7, false), + temp.path(), + content_limits(8 * MIB), + ) + .await; + fixture.directory_fault.store(fault, Ordering::SeqCst); + let error = store.hydrate(&reader).await.unwrap_err(); + assert_eq!( + error.code, + if matches!(fault, MISSING_SECOND_PAGE | MISSING_CHILD) { + SnapshotErrorCode::PathNotFound + } else { + SnapshotErrorCode::IntegrityError + }, + "fault {fault}" + ); + fixture.assert_no_content(); + unpublished(&store, &fixture.inner.advertised); + assert!(store.try_transaction().unwrap().is_some()); + assert!(reader.content_membership.get().is_none()); + settled_usage(&reader, 0).await; + fixture + .directory_fault + .store(DIRECTORY_OK, Ordering::SeqCst); + let report = store.hydrate(&reader).await.unwrap(); + assert_eq!(report.fetched, 1); + assert_eq!( + store.completion_kind().unwrap(), + Some(CompletionKind::FileClosure) + ); + } +} + +#[tokio::test] +async fn no_pages_partial_directory_walk_cancellation_mints_no_stream_authority() { + let temp = tempfile::tempdir().unwrap(); + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(3 * MIB as u64 + 7, false), + temp.path(), + content_limits(8 * MIB), + ) + .await; + let _release = ReleaseGuard(fixture.clone()); + fixture.hold_directory.store(true, Ordering::SeqCst); + let mut operation = Box::pin(store.hydrate(&reader)); + tokio::select! { + result = &mut operation => panic!("partial walk unexpectedly completed: {result:?}"), + _ = fixture.directory_entered.notified() => {}, + _ = tokio::time::sleep(Duration::from_secs(10)) => panic!("second directory page was never requested"), + } + assert_walk(&fixture, 1); + drop(operation); + fixture.release(); + fixture.assert_no_content(); + unpublished(&store, &fixture.inner.advertised); + assert!(store.try_transaction().unwrap().is_some()); + assert!(reader.content_membership.get().is_none()); + settled_usage(&reader, 0).await; + store.hydrate(&reader).await.unwrap(); + assert_eq!( + store.completion_kind().unwrap(), + Some(CompletionKind::FileClosure) + ); +} + +#[tokio::test] +async fn no_pages_map_chunk_and_end_errors_never_publish_and_retry_the_actual_stream() { + for fault in 0..=WRONG_END { + let temp = tempfile::tempdir().unwrap(); + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(3 * MIB as u64 + 7, false), + temp.path(), + content_limits(8 * MIB), + ) + .await; + fixture.wrong_map_size.store(fault == 0, Ordering::SeqCst); + fixture.chunk_fault.store(fault, Ordering::SeqCst); + assert_eq!( + store.hydrate(&reader).await.unwrap_err().code, + SnapshotErrorCode::DigestMismatch, + "fault {fault}" + ); + assert_walk(&fixture, 1); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 1); + assert_eq!( + fixture.inner.chunks.load(Ordering::SeqCst), + if fault == 0 { 0 } else { 2 } + ); + assert_eq!(fixture.inner.raw.load(Ordering::SeqCst), 0); + unpublished(&store, &fixture.inner.advertised); + assert!(store.try_transaction().unwrap().is_some()); + settled_usage(&reader, 0).await; + fixture.wrong_map_size.store(false, Ordering::SeqCst); + fixture.chunk_fault.store(CHUNK_OK, Ordering::SeqCst); + let before = fixture.inner.chunks.load(Ordering::SeqCst); + store.hydrate(&reader).await.unwrap(); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 2); + assert_eq!( + fixture.inner.chunks.load(Ordering::SeqCst) - before, + fixture.inner.map.chunk_count as usize + ); + assert_eq!( + store.completion_kind().unwrap(), + Some(CompletionKind::FileClosure) + ); + } +} + +#[tokio::test] +async fn no_pages_verified_chunks_with_wrong_whole_digest_never_complete() { + let temp = tempfile::tempdir().unwrap(); + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(3 * MIB as u64 + 7, true), + temp.path(), + content_limits(8 * MIB), + ) + .await; + assert_eq!( + store.hydrate(&reader).await.unwrap_err().code, + SnapshotErrorCode::DigestMismatch + ); + assert_eq!( + fixture.inner.chunks.load(Ordering::SeqCst) as u64, + fixture.inner.map.chunk_count + ); + assert_eq!(fixture.inner.raw.load(Ordering::SeqCst), 0); + unpublished(&store, &fixture.inner.advertised); + assert!(!store.blob_path(&id(&fixture.inner.whole)).unwrap().exists()); + settled_usage(&reader, 0).await; +} + +#[tokio::test] +async fn no_pages_output_budget_rejects_before_chunk_body_and_low_budget_stream_keeps_progress() { + let temp = tempfile::tempdir().unwrap(); + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(6 * MIB as u64 + 7, false), + temp.path(), + content_limits(MIB + 1024), + ) + .await; + assert_eq!( + store.hydrate(&reader).await.unwrap_err().code, + SnapshotErrorCode::LimitExceeded + ); + assert_walk(&fixture, 1); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 1); + assert_eq!(fixture.inner.chunks.load(Ordering::SeqCst), 0); + assert_eq!(fixture.inner.raw.load(Ordering::SeqCst), 0); + unpublished(&store, &fixture.inner.advertised); + settled_usage(&reader, 0).await; + + let temp = tempfile::tempdir().unwrap(); + let size = 20 * MIB as u64 + 7; + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(size, false), + temp.path(), + content_limits(3 * MIB), + ) + .await; + let report = store.hydrate(&reader).await.unwrap(); + assert_eq!((report.fetched, report.bytes_total), (1, size)); + assert_eq!( + fixture.inner.chunks.load(Ordering::SeqCst) as u64, + fixture.inner.map.chunk_count + ); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 1); + assert_eq!(fixture.inner.raw.load(Ordering::SeqCst), 0); + assert!(store + .verify_blob( + &id(&fixture.inner.whole), + size, + CasVerificationReason::CompletionAudit + ) + .unwrap()); + assert_eq!( + store.completion_kind().unwrap(), + Some(CompletionKind::FileClosure) + ); + settled_usage(&reader, 0).await; +} + +#[tokio::test] +async fn no_pages_cancelled_header_and_body_streams_release_only_their_paid_owners() { + for stage in [BEFORE_HEADERS, AFTER_CHUNK_BODY] { + let temp = tempfile::tempdir().unwrap(); + let (_server, fixture, reader, store) = open_online( + OnlineFixture::new(6 * MIB as u64 + 7, false), + temp.path(), + content_limits(8 * MIB), + ) + .await; + let _release = ReleaseGuard(fixture.clone()); + let (previous, marker, old_digest) = previous_complete(&store, temp.path()).await; + let manifest = crate::snapshot::online_file::CompletedOnlineManifest::walk(&reader) + .await + .unwrap(); + let token = manifest.range_file(&manifest.files()[0]).unwrap(); + let range = crate::snapshot::OwnedChunkedFile::open_online_path(&reader, token) + .await + .unwrap(); + let retained = range.read_range_owned(0, 32).await.unwrap(); + assert_eq!(retained.as_bytes(), &[pattern(0); 32]); + drop(range); + let baseline = reader.content_usage().output_bytes; + assert_eq!(baseline, 1024); + settled_usage(&reader, baseline).await; + let before = fixture.inner.chunks.load(Ordering::SeqCst); + let before_wire = reader.client().received_bytes(); + fixture.hold_chunk.store(1, Ordering::SeqCst); + fixture.chunk_stage.store(stage, Ordering::SeqCst); + let mut operation = Box::pin(store.hydrate(&reader)); + tokio::select! { + result = &mut operation => panic!("blocked stream unexpectedly completed: {result:?}"), + _ = fixture.chunk_entered.notified() => {}, + _ = tokio::time::sleep(Duration::from_secs(10)) => panic!("chunk gate was never reached"), + } + assert_eq!(fixture.inner.chunks.load(Ordering::SeqCst) - before, 2); + assert_eq!( + fixture.partial_bodies.load(Ordering::SeqCst), + usize::from(stage == AFTER_CHUNK_BODY) + ); + if stage == AFTER_CHUNK_BODY { + let header = mst2_codec::treeframe::HEADER_LEN as u64; + let minimum = 2 * (CHUNK_SIZE as u64 + header + 76) + header + 48; + // Keep polling the operation until the client's actual wire meter + // records both CHUNK frames; the second END remains withheld. + tokio::select! { + result = &mut operation => panic!("withheld END unexpectedly completed: {result:?}"), + result = tokio::time::timeout(Duration::from_secs(5), async { + while reader.client().received_bytes() - before_wire < minimum { + tokio::task::yield_now().await; + } + }) => result.expect("client did not receive the actual CHUNK body before cancellation"), + } + assert!(reader.client().received_bytes() - before_wire >= minimum); + } + let pending_usage = reader.content_usage(); + assert!(pending_usage.output_bytes > baseline); + assert!(pending_usage.output_bytes <= 8 * MIB); + assert!(pending_usage.construction_bytes <= 8 * MIB); + drop(operation); + fixture.release(); + settled_usage(&reader, baseline).await; + assert_eq!(retained.as_bytes(), &[pattern(0); 32]); + unpublished(&store, &fixture.inner.advertised); + assert!(store.try_transaction().unwrap().is_some()); + previous_unchanged(&previous, &marker, &old_digest); + drop(retained); + settled_usage(&reader, 0).await; + store.hydrate(&reader).await.unwrap(); + assert_eq!( + store.completion_kind().unwrap(), + Some(CompletionKind::FileClosure) + ); + previous_unchanged(&previous, &marker, &old_digest); + } +} + +async fn renewal_rejected( + fixture: &OnlineFixture, + reader: &SnapshotReader, + expected: SnapshotErrorCode, +) { + tokio::time::timeout(Duration::from_secs(5), async { + while fixture.renewals.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("real renewal endpoint was never reached"); + assert_eq!(reader.ensure_lease().await.unwrap_err().code, expected); +} + +#[tokio::test] +async fn no_pages_manifest_and_pending_chunk_replies_reject_real_403_and_410_renewals() { + for status in [StatusCode::FORBIDDEN, StatusCode::GONE] { + for at_map in [true, false] { + let temp = tempfile::tempdir().unwrap(); + let mut config = OnlineFixture::new(3 * MIB as u64 + 7, false); + config.short_lease = true; + config.renewal_status = status; + let (_server, fixture, reader, store) = + open_online(config, temp.path(), content_limits(8 * MIB)).await; + let _release = ReleaseGuard(fixture.clone()); + let expected = if status == StatusCode::FORBIDDEN { + SnapshotErrorCode::ScopeForbidden + } else { + SnapshotErrorCode::LeaseExpired + }; + fixture.hold_map.store(at_map, Ordering::SeqCst); + if !at_map { + fixture.hold_chunk.store(0, Ordering::SeqCst); + fixture.chunk_stage.store(BEFORE_HEADERS, Ordering::SeqCst); + } + let mut operation = Box::pin(store.hydrate(&reader)); + let entered = if at_map { + &fixture.map_entered + } else { + &fixture.chunk_entered + }; + tokio::select! { + result = &mut operation => panic!("held operation unexpectedly completed: {result:?}"), + _ = entered.notified() => {}, + _ = tokio::time::sleep(Duration::from_secs(10)) => panic!("post-manifest network gate was never reached"), + } + assert_walk(&fixture, 1); + assert_eq!(fixture.inner.maps.load(Ordering::SeqCst), 1); + assert_eq!( + fixture.inner.chunks.load(Ordering::SeqCst), + usize::from(!at_map) + ); + // Concurrently poll the operation so the actual HTTP response is + // still pending while the real background renewal latches failure. + tokio::select! { + result = &mut operation => panic!("pending response escaped before release: {result:?}"), + _ = renewal_rejected(&fixture, &reader, expected) => {}, + } + fixture.release(); + assert_eq!(operation.await.unwrap_err().code, expected); + assert_eq!(fixture.renewals.load(Ordering::SeqCst), 1); + assert_eq!(fixture.inner.raw.load(Ordering::SeqCst), 0); + unpublished(&store, &fixture.inner.advertised); + assert!(store.try_transaction().unwrap().is_some()); + assert!(reader.content_membership.get().is_none()); + settled_usage(&reader, 0).await; + } + } +} diff --git a/src/snapshot/durable_stream_tests.rs b/src/snapshot/durable_stream_tests.rs index e833888..9d00ed1 100644 --- a/src/snapshot/durable_stream_tests.rs +++ b/src/snapshot/durable_stream_tests.rs @@ -29,6 +29,9 @@ use tokio::sync::Notify; use super::{durability_tests::FaultGuard, *}; +#[path = "durable_online_tests.rs"] +mod online_retirement; + // This batch fixture exercises the coordinator's fixed final allocation while // the existing single-chunk fixture below keeps testing streaming CAS writes. async fn owned_chunks( @@ -974,15 +977,15 @@ async fn range_profile_rejects_above_eight_tib_before_request_and_accepts_bounda fixture.descriptor.metadata_root = page_id(&fixture.page); let (_server, fixture, reader, _store) = open_fixture(fixture, temp.path()).await; let digest = id(&fixture.advertised); - match crate::snapshot::ChunkedFile::open(&reader, "large.bin", &digest, size + 1).await { + match crate::snapshot::OwnedChunkedFile::open(&reader, "large.bin", &digest, size + 1).await { Ok(_) => panic!("over-profile range handle was accepted"), Err(error) => assert_eq!(error.code, SnapshotErrorCode::LimitExceeded), } assert_eq!(fixture.maps.load(Ordering::SeqCst), 0); - let handle = crate::snapshot::ChunkedFile::open(&reader, "large.bin", &digest, size) + let handle = crate::snapshot::OwnedChunkedFile::open(&reader, "large.bin", &digest, size) .await .unwrap(); - assert_eq!(handle.size, size); + assert_eq!(handle.size(), size); assert_eq!(handle.map_id(), id(&fixture.map.map_id())); assert_eq!(fixture.maps.load(Ordering::SeqCst), 1); assert_eq!(fixture.chunks.load(Ordering::SeqCst), 0); diff --git a/src/snapshot/frames.rs b/src/snapshot/frames.rs index f717ddd..c8f7eb3 100644 --- a/src/snapshot/frames.rs +++ b/src/snapshot/frames.rs @@ -239,6 +239,23 @@ pub struct VerifiedChunkMap { pub pages_root: [u8; 32], } +impl VerifiedChunkMap { + /// Length of chunk `index` derived from the fixed profile map. + pub fn chunk_len(&self, index: u64) -> Result { + let map = mst2_codec::chunkmap::ChunkMap::new( + parse_digest(&self.file_content_id)?, + self.file_size, + self.pages_root, + ) + .map_err(|error| { + SnapshotError::new(SnapshotErrorCode::Internal, format!("chunk codec: {error}")) + })?; + map.chunk_len(index).map_err(|error| { + SnapshotError::new(SnapshotErrorCode::Internal, format!("chunk codec: {error}")) + }) + } +} + /// One verified chunk, in request order. #[derive(Debug, Clone)] pub struct ChunkUnit { @@ -1256,6 +1273,27 @@ pub fn check_merkle_root(leaves: &[[u8; 32]], root: [u8; 32]) -> Result<(), Snap #[cfg(test)] mod tests { + #[test] + fn chunk_len_matches_the_map_rules() { + use mst2_codec::chunkmap::CHUNK_SIZE; + + use super::{parse_digest, VerifiedChunkMap}; + + let size = 2 * CHUNK_SIZE as u64 + 7; + let id = "sha256:".to_string() + &"0".repeat(64); + let map = VerifiedChunkMap { + file_content_id: id.clone(), + map_id: id, + file_size: size, + chunk_count: 3, + page_count: 1, + pages_root: [0u8; 32], + }; + assert_eq!(map.chunk_len(0).unwrap(), CHUNK_SIZE as u64); + assert_eq!(map.chunk_len(2).unwrap(), 7); + assert!(map.chunk_len(3).is_err()); + assert!(parse_digest(&map.file_content_id).is_ok()); + } use super::*; #[test] diff --git a/src/snapshot/lookup.rs b/src/snapshot/lookup.rs index f669eee..cecff88 100644 --- a/src/snapshot/lookup.rs +++ b/src/snapshot/lookup.rs @@ -2,8 +2,8 @@ //! or absence against the descriptor's metadata root. use super::{ + content_profile::MAX_FILE_SIZE, frames::parse_count, - range::MAX_FILE_SIZE, types::{LookupNode, SnapshotError, SnapshotErrorCode}, }; diff --git a/src/snapshot/mod.rs b/src/snapshot/mod.rs index 6a91a78..57f27e7 100644 --- a/src/snapshot/mod.rs +++ b/src/snapshot/mod.rs @@ -16,6 +16,7 @@ mod chunk_wire; pub mod client; pub mod closure; mod content; +mod content_profile; pub mod coordinator; mod descriptor_wire; mod directory; @@ -35,7 +36,6 @@ mod owned_reader; mod owned_transport; mod path_state; mod proven_file; -pub mod range; pub mod reader; mod resolve_receipt; mod resolve_wire; @@ -49,6 +49,7 @@ pub use auth::{AuthorizedSnapshotContext, CacheDomain}; pub use client::Mst2Client; pub use closure::{SnapshotClosureMeters, SnapshotDirectory, ValidatedSnapshotClosure}; pub use content::{ContentBudgetLimits, ContentBudgetUsage, VerifiedContent, VerifiedContentBatch}; +pub use content_profile::OBJECT_CAP; pub use coordinator::{FetchCoordinator, FetchCoordinatorCounts, FetchCoordinatorLimits}; pub use durable::{CompletionKind, DurableStore, HydrateReport, LocalCasRangeMeters, ViewMeta}; pub use frames::LeaseReleaseOutcome; @@ -59,7 +60,6 @@ pub use path_state::{ SnapshotPathState, }; pub use proven_file::{FileMembershipError, ProvenSnapshotFile}; -pub use range::{ChunkedFile, OBJECT_CAP}; pub use reader::{SnapshotFile, SnapshotReader}; pub use resolve_receipt::ResolveTraceReceipt; pub use types::{ diff --git a/src/snapshot/online_file.rs b/src/snapshot/online_file.rs index e347ddb..6234fa5 100644 --- a/src/snapshot/online_file.rs +++ b/src/snapshot/online_file.rs @@ -1,13 +1,71 @@ //! Private fixed-SID path authority for a completed online directory manifest. //! -//! This token is not an MTP2 membership proof. Only the public file-manifest -//! mount constructors mint it after their actual fixed-SID directory walk. +//! This token is not an MTP2 membership proof. The mount and hydration +//! constructors mint it only after their actual fixed-SID directory walk. //! Each path retains an independent online request and current lease check. -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; use super::{CacheDomain, SnapshotError, SnapshotErrorCode, SnapshotFile, SnapshotReader}; +/// Hydration may stream only entries from this completed online walk. No +/// caller-supplied manifest constructor or fixed-root conversion is provided. +pub(super) struct CompletedOnlineManifest { + files: Vec, + ranges: HashMap>, +} + +impl CompletedOnlineManifest { + pub(super) async fn walk(reader: &SnapshotReader) -> Result { + if reader.capabilities().features.metadata_pages { + return Err(SnapshotError::new( + SnapshotErrorCode::InvalidRequest, + "online manifest cannot replace fixed-root membership", + )); + } + // Do not mint any authority before every directory/cursor succeeds. + let files = reader.file_manifest_directory().await?; + reader.ensure_lease().await?; + let mut ranges = HashMap::new(); + for file in &files { + if file.size > super::OBJECT_CAP { + let authority = OnlineSnapshotFile::from_manifest_file(reader, file.clone())?; + if ranges.insert(file.rel_path.clone(), authority).is_some() { + return Err(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "completed online manifest repeats a file path", + )); + } + } + } + reader.local_lease_status()?; + Ok(Self { files, ranges }) + } + + pub(super) fn files(&self) -> &[SnapshotFile] { + &self.files + } + + pub(super) fn range_file( + &self, + file: &SnapshotFile, + ) -> Result, SnapshotError> { + let authority = self.ranges.get(&file.rel_path).ok_or_else(|| { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "file is not a large entry in the completed online manifest", + ) + })?; + if authority.file() != file { + return Err(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "stream tuple differs from the completed online manifest", + )); + } + Ok(authority.clone()) + } +} + #[derive(Debug)] pub(super) struct OnlineSnapshotFile { file: SnapshotFile, diff --git a/src/snapshot/owned_range.rs b/src/snapshot/owned_range.rs index 73fa7ba..06dd757 100644 --- a/src/snapshot/owned_range.rs +++ b/src/snapshot/owned_range.rs @@ -154,25 +154,45 @@ impl ChunkCache { type LeafEntry = (u64, Arc>); struct LeafCache { entries: [Option; CACHED_LEAVES], - next: usize, + len: usize, } impl LeafCache { fn new() -> Self { Self { entries: std::array::from_fn(|_| None), - next: 0, + len: 0, + } + } + fn remove(&mut self, slot: usize) -> LeafEntry { + let entry = self.entries[slot].take().unwrap(); + for position in slot..self.len - 1 { + self.entries[position] = self.entries[position + 1].take(); } + self.len -= 1; + entry } - fn get(&self, index: u64) -> Option>> { - self.entries + fn get(&mut self, index: u64) -> Option>> { + let slot = self.entries[..self.len] .iter() - .flatten() - .find(|entry| entry.0 == index) - .map(|entry| entry.1.clone()) + .position(|entry| entry.as_ref().unwrap().0 == index)?; + let entry = self.remove(slot); + let digests = entry.1.clone(); + self.entries[self.len] = Some(entry); + self.len += 1; + Some(digests) } fn insert(&mut self, index: u64, digests: Arc>) { - self.entries[self.next] = Some((index, digests)); - self.next = (self.next + 1) % CACHED_LEAVES; + if let Some(slot) = self.entries[..self.len] + .iter() + .position(|entry| entry.as_ref().unwrap().0 == index) + { + drop(self.remove(slot)); + } + if self.len == CACHED_LEAVES { + drop(self.remove(0)); + } + self.entries[self.len] = Some((index, digests)); + self.len += 1; } } @@ -232,7 +252,7 @@ impl OwnedChunkedFile { ) -> Result { reader.authorized_context().validate_relative_path(path)?; reader.client().validate_file_size(size)?; - if !(OBJECT_CAP + 1..=super::range::MAX_FILE_SIZE).contains(&size) { + if !(OBJECT_CAP + 1..=super::content_profile::MAX_FILE_SIZE).contains(&size) { return Err(SnapshotError::new( SnapshotErrorCode::LimitExceeded, "owned chunk range requires a large file within 8 TiB", @@ -278,7 +298,7 @@ impl OwnedChunkedFile { file: SnapshotFile, authority: FileAuthority, ) -> Result { - if !(OBJECT_CAP + 1..=super::range::MAX_FILE_SIZE).contains(&file.size) { + if !(OBJECT_CAP + 1..=super::content_profile::MAX_FILE_SIZE).contains(&file.size) { return Err(SnapshotError::new( SnapshotErrorCode::LimitExceeded, "owned chunk range requires a large file within 8 TiB", @@ -491,7 +511,7 @@ impl OwnedChunkedFile { async fn ensure_leaf(&self, page: u64) -> Result>, SnapshotError> { self.validate_file().await?; { - let cache = self.leaves.lock().await; + let mut cache = self.leaves.lock().await; self.reader.local_lease_status()?; if let Some(digests) = cache.get(page) { return Ok(digests); diff --git a/src/snapshot/owned_range_tests.rs b/src/snapshot/owned_range_tests.rs index 4230786..317cfed 100644 --- a/src/snapshot/owned_range_tests.rs +++ b/src/snapshot/owned_range_tests.rs @@ -713,7 +713,7 @@ async fn eight_tib_map_shape_opens_without_allocating_or_fetching_file_payload() let mut fixture = Fixture::new(1); fixture.map = ChunkMap::new( fixture.map.file_content_id, - super::super::range::MAX_FILE_SIZE, + super::super::content_profile::MAX_FILE_SIZE, fixture.map.pages_root, ) .unwrap(); @@ -732,7 +732,7 @@ async fn eight_tib_map_shape_opens_without_allocating_or_fetching_file_payload() &server.reader, "file", &digest, - super::super::range::MAX_FILE_SIZE + 1 + super::super::content_profile::MAX_FILE_SIZE + 1 ) .await .err() diff --git a/src/snapshot/proven_file.rs b/src/snapshot/proven_file.rs index 6820baf..bc9e70b 100644 --- a/src/snapshot/proven_file.rs +++ b/src/snapshot/proven_file.rs @@ -406,7 +406,7 @@ impl SnapshotReader { if entry.kind == EntryKind::Directory { return Err(not_file("selected path is a directory")); } - if entry.size > super::range::MAX_FILE_SIZE + if entry.size > super::content_profile::MAX_FILE_SIZE || entry.kind == EntryKind::Symlink && !(1..=4095).contains(&entry.size) { return Err(limit().into()); diff --git a/src/snapshot/proven_file_tests.rs b/src/snapshot/proven_file_tests.rs index 02940eb..14475be 100644 --- a/src/snapshot/proven_file_tests.rs +++ b/src/snapshot/proven_file_tests.rs @@ -1184,7 +1184,10 @@ async fn selected_membership_rejects_wrong_kinds_missing_paths_and_serving_profi assert!(server.reader.prove_file("../file000").await.is_err()); assert_eq!(server.fixture.requests.lock().unwrap().len(), before); for (kind, size) in [ - (EntryKind::Regular, super::super::range::MAX_FILE_SIZE + 1), + ( + EntryKind::Regular, + super::super::content_profile::MAX_FILE_SIZE + 1, + ), (EntryKind::Symlink, 0), (EntryKind::Symlink, 4096), ] { diff --git a/src/snapshot/range.rs b/src/snapshot/range.rs deleted file mode 100644 index d0f3ce4..0000000 --- a/src/snapshot/range.rs +++ /dev/null @@ -1,337 +0,0 @@ -//! Range reads over the fixed view (spec 07 §6). -//! -//! A FUSE `read(offset, size)` must fetch only the chunks that cover the -//! requested bytes — reading the first 4 KiB of a 2 GiB file may not -//! download the file. This module owns that path for large files: -//! -//! * the chunk map is fetched once and independently verified -//! (`map_id`, MCM2 field consistency) by the client; -//! * chunk-leaf pages are fetched on demand and checked against -//! `pages_root` with their Merkle proof, so the digests used to verify -//! chunk bytes are themselves authenticated; -//! * each chunk is verified against its leaf digest before it is cached or -//! sliced — a short read never becomes zero-padded kernel data; -//! * the whole-file digest remains the hydration (`DURABLE_COMPLETE`) -//! gate; per-chunk verification is what a range read can prove. -//! -//! Small files (≤ 256 KiB) are served whole through the OBJECT path: the -//! spec's small/large split, not a size threshold invented here. - -use std::{collections::VecDeque, sync::Arc}; - -use mst2_codec::chunkmap::{CHUNKS_PER_PAGE, CHUNK_SIZE}; - -use crate::snapshot::{ - frames::{ChunkRequest, VerifiedChunkMap}, - reader::SnapshotReader, - SnapshotError, SnapshotErrorCode, -}; - -/// Files at or below this size use the OBJECT path (spec 07 §2). -pub const OBJECT_CAP: u64 = 256 * 1024; -pub(crate) const MAX_FILE_SIZE: u64 = 8 * 1024 * 1024 * 1024 * 1024; -const CACHED_CHUNKS: usize = 16; -const CACHED_LEAVES: usize = 16; - -struct BoundedCache { - entries: VecDeque<(u64, Arc>)>, - capacity: usize, -} - -impl BoundedCache { - fn new(capacity: usize) -> Self { - Self { - entries: VecDeque::new(), - capacity, - } - } - - fn get(&mut self, index: u64) -> Option>> { - let slot = self.entries.iter().position(|(key, _)| *key == index)?; - let entry = self.entries.remove(slot)?; - let value = Arc::clone(&entry.1); - self.entries.push_back(entry); - Some(value) - } - - fn insert(&mut self, index: u64, value: Arc>) { - if let Some(slot) = self.entries.iter().position(|(key, _)| *key == index) { - self.entries.remove(slot); - } - if self.entries.len() == self.capacity { - self.entries.pop_front(); - } - self.entries.push_back((index, value)); - } -} - -/// A large file read through its chunk map, with verified per-chunk caching. -pub struct ChunkedFile { - reader: SnapshotReader, - path: String, - digest: String, - pub size: u64, - map: VerifiedChunkMap, - /// Leaf digests per chunk-map page, verified against `pages_root`. - leaves: tokio::sync::Mutex>, - /// Verified chunk bytes by index. - chunks: tokio::sync::Mutex>, -} - -impl ChunkedFile { - /// Fetch and verify the chunk map for one path in the fixed view. - pub async fn open( - reader: &SnapshotReader, - path: &str, - digest: &str, - size: u64, - ) -> Result { - reader.authorized_context().validate_relative_path(path)?; - reader.client().validate_file_size(size)?; - if size > MAX_FILE_SIZE { - return Err(SnapshotError::new( - SnapshotErrorCode::LimitExceeded, - "file size exceeds the 8 TiB serving profile", - )); - } - if size <= OBJECT_CAP { - return Err(SnapshotError::new( - SnapshotErrorCode::Internal, - "small files are read whole, not through the chunk map", - )); - } - reader.ensure_lease().await?; - let request_path = super::reader::ScopeRequestPath(path).to_string(); - let map = reader - .client - .chunk_map(reader.snapshot_id(), &request_path, digest) - .await?; - if map.file_size != size { - return Err(SnapshotError::new( - SnapshotErrorCode::DigestMismatch, - format!("chunk map size {} != view size {size}", map.file_size), - )); - } - Ok(ChunkedFile { - reader: reader.clone(), - path: request_path, - digest: digest.to_string(), - size, - map, - leaves: tokio::sync::Mutex::new(BoundedCache::new(CACHED_LEAVES)), - chunks: tokio::sync::Mutex::new(BoundedCache::new(CACHED_CHUNKS)), - }) - } - - pub fn map_id(&self) -> &str { - &self.map.map_id - } - - /// Bytes `[offset, offset+length)` of the file, clamped to EOF, with - /// every covering chunk verified before it is sliced. - /// This legacy Vec/cache path is caller-owned and outside owned byte quotas. - pub async fn read_range(&self, offset: u64, length: u64) -> Result, SnapshotError> { - if length == 0 || offset >= self.size { - return Ok(Vec::new()); - } - let end = offset.saturating_add(length).min(self.size); - let returned = end - offset; - if returned > crate::snapshot::client::MAX_BUFFERED_FILE_BYTES { - return Err(SnapshotError::new( - SnapshotErrorCode::LimitExceeded, - "range output exceeds the local 64 MiB budget; request a smaller range", - )); - } - let returned = usize::try_from(returned).map_err(|_| range_allocation_error())?; - self.reader - .authorized_context() - .validate_relative_path(&self.path)?; - self.reader.ensure_lease().await?; - let (start_chunk, end_chunk) = - mst2_codec::chunkmap::range_chunks(offset, end - offset, self.size) - .map_err(codec_err)?; - - let mut out = Vec::new(); - out.try_reserve_exact(returned) - .map_err(|_| range_allocation_error())?; - for index in start_chunk..=end_chunk { - // Keep this verified chunk alive while copying, even if another - // read evicts it. Output need not fit inside the retained cache. - let bytes = self.ensure_chunk(index).await?; - let chunk_start = index * CHUNK_SIZE as u64; - let from = if index == start_chunk { - (offset - chunk_start) as usize - } else { - 0 - }; - let to = if index == end_chunk { - (end - chunk_start) as usize - } else { - bytes.len() - }; - if from > to || to > bytes.len() { - return Err(SnapshotError::new( - SnapshotErrorCode::Internal, - format!( - "chunk {index}: slice {from}..{to} outside {} bytes", - bytes.len() - ), - )); - } - out.extend_from_slice(&bytes[from..to]); - } - Ok(out) - } - - /// True when every chunk of the file is already cached and verified. - pub async fn fully_cached(&self) -> bool { - self.chunks.lock().await.entries.len() as u64 == self.map.chunk_count - } - - /// Make chunk `index` available and verified, fetching the leaf page - /// that authenticates it first when needed. - async fn ensure_chunk(&self, index: u64) -> Result>, SnapshotError> { - if let Some(bytes) = self.chunks.lock().await.get(index) { - return Ok(bytes); - } - let page = index / CHUNKS_PER_PAGE as u64; - let digests = self.ensure_leaf_page(page).await?; - let slot = (index % CHUNKS_PER_PAGE as u64) as usize; - let want = digests.get(slot).copied().ok_or_else(|| { - SnapshotError::new( - SnapshotErrorCode::DigestMismatch, - format!("leaf page {page} has no digest for chunk {index}"), - ) - })?; - - self.reader - .authorized_context() - .validate_relative_path(&self.path)?; - self.reader.ensure_lease().await?; - let units = self - .reader - .client - .chunks( - self.reader.snapshot_id(), - &[ChunkRequest { - path: self.path.clone(), - expected_digest: self.digest.clone(), - map_id: self.map.map_id.clone(), - chunk_index: index, - }], - self.reader.encoding_hint(), - ) - .await?; - let unit = units - .into_iter() - .find(|u| u.chunk_index == index) - .ok_or_else(|| { - SnapshotError::new( - SnapshotErrorCode::DigestMismatch, - format!("no chunk {index} in the response"), - ) - })?; - // Independent verification: the bytes must hash to the digest the - // authenticated leaf advertises, and match the map's length. - let got = crate::snapshot::durable::digest_of(&unit.bytes); - if got != format!("sha256:{}", crate::snapshot::frames::hex32(&want)) { - return Err(SnapshotError::new( - SnapshotErrorCode::DigestMismatch, - format!("chunk {index} does not match its leaf digest"), - )); - } - let expected_len = self.map.chunk_len(index)?; - if unit.bytes.len() as u64 != expected_len { - return Err(SnapshotError::new( - SnapshotErrorCode::DigestMismatch, - format!( - "chunk {index} is {} bytes, map says {expected_len}", - unit.bytes.len() - ), - )); - } - let bytes = Arc::new(unit.bytes); - self.chunks.lock().await.insert(index, Arc::clone(&bytes)); - Ok(bytes) - } - - /// Fetch and proof-check one leaf page (cached per page). - async fn ensure_leaf_page(&self, page: u64) -> Result>, SnapshotError> { - if let Some(digests) = self.leaves.lock().await.get(page) { - return Ok(digests); - } - self.reader - .authorized_context() - .validate_relative_path(&self.path)?; - self.reader.ensure_lease().await?; - let leaf = self - .reader - .client - .chunk_map_page( - self.reader.snapshot_id(), - &self.path, - &self.digest, - &self.map, - page, - ) - .await?; - let digests = Arc::new(leaf.chunk_sha256); - self.leaves.lock().await.insert(page, Arc::clone(&digests)); - Ok(digests) - } -} - -impl VerifiedChunkMap { - /// Length of chunk `index` derived from the map (spec 07 §4). - pub fn chunk_len(&self, index: u64) -> Result { - let map = mst2_codec::chunkmap::ChunkMap::new( - crate::snapshot::frames::parse_digest(&self.file_content_id)?, - self.file_size, - self.pages_root, - ) - .map_err(codec_err)?; - map.chunk_len(index).map_err(codec_err) - } -} - -fn codec_err(e: mst2_codec::CodecError) -> SnapshotError { - SnapshotError::new(SnapshotErrorCode::Internal, format!("chunk codec: {e}")) -} - -fn range_allocation_error() -> SnapshotError { - SnapshotError::new( - SnapshotErrorCode::LimitExceeded, - "range output cannot fit in memory", - ) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn object_cap_matches_the_profile_split() { - // Spec 07 §2: ≤256 KiB uses OBJECT frames. - assert_eq!(OBJECT_CAP, 262_144); - } - - #[test] - fn chunk_len_matches_the_map_rules() { - use crate::snapshot::frames::parse_digest; - // 2 chunks + 7 bytes: last chunk is the positive remainder. - let size = 2 * CHUNK_SIZE as u64 + 7; - let id = "sha256:".to_string() + &"0".repeat(64); - let map = VerifiedChunkMap { - file_content_id: id.clone(), - map_id: id, - file_size: size, - chunk_count: 3, - page_count: 1, - pages_root: [0u8; 32], - }; - assert_eq!(map.chunk_len(0).unwrap(), CHUNK_SIZE as u64); - assert_eq!(map.chunk_len(2).unwrap(), 7); - assert!(map.chunk_len(3).is_err()); - assert!(parse_digest(&map.file_content_id).is_ok()); - } -} diff --git a/src/workspace/hydrate.rs b/src/workspace/hydrate.rs index 5e307ab..d85a64a 100644 --- a/src/workspace/hydrate.rs +++ b/src/workspace/hydrate.rs @@ -996,7 +996,7 @@ mod tests { } #[tokio::test] - async fn strict_http_paths_cover_owned_proven_and_compatibility_ranges_then_hydration() { + async fn strict_http_paths_cover_owned_and_proven_ranges_then_hydration() { let server = Server::new(Fixture::new(true, false, true)).await; let reader = server.reader().await; let bytes = &server.fixture.bodies["/large0"]; @@ -1039,19 +1039,16 @@ mod tests { .as_bytes(), second ); - let compatibility_range = - crate::snapshot::ChunkedFile::open(&reader, "large0", &digest, bytes.len() as u64) - .await - .unwrap(); assert_eq!( - compatibility_range - .read_range(2 * CHUNK_SIZE as u64, 7) + range + .read_range_owned(2 * CHUNK_SIZE as u64, 7) .await - .unwrap(), - bytes[2 * CHUNK_SIZE as usize..] + .unwrap() + .as_bytes(), + &bytes[2 * CHUNK_SIZE as usize..] ); - assert_eq!(server.fixture.map_calls.load(Ordering::SeqCst), 3); - assert_eq!(server.fixture.leaf_calls.load(Ordering::SeqCst), 3); + assert_eq!(server.fixture.map_calls.load(Ordering::SeqCst), 2); + assert_eq!(server.fixture.leaf_calls.load(Ordering::SeqCst), 2); assert_eq!(server.fixture.chunk_calls.load(Ordering::SeqCst), 6); let small = reader.prove_file("d0/f000").await.unwrap(); diff --git a/tests/mst2_authorized_paths_http.rs b/tests/mst2_authorized_paths_http.rs index 345b297..d3e6942 100644 --- a/tests/mst2_authorized_paths_http.rs +++ b/tests/mst2_authorized_paths_http.rs @@ -18,9 +18,8 @@ use axum::{ }; use mst2_codec::descriptor::ServingDescriptor; use scorpiofs::snapshot::{ - durable::digest_of, - range::{ChunkedFile, OBJECT_CAP}, - FetchCoordinator, Mst2Client, SnapshotErrorCode, SnapshotFile, SnapshotReader, + durable::digest_of, FetchCoordinator, Mst2Client, OwnedChunkedFile, SnapshotErrorCode, + SnapshotFile, SnapshotReader, OBJECT_CAP, }; use serde_json::{json, Value}; use tokio::{sync::Notify, task::JoinHandle}; @@ -217,7 +216,8 @@ async fn large_file_open_rejects_invalid_composed_paths_before_chunk_map_rpc() { .unwrap(); for (path, expected) in invalid_paths() { let error = - match ChunkedFile::open(&reader, &path, &digest_of(CONTENT), OBJECT_CAP + 1).await { + match OwnedChunkedFile::open(&reader, &path, &digest_of(CONTENT), OBJECT_CAP + 1).await + { Ok(_) => panic!("invalid range-read path was accepted: {path:?}"), Err(error) => error, }; diff --git a/tests/mst2_canonical_reader_http.rs b/tests/mst2_canonical_reader_http.rs index c4a5830..89e2988 100644 --- a/tests/mst2_canonical_reader_http.rs +++ b/tests/mst2_canonical_reader_http.rs @@ -26,8 +26,8 @@ use scorpiofs::snapshot::{ capabilities::CapabilityAdvertisement, durable::digest_of, frames::{parse_digest, MetadataPageItem}, - ChunkedFile, CompletionKind, DurableStore, LocalPinState, Mst2Client, OwnedChunkedFile, - ResolveDelivery, ResolveRequest, ResolveTarget, ScopeCache, SnapshotErrorCode, SnapshotReader, + CompletionKind, DurableStore, LocalPinState, Mst2Client, OwnedChunkedFile, ResolveDelivery, + ResolveRequest, ResolveTarget, ScopeCache, SnapshotErrorCode, SnapshotReader, }; use serde_json::{json, Value}; @@ -537,12 +537,16 @@ async fn canonical_ranges_use_page_index_and_split_actual_chunks_by_advertised_b .unwrap(); let bytes = reader.read_content(file, true).await.unwrap(); assert_eq!(bytes.as_bytes(), s.f.bodies["/large"]); - let range = ChunkedFile::open(&reader, "/large", &file.content_digest, file.size) + let range = OwnedChunkedFile::open(&reader, "/large", &file.content_digest, file.size) .await .unwrap(); assert_eq!( - range.read_range(CHUNK_SIZE as u64 - 2, 6).await.unwrap(), - vec![0x51; 6] + range + .read_range_owned(CHUNK_SIZE as u64 - 2, 6) + .await + .unwrap() + .as_bytes(), + &[0x51; 6] ); assert_eq!( s.calls("chunks") diff --git a/tests/mst2_range_budget_http.rs b/tests/mst2_range_budget_http.rs index e42f6d3..f63ba9b 100644 --- a/tests/mst2_range_budget_http.rs +++ b/tests/mst2_range_budget_http.rs @@ -21,10 +21,12 @@ use mst2_codec::{ leaf_proof, merkle_root, ChunkLeaf, ChunkMap, ProofSide, CHUNKS_PER_PAGE, CHUNK_SIZE, }, descriptor::ServingDescriptor, - treeframe::{ChunkPayload, EndPayload}, + metapage::{page_id, Entry, EntryKind, Page}, + treeframe::{ChunkPayload, EndPayload, MetaPayload}, }; use scorpiofs::snapshot::{ - client::MAX_BUFFERED_FILE_BYTES, ChunkedFile, Mst2Client, SnapshotErrorCode, SnapshotReader, + client::MAX_BUFFERED_FILE_BYTES, Mst2Client, OwnedChunkedFile, SnapshotErrorCode, + SnapshotReader, }; use serde_json::{json, Value}; @@ -72,6 +74,9 @@ fn base64(bytes: &[u8]) -> String { } struct Fixture { + descriptor: ServingDescriptor, + metadata_page: Vec, + metadata_requests: AtomicU64, map: ChunkMap, leaves: Vec, hashes: Vec<[u8; 32]>, @@ -84,17 +89,12 @@ struct Fixture { async fn capabilities() -> Json { Json(json!({ "protocol_versions": [2], "metadata_codecs": [1], "frame_encodings": ["identity"], - "features": {"resolve": true, "directory": true, "leases": true, "chunk_reads": true} + "features": {"resolve": true, "directory": true, "leases": true, "metadata_pages": true, "chunk_reads": true} })) } -async fn resolve() -> Json { - let descriptor = ServingDescriptor { - instance_uuid: *uuid::Uuid::parse_str(INSTANCE).unwrap().as_bytes(), - namespace_view_id: [0x22; 32], - scope: "/project".into(), - metadata_root: [1; 32], - }; +async fn resolve(State(fixture): State>) -> Json { + let descriptor = &fixture.descriptor; Json(json!({ "descriptor": { "schema_version": 2, "metadata_codec": 1, "instance_id": INSTANCE, @@ -107,6 +107,47 @@ async fn resolve() -> Json { })) } +async fn metadata( + State(fixture): State>, + Path(snapshot): Path, + body: Bytes, +) -> Response { + assert_eq!(snapshot, id(&fixture.descriptor.snapshot_id().unwrap())); + let request: Value = serde_json::from_slice(&body).unwrap(); + let items = request["items"].as_array().unwrap(); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["directory_path"], "/"); + assert_eq!(items[0]["route"], json!([])); + assert_eq!( + items[0]["expected_digest"], + id(&fixture.descriptor.metadata_root) + ); + fixture.metadata_requests.fetch_add(1, Ordering::SeqCst); + let mut wire = MetaPayload { + pages: vec![( + fixture.descriptor.metadata_root, + fixture.metadata_page.clone(), + )], + } + .encode(19, 0) + .unwrap(); + wire.extend( + EndPayload { + request_item_count: 1, + unique_unit_count: 1, + logical_bytes: fixture.metadata_page.len() as u64, + request_body_sha256: digest(&body), + } + .encode(19, 1), + ); + Response::builder() + .header("content-type", "application/vnd.mega.treeframe;version=2") + .header("x-mega-snapshot-id", snapshot) + .header("x-mega-request-digest", id(&digest(&body))) + .body(axum::body::Body::from(wire)) + .unwrap() +} + async fn map( State(fixture): State>, Path(snapshot): Path, @@ -201,7 +242,7 @@ impl Drop for Server { } } -async fn open(size: u64) -> (Server, Arc, SnapshotReader, ChunkedFile) { +async fn open(size: u64) -> (Server, Arc, SnapshotReader, OwnedChunkedFile) { let count = size.div_ceil(CHUNK); let full: Vec<_> = (0..4) .map(|i| digest(&vec![pattern(i); CHUNK_SIZE as usize])) @@ -228,7 +269,17 @@ async fn open(size: u64) -> (Server, Arc, SnapshotReader, ChunkedFile) .iter() .map(|leaf| leaf.leaf_hash().unwrap()) .collect(); + let metadata_page = + Page::build(&[Entry::file(EntryKind::Regular, b"file", size, CONTENT)]).unwrap(); let fixture = Arc::new(Fixture { + descriptor: ServingDescriptor { + instance_uuid: *uuid::Uuid::parse_str(INSTANCE).unwrap().as_bytes(), + namespace_view_id: [0x22; 32], + scope: "/project".into(), + metadata_root: page_id(&metadata_page), + }, + metadata_page, + metadata_requests: AtomicU64::new(0), map: ChunkMap::new(CONTENT, size, merkle_root(&hashes).unwrap()).unwrap(), leaves, hashes, @@ -240,6 +291,10 @@ async fn open(size: u64) -> (Server, Arc, SnapshotReader, ChunkedFile) let app = Router::new() .route("/api/v2/snapshots/capabilities", get(capabilities)) .route("/api/v2/snapshots/resolve", post(resolve)) + .route( + "/api/v2/snapshots/{snapshot}/metadata/pages", + post(metadata), + ) .route("/api/v2/snapshots/{snapshot}/chunk-map", get(map)) .route("/api/v2/snapshots/{snapshot}/chunk-map/pages", get(leaf)) .route("/api/v2/snapshots/{snapshot}/chunks", post(chunks)) @@ -252,9 +307,11 @@ async fn open(size: u64) -> (Server, Arc, SnapshotReader, ChunkedFile) let reader = SnapshotReader::resolve(Mst2Client::new(base), "/project", 600) .await .unwrap(); - let file = ChunkedFile::open(&reader, "file", &id(&CONTENT), size) + let file = OwnedChunkedFile::open(&reader, "file", &id(&CONTENT), size) .await .unwrap(); + assert_eq!(file.size(), size); + assert_eq!(fixture.metadata_requests.load(Ordering::SeqCst), 1); (server, fixture, reader, file) } @@ -271,22 +328,22 @@ async fn output_limit_is_checked_before_content_and_after_eof_clamping() { let (_server, fixture, reader, file) = open(size).await; for length in [MAX_BUFFERED_FILE_BYTES + 1, u64::MAX] { assert_eq!( - file.read_range(0, length).await.unwrap_err().code, + file.read_range_owned(0, length).await.unwrap_err().code, SnapshotErrorCode::LimitExceeded ); } assert!(fixture.leaf_requests.lock().unwrap().is_empty()); assert!(fixture.chunk_requests.lock().unwrap().is_empty()); assert!(file - .read_range(u64::MAX, u64::MAX) + .read_range_owned(u64::MAX, u64::MAX) .await .unwrap() .is_empty()); - let tail = file.read_range(size - 3, u64::MAX).await.unwrap(); + let tail = file.read_range_owned(size - 3, u64::MAX).await.unwrap(); assert_eq!(tail.len(), 3); - assert_bytes(&tail, size - 3); + assert_bytes(tail.as_bytes(), size - 3); assert_eq!(fixture.chunk_requests.lock().unwrap().as_slice(), &[80]); - let result = ChunkedFile::open(&reader, "file", &id(&CONTENT), size + 1).await; + let result = OwnedChunkedFile::open(&reader, "file", &id(&CONTENT), size + 1).await; assert_eq!( result.err().unwrap().code, SnapshotErrorCode::DigestMismatch @@ -296,20 +353,20 @@ async fn output_limit_is_checked_before_content_and_after_eof_clamping() { #[tokio::test] async fn output_larger_than_cache_stays_correct_and_evicted_chunk_is_reverified() { let (_server, fixture, _reader, file) = open(80 * CHUNK).await; - let bytes = file.read_range(0, 20 * CHUNK).await.unwrap(); - assert_bytes(&bytes, 0); + let bytes = file.read_range_owned(0, 20 * CHUNK).await.unwrap(); + assert_bytes(bytes.as_bytes(), 0); assert_eq!(bytes.len() as u64, 20 * CHUNK); assert_eq!(fixture.chunk_requests.lock().unwrap().len(), 20); - file.read_range(19 * CHUNK, 7).await.unwrap(); + file.read_range_owned(19 * CHUNK, 7).await.unwrap(); assert_eq!(fixture.chunk_requests.lock().unwrap().len(), 20); assert!(!file.fully_cached().await); fixture.corrupt_chunk.store(0, Ordering::SeqCst); assert_eq!( - file.read_range(0, 7).await.unwrap_err().code, + file.read_range_owned(0, 7).await.unwrap_err().code, SnapshotErrorCode::DigestMismatch ); fixture.corrupt_chunk.store(NO_CORRUPTION, Ordering::SeqCst); - assert_bytes(&file.read_range(0, 7).await.unwrap(), 0); + assert_bytes(file.read_range_owned(0, 7).await.unwrap().as_bytes(), 0); assert_eq!( fixture .chunk_requests @@ -329,22 +386,31 @@ async fn leaf_cache_refreshes_hits_and_rechecks_proof_after_eviction() { let (_server, fixture, _reader, file) = open(17 * page_bytes).await; for page in 0..16 { assert_bytes( - &file.read_range(page * page_bytes, 1).await.unwrap(), + file.read_range_owned(page * page_bytes, 1) + .await + .unwrap() + .as_bytes(), page * page_bytes, ); } // A second chunk on page zero refreshes that leaf without a leaf HTTP call. - file.read_range(CHUNK, 1).await.unwrap(); - file.read_range(16 * page_bytes, 1).await.unwrap(); + file.read_range_owned(CHUNK, 1).await.unwrap(); + file.read_range_owned(16 * page_bytes, 1).await.unwrap(); assert_eq!(fixture.leaf_requests.lock().unwrap().len(), 17); // Page one was least recently used; both it and its old chunk were evicted. fixture.corrupt_leaf.store(1, Ordering::SeqCst); assert_eq!( - file.read_range(page_bytes, 1).await.unwrap_err().code, + file.read_range_owned(page_bytes, 1).await.unwrap_err().code, SnapshotErrorCode::DigestMismatch ); fixture.corrupt_leaf.store(NO_CORRUPTION, Ordering::SeqCst); - assert_bytes(&file.read_range(page_bytes, 1).await.unwrap(), page_bytes); + assert_bytes( + file.read_range_owned(page_bytes, 1) + .await + .unwrap() + .as_bytes(), + page_bytes, + ); assert_eq!( fixture .leaf_requests @@ -363,9 +429,9 @@ async fn concurrent_reads_remain_correct_while_evicting_each_others_chunks() { let file = &file; let reads = (0..8).map(|i| async move { let offset = i * 8 * CHUNK + 13; - let bytes = file.read_range(offset, 3 * CHUNK + 7).await.unwrap(); + let bytes = file.read_range_owned(offset, 3 * CHUNK + 7).await.unwrap(); assert_eq!(bytes.len() as u64, 3 * CHUNK + 7); - assert_bytes(&bytes, offset); + assert_bytes(bytes.as_bytes(), offset); }); tokio::time::timeout( std::time::Duration::from_secs(30), @@ -377,7 +443,7 @@ async fn concurrent_reads_remain_correct_while_evicting_each_others_chunks() { // At most sixteen verified chunks remain, observable by refetching an old one. let before = fixture.chunk_requests.lock().unwrap().len(); for index in (0..8).map(|i| i * 8) { - file.read_range(index * CHUNK, 1).await.unwrap(); + file.read_range_owned(index * CHUNK, 1).await.unwrap(); } assert!(fixture.chunk_requests.lock().unwrap().len() > before); } diff --git a/tests/mst2_range_e2e.rs b/tests/mst2_range_e2e.rs index 25c0729..bff226d 100644 --- a/tests/mst2_range_e2e.rs +++ b/tests/mst2_range_e2e.rs @@ -8,7 +8,7 @@ //! Ignored by default; requires the running stack with the t08 fixture //! (`t08/chunked.bin`, 2 MiB + 7 bytes, byte i = (73i+11) % 256). -use scorpiofs::snapshot::{ChunkedFile, Mst2Client, SnapshotReader}; +use scorpiofs::snapshot::{Mst2Client, OwnedChunkedFile, SnapshotReader}; const CHUNK: u64 = 1 << 20; @@ -30,22 +30,18 @@ async fn range_read_transfers_only_the_covering_chunks() { let reader = SnapshotReader::resolve(client.clone(), "/project", 600) .await .expect("resolve"); - let manifest = reader.file_manifest().await.expect("manifest"); - let large = manifest - .iter() - .find(|f| f.rel_path == "t08/chunked.bin") - .expect("t08/chunked.bin seeded by the t08 oracle") - .clone(); - assert!(large.size > 2 * CHUNK, "fixture must exceed two chunks"); + let large = reader + .prove_file("/t08/chunked.bin") + .await + .expect("t08/chunked.bin has fixed-root membership in the t08 oracle"); + assert!( + large.file().size > 2 * CHUNK, + "fixture must exceed two chunks" + ); - let file = ChunkedFile::open( - &reader, - "/t08/chunked.bin", - &large.content_digest, - large.size, - ) - .await - .expect("chunk map"); + let file = OwnedChunkedFile::open_proven(&reader, large) + .await + .expect("chunk map"); // A 4 KiB read in the middle of chunk 1 fetches exactly one chunk — // counted in units, because the fixture's periodic byte pattern makes @@ -53,9 +49,12 @@ async fn range_read_transfers_only_the_covering_chunks() { let offset = CHUNK + 4096; let before_units = client.units_fetched(); let before_bytes = client.received_bytes(); - let bytes = file.read_range(offset, 4096).await.expect("range read"); + let bytes = file + .read_range_owned(offset, 4096) + .await + .expect("range read"); assert_eq!(bytes.len(), 4096); - for (i, b) in bytes.iter().enumerate() { + for (i, b) in bytes.as_bytes().iter().enumerate() { assert_eq!(*b, pattern_at(offset as usize + i), "byte {i} mismatch"); } assert_eq!( @@ -64,24 +63,30 @@ async fn range_read_transfers_only_the_covering_chunks() { "a 4 KiB read must fetch one chunk, not the file" ); assert!( - client.received_bytes() - before_bytes < large.size, + client.received_bytes() - before_bytes < file.size(), "a 4 KiB read must not transfer the whole {} byte file", - large.size + file.size() ); // The chunk is cached: the same range again costs nothing. let before = client.units_fetched(); - let again = file.read_range(offset, 4096).await.expect("cached read"); - assert_eq!(again, bytes); + let again = file + .read_range_owned(offset, 4096) + .await + .expect("cached read"); + assert_eq!(again.as_bytes(), bytes.as_bytes()); assert_eq!(client.units_fetched() - before, 0, "cached chunk refetched"); // Crossing a boundary fetches exactly the one chunk that was still // missing (chunk 1 is cached by the read above). let offset = CHUNK - 2048; let before = client.units_fetched(); - let bytes = file.read_range(offset, 4096).await.expect("boundary read"); + let bytes = file + .read_range_owned(offset, 4096) + .await + .expect("boundary read"); assert_eq!(bytes.len(), 4096); - for (i, b) in bytes.iter().enumerate() { + for (i, b) in bytes.as_bytes().iter().enumerate() { assert_eq!(*b, pattern_at(offset as usize + i)); } assert_eq!( @@ -91,10 +96,16 @@ async fn range_read_transfers_only_the_covering_chunks() { ); // The last (short) chunk is served exactly; a read past EOF clamps. - let tail_off = large.size - 7; - let tail = file.read_range(tail_off, 4096).await.expect("tail read"); + let tail_off = file.size() - 7; + let tail = file + .read_range_owned(tail_off, 4096) + .await + .expect("tail read"); assert_eq!(tail.len(), 7); - let past = file.read_range(large.size, 4096).await.expect("past EOF"); + let past = file + .read_range_owned(file.size(), 4096) + .await + .expect("past EOF"); assert!(past.is_empty()); - assert!(file.read_range(0, 0).await.unwrap().is_empty()); + assert!(file.read_range_owned(0, 0).await.unwrap().is_empty()); }