Stream gallery clips, and close three races around them
Four fixes from the latest review pass. The video gallery downloaded each clip into a blob before it could play, so playback waited on the whole file (tens to hundreds of MB), seeking was limited to what had arrived, and every viewed clip stayed pinned in the webview. The file route already streams and serves ranges; it just could not be a <video src> because it is bearer-gated. Mint a short-lived signed link instead (its own HMAC secret, 12 hour TTL, separate from the image links) and hand it to the element, which then fetches only the ranges it plays. That removes the blob budget, its LRU and every revoke on this page. The sd.cpp readiness probe accepted any process answering on the port, so a foreign server that grabbed the port between the bind check and the spawn was adopted as ours. Confirm the listener is our child before reporting ready, and stay best-effort (psutil missing, an unknown owner, or any probe error still passes) so the check can only reject a definitely foreign process. Dataset import held its lock for the extract but not for the upload path, so two concurrent uploads into the same folder interleaved; take the same lock and return 409. And reject Windows device names (CON, NUL, COM1..9, LPT1..9, with or without an extension) plus trailing periods in dataset names, which are unopenable on Windows.
This commit is contained in:
Daniel Han
parent
49b89de3dc
commit
63740bafbd
8 changed files with 636 additions and 243 deletions
|
|
@ -115,6 +115,27 @@ def _diagnostic_tail(
|
|||
_CANCEL_GRACE_S = 5.0
|
||||
|
||||
|
||||
def _has_ancestor(pid: int, ancestor_pid: int, *, max_depth: int = 8) -> bool:
|
||||
"""True if ``ancestor_pid`` is ``pid``'s parent (or grandparent, ...).
|
||||
|
||||
The listening socket can be held by a child of the process we spawned (a wrapper script, or a
|
||||
shell on Windows), so an exact pid match alone would reject our own server. Depth-capped so a
|
||||
pid-reuse cycle cannot loop."""
|
||||
try:
|
||||
import psutil
|
||||
|
||||
proc = psutil.Process(pid)
|
||||
for _ in range(max_depth):
|
||||
proc = proc.parent()
|
||||
if proc is None:
|
||||
return False
|
||||
if proc.pid == ancestor_pid:
|
||||
return True
|
||||
except Exception: # noqa: BLE001 -- gone / no permission: treat as unknown
|
||||
return False
|
||||
return False
|
||||
|
||||
|
||||
class SdCppServer:
|
||||
"""A resident ``sd-server`` subprocess plus the HTTP client that drives it."""
|
||||
|
||||
|
|
@ -283,7 +304,7 @@ class SdCppServer:
|
|||
logger.error("sd-server exited early during load (code %s)", code)
|
||||
return False
|
||||
try:
|
||||
if self._client.get(url, timeout = 2.0).status_code == 200:
|
||||
if self._client.get(url, timeout = 2.0).status_code == 200 and self._port_is_ours():
|
||||
return True
|
||||
except (*_TRANSPORT_ERRORS, httpx.TimeoutException):
|
||||
pass
|
||||
|
|
@ -291,6 +312,50 @@ class SdCppServer:
|
|||
logger.error("sd-server readiness timed out after %ss", timeout)
|
||||
return False
|
||||
|
||||
def _port_is_ours(self) -> bool:
|
||||
"""True unless the 200 at ``/v1/models`` demonstrably came from someone else's process.
|
||||
|
||||
``_find_free_port`` binds an ephemeral port, reads it and closes the socket, and sd-server
|
||||
binds it only AFTER loading the model -- minutes for a multi-gigabyte checkpoint. Another
|
||||
local process can take the port inside that window, and ``/v1/models`` is a stock
|
||||
OpenAI-compatible route that llama.cpp's own server (and a second Studio) answers 200 on,
|
||||
so readiness would pass and every generation would be posted to an unrelated listener.
|
||||
|
||||
Verifying the listener really belongs to our child closes that. Best-effort by design:
|
||||
psutil is optional, and reading another process' connections needs privileges on some
|
||||
platforms, so anything short of a definite "that port belongs to a DIFFERENT pid" keeps
|
||||
today's behaviour rather than failing a healthy start."""
|
||||
proc = self._process
|
||||
if proc is None or proc.pid is None:
|
||||
return True
|
||||
try:
|
||||
import psutil
|
||||
except Exception: # noqa: BLE001 -- optional dependency
|
||||
return True
|
||||
try:
|
||||
for conn in psutil.net_connections(kind = "inet"):
|
||||
laddr = getattr(conn, "laddr", None)
|
||||
if not laddr or getattr(laddr, "port", None) != self.port:
|
||||
continue
|
||||
if conn.status != psutil.CONN_LISTEN:
|
||||
continue
|
||||
owner = conn.pid
|
||||
if owner is None:
|
||||
continue # not visible to us; do not punish a healthy start
|
||||
if owner == proc.pid or _has_ancestor(owner, proc.pid):
|
||||
return True
|
||||
logger.error(
|
||||
"sd-server readiness port %s is held by pid %s, not our child %s; "
|
||||
"refusing to use it",
|
||||
self.port,
|
||||
owner,
|
||||
proc.pid,
|
||||
)
|
||||
return False
|
||||
except Exception as exc: # noqa: BLE001 -- permissions / platform quirks
|
||||
logger.debug("sd-server port ownership check unavailable: %s", exc)
|
||||
return True
|
||||
|
||||
def _drain_stdout(self, proc: subprocess.Popen) -> None:
|
||||
"""Drain stdout so the pipe never deadlocks; keep a tail for diagnostics and
|
||||
feed each line to the active generation's step callback."""
|
||||
|
|
|
|||
|
|
@ -1762,8 +1762,24 @@ async def diffusion_training_info(current_subject: str = Depends(get_current_sub
|
|||
_DATASET_NAME_RE = None # compiled lazily; module keeps its import block torch-free
|
||||
|
||||
|
||||
# Reserved in EVERY directory on Windows, with or without an extension (NUL.txt is NUL). The
|
||||
# superscript COM/LPT digits are recognised as digits by Win32 and are reserved too.
|
||||
# https://learn.microsoft.com/en-us/windows/win32/fileio/naming-a-file
|
||||
_WINDOWS_RESERVED_NAMES = frozenset(
|
||||
{"con", "prn", "aux", "nul"}
|
||||
| {f"com{d}" for d in "123456789¹²³"}
|
||||
| {f"lpt{d}" for d in "123456789¹²³"}
|
||||
)
|
||||
|
||||
|
||||
def _clean_diffusion_dataset_name(name: str) -> str:
|
||||
"""Validate a dataset folder name: a single path component, no traversal, printable."""
|
||||
"""Validate a dataset folder name: a single path component, no traversal, printable.
|
||||
|
||||
Windows path rules are applied on EVERY platform, not just Windows: a dataset created on one
|
||||
machine is opened on another, and both failures are silent or confusing. A reserved device name
|
||||
dies in mkdir with an unhandled OSError, and a trailing period is stripped by Win32
|
||||
normalization, so an upload to the "new" dataset 'photos.' would quietly write into the
|
||||
existing 'photos'."""
|
||||
import re
|
||||
|
||||
global _DATASET_NAME_RE
|
||||
|
|
@ -1778,6 +1794,23 @@ def _clean_diffusion_dataset_name(name: str) -> str:
|
|||
"dashes, spaces; no slashes), e.g. 'my-style-photos'."
|
||||
),
|
||||
)
|
||||
if cleaned.endswith("."):
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
detail = (
|
||||
"Dataset name cannot end with a period: Windows strips it, so this name would "
|
||||
f"open the existing '{cleaned.rstrip('.')}' dataset instead of a new one."
|
||||
),
|
||||
)
|
||||
# The stem alone is checked, since NUL.txt is the NUL device too.
|
||||
if cleaned.split(".", 1)[0].casefold() in _WINDOWS_RESERVED_NAMES:
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
detail = (
|
||||
f"'{cleaned}' is a reserved device name on Windows and cannot be a folder. "
|
||||
"Pick another dataset name."
|
||||
),
|
||||
)
|
||||
return cleaned
|
||||
|
||||
|
||||
|
|
@ -1803,174 +1836,192 @@ async def upload_diffusion_dataset(
|
|||
# write, so a name to external-directory symlink can't make the staged upload write outside root.
|
||||
folder = _resolve_dataset_folder(name, must_exist = False)
|
||||
folder.mkdir(parents = True, exist_ok = True)
|
||||
|
||||
limit_bytes = get_upload_limit_bytes()
|
||||
total_bytes = 0
|
||||
uploaded = 0
|
||||
allowed = _DIFFUSION_DATASET_IMAGE_EXTS | _DIFFUSION_DATASET_TEXT_EXTS
|
||||
# Validate every filename up front so a valid image ahead of a bad one isn't left on disk when the
|
||||
# 400 fires; the upload is all-or-nothing.
|
||||
names: list[str] = []
|
||||
for f in files:
|
||||
# Normalise to a safe basename. Path.name doesn't split on a backslash on POSIX, so a Windows
|
||||
# client sending a backslash path in the multipart filename would be stored verbatim; fold
|
||||
# backslashes first so the true basename is taken for both separators. The read/caption/delete
|
||||
# endpoints run the stored name through _safe_dataset_image_path, so a name still holding ".."
|
||||
# here would list an image the grid can never preview, caption, or delete.
|
||||
filename = Path((f.filename or "").replace("\\", "/")).name.strip().replace("\x00", "")
|
||||
ext = Path(filename).suffix.lower()
|
||||
if not filename or ".." in filename or ext not in allowed:
|
||||
exts = ", ".join(sorted(allowed))
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
detail = f"Unsupported file '{f.filename}'. Allowed: {exts}",
|
||||
)
|
||||
# Reject an EXACT duplicate name within THIS batch (two cat.png from different folders, or an API
|
||||
# client repeating a part). The same-name exemption below is for SEPARATE repeat uploads, a
|
||||
# deliberate overwrite; inside one batch the two parts are distinct files staged to the same
|
||||
# destination on EVERY filesystem, so the later replace would silently discard the earlier one.
|
||||
# Exact match only: a case VARIANT pair stays exempt per the stem guard.
|
||||
fname_cf = filename.casefold()
|
||||
if filename in names:
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
detail = (
|
||||
f"Duplicate file '{filename}' appears more than once in this upload. "
|
||||
"Files sharing a name would overwrite each other; rename one before "
|
||||
"uploading."
|
||||
),
|
||||
)
|
||||
# Reject a second IMAGE sharing this stem but differing by extension (sample.png vs sample.jpg):
|
||||
# both resolve to the same <stem>.txt sidecar (the kohya/diffusers convention the reader, editor
|
||||
# and delete paths use), so keeping both would silently share -- and corrupt -- one caption. Check
|
||||
# files already on disk and earlier images in THIS batch. Re-uploading the exact same name stays
|
||||
# an overwrite; caption/text files are exempt.
|
||||
if ext in _DIFFUSION_DATASET_IMAGE_EXTS:
|
||||
stem = Path(filename).stem
|
||||
# Compare stems (and the same-name guard) case-insensitively: on case-insensitive filesystems two
|
||||
# images whose stems differ only by case resolve to the SAME <stem>.txt sidecar, so a
|
||||
# case-sensitive check would let both corrupt one caption. A same-name case variant is exempt ONLY
|
||||
# when its stem also differs in case (one file on case-insensitive filesystems, separate sidecars
|
||||
# on Linux). An EXTENSION-case variant (cat.PNG vs cat.png) has equal stems, so it is rejected.
|
||||
stem_cf = stem.casefold()
|
||||
|
||||
def _shares_sidecar(other_name: str) -> bool:
|
||||
other = Path(other_name)
|
||||
if (
|
||||
other_name == filename
|
||||
or other.suffix.lower() not in _DIFFUSION_DATASET_IMAGE_EXTS
|
||||
or other.stem.casefold() != stem_cf
|
||||
):
|
||||
return False
|
||||
# A casefold-equal full name is exempt unless the stems match EXACTLY (extension-case variants
|
||||
# collide on one sidecar on case-sensitive filesystems).
|
||||
return other.stem == stem or other_name.casefold() != fname_cf
|
||||
|
||||
clash = next(
|
||||
(p.name for p in folder.iterdir() if p.is_file() and _shares_sidecar(p.name)),
|
||||
None,
|
||||
)
|
||||
if clash is None:
|
||||
clash = next((n for n in names if _shares_sidecar(n)), None)
|
||||
if clash is not None:
|
||||
# Serialize against a concurrent import into the SAME folder. The training interlock counts
|
||||
# mutations rather than excluding them, and only imports took this lock, so an upload could add
|
||||
# files while an import was materializing: the import's atomic promotion (os.rmdir + rename)
|
||||
# then failed on the now-non-empty folder and fell back to a per-file move, silently merging
|
||||
# the curated set with the uploaded one, and a failure partway through that move left a mixed
|
||||
# dataset the image_count > 0 idempotency check accepts as complete. The duplicate-stem
|
||||
# validation below reads the folder too, so it has to be inside the lock as well.
|
||||
_lock = _dataset_import_lock(folder)
|
||||
if not _lock.acquire(blocking = False):
|
||||
raise HTTPException(
|
||||
status_code = 409,
|
||||
detail = (
|
||||
f"An import into '{folder.name}' is already running. Wait for it to finish, "
|
||||
"then upload again."
|
||||
),
|
||||
)
|
||||
try:
|
||||
limit_bytes = get_upload_limit_bytes()
|
||||
total_bytes = 0
|
||||
uploaded = 0
|
||||
allowed = _DIFFUSION_DATASET_IMAGE_EXTS | _DIFFUSION_DATASET_TEXT_EXTS
|
||||
# Validate every filename up front so a valid image ahead of a bad one isn't left on disk when the
|
||||
# 400 fires; the upload is all-or-nothing.
|
||||
names: list[str] = []
|
||||
for f in files:
|
||||
# Normalise to a safe basename. Path.name doesn't split on a backslash on POSIX, so a Windows
|
||||
# client sending a backslash path in the multipart filename would be stored verbatim; fold
|
||||
# backslashes first so the true basename is taken for both separators. The read/caption/delete
|
||||
# endpoints run the stored name through _safe_dataset_image_path, so a name still holding ".."
|
||||
# here would list an image the grid can never preview, caption, or delete.
|
||||
filename = Path((f.filename or "").replace("\\", "/")).name.strip().replace("\x00", "")
|
||||
ext = Path(filename).suffix.lower()
|
||||
if not filename or ".." in filename or ext not in allowed:
|
||||
exts = ", ".join(sorted(allowed))
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
detail = f"Unsupported file '{f.filename}'. Allowed: {exts}",
|
||||
)
|
||||
# Reject an EXACT duplicate name within THIS batch (two cat.png from different folders, or an API
|
||||
# client repeating a part). The same-name exemption below is for SEPARATE repeat uploads, a
|
||||
# deliberate overwrite; inside one batch the two parts are distinct files staged to the same
|
||||
# destination on EVERY filesystem, so the later replace would silently discard the earlier one.
|
||||
# Exact match only: a case VARIANT pair stays exempt per the stem guard.
|
||||
fname_cf = filename.casefold()
|
||||
if filename in names:
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
detail = (
|
||||
f"Duplicate image name '{stem}'. '{clash}' is already in this "
|
||||
f"dataset; two images sharing a name would share one '{stem}.txt' "
|
||||
f"caption. Rename one before uploading."
|
||||
f"Duplicate file '{filename}' appears more than once in this upload. "
|
||||
"Files sharing a name would overwrite each other; rename one before "
|
||||
"uploading."
|
||||
),
|
||||
)
|
||||
names.append(filename)
|
||||
# Stage each file to a temp name and move it into place only once the whole batch is written, so a
|
||||
# mid-batch failure (size limit, disk error, disconnect) leaves the dataset untouched, including
|
||||
# any pre-existing same-name file a direct write would have truncated.
|
||||
staged: list[tuple[Path, Path]] = [] # (temp, final)
|
||||
committed = False
|
||||
try:
|
||||
for f, filename in zip(files, names):
|
||||
dest = folder / filename
|
||||
# A filename-independent temp name so a long (but valid) filename can't overflow NAME_MAX once the
|
||||
# staging suffix is added.
|
||||
tmp = folder / f".upload-{_uuid.uuid4().hex}.part"
|
||||
staged.append((tmp, dest))
|
||||
with open(tmp, "wb") as out:
|
||||
while chunk := await f.read(1024 * 1024):
|
||||
total_bytes += len(chunk)
|
||||
if total_bytes > limit_bytes:
|
||||
raise HTTPException(
|
||||
status_code = 413,
|
||||
detail = (
|
||||
"Dataset upload too large. "
|
||||
f"Maximum is {get_upload_limit_label()} per upload; "
|
||||
"add the remaining images in another batch."
|
||||
),
|
||||
)
|
||||
out.write(chunk)
|
||||
# Reject a decompression bomb before commit: a small compressible PNG can pass the byte limit yet
|
||||
# decode to huge pixels and OOM the trainer's latent cache, so bound each image's dimensions from
|
||||
# the header (mirrors diffusion._decode_b64_image).
|
||||
if Path(filename).suffix.lower() in _DIFFUSION_DATASET_IMAGE_EXTS:
|
||||
_validate_uploaded_training_image(tmp, filename)
|
||||
uploaded += 1
|
||||
# Re-check the interlock immediately before the commit: the entry guard only saw the pre-upload
|
||||
# state, so a /diffusion/start could have reserved the training slot while we were streaming.
|
||||
# Committing now would move images/captions underneath the trainer; a 409 here leaves the staged
|
||||
# temps to the finally below.
|
||||
_require_diffusion_dataset_mutable()
|
||||
# Commit every staged file as one transaction. A plain replace loop is not atomic across files: a
|
||||
# mid-loop failure leaves earlier destinations already overwritten while the request errors. Back
|
||||
# up each pre-existing destination first, then on any failure drop the versions this request
|
||||
# installed and restore every displaced original.
|
||||
backups: list[tuple[Path, Optional[Path]]] = [] # (dest, backup path or None)
|
||||
installed: list[Path] = []
|
||||
try:
|
||||
for tmp, dest in staged:
|
||||
backup: Optional[Path] = None
|
||||
if dest.exists():
|
||||
backup = folder / f".upload-backup-{_uuid.uuid4().hex}.part"
|
||||
dest.replace(backup)
|
||||
backups.append((dest, backup))
|
||||
tmp.replace(dest) # atomic on the same filesystem
|
||||
installed.append(dest)
|
||||
committed = True
|
||||
except BaseException:
|
||||
# Roll back: drop every new version, then restore every displaced original.
|
||||
for dest in reversed(installed):
|
||||
try:
|
||||
dest.unlink(missing_ok = True)
|
||||
except OSError:
|
||||
pass
|
||||
for dest, backup in reversed(backups):
|
||||
if backup is not None and backup.exists():
|
||||
try:
|
||||
backup.replace(dest)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
else:
|
||||
for _, backup in backups:
|
||||
if backup is not None:
|
||||
try:
|
||||
backup.unlink(missing_ok = True)
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
if not committed:
|
||||
for tmp, _ in staged:
|
||||
try:
|
||||
tmp.unlink(missing_ok = True)
|
||||
except OSError:
|
||||
pass
|
||||
# Reject a second IMAGE sharing this stem but differing by extension (sample.png vs sample.jpg):
|
||||
# both resolve to the same <stem>.txt sidecar (the kohya/diffusers convention the reader, editor
|
||||
# and delete paths use), so keeping both would silently share -- and corrupt -- one caption. Check
|
||||
# files already on disk and earlier images in THIS batch. Re-uploading the exact same name stays
|
||||
# an overwrite; caption/text files are exempt.
|
||||
if ext in _DIFFUSION_DATASET_IMAGE_EXTS:
|
||||
stem = Path(filename).stem
|
||||
# Compare stems (and the same-name guard) case-insensitively: on case-insensitive filesystems two
|
||||
# images whose stems differ only by case resolve to the SAME <stem>.txt sidecar, so a
|
||||
# case-sensitive check would let both corrupt one caption. A same-name case variant is exempt ONLY
|
||||
# when its stem also differs in case (one file on case-insensitive filesystems, separate sidecars
|
||||
# on Linux). An EXTENSION-case variant (cat.PNG vs cat.png) has equal stems, so it is rejected.
|
||||
stem_cf = stem.casefold()
|
||||
|
||||
summary = _diffusion_dataset_summary(folder)
|
||||
return DiffusionDatasetUploadResponse(
|
||||
name = cleaned,
|
||||
path = str(folder),
|
||||
image_count = summary.image_count,
|
||||
caption_count = summary.caption_count,
|
||||
uploaded = uploaded,
|
||||
)
|
||||
def _shares_sidecar(other_name: str) -> bool:
|
||||
other = Path(other_name)
|
||||
if (
|
||||
other_name == filename
|
||||
or other.suffix.lower() not in _DIFFUSION_DATASET_IMAGE_EXTS
|
||||
or other.stem.casefold() != stem_cf
|
||||
):
|
||||
return False
|
||||
# A casefold-equal full name is exempt unless the stems match EXACTLY (extension-case variants
|
||||
# collide on one sidecar on case-sensitive filesystems).
|
||||
return other.stem == stem or other_name.casefold() != fname_cf
|
||||
|
||||
clash = next(
|
||||
(p.name for p in folder.iterdir() if p.is_file() and _shares_sidecar(p.name)),
|
||||
None,
|
||||
)
|
||||
if clash is None:
|
||||
clash = next((n for n in names if _shares_sidecar(n)), None)
|
||||
if clash is not None:
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
detail = (
|
||||
f"Duplicate image name '{stem}'. '{clash}' is already in this "
|
||||
f"dataset; two images sharing a name would share one '{stem}.txt' "
|
||||
f"caption. Rename one before uploading."
|
||||
),
|
||||
)
|
||||
names.append(filename)
|
||||
# Stage each file to a temp name and move it into place only once the whole batch is written, so a
|
||||
# mid-batch failure (size limit, disk error, disconnect) leaves the dataset untouched, including
|
||||
# any pre-existing same-name file a direct write would have truncated.
|
||||
staged: list[tuple[Path, Path]] = [] # (temp, final)
|
||||
committed = False
|
||||
try:
|
||||
for f, filename in zip(files, names):
|
||||
dest = folder / filename
|
||||
# A filename-independent temp name so a long (but valid) filename can't overflow NAME_MAX once the
|
||||
# staging suffix is added.
|
||||
tmp = folder / f".upload-{_uuid.uuid4().hex}.part"
|
||||
staged.append((tmp, dest))
|
||||
with open(tmp, "wb") as out:
|
||||
while chunk := await f.read(1024 * 1024):
|
||||
total_bytes += len(chunk)
|
||||
if total_bytes > limit_bytes:
|
||||
raise HTTPException(
|
||||
status_code = 413,
|
||||
detail = (
|
||||
"Dataset upload too large. "
|
||||
f"Maximum is {get_upload_limit_label()} per upload; "
|
||||
"add the remaining images in another batch."
|
||||
),
|
||||
)
|
||||
out.write(chunk)
|
||||
# Reject a decompression bomb before commit: a small compressible PNG can pass the byte limit yet
|
||||
# decode to huge pixels and OOM the trainer's latent cache, so bound each image's dimensions from
|
||||
# the header (mirrors diffusion._decode_b64_image).
|
||||
if Path(filename).suffix.lower() in _DIFFUSION_DATASET_IMAGE_EXTS:
|
||||
_validate_uploaded_training_image(tmp, filename)
|
||||
uploaded += 1
|
||||
# Re-check the interlock immediately before the commit: the entry guard only saw the pre-upload
|
||||
# state, so a /diffusion/start could have reserved the training slot while we were streaming.
|
||||
# Committing now would move images/captions underneath the trainer; a 409 here leaves the staged
|
||||
# temps to the finally below.
|
||||
_require_diffusion_dataset_mutable()
|
||||
# Commit every staged file as one transaction. A plain replace loop is not atomic across files: a
|
||||
# mid-loop failure leaves earlier destinations already overwritten while the request errors. Back
|
||||
# up each pre-existing destination first, then on any failure drop the versions this request
|
||||
# installed and restore every displaced original.
|
||||
backups: list[tuple[Path, Optional[Path]]] = [] # (dest, backup path or None)
|
||||
installed: list[Path] = []
|
||||
try:
|
||||
for tmp, dest in staged:
|
||||
backup: Optional[Path] = None
|
||||
if dest.exists():
|
||||
backup = folder / f".upload-backup-{_uuid.uuid4().hex}.part"
|
||||
dest.replace(backup)
|
||||
backups.append((dest, backup))
|
||||
tmp.replace(dest) # atomic on the same filesystem
|
||||
installed.append(dest)
|
||||
committed = True
|
||||
except BaseException:
|
||||
# Roll back: drop every new version, then restore every displaced original.
|
||||
for dest in reversed(installed):
|
||||
try:
|
||||
dest.unlink(missing_ok = True)
|
||||
except OSError:
|
||||
pass
|
||||
for dest, backup in reversed(backups):
|
||||
if backup is not None and backup.exists():
|
||||
try:
|
||||
backup.replace(dest)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
else:
|
||||
for _, backup in backups:
|
||||
if backup is not None:
|
||||
try:
|
||||
backup.unlink(missing_ok = True)
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
if not committed:
|
||||
for tmp, _ in staged:
|
||||
try:
|
||||
tmp.unlink(missing_ok = True)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
summary = _diffusion_dataset_summary(folder)
|
||||
return DiffusionDatasetUploadResponse(
|
||||
name = cleaned,
|
||||
path = str(folder),
|
||||
image_count = summary.image_count,
|
||||
caption_count = summary.caption_count,
|
||||
uploaded = uploaded,
|
||||
)
|
||||
finally:
|
||||
_lock.release()
|
||||
|
||||
|
||||
# ── Dataset labeling (per-image caption editing) + one-click example imports ──
|
||||
|
|
|
|||
|
|
@ -18,9 +18,13 @@ here.
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import hashlib as _hashlib
|
||||
import hmac as _hmac
|
||||
import secrets as _secrets
|
||||
import time as _time
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Response
|
||||
from pydantic import ValidationError
|
||||
|
||||
from auth.authentication import get_current_subject
|
||||
|
|
@ -341,6 +345,84 @@ async def get_gallery_video_file(
|
|||
)
|
||||
|
||||
|
||||
# A clip is tens to hundreds of MB, so the gallery cannot fetch it into a blob the way it does a
|
||||
# PNG: that buffers the whole MP4 before playback starts, defeats seeking, and pins the bytes in the
|
||||
# webview for as long as the entry is cached -- one long high-resolution clip can exceed the whole
|
||||
# cache budget on its own. The /file route already streams and serves ranges; it just cannot be a
|
||||
# <video src> because it is bearer-gated. Mint a short-lived HMAC link instead, the same shape the
|
||||
# OpenAI images URLs use, and leave the bearer route untouched.
|
||||
#
|
||||
# 12 hours rather than the images' 1: these links are for our own UI (not an outside client with an
|
||||
# OpenAI-shaped contract), a <video> element re-requests bytes whenever the user seeks or replays,
|
||||
# and the per-process secret means a restart invalidates every outstanding link anyway.
|
||||
_VIDEO_LINK_TTL = 12 * 3600
|
||||
_VIDEO_LINK_SECRET = _secrets.token_bytes(32)
|
||||
|
||||
|
||||
def _sign_video_id(video_id: str) -> str:
|
||||
exp = int(_time.time()) + _VIDEO_LINK_TTL
|
||||
payload = f"{video_id}.{exp}"
|
||||
sig = _hmac.new(_VIDEO_LINK_SECRET, payload.encode(), _hashlib.sha256).hexdigest()
|
||||
return f"{payload}.{sig}"
|
||||
|
||||
|
||||
def _verify_video_link_token(token: str) -> Optional[str]:
|
||||
"""The video id a valid, unexpired token names, else None. A separate secret from the image
|
||||
links, so a token minted for one media type can never serve the other."""
|
||||
try:
|
||||
video_id, exp_s, sig = token.rsplit(".", 2)
|
||||
except ValueError:
|
||||
return None
|
||||
expected = _hmac.new(
|
||||
_VIDEO_LINK_SECRET, f"{video_id}.{exp_s}".encode(), _hashlib.sha256
|
||||
).hexdigest()
|
||||
if not _hmac.compare_digest(sig, expected):
|
||||
return None
|
||||
try:
|
||||
if int(exp_s) < int(_time.time()):
|
||||
return None
|
||||
except ValueError:
|
||||
return None
|
||||
return video_id
|
||||
|
||||
|
||||
@router.get("/video/gallery/{video_id}/signed-url")
|
||||
async def get_gallery_video_signed_url(
|
||||
video_id: str, current_subject: str = Depends(get_current_subject)
|
||||
):
|
||||
"""A directly playable, range-capable link for one clip (bearer-gated to mint, HMAC to use).
|
||||
|
||||
Returned as a relative URL so it works behind any proxy the page itself is served through."""
|
||||
from core.inference import video_gallery
|
||||
|
||||
path = await asyncio.to_thread(video_gallery.owned_video_path, video_id)
|
||||
if path is None:
|
||||
raise HTTPException(status_code = 404, detail = "Video not found.")
|
||||
token = _sign_video_id(video_id)
|
||||
return {"url": f"/api/inference/video/gallery/{video_id}/file-signed?token={token}"}
|
||||
|
||||
|
||||
@router.get("/video/gallery/{video_id}/file-signed")
|
||||
async def get_gallery_video_file_signed(video_id: str, token: str = Query(...)):
|
||||
"""Stream one gallery MP4 gated by the HMAC token instead of the bearer, so it can be a plain
|
||||
<video src> and the browser can range-request it. Same ownership gate as the bearer route, and
|
||||
the token names the single clip it may serve."""
|
||||
from core.inference import video_gallery
|
||||
|
||||
if _verify_video_link_token(token) != video_id:
|
||||
raise HTTPException(status_code = 401, detail = "Invalid or expired video link.")
|
||||
path = await asyncio.to_thread(video_gallery.owned_video_path, video_id)
|
||||
if path is None:
|
||||
raise HTTPException(status_code = 404, detail = "Video not found.")
|
||||
from fastapi.responses import FileResponse
|
||||
|
||||
return FileResponse(
|
||||
path,
|
||||
media_type = "video/mp4",
|
||||
headers = {"Cache-Control": "private, max-age=31536000, immutable"},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/video/gallery/{video_id}/export")
|
||||
async def export_gallery_video(
|
||||
video_id: str,
|
||||
|
|
|
|||
|
|
@ -815,3 +815,45 @@ def test_an_unreadable_sidecar_shadows_the_metadata_caption(client, ds_root):
|
|||
summary = next(d for d in info["datasets"] if d["name"] == "tombstone")
|
||||
assert summary["image_count"] == 2
|
||||
assert summary["caption_count"] == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", ["CON", "nul", "COM1", "lpt9", "NUL.txt", "aux.images"])
|
||||
def test_upload_rejects_windows_reserved_dataset_names(client, ds_root, bad):
|
||||
# Reserved in every directory on Windows, with or without an extension (NUL.txt is NUL), so
|
||||
# mkdir dies with an unhandled OSError there. Rejected on every platform, since a dataset made
|
||||
# on Linux gets opened on Windows.
|
||||
resp = _upload(client, bad, [("sample.png", _png_bytes())])
|
||||
assert resp.status_code == 400, resp.text
|
||||
assert "reserved" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
def test_upload_rejects_a_trailing_period_dataset_name(client, ds_root):
|
||||
# Win32 strips a trailing period, so 'photos.' opens the existing 'photos' dataset: an upload
|
||||
# meant for a new name would silently modify the old one.
|
||||
assert _upload(client, "photos", [("sample.png", _png_bytes())]).status_code == 200
|
||||
resp = _upload(client, "photos.", [("other.png", _png_bytes())])
|
||||
assert resp.status_code == 400, resp.text
|
||||
assert "period" in resp.json()["detail"].lower()
|
||||
# The existing dataset was not touched.
|
||||
assert sorted(p.name for p in (ds_root / "photos").iterdir()) == ["sample.png"]
|
||||
|
||||
|
||||
def test_upload_refuses_while_an_import_holds_the_same_folder(client, ds_root):
|
||||
# The training interlock counts mutations rather than excluding them, and only imports took the
|
||||
# per-folder lock, so an upload could add files while an import was materializing: the import's
|
||||
# atomic promotion then failed on the non-empty folder and merged the two sets.
|
||||
from routes.training import _dataset_import_lock
|
||||
|
||||
folder = ds_root / "shared-name"
|
||||
folder.mkdir(parents = True, exist_ok = True)
|
||||
lock = _dataset_import_lock(folder)
|
||||
assert lock.acquire(blocking = False)
|
||||
try:
|
||||
resp = _upload(client, "shared-name", [("sample.png", _png_bytes())])
|
||||
assert resp.status_code == 409, resp.text
|
||||
assert "import" in resp.json()["detail"].lower()
|
||||
assert not list(folder.glob("*.png"))
|
||||
finally:
|
||||
lock.release()
|
||||
# Released: the same upload now goes through.
|
||||
assert _upload(client, "shared-name", [("sample.png", _png_bytes())]).status_code == 200
|
||||
|
|
|
|||
|
|
@ -450,3 +450,112 @@ def test_diagnostic_tail_falls_back_to_the_last_lines():
|
|||
def test_diagnostic_tail_is_bounded():
|
||||
lines = ["error: " + "x" * 500 for _ in range(20)]
|
||||
assert len(srv._diagnostic_tail(lines)) <= 1500
|
||||
|
||||
|
||||
def test_readiness_refuses_a_port_held_by_another_process(patched):
|
||||
# _find_free_port picks an ephemeral port, closes the socket, and sd-server binds it only after
|
||||
# loading the model -- minutes for a big checkpoint. Another local process can take it in that
|
||||
# window, and /v1/models is a stock OpenAI route that llama.cpp's server also answers 200 on, so
|
||||
# readiness would pass and every generation would go to an unrelated listener.
|
||||
import types
|
||||
|
||||
popen = _FakePopen(lines = ["loading model"])
|
||||
patched.setattr(srv.subprocess, "Popen", lambda *a, **k: popen)
|
||||
s = _server_with(popen, _FakeClient(get = lambda url: _Resp(200, {"model": {}})))
|
||||
|
||||
fake_psutil = types.SimpleNamespace(
|
||||
CONN_LISTEN = "LISTEN",
|
||||
net_connections = lambda kind = "inet": [
|
||||
types.SimpleNamespace(
|
||||
laddr = types.SimpleNamespace(port = s.port),
|
||||
status = "LISTEN",
|
||||
pid = popen.pid + 1000, # somebody else
|
||||
)
|
||||
],
|
||||
Process = lambda pid: types.SimpleNamespace(parent = lambda: None),
|
||||
)
|
||||
patched.setitem(__import__("sys").modules, "psutil", fake_psutil)
|
||||
assert s._port_is_ours() is False
|
||||
|
||||
|
||||
def test_readiness_accepts_our_own_child_and_its_descendants(patched):
|
||||
import types
|
||||
|
||||
popen = _FakePopen()
|
||||
s = _server_with(popen, _FakeClient(get = lambda url: _Resp(200, {})))
|
||||
|
||||
def _conns(owner_pid):
|
||||
return [
|
||||
types.SimpleNamespace(
|
||||
laddr = types.SimpleNamespace(port = s.port), status = "LISTEN", pid = owner_pid
|
||||
)
|
||||
]
|
||||
|
||||
# The spawned pid itself.
|
||||
patched.setitem(
|
||||
__import__("sys").modules,
|
||||
"psutil",
|
||||
types.SimpleNamespace(
|
||||
CONN_LISTEN = "LISTEN",
|
||||
net_connections = lambda kind = "inet": _conns(popen.pid),
|
||||
Process = lambda pid: types.SimpleNamespace(parent = lambda: None),
|
||||
),
|
||||
)
|
||||
assert s._port_is_ours() is True
|
||||
|
||||
# A grandchild (wrapper script / shell) still counts as ours.
|
||||
child_pid = popen.pid + 7
|
||||
|
||||
def _process(pid):
|
||||
if pid == child_pid:
|
||||
return types.SimpleNamespace(
|
||||
parent = lambda: types.SimpleNamespace(pid = popen.pid, parent = lambda: None)
|
||||
)
|
||||
return types.SimpleNamespace(parent = lambda: None)
|
||||
|
||||
patched.setitem(
|
||||
__import__("sys").modules,
|
||||
"psutil",
|
||||
types.SimpleNamespace(
|
||||
CONN_LISTEN = "LISTEN",
|
||||
net_connections = lambda kind = "inet": _conns(child_pid),
|
||||
Process = _process,
|
||||
),
|
||||
)
|
||||
assert s._port_is_ours() is True
|
||||
|
||||
|
||||
def test_readiness_check_is_best_effort(patched):
|
||||
# No psutil, an unreadable owner pid, or a raising lookup must never fail a healthy start.
|
||||
import types
|
||||
|
||||
popen = _FakePopen()
|
||||
s = _server_with(popen, _FakeClient(get = lambda url: _Resp(200, {})))
|
||||
|
||||
patched.setitem(__import__("sys").modules, "psutil", None)
|
||||
assert s._port_is_ours() is True
|
||||
|
||||
def _boom(kind = "inet"):
|
||||
raise PermissionError("not allowed")
|
||||
|
||||
patched.setitem(
|
||||
__import__("sys").modules,
|
||||
"psutil",
|
||||
types.SimpleNamespace(CONN_LISTEN = "LISTEN", net_connections = _boom),
|
||||
)
|
||||
assert s._port_is_ours() is True
|
||||
|
||||
# Owner pid not visible (common for another user's process): unknown, so keep going.
|
||||
patched.setitem(
|
||||
__import__("sys").modules,
|
||||
"psutil",
|
||||
types.SimpleNamespace(
|
||||
CONN_LISTEN = "LISTEN",
|
||||
net_connections = lambda kind = "inet": [
|
||||
types.SimpleNamespace(
|
||||
laddr = types.SimpleNamespace(port = s.port), status = "LISTEN", pid = None
|
||||
)
|
||||
],
|
||||
),
|
||||
)
|
||||
assert s._port_is_ours() is True
|
||||
|
|
|
|||
|
|
@ -866,3 +866,77 @@ def test_video_load_guard_still_checks_diffusion_when_the_llm_probe_raises(clien
|
|||
assert resp.status_code == 409
|
||||
assert "training" in resp.json()["detail"].lower()
|
||||
assert video_routes is not None
|
||||
|
||||
|
||||
def test_signed_video_link_streams_without_a_bearer(client):
|
||||
# A clip is tens to hundreds of MB, so the gallery cannot fetch it into a blob the way it does a
|
||||
# PNG: that buffers the whole MP4 before playback, kills seeking, and pins the bytes for as long
|
||||
# as the entry is cached. The signed link makes the range-capable /file route usable as a plain
|
||||
# <video src>.
|
||||
client.post(
|
||||
"/api/inference/video/load",
|
||||
json = {"model_path": "unsloth/LTX-2.3-GGUF", "gguf_filename": "q.gguf"},
|
||||
)
|
||||
video = _generate_and_wait(client, {"prompt": "a"})
|
||||
vid = video["id"]
|
||||
|
||||
minted = client.get(f"/api/inference/video/gallery/{vid}/signed-url")
|
||||
assert minted.status_code == 200, minted.text
|
||||
url = minted.json()["url"]
|
||||
assert url.startswith(f"/api/inference/video/gallery/{vid}/file-signed?token=")
|
||||
|
||||
# Served with no Authorization header at all, and byte-identical to the bearer route.
|
||||
signed = client.get(url, headers = {})
|
||||
assert signed.status_code == 200
|
||||
assert signed.headers["content-type"] == "video/mp4"
|
||||
assert signed.content == client.get(f"/api/inference/video/gallery/{vid}/file").content
|
||||
|
||||
# Range requests work, which is the point: the player seeks instead of downloading everything.
|
||||
ranged = client.get(url, headers = {"Range": "bytes=0-3"})
|
||||
assert ranged.status_code == 206
|
||||
assert len(ranged.content) == 4
|
||||
|
||||
|
||||
def test_signed_video_link_rejects_tampering_and_other_ids(client):
|
||||
client.post(
|
||||
"/api/inference/video/load",
|
||||
json = {"model_path": "unsloth/LTX-2.3-GGUF", "gguf_filename": "q.gguf"},
|
||||
)
|
||||
first = _generate_and_wait(client, {"prompt": "a"})["id"]
|
||||
second = _generate_and_wait(client, {"prompt": "b"})["id"]
|
||||
token = client.get(f"/api/inference/video/gallery/{first}/signed-url").json()["url"].split(
|
||||
"token=", 1
|
||||
)[1]
|
||||
|
||||
# The token names exactly one clip.
|
||||
assert client.get(
|
||||
f"/api/inference/video/gallery/{second}/file-signed?token={token}"
|
||||
).status_code == 401
|
||||
# A flipped signature, a malformed token, and an expired one are all refused.
|
||||
assert client.get(
|
||||
f"/api/inference/video/gallery/{first}/file-signed?token={token[:-1]}x"
|
||||
).status_code == 401
|
||||
assert client.get(
|
||||
f"/api/inference/video/gallery/{first}/file-signed?token=nonsense"
|
||||
).status_code == 401
|
||||
from routes import video as video_routes
|
||||
|
||||
expired = video_routes._sign_video_id(first)
|
||||
payload, _sig = expired.rsplit(".", 1)
|
||||
stale_id, _exp = payload.rsplit(".", 1)
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
stale_payload = f"{stale_id}.1"
|
||||
stale_sig = hmac.new(
|
||||
video_routes._VIDEO_LINK_SECRET, stale_payload.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
assert client.get(
|
||||
f"/api/inference/video/gallery/{first}/file-signed?token={stale_payload}.{stale_sig}"
|
||||
).status_code == 401
|
||||
|
||||
|
||||
def test_signed_url_mint_is_bearer_gated_and_404s_for_an_unknown_clip(client):
|
||||
assert client.get(
|
||||
"/api/inference/video/gallery/does-not-exist/signed-url"
|
||||
).status_code == 404
|
||||
|
|
|
|||
|
|
@ -248,17 +248,22 @@ export async function clearVideoGallery(): Promise<void> {
|
|||
if (!res.ok) throw new Error(await readFastApiError(res));
|
||||
}
|
||||
|
||||
/** Fetch a gallery MP4 (auth-protected, so it can't be a plain <video src>) and wrap it
|
||||
* in an object URL. Callers must revoke the URL when done. Mirrors the images gallery. */
|
||||
export async function fetchGalleryVideoObjectUrl(
|
||||
url: string,
|
||||
): Promise<{ url: string; bytes: number }> {
|
||||
const res = await authFetch(url);
|
||||
/** A directly playable, range-capable URL for one gallery clip.
|
||||
*
|
||||
* NOT the blob treatment the images gallery uses: an MP4 is tens to hundreds of MB, so
|
||||
* `res.blob()` would download the whole clip before playback could start, defeat seeking, and
|
||||
* pin those bytes in the webview for as long as the entry is cached -- one long clip can exceed
|
||||
* the entire cache budget by itself. The backend's file route already streams and serves ranges;
|
||||
* it just cannot be a plain <video src> because it is bearer-gated, so mint a short-lived signed
|
||||
* link and let the element fetch only what it plays. */
|
||||
export async function fetchGalleryVideoSignedUrl(id: string): Promise<string> {
|
||||
const res = await authFetch(
|
||||
`/api/inference/video/gallery/${encodeURIComponent(id)}/signed-url`,
|
||||
);
|
||||
if (!res.ok) throw new Error(await readFastApiError(res));
|
||||
// The blob's size travels with the URL: the gallery cache is budgeted in bytes, and a clip's
|
||||
// size varies by two orders of magnitude, so the caller cannot estimate it.
|
||||
const blob = await res.blob();
|
||||
return { url: URL.createObjectURL(blob), bytes: blob.size };
|
||||
const body = (await res.json()) as { url?: string };
|
||||
if (!body.url) throw new Error("The server returned no video link.");
|
||||
return body.url;
|
||||
}
|
||||
|
||||
/** Server-side transcode for the Download menu (WebM / GIF). The backend 501s
|
||||
|
|
|
|||
|
|
@ -59,7 +59,6 @@ import { formatBytes, formatEta } from "@/features/hub/lib/format";
|
|||
import { useNavigate, useSearch } from "@tanstack/react-router";
|
||||
import { useStagedDownload } from "@/features/hub/download-manager";
|
||||
import { cn } from "@/lib/utils";
|
||||
import { BlobUrlCache } from "@/lib/blob-url-cache";
|
||||
import { diffusionRoutePick } from "@/lib/diffusion-route-pick";
|
||||
import { toast } from "@/lib/toast";
|
||||
|
||||
|
|
@ -72,7 +71,7 @@ import {
|
|||
clearVideoGallery,
|
||||
deleteGalleryVideo,
|
||||
fetchGalleryVideoExport,
|
||||
fetchGalleryVideoObjectUrl,
|
||||
fetchGalleryVideoSignedUrl,
|
||||
generateVideo,
|
||||
getVideoGallery,
|
||||
getVideoGenerateProgress,
|
||||
|
|
@ -125,29 +124,22 @@ const FALLBACK_RESOLUTION_PRESETS: Array<[number, number]> = [
|
|||
const FALLBACK_FRAME_STEP = 8;
|
||||
const FALLBACK_FPS = 24;
|
||||
|
||||
// Blob budget for cached clips. A clip runs from a few MB to a few hundred MB, and the page
|
||||
// stays mounted after its first visit, so an unbounded cache pinned everything the user ever
|
||||
// scrolled past for the rest of the session. 512 MB holds a comfortable working set (the strip's
|
||||
// visible cards plus their neighbours) while capping what the webview can be made to hold; the
|
||||
// playing and on-screen clips are never evicted, and anything dropped re-fetches on scroll-back.
|
||||
const VIDEO_BLOB_BUDGET_BYTES = 512 * 1024 * 1024;
|
||||
|
||||
// Module cache of the backend-persisted gallery, so a tab switch re-renders instantly.
|
||||
// Object URLs are revoked only on delete/eviction (not unmount), so they stay valid across
|
||||
// remounts.
|
||||
// The srcById entries are short-lived signed links, not object URLs: nothing is pinned in the
|
||||
// webview, so they survive a remount and need no budget, eviction or revoke. The clip's bytes
|
||||
// are streamed by the media element itself, which fetches ranges as it plays and drops what it
|
||||
// no longer needs -- the whole point of not blob-ing a file that runs to hundreds of MB.
|
||||
const galleryCache: {
|
||||
videos: GalleryVideo[];
|
||||
hasMore: boolean;
|
||||
selectedId: string | null;
|
||||
quant: string | null;
|
||||
srcById: BlobUrlCache;
|
||||
// Ids with a fetch in flight, so concurrent ensureSrc calls don't double-fetch
|
||||
// (and leak the duplicate object URL).
|
||||
srcById: Map<string, string>;
|
||||
// Ids with a mint in flight, so concurrent ensureSrc calls don't double-request.
|
||||
inflight: Set<string>;
|
||||
// Ids deleted while their MP4 was still downloading. A clip is tens to hundreds of MB, and a
|
||||
// fetch that lands after the delete has nothing left to revoke it: the card is gone, so the
|
||||
// blob would stay pinned for the rest of the session. Clear-all bumps the epoch instead of
|
||||
// listing every id.
|
||||
// Ids deleted while their link was still being minted, so a reply that lands after the delete
|
||||
// isn't cached for a card that no longer exists. Clear-all bumps the epoch instead of listing
|
||||
// every id.
|
||||
deleted: Set<string>;
|
||||
epoch: number;
|
||||
} = {
|
||||
|
|
@ -155,7 +147,7 @@ const galleryCache: {
|
|||
hasMore: false,
|
||||
selectedId: null,
|
||||
quant: null,
|
||||
srcById: new BlobUrlCache(VIDEO_BLOB_BUDGET_BYTES),
|
||||
srcById: new Map(),
|
||||
inflight: new Set(),
|
||||
deleted: new Set(),
|
||||
epoch: 0,
|
||||
|
|
@ -177,28 +169,29 @@ function exportFilename(video: GalleryVideo, format: VideoExportFormat = "mp4"):
|
|||
return `Unsloth_video_${stamp}_${video.seed}.${format}`;
|
||||
}
|
||||
|
||||
function saveBlobUrl(href: string, filename: string) {
|
||||
function saveLink(href: string, filename: string) {
|
||||
const link = document.createElement("a");
|
||||
link.href = href;
|
||||
link.download = filename;
|
||||
link.click();
|
||||
}
|
||||
|
||||
// MP4 saves the already-fetched original bytes; WebM / GIF are transcoded by
|
||||
// the backend on demand (501 with a readable reason when the codec is absent).
|
||||
// MP4 saves the original file straight from its signed link (same-origin, so the download
|
||||
// attribute is honoured); WebM / GIF are transcoded by the backend on demand (501 with a
|
||||
// readable reason when the codec is absent).
|
||||
async function downloadVideo(
|
||||
src: string,
|
||||
video: GalleryVideo,
|
||||
format: VideoExportFormat = "mp4",
|
||||
) {
|
||||
if (format === "mp4") {
|
||||
saveBlobUrl(src, exportFilename(video, format));
|
||||
saveLink(src, exportFilename(video, format));
|
||||
return;
|
||||
}
|
||||
const blob = await fetchGalleryVideoExport(video.id, format);
|
||||
const url = URL.createObjectURL(blob);
|
||||
try {
|
||||
saveBlobUrl(url, exportFilename(video, format));
|
||||
saveLink(url, exportFilename(video, format));
|
||||
} finally {
|
||||
setTimeout(() => URL.revokeObjectURL(url), 10_000);
|
||||
}
|
||||
|
|
@ -581,7 +574,7 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
if (!active) previewRef.current?.pause();
|
||||
}, [active]);
|
||||
const [srcById, setSrcById] = useState<Record<string, string>>(() =>
|
||||
galleryCache.srcById.toRecord(),
|
||||
Object.fromEntries(galleryCache.srcById),
|
||||
);
|
||||
// Guards a "load more" so a fast scroll can't fire several at once.
|
||||
const loadingMore = useRef(false);
|
||||
|
|
@ -698,38 +691,23 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
}
|
||||
}, [loadedModelKey, defaultSteps, defaultGuidance]);
|
||||
|
||||
// Fetch (once) the object URL for a record's MP4; cached across remounts. Same
|
||||
// auth-protected blob pattern the images gallery uses.
|
||||
// Mint (once) a playable link for a record's MP4; cached across remounts. Unlike the images
|
||||
// gallery this does NOT download the file: the link goes straight into the <video> element,
|
||||
// which streams ranges as it plays, so a clip starts on its first seconds instead of after a
|
||||
// full download and seeking works.
|
||||
const ensureSrc = useCallback(async (video: GalleryVideo) => {
|
||||
if (galleryCache.srcById.has(video.id) || galleryCache.inflight.has(video.id)) return;
|
||||
galleryCache.inflight.add(video.id);
|
||||
const epochAtStart = galleryCache.epoch;
|
||||
try {
|
||||
const { url, bytes } = await fetchGalleryVideoObjectUrl(video.url);
|
||||
// The record can be deleted (or the gallery cleared) while its MP4 is downloading. The
|
||||
// delete handler revoked whatever URL existed then, so caching this one would pin a blob
|
||||
// no card can ever release.
|
||||
if (galleryCache.deleted.has(video.id) || galleryCache.epoch !== epochAtStart) {
|
||||
URL.revokeObjectURL(url);
|
||||
return;
|
||||
}
|
||||
galleryCache.srcById.set(video.id, url, bytes);
|
||||
// Evict the coldest off-screen clips this one pushed over budget. On-screen cards and the
|
||||
// clip in the player are protected, so eviction is never visible; an evicted card re-fetches
|
||||
// if it is scrolled back to. The clip just fetched is protected too, or a single clip larger
|
||||
// than the whole budget would evict itself and re-fetch on every pass.
|
||||
const evicted = galleryCache.srcById.prune(
|
||||
new Set([video.id, ...visibleIds.current, galleryCache.selectedId ?? ""]),
|
||||
);
|
||||
const url = await fetchGalleryVideoSignedUrl(video.id);
|
||||
// The record can be deleted (or the gallery cleared) while the link is being minted;
|
||||
// caching it then would leave an entry for a card that no longer exists.
|
||||
if (galleryCache.deleted.has(video.id) || galleryCache.epoch !== epochAtStart) return;
|
||||
galleryCache.srcById.set(video.id, url);
|
||||
// The URL is cached above either way; skip the state update after unmount
|
||||
// (matches the other async callbacks in this file).
|
||||
if (isMounted.current) {
|
||||
setSrcById((prev) => {
|
||||
const next = { ...prev, [video.id]: url };
|
||||
for (const id of evicted) delete next[id];
|
||||
return next;
|
||||
});
|
||||
}
|
||||
if (isMounted.current) setSrcById((prev) => ({ ...prev, [video.id]: url }));
|
||||
} catch {
|
||||
// Leave it without a src; the card shows a placeholder.
|
||||
} finally {
|
||||
|
|
@ -737,34 +715,23 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
}
|
||||
}, []);
|
||||
|
||||
// Fetching a clip pulls its whole MP4 into an object URL that lives until the page closes, so
|
||||
// fetching a full gallery page (PAGE_SIZE records, each tens to hundreds of MB for a longer
|
||||
// clip) up front pinned hundreds of MB -- gigabytes over a few "load more" pages -- for cards
|
||||
// the user may never scroll to, and starved the one they were waiting on. Fetch a card as it
|
||||
// nears the viewport instead; the tile already shows a spinner until its src lands.
|
||||
// A card's poster frame only appears once its src lands, and each src costs a request, so a
|
||||
// full gallery page (PAGE_SIZE records) minted up front would queue PAGE_SIZE requests ahead
|
||||
// of the one clip the user is actually waiting on. Mint a card's link as it nears the viewport
|
||||
// instead; the tile shows a spinner until its src lands.
|
||||
// The cards are observed from here rather than through a ref on each tile: the tile is a
|
||||
// Tooltip trigger, whose asChild clone owns that ref. Re-runs per page of records, so cards
|
||||
// appended by "load more" are picked up and removed ones are dropped with the observer.
|
||||
const stripRef = useRef<HTMLDivElement | null>(null);
|
||||
// Ids currently intersecting the strip. The blob cache never evicts these, so pruning cannot
|
||||
// pull a clip out from under a visible card.
|
||||
const visibleIds = useRef<Set<string>>(new Set());
|
||||
useEffect(() => {
|
||||
const root = stripRef.current;
|
||||
if (!root || typeof IntersectionObserver === "undefined") return;
|
||||
const io = new IntersectionObserver(
|
||||
(entries) => {
|
||||
for (const entry of entries) {
|
||||
if (!entry.isIntersecting) continue;
|
||||
const id = (entry.target as HTMLElement).dataset.clipId;
|
||||
if (!id) continue;
|
||||
// Visibility is also the cache's recency and protection signal: an on-screen clip is
|
||||
// never evicted, and leaving the viewport makes it a candidate again.
|
||||
if (!entry.isIntersecting) {
|
||||
visibleIds.current.delete(id);
|
||||
continue;
|
||||
}
|
||||
visibleIds.current.add(id);
|
||||
galleryCache.srcById.touch(id);
|
||||
const clip = videos.find((v) => v.id === id);
|
||||
if (clip) void ensureSrc(clip);
|
||||
}
|
||||
|
|
@ -836,7 +803,7 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
|
||||
// WebM/GIF go through a server-side transcode that can take a few seconds
|
||||
// (and 501s with a readable reason when the codec is missing), so wrap the
|
||||
// helper with progress + error toasts; MP4 saves instantly.
|
||||
// helper with progress + error toasts; MP4 hands the link to the browser.
|
||||
const handleDownload = useCallback(
|
||||
async (src: string, video: GalleryVideo, format: "mp4" | "webm" | "gif") => {
|
||||
if (format === "mp4") {
|
||||
|
|
@ -864,9 +831,8 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
toast.error(err instanceof Error ? err.message : "Failed to delete video");
|
||||
return;
|
||||
}
|
||||
galleryCache.srcById.delete(id); // revokes the URL with the entry
|
||||
visibleIds.current.delete(id);
|
||||
// A fetch still in flight for this id must throw its blob away rather than cache it.
|
||||
galleryCache.srcById.delete(id);
|
||||
// A mint still in flight for this id must throw its link away rather than cache it.
|
||||
galleryCache.deleted.add(id);
|
||||
setSrcById((prev) => {
|
||||
const next = { ...prev };
|
||||
|
|
@ -884,9 +850,8 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
toast.error(err instanceof Error ? err.message : "Failed to clear gallery");
|
||||
return;
|
||||
}
|
||||
galleryCache.srcById.clear(); // revokes every cached URL
|
||||
visibleIds.current.clear();
|
||||
// Every fetch in flight now belongs to a cleared gallery, so their blobs are discarded on
|
||||
galleryCache.srcById.clear();
|
||||
// Every mint in flight now belongs to a cleared gallery, so their links are discarded on
|
||||
// arrival. The epoch covers ids this page never listed too.
|
||||
galleryCache.epoch += 1;
|
||||
galleryCache.videos = [];
|
||||
|
|
@ -1044,7 +1009,7 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
setBusy(null);
|
||||
setGenStep(null);
|
||||
if (p.phase === "completed" && p.video) {
|
||||
// Prepend the new clip (newest first) and load its blob.
|
||||
// Prepend the new clip (newest first) and mint its link.
|
||||
const clip = p.video;
|
||||
setVideos((prev) => [clip, ...prev.filter((v) => v.id !== clip.id)]);
|
||||
setSelectedId(clip.id);
|
||||
|
|
@ -1876,7 +1841,7 @@ export function VideoPage({ active = true }: { active?: boolean }) {
|
|||
</div>
|
||||
</>
|
||||
) : selected ? (
|
||||
// The selected record's blob is still loading -- spin in place.
|
||||
// The selected record's link has not landed yet -- spin in place.
|
||||
<div className="flex flex-col items-center gap-3 text-muted-foreground">
|
||||
<Spinner className="size-8" />
|
||||
<p className="text-sm">Loading…</p>
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue