fix(memory-storage): stop re-adds from duplicating and resetting pending requests

src/crawlee/storage_clients/_memory/_request_queue_client.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
-from collections import deque
-from contextlib import suppress
+from collections import OrderedDict
 from datetime import datetime, timezone
 from logging import getLogger
 from typing import TYPE_CHECKING
@@ -42,8 +41,12 @@ def __init__(
         """
         self._metadata = metadata
 
-        self._pending_requests = deque[Request]()
-        """Pending requests are those that have been added to the queue but not yet fetched for processing."""
+        self._pending_requests = OrderedDict[str, Request]()
+        """Pending requests are those that have been added to the queue but not yet fetched for processing.
+
+        Keyed by unique key and ordered from the front of the queue to its end, which keeps both fetching and
+        repositioning a request to the forefront O(1).
+        """
 
         self._handled_requests = dict[str, Request]()
         """Handled requests are those that have been processed and marked as handled."""
@@ -145,6 +148,9 @@ async def add_batch_of_requests(
         forefront: bool = False,
     ) -> AddRequestsResponse:
         processed_requests = []
+        new_total_request_count = self._metadata.total_request_count
+        new_pending_request_count = self._metadata.pending_request_count
+
         for request in requests:
             # Check if the request is already in the queue by unique_key.
             existing_request = self._requests_by_unique_key.get(request.unique_key)
@@ -175,35 +181,18 @@ async def add_batch_of_requests(
                 )
                 continue
 
-            # If the request is already in the queue but not handled, update it.
-            if was_already_present and existing_request:
-                # Update indexes.
-                self._requests_by_unique_key[request.unique_key] = request
-
-                # We only update `forefront` by updating its position by shifting it to the left.
-                if forefront:
-                    # Update the existing request with any new data and
-                    # remove old request from pending queue if it's there.
-                    with suppress(ValueError):
-                        self._pending_requests.remove(existing_request)
-
-                    # Add updated request back to queue.
-                    self._pending_requests.appendleft(request)
-
-            # Add the new request to the queue.
-            else:
-                if forefront:
-                    self._pending_requests.appendleft(request)
-                else:
-                    self._pending_requests.append(request)
-
-                # Update indexes.
+            # A new request is registered and appended to the end of the queue. A re-add of a still-pending
+            # request keeps the originally enqueued object: the incoming duplicate is typically a freshly built
+            # one that lost the state accumulated so far (e.g. `retry_count`).
+            if not was_already_present:
                 self._requests_by_unique_key[request.unique_key] = request
+                self._pending_requests[request.unique_key] = request
+                new_total_request_count += 1
+                new_pending_request_count += 1
 
-                await self._update_metadata(
-                    new_total_request_count=self._metadata.total_request_count + 1,
-                    new_pending_request_count=self._metadata.pending_request_count + 1,
-                )
+            # The only effect a re-add may have is repositioning the request to the front of the queue.
+            if forefront:
+                self._pending_requests.move_to_end(request.unique_key, last=False)
 
             processed_requests.append(
                 ProcessedRequest(
@@ -213,7 +202,12 @@ async def add_batch_of_requests(
                 )
             )
 
-        await self._update_metadata(update_accessed_at=True, update_modified_at=True)
+        await self._update_metadata(
+            update_accessed_at=True,
+            update_modified_at=True,
+            new_total_request_count=new_total_request_count,
+            new_pending_request_count=new_pending_request_count,
+        )
 
         return AddRequestsResponse(
             processed_requests=processed_requests,
@@ -222,22 +216,14 @@ async def add_batch_of_requests(
 
     @override
     async def fetch_next_request(self) -> Request | None:
-        while self._pending_requests:
-            request = self._pending_requests.popleft()
-
-            # Skip if already handled (shouldn't happen, but safety check).
-            if request.was_already_handled:
-                continue
-
-            # Skip if already in progress (shouldn't happen, but safety check).
-            if request.unique_key in self._in_progress_requests:
-                continue
+        if not self._pending_requests:
+            return None
 
-            # Mark as in progress.
-            self._in_progress_requests[request.unique_key] = request
-            return request
+        _, request = self._pending_requests.popitem(last=False)
 
-        return None
+        # Mark as in progress.
+        self._in_progress_requests[request.unique_key] = request
+        return request
 
     @override
     async def get_request(self, unique_key: str) -> Request | None:
@@ -290,11 +276,12 @@ async def reclaim_request(
         # Remove from in-progress.
         del self._in_progress_requests[request.unique_key]
 
-        # Add request back to pending queue.
+        # Add the request back to the pending queue. Unlike a re-add, a reclaim carries the state accumulated
+        # while the request was in progress, so both stores are updated with the reclaimed object.
+        self._requests_by_unique_key[request.unique_key] = request
+        self._pending_requests[request.unique_key] = request
         if forefront:
-            self._pending_requests.appendleft(request)
-        else:
-            self._pending_requests.append(request)
+            self._pending_requests.move_to_end(request.unique_key, last=False)
 
         # Update metadata timestamps.
         await self._update_metadata(update_modified_at=True)