feat(background): wire scaled redis backend into the facade + e2e claim-run-reattach proof

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

@@ -20,3 +20,20 @@ class BackgroundExecutionServiceFactory(ServiceFactory):
     @override
     def create(self, settings_service: SettingsService):
         return BackgroundExecutionService(settings_service)
+
+
+def select_background_backend(settings, *, client, job_service):
+    """Pick the scaled background backend per settings, or None for the default.
+
+    Scaled (redis) when ``settings.background_backend_is_scaled`` is True — a
+    separate ``langflow worker`` process drains the claim queue and publishes
+    live frames to redis Streams. Otherwise return None: the facade owns the
+    in-process executor + in-memory bus path directly (no separate backend
+    object). Backend selection follows the existing job_queue_type/redis
+    settings; see ``Settings.background_backend_is_scaled``.
+    """
+    if settings.background_backend_is_scaled:
+        from langflow.services.background_execution.redis_backend import RedisBackgroundQueue
+
+        return RedisBackgroundQueue(client=client, job_service=job_service)
+    return None

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

@@ -85,6 +85,16 @@ class RedisBackgroundQueue:
         """Hand a queued job id to a worker process via the claim queue."""
         await self.claim_queue.enqueue(job_id)
 
+    # ----------------------------------------------------------- worker claim
+
+    async def claim(self, *, block_ms: int = 1000) -> str | None:
+        """Claim a job id off the queue for a worker (delegates to the claim queue)."""
+        return await self.claim_queue.claim(block_ms=block_ms)
+
+    async def complete(self, job_id: str) -> None:
+        """Release a worker's lease on a job (delegates to the claim queue)."""
+        await self.claim_queue.complete(job_id)
+
     # ---------------------------------------------------------------- control
 
     async def stop(self, job_id: str, *, marker_ttl: int = 60) -> None:

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

@@ -58,21 +58,51 @@ class BackgroundExecutionService(Service):
         settings_service: SettingsService,
         *,
         frame_source_factory: FrameSourceFactory | None = None,
+        backend: Any | None = None,
     ) -> None:
         self.settings_service = settings_service
         self._settings = settings_service.settings
-        self._is_redis = self._settings.job_queue_type == "redis"
+        self._is_redis = self._settings.background_backend_is_scaled
+        # Scaled backend: the redis claim queue + Streams live bus + DB replay.
+        # When configured (redis), the API process only enqueues — a separate
+        # ``langflow worker`` process drains the queue and runs the JobRunner.
+        # Injected in tests; otherwise built lazily here from settings when the
+        # scaled backend is configured. In the default (asyncio) path it stays
+        # None and the in-process executor below runs jobs inside the API.
+        if backend is None and self._is_redis:
+            backend = self._build_scaled_backend()
+        self._backend = backend
         self._executor = InProcessExecutor(max_concurrency=self._settings.background_max_concurrency)
         self._bus = InMemoryLiveBus()
         # Injected in tests; defaulted to the real build loop by the route wiring.
         self._frame_source_factory = frame_source_factory
         self.set_ready()
 
+    @property
+    def _scaled(self) -> bool:
+        """True when a redis-backed scaled backend is wired behind this facade."""
+        return self._backend is not None
+
+    def _build_scaled_backend(self) -> Any:
+        """Build the redis-backed scaled backend from settings.
+
+        Reuses the worker's redis-client resolution (URL → host/port/db with the
+        cache-redis fallbacks) so the API enqueues to the exact redis a worker
+        drains, and ``select_background_backend`` so selection follows
+        ``background_backend_is_scaled``. Returns None in the default path.
+        """
+        from langflow.services.background_execution.factory import select_background_backend
+        from langflow.services.background_execution.worker import _build_redis_client
+        from langflow.services.deps import get_job_service
+
+        client = _build_redis_client(self._settings)
+        return select_background_backend(self._settings, client=client, job_service=get_job_service())
+
     async def start(self) -> None:
-        if self._is_redis:
-            # Phase 3 fills this in (redis queue + worker process + streams bus).
-            msg = "Redis background backend is not implemented yet (Phase 3)."
-            raise NotImplementedError(msg)
+        # Scaled mode: nothing to start in the API process — the worker process
+        # owns execution. (No NotImplementedError: the redis backend is wired.)
+        if self._scaled:
+            return
         await self._executor.start()
 
     async def stop(self) -> None:
@@ -108,10 +138,16 @@ class BackgroundExecutionService(Service):
             raise
         # Persist the submit request on the job row so a QUEUED job that survives
         # a restart is re-enqueued with its ORIGINAL inputs (input_value, tweaks,
-        # globals, etc.), not a reconstructed default. The startup sweep reads it
-        # back via ``_reconstruct_request``.
+        # globals, etc.), not a reconstructed default. The worker / startup sweep
+        # read it back via ``_reconstruct_request``.
         await job_service.update_job_metadata(job_id, {"request": request})
