refactor: Improve logging and retry handling in task management

Author: drazisil-codecov

apps/worker/services/lock_manager.py

@@ -120,12 +120,12 @@ def locked(self, lock_type: LockType, retry_num=0, max_retries: int | None = Non
                 log.error(
                     "Not retrying since we already had too many retries",
                     extra={
-                        "repoid": self.repoid,
                         "commitid": self.commitid,
                         "lock_name": lock_name,
                         "max_retries": max_retries,
-                        "retry_num": retry_num,
+                        "repoid": self.repoid,
                         "report_type": self.report_type.value,
+                        "retry_num": retry_num,
                     },
                     exc_info=True,
                 )
@@ -153,12 +153,12 @@ def locked(self, lock_type: LockType, retry_num=0, max_retries: int | None = Non
             log.warning(
                 "Unable to acquire lock",
                 extra={
-                    "repoid": self.repoid,
                     "commitid": self.commitid,
-                    "lock_name": lock_name,
                     "countdown": countdown,
-                    "retry_num": retry_num,
+                    "lock_name": lock_name,
                     "max_retries": max_retries,
+                    "repoid": self.repoid,
+                    "retry_num": retry_num,
                 },
             )
             raise LockRetry(countdown)

apps/worker/tasks/base.py

@@ -238,32 +238,62 @@ def _max_total_attempts(self, max_retries: int | None) -> int | None:
         """Convert max retries to max total attempts (includes initial attempt)."""
         return max_retries + 1 if max_retries is not None else None
 
-    def _get_total_attempts(self) -> int:
-        """Get total attempts including re-deliveries from visibility timeout expiration."""
+    def _get_retry_count(self) -> int:
+        """Get retry count safely, handling missing request or retries attribute."""
         if not hasattr(self, "request"):
             return 0
+        return self.request.retries if hasattr(self.request, "retries") else 0
+
+    def _get_request_headers(self) -> dict:
+        """Get request headers safely, handling missing request or get method."""
+        if not hasattr(self, "request") or self.request is None:
+            return {}
+        request = self.request
+        try:
+            if hasattr(request, "get") and callable(getattr(request, "get", None)):
+                return request.get("headers", {})
+        except (AttributeError, TypeError):
+            pass
+        return getattr(request, "headers", {})
+
+    def _expected_total_attempts(self) -> int:
+        """Get expected total attempts based on retry count (retries + initial attempt)."""
+        return self._get_retry_count() + 1
+
+    def _get_total_attempts(self) -> int:
+        """Get total attempts including re-deliveries from visibility timeout expiration.
+
+        Returns:
+            - Header value if present and valid (most accurate)
+            - retry_count + 1 if header missing/invalid (best guess based on retry count)
+            - 0 if request unavailable (rare, safe default for comparisons)
+
+        Returns int (not None) to be safe for comparisons and logging without null checks.
+        """
+        if not hasattr(self, "request") or self.request is None:
+            return 0
 
-        retry_count = self.request.retries if hasattr(self.request, "retries") else 0
-        headers = self.request.get("headers", {}) or {}
+        headers = self._get_request_headers()
         total_attempts_header = headers.get("total_attempts", None)
 
         if total_attempts_header is not None:
             try:
                 return int(total_attempts_header)
             except (ValueError, TypeError):
+                retry_count = self._get_retry_count()
                 log.warning(
                     "Invalid total_attempts header value",
                     extra={"value": total_attempts_header, "retry_count": retry_count},
                 )
-                return retry_count + 1
-        return retry_count + 1
+                return self._expected_total_attempts()
+        return self._expected_total_attempts()
 
     def _has_exceeded_max_attempts(self, max_retries: int | None) -> bool:
         """Check if task has exceeded max attempts (including re-deliveries)."""
         if max_retries is None:
             return False
 
-        current_retries = self.request.retries if hasattr(self, "request") else 0
+        current_retries = self._get_retry_count()
         total_attempts = self._get_total_attempts()
         max_total_attempts = self._max_total_attempts(max_retries)
 
@@ -279,31 +309,32 @@ def safe_retry(self, max_retries=None, countdown=None, exc=None, **kwargs):
         task_max_retries = max_retries if max_retries is not None else self.max_retries
 
         if self._has_exceeded_max_attempts(task_max_retries):
-            current_retries = self.request.retries if hasattr(self, "request") else 0
+            current_retries = self._get_retry_count()
             total_attempts = self._get_total_attempts()
             max_total_attempts = self._max_total_attempts(task_max_retries)
             log.error(
                 f"Task {self.name} exceeded max retries",
                 extra={
-                    "task_name": self.name,
                     "current_retries": current_retries,
-                    "total_attempts": total_attempts,
                     "max_retries": task_max_retries,
                     "max_total_attempts": max_total_attempts,
+                    "task_name": self.name,
+                    "total_attempts": total_attempts,
                 },
             )
             TASK_MAX_RETRIES_EXCEEDED_COUNTER.labels(task=self.name).inc()
             return False
 
-        current_retries = self.request.retries if hasattr(self, "request") else 0
+        current_retries = self._get_retry_count()
         if countdown is None:
             countdown = TASK_RETRY_BACKOFF_BASE_SECONDS * (2**current_retries)
 
         try:
             total_attempts = self._get_total_attempts()
             headers = {}
-            if hasattr(self, "request") and hasattr(self.request, "headers"):
-                headers.update(self.request.headers or {})
+            request_headers = self._get_request_headers()
+            if request_headers:
+                headers.update(request_headers)
             headers.update(kwargs.get("headers", {}) or {})
             headers["total_attempts"] = total_attempts + 1
             kwargs["headers"] = headers
@@ -351,13 +382,23 @@ def _analyse_error(self, exception: SQLAlchemyError, *args, **kwargs):
         )
 
     def _emit_queue_metrics(self):
-        created_timestamp = self.request.get("created_timestamp", None)
+        if not hasattr(self, "request") or self.request is None:
+            return
+        request = self.request
+        if not hasattr(request, "get"):
+            return
+        created_timestamp = request.get("created_timestamp", None)
         if created_timestamp:
             enqueued_time = datetime.fromisoformat(created_timestamp)
             now = datetime.now()
             delta = now - enqueued_time
 
-            queue_name = self.request.get("delivery_info", {}).get("routing_key", None)
+            delivery_info = request.get("delivery_info", {})
+            queue_name = (
+                delivery_info.get("routing_key", None)
+                if isinstance(delivery_info, dict)
+                else None
+            )
             time_in_queue_timer = TASK_TIME_IN_QUEUE.labels(
                 task=self.name, queue=queue_name
             )  # TODO is None a valid label value
@@ -377,36 +418,41 @@ def run(self, *args, **kwargs):
             task = get_current_task()
             if task and task.request:
                 log_context.task_name = task.name
-                log_context.task_id = task.request.id
-
-                # Track total attempts including re-deliveries from visibility timeout
-                # If this is a re-delivery (visibility timeout expired), log it
-                # Note: We can't modify headers here (they're read-only), but we can detect
-                # re-deliveries by checking if total_attempts is ahead of retry_count + 1
-                # (which indicates Redis re-queued the task without incrementing retry_count)
-                headers = task.request.get("headers", {}) or {}
-                total_attempts = headers.get("total_attempts", None)
-                retry_count = (
-                    task.request.retries if hasattr(task.request, "retries") else 0
-                )
-
-                if total_attempts is not None:
-                    try:
-                        total_attempts_int = int(total_attempts)
-                        if total_attempts_int > retry_count + 1:
-                            log.warning(
-                                f"Task {task.name} re-delivered (visibility timeout expired)",
+                task_id = getattr(task.request, "id", None)
+                if task_id:
+                    log_context.task_id = task_id
+
+                try:
+                    headers = self._get_request_headers()
+                    header_total_attempts = headers.get("total_attempts")
+                    if header_total_attempts is not None:
+                        try:
+                            total_attempts = int(header_total_attempts)
+                            retry_count = (
+                                getattr(task.request, "retries", 0)
+                                if hasattr(task.request, "retries")
+                                else 0
+                            )
+                            if total_attempts > retry_count + 1:
+                                log.warning(
+                                    f"Task {task.name} re-delivered (visibility timeout expired)",
+                                    extra={
+                                        "retry_count": retry_count,
+                                        "task_id": task_id,
+                                        "total_attempts": total_attempts,
+                                    },
+                                )
+                        except (ValueError, TypeError):
+                            log.debug(
+                                "Invalid total_attempts header, skipping re-delivery detection",
                                 extra={
-                                    "task_id": task.request.id,
-                                    "retry_count": retry_count,
-                                    "total_attempts": total_attempts_int,
+                                    "task_id": task_id,
+                                    "value": header_total_attempts,
                                 },
                             )
-                    except (ValueError, TypeError):
-                        log.debug(
-                            "Invalid total_attempts header, skipping re-delivery detection",
-                            extra={"task_id": task.request.id, "value": total_attempts},
-                        )
+                except Exception:
+                    # Silently ignore errors accessing request headers to avoid breaking task execution
+                    pass
 
             log_context.populate_from_sqlalchemy(db_session)
             set_log_context(log_context)
@@ -485,10 +531,8 @@ def wrap_up_dbsession(self, db_session):
     def on_retry(self, exc, task_id, args, kwargs, einfo):
         res = super().on_retry(exc, task_id, args, kwargs, einfo)
         self.task_retry_counter.inc()
-        # Track retry count for better observability
-        retry_count = self.request.retries if hasattr(self, "request") else 0
         TASK_RETRY_WITH_COUNT_COUNTER.labels(
-            task=self.name, retry_count=str(retry_count)
+            task=self.name, retry_count=str(self._get_retry_count())
         ).inc()
         return res
 

apps/worker/tasks/bundle_analysis_processor.py

@@ -223,8 +223,8 @@ def process_impl_within_lock(
                 log.exception(
                     "Failed to commit upload error state",
                     extra={
-                        "repoid": repoid,
                         "commit": commitid,
+                        "repoid": repoid,
                         "upload_id": upload.id_,
                     },
                 )

apps/worker/tasks/tests/unit/test_base.py

@@ -564,6 +564,9 @@ def test_safe_retry_fails_at_max_retries(self, mocker):
         task = SampleTask()
         task.request.retries = 10
         task.max_retries = 10
+        # Set up request.get() and headers for _get_request_headers()
+        task.request.get = lambda key, default=None: {} if key == "headers" else default
+        task.request.headers = {}
 
         mock_retry = mocker.patch.object(task, "retry")
 
@@ -836,7 +839,6 @@ class SampleTask(BaseCodecovTask, name="test.SampleTask"):
             def run_impl(self, dbsession):
                 return "success"
 
-        task = SampleTask()
         mock_current_task = MagicMock()
         mock_current_task.name = "test.SampleTask"
         mock_request = MagicMock()
@@ -848,6 +850,11 @@ def run_impl(self, dbsession):
         )
         mock_request.headers = {"total_attempts": 5}
         mock_current_task.request = mock_request
+        # Set up self.request on task instance using PropertyMock (request is a property)
+        # Must patch BEFORE creating task instance
+        mock_property = PropertyMock(return_value=mock_request)
+        mocker.patch.object(SampleTask, "request", mock_property, create=True)
+        task = SampleTask()
 
         mocker.patch("tasks.base.get_current_task", return_value=mock_current_task)
         mock_log_warning = mocker.patch("tasks.base.log.warning")
@@ -903,7 +910,6 @@ class SampleTask(BaseCodecovTask, name="test.SampleTask"):
             def run_impl(self, dbsession):
                 return "success"
 
-        task = SampleTask()
         mock_current_task = MagicMock()
         mock_current_task.name = "test.SampleTask"
         mock_request = MagicMock()
@@ -915,6 +921,11 @@ def run_impl(self, dbsession):
         )  # Invalid value
         mock_request.headers = {"total_attempts": "invalid"}
         mock_current_task.request = mock_request
+        # Set up self.request on task instance using PropertyMock (request is a property)
+        # Must patch BEFORE creating task instance
+        mock_property = PropertyMock(return_value=mock_request)
+        mocker.patch.object(SampleTask, "request", mock_property, create=True)
+        task = SampleTask()
 
         mocker.patch("tasks.base.get_current_task", return_value=mock_current_task)
         mock_log_warning = mocker.patch("tasks.base.log.warning")
@@ -972,7 +983,6 @@ class SampleTask(BaseCodecovTask, name="test.SampleTask"):
             def run_impl(self, dbsession):
                 return "success"
 
-        task = SampleTask()
         mock_current_task = MagicMock()
         mock_current_task.name = "test.SampleTask"
         mock_request = MagicMock()
@@ -984,6 +994,11 @@ def run_impl(self, dbsession):
         )
         mock_request.headers = {"total_attempts": None}
         mock_current_task.request = mock_request
