perf(blob): batch blob writes during migration to avoid per-blob fsync overhead

crates/admin/src/migrate_store_v2.rs

@@ -465,31 +465,36 @@ impl NewIndexWriterV2 {
         self.commit_tantivy()?;
 
         if !self.email_buf.is_empty() {
-            self.email_buf.sort_by(|a, b| a.0.cmp(&b.0));
+            let mut buf = std::mem::take(&mut self.email_buf);
-            self.email_buf.dedup_by(|a, b| a.0 == b.0);
+            buf.sort_by(|a, b| a.0.cmp(&b.0));
+            buf.dedup_by(|a, b| a.0 == b.0);
 
-            let count = self.email_buf.len();
+            let count = buf.len();
             let mut skipped = 0usize;
-            for (key, data) in &self.email_buf {
+            let mut batch: Vec<([u8; 32], Vec<u8>, Codec)> = Vec::with_capacity(buf.len());
-                if let Err(e) = self.engine.put(*key, data, Codec::Zstd) {
+            for (key, data) in buf {
-                    if matches!(e, bichon_blob::Error::ValueTooLarge { .. }) {
+                if data.len() > 100 * 1024 * 1024 {
-                        eprintln!(
+                    eprintln!(
-                            "{}",
+                        "{}",
-                            console::style(format!(
+                        console::style(format!(
-                                "WARN: skipping oversized email blob key={} ({} bytes)",
+                            "WARN: skipping oversized email blob key={} ({} bytes)",
-                                hex::encode(*key),
+                            hex::encode(key),
-                                data.len()
+                            data.len()
-                            ))
+                        ))
-                            .yellow()
+                        .yellow()
-                        );
+                    );
-                        skipped += 1;
+                    skipped += 1;
-                        continue;
+                    continue;
-                    }
-                    return Err(raise_error!(
-                        format!("blob engine put error: {e:#?}"),
-                        ErrorCode::InternalError
-                    ));
                 }
+                batch.push((key, data, Codec::Zstd));
+            }
+            if !batch.is_empty() {
+                self.engine.put_batch(&batch).map_err(|e| {
+                    raise_error!(
+                        format!("blob engine put_batch error: {e:#?}"),
+                        ErrorCode::InternalError
+                    )
+                })?;
             }
             println!("flushed {} email blobs to engine", count - skipped);
             if skipped > 0 {
@@ -498,44 +503,48 @@ impl NewIndexWriterV2 {
                     console::style(format!("skipped {} oversized email blobs", skipped)).yellow()
                 );
             }
-            self.email_buf.clear();
         }
 
         if !self.attachment_buf.is_empty() {
-            self.attachment_buf.sort_by(|a, b| a.0.cmp(&b.0));
+            let mut buf = std::mem::take(&mut self.attachment_buf);
-            self.attachment_buf.dedup_by(|a, b| a.0 == b.0);
+            buf.sort_by(|a, b| a.0.cmp(&b.0));
+            buf.dedup_by(|a, b| a.0 == b.0);
 
-            let count = self.attachment_buf.len();
+            let count = buf.len();
             let mut skipped = 0usize;
-            for (key, data) in &self.attachment_buf {
+            let mut batch: Vec<([u8; 32], Vec<u8>, Codec)> = Vec::with_capacity(buf.len());
-                if let Err(e) = self.engine.put(*key, data, Codec::Zstd) {
+            for (key, data) in buf {
-                    if matches!(e, bichon_blob::Error::ValueTooLarge { .. }) {
+                if data.len() > 100 * 1024 * 1024 {
-                        eprintln!(
+                    eprintln!(
-                            "{}",
+                        "{}",
-                            console::style(format!(
+                        console::style(format!(
-                                "WARN: skipping oversized attachment blob key={} ({} bytes)",
+                            "WARN: skipping oversized attachment blob key={} ({} bytes)",
-                                hex::encode(*key),
+                            hex::encode(key),
-                                data.len()
+                            data.len()
-                            ))
+                        ))
-                            .yellow()
+                        .yellow()
-                        );
+                    );
-                        skipped += 1;
+                    skipped += 1;
-                        continue;
+                    continue;
-                    }
-                    return Err(raise_error!(
-                        format!("blob engine put error: {e:#?}"),
-                        ErrorCode::InternalError
-                    ));
                 }
+                batch.push((key, data, Codec::Zstd));
+            }
+            if !batch.is_empty() {
+                self.engine.put_batch(&batch).map_err(|e| {
+                    raise_error!(
+                        format!("blob engine put_batch error: {e:#?}"),
+                        ErrorCode::InternalError
+                    )
+                })?;
             }
             println!("flushed {} attachment blobs to engine", count - skipped);
             if skipped > 0 {
                 eprintln!(
                     "{}",
-                    console::style(format!("skipped {} oversized attachment blobs", skipped)).yellow()
+                    console::style(format!("skipped {} oversized attachment blobs", skipped))
+                        .yellow()
                 );
             }
-            self.attachment_buf.clear();
         }
 
         Ok(())

crates/admin/src/migrate_v1.rs

@@ -33,7 +33,9 @@ fn migrate_keyspace(
     db: &Database,
     ks_name: &str,
     label: &str,
+    batch_size: usize,
 ) -> BichonResult<u64> {
+
     let ks = db
         .keyspace(ks_name, || {
             panic!("{ks_name} keyspace not found in fjall database")
@@ -53,6 +55,8 @@ fn migrate_keyspace(
     pb.set_message(format!("Scanning {label} blobs..."));
 
     let mut count: u64 = 0;
+    let mut batch: Vec<([u8; 32], Vec<u8>, Codec)> = Vec::with_capacity(batch_size);
+
     for item in ks.iter() {
         let (key_bytes, value) = item.into_inner().map_err(|e| {
             raise_error!(
@@ -64,31 +68,40 @@ fn migrate_keyspace(
         if value.is_empty() {
             continue;
         }
+        // MAX_VALUE_SIZE = 100 MB (bichon_blob::types)
+        if value.len() > 100 * 1024 * 1024 {
+            let raw_key = hex_key_to_raw(&key_bytes)?;
+            eprintln!(
+                "{}",
+                console::style(format!(
+                    "WARN: skipping oversized blob key={} ({} bytes)",
+                    hex::encode(raw_key),
+                    value.len()
+                ))
+                .yellow()
+            );
+            continue;
+        }
         let raw_key = hex_key_to_raw(&key_bytes)?;
-        if let Err(e) = engine.put(raw_key, &value, Codec::Zstd) {
+        batch.push((raw_key, value.to_vec(), Codec::Zstd));
-            if matches!(e, bichon_blob::Error::ValueTooLarge { .. }) {
+
-                eprintln!(
+        if batch.len() >= batch_size {
-                    "{}",
+            engine.put_batch(&batch).map_err(|e| {
-                    console::style(format!(
+                raise_error!(format!("{e:#?}"), ErrorCode::InternalError)
-                        "WARN: skipping oversized blob key={} ({} bytes)",
+            })?;
-                        hex::encode(raw_key),
+            count += batch.len() as u64;
-                        value.len()
-                    ))
-                    .yellow()
-                );
-                continue;
-            }
-            return Err(raise_error!(
-                format!("bichon-blob put error: {e:#?}"),
-                ErrorCode::InternalError
-            ));
-        }
-        count += 1;
-        if count % 1000 == 0 {
             pb.set_message(format!("{label}: {} blobs migrated...", count));
+            batch.clear();
         }
     }
 
+    if !batch.is_empty() {
+        engine.put_batch(&batch).map_err(|e| {
+            raise_error!(format!("{e:#?}"), ErrorCode::InternalError)
+        })?;
+        count += batch.len() as u64;
+    }
+
     pb.finish_with_message(format!("{label}: {} blobs migrated", count));
     Ok(count)
 }
@@ -173,6 +186,25 @@ pub fn handle_migrate_v1(theme: &ColorfulTheme) {
         return;
     }
 
+    let batch_size: usize = {
+        let input: String = Input::with_theme(theme)
+            .with_prompt("Enter batch size (affects memory usage, higher = faster but uses more RAM)")
+            .default("1000".to_string())
+            .validate_with(|s: &String| match s.trim().parse::<usize>() {
+                Ok(n) if n > 0 => Ok(()),
+                _ => Err("Please enter a valid positive number"),
+            })
+            .interact_text()
+            .unwrap_or("1000".to_string());
+        input.trim().parse::<usize>().unwrap_or(1000)
+    };
+
+    println!(
+        "{} Using batch size: {}\n",
+        style("✓").green(),
+        style(batch_size).cyan().bold()
+    );
+
     // Open old Fjall database (read-only by nature of the iter API)
     println!("\n{}", style("Opening fjall database...").dim());
     let db = match Database::open(FjallConfig::new(&fjall_path)) {
@@ -206,7 +238,7 @@ pub fn handle_migrate_v1(theme: &ColorfulTheme) {
     };
 
     // Migrate email blobs
-    let email_count = match migrate_keyspace(&engine, &db, "email", "Email") {
+    let email_count = match migrate_keyspace(&engine, &db, "email", "Email", batch_size) {
         Ok(n) => n,
         Err(e) => {
             println!("{}", style(format!("Email migration failed: {e:#?}")).red());
@@ -217,7 +249,7 @@ pub fn handle_migrate_v1(theme: &ColorfulTheme) {
 
     // Migrate attachment blobs
     let attach_count =
-        match migrate_keyspace(&engine, &db, "attachments", "Attachment") {
+        match migrate_keyspace(&engine, &db, "attachments", "Attachment", batch_size) {
             Ok(n) => n,
             Err(e) => {
                 println!(
@@ -349,7 +381,7 @@ mod tests {
             config.gc_interval_secs = 0;
             let engine = Engine::open(&blob_path, config).unwrap();
 
-            let count = migrate_keyspace(&engine, &fjall_db, "test_ks", "Test").unwrap();
+            let count = migrate_keyspace(&engine, &fjall_db, "test_ks", "Test", 100).unwrap();
             assert_eq!(count, expected.len() as u64);
 
             engine.flush().unwrap();
@@ -400,7 +432,7 @@ mod tests {
         config.gc_interval_secs = 0;
         let engine = Engine::open(&blob_path, config).unwrap();
 
-        let count = migrate_keyspace(&engine, &fjall_db, "empty_ks", "Empty").unwrap();
+        let count = migrate_keyspace(&engine, &fjall_db, "empty_ks", "Empty", 100).unwrap();
         assert_eq!(count, 0);
 
         engine.shutdown().unwrap();

crates/blob/src/engine.rs

@@ -400,6 +400,9 @@ impl Engine {
 
         self.shared.index_store.insert_batch(&records)?;
 
+        // Deduplicate: keep only the max offset per segment.
+        ends.sort_by(|a, b| a.0.cmp(&b.0).then(b.1.cmp(&a.1)));
+        ends.dedup_by(|a, b| a.0 == b.0);
         for (segment_id, entry_end) in &ends {
             inner.mark_indexed(*segment_id, *entry_end)?;
         }