Short term storage infrastructure.

lib/galaxy/app.py

@@ -9,6 +9,7 @@ import time
 from typing import (
     Any,
     Callable,
+    Dict,
     List,
     Tuple,
 )
@@ -117,6 +118,12 @@ from galaxy.visualization.genomes import Genomes
 from galaxy.visualization.plugins.registry import VisualizationsRegistry
 from galaxy.web import url_for
 from galaxy.web.proxy import ProxyManager
+from galaxy.web.short_term_storage import (
+    ShortTermStorageAllocator,
+    ShortTermStorageConfiguration,
+    ShortTermStorageManager,
+    ShortTermStorageMonitor,
+)
 from galaxy.web_stack import (
     application_stack_instance,
     ApplicationStack,
@@ -491,6 +498,21 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication):
         # Initialize the job management configuration
         self.job_config = self._register_singleton(jobs.JobConfiguration)
 
+        # Setup infrastructure for short term storage manager.
+        short_term_storage_config_kwds: Dict[str, Any] = {}
+        short_term_storage_config_kwds["short_term_storage_directory"] = self.config.short_term_storage_dir
+        short_term_storage_default_duration = self.config.short_term_storage_default_duration
+        short_term_storage_maximum_duration = self.config.short_term_storage_maximum_duration
+        if short_term_storage_default_duration is not None:
+            short_term_storage_config_kwds["default_storage_duration"] = short_term_storage_default_duration
+        if short_term_storage_maximum_duration is not None:
+            short_term_storage_config_kwds["maximum_storage_duration"] = short_term_storage_maximum_duration
+
+        short_term_storage_config = ShortTermStorageConfiguration(**short_term_storage_config_kwds)
+        short_term_storage_manager = ShortTermStorageManager(config=short_term_storage_config)
+        self._register_singleton(ShortTermStorageAllocator, short_term_storage_manager)
+        self._register_singleton(ShortTermStorageMonitor, short_term_storage_manager)
+
         # Tag handler
         self.tag_handler = self._register_singleton(GalaxyTagHandler)
         self.user_manager = self._register_singleton(UserManager)

lib/galaxy/app_unittest_utils/galaxy_mock.py

@@ -34,6 +34,12 @@ from galaxy.tools.data import ToolDataTableManager
 from galaxy.util import StructuredExecutionTimer
 from galaxy.util.bunch import Bunch
 from galaxy.util.dbkeys import GenomeBuilds
+from galaxy.web.short_term_storage import (
+    ShortTermStorageAllocator,
+    ShortTermStorageConfiguration,
+    ShortTermStorageManager,
+    ShortTermStorageMonitor,
+)
 from galaxy.web_stack import ApplicationStack
 
 
@@ -80,6 +86,10 @@ class MockApp(di.Container, GalaxyDataTestApp):
         self.name = kwargs.get("name", "galaxy")
         self[SharedModelMapping] = self.model
         self[GalaxyModelMapping] = self.model
+        sts_config = ShortTermStorageConfiguration(short_term_storage_directory=os.path.join(config.data_dir, "sts"))
+        sts_manager = ShortTermStorageManager(sts_config)
+        self[ShortTermStorageAllocator] = sts_manager
+        self[ShortTermStorageMonitor] = sts_manager
         self[galaxy_scoped_session] = self.model.context
         self.visualizations_registry = MockVisualizationsRegistry()
         self.tag_handler = tags.GalaxyTagHandler(self.model.context)

lib/galaxy/celery/__init__.py

@@ -108,6 +108,14 @@ def get_history_audit_table_prune_interval():
         return 3600
 
 
