refactor(cohorts): dedicated reconcile-completion marker topic

products/cohorts/backend/management/commands/compare_cohort_membership.py

@@ -35,6 +35,7 @@
 from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
+from itertools import chain
 from typing import Any, Optional
 from uuid import UUID
 
@@ -47,7 +48,13 @@
 from products.cohorts.backend.parity.classifier import ClassifierConfig, CohortComparison, classify_cohort, summarize
 from products.cohorts.backend.parity.eligibility import EMITTING_CLASSES, ScreenedCohort, screen_team
 from products.cohorts.backend.parity.fold import fold_membership_changes, members, reconcile_completeness_by_cohort
-from products.cohorts.backend.parity.kafka_io import DEFAULT_SHADOW_TOPIC, DrainStats, consumer_config, drain_topic
+from products.cohorts.backend.parity.kafka_io import (
+    DEFAULT_MARKER_TOPIC,
+    DEFAULT_SHADOW_TOPIC,
+    DrainStats,
+    consumer_config,
+    drain_topic,
+)
 from products.cohorts.backend.parity.oracle import (
     OracleSetTooLarge,
     load_day_counts,
@@ -390,7 +397,10 @@ def add_arguments(self, parser: CommandParser) -> None:
             help="old-pipeline only: O-bounded raw diff, no R-FRESH/R-STALE rules or suspect probe",
         )
         parser.add_argument("--shadow-topic", type=str, default=DEFAULT_SHADOW_TOPIC)
-        parser.add_argument("--new-kafka-hosts", type=str, default=None, help="Override shadow-topic bootstrap servers")
+        parser.add_argument("--marker-topic", type=str, default=DEFAULT_MARKER_TOPIC)
+        parser.add_argument(
+            "--new-kafka-hosts", type=str, default=None, help="Override bootstrap servers for both drained topics"
+        )
         parser.add_argument("--security-protocol", type=str, default=None, help="Override Kafka security protocol")
         parser.add_argument("--format", choices=["table", "json"], default="table")
         parser.add_argument("--max-messages", type=int, default=None, help="Cap on shadow messages drained")
@@ -479,14 +489,23 @@ def log(message: str) -> None:
             hosts_override=options["new_kafka_hosts"],
             security_protocol_override=options["security_protocol"],
         )
-        log(f"draining {options['shadow_topic']} from {since.isoformat()} via {config['bootstrap.servers']}")
+        log(
+            f"draining {options['shadow_topic']} and {options['marker_topic']} from {since.isoformat()} "
+            f"via {config['bootstrap.servers']}"
+        )
         drain_stats = DrainStats()
-        messages = drain_topic(
-            options["shadow_topic"],
-            config=config,
-            since=since,
-            stats=drain_stats,
-            max_messages=options["max_messages"],
+        # Reconcile markers ride their own topic, so completeness needs a second drain chained into
+        # the same fold. Without it every cohort reads as having no complete reconcile run, and the
+        # recompute and population oracles caveat their own numbers as unmeasurable.
+        messages = chain(
+            drain_topic(
+                options["shadow_topic"],
+                config=config,
+                since=since,
+                stats=drain_stats,
+                max_messages=options["max_messages"],
+            ),
+            drain_topic(options["marker_topic"], config=config, since=since, stats=DrainStats()),
         )
         # An explicit --at pins the comparison clock, so the fold has to converge to that instant too.
         new_state, fold_stats = fold_membership_changes(

products/cohorts/backend/migrations/0010_alter_cohortbackfillrun_marker_watch.py

@@ -0,0 +1,21 @@
+# Generated by Django 5.2.14 on 2026-07-31 20:33
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("cohorts", "0009_cohort_backfill_per_kind_uniqueness"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="cohortbackfillrun",
+            name="marker_watch",
+            field=models.JSONField(
+                blank=True,
+                help_text='Opaque watcher resume state written by the Rust seeder, shape {"schema":2,"topic":"...","positions":{...},"ends":{...}|null}. Django never interprets it.',
+                null=True,
+            ),
+        ),
+    ]

products/cohorts/backend/migrations/max_migration.txt

@@ -1 +1 @@
-0009_cohort_backfill_per_kind_uniqueness
+0010_alter_cohortbackfillrun_marker_watch

products/cohorts/backend/models/backfill.py

@@ -79,7 +79,7 @@ class CohortBackfillRun(TeamScopedRootMixin, UUIDTModel):
         blank=True,
         help_text=(
             "Opaque watcher resume state written by the Rust seeder, shape "
-            '{"schema":1,"positions":{...},"ends":{...}|null}. Django never interprets it.'
+            '{"schema":2,"topic":"...","positions":{...},"ends":{...}|null}. Django never interprets it.'
         ),
     )
     blocked_reason = models.TextField(blank=True, default="")

