fix(channels): address R14 deep-review findings
- Make the immediate-backfill guard SYMMETRIC: run_deep_backfill now also registers the channel it pages in _backfill_in_progress, so it and backfill_channel_now mutually exclude per-channel and never advance the same cursor concurrently (whichever started first). The interactive cap moves to a separate _immediate_running counter so the scheduler's channels don't false-trip it. - Attribute the immediate backfill's spend: wrap it in quota.attribute(user_id, VIDEOS_BACKFILL_FULL) (route passes user.id), so it shows under the user's full-backfill action, not the OTHER/system bucket. - Tests: the update_channel->background-task wiring + the demo guard (scheduled for a real user, NOT for demo); distinct N/S/L counts (2 short + 1 live) so a short/live swap fails; extracted channelDetailPollMs with a pure unit test of the poll-or-stop predicate. - a11y: role="img" on the fully-synced charm so screen readers announce it.
This commit is contained in:
@@ -477,7 +477,7 @@ def update_channel(
|
||||
# request returns right away; the channel page's live poll shows the count climb. The scheduler
|
||||
# still finishes a very large channel's tail on its normal runs (deep_requested stays set).
|
||||
if deep_turned_on:
|
||||
background.add_task(backfill_channel_now, channel_id)
|
||||
background.add_task(backfill_channel_now, channel_id, user.id)
|
||||
|
||||
return {
|
||||
"id": channel_id,
|
||||
|
||||
+39
-23
@@ -231,8 +231,12 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -
|
||||
for channel in channels:
|
||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||
break
|
||||
if channel.id in _immediate_backfill:
|
||||
continue # an interactive backfill_channel_now is paging this one — no cursor race
|
||||
# Claim the channel (or skip it if an interactive backfill already owns it) so the two
|
||||
# never advance the same cursor concurrently — symmetric with backfill_channel_now.
|
||||
with _backfill_lock:
|
||||
if channel.id in _backfill_in_progress:
|
||||
continue
|
||||
_backfill_in_progress.add(channel.id)
|
||||
try:
|
||||
videos_added += backfill_channel_deep(db, yt, channel, max_pages)
|
||||
processed += 1
|
||||
@@ -240,6 +244,9 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -
|
||||
except Exception:
|
||||
db.rollback()
|
||||
log.exception("Deep backfill failed for channel %s", channel.id)
|
||||
finally:
|
||||
with _backfill_lock:
|
||||
_backfill_in_progress.discard(channel.id)
|
||||
reconciled = reconcile_full_history(db)
|
||||
return {
|
||||
"channels_processed": processed,
|
||||
@@ -248,27 +255,32 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -
|
||||
}
|
||||
|
||||
|
||||
# Channels an interactive "get full history" is backfilling RIGHT NOW (backfill_channel_now runs in a
|
||||
# FastAPI BackgroundTasks threadpool thread; the scheduler runs in its own thread — hence the lock).
|
||||
# The scheduler's deep pass skips these so the two never page the same channel's cursor concurrently,
|
||||
# and the size cap bounds how much immediate backfill one burst can kick, letting the rest fall back
|
||||
# to the scheduler's fair, quota-paced runs.
|
||||
_immediate_backfill: set[str] = set()
|
||||
_immediate_lock = threading.Lock()
|
||||
# Channels being deep-backfilled RIGHT NOW by EITHER path — the interactive backfill_channel_now (a
|
||||
# FastAPI BackgroundTasks threadpool thread) OR the scheduler's run_deep_backfill (its own thread).
|
||||
# BOTH register the channel they're paging (under the lock) so the other skips it: no two runs page
|
||||
# the same channel's `backfill_cursor` concurrently, whichever started first. `_immediate_running`
|
||||
# counts only the interactive path, capped so one user's burst can't drain the shared budget (the
|
||||
# scheduler is already quota-paced). Single uvicorn process → one shared module state, lock suffices.
|
||||
_backfill_in_progress: set[str] = set()
|
||||
_immediate_running = 0
|
||||
_backfill_lock = threading.Lock()
|
||||
_MAX_CONCURRENT_IMMEDIATE = 2
|
||||
|
||||
|
||||
def backfill_channel_now(channel_id: str, max_pages: int = 60) -> None:
|
||||
def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 60) -> None:
|
||||
"""User-initiated "get full history": deep-backfill ONE channel to completion right now, in a
|
||||
background task with its OWN session, so the full catalogue lands in seconds instead of over the
|
||||
paced scheduler's 10-minute ticks (max_pages=5/channel/tick). Quota-guarded, and capped to
|
||||
_MAX_CONCURRENT_IMMEDIATE in flight so one user can't drain the shared backfill budget by opting
|
||||
many channels in at once — the overflow (and any very large channel's tail) is finished by the
|
||||
scheduler on its normal runs, since deep_requested stays set. Best-effort — logs and swallows."""
|
||||
with _immediate_lock:
|
||||
if channel_id in _immediate_backfill or len(_immediate_backfill) >= _MAX_CONCURRENT_IMMEDIATE:
|
||||
return # already running, or too many in flight — leave it to the scheduler
|
||||
_immediate_backfill.add(channel_id)
|
||||
paced scheduler's 10-minute ticks (max_pages=5/channel/tick). Quota-guarded, capped to
|
||||
_MAX_CONCURRENT_IMMEDIATE interactive runs so one user can't drain the shared backfill budget by
|
||||
opting many channels in at once (the overflow, and any very large channel's tail, is finished by
|
||||
the scheduler since deep_requested stays set), and mutually excluded per-channel with the scheduler
|
||||
via _backfill_in_progress. Best-effort — logs and swallows failures."""
|
||||
global _immediate_running
|
||||
with _backfill_lock:
|
||||
if channel_id in _backfill_in_progress or _immediate_running >= _MAX_CONCURRENT_IMMEDIATE:
|
||||
return # already backfilling this channel, or too many interactive runs — leave it to the scheduler
|
||||
_backfill_in_progress.add(channel_id)
|
||||
_immediate_running += 1
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
user = get_service_user(db)
|
||||
@@ -280,15 +292,19 @@ def backfill_channel_now(channel_id: str, max_pages: int = 60) -> None:
|
||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||
return
|
||||
try:
|
||||
with YouTubeClient(db, user) as yt:
|
||||
backfill_channel_deep(db, yt, channel, max_pages)
|
||||
reconcile_full_history(db)
|
||||
# Attribute the spend to the requesting user under the full-backfill action, so it
|
||||
# shows up correctly in the usage breakdown (not the default OTHER/system bucket).
|
||||
with quota.attribute(user_id, quota.QuotaAction.VIDEOS_BACKFILL_FULL):
|
||||
with YouTubeClient(db, user) as yt:
|
||||
backfill_channel_deep(db, yt, channel, max_pages)
|
||||
reconcile_full_history(db)
|
||||
except Exception:
|
||||
db.rollback()
|
||||
log.exception("Immediate backfill failed for channel %s", channel_id)
|
||||
finally:
|
||||
with _immediate_lock:
|
||||
_immediate_backfill.discard(channel_id)
|
||||
with _backfill_lock:
|
||||
_backfill_in_progress.discard(channel_id)
|
||||
_immediate_running -= 1
|
||||
|
||||
|
||||
def estimate_deep_backfill(db: Session, channels: list[Channel]) -> dict:
|
||||
|
||||
@@ -1,36 +1,37 @@
|
||||
"""R14 backend coverage: the channel-detail Normal/Shorts/Live counts (moved off the manager table),
|
||||
and the interactive "get full history" immediate-backfill guard that keeps one burst from racing the
|
||||
scheduler's cursor or draining the shared backfill budget."""
|
||||
the update_channel -> immediate-backfill wiring (incl. the demo guard), and the interactive
|
||||
"get full history" guard that keeps it from racing the scheduler's cursor or draining the budget."""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import app.routes.channels as channels_route
|
||||
import app.state as state
|
||||
from app.auth import current_user
|
||||
from app.db import get_db
|
||||
from app.main import app
|
||||
from app.models import Channel, User, Video
|
||||
from app.models import Channel, Subscription, User, Video
|
||||
from app.sync import runner
|
||||
|
||||
_PUB = datetime(2020, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def test_channel_detail_returns_normal_shorts_live_counts(db, monkeypatch):
|
||||
# published_at set so channel_detail skips the YouTube-enrich path (no quota / network in tests).
|
||||
ch = Channel(
|
||||
id="UC_nsl",
|
||||
title="N/S/L Channel",
|
||||
published_at=datetime(2020, 1, 1, tzinfo=timezone.utc),
|
||||
)
|
||||
ch = Channel(id="UC_nsl", title="N/S/L Channel", published_at=_PUB)
|
||||
db.add(ch)
|
||||
me = User(email="nsl@example.com")
|
||||
db.add(me)
|
||||
db.flush()
|
||||
# 2 normal, 1 short, 1 live → stored 4, normal = 4 − 1 (short) − 1 (live) = 2.
|
||||
# DISTINCT counts (2 shorts, 1 live) so a short<->live filter/field swap can't pass:
|
||||
# stored 5, short 2, live 1, normal = 5 - 2 - 1 = 2.
|
||||
db.add_all(
|
||||
[
|
||||
Video(id="v_norm1", channel_id=ch.id),
|
||||
Video(id="v_norm2", channel_id=ch.id),
|
||||
Video(id="v_short", channel_id=ch.id, is_short=True),
|
||||
Video(id="v_live", channel_id=ch.id, live_status="live"),
|
||||
Video(id="v_short1", channel_id=ch.id, is_short=True),
|
||||
Video(id="v_short2", channel_id=ch.id, is_short=True),
|
||||
Video(id="v_live1", channel_id=ch.id, live_status="live"),
|
||||
]
|
||||
)
|
||||
db.commit()
|
||||
@@ -46,48 +47,99 @@ def test_channel_detail_returns_normal_shorts_live_counts(db, monkeypatch):
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["stored_videos"] == 4
|
||||
assert body["count_short"] == 1
|
||||
assert body["stored_videos"] == 5
|
||||
assert body["count_short"] == 2
|
||||
assert body["count_live"] == 1
|
||||
assert body["count_normal"] == 2
|
||||
|
||||
|
||||
def _setup_channel_and_sub(db, *, email, is_demo=False):
|
||||
ch = Channel(
|
||||
id="UC_wire",
|
||||
title="Wire",
|
||||
published_at=_PUB,
|
||||
recent_synced_at=_PUB, # so update_channel skips the in-request recent backfill (no network)
|
||||
uploads_playlist_id="UU_wire",
|
||||
)
|
||||
db.add(ch)
|
||||
user = User(email=email, is_demo=is_demo)
|
||||
db.add(user)
|
||||
db.flush()
|
||||
db.add(Subscription(user_id=user.id, channel_id=ch.id, deep_requested=False))
|
||||
db.commit()
|
||||
return ch, user
|
||||
|
||||
|
||||
def _patch_deep_requested(db, monkeypatch, user):
|
||||
calls: list[tuple] = []
|
||||
# background.add_task runs after the response in TestClient; record what would be scheduled.
|
||||
monkeypatch.setattr(channels_route, "backfill_channel_now", lambda *a, **k: calls.append(a))
|
||||
monkeypatch.setattr(state, "_configured_cache", True)
|
||||
app.dependency_overrides[get_db] = lambda: db
|
||||
app.dependency_overrides[current_user] = lambda: user
|
||||
try:
|
||||
resp = TestClient(app).patch("/api/channels/UC_wire", json={"deep_requested": True})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
return resp, calls
|
||||
|
||||
|
||||
def test_deep_request_schedules_immediate_backfill_for_non_demo(db, monkeypatch):
|
||||
ch, me = _setup_channel_and_sub(db, email="wire@example.com")
|
||||
resp, calls = _patch_deep_requested(db, monkeypatch, me)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["deep_requested"] is True
|
||||
# scheduled exactly once, with (channel_id, user_id) — the args backfill_channel_now needs.
|
||||
assert calls == [(ch.id, me.id)]
|
||||
|
||||
|
||||
def test_deep_request_does_not_schedule_immediate_backfill_for_demo(db, monkeypatch):
|
||||
# The SHARED demo account must not be able to kick immediate full backfills (budget drain).
|
||||
_ch, demo = _setup_channel_and_sub(db, email="demo@example.com", is_demo=True)
|
||||
resp, calls = _patch_deep_requested(db, monkeypatch, demo)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert calls == [], "demo must not schedule an immediate backfill"
|
||||
# …and deep_requested is not persisted on the shared demo subscription either.
|
||||
assert resp.json()["deep_requested"] is False
|
||||
|
||||
|
||||
class TestImmediateBackfillGuard:
|
||||
"""backfill_channel_now coordinates with the paced scheduler through a small in-flight set: a
|
||||
second call for a channel already in flight is a no-op, the set is capped so a burst can't drain
|
||||
the shared budget, and run_deep_backfill skips a channel an immediate backfill is handling."""
|
||||
"""backfill_channel_now coordinates with the paced scheduler through _backfill_in_progress (per-
|
||||
channel mutual exclusion, both directions) + _immediate_running (a cap on interactive runs so a
|
||||
burst can't drain the shared budget)."""
|
||||
|
||||
def setup_method(self):
|
||||
runner._immediate_backfill.clear()
|
||||
runner._backfill_in_progress.clear()
|
||||
runner._immediate_running = 0
|
||||
|
||||
def teardown_method(self):
|
||||
runner._immediate_backfill.clear()
|
||||
runner._backfill_in_progress.clear()
|
||||
runner._immediate_running = 0
|
||||
|
||||
@staticmethod
|
||||
def _boom_session():
|
||||
# SessionLocal is only reached AFTER the guard admits the channel; if the guard short-circuits
|
||||
# (as it must here) this is never called, so raising proves the short-circuit.
|
||||
# Reached only AFTER the guard admits the channel; if the guard short-circuits (as it must
|
||||
# in these tests) this never runs, so raising proves the short-circuit.
|
||||
raise AssertionError("guard must short-circuit before opening a session")
|
||||
|
||||
def test_a_second_call_for_an_in_flight_channel_is_a_noop(self, monkeypatch):
|
||||
runner._immediate_backfill.add("UC_busy")
|
||||
runner._backfill_in_progress.add("UC_busy")
|
||||
monkeypatch.setattr(runner, "SessionLocal", self._boom_session)
|
||||
runner.backfill_channel_now("UC_busy") # returns immediately; _boom_session never runs
|
||||
assert runner._immediate_backfill == {"UC_busy"} # the in-flight entry is untouched
|
||||
runner.backfill_channel_now("UC_busy", 1) # returns immediately; _boom_session never runs
|
||||
assert runner._backfill_in_progress == {"UC_busy"} # the in-flight entry is untouched
|
||||
|
||||
def test_over_the_concurrency_cap_falls_back_to_the_scheduler(self, monkeypatch):
|
||||
for i in range(runner._MAX_CONCURRENT_IMMEDIATE):
|
||||
runner._immediate_backfill.add(f"UC_full_{i}")
|
||||
runner._immediate_running = runner._MAX_CONCURRENT_IMMEDIATE
|
||||
monkeypatch.setattr(runner, "SessionLocal", self._boom_session)
|
||||
runner.backfill_channel_now("UC_new") # over the cap → no-op, scheduler will handle it
|
||||
assert "UC_new" not in runner._immediate_backfill
|
||||
runner.backfill_channel_now("UC_new", 1) # over the cap → no-op, scheduler will handle it
|
||||
assert "UC_new" not in runner._backfill_in_progress
|
||||
|
||||
def test_scheduler_deep_pass_skips_an_in_flight_channel(self, db, monkeypatch):
|
||||
# A channel the scheduler's query WOULD pick (deep-requested, recent-synced, not done)…
|
||||
ch = Channel(
|
||||
id="UC_inflight",
|
||||
title="In-flight",
|
||||
recent_synced_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
recent_synced_at=_PUB,
|
||||
uploads_playlist_id="UU_inflight",
|
||||
backfill_done=False,
|
||||
)
|
||||
@@ -95,8 +147,6 @@ class TestImmediateBackfillGuard:
|
||||
me = User(email="svc@example.com")
|
||||
db.add(me)
|
||||
db.flush()
|
||||
from app.models import Subscription
|
||||
|
||||
db.add(Subscription(user_id=me.id, channel_id=ch.id, deep_requested=True))
|
||||
db.commit()
|
||||
|
||||
@@ -109,13 +159,13 @@ class TestImmediateBackfillGuard:
|
||||
)
|
||||
monkeypatch.setattr(runner, "reconcile_full_history", lambda _db: 0)
|
||||
|
||||
# …is skipped by run_deep_backfill while an immediate backfill owns it (no cursor race).
|
||||
runner._immediate_backfill.add(ch.id)
|
||||
# …is skipped by run_deep_backfill while an interactive backfill owns it (no cursor race).
|
||||
runner._backfill_in_progress.add(ch.id)
|
||||
runner.run_deep_backfill(db)
|
||||
assert paged == [], "the scheduler must not page a channel an immediate backfill is handling"
|
||||
|
||||
# …and IS paged once the immediate backfill releases it.
|
||||
runner._immediate_backfill.discard(ch.id)
|
||||
# …and IS paged once the interactive backfill releases it.
|
||||
runner._backfill_in_progress.discard(ch.id)
|
||||
runner.run_deep_backfill(db)
|
||||
assert paged == [ch.id]
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ import { useCallback, useEffect, useRef, useState } from "react";
|
||||
import { useTranslation } from "react-i18next";
|
||||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import { useLiveQuery } from "../lib/useLiveQuery";
|
||||
import { channelDetailPollMs } from "../lib/channelDetailPoll";
|
||||
import { qk } from "../lib/queryKeys";
|
||||
import { useNavigationActions } from "./NavigationProvider";
|
||||
import { PriorityStepper } from "./PriorityStepper";
|
||||
@@ -85,7 +86,7 @@ export default function ChannelPage({
|
||||
// scheduler tops up any tail on its ~10-minute ticks — then stop polling once the back-catalogue is
|
||||
// in. Same queryKey as before, so mutation invalidations still hit.
|
||||
const detail = useLiveQuery(qk.channel(channelId), () => api.channelDetail(channelId), {
|
||||
intervalMs: (d) => (d?.deep_requested && !d.backfill_done ? 4000 : false),
|
||||
intervalMs: channelDetailPollMs,
|
||||
});
|
||||
const ch = detail.data;
|
||||
useEffect(() => setBannerAttempt(0), [ch?.banner_url]); // fresh channel/banner → try again
|
||||
|
||||
@@ -52,6 +52,7 @@ function StoredCell({ c }: { c: ManagedChannel }) {
|
||||
{isFullySynced(c) && (
|
||||
<Tooltip hint={t("channels.row.fullySyncedHint")}>
|
||||
<Check
|
||||
role="img"
|
||||
className="w-3.5 h-3.5 text-accent shrink-0 cursor-help"
|
||||
aria-label={t("channels.row.fullySynced")}
|
||||
/>
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { channelDetailPollMs } from "./channelDetailPoll";
|
||||
import { type ChannelDetail } from "./api";
|
||||
|
||||
// Only deep_requested + backfill_done drive the decision; cast a partial for the pure predicate.
|
||||
const d = (over: Partial<ChannelDetail>): ChannelDetail => over as ChannelDetail;
|
||||
|
||||
describe("channelDetailPollMs", () => {
|
||||
it("polls at 4s while THIS user's full-history backfill is running", () => {
|
||||
expect(channelDetailPollMs(d({ deep_requested: true, backfill_done: false }))).toBe(4000);
|
||||
});
|
||||
|
||||
it("stops (false) once the back-catalogue is fully in", () => {
|
||||
expect(channelDetailPollMs(d({ deep_requested: true, backfill_done: true }))).toBe(false);
|
||||
});
|
||||
|
||||
it("stops (false) when the user hasn't requested full history", () => {
|
||||
expect(channelDetailPollMs(d({ deep_requested: false, backfill_done: false }))).toBe(false);
|
||||
});
|
||||
|
||||
it("stops (false) before the detail has loaded", () => {
|
||||
expect(channelDetailPollMs(undefined)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,8 @@
|
||||
import { type ChannelDetail } from "./api";
|
||||
|
||||
/** How often the channel page re-polls its detail (via useLiveQuery): every 4s while THIS user's
|
||||
* full-history backfill is running (`deep_requested && !backfill_done`) so the stored count + the
|
||||
* N/S/L breakdown climb on their own, and `false` (stop polling) once the back-catalogue is in or
|
||||
* the user never requested it. Extracted so the poll-or-stop decision is unit-testable. */
|
||||
export const channelDetailPollMs = (d: ChannelDetail | undefined): number | false =>
|
||||
d?.deep_requested && !d.backfill_done ? 4000 : false;
|
||||
Reference in New Issue
Block a user