From cc7044afafe371cf10c5fa929d8692c58faba9f8 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:12:23 -0800 Subject: [PATCH] call_edges: replace edges per file version Callers lookups miss callers at some commits a `--git` run indexed, and the `call_edges` row count differs between identical `--git` runs. `sync_for_files()` deletes every `call_edges` row of each path in a batch, at every version, before appending the batch's edges. A path has one version per commit that changed it, and parallel inserters write those versions in different batches, within one run or across runs. Each batch therefore deletes the edges of the path's other versions, which callers lookups need at the revisions that hold those versions. Which versions survive depends on batch order, which is why the row count varies. `sync_for_versions()` replaces `sync_for_files()` and deletes only the versions the batch writes, with one `IN` list over the key `concat(caller_file_hash, ':', caller_file)`. The key is exact because a git blob id holds no `:`. A process-global lock now serializes `call_edges` writes, as content upserts already are. Without it, concurrent deletes on one table conflict and retry, and two syncs of one version could both delete before either appends, leaving its rows twice. The backfill runs only on an empty table and appends without deleting. `tests/call_edge_versions.rs` fails on main and passes here. It indexes `c0..c1` and `c1..c2` as two `process_git_range()` calls and checks the callers at each commit. The same-version concurrency unit test fails on every run with the lock removed. --- src/database/call_edges.rs | 409 ++++++++++++++++++++++++++++++++++-- src/database/connection.rs | 18 +- tests/call_edge_versions.rs | 128 +++++++++++ 3 files changed, 525 insertions(+), 30 deletions(-) create mode 100644 tests/call_edge_versions.rs diff --git a/src/database/call_edges.rs b/src/database/call_edges.rs index e6c2cbe..0814424 100644 --- a/src/database/call_edges.rs +++ b/src/database/call_edges.rs @@ -8,19 +8,32 @@ // this table is a derived acceleration structure holding the same facts. // Every row carries the caller's file hash so readers apply the same // manifest check the LIKE path applies, and stale rows from older file -// versions stay invisible. Lifecycle follows the functions table: edges -// are deleted per file and rewritten whenever that file's functions are -// inserted, so a removed call's row cannot outlive its function row. +// versions stay invisible. Lifecycle follows the functions table: a file +// version's edges are deleted and rewritten whenever that version's +// functions are inserted, so a removed call's row cannot outlive its +// function row. Rows of the same path at other versions stay, and the +// manifest check hides them from revisions that do not hold that version. use anyhow::Result; use arrow::array::{Array, ArrayRef, Int64Builder, RecordBatch, StringArray, StringBuilder}; use futures::TryStreamExt; use lancedb::connection::Connection; use lancedb::query::{ExecutableQuery, QueryBase}; +use once_cell::sync::Lazy; use std::sync::Arc; +use tokio::sync::Mutex; use crate::database::get_column; use crate::types::{CallEdge, FunctionInfo}; +/// Process-global lock serializing every write to `call_edges`. +/// +/// Inserters write the table concurrently. Two deletes on it conflict and +/// retry, and a delete racing an append of the same file version can leave +/// that version's rows twice. Process-global because several +/// `CallEdgeStore` instances share one database, so a per-instance lock +/// would not serialize them. +static CALL_EDGE_WRITE_LOCK: Lazy> = Lazy::new(|| Mutex::new(())); + pub struct CallEdgeStore { connection: Connection, } @@ -108,21 +121,60 @@ impl CallEdgeStore { out } - /// Exact lifecycle: drop every edge credited to these files, then write - /// the fresh set. A removed call's row cannot survive its file's - /// re-index, and unchanged files are never touched. - pub async fn sync_for_files(&self, files: &[String], edges: Vec) -> Result<()> { + /// The (caller_file, caller_file_hash) pairs a batch of functions + /// covers, sorted and deduplicated: the file versions whose edges + /// [`Self::sync_for_versions`] replaces. + pub fn versions_of(functions: &[FunctionInfo]) -> Vec<(String, String)> { + let mut versions: Vec<(String, String)> = functions + .iter() + .map(|func| (func.file_path.clone(), func.git_file_hash.clone())) + .collect(); + versions.sort(); + versions.dedup(); + versions + } + + /// Exact lifecycle per file version: drop every edge credited to these + /// (caller_file, caller_file_hash) pairs, then write the fresh set. A + /// removed call's row cannot survive its version's re-index, and the + /// same path's rows at other versions are never touched. + pub async fn sync_for_versions( + &self, + versions: &[(String, String)], + edges: Vec, + ) -> Result<()> { + let _write_guard = CALL_EDGE_WRITE_LOCK.lock().await; let table = self.connection.open_table("call_edges").execute().await?; - if !files.is_empty() { - let list: Vec = files + if !versions.is_empty() { + // One IN list over a `hash:path` key: a single hash-set probe + // per row, where an OR of per-version comparisons costs one + // comparison per version per row. The key is exact because the + // hash comes first and a git blob id holds no `:`, so the first + // `:` splits every key the same way whatever the path holds. + // File paths and hashes come from the index itself, not user + // input. + let keys: Vec = versions .iter() - .map(|f| format!("'{}'", f.replace('\'', "''"))) + .map(|(file, hash)| { + format!( + "'{}:{}'", + hash.replace('\'', "''"), + file.replace('\'', "''") + ) + }) .collect(); - // File paths come from the index itself, not user input. table - .delete(&format!("caller_file IN ({})", list.join(", "))) + .delete(&format!( + "concat(caller_file_hash, ':', caller_file) IN ({})", + keys.join(", ") + )) .await?; } + Self::append_edges(&table, edges).await + } + + /// Append edge rows. The caller holds [`CALL_EDGE_WRITE_LOCK`]. + async fn append_edges(table: &lancedb::Table, edges: Vec) -> Result<()> { if edges.is_empty() { return Ok(()); } @@ -205,12 +257,12 @@ impl CallEdgeStore { calls: Some(calls.clone()), }) .collect(); - let mut files: Vec = sources.iter().map(|s| s.file.to_string()).collect(); - files.sort(); - files.dedup(); let edges = Self::edges_for_sources(&sources); let n = edges.len(); - self.sync_for_files(&files, edges).await?; + // The table is empty, so there is nothing to delete first. + let _write_guard = CALL_EDGE_WRITE_LOCK.lock().await; + let edges_table = self.connection.open_table("call_edges").execute().await?; + Self::append_edges(&edges_table, edges).await?; Ok(n) } @@ -260,3 +312,328 @@ impl CallEdgeStore { Ok(out) } } + +#[cfg(test)] +mod tests { + use super::*; + use lancedb::connection::LanceFileVersion; + use lancedb::database::listing::ListingDatabaseOptions; + + async fn fresh_store() -> (tempfile::TempDir, CallEdgeStore) { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("call-edges.lancedb"); + let options = ListingDatabaseOptions::builder() + .data_storage_version(LanceFileVersion::V2_2) + .build(); + let connection = lancedb::connect(db_path.to_str().unwrap()) + .database_options(&options) + .execute() + .await + .unwrap(); + crate::database::schema::SchemaManager::new(connection.clone()) + .create_all_tables() + .await + .unwrap(); + (dir, CallEdgeStore::new(connection)) + } + + /// One plain edge from `caller` in `file` at blob `hash` to `callee`. + fn edge(file: &str, hash: &str, caller: &str, callee: &str) -> StoredEdge { + StoredEdge { + caller_name: caller.to_string(), + caller_file: file.to_string(), + caller_line: 1, + caller_file_hash: hash.to_string(), + callee: callee.to_string(), + owner: None, + kind: "plain", + } + } + + fn version(file: &str, hash: &str) -> (String, String) { + (file.to_string(), hash.to_string()) + } + + /// Every row as (caller_file, caller_file_hash, caller_name, callee), + /// sorted, duplicates kept. + async fn all_rows(store: &CallEdgeStore) -> Vec<(String, String, String, String)> { + let table = store + .connection + .open_table("call_edges") + .execute() + .await + .unwrap(); + let batches = table + .query() + .execute() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let mut rows = Vec::new(); + for batch in &batches { + let col = |name: &str| -> &StringArray { get_column(batch, name).unwrap() }; + let (file, hash, caller, callee) = ( + col("caller_file"), + col("caller_file_hash"), + col("caller_name"), + col("callee"), + ); + for i in 0..batch.num_rows() { + rows.push(( + file.value(i).to_string(), + hash.value(i).to_string(), + caller.value(i).to_string(), + callee.value(i).to_string(), + )); + } + } + rows.sort(); + rows + } + + fn row(file: &str, hash: &str, caller: &str, callee: &str) -> (String, String, String, String) { + ( + file.to_string(), + hash.to_string(), + caller.to_string(), + callee.to_string(), + ) + } + + #[tokio::test] + async fn syncing_a_second_version_keeps_the_first() { + let (_dir, store) = fresh_store().await; + store + .sync_for_versions(&[version("a.c", "h1")], vec![edge("a.c", "h1", "f", "g")]) + .await + .unwrap(); + store + .sync_for_versions( + &[version("a.c", "h2")], + vec![edge("a.c", "h2", "f", "g"), edge("a.c", "h2", "f", "h")], + ) + .await + .unwrap(); + + assert_eq!( + all_rows(&store).await, + [ + row("a.c", "h1", "f", "g"), + row("a.c", "h2", "f", "g"), + row("a.c", "h2", "f", "h"), + ] + ); + } + + #[tokio::test] + async fn resyncing_a_version_replaces_its_edges() { + let (_dir, store) = fresh_store().await; + store + .sync_for_versions( + &[version("a.c", "h1"), version("a.c", "h2")], + vec![ + edge("a.c", "h1", "f", "g"), + edge("a.c", "h1", "f", "h"), + edge("a.c", "h2", "f", "g"), + ], + ) + .await + .unwrap(); + // h1 again, without its call to h. + store + .sync_for_versions(&[version("a.c", "h1")], vec![edge("a.c", "h1", "f", "g")]) + .await + .unwrap(); + + assert_eq!( + all_rows(&store).await, + [row("a.c", "h1", "f", "g"), row("a.c", "h2", "f", "g")] + ); + } + + #[tokio::test] + async fn same_blob_at_another_path_is_left_alone() { + let (_dir, store) = fresh_store().await; + store + .sync_for_versions( + &[version("a.c", "same"), version("b.c", "same")], + vec![edge("a.c", "same", "f", "g"), edge("b.c", "same", "f", "g")], + ) + .await + .unwrap(); + // a.c's version loses its edges; b.c holds the same blob and keeps + // its own. + store + .sync_for_versions(&[version("a.c", "same")], Vec::new()) + .await + .unwrap(); + + assert_eq!(all_rows(&store).await, [row("b.c", "same", "f", "g")]); + } + + #[tokio::test] + async fn paths_and_hashes_with_quotes_are_matched_literally() { + let (_dir, store) = fresh_store().await; + store + .sync_for_versions( + &[version("it's.c", "h'1"), version("its.c", "h1")], + vec![ + edge("it's.c", "h'1", "f", "g"), + edge("its.c", "h1", "f", "g"), + ], + ) + .await + .unwrap(); + store + .sync_for_versions(&[version("it's.c", "h'1")], Vec::new()) + .await + .unwrap(); + + assert_eq!(all_rows(&store).await, [row("its.c", "h1", "f", "g")]); + } + + /// A `:` in a path cannot make one version's key match another's, + /// because the hash before the first `:` holds none. + #[tokio::test] + async fn colons_in_paths_match_only_their_own_version() { + let (_dir, store) = fresh_store().await; + store + .sync_for_versions( + &[ + version("a.c", "h1"), + version("dir:a.c", "h1"), + version("a.c:x", "h1"), + ], + vec![ + edge("a.c", "h1", "f", "g"), + edge("dir:a.c", "h1", "f", "g"), + edge("a.c:x", "h1", "f", "g"), + ], + ) + .await + .unwrap(); + store + .sync_for_versions(&[version("dir:a.c", "h1")], Vec::new()) + .await + .unwrap(); + + assert_eq!( + all_rows(&store).await, + [row("a.c", "h1", "f", "g"), row("a.c:x", "h1", "f", "g")] + ); + } + + #[tokio::test] + async fn a_large_batch_of_versions_is_replaced_in_one_sync() { + let (_dir, store) = fresh_store().await; + // Past a `--git` batch's 512 files. + let count = 1025; + let versions: Vec<(String, String)> = (0..count) + .map(|i| version(&format!("f{i:04}.c"), "h1")) + .collect(); + let edges_to = |callee: &str| -> Vec { + versions + .iter() + .map(|(file, hash)| edge(file, hash, "f", callee)) + .collect() + }; + store + .sync_for_versions(&versions, edges_to("old")) + .await + .unwrap(); + store + .sync_for_versions(&versions, edges_to("new")) + .await + .unwrap(); + + let rows = all_rows(&store).await; + assert_eq!(rows.len(), count); + assert!(rows.iter().all(|(_, _, _, callee)| callee == "new")); + } + + #[test] + fn versions_of_is_sorted_and_deduplicated() { + let function = |file: &str, hash: &str, name: &str| FunctionInfo { + name: name.to_string(), + file_path: file.to_string(), + git_file_hash: hash.to_string(), + line_start: 1, + line_end: 3, + return_type: "int".to_string(), + parameters: Vec::new(), + body: String::new(), + calls: None, + types: None, + guard: None, + qualifier: None, + }; + let versions = CallEdgeStore::versions_of(&[ + function("b.c", "h1", "x"), + function("a.c", "h2", "y"), + function("b.c", "h1", "z"), + function("a.c", "h1", "w"), + ]); + assert_eq!( + versions, + [ + version("a.c", "h1"), + version("a.c", "h2"), + version("b.c", "h1") + ] + ); + } + + /// Two inserters syncing different versions of one path at once must + /// leave both versions' rows, each exactly once. Fresh versions every + /// round so one lucky interleaving cannot hide a lost or doubled row. + #[tokio::test] + async fn concurrent_syncs_of_different_versions_keep_every_row_once() { + const ROUNDS: usize = 20; + let (_dir, store) = fresh_store().await; + let mut expected = Vec::new(); + for round in 0..ROUNDS { + let (h1, h2) = (format!("r{round}a"), format!("r{round}b")); + let edges = |hash: &str| vec![edge("a.c", hash, "f", "g"), edge("a.c", hash, "f", "h")]; + for hash in [&h1, &h2] { + expected.push(row("a.c", hash, "f", "g")); + expected.push(row("a.c", hash, "f", "h")); + } + let (v1, v2) = ([version("a.c", &h1)], [version("a.c", &h2)]); + let (first, second) = tokio::join!( + store.sync_for_versions(&v1, edges(&h1)), + store.sync_for_versions(&v2, edges(&h2)), + ); + first.unwrap(); + second.unwrap(); + } + expected.sort(); + assert_eq!(all_rows(&store).await, expected); + } + + /// Two syncs of the same version at once must leave its rows once. + /// Indexing sends each blob once per run, so this holds the store to + /// its own contract rather than relying on that: without the write + /// lock both deletes run before either append and both appends land. + #[tokio::test] + async fn concurrent_syncs_of_one_version_leave_no_duplicates() { + const ROUNDS: usize = 20; + let (_dir, store) = fresh_store().await; + let mut expected = Vec::new(); + for round in 0..ROUNDS { + let hash = format!("r{round}"); + let edges = || vec![edge("a.c", &hash, "f", "g")]; + expected.push(row("a.c", &hash, "f", "g")); + let versions = [version("a.c", &hash)]; + let (first, second) = tokio::join!( + store.sync_for_versions(&versions, edges()), + store.sync_for_versions(&versions, edges()), + ); + first.unwrap(); + second.unwrap(); + } + expected.sort(); + assert_eq!(all_rows(&store).await, expected); + } +} diff --git a/src/database/connection.rs b/src/database/connection.rs index 22d8d48..1ea121f 100644 --- a/src/database/connection.rs +++ b/src/database/connection.rs @@ -1890,18 +1890,13 @@ impl DatabaseManager { // Normalized call edges for indexed reverse lookup, derived from // the same parsed calls the functions rows serialize. - let edge_files: Vec = { - let mut files: Vec = functions.iter().map(|f| f.file_path.clone()).collect(); - files.sort(); - files.dedup(); - files - }; + let edge_versions = crate::database::call_edges::CallEdgeStore::versions_of(&functions); let edges = crate::database::call_edges::CallEdgeStore::edges_for(&functions); // Insert functions self.function_store.insert_batch(functions).await?; self.call_edge_store - .sync_for_files(&edge_files, edges) + .sync_for_versions(&edge_versions, edges) .await?; self.object_macro_store.insert_batch(object_macros).await?; @@ -3675,16 +3670,11 @@ impl DatabaseManager { // Metadata-only insertion methods (skip content storage for performance) async fn insert_functions_metadata_only(&self, functions: Vec) -> Result<()> { // Insert functions metadata - let edge_files: Vec = { - let mut files: Vec = functions.iter().map(|f| f.file_path.clone()).collect(); - files.sort(); - files.dedup(); - files - }; + let edge_versions = crate::database::call_edges::CallEdgeStore::versions_of(&functions); let edges = crate::database::call_edges::CallEdgeStore::edges_for(&functions); self.function_store.insert_metadata_only(functions).await?; self.call_edge_store - .sync_for_files(&edge_files, edges) + .sync_for_versions(&edge_versions, edges) .await?; Ok(()) } diff --git a/tests/call_edge_versions.rs b/tests/call_edge_versions.rs new file mode 100644 index 0000000..59ee97d --- /dev/null +++ b/tests/call_edge_versions.rs @@ -0,0 +1,128 @@ +// SPDX-License-Identifier: MIT OR Apache-2.0 +// +// Callers of a function at each revision of a file indexed in separate +// `--git` runs: every version keeps its own call edges. +use semcode::{git, DatabaseManager}; +use std::path::Path; +use std::process::Command; +use std::sync::Arc; + +fn git_run(repo: &Path, args: &[&str]) { + let status = Command::new("git") + .args(args) + .current_dir(repo) + // Do not inherit developer's git configuration. + .env("GIT_CONFIG_GLOBAL", "/dev/null") + .env("GIT_CONFIG_SYSTEM", "/dev/null") + .env("GIT_AUTHOR_NAME", "Semcode Test") + .env("GIT_AUTHOR_EMAIL", "semcode@example.com") + .env("GIT_COMMITTER_NAME", "Semcode Test") + .env("GIT_COMMITTER_EMAIL", "semcode@example.com") + .status() + .unwrap(); + assert!(status.success(), "git {args:?} failed"); +} + +fn head_of(repo: &Path) -> String { + git::get_git_sha(repo).unwrap().unwrap() +} + +fn commit_all(repo: &Path, message: &str) -> String { + git_run(repo, &["add", "."]); + git_run(repo, &["commit", "-q", "-m", message]); + head_of(repo) +} + +/// `f` as written at one revision: its body calls each of `callees`. +fn caller_source(callees: &[&str]) -> String { + let calls: String = callees + .iter() + .map(|callee| format!("\ttotal += {callee}(total);\n")) + .collect(); + format!( + "int f(int seed)\n{{\n\ +\tint total = seed * 3 + 1;\n\ +{calls}\ +\tif (total < 0)\n\ +\t\treturn -1;\n\ +\treturn total;\n}}\n" + ) +} + +/// Names of the database's callers of `callee` at `revision`. +/// +/// Asks the index alone through the revision's manifest, so the working +/// copy cannot contribute an answer the index lost. +async fn callers_at(db: &DatabaseManager, callee: &str, revision: &str) -> Vec { + let manifest = db.git_manifest_cached(revision).await.unwrap(); + let mut names: Vec = db + .get_function_callers_with_manifest(callee, &manifest) + .await + .unwrap() + .into_iter() + .map(|(name, _file)| name) + .collect(); + names.sort(); + names.dedup(); + names +} + +/// `a.c` changes in c1 and again in c2, and each change is indexed by its +/// own `process_git_range()` call. The second run writes the edges of the +/// second version of `a.c`; the first version's edges must still answer +/// for c1. +/// +/// Two runs rather than one `c0..c2` range: one run could put both versions +/// in one batch, where the defect this guards against does not show. +#[tokio::test] +async fn each_file_version_keeps_its_callers_across_git_runs() { + let dir = tempfile::tempdir().unwrap(); + let repo = dir.path(); + + git_run(repo, &["init", "-q"]); + std::fs::write( + repo.join("lib.c"), + "int g(int x)\n{\n\treturn x + 1;\n}\n\nint h(int x)\n{\n\treturn x * 2;\n}\n", + ) + .unwrap(); + let c0 = commit_all(repo, "c0: lib.c"); + + std::fs::write(repo.join("a.c"), caller_source(&["g"])).unwrap(); + let c1 = commit_all(repo, "c1: f calls g"); + + std::fs::write(repo.join("a.c"), caller_source(&["g", "h"])).unwrap(); + let c2 = commit_all(repo, "c2: f calls g and h"); + + let db = Arc::new( + DatabaseManager::new( + repo.join(".semcode.db").to_str().unwrap(), + repo.to_string_lossy().into_owned(), + ) + .await + .unwrap(), + ); + db.create_tables().await.unwrap(); + + let extensions = ["c".to_string(), "h".to_string()]; + for range in [format!("{c0}..{c1}"), format!("{c1}..{c2}")] { + semcode::git_range::process_git_range( + &repo.to_path_buf(), + &range, + &extensions, + db.clone(), + false, + 1, + &[], + ) + .await + .unwrap(); + } + + assert_eq!(callers_at(&db, "g", &c1).await, ["f"], "callers of g at c1"); + assert_eq!(callers_at(&db, "g", &c2).await, ["f"], "callers of g at c2"); + assert_eq!(callers_at(&db, "h", &c2).await, ["f"], "callers of h at c2"); + assert!( + callers_at(&db, "h", &c1).await.is_empty(), + "f does not call h at c1" + ); +}