From 282be47a726ad9acf4971902215b398240d758d4 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 28 Sep 2026 15:19:28 +0000 Subject: [PATCH] fix(e2e): tolerate a missing system.replicas grant in mv_replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ObsessionDB's managed role cannot SELECT system.replicas, so the replica count added in #225 failed the empty-target e2e with ACCESS_DENIED. Fall back to one visibility sample, skip SYSTEM SYNC REPLICA only when that command is denied too, and keep parallel-replica disable plus empty-chunk replay. Co-authored-by: Marc Höffl --- .../src/mv-replay-plan.e2e.test.ts | 43 ++++++--- .../src/mv-replay-visibility.test.ts | 94 +++++++++++++++++++ .../src/mv-replay-visibility.ts | 66 +++++++++++++ packages/plugin-backfill/tsconfig.json | 2 +- 4 files changed, 189 insertions(+), 16 deletions(-) create mode 100644 packages/plugin-backfill/src/mv-replay-visibility.test.ts create mode 100644 packages/plugin-backfill/src/mv-replay-visibility.ts diff --git a/packages/plugin-backfill/src/mv-replay-plan.e2e.test.ts b/packages/plugin-backfill/src/mv-replay-plan.e2e.test.ts index 83e7c7f2..73345e0e 100644 --- a/packages/plugin-backfill/src/mv-replay-plan.e2e.test.ts +++ b/packages/plugin-backfill/src/mv-replay-plan.e2e.test.ts @@ -19,6 +19,11 @@ import { import { executeBackfill, type BackfillResult } from './async-backfill.js' import { buildChunkExecutionSql } from './chunking/sql.js' import { generateIdempotencyToken } from './chunking/utils/ids.js' +import { + resolveReplicaVisibility, + syncReplicaUnlessDenied, + type ReplicaVisibility, +} from './mv-replay-visibility.js' import { PlanSchema } from './options.js' import { buildBackfillPlan } from './planner.js' import type { Chunk, PlannerQuery } from './chunking/types.js' @@ -50,10 +55,10 @@ let runExecutor: ClickHouseExecutor let plannerQuery: PlannerQuery let liveEnv: LiveEnv let db: string -// Plain MergeTree has one copy of the data. Replicated/Shared engines need -// every active replica to attach the source parts before a chunk reads them. -let syncSourceReplica = false -let sourceReplicas = 1 +// Plain MergeTree has one copy of the data. Replicated/Shared engines sync +// each sampled replica before a chunk reads it, when those commands are granted. +let replicaVisibility: ReplicaVisibility = { kind: 'single-node', samples: 1 } +let syncReplicaDenied = false let sourceTable: string let targetTable: string let sourceFqn: string @@ -76,17 +81,23 @@ async function aggregateByBucket(fqn: string, valueExpr: string): Promise { let readySamples = 0 const ready = await pollUntil(async () => { const session = createLiveExecutor(liveEnv) try { - if (syncSourceReplica) { - await session.command( - `SYSTEM SYNC REPLICA ${quoteIdent(db)}.${quoteIdent(sourceTable)} LIGHTWEIGHT`, + if (replicaVisibility.kind !== 'single-node' && !syncReplicaDenied) { + const sync = await syncReplicaUnlessDenied(() => + session.command( + `SYSTEM SYNC REPLICA ${quoteIdent(db)}.${quoteIdent(sourceTable)} LIGHTWEIGHT`, + ), ) + if (sync === 'denied') syncReplicaDenied = true } const [row] = await session.query<{ cnt: string }>( `SELECT toString(count()) AS cnt FROM ${sourceFqn} SETTINGS select_sequential_consistency = 1`, @@ -96,8 +107,8 @@ async function waitUntilSourceVisible(): Promise { } finally { await session.close() } - }, (samples) => samples >= sourceReplicas, { timeoutMs: 45_000, intervalMs: 250 }) - expect(ready, `replicas that can see all ${SOURCE_ROWS} source rows`).toBeGreaterThanOrEqual(sourceReplicas) + }, (samples) => samples >= replicaVisibility.samples, { timeoutMs: 45_000, intervalMs: 250 }) + expect(ready, `replicas that can see all ${SOURCE_ROWS} source rows (${replicaVisibility.kind})`).toBeGreaterThanOrEqual(replicaVisibility.samples) } function chunkExecutionSql(planId: string, chunk: Chunk, plan: BackfillPlanState): string { @@ -195,15 +206,14 @@ beforeAll(async () => { const [engineRow] = await ddl.query<{ engine: string }>( `SELECT engine FROM system.tables WHERE database = '${db}' AND name = '${sourceTable}'`, ) - syncSourceReplica = /Shared|Replicated/.test(engineRow?.engine ?? '') - if (syncSourceReplica) { + replicaVisibility = await resolveReplicaVisibility(engineRow?.engine, async () => { const [replicaRow] = await ddl.query<{ active: string }>( `SELECT toString(active_replicas) AS active FROM system.replicas WHERE database = '${db}' AND table = '${sourceTable}'`, ) - sourceReplicas = Math.max(1, Number(replicaRow?.active ?? 0)) - } + return Number(replicaRow?.active ?? 0) + }) const rows = Array.from({ length: SOURCE_ROWS }, (_, i) => ({ id: i, @@ -312,6 +322,9 @@ describe('e2e: mv_replay backfill of an empty aggregate target (chkit#187)', () () => aggregateByBucket(targetFqn, 'sum(total)'), (rows) => Bun.deepEquals(rows, expected), ) - expect(actual, `rows written per chunk attempt: ${JSON.stringify(writtenAttempts)}`).toEqual(expected) + expect( + actual, + `rows written per chunk attempt: ${JSON.stringify(writtenAttempts)} (${replicaVisibility.kind}, syncDenied=${syncReplicaDenied})`, + ).toEqual(expected) }, 240_000) }) diff --git a/packages/plugin-backfill/src/mv-replay-visibility.test.ts b/packages/plugin-backfill/src/mv-replay-visibility.test.ts new file mode 100644 index 00000000..08041b41 --- /dev/null +++ b/packages/plugin-backfill/src/mv-replay-visibility.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, test } from 'bun:test' + +import { + engineNeedsReplicaSync, + isAccessDenied, + resolveReplicaVisibility, + syncReplicaUnlessDenied, +} from './mv-replay-visibility.js' + +const REPLICAS_DENIED = Object.assign( + new Error( + "default: Not enough privileges. To execute this query, it's necessary to have the grant SELECT(active_replicas, database, `table`) ON system.replicas.", + ), + { code: '497', type: 'ACCESS_DENIED' }, +) + +describe('mv_replay replica visibility', () => { + test('treats a plain MergeTree as a single copy of the data', () => { + expect(engineNeedsReplicaSync('MergeTree')).toBe(false) + expect(engineNeedsReplicaSync('SummingMergeTree')).toBe(false) + expect(engineNeedsReplicaSync(undefined)).toBe(false) + }) + + test('treats Shared and Replicated engines as multi-replica', () => { + expect(engineNeedsReplicaSync('SharedMergeTree')).toBe(true) + expect(engineNeedsReplicaSync('ReplicatedMergeTree')).toBe(true) + }) + + test('does not read system.replicas when the engine is not replicated', async () => { + let reads = 0 + const visibility = await resolveReplicaVisibility('MergeTree', async () => { + reads += 1 + return 4 + }) + expect(reads).toBe(0) + expect(visibility).toEqual({ kind: 'single-node', samples: 1 }) + }) + + test('uses the active replica count when system.replicas is readable', async () => { + const visibility = await resolveReplicaVisibility('SharedMergeTree', async () => 3) + expect(visibility).toEqual({ kind: 'counted', activeReplicas: 3, samples: 3 }) + }) + + test('requires at least one sample when the reported replica count is empty', async () => { + const visibility = await resolveReplicaVisibility('ReplicatedMergeTree', async () => 0) + expect(visibility).toEqual({ kind: 'counted', activeReplicas: 0, samples: 1 }) + }) + + test('falls back when system.replicas SELECT is denied', async () => { + const visibility = await resolveReplicaVisibility('SharedMergeTree', () => Promise.reject(REPLICAS_DENIED)) + expect(visibility).toEqual({ kind: 'grant-denied', samples: 1 }) + }) + + test('recognizes an ACCESS_DENIED payload that is not an Error instance', () => { + expect(isAccessDenied({ code: 497, type: 'ACCESS_DENIED' })).toBe(true) + }) + + test('recognizes a privilege message that lost its error code', () => { + expect(isAccessDenied(new Error('Not enough privileges. To execute this query, it is necessary to have the grant.'))).toBe(true) + }) + + test('does not treat other ClickHouse errors as a missing grant', () => { + const error = Object.assign(new Error('Database foo does not exist'), { code: '81', type: 'UNKNOWN_DATABASE' }) + expect(isAccessDenied(error)).toBe(false) + expect(isAccessDenied('ACCESS_DENIED')).toBe(false) + expect(isAccessDenied(undefined)).toBe(false) + }) + + test('rethrows replica reads that are not a missing grant', async () => { + await expect( + resolveReplicaVisibility('SharedMergeTree', () => Promise.reject(new Error('connection reset'))), + ).rejects.toThrow('connection reset') + }) + + test('reports SYSTEM SYNC REPLICA as denied without throwing', async () => { + const outcome = await syncReplicaUnlessDenied(() => Promise.reject(REPLICAS_DENIED)) + expect(outcome).toBe('denied') + }) + + test('reports SYSTEM SYNC REPLICA as synced when the command succeeds', async () => { + let ran = false + const outcome = await syncReplicaUnlessDenied(async () => { + ran = true + }) + expect(ran).toBe(true) + expect(outcome).toBe('synced') + }) + + test('rethrows sync failures that are not a missing grant', async () => { + await expect( + syncReplicaUnlessDenied(() => Promise.reject(new Error('Table is not replicated'))), + ).rejects.toThrow('Table is not replicated') + }) +}) diff --git a/packages/plugin-backfill/src/mv-replay-visibility.ts b/packages/plugin-backfill/src/mv-replay-visibility.ts new file mode 100644 index 00000000..e4d01286 --- /dev/null +++ b/packages/plugin-backfill/src/mv-replay-visibility.ts @@ -0,0 +1,66 @@ +// Visibility policy for the mv_replay empty-target e2e. Production backfill does +// not read system.replicas; this module stays out of the package build. + +const ACCESS_DENIED_CODE = '497' + +export type ReplicaVisibility = + | { kind: 'single-node'; samples: 1 } + | { kind: 'counted'; activeReplicas: number; samples: number } + | { kind: 'grant-denied'; samples: 1 } + +export type ReplicaSync = 'synced' | 'denied' + +// Shared and Replicated engines keep a copy of each part per replica. Plain +// MergeTree (the single-node verify service) does not. +export function engineNeedsReplicaSync(engine: string | undefined): boolean { + return /Shared|Replicated/.test(engine ?? '') +} + +// `readActiveReplicas` runs only for replicated/shared engines. A missing +// SELECT grant on system.replicas (ObsessionDB's managed role) is not fatal: +// one ready sample is enough to start, and empty-chunk replay covers a +// replica that sample did not visit. +export async function resolveReplicaVisibility( + engine: string | undefined, + readActiveReplicas: () => Promise, +): Promise { + if (!engineNeedsReplicaSync(engine)) return { kind: 'single-node', samples: 1 } + try { + const activeReplicas = await readActiveReplicas() + return { + kind: 'counted', + activeReplicas, + samples: samplesForActiveReplicas(activeReplicas), + } + } catch (error) { + if (!isAccessDenied(error)) throw error + return { kind: 'grant-denied', samples: 1 } + } +} + +// SYSTEM SYNC REPLICA is a different privilege from SELECT on system.replicas. +// When the managed role has it, sync the session we landed on. When it does +// not, keep going — the chunk settings and the empty-chunk replay do not +// need this command. +export async function syncReplicaUnlessDenied(sync: () => Promise): Promise { + try { + await sync() + return 'synced' + } catch (error) { + if (!isAccessDenied(error)) throw error + return 'denied' + } +} + +function samplesForActiveReplicas(activeReplicas: number): number { + if (!Number.isFinite(activeReplicas)) return 1 + return Math.max(1, Math.floor(activeReplicas)) +} + +export function isAccessDenied(error: unknown): boolean { + if (typeof error !== 'object' || error === null) return false + const record = error as { code?: unknown; type?: unknown; message?: unknown } + if (String(record.code ?? '') === ACCESS_DENIED_CODE) return true + if (String(record.type ?? '') === 'ACCESS_DENIED') return true + return typeof record.message === 'string' && /not enough privileges/i.test(record.message) +} diff --git a/packages/plugin-backfill/tsconfig.json b/packages/plugin-backfill/tsconfig.json index d38ac708..c18984ad 100644 --- a/packages/plugin-backfill/tsconfig.json +++ b/packages/plugin-backfill/tsconfig.json @@ -5,5 +5,5 @@ "outDir": "dist" }, "include": ["src/**/*.ts"], - "exclude": ["src/**/*.test.ts"] + "exclude": ["src/**/*.test.ts", "src/mv-replay-visibility.ts"] }