fix(downloads): floor max_concurrent + stable share order (R12 S5 review)
- Floor the effective max_concurrent at 1 on the read path (quota.resolve) and in the claim SQL (GREATEST(..., 1)). admin_set_quota already floors on write and the sysconfig default's min is 1, but a stray stored 0 (direct DB edit) would diverge the two "at limit" checks — the SQL claim-window exclusion can't see a user "capped" with zero running jobs, while at_concurrency_limit can — reopening the starvation the fairness fix closes. Keeping the effective limit >= 1 makes the two checks provably agree. - shared_with_me now orders by (created_at DESC, id DESC) to match list_downloads, so the newly-added 500 cap has a stable boundary among same-timestamp shares instead of a non-deterministic one. - 0062 upgrade loop: rename the unused loop var "table" to "_table" (matching the underscore-marked "_col"). Test: a stored max_concurrent=0 flooder no longer wedges the claim loop.
This commit is contained in:
@@ -42,7 +42,7 @@ _REDUNDANT: list[tuple[str, str, str]] = [
|
|||||||
|
|
||||||
|
|
||||||
def upgrade() -> None:
|
def upgrade() -> None:
|
||||||
for idx, table, _col in _REDUNDANT:
|
for idx, _table, _col in _REDUNDANT:
|
||||||
# IF EXISTS: a DB whose schema was stamped from a create_all (tests) won't have these once
|
# 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.
|
# the models drop index=True, so a plain drop_index would error there.
|
||||||
op.execute(f"DROP INDEX IF EXISTS {idx}")
|
op.execute(f"DROP INDEX IF EXISTS {idx}")
|
||||||
|
|||||||
@@ -43,10 +43,16 @@ class QuotaExceeded(Exception):
|
|||||||
def resolve(db: Session, user_id: int) -> QuotaLimits:
|
def resolve(db: Session, user_id: int) -> QuotaLimits:
|
||||||
row = db.get(DownloadQuota, user_id)
|
row = db.get(DownloadQuota, user_id)
|
||||||
if row is not None:
|
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(
|
return QuotaLimits(
|
||||||
max_bytes=sysconfig.get_int(db, "download_default_max_bytes"),
|
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"),
|
max_jobs=sysconfig.get_int(db, "download_default_max_jobs"),
|
||||||
unlimited=False,
|
unlimited=False,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -385,7 +385,7 @@ def shared_with_me(
|
|||||||
select(DownloadJob)
|
select(DownloadJob)
|
||||||
.join(DownloadShare, DownloadShare.job_id == DownloadJob.id)
|
.join(DownloadShare, DownloadShare.job_id == DownloadJob.id)
|
||||||
.where(DownloadShare.to_user_id == user.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)
|
.limit(_MAX_LIST)
|
||||||
)
|
)
|
||||||
.scalars()
|
.scalars()
|
||||||
|
|||||||
@@ -199,7 +199,7 @@ def _claim_job() -> int | None:
|
|||||||
WHERE r.status='running' AND r.user_id IS NOT NULL
|
WHERE r.status='running' AND r.user_id IS NOT NULL
|
||||||
GROUP BY r.user_id, q.max_concurrent, q.unlimited
|
GROUP BY r.user_id, q.max_concurrent, q.unlimited
|
||||||
HAVING COALESCE(q.unlimited, false) = false
|
HAVING COALESCE(q.unlimited, false) = false
|
||||||
AND count(*) >= COALESCE(q.max_concurrent, :dmc)
|
AND count(*) >= GREATEST(COALESCE(q.max_concurrent, :dmc), 1)
|
||||||
)
|
)
|
||||||
ORDER BY queue_pos, id
|
ORDER BY queue_pos, id
|
||||||
FOR UPDATE SKIP LOCKED
|
FOR UPDATE SKIP LOCKED
|
||||||
|
|||||||
@@ -80,6 +80,30 @@ def test_all_candidates_capped_returns_none(db):
|
|||||||
assert _claim_job() is None
|
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):
|
def test_unlimited_user_is_never_excluded(db):
|
||||||
u = User(email="unlimited@example.com")
|
u = User(email="unlimited@example.com")
|
||||||
db.add(u)
|
db.add(u)
|
||||||
|
|||||||
Reference in New Issue
Block a user