products/cohorts/backend/parity/fold.py

@@ -1,4 +1,4 @@
-"""Fold a shadow-topic message stream into converged per-cohort membership state.
+"""Fold a drained message stream into converged per-cohort membership state.
 
 Max last_updated per (cohort_id, person_id) wins, matching the argMax convergence of the
 old side's ClickHouse table. In practice that is also arrival order (messages are keyed

products/cohorts/backend/parity/kafka_io.py

@@ -1,4 +1,4 @@
-"""Kafka profile resolution and a bounded shadow-topic drain.
+"""Kafka profile resolution and a bounded topic drain.
 
 The drain reads each partition from `offsets_for_times(--since)` up to a high-watermark
 snapshot taken at start, so a still-producing topic terminates. Uses a throwaway
@@ -21,6 +21,8 @@
 from posthog.kafka_client.routing import get_profile_settings
 
 DEFAULT_SHADOW_TOPIC = "cohort_membership_changed_shadow"
+# Reconcile completion markers ride their own topic, on the same cluster as the shadow topic.
+DEFAULT_MARKER_TOPIC = "cohort_reconcile_markers"
 
 
 @dataclass
@@ -42,10 +44,10 @@ def consumer_config(
     hosts_override: Optional[str] = None,
     security_protocol_override: Optional[str] = None,
 ) -> dict[str, Any]:
-    """Consumer config for the shadow topic's cluster (Warpstream ingestion).
+    """Consumer config for the shadow and marker topics' cluster (Warpstream ingestion).
 
