fix(maintenance): one clock per run, so a video phase 1 recovers is not fetched again in phase 2
E32 review round 1 (B7). Each phase took its own now; phase 2's "not checked this run" test then let a just-recovered video through and spent another videos.list call on it.
This commit is contained in:
@@ -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.
|
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 import or_, select
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
@@ -129,14 +129,14 @@ def _notify_removed(db: Session, impact: dict[int, list[dict]]) -> int:
|
|||||||
return len(impact)
|
return len(impact)
|
||||||
|
|
||||||
|
|
||||||
def _recheck_flagged(db: Session, yt: YouTubeClient) -> dict:
|
def _recheck_flagged(db: Session, yt: YouTubeClient, now: datetime) -> dict:
|
||||||
"""Phase 1: recover or hard-delete videos already flagged unavailable."""
|
"""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()
|
flagged = db.execute(select(Video).where(Video.unavailable_since.is_not(None))).scalars().all()
|
||||||
if not flagged:
|
if not flagged:
|
||||||
return {"checked": 0, "recovered": 0, "deleted": 0, "notified": 0}
|
return {"checked": 0, "recovered": 0, "deleted": 0, "notified": 0}
|
||||||
progress.report(0, len(flagged), "recheck")
|
progress.report(0, len(flagged), "recheck")
|
||||||
items = {it["id"]: it for it in yt.get_videos([v.id for v in flagged])}
|
items = {it["id"]: it for it in yt.get_videos([v.id for v in flagged])}
|
||||||
now = _now()
|
|
||||||
recovered = 0
|
recovered = 0
|
||||||
to_delete: list[Video] = []
|
to_delete: list[Video] = []
|
||||||
for video in flagged:
|
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
|
"""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)."""
|
that have become unplayable (hidden immediately; deleted later if they don't recover)."""
|
||||||
batch = sysconfig.get_int(
|
batch = sysconfig.get_int(
|
||||||
@@ -181,7 +181,6 @@ def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict:
|
|||||||
) # admin override, else config default
|
) # admin override, else config default
|
||||||
flagged_new = 0
|
flagged_new = 0
|
||||||
checked = 0
|
checked = 0
|
||||||
now = _now()
|
|
||||||
# Process in videos.list-sized pages so we commit incrementally and stop on low quota.
|
# Process in videos.list-sized pages so we commit incrementally and stop on low quota.
|
||||||
while checked < batch:
|
while checked < batch:
|
||||||
if quota.backfill_reserve_reached():
|
if quota.backfill_reserve_reached():
|
||||||
@@ -224,10 +223,13 @@ def run_maintenance(db: Session) -> dict:
|
|||||||
user = get_service_user(db)
|
user = get_service_user(db)
|
||||||
if user is None:
|
if user is None:
|
||||||
return {"reason": "no service user"}
|
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:
|
with YouTubeClient(db, user) as yt:
|
||||||
phase1 = _recheck_flagged(db, yt)
|
phase1 = _recheck_flagged(db, yt, now)
|
||||||
phase2 = (
|
phase2 = (
|
||||||
_revalidate_rolling(db, yt)
|
_revalidate_rolling(db, yt, now)
|
||||||
if not quota.backfill_reserve_reached()
|
if not quota.backfill_reserve_reached()
|
||||||
else {"checked": 0, "flagged": 0, "skipped": "low quota"}
|
else {"checked": 0, "flagged": 0, "skipped": "low quota"}
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -45,7 +45,7 @@ def test_a_flagged_video_that_answers_again_is_unhidden(db, channel):
|
|||||||
)
|
)
|
||||||
db.add(v)
|
db.add(v)
|
||||||
db.commit()
|
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)
|
db.refresh(v)
|
||||||
assert v.unavailable_since is None and v.unavailable_reason is None
|
assert v.unavailable_since is None and v.unavailable_reason is None
|
||||||
assert out == {"checked": 1, "recovered": 1, "deleted": 0, "notified": 0}
|
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.add(v)
|
||||||
db.commit()
|
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 db.get(Video, "v1") is not None
|
||||||
assert out["deleted"] == 0
|
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()
|
db.commit()
|
||||||
out = maintenance._recheck_flagged(db, _YT({}))
|
out = maintenance._recheck_flagged(db, _YT({}), now_utc())
|
||||||
db.expire_all()
|
db.expire_all()
|
||||||
assert db.get(Video, "v1") is None and db.get(Video, "v2") is None
|
assert db.get(Video, "v1") is None and db.get(Video, "v2") is None
|
||||||
assert db.execute(select(PlaylistItem)).scalars().all() == [] # the cascade
|
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()
|
db.commit()
|
||||||
yt = _YT({"ok": _item("ok"), "live": _item("live", liveBroadcastContent="live")})
|
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()
|
db.expire_all()
|
||||||
assert db.get(Video, "gone").unavailable_reason == "removed"
|
assert db.get(Video, "gone").unavailable_reason == "removed"
|
||||||
assert db.get(Video, "ok").unavailable_since is None
|
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"))
|
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():
|
def test_an_upcoming_premiere_counts_as_abandoned_only_when_it_never_went_live():
|
||||||
past = (now_utc() - timedelta(days=3)).isoformat()
|
past = (now_utc() - timedelta(days=3)).isoformat()
|
||||||
soon = (now_utc() - timedelta(days=1)).isoformat()
|
soon = (now_utc() - timedelta(days=1)).isoformat()
|
||||||
|
|||||||
Reference in New Issue
Block a user