From f61eae8656f3ee1be810871f2fefbfc3a4d003bd Mon Sep 17 00:00:00 2001 From: kert Date: Fri, 11 Sep 2026 17:23:07 -0400 Subject: [PATCH] fix(bib): stop over-scanning on Store open, upsert_status, and sealed backfills (refs #680) - _init_schema skips the duplicate-attachments GROUP BY scan once the unique index already exists (checks sqlite_master first). - upsert_status merges extra_json per key instead of replacing the whole column: a stored key now survives unless the incoming payload sets that specific key to a non-empty value. - docket_counts is one grouped query (a single items scan, left-joined against the enriched-tag set) instead of two separate url LIKE scans. - fingerprint_files widens `except FileNotFoundError` to `except OSError` so NotADirectoryError/PermissionError are handled the same way as a plain missing file. - backfill_details and backfill_from_mirror shared an identical sealed-guard block; factored into one _sealed_backfill_skip helper. Adds coverage: extra_json per-key merge (both directions), docket_counts, the widened OSError catch, attach_file's explicit-title dedupe path (two different source files, same explicit title), a dedupe-migration conflict test for a missing kept file, and a test confirming _init_schema's new short-circuit actually skips the scan. --- src/bib/dockets.py | 5 +- src/bib/regulations_gov.py | 109 +++++++++++---------- src/bib/store.py | 30 +++++- tests/bib/test_dockets.py | 12 +++ tests/bib/test_regulations_gov_exercise.py | 44 +++++++++ tests/bib/test_store_attach_idempotent.py | 25 +++++ tests/bib/test_store_upsert_noop.py | 41 ++++++++ tests/dev/test_dedupe_attachments.py | 78 +++++++++++++++ 8 files changed, 288 insertions(+), 56 deletions(-) diff --git a/src/bib/dockets.py b/src/bib/dockets.py index 087d1ea..629f9f5 100644 --- a/src/bib/dockets.py +++ b/src/bib/dockets.py @@ -66,7 +66,10 @@ def fingerprint_files(paths: Iterable[Path]) -> str: try: st = p.stat() rows.append(f"{p.name}\x00{st.st_size}\x00{st.st_mtime_ns}") - except FileNotFoundError: + except OSError: + # Missing (FileNotFoundError), unreadable (PermissionError), + # or a broken symlink/removed parent dir — any of these + # means "not there" for fingerprinting purposes. rows.append(f"{p.name}\x00-1\x000") rows.sort() return hashlib.sha256("\n".join(rows).encode()).hexdigest() diff --git a/src/bib/regulations_gov.py b/src/bib/regulations_gov.py index 32491b3..ed14725 100644 --- a/src/bib/regulations_gov.py +++ b/src/bib/regulations_gov.py @@ -391,6 +391,42 @@ class Client: # ── Bib integration ──────────────────────────────────────────── +def _sealed_backfill_skip( + store: Store, docket: str, force: bool +) -> dict[str, int] | None: + """Sealed-docket short-circuit shared by :func:`backfill_details` and + :func:`backfill_from_mirror`: both return the same zero-work stats + dict, unchanged, when *docket* is sealed and *force* is false. + + Returns the stats dict to return immediately, or ``None`` when the + caller should proceed (no docket given, ``force``, unknown docket, + or a known-but-open docket). + """ + if not docket or force: + return None + d = store.docket_get(docket) + if d is None or not d.sealed: + return None + log.info( + "%s sealed %s (%s); backfill skipped", + docket, + d.sealed_at[:10], + d.seal_reason, + ) + print( + f"{docket}: sealed {d.sealed_at[:10]} ({d.seal_reason}); skipped — use --force", + flush=True, + ) + return { + "skipped_sealed": 1, + "seen": 0, + "enriched": 0, + "created": 0, + "attached": 0, + "errors": 0, + } + + def backfill_details( store: Store, client: Client, @@ -418,27 +454,9 @@ def backfill_details( so we stop hammering it. Commits land every ``commit_every`` items so a crash loses at most that many items of work. """ - if docket and not force: - d = store.docket_get(docket) - if d is not None and d.sealed: - log.info( - "%s sealed %s (%s); backfill skipped", - docket, - d.sealed_at[:10], - d.seal_reason, - ) - print( - f"{docket}: sealed {d.sealed_at[:10]} ({d.seal_reason}); skipped — use --force", - flush=True, - ) - return { - "skipped_sealed": 1, - "seen": 0, - "enriched": 0, - "created": 0, - "attached": 0, - "errors": 0, - } + skip = _sealed_backfill_skip(store, docket, force) + if skip is not None: + return skip con = store._con() # noqa: SLF001 # Resume rule: treat ``enriched:ok`` as the completion marker. # Using a tag (not just the abstract) lets us handle "see attached" @@ -711,27 +729,9 @@ def backfill_from_mirror( S3 fetches fan out over a thread pool; every Store/sqlite call stays on this thread. Idempotent: ``enriched:ok`` items are skipped. """ - if docket and not force: - d = store.docket_get(docket) - if d is not None and d.sealed: - log.info( - "%s sealed %s (%s); backfill skipped", - docket, - d.sealed_at[:10], - d.seal_reason, - ) - print( - f"{docket}: sealed {d.sealed_at[:10]} ({d.seal_reason}); skipped — use --force", - flush=True, - ) - return { - "skipped_sealed": 1, - "seen": 0, - "enriched": 0, - "created": 0, - "attached": 0, - "errors": 0, - } + skip = _sealed_backfill_skip(store, docket, force) + if skip is not None: + return skip con = store._con() # noqa: SLF001 existing: dict[str, tuple[str, bool]] = {} # cid -> (key, enriched) for key, url, enriched in con.execute( @@ -1030,18 +1030,25 @@ def walk_docket( def docket_counts(store: Store, docket_id: str) -> dict[str, int]: - """``{"comments": n, "enriched": n}`` from SQL only (no filesystem).""" + """``{"comments": n, "enriched": n}`` from SQL only (no filesystem). + + One grouped scan of ``items`` (a left join against the tagged-item + set) rather than two separate ``url LIKE`` scans. + """ con = store._con() # noqa: SLF001 like = f"https://www.regulations.gov/comment/{docket_id}-%" - comments = con.execute( - "SELECT count(*) FROM items WHERE url LIKE ?", (like,) - ).fetchone()[0] - enriched = con.execute( - """SELECT count(*) FROM items i WHERE i.url LIKE ? AND i.id IN ( - SELECT item_id FROM item_tags WHERE tag_id IN (SELECT id FROM tags WHERE name='enriched:ok'))""", + row = con.execute( + """SELECT count(*), + SUM(CASE WHEN et.item_id IS NOT NULL THEN 1 ELSE 0 END) + FROM items i + LEFT JOIN ( + SELECT item_id FROM item_tags + WHERE tag_id IN (SELECT id FROM tags WHERE name = 'enriched:ok') + ) et ON et.item_id = i.id + WHERE i.url LIKE ?""", (like,), - ).fetchone()[0] - return {"comments": comments, "enriched": enriched} + ).fetchone() + return {"comments": row[0], "enriched": row[1] or 0} # ── Internals ────────────────────────────────────────────────── diff --git a/src/bib/store.py b/src/bib/store.py index f28e21a..c73fa52 100644 --- a/src/bib/store.py +++ b/src/bib/store.py @@ -86,7 +86,15 @@ class Store: # The (item_id, filename) unique index can only exist once the # dedupe migration has run (dev/scripts/dedupe_attachments.py); # until then we leave it off rather than fail to open the store. + # Once it exists, sqlite_master says so directly — skip the + # GROUP BY scan (which otherwise runs on every Store() open). con = self._con() + indexed = con.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'index' " + "AND name = 'idx_attachments_item_filename'" + ).fetchone() + if indexed is not None: + return dup = con.execute( "SELECT 1 FROM attachments GROUP BY item_id, filename HAVING count(*) > 1 LIMIT 1" ).fetchone() @@ -266,9 +274,11 @@ class Store: * columns: an empty incoming value keeps the stored one (a list-walk row carries no body, and must not blank an enriched - abstract). ``extra_json`` is exempt — it is always a full - serialization, so an incoming ``{}`` is a real value and - replaces what is stored. + abstract). ``extra_json`` is merged per key instead of + column-wise: a stored key survives unless the incoming + payload sets that key to a non-empty value, so a list-walk + row's empty subclass fields (``{"effective_date": "", ...}``) + never blank an already-stored value. * tags and collections: union-merged with what is stored, never replaced. Removal goes through :meth:`remove_tag`. """ @@ -301,11 +311,22 @@ class Store: cur_row = current.to_row() # Merge, don't replace: an empty incoming value carries no # information (a list-walk row has no body), so the stored value - # wins. extra_json is always a full serialization and is exempt. + # wins. for c in self._COMPARE_COLS: if c != "extra_json" and not new_row.get(c) and cur_row.get(c): new_row[c] = cur_row[c] setattr(item, c, cur_row[c]) + + # extra_json is a full serialization of subclass-specific fields + # every time, so a column-wise empty check can't apply — merge + # the decoded dict per key instead: a stored key survives unless + # the incoming payload sets it to a non-empty value. + cur_extra = json.loads(cur_row.get("extra_json") or "{}") + new_extra = json.loads(new_row.get("extra_json") or "{}") + merged_extra = {**cur_extra, **{k: v for k, v in new_extra.items() if v}} + merged_extra_json = json.dumps(merged_extra) + new_row["extra_json"] = merged_extra_json + same_cols = all( new_row.get(c, "") == cur_row.get(c, "") for c in self._COMPARE_COLS ) @@ -319,6 +340,7 @@ class Store: item.stamp_access() row = item.to_row() row.pop("key", None) + row["extra_json"] = merged_extra_json row["tags"] = merged_tags row["collections"] = merged_cols self.update(ekey, **row) diff --git a/tests/bib/test_dockets.py b/tests/bib/test_dockets.py index 8aded44..6b5b7d1 100644 --- a/tests/bib/test_dockets.py +++ b/tests/bib/test_dockets.py @@ -96,6 +96,18 @@ class TestFingerprintFiles: assert fingerprint_files([]) == fingerprint_files([]) assert len(fingerprint_files([])) == 64 + def test_non_missing_oserror_is_recorded_not_fatal(self, tmp_path: Path): + """refs #680: widened from FileNotFoundError to OSError — a path + under a non-directory (NotADirectoryError, not + FileNotFoundError) must not raise, same as a missing file.""" + f = tmp_path / "not_a_dir.pdf" + f.write_bytes(b"x") + bogus = f / "nested.pdf" # stat() raises NotADirectoryError + assert fingerprint_files([bogus]) == fingerprint_files([bogus]) + assert fingerprint_files([bogus]) == fingerprint_files( + [tmp_path / "nested.pdf"] # same name, both "missing" + ) + class TestQuietDays: def test_default_when_section_missing(self, monkeypatch): diff --git a/tests/bib/test_regulations_gov_exercise.py b/tests/bib/test_regulations_gov_exercise.py index 772f327..66166dc 100644 --- a/tests/bib/test_regulations_gov_exercise.py +++ b/tests/bib/test_regulations_gov_exercise.py @@ -869,3 +869,47 @@ class TestBackfillAddTagGone: stats = backfill_details(store, api, limit=1, scratch_root=tmp_path) assert stats["enriched"] == 1 assert stats["attached"] == 0 + + +class TestDocketCounts: + def test_one_grouped_query(self, tmp_path): + """docket_counts counts comments and enriched ones from one scan + of items (refs #680) — not two separate ``url LIKE`` queries.""" + from bib.item import Source + from bib.regulations_gov import docket_counts + from bib.store import Store + + store = Store(tmp_path / "bib.sqlite") + k1 = store.create( + Source( + title="Comment 1", + url="https://www.regulations.gov/comment/CMS-2023-0001-0001", + ) + ) + k2 = store.create( + Source( + title="Comment 2", + url="https://www.regulations.gov/comment/CMS-2023-0001-0002", + ) + ) + store.create( + Source( + title="Other docket", + url="https://www.regulations.gov/comment/CMS-2099-9999-0001", + ) + ) + store.add_tags(k1, ["enriched:ok"]) + + assert k1 and k2 # both created without error + counts = docket_counts(store, "CMS-2023-0001") + assert counts == {"comments": 2, "enriched": 1} + store.close() + + def test_no_matches(self, tmp_path): + from bib.regulations_gov import docket_counts + from bib.store import Store + + store = Store(tmp_path / "bib.sqlite") + counts = docket_counts(store, "CMS-9999-9999") + assert counts == {"comments": 0, "enriched": 0} + store.close() diff --git a/tests/bib/test_store_attach_idempotent.py b/tests/bib/test_store_attach_idempotent.py index 403439c..f78642d 100644 --- a/tests/bib/test_store_attach_idempotent.py +++ b/tests/bib/test_store_attach_idempotent.py @@ -28,3 +28,28 @@ def test_different_filename_creates_second_row(tmp_path: Path): f2 = tmp_path / "b.pdf" f2.write_bytes(b"b") assert s.attach_file(key, f1) != s.attach_file(key, f2) + + +def test_explicit_title_dedupes_across_different_source_files(tmp_path: Path): + """refs #680: the dedupe key is (item_id, filename) where filename + is ``title or path.name`` — an explicit title must dedupe on its + own value even when the two calls point at physically different + source files (different path.name, different bytes).""" + s = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage") + key = s.create(Source(title="T", url="https://x/1")) + f1 = tmp_path / "original_name_1.pdf" + f1.write_bytes(b"first version") + f2 = tmp_path / "totally_different_name.pdf" + f2.write_bytes(b"second version, different bytes and path.name") + + k1 = s.attach_file(key, f1, title="Comment text") + k2 = s.attach_file(key, f2, title="Comment text") + + assert k1 == k2 + n = s._con().execute("SELECT count(*) FROM attachments").fetchone()[0] + assert n == 1 + # only the first file's bytes were ever copied into storage + assert len(list((tmp_path / "storage").iterdir())) == 1 + stored = next((tmp_path / "storage").iterdir()) + copied = next(stored.iterdir()) + assert copied.read_bytes() == b"first version" diff --git a/tests/bib/test_store_upsert_noop.py b/tests/bib/test_store_upsert_noop.py index a75fb8e..9acfa3f 100644 --- a/tests/bib/test_store_upsert_noop.py +++ b/tests/bib/test_store_upsert_noop.py @@ -102,3 +102,44 @@ def test_non_empty_incoming_still_updates(): key, _ = s.upsert_status(_item(abstract="v1")) _, status = s.upsert_status(_item(abstract="v2")) assert status == "updated" and s.get(key).abstract == "v2" + + +# ── extra_json merges per key (refs #680) ──────────────────────────── + + +def _rule(**kw): + from bib.item import Rule + + it = Rule(title="T", url="https://example.com/rule-1") + it.cms_id = kw.get("cms_id", "") + it.effective_date = kw.get("effective_date", "") + it.fr_page = kw.get("fr_page", "") + return it + + +def test_empty_incoming_extra_json_key_keeps_stored_key(): + """A list-walk Rule with a blank effective_date must not blank an + already-stored one — extra_json is merged per key, not replaced + wholesale.""" + s = _store() + key, _ = s.upsert_status(_rule(cms_id="CMS-1848-P", effective_date="2026-01-01")) + _, status = s.upsert_status(_rule(cms_id="CMS-1848-P", effective_date="")) + assert status == "unchanged" + stored = s.get(key) + assert stored.effective_date == "2026-01-01" + assert stored.cms_id == "CMS-1848-P" + + +def test_non_empty_incoming_extra_json_key_updates_that_key_only(): + s = _store() + key, _ = s.upsert_status(_rule(cms_id="CMS-1848-P", effective_date="2026-01-01")) + _, status = s.upsert_status( + _rule(cms_id="CMS-1848-P", effective_date="", fr_page="12345") + ) + assert status == "updated" + stored = s.get(key) + # New key applied... + assert stored.fr_page == "12345" + # ...but the blank incoming effective_date did not blank the stored one. + assert stored.effective_date == "2026-01-01" + assert stored.cms_id == "CMS-1848-P" diff --git a/tests/dev/test_dedupe_attachments.py b/tests/dev/test_dedupe_attachments.py index ed7db3e..b95d007 100644 --- a/tests/dev/test_dedupe_attachments.py +++ b/tests/dev/test_dedupe_attachments.py @@ -98,6 +98,27 @@ def test_refuses_group_with_differing_sizes(tmp_path: Path): assert rep.rows_removed == 0 and rep.skipped_conflicts == 1 +def test_conflict_when_kept_file_is_missing(tmp_path: Path): + """refs #680: the kept row's file being gone (e.g. an earlier + partial cleanup, or a manual delete) must not be treated as "no + conflict" just because a missing file's _size() sentinel (-1) + happens to also mean "ignore mismatches" for individual dupes — a + duplicate with real bytes on disk while the kept copy has none is + exactly the size-mismatch case this guard exists for.""" + s, _ = _seed(tmp_path) + (tmp_path / "storage" / "AAAAAAAA" / "attachment_1.pdf").unlink() + groups = mod.plan(s._con()) + assert len(groups) == 1 + assert groups[0].conflict is True + rep = mod.apply(s._con(), groups) + assert rep.rows_removed == 0 + assert rep.skipped_conflicts == 1 + # nothing touched: all three rows and both surviving files remain + assert s._con().execute("SELECT count(*) FROM attachments").fetchone()[0] == 3 + assert (tmp_path / "storage" / "BBBBBBBB" / "attachment_1.pdf").is_file() + assert (tmp_path / "storage" / "CCCCCCCC" / "attachment_1.pdf").is_file() + + def test_unique_index_created_only_when_clean(tmp_path: Path): s, _ = _seed(tmp_path) s.close() @@ -117,6 +138,63 @@ def test_unique_index_created_only_when_clean(tmp_path: Path): assert "idx_attachments_item_filename" in idx +class _ConnSpy: + """Wraps a real sqlite3.Connection to record GROUP BY calls. + + ``sqlite3.Connection`` is a C type and can't be monkeypatched + in-place, so this proxies everything through ``__getattr__`` except + ``execute`` (recorded) and attribute assignment (forwarded — needed + for ``row_factory``, which ``Store._con`` sets directly).""" + + def __init__(self, real: sqlite3.Connection) -> None: + object.__setattr__(self, "_real", real) + object.__setattr__(self, "group_by_calls", []) + + def execute(self, sql, *args, **kwargs): + if "GROUP BY item_id, filename" in sql: + self.group_by_calls.append(sql) + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + def __setattr__(self, name, value): + setattr(self._real, name, value) + + +def test_reopen_skips_group_by_scan_once_indexed(tmp_path: Path, monkeypatch): + """refs #680: Store._init_schema short-circuits via sqlite_master + once the unique index exists — a re-open must not re-run the + duplicate-attachments GROUP BY scan at all.""" + s, _ = _seed(tmp_path) + mod.apply(s._con(), mod.plan(s._con())) + s.close() + + # First reopen creates the unique index (no duplicates left). + s_setup = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage") + s_setup.close() + + import bib.store as store_mod + + real_connect = store_mod.sqlite3.connect + spies: list[_ConnSpy] = [] + + def fake_connect(path, *args, **kwargs): + spy = _ConnSpy(real_connect(path, *args, **kwargs)) + spies.append(spy) + return spy + + monkeypatch.setattr(store_mod.sqlite3, "connect", fake_connect) + + # A further reopen must skip the GROUP BY scan now that the index + # already exists. + s2 = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage") + s2.close() + + assert len(spies) == 1 + assert spies[0].group_by_calls == [] + + class _FailAfterFirstDelete: """Connection proxy that dies partway through the transaction."""