fix(sync): atomic install, stored-row validation, no local artifacts from remote manifests

From the full-repo Codex audit:
- Install moved the database to a backup, then renamed the snapshot in;
  a crash between the two left no database, and the next sync deleted the
  backup first. Replace it with a single rename (old file or new, never
  neither), and restore a backup an older version left behind.
- Validation trusted the release's row_count metadata, so a snapshot with
  an empty or missing holdings table passed and replaced good data. Also
  count the rows stored for the latest release.
- A manifest fetched over HTTP(S) could name a local download_url such as
  /dev/zero, read without a size limit. Remote manifests must give an
  HTTP(S) artifact; local reads take regular files only, within the size
  limit, and manifests are capped at 1 MiB.
- A same-date build with a different row count replaced the local file
  even when it did not list it in replaces_sqlite_sha256, so a replayed
  older build could undo a corrected one. Same-date rebuilds now go only
  through replaces_sqlite_sha256, which the publisher fills.

Verified against the published snapshot: fresh install, no-op re-sync,
and forced reinstall all succeed with no leftover files.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V3syn2fN2LNo5pvTSqc6eh

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

@@ -189,6 +189,7 @@ pub fn sync_snapshot<L>(
     // A body larger than the manifest says is rejected while downloading
     // (ureq fails once a body reaches its limit, hence the +1); the exact size
     // and SHA-256 are verified afterwards.
+    check_download_source(manifest_source, &manifest.snapshot.download_url)?;
     let bytes = read_bytes(
         &manifest.snapshot.download_url,
         "ownership snapshot SQLite",
@@ -506,17 +507,12 @@ fn build_sync_decision(
         };
     }
 
-    if latest.sha256 == snapshot.latest_release_sha256 {
+    // A same-date rebuild replaces the local file only through
-        if latest.row_count != snapshot.latest_row_count
+    // `replaces_sqlite_sha256` above; differing row counts alone would let a
-            && snapshot.release_count >= local.release_count
+    // replayed older build undo a corrected one.
+    if latest.sha256 == snapshot.latest_release_sha256
+        && latest.row_count == snapshot.latest_row_count
     {
-            return SyncDecision {
-                action: OwnershipSyncAction::Updated,
-                should_download: true,
-                reason: "local ownership database has the same latest release date but different row_count metadata".to_string(),
-            };
-        }
-
         if local.release_count < snapshot.release_count {
             return SyncDecision {
                 action: OwnershipSyncAction::Updated,
@@ -551,6 +547,18 @@ fn build_sync_decision(
     }
 }
 
+/// A manifest fetched over HTTP(S) must not name a local file: it could make
+/// sync read a device such as /dev/zero, or any local file. A local manifest
+/// (the user's own) may.
+fn check_download_source(manifest_source: &str, download_url: &str) -> Result<(), IdxError> {
+    if is_http_source(manifest_source) && !is_http_source(download_url) {
+        return Err(IdxError::InvalidInput(format!(
+            "ownership snapshot manifest from {manifest_source} names a non-HTTP(S) download_url: {download_url}"
+        )));
+    }
+    Ok(())
+}
+
 fn validate_downloaded_bytes(
     bytes: &[u8],
     snapshot: &OwnershipSnapshotArtifact,
@@ -626,6 +634,21 @@ fn validate_snapshot_db(path: &Path, snapshot: &OwnershipSnapshotArtifact) -> Re
             snapshot.latest_row_count, latest.row_count
         )));
     }
+    // Count the stored rows too: release metadata alone would accept a
+    // snapshot whose holdings table is empty or missing.
+    let stored_rows: i64 = conn
+        .query_row(
+            "SELECT COUNT(*) FROM ksei_holdings WHERE release_sha256 = ?1",
+            rusqlite::params![latest.sha256],
+            |row| row.get(0),
+        )
+        .map_err(|e| IdxError::DatabaseError(format!("failed counting snapshot holdings: {e}")))?;
+    if usize::try_from(stored_rows).ok() != Some(snapshot.latest_row_count) {
+        return Err(IdxError::ParseError(format!(
+            "downloaded snapshot holds {stored_rows} rows for its latest release; manifest expected {}",
+            snapshot.latest_row_count
+        )));
+    }
 
     let ticker_count: i64 = conn
         .query_row("SELECT COUNT(*) FROM tickers", [], |row| row.get(0))
