Merge 7092d533ea into c9e3bb7ca4
This commit is contained in:
levineam
GitHub
commit
cb726a0021
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise
|
|||
const canUsePgDump = !hasBackupTransforms(opts);
|
||||
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 () => {
|
||||
|
|
@ -540,12 +563,32 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise
|
|||
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 {
|
||||
lockHandle = acquireStorageTransactionLock({
|
||||
sourceRoot: opts.backupDir,
|
||||
operationKind: "backup",
|
||||
});
|
||||
} catch (error) {
|
||||
await closeSql();
|
||||
if (error instanceof StorageTransactionLockError) {
|
||||
throw error;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
const sqlFile = resolve(opts.backupDir, `${filenamePrefix}-${timestamp()}.sql`);
|
||||
const backupFile = `${sqlFile}.gz`;
|
||||
const writer = createBufferedTextFileWriter(sqlFile);
|
||||
let writer: ReturnType<typeof createBufferedTextFileWriter> | 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 {
|
||||
|
|
@ -555,13 +598,15 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise
|
|||
backupFile,
|
||||
connectTimeout,
|
||||
});
|
||||
await writer.abort();
|
||||
await activeWriter.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)) {
|
||||
|
|
@ -578,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);
|
||||
|
|
@ -936,19 +981,19 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise
|
|||
const nullifiedColumns = nullifiedColumnsByTable.get(currentTableKey) ?? new Set<string>();
|
||||
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;
|
||||
|
|
@ -965,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("");
|
||||
}
|
||||
|
|
@ -1018,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);
|
||||
|
|
@ -1026,16 +1071,19 @@ 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();
|
||||
operationError = error;
|
||||
await writer?.abort();
|
||||
if (existsSync(backupFile)) {
|
||||
try { unlinkSync(backupFile); } catch { /* ignore */ }
|
||||
}
|
||||
|
|
@ -1044,7 +1092,25 @@ 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 (error) {
|
||||
releaseError = error;
|
||||
}
|
||||
const teardownErrors = [operationError, closeError, releaseError].filter(
|
||||
(error): error is NonNullable<typeof error> => error !== null,
|
||||
);
|
||||
if (closeError || releaseError) {
|
||||
if (teardownErrors.length === 1) throw teardownErrors[0];
|
||||
throw new AggregateError(teardownErrors, "Database backup and storage transaction teardown failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string, unknown> = {}) {
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<Record<string, unknown>>;
|
||||
};
|
||||
|
||||
export type AcquireStorageTransactionLockOptions = {
|
||||
sourceRoot: string;
|
||||
operationKind: StorageTransactionOperationKind | string;
|
||||
allowDeadOwnerRecovery?: boolean;
|
||||
nowMs?: number;
|
||||
};
|
||||
|
||||
export class StorageTransactionLockError extends Error {
|
||||
readonly code: string;
|
||||
readonly details?: Record<string, unknown>;
|
||||
|
||||
constructor(code: string, message: string, details?: Record<string, unknown>) {
|
||||
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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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<typeof ownerPaths>,
|
||||
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<typeof ownerPaths>,
|
||||
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<Record<string, unknown>> = [];
|
||||
|
||||
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 };
|
||||
}
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ import {
|
|||
reconcilePendingMigrationHistory,
|
||||
formatDatabaseBackupResult,
|
||||
runDatabaseBackup,
|
||||
StorageTransactionLockError,
|
||||
authUsers,
|
||||
companies,
|
||||
companyMemberships,
|
||||
|
|
@ -863,6 +864,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 {
|
||||
|
|
|
|||
Loading…
Reference in New Issue