fix: shared reader locks, resolved lock paths, install-only sync lock

Third Codex (gpt-6.1-sol, xhigh) pass over 1e55138, each item verified
before fixing:

- readers now hold the ownership lock shared for the life of their
  connection, so a snapshot swap (exclusive) never happens under an
  open SQLite connection whose WAL cleanup could hit a later import.
- the lock file name uses the resolved database path, so a symlinked
  path shares the real database's lock; sync also installs at the
  resolved path, keeping a symlinked path a symlink (it used to be
  overwritten with a regular file).
- sync downloads without the lock and takes it only to recheck and
  install; snapshot downloads have connect (15 s) and total (10 min)
  timeouts. idx-mcp drops its in-process RwLock: idx-core's lock covers
  it across processes.
- XLSX parsing caps rows at 1,000,000 and checks a shared string's size
  before copying it.
- the temp snapshot's -wal/-shm files are removed instead of left in the
  data directory.

Remaining, by design: `changes` loads all rows of two releases from the
local database before filtering; its size is bounded by the snapshot.

CLI goldens unchanged. Live A/B vs 0.2.4: 16 passed; discover hits the
IDX 403 on both builds (exit 4 vs 1, intended).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/idx-core/src/investor.rs

@@ -47,17 +47,25 @@ impl Investor {
         }
     }
 
