From 94d50d15f212e0141f142530a4fc5b2dc9409e28 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Mon, 19 Oct 2026 10:24:58 -0700 Subject: [PATCH] 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. --- database/execution_history.py | 6 +++++- tests/test_execution_context_database.py | 1 + tests/test_execution_history_database.py | 18 +++++++++++++++++- tests/test_execution_output_database.py | 5 ++++- utils/execution_history_writer.py | 12 ++++++++++-- 5 files changed, 37 insertions(+), 5 deletions(-) diff --git a/database/execution_history.py b/database/execution_history.py index ff4e53c..d9c6cb6 100644 --- a/database/execution_history.py +++ b/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: diff --git a/tests/test_execution_context_database.py b/tests/test_execution_context_database.py index 64fcace..64f87cd 100644 --- a/tests/test_execution_context_database.py +++ b/tests/test_execution_context_database.py @@ -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) diff --git a/tests/test_execution_history_database.py b/tests/test_execution_history_database.py index c43a69b..eaddabe 100644 --- a/tests/test_execution_history_database.py +++ b/tests/test_execution_history_database.py @@ -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" diff --git a/tests/test_execution_output_database.py b/tests/test_execution_output_database.py index af16bfd..7cf3f7f 100644 --- a/tests/test_execution_output_database.py +++ b/tests/test_execution_output_database.py @@ -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( diff --git a/utils/execution_history_writer.py b/utils/execution_history_writer.py index 759be2c..248b49d 100644 --- a/utils/execution_history_writer.py +++ b/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: