Track pending_request_count in local metadata estimation

Author: Pijukatel

src/apify/storage_clients/_apify/_request_queue_shared_client.py

@@ -170,6 +170,7 @@ async def add_batch_of_requests(
                 new_request_count += 1
 
         self.metadata.total_request_count += new_request_count
+        self.metadata.pending_request_count += new_request_count
 
         return api_response
 
@@ -265,6 +266,7 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
             # 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
             cache_key = request.unique_key
@@ -312,6 +314,7 @@ async def reclaim_request(
                 # 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

src/apify/storage_clients/_apify/_request_queue_single_client.py

@@ -168,6 +168,7 @@ async def add_batch_of_requests(
             if not processed_request.was_already_present and not processed_request.was_already_handled:
                 new_request_count += 1
         self.metadata.total_request_count += new_request_count
+        self.metadata.pending_request_count += new_request_count
 
         return api_response
 
@@ -271,6 +272,7 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
         if request.handled_at is None:
             request.handled_at = datetime.now(tz=timezone.utc)
             self.metadata.handled_request_count += 1
+            self.metadata.pending_request_count -= 1
 
         if cached_request := self._requests_cache.get(request.unique_key):
             cached_request.handled_at = request.handled_at
@@ -333,6 +335,7 @@ async def reclaim_request(
             # 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}')