From b183fef8314251beabebf348dd8a260082ec5870 Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Thu, 4 Jun 2026 03:05:29 -0300 Subject: [PATCH] feat(workflows): persist submit request so re-enqueued QUEUED jobs replay original inputs --- .../services/background_execution/service.py | 18 ++- .../test_facade_hard_proof.py | 118 ++++++++++++++++++ 2 files changed, 132 insertions(+), 4 deletions(-) diff --git a/src/backend/base/langflow/services/background_execution/service.py b/src/backend/base/langflow/services/background_execution/service.py index 5ddbf4ae08..d7ce241c02 100644 --- a/src/backend/base/langflow/services/background_execution/service.py +++ b/src/backend/base/langflow/services/background_execution/service.py @@ -90,6 +90,11 @@ class BackgroundExecutionService(Service): user_id=user.id, dedupe_key=dedupe_key, ) + # Persist the submit request on the job row so a QUEUED job that survives + # a restart is re-enqueued with its ORIGINAL inputs (input_value, tweaks, + # globals, etc.), not a reconstructed default. The startup sweep reads it + # back via ``_reconstruct_request``. + await job_service.update_job_metadata(job_id, {"request": request}) await self._enqueue(job_id=job_id, flow_id=flow_id, request=request, user=user) return job_id @@ -214,13 +219,18 @@ class BackgroundExecutionService(Service): @staticmethod def _reconstruct_request(job: Job) -> dict[str, Any]: - """Rebuild the minimal request dict for a re-enqueued QUEUED job. + """Rebuild the request dict for a re-enqueued QUEUED job. - The original request body is not persisted on the job row, so re-enqueue - runs with defaults (langflow protocol, no chat input). ``job_metadata`` - may carry the original protocol/session if a future task persists it. + ``submit`` persists the original request body under + ``job_metadata["request"]`` so re-enqueue replays the ORIGINAL inputs + (input_value, tweaks, globals, files, partial-run ids, ...). Falls back + to a minimal default only for legacy rows written before the request was + persisted, so a pre-existing QUEUED job still re-runs rather than blocks. """ meta = job.job_metadata or {} + persisted = meta.get("request") + if isinstance(persisted, dict) and persisted: + return persisted return { "flow_id": str(job.flow_id), "mode": "background", diff --git a/src/backend/tests/unit/background_execution/test_facade_hard_proof.py b/src/backend/tests/unit/background_execution/test_facade_hard_proof.py index 07e0535859..b703b37691 100644 --- a/src/backend/tests/unit/background_execution/test_facade_hard_proof.py +++ b/src/backend/tests/unit/background_execution/test_facade_hard_proof.py @@ -140,3 +140,121 @@ async def test_sweep_orphans_fails_in_progress(hard_proof_job_service): assert job.status == JobStatus.FAILED assert job.error is not None assert job.error.get("type") == "worker_lost" + + +def _echo_input_factory(*, request, **_kwargs): + """Frame source that echoes ``request['input_value']`` into a durable event. + + Proves the re-enqueued QUEUED job replays its ORIGINAL inputs: the input it + actually runs with shows up in the durable ``job_events`` log, so a restart + that lost the in-memory request body would surface a different (defaulted) + value here. + """ + input_value = request.get("input_value") + + async def _source(**_kw): + # ``add_message`` is a durable langflow event, so it lands in job_events + # and survives for the post-run assertion. + yield _frame("add_message", {"input_value": input_value}) + yield _frame("end", {}) + + return _source + + +async def test_submit_persists_request_for_faithful_requeue(hard_proof_job_service): + """``submit`` persists the request body on the job row (job_metadata.request). + + This is the durable record the startup sweep reads to replay original inputs. + Proven on both real SQLite and real Postgres. The executor is stopped right + after submit so no run interferes with reading the persisted row. + """ + from langflow.services.background_execution.service import BackgroundExecutionService + from langflow.services.deps import get_settings_service + + job_service = hard_proof_job_service + original_input = f"original-input-{uuid4()}" + flow_id = uuid4() + + svc = BackgroundExecutionService( + settings_service=get_settings_service(), + frame_source_factory=_echo_input_factory, + ) + request = { + "flow_id": str(flow_id), + "mode": "background", + "stream_protocol": "langflow", + "input_value": original_input, + "session_id": "thread-restart", + "tweaks": {"ChatInput-x": {"foo": "bar"}}, + } + job_id = await svc.submit(flow_id=flow_id, request=request, user=_StubUser(uuid4())) + await svc.stop() + + job = await job_service.get_job_by_job_id(job_id) + assert job.job_metadata is not None + persisted = job.job_metadata.get("request") + # The whole request round-trips, not just input_value. + assert persisted == request + + +async def test_requeued_queued_job_replays_original_input(hard_proof_job_service): + """A QUEUED job that survives a restart re-runs with its ORIGINAL input. + + Models "queued, never started, then the process died": the durable row is + QUEUED with the request persisted exactly as ``submit`` writes it (create_job + + update_job_metadata(request)). A *fresh* facade (empty in-memory state, as + after a restart) sweeps and re-enqueues, and the run replays the ORIGINAL + ``input_value`` — visible in the durable ``job_events`` log. On both real + SQLite and real Postgres. + """ + import asyncio + + from langflow.services.background_execution.service import BackgroundExecutionService + from langflow.services.deps import get_settings_service + + job_service = hard_proof_job_service + original_input = f"original-input-{uuid4()}" + user_id = uuid4() + flow_id = uuid4() + job_id = uuid4() + request = { + "flow_id": str(flow_id), + "mode": "background", + "stream_protocol": "langflow", + "input_value": original_input, + "session_id": "thread-restart", + } + # Durable state a restart finds: a QUEUED row carrying the persisted request, + # written exactly the way ``submit`` writes it. + await job_service.create_job(job_id=job_id, flow_id=flow_id, user_id=user_id) + await job_service.update_job_metadata(job_id, {"request": request}) + + # Restart: a brand-new facade with empty in-memory state sweeps + re-enqueues. + restart_svc = BackgroundExecutionService( + settings_service=get_settings_service(), + frame_source_factory=_echo_input_factory, + ) + await restart_svc.start() + try: + await restart_svc.sweep_orphans_on_startup() + job = None + for _ in range(100): + job = await job_service.get_job_by_job_id(job_id) + if job.status == JobStatus.COMPLETED: + break + await asyncio.sleep(0.05) + assert job.status == JobStatus.COMPLETED + finally: + await restart_svc.stop() + + # The re-enqueued run replayed the ORIGINAL input, not a default. + events = await job_service.read_events(job_id, after_seq=0) + echoed = [e.payload.get("data", {}).get("input_value") for e in events if e.event_type == "add_message"] + assert echoed == [original_input] + + +class _StubUser: + """Minimal user carrying only ``id`` (all the facade submit path reads).""" + + def __init__(self, user_id): + self.id = user_id