-        await self._enqueue(job_id=job_id, flow_id=flow_id, request=request, user=user)
+        if self._scaled:
+            # Scaled mode: hand the QUEUED job id to a worker via the redis claim
+            # queue. The DB row stays the system of record; the API does NOT run
+            # the flow. The worker hydrates the request from the job row.
+            await self._backend.enqueue(str(job_id))
+        else:
+            await self._enqueue(job_id=job_id, flow_id=flow_id, request=request, user=user)
         return job_id
 
     @staticmethod
@@ -182,12 +218,29 @@ class BackgroundExecutionService(Service):
         # is empty, so ``reattach`` would replay durable rows then block forever on
         # ``while True: queue.get()`` waiting for a live tail that will never come.
         # Decide "finished" off the persisted status (the cross-restart source of
-        # truth), replay, and return.
+        # truth), replay, and return. The same holds cross-replica in scaled mode:
+        # a terminal job has nothing live left on the redis Stream.
         if job.status in _TERMINAL_STATUSES:
             for frame in await read_durable(last_seq):
                 yield frame.data
             return
 
+        # Scaled mode: any API replica serves reattach by replaying durable
+        # job_events (from the DB) then tailing the shared redis Stream. The
+        # backend yields durable event rows (carry .seq) and live _StreamFrames
+        # (payload is already SSE bytes the worker's RedisStreamLiveBus XADDed).
+        if self._scaled:
+            async for item in self._backend.events(str(job_id), last_event_id=last_seq):
+                seq = getattr(item, "seq", None)
+                if seq is not None:
+                    # Durable milestone row — re-frame through the SSE formatter
+                    # so replayed bytes match live frames (Last-Event-ID resume).
+                    yield self._row_to_frame(item)
+                else:
+                    # Live ephemeral frame from the Stream tail — already framed.
+                    yield item.payload
+            return
+
         async for frame in self._bus.reattach(str(job_id), last_seq=last_seq, read_durable=read_durable):
             yield frame.data
 
@@ -216,6 +269,13 @@ class BackgroundExecutionService(Service):
     async def stop_job(self, job_id: UUID, user: UserRead) -> None:
         # Enforces ownership (raises PermissionError on a cross-user/unknown job).
         await self._validate(job_id, user)
+        if self._scaled:
+            # Scaled mode: the owning worker runs in another process. backend.stop
+            # writes the durable STOP signal (the source of truth the worker polls
+            # at vertex boundaries) AND publishes the redis pub/sub fast-path so
+            # the owning worker reacts immediately instead of waiting for its poll.
+            await self._backend.stop(str(job_id))
+            return
         job_service = get_job_service()
         # Always write the durable STOP signal — even when the row currently
         # reads a finished status. An in-flight runner can write its terminal

src/backend/tests/unit/services/background_execution/test_factory_selection.py

@@ -0,0 +1,47 @@
+"""The facade factory selects the scaled redis backend when configured."""
+
+from __future__ import annotations
+
+import pytest
+from langflow.services.background_execution.factory import select_background_backend
+from langflow.services.background_execution.redis_backend import RedisBackgroundQueue
+from langflow.services.background_execution.service import BackgroundExecutionService
+from langflow.services.deps import get_settings_service
+
+
+def test_factory_selects_scaled_backend_when_configured():
+    class _Settings:
+        background_backend_is_scaled = True
+
+    backend = select_background_backend(_Settings(), client=object(), job_service=object())
+    assert isinstance(backend, RedisBackgroundQueue)
+
+
+def test_factory_returns_none_for_default_backend():
+    class _Settings:
+        background_backend_is_scaled = False
+
+    # The default (in-process) backend is owned by the facade itself, so the
+    # selector returns None: "no scaled backend, use the in-process path".
+    backend = select_background_backend(_Settings(), client=object(), job_service=object())
+    assert backend is None
+
+
+@pytest.mark.usefixtures("client")
+def test_facade_builds_scaled_backend_from_settings():
+    settings_service = get_settings_service()
+    settings = settings_service.settings
+    original = settings.job_queue_type
+    try:
+        # Default (asyncio): no scaled backend behind the facade.
+        settings.job_queue_type = "asyncio"
+        default_facade = BackgroundExecutionService(settings_service=settings_service)
+        assert default_facade._scaled is False
+
+        # Scaled (redis): the facade builds the redis backend itself.
+        settings.job_queue_type = "redis"
+        scaled_facade = BackgroundExecutionService(settings_service=settings_service)
+        assert scaled_facade._scaled is True
+        assert isinstance(scaled_facade._backend, RedisBackgroundQueue)
+    finally:
+        settings.job_queue_type = original

src/backend/tests/unit/services/background_execution/test_scaled_end_to_end_realredis.py