@@ -668,47 +691,34 @@ fn install_snapshot_file(temp_path: &Path, db_path: &Path) -> Result<(), IdxErro
         })?;
     }
 
+    // Versions before 0.3.3 moved the database to a backup before installing;
+    // a crash between the two renames left only the backup. Restore it, or
+    // drop it when the database is there.
     let backup_path = build_backup_path(db_path);
     if backup_path.exists() {
-        fs::remove_file(&backup_path).map_err(|e| {
+        let result = if db_path.exists() {
+            fs::remove_file(&backup_path)
+        } else {
+            fs::rename(&backup_path, db_path)
+        };
+        result.map_err(|e| {
             IdxError::Io(format!(
-                "failed to remove stale backup {}: {e}",
+                "failed to resolve leftover backup {}: {e}",
                 backup_path.display()
             ))
         })?;
     }
 
-    let had_existing = db_path.exists();
+    // One rename replaces the old database atomically: a crash leaves either
+    // the old file or the new one, never neither. Queries do not write to a
+    // synced file, so its sidecars hold no data and go first.
     remove_sqlite_sidecars(db_path)?;
-
+    fs::rename(temp_path, db_path).map_err(|e| {
-    if had_existing {
-        fs::rename(db_path, &backup_path).map_err(|e| {
         IdxError::Io(format!(
-                "failed to move existing ownership database {} to backup {}: {e}",
+            "failed to install ownership snapshot into {}: {e}",
-                db_path.display(),
-                backup_path.display()
-            ))
-        })?;
-    }
-
-    if let Err(err) = fs::rename(temp_path, db_path) {
-        if had_existing {
-            let _ = fs::rename(&backup_path, db_path);
-        }
-        return Err(IdxError::Io(format!(
-            "failed to install ownership snapshot into {}: {err}",
             db_path.display()
-        )));
-    }
-
-    if had_existing && backup_path.exists() {
-        fs::remove_file(&backup_path).map_err(|e| {
-            IdxError::Io(format!(
-                "failed to remove snapshot backup {}: {e}",
-                backup_path.display()
         ))
     })?;
-    }
 
     remove_sqlite_sidecars(db_path)?;
     Ok(())
@@ -772,17 +782,24 @@ fn read_text(source: &str, context: &str) -> Result<String, IdxError> {
             .header("Accept", "application/json,text/plain;q=0.9,*/*;q=0.8")
             .call()
             .map_err(|e| IdxError::Http(format!("failed to fetch {context}: {e}")))?;
-        let mut body = response.into_body();
+        return response
-        return body
+            .into_body()
+            .with_config()
+            .limit(MANIFEST_LIMIT_BYTES)
             .read_to_string()
             .map_err(|e| IdxError::Http(format!("failed reading {context} body: {e}")));
     }
 
     let path = local_source_path(source);
-    fs::read_to_string(&path)
+    let bytes = read_local_bounded(&path, context, MANIFEST_LIMIT_BYTES)?;
-        .map_err(|e| IdxError::Io(format!("failed to read {context} {}: {e}", path.display())))
+    String::from_utf8(bytes).map_err(|e| {
+        IdxError::ParseError(format!("{context} {} is not UTF-8: {e}", path.display()))
+    })
 }
 
+/// A manifest is a few KB; anything near this is not one.
+const MANIFEST_LIMIT_BYTES: u64 = 1 << 20;
+
 /// 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 {
@@ -816,9 +833,26 @@ fn read_bytes(source: &str, context: &str, limit_bytes: u64) -> Result<Vec<u8>,
             });
     }
 
-    let path = local_source_path(source);
+    read_local_bounded(&local_source_path(source), context, limit_bytes)
-    fs::read(&path)
+}
-        .map_err(|e| IdxError::Io(format!("failed to read {context} {}: {e}", path.display())))
+
+/// Reads at most `limit_bytes` of a regular file (not a device or FIFO); a
+/// larger file comes back truncated and fails the caller's size check.
+fn read_local_bounded(path: &Path, context: &str, limit_bytes: u64) -> Result<Vec<u8>, IdxError> {
+    let read_error = |e: std::io::Error| {
+        IdxError::Io(format!("failed to read {context} {}: {e}", path.display()))
+    };
+    let file = fs::File::open(path).map_err(read_error)?;
+    if !file.metadata().map_err(read_error)?.is_file() {
+        return Err(IdxError::InvalidInput(format!(
+            "{context} {} is not a regular file",
+            path.display()
+        )));
+    }
+    let mut bytes = Vec::new();
+    std::io::Read::read_to_end(&mut std::io::Read::take(file, limit_bytes), &mut bytes)
+        .map_err(read_error)?;
+    Ok(bytes)
 }
 
 fn is_http_source(source: &str) -> bool {
@@ -841,7 +875,8 @@ mod tests {
     use super::{
         DEFAULT_SNAPSHOT_MANIFEST_URL, OwnershipSnapshotArtifact, OwnershipSnapshotManifest,
         OwnershipSnapshotSource, OwnershipSyncAction, SNAPSHOT_MANIFEST_SCHEMA_VERSION,
-        build_sync_decision, parse_manifest, resolve_manifest_source_with,
+        build_backup_path, build_sync_decision, check_download_source, install_snapshot_file,
+        parse_manifest, read_local_bounded, resolve_manifest_source_with, validate_snapshot_db,
     };
     use crate::ownership::db::{ensure_schema, insert_release};
     use crate::ownership::types::OwnershipRelease;
@@ -984,6 +1019,105 @@ mod tests {
         assert!(!decision.should_download, "{}", decision.reason);
     }
 
+    #[test]
+    fn build_sync_decision_ignores_a_replayed_older_build_for_the_same_date() {
+        // Local holds corrected build B (7200 rows); an older build A for the
+        // same release (7154 rows) that does not list B must not undo it.
+        let snapshot = same_date_snapshot(Vec::new());
+        let mut local = same_release_local(&snapshot, &"c".repeat(64));
+        if let Some(release) = local.latest_release.as_mut() {
+            release.row_count = 7200;
+        }
+
+        let decision = build_sync_decision(Some(&local), &snapshot, false);
+        assert!(!decision.should_download, "{}", decision.reason);
+    }
+
+    #[test]
+    fn validation_rejects_a_snapshot_whose_holdings_are_missing() {
+        // Release metadata and manifest agree (7154 rows), but no holdings
+        // are stored: installing it would replace good data with nothing.
+        let snapshot = same_date_snapshot(Vec::new());
+        let path = std::env::temp_dir().join(format!("idx-empty-{}.sqlite", fastrand::u64(..)));
+        let conn = Connection::open(&path).unwrap();
+        ensure_schema(&conn).unwrap();
+        for (index, sha256) in ["1", "2"].into_iter().map(|c| c.repeat(64)).enumerate() {
+            let as_of_date = NaiveDate::from_ymd_opt(2026, 7 + index as u32, 28).unwrap();
+            insert_release(
+                &conn,
+                &OwnershipRelease {
+                    id: 0,
+                    source_url: None,
+                    sha256,
+                    as_of_date,
+                    row_count: 1,
+                    imported_at: 0,
+                },
+            )
+            .unwrap();
+        }
+        insert_release(
+            &conn,
+            &OwnershipRelease {
+                id: 0,
+                source_url: None,
+                sha256: snapshot.latest_release_sha256.clone(),
+                as_of_date: snapshot.latest_as_of_date,
+                row_count: snapshot.latest_row_count,
+                imported_at: 0,
+            },
+        )
+        .unwrap();
+        for code in 0..snapshot.ticker_count {
+            conn.execute(
+                "INSERT INTO tickers (code) VALUES (?1)",
+                [format!("T{code}")],
+            )
+            .unwrap();
+        }
+        drop(conn);
+
+        let err = validate_snapshot_db(&path, &snapshot).expect_err("no holdings stored");
+        assert!(err.to_string().contains("holds 0 rows"), "{err}");
+        std::fs::remove_file(&path).unwrap();
+    }
+
+    #[test]
+    fn remote_manifests_may_not_name_local_artifacts() {
+        let remote = "https://example.com/ownership-snapshot-manifest.json";
+        assert!(check_download_source(remote, "/dev/zero").is_err());
+        assert!(check_download_source(remote, "file:///etc/passwd").is_err());
+        assert!(check_download_source(remote, "https://example.com/a.sqlite").is_ok());
+        // The user's own local manifest may point at a local file.
+        assert!(check_download_source("/tmp/manifest.json", "/tmp/a.sqlite").is_ok());
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn local_reads_refuse_devices() {
+        let err = read_local_bounded(std::path::Path::new("/dev/zero"), "snapshot", 16)
+            .expect_err("a device is not a snapshot file");
+        assert!(err.to_string().contains("not a regular file"), "{err}");
+    }
+
+    #[test]
+    fn install_restores_a_backup_left_by_an_interrupted_older_install() {
+        let dir = std::env::temp_dir().join(format!("idx-install-{}", fastrand::u64(..)));
+        std::fs::create_dir_all(&dir).unwrap();
+        let db = dir.join("ownership.db");
+        // An interrupted pre-0.3.3 install: only the backup is left.
+        std::fs::write(build_backup_path(&db), b"previous").unwrap();
+        let temp = dir.join("incoming.tmp");
+        std::fs::write(&temp, b"snapshot").unwrap();
+
+        install_snapshot_file(&temp, &db).unwrap();
+
+        assert_eq!(std::fs::read(&db).unwrap(), b"snapshot");
+        assert!(!build_backup_path(&db).exists());
+        assert!(!temp.exists());
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
+
     #[test]
     fn build_sync_decision_keeps_a_locally_modified_database() {
         // Same release metadata but a file matching no published build (local

docs/OWNERSHIP_SYNC.md

@@ -62,7 +62,8 @@ Semantics:
   (`ownership-snapshot-YYYY-MM-DD-<sha256>.sqlite`), so consumers must always
   follow `download_url` and validate it with `sqlite_sha256` and `size_bytes`.
 - `sqlite_sha256` and `size_bytes` are validated before install.
-- `latest_*` and `release_count` are validated against the downloaded SQLite contents before replacement.
+- `latest_*` and `release_count` are validated against the downloaded SQLite contents before replacement, including the number of holdings rows actually stored for the latest release.
+- A manifest fetched over `http(s)://` must give an `http(s)://` `download_url`; only a local manifest may point at a local file. Local files must be regular files, and manifests are limited to 1 MiB.
 - `replaces_sqlite_sha256` (optional, omitted when empty) lists earlier builds
   published for the same `latest_as_of_date` that this snapshot supersedes, for
   example after a parser fix. Clients up to v0.2.x ignore it.
@@ -78,9 +79,11 @@ Without `--force`, sync behaves conservatively:
 - local DB file is byte-identical to a build in `replaces_sqlite_sha256`: replace
   (queries never write to a synced file, so this means it was not modified;
   committed changes still in the `-wal` file are checkpointed before hashing)
-- local latest release/date/count matches snapshot: no-op
+- local latest release/date/count matches snapshot: no-op (a same-date build
+  with a different row count replaces the local file only when it lists that
+  file in `replaces_sqlite_sha256`)
 - local latest release matches but has fewer releases than snapshot: replace to fill missing history
 - local DB newer than snapshot: no-op and keep local data
 - local DB diverges at the same latest date: no-op and require `--force`
 
-Replacement is staged through a temp file and validated before the local DB is swapped in.
+Replacement is staged through a temp file and validated before the local DB is swapped in with a single rename, so an interrupted sync leaves either the old database or the new one.