Revert "fix: stop silently swallowing exceptions in RQ mark_request_as_handled and reclaim_request"

This reverts commit 4bd1ca3.

Author: vdusek

src/apify/storage_clients/_apify/_request_queue_shared_client.py

@@ -218,24 +218,28 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
 
         if cached_request := self._requests_cache[request_id]:
             cached_request.was_already_handled = request.was_already_handled
-        # Update the request in the API
-        processed_request = await self._update_request(request)
-        processed_request.id = request_id
-        processed_request.unique_key = request.unique_key
-
-        # Update assumed handled count if this wasn't already handled
-        if not processed_request.was_already_handled:
-            self.metadata.handled_request_count += 1
-            self.metadata.pending_request_count -= 1
-
-        # Update the cache with the handled request
-        self._cache_request(
-            cache_key=request_id,
-            processed_request=processed_request,
-            hydrated_request=request,
-        )
+        try:
+            # Update the request in the API
+            processed_request = await self._update_request(request)
+            processed_request.id = request_id
+            processed_request.unique_key = request.unique_key
 
-        return processed_request
+            # Update assumed handled count if this wasn't already handled
+            if not processed_request.was_already_handled:
+                self.metadata.handled_request_count += 1
+                self.metadata.pending_request_count -= 1
+
+            # Update the cache with the handled request
+            self._cache_request(
+                cache_key=request_id,
+                processed_request=processed_request,
+                hydrated_request=request,
+            )
+        except Exception as exc:
+            logger.debug(f'Error marking request {request.unique_key} as handled: {exc!s}')
+            return None
+        else:
+            return processed_request
 
     async def reclaim_request(
         self,
@@ -251,30 +255,35 @@ async def reclaim_request(
 
         # Reclaim with lock to prevent race conditions that could lead to double processing of the same request.
         async with self._fetch_lock:
-            # Update the request in the API.
-            processed_request = await self._update_request(request, forefront=forefront)
-            processed_request.unique_key = request.unique_key
-
-            # If the request was previously handled, decrement our handled count since
-            # we're putting it back for processing.
-            if request.was_already_handled and not processed_request.was_already_handled:
-                self.metadata.handled_request_count -= 1
-                self.metadata.pending_request_count += 1
-
-            # Update the cache
-            cache_key = request.unique_key
-            self._cache_request(
-                cache_key,
-                processed_request,
-                hydrated_request=request,
-            )
+            try:
+                # Update the request in the API.
+                processed_request = await self._update_request(request, forefront=forefront)
+                processed_request.unique_key = request.unique_key
+
+                # If the request was previously handled, decrement our handled count since
+                # we're putting it back for processing.
+                if request.was_already_handled and not processed_request.was_already_handled:
+                    self.metadata.handled_request_count -= 1
+                    self.metadata.pending_request_count += 1
+
+                # Update the cache
+                cache_key = request.unique_key
+                self._cache_request(
+                    cache_key,
+                    processed_request,
+                    hydrated_request=request,
+                )
 
-            # If we're adding to the forefront, we need to check for forefront requests
-            # in the next list_head call
-            if forefront:
-                self._should_check_for_forefront_requests = True
+                # If we're adding to the forefront, we need to check for forefront requests
+                # in the next list_head call
+                if forefront:
+                    self._should_check_for_forefront_requests = True
 
-            return processed_request
+            except Exception as exc:
+                logger.debug(f'Error reclaiming request {request.unique_key}: {exc!s}')
+                return None
+            else:
+                return processed_request
 
     async def is_empty(self) -> bool:
         """Specific implementation of this method for the RQ shared access mode."""

src/apify/storage_clients/_apify/_request_queue_single_client.py

@@ -209,17 +209,22 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
         if cached_request := self._requests_cache.get(request_id):
             cached_request.handled_at = request.handled_at
 
-        # Update the request in the API
-        # Works as upsert - adds the request if it does not exist yet. (Local request that was handled before
-        # adding to the queue.)
-        processed_request = await self._update_request(request)
-        # Remember that we handled this request, to optimize local deduplication.
-        self._requests_already_handled.add(request_id)
-        # Remove request from cache. It will most likely not be needed.
-        self._requests_cache.pop(request_id)
-        self._requests_in_progress.discard(request_id)
-
-        return processed_request
+        try:
+            # Update the request in the API
+            # Works as upsert - adds the request if it does not exist yet. (Local request that was handled before
+            # adding to the queue.)
+            processed_request = await self._update_request(request)
+            # Remember that we handled this request, to optimize local deduplication.
+            self._requests_already_handled.add(request_id)
+            # Remove request from cache. It will most likely not be needed.
+            self._requests_cache.pop(request_id)
+            self._requests_in_progress.discard(request_id)
+
+        except Exception as exc:
+            logger.debug(f'Error marking request {request.unique_key} as handled: {exc!s}')
+            return None
+        else:
+            return processed_request
 
     async def reclaim_request(
         self,
@@ -236,28 +241,33 @@ async def reclaim_request(
         if request.was_already_handled:
             request.handled_at = None
 
-        # Make sure request is in the local cache. We might need it.
-        self._requests_cache[request_id] = request
-
-        # No longer in progress
-        self._requests_in_progress.discard(request_id)
-        # No longer handled
-        self._requests_already_handled.discard(request_id)
-
-        if forefront:
-            # Append to top of the local head estimation
-            self._head_requests.append(request_id)
-
-        processed_request = await self._update_request(request, forefront=forefront)
-        processed_request.id = request_id
-        processed_request.unique_key = request.unique_key
-        # If the request was previously handled, decrement our handled count since
-        # we're putting it back for processing.
-        if request.was_already_handled and not processed_request.was_already_handled:
-            self.metadata.handled_request_count -= 1
-            self.metadata.pending_request_count += 1
-
-        return processed_request
+        try:
+            # Make sure request is in the local cache. We might need it.
+            self._requests_cache[request_id] = request
+
+            # No longer in progress
+            self._requests_in_progress.discard(request_id)
+            # No longer handled
+            self._requests_already_handled.discard(request_id)
+
+            if forefront:
+                # Append to top of the local head estimation
+                self._head_requests.append(request_id)
+
+            processed_request = await self._update_request(request, forefront=forefront)
+            processed_request.id = request_id
+            processed_request.unique_key = request.unique_key
+            # If the request was previously handled, decrement our handled count since
+            # we're putting it back for processing.
+            if request.was_already_handled and not processed_request.was_already_handled:
+                self.metadata.handled_request_count -= 1
+                self.metadata.pending_request_count += 1
+
+        except Exception as exc:
+            logger.debug(f'Error reclaiming request {request.unique_key}: {exc!s}')
+            return None
+        else:
+            return processed_request
 
     async def is_empty(self) -> bool:
         """Specific implementation of this method for the RQ single access mode."""