From be529147a41321c86bba40d10b194e3b40ed1ca7 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 2 Jul 2026 16:42:44 +0200 Subject: [PATCH] perf!: Fix quadratic file rescans in the file system request queue BREAKING CHANGE: The file system request queue no longer emits the 'Request not found in state, skipping.' warning. The cache refresh no longer scans the queue directory, so orphaned request files (e.g. written by another process) are silently ignored. --- .../_file_system/_request_queue_client.py | 125 ++++++++---------- .../_file_system/test_fs_rq_client.py | 96 ++++++++++++++ 2 files changed, 151 insertions(+), 70 deletions(-) diff --git a/src/crawlee/storage_clients/_file_system/_request_queue_client.py b/src/crawlee/storage_clients/_file_system/_request_queue_client.py index b0c37ecf22..a95f5331ef 100644 --- a/src/crawlee/storage_clients/_file_system/_request_queue_client.py +++ b/src/crawlee/storage_clients/_file_system/_request_queue_client.py @@ -1,7 +1,6 @@ from __future__ import annotations import asyncio -import functools import json import shutil from collections import deque @@ -333,7 +332,8 @@ async def add_batch_of_requests( unprocessed_requests = list[UnprocessedRequest]() state = self._state.current_value - all_requests = state.forefront_requests | state.regular_requests + def is_in_queue(unique_key: str) -> bool: + return unique_key in state.forefront_requests or unique_key in state.regular_requests requests_to_enqueue = {} @@ -352,7 +352,7 @@ async def add_batch_of_requests( # Or if the request is already in the queue and the `forefront` flag is not used, we do not change the # position of the request. elif (request.unique_key in state.in_progress_requests) or ( - request.unique_key in all_requests and not forefront + is_in_queue(request.unique_key) and not forefront ): processed_requests.append( ProcessedRequest( @@ -368,7 +368,7 @@ async def add_batch_of_requests( # Process each request in the batch. for request in requests_to_enqueue.values(): # If the request is not already in the RQ, this is a new request. - if request.unique_key not in all_requests: + if not is_in_queue(request.unique_key): request_path = self._get_request_path(request.unique_key) # Add sequence number to ensure FIFO ordering using state. if forefront: @@ -384,6 +384,16 @@ async def add_batch_of_requests( request_data = await json_dumps(request.model_dump()) await atomic_write(request_path, request_data) + # A new forefront request belongs to the very front of the queue, so it can go straight + # to the cache without a full refresh, unless the cache is at capacity. New regular + # requests must not be appended to the cache: it may be truncated, so its tail is not + # necessarily the end of the queue; they are picked up by a refresh once the cache drains. + if forefront: + if len(self._request_cache) < self._MAX_REQUESTS_IN_CACHE: + self._request_cache.appendleft(request) + else: + self._request_cache_needs_refresh = True + # Update the metadata counts. new_total_request_count += 1 new_pending_request_count += 1 @@ -406,6 +416,9 @@ async def add_batch_of_requests( state.forefront_requests[request.unique_key] = state.forefront_sequence_counter state.forefront_sequence_counter += 1 + # The request may already sit elsewhere in the cache, so its position must be recomputed. + self._request_cache_needs_refresh = True + processed_requests.append( ProcessedRequest( unique_key=request.unique_key, @@ -431,10 +444,6 @@ async def add_batch_of_requests( new_pending_request_count=new_pending_request_count, ) - # Invalidate the cache if we added forefront requests. - if forefront: - self._request_cache_needs_refresh = True - # Invalidate is_empty cache. self._is_empty_cache = None @@ -506,8 +515,12 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | request_data = await json_dumps(request.model_dump()) await atomic_write(request_path, request_data) - # Update state: remove from in-progress and add to handled. + # Update state: remove from in-progress and pending, and add to handled. Dropping the key from + # the pending mappings keeps them (and the persisted state) sized by the backlog rather than by + # every request ever processed. state.in_progress_requests.discard(request.unique_key) + state.forefront_requests.pop(request.unique_key, None) + state.regular_requests.pop(request.unique_key, None) state.handled_requests.add(request.unique_key) # Update RQ metadata. @@ -696,63 +709,36 @@ async def _update_metadata( await atomic_write(self.path_to_metadata, data) async def _refresh_cache(self) -> None: - """Refresh the request cache from filesystem. + """Refresh the request cache from the filesystem. - This method loads up to _MAX_REQUESTS_IN_CACHE requests from the filesystem, - prioritizing forefront requests and maintaining proper ordering. + This method loads up to `_MAX_REQUESTS_IN_CACHE` requests from the filesystem, prioritizing forefront + requests and maintaining proper ordering. Only requests that are pending according to the state are + read, so files of already handled or in-progress requests are not touched at all. """ self._request_cache.clear() state = self._state.current_value - forefront_requests = list[tuple[Request, int]]() # (request, sequence) - regular_requests = list[tuple[Request, int]]() # (request, sequence) - - request_files = await self._get_request_files(self.path_to_rq) - - for request_file in request_files: - request = await self._parse_request_file(request_file) - - if request is None: - continue - - # Skip handled requests - if request.unique_key in state.handled_requests: - continue - - # Skip in-progress requests - if request.unique_key in state.in_progress_requests: - continue - - # Determine if request is forefront or regular based on state - if request.unique_key in state.forefront_requests: - sequence = state.forefront_requests[request.unique_key] - forefront_requests.append((request, sequence)) - elif request.unique_key in state.regular_requests: - sequence = state.regular_requests[request.unique_key] - regular_requests.append((request, sequence)) - else: - # Request not in state, skip it (might be orphaned) - logger.warning(f'Request {request.unique_key} not found in state, skipping.') - continue - - # Sort forefront requests by sequence (newest first for LIFO behavior). - forefront_requests.sort(key=lambda item: item[1], reverse=True) + def is_pending(unique_key: str) -> bool: + return unique_key not in state.handled_requests and unique_key not in state.in_progress_requests - # Sort regular requests by sequence (oldest first for FIFO behavior). - regular_requests.sort(key=lambda item: item[1], reverse=False) + # Forefront requests are fetched newest first (LIFO), regular requests oldest first (FIFO). + forefront_keys = sorted( + filter(is_pending, state.forefront_requests), + key=lambda unique_key: state.forefront_requests[unique_key], + reverse=True, + ) + regular_keys = sorted( + filter(is_pending, state.regular_requests), + key=lambda unique_key: state.regular_requests[unique_key], + ) - # Add forefront requests to the beginning of the cache (left side). Since forefront_requests are sorted - # by sequence (newest first), we need to add them in reverse order to maintain correct priority. - for request, _ in reversed(forefront_requests): + for unique_key in forefront_keys + regular_keys: if len(self._request_cache) >= self._MAX_REQUESTS_IN_CACHE: break - self._request_cache.appendleft(request) - # Add regular requests to the end of the cache (right side). - for request, _ in regular_requests: - if len(self._request_cache) >= self._MAX_REQUESTS_IN_CACHE: - break - self._request_cache.append(request) + request = await self._parse_request_file(self._get_request_path(unique_key)) + if request is not None: + self._request_cache.append(request) self._request_cache_needs_refresh = False @@ -787,25 +773,23 @@ async def _parse_request_file(cls, file_path: Path) -> Request | None: Returns: The parsed `Request` object or `None` if the file could not be read or parsed. """ - # Open the request file. + # Read the request file in a thread to avoid blocking the event loop. try: - file = await asyncio.to_thread(functools.partial(file_path.open, mode='r', encoding='utf-8')) + file_content = await asyncio.to_thread(file_path.read_text, encoding='utf-8') except FileNotFoundError: logger.warning(f'Request file "{file_path}" not found.') return None - # Read the file content and parse it as JSON. + # Parse the file content as JSON. try: - file_content = json.load(file) + parsed = json.loads(file_content) except json.JSONDecodeError as exc: logger.warning(f'Failed to parse request file {file_path}: {exc!s}') return None - finally: - await asyncio.to_thread(file.close) # Validate the content against the Request model. try: - return Request.model_validate(file_content) + return Request.model_validate(parsed) except ValidationError as exc: logger.warning(f'Failed to validate request file {file_path}: {exc!s}') return None @@ -830,16 +814,17 @@ async def _discover_existing_requests(self) -> None: if request is None: continue - # Add request to state as regular request (assign sequence numbers) - if request.unique_key not in state.regular_requests and request.unique_key not in state.forefront_requests: - # Assign as regular request with current sequence counter + # Already handled requests are tracked only for deduplication, not as pending work. + if request.handled_at is not None: + state.handled_requests.add(request.unique_key) + + # Add pending request to state as regular request (assign sequence numbers) + elif ( + request.unique_key not in state.regular_requests and request.unique_key not in state.forefront_requests + ): state.regular_requests[request.unique_key] = state.sequence_counter state.sequence_counter += 1 - # Check if request was already handled - if request.handled_at is not None: - state.handled_requests.add(request.unique_key) - @staticmethod def _get_file_base_name_from_unique_key(unique_key: str) -> str: """Generate a deterministic file name for a unique_key. diff --git a/tests/unit/storage_clients/_file_system/test_fs_rq_client.py b/tests/unit/storage_clients/_file_system/test_fs_rq_client.py index a3641f60bf..2839cc2610 100644 --- a/tests/unit/storage_clients/_file_system/test_fs_rq_client.py +++ b/tests/unit/storage_clients/_file_system/test_fs_rq_client.py @@ -259,3 +259,99 @@ async def test_get_request_does_not_mark_in_progress(rq_client: FileSystemReques next_request = await rq_client.fetch_next_request() assert next_request is not None assert next_request.unique_key == request.unique_key + + +async def test_forefront_add_fetch_handle_parses_linear_number_of_files( + rq_client: FileSystemRequestQueueClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Test that per-request forefront adds (the `RequestManagerTandem` pattern) do not rescan all request files.""" + parse_count = 0 + original_parse = type(rq_client)._parse_request_file + + async def counting_parse(file_path: Path) -> Request | None: + nonlocal parse_count + parse_count += 1 + return await original_parse(file_path) + + monkeypatch.setattr(type(rq_client), '_parse_request_file', staticmethod(counting_parse)) + + n = 25 + for i in range(n): + await rq_client.add_batch_of_requests([Request.from_url(f'https://example.com/{i}')], forefront=True) + request = await rq_client.fetch_next_request() + assert request is not None + await rq_client.mark_request_as_handled(request) + + # Before the fix, each forefront add invalidated the cache and each fetch re-parsed every request file + # ever written, giving n * (n + 1) / 2 parses in total. With the fix, new forefront requests go straight + # to the cache, so only the initial refresh parses anything. + assert parse_count <= 5 + + +async def test_cache_refresh_skips_handled_request_files( + rq_client: FileSystemRequestQueueClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Test that a cache refresh reads only pending request files, not files of already handled requests.""" + await rq_client.add_batch_of_requests([Request.from_url(f'https://example.com/{i}') for i in range(5)]) + for _ in range(5): + request = await rq_client.fetch_next_request() + assert request is not None + await rq_client.mark_request_as_handled(request) + + # The 5 handled request files stay on disk. Add 5 new requests; fetching the next one refreshes + # the cache, which must not parse the handled files again. + await rq_client.add_batch_of_requests([Request.from_url(f'https://example.com/new/{i}') for i in range(5)]) + + parse_count = 0 + original_parse = type(rq_client)._parse_request_file + + async def counting_parse(file_path: Path) -> Request | None: + nonlocal parse_count + parse_count += 1 + return await original_parse(file_path) + + monkeypatch.setattr(type(rq_client), '_parse_request_file', staticmethod(counting_parse)) + + request = await rq_client.fetch_next_request() + assert request is not None + assert parse_count == 5 + + +async def test_handled_requests_pruned_from_pending_state(rq_client: FileSystemRequestQueueClient) -> None: + """Test that handling a request removes it from the pending state mappings so the state stays bounded.""" + await rq_client.add_batch_of_requests([Request.from_url(f'https://example.com/{i}') for i in range(3)]) + + handled_keys = set() + for _ in range(3): + request = await rq_client.fetch_next_request() + assert request is not None + await rq_client.mark_request_as_handled(request) + handled_keys.add(request.unique_key) + + state = rq_client._state.current_value + assert not state.regular_requests + assert not state.forefront_requests + assert not state.in_progress_requests + assert state.handled_requests == handled_keys + + +async def test_handled_requests_deduplicated_after_reopen() -> None: + """Test that requests handled before a reopen are not served again and still deduplicate re-adds.""" + storage_client = FileSystemStorageClient() + client = await storage_client.create_rq_client(name='handled-reopen-test') + + request = Request.from_url('https://example.com/handled') + await client.add_batch_of_requests([request]) + fetched = await client.fetch_next_request() + assert fetched is not None + await client.mark_request_as_handled(fetched) + await client._state.persist_state() + + rq_id = (await client.get_metadata()).id + reopened = await storage_client.create_rq_client(id=rq_id) + + response = await reopened.add_batch_of_requests([request]) + assert response.processed_requests[0].was_already_handled is True + assert await reopened.fetch_next_request() is None + + await reopened.drop()