fix(e2e): tolerate missing system.replicas grant in mv_replay

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<Array<
 // select_sequential_consistency does not fetch parts that were never quorum
 // commits; it only hides or rejects blocks the quorum has not confirmed. A
 // replica that has not attached the partition therefore finishes the INSERT
-// with 0 written rows. Sample a fresh session per replica, and on
+// with 0 written rows. Sample a fresh session per active replica, and on
 // replicated/shared engines sync that session's replica before counting.
+// When system.replicas (or SYSTEM SYNC REPLICA) is not granted, this wait
+// only proves one session can see the source. enable_parallel_replicas = 0
+// and the empty-chunk replay below still cover a replica this wait missed.
 async function waitUntilSourceVisible(): Promise<void> {
   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<void> {
     } 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)
 })

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')
+  })
+})

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<number>,
+): Promise<ReplicaVisibility> {
+  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<void>): Promise<ReplicaSync> {
+  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)
+}

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"]
 }