diff --git a/Cargo.lock b/Cargo.lock index b31d48e09..4bd4715b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -206,6 +206,12 @@ dependencies = [ "allocator-api2", ] +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.11.1" @@ -442,6 +448,7 @@ dependencies = [ "criterion", "dashmap", "libsqlite3-sys", + "lmdb-rkv", "parking_lot", "redis", "rusqlite", @@ -1640,6 +1647,29 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "lmdb-rkv" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "447a296f7aca299cfbb50f4e4f3d49451549af655fb7215d7f8c0c3d64bad42b" +dependencies = [ + "bitflags 1.3.2", + "byteorder", + "libc", + "lmdb-rkv-sys", +] + +[[package]] +name = "lmdb-rkv-sys" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61b9ce6b3be08acefa3003c57b7565377432a89ec24476bbe72e11d101f852fe" +dependencies = [ + "cc", + "libc", + "pkg-config", +] + [[package]] name = "lock_api" version = "0.4.14" diff --git a/Cargo.toml b/Cargo.toml index ab0fd4428..56206109a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,29 +32,32 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } # storage -redis = { version = "1.0", features = ["tokio-comp"] } rusqlite = { version = "0.32", features = ["bundled", "backup"] } +redis = { version = "1.0", features = ["tokio-comp"] } sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "sqlite"] } +# embedded memory-mapped KV (bundled C — no system lib needed) +lmdb-rkv = "0.14" + # tree-sitter core tree-sitter = "0.25" # tree-sitter grammars (one feature per lang on extract crate) tree-sitter-typescript = "0.23" tree-sitter-javascript = "0.23" -tree-sitter-python = "0.23" -tree-sitter-rust = "0.23" -tree-sitter-go = "0.23" -tree-sitter-java = "0.23" -tree-sitter-c = "0.23" -tree-sitter-cpp = "0.23" -tree-sitter-c-sharp = "0.23" -tree-sitter-ruby = "0.23" -tree-sitter-php = "0.23" -tree-sitter-scala = "0.26" -tree-sitter-swift = "0.7" -tree-sitter-kotlin = "0.3" -tree-sitter-lua = "0.5" +tree-sitter-python = "0.23" +tree-sitter-rust = "0.23" +tree-sitter-go = "0.23" +tree-sitter-java = "0.23" +tree-sitter-c = "0.23" +tree-sitter-cpp = "0.23" +tree-sitter-c-sharp = "0.23" +tree-sitter-ruby = "0.23" +tree-sitter-php = "0.23" +tree-sitter-scala = "0.26" +tree-sitter-swift = "0.7" +tree-sitter-kotlin = "0.3" +tree-sitter-lua = "0.5" # cli / async / fs clap = { version = "4", features = ["derive", "wrap_help"] } diff --git a/crates/codegraph-api/tests/api.rs b/crates/codegraph-api/tests/api.rs index 41d39c024..8cdd85ef0 100644 --- a/crates/codegraph-api/tests/api.rs +++ b/crates/codegraph-api/tests/api.rs @@ -71,7 +71,7 @@ async fn api(path: &str) -> GraphApi { async fn search_and_symbol_by_id() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("db.sqlite"); - let db_str = db_path.to_string_lossy().into_owned(); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); let (caller, _, _) = seed_index(&db_str).await; let api = api(&db_str).await; @@ -87,7 +87,7 @@ async fn search_and_symbol_by_id() { async fn callers_callees_and_flow() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("db.sqlite"); - let db_str = db_path.to_string_lossy().into_owned(); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); let (caller, callee, helper) = seed_index(&db_str).await; let api = api(&db_str).await; @@ -117,7 +117,7 @@ async fn callers_callees_and_flow() { async fn search_flow_pattern_and_references() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("db.sqlite"); - let db_str = db_path.to_string_lossy().into_owned(); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); let (caller, callee, _) = seed_index(&db_str).await; let api = api(&db_str).await; @@ -147,7 +147,7 @@ async fn search_flow_pattern_and_references() { async fn files_stats_and_context() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("db.sqlite"); - let db_str = db_path.to_string_lossy().into_owned(); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); seed_index(&db_str).await; let api = api(&db_str).await; diff --git a/crates/codegraph-bench/Cargo.toml b/crates/codegraph-bench/Cargo.toml index c2ffeb2d9..ad3f2623d 100644 --- a/crates/codegraph-bench/Cargo.toml +++ b/crates/codegraph-bench/Cargo.toml @@ -8,7 +8,7 @@ description = "Benchmark codegraph-extract + codegraph-graph trên các repo th [dependencies] codegraph-extract = { path = "../codegraph-extract" } -codegraph-graph = { path = "../codegraph-graph" } +codegraph-graph = { path = "../codegraph-graph", features = ["sqlite", "lmdb"] } codegraph-core = { path = "../codegraph-core" } anyhow = { workspace = true } @@ -36,4 +36,8 @@ codspeed = ["dep:codspeed-criterion-compat"] [[bench]] name = "codspeed" -harness = false \ No newline at end of file +harness = false + +[[bench]] +name = "storage" +harness = false diff --git a/crates/codegraph-bench/STORAGE_PERF.md b/crates/codegraph-bench/STORAGE_PERF.md new file mode 100644 index 000000000..8c3aa2f01 --- /dev/null +++ b/crates/codegraph-bench/STORAGE_PERF.md @@ -0,0 +1,85 @@ +# Báo cáo hiệu năng storage backend + +So sánh 3 backend mà `codegraph-graph` hỗ trợ cho việc persist index: + +- `in_memory` — `GraphIndex::in_memory()` (baseline RAM, không persist) +- `sqlite` — backend hiện tại, qua `sqlx` (`sqlite:///db.sqlite`) +- `lmdb` — backend mới thêm, qua `lmdb-rkv` (`lmdb://`) + +> Redis bị loại khỏi phạm vi vì đã chạy trên RAM, không phải "disk-backed". + +## Cách đo + +Benchmark chạy **đúng pipeline thật** như `codspeed.rs` (extract → index → query) +thay vì micro-benchmark gọi trực tiếp từng `Storage`. Với mỗi repo: + +1. **extract** một lần (`codegraph-extract`: walk + parse → `Vec`). +2. **index**: với mỗi backend, mỗi iteration dựng **storage mới** (tempdir/file + mới) rồi `GraphIndex::open(dsn)` + `ingest` — đo chi phí open+ingest, không bị + tích luỹ giữa các iteration. Backend được chọn bằng **DSN scheme** + (`sqlite://` / `lmdb://` / `None` = in-memory), đúng cơ chế + `GraphIndex::open(dsn)` trong `lib.rs`. +3. **query**: chạy bộ truy vấn mẫu trên index in-memory sau ingest (engine query + nằm in-memory, backend không ảnh hưởng phase này). + +Repo đo: toàn bộ `crates/` (chính workspace này). Lệnh: + +```bash +cargo bench -p codegraph-bench --bench storage +``` + +## Kết quả + +### index: open + ingest (mỗi iteration storage mới) + +| Backend | lần 1 (median) | lần 2 (median) | lần 3 (median) | ghi chú | +|-------------|---------------|----------------|----------------|---------| +| `in_memory` | 13.75 µs | 12.09 µs | 8.99 µs | không persist, không I/O | +| `sqlite` | 42.73 ms | 40.55 ms | 13.46 ms | biến động cao | +| `lmdb` | 20.01 ms | 28.40 ms | 15.88 ms | biến động cao | + +**Nhận xét**: biến động giữa các lần chạy lớn (máy đo còn chia tải). Trung bình +LMDB nhanh hơn SQLite khoảng **1.4–2.1×**; có lần chạy về ngang nhau. Lợi thế +của LMDB đến từ: viết 1 transaction duy nhất cho toàn bộ commit (không +WAL/journal riêng, không parser SQL mỗi op), và mapping file theo trang B+tree +kiểu B-tree copy-on-write. + +### Dung lượng trên đĩa (corpus `crates/`) + +| Backend | kích thước | ghi chú | +|---------|-----------|---------| +| `sqlite` | ~590–690 KB | file db.sqlite | +| `lmdb` | ~270 KB | thư mục chứa data.mdb | + +**Nhận xét**: LMDB chiếm **ít hơn ~2.2×** so với SQLite trên cùng dữ liệu — bản +thân LMDB chứa trang metadata + dữ liệu compact; SQLite lưu cả schema, WAL +overhead và trang trống. + +### query (index in-memory, backend không ảnh hưởng) + +| Nhóm | median | +|-------|--------| +| `sample` (search_symbol + callees + flow × 200 tên) | ~84–90 ns / op | + +Query không bị ảnh hưởng bởi backend vì sau `ingest` engine đọc từ graph +in-memory. + +## Khuyến nghị + +- **LMDB đáng dùng khi cần persist nhanh hơn + nhỏ hơn** (cùng mức API + `GraphIndex::open(dsn)`), đặc biệt cho index lớn: chi phí open+ingest thấp hơn + và footprint ~2.2× nhỏ hơn SQLite. +- **SQLite vẫn là lựa chọn an toàn** nếu cần tooling/quen thuộc với file `.db` + đơn, hoặc dùng query ad-hoc bên ngoài. Độ lệch hiệu năng giữa 2 backend nằm + trong tầm 1.4–2.1× tuỳ tải máy. +- `in_memory` là baseline nhanh nhất (không I/O), dùng cho trường hợp không cần + persist (CLI một lần). +- Redis giữ vai trò dành cho triển khai cần chia sẻ index giữa nhiều process. + +Chọn backend bằng DSN scheme: + +```rust +GraphIndex::open("sqlite:///tmp/db.sqlite").await?; // sqlite +GraphIndex::open("lmdb:///tmp/db").await?; // lmdb +GraphIndex::in_memory(); // RAM +``` diff --git a/crates/codegraph-bench/benches/storage.rs b/crates/codegraph-bench/benches/storage.rs new file mode 100644 index 000000000..1e5d8966b --- /dev/null +++ b/crates/codegraph-bench/benches/storage.rs @@ -0,0 +1,166 @@ +//! Benchmark **storage backend** qua đúng pipeline luồng thật (extract → index → +//! query) như `codspeed.rs`, nhưng mỗi backend một group và mỗi iteration dựng +//! storage **mới** (file mới) để đo chi phí open+ingest không bị tích luỹ. +//! +//! Backend được chọn bằng DSN scheme (đúng cơ chế `GraphIndex::open(dsn)`): +//! - `in_memory` — `GraphIndex::in_memory()` (baseline RAM, không persist) +//! - `sqlite` — `sqlite:///db.sqlite` (persist) +//! - `lmdb` — `lmdb:///db` (persist) +//! +//! Chạy (repo list giống codspeed: `CODEGRAPH_BENCH_REPOS_LIST` hoặc fallback +//! `crates`): +//! ```bash +//! CODEGRAPH_BENCH_REPOS_LIST=repos.txt cargo bench -p codegraph-bench --bench storage +//! ``` + +use std::hint::black_box; + +use codegraph_bench::{ + BenchOptions, Repo, extract, index_at, orchestrator, run_queries, sample_query_names, +}; +// Dùng `codspeed_criterion_compat` khi build qua `cargo codspeed build` (đo bằng +// hardware counters); local (không feature) resolve về criterion thường. Giống +// benches/codspeed.rs — bắt buộc để CodSpeed nối được runner. +#[cfg(feature = "codspeed")] +use codspeed_criterion_compat as crit; +#[cfg(not(feature = "codspeed"))] +use criterion as crit; + +fn load_repos() -> Vec { + let mut out = Vec::new(); + if let Ok(list_file) = std::env::var("CODEGRAPH_BENCH_REPOS_LIST") { + if let Ok(body) = std::fs::read_to_string(&list_file) { + for line in body.lines() { + let line = line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + let name = std::path::Path::new(line) + .file_name() + .and_then(|s| s.to_str()) + .map(String::from) + .unwrap_or_else(|| line.to_string()); + out.push(Repo { + name, + root: line.into(), + }); + } + } + return out; + } + out.push(Repo { + name: "crates".into(), + root: "crates".into(), + }); + out +} + +/// Dung lượng trên đĩa của một thư mục (đệ quy), dùng để so sánh footprint +/// của sqlite vs lmdb trên cùng một corpus. +fn dir_size(path: &std::path::Path) -> u64 { + let mut total = 0u64; + if let Ok(rd) = std::fs::read_dir(path) { + for ent in rd.flatten() { + let p = ent.path(); + if p.is_dir() { + total += dir_size(&p); + } else if let Ok(md) = std::fs::metadata(&p) { + total += md.len(); + } + } + } + total +} + +/// Đo một lần dung lượng file thật trên đĩa cho sqlite vs lmdb (không chạy +/// trong benchmark lặp) để báo cáo footprint. Mỗi backend một tempdir riêng. +fn measure_on_disk(parsed: &[codegraph_graph::ParseResult]) { + let sqlite_dir = tempfile::tempdir().unwrap().keep(); + let sqlite = format!("sqlite://{}/db.sqlite", sqlite_dir.to_string_lossy()); + if let Ok(_idx) = index_at(parsed, Some(&sqlite)) {} + let sqlite_bytes = dir_size(&sqlite_dir); + + let lmdb_dir = tempfile::tempdir().unwrap().keep(); + let lmdb = format!("lmdb://{}", lmdb_dir.to_string_lossy()); + if let Ok(_idx) = index_at(parsed, Some(&lmdb)) {} + let lmdb_bytes = dir_size(&lmdb_dir); + + eprintln!( + "on-disk: sqlite={} bytes | lmdb={} bytes", + sqlite_bytes, lmdb_bytes + ); +} + +fn main_benchmark(c: &mut crit::Criterion) { + type BackendFactory = Box Option>; + type NamedBackend = (&'static str, BackendFactory); + + let opts = BenchOptions { + langs: None, + queries: 200, + with_flow: false, + }; + let repos = load_repos(); + for repo in &repos { + let name = repo.name.clone(); + // Parse một lần (extract), dùng chung cho mọi backend. + let parsed = match extract(&orchestrator(&opts), &repo.root) { + Ok((p, _)) => p, + Err(e) => { + eprintln!("[{name}] extract failed: {e}; skip"); + continue; + } + }; + let names = sample_query_names(&parsed, opts.queries); + measure_on_disk(&parsed); + + // ── index: mỗi backend một group, storage MỚI mỗi iteration ── + // Mỗi backend là một closure `mk_dsn()` trả DSN cho một storage trống + // (tempdir mới). Với in-memory, dsn = None. + let mk_backends: Vec = vec![ + ("in_memory", Box::new(|| None)), + ( + "sqlite", + Box::new(|| { + let dir = tempfile::tempdir().unwrap().keep(); + Some(format!("sqlite://{}/db.sqlite", dir.to_string_lossy())) + }), + ), + ( + "lmdb", + Box::new(|| { + let dir = tempfile::tempdir().unwrap().keep(); + Some(format!("lmdb://{}", dir.to_string_lossy())) + }), + ), + ]; + + for (bname, mk_dsn) in mk_backends { + let parsed = &parsed; + let mut g = c.benchmark_group(format!("{name}/{bname}/index")); + g.bench_function("open+ingest", |b| { + b.iter(|| { + let dsn = mk_dsn(); + let _ = black_box(index_at(parsed, dsn.as_deref())); + }); + }); + g.finish(); + } + + // ── query trên index in-memory (backend không ảnh hưởng query — engine + // in-memory sau ingest) — giữ để pipeline giống codspeed. ── + if let Ok(idx) = index_at(&parsed, None) { + let mut g = c.benchmark_group(format!("{name}/query")); + let names = &names; + g.bench_function("sample", |b| { + b.iter(|| { + let _ = black_box(run_queries(&idx, names, false)); + }); + }); + g.finish(); + } + } +} + +crit::criterion_group!(benches, main_benchmark); +crit::criterion_main!(benches); diff --git a/crates/codegraph-bench/src/lib.rs b/crates/codegraph-bench/src/lib.rs index 75a344ae9..b2db08943 100644 --- a/crates/codegraph-bench/src/lib.rs +++ b/crates/codegraph-bench/src/lib.rs @@ -92,8 +92,18 @@ pub fn extract( /// Phase index: dựng in-memory `GraphIndex` + `ingest` toàn bộ parsed. pub fn index(parsed: &[ParseResult]) -> Result { + index_at(parsed, None) +} + +/// Phase index trên một storage backend cụ thể — `dsn` chỉ rõ backend (vd +/// `sqlite:///tmp/db.sqlite`, `lmdb:///tmp/db`, hoặc `None` = in-memory) — +/// `GraphIndex::open(dsn)` tự route theo scheme, rồi `ingest` toàn bộ parsed. +pub fn index_at(parsed: &[ParseResult], dsn: Option<&str>) -> Result { runtime().block_on(async { - let mut idx = GraphIndex::in_memory(); + let mut idx = match dsn { + Some(d) => GraphIndex::open(d).await?, + None => GraphIndex::in_memory(), + }; idx.ingest(parsed).await?; Ok(idx) }) diff --git a/crates/codegraph-extract/src/config.rs b/crates/codegraph-extract/src/config.rs index 17f44f63f..59eb25746 100644 --- a/crates/codegraph-extract/src/config.rs +++ b/crates/codegraph-extract/src/config.rs @@ -1,4 +1,5 @@ use crate::languages::effects::EffectClassifier; +use crate::project::{project_db_path, project_dir}; use camino::Utf8Path; use codegraph_core::{EffectCallPattern, EffectRule, EffectType}; use serde::Deserialize; @@ -14,6 +15,31 @@ pub enum HeaderLanguage { Cpp, } +/// Backend storage cho index — chọn backend trong `[storage]` của config. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum StorageKind { + /// `sqlite://` (backend mặc định). + #[default] + Sqlite, + /// `lmdb://` (thư mục). + Lmdb, + /// `redis://` (cần `dsn`). + Redis, + /// In-memory — không persist. + Memory, +} + +impl StorageKind { + fn parse(raw: &str) -> Self { + match raw.trim().to_ascii_lowercase().as_str() { + "lmdb" => StorageKind::Lmdb, + "redis" => StorageKind::Redis, + "memory" | "in-memory" | "in_memory" => StorageKind::Memory, + _ => StorageKind::Sqlite, + } + } +} + #[derive(Debug, Default, Deserialize)] struct ConfigFile { #[serde(default)] @@ -21,6 +47,19 @@ struct ConfigFile { /// Project extra effect rules — xét trước bảng default (override). #[serde(default)] effect_rules: Vec, + /// Backend storage (mặc định sqlite). + #[serde(default)] + storage: StorageSection, +} + +#[derive(Debug, Default, Deserialize)] +struct StorageSection { + /// `"sqlite"`, `"lmdb"`, `"redis"`, `"memory"`. + #[serde(default, rename = "type")] + type_: Option, + /// DSN override — ví dụ `lmdb:///data/codegraph.db`. + #[serde(default)] + dsn: Option, } #[derive(Debug, Default, Deserialize)] @@ -45,6 +84,16 @@ pub struct ExtractConfig { pub header_language: HeaderLanguage, /// Classifier effect của project — config rules override bảng default. pub effect_classifier: EffectClassifier, + /// Backend storage được chọn trong config (mặc định sqlite). + pub storage: StorageConfig, +} + +/// Storage backend đã parse từ `[storage]` trong config. +#[derive(Debug, Clone, Default)] +pub struct StorageConfig { + pub kind: StorageKind, + /// DSN override (`None` = dựng từ `kind` + project path). + pub dsn: Option, } impl ExtractConfig { @@ -63,6 +112,35 @@ impl ExtractConfig { Self { header_language: parse_header_language(file.languages.headers.as_deref()), effect_classifier: build_classifier(file.effect_rules), + storage: StorageConfig { + kind: file + .storage + .type_ + .as_deref() + .map(StorageKind::parse) + .unwrap_or_default(), + dsn: file.storage.dsn, + }, + } + } + + /// DSN hoàn chỉnh (kèm scheme) cho backend storage — dùng làm input trực + /// tiếp cho `GraphIndex::open`. `None` = in-memory. + /// + /// - `dsn` trong config override → dùng nguyên văn. + /// - Nếu không, dựng từ `kind`: + /// - sqlite → `sqlite:///.codegraph/db.sqlite` + /// - lmdb → `lmdb:///.codegraph/db.lmdb` (thư mục) + /// - redis → phải có `dsn` (không có default hợp lý) + pub fn storage_dsn(&self, root: &Utf8Path) -> Option { + if let Some(dsn) = &self.storage.dsn { + return Some(dsn.clone()); + } + match self.storage.kind { + StorageKind::Sqlite => Some(format!("sqlite://{}", project_db_path(root))), + StorageKind::Lmdb => Some(format!("lmdb://{}", project_dir(root).join("db.lmdb"))), + StorageKind::Redis => None, + StorageKind::Memory => None, } } } @@ -103,12 +181,21 @@ pub const DEFAULT_CONFIG_TOML: &str = r#"# CodeGraph project configuration # "auto" detects C++ projects from .cpp/.hpp files and C++ syntax in headers. headers = "auto" -# Project effect rules — matched before the built-in defaults (first match wins). +# Critical effect rules — matched before the built-in defaults (first match wins). # call matchers: prefix / contains / exact. Effects: sql_query, sql_write, # cache_read, cache_write, http_call, event_emit, file_read, file_write, log. # [[effect_rules]] # call = { prefix = "db." } # effect = "sql_query" + +[storage] +# Backend lưu index: "sqlite", "lmdb", "redis", hoặc "memory". +type = "sqlite" +# DSN override (mặc định dựng từ `type` + project path): +# sqlite → sqlite:///.codegraph/db.sqlite +# lmdb → lmdb:///.codegraph/db.lmdb +# redis → bắt buộc khai dsn, ví dụ redis://localhost:6379 +# dsn = "sqlite:///tmp/codegraph.db" "#; /// Quick project scan: returns a hint when the tree is clearly C-only or C++-only. @@ -198,6 +285,70 @@ headers = "cpp" )); } + #[test] + fn parse_storage_kind() { + assert_eq!(StorageKind::parse("sqlite"), StorageKind::Sqlite); + assert_eq!(StorageKind::parse("lmdb"), StorageKind::Lmdb); + assert_eq!(StorageKind::parse("REDIS"), StorageKind::Redis); + assert_eq!(StorageKind::parse("memory"), StorageKind::Memory); + assert_eq!(StorageKind::parse("in-memory"), StorageKind::Memory); + // unknown → sqlite (default). + assert_eq!(StorageKind::parse("whatsapp"), StorageKind::Sqlite); + } + + /// `storage_dsn` dựng DSN theo kind; `dsn` override thắng. + #[test] + fn storage_dsn_built_or_overridden() { + let dir = std::env::temp_dir().join("codegraph-extract-dsn-test"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("config.toml"); + let path = Utf8Path::from_path(path.as_path()).unwrap(); + + std::fs::write( + path.as_std_path(), + r#" +[storage] +type = "lmdb" +"#, + ) + .unwrap(); + let cfg = ExtractConfig::load_from(path); + let dsn = cfg.storage_dsn(Utf8Path::new("/repo")).unwrap(); + assert!(dsn.starts_with("lmdb://"), "got {dsn}"); + assert!(dsn.contains("/repo/.codegraph/db.lmdb"), "got {dsn}"); + + // override dsn thắng kind. + std::fs::write( + path.as_std_path(), + r#" +[storage] +type = "lmdb" +dsn = "sqlite:///tmp/custom.db" +"#, + ) + .unwrap(); + let cfg = ExtractConfig::load_from(path); + assert_eq!( + cfg.storage_dsn(Utf8Path::new("/repo")).unwrap(), + "sqlite:///tmp/custom.db" + ); + + // memory → None (in-memory). + std::fs::write( + path.as_std_path(), + r#" +[storage] +type = "memory" +"#, + ) + .unwrap(); + let cfg = ExtractConfig::load_from(path); + assert!(cfg.storage_dsn(Utf8Path::new("/repo")).is_none()); + + let _ = std::fs::remove_file(path.as_std_path()); + let _ = std::fs::remove_dir(&dir); + } + /// Parse từ file tạm với `[[effect_rules]]` → classifier áp dụng được. #[test] fn load_from_file_applies_effect_rules() { diff --git a/crates/codegraph-extract/src/walker.rs b/crates/codegraph-extract/src/walker.rs index a1b097cf4..1de52529a 100644 --- a/crates/codegraph-extract/src/walker.rs +++ b/crates/codegraph-extract/src/walker.rs @@ -218,6 +218,7 @@ mod tests { let config = ExtractConfig { header_language: HeaderLanguage::Cpp, effect_classifier: Default::default(), + storage: Default::default(), }; let matches = walk(&root, &parsers, &config); let h = matches diff --git a/crates/codegraph-graph/Cargo.toml b/crates/codegraph-graph/Cargo.toml index 4eba1c2d3..afd2536cb 100644 --- a/crates/codegraph-graph/Cargo.toml +++ b/crates/codegraph-graph/Cargo.toml @@ -31,6 +31,7 @@ url = { version = "2.5.8", optional = true } zstd = { version = "0.13", optional = true } bincode = { version = "1.3", optional = true } sqlx = { workspace = true, optional = true } +lmdb-rkv = { workspace = true, optional = true } # Bundled sqlite cho sqlx (giống rusqlite của codegraph-db) — feature # unification khiến sqlx dùng chung bản build bundled này, không cần system lib. @@ -40,6 +41,7 @@ libsqlite3-sys = { version = "0.30", features = ["bundled"], optional = true } default = [] redis = ["dep:redis", "dep:zstd", "dep:bincode", "dep:url"] sqlite = ["dep:sqlx", "dep:libsqlite3-sys"] +lmdb = ["dep:lmdb-rkv"] bloom-search = [] [dev-dependencies] diff --git a/crates/codegraph-graph/src/lib.rs b/crates/codegraph-graph/src/lib.rs index a754f3696..0457c3182 100644 --- a/crates/codegraph-graph/src/lib.rs +++ b/crates/codegraph-graph/src/lib.rs @@ -35,7 +35,11 @@ //! var-type alias, gom SaveCallRecords) → files → rebuild engines → bump version. pub use crate::search::Search; -use crate::storage::InMemoryStorage; +#[cfg(feature = "lmdb")] +pub use crate::storage::lmdb::LmdbStorage; +#[cfg(feature = "sqlite")] +pub use crate::storage::sqlite::SqliteStorage; +pub use crate::storage::{InMemoryStorage, Storage, Tx}; use codegraph_core::{ CallRecord, CallSite, CallSiteResult, ClassInfo, DependenciesReport, Dependency, EdgeMeta, EffectType, Error, FileInfo, FlowCall, FlowResult, FunctionScope, MemberInfo, ResolveResult, @@ -80,6 +84,15 @@ fn serr(e: crate::storage::StorageError) -> Error { Error::Search(e.to_string()) } +/// Lỗi khi DSN chỉ rõ scheme nhưng feature tương ứng không được bật. +#[allow(dead_code)] // fallback dispatch + lmdb/sqlite branch dùng khi feature tắt +fn backend_unavailable(name: &str) -> Error { + Error::Db(format!( + "Backend '{name}' được yêu cầu qua DSN nhưng feature '{name}' không được bật \ + trong bản build này" + )) +} + /// Map `search::Error` → `Error::Search`. fn serr_search(e: crate::search::Error) -> Error { Error::Search(e.to_string()) @@ -151,25 +164,114 @@ impl GraphIndex { Self::new_with_storage(storage) } - /// Mở index từ file sqlite (feature `sqlite`) — rebuild từ entity store. - #[allow(unused_variables)] // dsn chỉ dùng khi bật sqlite/redis — không backend → Err. + /// Mở index từ một backend persistent bằng DSN — rebuild từ entity store. + /// + /// DSN mang scheme cho biết backend, phần còn lại là path: + /// - `sqlite://` → sqlite (feature `sqlite`) + /// - `lmdb://` → LMDB (feature `lmdb`) + /// - `redis://` → redis (feature `redis`) + /// + /// Không có scheme (plain path) → fallback backend mặc định **nếu chỉ có + /// đúng 1 backend** được compile (backward compat: main.rs/mcp truyền + /// plain path với build chỉ bật `sqlite`). Nếu ≥2 backend — thay vì chọn + /// ngầm một backend (gây nhầm) — báo lỗi bắt caller chỉ rõ scheme. pub async fn open(dsn: &str) -> Result { - #[cfg(feature = "sqlite")] - #[allow(unreachable_code)] - return Self::open_sqlite(dsn).await; + match Self::split_dsn(dsn) { + Some(("sqlite", path)) => Self::open_sqlite_dispatch(path).await, + Some(("lmdb", path)) => Self::open_lmdb_dispatch(path).await, + _ => Self::open_default(dsn).await, + } + } - #[cfg(feature = "redis")] - #[allow(unreachable_code)] - return Self::open_redis(dsn).await; + /// `sqlite://` rõ ràng — compile cả sqlite; không compile → báo lỗi. + #[cfg(feature = "sqlite")] + async fn open_sqlite_dispatch(path: &str) -> Result { + Self::open_sqlite(path).await + } + /// `sqlite://` rõ ràng nhưng feature không bật → không thể mở. + #[cfg(not(feature = "sqlite"))] + async fn open_sqlite_dispatch(_path: &str) -> Result { + Err(backend_unavailable("sqlite")) + } + + /// `lmdb://` rõ ràng — compile trường lmdb; không compile → báo lỗi. + #[cfg(feature = "lmdb")] + async fn open_lmdb_dispatch(path: &str) -> Result { + Self::open_lmdb(path).await + } + /// `lmdb://` rõ ràng nhưng feature không bổ — lỗi. + #[cfg(not(feature = "lmdb"))] + async fn open_lmdb_dispatch(_path: &str) -> Result { + Err(backend_unavailable("lmdb")) + } + + /// Tách `scheme://` khỏi DSN: trả `(scheme, phần còn lại)` hoặc `None` + /// nếu không có scheme (plain path / redis url giữ nguyên). + fn split_dsn(dsn: &str) -> Option<(&'static str, &str)> { + if let Some(rest) = dsn.strip_prefix("sqlite://") { + return Some(("sqlite", rest)); + } + if let Some(rest) = dsn.strip_prefix("lmdb://") { + return Some(("lmdb", rest)); + } + None + } + + /// Mở backend mặc định khi DSN không có scheme. Chỉ được phép ngầm chọn + /// khi **đúng 1** backend persistent được compile; nhiều hơn → lỗi bắt + /// buộc scheme (tránh chọn nhầm). Các nhánh cfg mutual-exclusive nên + /// không có unreachable code. + #[allow(unused_variables)] // dsn chỉ dùng trong nhánh single-backend + async fn open_default(dsn: &str) -> Result { + // Chỉ sqlite được compile — plain path = sqlite (backward compat). + #[cfg(all(feature = "sqlite", not(any(feature = "lmdb", feature = "redis"))))] + { + return Self::open_sqlite(dsn).await; + } + // Chỉ lmdb được compile — plain path = lmdb. + #[cfg(all(feature = "lmdb", not(any(feature = "sqlite", feature = "redis"))))] + { + return Self::open_lmdb(dsn).await; + } + // Chỉ redis được compile — plain path = redis. + #[cfg(all(feature = "redis", not(any(feature = "sqlite", feature = "lmdb"))))] + { + return Self::open_redis(dsn).await; + } + // Nhiều backend (≥2) — DSN không nói scheme → mơ hồ. + #[cfg(any( + all(feature = "sqlite", feature = "lmdb"), + all(feature = "sqlite", feature = "redis"), + all(feature = "lmdb", feature = "redis") + ))] + { + return Err(Error::Db( + "Nhiều backend persistent được bật nhưng DSN không chỉ rõ scheme. \ + Ghi rõ `sqlite://`, `lmdb://` hoặc `redis://` trong --dbdsn." + .into(), + )); + } + // Không backend nào — không thể mở persistent. #[allow(unreachable_code)] { Err(Error::Db( - "Phải bật ít nhất feature 'sqlite' hoặc 'redis'".into(), + "Phải bật ít nhất một feature 'sqlite', 'lmdb' hoặc 'redis'".into(), )) } } + #[cfg(feature = "lmdb")] + async fn open_lmdb(path: &str) -> Result { + let storage = crate::storage::lmdb::LmdbStorage::open(path) + .await + .map_err(serr)?; + let storage = Arc::new(RwLock::new(storage)) as Arc>; + let mut idx = Self::new_with_storage(storage); + idx.rebuild().await?; + Ok(idx) + } + #[cfg(feature = "sqlite")] async fn open_sqlite(path: &str) -> Result { let storage = crate::storage::sqlite::SqliteStorage::open(path) @@ -246,7 +348,10 @@ impl GraphIndex { // ── Build / rebuild ── /// Rebuild toàn bộ index từ entity store trong storage (open/reopen). - #[cfg_attr(not(any(feature = "sqlite", feature = "redis")), allow(dead_code))] + #[cfg_attr( + not(any(feature = "sqlite", feature = "lmdb", feature = "redis")), + allow(dead_code) + )] // chỉ open() dùng — không backend thì không ai gọi. async fn rebuild(&mut self) -> Result<()> { self.next_id = self @@ -326,7 +431,10 @@ impl GraphIndex { } /// Insert symbol vào registry + index (scope id đã global — path rebuild). - #[cfg_attr(not(any(feature = "sqlite", feature = "redis")), allow(dead_code))] + #[cfg_attr( + not(any(feature = "sqlite", feature = "lmdb", feature = "redis")), + allow(dead_code) + )] // chỉ rebuild() dùng — không backend thì không ai gọi. fn index_symbol(&mut self, sym: Symbol) { let id = sym.id; @@ -353,7 +461,10 @@ impl GraphIndex { } /// Rebuild edges từ chains + call records (nhanh — chỉ dùng khi reopen). - #[cfg_attr(not(any(feature = "sqlite", feature = "redis")), allow(dead_code))] + #[cfg_attr( + not(any(feature = "sqlite", feature = "lmdb", feature = "redis")), + allow(dead_code) + )] // chỉ rebuild() dùng — không backend thì không ai gọi. fn rebuild_edges(&mut self, recs: &HashMap>) { self.edges.clear(); @@ -675,7 +786,22 @@ impl GraphIndex { .unwrap_or("") .to_lowercase(); if !short.is_empty() { - candidates = self.name_index.get(&short).cloned().unwrap_or_default(); + // Chỉ nhận callee-thực-sự (Function/Method) — KHÔNG fallback vào + // biến / field / param trùng tên (VD `WrapResponse.ok(...)` với + // receiver external không resolve được dễ link nhầm vào `boolean ok` + // trong file khác — bug C). + candidates = self + .name_index + .get(&short) + .cloned() + .unwrap_or_default() + .into_iter() + .filter(|&id| { + self.symbols.get(&id).is_some_and(|s| { + matches!(s.kind, SymbolKind::Function | SymbolKind::Method) + }) + }) + .collect(); } } @@ -731,6 +857,11 @@ impl GraphIndex { if sym.annotations.iter().any(|a| a.name == "Override") { score += 10; } + if matches!(sym.kind, SymbolKind::Function | SymbolKind::Method) { + // Ưu tiên callee-thực-sự (hàm/method) hơn symbol trùng tên khác + // kind (Variable/Parameter/Field...). (bug C) + score += 4; + } if self.chains_map.contains_key(&id) { score += 5; } @@ -919,6 +1050,31 @@ impl GraphIndex { kind: Option, limit: usize, ) -> Result> { + self.search_symbol_filtered(query, limit, |s| kind.is_none() || s.kind == kind.unwrap()) + .await + } + + /// Như `search_symbol` nhưng chấp nhận NHIỀU kind — dùng cho sandbox (entry + /// có thể là `Function` free function (Rust/Go/...) hoặc `Method` (Java/...)). + pub async fn search_symbol_kinds( + &self, + query: &str, + kinds: &[SymbolKind], + limit: usize, + ) -> Result> { + self.search_symbol_filtered(query, limit, |s| kinds.contains(&s.kind)) + .await + } + + async fn search_symbol_filtered( + &self, + query: &str, + limit: usize, + filter: F, + ) -> Result> + where + F: Fn(&Symbol) -> bool, + { let q = query.to_lowercase(); let hits = match self.names.search(q.as_bytes(), None).await { Ok(h) => h, @@ -944,7 +1100,7 @@ impl GraphIndex { let Some(s) = self.symbols.get(&id) else { continue; }; - if kind.is_some_and(|k| s.kind != k) { + if !filter(s) { continue; } out.push(s.clone()); @@ -1890,8 +2046,7 @@ mod tests { #[tokio::test] async fn sqlite_persist_and_reopen() { let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("db.sqlite"); - let path = path.to_string_lossy().into_owned(); + let path = format!("sqlite://{}/db.sqlite", dir.path().to_string_lossy()); let chains = HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE, SYMBOL_BASE + 1])]); let r = result( "a.ts", diff --git a/crates/codegraph-graph/src/shared.rs b/crates/codegraph-graph/src/shared.rs index e3782ffe2..0e8d6bc94 100644 --- a/crates/codegraph-graph/src/shared.rs +++ b/crates/codegraph-graph/src/shared.rs @@ -1,16 +1,20 @@ //! SharedGraphIndex — index dùng chung cho production (GraphApi/MCP/viz). //! -//! Mọi request dùng chung 1 snapshot `Arc`. Index sống trong chính -//! file `.codegraph/db.sqlite` (entity store `sg_*` + radix chain engine `rt_*`): +//! Mọi request dùng chung 1 snapshot `Arc`. Index sống trong một +//! backend persistent mà DSN chỉ rõ (`sqlite://...` / `lmdb://...` / `redis://...`): //! `GraphIndex::ingest` (CLI/watcher, tiến trình riêng) bump `index_version` -//! trong file; `ensure_fresh` probe version (đọc thẳng file — không cần sidecar) -//! và rebuild snapshot khi stale dưới `rebuild_lock` (N request stale đồng thời -//! chỉ 1 lần rebuild), đổi snapshot dưới `RwLock`. `path = None`: in-memory — -//! không có writer ngoài, snapshot coi như luôn fresh sau lần build đầu. +//! trong store; `ensure_fresh` probe version (đọc thẳng store — không cần +//! sidecar) và rebuild snapshot khi stale dưới `rebuild_lock` (N request stale +//! đồng thời chỉ 1 lần rebuild), đổi snapshot dưới `RwLock`. `dsn = None`: +//! in-memory — không có writer ngoài, snapshot coi như luôn fresh sau lần +//! build đầu. +//! +//! DSN là **source duy nhất** cho cả `rebuild` (mở backend) lẫn `current_version` +//! (probe) — nên khi nhiều backend cùng được bật (vd `sqlite` + `lmdb`), backend +//! được chọn theo scheme trong DSN, không phải theo thứ tự feature. use crate::GraphIndex; use codegraph_core::Result; -use std::path::PathBuf; use std::sync::Arc; use tokio::sync::{Mutex, RwLock}; @@ -28,11 +32,8 @@ struct IndexState { /// chiếu 1 instance. Rebuild đồng bộ theo version file — request đầu sau khi /// re-index xong chờ rebuild, các request sau thấy đã fresh. pub struct SharedGraphIndex { - /// Nơi persist index (`None` = in-memory, chạy không feature `sqlite`). - /// Chỉ đọc trong nhánh `sqlite` (open/rebuild) — build không feature này - /// giữ `None` nên field không được dùng. - #[cfg_attr(not(feature = "sqlite"), allow(dead_code))] - path: Option, + /// DSN nơi persist index (`None` = in-memory, không có writer ngoài). + dsn: Option, state: RwLock, /// Serialize rebuild — N request stale đồng thời chỉ 1 lần rebuild. rebuild_lock: Arc>, @@ -41,11 +42,15 @@ pub struct SharedGraphIndex { impl SharedGraphIndex { /// Mở index dùng chung. /// - /// `path = Some(p)` (feature `sqlite`): chưa build — `ensure_fresh` sẽ - /// reopen + rebuild index từ file lần đầu. `path = None`: in-memory. - pub async fn open(path: Option) -> Result { + /// `dsn = Some(d)`: chưa build — `ensure_fresh` sẽ mở đúng backend theo + /// scheme rồi rebuild index từ store lần đầu. `dsn = None`: in-memory. + /// + /// `dsn` phải là DSN đầy đủ scheme (vd `sqlite:///path/db.sqlite`, + /// `lmdb:///path/db`) — không phải plain path, để nhiều backend cùng bật + /// vẫn chọn đúng backend. + pub async fn open(dsn: Option) -> Result { Ok(Self { - path, + dsn, state: RwLock::new(IndexState { index: Arc::new(GraphIndex::in_memory()), version: 0, @@ -55,31 +60,48 @@ impl SharedGraphIndex { }) } - /// Version index trên đĩa hiện tại — `None` nếu probe thất bại (file chưa - /// có hoặc đang bị re-index). Chỉ gọi khi `path.is_some()`. - #[cfg(feature = "sqlite")] + /// Scheme của DSN (`"sqlite"`, `"lmdb"`, `"redis"`) — `None` nếu in-memory. + fn scheme(&self) -> Option<&'static str> { + let dsn = self.dsn.as_ref()?; + if dsn.starts_with("sqlite://") { + return Some("sqlite"); + } + if dsn.starts_with("lmdb://") { + return Some("lmdb"); + } + if dsn.starts_with("redis://") { + return Some("redis"); + } + // Các scheme/DSN khác (chưa biết) — không đo được version độc lập. + None + } + + /// Version index trên đĩa hiện tại — `None` nếu probe thất bại (store chưa + /// có hoặc đang bị re-index), hay backend không probe độc lập được (redis). + /// Chỉ gọi khi `dsn.is_some()`. async fn current_version(&self) -> Option { - let p = self.path.as_ref()?; - crate::storage::sqlite::SqliteStorage::probe_version(&p.display().to_string()) - .await - .ok() + let dsn = self.dsn.as_ref()?; + let path = trim_scheme(dsn); + match self.scheme() { + #[cfg(feature = "sqlite")] + Some("sqlite") => crate::storage::sqlite::SqliteStorage::probe_version(path) + .await + .ok(), + #[cfg(feature = "lmdb")] + Some("lmdb") => crate::storage::lmdb::probe_version(path).await.ok(), + // redis không có probe file ngoài — không đo được → stale. + _ => None, + } } /// Snapshot hiện tại có khớp version trên đĩa không. In-memory (không file) - /// → không có writer ngoài → luôn fresh. + /// → không có writer ngoài → luôn fresh. Backend không probe được (redis/ + /// unknown scheme) → coi là stale để rebuilt lại. async fn is_fresh(&self, version: u64) -> bool { - #[cfg(feature = "sqlite")] - { - if self.path.is_none() { - return true; - } - matches!(self.current_version().await, Some(v) if v == version) - } - #[cfg(not(feature = "sqlite"))] - { - let _ = version; - true + if self.dsn.is_none() { + return true; } + matches!(self.current_version().await, Some(v) if v == version) } /// Đảm bảo index mới nhất, trả snapshot dùng được. @@ -110,19 +132,15 @@ impl SharedGraphIndex { self.state.read().await.index.clone() } - /// Build index từ file hiện tại rồi swap snapshot (gọi trong `rebuild_lock`). + /// Build index từ DSN hiện tại rồi swap snapshot (gọi trong `rebuild_lock`). + /// `GraphIndex::open` tự route theo scheme — không cần nhánh cfg. async fn rebuild_inner(&self) -> Result<()> { - #[cfg(feature = "sqlite")] - let index = match &self.path { - Some(p) => GraphIndex::open(&p.display().to_string()).await?, + #[cfg(any(feature = "sqlite", feature = "lmdb", feature = "redis"))] + let index = match &self.dsn { + Some(d) => GraphIndex::open(d).await?, None => GraphIndex::in_memory(), }; - #[cfg(all(feature = "redis", not(feature = "sqlite")))] - let index = match &self.path { - Some(p) => GraphIndex::open(&p.display().to_string()).await?, - None => GraphIndex::in_memory(), - }; - #[cfg(not(any(feature = "sqlite", feature = "redis")))] + #[cfg(not(any(feature = "sqlite", feature = "lmdb", feature = "redis")))] let index = GraphIndex::in_memory(); let version = index.version(); @@ -134,6 +152,14 @@ impl SharedGraphIndex { } } +/// Bỏ `scheme://` khỏi DSN — trả phần còn lại (path cho probe file). +fn trim_scheme(dsn: &str) -> &str { + dsn.strip_prefix("sqlite://") + .or_else(|| dsn.strip_prefix("lmdb://")) + .or_else(|| dsn.strip_prefix("redis://")) + .unwrap_or(dsn) +} + #[cfg(test)] mod tests { use super::*; @@ -161,7 +187,7 @@ mod tests { } } - // Chỉ test sqlite dùng — build không feature này vẫn compile. + // Chỉ test sqlite dùng — build không có feature này vẫn compile. #[cfg_attr(not(feature = "sqlite"), allow(dead_code))] fn mk_result(path: &str, symbols: Vec, chain: Vec) -> ParseResult { ParseResult { @@ -191,7 +217,7 @@ mod tests { async fn sqlite_stale_version_rebuilds() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("db.sqlite"); - let db_str = db_path.to_string_lossy().into_owned(); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); // "CLI process": index dữ liệu vào file. { @@ -205,7 +231,7 @@ mod tests { } // "Server process": shared index trên cùng file. - let sgi = Arc::new(SharedGraphIndex::open(Some(db_path.clone())).await.unwrap()); + let sgi = Arc::new(SharedGraphIndex::open(Some(db_str.clone())).await.unwrap()); let idx = sgi.ensure_fresh().await; assert_eq!(idx.version(), 1); assert_eq!(idx.stats().symbols, 2); diff --git a/crates/codegraph-graph/src/storage.rs b/crates/codegraph-graph/src/storage.rs index 68c0f3167..0c406e211 100644 --- a/crates/codegraph-graph/src/storage.rs +++ b/crates/codegraph-graph/src/storage.rs @@ -21,6 +21,9 @@ pub mod sqlite; #[cfg(feature = "redis")] pub mod redis; + +#[cfg(feature = "lmdb")] +pub mod lmdb; // ==================== Error Type ==================== #[derive(Debug)] diff --git a/crates/codegraph-graph/src/storage/lmdb.rs b/crates/codegraph-graph/src/storage/lmdb.rs new file mode 100644 index 000000000..686264d32 --- /dev/null +++ b/crates/codegraph-graph/src/storage/lmdb.rs @@ -0,0 +1,1140 @@ +//! LMDB-backed radix / entity storage (`lmdb-rkv`). +//! +//! Ánh xạ toàn bộ schema của sqlite (`rt_*` / `sg_*`) thành các named-database +//! trong một LMDB environment: mỗi bảng = một DBI, key/value pack LE 8-byte +//! giống sqlite (id/record/shard = `u64` LE). +//! +//! CHÚ Ý — mô hình concurrency: +//! - LMDB là sync/memory-mapped; các thao tác hoàn thành trong µs và KHÔNG +//! chờ `.await` giữa begin/commit, nên blocking executor không đáng kể so với +//! sqlx pool. +//! - `LmdbStorage` được `GraphIndex` bọc trong `Arc>` → mọi +//! mutation đã tuần tự hoá nên `read-modify-write` của children/shortcuts/ +//! counter không bao giờ va chạm giữa 2 writer. +//! - `tx.commit()` áp dụng buffer trong MỘT `RwTransaction` (atomic). + +use std::collections::HashMap; +use std::path::Path; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use codegraph_core::{FileInfo, Symbol}; +#[cfg(feature = "lmdb")] +use lmdb::EnvironmentFlags; +use lmdb::{Cursor, Database, DatabaseFlags, Environment, Transaction, WriteFlags}; + +use super::{EMPTY, Result, Storage, StorageError, Tx, TxOp, decode_chain, encode_chain}; + +/// Map lỗi LMDB → `StorageError`. +fn e(err: impl std::fmt::Display) -> StorageError { + StorageError::Internal(err.to_string()) +} + +// ── key/value packing (LE) ── + +#[inline] +fn k8(v: usize) -> [u8; 8] { + (v as u64).to_le_bytes() +} + +#[inline] +fn ku64(v: u64) -> [u8; 8] { + v.to_le_bytes() +} + +#[inline] +fn de_u64(b: &[u8]) -> u64 { + u64::from_le_bytes(b.try_into().expect("8-byte value")) +} + +// ── key chuỗi dài ── +// +// LMDB giới hạn key ≈ 511 byte (MDB_BAD_VALSIZE nếu vượt). Hai DBI dùng key là +// chuỗi dài (call_names, files) gặp tên/path > giới hạn. Khi đó ta ánh xạ chuỗi +// về key có độ dài cố định (marker 8B + FNV-1a 128-bit × 2 salt ~ổn định, va +// chạm ~2^-128) và lưu chuỗi gốc trong value để phục hồi lại đúng khi scan. + +const MAX_STR_KEY: usize = 440; + +fn fnv1a(h: u64, s: &str) -> u64 { + let mut h = h; + for &b in s.as_bytes() { + h ^= b as u64; + h = h.wrapping_mul(0x100000001b3); + } + h +} + +/// Key ổn định cho chuỗi: chuỗi ngắn dùng nguyên byte; dài → marker + hash. +fn str_key(s: &str) -> Vec { + if s.len() <= MAX_STR_KEY { + return s.as_bytes().to_vec(); + } + let mut v = Vec::with_capacity(24); + v.extend_from_slice(&u64::MAX.to_le_bytes()); + v.extend_from_slice(&fnv1a(0xcbf29ce484222325, s).to_le_bytes()); + v.extend_from_slice(&fnv1a(0x84222325cbf29ce4, s).to_le_bytes()); + v +} + +/// Value = `[u32 name_len] ++ name ++ payload` — name giữ nguyên phần key bị hash. +fn call_payload(name: &str, payload: &[u8]) -> Vec { + let mut v = Vec::with_capacity(4 + name.len() + payload.len()); + v.extend_from_slice(&(name.len() as u32).to_le_bytes()); + v.extend_from_slice(name.as_bytes()); + v.extend_from_slice(payload); + v +} + +/// Tách value `call_payload` → `(name, payload)`. +fn de_call_payload(v: &[u8]) -> (String, &[u8]) { + let n = u32::from_le_bytes(v[..4].try_into().expect("call payload len")) as usize; + let name = String::from_utf8_lossy(&v[4..4 + n]).into_owned(); + (name, &v[4 + n..]) +} + +/// Giá trị node = `prefix ++ record(8 LE)`. +fn node_val(prefix: &[u8], record: usize) -> Vec { + let mut v = Vec::with_capacity(prefix.len() + 8); + v.extend_from_slice(prefix); + v.extend_from_slice(&(record as u64).to_le_bytes()); + v +} + +fn de_node_val(v: &[u8]) -> (Vec, usize) { + let (p, r) = v.split_at(v.len() - 8); + (p.to_vec(), de_u64(r) as usize) +} + +/// Danh sách node id → bytes (mỗi phần tử `u64` LE). +fn list_val(list: &[usize]) -> Vec { + let mut v = Vec::with_capacity(list.len() * 8); + for &x in list { + v.extend_from_slice(&(x as u64).to_le_bytes()); + } + v +} + +fn de_list(v: &[u8]) -> Vec { + v.chunks_exact(8) + .map(|c| u64::from_le_bytes(c.try_into().unwrap()) as usize) + .collect() +} + +/// Thêm `x` vào danh sách (bỏ `EMPTY`, dedup, giữ sort) — mirror sqlite +/// `ORDER BY` + `ON CONFLICT DO NOTHING`. +fn push_unique(list: &mut Vec, x: usize) { + if x != EMPTY && !list.contains(&x) { + list.push(x); + list.sort_unstable(); + } +} + +// ── Tên DBI (schema — khớp bảng sqlite) ── + +const D_NODES: &str = "rt_nodes"; +const D_CHILDREN: &str = "rt_children"; +const D_ROOTS: &str = "rt_roots"; +const D_META: &str = "rt_meta"; +const D_KEYLEN: &str = "rt_keylen"; +const D_SHORTCUTS: &str = "rt_shortcuts"; +const D_CHAINS: &str = "rt_chains"; +const D_EDGES: &str = "rt_edge"; +const D_NODE_META: &str = "rt_node_meta"; +#[cfg(feature = "bloom-search")] +const D_BLOOMS: &str = "rt_node_blooms"; +const D_COUNTER: &str = "rt_counter"; +const D_SYMBOLS: &str = "sg_symbols"; +const D_NEXT_ID: &str = "sg_next_id"; +const D_CALL_RECORDS: &str = "sg_call_records"; +const D_CALL_NAMES: &str = "sg_call_names"; +const D_FILES: &str = "sg_files"; +const D_VERSION: &str = "sg_meta"; + +/// Key duy nhất cho các "row đơn" (counter / next_id / version) — mỗi DBI chỉ có 1 row. +const KEY_ONE: [u8; 8] = [0u8; 8]; + +// ── Env ── + +fn open_env(path: &str) -> Result> { + let p = Path::new(path); + std::fs::create_dir_all(p).map_err(e)?; + let mut b = Environment::new(); + b.set_max_dbs(32); // schema dùng ~17 named-db + b.set_max_readers(512); // locktable đủ chỗ cho runtime/mcp probe + request song song + b.set_map_size(1 << 30); // 1 GiB address space (LMDB chỉ commit trang thực đụng) + let env = b.open(p).map_err(e)?; + Ok(Arc::new(env)) +} + +#[cfg(feature = "lmdb")] +#[cfg_attr(feature = "sqlite", allow(dead_code))] // probe chỉ dùng khi lmdb là backend file +fn open_env_read_only(path: &str) -> lmdb::Result { + let mut b = Environment::new(); + b.set_flags(EnvironmentFlags::READ_ONLY); + b.set_max_dbs(32); + b.open(Path::new(path)) +} + +/// Cache read-only `Environment` theo path — 1 env dùng chung cho mọi `probe_version`. +/// +/// `Environment` là `Send + Sync` nên an toàn để dùng chung; env sống trọn +/// process (không drop) để locktable không bị mở/đóng lặp. +#[cfg(feature = "lmdb")] +#[cfg_attr(feature = "sqlite", allow(dead_code))] // probe chỉ dùng khi lmdb là backend file +fn probe_env(path: &str) -> lmdb::Result> { + let data = Path::new(path).join("data.mdb"); + if !data.is_file() { + return Err(lmdb::Error::NotFound); + } + static CACHE: std::sync::LazyLock>>> = + std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); + let mut cache = CACHE.lock().expect("probe env cache lock"); + if let Some(env) = cache.get(path) { + return Ok(env.clone()); + } + let env = Arc::new(open_env_read_only(path)?); + cache.insert(path.to_string(), env.clone()); + Ok(env) +} + +/// Đọc `version` từ file mà KHÔNG tạo file (nếu chưa có) — dùng bởi +/// `SharedGraphIndex::ensure_fresh` để dò stale. Mirror `SqliteStorage::probe_version`. +/// +/// Reuse env cache (`probe_env`) để không mở/đóng `Environment` mỗi lần gọi — +/// `MDB_BAD_RSLOT` xảy ra khi nhiều `Environment` cùng mở/đóng trên một locktable +/// (lock.mdb) khi nhiều request probe song song (runtime/mcp: mỗi request gọi qua +/// `ensure_fresh` → `current_version`). Cache theo path giữ 1 env read-only dùng +/// chung (sống trọn process) nên không còn tranh chấp slot reader. +#[cfg(feature = "lmdb")] +#[cfg_attr(feature = "sqlite", allow(dead_code))] // probe chỉ dùng khi lmdb là backend file +pub async fn probe_version(path: &str) -> Result { + let env = probe_env(path) + .map_err(|err| StorageError::Internal(format!("lmdb file not found: {path} ({err})")))?; + let db = env.open_db(Some(D_VERSION)).map_err(e)?; + let tx = env.begin_ro_txn().map_err(e)?; + match tx.get(db, &KEY_ONE).map(de_u64) { + Ok(v) => Ok(v), + Err(lmdb::Error::NotFound) => { + Err(StorageError::Internal("lmdb version row missing".into())) + } + Err(err) => Err(StorageError::Internal(err.to_string())), + } +} + +// ==================== LmdbStorage ==================== + +/// LMDB backend: `Arc` + handle (Copy) của từng DBI. +pub struct LmdbStorage { + env: Arc, + nodes: Database, + children: Database, + roots: Database, + meta: Database, + keylen: Database, + shortcuts: Database, + chains: Database, + edges: Database, + node_meta: Database, + #[cfg(feature = "bloom-search")] + blooms: Database, + counter: Database, + symbols: Database, + next_id: Database, + call_records: Database, + call_names: Database, + files: Database, + version: Database, +} + +impl LmdbStorage { + /// Mở (hoặc tạo mới nếu chưa có) LMDB tại thư mục `path`. Idempotent — + /// sentinel/counter chỉ seed nếu chưa có nên reopen giữ nguyên dữ liệu. + pub async fn open(path: &str) -> Result { + let env = open_env(path)?; + let s = Self::from_env(env)?; + s.init().await?; + Ok(s) + } + + fn from_env(env: Arc) -> Result { + let nodes = env + .create_db(Some(D_NODES), DatabaseFlags::empty()) + .map_err(e)?; + let children = env + .create_db(Some(D_CHILDREN), DatabaseFlags::empty()) + .map_err(e)?; + let roots = env + .create_db(Some(D_ROOTS), DatabaseFlags::empty()) + .map_err(e)?; + let meta = env + .create_db(Some(D_META), DatabaseFlags::empty()) + .map_err(e)?; + let keylen = env + .create_db(Some(D_KEYLEN), DatabaseFlags::empty()) + .map_err(e)?; + let shortcuts = env + .create_db(Some(D_SHORTCUTS), DatabaseFlags::empty()) + .map_err(e)?; + let chains = env + .create_db(Some(D_CHAINS), DatabaseFlags::empty()) + .map_err(e)?; + let edges = env + .create_db(Some(D_EDGES), DatabaseFlags::empty()) + .map_err(e)?; + let node_meta = env + .create_db(Some(D_NODE_META), DatabaseFlags::empty()) + .map_err(e)?; + #[cfg(feature = "bloom-search")] + let blooms = env + .create_db(Some(D_BLOOMS), DatabaseFlags::empty()) + .map_err(e)?; + let counter = env + .create_db(Some(D_COUNTER), DatabaseFlags::empty()) + .map_err(e)?; + let symbols = env + .create_db(Some(D_SYMBOLS), DatabaseFlags::empty()) + .map_err(e)?; + let next_id = env + .create_db(Some(D_NEXT_ID), DatabaseFlags::empty()) + .map_err(e)?; + let call_records = env + .create_db(Some(D_CALL_RECORDS), DatabaseFlags::empty()) + .map_err(e)?; + let call_names = env + .create_db(Some(D_CALL_NAMES), DatabaseFlags::empty()) + .map_err(e)?; + let files = env + .create_db(Some(D_FILES), DatabaseFlags::empty()) + .map_err(e)?; + let version = env + .create_db(Some(D_VERSION), DatabaseFlags::empty()) + .map_err(e)?; + Ok(Self { + env, + nodes, + children, + roots, + meta, + keylen, + shortcuts, + chains, + edges, + node_meta, + #[cfg(feature = "bloom-search")] + blooms, + counter, + symbols, + next_id, + call_records, + call_names, + files, + version, + }) + } + + /// Seed sentinel node 0 + counter/next_id/version nếu chưa tồn tại. + async fn init(&self) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + let db = self.nodes; + if matches!(tx.get(db, &k8(EMPTY)), Err(lmdb::Error::NotFound)) { + tx.put(db, &k8(EMPTY), &node_val(b"", 0), WriteFlags::empty()) + .map_err(e)?; + } + if matches!(tx.get(self.counter, &KEY_ONE), Err(lmdb::Error::NotFound)) { + tx.put(self.counter, &KEY_ONE, &ku64(1), WriteFlags::empty()) + .map_err(e)?; + } + if matches!(tx.get(self.next_id, &KEY_ONE), Err(lmdb::Error::NotFound)) { + // next_id bắt đầu từ SYMBOL_BASE (marker reserved 1..=99) — mirror sqlite. + tx.put(self.next_id, &KEY_ONE, &ku64(100), WriteFlags::empty()) + .map_err(e)?; + } + if matches!(tx.get(self.version, &KEY_ONE), Err(lmdb::Error::NotFound)) { + tx.put(self.version, &KEY_ONE, &ku64(0), WriteFlags::empty()) + .map_err(e)?; + } + tx.commit().map_err(e)?; + Ok(()) + } + + fn get_opt<'txn, K: AsRef<[u8]>>( + &self, + tx: &'txn impl Transaction, + db: Database, + key: &K, + ) -> Result> { + match tx.get(db, key) { + Ok(v) => Ok(Some(v)), + Err(lmdb::Error::NotFound) => Ok(None), + Err(err) => Err(StorageError::Internal(err.to_string())), + } + } +} + +// ==================== Storage impl ==================== + +#[async_trait] +impl Storage for LmdbStorage { + async fn new_node(&mut self, prefix: Vec, record: usize) -> Result { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + // Không có RETURNING — đọc-rồi-ghi counter trong cùng write tx; an toàn + // vì GraphIndex tuần tự hoá mọi writer qua RwLock. + let next = match self.get_opt(&tx, self.counter, &KEY_ONE)? { + Some(v) => de_u64(v), + None => 1, + }; + let id = next as usize; + tx.put(self.counter, &KEY_ONE, &ku64(next + 1), WriteFlags::empty()) + .map_err(e)?; + tx.put( + self.nodes, + &k8(id), + &node_val(&prefix, record), + WriteFlags::empty(), + ) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(id) + } + + async fn update_node( + &mut self, + id: usize, + prefix: Option>, + record: Option, + ) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + let key = k8(id); + let Some(cur) = self.get_opt(&tx, self.nodes, &key)?.map(de_node_val) else { + return Err(StorageError::BranchOutOfRange(id)); + }; + let (mut p, mut r) = cur; + if let Some(np) = prefix { + p = np; + } + if let Some(nr) = record { + r = nr; + } + tx.put(self.nodes, &key, &node_val(&p, r), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_node(&self, id: usize) -> Result<(Vec, usize)> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let Some(v) = self.get_opt(&tx, self.nodes, &k8(id))? else { + return Err(StorageError::BranchOutOfRange(id)); + }; + Ok(de_node_val(v)) + } + + async fn get_children(&self, id: usize) -> Result> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut out = match self.get_opt(&tx, self.children, &k8(id))? { + Some(v) => de_list(v), + None => Vec::new(), + }; + out.sort_unstable(); + Ok(out) + } + + #[cfg(feature = "bloom-search")] + async fn set_node_bloom(&mut self, id: usize, bloom: &[u8]) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.blooms, &k8(id), &bloom, WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + #[cfg(feature = "bloom-search")] + async fn get_node_bloom(&self, id: usize) -> Result>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self.get_opt(&tx, self.blooms, &k8(id))?.map(|b| b.to_vec())) + } + + async fn set_edge_data(&mut self, edge: usize, data: &[u8]) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.edges, &k8(edge), &data, WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_edge_data(&self, edge: usize) -> Result>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.edges, &k8(edge))? + .map(|v| v.to_vec())) + } + + async fn clear_edges(&mut self) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.clear_db(self.edges).map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn for_each_edge_data( + &self, + f: &mut (dyn for<'a> FnMut(usize, &'a [u8]) -> Result<()> + Send), + ) -> Result<()> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut cur = tx.open_ro_cursor(self.edges).map_err(e)?; + let mut rows: Vec<(Vec, Vec)> = Vec::new(); + for item in cur.iter() { + let (k, v) = item.map_err(e)?; + rows.push((k.to_vec(), v.to_vec())); + } + drop(cur); + drop(tx); + rows.sort_by(|a, b| a.0.cmp(&b.0)); + for (k, v) in rows { + f(de_u64(&k) as usize, &v)?; + } + Ok(()) + } + + async fn set_node_meta(&mut self, elem: usize, meta: &[u8]) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.node_meta, &k8(elem), &meta, WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_node_meta(&self, elem: usize) -> Result>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.node_meta, &k8(elem))? + .map(|v| v.to_vec())) + } + + async fn clear_node_meta(&mut self) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.clear_db(self.node_meta).map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn set_chain(&mut self, record: usize, chain: &[u64]) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put( + self.chains, + &k8(record), + &encode_chain(chain), + WriteFlags::empty(), + ) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_chain(&self, record: usize) -> Result>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.chains, &k8(record))? + .map(decode_chain)) + } + + async fn clear_chains(&mut self) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.clear_db(self.chains).map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn save_symbol(&mut self, sym: &Symbol) -> Result<()> { + let data = + serde_json::to_vec(sym).map_err(|err| StorageError::Internal(err.to_string()))?; + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.symbols, &ku64(sym.id), &data, WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn load_symbol(&self, id: u64) -> Result> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let Some(data) = self.get_opt(&tx, self.symbols, &ku64(id))? else { + return Ok(None); + }; + serde_json::from_slice(data) + .map(Some) + .map_err(|err| StorageError::Internal(err.to_string())) + } + + async fn load_all_symbols(&self) -> Result> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut cur = tx.open_ro_cursor(self.symbols).map_err(e)?; + let mut out = Vec::new(); + for item in cur.iter() { + let (_k, v) = item.map_err(e)?; + out.push( + serde_json::from_slice(v).map_err(|err| StorageError::Internal(err.to_string()))?, + ); + } + Ok(out) + } + + async fn save_next_id(&mut self, next: u64) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.next_id, &KEY_ONE, &ku64(next), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn load_next_id(&self) -> Result { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.next_id, &KEY_ONE)? + .map(de_u64) + .unwrap_or(100)) + } + + async fn all_chains(&self) -> Result)>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut cur = tx.open_ro_cursor(self.chains).map_err(e)?; + let mut out = Vec::new(); + for item in cur.iter() { + let (k, v) = item.map_err(e)?; + out.push((de_u64(k), v.to_vec())); + } + Ok(out) + } + + async fn set_call_records(&mut self, func: u64, records: &[u8]) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put( + self.call_records, + &ku64(func), + &records, + WriteFlags::empty(), + ) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_call_records(&self, func: u64) -> Result>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.call_records, &ku64(func))? + .map(|v| v.to_vec())) + } + + async fn all_call_records(&self) -> Result)>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut cur = tx.open_ro_cursor(self.call_records).map_err(e)?; + let mut out = Vec::new(); + for item in cur.iter() { + let (k, v) = item.map_err(e)?; + out.push((de_u64(k), v.to_vec())); + } + Ok(out) + } + + async fn set_call_name_index(&mut self, name: &str, sites: &[u8]) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put( + self.call_names, + &str_key(name), + &call_payload(name, sites), + WriteFlags::empty(), + ) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn load_call_name_index(&self, name: &str) -> Result>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.call_names, &str_key(name))? + .map(|v| de_call_payload(v).1.to_vec())) + } + + async fn all_call_name_indexes(&self) -> Result)>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut cur = tx.open_ro_cursor(self.call_names).map_err(e)?; + let mut out = Vec::new(); + for item in cur.iter() { + let (_k, v) = item.map_err(e)?; + let (name, sites) = de_call_payload(v); + out.push((name, sites.to_vec())); + } + Ok(out) + } + + async fn upsert_file(&mut self, f: &FileInfo) -> Result<()> { + let data = serde_json::to_vec(f).map_err(|err| StorageError::Internal(err.to_string()))?; + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.files, &str_key(&f.path), &data, WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn load_all_files(&self) -> Result> { + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut cur = tx.open_ro_cursor(self.files).map_err(e)?; + let mut out = Vec::new(); + for item in cur.iter() { + let (_k, v) = item.map_err(e)?; + out.push( + serde_json::from_slice(v).map_err(|err| StorageError::Internal(err.to_string()))?, + ); + } + Ok(out) + } + + async fn version(&self) -> Result { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.version, &KEY_ONE)? + .map(de_u64) + .unwrap_or(0)) + } + + async fn set_version(&mut self, v: u64) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.version, &KEY_ONE, &ku64(v), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn clear_entities(&mut self) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + for db in [self.symbols, self.call_records, self.call_names, self.files] { + tx.clear_db(db).map_err(e)?; + } + tx.put(self.next_id, &KEY_ONE, &ku64(100), WriteFlags::empty()) + .map_err(e)?; + tx.put(self.version, &KEY_ONE, &ku64(0), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn set_root(&mut self, shard: usize, root: usize) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.roots, &k8(shard), &k8(root), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_root(&self, shard: usize) -> Result { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.roots, &k8(shard))? + .map(de_u64) + .unwrap_or(EMPTY as u64) as usize) + } + + async fn set_meta(&mut self, record: usize, meta: &[u8]) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.meta, &k8(record), &meta, WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_meta(&self, record: usize) -> Result>> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.meta, &k8(record))? + .map(|v| v.to_vec())) + } + + async fn set_key_len(&mut self, record: usize, len: usize) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.put(self.keylen, &k8(record), &k8(len), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_key_len(&self, record: usize) -> Result> { + let tx = self.env.begin_ro_txn().map_err(e)?; + Ok(self + .get_opt(&tx, self.keylen, &k8(record))? + .map(de_u64) + .map(|v| v as usize)) + } + + async fn add_shortcut_node(&mut self, shard: usize, elem: &[u8], node_id: usize) -> Result<()> { + let mut key = k8(shard).to_vec(); + key.extend_from_slice(elem); + let mut tx = self.env.begin_rw_txn().map_err(e)?; + let mut list = match self.get_opt(&tx, self.shortcuts, &key)? { + Some(v) => de_list(v), + None => Vec::new(), + }; + push_unique(&mut list, node_id); + tx.put(self.shortcuts, &key, &list_val(&list), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + async fn get_shortcut_nodes(&self, shard: usize, elem: &[u8]) -> Result> { + let mut key = k8(shard).to_vec(); + key.extend_from_slice(elem); + let tx = self.env.begin_ro_txn().map_err(e)?; + let mut out = match self.get_opt(&tx, self.shortcuts, &key)? { + Some(v) => de_list(v), + None => Vec::new(), + }; + out.sort_unstable(); + Ok(out) + } + + async fn clear_shortcuts(&mut self) -> Result<()> { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + tx.clear_db(self.shortcuts).map_err(e)?; + tx.commit().map_err(e)?; + Ok(()) + } + + fn new_tx(&self) -> Box { + Box::new(LmdbTx { + env: self.env.clone(), + nodes: self.nodes, + children: self.children, + counter: self.counter, + nodes_pending: Vec::new(), + ops: Vec::new(), + }) + } +} + +// ==================== LmdbTx ==================== + +/// Transaction cho `LmdbStorage`: buffer mutation, áp dụng atomic trong một +/// `RwTransaction` tại `commit`. `new_node` cấp id ngay (bump counter như +/// sqlite `RETURNING`) nhưng row chỉ lộ khi commit. +pub struct LmdbTx { + env: Arc, + nodes: Database, + children: Database, + counter: Database, + nodes_pending: Vec<(usize, Vec, usize)>, + ops: Vec, +} + +#[async_trait] +impl Tx for LmdbTx { + async fn new_node(&mut self, prefix: Vec, record: usize) -> Result { + let mut tx = self.env.begin_rw_txn().map_err(e)?; + let next = match tx.get(self.counter, &KEY_ONE).map(de_u64) { + Ok(v) => v, + Err(lmdb::Error::NotFound) => 1, + Err(err) => return Err(StorageError::Internal(err.to_string())), + }; + tx.put(self.counter, &KEY_ONE, &ku64(next + 1), WriteFlags::empty()) + .map_err(e)?; + tx.commit().map_err(e)?; + let id = next as usize; + self.nodes_pending.push((id, prefix, record)); + Ok(id) + } + + async fn update_node( + &mut self, + id: usize, + prefix: Option>, + record: Option, + ) -> Result<()> { + self.ops.push(TxOp::UpdateNode { id, prefix, record }); + Ok(()) + } + + async fn add_child(&mut self, parent: usize, child: usize) -> Result<()> { + self.ops.push(TxOp::AddChild { parent, child }); + Ok(()) + } + + async fn move_child(&mut self, from: usize, to: usize, child: usize) -> Result<()> { + self.ops.push(TxOp::MoveChild { from, to, child }); + Ok(()) + } + + async fn commit(self: Box) -> Result<()> { + let LmdbTx { + env, + nodes, + children, + counter, + nodes_pending, + ops, + } = *self; + let mut tx = env.begin_rw_txn().map_err(e)?; + + // 1. Materialize node mới — để ops add/move trỏ tới hợp lệ. + for (id, prefix, record) in &nodes_pending { + tx.put( + nodes, + &k8(*id), + &node_val(prefix, *record), + WriteFlags::empty(), + ) + .map_err(e)?; + } + + // 2. Counter đã được bump ở new_node; giữ MAX như sqlite phòng writer khác. + if let Some(max_id) = nodes_pending.iter().map(|(id, _, _)| *id).max() { + let cur = match tx.get(counter, &KEY_ONE).map(de_u64) { + Ok(v) => v, + Err(lmdb::Error::NotFound) => 1, + Err(err) => return Err(StorageError::Internal(err.to_string())), + }; + let nxt = cur.max(max_id as u64 + 1); + tx.put(counter, &KEY_ONE, &ku64(nxt), WriteFlags::empty()) + .map_err(e)?; + } + + // 3. Áp dụng ops — children là read-modify-write trên KV; gộp theo parent + // để tránh đọc/ghi lặp nhiều lần cho cùng một node. + let mut child_map: HashMap> = HashMap::new(); + for op in &ops { + match op { + TxOp::AddChild { parent, child } => { + let list = child_map.entry(*parent).or_insert_with(|| { + tx.get(children, &k8(*parent)) + .map(de_list) + .unwrap_or_default() + }); + push_unique(list, *child); + } + TxOp::MoveChild { from, to, child } => { + if from != to { + if let Some(list) = child_map.get_mut(from) { + list.retain(|x| x != child); + } else { + let list = tx + .get(children, &k8(*from)) + .map(de_list) + .unwrap_or_default() + .into_iter() + .filter(|x| x != child) + .collect::>(); + child_map.insert(*from, list); + } + let list = child_map.entry(*to).or_insert_with(|| { + tx.get(children, &k8(*to)).map(de_list).unwrap_or_default() + }); + push_unique(list, *child); + } + } + TxOp::UpdateNode { id, prefix, record } => { + let key = k8(*id); + let Some((mut p, mut r)) = tx.get(nodes, &key).map(de_node_val).ok() else { + continue; + }; + if let Some(np) = prefix { + p = np.clone(); + } + if let Some(nr) = record { + r = *nr; + } + tx.put(nodes, &key, &node_val(&p, r), WriteFlags::empty()) + .map_err(e)?; + } + } + } + for (parent, list) in &child_map { + tx.put(children, &k8(*parent), &list_val(list), WriteFlags::empty()) + .map_err(e)?; + } + + tx.commit().map_err(e)?; + Ok(()) + } +} + +// ==================== Tests ==================== + +#[cfg(test)] +mod tests { + use super::*; + + fn tmp_path() -> (tempfile::TempDir, String) { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test.lmdb"); + let path = path.to_string_lossy().into_owned(); + (dir, path) + } + + #[tokio::test] + async fn test_new_node_and_get_node() { + let (_d, path) = tmp_path(); + let mut s = LmdbStorage::open(&path).await.unwrap(); + let id = s.new_node(b"hello".to_vec(), 42).await.unwrap(); + assert_ne!(id, EMPTY); + let (prefix, record) = s.get_node(id).await.unwrap(); + assert_eq!(prefix, b"hello"); + assert_eq!(record, 42); + } + + /// Node trong tx chưa lộ ra reader cho tới `commit`. + #[tokio::test] + async fn test_tx_atomic() { + let (_d, path) = tmp_path(); + let s = LmdbStorage::open(&path).await.unwrap(); + let mut tx = s.new_tx(); + let id = tx.new_node(b"x".to_vec(), 7).await.unwrap(); + // Chưa commit → đọc qua storage thấy "chưa có". + assert!(matches!( + s.get_node(id).await, + Err(StorageError::BranchOutOfRange(_)) + )); + + tx.add_child(EMPTY, id).await.unwrap(); + tx.commit().await.unwrap(); + + assert_eq!(s.get_node(id).await.unwrap(), (b"x".to_vec(), 7)); + assert_eq!(s.get_children(EMPTY).await.unwrap(), vec![id]); + } + + /// Move child từ parent này sang parent khác. + #[tokio::test] + async fn test_move_child() { + let (_d, path) = tmp_path(); + let s = LmdbStorage::open(&path).await.unwrap(); + let mut sa = s.new_tx(); + let a = sa.new_node(b"a".to_vec(), 1).await.unwrap(); + let b = sa.new_node(b"b".to_vec(), 2).await.unwrap(); + sa.add_child(EMPTY, a).await.unwrap(); + sa.add_child(EMPTY, b).await.unwrap(); + sa.commit().await.unwrap(); + + let mut tx = s.new_tx(); + tx.move_child(EMPTY, a, b).await.unwrap(); + tx.commit().await.unwrap(); + assert_eq!(s.get_children(EMPTY).await.unwrap(), vec![a]); + assert_eq!(s.get_children(a).await.unwrap(), vec![b]); + } + + /// Shortcut set đọc/ghi đúng, node id unique sort. + #[tokio::test] + async fn test_shortcut() { + let (_d, path) = tmp_path(); + let mut s = LmdbStorage::open(&path).await.unwrap(); + let elem = b"ab".to_vec(); + s.add_shortcut_node(0, &elem, 5).await.unwrap(); + s.add_shortcut_node(0, &elem, 3).await.unwrap(); + s.add_shortcut_node(0, &elem, 5).await.unwrap(); // dup — bị loại + assert_eq!(s.get_shortcut_nodes(0, &elem).await.unwrap(), vec![3, 5]); + } + + /// Meta/keylen ghi đọc như sqlite. + #[tokio::test] + async fn test_meta_and_keylen() { + let (_d, path) = tmp_path(); + let mut s = LmdbStorage::open(&path).await.unwrap(); + s.set_meta(1, b"m").await.unwrap(); + assert_eq!(s.get_meta(1).await.unwrap(), Some(b"m".to_vec())); + assert_eq!(s.get_meta(2).await.unwrap(), None); + s.set_key_len(1, 9).await.unwrap(); + assert_eq!(s.get_key_len(1).await.unwrap(), Some(9)); + } + + /// Dữ liệu tồn tại sau reopen + probe_version đọc đúng, không tạo file mới. + #[tokio::test] + async fn test_reopen_persists_and_probe() { + let (_d, path) = tmp_path(); + { + let mut s = LmdbStorage::open(&path).await.unwrap(); + s.new_node(b"hi".to_vec(), 1).await.unwrap(); + s.set_version(7).await.unwrap(); + } + let s = LmdbStorage::open(&path).await.unwrap(); + assert_eq!(s.version().await.unwrap(), 7); + + let (prefix, record) = s.get_node(1).await.unwrap(); + assert_eq!(prefix, b"hi"); + assert_eq!(record, 1); + + // probe_version đọc từ file hiện có (không tạo file mới). + assert_eq!(probe_version(&path).await.unwrap(), 7); + assert!(probe_version("definitely/missing.lmdb").await.is_err()); + } + + /// Regression MDB_BAD_RSLOT: nhiều reader probe song song trên cùng path + /// phải dùng chung env cache — không mở/đóng Environment mỗi lần gọi. + #[tokio::test] + async fn test_concurrent_probe_reuses_env() { + let (_d, path) = tmp_path(); + let (_d2, path2) = tmp_path(); + { + let mut s = LmdbStorage::open(&path).await.unwrap(); + s.set_version(5).await.unwrap(); + } + { + let mut s = LmdbStorage::open(&path2).await.unwrap(); + s.set_version(9).await.unwrap(); + } + + // Nhiều task probe song song trên 2 path khác nhau — mỗi path trả đúng + // version, và KHÔNG mở env mới mỗi lần (cache dùng chung → không BAD_RSLOT). + let mut tasks = Vec::new(); + for _ in 0..8 { + let p1 = path.clone(); + let p2 = path2.clone(); + tasks.push(tokio::spawn(async move { + for _ in 0..20 { + let v1 = probe_version(&p1).await.unwrap(); + let v2 = probe_version(&p2).await.unwrap(); + assert_eq!(v1, 5); + assert_eq!(v2, 9); + } + })); + } + for t in tasks { + t.await.unwrap(); + } + } + + /// Regression MDB_BAD_VALSIZE: key chuỗi > 511 byte bị LMDB từ chối; `str_key` + /// phải hash về key cố định 24B và giữ chuỗi gốc trong value để đọc lại đúng. + #[tokio::test] + async fn test_long_call_name_and_path_roundtrip() { + let (_d, path) = tmp_path(); + let mut s = LmdbStorage::open(&path).await.unwrap(); + + // call-name > 511 byte. + let long_call = format!("very{}long{}mangled", "t".repeat(300), "q".repeat(300)); + assert!(long_call.len() > 511); + s.set_call_name_index(&long_call, b"sites").await.unwrap(); + // Đọc lại nguyên payload dù key đã bị hash. + assert_eq!( + s.load_call_name_index(&long_call).await.unwrap().as_deref(), + Some(b"sites".as_slice()) + ); + // Scan trả đúng tên gốc. + let all = s.all_call_name_indexes().await.unwrap(); + assert_eq!(all.len(), 1); + assert_eq!(all[0].0, long_call); + assert_eq!(all[0].1, b"sites"); + + // path file > 511 byte. + let seg = "d".repeat(300); + let long_path = format!("src/{seg}/{seg}/mod.ts"); + assert!(long_path.len() > 511); + let f = FileInfo { + path: long_path.clone(), + language: "ts".into(), + bytes: 10, + lines: 1, + }; + s.upsert_file(&f).await.unwrap(); + let files = s.load_all_files().await.unwrap(); + assert_eq!(files.len(), 1); + assert_eq!(files[0].path, long_path); + } +} diff --git a/crates/codegraph-graph/src/storage/sqlite.rs b/crates/codegraph-graph/src/storage/sqlite.rs index 32daf65c7..b7d2aeeea 100644 --- a/crates/codegraph-graph/src/storage/sqlite.rs +++ b/crates/codegraph-graph/src/storage/sqlite.rs @@ -767,11 +767,17 @@ pub struct SqliteTx { impl Tx for SqliteTx { async fn new_node(&mut self, prefix: Vec, record: usize) -> Result { let mut conn = self.pool.acquire().await.map_err(db_err)?; - let next: i64 = sqlx::query_scalar("SELECT next FROM rt_counter WHERE id = 1") - .fetch_one(&mut *conn) - .await - .map_err(db_err)?; - let id = next as usize + self.nodes.len(); + // Cấp id atomic ngay tại lúc reservation — không `SELECT next` rồi tự + // tính (đọc-then-giữ nếu 2 tx/writer chạy song song trên cùng db sẽ cấp + // trùng id → `UNIQUE constraint failed: rt_nodes.id` — bug E). Bản thân + // các row vẫn được materialize ở commit, nhưng id đã unique toàn cục. + let next: i64 = sqlx::query_scalar( + "UPDATE rt_counter SET next = next + 1 WHERE id = 1 RETURNING next - 1", + ) + .fetch_one(&mut *conn) + .await + .map_err(db_err)?; + let id = next as usize; self.nodes.push((id, prefix, record)); Ok(id) } @@ -1018,6 +1024,48 @@ mod tests { assert_eq!(s.get_node(id).await.unwrap().1, 9); } + /// Regression bug E: `UNIQUE constraint failed: rt_nodes.id` khi nhiều tx + /// (2 writer / watcher + mcp chạy cùng db.sqlite) cấp id node song song. + /// `new_node` phải cấp id atomic qua `UPDATE rt_counter ... RETURNING`, + /// không đọc-then-tính (`SELECT next` + `next + nodes.len()`) dễ trùng. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_concurrent_tx_new_node_ids_unique() { + use std::collections::HashSet; + use std::sync::Arc; + let (_d, path) = tmp_path(); + let s = Arc::new(SqliteStorage::open(&path).await.unwrap()); + + let mut handles = Vec::new(); + for w in 0..8 { + let s = Arc::clone(&s); + handles.push(tokio::spawn(async move { + let mut tx = s.new_tx(); + let mut ids = Vec::new(); + for i in 0..8 { + let prefix = format!("w{w}-{i}").into_bytes(); + ids.push(tx.new_node(prefix, 1).await.unwrap()); + } + tx.commit().await.unwrap(); + ids + })); + } + + let mut all = Vec::new(); + for h in handles { + all.extend(h.await.unwrap()); + } + let unique: HashSet = all.iter().copied().collect(); + assert_eq!( + unique.len(), + all.len(), + "duplicate rt node ids allocated across concurrent transactions: {all:?}" + ); + // Toàn bộ node đã materialize hợp lệ (commit không UNIQUE-fail). + for id in all { + s.get_node(id).await.expect("committed node readable"); + } + } + #[tokio::test] async fn test_tx_move_child_migrates() { let (_d, path) = tmp_path(); diff --git a/crates/codegraph-graph/tests/lmdb.rs b/crates/codegraph-graph/tests/lmdb.rs new file mode 100644 index 000000000..b89223395 --- /dev/null +++ b/crates/codegraph-graph/tests/lmdb.rs @@ -0,0 +1,285 @@ +//! Integration tests cho backend LMDB (feature `lmdb`) — port subset của +//! `tests/sqlite.rs`. Khác sqlite (path = file), LMDB dùng path = thư mục. +//! +//! `SharedGraphIndex` routing theo scheme trong DSN (`lmdb://...`) — nên bộ +//! test này chạy được dù có bật sqlite hay không. + +#![cfg(feature = "lmdb")] + +use codegraph_core::{CallRecord, EffectType, SYMBOL_BASE, Symbol, SymbolKind}; +use codegraph_graph::GraphIndex; +use codegraph_graph::ParseResult; +use codegraph_graph::SharedGraphIndex; +use std::collections::HashMap; +use std::sync::Arc; + +fn sym(file: &str, name: &str, id: u64) -> Symbol { + Symbol { + id, + name: name.to_string(), + kind: SymbolKind::Function, + scope: codegraph_core::ScopeLevel::Global, + scope_id: 0, + type_ref: 0, + type_name: None, + file: file.to_string(), + line: 1, + end_line: 1, + signature: None, + doc: None, + annotations: Vec::new(), + language: "ts".to_string(), + } +} + +fn result( + path: &str, + symbols: Vec, + chains: HashMap>, + calls: Vec, +) -> ParseResult { + ParseResult { + path: path.to_string(), + language: "ts".to_string(), + bytes: 0, + lines: 0, + symbols, + chains, + calls, + } +} + +/// Ingest → reopen: entity + query surface sống lại từ file LMDB. +#[tokio::test] +async fn index_ingest_reopen_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let path = format!("lmdb://{}/db.lmdb", dir.path().to_string_lossy()); + + let calls = vec![CallRecord { + caller_id: SYMBOL_BASE, + call_name: "b".to_string(), + position: 1, + arg_exprs: vec!["x".to_string()], + line: 3, + condition: None, + is_loop_body: false, + effect: EffectType::None, + effect_desc: None, + target_class: None, + target_method: None, + }]; + let r = result( + "a.ts", + vec![ + sym("a.ts", "a", SYMBOL_BASE), + sym("a.ts", "b", SYMBOL_BASE + 1), + ], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE, SYMBOL_BASE + 1])]), + calls, + ); + { + let mut idx = GraphIndex::open(&path).await.unwrap(); + idx.ingest(&[r]).await.unwrap(); + assert_eq!(idx.version(), 1); + } + + let idx = GraphIndex::open(&path).await.unwrap(); + assert_eq!(idx.version(), 1); + assert_eq!(idx.stats().symbols, 2); + assert_eq!(idx.stats().chains, 1); + assert_eq!(idx.stats().edges, 1); + assert_eq!(idx.files().len(), 1); + assert_eq!(idx.files()[0].path, "a.ts"); + + let cees = idx.callees(SYMBOL_BASE).await.unwrap(); + assert_eq!(cees.len(), 1); + assert_eq!(cees[0].name, "b"); + let cers = idx.callers(SYMBOL_BASE + 1, 1).await.unwrap(); + assert_eq!(cers.len(), 1); + assert_eq!(cers[0].name, "a"); + + let flow = idx.flow(SYMBOL_BASE).await.unwrap(); + assert_eq!(flow.chain_desc, vec!["a", "b"]); + assert_eq!(flow.calls[0].line, 3); + + let sf = idx.search_flow(&[SYMBOL_BASE + 1]).await.unwrap(); + assert_eq!(sf.len(), 1); + assert_eq!(sf[0].function_name, "a"); +} + +/// Ingest rỗng = full wipe; version vẫn bump; wipe giữ trên đĩa sau reopen. +#[tokio::test] +async fn empty_ingest_wipes_store() { + let dir = tempfile::tempdir().unwrap(); + let path = format!("lmdb://{}/db.lmdb", dir.path().to_string_lossy()); + + let r = result( + "a.ts", + vec![sym("a.ts", "a", SYMBOL_BASE)], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE])]), + vec![], + ); + let mut idx = GraphIndex::open(&path).await.unwrap(); + idx.ingest(&[r]).await.unwrap(); + assert_eq!(idx.stats().symbols, 1); + + idx.ingest(&[]).await.unwrap(); + assert_eq!(idx.version(), 2); + assert_eq!(idx.stats().symbols, 0); + assert!(idx.symbol_by_id(SYMBOL_BASE).is_none()); + + let idx = GraphIndex::open(&path).await.unwrap(); + assert_eq!(idx.stats().symbols, 0); + assert_eq!(idx.version(), 2); +} + +/// SharedGraphIndex phát hiện stale qua version bump (dùng `LmdbStorage::probe_version`). +/// DSN có scheme `lmdb://` → shared mở đúng backend LMDB dù sqlite cũng bật. +#[tokio::test] +async fn shared_index_rebuilds_on_reindex() { + let dir = tempfile::tempdir().unwrap(); + let db_dir = dir.path().join("db.lmdb"); + let db_str = format!("lmdb://{}", db_dir.to_string_lossy()); + + { + let mut idx = GraphIndex::open(&db_str).await.unwrap(); + let r = result( + "a.ts", + vec![sym("a.ts", "a", SYMBOL_BASE)], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE])]), + vec![], + ); + idx.ingest(&[r]).await.unwrap(); + } + + let sgi = Arc::new(SharedGraphIndex::open(Some(db_str.clone())).await.unwrap()); + let idx = sgi.ensure_fresh().await; + assert_eq!(idx.version(), 1); + assert_eq!(idx.symbol_by_id(SYMBOL_BASE).unwrap().name, "a"); + + { + let mut idx = GraphIndex::open(&db_str).await.unwrap(); + let r = result( + "b.ts", + vec![sym("b.ts", "x", SYMBOL_BASE)], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE])]), + vec![], + ); + idx.ingest(&[r]).await.unwrap(); + } + let idx2 = sgi.ensure_fresh().await; + assert_eq!(idx2.version(), 2); + assert_eq!(idx2.stats().symbols, 1); + assert_eq!(idx2.symbol_by_id(SYMBOL_BASE).unwrap().name, "x"); +} + +/// 2 hàm cùng tên khác file → id global riêng, chain giữ nguyên, search trả đủ. +#[tokio::test] +async fn ingest_same_function_name_across_files_stays_distinct() { + let dir = tempfile::tempdir().unwrap(); + let db_dir = dir.path().join("db.lmdb"); + let db_str = format!("lmdb://{}", db_dir.to_string_lossy()); + + let r_store = result( + "store/store.go", + vec![sym("store/store.go", "process", SYMBOL_BASE)], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE])]), + vec![], + ); + let r_cache = result( + "cache/cache.go", + vec![sym("cache/cache.go", "process", SYMBOL_BASE)], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE])]), + vec![], + ); + + let mut idx = GraphIndex::open(&db_str).await.unwrap(); + idx.ingest(&[r_store, r_cache]).await.unwrap(); + + assert_eq!(idx.stats().symbols, 2); + let s1 = idx.symbol_by_id(SYMBOL_BASE).unwrap(); + let s2 = idx.symbol_by_id(SYMBOL_BASE + 1).unwrap(); + assert_eq!(s1.name, "process"); + assert_eq!(s2.name, "process"); + assert_eq!(s1.file, "store/store.go"); + assert_eq!(s2.file, "cache/cache.go"); + + assert_eq!( + idx.flow(SYMBOL_BASE).await.unwrap().chain_desc, + vec!["process"] + ); + assert_eq!( + idx.flow(SYMBOL_BASE + 1).await.unwrap().chain_desc, + vec!["process"] + ); + + let hits = idx + .search_symbol("process", Some(SymbolKind::Function), 10) + .await + .unwrap(); + assert_eq!(hits.len(), 2); + let mut files: Vec<&str> = hits.iter().map(|s| s.file.as_str()).collect(); + files.sort_unstable(); + assert_eq!(files, vec!["cache/cache.go", "store/store.go"]); +} + +/// Regression: LMDB giới hạn key ~511 byte (MDB_BAD_VALSIZE). Path file và +/// call-name vượt giới hạn phải vẫn ingest/reopen đúng (key bound + hash, +/// tên/phí giữ nguyên trong value). +#[tokio::test] +async fn long_path_and_call_name_survive_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let db_str = format!("lmdb://{}/db.lmdb", dir.path().to_string_lossy()); + + // path > 511 byte. + let long_seg = "d".repeat(280); // 280 + let long_path = ["src", &long_seg, &long_seg, "mod.ts"].join("/"); + assert!(long_path.len() > 511, "long_path len = {}", long_path.len()); + + // call_name > 511 byte (mangled symbol). + let long_call = format!( + "RTX{}MangledType0::method{}X", + "t".repeat(300), + "q".repeat(300) + ); + assert!(long_call.len() > 511); + + let calls = vec![CallRecord { + caller_id: SYMBOL_BASE, + call_name: long_call.clone(), + position: 1, + arg_exprs: vec!["x".to_string()], + line: 3, + condition: None, + is_loop_body: false, + effect: EffectType::None, + effect_desc: None, + target_class: None, + target_method: None, + }]; + let r = result( + &long_path, + vec![ + sym(&long_path, "a", SYMBOL_BASE), + sym(&long_path, "b", SYMBOL_BASE + 1), + ], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE, SYMBOL_BASE + 1])]), + calls, + ); + + { + let mut idx = GraphIndex::open(&db_str).await.unwrap(); + idx.ingest(&[r]).await.unwrap(); + assert_eq!(idx.version(), 1); + } + + let idx = GraphIndex::open(&db_str).await.unwrap(); + assert_eq!(idx.version(), 1); + assert_eq!(idx.files().len(), 1); + assert_eq!(idx.files()[0].path, long_path); + assert_eq!(idx.callees(SYMBOL_BASE).await.unwrap()[0].name, "b"); + // call-name index giữ nguyên tên dài sau reopen. + let hits = idx.search_flow(&[SYMBOL_BASE]).await.unwrap(); + assert_eq!(hits.len(), 1); + assert_eq!(hits[0].function_name, "a"); +} diff --git a/crates/codegraph-graph/tests/sqlite.rs b/crates/codegraph-graph/tests/sqlite.rs index b62b8aade..2c61f8f5a 100644 --- a/crates/codegraph-graph/tests/sqlite.rs +++ b/crates/codegraph-graph/tests/sqlite.rs @@ -31,6 +31,13 @@ fn sym(file: &str, name: &str, id: u64) -> Symbol { } } +/// Như `sym` nhưng cho phép chỉ định kind (Method/Variable/...). +fn sym_kind(file: &str, name: &str, id: u64, kind: SymbolKind) -> Symbol { + let mut s = sym(file, name, id); + s.kind = kind; + s +} + fn result( path: &str, symbols: Vec, @@ -53,8 +60,7 @@ fn result( #[tokio::test] async fn index_ingest_reopen_roundtrip() { let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("db.sqlite"); - let path = path.to_string_lossy().into_owned(); + let path = format!("sqlite://{}/db.sqlite", dir.path().to_string_lossy()); let calls = vec![CallRecord { caller_id: SYMBOL_BASE, @@ -114,8 +120,7 @@ async fn index_ingest_reopen_roundtrip() { #[tokio::test] async fn empty_ingest_wipes_store() { let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("db.sqlite"); - let path = path.to_string_lossy().into_owned(); + let path = format!("sqlite://{}/db.sqlite", dir.path().to_string_lossy()); let r = result( "a.ts", @@ -143,7 +148,7 @@ async fn empty_ingest_wipes_store() { async fn shared_index_rebuilds_on_reindex() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("db.sqlite"); - let db_str = db_path.to_string_lossy().into_owned(); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); // "CLI": index dữ liệu đầu. { @@ -158,7 +163,7 @@ async fn shared_index_rebuilds_on_reindex() { } // "Server": shared index trên cùng file. - let sgi = Arc::new(SharedGraphIndex::open(Some(db_path.clone())).await.unwrap()); + let sgi = Arc::new(SharedGraphIndex::open(Some(db_str.clone())).await.unwrap()); let idx = sgi.ensure_fresh().await; assert_eq!(idx.version(), 1); assert_eq!(idx.symbol_by_id(SYMBOL_BASE).unwrap().name, "a"); @@ -189,7 +194,7 @@ async fn shared_index_rebuilds_on_reindex() { async fn ingest_same_function_name_across_files_stays_distinct() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("db.sqlite"); - let db_str = db_path.to_string_lossy().into_owned(); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); // Hai package khác nhau (`store` và `cache`), mỗi package một hàm `process`. let r_store = result( @@ -237,3 +242,99 @@ async fn ingest_same_function_name_across_files_stays_distinct() { files.sort_unstable(); assert_eq!(files, vec!["cache/cache.go", "store/store.go"]); } + +/// Bug D: sandbox lookup entry phải tìm được cả Java `Method`, không chỉ Rust/ +/// Go free `Function`. `codegraph context getProfile` vốn dùng `kind=None` nên +/// resolve được — còn sandbox lọc `Some(SymbolKind::Function)` → "no function +/// matching". `search_symbol_kinds` phải trả về Method. +#[tokio::test] +async fn sandbox_search_kinds_finds_java_method() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("db.sqlite"); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); + + let r = result( + "UserController.java", + vec![sym_kind( + "UserController.java", + "getProfile", + SYMBOL_BASE, + SymbolKind::Method, + )], + HashMap::from([(SYMBOL_BASE, vec![SYMBOL_BASE])]), + vec![], + ); + let mut idx = GraphIndex::open(&db_str).await.unwrap(); + idx.ingest(&[r]).await.unwrap(); + + // Trước fix: lọc Function-only → bỏ Method → empty (sandbox fail). + let only_func = idx + .search_symbol("getProfile", Some(SymbolKind::Function), 1) + .await + .unwrap(); + assert!(only_func.is_empty()); + + // Fix: sandbox chấp nhận Function | Method. + let hits = idx + .search_symbol_kinds("getProfile", &[SymbolKind::Function, SymbolKind::Method], 1) + .await + .unwrap(); + assert_eq!(hits.len(), 1); + assert_eq!(hits[0].name, "getProfile"); + assert_eq!(hits[0].kind, SymbolKind::Method); +} + +/// Bug C: lời gọi method của receiver external không resolve được (`WrapResponse. +/// ok(...)`) KHÔNG được link nhầm vào local variable `boolean ok` trong file +/// khác (fallback tên ngắn từng trả bất kỳ symbol trùng tên, gồm Variable). +/// Chuỗi chỉ còn callee thật `selectDepartment`. +#[tokio::test] +async fn external_qualified_call_not_linked_to_local_variable() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("db.sqlite"); + let db_str = format!("sqlite://{}", db_path.to_string_lossy()); + + let a = SYMBOL_BASE; // getProfile (caller) + let b = SYMBOL_BASE + 1; // selectDepartment (callee thật) + let v = SYMBOL_BASE + 2; // local `boolean ok` (Variable) — KHÔNG được link + + let calls = vec![CallRecord { + caller_id: a, + call_name: "WrapResponse.ok".to_string(), + position: 1, // placeholder 0 trong chain + arg_exprs: vec![], + line: 2, + condition: None, + is_loop_body: false, + effect: EffectType::None, + effect_desc: None, + target_class: None, + target_method: None, + }]; + let r = result( + "UserController.java", + vec![ + sym_kind("UserController.java", "getProfile", a, SymbolKind::Method), + sym_kind( + "UserController.java", + "selectDepartment", + b, + SymbolKind::Method, + ), + sym_kind("HierarchyRefreshWorker.java", "ok", v, SymbolKind::Variable), + ], + HashMap::from([(a, vec![a, 0, b])]), + calls, + ); + let mut idx = GraphIndex::open(&db_str).await.unwrap(); + idx.ingest(&[r]).await.unwrap(); + + let cees = idx.callees(a).await.unwrap(); + assert!( + !cees.iter().any(|s| s.name == "ok"), + "external `WrapResponse.ok` must not resolve to the local `ok` variable" + ); + assert_eq!(cees.len(), 1, "chỉ còn callee thật của getProfile"); + assert_eq!(cees[0].name, "selectDepartment"); + assert_eq!(cees[0].file, "UserController.java"); +} diff --git a/crates/codegraph-mcp/src/lib.rs b/crates/codegraph-mcp/src/lib.rs index 6019b2331..e2c73b128 100644 --- a/crates/codegraph-mcp/src/lib.rs +++ b/crates/codegraph-mcp/src/lib.rs @@ -26,11 +26,8 @@ pub struct McpServer { } impl McpServer { - pub async fn new( - root: camino::Utf8PathBuf, - index_path: Option, - ) -> anyhow::Result { - let shared_index = Arc::new(SharedGraphIndex::open(index_path).await?); + pub async fn new(root: camino::Utf8PathBuf, dsn: Option) -> anyhow::Result { + let shared_index = Arc::new(SharedGraphIndex::open(dsn).await?); Ok(Self { root, shared_index, diff --git a/crates/codegraph-mcp/src/tools.rs b/crates/codegraph-mcp/src/tools.rs index be975cc4e..df81d91aa 100644 --- a/crates/codegraph-mcp/src/tools.rs +++ b/crates/codegraph-mcp/src/tools.rs @@ -2,7 +2,7 @@ use camino::{Utf8Path, Utf8PathBuf}; use codegraph_api::GraphApi; use codegraph_context::{ContextRequest, Format}; use codegraph_core::{is_marker, Error, Result, Symbol, SymbolKind, SymbolMatch}; -use codegraph_extract::{init_project, project_db_path, project_dir, ExtractStats, Orchestrator}; +use codegraph_extract::{init_project, project_dir, ExtractConfig, ExtractStats, Orchestrator}; use codegraph_graph::{GraphIndex, SharedGraphIndex}; use codegraph_sboxes::{compile_with_mocks, BranchPolicy, SboxConfig}; use serde_json::{json, Value}; @@ -635,11 +635,14 @@ pub async fn dispatch_admin(root: &Utf8Path, name: &str, args: Value) -> Result< } } -/// Full re-index: mở sqlite → `Orchestrator::index_all` (ingest = full re-index). -/// Không progress bar — MCP transport là stdout, tránh nhiễu JSON-RPC. +/// Full re-index: mở index theo backend config → `Orchestrator::index_all` +/// (ingest = full re-index). Không progress bar — MCP transport là stdout, +/// tránh nhiễu JSON-RPC. async fn run_index(root: &Utf8Path) -> Result { - let db_str = project_db_path(root).as_str().to_string(); - let mut idx = GraphIndex::open(&db_str).await?; + let mut idx = match ExtractConfig::load(root).storage_dsn(root) { + Some(dsn) => GraphIndex::open(&dsn).await?, + None => GraphIndex::in_memory(), + }; Orchestrator::with_registry() .index_all(root, &mut idx, None) .await @@ -740,7 +743,9 @@ pub async fn dispatch_sandbox( id } else { let q = arg_str(&args, "name")?; - let hits = idx.search_symbol(q, Some(SymbolKind::Function), 1).await?; + let hits = idx + .search_symbol_kinds(q, &[SymbolKind::Function, SymbolKind::Method], 1) + .await?; hits.first() .map(|s| s.id) .ok_or_else(|| Error::Invalid(format!("no function matching `{q}`")))? @@ -812,7 +817,7 @@ async fn run_sim( mocks: &[(String, String)], ) -> Result { let Some(sym) = idx - .search_symbol(entry_name, Some(SymbolKind::Function), 1) + .search_symbol_kinds(entry_name, &[SymbolKind::Function, SymbolKind::Method], 1) .await? .into_iter() .next() diff --git a/crates/codegraph/Cargo.toml b/crates/codegraph/Cargo.toml index 7ded2aef5..e33a15404 100644 --- a/crates/codegraph/Cargo.toml +++ b/crates/codegraph/Cargo.toml @@ -11,9 +11,9 @@ name = "codegraph" path = "src/main.rs" [dependencies] +codegraph-graph = { path = "../codegraph-graph", features = ["sqlite", "lmdb", "bloom-search"] } codegraph-core = { path = "../codegraph-core" } codegraph-extract = { path = "../codegraph-extract" } -codegraph-graph = { path = "../codegraph-graph", features = ["sqlite", "bloom-search"] } codegraph-context = { path = "../codegraph-context" } codegraph-mcp = { path = "../codegraph-mcp" } codegraph-installer = { path = "../codegraph-installer" } diff --git a/crates/codegraph/src/main.rs b/crates/codegraph/src/main.rs index 520732e26..5002092d1 100644 --- a/crates/codegraph/src/main.rs +++ b/crates/codegraph/src/main.rs @@ -143,7 +143,7 @@ fn main() -> Result<()> { } fn cmd_default(root: &Utf8Path) -> Result<()> { - if !db_path(root).exists() { + if !is_initialized(root) { use console::style; eprintln!(); eprintln!( @@ -172,12 +172,11 @@ fn cmd_default(root: &Utf8Path) -> Result<()> { } use console::style; - let db_str = db_path(root).as_str().to_string(); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build()?; let s = rt.block_on(async { - let idx = GraphIndex::open(&db_str).await?; + let idx = open_index(root).await?; Ok::<_, anyhow::Error>(idx.stats()) })?; eprintln!(); @@ -215,12 +214,29 @@ fn cmd_default(root: &Utf8Path) -> Result<()> { Ok(()) } -fn db_path(root: &Utf8Path) -> Utf8PathBuf { - codegraph_extract::project_db_path(root) +/// DSN (kèm scheme) của backend storage trong config — `None` = in-memory. +fn storage_dsn(root: &Utf8Path) -> Option { + codegraph_extract::ExtractConfig::load(root).storage_dsn(root) +} + +/// Mở index theo backend đã config (DSN scheme → `GraphIndex::open`). +async fn open_index(root: &Utf8Path) -> Result { + // `.codegraph/` đã được init (có config) — lúc này storage dsn đã biết. + match storage_dsn(root) { + Some(dsn) => Ok(GraphIndex::open(&dsn).await?), + None => Ok(GraphIndex::in_memory()), + } +} + +/// Workspace đã init chưa — dấu hiệu là thư mục `.codegraph/` tồn tại (do +/// `codegraph init` tạo). Backend-agnostic: không phụ thuộc db file tồn tại +/// (lmdb dùng thư mục, redis không có file địa phương). +fn is_initialized(root: &Utf8Path) -> bool { + codegraph_extract::project_dir(root).exists() } fn ensure_initialized(root: &Utf8Path) -> Result<()> { - if !db_path(root).exists() { + if !is_initialized(root) { use console::style; eprintln!(); eprintln!( @@ -250,15 +266,15 @@ fn ensure_initialized(root: &Utf8Path) -> Result<()> { Ok(()) } -/// Full re-index: mở sqlite → `Orchestrator::index_all` (ingest = full re-index). -fn block_on_index(root: &Utf8Path, db_path: &Utf8Path, progress: bool) -> Result { +/// Full re-index: mở index theo backend config → `Orchestrator::index_all` +/// (ingest = full re-index). +fn block_on_index(root: &Utf8Path, progress: bool) -> Result { let root = root.to_path_buf(); - let db_str = db_path.as_str().to_string(); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build()?; rt.block_on(async { - let mut idx = GraphIndex::open(&db_str).await?; + let mut idx = open_index(&root).await?; // Create progress bar if requested. let progress_bar = if progress { let bar = indicatif::ProgressBar::new(0); @@ -285,7 +301,7 @@ fn cmd_init(root: &Utf8Path, do_index: bool, show_progress: bool) -> Result<()> eprintln!("initialized {}", dir); if do_index { - let stats = block_on_index(root, &db_path(root), show_progress)?; + let stats = block_on_index(root, show_progress)?; eprintln!( "indexed {} files, {} symbols, {} chains, {} edges", stats.files, stats.symbols, stats.chains, stats.calls @@ -398,7 +414,7 @@ fn cmd_uninit(root: &Utf8Path) -> Result<()> { fn cmd_index(root: &Utf8Path, progress: bool) -> Result<()> { ensure_initialized(root)?; - let stats = block_on_index(root, &db_path(root), progress)?; + let stats = block_on_index(root, progress)?; eprintln!( "indexed {} files, {} symbols, {} chains, {} calls (skipped {})", stats.files, stats.symbols, stats.chains, stats.calls, stats.skipped @@ -408,12 +424,11 @@ fn cmd_index(root: &Utf8Path, progress: bool) -> Result<()> { fn cmd_status(root: &Utf8Path) -> Result<()> { ensure_initialized(root)?; - let db_str = db_path(root).as_str().to_string(); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build()?; let s = rt.block_on(async { - let idx = GraphIndex::open(&db_str).await?; + let idx = open_index(root).await?; Ok::<_, anyhow::Error>(idx.stats()) })?; println!("files: {}", s.files); @@ -425,13 +440,12 @@ fn cmd_status(root: &Utf8Path) -> Result<()> { fn cmd_query(root: &Utf8Path, q: &str, limit: u32) -> Result<()> { ensure_initialized(root)?; - let db_str = db_path(root).as_str().to_string(); let q = q.to_string(); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build()?; let hits = rt.block_on(async { - let idx = GraphIndex::open(&db_str).await?; + let idx = open_index(root).await?; Ok::<_, anyhow::Error>(idx.search_symbol(&q, None, limit as usize).await?) })?; for h in hits { @@ -451,13 +465,12 @@ fn cmd_files(root: &Utf8Path, prefix: Option<&str>) -> Result<()> { use std::io::Write; ensure_initialized(root)?; - let db_str = db_path(root).as_str().to_string(); let prefix = prefix.unwrap_or("").to_string(); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build()?; let files = rt.block_on(async { - let idx = GraphIndex::open(&db_str).await?; + let idx = open_index(root).await?; let all = idx.files(); Ok::<_, anyhow::Error>(if prefix.is_empty() { all @@ -478,7 +491,7 @@ fn cmd_files(root: &Utf8Path, prefix: Option<&str>) -> Result<()> { fn cmd_context(root: &Utf8Path, target: &str, depth: u32, include_source: bool) -> Result<()> { ensure_initialized(root)?; - let db_path = db_path(root); + let dsn = storage_dsn(root); let req = codegraph_context::ContextRequest { query: target.into(), depth, @@ -490,9 +503,7 @@ fn cmd_context(root: &Utf8Path, target: &str, depth: u32, include_source: bool) .enable_all() .build()?; let output = rt.block_on(async { - let sgi = Arc::new( - codegraph_graph::SharedGraphIndex::open(Some(db_path.into_std_path_buf())).await?, - ); + let sgi = Arc::new(codegraph_graph::SharedGraphIndex::open(dsn).await?); codegraph_context::build(&sgi, &req).await })?; print!("{}", output); @@ -504,14 +515,13 @@ fn cmd_serve(root: &Utf8Path, mcp: bool) -> Result<()> { return Err(anyhow!("only --mcp transport supported")); } ensure_initialized(root).context("init the index before serving")?; - let db_path = db_path(root); + let dsn = storage_dsn(root); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build()?; rt.block_on(async { - watcher::spawn(root.to_path_buf(), db_path.clone()); - let mcp_server = - McpServer::new(root.to_path_buf(), Some(db_path.into_std_path_buf())).await?; + watcher::spawn(root.to_path_buf(), dsn.clone()); + let mcp_server = McpServer::new(root.to_path_buf(), dsn).await?; mcp_server.run_stdio().await })?; Ok(()) @@ -524,7 +534,7 @@ fn cmd_sandbox(root: &Utf8Path, function: &str, args: &str, quiet: bool) -> Resu use codegraph_sboxes::SboxConfig; ensure_initialized(root)?; - let db_path = db_path(root); + let dsn = storage_dsn(root); let function = function.to_string(); let args: Vec = args .split(',') @@ -536,14 +546,12 @@ fn cmd_sandbox(root: &Utf8Path, function: &str, args: &str, quiet: bool) -> Resu .enable_all() .build()?; let (ret, trace, group_names) = rt.block_on(async { - let sgi = Arc::new( - codegraph_graph::SharedGraphIndex::open(Some(db_path.into_std_path_buf())).await?, - ); + let sgi = Arc::new(codegraph_graph::SharedGraphIndex::open(dsn).await?); let idx = sgi.ensure_fresh().await; // Resolve the entry function (substring, first function match). let hits = idx - .search_symbol(&function, Some(SymbolKind::Function), 1) + .search_symbol_kinds(&function, &[SymbolKind::Function, SymbolKind::Method], 1) .await?; let entry = hits .first() diff --git a/crates/codegraph/src/watcher.rs b/crates/codegraph/src/watcher.rs index f35dde52f..30af8a8fb 100644 --- a/crates/codegraph/src/watcher.rs +++ b/crates/codegraph/src/watcher.rs @@ -10,15 +10,17 @@ use std::time::Duration; /// Spawn a debounced watcher that full re-indexes the workspace on file changes. /// Runs on a background tokio task; cancellation when the runtime drops. -pub fn spawn(root: Utf8PathBuf, db_path: Utf8PathBuf) { +/// `dsn = None` (in-memory backend) → không có file ngoài để theo dõi, bỏ qua. +pub fn spawn(root: Utf8PathBuf, dsn: Option) { + let Some(dsn) = dsn else { return }; tokio::task::spawn_blocking(move || { - if let Err(e) = run(root, db_path) { + if let Err(e) = run(root, dsn) { tracing::error!("watcher error: {e}"); } }); } -fn run(root: Utf8PathBuf, db_path: Utf8PathBuf) -> Result<()> { +fn run(root: Utf8PathBuf, dsn: String) -> Result<()> { let (tx, rx) = std::sync::mpsc::channel::>(); let mut debouncer = new_debouncer( Duration::from_millis(500), @@ -57,9 +59,8 @@ fn run(root: Utf8PathBuf, db_path: Utf8PathBuf) -> Result<()> { } // Full re-index (đã chốt — bỏ incremental): bất kỳ thay đổi nào cũng // index lại toàn bộ (ingest reset + rebuild engine). - let db_str = db_path.as_str().to_string(); let result = handle.block_on(async { - let mut idx = GraphIndex::open(&db_str).await?; + let mut idx = GraphIndex::open(&dsn).await?; orch.index_all(&root, &mut idx, None).await }); match result {