0xrsydn/idx-cli
1
mirror of https://github.com/0xrsydn/idx-cli.git synced 2026-11-07 01:35:36 +00:00
Code Issues Projects Releases 1 Packages Wiki Activity

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

Browse source 
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>
This commit is contained in:
hermes 2026-10-14 22:39:09 +00:00
commit 5e0831d8fe
4 changed files with 133 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

90
crates/idx-core/src/investor.rs
View file

@ -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<fs::File, IdxError> {
/// 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<fs::File, IdxError> {
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<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)
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<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 {

35
crates/idx-core/src/ownership/snapshot.rs
View file

@ -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",

50
crates/idx-core/src/ownership/xlsx.rs
View file

@ -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 {
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<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(
"worksheet",
format!("cell text exceeds {MAX_CELL_TEXT_BYTES} bytes"),
));
return Err(text_cap_error());
}
current_row.cells.insert(column, text);
}

9
crates/idx-mcp/src/server.rs
View file

@ -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
}
}