Repository navigation
Conversation
ThreadJobExecutor never called telemetry.metrics.job_started()/job_ended(), so lk_agents_active_job_count had no writer when jobs ran under JobExecutorType.THREAD (the Windows default and the console/Jupyter path). The gauge's only writers were ProcJobExecutor.launch_job/_supervise_task, leaving the metric at 0 (and the labeled series absent entirely) for the whole lifetime of thread-executed jobs. Mirror the process executor's accounting: increment when a job is launched, roll back the increment if the StartJobRequest cannot be sent, and decrement once the runner thread exits while a job was assigned. Add a regression test that spies on job_started()/job_ended() and asserts the thread executor calls them around a job lifecycle.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| except Exception: | ||
| self._running_job = None | ||
| self._job_status = None | ||
| metrics.job_ended() |
There was a problem hiding this comment.
π‘ Failed dispatch double-decrements active jobs
If the runner exits during launch_job's send, _main_task can decrement the gauge before the send fails. The exception handler decrements it again, leaving a negative active-job count.
Learn more
The active-job gauge counts jobs across executors, so each increment needs exactly one decrement. A failed send can overlap with the runner's exit: the join callback wakes _main_task, which observes _running_job and calls job_ended. If the pending send then raises, this handler decrements the same job again. The resulting negative gauge persists and offsets counts from other jobs.
Example: A runner exits while the parent is awaiting the send of job A. The join handler reduces the count from 1 to 0; the send failure handler then reduces it from 0 to -1.
Recommended fix: Track whether the increment remains owned by launch_job or has been settled by _main_task, and make the decrement idempotent across both paths. Include a test that pauses a failing send until after the join callback runs.
Was this helpful? React with π or π to provide feedback.
| try: | ||
| await channel.asend_message(self._pch, start_req) | ||
| except Exception: |
There was a problem hiding this comment.
π‘ Cancelled dispatch leaves a phantom job
Cancelling launch_job during the send bypasses its Exception handler, leaving the gauge incremented and _running_job set. If the runner remains idle, the active-job count stays high until it exits.
Learn more
The parent writes a serialized job request and awaits the channel drain. Cancellation at that await raises asyncio.CancelledError, which the new except Exception does not catch. If cancellation happens before the request is delivered, the runner remains idle while the incremented gauge counts a job. _main_task only decrements it when the runner exits.
Example: A caller cancels dispatch for job A while its socket is draining and the runner has not received the complete request. No job starts, yet the gauge remains at 1 until the runner is closed.
Recommended fix: Define cancellation semantics for sends that may already have delivered a request. Either shield dispatch until its outcome is known, or handle cancellation while closing the executor and settling the gauge exactly once after the runner exits; avoid marking a potentially delivered job as unstarted.
Was this helpful? React with π or π to provide feedback.
The fake job's entrypoint could complete between launch_job() and the 'started' assertion, letting _main_task record 'ended' first. CI hit that window; keep the entrypoint parked on a threading.Event until the test has observed the running state, then release it and assert the final sequence.
This branch has not been deployed
1 participant
fix(ipc): update the active-job gauge in the thread job executor
What breaks
lk_agents_active_job_countnever moves for the thread job executor: the gauge reports0while a job is running, and stays0after it ends.ThreadJobExecutornever callstelemetry.metrics.job_started()/job_ended(), unlikeProcJobExecutorwhich calls them inlaunch_joband its supervise task.The thread executor is the default on Windows (
worker.pyfalls back toJobExecutorType.THREAD) and is forced by the devmode CLI, so on those deployments the active-job metric is permanently zero β dashboards and autoscaling rules that read it see an idle worker while jobs are running.Root cause
launch_job()(livekit-agents/livekit/agents/ipc/job_thread_executor.py) starts the runner and sends theStartJobRequestwithout touching the metrics module, and_main_task()completes after_join_futwithout decrementing. The gauge is therefore only ever updated by the process executor path.Fix
Mirror
ProcJobExecutorexactly:launch_job(): callmetrics.job_started()once the initialized check passes; if sending theStartJobRequestfails, reset_running_job/_job_statusand callmetrics.job_ended()before re-raising (no leaked increment)._main_task(): afterawait self._join_fut, if a job was running, callmetrics.job_ended().Tests
New test
tests/test_ipc.py::test_thread_executor_updates_active_job_metricspies ontelemetry.metrics.job_started/job_ended, runs a realThreadJobExecutorthrough_create_thread_proc, and asserts the gauge is incremented while the entrypoint is live and decremented afteraclose().AssertionError: job_started must be called while a job is running(both metric functions never called; the job provably runs).tests/test_ipc.py+tests/test_ipc_preload.pygreen (31 passed); ruff clean.The test is marked
no_concurrent(it monkeypatches process-global metric functions) so parallel thread-executor tests cannot pollute the spy. The change does not overlap the in-flight #7899 hunks in the same file (initialize()/aclose()/_monitor_task()); both patches apply independently.