From 133412f3d18d6c88b467db705fafa4bec261acea Mon Sep 17 00:00:00 2001 From: Andrew Levine Date: Fri, 11 Sep 2026 13:58:05 -0400 Subject: [PATCH 1/4] fix(SUP-3933): add producer storage transaction lock for backups Serialize runDatabaseBackup against RAID archival with the shared paperclip.storage-transaction-lock/v1 wire contract (.paperclip-stx-*). Acquire before backup/prune filesystem work, reverify before retention unlinks, and release with a ready participation marker. Aligns on-disk owner/marker shape with clawd so both sides mutually exclude. --- packages/db/src/backup-lib.ts | 54 +- .../db/src/backup-transaction-lock.test.ts | 198 +++++ packages/db/src/backup-transaction-lock.ts | 679 ++++++++++++++++++ packages/db/src/index.ts | 29 + 4 files changed, 956 insertions(+), 4 deletions(-) create mode 100644 packages/db/src/backup-transaction-lock.test.ts create mode 100644 packages/db/src/backup-transaction-lock.ts diff --git a/packages/db/src/backup-lib.ts b/packages/db/src/backup-lib.ts index f066f22e23..16dd50063f 100644 --- a/packages/db/src/backup-lib.ts +++ b/packages/db/src/backup-lib.ts @@ -1,5 +1,13 @@ import { createReadStream, createWriteStream, existsSync, mkdirSync, readdirSync, statSync, unlinkSync } from "node:fs"; import { basename, resolve } from "node:path"; +import { + acquireStorageTransactionLock, + assertStorageTransactionLockHeld, + isStorageTransactionReservedName, + releaseStorageTransactionLock, + type StorageTransactionLockHandle, + StorageTransactionLockError, +} from "./backup-transaction-lock.js"; import { createInterface } from "node:readline"; import { spawn } from "node:child_process"; import { open as openFile } from "node:fs/promises"; @@ -34,6 +42,8 @@ export type RunDatabaseBackupResult = { backupFile: string; sizeBytes: number; prunedCount: number; + /** Present when the backup transaction lock was acquired for this run. */ + lockTokenPrefix?: string; }; export type RunDatabaseRestoreOptions = { @@ -120,7 +130,12 @@ function monthlyRetentionCutoff(nowMs: number, monthlyMonths: number): number { * - Monthly tier: keep the NEWEST backup per calendar month for `monthlyMonths` months * - Everything else is deleted */ -function pruneOldBackups(backupDir: string, retention: BackupRetentionPolicy, filenamePrefix: string): number { +function pruneOldBackups( + backupDir: string, + retention: BackupRetentionPolicy, + filenamePrefix: string, + lockHandle?: StorageTransactionLockHandle, +): number { if (!existsSync(backupDir)) return 0; const now = Date.now(); @@ -132,10 +147,13 @@ function pruneOldBackups(backupDir: string, retention: BackupRetentionPolicy, fi const entries: BackupEntry[] = []; for (const name of readdirSync(backupDir)) { + // Never treat protocol artifacts as backup candidates. + if (isStorageTransactionReservedName(name)) continue; if (!name.startsWith(`${filenamePrefix}-`)) continue; if (!name.endsWith(".sql") && !name.endsWith(".sql.gz")) continue; const fullPath = resolve(backupDir, name); const stat = statSync(fullPath); + if (!stat.isFile()) continue; entries.push({ name, fullPath, mtimeMs: stat.mtimeMs }); } @@ -179,6 +197,10 @@ function pruneOldBackups(backupDir: string, retention: BackupRetentionPolicy, fi } for (const filePath of toDelete) { + // Reverify producer lock ownership before every irreversible retention unlink. + if (lockHandle) { + assertStorageTransactionLockHeld(lockHandle); + } unlinkSync(filePath); } @@ -533,6 +555,22 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise const canUsePgDump = !hasBackupTransforms(opts); const excludedTableNames = normalizeTableNameSet(opts.excludeTables); const nullifiedColumnsByTable = normalizeNullifyColumnMap(opts.nullifyColumns); + + // Acquire the source-root transaction lock before any backup/prune filesystem activity. + mkdirSync(opts.backupDir, { recursive: true }); + let lockHandle: StorageTransactionLockHandle; + try { + lockHandle = acquireStorageTransactionLock({ + sourceRoot: opts.backupDir, + operationKind: "backup", + }); + } catch (error) { + if (error instanceof StorageTransactionLockError) { + throw error; + } + throw error; + } + let sql = postgres(opts.connectionString, { max: 1, connect_timeout: connectTimeout }); let sqlClosed = false; const closeSql = async () => { @@ -540,7 +578,6 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise sqlClosed = true; await sql.end(); }; - mkdirSync(opts.backupDir, { recursive: true }); const sqlFile = resolve(opts.backupDir, `${filenamePrefix}-${timestamp()}.sql`); const backupFile = `${sqlFile}.gz`; const writer = createBufferedTextFileWriter(sqlFile); @@ -557,11 +594,13 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise }); await writer.abort(); const sizeBytes = statSync(backupFile).size; - const prunedCount = pruneOldBackups(opts.backupDir, retention, filenamePrefix); + assertStorageTransactionLockHeld(lockHandle); + const prunedCount = pruneOldBackups(opts.backupDir, retention, filenamePrefix, lockHandle); return { backupFile, sizeBytes, prunedCount, + lockTokenPrefix: lockHandle.token.slice(0, 8), }; } catch (error) { if (existsSync(backupFile)) { @@ -1026,13 +1065,15 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise await pipeline(sqlReadStream, createGzip(), gzWriteStream); unlinkSync(sqlFile); + assertStorageTransactionLockHeld(lockHandle); const sizeBytes = statSync(backupFile).size; - const prunedCount = pruneOldBackups(opts.backupDir, retention, filenamePrefix); + const prunedCount = pruneOldBackups(opts.backupDir, retention, filenamePrefix, lockHandle); return { backupFile, sizeBytes, prunedCount, + lockTokenPrefix: lockHandle.token.slice(0, 8), }; } catch (error) { await writer.abort(); @@ -1045,6 +1086,11 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise throw error; } finally { await closeSql(); + try { + releaseStorageTransactionLock(lockHandle, { participationState: "ready" }); + } catch { + // Release is best-effort on teardown; token mismatch means another owner already replaced us. + } } } diff --git a/packages/db/src/backup-transaction-lock.test.ts b/packages/db/src/backup-transaction-lock.test.ts new file mode 100644 index 0000000000..b2feaa476d --- /dev/null +++ b/packages/db/src/backup-transaction-lock.test.ts @@ -0,0 +1,198 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + STORAGE_TRANSACTION_LOCK_DIR_NAME, + STORAGE_TRANSACTION_LOCK_SCHEMA, + STORAGE_TRANSACTION_PARTICIPATION_NAME, + STORAGE_TRANSACTION_RESERVED_PREFIX, + StorageTransactionLockError, + acquireStorageTransactionLock, + assertStorageTransactionLockHeld, + isStorageTransactionReservedName, + parseStorageTransactionOwnerRecord, + readLocalBootId, + readLocalMachineId, + readProcessStartId, + readStorageTransactionParticipation, + releaseStorageTransactionLock, + resolveSourceRootRealPath, +} from "./backup-transaction-lock.js"; + +const cleanupRoots: string[] = []; + +afterEach(() => { + while (cleanupRoots.length > 0) { + fs.rmSync(cleanupRoots.pop()!, { recursive: true, force: true }); + } +}); + +function tempDir(prefix: string): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + cleanupRoots.push(dir); + return dir; +} + +function writeOwner(root: string, overrides: Record = {}) { + const realRoot = resolveSourceRootRealPath(root); + const ownerPath = path.join(realRoot, STORAGE_TRANSACTION_LOCK_DIR_NAME); + const owner = { + schema: STORAGE_TRANSACTION_LOCK_SCHEMA, + token: "FAKESECRET_g1h2i3j4k5l6m7n8o9p0", + machineId: readLocalMachineId(), + bootId: readLocalBootId(), + pid: 2_147_483_646, + processStartId: "dead-process-start-id", + operationKind: "backup", + acquiredAt: new Date(Date.now() - 60_000).toISOString(), + ...overrides, + }; + fs.writeFileSync(ownerPath, `${JSON.stringify(owner)}\n`, { mode: 0o600 }); + return owner; +} + +describe("paperclip.storage-transaction-lock/v1", () => { + it("acquires, holds, and token-safely releases a source-root lock", () => { + const root = tempDir("paperclip-tx-lock-basic-"); + const handle = acquireStorageTransactionLock({ + sourceRoot: root, + operationKind: "backup", + }); + + expect(handle.token.length).toBeGreaterThanOrEqual(16); + expect(fs.existsSync(handle.ownerPath)).toBe(true); + expect(path.basename(handle.ownerPath)).toBe(".paperclip-stx-owner"); + expect(path.basename(handle.participationPath)).toBe(".paperclip-stx-marker"); + assertStorageTransactionLockHeld(handle); + + const participation = readStorageTransactionParticipation(root); + expect(participation?.state).toBe("active"); + expect(participation?.schema).toBe(STORAGE_TRANSACTION_LOCK_SCHEMA); + + releaseStorageTransactionLock(handle); + expect(fs.existsSync(handle.ownerPath)).toBe(false); + expect(readStorageTransactionParticipation(root)?.state).toBe("ready"); + }); + + it("returns busy when a live local owner already holds the lock", () => { + const root = tempDir("paperclip-tx-lock-busy-"); + const first = acquireStorageTransactionLock({ + sourceRoot: root, + operationKind: "backup", + }); + + try { + expect(() => + acquireStorageTransactionLock({ + sourceRoot: root, + operationKind: "raid-archive", + }), + ).toThrow(StorageTransactionLockError); + + try { + acquireStorageTransactionLock({ + sourceRoot: root, + operationKind: "raid-archive", + }); + expect.unreachable("second acquire should fail"); + } catch (error) { + expect(error).toBeInstanceOf(StorageTransactionLockError); + expect((error as StorageTransactionLockError).code).toBe("busy"); + } + } finally { + releaseStorageTransactionLock(first); + } + }); + + it("emits a clawd-compatible owner and participation fixture shape", () => { + const root = tempDir("paperclip-tx-lock-fixture-"); + const handle = acquireStorageTransactionLock({ + sourceRoot: root, + operationKind: "backup", + }); + const raw = JSON.parse(fs.readFileSync(handle.ownerPath, "utf8")); + const parsed = parseStorageTransactionOwnerRecord(raw); + expect(parsed).not.toBeNull(); + expect(parsed!.schema).toBe(STORAGE_TRANSACTION_LOCK_SCHEMA); + expect(parsed!.token).toBe(handle.token); + expect(Object.keys(parsed!).sort()).toEqual([ + "acquiredAt", + "bootId", + "machineId", + "operationKind", + "pid", + "processStartId", + "schema", + "token", + ]); + expect(Object.keys(readStorageTransactionParticipation(root)!).sort()).toEqual([ + "schema", + "state", + "token", + "updatedAt", + ]); + releaseStorageTransactionLock(handle); + }); + + it("recovers a confirmed dead local owner exactly once under the recovery mutex", () => { + const root = tempDir("paperclip-tx-lock-dead-"); + writeOwner(root, { + machineId: readLocalMachineId(), + bootId: readLocalBootId(), + pid: 2_147_483_646, + processStartId: "dead-process-start-id", + }); + + const recovered = acquireStorageTransactionLock({ + sourceRoot: root, + operationKind: "backup", + allowDeadOwnerRecovery: true, + }); + expect(recovered.token).not.toBe("deadtoken0123456789abcdef012345"); + expect(recovered.owner.bootId).toBe(readLocalBootId()); + expect(recovered.owner.processStartId).toBe(readProcessStartId()); + assertStorageTransactionLockHeld(recovered); + releaseStorageTransactionLock(recovered); + }); + + it("refuses a foreign-machine owner without recovery", () => { + const root = tempDir("paperclip-tx-lock-foreign-"); + writeOwner(root, { machineId: "foreign-machine-id-not-local" }); + try { + acquireStorageTransactionLock({ sourceRoot: root, operationKind: "backup" }); + expect.unreachable("foreign owner should block"); + } catch (error) { + expect(error).toBeInstanceOf(StorageTransactionLockError); + expect((error as StorageTransactionLockError).code).toBe("foreign_owner"); + } + }); + + it("refuses to release when the caller token no longer matches", () => { + const root = tempDir("paperclip-tx-lock-token-"); + const first = acquireStorageTransactionLock({ + sourceRoot: root, + operationKind: "backup", + }); + const secondRoot = tempDir("paperclip-tx-lock-token-2-"); + // Replace owner under recovery by releasing and reacquiring, then forge handle. + const handle = acquireStorageTransactionLock({ + sourceRoot: secondRoot, + operationKind: "backup", + }); + const forged = { + ...handle, + token: "FAKESECRET_g2h3i4j5k6l7m8n9o0p1", + }; + expect(() => releaseStorageTransactionLock(forged)).toThrow(StorageTransactionLockError); + releaseStorageTransactionLock(handle); + releaseStorageTransactionLock(first); + }); + + it("excludes reserved protocol artifact names", () => { + expect(isStorageTransactionReservedName(STORAGE_TRANSACTION_LOCK_DIR_NAME)).toBe(true); + expect(isStorageTransactionReservedName(STORAGE_TRANSACTION_PARTICIPATION_NAME)).toBe(true); + expect(isStorageTransactionReservedName(`${STORAGE_TRANSACTION_RESERVED_PREFIX}owner.tmp.x`)).toBe(true); + expect(isStorageTransactionReservedName("paperclip-20260911-120000.sql.gz")).toBe(false); + }); +}); diff --git a/packages/db/src/backup-transaction-lock.ts b/packages/db/src/backup-transaction-lock.ts new file mode 100644 index 0000000000..2370463759 --- /dev/null +++ b/packages/db/src/backup-transaction-lock.ts @@ -0,0 +1,679 @@ +import { + closeSync, + existsSync, + fsyncSync, + linkSync, + lstatSync, + mkdirSync, + openSync, + readFileSync, + realpathSync, + renameSync, + unlinkSync, + writeFileSync, + writeSync, +} from "node:fs"; +import { randomUUID } from "node:crypto"; +import { join, resolve, sep } from "node:path"; +import { spawnSync } from "node:child_process"; + +/** Wire schema shared with clawd (byte-compatible; no cross-repo imports). */ +export const STORAGE_TRANSACTION_LOCK_SCHEMA = "paperclip.storage-transaction-lock/v1"; + +/** Reserved artifact prefix under a locked source root. All enumerators must exclude it. */ +export const STORAGE_TRANSACTION_RESERVED_PREFIX = ".paperclip-stx-"; +export const STORAGE_TRANSACTION_LOCK_DIR_NAME = `${STORAGE_TRANSACTION_RESERVED_PREFIX}owner`; +export const STORAGE_TRANSACTION_RECOVERY_MUTEX_NAME = `${STORAGE_TRANSACTION_RESERVED_PREFIX}recovery`; +export const STORAGE_TRANSACTION_PARTICIPATION_NAME = `${STORAGE_TRANSACTION_RESERVED_PREFIX}marker`; +export const STORAGE_TRANSACTION_TMP_PREFIX = STORAGE_TRANSACTION_RESERVED_PREFIX; +export const STORAGE_TRANSACTION_QUARANTINE_PREFIX = `${STORAGE_TRANSACTION_RESERVED_PREFIX}quarantine.`; + +const MAX_RECORD_BYTES = 64 * 1024; + +export type StorageTransactionOperationKind = + | "backup" + | "backup-prune" + | "raid-archive" + | "storage-transaction-recovery" + | "recovery"; + +export type StorageTransactionParticipationState = "active" | "ready"; + +export type StorageTransactionOwnerRecord = { + schema: typeof STORAGE_TRANSACTION_LOCK_SCHEMA; + token: string; + machineId: string; + bootId: string; + pid: number; + processStartId: string; + operationKind: string; + acquiredAt: string; +}; + +export type StorageTransactionParticipationRecord = { + schema: typeof STORAGE_TRANSACTION_LOCK_SCHEMA; + state: StorageTransactionParticipationState; + token: string; + updatedAt: string; +}; + +export type StorageTransactionLockHandle = { + sourceRootRealPath: string; + ownerPath: string; + recoveryMutexPath: string; + participationPath: string; + token: string; + owner: StorageTransactionOwnerRecord; + recoveries?: Array>; +}; + +export type AcquireStorageTransactionLockOptions = { + sourceRoot: string; + operationKind: StorageTransactionOperationKind | string; + allowDeadOwnerRecovery?: boolean; + nowMs?: number; +}; + +export class StorageTransactionLockError extends Error { + readonly code: string; + readonly details?: Record; + + constructor(code: string, message: string, details?: Record) { + super(message); + this.name = "StorageTransactionLockError"; + this.code = code; + this.details = details; + } +} + +function isErrno(error: unknown, code: string): boolean { + return Boolean(error && typeof error === "object" && (error as NodeJS.ErrnoException).code === code); +} + +function safeUnlink(filePath: string): void { + try { + unlinkSync(filePath); + } catch (error) { + if (!isErrno(error, "ENOENT")) throw error; + } +} + +function shellText(command: string, args: string[]): string | null { + const result = spawnSync(command, args, { + encoding: "utf8", + timeout: 2_000, + stdio: ["ignore", "pipe", "ignore"], + env: process.env, + }); + if (result.status !== 0) return null; + const out = String(result.stdout ?? "").trim(); + return out.length > 0 ? out : null; +} + +export function readLocalMachineId(): string | null { + for (const candidate of ["/etc/machine-id", "/var/lib/dbus/machine-id"]) { + try { + const value = readFileSync(candidate, "utf8").trim(); + if (value) return value; + } catch { + // try next + } + } + if (process.platform === "darwin") { + const output = shellText("ioreg", ["-rd1", "-c", "IOPlatformExpertDevice"]); + const matched = output && output.match(/"IOPlatformUUID"\s*=\s*"([^"]+)"/); + if (matched) return matched[1]!; + } + return null; +} + +export function readLocalBootId(): string | null { + if (process.platform === "linux") { + try { + return readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim() || null; + } catch { + return null; + } + } + if (process.platform === "darwin") return shellText("sysctl", ["-n", "kern.boottime"]); + return null; +} + +export function readProcessStartId(pid: number = process.pid): string | null { + if (process.platform === "linux") { + try { + const raw = readFileSync(`/proc/${pid}/stat`, "utf8"); + const closeParen = raw.lastIndexOf(")"); + if (closeParen < 0) return null; + return raw.slice(closeParen + 2).trim().split(/\s+/)[19] || null; + } catch { + return null; + } + } + if (process.platform === "darwin") return shellText("ps", ["-p", String(pid), "-o", "lstart="]); + return null; +} + +export function resolveStorageTransactionIdentity(): { + machineId: string; + bootId: string; + pid: number; + processStartId: string; +} { + const machineId = readLocalMachineId(); + const bootId = readLocalBootId(); + const processStartId = readProcessStartId(); + if (!machineId || !bootId || !processStartId) { + throw new StorageTransactionLockError( + "unreadable", + "Cannot establish stable machine, boot, and process-start identity", + ); + } + return { machineId, bootId, pid: process.pid, processStartId }; +} + +export function resolveSourceRootRealPath( + sourceRoot: string, + options: { createIfMissing?: boolean } = {}, +): string { + const createIfMissing = options.createIfMissing !== false; + const resolved = resolve(sourceRoot); + if (createIfMissing) mkdirSync(resolved, { recursive: true }); + try { + return realpathSync(resolved); + } catch (error) { + if (!createIfMissing && isErrno(error, "ENOENT")) return resolved; + throw error; + } +} + +function ownerPaths(sourceRootRealPath: string) { + return { + sourceRootRealPath, + ownerPath: join(sourceRootRealPath, STORAGE_TRANSACTION_LOCK_DIR_NAME), + recoveryMutexPath: join(sourceRootRealPath, STORAGE_TRANSACTION_RECOVERY_MUTEX_NAME), + participationPath: join(sourceRootRealPath, STORAGE_TRANSACTION_PARTICIPATION_NAME), + }; +} + +function readRegularFile(filePath: string): string | null { + let stat; + try { + stat = lstatSync(filePath); + } catch (error) { + if (isErrno(error, "ENOENT")) return null; + if (isErrno(error, "EPERM") || isErrno(error, "EACCES")) { + throw new StorageTransactionLockError("eperm", "Cannot inspect storage transaction artifact", { filePath }); + } + throw new StorageTransactionLockError("unreadable", "Cannot inspect storage transaction artifact", { filePath }); + } + if (!stat.isFile() || stat.isSymbolicLink() || stat.size <= 0 || stat.size > MAX_RECORD_BYTES) { + throw new StorageTransactionLockError( + "malformed", + "Storage transaction artifact is not a bounded regular file", + { filePath }, + ); + } + try { + return readFileSync(filePath, "utf8"); + } catch (error) { + if (isErrno(error, "EPERM") || isErrno(error, "EACCES")) { + throw new StorageTransactionLockError("eperm", "Cannot read storage transaction artifact", { filePath }); + } + throw new StorageTransactionLockError("unreadable", "Cannot read storage transaction artifact", { filePath }); + } +} + +export function parseStorageTransactionOwnerRecord( + value: unknown, +): StorageTransactionOwnerRecord | null { + if (!value || typeof value !== "object") return null; + const v = value as Record; + const valid = v.schema === STORAGE_TRANSACTION_LOCK_SCHEMA + && typeof v.machineId === "string" && v.machineId.length > 0 + && typeof v.bootId === "string" && v.bootId.length > 0 + && Number.isSafeInteger(v.pid) && (v.pid as number) > 0 + && typeof v.processStartId === "string" && v.processStartId.length > 0 + && typeof v.token === "string" && v.token.length > 0 + && typeof v.operationKind === "string" && v.operationKind.length > 0 + && typeof v.acquiredAt === "string" && Number.isFinite(Date.parse(v.acquiredAt)); + return valid ? (value as StorageTransactionOwnerRecord) : null; +} + +export function parseStorageTransactionParticipationRecord( + value: unknown, +): StorageTransactionParticipationRecord | null { + if (!value || typeof value !== "object") return null; + const v = value as Record; + const valid = v.schema === STORAGE_TRANSACTION_LOCK_SCHEMA + && (v.state === "active" || v.state === "ready") + && typeof v.token === "string" && v.token.length > 0 + && typeof v.updatedAt === "string" && Number.isFinite(Date.parse(v.updatedAt)); + return valid ? (value as StorageTransactionParticipationRecord) : null; +} + +function parseJson(raw: string, filePath: string): unknown { + try { + return JSON.parse(raw); + } catch { + throw new StorageTransactionLockError("malformed", "Storage transaction artifact is invalid JSON", { filePath }); + } +} + +function readOwnerRecord(filePath: string): StorageTransactionOwnerRecord | null { + const raw = readRegularFile(filePath); + if (raw == null) return null; + const parsed = parseStorageTransactionOwnerRecord(parseJson(raw, filePath)); + if (!parsed) { + throw new StorageTransactionLockError("malformed", "Storage transaction owner record is malformed", { filePath }); + } + return parsed; +} + +function readParticipationRecord(filePath: string): StorageTransactionParticipationRecord | null { + const raw = readRegularFile(filePath); + if (raw == null) return null; + const parsed = parseStorageTransactionParticipationRecord(parseJson(raw, filePath)); + if (!parsed) { + throw new StorageTransactionLockError( + "malformed", + "Storage transaction participation marker is malformed", + { filePath }, + ); + } + return parsed; +} + +export function checkStorageTransactionLiveness( + record: StorageTransactionOwnerRecord, + current: { machineId: string; bootId: string; pid: number; processStartId: string }, +): "alive" | "dead" | "foreign_machine" | "eperm" | "unreadable" { + if (record.machineId !== current.machineId) return "foreign_machine"; + if (record.bootId !== current.bootId) return "dead"; + if (record.pid === current.pid) { + return record.processStartId === current.processStartId ? "alive" : "dead"; + } + try { + process.kill(record.pid, 0); + } catch (error) { + if (isErrno(error, "ESRCH")) return "dead"; + if (isErrno(error, "EPERM")) return "eperm"; + return "unreadable"; + } + const observedStart = readProcessStartId(record.pid); + if (!observedStart) return "unreadable"; + return observedStart === record.processStartId ? "alive" : "dead"; +} + +function classifyExisting( + record: StorageTransactionOwnerRecord, + identity: { machineId: string; bootId: string; pid: number; processStartId: string }, +): "dead" { + const status = checkStorageTransactionLiveness(record, identity); + if (status === "dead") return "dead"; + if (status === "alive") { + throw new StorageTransactionLockError("busy", "Storage transaction lock is held by a live process", { + holderPid: record.pid, + }); + } + if (status === "foreign_machine") { + throw new StorageTransactionLockError("foreign_owner", "Storage transaction lock belongs to another machine"); + } + if (status === "eperm") { + throw new StorageTransactionLockError("eperm", "Storage transaction owner liveness is permission-denied"); + } + throw new StorageTransactionLockError("unreadable", "Storage transaction owner liveness is unknown"); +} + +function newRecord( + identity: { machineId: string; bootId: string; pid: number; processStartId: string }, + token: string, + operationKind: string, + nowMs: number, +): StorageTransactionOwnerRecord { + return { + schema: STORAGE_TRANSACTION_LOCK_SCHEMA, + ...identity, + token, + operationKind, + acquiredAt: new Date(nowMs).toISOString(), + }; +} + +function writeTempRecord(root: string, role: string, record: StorageTransactionOwnerRecord): string { + const filePath = join(root, `${STORAGE_TRANSACTION_RESERVED_PREFIX}${role}.tmp.${record.token}.${randomUUID()}`); + const fd = openSync(filePath, "wx", 0o600); + try { + writeSync(fd, `${JSON.stringify(record)}\n`, undefined, "utf8"); + fsyncSync(fd); + } finally { + closeSync(fd); + } + return filePath; +} + +function publishHardLink(tempPath: string, canonicalPath: string): void { + try { + linkSync(tempPath, canonicalPath); + } catch (error) { + if (isErrno(error, "EEXIST")) throw error; + if (["EXDEV", "EPERM", "ENOTSUP", "EOPNOTSUPP"].some((code) => isErrno(error, code))) { + throw new StorageTransactionLockError( + "unsupported_filesystem", + "Filesystem cannot publish the storage transaction lock atomically", + ); + } + throw error; + } + const source = lstatSync(tempPath); + const canonical = lstatSync(canonicalPath); + if (!source.isFile() || !canonical.isFile() || source.dev !== canonical.dev || source.ino !== canonical.ino) { + throw new StorageTransactionLockError("protocol_fault", "Published lock is not the expected hard-linked record"); + } +} + +function retireRecord( + filePath: string, + root: string, + role: string, + expectedToken: string, +): string { + const retired = join(root, `${STORAGE_TRANSACTION_RESERVED_PREFIX}${role}.${expectedToken}.${randomUUID()}`); + try { + renameSync(filePath, retired); + } catch (error) { + if (isErrno(error, "ENOENT")) { + throw new StorageTransactionLockError("busy", "Storage transaction ownership changed"); + } + throw error; + } + const moved = readOwnerRecord(retired); + if (!moved || moved.token !== expectedToken) { + throw new StorageTransactionLockError( + "protocol_fault", + `Retired ${role} record token does not match the observed owner`, + ); + } + return retired; +} + +function acquireRecoveryMutex( + paths: ReturnType, + identity: { machineId: string; bootId: string; pid: number; processStartId: string }, + token: string, + nowMs: number, +): { release: () => void } { + const record = newRecord(identity, token, "storage-transaction-recovery", nowMs); + const temp = writeTempRecord(paths.sourceRootRealPath, "recovery", record); + try { + try { + publishHardLink(temp, paths.recoveryMutexPath); + } catch (error) { + if (!isErrno(error, "EEXIST")) throw error; + const existing = readOwnerRecord(paths.recoveryMutexPath); + if (!existing) throw new StorageTransactionLockError("recovery_conflict", "Storage transaction recovery ownership changed"); + classifyExisting(existing, identity); + const reread = readOwnerRecord(paths.recoveryMutexPath); + if (!reread) throw new StorageTransactionLockError("recovery_conflict", "Storage transaction recovery ownership changed"); + classifyExisting(reread, identity); + const quarantined = retireRecord(paths.recoveryMutexPath, paths.sourceRootRealPath, "quarantine", reread.token); + safeUnlink(quarantined); + try { + publishHardLink(temp, paths.recoveryMutexPath); + } catch (publishError) { + if (isErrno(publishError, "EEXIST")) { + throw new StorageTransactionLockError("recovery_conflict", "Storage transaction recovery mutex is busy"); + } + throw publishError; + } + } + } finally { + safeUnlink(temp); + } + return { + release() { + const existing = readOwnerRecord(paths.recoveryMutexPath); + if (!existing || existing.token !== token) return; + const retired = retireRecord(paths.recoveryMutexPath, paths.sourceRootRealPath, "release", token); + safeUnlink(retired); + }, + }; +} + +function writeParticipation( + paths: ReturnType, + marker: StorageTransactionParticipationRecord, +): void { + const temp = join( + paths.sourceRootRealPath, + `${STORAGE_TRANSACTION_RESERVED_PREFIX}marker.tmp.${marker.token}.${randomUUID()}`, + ); + const fd = openSync(temp, "wx", 0o600); + try { + writeSync(fd, `${JSON.stringify(marker)}\n`, undefined, "utf8"); + fsyncSync(fd); + } finally { + closeSync(fd); + } + try { + renameSync(temp, paths.participationPath); + } catch (error) { + safeUnlink(temp); + throw error; + } +} + +function markerFor( + state: StorageTransactionParticipationState, + token: string, + nowMs: number, +): StorageTransactionParticipationRecord { + return { + schema: STORAGE_TRANSACTION_LOCK_SCHEMA, + state, + token, + updatedAt: new Date(nowMs).toISOString(), + }; +} + +export function isStorageTransactionReservedName(name: string): boolean { + return typeof name === "string" && name.startsWith(STORAGE_TRANSACTION_RESERVED_PREFIX); +} + +export function isStorageTransactionReservedPath(filePath: string, sourceRoot: string): boolean { + const root = resolveSourceRootRealPath(sourceRoot, { createIfMissing: false }); + const resolved = resolve(filePath); + const rel = resolved.startsWith(root + sep) + ? resolved.slice(root.length + 1) + : resolved === root + ? "" + : null; + if (rel == null) return false; + const top = rel.split(sep)[0] ?? ""; + return isStorageTransactionReservedName(top); +} + +/** + * Acquire the source-root storage transaction lock. + * Publishes a fully populated owner record via same-directory hardlink and holds + * it for the entire producer/archive transaction. + */ +export function acquireStorageTransactionLock( + options: AcquireStorageTransactionLockOptions, +): StorageTransactionLockHandle { + const sourceRootRealPath = resolveSourceRootRealPath(options.sourceRoot); + const paths = ownerPaths(sourceRootRealPath); + const identity = resolveStorageTransactionIdentity(); + const nowMs = options.nowMs ?? Date.now(); + const token = randomUUID(); + const owner = newRecord(identity, token, options.operationKind || "backup", nowMs); + const recoveries: Array> = []; + + const publishOwner = (): StorageTransactionLockHandle => { + const temp = writeTempRecord(sourceRootRealPath, "owner", owner); + try { + publishHardLink(temp, paths.ownerPath); + } finally { + safeUnlink(temp); + } + try { + writeParticipation(paths, markerFor("active", token, nowMs)); + } catch (error) { + const recovery = acquireRecoveryMutex(paths, identity, randomUUID(), Date.now()); + try { + const current = readOwnerRecord(paths.ownerPath); + if (current && current.token === token) { + const retired = retireRecord(paths.ownerPath, sourceRootRealPath, "release", token); + safeUnlink(retired); + } + } finally { + recovery.release(); + } + throw error; + } + return { + sourceRootRealPath, + ownerPath: paths.ownerPath, + recoveryMutexPath: paths.recoveryMutexPath, + participationPath: paths.participationPath, + token, + owner, + recoveries, + }; + }; + + try { + return publishOwner(); + } catch (error) { + if (!isErrno(error, "EEXIST")) throw error; + } + + const existing = readOwnerRecord(paths.ownerPath); + if (!existing) { + throw new StorageTransactionLockError("busy", "Storage transaction ownership changed; retry from fresh state"); + } + classifyExisting(existing, identity); + if (options.allowDeadOwnerRecovery === false) { + throw new StorageTransactionLockError("busy", "Dead-owner recovery is disabled"); + } + + const recovery = acquireRecoveryMutex(paths, identity, token, nowMs); + try { + const reread = readOwnerRecord(paths.ownerPath); + if (reread) { + classifyExisting(reread, identity); + const quarantined = retireRecord(paths.ownerPath, sourceRootRealPath, "quarantine", reread.token); + recoveries.push({ + sourceRootRealPath, + classification: "dead_local", + quarantined: true, + quarantinePath: quarantined, + recoveredToken: reread.token, + }); + safeUnlink(quarantined); + } + try { + return publishOwner(); + } catch (error) { + if (isErrno(error, "EEXIST")) { + throw new StorageTransactionLockError("busy", "Storage transaction lock was acquired by another owner"); + } + throw error; + } + } finally { + recovery.release(); + } +} + +export function assertStorageTransactionLockHeld(handle: StorageTransactionLockHandle): StorageTransactionOwnerRecord { + const current = readOwnerRecord(handle.ownerPath); + if (!current || current.token !== handle.token) { + throw new StorageTransactionLockError( + "token_mismatch", + "Storage transaction ownership no longer matches this handle", + { sourceRootRealPath: handle.sourceRootRealPath }, + ); + } + if (current.pid !== process.pid || current.processStartId !== readProcessStartId()) { + throw new StorageTransactionLockError( + "not_holder", + "Storage transaction owner identity no longer matches this process", + { sourceRootRealPath: handle.sourceRootRealPath }, + ); + } + return current; +} + +export function markStorageTransactionParticipationReady( + handle: StorageTransactionLockHandle, + nowMs: number = Date.now(), +): void { + assertStorageTransactionLockHeld(handle); + writeParticipation(ownerPaths(handle.sourceRootRealPath), markerFor("ready", handle.token, nowMs)); +} + +/** + * Token-safe release: removes only this holder's owner record and leaves a + * ready participation marker for enrolled archive consumers. + */ +export function releaseStorageTransactionLock( + handle: StorageTransactionLockHandle, + options: { participationState?: StorageTransactionParticipationState } = {}, +): void { + const identity = resolveStorageTransactionIdentity(); + const paths = ownerPaths(handle.sourceRootRealPath); + const recovery = acquireRecoveryMutex(paths, identity, randomUUID(), Date.now()); + try { + const current = readOwnerRecord(paths.ownerPath); + if (!current) return; + if (current.token !== handle.token) { + throw new StorageTransactionLockError( + "token_mismatch", + "Refusing to release a replacement storage transaction owner", + { sourceRootRealPath: handle.sourceRootRealPath }, + ); + } + if ((options.participationState || "ready") === "ready") { + writeParticipation(paths, markerFor("ready", handle.token, Date.now())); + } + const retired = retireRecord(paths.ownerPath, paths.sourceRootRealPath, "release", handle.token); + safeUnlink(retired); + } finally { + recovery.release(); + } +} + +export function readStorageTransactionParticipation( + sourceRoot: string, + options: { createIfMissing?: boolean } = {}, +): StorageTransactionParticipationRecord | null { + const sourceRootRealPath = resolveSourceRootRealPath(sourceRoot, { + createIfMissing: options.createIfMissing === true, + }); + try { + return readParticipationRecord(ownerPaths(sourceRootRealPath).participationPath); + } catch { + return null; + } +} + +export function classifyPaperclipParticipation(sourceRoot: string): { + ok: boolean; + reason: string | null; + sourceRootRealPath: string; + record?: StorageTransactionParticipationRecord | null; + newestSourceMtimeMs?: number | null; +} { + const sourceRootRealPath = resolveSourceRootRealPath(sourceRoot, { createIfMissing: false }); + let record: StorageTransactionParticipationRecord | null = null; + try { + record = readParticipationRecord(ownerPaths(sourceRootRealPath).participationPath); + } catch { + return { ok: false, reason: "participation_incompatible", sourceRootRealPath }; + } + if (!record) return { ok: false, reason: "participation_missing", sourceRootRealPath }; + if (record.state === "active") { + return { ok: false, reason: "participation_active", sourceRootRealPath, record }; + } + return { ok: true, reason: null, sourceRootRealPath, record }; +} diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index fb62bcf9cc..23d18d8730 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -29,6 +29,35 @@ export { type RunDatabaseBackupResult, type RunDatabaseRestoreOptions, } from "./backup-lib.js"; +export { + STORAGE_TRANSACTION_LOCK_SCHEMA, + STORAGE_TRANSACTION_RESERVED_PREFIX, + STORAGE_TRANSACTION_LOCK_DIR_NAME, + STORAGE_TRANSACTION_RECOVERY_MUTEX_NAME, + STORAGE_TRANSACTION_PARTICIPATION_NAME, + STORAGE_TRANSACTION_TMP_PREFIX, + STORAGE_TRANSACTION_QUARANTINE_PREFIX, + StorageTransactionLockError, + acquireStorageTransactionLock, + assertStorageTransactionLockHeld, + releaseStorageTransactionLock, + markStorageTransactionParticipationReady, + readStorageTransactionParticipation, + parseStorageTransactionOwnerRecord, + parseStorageTransactionParticipationRecord, + isStorageTransactionReservedName, + isStorageTransactionReservedPath, + resolveSourceRootRealPath, + readLocalMachineId, + readLocalBootId, + readProcessStartId, + resolveStorageTransactionIdentity, + classifyPaperclipParticipation, + type StorageTransactionLockHandle, + type StorageTransactionOwnerRecord, + type StorageTransactionParticipationRecord, + type StorageTransactionOperationKind, +} from "./backup-transaction-lock.js"; export { createEmbeddedPostgresLogBuffer, formatEmbeddedPostgresError, From 7092d533ea4afc0d553df3763fd8c4d043c7cfed Mon Sep 17 00:00:00 2001 From: Andrew Levine Date: Fri, 11 Sep 2026 15:06:08 -0400 Subject: [PATCH 4/4] fix(SUP-3933): harden backup lock teardown and busy schedule skip Acquire the storage transaction lock only after client construction so synchronous setup failures never publish an owner. Release/close errors now surface through teardown aggregation, and scheduled backups skip cleanly when the lock is busy. --- packages/db/src/backup-lib.test.ts | 38 ++++++++++++++++++++ packages/db/src/backup-lib.ts | 58 ++++++++++++++++++++---------- server/src/index.ts | 8 +++++ 3 files changed, 85 insertions(+), 19 deletions(-) diff --git a/packages/db/src/backup-lib.test.ts b/packages/db/src/backup-lib.test.ts index 13c5edc9d8..df33e46f3a 100644 --- a/packages/db/src/backup-lib.test.ts +++ b/packages/db/src/backup-lib.test.ts @@ -5,6 +5,10 @@ import { gunzipSync } from "node:zlib"; import { afterEach, describe, expect, it } from "vitest"; import postgres from "postgres"; import { createBufferedTextFileWriter, runDatabaseBackup, runDatabaseRestore } from "./backup-lib.js"; +import { + STORAGE_TRANSACTION_LOCK_DIR_NAME, + STORAGE_TRANSACTION_PARTICIPATION_NAME, +} from "./backup-transaction-lock.js"; import { ensurePostgresDatabase } from "./client.js"; import { getEmbeddedPostgresTestSupport, @@ -74,6 +78,40 @@ describe("createBufferedTextFileWriter", () => { }); }); +describe("runDatabaseBackup transaction teardown", () => { + it("does not publish an owner when database client setup throws synchronously", async () => { + const backupDir = createTempDir("paperclip-db-backup-setup-failure-"); + + await expect(runDatabaseBackup({ + connectionString: "not-a-url", + backupDir, + retention: { dailyDays: 7, weeklyWeeks: 4, monthlyMonths: 1 }, + backupEngine: "javascript", + })).rejects.toThrow(); + + expect(fs.existsSync(path.join(backupDir, STORAGE_TRANSACTION_LOCK_DIR_NAME))).toBe(false); + expect(fs.existsSync(path.join(backupDir, STORAGE_TRANSACTION_PARTICIPATION_NAME))).toBe(false); + }); + + it("releases an acquired owner when the backup connection fails", async () => { + const backupDir = createTempDir("paperclip-db-backup-connection-failure-"); + + await expect(runDatabaseBackup({ + connectionString: "postgres://paperclip:paperclip@127.0.0.1:1/paperclip", + backupDir, + retention: { dailyDays: 7, weeklyWeeks: 4, monthlyMonths: 1 }, + backupEngine: "javascript", + connectTimeoutSeconds: 1, + })).rejects.toThrow(); + + expect(fs.existsSync(path.join(backupDir, STORAGE_TRANSACTION_LOCK_DIR_NAME))).toBe(false); + expect(JSON.parse(fs.readFileSync( + path.join(backupDir, STORAGE_TRANSACTION_PARTICIPATION_NAME), + "utf8", + ))).toMatchObject({ state: "ready" }); + }); +}); + describeEmbeddedPostgres("runDatabaseBackup", () => { it( "keeps the newest backup for each retained calendar month", diff --git a/packages/db/src/backup-lib.ts b/packages/db/src/backup-lib.ts index 16dd50063f..74b936f274 100644 --- a/packages/db/src/backup-lib.ts +++ b/packages/db/src/backup-lib.ts @@ -556,7 +556,16 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise const excludedTableNames = normalizeTableNameSet(opts.excludeTables); const nullifiedColumnsByTable = normalizeNullifyColumnMap(opts.nullifyColumns); + let sql = postgres(opts.connectionString, { max: 1, connect_timeout: connectTimeout }); + let sqlClosed = false; + const closeSql = async () => { + if (sqlClosed) return; + sqlClosed = true; + await sql.end(); + }; + // Acquire the source-root transaction lock before any backup/prune filesystem activity. + // Client construction stays outside the hold so a synchronous setup error cannot strand it. mkdirSync(opts.backupDir, { recursive: true }); let lockHandle: StorageTransactionLockHandle; try { @@ -565,24 +574,21 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise operationKind: "backup", }); } catch (error) { + await closeSql(); if (error instanceof StorageTransactionLockError) { throw error; } throw error; } - let sql = postgres(opts.connectionString, { max: 1, connect_timeout: connectTimeout }); - let sqlClosed = false; - const closeSql = async () => { - if (sqlClosed) return; - sqlClosed = true; - await sql.end(); - }; const sqlFile = resolve(opts.backupDir, `${filenamePrefix}-${timestamp()}.sql`); const backupFile = `${sqlFile}.gz`; - const writer = createBufferedTextFileWriter(sqlFile); + let writer: ReturnType | null = null; + let operationError: unknown = null; try { + writer = createBufferedTextFileWriter(sqlFile); + const activeWriter = writer; if (backupEngine === "pg_dump" || (backupEngine === "auto" && canUsePgDump)) { await sql`SELECT 1`; try { @@ -592,7 +598,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise backupFile, connectTimeout, }); - await writer.abort(); + await activeWriter.abort(); const sizeBytes = statSync(backupFile).size; assertStorageTransactionLockHeld(lockHandle); const prunedCount = pruneOldBackups(opts.backupDir, retention, filenamePrefix, lockHandle); @@ -617,7 +623,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise await sql`SELECT 1`; - const emit = (line: string) => writer.emit(line); + const emit = (line: string) => activeWriter.emit(line); const emitStatement = (statement: string) => { emit(statement); emit(STATEMENT_BREAKPOINT); @@ -975,19 +981,19 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise const nullifiedColumns = nullifiedColumnsByTable.get(currentTableKey) ?? new Set(); if (effectiveBackupEngine !== "javascript" && nullifiedColumns.size === 0) { emit(`COPY ${qualifiedTableName} (${colNames}) FROM stdin;`); - await writer.writeRaw("\n"); + await activeWriter.writeRaw("\n"); const copySql = postgres(opts.connectionString, { max: 1, connect_timeout: connectTimeout }); try { const copyStream = await copySql .unsafe(`COPY ${qualifiedTableName} (${colNames}) TO STDOUT`) .readable(); for await (const chunk of copyStream) { - await writer.writeRaw(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk))); + await activeWriter.writeRaw(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk))); } } finally { await copySql.end(); } - await writer.writeRaw("\\.\n"); + await activeWriter.writeRaw("\\.\n"); emitStatementBoundary(); emit(""); continue; @@ -1004,7 +1010,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise ); emitStatement(`INSERT INTO ${qualifiedTableName} (${colNames}) VALUES (${values.join(", ")});`); } - await writer.drain(); + await activeWriter.drain(); } emit(""); } @@ -1057,7 +1063,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise emitStatement("COMMIT;"); emit(""); - await writer.close(); + await activeWriter.close(); // Compress the SQL file with gzip const sqlReadStream = createReadStream(sqlFile); @@ -1076,7 +1082,8 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise lockTokenPrefix: lockHandle.token.slice(0, 8), }; } catch (error) { - await writer.abort(); + operationError = error; + await writer?.abort(); if (existsSync(backupFile)) { try { unlinkSync(backupFile); } catch { /* ignore */ } } @@ -1085,11 +1092,24 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise } throw error; } finally { - await closeSql(); + let closeError: unknown = null; + try { + await closeSql(); + } catch (error) { + closeError = error; + } + let releaseError: unknown = null; try { releaseStorageTransactionLock(lockHandle, { participationState: "ready" }); - } catch { - // Release is best-effort on teardown; token mismatch means another owner already replaced us. + } catch (error) { + releaseError = error; + } + const teardownErrors = [operationError, closeError, releaseError].filter( + (error): error is NonNullable => error !== null, + ); + if (closeError || releaseError) { + if (teardownErrors.length === 1) throw teardownErrors[0]; + throw new AggregateError(teardownErrors, "Database backup and storage transaction teardown failed"); } } } diff --git a/server/src/index.ts b/server/src/index.ts index 723c118f55..0347102c31 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -32,6 +32,7 @@ import { reconcilePendingMigrationHistory, formatDatabaseBackupResult, runDatabaseBackup, + StorageTransactionLockError, authUsers, companies, companyMemberships, @@ -855,6 +856,13 @@ async function startServerWithDatabaseTeardown( ); return response; } catch (err) { + if (trigger === "scheduled" && err instanceof StorageTransactionLockError && err.code === "busy") { + logger.warn( + { backupDir: config.databaseBackupDir, trigger, reason: err.code }, + "Skipping scheduled database backup because the storage transaction lock is busy", + ); + return null; + } logger.error({ err, backupDir: config.databaseBackupDir, trigger }, `${label} database backup failed`); throw err; } finally {