fix: fix handling of loading empty metadata file for queue (#1042)

### Description

- Error handling if an empty metadata file was created

### Issues

- Closes: #1029

Author: Mantisus

src/crawlee/storage_clients/_memory/_creation_management.py

@@ -409,9 +409,16 @@ def _determine_storage_path(
                 metadata_path = os.path.join(entry.path, METADATA_FILENAME)
                 if os.access(metadata_path, os.F_OK):
                     with open(metadata_path, encoding='utf-8') as metadata_file:
-                        metadata = json.load(metadata_file)
-                    if (id and metadata.get('id') == id) or (name and metadata.get('name') == name):
-                        return entry.path
+                        try:
+                            metadata = json.load(metadata_file)
+                            if (id and metadata.get('id') == id) or (name and metadata.get('name') == name):
+                                return entry.path
+                        except Exception:
+                            logger.warning(
+                                f'Metadata of store entry "{entry.name}" for store {name or id} could not be parsed. '
+                                'The metadata file will be ignored.',
+                                exc_info=True,
+                            )
 
     # Check for default storage directory as a last resort
     if id == default_id:

src/crawlee/storage_clients/_memory/_key_value_store_client.py

@@ -334,13 +334,13 @@ async def persist_record(self, record: KeyValueStoreRecord) -> None:
             await asyncio.to_thread(f.close)
 
         if self._memory_storage_client.write_metadata:
-            f = await asyncio.to_thread(open, record_metadata_path, mode='wb')
+            metadata_f = await asyncio.to_thread(open, record_metadata_path, mode='wb')
 
             try:
                 record_metadata = KeyValueStoreRecordMetadata(key=record.key, content_type=record.content_type)
-                await asyncio.to_thread(f.write, record_metadata.model_dump_json(indent=2).encode('utf-8'))
+                await asyncio.to_thread(metadata_f.write, record_metadata.model_dump_json(indent=2).encode('utf-8'))
             finally:
-                await asyncio.to_thread(f.close)
+                await asyncio.to_thread(metadata_f.close)
 
     async def delete_persisted_record(self, record: KeyValueStoreRecord) -> None:
         """Delete the specified record from the key-value store."""

tests/unit/storage_clients/_memory/test_creation_management.py

@@ -1,15 +1,14 @@
 from __future__ import annotations
 
 import json
-import os
-from typing import TYPE_CHECKING
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
 
 from crawlee._consts import METADATA_FILENAME
 from crawlee.storage_clients._memory._creation_management import persist_metadata_if_enabled
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
 
 async def test_persist_metadata_skips_when_disabled(tmp_path: Path) -> None:
     await persist_metadata_if_enabled(data={'key': 'value'}, entity_directory=str(tmp_path), write_metadata=False)
@@ -18,17 +17,43 @@ async def test_persist_metadata_skips_when_disabled(tmp_path: Path) -> None:
 
 async def test_persist_metadata_creates_files_and_directories_when_enabled(tmp_path: Path) -> None:
     data = {'key': 'value'}
-    entity_directory = os.path.join(tmp_path, 'new_dir')
-    await persist_metadata_if_enabled(data=data, entity_directory=entity_directory, write_metadata=True)
-    assert os.path.exists(entity_directory)  # Check if directory was created
-    assert os.path.isfile(os.path.join(entity_directory, METADATA_FILENAME))  # Check if file was created
+    entity_directory = Path(tmp_path, 'new_dir')
+    await persist_metadata_if_enabled(data=data, entity_directory=str(entity_directory), write_metadata=True)
+    assert entity_directory.exists() is True  # Check if directory was created
+    assert (entity_directory / METADATA_FILENAME).is_file()  # Check if file was created
 
 
 async def test_persist_metadata_correctly_writes_data(tmp_path: Path) -> None:
     data = {'key': 'value'}
-    entity_directory = os.path.join(tmp_path, 'data_dir')
-    await persist_metadata_if_enabled(data=data, entity_directory=entity_directory, write_metadata=True)
-    metadata_path = os.path.join(entity_directory, METADATA_FILENAME)
+    entity_directory = Path(tmp_path, 'data_dir')
+    await persist_metadata_if_enabled(data=data, entity_directory=str(entity_directory), write_metadata=True)
+    metadata_path = entity_directory / METADATA_FILENAME
     with open(metadata_path) as f:  # noqa: ASYNC230
         content = f.read()
     assert json.loads(content) == data  # Check if correct data was written
+
+
+async def test_persist_metadata_rewrites_data_with_error(tmp_path: Path) -> None:
+    init_data = {'key': 'very_long_value'}
+    update_data = {'key': 'short_value'}
+    error_data = {'key': 'error'}
+
+    entity_directory = Path(tmp_path, 'data_dir')
+    metadata_path = entity_directory / METADATA_FILENAME
+
+    # write metadata with init_data
+    await persist_metadata_if_enabled(data=init_data, entity_directory=str(entity_directory), write_metadata=True)
+
+    # rewrite metadata with new_data
+    await persist_metadata_if_enabled(data=update_data, entity_directory=str(entity_directory), write_metadata=True)
+    with open(metadata_path) as f:  # noqa: ASYNC230
+        content = f.read()
+    assert json.loads(content) == update_data  # Check if correct data was rewritten
+
+    # raise interrupt between opening a file and writing
+    module_for_patch = 'crawlee.storage_clients._memory._creation_management.json_dumps'
+    with patch(module_for_patch, AsyncMock(side_effect=KeyboardInterrupt())), pytest.raises(KeyboardInterrupt):
+        await persist_metadata_if_enabled(data=error_data, entity_directory=str(entity_directory), write_metadata=True)
+    with open(metadata_path) as f:  # noqa: ASYNC230
+        content = f.read()
+    assert content == ''  # The file is empty after an error