docs(testing): add the concurrent SQLite test rule to the Windows preflight (#242)

The retired-owner projection tests park a history read at a barrier
while the test performs a full multipart restore, then release it. The
barrier gave up after 5 s and raised inside the route handler, so on a
slow Windows runner (a 694 s suite run) aiohttp answered with its
text/plain 500 and the test reported 'history route missing'.

Every wait already ends on an event; the ceiling now only stops a broken
test from hanging (60 s). With restore slowed by 5.5 s locally, all 8
subtests failed with text/plain before and pass after.

TESTING.md

@@ -134,6 +134,13 @@ Before changing paths, SQLite, streams, restore or cleanup:
   exercise the actual native branch, then prove handle release and successful
   cleanup. Do not hide failures with blanket skips or retries.
   See Python's [remove/unlink semantics](https://docs.python.org/3.13/library/os.html#os.remove).
+- Concurrent SQLite tests must not depend on how fast the host commits.
+  Managed connections use WAL with `synchronous=NORMAL` (see
+  `configure_connection`) so commits do not hold the write lock across a
+  flush. Make real overlap deterministic (for example hold the first writer
+  open with `FirstWriterHold`) and give peers that must all succeed the
+  ordinary busy timeout; cross-thread handoffs wait on events with generous
+  ceilings, never a short wall clock (#242).
 - Run native Windows CI early for these changes and include a Windows impact and
   evidence note in the PR. All twelve OS/Python combinations and every applicable
   exact-head check must pass before merge. Portable `ntpath`/`PureWindowsPath`

database/execution_history.py

@@ -25,6 +25,10 @@ except ImportError:
 
 
 _logger = get_logger("prompt_manager.database.execution_history")
+# Bounded wait for a history write lock. History is best effort and must
+# never stall ComfyUI's queue: a write that cannot start in time raises
+# "database is locked" and the writer records a gap instead of waiting.
+EXECUTION_BUSY_TIMEOUT_MS = 250
 _restore_lock = threading.Lock()
 _restore_observers: dict[str, dict[object, Callable[[], None]]] = {}
 
@@ -100,7 +104,7 @@ class ExecutionHistoryOperations:
                 "Execution history requires its own transaction"
             )
         timeout = conn.execute("PRAGMA busy_timeout").fetchone()[0]
-        conn.execute("PRAGMA busy_timeout=250")
+        conn.execute(f"PRAGMA busy_timeout={int(EXECUTION_BUSY_TIMEOUT_MS)}")
         try:
             conn.execute("BEGIN IMMEDIATE")
             try:

database/models.py

@@ -101,6 +101,24 @@ SQLITE_HEADER = b"SQLite format 3\x00"
 REQUIRED_PROMPT_COLUMNS = ("id", "text", "created_at")
 
 
+def configure_connection(conn: sqlite3.Connection) -> None:
+    """Apply the locking and durability settings every managed connection shares.
+
+    WAL lets readers run beside the single writer. In WAL mode
+    ``synchronous=NORMAL`` keeps the database consistent after an application
+    or ComfyUI crash and flushes only at checkpoints, so a commit no longer
+    holds SQLite's write lock across a disk flush. Under ``FULL`` a slow flush
+    (common on Windows) let other writers' polling busy handlers starve past
+    BUSY_TIMEOUT_SECONDS and fail with "database is locked" (#242). Without
+    WAL (for example an in-memory database) SQLite's FULL default is kept.
+    """
+    journal_mode = conn.execute("PRAGMA journal_mode = WAL").fetchone()[0]
+    conn.execute("PRAGMA foreign_keys = ON")
+    conn.execute(f"PRAGMA busy_timeout = {BUSY_TIMEOUT_SECONDS * 1000}")
+    if str(journal_mode).lower() == "wal":
+        conn.execute("PRAGMA synchronous = NORMAL")
+
+
 class PromptModel:
     """Database model for prompt storage and schema management."""
 
@@ -153,9 +171,7 @@ class PromptModel:
             try:
                 conn = sqlite3.connect(self.db_path, timeout=BUSY_TIMEOUT_SECONDS)
                 try:
-                    conn.execute("PRAGMA journal_mode = WAL")
-                    conn.execute(f"PRAGMA busy_timeout = {BUSY_TIMEOUT_SECONDS * 1000}")
-                    conn.execute("PRAGMA foreign_keys = ON")
+                    configure_connection(conn)
                     self._create_tables(conn)
                     self._create_indexes(conn)
                     conn.commit()
@@ -538,10 +554,11 @@ class PromptModel:
 
         Each thread gets its own ``sqlite3.Connection`` (kept in
         ``threading.local``), created on first use and reused by that thread
-        thereafter. Every connection is configured with WAL journaling,
-        ``row_factory = sqlite3.Row``, ``PRAGMA foreign_keys = ON`` and
-        ``PRAGMA busy_timeout`` of BUSY_TIMEOUT_SECONDS so concurrent writers
-        wait instead of failing. Callers use it as a context manager
+        thereafter. Every connection gets ``row_factory = sqlite3.Row`` and the
+        shared settings of ``configure_connection``: WAL journaling with
+        ``synchronous=NORMAL``, ``PRAGMA foreign_keys = ON`` and a
+        ``busy_timeout`` of BUSY_TIMEOUT_SECONDS so concurrent writers wait
+        instead of failing. Callers use it as a context manager
         (``with conn:``), which commits or rolls back that thread's transaction
         and leaves the connection open. No lock is held while a connection is in use.
 
@@ -558,9 +575,7 @@ class PromptModel:
             self.db_path, check_same_thread=False, timeout=BUSY_TIMEOUT_SECONDS
         )
         conn.row_factory = sqlite3.Row
