Raise error in manifest_files field is inconsistent with actual manifests listed in arrays (#1989)

Fix #1926

---------

Co-authored-by: Deepak Cherian <dcherian@users.noreply.github.com>

Author: li-em

icechunk/src/session.rs

@@ -170,6 +170,10 @@ pub enum SessionErrorKind {
     NoAmendForInitialCommit,
     #[error("failed to create manifest from chunk stream")]
     ManifestCreationError(#[from] Box<SessionError>),
+    #[error(
+        "inconsistent manifests detected: Snapshot will reference {snapshot} manifest, while array nodes will reference {nodes} manifests"
+    )]
+    ManifestsInconsistencyError { snapshot: usize, nodes: usize },
     #[error("failed to merge sessions: {0}")]
     SessionMerge(String),
     #[error("byte range {request:?} is out of bounds for chunk of length {chunk_length}")]
@@ -2851,15 +2855,22 @@ async fn do_flush(
     }
 
     // manifest_files & manifest_refs _must_ be consistent
-    debug_assert_eq!(
-        flush_data.manifest_files.iter().map(|x| x.id.clone()).collect::<HashSet<_>>(),
-        flush_data
-            .manifest_refs
-            .values()
-            .flatten()
-            .map(|x| x.object_id.clone())
-            .collect::<HashSet<_>>(),
-    );
+    let mfiles =
+        flush_data.manifest_files.iter().map(|x| x.id.clone()).collect::<HashSet<_>>();
+    let mrefs = flush_data
+        .manifest_refs
+        .values()
+        .flatten()
+        .map(|x| x.object_id.clone())
+        .collect::<HashSet<_>>();
+    if mfiles != mrefs {
+        return Err(SessionError::capture(
+            SessionErrorKind::ManifestsInconsistencyError {
+                snapshot: mfiles.len(),
+                nodes: mrefs.len(),
+            },
+        ));
+    }
 
     trace!("Building new snapshot");
     // gather and sort nodes:
@@ -6660,6 +6671,53 @@ mod tests {
         Ok(())
     }
 
+    #[tokio_test]
+    async fn manifest_files_consistency() -> Result<(), Box<dyn Error>> {
+        let repo = Arc::new(create_memory_store_repository(SpecVersionBin::V2).await);
+        let mut session = repo.writable_session("main").await?;
+        session
+            .add_array("/array".try_into().unwrap(), basic_shape(), None, Bytes::new())
+            .await?;
+
+        let properties = Default::default();
+        let manifest_config = ManifestConfig::default();
+        let mut flush_data = FlushProcess::new(
+            Arc::clone(&session.asset_manager),
+            &session.change_set,
+            &Snapshot::INITIAL_SNAPSHOT_ID,
+            &manifest_config,
+        );
+
+        // poke at flush_data to make inconsistent
+        flush_data.manifest_files.insert(ManifestFileInfo {
+            id: ManifestId::random(),
+            num_chunk_refs: 0,
+            size_bytes: 9000,
+        });
+
+        let res = do_flush(
+            flush_data,
+            "fail",
+            1,
+            properties,
+            false,
+            CommitMethod::NewCommit,
+            manifest_config.splitting(),
+        )
+        .await;
+
+        // verify it returns the right error
+        assert!(matches!(
+            res,
+            Err(SessionError {
+                kind: SessionErrorKind::ManifestsInconsistencyError { snapshot, nodes },
+                ..
+            }) if snapshot == 1 && nodes == 0
+        ));
+
+        Ok(())
+    }
+
     #[test]
     fn test_construct_valid_byte_range() {
         // chunk at offset 100, length 50 → valid absolute range is [100, 150]