diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index 8d8ce71f3..ffce4af6d 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -183,6 +183,25 @@ class ScheduledTaskService: return self._active_run_conflict_result(execution_thread_id) return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger) + # Global concurrent-run budget check for manual triggers. The poller + # enforces max_concurrent_runs via count_active_runs() before + # claim_due_tasks(); a manual trigger bypasses that path and must apply + # the same cap so it cannot push the active count above the limit. + # Like the poller's count this is a non-atomic fast path; the partial + # unique index uq_scheduled_task_run_active is the atomic arbiter that + # rejects a second active insert for the *same task*, but there is no + # DB-level constraint that caps the global count, so we treat this as a + # best-effort guard consistent with how the poller enforces the budget. + if trigger == "manual" and self._max_concurrent_runs > 0: + active = await self._task_run_repo.count_active_runs() + if active >= self._max_concurrent_runs: + return { + "outcome": "conflict", + "task_run_id": None, + "run_id": None, + "thread_id": execution_thread_id, + "error": "global concurrent-run limit reached", + } if self._multi_instance and trigger == "manual": task = await self._task_repo.claim_dispatch_lease( task["id"], diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py index cafc33bf8..4874a0fb9 100644 --- a/backend/tests/test_scheduled_task_service.py +++ b/backend/tests/test_scheduled_task_service.py @@ -1001,3 +1001,63 @@ async def test_malformed_launch_result_still_retains_active_slot(): first_row_id = run_repo.created[0]["run_record_id"] assert run_repo.rows[first_row_id]["status"] == "running" + + +@pytest.mark.asyncio +async def test_manual_trigger_rejected_when_global_budget_exhausted(): + """Manual trigger must return a conflict when max_concurrent_runs is reached.""" + launched = [] + + async def fake_launch(**kwargs): + launched.append(kwargs) + return {"run_id": "run-budget", "thread_id": kwargs["thread_id"]} + + row = _once_task_row(task_id="task-budget", status="enabled") + row.update({"schedule_type": "cron", "schedule_spec": {"cron": "* * * * *"}, "overlap_policy": "skip"}) + task_repo = DummyTaskRepo([row]) + # active_count equals max_concurrent_runs → budget is exhausted + run_repo = DummyRunRepo(active_count=3) + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + + result = await service.dispatch_task(row, now=datetime.now(UTC), trigger="manual") + + assert result["outcome"] == "conflict" + assert "limit" in result["error"] + assert launched == [] + assert run_repo.created is None + + +@pytest.mark.asyncio +async def test_manual_trigger_proceeds_when_global_budget_available(): + """Manual trigger must launch when active count is below max_concurrent_runs.""" + launched = [] + + async def fake_launch(**kwargs): + launched.append(kwargs) + return {"run_id": "run-ok", "thread_id": kwargs["thread_id"]} + + row = _once_task_row(task_id="task-ok", status="enabled") + row.update({"schedule_type": "cron", "schedule_spec": {"cron": "* * * * *"}, "overlap_policy": "skip"}) + task_repo = DummyTaskRepo([row]) + # active_count is 2, max_concurrent_runs is 3 → one slot left + run_repo = DummyRunRepo(active_count=2) + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + + result = await service.dispatch_task(row, now=datetime.now(UTC), trigger="manual") + + assert result["outcome"] == "launched" + assert len(launched) == 1