fix(quality): production hardening — race condition, unbounded queries, DRY, observability

- Fix embedding thread / rebuild race: wait for background embedding before
  clearing data in tool_rebuild_index to prevent orphaned vectors
- Batch-fetch unembedded nodes with LIMIT to bound memory usage
- Add LIMIT 200 to project_map key-symbols query (prevent full table scan)
- Enable RUST_LOG env filter for runtime log level control
- Extract FTS5 trigger SQL to shared function (was copy-pasted 4x in schema.rs)
- Extract is_test_symbol helper (was duplicated 3x in server.rs)
- Add anyOf required constraint to get_ast_node tool schema
- Replace bare .unwrap() with .expect() in parser cache
- Mark test-only functions with #[cfg(test)]: update_context_string, rrf_fusion
- Fix pre-existing clippy nonminimal_bool warning
- Bump version to 0.5.17

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: sdsrss

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "code-graph-mcp"
-version = "0.5.16"
+version = "0.5.17"
 edition = "2021"
 
 [features]
@@ -18,7 +18,7 @@ ignore = "0.4"
 rayon = "1"
 notify = "6"
 tracing = "0.1"
-tracing-subscriber = "0.3"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 anyhow = "1"
 bytemuck = "1"
 candle-core = { version = "0.9", optional = true }

Cargo.toml

