fix: close the remaining review findings (cross-process lock, partials)

Second Codex (gpt-6.1-sol, xhigh) pass over 60634d3, each item verified
before fixing:

- new: the sync recheck hashed an existing empty database before schema
  setup, so the first sync falsely reported an external change. Hash
  after setup; a published snapshot already has the schema and keeps
  its bytes.
- sync, import, and entity edits now hold a cross-process lock
  (<db>.lock, std File::lock), so a sync cannot replace the database
  under an import running in another process. Import validates and
  downloads before taking the lock, keeping its error order. MSRV
  1.85 -> 1.89 for File::lock.
- the row-count rebuild branch also never downgrades history.
- cache reads no longer delete corrupt entries (a concurrent writer may
  have just replaced them); the next put overwrites atomically.
- idx-mcp validates the changes limit before querying and runs each
  requested section or analysis once.
- XLSX parsing bounds total cell text, so one large shared string
  referenced from many cells cannot exhaust memory.
- PDF/archive percentages must be 0-100%; HHI uses saturating math.

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/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<T> = 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]

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<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();
+        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<Connection, IdxError> {
         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<MergeResult, IdxError> {
+        let _lock = self.write_lock()?;
         let conn = self.connect()?;
         let count = |table: &str| -> Result<i64, IdxError> {
             conn.query_row(
@@ -217,13 +242,17 @@ 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)
     }
 
     /// 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<ImportResult, IdxError> {
+        // 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)?;

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,

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

@@ -84,6 +84,7 @@ pub fn parse_id_percentage(s: &str) -> Result<i64, IdxError> {
     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);
     }
 }

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

@@ -348,8 +348,18 @@ fn inspect_local_db(db_path: &Path) -> Result<Option<LocalSnapshotState>, 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<Option<LocalSnapshotState>, 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!(

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

@@ -163,15 +163,18 @@ struct SheetRow {
 fn row_number(element: &BytesStart<'_>, fallback: usize) -> Result<usize, IdxError> {
     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<Vec<SheetRow>, 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<Vec<SheetRow
     let mut cell: Option<(String, String)> = 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<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"),
+                            ));
+                        }
                         current_row.cells.insert(column, text);
                     }
                 }