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).
This commit is contained in:
@@ -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,22 @@ 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),
|
||||
ThreadPoolExecutor(max_workers=4) as pool,
|
||||
):
|
||||
self.assertEqual(list(pool.map(write, starts)), [True] * 8)
|
||||
self.assertEqual(self.counts(db), (8, 1, 2, 8))
|
||||
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from contextlib import closing
|
||||
from dataclasses import replace
|
||||
@@ -13,6 +15,29 @@ from database.models import 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)
|
||||
|
||||
|
||||
SESSION = "11111111-1111-4111-8111-111111111111"
|
||||
JOB = "22222222-2222-4222-8222-222222222222"
|
||||
|
||||
|
||||
@@ -144,11 +144,19 @@ 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:
|
||||
|
||||
Reference in New Issue
Block a user