fix(sync): releases record stored rows; oversized manifests are errors

From the Codex re-gate of the 0.3.3 fixes:
- Imports recorded row_count as the rows parsed, but INSERT OR IGNORE
  drops a file's repeated ticker/investor rows, so the count could exceed
  what is stored, and the new stored-row validation would then reject a
  snapshot the publisher built from such a file. Record the rows stored.
- A local manifest over the 1 MiB limit was read truncated rather than
  rejected; read one byte more and fail when it does not fit.
- The backup-recovery test also passed on the old code; it now fails an
  install after the leftover backup was found and checks the old data is
  back.

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

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

@@ -355,7 +355,13 @@ pub fn write_ksei_release(
         }
 
         let inserted = insert_ksei_holdings_rows(conn, holdings)?;
-        let _ = insert_release(conn, release)?;
+        // Record the rows stored, not parsed: INSERT OR IGNORE drops repeated
+        // ticker/investor rows, and snapshot validation counts what is stored.
+        let stored = OwnershipRelease {
+            row_count: inserted,
+            ..release.clone()
+        };
+        let _ = insert_release(conn, &stored)?;
         Ok(inserted)
     })();
 
@@ -1616,6 +1622,38 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_release_records_the_rows_it_stored() {
+        // A file listing the same ticker/investor twice stores one row; the
+        // release metadata must say 1, or snapshot validation rejects it.
+        let conn = setup();
+        let bbca = upsert_ticker(&conn, "BBCA", None).unwrap();
+        write_release(
+            &conn,
+            "dup",
+            "2026-02-27",
+            1,
+            bbca,
+            &[(None, "ALPHA", 1000), (None, "ALPHA", 1000)],
+        );
+
+        let row_count: i64 = conn
+            .query_row(
+                "SELECT row_count FROM ownership_releases WHERE sha256 = 'dup'",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        let stored: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM ksei_holdings WHERE release_sha256 = 'dup'",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!((row_count, stored), (1, 1));
+    }
+
     #[test]
     fn changes_reject_a_date_without_a_release() {
         let conn = setup();

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

@@ -791,7 +791,14 @@ fn read_text(source: &str, context: &str) -> Result<String, IdxError> {
     }
 
     let path = local_source_path(source);
-    let bytes = read_local_bounded(&path, context, MANIFEST_LIMIT_BYTES)?;
+    // One byte over the limit tells an oversized manifest from one that fits.
+    let bytes = read_local_bounded(&path, context, MANIFEST_LIMIT_BYTES + 1)?;
+    if bytes.len() as u64 > MANIFEST_LIMIT_BYTES {
+        return Err(IdxError::InvalidInput(format!(
+            "{context} {} is larger than {MANIFEST_LIMIT_BYTES} bytes",
+            path.display()
+        )));
+    }
     String::from_utf8(bytes).map_err(|e| {
         IdxError::ParseError(format!("{context} {} is not UTF-8: {e}", path.display()))
     })
@@ -1105,19 +1112,38 @@ mod tests {
         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.
+        // An interrupted pre-0.3.3 install left only the backup. Even if this
+        // install then fails (its temp file is gone), the old data comes back
+        // instead of being deleted, as the old code did first.
         std::fs::write(build_backup_path(&db), b"previous").unwrap();
-        let temp = dir.join("incoming.tmp");
+        let missing_temp = dir.join("incoming.tmp");
+
+        assert!(install_snapshot_file(&missing_temp, &db).is_err());
+        assert_eq!(std::fs::read(&db).unwrap(), b"previous");
+
+        let temp = dir.join("incoming-2.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 local_manifests_over_the_limit_are_rejected_not_truncated() {
+        let path = std::env::temp_dir().join(format!("idx-manifest-{}.json", fastrand::u64(..)));
+        std::fs::write(
+            &path,
+            vec![b' '; (super::MANIFEST_LIMIT_BYTES + 1) as usize],
+        )
+        .unwrap();
+
+        let err = super::read_text(path.to_str().unwrap(), "manifest").unwrap_err();
+
+        assert!(err.to_string().contains("larger than"), "{err}");
+        std::fs::remove_file(&path).unwrap();
+    }
+
     #[test]
     fn build_sync_decision_keeps_a_locally_modified_database() {
         // Same release metadata but a file matching no published build (local