refactor(snapshot): retire the shared unaccounted range reader

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,
 };
 

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,
 };
 

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 {

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);
+    }
+}

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},
 };
 

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<super::OwnedChunkedFile, SnapshotError> {
+        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<HydrationRangeSource<'_>>,
     ) -> Result<HydrateReport, SnapshotError>
     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<B> = 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<BLarge> = 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<F, Fut, B>(
+async fn write_reader_blob_stream<F, Fut>(
     dir: &Path,
     file: &SnapshotFile,
     mut read_range: F,
 ) -> Result<(), SnapshotError>
 where
     F: FnMut(u64, u64) -> Fut,
-    Fut: Future<Output = Result<B, SnapshotError>>,
-    B: BlobBytes,
+    Fut: Future<Output = Result<Arc<super::VerifiedRange>, 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<VerifiedRange>` cannot implement `AsRef<[u8]>` through the standard
-/// library's `AsRef<T>` 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<u8> {
-    fn bytes(&self) -> &[u8] {
-        self.as_slice()
-    }
-}
-
-impl BlobBytes for Arc<crate::snapshot::VerifiedRange> {
-    fn bytes(&self) -> &[u8] {
-        self.as_bytes()
-    }
-}
-
 struct PendingBlob(PathBuf);
 
 impl Drop for PendingBlob {