Corrupt footer CRC must rebuild, not refuse the namespace

crates/engine/src/log/mod.rs

@@ -3012,6 +3012,60 @@ mod enospc_recovery_tests {
     }
 }
 
+#[cfg(test)]
+mod footer_crc_rebuild_tests {
+    use super::*;
+    use crate::log::config::LogConfig;
+    use crate::log::file::{FOOTER_TRAILER_LEN, data_filename};
+    use bytes::Bytes;
+    use tempfile::TempDir;
+
+    fn run<F: std::future::Future>(f: F) -> F::Output {
+        monoio::RuntimeBuilder::<monoio::FusionDriver>::new()
+            .enable_timer()
+            .build()
+            .expect("monoio runtime")
+            .block_on(f)
+    }
+
+    /// ARCHITECTURE.md: sealed footer CRC fail → rebuild from records.
+    /// `read_footer().await?` currently refuses the whole namespace.
+    #[test]
+    fn footer_crc_mismatch_falls_back_to_rebuild() {
+        run(async {
+            let dir = TempDir::new().unwrap();
+            let path = dir.path().to_path_buf();
+            let cfg = LogConfig {
+                rotate_threshold: 1 << 40,
+                fanout: 8,
+                value_sep_threshold: 1 << 20,
+            };
+            {
+                let log = NamespaceLog::open(path.clone(), cfg).await.unwrap();
+                log.put_full(Bytes::from_static(b"k"), b"v", &[], None)
+                    .await
+                    .unwrap();
+                log.seal_active_for_shutdown().await.unwrap();
+            }
+            let f0 = path.join(data_filename(0));
+            let mut bytes = std::fs::read(&f0).unwrap();
+            assert!(bytes.len() as u64 >= FOOTER_TRAILER_LEN);
+            // Trailer: [body_len 8][crc 8][magic 8]. Flip CRC, keep magic.
+            let crc_at = bytes.len() - 16;
+            bytes[crc_at] ^= 0x01;
+            std::fs::write(&f0, &bytes).unwrap();
+
+            let log = NamespaceLog::open(path, cfg)
+                .await
+                .expect("corrupt footer CRC must rebuild, not refuse the namespace");
+            assert!(
+                log.index.borrow().get(b"k").is_some(),
+                "rebuild from records must recover the key"
+            );
+        });
+    }
+}
+
 #[cfg(test)]
 mod fd_footprint_tests {
     use super::*;

crates/engine/src/log/recover.rs

@@ -54,11 +54,11 @@ pub async fn open_namespace(dir: PathBuf) -> Result<OpenedFiles> {
 
     for (file_id, path) in data_files {
         let file = LogFile::open_ro(path, file_id).await?;
-        match file.read_footer().await? {
-            Some(entries) => {
+        match classify_footer(&file).await? {
+            FooterRead::Ok(entries) => {
                 apply_footer_entries(&mut index, file_id, &entries);
             }
-            None => {
+            FooterRead::Missing | FooterRead::Corrupt => {
                 warn!(
                     file_id,
                     "sealed file footer missing or corrupt; rebuilding from records \
@@ -72,34 +72,25 @@ pub async fn open_namespace(dir: PathBuf) -> Result<OpenedFiles> {
 
     // Check if the highest-id file was cleanly sealed on shutdown (footer present).
     // If so, load it as a sealed file and open a fresh empty active file.
+    // Footer CRC/range errors are bitrot of a completed trailer (ARCHITECTURE.md
+    // fallback), not a torn active write — rebuild and rotate, don't refuse open.
     let highest = LogFile::open_ro(active_path.clone(), active_id).await?;
-    let active = match highest.read_footer().await? {
-        Some(entries) => {
+    let active = match classify_footer(&highest).await? {
+        FooterRead::Ok(entries) => {
             apply_footer_entries(&mut index, active_id, &entries);
             sealed.push(highest);
-            let next_id = active_id.checked_add(1).ok_or(EngineError::BadRecord {
-                offset: 0,
-                reason: "file_id overflow on clean-shutdown recovery",
-            })?;
-            if next_id >= u32::MAX - 100 {
-                warn!(
-                    file_id = next_id,
-                    remaining = u32::MAX - next_id,
-                    "file_id nearing u32::MAX; compact sealed files to reclaim IDs"
-                );
-            }
-            let new_path = active_path
-                .parent()
-                .ok_or(EngineError::BadRecord {
-                    offset: 0,
-                    reason: "namespace data_dir has no parent; cannot compute next-file path",
-                })?
-                .join(crate::log::file::data_filename(next_id));
-            let active = LogFile::open_rw(new_path, next_id).await?;
-            crate::log::file::sync_dir(&dir).await; // new active after clean-shutdown recovery
-            active
+            open_next_active(&dir, &active_path, active_id).await?
         }
-        None => {
+        FooterRead::Corrupt => {
+            warn!(
+                file_id = active_id,
+                "highest file footer corrupt; rebuilding from records and rotating"
+            );
+            rebuild_from_records(&highest, active_id, &mut index).await?;
+            sealed.push(highest);
+            open_next_active(&dir, &active_path, active_id).await?
+        }
+        FooterRead::Missing => {
             drop(highest);
             let active = LogFile::open_rw(active_path, active_id).await?;
             replay_active(&active, active_id, &mut index).await?;
@@ -114,6 +105,51 @@ pub async fn open_namespace(dir: PathBuf) -> Result<OpenedFiles> {
     })
 }
 
+enum FooterRead {
+    Ok(Vec<FooterEntry>),
+    Missing,
+    Corrupt,
+}
+
+async fn classify_footer(file: &LogFile) -> Result<FooterRead> {
+    match file.read_footer().await {
+        Ok(Some(entries)) => Ok(FooterRead::Ok(entries)),
+        Ok(None) => Ok(FooterRead::Missing),
+        Err(EngineError::CrcMismatch { .. }) | Err(EngineError::BadRecord { .. }) => {
+            Ok(FooterRead::Corrupt)
+        }
+        Err(e) => Err(e),
+    }
+}
+
+async fn open_next_active(
+    dir: &std::path::Path,
+    active_path: &std::path::Path,
+    active_id: u32,
+) -> Result<LogFile> {
+    let next_id = active_id.checked_add(1).ok_or(EngineError::BadRecord {
+        offset: 0,
+        reason: "file_id overflow on clean-shutdown recovery",
+    })?;
+    if next_id >= u32::MAX - 100 {
+        warn!(
+            file_id = next_id,
+            remaining = u32::MAX - next_id,
+            "file_id nearing u32::MAX; compact sealed files to reclaim IDs"
+        );
+    }
+    let new_path = active_path
+        .parent()
+        .ok_or(EngineError::BadRecord {
+            offset: 0,
+            reason: "namespace data_dir has no parent; cannot compute next-file path",
+        })?
+        .join(crate::log::file::data_filename(next_id));
+    let active = LogFile::open_rw(new_path, next_id).await?;
+    crate::log::file::sync_dir(dir).await;
+    Ok(active)
+}
+
 fn apply_footer_entries(index: &mut NsIndex, file_id: u32, entries: &[FooterEntry]) {
     for e in entries {
         let entry = IndexEntry::new(file_id, e.record_offset, e.record_size, e.tstamp_ms);