fix: prevent duplicate concurrent pods on unresponsive activation restart

  When the monitor detects an unresponsive activation and triggers a
  restart, _watch_pod_deletion falsely reports success if the watch
  stream times out without receiving a DELETED event. This allows
  _unresponsive_policy to restart the activation while the original
  pod is still running, producing ghost pods that compete for shared
  resources like Kafka consumer groups.

  Two fixes:

  1. _watch_pod_deletion now tracks whether a DELETED event was actually
     received. If the watch times out without one, it retries instead of
     reporting success. After all retries, it raises ContainerCleanupError.

  2. _unresponsive_policy now calls _cleanup explicitly via a new
     _handle_unresponsive_activation helper. If cleanup fails, the
     restart is skipped to avoid creating a duplicate pod. The activation
     is still marked FAILED regardless of cleanup outcome.
  3. adds a list_then_watch flow for deleting a pod so that we don't loose track
     of the pods we are trying to delete
  Fixes: AAP-81201

Author: AlexSCorey

src/aap_eda/services/activation/activation_manager.py

@@ -377,45 +377,52 @@ def _unresponsive_policy(self, check_type: str):
             "Unresponsive policy called for "
             f"activation id: {self.db_instance.id}",
         )
-        container_logger = self.container_logger_class(self.latest_instance.id)
-        if self.db_instance.restart_policy == RestartPolicy.NEVER:
-            LOGGER.info(
-                f"Activation id: {self.db_instance.id} "
-                f"Restart policy is set to {self.db_instance.restart_policy}."
-                "No restart policy is applied."
-            )
-            user_msg = (
-                "Activation is unresponsive. "
-                f"{check_type} check for ansible-rulebook timed out. "
-                "Restart policy is not applicable."
+        try:
+            self._handle_unresponsive_activation(check_type=check_type)
+        except engine_exceptions.ContainerCleanupError as e:
+            LOGGER.error(
+                f"Error occurred while handling unresponsive activation: {e}"
             )
-            container_logger.write(user_msg, flush=True)
-            self._fail_instance(user_msg)
-            self.set_status(
-                ActivationStatus.FAILED,
-                user_msg,
+            return
+
+        if not self.db_instance.restart_policy == RestartPolicy.NEVER:
+            system_restart_activation(
+                self.db_instance_type, self.db_instance.id, delay_seconds=1
             )
 
-        else:
-            LOGGER.info(
+    def _handle_unresponsive_activation(self, check_type: str):
+        try:
+            self._cleanup()
+        except engine_exceptions.ContainerCleanupError:
+            raise
+        finally:
+            container_logger = self.container_logger_class(
+                self.latest_instance.id
+            )
+            log_message = (
                 f"Activation id: {self.db_instance.id} "
                 f"Restart policy is set to {self.db_instance.restart_policy}."
-                "Restart policy is applied.",
             )
             user_msg = (
                 "Activation is unresponsive. "
                 f"{check_type} check for ansible-rulebook timed out. "
-                "Activation is going to be restarted."
             )
+
+            if self.db_instance.restart_policy == RestartPolicy.NEVER:
+                log_message += " Restart policy is not applicable."
+                user_msg += " Restart policy is not applicable."
+            else:
+                log_message += " Restart policy is applied."
+                user_msg += " Activation is going to be restarted."
+
+            LOGGER.info(log_message)
+
             container_logger.write(user_msg, flush=True)
             self._fail_instance(msg=user_msg)
             self.set_status(
                 ActivationStatus.FAILED,
                 user_msg,
             )
-            system_restart_activation(
-                self.db_instance_type, self.db_instance.id, delay_seconds=1
-            )
 
     def _missing_container_policy(self):
         LOGGER.info(

src/aap_eda/services/activation/engine/kubernetes.py

@@ -50,7 +50,7 @@
 
 K8S_API_RETRIES = 3
 K8S_API_RETRY_BACKOFF = 1.0
-K8S_API_TRANSIENT_STATUS_CODES = {401, 403, 500, 502, 503, 504}
+K8S_API_TRANSIENT_STATUS_CODES = {401, 403, 410, 500, 502, 503, 504}
 
 INVALID_IMAGE_NAME = "InvalidImageName"
 IMAGE_PULL_BACK_OFF = "ImagePullBackOff"
@@ -566,9 +566,22 @@ def _delete_job(self, log_handler: LogHandler) -> None:
         activation_job_name = activation_job.items[0].metadata.name
         self._delete_job_resource(activation_job_name, log_handler)
 
+    def _get_resource_version(self) -> str:
+        pod_list = self._call_k8s_api(
+            self.client.core_api.list_namespaced_pod,
+            namespace=self.namespace,
+            label_selector=f"job-name={self.job_name}",
+            error_cls=ContainerCleanupError,
+            description=f"list namespaced jobs {self.job_name}",
+        )
+        resource_version = pod_list.metadata.resource_version
+        return resource_version
+
     def _delete_job_resource(
         self, job_name: str, log_handler: LogHandler
     ) -> None:
+        resource_version = self._get_resource_version()
+
         result = self._call_k8s_api(
             self.client.batch_api.delete_namespaced_job,
             name=job_name,
@@ -581,32 +594,42 @@ def _delete_job_resource(
         if result.status == "Failure":
             raise ContainerCleanupError(f"{result}")
 
-        self._watch_pod_deletion(log_handler)
+        self._watch_pod_deletion(
+            log_handler, resource_version=resource_version
+        )
 
-    def _watch_pod_deletion(self, log_handler: LogHandler) -> None:
+    def _watch_pod_deletion(
+        self, log_handler: LogHandler, resource_version: str
+    ) -> None:
         """Watch for pod deletion with retry on transient errors."""
         desc = f"watch pod deletion {self.job_name}"
         last_exc = None
         for attempt in range(1, K8S_API_RETRIES + 1):
             watcher = watch.Watch()
+            pod_deleted = False
+            if attempt > 1:
+                resource_version = self._get_resource_version()
             try:
                 for event in watcher.stream(
                     self.client.core_api.list_namespaced_pod,
                     namespace=self.namespace,
                     label_selector=f"job-name={self.job_name}",
                     timeout_seconds=POD_DELETE_TIMEOUT,
+                    resource_version=resource_version,
                 ):
                     if event["type"] == "DELETED":
                         log_handler.write(
                             f"Pod '{self.job_name}' is deleted.",
                             flush=True,
                         )
+                        pod_deleted = True
                         break
-                log_handler.write(
-                    f"Job {self.job_name} is cleaned up.",
-                    flush=True,
-                )
-                return
+                if pod_deleted:
+                    log_handler.write(
+                        f"Job {self.job_name} is cleaned up.",
+                        flush=True,
+                    )
+                    return
             except ApiException as exc:
                 last_exc = exc
                 if exc.status == status.HTTP_404_NOT_FOUND:
@@ -660,7 +683,7 @@ def _process_pod_start_event(self, event) -> bool:
             )
         return False
 
-    def _wait_for_pod_to_start(self, _log_handler: LogHandler) -> None:
+    def _wait_for_pod_to_start(self, log_handler: LogHandler) -> None:
         LOGGER.info("Waiting for pod to start")
         desc = f"watch pod start {self.job_name}"
         last_exc = None

tests/integration/services/activation/engine/test_kubernetes.py

@@ -714,7 +714,11 @@ def test_delete_job(mock_watch, init_kubernetes_data, kubernetes_engine):
 
     with mock.patch.object(engine.client, "batch_api") as batch_api_mock:
         batch_api_mock.list_namespaced_job.return_value.items = [job_mock]
-        engine._delete_job(log_handler)
+        mock_watch.return_value.stream.return_value = [{"type": "DELETED"}]
+        with mock.patch.object(
+            engine, "_get_resource_version", return_value="1"
+        ):
+            engine._delete_job(log_handler)
 
         batch_api_mock.delete_namespaced_job.assert_called_once()
 
@@ -730,7 +734,10 @@ def test_delete_job(mock_watch, init_kubernetes_data, kubernetes_engine):
     with mock.patch.object(engine.client, "batch_api") as batch_api_mock:
         batch_api_mock.list_namespaced_job.return_value.items = [job_mock]
         mock_watch.return_value.stream.return_value = [event]
-        engine._delete_job(log_handler)
+        with mock.patch.object(
+            engine, "_get_resource_version", return_value="1"
+        ):
+            engine._delete_job(log_handler)
 
         log_messages = [
             record.log for record in models.RulebookProcessLog.objects.all()
@@ -1153,8 +1160,11 @@ def test_watch_pod_deletion_retries_on_transient(
 
     mock_watch.return_value.stream.side_effect = ApiException(status=503)
 
-    with pytest.raises(ContainerCleanupError, match="retries"):
-        engine._watch_pod_deletion(log_handler)
+    with mock.patch.object(
+        engine, "_get_resource_version", return_value="999"
+    ):
+        with pytest.raises(ContainerCleanupError, match="retries"):
+            engine._watch_pod_deletion(log_handler, resource_version="1")
 
     assert mock_watch.return_value.stream.call_count == K8S_API_RETRIES
     assert mock_watch.return_value.stop.call_count == K8S_API_RETRIES
@@ -1204,6 +1214,91 @@ def test_wait_for_pod_to_start_raises_on_non_transient(
     assert mock_watch.return_value.stream.call_count == 1
 
 
+@mock.patch("aap_eda.services.activation.engine.kubernetes.watch.Watch")
+@mock.patch("aap_eda.services.activation.engine.kubernetes.time.sleep")
+@pytest.mark.django_db
+def test_watch_pod_deletion_timeout_without_deleted_event(
+    sleep_mock,
+    mock_watch,
+    init_kubernetes_data,
+    kubernetes_engine,
+):
+    """_watch_pod_deletion raises ContainerCleanupError when watch times out
+    without receiving a DELETED event across all retries."""
+    engine = kubernetes_engine
+    engine.job_name = "test-job"
+    log_handler = DBLogger(init_kubernetes_data.activation_instance.id)
+
+    mock_watch.return_value.stream.return_value = iter(
+        [
+            {"type": "MODIFIED"},
+        ]
+    )
+
+    with mock.patch.object(
+        engine, "_get_resource_version", return_value="999"
+    ):
+        with pytest.raises(ContainerCleanupError, match="retries"):
+            engine._watch_pod_deletion(log_handler, resource_version="1")
+
+    assert mock_watch.return_value.stream.call_count == K8S_API_RETRIES
+    assert mock_watch.return_value.stop.call_count == K8S_API_RETRIES
+
+
+@mock.patch("aap_eda.services.activation.engine.kubernetes.watch.Watch")
+@pytest.mark.django_db
+def test_watch_pod_deletion_empty_stream_retries(
+    mock_watch,
+    init_kubernetes_data,
+    kubernetes_engine,
+):
+    """_watch_pod_deletion retries when the watch stream ends without any
+    events (simulating a pure timeout with no events)."""
+    engine = kubernetes_engine
+    engine.job_name = "test-job"
+    log_handler = DBLogger(init_kubernetes_data.activation_instance.id)
+
+    mock_watch.return_value.stream.return_value = iter([])
+
+    with mock.patch.object(
+        engine, "_get_resource_version", return_value="999"
+    ):
+        with pytest.raises(ContainerCleanupError, match="retries"):
+            engine._watch_pod_deletion(log_handler, resource_version="1")
+
+    assert mock_watch.return_value.stream.call_count == K8S_API_RETRIES
+
+
+@mock.patch("aap_eda.services.activation.engine.kubernetes.watch.Watch")
+@pytest.mark.django_db
+def test_watch_pod_deletion_success_on_deleted_event(
+    mock_watch,
+    init_kubernetes_data,
+    kubernetes_engine,
+):
+    """_watch_pod_deletion returns successfully when a DELETED event
+    is received."""
+    engine = kubernetes_engine
+    engine.job_name = "test-job"
+    log_handler = DBLogger(init_kubernetes_data.activation_instance.id)
+
+    mock_watch.return_value.stream.return_value = iter(
+        [
+            {"type": "MODIFIED"},
+            {"type": "DELETED"},
+        ]
+    )
+
+    engine._watch_pod_deletion(log_handler, resource_version="1")
+
+    assert mock_watch.return_value.stream.call_count == 1
+    log_messages = [
+        record.log for record in models.RulebookProcessLog.objects.all()
+    ]
+    assert f"Pod '{engine.job_name}' is deleted." in log_messages
+    assert f"Job {engine.job_name} is cleaned up." in log_messages
+
+
 @pytest.mark.django_db
 def test_process_pod_start_event_pending(kubernetes_engine):
     """_process_pod_start_event returns False for Pending."""