-        conn.execute("PRAGMA journal_mode = WAL")
-        conn.execute("PRAGMA foreign_keys = ON")
-        conn.execute(f"PRAGMA busy_timeout = {BUSY_TIMEOUT_SECONDS * 1000}")
+        configure_connection(conn)
         self._local.conn = conn
         self._local.generation = self._generation
         with self._conn_lock:

tests/test_api_history.py

@@ -30,6 +30,13 @@ from tests.test_execution_history_database import JOB, SESSION
 from utils.execution_contexts import ContextBundle, RecipeContext, TextEvidence
 from utils.execution_history_events import make_finish, make_start
 
+# Ceiling for the restore handoffs in assert_projection_releases_retired_owner.
+# Each wait ends on an event (the read reaching its barrier, the test releasing
+# it after a full multipart restore); the ceiling only stops a broken test from
+# hanging. A slow Windows runner took longer than 5 s for that restore, so the
+# blocked read raised inside its handler and returned a text/plain 500 (#242).
+HANDOFF_TIMEOUT_SECONDS = 60
+
 OTHER_SESSION = "aaaaaaaa-3333-4333-8333-333333333333"
 MISSING_RUN = "bbbbbbbb-4444-4444-8444-444444444444"
 
@@ -920,7 +927,7 @@ class HistoryAPITests(ImageAPITestCase):
             if armed and func.__name__ == "project":
                 armed = False
                 reached.set()
-                await asyncio.wait_for(release.wait(), 5)
+                await asyncio.wait_for(release.wait(), HANDOFF_TIMEOUT_SECONDS)
             return await executor(func, *args, **kwargs)
 
         def selected_connection():
@@ -928,7 +935,7 @@ class HistoryAPITests(ImageAPITestCase):
             if armed:
                 armed = False
                 loop.call_soon_threadsafe(reached.set)
-                if not release.wait(5):
+                if not release.wait(HANDOFF_TIMEOUT_SECONDS):
                     raise AssertionError("connection-acquisition barrier timed out")
             return connection()
 
@@ -992,7 +999,7 @@ class HistoryAPITests(ImageAPITestCase):
             )
             pending = asyncio.create_task(read)
             try:
-                await asyncio.wait_for(reached.wait(), 5)
+                await asyncio.wait_for(reached.wait(), HANDOFF_TIMEOUT_SECONDS)
                 form = FormData()
                 form.add_field("database_file", upload, filename="upload.db")
                 restored = await self.client.post("/prompt_manager/restore", data=form)
@@ -1001,7 +1008,9 @@ class HistoryAPITests(ImageAPITestCase):
                 self.assertEqual(len(old_db.model._connections), 0)
             finally:
                 release.set()
