diff --git a/Cargo.toml b/Cargo.toml index 5b2edf9..f91aa1b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ members = ["crates/idx-core", "crates/idx-cli", "crates/idx-mcp"] [workspace.package] version = "0.3.0" edition = "2024" -rust-version = "1.85" +rust-version = "1.89" license = "MIT" repository = "https://github.com/0xrsydn/idx-cli" homepage = "https://github.com/0xrsydn/idx-cli" diff --git a/README.md b/README.md index 9b9a5c1..ab10a92 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ helper tool below is available automatically. cargo install idx-cli ``` -Builds from source; requires Rust `1.85+`. +Builds from source; requires Rust `1.89+`. ### Runtime helper tools @@ -172,7 +172,7 @@ client: ```bash claude mcp add idx -- npx -y idx-mcp # Claude Code, via npm -cargo install idx-mcp # or build it (Rust 1.88+) +cargo install idx-mcp # or build it (Rust 1.89+) nix profile install github:0xrsydn/idx-cli#idx-mcp ``` diff --git a/crates/idx-core/src/cache.rs b/crates/idx-core/src/cache.rs index ad74752..5e1abb5 100644 --- a/crates/idx-core/src/cache.rs +++ b/crates/idx-core/src/cache.rs @@ -233,7 +233,8 @@ impl Cache { runtime::warn(format!( "corrupted cache entry for {data_type}/{symbol}, treating as miss: {e}" )); - let _ = fs::remove_file(&path); + // Not deleted: a concurrent writer may have just replaced it + // with a valid entry. The next put overwrites it atomically. return Ok(None); } }; @@ -244,7 +245,6 @@ impl Cache { entry.schema_version, CURRENT_SCHEMA_VERSION )); - let _ = fs::remove_file(&path); return Ok(None); } Ok(Some(entry)) @@ -352,7 +352,7 @@ mod tests { } #[test] - fn corrupted_cache_entry_returns_none_and_deletes_file() { + fn corrupted_cache_entry_is_a_miss_until_replaced() { let root = tmp(); let cache = Cache::at(root.clone()); @@ -371,12 +371,16 @@ mod tests { .expect("get should not error"); assert_eq!(result, None, "corrupted entry should be treated as miss"); - // The corrupted file should be deleted - assert!(!path.exists(), "corrupted file should be deleted"); + // Left for the next put to replace (deleting could race a writer). + cache + .put("quote", "CORRUPT.JK", &T { v: 7 }, 3600) + .expect("put"); + let result: Option = cache.get("quote", "CORRUPT.JK").expect("get"); + assert_eq!(result, Some(T { v: 7 })); } #[test] - fn corrupted_cache_entry_get_stale_returns_none_and_deletes_file() { + fn corrupted_cache_entry_get_stale_is_a_miss() { let root = tmp(); let cache = Cache::at(root.clone()); @@ -401,8 +405,7 @@ mod tests { "corrupted entry should be treated as miss in get_stale" ); - // The corrupted file should be deleted - assert!(!path.exists(), "corrupted file should be deleted"); + assert!(path.exists(), "left for the next put to replace"); } #[test] diff --git a/crates/idx-core/src/investor.rs b/crates/idx-core/src/investor.rs index c8285dc..56492b0 100644 --- a/crates/idx-core/src/investor.rs +++ b/crates/idx-core/src/investor.rs @@ -47,6 +47,29 @@ 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 { + 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(); + lock_name.push(".lock"); + let file = fs::OpenOptions::new() + .create(true) + .truncate(false) + .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}")))?; + Ok(file) + } + fn connect(&self) -> Result { db::open_db_at(&self.db_path()?) } @@ -172,11 +195,13 @@ impl Investor { } pub fn map_alias(&self, alias: &str, entity: &str) -> Result<(), IdxError> { + let _lock = self.write_lock()?; db::manual_map(&self.connect()?, 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 count = |table: &str| -> Result { conn.query_row( @@ -217,13 +242,17 @@ 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) } /// Imports one report. A release already imported (same SHA-256) is /// skipped unless `force`, which replaces it. pub fn import(&self, source: &ImportSource, force: bool) -> Result { + // Validate and download first (unchanged error order), then lock + // only around the database writes. let input = self.resolve_import_input(source)?; + let _lock = self.write_lock()?; let conn = self.connect()?; let sha256 = sha256_file(&input.path)?; diff --git a/crates/idx-core/src/ownership/db.rs b/crates/idx-core/src/ownership/db.rs index f298383..3646fee 100644 --- a/crates/idx-core/src/ownership/db.rs +++ b/crates/idx-core/src/ownership/db.rs @@ -1196,7 +1196,10 @@ pub fn compute_concentration(percentages_bps: &[i64]) -> ConcentrationMetrics { let top1_bps = values.first().copied().unwrap_or(0); let top3_bps: i64 = values.iter().take(3).sum(); let total_bps: i64 = values.iter().sum(); - let hhi: i64 = values.iter().map(|p| (p * p) / 10000).sum(); + let hhi: i64 = values + .iter() + .map(|p| p.saturating_mul(*p) / 10000) + .fold(0, i64::saturating_add); ConcentrationMetrics { top1_bps, diff --git a/crates/idx-core/src/ownership/entities.rs b/crates/idx-core/src/ownership/entities.rs index 4293eab..f2a9873 100644 --- a/crates/idx-core/src/ownership/entities.rs +++ b/crates/idx-core/src/ownership/entities.rs @@ -84,6 +84,7 @@ pub fn parse_id_percentage(s: &str) -> Result { whole_i .checked_mul(100) .and_then(|bps| bps.checked_add(frac_i)) + .filter(|bps| (0..=10_000).contains(bps)) .ok_or_else(|| IdxError::ParseError(format!("Indonesian percentage '{s}' is out of range"))) } @@ -460,6 +461,8 @@ mod tests { #[test] fn huge_percentages_are_parse_errors_not_overflow() { assert!(super::parse_id_percentage("9223372036854775807,99").is_err()); + assert!(super::parse_id_percentage("40000000,00").is_err()); + assert_eq!(super::parse_id_percentage("100,00").unwrap(), 10_000); assert_eq!(super::parse_id_percentage("54,94").unwrap(), 5494); } } diff --git a/crates/idx-core/src/ownership/snapshot.rs b/crates/idx-core/src/ownership/snapshot.rs index 6fe1a00..26c15c3 100644 --- a/crates/idx-core/src/ownership/snapshot.rs +++ b/crates/idx-core/src/ownership/snapshot.rs @@ -348,8 +348,18 @@ fn inspect_local_db(db_path: &Path) -> Result, IdxErr return Ok(None); } + // Schema setup first: it may write to an empty database, and hashing + // afterwards keeps two inspections of an unchanged database equal. A + // published snapshot already has the schema, so its bytes are untouched. + let releases = { + let conn = Connection::open(db_path).map_err(|e| { + IdxError::DatabaseError(format!("failed to open {}: {e}", db_path.display())) + })?; + db::ensure_schema(&conn)?; + db::query_releases(&conn)? + }; + let file_sha256 = if checkpoint_wal(db_path) { - // Hash before opening: schema setup may write to an empty database. let bytes = fs::read(db_path) .map_err(|e| IdxError::Io(format!("failed to read {}: {e}", db_path.display())))?; Some(sha256_hex(&bytes)) @@ -357,12 +367,6 @@ fn inspect_local_db(db_path: &Path) -> Result, IdxErr None }; - let conn = Connection::open(db_path).map_err(|e| { - IdxError::DatabaseError(format!("failed to open {}: {e}", db_path.display())) - })?; - db::ensure_schema(&conn)?; - let releases = db::query_releases(&conn)?; - Ok(Some(LocalSnapshotState { file_sha256, latest_release: releases.first().cloned(), @@ -487,7 +491,9 @@ fn build_sync_decision( } if latest.sha256 == snapshot.latest_release_sha256 { - if latest.row_count != snapshot.latest_row_count { + if latest.row_count != snapshot.latest_row_count + && snapshot.release_count >= local.release_count + { return SyncDecision { action: OwnershipSyncAction::Updated, should_download: true, @@ -1099,6 +1105,23 @@ mod tests { assert_eq!(release_count, 1); } + #[test] + fn inspecting_an_empty_database_twice_sees_the_same_state() { + let dir = std::env::temp_dir().join(format!( + "idx-snapshot-empty-{}-{}", + std::process::id(), + fastrand::u64(..) + )); + std::fs::create_dir_all(&dir).unwrap(); + let db_path = dir.join("ownership.db"); + std::fs::write(&db_path, b"").unwrap(); + + let first = super::inspect_local_db(&db_path).unwrap(); + let second = super::inspect_local_db(&db_path).unwrap(); + assert!(super::same_local_state(first.as_ref(), second.as_ref())); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn local_hash_covers_committed_changes_still_in_the_wal() { let dir = std::env::temp_dir().join(format!( diff --git a/crates/idx-core/src/ownership/xlsx.rs b/crates/idx-core/src/ownership/xlsx.rs index c150cfd..3488c4b 100644 --- a/crates/idx-core/src/ownership/xlsx.rs +++ b/crates/idx-core/src/ownership/xlsx.rs @@ -163,15 +163,18 @@ struct SheetRow { fn row_number(element: &BytesStart<'_>, fallback: usize) -> Result { for attribute in element.attributes() { let attribute = attribute.map_err(|e| xml_error("worksheet", e))?; - if attribute.key.local_name().as_ref() == b"r" { - if let Ok(number) = String::from_utf8_lossy(&attribute.value).trim().parse() { - return Ok(number); - } + if attribute.key.local_name().as_ref() == b"r" + && let Ok(number) = String::from_utf8_lossy(&attribute.value).trim().parse() + { + return Ok(number); } } Ok(fallback) } +/// Total text across all cells of one worksheet; real registers hold a few MB. +const MAX_CELL_TEXT_BYTES: usize = 256 * 1024 * 1024; + fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result, IdxError> { let mut reader = Reader::from_str(xml); let mut rows = Vec::new(); @@ -179,6 +182,7 @@ fn parse_sheet_rows(xml: &str, shared_strings: &[String]) -> Result = None; let mut value = String::new(); let mut in_value = false; + let mut cell_text_bytes = 0usize; loop { match reader.read_event() { @@ -231,6 +235,15 @@ 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"), + )); + } current_row.cells.insert(column, text); } } diff --git a/crates/idx-mcp/Cargo.toml b/crates/idx-mcp/Cargo.toml index 60285c3..4ad4c0d 100644 --- a/crates/idx-mcp/Cargo.toml +++ b/crates/idx-mcp/Cargo.toml @@ -3,8 +3,7 @@ name = "idx-mcp" description = "MCP server for Indonesian stock market (IDX) market, financial, and ownership data" version.workspace = true edition.workspace = true -# rmcp requires 1.88. -rust-version = "1.88" +rust-version.workspace = true license.workspace = true repository.workspace = true homepage.workspace = true diff --git a/crates/idx-mcp/src/server.rs b/crates/idx-mcp/src/server.rs index a6fe864..c68e0a8 100644 --- a/crates/idx-mcp/src/server.rs +++ b/crates/idx-mcp/src/server.rs @@ -171,7 +171,8 @@ earnings, analyst sentiment, insights, and news. Sections that fail are listed u )] async fn get_company(&self, params: Parameters) -> ToolResult { let CompanyParams { symbol, sections } = params.0; - let sections = sections.unwrap_or_else(|| CompanySection::ALL.to_vec()); + let mut sections = sections.unwrap_or_else(|| CompanySection::ALL.to_vec()); + dedup_in_order(&mut sections); if let Some(err) = empty_input("sections", §ions) { return err; } @@ -214,7 +215,8 @@ are listed under `errors`; the call fails only if every analysis fails." )] async fn analyze_stock(&self, params: Parameters) -> ToolResult { let AnalyzeParams { symbol, kinds } = params.0; - let kinds = kinds.unwrap_or_else(|| AnalysisKind::ALL.to_vec()); + let mut kinds = kinds.unwrap_or_else(|| AnalysisKind::ALL.to_vec()); + dedup_in_order(&mut kinds); if let Some(err) = empty_input("kinds", &kinds) { return err; } @@ -306,14 +308,10 @@ release dates, optionally for one ticker." limit, } = params.0; self.local(move |investor| { + let limit = bounded_limit("limit", limit, CHANGES_LIMIT, MAX_CHANGES_LIMIT)?; let mut changes = investor.changes(&from, &to, ticker.as_deref())?; let total = changes.len(); - changes.truncate(bounded_limit( - "limit", - limit, - CHANGES_LIMIT, - MAX_CHANGES_LIMIT, - )?); + changes.truncate(limit); Ok(json!({ "from": from, "to": to, "total": total, "changes": changes })) }) .await @@ -449,6 +447,18 @@ fn checked_symbols(symbols: Vec) -> Result, ToolResult> { Ok(symbols) } +/// Each section or analysis runs once, however often it is requested. +fn dedup_in_order(values: &mut Vec) { + let mut seen = Vec::with_capacity(values.len()); + values.retain(|value| { + let new = !seen.contains(value); + if new { + seen.push(*value); + } + new + }); +} + fn bounded_limit( field: &str, value: Option,