Vito
4ef236aaf7
* feat(history): define bounded output reference contracts * feat(database): migrate execution output evidence to schema v3 * feat(history): persist immutable run output references * test(history): isolate mocked node gallery monitor * feat(history): capture bounded output descriptors with shared leases * fix(history): authenticate payload owners and release failed captures * feat(history): attach output evidence to exact finalized attempts * test(history): verify run output evidence on real ComfyUI * fix(history): release idle worker payload references * docs(history): clarify defensive lease cleanup * test(history): align output fixtures with native path styles * test(history): separate restore completion from writer drain * test(history): bound batch drains by progress and total time
1015 lines
40 KiB
Python
1015 lines
40 KiB
Python
"""Real-thread, disposable-database coverage for the bounded history writer."""
|
|
|
|
from dataclasses import FrozenInstanceError, replace
|
|
from contextlib import nullcontext
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from uuid import uuid4
|
|
|
|
from tests.history_waits import drain_writer
|
|
from database import PromptDatabase
|
|
from utils.execution_history_events import (
|
|
LifecycleEvent,
|
|
RunFinish,
|
|
TerminalEvidence,
|
|
make_finish,
|
|
make_start,
|
|
)
|
|
from utils.execution_history_writer import HistoryWriter
|
|
|
|
|
|
class ManualClock:
|
|
def __init__(self):
|
|
self.value = 0.0
|
|
self.lock = threading.Lock()
|
|
|
|
def __call__(self):
|
|
with self.lock:
|
|
return self.value
|
|
|
|
def advance(self, seconds):
|
|
with self.lock:
|
|
self.value += seconds
|
|
|
|
|
|
class HistoryWriterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
logger = logging.getLogger("prompt_manager.execution_history_writer")
|
|
handlers, propagate, level = logger.handlers, logger.propagate, logger.level
|
|
logger.handlers = [logging.NullHandler()]
|
|
logger.propagate = False
|
|
logger.setLevel(logging.WARNING)
|
|
|
|
def restore_logging():
|
|
logger.handlers, logger.propagate, logger.level = handlers, propagate, level
|
|
|
|
self.addCleanup(restore_logging)
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.db = PromptDatabase(os.path.join(self.directory.name, "history.db"))
|
|
self.addCleanup(self.db.model.reset_schema_cache)
|
|
self.addCleanup(self.db.model.close_all)
|
|
self.clock = ManualClock()
|
|
self.session = str(uuid4())
|
|
self.job = str(uuid4())
|
|
self.releases = []
|
|
self.writers = []
|
|
self.addCleanup(self.cleanup_workers)
|
|
|
|
def cleanup_workers(self):
|
|
for release in self.releases:
|
|
release.set()
|
|
for writer in self.writers:
|
|
self.assertTrue(writer.close(3), "test-created worker leaked")
|
|
|
|
def wait_for(self, predicate):
|
|
deadline = time.monotonic() + 3
|
|
while time.monotonic() < deadline:
|
|
if predicate():
|
|
return
|
|
threading.Event().wait(0.005)
|
|
self.fail("finite deadline expired waiting for worker behavior")
|
|
|
|
def writer(self, factory=None, ready=True):
|
|
writer = HistoryWriter(factory or (lambda: self.db), clock=self.clock)
|
|
self.writers.append(writer)
|
|
writer.start()
|
|
if ready:
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
return writer
|
|
|
|
def start_command(self, item=1, epoch=None):
|
|
return make_start(
|
|
self.session,
|
|
((0, self.job), item),
|
|
epoch or self.db.get_execution_epoch(),
|
|
100,
|
|
)
|
|
|
|
def finish_command(self, start):
|
|
return make_finish(
|
|
start.session_id,
|
|
start.queue_item_id,
|
|
start.storage_epoch,
|
|
SimpleNamespace(messages=[("execution_success", {"prompt_id": self.job})]),
|
|
200,
|
|
)
|
|
|
|
def block(self):
|
|
entered, release = threading.Event(), threading.Event()
|
|
self.releases.append(release)
|
|
return entered, release
|
|
|
|
def sealed(self, writer, value="exact text", generation=None):
|
|
from utils.execution_context_budget import SealedContext
|
|
from utils.execution_contexts import (
|
|
ContextBundle,
|
|
RecipeContext,
|
|
PromptSnapshot,
|
|
TextEvidence,
|
|
bundle_payload_size,
|
|
)
|
|
|
|
field = TextEvidence(value, "recorded", "runtime_return", None)
|
|
snapshot = PromptSnapshot(field, field, field, field)
|
|
bundle = ContextBundle(
|
|
(RecipeContext("1", snapshot, snapshot, "inferred"),),
|
|
"complete",
|
|
"captured",
|
|
)
|
|
lease = writer.context_budget.reserve(bundle_payload_size(bundle))
|
|
return SealedContext(
|
|
bundle,
|
|
lease,
|
|
"complete",
|
|
"captured",
|
|
writer.status().capture_generation if generation is None else generation,
|
|
)
|
|
|
|
def test_context_success_releases_and_legacy_finish_preserves_capture(self):
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
sealed = self.sealed(writer)
|
|
self.assertTrue(writer.submit(start))
|
|
self.assertTrue(writer.submit_finalization(self.finish_command(start), sealed))
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "complete"
|
|
)
|
|
self.assertTrue(writer.submit(self.finish_command(start)))
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "complete"
|
|
)
|
|
|
|
def test_context_saturation_keeps_lifecycle_and_status(self):
|
|
from utils.execution_context_budget import SealedContext
|
|
from utils.execution_history_events import encode_command
|
|
|
|
writer = self.writer()
|
|
lease = writer.context_budget.reserve(8388608)
|
|
self.assertIsNone(writer.context_budget.reserve(1))
|
|
start = self.start_command()
|
|
self.assertTrue(writer.submit(start))
|
|
finish = self.finish_command(start)
|
|
self.assertLessEqual(len(encode_command(finish)), 2048)
|
|
self.assertTrue(
|
|
writer.submit_finalization(
|
|
finish, SealedContext(None, None, "unavailable", "budget_exhausted")
|
|
)
|
|
)
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["reason"], "budget_exhausted"
|
|
)
|
|
lease.release()
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
|
|
def test_outcome_survives_context_transaction_failure(self):
|
|
for error, ready in (
|
|
(ValueError("invalid evidence"), True),
|
|
(sqlite3.OperationalError("unavailable"), False),
|
|
):
|
|
with self.subTest(error=type(error)):
|
|
writer = self.writer()
|
|
start = self.start_command(10 if ready else 11)
|
|
self.assertTrue(writer.submit(start))
|
|
sealed = self.sealed(writer)
|
|
entered, release = self.block()
|
|
|
|
def fail_context(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(3)
|
|
raise error
|
|
|
|
with patch.object(
|
|
self.db, "write_execution_contexts_for_finish", fail_context
|
|
):
|
|
self.assertTrue(
|
|
writer.submit_finalization(self.finish_command(start), sealed)
|
|
)
|
|
self.assertTrue(entered.wait(1))
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertGreater(writer.context_budget.charged_bytes, 0)
|
|
release.set()
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertEqual(writer.status().accepting, ready)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
if not ready:
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"],
|
|
"not_captured",
|
|
)
|
|
self.assertTrue(writer.close())
|
|
|
|
def test_same_epoch_recovery_changes_capture_generation(self):
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
old_generation = writer.status().capture_generation
|
|
sealed = self.sealed(writer)
|
|
writer.invalidate()
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertEqual(writer.status().epoch, start.storage_epoch)
|
|
self.assertGreater(writer.status().capture_generation, old_generation)
|
|
self.assertTrue(writer.submit(start))
|
|
self.assertTrue(writer.submit_finalization(self.finish_command(start), sealed))
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
self.assertEqual(self.db.get_execution_contexts(start.run_id)["contexts"], [])
|
|
|
|
def test_restore_between_transactions_rejects_context(self):
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
self.db.write_execution_start(start)
|
|
backup = os.path.join(self.directory.name, "between.db")
|
|
self.assertTrue(self.db.model.backup_database(backup))
|
|
original = self.db.write_execution_finish
|
|
restore_started = threading.Event()
|
|
restore_finished = threading.Event()
|
|
restore_errors = []
|
|
|
|
def finish(command):
|
|
try:
|
|
result = original(command)
|
|
restore_started.set()
|
|
self.db.model.restore_from_file(backup)
|
|
return result
|
|
except Exception as error:
|
|
restore_errors.append(error)
|
|
raise
|
|
finally:
|
|
restore_finished.set()
|
|
|
|
with patch.object(self.db, "write_execution_finish", finish):
|
|
self.assertTrue(
|
|
writer.submit_finalization(
|
|
self.finish_command(start), self.sealed(writer)
|
|
)
|
|
)
|
|
# Disk restore must finish before the worker's short drain deadline.
|
|
self.assertTrue(
|
|
restore_finished.wait(30),
|
|
"finish/restore operation did not complete: "
|
|
f"restore_started={restore_started.is_set()}, {writer.status()}",
|
|
)
|
|
self.assertEqual(restore_errors, [])
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertNotEqual(self.db.get_execution_epoch(), start.storage_epoch)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
|
|
)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
self.assertFalse(writer.status().accepting)
|
|
|
|
def test_reject_discard_and_bounded_close_release_leases(self):
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
sealed = self.sealed(writer)
|
|
self.assertFalse(writer.submit_finalization(object(), sealed))
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
entered, release = self.block()
|
|
original = self.db.write_execution_start
|
|
|
|
def write(command):
|
|
entered.set()
|
|
release.wait(3)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", write):
|
|
self.assertTrue(writer.submit(start))
|
|
self.assertTrue(entered.wait(1))
|
|
for _ in range(1023):
|
|
self.assertTrue(
|
|
writer.submit_finalization(
|
|
self.finish_command(start), self.sealed(writer)
|
|
)
|
|
)
|
|
self.assertEqual(writer.status().pending, 1024)
|
|
self.assertFalse(
|
|
writer.submit_finalization(
|
|
self.finish_command(start), self.sealed(writer)
|
|
)
|
|
)
|
|
self.assertFalse(writer.close(0.01))
|
|
release.set()
|
|
self.wait_for(lambda: not writer.status().alive)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
|
|
)
|
|
|
|
def test_bounded_close_keeps_inflight_charge_and_releases_queued(self):
|
|
self.check_bounded_close_payloads()
|
|
|
|
def test_early_join_return_still_waits_for_deadline_and_releases_queued(self):
|
|
self.check_bounded_close_payloads(early_join=True)
|
|
|
|
def check_bounded_close_payloads(self, early_join=False):
|
|
from utils.execution_context_budget import ContextBudget
|
|
|
|
pool = ContextBudget(1000)
|
|
writer = HistoryWriter(lambda: self.db, clock=self.clock, context_budget=pool)
|
|
self.writers.append(writer)
|
|
writer.start()
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
start = self.start_command()
|
|
self.db.write_execution_start(start)
|
|
entered, release = self.block()
|
|
original = self.db.write_execution_contexts_for_finish
|
|
first = self.sealed(writer)
|
|
charge = first.lease.byte_count
|
|
|
|
def write(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(3)
|
|
return original(*args, **kwargs)
|
|
|
|
with patch.object(self.db, "write_execution_contexts_for_finish", write):
|
|
self.assertTrue(
|
|
writer.submit_finalization(self.finish_command(start), first)
|
|
)
|
|
self.assertTrue(entered.wait(1))
|
|
self.assertTrue(
|
|
writer.submit_finalization(
|
|
self.finish_command(start), self.sealed(writer)
|
|
)
|
|
)
|
|
native_join = writer._thread.join
|
|
close_clock = ManualClock()
|
|
|
|
def return_early(timeout):
|
|
# Keep the real worker blocked while simulating an early native join.
|
|
self.assertTrue(writer.status().alive)
|
|
self.assertFalse(writer.submit(self.finish_command(start)))
|
|
native_join(0)
|
|
close_clock.advance(min(timeout, 0.004))
|
|
|
|
def yield_wait(delay):
|
|
self.assertGreater(delay, 0)
|
|
self.assertLessEqual(delay, 0.001)
|
|
close_clock.advance(delay)
|
|
|
|
clock = (
|
|
patch(
|
|
"utils.execution_history_writer.time",
|
|
SimpleNamespace(monotonic=close_clock, sleep=yield_wait),
|
|
)
|
|
if early_join
|
|
else nullcontext()
|
|
)
|
|
join = (
|
|
patch.object(writer._thread, "join", return_early)
|
|
if early_join
|
|
else nullcontext()
|
|
)
|
|
with clock, join:
|
|
monotonic = close_clock if early_join else time.monotonic
|
|
before = monotonic()
|
|
self.assertFalse(writer.close(0.02))
|
|
elapsed = monotonic() - before
|
|
self.assertGreaterEqual(elapsed, 0.02)
|
|
self.assertLess(elapsed, 0.2)
|
|
self.assertEqual(pool.charged_bytes, charge)
|
|
replacement = HistoryWriter(lambda: self.db, context_budget=pool)
|
|
self.writers.append(replacement)
|
|
self.assertIsNone(replacement.context_budget.reserve(1000))
|
|
release.set()
|
|
self.wait_for(lambda: not writer.status().alive)
|
|
self.assertEqual(pool.charged_bytes, 0)
|
|
|
|
def test_close_drain_exits_without_sleeping_after_worker_join(self):
|
|
writer = self.writer()
|
|
clock = ManualClock()
|
|
native_join = writer._thread.join
|
|
|
|
def join(timeout):
|
|
native_join(3)
|
|
self.assertFalse(writer._thread.is_alive())
|
|
clock.advance(0.001)
|
|
|
|
def sleep(delay):
|
|
self.fail("Close slept after the worker exited")
|
|
|
|
with (
|
|
patch(
|
|
"utils.execution_history_writer.time",
|
|
SimpleNamespace(monotonic=clock, sleep=sleep),
|
|
),
|
|
patch.object(writer._thread, "join", join),
|
|
):
|
|
self.assertTrue(writer.close(0.02))
|
|
self.assertLess(clock(), 0.02)
|
|
self.assertFalse(writer.status().alive)
|
|
|
|
def test_same_epoch_invalidation_between_transactions_drops_payload(self):
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
self.db.write_execution_start(start)
|
|
original = self.db.write_execution_finish
|
|
|
|
def finish(command):
|
|
result = original(command)
|
|
writer.invalidate()
|
|
return result
|
|
|
|
with patch.object(self.db, "write_execution_finish", finish):
|
|
self.assertTrue(
|
|
writer.submit_finalization(
|
|
self.finish_command(start), self.sealed(writer)
|
|
)
|
|
)
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
|
|
)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
|
|
)
|
|
|
|
def test_invalid_seals_release_and_keep_valid_outcome(self):
|
|
from utils.execution_context_budget import ContextBudget
|
|
|
|
writer = self.writer()
|
|
cases = [
|
|
dict(state="private text" * 100),
|
|
dict(reason="arbitrary_exception"),
|
|
dict(capture_generation=True),
|
|
dict(capture_generation=-1),
|
|
dict(capture_generation=None),
|
|
dict(lease=None),
|
|
]
|
|
for item, changes in enumerate(cases, 20):
|
|
start = self.start_command(item)
|
|
self.db.write_execution_start(start)
|
|
sealed = self.sealed(writer)
|
|
malformed = replace(sealed, **changes)
|
|
if changes == {"lease": None}:
|
|
sealed.lease.release()
|
|
self.assertTrue(
|
|
writer.submit_finalization(self.finish_command(start), malformed)
|
|
)
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["reason"],
|
|
"invalid_evidence",
|
|
)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
foreign = ContextBudget(1000)
|
|
sealed = self.sealed(writer)
|
|
sealed.lease.release()
|
|
sealed = replace(sealed, lease=foreign.reserve(1000))
|
|
start = self.start_command(30)
|
|
self.db.write_execution_start(start)
|
|
self.assertTrue(writer.submit_finalization(self.finish_command(start), sealed))
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(foreign.charged_bytes, 0)
|
|
self.assertEqual(self.db.get_execution_contexts(start.run_id)["contexts"], [])
|
|
|
|
def test_custom_run_identity_receives_finalized_context(self):
|
|
writer = self.writer()
|
|
start = replace(self.start_command(), run_id=str(uuid4()))
|
|
self.assertTrue(writer.submit(start))
|
|
self.assertTrue(
|
|
writer.submit_finalization(self.finish_command(start), self.sealed(writer))
|
|
)
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "complete"
|
|
)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
|
|
def test_invalidation_during_context_preparation_fences_actual_transaction(self):
|
|
import database.execution_contexts as contexts_db
|
|
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
self.db.write_execution_start(start)
|
|
entered, release = self.block()
|
|
original = contexts_db._prepare
|
|
|
|
def prepare(bundle):
|
|
prepared = original(bundle)
|
|
entered.set()
|
|
release.wait(3)
|
|
return prepared
|
|
|
|
with patch.object(contexts_db, "_prepare", prepare):
|
|
self.assertTrue(
|
|
writer.submit_finalization(
|
|
self.finish_command(start), self.sealed(writer)
|
|
)
|
|
)
|
|
self.assertTrue(entered.wait(1))
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
generation = writer.status().capture_generation
|
|
writer.invalidate()
|
|
self.assertGreater(writer.status().capture_generation, generation)
|
|
release.set()
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
|
|
)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertEqual(writer.status().epoch, start.storage_epoch)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
|
|
)
|
|
|
|
def test_context_commit_serializes_invalidation_without_blocking_queue_or_close(
|
|
self,
|
|
):
|
|
import database.execution_contexts as contexts_db
|
|
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
self.db.write_execution_start(start)
|
|
entered, release = self.block()
|
|
invalidating, invalidated = threading.Event(), threading.Event()
|
|
original = contexts_db._write_capture
|
|
|
|
def capture(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(3)
|
|
return original(*args, **kwargs)
|
|
|
|
def invalidate():
|
|
invalidating.set()
|
|
writer.invalidate()
|
|
invalidated.set()
|
|
|
|
with patch.object(contexts_db, "_write_capture", capture):
|
|
self.assertTrue(
|
|
writer.submit_finalization(
|
|
self.finish_command(start), self.sealed(writer)
|
|
)
|
|
)
|
|
self.assertTrue(entered.wait(1))
|
|
invalidator = threading.Thread(target=invalidate)
|
|
invalidator.start()
|
|
try:
|
|
self.assertTrue(invalidating.wait(1))
|
|
self.assertFalse(invalidated.wait(0.03))
|
|
before = time.monotonic()
|
|
self.assertTrue(writer.status().accepting)
|
|
self.assertTrue(writer.submit(self.finish_command(start)))
|
|
self.assertLess(time.monotonic() - before, 0.1)
|
|
before = time.monotonic()
|
|
self.assertFalse(writer.close(0))
|
|
self.assertLess(time.monotonic() - before, 0.1)
|
|
self.assertGreater(writer.context_budget.charged_bytes, 0)
|
|
self.assertEqual(writer.status().pending, 1)
|
|
finally:
|
|
release.set()
|
|
invalidator.join(3)
|
|
self.assertFalse(invalidator.is_alive())
|
|
self.assertTrue(invalidated.is_set())
|
|
self.wait_for(lambda: not writer.status().alive)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
|
|
)
|
|
self.assertEqual(
|
|
self.db.get_execution_contexts(start.run_id)["state"], "complete"
|
|
)
|
|
self.assertEqual(writer.context_budget.charged_bytes, 0)
|
|
|
|
def test_fifo_single_worker_and_thread_local_connection_cleanup(self):
|
|
caller_conn = self.db.model.get_connection()
|
|
threads, timeouts = [], []
|
|
original = self.db.write_execution_start
|
|
|
|
def write(command):
|
|
threads.append(threading.get_ident())
|
|
timeouts.append(
|
|
self.db.model.get_connection()
|
|
.execute("PRAGMA busy_timeout")
|
|
.fetchone()[0]
|
|
)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", write):
|
|
writer = self.writer()
|
|
writer.start()
|
|
start = self.start_command()
|
|
self.assertTrue(writer.submit(start))
|
|
self.assertTrue(writer.submit(self.finish_command(start)))
|
|
self.wait_for(
|
|
lambda: (self.db.get_execution_run(start.run_id) or {}).get("outcome")
|
|
== "succeeded"
|
|
)
|
|
self.assertTrue(writer.submit(self.start_command(2)))
|
|
self.assertTrue(writer.close(2))
|
|
self.assertEqual(len(set(threads)), 1)
|
|
self.assertNotEqual(threads[0], threading.get_ident())
|
|
self.assertEqual(timeouts, [250, 250])
|
|
self.assertFalse(writer.status().alive)
|
|
self.assertFalse(writer.status().accepting)
|
|
self.assertEqual(writer.status().pending, 0)
|
|
self.assertEqual(
|
|
caller_conn.execute("PRAGMA busy_timeout").fetchone()[0], 30000
|
|
)
|
|
self.assertEqual(len(self.db.model._connections), 1)
|
|
with self.assertRaises(FrozenInstanceError):
|
|
writer.status().epoch = "changed"
|
|
self.assertFalse(writer.submit(start))
|
|
|
|
def test_blocked_initialization_never_accepts_or_blocks_producers(self):
|
|
entered, release = self.block()
|
|
|
|
def factory():
|
|
entered.set()
|
|
release.wait(3)
|
|
return self.db
|
|
|
|
writer = self.writer(factory, ready=False)
|
|
self.assertTrue(entered.wait(1))
|
|
before = time.monotonic()
|
|
self.assertFalse(writer.submit(self.start_command()))
|
|
self.assertIsNone(writer.status().epoch)
|
|
self.assertLess(time.monotonic() - before, 0.1)
|
|
self.assertIsNone(self.db.get_execution_run(self.start_command().run_id))
|
|
before = time.monotonic()
|
|
self.assertFalse(writer.close(0.03))
|
|
self.assertLess(time.monotonic() - before, 0.2)
|
|
release.set()
|
|
self.wait_for(lambda: not writer.status().alive)
|
|
self.assertFalse(writer.status().accepting)
|
|
self.assertEqual(len(self.db.model._connections), 1)
|
|
|
|
def test_saturation_bounds_all_pending_commands_and_coalesces_diagnostics(self):
|
|
entered, release = self.block()
|
|
original = self.db.write_execution_start
|
|
|
|
def write(command):
|
|
entered.set()
|
|
release.wait(3)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", write):
|
|
writer = self.writer()
|
|
command = self.start_command()
|
|
self.assertTrue(writer.submit(command))
|
|
self.assertTrue(entered.wait(1))
|
|
for _ in range(1023):
|
|
self.assertTrue(writer.submit(command))
|
|
for _ in range(100):
|
|
self.assertFalse(writer.submit(command))
|
|
self.assertEqual(writer.status().pending, 1024)
|
|
self.assertEqual(writer.status().recording_gaps, 100)
|
|
with self.assertLogs(
|
|
"prompt_manager.execution_history_writer", "WARNING"
|
|
) as logs:
|
|
release.set()
|
|
drain_writer(writer)
|
|
self.assertTrue(writer.close(2))
|
|
self.assertEqual(len(logs.records), 1)
|
|
|
|
def test_rejects_oversized_invalid_and_wrong_epoch_commands(self):
|
|
writer = self.writer()
|
|
start = self.start_command()
|
|
oversized = RunFinish(
|
|
self.session,
|
|
1,
|
|
start.storage_epoch,
|
|
200,
|
|
TerminalEvidence(
|
|
tuple(
|
|
LifecycleEvent(
|
|
"execution_error", self.job, i, "n" * 128, None, None
|
|
)
|
|
for i in range(20)
|
|
),
|
|
False,
|
|
False,
|
|
False,
|
|
),
|
|
)
|
|
for command in (
|
|
oversized,
|
|
object(),
|
|
replace(start, storage_epoch=str(uuid4())),
|
|
):
|
|
self.assertFalse(writer.submit(command))
|
|
self.assertEqual(writer.status().pending, 0)
|
|
self.assertEqual(writer.status().recording_gaps, 3)
|
|
self.assertIsNone(self.db.get_execution_run(start.run_id))
|
|
|
|
def test_failed_command_once_discards_backlog_and_waits_five_seconds(self):
|
|
entered, release = self.block()
|
|
probes, writes = [], []
|
|
original_epoch = self.db.get_execution_epoch
|
|
original_write = self.db.write_execution_start
|
|
|
|
def epoch():
|
|
probes.append(self.clock())
|
|
return original_epoch()
|
|
|
|
def write(command):
|
|
writes.append(command.queue_item_id)
|
|
if command.queue_item_id == 1:
|
|
entered.set()
|
|
release.wait(3)
|
|
raise sqlite3.OperationalError("private failure text")
|
|
return original_write(command)
|
|
|
|
start, queued, fresh = (self.start_command(i) for i in (1, 2, 3))
|
|
with (
|
|
patch.object(self.db, "get_execution_epoch", epoch),
|
|
patch.object(self.db, "write_execution_start", write),
|
|
):
|
|
writer = self.writer()
|
|
self.assertTrue(writer.submit(start))
|
|
self.assertTrue(entered.wait(1))
|
|
self.assertTrue(writer.submit(queued))
|
|
release.set()
|
|
self.wait_for(lambda: not writer.status().accepting)
|
|
self.assertFalse(writer.submit(fresh))
|
|
self.clock.advance(4.99)
|
|
threading.Event().wait(0.08)
|
|
self.assertEqual(probes, [0])
|
|
self.clock.advance(0.01)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertEqual(probes, [0, 5])
|
|
self.assertTrue(writer.submit(fresh))
|
|
self.wait_for(lambda: self.db.get_execution_run(fresh.run_id) is not None)
|
|
self.assertEqual(writes, [1, 3])
|
|
self.assertIsNone(self.db.get_execution_run(start.run_id))
|
|
self.assertIsNone(self.db.get_execution_run(queued.run_id))
|
|
|
|
def test_factory_failure_recovery_probes_are_bounded(self):
|
|
calls = []
|
|
|
|
def factory():
|
|
calls.append(self.clock())
|
|
if len(calls) < 3:
|
|
raise sqlite3.OperationalError("unavailable")
|
|
return self.db
|
|
|
|
writer = self.writer(factory, ready=False)
|
|
self.wait_for(lambda: len(calls) == 1)
|
|
self.clock.advance(4.99)
|
|
threading.Event().wait(0.08)
|
|
self.assertEqual(calls, [0])
|
|
self.clock.advance(0.01)
|
|
self.wait_for(lambda: len(calls) == 2)
|
|
self.clock.advance(4.99)
|
|
threading.Event().wait(0.08)
|
|
self.assertEqual(calls, [0, 5])
|
|
self.clock.advance(0.01)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertEqual(calls, [0, 5, 10])
|
|
command = self.start_command()
|
|
self.assertTrue(writer.submit(command))
|
|
self.wait_for(lambda: self.db.get_execution_run(command.run_id) is not None)
|
|
|
|
def test_failed_epoch_probe_closes_only_worker_connection_and_recovers(self):
|
|
command = self.start_command()
|
|
original = self.db.get_execution_epoch
|
|
failed = threading.Event()
|
|
|
|
def epoch():
|
|
if not failed.is_set():
|
|
failed.set()
|
|
raise sqlite3.DatabaseError("metadata unavailable")
|
|
return original()
|
|
|
|
with patch.object(self.db, "get_execution_epoch", epoch):
|
|
writer = self.writer(ready=False)
|
|
self.wait_for(lambda: writer.status().recording_gaps == 1)
|
|
self.assertIsNone(writer.status().epoch)
|
|
self.assertFalse(writer.submit(command))
|
|
self.assertEqual(len(self.db.model._connections), 1)
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertTrue(writer.submit(command))
|
|
self.wait_for(lambda: self.db.get_execution_run(command.run_id) is not None)
|
|
|
|
def test_invalidation_during_epoch_probe_cannot_publish_stale_readiness(self):
|
|
entered, release = self.block()
|
|
original = self.db.get_execution_epoch
|
|
stale_epoch = original()
|
|
fresh_epoch = str(uuid4())
|
|
|
|
def epoch():
|
|
result = original()
|
|
if not entered.is_set():
|
|
entered.set()
|
|
release.wait(3)
|
|
return result
|
|
|
|
with patch.object(self.db, "get_execution_epoch", epoch):
|
|
writer = self.writer(ready=False)
|
|
self.assertTrue(entered.wait(1))
|
|
with self.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"UPDATE execution_history_metadata SET storage_epoch=?",
|
|
(fresh_epoch,),
|
|
)
|
|
writer.invalidate()
|
|
release.set()
|
|
threading.Event().wait(0.08)
|
|
self.assertFalse(writer.status().accepting)
|
|
self.assertIsNone(writer.status().epoch)
|
|
self.assertFalse(writer.submit(self.start_command(epoch=stale_epoch)))
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().epoch == fresh_epoch)
|
|
command = self.start_command(epoch=fresh_epoch)
|
|
self.assertTrue(writer.submit(command))
|
|
self.wait_for(lambda: self.db.get_execution_run(command.run_id) is not None)
|
|
|
|
def test_held_sqlite_lock_does_not_block_submit_and_is_not_retried(self):
|
|
writer = self.writer()
|
|
command = self.start_command()
|
|
lock = sqlite3.connect(self.db.model.db_path)
|
|
self.addCleanup(lock.close)
|
|
lock.execute("BEGIN IMMEDIATE")
|
|
before = time.monotonic()
|
|
self.assertTrue(writer.submit(command))
|
|
self.assertLess(time.monotonic() - before, 0.1)
|
|
self.wait_for(lambda: not writer.status().accepting)
|
|
lock.rollback()
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertIsNone(self.db.get_execution_run(command.run_id))
|
|
fresh = self.start_command(2)
|
|
self.assertTrue(writer.submit(fresh))
|
|
self.wait_for(lambda: self.db.get_execution_run(fresh.run_id) is not None)
|
|
|
|
def test_external_epoch_rotation_drops_stale_commands_then_records_fresh(self):
|
|
writer = self.writer()
|
|
command = self.start_command()
|
|
new_epoch = str(uuid4())
|
|
with self.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"UPDATE execution_history_metadata SET storage_epoch=?", (new_epoch,)
|
|
)
|
|
self.assertTrue(writer.submit(command))
|
|
self.wait_for(lambda: not writer.status().accepting)
|
|
self.assertIsNone(writer.status().epoch)
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().epoch == new_epoch)
|
|
self.assertFalse(writer.submit(command))
|
|
fresh = self.start_command(2)
|
|
self.assertTrue(writer.submit(fresh))
|
|
self.wait_for(lambda: self.db.get_execution_run(fresh.run_id) is not None)
|
|
self.assertIsNone(self.db.get_execution_run(command.run_id))
|
|
|
|
def test_restore_invalidates_queued_and_inflight_commands(self):
|
|
backup = os.path.join(self.directory.name, "backup.db")
|
|
command = self.start_command()
|
|
self.db.write_execution_start(command)
|
|
self.assertTrue(self.db.model.backup_database(backup))
|
|
writer = self.writer()
|
|
entered, release = self.block()
|
|
original = self.db.write_execution_start
|
|
|
|
def write(command):
|
|
entered.set()
|
|
release.wait(3)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", write):
|
|
self.assertTrue(writer.submit(command))
|
|
self.assertTrue(entered.wait(1))
|
|
self.assertTrue(writer.submit(self.finish_command(command)))
|
|
self.db.model.restore_from_file(backup)
|
|
self.assertFalse(writer.status().accepting)
|
|
self.assertIsNone(writer.status().epoch)
|
|
self.assertFalse(writer.submit(command))
|
|
release.set()
|
|
self.wait_for(lambda: writer.status().pending == 0)
|
|
self.clock.advance(5)
|
|
self.wait_for(lambda: writer.status().accepting)
|
|
self.assertNotEqual(writer.status().epoch, command.storage_epoch)
|
|
self.assertEqual(
|
|
self.db.get_execution_run(command.run_id)["outcome"], "observed"
|
|
)
|
|
fresh = self.start_command(2)
|
|
timeouts = []
|
|
|
|
def write_fresh(command):
|
|
timeouts.append(
|
|
self.db.model.get_connection()
|
|
.execute("PRAGMA busy_timeout")
|
|
.fetchone()[0]
|
|
)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", write_fresh):
|
|
self.assertTrue(writer.submit(fresh))
|
|
self.assertTrue(writer.submit(self.finish_command(fresh)))
|
|
self.wait_for(
|
|
lambda: (self.db.get_execution_run(fresh.run_id) or {}).get("outcome")
|
|
== "succeeded"
|
|
)
|
|
self.assertEqual(timeouts, [250])
|
|
|
|
def test_bounded_close_discards_backlog_and_disposes_after_blocked_write(self):
|
|
entered, release = self.block()
|
|
original = self.db.write_execution_start
|
|
|
|
def write(command):
|
|
entered.set()
|
|
release.wait(3)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", write):
|
|
writer = self.writer()
|
|
first, queued = self.start_command(), self.start_command(2)
|
|
self.assertTrue(writer.submit(first))
|
|
self.assertTrue(entered.wait(1))
|
|
self.assertTrue(writer.submit(queued))
|
|
before = time.monotonic()
|
|
self.assertFalse(writer.close(0.03))
|
|
self.assertLess(time.monotonic() - before, 0.2)
|
|
release.set()
|
|
self.wait_for(lambda: not writer.status().alive)
|
|
self.assertIsNone(self.db.get_execution_run(queued.run_id))
|
|
self.assertEqual(writer.status().pending, 0)
|
|
self.assertEqual(len(self.db.model._connections), 1)
|
|
|
|
def test_close_before_start_is_terminal(self):
|
|
writer = HistoryWriter(lambda: self.fail("closed writer initialized database"))
|
|
self.assertTrue(writer.close())
|
|
writer.start()
|
|
self.assertFalse(writer.status().alive)
|
|
self.assertFalse(writer.submit(self.start_command()))
|
|
|
|
def test_external_capture_gaps_are_nonblocking_and_saturate_all_rejections(self):
|
|
writer = HistoryWriter(lambda: self.fail("producer accessed database"))
|
|
self.assertTrue(callable(getattr(writer, "record_gap", None)))
|
|
writer.record_gap()
|
|
self.assertEqual(writer.status().recording_gaps, 1)
|
|
with writer._lock:
|
|
writer._gaps = 2**63 - 2
|
|
writer.record_gap()
|
|
writer.record_gap()
|
|
writer.submit(self.start_command())
|
|
writer.submit(object())
|
|
self.assertEqual(writer.status().recording_gaps, 2**63 - 1)
|
|
self.assertTrue(writer.close())
|
|
|
|
def test_owner_invalidation_callback_runs_outside_both_locks_and_is_contained(self):
|
|
observations = []
|
|
writer = None
|
|
|
|
def callback():
|
|
acquired = [
|
|
lock.acquire(blocking=False)
|
|
for lock in (writer._lock, writer._context_commit_gate)
|
|
]
|
|
observations.append(tuple(acquired))
|
|
for lock, owned in zip(
|
|
(writer._lock, writer._context_commit_gate), acquired
|
|
):
|
|
if owned:
|
|
lock.release()
|
|
raise RuntimeError("callback failure cannot escape")
|
|
|
|
writer = HistoryWriter(lambda: self.db, on_invalidation=callback)
|
|
self.writers.append(writer)
|
|
writer.invalidate()
|
|
writer.close(0)
|
|
self.assertEqual(observations, [(True, True), (True, True)])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|