+        # Set up self.request on task instance using PropertyMock (request is a property)
+        # Must patch BEFORE creating task instance
+        mock_property = PropertyMock(return_value=mock_request)
+        mocker.patch.object(SampleTask, "request", mock_property, create=True)
+        task = SampleTask()
 
         mocker.patch("tasks.base.get_current_task", return_value=mock_current_task)
         mock_log_warning = mocker.patch("tasks.base.log.warning")

apps/worker/tasks/tests/unit/test_upload_finisher_task.py

@@ -831,6 +831,9 @@ def test_retry_on_report_lock(self, dbsession, mocker, mock_redis, mock_self_app
 
         task = UploadFinisherTask()
         task.request.retries = 0
+        # Set up request.get() and headers for _get_request_headers()
+        task.request.get = lambda key, default=None: {} if key == "headers" else default
+        task.request.headers = {}
 
         with pytest.raises(Retry):
             task.run_impl(
@@ -903,6 +906,9 @@ def test_die_on_finisher_lock(
 
         task = UploadFinisherTask()
         task.request.retries = 0
+        # Set up request.get() and headers for _get_request_headers()
+        task.request.get = lambda key, default=None: {} if key == "headers" else default
+        task.request.headers = {}
 
         # Task should call self.retry() which raises Retry exception
         with pytest.raises(Retry):

apps/worker/tasks/upload_finisher.py

@@ -471,12 +471,12 @@ def _process_reports_with_lock(
                 log.error(
                     "Upload finisher exceeded max retries",
                     extra={
-                        "repoid": repoid,
                         "commitid": commitid,
-                        "retry_count": self.request.retries,
-                        "total_attempts": total_attempts,
                         "max_retries": MAX_RETRIES,
                         "max_total_attempts": max_total_attempts,
+                        "repoid": repoid,
+                        "retry_count": self.request.retries,
+                        "total_attempts": total_attempts,
                     },
                 )
                 self._call_upload_breadcrumb_task(
@@ -499,8 +499,8 @@ def _process_reports_with_lock(
                 log.error(
                     "Failed to schedule retry for upload finisher",
                     extra={
-                        "repoid": repoid,
                         "commitid": commitid,
+                        "repoid": repoid,
                         "retry_count": self.request.retries,
                         "total_attempts": total_attempts,
                     },
@@ -601,12 +601,12 @@ def _handle_finisher_lock(
                 log.error(
                     "Upload finisher exceeded max retries (finisher lock)",
                     extra={
-                        "repoid": repoid,
                         "commitid": commitid,
-                        "retry_count": self.request.retries,
-                        "total_attempts": total_attempts,
                         "max_retries": MAX_RETRIES,
                         "max_total_attempts": max_total_attempts,
+                        "repoid": repoid,
+                        "retry_count": self.request.retries,
+                        "total_attempts": total_attempts,
                     },
                 )
                 self._call_upload_breadcrumb_task(
@@ -629,8 +629,8 @@ def _handle_finisher_lock(
                 log.error(
                     "Failed to schedule retry for upload finisher (finisher lock)",
                     extra={
-                        "repoid": repoid,
                         "commitid": commitid,
+                        "repoid": repoid,
                         "retry_count": self.request.retries,
                         "total_attempts": total_attempts,
                     },