Backfill for streaming behavioral cohorts

Problem

The streaming cohorts pipeline (cohort-event-shuffler + cohort-stream-processor) computes membership only from events it sees live. It has no memory of anything that happened before it started watching. Five real situations depend on history it can never observe live:

None of this is built today. The flag serving gate Cohort.is_flag_compatible already requires last_backfill_events_at to be set for a behavioral cohort, but nothing in production writes that column, so the gate stays permanently closed.

Core idea

We compute the historical state with the exact same evaluator the live processor uses, aggregate it into small mergeable tiles, and feed those tiles into the processor through a new topic. The processor merges them into its live state idempotently. Because batch and live can overlap, we make the two domains disjoint by construction and absorb any overlap with an idempotent merge.

Django (products/cohorts)                    ClickHouse events history
  detects edit / new cohort / enablement          |
  enqueues a pinned BackfillRun ---------> cohort-seeder (new Rust binary)
  compare-and-swap stamps readiness <--+     reuses the live Stage 1 crate + Rust HogVM
                                       |     scans team + day + event-name, evaluates each event
                                       |     aggregates matches into (person, condition, day) tiles
                                       |          |
                                       |     cohort_stream_seed_events (64p, same murmur2 key)
                                       |          |
                                       |     cohort-stream-processor (new 5th consumer)
                                       |          merges tiles into daily-bucket state (take the max)
                                       +---- reconcile: full membership snapshot when seeding finishes
                                                  |
                                             cohort_membership_changed(_shadow) -> harness / PG consumer

How it works

1. Seeder. A new small Rust binary scans ClickHouse once per run, filtered to the union of event names the team's cohorts reference. It evaluates every event with the same Stage 1 crate and Rust HogVM the processor runs live, so matching is identical by construction. It resolves each event to its canonical person at scan time (avoiding the stale person_id trap on the events row).

2. Day tiles. Instead of one message per matched event, the seeder aggregates matches into one tile per (person, condition_hash, day) carrying an absolute matched count for that day. This shrinks the payload roughly 6x versus raw events and is what makes a 30 day team scale backfill run in hours to a day rather than weeks.

3. Seed topic. Tiles go to a new cohort_stream_seed_events topic with 64 partitions and the same murmur2("{team}:{person}") key as the live topic, so a person's tiles always land on the partition that owns that person's state.

4. Processor 5th consumer. The processor gains a seed consumer that routes each tile to the owning partition worker. The worker merges the tile into the existing daily buckets by taking the max at that day index, then reruns the membership evaluation. It never touches the offset bookkeeping the live path uses (see correctness below).

5. Reconcile snapshot. When seeding finishes, the seeder emits a control message. Each partition worker scans its membership range for the cohort and emits the current membership for every person unconditionally. This heals any downstream drift, then the coordinator stamps readiness and the flag gate opens.

Why it is correct

The hard part is stitching batch history to the live stream without double counting. Four ideas carry it:

• Disjoint domains. Seeds cover only full days strictly before a pinned boundary, and only events that had already arrived at scan time. Live keeps counting everything from the boundary onward. The two sets do not overlap by construction.

• Idempotent merge. Tiles carry an absolute count and are applied with a max, so re-delivering any tile any number of times in any order is a no-op. Reordering does not matter.

• Apply fence. A tile is applied only after the worker's live consumption has passed the tile's scan point. This orders live before tile, so the only remaining overlap (a late arriving event) is absorbed by the max instead of being added twice. This is what protects the DR case, where a wiped processor replays backlog at the same time as seeding.

• Reconcile heals the pipeline's at most once output. The pipeline can commit correct state while a membership change fails to publish downstream, so a state versus state diff would see nothing to fix. The reconcile snapshot emits full membership unconditionally, so downstream converges via a simple mark and sweep keyed by run.

Readiness is only stamped with a conditional update guarded by the cohort's current filters_hash, so a run that was superseded by an edit mid flight can never open the gate for a definition it did not actually backfill.

Why these choices

• Day tiles, not raw replay. Measured on team 2 (2026-07-10): 1.9B events over 30 days but only 302M (person, event, day) triples. Raw replay through the processor at its demonstrated ceiling would take roughly 18 days of wall clock; tiles bring it to hours to a day.
• Same evaluator, not SQL. Reimplementing leaf matching in ClickHouse SQL would introduce a second evaluator, and every HogVM, timezone, or coercion divergence would become silent state corruption. This initiative exists precisely because such divergences are real, so the seeder runs the production evaluator and ClickHouse only pre filters by team, day, and event name.
• Not importing legacy membership. That would contaminate the parity comparison with the old pipeline's known errors.