Make provider deletion durable with generation fencing - #1677
Conversation
Replace unbounded provider-wide ClickHouse deletion with a generation-fenced transactional outbox and checkpointed BullMQ batches. Includes PostgreSQL and ClickHouse migrations. Historical projection materialization remains operator-controlled, and pre-migration deletion requests must be resubmitted after deployment.
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
|
The PR looks solid and introduces a robust "generation" concept to handle provider data deletion and stale data insertion. The race condition handling in the ClickHouse sink is particularly well-thought-out. I've noted two main points:
Otherwise, the changes are correct and well-tested. 🤖 Reviewed by codereviewbot.ai - Catch bugs before your team does. |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
📝 WalkthroughWalkthroughThis PR introduces generation-fenced provider-data deletion through PostgreSQL outbox records, BullMQ dispatch, checkpointed ClickHouse tombstoning, and acknowledgement handling. It also propagates generations through metric-stream ingestion, updates migrations and tests, and adds operational documentation. ChangesProvider data deletion workflow
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 1 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (1 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Storybook previews for This comment updates automatically on each PR push. |
|
[Review Note] File: The |
![codereviewbot-ai[bot]](https://avatars.githubusercontent.com/in/309638?s=60&v=4){kind=small}
Format the migration for the CI SQLFluff policy and build the outbox index concurrently. Use synchronous ClickHouse mutations for integration cleanup because lightweight deletes reject tables with projections.
|
The introduction of the provider data generation and outbox pattern is a solid approach to handling asynchronous deletions in ClickHouse while maintaining consistency. I've noted a few areas for improvement:
The rest of the changes, including the SQL migrations and the outbox processor job, look well-implemented. 🤖 Reviewed by codereviewbot.ai - Catch bugs before your team does. |
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
|
[Review Note] File: The ClickHouse query here does not filter by Since To ensure immediate consistency in the UI after a deletion, consider filtering by the current generation. Since the current generation is stored in Postgres, you might need to fetch it first or join with the WHERE user_id = {userId:UUID}
AND provider_id = {providerId:String}
AND generation >= {currentGeneration:UInt64} |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/metric-stream/write-metric-stream.ts (1)
6-44: 🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy liftN+1 database queries when resolving provider generation.
writeMetricStreamRowsresolves provider generations sequentially per missing key, which causes O(N) queries when receiving generic arrays spanning multiple providers (intra-batch). Additionally, because sync repositories callwriteMetricStreamRowsinside their batching loops, identical queries are repeated across every batch for the exact same provider (inter-batch). For high-frequency data streams (e.g., IMU syncs mapping hundreds of thousands of samples into chunks of 1000), this creates a massive database latency bottleneck in background workers.
src/metric-stream/write-metric-stream.ts#L6-L44: UpdateWriteMetricStreamRowsOptionsto accept an optionalgenerationCache?: Map<string, number>, thread it throughaddProviderDataGenerations, and refactor the generation lookup to fetch missing keys concurrently viaPromise.allrather than using a sequentialfor...ofloop.packages/server/src/repositories/activity-recording-repository.ts#L157-L157: Initialize aconst generationCache = new Map<string, number>();immediately before theinput.samplesloop and pass it in thewriteMetricStreamRowsoptions.packages/server/src/repositories/ble-heart-rate-sync-repository.ts#L104-L104: Initialize and pass a sharedgenerationCachearound theinsertSampleBatchloop.packages/server/src/repositories/health-kit-sync-repository.ts#L393-L393: Initialize and pass a sharedgenerationCachearound theprocessBodyMeasurementsloop.packages/server/src/repositories/health-kit-sync-repository.ts#L504-L504: Initialize and pass a sharedgenerationCachearound theprocessMetricStreamloop.packages/server/src/repositories/inertial-measurement-unit-sync-repository.ts#L94-L98: Initialize and pass a sharedgenerationCachearound theinsertBatchloop.packages/server/src/repositories/watch-altitude-sync-repository.ts#L81-L81: Initialize and pass a sharedgenerationCachearound theinsertSampleBatchloop.packages/server/src/repositories/whoop-ble-sync-repository.ts#L103-L103: Initialize and pass a sharedgenerationCachearound theinsertRealtimeDataBatchloop.⚡ Proposed fix for `src/metric-stream/write-metric-stream.ts`
export interface WriteMetricStreamRowsOptions { database: Database; publisher: MetricStreamEventPublisher; rows: readonly MetricStreamRowInput[]; + generationCache?: Map<string, number>; } ... export async function addProviderDataGenerations( database: Database, rows: readonly MetricStreamRowInput[], + cache?: Map<string, number>, ): Promise<MetricStreamRowInput[]> { - const generationsByProvider = new Map<string, number>(); - for (const row of rows) { - const providerKey = `${row.userId}\0${row.providerId}`; - if (!generationsByProvider.has(providerKey)) { - generationsByProvider.set( - providerKey, - await getProviderDataGeneration(database, row.userId, row.providerId), - ); - } - } + const generationsByProvider = cache ?? new Map<string, number>(); + + const missingKeys = new Set<string>(); + for (const row of rows) { + const providerKey = `${row.userId}\0${row.providerId}`; + if (!generationsByProvider.has(providerKey)) { + missingKeys.add(providerKey); + } + } + + if (missingKeys.size > 0) { + await Promise.all( + Array.from(missingKeys).map(async (key) => { + const [userId, providerId] = key.split("\0"); + const generation = await getProviderDataGeneration(database, userId!, providerId!); + generationsByProvider.set(key, generation); + }) + ); + } return rows.map((row) => { ... export async function writeMetricStreamRows( options: WriteMetricStreamRowsOptions, ): Promise<WriteMetricStreamRowsResult> { - const rowsWithGeneration = await addProviderDataGenerations(options.database, options.rows); + const rowsWithGeneration = await addProviderDataGenerations( + options.database, + options.rows, + options.generationCache + );🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/metric-stream/write-metric-stream.ts` around lines 6 - 44, Eliminate repeated provider-generation queries by adding an optional shared generationCache to WriteMetricStreamRowsOptions and threading it through addProviderDataGenerations; resolve uncached provider keys concurrently with Promise.all, then populate and reuse the cache. In activity-recording-repository.ts:157, ble-heart-rate-sync-repository.ts:104, health-kit-sync-repository.ts:393 and :504, inertial-measurement-unit-sync-repository.ts:94-98, watch-altitude-sync-repository.ts:81, and whoop-ble-sync-repository.ts:103, create one Map before the respective batching loop and pass it to each writeMetricStreamRows or insert batch call.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@packages/server/src/jobs/process-provider-data-deletion-job.integration.test.ts`:
- Around line 105-121: Parse the results of the generation and acknowledgement
ClickHouse queries with the existing Zod coercion pattern before asserting them.
Update the assertions around generationResult and acknowledgementResult so
aggregate values such as max(generation) and count() are converted from their
runtime representation to numbers, while preserving the expected values of 1.
In `@packages/server/src/repositories/provider-detail-repository.ts`:
- Around line 780-790: The transaction cleanup loop around transaction.execute
must not swallow undefined_table errors because the transaction remains aborted.
Require the listed tables to exist, or isolate each optional delete with a
savepoint that can be rolled back before continuing; preserve the outbox
insert's executability. Add a real-PostgreSQL integration test with a minimal
fixture covering the intended missing-table behavior.
In `@src/jobs/provider-data-deletion-outbox.ts`:
- Around line 18-23: Update the provider deletion outbox flow around
enqueueProviderDataDeletion and markProviderDataDeletionDispatched to handle
terminal queue failures after all attempts are exhausted. Reset the
corresponding request to pending or explicitly requeue it when the worker
reports a terminal failure, so it becomes pollable again instead of remaining
dispatched. Keep successful dispatch handling unchanged.
In `@src/jobs/worker.ts`:
- Around line 167-175: The providerDataDeletionWorker currently trusts the
TypeScript payload type without runtime validation. Define a Zod schema covering
the complete ProviderDataDeletionJobData payload, including checkpoint, parse
job.data at the worker boundary, and pass the parsed result to
processProviderDataDeletionJob while preserving the existing worker context and
callbacks.
---
Outside diff comments:
In `@src/metric-stream/write-metric-stream.ts`:
- Around line 6-44: Eliminate repeated provider-generation queries by adding an
optional shared generationCache to WriteMetricStreamRowsOptions and threading it
through addProviderDataGenerations; resolve uncached provider keys concurrently
with Promise.all, then populate and reuse the cache. In
activity-recording-repository.ts:157, ble-heart-rate-sync-repository.ts:104,
health-kit-sync-repository.ts:393 and :504,
inertial-measurement-unit-sync-repository.ts:94-98,
watch-altitude-sync-repository.ts:81, and whoop-ble-sync-repository.ts:103,
create one Map before the respective batching loop and pass it to each
writeMetricStreamRows or insert batch call.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: d60699ae-4ddd-4d73-9b14-751272c70e3c
📒 Files selected for processing (66)
analytics/models/read_models/activity_sensor_summary_rows.sqldocs/README.mddocs/production-incident-baseline.mddocs/provider-data-deletion-runbook.mddrizzle/0050_provider_data_deletion_outbox.sqlpackage.jsonpackages/server/src/jobs/process-provider-data-deletion-job.integration.test.tspackages/server/src/repositories/activity-recording-repository.test.tspackages/server/src/repositories/activity-recording-repository.tspackages/server/src/repositories/ble-heart-rate-sync-repository.tspackages/server/src/repositories/health-kit-sync-repository.test.tspackages/server/src/repositories/health-kit-sync-repository.tspackages/server/src/repositories/inertial-measurement-unit-sync-repository.tspackages/server/src/repositories/provider-detail-repository.integration.test.tspackages/server/src/repositories/provider-detail-repository.test.tspackages/server/src/repositories/provider-detail-repository.tspackages/server/src/repositories/watch-altitude-sync-repository.tspackages/server/src/repositories/whoop-ble-sync-repository.tspackages/server/src/routers/activity-recording.test.tspackages/server/src/routers/ble-heart-rate-sync.test.tspackages/server/src/routers/clickhouse-integration-test-helpers.tspackages/server/src/routers/health-kit-sync-processors.test.tspackages/server/src/routers/health-kit-sync-processors.tspackages/server/src/routers/health-kit-sync.test.tspackages/server/src/routers/inertial-measurement-unit-sync.test.tspackages/server/src/routers/provider-detail.test.tspackages/server/src/routers/provider-detail.tspackages/server/src/routers/watch-altitude-sync.test.tspackages/server/src/routers/whoop-ble-sync.test.tssrc/db/clickhouse-metric-stream-bootstrap.tssrc/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.tssrc/db/clickhouse-migrations/0046_provider_data_generation.tssrc/db/clickhouse-migrations/registry.test.tssrc/db/clickhouse-migrations/registry.tssrc/db/clickhouse.tssrc/db/metric-stream-writer.test.tssrc/db/metric-stream-writer.tssrc/db/provider-data-deletion.test.tssrc/db/provider-data-deletion.tssrc/db/schema/events.tssrc/db/typed-sql.test.tssrc/db/typed-sql.tssrc/jobs/process-activity-delete-analytics-job.test.tssrc/jobs/process-activity-delete-analytics-job.tssrc/jobs/process-provider-data-deletion-job.test.tssrc/jobs/process-provider-data-deletion-job.tssrc/jobs/provider-data-deletion-outbox.test.tssrc/jobs/provider-data-deletion-outbox.tssrc/jobs/queues.test.tssrc/jobs/queues.tssrc/jobs/worker.test.tssrc/jobs/worker.tssrc/metric-stream/clickhouse-sink.test.tssrc/metric-stream/clickhouse-sink.tssrc/metric-stream/clickhouse-table.tssrc/metric-stream/events.test.tssrc/metric-stream/events.tssrc/metric-stream/write-metric-stream.test.tssrc/metric-stream/write-metric-stream.tssrc/providers/apple-health/db-insertion.integration.test.tssrc/providers/apple-health/db-insertion.test.tssrc/providers/fitbit/provider.test.tssrc/providers/garmin.test.tssrc/providers/oura.test.tssrc/providers/ride-with-gps-ext.test.tssrc/providers/strava.test.ts
Batch fencing-token reads, validate Redis payloads, fail fast on missing tables, and redrive terminal BullMQ failures. Derive ClickHouse tombstone versions from stored rows.
|
🤖 Review skipped: Repository rate limit exceeded. Free accounts are limited to 2 reviews per 4 hours per repository. Upgrade to a paid plan for unlimited reviews. |
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
|
Review disposition for the outside-diff cross-batch generation-cache suggestion in this review:
|
|
🤖 Review skipped: Repository rate limit exceeded. Free accounts are limited to 2 reviews per 4 hours per repository. Upgrade to a paid plan for unlimited reviews. |
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
Keep provider and sync unit fixtures faithful to the batched generation lookup without weakening production fencing.
|
🤖 Review skipped: Repository rate limit exceeded. Free accounts are limited to 2 reviews per 4 hours per repository. Upgrade to a paid plan for unlimited reviews. |
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/production-incident-baseline.md`:
- Around line 13856-13860: Update the validation bullet in the production
incident baseline to cite an authoritative successful CI run or primary
test/mutation report for the stated test totals, integration results, deletion
tests, dry-run count, and 100% mutation score. Retain the claims only when each
is verifiable from the linked evidence, and follow the documentation citation
requirement for content under docs/.
In `@packages/server/src/repositories/health-kit-sync-repository.test.ts`:
- Around line 14-22: Move resolveProviderDataGenerationsForTest and its
provider-data-deletion mock setup out of shared src/test/test-helpers.ts imports
and into colocated test-helper files for
packages/server/src/repositories/health-kit-sync-repository.test.ts#L14-L22,
packages/server/src/routers/watch-altitude-sync.test.ts#L3-L11,
packages/server/src/routers/whoop-ble-sync.test.ts#L3-L11,
src/providers/amazfit-zepp.test.ts#L21-L26,
src/providers/apple-health/db-insertion.test.ts#L24-L29,
src/providers/apple-health/import.test.ts#L7-L12,
src/providers/eight-sleep.test.ts#L15-L20,
src/providers/fitbit/provider.test.ts#L21-L26, and
src/providers/garmin.test.ts#L19-L24. Update each test to use its directory’s
local test-helpers.ts while preserving the existing getProviderDataGenerations
mock behavior.
In `@src/jobs/worker.ts`:
- Around line 170-188: Wrap providerDataDeletionJobDataSchema.parse in the
worker processor and convert validation failures into BullMQ’s imported
UnrecoverableError so malformed jobs are not retried or redriven. Update the
providerDataDeletionWorker failed handler to accept the original error
parameter, preserve retry handling, and call captureException for terminal or
unrecoverable failures, including the original job error rather than only retry
errors.
In `@src/metric-stream/clickhouse-sink.integration.test.ts`:
- Around line 107-108: Update the row expectation in the ClickHouse integration
test to assert latest_version as the string "1", matching the declared result
type and ClickHouse UInt64 client behavior; leave is_deleted unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 908c2b13-ae49-4a5f-97eb-2c245b4fe0bb
📒 Files selected for processing (44)
analytics/models/read_models/read_model_microbatch.sql.test.tscspell.jsondocs/production-incident-baseline.mddrizzle/0050_provider_data_deletion_outbox.sqlpackages/server/src/jobs/process-provider-data-deletion-job.integration.test.tspackages/server/src/repositories/activity-recording-repository.test.tspackages/server/src/repositories/health-kit-sync-repository.test.tspackages/server/src/repositories/heart-rate-repository.integration.test.tspackages/server/src/repositories/provider-detail-repository.integration.test.tspackages/server/src/repositories/provider-detail-repository.test.tspackages/server/src/repositories/provider-detail-repository.tspackages/server/src/routers/activity-recording.test.tspackages/server/src/routers/ble-heart-rate-sync.test.tspackages/server/src/routers/health-kit-sync-processors.test.tspackages/server/src/routers/health-kit-sync.test.tspackages/server/src/routers/inertial-measurement-unit-sync.test.tspackages/server/src/routers/watch-altitude-sync.test.tspackages/server/src/routers/whoop-ble-sync.test.tssrc/db/metric-stream-writer.test.tssrc/db/provider-data-deletion.integration.test.tssrc/db/provider-data-deletion.test.tssrc/db/provider-data-deletion.tssrc/jobs/queues.test.tssrc/jobs/queues.tssrc/jobs/worker.test.tssrc/jobs/worker.tssrc/metric-stream/clickhouse-sink.integration.test.tssrc/metric-stream/clickhouse-sink.test.tssrc/metric-stream/clickhouse-sink.tssrc/metric-stream/write-metric-stream.test.tssrc/metric-stream/write-metric-stream.tssrc/providers/amazfit-zepp.test.tssrc/providers/apple-health/db-insertion.test.tssrc/providers/apple-health/import.test.tssrc/providers/eight-sleep.test.tssrc/providers/fitbit/provider.test.tssrc/providers/garmin.test.tssrc/providers/oura.test.tssrc/providers/peloton.test.tssrc/providers/ride-with-gps-ext.test.tssrc/providers/strava.test.tssrc/providers/whoop.test.tssrc/providers/withings.test.tssrc/test/test-helpers.ts
💤 Files with no reviewable changes (1)
- src/metric-stream/clickhouse-sink.test.ts
Exercise interval dispatch, overlap fencing, error recovery, and shutdown so the lifecycle is mutation-covered.
|
🤖 Review skipped: Repository rate limit exceeded. Free accounts are limited to 2 reviews per 4 hours per repository. Upgrade to a paid plan for unlimited reviews. |
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
Classify malformed BullMQ jobs as unrecoverable, keep generation mocks colocated, and align ClickHouse test types with observed results.
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
|
🤖 Review skipped: Repository rate limit exceeded. Free accounts are limited to 2 reviews per 4 hours per repository. Upgrade to a paid plan for unlimited reviews. |
Exercise null, callable, and malformed record result shapes so the typed SQL guard is mutation-complete.
|
🤖 Review skipped: Repository rate limit exceeded. Free accounts are limited to 2 reviews per 4 hours per repository. Upgrade to a paid plan for unlimited reviews. |
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
Exercise persistence edge cases and cached queue creation, reuse, connection wiring, and shutdown for the deletion workflow.
|
🤖 Review aborted: the PR is too complex or took too long to analyze. |
There was a problem hiding this comment.
Asherlc has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
Summary