From 80fd7623d3ebb065d32df4ef5d5e238b4f2b82b6 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 23 Jul 2026 23:21:56 +0100 Subject: [PATCH 1/2] fix(webapp): keep the rest of a batch when one run or span has un-ingestable JSON When a run's output/payload or a trace span's attributes contained JSON ClickHouse could not ingest (e.g. an object nested past its depth limit), the analytics insert aborted the whole batch. Because runs and events are written in per-batch inserts, one bad row made every other run or span in that batch go missing from the runs list, traces, and logs, even though the Postgres rows were fine. Both ClickHouse write paths now fall back to retrying the insert with input_format_allow_errors so ClickHouse skips only the rows it can't parse and commits the rest. The healthy insert path is unchanged; a successful row isolation logs at info, only a failed isolation insert is an error. --- ...uns-list-batch-drop-on-unparseable-json.md | 6 + .../services/runsReplicationService.server.ts | 264 +++++++++++------ .../clickhouseEventRepository.server.ts | 223 ++++++++++----- ...ckhouseEventRepositoryJsonRecovery.test.ts | 123 ++++++++ .../test/runsReplicationBenchmark.producer.ts | 23 ++ ...nsReplicationJsonRecoveryBenchmark.test.ts | 267 ++++++++++++++++++ .../runsReplicationService.part10.test.ts | 143 ++++++++++ 7 files changed, 885 insertions(+), 164 deletions(-) create mode 100644 .server-changes/fix-runs-list-batch-drop-on-unparseable-json.md create mode 100644 apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts create mode 100644 apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts create mode 100644 apps/webapp/test/runsReplicationService.part10.test.ts diff --git a/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md b/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md new file mode 100644 index 00000000000..2f35f93733e --- /dev/null +++ b/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fixed a rare case where a single run or span carrying data that could not be ingested would make other runs or trace events in the same batch go missing from the runs list, traces, and logs. Now only the affected item is dropped and everything else is stored normally. diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts index 732665461dd..5ffb38bd5f8 100644 --- a/apps/webapp/app/services/runsReplicationService.server.ts +++ b/apps/webapp/app/services/runsReplicationService.server.ts @@ -1,6 +1,7 @@ import type { ClickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { type ClickHouse, + type ClickHouseSettings, type PayloadInsertArray, type TaskRunInsertArray, composeTaskRunVersion, @@ -175,14 +176,31 @@ export class RunsReplicationService { private _disableErrorFingerprinting: boolean; /** - * Counts batches that hit a ClickHouse `Cannot parse JSON object` failure - * that survived one sanitize-retry. These batches are dropped on the floor - * (returning success-ish to the caller so the retry layer doesn't spin on - * the same deterministic failure), and we track the drop count for - * observability. Counter only — does not gate behaviour. + * Counts batches dropped whole because even the row-isolation retry failed + * to insert anything (an unexpected terminal case). The common recovery now + * isolates the un-ingestable rows and lands the rest, so this should stay at + * zero in practice. Counter only — does not gate behaviour. */ private _permanentlyDroppedBatches = 0; + /** + * Counts batches that took the row-isolation recovery path — a + * `Cannot parse JSON object` failure the sanitizer could not repair, where we + * re-inserted with `input_format_allow_errors_*` so ClickHouse skipped the + * un-ingestable rows and landed the rest. Reliable per-event signal that user + * data is hitting the ceiling. + */ + private _rowIsolationRecoveries = 0; + + /** + * Best-effort count of individual rows ClickHouse skipped across those + * row-isolation retries, derived from the insert summary's `written_rows` + * when the client surfaces it (it is not always populated under concurrent + * inserts, so treat this as a lower bound; `_rowIsolationRecoveries` is the + * dependable event count). + */ + private _permanentlyDroppedRows = 0; + // Metrics private _replicationLagHistogram: Histogram; private _batchesFlushedCounter: Counter; @@ -416,6 +434,16 @@ export class RunsReplicationService { return this._permanentlyDroppedBatches; } + /** Exposed for tests and metrics — batches that took the row-isolation recovery path. */ + get rowIsolationRecoveries() { + return this._rowIsolationRecoveries; + } + + /** Exposed for tests and metrics — best-effort count of individual rows ClickHouse skipped during row-isolation retries. */ + get permanentlyDroppedRows() { + return this._permanentlyDroppedRows; + } + public async shutdown() { if (this._isShuttingDown) return; @@ -870,15 +898,11 @@ export class RunsReplicationService { payloadError = plErr; } - // Only count rows that actually landed in ClickHouse. `kind: "dropped"` - // means the recovery wrapper bailed (sanitizer no-op or sanitize-retry - // still failed) — those rows never made it, so they must not show up - // as successful inserts in the per-batch counter. - if (!trErr && trOutcome?.kind !== "dropped") { - this._taskRunsInsertedCounter.add(group.taskRunInserts.length); + if (!trErr && trOutcome && trOutcome.kind !== "dropped") { + this._taskRunsInsertedCounter.add(landedRowCount(group.taskRunInserts.length, trOutcome)); } - if (!plErr && plOutcome?.kind !== "dropped") { - this._payloadsInsertedCounter.add(group.payloadInserts.length); + if (!plErr && plOutcome && plOutcome.kind !== "dropped") { + this._payloadsInsertedCounter.add(landedRowCount(group.payloadInserts.length, plOutcome)); } } @@ -1034,10 +1058,14 @@ export class RunsReplicationService { return; } return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => { - const doInsert = async () => { + const doInsert = async (extraSettings?: ClickHouseSettings) => { const [insertError, insertResult] = await clickhouse.taskRuns.insertCompactArrays( taskRunInserts, - { params: { clickhouse_settings: this.#getClickhouseInsertSettings() } } + { + params: { + clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings }, + }, + } ); if (insertError) { this.logger.error("Error inserting task run inserts attempt", { @@ -1068,10 +1096,14 @@ export class RunsReplicationService { return; } return await startSpan(this._tracer, "insertPayloadInserts", async (span) => { - const doInsert = async () => { + const doInsert = async (extraSettings?: ClickHouseSettings) => { const [insertError, insertResult] = await clickhouse.taskRuns.insertPayloadsCompactArrays( payloadInserts, - { params: { clickhouse_settings: this.#getClickhouseInsertSettings() } } + { + params: { + clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings }, + }, + } ); if (insertError) { this.logger.error("Error inserting payload inserts attempt", { @@ -1094,36 +1126,38 @@ export class RunsReplicationService { } /** - * Wraps a ClickHouse insert with reactive UTF-16 sanitization for - * `Cannot parse JSON object` rejections. Mirrors the pattern from - * `ClickhouseEventRepository.#insertWithJsonParseRecovery` introduced - * in #3659 — same root cause (lone UTF-16 surrogates in user-provided - * JSON), same recovery shape: + * Wraps a ClickHouse insert with recovery for `Cannot parse JSON object` + * rejections so one un-ingestable row never drops its whole batch. Builds on + * the sanitize pattern from `ClickhouseEventRepository.#insertWithJsonParseRecovery` + * (#3659) and adds a final row-isolation fallback: * - * 1. Try the insert. Healthy batches pay zero scan cost. - * 2. On parse error, walk the whole batch via `sanitizeRows` and - * replace any lone-surrogate string with `"[invalid-utf16]"`. - * 3. Retry once. If the sanitizer found nothing or the retry also - * fails with the same error class, drop the batch loudly and - * return — do NOT rethrow, otherwise the surrounding - * `#insertWithRetry` layer would spin three more times on the - * same deterministic failure. - * 4. Non-parse errors propagate unchanged so the existing - * transient-retry path still handles them. - * - * The whole-batch scan (rather than slicing on the `at row N` hint) is - * deliberate: `at row N` semantics under `input_format_parallel_parsing` - * aren't stable enough to safely skip rows. The cost is bounded because - * `detectBadJsonStrings` exits in O(1) for clean strings. + * 1. Try the insert. Healthy batches pay zero recovery cost. + * 2. On a parse error, walk the batch via `sanitizeRows` and losslessly + * repair what we can in place (lone UTF-16 surrogates → `"[invalid-utf16]"`, + * out-of-range integers → their string form), then retry once. A clean + * retry keeps every row. + * 3. If the sanitizer found nothing to fix, or the retry still hits the same + * parse error (a structurally un-ingestable row, e.g. JSON nested past + * ClickHouse's `input_format_json_max_depth`), retry once more with + * `input_format_allow_errors_*` so ClickHouse skips only the un-parseable + * rows and lands the rest. Verified against ClickHouse 26.2 (the cloud + * version): the good rows land, the JSONCompactEachRowWithNames header is + * unaffected, both sync and async. + * 4. Only if that row-isolation insert itself fails do we drop the batch, + * and never rethrow a parse error — otherwise the surrounding + * `#insertWithRetry` layer would spin on the same deterministic failure. + * 5. Non-parse errors propagate unchanged so the transient-retry path still + * handles them. */ async #insertWithJsonParseRecovery( rows: T[], - doInsert: () => Promise, + doInsert: (extraSettings?: ClickHouseSettings) => Promise, contextLabel: string, attempt: number ): Promise< | { kind: "inserted"; insertResult: unknown } | { kind: "sanitized"; insertResult: unknown } + | { kind: "isolated"; insertResult: unknown; rowsSkipped: number | null } | { kind: "dropped" } > { try { @@ -1131,65 +1165,93 @@ export class RunsReplicationService { } catch (firstError) { if (!isClickHouseJsonParseError(firstError)) throw firstError; - const firstMessage = - typeof firstError === "object" && firstError !== null && "message" in firstError - ? String((firstError as { message?: unknown }).message ?? "") - : String(firstError); - + const firstMessage = errorMessage(firstError); const rowHint = parseRowNumberFromError(firstMessage); const { rowsTouched, fieldsSanitized } = sanitizeRows(rows); - if (fieldsSanitized === 0) { - this._permanentlyDroppedBatches += 1; - this.logger.error( - "Dropped batch — ClickHouse JSON parse error but sanitizer found nothing to fix", - { - contextLabel, - attempt, - batchSize: rows.length, - clickhouseRowHint: rowHint, - permanentlyDroppedBatches: this._permanentlyDroppedBatches, - sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), - clickhouseError: firstMessage.split("\n")[0], - } - ); - return { kind: "dropped" }; + if (fieldsSanitized > 0) { + this.logger.warn("Sanitizing batch after ClickHouse JSON parse error", { + contextLabel, + attempt, + batchSize: rows.length, + clickhouseRowHint: rowHint, + rowsTouched, + fieldsSanitized, + clickhouseError: firstMessage.split("\n")[0], + }); + + try { + return { kind: "sanitized", insertResult: await doInsert() }; + } catch (retryError) { + if (!isClickHouseJsonParseError(retryError)) throw retryError; + } } - this.logger.warn("Sanitizing batch after ClickHouse JSON parse error", { - contextLabel, - attempt, - batchSize: rows.length, - clickhouseRowHint: rowHint, - rowsTouched, - fieldsSanitized, - clickhouseError: firstMessage.split("\n")[0], + return await this.#insertIsolatingBadRows(rows, doInsert, contextLabel, attempt, firstMessage); + } + } + + /** + * Last-resort recovery: re-run the insert with `input_format_allow_errors_*` + * so ClickHouse skips the rows it cannot parse and lands the rest, rather than + * aborting the whole INSERT. Forces a synchronous insert so `written_rows` + * reflects exactly what landed and we can count what was lost. + */ + async #insertIsolatingBadRows( + rows: T[], + doInsert: (extraSettings?: ClickHouseSettings) => Promise, + contextLabel: string, + attempt: number, + firstMessage: string + ): Promise< + { kind: "isolated"; insertResult: unknown; rowsSkipped: number | null } | { kind: "dropped" } + > { + try { + const insertResult = await doInsert({ + input_format_allow_errors_num: String(rows.length), + input_format_allow_errors_ratio: 1, + async_insert: 0, }); - try { - return { kind: "sanitized", insertResult: await doInsert() }; - } catch (retryError) { - if (!isClickHouseJsonParseError(retryError)) throw retryError; - - this._permanentlyDroppedBatches += 1; - const retryMessage = - typeof retryError === "object" && retryError !== null && "message" in retryError - ? String((retryError as { message?: unknown }).message ?? "") - : String(retryError); - this.logger.error( - "Dropped batch after sanitize-retry still hit ClickHouse JSON parse error", - { - contextLabel, - attempt, - batchSize: rows.length, - permanentlyDroppedBatches: this._permanentlyDroppedBatches, - sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), - firstError: firstMessage.split("\n")[0], - retryError: retryMessage.split("\n")[0], - } - ); - return { kind: "dropped" }; + this._rowIsolationRecoveries += 1; + const writtenRows = extractWrittenRows(insertResult); + const rowsSkipped = writtenRows === null ? null : Math.max(0, rows.length - writtenRows); + if (rowsSkipped !== null) { + this._permanentlyDroppedRows += rowsSkipped; } + + this.logger.info( + "Isolated un-ingestable rows after ClickHouse JSON parse error — landed the rest of the batch", + { + contextLabel, + attempt, + batchSize: rows.length, + rowsSkipped, + rowIsolationRecoveries: this._rowIsolationRecoveries, + permanentlyDroppedRows: this._permanentlyDroppedRows, + sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), + clickhouseError: firstMessage.split("\n")[0], + } + ); + + return { kind: "isolated", insertResult, rowsSkipped }; + } catch (isolationError) { + if (!isClickHouseJsonParseError(isolationError)) throw isolationError; + + this._permanentlyDroppedBatches += 1; + this.logger.error( + "Dropped batch — row-isolation insert failed after ClickHouse JSON parse error", + { + contextLabel, + attempt, + batchSize: rows.length, + permanentlyDroppedBatches: this._permanentlyDroppedBatches, + sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), + firstError: firstMessage.split("\n")[0], + isolationError: errorMessage(isolationError).split("\n")[0], + } + ); + return { kind: "dropped" }; } } @@ -1542,3 +1604,27 @@ function lsnToUInt64(lsn: string): bigint { const [seg, off] = lsn.split("/"); return (BigInt("0x" + seg) << 32n) | BigInt("0x" + off); } + +function errorMessage(err: unknown): string { + return typeof err === "object" && err !== null && "message" in err + ? String((err as { message?: unknown }).message ?? "") + : String(err); +} + +function landedRowCount( + groupSize: number, + outcome: { kind: string; rowsSkipped?: number | null } +): number { + if (outcome.kind === "isolated" && typeof outcome.rowsSkipped === "number") { + return Math.max(0, groupSize - outcome.rowsSkipped); + } + return groupSize; +} + +function extractWrittenRows(insertResult: unknown): number | null { + const written = (insertResult as { summary?: { written_rows?: string } } | null | undefined) + ?.summary?.written_rows; + if (written === undefined) return null; + const parsed = Number(written); + return Number.isFinite(parsed) ? parsed : null; +} diff --git a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts index 573097005ed..417b9bffa7f 100644 --- a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts @@ -1,5 +1,6 @@ import type { ClickHouse, + ClickHouseSettings, LlmMetricsV1Input, MetricsV1Input, TaskEventDetailedSummaryV1Result, @@ -121,14 +122,31 @@ export class ClickhouseEventRepository implements IEventRepository { private _tracer: Tracer; private _version: "v1" | "v2"; /** - * Counts batches that hit a ClickHouse JSON parse failure that survived - * one sanitize-retry. These batches are dropped on the floor (the scheduler - * is told the flush "succeeded" so its queue counter doesn't leak), and we - * track the drop count for observability. + * Counts batches dropped whole because even the row-isolation retry failed + * to insert anything (an unexpected terminal case). The common recovery now + * isolates the un-ingestable rows and lands the rest, so this should stay at + * zero in practice. */ private _permanentlyDroppedBatches = 0; private readonly _droppedBatchesCounter: Counter; + /** + * Counts batches that took the row-isolation recovery path — a + * `Cannot parse JSON object` failure the sanitizer could not repair, where we + * re-inserted with `input_format_allow_errors_*` so ClickHouse skipped the + * un-ingestable rows and landed the rest. Reliable per-event signal. + */ + private _rowIsolationRecoveries = 0; + private readonly _rowsIsolatedCounter: Counter; + + /** + * Best-effort count of individual rows ClickHouse skipped across those + * row-isolation retries, from the insert summary's `written_rows` when the + * client surfaces it (not always populated under concurrent inserts, so treat + * as a lower bound; `_rowIsolationRecoveries` is the dependable event count). + */ + private _permanentlyDroppedRows = 0; + constructor(config: ClickhouseEventRepositoryConfig) { this._clickhouse = config.clickhouse; this._config = config; @@ -140,6 +158,11 @@ export class ClickhouseEventRepository implements IEventRepository { description: "Batches permanently dropped after an unrecoverable ClickHouse JSON parse error", unit: "batches", }); + this._rowsIsolatedCounter = meter.createCounter("ingest.flush.rows_isolated", { + description: + "Batches that recovered by isolating un-ingestable rows (landed the rest) after a ClickHouse JSON parse error", + unit: "batches", + }); this._flushScheduler = new DynamicFlushScheduler({ name: `task_events_${this._version}`, @@ -194,6 +217,16 @@ export class ClickhouseEventRepository implements IEventRepository { return this._permanentlyDroppedBatches; } + /** Exposed for tests and metrics — batches that took the row-isolation recovery path. */ + get rowIsolationRecoveries() { + return this._rowIsolationRecoveries; + } + + /** Exposed for tests and metrics — best-effort count of individual rows ClickHouse skipped during row-isolation retries. */ + get permanentlyDroppedRows() { + return this._permanentlyDroppedRows; + } + /** * Clamps a start time (in nanoseconds) to now if it's too far in the past. * Returns the clamped value as a bigint. @@ -262,10 +295,10 @@ export class ClickhouseEventRepository implements IEventRepository { ? this._clickhouse.taskEventsV2.insert : this._clickhouse.taskEvents.insert; - const doInsert = async () => { + const doInsert = async (extraSettings?: ClickHouseSettings) => { const [insertError, insertResult] = await insertFn(events, { params: { - clickhouse_settings: this.#getClickhouseInsertSettings(), + clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings }, }, }); if (insertError) throw insertError; @@ -296,10 +329,10 @@ export class ClickhouseEventRepository implements IEventRepository { } async #flushLlmMetricsBatch(flushId: string, rows: LlmMetricsV1Input[]) { - const doInsert = async () => { + const doInsert = async (extraSettings?: ClickHouseSettings) => { const [insertError, insertResult] = await this._clickhouse.llmMetrics.insert(rows, { params: { - clickhouse_settings: this.#getClickhouseInsertSettings(), + clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings }, }, }); if (insertError) throw insertError; @@ -324,28 +357,35 @@ export class ClickhouseEventRepository implements IEventRepository { } /** - * Wraps a ClickHouse insert callable with reactive UTF-16 sanitization. + * Wraps a ClickHouse insert with recovery for `Cannot parse JSON object` + * rejections so one un-ingestable event never drops its whole batch. * - * On a `Cannot parse JSON object` failure: - * 1. Sanitize the batch from `max(0, parsedRowN - 1)` onwards (rows - * before the failing one parsed fine — known good). - * 2. Retry the insert once with the sanitized batch. - * 3. If the retry still fails with the same error class, log loudly, - * increment `permanentlyDroppedBatches`, and return without - * throwing — the scheduler's transient-retry path would just repeat - * the same deterministic failure. + * 1. Try the insert. Healthy batches pay zero recovery cost. + * 2. On a parse error, `sanitizeRows` losslessly repairs what it can in + * place (lone UTF-16 surrogates, out-of-range integers), then retries + * once. A clean retry keeps every row. + * 3. If the sanitizer found nothing, or the retry still hits the same parse + * error (a structurally un-ingestable row, e.g. JSON nested past + * ClickHouse's `input_format_json_max_depth`), retry once more with + * `input_format_allow_errors_*` so ClickHouse skips only the un-parseable + * rows and lands the rest. Verified against ClickHouse 26.2 (the cloud + * version) with the JSONEachRow insert format. + * 4. Only if that row-isolation insert itself fails do we drop the batch, + * and never rethrow a parse error — the scheduler would just repeat the + * same deterministic failure. * - * Non-parse errors propagate unchanged so the scheduler's existing - * backoff/retry behaviour still handles transient network or CH issues. + * Non-parse errors propagate unchanged so the scheduler's backoff/retry still + * handles transient network or CH issues. */ async #insertWithJsonParseRecovery( flushId: string, rows: T[], - doInsert: () => Promise, + doInsert: (extraSettings?: ClickHouseSettings) => Promise, contextLabel: string ): Promise< | { kind: "inserted"; insertResult: unknown } | { kind: "sanitized"; insertResult: unknown } + | { kind: "isolated"; insertResult: unknown; rowsSkipped: number | null } | { kind: "dropped" } > { try { @@ -353,73 +393,92 @@ export class ClickhouseEventRepository implements IEventRepository { } catch (firstError) { if (!isClickHouseJsonParseError(firstError)) throw firstError; - const firstMessage = - typeof firstError === "object" && firstError !== null && "message" in firstError - ? String((firstError as { message?: unknown }).message ?? "") - : String(firstError); - - // Sanitize the whole batch. ClickHouse's `at row N` index is logged - // for observability but not used to slice — its semantics under - // parallel parsing are not stable enough to safely skip rows. + const firstMessage = errorMessage(firstError); const rowHint = parseRowNumberFromError(firstMessage); const { rowsTouched, fieldsSanitized } = sanitizeRows(rows); - // Sanitizer found nothing to fix → retrying the exact same batch is - // guaranteed to hit the same deterministic parse failure. Skip the - // wasted ClickHouse round-trip and drop loudly. Throwing instead would - // hand the failure back to the scheduler's 3× transient-retry loop — - // exactly the retry storm this wrapper is designed to avoid. - if (fieldsSanitized === 0) { - this._permanentlyDroppedBatches += 1; - this._droppedBatchesCounter.add(1, { table: contextLabel }); - logger.error( - "Dropped batch — ClickHouse JSON parse error but sanitizer found nothing to fix", - { - flushId, - contextLabel, - batchSize: rows.length, - clickhouseRowHint: rowHint, - permanentlyDroppedBatches: this._permanentlyDroppedBatches, - sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), - clickhouseError: firstMessage.split("\n")[0], - } - ); - return { kind: "dropped" }; + if (fieldsSanitized > 0) { + logger.warn("Sanitizing batch after ClickHouse JSON parse error", { + flushId, + contextLabel, + batchSize: rows.length, + clickhouseRowHint: rowHint, + rowsTouched, + fieldsSanitized, + clickhouseError: firstMessage.split("\n")[0], + }); + + try { + return { kind: "sanitized", insertResult: await doInsert() }; + } catch (retryError) { + if (!isClickHouseJsonParseError(retryError)) throw retryError; + } } - logger.warn("Sanitizing batch after ClickHouse JSON parse error", { - flushId, - contextLabel, - batchSize: rows.length, - clickhouseRowHint: rowHint, - rowsTouched, - fieldsSanitized, - clickhouseError: firstMessage.split("\n")[0], + return await this.#insertIsolatingBadRows(flushId, rows, doInsert, contextLabel, firstMessage); + } + } + + /** + * Last-resort recovery: re-run the insert with `input_format_allow_errors_*` + * so ClickHouse skips the rows it cannot parse and lands the rest, rather than + * aborting the whole INSERT. Forces a synchronous insert so `written_rows` + * reflects what landed and we can count what was lost. + */ + async #insertIsolatingBadRows( + flushId: string, + rows: T[], + doInsert: (extraSettings?: ClickHouseSettings) => Promise, + contextLabel: string, + firstMessage: string + ): Promise< + { kind: "isolated"; insertResult: unknown; rowsSkipped: number | null } | { kind: "dropped" } + > { + try { + const insertResult = await doInsert({ + input_format_allow_errors_num: String(rows.length), + input_format_allow_errors_ratio: 1, + async_insert: 0, }); - try { - return { kind: "sanitized", insertResult: await doInsert() }; - } catch (retryError) { - if (!isClickHouseJsonParseError(retryError)) throw retryError; - - this._permanentlyDroppedBatches += 1; - this._droppedBatchesCounter.add(1, { table: contextLabel }); - const retryMessage = - typeof retryError === "object" && retryError !== null && "message" in retryError - ? String((retryError as { message?: unknown }).message ?? "") - : String(retryError); - logger.error("Dropped batch after sanitize-retry still hit ClickHouse JSON parse error", { + this._rowIsolationRecoveries += 1; + this._rowsIsolatedCounter.add(1, { table: contextLabel }); + const writtenRows = extractWrittenRows(insertResult); + const rowsSkipped = writtenRows === null ? null : Math.max(0, rows.length - writtenRows); + if (rowsSkipped !== null) { + this._permanentlyDroppedRows += rowsSkipped; + } + + logger.info( + "Isolated un-ingestable rows after ClickHouse JSON parse error — landed the rest of the batch", + { flushId, contextLabel, batchSize: rows.length, - permanentlyDroppedBatches: this._permanentlyDroppedBatches, + rowsSkipped, + rowIsolationRecoveries: this._rowIsolationRecoveries, + permanentlyDroppedRows: this._permanentlyDroppedRows, sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), - firstError: firstMessage.split("\n")[0], - retryError: retryMessage.split("\n")[0], - }); + clickhouseError: firstMessage.split("\n")[0], + } + ); - return { kind: "dropped" }; - } + return { kind: "isolated", insertResult, rowsSkipped }; + } catch (isolationError) { + if (!isClickHouseJsonParseError(isolationError)) throw isolationError; + + this._permanentlyDroppedBatches += 1; + this._droppedBatchesCounter.add(1, { table: contextLabel }); + logger.error("Dropped batch — row-isolation insert failed after ClickHouse JSON parse error", { + flushId, + contextLabel, + batchSize: rows.length, + permanentlyDroppedBatches: this._permanentlyDroppedBatches, + sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), + firstError: firstMessage.split("\n")[0], + isolationError: errorMessage(isolationError).split("\n")[0], + }); + return { kind: "dropped" }; } } @@ -2942,3 +3001,17 @@ function formatClickhouseUnsignedIntegerString(value: number | bigint): string { return Math.floor(value).toString(); } + +function errorMessage(err: unknown): string { + return typeof err === "object" && err !== null && "message" in err + ? String((err as { message?: unknown }).message ?? "") + : String(err); +} + +function extractWrittenRows(insertResult: unknown): number | null { + const written = (insertResult as { summary?: { written_rows?: string } } | null | undefined) + ?.summary?.written_rows; + if (written === undefined) return null; + const parsed = Number(written); + return Number.isFinite(parsed) ? parsed : null; +} diff --git a/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts b/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts new file mode 100644 index 00000000000..06786126485 --- /dev/null +++ b/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts @@ -0,0 +1,123 @@ +import { ClickHouse, type TaskEventV2Input } from "@internal/clickhouse"; +import { clickhouseTest } from "@internal/testcontainers"; +import { describe, expect } from "vitest"; +import { z } from "zod"; +import { + ClickhouseEventRepository, + convertDateToClickhouseDateTime, +} from "~/v3/eventRepository/clickhouseEventRepository.server"; + +const TIMEOUT_MS = 60_000; + +function deeplyNested(depth: number): Record { + let node: Record = { leaf: 1 }; + for (let i = 0; i < depth; i++) { + node = { [`k${i}`]: node }; + } + return node; +} + +function startTime(baseMs: number, offsetMs: number): string { + const ns = ((BigInt(baseMs) + BigInt(offsetMs)) * 1_000_000n).toString(); + return `${ns.substring(0, 10)}.${ns.substring(10)}`; +} + +describe("ClickhouseEventRepository JSON parse recovery", () => { + clickhouseTest( + "lands the good events when one event in the batch has ClickHouse-unparseable attributes", + async ({ clickhouseContainer }) => { + const clickhouse = new ClickHouse({ + url: clickhouseContainer.getConnectionUrl(), + logLevel: "error", + }); + + const repository = new ClickhouseEventRepository({ + clickhouse, + version: "v2", + insertStrategy: "insert", + batchSize: 100, + flushInterval: 200, + }); + + const environmentId = "env_json_recovery_test"; + const organizationId = "org_json_recovery_test"; + const projectId = "proj_json_recovery_test"; + const traceId = "b".repeat(32); + const baseMs = Date.now(); + const expiresAt = convertDateToClickhouseDateTime( + new Date(baseMs + 365 * 24 * 60 * 60 * 1000) + ); + + function makeRow(i: number, attributes: unknown): TaskEventV2Input { + return { + environment_id: environmentId, + organization_id: organizationId, + project_id: projectId, + task_identifier: "json-recovery-task", + run_id: `run_${i}`, + start_time: startTime(baseMs, i), + duration: "1000000", + trace_id: traceId, + span_id: `span_recovery_${String(i).padStart(6, "0")}`, + parent_span_id: "", + message: `event ${i}`, + kind: "SPAN", + status: "OK", + attributes, + metadata: "{}", + expires_at: expiresAt, + }; + } + + const goodSpanIds: string[] = []; + let poisonSpanId = ""; + const rows: TaskEventV2Input[] = []; + for (let i = 0; i < 5; i++) { + const isPoison = i === 2; + const row = makeRow(i, isPoison ? deeplyNested(1500) : { ok: true, i }); + rows.push(row); + if (isPoison) { + poisonSpanId = row.span_id; + } else { + goodSpanIds.push(row.span_id); + } + } + + try { + (repository as any).addToBatch(rows); + + const queryEvents = clickhouse.reader.query({ + name: "event-recovery-check", + query: + "SELECT span_id FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}", + schema: z.object({ span_id: z.string() }), + params: z.object({ env_id: z.string() }), + }); + + const landedIds = await vi.waitFor( + async () => { + const [queryError, resultRows] = await queryEvents({ env_id: environmentId }); + expect(queryError).toBeNull(); + const ids = new Set((resultRows ?? []).map((r) => r.span_id)); + for (const id of goodSpanIds) { + expect(ids.has(id)).toBe(true); + } + return ids; + }, + { timeout: 30_000, interval: 250 } + ); + + for (const id of goodSpanIds) { + expect(landedIds.has(id)).toBe(true); + } + expect(landedIds.has(poisonSpanId)).toBe(false); + + expect(repository.permanentlyDroppedBatches).toBe(0); + expect(repository.rowIsolationRecoveries).toBeGreaterThanOrEqual(1); + } finally { + await (repository as any)._flushScheduler?.shutdown?.(); + } + }, + TIMEOUT_MS + ); +}); diff --git a/apps/webapp/test/runsReplicationBenchmark.producer.ts b/apps/webapp/test/runsReplicationBenchmark.producer.ts index 547d1318c46..efa981a3bf6 100644 --- a/apps/webapp/test/runsReplicationBenchmark.producer.ts +++ b/apps/webapp/test/runsReplicationBenchmark.producer.ts @@ -15,6 +15,16 @@ interface ProducerConfig { numRuns: number; errorRate: number; // 0.07 = 7% batchSize: number; + poisonRate?: number; + poisonDepth?: number; +} + +function deeplyNestedOutput(depth: number): Record { + let node: Record = { leaf: 1 }; + for (let i = 0; i < depth; i++) { + node = { [`k${i}`]: node }; + } + return node; } // Error templates for realistic variety @@ -108,6 +118,9 @@ async function runProducer(config: ProducerConfig) { const startTime = performance.now(); let created = 0; let withErrors = 0; + let poisoned = 0; + const poisonRate = config.poisonRate ?? 0; + const poisonDepth = config.poisonDepth ?? 1500; // Process in batches to avoid overwhelming the database for (let batch = 0; batch < Math.ceil(config.numRuns / config.batchSize); batch++) { @@ -120,6 +133,8 @@ async function runProducer(config: ProducerConfig) { const hasError = Math.random() < config.errorRate; const status = hasError ? "COMPLETED_WITH_ERRORS" : "COMPLETED_SUCCESSFULLY"; + const isPoison = Math.random() < poisonRate; + const runData: any = { friendlyId: `run_bench_${Date.now()}_${i}`, taskIdentifier: `benchmark-task-${i % 10}`, // Vary task identifiers @@ -142,6 +157,12 @@ async function runProducer(config: ProducerConfig) { withErrors++; } + if (isPoison) { + runData.output = JSON.stringify(deeplyNestedOutput(poisonDepth)); + runData.outputType = "application/json"; + poisoned++; + } + runs.push(runData); } @@ -168,6 +189,7 @@ async function runProducer(config: ProducerConfig) { console.log(`[Producer] Completed:`); console.log(` - Total runs: ${created}`); console.log(` - With errors: ${withErrors} (${((withErrors / created) * 100).toFixed(1)}%)`); + console.log(` - Poisoned: ${poisoned} (${((poisoned / created) * 100).toFixed(1)}%)`); console.log(` - Duration: ${duration.toFixed(0)}ms`); console.log(` - Throughput: ${throughput.toFixed(0)} runs/sec`); @@ -178,6 +200,7 @@ async function runProducer(config: ProducerConfig) { stats: { created, withErrors, + poisoned, duration, throughput, }, diff --git a/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts b/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts new file mode 100644 index 00000000000..8186e145ec6 --- /dev/null +++ b/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts @@ -0,0 +1,267 @@ +import { ClickHouse } from "@internal/clickhouse"; +import { replicationContainerTest } from "@internal/testcontainers"; +import { fork } from "node:child_process"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; +import { setTimeout } from "node:timers/promises"; +import { z } from "zod"; +import { RunsReplicationService } from "~/services/runsReplicationService.server"; +import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory"; +import { createInMemoryMetrics, createInMemoryTracing } from "./utils/tracing"; + +vi.setConfig({ testTimeout: 300_000 }); + +const CONFIG = { + NUM_RUNS: parseInt(process.env.BENCHMARK_NUM_RUNS || "5000", 10), + PRODUCER_BATCH_SIZE: 100, + FLUSH_BATCH_SIZE: 50, + FLUSH_INTERVAL_MS: 100, + MAX_FLUSH_CONCURRENCY: 4, + REPLICATION_TIMEOUT_MS: 180_000, + POISON_RATE: parseFloat(process.env.BENCHMARK_POISON_RATE || "0.05"), +}; + +class ELUMonitor { + private samples: number[] = []; + private interval: NodeJS.Timeout | null = null; + + start(intervalMs = 100) { + this.samples = []; + performance.eventLoopUtilization(); + this.interval = setInterval(() => { + this.samples.push(performance.eventLoopUtilization().utilization * 100); + }, intervalMs); + } + + stop() { + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + } + if (this.samples.length === 0) return { mean: 0, p50: 0, p95: 0, p99: 0, samples: 0 }; + const sorted = [...this.samples].sort((a, b) => a - b); + const mean = sorted.reduce((s, v) => s + v, 0) / sorted.length; + return { + mean, + p50: sorted[Math.floor(sorted.length * 0.5)], + p95: sorted[Math.floor(sorted.length * 0.95)], + p99: sorted[Math.floor(sorted.length * 0.99)], + samples: sorted.length, + }; + } +} + +function runProducer(config: { + postgresUrl: string; + organizationId: string; + projectId: string; + environmentId: string; + numRuns: number; + errorRate: number; + batchSize: number; + poisonRate: number; + poisonDepth: number; +}): Promise<{ created: number; withErrors: number; poisoned: number; duration: number }> { + return new Promise((resolve, reject) => { + const producerPath = path.join(__dirname, "runsReplicationBenchmark.producer.ts"); + const child = fork(producerPath, [JSON.stringify(config)], { + stdio: ["ignore", "pipe", "pipe", "ipc"], + execArgv: ["-r", "tsx/cjs"], + }); + child.stdout?.on("data", (d) => console.log(d.toString().trim())); + child.stderr?.on("data", (d) => console.error(d.toString().trim())); + child.on("message", (m: any) => { + if (m.type === "complete") resolve(m.stats); + else if (m.type === "error") reject(new Error(m.error)); + }); + child.on("error", reject); + child.on("exit", (code) => { + if (code !== 0) reject(new Error(`Producer exited with code ${code}`)); + }); + }); +} + +async function waitForCount( + clickhouse: ClickHouse, + organizationId: string, + expectedCount: number, + timeoutMs: number +): Promise<{ duration: number; count: number }> { + const startTime = performance.now(); + const deadline = startTime + timeoutMs; + const queryRuns = clickhouse.reader.query({ + name: "bench-count", + query: + "SELECT count(*) as count FROM trigger_dev.task_runs_v2 WHERE organization_id = {org_id:String}", + schema: z.object({ count: z.number() }), + params: z.object({ org_id: z.string() }), + }); + while (performance.now() < deadline) { + const [error, result] = await queryRuns({ org_id: organizationId }); + if (error) throw new Error(`Failed to query ClickHouse: ${error.message}`); + const count = result?.[0]?.count || 0; + if (count >= expectedCount) return { duration: performance.now() - startTime, count }; + await setTimeout(500); + } + const [, result] = await queryRuns({ org_id: organizationId }); + return { duration: performance.now() - startTime, count: result?.[0]?.count || 0 }; +} + +async function runScenario( + name: string, + poisonRate: number, + ctx: { clickhouseContainer: any; redisOptions: any; postgresContainer: any; prisma: any } +) { + const { clickhouseContainer, redisOptions, postgresContainer, prisma } = ctx; + + const organization = await prisma.organization.create({ + data: { title: `bench-${name}`, slug: `bench-${name}` }, + }); + const project = await prisma.project.create({ + data: { + name: `bench-${name}`, + slug: `bench-${name}`, + organizationId: organization.id, + externalRef: `bench-${name}`, + }, + }); + const runtimeEnvironment = await prisma.runtimeEnvironment.create({ + data: { + slug: `bench-${name}`, + type: "DEVELOPMENT", + projectId: project.id, + organizationId: organization.id, + apiKey: `bench-${name}`, + pkApiKey: `bench-${name}`, + shortcode: `bench-${name}`, + }, + }); + + const clickhouse = new ClickHouse({ + url: clickhouseContainer.getConnectionUrl(), + name: `bench-${name}`, + compression: { request: true }, + logLevel: "error", + }); + + const { tracer } = createInMemoryTracing(); + const metricsHelper = createInMemoryMetrics(); + + const service = new RunsReplicationService({ + clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse), + pgConnectionUrl: postgresContainer.getConnectionUri(), + serviceName: `bench-${name}`, + slotName: `bench_${name.replace(/-/g, "_")}`, + publicationName: `bench_${name.replace(/-/g, "_")}_pub`, + redisOptions, + maxFlushConcurrency: CONFIG.MAX_FLUSH_CONCURRENCY, + flushIntervalMs: CONFIG.FLUSH_INTERVAL_MS, + flushBatchSize: CONFIG.FLUSH_BATCH_SIZE, + leaderLockTimeoutMs: 10000, + leaderLockExtendIntervalMs: 2000, + ackIntervalSeconds: 10, + tracer, + meter: metricsHelper.meter, + logLevel: "error", + }); + + await service.start(); + + const elu = new ELUMonitor(); + elu.start(100); + + let producerStats!: { created: number; withErrors: number; poisoned: number; duration: number }; + let replication!: { duration: number; count: number }; + let eluStats!: ReturnType; + + try { + producerStats = await runProducer({ + postgresUrl: postgresContainer.getConnectionUri(), + organizationId: organization.id, + projectId: project.id, + environmentId: runtimeEnvironment.id, + numRuns: CONFIG.NUM_RUNS, + errorRate: 0.07, + batchSize: CONFIG.PRODUCER_BATCH_SIZE, + poisonRate, + poisonDepth: 1500, + }); + + const expectedLanded = producerStats.created - producerStats.poisoned; + replication = await waitForCount( + clickhouse, + organization.id, + expectedLanded, + CONFIG.REPLICATION_TIMEOUT_MS + ); + } finally { + eluStats = elu.stop(); + await service.stop(); + await metricsHelper.shutdown(); + } + + const throughput = (replication.count / replication.duration) * 1000; + const expectedLanded = producerStats.created - producerStats.poisoned; + + console.log(`\n${"=".repeat(72)}`); + console.log(`SCENARIO: ${name} (poison rate ${(poisonRate * 100).toFixed(1)}%)`); + console.log(`${"=".repeat(72)}`); + console.log(`Produced: ${producerStats.created} runs (${producerStats.poisoned} poisoned)`); + console.log(`Landed in CH: ${replication.count} / expected ${expectedLanded}`); + console.log(`Repl dur: ${replication.duration.toFixed(0)}ms`); + console.log(`Throughput: ${throughput.toFixed(0)} runs/sec`); + console.log(`Row-isolation recoveries: ${service.rowIsolationRecoveries}`); + console.log(`Dropped rows (best-effort): ${service.permanentlyDroppedRows}`); + console.log(`Dropped batches (whole): ${service.permanentlyDroppedBatches}`); + console.log(`ELU mean=${eluStats.mean.toFixed(2)}% p50=${eluStats.p50.toFixed(2)}% p95=${eluStats.p95.toFixed(2)}% p99=${eluStats.p99.toFixed(2)}% (${eluStats.samples} samples)`); + console.log(`${"=".repeat(72)}\n`); + + return { name, poisonRate, producerStats, replication, eluStats, throughput, expectedLanded, service }; +} + +describe("RunsReplicationService JSON-recovery ELU benchmark", () => { + replicationContainerTest.skipIf(process.env.BENCHMARKS_ENABLED !== "1")( + "measures ELU on the healthy hot path vs the row-isolation recovery path", + async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => { + await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`); + + const healthy = await runScenario("healthy-0pct", 0, { + clickhouseContainer, + redisOptions, + postgresContainer, + prisma, + }); + + const poisoned = await runScenario("poison", CONFIG.POISON_RATE, { + clickhouseContainer, + redisOptions, + postgresContainer, + prisma, + }); + + const eluMeanDelta = + ((poisoned.eluStats.mean - healthy.eluStats.mean) / healthy.eluStats.mean) * 100; + const eluP99Delta = + ((poisoned.eluStats.p99 - healthy.eluStats.p99) / healthy.eluStats.p99) * 100; + + console.log(`\n${"=".repeat(72)}`); + console.log("COMPARISON (healthy 0% -> poison)"); + console.log(`${"=".repeat(72)}`); + console.log( + `ELU mean: ${healthy.eluStats.mean.toFixed(2)}% -> ${poisoned.eluStats.mean.toFixed(2)}% (${eluMeanDelta > 0 ? "+" : ""}${eluMeanDelta.toFixed(1)}%)` + ); + console.log( + `ELU p99: ${healthy.eluStats.p99.toFixed(2)}% -> ${poisoned.eluStats.p99.toFixed(2)}% (${eluP99Delta > 0 ? "+" : ""}${eluP99Delta.toFixed(1)}%)` + ); + console.log(`${"=".repeat(72)}\n`); + + expect(healthy.replication.count).toBe(healthy.expectedLanded); + expect(healthy.service.rowIsolationRecoveries).toBe(0); + expect(healthy.service.permanentlyDroppedBatches).toBe(0); + + expect(poisoned.replication.count).toBe(poisoned.expectedLanded); + expect(poisoned.service.permanentlyDroppedBatches).toBe(0); + expect(poisoned.service.rowIsolationRecoveries).toBeGreaterThanOrEqual(1); + } + ); +}); diff --git a/apps/webapp/test/runsReplicationService.part10.test.ts b/apps/webapp/test/runsReplicationService.part10.test.ts new file mode 100644 index 00000000000..3f269430c42 --- /dev/null +++ b/apps/webapp/test/runsReplicationService.part10.test.ts @@ -0,0 +1,143 @@ +import { ClickHouse } from "@internal/clickhouse"; +import { replicationContainerTest } from "@internal/testcontainers"; +import { z } from "zod"; +import { RunsReplicationService } from "~/services/runsReplicationService.server"; +import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory"; +import { createInMemoryTracing } from "./utils/tracing"; + +vi.setConfig({ testTimeout: 60_000 }); + +function deeplyNested(depth: number): Record { + let node: Record = { leaf: 1 }; + for (let i = 0; i < depth; i++) { + node = { [`k${i}`]: node }; + } + return node; +} + +describe("RunsReplicationService (part 10/10) — JSON parse recovery", () => { + replicationContainerTest( + "lands the good runs when one run in the batch has ClickHouse-unparseable JSON output", + async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => { + await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`); + + const clickhouse = new ClickHouse({ + url: clickhouseContainer.getConnectionUrl(), + name: "runs-replication", + compression: { request: true }, + logLevel: "warn", + }); + + const { tracer } = createInMemoryTracing(); + + const runsReplicationService = new RunsReplicationService({ + clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse), + pgConnectionUrl: postgresContainer.getConnectionUri(), + serviceName: "runs-replication", + slotName: "task_runs_to_clickhouse_v1", + publicationName: "task_runs_to_clickhouse_v1_publication", + redisOptions, + maxFlushConcurrency: 1, + flushIntervalMs: 500, + flushBatchSize: 50, + leaderLockTimeoutMs: 5000, + leaderLockExtendIntervalMs: 1000, + ackIntervalSeconds: 5, + tracer, + logLevel: "warn", + }); + + await runsReplicationService.start(); + + const organization = await prisma.organization.create({ + data: { title: "test", slug: "test" }, + }); + const project = await prisma.project.create({ + data: { + name: "test", + slug: "test", + organizationId: organization.id, + externalRef: "test", + }, + }); + const runtimeEnvironment = await prisma.runtimeEnvironment.create({ + data: { + slug: "test", + type: "DEVELOPMENT", + projectId: project.id, + organizationId: organization.id, + apiKey: "test", + pkApiKey: "test", + shortcode: "test", + }, + }); + + const goodRunIds: string[] = []; + let poisonRunId = ""; + + for (let i = 0; i < 5; i++) { + const isPoison = i === 2; + const output = isPoison + ? JSON.stringify(deeplyNested(1500)) + : JSON.stringify({ ok: true, i }); + + const run = await prisma.taskRun.create({ + data: { + friendlyId: `run_batchdrop_${i}`, + taskIdentifier: "my-task", + payload: JSON.stringify({ i }), + payloadType: "application/json", + output, + outputType: "application/json", + traceId: `trace_${i}`, + spanId: `span_${i}`, + queue: "test", + status: "COMPLETED_SUCCESSFULLY", + runtimeEnvironmentId: runtimeEnvironment.id, + projectId: project.id, + organizationId: organization.id, + environmentType: "DEVELOPMENT", + engine: "V2", + }, + }); + + if (isPoison) { + poisonRunId = run.id; + } else { + goodRunIds.push(run.id); + } + } + + const queryRuns = clickhouse.reader.query({ + name: "runs-replication-batchdrop", + query: + "SELECT run_id FROM trigger_dev.task_runs_v2 WHERE organization_id = {org_id:String}", + schema: z.object({ run_id: z.string() }), + params: z.object({ org_id: z.string() }), + }); + + const landedIds = await vi.waitFor( + async () => { + const [queryError, rows] = await queryRuns({ org_id: organization.id }); + expect(queryError).toBeNull(); + const ids = new Set((rows ?? []).map((r) => r.run_id)); + for (const id of goodRunIds) { + expect(ids.has(id)).toBe(true); + } + return ids; + }, + { timeout: 30_000, interval: 250 } + ); + + for (const id of goodRunIds) { + expect(landedIds.has(id)).toBe(true); + } + expect(landedIds.has(poisonRunId)).toBe(false); + + expect(runsReplicationService.permanentlyDroppedBatches).toBe(0); + expect(runsReplicationService.rowIsolationRecoveries).toBeGreaterThanOrEqual(1); + + await runsReplicationService.stop(); + } + ); +}); From 6848c4462a2c93c9c99c76c821d77b7280eee932 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 24 Jul 2026 09:29:53 +0100 Subject: [PATCH 2/2] chore(webapp): rename batch row-isolation metric and emit run-replication drop/isolation counters Rename the event repository counter from ingest.flush.rows_isolated to ingest.flush.batches_row_isolated (it counts recovered batches, not rows, and the OTLP->Prometheus suffix would read as a row count otherwise). Add matching runs_replication.batches_row_isolated and runs_replication.batches_dropped OTEL counters so run-replication drops and isolations are alertable in production, not just readable in tests. --- .../services/runsReplicationService.server.ts | 26 ++++++++++++- .../clickhouseEventRepository.server.ts | 37 ++++++++++++------- ...nsReplicationJsonRecoveryBenchmark.test.ts | 15 +++++++- 3 files changed, 61 insertions(+), 17 deletions(-) diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts index 5ffb38bd5f8..9eb3b895d07 100644 --- a/apps/webapp/app/services/runsReplicationService.server.ts +++ b/apps/webapp/app/services/runsReplicationService.server.ts @@ -209,6 +209,8 @@ export class RunsReplicationService { private _payloadsInsertedCounter: Counter; private _insertRetriesCounter: Counter; private _eventsProcessedCounter: Counter; + private _rowIsolatedBatchesCounter: Counter; + private _droppedBatchesCounter: Counter; private _flushDurationHistogram: Histogram; public readonly events: EventEmitter; @@ -262,6 +264,20 @@ export class RunsReplicationService { description: "Replication events processed (inserts, updates, deletes)", }); + this._rowIsolatedBatchesCounter = this._meter.createCounter( + "runs_replication.batches_row_isolated", + { + description: + "Batches recovered by isolating un-ingestable rows (landed the rest) after a ClickHouse JSON parse error", + unit: "batches", + } + ); + + this._droppedBatchesCounter = this._meter.createCounter("runs_replication.batches_dropped", { + description: "Batches permanently dropped after an unrecoverable ClickHouse JSON parse error", + unit: "batches", + }); + this._flushDurationHistogram = this._meter.createHistogram( "runs_replication.flush_duration_ms", { @@ -1187,7 +1203,13 @@ export class RunsReplicationService { } } - return await this.#insertIsolatingBadRows(rows, doInsert, contextLabel, attempt, firstMessage); + return await this.#insertIsolatingBadRows( + rows, + doInsert, + contextLabel, + attempt, + firstMessage + ); } } @@ -1214,6 +1236,7 @@ export class RunsReplicationService { }); this._rowIsolationRecoveries += 1; + this._rowIsolatedBatchesCounter.add(1, { table: contextLabel }); const writtenRows = extractWrittenRows(insertResult); const rowsSkipped = writtenRows === null ? null : Math.max(0, rows.length - writtenRows); if (rowsSkipped !== null) { @@ -1239,6 +1262,7 @@ export class RunsReplicationService { if (!isClickHouseJsonParseError(isolationError)) throw isolationError; this._permanentlyDroppedBatches += 1; + this._droppedBatchesCounter.add(1, { table: contextLabel }); this.logger.error( "Dropped batch — row-isolation insert failed after ClickHouse JSON parse error", { diff --git a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts index 417b9bffa7f..8bee56dba82 100644 --- a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts @@ -137,7 +137,7 @@ export class ClickhouseEventRepository implements IEventRepository { * un-ingestable rows and landed the rest. Reliable per-event signal. */ private _rowIsolationRecoveries = 0; - private readonly _rowsIsolatedCounter: Counter; + private readonly _rowIsolatedBatchesCounter: Counter; /** * Best-effort count of individual rows ClickHouse skipped across those @@ -158,9 +158,9 @@ export class ClickhouseEventRepository implements IEventRepository { description: "Batches permanently dropped after an unrecoverable ClickHouse JSON parse error", unit: "batches", }); - this._rowsIsolatedCounter = meter.createCounter("ingest.flush.rows_isolated", { + this._rowIsolatedBatchesCounter = meter.createCounter("ingest.flush.batches_row_isolated", { description: - "Batches that recovered by isolating un-ingestable rows (landed the rest) after a ClickHouse JSON parse error", + "Batches recovered by isolating un-ingestable rows (landed the rest) after a ClickHouse JSON parse error", unit: "batches", }); @@ -415,7 +415,13 @@ export class ClickhouseEventRepository implements IEventRepository { } } - return await this.#insertIsolatingBadRows(flushId, rows, doInsert, contextLabel, firstMessage); + return await this.#insertIsolatingBadRows( + flushId, + rows, + doInsert, + contextLabel, + firstMessage + ); } } @@ -442,7 +448,7 @@ export class ClickhouseEventRepository implements IEventRepository { }); this._rowIsolationRecoveries += 1; - this._rowsIsolatedCounter.add(1, { table: contextLabel }); + this._rowIsolatedBatchesCounter.add(1, { table: contextLabel }); const writtenRows = extractWrittenRows(insertResult); const rowsSkipped = writtenRows === null ? null : Math.max(0, rows.length - writtenRows); if (rowsSkipped !== null) { @@ -469,15 +475,18 @@ export class ClickhouseEventRepository implements IEventRepository { this._permanentlyDroppedBatches += 1; this._droppedBatchesCounter.add(1, { table: contextLabel }); - logger.error("Dropped batch — row-isolation insert failed after ClickHouse JSON parse error", { - flushId, - contextLabel, - batchSize: rows.length, - permanentlyDroppedBatches: this._permanentlyDroppedBatches, - sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), - firstError: firstMessage.split("\n")[0], - isolationError: errorMessage(isolationError).split("\n")[0], - }); + logger.error( + "Dropped batch — row-isolation insert failed after ClickHouse JSON parse error", + { + flushId, + contextLabel, + batchSize: rows.length, + permanentlyDroppedBatches: this._permanentlyDroppedBatches, + sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024), + firstError: firstMessage.split("\n")[0], + isolationError: errorMessage(isolationError).split("\n")[0], + } + ); return { kind: "dropped" }; } } diff --git a/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts b/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts index 8186e145ec6..254237a6022 100644 --- a/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts +++ b/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts @@ -213,10 +213,21 @@ async function runScenario( console.log(`Row-isolation recoveries: ${service.rowIsolationRecoveries}`); console.log(`Dropped rows (best-effort): ${service.permanentlyDroppedRows}`); console.log(`Dropped batches (whole): ${service.permanentlyDroppedBatches}`); - console.log(`ELU mean=${eluStats.mean.toFixed(2)}% p50=${eluStats.p50.toFixed(2)}% p95=${eluStats.p95.toFixed(2)}% p99=${eluStats.p99.toFixed(2)}% (${eluStats.samples} samples)`); + console.log( + `ELU mean=${eluStats.mean.toFixed(2)}% p50=${eluStats.p50.toFixed(2)}% p95=${eluStats.p95.toFixed(2)}% p99=${eluStats.p99.toFixed(2)}% (${eluStats.samples} samples)` + ); console.log(`${"=".repeat(72)}\n`); - return { name, poisonRate, producerStats, replication, eluStats, throughput, expectedLanded, service }; + return { + name, + poisonRate, + producerStats, + replication, + eluStats, + throughput, + expectedLanded, + service, + }; } describe("RunsReplicationService JSON-recovery ELU benchmark", () => {