perf(file-system): Fix quadratic file rescans in the FS request queue

[vdusek]

Problem

The file system request queue client was O(n²):

• Every cache refresh globbed and parsed every request file ever written (handled files are never deleted from disk).
• forefront=True invalidated the whole cache, and RequestManagerTandem adds every loader request with forefront=True, which triggered a full rescan per processed request.
• The blocking json.load ran on the event loop.

Processing a 10k RequestList through the tandem meant roughly 50M file parses.

Changes

• _refresh_cache now derives pending requests from the recoverable state's key→sequence maps and reads only those files, instead of globbing and parsing the whole directory.
• New forefront requests go straight to the front of the in-memory cache. The cache is invalidated only when an existing request is repositioned, or when it's at capacity.
• Handled requests are pruned from the pending state maps (they stay in handled_requests for deduplication), so cache refreshes, is_empty, and the persisted state blob scale with the backlog rather than with every request ever processed.
• _parse_request_file reads the file in a single asyncio.to_thread call instead of running json.load on the event loop.

Results

Processing 200 requests through the tandem pattern went from 20,100 file parses (2.8 s) to 1 (0.3 s), and scaling is now linear (2,000 requests in about 2.6 s). Ordering semantics (forefront LIFO, regular FIFO), deduplication, and crash-recovery behavior are preserved and covered by new regression tests.

Logging changes

The file system request queue no longer emits the Request <unique_key> 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.

[codecov]

Codecov Report

❌ Patch coverage is 95.83333% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 93.44%. Comparing base (7780e78) to head (be52914).
⚠️ Report is 46 commits behind head on master.

Files with missing lines	Patch %	Lines
...rage_clients/_file_system/_request_queue_client.py	95.83%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #2011      +/-   ##
==========================================
+ Coverage   93.35%   93.44%   +0.09%     
==========================================
  Files         179      179              
  Lines       12482    12467      -15     
==========================================
- Hits        11652    11650       -2     
+ Misses        830      817      -13     

Flag	Coverage Δ
unit	93.44% <95.83%> (+0.09%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Mantisus]

LGTM. Only few suggestions.

[Mantisus]

Let's add a log warning when there is no file for the unique_key.

[Mantisus]

We can probably skip this if self._request_cache_needs_refresh is already True.

[Mantisus]

Maybe we could use unittest.mock.patch for this?

For example

from unittest.mock import patch

async def test_forefront_add_fetch_handle_parses_linear_number_of_files(
    rq_client: FileSystemRequestQueueClient,
) -> None:
    with patch.object(
        rq_client,
        '_parse_request_file',
        wraps=rq_client._parse_request_file,
    ) as call_counter:

        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)

    assert call_counter.call_count <= 5