diff --git a/indexnow.py b/indexnow.py index a9323bd..857b4be 100644 --- a/indexnow.py +++ b/indexnow.py @@ -133,6 +133,23 @@ def _write_state(sig: str) -> None: log.warning("could not persist IndexNow state to %s", _STATE_PATH) +def submit_if_changed(base_url: str | None = None) -> dict | None: + """Submit every indexable page, but only when the URL set changed since the + last successful submission — the shared logic behind the CLI's --if-changed + flag and the worker scheduler's periodic ping (notifications/scheduler.py), so + a code-only deploy or an unremarkable scheduled tick never resubmits. Returns + the submit_all() result, or None when skipped (unchanged, or a partial/failed + submission that leaves the prior state in place so the next attempt retries).""" + base = base_url or os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org") + sig = url_signature() + if sig == _read_state(): + return None + result = submit_all(base) + if result["total"] and result["submitted"] == result["total"]: + _write_state(sig) # only remember a fully-successful submission + return result + + if __name__ == "__main__": import sys @@ -145,14 +162,16 @@ if __name__ == "__main__": base = (positional[0] if positional else "") or os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org") logging.basicConfig(level=logging.INFO) - sig = url_signature() - if if_changed and sig == _read_state(): - print("IndexNow: URL set unchanged since last submission — skipping.") - sys.exit(0) - print(f"IndexNow key: {key()} (served at {base.rstrip('/')}/{key()}.txt)") - result = submit_all(base) + if if_changed: + result = submit_if_changed(base) + if result is None: + print("IndexNow: URL set unchanged since last submission — skipping.") + sys.exit(0) + else: + result = submit_all(base) print(f"Submitted {result['submitted']}/{result['total']} URLs " f"in {result['batches']} batch(es); statuses={result['statuses']}") - if result["total"] and result["submitted"] == result["total"]: + if result["total"] and result["submitted"] == result["total"] and not if_changed: + sig = url_signature() _write_state(sig) # only remember a fully-successful submission diff --git a/notifications/scheduler.py b/notifications/scheduler.py new file mode 100644 index 0000000..2677f33 --- /dev/null +++ b/notifications/scheduler.py @@ -0,0 +1,68 @@ +"""Recurring worker-tier jobs: keep the curated city set warm and ping IndexNow +on a schedule, so neither needs a human to run it after every deploy anymore. + +Only the process that already won leader election for the notifier runs this +too (gated by the same check in web/app.py's _lifespan) — APScheduler has no +cross-host coordination of its own, so running it unguarded on every worker +replica would repeat every job once per replica, same reason the notifier +itself needs a leader. + +APScheduler, not a Postgres-native queue: there is exactly one process that +ever runs this, no fan-out/backpressure/dead-letter need, and it needs no new +infrastructure. procrastinate/pgqueuer are the documented upgrade path if that +changes — see docs/architecture/repo-topology-and-infrastructure.md §7. +""" +import os + +from apscheduler.schedulers.background import BackgroundScheduler + +# How often the curated city set is (re-)warmed. warm_cities.py's own docstring +# frames it as a deploy-time script ("Run at/after deploy") because a warm cache +# mostly just stays warm; scheduling it periodically instead catches a newly +# added city, or one that fell out of cache, without waiting for the next +# deploy. Daily is plenty — every already-cached city is a cheap skip. +WARM_CITIES_INTERVAL_HOURS = int(os.environ.get("THERMOGRAPH_WARM_CITIES_INTERVAL_HOURS", "24") or 24) + +# How often to check whether the indexable URL set changed and, if so, ping +# IndexNow. submit_if_changed() is a cheap no-op when nothing changed, so this +# can run more often than warm_cities without spending anything on most ticks. +INDEXNOW_INTERVAL_HOURS = int(os.environ.get("THERMOGRAPH_INDEXNOW_INTERVAL_HOURS", "6") or 6) + +_scheduler: "BackgroundScheduler | None" = None + + +def _warm_cities_job() -> None: + import warm_cities # local import: a script-style top-level module — only + # pay its import cost on the process that runs the job + warm_cities.main() + + +def _indexnow_job() -> None: + import indexnow + indexnow.submit_if_changed() + + +def start() -> None: + """Start the scheduler (idempotent, mirroring notify.start()). The caller is + responsible for the leader-election gate — see web/app.py's _lifespan.""" + global _scheduler + if _scheduler is not None: + return + sched = BackgroundScheduler(daemon=True) + # No next_run_time override: the default first run is one interval from now, + # not immediately — warm_cities already runs at deploy time (warm_cities.py / + # the deploy hook), so an immediate duplicate run on every worker boot/restart + # would just be wasted work. + sched.add_job(_warm_cities_job, "interval", hours=WARM_CITIES_INTERVAL_HOURS, + id="warm_cities") + sched.add_job(_indexnow_job, "interval", hours=INDEXNOW_INTERVAL_HOURS, + id="indexnow") + sched.start() + _scheduler = sched + + +def stop() -> None: + global _scheduler + if _scheduler is not None: + _scheduler.shutdown(wait=False) + _scheduler = None diff --git a/requirements.txt b/requirements.txt index 987dc06..d609f17 100644 --- a/requirements.txt +++ b/requirements.txt @@ -22,3 +22,6 @@ jinja2==3.1.6 # Structured SSR copy (glossary, static-page SEO meta) — see # backend/web/content_loader.py and content/. PyYAML==6.0.3 +# Recurring worker-tier jobs — city warming, IndexNow pings (see +# notifications/scheduler.py). In-process; no broker/queue needed at this scale. +apscheduler==3.11.3 diff --git a/tests/notifications/test_scheduler.py b/tests/notifications/test_scheduler.py new file mode 100644 index 0000000..3aeb929 --- /dev/null +++ b/tests/notifications/test_scheduler.py @@ -0,0 +1,79 @@ +"""Tests for the recurring worker-tier jobs (scheduler.py): job registration, +cadence from env, idempotent start/stop, and that each job calls the right +underlying function. No real hours-long wait — inspects APScheduler's own job +registry rather than letting anything actually fire.""" +import datetime + +import pytest + +from notifications import scheduler + + +@pytest.fixture(autouse=True) +def _clean_scheduler(): + """Every test starts from "not running" and leaves nothing behind, even on + failure — scheduler._scheduler is module-global state shared across tests.""" + scheduler.stop() + yield + scheduler.stop() + + +def test_start_registers_both_jobs_with_configured_intervals(monkeypatch): + monkeypatch.setattr(scheduler, "WARM_CITIES_INTERVAL_HOURS", 24) + monkeypatch.setattr(scheduler, "INDEXNOW_INTERVAL_HOURS", 6) + scheduler.start() + jobs = {j.id: j for j in scheduler._scheduler.get_jobs()} + assert set(jobs) == {"warm_cities", "indexnow"} + assert jobs["warm_cities"].trigger.interval == datetime.timedelta(hours=24) + assert jobs["indexnow"].trigger.interval == datetime.timedelta(hours=6) + assert jobs["warm_cities"].func is scheduler._warm_cities_job + assert jobs["indexnow"].func is scheduler._indexnow_job + + +def test_start_does_not_run_a_job_immediately(): + """warm_cities already runs at deploy time; an immediate duplicate run on + every worker boot/restart would just be wasted work — the first scheduled + run must be one interval away, not "now".""" + scheduler.start() + now = datetime.datetime.now(datetime.timezone.utc) + for job in scheduler._scheduler.get_jobs(): + assert job.next_run_time > now + + +def test_start_is_idempotent(monkeypatch): + scheduler.start() + first = scheduler._scheduler + calls = [] + monkeypatch.setattr(scheduler.BackgroundScheduler, "__init__", + lambda self, **kw: calls.append(1) or None) + scheduler.start() + assert scheduler._scheduler is first + assert calls == [] # a second BackgroundScheduler was never constructed + + +def test_stop_shuts_down_and_clears_state(): + scheduler.start() + assert scheduler._scheduler is not None + scheduler.stop() + assert scheduler._scheduler is None + + +def test_stop_without_start_is_a_safe_no_op(): + scheduler.stop() # must not raise + assert scheduler._scheduler is None + + +def test_warm_cities_job_calls_warm_cities_main(monkeypatch): + import warm_cities + calls = [] + monkeypatch.setattr(warm_cities, "main", lambda: calls.append(1)) + scheduler._warm_cities_job() + assert calls == [1] + + +def test_indexnow_job_calls_submit_if_changed(monkeypatch): + import indexnow + calls = [] + monkeypatch.setattr(indexnow, "submit_if_changed", lambda: calls.append(1)) + scheduler._indexnow_job() + assert calls == [1] diff --git a/tests/test_indexnow.py b/tests/test_indexnow.py new file mode 100644 index 0000000..d2a05c2 --- /dev/null +++ b/tests/test_indexnow.py @@ -0,0 +1,59 @@ +"""submit_if_changed: the shared skip-when-unchanged logic behind the CLI's +--if-changed flag and the worker scheduler's periodic ping (no network — +submit_all/url_signature are stubbed).""" +import indexnow + + +def _ok(total=3): + return {"submitted": total, "total": total, "batches": 1, "statuses": [200]} + + +def test_skips_when_url_set_unchanged(monkeypatch, tmp_path): + monkeypatch.setattr(indexnow, "_STATE_PATH", str(tmp_path / "state.txt")) + monkeypatch.setattr(indexnow, "url_signature", lambda: "same-sig") + (tmp_path / "state.txt").write_text("same-sig") + + def boom(*a, **k): + raise AssertionError("must not submit when the URL set is unchanged") + + monkeypatch.setattr(indexnow, "submit_all", boom) + assert indexnow.submit_if_changed("https://example.org") is None + + +def test_submits_and_persists_state_when_changed(monkeypatch, tmp_path): + state = tmp_path / "state.txt" + monkeypatch.setattr(indexnow, "_STATE_PATH", str(state)) + monkeypatch.setattr(indexnow, "url_signature", lambda: "new-sig") + state.write_text("old-sig") + + calls = [] + monkeypatch.setattr(indexnow, "submit_all", lambda base, **kw: calls.append(base) or _ok()) + result = indexnow.submit_if_changed("https://example.org") + assert calls == ["https://example.org"] + assert result["submitted"] == result["total"] == 3 + assert state.read_text() == "new-sig" + + +def test_partial_failure_does_not_persist_state(monkeypatch, tmp_path): + """A partial/failed submission must leave the prior state in place so the + next attempt retries, rather than silently giving up on the missed URLs.""" + state = tmp_path / "state.txt" + monkeypatch.setattr(indexnow, "_STATE_PATH", str(state)) + monkeypatch.setattr(indexnow, "url_signature", lambda: "new-sig") + state.write_text("old-sig") + + monkeypatch.setattr(indexnow, "submit_all", + lambda base, **kw: {"submitted": 1, "total": 3, "batches": 1, "statuses": [429]}) + result = indexnow.submit_if_changed("https://example.org") + assert result["submitted"] < result["total"] + assert state.read_text() == "old-sig" + + +def test_default_base_url_falls_back_to_env(monkeypatch, tmp_path): + monkeypatch.setattr(indexnow, "_STATE_PATH", str(tmp_path / "state.txt")) + monkeypatch.setattr(indexnow, "url_signature", lambda: "sig") + monkeypatch.setenv("THERMOGRAPH_BASE_URL", "https://from-env.example") + calls = [] + monkeypatch.setattr(indexnow, "submit_all", lambda base, **kw: calls.append(base) or _ok()) + indexnow.submit_if_changed() + assert calls == ["https://from-env.example"] diff --git a/web/app.py b/web/app.py index 3bed0ae..acc3583 100644 --- a/web/app.py +++ b/web/app.py @@ -27,6 +27,7 @@ from accounts import db from data import grid from core import metrics from notifications import notify +from notifications import scheduler import paths from data import places from core import singleton @@ -159,8 +160,12 @@ async def _lifespan(app): # N-replica web tier can scale without also scaling notifier instances. if _should_run_notifier(): notify.start() + # Same leader gate as the notifier above — the recurring worker jobs + # (city warming, IndexNow) must likewise run in exactly one process. + scheduler.start() yield notify.stop() + scheduler.stop() app = FastAPI(title="Thermograph", version="0.2.0", lifespan=_lifespan)