[Frontend] Configurable video storage backends with TTL (#2531)

Signed-off-by: Ian Eaves <ian.k.eaves@gmail.com>

docs/serving/videos_api.md

@@ -195,13 +195,16 @@ curl -X POST http://localhost:8091/v1/videos/sync \
 
 ## Storage
 
-Set `VLLM_OMNI_STORAGE_PATH` to control where asynchronous video outputs are
+Set `VLLM_OMNI_SERVER_STORAGE__PATH` to control where asynchronous video outputs are
 stored:
 
 ```bash
-export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos
+export VLLM_OMNI_SERVER_STORAGE__PATH=/var/tmp/vllm-omni-videos
 ```
 
+> `VLLM_OMNI_STORAGE_PATH` is deprecated and will be removed in a future release;
+> use `VLLM_OMNI_SERVER_STORAGE__PATH` instead.
+
 ## Model-Specific Examples
 
 For complete text-to-video, image-to-video, and model-specific video-to-video

docs/user_guide/examples/online_serving/image_to_video.md

@@ -84,16 +84,26 @@ curl -X POST http://localhost:8091/v1/videos/sync \
 Generated video files are stored on local disk by the async video API.
 Local file storage behavior can be controlled via the following environment variables:
 
-- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`)
-- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`)
+- `VLLM_OMNI_SERVER_STORAGE__PATH`: directory used for generated files
+  (default: `/tmp/storage`)
+- `VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY`: max concurrent save/delete/open
+  operations (default: `4`)
+- `VLLM_OMNI_SERVER_STORAGE__FILE_TTL`: optional TTL for generated files in
+  seconds
+- `VLLM_OMNI_SERVER_STORAGE__TTL_SWEEP_INTERVAL`: optional sweep interval in
+  seconds for enforcing file TTL; defaults to `300` when TTL is set
 
 Example:
 
 ```bash
-export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos
-export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8
+export VLLM_OMNI_SERVER_STORAGE__PATH=/var/tmp/vllm-omni-videos
+export VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY=8
+export VLLM_OMNI_SERVER_STORAGE__FILE_TTL=86400
+export VLLM_OMNI_SERVER_STORAGE__TTL_SWEEP_INTERVAL=300
 ```
 
+See also: [Configuration Options](../../../configuration/README.md)
+
 ## API Calls
 
 ### Method 1: Using curl

docs/user_guide/examples/online_serving/text_to_video.md

@@ -90,16 +90,26 @@ curl -X POST http://localhost:8091/v1/videos/sync \
 Generated video files are stored on local disk by the async video API.
 Local file storage behavior can be controlled via the following environment variables:
 
-- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`)
-- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`)
+- `VLLM_OMNI_SERVER_STORAGE__PATH`: directory used for generated files
+  (default: `/tmp/storage`)
+- `VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY`: max concurrent save/delete/open
+  operations (default: `4`)
+- `VLLM_OMNI_SERVER_STORAGE__FILE_TTL`: optional TTL for generated files in
+  seconds
+- `VLLM_OMNI_SERVER_STORAGE__TTL_SWEEP_INTERVAL`: optional sweep interval in
+  seconds for enforcing file TTL; defaults to `300` when TTL is set
 
 Example:
 
 ```bash
-export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos
-export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8
+export VLLM_OMNI_SERVER_STORAGE__PATH=/var/tmp/vllm-omni-videos
+export VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY=8
+export VLLM_OMNI_SERVER_STORAGE__FILE_TTL=86400
+export VLLM_OMNI_SERVER_STORAGE__TTL_SWEEP_INTERVAL=300
 ```
 
+See also: [Configuration Options](../../../configuration/README.md)
+
 ## API Calls
 
 ### Method 1: Using curl

examples/online_serving/image_to_video/README.md

@@ -120,14 +120,17 @@ Use `/v1/videos/sync` if you want to write the MP4 directly to a file. `POST /v1
 Generated video files are stored on local disk by the async video API.
 Local file storage behavior can be controlled via the following environment variables:
 
-- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`)
-- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`)
+- `VLLM_OMNI_SERVER_STORAGE__PATH`: directory used for generated files (default: `/tmp/storage`)
+- `VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY`: max concurrent save/delete operations (default: `4`)
+
+`VLLM_OMNI_STORAGE_PATH` and `VLLM_OMNI_STORAGE_MAX_CONCURRENCY` are deprecated and will be
+removed in a future release; use the names above instead.
 
 Example:
 
 ```bash
