refactor(api): review round 2 for the DB-backed scaled backend

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.

Author: ogabrielluiz

src/backend/base/langflow/alembic/versions/4b8f2c6d9e1a_add_job_claim_scan_index.py

@@ -0,0 +1,42 @@
+"""add_job_claim_scan_index
+
+Revision ID: 4b8f2c6d9e1a
+Revises: d19e7b3c5a42
+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 = "4b8f2c6d9e1a"  # pragma: allowlist secret
+down_revision: str | Sequence[str] | None = "d19e7b3c5a42"  # 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. The job table has no retention, so
+    without this index the hottest query is an ever-growing scan+sort.
+    """
+    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)
+
+
+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")

src/backend/base/langflow/services/background_execution/db_backend.py

@@ -37,6 +37,10 @@
 
 # Durable statuses that mean the run is over (the event tail drains and ends).
 _TERMINAL_STATUSES = frozenset({JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED, JobStatus.TIMED_OUT})
+# Statuses that end the event tail: terminal runs, plus SUSPENDED — a paused
+# run will produce no more durable frames until it is resumed, so a tail
+# waiting on it would poll forever (mirrors the default backend's is_done).
+_TAIL_END_STATUSES = _TERMINAL_STATUSES | {JobStatus.SUSPENDED}
 
 
 def _coerce_uuid(job_id: Any) -> Any:
@@ -180,9 +184,17 @@ async def events(self, job_id: str, last_event_id: int = 0) -> AsyncIterator[Any
         flips FAILED, then appends ``run_failed``), so a tail that returned on
         the first terminal observation could end without the terminal
         milestone. On observing a terminal status the tail therefore drains,
-        waits one more poll interval, drains again, and ends — the append
-        happens within milliseconds of the flip, so the bounded grace pass
-        closes the window without risking an unterminated stream.
+        waits one more poll interval, drains again, and ends. The gap between
+        flip and append is a handful of sequential DB round trips (widest on
+        the CANCELLED path), comfortably inside the default 0.5s interval —
+        an operator tuning ``background_poll_interval_s`` very low narrows the
+        grace window with it, degrading to a status-only stream end (never a
+        hang).
+
+        A job that no worker ever claims stays QUEUED and the tail keeps
+        polling: deliberately client-bounded (the job may start any moment;
+        two PK-indexed queries per interval), matching the semantics of
+        waiting on a queued run.
         """
         # ponytail: poll-based tail (<= poll_interval_s added latency per
         # milestone); switch to pg LISTEN/NOTIFY wakes if that ever matters.
@@ -202,7 +214,7 @@ async def _drain() -> AsyncIterator[Any]:
                 yield event
 
             job = await self._job_service.get_job_by_job_id(durable_id)
-            if job is None or job.status in _TERMINAL_STATUSES:
+            if job is None or job.status in _TAIL_END_STATUSES:
                 # Grace pass: drain, give the terminal-event append one poll
                 # interval to land, drain once more, then end the stream.
                 async for event in _drain():

src/backend/base/langflow/services/background_execution/service.py

@@ -232,10 +232,12 @@ async def submit(self, *, flow_id: UUID, request: dict[str, Any], user: UserRead
         # carry inline secrets (API keys), and storing them plaintext in the
         # durable ``job`` table (JSONB on Postgres) widens the blast radius of any
         # DB read (backup, ops access, a SQL-injection elsewhere) beyond the
-        # live-only handling globals get on the sync path. Tradeoff: a background
-        # re-enqueue after a restart drops inline globals — reference STORED global
-        # variables by name for background runs rather than passing secrets inline.
-        # The live in-memory run below still uses the full ``request``.
+        # live-only handling globals get on the sync path. Tradeoff: inline
+        # globals are dropped on EVERY scaled-mode run (the worker always
+        # hydrates from this redacted row) and on a default-mode re-enqueue
+        # after a restart — reference STORED global variables by name for
+        # background runs rather than passing secrets inline. Only the
+        # default-mode live in-process run below uses the full ``request``.
         await job_service.update_job_metadata(job_id, {"request": self._redact_request(request)})
         # After create_job so an idempotent retry returns the existing job instead of
         # cancelling it; the new job is QUEUED, so the suspended-only query skips it.
@@ -488,7 +490,7 @@ async def sweep_orphans_on_startup(self) -> None:
         worker_lost + terminal event). QUEUED workflow rows never started, so
         under at-least-once we re-enqueue them onto this worker's executor with a
         reconstructed request. Best-effort per job so one bad row can't block the
-        rest. Redis backend reconciles via its own watchdog.
+        rest. The scaled backend reconciles via its worker-side watchdog.
         """
         if self._is_scaled_configured:
             return

src/backend/base/langflow/services/database/models/jobs/model.py

@@ -3,7 +3,7 @@
 from typing import Any
 from uuid import UUID, uuid4
 
-from sqlalchemy import JSON, Column, DateTime, Integer, String, Text, UniqueConstraint
+from sqlalchemy import JSON, Column, DateTime, Index, Integer, String, Text, UniqueConstraint
 from sqlalchemy import Enum as SQLEnum
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlmodel import Field, SQLModel
@@ -92,6 +92,12 @@ class JobBase(SQLModel):
 
 class Job(JobBase, table=True):  # type: ignore[call-arg]
     __tablename__ = "job"
+    # The scaled backend's claim poll (claim_next_queued_lease) filters on
+    # status + type and sorts by created_timestamp on every worker poll; the
+    # watchdog scans status + type on its interval. The job table has no
+    # retention, so without this composite the hottest query degrades into an
+    # ever-growing scan+sort over terminal rows.
+    __table_args__ = (Index("ix_job_claim_scan", "status", "type", "created_timestamp"),)
 
 
 class JobEvent(SQLModel, table=True):  # type: ignore[call-arg]

src/lfx/src/lfx/services/settings/groups/runtime.py

@@ -97,7 +97,9 @@ class RuntimeSettings(BaseModel):
     the work queue: the API only persists the QUEUED row, and separate
     ``langflow worker`` processes lease-claim and run jobs against the SAME
     database, so background load runs off the API workers and scales
-    horizontally. No broker is needed — the database is the queue."""
+    horizontally. No broker is needed — the database is the queue. Every worker
+    must reach that one database: use Postgres for multi-host fleets (SQLite's
+    WAL is host-local, so scaled + SQLite only works on a single machine)."""
     background_poll_interval_s: float = Field(default=0.5, gt=0)
     """How often a scaled-mode event tail polls ``job_events`` for new durable
     frames while a job is live. Bounds the added reattach latency per milestone.