fix deadlock in KVS

Author: vdusek

src/crawlee/storage_clients/_file_system/_key_value_store_client.py

@@ -151,9 +151,7 @@ async def open(
 
         # Get a new instance by name.
         else:
-            kvs_path = (
-                kvs_base_path / cls._STORAGE_SUBSUBDIR_DEFAULT if name is None else kvs_base_path / name
-            )
+            kvs_path = kvs_base_path / cls._STORAGE_SUBSUBDIR_DEFAULT if name is None else kvs_base_path / name
             metadata_path = kvs_path / METADATA_FILENAME
 
             # If the key-value store directory exists, reconstruct the client from the metadata file.
@@ -235,7 +233,7 @@ async def get_value(self, *, key: str) -> KeyValueStoreRecord | None:
             return None
 
         # Read the metadata file
-        async with self._lock:
+        async with self._lock:  # DEADLOCK
             file = await asyncio.to_thread(open, record_metadata_filepath)
             try:
                 metadata_content = json.load(file)
@@ -364,41 +362,42 @@ async def iterate_keys(
         if not self.path_to_kvs.exists():
             return
 
-        count = 0
+        # List and sort all files *inside* a brief lock, then release it immediately:
         async with self._lock:
-            # Get all files in the KVS directory, sorted alphabetically
             files = sorted(await asyncio.to_thread(list, self.path_to_kvs.glob('*')))
 
-            for file_path in files:
-                # Skip the main metadata file
-                if file_path.name == METADATA_FILENAME:
-                    continue
+        count = 0
 
-                # Only process metadata files for records
-                if not file_path.name.endswith(f'.{METADATA_FILENAME}'):
-                    continue
+        for file_path in files:
+            # Skip the main metadata file
+            if file_path.name == METADATA_FILENAME:
+                continue
 
-                # Extract the base key name from the metadata filename
-                key_name = self._decode_key(file_path.name[: -len(f'.{METADATA_FILENAME}')])
+            # Only process metadata files for records
+            if not file_path.name.endswith(f'.{METADATA_FILENAME}'):
+                continue
 
-                # Apply exclusive_start_key filter if provided
-                if exclusive_start_key is not None and key_name <= exclusive_start_key:
-                    continue
+            # Extract the base key name from the metadata filename
+            key_name = self._decode_key(file_path.name[: -len(f'.{METADATA_FILENAME}')])
 
-                # Try to read and parse the metadata file
-                try:
-                    metadata_content = await asyncio.to_thread(file_path.read_text, encoding='utf-8')
-                    metadata_dict = json.loads(metadata_content)
-                    record_metadata = KeyValueStoreRecordMetadata(**metadata_dict)
+            # Apply exclusive_start_key filter if provided
+            if exclusive_start_key is not None and key_name <= exclusive_start_key:
+                continue
+
+            # Try to read and parse the metadata file
+            try:
+                metadata_content = await asyncio.to_thread(file_path.read_text, encoding='utf-8')
+                metadata_dict = json.loads(metadata_content)
+                record_metadata = KeyValueStoreRecordMetadata(**metadata_dict)
 
-                    yield record_metadata
+                yield record_metadata
 
-                    count += 1
-                    if limit and count >= limit:
-                        break
+                count += 1
+                if limit and count >= limit:
+                    break
 
-                except (json.JSONDecodeError, ValidationError) as e:
-                    logger.warning(f'Failed to parse metadata file {file_path}: {e}')
+            except (json.JSONDecodeError, ValidationError) as e:
+                logger.warning(f'Failed to parse metadata file {file_path}: {e}')
 
         # Update accessed_at timestamp
         await self._update_metadata(update_accessed_at=True)

src/crawlee/storage_clients/_file_system/_request_queue_client.py

@@ -171,9 +171,7 @@ async def open(
 
         # Get a new instance by name.
         else:
-            rq_path = (
-                rq_base_path / cls._STORAGE_SUBSUBDIR_DEFAULT if name is None else rq_base_path / name
-            )
+            rq_path = rq_base_path / cls._STORAGE_SUBSUBDIR_DEFAULT if name is None else rq_base_path / name
             metadata_path = rq_path / METADATA_FILENAME
 
             # If the RQ directory exists, reconstruct the client from the metadata file.

tests/unit/crawlers/_http/test_http_crawler.py

@@ -543,9 +543,9 @@ async def request_handler(context: HttpCrawlingContext) -> None:
     assert snapshot.html == HELLO_WORLD.decode('utf8')
 
 
-@pytest.mark.skip(reason='TODO: broken, freezing')
 async def test_error_snapshot_through_statistics(server_url: URL) -> None:
-    crawler = HttpCrawler(statistics=Statistics.with_default_state(save_error_snapshots=True))
+    statistics = Statistics.with_default_state(save_error_snapshots=True)
+    crawler = HttpCrawler(statistics=statistics)
 
     @crawler.router.default_handler
     async def request_handler(context: HttpCrawlingContext) -> None:

tests/unit/crawlers/_playwright/test_playwright_crawler.py

@@ -553,7 +553,7 @@ async def request_handler(context: PlaywrightCrawlingContext) -> None:
     assert snapshot.html == HELLO_WORLD.decode('utf-8')
 
 
-@pytest.mark.skip(reason='TODO: broken, freezing')
+# TODO: failing
 async def test_error_snapshot_through_statistics(server_url: URL) -> None:
     """Test correct use of error snapshotter by the Playwright crawler.