-export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos
-export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8
+export VLLM_OMNI_SERVER_STORAGE__PATH=/var/tmp/vllm-omni-videos
+export VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY=8
 ```
 
 ## API Calls

examples/online_serving/text_to_video/README.md

@@ -87,14 +87,17 @@ curl -X POST http://localhost:8091/v1/videos/sync \
 Generated video files are stored on local disk by the async video API.
 Local file storage behavior can be controlled via the following environment variables:
 
-- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`)
-- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`)
+- `VLLM_OMNI_SERVER_STORAGE__PATH`: directory used for generated files (default: `/tmp/storage`)
+- `VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY`: max concurrent save/delete operations (default: `4`)
+
+`VLLM_OMNI_STORAGE_PATH` and `VLLM_OMNI_STORAGE_MAX_CONCURRENCY` are deprecated and will be
+removed in a future release; use the names above instead.
 
 Example:
 
 ```bash
-export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos
-export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8
+export VLLM_OMNI_SERVER_STORAGE__PATH=/var/tmp/vllm-omni-videos
+export VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY=8
 ```
 
 ## API Calls

requirements/common.txt

@@ -28,3 +28,5 @@ aenum==3.1.16
 pyzmq>=25.0.0
 janus>=1.0.0
 msgpack>=1.0.0
+pydantic>=2.1.0
+pydantic-settings>=2.0.0

tests/config/test_server_settings.py

@@ -0,0 +1,84 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Unit tests for ServerSettings env-var parameterization."""
+
+import pytest
+
+from vllm_omni.config.server_settings import ServerSettings
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+ENV_VARS = [
+    "VLLM_OMNI_SERVER_STORAGE__PATH",
+    "VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY",
+    "VLLM_OMNI_SERVER_STORAGE__FILE_TTL",
+    "VLLM_OMNI_SERVER_STORAGE__TTL_SWEEP_INTERVAL",
+    "VLLM_OMNI_STORAGE_PATH",
+    "VLLM_OMNI_STORAGE_MAX_CONCURRENCY",
+]
+
+
+@pytest.fixture(autouse=True)
+def clean_env(monkeypatch):
+    for name in ENV_VARS:
+        monkeypatch.delenv(name, raising=False)
+
+
+def test_defaults_when_no_env_set():
+    storage = ServerSettings().storage
+
+    assert storage.path == "/tmp/storage"
+    assert storage.file_concurrency == 4
+    assert storage.file_ttl is None
+    assert storage.ttl_sweep_interval is None
+
+
+def test_nested_env_vars_populate_storage(monkeypatch):
+    monkeypatch.setenv("VLLM_OMNI_SERVER_STORAGE__PATH", "/var/storage")
+    monkeypatch.setenv("VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY", "8")
+    monkeypatch.setenv("VLLM_OMNI_SERVER_STORAGE__FILE_TTL", "120")
+    monkeypatch.setenv("VLLM_OMNI_SERVER_STORAGE__TTL_SWEEP_INTERVAL", "30")
+
+    storage = ServerSettings().storage
+
+    assert storage.path == "/var/storage"
+    assert storage.file_concurrency == 8
+    assert storage.file_ttl == 120
+    assert storage.ttl_sweep_interval == 30
+
+
+def test_ttl_sweep_interval_defaults_when_ttl_set(monkeypatch):
+    monkeypatch.setenv("VLLM_OMNI_SERVER_STORAGE__FILE_TTL", "60")
+
+    storage = ServerSettings().storage
+
+    assert storage.file_ttl == 60
+    assert storage.ttl_sweep_interval == 300
+
+
+def test_deprecated_path_alias_populates_and_warns(monkeypatch):
+    monkeypatch.setenv("VLLM_OMNI_STORAGE_PATH", "/legacy/path")
+
+    with pytest.warns(DeprecationWarning, match="VLLM_OMNI_STORAGE_PATH"):
+        storage = ServerSettings().storage
+
+    assert storage.path == "/legacy/path"
+
+
+def test_deprecated_concurrency_alias_populates_and_warns(monkeypatch):
+    monkeypatch.setenv("VLLM_OMNI_STORAGE_MAX_CONCURRENCY", "16")
+
+    with pytest.warns(DeprecationWarning, match="VLLM_OMNI_STORAGE_MAX_CONCURRENCY"):
+        storage = ServerSettings().storage
+
+    assert storage.file_concurrency == 16
+
+
+def test_new_env_var_wins_over_deprecated_alias(monkeypatch):
+    monkeypatch.setenv("VLLM_OMNI_SERVER_STORAGE__PATH", "/new/path")
+    monkeypatch.setenv("VLLM_OMNI_STORAGE_PATH", "/legacy/path")
+
+    with pytest.warns(DeprecationWarning, match="VLLM_OMNI_STORAGE_PATH"):
+        storage = ServerSettings().storage
+
+    assert storage.path == "/new/path"

tests/entrypoints/openai_api/test_storage_manager.py

@@ -0,0 +1,94 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+Unit tests for the OpenAI video storage managers.
+"""
+
+import asyncio
+import os
+import time
+
+import pytest
+
+from vllm_omni.entrypoints.openai import storage as storage_module
+from vllm_omni.entrypoints.openai.storage import FileStorageHandle, LocalStorageManager, LocalStorageTTLManager
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_local_storage_open_returns_file_handle_for_saved_key(tmp_path):
+    storage = LocalStorageManager(storage_path=str(tmp_path / "storage"))
+    save_context = asyncio.run(storage.save(b"video-bytes", "video-123"))
+
+    handle = asyncio.run(storage.open(save_context.key))
+
+    assert isinstance(handle, FileStorageHandle)
+    assert handle.path == os.path.join(storage.storage_path, "video-123")
+    with open(handle.path, "rb") as file:
+        assert file.read() == b"video-bytes"
+
+
+def test_local_storage_open_returns_none_for_missing_key(tmp_path):
+    storage = LocalStorageManager(storage_path=str(tmp_path / "storage"))
+
+    handle = asyncio.run(storage.open("missing-video"))
+
+    assert handle is None
+
+
+def test_local_storage_ttl_save_sets_expiration_metadata(tmp_path, monkeypatch):
+    storage = LocalStorageTTLManager(
+        storage_path=str(tmp_path / "storage"),
+        max_concurrency=1,
+        ttl_seconds=60,
+        sweep_interval_seconds=300,
+    )
+    monkeypatch.setattr(storage_module.time, "time", lambda: 1_700_000_000)
+
+    save_context = asyncio.run(storage.save(b"video-bytes", "video-ttl"))
+
+    assert save_context.key == "video-ttl"
+    assert save_context.created_at == 1_700_000_000
+    assert save_context.expires_at == 1_700_000_060
+
+
+def test_local_storage_ttl_sweeper_removes_expired_file(tmp_path):
+    storage = LocalStorageTTLManager(
+        storage_path=str(tmp_path / "storage"),
+        max_concurrency=1,
+        ttl_seconds=1,
+        sweep_interval_seconds=60,
+    )
+
+    async def setup_file() -> tuple[str, str]:
+        save_context = await storage.save(b"video-bytes", "video-expired")
+        file_path = storage.get_full_file_path(save_context.key)
+        return save_context.key, file_path
+
+    storage_key, file_path = asyncio.run(setup_file())
+    expired_mtime = time.time() - 10
+    os.utime(file_path, (expired_mtime, expired_mtime))
+    assert os.path.exists(file_path)
+
+    deleted = asyncio.run(storage._sweep_once(time.time() - 1))
+
+    assert deleted == 1
+    assert not os.path.exists(file_path)
+    assert asyncio.run(storage.open(storage_key)) is None
+
+
+def test_local_storage_ttl_sweeper_keeps_files_when_path_missing(tmp_path):
+    storage = LocalStorageTTLManager(
+        storage_path=str(tmp_path / "storage"),
+        max_concurrency=1,
+        ttl_seconds=1,
+        sweep_interval_seconds=60,
+    )
+
+    missing_path = storage.get_full_file_path("video-missing")
+    assert not os.path.exists(missing_path)
+
+    deleted = asyncio.run(storage._sweep_once(time.time() - 1))
+
+    assert deleted == 0
+    assert not os.path.exists(missing_path)

tests/entrypoints/openai_api/test_video_server.py

@@ -1020,6 +1020,31 @@ def test_invalid_lora_returns_400(test_client):
     assert "lora object" in failed["error"]["message"].lower()
 
 
+def test_failed_generation_awaits_storage_cleanup(test_client, isolated_video_backends, mocker: MockerFixture):
+    """Regression (merge seam): when async generation raises, the failure handler
+    must ``await _cleanup_video(video_id)`` (single-arg, async) and still record
+    FAILED. Upstream carried a sync ``_cleanup_video(video_id, output_path)`` whose
+    stale call in the generic handler sat outside the conflict markers; against the
+    PR's async storage manager that raised NameError before the FAILED update,
+    wedging the job in IN_PROGRESS and orphaning the artifact."""
+    _store, _tasks, storage = isolated_video_backends
+    delete_spy = mocker.spy(storage, "delete")
+    mocker.patch.object(
+        OmniOpenAIServingVideo,
+        "generate_video_bytes",
+        side_effect=RuntimeError("GPU exploded"),
+    )
+
+    response = test_client.post("/v1/videos", data={"prompt": "will fail"})
+    assert response.status_code == 200
+    video_id = response.json()["id"]
+
+    failed = _wait_for_status(test_client, video_id, VideoGenerationStatus.FAILED.value)
+    assert failed["error"]["code"] == 500
+    assert "GPU exploded" in failed["error"]["message"]
+    delete_spy.assert_called_once_with(video_id)
+
+
 def test_async_guardrail_error_returns_400_on_retrieve(test_client, mocker: MockerFixture):
     mocker.patch.object(
         OmniOpenAIServingVideo,
@@ -1178,7 +1203,7 @@ def test_delete_completed_job_removes_file_and_metadata(test_client, mocker: Moc
     final = _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value)
     file_name = final["file_name"]
     assert file_name is not None
-    file_path = os.path.join(api_server.STORAGE_MANAGER.storage_path, file_name)
+    file_path = os.path.join(api_server.STORAGE_MANAGER.storage_path, video_id)
     assert os.path.exists(file_path)
 
     delete_resp = test_client.delete(f"/v1/videos/{video_id}")
@@ -1189,6 +1214,30 @@ def test_delete_completed_job_removes_file_and_metadata(test_client, mocker: Moc
     assert not os.path.exists(file_path)
 
 
+def test_download_completed_job_uses_storage_open_and_download_name(test_client, mocker: MockerFixture):
+    video_bytes = b"stored-video-data"
+    mocker.patch(
+        "vllm_omni.entrypoints.openai.serving_video._encode_video_bytes",
+        return_value=video_bytes,
+    )
+    create_resp = test_client.post("/v1/videos", data={"prompt": "Download this video"})
+    assert create_resp.status_code == 200
+    video_id = create_resp.json()["id"]
+
+    final = _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value)
+    file_name = final["file_name"]
+    assert file_name == f"{video_id}.mp4"
+
+    storage_path = os.path.join(api_server.STORAGE_MANAGER.storage_path, video_id)
+    assert os.path.exists(storage_path)
+
+    response = test_client.get(f"/v1/videos/{video_id}/content")
+    assert response.status_code == 200
+    assert response.content == video_bytes
+    assert response.headers["content-type"] == "video/mp4"
+    assert file_name in response.headers["content-disposition"]
+
+
 def test_delete_in_progress_job_cancels_task_and_removes_metadata(test_client):
     handler = BlockingVideoHandler()
     test_client.app.state.openai_serving_video = handler

vllm_omni/config/server_settings.py

@@ -0,0 +1,58 @@
+import os
+import warnings
+from typing import Literal
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+def _resolve_deprecated_env(old_name: str, new_name: str, value):
+    old = os.environ.get(old_name)
+    if old is None:
+        return value
+    warnings.warn(
+        f"{old_name} is deprecated and will be removed in a future release; use {new_name} instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+    return value if new_name in os.environ else old
+
+
+class FileBackend(BaseModel):
+    type: Literal["file"] = "file"
+    path: str = Field(default="/tmp/storage", validate_default=True, description="Local path to store completed files.")
+    file_concurrency: int = Field(
+        default=4, validate_default=True, description="Maximum number of file operations permitted at a time"
+    )
+    file_ttl: int | None = Field(
+        default=None, description="Optional TTL (in seconds) configuration settings for locally stored files."
+    )
+    ttl_sweep_interval: int | None = Field(
+        default=None, description="Optional frequency (in seconds) to enforce file TTLs."
+    )
+
+    @field_validator("path", mode="before")
+    @classmethod
+    def _migrate_path(cls, value):
+        return _resolve_deprecated_env("VLLM_OMNI_STORAGE_PATH", "VLLM_OMNI_SERVER_STORAGE__PATH", value)
+
+    @field_validator("file_concurrency", mode="before")
+    @classmethod
+    def _migrate_file_concurrency(cls, value):
+        return _resolve_deprecated_env(
+            "VLLM_OMNI_STORAGE_MAX_CONCURRENCY", "VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY", value
+        )
+
+    @model_validator(mode="after")
+    def set_default_ttl_sweep_interval(self) -> "FileBackend":
+        if self.file_ttl is not None and self.ttl_sweep_interval is None:
+            self.ttl_sweep_interval = 300
+        return self
+
+
+class ServerSettings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="VLLM_OMNI_SERVER_", env_nested_delimiter="__")
+    storage: FileBackend = Field(default_factory=FileBackend)
+
+
+SERVER_SETTINGS_CONFIG = ServerSettings()

vllm_omni/entrypoints/openai/api_server.py

@@ -137,7 +137,7 @@ from vllm_omni.entrypoints.openai.stage_params import (
     build_stage_sampling_params_list,
     get_default_sampling_params_list,
 )
-from vllm_omni.entrypoints.openai.storage import STORAGE_MANAGER
+from vllm_omni.entrypoints.openai.storage import STORAGE_MANAGER, FileStorageHandle
 from vllm_omni.entrypoints.openai.stores import VIDEO_STORE, VIDEO_TASKS
 from vllm_omni.entrypoints.openai.utils import get_stage_type, parse_lora_request
 from vllm_omni.entrypoints.openai.video_api_utils import decode_audio_url, decode_input_reference
@@ -474,6 +474,9 @@ async def omni_run_server_worker(listen_address, sock, args, client_config=None,
 
         await omni_init_app_state(engine_client, app.state, args)
 
+        # Start background processes
+        await STORAGE_MANAGER.start()
+
         # Conditionally register profiler endpoints based on stage YAML configs
         stage_configs = engine_client.stage_configs if hasattr(engine_client, "stage_configs") else None
         if _should_enable_profiler_endpoints(stage_configs):
@@ -2690,12 +2693,11 @@ def _video_error_from_exception(exc: Exception) -> VideoError:
     )
 
 
-def _cleanup_video(video_id: str, output_path: str | None):
+async def _cleanup_video(video_id: str):
     try:
-        if output_path is not None:
-            os.remove(output_path)
-    except OSError:
-        logger.warning("Failed to cleanup partial video file '%s' for id=%s", output_path, video_id)
+        await STORAGE_MANAGER.delete(video_id)
+    except Exception:
+        logger.warning("Failed to cleanup partial video file '%s'", video_id)
 
 
 async def _run_video_generation_job(
@@ -2714,7 +2716,6 @@ async def _run_video_generation_job(
 
     await VIDEO_STORE.update_fields(video_id, {"status": VideoGenerationStatus.IN_PROGRESS})
     started_at = time.perf_counter()
-    output_path = None
     try:
         video_bytes, stage_durations, peak_memory_mb, action = await handler.generate_video_bytes(
             request,
@@ -2724,27 +2725,27 @@ async def _run_video_generation_job(
             reference_audio=reference_audio,
         )
 
-        file_name = f"{video_id}.{job.file_extension}"
-        output_path = await STORAGE_MANAGER.save(video_bytes, file_name)
-        logger.info("Video request %s persisted %s output file.", video_id, output_path)
+        save_context = await STORAGE_MANAGER.save(video_bytes, video_id)
+        logger.info("Video request %s persisted %s output file.", video_id, save_context.key)
 
-        await VIDEO_STORE.update_fields(
-            video_id,
-            {
-                "status": VideoGenerationStatus.COMPLETED,
-                "progress": 100,
-                "file_name": file_name,
-                "completed_at": int(time.time()),
-                "inference_time_s": time.perf_counter() - started_at,
-                "stage_durations": stage_durations,
-                "peak_memory_mb": peak_memory_mb,
-                "action": action,
-            },
-        )
+        updated_fields = {
+            "status": VideoGenerationStatus.COMPLETED,
+            "progress": 100,
+            "file_name": f"{video_id}.{job.file_extension}",
+            "completed_at": save_context.created_at,
+            "inference_time_s": time.perf_counter() - started_at,
+            "stage_durations": stage_durations,
+            "peak_memory_mb": peak_memory_mb,
+            "action": action,
+        }
+        if save_context.expires_at is not None:
+            updated_fields["expires_at"] = save_context.expires_at
+
+        await VIDEO_STORE.update_fields(video_id, updated_fields)
     except (EngineGenerateError, EngineDeadError) as exc:
         logger.exception("Video generation failed (engine error) for id=%s", video_id)
 
-        _cleanup_video(video_id, output_path)
+        await _cleanup_video(video_id)
         await VIDEO_STORE.update_fields(
             video_id,
             {
@@ -2764,7 +2765,7 @@ async def _run_video_generation_job(
     except Exception as exc:
         logger.exception("Video generation failed for id=%s", video_id)
 
-        _cleanup_video(video_id, output_path)
+        await _cleanup_video(video_id)
         await VIDEO_STORE.update_fields(
             video_id,
             {
@@ -2775,7 +2776,7 @@ async def _run_video_generation_job(
             },
         )
     except asyncio.CancelledError:
-        _cleanup_video(video_id, output_path)
+        await _cleanup_video(video_id)
         await VIDEO_STORE.pop(video_id)
         raise
     finally:
@@ -3163,7 +3164,7 @@ async def delete_video(video_id: str) -> VideoDeleteResponse:
     elif job.status is VideoGenerationStatus.FAILED:
         if job.file_name is not None:
             try:
-                await STORAGE_MANAGER.delete(job.file_name)
+                await STORAGE_MANAGER.delete(video_id)
             except Exception:
                 logger.warning("Failed to delete stored artifact for failed video job %s", video_id, exc_info=True)
 
@@ -3173,13 +3174,13 @@ async def delete_video(video_id: str) -> VideoDeleteResponse:
     if job.file_name is None:
         raise HTTPException(status_code=409, detail="Video output not yet available. Please try again later.")
 
-    await STORAGE_MANAGER.delete(job.file_name)
+    await STORAGE_MANAGER.delete(video_id)
     await VIDEO_STORE.pop(video_id)
     return VideoDeleteResponse(id=job.id, deleted=True)
 
 
 @router.get("/v1/videos/{video_id}/content")
-async def download_video(video_id: str) -> FileResponse:
+async def download_video(video_id: str) -> Response:
     """Download the generated file for a completed video job.
 
     Args:
@@ -3202,11 +3203,19 @@ async def download_video(video_id: str) -> FileResponse:
     if not job.file_name:
         raise HTTPException(status_code=404, detail="Generation is still in-progress")
 
-    full_path = STORAGE_MANAGER.get_full_file_path(job.file_name)
-    if not os.path.exists(full_path):
+    file_handle = await STORAGE_MANAGER.open(video_id)
+    if file_handle is None:
         raise HTTPException(status_code=404, detail="Generated video file not found on disk")
 
-    return FileResponse(path=full_path, media_type=job.media_type, filename=job.file_name)
+    file_name = job.file_name or f"{video_id}.{job.file_extension}"
+    if isinstance(file_handle, FileStorageHandle):
+        response = FileResponse(path=file_handle.path, media_type=job.media_type, filename=file_name)
+    else:
+        raise HTTPException(
+            status_code=500, detail=f"Server generated an unsupported file storage handle for file id {video_id}"
+        )
+
+    return response
 
 
 @profiler_router.post("/start_profile")

vllm_omni/entrypoints/openai/storage.py

@@ -1,32 +1,78 @@
 import asyncio
+import contextlib
 import os
+import stat
+import time
 from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
 from tempfile import NamedTemporaryFile
+from typing import Generic, Literal, TypeVar
+
+from vllm.logger import init_logger
+
+from vllm_omni.config.server_settings import SERVER_SETTINGS_CONFIG, FileBackend
+
+logger = init_logger(__name__)
 
 
-class StorageBaseManager(ABC):
+@dataclass
+class SaveContext:
+    key: str
+    created_at: int
+    expires_at: int | None = None
+
+
+@dataclass(frozen=True)
+class BaseStorageHandle:
+    kind: str
+
+
+@dataclass(frozen=True)
+class FileStorageHandle(BaseStorageHandle):
+    path: str
+    kind: Literal["path"] = field(default="path", init=False)
+
+
+K = TypeVar("K", bound=BaseStorageHandle, covariant=True)
+
+
+class StorageBaseManager(Generic[K], ABC):
     @abstractmethod
-    async def save(self, *args, **kwargs) -> str:
+    async def save(self, *args, **kwargs) -> SaveContext:
         pass
 
     @abstractmethod
     async def delete(self, *args, **kwargs) -> bool:
         pass
 
+    async def start(self, *args, **kwargs) -> None:
+        pass
 
-class LocalStorageManager(StorageBaseManager):
-    def __init__(self, storage_path: str | None = None):
-        if storage_path is None:
-            storage_path = os.getenv("VLLM_OMNI_STORAGE_PATH", "/tmp/storage")
-        self.storage_path = storage_path
+    async def stop(self, *args, **kwargs) -> None:
+        pass
+
+    @abstractmethod
+    async def open(self, storage_key: str) -> K | None:
+        pass
+
+
+class LocalStorageManager(StorageBaseManager[FileStorageHandle]):
+    def __init__(self, storage_path: str, max_concurrency: int = 4):
+        self.storage_path = os.path.realpath(storage_path)
         os.makedirs(self.storage_path, exist_ok=True)
 
-        max_concurrency = int(os.getenv("VLLM_OMNI_STORAGE_MAX_CONCURRENCY", "4"))
         self._io_semaphore = asyncio.Semaphore(max(1, max_concurrency))
 
-    def _save_sync(self, data: bytes, file_name: str) -> str:
+    async def open(self, storage_key: str) -> FileStorageHandle | None:
+        local_file = self.get_full_file_path(storage_key)
+        if not os.path.exists(local_file):
+            return None
+        return FileStorageHandle(path=local_file)
+
+    def _save_sync(self, data: bytes, file_name: str) -> SaveContext:
         filename = self.get_full_file_path(file_name)
         tmp_name: str | None = None
+
         try:
             with NamedTemporaryFile("wb", dir=self.storage_path, delete=False) as f:
                 tmp_name = f.name
@@ -34,7 +80,8 @@ class LocalStorageManager(StorageBaseManager):
                 f.flush()
                 os.fsync(f.fileno())
             os.replace(tmp_name, filename)
-            return filename
+            response = SaveContext(key=file_name, created_at=int(time.time()))
+            return response
         except Exception:
             if tmp_name is not None:
                 try:
@@ -43,7 +90,7 @@ class LocalStorageManager(StorageBaseManager):
                     pass
             raise
 
-    async def save(self, data: bytes, file_name: str) -> str:
+    async def save(self, data: bytes, file_name: str) -> SaveContext:
         async with self._io_semaphore:
             return await asyncio.to_thread(self._save_sync, data, file_name)
 
@@ -62,9 +109,102 @@ class LocalStorageManager(StorageBaseManager):
         return os.path.exists(self.get_full_file_path(file_name))
 
     def get_full_file_path(self, file_name: str) -> str:
-        return os.path.join(self.storage_path, file_name)
+        full = os.path.realpath(os.path.join(self.storage_path, file_name))
+
+        if os.path.commonpath([self.storage_path, full]) != self.storage_path:
+            raise ValueError(f"Illegal storage key: {file_name!r}")
+
+        return full
 
 
-# Should implement TTL management on local file storage
-# and recovery on restart in the *STORE's
-STORAGE_MANAGER = LocalStorageManager()
+class LocalStorageTTLManager(LocalStorageManager):
+    def __init__(self, ttl_seconds: int, sweep_interval_seconds: int, *args, **kwargs):
+        if ttl_seconds <= 0:
+            raise ValueError("`ttl_seconds` must be greater than or equal to 1.")
+        if sweep_interval_seconds <= 0:
+            raise ValueError("`sweep_interval_seconds` must be greater than or equal to 1.")
+
+        self._ttl_seconds = ttl_seconds
+        self._sweep_interval_seconds = sweep_interval_seconds
+        self._sweeper_task: asyncio.Task[None] | None = None
+
+        super().__init__(*args, **kwargs)
+
+    async def save(self, data: bytes, file_name: str) -> SaveContext:
+        result = await super().save(data, file_name)
+        result.expires_at = result.created_at + self._ttl_seconds
+        return result
+
+    async def _sweep_once(self, cutoff: float) -> int:
+        expired: list[str] = []
+        for entry in os.scandir(self.storage_path):
+            try:
+                if not entry.is_file(follow_symlinks=False):
+                    continue
+                if entry.stat(follow_symlinks=False).st_mtime < cutoff:
+                    expired.append(entry.path)
+            except FileNotFoundError:
+                logger.debug("TTL sweep skipped %s; file removed during scan", entry.path)
+            except OSError:
+                logger.warning("TTL sweep failed to inspect %s", entry.path, exc_info=True)
+
+        deleted = 0
+        for path in expired:
+            try:
+                async with self._io_semaphore:
+                    st = await asyncio.to_thread(os.stat, path, follow_symlinks=False)
+                    if not stat.S_ISREG(st.st_mode):
+                        continue
+                    if st.st_mtime >= cutoff:
+                        continue
+                    await asyncio.to_thread(os.remove, path)
+                    deleted += 1
+            except FileNotFoundError:
+                logger.debug("TTL sweep skipped %s; file already removed", path)
+            except OSError:
+                logger.warning("TTL sweep failed to delete expired file %s", path, exc_info=True)
+
+        return deleted
+
+    async def _sweep_loop(self) -> None:
+        while True:
+            try:
+                cutoff = time.time() - self._ttl_seconds
+                await self._sweep_once(cutoff)
+            except Exception:
+                logger.exception("TTL sweep failed for storage path %s", self.storage_path)
+            await asyncio.sleep(self._sweep_interval_seconds)
+
+    async def start(self) -> None:
+        if self._sweeper_task is None or self._sweeper_task.done():
+            self._sweeper_task = asyncio.create_task(self._sweep_loop())
+
+    async def stop(self) -> None:
+        if self._sweeper_task is None:
+            return
+        self._sweeper_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await self._sweeper_task
+        self._sweeper_task = None
+
+
+def get_storage_manager(storage_config: FileBackend) -> StorageBaseManager[FileStorageHandle]:
+    if isinstance(storage_config, FileBackend):
+        if storage_config.file_ttl is not None and storage_config.ttl_sweep_interval is not None:
+            manager = LocalStorageTTLManager(
+                storage_path=storage_config.path,
+                max_concurrency=storage_config.file_concurrency,
+                ttl_seconds=storage_config.file_ttl,
+                sweep_interval_seconds=storage_config.ttl_sweep_interval,
+            )
+        else:
+            manager = LocalStorageManager(
+                storage_path=storage_config.path, max_concurrency=storage_config.file_concurrency
+            )
+    else:
+        raise ValueError("No supported storage managers")
+
+    return manager
+
+
+STORAGE_MANAGER = get_storage_manager(SERVER_SETTINGS_CONFIG.storage)