perf(mst2): stream authenticated raw blobs with owned chunk credit

src/api/router/snapshot_content.rs

@@ -27,15 +27,15 @@ use crate::ceres::snapshot::{
 };
 
 pub(super) struct ResolvedFileMetadata {
-    fs_kind: FsKind,
-    oid: String,
+    pub(super) fs_kind: FsKind,
+    pub(super) oid: String,
     pub(super) digest: [u8; 32],
     pub(super) size: u64,
     fact: crate::callisto::mst2_verified_object::Model,
 }
 
 #[allow(clippy::result_large_err)]
-async fn fixed_root_tree<T: crate::ceres::api_service::ApiHandler + ?Sized>(
+pub(super) async fn fixed_root_tree<T: crate::ceres::api_service::ApiHandler + ?Sized>(
     handler: &T,
     oid: &str,
 ) -> Result<git_internal::internal::object::tree::Tree, Response> {
@@ -50,7 +50,7 @@ async fn fixed_root_tree<T: crate::ceres::api_service::ApiHandler + ?Sized>(
 }
 
 #[allow(clippy::result_large_err)]
-async fn resolve_file_metadata<T: crate::ceres::api_service::ApiHandler + ?Sized>(
+pub(super) async fn resolve_file_metadata<T: crate::ceres::api_service::ApiHandler + ?Sized>(
     handler: &T,
     root_tree: &git_internal::internal::object::tree::Tree,
     scope: &str,
@@ -479,7 +479,7 @@ async fn project_for<T: crate::ceres::api_service::ApiHandler + ?Sized>(
 }
 
 #[allow(clippy::result_large_err)]
-async fn project_resolved<T: crate::ceres::api_service::ApiHandler + ?Sized>(
+pub(super) async fn project_resolved<T: crate::ceres::api_service::ApiHandler + ?Sized>(
     handler: &T,
     f: &ResolvedFileMetadata,
 ) -> Result<std::sync::Arc<crate::jupiter::storage::native_chunk_map::PersistedChunkMap>, Response>
@@ -789,7 +789,7 @@ struct ResolvedChunk {
     map_id: String,
 }
 
-fn content_read_error(error: crate::common::errors::MegaError) -> SnapshotError {
+pub(super) fn content_read_error(error: crate::common::errors::MegaError) -> SnapshotError {
     use crate::common::errors::MegaError;
     let code = match &error {
         MegaError::ObjStorageNotFound(_) => SnapshotErrorCode::ObjectUnavailable,

src/api/router/snapshot_content_tests.rs

@@ -88,6 +88,9 @@ mod bounded_chunks;
 #[path = "snapshot_persisted_chunk_map_tests.rs"]
 mod persisted_chunk_maps;
 
+#[path = "snapshot_raw_blob_tests.rs"]
+mod raw_blob;
+
 #[path = "snapshot_session_tests.rs"]
 mod durable_sessions;
 
@@ -111,6 +114,7 @@ struct ReadCounts {
     range: AtomicUsize,
     bytes: AtomicUsize,
     object_fault: std::sync::Mutex<Option<bounded_objects::StreamFault>>,
+    object_size_override: std::sync::Mutex<Option<i64>>,
     chunk_faults: std::sync::Mutex<Vec<bounded_chunks::ChunkFault>>,
     receipt_reads: AtomicUsize,
     receipt_writes: AtomicUsize,
@@ -244,7 +248,10 @@ impl MegaObjectStorage for CountingStorage {
                 },
             ));
         }
-        let (stream, meta) = self.inner.inner.get_stream(key).await?;
+        let (stream, mut meta) = self.inner.inner.get_stream(key).await?;
+        if let Some(size) = *self.counts.object_size_override.lock().unwrap() {
+            meta.size = size;
+        }
         let fault = self.counts.object_fault.lock().unwrap().clone();
         let stream = match fault {
             Some(fault) if fault.oid == key.key => fault.stream(),
@@ -904,7 +911,8 @@ async fn mst2_fixed_head_uses_verified_facts_without_body_reads_and_preserves_ra
         .await
         .unwrap();
     assert_eq!(bytes.as_ref(), fixture.raw);
-    fixture.counts.assert(1, fixture.raw.len());
+    fixture.counts.assert(2, 2 * fixture.raw.len());
+    assert_eq!(fixture.counts.receipt_writes.load(Ordering::SeqCst), 1);
 }
 
 #[tokio::test]

src/api/router/snapshot_objects_bounded_tests.rs

@@ -26,6 +26,14 @@ pub(super) enum FaultKind {
         release: Arc<Notify>,
         drops: Arc<AtomicUsize>,
     },
+    HeldFragment {
+        prefix: Bytes,
+        fragment: Option<Bytes>,
+        entered: Arc<Notify>,
+        release: Arc<Notify>,
+        tail_polls: Arc<AtomicUsize>,
+        drops: Arc<AtomicUsize>,
+    },
 }
 
 struct DropCount(Arc<AtomicUsize>);
@@ -39,6 +47,46 @@ impl Drop for DropCount {
 impl StreamFault {
     pub(super) fn stream(self) -> ObjectByteStream {
         match self.kind {
+            FaultKind::HeldFragment {
+                prefix,
+                fragment,
+                entered,
+                release,
+                tail_polls,
+                drops,
+            } => Box::pin(futures::stream::unfold(
+                (
+                    prefix,
+                    fragment,
+                    entered,
+                    release,
+                    tail_polls,
+                    DropCount(drops),
+                    0u8,
+                ),
+                |(prefix, fragment, entered, release, tail_polls, owner, turn)| async move {
+                    match turn {
+                        0 => Some((
+                            Ok(prefix.clone()),
+                            (prefix, fragment, entered, release, tail_polls, owner, 1),
+                        )),
+                        1 => {
+                            entered.notify_one();
+                            release.notified().await;
+                            fragment.clone().map(|part| {
+                                (
+                                    Ok(part),
+                                    (prefix, fragment, entered, release, tail_polls, owner, 2),
+                                )
+                            })
+                        }
+                        _ => {
+                            tail_polls.fetch_add(1, Ordering::SeqCst);
+                            std::future::pending().await
+                        }
+                    }
+                },
+            )),
             FaultKind::HeldThenError {
                 entered,
                 release,