From 50324d8f3a2a864e6fd4eb52f751f641410b0814 Mon Sep 17 00:00:00 2001 From: Arturs Sosins Date: Wed, 14 Oct 2026 16:00:23 +0300 Subject: [PATCH] ClickHouse 26.8: pin synchronous inserts, 26.8-safe fixtures, CI on 26.8 Countly's ClickHouse runs its default profile with async_insert=1, so the staging inserts silently became async inserts. The verify-then-attach count right after an insert relies on an ack meaning the rows are written, which then held only while wait_for_async_insert stays on (an env knob there). The client now sets async_insert=0 itself. ts/cd/lu are already written as quoted datetime strings, so the 26.8 change that reads a bare JSON number for DateTime64 as seconds does not reach the migrator's writes, with or without input_format_read_datetime_number_as_raw_value. Test fixtures did write bare millisecond numbers into ts in VALUES, which stores 9999-12-31 on a 26.8 server without that setting; they now use fromUnixTimestamp64Milli. The ledger e2e test now creates the platform's drill_events DDL (codecs, cd index, object serialization settings) and asserts exact stored ts/cd/lu and partitions, no 9999 rows, and that every insert ran with async_insert=0. CI moves from ClickHouse 26.4 to the stock 26.8 image (no Countly profile, the strictest case); the opt-in outage test defaults to 26.8 too. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 11 ++-- src/target/staging-manager.ts | 11 ++++ tests/integration/ledger-engine.test.ts | 59 ++++++++++++++++++- .../multi-collection-and-rebuild.test.ts | 10 ++-- tests/integration/outage-chaos.test.ts | 2 +- 5 files changed, 80 insertions(+), 13 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 10d26b4..c783b15 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,7 +36,7 @@ jobs: run: npx vitest run tests/differential integration: - name: Integration (MongoDB 7 + ClickHouse 26.4) + name: Integration (MongoDB 7 + ClickHouse 26.8) runs-on: ubuntu-latest services: mongo: @@ -46,10 +46,13 @@ jobs: --health-cmd "mongosh --quiet --eval 'db.runCommand({ping:1})'" --health-interval 5s --health-timeout 5s --health-retries 30 clickhouse: - # Production ClickHouse version (deploy/compose/images.standard.env - # in countly-platform). 26.4 requires a password for non-localhost + # Production ClickHouse minor (deploy/compose/components/clickhouse.yml + # in countly-platform; its floor is 26.8.6). The stock image has no + # Countly profile, so input_format_read_datetime_number_as_raw_value + # stays 0 here: the strictest case for the 26.8 DateTime64 change, as + # on an external ClickHouse. It requires a password for non-localhost # clients — the tests read TEST_CLICKHOUSE_PASSWORD. - image: clickhouse/clickhouse-server:26.4 + image: clickhouse/clickhouse-server:26.8 env: CLICKHOUSE_PASSWORD: ci-pass ports: ['8123:8123'] diff --git a/src/target/staging-manager.ts b/src/target/staging-manager.ts index b946ac0..b1e7ce3 100644 --- a/src/target/staging-manager.ts +++ b/src/target/staging-manager.ts @@ -45,6 +45,17 @@ export class StagingManager { compression: { request: true }, // Synchronous inserts: an acked insert is parsed, validated, and visible. clickhouse_settings: { + // Pinned, not inherited: Countly's ClickHouse runs its `default` + // profile with async_insert=1 (and wait_for_async_insert is an env + // knob there). With wait off an ack would precede the flush, and the + // verify-then-attach count right after it would race the buffer. + async_insert: 0, + // ts/lu/cd are sent as quoted 'yyyy-MM-dd HH:mm:ss.SSS' strings, never + // bare numbers: from ClickHouse 26.8 a bare number in JSON input for a + // DateTime64 column is read as seconds unless the session or profile + // sets input_format_read_datetime_number_as_raw_value=1 (a setting + // older servers reject), and a millisecond epoch lands on 9999-12-31. + // Strings parse the same on every version with or without it. date_time_input_format: 'best_effort', optimize_on_insert: 0, // Chunks are cut on cd (server insert time) but tables partition by diff --git a/tests/integration/ledger-engine.test.ts b/tests/integration/ledger-engine.test.ts index e9d1ca3..6862538 100644 --- a/tests/integration/ledger-engine.test.ts +++ b/tests/integration/ledger-engine.test.ts @@ -304,6 +304,8 @@ describe('ledger engine end-to-end', () => { ch = createClient({ url: CH_URL, password: CH_PASSWORD }); await ch.command({ query: `CREATE DATABASE IF NOT EXISTS ${DB}` }); await ch.command({ query: `DROP TABLE IF EXISTS ${DB}.drill_events` }); + // countly-platform plugins/clickhouse/api/sql/01-drill_events.sql as + // shipped with ClickHouse 26.8 (codecs, cd index, object serialization). await ch.command({ query: `CREATE TABLE ${DB}.drill_events ( \`a\` LowCardinality(String), \`e\` LowCardinality(String), \`n\` String, @@ -311,8 +313,16 @@ describe('ledger engine end-to-end', () => { \`_id\` String, \`ts\` DateTime64(3), \`up\` JSON(max_dynamic_paths = 32), \`custom\` Nullable(JSON(max_dynamic_paths = 0)), \`cmp\` Nullable(JSON(max_dynamic_paths = 0)), \`sg\` JSON(max_dynamic_paths = 0), \`c\` UInt32, \`s\` Float64, \`dur\` Float64, - \`lu\` Nullable(DateTime64(3)), \`cd\` DateTime64(3) DEFAULT now64(3)) - ENGINE = MergeTree PARTITION BY toYYYYMM(ts, 'UTC') ORDER BY (a, e, n, ts)`, + \`lu\` Nullable(DateTime64(3)) CODEC (Delta, LZ4), + \`cd\` DateTime64(3) DEFAULT now64(3) CODEC (Delta, LZ4), + INDEX cd_minmax (cd) TYPE minmax GRANULARITY 1) + ENGINE = MergeTree PARTITION BY toYYYYMM(ts, 'UTC') ORDER BY (a, e, n, ts) + SETTINGS index_granularity = 8192, + object_serialization_version = 'v3', + object_shared_data_serialization_version = 'advanced', + object_shared_data_serialization_version_for_zero_level_parts = 'map_with_buckets', + object_shared_data_buckets_for_compact_part = 8, + object_shared_data_buckets_for_wide_part = 32`, }); // Seed: clean docs + 3 transform-poisoned docs (bad ts) + 1 coercion doc @@ -332,7 +342,8 @@ describe('ledger engine end-to-end', () => { } docs.push({ _id: 'coerce_me', a: 'app1', e: 'big_int_event', uid: 'u9', did: 'd9', - ts: base + 1, cd: new Date(base + 1), sg: { order_id: 9.2e25, weird: Number.POSITIVE_INFINITY, user_ref: 5.2601586211929e19 }, c: 1, + ts: base + 1, cd: new Date(base + 1), lu: new Date(base + 2), + sg: { order_id: 9.2e25, weird: Number.POSITIVE_INFINITY, user_ref: 5.2601586211929e19 }, c: 1, }); // Docs with no cd value — must be picked up by the null-cd sweep chunk docs.push({ _id: 'nocd_1', a: 'app1', e: 'legacy_event', uid: 'u1', did: 'd', ts: base - 86_400_000 }); @@ -393,6 +404,7 @@ describe('ledger engine end-to-end', () => { }); it('migrates exactly: clean docs land once, poisoned docs go to DLQ with raw docs, coercions counted', async () => { + const runStartSec = Math.floor(Date.now() / 1000) - 1; await orchestrator.run(); const res = await ch.query({ @@ -404,6 +416,47 @@ describe('ledger engine end-to-end', () => { expect(Number(row.t)).toBe(CLEAN_DOCS + 4); expect(Number(row.u)).toBe(CLEAN_DOCS + 4); // zero duplicates + // Millisecond timestamps keep their value. From ClickHouse 26.8 a writer + // sending them as bare JSON numbers without + // input_format_read_datetime_number_as_raw_value=1 lands on 9999-12-31. + const dates = await ch.query({ + query: `SELECT _id, toUnixTimestamp64Milli(ts) AS ts, toUnixTimestamp64Milli(cd) AS cd, + toUnixTimestamp64Milli(lu) AS lu, _partition_id AS p + FROM ${DB}.drill_events WHERE _id IN ('doc_7', 'coerce_me', 'nocd_1') ORDER BY _id`, + format: 'JSONEachRow', + }); + const base = Date.UTC(2026, 0, 1); + // Int64 comes back unquoted from 26.x (output_format_json_quote_64bit_integers=0), quoted before. + const num = (v: unknown): number | null => (v === null ? null : Number(v)); + const got = (await dates.json>()) + .map((r) => ({ _id: r._id, ts: num(r.ts), cd: num(r.cd), lu: num(r.lu), p: r.p })); + expect(got).toEqual([ + { _id: 'coerce_me', ts: base + 1, cd: base + 1, lu: base + 2, p: '202601' }, + { _id: 'doc_7', ts: base + 7 * 60_000, cd: base + 7 * 60_000, lu: null, p: '202601' }, + { _id: 'nocd_1', ts: base - 86_400_000, cd: base - 86_400_000, lu: null, p: '202512' }, + ]); + const poisoned = await ch.query({ + query: `SELECT count() AS c FROM ${DB}.drill_events + WHERE toYear(ts) >= 9999 OR toYear(cd) >= 9999 OR toYear(assumeNotNull(lu)) >= 9999`, + format: 'JSONEachRow', + }); + expect(Number((await poisoned.json<{ c: string }>())[0].c)).toBe(0); + + // Every insert ran synchronously, whatever the server profile says + // (Countly's sets async_insert=1). + await ch.command({ query: 'SYSTEM FLUSH LOGS' }); + const inserts = await ch.query({ + query: `SELECT Settings['async_insert'] AS ai, count() AS c FROM system.query_log + WHERE type = 'QueryFinish' AND query_kind = 'Insert' AND has(databases, {db:String}) + AND event_time >= toDateTime({since:UInt32}) + GROUP BY ai`, + query_params: { db: DB, since: runStartSec }, + format: 'JSONEachRow', + }); + const byMode = await inserts.json<{ ai: string; c: string }>(); + expect(byMode.map((r) => r.ai)).toEqual(['0']); + expect(Number(byMode[0].c)).toBeGreaterThan(0); + // Null-cd docs arrived via the sweep chunk const nocd = await ch.query({ query: `SELECT count() AS c FROM ${DB}.drill_events WHERE _id LIKE 'nocd_%'`, diff --git a/tests/integration/multi-collection-and-rebuild.test.ts b/tests/integration/multi-collection-and-rebuild.test.ts index 1211320..8fd2665 100644 --- a/tests/integration/multi-collection-and-rebuild.test.ts +++ b/tests/integration/multi-collection-and-rebuild.test.ts @@ -209,8 +209,8 @@ describe('multi-collection scoping + ledger rebuild', () => { await mc.db(DB).collection('mig_ranges').deleteMany({}); await ch.command({ query: `INSERT INTO ${DB}.drill_events (a, e, n, uid, did, _id, ts, cd) - VALUES ('${APP}', '[CLY]_custom', '${EV1}', 'u_live', 'd_live', 'live_1', ${BASE + 40 * 86_400_000}, fromUnixTimestamp64Milli(${BASE + 40 * 86_400_000})) - , ('${APP}', '[CLY]_custom', '${EV2}', 'u_live', 'd_live', 'live_2', ${BASE + 41 * 86_400_000}, fromUnixTimestamp64Milli(${BASE + 41 * 86_400_000}))`, + VALUES ('${APP}', '[CLY]_custom', '${EV1}', 'u_live', 'd_live', 'live_1', fromUnixTimestamp64Milli(${BASE + 40 * 86_400_000}), fromUnixTimestamp64Milli(${BASE + 40 * 86_400_000})) + , ('${APP}', '[CLY]_custom', '${EV2}', 'u_live', 'd_live', 'live_2', fromUnixTimestamp64Milli(${BASE + 41 * 86_400_000}), fromUnixTimestamp64Milli(${BASE + 41 * 86_400_000}))`, }); const progress = newRebuildProgress(); @@ -278,7 +278,7 @@ describe('multi-collection scoping + ledger rebuild', () => { // same event) — that shared ts month is what makes per-partition // duplicate scanning exact. const mk = (id: string, tsMs: number, cdMs: number) => - `('${APP}', '[CLY]_custom', '${EV1}', 'u_dup', 'd_dup', '${id}', ${tsMs}, fromUnixTimestamp64Milli(${cdMs}))`; + `('${APP}', '[CLY]_custom', '${EV1}', 'u_dup', 'd_dup', '${id}', fromUnixTimestamp64Milli(${tsMs}), fromUnixTimestamp64Milli(${cdMs}))`; // 1) live at-least-once redelivery: same _id twice, NO migrated copy await ch.command({ query: `INSERT INTO ${DB}.drill_events (a, e, n, uid, did, _id, ts, cd) @@ -305,7 +305,7 @@ describe('multi-collection scoping + ledger rebuild', () => { // (both written by migration — same doc migrated twice) await ch.command({ query: `INSERT INTO ${DB}.drill_events (a, e, n, uid, did, _id, ts, cd) - VALUES ('${APP}', '[CLY]_custom', '${EV1}', 'u_dup', 'd_dup', 'p_6', ${BASE + 6 * 60_000}, fromUnixTimestamp64Milli(${BASE + 6 * 60_000 + 1}))`, + VALUES ('${APP}', '[CLY]_custom', '${EV1}', 'u_dup', 'd_dup', 'p_6', fromUnixTimestamp64Milli(${BASE + 6 * 60_000}), fromUnixTimestamp64Milli(${BASE + 6 * 60_000 + 1}))`, }); verify = await orchestrator.verifyMigration(); expect(verify.migrationDuplicates).toBe(1); @@ -553,7 +553,7 @@ describe('multi-collection scoping + ledger rebuild', () => { // live retry copy: same _id, same ts (same partition), NOT migrated await ch.command({ query: `INSERT INTO ${DB}.drill_events (a, e, n, uid, did, _id, ts, cd) - VALUES ('${APP}', '[CLY]_custom', '${EV1}', 'u', 'd', 'retry_victim', ${tsMs}, fromUnixTimestamp64Milli(${Date.now()}))`, + VALUES ('${APP}', '[CLY]_custom', '${EV1}', 'u', 'd', 'retry_victim', fromUnixTimestamp64Milli(${tsMs}), fromUnixTimestamp64Milli(${Date.now()}))`, }); expect(await staging.countLiveMatchingStaged(stagingTable, partitionId)).toBe(0); // id-only matching: 1 → skipped attach → data loss diff --git a/tests/integration/outage-chaos.test.ts b/tests/integration/outage-chaos.test.ts index f3fa414..2d49038 100644 --- a/tests/integration/outage-chaos.test.ts +++ b/tests/integration/outage-chaos.test.ts @@ -124,7 +124,7 @@ describe.skipIf(!ENABLED)('backing-service outage chaos (CHAOS_OUTAGE=1, dedicat // dedicated throwaway services — NEVER the shared dev-stack containers sh(`docker rm -f ${MONGO_CTR} ${CH_CTR} 2>/dev/null || true`); const mongoImage = process.env.CHAOS_MONGO_IMAGE ?? 'mirror.gcr.io/library/mongo:8'; - const chImage = process.env.CHAOS_CH_IMAGE ?? 'mirror.gcr.io/clickhouse/clickhouse-server:26.4'; + const chImage = process.env.CHAOS_CH_IMAGE ?? 'mirror.gcr.io/clickhouse/clickhouse-server:26.8'; sh(`docker run -d --name ${MONGO_CTR} -p 27117:27017 ${mongoImage}`); sh(`docker run -d --name ${CH_CTR} -p 8223:8123 -e CLICKHOUSE_PASSWORD=${CH_PASSWORD} ${chImage}`);