feat(cohorts): sweep membership rows a completed reconcile did not assert

nodejs/src/cdp/config.ts

@@ -186,6 +186,17 @@ export type CdpConfig = ClickhouseConfig & {
     CYCLOTRON_NODE_RESCHEDULE_MAX_CHUNKS_PER_CALL: number
     CYCLOTRON_NODE_RESCHEDULE_CHUNK_SLEEP_MS: number
 
+    // Mark-and-sweep deletion of cohort_membership rows a completed reconcile run did not re-assert.
+    // Two flags so a rolling restart never mixes sweeping pods with pods that write no versions:
+    // version writes turn on first and off last, sweeping turns on last and off first. A pod that
+    // writes without versions preserves a row's old stamp, which a sweeping pod then reads as stale.
+    COHORT_MEMBERSHIP_VERSION_WRITES_ENABLED: boolean
+    COHORT_MEMBERSHIP_SWEEP_ENABLED: boolean
+    COHORT_MEMBERSHIP_SWEEP_INTERVAL_MS: number
+    COHORT_MEMBERSHIP_SWEEP_BATCH_SIZE: number
+    COHORT_MEMBERSHIP_SWEEP_CLAIM_TIMEOUT_MS: number
+    COHORT_MEMBERSHIP_SWEEP_ABANDON_AFTER_DAYS: number
+
     // Email reputation evaluator (daily Temporal-scheduled bounce/complaint snapshots for workflows email)
 }
 
@@ -345,5 +356,23 @@ export function getDefaultCdpConfig(): CdpConfig {
         CYCLOTRON_NODE_RESCHEDULE_CHUNK_SIZE: 5000,
         CYCLOTRON_NODE_RESCHEDULE_MAX_CHUNKS_PER_CALL: 20,
         CYCLOTRON_NODE_RESCHEDULE_CHUNK_SLEEP_MS: 100,
+
+        // Stays off until the schema migrations have applied; a pod reaching an unmigrated
+        // database must degrade to the pre-sweep write shape, not crash on a missing column.
+        COHORT_MEMBERSHIP_VERSION_WRITES_ENABLED: false,
+        // Stays off until the processor writes membership to the live topic and the whole fleet
+        // persists row versions. Sweeping before either would delete rows nothing re-asserts.
+        //
+        // The cohort stream pipeline (cohort-stream-processor and its topics, including the
+        // marker topic) is deployed US-only, like the rest of the cohort streaming system. This
+        // flag follows that scoping: environments without the pipeline keep it off, and the
+        // marker topic must never be created just to let the flag turn on — with no processor
+        // producing markers there, the sweeper would idle forever. start() fails fast when the
+        // flag is set and the marker topic does not exist.
+        COHORT_MEMBERSHIP_SWEEP_ENABLED: false,
+        COHORT_MEMBERSHIP_SWEEP_INTERVAL_MS: 60000,
+        COHORT_MEMBERSHIP_SWEEP_BATCH_SIZE: 1000,
+        COHORT_MEMBERSHIP_SWEEP_CLAIM_TIMEOUT_MS: 300000,
+        COHORT_MEMBERSHIP_SWEEP_ABANDON_AFTER_DAYS: 3,
     }
 }

nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.ts

@@ -9,7 +9,7 @@ import {
     KAFKA_PERSON,
     KAFKA_PERSON_DISTINCT_ID,
 } from '~/common/config/kafka-topics'
-import { KafkaConsumerInterface, RdKafkaConsumerConfig, createKafkaConsumer } from '~/common/kafka/consumer'
+import { KafkaConsumerInterface, START_AT_LATEST, createKafkaConsumer } from '~/common/kafka/consumer'
 import { InternalCaptureEvent } from '~/common/services/internal-capture'
 import { instrumentFn, instrumented } from '~/common/tracing/tracing-utils'
 import { parseJSON } from '~/common/utils/json-parse'
@@ -46,7 +46,7 @@ import { counterParseError } from './metrics'
 // volume) topic backlog. Once the group has committed offsets, both consumer impls resume from them
 // regardless of this value — so steady-state recovery is committed-offset resume; latest only governs
 // the first bootstrap and offset-loss edge cases (covered by the deferred lag alerting follow-up).
