Stabilize result ordering and streaming evidence

crates/remora-result/tests/substrate.rs

@@ -7,8 +7,10 @@ use remora_result::{
     MANIFEST_SCHEMA_VERSION, PageSize, ResultError, ResultInventory, ResultMember,
     migrate as result_migrate,
 };
+use remora_stream::TargetBody;
 use remora_target::{
-    ContentSha256, CreateObjectRequest, ObjectIdentity, TargetStore, TargetStoreConfig, hash_bytes,
+    ContentSha256, CreateObjectRequest, ObjectIdentity, StoredObject, TargetStore,
+    TargetStoreConfig, hash_bytes,
 };
 use serde_json::Value;
 use sqlx::PgPool;
@@ -223,47 +225,38 @@ async fn integrity_stale_attempt_and_descriptor_version_fail_closed() {
 
 #[tokio::test]
 async fn pages_large_ordered_inventory_without_materializing_all_rows() {
-    let Some(pool) = test_pool().await else {
+    let (Some(pool), Some(target)) = (test_pool().await, target_store()) else {
         return;
     };
-    let inventory = ResultInventory::new(pool.clone());
+    let inventory = ResultInventory::new(pool);
     let attempt = unique_id("attempt");
     inventory.create(&attempt).await.unwrap();
-    let digest = hash_bytes(b"");
     for index in (0..2_500_u32).rev() {
         let member_id = OpaqueId::parse(format!("member-{index:08}")).unwrap();
+        let stored = target
+            .put_create_only(
+                &CreateObjectRequest::for_attempt_object(&attempt, &member_id, None),
+                Vec::new(),
+            )
+            .await
+            .unwrap();
         let member = ResultMember::new(
             member_id,
             &format!("inventory/{index:08}"),
-            0,
-            member_identity(&attempt, &unique_id("object")),
-            digest.clone(),
+            stored.size,
+            stored.identity,
+            stored.content_sha256,
         )
         .unwrap();
         inventory.add_member(&attempt, &member).await.unwrap();
     }
-    inventory.close(&attempt).await.unwrap();
-    let mut committed = placeholder_publication(&attempt);
-    committed.member_count = 2_500;
-    sqlx::query(
-        "UPDATE result_inventories SET
-            state = 'sealed', manifest_schema_version = $2, manifest_id = $3,
-            manifest_store = $4, manifest_bucket = $5, manifest_key = $6,
-            manifest_version_id = $7, manifest_sha256 = $8, manifest_size = $9
-         WHERE attempt_id = $1",
-    )
-    .bind(attempt.as_str())
-    .bind(committed.schema_version as i32)
-    .bind(committed.manifest_id.as_str())
-    .bind(&committed.manifest.identity.target_store_id)
-    .bind(&committed.manifest.identity.bucket)
-    .bind(&committed.manifest.identity.key)
-    .bind(&committed.manifest.identity.version_id)
-    .bind(committed.manifest.content_sha256.as_str())
-    .bind(committed.manifest.size as i64)
-    .execute(&pool)
-    .await
-    .unwrap();
+    let (committed, metrics) = inventory
+        .create_manifest_with_metrics(&attempt, &unique_id("manifest"), &target)
+        .await
+        .unwrap();
+    assert_eq!(metrics.members_streamed, 2_500);
+    assert!(metrics.max_encoded_member_bytes < 1_024);
+    assert_eq!(committed.member_count, 2_500);
 
     let mut after = None;
     let mut observed = 0_u32;
@@ -284,6 +277,87 @@ async fn pages_large_ordered_inventory_without_materializing_all_rows() {
     assert_eq!(observed, 2_500);
 }
 
+#[tokio::test]
+async fn binary_path_order_is_identical_in_manifest_and_pages() {
+    let (Some(pool), Some(target)) = (test_pool().await, target_store()) else {
+        return;
+    };
+    let inventory = ResultInventory::new(pool);
+    let attempt = unique_id("attempt");
+    inventory.create(&attempt).await.unwrap();
+    for path in ["한/file", "é/file", "a/file", "Z/file"] {
+        let member = stored_member(&target, &attempt, path, b"").await;
+        inventory.add_member(&attempt, &member).await.unwrap();
+    }
+    let committed = inventory
+        .create_manifest(&attempt, &unique_id("manifest"), &target)
+        .await
+        .unwrap();
+    let manifest: Value = serde_json::from_slice(
+        &target
+            .read_exact(&committed.manifest.identity)
+            .await
+            .unwrap(),
+    )
+    .unwrap();
+    let manifest_paths = manifest["members"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|member| member["path"].as_str().unwrap())
+        .collect::<Vec<_>>();
+    assert_eq!(manifest_paths, ["Z/file", "a/file", "é/file", "한/file"]);
+    let page = inventory
+        .page_members(&committed, None, PageSize::new(10).unwrap())
+        .await
+        .unwrap();
+    let page_paths = page
+        .items
+        .iter()
+        .map(ResultMember::path)
+        .collect::<Vec<_>>();
+    assert_eq!(page_paths, manifest_paths);
+}
+
+#[tokio::test]
+async fn large_member_integrity_validation_keeps_manifest_allocation_member_bounded() {
+    let (Some(pool), Some(target)) = (test_pool().await, target_store()) else {
+        return;
+    };
+    let inventory = ResultInventory::new(pool);
+    let attempt = unique_id("attempt");
+    let member_id = unique_id("member");
+    inventory.create(&attempt).await.unwrap();
+    let request = CreateObjectRequest::for_attempt_object(&attempt, &member_id, None);
+    let mut body = target.open_create_only(&request).await.unwrap();
+    let chunk = vec![0x5a; 1024 * 1024];
+    for _ in 0..32 {
+        body.write_all(&chunk).await.unwrap();
+    }
+    let written = body.finish_object().await.unwrap();
+    let stored = StoredObject {
+        identity: written.identity,
+        content_sha256: written.content_sha256,
+        size: written.bytes_written,
+    };
+    let member = ResultMember::new(
+        member_id,
+        "large/member.bin",
+        stored.size,
+        stored.identity,
+        stored.content_sha256,
+    )
+    .unwrap();
+    inventory.add_member(&attempt, &member).await.unwrap();
+    let (committed, metrics) = inventory
+        .create_manifest_with_metrics(&attempt, &unique_id("manifest"), &target)
+        .await
+        .unwrap();
+    assert_eq!(committed.total_bytes, 32 * 1024 * 1024);
+    assert_eq!(metrics.members_streamed, 1);
+    assert!(metrics.max_encoded_member_bytes < 1_024);
+}
+
 fn member_identity(attempt: &OpaqueId, object: &OpaqueId) -> ObjectIdentity {
     ObjectIdentity {
         target_store_id: "fixture-store".into(),