-    The shadow topic is not in the Django routing table, so the INGESTION profile is
-    resolved explicitly; overrides serve local runs against a dev stack.
+    Neither topic is in the Django routing table, so the INGESTION profile is resolved
+    explicitly; overrides serve local runs against a dev stack.
     """
     profile = get_profile_settings(profile=KafkaClusterProfile.INGESTION)
     hosts = hosts_override or ",".join(profile.hosts)

rust/cohort-core/src/seed/reconcile.rs

@@ -309,9 +309,9 @@ fn deserialize_schema_version<'de, D: Deserializer<'de>>(deserializer: D) -> Res
 pub(super) const RECONCILE_COMPLETE_KIND: &str = "reconcile_complete";
 
 /// A completion certificate emitted after one partition's reconcile snapshot is durable. Produced by
-/// the stream processor onto the membership topic and folded by the seeder's marker watcher, so it
-/// lives here — the shared seed contract — rather than in either crate. Field order is the wire order;
-/// the golden test below pins the exact bytes both ends depend on.
+/// the stream processor onto the dedicated reconcile-marker topic and folded by the seeder's marker
+/// watcher, so it lives here — the shared seed contract — rather than in either crate. Field order is
+/// the wire order; the golden test below pins the exact bytes both ends depend on.
 ///
 /// Deliberately carries no dispatch epoch. A marker is a run-scoped durable fact: that partition's
 /// snapshot drained under this run's pinned filters. If a re-dispatch's watcher folds a late marker

rust/cohort-seeder/src/app/completion.rs

@@ -68,6 +68,56 @@ use super::watch::{WatchDirective, WatchDirectives};
 /// Timeout for the blocking Kafka metadata and watermark calls the dispatch makes.
 const KAFKA_METADATA_TIMEOUT: Duration = Duration::from_secs(10);
 
+/// Prove the marker topic exists before anything anchors on it: the daemon before it arms either
+/// completion half, the CLI before it claims a run. [`dispatch_and_record`]'s first act is to capture
+/// this topic's watermarks and it owns the claim by then, so an unprovisioned topic would otherwise
+/// park the run in `reconciling` with no dispatch record. Shared by both binaries so the diagnosis
+/// and the guidance stay the same wherever an operator meets them.
+pub async fn verify_marker_topic(
+    producer: &SeedTileProducer,
+    topic: &str,
+) -> Result<(), MarkerTopicError> {
+    let probe = producer.clone();
+    let probed = topic.to_string();
+    let joined = tokio::task::spawn_blocking(move || {
+        probe.verify_topic_reachable(&probed, KAFKA_METADATA_TIMEOUT)
+    })
+    .await;
+    match joined {
+        Err(source) => Err(MarkerTopicError::Join {
+            topic: topic.to_string(),
+            source,
+        }),
+        Ok(Err(source)) => Err(MarkerTopicError::Unreachable {
+            topic: topic.to_string(),
+            source,
+        }),
+        Ok(Ok(())) => Ok(()),
+    }
+}
+
+/// Why the marker-topic preflight failed. The operator guidance rides the error rather than each call
+/// site, so the daemon and the CLI cannot drift apart on what to do about it.
+#[derive(Debug, thiserror::Error)]
+pub enum MarkerTopicError {
+    #[error("joining the {topic:?} reachability probe")]
+    Join {
+        topic: String,
+        #[source]
+        source: JoinError,
+    },
+    #[error(
+        "reconcile marker topic {topic:?} is unreachable. Provision it before dispatching a run or \
+         arming either completion half, or leave SEEDER_RECONCILE_AUTO_DISPATCH_ENABLED and \
+         SEEDER_RECONCILE_OBSERVER_ENABLED off to boot without it."
+    )]
+    Unreachable {
+        topic: String,
+        #[source]
+        source: CaptureOffsetsError,
+    },
+}
+
 /// Whether automatic dispatch is armed. `Enabled` carries the register-backfill attestation, so the
 /// driver cannot be constructed without it.
 #[derive(Debug, Clone, Copy)]
@@ -157,7 +207,7 @@ enum DispatchKind {
 struct DispatchArm {
     pool: PgPool,
     producer: SeedTileProducer,
-    membership_topic: Arc<str>,
+    marker_topic: Arc<str>,
     max_inflight: NonZeroUsize,
     /// Caps how many dispatch tasks this replica runs at once. Every task produces
     /// `cohorts × COHORT_PARTITION_COUNT` tiles through the producer the chunk pipeline shares, so
@@ -185,6 +235,9 @@ pub struct CompletionDriver {
     /// The backfill kinds discovery binds. `[Behavioral]` unless the person seed path is on, which
     /// keeps a person run invisible to this protocol while its gate is off.
     kinds: &'static [RunKind],
+    /// The marker topic every persisted watch is anchored to. A run whose record names a different
+    /// one classifies as [`UndispatchedReason::TopicChanged`] and re-dispatches.
+    marker_topic: String,
     dispatch: Option<DispatchArm>,
     observe: Option<ObserveArm>,
     in_flight: Arc<Mutex<HashSet<RunId>>>,
@@ -196,11 +249,17 @@ pub struct CompletionDriver {
 
 impl CompletionDriver {
     /// Base driver with neither half armed. `main` adds the halves the config enables.
-    pub fn new(pool: PgPool, allowlist: TeamAllowlist, kinds: &'static [RunKind]) -> Self {
+    pub fn new(
+        pool: PgPool,
+        allowlist: TeamAllowlist,
+        kinds: &'static [RunKind],
+        marker_topic: String,
+    ) -> Self {
         Self {
             pool,
             allowlist,
             kinds,
+            marker_topic,
             dispatch: None,
             observe: None,
             in_flight: Arc::new(Mutex::new(HashSet::new())),
@@ -212,15 +271,14 @@ impl CompletionDriver {
     pub fn with_dispatch(
         mut self,
         producer: SeedTileProducer,
-        membership_topic: String,
         max_inflight: NonZeroUsize,
         max_concurrent_dispatches: NonZeroUsize,
         register_backfill: RegisterBackfillConfirmation,
     ) -> Self {
         self.dispatch = Some(DispatchArm {
             pool: self.pool.clone(),
             producer,
-            membership_topic: Arc::from(membership_topic),
+            marker_topic: Arc::from(self.marker_topic.as_str()),
             max_inflight,
             dispatch_slots: Arc::new(Semaphore::new(max_concurrent_dispatches.get())),
             register_backfill,
@@ -237,7 +295,10 @@ impl CompletionDriver {
         directives: watch::Sender<WatchDirectives>,
     ) -> Self {
         self.observe = Some(ObserveArm {
-            store: Arc::new(PgObservationStore::new(self.pool.clone())),
+            store: Arc::new(PgObservationStore::new(
+                self.pool.clone(),
+                self.marker_topic.clone(),
+            )),
             committed,
             topic_ends,
             directives,
@@ -249,13 +310,16 @@ impl CompletionDriver {
     /// and observe the reconciling ones (observe arm). A discovery failure is logged and retried next
     /// tick; the dark path (driver absent) costs zero queries.
     pub async fn tick(&self) {
-        let discovered = match discover_completions(&self.pool, &self.allowlist, self.kinds).await {
-            Ok(discovered) => discovered,
-            Err(error) => {
-                warn!(error = %error, "completion discovery failed");
-                return;
-            }
-        };
+        let discovered =
+            match discover_completions(&self.pool, &self.allowlist, self.kinds, &self.marker_topic)
+                .await
+            {
+                Ok(discovered) => discovered,
+                Err(error) => {
+                    warn!(error = %error, "completion discovery failed");
+                    return;
+                }
+            };
 
         let now = Utc::now();
         // Per kind, not summed: "person runs are piling up undispatched" is the failure this protocol
@@ -343,7 +407,8 @@ impl CompletionDriver {
                     // read as stalled the instant the dispatch record breaks.
                     track_oldest(&mut oldest_stalled, since);
                     if first_sighting {
-                        counter!(RECONCILE_RECORD_INVALID).increment(1);
+                        counter!(RECONCILE_RECORD_INVALID, "reason" => reason.as_str())
+                            .increment(1);
                     }
                     // Only the dispatch arm can heal an undispatched run; observe-only leaves it for a
                     // CLI re-dispatch.
@@ -753,7 +818,7 @@ async fn run_dispatch(context: &DispatchArm, run_id: RunId, run_kind: RunKind, k
     match dispatch_and_record(
         &context.pool,
         &context.producer,
-        &context.membership_topic,
+        &context.marker_topic,
         context.max_inflight,
         KAFKA_METADATA_TIMEOUT,
         certified,
@@ -802,15 +867,15 @@ pub struct RecordedDispatch {
 pub async fn dispatch_and_record(
     pool: &PgPool,
     producer: &SeedTileProducer,
-    membership_topic: &str,
+    marker_topic: &str,
     max_inflight: NonZeroUsize,
     metadata_timeout: Duration,
     certified: CertifiedDispatch,
     claim: ReconcilingClaim,
 ) -> Result<RecordedDispatch, DispatchRecordError> {
     let positions = {
         let producer = producer.clone();
-        let topic = membership_topic.to_string();
+        let topic = marker_topic.to_string();
         tokio::task::spawn_blocking(move || {
             producer.capture_topic_offsets(&topic, metadata_timeout)
         })
@@ -830,6 +895,7 @@ pub async fn dispatch_and_record(
 
     let hwms = ReconcileHwms::try_from(&receipt).map_err(DispatchRecordError::Hwms)?;
     let watch = MarkerWatch {
+        topic: marker_topic.to_string(),
         positions,
         ends: None,
     };
@@ -857,7 +923,7 @@ impl TryFrom<&ReconcileDispatchReceipt> for ReconcileHwms {
 pub enum DispatchRecordError {
     #[error("joining the offset-capture task")]
     CaptureTask(#[source] JoinError),
-    #[error("capturing the membership topic start positions")]
+    #[error("capturing the marker topic start positions")]
     CaptureOffsets(#[source] CaptureOffsetsError),
     #[error("producing the reconcile control tiles")]
     Execute(#[source] ReconcileDispatchError),