+def get_cleanup_short_term_storage_interval():
+    config = get_config()
+    if config:
+        return config.short_term_storage_cleanup_interval
+    else:
+        return 3600
+
+
 broker = get_broker()
 backend = get_backend()
 celery_app_kwd: Dict[str, Any] = {
@@ -118,14 +126,26 @@ if backend:
     celery_app_kwd["backend"] = backend
 
 celery_app = Celery("galaxy", **celery_app_kwd)
+
+# setup cron like tasks...
+beat_schedule: Dict[str, Dict[str, Any]] = {}
+
 prune_interval = get_history_audit_table_prune_interval()
 if prune_interval > 0:
-    celery_app.conf.beat_schedule = {
-        "prune-history-audit-table": {
-            "task": "galaxy.celery.tasks.prune_history_audit_table",
-            "schedule": prune_interval,
-        },
+    beat_schedule["prune-history-audit-table"] = {
+        "task": f"{MAIN_TASK_MODULE}.prune_history_audit_table",
+        "schedule": prune_interval,
     }
+
+cleanup_interval = get_cleanup_short_term_storage_interval()
+if cleanup_interval > 0:
+    beat_schedule["cleanup-short-term-storage"] = {
+        "task": f"{MAIN_TASK_MODULE}.cleanup_short_term_storage",
+        "schedule": cleanup_interval,
+    }
+
+if beat_schedule:
+    celery_app.conf.beat_schedule = beat_schedule
 celery_app.conf.timezone = "UTC"
 
 

lib/galaxy/celery/tasks.py

@@ -6,6 +6,7 @@ from galaxy.managers.model_stores import ModelStoreManager
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.schema.tasks import SetupHistoryExportJob
 from galaxy.util.custom_logging import get_logger
+from galaxy.web.short_term_storage import ShortTermStorageMonitor
 
 log = get_logger(__name__)
 
@@ -52,3 +53,9 @@ def export_history(
 def prune_history_audit_table(sa_session: galaxy_scoped_session):
     """Prune ever growing history_audit table."""
     model.HistoryAudit.prune(sa_session)
+
+
+@galaxy_task(action="clean up short term storage")
+def cleanup_short_term_storage(storage_monitor: ShortTermStorageMonitor):
+    """Cleanup short term storage."""
+    storage_monitor.cleanup()

lib/galaxy/config/schemas/config_schema.yml

@@ -493,6 +493,34 @@ mapping:
           changes are found, modified tours are automatically reloaded. Takes the same values as the
           'watch_tools' option.
 
+      short_term_storage_dir:
+        type: str
+        default: short_term_web_storage
+        path_resolves_to: cache_dir
+        required: false
+        desc: |
+          Location of files available for a short time as downloads (short term storage).
+
+      short_term_storage_default_duration:
+        type: int
+        required: false
+        desc: |
+          Default duration before short term storage files can be cleaned up.
+
+      short_term_storage_maximum_duration:
+        type: int
+        required: false
+        desc: |
+          Default duration before short term storage files can be cleaned up.
+
+      short_term_storage_cleanup_interval:
+        type: int
+        required: false
+        default: 3600
+        desc: |
+          How many seconds between instances of short term storage being cleaned up in default
+          Celery task configuration.
+
       file_sources_config_file:
         type: str
         default: file_sources_conf.yml

lib/galaxy/exceptions/__init__.py

@@ -59,6 +59,11 @@ class ObjectInvalid(Exception):
 # Please keep the exceptions ordered by status code
 
 
+class NoContentException(MessageException):
+    status_code = 204
+    err_code = error_codes_by_name["NO_CONTENT_GENERIC"]
+
+
 class ActionInputError(MessageException):
     status_code = 400
     err_code = error_codes_by_name["USER_REQUEST_INVALID_PARAMETER"]

lib/galaxy/exceptions/error_codes.json

@@ -4,6 +4,11 @@
       "code": 0,
       "message": "Unknown error occurred while processing request."
    },
+   {
+      "name": "NO_CONTENT_GENERIC",
+      "code": 204001,
+      "message": "Galaxy has not content to associate with this request."
+   },
    {
       "name": "USER_CANNOT_RUN_AS",
       "code": 400001,

lib/galaxy/exceptions/error_codes.py

@@ -16,7 +16,7 @@ UNKNOWN_ERROR_MESSAGE = "Unknown error occurred while processing request."
 class ErrorCode:
     """Small class allowing object representation for error descriptions loaded from JSON."""
 
-    def __init__(self, code, default_error_message):
+    def __init__(self, code: int, default_error_message: str):
         """Construct a :class:`ErrorCode` from supplied integer and error message."""
         self.code = code
         self.default_error_message = default_error_message or UNKNOWN_ERROR_MESSAGE
@@ -31,7 +31,7 @@ class ErrorCode:
 
     def __int__(self):
         """Return the error code integer."""
-        return int(self.code)
+        return self.code
 
 
 def _from_dict(entry):
@@ -44,8 +44,10 @@ def _from_dict(entry):
 
 error_codes_json = resource_string(__package__, "error_codes.json")
 error_codes_by_name: Dict[str, ErrorCode] = {}
+error_codes_by_int_code: Dict[int, ErrorCode] = {}
 
 for entry in loads(error_codes_json):
     name, error_code_obj = _from_dict(entry)
     globals()[name] = error_code_obj
     error_codes_by_name[name] = error_code_obj
+    error_codes_by_int_code[error_code_obj.code] = error_code_obj

lib/galaxy/web/short_term_storage/__init__.py

@@ -0,0 +1,295 @@
+import abc
+import contextlib
+import json
+import os
+import shutil
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import (
+    Any,
+    Dict,
+    Optional,
+    Union,
+)
+from uuid import uuid4
+
+from galaxy.exceptions import (
+    InternalServerError,
+    MessageException,
+    NoContentException,
+)
+from galaxy.exceptions.error_codes import error_codes_by_int_code
+from galaxy.util import (
+    directory_hash_id,
+    is_uuid,
+    safe_makedirs,
+)
+from galaxy.web.framework.decorators import api_error_message
+
+now = datetime.utcnow
+DEFAULT_STORAGE_DURATION = 24 * 60 * 60  # store for a day by default
+
+
+OptionalNumberT = Optional[Union[int, float]]
+
+
+@dataclass
+class ShortTermStorageConfiguration:
+    short_term_storage_directory: str
+    default_storage_duration: OptionalNumberT = None
+    maximum_storage_duration: OptionalNumberT = None
+
+
+@dataclass
+class ShortTermStorageTargetSecurity:
+    user_id: Optional[int] = None
+    session_id: Optional[int] = None
+
+    def to_dict(self) -> Dict[str, Optional[int]]:
+        return {
+            "user_id": self.user_id,
+            "session_id": self.session_id,
+        }
+
+    @classmethod
+    def from_dict(self, as_dict: Dict[str, Optional[int]]) -> "ShortTermStorageTargetSecurity":
+        return ShortTermStorageTargetSecurity(
+            user_id=as_dict.get("user_id"),
+            session_id=as_dict.get("session_id"),
+        )
+
+
+@dataclass
+class ShortTermStorageTarget:
+    request_id: str  # uuid
+    raw_path: str
+
+    @property
+    def path(self):
+        return Path(self.raw_path)
+
+
+@dataclass
+class ShortTermStorageServeCompletedInformation:
+    target: ShortTermStorageTarget
+    mime_type: str
+    filename: str
+    security: ShortTermStorageTargetSecurity
+
+
+@dataclass
+class ShortTermStorageServeCancelledInformation:
+    target: ShortTermStorageTarget
+    status_code: int
+    exception: Optional[Dict[str, Any]]
+
+    @property
+    def message_exception(self) -> MessageException:
+        serialized_exception = self.exception
+        if not serialized_exception:
+            raise NoContentException()
+        exception_obj = MessageException()
+        exception_obj.status_code = self.status_code
+        exception_obj.err_code = error_codes_by_int_code[serialized_exception["err_code"]]
+        exception_obj.err_msg = serialized_exception["err_msg"]
+        return exception_obj
+
+
+ShortTermStorageServeInformation = Union[
+    ShortTermStorageServeCompletedInformation, ShortTermStorageServeCancelledInformation
+]
+
+
+class ShortTermStorageAllocator(metaclass=abc.ABCMeta):
+
+    # TODO: Implement upstream_mod_zip=False, upstream_gzip=False - in initial request and serving...
+    @abc.abstractmethod
+    def new_target(
+        self,
+        filename: str,
+        mime_type: str,
+        duration: Optional[int] = None,
+        security: Optional[ShortTermStorageTargetSecurity] = None,
+    ) -> ShortTermStorageTarget:
+        """Return a new ShortTermStorageTarget for this short term file request."""
+
+
+class ShortTermStorageMonitor(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def is_ready(self, target: ShortTermStorageTarget) -> bool:
+        """Check if storage is ready."""
+
+    @abc.abstractmethod
+    def get_serve_info(self, target: ShortTermStorageTarget) -> ShortTermStorageServeInformation:
+        """Get information required to serve this short term storage target."""
+
+    @abc.abstractmethod
+    def finalize(self, target: ShortTermStorageTarget) -> None:
+        """Indicate the file is ready to be served."""
+
+    @abc.abstractmethod
+    def cancel(self, target: ShortTermStorageTarget, exception: Optional[MessageException] = None) -> None:
+        """Store metadata for failed task.
+
+        Implementation is responsible for indicating target is finalized as well.
+        """
+
+    @abc.abstractmethod
+    def target_path(self, target: ShortTermStorageTarget) -> Path:
+        """Return fully qualified path on this server for specified target."""
+
+    @abc.abstractmethod
+    def cleanup(self) -> None:
+        """Cleanup old requests."""
+
+    @abc.abstractmethod
+    def recover_target(self, request_id: str) -> ShortTermStorageTarget:
+        """Return an existing ShortTermStorageTarget from a specified request_id."""
+
+
+class ShortTermStorageManager(ShortTermStorageAllocator, ShortTermStorageMonitor):
+    def __init__(self, config: ShortTermStorageConfiguration):
+        self._config = config
+
+    def new_target(
+        self,
+        filename: str,
+        mime_type: str,
+        duration: OptionalNumberT = None,
+        security: Optional[ShortTermStorageTargetSecurity] = None,
+    ) -> ShortTermStorageTarget:
+        if security is None:
+            security = ShortTermStorageTargetSecurity()
+        request_id = str(uuid4())
+        target_directory = self._directory(request_id)
+        target = ShortTermStorageTarget(request_id=request_id, raw_path=str(target_directory / "target"))
+        safe_makedirs(target_directory)
+        duration = duration or self._config.default_storage_duration or DEFAULT_STORAGE_DURATION
+        maximum_storage_duration = self._config.maximum_storage_duration
+        if duration and maximum_storage_duration and duration > maximum_storage_duration:
+            duration = maximum_storage_duration
+        # optimize by placing the deletion time outside JSON as new file...
+        request_info = {
+            "filename": filename,
+            "mime_type": mime_type,
+            "duration": duration,
+            "created": str(now()),
+            "security": security.to_dict(),
+        }
+        self._store_metadata(target_directory, "request", request_info)
+        return target
+
+    def recover_target(self, request_id: str) -> ShortTermStorageTarget:
+        assert is_uuid(request_id)  # secure the directory structure...
+        target_directory = self._directory(request_id)
+        target = ShortTermStorageTarget(request_id=request_id, raw_path=str(target_directory / "target"))
+        return target
+
+    def is_ready(self, target: ShortTermStorageTarget) -> bool:
+        """Check if storage is ready."""
+        return self._finalized_path(target).exists()
+
+    def get_serve_info(self, target: ShortTermStorageTarget) -> ShortTermStorageServeInformation:
+        """Get information required to serve this short term storage target."""
+        cancelled = self._cancelled_path(target)
+        serve_info: ShortTermStorageServeInformation
+        target_directory = self._directory(target)
+        if cancelled.exists():
+            exception_metadata = self._load_metadata(target_directory, "cancelled")
+            serve_info = ShortTermStorageServeCancelledInformation(
+                target=target,
+                status_code=exception_metadata["status_code"],
+                exception=exception_metadata["exception"],
+            )
+        else:
+            request_metadata = self._load_metadata(target_directory, "request")
+            serve_info = ShortTermStorageServeCompletedInformation(
+                target=target,
+                filename=request_metadata["filename"],
+                mime_type=request_metadata["mime_type"],
+                security=ShortTermStorageTargetSecurity.from_dict(request_metadata["security"]),
+            )
+        return serve_info
+
+    def cancel(self, target: ShortTermStorageTarget, exception: Optional[MessageException] = None):
+        """Write metadata for failed task."""
+        if exception:
+            exception_json = {
+                "status_code": exception.status_code,
+                "exception": api_error_message(None, exception=exception),
+            }
+        else:
+            exception_json = {"status_code": 204, "exception": None}  # NO CONTENT
+        self._store_metadata(self._directory(target), "cancelled", exception_json)
+        self.finalize(target)
+
+    def finalize(self, target: ShortTermStorageTarget) -> None:
+        """Indicate the file is ready to be served."""
+        self._finalized_path(target).touch()
+
+    def target_path(self, target: ShortTermStorageTarget) -> Path:
+        return self._directory(target) / "target"
+
+    def _cancelled_path(self, target: ShortTermStorageTarget) -> Path:
+        return self._directory(target) / "cancelled.json"
+
+    def _finalized_path(self, target: ShortTermStorageTarget) -> Path:
+        return self._directory(target) / "finalized"
+
+    def _store_metadata(self, target_directory: Path, meta_name: str, meta_value: Any):
+        meta_path = target_directory / f"{meta_name}.json"
+        with open(meta_path, "w") as f:
+            json.dump(meta_value, f)
+
+    def _load_metadata(self, target_directory: Path, meta_name: str):
+        meta_path = target_directory / f"{meta_name}.json"
+        with open(meta_path, "r") as f:
+            return json.load(f)
+
+    def _directory(self, target: Union[str, ShortTermStorageTarget]) -> Path:
+        if isinstance(target, ShortTermStorageTarget):
+            request_id = target.request_id
+        else:
+            request_id = target
+        relative_directory = directory_hash_id(request_id) + [request_id]
+        return self._root.joinpath(*relative_directory)
+
+    def _cleanup_if_needed(self, request_id: str):
+        request_metadata = self._load_metadata(self._directory(request_id), "request")
+        duration = request_metadata["duration"]
+        creation_datetime_str = request_metadata["created"]
+        unprintStrptimeFmt = "%Y-%m-%d %H:%M:%S.%f"
+        creation_datetime = datetime.strptime(creation_datetime_str, unprintStrptimeFmt)
+        request_time = now() - creation_datetime
+        request_seconds = request_time.total_seconds()
+        if request_seconds > duration:
+            self._delete(request_id)
+
+    def _delete(self, request_id: str):
+        shutil.rmtree(self._directory(request_id))
+
+    def cleanup(self):
+        for directory in self._root.glob("*/*/*/*"):
+            request_id = os.path.basename(directory)
+            if not is_uuid(request_id):
+                continue
+            self._cleanup_if_needed(request_id)
+
+    @property
+    def _root(self) -> Path:
+        return Path(self._config.short_term_storage_directory)
+
+
+@contextlib.contextmanager
+def storage_context(short_term_storage_request_id: str, short_term_storage_monitor: ShortTermStorageMonitor):
+    target = short_term_storage_monitor.recover_target(short_term_storage_request_id)
+    try:
+        yield target
+    except MessageException as e:
+        short_term_storage_monitor.cancel(target, exception=e)
+        raise
+    except Exception as e:
+        short_term_storage_monitor.cancel(target, exception=InternalServerError(f"Unknown error: {e}"))
+        raise
+    short_term_storage_monitor.finalize(target)

lib/galaxy/webapps/galaxy/api/short_term_storage.py

@@ -0,0 +1,58 @@
+"""
+API operations around galaxy.web.short_term_storage infrastructure.
+"""
+from starlette.responses import FileResponse
+
+from galaxy.web.short_term_storage import (
+    ShortTermStorageMonitor,
+    ShortTermStorageServeCancelledInformation,
+    ShortTermStorageServeCompletedInformation,
+)
+from . import (
+    depends,
+    Router,
+)
+
+router = Router(tags=["short_term_storage"])
+
+
+@router.cbv
+class FastAPIShortTermStorage:
+    # typing here is not ideal, mypy is the issue xref https://github.com/python/mypy/issues/5374
+    short_term_storage_monitor: ShortTermStorageMonitor = depends(ShortTermStorageMonitor)  # type: ignore[misc]
+
+    @router.get(
+        "/api/short_term_storage/{storage_request_id}/ready",
+        summary="Determine if specified storage request ID is ready for download.",
+        response_description="Boolean indicating if the storage is ready.",
+    )
+    def is_ready(self, storage_request_id: str) -> bool:
+        storage_target = self.short_term_storage_monitor.recover_target(storage_request_id)
+        return self.short_term_storage_monitor.is_ready(storage_target)
+
+    @router.get(
+        "/api/short_term_storage/{storage_request_id}",
+        summary="Serve the staged download specified by request ID.",
+        response_description="Raw contents of the file.",
+        response_class=FileResponse,
+        responses={
+            200: {
+                "description": "The archive file containing the History.",
+            },
+            204: {
+                "description": "Request was cancelled without an exception condition recorded.",
+            },
+        },
+    )
+    def serve(self, storage_request_id: str):
+        storage_target = self.short_term_storage_monitor.recover_target(storage_request_id)
+        serve_info = self.short_term_storage_monitor.get_serve_info(storage_target)
+        if isinstance(serve_info, ShortTermStorageServeCompletedInformation):
+            return FileResponse(
+                path=serve_info.target.path,
+                media_type=serve_info.mime_type,
+                filename=serve_info.filename,
+            )
+
+        assert isinstance(serve_info, ShortTermStorageServeCancelledInformation)
+        raise serve_info.message_exception

packages/web_framework/setup.py

@@ -40,6 +40,7 @@ PACKAGES = [
     "galaxy.web.framework.middleware",
     "galaxy.web.legacy_framework",
     "galaxy.web.proxy",
+    "galaxy.web.short_term_storage",
 ]
 ENTRY_POINTS = """
         [console_scripts]

packages/web_stack/requirements.txt

@@ -1,2 +1,3 @@
 SQLAlchemy>=1.4.25,<2
 galaxy-util
+dataclasses; python_version < "3.7"

test/integration/test_config_defaults.py

@@ -85,6 +85,7 @@ PATH_CONFIG_PROPERTIES = [
     "shed_tool_config_file",
     "shed_tool_data_path",
     "shed_tool_data_table_config",
+    "short_term_storage_dir",
     "template_cache_path",
     "tool_data_path",
     "tool_dependency_cache_dir",
@@ -126,6 +127,7 @@ RESOLVE = {
     "shed_tool_config_file": "managed_config_dir",
     "shed_tool_data_path": "tool_data_path",
     "shed_tool_data_table_config": "managed_config_dir",
+    "short_term_storage_directory": "cache_dir",
     "tool_data_path": "root_dir",
     "tool_path": "root_dir",
     "tool_sheds_config_file": "config_dir",

test/unit/config/test_config_values.py

@@ -143,6 +143,7 @@ class ExpectedValues:
             "shed_tool_config_file": self._in_managed_config_dir("shed_tool_conf.xml"),
             "shed_tool_data_path": self._in_root_or_data_dir("tool-data"),
             "shed_tool_data_table_config": self._in_managed_config_dir("shed_tool_data_table_conf.xml"),
+            "short_term_storage_dir": self._in_cache_dir("short_term_web_storage"),
             "template_cache_path": self._in_cache_dir("compiled_templates"),
             "tool_cache_data_dir": self._in_cache_dir("tool_cache"),
             "tool_config_file": self._in_sample_dir("tool_conf.xml.sample"),

test/unit/web_framework/test_short_term_storage.py

@@ -0,0 +1,123 @@
+import time
+
+from galaxy.exceptions import MessageException
+from galaxy.web.short_term_storage import (
+    ShortTermStorageConfiguration,
+    ShortTermStorageManager,
+    ShortTermStorageServeCancelledInformation,
+    ShortTermStorageServeCompletedInformation,
+    ShortTermStorageTargetSecurity,
+    storage_context,
+)
+
+TEST_FILENAME = "moo.txt"
+TEST_MIME_TYPE = "text/plain"
+TEST_SLEEP_DURATION = 0.25
+
+
+def test_typical_usage(tmpdir):
+    config = ShortTermStorageConfiguration(short_term_storage_directory=tmpdir)
+    manager = ShortTermStorageManager(config=config)
+    security = ShortTermStorageTargetSecurity(
+        user_id=12,
+    )
+    short_term_storage_target = manager.new_target(
+        TEST_FILENAME,
+        TEST_MIME_TYPE,
+        security=security,
+    )
+    assert short_term_storage_target
+    assert not manager.is_ready(short_term_storage_target)
+    with open(short_term_storage_target.raw_path, "w") as f:
+        f.write("Moo Cow!!!")
+
+    # exercise recovering target from request id...
+    short_term_storage_target = manager.recover_target(short_term_storage_target.request_id)
+
+    manager.finalize(short_term_storage_target)
+    assert manager.is_ready(short_term_storage_target)
+    serve_info = manager.get_serve_info(short_term_storage_target)
+    assert serve_info
+    assert isinstance(serve_info, ShortTermStorageServeCompletedInformation)
+    assert serve_info.security.user_id == 12
+    with serve_info.target.path.open() as f:
+        assert f.read() == "Moo Cow!!!"
+
+
+def test_cancel_with_message_exception(tmpdir):
+    config = ShortTermStorageConfiguration(short_term_storage_directory=tmpdir)
+    manager = ShortTermStorageManager(config=config)
+    short_term_storage_target = manager.new_target(
+        TEST_FILENAME,
+        TEST_MIME_TYPE,
+    )
+    assert short_term_storage_target
+    assert not manager.is_ready(short_term_storage_target)
+    exception = MessageException("moo cow")
+    manager.cancel(short_term_storage_target, exception=exception)
+    assert manager.is_ready(short_term_storage_target)
+    serve_info = manager.get_serve_info(short_term_storage_target)
+    assert serve_info
+    assert isinstance(serve_info, ShortTermStorageServeCancelledInformation)
+    exc = serve_info.message_exception
+    assert exc.err_code.code == 0
+
+
+def test_cancel_with_arbitrary_exception(tmpdir):
+    config = ShortTermStorageConfiguration(short_term_storage_directory=tmpdir)
+    manager = ShortTermStorageManager(config=config)
+    short_term_storage_target = manager.new_target(
+        TEST_FILENAME,
+        TEST_MIME_TYPE,
+    )
+    assert short_term_storage_target
+    assert not manager.is_ready(short_term_storage_target)
+    exception_raised = False
+    try:
+        with storage_context(short_term_storage_target.request_id, manager):
+            raise Exception("Moo Cow...")
+    except Exception:
+        exception_raised = True
+    assert exception_raised
+    assert manager.is_ready(short_term_storage_target)
+    serve_info = manager.get_serve_info(short_term_storage_target)
+    assert serve_info
+    assert isinstance(serve_info, ShortTermStorageServeCancelledInformation)
+    exc = serve_info.message_exception
+    assert exc.status_code == 500
+    assert exc.err_code.code == 500001
+    assert "Moo Cow..." in str(exc)
+
+
+def test_cleanup(tmpdir):
+    config = ShortTermStorageConfiguration(
+        short_term_storage_directory=tmpdir,
+        maximum_storage_duration=TEST_SLEEP_DURATION / 10,
+    )
+    manager = ShortTermStorageManager(config=config)
+    short_term_storage_target = manager.new_target(
+        TEST_FILENAME,
+        TEST_MIME_TYPE,
+    )
+    short_term_storage_target.path.touch()
+    assert short_term_storage_target.path.exists()
+    time.sleep(TEST_SLEEP_DURATION)
+    manager.cleanup()
+    assert not short_term_storage_target.path.exists()
+
+
+def test_cleanup_no_op_if_duration_not_reached(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()
+    assert short_term_storage_target.path.exists()
+    time.sleep(TEST_SLEEP_DURATION)
+    manager.cleanup()
+    assert short_term_storage_target.path.exists()