mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-25 18:09:16 +08:00
test(background): real langflow worker OS subprocess hard proofs (e2e, kill -9 watchdog, pub/sub stop)
This commit is contained in:
ogabrielluiz
@ -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
|
||||
@ -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()
|
||||
@ -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()
|
||||
Reference in New Issue
Block a user