Suite 36: beat last_run_at persistence root cause + ingestion funnel dead-ends (36.1–36.4)

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

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,

backendServer/devtools/templates/devtools/monitor.html

@@ -247,6 +247,7 @@ <h2>Collectors</h2>
         <th class="num">Held for review</th>
         <th class="num">Rejected</th>
         <th class="num">Approved</th>
+        <th class="num">No town</th>
         <th class="num">Published</th>
         <th></th>
       </tr>
@@ -274,10 +275,11 @@ <h2>Collectors</h2>
         <td class="num funnel-cell{% if not c.funnel.held_for_review %} zero{% endif %}">{{ c.funnel.held_for_review }}</td>
         <td class="num funnel-cell{% if not c.funnel.rejected %} zero{% endif %}">{{ c.funnel.rejected }}</td>
         <td class="num funnel-cell{% if not c.funnel.approved %} zero{% endif %}">{{ c.funnel.approved }}</td>
+        <td class="num funnel-cell{% if not c.funnel.no_town %} zero{% endif %}"{% if c.no_town_note %} title="{{ c.no_town_note }}"{% endif %}>{{ c.funnel.no_town }}{% if c.no_town_note %}<span class="zero-note"> &mdash; {{ c.no_town_note }}</span>{% endif %}</td>
         <td class="num funnel-cell{% if not c.funnel.published %} zero{% endif %}">{{ c.funnel.published }}</td>
         <td><button type="button" class="btn-probe" data-probe-source-id="{{ c.id }}">Probe</button></td>
       </tr>
-      <tr class="drilldown-row" data-for="collector:{{ c.id }}" style="display:none;"><td colspan="15"></td></tr>
+      <tr class="drilldown-row" data-for="collector:{{ c.id }}" style="display:none;"><td colspan="16"></td></tr>
       {% endfor %}
     </tbody>
   </table>
@@ -302,6 +304,7 @@ <h2>Broadcast — inbound (direct submissions)</h2>
         <th class="num">Held for review</th>
         <th class="num">Rejected</th>
         <th class="num">Approved</th>
+        <th class="num">No town</th>
         <th class="num">Published</th>
         <th></th>
       </tr>
@@ -325,10 +328,11 @@ <h2>Broadcast — inbound (direct submissions)</h2>
         <td class="num funnel-cell{% if not c.funnel.held_for_review %} zero{% endif %}">{{ c.funnel.held_for_review }}</td>
         <td class="num funnel-cell{% if not c.funnel.rejected %} zero{% endif %}">{{ c.funnel.rejected }}</td>
         <td class="num funnel-cell{% if not c.funnel.approved %} zero{% endif %}">{{ c.funnel.approved }}</td>
+        <td class="num funnel-cell{% if not c.funnel.no_town %} zero{% endif %}"{% if c.no_town_note %} title="{{ c.no_town_note }}"{% endif %}>{{ c.funnel.no_town }}{% if c.no_town_note %}<span class="zero-note"> &mdash; {{ c.no_town_note }}</span>{% endif %}</td>
         <td class="num funnel-cell{% if not c.funnel.published %} zero{% endif %}">{{ c.funnel.published }}</td>
         <td><button type="button" class="btn-probe" data-probe-source-id="{{ c.id }}">Probe</button></td>
       </tr>
-      <tr class="drilldown-row" data-for="inbound:{{ c.id }}" style="display:none;"><td colspan="14"></td></tr>
+      <tr class="drilldown-row" data-for="inbound:{{ c.id }}" style="display:none;"><td colspan="15"></td></tr>
       {% endfor %}
     </tbody>
   </table>

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,15 +217,19 @@ 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)
 
         b = rows["Collector B"]
         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)