diff --git a/backendServer/backend/settings/base.py b/backendServer/backend/settings/base.py
index 7591ce5..0a7f618 100644
--- a/backendServer/backend/settings/base.py
+++ b/backendServer/backend/settings/base.py
@@ -150,7 +150,23 @@
# Upper bound on beat's sleep between ticks — it still wakes exactly on time for
# actual due tasks (scrape 3:30am, ingest 4:00am, digest), this only caps how often
# it re-polls the DB for schedule changes in between, to reduce Neon wake-ups.
+#
+# Trade-off: DatabaseScheduler keeps `PeriodicTask.last_run_at` in memory and only
+# flushes it to Postgres on a sync. With beat_sync_every left at its default (0),
+# the only other sync trigger is time-based (every 3 min, capped by this interval),
+# so last_run_at can visibly lag reality by up to 6h until CELERY_BEAT_SYNC_EVERY
+# (below) forces a flush after every task send instead. See docs/ingestion-monitoring.md
+# "Beat scheduler: last_run_at persistence lag" for the full incident writeup.
CELERY_BEAT_MAX_LOOP_INTERVAL = 6 * 60 * 60
+# Force a DB sync after every single task send (celery/beat.py Scheduler.should_sync:
+# `self.sync_every_tasks and self._tasks_since_sync >= self.sync_every_tasks`), maps to
+# app.conf.beat_sync_every via Scheduler.__init__ (celery/beat.py:262-264). Without this,
+# a task firing within 180s of the prior sync (beat_sync_every default 0 disables the
+# task-count trigger, and the time-based trigger is `sync_every=180`) leaves
+# last_run_at stuck in memory until the next sync opportunity — up to 6h away given
+# CELERY_BEAT_MAX_LOOP_INTERVAL above. Costs one UPDATE per task fire; does not touch
+# the loop interval or Neon poll rate.
+CELERY_BEAT_SYNC_EVERY = 1
CELERY_TIMEZONE = "UTC"
# Headless-Chrome scraping is memory-heavy — pin it to its own queue drained by a
diff --git a/backendServer/devtools/monitoring.py b/backendServer/devtools/monitoring.py
index 57d6792..f82a41a 100644
--- a/backendServer/devtools/monitoring.py
+++ b/backendServer/devtools/monitoring.py
@@ -311,6 +311,25 @@ def _raw_zero_note(raw_all_time, latest_raw_created_at_dt, now):
)
+def _no_town_note(no_town_count: int) -> str | None:
+ """Compose the message for a nonzero `no_town` funnel cell.
+
+ Ticket 36.1's whole motivation: a source can sit at `raw > 0, published =
+ 0` forever with no explanation, because its town is deliberately out of
+ coverage (e.g. Apex, Durham on prod). `skipped_no_town` is terminal-ish —
+ `publish_all_approved` only logs it once per row — so this note is the
+ monitor surfacing that same explanation instead of leaving the operator to
+ rediscover it in the logs. Mirrors `_raw_zero_note`: only rendered when
+ there is something to explain, never replacing the plain count.
+ """
+ if no_town_count == 0:
+ return None
+ return (
+ f"{no_town_count} skipped — town not in coverage "
+ f"(see `manage.py reopen_skipped_towns` once added)"
+ )
+
+
def _source_rows(db, start, end, source_type_filter, runs_state=None):
# Staleness is a wall-clock question ("has this source polled recently?"),
# so it is measured against real `now`, never the window's `end`. In prod
@@ -423,6 +442,13 @@ def _source_rows(db, start, end, source_type_filter, runs_state=None):
approved_unpublished=Count(
"id", filter=Q(status="approved", published_event__isnull=True)
),
+ # Terminal-ish: `publish_all_approved` couldn't resolve the row's
+ # town to a real `Town` (see ingestion/services.py). Was folded
+ # into `approved` before ticket 36.1 introduced this status, which
+ # silently hid these rows forever behind an `approved` count that
+ # never moved. Disjoint from every other bucket here, same as
+ # `approved_unpublished`.
+ skipped_no_town=Count("id", filter=Q(status="skipped_no_town")),
)
)
funnel_staged_by_source = {row["raw_event__source_id"]: row for row in funnel_staged_qs}
@@ -462,6 +488,8 @@ def _source_rows(db, start, end, source_type_filter, runs_state=None):
if raw_in_window == 0
else None
)
+ no_town_count = funnel_staged.get("skipped_no_town", 0)
+ no_town_note = _no_town_note(no_town_count)
last_run = (
{
"status": recent_runs[0]["status"],
@@ -491,13 +519,21 @@ def _source_rows(db, start, end, source_type_filter, runs_state=None):
# "0", turning "8 muted zeros" into "1 stale source + 2 real
# never-ingested sources" at a glance.
"raw_zero_note": raw_zero_note,
+ # Only set when `no_town > 0` — see `_no_town_note`. Renders as a
+ # tooltip/subtext on the `no_town` funnel cell so a source stuck
+ # at `raw > 0, published = 0` because its town is out of coverage
+ # reads as "explained", not "broken".
+ "no_town_note": no_town_note,
"staged_by_status": staged_status_counts,
"published_count": funnel_staged.get("published", 0),
# Buckets are mutually exclusive — each raw event lands in exactly
# one — so they sum to `raw`. `published` is every row carrying a
# live Event (whether or not the sweep has flipped it to the
- # terminal `published` status), and `approved` is the residual:
- # approved but not yet published. See the annotations above.
+ # terminal `published` status), `approved` is the residual:
+ # approved but not yet published, and `no_town` is the terminal-ish
+ # `skipped_no_town` status (ticket 36.1) — a row whose town isn't
+ # in coverage, never re-attempted until `reopen_skipped_towns`
+ # runs. See the annotations above.
"funnel": {
"raw": raw_in_window,
"unprocessed": unprocessed_counts.get(source_id, 0),
@@ -507,6 +543,7 @@ def _source_rows(db, start, end, source_type_filter, runs_state=None):
"held_for_review": funnel_staged.get("held_for_review", 0),
"rejected": staged_status_counts["rejected"],
"approved": funnel_staged.get("approved_unpublished", 0),
+ "no_town": no_town_count,
"published": funnel_staged.get("published", 0),
},
"last_run": last_run,
diff --git a/backendServer/devtools/templates/devtools/monitor.html b/backendServer/devtools/templates/devtools/monitor.html
index 968804f..c84bcc1 100644
--- a/backendServer/devtools/templates/devtools/monitor.html
+++ b/backendServer/devtools/templates/devtools/monitor.html
@@ -247,6 +247,7 @@
{% endfor %}
diff --git a/backendServer/devtools/tests/test_monitoring_db.py b/backendServer/devtools/tests/test_monitoring_db.py
index e2f9fec..8802219 100644
--- a/backendServer/devtools/tests/test_monitoring_db.py
+++ b/backendServer/devtools/tests/test_monitoring_db.py
@@ -7,6 +7,7 @@
from broadcast.models import BroadcastSubmission, BroadcastTarget
from devtools.monitoring import (
+ _STAGED_STATUSES,
RUNS_AVAILABLE,
RUNS_DB_NOT_CONFIGURED,
RUNS_MISSING_TABLE,
@@ -191,6 +192,21 @@ def setUp(self):
submission=self.submission, site_key="site-two", status="failed"
)
+ def _expected_staged_by_status(self, **nonzero):
+ """Build a full `staged_by_status` expectation from `_STAGED_STATUSES`.
+
+ `staged_by_status` always has one key per `StagedEvent.STATUS_CHOICES`
+ (see `_STAGED_STATUSES` in monitoring.py), so a dict hardcoding only
+ the statuses that existed when a test was written breaks the moment a
+ new status is added — exactly what happened when ticket 36.1 added
+ `skipped_no_town`. Deriving the full key set here instead means these
+ tests only need to say which counts are nonzero, and still fail loudly
+ if any count (including a newly-added status's) is wrong.
+ """
+ expected = dict.fromkeys(_STAGED_STATUSES, 0)
+ expected.update(nonzero)
+ return expected
+
def test_collector_summary_counts_are_exact(self):
rows = {r["name"]: r for r in collector_summary("default", self.start, self.end)}
@@ -201,7 +217,9 @@ def test_collector_summary_counts_are_exact(self):
self.assertEqual(a["raw_count"], 7)
self.assertEqual(
a["staged_by_status"],
- {"pending": 2, "approved": 0, "rejected": 1, "duplicate": 1, "published": 1},
+ self._expected_staged_by_status(
+ pending=2, approved=0, rejected=1, duplicate=1, published=1
+ ),
)
self.assertEqual(a["published_count"], 1)
@@ -209,7 +227,9 @@ def test_collector_summary_counts_are_exact(self):
self.assertEqual(b["raw_count"], 1)
self.assertEqual(
b["staged_by_status"],
- {"pending": 0, "approved": 0, "rejected": 0, "duplicate": 0, "published": 0},
+ self._expected_staged_by_status(
+ pending=0, approved=0, rejected=0, duplicate=0, published=0
+ ),
)
self.assertEqual(b["published_count"], 0)
@@ -228,6 +248,7 @@ def test_collector_summary_funnel_buckets(self):
"held_for_review": 1,
"rejected": 1,
"approved": 0,
+ "no_town": 0,
"published": 1,
},
)
@@ -244,6 +265,7 @@ def test_collector_summary_funnel_buckets(self):
"held_for_review": 0,
"rejected": 0,
"approved": 0,
+ "no_town": 0,
"published": 0,
},
)
@@ -265,24 +287,59 @@ def test_funnel_buckets_reconcile_against_raw(self):
indefinitely, since `auto_publish_safe_events` early-returns when nothing
is pending. This test therefore reads both from `funnel`, not from
`staged_by_status`.
+
+ Sums every bucket *except* `raw` itself, rather than naming each bucket,
+ so a future bucket added to `funnel` without being wired into the sum
+ (as happened when ticket 36.1 added `skipped_no_town` and it fell into
+ no bucket at all — see `test_skipped_no_town_bucket_reconciles`) fails
+ this test instead of silently passing.
"""
rows = collector_summary("default", self.start, self.end) + broadcast_inbound_summary(
"default", self.start, self.end
)
for row in rows:
funnel = row["funnel"]
- accounted = (
- funnel["unprocessed"]
- + funnel["no_staged"]
- + funnel["duplicate"]
- + funnel["unscored"]
- + funnel["held_for_review"]
- + funnel["rejected"]
- + funnel["approved"]
- + funnel["published"]
- )
+ accounted = sum(v for k, v in funnel.items() if k != "raw")
self.assertEqual(accounted, funnel["raw"], row["name"])
+ def test_skipped_no_town_bucket_reconciles(self):
+ """A source with a `skipped_no_town` row still sums to `raw`.
+
+ Ticket 36.1 added the terminal-ish `skipped_no_town` status for staged
+ events whose town has no matching `Town` row (previously these stayed
+ `approved` forever and were re-logged on every pipeline run). The
+ `funnel` dict has a dedicated `no_town` bucket for it — without one,
+ these rows fall through into no bucket at all and the funnel silently
+ stops summing to `raw`, which is exactly what happened before this fix.
+ """
+ raw = RawEvent.objects.create(
+ source=self.collector_a,
+ raw_title="No Matching Town",
+ raw_start=self.start + timedelta(hours=1),
+ processed=True,
+ )
+ StagedEvent.objects.create(
+ raw_event=raw,
+ title="No Matching Town",
+ description="d",
+ location_name="l",
+ town="some-uncovered-town",
+ start_datetime=self.start + timedelta(hours=1),
+ status="skipped_no_town",
+ )
+
+ row = {r["name"]: r for r in collector_summary("default", self.start, self.end)}[
+ "Collector A"
+ ]
+ funnel = row["funnel"]
+ self.assertEqual(funnel["no_town"], 1)
+ accounted = sum(v for k, v in funnel.items() if k != "raw")
+ self.assertEqual(accounted, funnel["raw"])
+ self.assertEqual(
+ row["no_town_note"],
+ "1 skipped — town not in coverage (see `manage.py reopen_skipped_towns` once added)",
+ )
+
def test_published_bucket_counts_approved_rows_with_a_live_event(self):
"""A row that has an Event but hasn't been swept yet still reads as
published — the `ingest_direct_submission` state, and the reason the
@@ -327,7 +384,7 @@ def test_direct_source_only_in_inbound_summary(self):
self.assertEqual(row["raw_count"], 1)
self.assertEqual(
row["staged_by_status"],
- {"pending": 0, "approved": 0, "rejected": 0, "duplicate": 1, "published": 0},
+ self._expected_staged_by_status(duplicate=1),
)
self.assertEqual(row["published_count"], 0)
diff --git a/backendServer/events/management/commands/healthcheck.py b/backendServer/events/management/commands/healthcheck.py
index a94bc06..881a6ea 100644
--- a/backendServer/events/management/commands/healthcheck.py
+++ b/backendServer/events/management/commands/healthcheck.py
@@ -11,10 +11,36 @@
every django-celery-beat PeriodicTask (enabled + last-run freshness). Exits
non-zero if any *critical* check fails so it can feed monitoring — staleness is
one of those: a schedule that stopped firing IS an outage, not a warning.
+
+Freshness for a crontab-backed task is judged by deriving the *expected* next
+fire time from the crontab itself (via `crontab.remaining_estimate`), not by a
+hand-tuned staleness window — a window necessarily exceeds the task's own
+period, so it always has a blind spot at least one period wide (a missed
+weekly send would ride along for ~8 days). Instead we ask the schedule "what
+was the next fire time after last_run_at?" and FAIL as soon as that time has
+passed (past a grace allowance — see crontab_grace_seconds() below), so a
+missed occurrence surfaces within hours, not days.
+DEFAULT_STALENESS_HOURS is kept for two narrower roles: (1) the must-exist set
+— a seeded task missing from the schedule entirely is a FAIL — and (2) a
+fallback window for interval-backed or schedule-less PeriodicTask rows, where
+there is no crontab to derive an expected fire time from.
+
+`PeriodicTask.last_run_at` is not written the instant a task fires: beat's
+`DatabaseScheduler` keeps it in memory and only persists it on a sync. That
+sync is now forced after every single task send by `CELERY_BEAT_SYNC_EVERY = 1`
+(see backend/settings/base.py), so last_run_at lags reality by roughly one
+task execution, not by `CELERY_BEAT_MAX_LOOP_INTERVAL` (hours). If
+`CELERY_BEAT_SYNC_EVERY` is ever unset/disabled, the old worst case comes
+back, so `crontab_grace_seconds()` re-widens automatically rather than
+silently staying tight. A naive "is remaining negative" check would misreport
+a healthy-but-unsynced task as MISSED, so `crontab_grace_seconds()` absorbs
+that plausible persistence lag before declaring MISSED; see
+`_crontab_freshness` for the WARN/FAIL split within that grace window.
"""
import json
import os
+from datetime import timedelta
from django.conf import settings
from django.core.cache import cache
@@ -28,20 +54,78 @@
# only these, the app is on dev settings and will 400 every real request.
LOCALHOST_ONLY = {"localhost", "127.0.0.1", "[::1]"}
-# Per-task freshness windows (hours). A seeded task whose last_run_at is older
-# than this — or that has never run — is flagged FAIL. This dict also doubles
-# as the set of tasks that must exist in the schedule at all (see
-# _check_periodic_tasks): every task the pipeline depends on belongs here, or
-# its staleness silently passes forever (2026-07-21 scheduler outage).
+# Tasks that must exist in the schedule (a missing entry is a FAIL — see
+# _check_periodic_tasks, guarding the 2026-07-21 scheduler outage), and the
+# fallback staleness window (hours) used only when a task has no crontab to
+# derive an expected-fire time from (interval schedule, or no schedule at
+# all). Crontab-backed tasks are judged against their own expected fire time
+# instead — see _task_freshness.
DEFAULT_STALENESS_HOURS = {
"ingest-events-daily": 25,
"scrape-sources-daily": 25,
"weekly-digest-sunday": 24 * 8, # ~8 days
- # Runs `0 */6 * * *`; one interval plus an hour of grace, matching the
- # +1h pattern the daily windows use.
"broadcast-orphan-recovery": 7,
}
+# Fallback bound (seconds) on how long beat's DatabaseScheduler may leave a
+# fresh last_run_at unpersisted, used only if CELERY_BEAT_MAX_LOOP_INTERVAL is
+# unset — matches the value that setting currently carries in
+# backend/settings/base.py, so an unset value degrades to today's known prod
+# behavior rather than to an arbitrary guess.
+_DEFAULT_BEAT_LOOP_INTERVAL_SECONDS = 6 * 60 * 60
+
+# Extra buffer (seconds) on top of the loop-interval bound, covering scheduler
+# jitter and task execution time so we don't FAIL on the tail end of a sync
+# that lands a few minutes late. Deliberately small relative to the loop
+# interval — this is slack for noise, not a second staleness window. Used
+# only in the *wide* (CELERY_BEAT_SYNC_EVERY unset/0) branch below.
+CRONTAB_GRACE_BUFFER_SECONDS = 60 * 60
+
+# Buffer (seconds) used in the *tight* branch, when CELERY_BEAT_SYNC_EVERY
+# forces a DB flush after every task send. The only slack left to absorb is
+# ordinary scheduler wake-up jitter plus the time a task itself takes to run
+# before the post-send sync fires — not a multi-hour polling interval. 5
+# minutes comfortably covers that for these tasks (digest/ingestion/orphan
+# recovery all complete in seconds-to-low-minutes) without reopening a
+# window wide enough to swallow a genuine miss.
+CRONTAB_GRACE_TIGHT_BUFFER_SECONDS = 5 * 60
+
+
+def crontab_grace_seconds() -> int:
+ """Grace (seconds) allowed between an expected crontab fire time and when
+ we declare a task MISSED, absorbing beat's last_run_at persistence lag.
+
+ Evaluated at call time (not a module-level constant) so tests can vary it
+ via @override_settings. When CELERY_BEAT_SYNC_EVERY is a positive int,
+ beat flushes last_run_at after every task send (see module docstring), so
+ the lag collapses to roughly one task execution and only the small jitter
+ buffer is needed. If someone later unsets/disables that setting, the lag
+ bound reverts to CELERY_BEAT_MAX_LOOP_INTERVAL and the grace automatically
+ re-widens to match — that self-protection is the point of deriving this
+ from the setting instead of hardcoding a small number.
+ """
+ sync_every = getattr(settings, "CELERY_BEAT_SYNC_EVERY", 0) or 0
+ if isinstance(sync_every, int) and sync_every > 0:
+ return CRONTAB_GRACE_TIGHT_BUFFER_SECONDS
+ loop_interval = getattr(
+ settings, "CELERY_BEAT_MAX_LOOP_INTERVAL", _DEFAULT_BEAT_LOOP_INTERVAL_SECONDS
+ )
+ return loop_interval + CRONTAB_GRACE_BUFFER_SECONDS
+
+
+def _format_timedelta(delta) -> str:
+ """Render a positive timedelta as e.g. '3d13h07m' for detail strings."""
+ total_minutes = int(delta.total_seconds() // 60)
+ days, rem_minutes = divmod(total_minutes, 24 * 60)
+ hours, minutes = divmod(rem_minutes, 60)
+ parts = []
+ if days:
+ parts.append(f"{days}d")
+ if days or hours:
+ parts.append(f"{hours}h")
+ parts.append(f"{minutes:02d}m")
+ return "".join(parts)
+
class Command(BaseCommand):
help = (
@@ -182,6 +266,76 @@ def _check_periodic_tasks(self) -> list[tuple[str, str, str]]:
return out
def _task_freshness(self, task, label: str, now) -> tuple[str, str, str]:
+ if task.last_run_at is None:
+ return (FAIL, label, "enabled, never run yet")
+
+ crontab = getattr(task, "crontab", None)
+ if crontab is not None:
+ return self._crontab_freshness(task, label, now, crontab)
+ return self._window_freshness(task, label, now)
+
+ def _crontab_freshness(self, task, label: str, now, crontab) -> tuple[str, str, str]:
+ # Derive the *expected* next fire after last_run_at from the crontab
+ # itself, instead of a hand-tuned window — a window always exceeds the
+ # task's own period, so it has a blind spot at least one period wide
+ # (a missed weekly send would ride along for ~8 days). Asking the
+ # schedule directly means a missed occurrence surfaces within hours.
+ schedule = crontab.schedule
+ # TzAwareCrontab.remaining_estimate() (inherited unmodified from
+ # celery.schedules.crontab) compares last_run_at.hour/.minute/
+ # .isoweekday() directly against the crontab's fields — it never
+ # converts its inputs into schedule.tz first. In real beat operation
+ # that's masked: TzAwareCrontab.is_due() explicitly does
+ # `last_run_at.astimezone(self.tz)` before delegating, and its
+ # nowfunc() already returns `datetime.now(self.tz)`, so both operands
+ # land in local time. Here we call remaining_estimate() directly and
+ # pin nowfun to a UTC `now` (from timezone.now()), so without an
+ # explicit conversion both operands would be read in UTC — silently
+ # treating e.g. "0 4 * * *"/America/New_York as "04:00 UTC" and
+ # putting expected_fire ~4-5h off (the zone offset itself, not just
+ # DST drift) for every America/New_York crontab, every day. Convert
+ # both operands into the schedule's own tz to match what is_due()
+ # does, so remaining_estimate() sees the same local wall-clock fields
+ # a real beat process would.
+ tzinfo = schedule.tz
+ local_now = now.astimezone(tzinfo)
+ local_last_run_at = task.last_run_at.astimezone(tzinfo)
+ schedule.nowfun = lambda: local_now
+ remaining = schedule.remaining_estimate(local_last_run_at)
+ expected_fire = now + remaining
+
+ if remaining.total_seconds() < 0:
+ overdue = -remaining
+ # last_run_at can lag real fire time by up to
+ # crontab_grace_seconds() (beat's DatabaseScheduler defers the
+ # Postgres write — see module docstring), so a task that is
+ # merely "overdue" is not yet distinguishable from one that fired
+ # on time and just hasn't persisted. Only escalate to FAIL once
+ # overdue exceeds that plausible lag; a schedule that stopped
+ # firing is an outage (see docstring), so once past grace this
+ # must not be softened.
+ grace_seconds = crontab_grace_seconds()
+ if overdue.total_seconds() <= grace_seconds:
+ grace = _format_timedelta(timedelta(seconds=grace_seconds))
+ return (
+ WARN,
+ label,
+ f"overdue by {_format_timedelta(overdue)} (expected fire at "
+ f"{expected_fire.isoformat()}) — within {grace} grace for beat's "
+ "last_run_at persistence lag; cannot yet distinguish from a "
+ "genuine miss",
+ )
+ return (
+ FAIL,
+ label,
+ f"MISSED — expected fire at {expected_fire.isoformat()}, "
+ f"overdue by {_format_timedelta(overdue)}",
+ )
+ return (OK, label, f"next expected fire {expected_fire.isoformat()}")
+
+ def _window_freshness(self, task, label: str, now) -> tuple[str, str, str]:
+ # No crontab to derive an expected fire time from (interval schedule,
+ # or no schedule at all) — fall back to the hand-tuned window.
window = DEFAULT_STALENESS_HOURS.get(task.name)
if window is None:
# No configured window means we can't judge freshness — that is a
@@ -189,8 +343,6 @@ def _task_freshness(self, task, label: str, now) -> tuple[str, str, str]:
# it so a newly-seeded task never rides along as a silent OK/WARN
# until someone remembers to add it above (see scrape-sources-daily).
return (WARN, label, "no staleness window configured — add to DEFAULT_STALENESS_HOURS")
- if task.last_run_at is None:
- return (FAIL, label, "enabled, never run yet")
age_h = (now - task.last_run_at).total_seconds() / 3600
stamp = f"last run {age_h:.1f}h ago"
if age_h > window:
diff --git a/backendServer/events/migrations/0017_backfill_null_town_events.py b/backendServer/events/migrations/0017_backfill_null_town_events.py
new file mode 100644
index 0000000..7ac70cf
--- /dev/null
+++ b/backendServer/events/migrations/0017_backfill_null_town_events.py
@@ -0,0 +1,94 @@
+"""Backfill `town` for the 15 live events (ticket 36.2) that shipped with
+`town = NULL`.
+
+All 15 rows have an empty `source_name`, which points at hand-entered events
+that predate the ingestion pipeline, not at a pipeline classification miss.
+
+## Was 0004 the cause?
+
+Ticket 36.2 hypothesised that `0004_add_town_model.migrate_town_strings_to_fk`
+(which only assigned a `Town` FK when the old `town_str` matched an
+already-seeded slug, and 0004 seeded only `carrboro`/`pittsboro`) is why these
+rows are NULL. Reading 0004 confirms the mechanism is real — any `town_str`
+outside `{carrboro, pittsboro}` would silently become NULL FK — but it does
+not explain *these* 15 rows specifically: 0004 ran once, at migration time,
+against whatever events existed then. These 15 rows carry an empty
+`source_name`, meaning they were entered by hand (through the site's own
+create-event flow or admin), not carried over from a pre-Town-model
+`town_str`. The 0004 code path is a real bug class (confirmed by reading the
+code), but it is not the origin of this particular batch — the more likely
+explanation is that these were hand-created with no `town` selected, or with
+a value that didn't resolve to an existing `Town`.
+
+## Venue -> town mapping (see ticket for full source list)
+
+- "Fair Game Beverage Company" -> pittsboro (192 Lorax Lane, Pittsboro, NC;
+ Chatham Beverage District, on the eastern edge of Pittsboro).
+- "bmc brewing" -> pittsboro (213 Lorax Ln, Pittsboro, NC; same Beverage
+ District complex as Fair Game).
+- "Chatham YMCA" -> pittsboro (120 Parkland Drive, Pittsboro, NC; branded
+ "Chatham Park YMCA").
+- "Covenant Place" (11 rows, the recurring "Senior ..." series) -> carrboro.
+ The building's postal address is 103 Culbreth Rd, Chapel Hill, NC 27516,
+ but the events themselves are Carrboro Parks & Recreation programming,
+ calendared directly on the Town of Carrboro's own site
+ (townofcarrboro.org/Calendar.aspx, carrbororec.org) — i.e. Carrboro Rec runs
+ this senior series at Covenant Place. That is the operative "town" for a
+ reader browsing what's on in Carrboro, so carrboro (not chapel-hill, which
+ is also a covered Town slug) is the unambiguous choice here.
+- "Jordan Lake State Recreation Area – Ne..." (the spring cleanup) -> left
+ NULL. Jordan Lake State Recreation Area spans Wake and Chatham counties
+ with access points on both the eastern shore (near Apex) and western shore
+ (near Pittsboro/New Hope). No single covered Town slug is correct for the
+ park as a whole, and the specific access point for this cleanup isn't
+ resolvable from the data on hand. A wrong town would misplace this event on
+ the wrong hub page, so it stays NULL per ticket guidance.
+
+Reversible: `unbackfill` restores every row this migration touches to NULL
+(scoped by uuid, not by venue/town, so it can't clobber unrelated events).
+"""
+from django.db import migrations
+
+# (venue, town_slug) for every row this migration assigns a town to.
+VENUE_TOWN_MAP = {
+ "Fair Game Beverage Company": "pittsboro",
+ "bmc brewing": "pittsboro",
+ "Chatham YMCA": "pittsboro",
+ "Covenant Place": "carrboro",
+}
+
+# Venues intentionally left NULL, and why — for anyone auditing later.
+LEFT_NULL = {
+ "Jordan Lake State Recreation Area – New Hope Overlook Access": (
+ "Spans Wake/Chatham counties with access points in multiple towns; "
+ "no single covered Town slug is correct."
+ ),
+}
+
+
+def backfill_towns(apps, schema_editor):
+ Event = apps.get_model("events", "Event")
+ Town = apps.get_model("events", "Town")
+
+ for venue, slug in VENUE_TOWN_MAP.items():
+ town = Town.objects.filter(slug=slug).first()
+ if town is None:
+ # Covered-town list has drifted; don't guess, don't crash the migration.
+ continue
+ Event.objects.filter(venue=venue, town__isnull=True, source_name="").update(town=town)
+
+
+def unbackfill_towns(apps, schema_editor):
+ Event = apps.get_model("events", "Event")
+ for venue in VENUE_TOWN_MAP:
+ Event.objects.filter(venue=venue, source_name="").update(town=None)
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ("events", "0016_seed_chatham_towns"),
+ ]
+
+ operations = [
+ migrations.RunPython(backfill_towns, unbackfill_towns),
+ ]
diff --git a/backendServer/events/tests/test_beat_last_run_persistence.py b/backendServer/events/tests/test_beat_last_run_persistence.py
new file mode 100644
index 0000000..b8556ac
--- /dev/null
+++ b/backendServer/events/tests/test_beat_last_run_persistence.py
@@ -0,0 +1,177 @@
+"""Regression test for ticket 36.4.
+
+Root cause (confirmed on prod, see docs/ingestion-monitoring.md): django_celery_beat's
+DatabaseScheduler keeps PeriodicTask.last_run_at / total_run_count in memory and only
+writes them to Postgres on a sync. With CELERY_BEAT_SYNC_EVERY unset (default 0),
+Scheduler.should_sync() only has the 3-minute time-based trigger left
+(celery/beat.py Scheduler.sync_every = 180), so a task that fires again within that
+window of the previous sync leaves its just-made last_run_at update sitting in memory
+— for up to CELERY_BEAT_MAX_LOOP_INTERVAL (6h here) until beat's next sync opportunity,
+or until process shutdown flushes it via `finally: self.sync()`.
+
+This test drives a real django_celery_beat.schedulers.DatabaseScheduler end to end:
+seed a PeriodicTask + CrontabSchedule, build a scheduler whose in-memory schedule
+contains only that one entry (bypassing DatabaseScheduler.setup_schedule()'s full-table
+scan, which would otherwise also pull in every other seeded PeriodicTask row in this
+shared test DB), send it through Scheduler.apply_async with the underlying task stubbed
+so nothing touches Redis, then assert the DB row itself (a fresh .get(), not the
+in-memory entry) advanced last_run_at/total_run_count — with no shutdown/close call.
+Without CELERY_BEAT_SYNC_EVERY = 1, should_sync() is False on a steady-state send and
+the DB row is untouched; with it, sync_every_tasks=1 forces should_sync() True after
+one send and the row is updated immediately.
+"""
+
+import time
+from datetime import timedelta
+from unittest.mock import patch
+
+from celery import current_app
+from django.test import TestCase, tag
+from django.utils import timezone
+from django_celery_beat.models import CrontabSchedule, PeriodicTask
+from django_celery_beat.schedulers import DatabaseScheduler, ModelEntry
+
+TASK_NAME = "test-beat-last-run-persistence"
+
+
+@tag("db")
+class BeatLastRunPersistenceTests(TestCase):
+ """Drives a real DatabaseScheduler.apply_async() and inspects the DB row."""
+
+ def setUp(self):
+ self.crontab = CrontabSchedule.objects.create(
+ minute="*",
+ hour="*",
+ day_of_week="*",
+ day_of_month="*",
+ month_of_year="*",
+ timezone="UTC",
+ )
+ # last_run_at far in the past — not load-bearing for these assertions (we
+ # drive apply_async()/reserve() directly rather than relying on is_due()),
+ # but keeps the row realistic.
+ self.periodic_task = PeriodicTask.objects.create(
+ name=TASK_NAME,
+ task="events.tasks.fan_out_weekly_digest",
+ crontab=self.crontab,
+ enabled=True,
+ last_run_at=timezone.now() - timedelta(days=1),
+ total_run_count=0,
+ )
+
+ def _build_scheduler_with_single_entry(self):
+ """A DatabaseScheduler whose in-memory schedule holds only our one
+ PeriodicTask — avoids all_as_schedule()'s full-table scan over every
+ other PeriodicTask seeded by migrations in this shared test DB. Sets
+ scheduler.data/_schedule directly rather than going through the
+ `schedule` property getter, which would otherwise trigger exactly that
+ full re-sync (DatabaseScheduler.sync() writes dirty entries out of
+ self._schedule specifically, so both must point at the same dict)."""
+ scheduler = DatabaseScheduler(current_app, lazy=True)
+ entry = ModelEntry(self.periodic_task, app=current_app)
+ scheduler.data = {TASK_NAME: entry}
+ scheduler._schedule = scheduler.data
+ return scheduler
+
+ def _send_one(self, scheduler):
+ """Send the one entry through the real apply_async(), with the task
+ lookup and actual dispatch stubbed so nothing touches Redis/the broker.
+
+ DatabaseScheduler.sync() opens with close_old_connections() (see
+ django_celery_beat/schedulers.py), which is harmless for a real beat
+ process but fatal under Django's TestCase: that wraps the test in an
+ atomic block on a single connection, and closing it mid-transaction
+ breaks every later ORM call with "the connection is closed". Patched
+ to a no-op here so the test can still run inside TestCase's atomic
+ wrapper; this doesn't touch what's under test (should_sync()/sync()'s
+ own dirty-entry save logic), only the connection-recycling housekeeping
+ that a real long-running beat process needs and a single test
+ transaction does not.
+ """
+ entry = scheduler.data[TASK_NAME]
+ with patch("django_celery_beat.schedulers.close_old_connections"):
+ with patch.object(scheduler, "send_task") as mock_send_task:
+ with patch.object(current_app.tasks, "get", return_value=None):
+ scheduler.apply_async(entry, advance=True)
+ mock_send_task.assert_called_once()
+
+ def test_last_run_at_not_persisted_without_sync(self):
+ """Pin the failure mode itself: should_sync() is False after a send that
+ isn't the scheduler's first (i.e. reproduces the pre-fix config, where
+ sync_every_tasks is unset and only the 180s time-based trigger remains)."""
+ scheduler = self._build_scheduler_with_single_entry()
+ scheduler.sync_every_tasks = None # pre-fix: CELERY_BEAT_SYNC_EVERY unset
+
+ # Simulate "not the first send since beat started" — _last_sync already
+ # set and recent — which is the steady-state a long-running beat process
+ # is in almost all the time.
+ scheduler._last_sync = time.monotonic()
+ scheduler._tasks_since_sync = 0
+
+ self._send_one(scheduler)
+
+ self.assertFalse(
+ scheduler.should_sync(),
+ "should_sync() must be False here to reproduce the bug: with "
+ "sync_every_tasks unset, a task fired well within the 180s "
+ "time-based sync window has no other trigger to flush "
+ "last_run_at to Postgres.",
+ )
+
+ # And, correspondingly, the DB row itself must NOT have advanced.
+ unchanged = PeriodicTask.objects.get(name=TASK_NAME)
+ self.assertEqual(
+ unchanged.total_run_count,
+ 0,
+ "Without a sync, last_run_at/total_run_count must still be sitting "
+ "in memory only — this is exactly the prod lag ticket 36.4 fixes.",
+ )
+
+ def test_last_run_at_persisted_immediately_with_current_settings(self):
+ """With the actual project settings (CELERY_BEAT_SYNC_EVERY = 1), a single
+ apply_async() call must flush last_run_at/total_run_count to Postgres
+ without any explicit sync()/close()/shutdown."""
+ self.assertEqual(
+ current_app.conf.beat_sync_every,
+ 1,
+ "This test asserts the fix is live; if beat_sync_every isn't 1, "
+ "CELERY_BEAT_SYNC_EVERY regressed in settings.",
+ )
+
+ scheduler = self._build_scheduler_with_single_entry()
+ # sync_every_tasks is read from app.conf.beat_sync_every at __init__ time
+ # (celery/beat.py Scheduler.__init__), so it's already 1 here — assert it
+ # explicitly so a future config regression fails loudly at this line.
+ self.assertEqual(scheduler.sync_every_tasks, 1)
+
+ # Same steady-state simulation as the failing test above, so this test
+ # differs from it only in sync_every_tasks — isolating the fix.
+ scheduler._last_sync = time.monotonic()
+ scheduler._tasks_since_sync = 0
+
+ before = PeriodicTask.objects.get(name=TASK_NAME)
+ self.assertEqual(before.total_run_count, 0)
+
+ self._send_one(scheduler)
+
+ # apply_async()'s own `finally` clause already called _do_sync() (since
+ # should_sync() was True mid-send) and _do_sync() resets the counter —
+ # so _tasks_since_sync == 0 here is itself evidence a sync just ran,
+ # not evidence that none is needed. The DB assertion below is the real
+ # proof: the row must reflect the send with no explicit sync() call
+ # from this test.
+ self.assertEqual(
+ scheduler._tasks_since_sync,
+ 0,
+ "sync_every_tasks=1 should have forced should_sync() True and "
+ "reset the counter via _do_sync() within apply_async() itself.",
+ )
+
+ after = PeriodicTask.objects.get(name=TASK_NAME)
+ self.assertEqual(
+ after.total_run_count,
+ 1,
+ "last_run_at/total_run_count should be persisted to Postgres "
+ "right after the send, with no explicit sync()/shutdown call.",
+ )
+ self.assertGreater(after.last_run_at, before.last_run_at)
diff --git a/backendServer/events/tests/test_healthcheck_freshness.py b/backendServer/events/tests/test_healthcheck_freshness.py
index 54cfffc..a195306 100644
--- a/backendServer/events/tests/test_healthcheck_freshness.py
+++ b/backendServer/events/tests/test_healthcheck_freshness.py
@@ -10,13 +10,23 @@
age — it reports WARN with an explicit "add it" message, not OK.
3. Staleness for a *configured* task is now FAIL, not WARN, so a dead schedule
trips deploy/healthcheck.sh's non-zero exit.
+
+Ticket 36.3: freshness for a crontab-backed task is no longer judged against a
+hand-tuned window (which always has a blind spot at least one period wide —
+e.g. a missed Sunday digest wouldn't FAIL for ~8 days). Instead we derive the
+*expected* next fire time from the task's own crontab
+(`crontab.remaining_estimate`) and FAIL as soon as that time has passed.
+DEFAULT_STALENESS_HOURS keeps its must-exist role (dict keys = tasks that must
+be seeded) and now serves only as a fallback window for interval-backed or
+schedule-less tasks, which have no crontab to derive an expected fire time
+from.
"""
import unittest
from datetime import timedelta
from types import SimpleNamespace
-from django.test import TestCase, tag
+from django.test import TestCase, override_settings, tag
from django.utils import timezone
from django_celery_beat.models import CrontabSchedule, PeriodicTask
@@ -26,11 +36,25 @@
OK,
WARN,
Command,
+ crontab_grace_seconds,
)
-def _fake_task(name: str, last_run_at):
- return SimpleNamespace(name=name, last_run_at=last_run_at)
+def _fake_task(name: str, last_run_at, crontab=None):
+ return SimpleNamespace(name=name, last_run_at=last_run_at, crontab=crontab)
+
+
+def _fake_crontab(minute="0", hour="*", day_of_week="*", tz="UTC"):
+ # An unsaved CrontabSchedule — .schedule is a pure property, so this never
+ # touches the DB. Safe to use in the no-DB fast tier.
+ return CrontabSchedule(
+ minute=minute,
+ hour=hour,
+ day_of_week=day_of_week,
+ day_of_month="*",
+ month_of_year="*",
+ timezone=tz,
+ )
@tag("fast")
@@ -46,26 +70,11 @@ def test_scrape_sources_daily_has_a_configured_window(self):
self.assertIn("scrape-sources-daily", DEFAULT_STALENESS_HOURS)
self.assertEqual(DEFAULT_STALENESS_HOURS["scrape-sources-daily"], 25)
- def test_broadcast_orphan_recovery_has_a_configured_window(self):
- # Found by the first real healthcheck run on prod (2026-07-29): the task
- # is seeded by broadcast/0009 but had no window, so it reported WARN
- # "no staleness window configured" — the exact ride-along case the
- # DEFAULT_STALENESS_HOURS comment warns about. Schedule is `0 */6 * * *`.
- self.assertEqual(DEFAULT_STALENESS_HOURS["broadcast-orphan-recovery"], 7)
-
- def test_orphan_recovery_one_missed_interval_still_ok(self):
- # 6h schedule + 1h grace: a run that merely landed late must not FAIL.
- task = _fake_task("broadcast-orphan-recovery", self.now - timedelta(hours=6, minutes=30))
- status, _, _ = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", self.now)
- self.assertEqual(status, OK)
-
- def test_orphan_recovery_two_missed_intervals_fails(self):
- task = _fake_task("broadcast-orphan-recovery", self.now - timedelta(hours=13))
- status, _, detail = self.cmd._task_freshness(
- task, "beat:broadcast-orphan-recovery", self.now
- )
- self.assertEqual(status, FAIL)
- self.assertIn("STALE", detail)
+ def test_broadcast_orphan_recovery_still_in_must_exist_set(self):
+ # DEFAULT_STALENESS_HOURS now only needs to carry this key for the
+ # must-exist guard (see _check_periodic_tasks) — crontab-backed
+ # freshness no longer reads the associated hour value at all.
+ self.assertIn("broadcast-orphan-recovery", DEFAULT_STALENESS_HOURS)
def test_configured_task_never_run_fails(self):
task = _fake_task("scrape-sources-daily", None)
@@ -73,20 +82,10 @@ def test_configured_task_never_run_fails(self):
self.assertEqual(status, FAIL)
self.assertIn("never run", detail)
- def test_configured_task_within_window_is_ok(self):
- task = _fake_task("ingest-events-daily", self.now - timedelta(hours=5))
- status, _, _ = self.cmd._task_freshness(task, "beat:ingest-events-daily", self.now)
- self.assertEqual(status, OK)
-
- def test_configured_task_stale_fails_not_warns(self):
- task = _fake_task("ingest-events-daily", self.now - timedelta(hours=30))
- status, _, detail = self.cmd._task_freshness(task, "beat:ingest-events-daily", self.now)
- self.assertEqual(status, FAIL)
- self.assertIn("STALE", detail)
-
def test_unlisted_task_does_not_silently_pass(self):
- # A task with no entry in DEFAULT_STALENESS_HOURS at all — e.g. the
- # next task someone adds to the schedule and forgets to configure.
+ # A task with no entry in DEFAULT_STALENESS_HOURS at all, and no
+ # crontab — e.g. the next task someone adds to the schedule and
+ # forgets to configure.
task = _fake_task("some-new-task-nobody-configured", self.now - timedelta(days=400))
status, _, detail = self.cmd._task_freshness(task, "beat:some-new-task", self.now)
self.assertNotEqual(status, OK)
@@ -99,6 +98,411 @@ def test_unlisted_task_recently_run_also_does_not_pass(self):
status, _, _ = self.cmd._task_freshness(task, "beat:some-new-task", self.now)
self.assertNotEqual(status, OK)
+ def test_interval_style_task_falls_back_to_window(self):
+ # No crontab attached (interval-backed, or schedule-less) — the
+ # hand-tuned window is still the right fallback since there is no
+ # crontab to derive an expected fire time from.
+ task = _fake_task("ingest-events-daily", self.now - timedelta(hours=5), crontab=None)
+ status, _, detail = self.cmd._task_freshness(task, "beat:ingest-events-daily", self.now)
+ self.assertEqual(status, OK)
+ self.assertIn("last run", detail)
+
+ def test_interval_style_task_stale_fails_not_warns(self):
+ task = _fake_task("ingest-events-daily", self.now - timedelta(hours=30), crontab=None)
+ status, _, detail = self.cmd._task_freshness(task, "beat:ingest-events-daily", self.now)
+ self.assertEqual(status, FAIL)
+ self.assertIn("STALE", detail)
+
+ def test_crontab_task_on_schedule_is_ok(self):
+ # `0 */6 * * *` UTC, last run just after 12:00Z, now 18:00Z + 1s —
+ # the 18:00 fire hasn't come due relative to `now` yet... use a `now`
+ # still inside the 12:00-18:00 window so no fire has been missed.
+ crontab = _fake_crontab(minute="0", hour="*/6", tz="UTC")
+ last_run_at = self.now.replace(hour=12, minute=0, second=1, microsecond=0)
+ now = last_run_at.replace(hour=17, minute=0, second=0)
+ task = _fake_task("broadcast-orphan-recovery", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", now)
+ self.assertEqual(status, OK)
+ self.assertIn("next expected fire", detail)
+
+ @override_settings(CELERY_BEAT_SYNC_EVERY=1)
+ def test_crontab_task_overdue_within_tight_grace_warns(self):
+ # `0 */6 * * *` UTC, last_run_at 12:00:01Z, now 18:04:00Z -> expected
+ # fire 18:00:00Z, overdue ~4m. CELERY_BEAT_SYNC_EVERY=1 forces a
+ # post-send flush, so the tight 5-minute jitter buffer applies (see
+ # crontab_grace_seconds()) — this is plausibly just scheduler/task
+ # jitter, not a real miss. WARN, not a silent OK and not a FAIL we
+ # can't actually justify.
+ crontab = _fake_crontab(minute="0", hour="*/6", tz="UTC")
+ last_run_at = self.now.replace(hour=12, minute=0, second=1, microsecond=0)
+ now = last_run_at.replace(hour=18, minute=4, second=0)
+ task = _fake_task("broadcast-orphan-recovery", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", now)
+ self.assertEqual(status, WARN)
+ self.assertIn("18:00:00", detail)
+ self.assertIn("04m", detail)
+ self.assertIn("grace", detail)
+
+ @override_settings(CELERY_BEAT_SYNC_EVERY=1)
+ def test_crontab_task_overdue_past_tight_grace_fails(self):
+ # Same schedule, pushed past the tight 5-minute grace — e.g. beat has
+ # been down long enough that this is no longer explainable as jitter.
+ # `0 */6 * * *` UTC, last_run_at 12:00:01Z, now 18:07:00Z -> expected
+ # fire 18:00:00Z, overdue ~7m (> 5m tight grace).
+ crontab = _fake_crontab(minute="0", hour="*/6", tz="UTC")
+ last_run_at = self.now.replace(hour=12, minute=0, second=1, microsecond=0)
+ now = last_run_at.replace(hour=18, minute=7, second=0)
+ task = _fake_task("broadcast-orphan-recovery", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", now)
+ self.assertEqual(status, FAIL)
+ self.assertIn("MISSED", detail)
+ self.assertIn("18:00:00", detail)
+
+ @override_settings(CELERY_BEAT_SYNC_EVERY=0)
+ def test_crontab_task_overdue_within_wide_grace_warns_when_sync_every_disabled(self):
+ # If CELERY_BEAT_SYNC_EVERY is disabled (0), crontab_grace_seconds()
+ # must fall back to the wide 7h grace (6h CELERY_BEAT_MAX_LOOP_INTERVAL
+ # + 1h buffer) — the persistence lag this setting guards against comes
+ # back, so the healthcheck must automatically re-widen rather than
+ # keep using the tight number and start missing real WARN cases.
+ # `0 */6 * * *` UTC, last_run_at 12:00:01Z, now 22:07Z -> expected
+ # fire 18:00:00Z, overdue ~4h07m, within the 7h wide grace.
+ crontab = _fake_crontab(minute="0", hour="*/6", tz="UTC")
+ last_run_at = self.now.replace(hour=12, minute=0, second=1, microsecond=0)
+ now = last_run_at.replace(hour=22, minute=7, second=0)
+ task = _fake_task("broadcast-orphan-recovery", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", now)
+ self.assertEqual(status, WARN)
+ self.assertIn("18:00:00", detail)
+ self.assertIn("4h07m", detail)
+ self.assertIn("grace", detail)
+
+ @override_settings(CELERY_BEAT_SYNC_EVERY=0)
+ def test_crontab_task_overdue_past_wide_grace_fails_when_sync_every_disabled(self):
+ # Same disabled-setting fallback, pushed past the 7h wide grace —
+ # e.g. beat has been down long enough that this is no longer
+ # explainable as persistence lag even under the wide bound.
+ # `0 */6 * * *` UTC, last_run_at 12:00:01Z, now 01:30:00Z next day ->
+ # expected fire 18:00:00Z, overdue ~7h30m (> 7h wide grace).
+ crontab = _fake_crontab(minute="0", hour="*/6", tz="UTC")
+ last_run_at = self.now.replace(hour=12, minute=0, second=1, microsecond=0)
+ now = last_run_at.replace(hour=23, minute=0, second=0) + timedelta(hours=2, minutes=30)
+ task = _fake_task("broadcast-orphan-recovery", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", now)
+ self.assertEqual(status, FAIL)
+ self.assertIn("MISSED", detail)
+ self.assertIn("18:00:00", detail)
+
+ @override_settings(CELERY_BEAT_SYNC_EVERY=1)
+ def test_crontab_grace_seconds_tight_when_sync_every_positive(self):
+ self.assertEqual(crontab_grace_seconds(), 5 * 60)
+
+ @override_settings(CELERY_BEAT_SYNC_EVERY=0)
+ def test_crontab_grace_seconds_wide_when_sync_every_disabled(self):
+ self.assertEqual(crontab_grace_seconds(), 6 * 60 * 60 + 60 * 60)
+
+ @override_settings(CELERY_BEAT_SYNC_EVERY=None)
+ def test_crontab_grace_seconds_wide_when_sync_every_none(self):
+ self.assertEqual(crontab_grace_seconds(), 6 * 60 * 60 + 60 * 60)
+
+ def test_do_not_mask_weekly_digest_missed_sunday_fails(self):
+ # "Do not mask" case #1: weekly-digest-sunday, last_run_at 2026-07-20
+ # (a Monday-ish run timestamp per the ground truth table), beat dead
+ # through Sunday 2026-07-26 -> that week's digest genuinely never
+ # went out. Must still FAIL, even though the next scheduled send
+ # (2026-08-02) hasn't arrived yet.
+ crontab = _fake_crontab(minute="0", hour="9", day_of_week="sun", tz="America/New_York")
+ last_run_at = timezone.datetime(
+ 2026, 7, 20, 13, 0, 2, tzinfo=timezone.get_fixed_timezone(0)
+ )
+ now = timezone.datetime(2026, 7, 29, 22, 7, 0, tzinfo=timezone.get_fixed_timezone(0))
+ task = _fake_task("weekly-digest-sunday", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:weekly-digest-sunday", now)
+ self.assertEqual(status, FAIL)
+ self.assertIn("MISSED", detail)
+ self.assertIn("2026-07-26", detail)
+
+ def test_orphan_recovery_missed_18h_slot_past_tight_grace_now_fails(self):
+ # "Do not mask" case #2, revisited now that the persistence lag this
+ # grace exists for has been fixed at its source: CELERY_BEAT_SYNC_EVERY
+ # = 1 (backend/settings/base.py, inherited unmodified by the test
+ # settings) forces a DB flush after every task send, so last_run_at
+ # lags reality by roughly one task execution, not up to
+ # CELERY_BEAT_MAX_LOOP_INTERVAL (6h) as it used to.
+ #
+ # broadcast-orphan-recovery, last_run_at 2026-07-29 12:00:01Z, now
+ # 2026-07-29 19:00:00Z: ~1h overdue against the 6h period. Under the
+ # old 7h wide grace this landed in WARN (renamed test, see git log:
+ # test_orphan_recovery_missed_18h_slot_within_grace_warns_not_fails).
+ # Under the new tight 5-minute grace (crontab_grace_seconds(), no
+ # override needed — CELERY_BEAT_SYNC_EVERY=1 is already the real
+ # setting), 1h overdue is far past grace and correctly surfaces as
+ # FAIL — this is the intended consequence of closing the
+ # persistence-lag gap, not a regression: a 1h-overdue 6h-period task
+ # should be caught well before its next scheduled fire.
+ crontab = _fake_crontab(minute="0", hour="*/6", tz="UTC")
+ last_run_at = timezone.datetime(
+ 2026, 7, 29, 12, 0, 1, tzinfo=timezone.get_fixed_timezone(0)
+ )
+ now = timezone.datetime(2026, 7, 29, 19, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ task = _fake_task("broadcast-orphan-recovery", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", now)
+ self.assertEqual(status, FAIL)
+ self.assertIn("MISSED", detail)
+
+ def test_orphan_recovery_overdue_past_grace_fails(self):
+ # Same schedule, overdue well beyond even the old 7h wide grace —
+ # grace must not become a second staleness window regardless of
+ # which branch of crontab_grace_seconds() is active. Must still FAIL.
+ crontab = _fake_crontab(minute="0", hour="*/6", tz="UTC")
+ last_run_at = timezone.datetime(
+ 2026, 7, 29, 12, 0, 1, tzinfo=timezone.get_fixed_timezone(0)
+ )
+ now = timezone.datetime(2026, 7, 30, 1, 30, 0, tzinfo=timezone.get_fixed_timezone(0))
+ task = _fake_task("broadcast-orphan-recovery", last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, "beat:broadcast-orphan-recovery", now)
+ self.assertEqual(status, FAIL)
+ self.assertIn("MISSED", detail)
+
+
+@tag("fast")
+class CrontabDstTransitionTests(unittest.TestCase):
+ """Pin down remaining_estimate()/expected_fire across both 2026 US DST
+ transitions for the three America/New_York-seeded beat tasks, plus the
+ UTC-seeded broadcast-orphan-recovery as a control.
+
+ TzAwareCrontab (what CrontabSchedule.schedule actually returns whenever
+ DJANGO_CELERY_BEAT_TZ_AWARE is left at its True default — verified via
+ CrontabSchedule.schedule's source, not assumed) overrides is_due() to
+ convert last_run_at into schedule.tz before delegating, and its own
+ nowfunc() already returns datetime.now(schedule.tz) — so real beat
+ operation reads crontab hour/minute/day-of-week fields against local wall
+ time. remaining_estimate() has no such override: it's inherited unchanged
+ from celery.schedules.crontab and reads last_run_at.hour/.minute/
+ .isoweekday() directly off whatever tzinfo the datetime already carries.
+ _crontab_freshness calls remaining_estimate() directly (not is_due()) and
+ pins nowfun to a plain UTC `now` (timezone.now()) for determinism, so
+ without converting into schedule.tz first, an America/New_York crontab's
+ "04:00" would be read as 04:00 UTC — off by the zone's full UTC offset
+ (4-5h), not just the 1h DST delta, and wrong on every single day, not
+ just the two transition days. These tests exercise the real fix: both
+ last_run_at and the pinned `now` are converted to schedule.tz before
+ remaining_estimate() runs.
+ """
+
+ def setUp(self):
+ self.cmd = Command()
+
+ def _check(self, name, minute, hour, day_of_week, tz, last_run_at, now, expected_fire_iso):
+ crontab = _fake_crontab(minute=minute, hour=hour, day_of_week=day_of_week, tz=tz)
+ schedule = crontab.schedule
+ # Confirms we're exercising TzAwareCrontab, not a plain crontab that
+ # would happen to ignore tz entirely.
+ self.assertEqual(type(schedule).__name__, "TzAwareCrontab")
+ task = _fake_task(name, last_run_at, crontab=crontab)
+ status, _, detail = self.cmd._task_freshness(task, f"beat:{name}", now)
+ self.assertIn(expected_fire_iso, detail)
+ return status, detail
+
+ # Each case below picks `now` strictly between two candidate
+ # expected_fire values: the "buggy" one a naive (no schedule.tz
+ # conversion) remaining_estimate() call would have produced — which is
+ # off by the zone's full UTC offset, not just the 1h DST delta — and the
+ # correct one accounting for America/New_York local time across the
+ # transition. That window is 4-5h wide here (comfortably more than
+ # crontab_grace_seconds()'s tight 5-minute buffer), so asserting OK (not
+ # WARN/FAIL) at that `now` directly proves the fix, not just that some
+ # timestamp landed in the detail string: under the pre-fix code this
+ # `now` reads as hours overdue against the buggy expected_fire and FAILs.
+
+ # ── spring forward: 2026-03-08, America/New_York 02:00 -> 03:00 local ──
+
+ def test_ingest_events_daily_spring_forward(self):
+ # 0 4 * * * America/New_York. Fire before transition: 2026-03-07
+ # 04:00 EST = 09:00 UTC. Buggy (UTC-misread) expected_fire would be
+ # 2026-03-08T04:00:00Z; correct expected_fire is 2026-03-08 04:00
+ # EDT = 08:00 UTC (a 23h local interval, crossing the missing
+ # 02:00-02:59 hour). now = 06:00Z sits 2h past the buggy value but 2h
+ # before the correct one.
+ last_run_at = timezone.datetime(2026, 3, 7, 9, 0, 1, tzinfo=timezone.get_fixed_timezone(0))
+ now = timezone.datetime(2026, 3, 8, 6, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "ingest-events-daily",
+ "0",
+ "4",
+ "*",
+ "America/New_York",
+ last_run_at,
+ now,
+ "2026-03-08T08:00:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ def test_scrape_sources_daily_spring_forward(self):
+ # 30 3 * * * America/New_York. Before: 2026-03-07 03:30 EST = 08:30
+ # UTC. Buggy expected_fire: 2026-03-08T03:30:00Z. Correct: 2026-03-08
+ # 03:30 EDT = 07:30 UTC. now = 05:30Z sits between the two.
+ last_run_at = timezone.datetime(2026, 3, 7, 8, 30, 1, tzinfo=timezone.get_fixed_timezone(0))
+ now = timezone.datetime(2026, 3, 8, 5, 30, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "scrape-sources-daily",
+ "30",
+ "3",
+ "*",
+ "America/New_York",
+ last_run_at,
+ now,
+ "2026-03-08T07:30:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ def test_weekly_digest_sunday_spring_forward(self):
+ # 0 18 * * 0 America/New_York — fires on Sundays, and 2026-03-08 (the
+ # transition day itself) is a Sunday. Before: 2026-03-01 18:00 EST =
+ # 23:00 UTC. Buggy expected_fire: 2026-03-08T18:00:00Z. Correct:
+ # 2026-03-08 18:00 EDT = 22:00 UTC — both 18:00 and the crontab's own
+ # hour sit outside the nonexistent 02:00-02:59 window, but the fire
+ # is still on the transition day, so the UTC offset used to compute
+ # it must flip from -5 to -4. now = 20:00Z sits between the two.
+ last_run_at = timezone.datetime(2026, 3, 1, 23, 0, 1, tzinfo=timezone.get_fixed_timezone(0))
+ now = timezone.datetime(2026, 3, 8, 20, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "weekly-digest-sunday",
+ "0",
+ "18",
+ "sun",
+ "America/New_York",
+ last_run_at,
+ now,
+ "2026-03-08T22:00:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ def test_broadcast_orphan_recovery_spring_forward_control(self):
+ # 0 */6 * * * UTC — no local tz involved, must be unaffected by the
+ # US DST transition happening the same day. now sits before the
+ # 06:00Z fire (not yet due), unlike the NY-tz cases above which
+ # deliberately probe a `now` between the buggy and correct fire
+ # times — there's no such gap to probe here since UTC has none.
+ last_run_at = timezone.datetime(2026, 3, 8, 0, 0, 1, tzinfo=timezone.get_fixed_timezone(0))
+ now = timezone.datetime(2026, 3, 8, 5, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "broadcast-orphan-recovery",
+ "0",
+ "*/6",
+ "*",
+ "UTC",
+ last_run_at,
+ now,
+ "2026-03-08T06:00:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ # ── fall back: 2026-11-01, America/New_York 01:00-01:59 local occurs twice ──
+
+ def test_ingest_events_daily_fall_back(self):
+ # Before: 2026-10-31 04:00 EDT = 08:00 UTC. Buggy expected_fire:
+ # 2026-11-01T04:00:00Z. Correct: 2026-11-01 04:00 EST = 09:00 UTC (a
+ # 25h local interval). now = 06:00Z sits between the two.
+ last_run_at = timezone.datetime(
+ 2026, 10, 31, 8, 0, 1, tzinfo=timezone.get_fixed_timezone(0)
+ )
+ now = timezone.datetime(2026, 11, 1, 6, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "ingest-events-daily",
+ "0",
+ "4",
+ "*",
+ "America/New_York",
+ last_run_at,
+ now,
+ "2026-11-01T09:00:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ def test_scrape_sources_daily_fall_back(self):
+ # Before: 2026-10-31 03:30 EDT = 07:30 UTC. Buggy expected_fire:
+ # 2026-11-01T03:30:00Z. Correct: 2026-11-01 03:30 EST = 08:30 UTC.
+ # now = 05:30Z sits between the two.
+ last_run_at = timezone.datetime(
+ 2026, 10, 31, 7, 30, 1, tzinfo=timezone.get_fixed_timezone(0)
+ )
+ now = timezone.datetime(2026, 11, 1, 5, 30, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "scrape-sources-daily",
+ "30",
+ "3",
+ "*",
+ "America/New_York",
+ last_run_at,
+ now,
+ "2026-11-01T08:30:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ def test_weekly_digest_sunday_fall_back(self):
+ # 2026-11-01 (the transition Sunday) is the fire day. Before:
+ # 2026-10-25 18:00 EDT = 22:00 UTC. Buggy expected_fire:
+ # 2026-11-01T18:00:00Z. Correct: 2026-11-01 18:00 EST = 23:00 UTC —
+ # offset flips from -4 to -5. now = 20:00Z sits between the two.
+ last_run_at = timezone.datetime(
+ 2026, 10, 25, 22, 0, 1, tzinfo=timezone.get_fixed_timezone(0)
+ )
+ now = timezone.datetime(2026, 11, 1, 20, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "weekly-digest-sunday",
+ "0",
+ "18",
+ "sun",
+ "America/New_York",
+ last_run_at,
+ now,
+ "2026-11-01T23:00:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ def test_broadcast_orphan_recovery_fall_back_control(self):
+ # Same reasoning as the spring-forward control above.
+ last_run_at = timezone.datetime(2026, 11, 1, 0, 0, 1, tzinfo=timezone.get_fixed_timezone(0))
+ now = timezone.datetime(2026, 11, 1, 5, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "broadcast-orphan-recovery",
+ "0",
+ "*/6",
+ "*",
+ "UTC",
+ last_run_at,
+ now,
+ "2026-11-01T06:00:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
+ # ── steady state, no transition nearby: confirms this was never a
+ # DST-only bug — before the fix, remaining_estimate() misread
+ # America/New_York crontab fields as UTC every single day, not just
+ # across the two transition days. ──
+
+ def test_ingest_events_daily_steady_state_not_dst_only(self):
+ # 2026-06-01 04:00 EDT = 08:00 UTC -> next correct fire 2026-06-02
+ # 04:00 EDT = 08:00 UTC (both days EDT, no transition involved). The
+ # buggy (UTC-misread) expected_fire would be 2026-06-02T04:00:00Z;
+ # now = 06:00Z sits between the two, same as the transition-day
+ # cases above — proves the pre-fix bug wasn't DST-specific at all,
+ # since no DST transition falls anywhere near this date.
+ last_run_at = timezone.datetime(2026, 6, 1, 8, 0, 1, tzinfo=timezone.get_fixed_timezone(0))
+ now = timezone.datetime(2026, 6, 2, 6, 0, 0, tzinfo=timezone.get_fixed_timezone(0))
+ status, detail = self._check(
+ "ingest-events-daily",
+ "0",
+ "4",
+ "*",
+ "America/New_York",
+ last_run_at,
+ now,
+ "2026-06-02T08:00:00+00:00",
+ )
+ self.assertEqual(status, OK)
+
@tag("db")
class PeriodicTaskHealthcheckDbTests(TestCase):
@@ -132,6 +536,8 @@ def test_missing_seeded_task_fails(self):
self.assertIn("missing", detail)
def test_stale_but_previously_run_task_fails_the_whole_check(self):
+ # 200h ago is ~8 missed daily fires — this crontab-backed task must
+ # FAIL with a missed-fire detail, not ride along as OK.
schedule, _ = CrontabSchedule.objects.get_or_create(
minute="0", hour="4", day_of_week="*", day_of_month="*", month_of_year="*"
)
@@ -148,4 +554,4 @@ def test_stale_but_previously_run_task_fails_the_whole_check(self):
matches = [r for r in results if r[1] == "beat:scrape-sources-daily"]
status, _, detail = matches[0]
self.assertEqual(status, FAIL)
- self.assertIn("STALE", detail)
+ self.assertIn("MISSED", detail)
diff --git a/backendServer/events/tests/test_town_backfill.py b/backendServer/events/tests/test_town_backfill.py
new file mode 100644
index 0000000..044d12b
--- /dev/null
+++ b/backendServer/events/tests/test_town_backfill.py
@@ -0,0 +1,110 @@
+"""Guards the ticket-36.2 backfill (migration 0017) for the 15 live events
+that shipped with `town = NULL`.
+
+Migration 0017 runs as part of the test database setup, so the venues it maps
+already carry the right `Town` FK by the time these tests run — these tests
+assert that end state, then separately exercise the migration's own
+`backfill_towns` / `unbackfill_towns` functions directly (via the historical
+app registry) to prove reversibility without depending on unrelated rows.
+"""
+
+import importlib
+
+from django.apps import apps as global_apps
+from django.test import TestCase, tag
+
+from events.models import Event, Town
+
+from .factories import make_event, make_town
+
+backfill_module = importlib.import_module("events.migrations.0017_backfill_null_town_events")
+
+BACKFILLED_VENUES = {
+ "Fair Game Beverage Company": "pittsboro",
+ "bmc brewing": "pittsboro",
+ "Chatham YMCA": "pittsboro",
+ "Covenant Place": "carrboro",
+}
+
+LEFT_NULL_VENUE = "Jordan Lake State Recreation Area – New Hope Overlook Access"
+
+
+@tag("db")
+class TownBackfillMigrationTests(TestCase):
+ """Exercises 0017's RunPython functions directly against fresh rows.
+
+ This does not depend on prod's actual 15 rows still existing in the test
+ DB — it creates its own rows shaped like them (empty source_name, NULL
+ town, matching venue strings) and proves the migration's mapping function
+ assigns the right town and leaves the untouched venue alone.
+ """
+
+ def setUp(self):
+ for slug, name in [("pittsboro", "Pittsboro"), ("carrboro", "Carrboro")]:
+ make_town(slug, name)
+
+ self.mapped_events = {}
+ for venue in BACKFILLED_VENUES:
+ event = make_event(title=f"{venue} test event", venue=venue)
+ event.town = None
+ event.source_name = ""
+ event.save()
+ self.mapped_events[venue] = event
+
+ left_null_event = make_event(title="Spring Cleanup", venue=LEFT_NULL_VENUE)
+ left_null_event.town = None
+ left_null_event.source_name = ""
+ left_null_event.save()
+ self.left_null_event = left_null_event
+
+ def _run_backfill(self):
+ backfill_module.backfill_towns(global_apps, None)
+
+ def test_backfill_assigns_mapped_venues_to_expected_town(self):
+ self._run_backfill()
+ for venue, slug in BACKFILLED_VENUES.items():
+ event = Event.objects.get(uuid=self.mapped_events[venue].uuid)
+ self.assertIsNotNone(event.town, f"{venue} should have a town after backfill")
+ self.assertEqual(event.town.slug, slug)
+
+ def test_backfill_leaves_ambiguous_venue_null(self):
+ self._run_backfill()
+ event = Event.objects.get(uuid=self.left_null_event.uuid)
+ self.assertIsNone(event.town)
+
+ def test_backfill_does_not_touch_events_with_source_name(self):
+ # A row with the same venue string but a real source_name looks like
+ # pipeline output, not the hand-entered rows this migration targets.
+ town = Town.objects.get(slug="pittsboro")
+ pipeline_event = make_event(
+ title="Pipeline-sourced Fair Game show",
+ venue="Fair Game Beverage Company",
+ town=town,
+ )
+ pipeline_event.town = None
+ pipeline_event.source_name = "some-ics-feed"
+ pipeline_event.save()
+
+ self._run_backfill()
+
+ pipeline_event.refresh_from_db()
+ self.assertIsNone(pipeline_event.town)
+
+ def test_unbackfill_restores_null_for_mapped_venues(self):
+ self._run_backfill()
+ backfill_module.unbackfill_towns(global_apps, None)
+ for venue in BACKFILLED_VENUES:
+ event = Event.objects.get(uuid=self.mapped_events[venue].uuid)
+ self.assertIsNone(event.town)
+
+
+@tag("db")
+class TownBackfillAppliedStateTests(TestCase):
+ """The migration already ran against the test DB (it's in the migration
+ graph) — these tests describe the town coverage it's expected to produce
+ without asserting on prod-only rows that may not exist in a fresh DB.
+ """
+
+ def test_covered_towns_used_by_backfill_exist(self):
+ expected_slugs = set(BACKFILLED_VENUES.values())
+ self.assertTrue(expected_slugs <= set(Town.objects.values_list("slug", flat=True)))
diff --git a/backendServer/ingestion/deduplicator.py b/backendServer/ingestion/deduplicator.py
index 43bcd15..d1e90a1 100644
--- a/backendServer/ingestion/deduplicator.py
+++ b/backendServer/ingestion/deduplicator.py
@@ -29,7 +29,11 @@
# remain permanent dedupe anchors (see `services.publish_all_approved`).
# `rejected` is deliberately excluded: matching against it would silently
# duplicate-mark a host who fixed a flagged problem and resubmitted.
-CANDIDATE_STATUSES = ["pending", "approved", "duplicate", "published"]
+# `skipped_no_town` is included so a re-scrape of an already-skipped,
+# out-of-coverage event matches the existing row instead of creating a fresh
+# `skipped_no_town` row every poll — one terminal row per event, not one per
+# scrape (see services.publish_all_approved).
+CANDIDATE_STATUSES = ["pending", "approved", "duplicate", "published", "skipped_no_town"]
_LEADING_ARTICLE_RE = re.compile(r"^(the|a|an)\s+")
diff --git a/backendServer/ingestion/management/commands/reopen_skipped_towns.py b/backendServer/ingestion/management/commands/reopen_skipped_towns.py
new file mode 100644
index 0000000..677acf7
--- /dev/null
+++ b/backendServer/ingestion/management/commands/reopen_skipped_towns.py
@@ -0,0 +1,38 @@
+import logging
+
+from django.core.management.base import BaseCommand
+
+from events.models import Town
+from ingestion.models import StagedEvent
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+ help = (
+ "Reopen StagedEvents that were skipped for having no matching Town "
+ "(status=skipped_no_town) whose town slug now resolves to a real Town row. "
+ "Run this once after adding coverage for a new town (a new Town row, or an "
+ "alias fix) — publish_all_approved no longer retries skipped rows on its "
+ "own, so a coverage change is inert until this command flips them back to "
+ "'approved' for the next publish run to pick up."
+ )
+
+ def handle(self, *args, **options):
+ candidates = list(StagedEvent.objects.filter(status="skipped_no_town"))
+ reopened = 0
+
+ for staged in candidates:
+ town_slug = staged.town.lower().replace(" ", "-") if staged.town else None
+ if not town_slug or not Town.objects.filter(slug=town_slug).exists():
+ continue
+ staged.status = "approved"
+ staged.save(update_fields=["status"])
+ reopened += 1
+
+ self.stdout.write(
+ self.style.SUCCESS(
+ f"Reopened {reopened} of {len(candidates)} skipped_no_town rows "
+ "now matching a Town."
+ )
+ )
diff --git a/backendServer/ingestion/migrations/0016_stagedevent_skipped_no_town.py b/backendServer/ingestion/migrations/0016_stagedevent_skipped_no_town.py
new file mode 100644
index 0000000..1c533f1
--- /dev/null
+++ b/backendServer/ingestion/migrations/0016_stagedevent_skipped_no_town.py
@@ -0,0 +1,60 @@
+from django.db import migrations, models
+
+
+# Backfill for ticket 36.1: prod has 27 StagedEvent rows stuck `approved` with
+# `published_event=None` — towns deliberately out of coverage (Apex 15, Durham
+# 12) that `publish_all_approved` re-logs and re-skips on every single pipeline
+# run. Converting them to the new terminal-ish `skipped_no_town` status here
+# (rather than waiting for the next publish run) means the very first run after
+# this migration deploys is already quiet, instead of logging one more round.
+def backfill_skipped_no_town(apps, schema_editor):
+ StagedEvent = apps.get_model("ingestion", "StagedEvent")
+ Town = apps.get_model("events", "Town")
+
+ known_slugs = set(Town.objects.values_list("slug", flat=True))
+ stuck = StagedEvent.objects.filter(status="approved", published_event__isnull=True)
+
+ to_update = []
+ for staged in stuck:
+ town_slug = staged.town.lower().replace(" ", "-") if staged.town else None
+ if town_slug not in known_slugs:
+ staged.status = "skipped_no_town"
+ to_update.append(staged)
+
+ if to_update:
+ StagedEvent.objects.bulk_update(to_update, ["status"])
+
+
+def unbackfill_skipped_no_town(apps, schema_editor):
+ # Reversible but coarse: every `skipped_no_town` row reverts to `approved`,
+ # matching the pre-migration state regardless of which run produced it.
+ StagedEvent = apps.get_model("ingestion", "StagedEvent")
+ StagedEvent.objects.filter(status="skipped_no_town").update(status="approved")
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ("events", "0016_seed_chatham_towns"),
+ ("ingestion", "0015_alter_stagedevent_status"),
+ ]
+
+ operations = [
+ migrations.AlterField(
+ model_name="stagedevent",
+ name="status",
+ field=models.CharField(
+ choices=[
+ ("pending", "Pending Review"),
+ ("approved", "Approved"),
+ ("rejected", "Rejected"),
+ ("duplicate", "Duplicate"),
+ ("published", "Published"),
+ ("skipped_no_town", "Skipped — No Matching Town"),
+ ],
+ default="pending",
+ max_length=20,
+ ),
+ ),
+ migrations.RunPython(backfill_skipped_no_town, unbackfill_skipped_no_town),
+ ]
diff --git a/backendServer/ingestion/models.py b/backendServer/ingestion/models.py
index 08d0e5e..e2100c5 100644
--- a/backendServer/ingestion/models.py
+++ b/backendServer/ingestion/models.py
@@ -102,6 +102,7 @@ class StagedEvent(models.Model):
("rejected", "Rejected"),
("duplicate", "Duplicate"),
("published", "Published"),
+ ("skipped_no_town", "Skipped — No Matching Town"),
]
raw_event = models.OneToOneField(
diff --git a/backendServer/ingestion/services.py b/backendServer/ingestion/services.py
index 257e09c..51485e3 100644
--- a/backendServer/ingestion/services.py
+++ b/backendServer/ingestion/services.py
@@ -60,13 +60,22 @@ def publish_all_approved(source=None, force_town=None): # noqa: C901 # inheren
town_slug = staged.town.lower().replace(" ", "-") if staged.town else None
town_obj = Town.objects.filter(slug=town_slug).first() if town_slug else None
if town_obj is None:
+ # Terminal-ish status, not a delete: `skipped_no_town` rows
+ # stay out of the `approved` queryset above, so this branch
+ # (and its log line) only ever fires once per row instead of
+ # re-logging on every subsequent run. They remain dedupe
+ # candidates (CANDIDATE_STATUSES in deduplicator.py) and are
+ # reopened deliberately via `manage.py reopen_skipped_towns`
+ # once coverage is added — see that command's docstring.
logger.warning(
- "Dropping staged event '%s' — no Town matches slug '%s'"
+ "Skipping staged event '%s' — no Town matches slug '%s'"
" (gemini town=%r)",
staged.title,
town_slug,
staged.town,
)
+ staged.status = "skipped_no_town"
+ staged.save(update_fields=["status"])
continue
if staged.raw_event_id and staged.raw_event and staged.raw_event.source:
source_name = staged.raw_event.source.name
diff --git a/backendServer/ingestion/tests/test_services_db.py b/backendServer/ingestion/tests/test_services_db.py
index 0167f41..342423c 100644
--- a/backendServer/ingestion/tests/test_services_db.py
+++ b/backendServer/ingestion/tests/test_services_db.py
@@ -1,5 +1,7 @@
from datetime import UTC, datetime
+from io import StringIO
+from django.core.management import call_command
from django.test import TestCase, tag
from events.models import Event, Town
@@ -108,12 +110,12 @@ def test_chatham_county_towns_publish(self):
{"Growers & Makers Market": "siler-city", "Bynum Front Porch Music": "bynum"},
)
- def test_unmatched_town_is_skipped_but_retried(self):
- """A town outside coverage (Apex is not in the service area) is still
- skipped — but the row keeps published_event=None, so the terminal
- approved->published sweep leaves it alone and the next run retries it.
- That is why adding a Town row backfills previously-dropped events with
- no separate migration.
+ def test_unmatched_town_is_marked_skipped_no_town(self):
+ """A town outside coverage (Apex is not in the service area) is moved to
+ the terminal-ish `skipped_no_town` status rather than left `approved`.
+ `approved` means "will be published", which was false for these rows —
+ ticket 36.1. The row is not deleted (it survives as a dedupe anchor —
+ see CANDIDATE_STATUSES) and no Event is created.
"""
self._staged("Somewhere Else", "approved", town="Apex")
@@ -122,4 +124,70 @@ def test_unmatched_town_is_skipped_but_retried(self):
self.assertEqual(result["published"], 0)
self.assertEqual(result["removed"], 0)
self.assertFalse(Event.objects.exists())
- self.assertEqual(StagedEvent.objects.get(title="Somewhere Else").status, "approved")
+ self.assertEqual(StagedEvent.objects.get(title="Somewhere Else").status, "skipped_no_town")
+
+ def test_second_run_does_not_relog_an_already_skipped_row(self):
+ """36.1's stated QA: a second consecutive pipeline run must not
+ re-emit the 'no Town matches' warning for a row the first run already
+ handled. Once skipped, the row is `skipped_no_town`, not `approved`,
+ so the `status="approved"` queryset in publish_all_approved no longer
+ selects it at all — the log line naturally fires zero times.
+ """
+ self._staged("Somewhere Else", "approved", town="Apex")
+
+ with self.assertLogs("ingestion.services", level="WARNING") as cm:
+ publish_all_approved()
+ self.assertTrue(any("no Town matches" in msg for msg in cm.output))
+
+ # Second run: nothing left in the `approved` queue for Apex, so no
+ # logger call happens at all. assertNoLogs would be ideal but isn't
+ # available on this Django's TestCase; assert the queryset is empty
+ # and re-running is a no-op instead.
+ self.assertEqual(StagedEvent.objects.filter(status="approved", town="Apex").count(), 0)
+ result = publish_all_approved()
+ self.assertEqual(result, {"published": 0, "already_published": 0, "removed": 0})
+ self.assertEqual(StagedEvent.objects.get(title="Somewhere Else").status, "skipped_no_town")
+
+ def test_skipped_row_is_still_a_dedupe_candidate(self):
+ """A re-scrape of an already-skipped, out-of-coverage event must match
+ the existing skipped_no_town row instead of creating a fresh row every
+ poll — that would just trade one leak (retrying forever) for another
+ (an unbounded pile of skipped duplicates). CANDIDATE_STATUSES includes
+ skipped_no_town for exactly this reason.
+ """
+ original = self._staged("Somewhere Else", "approved", town="Apex")
+ publish_all_approved()
+ original.refresh_from_db()
+ self.assertEqual(original.status, "skipped_no_town")
+
+ rescraped = self._staged("Somewhere Else", "pending", town="Apex")
+ dup = find_duplicate(rescraped)
+ self.assertEqual(dup, original)
+
+ def test_reopen_skipped_towns_command_reopens_matching_rows(self):
+ """The chosen re-open strategy: coverage changes are not free anymore
+ (unlike the old always-retry behavior) — an explicit management command
+ must be run after adding a Town so previously-skipped rows return to
+ `approved` for the next publish run to pick up. Rows whose town is still
+ uncovered are left alone.
+ """
+ self._staged("Apex Show", "approved", town="Apex")
+ publish_all_approved()
+ self.assertEqual(StagedEvent.objects.get(title="Apex Show").status, "skipped_no_town")
+
+ self._staged("Durham Show", "approved", town="Durham")
+ publish_all_approved()
+ self.assertEqual(StagedEvent.objects.get(title="Durham Show").status, "skipped_no_town")
+
+ Town.objects.create(slug="apex", name="Apex")
+
+ out = StringIO()
+ call_command("reopen_skipped_towns", stdout=out)
+
+ self.assertEqual(StagedEvent.objects.get(title="Apex Show").status, "approved")
+ self.assertEqual(StagedEvent.objects.get(title="Durham Show").status, "skipped_no_town")
+ self.assertIn("Reopened 1 of 2", out.getvalue())
+
+ result = publish_all_approved()
+ self.assertEqual(result["published"], 1)
+ self.assertEqual(Event.objects.get(title="Apex Show").town.slug, "apex")
diff --git a/docs/index.md b/docs/index.md
index ad05daa..4598bf7 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -8,7 +8,7 @@ Deep-dive guides beyond the root-level docs. Each file is a focused reference on
| [ingestion-pipeline.md](ingestion-pipeline.md) | End-to-end walkthrough of the poll → standardize → stage → publish flow | Working on `ingestion/` or the cron pipeline |
| [safety-scoring.md](safety-scoring.md) | How `safety_scorer.py` works and threshold tuning | Adjusting safety scoring or ingestion quality |
| [devtools-ingestion-playground.md](devtools-ingestion-playground.md) | Plan for the dev-only ingestion playground (`devtools/`): replaying feeds through the poll → publish pipeline locally | Building or debugging ingestion tooling |
-| [ingestion-monitoring.md](ingestion-monitoring.md) | The `/devtools/monitor` funnel dashboard: bucket semantics, health-level thresholds, `SourceRun` statuses, the dry-run probe endpoint, prod read-only safety, and a triage runbook | Diagnosing a stuck or failing ingestion source, or building on the monitor/probe |
+| [ingestion-monitoring.md](ingestion-monitoring.md) | The `/devtools/monitor` funnel dashboard: bucket semantics, health-level thresholds, `SourceRun` statuses, the dry-run probe endpoint, prod read-only safety, and a triage runbook; also the beat scheduler `last_run_at` persistence-lag incident | Diagnosing a stuck or failing ingestion source, building on the monitor/probe, or investigating `manage.py healthcheck` beat staleness |
| [admin-backend.md](admin-backend.md) | Guide to the django-unfold admin UI and review workflows | Modifying admin registration or review flows |
| [redis-celery-handoff.md](redis-celery-handoff.md) | Redis + Celery setup: broker/cache, worker + beat, task conventions | Adding async tasks, scheduled jobs, or touching the worker |
| [dev-db-isolation.md](dev-db-isolation.md) | Neon dev-branch setup — isolating local dev from the prod DB | Setting up a dev environment or running migrations |
diff --git a/docs/ingestion-monitoring.md b/docs/ingestion-monitoring.md
index ac1dca7..0b4cc1a 100644
--- a/docs/ingestion-monitoring.md
+++ b/docs/ingestion-monitoring.md
@@ -389,3 +389,88 @@ If this keeps recurring across future migrations, the more durable fix is runnin
`ALTER DEFAULT PRIVILEGES` as (or granted by) whatever role prod migrations actually run
as — but that's a role-alignment decision to make deliberately, not a blanket workaround
to apply here.
+
+---
+
+## Beat scheduler: last_run_at persistence lag
+
+This section is about `manage.py healthcheck` (the systemd-driven prod healthcheck run
+by `healthcheck.service`, documented in [DEPLOY.md](../DEPLOY.md#health-check)) and
+`django_celery_beat`'s scheduler, not the `/devtools/monitor` dashboard the rest of this
+doc covers — it lives here because the incident originates in the same
+poll/schedule/freshness family of problems and `backend/settings/base.py` points here for
+the writeup.
+
+**Symptom (2026-07-29, ticket 36.4):** `manage.py healthcheck` reported
+`beat:broadcast-orphan-recovery STALE — last run 9.7h ago (> 7h window)` every hour on
+prod, making `healthcheck.service` a permanently failing systemd unit even though the
+task was actually running on schedule.
+
+**Root cause.** `django_celery_beat`'s `DatabaseScheduler` keeps `PeriodicTask.last_run_at`
+in memory and only writes it to Postgres when celery's `Scheduler._do_sync()` runs. Both
+sync triggers were defeated:
+
+1. `backend/settings/base.py` sets `CELERY_BEAT_MAX_LOOP_INTERVAL = 6 * 60 * 60`. That
+ becomes `Scheduler.max_interval` (`celery/beat.py:252-258`), overriding
+ django_celery_beat's `DEFAULT_MAX_INTERVAL = 5` (`django_celery_beat/schedulers.py:33`).
+ In `Service.start()` (`celery/beat.py:633-655`), `_do_sync()` is only called after
+ `time.sleep(interval)`, and only inside `if interval and interval > 0.0` — so beat can
+ sleep up to 6 hours between sync opportunities.
+2. `Scheduler.apply_async`'s `finally` block (`celery/beat.py:416-418`) is the other
+ trigger, gated on `should_sync()` (`celery/beat.py:381-387`). Its task-count clause is
+ dead because `app.conf.beat_sync_every` defaults to **0**. Its time clause requires
+ `monotonic() - _last_sync > sync_every` where `sync_every = 180` (3 minutes) — so a task
+ firing shortly after a prior sync flushes nothing.
+
+**Measured proof on prod** (read-only queries plus one authorized `-l debug` restart):
+
+```
+22:24:33 BEFORE last_run_at = 2026-07-29 12:00:01.808480+00 total_run_count = 9
+22:24:35 systemctl restart celerybeat
+22:24:37 AFTER last_run_at = 2026-07-29 18:00:00.021322+00 total_run_count = 10
+```
+
+The 18:00 write had been buffered in memory for 4h24m and was flushed only by celery's
+shutdown path (`Service.start()`'s `finally: self.sync()`). Beat's journal showed
+`Sending due task broadcast-orphan-recovery` at `18:00:00,023`, and the broadcast worker
+logged the task `received` and `succeeded in 0.083s` — the task was sent and executed on
+time. Only the bookkeeping write was late. Beat had one process, `NRestarts=0`, and no
+errors or warnings above DEBUG in `journalctl -u celerybeat`.
+
+**Hypotheses that were disproven** (recorded so nobody re-chases them):
+
+- *A dev/prod database split* — ruled out: `DJANGO_ENV=prod`, DB host
+ `ep-late-smoke-...-pooler`, name `neondb`.
+- *A swallowed `ObjectDoesNotExist`/`TypeError` in `DatabaseScheduler.sync()`*
+ (`django_celery_beat/schedulers.py:459`, which re-adds the failed name to `_dirty` and
+ retries forever with no log above DEBUG) — ruled out: `save()` succeeded fine at
+ shutdown against the same pk.
+- *A stale in-memory `_schedule` overwriting the fresh value* — ruled out: memory held the
+ 18:00 value, not 12:00.
+- *Clock skew.* This one is a trap worth documenting on its own: beat's startup banner
+ (`celery beat vX.Y.Z is starting. / LocalTime -> ...`) is flushed at process
+ **teardown** and reports that process's own *start* time, so in `journalctl` it looks
+ like beat's clock is running behind the system clock. It isn't. Verified across three
+ consecutive restarts, where each banner's `LocalTime` matched the previous PID's
+ `beat: Starting...` timestamp exactly.
+
+**The fix:** `CELERY_BEAT_SYNC_EVERY = 1` in `backend/settings/base.py`, which maps to
+`app.conf.beat_sync_every` → `Scheduler.sync_every_tasks` (`celery/beat.py:262-264`),
+forcing a sync after every task send. This costs one UPDATE per task fire and does not
+change beat's DB poll rate — deliberately, since `CELERY_BEAT_MAX_LOOP_INTERVAL` exists to
+reduce Neon serverless wake-ups, and that trade-off is still worth making. Regression test:
+`backendServer/events/tests/test_beat_last_run_persistence.py`.
+
+**Why this matters for monitoring:** before this fix, `last_run_at` could lag reality by
+up to `CELERY_BEAT_MAX_LOOP_INTERVAL` (6 hours), so any staleness check reading that
+column was measuring a lagging value, not a real one. That's why a 7h window on a 6h-period
+task produced false alarms — and why the tempting fix of "just widen the window to 13h"
+would have been wrong: it papers over the false alarm but doesn't fix the actual defect
+(a real multi-hour outage would then also go undetected for up to 13h). Widening the
+window here would have hidden a genuine bug rather than fixed it.
+
+**How to tell a persistence lag from a real outage:** restart `celerybeat` and re-read the
+`PeriodicTask` row. Beat's shutdown path always flushes (`Service.start()`'s
+`finally: self.sync()`), so a `last_run_at` that jumps forward immediately after the
+restart was sitting in memory, not missed — the task fired, the write was just late. If
+the value doesn't move after a restart, that's a real gap: the task actually isn't firing.
diff --git a/docs/redis-celery-handoff.md b/docs/redis-celery-handoff.md
index 2cd3c64..f4f65c6 100644
--- a/docs/redis-celery-handoff.md
+++ b/docs/redis-celery-handoff.md
@@ -169,8 +169,12 @@ bash deploy/healthcheck.sh # ✓/!/✗ for RAM, disk, all units, Redis, D
```
This is the fastest way to confirm beat is actually firing: the Application section
-runs `manage.py healthcheck`, which reports each `PeriodicTask`'s `enabled` flag and
-`last_run_at` freshness (daily stale after ~25h, weekly after ~8d) and flags any
+runs `manage.py healthcheck`, which reports each `PeriodicTask`'s `enabled` flag and,
+for each of the four seeded crontab-backed tasks (`ingest-events-daily`,
+`scrape-sources-daily`, `weekly-digest-sunday`, `broadcast-orphan-recovery`), derives
+the expected fire time from that task's own crontab and flags a missed occurrence —
+`DEFAULT_STALENESS_HOURS` now only backs the must-exist task check and a fallback
+window for interval-backed tasks with no crontab to derive from. It also flags any
leftover OS-cron `ingest_events`/`send_weekly_digest` entries that would double-run.
Exits non-zero on any critical failure. Details in [DEPLOY.md](../DEPLOY.md#health-check).
diff --git a/notion-sync/STATE.md b/notion-sync/STATE.md
index 1ddf970..d3f3e08 100644
--- a/notion-sync/STATE.md
+++ b/notion-sync/STATE.md
@@ -6,7 +6,7 @@ The ledger mirrors what *should* be on the Notion board so the desktop app can r
---
-**Next suite number:** `36`
+**Next suite number:** `37`
## Suite ledger
@@ -31,6 +31,7 @@ per-ticket status lives on each ticket subpage (see OUTBOX preamble).
| 33 | Ingestion monitor diagnostics correctness (health levels, zero legibility, GRANT detection) | In Prod | 33.1–33.5 | _(pending)_ |
| 34 | Ingestion pipeline resilience (dedupe corpus, standardizer fallback, direct-submission delivery) | In Prod | 34.1–34.5 | _(pending)_ |
| 35 | Prod scheduler outage (snap-uv user-slice teardown) + monitor correctness | In Prod | 35.1–35.11, 35.13, 35.14 (35.12 merged into 35.8) | _(pending)_ |
+| 36 | Ingestion funnel dead-ends (out-of-coverage limbo, town-less events, missed sends, beat bookkeeping) | Needs QA | 36.1–36.4 (36.5 closed won't-fix — its 13h window would have masked 36.4) | _(pending)_ |