feat(cohorts): harden realtime backfill observability and failure handling

posthog/tasks/calculate_cohort.py

@@ -25,11 +25,13 @@
 from posthog.tasks.utils import CeleryQueue
 
 from products.cohorts.backend.backfill.finalize import finalize_backfill_runs
+from products.cohorts.backend.backfill.observe import publish_backfill_run_gauges
 from products.cohorts.backend.backfill.runs import (
+    BackfillRefusalReason,
+    attempt_backfill_run_for_cohort,
+    attempt_person_backfill_run_for_cohort,
     check_person_run_preconditions,
     check_run_preconditions,
-    create_backfill_run_for_cohort,
-    create_person_backfill_run_for_cohort,
 )
 from products.cohorts.backend.models.backfill import CohortBackfillKind
 from products.cohorts.backend.models.calculation_history import CohortCalculationHistory
@@ -919,6 +921,31 @@ def collect_cohort_query_stats(
     labelnames=["backfill_kind", "outcome"],
 )
 
+# Refusals grouped by the operator response they call for, not one label per reason: a budget
+# refusal means raise the budget or wait for in-flight runs, an occupied slot means an active run
+# already covers the cohort, and the rest are the cohort not being a candidate for this trigger.
+# Collapsing them into a single `refused`, as this used to, made the first two indistinguishable.
+# The full reason stays on the log line, so cardinality does not have to carry it.
+#
+# An occupied slot is the expected outcome of an ordinary backfill, not a wedge: a team-enablement
+# run covers every cohort on the team, and each cohort saved during one refuses this way. Read
+# `posthog_cohort_backfill_oldest_active_run_age_seconds` to tell a stuck run from normal progress.
+COHORT_BACKFILL_REFUSAL_OUTCOMES: dict[BackfillRefusalReason, str] = {
+    BackfillRefusalReason.OVER_BUDGET: "refused_over_budget",
+    BackfillRefusalReason.RUN_SLOT_OCCUPIED: "refused_slot_occupied",
+    BackfillRefusalReason.PARTICIPATION_ACTIVE: "refused_slot_occupied",
+    BackfillRefusalReason.SLOT_RACE: "refused_slot_occupied",
+    BackfillRefusalReason.TEAM_NOT_REALTIME: "refused_ineligible",
+    BackfillRefusalReason.COHORT_MISSING: "refused_ineligible",
+    BackfillRefusalReason.COHORT_INELIGIBLE: "refused_ineligible",
+    BackfillRefusalReason.INVALID_HORIZON: "refused_ineligible",
+    BackfillRefusalReason.PINNING_CAP_EXCEEDED: "refused_ineligible",
+    # Not transient: the sizing scan's read cap is deterministic, so a cohort that trips it trips it
+    # again on every later trigger until someone raises the limit. Nothing clears on its own.
+    BackfillRefusalReason.SIZING_SCAN_CAP_EXCEEDED: "refused_ineligible",
+    BackfillRefusalReason.DEFINITION_CHANGED: "refused_transient",
+}
+
 
 # acks_late: a countdown message acked on receipt sits in one worker's memory for the whole window
 # and dies with it, after the edit's supersession already ran. The creators refuse duplicates, so
@@ -972,19 +999,24 @@ def trigger_cohort_backfill_run_task(team_id: int, cohort_id: int, trigger_kind:
                 missing=missing,
             )
             return
