diff --git a/crates/idx-core/src/investor.rs b/crates/idx-core/src/investor.rs index 56492b0..001dc5c 100644 --- a/crates/idx-core/src/investor.rs +++ b/crates/idx-core/src/investor.rs @@ -47,17 +47,25 @@ impl Investor { } } - /// Cross-process lock held for every write (sync, import, entity edits), - /// so a sync cannot replace the database under a concurrent import, or the - /// other way around. Released when the returned file is dropped. - fn write_lock(&self) -> Result { + /// Cross-process lock beside the database. Writers (sync install, import, + /// entity edits) hold it exclusively; readers hold it shared, so a + /// snapshot swap never happens under an open connection. Released when + /// the returned file is dropped. + fn lock_file(&self, exclusive: bool) -> Result { let db_path = self.db_path()?; - if let Some(parent) = db_path.parent() - && !parent.as_os_str().is_empty() - { - 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(); + let parent = match db_path.parent() { + 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()))?; + // 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() - .map_err(|e| IdxError::Io(format!("failed to lock ownership database: {e}")))?; + if exclusive { + file.lock() + } else { + file.lock_shared() + } + .map_err(|e| IdxError::Io(format!("failed to lock ownership database: {e}")))?; Ok(file) } - fn connect(&self) -> Result { + fn write_lock(&self) -> Result { + self.lock_file(true) + } + + /// A read connection holding the shared lock for its lifetime. + fn connect(&self) -> Result { + 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 { db::open_db_at(&self.db_path()?) } pub fn releases(&self) -> Result, 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 { 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, 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, 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, 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, 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, Vec), IdxError> { - graph::query_ownership_graph(&self.connect()?, root, depth) + graph::query_ownership_graph(&*self.connect()?, root, depth) } pub fn unresolved(&self, limit: usize) -> Result, 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 { let _lock = self.write_lock()?; - let conn = self.connect()?; + let conn = self.open_db()?; let count = |table: &str| -> Result { conn.query_row( &format!("SELECT COUNT(*) FROM {table} WHERE entity_id = ?1"), @@ -242,8 +268,7 @@ impl Investor { ) -> Result { 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) + snapshot::sync_snapshot(&manifest_source, &db_path, force, || self.write_lock()) } /// 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, } +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 { diff --git a/crates/idx-core/src/ownership/snapshot.rs b/crates/idx-core/src/ownership/snapshot.rs index 26c15c3..9768c53 100644 --- a/crates/idx-core/src/ownership/snapshot.rs +++ b/crates/idx-core/src/ownership/snapshot.rs @@ -144,11 +144,20 @@ pub fn parse_manifest(raw: &str) -> Result 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( manifest_source: &str, db_path: &Path, force: bool, + lock: impl FnOnce() -> Result, ) -> Result { + // 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 { 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 { .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, IdxError> { if is_http_source(source) { - let response = ureq::get(source) + let response = http_agent() + .get(source) .header("User-Agent", USER_AGENT) .header( "Accept", diff --git a/crates/idx-core/src/ownership/xlsx.rs b/crates/idx-core/src/ownership/xlsx.rs index 3488c4b..40e0a37 100644 --- a/crates/idx-core/src/ownership/xlsx.rs +++ b/crates/idx-core/src/ownership/xlsx.rs @@ -172,8 +172,27 @@ fn row_number(element: &BytesStart<'_>, fallback: usize) -> Result 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, IdxError> { let mut reader = Reader::from_str(xml); @@ -188,6 +207,7 @@ fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result 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 in_value = true, _ => {} }, - Ok(Event::Empty(e)) if e.local_name().as_ref() == b"row" => rows.push(SheetRow { - number: row_number(&e, rows.len() + 1)?, - cells: HashMap::new(), - }), + Ok(Event::Empty(e)) if e.local_name().as_ref() == b"row" => { + check_row_count(rows.len())?; + 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().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 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( - "worksheet", - format!("cell text exceeds {MAX_CELL_TEXT_BYTES} bytes"), - )); + return Err(text_cap_error()); } current_row.cells.insert(column, text); } diff --git a/crates/idx-mcp/src/server.rs b/crates/idx-mcp/src/server.rs index c68e0a8..eba79fa 100644 --- a/crates/idx-mcp/src/server.rs +++ b/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) -> 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 } }