fix(api): восстановить разделы PostgreSQL в одной транзакции
This commit is contained in:
1 parent
c24ad7810c
commit
f3e205950f
10 files changed
+261
-10
No files matched your search
@@ -1,6 +1,6 @@
|
||||
import { randomUUID } from "node:crypto"
|
||||
import { spawn } from "node:child_process"
|
||||
import { mkdir, rm, writeFile } from "node:fs/promises"
|
||||
import { chmod, mkdir, rm, writeFile } from "node:fs/promises"
|
||||
import os from "node:os"
|
||||
import path from "node:path"
|
||||
import { executeSystemDatabaseRestore, validateSystemDatabaseRestoreArchive, type SystemDatabaseRestoreOutcome } from "./system-database-restore-execution.js"
|
||||
@@ -11,6 +11,7 @@ import { executeDatabaseOperation, type DatabaseOperationWarnings } from "./syst
|
||||
import { restartSchedulerAfterDatabaseOperation, stopSchedulerForDatabaseOperation } from "./scheduler.js"
|
||||
import { pauseRipeBackgroundWork, resumeRipeBackgroundWork } from "./traffic-flow-ripe.js"
|
||||
import { invalidateSystemDatabaseCaches } from "./system-database-cache.js"
|
||||
import { restorePartitionedDatabase } from "./system-database-partition-restore.js"
|
||||
import {
|
||||
restartTrafficFlowListenerAfterDatabaseOperation,
|
||||
stopTrafficFlowListenerForDatabaseOperation,
|
||||
@@ -78,12 +79,18 @@ export async function restoreSystemDatabaseBackup(buffer: Buffer): Promise<Syste
|
||||
return await withDatabaseOperation(() => executeSystemDatabaseRestore(buffer, env.DATABASE_URL, {
|
||||
allocate: async () => {
|
||||
const tempDir = path.join(os.tmpdir(), "mmapp-db-restore")
|
||||
await mkdir(tempDir, { recursive: true })
|
||||
await mkdir(tempDir, { recursive: true, mode: 0o700 })
|
||||
await chmod(tempDir, 0o700)
|
||||
return path.join(tempDir, `restore-${randomUUID()}.dump`)
|
||||
},
|
||||
write: (file, data) => writeFile(file, data),
|
||||
run: async args => { await spawnCollected("pg_restore", args) },
|
||||
remove: file => rm(file, { force: true }),
|
||||
write: (file, data) => writeFile(file, data, { mode: 0o600 }),
|
||||
run: async args => {
|
||||
await restorePartitionedDatabase(args[args.length - 1], env.DATABASE_URL, {
|
||||
command: spawnCollected,
|
||||
prepareSql: file => writeFile(file, "", { mode: 0o600 }),
|
||||
})
|
||||
},
|
||||
remove: async file => { await Promise.all([rm(file, { force: true }), rm(`${file}.sql`, { force: true })]) },
|
||||
invalidate: invalidateSystemDatabaseCaches,
|
||||
}))
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { test } from "node:test"
|
||||
import { partitionRestorePrelude, restorePartitionedDatabase } from "./system-database-partition-restore.js"
|
||||
const toc = "1; 1259 1 TABLE public servers mmapp\n2; 1259 2 TABLE public schema_migrations mmapp\n3; 1259 3 TABLE public server_snapshots mmapp\n"
|
||||
|
||||
test("preparation covers only known application parents present in the archive", () => {
|
||||
const sql = partitionRestorePrelude(toc + "4; 1259 4 TABLE other flow_buckets owner\n5; 1259 5 TABLE public custom_table owner\n")
|
||||
assert.ok(sql.includes("IN ('server_snapshots')"))
|
||||
assert.ok(sql.includes("c.conparentid = 0"))
|
||||
assert.ok(sql.includes("format('ALTER TABLE %I.%I DROP CONSTRAINT %I CASCADE'"))
|
||||
assert.equal(sql.includes("flow_buckets"), false)
|
||||
assert.equal(sql.includes("custom_table"), false)
|
||||
assert.throws(() => partitionRestorePrelude("1; 1259 1 TABLE public unrelated owner\n"), /обязательные таблицы/)
|
||||
})
|
||||
|
||||
test("archive decoding precedes one transactional psql execution with fail-fast SQL", async () => {
|
||||
const calls: string[] = []
|
||||
await restorePartitionedDatabase("fixture.dump", "fixture-db", {
|
||||
prepareSql: async file => { assert.equal(file, "fixture.dump.sql"); calls.push("prepare") },
|
||||
command: async (command, args) => {
|
||||
calls.push(command)
|
||||
if (command === "pg_restore" && args[0] === "--list") return Buffer.from(toc)
|
||||
if (command === "psql") {
|
||||
assert.deepEqual(args.slice(0, 6), ["-X", "--single-transaction", "--set", "ON_ERROR_STOP=on", "-d", "fixture-db"])
|
||||
assert.equal(args[6], "-c"); assert.ok(args[7].includes("DROP CONSTRAINT"))
|
||||
assert.deepEqual(args.slice(8), ["-f", "fixture.dump.sql"])
|
||||
}
|
||||
return Buffer.alloc(0)
|
||||
},
|
||||
})
|
||||
assert.deepEqual(calls, ["pg_restore", "prepare", "pg_restore", "psql"])
|
||||
})
|
||||
|
||||
test("invalid or undecodable archive never starts target SQL", async () => {
|
||||
for (const phase of ["list", "decode"]) {
|
||||
let executions = 0
|
||||
await assert.rejects(restorePartitionedDatabase("fixture.dump", "fixture-db", {
|
||||
prepareSql: async () => {},
|
||||
command: async (command, args) => {
|
||||
if (command === "psql") executions++
|
||||
if ((phase === "list" && args[0] === "--list") || (phase === "decode" && args[0] === "--clean")) throw Error("archive failed")
|
||||
return Buffer.from(toc)
|
||||
},
|
||||
}), /archive failed/)
|
||||
assert.equal(executions, 0)
|
||||
}
|
||||
})
|
||||
@@ -0,0 +1,47 @@
|
||||
import { PARTITION_SPECS } from "../db/partitions.js"
|
||||
|
||||
/** Only application parents actually present in the archive may be prepared. */
|
||||
export function partitionRestorePrelude(toc: string): string {
|
||||
const tables = new Set<string>()
|
||||
for (const line of toc.split(/\r?\n/)) {
|
||||
const match = /^\d+;\s+\d+\s+\d+\s+TABLE public ([a-z_][a-z_0-9]*)\s/.exec(line)
|
||||
if (match) tables.add(match[1])
|
||||
}
|
||||
if (!tables.has("servers") || !tables.has("schema_migrations")) {
|
||||
throw new Error("Архив не содержит обязательные таблицы базы приложения")
|
||||
}
|
||||
const parents = PARTITION_SPECS.map(s => s.parent).filter(name => tables.has(name))
|
||||
if (!parents.length) return "SELECT 1;"
|
||||
return `DO $mm_restore$
|
||||
DECLARE r record;
|
||||
BEGIN
|
||||
FOR r IN
|
||||
SELECT n.nspname, t.relname, c.conname
|
||||
FROM pg_constraint c
|
||||
JOIN pg_class t ON t.oid = c.conrelid
|
||||
JOIN pg_namespace n ON n.oid = t.relnamespace
|
||||
WHERE n.nspname = 'public' AND t.relkind = 'p'
|
||||
AND c.conparentid = 0 AND c.contype IN ('p', 'u')
|
||||
AND t.relname IN (${parents.map(name => `'${name}'`).join(", ")})
|
||||
LOOP
|
||||
EXECUTE format('ALTER TABLE %I.%I DROP CONSTRAINT %I CASCADE', r.nspname, r.relname, r.conname);
|
||||
END LOOP;
|
||||
END $mm_restore$;`
|
||||
}
|
||||
|
||||
interface Dependencies {
|
||||
command: (command: string, args: string[]) => Promise<Buffer>
|
||||
prepareSql: (file: string) => Promise<void>
|
||||
}
|
||||
|
||||
export async function restorePartitionedDatabase(file: string, databaseUrl: string, deps: Dependencies): Promise<void> {
|
||||
const toc = await deps.command("pg_restore", ["--list", file])
|
||||
const prelude = partitionRestorePrelude(toc.toString("utf8"))
|
||||
const sqlFile = `${file}.sql`
|
||||
await deps.prepareSql(sqlFile)
|
||||
// Fully decode before connecting to the target database; never rewrite archive SQL/data.
|
||||
await deps.command("pg_restore", ["--clean", "--if-exists", "--file", sqlFile, file])
|
||||
// psql wraps both -c and -f in ONE transaction and rolls back on a SQL error.
|
||||
await deps.command("psql", ["-X", "--single-transaction", "--set", "ON_ERROR_STOP=on",
|
||||
"-d", databaseUrl, "-c", prelude, "-f", sqlFile])
|
||||
}
|
||||
Reference in new issue
Block a user