-        run = (
-            create_person_backfill_run_for_cohort(team_id, cohort_id, trigger_kind)
+        attempt = (
+            attempt_person_backfill_run_for_cohort(team_id, cohort_id, trigger_kind)
             if person
-            else create_backfill_run_for_cohort(team_id, cohort_id, trigger_kind)
+            else attempt_backfill_run_for_cohort(team_id, cohort_id, trigger_kind)
         )
+        run = attempt.run
         if run is None:
-            COHORT_BACKFILL_TRIGGER_TASK_COUNTER.labels(backfill_kind=backfill_kind, outcome="refused").inc()
+            # An unmapped reason falls back to the old flat `refused` rather than vanishing, so
+            # adding an enum member can never silently drop refusals out of the metric.
+            outcome = COHORT_BACKFILL_REFUSAL_OUTCOMES.get(attempt.reason, "refused") if attempt.reason else "refused"
+            COHORT_BACKFILL_TRIGGER_TASK_COUNTER.labels(backfill_kind=backfill_kind, outcome=outcome).inc()
             logger.info(
                 "skipping_cohort_backfill_run_task",
                 cohort_id=cohort_id,
                 team_id=team_id,
                 trigger_kind=trigger_kind,
                 backfill_kind=backfill_kind,
+                refusal_reason=attempt.reason,
             )
             return
         COHORT_BACKFILL_TRIGGER_TASK_COUNTER.labels(backfill_kind=backfill_kind, outcome="created").inc()
@@ -1010,6 +1042,17 @@ def trigger_cohort_backfill_run_task(team_id: int, cohort_id: int, trigger_kind:
         raise
 
 
+@shared_task(ignore_result=True)
+def publish_cohort_backfill_run_gauges() -> None:
+    """Publish backfill run/chunk state for alerting.
+
+    Ungated on purpose, unlike ``finalize_cohort_backfill_runs``: a stalled run has to be alertable
+    while the finalizer is still dark, and the seeder's own metrics cannot see the transitions
+    Django owns.
+    """
+    publish_backfill_run_gauges()
+
+
 @shared_task(ignore_result=True)
 def finalize_cohort_backfill_runs() -> None:
     """Terminalize behavioral backfill runs the Rust seeder has fully observed. Gated off by

posthog/tasks/scheduled.py

@@ -11,7 +11,7 @@
 from posthog.clickhouse.client.execute_async import QueryStatusManager
 from posthog.tasks.ai_observability_usage_report import send_ai_observability_usage_reports
 from posthog.tasks.auth_token_cache_verification import verify_and_fix_auth_token_cache_task
-from posthog.tasks.calculate_cohort import finalize_cohort_backfill_runs
+from posthog.tasks.calculate_cohort import finalize_cohort_backfill_runs, publish_cohort_backfill_run_gauges
 from posthog.tasks.email import (
     EXTERNAL_DATA_DIGEST_DAY_BOUNDARY_HOUR_UTC,
     send_hog_functions_daily_digest,
@@ -698,6 +698,13 @@ def setup_periodic_tasks(sender: Celery, **kwargs: Any) -> None:
         name="finalize cohort backfill runs",
     )
 
+    add_periodic_task_with_expiry(
+        sender,
+        crontab(minute="*/2"),
+        publish_cohort_backfill_run_gauges.s(),
+        name="publish cohort backfill run gauges",
+    )
+
     add_periodic_task_with_expiry(
         sender,
         crontab(minute="*/2"),

posthog/tasks/test/__snapshots__/test_task_registration.ambr

@@ -18,6 +18,7 @@
     'posthog.tasks.calculate_cohort.insert_cohort_from_feature_flag',
     'posthog.tasks.calculate_cohort.insert_cohort_from_filters',
     'posthog.tasks.calculate_cohort.insert_cohort_from_query',
+    'posthog.tasks.calculate_cohort.publish_cohort_backfill_run_gauges',
     'posthog.tasks.calculate_cohort.trigger_cohort_backfill_run_task',
     'posthog.tasks.comment_slack_sync.backfill_comment_slack_thread',
     'posthog.tasks.comment_slack_sync.ingest_slack_discussion_reply',

posthog/tasks/test/test_calculate_cohort.py

@@ -15,6 +15,8 @@
 
 from posthog.exceptions import ClickHouseAtCapacity
 from posthog.tasks.calculate_cohort import (
+    COHORT_BACKFILL_REFUSAL_OUTCOMES,
+    COHORT_BACKFILL_TRIGGER_TASK_COUNTER,
     COHORT_STUCK_COUNT_GAUGE,
     COHORTS_STALE_COUNT_GAUGE,
     COHORTS_TOTAL_GAUGE,
@@ -31,6 +33,8 @@
 )
 from posthog.test.persons import create_person
 
+from products.cohorts.backend.backfill.runs import BackfillRefusalReason
+from products.cohorts.backend.backfill.sizing import PersonSeedEstimate
 from products.cohorts.backend.models.backfill import CohortBackfillKind, CohortBackfillRun
 from products.cohorts.backend.models.cohort import Cohort, CohortType
 from products.cohorts.backend.models.util import count_cohort_members, list_cohort_member_ids
@@ -1165,6 +1169,54 @@ def test_trigger_backfill_run_task_skips_when_the_person_budget_is_unset(self) -
         estimate.assert_not_called()
         self.assertFalse(CohortBackfillRun.objects.for_team(self.team.pk).exists())
 
+    def test_trigger_backfill_run_task_labels_a_budget_refusal_apart_from_a_blocked_slot(self) -> None:
+        # These two refusals call for opposite operator responses — raise the budget vs. go unwedge
+        # a stuck run — so the single flat `refused` outcome could not drive either alert.
+        cohort = self._backfillable_cohort()
+
+        with (
+            patch.object(COHORT_BACKFILL_TRIGGER_TASK_COUNTER, "labels") as counter,
+            override_settings(**{**BACKFILL_TASK_SETTINGS, "BEHAVIORAL_BACKFILL_PERSON_TOPIC_BYTES_BUDGET": 1}),
+            patch("products.cohorts.backend.backfill.runs.estimate_person_seed_topic_bytes") as estimate,
+        ):
+            estimate.return_value = PersonSeedEstimate(
+                estimated_persons=10,
+                pinned_condition_count=1,
+                bytes_per_seed=294,
+                estimated_topic_bytes=2_940,
+                budget_bytes=1,
+            )
+            trigger_cohort_backfill_run_task(
+                self.team.pk, cohort.pk, "cohort_created", CohortBackfillKind.PERSON_PROPERTY.value
+            )
+
+        counter.assert_called_once_with(
+            backfill_kind=CohortBackfillKind.PERSON_PROPERTY.value, outcome="refused_over_budget"
+        )
+
+        with override_settings(**BACKFILL_TASK_SETTINGS):
+            trigger_cohort_backfill_run_task(
+                self.team.pk, cohort.pk, "cohort_created", CohortBackfillKind.BEHAVIORAL.value
+            )
+            with patch.object(COHORT_BACKFILL_TRIGGER_TASK_COUNTER, "labels") as counter:
+                trigger_cohort_backfill_run_task(
+                    self.team.pk, cohort.pk, "cohort_edited", CohortBackfillKind.BEHAVIORAL.value
+                )
+
+        counter.assert_called_once_with(
+            backfill_kind=CohortBackfillKind.BEHAVIORAL.value, outcome="refused_slot_occupied"
+        )
+
+    def test_every_backfill_refusal_reason_has_a_trigger_outcome(self) -> None:
+        # A reason with no mapping entry falls back to the flat `refused`, which reads as an
+        # unclassified refusal rather than an omission. Catch the drift here instead.
+        self.assertEqual(set(COHORT_BACKFILL_REFUSAL_OUTCOMES), set(BackfillRefusalReason))
+        # The alert rules match these literals, so a rename has to break a test, not a dashboard.
+        self.assertEqual(
+            set(COHORT_BACKFILL_REFUSAL_OUTCOMES.values()),
+            {"refused_over_budget", "refused_slot_occupied", "refused_ineligible", "refused_transient"},
+        )
+
     def test_trigger_backfill_run_task_rechecks_the_allowlist_at_execution_time(self) -> None:
         # Tasks sit in the queue for the debounce countdown, so an operator shrinking the allowlist
         # during an incident has to stop those too, not only new enqueues.

products/cohorts/backend/backfill/finalize.py

@@ -69,6 +69,9 @@
     # labels: "shortfall" (an outcome was missing), "error" (the pass raised), "gated" (a person
     # run parked behind the readiness gate)
     ["reason"],
+    # This is a whole-fleet snapshot served from each worker pod's own registry, so `max` holds a
+    # drained reason high until the task lands again on the pod that wrote the high reading.
+    # `observe.py` pushes its equivalent gauges under one job name instead; this one wants the same.
     multiprocess_mode="max",
 )
 

products/cohorts/backend/backfill/observe.py

@@ -0,0 +1,161 @@
+"""Publishes backfill run and chunk state as Prometheus gauges.
+
+The seeder owns the richest telemetry about a run — per chunk, per partition, reconcile liveness —
+but all of it is the seeder's own view: counters that restart with the pod, and blind to every
+transition Django owns. Supersession, cancellation, a `blocked` run parked on a missing attestation,
+and the finalizer's own moves never reach it. These gauges read the runs table instead, so a stalled
+run or a chunk that exhausted its retries is visible whichever side put it there, and survives a
+seeder deploy.
+
+Read the age gauge with the finalizer's state in mind: while `BEHAVIORAL_BACKFILL_FINALIZER_ENABLED`
+is off, a behavioral run that finishes seeding parks in `reconciling`, which counts as active, so
+that slice climbs forever for work that is already done. Until the finalizer is on, only the
+`seeding`, `awaiting_boundary` and `blocked` slices carry an alertable age.
+
+Deliberately not folded into ``finalize_backfill_runs``: that returns early while the finalizer is
+off, and the alerts have to work with it still dark.
+"""
+
+from dataclasses import field
+from datetime import timedelta
+
+from django.db.models import Count, Min
+from django.utils import timezone as django_timezone
+
+from prometheus_client import CollectorRegistry, Gauge
+
+from posthog.dataclasses import frozen
+from posthog.metrics import pushed_metrics_registry
+
+from products.cohorts.backend.models.backfill import (
+    ACTIVE_COHORT_BACKFILL_RUN_STATUSES,
+    TERMINAL_COHORT_BACKFILL_RUN_STATUSES,
+    CohortBackfillChunk,
+    CohortBackfillChunkStatus,
+    CohortBackfillKind,
+    CohortBackfillRun,
+)
+
+RECENT_WINDOW = timedelta(hours=1)
+
+PUSH_JOB_NAME = "cohort_backfill_observe"
+
+_KINDS = tuple(CohortBackfillKind)
+
+
+@frozen
+class ObservationPass:
+    """One pass's readings of the runs table, before they are published."""
+
+    active_runs: dict[tuple[str, str], int] = field(default_factory=dict)
+    oldest_active_age_seconds: dict[tuple[str, str], float] = field(default_factory=dict)
+    failed_chunks: dict[str, int] = field(default_factory=dict)
+    recent_runs: dict[tuple[str, str], int] = field(default_factory=dict)
+
+
+def publish_backfill_run_gauges() -> ObservationPass:
+    """One observation pass over every team's backfill runs and chunks."""
+    result = _observe()
+    # Pushed, rather than served from the worker's own registry: this is a whole-fleet snapshot, and
+    # whichever worker the task lands on is the only one holding it. Every other worker keeps serving
+    # whatever it last wrote, and no multiprocess mode reaches across pods, so a `max by (status,
+    # kind)` alert stays lit until the task comes back round to the pod with the stale reading. One
+    # push under one job name replaces the whole group atomically instead. The tradeoff is this
+    # helper's usual one: if beat stops, the pushgateway keeps serving the last push, which
+    # `push_time_seconds` covers.
+    with pushed_metrics_registry(PUSH_JOB_NAME) as registry:
+        _publish(result, registry)
+    return result
+
+
+def _observe() -> ObservationPass:
+    now = django_timezone.now()
+    result = ObservationPass()
+
+    # Seed every combination with a zero first, so each pass pushes a complete group. A slice left
+    # out while it is empty vanishes from the push the moment it drains, and an alert expression
+    # reading it goes from a value to no data rather than to zero.
+    for kind in _KINDS:
+        result.failed_chunks[kind] = 0
+        for status in ACTIVE_COHORT_BACKFILL_RUN_STATUSES:
+            result.active_runs[(status, kind)] = 0
+            result.oldest_active_age_seconds[(status, kind)] = 0.0
+        for status in TERMINAL_COHORT_BACKFILL_RUN_STATUSES:
+            result.recent_runs[(status, kind)] = 0
+
+    # Cross-team on purpose, like the finalizer: these gauges serve the whole fleet, and there is no
+    # per-team consumer. No index serves a team-less status predicate — `cohort_bfr_team_status_idx`
+    # is team-prefixed, which is the same reason the finalizer needed `cohort_bfr_reconciling_idx`.
+    # Fine while the trigger allowlist is narrow; revisit with a partial index if that widens.
+    for row in (
+        CohortBackfillRun.objects.unscoped()
+        .filter(status__in=ACTIVE_COHORT_BACKFILL_RUN_STATUSES)
+        .values("status", "backfill_kind")
+        .annotate(runs=Count("id"), oldest=Min("created_at"))
+    ):
+        key = (row["status"], row["backfill_kind"])
+        result.active_runs[key] = row["runs"]
+        result.oldest_active_age_seconds[key] = max(0.0, (now - row["oldest"]).total_seconds())
+
+    for row in (
+        CohortBackfillChunk.objects.unscoped()
+        .filter(
+            status=CohortBackfillChunkStatus.FAILED,
+            run__status__in=ACTIVE_COHORT_BACKFILL_RUN_STATUSES,
+        )
+        .values("run__backfill_kind")
+        .annotate(chunks=Count("id"))
+    ):
+        result.failed_chunks[row["run__backfill_kind"]] = row["chunks"]
+
+    for row in (
+        CohortBackfillRun.objects.unscoped()
+        .filter(status__in=TERMINAL_COHORT_BACKFILL_RUN_STATUSES, finished_at__gte=now - RECENT_WINDOW)
+        .values("status", "backfill_kind")
+        .annotate(runs=Count("id"))
+    ):
+        result.recent_runs[(row["status"], row["backfill_kind"])] = row["runs"]
+
+    return result
+
+
+def _publish(result: ObservationPass, registry: CollectorRegistry) -> None:
+    runs_active = Gauge(
+        "posthog_cohort_backfill_runs_active",
+        "Backfill runs in each active status, by backfill kind",
+        ["status", "kind"],
+        registry=registry,
+    )
+    oldest_active_run_age = Gauge(
+        "posthog_cohort_backfill_oldest_active_run_age_seconds",
+        "Age of the oldest backfill run in each active status, by backfill kind",
+        ["status", "kind"],
+        registry=registry,
+    )
+    chunks_failed = Gauge(
+        "posthog_cohort_backfill_chunks_failed",
+        # Retry churn, not wedged work: a chunk under the attempt cap is reclaimed and retried, and
+        # one past it takes its whole run out of the active set within a poll. A run wedged that way
+        # surfaces as `runs_recent{status="failed"}`.
+        "Chunks in `failed` status on a still-active run, by backfill kind — retry activity",
+        ["kind"],
+        registry=registry,
+    )
+    runs_recent = Gauge(
+        "posthog_cohort_backfill_runs_recent",
+        # The one to alert on, not `runs_finalized_total`: supersession, the readiness supersede and
+        # the seeder's own `fail_run` all terminalize a run without the finalizer counting it, so
+        # that counter measures finalizer throughput while this measures run outcomes.
+        "Backfill runs terminalized in the last hour, by terminal status and backfill kind",
+        ["status", "kind"],
+        registry=registry,
+    )
+
+    for (status, kind), runs in result.active_runs.items():
+        runs_active.labels(status=status, kind=kind).set(runs)
+    for (status, kind), age in result.oldest_active_age_seconds.items():
+        oldest_active_run_age.labels(status=status, kind=kind).set(age)
+    for kind, chunks in result.failed_chunks.items():
+        chunks_failed.labels(kind=kind).set(chunks)
+    for (status, kind), runs in result.recent_runs.items():
+        runs_recent.labels(status=status, kind=kind).set(runs)