diff --git a/scripts/cleanup_event_storm_backlog.py b/scripts/cleanup_event_storm_backlog.py new file mode 100644 index 000000000..97d53a770 --- /dev/null +++ b/scripts/cleanup_event_storm_backlog.py @@ -0,0 +1,223 @@ +#!/usr/bin/env python3 +"""One-off cleanup of the event-storm instance backlog (JVNAUTOSCI-2507). + +Self-sustaining event cascades (paper_recommendation_evaluation_workflow and +episode_evaluation_workflow re-trigger themselves via their own +text_relation.* / relationship.* / workflow.instance_terminal bindings) left a +large backlog of pending / paused / orphaned-running instances that never drain. + +This script cancels that backlog with a *direct* bulk ``update_many`` on the +workflow_instances collection. It deliberately does NOT route through +WorkflowInstanceManager.mark_cancelled, because that path emits a +``workflow.instance_terminal`` event which would spawn a fresh +episode_evaluation instance per cancellation — i.e. cancelling the backlog the +normal way would feed the very storm being drained. A direct update emits no +events (there is no change-stream watcher on the collection), so it is safe. + +Dry-run by default: prints the counts it WOULD cancel and changes nothing. +Pass --execute to perform the cancellation. + +Examples: + # Inspect scope (read-only) + .venv/bin/python scripts/cleanup_event_storm_backlog.py + + # Cancel pending + paused backlog for the two storm workflows + .venv/bin/python scripts/cleanup_event_storm_backlog.py --execute + + # Also cancel orphaned running instances (expired / missing lock) + .venv/bin/python scripts/cleanup_event_storm_backlog.py \ + --include-stale-running --execute +""" + +from __future__ import annotations + +import argparse +import os +import sys +from datetime import datetime, timezone + +# Allow running from the repo root without installation. +sys.path.insert(0, ".") + +# The shared Atlas cluster is heavily loaded by the very storm this script +# drains, so the default 5-10s socket timeout trips on large indexed counts. +# Raise it for this one-off maintenance run (before the client is built). +os.environ.setdefault("MONGO_SOCKET_TIMEOUT_MS", "120000") +os.environ.setdefault("MONGO_CONNECT_TIMEOUT_MS", "20000") + +from src.backend.db.mongo_client import get_db # noqa: E402 + +# Server-side cap so a slow count cannot pin a cluster connection indefinitely. +_COUNT_MAX_TIME_MS = 90000 + +WORKFLOW_INSTANCES_COLLECTION = "workflow_instances" + +DEFAULT_STORM_WORKFLOWS = ( + "#V#paper_recommendation_evaluation_workflow", + "#V#episode_evaluation_workflow", +) +DEFAULT_STATUSES = ("pending", "paused") +RUNNING_STATUS = "running" + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--workflow-id", + action="append", + dest="workflow_ids", + help=( + "Workflow concept id to clean up (repeatable). " + f"Defaults to: {', '.join(DEFAULT_STORM_WORKFLOWS)}" + ), + ) + parser.add_argument( + "--status", + action="append", + dest="statuses", + help=( + "Instance status to cancel (repeatable). " + f"Defaults to: {', '.join(DEFAULT_STATUSES)}" + ), + ) + parser.add_argument( + "--include-stale-running", + action="store_true", + help=( + "Also cancel running instances whose lock has expired or is absent " + "(orphaned mid-flight). Running instances with a live lock are never " + "touched." + ), + ) + parser.add_argument( + "--batch-size", + type=int, + default=2000, + help="Max documents per update_many batch (default 2000).", + ) + parser.add_argument( + "--execute", + action="store_true", + help="Perform the cancellation. Without this flag the script is read-only.", + ) + return parser.parse_args() + + +def _cancel_set(now: datetime) -> dict: + return { + "$set": { + "status": "cancelled", + "completed_at": now, + "locked_by": None, + "lock_expires_at": None, + "progress_message": "cancelled:event_storm_backlog_cleanup:JVNAUTOSCI-2507", + "progress_updated_at": now, + } + } + + +def _safe_count(coll, query: dict) -> int | None: + """Best-effort count; returns None if the loaded cluster cannot serve it.""" + try: + return int(coll.count_documents(query, maxTimeMS=_COUNT_MAX_TIME_MS)) + except Exception as exc: # noqa: BLE001 - report and continue + print(f" (count unavailable: {type(exc).__name__})") + return None + + +def _bulk_cancel( + coll, query: dict, *, batch_size: int, execute: bool +) -> tuple[int | None, int]: + """Return (matched_count_or_None, cancelled_count). + + Dry-run reports the (best-effort) match count and cancels nothing. Execute + cancels in index-bounded ``_id`` batches; because each batch flips the + documents out of the matching set, the loop drains without needing an + up-front full count (which the loaded cluster may not be able to serve). + Uses a direct update_many, NOT mark_cancelled, so no instance_terminal + events are emitted (that would re-spawn episode_evaluation instances). + """ + if not execute: + return _safe_count(coll, query), 0 + now = datetime.now(timezone.utc) + cancelled = 0 + batch_no = 0 + while True: + ids = [doc["_id"] for doc in coll.find(query, {"_id": 1}).limit(batch_size)] + if not ids: + break + result = coll.update_many({"_id": {"$in": ids}}, _cancel_set(now)) + cancelled += int(result.modified_count) + batch_no += 1 + print(f" batch {batch_no}: +{result.modified_count} (running {cancelled})", + flush=True) + if len(ids) < batch_size: + break + return cancelled, cancelled + + +def main() -> int: + args = _parse_args() + workflow_ids = list(args.workflow_ids or DEFAULT_STORM_WORKFLOWS) + statuses = list(args.statuses or DEFAULT_STATUSES) + + db = get_db() + if db is None: + print("ERROR: database unavailable", file=sys.stderr) + return 1 + coll = db[WORKFLOW_INSTANCES_COLLECTION] + + mode = "EXECUTE" if args.execute else "DRY-RUN" + print(f"=== event-storm backlog cleanup [{mode}] ===") + print(f"workflows: {workflow_ids}") + print(f"statuses: {statuses}" + f"{' + stale running' if args.include_stale_running else ''}") + print() + + now = datetime.now(timezone.utc) + verb = "cancelled" if args.execute else "would cancel" + grand_total = 0 + + def _report(label: str, matched: int | None, cancelled: int) -> None: + nonlocal grand_total + if args.execute: + grand_total += cancelled + print(f" {label}: {verb} {cancelled}") + else: + shown = "unknown" if matched is None else matched + if isinstance(matched, int): + grand_total += matched + print(f" {label}: {verb} {shown}") + + for workflow_id in workflow_ids: + query = {"workflow_id": workflow_id, "status": {"$in": statuses}} + matched, cancelled = _bulk_cancel( + coll, query, batch_size=args.batch_size, execute=args.execute + ) + _report(f"{workflow_id} [{','.join(statuses)}]", matched, cancelled) + + if args.include_stale_running: + stale_query = { + "workflow_id": workflow_id, + "status": RUNNING_STATUS, + "$or": [ + {"lock_expires_at": {"$lt": now}}, + {"lock_expires_at": None}, + {"lock_expires_at": {"$exists": False}}, + ], + } + s_matched, s_cancelled = _bulk_cancel( + coll, stale_query, batch_size=args.batch_size, execute=args.execute + ) + _report(f"{workflow_id} [stale running]", s_matched, s_cancelled) + + print() + print(f"TOTAL {'cancelled' if args.execute else 'to cancel (best-effort)'}: " + f"{grand_total}") + if not args.execute: + print("\n(dry run — nothing changed; re-run with --execute to apply)") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/backend/workflows/durable/instance_manager.py b/src/backend/workflows/durable/instance_manager.py index ac7117a98..66a33d36e 100644 --- a/src/backend/workflows/durable/instance_manager.py +++ b/src/backend/workflows/durable/instance_manager.py @@ -813,6 +813,66 @@ def get_event_instance( ) return self._instance_from_doc(doc) + def find_instance_id_by_event_key( + self, event_idempotency_key: str + ) -> str | None: + """Return the instance_id already created for an event idempotency key. + + Cheap unique-index lookup used to keep idempotent event redelivery from + being suppressed by the event-storm backlog damper (JVNAUTOSCI-2507): + a redelivered event must still resolve to its existing instance. + """ + key = str(event_idempotency_key or "").strip() + if not key: + return None + coll = self._get_instances_collection() + if coll is None: + return None + doc = coll.find_one({"event_idempotency_key": key}, {"instance_id": 1}) + if doc and isinstance(doc.get("instance_id"), str): + return doc["instance_id"] + return None + + def count_recent_event_backlog( + self, + *, + workflow_id: str, + source_event_type: str, + statuses: Iterable[WorkflowInstanceStatus | str], + since_utc: datetime, + ) -> int: + """Count recent event-triggered instances for a (workflow, event-type) pair. + + Supports the event-storm backlog damper (JVNAUTOSCI-2507). The query is + served by the ``workflow_created`` index (workflow_id + created_at) and is + always time-bounded via ``since_utc`` so it cannot turn into an unindexed + full scan on the shared cluster. + """ + workflow_id_clean = str(workflow_id or "").strip() + source_event_type_clean = str(source_event_type or "").strip() + if not workflow_id_clean or not source_event_type_clean: + return 0 + status_values = self._normalise_status_filter(statuses) + coll = self._get_instances_collection() + if coll is None: + return 0 + query: dict[str, Any] = { + "workflow_id": workflow_id_clean, + "source_event_type": source_event_type_clean, + "created_at": {"$gte": since_utc}, + } + if status_values: + query["status"] = {"$in": status_values} + try: + return int(coll.count_documents(query)) + except Exception: + logger.warning( + "[event_storm_damper] backlog count failed for %s/%s", + workflow_id_clean, + source_event_type_clean, + ) + return 0 + def get_instance(self, instance_id: str) -> WorkflowInstance | None: """Load a workflow instance by ID. diff --git a/src/backend/workflows/durable/workflow_instance_submission_service.py b/src/backend/workflows/durable/workflow_instance_submission_service.py index 20d40e13a..85cf21c72 100644 --- a/src/backend/workflows/durable/workflow_instance_submission_service.py +++ b/src/backend/workflows/durable/workflow_instance_submission_service.py @@ -24,8 +24,10 @@ from functools import lru_cache import hashlib import json +import logging import os import threading +from datetime import datetime, timedelta, timezone from time import monotonic, perf_counter from typing import Any, Callable, Dict, List, Mapping, Sequence @@ -57,6 +59,8 @@ ) from .instance_manager import WorkflowInstanceManager +logger = logging.getLogger(__name__) + _RUNNABLE_CACHE_TTL_ENV = "VON_WORKFLOW_RUNNABLE_CACHE_TTL_SECONDS" _RUNNABLE_CACHE_MAX_ENTRIES_ENV = "VON_WORKFLOW_RUNNABLE_CACHE_MAX_ENTRIES" _DEFAULT_RUNNABLE_CACHE_TTL_SECONDS = 45.0 @@ -64,6 +68,20 @@ _MAX_RUNNABLE_CACHE_ENTRIES = 2048 _MIN_RUNNABLE_CACHE_ENTRIES = 8 +# Event-storm backlog damper (JVNAUTOSCI-2507): suppress NEW event-triggered +# instance creation for a (workflow_id, source_event_type) pair once an +# unprocessed backlog of that pair already exists. Self-sustaining event +# cascades (e.g. paper_recommendation/episode workflows that rewrite graph +# relations, re-triggering themselves) otherwise create hundreds of pending +# instances that never drain. This is a high-watermark guard, not a per-event +# dedupe: at the default limit, normal low-volume event processing is +# unaffected. Set the limit to <= 0 to disable. +_EVENT_BACKLOG_LIMIT_ENV = "VON_WORKFLOW_EVENT_BACKLOG_LIMIT" +_EVENT_BACKLOG_WINDOW_ENV = "VON_WORKFLOW_EVENT_BACKLOG_WINDOW_SECONDS" +_DEFAULT_EVENT_BACKLOG_LIMIT = 50 +_DEFAULT_EVENT_BACKLOG_WINDOW_SECONDS = 21600.0 # 6h; bounds the backlog count +_EVENT_BACKLOG_PENDING_STATUSES = ("pending", "running") + @dataclass(frozen=True) class WorkflowRunnableVerification: @@ -248,6 +266,96 @@ def _read_runnable_cache_max_entries() -> int: ) +def _read_event_backlog_limit() -> int: + """Backlog high-watermark above which new event-triggered creation is + suppressed. ``<= 0`` disables the damper (JVNAUTOSCI-2507).""" + return _read_int_env( + _EVENT_BACKLOG_LIMIT_ENV, + default=_DEFAULT_EVENT_BACKLOG_LIMIT, + minimum=0, + maximum=100000, + ) + + +def _read_event_backlog_window_seconds() -> float: + return _read_float_env( + _EVENT_BACKLOG_WINDOW_ENV, + default=_DEFAULT_EVENT_BACKLOG_WINDOW_SECONDS, + minimum=60.0, + maximum=604800.0, + ) + + +def _event_backlog_throttle_result( + *, + manager: "WorkflowInstanceManager", + workflow_id: str, + source_event_type: str, + event_idempotency_key: str, + verification_payload: Mapping[str, Any], +) -> "WorkflowInstanceSubmissionResult | None": + """Event-storm backlog damper (JVNAUTOSCI-2507). + + Returns a ``throttled_event_backlog`` submission result when an unprocessed + backlog already exists for the ``(workflow_id, source_event_type)`` pair, so + a self-sustaining event cascade cannot keep minting instances that never + drain. Returns ``None`` (allow creation) when the damper is disabled, when + the event is an idempotent redelivery of an already-created instance, or + when the backlog is below the configured high-watermark. + """ + + limit = _read_event_backlog_limit() + if limit <= 0: + return None + # Idempotent redelivery of the same event must resolve to its existing + # instance, never be throttled. + try: + if manager.find_instance_id_by_event_key(event_idempotency_key): + return None + except Exception: + # If the idempotency probe fails, fall through to the backlog check + # rather than risk double-suppressing or double-creating. + pass + + window_seconds = _read_event_backlog_window_seconds() + since_utc = datetime.now(timezone.utc) - timedelta(seconds=window_seconds) + backlog = manager.count_recent_event_backlog( + workflow_id=workflow_id, + source_event_type=source_event_type, + statuses=_EVENT_BACKLOG_PENDING_STATUSES, + since_utc=since_utc, + ) + if backlog < limit: + return None + + logger.warning( + "[event_storm_damper] suppressed event-triggered instance for " + "workflow=%s source_event_type=%s: backlog=%d (pending/running within " + "%.0fs) >= limit=%d", + workflow_id, + source_event_type, + backlog, + window_seconds, + limit, + ) + return WorkflowInstanceSubmissionResult( + success=False, + workflow_id=workflow_id, + status="throttled_event_backlog", + instance_id=None, + error_code="event_backlog_throttled", + error=( + f"Event-triggered instance for workflow '{workflow_id}' on " + f"'{source_event_type}' was suppressed: {backlog} unprocessed " + f"instance(s) already pending within the backlog window " + f"(limit {limit}). The existing backlog must drain before new " + "event-triggered work is created." + ), + verification=dict(verification_payload), + created_new=False, + ) + + def _stable_json_hash(payload: Mapping[str, Any]) -> str: encoded = json.dumps( payload, @@ -1221,6 +1329,15 @@ def submit_verified_workflow_instance( ) created_new = True if use_event_idempotency_submission: + throttled = _event_backlog_throttle_result( + manager=manager, + workflow_id=workflow_id, + source_event_type=str(source_event_type).strip(), + event_idempotency_key=str(event_idempotency_key).strip(), + verification_payload=verification_payload, + ) + if throttled is not None: + return throttled instance_id, created_new = manager.create_instance_for_event( workflow_id=workflow_id, user_id=resolved_user_id, diff --git a/tests/backend/test_event_storm_damper.py b/tests/backend/test_event_storm_damper.py new file mode 100644 index 000000000..c95100a75 --- /dev/null +++ b/tests/backend/test_event_storm_damper.py @@ -0,0 +1,114 @@ +"""Tests for the event-storm backlog damper (JVNAUTOSCI-2507). + +The damper suppresses NEW event-triggered workflow-instance creation for a +``(workflow_id, source_event_type)`` pair once an unprocessed backlog already +exists, so a self-sustaining event cascade (e.g. paper_recommendation / +episode_evaluation workflows that rewrite graph relations and re-trigger +themselves) cannot keep minting instances that never drain. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest + +from src.backend.workflows.durable import workflow_instance_submission_service as svc + + +def _fake_manager(*, existing_key: str | None, backlog: int) -> MagicMock: + manager = MagicMock() + manager.find_instance_id_by_event_key.return_value = existing_key + manager.count_recent_event_backlog.return_value = backlog + return manager + + +def _throttle(manager: MagicMock): + return svc._event_backlog_throttle_result( + manager=manager, + workflow_id="#V#paper_recommendation_evaluation_workflow", + source_event_type="text_relation.upserted", + event_idempotency_key="evt:text_relation.upserted:abc123", + verification_payload={"runnable_verification_success": True}, + ) + + +# --------------------------------------------------------------------------- +# Env readers +# --------------------------------------------------------------------------- + + +def test_backlog_limit_default(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(svc._EVENT_BACKLOG_LIMIT_ENV, raising=False) + assert svc._read_event_backlog_limit() == svc._DEFAULT_EVENT_BACKLOG_LIMIT + + +def test_backlog_limit_override_and_disable(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(svc._EVENT_BACKLOG_LIMIT_ENV, "5") + assert svc._read_event_backlog_limit() == 5 + monkeypatch.setenv(svc._EVENT_BACKLOG_LIMIT_ENV, "0") + assert svc._read_event_backlog_limit() == 0 # disabled sentinel + + +def test_backlog_window_clamped(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(svc._EVENT_BACKLOG_WINDOW_ENV, "5") # below minimum 60 + assert svc._read_event_backlog_window_seconds() == 60.0 + + +# --------------------------------------------------------------------------- +# Throttle decision +# --------------------------------------------------------------------------- + + +def test_disabled_when_limit_non_positive(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(svc._EVENT_BACKLOG_LIMIT_ENV, "0") + manager = _fake_manager(existing_key=None, backlog=10_000) + assert _throttle(manager) is None + manager.count_recent_event_backlog.assert_not_called() + + +def test_idempotent_redelivery_not_throttled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(svc._EVENT_BACKLOG_LIMIT_ENV, "1") + # An instance already exists for this exact event key: it must resolve to + # that instance, not be suppressed, even with a huge backlog. + manager = _fake_manager(existing_key="instance-existing", backlog=10_000) + assert _throttle(manager) is None + manager.count_recent_event_backlog.assert_not_called() + + +def test_below_limit_allows_creation(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(svc._EVENT_BACKLOG_LIMIT_ENV, "50") + manager = _fake_manager(existing_key=None, backlog=49) + assert _throttle(manager) is None + + +def test_at_or_above_limit_throttles(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(svc._EVENT_BACKLOG_LIMIT_ENV, "50") + manager = _fake_manager(existing_key=None, backlog=50) + result = _throttle(manager) + assert result is not None + assert result.success is False + assert result.status == "throttled_event_backlog" + assert result.error_code == "event_backlog_throttled" + assert result.instance_id is None + assert result.created_new is False + + +def test_throttle_query_is_time_bounded(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(svc._EVENT_BACKLOG_LIMIT_ENV, "10") + monkeypatch.setenv(svc._EVENT_BACKLOG_WINDOW_ENV, "3600") + manager = _fake_manager(existing_key=None, backlog=0) + before = datetime.now(timezone.utc) + _throttle(manager) + kwargs = manager.count_recent_event_backlog.call_args.kwargs + assert kwargs["workflow_id"] == "#V#paper_recommendation_evaluation_workflow" + assert kwargs["source_event_type"] == "text_relation.upserted" + assert tuple(kwargs["statuses"]) == svc._EVENT_BACKLOG_PENDING_STATUSES + # since_utc must be a bounded lower bound in the past, never unbounded. + assert kwargs["since_utc"] < before + assert (before - kwargs["since_utc"]).total_seconds() == pytest.approx( + 3600, abs=30 + ) diff --git a/tests/backend/test_workflow_instance_submission_service.py b/tests/backend/test_workflow_instance_submission_service.py index 5beff2832..77e95d0a7 100644 --- a/tests/backend/test_workflow_instance_submission_service.py +++ b/tests/backend/test_workflow_instance_submission_service.py @@ -1245,6 +1245,45 @@ def test_submit_verified_workflow_instance_uses_event_idempotent_creation() -> N assert mock_verify.call_count == 2 +def test_submit_verified_workflow_instance_throttles_event_backlog( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """JVNAUTOSCI-2507: a new event-triggered submission is suppressed once an + unprocessed backlog for the (workflow, event-type) pair exceeds the limit.""" + monkeypatch.setenv("VON_WORKFLOW_EVENT_BACKLOG_LIMIT", "5") + manager = MagicMock() + manager.find_instance_id_by_event_key.return_value = None # genuinely new event + manager.count_recent_event_backlog.return_value = 5 # at the limit + verification = _make_verification() + definition = _make_definition(include_action=True) + + with patch( + "src.backend.workflows.durable.workflow_instance_submission_service.verify_workflow_runnable", + side_effect=[verification, verification], + ), patch( + "src.backend.workflows.durable.registry_factory.get_shared_workflow_registry_read_only", + return_value=_make_registry(definition), + ): + result = submit_verified_workflow_instance( + manager=manager, + workflow_id="#V#candidate_workflow", + user_id="#V#user_alice", + org_id="#V#org_nao", + namespace="#V#user_alice/#V#org_nao", + inputs={"seed": "abc-123"}, + source_event_type="text_relation.upserted", + source_event_id="rel-1", + event_idempotency_key="evt:text_relation.upserted:rel-1", + ) + + assert result.success is False + assert result.status == "throttled_event_backlog" + assert result.error_code == "event_backlog_throttled" + assert result.created_new is False + manager.create_instance_for_event.assert_not_called() + manager.create_instance.assert_not_called() + + def test_submit_verified_workflow_instance_preserves_idempotent_reuse_without_postflight_failure() -> None: manager = MagicMock() manager.create_instance_for_event.return_value = ("instance-evt-existing", False)