Fix WAL recovery test and improve error handling

- Enable test_recovery by setting WALRUS_DATA_DIR env var
- Use test_helpers for proper schema-compatible test batches
- Add #[serial] to prevent test isolation issues
- Improve error handling in wal.rs persist_topic()
- Remove explicit shutdown to avoid premature WAL consumption

Author: tonyalaribe

src/buffered_write_layer.rs

@@ -560,8 +560,8 @@ impl BufferedWriteLayer {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use arrow::array::{Int64Array, StringViewArray};
-    use arrow::datatypes::{DataType, Field, Schema};
+    use crate::test_utils::test_helpers::{json_to_batch, test_span};
+    use serial_test::serial;
     use std::path::PathBuf;
     use tempfile::tempdir;
 
@@ -571,14 +571,14 @@ mod tests {
         Arc::new(cfg)
     }
 
-    fn create_test_batch() -> RecordBatch {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("id", DataType::Int64, false),
-            Field::new("name", DataType::Utf8View, false),
-        ]));
-        let id_array = Int64Array::from(vec![1, 2, 3]);
-        let name_array = StringViewArray::from(vec!["a", "b", "c"]);
-        RecordBatch::try_new(schema, vec![Arc::new(id_array), Arc::new(name_array)]).unwrap()
+    fn create_test_batch(project_id: &str) -> RecordBatch {
+        // Use test_span helper which creates data matching the default schema
+        json_to_batch(vec![
+            test_span("test1", "span1", project_id),
+            test_span("test2", "span2", project_id),
+            test_span("test3", "span3", project_id),
+        ])
+        .unwrap()
     }
 
     #[tokio::test]
@@ -592,7 +592,7 @@ mod tests {
         let table = format!("t{}", test_id);
 
         let layer = BufferedWriteLayer::with_config(cfg).unwrap();
-        let batch = create_test_batch();
+        let batch = create_test_batch(&project);
 
         layer.insert(&project, &table, vec![batch.clone()]).await.unwrap();
 
@@ -601,15 +601,16 @@ mod tests {
         assert_eq!(results[0].num_rows(), 3);
     }
 
-    // NOTE: This test is ignored because walrus-rust creates new files for each instance
-    // rather than discovering existing files from previous instances in the same directory.
-    // This is a limitation of the walrus library, not our code.
-    #[ignore]
+    #[serial]
     #[tokio::test]
     async fn test_recovery() {
         let dir = tempdir().unwrap();
         let cfg = create_test_config(dir.path().to_path_buf());
 
+        // SAFETY: walrus-rust reads WALRUS_DATA_DIR from environment. We use #[serial]
+        // to prevent concurrent access to this process-global state.
+        unsafe { std::env::set_var("WALRUS_DATA_DIR", &cfg.core.walrus_data_dir) };
+
         // Use unique but short project/table names (walrus has metadata size limit)
         let test_id = &uuid::Uuid::new_v4().to_string()[..4];
         let project = format!("r{}", test_id);
@@ -618,10 +619,9 @@ mod tests {
         // First instance - write data
         {
             let layer = BufferedWriteLayer::with_config(Arc::clone(&cfg)).unwrap();
-            let batch = create_test_batch();
+            let batch = create_test_batch(&project);
             layer.insert(&project, &table, vec![batch]).await.unwrap();
-            // Shutdown to ensure WAL is synced
-            layer.shutdown().await.unwrap();
+            // Layer drops here - WAL data should be persisted
         }
 
         // Second instance - recover from WAL
@@ -648,7 +648,7 @@ mod tests {
         let layer = BufferedWriteLayer::with_config(cfg).unwrap();
 
         // First insert should succeed
-        let batch = create_test_batch();
+        let batch = create_test_batch(&project);
         layer.insert(&project, &table, vec![batch]).await.unwrap();
 
         // Verify reservation is released (should be 0 after successful insert)

src/database.rs

@@ -1736,7 +1736,9 @@ impl ProjectRoutingTable {
 
         // Determine target schema based on projection
         let target_schema = match projection {
-            Some(proj) => Arc::new(arrow_schema::Schema::new(proj.iter().map(|&idx| self.schema.field(idx).clone()).collect::<Vec<_>>())),
+            Some(proj) => Arc::new(arrow_schema::Schema::new(
+                proj.iter().map(|&idx| self.schema.field(idx).clone()).collect::<Vec<_>>(),
+            )),
             None => self.schema.clone(),
         };
 

src/wal.rs

@@ -210,10 +210,18 @@ impl WalManager {
     fn persist_topic(&self, topic: &str) {
         if self.known_topics.insert(topic.to_string()) {
             let meta_dir = self.data_dir.join(".timefusion_meta");
-            let _ = std::fs::create_dir_all(&meta_dir);
-            if let Ok(mut file) = std::fs::OpenOptions::new().create(true).append(true).open(meta_dir.join("topics")) {
-                use std::io::Write;
-                let _ = writeln!(file, "{}", topic);
+            if let Err(e) = std::fs::create_dir_all(&meta_dir) {
+                warn!("Failed to create WAL meta dir {:?}: {}", meta_dir, e);
+                return;
+            }
+            match std::fs::OpenOptions::new().create(true).append(true).open(meta_dir.join("topics")) {
+                Ok(mut file) => {
+                    use std::io::Write;
+                    if let Err(e) = writeln!(file, "{}", topic) {
+                        warn!("Failed to write topic '{}' to index: {}", topic, e);
+                    }
+                }
+                Err(e) => warn!("Failed to open topics file: {}", e),
             }
         }
     }
@@ -414,7 +422,10 @@ fn serialize_record_batch(batch: &RecordBatch) -> Result<Vec<u8>, WalError> {
 
 fn deserialize_record_batch(data: &[u8], schema: &SchemaRef) -> Result<RecordBatch, WalError> {
     if data.len() > MAX_BATCH_SIZE {
-        return Err(WalError::BatchTooLarge { size: data.len(), max: MAX_BATCH_SIZE });
+        return Err(WalError::BatchTooLarge {
+            size: data.len(),
+            max: MAX_BATCH_SIZE,
+        });
     }
 
     let (compact, _): (CompactBatch, _) = bincode::decode_from_slice(data, BINCODE_CONFIG)?;
@@ -451,7 +462,10 @@ fn deserialize_wal_entry(data: &[u8]) -> Result<WalEntry, WalError> {
                 return Err(WalError::TooShort { len: data.len() });
             }
             if data[4] != WAL_VERSION {
-                return Err(WalError::UnsupportedVersion { version: data[4], expected: WAL_VERSION });
+                return Err(WalError::UnsupportedVersion {
+                    version: data[4],
+                    expected: WAL_VERSION,
+                });
             }
             WalOperation::try_from(data[5])?;
             let (entry, _): (WalEntry, _) = bincode::decode_from_slice(&data[6..], BINCODE_CONFIG)?;