ClickHouse 26.8: pin synchronous inserts, 26.8-safe fixtures, CI on 26.8

.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']

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

tests/integration/ledger-engine.test.ts

@@ -304,15 +304,25 @@ 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,
         \`uid\` String, \`uid_canon\` Nullable(String), \`did\` String, \`lsid\` Nullable(String),
         \`_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<Record<string, unknown>>())
+      .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_%'`,

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
 

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}`);