[23.0] Fix Short Term Storage cleanup when metadata is lost

lib/galaxy/web/short_term_storage/__init__.py

@@ -252,6 +252,12 @@ def _load_metadata(self, target_directory: Path, meta_name: str):
         with open(meta_path) as f:
             return json.load(f)
 
+    def _load_metadata_safe(self, target_directory: Path, meta_name: str):
+        try:
+            return self._load_metadata(target_directory, meta_name)
+        except ObjectNotFound:
+            return None
+
     def _directory(self, target: Union[UUID, ShortTermStorageTarget]) -> Path:
         if isinstance(target, ShortTermStorageTarget):
             request_id = target.request_id
@@ -261,7 +267,14 @@ def _directory(self, target: Union[UUID, ShortTermStorageTarget]) -> Path:
         return self._root.joinpath(*relative_directory)
 
     def _cleanup_if_needed(self, request_id: UUID):
-        request_metadata = self._load_metadata(self._directory(request_id), "request")
+        target_directory = self._directory(request_id)
+        if not target_directory.exists():
+            return  # Nothing to clean
+        request_metadata = self._load_metadata_safe(target_directory, "request")
+        if request_metadata is None:
+            # Delete if metadata is lost
+            self._delete(request_id)
+            return
         duration = request_metadata["duration"]
         creation_datetime_str = request_metadata["created"]
         unprintStrptimeFmt = "%Y-%m-%d %H:%M:%S.%f"
@@ -272,7 +285,7 @@ def _cleanup_if_needed(self, request_id: UUID):
             self._delete(request_id)
 
     def _delete(self, request_id: UUID):
-        shutil.rmtree(self._directory(request_id))
+        shutil.rmtree(self._directory(request_id), ignore_errors=True)
 
     def cleanup(self):
         for directory in self._root.glob("*/*/*/*"):

test/unit/web_framework/test_short_term_storage.py

@@ -1,4 +1,5 @@
 import time
+from pathlib import Path
 
 import pytest
 
@@ -128,6 +129,28 @@ def test_cleanup_no_op_if_duration_not_reached(tmpdir):
     assert short_term_storage_target.path.exists()
 
 
+def test_cleanup_if_metadata_not_found(tmpdir):
+    config = ShortTermStorageConfiguration(
+        short_term_storage_directory=tmpdir,
+        maximum_storage_duration=100,
+    )
+    manager = ShortTermStorageManager(config=config)
+    short_term_storage_target = manager.new_target(
+        TEST_FILENAME,
+        TEST_MIME_TYPE,
+    )
+    short_term_storage_target.path.touch()
+    # Force metadata file deletion only
+    parent_directory_path = Path(short_term_storage_target.path.parent)
+    for metadata_file_path in parent_directory_path.glob("*.json"):
+        metadata_file_path.unlink()
+    assert short_term_storage_target.path.exists()
+
+    manager.cleanup()
+    # If the metadata is lost, the target will be deleted
+    assert not short_term_storage_target.path.exists()
+
+
 def test_serve_non_existent_raises_object_not_found(tmpdir):
     config = ShortTermStorageConfiguration(
         short_term_storage_directory=tmpdir,