feat(cohorts): add version and sweep tables to cohort_membership

rust/behavioral_cohorts_migrations/20260826000001_add_version_to_cohort_membership.sql

@@ -0,0 +1,21 @@
+-- The pipeline's canonical last-writer-wins version for a membership row: the producer's
+-- monotonic `last_updated` stamp, the same value ClickHouse's ReplacingMergeTree orders on.
+-- `last_updated` here stays consumer-write-time, so it cannot serve that role.
+--
+-- `-infinity` means "older than anything the pipeline has stamped", which is what pre-feature
+-- rows are. NULL cannot carry that meaning: the sweep deletes rows matching `version < cutoff`,
+-- and NULL fails that test, so the rows the sweep exists to delete would be the ones it never
+-- sees. A constant default is catalog-only on PG 11+ (stored as attmissingval and synthesized at
+-- read time), so NOT NULL here costs no rewrite, no backfill, and no validation scan.
+--
+-- Readers that format this column must handle the sentinel: `to_char` returns NULL for an
+-- infinite timestamp instead of a formatted string.
+--
+-- The DDL is catalog-only, but it still takes a brief ACCESS EXCLUSIVE lock: queueing for it
+-- behind a long-running statement parks every subsequent reader behind us. lock_timeout bounds
+-- that wait; on timeout the per-file transaction aborts, nothing is recorded, and the next
+-- migration run retries this idempotent file.
+
+SET LOCAL lock_timeout = '5s';
+
+ALTER TABLE cohort_membership ADD COLUMN IF NOT EXISTS version TIMESTAMP NOT NULL DEFAULT '-infinity';

rust/behavioral_cohorts_migrations/20260826000002_create_sweep_tables.sql

@@ -0,0 +1,78 @@
+-- Coordination state for mark-and-sweep deletion of stale `cohort_membership` rows.
+--
+-- A reconcile run emits one `origin=reconcile` row per person holding a `cf_stage2` register row
+-- for the cohort, and then emits one completion marker per processor partition. Persons with no
+-- register row get nothing, and their `cohort_membership` rows are exactly what the sweep deletes.
+-- Once every partition's marker has arrived AND the membership consumer has passed the snapshot
+-- those markers certify, the cohort's rows older than the run can be deleted: the run just
+-- re-asserted everything that is still true.
+CREATE TABLE IF NOT EXISTS cohort_membership_sweeps (
+    run_id UUID NOT NULL,
+    cohort_id BIGINT NOT NULL,
+    team_id BIGINT NOT NULL,
+    -- i64 bitmap of the partitions whose completion marker has arrived, same convention as the
+    -- seeder's `reconcile_marker_bits`: all 64 bits set reads as -1. Bit union is monotone, so
+    -- marker redelivery is a no-op.
+    marker_bits BIGINT NOT NULL DEFAULT 0,
+    -- The lowest stamp carried by the run's own completion markers. Bounds how far the marker set
+    -- proves the run got, which is above the rows it emitted, never below.
+    min_marker_version TIMESTAMP,
+    -- The lowest stamp carried by the snapshot rows the run emitted. This is the sweep threshold:
+    -- rows below it were not re-asserted by the run. `TIMESTAMP`, not `TIMESTAMPTZ`, because both
+    -- sort against `cohort_membership.version`, which carries the producer's UTC stamp verbatim.
+    min_snapshot_version TIMESTAMP,
+    -- Observability only. Double-counts on replay, so it must never gate a sweep.
+    snapshot_rows BIGINT NOT NULL DEFAULT 0,
+    -- Membership-topic high watermarks at the moment the marker set completed, per partition.
+    membership_hwms JSONB,
+    -- Where those watermarks were read from. Offsets only mean something against the cluster and
+    -- topic that produced them, and both can move under a run: the consumer's brokers change, or
+    -- the processor's output topic flips from the shadow name to the real one. Without this the
+    -- gate would compare a captured watermark against progress from a different feed and pass
+    -- vacuously. Recording the provenance lets the sweep refuse a run it can no longer verify.
+    membership_cluster TEXT,
+    membership_topic TEXT,
+    status TEXT NOT NULL DEFAULT 'collecting',
+    -- The wall-clock columns are TIMESTAMPTZ so that they name the same instant on every pod. A
+    -- bare TIMESTAMP stores whatever the writing session's TimeZone produced, so two pods that
+    -- disagree on that setting would disagree about when a claim lease expired.
+    --
+    -- ready_at is when the marker set completed and the watermarks were captured. Never rewritten
+    -- after, so gate-wait observability survives partial sweeps that bounce the row back to
+    -- 'ready'.
+    ready_at TIMESTAMPTZ,
+    claimed_at TIMESTAMPTZ,
+    -- Fences the claiming pod: the heartbeat and the finish update require the token, so a
+    -- straggler whose claim timed out cannot overwrite the pod that reclaimed the run.
+    claim_token UUID,
+    swept_rows BIGINT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (run_id, cohort_id),
+    CONSTRAINT cohort_membership_sweeps_status_check
+        CHECK (status IN ('collecting', 'ready', 'sweeping', 'swept', 'abandoned'))
+);
+
+-- Every pod scans for claimable runs on each sweep tick.
+CREATE INDEX IF NOT EXISTS idx_cohort_membership_sweeps_status
+    ON cohort_membership_sweeps (status, created_at);
+
+-- How far the membership consumer group has actually applied each partition, aggregated across
+-- pods. A sweep waits until this passes the high watermarks its run captured, so it can never
+-- delete rows whose re-asserting snapshot messages are still in flight.
+CREATE TABLE IF NOT EXISTS cohort_membership_consumer_progress (
+    -- The feed the consumer read the offsets from: broker list plus topic. Offsets only mean
+    -- something against the exact feed that produced them. After a cluster move, rows keyed to
+    -- the old brokers would report large retained offsets against the new cluster's small
+    -- watermarks; after a topic rename on the same brokers, the old topic's offsets would stand
+    -- in for the new one's. Either way the gate would pass vacuously. Keying by both makes a
+    -- move start from no progress, which can only hold the gate closed, never open it early.
+    cluster TEXT NOT NULL,
+    topic TEXT NOT NULL,
+    partition INT NOT NULL,
+    -- The next offset to consume, not the last consumed one: Kafka's own committed-offset
+    -- convention, which makes the comparison against a high watermark exact.
+    next_offset BIGINT NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (cluster, topic, partition)
+);

