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.
This commit is contained in:
Vito Sansevero
2026-10-19 10:24:58 -07:00
parent 06c26e0a41
commit 94d50d15f2
5 changed files with 37 additions and 5 deletions
+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:
+1
View File
@@ -176,6 +176,7 @@ class ContextDatabaseTests(unittest.TestCase):
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)
+17 -1
View File
@@ -6,12 +6,14 @@ 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
@@ -38,6 +40,20 @@ class FirstWriterHold:
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"
+4 -1
View File
@@ -159,7 +159,10 @@ class OutputDatabaseTests(unittest.TestCase):
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: