fix(db): prevent duplicate event inserts on lock expiry and partial-batch retry#417
fix(db): prevent duplicate event inserts on lock expiry and partial-batch retry#417raonitimo wants to merge 1 commit into
Conversation
…atch retry The event buffer can insert the same events into ClickHouse twice under two conditions: 1. Flush-lock expiry. tryFlush() holds a Redis flush lock with a 60s TTL, but a ClickHouse insert may run up to max_execution_time = 300s server-side. A slow insert can outlive the lock, letting a second worker replica acquire it and re-flush the same still-queued events. The `events` table has no dedup key, so this produces duplicate rows. 2. Partial-batch retry. processBuffer() inserted every chunk but only trimmed the whole slice from the Redis queue in a single ltrim at the very end. If a later chunk's insert threw, the error propagated before the trim ran, so the entire batch — including chunks already inserted — was retried on the next cycle, duplicating the already-committed chunks. Fixes: - Raise the flush-lock TTL to 360s so it cannot expire while an insert is in flight (comfortably above the 300s server-side insert ceiling). - Trim each chunk from the front of the queue immediately after its insert succeeds, and publish that chunk's realtime notification then. A mid-batch failure now leaves only the untrimmed remainder to retry; committed chunks are never replayed. - Send a deterministic per-chunk insert_deduplication_token (sha256 of the chunk's rows) as defense-in-depth, so an identical insert is rejected by ClickHouse even if the lock assumption is ever violated (e.g. a replica race on a replicated table). Because a safe front-of-queue trim requires committing chunks in order, this buffer's chunk inserts are now sequential rather than run through parallelLimit; the other buffers keep their parallel inserts. Adds unit tests for the partial-batch-retry path and the deterministic token. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
📝 WalkthroughWalkthroughChangesEvent buffer flush reliability
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant EventBuffer
participant Redis
participant ClickHouse
participant RealtimeNotifications
EventBuffer->>Redis: Read queued JSONEachRow lines
EventBuffer->>ClickHouse: Insert one chunk with deduplication token
ClickHouse-->>EventBuffer: Confirm chunk insert
EventBuffer->>Redis: Trim committed chunk
EventBuffer->>RealtimeNotifications: Publish per-project counts
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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/db/src/buffers/event-buffer.ts`:
- Around line 256-268: Handle the promise returned by publishEvent in the batch
publishing loop so Redis publish failures are caught and logged rather than
becoming unhandled rejections. Update the loop around countByProject and
publishEvent, preserving the existing per-project event payload and allowing the
buffer flush to continue for other projects.
- Around line 177-194: Update the event flush path using deduplicationToken so
ClickHouse insertion is synchronous and completes before the Redis ltrim commit
point. Override the settings from getClickhouseSettings() for this insert to
disable async_insert and wait for the insert to finish; do not rely on
async-insert deduplication for correctness.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 6e6a31a9-4c37-42f9-9f08-32db61b26412
📒 Files selected for processing (3)
packages/db/src/buffers/base-buffer.tspackages/db/src/buffers/event-buffer.test.tspackages/db/src/buffers/event-buffer.ts
| /** | ||
| * Deterministic idempotency token for a chunk of raw JSONEachRow lines. | ||
| * The same chunk content always hashes to the same token, so if the exact | ||
| * chunk is inserted twice — e.g. a lock-expiry double-flush, or a retry | ||
| * after a partial batch failure — ClickHouse can reject the duplicate | ||
| * block via `insert_deduplication_token`. This is defense-in-depth on top | ||
| * of the per-chunk commit below; on replicated tables it also covers a | ||
| * race where two replicas flush the same queue slice. | ||
| */ | ||
| private deduplicationToken(lines: string[]): string { | ||
| const hash = createHash('sha256'); | ||
| for (const line of lines) { | ||
| hash.update(line); | ||
| hash.update('\n'); | ||
| } | ||
| return hash.digest('hex'); | ||
| } | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether BUFFER_ASYNC_INSERTS is used in deployment/env config, and whether the
# `events` table DDL sets non_replicated_deduplication_window (if not Replicated).
rg -n 'BUFFER_ASYNC_INSERTS' --type=ts -C2
rg -n 'CREATE TABLE.*events|non_replicated_deduplication_window|ReplicatedMergeTree' -g '*.sql' -g '*.ts'Repository: Openpanel-dev/openpanel
Length of output: 161
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- files of interest ---'
git ls-files | rg '^(packages/db/src/(buffers|clickhouse)|.*(events|schema|migration|ddl).*)'
echo
echo '--- locate clickhouse/event references ---'
rg -n --hidden --no-ignore-vcs \
'insert_deduplication_token|async_insert|BUFFER_ASYNC_INSERTS|non_replicated_deduplication_window|ReplicatedMergeTree|MergeTree|events\b' \
packages db . 2>/dev/null | head -n 300
echo
echo '--- event-buffer file outline ---'
ast-grep outline packages/db/src/buffers/event-buffer.ts --view expanded || true
echo
echo '--- base-buffer file outline ---'
ast-grep outline packages/db/src/buffers/base-buffer.ts --view expanded || trueRepository: Openpanel-dev/openpanel
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- base-buffer outline ---'
ast-grep outline packages/db/src/buffers/base-buffer.ts --view expanded || true
echo
echo '--- event-buffer outline ---'
ast-grep outline packages/db/src/buffers/event-buffer.ts --view expanded || true
echo
echo '--- init clickhouse migration outline ---'
ast-grep outline packages/db/code-migrations/3-init-ch.ts --view expanded || true
echo
echo '--- event-buffer relevant lines ---'
cat -n packages/db/src/buffers/event-buffer.ts | sed -n '150,260p'
echo
echo '--- base-buffer relevant lines ---'
cat -n packages/db/src/buffers/base-buffer.ts | sed -n '1,220p'
echo
echo '--- init clickhouse migration relevant lines ---'
cat -n packages/db/code-migrations/3-init-ch.ts | sed -n '1,260p'Repository: Openpanel-dev/openpanel
Length of output: 28575
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- createTable helper outline ---'
ast-grep outline packages/db/src/clickhouse/migration.ts --view expanded || true
echo
echo '--- createTable helper relevant lines ---'
cat -n packages/db/src/clickhouse/migration.ts | sed -n '1,260p'
echo
echo '--- search for events_replicated / replicated engine usage ---'
rg -n 'events_replicated|ReplicatedMergeTree|ReplacingMergeTree|MergeTree\(' packages/db/src packages/db/code-migrations packages/db/prisma/migrations -g '!**/node_modules/**'Repository: Openpanel-dev/openpanel
Length of output: 13346
🌐 Web query:
ClickHouse insert_deduplication_token async_insert ignored deduplication replicated MergeTree non_replicated_deduplication_window
💡 Result:
In ClickHouse, the interaction between insert deduplication settings and asynchronous inserts works as follows: 1. insert_deduplication_token and async_insert Previously, insert_deduplication_token was not supported with async_insert, but this was addressed in version 23.7, where support for insert_deduplication_token was added for asynchronous inserts [1][2]. When using async_insert, the deduplication token acts as a user-defined identifier to prevent duplicate block insertion, taking precedence over the automatic data digest (checksum) method [3]. 2. async_insert and Deduplication By default, automatic block-level deduplication is disabled for asynchronous inserts unless explicitly enabled [4]. The setting async_insert_deduplicate=1 is required to enable deduplication for asynchronous inserts in ReplicatedMergeTree tables [5][6]. It is recommended to use wait_for_async_insert=1 when deduplication is required to ensure that retries are handled reliably and the client is aware of the final status of the insert [4]. 3. non_replicated_deduplication_window This setting enables block-level deduplication for non-replicated MergeTree tables [7][8]. By default, it is set to 0 (disabled) for non-replicated tables for backward compatibility [8]. When set to a positive integer, ClickHouse stores the hash sums of the most recently inserted blocks locally in a file to detect and ignore duplicates [7][8]. This mechanism functions similarly to the deduplication process in replicated tables, though hash sums are managed locally rather than via ClickHouse Keeper [7]. In summary, while async_insert once ignored deduplication tokens, modern versions (23.7+) support them [1][2]. For non-replicated tables, deduplication must be explicitly enabled using the non_replicated_deduplication_window setting [8].
Citations:
- 1: insert_deduplication_token does not work with async insert ClickHouse/ClickHouse#55539
- 2: support insert_deduplication_token for async insert ClickHouse/ClickHouse#52136
- 3: https://clickhouse.com/docs/guides/developer/deduplicating-inserts-on-retries
- 4: https://clickhouse.com/docs/optimize/asynchronous-inserts
- 5: https://dev.to/cheng-w/clickhouse-source-code-analysis-insert-deduplication-3jep
- 6: https://clickhouse.com/docs/operations/settings/settings
- 7: https://clickhouse.com/docs/operations/settings/merge-tree-settings
- 8: https://kb.altinity.com/altinity-kb-schema-design/insert_deduplication/
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- cluster/self-hosting helpers ---'
rg -n 'getIsCluster|getIsSelfHosting|CLICKHOUSE_CLUSTER|SELF_HOSTING|BUFFER_ASYNC_INSERTS' packages/db/src packages/db/code-migrations -g '!**/node_modules/**'
echo
echo '--- helper source ---'
cat -n packages/db/code-migrations/helpers.ts | sed -n '1,220p'
echo
echo '--- event-buffer insertion context ---'
cat -n packages/db/src/buffers/event-buffer.ts | sed -n '228,250p'Repository: Openpanel-dev/openpanel
Length of output: 5860
🌐 Web query:
ClickHouse docs insert_deduplication_token async_insert support 23.7 async_insert_deduplicate wait_for_async_insert deduplication token
💡 Result:
Support for using insert_deduplication_token with asynchronous inserts was introduced in ClickHouse version 23.7 [1]. Key details regarding this functionality: 1. Enabling Deduplication for Async Inserts: By default, deduplication is disabled for asynchronous inserts. To enable it, you must explicitly set async_insert_deduplicate = 1 [2][3]. 2. Functionality: Once async_insert_deduplicate is enabled, ClickHouse allows the use of insert_deduplication_token to control deduplication on a per-insert basis, similar to how it functions with standard synchronous inserts [1][4][5]. 3. Interaction with Materialized Views: Historically, deduplication with asynchronous inserts has faced limitations when dependent materialized views are involved [6][7]. While initial support for insert_deduplication_token was added in 23.7, end-to-end deduplication—specifically ensuring that materialized views are handled consistently during asynchronous insert retries—was further refined in later releases (e.g., ClickHouse 26.1) [6][5]. 4. Best Practices: - It is strongly recommended to use async_insert=1 in combination with wait_for_async_insert=1 to ensure reliability, as wait_for_async_insert=0 (fire-and-forget) can lead to silent failures and makes it difficult for the client to confirm whether data was successfully persisted [2]. - If using insert_deduplication_token, ensure that async_insert_deduplicate is active, otherwise the token may be ignored for asynchronous operations [8][1]. For further implementation, you can specify these settings within the SETTINGS clause of your INSERT query [2]: INSERT INTO YourTable SETTINGS async_insert=1, async_insert_deduplicate=1, wait_for_async_insert=1 VALUES (...)
Citations:
- 1: insert_deduplication_token does not work with async insert ClickHouse/ClickHouse#55539
- 2: https://clickhouse.com/docs/optimize/asynchronous-inserts
- 3: https://altinity.com/blog/using-async-inserts-for-peak-data-loading-rates-in-clickhouse
- 4: support insert_deduplication_token for async insert ClickHouse/ClickHouse#52136
- 5: Proper deduplication of async inserts. ClickHouse/ClickHouse#38075
- 6: https://clickhouse.com/blog/clickhouse-release-26-01
- 7: coordinate deduplication logic among use cases when async insert is ON by default ClickHouse/ClickHouse#76297
- 8: Making insert_deduplication_token work with async_insert ClickHouse/ClickHouse#52018
Don't rely on async inserts for this flush path. events only dedupes on replicated tables (or non-replicated tables with non_replicated_deduplication_window), and getClickhouseSettings() enables async_insert without async_insert_deduplicate/wait_for_async_insert. That makes ch.insert() return before ClickHouse has committed the block, so the Redis ltrim is no longer a safe commit point.
🤖 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 `@packages/db/src/buffers/event-buffer.ts` around lines 177 - 194, Update the
event flush path using deduplicationToken so ClickHouse insertion is synchronous
and completes before the Redis ltrim commit point. Override the settings from
getClickhouseSettings() for this insert to disable async_insert and wait for the
insert to finish; do not rely on async-insert deduplication for correctness.
| const countByProject = new Map<string, number>(); | ||
| for (const line of chunk) { | ||
| const projectId = extractProjectId(line); | ||
| if (projectId) { | ||
| countByProject.set( | ||
| projectId, | ||
| (countByProject.get(projectId) ?? 0) + 1, | ||
| ); | ||
| } | ||
| } | ||
| if ((i + 1) % yieldEvery === 0) { | ||
| await this.yieldToEventLoop(); | ||
| for (const [projectId, count] of countByProject) { | ||
| publishEvent('events', 'batch', { projectId, count }); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
Unhandled rejection risk: publishEvent calls are fire-and-forget.
publishEvent('events', 'batch', ...) at Line 267 returns a promise (packages/redis/publisher.ts forwards to redis.publish(...)), which is neither awaited nor .catch()-handled here. If the Redis pub/sub publish rejects (e.g. transient connection issue), this becomes an unhandled promise rejection, which by default crashes the Node process — a far worse outcome than the duplicate-insert bug this PR fixes, and it would take down the whole worker (all buffers), not just this flush.
🔒 Swallow/log publish errors instead of letting them propagate unhandled
for (const [projectId, count] of countByProject) {
- publishEvent('events', 'batch', { projectId, count });
+ publishEvent('events', 'batch', { projectId, count }).catch((err) => {
+ this.logger.warn({ err, projectId, count }, 'Failed to publish event batch notification');
+ });
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const countByProject = new Map<string, number>(); | |
| for (const line of chunk) { | |
| const projectId = extractProjectId(line); | |
| if (projectId) { | |
| countByProject.set( | |
| projectId, | |
| (countByProject.get(projectId) ?? 0) + 1, | |
| ); | |
| } | |
| } | |
| if ((i + 1) % yieldEvery === 0) { | |
| await this.yieldToEventLoop(); | |
| for (const [projectId, count] of countByProject) { | |
| publishEvent('events', 'batch', { projectId, count }); | |
| } | |
| const countByProject = new Map<string, number>(); | |
| for (const line of chunk) { | |
| const projectId = extractProjectId(line); | |
| if (projectId) { | |
| countByProject.set( | |
| projectId, | |
| (countByProject.get(projectId) ?? 0) + 1, | |
| ); | |
| } | |
| } | |
| for (const [projectId, count] of countByProject) { | |
| publishEvent('events', 'batch', { projectId, count }).catch((err) => { | |
| this.logger.warn({ err, projectId, count }, 'Failed to publish event batch notification'); | |
| }); | |
| } |
🤖 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 `@packages/db/src/buffers/event-buffer.ts` around lines 256 - 268, Handle the
promise returned by publishEvent in the batch publishing loop so Redis publish
failures are caught and logged rather than becoming unhandled rejections. Update
the loop around countByProject and publishEvent, preserving the existing
per-project event payload and allowing the buffer flush to continue for other
projects.
1 participant
Problem
The event buffer (
packages/db/src/buffers) can insert the same events into ClickHouse twice under two independent conditions. Since theeventstable has no dedup key, each one produces duplicate rows.1. Flush-lock expiry.
BaseBuffer.tryFlush()holds the Redis flush lock with a 60s TTL (lockTimeout = 60), but a ClickHouse insert can run up tomax_execution_time = 300sserver-side (INSERT_DEFAULT_SETTINGSinclickhouse/client.ts), and a flush inserts several chunks in sequence. A slow flush can outlive the 60s lock, letting a second worker replica acquire the lock and concurrently re-flush the same still-queued events.2. Partial-batch retry.
EventBuffer.processBuffer()inserted every chunk but only trimmed the whole slice from the Redis queue with a singleltrimat the very end. If a later chunk's insert threw, the error propagated before the trim, so the entire batch — including chunks that had already been inserted — was left on the queue and retried on the next cycle, duplicating the committed chunks.Fix
insert_deduplication_token(sha256 of the chunk's rows) as defense-in-depth, so an identical insert is rejected by ClickHouse even if the lock assumption is ever violated (e.g. a replica race on a replicated table).Because a safe front-of-queue trim requires committing chunks in order, this buffer's chunk inserts are now sequential rather than run through
parallelLimit. The other buffers (profile/replay/group) keep their parallel inserts. The raw JSONEachRow passthrough (noJSON.parseon the hot path) is preserved.Tests
Extends
event-buffer.test.ts:All 20 tests in the file pass locally (
vitest run src/buffers/event-buffer.test.ts, Redis running, ClickHouse mocked).Notes
NEVER CALL FORMATguidance, I did not run the formatter; the diff is limited to the logical change..changesetsetup).Summary by CodeRabbit