-    /// Cross-process lock held for every write (sync, import, entity edits),
+    /// Cross-process lock beside the database. Writers (sync install, import,
-    /// so a sync cannot replace the database under a concurrent import, or the
+    /// entity edits) hold it exclusively; readers hold it shared, so a
-    /// other way around. Released when the returned file is dropped.
+    /// snapshot swap never happens under an open connection. Released when
-    fn write_lock(&self) -> Result<fs::File, IdxError> {
+    /// the returned file is dropped.
+    fn lock_file(&self, exclusive: bool) -> Result<fs::File, IdxError> {
         let db_path = self.db_path()?;
-        if let Some(parent) = db_path.parent()
+        let parent = match db_path.parent() {
-            && !parent.as_os_str().is_empty()
+            Some(parent) if !parent.as_os_str().is_empty() => parent.to_path_buf(),
-        {
+            _ => PathBuf::from("."),
-            fs::create_dir_all(parent).map_err(|e| IdxError::Io(e.to_string()))?;
+        };
-        }
+        fs::create_dir_all(&parent).map_err(|e| IdxError::Io(e.to_string()))?;
-        let mut lock_name = db_path.as_os_str().to_os_string();
+        // Resolve symlinks, so every path to one database shares one lock.
+        let real = match fs::canonicalize(&db_path) {
+            Ok(real) => real,
+            Err(_) => fs::canonicalize(&parent)
+                .map_err(|e| IdxError::Io(e.to_string()))?
+                .join(db_path.file_name().unwrap_or_default()),
+        };
+        let mut lock_name = real.into_os_string();
         lock_name.push(".lock");
         let file = fs::OpenOptions::new()
             .create(true)
@@ -65,23 +73,41 @@ impl Investor {
             .write(true)
             .open(PathBuf::from(lock_name))
             .map_err(|e| IdxError::Io(format!("failed to open ownership lock: {e}")))?;
-        file.lock()
+        if exclusive {
-            .map_err(|e| IdxError::Io(format!("failed to lock ownership database: {e}")))?;
+            file.lock()
+        } else {
+            file.lock_shared()
+        }
+        .map_err(|e| IdxError::Io(format!("failed to lock ownership database: {e}")))?;
         Ok(file)
     }
 
-    fn connect(&self) -> Result<Connection, IdxError> {
+    fn write_lock(&self) -> Result<fs::File, IdxError> {
+        self.lock_file(true)
+    }
+
+    /// A read connection holding the shared lock for its lifetime.
+    fn connect(&self) -> Result<ReadConnection, IdxError> {
+        let lock = self.lock_file(false)?;
+        Ok(ReadConnection {
+            conn: self.open_db()?,
+            _lock: lock,
+        })
+    }
+
+    /// For writers, which already hold the exclusive lock.
+    fn open_db(&self) -> Result<Connection, IdxError> {
         db::open_db_at(&self.db_path()?)
     }
 
     pub fn releases(&self) -> Result<Vec<OwnershipRelease>, IdxError> {
-        db::query_releases(&self.connect()?)
+        db::query_releases(&*self.connect()?)
     }
 
     /// Whether the database has data, and which releases it holds.
     pub fn status(&self) -> Result<OwnershipStatus, IdxError> {
         let db_path = self.db_path()?;
-        let releases = db::query_releases(&db::open_db_at(&db_path)?)?;
+        let releases = db::query_releases(&*self.connect()?)?;
         Ok(OwnershipStatus {
             db_path: db_path.display().to_string(),
             populated: !releases.is_empty(),
@@ -139,7 +165,7 @@ impl Investor {
     }
 
     pub fn search(&self, query: &str, limit: usize) -> Result<Vec<Entity>, IdxError> {
-        search::fts_search(&self.connect()?, query, limit)
+        search::fts_search(&*self.connect()?, query, limit)
     }
 
     pub fn cross_holders(
@@ -147,7 +173,7 @@ impl Investor {
         min_tickers: usize,
         top: usize,
     ) -> Result<Vec<CrossHolderRow>, IdxError> {
-        db::query_cross_holders(&self.connect()?, min_tickers, top)
+        db::query_cross_holders(&*self.connect()?, min_tickers, top)
     }
 
     pub fn concentration(
@@ -155,7 +181,7 @@ impl Investor {
         by: ConcentrationRank,
         top: usize,
     ) -> Result<Vec<(String, ConcentrationMetrics)>, IdxError> {
-        db::query_concentration(&self.connect()?, by.as_str(), top)
+        db::query_concentration(&*self.connect()?, by.as_str(), top)
     }
 
     /// Institutional (Bing) flow; `None` when no flow data exists.
@@ -174,7 +200,7 @@ impl Investor {
         to: &str,
         ticker: Option<&str>,
     ) -> Result<Vec<ChangeRow>, IdxError> {
-        let mut rows = db::query_changes(&self.connect()?, from, to)?;
+        let mut rows = db::query_changes(&*self.connect()?, from, to)?;
         if let Some(ticker) = ticker {
             let ticker = ticker.trim().to_uppercase();
             rows.retain(|row| row.ticker_code == ticker);
@@ -187,22 +213,22 @@ impl Investor {
         root: &str,
         depth: usize,
     ) -> Result<(Vec<GraphNode>, Vec<GraphEdge>), IdxError> {
-        graph::query_ownership_graph(&self.connect()?, root, depth)
+        graph::query_ownership_graph(&*self.connect()?, root, depth)
     }
 
     pub fn unresolved(&self, limit: usize) -> Result<Vec<UnresolvedRow>, IdxError> {
-        db::list_unresolved(&self.connect()?, limit)
+        db::list_unresolved(&*self.connect()?, limit)
     }
 
     pub fn map_alias(&self, alias: &str, entity: &str) -> Result<(), IdxError> {
         let _lock = self.write_lock()?;
-        db::manual_map(&self.connect()?, alias, entity)
+        db::manual_map(&self.open_db()?, alias, entity)
     }
 
     /// Merges entity `merge` into `keep`, reporting how many rows moved.
     pub fn merge_entities(&self, keep: i64, merge: i64) -> Result<MergeResult, IdxError> {
         let _lock = self.write_lock()?;
-        let conn = self.connect()?;
+        let conn = self.open_db()?;
         let count = |table: &str| -> Result<i64, IdxError> {
             conn.query_row(
                 &format!("SELECT COUNT(*) FROM {table} WHERE entity_id = ?1"),
@@ -242,8 +268,7 @@ impl Investor {
     ) -> Result<OwnershipSyncResult, IdxError> {
         let manifest_source = snapshot::resolve_manifest_source(manifest)?;
         let db_path = self.db_path()?;
-        let _lock = self.write_lock()?;
+        snapshot::sync_snapshot(&manifest_source, &db_path, force, || self.write_lock())
-        snapshot::sync_snapshot(&manifest_source, &db_path, force)
     }
 
     /// Imports one report. A release already imported (same SHA-256) is
@@ -253,7 +278,7 @@ impl Investor {
         // only around the database writes.
         let input = self.resolve_import_input(source)?;
         let _lock = self.write_lock()?;
-        let conn = self.connect()?;
+        let conn = self.open_db()?;
 
         let sha256 = sha256_file(&input.path)?;
         if !force && db::release_exists(&conn, &sha256)? {
@@ -441,6 +466,19 @@ pub struct OwnershipStatus {
     pub releases: Vec<OwnershipRelease>,
 }
 
+struct ReadConnection {
+    conn: Connection,
+    _lock: fs::File,
+}
+
+impl std::ops::Deref for ReadConnection {
+    type Target = Connection;
+
+    fn deref(&self) -> &Connection {
+        &self.conn
+    }
+}
+
 /// Holder sources for `Investor::ticker`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum HolderSource {

crates/idx-core/src/ownership/snapshot.rs

@@ -144,11 +144,20 @@ pub fn parse_manifest(raw: &str) -> Result<OwnershipSnapshotManifest, IdxError>
     Ok(manifest)
 }
 
-pub fn sync_snapshot(
+/// `lock` takes the database's exclusive writer lock; it is held only while
+/// rechecking and installing, never during the downloads.
+pub fn sync_snapshot<L>(
     manifest_source: &str,
     db_path: &Path,
     force: bool,
+    lock: impl FnOnce() -> Result<L, IdxError>,
 ) -> Result<OwnershipSyncResult, IdxError> {
+    // Results report the configured path; files are read and replaced at the
+    // resolved one, so a symlinked path updates its target instead of being
+    // overwritten with a regular file.
+    let configured_db_path = db_path;
+    let resolved_db_path = fs::canonicalize(db_path).unwrap_or_else(|_| db_path.to_path_buf());
+    let db_path = resolved_db_path.as_path();
     let manifest = fetch_manifest(manifest_source)?;
     let local_state = inspect_local_db(db_path)?;
     let decision = build_sync_decision(local_state.as_ref(), &manifest.snapshot, force);
@@ -157,7 +166,7 @@ pub fn sync_snapshot(
         return Ok(OwnershipSyncResult {
             action: decision.action,
             manifest: manifest_source.to_string(),
-            db_path: db_path.display().to_string(),
+            db_path: configured_db_path.display().to_string(),
             snapshot_version: manifest.snapshot.version.clone(),
             latest_as_of_date: manifest.snapshot.latest_as_of_date,
             release_count: manifest.snapshot.release_count,
@@ -202,6 +211,7 @@ pub fn sync_snapshot(
 
     let install_result = (|| -> Result<(), IdxError> {
         validate_snapshot_db(&temp_path, &manifest.snapshot)?;
+        let _lock = lock()?;
         // The decision was made before the download; an import or edit that
         // landed since must not be overwritten without --force.
         if !force && !same_local_state(local_state.as_ref(), inspect_local_db(db_path)?.as_ref()) {
@@ -215,13 +225,14 @@ pub fn sync_snapshot(
 
     if install_result.is_err() {
         let _ = fs::remove_file(&temp_path);
+        let _ = remove_sqlite_sidecars(&temp_path);
     }
     install_result?;
 
     Ok(OwnershipSyncResult {
         action: decision.action,
         manifest: manifest_source.to_string(),
-        db_path: db_path.display().to_string(),
+        db_path: configured_db_path.display().to_string(),
         snapshot_version: manifest.snapshot.version.clone(),
         latest_as_of_date: manifest.snapshot.latest_as_of_date,
         release_count: manifest.snapshot.release_count,
@@ -639,6 +650,8 @@ fn build_temp_path(db_path: &Path) -> PathBuf {
 }
 
 fn install_snapshot_file(temp_path: &Path, db_path: &Path) -> Result<(), IdxError> {
+    // Validation opened the temp database; its sidecars are named after it.
+    remove_sqlite_sidecars(temp_path)?;
     if let Some(parent) = db_path.parent()
         && !parent.as_os_str().is_empty()
     {
@@ -748,7 +761,8 @@ fn sha256_hex(bytes: &[u8]) -> String {
 
 fn read_text(source: &str, context: &str) -> Result<String, IdxError> {
     if is_http_source(source) {
-        let response = ureq::get(source)
+        let response = http_agent()
+            .get(source)
             .header("User-Agent", USER_AGENT)
             .header("Accept", "application/json,text/plain;q=0.9,*/*;q=0.8")
             .call()
@@ -764,9 +778,20 @@ fn read_text(source: &str, context: &str) -> Result<String, IdxError> {
         .map_err(|e| IdxError::Io(format!("failed to read {context} {}: {e}", path.display())))
 }
 
+/// Downloads give up rather than hang: a stalled connection must not keep
+/// sync (and anything waiting on it) blocked forever.
+fn http_agent() -> ureq::Agent {
+    ureq::Agent::config_builder()
+        .timeout_connect(Some(std::time::Duration::from_secs(15)))
+        .timeout_global(Some(std::time::Duration::from_secs(600)))
+        .build()
+        .into()
+}
+
 fn read_bytes(source: &str, context: &str, limit_bytes: u64) -> Result<Vec<u8>, IdxError> {
     if is_http_source(source) {
-        let response = ureq::get(source)
+        let response = http_agent()
+            .get(source)
             .header("User-Agent", USER_AGENT)
             .header(
                 "Accept",

crates/idx-core/src/ownership/xlsx.rs

@@ -172,8 +172,27 @@ fn row_number(element: &BytesStart<'_>, fallback: usize) -> Result<usize, IdxErr
     Ok(fallback)
 }
 
-/// Total text across all cells of one worksheet; real registers hold a few MB.
+/// Bounds on one worksheet. Real above-1% registers hold about 7,000 rows
+/// and a few MB of text.
 const MAX_CELL_TEXT_BYTES: usize = 256 * 1024 * 1024;
+const MAX_SHEET_ROWS: usize = 1_000_000;
+
+fn text_cap_error() -> IdxError {
+    xml_error(
+        "worksheet",
+        format!("cell text exceeds {MAX_CELL_TEXT_BYTES} bytes"),
+    )
+}
+
+fn check_row_count(rows: usize) -> Result<(), IdxError> {
+    if rows >= MAX_SHEET_ROWS {
+        return Err(xml_error(
+            "worksheet",
+            format!("more than {MAX_SHEET_ROWS} rows"),
+        ));
+    }
+    Ok(())
+}
 
 fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result<Vec<SheetRow>, IdxError> {
     let mut reader = Reader::from_str(xml);
@@ -188,6 +207,7 @@ fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result<Vec<SheetRow
         match reader.read_event() {
             Ok(Event::Start(e)) => match e.local_name().as_ref() {
                 b"row" => {
+                    check_row_count(rows.len())?;
                     row = Some(SheetRow {
                         number: row_number(&e, rows.len() + 1)?,
                         cells: HashMap::new(),
@@ -200,10 +220,13 @@ fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result<Vec<SheetRow
                 b"v" | b"t" if cell.is_some() => in_value = true,
                 _ => {}
             },
-            Ok(Event::Empty(e)) if e.local_name().as_ref() == b"row" => rows.push(SheetRow {
+            Ok(Event::Empty(e)) if e.local_name().as_ref() == b"row" => {
-                number: row_number(&e, rows.len() + 1)?,
+                check_row_count(rows.len())?;
-                cells: HashMap::new(),
+                rows.push(SheetRow {
-            }),
+                    number: row_number(&e, rows.len() + 1)?,
+                    cells: HashMap::new(),
+                })
+            }
             Ok(Event::Text(text)) if in_value => {
                 let unescaped = text.unescape().map_err(|e| xml_error("worksheet", e))?;
                 value.push_str(&unescaped);
@@ -219,12 +242,18 @@ fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result<Vec<SheetRow
                                 let index = value.trim().parse::<usize>().map_err(|e| {
                                     xml_error("worksheet", format!("bad shared string index: {e}"))
                                 })?;
-                                shared_strings.get(index).cloned().ok_or_else(|| {
+                                let shared = shared_strings.get(index).ok_or_else(|| {
                                     xml_error(
                                         "worksheet",
                                         format!("shared string {index} out of range"),
                                     )
-                                })?
+                                })?;
+                                // Check before copying: one large string may be
+                                // referenced from many cells.
+                                if cell_text_bytes + shared.len() > MAX_CELL_TEXT_BYTES {
+                                    return Err(text_cap_error());
+                                }
+                                shared.clone()
                             }
                             // Excel stores the share code TRUE (TRINITI DINAMIK)
                             // as a boolean cell holding 1.
@@ -235,14 +264,9 @@ fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result<Vec<SheetRow
                             },
                             _ => value.clone(),
                         };
-                        // Shared strings are bounded per entry, but one large
-                        // string referenced from many cells is copied each time.
                         cell_text_bytes += text.len();
                         if cell_text_bytes > MAX_CELL_TEXT_BYTES {
-                            return Err(xml_error(
+                            return Err(text_cap_error());
-                                "worksheet",
-                                format!("cell text exceeds {MAX_CELL_TEXT_BYTES} bytes"),
-                            ));
                         }
                         current_row.cells.insert(column, text);
                     }

crates/idx-mcp/src/server.rs

@@ -15,7 +15,7 @@ use rmcp::model::{CallToolResult, Implementation, ServerCapabilities, ServerConf
 use rmcp::{ErrorData, ServerHandler, tool, tool_handler, tool_router};
 use serde::Serialize;
 use serde_json::{Map, Value, json};
-use tokio::sync::{RwLock, Semaphore};
+use tokio::sync::Semaphore;
 
 use crate::params::{
     AnalysisKind, AnalyzeParams, ChangesParams, CompanyParams, CompanySection, EntityParams,
@@ -36,9 +36,6 @@ const MAX_CHANGES_LIMIT: usize = 2000;
 
 /// Tool calls running at once; the rest wait their turn.
 static CALL_SLOTS: Semaphore = Semaphore::const_new(4);
-/// `ownership_sync` replaces the database file: it runs alone, while
-/// ownership reads may run together.
-static OWNERSHIP: RwLock<()> = RwLock::const_new(());
 
 const INSTRUCTIONS: &str = "Indonesian stock market (IDX) data. Market and financial \
 tools fetch live data (cached on disk) and wrap it as {data, source, fetched_at, stale}; \
@@ -92,7 +89,6 @@ impl IdxServer {
         } else {
             "run `idx ownership sync`, or restart idx-mcp with --allow-sync"
         };
-        let _read = OWNERSHIP.read().await;
         blocking(move || {
             if !investor.status()?.populated {
                 return Err(IdxError::NoMarketData(format!(
@@ -323,7 +319,6 @@ KSEI release dates."
     )]
     async fn ownership_status(&self) -> ToolResult {
         let investor = Investor::new(&self.config);
-        let _read = OWNERSHIP.read().await;
         blocking(move || investor.status()).await
     }
 
@@ -334,7 +329,7 @@ the local database."
     async fn ownership_sync(&self, params: Parameters<SyncParams>) -> ToolResult {
         let SyncParams { force } = params.0;
         let investor = Investor::new(&self.config);
-        let _write = OWNERSHIP.write().await;
+        // idx-core's database lock keeps the swap away from open readers.
         blocking(move || investor.sync(None, force)).await
     }
 }