diff --git a/backend/app/sync/maintenance.py b/backend/app/sync/maintenance.py index 2548d7dd..caebde45 100644 --- a/backend/app/sync/maintenance.py +++ b/backend/app/sync/maintenance.py @@ -20,7 +20,7 @@ Quota: phase 1 costs ~(flagged / 50) units, phase 2 ~(batch / 50). Both respect backfill reserve as the other jobs so interactive syncs never starve. """ -from datetime import timedelta +from datetime import datetime, timedelta from sqlalchemy import or_, select from sqlalchemy.orm import Session @@ -129,14 +129,14 @@ def _notify_removed(db: Session, impact: dict[int, list[dict]]) -> int: return len(impact) -def _recheck_flagged(db: Session, yt: YouTubeClient) -> dict: - """Phase 1: recover or hard-delete videos already flagged unavailable.""" +def _recheck_flagged(db: Session, yt: YouTubeClient, now: datetime) -> dict: + """Phase 1: recover or hard-delete videos already flagged unavailable. `now` is the run's: + phase 2 skips a video stamped with it, so one recovered here is not fetched again.""" flagged = db.execute(select(Video).where(Video.unavailable_since.is_not(None))).scalars().all() if not flagged: return {"checked": 0, "recovered": 0, "deleted": 0, "notified": 0} progress.report(0, len(flagged), "recheck") items = {it["id"]: it for it in yt.get_videos([v.id for v in flagged])} - now = _now() recovered = 0 to_delete: list[Video] = [] for video in flagged: @@ -173,7 +173,7 @@ def _recheck_flagged(db: Session, yt: YouTubeClient) -> dict: } -def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict: +def _revalidate_rolling(db: Session, yt: YouTubeClient, now: datetime) -> dict: """Phase 2: re-check the least-recently-checked currently-available videos and flag any that have become unplayable (hidden immediately; deleted later if they don't recover).""" batch = sysconfig.get_int( @@ -181,7 +181,6 @@ def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict: ) # admin override, else config default flagged_new = 0 checked = 0 - now = _now() # Process in videos.list-sized pages so we commit incrementally and stop on low quota. while checked < batch: if quota.backfill_reserve_reached(): @@ -224,10 +223,13 @@ def run_maintenance(db: Session) -> dict: user = get_service_user(db) if user is None: return {"reason": "no service user"} + # One `now` for the whole run: each phase taking its own let a video phase 1 had just recovered + # (stamped with phase 1's) pass phase 2's "not checked this run" test and be fetched twice. + now = _now() with YouTubeClient(db, user) as yt: - phase1 = _recheck_flagged(db, yt) + phase1 = _recheck_flagged(db, yt, now) phase2 = ( - _revalidate_rolling(db, yt) + _revalidate_rolling(db, yt, now) if not quota.backfill_reserve_reached() else {"checked": 0, "flagged": 0, "skipped": "low quota"} ) diff --git a/backend/tests/test_maintenance_lifecycle.py b/backend/tests/test_maintenance_lifecycle.py index ec725568..5973065e 100644 --- a/backend/tests/test_maintenance_lifecycle.py +++ b/backend/tests/test_maintenance_lifecycle.py @@ -45,7 +45,7 @@ def test_a_flagged_video_that_answers_again_is_unhidden(db, channel): ) db.add(v) db.commit() - out = maintenance._recheck_flagged(db, _YT({"v1": _item("v1")})) + out = maintenance._recheck_flagged(db, _YT({"v1": _item("v1")}), now_utc()) db.refresh(v) assert v.unavailable_since is None and v.unavailable_reason is None assert out == {"checked": 1, "recovered": 1, "deleted": 0, "notified": 0} @@ -60,7 +60,7 @@ def test_a_video_still_missing_inside_its_grace_period_is_kept_hidden(db, channe ) db.add(v) db.commit() - out = maintenance._recheck_flagged(db, _YT({})) + out = maintenance._recheck_flagged(db, _YT({}), now_utc()) assert db.get(Video, "v1") is not None assert out["deleted"] == 0 @@ -92,7 +92,7 @@ def test_past_the_grace_period_it_is_deleted_and_each_user_who_kept_it_told_once ] ) db.commit() - out = maintenance._recheck_flagged(db, _YT({})) + out = maintenance._recheck_flagged(db, _YT({}), now_utc()) db.expire_all() assert db.get(Video, "v1") is None and db.get(Video, "v2") is None assert db.execute(select(PlaylistItem)).scalars().all() == [] # the cascade @@ -118,7 +118,7 @@ def test_rolling_revalidation_flags_the_missing_and_keeps_a_live_stream(db, chan ) db.commit() yt = _YT({"ok": _item("ok"), "live": _item("live", liveBroadcastContent="live")}) - out = maintenance._revalidate_rolling(db, yt) + out = maintenance._revalidate_rolling(db, yt, now_utc()) db.expire_all() assert db.get(Video, "gone").unavailable_reason == "removed" assert db.get(Video, "ok").unavailable_since is None @@ -127,6 +127,41 @@ def test_rolling_revalidation_flags_the_missing_and_keeps_a_live_stream(db, chan assert all(db.get(Video, v).last_checked_at is not None for v in ("ok", "gone", "live")) +def test_a_run_asks_youtube_about_each_video_once(db, channel, monkeypatch, make_user): + # Phase 1 recovers v1; phase 2 must not take it as "not checked this run" and fetch it again. + monkeypatch.setattr(maintenance.sysconfig, "get_int", lambda db, key: 10) + monkeypatch.setattr(maintenance.quota, "backfill_reserve_reached", lambda: False) + yt = _YT({"v1": _item("v1"), "v2": _item("v2")}) + + class _Client: + def __init__(self, *_): + pass + + def __enter__(self): + return yt + + def __exit__(self, *_): + return False + + monkeypatch.setattr(maintenance, "YouTubeClient", _Client) + monkeypatch.setattr(maintenance, "get_service_user", lambda db: make_user()) + db.add_all( + [ + Video( + id="v1", + channel_id=channel.id, + unavailable_since=now_utc(), + unavailable_reason="removed", + ), + Video(id="v2", channel_id=channel.id), + ] + ) + db.commit() + maintenance.run_maintenance(db) + asked = [vid for batch in yt.asked for vid in batch] + assert sorted(asked) == ["v1", "v2"] + + def test_an_upcoming_premiere_counts_as_abandoned_only_when_it_never_went_live(): past = (now_utc() - timedelta(days=3)).isoformat() soon = (now_utc() - timedelta(days=1)).isoformat()