Added support to Multiplexer and Manager to terminate durable workers

sanic/worker/manager.py

@@ -408,6 +408,23 @@ def remove_worker(self, worker: Worker) -> None:
         logger.info("Removed worker %s", worker.ident)
         del worker
 
+    def terminate_durable_worker(self, ident: str) -> None:
+        """Terminate a durable worker and its processes.
+
+        Args:
+            ident (str): The name of the durable process to terminate.
+        """
+        worker = self.durable.get(ident)
+        if not worker:
+            error_logger.error(
+                "Durable worker termination failed (%s not found).", worker
+            )
+            return
+        for process in worker.processes:
+            process.terminate()
+        self.remove_worker(worker)
+        logger.info("Terminated durable worker %s", ident)
+
     @property
     def pid(self):
         """Get the process ID of the main process."""
@@ -452,7 +469,14 @@ def _poll_monitor(self) -> Optional[MonitorCycle]:
             elif message == "__TERMINATE__":
                 self._handle_terminate()
                 return MonitorCycle.BREAK
-            elif isinstance(message, tuple) and len(message) == 7:
+            elif isinstance(message, str) and message.startswith(
+                "__TERMINATE_WORKER__"
+            ):
+                self._handle_terminate_worker(
+                    *message.split(":")[1].split(",")
+                )
+                return MonitorCycle.CONTINUE
+            elif isinstance(message, tuple) and len(message) == 8:
                 self._handle_manage(*message)  # type: ignore
                 return MonitorCycle.CONTINUE
             elif not isinstance(message, str):
@@ -466,6 +490,10 @@ def _poll_monitor(self) -> Optional[MonitorCycle]:
     def _handle_terminate(self) -> None:
         self.shutdown()
 
+    def _handle_terminate_worker(self, *args) -> None:
+        for ident in args:
+            self.terminate_durable_worker(ident)
+
     def _handle_message(self, message: str) -> Optional[MonitorCycle]:
         logger.debug(
             "Incoming monitor message: %s",

sanic/worker/multiplexer.py

@@ -146,6 +146,14 @@ def terminate(self, early: bool = False):
         message = "__TERMINATE_EARLY__" if early else "__TERMINATE__"
         self._monitor_publisher.send(message)
 
+    def terminate_worker(self, ident: str):
+        """Terminate a worker.
+
+        Args:
+            ident (str): The name of the worker to terminate.
+        """
+        self._monitor_publisher.send(f"__TERMINATE_WORKER__:{ident}")
+
     @property
     def pid(self) -> int:
         """The process ID of the worker."""

tests/worker/test_manager.py

@@ -50,7 +50,7 @@ def test_terminate(os_mock: Mock):
 
 
 @patch("sanic.worker.process.os")
-def test_shutown(os_mock: Mock):
+def test_shutdown(os_mock: Mock):
     process = Mock()
     process.pid = 1234
     process.is_alive.return_value = True
@@ -417,6 +417,22 @@ def test_remove_worker(manager: WorkerManager, caplog):
     assert ("sanic.error", 40, message) in caplog.record_tuples
 
 
+def test_terminate_durable_worker(manager: WorkerManager, caplog):
+    worker = manager.manage("TEST", fake_serve, kwargs={})
+
+    assert "Sanic-TEST-0" in worker.worker_state
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+
+    manager.terminate_durable_worker("WRONG")
+    message = "Durable worker termination failed because WRONG not found."
+
+    assert "Sanic-TEST-0" in worker.worker_state
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+    assert ("sanic.error", 40, message) in caplog.record_tuples
+
+
 def test_remove_untracked_worker(manager: WorkerManager, caplog):
     caplog.set_level(20)
     worker = manager.manage("TEST", fake_serve, kwargs={}, tracked=False)

tests/worker/test_multiplexer.py

@@ -102,6 +102,13 @@ def test_terminate(monitor_publisher: Mock, m: WorkerMultiplexer):
     monitor_publisher.send.assert_called_once_with("__TERMINATE__")
 
 
+def test_terminate_worker(monitor_publisher: Mock, m: WorkerMultiplexer):
+    m.terminate_worker("foo,bar")
+    monitor_publisher.send.assert_called_once_with(
+        "__TERMINATE_WORKER__:foo,bar"
+    )
+
+
 def test_scale(monitor_publisher: Mock, m: WorkerMultiplexer):
     m.scale(99)
     monitor_publisher.send.assert_called_once_with("__SCALE__:99")