Fixes to RequestQueue (#82)

This fixes three bugs in `RequestQueue`:
- when loading a queue from disk, `orderNo`s were not correctly
converted back to `Decimal`s
- when adding requests to the queue, even handled requests were added to
the queue head
- when opening an already populated queue, if you first added another
request to it before you called `fetch_next_request()`, the queue head
would not contain the requests that were in the queue before you opened
it

It also adds an E2E test which catches all three of these things.

Author: fnesveda

CHANGELOG.md

@@ -8,6 +8,10 @@ Changelog
 
 - started triggering base Docker image builds when releasing a new version
 
+### Fixed
+
+- fixed `RequestQueue` not loading requests from an existing queue properly
+
 [0.2.0](../../releases/tag/v0.2.0) - 2023-03-06
 -----------------------------------------------
 

src/apify/_memory_storage/resource_clients/request_queue.py

@@ -437,6 +437,8 @@ def _create_from_directory(
 
                 with open(os.path.join(storage_directory, entry.name)) as f:
                     request = json.load(f)
+                    if request.get('orderNo'):
+                        request['orderNo'] = Decimal(request.get('orderNo'))
                 entries.append(request)
 
         new_client = cls(

src/apify/storages/request_queue.py

@@ -166,7 +166,8 @@ async def add_request(self, request: Dict, *, forefront: bool = False) -> Dict:
         self._cache_request(cache_key, queue_operation_info)
 
         request_id, was_already_present = queue_operation_info['requestId'], queue_operation_info['wasAlreadyPresent']
-        if not was_already_present and request_id not in self._in_progress and self._recently_handled.get(request_id) is None:
+        is_handled = request.get('handledAt') is not None
+        if not is_handled and not was_already_present and request_id not in self._in_progress and self._recently_handled.get(request_id) is None:
             self._assumed_total_count += 1
 
             self._maybe_add_request_to_queue_head(request_id, forefront)
@@ -520,4 +521,6 @@ async def open(
         Returns:
             RequestQueue: An instance of the `RequestQueue` class for the given ID or name.
         """
-        return await super().open(id=id, name=name, force_cloud=force_cloud, config=config)
+        queue = await super().open(id=id, name=name, force_cloud=force_cloud, config=config)
+        await queue._ensure_head_is_non_empty()
+        return queue

tests/unit/actor/test_actor_memory_storage_e2e.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from typing import Callable
 
 import pytest
@@ -8,7 +9,7 @@
 
 
 @pytest.mark.parametrize('purge_on_start', [True, False])
-async def test_actor_memory_storage_client_e2e(
+async def test_actor_memory_storage_client_key_value_store_e2e(
     monkeypatch: pytest.MonkeyPatch,
     purge_on_start: bool,
     reset_default_instances: Callable[[], None],
@@ -17,7 +18,7 @@ async def test_actor_memory_storage_client_e2e(
     The second run attempts to access data created by the first one.
     We run 2 configurations with different `purge_on_start`."""
     # Configure purging env var
-    monkeypatch.setenv(ApifyEnvVars.PURGE_ON_START, 'true' if purge_on_start else 'false')
+    monkeypatch.setenv(ApifyEnvVars.PURGE_ON_START, f'{int(purge_on_start)}')
     # Store old storage client so we have the object reference for comparison
     old_client = StorageClientManager.get_storage_client()
     async with Actor:
@@ -45,3 +46,77 @@ async def test_actor_memory_storage_client_e2e(
         else:
             assert default_value == 'default value'
         assert non_default_value == 'non-default value'
+
+
+@pytest.mark.parametrize('purge_on_start', [True, False])
+async def test_actor_memory_storage_client_request_queue_e2e(
+    monkeypatch: pytest.MonkeyPatch,
+    purge_on_start: bool,
+    reset_default_instances: Callable[[], None],
+) -> None:
+    """This test simulates two clean runs using memory storage.
+    The second run attempts to access data created by the first one.
+    We run 2 configurations with different `purge_on_start`."""
+    # Configure purging env var
+    monkeypatch.setenv(ApifyEnvVars.PURGE_ON_START, f'{int(purge_on_start)}')
+    async with Actor:
+        # Add some requests to the default queue
+        default_queue = await Actor.open_request_queue()
+        for i in range(6):
+            request_url = f'http://example.com/{i}'
+            forefront = i % 3 == 1
+            was_handled = i % 3 == 2
+            await default_queue.add_request({
+                'uniqueKey': str(i),
+                'url': request_url,
+                'handledAt': datetime.now(timezone.utc) if was_handled else None,
+            }, forefront=forefront)
+
+    # We simulate another clean run, we expect the memory storage to read from the local data directory
+    # Default storages are purged based on purge_on_start parameter.
+    reset_default_instances()
+
+    async with Actor:
+        # Add some more requests to the default queue
+        default_queue = await Actor.open_request_queue()
+        for i in range(6, 12):
+            request_url = f'http://example.com/{i}'
+            forefront = i % 3 == 1
+            was_handled = i % 3 == 2
+            await default_queue.add_request({
+                'uniqueKey': str(i),
+                'url': request_url,
+                'handledAt': datetime.now(timezone.utc) if was_handled else None,
+            }, forefront=forefront)
+
+        queue_info = await default_queue.get_info()
+        assert queue_info is not None
+
+        # If the queue was purged between the runs, only the requests from the second run should be present, in the right order
+        if purge_on_start:
+            assert queue_info.get('totalRequestCount') == 6
+            assert queue_info.get('handledRequestCount') == 2
+
+            expected_pending_request_order = [10, 7, 6, 9]
+            for request_number in expected_pending_request_order:
+                next_request = await default_queue.fetch_next_request()
+                assert next_request is not None
+                assert next_request.get('uniqueKey') == f'{request_number}'
+                assert next_request.get('url') == f'http://example.com/{request_number}'
+
+            next_request = await default_queue.fetch_next_request()
+            assert next_request is None
+        # If the queue was NOT purged between the runs, all the requests should be in the queue in the right order
+        else:
+            assert queue_info.get('totalRequestCount') == 12
+            assert queue_info.get('handledRequestCount') == 4
+
+            expected_pending_request_order = [10, 7, 4, 1, 0, 3, 6, 9]
+            for request_number in expected_pending_request_order:
+                next_request = await default_queue.fetch_next_request()
+                assert next_request is not None
+                assert next_request.get('uniqueKey') == f'{request_number}'
+                assert next_request.get('url') == f'http://example.com/{request_number}'
+
+            next_request = await default_queue.fetch_next_request()
+            assert next_request is None