diff --git a/docs/serving/videos_api.md b/docs/serving/videos_api.md index 4d156bdf0..95b5dbe77 100644 --- a/docs/serving/videos_api.md +++ b/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 diff --git a/docs/user_guide/examples/online_serving/image_to_video.md b/docs/user_guide/examples/online_serving/image_to_video.md index 781f0c2a5..2dbc57633 100644 --- a/docs/user_guide/examples/online_serving/image_to_video.md +++ b/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 diff --git a/docs/user_guide/examples/online_serving/text_to_video.md b/docs/user_guide/examples/online_serving/text_to_video.md index b918aac19..d4d520d5e 100644 --- a/docs/user_guide/examples/online_serving/text_to_video.md +++ b/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 diff --git a/examples/online_serving/image_to_video/README.md b/examples/online_serving/image_to_video/README.md index 285eeb279..f18cea1c8 100644 --- a/examples/online_serving/image_to_video/README.md +++ b/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 diff --git a/examples/online_serving/text_to_video/README.md b/examples/online_serving/text_to_video/README.md index c01e0602f..35b7f0284 100644 --- a/examples/online_serving/text_to_video/README.md +++ b/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 diff --git a/requirements/common.txt b/requirements/common.txt index 469563d3a..6d9dd63ba 100644 --- a/requirements/common.txt +++ b/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 diff --git a/tests/config/test_server_settings.py b/tests/config/test_server_settings.py new file mode 100644 index 000000000..53c15524c --- /dev/null +++ b/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" diff --git a/tests/entrypoints/openai_api/test_storage_manager.py b/tests/entrypoints/openai_api/test_storage_manager.py new file mode 100644 index 000000000..30e505766 --- /dev/null +++ b/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) diff --git a/tests/entrypoints/openai_api/test_video_server.py b/tests/entrypoints/openai_api/test_video_server.py index b54afd54a..dbf766391 100644 --- a/tests/entrypoints/openai_api/test_video_server.py +++ b/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 diff --git a/vllm_omni/config/server_settings.py b/vllm_omni/config/server_settings.py new file mode 100644 index 000000000..bb63712ba --- /dev/null +++ b/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() diff --git a/vllm_omni/entrypoints/openai/api_server.py b/vllm_omni/entrypoints/openai/api_server.py index 627a10705..35b9dc20f 100755 --- a/vllm_omni/entrypoints/openai/api_server.py +++ b/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") diff --git a/vllm_omni/entrypoints/openai/storage.py b/vllm_omni/entrypoints/openai/storage.py index 52aef8948..a36688956 100644 --- a/vllm_omni/entrypoints/openai/storage.py +++ b/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)