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,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]()
@@ -175,27 +174,19 @@ async def add_batch_of_requests(
                 )
                 continue
 
-            # If the request is already in the queue but not handled, update it.
+            # If the request is already in the queue but not handled, update it; forefront moves it to the front.
             if was_already_present and existing_request:
-                # Update indexes.
                 self._requests_by_unique_key[request.unique_key] = request
+                self._pending_requests[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)
+                    self._pending_requests.move_to_end(request.unique_key, last=False)
 
             # Add the new request to the queue.
             else:
+                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 indexes.
                 self._requests_by_unique_key[request.unique_key] = request
@@ -223,7 +214,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()
+            _, request = self._pending_requests.popitem(last=False)
 
             # Skip if already handled (shouldn't happen, but safety check).
             if request.was_already_handled:
@@ -290,11 +281,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)

tests/unit/storage_clients/_memory/test_memory_rq_client.py

@@ -93,3 +93,115 @@ async def test_memory_metadata_updates(rq_client: MemoryRequestQueueClient) -> N
     assert metadata.created_at == initial_created
     assert metadata.modified_at > initial_modified
     assert metadata.accessed_at > accessed_after_read
+
+
+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)
+
+    for _ in range(10):
+        await rq_client.add_batch_of_requests(requests, forefront=True)
+
+    fetched = []
+    while (request := await rq_client.fetch_next_request()) is not None:
+        fetched.append(request)
+        await rq_client.mark_request_as_handled(request)
+
+    assert len(fetched) == len(requests)
+    assert {request.unique_key for request in fetched} == {request.unique_key for request in requests}
+
+
+async def test_forefront_readd_preserves_order_and_dedup(rq_client: MemoryRequestQueueClient) -> None:
+    """Test that repositioning already-pending requests to the forefront keeps LIFO order and dedup."""
+    requests = [Request.from_url(f'https://example.com/{i}') for i in range(3)]
+    await rq_client.add_batch_of_requests(requests)
+
+    # Re-add a subset (0 and 1) to the forefront while still pending. Request 1 is added last, so it must
+    # end up at the very front, followed by request 0, then the untouched regular request 2.
+    await rq_client.add_batch_of_requests(requests[:2], forefront=True)
+
+    fetched_urls = []
+    while (request := await rq_client.fetch_next_request()) is not None:
+        fetched_urls.append(request.url)
+        await rq_client.mark_request_as_handled(request)
+
+    assert fetched_urls == [
+        'https://example.com/1',
+        'https://example.com/0',
+        'https://example.com/2',
+    ]
+
+    # No stale duplicates should linger after all live requests are drained.
+    assert await rq_client.is_empty() is True
+    assert await rq_client.is_finished() is True
+
+
+async def test_regular_readd_of_pending_request_is_not_dropped(rq_client: MemoryRequestQueueClient) -> None:
+    """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])
+
+    # The request must still be pending and fetchable exactly once, and the counts must stay consistent.
+    assert await rq_client.is_empty() is False
+
+    fetched = await rq_client.fetch_next_request()
+    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
+    assert await rq_client.is_empty() is True
+    assert await rq_client.is_finished() is True
+
+    metadata = await rq_client.get_metadata()
+    assert metadata.total_request_count == 1
+    assert metadata.pending_request_count == 0
+    assert metadata.handled_request_count == 1
+
+
+async def test_regular_readd_does_not_reorder_pending_queue(rq_client: MemoryRequestQueueClient) -> None:
+    """Test that a regular re-add of an already-pending request leaves the FIFO order untouched."""
+    requests = [Request.from_url(f'https://example.com/{i}') for i in range(3)]
+    await rq_client.add_batch_of_requests(requests)
+
+    # Re-add the first request (still pending) without `forefront`; it must stay in its original position.
+    await rq_client.add_batch_of_requests([requests[0]])
+
+    fetched_urls = []
+    while (request := await rq_client.fetch_next_request()) is not None:
+        fetched_urls.append(request.url)
+        await rq_client.mark_request_as_handled(request)
+
+    assert fetched_urls == [
+        'https://example.com/0',
+        '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