feat(api): DB-backed scaled background execution backend + langflow worker - #13508
feat(api): DB-backed scaled background execution backend + langflow worker#13508ogabrielluiz wants to merge 62 commits into
Conversation
Rebased onto release-1.10.0. The base independently rebuilt the v2 workflows backend (RBAC, body globals, share-aware fetch); keep our forward design and conform its auth to that work: 1. Auth: keep get_current_user_for_workflow (session-or-API-key authN that does not hold a DB connection during the inline run, avoiding the SQLite lock contention api_key_security would cause) and enforce the base's RBAC on top: ensure_flow_permission(EXECUTE) before run, (READ) before status reconstruct, with widen_for_shares fetch. 2. Port the base's request-body globals onto the v2 WorkflowRunRequest. The X-LANGFLOW-GLOBAL-VAR-* headers stay supported (the Responses API passes globals that way); body globals win on conflict. Converters echo the effective globals via effective_globals. 3. Public endpoint keeps the v1 build_public_tmp posture (access_type==PUBLIC, run-as-owner); RBAC applies to the authenticated endpoint only. 4. Preserve the base's post-build KB-cache invalidation in the AG-UI build path. The endpoint, AG-UI bridge, pluggable stream adapters, public endpoint, and re-attach are unchanged.
The synchronous /api/v2/workflows response keyed every result under its component id, so reading the answer meant knowing an id you can't predict. Surface two additive fields: - output_text: the flow's single text answer (ChatOutput/TextOutput). None when the flow has zero or multiple text outputs, so callers read outputs rather than the shortcut guessing which channel is the answer. - session_id: echoes the resolved session so chat/memory callers can continue the same thread (v1 /run returned this; v2 had dropped it). outputs is unchanged, so this is non-breaking.
…ponse
Pin the sync-response shortcuts on the v2 workflows endpoint:
- output_text surfaces the lone ChatOutput/TextOutput text and stays None for
non-output message nodes, data-only flows, and multi-text flows
- session_id echoes the resolved session; the error response exposes neither
- each outputs entry exposes only {type, status, content, metadata}, with the
component id carried by the dict key
Also drop the component_id kwarg the converter passed to ComponentOutput, which
has no such field and silently dropped it.
Replace the flat output_text shortcut with an `output` object carrying the resolved text answer plus a `reason` that explains why it resolved that way (single/multiple/none/non_string/failed), so a null answer is always diagnosable instead of silently None. `reason` follows the LLM-domain finish_reason/stop_reason convention, distinct from the lifecycle status. Also add `display_name` to each ComponentOutput (the stable component id stays the dict key) and a computed `has_errors` flag derived from errors.
Let a sync caller name the output(s) they want via output_ids so output.text resolves deterministically (reason=single) on multi-output flows instead of going null. Selection is steer-only: it picks the answer among the named outputs without filtering the outputs map. Invalid ids are rejected with 422 before the flow runs (and before any job row is created), so a typo costs no compute. Resolution considers selected outputs that actually fired, so branching flows resolve to whichever candidate ran.
Give v2-workflows sync and the langflow stream protocol one parser. The stream now emits a normalized "output" event per terminal output carrying an OutputEvent (the ComponentOutput shape sync returns in outputs[id], plus component_id). A shared build_component_output() backs both the sync converter and the adapter, and the build loop ships authoritative vertex metadata as an additive output_meta key on end_vertex (existing consumers read build_data and ignore it). This is access-pattern parity (one parser, same fields, same terminal set), not byte-identical content: the stream reuses the v1 build path whose display serialization differs from sync's run_graph output.
…ackend) Turns v2 mode:background into a durable, in-API background execution service behind a BackgroundExecutionService facade. Adds the store layer (result/error columns, job_events durable milestone log, execution_signals control, heartbeat/lease, 3 migrations), the default backend (bounded executor, runner, in-memory live bus, liveness-aware single-flight orphan sweep), the v2 endpoint rewiring, and the real-instance test harness. Needs no new infra; works on the SQLite single-process install. The redis-scaled worker backend is stacked on top in a follow-up PR.
WalkthroughThe PR replaces Redis-based scaled background execution with a PostgreSQL-backed queue. It adds explicit backend settings, lease-based workers, durable event streaming, watchdog recovery, a worker CLI command, migrations, load testing, and real-service validation. ChangesScaled background execution
Estimated code review effort: 4 (Complex) | ~60 minutes Suggested labels: Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant API as BackgroundExecutionService
participant DB as PostgreSQL
participant Worker as langflow worker
participant Runner as WorkerJobRunner
API->>DB: Store queued job
Worker->>DB: Claim lease
Worker->>Runner: Hydrate and execute job
Runner->>DB: Store status and events
API->>DB: Replay durable events
🚥 Pre-merge checks | ✅ 6 | ❌ 3❌ Failed checks (3 warnings)
✅ Passed checks (6 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 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 |
|
✅ Migration Validation Passed All migrations follow the Expand-Contract pattern correctly. |
github-actions
Bot
added
enhancement
The hard_proof marker name was a vibe word that said nothing about what the tests need. Rename it to real_services everywhere: the pytest marker registration, the *_hard_proof.py test files, the Makefile target (real_services_tests), the -m selector in migration-validation.yml, and the CI job. real_services says what these tests require: real Postgres + Redis + worker subprocesses. (integration was already taken for the external-API suite under tests/integration.)
…rker Stacks on the default backend. Adds the opt-in redis backend: a langflow worker CLI claiming jobs off a redis list, a redis Streams live bus for cross-replica reattach, and a lease+heartbeat+periodic watchdog (liveness-aware reconcile, atomic retry accounting). Same facade; default behavior unchanged when no redis is configured. Proven on real redis incl. a real worker subprocess, kill-9 watchdog reconcile, and a Locust load test (22,841 req / 0 failures / ~2.5x throughput vs in-process). Includes the v2 background locustfile.
…ting prints The events.quitting listener took _environment as a required positional arg, but locust passes environment as a keyword, so the handler raised TypeError at shutdown and the submitted/completed summary never printed (locust exited 2). Accept the event kwargs via **_kwargs.
… in scaled worker tests Follows the marker rename on the base PR: updates the scaled worker subprocess tests and the subprocess harness to the real_services marker.
e2ccbaf to
2f961f5
Compare
github-actions
Bot
added
enhancement
…kflows-agui # Conflicts: # src/frontend/src/stores/flowStore.ts
The v2 public endpoint only ran validate_flow_for_current_settings and skipped validate_public_flow_no_code_execution, which the v1 build_public_tmp path applies. A public flow containing a Python interpreter/REPL (or the legacy Python Code Structured tool, Smart Transform lambda) was therefore an unauthenticated server-side code-execution primitive (report H1-3754930). Mirror v1: import the validator and call it right after the public-access gate. PublicFlowValidationError subclasses CustomComponentValidationError, so the existing handler already sanitizes it to a 400 'This flow cannot be executed.' without leaking the blocked component class names. Add a non-mocking test that builds a public flow with a real PythonREPLComponent and asserts the sanitized 400 (verified RED: returns 200 without the gate). LE-1389
…rtex builds A completed background job's GET status 500'd with 'No vertex builds found for job_id'. The background build path differed from the sync path twice: 1. generate_flow_events minted a fresh run_id instead of using job_id, so vertex builds were keyed by an id the status query never uses. Thread run_id through _stream_event_frames -> generate_flow_events and pass job_id from the background buffer so graph.run_id == job_id (the sync path already does graph.set_run_id(job_id)). 2. The SSE build loop (build_vertices) only persisted builds when log_builds was set and never passed job_id. Tie log_builds to job-tracked runs (run_id present) and pass job_id=graph.run_id on the persist call. Job-tracked runs also persist streaming terminal vertices so reconstruction is complete; the live build path (run_id is None) keeps its original behavior, so the v1 build path is unchanged. Test: a real background run polled to completion, then GET status asserts a reconstructed 200 (verified RED: 500 'No vertex builds found' before the fix). Covers the non-streaming flow. v1 build path unchanged (35 build tests pass); AG-UI suite 46 pass. LE-1389
Adopt 1.11.0's composed group-mixin Settings for lfx settings/base.py (all v2 settings already present in the new groups); port the Python 3.14 cors_origins ['*']->'*' fix into groups/security.py.
…ows-bg-default Port the 6 background-execution settings and background_backend_is_scaled into 1.11.0's RuntimeSettings group mixin; union the BACKGROUND_EXECUTION_SERVICE / TELEMETRY_WRITER_SERVICE additions in schema.py and deps.py.
…workflows-bg-scaled
…thon 3.10 The bounded executor's worker awaited the in-flight job task with a bare `await task`. On Python 3.10, when stop() cancels a worker while its job task is finishing, the awaiter's wakeup is lost and the event loop idles forever in select(), deadlocking stop(). Await via a done-callback Event (the same mechanism stop()'s own asyncio.gather already uses), which delivers the wakeup reliably; task.result() preserves the cancellation and exception semantics of the bare await.
|
@ogabrielluiz PR Review — scaled background execution (Redis claim queue + Streams live bus + DB replay)Overall this is careful work — the durable-milestone/Stream dedup is correct and well-tested, the lease/watchdog reconciliation is thought through, and the cross-replica reattach contract is sound. My concerns are about how it behaves on a single shared Redis cluster used by the whole fleet (every 🔴 Blocking1. Fix: co-locate the two keys with a hash tag, e.g., 2. A single global pending key caps the entire fleet's throughput on one shard. Shard the claim queue into N logical lists, each with its own hash tag ( 3. Keys aren't namespaced, which is the only isolation boundary on a shared cluster. 🟡 Should fix4. Every executor runs its own reconciler against the same shard + DB. 5. 6. Heartbeat writes to the durable store. 7. Shared-cluster memory / noisy neighbor. 8. 🟢 Minor/follow-up9. 10. In-process state in the live bus. 11. At least once retry appends a second run's milestones. A retried job continues 12. No startup health-check / fail-fast. 13. Cosmetic. Dead Summary: #1 and #2 are the ones I'd want resolved before this runs on the shared cluster — both are fixed by the same change (hash-tagged, sharded, deployment-namespaced queue keys, ideally behind a queue abstraction so Redis is swappable), which also restores logical isolation (#3) when Redis itself isn't isolated. After that, moving reconciliation to a single leader (#4) and leases to the cache (#6) are the next most impactful. Happy to pair on the key scheme — I can sketch the sharded/hash-tagged layout and a queue port interface that wraps the current implementation as the first adapter if useful. |
|
Upon investigating the setup, here is a quick excerpt distilled to explain the cross-slot issue failing claims and how we can support a sharded cluster setup with dedicated consumer groups: Point 1 in plain termsThe one-sentence version: the To decide which shard owns a key, Redis doesn't use random assignment. It defines exactly 16,384 "slots" and gives each shard a contiguous range of them: When you read or write a key, Redis computes:
The rule that bites usA command that touches multiple keys (like If the keys land on different slots, Redis doesn't try to be clever — it just rejects the command: Why this code trips it
Two different key names → two different CRC16 values → almost certainly two different slots: So on a real cluster, every claim fails and no worker can pick up a job. (On a single non-clustered Redis there are no slots to span, so it works — which is why this can pass in dev and break in production.) The fix: hash tagsRedis gives you an escape hatch. If a key name contains text inside The The sharding problem, visualizedHere's the subtlety that a single shared cluster (scaled up for capacity) makes important: fixing CROSSSLOT and getting good throughput pull in opposite directions. Here are both states. State 1 — after the hash-tag fix (correct, but everything on one shard)We forced Correctness is fine, but the whole fleet's claim traffic funnels through one shard. Adding shards (scaling the cluster up) does nothing — slots A and C sit idle because no key ever routes there. You've bought a 3-machine cluster and you're using 1 machine for the queue. This is the "single hot key" ceiling. State 2 — sharded queue (correct AND uses the whole cluster)Instead of one queue, create N queues, each with its own hash tag, and have each pod pick one (by pod number, or at random). Crucially, each queue's own pending+processing still share a tag, so each individual Two invariants at once:
Why the event streams are already fineWorth noting the per-job event streams don't have this problem, because each one is a single key touched by single-key commands ( So the streams already do what we want; it's only the two-key claim queue that needs the hash-tag-plus-sharding treatment. The mental model to keep
|
Brings the v2 workflow router seam (#13816) and the rest of release onto the durable background execution branch. Same rule as the stacked branches: take release's structure and hygiene, keep the durable semantics on top. api/v2/workflow.py is now release's seam skeleton (resolve_flow_for_execution, authorize_flow_action, run_sync_with_mapping, build_stream_response) with the durable route bodies (status reconstruct with the Job.result fallback, the stop signal ordering, service.events replay) and the durable facade on background submit. Helpers moved to release's workflow_execution.py. Two things the plain merge would have dropped: - Release's workflow_execution.py does not pass tweaks to generate_flow_events, so taking it wholesale silently reverted 02e06fe (apply request tweaks on the streaming and background paths). Restored. - idempotency_key never survived the parse boundary: ParsedWorkflowRun did not carry the field, so the dedupe and DuplicateJobError -> 409 could never fire through the seam. Threaded through. Durable runs also key vertex builds by job_id (run_id=str(job_id)) so a completed job's status reconstructs its outputs and recovers the session_id instead of falling back to the leaner Job.result rebuild. Migration chain is re-parented onto release's head, so there is a single alembic head.
github-actions
Bot
added
enhancement
There was a problem hiding this comment.
Pull request overview
Adds an opt-in Redis-scaled background execution backend plus a langflow worker CLI process so background workflows execute out-of-process and can be distributed across replicas, while keeping the API surface behind the existing BackgroundExecutionService facade.
Changes:
- Implemented a Redis-backed claim queue + worker loop with periodic lease/watchdog reconciliation and DB+Stream reattach semantics.
- Added a Redis Streams live bus producer for cross-replica event tailing with batched TTL refresh.
- Added extensive real-Redis / real-Postgres / real-subprocess proofs and a Locust load test for background execution.
Reviewed changes
Copilot reviewed 32 out of 32 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| src/backend/base/langflow/main.py | Adds langflow worker CLI command wiring services + worker loop. |
| src/backend/base/langflow/services/background_execution/worker.py | Implements worker job runner, claim loop, and periodic watchdog. |
| src/backend/base/langflow/services/background_execution/redis_backend.py | Adds scaled backend: claim queue, durable replay, Stream tail + dedup, stop via durable signals, watchdog reconcile. |
| src/backend/base/langflow/services/background_execution/redis_queue.py | Adds BRPOPLPUSH-based claim queue with processing list semantics. |
| src/backend/base/langflow/services/background_execution/redis_live_bus.py | Adds Redis Streams producer (XADD) with batched TTL refresh. |
| src/backend/tests/unit/services/background_execution/_subprocess_harness.py | Adds real-process harness for spawning langflow worker and seeding real flows. |
| src/backend/tests/unit/services/background_execution/test_build_worker.py | Tests build_worker wiring and shared redis client between backend and bus. |
| src/backend/tests/unit/services/background_execution/test_control_fastpath.py | Verifies stop writes only durable STOP signal (no pub/sub marker). |
| src/backend/tests/unit/services/background_execution/test_facade_teardown_realredis.py | Tests scaled facade teardown closes the redis client. |
| src/backend/tests/unit/services/background_execution/test_factory_selection.py | Tests backend factory selection based on scaled settings. |
| src/backend/tests/unit/services/background_execution/test_lease_watchdog.py | Unit-tests watchdog reconciliation semantics (fakeredis). |
| src/backend/tests/unit/services/background_execution/test_live_bus_ttl_realredis.py | Proves TTL refresh is batched (real Redis). |
| src/backend/tests/unit/services/background_execution/test_periodic_watchdog_realredis.py | Proves periodic watchdog reaps stale leases without restarts (real Redis/DB). |
| src/backend/tests/unit/services/background_execution/test_queued_strand_realredis.py | Proves recovery of stranded QUEUED rows not present on redis lists. |
| src/backend/tests/unit/services/background_execution/test_reattach_dedup_realredis.py | Proves reattach dedup across DB replay + Stream tail (exactly-once milestones). |
| src/backend/tests/unit/services/background_execution/test_reattach_realredis.py | Proves DB replay + Stream tail works cross-replica (real Redis). |
| src/backend/tests/unit/services/background_execution/test_redis_backend.py | Unit-tests enqueue + replay behavior (fakeredis + fake JobService). |
| src/backend/tests/unit/services/background_execution/test_redis_claim_queue.py | Unit-tests enqueue/claim/complete FIFO semantics (fakeredis). |
| src/backend/tests/unit/services/background_execution/test_redis_claim_queue_realredis.py | Proves BRPOPLPUSH blocking claim semantics + disjoint workers (real Redis). |
| src/backend/tests/unit/services/background_execution/test_redis_fixture.py | Documents/validates real-Redis fixture skip/roundtrip behavior. |
| src/backend/tests/unit/services/background_execution/test_redis_live_bus_realredis.py | Proves published stream frames are consumable by a reattaching backend (real Redis). |
| src/backend/tests/unit/services/background_execution/test_scaled_end_to_end_realredis.py | End-to-end scaled submit → worker run → cross-replica reattach (real Redis + DB). |
| src/backend/tests/unit/services/background_execution/test_scaled_liveness_realredis.py | Proves lease-aware liveness guard + concurrent reconcile correctness (real Redis + DB). |
| src/backend/tests/unit/services/background_execution/test_scaled_stop_latency_realredis.py | Proves STOP latency bounded by vertex-boundary polling cadence. |
| src/backend/tests/unit/services/background_execution/test_scaled_stop_realredis.py | Proves durable STOP leads to CANCELLED and signal consumption (real Redis + DB). |
| src/backend/tests/unit/services/background_execution/test_worker_death_realredis.py | Proves watchdog recovery behavior when worker dies after claim (real Redis + DB). |
| src/backend/tests/unit/services/background_execution/test_worker_loop.py | Tests worker claim loop behavior and lease release on runner errors. |
| src/backend/tests/unit/services/background_execution/test_worker_runner.py | Tests WorkerJobRunner hydrates a durable job and drives JobRunner to terminal state. |
| src/backend/tests/unit/services/background_execution/test_worker_subprocess_e2e.py | Real subprocess E2E: worker completes job; kill -9 reconciled by watchdog. |
| src/backend/tests/unit/services/background_execution/test_worker_subprocess_stop.py | Real subprocess proof: API-side stop_job cancels a job in a separate worker. |
| src/backend/tests/unit/background_execution/test_side_effect_safety.py | Proves at-most-once vs retry-safe semantics with a real side-effecting component. |
| src/backend/tests/locust/v2_background_locustfile.py | Adds Locust load test for v2 background workflows submit/poll accounting. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| @pytest.mark.real_services | ||
| async def test_real_worker_subprocess_honors_pubsub_stop(): | ||
| """API-side stop_job CANCELs a job running in a SEPARATE worker OS process.""" | ||
| pg, redis = _require_real_instances() |
| * Real Postgres (``LANGFLOW_TEST_DATABASE_URI``) as the durable store, shared | ||
| between the test-side API facade and the worker subprocess. | ||
| * Real Redis (``LANGFLOW_TEST_REDIS_URL``, a dedicated DB index) as the claim | ||
| queue + Streams live bus + cancel pub/sub. | ||
| * A real no-LLM flow (ChatInput -> ChatOutput) seeded into the shared DB, so the | ||
| worker's PRODUCTION ``_default_frame_source_factory`` builds and runs a real | ||
| graph — not a scripted source. |
| def test_real_redis_skips_without_env(real_redis_url): | ||
| # real_redis_url returns None when env is unset; documents the skip contract. | ||
| if real_redis_url is None: | ||
| pytest.skip("LANGFLOW_TEST_REDIS_URL not set") | ||
| assert real_redis_url.startswith("redis://") or os.environ.get("LANGFLOW_TEST_REDIS_URL") |
| # Capture the live pooled connection so we can prove it was disconnected. | ||
| pooled = client.connection_pool._available_connections[0] | ||
| assert pooled._writer is not None | ||
|
|
github-actions
Bot
added
enhancement
|
On the CROSSSLOT blocker — I think there's a fix that's structural rather than a hash-tag patch, and it also happens to answer the single-shard cap and the "swappable port" concern at the same time: move the claim queue from lists to Streams with a consumer group. The claim protocol becomes: Why this kills CROSSSLOT for good: every one of those commands touches exactly one key. There is no multi-key operation left to span slots, so the error isn't "fixed" — it becomes unrepresentable. A hash tag on On the single-shard cap: shard into N streams, each with its own tag — The part I'd want to get right — this must not quietly become at-least-once. The PR deliberately picks at-most-once for in-flight work, and
So the lease/heartbeat/attempt accounting doesn't go away — Redis stops owning the recovery trigger, and the DB keeps owning the recovery decision. What comes for free: the PEL gives per-consumer attribution and idle time via Happy to help with the |
The durable job table becomes the work queue itself: submit persists the QUEUED row and separate langflow worker processes lease-claim it off the shared database via the existing exact-heartbeat conditional UPDATE, so no broker is involved anywhere. Cross-replica reattach replays job_events and polls the same table for new durable milestones; stop stays the durable ExecutionSignal row. Deleting the broker deletes its failure modes: there is no pending or processing list to drift from the DB (stranded-QUEUED recovery has no equivalent, a QUEUED row is enqueued by definition) and no LREM token protocol (the conditional UPDATE is the single flight). The lease watchdog now scans stale IN_PROGRESS rows directly, keeping the same at-most-once default and opt-in bounded retry semantics. Backend selection is the explicit background_backend setting (default or scaled) instead of riding job_queue_type=redis; scaled needs nothing but the database the API already uses. The real-service proofs (a real worker OS subprocess completing a real graph, kill -9 reconciliation, cross-process durable stop) now run against Postgres alone, and the former real-redis test fleet runs against the ordinary test DB.
# Conflicts: # .github/workflows/migration-validation.yml # .secrets.baseline # src/backend/base/langflow/api/v2/workflow.py # src/backend/base/langflow/services/background_execution/executor.py # src/backend/base/langflow/services/background_execution/factory.py # src/backend/base/langflow/services/background_execution/live_bus.py # src/backend/base/langflow/services/background_execution/runner.py # src/backend/base/langflow/services/background_execution/service.py # src/backend/base/langflow/services/database/models/__init__.py # src/backend/base/langflow/services/database/models/jobs/__init__.py # src/backend/base/langflow/services/database/models/jobs/model.py # src/backend/base/langflow/services/jobs/service.py # src/backend/tests/unit/api/v2/test_workflow_background.py # src/backend/tests/unit/background_execution/conftest.py # src/backend/tests/unit/background_execution/test_head_to_head_deltas.py # src/backend/tests/unit/background_execution/test_real_services_ci_wiring.py # src/backend/tests/unit/background_execution/test_real_services_harness.py # src/backend/tests/unit/services/background_execution/test_live_bus.py # src/backend/tests/unit/services/background_execution/test_runner.py # src/backend/tests/unit/services/background_execution/test_service.py # src/lfx/src/lfx/services/settings/groups/runtime.py # src/lfx/src/lfx/workflow/adapters/langflow.py # src/lfx/tests/unit/services/settings/test_background_scaled_settings.py # src/lfx/tests/unit/services/settings/test_settings_composition.py
Index the claim poll: composite (status, type, created_timestamp) on the job table, declared on the model and added in a guarded EXPAND migration — the claim SELECT and watchdog scan run per worker poll on a table with no retention, so an unindexed scan grows without bound. Operational hardening from the production-lens review: log once at facade wiring when scaled mode is active (a fleet with zero workers otherwise queues silently), state honestly that inline request globals are dropped on every scaled run, document the single-host limit of scaled + SQLite, end the event tail on SUSPENDED (a paused run produces no more frames until resume), and describe the grace-pass window and the deliberately client-bounded QUEUED tail in the events docstring.
ogabrielluiz
changed the title
github-actions
Bot
added
enhancement
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (20)
src/backend/tests/unit/services/background_execution/_subprocess_harness.py (3)
157-161: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse
asyncio.get_running_loop()inside the coroutine.
get_event_loop()is the legacy accessor and its behavior outside a running loop is deprecated. This code always runs inside a running loop, so callget_running_loop()once and reuse it.♻️ Proposed change
- import asyncio - - deadline = asyncio.get_event_loop().time() + timeout + import asyncio + + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout last = None - while asyncio.get_event_loop().time() < deadline: + while loop.time() < deadline:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/_subprocess_harness.py` around lines 157 - 161, Update the coroutine’s deadline loop to obtain the running event loop once with asyncio.get_running_loop(), then reuse that loop for both time checks instead of repeatedly calling asyncio.get_event_loop().
191-201: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueMove the blocking process waits off the event loop.
proc.wait(timeout=10)andproc.wait(timeout=5)are synchronous. In an async teardown they block the loop for up to 15 seconds per process, which also stalls any other pending task. Wrap them withasyncio.to_thread.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/_subprocess_harness.py` around lines 191 - 201, Update the async teardown flow around the two proc.wait calls in the process cleanup loop to execute each blocking wait via asyncio.to_thread, preserving the existing timeouts, exception suppression, and SIGKILL fallback behavior.
174-186: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
drain_worker_outputterminates live workers as a side effect.The function is documented as best-effort diagnostics, but it sends
terminate()to any running worker so thatread()can reach EOF. A caller that drains output for context therefore changes the test state. The file-redirect change proposed forspawn_workerremoves the need for this. Until then, rename the method or state the side effect in the docstring.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/_subprocess_harness.py` around lines 174 - 186, The drain_worker_output method currently terminates live worker processes as a diagnostic side effect. Remove the proc.terminate() behavior so draining output does not alter test state, relying on the file-redirect output path introduced by spawn_worker; preserve the existing best-effort collection and fallback behavior.src/backend/tests/unit/services/background_execution/test_worker_loop.py (1)
53-53: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBound the loop await so a regression fails instead of hanging.
Both tests rely on the runner to set
stop_event. If a change breaks that path,run_worker_loopnever returns and the test blocks until the CI job times out. Wrap the call withasyncio.wait_for.💚 Proposed change
- await run_worker_loop(backend, runner, stop_event=stop_event, idle_block_ms=10) + await asyncio.wait_for( + run_worker_loop(backend, runner, stop_event=stop_event, idle_block_ms=10), timeout=10.0 + )Also applies to: 72-72
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/test_worker_loop.py` at line 53, Bound the await of run_worker_loop in both affected tests with asyncio.wait_for, using an appropriate timeout so the tests fail promptly if stop_event is never set. Preserve the existing backend, runner, stop_event, and idle_block_ms arguments.src/backend/base/langflow/__main__.py (1)
865-872: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueCheck the backend setting before you initialize services.
initialize_services()runs first, then the command exits when the backend is notscaled. The early exit leaves the initialized services without teardown. The setting is readable fromget_settings_service()before initialization, so the guard can run first and the process can exit without side effects.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/__main__.py` around lines 865 - 872, In _run_worker, retrieve and validate get_settings_service().settings.background_backend_is_scaled before calling initialize_services(). Preserve the existing message and typer.Exit(code=1) for non-scaled backends, and only initialize services after the guard passes.src/backend/base/langflow/services/background_execution/worker.py (6)
161-162: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winGuard the startup reconcile so a transient database error does not abort worker boot.
The periodic pass at Line 127 suppresses errors, but the startup pass does not. If the database is briefly unreachable while the worker starts, this raises and the
langflow workerprocess exits with a traceback before it claims any job. The periodic watchdog would have reconciled the same rows on its first interval.♻️ Proposed change
- # Startup reconcile: requeue work lost by a previously-crashed worker. - await backend.requeue_lost(lease_ttl_s=lease_ttl_s) + # Startup reconcile: requeue work lost by a previously-crashed worker. + # Best-effort: a transient database error must not stop the worker from + # booting, because the periodic watchdog reconciles the same rows. + try: + await backend.requeue_lost(lease_ttl_s=lease_ttl_s) + except Exception as exc: # noqa: BLE001 + await logger.aexception(f"Worker: startup reconcile failed: {exc}")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/worker.py` around lines 161 - 162, Guard the startup requeue_lost call in the worker boot flow so transient database errors are caught and logged without aborting worker startup. Match the existing error-suppression behavior used by the periodic reconciliation pass, while preserving the startup reconciliation attempt and allowing the worker to continue claiming jobs.
78-82: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMove the request/user hydration helpers out of the service class.
WorkerJobRunner.runreaches into two private statics ofBackgroundExecutionService. Both are pure functions over aJobrow. Move them to a module-level helper (for examplebackground_execution/hydrate.py) and let both the facade and the worker call the public function. This removes the twonoqa: SLF001suppressions and makes the contract explicit.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/worker.py` around lines 78 - 82, Move the request and user hydration logic currently exposed through BackgroundExecutionService._reconstruct_request and _user_stub into public module-level helpers in a dedicated hydration module. Update BackgroundExecutionService and WorkerJobRunner.run to call those helpers directly, preserving their existing behavior and removing both SLF001 suppressions.
176-198: 🩺 Stability & Availability | 🔵 TrivialDocument the shutdown drain semantics.
The loop checks
stop_eventonly between claims, so SIGTERM lets the current job finish. This is the correct choice for at-most-once work. Under an orchestrator with a fixed termination grace period, a long job is killed mid-flight and the row staysIN_PROGRESSuntil the watchdog reaps it. State the expected relationship between the deployment grace period andbackground_job_timeoutin the operator documentation, soterminationGracePeriodSecondscan be set accordingly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/worker.py` around lines 176 - 198, Document the worker shutdown behavior around the loop containing runner.run: stopping prevents new claims but allows the current job to finish, while forced termination may leave it IN_PROGRESS until the watchdog reaps it. In the operator documentation, state that terminationGracePeriodSeconds should be set longer than background_job_timeout so the configured grace period accommodates job completion and recovery.
206-206: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd the return type annotation.
build_workerreturns a 3-tuple, but the signature has no return type, so callers getAny. Annotate it so mypy checks the unpack at the call sites.♻️ Proposed change
-async def build_worker(*, owner: str | None = None): +async def build_worker( + *, owner: str | None = None +) -> tuple[Any, WorkerJobRunner, Callable[[], Awaitable[None]]]:Add
Awaitableto theTYPE_CHECKINGimport fromcollections.abc.As per coding guidelines: "Python code must pass mypy type checking via
make lint".🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/worker.py` at line 206, Annotate the async build_worker function with its concrete three-tuple return type so mypy can validate callers’ unpacking. Add Awaitable to the TYPE_CHECKING collections.abc imports as suggested, and express the returned tuple’s element types using the existing worker symbols without changing runtime behavior.Source: Coding guidelines
109-132: 🩺 Stability & Availability | 🔵 TrivialConsider bounding watchdog concurrency across the fleet.
Every worker process runs this loop, so each
requeue_lostpass scans allIN_PROGRESSworkflow jobs on the shared database atwatchdog_interval_s. The reconcile writes are single-flight, so correctness holds. The read cost grows linearly with the worker count. With a large fleet and a short interval this becomes a repeated full scan of in-flight rows.Options: elect one reconciler (advisory lock or leader row), stagger the first pass with a random offset per worker, or scale the effective interval by fleet size.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/worker.py` around lines 109 - 132, Bound concurrent watchdog reconciliation across workers by adding coordination around the requeue_lost call in _watchdog_loop, using an advisory lock or leader-row mechanism so only one worker performs each scan at a time. Preserve the existing periodic loop, stop-event handling, and best-effort error suppression.
19-19: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePromote
_coerce_uuidto a shared public helper.
worker.pyimports a private symbol fromdb_backend.py. The worker is a separate module boundary, so this couples it to a private detail. Rename the helper tocoerce_uuidin a shared module (or re-export it) and import the public name.Also applies to: 187-189
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/worker.py` at line 19, Promote the private _coerce_uuid helper used by worker.py to the public name coerce_uuid in a shared module or via an explicit re-export from db_backend.py. Update the helper definition and every reference, including the import and usages in worker.py, so no cross-module code depends on the underscored symbol.src/backend/tests/unit/services/background_execution/test_worker_runner.py (1)
50-74: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for the error paths of
WorkerJobRunner.run.The test covers the success path only. Two reachable branches stay untested: an unknown
job_id(the runner logs and returns without raising) and a frame source that raises mid-stream (the durable row must reachFAILED). Both are short tests against the same realJobService.As per path instructions: tests should include "coverage for positive, negative, edge, and error cases".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/test_worker_runner.py` around lines 50 - 74, Extend the tests around WorkerJobRunner.run with separate cases for an unknown job_id and a frame source that raises during streaming. Assert the unknown-job path returns without raising, and verify the mid-stream failure updates the durable job row to JobStatus.FAILED using the same real JobService setup.Source: Path instructions
src/backend/tests/unit/services/background_execution/test_build_worker.py (1)
17-25: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert the settings-derived lease and poll wiring.
build_workerreadsbackground_lease_ttl_sandbackground_poll_interval_sfrom settings and passes them toDBBackgroundQueue. The test does not check those two values, so a swapped or dropped argument would still pass. Compare both againstget_settings_service().settings.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/test_build_worker.py` around lines 17 - 25, Extend the build_worker test assertions to validate the settings-derived lease and poll values passed to DBBackgroundQueue. Compare the backend’s corresponding lease-TTL and poll-interval attributes with get_settings_service().settings.background_lease_ttl_s and background_poll_interval_s, while preserving the existing backend, runner, teardown, bus, and owner assertions.src/backend/tests/unit/services/background_execution/test_lease_watchdog.py (1)
28-29: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRedundant
@pytest.mark.asynciomarkers. Based on learnings, this repository configures pytest-asyncio withasyncio_mode = 'auto'inpyproject.toml, so async tests are auto-detected and the markers add no behavior.
src/backend/tests/unit/services/background_execution/test_lease_watchdog.py#L28-L29: remove the@pytest.mark.asynciodecorator here and on the other five tests in the file (Lines 48, 63, 84, 116, 140).src/backend/tests/unit/services/background_execution/test_worker_loop.py#L47-L47: remove the@pytest.mark.asynciodecorator here and on Line 59.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/test_lease_watchdog.py` around lines 28 - 29, Remove the redundant pytest.mark.asyncio decorators from all six async tests in src/backend/tests/unit/services/background_execution/test_lease_watchdog.py, including the tests anchored by the existing decorators and those at lines 48, 63, 84, 116, and 140. Also remove the decorators from both affected tests in src/backend/tests/unit/services/background_execution/test_worker_loop.py, at lines 47 and 59; leave the async test implementations unchanged.Source: Learnings
src/backend/tests/unit/services/background_execution/test_control_fastpath.py (1)
13-15: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the redundant
@pytest.mark.asynciodecorator.
pyproject.tomlsetsasyncio_mode = "auto", so async tests are detected automatically and this marker is unnecessary.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/test_control_fastpath.py` around lines 13 - 15, Remove the redundant `@pytest.mark.asyncio` decorator from test_stop_writes_durable_signal, while keeping the client fixture marker and test function unchanged.Source: Learnings
src/backend/tests/unit/services/background_execution/test_db_backend.py (1)
18-18: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDrop the redundant
pytest.mark.asyncio.This repository configures
asyncio_mode = 'auto'inpyproject.toml, so async tests are auto-detected withoutpytest.mark.asyncio. Other test files in this suite (for exampletest_lease_watchdog.py) define async test functions without this marker. Remove it frompytestmarkfor consistency.🧹 Proposed fix
-pytestmark = [pytest.mark.usefixtures("client"), pytest.mark.asyncio] +pytestmark = [pytest.mark.usefixtures("client")]Based on learnings, "pytest-asyncio is configured with asyncio_mode = 'auto' in pyproject.toml... you do not need to decorate test functions or classes with pytest.mark.asyncio."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/background_execution/test_db_backend.py` at line 18, Update the module-level pytestmark in test_db_backend.py to remove the redundant pytest.mark.asyncio entry, while preserving the client fixture marker and automatic async test detection.Source: Learnings
src/backend/base/langflow/services/jobs/service.py (1)
591-599: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid the N+1 read pattern in the watchdog scan.
in_progress_workflow_job_idsreturns onlyjob_idvalues. Its only consumer,DBBackgroundQueue.requeue_lostinsrc/backend/base/langflow/services/background_execution/db_backend.py, re-fetches eachJobindividually withget_job_by_job_idright after. Return fullJobrows here (select(Job).where(...)) instead of ids-only, so the watchdog scan does one query instead of1 + Nqueries per reconcile pass. This runs on every watchdog interval across the worker fleet, so the extra round trips scale with both job count and worker count.♻️ Proposed refactor
- async def in_progress_workflow_job_ids(self) -> list[UUID]: - """Return the ids of every IN_PROGRESS workflow job (for the watchdog).""" - async with session_scope() as session: - stmt = select(Job.job_id).where( - Job.status == JobStatus.IN_PROGRESS, - Job.type == JobType.WORKFLOW, - ) - result = await session.exec(stmt) - return list(result.all()) + async def in_progress_workflow_jobs(self) -> list[Job]: + """Return every IN_PROGRESS workflow Job row (for the watchdog).""" + async with session_scope() as session: + stmt = select(Job).where( + Job.status == JobStatus.IN_PROGRESS, + Job.type == JobType.WORKFLOW, + ) + result = await session.exec(stmt) + return list(result.all())🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/jobs/service.py` around lines 591 - 599, Update in_progress_workflow_job_ids to select and return full Job entities rather than only Job.job_id values, preserving the existing IN_PROGRESS and WORKFLOW filters. Adjust its return annotation and ensure DBBackgroundQueue.requeue_lost consumes the returned Job objects directly instead of re-fetching each job with get_job_by_job_id.src/backend/base/langflow/services/background_execution/db_backend.py (1)
91-109: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winMake the FIFO claim candidate window scale with the worker fleet.
claim_next_queued_lease's hardcodedcandidates=5default is never overridden anywhere in the call chain, even though its own docstring says the candidate count should be "slightly larger than the worker fleet." With more than ~5 concurrently polling workers, a claim round can returnNoneeven while further QUEUED rows exist, capping throughput in exactly the multi-worker deployments this PR targets.
src/backend/base/langflow/services/background_execution/db_backend.py#L91-L109: thread acandidatesvalue (e.g. a newDBBackgroundQueue.__init__parameter sourced from a setting) into theclaim_next_queued_lease(...)call instead of relying on the hardcoded default.src/backend/base/langflow/services/jobs/service.py#L601-L631: keepcandidatesas the tunable parameter it already is; the fix is exposing it through the call chain, not changing this signature.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/db_backend.py` around lines 91 - 109, Expose a configurable candidates value through DBBackgroundQueue.__init__ and pass it to claim_next_queued_lease from claim, sourcing it from the appropriate setting and defaulting safely for existing callers. In src/backend/base/langflow/services/background_execution/db_backend.py lines 91-109, update the queue configuration and claim call; in src/backend/base/langflow/services/jobs/service.py lines 601-631, make no direct change because its existing candidates parameter is already tunable and should remain unchanged.src/backend/tests/unit/background_execution/test_real_services_ci_wiring.py (1)
30-36: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAssert the absence of a redis service to pin the stated intent.
The comment states that no redis service belongs in this job. Add the negative assertion so a re-added broker service fails the test.
♻️ Proposed addition
assert any(img.startswith("postgres:") for img in images), images + assert not any("redis" in img for img in images), images🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/background_execution/test_real_services_ci_wiring.py` around lines 30 - 36, Update test_real_services_job_has_postgres_service to explicitly assert that no service image starts with “redis:”, preserving the existing PostgreSQL assertion and the stated real-services job wiring intent.src/backend/tests/locust/v2_background_locustfile.py (1)
136-147: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDo not send an extra request to record the deadline failure.
The block sends one more real GET under the same request name only to call
poll.failure. That extra sample changes the poll latency and count statistics. Fire a request event directly instead, or reuse the last poll context manager to mark the failure.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/locust/v2_background_locustfile.py` around lines 136 - 147, Update the deadline-exceeded handling in the locust task to avoid issuing another GET request to record the failure. Reuse the most recent poll response context to call failure, or fire the appropriate request event directly, while preserving the deadline_exceeded counter and failure reporting.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/backend/base/langflow/__main__.py`:
- Around line 848-850: Update the idle_block_ms Typer option to enforce a
minimum value of 1, so --idle-block-ms rejects zero and negative inputs while
preserving the existing default and help text.
In
`@src/backend/base/langflow/alembic/versions/4b8f2c6d9e1a_add_job_claim_scan_index.py`:
- Around line 32-35: Update the upgrade logic around the existing index check to
use PostgreSQL’s concurrent index creation outside the Alembic transaction,
while retaining a safe non-PostgreSQL fallback using the normal index operation.
Preserve the existing _INDEX name, job table, columns, uniqueness, and
idempotent existing-index check.
In `@src/backend/tests/locust/v2_background_locustfile.py`:
- Around line 1-16: The docstrings contain stale Redis references after the
database-backed migration. In
src/backend/tests/locust/v2_background_locustfile.py lines 1-16, describe scaled
mode as using the database-backed worker claim path; in
src/backend/tests/unit/background_execution/test_side_effect_safety.py lines
13-16, replace the real Redis backend wording with DBBackgroundQueue, matching
the instance constructed by the test.
- Around line 112-133: Guard the 200-response body parsing in the polling flow
around poll.json() so malformed or non-JSON responses are handled as poll errors
and increment poll_errors instead of escaping the with-block. Also make the
failure message in the fallback branch safe when poll.text is None by using an
empty-string fallback before slicing.
- Around line 78-85: The workflow request in the locust user should not pass
per-request timeout arguments to self.client.post or self.client.get. Set the
FastHttpUser class attributes network_timeout and connection_timeout to
REQUEST_TIMEOUT, then remove the timeout keyword from both request calls while
preserving their existing response handling.
In `@src/backend/tests/unit/background_execution/conftest.py`:
- Around line 37-40: Update the URL normalization logic around the raw URL
checks so psycopg2 URLs are not mistaken for psycopg URLs: match the exact async
psycopg dialect prefix, and normalize postgresql+psycopg2 URLs to the
async-compatible postgresql+psycopg form before returning. Preserve unchanged
URLs that already use the exact psycopg prefix.
In `@src/backend/tests/unit/services/background_execution/_subprocess_harness.py`:
- Around line 134-142: Update the subprocess setup around Popen and
drain_worker_output so worker stdout/stderr is redirected to a temporary log
file instead of an unread PIPE, preventing the worker from blocking on pipe
capacity. Track the file handle or path per process, and have
drain_worker_output read the available log-file tail for diagnostics without
terminating a live worker merely to reach EOF.
In `@src/backend/tests/unit/services/background_execution/test_scaled_stop.py`:
- Around line 76-98: Ensure both helper tasks always set stop_event in a finally
block and bound the worker-loop await with a 30-second timeout. In
src/backend/tests/unit/services/background_execution/test_scaled_stop.py lines
76-98, update drive_stop_then_wait and wrap run_worker_loop with
asyncio.wait_for; apply the same changes to stop_when_done and run_worker_loop
in
src/backend/tests/unit/services/background_execution/test_scaled_end_to_end.py
lines 71-89.
In `@src/lfx/tests/unit/services/settings/test_background_scaled_settings.py`:
- Around line 31-34: Extend test_background_poll_interval_default to verify that
LANGFLOW_BACKGROUND_POLL_INTERVAL_S values of 0 and a negative number are
rejected by Settings, preserving the existing assertions for the default and
valid 0.1 interval.
---
Nitpick comments:
In `@src/backend/base/langflow/__main__.py`:
- Around line 865-872: In _run_worker, retrieve and validate
get_settings_service().settings.background_backend_is_scaled before calling
initialize_services(). Preserve the existing message and typer.Exit(code=1) for
non-scaled backends, and only initialize services after the guard passes.
In `@src/backend/base/langflow/services/background_execution/db_backend.py`:
- Around line 91-109: Expose a configurable candidates value through
DBBackgroundQueue.__init__ and pass it to claim_next_queued_lease from claim,
sourcing it from the appropriate setting and defaulting safely for existing
callers. In
src/backend/base/langflow/services/background_execution/db_backend.py lines
91-109, update the queue configuration and claim call; in
src/backend/base/langflow/services/jobs/service.py lines 601-631, make no direct
change because its existing candidates parameter is already tunable and should
remain unchanged.
In `@src/backend/base/langflow/services/background_execution/worker.py`:
- Around line 161-162: Guard the startup requeue_lost call in the worker boot
flow so transient database errors are caught and logged without aborting worker
startup. Match the existing error-suppression behavior used by the periodic
reconciliation pass, while preserving the startup reconciliation attempt and
allowing the worker to continue claiming jobs.
- Around line 78-82: Move the request and user hydration logic currently exposed
through BackgroundExecutionService._reconstruct_request and _user_stub into
public module-level helpers in a dedicated hydration module. Update
BackgroundExecutionService and WorkerJobRunner.run to call those helpers
directly, preserving their existing behavior and removing both SLF001
suppressions.
- Around line 176-198: Document the worker shutdown behavior around the loop
containing runner.run: stopping prevents new claims but allows the current job
to finish, while forced termination may leave it IN_PROGRESS until the watchdog
reaps it. In the operator documentation, state that
terminationGracePeriodSeconds should be set longer than background_job_timeout
so the configured grace period accommodates job completion and recovery.
- Line 206: Annotate the async build_worker function with its concrete
three-tuple return type so mypy can validate callers’ unpacking. Add Awaitable
to the TYPE_CHECKING collections.abc imports as suggested, and express the
returned tuple’s element types using the existing worker symbols without
changing runtime behavior.
- Around line 109-132: Bound concurrent watchdog reconciliation across workers
by adding coordination around the requeue_lost call in _watchdog_loop, using an
advisory lock or leader-row mechanism so only one worker performs each scan at a
time. Preserve the existing periodic loop, stop-event handling, and best-effort
error suppression.
- Line 19: Promote the private _coerce_uuid helper used by worker.py to the
public name coerce_uuid in a shared module or via an explicit re-export from
db_backend.py. Update the helper definition and every reference, including the
import and usages in worker.py, so no cross-module code depends on the
underscored symbol.
In `@src/backend/base/langflow/services/jobs/service.py`:
- Around line 591-599: Update in_progress_workflow_job_ids to select and return
full Job entities rather than only Job.job_id values, preserving the existing
IN_PROGRESS and WORKFLOW filters. Adjust its return annotation and ensure
DBBackgroundQueue.requeue_lost consumes the returned Job objects directly
instead of re-fetching each job with get_job_by_job_id.
In `@src/backend/tests/locust/v2_background_locustfile.py`:
- Around line 136-147: Update the deadline-exceeded handling in the locust task
to avoid issuing another GET request to record the failure. Reuse the most
recent poll response context to call failure, or fire the appropriate request
event directly, while preserving the deadline_exceeded counter and failure
reporting.
In `@src/backend/tests/unit/background_execution/test_real_services_ci_wiring.py`:
- Around line 30-36: Update test_real_services_job_has_postgres_service to
explicitly assert that no service image starts with “redis:”, preserving the
existing PostgreSQL assertion and the stated real-services job wiring intent.
In `@src/backend/tests/unit/services/background_execution/_subprocess_harness.py`:
- Around line 157-161: Update the coroutine’s deadline loop to obtain the
running event loop once with asyncio.get_running_loop(), then reuse that loop
for both time checks instead of repeatedly calling asyncio.get_event_loop().
- Around line 191-201: Update the async teardown flow around the two proc.wait
calls in the process cleanup loop to execute each blocking wait via
asyncio.to_thread, preserving the existing timeouts, exception suppression, and
SIGKILL fallback behavior.
- Around line 174-186: The drain_worker_output method currently terminates live
worker processes as a diagnostic side effect. Remove the proc.terminate()
behavior so draining output does not alter test state, relying on the
file-redirect output path introduced by spawn_worker; preserve the existing
best-effort collection and fallback behavior.
In `@src/backend/tests/unit/services/background_execution/test_build_worker.py`:
- Around line 17-25: Extend the build_worker test assertions to validate the
settings-derived lease and poll values passed to DBBackgroundQueue. Compare the
backend’s corresponding lease-TTL and poll-interval attributes with
get_settings_service().settings.background_lease_ttl_s and
background_poll_interval_s, while preserving the existing backend, runner,
teardown, bus, and owner assertions.
In
`@src/backend/tests/unit/services/background_execution/test_control_fastpath.py`:
- Around line 13-15: Remove the redundant `@pytest.mark.asyncio` decorator from
test_stop_writes_durable_signal, while keeping the client fixture marker and
test function unchanged.
In `@src/backend/tests/unit/services/background_execution/test_db_backend.py`:
- Line 18: Update the module-level pytestmark in test_db_backend.py to remove
the redundant pytest.mark.asyncio entry, while preserving the client fixture
marker and automatic async test detection.
In `@src/backend/tests/unit/services/background_execution/test_lease_watchdog.py`:
- Around line 28-29: Remove the redundant pytest.mark.asyncio decorators from
all six async tests in
src/backend/tests/unit/services/background_execution/test_lease_watchdog.py,
including the tests anchored by the existing decorators and those at lines 48,
63, 84, 116, and 140. Also remove the decorators from both affected tests in
src/backend/tests/unit/services/background_execution/test_worker_loop.py, at
lines 47 and 59; leave the async test implementations unchanged.
In `@src/backend/tests/unit/services/background_execution/test_worker_loop.py`:
- Line 53: Bound the await of run_worker_loop in both affected tests with
asyncio.wait_for, using an appropriate timeout so the tests fail promptly if
stop_event is never set. Preserve the existing backend, runner, stop_event, and
idle_block_ms arguments.
In `@src/backend/tests/unit/services/background_execution/test_worker_runner.py`:
- Around line 50-74: Extend the tests around WorkerJobRunner.run with separate
cases for an unknown job_id and a frame source that raises during streaming.
Assert the unknown-job path returns without raising, and verify the mid-stream
failure updates the durable job row to JobStatus.FAILED using the same real
JobService setup.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: badef48e-1646-4e0c-bf40-046b82ef4a35
📒 Files selected for processing (34)
.github/workflows/migration-validation.ymlsrc/backend/base/langflow/__main__.pysrc/backend/base/langflow/alembic/versions/4b8f2c6d9e1a_add_job_claim_scan_index.pysrc/backend/base/langflow/services/background_execution/db_backend.pysrc/backend/base/langflow/services/background_execution/factory.pysrc/backend/base/langflow/services/background_execution/live_bus.pysrc/backend/base/langflow/services/background_execution/service.pysrc/backend/base/langflow/services/background_execution/worker.pysrc/backend/base/langflow/services/database/models/jobs/model.pysrc/backend/base/langflow/services/jobs/service.pysrc/backend/tests/locust/v2_background_locustfile.pysrc/backend/tests/unit/background_execution/conftest.pysrc/backend/tests/unit/background_execution/test_head_to_head_deltas.pysrc/backend/tests/unit/background_execution/test_real_services_ci_wiring.pysrc/backend/tests/unit/background_execution/test_real_services_harness.pysrc/backend/tests/unit/background_execution/test_side_effect_safety.pysrc/backend/tests/unit/services/background_execution/_subprocess_harness.pysrc/backend/tests/unit/services/background_execution/conftest.pysrc/backend/tests/unit/services/background_execution/test_build_worker.pysrc/backend/tests/unit/services/background_execution/test_control_fastpath.pysrc/backend/tests/unit/services/background_execution/test_db_backend.pysrc/backend/tests/unit/services/background_execution/test_factory_selection.pysrc/backend/tests/unit/services/background_execution/test_lease_watchdog.pysrc/backend/tests/unit/services/background_execution/test_scaled_end_to_end.pysrc/backend/tests/unit/services/background_execution/test_scaled_stop.pysrc/backend/tests/unit/services/background_execution/test_service.pysrc/backend/tests/unit/services/background_execution/test_standalone_slice.pysrc/backend/tests/unit/services/background_execution/test_worker_loop.pysrc/backend/tests/unit/services/background_execution/test_worker_runner.pysrc/backend/tests/unit/services/background_execution/test_worker_subprocess_e2e.pysrc/backend/tests/unit/services/background_execution/test_worker_subprocess_stop.pysrc/lfx/src/lfx/services/settings/groups/runtime.pysrc/lfx/tests/unit/services/settings/test_background_scaled_settings.pysrc/lfx/tests/unit/services/settings/test_settings_composition.py
💤 Files with no reviewable changes (1)
- src/backend/tests/unit/services/background_execution/conftest.py
| idle_block_ms: int = typer.Option( | ||
| 1000, help="Claim poll window in milliseconds (idle sleep between claim attempts)." | ||
| ), |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Typer typer.Option min max integer range validation
💡 Result:
In Typer, you can validate numeric ranges for CLI options and arguments by passing min and max parameters directly to typer.Option or typer.Argument [1][2]. Typer handles these constraints automatically, displaying an error message if the input falls outside the specified bounds [1][3]. Example usage: from typing import Annotated import typer def main( user_id: Annotated[int, typer.Option(min=1, max=1000)]): print(f"ID is {user_id}") If you want to automatically adjust out-of-bounds values to the nearest valid boundary (minimum or maximum) instead of raising an error, you can enable the clamp parameter [1][3]: def main( rank: Annotated[int, typer.Option(min=1, max=10, clamp=True)]): print(f"Rank is {rank}") Because Typer is built on top of Click, these numeric validations are implemented via Click's range handling [4][5]. While you can use Click's explicit click.IntRange or click.FloatRange types within Typer, the idiomatic Typer approach is to use the min, max, and clamp parameters directly within the Annotated type hint [1][2]. For more complex validation logic that falls outside of simple range checking, you can implement custom parameter callbacks [6].
Citations:
- 1: https://typer.tiangolo.com/tutorial/parameter-types/number/
- 2: https://typer.tiangolo.com/reference/parameters/
- 3: https://github.com/tiangolo/typer/blob/master/typer/params.py
- 4: https://github.com/pallets/click/blob/main/src/click/types.py
- 5: https://click.palletsprojects.com/en/stable/parameter-types/
- 6: https://github.com/pallets/click/blob/main/docs/advanced.md
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate file =="
fd -a '__main__.py$' . | sed 's#^\./##' | head -20
echo "== relevant lines =="
sed -n '830,865p' src/backend/base/langflow/__main__.py
echo "== dependency versions for typer/click =="
for f in pyproject.toml uv.lock poetry.lock requirements*.txt; do
if [ -f "$f" ]; then
echo "-- $f --"
rg -n "typer|click" "$f" | head -40 || true
fi
done
echo "== option signature context across file =="
rg -n "idle_block_ms|def .*worker|`@app.command`|typer\.Option|Annotated\[" src/backend/base/langflow/__main__.py | head -80Repository: langflow-ai/langflow
Length of output: 18711
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== worker loop and idle usage =="
sed -n '865,898p' src/backend/base/langflow/__main__.py
echo "== claim implementation references =="
rg -n "async def .*claim|def .*claim|idle_block_ms|idle_block" src/backend -S | head -120
echo "== packages installed in sandbox =="
python3 - <<'PY'
try:
import typer
import click
print("typer", typer.__version__)
print("click", click.__version__)
import inspect
print("typer.Option signature:")
print(inspect.signature(typer.Option))
print("click types IntRange:")
print(callable(getattr(click.types, "IntRange", None)))
except Exception as e:
print("IMPORT_ERROR", repr(e))
PY
echo "== static verifier: option signatures have range constraints =="
python3 - <<'PY'
from pathlib import Path
import ast
path = Path("src/backend/base/langflow/__main__.py")
tree = ast.parse(path.read_text())
options_no_constraints = []
options_with_constraints = []
for n in ast.walk(tree):
if isinstance(n, ast.arguments):
for name in ("log_level", "env_file", "idle_block_ms"):
for arg in n.args + n.kwonlyargs:
if arg.arg == name:
defn = None
for p in n.defaults + n.kw_defaults:
if isinstance(p, ast.Call) and getattr(p.func, "attr", "") == "Option":
defn = p
break
if defn is None:
continue
constraints = []
for kw in defn.keywords:
if kw.arg in ("min", "max"):
constraints.append(kw.arg)
line = getattr(defn, "lineno", 0)
label = f"{arg.arg}: line {line}"
(options_with_constraints if constraints else options_no_constraints).append((label, constraints, ast.get_source_segment(path.read_text(), defn)))
break
print("with constraints:")
for item in options_with_constraints: print(item[0], item[1], item[2])
print("no constraints at typer.Option:")
for item in options_no_constraints: print(item[0], item[2].replace("\\n", " ")[:200])
PYRepository: langflow-ai/langflow
Length of output: 10713
🌐 Web query:
pydantic v2 model_config validate_assignment false default value assigned to parameter __init__ extra ignore
💡 Result:
In Pydantic v2, the validate_assignment configuration and extra configuration (specifically extra='ignore') operate on different stages of the model lifecycle, and validate_assignment=False (which is the default) does not disable validation during model initialization [1][2][3]. 1. validate_assignment: This setting specifically controls whether validation occurs when an attribute is modified after the model instance has already been created [1][3]. It does not affect the initial validation that occurs when calling the model's constructor [1][3]. Even when set to False, Pydantic will still enforce all field validations during the initial instantiation [1][3]. 2. extra="ignore": When configured as model_config = ConfigDict(extra='ignore'), this setting dictates how the model handles keyword arguments passed to the constructor that do not correspond to defined model fields [2][4]. With 'ignore' (which is the default behavior), any such extra parameters are silently discarded and are not stored in the model instance [2][4]. In summary, setting validate_assignment=False does not stop Pydantic from validating required fields or types during the init call, nor does it affect how the model handles extra parameters [1][3]. If you pass extra parameters to a model with extra='ignore', they are simply dropped during initialization, regardless of your validate_assignment setting [2][4].
Citations:
- 1: https://pydantic.dev/docs/validation/latest/api/pydantic/config/
- 2: https://pydantic.dev/docs/validation/2.12/api/pydantic/config/
- 3:
validate_assignment=Falsehas no effect pydantic/pydantic#9807 - 4: https://github.com/pydantic/pydantic/blob/main/docs/concepts/models.md
Bound --idle-block-ms to a positive value.
This CLI option accepts 0 and negative values, and the worker loop passes it directly to claim(). With a non-positive poll window, claim() can return immediately when work is available and then sleep for zero, which makes the worker loop poll the shared job table without throttling. Add min=1 so Typer rejects invalid values.
🛡️ Proposed fix
idle_block_ms: int = typer.Option(
- 1000, help="Claim poll window in milliseconds (idle sleep between claim attempts)."
+ 1000, min=1, help="Claim poll window in milliseconds (idle sleep между claim attempts)."
),📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| idle_block_ms: int = typer.Option( | |
| 1000, help="Claim poll window in milliseconds (idle sleep between claim attempts)." | |
| ), | |
| idle_block_ms: int = typer.Option( | |
| 1000, min=1, help="Claim poll window in milliseconds (idle sleep between claim attempts)." | |
| ), |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/base/langflow/__main__.py` around lines 848 - 850, Update the
idle_block_ms Typer option to enforce a minimum value of 1, so --idle-block-ms
rejects zero and negative inputs while preserving the existing default and help
text.
| conn = op.get_bind() | ||
| existing = {ix["name"] for ix in sa.inspect(conn).get_indexes("job")} | ||
| if _INDEX not in existing: | ||
| op.create_index(_INDEX, "job", ["status", "type", "created_timestamp"], unique=False) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 5 \
'autocommit_block|CREATE INDEX CONCURRENTLY|postgresql_concurrently|create_index' \
src/backend/base/langflow/alembic
rg -n -C 5 \
'transaction_per_migration|transactional_ddl' \
alembic.ini pyproject.toml src/backendRepository: langflow-ai/langflow
Length of output: 50378
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== target migration =="
cat -n src/backend/base/langflow/alembic/versions/4b8f2c6d9e1a_add_job_claim_scan_index.py
echo
echo "== alembic config migration settings =="
for f in alembic.ini pyproject.toml setup.py setup.cfg tox.ini; do
if [ -f "$f" ]; then
echo "-- $f --"
rg -n 'transaction_per_migration|sqlalchemy.url|setenv|pythonpath|script_location|exclude_tag_prefix|sqlalchemy.url|sqlite|postgresql' "$f" || true
fi
done
echo
echo "== references to job table in services/migration models =="
rg -n '"job"|table_name=.job|class .*Job|job\.(status|type|created_timestamp)|status.*type.*created|claim|lease|heartbeat' src/backend -g '*.py' | head -n 200
echo
echo "== create_index conventions using index existence checks =="
python3 - <<'PY'
from pathlib import Path
import re
target = Path('src/backend/base/langflow/alembic/versions/4b8f2c6d9e1a_add_job_claim_scan_index.py')
text = target.read_text()
print('target file exists:', target.exists())
print('has get_bind:', 'op.get_bind()' in text)
print('has inspect get_indexes:', 'sa.inspect(conn).get_indexes("job")' in text)
print('has op.create_index on job:', bool(re.search(r'op\.create_index\s*\(\s*_INDEX\s*,\s*"job"', text)))
print('has concurrent keywords:', any(k in text for k in ['CREATE INDEX CONCURRENTLY', 'postgresql_concurrently', 'postgresql_with']))
PYRepository: langflow-ai/langflow
Length of output: 27687
Build the job claim index concurrently for PostgreSQL.
op.create_index issues a normal CREATE INDEX, which holds locks that can block writes to the active job table while the index builds. Use a PostgreSQL-specific concurrent build outside an Alembic transaction block, with a safe non-PostgreSQL alternative, because normal PostgreSQL index creation blocks INSERT, UPDATE, and DELETE until it completes.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/backend/base/langflow/alembic/versions/4b8f2c6d9e1a_add_job_claim_scan_index.py`
around lines 32 - 35, Update the upgrade logic around the existing index check
to use PostgreSQL’s concurrent index creation outside the Alembic transaction,
while retaining a safe non-PostgreSQL fallback using the normal index operation.
Preserve the existing _INDEX name, job table, columns, uniqueness, and
idempotent existing-index check.
| """Locust load test for the v2 background execution service. | ||
|
|
||
| Each task submits a background workflow job and polls it to a terminal state, | ||
| proving the background API holds under load with real ``langflow worker`` | ||
| processes draining a redis claim queue (scaled mode) -- or the in-process | ||
| executor (default mode) for a head-to-head comparison. | ||
|
|
||
| Submit: POST /api/v2/workflows {"flow_id", "input_value", "mode": "background"} | ||
| -> {"job_id", "flow_id", "status": "queued"} | ||
| Poll: GET /api/v2/workflows?job_id=<job_id> | ||
| -> 200 with status in {queued, in_progress, completed} ; 500 on failure. | ||
|
|
||
| A request is recorded as a success only when its job reaches COMPLETED inside | ||
| the bounded deadline. Submitted vs completed counts are tracked and printed in | ||
| the final summary so correctness can be cross-checked against the ``job`` table. | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Stale Redis wording remains in two docstrings after the database-backed migration. The scaled backend now uses the database as the queue, but both docstrings still describe Redis.
src/backend/tests/locust/v2_background_locustfile.py#L1-L16: replace "draining a redis claim queue (scaled mode)" with the database-backed worker claim path.src/backend/tests/unit/background_execution/test_side_effect_safety.py#L13-L16: replace "the retry-safe requeue branch uses the real redis backend" withDBBackgroundQueue, which line 89 constructs.
📍 Affects 2 files
src/backend/tests/locust/v2_background_locustfile.py#L1-L16(this comment)src/backend/tests/unit/background_execution/test_side_effect_safety.py#L13-L16
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/locust/v2_background_locustfile.py` around lines 1 - 16,
The docstrings contain stale Redis references after the database-backed
migration. In src/backend/tests/locust/v2_background_locustfile.py lines 1-16,
describe scaled mode as using the database-backed worker claim path; in
src/backend/tests/unit/background_execution/test_side_effect_safety.py lines
13-16, replace the real Redis backend wording with DBBackgroundQueue, matching
the instance constructed by the test.
| with self.client.post( | ||
| "/api/v2/workflows", | ||
| json=body, | ||
| headers=self.headers, | ||
| name="POST /api/v2/workflows (submit)", | ||
| timeout=REQUEST_TIMEOUT, | ||
| catch_response=True, | ||
| ) as resp: |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Inspect the installed/pinned locust version and FastHttpSession.request signature.
fd -t f 'pyproject.toml|uv.lock|requirements.*txt' | xargs rg -n 'locust' | head -50
pip download locust==2.43.4 --no-deps --no-binary :all: -d /tmp/locust_src 2>/dev/null | tail -2
fd -t f 'fasthttp.py' /tmp/locust_src 2>/dev/null | head -5Repository: langflow-ai/langflow
Length of output: 1161
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
try:
import locust.__version__
from locust import FastHttpUser
from inspect import signature
print("LOCUST_VERSION", locust.__version__)
print("FASTHTTPUSER_SIG", signature(FastHttpUser.client.get))
print("FASTHTTPUSER_SIG_POST", signature(FastHttpUser.client.post))
print("HAS_REQUEST", hasattr(FastHttpUser.client, "request"))
if hasattr(FastHttpUser.client, "request"):
print("FASTHTTPSESSION_SIG", signature(FastHttpUser.client.request))
except Exception as e:
print("ERROR", type(e).__name__, str(e))
PY
python3 - <<'PY'
import inspect
try:
from locust.contrib.fasthttp import FastHttpUser
cls = FastHttpUser
print("client_type", cls.client_type)
print("cls attrs network/connection:", (getattr(cls, "network_timeout", getattr(cls, "timeout", None)), getattr(cls, "connection_timeout", None)))
print("client_module", type(cls.client).__module__)
print("client_class", type(cls.client).__name__)
print("client_request", type(cls.client).request)
print("client_request_sig", inspect.signature(type(cls.client).request))
for name in dir(type(cls.client)):
if name.startswith("request"):
print("request_attr", name, getattr(type(cls.client), name, None))
except Exception as e:
print("ERR", type(e).__name__, e)
PYRepository: langflow-ai/langflow
Length of output: 258
🏁 Script executed:
#!/bin/bash
set -euo pipefail
tmp="/tmp/locust-check"
rm -rf "$tmp"
python3 - <<'PY' "$tmp" locust==2.43.4
import sys, os, tarfile, tempfile, urllib.request
tmpdir = sys.argv[1]
version = sys.argv[2]
url = "https://files.pythonhosted.org/packages/52/be/6df1c778f673e1e2d785f262d20a4e11d985f62442466d7ae434b66a587/locust-2.43.4.tar.gz"
tarball = os.path.join(tmpdir, version + ".tar.gz")
print("DOWNLOAD", url)
urllib.request.urlretrieve(url, tarball)
os.makedirs(tarball + ".extracted", exist_ok=True)
print("DOWNLOAD_OK", os.path.exists(tarball))
PY
T="$tmp/locust-check/${2:-locust==2.43.4}.tar.gz.extracted"
python3 <<'PY' "$T"
import sys, ast, inspect
path = sys.argv[1]
# Locate installed-like module code from the unpack tarball without import.
importers = []
for dirpath, _, filenames in os.walk(path):
if "__init__.py" in filenames:
rel = os.path.relpath(os.path.join(dirpath, "__init__.py"), path)
importers.append(rel.replace(os.sep, ".").replace(".__init__", ""))
root_import = next((i for i in importers if i.endswith("locust") or i == "locust"), None)
print("ROOT_IMPORT", root_import)
if not root_import:
raise SystemExit("no locust package")
# Use AST to find FastHttpSession request function signature and relevant request forwarding lines.
from pathlib import Path
src = Path(path).expanduser()
fasthttp = None
for py in src.rglob("fasthttp.py"):
if "locust" in str(py):
fasthttp = py
break
print("FASTHTTP", fasthttp)
if not fasthttp:
raise SystemExit("no fasthttp.py")
code = fasthttp.read_text()
tree = ast.parse(code)
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef) and node.name == "FastHttpSession":
print("CLASS", node.name, node.lineno)
for item in node.body:
if isinstance(item, ast.FunctionDef) and item.name == "request":
print("request_lineno", item.lineno)
print("request_args", ast.Module(body=[item], type_ignores=[]).ast_dump())
for stmt in ast.walk(item):
if isinstance(stmt, ast.Call):
func = stmt
if isinstance(func, ast.Attribute) and func.attr in ("request", "get", "post", "head", "put", "patch", "delete", "json"):
# Print request calls with kwargs or **kwargs
if stmt.keywords or any(isinstance(a, ast.Starred) for a in stmt.args):
snip = ast.get_source_segment(code, stmt)
print("REQUEST_CALL", stmt.lineno, stmt.col_offset, snip)
PYRepository: langflow-ai/langflow
Length of output: 2709
🏁 Script executed:
#!/bin/bash
set -euo pipefail
tmp="/tmp/locust-check"
rm -rf "$tmp"
mkdir -p "$tmp"
python3 - <<'PY' "$tmp" locust==2.43.4
import sys, os, urllib.request, tarfile
tmpdir = sys.argv[1]
version = sys.argv[2]
url = "https://files.pythonhosted.org/packages/52/be/6df1c778f673e1e2d785f262d20a4e11d985f62442466d7ae434b66a587/locust-2.43.4.tar.gz"
req = urllib.request.Request(url)
req.add_header("User-Agent", "curious-verifier")
urllib.request.urlretrieve(url, tmpdir + "/" + version + ".tar.gz")
print("toggled")
PY
T="$tmp/locust-check/locust==2.43.4.tar.gz.extracted"
python3 <<'PY' "$T"
import sys, ast, os
path = sys.argv[1]
os.makedirs(path, exist_ok=True)
with open(path+"/"+os.listdir(tmp)[0]) as f: pass
PY
python3 <<'PY'
import tarfile, ast, os
tmp = "/tmp/locust-check"
for fn in os.listdir(tmp):
if fn.endswith(".tar.gz"):
tb = tarfile.open(os.path.join(tmp, fn), "r:gz")
members = [m for m in tb.getmembers() if m.name.endswith(("locust/contrib/fasthttp.py", "locust/contrib/fasthttp.pyi"))]
print(members)
for m in members:
data = tb.extractfile(m).read().decode("utf-8", "replace")
print("---", m.name, "lines", len(data.splitlines()))
tree = ast.parse(data)
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == "request":
line = data.splitlines()[node.lineno-1] if node.lineno <= len(data.splitlines()) else "<missing line>"
print("FOUND_REQUEST", m.name, node.lineno, "args=", ast.dump(node.args))
print("FIRST_LINE", line[:200])
# search for timeout/params usage
for i,l in enumerate(data.splitlines(), 1):
if "timeout" in l or "params" in l:
print("MATCH", m.name, i, l.strip())
break
PYRepository: langflow-ai/langflow
Length of output: 2573
🌐 Web query:
Locust 2.43.4 FastHttpSession request signature timeout params
💡 Result:
In Locust 2.43.4, the FastHttpSession class (part of locust.contrib.fasthttp) does not utilize a specific signature parameter named timeout directly within its request method signature to manage request timeouts. Instead, FastHttpSession manages timeouts via instance-level attributes that are passed during initialization [1][2]. Specifically, the following timeout-related parameters are available on the FastHttpSession instance: connection_timeout: A float value representing the time in seconds to wait for a connection to be established (default is 60.0) [1][2]. network_timeout: A float value representing the time in seconds to wait for data (default is 60.0) [1][2]. These parameters are typically configured on the FastHttpUser class or instance, which then passes them to the underlying FastHttpSession [2][3][4]. If you need to override these settings, you should set them as attributes on your FastHttpUser class: class MyUser(FastHttpUser): connection_timeout = 30.0 network_timeout = 30.0 Note that while the standard HttpSession.request method (from the requests library) accepts a timeout argument directly, FastHttpSession is designed for higher performance and uses a different client implementation (gevent-based) that relies on the aforementioned instance attributes for timeout control [1][5][2].
Citations:
- 1: https://docs.locust.io/en/2.43.2/%5Fmodules/locust/contrib/fasthttp.html
- 2: https://docs.locust.io/en/2.42.4/increase-performance.html
- 3: https://docs.locust.io/en/stable/increase-performance.html
- 4: https://docs.locust.io/en/2.19.0/_modules/locust/contrib/fasthttp.html
- 5: https://docs.locust.io/en/2.43.3/%5Fmodules/locust/clients.html
🌐 Web query:
Locust FastHttpUser client accept **kwargs request timeout params
💡 Result:
Yes, the Locust FastHttpUser client supports passing a timeout parameter through the kwargs in its request methods [1][2][3]. The FastHttpUser class is designed to provide a similar API to HttpUser [1][2]. Its request methods (such as get, post, request, etc.) accept additional arguments via kwargs, which are passed to the underlying HTTP client implementation [2][3]. For the timeout parameter specifically, the documentation indicates it can be provided as a float (representing seconds) or as a tuple (connect timeout, read timeout) [1][4][5]. Example usage: self.client.get("/endpoint", timeout=(5.0, 10.0)) While FastHttpUser uses geventhttpclient instead of python-requests, the Locust wrapper maps these keyword arguments to the appropriate configuration for the underlying client [1][2]. You can also configure default timeouts for the entire User class by setting the connection_timeout and network_timeout class attributes [2][6].
Citations:
- 1: https://docs.locust.io/en/2.31.4/api.html
- 2: https://docs.locust.io/en/2.43.1/increase-performance.html
- 3: https://docs.locust.io/en/2.34.1/api.html
- 4: https://github.com/locustio/locust/blob/master/locust/clients.py
- 5: https://docs.locust.io/en/2.32.1/_modules/locust/clients.html
- 6: https://github.com/locustio/locust/blob/cac9add64664aa5ec18573f074cca377cfe90400/locust/contrib/fasthttp.py
🌐 Web query:
Locust FastHttpSession source kwargs timeout params 2.43.3 FastHttpSession request
💡 Result:
In Locust 2.43.3, the FastHttpSession.request method accepts a kwargs parameter to handle extra arguments [1]. While the documented explicit parameters for request (such as method, url, name, data, json, headers, auth, stream, allow_redirects, and catch_response) do not include an explicit timeout parameter, the kwargs are passed down to the underlying HTTP client implementation [1][2]. Specifically, the FastHttpSession.request method calls self._send_request_safe_mode(..., kwargs), which forwards these arguments [2]. The underlying HTTP library used by FastHttpSession is typically geventhttpclient, which uses different timeout mechanisms (e.g., network_timeout and connection_timeout) configured during the initialization of FastHttpSession [3][4]. If you are looking to set request-specific timeouts, note that unlike the standard HttpSession (which wraps the requests library), FastHttpSession is designed for high performance using geventhttpclient and may not support the same per-request timeout keyword arguments that requests.request does. In FastHttpSession, timeouts are generally managed at the session level via parameters like network_timeout and connection_timeout provided during the instantiation of the FastHttpUser or the FastHttpSession object itself [3][4]. Passing a timeout argument via kwargs in request may have no effect depending on the underlying implementation constraints [5].
Citations:
- 1: https://docs.locust.io/en/2.43.3/api.html
- 2: https://docs.locust.io/en/stable/%5Fmodules/locust/contrib/fasthttp.html
- 3: https://docs.locust.io/en/2.43.3/_modules/locust/contrib/fasthttp.html
- 4: https://docs.locust.io/en/2.43.2/%5Fmodules/locust/contrib/fasthttp.html
- 5: https://docs.locust.io/en/2.37.4/%5Fmodules/locust/contrib/fasthttp.html
🌐 Web query:
Locust 2.43.4 locust clients py FastHttpSession request timeout params kwargs
💡 Result:
In Locust 2.43.4, the FastHttpSession class (used by FastHttpUser) acts as a high-performance alternative to the standard HttpSession [1][2]. For request timeouts when using FastHttpSession, note that the timeout parameter found in the standard requests-based HttpSession is not directly supported in the same way within the FastHttpSession.request method [3][4]. Instead, FastHttpUser handles connection and network timeouts at the client level using specific attributes that are passed to the underlying FastHttpSession upon instantiation [5][6][1]: 1. connection_timeout: Controls the time allowed to establish a connection (default is 60.0 seconds) [6][1]. 2. network_timeout: Controls the time allowed for network operations (default is 60.0 seconds) [6][1]. These attributes can be configured in your FastHttpUser subclass before it is instantiated [6][1]: class MyUser(FastHttpUser): connection_timeout = 10.0 network_timeout = 20.0 While the request method accepts kwargs, these are typically passed to the underlying geventhttpclient and do not support the same timeout dictionary or tuple structures used by the requests library [3][4]. Users requiring per-request timeout flexibility should be aware that FastHttpSession is optimized for performance by using a persistent connection pool, and per-request timeout overrides are not natively exposed through the request method signature in the same manner as the standard requests client [6][1][2].
Citations:
- 1: https://docs.locust.io/en/latest/_modules/locust/contrib/fasthttp.html
- 2: https://docs.locust.io/en/stable/increase-performance.html
- 3: https://docs.locust.io/en/2.43.3/api.html
- 4: https://docs.locust.io/en/stable/api.html
- 5: https://docs.locust.io/en/2.43.3/_modules/locust/contrib/fasthttp.html
- 6: https://github.com/locustio/locust/blob/master/locust/contrib/fasthttp.py
Use class-level timeout attributes for FastHttpUser requests.
Locust’s FastHttpSession does not support per-request timeout; unsupported kwargs can be forwarded to geventhttpclient without applying Locust’s configured network_timeout / connection_timeout. Set both class attributes to REQUEST_TIMEOUT instead of passing timeout= into self.client.get / self.client.post.
Also applies to: 104-111
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/locust/v2_background_locustfile.py` around lines 78 - 85,
The workflow request in the locust user should not pass per-request timeout
arguments to self.client.post or self.client.get. Set the FastHttpUser class
attributes network_timeout and connection_timeout to REQUEST_TIMEOUT, then
remove the timeout keyword from both request calls while preserving their
existing response handling.
| code = poll.status_code | ||
| if code == 200: | ||
| status = (poll.json() or {}).get("status") | ||
| if status == "completed": | ||
| poll.success() | ||
| _bump("completed") | ||
| return | ||
| # queued / in_progress: still working, keep polling. | ||
| poll.success() | ||
| elif code == 500: | ||
| # Failed job surfaces as 500 JOB_FAILED on the status route. | ||
| poll.success() # the poll call itself worked; the job failed | ||
| _bump("failed") | ||
| return | ||
| elif code == 408: | ||
| poll.success() | ||
| _bump("failed") | ||
| return | ||
| else: | ||
| _bump("poll_errors") | ||
| poll.failure(f"poll {code}: {poll.text[:200]}") | ||
| return |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Guard the poll body parse like the submit body parse.
Line 114 calls poll.json() without a guard. A 200 response with a non-JSON body raises inside the with-block. The task then ends without a poll_errors bump, so the final accounting under-reports. Line 132 also indexes poll.text, which can be None on an empty body.
🛡️ Proposed fix
code = poll.status_code
if code == 200:
- status = (poll.json() or {}).get("status")
+ try:
+ status = (poll.json() or {}).get("status")
+ except Exception as exc: # noqa: BLE001
+ _bump("poll_errors")
+ poll.failure(f"poll body parse error: {exc}")
+ return
if status == "completed":🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/locust/v2_background_locustfile.py` around lines 112 - 133,
Guard the 200-response body parsing in the polling flow around poll.json() so
malformed or non-JSON responses are handled as poll errors and increment
poll_errors instead of escaping the with-block. Also make the failure message in
the fallback branch safe when poll.text is None by using an empty-string
fallback before slicing.
| if "+psycopg" in raw: | ||
| return raw | ||
| if raw.startswith("postgresql+asyncpg://"): | ||
| return raw.replace("postgresql+asyncpg://", "postgresql+psycopg://", 1) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
The substring check also matches postgresql+psycopg2://.
"+psycopg" in raw is true for a psycopg2 URL. psycopg2 is a sync driver, so create_async_engine(real_services_db_url) in test_real_services_harness.py fails for that input. Match the exact dialect prefix and normalize psycopg2 as well.
🛠️ Proposed fix
- if "+psycopg" in raw:
+ if raw.startswith("postgresql+psycopg://"):
return raw
+ if raw.startswith("postgresql+psycopg2://"):
+ return raw.replace("postgresql+psycopg2://", "postgresql+psycopg://", 1)
if raw.startswith("postgresql+asyncpg://"):
return raw.replace("postgresql+asyncpg://", "postgresql+psycopg://", 1)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if "+psycopg" in raw: | |
| return raw | |
| if raw.startswith("postgresql+asyncpg://"): | |
| return raw.replace("postgresql+asyncpg://", "postgresql+psycopg://", 1) | |
| if raw.startswith("postgresql+psycopg://"): | |
| return raw | |
| if raw.startswith("postgresql+psycopg2://"): | |
| return raw.replace("postgresql+psycopg2://", "postgresql+psycopg://", 1) | |
| if raw.startswith("postgresql+asyncpg://"): | |
| return raw.replace("postgresql+asyncpg://", "postgresql+psycopg://", 1) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/unit/background_execution/conftest.py` around lines 37 -
40, Update the URL normalization logic around the raw URL checks so psycopg2
URLs are not mistaken for psycopg URLs: match the exact async psycopg dialect
prefix, and normalize postgresql+psycopg2 URLs to the async-compatible
postgresql+psycopg form before returning. Preserve unchanged URLs that already
use the exact psycopg prefix.
| proc = subprocess.Popen( # noqa: S603 | ||
| ["uv", "run", "langflow", "worker", "--idle-block-ms", str(idle_block_ms)], # noqa: S607 | ||
| env=env, | ||
| stdout=subprocess.PIPE, | ||
| stderr=subprocess.STDOUT, | ||
| start_new_session=True, | ||
| ) | ||
| self.procs.append(proc) | ||
| return proc |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
The unread stdout pipe can block the worker subprocess.
Nothing reads proc.stdout while the worker runs. drain_worker_output reads it only on failure or teardown. When the worker writes more than the OS pipe buffer (commonly 64 KiB), its write blocks, the claim loop stops, and wait_for_status fails with a timeout that hides the real cause. A real graph build with logging reaches that volume quickly.
Redirect the worker output to a temporary file, then read the file for diagnostics. A reader thread per process is the alternative.
🐛 Proposed direction
+import tempfile
...
- proc = subprocess.Popen( # noqa: S603
- ["uv", "run", "langflow", "worker", "--idle-block-ms", str(idle_block_ms)], # noqa: S607
- env=env,
- stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT,
- start_new_session=True,
- )
- self.procs.append(proc)
+ log = tempfile.NamedTemporaryFile(suffix=".worker.log", delete=False) # noqa: SIM115
+ proc = subprocess.Popen( # noqa: S603
+ ["uv", "run", "langflow", "worker", "--idle-block-ms", str(idle_block_ms)], # noqa: S607
+ env=env,
+ stdout=log,
+ stderr=subprocess.STDOUT,
+ start_new_session=True,
+ )
+ self.logs[proc.pid] = log.name
+ self.procs.append(proc)drain_worker_output then reads the tail of each log file and no longer needs to terminate a live worker to reach EOF.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/unit/services/background_execution/_subprocess_harness.py`
around lines 134 - 142, Update the subprocess setup around Popen and
drain_worker_output so worker stdout/stderr is redirected to a temporary log
file instead of an unread PIPE, preventing the worker from blocking on pipe
capacity. Track the file handle or path per process, and have
drain_worker_output read the available log-file tail for diagnostics without
terminating a live worker merely to reach EOF.
| async def drive_stop_then_wait(): | ||
| # Wait until the run is in-flight, then stop it through the facade. | ||
| await asyncio.wait_for(in_flight.wait(), timeout=10.0) | ||
| await facade.stop_job(job_id, active_user) | ||
| # Wait until the durable row reaches a terminal state, then end the loop. | ||
| for _ in range(200): | ||
| refreshed = await jobs.get_job_by_job_id(job_id) | ||
| if refreshed.status in {JobStatus.CANCELLED, JobStatus.COMPLETED, JobStatus.FAILED}: | ||
| stop_event.set() | ||
| return | ||
| await asyncio.sleep(0.05) | ||
| stop_event.set() | ||
|
|
||
| driver = asyncio.create_task(drive_stop_then_wait()) | ||
| await run_worker_loop( | ||
| worker_backend, | ||
| worker_runner, | ||
| stop_event=stop_event, | ||
| idle_block_ms=50, | ||
| job_service=jobs, | ||
| owner="worker:w1", | ||
| ) | ||
| await driver |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
A failing driver task hangs the worker-loop tests. In both tests a helper task is the only writer of stop_event, and the test awaits run_worker_loop without a timeout. If the helper raises, stop_event stays clear, the awaited loop never returns, and the helper's exception is never reported. The fix is the same at both sites: set stop_event in a finally block and bound the loop await.
src/backend/tests/unit/services/background_execution/test_scaled_stop.py#L76-L98: wrap the body ofdrive_stop_then_waitintry/finallywithstop_event.set()in thefinally, and callawait asyncio.wait_for(run_worker_loop(...), timeout=30.0).src/backend/tests/unit/services/background_execution/test_scaled_end_to_end.py#L71-L89: apply the sametry/finallytostop_when_doneand the sameasyncio.wait_forbound on therun_worker_loopcall.
📍 Affects 2 files
src/backend/tests/unit/services/background_execution/test_scaled_stop.py#L76-L98(this comment)src/backend/tests/unit/services/background_execution/test_scaled_end_to_end.py#L71-L89
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/unit/services/background_execution/test_scaled_stop.py`
around lines 76 - 98, Ensure both helper tasks always set stop_event in a
finally block and bound the worker-loop await with a 30-second timeout. In
src/backend/tests/unit/services/background_execution/test_scaled_stop.py lines
76-98, update drive_stop_then_wait and wrap run_worker_loop with
asyncio.wait_for; apply the same changes to stop_when_done and run_worker_loop
in
src/backend/tests/unit/services/background_execution/test_scaled_end_to_end.py
lines 71-89.
| def test_background_poll_interval_default(monkeypatch): | ||
| assert Settings().background_poll_interval_s == 0.5 | ||
| monkeypatch.setenv("LANGFLOW_BACKGROUND_POLL_INTERVAL_S", "0.1") | ||
| assert Settings().background_poll_interval_s == 0.1 |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate relevant files =="
git ls-files | rg '(^|/)test_background_scaled_settings\.py$|(^|/)settings.*\.py$|runtime.*settings.*\.py$|background.*settings.*\.py$' || true
echo
echo "== target test excerpt =="
target="src/lfx/tests/unit/services/settings/test_background_scaled_settings.py"
if [ -f "$target" ]; then
nl -ba "$target" | sed -n '1,120p'
fi
echo
echo "== search RuntimeSettings / background_poll_interval_s =="
rg -n "RuntimeSettings|background_poll_interval_s|LANGFLOW_BACKGROUND_POLL_INTERVAL_S" -S .Repository: langflow-ai/langflow
Length of output: 2553
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== target test excerpt =="
target="src/lfx/tests/unit/services/settings/test_background_scaled_settings.py"
if [ -f "$target" ]; then
awk '{printf "%6d %s\n", NR, $0}' "$target" | sed -n '1,120p'
fi
echo
echo "== search RuntimeSettings / background_poll_interval_s =="
rg -n "RuntimeSettings|background_poll_interval_s|LANGFLOW_BACKGROUND_POLL_INTERVAL_S" -S src/lfx src/backend || true
echo
echo "== inspect candidate settings files around relevant symbols =="
pgrep -f "RuntimeSettings|background_poll_interval_s" $(git ls-files 'src/lfx/src/lfx/services/settings/*.py' 'src/lfx/src/lfx/settings.py' | tr '\n' ' ') || trueRepository: langflow-ai/langflow
Length of output: 4660
Test rejected polling intervals.
background_poll_interval_s requires gt=0, so add tests for LANGFLOW_BACKGROUND_POLL_INTERVAL_S=0 and a negative value. The current test covers only valid values.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/lfx/tests/unit/services/settings/test_background_scaled_settings.py`
around lines 31 - 34, Extend test_background_poll_interval_default to verify
that LANGFLOW_BACKGROUND_POLL_INTERVAL_S values of 0 and a negative number are
rejected by Settings, preserving the existing assertions for the default and
valid 0.1 interval.
Source: Coding guidelines
Second of a two-PR stack, stacked on the durable default backend PR. This adds the opt-in scaled backend so background flows run off the API workers, decoupled and distributed across replicas. Same
BackgroundExecutionServicefacade, the endpoint never knows which backend is active. The queue is the database itself: no broker, no redis — deployments need nothing beyond the DB the API already uses.What it adds
langflow workerCLI process that lease-claims QUEUED job rows off the shared database (claim_next_queued_lease, the same exact-heartbeat conditional UPDATE every other single-flight guard uses), runs the same runner the default backend uses, and persists durable milestones tojob_eventsso any API replica can reattach to any job.job_eventsfrom the DB, then poll the same table past the replayed boundary (no gap, no double delivery,Last-Event-IDcontract preserved). A grace pass on terminal status covers the paths that flip the status before appending their terminal milestone. Tails end on SUSPENDED too, so a HITL pause never strands a reader.execution_signalsSTOP row that the worker polls at vertex boundaries.(status, type, created_timestamp)index backing the claim poll and watchdog scan (the job table has no retention, so the hottest query must not degrade into a growing scan).Because the queue and the system of record are one table, the broker failure modes are structurally impossible: there is no pending/processing list to drift from the DB (a QUEUED row is enqueued by definition, so stranded-queue recovery has no equivalent) and no token-exchange protocol (the conditional UPDATE is the single flight).
Selection is the explicit
LANGFLOW_BACKGROUND_BACKEND=scaledsetting, decoupled fromjob_queue_type(redis for v1 build events no longer implies anything about the background backend). The facade logs once at wiring time in scaled mode so a fleet with zero workers is diagnosable. Scaled + SQLite works on a single host; use Postgres for multi-host fleets.Semantics
Same guarantees as the default backend, now across processes: at-least-once for not-yet-started work, at-most-once for in-flight work by default (a crashed worker's job becomes FAILED
worker_lost), and opt-in bounded retry per flow with atomic attempt accounting so concurrent watchdogs can't push a job pastmax_attempts.Tradeoffs made deliberately: claim latency is bounded by the worker poll window (
--idle-block-ms, default 1s) and reattach milestone latency bybackground_poll_interval_s(default 0.5s) instead of a broker's push; inline requestglobalsare dropped on every scaled run (the worker hydrates from the redacted persisted row — use stored global variables for background runs).What's proven
The former real-redis test fleet now runs against the ordinary test DB (no external service), and the real-service tier needs only Postgres:
langflow workerOS subprocess claims, builds, and completes a real graph off a shared Postgres while the API side reattaches; akill -9mid-job is reconciled by the watchdog (fresh-lease guard proven first); a durable stop reaches the separate worker and lands CANCELLED with the signal consumed.run_failedevent, racing lease-claims on one row produce exactly one winner.Three review rounds (specialist + adversarial validation per round) ran on this rework; all findings are fixed in-branch. Not yet redone: the Locust saturation run from the redis era — those numbers no longer apply and a fresh head-to-head against the in-process default is pending.
Summary by CodeRabbit
New Features
Bug Fixes
Tests