@@ -1,6 +1,6 @@
 {
   "name": "@sdsrs/code-graph",
-  "version": "0.5.16",
+  "version": "0.5.17",
   "description": "MCP server that indexes codebases into an AST knowledge graph with semantic search, call graph traversal, and HTTP route tracing",
   "license": "MIT",
   "repository": {

package.json

@@ -63,6 +63,10 @@ fn print_help() {
 
 fn run_serve() -> Result<()> {
     tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"))
+        )
         .with_writer(io::stderr)
         .init();
 

src/main.rs

@@ -25,6 +25,11 @@ fn required_str<'a>(args: &'a serde_json::Value, key: &str) -> Result<&'a str> {
     Ok(s)
 }
 
+/// Whether a symbol is a test-only symbol (by name or file path convention).
+fn is_test_symbol(name: &str, file_path: &str) -> bool {
+    name.starts_with("test_") || file_path.starts_with("tests/")
+}
+
 /// Parse route input like "GET /api/users" into (Some("GET"), "/api/users").
 /// If no method prefix, returns (None, original_path).
 fn parse_route_input(input: &str) -> (Option<String>, &str) {
@@ -356,27 +361,29 @@ impl McpServer {
                     None => return Ok(()),
                 };
                 let db = Database::open_with_vec(&db_path)?;
-                let unembedded = queries::get_unembedded_nodes(db.conn())?;
-                if unembedded.is_empty() {
-                    return Ok(());
-                }
 
-                let total = unembedded.len();
-                tracing::info!("[embed-bg] Embedding {} nodes in background...", total);
+                const EMBED_BATCH: usize = 32;
+                let mut total_embedded = 0usize;
                 let t0 = std::time::Instant::now();
 
-                const EMBED_PROGRESS_BATCH: usize = 32;
-                for (i, chunk) in unembedded.chunks(EMBED_PROGRESS_BATCH).enumerate() {
+                loop {
+                    let chunk = queries::get_unembedded_nodes(db.conn(), EMBED_BATCH)?;
+                    if chunk.is_empty() {
+                        break;
+                    }
+
                     let tx = db.conn().unchecked_transaction()?;
-                    embed_and_store_batch(&db, &model, chunk)?;
+                    embed_and_store_batch(&db, &model, &chunk)?;
                     tx.commit()?;
 
-                    let done = ((i + 1) * EMBED_PROGRESS_BATCH).min(total);
-                    tracing::info!("[embed-bg] Progress: {}/{} nodes", done, total);
+                    total_embedded += chunk.len();
+                    tracing::info!("[embed-bg] Progress: {} nodes embedded", total_embedded);
                 }
 
-                tracing::info!("[embed-bg] Complete: {} nodes in {:.1}s",
-                    total, t0.elapsed().as_secs_f64());
+                if total_embedded > 0 {
+                    tracing::info!("[embed-bg] Complete: {} nodes in {:.1}s",
+                        total_embedded, t0.elapsed().as_secs_f64());
+                }
                 Ok(())
             })();
 
@@ -438,7 +445,7 @@ impl McpServer {
     fn resolve_fuzzy_name(&self, name: &str) -> Result<FuzzyResolution> {
         let candidates: Vec<_> = queries::find_functions_by_fuzzy_name(self.db.conn(), name)?
             .into_iter()
-            .filter(|c| !c.name.starts_with("test_") && !c.file_path.starts_with("tests/"))
+            .filter(|c| !is_test_symbol(&c.name, &c.file_path))
             .collect();
         if candidates.len() == 1 {
             Ok(FuzzyResolution::Unique(candidates.into_iter().next().unwrap().name))
@@ -1133,7 +1140,7 @@ impl McpServer {
         // If exact match returns empty (only seed node, no edges), try fuzzy name resolution
         let has_edges = results.iter().any(|n| n.depth > 0);
         let has_seed = results.iter().any(|n| n.depth == 0);
-        if !has_edges && !(has_seed && file_path.is_some()) {
+        if !(has_edges || has_seed && file_path.is_some()) {
             match self.resolve_fuzzy_name(function_name)? {
                 FuzzyResolution::Unique(resolved) => {
                     let results2 = crate::graph::query::get_call_graph(
@@ -1177,7 +1184,7 @@ impl McpServer {
         results: &[crate::graph::query::CallGraphNode],
     ) -> Result<serde_json::Value> {
         let is_test = |n: &&crate::graph::query::CallGraphNode| {
-            n.name.starts_with("test_") || n.file_path.starts_with("tests/")
+            is_test_symbol(&n.name, &n.file_path)
         };
         let all_nodes: Vec<serde_json::Value> = results.iter()
             .filter(|n| n.depth > 0 && !is_test(n))
@@ -1585,6 +1592,16 @@ impl McpServer {
         let project_root = self.project_root.as_ref()
             .ok_or_else(|| anyhow!("No project root configured"))?;
 
+        // Wait for background embedding to finish before clearing data
+        // to avoid race where embedding thread writes vectors for deleted nodes
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+        while self.embedding_in_progress.load(Ordering::Acquire) {
+            if std::time::Instant::now() > deadline {
+                return Err(anyhow!("Background embedding still in progress. Try again shortly."));
+            }
+            std::thread::sleep(std::time::Duration::from_millis(100));
+        }
+
         // Clear all data in a single transaction (CASCADE handles nodes→edges)
         {
             let tx = self.db.conn().unchecked_transaction()?;
@@ -1685,7 +1702,7 @@ impl McpServer {
 
         // Separate production callers from test callers
         let is_test = |c: &&queries::CallerWithRouteInfo| {
-            c.name.starts_with("test_") || c.file_path.starts_with("tests/")
+            is_test_symbol(&c.name, &c.file_path)
         };
         let prod_callers: Vec<_> = callers.iter().filter(|c| !is_test(c)).collect();
         let test_callers: Vec<_> = callers.iter().filter(|c| is_test(c)).collect();

src/mcp/server.rs

@@ -73,7 +73,11 @@ impl ToolRegistry {
                         "node_id": { "type": "number", "description": "Node ID (alternative to file_path+symbol_name)" },
                         "include_references": { "type": "boolean", "description": "Include callers/callees (default false)" },
                         "context_lines": { "type": "number", "description": "Surrounding source lines to include (default 0)" }
-                    }
+                    },
+                    "anyOf": [
+                        { "required": ["symbol_name"] },
+                        { "required": ["node_id"] }
+                    ]
                 }),
             },
             ToolDefinition {

src/mcp/tools.rs

@@ -38,7 +38,8 @@ pub fn parse_tree(source: &str, language: &str) -> Result<tree_sitter::Tree> {
             p.set_language(&lang)?;
             cache.insert(language.to_string(), p);
         }
-        let parser = cache.get_mut(language).unwrap();
+        let parser = cache.get_mut(language)
+            .expect("parser was just inserted");
         parser.parse(source, None)
             .ok_or_else(|| anyhow!("parse failed or timed out"))
     })

src/parser/treesitter.rs

@@ -36,6 +36,7 @@ pub fn weighted_rrf_fusion(
     results
 }
 
+#[cfg(test)]
 pub fn rrf_fusion(
     fts_results: &[SearchResult],
     vec_results: &[SearchResult],

src/search/fusion.rs

@@ -237,6 +237,7 @@ pub fn delete_nodes_by_file(conn: &Connection, file_id: i64) -> Result<()> {
     Ok(())
 }
 
+#[cfg(test)]
 pub fn update_context_string(conn: &Connection, node_id: i64, context_string: &str) -> Result<()> {
     conn.execute(
         "UPDATE nodes SET context_string = ?1 WHERE id = ?2",
@@ -882,13 +883,15 @@ pub fn get_import_tree(
 // --- Unembedded nodes ---
 
 /// Get (node_id, context_string) for nodes that have context strings but no vectors.
-pub fn get_unembedded_nodes(conn: &Connection) -> Result<Vec<(i64, String)>> {
+/// Returns at most `limit` rows per call to bound memory usage.
+pub fn get_unembedded_nodes(conn: &Connection, limit: usize) -> Result<Vec<(i64, String)>> {
     let mut stmt = conn.prepare(
         "SELECT n.id, n.context_string FROM nodes n
          LEFT JOIN node_vectors v ON v.node_id = n.id
-         WHERE n.context_string IS NOT NULL AND v.node_id IS NULL"
+         WHERE n.context_string IS NOT NULL AND v.node_id IS NULL
+         LIMIT ?1"
     )?;
-    let rows = stmt.query_map([], |row| {
+    let rows = stmt.query_map([limit as i64], |row| {
         Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?))
     })?;
     rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
@@ -1007,7 +1010,8 @@ pub fn get_project_map(conn: &Connection) -> Result<(Vec<ModuleStats>, Vec<Modul
                AND f.path NOT LIKE 'tests/%' \
                AND f.path NOT LIKE '%_test.%' \
              GROUP BY n.id \
-             ORDER BY cnt DESC";
+             ORDER BY cnt DESC \
+             LIMIT 200";
         let mut stmt = conn.prepare(sql)?;
         let rows = stmt.query_map([REL_CALLS], |row| {
             Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))

src/storage/queries.rs

@@ -71,6 +71,25 @@ CREATE INDEX IF NOT EXISTS idx_edges_source_rel ON edges(source_id, relation);
 CREATE INDEX IF NOT EXISTS idx_edges_target_rel ON edges(target_id, relation);
 "#;
 
+/// FTS5 sync trigger SQL. Shared by migrations that recreate the FTS5 table,
+/// ensuring trigger definitions stay consistent with CREATE_TABLES.
+fn fts5_triggers_sql() -> &'static str {
+    "CREATE TRIGGER IF NOT EXISTS nodes_ai AFTER INSERT ON nodes BEGIN
+        INSERT INTO nodes_fts(rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
+        VALUES (new.id, new.name, new.qualified_name, new.code_content, new.context_string, new.doc_comment, new.name_tokens, new.return_type, new.param_types);
+    END;
+    CREATE TRIGGER IF NOT EXISTS nodes_ad AFTER DELETE ON nodes BEGIN
+        INSERT INTO nodes_fts(nodes_fts, rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
+        VALUES ('delete', old.id, old.name, old.qualified_name, old.code_content, old.context_string, old.doc_comment, old.name_tokens, old.return_type, old.param_types);
+    END;
+    CREATE TRIGGER IF NOT EXISTS nodes_au AFTER UPDATE ON nodes BEGIN
+        INSERT INTO nodes_fts(nodes_fts, rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
+        VALUES ('delete', old.id, old.name, old.qualified_name, old.code_content, old.context_string, old.doc_comment, old.name_tokens, old.return_type, old.param_types);
+        INSERT INTO nodes_fts(rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
+        VALUES (new.id, new.name, new.qualified_name, new.code_content, new.context_string, new.doc_comment, new.name_tokens, new.return_type, new.param_types);
+    END;"
+}
+
 /// Migrate from schema v1 to v2. Must be called within a transaction.
 pub fn migrate_v1_to_v2(conn: &rusqlite::Connection) -> anyhow::Result<()> {
     tracing::info!("[schema] Migrating v1 → v2: adding name_tokens, return_type, param_types");
@@ -93,23 +112,9 @@ pub fn migrate_v1_to_v2(conn: &rusqlite::Connection) -> anyhow::Result<()> {
             name, qualified_name, code_content, context_string, doc_comment,
             name_tokens, return_type, param_types,
             content='nodes', content_rowid='id'
-        );
-
-        CREATE TRIGGER IF NOT EXISTS nodes_ai AFTER INSERT ON nodes BEGIN
-            INSERT INTO nodes_fts(rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES (new.id, new.name, new.qualified_name, new.code_content, new.context_string, new.doc_comment, new.name_tokens, new.return_type, new.param_types);
-        END;
-        CREATE TRIGGER IF NOT EXISTS nodes_ad AFTER DELETE ON nodes BEGIN
-            INSERT INTO nodes_fts(nodes_fts, rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES ('delete', old.id, old.name, old.qualified_name, old.code_content, old.context_string, old.doc_comment, old.name_tokens, old.return_type, old.param_types);
-        END;
-        CREATE TRIGGER IF NOT EXISTS nodes_au AFTER UPDATE ON nodes BEGIN
-            INSERT INTO nodes_fts(nodes_fts, rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES ('delete', old.id, old.name, old.qualified_name, old.code_content, old.context_string, old.doc_comment, old.name_tokens, old.return_type, old.param_types);
-            INSERT INTO nodes_fts(rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES (new.id, new.name, new.qualified_name, new.code_content, new.context_string, new.doc_comment, new.name_tokens, new.return_type, new.param_types);
-        END;"
+        );"
     )?;
+    conn.execute_batch(fts5_triggers_sql())?;
 
     conn.execute_batch("INSERT INTO nodes_fts(nodes_fts) VALUES('rebuild');")?;
 
@@ -167,23 +172,9 @@ pub fn migrate_v3_to_v4(conn: &rusqlite::Connection) -> anyhow::Result<()> {
             name_tokens, return_type, param_types,
             content='nodes', content_rowid='id',
             tokenize='porter unicode61'
-        );
-
-        CREATE TRIGGER IF NOT EXISTS nodes_ai AFTER INSERT ON nodes BEGIN
-            INSERT INTO nodes_fts(rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES (new.id, new.name, new.qualified_name, new.code_content, new.context_string, new.doc_comment, new.name_tokens, new.return_type, new.param_types);
-        END;
-        CREATE TRIGGER IF NOT EXISTS nodes_ad AFTER DELETE ON nodes BEGIN
-            INSERT INTO nodes_fts(nodes_fts, rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES ('delete', old.id, old.name, old.qualified_name, old.code_content, old.context_string, old.doc_comment, old.name_tokens, old.return_type, old.param_types);
-        END;
-        CREATE TRIGGER IF NOT EXISTS nodes_au AFTER UPDATE ON nodes BEGIN
-            INSERT INTO nodes_fts(nodes_fts, rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES ('delete', old.id, old.name, old.qualified_name, old.code_content, old.context_string, old.doc_comment, old.name_tokens, old.return_type, old.param_types);
-            INSERT INTO nodes_fts(rowid, name, qualified_name, code_content, context_string, doc_comment, name_tokens, return_type, param_types)
-            VALUES (new.id, new.name, new.qualified_name, new.code_content, new.context_string, new.doc_comment, new.name_tokens, new.return_type, new.param_types);
-        END;"
+        );"
     )?;
+    conn.execute_batch(fts5_triggers_sql())?;
 
     conn.execute_batch("INSERT INTO nodes_fts(nodes_fts) VALUES('rebuild');")?;