Repository navigation
feat(api): DB-backed scaled background execution backend + langflow worker #13508
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -996,6 +996,68 @@ def print_banner(host: str, port: int, protocol: str) -> None: | |||||||||||||
| logger.info(f"Open Langflow: {protocol}://{access_host}:{port}") | ||||||||||||||
|
|
||||||||||||||
|
|
||||||||||||||
| @app.command() | ||||||||||||||
| def worker( | ||||||||||||||
| log_level: str | None = typer.Option(None, help="Logging level."), | ||||||||||||||
| env_file: Path | None = typer.Option(None, help="Path to the .env file."), | ||||||||||||||
| idle_block_ms: int = typer.Option( | ||||||||||||||
| 1000, help="Claim poll window in milliseconds (idle sleep between claim attempts)." | ||||||||||||||
| ), | ||||||||||||||
|
Comment on lines
+1003
to
+1005
Contributor
There was a problem hiding this comment. 🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🌐 Web query:
💡 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:
🏁 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:
💡 Result: In Pydantic v2, the Citations:
Bound This CLI option accepts 🛡️ 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
Suggested change
🤖 Prompt for AI Agents |
||||||||||||||
| ) -> None: | ||||||||||||||
| """Run a Langflow background worker that drains queued jobs off the database. | ||||||||||||||
|
|
||||||||||||||
| Requires LANGFLOW_BACKGROUND_BACKEND=scaled and the same database as the | ||||||||||||||
| API. Each worker lease-claims QUEUED workflow jobs off the shared job table, | ||||||||||||||
| runs them through the background JobRunner (durable milestones land in | ||||||||||||||
| job_events, which any API replica's event tail reads), and releases the | ||||||||||||||
| lease. Run as many worker processes as you need horizontal capacity for. | ||||||||||||||
| """ | ||||||||||||||
| if env_file: | ||||||||||||||
| load_dotenv(env_file, override=True) | ||||||||||||||
| if log_level: | ||||||||||||||
| configure(log_level=log_level) | ||||||||||||||
|
|
||||||||||||||
| async def _run_worker() -> None: | ||||||||||||||
| from langflow.services.background_execution.worker import build_worker, run_worker_loop | ||||||||||||||
|
|
||||||||||||||
| await initialize_services() | ||||||||||||||
| settings = get_settings_service().settings | ||||||||||||||
| if not settings.background_backend_is_scaled: | ||||||||||||||
| typer.echo("LANGFLOW_BACKGROUND_BACKEND must be 'scaled' to run a worker.") | ||||||||||||||
| raise typer.Exit(code=1) | ||||||||||||||
|
|
||||||||||||||
| from uuid import uuid4 | ||||||||||||||
|
|
||||||||||||||
| from langflow.services.deps import get_job_service | ||||||||||||||
|
|
||||||||||||||
| # Process-unique owner so this worker's heartbeats are attributable, and | ||||||||||||||
| # a periodic watchdog so a dead worker's in-flight job is reaped under a | ||||||||||||||
| # steady fleet WITHOUT requiring a restart. | ||||||||||||||
| owner = f"worker:{os.getpid()}:{uuid4().hex[:8]}" | ||||||||||||||
| backend, worker_runner, teardown = await build_worker(owner=owner) | ||||||||||||||
| stop_event = asyncio.Event() | ||||||||||||||
| loop = asyncio.get_running_loop() | ||||||||||||||
| for sig in (signal.SIGTERM, signal.SIGINT): | ||||||||||||||
| with suppress(NotImplementedError): | ||||||||||||||
| loop.add_signal_handler(sig, stop_event.set) | ||||||||||||||
| logger.info("Langflow worker started; draining queued jobs off the database.") | ||||||||||||||
| try: | ||||||||||||||
| await run_worker_loop( | ||||||||||||||
| backend, | ||||||||||||||
| worker_runner, | ||||||||||||||
| stop_event=stop_event, | ||||||||||||||
| idle_block_ms=idle_block_ms, | ||||||||||||||
| job_service=get_job_service(), | ||||||||||||||
| owner=owner, | ||||||||||||||
| lease_ttl_s=settings.background_lease_ttl_s, | ||||||||||||||
| watchdog_interval_s=settings.background_watchdog_interval_s, | ||||||||||||||
| ) | ||||||||||||||
| finally: | ||||||||||||||
| await teardown() | ||||||||||||||
|
|
||||||||||||||
| asyncio.run(_run_worker()) | ||||||||||||||
|
|
||||||||||||||
|
|
||||||||||||||
| @app.command() | ||||||||||||||
| def superuser( | ||||||||||||||
| username: str = typer.Option( | ||||||||||||||
|
|
||||||||||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| """add_job_claim_scan_index | ||
|
|
||
| Revision ID: a4e8c2b6d931 | ||
| Revises: 1d8d193028c8 | ||
| Create Date: 2026-07-23 12:00:00.000000 | ||
|
|
||
| Phase: EXPAND | ||
| """ | ||
|
|
||
| from collections.abc import Sequence | ||
|
|
||
| import sqlalchemy as sa | ||
| from alembic import op | ||
|
|
||
| # revision identifiers, used by Alembic. | ||
| revision: str = "a4e8c2b6d931" # pragma: allowlist secret | ||
| down_revision: str | Sequence[str] | None = "1d8d193028c8" # pragma: allowlist secret | ||
| branch_labels: str | Sequence[str] | None = None | ||
| depends_on: str | Sequence[str] | None = None | ||
|
|
||
| _INDEX = "ix_job_claim_scan" | ||
|
|
||
|
|
||
| def upgrade() -> None: | ||
| """Composite index for the scaled backend's claim poll and watchdog scan. | ||
|
|
||
| ``claim_next_queued_lease`` filters ``status`` + ``type`` and sorts by | ||
| ``created_timestamp`` on every worker poll; the watchdog scans | ||
| ``status`` + ``type`` on its interval. ``ix_job_status_created`` cannot serve | ||
| either: the ``type`` filter sits between its two sorted columns. Without this | ||
| index the hottest query in the fleet is a scan+sort over every row retention | ||
| has not yet removed. | ||
| """ | ||
| 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) | ||
|
Comment on lines
+34
to
+37
Contributor
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.
🤖 Prompt for AI Agents |
||
|
|
||
|
|
||
| def downgrade() -> None: | ||
| conn = op.get_bind() | ||
| existing = {ix["name"] for ix in sa.inspect(conn).get_indexes("job")} | ||
| if _INDEX in existing: | ||
| op.drop_index(_INDEX, table_name="job") | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@ogabrielluiz Things I have found and also sharing some opinions; please let me know what you think about them:
Review:
Overall, this is a good simplification. Making the durable
jobtable the queue removes the second system of record, and the conditional-UPDATE (compare-and-swap) primitive is used consistently across claim / fail / retry/resume. The concurrency reasoning holds up — I traced the claim race and two workers cannot both run a job, even though the row staysQUEUEDduring the claim window.My main reservation is database pressure, not correctness. Details below, plus two bugs in the suspended-job path.
Bugs
1. In scaled mode, stopping a SUSPENDED job does nothing
supersede_suspended_runsandstop_jobboth short-circuit tobackend.stop()when scaled:DBBackgroundQueue.stop()only writes a STOP signal:But a STOP signal is only ever read by a live runner (
runner._stop_requested, polled at durable frame boundaries). ASUSPENDEDjob has already exited_driveand has no worker attached. I grepped every STOP consumer — the only two arerunner._stop_requestedand_cancel_suspended, and the scaled branch skips the latter. Nothing reconciles aSUSPENDEDrow against a pending STOP.Consequences in scaled mode:
SUSPENDEDforever (background_input_deadline_sisNoneby default, so no sweep reaps it).supersede_suspended_runsexists to prevent._drivehits its first durable frame,_stop_requestedfinds the stale STOP, and the resumed run immediately self-cancels. This is the failure modeconsume_signals' own docstring warns about.Same shape in
stop_job— the scaled branch returns before theSUSPENDEDcheck, so a user stop returns success on a job that never leavesSUSPENDED.Suggested fix — branch on status, not on backend. A suspended job has no live worker in any mode, so
_cancel_suspendedis correct everywhere:2.
supersede_suspended_runsreports success unconditionally when scaledIn the scaled branch
cancelled.append(stale_job_id)runs regardless of outcome, while the default branch only appends when_cancel_suspendedactually won the CAS. Callers are told cleanup happened when it didn't. Fixed by the change above.Main concern: database pressure
This design trades broker infrastructure for steady DB load. That's a reasonable trade, but a few things should land before this is run at scale.
3. No retention on
joborjob_events(please treat as a follow-up blocker)The index migration comment says it outright:
The index helps, but nothing purges terminal rows and
job_eventsgrows one row per durable milestone forever. Over months the claim poll sorts out of an ever-growing table and the events table dominates storage. A scaled backend needs a companion reaper (age-based delete of terminal jobs + their events, or partitioning). Happy for it to be a separate PR, but it should be tracked before anyone enablesscaledin production.4. Watchdog runs in every worker with no jitter — periodic DB spikes
_watchdog_loopruns in every worker process on a fixed 15s interval, and each pass is a fullIN_PROGRESSscan plus aget_job_by_job_idper row:With 20 workers the same orphan set is read 20x. Correctness is fine (the conditional UPDATE single-flights the actual fix), but workers started together drift into lockstep and you get a synchronised burst every 15s rather than smooth load.
Two cheap improvements:
Longer term, consider whether the lease watchdog needs to run in every worker at all, or whether one holder (advisory lock / leader election) is enough — 20 workers duplicating the same scan is pure read amplification.
5.
candidatesis hardcoded at 5DBBackgroundQueue.claim()never passescandidates, so it's always 5. Fetching a small batch is a nice way to let workers spread across distinct jobs instead of all fighting over the head of the queue — but past ~5 workers the extras collide again, lose the CAS, and back off a fullidle_block_ms. Either make it settings-driven and scaled to the expected fleet, or land theFOR UPDATE SKIP LOCKEDpath the ponytail comment already proposes.6. Write amplification per milestone (noting, not asking for a change)
Each durable frame costs three round-trips: the stop-check
SELECT,SELECT max(seq), and theINSERT. Fine for typical flows; worth knowing it scales with milestones/sec across the fleet, and that PGNOTIFYwould not reduce it. On SQLite this is the part that will contend first —append_eventalready has a "database is locked" retry loop, which tells the story. Givenscaled+ SQLite is single-host only, consider a startup warning rather than only a docstring note.7. On PG LISTEN/NOTIFY
The
# ponytail:markers already point at this and I agree with the direction — it removes idle claim polling and the ~4 queries/sec per connected event-tail client. Two things worth deciding up front when it happens:LISTENholds a connection open. It must be one listener per process fanning out to in-process subscribers, not one per SSE client, or you swap query load for connection exhaustion.Not asking for it here. My suggested order is retention → watchdog jitter → NOTIFY.
Making room for future backends (NATS JetStream and friends)
We'd like to add a JetStream backend later for durability plus higher throughput. The seam is nearly right, but two things would make that genuinely additive rather than a refactor:
a) Formalise the backend contract. It's currently duck-typed with no base class. An explicit
Protocoldocuments what a new backend must satisfy and makes the surface reviewable:b) Replace the
if self._scaled:booleans with polymorphic calls. There are 7 of them in the facade._scaledis really "is there a backend object", so a third backend would have to thread through the same boolean. The resume path is the clearest example — the facade currently reaches past the backend intojob_service:That would read better as a backend method (
await self._backend.hand_back(job_id, owner=self._owner)), with the default path supplying an in-process implementation.One design note we'd want agreement on before writing a JetStream backend: is the broker the queue of record, or just a dispatch hint? If JetStream only wakes workers and the DB CAS remains the real claim, every semantic in this PR holds unchanged and the new backend is a clean drop-in. If JetStream owns the queue, there are now two lease mechanisms — JetStream
AckWaitredelivery and the DB heartbeat/watchdog — which can disagree and redeliver a job a live worker still owns. That's exactly the drift class this PR removes by having one system of record, so we'd want to keep the DB as the authority and use the broker for dispatch and live event fan-out only.Minor
stop()callsself._executor.stop()even in scaled mode, wherestart()never started it. I checked and it's a safe no-op (_workersis empty), so cosmetic only — butstart()guards onself._scaledandstop()doesn't, which reads as an oversight.execute_with_statusand the terminal event by the runner'sfinally(and byrequeue_lostforworker_lost). A crash between them leaves a terminal row with no terminal event. Theevents()grace pass handles it (the tail ends on status), so this is fine — flagging only to confirm it was deliberate.retry_requeue_claimleaves staleowner/heartbeat_atin metadata when flipping back toQUEUED. Harmless because the next claimant's staleness check overwrites both, but worth a comment so nobody later keys offowneron aQUEUEDrow.Things I checked that are correct
claim_queued_lease) is sound — theheartbeat_atguard means two workers racing the same row produce exactly onerowcount == 1, including theprior_hb is Nonecase for fresh submits.QUEUEDthrough the claim window is the right call: a worker that dies before starting leaves a re-claimable row instead of a strandedIN_PROGRESS.requeue_lost's control flow is subtle but correct — thecontinueplacement means a lost retry race falls through to "someone else requeued it" rather than double-failing.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hey @dkaushik94, thanks for the review. Both bugs were real, good catches.
Fixed, taking your suggestion of branching on status instead of backend:
supersede_suspended_runsnow always goes through_cancel_suspended(which also fixes the unconditional success reporting), andstop_jobchecks SUSPENDED before the scaled branch, so the lost-claim fallthrough lands on the durable STOP for a run a resume just revived. The tests reproduce both failures first (row stuck SUSPENDED, unconsumed STOP left behind) and pass with the fix.The pressure items:
background_claim_candidates, default 5) with sizing guidance in its docstring. SKIP LOCKED stays the marked upgrade path.The seam: added a
BackgroundBackendprotocol with the DB-as-sole-system-of-record invariant in its docstring. For collapsing the_scaledbooleans into polymorphic calls, agreed, and rather than fold that refactor into this diff mid-review I'll do it as a PR stacked right on top of this one, building the in-process backend object for the default path.On your design question: the DB stays the queue of record. The broker is dispatch and event fan-out only, and the CAS stays the only claim. Two lease mechanisms that can disagree is exactly the drift this PR exists to remove, so a JetStream backend should be a clean drop-in under the protocol.
On the terminal status/event gap: it is deliberate (the
events()grace pass ends tails on status, and there are tests pinning it), but you're right that a crash in that window leaves the event log permanently missing its terminal milestone, which matters for replay and timelines even though nothing hangs. I see three ways to treat it: leave it as an accepted crash artifact; teach the existing sweep to append the missing terminal event when it finds a terminal row without one (small, off the hot path, history self-heals); or extend thefail_queued_jobsingle-transaction pattern toexecute_with_statusandfail_in_progress_job(structural, but it rewires the runner's finally path and adds seq contention to every run end). I lean toward the sweep heal. What do you think?retry_requeue_claimnow carries a comment saying owner/heartbeat_at are kept on purpose and not to key off owner on a QUEUED row, andstop()mirrorsstart()'s scaled guard.