Merge pull request #2496 from debpalash/triage/downloads-storage

fix(downloads): validated ranges/resume, atomic snapshots, portable filenames, disk-full guidance (#2376 #2451 #2452 #2453 #2479)
This commit is contained in:
Palash Debnath
2026-10-01 21:51:04 +05:30
committed by GitHub
25 changed files with 1238 additions and 70 deletions
+3
View File
@@ -9,6 +9,7 @@ metadata and the backend fallback mirror it.
## [Unreleased]
**Highlights**
- Reserve migration snapshot counters atomically (#2453) — thanks @rudycelekli!
- Ask VoiceStudio Agent adds chat, harness selection, feature presets, read-only planning and autopilot app actions without a source checkout (#2407)
- Home credits contributors with over 10 commits in three responsive rows of round avatars stacked from right to left with an All contributors link, with GitHub and X links on Palash's hover card (#2407)
- The VoiceStudio.sh Open Source title opens a website preview below the clicked item, within the right content area, with navigation and external-browser controls (#2407)
@@ -86,6 +87,8 @@ metadata and the backend fallback mirror it.
- The Twilio guide and integration directory describe the guided setup and in-app integration pages (#2304)
### Fixed
- Count interrupted sidecar installs and report unreadable engine scans in Storage usage (#2479) — thanks @rudycelekli!
- Dub assembly, cached segments and audio tools can read generated WAV files when TorchCodec is missing or cannot load (#2379) — thanks @tokutei58301-boop!
- Remote API-key and share-PIN clients can record and read export history while native filesystem operations stay local (#2383, #2384) — thanks @sedatdagg!
- Concurrent job events receive unique sequence numbers (#2384) — thanks @sedatdagg!
+2 -1
View File
@@ -24,6 +24,7 @@ from pydantic import BaseModel
from core.config import DATA_DIR
from core import failure
from core.logging_utils import log_safe
from core.path_security import portable_filename
from core.file_cleanup import FileCleanupError, unlink_if_present
from services.dub_batching import (
BATCH_WIDTH_ENV,
@@ -1157,7 +1158,7 @@ def download_batch_output(job_id: str, lang: str):
if not path or not os.path.exists(path):
raise HTTPException(404, f"No output for language '{lang}'")
filename = f"{os.path.splitext(job['filename'])[0]}_{lang}.mp4"
filename = portable_filename(f"{os.path.splitext(job['filename'])[0]}_{lang}.mp4", "output.mp4")
return FileResponse(
path,
media_type="video/mp4",
+4 -4
View File
@@ -13,7 +13,7 @@ from typing import Optional
from core.config import DUB_DIR
from core.http_headers import content_disposition
from core.logging_utils import log_safe
from core.path_security import UnsafePath, resolve_within
from core.path_security import UnsafePath, portable_filename, resolve_within
from core.tasks import task_manager
from fastapi import APIRouter, Header, HTTPException, Query, Request, Response
from fastapi.responses import FileResponse, StreamingResponse
@@ -1888,7 +1888,7 @@ async def dub_export_srt(
srt_content = "\n".join(srt_lines)
base_name = os.path.splitext(job.get('filename', 'video'))[0]
suffix = "_dual" if dual else ""
dl_name = f"subtitles_{base_name}{suffix}.srt"
dl_name = portable_filename(f"subtitles_{base_name}{suffix}.srt", "subtitles")
return Response(
content=srt_content,
media_type="text/plain",
@@ -1947,7 +1947,7 @@ async def dub_export_vtt(
vtt_content = "\n".join(vtt_lines)
base_name = os.path.splitext(job.get('filename', 'video'))[0]
suffix = "_dual" if dual else ""
dl_name = f"subtitles_{base_name}{suffix}.vtt"
dl_name = portable_filename(f"subtitles_{base_name}{suffix}.vtt", "subtitles")
return Response(
content=vtt_content,
media_type="text/vtt",
@@ -1991,7 +1991,7 @@ async def dub_export_ass(
]
base_name = os.path.splitext(job.get('filename', 'video'))[0]
dl_name = f"subtitles_{base_name}_karaoke.ass"
dl_name = portable_filename(f"subtitles_{base_name}_karaoke.ass", "subtitles.ass")
return Response(
content=build_ass(segments),
media_type="text/plain",
+12 -1
View File
@@ -21,7 +21,7 @@ from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from core import prefs
from core.failure import is_hf_connectivity_error
from core.failure import is_disk_full_error, is_hf_connectivity_error
from services.hf_revisions import revision_for
from utils import hf_progress
from utils import download_aggregator
@@ -33,6 +33,7 @@ from .models import ( # noqa: F401
KNOWN_MODELS,
invalidate_cache,
snapshot_is_complete,
disk_full_message,
disk_space_error,
_MIN_WEIGHT_BYTES,
_WEIGHT_FLOORS,
@@ -476,6 +477,10 @@ def _is_retryable_download_error(exc: BaseException) -> bool:
if isinstance(exc, _InstallCancelled):
return False
if is_disk_full_error(exc):
# A full volume does not heal between backoff sleeps; five retries only
# delay the actionable message (and re-preallocate the whole file).
return False
if isinstance(exc, HfHubHTTPError):
# An auth / not-found / gone answer from the Hub is a settled verdict:
# the token is wrong, the repo is gated, or it isn't there. Retrying
@@ -770,6 +775,8 @@ async def install_model(req: InstallModelRequest):
except _InstallCancelled:
raise
except Exception as _seg_err:
if is_disk_full_error(_seg_err):
raise # the plain path would hit the same wall
_segmented_off, _seg_reraise = _segmented_retry_plan(
_seg_err, _attempt, _max_attempts
)
@@ -915,6 +922,9 @@ async def install_model(req: InstallModelRequest):
from core.failure import append_hint, classify
_error = append_hint(str(e))
_docs_topic = classify(str(e))
if is_disk_full_error(e):
_error = disk_full_message()
_docs_topic = "DISK_SPACE_LOW"
# Gated catalogue entries own their recovery topic. Hugging Face
# uses several exception wordings for the same access verdict, so
# the UI must not depend on parsing an English 401/403 message.
@@ -931,6 +941,7 @@ async def install_model(req: InstallModelRequest):
"HF_AUTH_FAILED",
"PYANNOTE_LICENSE_REQUIRED",
"POCKETTTS_GATED_WEIGHTS",
"DISK_SPACE_LOW", # freeing space, not waiting, is the fix
}:
_install_cooldowns.pop(req.repo_id, None)
_install_failures[req.repo_id] = {
+13
View File
@@ -276,6 +276,19 @@ def disk_space_error(to_download_bytes: "int | None", *, cache_dir: "str | None"
)
def disk_full_message(*, cache_dir: "str | None" = None) -> str:
"""Actionable text for a download that ran out of space *mid-flight*
(the preflight guard can't see other writers or an unknown plan size)."""
cache = cache_dir or hf_cache_dir()
free = disk_free_bytes(cache)
have = f"only {free / _GIB:.1f} GB is free" if free > 0 else "the disk is full"
return (
f"The disk filled up while downloading: {have} at {cache}. Free up space "
"(or move the model cache to a bigger volume) and retry; the download "
"resumes from the part that already finished."
)
def _repo_dir_name(repo_id: str) -> str:
"""HF cache dir name for a repo: 'k2-fsa/OmniVoice' → 'models--k2-fsa--OmniVoice'."""
return "models--" + repo_id.replace("/", "--")
+103 -25
View File
@@ -114,6 +114,78 @@ def _next_counter(db_path: str, safe_version: str) -> int:
return highest + 1
def _reserve_snapshot(db_path: str, safe_version: str) -> tuple[str, str]:
"""Claim a counter atomically across threads/processes before copying.
Reservations are not backup files. An abandoned reservation consumes one
counter but never gets offered as recovery data or overwritten by a retry.
"""
counter = _next_counter(db_path, safe_version)
while True:
target = f"{db_path}.backup-{safe_version}-{counter}"
reservation = target + ".reserve"
try:
fd = os.open(reservation, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
except FileExistsError:
counter += 1
continue
try:
os.write(fd, str(os.getpid()).encode("ascii")) # ownership for stale_reservations
finally:
os.close(fd)
if os.path.exists(target):
os.unlink(reservation)
counter += 1
continue
return target, reservation
#: Fallback only for a reservation whose owner PID was never recorded (a writer
#: that died between creating the file and writing its PID).
_RESERVATION_MAX_AGE_S = 24 * 3600
def _reservation_owner(path: str) -> int | None:
try:
with open(path, "r", encoding="ascii") as fh:
pid = int(fh.read().strip())
except (OSError, ValueError):
return None
return pid if pid > 0 else None
def stale_reservations(db_path: str) -> list[str]:
"""``.reserve`` markers whose writer is confirmed gone.
A reservation owned by a live process (including this one, other threads)
is never listed, however old: a snapshot can outlive a long suspend, and
removing its marker would let another startup claim the same counter.
``_pid_alive`` errs toward "alive". Only a marker with no recorded owner
falls back to an age check."""
directory = os.path.dirname(os.path.abspath(db_path)) or "."
base = os.path.basename(db_path)
try:
names = os.listdir(directory)
except OSError:
return []
stale = []
now = time.time()
for name in names:
if not (name.startswith(base + ".backup-") and name.endswith(".reserve")):
continue
if not _BACKUP_SUFFIX_RE.fullmatch(name[: -len(".reserve")][len(base):]):
continue
path = os.path.join(directory, name)
owner = _reservation_owner(path)
if owner is not None:
if owner == os.getpid() or _pid_alive(owner):
continue
stale.append(path)
elif now - _mtime(path) > _RESERVATION_MAX_AGE_S:
stale.append(path)
return stale
def stale_partial_backups(db_path: str) -> list[str]:
"""``<db>.backup-<version>-<n>.part-<pid>`` files whose writer is gone.
@@ -158,13 +230,13 @@ def prune_backups(db_path: str, keep: int = KEEP_BACKUPS) -> list[str]:
"""Delete all but the ``keep`` newest backups, plus torn partial
snapshots from earlier runs. Returns deleted paths."""
deleted = []
for path in stale_partial_backups(db_path):
for path in stale_partial_backups(db_path) + stale_reservations(db_path):
try:
os.remove(path)
deleted.append(path)
logger.info("Removed torn partial DB backup %s", path)
logger.info("Removed stale DB backup leftover %s", path)
except OSError as exc:
logger.warning("Could not remove partial DB backup %s: %s", path, exc)
logger.warning("Could not remove DB backup leftover %s: %s", path, exc)
for path in list_backups(db_path)[keep:]:
try:
os.remove(path)
@@ -195,32 +267,38 @@ def snapshot_before_migration(db_path: str, version: str) -> str | None:
return None
safe_version = _sanitize_version(version)
target = f"{db_path}.backup-{safe_version}-{_next_counter(db_path, safe_version)}"
target, reservation = _reserve_snapshot(db_path, safe_version)
tmp = f"{target}.part-{os.getpid()}"
src = sqlite3.connect(db_path)
try:
dst = sqlite3.connect(tmp)
src = sqlite3.connect(db_path)
try:
# Online backup: consistent snapshot including WAL contents.
src.backup(dst)
dst.commit()
dst = sqlite3.connect(tmp)
try:
# Online backup: consistent snapshot including WAL contents.
src.backup(dst)
dst.commit()
finally:
dst.close()
except BaseException:
try:
os.remove(tmp)
except OSError:
pass # best effort: the torn temp is pruned later; the backup error below matters more
raise
finally:
dst.close()
except BaseException:
src.close()
os.replace(tmp, target)
# A same-second rotation must still rank the new file newest.
try:
os.remove(tmp)
now = time.time()
os.utime(target, (now, now))
except OSError:
pass
raise
pass # ordering hint only; the snapshot itself is already in place
logger.info("Pre-migration DB backup written: %s (%.1f MB)", target, size / (1024 * 1024))
prune_backups(db_path)
return target
finally:
src.close()
os.replace(tmp, target)
# A same-second rotation must still rank the new file newest.
try:
now = time.time()
os.utime(target, (now, now))
except OSError:
pass
logger.info("Pre-migration DB backup written: %s (%.1f MB)", target, size / (1024 * 1024))
prune_backups(db_path)
return target
try:
os.unlink(reservation)
except OSError:
pass # marker already gone; pruning handles any leftover
+39
View File
@@ -15,6 +15,7 @@ Guarantees:
"""
from __future__ import annotations
import errno
import os
import platform
import re
@@ -274,6 +275,44 @@ _HF_CONTEXT_MARKERS = (
)
_DISK_FULL_SIGNATURES = (
"errno 28", "no space left",
# Linux says "Disk quota exceeded"; macOS says "Disc quota exceeded".
"disk quota exceeded", "disc quota exceeded",
"winerror 112", "winerror 39", "not enough space on the disk",
)
# EDQUOT is 122 on Linux and 69 on macOS (absent on Windows) — use the platform's.
_DISK_FULL_ERRNOS = frozenset(
n for n in (getattr(errno, "ENOSPC", None), getattr(errno, "EDQUOT", None)) if n is not None
)
def is_disk_full_error(reason: "BaseException | str | None") -> bool:
"""True when an install/download died because the volume is full.
Accepts an exception (checks ``errno``/``winerror`` through the
``__cause__``/``__context__`` chain, so a wrapped ``OSError`` still counts) or
the text of one. Disk-full is not transient: retrying with backoff only
delays the message, and the generic "network hiccup" hints are wrong for it.
Never raises."""
try:
if isinstance(reason, BaseException):
seen: set[int] = set()
exc: "BaseException | None" = reason
while exc is not None and id(exc) not in seen:
seen.add(id(exc))
if getattr(exc, "errno", None) in _DISK_FULL_ERRNOS or getattr(exc, "winerror", None) in (39, 112):
return True
if any(sig in str(exc).lower() for sig in _DISK_FULL_SIGNATURES):
return True
exc = exc.__cause__ or exc.__context__
return False
low = (reason or "").lower()
return any(sig in low for sig in _DISK_FULL_SIGNATURES)
except Exception:
return False
def is_hf_connectivity_error(reason: Optional[str]) -> bool:
"""True when *reason* looks like a network/connectivity failure of an HF
download (DNS, refused/reset connections, timeouts, hub locate errors).
+33
View File
@@ -49,6 +49,39 @@ def safe_filename(value: object) -> str:
return name
_PORTABLE_INVALID_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f\x7f]')
def portable_filename(value: object, default: str = "file", max_bytes: int = 200) -> str:
"""Turn arbitrary text (a video title, a voice name) into a filename every
desktop OS accepts. ``safe_filename`` *validates*; this *repairs*.
Windows rejects ``< > : " / \\ | ? *``, control characters, trailing
dots/spaces and device names (``CON``, ``NUL``...) with ``[Errno 22] Invalid
argument``; titles like ``"How to X: a guide?"`` hit that on the first
export. The extension survives truncation, which is by UTF-8 bytes so a CJK
title cannot overrun the 255-byte name limit of ext4/APFS/NTFS.
"""
name = _PORTABLE_INVALID_CHARS.sub("_", str(value or "")).strip(" .")
stem, dot, ext = name.rpartition(".")
if not dot or not stem or len(ext) > 16:
stem, ext = name, ""
else:
ext = "." + ext
budget = max(2, max_bytes - len(ext.encode("utf-8")))
def _fit(text: str) -> str:
return text.encode("utf-8")[:budget].decode("utf-8", "ignore").rstrip(" .")
stem = _fit(stem) or default
if not stem.strip("_ "):
stem = default
# Check device names AFTER truncation: cutting a long stem can expose "CON".
if stem.split(".", 1)[0].rstrip().upper() in _WINDOWS_RESERVED_NAMES:
stem = _fit("_" + stem)
return stem + ext
def resolve_within(root: os.PathLike[str] | str, value: os.PathLike[str] | str) -> Path:
"""Resolve *value* beneath *root*, rejecting traversal and symlink escapes.
+39 -5
View File
@@ -21,6 +21,7 @@ from __future__ import annotations
import asyncio
import json
import os
import re
from typing import Callable, Optional
import httpx
@@ -100,11 +101,23 @@ def _manifest_path(part: str) -> str:
def _load_done(part: str, size: int) -> set[tuple[int, int]]:
try:
if os.path.getsize(part) != size:
return set() # missing/truncated bytes cannot be certified by a sidecar
with open(_manifest_path(part)) as f:
data = json.load(f)
if data.get("size") != size:
if not isinstance(data, dict) or data.get("size") != size:
return set()
return {tuple(s) for s in data.get("done", [])}
ranges = data.get("done", [])
if not isinstance(ranges, list):
return set()
done = set()
for segment in ranges:
if (not isinstance(segment, list) or len(segment) != 2
or any(type(value) is not int for value in segment)
or not 0 <= segment[0] <= segment[1] < size):
return set()
done.add(tuple(segment))
return done
except (OSError, ValueError):
return set()
@@ -170,6 +183,12 @@ async def segmented_download(
headers = {**_auth_headers(final_url, token), "Range": f"bytes={start}-{end}"}
async with client.stream("GET", final_url, headers=headers) as r:
r.raise_for_status()
match = re.fullmatch(r"bytes\s+([0-9]+)-([0-9]+)/([0-9]+|\*)",
r.headers.get("content-range", "").strip(), re.IGNORECASE)
if (r.status_code != 206 or match is None
or (int(match[1]), int(match[2])) != (start, end)
or (match[3] != "*" and int(match[3]) != size)):
raise ValueError(f"invalid response range for bytes {start}-{end}/{size}")
got = 0
with open(part, "r+b") as fh:
fh.seek(start)
@@ -231,11 +250,26 @@ async def _stream_single(client, url, token, part, on_bytes, cancelled) -> None:
on_bytes(len(chunk))
def _invalidate_done(part: str) -> None:
try:
os.remove(_manifest_path(part))
except FileNotFoundError:
pass # already absent: the goal of invalidation is met
def _preallocate(part: str, size: int) -> None:
# Create/extend the file to `size` so segment writes can seek to offsets.
with open(part, "a+b") as fh:
# A checkpoint only certifies the original bytes, not a resized replacement.
try:
fh = open(part, "r+b")
except FileNotFoundError:
# Invalidate before creating a missing partial, too: an unlink failure
# must leave both the original partial state and destination untouched.
_invalidate_done(part)
fh = open(part, "a+b")
with fh:
fh.seek(0, os.SEEK_END)
if fh.tell() < size:
if fh.tell() != size:
_invalidate_done(part)
fh.truncate(size)
+47 -4
View File
@@ -57,7 +57,7 @@ import time
from collections import deque
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, NamedTuple, Optional
from typing import Callable, Iterable, NamedTuple, Optional
from core.config import DATA_DIR
from core.contained_subprocess import OwnedPopen, WindowsJobPopen, spawn_owned
@@ -939,6 +939,35 @@ def _job_step(job: dict, step_id: str) -> dict:
return next(s for s in job["steps"] if s["id"] == step_id)
_DISK_FULL_REMEDIATION = (
"The disk is full. Free up space (or move VoiceStudio's data directory to a "
"larger volume), then re-run the install — it resumes from where it stopped."
)
def _redact(exc: BaseException) -> str:
"""Exception text with home-directory paths and tokens scrubbed, since a
step error is persisted in the job record and log."""
from core.failure import sanitize
return sanitize(str(exc))
def _is_disk_full(exc: BaseException) -> bool:
from core.failure import is_disk_full_error
return is_disk_full_error(exc)
def _output_shows_disk_full(lines: Iterable[str]) -> bool:
"""uv reports a full volume only as output text with a bare non-zero exit.
Pass the failing process's own output, never the shared job log: an earlier
step's recovered "No space left" line must not relabel a later failure."""
from core.failure import is_disk_full_error
return is_disk_full_error("\n".join(lines))
def _log(job: dict, line: str) -> None:
line = line.rstrip()
if line:
@@ -950,7 +979,7 @@ def _log(job: dict, line: str) -> None:
def _serialize_job(job: Optional[dict]) -> Optional[dict]:
if job is None:
return None
out = dict(job)
out = {k: v for k, v in job.items() if not k.startswith("_")}
with _log_lock:
out["log"] = list(job["log"])
out["steps"] = [dict(s) for s in job["steps"]]
@@ -1163,8 +1192,10 @@ def _run_install(spec: SidecarSpec, job: dict) -> None:
raise
except Exception as exc: # noqa: BLE001 — surfaced into the job
step["state"] = "error"
if _is_disk_full(exc):
raise _StepError(f"{type(exc).__name__}: {_redact(exc)}", _DISK_FULL_REMEDIATION) from exc
raise _StepError(
f"{type(exc).__name__}: {exc}",
f"{type(exc).__name__}: {_redact(exc)}",
"Re-run the install — it resumes from where it stopped. If it "
f"keeps failing, see {spec.docs_path} for the manual steps.",
) from exc
@@ -1505,6 +1536,7 @@ def _step_install_deps(spec: SidecarSpec, job: dict) -> None:
target += list(UV_PIP_CU128_ARGS)
# Always `--python <this engine's venv>`: the install can only ever land in
# the venv this engine owns, never the app's interpreter.
job.pop("_last_run_output", None)
rc = _run_logged(
job,
[uv, "pip", "install", "--python", str(py), *target],
@@ -1512,6 +1544,9 @@ def _step_install_deps(spec: SidecarSpec, job: dict) -> None:
env=uv_subprocess_env(Path(DATA_DIR) / "engines"),
)
if rc != 0:
if _output_shows_disk_full(job.get("_last_run_output") or ()):
raise _StepError(f"uv pip install failed (exit {rc}): no space left on device.",
_DISK_FULL_REMEDIATION)
hint = (
"Usually a network hiccup — re-run the install to resume. Behind a "
"proxy, set HTTPS_PROXY in Settings → Environment first."
@@ -1681,8 +1716,10 @@ def _step_fetch_weights(spec: SidecarSpec, job: dict) -> None:
try:
snapshot_download(**kwargs) # nosec B615 — deliberate default-branch policy, see above
except Exception as exc:
if _is_disk_full(exc):
raise _StepError(f"Model weight download failed: {_redact(exc)}", _DISK_FULL_REMEDIATION) from exc
raise _StepError(
f"Model weight download failed: {exc}",
f"Model weight download failed: {_redact(exc)}",
"Re-run the install — the download resumes where it stopped. "
"Check Settings → Network (HF endpoint / proxy) if it keeps failing.",
) from exc
@@ -1727,6 +1764,9 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
Returns the exit code; -1 on timeout (process tree killed) or spawn
failure. argv-list only — never a shell string — so paths with spaces
are safe on every platform. ``env=None`` inherits the parent environment.
This process's own lines are also kept in ``job["_last_run_output"]`` (reset
per call, hidden from the status payload) so a caller can diagnose its own
failure without reading earlier steps' log output.
The stdout drain runs on its own daemon thread and the main flow blocks
on ``proc.wait(timeout=…)``. That bounds the step even when a grandchild
@@ -1738,6 +1778,8 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
# starts. POSIX links it to backend death through a control pipe; Windows
# retains a kill-on-close Job handle in this backend process.
popen_kwargs = _install_containment_kwargs()
tail: deque[str] = deque(maxlen=60)
job["_last_run_output"] = tail
try:
proc = spawn_owned(
argv,
@@ -1758,6 +1800,7 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
assert proc.stdout is not None
for line in proc.stdout:
_log(job, line)
tail.append(line)
except (OSError, ValueError):
pass # pipe closed by the timeout kill — nothing left to read
+57 -15
View File
@@ -23,6 +23,7 @@ from __future__ import annotations
import glob
import os
import shutil
import stat
import sys
import tempfile
import threading
@@ -62,6 +63,16 @@ def default_engines_dir() -> str:
return str(Path(DATA_DIR) / "engines")
def _has_venv(entry_path: str) -> bool:
"""True when ``<entry>/.venv`` is a directory. Only a *missing* path means
"no venv"; a permission or I/O error propagates so the caller can mark the
scan incomplete instead of silently reclassifying an installed engine."""
try:
return stat.S_ISDIR(os.stat(os.path.join(entry_path, ".venv")).st_mode)
except (FileNotFoundError, NotADirectoryError):
return False
def _engines_child_name(engines_dir: str, data_dir: str) -> str | None:
"""Basename of ``engines_dir`` when it is a direct child of ``data_dir`` —
so the data category can skip it and not double-count what the engine-venv
@@ -200,7 +211,7 @@ def build_report(
})
def _finish(category_id: str, cat: dict, complete: bool, err_path: str | None) -> None:
cat["complete"] = complete
cat["complete"] = complete and err_path is None
if not complete:
_warn_unreadable(category_id, cat["path"], "timeout")
if err_path is not None:
@@ -263,6 +274,33 @@ def build_report(
_finish("hf_cache", hf_cat, hf_complete, hf_err)
categories.append(hf_cat)
# Use one ownership snapshot for both categories: only installed sidecars
# belong to engine_venvs. Interrupted installs inside DATA_DIR remain data.
engine_err: str | None = None
try:
with os.scandir(engines_dir) as it:
engine_entries = list(it)
except FileNotFoundError:
engine_entries = []
except OSError:
engine_entries = []
engine_err = engines_dir
engine_dirs = []
unclassified_engines = set()
for e in engine_entries:
# Resolve a transient inspection failure before either category claims
# the entry. Persistently unknown ownership remains an incomplete scan.
for _ in range(2):
try:
if e.is_dir(follow_symlinks=False) and _has_venv(e.path):
engine_dirs.append(e.path)
break
except OSError:
engine_err = engine_err or e.path
else:
unclassified_engines.add(e.path)
engine_dirs.sort()
# ── 2. App data dir, broken into subtotals ─────────────────────────────
deadline = time.monotonic() + category_timeout
data_complete = True
@@ -271,10 +309,11 @@ def build_report(
claimed: set[str] = set()
# When sidecar engines live under DATA_DIR/engines, the engine-venv category
# below owns that subtree — claim it here so it isn't also swept into "other".
# below owns installed sidecars; count the unclaimed remainder separately.
engines_child = _engines_child_name(engines_dir, data_dir)
if engines_child:
claimed.add(engines_child)
data_err = engine_err
for name in _DATA_CHILD_DIRS:
p = os.path.join(data_dir, name)
@@ -322,6 +361,20 @@ def build_report(
except OSError:
if os.path.exists(data_dir):
data_err = data_err or data_dir
if engines_child:
for e in engine_entries:
if e.path in engine_dirs or e.path in unclassified_engines:
continue
try:
if e.is_dir(follow_symlinks=False):
size, ok, err = _dir_size(e.path, deadline)
other_bytes += size
data_complete = data_complete and ok
data_err = data_err or err
else:
other_bytes += e.stat(follow_symlinks=False).st_size
except OSError:
data_err = data_err or e.path
children.append({"id": "other", "path": data_dir, "bytes": other_bytes, "complete": True})
data_cat = {
@@ -338,21 +391,10 @@ def build_report(
deadline = time.monotonic() + category_timeout
venv_total = 0
venv_complete = True
venv_err: str | None = None
venv_err: str | None = engine_err
venv_items: list[dict] = []
try:
with os.scandir(engines_dir) as it:
engine_dirs = sorted(e.path for e in it if e.is_dir(follow_symlinks=False))
except OSError:
engine_dirs = []
for edir in engine_dirs:
# A sidecar install is the venv PLUS a git checkout PLUS multi-GB weights
# (`checkpoints/`) — measure the whole `<id>` dir, not just `.venv`, or the
# weights (usually the bulk) go uncounted now that the data category no
# longer sweeps this subtree into "other". Only real installs have a venv,
# so that gate still skips a bare/interrupted dir.
if not os.path.isdir(os.path.join(edir, ".venv")):
continue
# Measure the whole installed sidecar: environment, checkout and weights.
size, ok, err = _dir_size(edir, deadline)
venv_total += size
venv_complete = venv_complete and ok
+4
View File
@@ -15,3 +15,7 @@ The application data location uses Electron's native folder picker and a main-pr
Remove all data scans the backend data root, Electron runtime/configuration, logs, durable environment and model cache with real sizes. Shared Hugging Face caches remain an explicit opt-in. After typed confirmation, main rescans and validates every root, stops the backend and hands the exact plan to the signed desktop helper. The helper canonicalizes every path, waits for Electron to exit, then removes the locked Chromium/runtime tree without following a path alias outside VoiceStudio-owned data. A failed helper launch restores the backend and keeps the app open; a successful handoff quits immediately. Removing the installed application binary remains the operating system's normal uninstall step.
`electron/tests/storage-settings-smoke.mjs` verifies warning/partial-scan display, folder reveal, cancel/confirm cleanup, partial cleanup failure and backup status with mocked mutations. Live read-only reports returned all four categories, one volume and an existing backup. No actual files were deleted. Tauri storage regression tests pass after shared-helper extraction. Connection, performance and privacy browser tests also pass after the settings-navigation refactor.
Interrupted sidecar installs without an environment remain included in application data. Completed sidecar environments, checkouts and weights are counted once in the engine category.
An unreadable engine directory or entry produces an incomplete report and a warning; unavailable bytes are not presented as a complete empty footprint.
+12
View File
@@ -694,6 +694,14 @@ order:
- macOS/Linux: `export HF_ENDPOINT=https://hf-mirror.com`
- Windows (PowerShell): `[Environment]::SetEnvironmentVariable("HF_ENDPOINT","https://hf-mirror.com","User")`
A segmented download refuses a response whose status or Content-Range does not match the requested bytes and file size. An invalid response is not published as the model file; retry through a server or mirror that supports correct byte ranges.
Segmented download resume records are reused only with an existing partial file of the expected size and valid byte-range entries. If a partial file is missing, truncated, or oversized, or its sidecar is malformed, the download fetches those bytes again instead of treating preallocated zeros as completed data. Oversized partial files are resized before restarting so old trailing bytes cannot prevent verification of the new download. Stale checkpoints are removed before resizing or recreating partial files, so a failed fetch cannot make the next retry trust stale or zero-filled bytes. If that stale checkpoint cannot be removed, the restart stops before changing the partial file or destination.
**Disk filled up mid-install.** A model or engine install that runs out of space stops immediately (it is not retried with backoff) and reports how much space is free and where; free space or move the model cache to a larger volume, then retry. The download resumes from the part that already finished. During first-run setup, the one-time `uv` installer download is attempted up to three times (two retries) on connection resets, timeouts, and HTTP 5xx/429 before it reports failure.
**Exports named after a video title.** Download and export names built from a video title replace characters Windows rejects (`< > : " / \ | ? *`, control characters, trailing dots/spaces, device names such as `CON`) with `_` on every OS, so `How to X: a guide?` exports as `How to X_ a guide_` instead of failing with `[Errno 22] Invalid argument`.
**Manual fallback** (if downloads keep failing), pull the weights yourself into
the same cache, then relaunch:
@@ -1322,3 +1330,7 @@ prove that torch and torchvision versions are mismatched. Save the diagnostic
bundle and check package versions in the environment running the backend before
reinstalling anything. Faster Whisper is an alternative when only transcription
is affected; it does not diagnose or repair the original environment.
## Concurrent migration backups
Concurrent pre-migration database snapshots reserve distinct backup counters before copying. Reservation files are not recovery backups. A reservation left by an interrupted writer is skipped by subsequent snapshots rather than reused.
+3 -2
View File
@@ -1,4 +1,5 @@
import { registerSiteBrowser } from './site-browser';
import { portableFilename } from './portable-filename';
import { saveFiltersFor } from './save-filters';
import { resolveBackendDownloadUrl } from './backend-download';
import {
@@ -150,7 +151,7 @@ function assertSaveAudioRequest(value: unknown): SaveAudioRequest {
}
return {
url: assertString(req.url, 'audio url'),
suggestedName: assertString(req.suggestedName, 'file name'),
suggestedName: portableFilename(assertString(req.suggestedName, 'file name')),
method: req.method,
};
}
@@ -161,7 +162,7 @@ function assertSaveDataRequest(value: unknown): SaveDataRequest {
if (!(req.data instanceof Uint8Array)) throw new Error('Invalid file data');
return {
data: req.data,
suggestedName: assertString(req.suggestedName, 'file name'),
suggestedName: portableFilename(assertString(req.suggestedName, 'file name')),
};
}
@@ -0,0 +1,29 @@
// @vitest-environment node
import { expect, it } from 'vitest';
import { portableFilename } from './portable-filename';
it('replaces characters Windows rejects and keeps the extension', () => {
expect(portableFilename('How to X: a guide?_en.mp4')).toBe('How to X_ a guide__en.mp4');
expect(portableFilename('a<b>c|d"e*f.srt')).toBe('a_b_c_d_e_f.srt');
});
it('repairs trailing dots/spaces, device names and empty names', () => {
expect(portableFilename('clip. .')).toBe('clip');
expect(portableFilename('CON.wav')).toBe('_CON.wav');
expect(portableFilename('???', 'audio')).toBe('audio');
expect(portableFilename('', 'audio.wav')).toBe('audio.wav');
});
it('repairs a device name exposed by truncation', () => {
const name = portableFilename(`CON${' '.repeat(197)}x.wav`);
expect(name.split('.')[0].toUpperCase()).not.toBe('CON');
expect(name.endsWith('.wav')).toBe(true);
expect(new TextEncoder().encode(name).length).toBeLessThanOrEqual(200);
});
it('truncates by UTF-8 bytes without splitting a character or losing the extension', () => {
const name = portableFilename('\u65e5\u672c\u8a9e'.repeat(60) + '.mp4');
expect(new TextEncoder().encode(name).length).toBeLessThanOrEqual(200);
expect(name.endsWith('.mp4')).toBe(true);
expect(name).not.toContain('�');
});
+43
View File
@@ -0,0 +1,43 @@
// Mirrors backend/core/path_security.portable_filename: one rule on every OS so an
// export named after a video title ("How to X: a guide?") cannot hit Windows'
// [Errno 22] Invalid argument. Keep the two in sync (tests cover the same cases).
const INVALID = /[<>:"/\\|?*\u0000-\u001f\u007f]/g;
const RESERVED = new Set([
'CON',
'PRN',
'AUX',
'NUL',
...Array.from({ length: 9 }, (_, i) => `COM${i + 1}`),
...Array.from({ length: 9 }, (_, i) => `LPT${i + 1}`),
]);
const encoder = new TextEncoder();
function truncateBytes(text: string, budget: number): string {
let out = '';
let used = 0;
for (const ch of text) {
const size = encoder.encode(ch).length;
if (used + size > budget) break;
out += ch;
used += size;
}
return out;
}
export function portableFilename(value: string, fallback = 'file', maxBytes = 200): string {
const name = value.replace(INVALID, '_').replace(/^[ .]+|[ .]+$/g, '');
const dot = name.lastIndexOf('.');
let stem = name;
let ext = '';
if (dot > 0 && name.length - dot - 1 <= 16 && name.length - dot > 1) {
stem = name.slice(0, dot);
ext = name.slice(dot);
}
const budget = Math.max(2, maxBytes - encoder.encode(ext).length);
const fit = (text: string) => truncateBytes(text, budget).replace(/[ .]+$/, '');
stem = fit(stem) || fallback;
if (!stem.replace(/[_ ]/g, '')) stem = fallback;
// Check device names AFTER truncation: cutting a long stem can expose "CON".
if (RESERVED.has(stem.split('.')[0].trimEnd().toUpperCase())) stem = fit(`_${stem}`);
return stem + ext;
}
+25 -2
View File
@@ -75,10 +75,33 @@ it('rejects oversized scripts and HTTP failures', async () => {
await expect(
downloadRuntimeInstaller('https://astral.sh', {}, new AbortController().signal),
).rejects.toThrow('size limit');
reply(503, 'unavailable');
for (let i = 0; i < 3; i += 1) reply(503, 'unavailable');
await expect(
downloadRuntimeInstaller('https://astral.sh', {}, new AbortController().signal),
downloadRuntimeInstaller('https://astral.sh', {}, new AbortController().signal, 0),
).rejects.toThrow('(503)');
expect(get).toHaveBeenCalledTimes(4); // 1 oversized + 3 bounded attempts
});
it('retries transient failures so a flaky first-run network still bootstraps', async () => {
reply(503, 'unavailable');
vi.mocked(get).mockImplementationOnce((() => {
const request = new EventEmitter();
queueMicrotask(() =>
request.emit('error', Object.assign(new Error('socket hang up'), { code: 'ECONNRESET' })),
);
return request;
}) as unknown as typeof get);
reply(200, '# installer');
expect(
await downloadRuntimeInstaller('https://astral.sh', {}, new AbortController().signal, 0),
).toBe('# installer');
expect(get).toHaveBeenCalledTimes(3);
});
it('does not retry definitive answers such as HTTP 404', async () => {
reply(404, 'missing');
await expect(
downloadRuntimeInstaller('https://astral.sh', {}, new AbortController().signal, 0),
).rejects.toThrow('(404)');
expect(get).toHaveBeenCalledTimes(1);
});
it('does not start cancelled downloads', async () => {
const controller = new AbortController();
+69 -8
View File
@@ -43,14 +43,74 @@ export function setupProxyForUrl(raw: string, env: NodeJS.ProcessEnv): string {
return candidate;
}
/** Download executable installer text only over HTTPS, including redirects. */
const INSTALLER_ATTEMPTS = 3;
const INSTALLER_ATTEMPT_TIMEOUT_MS = 60_000;
/** Marks failures a fresh attempt can plausibly cure (resets, DNS blips, 5xx, 429). */
class TransientDownloadError extends Error {}
function pause(ms: number, signal: AbortSignal): Promise<void> {
return new Promise((resolve, reject) => {
if (signal.aborted) return reject(signal.reason);
const timer = setTimeout(() => {
signal.removeEventListener('abort', onAbort);
resolve();
}, ms);
const onAbort = () => {
clearTimeout(timer);
reject(signal.reason);
};
signal.addEventListener('abort', onAbort, { once: true });
});
}
/** Download executable installer text only over HTTPS, including redirects.
* First-run setup must survive a flaky network, so transient failures get a
* short bounded retry; HTTP 4xx, size/redirect violations and cancellation do not. */
export async function downloadRuntimeInstaller(
url: string,
env: NodeJS.ProcessEnv,
signal: AbortSignal,
backoffMs = 1_000,
): Promise<string> {
const agent = new ProxyAgent({ getProxyForUrl: (target) => setupProxyForUrl(target, env) });
const bounded = AbortSignal.any([signal, AbortSignal.timeout(60_000)]);
try {
for (let attempt = 1; ; attempt += 1) {
try {
return await downloadOnce(url, agent, signal);
} catch (error) {
signal.throwIfAborted();
if (attempt >= INSTALLER_ATTEMPTS || !isTransient(error)) throw error;
await pause(backoffMs * attempt, signal);
}
}
} finally {
agent.destroy();
}
}
function isTransient(error: unknown): boolean {
if (error instanceof TransientDownloadError) return true;
const code = (error as NodeJS.ErrnoException | undefined)?.code;
// The per-attempt timeout surfaces as an AbortError/TimeoutError; user cancel was
// already rethrown by the caller.
const name = (error as Error | undefined)?.name;
return (
name === 'TimeoutError' ||
name === 'AbortError' ||
[
'ECONNRESET',
'ECONNREFUSED',
'ETIMEDOUT',
'EAI_AGAIN',
'ENOTFOUND',
'EPIPE',
'ECONNABORTED',
].includes(code ?? '')
);
}
async function downloadOnce(url: string, agent: ProxyAgent, signal: AbortSignal): Promise<string> {
const bounded = AbortSignal.any([signal, AbortSignal.timeout(INSTALLER_ATTEMPT_TIMEOUT_MS)]);
async function download(target: string, redirects = 0): Promise<string> {
if (new URL(target).protocol !== 'https:') throw new Error('uv installer requires HTTPS');
bounded.throwIfAborted();
@@ -72,7 +132,12 @@ export async function downloadRuntimeInstaller(
}
if (status !== 200) {
response.resume();
reject(new Error(`uv installer download failed (${status})`));
const message = `uv installer download failed (${status})`;
reject(
status >= 500 || status === 429
? new TransientDownloadError(message)
: new Error(message),
);
return;
}
const chunks: Buffer[] = [];
@@ -89,9 +154,5 @@ export async function downloadRuntimeInstaller(
request.on('error', reject);
});
}
try {
return await download(url);
} finally {
agent.destroy();
}
return download(url);
}
+96
View File
@@ -0,0 +1,96 @@
"""Concurrent migration snapshots must each retain a usable recovery file."""
import concurrent.futures
import sqlite3
import threading
def test_concurrent_snapshots_cannot_share_a_recovery_slot(tmp_path, monkeypatch):
from core import db_backup
path = tmp_path / 'voices.db'
with sqlite3.connect(path) as connection:
connection.execute('CREATE TABLE voices(name TEXT)')
connection.execute("INSERT INTO voices VALUES ('keep my voice')")
counter = db_backup._next_counter
boundary = threading.Barrier(2)
def simultaneous_counter(*args):
value = counter(*args)
boundary.wait(timeout=5)
return value
monkeypatch.setattr(db_backup, '_next_counter', simultaneous_counter)
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(db_backup.snapshot_before_migration, str(path), '1.0') for _ in range(2)]
results = [future.result(timeout=15) for future in futures]
assert len(set(results)) == 2
assert set(db_backup.list_backups(str(path))) == set(results)
for backup in results:
with sqlite3.connect(backup) as connection:
assert connection.execute('SELECT name FROM voices').fetchall() == [('keep my voice',)]
assert not list(tmp_path.glob('*.reserve'))
def test_abandoned_reservation_is_not_reused_or_listed(tmp_path):
from core import db_backup
path = tmp_path / 'voices.db'
with sqlite3.connect(path) as connection:
connection.execute('CREATE TABLE voices(name TEXT)')
abandoned = tmp_path / 'voices.db.backup-1.0-1.reserve'
abandoned.touch()
backup = db_backup.snapshot_before_migration(str(path), '1.0')
assert backup.endswith('.backup-1.0-2')
assert db_backup.list_backups(str(path)) == [backup]
assert abandoned.exists()
def test_failed_copy_releases_its_reservation(tmp_path, monkeypatch):
from core import db_backup
import pytest
path = tmp_path / 'voices.db'
with sqlite3.connect(path) as connection:
connection.execute('CREATE TABLE voices(name TEXT)')
connect = sqlite3.connect
def fail_destination(filename, *args, **kwargs):
if '.part-' in str(filename):
raise sqlite3.OperationalError('destination unavailable')
return connect(filename, *args, **kwargs)
monkeypatch.setattr(sqlite3, 'connect', fail_destination)
with pytest.raises(sqlite3.OperationalError, match='destination unavailable'):
db_backup.snapshot_before_migration(str(path), '1.0')
assert not list(tmp_path.glob('*.reserve'))
assert db_backup.list_backups(str(path)) == []
def test_reservations_are_pruned_only_when_the_owner_is_gone(tmp_path, monkeypatch):
import os
import time
from core import db_backup
path = tmp_path / 'voices.db'
path.write_bytes(b'')
dead = tmp_path / 'voices.db.backup-1.0-1.reserve'
dead.write_text('999999')
live_old = tmp_path / 'voices.db.backup-1.0-2.reserve' # owner alive, however old
live_old.write_text('4242')
mine = tmp_path / 'voices.db.backup-1.0-3.reserve'
mine.write_text(str(os.getpid()))
ownerless_old = tmp_path / 'voices.db.backup-1.0-4.reserve'
ownerless_old.write_bytes(b'')
ownerless_fresh = tmp_path / 'voices.db.backup-1.0-5.reserve'
ownerless_fresh.write_bytes(b'')
past = time.time() - 48 * 3600
for p in (dead, live_old, ownerless_old):
os.utime(p, (past, past))
monkeypatch.setattr(db_backup, '_pid_alive', lambda pid: pid == 4242)
db_backup.prune_backups(str(path))
assert sorted(p.name for p in tmp_path.iterdir() if p.name.endswith('.reserve')) == sorted(
[live_old.name, mine.name, ownerless_fresh.name]
)
+54
View File
@@ -0,0 +1,54 @@
"""Actual HTTP ranges must prove their identity before a model is published."""
import asyncio
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import pytest
@pytest.mark.parametrize('mode', ['wrong_start', 'wrong_total', 'ignored_range', 'valid', 'valid_unknown_total'])
def test_wrong_range_response_never_commits_model(tmp_path, mode):
from services.segmented_download import segmented_download
content = b'A' * (4 * 1024 * 1024) + b'B' * (4 * 1024 * 1024)
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_HEAD(self):
self.send_response(200)
self.send_header('Content-Length', str(len(content)))
self.send_header('Accept-Ranges', 'bytes')
self.end_headers()
def do_GET(self):
lo, hi = map(int, self.headers['Range'].removeprefix('bytes=').split('-'))
self.send_response(200 if mode == 'ignored_range' else 206)
advertised_lo = lo + 1 if mode == 'wrong_start' else lo
total = '*' if mode == 'valid_unknown_total' else len(content) + 1 if mode == 'wrong_total' else len(content)
self.send_header('Content-Range', f'bytes {advertised_lo}-{hi}/{total}')
self.send_header('Content-Length', str(hi - lo + 1))
self.end_headers()
try:
self.wfile.write(content[lo:hi + 1] if mode.startswith('valid') else b'X' * (hi - lo + 1))
except (BrokenPipeError, ConnectionResetError):
pass # Invalid headers may be rejected before the response body.
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
destination = tmp_path / 'model.bin'
try:
call = segmented_download(f'http://127.0.0.1:{server.server_port}/model', str(destination), num_connections=2)
if mode.startswith('valid'):
asyncio.run(call)
assert destination.read_bytes() == content
else:
with pytest.raises(ValueError, match='range'):
asyncio.run(call)
assert not destination.exists()
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
+238
View File
@@ -0,0 +1,238 @@
"""A checkpoint cannot vouch for model bytes that no longer exist."""
import asyncio
import hashlib
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import httpx
import pytest
@pytest.mark.parametrize('partial', [None, b'abc', b'abcdefgh'])
def test_completed_ranges_require_their_original_complete_partial(tmp_path, partial):
from services.segmented_download import segmented_download
destination = tmp_path / 'model.bin'
path = tmp_path / 'model.bin.part'
if partial is not None:
path.write_bytes(partial)
(tmp_path / 'model.bin.part.done').write_text(json.dumps({'size': 8, 'done': [[0, 7]]}))
fetched = []
def origin(request):
if request.method == 'HEAD':
return httpx.Response(200, headers={'Content-Length': '8', 'Accept-Ranges': 'bytes'})
fetched.append(request.headers['Range'])
return httpx.Response(206, headers={'Content-Range': 'bytes 0-7/8'}, content=b'abcdefgh')
async def download():
async with httpx.AsyncClient(transport=httpx.MockTransport(origin)) as client:
await segmented_download('https://model.example/file', str(destination), client=client)
asyncio.run(download())
assert destination.read_bytes() == b'abcdefgh'
assert fetched == ([] if partial == b'abcdefgh' else ['bytes=0-7'])
@pytest.mark.parametrize('record', [[], {'size': 8, 'done': None}, {'size': 8, 'done': [False]}, {'size': 8, 'done': [[0, 7], [0, 'seven']]}])
def test_malformed_checkpoint_restarts_download(tmp_path, record):
from services.segmented_download import segmented_download
destination = tmp_path / 'model.bin'
(tmp_path / 'model.bin.part').write_bytes(b'oldbytes')
(tmp_path / 'model.bin.part.done').write_text(json.dumps(record))
def origin(request):
if request.method == 'HEAD':
return httpx.Response(200, headers={'Content-Length': '8', 'Accept-Ranges': 'bytes'})
return httpx.Response(206, headers={'Content-Range': 'bytes 0-7/8'}, content=b'newbytes')
async def download():
async with httpx.AsyncClient(transport=httpx.MockTransport(origin)) as client:
await segmented_download('https://model.example/file', str(destination), client=client)
asyncio.run(download())
assert destination.read_bytes() == b'newbytes'
def test_oversized_partial_restarts_from_real_http_ranges(tmp_path):
from services.segmented_download import segmented_download
content = b'abcdefgh'
destination = tmp_path / 'model.bin'
destination.write_bytes(b'previous model')
partial = tmp_path / 'model.bin.part'
partial.write_bytes(content + b'old trailing bytes')
checkpoint = tmp_path / 'model.bin.part.done'
checkpoint.write_text(json.dumps({'size': len(content), 'done': [[0, 7]]}))
fetched = []
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_HEAD(self):
self.send_response(200)
self.send_header('Content-Length', str(len(content)))
self.send_header('Accept-Ranges', 'bytes')
self.end_headers()
def do_GET(self):
fetched.append(self.headers['Range'])
self.send_response(206)
self.send_header('Content-Range', 'bytes 0-7/8')
self.send_header('Content-Length', str(len(content)))
self.end_headers()
self.wfile.write(content)
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
worker = threading.Thread(target=server.serve_forever, daemon=True)
worker.start()
async def download():
async with httpx.AsyncClient(trust_env=False, timeout=5) as client:
await segmented_download(
f'http://127.0.0.1:{server.server_port}/model', str(destination),
expected_etag=hashlib.sha256(content).hexdigest(), client=client,
)
try:
asyncio.run(download())
finally:
server.shutdown()
server.server_close()
worker.join(timeout=5)
assert fetched == ['bytes=0-7']
assert destination.read_bytes() == content
assert destination.stat().st_size == len(content)
assert not partial.exists()
assert not checkpoint.exists()
@pytest.fixture
def retry_range_origin():
content = b'newbytes'
fetched = []
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_HEAD(self):
self.send_response(200)
self.send_header('Content-Length', str(len(content)))
self.send_header('Accept-Ranges', 'bytes')
self.end_headers()
def do_GET(self):
fetched.append(self.headers['Range'])
if len(fetched) == 1:
self.send_response(503)
self.send_header('Content-Length', '0')
self.end_headers()
return
self.send_response(206)
self.send_header('Content-Range', 'bytes 0-7/8')
self.send_header('Content-Length', str(len(content)))
self.end_headers()
self.wfile.write(content)
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
worker = threading.Thread(target=server.serve_forever, daemon=True)
worker.start()
try:
yield f'http://127.0.0.1:{server.server_port}/model', content, fetched
finally:
server.shutdown()
server.server_close()
worker.join(timeout=5)
@pytest.mark.parametrize('partial', [None, b'old', b'oldbytes plus trailing bytes'])
@pytest.mark.parametrize('etag', [None, 'opaque-not-a-sha256'])
def test_retry_cannot_retrust_checkpoint_after_resizing(tmp_path, retry_range_origin, partial, etag):
from services.segmented_download import segmented_download
url, content, fetched = retry_range_origin
destination = tmp_path / 'model.bin'
destination.write_bytes(b'previous verified model')
path = tmp_path / 'model.bin.part'
if partial is not None:
path.write_bytes(partial)
checkpoint = tmp_path / 'model.bin.part.done'
checkpoint.write_text(json.dumps({'size': 8, 'done': [[0, 7]]}))
async def download():
async with httpx.AsyncClient(trust_env=False, timeout=5) as client:
await segmented_download(url, str(destination), expected_etag=etag, client=client)
with pytest.raises(httpx.HTTPStatusError) as first:
asyncio.run(download())
assert first.value.response.status_code == 503
assert destination.read_bytes() == b'previous verified model'
assert path.stat().st_size == len(content)
checkpoint_survived_first_attempt = checkpoint.exists()
asyncio.run(download())
assert fetched == ['bytes=0-7', 'bytes=0-7']
assert destination.read_bytes() == content
assert not checkpoint_survived_first_attempt
assert not path.exists()
assert not checkpoint.exists()
def test_valid_complete_checkpoint_resumes_without_a_get(tmp_path, retry_range_origin):
from services.segmented_download import segmented_download
url, content, fetched = retry_range_origin
destination = tmp_path / 'model.bin'
(tmp_path / 'model.bin.part').write_bytes(content)
(tmp_path / 'model.bin.part.done').write_text(json.dumps({'size': 8, 'done': [[0, 7]]}))
async def download():
async with httpx.AsyncClient(trust_env=False, timeout=5) as client:
await segmented_download(url, str(destination), client=client)
asyncio.run(download())
assert destination.read_bytes() == content
assert fetched == []
@pytest.mark.parametrize('partial', [None, b'old', b'oldbytes plus trailing bytes'])
def test_checkpoint_unlink_failure_preserves_partial_and_destination(tmp_path, retry_range_origin, monkeypatch, partial):
import services.segmented_download as sd
url, _, fetched = retry_range_origin
destination = tmp_path / 'model.bin'
destination.write_bytes(b'previous verified model')
path = tmp_path / 'model.bin.part'
if partial is not None:
path.write_bytes(partial)
checkpoint = tmp_path / 'model.bin.part.done'
record = json.dumps({'size': 8, 'done': [[0, 7]]})
checkpoint.write_text(record)
original_remove = sd.os.remove
def refuse_checkpoint(filename):
if str(filename) == str(checkpoint):
raise PermissionError('checkpoint cannot be removed')
return original_remove(filename)
monkeypatch.setattr(sd.os, 'remove', refuse_checkpoint)
async def download():
async with httpx.AsyncClient(trust_env=False, timeout=5) as client:
await sd.segmented_download(url, str(destination), client=client)
with pytest.raises(PermissionError, match='checkpoint cannot be removed'):
asyncio.run(download())
assert fetched == []
assert destination.read_bytes() == b'previous verified model'
assert checkpoint.read_text() == record
if partial is None:
assert not path.exists()
else:
assert path.read_bytes() == partial
+3 -3
View File
@@ -39,7 +39,7 @@ def _ranged_handler(payload=PAYLOAD, *, accept_ranges=True, record=None):
if rng and accept_ranges:
lo, hi = rng.replace("bytes=", "").split("-")
lo, hi = int(lo), int(hi)
return httpx.Response(206, content=payload[lo:hi + 1])
return httpx.Response(206, headers={"Content-Range": f"bytes {lo}-{hi}/{len(payload)}"}, content=payload[lo:hi + 1])
return httpx.Response(200, content=payload)
return handler
@@ -177,7 +177,7 @@ def test_concurrency_stays_at_num_connections(tmp_path, monkeypatch):
except asyncio.TimeoutError:
pass
lo, hi = request.headers["range"].replace("bytes=", "").split("-")
return httpx.Response(206, content=PAYLOAD[int(lo):int(hi) + 1])
return httpx.Response(206, headers={"Content-Range": f"bytes {lo}-{hi}/{len(PAYLOAD)}"}, content=PAYLOAD[int(lo):int(hi) + 1])
finally:
state["inflight"] -= 1
@@ -215,7 +215,7 @@ def test_dropped_connection_resumes_from_manifest(tmp_path, monkeypatch):
raise httpx.RemoteProtocolError("peer closed connection", request=request)
lo, hi = request.headers["range"].replace("bytes=", "").split("-")
served.append(int(hi) - int(lo) + 1)
return httpx.Response(206, content=PAYLOAD[int(lo):int(hi) + 1])
return httpx.Response(206, headers={"Content-Range": f"bytes {lo}-{hi}/{len(PAYLOAD)}"}, content=PAYLOAD[int(lo):int(hi) + 1])
with pytest.raises(httpx.RemoteProtocolError):
_download(handler, dest, expected_size=len(PAYLOAD), num_connections=2)
+94
View File
@@ -0,0 +1,94 @@
"""A full disk mid-install must fail fast with an actionable message (not retry
five times, not blame the network)."""
import errno
import pytest
@pytest.mark.parametrize(
"reason",
[
OSError(errno.ENOSPC, "No space left on device"),
OSError("[Errno 28] No space left on device: '/x/blobs/a.incomplete'"),
"[WinError 112] There is not enough space on the disk",
"OSError: [Errno 122] Disk quota exceeded",
"OSError: [Errno 69] Disc quota exceeded",
],
)
def test_disk_full_is_recognised(reason):
from core.failure import is_disk_full_error
assert is_disk_full_error(reason)
def test_disk_full_is_found_through_a_wrapped_cause():
from core.failure import is_disk_full_error
try:
try:
raise OSError(errno.ENOSPC, "No space left on device")
except OSError as inner:
raise RuntimeError("download failed") from inner
except RuntimeError as wrapped:
assert is_disk_full_error(wrapped)
@pytest.mark.skipif(not hasattr(errno, "EDQUOT"), reason="EDQUOT is absent on Windows")
def test_platform_quota_errno_is_disk_full():
from core.failure import is_disk_full_error
assert is_disk_full_error(OSError(errno.EDQUOT, "quota"))
def test_other_failures_are_not_disk_full():
from core.failure import is_disk_full_error
assert not is_disk_full_error(ConnectionResetError("peer closed connection"))
assert not is_disk_full_error("Not enough disk space to install: needs 3 GB")
assert not is_disk_full_error(None)
def test_model_install_does_not_retry_a_full_disk():
from api.routers.setup.download import _is_retryable_download_error
assert not _is_retryable_download_error(OSError(errno.ENOSPC, "No space left on device"))
assert _is_retryable_download_error(ConnectionResetError(errno.ECONNRESET, "reset"))
def test_disk_full_message_names_free_space_and_cache(tmp_path):
from api.routers.setup.models import disk_full_message
text = disk_full_message(cache_dir=str(tmp_path))
assert str(tmp_path) in text and "Free up space" in text
def test_sidecar_install_reports_full_disk_from_uv_output():
from collections import deque
from services import sidecar_install as si
assert si._output_shows_disk_full(
deque(["Downloading torch", "error: failed to write: No space left on device (os error 28)"])
)
assert not si._output_shows_disk_full(deque(["connection reset by peer"]))
def test_earlier_recovered_disk_full_does_not_relabel_a_later_uv_failure(monkeypatch):
"""A clone that logged ENOSPC then recovered stays in the shared job log;
only the failing process's own output may decide the diagnosis."""
from collections import deque
from services import sidecar_install as si
job = {"engine_id": "x", "log": deque(["clone: No space left on device", "fallback clone ok"])}
class _Proc:
stdout = iter(["error: no matching distribution\n"])
def wait(self, timeout=None):
return 1
monkeypatch.setattr(si, "spawn_owned", lambda *a, **k: _Proc())
assert si._run_logged(job, ["uv"], timeout=5) == 1
assert list(job["_last_run_output"]) == ["error: no matching distribution\n"]
assert not si._output_shows_disk_full(job["_last_run_output"])
+53
View File
@@ -0,0 +1,53 @@
"""Titles become filenames; Windows rejects `<>:"|?*` with [Errno 22] (#2376 class)."""
import pathlib
import re
import pytest
BACKEND = pathlib.Path(__file__).resolve().parents[1] / "backend"
@pytest.mark.parametrize(
("raw", "expected"),
[
("How to X: a guide?_en.mp4", "How to X_ a guide__en.mp4"),
('a<b>c|d"e*f.srt', "a_b_c_d_e_f.srt"),
("clip. .", "clip"),
("CON.wav", "_CON.wav"),
("lpt1", "_lpt1"),
("???", "file"),
("", "file"),
("tab\there\x00.wav", "tab_here_.wav"),
],
)
def test_portable_filename_repairs_windows_hostile_names(raw, expected):
from core.path_security import portable_filename
assert portable_filename(raw) == expected
def test_reserved_device_name_exposed_by_truncation_is_still_repaired():
from core.path_security import portable_filename
name = portable_filename("CON" + " " * 197 + "x.wav")
assert name.split(".")[0].upper() != "CON" and name.endswith(".wav")
assert len(name.encode("utf-8")) <= 200
def test_portable_filename_truncates_by_bytes_and_keeps_extension():
from core.path_security import portable_filename
name = portable_filename("日本語" * 60 + ".mp4")
assert len(name.encode("utf-8")) <= 200
assert name.endswith(".mp4") and "�" not in name
def test_download_names_derived_from_titles_are_never_raw():
"""A job's filename is the video title; every route that turns it into a
download name must go through portable_filename (or an alnum allowlist)."""
offenders = []
for rel in ("api/routers/dub_export.py", "api/routers/batch.py"):
text = (BACKEND / rel).read_text(encoding="utf-8")
for match in re.finditer(r'^\s*(?:dl_)?filename\s*=\s*f"[^"\n]*\{[^}]*(?:base_name|filename)[^}]*\}[^\n]*$|^\s*dl_name\s*=\s*f"[^\n]*\{base_name\}[^\n]*$', text, re.M):
offenders.append(f"{rel}: {match.group(0).strip()}")
assert not offenders, offenders
+163
View File
@@ -0,0 +1,163 @@
import os
import pytest
def test_partial_managed_engines_remain_in_data_footprint(tmp_path):
from services.storage_report import build_report
data = tmp_path / "data"
engines = data / "engines"
installed = engines / "installed"
(installed / ".venv").mkdir(parents=True)
(installed / "weights.bin").write_bytes(b"i" * 70)
partial = engines / "interrupted" / "checkpoints"
partial.mkdir(parents=True)
(partial / "weights.bin").write_bytes(b"p" * 110)
(engines / "install.log").write_bytes(b"log")
report = build_report(data_dir=str(data), engines_dir=str(engines),
hf_cache_dir=str(tmp_path / "hf"), temp_root=str(tmp_path / "tmp"))
categories = {c["id"]: c for c in report["categories"]}
assert categories["engine_venvs"]["bytes"] == 70
other = next(c for c in categories["data"]["children"] if c["id"] == "other")
assert other["bytes"] == 113
assert sum(c["bytes"] for c in categories.values()) == 183
def test_unreadable_managed_engines_are_reported(tmp_path, monkeypatch):
from services import storage_report
data = tmp_path / "data"
engines = data / "engines"
engines.mkdir(parents=True)
real_scandir = storage_report.os.scandir
def scandir(path):
if str(path) == str(engines):
raise PermissionError("unreadable engines")
return real_scandir(path)
monkeypatch.setattr(storage_report.os, "scandir", scandir)
report = storage_report.build_report(data_dir=str(data), engines_dir=str(engines),
hf_cache_dir=str(tmp_path / "hf"), temp_root=str(tmp_path / "tmp"))
warned = {w["category_id"] for w in report["warnings"] if w["kind"] == "unreadable"}
assert {"data", "engine_venvs"} <= warned
assert all(not c["complete"] for c in report["categories"] if c["id"] in warned)
@pytest.mark.parametrize("failing_call", [1, 2])
def test_engine_entry_inspection_errors_do_not_abort_report(tmp_path, monkeypatch, failing_call):
from contextlib import contextmanager
from services import storage_report
data = tmp_path / "data"
engines = data / "engines"
partial = engines / "partial"
partial.mkdir(parents=True)
(partial / "weights.bin").write_bytes(b"saved")
real_scandir = storage_report.os.scandir
class Entry:
def __init__(self, entry):
self.entry = entry
self.calls = 0
self.path, self.name = entry.path, entry.name
def is_dir(self, **kwargs):
self.calls += 1
if self.calls == failing_call:
raise PermissionError("entry inaccessible")
return self.entry.is_dir(**kwargs)
def stat(self, **kwargs):
return self.entry.stat(**kwargs)
@contextmanager
def engine_scan():
with real_scandir(engines) as entries:
yield iter([Entry(e) for e in entries])
def scandir(path):
return engine_scan() if str(path) == str(engines) else real_scandir(path)
monkeypatch.setattr(storage_report.os, "scandir", scandir)
report = storage_report.build_report(data_dir=str(data), engines_dir=str(engines),
hf_cache_dir=str(tmp_path / "hf"), temp_root=str(tmp_path / "tmp"))
assert any(w["kind"] == "unreadable" and w["path"] == str(partial) for w in report["warnings"])
@pytest.mark.skipif(os.name != "posix", reason="native POSIX permission reproduction")
def test_native_unreadable_engines_emit_partial_report(tmp_path):
from services import storage_report
data = tmp_path / "data"
engines = data / "engines"
engines.mkdir(parents=True)
(engines / "partial.bin").write_bytes(b"unseen")
engines.chmod(0)
try:
try:
with os.scandir(engines):
pass
except PermissionError:
pass
else:
pytest.skip("current user bypasses directory permissions")
report = storage_report.build_report(data_dir=str(data), engines_dir=str(engines),
hf_cache_dir=str(tmp_path / "hf"), temp_root=str(tmp_path / "tmp"))
warnings = [w for w in report["warnings"] if w["kind"] == "unreadable"]
assert {w["category_id"] for w in warnings} == {"data", "engine_venvs"}
assert all(not c["complete"] for c in report["categories"] if c["id"] in {"data", "engine_venvs"})
finally:
engines.chmod(0o700)
def test_transient_installed_entry_error_preserves_category_ownership(tmp_path, monkeypatch):
from contextlib import contextmanager
from services import storage_report
data = tmp_path / "data"
engines = data / "engines"
installed = engines / "installed"
(installed / ".venv").mkdir(parents=True)
(installed / "weights.bin").write_bytes(b"installed weights")
real_scandir = storage_report.os.scandir
class Entry:
def __init__(self, entry):
self.entry = entry
self.path, self.name = entry.path, entry.name
self.calls = 0
def is_dir(self, **kwargs):
self.calls += 1
if self.calls == 1:
raise PermissionError("transient entry inspection failure")
return self.entry.is_dir(**kwargs)
def stat(self, **kwargs):
return self.entry.stat(**kwargs)
@contextmanager
def scan_engines():
with real_scandir(engines) as entries:
yield iter([Entry(e) for e in entries])
def scandir(path):
return scan_engines() if str(path) == str(engines) else real_scandir(path)
monkeypatch.setattr(storage_report.os, "scandir", scandir)
report = storage_report.build_report(data_dir=str(data), engines_dir=str(engines),
hf_cache_dir=str(tmp_path / "hf"), temp_root=str(tmp_path / "tmp"))
categories = {c["id"]: c for c in report["categories"]}
assert categories["engine_venvs"]["bytes"] == len(b"installed weights")
other = next(c for c in categories["data"]["children"] if c["id"] == "other")
assert other["bytes"] == 0
assert any(w["kind"] == "unreadable" and w["path"] == str(installed) for w in report["warnings"])
def test_unreadable_venv_is_not_treated_as_missing(tmp_path, monkeypatch):
from services import storage_report
data = tmp_path / "data"
engines = data / "engines"
installed = engines / "installed"
(installed / ".venv").mkdir(parents=True)
(installed / "model.bin").write_bytes(b"installed bytes")
venv = str(installed / ".venv")
real_stat = storage_report.os.stat
def stat(path, *args, **kwargs):
if str(path) == venv:
raise PermissionError("traversal denied")
return real_stat(path, *args, **kwargs)
monkeypatch.setattr(storage_report.os, "stat", stat)
report = storage_report.build_report(data_dir=str(data), engines_dir=str(engines),
hf_cache_dir=str(tmp_path / "hf"), temp_root=str(tmp_path / "tmp"))
categories = {c["id"]: c for c in report["categories"]}
assert not categories["engine_venvs"]["complete"]
other = next(c for c in categories["data"]["children"] if c["id"] == "other")
assert other["bytes"] == 0 # not silently reclassified as application data
assert any(w["kind"] == "unreadable" and w["path"] == str(installed) for w in report["warnings"])