fix(downloads): a download larger than the space left says so, not "processing failed"

yt-dlp refuses a file announced over `max_filesize` by logging one line and returning with no file
(measured, and pinned in test_formats against a local server): the worker found no output and
failed the job as `postprocess`. `formats.watching_size_cap` now sees that line through our yt-dlp
logger, and the download (and, should it ever happen, a YouTube clip's range fetch) fails as the
owner's refusal with DOWNLOAD_TOO_BIG_MESSAGE, classified `quota_exceeded`. The download's quota
copy says it does not fit in what is left — true at a full cap too — instead of "Storage full".
A pre-existing type error the change touched is fixed (baseline down by one).

backend/tests/test_clip.py

@@ -15,7 +15,7 @@ from app import worker as workermod
 from app.config import settings
 from app.downloads import clip as clipmod
 from app.downloads import errors as dlerrors
-from app.downloads import pipeline, quota, runner, service
+from app.downloads import formats, pipeline, quota, runner, service
 from app.downloads import probe as probemod
 from app.downloads.edit import INPUT_GUARD
 from app.models import Channel, DownloadJob, MediaAsset, PlexItem, PlexLibrary, PlexShow, Video
@@ -705,6 +705,30 @@ def test_a_clip_larger_than_what_its_owner_has_left_is_refused_and_not_placed(
     )
 
 
+def test_a_youtube_clip_range_yt_dlp_refuses_for_its_cap_says_it_is_too_big(
+    db, make_user, as_user, yt_video, worker, monkeypatch
+):
+    # Should yt-dlp's own `max_filesize` refuse the range, it returns with no file (test_formats):
+    # the owner's storage, not a processing fault, and nothing is cut.
+    def refused(*_a, max_bytes=None, **_k):
+        formats._YDL_LOG.debug(
+            f"[download] File is larger than max-filesize (9000 bytes > {max_bytes} bytes). Aborting."
+        )
+        return {"requested_downloads": [{}]}
+
+    monkeypatch.setattr(pipeline, "_extract_with_fallback", refused)
+    job = _clip(as_user(make_user()), source="youtube", ref=yt_video).json()
+    monkeypatch.setattr(quota, "remaining_bytes", lambda _db, _uid: 500)
+    job, asset = worker["run"](job)
+    assert (job.status, job.error_code, job.error, asset.status) == (
+        "error",
+        "quota_exceeded",
+        dlerrors.CLIP_TOO_BIG_MESSAGE,
+        "pending",
+    )
+    assert "cmd" not in worker
+
+
 def test_a_plex_clip_of_a_silent_file_is_cut_silent(db, make_user, as_user, plex_episode, worker):
     # The player names the track it plays 0 even when the file has none.
     worker["probe"] = {"streams": [{"codec_type": "video"}]}

backend/tests/test_download_errors.py

@@ -65,6 +65,7 @@ def test_our_own_worker_refusals_classify_as_themselves():
     assert classify(dlerrors.QUOTA_FULL_MESSAGE) == "quota_exceeded"
     assert classify(dlerrors.CLIP_TOO_BIG_MESSAGE) == "quota_exceeded"
     assert classify(dlerrors.CONVERTED_TOO_BIG_MESSAGE) == "quota_exceeded"
+    assert classify(dlerrors.DOWNLOAD_TOO_BIG_MESSAGE) == "quota_exceeded"
 
 
 def test_the_redacted_network_text_classifies_back_to_network():

backend/tests/test_download_limits.py

@@ -11,7 +11,8 @@ its OWN session, so every test COMMITS its setup before calling it.
 """
 
 from app.config import settings
-from app.downloads import pipeline, quota
+from app.downloads import errors as dlerrors
+from app.downloads import formats, pipeline, quota
 from app.models import DownloadJob, DownloadQuota, MediaAsset, User
 from app.worker import _claim_job
 
@@ -297,3 +298,49 @@ def test_a_refusal_for_one_user_does_not_poison_the_shared_asset(db, monkeypatch
     assert db.get(MediaAsset, shared.id).status == "pending"
     assert db.get(MediaAsset, shared.id).error is None
     assert db.get(DownloadJob, theirs.id).status == "queued"
+
+
+def test_a_download_over_the_space_left_says_so_and_releases_the_shared_asset(
+    db, monkeypatch, tmp_path
+):
+    # yt-dlp refuses a file over `max_filesize` by saying so and returning with no file (pinned in
+    # test_formats). That is the owner's storage, not a processing fault: their refusal, in words
+    # true of a cap with space still left, and the shared asset goes back for the next job.
+    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    u = User(email="nearcap@example.com")
+    db.add(u)
+    db.flush()
+    db.add(
+        DownloadQuota(user_id=u.id, max_bytes=1000, max_concurrent=2, max_jobs=100, unlimited=False)
+    )
+    held = _sized_asset(db, 400, "near-held")
+    _job(db, u.id, queue_pos=0, status="done", asset_id=held.id)
+    shared = MediaAsset(
+        source_kind="youtube",
+        source_ref="y" * 11,
+        format_sig="s",
+        status="downloading",
+        ref_count=1,
+    )
+    db.add(shared)
+    db.flush()
+    mine = _job(db, u.id, queue_pos=1, status="running", asset_id=shared.id)
+    db.commit()
+    caps = []
+
+    def refuses_over_the_cap(*args, max_bytes=None, **kwargs):
+        caps.append(max_bytes)
+        formats._YDL_LOG.debug(
+            f"[download] File is larger than max-filesize (2000 bytes > {max_bytes} bytes). Aborting."
+        )
+        return {"id": "y" * 11, "requested_downloads": [{}]}
+
+    monkeypatch.setattr(pipeline, "_extract_with_fallback", refuses_over_the_cap)
+    pipeline._download(mine.id, shared.id, "youtube", "y" * 11, {}, u.id)
+
+    db.expire_all()
+    job = db.get(DownloadJob, mine.id)
+    assert caps == [600]
+    assert (job.status, job.error_code) == ("error", "quota_exceeded")
+    assert job.error == dlerrors.DOWNLOAD_TOO_BIG_MESSAGE
+    assert db.get(MediaAsset, shared.id).status == "pending"

backend/tests/test_formats.py

@@ -3,6 +3,13 @@ playability guarantee) and must SELECT H.264 (not merely prefer it), so shared d
 every browser incl. iOS/Safari. Non-mp4 containers and non-H.264 vcodecs are coerced, not honoured.
 See the iPad AV1 broken-player fix + the format narrowing."""
 
+import contextlib
+import functools
+import http.server
+import threading
+
+import yt_dlp
+
 from app.downloads import edit as editmod
 from app.downloads import formats
 
@@ -162,3 +169,40 @@ def test_the_mp3_fix_moved_the_audio_cache_identity(monkeypatch):
     current = formats.format_sig(audio)
     monkeypatch.setattr(formats, "_AUDIO_SIG_VERSION", 4)
     assert formats.format_sig(audio) != current
+
+
+class _QuietFiles(http.server.SimpleHTTPRequestHandler):
+    def log_message(self, *args):  # the test's output is yt-dlp's, not the server's
+        pass
+
+    def handle(self):
+        # yt-dlp hangs up on a file over the cap as soon as it reads its size.
+        with contextlib.suppress(ConnectionError):
+            super().handle()
+
+
+def test_yt_dlp_refuses_a_file_over_the_cap_without_raising_and_the_watch_sees_it(tmp_path):
+    # The boundary `watching_size_cap` reads, held against the yt-dlp this image pins: over
+    # `max_filesize` its HTTP downloader logs one line and RETURNS — no exception, no file — so
+    # without the watch the pipeline would only find "no output file" and call it a processing fault.
+    served = tmp_path / "served"
+    served.mkdir()
+    (served / "v.mp4").write_bytes(b"\0" * 200_000)
+    srv = http.server.ThreadingHTTPServer(
+        ("127.0.0.1", 0), functools.partial(_QuietFiles, directory=str(served))
+    )
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    url = f"http://127.0.0.1:{srv.server_port}/v.mp4"
+    try:
+        for cap, refused in ((1_000, True), (10_000_000, False)):
+            out = tmp_path / f"out-{cap}"
+            opts = formats.build_ydl_opts(
+                {}, str(out / "%(id)s.%(ext)s"), lambda d: None, max_bytes=cap
+            )
+            with formats.watching_size_cap() as over, yt_dlp.YoutubeDL(opts) as ydl:
+                ydl.extract_info(url, download=True)
+            assert over[0] is refused, cap
+            files = [p for p in out.glob("*") if p.suffix == ".mp4"] if out.exists() else []
+            assert bool(files) is not refused, cap
+    finally:
+        srv.shutdown()