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
761 lines
30 KiB
Python
761 lines
30 KiB
Python
"""Actual dequeue and completion capture through the owned SQLite worker."""
|
|
|
|
from collections import namedtuple
|
|
import logging
|
|
import gc
|
|
import contextlib
|
|
import io
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import sys
|
|
import types
|
|
import weakref
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from tests.history_waits import drain_writer
|
|
from database import PromptDatabase
|
|
from utils import usage_tracking
|
|
from utils.execution_history_writer import HistoryWriter
|
|
|
|
try:
|
|
from utils.execution_history import register_history_hook
|
|
except ImportError:
|
|
register_history_hook = None
|
|
|
|
Status = namedtuple("ExecutionStatus", "status_str completed messages")
|
|
JOB = "a0c26da3-eb26-405c-b143-c662a84c3112"
|
|
|
|
|
|
class HostQueue:
|
|
def __init__(self):
|
|
self.results = []
|
|
self.done_result = object()
|
|
self.done_calls = []
|
|
|
|
def get(self, *args, **kwargs):
|
|
return self.results.pop(0) if self.results else None
|
|
|
|
def task_done(self, *args, **kwargs):
|
|
self.done_calls.append((args, kwargs))
|
|
return self.done_result
|
|
|
|
|
|
def item(item_id, job=JOB, graph=None, extra=None, sensitive=None):
|
|
return ((0, job, graph or {}, extra or {}, [], sensitive or {}), item_id)
|
|
|
|
|
|
def status(kind="execution_success", job=JOB):
|
|
return Status("success", True, [(kind, {"prompt_id": job, "timestamp": 20})])
|
|
|
|
|
|
class HistoryHooksTests(unittest.TestCase):
|
|
def setUp(self):
|
|
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.queue = HostQueue()
|
|
self.server = SimpleNamespace(prompt_queue=self.queue)
|
|
self.registrations, self.releases = [], []
|
|
self.addCleanup(self.cleanup_registrations)
|
|
for name in ("execution_history_writer", "execution_history"):
|
|
logger = logging.getLogger("prompt_manager." + name)
|
|
previous = logger.handlers, logger.propagate, logger.level
|
|
logger.handlers, logger.propagate, logger.level = (
|
|
[logging.NullHandler()],
|
|
False,
|
|
logging.WARNING,
|
|
)
|
|
self.addCleanup(self.restore_logger, logger, previous)
|
|
|
|
def restore_logger(self, logger, previous):
|
|
logger.handlers, logger.propagate, logger.level = previous
|
|
|
|
def cleanup_registrations(self):
|
|
for registration in self.registrations:
|
|
registration.writer.invalidate()
|
|
for release in self.releases:
|
|
release.set()
|
|
for registration in self.registrations:
|
|
self.assertTrue(registration.dispose(3), "owned 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 recorded history")
|
|
|
|
def register(self, factory=None, ready=True):
|
|
self.assertTrue(
|
|
callable(register_history_hook), "history registration is absent"
|
|
)
|
|
registration = register_history_hook(self.server, factory or (lambda: self.db))
|
|
self.assertIsNotNone(registration)
|
|
self.registrations.append(registration)
|
|
if ready:
|
|
self.wait_for(lambda: registration.writer.status().accepting)
|
|
return registration
|
|
|
|
def dequeue(self, item_id=0, **kwargs):
|
|
result = item(item_id, **kwargs)
|
|
self.queue.results.append(result)
|
|
self.assertIs(self.queue.get(), result)
|
|
return result
|
|
|
|
def rows(self):
|
|
return [
|
|
dict(row)
|
|
for row in self.db.model.get_connection()
|
|
.execute("SELECT * FROM execution_runs ORDER BY queue_item_id")
|
|
.fetchall()
|
|
]
|
|
|
|
def drain(self, registration):
|
|
drain_writer(registration.writer, clock=time.monotonic)
|
|
|
|
def test_package_history_initializes_when_monitor_and_usage_setup_fail(self):
|
|
try:
|
|
from tests.test_init_package import (
|
|
fake_server_module,
|
|
load_package,
|
|
)
|
|
except ImportError:
|
|
from test_init_package import (
|
|
fake_server_module,
|
|
load_package,
|
|
)
|
|
name = "prompt_manager_pkg_history_independent"
|
|
server = fake_server_module()
|
|
server.PromptServer.instance.prompt_queue = self.queue
|
|
server.PromptServer.instance.app = SimpleNamespace(middlewares=[])
|
|
|
|
def reject_usage(_):
|
|
raise RuntimeError("usage registration unavailable")
|
|
|
|
server.PromptServer.instance.add_on_prompt_handler = reject_usage
|
|
output = os.path.join(self.directory.name, "output")
|
|
os.makedirs(output)
|
|
monitor_module = types.ModuleType(name + ".utils.image_monitor")
|
|
# Missing monitor import fails before shared database initialization.
|
|
with (
|
|
patch.dict(sys.modules, {monitor_module.__name__: monitor_module}),
|
|
contextlib.redirect_stdout(io.StringIO()),
|
|
self.assertLogs("prompt_manager.init", level="ERROR"),
|
|
):
|
|
with load_package(name, self.db.model.db_path, output, server) as (
|
|
package,
|
|
_,
|
|
):
|
|
registration = getattr(package, "_history_registration", None)
|
|
self.assertIsNotNone(
|
|
registration, "history setup depended on another hook"
|
|
)
|
|
self.assertIsNone(package._global_db)
|
|
self.wait_for(lambda: registration.writer.status().accepting)
|
|
self.dequeue(0)
|
|
self.queue.task_done(0, {}, status())
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows()[0]["outcome"], "succeeded")
|
|
|
|
def test_distinct_actual_attempts_for_reused_job_and_cached_rerun(self):
|
|
registration = self.register()
|
|
self.dequeue(0)
|
|
self.queue.task_done(0, {"outputs": {}}, status("execution_error"))
|
|
self.dequeue(1)
|
|
self.queue.task_done(item_id=1, history_result={}, status=status())
|
|
self.drain(registration)
|
|
rows = self.rows()
|
|
self.assertEqual(len(rows), 2)
|
|
self.assertEqual([r["job_id"] for r in rows], [JOB, JOB])
|
|
self.assertEqual([r["outcome"] for r in rows], ["failed", "succeeded"])
|
|
self.assertNotEqual(rows[0]["run_id"], rows[1]["run_id"])
|
|
|
|
def test_later_conflicting_completion_resolves_completed_stored_identity(self):
|
|
registration = self.register()
|
|
self.dequeue(0)
|
|
self.queue.task_done(0, {}, status())
|
|
self.drain(registration)
|
|
self.queue.task_done(0, {}, status("execution_error"))
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows()[0]["outcome"], "unknown")
|
|
self.assertEqual(self.rows()[0]["reason"], "conflicting_terminal")
|
|
|
|
def test_repeated_registration_cleanup_and_reregistration_keep_session_and_wrappers(
|
|
self,
|
|
):
|
|
before = self.queue.get
|
|
calls = []
|
|
|
|
def unrelated_before(*a, **k):
|
|
calls.append("before")
|
|
return before(*a, **k)
|
|
|
|
self.queue.get = unrelated_before
|
|
first = self.register()
|
|
self.assertIs(register_history_hook(self.server, lambda: self.db), first)
|
|
observed = self.queue.get
|
|
|
|
def unrelated_after(*a, **k):
|
|
calls.append("after")
|
|
return observed(*a, **k)
|
|
|
|
self.queue.get = unrelated_after
|
|
self.dequeue(0)
|
|
self.queue.task_done(0, {}, status())
|
|
self.assertTrue(first.dispose())
|
|
self.assertTrue(first.dispose())
|
|
self.assertIs(self.queue.get, unrelated_after)
|
|
second = self.register()
|
|
self.dequeue(1)
|
|
self.queue.task_done(1, {}, status())
|
|
self.drain(second)
|
|
self.assertEqual(len(self.rows()), 2)
|
|
self.assertEqual(self.rows()[0]["session_id"], self.rows()[1]["session_id"])
|
|
self.assertEqual(calls, ["after", "before"] * 2)
|
|
|
|
def test_concurrent_handle_capacity_never_reassigns_active_attempts(self):
|
|
registration = self.register()
|
|
self.queue.results.extend(item(index) for index in range(129))
|
|
ready = threading.Barrier(130)
|
|
errors = []
|
|
|
|
def dequeue():
|
|
try:
|
|
ready.wait(3)
|
|
self.queue.get()
|
|
except Exception as error:
|
|
errors.append(error)
|
|
|
|
threads = [threading.Thread(target=dequeue) for _ in range(129)]
|
|
for thread in threads:
|
|
thread.start()
|
|
ready.wait(3)
|
|
for thread in threads:
|
|
thread.join(3)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(errors, [])
|
|
self.drain(registration)
|
|
rows = self.rows()
|
|
admitted = {row["queue_item_id"] for row in rows}
|
|
self.assertEqual(len(rows), 128)
|
|
missing = (set(range(129)) - admitted).pop()
|
|
oldest = min(admitted)
|
|
self.queue.task_done(oldest, {}, status())
|
|
self.queue.task_done(missing, {}, status())
|
|
self.drain(registration)
|
|
by_item = {row["queue_item_id"]: row for row in self.rows()}
|
|
self.assertEqual(by_item[oldest]["outcome"], "succeeded")
|
|
self.assertNotIn(missing, by_item)
|
|
self.assertGreater(registration.writer.status().recording_gaps, 0)
|
|
self.dequeue(129)
|
|
self.queue.task_done(129, {}, status())
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows()[-1]["queue_item_id"], 129)
|
|
|
|
def test_active_handles_do_not_retain_graph_extra_or_sensitive_envelopes(self):
|
|
class Envelope(dict):
|
|
pass
|
|
|
|
registration = self.register()
|
|
graph, extra, sensitive = (
|
|
Envelope({"g": object()}),
|
|
Envelope({"e": object()}),
|
|
Envelope({"s": object()}),
|
|
)
|
|
references = [weakref.ref(value) for value in (graph, extra, sensitive)]
|
|
self.dequeue(0, graph=graph, extra=extra, sensitive=sensitive)
|
|
self.drain(registration)
|
|
del graph, extra, sensitive
|
|
gc.collect()
|
|
self.assertEqual([reference() for reference in references], [None, None, None])
|
|
self.queue.task_done(0, {}, status())
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows()[0]["outcome"], "succeeded")
|
|
|
|
def test_capture_uses_no_database_methods_on_the_host_thread(self):
|
|
registration = self.register()
|
|
host_thread = threading.get_ident()
|
|
original = self.db.model.get_connection
|
|
|
|
def connection():
|
|
if threading.get_ident() == host_thread:
|
|
self.fail("history queried SQLite on the host thread")
|
|
return original()
|
|
|
|
with patch.object(self.db.model, "get_connection", connection):
|
|
self.dequeue(0)
|
|
self.queue.task_done(0, {}, status())
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows()[0]["outcome"], "succeeded")
|
|
|
|
def test_capture_never_copies_or_serializes_sensitive_objects(self):
|
|
class Sensitive:
|
|
def __deepcopy__(self, memo):
|
|
raise AssertionError("sensitive copy")
|
|
|
|
def __repr__(self):
|
|
raise AssertionError("sensitive serialization")
|
|
|
|
registration = self.register()
|
|
secret = Sensitive()
|
|
graph, extra, sensitive = {"graph": secret}, {"extra": secret}, {"auth": secret}
|
|
result = self.dequeue(0, graph=graph, extra=extra, sensitive=sensitive)
|
|
payload = {"outputs": secret, "meta": secret}
|
|
completion = status()
|
|
completion.messages[0][1].update(
|
|
{"traceback": secret, "current_inputs": secret}
|
|
)
|
|
self.assertIs(
|
|
self.queue.task_done(0, payload, completion), self.queue.done_result
|
|
)
|
|
self.drain(registration)
|
|
self.assertIs(result[0][2], graph)
|
|
self.assertIs(result[0][3], extra)
|
|
self.assertIs(result[0][5], sensitive)
|
|
self.assertEqual(len(self.rows()), 1)
|
|
self.assertEqual(self.rows()[0]["outcome"], "succeeded")
|
|
self.assertEqual(
|
|
set(completion.messages[0][1]),
|
|
{"prompt_id", "timestamp", "traceback", "current_inputs"},
|
|
)
|
|
|
|
def test_callbacks_return_while_writer_blocked_and_disposal_is_bounded(self):
|
|
entered, release = threading.Event(), threading.Event()
|
|
self.releases.append(release)
|
|
original = self.db.write_execution_start
|
|
|
|
def blocked(command):
|
|
entered.set()
|
|
release.wait(5)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", blocked):
|
|
registration = self.register()
|
|
self.dequeue(0)
|
|
self.assertTrue(entered.wait(1))
|
|
began = time.monotonic()
|
|
self.queue.task_done(0, {}, status())
|
|
self.dequeue(1)
|
|
self.assertLess(time.monotonic() - began, 0.2)
|
|
began = time.monotonic()
|
|
self.assertFalse(registration.dispose())
|
|
self.assertGreaterEqual(time.monotonic() - began, 1.8)
|
|
self.assertLess(time.monotonic() - began, 2.3)
|
|
release.set()
|
|
self.assertTrue(registration.dispose(2))
|
|
|
|
def test_restore_disables_capture_and_clears_handles_before_fresh_epoch(self):
|
|
try:
|
|
from tests.test_execution_history_writer import ManualClock
|
|
except ImportError:
|
|
from test_execution_history_writer import ManualClock
|
|
clock = ManualClock()
|
|
with patch(
|
|
"utils.execution_history.HistoryWriter",
|
|
lambda factory, **kwargs: HistoryWriter(factory, clock=clock, **kwargs),
|
|
):
|
|
registration = self.register()
|
|
for index in range(128):
|
|
self.dequeue(index)
|
|
self.drain(registration)
|
|
old_epoch = registration.writer.status().epoch
|
|
backup = os.path.join(self.directory.name, "backup.db")
|
|
self.assertTrue(self.db.model.backup_database(backup))
|
|
self.db.model.restore_from_file(backup)
|
|
self.assertFalse(registration.writer.status().accepting)
|
|
self.assertIsNone(registration.writer.status().epoch)
|
|
self.queue.task_done(0, {}, status())
|
|
self.dequeue(128)
|
|
self.assertGreater(registration.writer.status().recording_gaps, 0)
|
|
clock.advance(5)
|
|
self.wait_for(lambda: registration.writer.status().accepting)
|
|
self.assertNotEqual(registration.writer.status().epoch, old_epoch)
|
|
# A late final descriptor cannot complete rows from the restored epoch.
|
|
self.queue.task_done(0, {}, status())
|
|
self.dequeue(129)
|
|
self.queue.task_done(129, {}, status())
|
|
self.drain(registration)
|
|
rows = self.rows()
|
|
self.assertEqual(len(rows), 129)
|
|
self.assertEqual(rows[0]["outcome"], "observed")
|
|
self.assertEqual(rows[-1]["queue_item_id"], 129)
|
|
self.assertEqual(rows[-1]["outcome"], "succeeded")
|
|
|
|
def test_unavailable_database_and_invalid_identity_never_escape_host(self):
|
|
entered = threading.Event()
|
|
|
|
def broken():
|
|
entered.set()
|
|
raise RuntimeError("secret unavailable database")
|
|
|
|
registration = self.register(broken, ready=False)
|
|
self.assertTrue(entered.wait(1))
|
|
result = self.dequeue(0)
|
|
self.assertEqual(result[1], 0)
|
|
self.queue.task_done(0, {}, status())
|
|
self.assertEqual(self.rows(), [])
|
|
self.assertGreater(registration.writer.status().recording_gaps, 0)
|
|
self.assertTrue(registration.dispose())
|
|
registration = self.register()
|
|
self.dequeue(0, job="invalid id")
|
|
self.queue.task_done(True, {}, object())
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows(), [])
|
|
self.assertGreater(registration.writer.status().recording_gaps, 0)
|
|
|
|
def test_invalid_evidence_is_recorded_unknown_and_raising_status_is_isolated(self):
|
|
registration = self.register()
|
|
self.dequeue(0)
|
|
self.queue.task_done(0, {}, object())
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows()[0]["outcome"], "unknown")
|
|
self.assertEqual(self.rows()[0]["reason"], "invalid_evidence")
|
|
|
|
class BrokenStatus:
|
|
@property
|
|
def messages(self):
|
|
raise RuntimeError("secret failure")
|
|
|
|
self.dequeue(1)
|
|
self.queue.task_done(1, {}, BrokenStatus())
|
|
self.drain(registration)
|
|
self.assertEqual(self.rows()[1]["outcome"], "observed")
|
|
self.assertGreater(registration.writer.status().recording_gaps, 0)
|
|
|
|
def test_unsupported_queue_does_not_create_writer_or_submit_fallback(self):
|
|
self.assertTrue(
|
|
callable(register_history_hook), "history registration is absent"
|
|
)
|
|
for host in (
|
|
object(),
|
|
SimpleNamespace(prompt_queue=SimpleNamespace(get=lambda: None)),
|
|
):
|
|
self.assertIsNone(
|
|
register_history_hook(host, lambda: self.fail("DB touched"))
|
|
)
|
|
|
|
def test_existing_usage_callback_counts_once_alongside_history(self):
|
|
registration = self.register()
|
|
seen = []
|
|
self.assertTrue(usage_tracking._wrap_queue_get(self.queue, seen.append))
|
|
self.assertFalse(usage_tracking._wrap_queue_get(self.queue, seen.append))
|
|
self.dequeue(0)
|
|
self.queue.task_done(0, {}, status())
|
|
self.drain(registration)
|
|
self.assertEqual(seen, [JOB])
|
|
self.assertEqual(len(self.rows()), 1)
|
|
|
|
def test_fixture_cleanup_discards_saturation_before_release_and_database_close(
|
|
self,
|
|
):
|
|
from utils.execution_history_events import make_start
|
|
|
|
entered, release, rescue = (threading.Event() for _ in range(3))
|
|
self.releases.append(release)
|
|
pending_at_release, writes = [], []
|
|
original = self.db.write_execution_start
|
|
|
|
def blocked(command):
|
|
entered.set()
|
|
release.wait(5)
|
|
pending = registration.writer.status().pending
|
|
pending_at_release.append(pending)
|
|
# Model a slow drain without executing 1,024 SQLite transactions.
|
|
if pending > 1:
|
|
rescue.wait(5)
|
|
writes.append(command.run_id)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", blocked):
|
|
registration = self.register()
|
|
try:
|
|
self.dequeue(1)
|
|
self.assertTrue(entered.wait(1))
|
|
epoch = registration.writer.status().epoch
|
|
for i in range(1023):
|
|
command = make_start(
|
|
registration._session_id, item(1000 + i), epoch, 0
|
|
)
|
|
self.assertTrue(registration.writer.submit(command))
|
|
self.assertEqual(registration.writer.status().pending, 1024)
|
|
before = time.monotonic()
|
|
self.cleanup_registrations()
|
|
self.assertLess(time.monotonic() - before, 2)
|
|
self.assertEqual(pending_at_release, [1])
|
|
self.assertEqual(len(writes), 1)
|
|
self.assertFalse(registration.writer.status().alive)
|
|
self.assertEqual(registration.writer.status().pending, 0)
|
|
self.db.model.close_all()
|
|
finally:
|
|
registration.writer.invalidate()
|
|
rescue.set()
|
|
release.set()
|
|
self.assertTrue(
|
|
registration.dispose(3), "regression rescue leaked worker"
|
|
)
|
|
|
|
def test_rejected_new_start_revokes_previous_worker_capability(self):
|
|
from utils.execution_context_capture import active_capture
|
|
from utils.execution_history_events import make_start
|
|
|
|
entered, release = threading.Event(), threading.Event()
|
|
self.releases.append(release)
|
|
original = self.db.write_execution_start
|
|
|
|
def blocked(command):
|
|
entered.set()
|
|
release.wait(5)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", blocked):
|
|
registration = self.register()
|
|
self.dequeue(1)
|
|
self.assertTrue(entered.wait(1))
|
|
previous = active_capture()
|
|
self.assertIsNotNone(previous)
|
|
epoch = registration.writer.status().epoch
|
|
for i in range(1023):
|
|
command = make_start(registration._session_id, item(1000 + i), epoch, 0)
|
|
self.assertTrue(registration.writer.submit(command))
|
|
self.assertEqual(registration.writer.status().pending, 1024)
|
|
self.dequeue(2)
|
|
self.assertIsNone(active_capture())
|
|
self.assertFalse(previous[0].valid(previous[1]))
|
|
registration.writer.invalidate()
|
|
release.set()
|
|
self.assertTrue(registration.dispose(3), "owned worker leaked")
|
|
|
|
def test_late_finish_does_not_clear_new_attempt_binding(self):
|
|
from utils.execution_context_capture import active_capture
|
|
|
|
registration = self.register()
|
|
self.dequeue(1)
|
|
self.dequeue(2)
|
|
current = active_capture()
|
|
self.assertIsNotNone(current)
|
|
self.queue.task_done(1, {}, status())
|
|
self.assertEqual(active_capture(), current)
|
|
# An identical callback replay keeps the already accepted binding.
|
|
registration._dequeue(item(2))
|
|
self.assertEqual(active_capture(), current)
|
|
self.queue.task_done(2, {}, status())
|
|
self.assertIsNone(active_capture())
|
|
|
|
def test_invalidation_releases_capture_immediately_outside_writer_locks(self):
|
|
from utils.execution_context_capture import active_capture
|
|
|
|
registration = self.register()
|
|
self.dequeue(1)
|
|
current = active_capture()
|
|
pool = registration.writer.context_budget
|
|
self.assertGreater(pool.charged_bytes, 0)
|
|
registration.writer.invalidate()
|
|
self.assertEqual(pool.charged_bytes, 0)
|
|
self.assertFalse(current[0].valid(current[1]))
|
|
|
|
def test_replayed_older_accepted_dequeue_cannot_revoke_current_binding(self):
|
|
from utils.execution_context_capture import active_capture
|
|
|
|
registration = self.register()
|
|
old = self.dequeue(1)
|
|
self.dequeue(2)
|
|
current = active_capture()
|
|
registration._dequeue(old)
|
|
self.assertEqual(active_capture(), current)
|
|
|
|
def test_broken_finish_releases_active_capture(self):
|
|
from utils.execution_context_capture import active_capture
|
|
|
|
registration = self.register()
|
|
self.dequeue(1)
|
|
|
|
class Broken:
|
|
@property
|
|
def messages(self):
|
|
raise RuntimeError("broken status")
|
|
|
|
self.queue.task_done(1, {}, Broken())
|
|
self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
|
|
self.assertIsNone(active_capture())
|
|
|
|
def test_finished_accepted_replay_preserves_new_same_uuid_attempt(self):
|
|
from utils.execution_context_capture import active_capture
|
|
|
|
registration = self.register()
|
|
old = self.dequeue(1)
|
|
self.queue.task_done(1, {}, status())
|
|
self.dequeue(2)
|
|
current = active_capture()
|
|
registration._dequeue(old)
|
|
self.assertEqual(active_capture(), current)
|
|
self.queue.task_done(2, {}, status())
|
|
self.drain(registration)
|
|
self.assertEqual(len(self.rows()), 2)
|
|
|
|
def test_observed_upstream_survives_downstream_failure_in_database(self):
|
|
from utils.execution_context_capture import record_runtime_context
|
|
|
|
g = {
|
|
"p": {
|
|
"class_type": "PromptManagerPair",
|
|
"inputs": {"text": "raw", "negative_text": ""},
|
|
},
|
|
"s": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["p", 0], "negative": ["p", 1]},
|
|
},
|
|
}
|
|
registration = self.register()
|
|
self.queue.results.append(((0, JOB, g, {}, ["s"]), 1))
|
|
self.queue.get()
|
|
host = types.ModuleType("comfy_execution.utils")
|
|
host.get_executing_context = lambda: SimpleNamespace(
|
|
prompt_id=JOB, node_id="p", list_index=0
|
|
)
|
|
with patch.dict(sys.modules, {"comfy_execution.utils": host}):
|
|
record_runtime_context(
|
|
"PromptManagerPair",
|
|
" raw ",
|
|
" pre ",
|
|
"",
|
|
"pre raw TRIGGER",
|
|
role="positive",
|
|
)
|
|
record_runtime_context("PromptManagerPair", "", "", "", "", role="negative")
|
|
self.queue.task_done(1, {}, status("execution_error"))
|
|
self.drain(registration)
|
|
row = self.rows()[0]
|
|
self.assertEqual(row["outcome"], "failed")
|
|
result = self.db.get_execution_contexts(row["run_id"])
|
|
self.assertEqual(result["state"], "complete")
|
|
self.assertEqual(
|
|
result["contexts"][0]["positive"],
|
|
{
|
|
"raw": " raw ",
|
|
"effective": "pre raw TRIGGER",
|
|
"prepend": " pre ",
|
|
"append": "",
|
|
},
|
|
)
|
|
self.assertEqual(result["contexts"][0]["negative"]["effective"], "")
|
|
|
|
def test_close_releases_capture_while_writer_is_blocked(self):
|
|
entered, release = threading.Event(), threading.Event()
|
|
self.releases.append(release)
|
|
original = self.db.write_execution_start
|
|
|
|
def blocked(command):
|
|
entered.set()
|
|
release.wait(3)
|
|
return original(command)
|
|
|
|
with patch.object(self.db, "write_execution_start", blocked):
|
|
registration = self.register()
|
|
self.dequeue(1)
|
|
self.assertTrue(entered.wait(1))
|
|
self.assertGreater(registration.writer.context_budget.charged_bytes, 0)
|
|
self.assertFalse(registration.writer.close(0))
|
|
self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
|
|
release.set()
|
|
|
|
def test_reregistration_cannot_rebind_already_observed_queue_item(self):
|
|
from utils.execution_context_capture import active_capture
|
|
|
|
first = self.register()
|
|
self.dequeue(10)
|
|
self.queue.task_done(10, {}, status())
|
|
self.drain(first)
|
|
self.assertTrue(first.dispose())
|
|
second = self.register()
|
|
self.dequeue(10)
|
|
self.assertIsNone(active_capture())
|
|
self.queue.task_done(10, {}, status())
|
|
self.drain(second)
|
|
self.assertEqual(len(self.rows()), 1)
|
|
|
|
def test_cached_executor_subscription_persists_strings_and_disposes_cooperatively(
|
|
self,
|
|
):
|
|
from utils.execution_context_host import subscribe_executor_messages
|
|
|
|
calls = []
|
|
execution = types.ModuleType("execution")
|
|
caching = types.ModuleType("comfy_execution.caching")
|
|
package = types.ModuleType("comfy_execution")
|
|
entry_type = namedtuple("CacheEntry", "ui outputs")
|
|
|
|
class PromptExecutor:
|
|
def add_message(self, event, data, broadcast):
|
|
calls.append(event)
|
|
return "host-result"
|
|
|
|
class HierarchicalCache:
|
|
def get_local(self, node_id):
|
|
return entry_type(None, [object(), object(), ["old trigger"], [""]])
|
|
|
|
execution.PromptExecutor, execution.CacheEntry = PromptExecutor, entry_type
|
|
caching.HierarchicalCache = HierarchicalCache
|
|
package.caching = caching
|
|
with patch.dict(
|
|
sys.modules,
|
|
{
|
|
"execution": execution,
|
|
"comfy_execution": package,
|
|
"comfy_execution.caching": caching,
|
|
},
|
|
):
|
|
observer_calls = []
|
|
independent = subscribe_executor_messages(
|
|
"independent", lambda *args: observer_calls.append(args)
|
|
)
|
|
try:
|
|
registration = self.register()
|
|
g = {
|
|
"p": {
|
|
"class_type": "PromptManagerPair",
|
|
"inputs": {"text": "raw", "negative_text": ""},
|
|
},
|
|
"s": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["p", 0], "negative": ["p", 1]},
|
|
},
|
|
}
|
|
self.queue.results.append(((0, JOB, g, {}, ["s"]), 1))
|
|
self.queue.get()
|
|
executor = PromptExecutor()
|
|
executor.server = self.server
|
|
executor.caches = SimpleNamespace(outputs=HierarchicalCache())
|
|
self.assertEqual(
|
|
executor.add_message("execution_start", {"prompt_id": JOB}, False),
|
|
"host-result",
|
|
)
|
|
executor.add_message(
|
|
event="execution_cached",
|
|
data={"prompt_id": JOB, "nodes": ["p"]},
|
|
broadcast=False,
|
|
)
|
|
self.queue.task_done(1, {}, status())
|
|
self.drain(registration)
|
|
result = self.db.get_execution_contexts(self.rows()[0]["run_id"])
|
|
self.assertEqual(
|
|
result["contexts"][0]["positive"]["effective"], "old trigger"
|
|
)
|
|
self.assertEqual(result["contexts"][0]["negative"]["effective"], "")
|
|
self.assertTrue(registration.dispose())
|
|
executor.add_message(
|
|
"execution_cached", {"prompt_id": JOB, "nodes": ["p"]}, False
|
|
)
|
|
self.assertEqual(len(observer_calls), 3)
|
|
finally:
|
|
independent()
|
|
self.assertEqual(
|
|
calls, ["execution_start", "execution_cached", "execution_cached"]
|
|
)
|