feat(api/v2): durable + distributed background execution service for workflows - #13505
Closed
ogabrielluiz wants to merge 78 commits into
Closed
feat(api/v2): durable + distributed background execution service for workflows#13505ogabrielluiz wants to merge 78 commits into
ogabrielluiz wants to merge 78 commits into
Conversation
…te/postgres/redis fixtures, make + CI)
…s; stabilize stop e2e
…ree under concurrency)
…play original inputs
…make resume contract deterministic
The worker used asyncio.Task.cancelling() to tell a job-cancel apart from a worker-cancel, but that API is Python 3.11+. On 3.10 the first job cancellation raised AttributeError inside the cancel handler and permanently killed the worker, hanging later jobs QUEUED. Replace it with an explicit _closed flag set by stop(): a CancelledError while shutting down re-raises to exit the worker, otherwise it is a per-job cancel that is swallowed so the pool keeps serving.
…tic (direct claim race)
… harness teardown to prevent state leak
…v run' child python cannot leak
Submits POST /api/v2/workflows with mode=background, polls GET /api/v2/workflows?job_id=... to a terminal status, and records a success only when the job reaches COMPLETED. Tracks submitted vs completed counts in the quitting summary so the run can be cross-checked against the job table for lost or stuck jobs.
Lease+heartbeat liveness in job_metadata (no new column): heartbeat() stamps owner+timestamp via shallow merge; is_lease_stale() tells a live in-flight run from a genuinely orphaned one; increment_attempt_if() is an atomic conditional bump so concurrent reconcilers cannot push a job past max_attempts.
create_job's dedupe count was global, so a client-controlled idempotency_key let user A collide with / DoS user B's key and the 500 leaked its existence. Scope the count by user_id (ownerless rows keep a global floor for AUTO_LOGIN), matching the already-user-scoped lookup. Defense-in-depth: map DuplicateJobError to a 409 with a generic body in execute_workflow_background so the residual create/lookup race cannot 500-leak the key.
sweep_orphans only fails IN_PROGRESS rows whose heartbeat is stale/absent, so a booting worker can no longer flip a sibling's actively-running, freshly- heartbeated job FAILED(worker_lost) under gunicorn -w N. The runner now refreshes a job-row heartbeat (owner+ts) on an interval while in flight. The startup sweep is single-flighted with a FileLock (the starter-projects pattern), and its terminal-event insert reuses append_event so a seq collision can't roll back the whole sweep.
…UEUED-strand recovery requeue_lost only reconciles processing-list ids whose lease is stale/absent, so a booting/scaled-up worker can no longer re-claim and double-run (or fail) a job a live worker is mid-running. A periodic watchdog runs requeue_lost on an interval inside run_worker_loop so a dead worker's in-flight job is reaped under a steady fleet WITHOUT a restart. The worker stamps a heartbeat on claim and the in-flight runner keeps it fresh. Retry-safe attempt accounting is now an atomic conditional increment (increment_attempt_if) so concurrent reconcilers cannot exceed max_attempts; the LREM count is the single-flight token so two watchdogs cannot double-enqueue. recover_stranded_queued re-enqueues QUEUED rows present on neither redis list (API-crash window) so a job is never stuck QUEUED forever.
…esult consistency execute_with_status writes the TIMED_OUT/CANCELLED status but no durable error blob or terminal job_events row, breaking the design invariant that every terminal path writes result/error + a terminal event. The runner now backfills run_timed_out (+error) / run_cancelled in its finally. A late stop that wins over a racing completion clears the completed-run result and sets error=cancelled so a CANCELLED row never carries a completed result.
…ing IN_PROGRESS The startup re-enqueue flipped a QUEUED row to IN_PROGRESS at claim time, so a crash before the runner started left a stranded IN_PROGRESS the next sweep failed worker_lost -- a job that never ran ended FAILED. claim_queued_lease stamps a single-flight owner+heartbeat WITHOUT changing status, so the row stays QUEUED (re-runnable) and the runner's execute_with_status performs the real QUEUED->IN_PROGRESS flip only when it actually starts.
…roofs Drain the claim queue after a concurrent-reconcile race and assert the lost job is claimable exactly once (a side-effect-style run counter), so a draining worker re-runs it at most once more, never twice.
…e frames The live SSE stream baked its own per-frame stream-seq counter into the id: line while durable replay used job_events.seq, two different namespaces. A client reconnecting with a live Last-Event-ID resumed against the wrong cursor and silently missed or duplicated milestones. The runner now re-stamps every published frame's id: with its durable seq (the row seq for a milestone, the last milestone's seq for an ephemeral token) so live ids and durable replay ids share one cursor and a mid-run reattach resumes exactly after the durable seq.
…ch seam The scaled events() replayed durable job_events from the DB then tailed the redis Stream from 0-0 with no seq filter, so every milestone the worker wrote to BOTH the DB and the Stream was delivered twice on reattach. Track the highest durable seq replayed and skip any Stream frame whose worker-stamped seq is <= highest, mirroring the in-memory bus's dedup-at-the-seam rule, so each milestone is delivered exactly once on both paths.
_row_to_frame serialized durable payloads with json.dumps default (spaced) separators, but the agui live path uses pydantic model_dump_json (compact separators), so replayed agui frames were not byte-identical to live frames. Pass the run's stream protocol (read off the persisted submit request) into _row_to_frame and pick compact separators for agui, spaced for langflow, so replay == live byte-for-byte on both wires.
…r job close() set _closed[job_id]=True and nothing ever evicted it, so a long-lived API process leaked one key per completed job. The marker only needs to outlive close() until existing subscribers drain (a late reattach to a finished job is gated on the persisted JobStatus by the facade, not this in-process flag), so set it only when subscribers exist and evict it with the last subscriber drop.
…down In scaled mode the facade built a redis client for the backend but teardown() only stopped the executor, leaking the API replica's background-execution connection pool on shutdown. Add a backend teardown() that closes its client and call it from the facade teardown (the worker process already closes its own).
…r-XADD RedisStreamLiveBus called expire() on every XADD, doubling redis round-trips per token frame and capping single-job streaming throughput. Mirror the v1 Streams bridge: refresh the TTL on the first frame, then every 100 frames or 30s, and always on close, preserving the TTL semantics at ~1/100 the EXPIRE rate.
… terminate The first _closed bound evicted the marker on last-subscriber drop and skipped marking when no subscriber existed, which broke a direct reattach to a finished job that closed with no subscriber (it blocked on a tail that never produces). Always record close() and bound the map with an LRU cap instead, so the standalone reattach-terminates contract holds while a long-lived process still cannot leak one key per job.
…s the path stop() set a redis cancel marker and PUBLISHed on a cancel channel as a fast path, but the background worker (run_worker_loop -> WorkerJobRunner -> JobRunner) never subscribes to that channel or checks the marker — only the v1 RedisJobQueueService dispatcher does, which the worker does not run. So the marker + PUBLISH were no-ops in production, a misleading dead fast-path. Remove them: the durable STOP signal polled at each vertex boundary is the single mechanism. Add a real-redis latency test proving scaled stop is bounded by the durable poll cadence, and drop the v1-dispatcher fast-path test that gave false confidence.
submit persisted the full request body on the durable job row for faithful replay, but request-level globals can carry inline secrets (API keys), landing plaintext in the job table (JSONB on Postgres) and widening the blast radius of any DB read. Redact globals from the persisted copy (the live in-memory run still uses them). Tradeoff: a background re-enqueue and a scaled worker run drop inline globals, so reference stored global variables by name for background runs rather than passing secrets inline.
…ng QUEUED atomically
Contributor
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
|
✅ Migration Validation Passed All migrations follow the Expand-Contract pattern correctly. |
This was referenced Jun 4, 2026
Contributor
Author
Contributor
Author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Builds on #13307. Turns the v2
mode: backgroundstub into an actual background execution service: durable, decoupled, and distributed, behind oneBackgroundExecutionServicefacade so the endpoint never knows which backend is running.What it does
The old
mode: backgroundran the flow as a fire-and-forgetasyncio.create_taskin the API worker, buffered SSE frames in a process-local dict, and persisted only job metadata. A restart killed the run, reattach/stop only worked on the worker that served the POST, and a completed job's output vanished. This replaces that with one facade over five small primitives (store / queue / runner / live bus / control) and two wirings:execution_signalsDB row, and a liveness-aware single-flight startup sweep. State, result, error, and a seq-ordered durable milestone log are all persisted, so a reattach replays the run and a completed job's result is fetchable forever.langflow workerprocess claims jobs off a redis list, runs the same runner, and publishes live frames to a redis Stream so any API replica can reattach. A lease + heartbeat + periodic watchdog reconciles a dead worker without needing a restart.Same
WorkflowJobResponseand SSE /Last-Event-IDcontract as today. The change is additive: a client that works now keeps working, it just stops losing jobs.Data model
Three migrations:
result/errorcolumns onjob, a newjob_eventstable (the durable milestone log,UNIQUE(job_id, seq),seqis theLast-Event-IDcursor), and a newexecution_signalstable (cross-worker stop without redis). No newJobStatusvalues.Semantics
At-least-once for not-yet-started (QUEUED) work, at-most-once for in-flight work by default (a crashed run becomes FAILED, not silently re-run, since flows are not generally idempotent), and opt-in bounded retry per flow. Durable result/error plus a terminal
job_eventsrow on every terminal path, including TIMED_OUT and CANCELLED. Cooperative stop at vertex boundaries.What's proven
Everything was built TDD against real Postgres + real redis (no fakes on the durability/distribution paths). The full suite is 268 passing on real instances. Beyond unit/integration:
langflow workerOS subprocess claims, builds, and completes a real graph while the API side reattaches; akill -9mid-job is reconciled by the watchdog; a durable stop reaches the separate worker.There were three rounds of review along the way; the findings (a Python 3.10 cancellation bug, a restart reattach hang, a multi-worker double-run in the orphan recovery, a
Last-Event-IDseq namespace mismatch, a cross-tenant idempotency key collision, and an attempt-cap race) are all fixed with regression tests.Notes / follow-ups
JobStatus) and the seq-based event log are shaped so a later resume-from-checkpoint (the feat: graph lifecycle + checkpointing and execution resume #12387 direction) can drop in.globalsare redacted from the persisted re-enqueue request, so a scaled worker run or a restart re-run uses named stored variables rather than secrets passed inline. Documented in the design doc.RedisJobQueueServiceStreams/owner/watchdog machinery, so it lines up with thefeat_redis_job_queue/add-workerstracks.Design doc and the TDD implementation plan are in my notes, happy to share if useful.