Merge remote-tracking branch 'origin/main' into feat/249-system-maintenance

database/review.py

@@ -90,6 +90,30 @@ _PENDING_AT = (
     "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.resolver_version=? "
     "AND i.state='pending' LIMIT 1"
 )
+# Bytes verified at this stat whose item carries a decision (any version).
+_DECIDED_AT = (
+    f"SELECT 1 FROM {FILES_TABLE} f "  # nosec B608
+    f"JOIN {BYTES_TABLE} b ON b.file_id=f.id "
+    f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id "
+    "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? "
+    "AND (i.state<>'pending' OR i.revision<>0 OR EXISTS (SELECT 1 FROM "
+    f"{DECISIONS_TABLE} d WHERE d.item_id=i.id)) LIMIT 1"
+)
+# The other items of one byte record (another version or evidence), with
+# whether each may be superseded: pending, revision 0, no decision row.
+_OTHER_ITEMS_OF_BYTES = (
+    "SELECT i.id, (i.state='pending' AND i.revision=0 AND NOT EXISTS "  # nosec B608
+    f"(SELECT 1 FROM {DECISIONS_TABLE} d WHERE d.item_id=i.id)) "
+    f"FROM {ITEMS_TABLE} i WHERE i.byte_id=? "
+    "AND NOT (i.resolver_version=? AND i.evidence_hash=?)"
+)
+# Pending items of the bytes verified at one stat, with their versions.
+_PENDING_AT_STAT = (
+    f"SELECT i.id,i.resolver_version FROM {FILES_TABLE} f "  # nosec B608
+    f"JOIN {BYTES_TABLE} b ON b.file_id=f.id "
+    f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id "
+    "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.state='pending'"
+)
 # Pending items of one file: the candidates a link settles.
 _PENDING_OF_LOCATOR = (
     f"SELECT i.id FROM {FILES_TABLE} f "  # nosec B608
@@ -274,6 +298,12 @@ class ReviewOperations:
         ``LibraryCommandError("storage-changed")`` when the storage epoch is
         no longer ``epoch`` (a restore since the capture) and ``ValueError``
         for invalid facts or evidence.
+
+        One file never holds two pending items for the same bytes: the new
+        item supersedes the byte record's other items (an older resolver or
+        semantics version) when every one is pending and undecided (the safe
+        delete of the self-heal); when one carries a decision or a later
+        revision, it is kept and nothing new is stored (False).
         """
         _validate_target(locator, sha256, size, mtime_ns)
         _validate_spelling(spelled_path, locator)
@@ -297,6 +327,18 @@ class ReviewOperations:
             )
             if created or moved:
                 conn.execute(_BUMP_IMAGE_INDEX)
+            # The byte record is created or refreshed even when the item is
+            # refused below (an observe would do the same); that commits
+            # with the image index bump, and the review queue is unchanged.
+            others = conn.execute(
+                _OTHER_ITEMS_OF_BYTES, (byte_id, resolver_version, evidence_hash)
+            ).fetchall()
+            if any(not removable for _id, removable in others):
+                return False
+            superseded = sum(
+                conn.execute(_SETTLE_ITEM, (item_id,)).rowcount
+                for item_id, _removable in others
+            )
             cursor = conn.execute(
                 _INSERT_ITEM,
                 (
@@ -314,9 +356,9 @@ class ReviewOperations:
             )
             inserted = cursor.rowcount > 0
             refreshed = _refresh_spelling(conn, locator, spelled_path)
-            if inserted or refreshed:
+            if inserted or refreshed or superseded:
                 conn.execute(_BUMP_REVIEW_INDEX)
-        if inserted or refreshed:
+        if inserted or refreshed or superseded:
             notify_review_change(self.model.db_path)
         return inserted
 
@@ -389,6 +431,19 @@ class ReviewOperations:
         )
         return row is not None
 
+    def review_decided_at(self, locator: str, size: int, mtime_ns: int) -> bool:
+        """True when bytes verified at this stat carry a decided item.
+
+        Any version: a newer resolver or table never adds an item beside a
+        decision, so the rescan need not copy those bytes again.
+        """
+        row = (
+            self.model.get_connection()
+            .execute(_DECIDED_AT, (locator, size, mtime_ns))
+            .fetchone()
+        )
+        return row is not None
+
     def review_pending_at(
         self, locator: str, size: int, mtime_ns: int, resolver_version: int
     ) -> bool:
@@ -445,6 +500,38 @@ class ReviewOperations:
             notify_review_change(self.model.db_path)
         return removed
 
+    def settle_stale_review_items(
+        self, locator: str, size: int, mtime_ns: int, versions
+    ) -> int:
+        """Remove pending, undecided items of the bytes verified at this stat
+        whose ``resolver_version`` is not one of ``versions``.
+
+        The caller found these bytes untraced under the current versions, so
+        an older version's item could only refuse Apply. The same safe delete
+        as the other settles; one indexed read when nothing is pending.
+        """
+        _validate_locator(locator)
+        current = set(versions)
+        conn = self.model.get_connection()
+        stale = [
+            item_id
+            for item_id, version in conn.execute(
+                _PENDING_AT_STAT, (locator, size, mtime_ns)
+            )
+            if version not in current
+        ]
+        if not stale:
+            return 0
+        removed = 0
+        with self._pair_transaction() as conn:
+            for item_id in stale:
+                removed += conn.execute(_SETTLE_ITEM, (item_id,)).rowcount
+            if removed:
+                conn.execute(_BUMP_REVIEW_INDEX)
+        if removed:
+            notify_review_change(self.model.db_path)
+        return removed
+
     def settle_untraced_review_items(self, locators) -> int:
         """Remove the pending, undecided items that offer nothing to choose.