Repository navigation
feat(cohorts): sweep membership rows a completed reconcile did not assert - #89797
Conversation
github-actions
Bot
added
feature/cohorts
👀 Auto-assigned reviewersThese soft owners were skipped because they only have minor changes here. Nothing blocks merge, so self-assign if you'd like a look:
Soft owners come from each directory's |
🤖 CI report✅ Trunk lane — non-backend laneThis PR is assigned to the non-backend lane. It does not run backend Python tests and may merge in parallel with PRs in other lanes. |
3ed6c3a to
6cb8d43
Compare
matheus-vb
changed the title
7358e7a to
356bf1f
Compare
🦔 PostHog Review reviewed this pull requestFound 0 must fix, 6 should fix, 2 consider. Published 8 findings (view the review). |
|
PostHog Review alpha 🦔 If you find any issues helpful - please reply "valid", "invalid", etc., for evaluation purposes 🙏 |
![posthog[bot]](https://avatars.githubusercontent.com/in/1203049?s=60&v=4){kind=small}
| WITH candidates AS ( | ||
| SELECT id FROM cohort_membership | ||
| WHERE team_id = $1 AND cohort_id = $2 | ||
| AND version < $3::timestamp | ||
| AND (version, id) > ($5::timestamp, $6::bigint) | ||
| ORDER BY version, id | ||
| LIMIT $4 | ||
| FOR UPDATE SKIP LOCKED | ||
| ) | ||
| DELETE FROM cohort_membership m | ||
| USING candidates c | ||
| WHERE m.id = c.id AND m.version < $3::timestamp | ||
| RETURNING to_char(m.version, 'YYYY-MM-DD HH24:MI:SS.US') AS version, m.id |
There was a problem hiding this comment.
A current versionless update can be swept as stale
Why we think it's a valid issue
- Checked: whether a versionless change can reach the upsert, what it leaves in the
versioncolumn, and whether the sweep predicate can then match that row. - Found: the mechanism is real.
upsertMembershipRowsatcdp-cohort-membership.consumer.ts:365-368setsversion = COALESCE(EXCLUDED.version, cohort_membership.version)and admits the write throughEXCLUDED.version IS NULL, so a versionlessenteredflipsin_cohortand refresheslast_updatedwhile the stale stamp survives. The sweep atsweeper.service.ts:489-490tests onlyteam_id,cohort_id, andversion < $3::timestamp— it never readsin_cohortorlast_updated— so that row is deleted even though it holds a current membership the run did not assert. The branch is live behavior the author tests, not dead code:cdp-cohort-membership.consumer.test.ts:341asserts "a versionless change applies without moving the version". - Found: this PR makes a stated invariant wrong. The comment at
cdp-cohort-membership.consumer.ts:505-508says an off-format stamp degrades to no version and "loses only the ordering: the membership transition itself still applies". With the sweep in place that is no longer true — the transition applies and is then silently deleted. The off-format valve exists precisely because the author judged format drift worth building for. - Found: no trigger exists today, which is what caps the severity. The topic has one producer:
CohortMembershipChange.last_updatedis a non-optionalStringatrust/cohort-stream-processor/src/producer/mod.rs:53, with noskip_serializing_if(unlikeoriginandrun_idtwo fields below), and every emitting path threads a stamp in —map_transitionat:118, pluscascade_path.rs:149,stage2_path.rs:110, andperson_seed_path.rs:511. Python only declares the topic and consumes it into ClickHouse;cohort-seedernever builds this struct. - Found: the downgrade route is closed too.
PRODUCER_VERSION_FORMATatsweeper.service.ts:23matches exactly whatformat_last_updatedemits with%Y-%m-%d %H:%M:%S%.6f(rust/cohort-stream-processor/src/producer/mod.rs:156-161), so_parseAndValidateBatchnever strips a stamp from a current producer. - Impact: confirmed silent data loss if a producer ever omits or malforms the stamp — a current member's row is deleted, and flag evaluation reads them as a non-member until the next reconcile or the next versioned write.
offFormatVersionswould rise at the same time, so the cause is visible, but the deletion itself is not. - Priority: lowered from
must_fix. The defect needs an upstream producer regression before it can fire, and the Rust type rules out the primary route, so nothing reaches a user from this PR as it stands. It stays on record because the trigger is a valve this PR relies on, the failure mode is deletion rather than the benign degradation the code comment promises, and the fix is small.
Issue description
The sweep classifies a row only by its preserved version. The consumer intentionally applies a versionless change without changing that field. A current versionless entered event can therefore leave an old version below this run's threshold. This DELETE removes the current membership.
Suggested fix
Record versionless writes in a separate field that the sweep can test. Another safe option makes these rows non-sweepable until a versioned write replaces them. Add a race test where a versionless entered change arrives after the snapshot and before deletion.
Prompt to fix with AI (copy-paste)
## Context
@nodejs/src/cdp/services/cohort-membership/sweeper.service.ts#L487-499
<issue_description>
The sweep classifies a row only by its preserved `version`. The consumer intentionally applies a versionless change without changing that field. A current versionless `entered` event can therefore leave an old version below this run's threshold. This DELETE removes the current membership.
</issue_description>
<issue_validation>
- **Checked:** whether a versionless change can reach the upsert, what it leaves in the `version` column, and whether the sweep predicate can then match that row.
- **Found:** the mechanism is real. `upsertMembershipRows` at `cdp-cohort-membership.consumer.ts:365-368` sets `version = COALESCE(EXCLUDED.version, cohort_membership.version)` and admits the write through `EXCLUDED.version IS NULL`, so a versionless `entered` flips `in_cohort` and refreshes `last_updated` while the stale stamp survives. The sweep at `sweeper.service.ts:489-490` tests only `team_id`, `cohort_id`, and `version < $3::timestamp` — it never reads `in_cohort` or `last_updated` — so that row is deleted even though it holds a current membership the run did not assert. The branch is live behavior the author tests, not dead code: `cdp-cohort-membership.consumer.test.ts:341` asserts "a versionless change applies without moving the version".
- **Found:** this PR makes a stated invariant wrong. The comment at `cdp-cohort-membership.consumer.ts:505-508` says an off-format stamp degrades to no version and "loses only the ordering: the membership transition itself still applies". With the sweep in place that is no longer true — the transition applies and is then silently deleted. The off-format valve exists precisely because the author judged format drift worth building for.
- **Found:** no trigger exists today, which is what caps the severity. The topic has one producer: `CohortMembershipChange.last_updated` is a non-optional `String` at `rust/cohort-stream-processor/src/producer/mod.rs:53`, with no `skip_serializing_if` (unlike `origin` and `run_id` two fields below), and every emitting path threads a stamp in — `map_transition` at `:118`, plus `cascade_path.rs:149`, `stage2_path.rs:110`, and `person_seed_path.rs:511`. Python only declares the topic and consumes it into ClickHouse; `cohort-seeder` never builds this struct.
- **Found:** the downgrade route is closed too. `PRODUCER_VERSION_FORMAT` at `sweeper.service.ts:23` matches exactly what `format_last_updated` emits with `%Y-%m-%d %H:%M:%S%.6f` (`rust/cohort-stream-processor/src/producer/mod.rs:156-161`), so `_parseAndValidateBatch` never strips a stamp from a current producer.
- **Impact:** confirmed silent data loss if a producer ever omits or malforms the stamp — a current member's row is deleted, and flag evaluation reads them as a non-member until the next reconcile or the next versioned write. `offFormatVersions` would rise at the same time, so the cause is visible, but the deletion itself is not.
- **Priority:** lowered from `must_fix`. The defect needs an upstream producer regression before it can fire, and the Rust type rules out the primary route, so nothing reaches a user from this PR as it stands. It stays on record because the trigger is a valve this PR relies on, the failure mode is deletion rather than the benign degradation the code comment promises, and the fix is small.
</issue_validation>
## Task
Investigate the issue and solve it
<potential_solution>
Record versionless writes in a separate field that the sweep can test. Another safe option makes these rows non-sweepable until a versioned write replaces them. Add a race test where a versionless `entered` change arrives after the snapshot and before deletion.
</potential_solution>
There was a problem hiding this comment.
Leaving this as is. No producer today can emit a versionless or malformed stamp, so this race cannot fire, and the parser comment now states that a degraded row stays sweepable until a versioned write lands. A real fix needs a new column marking rows touched by an unordered write, and we would rather not grow the schema for a case the producer contract rules out.
haacked
left a comment
haacked
left a comment
There was a problem hiding this comment.
Two blocking issues inline, both about what happens around the flag rather than inside the sweep.
d7b5bed to
f5c2925
Compare
4a5ef56 to
b9f830e
Compare
haacked
left a comment
haacked
left a comment
There was a problem hiding this comment.
Everything from the last round landed. Two blocking inline, and one of them is just a stale bullet in the description.
b9f830e to
0a1d96b
Compare
…sert A person who stops matching an edited cohort gets no live transition, so their in_cohort = true row survives forever. A reconcile run replays full membership and certifies itself with per-partition markers; the sweep deletes what the run did not re-assert. Rows are stamped with the producer version, and the oldest version a run asserted bounds the sweep threshold. Rows, marks, and consumer progress commit in one transaction. Deletes wait until the snapshot minimum is known and progress passes the marker-completion watermarks; progress also refreshes from committed offsets so quiet partitions cannot stall the gate. Gated by COHORT_MEMBERSHIP_SWEEP_ENABLED, off by default. Version persistence stays ungated: a mixed fleet writing NULLs would get current members swept.
0a1d96b to
17feec1
Compare
|
😎 Stack merged successfully - details. |
haacked
left a comment
haacked
left a comment
There was a problem hiding this comment.
Round four, and it all landed. Nothing blocking; everything inline is follow-up now that this is merged.
| topic: KAFKA_COHORT_MEMBERSHIP_CHANGED, | ||
| }) | ||
|
|
||
| if (config.COHORT_MEMBERSHIP_SWEEP_ENABLED && !config.COHORT_MEMBERSHIP_VERSION_WRITES_ENABLED) { |
There was a problem hiding this comment.
suggestion: The PR description says COHORT_MEMBERSHIP_SWEEP_ENABLED gates the version stamps, the snapshot marks, and the progress rows. The merged code actually gates those with COHORT_MEMBERSHIP_VERSION_WRITES_ENABLED instead: the dedupe rule at cdp-cohort-membership.consumer.ts:301, the write shape at :316, and the progress rows at :548. An operator following the outdated description sets COHORT_MEMBERSHIP_SWEEP_ENABLED on its own, but the constructor throws when sweeping is on without version writes (cdp-cohort-membership.consumer.ts:166-173), so the pod crashes at startup (server.ts:376) and membership ingestion stops.
The ordering rule has nowhere else to live besides the PR description and the comment at config.ts:189-192. Fix the second bullet in the PR description to this:
- Two flags, both defaulting to false.
COHORT_MEMBERSHIP_VERSION_WRITES_ENABLEDgates the version stamps, the snapshot marks and the progress rows; with it off the consumer keeps the exact pre-sweep write shape, so the migrations gate that flip rather than the deploy.COHORT_MEMBERSHIP_SWEEP_ENABLEDgates the marker consumer and the sweep, and a pod refuses to start when it is set without version writes. Turn version writes on first and off last, and sweeping on last and off first: a pod writing without versions leaves a re-asserted row carrying its old stamp, which a sweeping pod reads as deletable.
| private async readConsumerProgress(): Promise<Map<number, number>> { | ||
| const result = await this.postgres.query<{ partition: number; next_offset: string }>( | ||
| PostgresUse.BEHAVIORAL_COHORTS_RW, | ||
| 'SELECT partition, next_offset FROM cohort_membership_consumer_progress WHERE cluster = $1 AND topic = $2', |
There was a problem hiding this comment.
suggestion: The test doesn't check that the cluster and topic filters work, so a topic change on the same brokers could open the gate on the old topic's offsets. readConsumerProgress filters progress rows by cluster and topic, but every progress row in the suite gets inserted with the same CLUSTER and TOPIC: insertProgress at sweeper.service.serial.test.ts:128 hardcodes both. Dropping either filter changes no assertion.
The rows it would then pick up are real. collectGarbage only removes progress rows after 30 days, so after the processor's output topic flips from the shadow name to the live one, the old topic's much larger offsets sit in the table for a month while a fresh run captured on the new topic passes the identity check at sweeper.service.ts:462. The gate would see progress past every watermark and delete rows whose re-asserting snapshot messages are still queued.
Add cluster and topic parameters to insertProgress, then seed off-feed rows for partitions 0 and 3 at offsets past the watermarks before any in-feed row exists in should hold the sweep until consumer progress passes every captured watermark (sweeper.service.serial.test.ts:210). If the filter breaks, the test catches it: blocked: 1 becomes swept: 1 rather than racing on Map insertion order.
| } | ||
|
|
||
| /** Returns the deleted row count, or null when the run was refused rather than swept. */ | ||
| private async sweep(run: ClaimableRun): Promise<number | null> { |
There was a problem hiding this comment.
suggestion: A reconcile run whose snapshot minimum collapsed to -infinity reaches abandonThresholdless, whose comment says it is unreachable, and it gets logged as an error with sweepsAbandoned{reason='no_threshold'}.
collectSnapshotMarks folds VERSIONLESS into min_snapshot_version when a reconcile change arrives without a stamp (cdp-cohort-membership.consumer.ts:127), and logs "collapsing the run threshold", so the design expects the run to sweep and delete nothing. The run is claimable because -infinity is not NULL and that is all claimablePredicate requires. But the repo installs a TIMESTAMP parser at common/utils/db/postgres.ts:37-39 that runs the value through Luxon's fromSQL, which cannot parse -infinity and returns null, so LEAST(...) AS threshold at sweeper.service.ts:115 comes back null and sweep() takes the !run.threshold branch at :527.
Nothing is deleted either way, so the data outcome is safe. What is wrong is the reporting: an operator watching sweepsAbandoned{reason='no_threshold'} and a logger.error sees what the comment tells them is impossible, and the run lands in abandoned rather than the swept state the collapse was designed to produce. The same parser was already worked around for the version column, via the COALESCE(to_char(...)) in the delete's RETURNING at sweeper.service.ts:609-611; the threshold projection has the same exposure and no such handling.
Add a sweeper case with min_snapshot_version = '-infinity' that asserts zero deletions and the terminal state the run actually reaches, plus a correction to the abandonThresholdless comment.
| return this.rdKafkaConsumer.isConnected() ? this.rdKafkaConsumer.assignments() : [] | ||
| } | ||
|
|
||
| public async queryWatermarkOffsets(topic: string, partition: number, timeout = 10000): Promise<[number, number]> { |
There was a problem hiding this comment.
suggestion: The two Kafka consumer implementations now carry the same three metadata reads copied line for line, so a fix applied to one leaves the other wrong and nothing catches it. queryWatermarkOffsets, committedOffsets, and getPartitionsForTopic at consumer-v2.ts:212-250 are byte-identical to the versions at consumer-v1.ts:365-403, down to the Not connected throw, the paired captureException and logger.error calls, and the 10000 ms default timeout. The only difference is the parameter name in the meta.topics.find callback. Before this change the only method duplicated verbatim across the two classes was the three-line assignments.
Every method added to the interface at common/kafka/consumer/index.ts:25-26 has to be written out in both classes, and this change adds three at once. The second copy is not short-lived either, because CONSUMER_USE_V2 defaults to false and v1 is the path that actually runs.
None of the three touches anything implementation-specific; each takes the raw RdKafkaConsumer and calls straight through it. Extract them to a shared module beside metrics.ts:
// common/kafka/consumer/metadata.ts
export async function readCommittedOffsets(
consumer: RdKafkaConsumer,
timeout: number
): Promise<TopicPartitionOffset[]> {
if (!consumer.isConnected()) {
return []
}
return await promisifyCallback<TopicPartitionOffset[]>((cb) => consumer.committed(timeout, cb))
}Each class's method then collapses to return await readCommittedOffsets(this.rdKafkaConsumer, timeout).
|
|
||
| expect(await readSweep()).toMatchObject({ status: 'collecting', membership_hwms: null }) | ||
|
|
||
| // Redelivery must be a no-op, not an increment. |
There was a problem hiding this comment.
suggestion: The assertion under "Redelivery must be a no-op, not an increment" cannot fail. The line above it has already applied partitions 0 through 62, so redelivering partition 5 touches a bit that is already set, and the assertion that follows only repeats the status: 'collecting' that was asserted four lines earlier.
Both fold sites are invisible to it. If the in-memory fold at sweeper.service.ts:257 added instead of OR-ing, 32 + 32 would land on bit 6, which is already set. If the SQL fold at sweeper.service.ts:278 added, the sum would overflow bigint, the per-entry catch at :289 would swallow the error, and the row would be unchanged. A fold that is not idempotent can carry into a bit no marker covers, which either strands the run forever or completes it without every partition's marker and lets the sweep delete rows for partitions still emitting.
Apply markerMessage(COHORT_PARTITION_COUNT - 1) twice in one batch and assert marker_bits: '-1'. If the fold logic breaks, the test catches it instead of silently passing.
| // version writes off there is no SQL guard to agree with, so the last message in | ||
| // Kafka order wins and the deploy stays a behavioral no-op until the flag flips. | ||
| const versionAware = this.config.COHORT_MEMBERSHIP_VERSION_WRITES_ENABLED | ||
| const deduped = new Map<string, CohortMembershipChange>() |
There was a problem hiding this comment.
suggestion: The batch dedupe rule can only be tested through Postgres, so the four it.each cases covering it each spend a database round trip on what is pure in-memory logic. The pick sits inline in persistCohortMembershipChanges at cdp-cohort-membership.consumer.ts:302-314, while the file's other two batch folds, collectSnapshotMarks and partitionProgressFromMessages, are module-level pure functions of the same shape.
Moving it out matches them and gives the four-way condition a name, a signature, and a home for the six-line comment that currently explains it:
/** … the existing comment moves here … */
function dedupeChanges(
changes: CohortMembershipChange[],
versionAware: boolean
): CohortMembershipChange[] {The call site becomes const deduped = dedupeChanges(changes, versionAware), and the check on the next line can then use the same versionAware local instead of re-reading this.config.COHORT_MEMBERSHIP_VERSION_WRITES_ENABLED at :316.
| deps | ||
| ) | ||
| teamId = Number.parseInt(new UUIDT().toString().replaceAll('-', '').slice(-7), 16) | ||
| // Progress rows key on (cluster, partition) and nothing truncates between tests, so a |
There was a problem hiding this comment.
nit: This comment says progress rows key on (cluster, partition), but the primary key is (cluster, topic, partition) (20260826000002_create_sweep_tables.sql:77, and the ON CONFLICT at cdp-cohort-membership.consumer.ts:469). Add topic to the comment so the next person adjusting test isolation reads the real key.
| // 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 |
There was a problem hiding this comment.
nit: const startAtLatest = START_AT_LATEST is a second name for the imported value that changes nothing about it, and the four createKafkaConsumer calls at :179, :186, :193, and :200 still pass the alias, so a reader chasing the argument lands here before reaching the export. Pass START_AT_LATEST at those four sites and delete this line, keeping the comment above it since it explains why this consumer starts at the tip.
| (this.consumerConfig['auto.offset.reset' as keyof ConsumerGlobalConfig] as | ||
| | 'earliest' | ||
| | 'latest' | ||
| | undefined) ?? 'earliest' |
There was a problem hiding this comment.
nit: This hardcodes 'earliest' while v2 reads the same value from DEFAULT_AUTO_OFFSET_RESET (consumer-v2.ts:47), so changing one leaves the two implementations disagreeing about where a consumer starts. Move that constant to index.ts and have both createConsumer bodies read it.
| expect(result.rows).toHaveLength(1) | ||
| expect(result.rows[0].in_cohort).toBe(false) | ||
| }) | ||
| it.each([ |
There was a problem hiding this comment.
nit: The only equal-stamp case in the dedupe table runs with version writes off, so nothing asserts that a later message with an identical stamp still wins on the version-writes path. Add one more row in the table: flag on, two equal stamps, last in Kafka order wins; changing >= to > at cdp-cohort-membership.consumer.ts:945 would make the earlier message win instead, which the test would catch.
Problem
Someone who stops matching an edited cohort still evaluates as a member, and nothing ever corrects it.
The membership consumer only upserts, so a person who no longer matches gets no live transition and their row survives.
Builds on the schema PR below this one in the stack.
Closes #88014.
Changes
-infinity, which sorts below every producer stamp, so no separate cleanup is needed.COHORT_MEMBERSHIP_SWEEP_ENABLEDgates the marker consumer, the sweep, and every new write, and defaults to false. With the flag off the consumer keeps the exact pre-sweep write shape, so the migrations gate the flag flip, not the deploy.auto.offset.resetoverride instead of pinning every consumer toearliest. This matches v2, which already honored it.latestso a fresh group skips the person topic backlog.KafkaConsumerInterfacegains three read methods. The v2 consumer implements all three; v1 already had two and only addscommittedOffsets.Nothing is user visible until the flag flips in charts, which happens after the processor writes membership to the live topic.
How did you test this code?
New sweeper tests cover the marker bitmap under redelivery, the progress gate, and stale claim recovery. They also cover the delete predicate scoping by team and cohort, and the two refusals that stop a run from deleting a whole cohort.
New consumer tests cover version persistence across all three origins and the last writer wins guard on replay, including a versionless change against a versioned row. They also cover snapshot minimum tracking across batches, progress that never moves backwards, partial partition metadata refusal, committed offset filtering, the flag-off write shape, and a marker batch failure staying contained.
Each assertion in the original suites was checked by breaking the implementation and confirming the intended test failed. Reverting the gate, the version guard, the snapshot minimum, and both refusals each failed exactly one group. The review-round tests were not break-checked.
nodejs/src/cdp/dev/cohort-sweep-e2e.test.tsdrives the whole chain against real Kafka and Postgres on a dev stack. It lives underdev/so CI skips it, and it passed against both consumer implementations.Not run: CI, and any deployed environment.
👉 Stay up-to-date with PostHog coding conventions for a smoother review.
Automatic notifications
Docs update
None.
🤖 Agent context
Autonomy: Human-driven (agent-assisted)
Authored with Claude Code (Opus) from a design document for issue #88014. Review fixes applied with Claude Code (Fable) across two review rounds.
Skills invoked:
/implement-plan,/writing-tests,/writing-pr-descriptions, andcode-review-v1for an adversarial pass.Two designs from the document did not survive contact with a running stack. The gate originally read progress only from consumed batches, which deadlocked on any partition holding retained messages with no new traffic. Committed offsets replaced it. The threshold originally fell back to the marker minimum through
LEAST. That value sits above the rows a run asserted, so a run with no snapshot marks would have deleted a whole cohort. Such a run is now refused.The end to end harness found the first of those, and the review found the second. A second review round moved the version writes behind the flag, dropped NULL rows from the sweep predicate, and hardened the marker path against transient failures.