fix(images): keep the maintenance link reader, refine the one-step fallback (#354)

The one-step reader was named _link_rows, which shadowed
MaintenanceOperations._link_rows on PromptDatabase and broke the Missing
links preview and apply; it is now _one_step_link_rows, and a test pins
that the maintenance reader is not shadowed.

An interrupted read (Move library retiring the connection) is raised as
before instead of falling back and spending the one warning; a JSON text
over SQLite's length limit (DataError) streams like the other value
failures. The equivalence test compares value types per path rather than
by row order, and a test checks that the read leaves no WAL snapshot
pinned.

database/operations.py

@@ -2597,11 +2597,14 @@ class PromptDatabase(
         grows with the database's matching linked rows (about 530 bytes each,
         freed on return); the result is bounded by *wanted* (the listing's
         normalized paths) when given. Where the JSON step fails (no JSON
-        functions, a BLOB value) the rows are streamed as before.
+        functions, a BLOB value, a JSON text over SQLite's length limit) the
+        rows are streamed as before.
         """
         sql, params = link_filter_query(q, category, tags, every_link=every_link)
         with self.model.get_connection() as conn:
-            rows = None if _one_step_unsupported else self._link_rows(conn, sql, params)
+            rows = None
+            if not _one_step_unsupported:
+                rows = self._one_step_link_rows(conn, sql, params)
             if rows is None:
                 return _streamed_link_matches(conn, sql, params, wanted)
         return {
@@ -2610,10 +2613,14 @@ class PromptDatabase(
             if wanted is None or path in wanted
         }
 
-    def _link_rows(
+    def _one_step_link_rows(
         self, conn: sqlite3.Connection, sql: str, params: Dict[str, Any]
     ) -> Optional[List[List[Any]]]:
-        """*sql*'s rows from one JSON value, or None when this read must stream."""
+        """*sql*'s rows from one JSON value, or None when this read must stream.
+
+        An interrupted read (Move library retiring the connection) is raised
+        as before, never retried row by row.
+        """
         global _one_step_unsupported, _one_step_warned
         error: Optional[str] = None
         try:
@@ -2622,8 +2629,11 @@ class PromptDatabase(
                 ((document,),) = cursor.fetchall()
             finally:
                 cursor.close()  # resets the statement: its snapshot ends now
-        except sqlite3.OperationalError as failure:
+        except (sqlite3.OperationalError, sqlite3.DataError) as failure:
+            # DataError: the JSON text is longer than SQLITE_LIMIT_LENGTH.
             error = str(failure)
+            if error == "interrupted":
+                raise
             if error.startswith("no such function"):
                 _one_step_unsupported = True
         else:

tests/test_database_link_matches.py

@@ -9,6 +9,7 @@ the fallback, on every filter and on odd values.
 
 import itertools
 import os
+import sqlite3
 import sys
 import tempfile
 import unittest
@@ -209,8 +210,8 @@ class TestOneStepLinkRead(LinkMatchesTestCase):
                 )
                 self.assertEqual(actual, expected)
                 self.assertEqual(
-                    [type(value) for row in actual.values() for value in row],
-                    [type(value) for row in expected.values() for value in row],
+                    {path: tuple(map(type, row)) for path, row in actual.items()},
+                    {path: tuple(map(type, row)) for path, row in expected.items()},
                 )
                 found = self.selects()
                 self.assertEqual(len(found), 1, found)
@@ -238,6 +239,23 @@ class TestOneStepLinkRead(LinkMatchesTestCase):
             {},
         )
 
+    def test_read_releases_its_snapshot(self):
+        # A statement left open after the read would pin its WAL snapshot, and
+        # a TRUNCATE checkpoint from another connection would report busy.
+        self.seed()
+        self.assertTrue(self.db.get_image_link_matches("", None, [], every_link=True))
+        other = sqlite3.connect(str(self.root / "prompts.db"))
+        try:
+            other.execute("UPDATE prompts SET rating=2 WHERE id=1")
+            other.commit()
+            busy, _log, _done = other.execute(
+                "PRAGMA wal_checkpoint(TRUNCATE)"
+            ).fetchone()
+        finally:
+            other.close()
+        self.assertEqual(busy, 0)
+        self.assertEqual(os.path.getsize(self.root / "prompts.db-wal"), 0)
+
     def test_wrapped_query_keeps_the_ambiguity_probe_an_index_search(self):
         self.seed()
         sql, params = link_filter_query("lantern", "cat-a", ["t1"], every_link=True)
@@ -330,13 +348,51 @@ class TestStreamedFallback(LinkMatchesTestCase):
                 self.db.get_image_link_matches("", None, [], every_link=True), expected
             )
 
+    def test_json_text_over_the_length_limit_streams(self):
+        self.seed()
+        expected = self.streamed("", None, [], True)
+        conn = self.db.model.get_connection()
+        limit = conn.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, 200)
+        try:
+            with self.assertLogs(LOGGER, "WARNING") as logged:
+                actual = self.db.get_image_link_matches("", None, [], every_link=True)
+        finally:
+            conn.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, limit)
+        self.assertEqual(actual, expected)
+        self.assertIn("too big", logged.output[0])
+        self.assertFalse(operations._one_step_unsupported)
+
+    def test_an_interrupted_read_raises_without_a_fallback(self):
+        # Move library interrupts a retired connection: the read fails as
+        # before, and the one warning is kept for a real fallback.
+        self.seed()
+        conn = self.db.model.get_connection()
+        conn.set_progress_handler(lambda: 1, 1)
+        self.statements.clear()
+        try:
+            with (
+                self.assertNoLogs(LOGGER, "WARNING"),
+                self.assertRaisesRegex(sqlite3.OperationalError, "^interrupted$"),
+            ):
+                self.db.get_image_link_matches("", None, [], every_link=True)
+        finally:
+            conn.set_progress_handler(None, 1)
+        self.assertFalse(operations._one_step_warned)
+        self.assertFalse(operations._one_step_unsupported)
+        self.assertFalse([s for s in self.statements if "json_group_array" not in s])
+
+    def test_the_maintenance_link_rows_reader_is_not_shadowed(self):
+        from database.maintenance import MaintenanceOperations
+
+        self.assertIs(PromptDatabase._link_rows, MaintenanceOperations._link_rows)
+
     def test_a_failed_streamed_read_still_raises(self):
         with (
             patch.object(operations, "_LINK_MATCHES_JSON_SQL", MISSING_FUNCTION_SQL),
             patch.object(operations, "_LINK_FILTER_SQL", "SELECT broken FROM nowhere"),
             self.assertLogs(LOGGER, "WARNING"),
         ):
-            with self.assertRaises(operations.sqlite3.OperationalError):
+            with self.assertRaises(sqlite3.OperationalError):
                 self.db.get_image_link_matches("", None, [], every_link=True)