fix(memory-storage): preserve request queue semantics

Replace lazy deque tombstones with an ordered mapping so forefront repositioning remains constant-time without stale-entry growth. Keep request updates on regular re-adds, update reclaimed request instances in the index, and leave is_empty side-effect free.

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

Author: anxkhn

src/crawlee/storage_clients/_memory/_request_queue_client.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections import deque
+from collections import OrderedDict
 from datetime import datetime, timezone
 from logging import getLogger
 from typing import TYPE_CHECKING
@@ -41,7 +41,7 @@ def __init__(
         """
         self._metadata = metadata
 
-        self._pending_requests = deque[Request]()
+        self._pending_requests = OrderedDict[str, Request]()
         """Pending requests are those that have been added to the queue but not yet fetched for processing."""
 
         self._handled_requests = dict[str, Request]()
@@ -174,23 +174,21 @@ async def add_batch_of_requests(
                 )
                 continue
 
-            # If the request is already in the queue but not handled, we only reposition it when `forefront`
-            # is set; a regular re-add leaves the already-pending entry untouched.
+            # If the request is already in the queue but not handled, update it without changing its position.
             if was_already_present and existing_request:
+                self._requests_by_unique_key[request.unique_key] = request
+                self._pending_requests[request.unique_key] = request
+
                 if forefront:
-                    # Move the request to the front. The old entry is left in the deque instead of being
-                    # located and removed (`deque.remove` is O(n), which makes a batch of forefront re-adds
-                    # O(n^2)); registering the new object here supersedes the old one, and the stale entry is
-                    # skipped lazily by `fetch_next_request` and `is_empty`.
-                    self._requests_by_unique_key[request.unique_key] = request
-                    self._pending_requests.appendleft(request)
+                    self._pending_requests.move_to_end(request.unique_key, last=False)
 
             # Add the new request to the queue.
             else:
                 if forefront:
-                    self._pending_requests.appendleft(request)
+                    self._pending_requests[request.unique_key] = request
+                    self._pending_requests.move_to_end(request.unique_key, last=False)
                 else:
-                    self._pending_requests.append(request)
+                    self._pending_requests[request.unique_key] = request
 
                 # Update indexes.
                 self._requests_by_unique_key[request.unique_key] = request
@@ -218,12 +216,7 @@ 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 stale entries left behind when a request was repositioned to the forefront while already
-            # pending. Only the object currently registered for the unique key is live.
-            if self._requests_by_unique_key.get(request.unique_key) is not request:
-                continue
+            _, request = self._pending_requests.popitem(last=False)
 
             # Skip if already handled (shouldn't happen, but safety check).
             if request.was_already_handled:
@@ -290,11 +283,13 @@ async def reclaim_request(
         # Remove from in-progress.
         del self._in_progress_requests[request.unique_key]
 
+        # Update index with the possibly modified request.
+        self._requests_by_unique_key[request.unique_key] = request
+
         # Add request back to pending queue.
+        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)
@@ -309,13 +304,6 @@ async def reclaim_request(
     async def is_empty(self) -> bool:
         await self._update_metadata(update_accessed_at=True)
 
-        # Discard stale entries left at the front by forefront repositioning; a live request is always
-        # enqueued ahead of the stale entry it supersedes, so pruning stops at the first live request.
-        while self._pending_requests and (
-            self._requests_by_unique_key.get(self._pending_requests[0].unique_key) is not self._pending_requests[0]
-        ):
-            self._pending_requests.popleft()
-
         # Queue is empty if there are no pending requests.
         return len(self._pending_requests) == 0
 

tests/unit/storage_clients/_memory/test_memory_rq_client.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-from collections import deque
 from typing import TYPE_CHECKING
 
 import pytest
@@ -96,26 +95,15 @@ async def test_memory_metadata_updates(rq_client: MemoryRequestQueueClient) -> N
     assert metadata.accessed_at > accessed_after_read
 
 
-async def test_forefront_readd_repositions_without_deque_scan(rq_client: MemoryRequestQueueClient) -> None:
-    """Test that re-adding pending requests to the forefront does not do an O(n) `deque.remove` scan."""
-
-    class CountingDeque(deque):  # type: ignore[type-arg]
-        remove_calls = 0
-
-        def remove(self, value: object) -> None:
-            CountingDeque.remove_calls += 1
-            super().remove(value)
-
-    rq_client._pending_requests = CountingDeque(rq_client._pending_requests)
-
+async def test_forefront_readd_does_not_grow_pending_requests(rq_client: MemoryRequestQueueClient) -> None:
+    """Test that repeatedly repositioning pending requests does not create duplicate entries."""
     requests = [Request.from_url(f'https://example.com/{i}') for i in range(20)]
     await rq_client.add_batch_of_requests(requests)
 
-    # Re-add the same, still-pending requests with `forefront=True`. Previously each re-add scanned the
-    # whole pending deque with `deque.remove` to reposition the existing entry, which is O(n) per request.
-    await rq_client.add_batch_of_requests(requests, forefront=True)
+    for _ in range(10):
+        await rq_client.add_batch_of_requests(requests, forefront=True)
 
-    assert CountingDeque.remove_calls == 0
+    assert len(rq_client._pending_requests) == len(requests)
 
 
 async def test_forefront_readd_preserves_order_and_dedup(rq_client: MemoryRequestQueueClient) -> None:
@@ -144,18 +132,14 @@ async def test_forefront_readd_preserves_order_and_dedup(rq_client: MemoryReques
 
 
 async def test_regular_readd_of_pending_request_is_not_dropped(rq_client: MemoryRequestQueueClient) -> None:
-    """Test that a regular (non-forefront) re-add of a still-pending request keeps it fetchable.
-
-    The lazy-tombstone skip in `fetch_next_request`/`is_empty` keys off object identity, so a regular re-add
-    must not repoint the registered object away from the entry still sitting in the pending deque, otherwise
-    the genuinely-live request would be treated as stale and silently dropped.
-    """
+    """Test that a regular re-add updates a still-pending request without dropping it."""
     original = Request.from_url('https://example.com/page')
     await rq_client.add_batch_of_requests([original])
 
     # Re-add the same URL while still pending, as a distinct object (as the higher-level API does when it
     # rebuilds requests). `forefront` defaults to False.
     duplicate = Request.from_url('https://example.com/page')
+    duplicate.user_data['version'] = 2
     assert duplicate is not original
     assert duplicate.unique_key == original.unique_key
     await rq_client.add_batch_of_requests([duplicate])
@@ -164,8 +148,8 @@ async def test_regular_readd_of_pending_request_is_not_dropped(rq_client: Memory
     assert await rq_client.is_empty() is False
 
     fetched = await rq_client.fetch_next_request()
-    assert fetched is not None
-    assert fetched.url == 'https://example.com/page'
+    assert fetched is duplicate
+    assert fetched.user_data['version'] == 2
     await rq_client.mark_request_as_handled(fetched)
 
     assert await rq_client.fetch_next_request() is None
@@ -196,3 +180,22 @@ async def test_regular_readd_does_not_reorder_pending_queue(rq_client: MemoryReq
         'https://example.com/1',
         'https://example.com/2',
     ]
+
+
+async def test_reclaim_modified_request_after_forefront_readd(rq_client: MemoryRequestQueueClient) -> None:
+    """Test reclaiming a modified request after it was repositioned to the forefront."""
+    request = Request.from_url('https://example.com/page')
+    await rq_client.add_batch_of_requests([request])
+    await rq_client.add_batch_of_requests([request], forefront=True)
+
+    fetched = await rq_client.fetch_next_request()
+    assert fetched is request
+
+    modified = request.model_copy(deep=True)
+    modified.user_data['reclaimed'] = True
+    await rq_client.reclaim_request(modified)
+
+    assert await rq_client.is_empty() is False
+    reclaimed = await rq_client.fetch_next_request()
+    assert reclaimed is modified
+    assert reclaimed.user_data['reclaimed'] is True