fix(replay): clamp ref flush window, real setting name, stable log_comment

Review follow-ups on the session_replay_blocks ref insert settings:

- REPLAY_BLOCKS_ASYNC_INSERT_TIMEOUT_MS is clamped to [200, 10000] ms with
  a warning (0 still means ClickHouse's adaptive default). A value at or
  above REPLAY_BLOCKS_TIMEOUT_MS (20 s) would abort every ref insert, and a
  non-numeric value such as "2s" now falls back to the 2 s default instead
  of parsing as a 2 ms window.
- Use async_insert_busy_timeout_max_ms (the real name; _ms is an alias the
  client types do not know).
- Pass a fixed log_comment on ref inserts. ClickHouse only coalesces async
  inserts with identical settings, and the client wrapper stamps a
  per-request log_comment (trace_id, span_id), so every ref insert was its
  own bucket and its own part regardless of the busy timeout. Caller
  settings override the wrapper's.
- Biome import formatting in the test.

Author: ayushjhanwar-png

packages/db/src/services/replay-block.service.test.ts

@@ -29,9 +29,11 @@ vi.mock('../clickhouse/client', () => ({
   ch: { insert: (args: any) => insertMock(args) },
 }));
 
-const { writeSessionReplayBlocks, replayBlockRefsInsertSettings } = await import(
-  './replay-block.service'
-);
+const {
+  writeSessionReplayBlocks,
+  replayBlockRefsInsertSettings,
+  parseRefsAsyncInsertTimeoutMs,
+} = await import('./replay-block.service');
 const { decompressReplayBlock } = await import('../blob/replay-blocks');
 
 const chunk = (
@@ -218,15 +220,57 @@ describe('writeSessionReplayBlocks', () => {
       async_insert: 1,
       wait_for_async_insert: 1,
       async_insert_use_adaptive_busy_timeout: 0,
-      async_insert_busy_timeout_ms: 2000,
+      async_insert_busy_timeout_max_ms: 2000,
     });
     // 0 falls back to ClickHouse's adaptive default
     const off = replayBlockRefsInsertSettings(0);
-    expect(off).not.toHaveProperty('async_insert_busy_timeout_ms');
+    expect(off).not.toHaveProperty('async_insert_busy_timeout_max_ms');
     expect(off).not.toHaveProperty('async_insert_use_adaptive_busy_timeout');
     expect(off.wait_for_async_insert).toBe(1);
   });
 
