From 481a7a4ba8331bdce34356e7d08de053799d73eb Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Thu, 4 Jun 2026 20:53:39 -0300 Subject: [PATCH] feat(background-execution): emit worker_lost + orphan reconciliation metrics --- .../services/background_execution/metrics.py | 10 +++++ .../background_execution/redis_backend.py | 5 +++ .../base/langflow/services/jobs/service.py | 9 +++++ .../unit/services/jobs/test_sweep_liveness.py | 38 +++++++++++++++++++ 4 files changed, 62 insertions(+) diff --git a/src/backend/base/langflow/services/background_execution/metrics.py b/src/backend/base/langflow/services/background_execution/metrics.py index f24eef93a6..fb08c56773 100644 --- a/src/backend/base/langflow/services/background_execution/metrics.py +++ b/src/backend/base/langflow/services/background_execution/metrics.py @@ -7,6 +7,16 @@ from lfx.log.logger import logger from langflow.services.deps import get_telemetry_service +def current_backend() -> str: + """Best-effort: 'scaled' when the redis job queue is active, else 'default'. Never raises.""" + try: + from langflow.services.deps import get_settings_service + + return "scaled" if get_settings_service().settings.background_backend_is_scaled else "default" + except Exception: # noqa: BLE001 + return "default" + + def _emit(fn_name: str, name: str, labels: dict, value: float | None = None) -> None: try: ot = get_telemetry_service().ot 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 e51cf10f28..687b804d30 100644 --- a/src/backend/base/langflow/services/background_execution/redis_backend.py +++ b/src/backend/base/langflow/services/background_execution/redis_backend.py @@ -21,6 +21,7 @@ import contextlib import uuid from typing import TYPE_CHECKING, Any +from langflow.services.background_execution import metrics as bg_metrics from langflow.services.background_execution.redis_queue import RedisJobClaimQueue from langflow.services.database.models.jobs.model import JobStatus, SignalType from langflow.services.job_queue.service import RedisQueueWrapper @@ -200,6 +201,10 @@ class RedisBackgroundQueue: # (with a finished timestamp) explicitly. await self._job_service.update_job_status(job.job_id, JobStatus.FAILED, finished_timestamp=True) await self._job_service.set_error(job.job_id, {"type": "worker_lost"}) + # One emit pair per reconciled orphan (best-effort, never raises). + # This path only runs in scaled mode, so the backend label is literal. + bg_metrics.emit_job_failed(reason="worker_lost", backend="scaled") + bg_metrics.emit_orphan_reconciled(backend="scaled") await self.claim_queue.complete(job_id) return requeued diff --git a/src/backend/base/langflow/services/jobs/service.py b/src/backend/base/langflow/services/jobs/service.py index 3c790c23ae..87de4433cb 100644 --- a/src/backend/base/langflow/services/jobs/service.py +++ b/src/backend/base/langflow/services/jobs/service.py @@ -569,11 +569,20 @@ class JobService(Service): session.add(job) reconciled.append(job.job_id) await session.flush() + # Function-local import: background_execution imports from jobs, so a + # module-level import here would risk a cycle. The metrics module itself + # only pulls in deps + logger, so the local import is cheap and safe. + from langflow.services.background_execution import metrics as bg_metrics + + backend = bg_metrics.current_backend() # Append the terminal milestone via append_event (its own session) so the # IntegrityError/seq-collision retry applies: a seq collision with a # concurrent appender can no longer roll back the whole sweep. for job_id in reconciled: await self.append_event(job_id, "run_failed", dict(error_payload)) + # One emit pair per reconciled orphan (best-effort, never raises). + bg_metrics.emit_job_failed(reason="worker_lost", backend=backend) + bg_metrics.emit_orphan_reconciled(backend=backend) return reconciled async def get_latest_jobs_by_asset_ids(self, asset_ids: Sequence[UUID | str]) -> dict[UUID, Job]: diff --git a/src/backend/tests/unit/services/jobs/test_sweep_liveness.py b/src/backend/tests/unit/services/jobs/test_sweep_liveness.py index 89bebd09a6..d9ce678c21 100644 --- a/src/backend/tests/unit/services/jobs/test_sweep_liveness.py +++ b/src/backend/tests/unit/services/jobs/test_sweep_liveness.py @@ -62,3 +62,41 @@ async def test_sweep_still_fails_heartbeatless_in_progress(): events = await service.read_events(orphan) assert len(events) == 1 assert events[0].event_type == "run_failed" + + +@pytest.mark.usefixtures("client") +async def test_sweep_emits_worker_lost_metrics_without_raising(): + """Reconciling a stale-heartbeat orphan fires the worker_lost emit pair. + + The emit helpers are best-effort (they swallow their own errors), so the + contract here is: the sweep still reconciles the orphan AND the emit calls + do not raise into the sweep. We do not assert OTel counter values (the + metrics are write-only); a clean run that preserves the existing FAILED / + worker_lost behavior is the proof. + """ + service = JobService() + stale = uuid4() + fresh = uuid4() + + # Stale heartbeat -> real orphan that must be reconciled (and emitted for). + await service.create_job(job_id=stale, flow_id=uuid4(), user_id=uuid4()) + await service.update_job_status(stale, JobStatus.IN_PROGRESS) + old = (datetime.now(timezone.utc) - timedelta(seconds=300)).isoformat() + await service.update_job_metadata(stale, {"owner": "worker-dead", "heartbeat_at": old}) + + # Fresh heartbeat -> liveness gate keeps it out of reconciliation (no emit). + await service.create_job(job_id=fresh, flow_id=uuid4(), user_id=uuid4()) + await service.update_job_status(fresh, JobStatus.IN_PROGRESS) + await service.heartbeat(fresh, owner="worker-live") + + swept = await service.sweep_orphans(lease_ttl_s=30.0) + + assert stale in swept + assert fresh not in swept + + stale_job = await service.get_job_by_job_id(stale) + assert stale_job.status == JobStatus.FAILED + assert (stale_job.error or {}).get("type") == "worker_lost" + + fresh_job = await service.get_job_by_job_id(fresh) + assert fresh_job.status == JobStatus.IN_PROGRESS