-const startAtLatest = { ['auto.offset.reset' as keyof RdKafkaConsumerConfig]: 'latest' as never }
+const startAtLatest = START_AT_LATEST
 
 const counterHogflowMatcherCandidatesEvaluated = new Counter({
     name: 'cdp_hogflow_matcher_candidates_evaluated',

nodejs/src/cdp/dev/cohort-sweep-e2e.test.ts

@@ -0,0 +1,136 @@
+import { KAFKA_COHORT_MEMBERSHIP_CHANGED, KAFKA_COHORT_RECONCILE_MARKERS } from '~/common/config/kafka-topics'
+import { KafkaProducerWrapper } from '~/common/kafka/producer'
+import { closeHub, createHub } from '~/common/utils/db/hub'
+import { PostgresUse } from '~/common/utils/db/postgres'
+import { UUIDT } from '~/common/utils/utils'
+import { createCdpConsumerDeps } from '~/tests/helpers/cdp'
+import { waitForExpect } from '~/tests/helpers/expectations'
+import { resetBehavioralCohortsDatabase } from '~/tests/helpers/sql'
+import { Hub } from '~/types'
+
+import { CdpCohortMembershipConsumer } from '../consumers/cdp-cohort-membership.consumer'
+import { COHORT_PARTITION_COUNT } from '../services/cohort-membership/sweeper.service'
+
+// Drives the whole chain against the running dev stack: real Kafka topics, the real membership
+// consumer, the real marker consumer, and the real sweep loop. Lives under dev/ so CI skips it.
+jest.setTimeout(180_000)
+
+describe('cohort membership mark and sweep, end to end', () => {
+    const TEAM_ID = 987654
+    const COHORT_ID = 5150
+    const SNAPSHOT_VERSION = '2026-08-26 10:00:00.000000'
+    const STALE_VERSION = '2026-08-26 09:00:00.000000'
+
+    let hub: Hub
+    let consumer: CdpCohortMembershipConsumer
+    let producer: KafkaProducerWrapper
+
+    afterEach(async () => {
+        await consumer?.stop()
+        await producer?.disconnect()
+        await closeHub(hub)
+    })
+
+    it('deletes rows the completed reconcile run did not re-assert', async () => {
+        const runId = new UUIDT().toString()
+        const stillMember = new UUIDT().toString()
+        const droppedOut = new UUIDT().toString()
+
+        hub = await createHub()
+        await resetBehavioralCohortsDatabase(hub.postgres)
+
+        // Someone who matched before the cohort was edited. No live transition ever arrives for
+        // them, so without a sweep this row claims membership forever.
+        await hub.postgres.query(
+            PostgresUse.BEHAVIORAL_COHORTS_RW,
+            `INSERT INTO cohort_membership (team_id, cohort_id, person_id, in_cohort, version)
+             VALUES ($1, $2, $3, true, $4::timestamp)`,
+            [TEAM_ID, COHORT_ID, droppedOut, STALE_VERSION],
+            'seedStaleRow'
+        )
+
+        consumer = new CdpCohortMembershipConsumer(
+            {
+                ...hub,
+                COHORT_MEMBERSHIP_VERSION_WRITES_ENABLED: true,
+                COHORT_MEMBERSHIP_SWEEP_ENABLED: true,
+                COHORT_MEMBERSHIP_SWEEP_INTERVAL_MS: 1000,
+            },
+            createCdpConsumerDeps(hub)
+        )
+        await consumer.start()
+
+        producer = await KafkaProducerWrapper.create(hub.KAFKA_CLIENT_RACK)
+
+        await producer.produce({
+            topic: KAFKA_COHORT_MEMBERSHIP_CHANGED,
+            key: Buffer.from(stillMember),
+            value: Buffer.from(
+                JSON.stringify({
+                    team_id: TEAM_ID,
+                    cohort_id: COHORT_ID,
+                    person_id: stillMember,
+                    last_updated: SNAPSHOT_VERSION,
+                    status: 'entered',
+                    origin: 'reconcile',
+                    run_id: runId,
+                })
+            ),
+        })
+        await producer.flush()
+
+        // The processor produces a marker only after its partition's membership rows are acked.
+        await waitForExpect(async () => {
+            const applied = await hub.postgres.query(
+                PostgresUse.BEHAVIORAL_COHORTS_RW,
+                'SELECT person_id FROM cohort_membership WHERE person_id = $1',
+                [stillMember],
+                'awaitSnapshot'
+            )
+            expect(applied.rows).toHaveLength(1)
+        }, 60_000)
+
+        for (let partition = 0; partition < COHORT_PARTITION_COUNT; partition++) {
+            await producer.produce({
+                topic: KAFKA_COHORT_RECONCILE_MARKERS,
+                key: Buffer.from(`${TEAM_ID}:${COHORT_ID}:${runId}:${partition}`),
+                value: Buffer.from(
+                    JSON.stringify({
+                        type: 'reconcile_complete',
+                        team_id: TEAM_ID,
+                        cohort_id: COHORT_ID,
+                        partition,
+                        run_id: runId,
+                        last_updated: SNAPSHOT_VERSION,
+                    })
+                ),
+            })
+        }
+        await producer.flush()
+
+        // Both checks sit in the same wait: the rows go before the ledger is marked done, so
+        // asserting the ledger separately would race the sweep's own last statement.
+        await waitForExpect(async () => {
+            const members = await hub.postgres.query(
+                PostgresUse.BEHAVIORAL_COHORTS_RW,
+                'SELECT person_id FROM cohort_membership WHERE team_id = $1 AND cohort_id = $2',
+                [TEAM_ID, COHORT_ID],
+                'readMembers'
+            )
+            expect(members.rows.map((row: { person_id: string }) => row.person_id)).toEqual([stillMember])
+
+            const sweeps = await hub.postgres.query(
+                PostgresUse.BEHAVIORAL_COHORTS_RW,
+                'SELECT marker_bits, status, swept_rows, snapshot_rows FROM cohort_membership_sweeps WHERE run_id = $1',
+                [runId],
+                'readSweep'
+            )
+            expect(sweeps.rows[0]).toMatchObject({
+                marker_bits: '-1',
+                status: 'swept',
+                swept_rows: '1',
+                snapshot_rows: '1',
+            })
+        }, 120_000)
+    })
+})