scripts/sync-read-replica-schema.ts
Rama main · réplica de lectura del repositorio original.
import type { CloudSqlDataApiResponse } from '../read_replicate/cloud_sql_data_api_response.ts'
import type {
ReadReplicaSchemaSyncClient,
ReadReplicaSchemaSyncPlan,
ReadReplicaSchemaSyncStatement,
} from '../read_replicate/schema_additive_sync.ts'
import type { Queryable } from '../read_replicate/schema_catalog.ts'
import type { SchemaCompatibilityIssue } from '../read_replicate/schema_compatibility.ts'
import { randomUUID } from 'node:crypto'
import { Client } from 'pg'
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import process from 'node:process'
import { assertCloudSqlDataApiResponseSucceeded } from '../read_replicate/cloud_sql_data_api_response.ts'
import { ALLOWED_NEW_REPLICA_TABLES, planReadReplicaSchemaSync } from '../read_replicate/schema_additive_sync.ts'
import {
READ_REPLICA_SCHEMA_CATALOG_SQL,
readReplicaSchemaCatalog,
REPLICA_TABLES,
stableStringify,
} from '../read_replicate/schema_catalog.ts'
import { readReplicaSchemaCatalogFromMigrations } from '../read_replicate/schema_catalog_from_migrations.ts'
import { readReplicaSubscriberCompatibilityIssues } from '../read_replicate/schema_compatibility.ts'
interface BunSubprocess {
exitCode: number | null
exited: Promise<number>
kill: () => void
stderr: ReadableStream<Uint8Array>
stdout: ReadableStream<Uint8Array>
}
interface BunRuntime {
spawn: (command: string[], options: unknown) => BunSubprocess
}
interface DataApiResponse extends CloudSqlDataApiResponse {
results?: Array<{
columns?: Array<{ name?: string }>
partialResult?: boolean
rows?: Array<{ values?: Array<{ value?: string, nullValue?: boolean }> }>
}>
}
interface GoogleDataApiConfig {
project: string
instance: string
database: string
}
const bunRuntime = (globalThis as unknown as { Bun: BunRuntime }).Bun
const DEFAULT_SYNC_MAX_SECONDS = 30 * 60
const CATALOG_QUERY_BUFFER_MS = 5000
const GOOGLE_DATA_API_MAX_SECONDS = 30
const GOOGLE_DATA_API_PROCESS_GRACE_SECONDS = 5
const GOOGLE_DATA_API_REQUEST_LIMIT_BYTES = 500_000
const GOOGLE_DATA_API_REQUEST_METADATA_BUFFER_BYTES = 1_024
const GOOGLE_DATA_API_RESPONSE_LIMIT_BYTES = 10_000_000
const IMPORT_CLEANUP_MAX_SECONDS = 60
const POSTGRES_IMPORT_USER = 'postgres'
const GCS_IMPORT_BUCKET = 'capgo-read-replica-schema-import-394818'
// Included in Capgo release scope so replica schema sync fixes cut deploy tags.
const GOOGLE_READ_REPLICA: GoogleDataApiConfig = {
project: 'capgo-394818',
instance: 'eu-2',
database: 'postgres',
}
const READ_REPLICA_SUBSCRIPTION = 'capgo_google_eu_2_sub'
// Fixed statements only (plus ANALYZE of known replicated tables): the release
// job never sends other SQL this way.
export const READ_REPLICA_SUBSCRIBED_TABLES_SQL = `SELECT c.relname AS table_name, r.srsubstate AS state FROM pg_catalog.pg_subscription_rel AS r JOIN pg_catalog.pg_subscription AS s ON s.oid = r.srsubid JOIN pg_catalog.pg_class AS c ON c.oid = r.srrelid JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace WHERE s.subname = '${READ_REPLICA_SUBSCRIPTION}' AND n.nspname = 'public'`
// One statement per request runs outside a transaction block, which
// REFRESH PUBLICATION requires. copy_data only copies newly added tables.
export const READ_REPLICA_REFRESH_SUBSCRIPTION_SQL = `ALTER SUBSCRIPTION ${READ_REPLICA_SUBSCRIPTION} REFRESH PUBLICATION WITH (copy_data = true)`
/** Statistics for tables the refresh just copied, so the first reads get index plans. */
export function readReplicaAnalyzeSql(tables: readonly string[]): string {
const known = new Set<string>(REPLICA_TABLES)
const unknown = tables.filter(table => !known.has(table))
if (unknown.length)
throw new Error(`Not a replicated table: ${unknown.join(', ')}`)
return `ANALYZE ${tables.map(table => `public."${table}"`).join(', ')}`
}
/** Replicated tables the subscription does not stream yet. */
export function unsubscribedReplicaTables(subscribed: readonly string[]): string[] {
const known = new Set(subscribed)
return REPLICA_TABLES.filter(table => !known.has(table))
}
/** Replicated tables whose initial copy is not finished ('r' = ready). */
export function copyingReplicaTables(rows: readonly { table_name: string, state: string }[]): string[] {
const ready = new Set(rows.filter(row => row.state === 'r').map(row => row.table_name))
return REPLICA_TABLES.filter(table => !ready.has(table))
}
const SUBSCRIPTION_COPY_POLL_MS = 15_000
/**
* Runs after primary migrations: a table the migrations just added to the
* publication is only streamed once the subscription is refreshed. Tables
* the subscriber already streams are not copied again.
*/
async function refreshSubscription(): Promise<void> {
const deadline = Date.now() + DEFAULT_SYNC_MAX_SECONDS * 1000
const { project, instance, database } = GOOGLE_READ_REPLICA
const subscribed = async () => dataApiRows(
await executeGoogleSql(project, instance, database, READ_REPLICA_SUBSCRIBED_TABLES_SQL, deadline),
).map(row => ({ table_name: String(row.table_name), state: String(row.state) }))
const missing = unsubscribedReplicaTables((await subscribed()).map(row => row.table_name))
if (missing.length) {
console.log(`Refreshing ${READ_REPLICA_SUBSCRIPTION} for: ${missing.join(', ')}`)
await executeGoogleSql(project, instance, database, READ_REPLICA_REFRESH_SUBSCRIPTION_SQL, deadline)
}
// Workers that read the replica deploy after this job: wait for the copy.
for (;;) {
const rows = await subscribed()
const stillMissing = unsubscribedReplicaTables(rows.map(row => row.table_name))
if (stillMissing.length) {
throw new Error(
`Subscription ${READ_REPLICA_SUBSCRIPTION} still does not stream ${stillMissing.join(', ')}; check that the primary publication includes them.`,
)
}
const copying = copyingReplicaTables(rows)
if (!copying.length)
break
if (Date.now() + SUBSCRIPTION_COPY_POLL_MS > deadline)
throw new Error(`Read-replica initial copy still running for ${copying.join(', ')}`)
console.log(`Waiting for the read-replica initial copy of: ${copying.join(', ')}`)
await new Promise(resolve => setTimeout(resolve, SUBSCRIPTION_COPY_POLL_MS))
}
if (missing.length)
await executeGoogleSql(project, instance, database, readReplicaAnalyzeSql(missing), deadline)
console.log('Read-replica subscription streams every replicated table.')
}
async function main(): Promise<void> {
if (process.argv.includes('--refresh-subscription')) {
await refreshSubscription()
return
}
const deadline = Date.now() + DEFAULT_SYNC_MAX_SECONDS * 1000
const dryRun = process.argv.includes('--dry-run')
console.log(
'Building the read-replica schema catalog from local migrations through Tinbase/PGlite...',
)
const expected = await readReplicaSchemaCatalogFromMigrations()
const replicaRead = await connectGoogleReplica(deadline)
const replica = googleDataApiClient(GOOGLE_READ_REPLICA, deadline)
try {
const actual = await readReplicaSchemaCatalog(replicaRead)
const plan = planReadReplicaSchemaSync(expected, actual)
// Both checks run before the SQL object is uploaded or Cloud SQL is asked to
// import anything. The import receives this exact already-reviewed plan.
assertGoogleReadReplicaSchemaPlan(plan)
const preflightIssues = preflightCompatibilityIssues(expected, actual, plan)
if (preflightIssues.length) {
throw new Error(
`Read-replica schema preflight found residual drift: ${stableStringify(preflightIssues)}`,
)
}
if (dryRun) {
console.log('Read-replica schema dry run result:')
console.log(stableStringify({
statements: plan.statements.map(({ kind, table, name }) => ({ kind, table, name })),
skipped: plan.skipped,
}))
return
}
if (plan.statements.length) {
const applyPlan = replica.applyReadReplicaSchemaPlan
if (!applyPlan) {
throw new Error(
'Cloud SQL schema reconciliation requires a server-side plan importer.',
)
}
await applyPlan(plan)
}
const finalActual = await readReplicaSchemaCatalog(replicaRead)
const issues = readReplicaSubscriberCompatibilityIssues(expected, finalActual)
const result = {
applied: plan.statements.map(({ kind, table, name }) => ({ kind, table, name })),
skipped: plan.skipped,
}
if (issues.length) {
console.error(
'::error title=Read-replica schema did not converge::Cloud SQL server-side import completed with residual drift.',
)
console.error(
stableStringify({ error: 'schema_not_converged', ...result, issues }),
)
process.exitCode = 1
return
}
console.log('Read-replica Cloud SQL server-side import result:')
console.log(stableStringify({ ...result, issues }))
console.log(
'Read replica accepts the schema derived from local migrations before primary migrations.',
)
}
finally {
await replicaRead.close()
}
}
async function connectGoogleReplica(deadline: number): Promise<Queryable & { close: () => Promise<void> }> {
const client = new Client({
connectionString: replicaConnectionString(),
connectionTimeoutMillis: 10_000,
})
await client.connect()
const bounded = deadlineBoundCatalogClient({
query: (queryText, values) => client.query(queryText, values),
}, deadline)
return {
query: (queryText, values) => bounded.query(queryText, values),
close: () => client.end(),
}
}
function deadlineBoundCatalogClient(client: Queryable, deadline: number): Queryable {
return {
async query(queryText: string, values?: unknown[]) {
if (queryText !== READ_REPLICA_SCHEMA_CATALOG_SQL)
return client.query(queryText, values)
const timeoutMs = remainingBudgetMs(deadline, 'read the schema catalog')
await client.query(`SET statement_timeout = ${timeoutMs}`)
try {
return await client.query(queryText, values)
}
finally {
await client.query('RESET statement_timeout')
}
},
}
}
function replicaConnectionString(): string {
const connectionString = process.env.READ_REPLICATE_GOOGLE_EU1
?? process.env.READ_REPLICA_DB_URL
?? process.env.GOOGLE_READ_REPLICA_DB_URL
if (!connectionString) {
throw new Error(
'Set READ_REPLICATE_GOOGLE_EU1 to the direct Google subscriber PostgreSQL URL.',
)
}
return connectionString
}
export function preflightCompatibilityIssues(
expected: unknown,
actual: unknown,
plan: ReadReplicaSchemaSyncPlan,
): SchemaCompatibilityIssue[] {
return readReplicaSubscriberCompatibilityIssues(expected, actual).filter(issue =>
!plan.statements.some(statement => statementResolvesCompatibilityIssue(statement, issue)),
)
}
function statementResolvesCompatibilityIssue(
statement: ReadReplicaSchemaSyncStatement,
issue: SchemaCompatibilityIssue,
): boolean {
// A created allowlisted table brings its columns, constraints, primary key
// index and identity sequence.
const created = statement.kind === 'table' ? ALLOWED_NEW_REPLICA_TABLES[statement.table] : undefined
if (created) {
return (issue.kind === 'table' && issue.object === statement.table)
|| ((issue.kind === 'column' || issue.kind === 'constraint') && issue.object.startsWith(`${statement.table}.`))
|| (issue.kind === 'index' && created.catalog.indexes.some(index => index.name === issue.object))
|| (issue.kind === 'sequence' && created.catalog.sequences.some(sequence => sequence.name === issue.object))
}
switch (issue.kind) {
case 'column':
return statement.kind === issue.kind
&& issue.object === `${statement.table}.${statement.name}`
case 'constraint':
return (
(statement.kind === 'constraint'
|| statement.kind === 'drop_check_constraint'
|| statement.kind === 'check_constraint')
&& issue.object === `${statement.table}.${statement.name}`
)
case 'type':
case 'sequence':
case 'function':
return statement.kind === issue.kind && issue.object === statement.name
case 'index':
return (
(statement.kind === 'index'
|| statement.kind === 'invalid_index'
|| statement.kind === 'drop_index')
&& issue.object === statement.name
)
default:
return false
}
}
function googleDataApiClient(
config: GoogleDataApiConfig,
deadline: number,
): ReadReplicaSchemaSyncClient {
const { project, instance, database } = config
return {
assertCanApplyReadReplicaSchemaPlan(plan) {
assertGoogleReadReplicaSchemaPlan(plan)
},
async applyReadReplicaSchemaPlan(plan) {
const {
preIndexAtomicStatements,
indexStatements,
postIndexAtomicStatements,
} = partitionReadReplicaImportStatements(plan.statements)
// New tables and columns/types/sequences first, then indexes, then
// USING INDEX attaches.
// Index DDL is imported outside BEGIN/COMMIT and without CONCURRENTLY
// because Cloud SQL managed SQL import is transactional.
if (preIndexAtomicStatements.length) {
await importReadReplicaSchemaTransaction(
config,
renderReadReplicaImportTransaction(preIndexAtomicStatements),
deadline,
)
}
if (indexStatements.length) {
await importReadReplicaSchemaTransaction(
config,
renderReadReplicaIndexImport(indexStatements),
deadline,
)
}
if (postIndexAtomicStatements.length) {
await importReadReplicaSchemaTransaction(
config,
renderReadReplicaImportTransaction(postIndexAtomicStatements),
deadline,
)
}
},
async query(queryText: string, values?: unknown[]) {
// The Data API has no session affinity and rejects requests over 0.5 MB or
// responses over 10 MB. It is only used for bounded selected-schema reads.
if (
queryText.startsWith('SET statement_timeout')
|| queryText === 'RESET statement_timeout'
) {
return { rows: [] }
}
assertGoogleDataApiCatalogQuery(queryText)
const sql = renderCatalogQueryWithStaticValues(queryText, values)
return {
rows: dataApiRows(
await executeGoogleSql(project, instance, database, sql, deadline),
),
}
},
}
}
export function assertGoogleReadReplicaSchemaPlan(
plan: ReadReplicaSchemaSyncPlan,
): void {
const skipped = plan.skipped[0]
if (skipped) {
throw new Error(
`Cloud SQL server-side import cannot reconcile skipped ${skipped.kind} ${skipped.table}.${skipped.name} (${skipped.reason}) before primary migrations.`,
)
}
for (const statement of plan.statements)
assertGoogleReadReplicaSchemaStatement(statement)
}
function assertGoogleReadReplicaSchemaStatement(
statement: ReadReplicaSchemaSyncStatement,
): void {
if (statement.sql.includes(';') || statement.sql.includes('\0')) {
throw new Error(
`Cloud SQL server-side import rejected unsafe ${statement.kind} ${statement.table}.${statement.name}.`,
)
}
switch (statement.kind) {
case 'table':
assertSelectedReplicaTable(statement)
if (
statement.name !== statement.table
|| statement.sql !== ALLOWED_NEW_REPLICA_TABLES[statement.table]?.sql
) {
throw new Error(
`Cloud SQL server-side import cannot create unsupported table ${statement.table}.`,
)
}
return
case 'column':
assertSelectedReplicaTable(statement)
assertColumnStatement(statement)
return
case 'constraint':
assertSelectedReplicaTable(statement)
assertConstraintStatement(statement)
return
case 'drop_check_constraint':
assertSelectedReplicaTable(statement)
assertDropCheckConstraintStatement(statement)
return
case 'check_constraint':
assertSelectedReplicaTable(statement)
assertCheckConstraintStatement(statement)
return
case 'type':
assertTypeStatement(statement)
return
case 'sequence':
assertSequenceStatement(statement)
return
case 'index':
case 'invalid_index':
case 'drop_index':
assertSelectedReplicaTable(statement)
assertIndexStatement(statement)
return
default:
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function assertColumnStatement(statement: ReadReplicaSchemaSyncStatement): void {
const prefix = `ALTER TABLE public.${quoteSqlIdentifier(statement.table)} ADD COLUMN IF NOT EXISTS ${quoteSqlIdentifier(statement.name)} `
if (
!isSafeIdentifier(statement.name)
|| !statement.sql.startsWith(prefix)
|| !isSafeSchemaFragment(statement.sql.slice(prefix.length))
) {
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function assertConstraintStatement(statement: ReadReplicaSchemaSyncStatement): void {
const expected = `ALTER TABLE public.${quoteSqlIdentifier(statement.table)} ADD CONSTRAINT ${quoteSqlIdentifier(statement.name)} `
const primaryKey = `PRIMARY KEY USING INDEX ${quoteSqlIdentifier(statement.name)}`
const unique = `UNIQUE USING INDEX ${quoteSqlIdentifier(statement.name)}`
const tail = statement.sql.slice(expected.length)
if (
!isSafeQuotedIdentifier(statement.name)
|| !statement.sql.startsWith(expected)
|| (tail !== primaryKey && tail !== unique)
) {
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function assertDropCheckConstraintStatement(
statement: ReadReplicaSchemaSyncStatement,
): void {
const expected = `ALTER TABLE public.${quoteSqlIdentifier(statement.table)} DROP CONSTRAINT IF EXISTS ${quoteSqlIdentifier(statement.name)}`
if (
!isSafeQuotedIdentifier(statement.name)
|| statement.sql !== expected
) {
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function assertCheckConstraintStatement(
statement: ReadReplicaSchemaSyncStatement,
): void {
const expectedPrefix = `ALTER TABLE public.${quoteSqlIdentifier(statement.table)} ADD CONSTRAINT ${quoteSqlIdentifier(statement.name)} CHECK (`
if (
!isSafeQuotedIdentifier(statement.name)
|| !statement.sql.startsWith(expectedPrefix)
|| !isSafeCheckConstraintTail(statement.sql.slice(expectedPrefix.length))
) {
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function isSafeCheckConstraintTail(value: string): boolean {
const body = value.endsWith(' NOT VALID')
? value.slice(0, -' NOT VALID'.length)
: value
return (
body.length > 0
&& body.endsWith(')')
&& !value.includes('\0')
&& !value.includes(';')
&& !value.includes('--')
&& !value.includes('/*')
&& !value.includes('*/')
)
}
function assertTypeStatement(statement: ReadReplicaSchemaSyncStatement): void {
const identifier = quoteSqlIdentifier(statement.name)
const createPrefix = `CREATE TYPE public.${identifier} AS `
const alterPrefix = `ALTER TYPE public.${identifier} ADD `
const validCreate = statement.sql.startsWith(createPrefix)
&& isSafeTypeDefinition(statement.sql.slice(createPrefix.length))
const validAlter = statement.sql.startsWith(alterPrefix)
&& isSafeTypeAlteration(statement.sql.slice(alterPrefix.length))
if (
statement.table !== 'public'
|| !isSafeIdentifier(statement.name)
|| (!validCreate && !validAlter)
) {
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function assertSequenceStatement(statement: ReadReplicaSchemaSyncStatement): void {
const identifier = quoteSqlIdentifier(statement.name)
const createPrefix = `CREATE SEQUENCE IF NOT EXISTS public.${identifier} `
const alterPrefix = `ALTER SEQUENCE public.${identifier} `
const options = statement.sql.startsWith(createPrefix)
? statement.sql.slice(createPrefix.length)
: statement.sql.startsWith(alterPrefix)
? statement.sql.slice(alterPrefix.length)
: undefined
if (
statement.table !== 'public'
|| !isSafeIdentifier(statement.name)
|| !options
|| !isSafeSequenceOptions(options)
) {
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function assertSelectedReplicaTable(statement: ReadReplicaSchemaSyncStatement): void {
if (!REPLICA_TABLES.includes(statement.table as never)) {
throw new Error(
`Cloud SQL server-side import rejected non-subscriber table ${statement.table}.`,
)
}
}
function isIndexImportStatement(
statement: ReadReplicaSchemaSyncStatement,
): boolean {
return (
statement.kind === 'index'
|| statement.kind === 'invalid_index'
|| statement.kind === 'drop_index'
)
}
export function partitionReadReplicaImportStatements(
statements: readonly ReadReplicaSchemaSyncStatement[],
): {
preIndexAtomicStatements: ReadReplicaSchemaSyncStatement[]
indexStatements: ReadReplicaSchemaSyncStatement[]
postIndexAtomicStatements: ReadReplicaSchemaSyncStatement[]
} {
const preIndexAtomicStatements: ReadReplicaSchemaSyncStatement[] = []
const indexStatements: ReadReplicaSchemaSyncStatement[] = []
const postIndexAtomicStatements: ReadReplicaSchemaSyncStatement[] = []
for (const statement of statements) {
if (isIndexImportStatement(statement))
indexStatements.push(statement)
else if (statement.kind === 'constraint')
postIndexAtomicStatements.push(statement)
else
preIndexAtomicStatements.push(statement)
}
return {
preIndexAtomicStatements,
indexStatements,
postIndexAtomicStatements,
}
}
function assertIndexStatement(statement: ReadReplicaSchemaSyncStatement): void {
if (!isSafeQuotedIdentifier(statement.name)) {
throw new Error(
`Cloud SQL server-side import cannot apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
const quotedName = quoteSqlIdentifier(statement.name)
const quotedTable = quoteSqlIdentifier(statement.table)
const createPrefix = `CREATE INDEX CONCURRENTLY IF NOT EXISTS ${quotedName} ON public.${quotedTable} `
const createUniquePrefix = `CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS ${quotedName} ON public.${quotedTable} `
const dropSql = `DROP INDEX CONCURRENTLY IF EXISTS public.${quotedName}`
const reindexSql = `REINDEX INDEX CONCURRENTLY public.${quotedName}`
if (statement.kind === 'drop_index') {
if (statement.sql !== dropSql) {
throw new Error(
`Cloud SQL server-side import cannot apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
return
}
if (statement.kind === 'invalid_index') {
if (statement.sql !== reindexSql) {
throw new Error(
`Cloud SQL server-side import cannot apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
return
}
const createTail = statement.sql.startsWith(createPrefix)
? statement.sql.slice(createPrefix.length)
: statement.sql.startsWith(createUniquePrefix)
? statement.sql.slice(createUniquePrefix.length)
: undefined
if (!createTail || !isSafeSchemaFragment(createTail)) {
throw new Error(
`Cloud SQL server-side import cannot apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
}
function isSafeIdentifier(value: string): boolean {
return /^[A-Za-z_]\w*$/u.test(value)
}
function isSafeQuotedIdentifier(value: string): boolean {
return value.length > 0 && !value.includes('\0')
}
function isSafeSchemaFragment(value: string): boolean {
return /^[\w #.()[\],:'"{}+\-=<>!]+$/u.test(value)
}
function isSafeTypeDefinition(value: string): boolean {
return (
(value.startsWith('ENUM (') || value.startsWith('('))
&& value.endsWith(')')
&& /^[\w .()[\],:'"{}+\-=<>!\\]+$/u.test(value)
)
}
function isSafeTypeAlteration(value: string): boolean {
return (
(value.startsWith('VALUE IF NOT EXISTS ') || value.startsWith('ATTRIBUTE '))
&& /^[\w .()[\],:'"{}+\-=<>!\\]+$/u.test(value)
)
}
function isSafeSequenceOptions(value: string): boolean {
const match = /^AS (?:smallint|integer|bigint) START WITH -?\d+ INCREMENT BY -?\d+ MINVALUE -?\d+ MAXVALUE -?\d+ CACHE \d+ (?:CYCLE|NO CYCLE) (?:OWNED BY NONE|OWNED BY public\."([A-Za-z_]\w*)"\."[A-Za-z_]\w*")$/u.exec(value)
if (!match)
return false
return !match[1] || REPLICA_TABLES.includes(match[1] as never)
}
export function assertGoogleDataApiCatalogQuery(queryText: string): void {
if (queryText !== READ_REPLICA_SCHEMA_CATALOG_SQL) {
throw new Error(
'Cloud SQL Data API schema adapter only permits selected-schema catalog reads outside the server-side import.',
)
}
}
export function renderReadReplicaImportTransaction(
statements: readonly ReadReplicaSchemaSyncStatement[],
): string {
if (!statements.length) {
throw new Error(
'Read-replica server-side import requires at least one reviewed statement',
)
}
for (const statement of statements) {
if (isIndexImportStatement(statement)) {
throw new Error(
`Cloud SQL server-side import cannot atomically apply ${statement.kind} ${statement.table}.${statement.name} before primary migrations.`,
)
}
assertGoogleReadReplicaSchemaStatement(statement)
}
return `BEGIN;\n${statements.map(statement => statement.sql).join(';\n')};\nCOMMIT;`
}
export function renderReadReplicaIndexImport(
statements: readonly ReadReplicaSchemaSyncStatement[],
): string {
if (!statements.length) {
throw new Error(
'Read-replica server-side index import requires at least one reviewed statement',
)
}
for (const statement of statements) {
if (!isIndexImportStatement(statement)) {
throw new Error(
`Cloud SQL server-side index import rejected non-index ${statement.kind} ${statement.table}.${statement.name}.`,
)
}
assertGoogleReadReplicaSchemaStatement(statement)
}
// Cloud SQL managed SQL import is transactional, so strip CONCURRENTLY.
// Keep planner statements validated as CONCURRENTLY above.
return `${statements.map(statement => renderCloudSqlIndexStatement(statement.sql)).join(';\n')};`
}
function renderCloudSqlIndexStatement(sql: string): string {
return sql
.replace(/^CREATE UNIQUE INDEX CONCURRENTLY /u, 'CREATE UNIQUE INDEX ')
.replace(/^CREATE INDEX CONCURRENTLY /u, 'CREATE INDEX ')
.replace(/^DROP INDEX CONCURRENTLY /u, 'DROP INDEX ')
.replace(/^REINDEX INDEX CONCURRENTLY /u, 'REINDEX INDEX ')
}
async function importReadReplicaSchemaTransaction(
config: GoogleDataApiConfig,
sql: string,
deadline: number,
): Promise<void> {
const workingDirectory = await mkdtemp(
join(tmpdir(), 'capgo-read-replica-schema-'),
)
const object = `schema-${randomUUID()}.sql`
const uri = `gs://${GCS_IMPORT_BUCKET}/${object}`
let uploaded = false
let operationError: unknown
try {
const localSqlPath = join(workingDirectory, 'schema.sql')
await writeFile(localSqlPath, sql, { mode: 0o600 })
await runGcloudCommand(
[
'gcloud',
'storage',
'cp',
localSqlPath,
uri,
'--quiet',
],
deadline,
'Upload the reviewed read-replica schema transaction',
)
uploaded = true
await runGcloudCommand(
[
'gcloud',
'sql',
'import',
'sql',
config.instance,
uri,
`--project=${config.project}`,
`--database=${config.database}`,
`--user=${POSTGRES_IMPORT_USER}`,
'--quiet',
],
deadline,
'Run the Cloud SQL server-side schema import',
)
}
catch (error) {
operationError = error
throw error
}
finally {
try {
if (uploaded) {
await runGcloudCommand(
[
'gcloud',
'storage',
'rm',
uri,
'--quiet',
],
Date.now() + IMPORT_CLEANUP_MAX_SECONDS * 1000,
'Remove the temporary read-replica schema object',
)
}
}
catch (cleanupError) {
const message = cleanupError instanceof Error
? cleanupError.message
: String(cleanupError)
if (operationError)
console.error(`::warning title=Temporary import object cleanup failed::${message}`)
else
throw cleanupError
}
finally {
await rm(workingDirectory, { force: true, recursive: true })
}
}
}
async function runGcloudCommand(
command: string[],
deadline: number,
operation: string,
): Promise<void> {
const timeoutMs = remainingBudgetMs(deadline, operation)
let child: BunSubprocess | undefined
let timeout: ReturnType<typeof setTimeout> | undefined
let timedOut = false
try {
const childProcess = bunRuntime.spawn(command, {
env: process.env,
stderr: 'pipe',
stdout: 'pipe',
})
child = childProcess
timeout = setTimeout(() => {
timedOut = true
childProcess.kill()
}, timeoutMs)
const [exitCode, stdout, stderr] = await Promise.all([
childProcess.exited,
new Response(childProcess.stdout).text(),
new Response(childProcess.stderr).text(),
])
if (timedOut)
throw new Error(`${operation} exceeded the remaining release time budget.`)
if (exitCode !== 0)
throw new Error(`${operation} failed: ${commandOutput(stderr || stdout)}`)
}
finally {
if (timeout)
clearTimeout(timeout)
if (child?.exitCode === null) {
child.kill()
await child.exited
}
}
}
async function executeGoogleSql(
project: string,
instance: string,
database: string,
sql: string,
deadline: number,
): Promise<DataApiResponse> {
assertDataApiRequestFitsLimit(database, sql)
let child: BunSubprocess | undefined
let timeout: ReturnType<typeof setTimeout> | undefined
let timedOut = false
try {
const commandTimeoutMs = Math.min(
remainingBudgetMs(deadline, 'read the selected schema catalog'),
(GOOGLE_DATA_API_MAX_SECONDS + GOOGLE_DATA_API_PROCESS_GRACE_SECONDS)
* 1000,
)
const command = bunRuntime.spawn(
[
'gcloud',
'sql',
'instances',
'execute-sql',
instance,
`--project=${project}`,
`--database=${database}`,
`--sql=${sql}`,
'--format=json',
'--partial_result_mode=FAIL_PARTIAL_RESULT',
'--quiet',
],
{ stdout: 'pipe', stderr: 'pipe', env: process.env },
)
child = command
timeout = setTimeout(() => {
timedOut = true
command.kill()
}, commandTimeoutMs)
const [exitCode, stdout, stderr] = await Promise.all([
command.exited,
new Response(command.stdout).text(),
new Response(command.stderr).text(),
])
if (timedOut) {
throw new Error(
`Cloud SQL Data API command exceeded its ${GOOGLE_DATA_API_MAX_SECONDS}-second SQL limit.`,
)
}
if (exitCode !== 0)
throw dataApiCommandError(stderr || stdout)
const response = JSON.parse(stdout) as DataApiResponse
assertCloudSqlDataApiResponseSucceeded(
response,
dataApiCommandError,
)
assertCompleteDataApiResponse(response)
return response
}
finally {
if (timeout)
clearTimeout(timeout)
if (child?.exitCode === null) {
child.kill()
await child.exited
}
}
}
function assertDataApiRequestFitsLimit(database: string, sql: string): void {
const payloadBytes = new TextEncoder().encode(
JSON.stringify({
database,
partialResultMode: 'FAIL_PARTIAL_RESULT',
sqlStatement: sql,
}),
).byteLength
const maximumPayloadBytes
= GOOGLE_DATA_API_REQUEST_LIMIT_BYTES
- GOOGLE_DATA_API_REQUEST_METADATA_BUFFER_BYTES
if (payloadBytes > maximumPayloadBytes) {
throw new Error(
`Cloud SQL Data API SQL payload is ${payloadBytes} bytes and exceeds the safe ${maximumPayloadBytes}-byte budget below its 0.5 MB request limit. Keep the selected schema catalog bounded.`,
)
}
}
function assertCompleteDataApiResponse(response: DataApiResponse): void {
if (response.results?.some(result => result.partialResult)) {
throw new Error(
`Cloud SQL Data API returned a partial result. It refuses catalog responses at or above its ${GOOGLE_DATA_API_RESPONSE_LIMIT_BYTES / 1_000_000} MB limit.`,
)
}
}
function dataApiRows(response: DataApiResponse): Record<string, any>[] {
const result = response.results?.[0]
const columns = result?.columns ?? []
return (result?.rows ?? []).map((row) => {
const mapped: Record<string, any> = {}
for (const [index, column] of columns.entries()) {
const value = row.values?.[index]
if (!column.name)
continue
if (value?.nullValue) {
mapped[column.name] = null
continue
}
mapped[column.name]
= column.name === 'catalog' && value?.value
? JSON.parse(value.value)
: value?.value
}
return mapped
})
}
function renderCatalogQueryWithStaticValues(
queryText: string,
values: unknown[] | undefined,
): string {
const parameters = values ?? []
if (parameters.length !== 5) {
throw new Error(
'Read-replica schema catalog requires five selected-schema parameter arrays.',
)
}
let sql = queryText
for (const [index, value] of parameters.entries()) {
const placeholder = `$${index + 1}::text[]`
if (!sql.includes(placeholder)) {
throw new Error(
'Read-replica schema catalog query did not contain its expected parameter placeholders.',
)
}
sql = sql.replaceAll(placeholder, postgresTextArray(value))
}
return sql
}
function postgresTextArray(value: unknown): string {
if (
!Array.isArray(value)
|| !value.every(entry => typeof entry === 'string')
) {
throw new Error(
'Read-replica schema catalog parameters must be arrays of selected object names.',
)
}
return `ARRAY[${value.map(quoteSqlText).join(', ')}]::text[]`
}
function quoteSqlText(value: string): string {
if (value.includes('\0')) {
throw new Error(
'Read-replica schema catalog parameters cannot contain null bytes.',
)
}
return `'${value.replaceAll('\'', '\'\'')}'`
}
function quoteSqlIdentifier(value: string): string {
if (!isSafeQuotedIdentifier(value))
throw new Error('Read-replica schema identifier must be valid')
return `"${value.replaceAll('"', '""')}"`
}
function remainingBudgetMs(deadline: number, operation: string): number {
const remainingMs = deadline - Date.now() - CATALOG_QUERY_BUFFER_MS
if (remainingMs <= 0) {
throw new Error(
`Read-replica schema sync exceeded max duration before it could ${operation}.`,
)
}
return remainingMs
}
function dataApiCommandError(value: string): Error {
const output = commandOutput(value)
if (
/0\.5\s*mb|request.*(?:size|limit)|payload.*(?:size|limit)/i.test(output)
) {
return new Error(
`Cloud SQL Data API rejected the SQL request at its 0.5 MB request limit. Keep the selected schema catalog bounded. gcloud output: ${output}`,
)
}
if (
/10\s*mb|partial.?result|response.*(?:size|limit)|result.*(?:size|limit)/i.test(
output,
)
) {
return new Error(
`Cloud SQL Data API rejected the response at its 10 MB response limit. Keep the selected schema catalog bounded. gcloud output: ${output}`,
)
}
return new Error(`Cloud SQL Data API query failed: ${output}`)
}
function commandOutput(value: string): string {
const message = value.trim().replaceAll(/\s+/g, ' ')
return message ? message.slice(0, 4096) : 'no diagnostic output'
}
if (import.meta.main) {
try {
await main()
}
catch (error) {
const message = error instanceof Error ? error.message : String(error)
console.error(`::error title=Read-replica schema sync failed::${message}`)
process.exitCode = 1
}
}