diff --git a/src/backend/base/langflow/services/background_execution/factory.py b/src/backend/base/langflow/services/background_execution/factory.py index 8b90214ba7..4233fb0e76 100644 --- a/src/backend/base/langflow/services/background_execution/factory.py +++ b/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 diff --git a/src/backend/base/langflow/services/background_execution/redis_backend.py b/src/backend/base/langflow/services/background_execution/redis_backend.py index 980b06718c..035f699755 100644 --- a/src/backend/base/langflow/services/background_execution/redis_backend.py +++ b/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: diff --git a/src/backend/base/langflow/services/background_execution/service.py b/src/backend/base/langflow/services/background_execution/service.py index 2e1c3116d0..bafedfda17 100644 --- a/src/backend/base/langflow/services/background_execution/service.py +++ b/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 diff --git a/src/backend/tests/unit/services/background_execution/test_factory_selection.py b/src/backend/tests/unit/services/background_execution/test_factory_selection.py new file mode 100644 index 0000000000..a4ee6e8c52 --- /dev/null +++ b/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 diff --git a/src/backend/tests/unit/services/background_execution/test_scaled_end_to_end_realredis.py b/src/backend/tests/unit/services/background_execution/test_scaled_end_to_end_realredis.py new file mode 100644 index 0000000000..6ed8b3c668 --- /dev/null +++ b/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)