+  it('ref inserts carry identical settings (incl. a fixed log_comment) so CH can coalesce them', async () => {
+    await writeSessionReplayBlocks([
+      { line: chunk('A', 0, '2026-10-01 10:00:00.000'), offset: 1 },
+    ]);
+    await writeSessionReplayBlocks([
+      { line: chunk('B', 0, '2026-10-01 10:00:05.000'), offset: 2 },
+    ]);
+    const [first, second] = insertMock.mock.calls.map(
+      (c) => c[0].clickhouse_settings,
+    );
+    expect(first).toEqual(second);
+    // an explicit log_comment, so the client wrapper's per-request one
+    // (trace_id/span_id) does not split every insert into its own bucket
+    expect(typeof first.log_comment).toBe('string');
+    const comment = JSON.parse(first.log_comment);
+    expect(comment).not.toHaveProperty('trace_id');
+    expect(comment).not.toHaveProperty('span_id');
+    expect(comment).not.toHaveProperty('user_id');
+  });
+
+  it('clamps REPLAY_BLOCKS_ASYNC_INSERT_TIMEOUT_MS to [200, 10000] ms and warns', () => {
+    const warn = vi.fn();
+    expect(parseRefsAsyncInsertTimeoutMs(undefined, warn)).toBe(2000);
+    expect(parseRefsAsyncInsertTimeoutMs('', warn)).toBe(2000);
+    expect(parseRefsAsyncInsertTimeoutMs('0', warn)).toBe(0); // adaptive
+    expect(parseRefsAsyncInsertTimeoutMs('1500', warn)).toBe(1500);
+    expect(parseRefsAsyncInsertTimeoutMs('200', warn)).toBe(200);
+    expect(parseRefsAsyncInsertTimeoutMs('10000', warn)).toBe(10_000);
+    expect(warn).not.toHaveBeenCalled();
+
+    expect(parseRefsAsyncInsertTimeoutMs('2', warn)).toBe(200);
+    expect(parseRefsAsyncInsertTimeoutMs('20000', warn)).toBe(10_000);
+    expect(warn).toHaveBeenCalledTimes(2);
+
+    // not a whole number of ms (a typo like "2s"): default, not a 2 ms window
+    warn.mockClear();
+    expect(parseRefsAsyncInsertTimeoutMs('2s', warn)).toBe(2000);
+    expect(parseRefsAsyncInsertTimeoutMs('-5', warn)).toBe(2000);
+    expect(parseRefsAsyncInsertTimeoutMs('abc', warn)).toBe(2000);
+    expect(warn).toHaveBeenCalledTimes(3);
+  });
+
   it('propagates an append failure (caller decides) and inserts no refs', async () => {
     appendMock.mockRejectedValueOnce(new Error('azure down'));
     await expect(

packages/db/src/services/replay-block.service.ts

@@ -1,3 +1,4 @@
+import { createLogger } from '@openpanel/logger';
 import {
   appendReplayBlock,
   compressReplayBlock,
@@ -137,25 +138,71 @@ const BLOCK_WRITE_CONCURRENCY = (() => {
 // window turns that into roughly one part per node per window. The consumer
 // waits for the flush (wait_for_async_insert) before resolving offsets, so the
 // window is added latency per Kafka batch, well inside the batch deadline.
-// 0 = ClickHouse's default (adaptive) behaviour.
-const REFS_ASYNC_INSERT_TIMEOUT_MS = (() => {
-  const n = Number.parseInt(
-    process.env.REPLAY_BLOCKS_ASYNC_INSERT_TIMEOUT_MS ?? '',
-    10,
+// 0 = ClickHouse's default (adaptive) behaviour. Other values are clamped to
+// [MIN, MAX]: at or above REPLAY_BLOCKS_TIMEOUT_MS (20 s) every ref insert
+// would abort, and a typo like "2s" must not become a 2 ms window.
+export const REFS_ASYNC_INSERT_TIMEOUT_DEFAULT_MS = 2000;
+export const REFS_ASYNC_INSERT_TIMEOUT_MIN_MS = 200;
+export const REFS_ASYNC_INSERT_TIMEOUT_MAX_MS = 10_000;
+
+/** Parse REPLAY_BLOCKS_ASYNC_INSERT_TIMEOUT_MS (exported for tests). */
+export function parseRefsAsyncInsertTimeoutMs(
+  raw: string | undefined,
+  warn: (msg: string, meta: Record<string, unknown>) => void = () => {},
+): number {
+  const value = (raw ?? '').trim();
+  if (value === '') return REFS_ASYNC_INSERT_TIMEOUT_DEFAULT_MS;
+  if (!/^\d+$/.test(value)) {
+    warn(
+      'REPLAY_BLOCKS_ASYNC_INSERT_TIMEOUT_MS is not a whole number of ms; using the default',
+      { value: raw, usedMs: REFS_ASYNC_INSERT_TIMEOUT_DEFAULT_MS },
+    );
+    return REFS_ASYNC_INSERT_TIMEOUT_DEFAULT_MS;
+  }
+  const n = Number.parseInt(value, 10);
+  if (n === 0) return 0;
+  const clamped = Math.min(
+    REFS_ASYNC_INSERT_TIMEOUT_MAX_MS,
+    Math.max(REFS_ASYNC_INSERT_TIMEOUT_MIN_MS, n),
   );
-  return Number.isFinite(n) && n >= 0 ? n : 2000;
-})();
+  if (clamped !== n) {
+    warn('REPLAY_BLOCKS_ASYNC_INSERT_TIMEOUT_MS out of range; clamped', {
+      value: n,
+      usedMs: clamped,
+      minMs: REFS_ASYNC_INSERT_TIMEOUT_MIN_MS,
+      maxMs: REFS_ASYNC_INSERT_TIMEOUT_MAX_MS,
+    });
+  }
+  return clamped;
+}
+
+const REFS_ASYNC_INSERT_TIMEOUT_MS = parseRefsAsyncInsertTimeoutMs(
+  process.env.REPLAY_BLOCKS_ASYNC_INSERT_TIMEOUT_MS,
+  (msg, meta) => createLogger({ name: 'replay-blocks' }).warn(msg, meta),
+);
+
+// ClickHouse only coalesces async inserts whose settings are identical. The
+// client wrapper stamps a per-request log_comment (trace_id, span_id), which
+// made every ref insert its own async-insert bucket and so its own part,
+// whatever the busy timeout. A fixed log_comment here (caller settings win over
+// the wrapper's) keeps the settings identical across inserts; per-insert trace
+// attribution means little once inserts are flushed together anyway.
+const REFS_LOG_COMMENT = JSON.stringify({
+  service: process.env.OTEL_SERVICE_NAME ?? 'openpanel-unknown',
+  endpoint: 'replay.insertReplayBlocks',
+});
 
 /** ClickHouse settings for a refs insert (exported for tests). */
 export function replayBlockRefsInsertSettings(timeoutMs: number) {
   return {
     async_insert: 1,
     wait_for_async_insert: 1,
     input_format_skip_unknown_fields: 1,
+    log_comment: REFS_LOG_COMMENT,
     ...(timeoutMs > 0
       ? {
           async_insert_use_adaptive_busy_timeout: 0,
-          async_insert_busy_timeout_ms: timeoutMs,
+          async_insert_busy_timeout_max_ms: timeoutMs,
         }
       : {}),
   } as const;