diff --git a/src/backend/base/langflow/services/background_execution/runner.py b/src/backend/base/langflow/services/background_execution/runner.py index f800e2cbda..64648b221e 100644 --- a/src/backend/base/langflow/services/background_execution/runner.py +++ b/src/backend/base/langflow/services/background_execution/runner.py @@ -111,8 +111,41 @@ class JobRunner: with contextlib.suppress(Exception): if await asyncio.shield(self._reconcile_stop(job_id)): await logger.adebug(f"Background job {job_id} reconciled to CANCELLED after a racing stop") + # Every terminal path writes result/error + a terminal job_events row + # (design §8). COMPLETED/FAILED already do via _drive; TIMED_OUT and + # CANCELLED do not, so backfill their error blob + terminal milestone + # here (after the stop reconcile so a late-stop CANCELLED is included). + with contextlib.suppress(Exception): + await asyncio.shield(self._finalize_terminal_event(job_id)) await self._bus.close(str(job_id)) + async def _finalize_terminal_event(self, job_id: UUID) -> None: + """Backfill the error blob + terminal event for TIMED_OUT / CANCELLED. + + ``execute_with_status`` writes the TIMED_OUT/CANCELLED status but no + durable error blob or terminal ``job_events`` row, so a consumer keying + on a terminal event TYPE (not just stream close) would not find one. This + appends ``run_timed_out`` / ``run_cancelled`` and, for a CANCELLED row + that a racing completion left with a populated ``result``, clears the + result and sets ``error={type: cancelled}`` so the terminal state is + internally consistent (no CANCELLED row carrying a completed-run result). + """ + job = await self._jobs.get_job_by_job_id(job_id) + if job is None: + return + if job.status == JobStatus.TIMED_OUT: + if job.error is None: + await self._jobs.set_error(job_id, {"type": "timed_out"}) + await self._jobs.append_event(job_id, "run_timed_out", {"type": "timed_out"}) + elif job.status == JobStatus.CANCELLED: + # A late stop that won over a genuine completion may have left a + # completed-run result behind — overwrite it so the row is consistent. + if job.result is not None: + await self._jobs.set_result(job_id, None) + if job.error is None: + await self._jobs.set_error(job_id, {"type": "cancelled"}) + await self._jobs.append_event(job_id, "run_cancelled", {"type": "cancelled"}) + def _start_heartbeat(self, job_id: UUID) -> asyncio.Task | None: """Spawn the periodic heartbeat task for a run (None when owner unset). diff --git a/src/backend/base/langflow/services/jobs/service.py b/src/backend/base/langflow/services/jobs/service.py index da481ee9de..892da534c5 100644 --- a/src/backend/base/langflow/services/jobs/service.py +++ b/src/backend/base/langflow/services/jobs/service.py @@ -225,9 +225,13 @@ class JobService(Service): await session.flush() return job - async def set_result(self, job_id: UUID, result: dict) -> Job | None: + async def set_result(self, job_id: UUID, result: dict | None) -> Job | None: """Persist the durable terminal result blob for a job. + ``None`` clears a previously-written result (used when a late stop + reconciles a racing completion to CANCELLED and the completed-run result + must not linger on the terminal row). + Returns the updated Job, or None if the row does not exist. """ async with session_scope() as session: diff --git a/src/backend/tests/unit/services/background_execution/test_runner.py b/src/backend/tests/unit/services/background_execution/test_runner.py index 1151795a5b..6a4a73a4dd 100644 --- a/src/backend/tests/unit/services/background_execution/test_runner.py +++ b/src/backend/tests/unit/services/background_execution/test_runner.py @@ -99,6 +99,91 @@ async def test_runner_finalizes_failed_and_writes_error(active_user): assert job.error is not None +async def test_runner_timeout_writes_terminal_event_and_error(active_user): + """TIMED_OUT must write an error blob AND a terminal job_events row (design §8).""" + job_service = get_job_service() + job_id = await _make_job(uuid4(), active_user.id) + + async def source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]: + yield _frame("build_start", {}) + # Run past the job timeout so asyncio.wait_for raises TimeoutError. + await asyncio.sleep(5) + yield _frame("end", {}) + + bus = InMemoryLiveBus() + adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t")) + runner = JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source, job_timeout=0.2) + + await asyncio.wait_for(runner.run(job_id=job_id, source_kwargs={}), timeout=10) + + job = await job_service.get_job_by_job_id(job_id) + assert job.status == JobStatus.TIMED_OUT + # Error blob is populated on timeout (was NULL before the fix). + assert job.error is not None + assert job.error.get("type") == "timed_out" + # A terminal milestone is on the durable log. + events = await job_service.read_events(job_id) + assert any(e.event_type == "run_timed_out" for e in events) + + +async def test_runner_cancel_writes_terminal_event(active_user): + """CANCELLED must write a terminal job_events row (design §8).""" + job_service = get_job_service() + job_id = await _make_job(uuid4(), active_user.id) + + from langflow.services.database.models.jobs.model import SignalType + + gate = asyncio.Event() + + async def source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]: + yield _frame("build_start", {}) + await gate.wait() + yield _frame("end_vertex", {"id": "n1"}) + yield _frame("end", {}) + + bus = InMemoryLiveBus() + adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t")) + runner = JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source) + + run_task = asyncio.create_task(runner.run(job_id=job_id, source_kwargs={})) + await asyncio.sleep(0.1) + await job_service.write_signal(job_id, SignalType.STOP) + gate.set() + await asyncio.wait_for(run_task, timeout=5) + + job = await job_service.get_job_by_job_id(job_id) + assert job.status == JobStatus.CANCELLED + events = await job_service.read_events(job_id) + assert any(e.event_type == "run_cancelled" for e in events) + + +async def test_late_stop_after_completion_clears_result(active_user): + """A stop reconciled to CANCELLED after a genuine completion must not keep result.""" + job_service = get_job_service() + job_id = await _make_job(uuid4(), active_user.id) + + from langflow.services.database.models.jobs.model import SignalType + + async def source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]: + # The flow completes cleanly (set_result + COMPLETED), but a STOP lands in + # the race window before the runner's terminal reconcile. + await job_service.write_signal(job_id, SignalType.STOP) + yield _frame("end", {}) + + bus = InMemoryLiveBus() + adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t")) + runner = JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source) + + await runner.run(job_id=job_id, source_kwargs={}) + + job = await job_service.get_job_by_job_id(job_id) + # Stop wins over the racing completion... + assert job.status == JobStatus.CANCELLED + # ...and the terminal state is internally consistent: no completed result blob. + assert job.result is None + assert (job.error or {}).get("type") == "cancelled" + + async def test_runner_stops_at_signal_boundary(active_user): job_service = get_job_service() job_id = await _make_job(uuid4(), active_user.id)