From 18774deded7517c43d0da8476cd79f8aacf262bc Mon Sep 17 00:00:00 2001 From: Raoni Timo Date: Tue, 21 Jul 2026 19:59:47 -0300 Subject: [PATCH] fix(db): prevent duplicate event inserts on lock expiry and partial-batch retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- packages/db/src/buffers/base-buffer.ts | 10 +- packages/db/src/buffers/event-buffer.test.ts | 97 ++++++++++++++ packages/db/src/buffers/event-buffer.ts | 127 +++++++++++-------- 3 files changed, 182 insertions(+), 52 deletions(-) diff --git a/packages/db/src/buffers/base-buffer.ts b/packages/db/src/buffers/base-buffer.ts index 5c279788a..9e4919387 100644 --- a/packages/db/src/buffers/base-buffer.ts +++ b/packages/db/src/buffers/base-buffer.ts @@ -59,7 +59,15 @@ export class BaseBuffer { name: string; logger: ILogger; lockKey: string; - lockTimeout = 60; + // Flush-lock TTL, in seconds. It MUST comfortably exceed the maximum time a + // flush can spend inserting into ClickHouse, otherwise the lock can expire + // mid-flush and a second worker replica acquires it and re-flushes the same + // still-queued rows -> duplicate inserts (the `events` table has no dedup + // key). The server-side insert ceiling is max_execution_time = 300s (see + // INSERT_DEFAULT_SETTINGS in clickhouse/client.ts), and a flush inserts + // several chunks in sequence, so 60s was far too short. 360s keeps the lock + // held past the worst-case insert while still self-healing if a worker dies. + lockTimeout = 360; onFlush: () => Promise | void; enableParallelProcessing: boolean; diff --git a/packages/db/src/buffers/event-buffer.test.ts b/packages/db/src/buffers/event-buffer.test.ts index 66e95f130..de6626573 100644 --- a/packages/db/src/buffers/event-buffer.test.ts +++ b/packages/db/src/buffers/event-buffer.test.ts @@ -305,6 +305,103 @@ describe('EventBuffer', () => { expect(await eventBuffer.getBufferSize()).toBe(5); }); + it('does not replay already-committed chunks when a later chunk fails', async () => { + const prev = process.env.EVENT_BUFFER_CHUNK_SIZE; + process.env.EVENT_BUFFER_CHUNK_SIZE = '2'; + const eb = new EventBuffer(); + + for (let i = 0; i < 4; i++) { + eb.add({ + project_id: 'p13', + name: `event${i}`, + created_at: new Date(Date.now() + i).toISOString(), + } as any); + } + await eb.flush(); + expect(await eb.getBufferSize()).toBe(4); + + // First chunk inserts fine; the second chunk fails mid-batch. + const insertSpy = vi + .spyOn(ch, 'insert') + .mockResolvedValueOnce(undefined as any) + .mockRejectedValueOnce(new Error('ClickHouse unavailable')); + + await expect(eb.processBuffer()).rejects.toThrow('ClickHouse unavailable'); + + // The committed chunk is trimmed; the failed chunk is retained. Crucially + // the committed chunk is NOT left in the queue, so it cannot be replayed. + expect(insertSpy).toHaveBeenCalledTimes(2); + expect(await eb.getBufferSize()).toBe(2); + + // Retry: only the remaining chunk is inserted — committed events are never + // re-sent, so no duplicate rows. + insertSpy.mockReset(); + insertSpy.mockResolvedValue(undefined as any); + + await eb.processBuffer(); + + expect(insertSpy).toHaveBeenCalledTimes(1); + const retryLines = await streamToLines( + insertSpy.mock.calls[0]![0].values as Readable + ); + const retryNames = retryLines.map((line) => JSON.parse(line).name); + expect(retryNames).toEqual(['event2', 'event3']); + expect(await eb.getBufferSize()).toBe(0); + + if (prev === undefined) { + delete process.env.EVENT_BUFFER_CHUNK_SIZE; + } else { + process.env.EVENT_BUFFER_CHUNK_SIZE = prev; + } + insertSpy.mockRestore(); + }); + + it('sends a deterministic dedup token so retried inserts are idempotent', async () => { + const makeEvents = () => + [ + { + project_id: 'p14', + name: 'event_a', + created_at: new Date(1000).toISOString(), + }, + { + project_id: 'p14', + name: 'event_b', + created_at: new Date(2000).toISOString(), + }, + ] as any[]; + + const insertSpy = vi + .spyOn(ch, 'insert') + .mockResolvedValue(undefined as any); + + const eb1 = new EventBuffer(); + eb1.bulkAdd(makeEvents()); + await eb1.flush(); + await eb1.processBuffer(); + + const token1 = (insertSpy.mock.calls[0]![0] as any).clickhouse_settings + ?.insert_deduplication_token as string; + + insertSpy.mockClear(); + + // Same content queued again (e.g. a lock-expiry double-flush): the token + // must match so ClickHouse can reject the duplicate insert. + const eb2 = new EventBuffer(); + eb2.bulkAdd(makeEvents()); + await eb2.flush(); + await eb2.processBuffer(); + + const token2 = (insertSpy.mock.calls[0]![0] as any).clickhouse_settings + ?.insert_deduplication_token as string; + + expect(typeof token1).toBe('string'); + expect(token1.length).toBeGreaterThan(0); + expect(token1).toBe(token2); + + insertSpy.mockRestore(); + }); + it('retains events in queue when ClickHouse insert fails', async () => { eventBuffer.add({ project_id: 'p12', diff --git a/packages/db/src/buffers/event-buffer.ts b/packages/db/src/buffers/event-buffer.ts index d29003f63..6156a3075 100644 --- a/packages/db/src/buffers/event-buffer.ts +++ b/packages/db/src/buffers/event-buffer.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto'; import { getRedisCache, publishEvent } from '@openpanel/redis'; import { ch, chQuery } from '../clickhouse/client'; import type { IClickhouseEvent } from '../services/event.service'; @@ -173,6 +174,24 @@ export class EventBuffer extends BaseBuffer { return this.queueKey; } + /** + * 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'); + } + async processBuffer() { const redis = getRedisCache(); @@ -189,64 +208,70 @@ export class EventBuffer extends BaseBuffer { return; } - // We don't need to JSON.parse the events at all — they're already - // valid JSONEachRow lines (one stringified event per Redis entry). - // The client's custom `json.stringify` (set in CLICKHOUSE_OPTIONS) - // passes strings through unchanged, so the bytes go straight from - // Redis → CH HTTP body. This skips: - // - JSON.parse × N (50–300ms for N=100k) - // - The @clickhouse/client's internal JSON.stringify × N (same) - // - All the intermediate object allocations (saves ~200MB heap) + // Process chunks in queue order, committing each one before the next. + // A chunk is trimmed from the front of the queue only AFTER its insert + // succeeds, and its per-project realtime notification is published only + // then. If a chunk insert throws, we stop and let the error propagate to + // tryFlush: already-committed chunks stay trimmed, and only the untrimmed + // remainder is retried on the next cycle. Previously every chunk was + // inserted and the whole slice was trimmed in a single `ltrim` at the + // very end, so a failure on a later chunk replayed the already-inserted + // earlier chunks on retry — duplicate rows, since the `events` table has + // no dedup key. // - // We still need `project_id` per row for the per-project pub/sub. - // extractProjectId() does an indexOf-based fast path that's ~50× - // faster than JSON.parse, and falls back to a real parse on the - // rare line where `project_id` appears more than once (e.g. a - // user-supplied `properties.project_id`) — so the count is always - // attributed to the top-level field, never a nested one. - const countByProject = new Map(); - const yieldEvery = this.getYieldInterval(queueEvents.length, { - min: 1000, - max: 5000, - }); - for (let i = 0; i < queueEvents.length; i++) { - const projectId = extractProjectId(queueEvents[i]!); - if (projectId) { - countByProject.set( - projectId, - (countByProject.get(projectId) ?? 0) + 1, - ); + // We never JSON.parse the events: they're already valid JSONEachRow lines + // (one stringified event per Redis entry), and the client's custom + // `json.stringify` (set in CLICKHOUSE_OPTIONS) passes strings through + // unchanged, so the bytes go straight from Redis → CH HTTP body. + // `project_id` for the pub/sub is pulled with extractProjectId()'s + // indexOf fast path (falls back to JSON.parse only on the rare line where + // `project_id` appears more than once). + const chunks = this.chunks(queueEvents, this.chunkSize); + let rowsProcessed = 0; + let chInsertMs = 0; + let trimMs = 0; + + for (const chunk of chunks) { + const chStart = performance.now(); + await ch.insert({ + table: 'events', + // Stream the raw JSONEachRow lines straight through — already + // serialized in Redis, no client-side parse/stringify needed. + values: this.jsonEachRowStream(chunk), + format: 'JSONEachRow', + clickhouse_settings: { + ...this.getClickhouseSettings(), + insert_deduplication_token: this.deduplicationToken(chunk), + }, + }); + chInsertMs += performance.now() - chStart; + + // Commit this chunk: remove exactly the entries we just inserted from + // the front of the queue. Order is preserved because we always insert + // and trim the head chunk first. + const trimStart = performance.now(); + await redis.ltrim(this.queueKey, chunk.length, -1); + trimMs += performance.now() - trimStart; + + const countByProject = new Map(); + 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 chStart = performance.now(); - await this.parallelLimit( - this.chunks(queueEvents, this.chunkSize), - (chunk) => - ch.insert({ - table: 'events', - // Stream the raw JSONEachRow lines straight through — already - // serialized in Redis, no client-side parse/stringify needed. - values: this.jsonEachRowStream(chunk), - format: 'JSONEachRow', - clickhouse_settings: this.getClickhouseSettings(), - }), - ); - const chInsertMs = performance.now() - chStart; - - for (const [projectId, count] of countByProject) { - publishEvent('events', 'batch', { projectId, count }); + rowsProcessed += chunk.length; } - const trimStart = performance.now(); - await redis.ltrim(this.queueKey, queueEvents.length, -1); - const trimMs = performance.now() - trimStart; - this.reportFlushStats({ - rowsProcessed: queueEvents.length, + rowsProcessed, phases: { lrangeMs, chInsertMs, trimMs }, }); }