fix(ipc): back off failed process initialization

livekit-agents/livekit/agents/ipc/proc_pool.py

@@ -23,6 +23,8 @@
 ]
 
 MAX_CONCURRENT_INITIALIZATIONS = min(math.ceil(get_cpu_monitor().cpu_count()), 4)
+PROCESS_INIT_RETRY_INITIAL_DELAY = 1.0
+PROCESS_INIT_RETRY_MAX_DELAY = 30.0
 
 
 class ProcPool(utils.EventEmitter[EventTypes]):
@@ -66,14 +68,16 @@ def __init__(
         self._init_sem = asyncio.Semaphore(MAX_CONCURRENT_INITIALIZATIONS)
         self._warmed_proc_queue = asyncio.Queue[JobExecutor]()
         self._executors: list[JobExecutor] = []
-        self._spawn_tasks: set[asyncio.Task[None]] = set()
+        self._spawn_tasks: set[asyncio.Task[bool]] = set()
         self._close_tasks: set[asyncio.Task[None]] = set()
         self._monitor_tasks: set[asyncio.Task[None]] = set()
         self._started = False
         self._closed = False
 
         self._idle_ready = asyncio.Event()
         self._jobs_waiting_for_process = 0
+        self._process_init_retry_delay = 0.0
+        self._next_idle_process_spawn_at = 0.0
 
     @property
     def processes(self) -> list[JobExecutor]:
@@ -110,6 +114,47 @@ async def aclose(self) -> None:
         self._closed = True
         await aio.cancel_and_wait(self._main_atask)
 
+    def _spawn_process(self) -> None:
+        task = asyncio.create_task(self._proc_spawn_task())
+        self._spawn_tasks.add(task)
+        task.add_done_callback(self._on_process_spawn_done)
+
+    def _on_process_spawn_done(self, task: asyncio.Task[bool]) -> None:
+        self._spawn_tasks.discard(task)
+        if task.cancelled():
+            return
+
+        exception = task.exception()
+        if self._closed:
+            return
+
+        initialized = exception is None and task.result()
+        if initialized:
+            self._process_init_retry_delay = 0.0
+            # Keep any active deadline set by another failed task in the same batch.
+            return
+
+        now = self._loop.time()
+        started_retry_round = now >= self._next_idle_process_spawn_at
+        if started_retry_round:
+            retry_delay = min(
+                max(PROCESS_INIT_RETRY_INITIAL_DELAY, self._process_init_retry_delay * 2),
+                PROCESS_INIT_RETRY_MAX_DELAY,
+            )
+        else:
+            retry_delay = max(PROCESS_INIT_RETRY_INITIAL_DELAY, self._process_init_retry_delay)
+
+        self._process_init_retry_delay = retry_delay
+        self._next_idle_process_spawn_at = max(
+            self._next_idle_process_spawn_at,
+            now + retry_delay,
+        )
+        if started_retry_round:
+            logger.warning(
+                "backing off idle process replenishment after an initialization failure",
+                extra={"retry_delay": retry_delay},
+            )
+
     async def _acquire_proc(self, job_id: str) -> JobExecutor:
         MAX_ACQUIRE_ATTEMPTS = 3
 
@@ -119,9 +164,7 @@ async def _acquire_proc(self, job_id: str) -> JobExecutor:
                 and len(self._spawn_tasks) < self._jobs_waiting_for_process
             ):
                 # spawn a new process if there are no idle processes