@@ -0,0 +1,118 @@
+"""Real-redis + real-DB end-to-end: facade submit (scaled) -> worker runs -> COMPLETED.
+
+This is the integration capstone. A scaled-mode facade submits a job (which only
+enqueues onto the redis claim queue — no in-process execution), a worker loop
+claims it and runs the real JobRunner via WorkerJobRunner with a scripted frame
+source (no LLM), and the durable job row reaches COMPLETED with a replayable
+event log. A SECOND facade instance (a different API replica) then reattaches and
+replays the durable milestones — cross-replica, no gap.
+
+Uses real redis (LANGFLOW_TEST_REDIS_URL) + the migrated test DB. No mocks of our
+code: the only injected piece is the scripted frame source standing in for a live
+graph build (same pattern as test_service.py).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from typing import TYPE_CHECKING
+
+import pytest
+from langflow.services.background_execution.redis_backend import RedisBackgroundQueue
+from langflow.services.background_execution.redis_live_bus import RedisStreamLiveBus
+from langflow.services.background_execution.service import BackgroundExecutionService
+from langflow.services.background_execution.worker import WorkerJobRunner, run_worker_loop
+from langflow.services.database.models.jobs.model import JobStatus
+from langflow.services.deps import get_job_service, get_settings_service
+from langflow.services.job_queue.service import _STREAM_PREFIX
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+pytestmark = pytest.mark.usefixtures("client")
+
+
+def _frame(event_type: str, data: dict) -> tuple[bytes, str]:
+    return (json.dumps({"event": event_type, "data": data}).encode("utf-8"), event_type)
+
+
+async def _scripted_source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]:
+    yield _frame("build_start", {})
+    yield _frame("end_vertex", {"id": "n1"})
+    yield _frame("end", {})
+
+
+def _scaled_settings():
+    settings = get_settings_service().settings
+    # Force scaled selection for this facade instance without touching global env.
+    settings.job_queue_type = "redis"
+    return settings
+
+
+async def test_submit_then_worker_runs_to_completion(real_redis, real_redis_url, active_user):
+    from redis.asyncio import StrictRedis
+
+    jobs = get_job_service()
+    settings = _scaled_settings()
+    prefix = real_redis._bgtest_prefix
+
+    # API replica A: scaled facade wired to the real redis backend.
+    backend_a = RedisBackgroundQueue(client=real_redis, job_service=jobs, startup_grace_s=10.0)
+    backend_a.claim_queue.pending_key = f"{prefix}pending"
+    backend_a.claim_queue.processing_key = f"{prefix}processing"
+    facade_a = BackgroundExecutionService(settings_service=get_settings_service(), backend=backend_a)
+
+    flow_id = uuid.uuid4()
+    job_id = await facade_a.submit(flow_id=flow_id, request={"stream_protocol": "langflow"}, user=active_user)
+
+    # In scaled mode the API process must NOT execute — the job sits QUEUED on the
+    # claim queue until a worker picks it up.
+    queued = await jobs.get_job_by_job_id(job_id)
+    assert queued.status == JobStatus.QUEUED
+    pending = await real_redis.lrange(backend_a.claim_queue.pending_key, 0, -1)
+    assert str(job_id).encode() in pending
+
+    # Worker process (separate task): claim + run the real JobRunner with a
+    # scripted source, publishing live frames to redis Streams.
+    live_bus = RedisStreamLiveBus(real_redis, ttl=60)
+    worker_runner = WorkerJobRunner(
+        settings=settings,
+        live_bus=live_bus,
+        frame_source_factory=lambda **_kw: _scripted_source,
+    )
+    stop_event = asyncio.Event()
+
+    async def stop_when_done():
+        for _ in range(200):
+            refreshed = await jobs.get_job_by_job_id(job_id)
+            if refreshed.status == JobStatus.COMPLETED:
+                stop_event.set()
+                return
+            await asyncio.sleep(0.05)
+        stop_event.set()
+
+    driver = asyncio.create_task(stop_when_done())
+    await run_worker_loop(backend_a, worker_runner, stop_event=stop_event, idle_block_ms=50)
+    await driver
+
+    refreshed = await jobs.get_job_by_job_id(job_id)
+    assert refreshed.status == JobStatus.COMPLETED
+    assert refreshed.result is not None
+    # Lease released.
+    assert str(job_id) not in await backend_a.claim_queue.processing_ids()
+
+    # API replica B: a different facade instance on a SEPARATE connection that
+    # never ran the job reattaches and replays the durable milestones.
+    client_b = StrictRedis.from_url(real_redis_url)
+    backend_b = RedisBackgroundQueue(client=client_b, job_service=jobs, stream_ttl=60, startup_grace_s=10.0)
+    facade_b = BackgroundExecutionService(settings_service=get_settings_service(), backend=backend_b)
+    try:
+        seen = [chunk async for chunk in facade_b.events(job_id, last_event_id=None, user=active_user)]
+    finally:
+        await client_b.aclose()
+        await real_redis.delete(f"{_STREAM_PREFIX}{job_id}")
+
+    assert any(b"build_start" in c for c in seen)
+    assert any(b"end_vertex" in c for c in seen)