diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index 4efa0aa9..27f22ca5 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -7,6 +7,7 @@ routing, retries, and history all aligned with the rest of the app. """ import asyncio +import time import logging from datetime import datetime, timedelta, timezone from typing import Optional @@ -21,6 +22,13 @@ logger = logging.getLogger(__name__) # In-process map: workflow_id -> currently running run id. Prevents two overlapping fires for the same workflow (e.g. cron tick races a manual Run button) without serializing across the whole executor. _running: dict[str, str] = {} +# Global admission on top of the per-workflow guard: every run is a full agent, and an agent's +# browsers and apps are exempt from the renderer budget by design (sleeping a working agent's browser +# blinds it), so the ONLY thing bounding total pressure is how many runs exist at once. A library of +# 30 workflows whose schedules drift into alignment must queue, not stampede. +MAX_CONCURRENT_RUNS = 3 +ADMISSION_WAIT_S = 600.0 +ADMISSION_POLL_S = 2.0 _running_lock = asyncio.Lock() @@ -211,6 +219,25 @@ async def execute( # the scheduler, an agent tool, an invoke, a retry, the Run Now route, or a stale in-flight handle. # Guarding this per call site left every unguarded caller able to fire it, which is the field report # of a toggled-off workflow running itself. Turn it back on to run it. + # Wait for a global slot BEFORE the off-means-off guard, so the guard runs on fresh state after + # a possibly long wait (a workflow paused while queueing still gets refused, not run). + p_admit_start = time.monotonic() + while len(_running) >= MAX_CONCURRENT_RUNS: + if time.monotonic() - p_admit_start >= ADMISSION_WAIT_S: + p_busy = WorkflowRun( + workflow_id=wf.id, status="skipped", + error=f"{MAX_CONCURRENT_RUNS} workflows already running; gave up after {int(ADMISSION_WAIT_S)}s", + scheduled_for=scheduled_for, started_at=datetime.now(), + finished_at=datetime.now(), triggered_by=triggered_by, + ) + p_wf_now = storage.get_workflow(wf.id) + if p_wf_now is not None and p_wf_now.deleted_at is None: + try: + storage.record_run(p_busy) + except Exception: + logger.debug("could not record the admission-skip row", exc_info=True) + return p_busy + await asyncio.sleep(ADMISSION_POLL_S) p_live = storage.get_workflow(wf.id) p_refusal = None if p_live is None: diff --git a/backend/tests/test_workflow_run_admission.py b/backend/tests/test_workflow_run_admission.py new file mode 100644 index 00000000..f002abf4 --- /dev/null +++ b/backend/tests/test_workflow_run_admission.py @@ -0,0 +1,102 @@ +"""No stampede: a due fire beyond the global cap queues, and gives up honestly, never silently. + +Every workflow run is a full agent, and a working agent's browsers and apps are exempt from the +renderer budget on purpose (sleeping them blinds the agent). So the ONLY bound on total pressure is +how many runs exist at once, and before this the scheduler would happily start one agent per due +workflow: thirty accumulated workflows drifting into schedule alignment meant thirty agents. +""" + +import asyncio + +import pytest + +from backend.apps.workflows import executor, storage +from backend.apps.workflows.models import Workflow, WorkflowStep + + +def p_make() -> Workflow: + return Workflow( + title="queued", + steps=[WorkflowStep(text="do it")], + schedule={"enabled": True, "kind": "interval", "every_minutes": 45}, + ) + + +@pytest.fixture(autouse=True) +def p_isolated(tmp_path, monkeypatch): + monkeypatch.setattr(storage, "DATA_DIR", str(tmp_path / "workflows")) + monkeypatch.setattr(storage, "RUNS_DIR", str(tmp_path / "workflows" / "runs")) + monkeypatch.setattr(storage, "PAUSED_FILE", str(tmp_path / "workflows" / "paused.json")) + monkeypatch.setattr(storage, "MISSED_FILE", str(tmp_path / "workflows" / "missed.json")) + monkeypatch.setattr(storage, "_workflow_cache", {}) + monkeypatch.setattr(storage, "_runs_cache", {}) + monkeypatch.setattr(storage, "_missed_cache", []) + monkeypatch.setattr(storage, "p_deleted_ids", set(), raising=False) + monkeypatch.setattr(storage, "_cache_loaded", True) + monkeypatch.setattr(executor, "_running", {}) + yield + + +def test_at_the_cap_a_fire_waits_then_skips_with_an_honest_row(monkeypatch): + monkeypatch.setattr(executor, "MAX_CONCURRENT_RUNS", 2) + monkeypatch.setattr(executor, "ADMISSION_WAIT_S", 0.05) + monkeypatch.setattr(executor, "ADMISSION_POLL_S", 0.01) + monkeypatch.setattr(executor, "_running", {"other-a": "r1", "other-b": "r2"}) + + wf = storage.save_workflow(p_make()) + run = asyncio.run(executor.execute(wf, triggered_by="schedule")) + + assert run.status == "skipped" + assert "already running" in (run.error or "") + rows = storage.list_runs(wf.id) + assert len(rows) == 1 and rows[0].status == "skipped", "the give-up must be visible in History" + + +def test_a_freed_slot_lets_the_queued_fire_proceed(monkeypatch): + """The wait is a queue, not a rejection: the moment a slot frees, the run goes ahead and reaches + the normal guard path (here: refused as paused, which proves it got past admission).""" + monkeypatch.setattr(executor, "MAX_CONCURRENT_RUNS", 1) + monkeypatch.setattr(executor, "ADMISSION_WAIT_S", 5.0) + monkeypatch.setattr(executor, "ADMISSION_POLL_S", 0.01) + monkeypatch.setattr(executor, "_running", {"other-a": "r1"}) + + wf = storage.save_workflow(p_make()) + live = storage.get_workflow(wf.id) + live.schedule.enabled = False + storage.save_workflow(live) + + async def scenario(): + async def free_slot_soon(): + await asyncio.sleep(0.05) + executor._running.clear() + asyncio.ensure_future(free_slot_soon()) + return await executor.execute(wf, triggered_by="schedule") + + run = asyncio.run(scenario()) + assert run.status == "skipped" + assert "paused" in (run.error or "").lower(), ( + f"expected the post-wait guard to refuse the paused workflow, got {run.error!r}" + ) + + +def test_the_guard_runs_on_state_AFTER_the_wait(monkeypatch): + """A workflow deleted while queueing must be refused, not run: admission sits before the + off-means-off guard precisely so the guard sees post-wait truth.""" + monkeypatch.setattr(executor, "MAX_CONCURRENT_RUNS", 1) + monkeypatch.setattr(executor, "ADMISSION_WAIT_S", 5.0) + monkeypatch.setattr(executor, "ADMISSION_POLL_S", 0.01) + monkeypatch.setattr(executor, "_running", {"other-a": "r1"}) + + wf = storage.save_workflow(p_make()) + + async def scenario(): + async def delete_then_free(): + await asyncio.sleep(0.05) + storage.delete_workflow(wf.id) + executor._running.clear() + asyncio.ensure_future(delete_then_free()) + return await executor.execute(wf, triggered_by="schedule") + + run = asyncio.run(scenario()) + assert run.status == "skipped" + assert "delete" in (run.error or "").lower()