From 024240e6bb16f62815ecf956db26b0e676297b46 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Sun, 28 Jun 2026 18:29:59 -0700
Subject: [PATCH 01/24] Don't Quit Children Until Replaced

---
 .../src/taskbroker_client/worker/worker.py    | 157 ++++++++-
 .../taskbroker_client/worker/workerchild.py   |  57 ++--
 clients/python/tests/worker/test_worker.py    | 317 ++++++++++++++++--
 3 files changed, 465 insertions(+), 66 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 690b6514..2541f7ca 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -8,10 +8,13 @@
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
 from multiprocessing.context import ForkContext, ForkServerContext, SpawnContext
 from multiprocessing.process import BaseProcess
+from multiprocessing.synchronize import Event
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable, List
+from typing import TYPE_CHECKING, Any, Callable, List, Literal
+from uuid import UUID, uuid4
 
 import grpc
 from grpc_health.v1 import health, health_pb2, health_pb2_grpc
@@ -42,7 +45,7 @@
     parse_rpc_secret_list,
 )
 from taskbroker_client.worker.push_clients import BatchPushTaskbrokerClient, PushTaskbrokerClient
-from taskbroker_client.worker.workerchild import child_process
+from taskbroker_client.worker.workerchild import ChildMessage, child_process
 
 if TYPE_CHECKING:
     ServerInterceptor = grpc.ServerInterceptor[Any, Any]
@@ -122,6 +125,16 @@ class RequeueException(Exception):
     pass
 
 
+ChildLifecycleState = Literal["pending", "ready", "exiting"]
+
+
+@dataclass
+class TrackedChild:
+    process: BaseProcess
+    state: ChildLifecycleState
+    release_event: Event
+
+
 class PushTaskWorker:
     _mp_context: ForkContext | SpawnContext | ForkServerContext
 
@@ -800,9 +813,9 @@ def __init__(
         self._processed_tasks: multiprocessing.Queue[ProcessingResult] = self._mp_context.Queue(
             maxsize=result_queue_maxsize
         )
-        self._children: list[BaseProcess] = []
+        self._children: dict[UUID, TrackedChild] = {}
+        self._children_lock = threading.Lock()
         self._shutdown_event = self._mp_context.Event()
-        self._ready_counter = self._mp_context.Value("i", 0)
         self._result_thread: threading.Thread | None = None
         self._metrics_thread: threading.Thread | None = None
         self._spawn_children_thread: threading.Thread | None = None
@@ -810,7 +823,10 @@ def __init__(
     @property
     def ready_count(self) -> int:
         """Number of children that have finished warming up and are consuming."""
-        return self._ready_counter.value
+        with self._children_lock:
+            return sum(
+                1 for tracked_child in self._children.values() if tracked_child.state == "ready"
+            )
 
     def send_results(self, results: list[ProcessingResult], is_draining: bool = False) -> None:
         """
@@ -917,16 +933,106 @@ def result_thread() -> None:
     def start_spawn_children_thread(self) -> None:
         def spawn_children_thread() -> None:
             logger.debug("taskworker.worker.spawn_children_thread.started")
+
+            # Queue of incoming message from children
+            messages: multiprocessing.Queue[ChildMessage] = self._mp_context.Queue()
+
             while not self._shutdown_event.is_set():
-                self._children = [child for child in self._children if child.is_alive()]
-                if len(self._children) >= self._concurrency:
-                    time.sleep(0.1)
-                    continue
-                for i in range(self._concurrency - len(self._children)):
+                # Read any events that may have come in since the last loop iteration
+                received: list[ChildMessage] = []
+
+                while True:
+                    try:
+                        message = messages.get(block=False)
+                        received.append(message)
+                    except queue.Empty:
+                        break
+
+                with self._children_lock:
+                    for child_id, tracked_child in list(self._children.items()):
+                        if tracked_child.process.is_alive():
+                            continue
+
+                        tracked_child.process.join(timeout=0)
+                        del self._children[child_id]
+                        logger.info(
+                            "taskworker.child_exited",
+                            extra={
+                                "pid": tracked_child.process.pid,
+                                "child_id": str(child_id),
+                                "exitcode": tracked_child.process.exitcode,
+                                "state": tracked_child.state,
+                                "processing_pool": self._processing_pool_name,
+                            },
+                        )
+
+                    for message in received:
+                        message_child = self._children.get(message.child_id)
+                        if message_child is None:
+                            logger.debug(
+                                "taskworker.child_message.unknown_child",
+                                extra={
+                                    "child_id": str(message.child_id),
+                                    "event": message.event,
+                                    "processing_pool": self._processing_pool_name,
+                                },
+                            )
+                            continue
+
+                        if message.event == "ready":
+                            if message_child.state == "pending":
+                                message_child.state = "ready"
+
+                            else:
+                                logger.debug(
+                                    "taskworker.child_message.duplicate_ready",
+                                    extra={
+                                        "child_id": str(message.child_id),
+                                        "state": message_child.state,
+                                        "processing_pool": self._processing_pool_name,
+                                    },
+                                )
+
+                        elif message.event == "exiting":
+                            if message_child.state == "ready":
+                                message_child.state = "exiting"
+                            elif message_child.state != "exiting":
+                                logger.debug(
+                                    "taskworker.child_message.exiting_before_ready",
+                                    extra={
+                                        "child_id": str(message.child_id),
+                                        "state": message_child.state,
+                                        "processing_pool": self._processing_pool_name,
+                                    },
+                                )
+
+                    ready_count = sum(
+                        1
+                        for tracked_child in self._children.values()
+                        if tracked_child.state == "ready"
+                    )
+                    if ready_count >= self._concurrency:
+                        for tracked_child in self._children.values():
+                            if tracked_child.state == "exiting":
+                                tracked_child.release_event.set()
+
+                    non_draining_count = sum(
+                        1
+                        for tracked_child in self._children.values()
+                        if tracked_child.state in {"pending", "ready"}
+                    )
+                    children_to_spawn = max(self._concurrency - non_draining_count, 0)
+
+                # If there aren't enough pending or ready processes, spawn more.
+                for _ in range(children_to_spawn):
+                    child_id = uuid4()
+                    release_event = self._mp_context.Event()
+
                     process = self._mp_context.Process(
-                        name=f"taskworker-child-{i}",
+                        name=f"taskworker-child-{child_id}",
                         target=child_process,
                         args=(
+                            child_id,
                             self._app_module,
                             self._child_tasks,
                             self._processed_tasks,
@@ -935,20 +1041,36 @@ def spawn_children_thread() -> None:
                             self._processing_pool_name,
                             self._process_type,
                             self._skip_awaiting_futures,
-                            self._ready_counter,
+                            messages,
+                            release_event,
                         ),
                     )
+
                     process.start()
-                    self._children.append(process)
+
+                    with self._children_lock:
+                        self._children[child_id] = TrackedChild(
+                            process=process,
+                            state="pending",
+                            release_event=release_event,
+                        )
+
                     logger.info(
                         "taskworker.spawn_child",
-                        extra={"pid": process.pid, "processing_pool": self._processing_pool_name},
+                        extra={
+                            "pid": process.pid,
+                            "child_id": str(child_id),
+                            "processing_pool": self._processing_pool_name,
+                        },
                     )
+
                     self._metrics.incr(
                         "taskworker.worker.spawn_child",
                         tags={"processing_pool": self._processing_pool_name},
                     )
 
+                time.sleep(0.1)
+
         self._spawn_children_thread = threading.Thread(
             name="spawn-children", target=spawn_children_thread, daemon=True
         )
@@ -1005,9 +1127,12 @@ def shutdown(self) -> None:
             self._spawn_children_thread.join()
 
         logger.info("taskworker.worker.shutdown.children")
-        for child in self._children:
+        with self._children_lock:
+            children = [tracked_child.process for tracked_child in self._children.values()]
+
+        for child in children:
             child.terminate()
-        for child in self._children:
+        for child in children:
             child.join(WORKER_CHILD_JOIN_TIMEOUT_SEC)
             if child.is_alive():
                 child.kill()
diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py
index 5395920c..0a4e5682 100644
--- a/clients/python/src/taskbroker_client/worker/workerchild.py
+++ b/clients/python/src/taskbroker_client/worker/workerchild.py
@@ -2,6 +2,7 @@
 
 import contextlib
 import logging
+import multiprocessing
 import queue
 import signal
 import threading
@@ -11,10 +12,8 @@
 from functools import partial
 from multiprocessing.synchronize import Event
 from types import FrameType
-from typing import TYPE_CHECKING, Any
-
-if TYPE_CHECKING:
-    from multiprocessing.sharedctypes import Synchronized
+from typing import Any, Literal
+from uuid import UUID
 
 # XXX: Don't import any modules that will import django here, do those within child_process
 import msgpack
@@ -166,7 +165,14 @@ def _log_task_retry_exhausted(
     )
 
 
+@dataclass(frozen=True)
+class ChildMessage:
+    child_id: UUID
+    event: Literal["ready"] | Literal["exiting"]
+
+
 def child_process(
+    child_id: UUID,
     app_module: str,
     child_tasks: queue.Queue[InflightTaskActivation],
     processed_tasks: queue.Queue[ProcessingResult],
@@ -175,7 +181,8 @@ def child_process(
     processing_pool_name: str,
     process_type: str,
     skip_awaiting_futures: bool,
-    ready_counter: "Synchronized[int] | None" = None,
+    messages: multiprocessing.Queue[ChildMessage],
+    parent_release: Event,
 ) -> None:
     """
     The entrypoint for spawned worker children.
@@ -351,18 +358,30 @@ def check_task_future_completion(
         )
         _future_completion_thread.start()
 
+        waiting_for_parent_release = False
+
         while not shutdown_event.is_set() and not local_shutdown.is_set():
             if max_task_count and processed_task_count >= max_task_count:
-                metrics.incr(
-                    "taskworker.worker.max_task_count_reached",
-                    tags={"count": processed_task_count, "processing_pool": processing_pool_name},
-                )
-                logger.info(
-                    "taskworker.max_task_count_reached", extra={"count": processed_task_count}
-                )
-                # Still set the shutdown signal to trigger shutdown of the future checker thread
-                local_shutdown.set()
-                break
+                if not waiting_for_parent_release:
+                    metrics.incr(
+                        "taskworker.worker.max_task_count_reached",
+                        tags={
+                            "count": processed_task_count,
+                            "processing_pool": processing_pool_name,
+                        },
+                    )
+
+                    logger.info(
+                        "taskworker.max_task_count_reached", extra={"count": processed_task_count}
+                    )
+
+                    # Tell the parent this warmed child can be replaced.
+                    messages.put_nowait(ChildMessage(child_id, "exiting"))
+                    waiting_for_parent_release = True
+
+                if parent_release.is_set():
+                    local_shutdown.set()
+                    break
 
             try:
                 inflight = child_tasks.get(timeout=1.0)
@@ -806,11 +825,9 @@ def _task_execution_complete(
             futures_start_time,
         )
 
-    # Signal that this child has finished warmup and ready to consume tasks. The parent uses this
-    # to gate the gRPC SERVING health signal. Monotonic by design
-    if ready_counter is not None:
-        with ready_counter.get_lock():
-            ready_counter.value += 1
+    # Signal that this child has warmed up and is ready to consume tasks
+    # The parent uses this to gate the gRPC SERVING health signal. Monotonic by design
+    messages.put_nowait(ChildMessage(child_id, "ready"))
 
     # Run the worker loop
     run_worker(
diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py
index bcf6186a..0f68a397 100644
--- a/clients/python/tests/worker/test_worker.py
+++ b/clients/python/tests/worker/test_worker.py
@@ -8,9 +8,11 @@
 from concurrent.futures import Future
 from datetime import datetime
 from multiprocessing import Event, get_context
+from multiprocessing.synchronize import Event as MultiprocessingEvent
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 from unittest import TestCase, mock
+from uuid import uuid4
 
 import grpc
 import msgpack
@@ -45,7 +47,8 @@
     TaskWorkerProcessingPool,
     WorkerServicer,
 )
-from taskbroker_client.worker.workerchild import child_process
+from taskbroker_client.worker.workerchild import ChildMessage
+from taskbroker_client.worker.workerchild import child_process as _child_process
 
 SIMPLE_TASK = InflightTaskActivation(
     host="localhost:50051",
@@ -266,6 +269,36 @@ def _make_processing_result(task_id: str) -> ProcessingResult:
     )
 
 
+def child_process(
+    app_module: str,
+    child_tasks: queue.Queue[InflightTaskActivation],
+    processed_tasks: queue.Queue[ProcessingResult],
+    shutdown_event: MultiprocessingEvent,
+    max_task_count: int | None,
+    processing_pool_name: str,
+    process_type: str,
+    skip_awaiting_futures: bool,
+) -> None:
+    ctx = get_context("fork")
+    messages = ctx.Queue()
+    parent_release = ctx.Event()
+    parent_release.set()
+
+    _child_process(
+        uuid4(),
+        app_module,
+        child_tasks,
+        processed_tasks,
+        shutdown_event,
+        max_task_count,
+        processing_pool_name,
+        process_type,
+        skip_awaiting_futures,
+        messages,
+        parent_release,
+    )
+
+
 class _SendResultCapture:
     def __init__(self) -> None:
         self.send_calls: list[tuple[list[ProcessingResult], bool]] = []
@@ -302,6 +335,87 @@ def _make_result_thread_pool(
     )
 
 
+class _FakeProcess:
+    _next_pid = 1
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.name = kwargs["name"]
+        self.target = kwargs["target"]
+        self.args = kwargs["args"]
+        self.pid = _FakeProcess._next_pid
+        _FakeProcess._next_pid += 1
+        self.exitcode: int | None = None
+        self.started = False
+        self.alive = False
+        self.terminated = False
+        self.killed = False
+        self.join_calls: list[float | None] = []
+
+    def start(self) -> None:
+        self.started = True
+        self.alive = True
+
+    def is_alive(self) -> bool:
+        return self.alive
+
+    def join(self, timeout: float | None = None) -> None:
+        self.join_calls.append(timeout)
+
+    def terminate(self) -> None:
+        self.terminated = True
+        self.alive = False
+        self.exitcode = -signal.SIGTERM
+
+    def kill(self) -> None:
+        self.killed = True
+        self.alive = False
+        self.exitcode = -signal.SIGKILL
+
+
+class _FakeContext:
+    def __init__(self) -> None:
+        self.processes: list[_FakeProcess] = []
+        self.queues: list[queue.Queue[Any]] = []
+
+    def Queue(self, maxsize: int = 0) -> queue.Queue[Any]:
+        created: queue.Queue[Any] = queue.Queue(maxsize=maxsize)
+        self.queues.append(created)
+        return created
+
+    def Event(self) -> threading.Event:
+        return threading.Event()
+
+    def Process(self, *args: Any, **kwargs: Any) -> _FakeProcess:
+        process = _FakeProcess(*args, **kwargs)
+        self.processes.append(process)
+        return process
+
+
+def _make_fake_context_pool(
+    fake_context: _FakeContext,
+    *,
+    concurrency: int = 1,
+) -> TaskWorkerProcessingPool:
+    return TaskWorkerProcessingPool(
+        app_module="examples.app:app",
+        send_result_fn=lambda x, y: None,
+        mp_context=fake_context,  # type: ignore[arg-type]
+        max_child_task_count=1,
+        concurrency=concurrency,
+        processing_pool_name="test",
+        process_type="fork",
+    )
+
+
+def _wait_for(condition: Callable[[], bool], timeout: float = 5) -> None:
+    start = time.time()
+    while time.time() - start < timeout:
+        if condition():
+            return
+        time.sleep(0.01)
+    raise AssertionError("Timed out waiting for condition")
+
+
 class TestTaskWorker(TestCase):
     def test_fetch_task(self) -> None:
         taskworker = TaskWorker(
@@ -359,13 +473,11 @@ def test_run_once_no_next_task(self) -> None:
 
             taskworker.shutdown()
             assert mock_client.get_task.called
-            assert mock_client.update_task.call_count == 1
-            assert mock_client.update_task.call_args.args[0].host == "localhost:50051"
-            assert mock_client.update_task.call_args.args[0].task_id == SIMPLE_TASK.activation.id
-            assert (
-                mock_client.update_task.call_args.args[0].status == TASK_ACTIVATION_STATUS_COMPLETE
-            )
-            assert mock_client.update_task.call_args.args[1] is None
+            assert mock_client.update_task.call_count >= 1
+            first_update = mock_client.update_task.call_args_list[0]
+            assert first_update.args[0].host == "localhost:50051"
+            assert first_update.args[0].task_id == SIMPLE_TASK.activation.id
+            assert first_update.args[0].status == TASK_ACTIVATION_STATUS_COMPLETE
 
     def test_run_once_with_next_task(self) -> None:
         # Cover the scenario where update_task returns the next task which should
@@ -401,13 +513,11 @@ def update_task_response(*args: Any, **kwargs: Any) -> InflightTaskActivation |
 
             taskworker.shutdown()
             assert mock_client.get_task.called
-            assert mock_client.update_task.call_count == 2
-            assert mock_client.update_task.call_args.args[0].host == "localhost:50051"
-            assert mock_client.update_task.call_args.args[0].task_id == SIMPLE_TASK.activation.id
-            assert (
-                mock_client.update_task.call_args.args[0].status == TASK_ACTIVATION_STATUS_COMPLETE
-            )
-            assert mock_client.update_task.call_args.args[1] is None
+            assert mock_client.update_task.call_count >= 2
+            for update_call in mock_client.update_task.call_args_list[:2]:
+                assert update_call.args[0].host == "localhost:50051"
+                assert update_call.args[0].task_id == SIMPLE_TASK.activation.id
+                assert update_call.args[0].status == TASK_ACTIVATION_STATUS_COMPLETE
 
     def test_run_once_with_update_failure(self) -> None:
         # Cover the scenario where update_task fails a few times in a row
@@ -665,9 +775,14 @@ def _make_push_worker(**kwargs: Any) -> PushTaskWorker:
 
 def test_await_children_warm_returns_when_ready() -> None:
     taskworker = _make_push_worker(concurrency=4, warmup_timeout=5)
-    taskworker.worker_pool._ready_counter.value = 4
 
-    with mock.patch.object(taskworker, "_metrics") as mock_metrics:
+    with (
+        mock.patch.object(taskworker, "_metrics") as mock_metrics,
+        mock.patch.object(
+            TaskWorkerProcessingPool, "ready_count", new_callable=mock.PropertyMock
+        ) as ready_count,
+    ):
+        ready_count.return_value = 4
         start = time.time()
         taskworker._await_children_warm()
         elapsed = time.time() - start
@@ -687,8 +802,6 @@ def test_await_children_warm_returns_when_ready() -> None:
 
 def test_await_children_warm_times_out() -> None:
     taskworker = _make_push_worker(concurrency=4, warmup_timeout=0.1)
-    # Never becomes ready.
-    taskworker.worker_pool._ready_counter.value = 0
 
     with mock.patch.object(taskworker, "_metrics") as mock_metrics:
         start = time.time()
@@ -703,16 +816,23 @@ def test_await_children_warm_times_out() -> None:
 
 def test_await_children_warm_unblocks_when_children_warm() -> None:
     taskworker = _make_push_worker(concurrency=2, warmup_timeout=5)
-    taskworker.worker_pool._ready_counter.value = 0
+    ready_child_count = 0
 
     def warm_up() -> None:
+        nonlocal ready_child_count
         time.sleep(0.2)
-        taskworker.worker_pool._ready_counter.value = 2
+        ready_child_count = 2
 
     warmer = threading.Thread(target=warm_up)
     warmer.start()
     try:
-        with mock.patch.object(taskworker, "_metrics") as mock_metrics:
+        with (
+            mock.patch.object(taskworker, "_metrics") as mock_metrics,
+            mock.patch.object(
+                TaskWorkerProcessingPool, "ready_count", new_callable=mock.PropertyMock
+            ) as ready_count,
+        ):
+            ready_count.side_effect = lambda: ready_child_count
             start = time.time()
             taskworker._await_children_warm()
             elapsed = time.time() - start
@@ -733,7 +853,6 @@ def test_start_does_not_serve_when_shutdown_during_warmup() -> None:
 
     taskworker = _make_push_worker(concurrency=2, warmup_timeout=5)
     # Children never warm, and shutdown is requested before start() runs.
-    taskworker.worker_pool._ready_counter.value = 0
     taskworker._grpc_sync_event.set()
 
     fake_health = mock.MagicMock()
@@ -768,6 +887,82 @@ def test_start_does_not_serve_when_shutdown_during_warmup() -> None:
     fake_server.wait_for_termination.assert_not_called()
 
 
+def test_spawn_children_counts_pending_children_toward_concurrency() -> None:
+    fake_context = _FakeContext()
+    pool = _make_fake_context_pool(fake_context, concurrency=2)
+
+    pool.start_spawn_children_thread()
+    try:
+        _wait_for(lambda: len(fake_context.processes) == 2)
+        time.sleep(0.25)
+
+        assert len(fake_context.processes) == 2
+        assert pool.ready_count == 0
+    finally:
+        pool.shutdown()
+
+
+def test_spawn_children_releases_draining_child_after_replacement_ready() -> None:
+    fake_context = _FakeContext()
+    pool = _make_fake_context_pool(fake_context, concurrency=1)
+
+    pool.start_spawn_children_thread()
+    try:
+        _wait_for(lambda: len(fake_context.processes) == 1)
+        messages = fake_context.queues[-1]
+        first_process = fake_context.processes[0]
+        first_child_id = first_process.args[0]
+        first_release = first_process.args[-1]
+
+        messages.put(ChildMessage(first_child_id, "ready"))
+        _wait_for(lambda: pool.ready_count == 1)
+
+        messages.put(ChildMessage(first_child_id, "exiting"))
+        _wait_for(lambda: len(fake_context.processes) == 2)
+
+        assert not first_release.is_set()
+
+        second_process = fake_context.processes[1]
+        second_child_id = second_process.args[0]
+        messages.put(ChildMessage(second_child_id, "ready"))
+
+        _wait_for(first_release.is_set)
+    finally:
+        pool.shutdown()
+
+
+def test_spawn_children_replaces_pending_child_that_dies_before_ready() -> None:
+    fake_context = _FakeContext()
+    pool = _make_fake_context_pool(fake_context, concurrency=1)
+
+    pool.start_spawn_children_thread()
+    try:
+        _wait_for(lambda: len(fake_context.processes) == 1)
+        first_process = fake_context.processes[0]
+        first_process.alive = False
+        first_process.exitcode = 1
+
+        _wait_for(lambda: len(fake_context.processes) == 2)
+
+        assert first_process.join_calls
+        assert fake_context.processes[1].started
+    finally:
+        pool.shutdown()
+
+
+def test_shutdown_terminates_all_tracked_children() -> None:
+    fake_context = _FakeContext()
+    pool = _make_fake_context_pool(fake_context, concurrency=2)
+
+    pool.start_spawn_children_thread()
+    _wait_for(lambda: len(fake_context.processes) == 2)
+
+    pool.shutdown()
+
+    assert all(process.terminated for process in fake_context.processes)
+    assert all(process.join_calls for process in fake_context.processes)
+
+
 class TestWorkerServicer(TestCase):
     def test_push_task_success(self) -> None:
         taskworker = PushTaskWorker(
@@ -893,15 +1088,19 @@ def test_child_process_complete(mock_capture_checkin: mock.MagicMock) -> None:
     assert mock_capture_checkin.call_count == 0
 
 
-def test_child_process_increments_ready_counter() -> None:
+def test_child_process_emits_ready_message() -> None:
     todo: queue.Queue[InflightTaskActivation] = queue.Queue()
     processed: queue.Queue[ProcessingResult] = queue.Queue()
     shutdown = Event()
     ctx = get_context("fork")
-    ready_counter = ctx.Value("i", 0)
+    child_id = uuid4()
+    messages = ctx.Queue()
+    parent_release = ctx.Event()
+    parent_release.set()
 
     todo.put(SIMPLE_TASK)
-    child_process(
+    _child_process(
+        child_id,
         "examples.app:app",
         todo,
         processed,
@@ -910,11 +1109,69 @@ def test_child_process_increments_ready_counter() -> None:
         processing_pool_name="test",
         process_type="fork",
         skip_awaiting_futures=False,
-        ready_counter=ready_counter,
+        messages=messages,
+        parent_release=parent_release,
+    )
+
+    # The child signals readiness once warmup is done, before consuming.
+    message = messages.get(timeout=1)
+    assert message == ChildMessage(child_id, "ready")
+
+
+@mock.patch("taskbroker_client.worker.workerchild.capture_checkin")
+def test_child_process_emits_exiting_once_and_continues_until_release(
+    mock_capture_checkin: mock.MagicMock,
+) -> None:
+    shutdown = Event()
+    ctx = get_context("fork")
+    child_id = uuid4()
+    todo = ctx.Queue()
+    processed = ctx.Queue()
+    messages = ctx.Queue()
+    parent_release = ctx.Event()
+
+    todo.put(SIMPLE_TASK)
+    process = ctx.Process(
+        target=_child_process,
+        args=(
+            child_id,
+            "examples.app:app",
+            todo,
+            processed,
+            shutdown,
+            1,
+            "test",
+            "fork",
+            False,
+            messages,
+            parent_release,
+        ),
     )
+    process.start()
+    try:
+        ready_message = messages.get(timeout=5)
+        exiting_message = messages.get(timeout=5)
+
+        assert ready_message == ChildMessage(child_id, "ready")
+        assert exiting_message == ChildMessage(child_id, "exiting")
+        assert processed.get(timeout=5).task_id == SIMPLE_TASK.activation.id
 
-    # The child increments the counter once warmup is done, before consuming.
-    assert ready_counter.value == 1
+        todo.put(SIMPLE_TASK)
+        assert processed.get(timeout=5).task_id == SIMPLE_TASK.activation.id
+
+        time.sleep(0.2)
+        assert process.is_alive()
+        assert messages.empty()
+
+        parent_release.set()
+        process.join(timeout=5)
+        assert not process.is_alive()
+    finally:
+        if process.is_alive():
+            process.terminate()
+            process.join(timeout=5)
+
+    assert mock_capture_checkin.call_count == 0
 
 
 def test_child_process_remove_start_time_kwargs() -> None:

From 78f0fce74554f03c8c050e5e1cd67ff96a153738 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Sun, 28 Jun 2026 19:23:34 -0700
Subject: [PATCH 02/24] Add Fake Sleep to Child Process

---
 clients/python/src/taskbroker_client/worker/workerchild.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py
index 0a4e5682..d63a1032 100644
--- a/clients/python/src/taskbroker_client/worker/workerchild.py
+++ b/clients/python/src/taskbroker_client/worker/workerchild.py
@@ -4,6 +4,7 @@
 import logging
 import multiprocessing
 import queue
+import random
 import signal
 import threading
 import time
@@ -375,7 +376,7 @@ def check_task_future_completion(
                         "taskworker.max_task_count_reached", extra={"count": processed_task_count}
                     )
 
-                    # Tell the parent this warmed child can be replaced.
+                    # Tell the parent this child can be replaced.
                     messages.put_nowait(ChildMessage(child_id, "exiting"))
                     waiting_for_parent_release = True
 
@@ -825,6 +826,9 @@ def _task_execution_complete(
             futures_start_time,
         )
 
+    seconds = random.randint(30, 50)
+    time.sleep(seconds)
+
     # Signal that this child has warmed up and is ready to consume tasks
     # The parent uses this to gate the gRPC SERVING health signal. Monotonic by design
     messages.put_nowait(ChildMessage(child_id, "ready"))

From 8904e3cebc26ab0be291c3df309a3898fe1f4384 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Sun, 28 Jun 2026 20:34:41 -0700
Subject: [PATCH 03/24] Remove Fake Sleep

---
 clients/python/src/taskbroker_client/worker/workerchild.py | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py
index d63a1032..bfa15df8 100644
--- a/clients/python/src/taskbroker_client/worker/workerchild.py
+++ b/clients/python/src/taskbroker_client/worker/workerchild.py
@@ -4,7 +4,6 @@
 import logging
 import multiprocessing
 import queue
-import random
 import signal
 import threading
 import time
@@ -826,9 +825,6 @@ def _task_execution_complete(
             futures_start_time,
         )
 
-    seconds = random.randint(30, 50)
-    time.sleep(seconds)
-
     # Signal that this child has warmed up and is ready to consume tasks
     # The parent uses this to gate the gRPC SERVING health signal. Monotonic by design
     messages.put_nowait(ChildMessage(child_id, "ready"))

From 4d8a1008860ac9f2caefdcdeb3576edc74df62ce Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Sun, 28 Jun 2026 21:25:11 -0700
Subject: [PATCH 04/24] Handle Process Start Failure

---
 .../src/taskbroker_client/worker/worker.py    | 27 +++++++++++++++++--
 1 file changed, 25 insertions(+), 2 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 2541f7ca..537cadb0 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1046,7 +1046,27 @@ def spawn_children_thread() -> None:
                         ),
                     )
 
-                    process.start()
+                    try:
+                        process.start()
+                    except Exception as e:
+                        logger.exception(
+                            "taskworker.spawn_child.failed",
+                            extra={
+                                "child_id": str(child_id),
+                                "error": e,
+                                "processing_pool": self._processing_pool_name,
+                            },
+                        )
+
+                        self._metrics.incr(
+                            "taskworker.worker.spawn_child",
+                            tags={
+                                "processing_pool": self._processing_pool_name,
+                                "result": "failure",
+                            },
+                        )
+
+                        continue
 
                     with self._children_lock:
                         self._children[child_id] = TrackedChild(
@@ -1066,7 +1086,10 @@ def spawn_children_thread() -> None:
 
                     self._metrics.incr(
                         "taskworker.worker.spawn_child",
-                        tags={"processing_pool": self._processing_pool_name},
+                        tags={
+                            "processing_pool": self._processing_pool_name,
+                            "result": "success",
+                        },
                     )
 
                 time.sleep(0.1)

From 01b3550659e25a8fb6c54a59535b5746742c0195 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 06:36:20 -0700
Subject: [PATCH 05/24] Formatting

---
 .../src/taskbroker_client/worker/worker.py     | 18 ++++++++----------
 1 file changed, 8 insertions(+), 10 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 537cadb0..4410ee53 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -13,7 +13,7 @@
 from multiprocessing.process import BaseProcess
 from multiprocessing.synchronize import Event
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable, List, Literal
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Literal
 from uuid import UUID, uuid4
 
 import grpc
@@ -125,14 +125,14 @@ class RequeueException(Exception):
     pass
 
 
-ChildLifecycleState = Literal["pending", "ready", "exiting"]
+ChildState = Literal["pending", "ready", "exiting"]
 
 
 @dataclass
 class TrackedChild:
     process: BaseProcess
-    state: ChildLifecycleState
-    release_event: Event
+    state: ChildState
+    release: Event
 
 
 class PushTaskWorker:
@@ -813,7 +813,7 @@ def __init__(
         self._processed_tasks: multiprocessing.Queue[ProcessingResult] = self._mp_context.Queue(
             maxsize=result_queue_maxsize
         )
-        self._children: dict[UUID, TrackedChild] = {}
+        self._children: Dict[UUID, TrackedChild] = {}
         self._children_lock = threading.Lock()
         self._shutdown_event = self._mp_context.Event()
         self._result_thread: threading.Thread | None = None
@@ -824,9 +824,7 @@ def __init__(
     def ready_count(self) -> int:
         """Number of children that have finished warming up and are consuming."""
         with self._children_lock:
-            return sum(
-                1 for tracked_child in self._children.values() if tracked_child.state == "ready"
-            )
+            return sum(1 for c in self._children.values() if c.state == "ready")
 
     def send_results(self, results: list[ProcessingResult], is_draining: bool = False) -> None:
         """
@@ -1014,7 +1012,7 @@ def spawn_children_thread() -> None:
                     if ready_count >= self._concurrency:
                         for tracked_child in self._children.values():
                             if tracked_child.state == "exiting":
-                                tracked_child.release_event.set()
+                                tracked_child.release.set()
 
                     non_draining_count = sum(
                         1
@@ -1072,7 +1070,7 @@ def spawn_children_thread() -> None:
                         self._children[child_id] = TrackedChild(
                             process=process,
                             state="pending",
-                            release_event=release_event,
+                            release=release_event,
                         )
 
                     logger.info(

From 70eddc661af3d1a5dd9873c3324dbd4865d3f527 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 06:51:57 -0700
Subject: [PATCH 06/24] Readability

---
 .../src/taskbroker_client/worker/worker.py    | 108 +++++++++---------
 1 file changed, 53 insertions(+), 55 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 4410ee53..5dc2e1e4 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -937,7 +937,7 @@ def spawn_children_thread() -> None:
 
             while not self._shutdown_event.is_set():
                 # Read any events that may have come in since the last loop iteration
-                received: list[ChildMessage] = []
+                received: List[ChildMessage] = []
 
                 while True:
                     try:
@@ -947,84 +947,81 @@ def spawn_children_thread() -> None:
                         break
 
                 with self._children_lock:
-                    for child_id, tracked_child in list(self._children.items()):
-                        if tracked_child.process.is_alive():
+                    children = list(self._children.items())
+
+                    for cid, c in children:
+                        if c.process.is_alive():
                             continue
 
-                        tracked_child.process.join(timeout=0)
-                        del self._children[child_id]
+                        c.process.join(timeout=0)
+                        self._children.pop(cid)
+
                         logger.info(
-                            "taskworker.child_exited",
+                            "taskworker.child.exited",
                             extra={
-                                "pid": tracked_child.process.pid,
-                                "child_id": str(child_id),
-                                "exitcode": tracked_child.process.exitcode,
-                                "state": tracked_child.state,
+                                "pid": c.process.pid,
+                                "cid": str(cid),
+                                "exitcode": c.process.exitcode,
+                                "state": c.state,
                                 "processing_pool": self._processing_pool_name,
                             },
                         )
 
                     for message in received:
-                        message_child = self._children.get(message.child_id)
-                        if message_child is None:
-                            logger.debug(
+                        child = self._children.get(message.child_id)
+
+                        if child is None:
+                            logger.warning(
                                 "taskworker.child_message.unknown_child",
                                 extra={
-                                    "child_id": str(message.child_id),
+                                    "cid": str(message.child_id),
                                     "event": message.event,
                                     "processing_pool": self._processing_pool_name,
                                 },
                             )
+
                             continue
 
                         if message.event == "ready":
-                            if message_child.state == "pending":
-                                message_child.state = "ready"
-
-                            else:
-                                logger.debug(
-                                    "taskworker.child_message.duplicate_ready",
+                            if child.state != "pending":
+                                logger.warning(
+                                    "taskworker.child.ready_not_pending",
                                     extra={
-                                        "child_id": str(message.child_id),
-                                        "state": message_child.state,
+                                        "cid": str(message.child_id),
+                                        "state": child.state,
                                         "processing_pool": self._processing_pool_name,
                                     },
                                 )
 
+                            child.state = "ready"
+
                         elif message.event == "exiting":
-                            if message_child.state == "ready":
-                                message_child.state = "exiting"
-                            elif message_child.state != "exiting":
-                                logger.debug(
-                                    "taskworker.child_message.exiting_before_ready",
+                            if child.state != "ready":
+                                logger.warning(
+                                    "taskworker.child.exiting_not_ready",
                                     extra={
-                                        "child_id": str(message.child_id),
-                                        "state": message_child.state,
+                                        "cid": str(message.child_id),
+                                        "state": child.state,
                                         "processing_pool": self._processing_pool_name,
                                     },
                                 )
 
-                    ready_count = sum(
-                        1
-                        for tracked_child in self._children.values()
-                        if tracked_child.state == "ready"
-                    )
-                    if ready_count >= self._concurrency:
-                        for tracked_child in self._children.values():
-                            if tracked_child.state == "exiting":
-                                tracked_child.release.set()
-
-                    non_draining_count = sum(
-                        1
-                        for tracked_child in self._children.values()
-                        if tracked_child.state in {"pending", "ready"}
-                    )
-                    children_to_spawn = max(self._concurrency - non_draining_count, 0)
+                            child.state = "exiting"
+
+                    ready = sum(1 for c in self._children.values() if c.state == "ready")
 
-                # If there aren't enough pending or ready processes, spawn more.
+                    if ready >= self._concurrency:
+                        for child in self._children.values():
+                            if child.state == "exiting":
+                                child.release.set()
+
+                    not_exiting = sum(1 for c in self._children.values() if c.state != "exiting")
+                    children_to_spawn = max(self._concurrency - not_exiting, 0)
+
+                # If there aren't enough pending or ready processes, spawn more
                 for _ in range(children_to_spawn):
                     child_id = uuid4()
-                    release_event = self._mp_context.Event()
+                    release = self._mp_context.Event()
 
                     process = self._mp_context.Process(
                         name=f"taskworker-child-{child_id}",
@@ -1040,7 +1037,7 @@ def spawn_children_thread() -> None:
                             self._process_type,
                             self._skip_awaiting_futures,
                             messages,
-                            release_event,
+                            release,
                         ),
                     )
 
@@ -1048,16 +1045,16 @@ def spawn_children_thread() -> None:
                         process.start()
                     except Exception as e:
                         logger.exception(
-                            "taskworker.spawn_child.failed",
+                            "taskworker.child.spawn.failed",
                             extra={
-                                "child_id": str(child_id),
+                                "cid": str(child_id),
                                 "error": e,
                                 "processing_pool": self._processing_pool_name,
                             },
                         )
 
                         self._metrics.incr(
-                            "taskworker.worker.spawn_child",
+                            "taskworker.worker.child.spawn",
                             tags={
                                 "processing_pool": self._processing_pool_name,
                                 "result": "failure",
@@ -1070,20 +1067,20 @@ def spawn_children_thread() -> None:
                         self._children[child_id] = TrackedChild(
                             process=process,
                             state="pending",
-                            release=release_event,
+                            release=release,
                         )
 
                     logger.info(
-                        "taskworker.spawn_child",
+                        "taskworker.child.spawn",
                         extra={
                             "pid": process.pid,
-                            "child_id": str(child_id),
+                            "cid": str(child_id),
                             "processing_pool": self._processing_pool_name,
                         },
                     )
 
                     self._metrics.incr(
-                        "taskworker.worker.spawn_child",
+                        "taskworker.worker.child.spawn",
                         tags={
                             "processing_pool": self._processing_pool_name,
                             "result": "success",
@@ -1095,6 +1092,7 @@ def spawn_children_thread() -> None:
         self._spawn_children_thread = threading.Thread(
             name="spawn-children", target=spawn_children_thread, daemon=True
         )
+
         self._spawn_children_thread.start()
 
     def push_task(self, inflight: InflightTaskActivation, timeout: float | None = None) -> bool:

From 60f805a45f6c703e128f7ebe6268d11367205e75 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 07:35:01 -0700
Subject: [PATCH 07/24] Try State Management Again

---
 .../src/taskbroker_client/worker/worker.py    | 22 ++++++++++++++-----
 1 file changed, 17 insertions(+), 5 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 5dc2e1e4..eabcf3c2 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1009,17 +1009,29 @@ def spawn_children_thread() -> None:
                             child.state = "exiting"
 
                     ready = sum(1 for c in self._children.values() if c.state == "ready")
+                    exiting = sum(1 for c in self._children.values() if c.state == "exiting")
+
+                    desired = self._concurrency
+
+                    if ready + exiting > desired:
+                        # We have too many active children
+                        releasable = ready + exiting - desired
+                        released = 0
 
-                    if ready >= self._concurrency:
                         for child in self._children.values():
+                            if released == releasable:
+                                break
+
                             if child.state == "exiting":
                                 child.release.set()
+                                released += 1
 
-                    not_exiting = sum(1 for c in self._children.values() if c.state != "exiting")
-                    children_to_spawn = max(self._concurrency - not_exiting, 0)
+                        needed = 0
+                    else:
+                        # We may not have enough active children
+                        needed = desired - ready + exiting
 
-                # If there aren't enough pending or ready processes, spawn more
-                for _ in range(children_to_spawn):
+                for _ in range(needed):
                     child_id = uuid4()
                     release = self._mp_context.Event()
 

From 12b2d1949ce6ab9324c1e7e9f34ccd7fd9235dcf Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 08:34:19 -0700
Subject: [PATCH 08/24] Fix Child Replacement Logic

---
 clients/python/src/taskbroker_client/worker/worker.py | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index eabcf3c2..60d7ee89 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1008,14 +1008,16 @@ def spawn_children_thread() -> None:
 
                             child.state = "exiting"
 
+                    pending = sum(1 for c in self._children.values() if c.state == "pending")
                     ready = sum(1 for c in self._children.values() if c.state == "ready")
                     exiting = sum(1 for c in self._children.values() if c.state == "exiting")
 
                     desired = self._concurrency
+                    active = ready + exiting
 
-                    if ready + exiting > desired:
+                    if active > desired:
                         # We have too many active children
-                        releasable = ready + exiting - desired
+                        releasable = active - desired
                         released = 0
 
                         for child in self._children.values():
@@ -1029,7 +1031,10 @@ def spawn_children_thread() -> None:
                         needed = 0
                     else:
                         # We may not have enough active children
-                        needed = desired - ready + exiting
+                        missing = desired - active
+
+                        # If we have N pending children and M needed, then we must spawn another M - N
+                        needed = max(missing - pending, 0)
 
                 for _ in range(needed):
                     child_id = uuid4()

From dadff179fbd55d8f35b4ec1092cdf81b75101ced Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 08:40:48 -0700
Subject: [PATCH 09/24] Newly Exiting Children Should Spawn Pending Children

---
 clients/python/src/taskbroker_client/worker/worker.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 60d7ee89..745127f3 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -938,6 +938,7 @@ def spawn_children_thread() -> None:
             while not self._shutdown_event.is_set():
                 # Read any events that may have come in since the last loop iteration
                 received: List[ChildMessage] = []
+                needed = 0
 
                 while True:
                     try:
@@ -1007,6 +1008,7 @@ def spawn_children_thread() -> None:
                                 )
 
                             child.state = "exiting"
+                            needed += 1
 
                     pending = sum(1 for c in self._children.values() if c.state == "pending")
                     ready = sum(1 for c in self._children.values() if c.state == "ready")
@@ -1027,14 +1029,12 @@ def spawn_children_thread() -> None:
                             if child.state == "exiting":
                                 child.release.set()
                                 released += 1
-
-                        needed = 0
                     else:
                         # We may not have enough active children
                         missing = desired - active
 
                         # If we have N pending children and M needed, then we must spawn another M - N
-                        needed = max(missing - pending, 0)
+                        needed += max(missing - pending, 0)
 
                 for _ in range(needed):
                     child_id = uuid4()

From f82d5caf91364004ffb72d7e31c13a92a69afac2 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 08:46:30 -0700
Subject: [PATCH 10/24] Change Exiting Children Comment

---
 clients/python/src/taskbroker_client/worker/worker.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 745127f3..88b66461 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1018,7 +1018,7 @@ def spawn_children_thread() -> None:
                     active = ready + exiting
 
                     if active > desired:
-                        # We have too many active children
+                        # We have too many exiting children, we can release some of them
                         releasable = active - desired
                         released = 0
 

From f7a356e1382ff11eb8d74847423939aff6dd61b9 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 08:55:54 -0700
Subject: [PATCH 11/24] Fix Needed Count Again

---
 .../python/src/taskbroker_client/worker/worker.py  | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 88b66461..009c118b 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -938,7 +938,6 @@ def spawn_children_thread() -> None:
             while not self._shutdown_event.is_set():
                 # Read any events that may have come in since the last loop iteration
                 received: List[ChildMessage] = []
-                needed = 0
 
                 while True:
                     try:
@@ -1008,12 +1007,14 @@ def spawn_children_thread() -> None:
                                 )
 
                             child.state = "exiting"
-                            needed += 1
 
                     pending = sum(1 for c in self._children.values() if c.state == "pending")
                     ready = sum(1 for c in self._children.values() if c.state == "ready")
                     exiting = sum(1 for c in self._children.values() if c.state == "exiting")
 
+                    # We initially assume all exiting children are new and require replacement
+                    needed = exiting
+
                     desired = self._concurrency
                     active = ready + exiting
 
@@ -1029,12 +1030,15 @@ def spawn_children_thread() -> None:
                             if child.state == "exiting":
                                 child.release.set()
                                 released += 1
+
+                                # No replacement needed
+                                needed -= 1
                     else:
                         # We may not have enough active children
-                        missing = desired - active
+                        needed += desired - active
 
-                        # If we have N pending children and M needed, then we must spawn another M - N
-                        needed += max(missing - pending, 0)
+                    # If N are needed and M are already pending, we only need to spawn N - M new children
+                    needed = max(needed - pending, 0)
 
                 for _ in range(needed):
                     child_id = uuid4()

From 82dae9a05dac50991d98630d460fe2748cd898c5 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 09:03:33 -0700
Subject: [PATCH 12/24] Move Counting Logic into Separate Method

---
 .../src/taskbroker_client/worker/worker.py    | 68 ++++++++++---------
 1 file changed, 37 insertions(+), 31 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 009c118b..e89b8453 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -826,6 +826,39 @@ def ready_count(self) -> int:
         with self._children_lock:
             return sum(1 for c in self._children.values() if c.state == "ready")
 
+    def _plan_child_pool_changes(self) -> tuple[int, list[TrackedChild]]:
+        """
+        Determine how to move the child pool toward the desired concurrency.
+
+        The caller must hold `_children_lock`.
+
+        Exiting children are still consuming tasks until released, so they count
+        as serving capacity. Pending children are not serving yet, but they count
+        as replacement capacity that has already been requested.
+        """
+        pending_children = [c for c in self._children.values() if c.state == "pending"]
+        ready_children = [c for c in self._children.values() if c.state == "ready"]
+        exiting_children = [c for c in self._children.values() if c.state == "exiting"]
+
+        desired_serving_capacity = self._concurrency
+        current_serving_capacity = len(ready_children) + len(exiting_children)
+
+        excess_serving_capacity = max(current_serving_capacity - desired_serving_capacity, 0)
+        children_to_release = exiting_children[:excess_serving_capacity]
+
+        kept_exiting_count = len(exiting_children) - len(children_to_release)
+
+        # Keep enough pending children to both reach desired concurrency and replace
+        # exiting children that are still serving during their grace period
+        target_pending_count = max(
+            desired_serving_capacity - len(ready_children),
+            kept_exiting_count,
+        )
+
+        children_to_spawn = max(target_pending_count - len(pending_children), 0)
+
+        return children_to_spawn, children_to_release
+
     def send_results(self, results: list[ProcessingResult], is_draining: bool = False) -> None:
         """
         Call the passed in function. If is_draining is True, the function should not fetch a new task.
@@ -1008,39 +1041,12 @@ def spawn_children_thread() -> None:
 
                             child.state = "exiting"
 
-                    pending = sum(1 for c in self._children.values() if c.state == "pending")
-                    ready = sum(1 for c in self._children.values() if c.state == "ready")
-                    exiting = sum(1 for c in self._children.values() if c.state == "exiting")
-
-                    # We initially assume all exiting children are new and require replacement
-                    needed = exiting
-
-                    desired = self._concurrency
-                    active = ready + exiting
-
-                    if active > desired:
-                        # We have too many exiting children, we can release some of them
-                        releasable = active - desired
-                        released = 0
-
-                        for child in self._children.values():
-                            if released == releasable:
-                                break
-
-                            if child.state == "exiting":
-                                child.release.set()
-                                released += 1
-
-                                # No replacement needed
-                                needed -= 1
-                    else:
-                        # We may not have enough active children
-                        needed += desired - active
+                    children_to_spawn, children_to_release = self._plan_child_pool_changes()
 
-                    # If N are needed and M are already pending, we only need to spawn N - M new children
-                    needed = max(needed - pending, 0)
+                    for child in children_to_release:
+                        child.release.set()
 
-                for _ in range(needed):
+                for _ in range(children_to_spawn):
                     child_id = uuid4()
                     release = self._mp_context.Event()
 

From 62ad0820fe5ea9fa29db17e96646ca44060382ff Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 09:07:03 -0700
Subject: [PATCH 13/24] Minor Edits

---
 clients/python/src/taskbroker_client/worker/worker.py      | 5 ++---
 clients/python/src/taskbroker_client/worker/workerchild.py | 3 +--
 2 files changed, 3 insertions(+), 5 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index e89b8453..1c06f0ee 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1098,7 +1098,7 @@ def spawn_children_thread() -> None:
                         )
 
                     logger.info(
-                        "taskworker.child.spawn",
+                        "taskworker.spawn_child",
                         extra={
                             "pid": process.pid,
                             "cid": str(child_id),
@@ -1107,7 +1107,7 @@ def spawn_children_thread() -> None:
                     )
 
                     self._metrics.incr(
-                        "taskworker.worker.child.spawn",
+                        "taskworker.worker.spawn_child",
                         tags={
                             "processing_pool": self._processing_pool_name,
                             "result": "success",
@@ -1119,7 +1119,6 @@ def spawn_children_thread() -> None:
         self._spawn_children_thread = threading.Thread(
             name="spawn-children", target=spawn_children_thread, daemon=True
         )
-
         self._spawn_children_thread.start()
 
     def push_task(self, inflight: InflightTaskActivation, timeout: float | None = None) -> bool:
diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py
index bfa15df8..cc4fffb1 100644
--- a/clients/python/src/taskbroker_client/worker/workerchild.py
+++ b/clients/python/src/taskbroker_client/worker/workerchild.py
@@ -825,8 +825,7 @@ def _task_execution_complete(
             futures_start_time,
         )
 
-    # Signal that this child has warmed up and is ready to consume tasks
-    # The parent uses this to gate the gRPC SERVING health signal. Monotonic by design
+    # Tell the parent that this child has warmed up and is ready to consume tasks
     messages.put_nowait(ChildMessage(child_id, "ready"))
 
     # Run the worker loop

From cf8ae27daf716d37a762b01817b9c2723bddfce9 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 13:53:17 -0700
Subject: [PATCH 14/24] Simplify Concurrency Management

---
 .../src/taskbroker_client/worker/worker.py    | 109 ++++++------------
 .../taskbroker_client/worker/workerchild.py   |   4 +-
 clients/python/tests/worker/test_worker.py    |   8 +-
 3 files changed, 43 insertions(+), 78 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 1c06f0ee..1aa87415 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -7,13 +7,14 @@
 import signal
 import threading
 import time
+from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from multiprocessing.context import ForkContext, ForkServerContext, SpawnContext
 from multiprocessing.process import BaseProcess
 from multiprocessing.synchronize import Event
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Literal
+from typing import TYPE_CHECKING, Any, Callable, Deque, Dict, List, Literal
 from uuid import UUID, uuid4
 
 import grpc
@@ -125,7 +126,7 @@ class RequeueException(Exception):
     pass
 
 
-ChildState = Literal["pending", "ready", "exiting"]
+ChildState = Literal["pending", "running", "exiting"]
 
 
 @dataclass
@@ -145,6 +146,7 @@ def __init__(
         max_child_task_count: int | None = None,
         namespace: str | None = None,
         concurrency: int = 1,
+        min_concurrency: int = 1,
         child_tasks_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         result_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         rebalance_after: int = DEFAULT_REBALANCE_AFTER,
@@ -176,6 +178,7 @@ def __init__(
             send_result_fn=self._send_results,
             max_child_task_count=max_child_task_count,
             concurrency=concurrency,
+            min_concurrency=min_concurrency,
             child_tasks_queue_maxsize=child_tasks_queue_maxsize,
             result_queue_maxsize=result_queue_maxsize,
             processing_pool_name=processing_pool_name,
@@ -574,6 +577,7 @@ def __init__(
         max_child_task_count: int | None = None,
         namespace: str | None = None,
         concurrency: int = 1,
+        min_concurrency: int = 1,
         child_tasks_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         result_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         rebalance_after: int = DEFAULT_REBALANCE_AFTER,
@@ -601,6 +605,7 @@ def __init__(
             send_result_fn=self._send_results,
             max_child_task_count=max_child_task_count,
             concurrency=concurrency,
+            min_concurrency=min_concurrency,
             child_tasks_queue_maxsize=child_tasks_queue_maxsize,
             result_queue_maxsize=result_queue_maxsize,
             processing_pool_name=processing_pool_name,
@@ -783,6 +788,7 @@ def __init__(
         mp_context: ForkContext | SpawnContext | ForkServerContext,
         max_child_task_count: int | None = None,
         concurrency: int = 1,
+        min_concurrency: int = 1,
         child_tasks_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         result_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         processing_pool_name: str | None = None,
@@ -792,6 +798,7 @@ def __init__(
         skip_awaiting_futures: bool = True,
     ) -> None:
         self._concurrency = concurrency
+        self._min_concurrency = min_concurrency
         self._processing_pool_name = processing_pool_name or "unknown"
         self._pod_name = pod_name or "unknown"
         self._update_in_batches = update_in_batches
@@ -824,40 +831,7 @@ def __init__(
     def ready_count(self) -> int:
         """Number of children that have finished warming up and are consuming."""
         with self._children_lock:
-            return sum(1 for c in self._children.values() if c.state == "ready")
-
-    def _plan_child_pool_changes(self) -> tuple[int, list[TrackedChild]]:
-        """
-        Determine how to move the child pool toward the desired concurrency.
-
-        The caller must hold `_children_lock`.
-
-        Exiting children are still consuming tasks until released, so they count
-        as serving capacity. Pending children are not serving yet, but they count
-        as replacement capacity that has already been requested.
-        """
-        pending_children = [c for c in self._children.values() if c.state == "pending"]
-        ready_children = [c for c in self._children.values() if c.state == "ready"]
-        exiting_children = [c for c in self._children.values() if c.state == "exiting"]
-
-        desired_serving_capacity = self._concurrency
-        current_serving_capacity = len(ready_children) + len(exiting_children)
-
-        excess_serving_capacity = max(current_serving_capacity - desired_serving_capacity, 0)
-        children_to_release = exiting_children[:excess_serving_capacity]
-
-        kept_exiting_count = len(exiting_children) - len(children_to_release)
-
-        # Keep enough pending children to both reach desired concurrency and replace
-        # exiting children that are still serving during their grace period
-        target_pending_count = max(
-            desired_serving_capacity - len(ready_children),
-            kept_exiting_count,
-        )
-
-        children_to_spawn = max(target_pending_count - len(pending_children), 0)
-
-        return children_to_spawn, children_to_release
+            return sum(1 for c in self._children.values() if c.state == "running")
 
     def send_results(self, results: list[ProcessingResult], is_draining: bool = False) -> None:
         """
@@ -968,6 +942,9 @@ def spawn_children_thread() -> None:
             # Queue of incoming message from children
             messages: multiprocessing.Queue[ChildMessage] = self._mp_context.Queue()
 
+            # Queue of children that want to exit
+            exiting: Deque[UUID] = deque()
+
             while not self._shutdown_event.is_set():
                 # Read any events that may have come in since the last loop iteration
                 received: List[ChildMessage] = []
@@ -1003,50 +980,38 @@ def spawn_children_thread() -> None:
                     for message in received:
                         child = self._children.get(message.child_id)
 
-                        if child is None:
-                            logger.warning(
-                                "taskworker.child_message.unknown_child",
-                                extra={
-                                    "cid": str(message.child_id),
-                                    "event": message.event,
-                                    "processing_pool": self._processing_pool_name,
-                                },
-                            )
+                        # If we received a message from a child, we MUST be tracking that child
+                        assert child is not None
 
-                            continue
+                        # This child is now running
+                        if message.event == "running":
+                            child.state = "running"
 
-                        if message.event == "ready":
-                            if child.state != "pending":
-                                logger.warning(
-                                    "taskworker.child.ready_not_pending",
-                                    extra={
-                                        "cid": str(message.child_id),
-                                        "state": child.state,
-                                        "processing_pool": self._processing_pool_name,
-                                    },
-                                )
+                        # This child wants to exit, but we may not have enough running children to shut down right away
+                        elif message.event == "exiting":
+                            exiting.append(message.child_id)
 
-                            child.state = "ready"
+                    while True:
+                        # Compute how many children are still running
+                        running = sum([1 for c in self._children.values() if c.state == "running"])
 
-                        elif message.event == "exiting":
-                            if child.state != "ready":
-                                logger.warning(
-                                    "taskworker.child.exiting_not_ready",
-                                    extra={
-                                        "cid": str(message.child_id),
-                                        "state": child.state,
-                                        "processing_pool": self._processing_pool_name,
-                                    },
-                                )
+                        if running <= self._min_concurrency:
+                            # We cannot shut down any more children without falling below minimum concurrency
+                            break
 
-                            child.state = "exiting"
+                        # We can shut down another child without falling below minimum concurrency
+                        child_id = exiting.popleft()
 
-                    children_to_spawn, children_to_release = self._plan_child_pool_changes()
+                        self._children[child_id].state = "exiting"
+                        self._children[child_id].release.set()
 
-                    for child in children_to_release:
-                        child.release.set()
+                # How many children do we need to spawn?
+                spawned = sum(
+                    [1 for c in self._children.values() if c.state in ["pending", "running"]]
+                )
+                needed = max(self._concurrency - spawned, 0)
 
-                for _ in range(children_to_spawn):
+                for _ in range(needed):
                     child_id = uuid4()
                     release = self._mp_context.Event()
 
diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py
index cc4fffb1..cccf7d3a 100644
--- a/clients/python/src/taskbroker_client/worker/workerchild.py
+++ b/clients/python/src/taskbroker_client/worker/workerchild.py
@@ -168,7 +168,7 @@ def _log_task_retry_exhausted(
 @dataclass(frozen=True)
 class ChildMessage:
     child_id: UUID
-    event: Literal["ready"] | Literal["exiting"]
+    event: Literal["running"] | Literal["exiting"]
 
 
 def child_process(
@@ -826,7 +826,7 @@ def _task_execution_complete(
         )
 
     # Tell the parent that this child has warmed up and is ready to consume tasks
-    messages.put_nowait(ChildMessage(child_id, "ready"))
+    messages.put_nowait(ChildMessage(child_id, "running"))
 
     # Run the worker loop
     run_worker(
diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py
index 0f68a397..c70d0f10 100644
--- a/clients/python/tests/worker/test_worker.py
+++ b/clients/python/tests/worker/test_worker.py
@@ -914,7 +914,7 @@ def test_spawn_children_releases_draining_child_after_replacement_ready() -> Non
         first_child_id = first_process.args[0]
         first_release = first_process.args[-1]
 
-        messages.put(ChildMessage(first_child_id, "ready"))
+        messages.put(ChildMessage(first_child_id, "running"))
         _wait_for(lambda: pool.ready_count == 1)
 
         messages.put(ChildMessage(first_child_id, "exiting"))
@@ -924,7 +924,7 @@ def test_spawn_children_releases_draining_child_after_replacement_ready() -> Non
 
         second_process = fake_context.processes[1]
         second_child_id = second_process.args[0]
-        messages.put(ChildMessage(second_child_id, "ready"))
+        messages.put(ChildMessage(second_child_id, "running"))
 
         _wait_for(first_release.is_set)
     finally:
@@ -1115,7 +1115,7 @@ def test_child_process_emits_ready_message() -> None:
 
     # The child signals readiness once warmup is done, before consuming.
     message = messages.get(timeout=1)
-    assert message == ChildMessage(child_id, "ready")
+    assert message == ChildMessage(child_id, "running")
 
 
 @mock.patch("taskbroker_client.worker.workerchild.capture_checkin")
@@ -1152,7 +1152,7 @@ def test_child_process_emits_exiting_once_and_continues_until_release(
         ready_message = messages.get(timeout=5)
         exiting_message = messages.get(timeout=5)
 
-        assert ready_message == ChildMessage(child_id, "ready")
+        assert ready_message == ChildMessage(child_id, "running")
         assert exiting_message == ChildMessage(child_id, "exiting")
         assert processed.get(timeout=5).task_id == SIMPLE_TASK.activation.id
 

From 26978a327161de8626223b3990644bcf00e21600 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:00:19 -0700
Subject: [PATCH 15/24] Formatting

---
 clients/python/src/taskbroker_client/worker/worker.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 1aa87415..7dac4ce1 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1006,9 +1006,7 @@ def spawn_children_thread() -> None:
                         self._children[child_id].release.set()
 
                 # How many children do we need to spawn?
-                spawned = sum(
-                    [1 for c in self._children.values() if c.state in ["pending", "running"]]
-                )
+                spawned = sum([1 for c in self._children.values() if c.state != "exiting"])
                 needed = max(self._concurrency - spawned, 0)
 
                 for _ in range(needed):

From f184e463b6990760ffcca9e8da2bcd1035517dfd Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:01:33 -0700
Subject: [PATCH 16/24] Compute Spawned w/Lock

---
 clients/python/src/taskbroker_client/worker/worker.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 7dac4ce1..3099d35d 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1005,8 +1005,9 @@ def spawn_children_thread() -> None:
                         self._children[child_id].state = "exiting"
                         self._children[child_id].release.set()
 
+                    spawned = sum([1 for c in self._children.values() if c.state != "exiting"])
+
                 # How many children do we need to spawn?
-                spawned = sum([1 for c in self._children.values() if c.state != "exiting"])
                 needed = max(self._concurrency - spawned, 0)
 
                 for _ in range(needed):

From 18a6502f3558f34aef98f070a62e828fb6c3aaa2 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:09:02 -0700
Subject: [PATCH 17/24] Check Exiting Queue Before Popping

---
 clients/python/src/taskbroker_client/worker/worker.py | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 0b452004..c9064e77 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -999,19 +999,22 @@ def spawn_children_thread() -> None:
 
                     while True:
                         # Compute how many children are still running
-                        running = sum([1 for c in self._children.values() if c.state == "running"])
+                        running = sum(1 for c in self._children.values() if c.state == "running")
 
                         if running <= self._min_concurrency:
                             # We cannot shut down any more children without falling below minimum concurrency
                             break
 
-                        # We can shut down another child without falling below minimum concurrency
+                        if not exiting:
+                            # No children are waiting to exit
+                            break
+
                         child_id = exiting.popleft()
 
                         self._children[child_id].state = "exiting"
                         self._children[child_id].release.set()
 
-                    spawned = sum([1 for c in self._children.values() if c.state != "exiting"])
+                    spawned = sum(1 for c in self._children.values() if c.state != "exiting")
 
                 # How many children do we need to spawn?
                 needed = max(self._concurrency - spawned, 0)

From fde094e5e6cb0dbe3ae10b73710455f136f803a7 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:24:27 -0700
Subject: [PATCH 18/24] Fix Tests

---
 clients/python/tests/worker/test_worker.py | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py
index ba8d8fe2..6b852fa8 100644
--- a/clients/python/tests/worker/test_worker.py
+++ b/clients/python/tests/worker/test_worker.py
@@ -1091,7 +1091,7 @@ def test_child_process_complete(mock_capture_checkin: mock.MagicMock) -> None:
     assert mock_capture_checkin.call_count == 0
 
 
-def test_child_process_emits_ready_message() -> None:
+def test_child_process_emits_running_message() -> None:
     todo: queue.Queue[InflightTaskActivation] = queue.Queue()
     processed: queue.Queue[ProcessingResult] = queue.Queue()
     shutdown = Event()
@@ -1117,7 +1117,7 @@ def test_child_process_emits_ready_message() -> None:
         parent_release=parent_release,
     )
 
-    # The child signals readiness once warmup is done, before consuming.
+    # The child signals readiness once warmup is done, before consuming
     message = messages.get(timeout=1)
     assert message == ChildMessage(child_id, "running")
 
@@ -1147,16 +1147,17 @@ def test_child_process_emits_exiting_once_and_continues_until_release(
             "test",
             "fork",
             False,
+            0.1,
             messages,
             parent_release,
         ),
     )
     process.start()
     try:
-        ready_message = messages.get(timeout=5)
+        running_message = messages.get(timeout=5)
         exiting_message = messages.get(timeout=5)
 
-        assert ready_message == ChildMessage(child_id, "running")
+        assert running_message == ChildMessage(child_id, "running")
         assert exiting_message == ChildMessage(child_id, "exiting")
         assert processed.get(timeout=5).task_id == SIMPLE_TASK.activation.id
 

From dfce05007dbf7ecd174b2f2415b76135c2f27ad0 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:27:47 -0700
Subject: [PATCH 19/24] Handle Unknown Child Message

---
 .../python/src/taskbroker_client/worker/worker.py    | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index c9064e77..a2800a89 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -987,7 +987,17 @@ def spawn_children_thread() -> None:
                         child = self._children.get(message.child_id)
 
                         # If we received a message from a child, we MUST be tracking that child
-                        assert child is not None
+                        if child is None:
+                            logger.warning(
+                                "taskworker.child_message.unknown_child",
+                                extra={
+                                    "cid": str(message.child_id),
+                                    "event": message.event,
+                                    "processing_pool": self._processing_pool_name,
+                                },
+                            )
+
+                            continue
 
                         # This child is now running
                         if message.event == "running":

From 0f61d82b84cd7b0476186f526115b38c273c2a88 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:39:33 -0700
Subject: [PATCH 20/24] Address Possible Deadlock Scenario

---
 clients/python/src/taskbroker_client/worker/worker.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index a2800a89..cc4391c0 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -803,7 +803,12 @@ def __init__(
         future_checking_frequency: float = 0.1,
     ) -> None:
         self._concurrency = concurrency
-        self._min_concurrency = min_concurrency
+
+        if min_concurrency < concurrency:
+            self._min_concurrency = min_concurrency
+        else:
+            raise ValueError("Minimum concurrency must be strictly below concurrency")
+
         self._processing_pool_name = processing_pool_name or "unknown"
         self._pod_name = pod_name or "unknown"
         self._update_in_batches = update_in_batches
@@ -1012,7 +1017,7 @@ def spawn_children_thread() -> None:
                         running = sum(1 for c in self._children.values() if c.state == "running")
 
                         if running <= self._min_concurrency:
-                            # We cannot shut down any more children without falling below minimum concurrency
+                            # Cannot shut down any more children without falling below minimum concurrency (guaranteed < concurrency)
                             break
 
                         if not exiting:

From ad33a41006dcf405add84e60876b08d00359c183 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:42:18 -0700
Subject: [PATCH 21/24] Default `min_concurrency` to 0

---
 clients/python/src/taskbroker_client/worker/worker.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index cc4391c0..744cdd2d 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -146,7 +146,7 @@ def __init__(
         max_child_task_count: int | None = None,
         namespace: str | None = None,
         concurrency: int = 1,
-        min_concurrency: int = 1,
+        min_concurrency: int = 0,
         child_tasks_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         result_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         rebalance_after: int = DEFAULT_REBALANCE_AFTER,
@@ -579,7 +579,7 @@ def __init__(
         max_child_task_count: int | None = None,
         namespace: str | None = None,
         concurrency: int = 1,
-        min_concurrency: int = 1,
+        min_concurrency: int = 0,
         child_tasks_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         result_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         rebalance_after: int = DEFAULT_REBALANCE_AFTER,
@@ -792,7 +792,7 @@ def __init__(
         mp_context: ForkContext | SpawnContext | ForkServerContext,
         max_child_task_count: int | None = None,
         concurrency: int = 1,
-        min_concurrency: int = 1,
+        min_concurrency: int = 0,
         child_tasks_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         result_queue_maxsize: int = DEFAULT_WORKER_QUEUE_SIZE,
         processing_pool_name: str | None = None,

From 02b4ac9dfdee1fc343f6c386ad7928d8d17728d9 Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Mon, 29 Jun 2026 14:49:23 -0700
Subject: [PATCH 22/24] Try Fixing Tests

---
 clients/python/tests/worker/test_worker.py | 41 ++++++++++++++++++----
 1 file changed, 34 insertions(+), 7 deletions(-)

diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py
index 6b852fa8..5eb92829 100644
--- a/clients/python/tests/worker/test_worker.py
+++ b/clients/python/tests/worker/test_worker.py
@@ -397,6 +397,7 @@ def _make_fake_context_pool(
     fake_context: _FakeContext,
     *,
     concurrency: int = 1,
+    min_concurrency: int = 0,
 ) -> TaskWorkerProcessingPool:
     return TaskWorkerProcessingPool(
         app_module="examples.app:app",
@@ -404,6 +405,7 @@ def _make_fake_context_pool(
         mp_context=fake_context,  # type: ignore[arg-type]
         max_child_task_count=1,
         concurrency=concurrency,
+        min_concurrency=min_concurrency,
         processing_pool_name="test",
         process_type="fork",
     )
@@ -904,30 +906,55 @@ def test_spawn_children_counts_pending_children_toward_concurrency() -> None:
         pool.shutdown()
 
 
-def test_spawn_children_releases_draining_child_after_replacement_ready() -> None:
+def test_spawn_children_releases_draining_child_above_min_concurrency() -> None:
     fake_context = _FakeContext()
-    pool = _make_fake_context_pool(fake_context, concurrency=1)
+    pool = _make_fake_context_pool(fake_context, concurrency=2, min_concurrency=1)
 
     pool.start_spawn_children_thread()
     try:
-        _wait_for(lambda: len(fake_context.processes) == 1)
+        _wait_for(lambda: len(fake_context.processes) == 2)
         messages = fake_context.queues[-1]
         first_process = fake_context.processes[0]
         first_child_id = first_process.args[0]
         first_release = first_process.args[-1]
 
         messages.put(ChildMessage(first_child_id, "running"))
-        _wait_for(lambda: pool.ready_count == 1)
+        second_process = fake_context.processes[1]
+        second_child_id = second_process.args[0]
+        messages.put(ChildMessage(second_child_id, "running"))
+        _wait_for(lambda: pool.ready_count == 2)
 
         messages.put(ChildMessage(first_child_id, "exiting"))
-        _wait_for(lambda: len(fake_context.processes) == 2)
+        _wait_for(first_release.is_set)
 
-        assert not first_release.is_set()
+        _wait_for(lambda: len(fake_context.processes) == 3)
+    finally:
+        pool.shutdown()
+
+
+def test_spawn_children_defers_draining_child_at_min_concurrency() -> None:
+    fake_context = _FakeContext()
+    pool = _make_fake_context_pool(fake_context, concurrency=2, min_concurrency=1)
+
+    pool.start_spawn_children_thread()
+    try:
+        _wait_for(lambda: len(fake_context.processes) == 2)
+        messages = fake_context.queues[-1]
+        first_process = fake_context.processes[0]
+        first_child_id = first_process.args[0]
+        first_release = first_process.args[-1]
 
         second_process = fake_context.processes[1]
         second_child_id = second_process.args[0]
-        messages.put(ChildMessage(second_child_id, "running"))
 
+        messages.put(ChildMessage(first_child_id, "running"))
+        _wait_for(lambda: pool.ready_count == 1)
+
+        messages.put(ChildMessage(first_child_id, "exiting"))
+        time.sleep(0.25)
+        assert not first_release.is_set()
+
+        messages.put(ChildMessage(second_child_id, "running"))
         _wait_for(first_release.is_set)
     finally:
         pool.shutdown()

From b37ee867112f1f47ebdaebdef6a7e2b4284d19ef Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Tue, 30 Jun 2026 07:17:56 -0700
Subject: [PATCH 23/24] Add Child Management Metrics

---
 .../src/taskbroker_client/worker/worker.py    | 91 ++++++++++++++-----
 .../taskbroker_client/worker/workerchild.py   | 13 ++-
 2 files changed, 76 insertions(+), 28 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 744cdd2d..4d83eb13 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -832,6 +832,7 @@ def __init__(
             maxsize=result_queue_maxsize
         )
         self._children: Dict[UUID, TrackedChild] = {}
+        self._exiting_children: Deque[UUID] = deque()
         self._children_lock = threading.Lock()
         self._shutdown_event = self._mp_context.Event()
         self._result_thread: threading.Thread | None = None
@@ -844,6 +845,60 @@ def ready_count(self) -> int:
         with self._children_lock:
             return sum(1 for c in self._children.values() if c.state == "running")
 
+    def _emit_periodic_metrics(self) -> None:
+        tags = {
+            "processing_pool": self._processing_pool_name,
+            "pod_name": self._pod_name,
+        }
+
+        # Emit queue size metrics
+        try:
+            # Method 'qsize' not implemented on all platforms, such as macOS
+            self._metrics.gauge(
+                "taskworker.child_tasks.size",
+                float(self._child_tasks.qsize()),
+                tags=tags,
+            )
+
+            self._metrics.gauge(
+                "taskworker.processed_tasks.size",
+                float(self._processed_tasks.qsize()),
+                tags=tags,
+            )
+        except Exception as e:
+            logger.debug(
+                "taskworker.worker.queue_gauges.error",
+                extra={"error": e, "processing_pool": self._processing_pool_name},
+            )
+
+        # Count the number of children in each state and waiting for exit
+        with self._children_lock:
+            state_counts: dict[ChildState, int] = {
+                "pending": 0,
+                "running": 0,
+                "exiting": 0,
+            }
+
+            for child in self._children.values():
+                state_counts[child.state] += 1
+
+            exiting_children = len(self._exiting_children)
+
+        # Emit number of children in each state
+        for state, count in state_counts.items():
+            self._metrics.gauge(
+                "taskworker.worker.children",
+                float(count),
+                tags={**tags, "state": state},
+            )
+
+        # Emit number of children waiting to exit
+        self._metrics.gauge(
+            "taskworker.worker.exiting_children.size",
+            float(exiting_children),
+            tags=tags,
+        )
+
     def send_results(self, results: list[ProcessingResult], is_draining: bool = False) -> None:
         """
         Call the passed in function. If is_draining is True, the function should not fetch a new task.
@@ -870,25 +925,9 @@ def start_metrics_thread(self) -> None:
         """
 
         def metrics_thread() -> None:
-            tags = {
-                "processing_pool": self._processing_pool_name,
-                "pod_name": self._pod_name,
-            }
-
             while True:
                 try:
-                    # 'qsize' is not implemented on all platforms, such as macOS
-                    self._metrics.gauge(
-                        "taskworker.child_tasks.size",
-                        float(self._child_tasks.qsize()),
-                        tags=tags,
-                    )
-
-                    self._metrics.gauge(
-                        "taskworker.processed_tasks.size",
-                        float(self._processed_tasks.qsize()),
-                        tags=tags,
-                    )
+                    self._emit_periodic_metrics()
                 except Exception as e:
                     logger.debug(
                         "taskworker.worker.queue_gauges.error",
@@ -953,9 +992,6 @@ def spawn_children_thread() -> None:
             # Queue of incoming message from children
             messages: multiprocessing.Queue[ChildMessage] = self._mp_context.Queue()
 
-            # Queue of children that want to exit
-            exiting: Deque[UUID] = deque()
-
             while not self._shutdown_event.is_set():
                 # Read any events that may have come in since the last loop iteration
                 received: List[ChildMessage] = []
@@ -1010,7 +1046,7 @@ def spawn_children_thread() -> None:
 
                         # This child wants to exit, but we may not have enough running children to shut down right away
                         elif message.event == "exiting":
-                            exiting.append(message.child_id)
+                            self._exiting_children.append(message.child_id)
 
                     while True:
                         # Compute how many children are still running
@@ -1020,14 +1056,19 @@ def spawn_children_thread() -> None:
                             # Cannot shut down any more children without falling below minimum concurrency (guaranteed < concurrency)
                             break
 
-                        if not exiting:
+                        if not self._exiting_children:
                             # No children are waiting to exit
                             break
 
-                        child_id = exiting.popleft()
+                        child_id = self._exiting_children.popleft()
+                        child = self._children.get(child_id)
+
+                        # Child may have died already
+                        if child is None:
+                            continue
 
-                        self._children[child_id].state = "exiting"
-                        self._children[child_id].release.set()
+                        child.state = "exiting"
+                        child.release.set()
 
                     spawned = sum(1 for c in self._children.values() if c.state != "exiting")
 
diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py
index bef1bd43..b88be6e3 100644
--- a/clients/python/src/taskbroker_client/worker/workerchild.py
+++ b/clients/python/src/taskbroker_client/worker/workerchild.py
@@ -367,11 +367,12 @@ def check_task_future_completion(
         )
         _future_completion_thread.start()
 
-        waiting_for_parent_release = False
+        # Track when exit was initiated for metrics
+        exit_initiated: float | None = None
 
         while not shutdown_event.is_set() and not local_shutdown.is_set():
             if max_task_count and processed_task_count >= max_task_count:
-                if not waiting_for_parent_release:
+                if exit_initiated is None:
                     metrics.incr(
                         "taskworker.worker.max_task_count_reached",
                         tags={
@@ -386,9 +387,15 @@ def check_task_future_completion(
 
                     # Tell the parent this child can be replaced.
                     messages.put_nowait(ChildMessage(child_id, "exiting"))
-                    waiting_for_parent_release = True
+                    exit_initiated = time.monotonic()
 
                 if parent_release.is_set():
+                    metrics.distribution(
+                        "taskworker.worker.child.parent_release_wait_duration",
+                        time.monotonic() - exit_initiated,
+                        tags={"processing_pool": processing_pool_name},
+                    )
+
                     local_shutdown.set()
                     break
 

From 92bb70a0038d59418dc7e30edf21caac34e266ec Mon Sep 17 00:00:00 2001
From: james-mcnulty <george.berdovskiy@sentry.io>
Date: Tue, 30 Jun 2026 14:16:05 -0700
Subject: [PATCH 24/24] Fix Possible Process Start Race Condition

---
 .../src/taskbroker_client/worker/worker.py       | 16 +++++++++-------
 1 file changed, 9 insertions(+), 7 deletions(-)

diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py
index 82dccc8b..7f89d5d1 100644
--- a/clients/python/src/taskbroker_client/worker/worker.py
+++ b/clients/python/src/taskbroker_client/worker/worker.py
@@ -1149,6 +1149,15 @@ def spawn_children_thread() -> None:
 
                     try:
                         process.start()
+
+                        with self._children_lock:
+                            child = TrackedChild(
+                                process=process,
+                                state="pending",
+                                release=release,
+                            )
+
+                            self._children[child_id] = child
                     except Exception as e:
                         logger.exception(
                             "taskworker.child.spawn.failed",
@@ -1169,13 +1178,6 @@ def spawn_children_thread() -> None:
 
                         continue
 
-                    with self._children_lock:
-                        self._children[child_id] = TrackedChild(
-                            process=process,
-                            state="pending",
-                            release=release,
-                        )
-
                     logger.info(
                         "taskworker.spawn_child",
                         extra={