-                task = asyncio.create_task(self._proc_spawn_task())
-                self._spawn_tasks.add(task)
-                task.add_done_callback(self._spawn_tasks.discard)
+                self._spawn_process()
 
             if self._warmed_proc_queue.empty():
                 logger.warning(
@@ -199,7 +242,7 @@ def target_idle_processes(self) -> int:
         return self._target_idle_processes
 
     @utils.log_exceptions(logger=logger)
-    async def _proc_spawn_task(self) -> None:
+    async def _proc_spawn_task(self) -> bool:
         proc: JobExecutor
         if self._job_executor_type == JobExecutorType.THREAD:
             proc = job_thread_executor.ThreadJobExecutor(
@@ -262,11 +305,12 @@ async def _proc_spawn_task(self) -> None:
             self._executors.remove(proc)
             await proc.aclose()
             self.emit("process_closed", proc)
-            return
+            return False
 
         monitor_task = asyncio.create_task(self._monitor_process_task(proc))
         self._monitor_tasks.add(monitor_task)
         monitor_task.add_done_callback(self._monitor_tasks.discard)
+        return True
 
     @utils.log_exceptions(logger=logger)
     async def _monitor_process_task(self, proc: JobExecutor) -> None:
@@ -281,16 +325,20 @@ async def _main_task(self) -> None:
         try:
             while not self._closed:
                 current_pending = self._warmed_proc_queue.qsize() + len(self._spawn_tasks)
-                target = max(
-                    min(self._target_idle_processes, self._default_num_idle_processes),
-                    self._jobs_waiting_for_process,
+                idle_target = min(
+                    self._target_idle_processes,
+                    self._default_num_idle_processes,
                 )
-                to_spawn = target - current_pending
+                jobs_to_spawn = max(self._jobs_waiting_for_process - current_pending, 0)
+
+                for _ in range(jobs_to_spawn):
+                    self._spawn_process()
 
-                for _ in range(to_spawn):
-                    task = asyncio.create_task(self._proc_spawn_task())
-                    self._spawn_tasks.add(task)
-                    task.add_done_callback(self._spawn_tasks.discard)
+                current_pending += jobs_to_spawn
+                if self._loop.time() >= self._next_idle_process_spawn_at:
+                    idle_to_spawn = max(idle_target - current_pending, 0)
+                    for _ in range(idle_to_spawn):
+                        self._spawn_process()
 
                 await asyncio.sleep(0.1)
         except asyncio.CancelledError:

tests/test_ipc.py

@@ -274,6 +274,26 @@ async def _wait_for_elements(q: asyncio.Queue, num_elements: int) -> None:
         await q.get()
 
 
+def _create_scheduler_test_pool(*, num_idle_processes: int) -> ipc.proc_pool.ProcPool:
+    return ipc.proc_pool.ProcPool(
+        initialize_process_fnc=_initialize_proc,
+        job_entrypoint_fnc=_job_entrypoint,
+        session_end_fnc=None,
+        simulation_end_fnc=None,
+        num_idle_processes=num_idle_processes,
+        job_executor_type=job.JobExecutorType.THREAD,
+        initialize_timeout=1.0,
+        close_timeout=1.0,
+        session_end_timeout=1.0,
+        inference_executor=None,
+        memory_warn_mb=0,
+        memory_limit_mb=0,
+        http_proxy=None,
+        mp_ctx=mp.get_context("spawn"),
+        loop=asyncio.get_running_loop(),
+    )
+
+
 async def test_proc_pool():
     mp_ctx = mp.get_context("spawn")
     loop = asyncio.get_running_loop()
@@ -404,7 +424,7 @@ def _process_closed(proc: ipc.job_proc_executor.ProcJobExecutor):
     await _wait_for_elements(start_q, num_idle_processes)
     await _wait_for_elements(close_q, num_idle_processes)
 
-    # retry batch should also timeout and be killed
+    # retry batch should also timeout and be killed after the initialization backoff
     await _wait_for_elements(start_q, num_idle_processes)
     await _wait_for_elements(close_q, num_idle_processes)
 
@@ -417,6 +437,122 @@ def _process_closed(proc: ipc.job_proc_executor.ProcJobExecutor):
         assert exitcode != 0, "process should have been killed"
 
 
+async def test_proc_pool_failed_spawn_uses_bounded_round_backoff():
+    pool = _create_scheduler_test_pool(num_idle_processes=1)
+    loop = asyncio.get_running_loop()
+
+    async def _spawn_result(initialized: bool) -> bool:
+        return initialized
+
+    for expected_delay in [1.0, 2.0, 4.0, 8.0, 16.0, 30.0, 30.0]:
+        pool._next_idle_process_spawn_at = loop.time() - 1.0
+        task = asyncio.create_task(_spawn_result(False))
+        pool._spawn_tasks.add(task)
+        await task
+        before_callback = loop.time()
+        pool._on_process_spawn_done(task)
+
+        assert task not in pool._spawn_tasks
+        assert pool._process_init_retry_delay == expected_delay
+        assert pool._next_idle_process_spawn_at - before_callback == pytest.approx(
+            expected_delay, abs=0.05
+        )
+
+    retry_at = pool._next_idle_process_spawn_at
+    task = asyncio.create_task(_spawn_result(True))
+    pool._spawn_tasks.add(task)
+    await task
+    pool._on_process_spawn_done(task)
+
+    assert pool._process_init_retry_delay == 0.0
+    assert pool._next_idle_process_spawn_at == retry_at
+
+    # Multiple failures inside one active retry round must not scale with pool size.
+    pool._next_idle_process_spawn_at = loop.time() - 1.0
+    first_failure = asyncio.create_task(_spawn_result(False))
+    pool._spawn_tasks.add(first_failure)
+    await first_failure
+    pool._on_process_spawn_done(first_failure)
+    first_retry_at = pool._next_idle_process_spawn_at
+
+    same_round_failure = asyncio.create_task(_spawn_result(False))
+    pool._spawn_tasks.add(same_round_failure)
+    await same_round_failure
+    pool._on_process_spawn_done(same_round_failure)
+
+    assert pool._process_init_retry_delay == 1.0
+    assert pool._next_idle_process_spawn_at >= first_retry_at
+
+
+async def test_proc_pool_spawn_exception_sets_backoff():
+    pool = _create_scheduler_test_pool(num_idle_processes=1)
+
+    async def _raise_during_spawn() -> bool:
+        raise RuntimeError("simulated spawn failure")
+
+    task = asyncio.create_task(_raise_during_spawn())
+    pool._spawn_tasks.add(task)
+    await asyncio.wait([task])
+    pool._on_process_spawn_done(task)
+
+    assert task not in pool._spawn_tasks
+    assert pool._process_init_retry_delay == 1.0
+
+
+async def test_proc_pool_backoff_only_delays_idle_replenishment(monkeypatch):
+    pool = _create_scheduler_test_pool(num_idle_processes=4)
+    attempts = 0
+    block_spawn = asyncio.Event()
+
+    async def _blocked_spawn() -> bool:
+        nonlocal attempts
+        attempts += 1
+        await block_spawn.wait()
+        return True
+
+    monkeypatch.setattr(pool, "_proc_spawn_task", _blocked_spawn)
+    pool._next_idle_process_spawn_at = asyncio.get_running_loop().time() + 10.0
+    pool._started = True
+    pool._main_atask = asyncio.create_task(pool._main_task())
+
+    try:
+        await asyncio.sleep(0.2)
+        assert attempts == 0
+
+        pool._jobs_waiting_for_process = 1
+        await _poll_until(lambda: attempts == 1, timeout=1.0)
+        await asyncio.sleep(0.2)
+
+        # The waiting job bypasses the cooldown, but the other three idle slots do not.
+        assert attempts == 1
+    finally:
+        await pool.aclose()
+
+
+async def test_proc_pool_resumes_idle_replenishment_after_backoff(monkeypatch):
+    pool = _create_scheduler_test_pool(num_idle_processes=2)
+    attempts = 0
+    block_spawn = asyncio.Event()
+
+    async def _blocked_spawn() -> bool:
+        nonlocal attempts
+        attempts += 1
+        await block_spawn.wait()
+        return True
+
+    monkeypatch.setattr(pool, "_proc_spawn_task", _blocked_spawn)
+    pool._next_idle_process_spawn_at = asyncio.get_running_loop().time() + 1.0
+    pool._started = True
+    pool._main_atask = asyncio.create_task(pool._main_task())
+
+    try:
+        await asyncio.sleep(0.05)
+        assert attempts == 0
+        await _poll_until(lambda: attempts == 2, timeout=2.0)
+    finally:
+        await pool.aclose()
+
+
 async def test_proc_pool_launch_job_raises_when_all_spawns_fail():
     """When every spawn task fails to initialize, launch_job should raise
     instead of hanging on an empty warmed-process queue. Reproduces #5868.