mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
fix(db): preserve binary data in checkpoint backups
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
0748b90062
commit
7a43aa31b9
16 files changed
+748
-120
No files matched your search
@@ -0,0 +1,80 @@
|
||||
name: Backup checkpoint qualification
|
||||
on:
|
||||
pull_request:
|
||||
paths:
|
||||
- 'packages/db/**'
|
||||
- 'server/src/services/*backup*'
|
||||
- 'server/src/__tests__/*backup*'
|
||||
- 'server/src/__tests__/fixtures/database-backup-worker.ts'
|
||||
- 'server/src/index.ts'
|
||||
- 'server/scripts/*backup*'
|
||||
- 'server/scripts/lib/backup-qualification.ts'
|
||||
- 'server/vitest.backup-qualification.config.ts'
|
||||
- '.github/workflows/backup-checkpoint-qualification.yml'
|
||||
- 'package.json'
|
||||
- 'pnpm-lock.yaml'
|
||||
push:
|
||||
branches: [master]
|
||||
permissions:
|
||||
contents: read
|
||||
jobs:
|
||||
backup-checkpoint:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
services:
|
||||
postgres:
|
||||
image: postgres:17
|
||||
env:
|
||||
POSTGRES_USER: paperclip
|
||||
POSTGRES_PASSWORD: backup-qualification-fixture
|
||||
POSTGRES_DB: backup_source_test
|
||||
ports: ['5432:5432']
|
||||
options: >-
|
||||
--health-cmd "pg_isready -U paperclip -d backup_source_test"
|
||||
--health-interval 5s --health-timeout 5s --health-retries 12
|
||||
env:
|
||||
PAPERCLIP_BACKUP_CHECKPOINT_SOURCE_TEST_DATABASE_URL: postgres://paperclip:backup-qualification-fixture@127.0.0.1:5432/backup_source_test
|
||||
PAPERCLIP_BACKUP_CHECKPOINT_RESTORE_TEST_DATABASE_URL: postgres://paperclip:backup-qualification-fixture@127.0.0.1:5432/backup_restore_test
|
||||
PAPERCLIP_BACKUP_QUALIFICATION_RECEIPT: ${{ github.workspace }}/backup-checkpoint-evidence/qualification.json
|
||||
steps:
|
||||
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
|
||||
- uses: pnpm/action-setup@0977fd99725f1db4007ccb2928dbb4e90d06cc86
|
||||
with:
|
||||
version: 9.15.4
|
||||
- uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020
|
||||
with:
|
||||
node-version: '24'
|
||||
cache: pnpm
|
||||
- run: pnpm install --frozen-lockfile --ignore-scripts
|
||||
- name: Build workspace import prerequisites
|
||||
run: |
|
||||
pnpm --filter @paperclipai/shared build
|
||||
pnpm --filter @paperclipai/db build
|
||||
pnpm --filter @paperclipai/plugin-sdk ensure-build-deps
|
||||
- name: Use the service image's matching real PostgreSQL clients
|
||||
env:
|
||||
PG_CONTAINER: ${{ job.services.postgres.id }}
|
||||
run: |
|
||||
mkdir -p "$RUNNER_TEMP/backup-clients" backup-checkpoint-evidence
|
||||
for binary in pg_dump psql; do
|
||||
printf '#!/bin/sh\nexec docker exec --interactive %s %s "$@"\n' "$PG_CONTAINER" "$binary" > "$RUNNER_TEMP/backup-clients/$binary"
|
||||
chmod 700 "$RUNNER_TEMP/backup-clients/$binary"
|
||||
done
|
||||
echo "PAPERCLIP_PG_DUMP_PATH=$RUNNER_TEMP/backup-clients/pg_dump" >> "$GITHUB_ENV"
|
||||
echo "PAPERCLIP_PSQL_PATH=$RUNNER_TEMP/backup-clients/psql" >> "$GITHUB_ENV"
|
||||
"$RUNNER_TEMP/backup-clients/pg_dump" --version
|
||||
"$RUNNER_TEMP/backup-clients/psql" --version
|
||||
"$RUNNER_TEMP/backup-clients/psql" "$PAPERCLIP_BACKUP_CHECKPOINT_SOURCE_TEST_DATABASE_URL" --set=ON_ERROR_STOP=1 -c 'CREATE DATABASE backup_restore_test'
|
||||
- name: Migrate only the owned source fixture
|
||||
env:
|
||||
DATABASE_URL: ${{ env.PAPERCLIP_BACKUP_CHECKPOINT_SOURCE_TEST_DATABASE_URL }}
|
||||
run: pnpm --filter @paperclipai/db exec tsx src/migrate.ts
|
||||
- name: Run required real checkpoint and restore proof
|
||||
run: pnpm test:backup-checkpoint
|
||||
- name: Attach qualification receipt
|
||||
if: always()
|
||||
uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
|
||||
with:
|
||||
name: backup-checkpoint-evidence
|
||||
path: backup-checkpoint-evidence/
|
||||
if-no-files-found: warn
|
||||
@@ -328,6 +328,56 @@ This does not replace filesystem backups, encryption-key recovery or off-provide
|
||||
disaster recovery. Validate restoration of a staging checkpoint before enabling
|
||||
the flag on customer instances.
|
||||
|
||||
#### Repeatable local checkpoint qualification
|
||||
|
||||
Run the qualification command only against two dedicated, disposable PostgreSQL
|
||||
DBs with names that end in `_test`. Provision the source and an empty restore
|
||||
DB first. Use a migrated source when testing the full sleep-safety scan. The
|
||||
command does not create, drop, or migrate databases. It creates a small fixture
|
||||
schema in the source. It refuses an existing fixture and any nonempty restore
|
||||
target. The restore engine uses `--clean`, so never use a DB that holds work.
|
||||
|
||||
Set `PAPERCLIP_BACKUP_CHECKPOINT_SOURCE_TEST_DATABASE_URL` and
|
||||
`PAPERCLIP_BACKUP_CHECKPOINT_RESTORE_TEST_DATABASE_URL`. The command rejects
|
||||
partial configuration, unsafe database names, URL database overrides, and the
|
||||
same actual server/database even through different host names. Set
|
||||
`PAPERCLIP_PG_DUMP_PATH` and `PAPERCLIP_PSQL_PATH` when the matching PostgreSQL
|
||||
binaries are outside `PATH`.
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec tsx scripts/qualify-idle-database-backup.ts /absolute/new/output-directory pg_dump
|
||||
```
|
||||
|
||||
Use a fresh source/restore pair and output directory for a second run with
|
||||
`javascript` as the last argument. This exercises the supported fallback engine
|
||||
separately. The command verifies the full archive, syncs its bytes and directory,
|
||||
restores it, checks relational and binary (including empty bytes)/Unicode/null values, enforces the
|
||||
restored foreign key, and checks identity sequence progress. It writes a JSON
|
||||
receipt with binary versions, database names, archive size/hash, and results.
|
||||
It does not write credentials to the receipt. The fixture rows remain available
|
||||
for inspection. Dispose of these owned databases through the normal test-owner
|
||||
workflow after closing their clients.
|
||||
|
||||
The integration suite is
|
||||
`server/src/__tests__/database-backup-checkpoint-qualification.integration.ts`. Local
|
||||
runs use `pnpm test:backup-checkpoint`, require the same explicit URL pair, and use
|
||||
bounded clients. The dedicated Backup checkpoint qualification CI job provisions
|
||||
both fixtures and matching PostgreSQL clients. Missing URLs or clients fail this
|
||||
required integration invocation; it is separate from the default unit suite. The suite
|
||||
also tests the production backup owner under slow dumps, changed drain ownership,
|
||||
lease expiry, changed work generation, interrupted checkpoint creation, failed
|
||||
wake backup, and successful marker recovery. The filesystem and scheduling
|
||||
suites separately verify sync ordering, integrity failure, retry, drain/resume,
|
||||
and the normal interval after wake. Set `PAPERCLIP_BACKUP_QUALIFICATION_RECEIPT`
|
||||
to retain its JSON evidence and temporary archives.
|
||||
|
||||
These results prove a local logical backup and restore. A writable directory and
|
||||
successful `fsync` do not prove that a provider volume stays attached across a
|
||||
compute stop. Before enabling checkpoint mode, separately verify the deployed
|
||||
backup flags, provider volume identity and mount, and archive continuity across
|
||||
a real stop/restart. A missing mount can appear as a new writable local directory;
|
||||
local tests must not certify that as persistent storage.
|
||||
|
||||
The report inspects accounting and sandbox-cleanup spools directly. Any entry,
|
||||
including malformed JSON, temporary files and failed write probes, blocks sleep.
|
||||
Read errors and invalid spool paths return unknown. Recovery readers that skip
|
||||
|
||||
@@ -24,6 +24,7 @@
|
||||
"typecheck": "pnpm run preflight:workspace-links && pnpm -r typecheck",
|
||||
"typecheck:build-gaps": "pnpm run preflight:workspace-links && pnpm --filter @paperclipai/plugin-sdk ensure-build-deps && pnpm --filter @paperclipai/server build && node scripts/run-typecheck-build-gaps.mjs",
|
||||
"test": "pnpm run test:run",
|
||||
"test:backup-checkpoint": "pnpm --filter @paperclipai/server exec vitest run --config vitest.backup-qualification.config.ts",
|
||||
"test:watch": "pnpm run preflight:workspace-links && vitest",
|
||||
"test:run": "pnpm run preflight:workspace-links && node scripts/run-vitest-stable.mjs",
|
||||
"test:runner-acceptance": "vitest run --config tests/runner-acceptance/vitest.config.ts",
|
||||
|
||||
@@ -288,6 +288,9 @@ function formatSqlValue(
|
||||
if (dataType === "json" || dataType === "jsonb") {
|
||||
return formatSqlLiteral(JSON.stringify(val));
|
||||
}
|
||||
if (dataType === "bytea" && val instanceof Uint8Array) {
|
||||
return `decode('${Buffer.from(val).toString("hex")}', 'hex')`;
|
||||
}
|
||||
if (typeof val === "boolean") return val ? "true" : "false";
|
||||
if (typeof val === "number") return String(val);
|
||||
if (val instanceof Date) return formatSqlLiteral(val.toISOString());
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
import { randomBytes, createHash } from "node:crypto";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { createDb, runDatabaseBackup, runDatabaseRestore, type RunDatabaseBackupOptions } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../src/__tests__/helpers/external-test-database.js";
|
||||
import { verifyIdleDatabaseBackup } from "../../src/services/idle-database-backup.js";
|
||||
|
||||
export const SOURCE_ENV = "PAPERCLIP_BACKUP_CHECKPOINT_SOURCE_TEST_DATABASE_URL";
|
||||
export const RESTORE_ENV = "PAPERCLIP_BACKUP_CHECKPOINT_RESTORE_TEST_DATABASE_URL";
|
||||
export const retention = { dailyDays: 14, weeklyWeeks: 4, monthlyMonths: 6 };
|
||||
|
||||
export function resolveBackupQualificationPair(source = process.env[SOURCE_ENV], restore = process.env[RESTORE_ENV]) {
|
||||
const sourceUrl = resolveExternalTestDatabaseUrl(SOURCE_ENV, source);
|
||||
const restoreUrl = resolveExternalTestDatabaseUrl(RESTORE_ENV, restore);
|
||||
if (!sourceUrl && !restoreUrl) return undefined;
|
||||
if (!sourceUrl || !restoreUrl) throw new Error("Backup qualification requires both source and restore test database URLs");
|
||||
const a = new URL(sourceUrl), b = new URL(restoreUrl);
|
||||
if (a.hostname.toLowerCase() === b.hostname.toLowerCase() && (a.port || "5432") === (b.port || "5432")
|
||||
&& decodeURIComponent(a.pathname) === decodeURIComponent(b.pathname)) throw new Error("Backup source and restore databases must differ");
|
||||
return { sourceUrl, restoreUrl };
|
||||
}
|
||||
|
||||
export type QualificationPair = NonNullable<ReturnType<typeof resolveBackupQualificationPair>>;
|
||||
|
||||
export async function assertDistinctEmptyRestore(pair: QualificationPair) {
|
||||
const source = createDb(pair.sourceUrl, { maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0, connectTimeoutSeconds: 5 }).$client;
|
||||
const target = createDb(pair.restoreUrl, { maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0, connectTimeoutSeconds: 5 }).$client;
|
||||
const key = randomBytes(8); const a = key.readInt32BE(0), b = key.readInt32BE(4);
|
||||
let sourceLocked = false, targetLocked = false;
|
||||
try {
|
||||
await source`SELECT pg_advisory_lock(${a}, ${b})`; sourceLocked = true;
|
||||
const [result] = await target`SELECT pg_try_advisory_lock(${a}, ${b}) AS acquired`;
|
||||
targetLocked = result!.acquired === true;
|
||||
if (!targetLocked) throw new Error("Backup source and restore resolve to the same actual database");
|
||||
const rows = await target`SELECT 1 FROM pg_namespace WHERE nspname NOT IN ('public', 'information_schema') AND nspname NOT LIKE 'pg_%'
|
||||
UNION ALL SELECT 1 FROM pg_depend
|
||||
WHERE refclassid = 'pg_namespace'::regclass AND refobjid = 'public'::regnamespace LIMIT 1`;
|
||||
if (rows.length) throw new Error("Backup restore target must be empty before restore");
|
||||
} finally {
|
||||
try {
|
||||
await Promise.all([
|
||||
...(targetLocked ? [target`SELECT pg_advisory_unlock(${a}, ${b})`] : []),
|
||||
...(sourceLocked ? [source`SELECT pg_advisory_unlock(${a}, ${b})`] : []),
|
||||
]);
|
||||
} finally { await Promise.all([source.end({ timeout: 5 }), target.end({ timeout: 5 })]); }
|
||||
}
|
||||
}
|
||||
|
||||
export async function seedBackupQualificationFixture(sourceUrl: string) {
|
||||
const source = createDb(sourceUrl, { maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0, connectTimeoutSeconds: 5 }).$client;
|
||||
try {
|
||||
await source`CREATE SCHEMA backup_qualification`;
|
||||
await source`CREATE TABLE backup_qualification.parents (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, value text NOT NULL)`;
|
||||
await source`CREATE TABLE backup_qualification.children (id integer PRIMARY KEY, parent_id bigint REFERENCES backup_qualification.parents(id),
|
||||
value text, bytes bytea)`;
|
||||
await source.begin(async tx => {
|
||||
await tx`INSERT INTO backup_qualification.parents (value) VALUES ('last committed state — 雪 ☃')`;
|
||||
await tx`INSERT INTO backup_qualification.children VALUES (1, 1, NULL, decode('00ff10275c', 'hex')), (2, 1, 'empty bytes', decode('', 'hex')), (3, 1, 'null bytes', NULL)`;
|
||||
});
|
||||
} finally { await source.end({ timeout: 5 }); }
|
||||
}
|
||||
|
||||
export async function inspectRestoredFixture(restoreUrl: string) {
|
||||
const target = createDb(restoreUrl, { maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0, connectTimeoutSeconds: 5 }).$client;
|
||||
try {
|
||||
const [row] = await target`SELECT p.id::text AS id, p.value, c.value AS nullable, encode(c.bytes, 'hex') AS binary
|
||||
FROM backup_qualification.parents p JOIN backup_qualification.children c ON c.parent_id = p.id WHERE c.id = 1`;
|
||||
if (!row || row.id !== "1" || row.value !== "last committed state — 雪 ☃" || row.nullable !== null || row.binary !== "00ff10275c") {
|
||||
throw new Error("Restored relational, Unicode, binary or null fixture differs");
|
||||
}
|
||||
const [empty] = await target`SELECT octet_length(bytes) AS size FROM backup_qualification.children WHERE id = 2`;
|
||||
if (empty?.size !== 0) throw new Error("Restored empty binary value differs");
|
||||
const [nullBytes] = await target`SELECT bytes IS NULL AS is_null FROM backup_qualification.children WHERE id = 3`;
|
||||
if (nullBytes?.is_null !== true) throw new Error("Restored null binary value differs");
|
||||
const [next] = await target`INSERT INTO backup_qualification.parents (value) VALUES ('sequence after restore') RETURNING id::text`;
|
||||
if (next?.id !== "2") throw new Error("Restored identity sequence did not resume at 2");
|
||||
try {
|
||||
await target`INSERT INTO backup_qualification.children VALUES (999, 999, NULL, decode('', 'hex'))`;
|
||||
throw new Error("Restored foreign key was not enforced");
|
||||
} catch (error) { if ((error as { code?: string }).code !== "23503") throw error; }
|
||||
return { relational: true, unicode: true, binary: true, emptyBinary: true, nullBinary: true, nullable: true, nextIdentity: next.id, foreignKeyEnforced: true };
|
||||
} finally { await target.end({ timeout: 5 }); }
|
||||
}
|
||||
|
||||
export async function qualifyBackupRestore(pair: QualificationPair, backupDir: string, engine: RunDatabaseBackupOptions["backupEngine"]) {
|
||||
await assertDistinctEmptyRestore(pair);
|
||||
const archive = await runDatabaseBackup({ connectionString: pair.sourceUrl, backupDir, retention,
|
||||
backupEngine: engine, verifyBeforePrune: verifyIdleDatabaseBackup });
|
||||
await assertDistinctEmptyRestore(pair);
|
||||
await runDatabaseRestore({ connectionString: pair.restoreUrl, backupFile: archive.backupFile });
|
||||
return { engine, archive: archive.backupFile, sizeBytes: archive.sizeBytes,
|
||||
sha256: createHash("sha256").update(await readFile(archive.backupFile)).digest("hex"),
|
||||
restored: await inspectRestoredFixture(pair.restoreUrl) };
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import { execFileSync } from "node:child_process";
|
||||
import path from "node:path";
|
||||
import { mkdir, writeFile } from "node:fs/promises";
|
||||
import { resolveBackupQualificationPair, assertDistinctEmptyRestore, seedBackupQualificationFixture, qualifyBackupRestore } from "./lib/backup-qualification.js";
|
||||
|
||||
const pair = resolveBackupQualificationPair();
|
||||
if (!pair) throw new Error("Set both PAPERCLIP_BACKUP_CHECKPOINT_SOURCE_TEST_DATABASE_URL and PAPERCLIP_BACKUP_CHECKPOINT_RESTORE_TEST_DATABASE_URL");
|
||||
const output = process.argv[2];
|
||||
const engine = process.argv[3] ?? "pg_dump";
|
||||
if (!output || !["pg_dump", "javascript"].includes(engine)) throw new Error("Usage: qualify-idle-database-backup.ts <new-output-directory> [pg_dump|javascript]");
|
||||
await mkdir(output);
|
||||
await assertDistinctEmptyRestore(pair);
|
||||
await seedBackupQualificationFixture(pair.sourceUrl);
|
||||
const binaries = Object.fromEntries([process.env.PAPERCLIP_PG_DUMP_PATH || "pg_dump", process.env.PAPERCLIP_PSQL_PATH || "psql"]
|
||||
.map(binary => [binary, execFileSync(binary, ["--version"], { encoding: "utf8" }).trim()]));
|
||||
const result = await qualifyBackupRestore(pair, output, engine as "pg_dump" | "javascript");
|
||||
const receipt = { version: 1, qualifiedAt: new Date().toISOString(), sourceDatabase: new URL(pair.sourceUrl).pathname.slice(1),
|
||||
restoreDatabase: new URL(pair.restoreUrl).pathname.slice(1), binaries, ...result,
|
||||
limits: ["Local logical restore proof; persistent provider volume and deployed sleep/wake require separate qualification"] };
|
||||
await writeFile(path.join(output, "qualification.json"), `${JSON.stringify(receipt, null, 2)}\n`, { flag: "wx" });
|
||||
process.stdout.write(`${JSON.stringify(receipt, null, 2)}\n`);
|
||||
@@ -0,0 +1,188 @@
|
||||
import { execFileSync, spawn } from "node:child_process";
|
||||
import { existsSync } from "node:fs";
|
||||
import { chmod, mkdir, mkdtemp, readFile, rm, utimes, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { gzipSync, gunzipSync } from "node:zlib";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { createDb, runDatabaseBackup, runDatabaseRestore } from "@paperclipai/db";
|
||||
import { createDatabaseBackupOwner } from "../services/database-backup-owner.js";
|
||||
import { idleBackupWakeMarker, verifyIdleDatabaseBackup } from "../services/idle-database-backup.js";
|
||||
import { beginIdleTrackedWork, idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js";
|
||||
import { assertDistinctEmptyRestore, inspectRestoredFixture, qualifyBackupRestore, resolveBackupQualificationPair, retention, seedBackupQualificationFixture } from "../../scripts/lib/backup-qualification.js";
|
||||
|
||||
const configuredPair = resolveBackupQualificationPair();
|
||||
if (!configuredPair) throw new Error("Backup qualification requires an explicit dedicated source/restore test URL pair");
|
||||
const pair = configuredPair;
|
||||
describe("real automatic backup checkpoint qualification", () => {
|
||||
let directory: string;
|
||||
let source: ReturnType<typeof createDb>;
|
||||
const logger = { info() {}, warn() {}, error() {} };
|
||||
const receipts: unknown[] = [];
|
||||
const owner = () => createDatabaseBackupOwner({ connectionString: pair.sourceUrl, backupDir: directory, getRetention: async () => retention, logger });
|
||||
beforeAll(async () => {
|
||||
const receiptFile = process.env.PAPERCLIP_BACKUP_QUALIFICATION_RECEIPT;
|
||||
const parent = receiptFile ? path.dirname(receiptFile) : os.tmpdir();
|
||||
await mkdir(parent, { recursive: true });
|
||||
directory = await mkdtemp(path.join(parent, "backup-checkpoint-qualification-"));
|
||||
source = createDb(pair.sourceUrl, { maxConnections: 1 });
|
||||
await assertDistinctEmptyRestore(pair);
|
||||
await seedBackupQualificationFixture(pair.sourceUrl);
|
||||
}, 120_000);
|
||||
afterAll(async () => {
|
||||
try {
|
||||
const receiptFile = process.env.PAPERCLIP_BACKUP_QUALIFICATION_RECEIPT;
|
||||
if (receiptFile) await writeFile(receiptFile, JSON.stringify({ version: 1, cases: receipts, directory,
|
||||
sourceDatabase: new URL(pair.sourceUrl).pathname.slice(1), restoreDatabase: new URL(pair.restoreUrl).pathname.slice(1) }, null, 2));
|
||||
} finally {
|
||||
await source?.$client.end({ timeout: 5 });
|
||||
if (directory && !process.env.PAPERCLIP_BACKUP_QUALIFICATION_RECEIPT) await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
async function emptyRestore() {
|
||||
const target = createDb(pair.restoreUrl, { maxConnections: 1 }).$client;
|
||||
try {
|
||||
const tables = await target`SELECT n.nspname, c.relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
|
||||
WHERE c.relkind IN ('r', 'p') AND n.nspname <> 'information_schema' AND n.nspname NOT LIKE 'pg_%'`;
|
||||
for (const { nspname, relname } of tables) await target`DROP TABLE IF EXISTS ${target(nspname)}.${target(relname)} CASCADE`;
|
||||
const schemas = await target`SELECT nspname FROM pg_namespace WHERE nspname NOT IN ('information_schema') AND nspname NOT LIKE 'pg_%'`;
|
||||
for (const { nspname } of schemas) await target`DROP SCHEMA ${target(nspname)} CASCADE`;
|
||||
await target`CREATE SCHEMA public`;
|
||||
} finally { await target.end({ timeout: 5 }); }
|
||||
}
|
||||
it("rejects the same actual database behind a hostname alias and nonempty targets", async () => {
|
||||
const alias = new URL(pair.sourceUrl); alias.hostname = alias.hostname === "localhost" ? "127.0.0.1" : "localhost";
|
||||
await expect(assertDistinctEmptyRestore({ ...pair, restoreUrl: alias.toString() })).rejects.toThrow("same actual database");
|
||||
const target = createDb(pair.restoreUrl, { maxConnections: 1 }).$client;
|
||||
try {
|
||||
for (const [create, drop] of [
|
||||
["CREATE TABLE guard_fixture (id integer)", "DROP TABLE guard_fixture"],
|
||||
["CREATE TYPE guard_fixture AS ENUM ('retained')", "DROP TYPE guard_fixture"],
|
||||
["CREATE FUNCTION guard_fixture() RETURNS integer LANGUAGE sql AS 'SELECT 1'", "DROP FUNCTION guard_fixture()"],
|
||||
]) {
|
||||
await target.unsafe(create!);
|
||||
try { await expect(assertDistinctEmptyRestore(pair)).rejects.toThrow("must be empty"); }
|
||||
finally { await target.unsafe(drop!); }
|
||||
}
|
||||
} finally { await target.end({ timeout: 5 }); }
|
||||
});
|
||||
|
||||
it.each(["pg_dump", "javascript"] as const)("restores the real %s archive with relationships, values and sequence progress", async engine => {
|
||||
if (engine === "javascript") await emptyRestore();
|
||||
const versions = Object.fromEntries([process.env.PAPERCLIP_PG_DUMP_PATH || "pg_dump", process.env.PAPERCLIP_PSQL_PATH || "psql"]
|
||||
.map(binary => [binary, execFileSync(binary, ["--version"], { encoding: "utf8" }).trim()]));
|
||||
const result = await qualifyBackupRestore(pair, directory, engine);
|
||||
expect(result.restored).toEqual({ relational: true, unicode: true, binary: true, emptyBinary: true, nullBinary: true, nullable: true, nextIdentity: "2", foreignKeyEnforced: true });
|
||||
expect(gunzipSync(await readFile(result.archive)).length).toBeGreaterThan(0);
|
||||
receipts.push({ ...result, versions });
|
||||
}, 120_000);
|
||||
|
||||
it("restores the real JavaScript fallback after pg_dump is unavailable", async () => {
|
||||
await emptyRestore();
|
||||
const previous = process.env.PAPERCLIP_PG_DUMP_PATH;
|
||||
process.env.PAPERCLIP_PG_DUMP_PATH = path.join(directory, "missing-pg-dump");
|
||||
try {
|
||||
const result = await qualifyBackupRestore(pair, directory, "auto");
|
||||
expect(result.restored.foreignKeyEnforced).toBe(true);
|
||||
receipts.push({ ...result, case: "automatic-javascript-fallback", pgDumpUnavailable: true });
|
||||
} finally {
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_PG_DUMP_PATH; else process.env.PAPERCLIP_PG_DUMP_PATH = previous;
|
||||
}
|
||||
}, 120_000);
|
||||
|
||||
it("permits sleep only after the production owner finishes a real checkpoint and restores its just-committed state", async () => {
|
||||
await emptyRestore();
|
||||
const clock = Date.now();
|
||||
const hold: IdleSleepDrainStatus = { ownerId: "checkpoint-owner", draining: true, startedAt: new Date(clock - 1), expiresAt: new Date(clock + 120_000), activeRuns: 0, pendingWakes: 0 };
|
||||
const run = owner(); let archive: string | undefined;
|
||||
expect(await readIdleSleepSafety(source, () => hold, () => Date.now(), hold.ownerId, async () => "none", undefined,
|
||||
async () => { const result = await run("idle"); archive = result!.backupFile; return Boolean(result); }))
|
||||
.toEqual({ version: 1, backgroundWork: "none" });
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
await assertDistinctEmptyRestore(pair);
|
||||
await runDatabaseRestore({ connectionString: pair.restoreUrl, backupFile: archive! });
|
||||
expect((await inspectRestoredFixture(pair.restoreUrl)).nextIdentity).toBe("2");
|
||||
receipts.push({ case: "real-owner-safety-and-restore", archive, markerRetained: true });
|
||||
}, 120_000);
|
||||
|
||||
it.each(["expiry", "owner", "generation"])("keeps a slow real dump active and fences a changed %s", async change => {
|
||||
let clock = Date.now();
|
||||
let hold: IdleSleepDrainStatus = { ownerId: "slow-owner", draining: true, startedAt: new Date(clock - 1), expiresAt: new Date(clock + 120_000), activeRuns: 0, pendingWakes: 0 };
|
||||
const barrier = path.join(directory, `slow-${change}.ready`), release = path.join(directory, `slow-${change}.release`);
|
||||
const wrapper = path.join(directory, `slow-${change}.mjs`);
|
||||
const previous = process.env.PAPERCLIP_PG_DUMP_PATH;
|
||||
await writeFile(wrapper, `#!${process.execPath}\nimport { spawnSync } from 'node:child_process';\nimport { existsSync, writeFileSync } from 'node:fs';\nconst result = spawnSync(${JSON.stringify(previous || "pg_dump")}, process.argv.slice(2), { stdio: ['ignore', 'inherit', 'inherit'] });\nif (result.status !== 0) process.exit(result.status ?? 1);\nwriteFileSync(${JSON.stringify(barrier)}, 'streamed');\nsetInterval(() => { if (existsSync(${JSON.stringify(release)})) process.exit(0); }, 10);\n`);
|
||||
await chmod(wrapper, 0o700); process.env.PAPERCLIP_PG_DUMP_PATH = wrapper;
|
||||
const run = owner();
|
||||
const safety = readIdleSleepSafety(source, () => hold, () => clock, "slow-owner", async () => "none", undefined, async () => Boolean(await run("idle")));
|
||||
let result;
|
||||
try {
|
||||
await vi.waitFor(() => expect(existsSync(barrier)).toBe(true), { timeout: 60_000 });
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
expect(await run("scheduled")).toBeNull();
|
||||
if (change === "expiry") clock += 120_000;
|
||||
if (change === "owner") hold = { ...hold, ownerId: "replacement" };
|
||||
if (change === "generation") beginIdleTrackedWork()();
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
} finally {
|
||||
await writeFile(release, "release owned dump");
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_PG_DUMP_PATH; else process.env.PAPERCLIP_PG_DUMP_PATH = previous;
|
||||
result = await safety;
|
||||
}
|
||||
expect(result).toEqual({ version: 1, backgroundWork: "unknown" });
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
receipts.push({ case: `slow-real-dump-${change}`, activeUntilSettled: true, sleepRefused: true });
|
||||
}, 120_000);
|
||||
|
||||
it.each(["auto", "javascript"] as const)("preserves prior recovery points after %s verification failure without fallback or pruning", async engine => {
|
||||
const first = await owner()("idle"); const firstBytes = await readFile(first!.backupFile);
|
||||
const history = path.join(directory, "paperclip-history.sql.gz"); const historyBytes = gzipSync("SELECT 'history';");
|
||||
await writeFile(history, historyBytes); await utimes(history, new Date(0), new Date(0));
|
||||
const failure = new Error("qualification fsync boundary failed"); let verifications = 0;
|
||||
await expect(runDatabaseBackup({ connectionString: pair.sourceUrl, backupDir: directory, retention, backupEngine: engine,
|
||||
verifyBeforePrune: async archive => { verifications++; await verifyIdleDatabaseBackup(archive); throw failure; } })).rejects.toBe(failure);
|
||||
expect(verifications).toBe(1);
|
||||
expect(await readFile(first!.backupFile)).toEqual(firstBytes);
|
||||
expect(await readFile(history)).toEqual(historyBytes);
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
receipts.push({ case: "verification-failure", previousArchive: first!.backupFile, previousArchiveRetained: true, engine, verificationCalls: verifications });
|
||||
}, 120_000);
|
||||
|
||||
it("retains an interrupted checkpoint, refuses a failed wake backup, then recovers through the same production owner", async () => {
|
||||
const first = await owner()("idle"); const firstBytes = await readFile(first!.backupFile);
|
||||
const barrier = path.join(directory, "dump-streamed");
|
||||
const wrapper = path.join(directory, "slow-pg-dump.mjs");
|
||||
const realPgDump = process.env.PAPERCLIP_PG_DUMP_PATH || "pg_dump";
|
||||
await writeFile(wrapper, `#!${process.execPath}\nimport { spawnSync } from 'node:child_process';\nimport { writeFileSync } from 'node:fs';\nconst result = spawnSync(${JSON.stringify(realPgDump)}, process.argv.slice(2), { stdio: ['ignore', 'inherit', 'inherit'] });\nif (result.status !== 0) process.exit(result.status ?? 1);\nwriteFileSync(${JSON.stringify(barrier)}, 'real pg_dump exited; pipeline remains open');\nsetInterval(() => {}, 1000);\n`);
|
||||
await chmod(wrapper, 0o700);
|
||||
const worker = fileURLToPath(new URL("./fixtures/database-backup-worker.ts", import.meta.url));
|
||||
const tsx = fileURLToPath(import.meta.resolve("tsx"));
|
||||
const child = spawn(process.execPath, ["--import", tsx, worker], { detached: true, stdio: ["ignore", "ignore", "pipe"],
|
||||
env: { ...process.env, PAPERCLIP_PG_DUMP_PATH: wrapper, PAPERCLIP_BACKUP_WORKER_DATABASE_URL: pair.sourceUrl, PAPERCLIP_BACKUP_WORKER_DIRECTORY: directory } });
|
||||
let stderr = ""; child.stderr!.on("data", chunk => { stderr += String(chunk); });
|
||||
const exited = new Promise<void>((resolve, reject) => { child.once("error", reject); child.once("exit", () => resolve()); });
|
||||
try {
|
||||
await vi.waitFor(() => { if (child.exitCode !== null) throw new Error(`Owned worker exited: ${stderr}`); expect(existsSync(barrier)).toBe(true); }, { timeout: 60_000 });
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
} finally {
|
||||
if (child.pid && child.exitCode === null && child.signalCode === null) {
|
||||
try { process.kill(-child.pid, "SIGKILL"); }
|
||||
catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; }
|
||||
}
|
||||
await exited;
|
||||
}
|
||||
expect(await readFile(first!.backupFile)).toEqual(firstBytes);
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
const unavailable = new URL(pair.sourceUrl); unavailable.port = "1";
|
||||
await expect(createDatabaseBackupOwner({ connectionString: unavailable.toString(), backupDir: directory, getRetention: async () => retention, logger })("scheduled")).rejects.toThrow();
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
await owner()("scheduled");
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(false);
|
||||
expect(await readFile(first!.backupFile)).toEqual(firstBytes);
|
||||
receipts.push({ case: "owned-child-interruption-wake-recovery", markerSurvived: true, priorArchiveRetained: true, wakeBackupVerified: true });
|
||||
}, 120_000);
|
||||
});
|
||||
@@ -0,0 +1,133 @@
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { gzipSync } from "node:zlib";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { runDatabaseBackup } from "@paperclipai/db";
|
||||
import { createDatabaseBackupOwner } from "../services/database-backup-owner.js";
|
||||
import { idleBackupWakeMarker, markIdleBackupWakeRequired } from "../services/idle-database-backup.js";
|
||||
import { beginIdleTrackedWork, idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { registerDatabaseBackupWork } from "../services/database-backup-work.js";
|
||||
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
|
||||
|
||||
vi.mock("@paperclipai/db", async importOriginal => ({ ...await importOriginal<typeof import("@paperclipai/db")>(), runDatabaseBackup: vi.fn() }));
|
||||
let directory: string;
|
||||
const policy = { dailyDays: 14, weeklyWeeks: 4, monthlyMonths: 6 };
|
||||
const logger = { info: vi.fn(), warn: vi.fn(), error: vi.fn() };
|
||||
function owner(getRetention = async () => policy) {
|
||||
return createDatabaseBackupOwner({ connectionString: "unused", backupDir: directory, getRetention, logger });
|
||||
}
|
||||
async function validArchive(options: Parameters<typeof runDatabaseBackup>[0]) {
|
||||
const backupFile = path.join(directory, "fixture.sql.gz");
|
||||
const bytes = gzipSync("CREATE TABLE fixture (id integer);\n");
|
||||
await writeFile(backupFile, bytes);
|
||||
await options.verifyBeforePrune?.(backupFile);
|
||||
return { backupFile, sizeBytes: bytes.length, prunedCount: 0 };
|
||||
}
|
||||
beforeEach(async () => {
|
||||
directory = await mkdtemp(path.join(os.tmpdir(), "backup-owner-"));
|
||||
vi.mocked(runDatabaseBackup).mockReset().mockImplementation(validArchive);
|
||||
});
|
||||
afterEach(async () => { await rm(directory, { recursive: true, force: true }); vi.restoreAllMocks(); });
|
||||
|
||||
describe("database backup owner", () => {
|
||||
it("marks before retention access and retains the marker until a verified wake backup", async () => {
|
||||
const run = owner(async () => {
|
||||
expect(await readFile(idleBackupWakeMarker(directory), "utf8")).toContain("restarts");
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
return policy;
|
||||
});
|
||||
const before = idleWorkSnapshot();
|
||||
const checkpoint = await run("idle");
|
||||
expect(checkpoint).toMatchObject({ trigger: "idle", retention: policy, backupDir: directory });
|
||||
expect(idleWorkSnapshot()).toEqual({ active: before.active, generation: before.generation + 2 });
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
await owner()("scheduled");
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(false);
|
||||
expect(runDatabaseBackup).toHaveBeenCalledWith(expect.objectContaining({ retention: policy, verifyBeforePrune: expect.any(Function) }));
|
||||
});
|
||||
|
||||
it("keeps one receipt and single-flight exclusion until the actual operation settles", async () => {
|
||||
let finish!: () => void;
|
||||
const gate = new Promise<void>(resolve => { finish = resolve; });
|
||||
vi.mocked(runDatabaseBackup).mockImplementationOnce(async options => { await gate; return validArchive(options); });
|
||||
const run = owner(); const backup = run("idle");
|
||||
await vi.waitFor(() => expect(runDatabaseBackup).toHaveBeenCalledOnce());
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
expect(await run("scheduled")).toBeNull();
|
||||
await expect(run("manual")).rejects.toMatchObject({ status: 409 });
|
||||
await expect(run("idle")).rejects.toMatchObject({ status: 409 });
|
||||
finish(); await backup;
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
|
||||
it.each(["dump", "integrity", "retention"])("retains the marker and releases the receipt after %s failure", async failure => {
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
if (failure === "dump") vi.mocked(runDatabaseBackup).mockRejectedValueOnce(new Error("dump failed"));
|
||||
if (failure === "integrity") vi.mocked(runDatabaseBackup).mockImplementationOnce(async options => {
|
||||
const backupFile = path.join(directory, "invalid.sql.gz"); await writeFile(backupFile, "corrupt");
|
||||
await options.verifyBeforePrune!(backupFile); throw new Error("unreachable");
|
||||
});
|
||||
const run = owner(async () => { if (failure === "retention") throw new Error("retention unavailable"); return policy; });
|
||||
await expect(run("scheduled")).rejects.toThrow();
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
await owner()("scheduled");
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(false);
|
||||
});
|
||||
|
||||
it.each(["expiry", "owner", "generation"])("rejects sleep after %s changes during the production owner's backup", async change => {
|
||||
let now = Date.now();
|
||||
let hold: IdleSleepDrainStatus = { ownerId: "owner", draining: true, startedAt: new Date(now - 1), expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 };
|
||||
let finish!: () => void; const gate = new Promise<void>(resolve => { finish = resolve; });
|
||||
vi.mocked(runDatabaseBackup).mockImplementationOnce(async options => { await gate; return validArchive(options); });
|
||||
const db = { transaction: async (fn: (tx: unknown) => unknown) => fn({ execute: async () => [{ blocked: false }] }) } as unknown as Db;
|
||||
const run = owner();
|
||||
const safety = readIdleSleepSafety(db, () => hold, () => now, "owner", async () => "none", undefined, async () => Boolean(await run("idle")));
|
||||
await vi.waitFor(() => expect(runDatabaseBackup).toHaveBeenCalledOnce());
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
if (change === "expiry") now += 60_000;
|
||||
if (change === "owner") hold = { ...hold, ownerId: "replacement" };
|
||||
if (change === "generation") beginIdleTrackedWork()();
|
||||
finish();
|
||||
expect(await safety).toEqual({ version: 1, backgroundWork: "unknown" });
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(true);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
|
||||
it("resumes the normal interval after a verified wake backup and retains a due slot through drain", async () => {
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
const run = owner(); await run("scheduled");
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(false);
|
||||
vi.useFakeTimers();
|
||||
const execute = vi.fn(async () => []);
|
||||
let canRun = true;
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: installDatabaseWorkSignals({ execute, transaction: vi.fn() }), canRun: () => canRun, onError: error => { throw error; } });
|
||||
try {
|
||||
await registerDatabaseBackupWork(coordinator, { intervalMs: 1000, run: () => run("scheduled") }).ready;
|
||||
await vi.advanceTimersByTimeAsync(999);
|
||||
expect(runDatabaseBackup).toHaveBeenCalledOnce();
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
canRun = false;
|
||||
await vi.advanceTimersByTimeAsync(3001);
|
||||
expect(runDatabaseBackup).toHaveBeenCalledOnce();
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
canRun = true;
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
|
||||
expect(runDatabaseBackup).toHaveBeenCalledTimes(2);
|
||||
expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now());
|
||||
} finally { await coordinator.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("fails closed if the storage path is no longer a directory", async () => {
|
||||
await rm(directory, { recursive: true }); await writeFile(directory, "mount unavailable");
|
||||
await expect(owner()("idle")).rejects.toThrow();
|
||||
expect(runDatabaseBackup).not.toHaveBeenCalled();
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,17 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { resolveBackupQualificationPair } from "../../scripts/lib/backup-qualification.js";
|
||||
|
||||
describe("backup qualification URL guard", () => {
|
||||
const source = "postgres://user:synthetic@localhost/source_test";
|
||||
const target = "postgres://user:synthetic@localhost/restore_test";
|
||||
it("accepts a complete distinct pair and leaves CI fallback explicit", () => {
|
||||
expect(resolveBackupQualificationPair("", "")).toBeUndefined();
|
||||
expect(resolveBackupQualificationPair(source, target)).toEqual({ sourceUrl: source, restoreUrl: target });
|
||||
});
|
||||
it.each([[source, ""], ["", target], [source, source], [source, source.replace("user:synthetic", "other:secret")],
|
||||
[source, source.replace("localhost", "LOCALHOST:5432")], [source, target.replace("restore_test", "production")],
|
||||
[source, `${target}?options=-csearch_path=public`]])("rejects unsafe configuration before opening clients", (a, b) => {
|
||||
expect(() => resolveBackupQualificationPair(a, b)).toThrow();
|
||||
try { resolveBackupQualificationPair(a, b); } catch (error) { expect(String(error)).not.toContain("synthetic"); }
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,8 @@
|
||||
import { createDatabaseBackupOwner } from "../../services/database-backup-owner.js";
|
||||
const run = createDatabaseBackupOwner({
|
||||
connectionString: process.env.PAPERCLIP_BACKUP_WORKER_DATABASE_URL!, backupDir: process.env.PAPERCLIP_BACKUP_WORKER_DIRECTORY!,
|
||||
getRetention: async () => ({ dailyDays: 14, weeklyWeeks: 4, monthlyMonths: 6 }),
|
||||
logger: { info() {}, warn() {}, error() {} },
|
||||
});
|
||||
await run("idle");
|
||||
throw new Error("Interrupted backup fixture unexpectedly completed");
|
||||
@@ -0,0 +1,48 @@
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { gzipSync } from "node:zlib";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "../services/idle-database-backup.js";
|
||||
|
||||
const syncs = vi.hoisted(() => ({ events: [] as string[], fail: "" }));
|
||||
vi.mock("node:fs/promises", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("node:fs/promises")>();
|
||||
return { ...actual, open: async (...args: Parameters<typeof actual.open>) => {
|
||||
const handle = await actual.open(...args);
|
||||
const name = (await handle.stat()).isDirectory() ? "directory" : path.basename(String(args[0]));
|
||||
const sync = handle.sync.bind(handle);
|
||||
handle.sync = async () => {
|
||||
syncs.events.push(name);
|
||||
if (syncs.fail === name) throw new Error(`injected ${name} fsync failure`);
|
||||
await sync();
|
||||
};
|
||||
return handle;
|
||||
} };
|
||||
});
|
||||
let directory: string;
|
||||
beforeEach(async () => { directory = await mkdtemp(path.join(os.tmpdir(), "backup-sync-")); syncs.events = []; syncs.fail = ""; });
|
||||
afterEach(async () => { await rm(directory, { recursive: true, force: true }); });
|
||||
describe("idle checkpoint persistence barriers", () => {
|
||||
it("syncs the marker, archive and directory entries before reporting completion", async () => {
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
expect(syncs.events).toEqual([".idle-backup-wake-required", "directory"]);
|
||||
const archive = path.join(directory, "checkpoint.sql.gz"); await writeFile(archive, gzipSync("SELECT 1;"));
|
||||
await verifyIdleDatabaseBackup(archive);
|
||||
expect(syncs.events.slice(-2)).toEqual(["checkpoint.sql.gz", "directory"]);
|
||||
await clearIdleBackupWakeRequired(directory);
|
||||
expect(syncs.events.at(-1)).toBe("directory");
|
||||
await expect(readFile(idleBackupWakeMarker(directory))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
it.each(["checkpoint.sql.gz", "directory"])("rejects %s fsync failure and retains the wake obligation", async boundary => {
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
const archive = path.join(directory, "checkpoint.sql.gz"); await writeFile(archive, gzipSync("SELECT 1;"));
|
||||
syncs.fail = boundary;
|
||||
await expect(verifyIdleDatabaseBackup(archive)).rejects.toThrow(`injected ${boundary} fsync failure`);
|
||||
expect(await readFile(idleBackupWakeMarker(directory), "utf8")).toContain("restarts");
|
||||
});
|
||||
it.each([".idle-backup-wake-required", "directory"])("rejects marker %s fsync failure", async boundary => {
|
||||
syncs.fail = boundary;
|
||||
await expect(markIdleBackupWakeRequired(directory)).rejects.toThrow(`injected ${boundary} fsync failure`);
|
||||
});
|
||||
});
|
||||
@@ -1,10 +1,8 @@
|
||||
import { mkdtemp, readFile, rm, symlink, utimes, writeFile } from "node:fs/promises";
|
||||
import { mkdtemp, readFile, rm, symlink, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { gzipSync } from "node:zlib";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { createDb, ensurePostgresDatabase, getEmbeddedPostgresTestSupport, runDatabaseBackup, runDatabaseRestore, startEmbeddedPostgresTestDatabase } from "@paperclipai/db";
|
||||
import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "../services/idle-database-backup.js";
|
||||
|
||||
let directory: string;
|
||||
@@ -48,45 +46,3 @@ describe("idle database backup persistence", () => {
|
||||
expect(await readFile(target)).toEqual(gzipSync("SELECT 1;"));
|
||||
});
|
||||
});
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("idle backup restore", () => {
|
||||
it("restores the last pre-sleep state into a separate database", async () => {
|
||||
const source = await startEmbeddedPostgresTestDatabase("idle-backup-restore-");
|
||||
const sourceDb = createDb(source.connectionString);
|
||||
const targetUrl = new URL(source.connectionString);
|
||||
targetUrl.pathname = "/idle_restore";
|
||||
try {
|
||||
await sourceDb.execute(sql`CREATE TABLE backup_fixture (id integer PRIMARY KEY, value text NOT NULL)`);
|
||||
await sourceDb.execute(sql`INSERT INTO backup_fixture VALUES (1, 'last committed state')`);
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
const result = await runDatabaseBackup({
|
||||
connectionString: source.connectionString, backupDir: directory,
|
||||
retention: { dailyDays: 14, weeklyWeeks: 4, monthlyMonths: 6 },
|
||||
verifyBeforePrune: verifyIdleDatabaseBackup,
|
||||
});
|
||||
await verifyIdleDatabaseBackup(result.backupFile);
|
||||
await ensurePostgresDatabase(source.connectionString, "idle_restore");
|
||||
await runDatabaseRestore({ connectionString: targetUrl.toString(), backupFile: result.backupFile });
|
||||
const target = createDb(targetUrl.toString());
|
||||
expect(await target.execute(sql`SELECT id, value FROM backup_fixture`)).toEqual([{ id: 1, value: "last committed state" }]);
|
||||
const checkpointBytes = await readFile(result.backupFile);
|
||||
expect(await readFile(idleBackupWakeMarker(directory), "utf8")).toContain("restarts");
|
||||
// A failed verification must not delete existing recovery points, even
|
||||
// when their age would make them eligible for the configured pruning.
|
||||
const history = path.join(directory, "paperclip-history.sql.gz");
|
||||
const historyBytes = gzipSync("SELECT 'historical recovery point';");
|
||||
await writeFile(history, historyBytes);
|
||||
await utimes(history, new Date(0), new Date(0));
|
||||
await expect(runDatabaseBackup({
|
||||
connectionString: source.connectionString, backupDir: directory,
|
||||
backupEngine: "javascript", retention: { dailyDays: 7, weeklyWeeks: 4, monthlyMonths: 1 },
|
||||
verifyBeforePrune: async () => { throw new Error("verification failed"); },
|
||||
})).rejects.toThrow("verification failed");
|
||||
expect(await readFile(history)).toEqual(historyBytes);
|
||||
expect(await readFile(result.backupFile)).toEqual(checkpointBytes);
|
||||
} finally {
|
||||
await source.cleanup();
|
||||
}
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -1232,8 +1232,8 @@ describe("startServer feedback export wiring", () => {
|
||||
databaseBackupService: { runManualBackup(): Promise<unknown> };
|
||||
};
|
||||
const backup = options.databaseBackupService.runManualBackup();
|
||||
await Promise.resolve();
|
||||
expect(idleWorkSnapshot().active).toBe(before + 1);
|
||||
await vi.waitFor(() => expect(resolveBackup).toBeTypeOf("function"));
|
||||
if (fails) {
|
||||
const rejection = expect(backup).rejects.toThrow("backup failed");
|
||||
rejectBackup(new Error("backup failed"));
|
||||
|
||||
+8
-74
@@ -61,7 +61,8 @@ import { deploymentNotifications } from "./services/deployment-notifications.js"
|
||||
import { registerTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./services/terminal-workspace-work.js";
|
||||
import { isIdleTaskDrainActive, readTaskDrain, beginIdleTrackedWork, trackIdleWork, registerTaskDrainExpiryWork } from "./services/task-admission.js";
|
||||
import { markIdleStartupComplete, registerIdleLocalWorkProbe } from "./services/idle-local-work.js";
|
||||
import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "./services/idle-database-backup.js";
|
||||
import { idleBackupWakeMarker } from "./services/idle-database-backup.js";
|
||||
import { createDatabaseBackupOwner } from "./services/database-backup-owner.js";
|
||||
import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js";
|
||||
import { createCloudWarmStandby } from "./services/cloud-warm-standby.js";
|
||||
import { chatCompletionDeliveryService } from "./services/chat-completion-delivery.js";
|
||||
@@ -103,8 +104,6 @@ import {
|
||||
createEmbeddedPostgresLogBuffer,
|
||||
prepareEmbeddedPostgresNativeRuntime,
|
||||
reconcilePendingMigrationHistory,
|
||||
formatDatabaseBackupResult,
|
||||
runDatabaseBackup,
|
||||
authUsers,
|
||||
companies,
|
||||
companyMemberships,
|
||||
@@ -206,10 +205,6 @@ import {
|
||||
type EmbeddedPostgresSupervisor,
|
||||
type SupervisedEmbeddedPostgres,
|
||||
} from "./embedded-postgres-supervisor.js";
|
||||
import type {
|
||||
InstanceDatabaseBackupRunResult,
|
||||
InstanceDatabaseBackupTrigger,
|
||||
} from "./routes/instance-database-backups.js";
|
||||
|
||||
type BetterAuthSessionUser = {
|
||||
id: string;
|
||||
@@ -912,75 +907,14 @@ async function startServerWithDatabaseTeardown(
|
||||
resolve(config.databaseBackupDir, "db-backup-to-s3.failure"),
|
||||
resolve(config.databaseBackupDir, "..", "db-backup-to-s3.failure"),
|
||||
];
|
||||
let databaseBackupInFlight = false;
|
||||
const idleDatabaseBackupEnabled = config.databaseBackupEnabled
|
||||
&& process.env.PAPERCLIP_DB_BACKUP_IDLE_CHECKPOINT_ENABLED === "1";
|
||||
const runServerDatabaseBackup = async (
|
||||
trigger: InstanceDatabaseBackupTrigger,
|
||||
): Promise<InstanceDatabaseBackupRunResult | null> => {
|
||||
if (databaseBackupInFlight) {
|
||||
const message = "Database backup already in progress";
|
||||
if (trigger === "scheduled") {
|
||||
logger.warn("Skipping scheduled database backup because a previous backup is still running");
|
||||
return null;
|
||||
}
|
||||
throw conflict(message);
|
||||
}
|
||||
|
||||
databaseBackupInFlight = true;
|
||||
const finishIdleBackup = beginIdleTrackedWork();
|
||||
const startedAt = new Date();
|
||||
const startedAtMs = Date.now();
|
||||
const label = trigger === "manual" ? "Manual" : trigger === "idle" ? "Pre-sleep" : "Automatic";
|
||||
try {
|
||||
if (trigger === "idle") await markIdleBackupWakeRequired(config.databaseBackupDir);
|
||||
logger.info({ backupDir: config.databaseBackupDir, trigger }, `${label} database backup starting`);
|
||||
// Read retention from Instance Settings (DB) so changes take effect without restart.
|
||||
const generalSettings = await backupSettingsSvc.getGeneral();
|
||||
const retention = generalSettings.backupRetention;
|
||||
const verifyCheckpoint = trigger === "idle" || existsSync(idleBackupWakeMarker(config.databaseBackupDir));
|
||||
|
||||
const result = await runDatabaseBackup({
|
||||
connectionString: activeDatabaseConnectionString,
|
||||
backupDir: config.databaseBackupDir,
|
||||
retention,
|
||||
filenamePrefix: "paperclip",
|
||||
verifyBeforePrune: verifyCheckpoint ? verifyIdleDatabaseBackup : undefined,
|
||||
});
|
||||
if (verifyCheckpoint) {
|
||||
if (trigger !== "idle") await clearIdleBackupWakeRequired(config.databaseBackupDir);
|
||||
}
|
||||
const finishedAt = new Date();
|
||||
const response: InstanceDatabaseBackupRunResult = {
|
||||
...result,
|
||||
trigger,
|
||||
backupDir: config.databaseBackupDir,
|
||||
retention,
|
||||
startedAt: startedAt.toISOString(),
|
||||
finishedAt: finishedAt.toISOString(),
|
||||
durationMs: Date.now() - startedAtMs,
|
||||
};
|
||||
logger.info(
|
||||
{
|
||||
backupFile: result.backupFile,
|
||||
sizeBytes: result.sizeBytes,
|
||||
prunedCount: result.prunedCount,
|
||||
backupDir: config.databaseBackupDir,
|
||||
retention,
|
||||
trigger,
|
||||
durationMs: response.durationMs,
|
||||
},
|
||||
`${label} database backup complete: ${formatDatabaseBackupResult(result)}`,
|
||||
);
|
||||
return response;
|
||||
} catch (err) {
|
||||
logger.error({ err, backupDir: config.databaseBackupDir, trigger }, `${label} database backup failed`);
|
||||
throw err;
|
||||
} finally {
|
||||
databaseBackupInFlight = false;
|
||||
finishIdleBackup();
|
||||
}
|
||||
};
|
||||
const runServerDatabaseBackup = createDatabaseBackupOwner({
|
||||
connectionString: activeDatabaseConnectionString,
|
||||
backupDir: config.databaseBackupDir,
|
||||
getRetention: async () => (await backupSettingsSvc.getGeneral()).backupRetention,
|
||||
logger,
|
||||
});
|
||||
const pluginWorkerManager = createPluginWorkerManager();
|
||||
const heartbeat = config.heartbeatSchedulerEnabled
|
||||
? heartbeatService(db as any, { pluginWorkerManager })
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
import { existsSync } from "node:fs";
|
||||
import { formatDatabaseBackupResult, runDatabaseBackup, type BackupRetentionPolicy } from "@paperclipai/db";
|
||||
import { conflict } from "../errors.js";
|
||||
import type { InstanceDatabaseBackupRunResult, InstanceDatabaseBackupTrigger } from "../routes/instance-database-backups.js";
|
||||
import { beginIdleTrackedWork } from "./task-admission.js";
|
||||
import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "./idle-database-backup.js";
|
||||
|
||||
type BackupLogger = {
|
||||
info: (details: object, message: string) => void;
|
||||
warn: (message: string) => void;
|
||||
error: (details: object, message: string) => void;
|
||||
};
|
||||
|
||||
export function createDatabaseBackupOwner(options: {
|
||||
connectionString: string;
|
||||
backupDir: string;
|
||||
getRetention: () => Promise<BackupRetentionPolicy>;
|
||||
logger: BackupLogger;
|
||||
}) {
|
||||
let databaseBackupInFlight = false;
|
||||
return async (
|
||||
trigger: InstanceDatabaseBackupTrigger,
|
||||
): Promise<InstanceDatabaseBackupRunResult | null> => {
|
||||
if (databaseBackupInFlight) {
|
||||
const message = "Database backup already in progress";
|
||||
if (trigger === "scheduled") {
|
||||
options.logger.warn("Skipping scheduled database backup because a previous backup is still running");
|
||||
return null;
|
||||
}
|
||||
throw conflict(message);
|
||||
}
|
||||
|
||||
databaseBackupInFlight = true;
|
||||
const finishIdleBackup = beginIdleTrackedWork();
|
||||
const startedAt = new Date();
|
||||
const startedAtMs = Date.now();
|
||||
const label = trigger === "manual" ? "Manual" : trigger === "idle" ? "Pre-sleep" : "Automatic";
|
||||
try {
|
||||
if (trigger === "idle") await markIdleBackupWakeRequired(options.backupDir);
|
||||
options.logger.info({ backupDir: options.backupDir, trigger }, `${label} database backup starting`);
|
||||
const retention = await options.getRetention();
|
||||
const verifyCheckpoint = trigger === "idle" || existsSync(idleBackupWakeMarker(options.backupDir));
|
||||
|
||||
const result = await runDatabaseBackup({
|
||||
connectionString: options.connectionString,
|
||||
backupDir: options.backupDir,
|
||||
retention,
|
||||
filenamePrefix: "paperclip",
|
||||
verifyBeforePrune: verifyCheckpoint ? verifyIdleDatabaseBackup : undefined,
|
||||
});
|
||||
if (verifyCheckpoint) {
|
||||
if (trigger !== "idle") await clearIdleBackupWakeRequired(options.backupDir);
|
||||
}
|
||||
const finishedAt = new Date();
|
||||
const response: InstanceDatabaseBackupRunResult = {
|
||||
...result,
|
||||
trigger,
|
||||
backupDir: options.backupDir,
|
||||
retention,
|
||||
startedAt: startedAt.toISOString(),
|
||||
finishedAt: finishedAt.toISOString(),
|
||||
durationMs: Date.now() - startedAtMs,
|
||||
};
|
||||
options.logger.info(
|
||||
{
|
||||
backupFile: result.backupFile,
|
||||
sizeBytes: result.sizeBytes,
|
||||
prunedCount: result.prunedCount,
|
||||
backupDir: options.backupDir,
|
||||
retention,
|
||||
trigger,
|
||||
durationMs: response.durationMs,
|
||||
},
|
||||
`${label} database backup complete: ${formatDatabaseBackupResult(result)}`,
|
||||
);
|
||||
return response;
|
||||
} catch (err) {
|
||||
options.logger.error({ err, backupDir: options.backupDir, trigger }, `${label} database backup failed`);
|
||||
throw err;
|
||||
} finally {
|
||||
databaseBackupInFlight = false;
|
||||
finishIdleBackup();
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
import { defineConfig } from "vitest/config";
|
||||
import base from "./vitest.config.js";
|
||||
export default defineConfig({
|
||||
...base,
|
||||
test: {
|
||||
...base.test,
|
||||
include: ["src/__tests__/database-backup-checkpoint-qualification.integration.ts"],
|
||||
fileParallelism: false,
|
||||
maxWorkers: 1,
|
||||
},
|
||||
});
|
||||
Reference in new issue
Block a user