diff --git a/src/backend/tests/unit/services/background_execution/_subprocess_harness.py b/src/backend/tests/unit/services/background_execution/_subprocess_harness.py new file mode 100644 index 0000000000..53d9e01432 --- /dev/null +++ b/src/backend/tests/unit/services/background_execution/_subprocess_harness.py @@ -0,0 +1,264 @@ +"""Real-process hard-proof harness: a REAL ``langflow worker`` OS subprocess. + +Phase 4 proved the scaled claim/run/reattach/stop paths in-process (same event +loop). This closes that honest caveat with an ACTUAL separate process: + +* 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. + +The worker is launched with ``subprocess.Popen(["uv", "run", "langflow", +"worker", ...])`` and torn down in a finally. Every wait is bounded by a deadline +so a regression FAILS loudly instead of hanging. +""" + +from __future__ import annotations + +import contextlib +import os +import subprocess +import uuid +from dataclasses import dataclass, field +from urllib.parse import urlparse, urlunparse + +# Async driver the project actually ships for Postgres (see pyproject: +# sqlalchemy[postgresql_psycopg]). psycopg v3 accepts the ``options`` connect +# arg the DatabaseService sets for Postgres; asyncpg does not. +_PSYCOPG_PREFIX = "postgresql+psycopg://" + +# Default claim-queue keys the worker drains (RedisJobClaimQueue defaults). The +# test-side facade enqueues to these same keys; isolation comes from a dedicated +# redis DB index that is flushed per test. +_PENDING_KEY = "langflow:bg:pending" +_PROCESSING_KEY = "langflow:bg:processing" + + +def psycopg_url(raw: str) -> str: + """Normalize a Postgres URL to the async psycopg (v3) driver.""" + if raw.startswith(_PSYCOPG_PREFIX): + return raw + for prefix in ("postgresql+asyncpg://", "postgresql+psycopg2://", "postgresql://", "postgres://"): + if raw.startswith(prefix): + return _PSYCOPG_PREFIX + raw[len(prefix) :] + return raw + + +def redis_db_url(base_url: str, db_index: int) -> str: + """Return ``base_url`` pointed at a specific redis DB index (path = /N).""" + parsed = urlparse(base_url) + return urlunparse(parsed._replace(path=f"/{db_index}")) + + +@dataclass +class WorkerHarness: + """Bound state for a real-process worker proof. Build via ``setup``.""" + + db_url: str + redis_url: str + job_service: object + procs: list[subprocess.Popen] = field(default_factory=list) + _client: object | None = None + _prior_db_env: str | None = None + _prior_db_env_set: bool = False + + async def seed_real_flow(self, *, input_value: str = "hello") -> tuple[uuid.UUID, uuid.UUID]: + """Seed a real user + a real ChatInput->ChatOutput flow; return (user_id, flow_id). + + The flow ``data`` is produced by dumping a REAL connected Graph, so the + worker builds the same graph the canvas would. + """ + from langflow.services.database.models.flow.model import Flow + from langflow.services.database.models.user.model import User + from langflow.services.deps import session_scope + from lfx.components.input_output import ChatInput, ChatOutput + from lfx.graph import Graph + + chat_input = ChatInput(_id=f"ChatInput-{uuid.uuid4().hex[:6]}") + chat_input.set(input_value=input_value) + chat_output = ChatOutput(_id=f"ChatOutput-{uuid.uuid4().hex[:6]}") + chat_output.set(input_value=chat_input.message_response) + graph = Graph(start=chat_input, end=chat_output) + flow_data = graph.dump()["data"] + + user_id = uuid.uuid4() + flow_id = uuid.uuid4() + async with session_scope() as session: + session.add(User(id=user_id, username=f"wkr-{user_id}", password="x", is_active=True)) # noqa: S106 + session.add(Flow(id=flow_id, name=f"wkr-{flow_id}", data=flow_data, user_id=user_id)) + await session.commit() + return user_id, flow_id + + async def seed_slow_flow(self) -> tuple[uuid.UUID, uuid.UUID]: + """Seed a real multi-vertex no-LLM flow (MemoryChatbotNoLLM starter). + + Prompt + ChatInput + ChatOutput + Memory + TypeConverter cross several + vertex boundaries, so the build loop is reliably mid-flight long enough to + SIGKILL the worker or land a stop. Returns (user_id, flow_id). + """ + import json + from pathlib import Path + + from langflow.services.database.models.flow.model import Flow + from langflow.services.database.models.user.model import User + from langflow.services.deps import session_scope + + data_path = Path(__file__).parents[3] / "data" / "MemoryChatbotNoLLM.json" + flow_data = json.loads(data_path.read_text(encoding="utf-8"))["data"] + user_id = uuid.uuid4() + flow_id = uuid.uuid4() + async with session_scope() as session: + session.add(User(id=user_id, username=f"slow-{user_id}", password="x", is_active=True)) # noqa: S106 + session.add(Flow(id=flow_id, name=f"slow-{flow_id}", data=flow_data, user_id=user_id)) + await session.commit() + return user_id, flow_id + + async def submit_job(self, *, flow_id: uuid.UUID, user_id: uuid.UUID, input_value: str = "hello") -> uuid.UUID: + """Persist a QUEUED job + its request and enqueue it on the claim queue. + + Mirrors exactly what the scaled facade ``submit`` does (create_job + + update_job_metadata({"request": ...}) + LPUSH), so the worker hydrates a + production-shaped job row. + """ + job_id = uuid.uuid4() + await self.job_service.create_job(job_id=job_id, flow_id=flow_id, user_id=user_id) + request = {"flow_id": str(flow_id), "stream_protocol": "langflow", "input_value": input_value} + await self.job_service.update_job_metadata(job_id, {"request": request}) + await self.client.lpush(_PENDING_KEY, str(job_id)) + return job_id + + @property + def client(self): + if self._client is None: + from redis.asyncio import StrictRedis + + self._client = StrictRedis.from_url(self.redis_url) + return self._client + + def spawn_worker(self, *, idle_block_ms: int = 200) -> subprocess.Popen: + """Launch a REAL ``langflow worker`` OS subprocess against the shared store.""" + env = dict(os.environ) + env["LANGFLOW_JOB_QUEUE_TYPE"] = "redis" + env["LANGFLOW_DATABASE_URL"] = self.db_url + env["LANGFLOW_REDIS_QUEUE_URL"] = self.redis_url + # Keep the worker quiet + deterministic; no telemetry/tracing. + env["LANGFLOW_DEACTIVATE_TRACING"] = "true" + env["DO_NOT_TRACK"] = "true" + # A REAL OS subprocess is the whole point of this proof (closes the Phase + # 4 in-process caveat); the argv is a fixed literal, not untrusted input. + 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, + ) + self.procs.append(proc) + return proc + + async def wait_for_status(self, job_id: uuid.UUID, targets: set, *, timeout: float = 60.0): + """Poll the durable job row until its status is in ``targets``, bounded by ``timeout``. + + Raises AssertionError on timeout so a regression fails loudly. The worker + subprocess output is included in the failure message so a worker that + failed to boot or claim is diagnosable. + """ + import asyncio + + deadline = asyncio.get_event_loop().time() + timeout + last = None + while asyncio.get_event_loop().time() < deadline: + job = await self.job_service.get_job_by_job_id(job_id) + last = job.status if job else None + if last in targets: + return job + # Surface an early worker crash instead of waiting the full timeout. + dead = [p for p in self.procs if p.poll() is not None] + if dead and not any(p.poll() is None for p in self.procs): + break + await asyncio.sleep(0.2) + msg = f"job {job_id} never reached {targets}; last status={last}\n{self.drain_worker_output()}" + raise AssertionError(msg) + + def drain_worker_output(self) -> str: + """Return any buffered stdout/stderr from the worker subprocesses (best-effort).""" + chunks = [] + for proc in self.procs: + if proc.stdout is None: + continue + with contextlib.suppress(Exception): + if proc.poll() is None: + proc.terminate() + out = proc.stdout.read() + if out: + chunks.append(f"--- worker pid={proc.pid} output ---\n{out.decode(errors='replace')[-4000:]}") + return "\n".join(chunks) if chunks else "(no worker output captured)" + + async def teardown(self) -> None: + for proc in self.procs: + with contextlib.suppress(Exception): + if proc.poll() is None: + proc.terminate() + for proc in self.procs: + with contextlib.suppress(Exception): + proc.wait(timeout=10) + if proc.poll() is None: + with contextlib.suppress(Exception): + proc.kill() + proc.wait(timeout=5) + if self._client is not None: + with contextlib.suppress(Exception): + await self._client.flushdb() + with contextlib.suppress(Exception): + await self._client.aclose() + # Restore the LANGFLOW_DATABASE_URL env so it does not leak to later tests. + if self._prior_db_env_set: + os.environ["LANGFLOW_DATABASE_URL"] = self._prior_db_env # type: ignore[assignment] + else: + os.environ.pop("LANGFLOW_DATABASE_URL", None) + + +async def setup_worker_harness(pg_uri: str, redis_base_url: str, *, redis_db: int = 14) -> WorkerHarness: + """Stand up the shared store + a JobService bound to it. Caller owns teardown. + + Migrates (creates) the schema on the shared Postgres via the async psycopg + engine and binds ``session_scope`` to it, so both the test-side facade and + the worker subprocess read the SAME durable store. + """ + from langflow.services.database.factory import DatabaseServiceFactory + from langflow.services.deps import get_settings_service + from langflow.services.jobs.service import JobService + from lfx.services.manager import get_service_manager + from lfx.services.schema import ServiceType + + db_url = psycopg_url(pg_uri) + redis_url = redis_db_url(redis_base_url, redis_db) + + # The Settings.database_url ``mode='before'`` validator discards a directly + # assigned value and re-derives a config-dir sqlite path UNLESS + # LANGFLOW_DATABASE_URL is set. Set it in the test process so both the + # test-side engine AND the inherited worker subprocess bind to the SAME pg. + prior_db_env = os.environ.get("LANGFLOW_DATABASE_URL") + prior_db_env_set = "LANGFLOW_DATABASE_URL" in os.environ + os.environ["LANGFLOW_DATABASE_URL"] = db_url + + settings_service = get_settings_service() + settings_service.settings.database_url = db_url + manager = get_service_manager() + manager.services.pop(ServiceType.DATABASE_SERVICE, None) + db_service = DatabaseServiceFactory().create(settings_service) + manager.services[ServiceType.DATABASE_SERVICE] = db_service + await db_service.create_db_and_tables() + + harness = WorkerHarness( + db_url=db_url, + redis_url=redis_url, + job_service=JobService(), + _prior_db_env=prior_db_env, + _prior_db_env_set=prior_db_env_set, + ) + # Flush the dedicated redis DB so a prior run cannot leak claim-queue ids. + await harness.client.flushdb() + return harness diff --git a/src/backend/tests/unit/services/background_execution/test_worker_subprocess_e2e.py b/src/backend/tests/unit/services/background_execution/test_worker_subprocess_e2e.py new file mode 100644 index 0000000000..27642a8302 --- /dev/null +++ b/src/backend/tests/unit/services/background_execution/test_worker_subprocess_e2e.py @@ -0,0 +1,151 @@ +"""REAL-PROCESS hard proof: a separate ``langflow worker`` OS process runs a job. + +Closes the Phase 4 caveat that the worker ran in-process. Here an ACTUAL +``subprocess.Popen(["uv", "run", "langflow", "worker"])`` claims a job off real +redis, builds a REAL ChatInput->ChatOutput graph from the shared Postgres, runs +it to COMPLETED, and the API-side facade reattaches and replays the durable +milestones the worker persisted. Then a second job is killed with ``kill -9`` +mid-flight and the lease watchdog reconciles the stranded job. + +Requires both LANGFLOW_TEST_DATABASE_URI (real Postgres) and +LANGFLOW_TEST_REDIS_URL (real Redis). Skips otherwise. Every wait is bounded. +""" + +from __future__ import annotations + +import asyncio +import os +import signal + +import pytest +from langflow.services.background_execution.redis_backend import RedisBackgroundQueue +from langflow.services.background_execution.service import BackgroundExecutionService +from langflow.services.database.models.jobs.model import JobStatus +from langflow.services.deps import get_settings_service + +from ._subprocess_harness import setup_worker_harness + + +class _StubUser: + def __init__(self, user_id): + self.id = user_id + + +def _require_real_instances(): + pg = os.environ.get("LANGFLOW_TEST_DATABASE_URI") + redis = os.environ.get("LANGFLOW_TEST_REDIS_URL") + if not pg: + pytest.skip("LANGFLOW_TEST_DATABASE_URI not set; real-process proof requires real Postgres") + if not redis: + pytest.skip("LANGFLOW_TEST_REDIS_URL not set; real-process proof requires real Redis") + return pg, redis + + +@pytest.mark.hard_proof +async def test_real_worker_subprocess_runs_job_and_facade_reattaches(): + """A REAL worker process claims + builds + completes a real graph; facade reattaches.""" + pg, redis = _require_real_instances() + harness = await setup_worker_harness(pg, redis, redis_db=14) + try: + from redis.asyncio import StrictRedis + + user_id, flow_id = await harness.seed_real_flow(input_value="hello-subproc") + job_id = await harness.submit_job(flow_id=flow_id, user_id=user_id, input_value="hello-subproc") + + # Sanity: the API process did NOT run it — it sits QUEUED on the claim queue. + queued = await harness.job_service.get_job_by_job_id(job_id) + assert queued.status == JobStatus.QUEUED + + # Launch the REAL separate worker OS process. + proc = harness.spawn_worker(idle_block_ms=200) + assert proc.poll() is None, "worker subprocess failed to start" + + # The SEPARATE process claims + builds + completes the real graph. + job = await harness.wait_for_status(job_id, {JobStatus.COMPLETED}, timeout=90.0) + assert job.status == JobStatus.COMPLETED + assert job.result is not None + + # Durable milestones the worker persisted are replayable by ANY API replica. + events = await harness.job_service.read_events(job_id, after_seq=0) + types = [e.event_type for e in events] + assert types, "worker persisted no durable milestones" + assert "end" in types, f"no terminal milestone; types={types}" + # Token deltas stay ephemeral (not durable). + assert "token" not in types, "token deltas must not be persisted" + + # API-side facade (scaled backend on a SEPARATE redis connection) reattaches + # and replays the durable milestones the worker wrote — cross-process. + client_b = StrictRedis.from_url(harness.redis_url) + backend_b = RedisBackgroundQueue( + client=client_b, job_service=harness.job_service, stream_ttl=60, startup_grace_s=5.0 + ) + get_settings_service().settings.job_queue_type = "redis" + facade_b = BackgroundExecutionService(settings_service=get_settings_service(), backend=backend_b) + try: + + async def _collect(): + return [c async for c in facade_b.events(job_id, last_event_id=None, user=_StubUser(user_id))] + + replayed = await asyncio.wait_for(_collect(), timeout=15.0) + finally: + await client_b.aclose() + body = b"".join(replayed) + assert b"add_message" in body or b"end_vertex" in body, "facade replayed no durable worker frames" + + print( # noqa: T201 + f"PROOF[subprocess/e2e]: REAL `langflow worker` (pid={proc.pid}) claimed + built + " + f"completed a real graph; facade reattach replayed {len(replayed)} durable frames; " + f"durable types={types}" + ) + finally: + await harness.teardown() + + +@pytest.mark.hard_proof +async def test_real_worker_kill9_midjob_reconciled_by_watchdog(): + """Kill -9 a REAL worker mid-job; the lease watchdog fails the stranded job worker_lost.""" + pg, redis = _require_real_instances() + harness = await setup_worker_harness(pg, redis, redis_db=13) + try: + # A real multi-vertex flow so the worker is reliably mid-build when we + # SIGKILL it. Default (not retry-safe) => at-most-once: worker_lost. + user_id, flow_id = await harness.seed_slow_flow() + job_id = await harness.submit_job(flow_id=flow_id, user_id=user_id, input_value="kill9") + + proc = harness.spawn_worker(idle_block_ms=200) + + # Wait until the worker has CLAIMED + started the job (IN_PROGRESS), then + # kill -9 it so it cannot finalize the row (true worker death mid-flight). + await harness.wait_for_status(job_id, {JobStatus.IN_PROGRESS, JobStatus.COMPLETED}, timeout=90.0) + job = await harness.job_service.get_job_by_job_id(job_id) + if job.status == JobStatus.COMPLETED: + pytest.skip("graph completed before we could kill the worker mid-flight (too fast on this host)") + + os.kill(proc.pid, signal.SIGKILL) + proc.wait(timeout=10) + assert proc.returncode is not None, "worker did not die on SIGKILL" + # The durable row is now a stranded IN_PROGRESS orphan with a live lease on + # the redis processing list (the dead worker never released it). + + # A fresh backend instance acts as a second replica / restarted worker and + # runs the lease watchdog reconcile. Default (not retry-safe) => worker_lost. + from redis.asyncio import StrictRedis + + client = StrictRedis.from_url(harness.redis_url) + backend = RedisBackgroundQueue(client=client, job_service=harness.job_service, startup_grace_s=5.0) + try: + stranded = await backend.claim_queue.processing_ids() + assert str(job_id) in stranded, f"job not on the processing list after kill: {stranded}" + await backend.requeue_lost() + finally: + await client.aclose() + + job = await harness.job_service.get_job_by_job_id(job_id) + assert job.status == JobStatus.FAILED, f"stranded job not failed: {job.status}" + assert (job.error or {}).get("type") == "worker_lost", f"error={job.error}" + print( # noqa: T201 + f"PROOF[subprocess/kill9]: kill -9 the REAL worker (pid={proc.pid}) mid-job -> " + f"lease watchdog reconciled stranded job to FAILED(worker_lost)" + ) + finally: + await harness.teardown() diff --git a/src/backend/tests/unit/services/background_execution/test_worker_subprocess_stop.py b/src/backend/tests/unit/services/background_execution/test_worker_subprocess_stop.py new file mode 100644 index 0000000000..81b4837218 --- /dev/null +++ b/src/backend/tests/unit/services/background_execution/test_worker_subprocess_stop.py @@ -0,0 +1,90 @@ +"""REAL-PROCESS hard proof: pub/sub STOP from the API reaches a separate worker. + +Closes the Phase 4 caveat that stop was delivered in-process. Here an ACTUAL +``langflow worker`` OS subprocess runs a slowish multi-vertex real flow; the +API-side facade calls ``stop_job`` (durable STOP signal + redis pub/sub +fast-path); the SEPARATE worker process honors the stop and the job ends +CANCELLED with the STOP signal consumed. + +Requires real Postgres + real Redis. Skips otherwise. Every wait is bounded. +""" + +from __future__ import annotations + +import os + +import pytest +from langflow.services.background_execution.redis_backend import RedisBackgroundQueue +from langflow.services.background_execution.service import BackgroundExecutionService +from langflow.services.database.models.jobs.model import JobStatus, SignalType +from langflow.services.deps import get_settings_service + +from ._subprocess_harness import setup_worker_harness + + +class _StubUser: + def __init__(self, user_id): + self.id = user_id + + +def _require_real_instances(): + pg = os.environ.get("LANGFLOW_TEST_DATABASE_URI") + redis = os.environ.get("LANGFLOW_TEST_REDIS_URL") + if not pg: + pytest.skip("LANGFLOW_TEST_DATABASE_URI not set; real-process proof requires real Postgres") + if not redis: + pytest.skip("LANGFLOW_TEST_REDIS_URL not set; real-process proof requires real Redis") + return pg, redis + + +@pytest.mark.hard_proof +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() + harness = await setup_worker_harness(pg, redis, redis_db=12) + try: + # A real multi-vertex no-LLM flow whose build crosses several vertex + # boundaries — each a point where the runner polls the durable STOP. + user_id, flow_id = await harness.seed_slow_flow() + job_id = await harness.submit_job(flow_id=flow_id, user_id=user_id, input_value="stop-me") + + # API-side scaled facade (writes the durable STOP + publishes pub/sub). + from redis.asyncio import StrictRedis + + api_client = StrictRedis.from_url(harness.redis_url) + backend = RedisBackgroundQueue(client=api_client, job_service=harness.job_service, startup_grace_s=5.0) + get_settings_service().settings.job_queue_type = "redis" + facade = BackgroundExecutionService(settings_service=get_settings_service(), backend=backend) + + proc = harness.spawn_worker(idle_block_ms=200) + + try: + # Wait until the SEPARATE worker has started the job (IN_PROGRESS). + await harness.wait_for_status( + job_id, {JobStatus.IN_PROGRESS, JobStatus.COMPLETED, JobStatus.CANCELLED}, timeout=90.0 + ) + job = await harness.job_service.get_job_by_job_id(job_id) + if job.status != JobStatus.IN_PROGRESS: + pytest.skip(f"job reached {job.status} before stop could land (too fast on this host)") + + # API-side stop: durable STOP signal + redis pub/sub fast-path to the + # owning worker process. + await facade.stop_job(job_id, _StubUser(user_id)) + + # The SEPARATE worker honors the stop at a vertex boundary -> CANCELLED. + final = await harness.wait_for_status(job_id, {JobStatus.CANCELLED}, timeout=60.0) + assert final.status == JobStatus.CANCELLED, f"worker did not honor stop: {final.status}" + + # The STOP signal was consumed by the worker's terminal reconcile. + leftover = [ + s for s in await harness.job_service.unconsumed_signals(job_id) if s.signal_type == SignalType.STOP + ] + assert leftover == [], "STOP signal was not consumed by the worker" + print( # noqa: T201 + f"PROOF[subprocess/stop]: API stop_job -> SEPARATE worker (pid={proc.pid}) honored " + f"durable STOP + pub/sub -> job CANCELLED, signal consumed" + ) + finally: + await api_client.aclose() + finally: + await harness.teardown()