fix(dev): dedupe unlinks after commit; report removed keys (refs #615)

apply() deleted storage copies inside the BEGIN/COMMIT, so a failure
partway through rolled the rows back and left the surviving rows
pointing at files that were already gone. Collect the doomed paths
during the transaction, COMMIT, then unlink — rows and files can now
only be lost together.

--apply also prints how many keys it removed (first 20), and main()
closes the connection it opens.

dev/scripts/dedupe_attachments.py

@@ -92,7 +92,16 @@ def _size(path: str) -> int:
 
 
 def apply(con: sqlite3.Connection, groups: list[Group]) -> Report:
+    """Delete the duplicate rows, then their storage copies.
+
+    Rows first, files second, with the COMMIT in between: a failure
+    mid-transaction rolls the rows back, and rows that still exist must
+    still have their files. Unlinking inside the transaction would leave
+    surviving rows pointing at nothing — the one outcome this cleanup
+    must never produce.
+    """
     rep = Report(groups=len(groups))
+    doomed: list[Path] = []
     con.execute("BEGIN")
     try:
         for g in groups:
@@ -108,19 +117,20 @@ def apply(con: sqlite3.Connection, groups: list[Group]) -> Report:
                 rep.rows_removed += 1
                 rep.removed_keys.append(key)
                 if not still_referenced:
-                    p = Path(path)
-                    if p.is_file():
-                        rep.bytes_freed += p.stat().st_size
-                        p.unlink()
-                        rep.files_removed += 1
-                        try:
-                            p.parent.rmdir()  # the per-key dir, if now empty
-                        except OSError:
-                            pass
+                    doomed.append(Path(path))
         con.execute("COMMIT")
     except Exception:
         con.execute("ROLLBACK")
         raise
+    for p in doomed:
+        if p.is_file():
+            rep.bytes_freed += p.stat().st_size
+            p.unlink()
+            rep.files_removed += 1
+            try:
+                p.parent.rmdir()  # the per-key dir, if now empty
+            except OSError:
+                pass
     return rep
 
 
@@ -143,20 +153,28 @@ def main(argv: list[str] | None = None) -> int:
 
         db = str(path("db.bib"))
     con = sqlite3.connect(db, isolation_level=None)
-    groups = plan(con)
-    conflicts = sum(1 for g in groups if g.conflict)
-    rows = sum(len(g.remove) for g in groups)
-    print(
-        f"{db}: {len(groups)} duplicate groups, {rows} rows to remove, {conflicts} conflicts (size mismatch, skipped)"
-    )
-    if not args.apply:
-        print("dry run — pass --apply to delete")
-        return 0
-    rep = apply(con, groups)
+    try:
+        groups = plan(con)
+        conflicts = sum(1 for g in groups if g.conflict)
+        rows = sum(len(g.remove) for g in groups)
+        print(
+            f"{db}: {len(groups)} duplicate groups, {rows} rows to remove, "
+            f"{conflicts} conflicts (size mismatch, skipped)"
+        )
+        if not args.apply:
+            print("dry run — pass --apply to delete")
+            return 0
+        rep = apply(con, groups)
+    finally:
+        con.close()
     print(
         f"removed rows={rep.rows_removed} files={rep.files_removed} "
         f"freed={rep.bytes_freed / 1e6:.1f} MB skipped_conflicts={rep.skipped_conflicts}"
     )
+    if rep.removed_keys:
+        head = ", ".join(rep.removed_keys[:20])
+        more = " …" if len(rep.removed_keys) > 20 else ""
+        print(f"removed keys ({len(rep.removed_keys)}): {head}{more}")
     print("reopen the store once (any `stack bib` command) to create the unique index")
     return 0
 

tests/dev/test_dedupe_attachments.py

@@ -115,3 +115,44 @@ def test_unique_index_created_only_when_clean(tmp_path: Path):
         for r in s3._con().execute("SELECT name FROM sqlite_master WHERE type='index'")
     }
     assert "idx_attachments_item_filename" in idx
+
+
+class _FailAfterFirstDelete:
+    """Connection proxy that dies partway through the transaction."""
+
+    def __init__(self, con: sqlite3.Connection) -> None:
+        self._con = con
+        self.deletes = 0
+
+    def execute(self, sql: str, *args):
+        if sql.lstrip().upper().startswith("DELETE"):
+            self.deletes += 1
+            if self.deletes == 2:
+                raise sqlite3.OperationalError("disk I/O error")
+        return self._con.execute(sql, *args)
+
+
+def test_apply_removes_no_file_when_the_transaction_fails(tmp_path: Path):
+    """A rollback restores the rows — so the files they point at must
+    still be there. Unlinking inside the transaction loses data."""
+    import pytest
+
+    s, _ = _seed(tmp_path)
+    groups = mod.plan(s._con())
+    flaky = _FailAfterFirstDelete(s._con())
+    with pytest.raises(sqlite3.OperationalError):
+        mod.apply(flaky, groups)
+    assert s._con().execute("SELECT count(*) FROM attachments").fetchone()[0] == 3
+    for k in ("AAAAAAAA", "BBBBBBBB", "CCCCCCCC"):
+        assert (tmp_path / "storage" / k / "attachment_1.pdf").is_file()
+
+
+def test_main_apply_reports_removed_keys(tmp_path: Path, capsys):
+    s, _ = _seed(tmp_path)
+    s.close()
+    rc = mod.main(["--db", str(tmp_path / "bib.sqlite"), "--apply"])
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "removed rows=2" in out
+    assert "removed keys (2)" in out
+    assert "BBBBBBBB" in out and "CCCCCCCC" in out