fix: stop silently swallowing exceptions in RQ mark_request_as_handled and reclaim_request

Both the single and shared request queue clients caught all exceptions
in `mark_request_as_handled` and `reclaim_request`, logged them at
DEBUG level, and returned `None`. This made it impossible for callers
to distinguish between "request was not in progress" (legitimate None
per the base class contract) and "API call failed" (swallowed error).

Silently swallowing errors also meant metadata counters could drift
from platform state without any visible indication of the problem.

Remove the catch-all `except Exception` blocks and let exceptions
propagate naturally so callers can handle failures appropriately.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: vdusek

src/apify/storage_clients/_apify/_request_queue_shared_client.py

@@ -218,28 +218,24 @@ 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
-        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
-
-            # 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 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,
+        )
 
-            # 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
+        return processed_request
 
     async def reclaim_request(
         self,
@@ -255,35 +251,30 @@ 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:
-            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,
-                )
+            # Update the request in the API.
+            processed_request = await self._update_request(request, forefront=forefront)
+            processed_request.unique_key = request.unique_key
 
-                # 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 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
+            # 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
+
+            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,22 +209,17 @@ 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
 
-        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
+        # 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
 
     async def reclaim_request(
         self,
@@ -241,33 +236,28 @@ async def reclaim_request(
         if request.was_already_handled:
             request.handled_at = None
 
-        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
+        # 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
 
     async def is_empty(self) -> bool:
         """Specific implementation of this method for the RQ single access mode."""