diff --git a/Cargo.lock b/Cargo.lock index 58da3ff..bd26c34 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -303,16 +303,23 @@ version = "1.6.2" dependencies = [ "bichon-core", "bichon-memdb", + "bytes 1.12.0", + "chrono", "console", "dialoguer", + "fjall", "indicatif", "itertools 0.15.0", + "mail-parser", "native_db", "native_model", "serde", "serde_json", "snafu", + "tantivy", "tokio", + "tracing", + "uuid", ] [[package]] @@ -361,6 +368,7 @@ version = "1.6.2" dependencies = [ "async-imap", "base64 0.22.1", + "bichon-blob", "bichon-memdb", "blake3", "bytes 1.12.0", @@ -373,7 +381,6 @@ dependencies = [ "deunicode", "email_address", "encoding_rs", - "fjall", "futures", "governor", "hex", @@ -3581,9 +3588,9 @@ dependencies = [ [[package]] name = "quick_cache" -version = "0.6.21" +version = "0.6.24" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a70b1b8b47e31d0498ecbc3c5470bb931399a8bfed1fd79d1717a61ce7f96e3" +checksum = "b9c6658afe513a3b484e3abfdaa0d03ef3c0bbf017542c178dd55f94eb3051f9" dependencies = [ "equivalent", "hashbrown 0.16.1", @@ -5370,9 +5377,9 @@ checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" [[package]] name = "varint-rs" -version = "2.2.0" +version = "2.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8f54a172d0620933a27a4360d3db3e2ae0dd6cceae9730751a036bbf182c4b23" +checksum = "bfa6c38708f6257f1ec2ca7e5a11f9bbf58a27d7060078b6b333624968183d96" [[package]] name = "vcpkg" @@ -6054,9 +6061,9 @@ dependencies = [ [[package]] name = "xxhash-rust" -version = "0.8.15" +version = "0.8.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fdd20c5420375476fbd4394763288da7eb0cc0b8c11deed431a91562af7335d3" +checksum = "4d93c89cdc2d3a63c3ec48ffe926931bdc069eafa8e4402fe6d8f790c9d1e576" [[package]] name = "yasna" diff --git a/crates/admin/Cargo.toml b/crates/admin/Cargo.toml index 5184ed5..02f34d1 100644 --- a/crates/admin/Cargo.toml +++ b/crates/admin/Cargo.toml @@ -17,4 +17,11 @@ serde_json.workspace = true itertools.workspace = true snafu.workspace = true -bichon-memdb.workspace = true \ No newline at end of file +bichon-memdb.workspace = true +fjall.workspace = true +mail-parser.workspace = true +bytes.workspace = true +uuid.workspace = true +tantivy = { version = "0.26.1", features = ["zstd-compression", "quickwit"] } +chrono.workspace = true +tracing.workspace = true \ No newline at end of file diff --git a/crates/core/src/migrate/legacy/fields.rs b/crates/admin/src/legacy/fields.rs similarity index 100% rename from crates/core/src/migrate/legacy/fields.rs rename to crates/admin/src/legacy/fields.rs diff --git a/crates/core/src/migrate/legacy/mod.rs b/crates/admin/src/legacy/mod.rs similarity index 100% rename from crates/core/src/migrate/legacy/mod.rs rename to crates/admin/src/legacy/mod.rs diff --git a/crates/core/src/migrate/legacy/schema.rs b/crates/admin/src/legacy/schema.rs similarity index 98% rename from crates/core/src/migrate/legacy/schema.rs rename to crates/admin/src/legacy/schema.rs index 9a0d762..1fcfd53 100644 --- a/crates/core/src/migrate/legacy/schema.rs +++ b/crates/admin/src/legacy/schema.rs @@ -1,6 +1,6 @@ use tantivy::schema::{FacetOptions, Field, Schema, FAST, INDEXED, STORED, STRING, TEXT}; -use crate::migrate::legacy::fields::{EmlFields, EnvelopeFields, *}; +use crate::legacy::fields::{EmlFields, EnvelopeFields, *}; pub struct SchemaTools; diff --git a/crates/admin/src/main.rs b/crates/admin/src/main.rs index 15655a1..8f95a00 100644 --- a/crates/admin/src/main.rs +++ b/crates/admin/src/main.rs @@ -21,8 +21,10 @@ use dialoguer::{theme::ColorfulTheme, Select}; use crate::{migrate::handle_migration, reset::handle_reset_password}; +pub mod legacy; pub mod meta; pub mod migrate; +pub mod migrate_store; pub mod reset; diff --git a/crates/admin/src/migrate.rs b/crates/admin/src/migrate.rs index 0005e5f..2db47c4 100644 --- a/crates/admin/src/migrate.rs +++ b/crates/admin/src/migrate.rs @@ -1,12 +1,23 @@ -use std::path::{Path, PathBuf}; +use std::{collections::HashMap, path::PathBuf}; -use bichon_core::migrate::{ - count_eml_segments, do_migrate_segment, is_tantivy_index_dir, - store::{LegacyDirs, NewDirs, NewIndexWriter}, +use bichon_core::{ + error::{code::ErrorCode, BichonResult}, + migrate::is_tantivy_index_dir, + raise_error, }; use console::style; use dialoguer::{theme::ColorfulTheme, Confirm, Input}; use indicatif::{ProgressBar, ProgressStyle}; +use tantivy::{ + collector::TopDocs, + columnar::Column, + query::TermQuery, + schema::{IndexRecordOption, Value}, + DocAddress, Index, TantivyDocument, Term, +}; + +use crate::legacy::schema::SchemaTools; +use crate::migrate_store::{LegacyDirs, NewDirs, NewIndexWriter}; pub fn handle_migration(theme: &ColorfulTheme) { println!( @@ -21,7 +32,7 @@ pub fn handle_migration(theme: &ColorfulTheme) { style( "This tool migrates data from the legacy v0.3.7 Tantivy-based storage \ architecture to the new v1.x \ - separated index and Fjall-backed storage format." + separated index and blob-backed storage format." ) .dim() ); @@ -35,8 +46,8 @@ pub fn handle_migration(theme: &ColorfulTheme) { New v1.x architecture:\n\ • mail indexes stored in Tantivy\n\ • attachment indexes stored in Tantivy\n\ - • raw message data stored in Fjall\n\ - • attachment blobs stored in Fjall" + • raw message data stored in blob engine\n\ + • attachment blobs stored in blob engine" ) .dim() ); @@ -54,7 +65,7 @@ pub fn handle_migration(theme: &ColorfulTheme) { let root_dir_str: String = Input::with_theme(theme) .with_prompt("Enter --bichon-root-dir (same value used by the old server)") .validate_with(|input: &String| -> Result<(), &str> { - let path = Path::new(input); + let path = PathBuf::from(input); if !path.is_absolute() { return Err("Path must be absolute."); } @@ -81,7 +92,7 @@ pub fn handle_migration(theme: &ColorfulTheme) { if input.is_empty() { return Ok(()); } - let path = Path::new(input); + let path = PathBuf::from(input); if !path.is_absolute() { return Err("Path must be absolute."); } @@ -119,7 +130,7 @@ pub fn handle_migration(theme: &ColorfulTheme) { if input.is_empty() { return Ok(()); } - let path = Path::new(input); + let path = PathBuf::from(input); if !path.is_absolute() { return Err("Path must be absolute."); } @@ -447,3 +458,210 @@ pub fn is_legacy_data_layout_with_paths( Ok(envelope_result || eml_result) } + +/// Return the number of segments in the legacy EML Tantivy index. +/// Each segment can be passed to `do_migrate_segment` for bounded-memory batch migration. +pub fn count_eml_segments(legacy: &LegacyDirs) -> BichonResult { + let eml_index = Index::open_in_dir(&legacy.eml_dir) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + let reader = eml_index + .reader() + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + let searcher = reader.searcher(); + Ok(searcher.segment_readers().len()) +} + +/// Migrate all documents from a single EML segment to the new storage layout. +/// +/// This is the core of the batch migration strategy: each Process B invocation +/// handles exactly one EML segment, so peak memory is bounded by that segment's +/// size regardless of the total archive size. +pub fn do_migrate_segment( + batch_size: u32, + legacy: LegacyDirs, + writer: &mut NewIndexWriter, + segment_index: usize, + mut on_progress: F, +) -> BichonResult<()> +where + F: FnMut(&str), +{ + // ── open legacy indices ──────────────────────────────────────────── + let envelope_index = Index::open_in_dir(&legacy.envelope_dir) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + let eml_index = Index::open_in_dir(&legacy.eml_dir) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + + let envelope_reader = envelope_index + .reader() + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + let eml_reader = eml_index + .reader() + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + + let envelope_searcher = envelope_reader.searcher(); + let eml_searcher = eml_reader.searcher(); + + let ef = SchemaTools::envelope_fields(); + let mf = SchemaTools::eml_fields(); + + let eml_segments = eml_searcher.segment_readers(); + let eml_segment = eml_segments.get(segment_index).ok_or_else(|| { + raise_error!( + format!( + "segment index {} out of range ({} segments)", + segment_index, + eml_segments.len() + ), + ErrorCode::InternalError + ) + })?; + + let num_docs = eml_segment.num_docs(); + if num_docs == 0 { + on_progress("TOTAL:0"); + on_progress("DONE:0:0"); + return Ok(()); + } + + on_progress(&format!("TOTAL:{}", num_docs)); + + let max_doc = eml_segment.max_doc(); + let ff = eml_segment.fast_fields(); + let f_id_col: Column = ff.u64("id").map_err(|e| { + raise_error!( + format!("failed to open f_id fast field: {e:#?}"), + ErrorCode::InternalError + ) + })?; + + // ── Phase 1: build eid → (uid, internal_date) from envelope, then drop it ── + let mut envelope_map: HashMap = HashMap::with_capacity(num_docs as usize); + + let mut env_scanned = 0u32; + let mut env_skipped = 0u32; + for doc_id in 0..max_doc { + if eml_segment.is_deleted(doc_id) { + continue; + } + let eid = f_id_col.values.get_val(doc_id); + + let term = Term::from_field_u64(ef.f_id, eid); + let query = TermQuery::new(term, IndexRecordOption::Basic); + let hits: Vec<(_, DocAddress)> = envelope_searcher + .search(&query, &TopDocs::with_limit(1).order_by_score()) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + + if let Some((_, addr)) = hits.first() { + let env_doc: TantivyDocument = envelope_searcher + .doc(*addr) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + let uid = env_doc + .get_first(ef.f_uid) + .and_then(|v| v.as_u64()) + .unwrap_or(0) as u32; + let internal_date = env_doc + .get_first(ef.f_internal_date) + .and_then(|v| v.as_i64()) + .unwrap_or(0); + envelope_map.insert(eid, (uid, internal_date)); + env_scanned += 1; + } else { + env_skipped += 1; + } + + if env_scanned % 10 == 0 { + on_progress(&format!( + "PHASE1:{}/{} skipped:{}", + env_scanned, max_doc, env_skipped + )); + } + } + + // Free the envelope index before the heavy EML processing. + drop(envelope_searcher); + drop(envelope_reader); + drop(envelope_index); + + // ── Phase 2: process EML docs, streaming one at a time ───────────── + let mut total_migrated = 0usize; + let mut total_skipped = 0usize; + + let mut chunk_start = 0u32; + + while chunk_start < max_doc { + let chunk_end = (chunk_start + batch_size).min(max_doc); + let store_reader = eml_segment + .get_store_reader(2) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + + for doc_id in chunk_start..chunk_end { + if eml_segment.is_deleted(doc_id) { + continue; + } + + let eid = f_id_col.values.get_val(doc_id); + + let (uid, internal_date) = match envelope_map.get(&eid) { + Some(v) => *v, + None => { + on_progress(&format!("WARN: eid {} envelope not found", eid)); + total_skipped += 1; + continue; + } + }; + + let eml_doc: TantivyDocument = store_reader + .get(doc_id) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + + let account_id = match eml_doc.get_first(mf.f_account_id).and_then(|v| v.as_u64()) { + Some(v) => v, + None => { + on_progress(&format!("WARN: eid {} account_id missing", eid)); + total_skipped += 1; + continue; + } + }; + let mailbox_id = eml_doc + .get_first(mf.f_mailbox_id) + .and_then(|v| v.as_u64()) + .unwrap_or(0); + + // Borrow directly from eml_doc — no .to_vec() clone. + let eml_bytes = match eml_doc.get_first(mf.f_eml).and_then(|v| v.as_bytes()) { + Some(b) => b, + None => { + on_progress(&format!("WARN: eid {} eml bytes missing", eid)); + total_skipped += 1; + continue; + } + }; + + if let Err(e) = writer.ingest(eml_bytes, account_id, mailbox_id, uid, internal_date) { + on_progress(&format!( + "ERROR: Account {} eid {} ingest failed: {}", + account_id, eid, e + )); + total_skipped += 1; + continue; + } + + total_migrated += 1; + + if total_migrated % 10 == 0 || total_migrated as u32 == num_docs { + on_progress(&format!("PROGRESS:{}:{}", total_migrated, num_docs)); + } + } + + drop(store_reader); + + // Flush blob buffers via ingestion API — bypasses memtable/WAL. + writer.flush_fjall_buffers()?; + + chunk_start = chunk_end; + } + + on_progress(&format!("DONE:{}:{}", total_migrated, total_skipped)); + Ok(()) +} diff --git a/crates/core/src/migrate/store.rs b/crates/admin/src/migrate_store.rs similarity index 99% rename from crates/core/src/migrate/store.rs rename to crates/admin/src/migrate_store.rs index 8dc203f..792e674 100644 --- a/crates/core/src/migrate/store.rs +++ b/crates/admin/src/migrate_store.rs @@ -3,7 +3,7 @@ use std::{path::PathBuf, time::Instant}; use bytes::Bytes; use mail_parser::MimeHeaders; -use crate::{ +use bichon_core::{ envelope::extractor::extract_references, message::content::AttachmentInfo, store::tantivy::tokenizers::EuroTokenizer, utils::compute_content_hash, }; @@ -16,7 +16,7 @@ use mail_parser::MessageParser; use tantivy::{indexer::NoMergePolicy, Index, IndexWriter, TantivyDocument}; use uuid::Uuid; -use crate::{ +use bichon_core::{ common::AddrVec, envelope::extractor::{compute_thread_id, generate_message_id}, error::{code::ErrorCode, BichonResult}, @@ -284,7 +284,7 @@ impl NewIndexWriter { .or_else(|| { message .body_html(0) - .map(|html| crate::utils::html::extract_text(html.into_owned())) + .map(|html| bichon_core::utils::html::extract_text(html.into_owned())) }) .unwrap_or_default(); let text = text.split_whitespace().collect::>().join(" "); diff --git a/crates/blob/src/engine.rs b/crates/blob/src/engine.rs index 6ecbe26..0f8b607 100644 --- a/crates/blob/src/engine.rs +++ b/crates/blob/src/engine.rs @@ -188,8 +188,9 @@ impl Engine { self.shared.config.compression_level, ); + let original_len = value.len() as u32; let (segment_id, offset, data_size) = - inner.append_entry(key, &data, 0, actual_codec)?; + inner.append_entry(key, &data, original_len, 0, actual_codec)?; let record = IndexRecord::new(key, segment_id, offset, data_size, 0); self.shared.bucket_store.insert(record)?; @@ -226,7 +227,7 @@ impl Engine { let mut inner = self.shared.inner.write().unwrap(); let (segment_id, offset, data_size) = - inner.append_entry(*key, &[], 1, Codec::None)?; + inner.append_entry(*key, &[], 0, 1, Codec::None)?; let record = IndexRecord::new(*key, segment_id, offset, data_size, 1); self.shared.bucket_store.insert(record)?; @@ -256,7 +257,7 @@ impl Engine { for key in keys { let (segment_id, offset, data_size) = - inner.append_entry(*key, &[], 1, Codec::None)?; + inner.append_entry(*key, &[], 0, 1, Codec::None)?; let entry_end = offset + ENTRY_HEADER_SIZE as u64 + data_size as u64; records.push(IndexRecord::new(*key, segment_id, offset, data_size, 1)); @@ -298,8 +299,9 @@ impl Engine { self.shared.config.compression_level, ); + let original_len = value.len() as u32; let (segment_id, offset, data_size) = - inner.append_entry(*key, &data, 0, actual_codec)?; + inner.append_entry(*key, &data, original_len, 0, actual_codec)?; let entry_end = offset + ENTRY_HEADER_SIZE as u64 + data_size as u64; records.push(IndexRecord::new(*key, segment_id, offset, data_size, 0)); @@ -453,6 +455,7 @@ impl EngineInner { &mut self, key: [u8; 32], data: &[u8], + raw_size: u32, flags: u8, codec: Codec, ) -> Result<(u32, u64, u32)> { @@ -464,7 +467,7 @@ impl EngineInner { let entry = if flags == 1 { Entry::tombstone(key) } else { - Entry::new(key, data, flags, codec) + Entry::new(key, data, raw_size, flags, codec) }; let data_size = entry.data.len() as u32; diff --git a/crates/blob/src/segment.rs b/crates/blob/src/segment.rs index 0d9ecb1..7f3808e 100644 --- a/crates/blob/src/segment.rs +++ b/crates/blob/src/segment.rs @@ -19,13 +19,14 @@ pub struct Entry { impl Entry { /// Create a normal data entry. - pub fn new(key: [u8; 32], raw_data: &[u8], flags: u8, codec: Codec) -> Self { + /// `raw_size` is the original uncompressed size; `data` is what goes to disk. + pub fn new(key: [u8; 32], data: &[u8], raw_size: u32, flags: u8, codec: Codec) -> Self { Self { flags, codec, key, - raw_size: raw_data.len() as u32, - data: raw_data.to_vec(), + raw_size, + data: data.to_vec(), } } @@ -460,7 +461,7 @@ mod tests { let key = [0xAAu8; 32]; let data = b"hello world".to_vec(); - let entry = Entry::new(key, &data, 0, Codec::None); + let entry = Entry::new(key, &data, data.len() as u32, 0, Codec::None); { let mut writer = SegmentWriter::create(path.clone(), 1).unwrap(); writer.append(&entry).unwrap(); @@ -506,7 +507,7 @@ mod tests { .map(|i| { let mut key = [0u8; 32]; key[0] = i; - Entry::new(key, &vec![i; 100], 0, Codec::None) + Entry::new(key, &vec![i; 100], 100, 0, Codec::None) }) .collect(); diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index 1b92d13..9794b0d 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -63,7 +63,7 @@ bytes.workspace = true mail-send.workspace = true blake3.workspace = true uuid.workspace = true -fjall.workspace = true +bichon-blob.workspace = true tracing-log.workspace = true tokio-util.workspace = true whichlang = "0.1.1" diff --git a/crates/core/src/migrate/mod.rs b/crates/core/src/migrate/mod.rs index 05308ef..10acd69 100644 --- a/crates/core/src/migrate/mod.rs +++ b/crates/core/src/migrate/mod.rs @@ -1,24 +1,24 @@ -use std::{collections::HashMap, path::PathBuf}; +use std::path::{Path, PathBuf}; -use crate::{ - error::{code::ErrorCode, BichonResult}, - migrate::{ - legacy::schema::SchemaTools, - store::{LegacyDirs, NewIndexWriter}, - }, - raise_error, - settings::cli::SETTINGS, -}; -use tantivy::{ - collector::TopDocs, - columnar::Column, - query::TermQuery, - schema::{IndexRecordOption, Value}, - DocAddress, Index, TantivyDocument, Term, -}; +use crate::settings::cli::SETTINGS; -pub mod legacy; -pub mod store; +/// Current storage layout version. +/// - 1: fjall-based blob storage (post v0.3.7 migration) +/// - 2: bichon-blob based storage +pub const CURRENT_STORAGE_VERSION: u32 = 2; + +const VERSION_FILE: &str = "STORAGE_VERSION"; + +/// Read the storage layout version from `root_dir/STORAGE_VERSION`. +pub fn read_storage_version(root_dir: &Path) -> Option { + let content = std::fs::read_to_string(root_dir.join(VERSION_FILE)).ok()?; + content.trim().parse().ok() +} + +/// Write the storage layout version to `root_dir/STORAGE_VERSION`. +pub fn write_storage_version(root_dir: &Path, version: u32) -> std::io::Result<()> { + std::fs::write(root_dir.join(VERSION_FILE), format!("{}\n", version)) +} pub fn is_tantivy_index_dir(dir: &PathBuf) -> std::io::Result { if !dir.exists() || !dir.is_dir() { @@ -47,28 +47,17 @@ pub fn is_tantivy_index_dir(dir: &PathBuf) -> std::io::Result { Ok(has_meta_json && match_count >= 3) } -/// Return the number of segments in the legacy EML Tantivy index. -/// Each segment can be passed to `do_migrate_segment` for bounded-memory batch migration. -pub fn count_eml_segments(legacy: &LegacyDirs) -> BichonResult { - let eml_index = Index::open_in_dir(&legacy.eml_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let reader = eml_index - .reader() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let searcher = reader.searcher(); - Ok(searcher.segment_readers().len()) -} - +/// Check whether the data layout is compatible with the current server. +/// Returns `false` only when legacy v0.3.7 data is detected and migration is required. pub fn check_data_status() -> std::io::Result { let root_dir = PathBuf::from(&SETTINGS.bichon_root_dir); - let new_indices_base = SETTINGS - .bichon_index_dir - .as_ref() - .map(PathBuf::from) - .unwrap_or_else(|| root_dir.clone()); - let new_indices_path = new_indices_base.join("bichon-indices"); + // 1. Version file takes precedence + if let Some(version) = read_storage_version(&root_dir) { + return Ok(version >= 1); + } + // 2. No version file — check for existing v1.x-style storage (fjall era) let new_data_base = SETTINGS .bichon_data_dir .as_ref() @@ -76,14 +65,13 @@ pub fn check_data_status() -> std::io::Result { .unwrap_or_else(|| root_dir.clone()); let new_storage_path = new_data_base.join("bichon-storage"); - let has_new_indices = is_tantivy_index_dir(&new_indices_path.join("attachment_metadata"))? - && is_tantivy_index_dir(&new_indices_path.join("mail_metadata"))?; - let has_new_storage = is_dir_not_empty(&new_storage_path)?; - - if has_new_indices && has_new_storage { + if is_dir_not_empty(&new_storage_path)? { + // Existing v1.x install predates version file — mark it + let _ = write_storage_version(&root_dir, 1); return Ok(true); } + // 3. Check for legacy v0.3.7 Tantivy layout let legacy_index_root = SETTINGS .bichon_index_dir .as_ref() @@ -99,9 +87,9 @@ pub fn check_data_status() -> std::io::Result { let has_legacy_data = is_tantivy_index_dir(&legacy_data_root)?; if has_legacy_index || has_legacy_data { - Ok(false) + Ok(false) // Needs migration } else { - Ok(true) + Ok(true) // Fresh install } } @@ -112,213 +100,3 @@ fn is_dir_not_empty(path: &PathBuf) -> std::io::Result { let mut entries = std::fs::read_dir(path)?; Ok(entries.next().is_some()) } - -/// Migrate all documents from a single EML segment to the new storage layout. -/// -/// This is the core of the batch migration strategy: each Process B invocation -/// handles exactly one EML segment, so peak memory is bounded by that segment's -/// size regardless of the total archive size. -pub fn do_migrate_segment( - batch_size: u32, - legacy: LegacyDirs, - writer: &mut NewIndexWriter, - segment_index: usize, - mut on_progress: F, -) -> BichonResult<()> -where - F: FnMut(&str), -{ - // ── open legacy indices ──────────────────────────────────────────── - let envelope_index = Index::open_in_dir(&legacy.envelope_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let eml_index = Index::open_in_dir(&legacy.eml_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - let envelope_reader = envelope_index - .reader() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let eml_reader = eml_index - .reader() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - let envelope_searcher = envelope_reader.searcher(); - let eml_searcher = eml_reader.searcher(); - - let ef = SchemaTools::envelope_fields(); - let mf = SchemaTools::eml_fields(); - - let eml_segments = eml_searcher.segment_readers(); - let eml_segment = eml_segments.get(segment_index).ok_or_else(|| { - raise_error!( - format!( - "segment index {} out of range ({} segments)", - segment_index, - eml_segments.len() - ), - ErrorCode::InternalError - ) - })?; - - let num_docs = eml_segment.num_docs(); - if num_docs == 0 { - on_progress("TOTAL:0"); - on_progress("DONE:0:0"); - return Ok(()); - } - - on_progress(&format!("TOTAL:{}", num_docs)); - - let max_doc = eml_segment.max_doc(); - let ff = eml_segment.fast_fields(); - let f_id_col: Column = ff.u64("id").map_err(|e| { - raise_error!( - format!("failed to open f_id fast field: {e:#?}"), - ErrorCode::InternalError - ) - })?; - - // ── Phase 1: build eid → (uid, internal_date) from envelope, then drop it ── - let mut envelope_map: HashMap = HashMap::with_capacity(num_docs as usize); - - let mut env_scanned = 0u32; - let mut env_skipped = 0u32; - for doc_id in 0..max_doc { - if eml_segment.is_deleted(doc_id) { - continue; - } - let eid = f_id_col.values.get_val(doc_id); - - let term = Term::from_field_u64(ef.f_id, eid); - let query = TermQuery::new(term, IndexRecordOption::Basic); - let hits: Vec<(_, DocAddress)> = envelope_searcher - .search(&query, &TopDocs::with_limit(1).order_by_score()) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - if let Some((_, addr)) = hits.first() { - let env_doc: TantivyDocument = envelope_searcher - .doc(*addr) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let uid = env_doc - .get_first(ef.f_uid) - .and_then(|v| v.as_u64()) - .unwrap_or(0) as u32; - let internal_date = env_doc - .get_first(ef.f_internal_date) - .and_then(|v| v.as_i64()) - .unwrap_or(0); - envelope_map.insert(eid, (uid, internal_date)); - env_scanned += 1; - } else { - env_skipped += 1; - } - - if env_scanned % 10 == 0 { - on_progress(&format!( - "PHASE1:{}/{} skipped:{}", - env_scanned, max_doc, env_skipped - )); - } - } - - // Free the envelope index before the heavy EML processing. - drop(envelope_searcher); - drop(envelope_reader); - drop(envelope_index); - - // ── Phase 2: process EML docs, streaming one at a time ───────────── - let mut total_migrated = 0usize; - let mut total_skipped = 0usize; - - // Recreate the StoreReader periodically to bound any internal caches. - //const CHUNK_SIZE: u32 = 3000; - let mut chunk_start = 0u32; - - while chunk_start < max_doc { - let chunk_end = (chunk_start + batch_size).min(max_doc); - let store_reader = eml_segment - .get_store_reader(2) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - for doc_id in chunk_start..chunk_end { - if eml_segment.is_deleted(doc_id) { - continue; - } - - let eid = f_id_col.values.get_val(doc_id); - - let (uid, internal_date) = match envelope_map.get(&eid) { - Some(v) => *v, - None => { - on_progress(&format!("WARN: eid {} envelope not found", eid)); - total_skipped += 1; - continue; - } - }; - - let eml_doc: TantivyDocument = store_reader - .get(doc_id) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - let account_id = match eml_doc.get_first(mf.f_account_id).and_then(|v| v.as_u64()) { - Some(v) => v, - None => { - on_progress(&format!("WARN: eid {} account_id missing", eid)); - total_skipped += 1; - continue; - } - }; - let mailbox_id = eml_doc - .get_first(mf.f_mailbox_id) - .and_then(|v| v.as_u64()) - .unwrap_or(0); - - // Borrow directly from eml_doc — no .to_vec() clone. - let eml_bytes = match eml_doc.get_first(mf.f_eml).and_then(|v| v.as_bytes()) { - Some(b) => b, - None => { - on_progress(&format!("WARN: eid {} eml bytes missing", eid)); - total_skipped += 1; - continue; - } - }; - - if let Err(e) = writer.ingest(eml_bytes, account_id, mailbox_id, uid, internal_date) { - on_progress(&format!( - "ERROR: Account {} eid {} ingest failed: {}", - account_id, eid, e - )); - total_skipped += 1; - continue; - } - - total_migrated += 1; - - if total_migrated % 10 == 0 || total_migrated as u32 == num_docs { - on_progress(&format!("PROGRESS:{}:{}", total_migrated, num_docs)); - } - } - - drop(store_reader); - - // Flush Fjall buffers via ingestion API — bypasses memtable/WAL. - writer.flush_fjall_buffers()?; - - chunk_start = chunk_end; - } - - on_progress(&format!("DONE:{}:{}", total_migrated, total_skipped)); - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_real_tantivy_dir() { - let path = PathBuf::from(r"D:\test-data\envelope"); - let result = is_tantivy_index_dir(&path).unwrap(); - println!("is tantivy index dir: {}", result); - assert!(result); - } -} diff --git a/crates/core/src/settings/dir.rs b/crates/core/src/settings/dir.rs index 0578117..359d2df 100644 --- a/crates/core/src/settings/dir.rs +++ b/crates/core/src/settings/dir.rs @@ -17,6 +17,7 @@ // along with this program. If not, see . use crate::context::Initialize; +use crate::migrate::{write_storage_version, CURRENT_STORAGE_VERSION}; use crate::settings::cli::SETTINGS; use crate::{ error::{code::ErrorCode, BichonResult}, @@ -62,6 +63,14 @@ impl Initialize for DataDirManager { .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; std::fs::create_dir_all(&DATA_DIR_MANAGER.storage_dir) .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + // Write STORAGE_VERSION on fresh install (no existing data) + let version_path = DATA_DIR_MANAGER.root_dir.join("STORAGE_VERSION"); + if !version_path.exists() && !DATA_DIR_MANAGER.storage_dir.join("blobs").exists() { + write_storage_version(&DATA_DIR_MANAGER.root_dir, CURRENT_STORAGE_VERSION) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + } + Ok(()) } } diff --git a/crates/core/src/store/blob.rs b/crates/core/src/store/blob.rs index 34245de..d1852f0 100644 --- a/crates/core/src/store/blob.rs +++ b/crates/core/src/store/blob.rs @@ -16,17 +16,17 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use crate::raise_error; use crate::{ common::signal::SIGNAL_MANAGER, envelope::extractor::reattach_eml_content_self_healing, error::{code::ErrorCode, BichonResult}, settings::dir::DATA_DIR_MANAGER, }; -use crate::raise_error; +use bichon_blob::{Codec, Config, Engine}; use bytes::Bytes; -use fjall::{CompressionType, Database, Keyspace, KeyspaceCreateOptions, KvSeparationOptions, config::{BlockSizePolicy, CompressionPolicy}}; -use std::{io::Cursor, sync::LazyLock}; +use std::{io::Cursor, sync::Arc, sync::LazyLock}; use tokio::{ sync::{mpsc, Mutex}, task::{self, JoinHandle}, @@ -41,33 +41,48 @@ pub struct DetachedEmail { pub struct BlobManager { sender: mpsc::Sender, - db: Database, - email_keyspace: Keyspace, - attachments_keyspace: Keyspace, + engine: Arc, handle: Mutex>>, } +fn hex_to_key(hex: &str) -> BichonResult<[u8; 32]> { + let mut key = [0u8; 32]; + hex::decode_to_slice(hex, &mut key).map_err(|e| { + raise_error!( + format!("invalid content hash '{hex}': {e:#?}"), + ErrorCode::InternalError + ) + })?; + Ok(key) +} + impl BlobManager { pub async fn shutdown(&self) { let mut guard = self.handle.lock().await; if let Some(handle) = guard.take() { let _ = handle.await; } + if let Err(e) = self.engine.shutdown() { + tracing::error!("blob engine shutdown error: {}", e); + } } - fn process_detached_email( - eml: DetachedEmail, - email_ks: &Keyspace, - attach_ks: &Keyspace, - ) { + fn process_detached_email(eml: DetachedEmail, engine: &Engine) { let (email_hash, email_data) = eml.email; - match email_ks.contains_key(&email_hash) { + let email_key = match hex_to_key(&email_hash) { + Ok(k) => k, + Err(e) => { + tracing::error!("{:#?}", e); + return; + } + }; + match engine.exists(&email_key) { Ok(false) => { - if let Err(e) = email_ks.insert(email_hash, email_data) { - tracing::error!("CRITICAL: Failed to insert email: {:?}", e); + if let Err(e) = engine.put(email_key, &email_data, Codec::Lz4) { + tracing::error!("CRITICAL: Failed to insert email blob: {:?}", e); } } - Err(e) => tracing::error!("Fjall email_ks error: {:?}", e), + Err(e) => tracing::error!("blob engine error: {:?}", e), Ok(true) => { tracing::debug!("Email blob already exists (dedup): {}", &email_hash); } @@ -75,13 +90,20 @@ impl BlobManager { if let Some(attachments) = eml.attachments { for (a_hash, a_data) in attachments { - match attach_ks.contains_key(&a_hash) { + let a_key = match hex_to_key(&a_hash) { + Ok(k) => k, + Err(e) => { + tracing::error!("{:#?}", e); + continue; + } + }; + match engine.exists(&a_key) { Ok(false) => { - if let Err(e) = attach_ks.insert(a_hash, a_data) { - tracing::error!("CRITICAL: Failed to insert attachment: {:?}", e); + if let Err(e) = engine.put(a_key, &a_data, Codec::Lz4) { + tracing::error!("CRITICAL: Failed to insert attachment blob: {:?}", e); } } - Err(e) => tracing::error!("Fjall attach_ks error: {:?}", e), + Err(e) => tracing::error!("blob engine error: {:?}", e), Ok(true) => { tracing::debug!("Attachment blob already exists (dedup): {}", &a_hash); } @@ -91,57 +113,21 @@ impl BlobManager { } pub fn new() -> Self { - let db = Database::builder(&DATA_DIR_MANAGER.storage_dir) - .cache_size(64 * 1024 * 1024) - .max_cached_files(Some(400)) - .journal_compression(CompressionType::None) - .max_journaling_size(64 * 1024 * 1024) - .open() - .expect("Failed to initialize Fjall database: Check if the directory exists and has write permissions."); + let blob_dir = DATA_DIR_MANAGER.storage_dir.join("blobs"); + let mut config = Config::default(); + config.default_codec = Codec::Zstd; + config.compress_threshold = 1024; + config.flush_interval_secs = 60; + + let engine = Engine::open(&blob_dir, config) + .expect("Failed to initialize blob engine: Check disk space and permissions."); + + let engine = Arc::new(engine); - let email_keyspace = db - .keyspace("email", || { - KeyspaceCreateOptions::default() - .max_memtable_size(16 * 1024 * 1024) - .data_block_size_policy(BlockSizePolicy::all(4 * 1024)) - .data_block_compression_policy( - CompressionPolicy::all(CompressionType::Lz4) - ) - .with_kv_separation(Some( - KvSeparationOptions::default() - .separation_threshold(1024) - .compression(CompressionType::Lz4) - .file_target_size(512 * 1024 * 1024) - .staleness_threshold(0.5) - .age_cutoff(0.6), - )) - }) - .expect("Failed to open 'email' keyspace: The partition metadata might be corrupted or inaccessible."); - - let attachments_keyspace = db - .keyspace("attachments", || { - KeyspaceCreateOptions::default() - .data_block_size_policy(BlockSizePolicy::all(4 * 1024)) - .data_block_compression_policy( - CompressionPolicy::all(CompressionType::Lz4) - ) - .with_kv_separation(Some( - KvSeparationOptions::default() - .separation_threshold(1024) - .compression(CompressionType::Lz4) - .file_target_size(512 * 1024 * 1024) - .staleness_threshold(0.5) - .age_cutoff(0.6), - )) - .max_memtable_size(16 * 1024 * 1024) - }) - .expect("Failed to open 'attachments' keyspace: Check disk space for blob storage initialization."); - let (sender, mut receiver) = mpsc::channel::(100); - let email_ks = email_keyspace.clone(); - let attach_ks = attachments_keyspace.clone(); + let engine_bg = Arc::clone(&engine); let handler = task::spawn(async move { let mut shutdown = SIGNAL_MANAGER.subscribe(); loop { @@ -153,18 +139,17 @@ impl BlobManager { while let Ok(next_eml) = receiver.try_recv() { batch.push(next_eml); } - let email_ks = email_ks.clone(); - let attach_ks = attach_ks.clone(); + let engine_bg = Arc::clone(&engine_bg); if let Err(e) = tokio::task::spawn_blocking(move || { for eml in batch { - Self::process_detached_email(eml, &email_ks, &attach_ks); + Self::process_detached_email(eml, &engine_bg); } }).await { tracing::error!("BlobManager: spawn_blocking join error: {:#?}", e); } } None => { - tracing::info!("BlobManager: All senders dropped, closing storage."); + tracing::info!("BlobManager: All senders dropped, closing blob storage."); break; } } @@ -180,17 +165,16 @@ impl BlobManager { remaining.len() ); if !remaining.is_empty() { - let email_ks = email_ks.clone(); - let attach_ks = attach_ks.clone(); + let engine_bg = Arc::clone(&engine_bg); if let Err(e) = tokio::task::spawn_blocking(move || { for eml in remaining { - Self::process_detached_email(eml, &email_ks, &attach_ks); + Self::process_detached_email(eml, &engine_bg); } }).await { tracing::error!("BlobManager: shutdown spawn_blocking join error: {:#?}", e); } } - tracing::info!("BlobManager: All remaining tasks processed. Closing Fjall."); + tracing::info!("BlobManager: All remaining tasks processed. Closing blob engine."); break; } } @@ -199,30 +183,30 @@ impl BlobManager { Self { sender, - db, - email_keyspace, - attachments_keyspace, + engine, handle: Mutex::new(Some(handler)), } } pub async fn queue(&self, email: DetachedEmail) { if let Err(e) = self.sender.send(email).await { - tracing::error!("BlobManager channel closed, email lost: {:#?}", e); + tracing::error!("BlobManager channel closed, email lost: {:#?}", e); } } pub fn get_email(&self, content_hash: &str) -> BichonResult> { - self.email_keyspace - .get(content_hash) - .map(|user_value| user_value.map(|s| s.into())) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError)) + self.get(content_hash) } pub fn get_attachment(&self, content_hash: &str) -> BichonResult> { - self.attachments_keyspace - .get(content_hash) - .map(|user_value| user_value.map(|s| s.into())) + self.get(content_hash) + } + + fn get(&self, content_hash: &str) -> BichonResult> { + let key = hex_to_key(content_hash)?; + self.engine + .get(&key) + .map(|v| v.map(Bytes::from)) .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError)) } @@ -235,17 +219,23 @@ impl BlobManager { I1: IntoIterator, I1::Item: AsRef, I2: IntoIterator, - I2::Item: AsRef { - let mut batch = self.db.batch(); - for hash in email_content_hashes { - batch.remove(&self.email_keyspace, hash.as_ref()); + I2::Item: AsRef, + { + let mut keys: Vec<[u8; 32]> = email_content_hashes + .into_iter() + .map(|h| hex_to_key(h.as_ref())) + .collect::>()?; + + for h in attachment_content_hashes { + keys.push(hex_to_key(h.as_ref())?); } - for hash in attachment_content_hashes { - batch.remove(&self.attachments_keyspace, hash.as_ref()); + + if !keys.is_empty() { + self.engine + .delete_batch(&keys) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; } - batch - .commit() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + Ok(()) } }