release: 0.63.2 (r12-s5-backend-residue)
This commit is contained in:
@@ -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])
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
+10
-10
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+20
-3
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user