diff --git a/crates/admin/src/migrate_store_v2.rs b/crates/admin/src/migrate_store_v2.rs index dd2c00a..4bfa9eb 100644 --- a/crates/admin/src/migrate_store_v2.rs +++ b/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)); - self.email_buf.dedup_by(|a, b| a.0 == b.0); + let mut buf = std::mem::take(&mut self.email_buf); + 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 { - if let Err(e) = self.engine.put(*key, data, Codec::Zstd) { - if matches!(e, bichon_blob::Error::ValueTooLarge { .. }) { - eprintln!( - "{}", - console::style(format!( - "WARN: skipping oversized email blob key={} ({} bytes)", - hex::encode(*key), - data.len() - )) - .yellow() - ); - skipped += 1; - continue; - } - return Err(raise_error!( - format!("blob engine put error: {e:#?}"), - ErrorCode::InternalError - )); + let mut batch: Vec<([u8; 32], Vec, Codec)> = Vec::with_capacity(buf.len()); + for (key, data) in buf { + if data.len() > 100 * 1024 * 1024 { + eprintln!( + "{}", + console::style(format!( + "WARN: skipping oversized email blob key={} ({} bytes)", + hex::encode(key), + data.len() + )) + .yellow() + ); + skipped += 1; + continue; } + 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)); - self.attachment_buf.dedup_by(|a, b| a.0 == b.0); + let mut buf = std::mem::take(&mut self.attachment_buf); + 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 { - if let Err(e) = self.engine.put(*key, data, Codec::Zstd) { - if matches!(e, bichon_blob::Error::ValueTooLarge { .. }) { - eprintln!( - "{}", - console::style(format!( - "WARN: skipping oversized attachment blob key={} ({} bytes)", - hex::encode(*key), - data.len() - )) - .yellow() - ); - skipped += 1; - continue; - } - return Err(raise_error!( - format!("blob engine put error: {e:#?}"), - ErrorCode::InternalError - )); + let mut batch: Vec<([u8; 32], Vec, Codec)> = Vec::with_capacity(buf.len()); + for (key, data) in buf { + if data.len() > 100 * 1024 * 1024 { + eprintln!( + "{}", + console::style(format!( + "WARN: skipping oversized attachment blob key={} ({} bytes)", + hex::encode(key), + data.len() + )) + .yellow() + ); + skipped += 1; + continue; } + 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(()) diff --git a/crates/admin/src/migrate_v1.rs b/crates/admin/src/migrate_v1.rs index 555996b..350a9e7 100644 --- a/crates/admin/src/migrate_v1.rs +++ b/crates/admin/src/migrate_v1.rs @@ -33,7 +33,9 @@ fn migrate_keyspace( db: &Database, ks_name: &str, label: &str, + batch_size: usize, ) -> BichonResult { + 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, 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) { - if matches!(e, bichon_blob::Error::ValueTooLarge { .. }) { - eprintln!( - "{}", - console::style(format!( - "WARN: skipping oversized blob key={} ({} bytes)", - hex::encode(raw_key), - value.len() - )) - .yellow() - ); - continue; - } - return Err(raise_error!( - format!("bichon-blob put error: {e:#?}"), - ErrorCode::InternalError - )); - } - count += 1; - if count % 1000 == 0 { + batch.push((raw_key, value.to_vec(), Codec::Zstd)); + + if batch.len() >= batch_size { + engine.put_batch(&batch).map_err(|e| { + raise_error!(format!("{e:#?}"), ErrorCode::InternalError) + })?; + count += batch.len() as u64; 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::() { + Ok(n) if n > 0 => Ok(()), + _ => Err("Please enter a valid positive number"), + }) + .interact_text() + .unwrap_or("1000".to_string()); + input.trim().parse::().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(); diff --git a/crates/blob/src/engine.rs b/crates/blob/src/engine.rs index 609c827..5656b5f 100644 --- a/crates/blob/src/engine.rs +++ b/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)?; }