diff --git a/VERSION b/VERSION index 1f6db1c8..b0ac4d72 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.63.1 \ No newline at end of file +0.63.2 \ No newline at end of file diff --git a/backend/alembic/versions/0062_drop_redundant_idx.py b/backend/alembic/versions/0062_drop_redundant_idx.py new file mode 100644 index 00000000..198e980d --- /dev/null +++ b/backend/alembic/versions/0062_drop_redundant_idx.py @@ -0,0 +1,53 @@ +"""drop 10 redundant single-column FK indexes (R12 S5) + +Each of these columns already leads a composite UNIQUE constraint, whose btree covers leftmost-prefix +lookups on that column (and the ON DELETE CASCADE reverse lookups too) — so the extra single-column +`index=True` was pure write-amplification. Dropping them costs nothing on reads and saves an index +write per insert/update. Fast (index-only drop, no table rewrite). Reversible (downgrade recreates). + +Redundant index -> the composite that already covers it: + ix_subscriptions_user_id <- uq_user_channel (user_id, channel_id) + ix_video_states_user_id <- uq_user_video (user_id, video_id) + ix_search_finds_user_id <- uq_user_search_video (user_id, video_id) + ix_explored_channels_user_id <- uq_user_explored_channel (user_id, channel_id) + ix_blocked_channels_user_id <- uq_user_blocked_channel (user_id, channel_id) + ix_saved_views_user_id <- uq_user_saved_view (user_id, name) + ix_playlist_items_playlist_id <- uq_playlist_video (playlist_id, video_id) + ix_download_shares_job_id <- uq_download_share (job_id, to_user_id) + ix_plex_states_user_id <- uq_plex_user_item (user_id, item_id) + ix_plex_playlist_items_playlist_id <- uq_plex_playlist_item (playlist_id, item_id) +""" +from typing import Sequence, Union + +from alembic import op + +revision: str = "0062_drop_redundant_idx" +down_revision: Union[str, None] = "0061_dl_job_error_code" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +# (index_name, table_name, column) +_REDUNDANT: list[tuple[str, str, str]] = [ + ("ix_subscriptions_user_id", "subscriptions", "user_id"), + ("ix_video_states_user_id", "video_states", "user_id"), + ("ix_search_finds_user_id", "search_finds", "user_id"), + ("ix_explored_channels_user_id", "explored_channels", "user_id"), + ("ix_blocked_channels_user_id", "blocked_channels", "user_id"), + ("ix_saved_views_user_id", "saved_views", "user_id"), + ("ix_playlist_items_playlist_id", "playlist_items", "playlist_id"), + ("ix_download_shares_job_id", "download_shares", "job_id"), + ("ix_plex_states_user_id", "plex_states", "user_id"), + ("ix_plex_playlist_items_playlist_id", "plex_playlist_items", "playlist_id"), +] + + +def upgrade() -> None: + for idx, _table, _col in _REDUNDANT: + # IF EXISTS: a DB whose schema was stamped from a create_all (tests) won't have these once + # the models drop index=True, so a plain drop_index would error there. + op.execute(f"DROP INDEX IF EXISTS {idx}") + + +def downgrade() -> None: + for idx, table, col in _REDUNDANT: + op.create_index(idx, table, [col]) diff --git a/backend/app/downloads/quota.py b/backend/app/downloads/quota.py index b60fcb75..9a87f040 100644 --- a/backend/app/downloads/quota.py +++ b/backend/app/downloads/quota.py @@ -43,10 +43,16 @@ class QuotaExceeded(Exception): def resolve(db: Session, user_id: int) -> QuotaLimits: row = db.get(DownloadQuota, user_id) if row is not None: - return QuotaLimits(row.max_bytes, row.max_concurrent, row.max_jobs, row.unlimited) + # Floor max_concurrent at 1 (admin_set_quota already does on write; the sysconfig default's + # min is 1). A stored 0 is nonsensical for a concurrency limit AND would diverge the two + # "at limit" checks — the worker's claim-window exclusion vs at_concurrency_limit — since a + # 0-limit user is "capped" with zero running jobs, which the SQL exclusion can't see. Keep + # them consistent by never letting the effective limit drop below 1. (To BLOCK a user's + # downloads, suspend them or set max_jobs=0, not max_concurrent=0.) + return QuotaLimits(row.max_bytes, max(1, row.max_concurrent), row.max_jobs, row.unlimited) return QuotaLimits( max_bytes=sysconfig.get_int(db, "download_default_max_bytes"), - max_concurrent=sysconfig.get_int(db, "download_default_max_concurrent"), + max_concurrent=max(1, sysconfig.get_int(db, "download_default_max_concurrent")), max_jobs=sysconfig.get_int(db, "download_default_max_jobs"), unlimited=False, ) diff --git a/backend/app/models.py b/backend/app/models.py index d871b8ea..3bf35918 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -219,7 +219,7 @@ class Subscription(Base, TimestampMixin): id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column( - ForeignKey("users.id", ondelete="CASCADE"), index=True + ForeignKey("users.id", ondelete="CASCADE") ) channel_id: Mapped[str] = mapped_column( ForeignKey("channels.id", ondelete="CASCADE"), index=True @@ -388,7 +388,7 @@ class VideoState(Base, UpdatedAtMixin): id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column( - ForeignKey("users.id", ondelete="CASCADE"), index=True + ForeignKey("users.id", ondelete="CASCADE") ) video_id: Mapped[str] = mapped_column( ForeignKey("videos.id", ondelete="CASCADE"), index=True @@ -418,7 +418,7 @@ class SearchFind(Base, TimestampMixin): id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column( - ForeignKey("users.id", ondelete="CASCADE"), index=True + ForeignKey("users.id", ondelete="CASCADE") ) video_id: Mapped[str] = mapped_column( ForeignKey("videos.id", ondelete="CASCADE"), index=True @@ -443,7 +443,7 @@ class ExploredChannel(Base, TimestampMixin): id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column( - ForeignKey("users.id", ondelete="CASCADE"), index=True + ForeignKey("users.id", ondelete="CASCADE") ) channel_id: Mapped[str] = mapped_column( ForeignKey("channels.id", ondelete="CASCADE"), index=True @@ -462,7 +462,7 @@ class BlockedChannel(Base, TimestampMixin): id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column( - ForeignKey("users.id", ondelete="CASCADE"), index=True + ForeignKey("users.id", ondelete="CASCADE") ) channel_id: Mapped[str] = mapped_column( ForeignKey("channels.id", ondelete="CASCADE"), index=True @@ -619,7 +619,7 @@ class PlaylistItem(Base): id: Mapped[int] = mapped_column(primary_key=True) playlist_id: Mapped[int] = mapped_column( - ForeignKey("playlists.id", ondelete="CASCADE"), index=True + ForeignKey("playlists.id", ondelete="CASCADE") ) video_id: Mapped[str] = mapped_column( ForeignKey("videos.id", ondelete="CASCADE"), index=True @@ -644,7 +644,7 @@ class SavedView(Base, TimestampMixin, UpdatedAtMixin): id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column( - ForeignKey("users.id", ondelete="CASCADE"), index=True + ForeignKey("users.id", ondelete="CASCADE") ) name: Mapped[str] = mapped_column(String(80)) filters: Mapped[dict] = mapped_column(JSON) @@ -893,7 +893,7 @@ class DownloadShare(Base, TimestampMixin): id: Mapped[int] = mapped_column(primary_key=True) job_id: Mapped[int] = mapped_column( - ForeignKey("download_jobs.id", ondelete="CASCADE"), index=True + ForeignKey("download_jobs.id", ondelete="CASCADE") ) from_user_id: Mapped[int] = mapped_column( ForeignKey("users.id", ondelete="CASCADE"), index=True @@ -1191,7 +1191,7 @@ class PlexState(Base, UpdatedAtMixin): id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column( - ForeignKey("users.id", ondelete="CASCADE"), index=True + ForeignKey("users.id", ondelete="CASCADE") ) item_id: Mapped[int] = mapped_column( ForeignKey("plex_items.id", ondelete="CASCADE"), index=True @@ -1260,7 +1260,7 @@ class PlexPlaylistItem(Base): id: Mapped[int] = mapped_column(primary_key=True) playlist_id: Mapped[int] = mapped_column( - ForeignKey("plex_playlists.id", ondelete="CASCADE"), index=True + ForeignKey("plex_playlists.id", ondelete="CASCADE") ) item_id: Mapped[int] = mapped_column( ForeignKey("plex_items.id", ondelete="CASCADE"), index=True diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index aa10e5ce..e1fe1503 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -40,6 +40,11 @@ admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"] _VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$") +# Cap the per-user job listings so a heavy library can't return an unbounded result set. The +# Download Center renders newest-first; 500 is far beyond any real manual-download history (and +# matches the admin listing's own cap). +_MAX_LIST = 500 + # --- source + serialization ---------------------------------------------------------------- @@ -334,6 +339,7 @@ def list_downloads( select(DownloadJob) .where(DownloadJob.user_id == user.id) .order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()) + .limit(_MAX_LIST) ) .scalars() .all() @@ -379,7 +385,8 @@ def shared_with_me( select(DownloadJob) .join(DownloadShare, DownloadShare.job_id == DownloadJob.id) .where(DownloadShare.to_user_id == user.id) - .order_by(DownloadShare.created_at.desc()) + .order_by(DownloadShare.created_at.desc(), DownloadJob.id.desc()) + .limit(_MAX_LIST) ) .scalars() .all() @@ -881,7 +888,7 @@ def admin_list_downloads( ) -> list[dict]: jobs = ( db.execute( - select(DownloadJob).order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()).limit(500) + select(DownloadJob).order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()).limit(_MAX_LIST) ) .scalars() .all() diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 5591ee77..7e555f11 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -41,6 +41,8 @@ VALID_STATES = {"new", "watched", "hidden"} # Resume-position thresholds (mirror the client): positions below this are "didn't # really start" and within this of the end are "basically finished" — neither is worth # storing, so we clear the position instead (the video shows no progress bar). +# NB: the Plex player intentionally uses a LARGER finish margin (30s) because it marks +# long-form items watched, not just clears a resume bar — see routes/plex/playback.py. PROGRESS_MIN_SECONDS = 5 FINISH_MARGIN_SECONDS = 10 diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index cbb7d6e2..42495705 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -158,6 +158,11 @@ def subtitle( return Response(subtitles.shift_vtt(vtt, offset), media_type="text/vtt") +# Resume/watched thresholds. _PROGRESS_MIN_S matches the feed's PROGRESS_MIN_SECONDS (5), but +# _FINISH_MARGIN_S is DELIBERATELY larger than the feed's FINISH_MARGIN_SECONDS (30 vs 10): this is +# long-form content (movies/episodes with credits) and crossing it MARKS THE ITEM WATCHED, whereas +# the feed's margin only clears a short video's resume bar — different domains, intentional divergence, +# not drift. Keep them separate on purpose (see routes/feed.py). _PROGRESS_MIN_S = 5 _FINISH_MARGIN_S = 30 diff --git a/backend/app/worker.py b/backend/app/worker.py index 96c56956..0749bd42 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -177,19 +177,36 @@ def _recover_orphans() -> None: def _claim_job() -> int | None: """Claim the oldest queued job whose user is under their per-user concurrency limit. - Fetches a window of locked candidates (FOR UPDATE SKIP LOCKED) and picks the first eligible - one, so one user flooding the queue can't starve others beyond their max_concurrent.""" + Users already AT their concurrency limit are excluded IN THE SQL, so one user flooding the + queue can't monopolize the fetch window and starve everyone behind them. (The old fixed + LIMIT-50 window did exactly that: 50 queued jobs from a single capped user filled the window, + `chosen` came back None, and every other user's work — sitting past position 50 — was never + seen.) The per-user limit is the quota row's `max_concurrent` (else the sysconfig default); + `unlimited` users are never capped. A final Python `at_concurrency_limit` re-check still guards + the tiny race between the exclusion subquery and the claim (it can only ever reject, never + over-claim), so correctness never depends on the two observations agreeing.""" with SessionLocal() as db: + default_mc = sysconfig.get_int(db, "download_default_max_concurrent") rows = db.execute( text( """ SELECT id, user_id FROM download_jobs WHERE status='queued' + AND user_id NOT IN ( + SELECT r.user_id + FROM download_jobs r + LEFT JOIN download_quotas q ON q.user_id = r.user_id + WHERE r.status='running' AND r.user_id IS NOT NULL + GROUP BY r.user_id, q.max_concurrent, q.unlimited + HAVING COALESCE(q.unlimited, false) = false + AND count(*) >= GREATEST(COALESCE(q.max_concurrent, :dmc), 1) + ) ORDER BY queue_pos, id FOR UPDATE SKIP LOCKED LIMIT 50 """ - ) + ), + {"dmc": default_mc}, ).fetchall() chosen = next( (jid for jid, uid in rows if not quota.at_concurrency_limit(db, uid)), None diff --git a/backend/tests/test_download_limits.py b/backend/tests/test_download_limits.py new file mode 100644 index 00000000..1b1a1663 --- /dev/null +++ b/backend/tests/test_download_limits.py @@ -0,0 +1,133 @@ +"""DB-backed tests for R12 · S5 backend residue — the two download limits: + +- **#2 claim fairness** (`worker._claim_job`): users already AT their per-user concurrency limit + are excluded IN THE SQL, so one user flooding the queue past the fetch window can't starve + everyone behind them. The old fixed `LIMIT 50` window did exactly that. +- **#3 list cap** (`routes.downloads.list_downloads` / `shared_with_me`): the per-user listings + are bounded so a heavy library can't return an unbounded result set. + +Uses the `db` fixture (real Postgres) — skipped when no database is reachable. `_claim_job` runs in +its OWN session, so every test COMMITS its setup before calling it. +""" +from app.models import DownloadJob, DownloadQuota, User +from app.worker import _claim_job + + +def _job(db, user_id, queue_pos, status="queued"): + j = DownloadJob( + user_id=user_id, + source_kind="youtube", + source_ref="x" * 11, + status=status, + queue_pos=queue_pos, + profile_snapshot={}, + ) + db.add(j) + return j + + +def _quota(db, user_id, max_concurrent, unlimited=False): + db.add( + DownloadQuota( + user_id=user_id, + max_bytes=10**12, + max_concurrent=max_concurrent, + max_jobs=10**6, + unlimited=unlimited, + ) + ) + + +# --- #2 · claim fairness -------------------------------------------------------------------- + +def test_capped_user_flood_does_not_starve_users_behind_the_window(db): + a = User(email="flooder@example.com") + b = User(email="waiter@example.com") + db.add_all([a, b]) + db.flush() + _quota(db, a.id, max_concurrent=1) # A: 1 concurrent, already running one -> all A's queued are ineligible + _quota(db, b.id, max_concurrent=3) + _job(db, a.id, queue_pos=0, status="running") + for i in range(1, 61): # A floods the FRONT with far more than the 50-row fetch window + _job(db, a.id, queue_pos=i) + bjob = _job(db, b.id, queue_pos=100) # B sits behind the flood + db.commit() + + assert _claim_job() == bjob.id # B is served despite A's 60-job flood filling the old window + db.expire_all() + assert db.get(DownloadJob, bjob.id).status == "running" + + +def test_eligible_user_claims_oldest_by_queue_pos(db): + u = User(email="solo@example.com") + db.add(u) + db.flush() + _quota(db, u.id, max_concurrent=2) + first = _job(db, u.id, queue_pos=1) + _job(db, u.id, queue_pos=2) + db.commit() + assert _claim_job() == first.id + + +def test_all_candidates_capped_returns_none(db): + u = User(email="capped@example.com") + db.add(u) + db.flush() + _quota(db, u.id, max_concurrent=1) + _job(db, u.id, queue_pos=0, status="running") # at the cap + _job(db, u.id, queue_pos=1) # queued but ineligible + db.commit() + assert _claim_job() is None + + +def test_stored_zero_concurrency_is_floored_not_starving(db): + # A stray max_concurrent=0 is only reachable by a direct DB edit (admin_set_quota floors at 1), + # but it must not wedge the claim loop. The effective limit floors to 1 on BOTH the read path + # (resolve -> at_concurrency_limit) and in the claim SQL (GREATEST(..., 1)), so the two "at + # limit" checks agree: the user's own first job is claimable and its flood never monopolizes + # the window with permanently-ineligible jobs. (Before the floor, _claim_job returned None here + # — the guard rejected all 60 of A's jobs — and any user behind them starved.) + a = User(email="zero@example.com") + b = User(email="behind@example.com") + db.add_all([a, b]) + db.flush() + _quota(db, a.id, max_concurrent=0) # nonsensical stored value + _quota(db, b.id, max_concurrent=2) + for i in range(60): + _job(db, a.id, queue_pos=i) + _job(db, b.id, queue_pos=100) + db.commit() + + claimed = _claim_job() + assert claimed is not None # the loop is not wedged + db.expire_all() + assert db.get(DownloadJob, claimed).user_id == a.id # A's own first job runs (floored to 1) + + +def test_unlimited_user_is_never_excluded(db): + u = User(email="unlimited@example.com") + db.add(u) + db.flush() + _quota(db, u.id, max_concurrent=1, unlimited=True) + for i in range(3): # 3 running, well past max_concurrent — but unlimited bypasses the cap + _job(db, u.id, queue_pos=i, status="running") + q = _job(db, u.id, queue_pos=10) + db.commit() + assert _claim_job() == q.id + + +# --- #3 · list cap -------------------------------------------------------------------------- + +def test_list_downloads_is_capped(db, monkeypatch): + from app.routes import downloads as dl + + u = User(email="heavy@example.com") + db.add(u) + db.flush() + for i in range(3): + _job(db, u.id, queue_pos=i, status="done") + db.commit() + + monkeypatch.setattr(dl, "_MAX_LIST", 2) + rows = dl.list_downloads(user=u, db=db) + assert len(rows) == 2 # bounded by the cap, not the 3 rows on disk diff --git a/frontend/src/lib/releaseNotes.ts b/frontend/src/lib/releaseNotes.ts index c7bc720c..c91569c9 100644 --- a/frontend/src/lib/releaseNotes.ts +++ b/frontend/src/lib/releaseNotes.ts @@ -14,6 +14,14 @@ export interface ReleaseEntry { } export const RELEASE_NOTES: ReleaseEntry[] = [ + { + version: "0.63.2", + date: "2026-08-13", + chores: [ + "Fairer download queue: when one person queues a very large batch, the worker no longer lets it crowd out everyone else's downloads — each person's work is picked up as their own slots free up.", + "Internal database and API cleanup (dropped redundant indexes; bounded the download list queries). No change to how anything looks or works.", + ], + }, { version: "0.63.1", date: "2026-08-13",