rust/behavioral_cohorts_migrations/20260826000003_add_cohort_membership_sweep_index.sql

@@ -0,0 +1,29 @@
+-- no-transaction
+--
+-- The sweep pages through one cohort's rows below a version. The unique constraint on
+-- (team_id, cohort_id, person_id) can only prefix-match the cohort, leaving the version filter to
+-- discard most of what it reads; carrying version in the index lets each page stop early.
+--
+-- `id` is in the index because the pager orders by (version, id) and resumes from a
+-- (version, id) > (cursor) row comparison. Postgres turns a row comparison into an index
+-- condition only when every compared column is indexed; without `id` it truncates to
+-- `version >= cursor`, which bounds nothing when candidates share one version, and each page
+-- then sorts the whole remaining set. Every pre-feature row shares the '-infinity' default, so
+-- the first sweep after rollout is exactly that case.
+--
+-- CONCURRENTLY, alone in its own no-transaction file: a plain CREATE INDEX takes SHARE on
+-- cohort_membership, which conflicts with the ROW EXCLUSIVE every consumer upsert needs, and sqlx
+-- would hold it until the whole file commits. This table is on the membership consumer's hot write
+-- path, so a blocking build stalls every batch for the duration.
+--
+-- Recovery note: if this CONCURRENTLY build is ever interrupted, it leaves the index INVALID and a
+-- rerun's IF NOT EXISTS will NOT rebuild it. A drop-first statement can't fix that here, because
+-- sqlx runs a no-transaction migration as one implicitly-transactional batch, so CONCURRENTLY
+-- statements must be alone in their file, and a separate drop migration wouldn't re-run on retry.
+-- Nothing reports that state: the migration run that left it INVALID already failed, and every
+-- later run skips the file and reports success, so check for it before assuming the index is there.
+--   SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass('idx_cohort_membership_sweep');
+-- On `f`, recover manually, then re-run migrations:
+--   DROP INDEX CONCURRENTLY idx_cohort_membership_sweep;
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_cohort_membership_sweep
+    ON cohort_membership (team_id, cohort_id, version, id);

rust/behavioral_cohorts_migrations/20260826000004_tune_cohort_membership_autovacuum.sql

@@ -0,0 +1,15 @@
+-- The sweep index puts `version` under an index, and the membership consumer rewrites `version` on
+-- every message, so those upserts can no longer take the heap-only tuple path: each one writes a
+-- new heap tuple plus an entry in all four indexes, and leaves the old entries as dead tuples.
+-- Before the index, the upsert only changed `in_cohort` and `last_updated`, neither of them
+-- indexed, so it reused the heap page and touched no index at all.
+--
+-- Default autovacuum waits for 20% dead tuples, which is a large amount of bloat on a table the
+-- flag evaluation path reads under a 1000ms timeout. 2% keeps the index scans on live tuples.
+--
+-- SET (...) takes SHARE UPDATE EXCLUSIVE, which does not conflict with the consumer's writes.
+
+ALTER TABLE cohort_membership SET (
+    autovacuum_vacuum_scale_factor = 0.02,
+    autovacuum_analyze_scale_factor = 0.02
+);