peter
1b1e3cdcf2
A row's one dot said how the LAST run went, and an interrupted run fell through to "idle" — a grey dot that read as "no run recorded". The row now leads with its last three runs, newest first, each named for a screen reader and in its tooltip (how it went, when, who started it); an interrupted run is a ring, set apart from a skipped run and a failure, and its line says the server restarted under it. A channel's full-history backfill gets its own "Ad-hoc runs" block: its runs, its channel and what it fetched, no interval and no run button — it needs its target. AD_HOC_JOBS names every such job; the backend-id contract holds their names in both languages, and run_ad_hoc refuses any other id.
414 lines
16 KiB
Python
414 lines
16 KiB
Python
"""The scheduler's run path, DB-lane: the single-run guard, and every run as a durable `job_runs` row
|
|
— its outcome as data, what it spent, the lines it logged — that outlives the process.
|
|
|
|
_claim_running is the single-run guard that both the scheduled fire and the manual trigger go
|
|
through (at the top of _job): holding _activity_lock across the check-and-set means exactly one
|
|
caller can hold a job's "running" claim at a time, and it's claimable again once released."""
|
|
|
|
import logging
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app import jobruns, logs, logsink, quota, scheduler, state
|
|
from app.models import AuditLog, Channel, JobRun, Notification, QuotaEvent
|
|
from app.runstatus import RunStatus, RunTrigger
|
|
|
|
_JOB = "unit_test_claim" # not a real job id — exercises the primitive without touching live jobs
|
|
|
|
|
|
def _release():
|
|
scheduler._record(_JOB, running=False)
|
|
|
|
|
|
def test_claim_is_exclusive_until_released():
|
|
_release()
|
|
# First caller claims it.
|
|
assert scheduler._claim_running(_JOB) is True
|
|
# A second caller is refused while the first still holds it (the TOCTOU the fix closes).
|
|
assert scheduler._claim_running(_JOB) is False
|
|
assert scheduler._claim_running(_JOB) is False
|
|
# _job's finally releases the claim...
|
|
_release()
|
|
# ...and now it can be claimed again.
|
|
assert scheduler._claim_running(_JOB) is True
|
|
_release()
|
|
|
|
|
|
# ---- the result as data --------------------------------------------------------------------------
|
|
|
|
|
|
def _snapshot_job(db, job_id):
|
|
return next(j for j in scheduler.scheduler_snapshot(db)["jobs"] if j["id"] == job_id)
|
|
|
|
|
|
def _runs(db, job=None) -> list[JobRun]:
|
|
db.expire_all()
|
|
q = select(JobRun).order_by(JobRun.id)
|
|
if job is not None:
|
|
q = q.where(JobRun.job == job)
|
|
return list(db.execute(q).scalars())
|
|
|
|
|
|
def _manual(actor_id, name, fn):
|
|
token = scheduler._manual_actor.set(actor_id)
|
|
try:
|
|
scheduler._job(name, fn)
|
|
finally:
|
|
scheduler._manual_actor.reset(token)
|
|
|
|
|
|
def test_a_run_keeps_its_result_as_data_for_the_readers_language(db, make_user):
|
|
"""The dashboard and the manual-run notice showed `_summarize`'s English one-liner ("new=3,
|
|
failed=0", a Python repr for a nested phase). The result rides along as data, so the client names
|
|
it; the one-liner stays as the fallback."""
|
|
admin = make_user("sched@example.com", role="admin")
|
|
_manual(admin.id, "rss_poll", lambda _db: {"new": 3, "failed": 0, "fallback": 1})
|
|
_manual(admin.id, "enrich", lambda _db: 5) # enrich returns a bare count
|
|
|
|
rss = _snapshot_job(db, "rss_poll")
|
|
assert rss["result"] == {"new": 3, "failed": 0, "fallback": 1}
|
|
assert rss["last_result"] == "new=3, failed=0, fallback=1"
|
|
assert _snapshot_job(db, "enrich")["result"] == {"enriched": 5}
|
|
|
|
notices = db.execute(
|
|
select(Notification).where(Notification.user_id == admin.id).order_by(Notification.id)
|
|
).scalars()
|
|
assert [n.data["result"] for n in notices] == [
|
|
{"new": 3, "failed": 0, "fallback": 1},
|
|
{"enriched": 5},
|
|
]
|
|
|
|
|
|
def test_a_failed_run_drops_the_previous_result(db):
|
|
scheduler._job("rss_poll", lambda _db: {"new": 1})
|
|
scheduler._job("rss_poll", lambda _db: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
rss = _snapshot_job(db, "rss_poll")
|
|
assert rss["status"] == "error"
|
|
assert rss["last_error"] == "boom"
|
|
assert rss["result"] is None
|
|
|
|
|
|
def test_a_failure_the_job_names_rides_along_as_data(db, make_user):
|
|
"""A failed RSS poll reached the Hungarian UI as the server's English sentence. A failure the job
|
|
names (`progress.JobFailed`) is stored as data beside that sentence — on the run's row, the
|
|
dashboard and the manual-run notice — so the client says it in the reader's language."""
|
|
from app import progress
|
|
|
|
def outage(_db):
|
|
raise progress.JobFailed("rss_all_failed", "RSS poll failed", total=2, cause="transport")
|
|
|
|
admin = make_user("failure@example.com", role="admin")
|
|
_manual(admin.id, "rss_poll", outage)
|
|
|
|
named = {"failure": "rss_all_failed", "total": 2, "cause": "transport"}
|
|
rss = _snapshot_job(db, "rss_poll")
|
|
assert rss["last_error"] == "RSS poll failed"
|
|
assert rss["result"] == named
|
|
notice = db.execute(select(Notification).where(Notification.user_id == admin.id)).scalar_one()
|
|
assert notice.data["result"] == named
|
|
(run,) = _runs(db, "rss_poll")
|
|
assert (run.status, run.error, run.failure) == ("error", "RSS poll failed", named)
|
|
|
|
|
|
# ---- the durable row -----------------------------------------------------------------------------
|
|
|
|
|
|
def test_every_run_is_a_row_with_who_started_it_and_none_in_the_audit_log(db, make_user):
|
|
"""A no-op scheduled poll was not written anywhere (the audit rule kept only changes), so "did
|
|
it run at 03:00?" had no answer. Every run is a row now; the audit log keeps admin actions only."""
|
|
admin = make_user("runs@example.com", role="admin")
|
|
scheduler._job("rss_poll", lambda _db: {"new": 0})
|
|
_manual(admin.id, "rss_poll", lambda _db: {"new": 2})
|
|
|
|
scheduled, manual = _runs(db, "rss_poll")
|
|
assert (scheduled.trigger, scheduled.actor_id, scheduled.status) == ("scheduled", None, "ok")
|
|
assert scheduled.result == {"new": 0}
|
|
assert (manual.trigger, manual.actor_id, manual.summary) == ("manual", admin.id, "new=2")
|
|
assert manual.finished_at is not None and manual.finished_at >= manual.started_at
|
|
assert db.execute(select(AuditLog)).scalars().all() == []
|
|
|
|
|
|
def test_the_last_run_survives_a_restart(db):
|
|
"""The dashboard's "last run" lived in the process's memory: a deploy blanked every job."""
|
|
scheduler._job("plex_sync", lambda _db: {"items": 4})
|
|
with scheduler._activity_lock:
|
|
scheduler._activity.clear() # what a new process starts with
|
|
|
|
plex = _snapshot_job(db, "plex_sync")
|
|
assert (plex["status"], plex["result"], plex["last_trigger"]) == (
|
|
"ok",
|
|
{"items": 4},
|
|
"scheduled",
|
|
)
|
|
assert plex["last_finished"] is not None
|
|
|
|
|
|
def test_a_row_shows_its_last_three_runs_newest_first_an_interrupted_one_included(db):
|
|
"""The row said only how the LAST run went; an interrupted run read as "no run recorded"."""
|
|
oldest = jobruns.start("shorts", RunTrigger.SCHEDULED, None)
|
|
jobruns.interrupt_leftovers()
|
|
scheduler._job("shorts", lambda _db: {"probed": 1})
|
|
scheduler._job("shorts", lambda _db: {"probed": 2})
|
|
scheduler._job("shorts", lambda _db: 1 / 0)
|
|
scheduler._job("rss_poll", lambda _db: {"new": 0}) # another job's run stays on its own row
|
|
|
|
recent = _snapshot_job(db, "shorts")["recent"]
|
|
assert [r["status"] for r in recent] == ["error", "ok", "ok"]
|
|
assert oldest not in [r["id"] for r in recent] # the fourth-newest is not shown
|
|
assert recent[0]["trigger"] == "scheduled" and recent[0]["finished_at"] is not None
|
|
assert [r["status"] for r in _snapshot_job(db, "rss_poll")["recent"]] == ["ok"]
|
|
|
|
jobruns.start("shorts", RunTrigger.SCHEDULED, None)
|
|
jobruns.interrupt_leftovers()
|
|
shorts = _snapshot_job(db, "shorts")
|
|
assert shorts["status"] == "interrupted"
|
|
assert [r["status"] for r in shorts["recent"]] == ["interrupted", "error", "ok"]
|
|
|
|
|
|
def test_a_paused_sync_is_a_skipped_run(db):
|
|
state.set_sync_paused(db, True)
|
|
db.commit()
|
|
scheduler._job("rss_poll", lambda _db: pytest.fail("a paused run must not do its work"))
|
|
(run,) = _runs(db, "rss_poll")
|
|
assert (run.status, run.result) == ("skipped", {"skipped": "sync paused"})
|
|
|
|
|
|
def test_a_run_left_running_by_a_stopped_process_is_interrupted_and_stays_so(db):
|
|
"""A deploy mid-run left nothing behind before; now the row would say `running` forever. The
|
|
next startup closes it, and the stale thread's own finish cannot rewrite it."""
|
|
run_id = jobruns.start("backfill", RunTrigger.SCHEDULED, None)
|
|
assert jobruns.interrupt_leftovers() == 1
|
|
|
|
(run,) = _runs(db, "backfill")
|
|
assert run.status == "interrupted" and run.finished_at is not None
|
|
jobruns.finish(run_id, RunStatus.OK, summary="late")
|
|
(run,) = _runs(db, "backfill")
|
|
assert (run.status, run.summary) == ("interrupted", None)
|
|
|
|
|
|
def test_a_run_knows_what_it_spent_and_a_request_spend_is_nobodys_run(db):
|
|
def spends(_db):
|
|
quota.record_usage(5)
|
|
quota.record_usage(3)
|
|
return {"enriched": 2}
|
|
|
|
scheduler._job("enrich", spends)
|
|
quota.record_usage(11) # a request's own spend, outside any run
|
|
|
|
(run,) = _runs(db, "enrich")
|
|
assert run.quota_units == 8
|
|
events = db.execute(select(QuotaEvent.units, QuotaEvent.job_run_id)).all()
|
|
assert sorted(events) == [(3, run.id), (5, run.id), (11, None)]
|
|
assert _snapshot_job(db, "enrich")["quota_units"] == 8
|
|
|
|
|
|
class _Collect(logging.Handler):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.addFilter(logs.ContextFilter())
|
|
self.records: list[logging.LogRecord] = []
|
|
|
|
def emit(self, record):
|
|
self.records.append(record)
|
|
|
|
|
|
def test_a_runs_log_lines_carry_its_id(db):
|
|
"""So a run's row can open exactly the lines it logged (the log table's `run_id`)."""
|
|
probe = logging.getLogger("siftlode.test.run")
|
|
handler = _Collect()
|
|
probe.addHandler(handler)
|
|
probe.setLevel(logging.INFO)
|
|
try:
|
|
scheduler._job("autotag", lambda _db: probe.info("tagging") or {"tagged": 1})
|
|
probe.info("after the run")
|
|
finally:
|
|
probe.removeHandler(handler)
|
|
|
|
(run,) = _runs(db, "autotag")
|
|
inside, after = (logs.record_fields(r).get("run_id") for r in handler.records)
|
|
assert (inside, after) == (run.id, None)
|
|
# …and it reaches the stored row, whose columns are the model's.
|
|
assert logsink.to_row(logs.record_fields(handler.records[0]))["run_id"] == run.id
|
|
|
|
|
|
def test_retention_reaps_finished_runs_only(db):
|
|
old = datetime.now(timezone.utc) - timedelta(days=40)
|
|
db.add_all(
|
|
[
|
|
JobRun(
|
|
job="rss_poll",
|
|
trigger="scheduled",
|
|
status="ok",
|
|
started_at=old,
|
|
finished_at=old,
|
|
),
|
|
JobRun(job="backfill", trigger="scheduled", status="running", started_at=old),
|
|
JobRun(
|
|
job="rss_poll",
|
|
trigger="scheduled",
|
|
status="ok",
|
|
started_at=datetime.now(timezone.utc),
|
|
finished_at=datetime.now(timezone.utc),
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
assert jobruns.gc(db, 0) == 0 # 0 = keep forever
|
|
assert jobruns.gc(db, 30) == 1
|
|
db.commit()
|
|
assert sorted((r.job, r.status) for r in _runs(db)) == [
|
|
("backfill", "running"),
|
|
("rss_poll", "ok"),
|
|
]
|
|
|
|
|
|
# ---- "run now" through the scheduler --------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def live_scheduler(db, monkeypatch):
|
|
"""This process's scheduler, started as the API starts it (on the migrated schema `db` brings),
|
|
with interval jobs off."""
|
|
monkeypatch.setattr(scheduler.settings, "scheduler_enabled", False)
|
|
scheduler.start_scheduler()
|
|
try:
|
|
yield scheduler._scheduler
|
|
finally:
|
|
scheduler.shutdown_scheduler()
|
|
|
|
|
|
def _wait_finished(db, job, timeout=10.0) -> JobRun:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
runs = [r for r in _runs(db, job) if r.status != "running"]
|
|
if runs:
|
|
return runs[-1]
|
|
time.sleep(0.05)
|
|
raise AssertionError(f"{job} did not finish")
|
|
|
|
|
|
def test_run_now_goes_through_the_scheduler_even_with_interval_jobs_off(
|
|
db, make_user, live_scheduler, monkeypatch
|
|
):
|
|
admin = make_user("now@example.com", role="admin")
|
|
monkeypatch.setitem(
|
|
scheduler.JOB_FUNCS, "shorts", lambda: scheduler._job("shorts", lambda _db: {"probed": 3})
|
|
)
|
|
assert live_scheduler.get_jobs() == [] # SCHEDULER_ENABLED=false: no interval job
|
|
live_scheduler.pause() # hold the one-shot so the second click finds it pending
|
|
assert scheduler.trigger_job("shorts", admin.id) == "started"
|
|
assert scheduler.trigger_job("shorts", admin.id) == "already_running"
|
|
assert [j.id for j in live_scheduler.get_jobs()] == ["shorts@now"]
|
|
live_scheduler.resume()
|
|
|
|
run = _wait_finished(db, "shorts")
|
|
assert (run.trigger, run.actor_id, run.result) == ("manual", admin.id, {"probed": 3})
|
|
|
|
|
|
def test_an_ad_hoc_run_is_the_users_run_of_its_target_with_its_spend_and_failure(
|
|
db, make_user, live_scheduler
|
|
):
|
|
"""The one-channel deep backfill ran as a BackgroundTask nothing recorded: its spend counted as
|
|
nobody's run and a failure lived only in the log."""
|
|
user = make_user("adhoc@example.com")
|
|
|
|
def work():
|
|
quota.record_usage(3)
|
|
return {"videos_added": 50}
|
|
|
|
scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, work, actor_id=user.id, target="UC_a")
|
|
run = _wait_finished(db, "channel_backfill")
|
|
assert (run.trigger, run.actor_id, run.target) == ("ad_hoc", user.id, "UC_a")
|
|
assert (run.status, run.result, run.quota_units) == ("ok", {"videos_added": 50}, 3)
|
|
|
|
def fail():
|
|
raise RuntimeError("quota exceeded")
|
|
|
|
scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, fail, actor_id=user.id, target="UC_b")
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline and len(_finished(db, "channel_backfill")) < 2:
|
|
time.sleep(0.05)
|
|
failed = next(r for r in _runs(db, "channel_backfill") if r.target == "UC_b")
|
|
assert (failed.status, failed.error) == ("error", "quota exceeded")
|
|
|
|
|
|
def test_the_page_shows_the_ad_hoc_runs_beside_the_jobs_with_their_channel(db, make_user):
|
|
user = make_user("shown@example.com")
|
|
db.add(Channel(id="UC_shown", title="BBC Earth"))
|
|
db.commit()
|
|
scheduler._ad_hoc("channel_backfill", lambda: {"skipped": "at capacity"}, user.id, "UC_none")
|
|
scheduler._ad_hoc("channel_backfill", lambda: {"videos_added": 69}, user.id, "UC_shown")
|
|
|
|
(row,) = scheduler.scheduler_snapshot(db)["ad_hoc"]
|
|
assert (row["id"], row["running"], row["target"], row["target_title"]) == (
|
|
"channel_backfill",
|
|
False,
|
|
"UC_shown",
|
|
"BBC Earth",
|
|
)
|
|
assert (row["status"], row["result"], row["last_trigger"]) == (
|
|
"ok",
|
|
{"videos_added": 69},
|
|
"ad_hoc",
|
|
)
|
|
assert len(row["recent"]) == 2
|
|
assert all(j["id"] != "channel_backfill" for j in scheduler.scheduler_snapshot(db)["jobs"])
|
|
|
|
jobruns.start("channel_backfill", RunTrigger.AD_HOC, user.id, "UC_shown")
|
|
assert scheduler.scheduler_snapshot(db)["ad_hoc"][0]["running"] is True
|
|
|
|
|
|
def _finished(db, job) -> list[JobRun]:
|
|
return [r for r in _runs(db, job) if r.status != "running"]
|
|
|
|
|
|
def test_an_ad_hoc_job_outside_the_named_list_is_refused(live_scheduler):
|
|
"""The page names every ad-hoc job from AD_HOC_JOBS; one outside it would show as a bare id."""
|
|
with pytest.raises(KeyError):
|
|
scheduler.run_ad_hoc("unnamed_job", dict, actor_id=1, target="x")
|
|
assert live_scheduler.get_jobs() == []
|
|
|
|
|
|
def test_a_second_ad_hoc_request_before_the_first_starts_is_that_same_run(
|
|
db, make_user, live_scheduler
|
|
):
|
|
user = make_user("twice@example.com")
|
|
live_scheduler.pause() # hold the one-shots so the second request finds the first pending
|
|
for target in ("UC_x", "UC_x", "UC_y"):
|
|
scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, dict, actor_id=user.id, target=target)
|
|
assert sorted(j.id for j in live_scheduler.get_jobs()) == [
|
|
"channel_backfill@UC_x",
|
|
"channel_backfill@UC_y",
|
|
]
|
|
live_scheduler.remove_all_jobs() # never run: a run finishing after the test would land in the next
|
|
|
|
|
|
def test_a_manual_run_leaves_no_actor_behind_on_its_thread(db, make_user):
|
|
"""The scheduler's pool threads are reused: an actor a manual run left set made the next
|
|
scheduled run on that thread an admin's "manual" run, with a notice in their inbox."""
|
|
admin = make_user("pool@example.com", role="admin")
|
|
seen = []
|
|
scheduler._as_actor(lambda: seen.append(scheduler._manual_actor.get()), admin.id)()
|
|
assert seen == [admin.id]
|
|
assert scheduler._manual_actor.get() is None # this same thread's next task runs as nobody
|
|
|
|
|
|
def test_run_now_refuses_a_job_already_running(live_scheduler):
|
|
assert scheduler._claim_running("maintenance") is True
|
|
try:
|
|
assert scheduler.trigger_job("maintenance") == "already_running"
|
|
finally:
|
|
scheduler._record("maintenance", running=False)
|
|
|
|
|
|
def test_startup_closes_what_the_previous_process_left_running(db, monkeypatch):
|
|
jobruns.start("plex_sync", RunTrigger.MANUAL, None)
|
|
monkeypatch.setattr(scheduler.settings, "scheduler_enabled", False)
|
|
scheduler.start_scheduler()
|
|
scheduler.shutdown_scheduler()
|
|
(run,) = _runs(db, "plex_sync")
|
|
assert run.status == "interrupted"
|