Run video generation as a background job so secure mode's tunnel cap cannot 524 it
POST /video/generate previously held the response open for the whole generation (multi-minute for 720p), so in --secure mode the Cloudflare quick tunnel's ~100s origin-response cap returned a 524 while the server kept generating, and the frontend treated the run as failed. Generation now follows the same return-at-once pattern as /video/load: begin_generate validates synchronously (409 on no model or on a second concurrent generate via a new busy sentinel) and runs the existing generate + gallery-persist pipeline, with the route's exact error mapping, on a daemon thread. GET /video/generate-progress gains optional terminal fields: phase completed carries the saved gallery record, phase failed a client-safe error; active only drops together with a terminal phase. The cancel event is registered before the worker starts so /video/generate/cancel keeps working across the whole job. VideoGenerateResponse becomes an accepted acknowledgement (status started, video kept as an always-null compat field). The video page fires the POST, then drives completion off the progress poll it already runs (completed prepends the clip, failed surfaces the error, the cancelled sentinel stays toast-free). The API-key training-start guards now also probe the video backend for an in-flight background clip, since it is no longer visible as an in-flight HTTP request to the keep-warm counter. Route tests keep the fake backend for load/generate/status but inherit the real job machinery, covering immediate accept, concurrent 409, the terminal completed record, sanitized/ValueError/cancelled failures, and cancel of a running job.
This commit is contained in:
Daniel Han
parent
783c0c1af6
commit
daaac9e10b
9 changed files with 454 additions and 127 deletions
|
|
@ -61,12 +61,15 @@ _INFERENCE_SUFFIXES = (
|
|||
"/responses",
|
||||
"/generate/stream", # Studio's own streaming route on the same llama-server
|
||||
"/audio/generate", # direct GGUF TTS; can outlive the idle TTL
|
||||
# Image/video generation holds a multi-GB diffusion/video pipeline for the whole request.
|
||||
# Tracking them here lets other_inference_request_count() see an in-flight generation, so an
|
||||
# Image generation holds a multi-GB diffusion pipeline for the whole request.
|
||||
# Tracking it here lets other_inference_request_count() see an in-flight generation, so an
|
||||
# API-key training start is refused (409) before its unload cancels the generation. endswith
|
||||
# so the GET *-progress and */cancel variants are not matched.
|
||||
"/images/generate", # /api/inference/images/generate
|
||||
"/images/generations", # /v1/images/generations (+ /api/inference/images/generations)
|
||||
# Video generation runs as a background job (the POST returns at once), so this entry only
|
||||
# covers the brief accept request; the training-start guards additionally probe the video
|
||||
# backend's generate-progress for an in-flight background clip.
|
||||
"/video/generate", # /api/inference/video/generate
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -81,6 +81,7 @@ from .diffusion_transformer_quant import (
|
|||
from .diffusion_precision import normalize_te_quant, quantize_text_encoders
|
||||
from .video_families import (
|
||||
VIDEO_CANCELLED_MSG,
|
||||
VIDEO_GENERATION_BUSY_MSG,
|
||||
VIDEO_NOT_LOADED_MSG,
|
||||
VideoFamily,
|
||||
default_video_generation_params,
|
||||
|
|
@ -381,6 +382,10 @@ class VideoBackend:
|
|||
self._active_generate_cancel: Optional[threading.Event] = None
|
||||
# Generation progress, written by the step callback / phase transitions.
|
||||
self._gen: dict[str, Any] = {"active": False}
|
||||
# True from begin_generate() until its worker records a terminal state, so
|
||||
# a second begin_generate() is refused while the first still runs (or is
|
||||
# about to run: generate() only sets _gen after taking its locks).
|
||||
self._generate_job_active = False
|
||||
|
||||
# ── validation ───────────────────────────────────────────────────────────
|
||||
|
||||
|
|
@ -1474,6 +1479,164 @@ class VideoBackend:
|
|||
except Exception: # noqa: BLE001 -- reset is best-effort, never fail a generation
|
||||
pass
|
||||
|
||||
def begin_generate(
|
||||
self,
|
||||
*,
|
||||
prompt: str,
|
||||
negative_prompt: Optional[str] = None,
|
||||
width: Optional[int] = None,
|
||||
height: Optional[int] = None,
|
||||
num_frames: Optional[int] = None,
|
||||
fps: Optional[int] = None,
|
||||
steps: Optional[int] = None,
|
||||
guidance: Optional[float] = None,
|
||||
guidance_2: Optional[float] = None,
|
||||
seed: Optional[int] = None,
|
||||
) -> None:
|
||||
"""Validate cheaply, then run generate + gallery persist on a daemon thread.
|
||||
|
||||
Returns at once, mirroring begin_load: a clip takes minutes to denoise, and
|
||||
a proxy in front of Studio (secure mode's Cloudflare tunnel) caps the origin
|
||||
response window near 100 seconds, so the HTTP call must not span the
|
||||
generation. The terminal outcome (phase "completed" with the saved gallery
|
||||
record, or "failed" with a client-safe error) is reported by
|
||||
generate_progress(); cancel_generate() keeps working against the job.
|
||||
Raises RuntimeError with VIDEO_NOT_LOADED_MSG / VIDEO_GENERATION_BUSY_MSG
|
||||
sentinels the route maps to 409.
|
||||
"""
|
||||
cancel = threading.Event()
|
||||
with self._lock:
|
||||
if self._state is None:
|
||||
raise RuntimeError(VIDEO_NOT_LOADED_MSG)
|
||||
if self._generate_job_active:
|
||||
raise RuntimeError(VIDEO_GENERATION_BUSY_MSG)
|
||||
self._generate_job_active = True
|
||||
# Register the cancel event BEFORE the worker starts so a cancel (or an
|
||||
# unload) that lands in the spawn window still stops the run instead of
|
||||
# returning "nothing to cancel".
|
||||
self._active_generate_cancel = cancel
|
||||
self._gen = {
|
||||
"active": True,
|
||||
"phase": "queued",
|
||||
"step": 0,
|
||||
"total": 0,
|
||||
"eta_seconds": None,
|
||||
}
|
||||
threading.Thread(
|
||||
target = self._run_generate,
|
||||
kwargs = dict(
|
||||
prompt = prompt,
|
||||
negative_prompt = negative_prompt,
|
||||
width = width,
|
||||
height = height,
|
||||
num_frames = num_frames,
|
||||
fps = fps,
|
||||
steps = steps,
|
||||
guidance = guidance,
|
||||
guidance_2 = guidance_2,
|
||||
seed = seed,
|
||||
cancel_event = cancel,
|
||||
),
|
||||
daemon = True,
|
||||
).start()
|
||||
|
||||
def _run_generate(self, *, cancel_event: threading.Event, **gen_kwargs: Any) -> None:
|
||||
"""begin_generate's worker: generate, persist to the gallery, record the
|
||||
terminal state where generate_progress() reports it. The error mapping is
|
||||
the exact one the route applied when the call was synchronous: ValueError
|
||||
text is client input feedback, sentinel RuntimeErrors pass through, and any
|
||||
other failure is logged server-side and reported as a generic message so
|
||||
internals (CUDA state, paths) never reach the client."""
|
||||
from . import video_gallery
|
||||
|
||||
try:
|
||||
result = self.generate(cancel_event = cancel_event, **gen_kwargs)
|
||||
except ValueError as exc:
|
||||
self._finish_generate_job(cancel_event = cancel_event, error = str(exc))
|
||||
return
|
||||
except RuntimeError as exc:
|
||||
msg = str(exc)
|
||||
if msg not in (VIDEO_NOT_LOADED_MSG, VIDEO_CANCELLED_MSG):
|
||||
logger.error("video.generate_failed: %s", exc, exc_info = True)
|
||||
msg = "Video generation failed."
|
||||
self._finish_generate_job(cancel_event = cancel_event, error = msg)
|
||||
return
|
||||
except Exception as exc: # noqa: BLE001 -- worker thread: never propagate
|
||||
logger.error("video.generate_failed: %s", exc, exc_info = True)
|
||||
self._finish_generate_job(cancel_event = cancel_event, error = "Video generation failed.")
|
||||
return
|
||||
|
||||
# Persist the clip with its full recipe as the JSON sidecar the gallery reads back.
|
||||
created_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||||
try:
|
||||
record = video_gallery.save(
|
||||
result["mp4_bytes"],
|
||||
{
|
||||
"prompt": gen_kwargs["prompt"],
|
||||
"negative_prompt": gen_kwargs.get("negative_prompt"),
|
||||
"width": result["width"],
|
||||
"height": result["height"],
|
||||
"num_frames": result["num_frames"],
|
||||
"fps": result["fps"],
|
||||
"duration_s": result["duration_s"],
|
||||
"steps": result["steps"],
|
||||
"guidance": result["guidance"],
|
||||
"guidance_2": gen_kwargs.get("guidance_2"),
|
||||
"seed": result["seed"],
|
||||
"has_audio": result["has_audio"],
|
||||
"model": result["repo_id"],
|
||||
"created_at": created_at,
|
||||
},
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 -- disk failure must reach the poller
|
||||
logger.error("video.persist_failed: %s", exc)
|
||||
self._finish_generate_job(
|
||||
cancel_event = cancel_event, error = "Failed to save the generated video."
|
||||
)
|
||||
return
|
||||
self._finish_generate_job(
|
||||
cancel_event = cancel_event, video = record, total = result["steps"]
|
||||
)
|
||||
|
||||
def _finish_generate_job(
|
||||
self,
|
||||
*,
|
||||
cancel_event: Optional[threading.Event] = None,
|
||||
video: Optional[dict] = None,
|
||||
error: Optional[str] = None,
|
||||
total: int = 0,
|
||||
) -> None:
|
||||
"""Record a job's terminal state as one atomic swap. The terminal dict
|
||||
replaces the live-progress one so a poll can never mix fields from both,
|
||||
and the busy flag drops in the same critical section so the earliest
|
||||
moment a new begin_generate() can start is after the outcome is visible."""
|
||||
with self._lock:
|
||||
self._generate_job_active = False
|
||||
if cancel_event is not None and self._active_generate_cancel is cancel_event:
|
||||
# generate() clears its own registration; this covers a job whose
|
||||
# worker failed before (or without) reaching generate()'s finally.
|
||||
# Identity-guarded so a direct generate() that registered its own
|
||||
# event in the meantime keeps its cancel handle.
|
||||
self._active_generate_cancel = None
|
||||
if error is not None:
|
||||
self._gen = {
|
||||
"active": False,
|
||||
"phase": "failed",
|
||||
"error": error,
|
||||
"step": 0,
|
||||
"total": 0,
|
||||
"eta_seconds": None,
|
||||
}
|
||||
else:
|
||||
self._gen = {
|
||||
"active": False,
|
||||
"phase": "completed",
|
||||
"video": video,
|
||||
"step": total,
|
||||
"total": total,
|
||||
"eta_seconds": None,
|
||||
}
|
||||
|
||||
def generate(
|
||||
self,
|
||||
*,
|
||||
|
|
@ -1487,9 +1650,12 @@ class VideoBackend:
|
|||
guidance: Optional[float] = None,
|
||||
guidance_2: Optional[float] = None,
|
||||
seed: Optional[int] = None,
|
||||
cancel_event: Optional[threading.Event] = None,
|
||||
) -> dict[str, Any]:
|
||||
import torch
|
||||
cancel = threading.Event()
|
||||
# begin_generate passes the event it already registered (so a cancel in the
|
||||
# spawn window is honoured); a direct call makes its own.
|
||||
cancel = cancel_event if cancel_event is not None else threading.Event()
|
||||
with self._generate_lock:
|
||||
with self._lock:
|
||||
state = self._state
|
||||
|
|
@ -1710,7 +1876,14 @@ class VideoBackend:
|
|||
pass
|
||||
|
||||
def generate_progress(self) -> dict[str, Any]:
|
||||
gen = dict(self._gen)
|
||||
with self._lock:
|
||||
gen = dict(self._gen)
|
||||
# generate() swaps in a bare {"active": False} on its own exit paths
|
||||
# before the job worker records the terminal dict; report the job as
|
||||
# still active across that gap so a poller only sees active drop
|
||||
# together with a terminal phase ("completed" / "failed").
|
||||
if self._generate_job_active:
|
||||
gen["active"] = True
|
||||
gen.setdefault("active", False)
|
||||
return gen
|
||||
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ from typing import Optional
|
|||
# these EXACTLY to return 409 (client-recoverable) instead of a sanitized 500.
|
||||
VIDEO_NOT_LOADED_MSG = "No video model is loaded."
|
||||
VIDEO_CANCELLED_MSG = "Video generation was cancelled."
|
||||
VIDEO_GENERATION_BUSY_MSG = "A video generation is already in progress."
|
||||
|
||||
|
||||
@dataclass(frozen = True)
|
||||
|
|
|
|||
|
|
@ -2499,9 +2499,21 @@ class GalleryVideo(BaseModel):
|
|||
|
||||
|
||||
class VideoGenerateResponse(BaseModel):
|
||||
"""The persisted gallery record for one generation call."""
|
||||
"""Acknowledgement that a generation was accepted and started.
|
||||
|
||||
video: GalleryVideo = Field(..., description = "Saved record for the generated clip")
|
||||
Generation runs as a background job (a clip takes minutes, and secure mode's
|
||||
tunnel caps the origin response window near 100 seconds, so the POST cannot
|
||||
span it). The saved gallery record arrives via GET /video/generate-progress
|
||||
when its phase reaches "completed"."""
|
||||
|
||||
status: Literal["started"] = Field(
|
||||
"started", description = "Discriminator: the generation job was started"
|
||||
)
|
||||
video: Optional[GalleryVideo] = Field(
|
||||
None,
|
||||
description = "Always null (kept for response-shape compatibility); the saved "
|
||||
"record is delivered by generate-progress on completion",
|
||||
)
|
||||
|
||||
|
||||
class VideoGalleryListResponse(BaseModel):
|
||||
|
|
@ -2512,13 +2524,23 @@ class VideoGalleryListResponse(BaseModel):
|
|||
|
||||
|
||||
class VideoGenerateProgressResponse(BaseModel):
|
||||
"""Live progress for an in-flight video generation."""
|
||||
"""Live progress for an in-flight video generation, plus the terminal outcome
|
||||
of the background job POST /video/generate started."""
|
||||
|
||||
active: bool = Field(False, description = "Whether a generation is running")
|
||||
phase: Optional[str] = Field(None, description = "Current phase: denoise | export | null")
|
||||
phase: Optional[str] = Field(
|
||||
None,
|
||||
description = "Current phase: queued | denoise | export | completed | failed | null",
|
||||
)
|
||||
step: int = Field(0, description = "Denoising steps completed so far")
|
||||
total: int = Field(0, description = "Total denoising steps for this run")
|
||||
eta_seconds: Optional[float] = Field(None, description = "Estimated seconds remaining")
|
||||
video: Optional[GalleryVideo] = Field(
|
||||
None, description = "Saved gallery record when phase is 'completed'"
|
||||
)
|
||||
error: Optional[str] = Field(
|
||||
None, description = "Client-safe failure detail when phase is 'failed'"
|
||||
)
|
||||
|
||||
|
||||
class VideoLoadProgressResponse(BaseModel):
|
||||
|
|
|
|||
|
|
@ -134,6 +134,21 @@ async def get_visible_hardware_utilization(current_subject: str = Depends(get_cu
|
|||
return get_visible_gpu_utilization()
|
||||
|
||||
|
||||
def _background_video_generation_active() -> bool:
|
||||
"""Whether a video clip is generating on the video backend's worker thread.
|
||||
|
||||
POST /video/generate returns at once and generates in the background, so an
|
||||
in-flight clip is invisible to the keep-warm in-flight request count the
|
||||
API-key training guards consult; ask the backend directly. Best-effort: a
|
||||
probe failure must never block a training start."""
|
||||
try:
|
||||
from core.inference.video import get_video_backend
|
||||
return bool(get_video_backend().generate_progress().get("active"))
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning("Could not check video generation state for training guard: %s", e)
|
||||
return False
|
||||
|
||||
|
||||
@router.post("/start")
|
||||
async def start_training(
|
||||
request: TrainingStartRequest,
|
||||
|
|
@ -156,7 +171,10 @@ async def start_training(
|
|||
# session is not yet special-cased.)
|
||||
if via_api_key is True:
|
||||
from core.inference.llama_keepwarm import other_inference_request_count
|
||||
if other_inference_request_count(current_request_counted = False) > 0:
|
||||
if (
|
||||
other_inference_request_count(current_request_counted = False) > 0
|
||||
or _background_video_generation_active()
|
||||
):
|
||||
raise HTTPException(
|
||||
status_code = 409,
|
||||
detail = (
|
||||
|
|
@ -1261,7 +1279,10 @@ async def start_diffusion_training(
|
|||
# a diffusion start cannot silently drop an active API inference request.
|
||||
if via_api_key is True:
|
||||
from core.inference.llama_keepwarm import other_inference_request_count
|
||||
if other_inference_request_count(current_request_counted = False) > 0:
|
||||
if (
|
||||
other_inference_request_count(current_request_counted = False) > 0
|
||||
or _background_video_generation_active()
|
||||
):
|
||||
raise HTTPException(
|
||||
status_code = 409,
|
||||
detail = (
|
||||
|
|
|
|||
|
|
@ -8,15 +8,16 @@ these routes mirror the /images/* routes one-for-one: the same validate-before-e
|
|||
load ordering, the same GPU arbiter handoff (VIDEO owner in place of DIFFUSION),
|
||||
the same error boundary mapping backend exceptions to HTTP, and the same gallery
|
||||
CRUD shape. The backend runs in-process and is synchronous, so the blocking
|
||||
load/generate/unload calls are offloaded with asyncio.to_thread to keep the event
|
||||
loop free. This module is the single error boundary: backend methods raise, we
|
||||
map to HTTP here.
|
||||
calls are offloaded with asyncio.to_thread to keep the event loop free; the slow
|
||||
operations (load AND generate) run as background jobs whose begin_* calls return
|
||||
at once, with progress + terminal outcome polled from their *-progress routes.
|
||||
This module is the single error boundary: backend methods raise, we map to HTTP
|
||||
here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
from pydantic import ValidationError
|
||||
|
|
@ -145,14 +146,18 @@ async def video_load_progress(current_subject: str = Depends(get_current_subject
|
|||
async def generate_video(
|
||||
request: VideoGenerateRequest, current_subject: str = Depends(get_current_subject)
|
||||
):
|
||||
from core.inference import video_gallery
|
||||
"""Start a generation job and return at once (the begin_load pattern): a clip
|
||||
takes minutes, and secure mode's tunnel caps the origin response window near
|
||||
100 seconds, so the response must not span the generation. The worker runs the
|
||||
generate + gallery-persist pipeline; the terminal outcome (completed with the
|
||||
saved record / failed with a client-safe error) arrives via generate-progress."""
|
||||
from core.inference.video import get_video_backend
|
||||
from core.inference.video_families import VIDEO_CANCELLED_MSG, VIDEO_NOT_LOADED_MSG
|
||||
from core.inference.video_families import VIDEO_GENERATION_BUSY_MSG, VIDEO_NOT_LOADED_MSG
|
||||
|
||||
backend = get_video_backend()
|
||||
try:
|
||||
result = await asyncio.to_thread(
|
||||
backend.generate,
|
||||
await asyncio.to_thread(
|
||||
backend.begin_generate,
|
||||
prompt = request.prompt,
|
||||
negative_prompt = request.negative_prompt,
|
||||
width = request.width,
|
||||
|
|
@ -165,53 +170,19 @@ async def generate_video(
|
|||
seed = request.seed,
|
||||
)
|
||||
except ValueError as exc:
|
||||
# Bad client input (a workflow the loaded family doesn't support) -- a 400 with
|
||||
# the reason, not a generic 500.
|
||||
# Bad client input -- a 400 with the reason, not a generic 500.
|
||||
raise HTTPException(status_code = 400, detail = str(exc))
|
||||
except RuntimeError as exc:
|
||||
# Only "no model loaded" / user-cancelled are client-state (409). Match the
|
||||
# sentinels exactly, not as a substring, so an execution failure that merely
|
||||
# contains "cancelled" can't misroute to 409 and leak that output.
|
||||
# Only "no model loaded" / "already generating" are client-state (409).
|
||||
# Match the sentinels exactly, not as a substring, so an unrelated failure
|
||||
# can't misroute to 409 and leak its message.
|
||||
msg = str(exc)
|
||||
if msg in (VIDEO_NOT_LOADED_MSG, VIDEO_CANCELLED_MSG):
|
||||
if msg in (VIDEO_NOT_LOADED_MSG, VIDEO_GENERATION_BUSY_MSG):
|
||||
raise HTTPException(status_code = 409, detail = msg)
|
||||
logger.error("video.generate_failed: %s", exc, exc_info = True)
|
||||
raise HTTPException(status_code = 500, detail = "Video generation failed.")
|
||||
except Exception as exc:
|
||||
logger.error("video.generate_failed: %s", exc, exc_info = True)
|
||||
raise HTTPException(status_code = 500, detail = "Video generation failed.")
|
||||
|
||||
# Persist the clip with its full recipe as the JSON sidecar the gallery reads back.
|
||||
created_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||||
|
||||
def _persist() -> dict:
|
||||
return video_gallery.save(
|
||||
result["mp4_bytes"],
|
||||
{
|
||||
"prompt": request.prompt,
|
||||
"negative_prompt": request.negative_prompt,
|
||||
"width": result["width"],
|
||||
"height": result["height"],
|
||||
"num_frames": result["num_frames"],
|
||||
"fps": result["fps"],
|
||||
"duration_s": result["duration_s"],
|
||||
"steps": result["steps"],
|
||||
"guidance": result["guidance"],
|
||||
"guidance_2": request.guidance_2,
|
||||
"seed": result["seed"],
|
||||
"has_audio": result["has_audio"],
|
||||
"model": result["repo_id"],
|
||||
"created_at": created_at,
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
record = await asyncio.to_thread(_persist)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.error("video.persist_failed: %s", exc)
|
||||
raise HTTPException(status_code = 500, detail = "Failed to save the generated video.")
|
||||
|
||||
return VideoGenerateResponse(video = GalleryVideo(**record))
|
||||
return VideoGenerateResponse()
|
||||
|
||||
|
||||
@router.get("/video/generate-progress", response_model = VideoGenerateProgressResponse)
|
||||
|
|
|
|||
|
|
@ -12,6 +12,9 @@ video_gallery code.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
|
@ -20,7 +23,11 @@ import core.inference.gpu_arbiter as gpu_arbiter
|
|||
import core.inference.video as video_module
|
||||
import core.inference.video_gallery as gallery_module
|
||||
from auth.authentication import get_current_subject
|
||||
from core.inference.video_families import VIDEO_CANCELLED_MSG, VIDEO_NOT_LOADED_MSG
|
||||
from core.inference.video_families import (
|
||||
VIDEO_CANCELLED_MSG,
|
||||
VIDEO_GENERATION_BUSY_MSG,
|
||||
VIDEO_NOT_LOADED_MSG,
|
||||
)
|
||||
from routes.video import router as video_router
|
||||
|
||||
|
||||
|
|
@ -60,14 +67,29 @@ def _unloaded_status():
|
|||
}
|
||||
|
||||
|
||||
class _FakeBackend:
|
||||
class _FakeBackend(video_module.VideoBackend):
|
||||
"""Overrides the heavy load/generate/status surface but INHERITS the real
|
||||
begin_generate / _run_generate / generate_progress / cancel_generate job
|
||||
machinery, so the asynchronous generate contract (immediate accept, busy
|
||||
guard, terminal completed/failed state, cancel) is exercised for real."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.loaded = False
|
||||
super().__init__()
|
||||
self.last_load_kwargs: dict = {}
|
||||
# Repo ids of in-flight (not yet committed) loads; empty tuple = none. The unload
|
||||
# route reads this to keep VIDEO ownership while a concurrent load is still loading.
|
||||
self.loading: tuple = ()
|
||||
|
||||
# The real backend keys "loaded" off its committed pipeline state (_state); map
|
||||
# the fake's flag onto it so the inherited begin_generate sees the same thing.
|
||||
@property
|
||||
def loaded(self) -> bool:
|
||||
return self._state is not None
|
||||
|
||||
@loaded.setter
|
||||
def loaded(self, value: bool) -> None:
|
||||
self._state = object() if value else None
|
||||
|
||||
def loading_repo_ids(self) -> tuple:
|
||||
return tuple(self.loading)
|
||||
|
||||
|
|
@ -130,6 +152,7 @@ class _FakeBackend:
|
|||
*,
|
||||
prompt,
|
||||
seed = None,
|
||||
cancel_event = None,
|
||||
**kwargs,
|
||||
):
|
||||
if not self.loaded:
|
||||
|
|
@ -148,12 +171,6 @@ class _FakeBackend:
|
|||
"guidance": 4.0 if kwargs.get("guidance") is None else kwargs.get("guidance"),
|
||||
}
|
||||
|
||||
def generate_progress(self):
|
||||
return {"active": False}
|
||||
|
||||
def cancel_generate(self):
|
||||
return False
|
||||
|
||||
def unload(self):
|
||||
self.loaded = False
|
||||
return _unloaded_status()
|
||||
|
|
@ -203,6 +220,33 @@ def client(monkeypatch, tmp_path):
|
|||
return TestClient(app)
|
||||
|
||||
|
||||
def _wait_terminal(client, timeout = 5.0) -> dict:
|
||||
"""Poll generate-progress until the background job records a terminal phase.
|
||||
Generation is asynchronous now (the POST returns as soon as the job starts),
|
||||
so its outcome is only observable here."""
|
||||
deadline = time.monotonic() + timeout
|
||||
progress: dict = {}
|
||||
while time.monotonic() < deadline:
|
||||
progress = client.get("/api/inference/video/generate-progress").json()
|
||||
if progress.get("phase") in ("completed", "failed"):
|
||||
return progress
|
||||
time.sleep(0.01)
|
||||
raise AssertionError(f"generation never reached a terminal state: {progress}")
|
||||
|
||||
|
||||
def _generate_and_wait(client, payload) -> dict:
|
||||
"""Start a generation, assert the immediate accepted response, and return the
|
||||
saved gallery record the completed progress state carries."""
|
||||
resp = client.post("/api/inference/video/generate", json = payload)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["status"] == "started" and body["video"] is None
|
||||
progress = _wait_terminal(client)
|
||||
assert progress["phase"] == "completed", progress
|
||||
assert progress["active"] is False and progress["error"] is None
|
||||
return progress["video"]
|
||||
|
||||
|
||||
def test_load_happy_path_and_arbiter_acquired(client, monkeypatch):
|
||||
# Force the device to cuda so the load takes the GPU arbiter, and record the acquire.
|
||||
import types
|
||||
|
|
@ -277,11 +321,10 @@ def test_load_threads_transformer_quant_and_guidance_2(client):
|
|||
kwargs = video_module.get_video_backend().last_load_kwargs
|
||||
assert kwargs.get("transformer_quant") == "fp8"
|
||||
|
||||
gen = client.post(
|
||||
"/api/inference/video/generate",
|
||||
json = {"prompt": "a sloth", "guidance": 5.0, "guidance_2": 3.0},
|
||||
video = _generate_and_wait(
|
||||
client, {"prompt": "a sloth", "guidance": 5.0, "guidance_2": 3.0}
|
||||
)
|
||||
assert gen.status_code == 200
|
||||
assert video["guidance"] == 5.0 and video["guidance_2"] == 3.0
|
||||
|
||||
|
||||
def test_load_rejects_bad_transformer_quant_422(client):
|
||||
|
|
@ -337,16 +380,14 @@ def test_load_progress_route(client):
|
|||
assert ready.json()["phase"] == "ready"
|
||||
|
||||
|
||||
def test_generate_happy_path_persists_and_returns_record(client):
|
||||
def test_generate_happy_path_persists_and_reports_record(client):
|
||||
client.post(
|
||||
"/api/inference/video/load",
|
||||
json = {"model_path": "unsloth/LTX-2.3-GGUF", "gguf_filename": "q.gguf"},
|
||||
)
|
||||
gen = client.post(
|
||||
"/api/inference/video/generate", json = {"prompt": "a sloth surfing", "seed": 7}
|
||||
)
|
||||
assert gen.status_code == 200
|
||||
video = gen.json()["video"]
|
||||
# The POST returns at once ("started"); the saved record arrives through the
|
||||
# generate-progress terminal state (asserted inside the helper).
|
||||
video = _generate_and_wait(client, {"prompt": "a sloth surfing", "seed": 7})
|
||||
assert video["seed"] == 7 and video["prompt"] == "a sloth surfing" and video["id"]
|
||||
assert video["has_audio"] is True
|
||||
assert video["model"] == "unsloth/LTX-2.3-GGUF"
|
||||
|
|
@ -369,7 +410,9 @@ def test_generate_without_load_returns_409(client):
|
|||
assert resp.json()["detail"] == VIDEO_NOT_LOADED_MSG
|
||||
|
||||
|
||||
def test_generate_cancelled_returns_409(client, monkeypatch):
|
||||
def test_generate_cancelled_reports_failed_with_sentinel(client, monkeypatch):
|
||||
# A cancel mid-run surfaces as the job's terminal failed state carrying the exact
|
||||
# sentinel (the frontend suppresses the toast on it), not as an HTTP error.
|
||||
backend = video_module.get_video_backend()
|
||||
backend.loaded = True
|
||||
|
||||
|
|
@ -378,13 +421,16 @@ def test_generate_cancelled_returns_409(client, monkeypatch):
|
|||
|
||||
monkeypatch.setattr(backend, "generate", _cancel)
|
||||
resp = client.post("/api/inference/video/generate", json = {"prompt": "p"})
|
||||
assert resp.status_code == 409
|
||||
assert resp.json()["detail"] == VIDEO_CANCELLED_MSG
|
||||
assert resp.status_code == 200
|
||||
progress = _wait_terminal(client)
|
||||
assert progress["phase"] == "failed"
|
||||
assert progress["error"] == VIDEO_CANCELLED_MSG
|
||||
assert progress["active"] is False
|
||||
|
||||
|
||||
def test_generate_pipeline_error_returns_sanitized_500(client, monkeypatch):
|
||||
# A loaded model that fails mid-pipeline (CUDA OOM) is a server failure: 500 with a
|
||||
# generic message, not a 409 echoing the raw exception.
|
||||
def test_generate_pipeline_error_reports_sanitized_failure(client, monkeypatch):
|
||||
# A loaded model that fails mid-pipeline (CUDA OOM) is a server failure: the job's
|
||||
# terminal state carries a generic message, never the raw exception.
|
||||
backend = video_module.get_video_backend()
|
||||
backend.loaded = True
|
||||
|
||||
|
|
@ -393,12 +439,15 @@ def test_generate_pipeline_error_returns_sanitized_500(client, monkeypatch):
|
|||
|
||||
monkeypatch.setattr(backend, "generate", _oom)
|
||||
resp = client.post("/api/inference/video/generate", json = {"prompt": "p"})
|
||||
assert resp.status_code == 500
|
||||
assert resp.json()["detail"] == "Video generation failed."
|
||||
assert "CUDA" not in resp.json()["detail"]
|
||||
assert resp.status_code == 200
|
||||
progress = _wait_terminal(client)
|
||||
assert progress["phase"] == "failed"
|
||||
assert progress["error"] == "Video generation failed."
|
||||
assert "CUDA" not in progress["error"]
|
||||
|
||||
|
||||
def test_generate_value_error_returns_400(client, monkeypatch):
|
||||
def test_generate_value_error_reports_reason(client, monkeypatch):
|
||||
# Bad client input is feedback: the terminal failed state carries the reason.
|
||||
backend = video_module.get_video_backend()
|
||||
backend.loaded = True
|
||||
|
||||
|
|
@ -407,14 +456,48 @@ def test_generate_value_error_returns_400(client, monkeypatch):
|
|||
|
||||
monkeypatch.setattr(backend, "generate", _bad)
|
||||
resp = client.post("/api/inference/video/generate", json = {"prompt": "p"})
|
||||
assert resp.status_code == 400
|
||||
assert "not supported" in resp.json()["detail"]
|
||||
assert resp.status_code == 200
|
||||
progress = _wait_terminal(client)
|
||||
assert progress["phase"] == "failed"
|
||||
assert "not supported" in progress["error"]
|
||||
|
||||
|
||||
def test_generate_concurrent_second_returns_409(client, monkeypatch):
|
||||
# While a job is running, a second generate is refused synchronously with the busy
|
||||
# sentinel; the first job still completes and persists once released.
|
||||
backend = video_module.get_video_backend()
|
||||
backend.loaded = True
|
||||
release = threading.Event()
|
||||
real_generate = _FakeBackend.generate
|
||||
|
||||
def _slow(**kwargs):
|
||||
assert release.wait(5)
|
||||
return real_generate(backend, **kwargs)
|
||||
|
||||
monkeypatch.setattr(backend, "generate", _slow)
|
||||
first = client.post("/api/inference/video/generate", json = {"prompt": "a", "seed": 1})
|
||||
assert first.status_code == 200 and first.json()["status"] == "started"
|
||||
|
||||
second = client.post("/api/inference/video/generate", json = {"prompt": "b"})
|
||||
assert second.status_code == 409
|
||||
assert second.json()["detail"] == VIDEO_GENERATION_BUSY_MSG
|
||||
|
||||
running = client.get("/api/inference/video/generate-progress").json()
|
||||
assert running["active"] is True
|
||||
|
||||
release.set()
|
||||
progress = _wait_terminal(client)
|
||||
assert progress["phase"] == "completed" and progress["video"]["seed"] == 1
|
||||
# With the job finished, a new generate is accepted again.
|
||||
assert _generate_and_wait(client, {"prompt": "c", "seed": 2})["seed"] == 2
|
||||
|
||||
|
||||
def test_generate_progress_route(client):
|
||||
resp = client.get("/api/inference/video/generate-progress")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["active"] is False
|
||||
body = resp.json()
|
||||
assert body["active"] is False
|
||||
assert body["phase"] is None and body["video"] is None and body["error"] is None
|
||||
|
||||
|
||||
def test_cancel_generation_route(client):
|
||||
|
|
@ -423,6 +506,31 @@ def test_cancel_generation_route(client):
|
|||
assert resp.json()["cancelled"] is False
|
||||
|
||||
|
||||
def test_cancel_running_job(client, monkeypatch):
|
||||
# Cancel still works against the background job: begin_generate registers the
|
||||
# cancel event before the worker starts, so the cancel route reports True at
|
||||
# once and the job lands in the failed(cancelled) terminal state.
|
||||
backend = video_module.get_video_backend()
|
||||
backend.loaded = True
|
||||
|
||||
def _wait_for_cancel(*, cancel_event = None, **kwargs):
|
||||
assert cancel_event is not None and cancel_event.wait(5)
|
||||
raise RuntimeError(VIDEO_CANCELLED_MSG)
|
||||
|
||||
monkeypatch.setattr(backend, "generate", _wait_for_cancel)
|
||||
resp = client.post("/api/inference/video/generate", json = {"prompt": "p"})
|
||||
assert resp.status_code == 200
|
||||
|
||||
cancelled = client.post("/api/inference/video/generate/cancel")
|
||||
assert cancelled.status_code == 200 and cancelled.json()["cancelled"] is True
|
||||
|
||||
progress = _wait_terminal(client)
|
||||
assert progress["phase"] == "failed"
|
||||
assert progress["error"] == VIDEO_CANCELLED_MSG
|
||||
# Nothing was persisted for the cancelled run.
|
||||
assert client.get("/api/inference/video/gallery").json()["videos"] == []
|
||||
|
||||
|
||||
def test_file_endpoint_404_for_bad_id(client):
|
||||
resp = client.get("/api/inference/video/gallery/does-not-exist/file")
|
||||
assert resp.status_code == 404
|
||||
|
|
@ -433,8 +541,8 @@ def test_delete_and_clear(client):
|
|||
"/api/inference/video/load",
|
||||
json = {"model_path": "unsloth/LTX-2.3-GGUF", "gguf_filename": "q.gguf"},
|
||||
)
|
||||
first = client.post("/api/inference/video/generate", json = {"prompt": "a"}).json()["video"]
|
||||
second = client.post("/api/inference/video/generate", json = {"prompt": "b"}).json()["video"]
|
||||
first = _generate_and_wait(client, {"prompt": "a"})
|
||||
second = _generate_and_wait(client, {"prompt": "b"})
|
||||
assert len(client.get("/api/inference/video/gallery").json()["videos"]) == 2
|
||||
|
||||
# Delete one, then confirm it 404s and the other remains.
|
||||
|
|
@ -455,7 +563,7 @@ def test_gallery_pagination(client):
|
|||
json = {"model_path": "unsloth/LTX-2.3-GGUF", "gguf_filename": "q.gguf"},
|
||||
)
|
||||
for i in range(5):
|
||||
client.post("/api/inference/video/generate", json = {"prompt": f"clip {i}", "seed": i})
|
||||
_generate_and_wait(client, {"prompt": f"clip {i}", "seed": i})
|
||||
page1 = client.get("/api/inference/video/gallery?limit=2&offset=0").json()
|
||||
assert len(page1["videos"]) == 2 and page1["has_more"] is True
|
||||
last = client.get("/api/inference/video/gallery?limit=2&offset=4").json()
|
||||
|
|
@ -571,7 +679,7 @@ def test_export_endpoint_validation(client, monkeypatch):
|
|||
"/api/inference/video/load",
|
||||
json = {"model_path": "unsloth/LTX-2.3-GGUF", "gguf_filename": "q.gguf"},
|
||||
)
|
||||
video = client.post("/api/inference/video/generate", json = {"prompt": "a"}).json()["video"]
|
||||
video = _generate_and_wait(client, {"prompt": "a"})
|
||||
resp = client.get(f"/api/inference/video/gallery/{video['id']}/export?format=webm")
|
||||
assert resp.status_code == 501
|
||||
assert "PyAV" in resp.json()["detail"]
|
||||
|
|
|
|||
|
|
@ -60,11 +60,16 @@ export interface VideoStatus {
|
|||
|
||||
export interface VideoGenerateProgress {
|
||||
active: boolean;
|
||||
// "denoise" | "export" | null.
|
||||
// "queued" | "denoise" | "export" | "completed" | "failed" | null. The terminal
|
||||
// phases carry the outcome of the background job POST /video/generate started.
|
||||
phase?: string | null;
|
||||
step: number;
|
||||
total: number;
|
||||
eta_seconds?: number | null;
|
||||
// Saved gallery record when phase is "completed".
|
||||
video?: GalleryVideo | null;
|
||||
// Client-safe failure detail when phase is "failed".
|
||||
error?: string | null;
|
||||
}
|
||||
|
||||
export interface VideoLoadProgress {
|
||||
|
|
@ -144,8 +149,12 @@ export interface GalleryVideo {
|
|||
created_at: string;
|
||||
}
|
||||
|
||||
// Acknowledgement that the generation job started; the saved record arrives via
|
||||
// getVideoGenerateProgress when its phase reaches "completed".
|
||||
export interface VideoGenerateResponse {
|
||||
video: GalleryVideo;
|
||||
status: "started";
|
||||
// Always null (kept for response-shape compatibility).
|
||||
video?: GalleryVideo | null;
|
||||
}
|
||||
|
||||
async function parseJson<T>(response: Response): Promise<T> {
|
||||
|
|
@ -177,6 +186,9 @@ export async function loadVideoModel(body: VideoLoadRequest): Promise<VideoStatu
|
|||
);
|
||||
}
|
||||
|
||||
/** Start a generation job. Returns as soon as the backend accepts it (the clip takes
|
||||
* minutes, and secure mode's tunnel caps responses near 100s, so the POST cannot span
|
||||
* the generation); poll getVideoGenerateProgress for completion. */
|
||||
export async function generateVideo(
|
||||
body: VideoGenerateRequest,
|
||||
): Promise<VideoGenerateResponse> {
|
||||
|
|
|
|||
|
|
@ -1098,11 +1098,56 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
|
||||
setBusy("generating");
|
||||
setGenStep(null);
|
||||
// Poll the backend's per-step progress so the bar tracks the live denoising steps and
|
||||
// the encode phase.
|
||||
// The POST only STARTS the job and returns at once (a clip takes minutes, and
|
||||
// secure mode's tunnel caps responses near 100s, so completion cannot ride the
|
||||
// POST). A synchronous rejection (no model / already generating / bad input)
|
||||
// still surfaces here; everything after acceptance arrives via the poll.
|
||||
try {
|
||||
await generateVideo({
|
||||
prompt: prompt.trim(),
|
||||
// Only send a negative prompt when guidance uses it, so the recipe doesn't record
|
||||
// one the model ignored.
|
||||
negative_prompt: guidance > 0 ? negativePrompt.trim() || undefined : undefined,
|
||||
width: w,
|
||||
height: h,
|
||||
num_frames: numFrames,
|
||||
fps,
|
||||
steps,
|
||||
guidance,
|
||||
seed: resolvedSeed,
|
||||
});
|
||||
} catch (err) {
|
||||
if (!isMounted.current) return;
|
||||
toast.error(err instanceof Error ? err.message : "Video generation failed");
|
||||
setBusy(null);
|
||||
setGenStep(null);
|
||||
return;
|
||||
}
|
||||
// Poll the backend's per-step progress so the bar tracks the live denoising steps
|
||||
// and the encode phase, and drive completion off the terminal phase: "completed"
|
||||
// carries the saved gallery record, "failed" the client-safe error.
|
||||
genPollTimer.current = setInterval(async () => {
|
||||
try {
|
||||
const p = await getVideoGenerateProgress();
|
||||
if (p.phase === "completed" || p.phase === "failed") {
|
||||
if (genPollTimer.current) clearInterval(genPollTimer.current);
|
||||
genPollTimer.current = null;
|
||||
if (!isMounted.current) return;
|
||||
setBusy(null);
|
||||
setGenStep(null);
|
||||
if (p.phase === "completed" && p.video) {
|
||||
// Prepend the new clip (newest first) and load its blob.
|
||||
const clip = p.video;
|
||||
setVideos((prev) => [clip, ...prev.filter((v) => v.id !== clip.id)]);
|
||||
setSelectedId(clip.id);
|
||||
void ensureSrc(clip);
|
||||
} else if (p.phase === "failed") {
|
||||
const msg = p.error || "Video generation failed";
|
||||
// The user's own Cancel surfaces as the backend's cancelled sentinel; not an error.
|
||||
if (!msg.toLowerCase().includes("cancelled")) toast.error(msg);
|
||||
}
|
||||
return;
|
||||
}
|
||||
setGenStep((prev) => {
|
||||
if (!p.active) return null;
|
||||
if (
|
||||
|
|
@ -1118,35 +1163,6 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
// transient; keep polling
|
||||
}
|
||||
}, 300);
|
||||
try {
|
||||
const res = await generateVideo({
|
||||
prompt: prompt.trim(),
|
||||
// Only send a negative prompt when guidance uses it, so the recipe doesn't record
|
||||
// one the model ignored.
|
||||
negative_prompt: guidance > 0 ? negativePrompt.trim() || undefined : undefined,
|
||||
width: w,
|
||||
height: h,
|
||||
num_frames: numFrames,
|
||||
fps,
|
||||
steps,
|
||||
guidance,
|
||||
seed: resolvedSeed,
|
||||
});
|
||||
if (!isMounted.current) return;
|
||||
// Prepend the new clip (newest first) and load its blob.
|
||||
setVideos((prev) => [res.video, ...prev.filter((v) => v.id !== res.video.id)]);
|
||||
setSelectedId(res.video.id);
|
||||
void ensureSrc(res.video);
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : "Video generation failed";
|
||||
// The user's own Cancel comes back as the backend's 409 sentinel; not an error.
|
||||
if (!msg.toLowerCase().includes("cancelled")) toast.error(msg);
|
||||
} finally {
|
||||
if (genPollTimer.current) clearInterval(genPollTimer.current);
|
||||
genPollTimer.current = null;
|
||||
setBusy(null);
|
||||
setGenStep(null);
|
||||
}
|
||||
}, [
|
||||
prompt,
|
||||
negativePrompt,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue