Centralize failed subhandler check in StateBasedSubHandler

Author: tomach

crate/operator/restore_backup.py

@@ -21,7 +21,6 @@
 
 import abc
 import asyncio
-import json
 import logging
 import re
 from contextlib import asynccontextmanager
@@ -44,7 +43,6 @@
 from crate.operator.constants import (
     API_GROUP,
     GC_USERNAME,
-    KOPF_STATE_STORE_PREFIX,
     RESOURCE_CRATEDB,
     SYSTEM_USERNAME,
     BackupStorageProvider,
@@ -925,21 +923,9 @@ async def handle(  # type: ignore
         """
 
         # check whether any dependent handlers have failed.
-        failed_handlers = []
         annotations = kwargs["annotations"]
         logger.info("RestoreInternalTablesSubHandler checking handler status...")
-        for dep in self.depends_on:
-            key = kopf.AnnotationsProgressStorage(
-                v1=False, prefix=KOPF_STATE_STORE_PREFIX
-            ).make_v2_key(dep)
-
-            status_str = annotations.get(key)
-            logger.info(f"[{dep}] annotation raw value: {status_str}")
-            if status_str:
-                parsed = json.loads(status_str)
-                if not parsed.get("success"):
-                    failed_handlers.append(dep)
-        logger.info(f"Failed handlers: {failed_handlers}")
+        failed_handlers = self._get_failed_dependent_handlers(annotations, logger)
 
         async with GlobalApiClient() as api_client:
             core = CoreV1Api(api_client)

crate/operator/rollback.py

@@ -42,18 +42,7 @@ async def handle(
 
         logger.info("Evaluating rollback conditions...")
 
-        failed_handlers = []
-        for dep in self.depends_on:
-            key = kopf.AnnotationsProgressStorage(
-                v1=False, prefix=KOPF_STATE_STORE_PREFIX
-            ).make_v2_key(dep)
-
-            status_str = annotations.get(key)
-            logger.info(f"[{dep}] annotation raw value: {status_str}")
-            if status_str:
-                parsed = json.loads(status_str)
-                if not parsed.get("success"):
-                    failed_handlers.append(dep)
+        failed_handlers = self._get_failed_dependent_handlers(annotations, logger)
 
         if failed_handlers:
             logger.info(f"Rollback triggered due to failed handlers: {failed_handlers}")

crate/operator/utils/kopf.py

@@ -25,7 +25,7 @@
 import logging
 from datetime import datetime, timezone
 from functools import wraps
-from typing import Any, Callable, Optional, TypedDict
+from typing import Any, Callable, List, Optional, TypedDict
 
 import kopf
 
@@ -271,6 +271,41 @@ def _init_handler_starttime(
                 "ref": self.ref,
             }
 
+    def _get_failed_dependent_handlers(
+        self, annotations: dict, logger: logging.Logger
+    ) -> List[str]:
+        """
+        Determine which dependent handlers have failed based on kopf annotations.
+
+        This inspects the progress annotations stored by kopf for each dependency in
+        ``self.depends_on`` and returns a list of handlers that did not succeed.
+
+        Used by subhandlers that need to decide whether to perform cleanup or rollback
+        actions depending on the outcome of previous handlers.
+        """
+        failed_handlers: List[str] = []
+        logger.info("Checking dependent handler status...")
+
+        for dep in self.depends_on:
+            key = kopf.AnnotationsProgressStorage(
+                v1=False, prefix=KOPF_STATE_STORE_PREFIX
+            ).make_v2_key(dep)
+
+            status_str = annotations.get(key)
+            logger.info(f"[{dep}] annotation raw value: {status_str}")
+            if not status_str:
+                continue
+
+            try:
+                parsed = json.loads(status_str)
+                if not parsed.get("success"):
+                    failed_handlers.append(dep)
+            except json.JSONDecodeError as e:
+                logger.warning(f"Failed to parse status annotation for {dep}: {e}")
+
+        logger.info(f"Failed handlers detected: {failed_handlers}")
+        return failed_handlers
+
 
 async def send_webhook_notification(
     namespace: str,