From 9bda0c6cd75f37cdc71eb5265a0924d4823b0c6e Mon Sep 17 00:00:00 2001 From: Arya Venkatesan Date: Thu, 30 Jul 2026 01:21:40 -0400 Subject: [PATCH 1/9] fix(ingestion): terminal status for out-of-coverage towns (36.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A staged event whose town had no `Town` row stayed `approved` forever and re-logged a warning on every pipeline run — 27 rows on prod (Apex 15, Durham 12) burning 27 unactionable log lines per run. That is how Siler City's dropped events went unnoticed for weeks (35.14). Adds `skipped_no_town`, set at the skip point, so the row leaves the `approved` queryset and the warning fires exactly once. Added to `deduplicator.CANDIDATE_STATUSES` so a re-scrape collapses onto the existing row instead of inserting a fresh one each poll. Coverage fixes are no longer free: `manage.py reopen_skipped_towns` re-opens rows once a `Town` lands. Migration backfills the 27 existing rows. Includes the devtools funnel fix in the same commit deliberately — the new status fell into no funnel bucket, silently breaking the documented "buckets sum to raw" invariant and dropping 27 rows out of the monitor. Splitting these would plant a commit with failing devtools tests. The monitor now surfaces a `no_town` bucket with an explanatory note, which is what 36.1 wanted: `raw > 0, published = 0` now says why. The reconcile test sums buckets dynamically so a future unwired bucket fails instead of passing silently. Co-Authored-By: Claude --- backendServer/devtools/monitoring.py | 41 ++++++++- .../devtools/templates/devtools/monitor.html | 8 +- .../devtools/tests/test_monitoring_db.py | 83 ++++++++++++++++--- backendServer/ingestion/deduplicator.py | 6 +- .../commands/reopen_skipped_towns.py | 38 +++++++++ .../0016_stagedevent_skipped_no_town.py | 60 ++++++++++++++ backendServer/ingestion/models.py | 1 + backendServer/ingestion/services.py | 11 ++- .../ingestion/tests/test_services_db.py | 82 ++++++++++++++++-- 9 files changed, 304 insertions(+), 26 deletions(-) create mode 100644 backendServer/ingestion/management/commands/reopen_skipped_towns.py create mode 100644 backendServer/ingestion/migrations/0016_stagedevent_skipped_no_town.py 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 @@

Collectors

Held for review Rejected Approved + No town Published @@ -274,10 +275,11 @@

Collectors

{{ c.funnel.held_for_review }} {{ c.funnel.rejected }} {{ c.funnel.approved }} + {{ c.funnel.no_town }}{% if c.no_town_note %} — {{ c.no_town_note }}{% endif %} {{ c.funnel.published }} - + {% endfor %} @@ -302,6 +304,7 @@

Broadcast — inbound (direct submissions)

Held for review Rejected Approved + No town Published @@ -325,10 +328,11 @@

Broadcast — inbound (direct submissions)

{{ c.funnel.held_for_review }} {{ c.funnel.rejected }} {{ c.funnel.approved }} + {{ c.funnel.no_town }}{% if c.no_town_note %} — {{ c.no_town_note }}{% endif %} {{ c.funnel.published }} - + {% 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/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") From 961efd83c4809c4c84d64cd88c92fa64e4de6cbe Mon Sep 17 00:00:00 2001 From: Arya Venkatesan Date: Thu, 30 Jul 2026 01:21:40 -0400 Subject: [PATCH 2/9] fix(events): backfill 14 of the 15 live events with town = NULL (36.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All 15 had an empty `source_name`, so they predate the ingestion pipeline — hand-entered rows whose town was never resolved. 11 of 15 are one recurring "Senior ..." series at Covenant Place. Mapping: Fair Game Beverage Company, bmc brewing and Chatham YMCA -> pittsboro; Covenant Place -> carrboro (Carrboro Parks & Rec programming, per product decision, though the building's postal address is Chapel Hill). The Jordan Lake spring-cleanup row is left NULL on purpose: the recreation area spans Wake and Chatham with access points in different towns, and a wrong town puts an event on the wrong hub page. `Event.town` stays nullable — ingestion legitimately produces unmatched towns (36.1). Co-Authored-By: Claude --- .../0017_backfill_null_town_events.py | 94 +++++++++++++++ .../events/tests/test_town_backfill.py | 110 ++++++++++++++++++ 2 files changed, 204 insertions(+) create mode 100644 backendServer/events/migrations/0017_backfill_null_town_events.py create mode 100644 backendServer/events/tests/test_town_backfill.py 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_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))) From 27b5eff33d651f0b9a49abfa62865ccb140adfe5 Mon Sep 17 00:00:00 2001 From: Arya Venkatesan Date: Thu, 30 Jul 2026 01:22:10 -0400 Subject: [PATCH 3/9] fix(celery): make beat persist last_run_at promptly (36.4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of the hourly `healthcheck.service` failure. django-celery-beat holds `PeriodicTask.last_run_at` in memory and writes it only when `Scheduler._do_sync()` runs. Both sync triggers were defeated: - `CELERY_BEAT_MAX_LOOP_INTERVAL = 6h` becomes `Scheduler.max_interval` (celery/beat.py:253-256), overriding django-celery-beat's DEFAULT_MAX_INTERVAL of 5s. `Service.start()` (beat.py:645-652) calls `_do_sync()` only after `time.sleep(interval)`, so beat can sleep 6h between sync opportunities. - `apply_async`'s finally (beat.py:416-418) is gated on `should_sync()`, whose task-count clause is dead because `beat_sync_every` defaults to 0. A task firing shortly after a prior sync flushes nothing. Measured on prod: a write made at 18:00:00 was still absent 4h24m later and appeared the instant celerybeat was restarted (shutdown's `finally: self.sync()` flushed it) — 12:00:01/runs=9 -> 18:00:00.021/runs=10. The task was sent AND executed on time; only the bookkeeping was late. Fix is `CELERY_BEAT_SYNC_EVERY = 1`, forcing a flush after every send. Costs one UPDATE per fire and deliberately does not touch the loop interval, which exists to reduce Neon serverless wake-ups. Note this means PR #36's 7h staleness window was not "too tight" — it was comparing against a column that lagged by up to 6h. Widening it to 13h (the original 36.5) would have turned a true positive into a pass and buried this bug. Co-Authored-By: Claude --- backendServer/backend/settings/base.py | 16 ++ .../tests/test_beat_last_run_persistence.py | 177 ++++++++++++++++++ docs/index.md | 2 +- docs/ingestion-monitoring.md | 85 +++++++++ 4 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 backendServer/events/tests/test_beat_last_run_persistence.py 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/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/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. From f7356ec6c50a56a465264d91ec37ad71772219a0 Mon Sep 17 00:00:00 2001 From: Arya Venkatesan Date: Thu, 30 Jul 2026 01:22:10 -0400 Subject: [PATCH 4/9] fix(healthcheck): derive beat freshness from each task's crontab (36.3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A staleness window must exceed the task's period, so every task had a blind spot at least one period wide — a missed weekly digest rode along for ~8 days. Now `_task_freshness` asks the schedule itself: expected_next = now + schedule.remaining_estimate(last_run_at) `DEFAULT_STALENESS_HOURS` is demoted to (a) the must-exist task set, which still catches a seeded task vanishing from the schedule (2026-07-21 outage), and (b) a fallback window for interval-backed tasks only. Grace is keyed to `CELERY_BEAT_SYNC_EVERY`: 5 minutes when it is set (the persistence lag is gone — see 36.4), automatically re-widening to MAX_LOOP_INTERVAL + 1h if it is ever removed. Evaluated at call time so `@override_settings` can reach both branches. Also fixes a bug the tightened grace exposed. `TzAwareCrontab` converts `last_run_at` into the schedule's tz inside `is_due()` (tzcrontab.py:39) but inherits `remaining_estimate`/`remaining_delta` unmodified from celery, which read `.hour`/`.minute`/`.isoweekday()` off whatever tzinfo the datetime carries (celery/schedules.py:577-605). Calling `remaining_estimate()` with UTC datetimes therefore compared America/New_York crontab fields against UTC clock fields — a permanent 4-5h error every day for ingest-events-daily, scrape-sources-daily and weekly-digest-sunday. Not a DST-only bug; it reproduces mid-June. The old 7h grace had been silently absorbing it. Fixed by converting both operands via `.astimezone(schedule.tz)`, mirroring `is_due()`, rather than widening grace back out. Detection latency: ~8 days -> ~5 min for a missed weekly digest. Co-Authored-By: Claude --- .../events/management/commands/healthcheck.py | 170 ++++++- .../tests/test_healthcheck_freshness.py | 480 ++++++++++++++++-- docs/redis-celery-handoff.md | 8 +- 3 files changed, 610 insertions(+), 48 deletions(-) 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/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/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). From 590781b66e8ca279b66c740557683bc914bb1dcc Mon Sep 17 00:00:00 2001 From: Arya Venkatesan Date: Thu, 30 Jul 2026 01:24:30 -0400 Subject: [PATCH 5/9] =?UTF-8?q?chore(notion):=20suite=2036=20=E2=86=92=20N?= =?UTF-8?q?eeds=20QA,=2036.5=20closed=20won't-fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 36.1–36.4 built and green. 36.5 recorded as won't-fix: widening the orphan-recovery staleness window to 13h would have converted a true positive into a pass and buried 36.4's root cause. OUTBOX.md carries the full change blocks (gitignored, so it does not travel with this commit). Note the board still shows every suite back to 17 as _(pending)_ — the desktop app has never synced. Co-Authored-By: Claude --- notion-sync/STATE.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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)_ |