Compare commits

...
Author SHA1 Message Date
Vito Sansevero e471a47d1f docs(testing): add the concurrent SQLite test rule to the Windows preflight (#242) 2026-10-19 10:27:31 -07:00
Vito Sansevero deeacea700 test(history): wait on restore handoff events, not a 5 s clock (#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.
2026-10-19 10:27:11 -07:00
Vito Sansevero 94d50d15f2 test(history): let concurrent replay peers queue like ordinary writers (#242)
History writes deliberately give up after a bounded 250 ms wait so they
never stall ComfyUI; the existing bounded-timeout test still proves that.
The concurrent replay tests assert that every peer succeeds, which is a
serialization and first-capture property, so they now raise the budget
to the shared 30 s busy timeout while the first writer is held open.
The budget is one named constant, EXECUTION_BUSY_TIMEOUT_MS, shared by
the transaction helper and the background writer.

Before: 20/20 'database is locked' with the held writer. After: 0/20.
2026-10-19 10:24:58 -07:00
Vito Sansevero 06c26e0a41 test(history): hold the first concurrent writer open past the bounded wait (#242)
The concurrent replay tests expected every peer to succeed, but each peer
only waits 250 ms for BEGIN IMMEDIATE by design. On a slow Windows runner
three queued commits exceeded that and a peer raised 'database is
locked'. Holding the first writer's transaction open for 500 ms makes the
same failure deterministic on any host (20/20 locally).
2026-10-19 10:23:20 -07:00
Vito Sansevero ba4e958804 fix(database): stop WAL commits from holding the write lock across a flush (#242)
Every managed connection now goes through configure_connection: WAL,
foreign keys, the 30 s busy timeout and, when WAL is active,
synchronous=NORMAL. Under FULL each commit flushed to disk while it held
SQLite's only write lock; on slow Windows disks a burst of writers (a
scan beside ComfyUI saves, or the concurrency tests) starved a polling
waiter past the busy timeout and it failed with 'database is locked'.

With a 60 ms emulated flush on Linux, 8 threads x 50 save+use calls went
from 49.6 s with a 32.1 s worst call (and 'database is locked' in the
unit test) to 0.7 s with a 0.4 s worst call. NORMAL in WAL mode keeps the
database consistent after an application crash; only an OS crash or
power loss can drop the most recent commits. In-memory or non-WAL
databases keep SQLite's FULL default.
2026-10-19 10:22:17 -07:00
Vito Sansevero a574f57cd8 test(database): require identical WAL/NORMAL settings on managed connections (#242)
Concurrent writers failed with 'database is locked' on Windows runners
despite the 30 s busy timeout: every WAL commit flushed to disk while it
held the write lock, so polling waiters starved. These tests pin the
shared per-connection settings, including synchronous=NORMAL in WAL mode.
2026-10-19 10:21:18 -07:00
9 changed files with 202 additions and 22 deletions
+7
View File
@@ -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`
+5 -1
View File
@@ -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:
+25 -10
View File
@@ -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:
+13 -4
View File
@@ -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(
+73
View File
@@ -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()
+13 -1
View File
@@ -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))
+42 -1
View File
@@ -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"
+14 -3
View File
@@ -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(
+10 -2
View File
@@ -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: