fix(worker): keep pull worker healthy while queue is full

clients/python/src/taskbroker_client/worker/client.py

@@ -309,7 +309,7 @@ def __init__(
         self._timestamp_since_touch_lock = threading.Lock()
         self._timestamp_since_touch = 0.0
 
-    def _emit_health_check(self) -> None:
+    def emit_health_check(self) -> None:
         if self._health_check_settings is None:
             return
 
@@ -403,7 +403,7 @@ def get_task(self, namespace: str | None = None) -> InflightTaskActivation | Non
         If a namespace is provided, only tasks for that namespace will be fetched.
         This will return None if there are no tasks to fetch.
         """
-        self._emit_health_check()
+        self.emit_health_check()
 
         request = GetTaskRequest(application=self._application, namespace=namespace)
         try:
@@ -455,7 +455,7 @@ def update_task(
 
         The return value is the next task that should be executed.
         """
-        self._emit_health_check()
+        self.emit_health_check()
         if fetch_next_task is not None:
             fetch_next_task.application = self._application
 

clients/python/src/taskbroker_client/worker/worker.py

@@ -663,6 +663,7 @@ def _add_task(self) -> bool:
         Add a task to child tasks queue. Returns False if no new task was fetched.
         """
         if self.worker_pool.is_worker_full():
+            self.client.emit_health_check()
             self._metrics.incr(
                 "taskworker.worker.add_tasks.child_tasks_full",
                 tags={"processing_pool": self._processing_pool_name},

clients/python/tests/worker/test_worker.py

@@ -723,6 +723,35 @@ def test_constructor_push_mode(self) -> None:
         self.assertEqual(taskworker._grpc_port, 50099)
 
 
+def test_pull_worker_health_check_touches_while_full(tmp_path: Path) -> None:
+    health_check_path = tmp_path / "health"
+    taskworker = TaskWorker(
+        app_module="examples.app:app",
+        broker_hosts=["127.0.0.1:50051"],
+        max_child_task_count=100,
+        process_type="fork",
+        health_check_file_path=str(health_check_path),
+        health_check_sec_per_touch=1,
+    )
+
+    with (
+        mock.patch.object(taskworker.worker_pool, "is_worker_full", return_value=True),
+        mock.patch.object(taskworker.client, "get_task") as mock_get_task,
+        mock.patch("taskbroker_client.worker.worker.time.sleep"),
+        mock.patch("taskbroker_client.worker.client.time") as mock_time,
+    ):
+        mock_time.time.return_value = 1
+        taskworker.run_once()
+        assert health_check_path.exists()
+
+        health_check_path.unlink()
+        mock_time.time.return_value = 3
+        taskworker.run_once()
+
+    assert health_check_path.exists()
+    mock_get_task.assert_not_called()
+
+
 def test_push_worker_health_check_touches_while_idle(tmp_path: Path) -> None:
     taskworker = PushTaskWorker(
         app_module="examples.app:app",