-                response, body = await asyncio.wait_for(pending, 5)
+                response, body = await asyncio.wait_for(
+                    pending, HANDOFF_TIMEOUT_SECONDS
+                )
 
         if failure is not None:
             self.assert_error(

tests/test_database_concurrency.py

@@ -7,13 +7,16 @@ another thread's half-done write.
 """
 
 import os
+import sqlite3
 import sys
 import tempfile
 import threading
 import unittest
+from unittest.mock import patch
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+from database import models
 from database.operations import PromptDatabase
 from utils.hashing import generate_prompt_hash
 
@@ -126,5 +129,75 @@ class TestConcurrentWrites(ConcurrencyTestCase):
         self.assertEqual(prompt["run_count"], THREADS * CALLS_PER_THREAD)
 
 
+def connection_settings(conn):
+    return tuple(
+        conn.execute(f"PRAGMA {name}").fetchone()[0]
+        for name in ("journal_mode", "synchronous", "busy_timeout", "foreign_keys")
+    )
+
+
+# WAL journal, synchronous=NORMAL (1), the shared busy timeout, foreign keys on.
+MANAGED_SETTINGS = ("wal", 1, models.BUSY_TIMEOUT_SECONDS * 1000, 1)
+
+
+class TestConnectionConfiguration(ConcurrencyTestCase):
+    """Every managed connection shares one locking and durability setup.
+
+    With synchronous=FULL each WAL commit waits for a disk flush while it holds
+    SQLite's single write lock. On slow disks (Windows runners) the other
+    writers' polling busy handlers then starve past the busy timeout and fail
+    with "database is locked" (#242). In WAL mode synchronous=NORMAL keeps the
+    database consistent after an application crash and only flushes at
+    checkpoints, so commits stop holding the lock across a flush.
+    """
+
+    def test_every_thread_connection_uses_the_managed_settings(self):
+        seen = []
+
+        def grab(_index):
+            seen.append(connection_settings(self.db.model.get_connection()))
+
+        self.assertEqual(self._run_threads(grab), [])
+        seen.append(connection_settings(self.db.model.get_connection()))
+        self.db.close_all()
+        seen.append(connection_settings(self.db.model.get_connection()))
+        self.assertEqual(seen, [MANAGED_SETTINGS] * (THREADS + 2))
+
+    def test_schema_connection_uses_the_managed_settings(self):
+        fd, path = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        seen = []
+        create_tables = models.PromptModel._create_tables
+
+        def record(model, conn):
+            seen.append(connection_settings(conn))
+            return create_tables(model, conn)
+
+        fresh = None
+        try:
+            with patch.object(
+                models.PromptModel, "_create_tables", autospec=True, side_effect=record
+            ):
+                fresh = PromptDatabase(path)
+        finally:
+            if fresh is not None:
+                fresh.close_all()
+            for suffix in ("", "-wal", "-shm"):
+                if os.path.exists(path + suffix):
+                    os.unlink(path + suffix)
+        self.assertEqual(seen, [MANAGED_SETTINGS])
+
+    def test_normal_sync_is_only_used_with_a_wal_journal(self):
+        # An in-memory database cannot use WAL; keep SQLite's FULL default there.
+        conn = sqlite3.connect(":memory:")
+        try:
+            models.configure_connection(conn)
+            self.assertEqual(
+                connection_settings(conn), ("memory", 2, *MANAGED_SETTINGS[2:])
+            )
+        finally:
+            conn.close()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_execution_context_database.py

@@ -11,6 +11,7 @@ import textwrap
 from unittest.mock import patch
 import unittest
 
+from database import execution_contexts
 from database.execution_history import HistoryEpochChanged
 from database.execution_context_schema import context_schema_error
 from database.execution_contexts import LIVE_HINT_PROMPT_IDS_SQL
@@ -161,12 +162,23 @@ class ContextDatabaseTests(unittest.TestCase):
         for start in starts:
             db.write_execution_start(start)
 
+        hold = history_tests.FirstWriterHold()
+        write_capture = execution_contexts._write_capture
+
+        def held_open(*args, **kwargs):
+            hold()
+            return write_capture(*args, **kwargs)
+
         def write(start):
             return db.write_execution_contexts(
                 start.run_id, start.storage_epoch, bundle()
             )
 
-        with ThreadPoolExecutor(max_workers=4) as pool:
+        with (
+            patch.object(execution_contexts, "_write_capture", side_effect=held_open),
+            history_tests.queued_writer_budget(),
+            ThreadPoolExecutor(max_workers=4) as pool,
+        ):
             self.assertEqual(list(pool.map(write, starts)), [True] * 8)
         self.assertEqual(self.counts(db), (8, 1, 2, 8))
 

tests/test_execution_history_database.py

@@ -2,17 +2,58 @@
 
 import sqlite3
 import tempfile
+import threading
+import time
 import unittest
 from contextlib import closing
+from unittest.mock import patch
 from dataclasses import replace
 from pathlib import Path
 from types import SimpleNamespace
 from uuid import UUID
 
-from database.models import PromptModel
+from database import execution_history
+from database.models import BUSY_TIMEOUT_SECONDS, PromptModel
 from database.operations import PromptDatabase
 from utils.execution_history_events import make_finish, make_start
 
+# Longer than the 250 ms bounded writer wait, as a slow Windows commit can be.
+WRITER_HOLD_SECONDS = 0.5
+
+
+class FirstWriterHold:
+    """Keep the first write transaction that calls it open for a while.
+
+    Concurrent writer tests use it so the other peers really queue behind an
+    open BEGIN IMMEDIATE, instead of depending on how fast the host commits.
+    """
+
+    def __init__(self, seconds=WRITER_HOLD_SECONDS):
+        self.seconds = seconds
+        self._lock = threading.Lock()
+        self._held = False
+
+    def __call__(self):
+        with self._lock:
+            hold, self._held = not self._held, True
+        if hold:
+            time.sleep(self.seconds)
+
+
+def queued_writer_budget():
+    """Let concurrent test peers wait like ordinary writers do.
+
+    Production history writes give up after EXECUTION_BUSY_TIMEOUT_MS so they
+    never stall ComfyUI, which the bounded writer timeout test covers. Tests
+    that assert every concurrent peer succeeds are about serialization and
+    first-capture semantics, so they must not also depend on how quickly the
+    host commits the writers queued ahead of them.
+    """
+    return patch.object(
+        execution_history, "EXECUTION_BUSY_TIMEOUT_MS", BUSY_TIMEOUT_SECONDS * 1000
+    )
+
+
 SESSION = "11111111-1111-4111-8111-111111111111"
 JOB = "22222222-2222-4222-8222-222222222222"
 

tests/test_execution_output_database.py

@@ -144,14 +144,25 @@ class OutputDatabaseTests(unittest.TestCase):
         finish = self.finish(start)
         peers = [self.database() for _ in range(4)]
         barrier = Barrier(len(peers))
+        hold = history_tests.FirstWriterHold()
+
+        @contextmanager
+        def held_open():
+            hold()
+            yield True
 
         def write(index):
-            barrier.wait(timeout=5)
+            barrier.wait(timeout=60)
             return peers[index].write_execution_outputs_for_finish(
-                finish, bundle("first.png" if index % 2 else "second.png")
+                finish,
+                bundle("first.png" if index % 2 else "second.png"),
+                generation_guard=held_open,
             )
 
-        with ThreadPoolExecutor(max_workers=4) as executor:
+        with (
+            history_tests.queued_writer_budget(),
+            ThreadPoolExecutor(max_workers=4) as executor,
+        ):
             self.assertEqual(list(executor.map(write, range(4))), [True] * 4)
         capture = db.get_execution_outputs(start.run_id)
         self.assertEqual(

utils/execution_history_writer.py

@@ -12,12 +12,14 @@ from typing import Callable, Iterator
 
 try:
     from ..database import PromptDatabase
+    from ..database import execution_history as history_storage
     from ..database.execution_history import (
         HistoryEpochChanged,
         subscribe_history_restore,
     )
 except ImportError:
     from database import PromptDatabase
+    from database import execution_history as history_storage
     from database.execution_history import (
         HistoryEpochChanged,
         subscribe_history_restore,
@@ -409,7 +411,10 @@ class HistoryWriter:
                             unsubscribe = subscribe_history_restore(
                                 db.model.db_path, self.invalidate
                             )
-                        db.model.get_connection().execute("PRAGMA busy_timeout=250")
+                        db.model.get_connection().execute(
+                            "PRAGMA busy_timeout="
+                            f"{int(history_storage.EXECUTION_BUSY_TIMEOUT_MS)}"
+                        )
                         epoch = db.get_execution_epoch()
                         with self._lock:
                             if not self._closing and generation == self._generation:
@@ -429,7 +434,10 @@ class HistoryWriter:
                         command = command.command
                     try:
                         # Restore may have reopened this thread's managed connection.
-                        db.model.get_connection().execute("PRAGMA busy_timeout=250")
+                        db.model.get_connection().execute(
+                            "PRAGMA busy_timeout="
+                            f"{int(history_storage.EXECUTION_BUSY_TIMEOUT_MS)}"
+                        )
                         if isinstance(command, RunStart):
                             db.write_execution_start(command)
                         else: