mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
fix: checkpoint database backups before idle sleep
Add opt-in backup checkpoints under the owned idle hold. Verify and sync each archive before pruning history, retain backup work through completion, and take a fresh backup after restart using a persistent marker. Preserve automatic backups and the instance retention policy. Tested idle/route/spool checks, PostgreSQL backup restore and retention failure boundaries, recursive typecheck, and build. Full suite is in progress. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
4265cb3a2b
commit
e743a8ea50
14 files changed
+333
-20
No files matched your search
@@ -0,0 +1,92 @@
|
||||
import { mkdtemp, readFile, rm, symlink, utimes, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { gzipSync } from "node:zlib";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { createDb, ensurePostgresDatabase, getEmbeddedPostgresTestSupport, runDatabaseBackup, runDatabaseRestore, startEmbeddedPostgresTestDatabase } from "@paperclipai/db";
|
||||
import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "../services/idle-database-backup.js";
|
||||
|
||||
let directory: string;
|
||||
beforeEach(async () => { directory = await mkdtemp(path.join(os.tmpdir(), "idle-backup-")); });
|
||||
afterEach(async () => { await rm(directory, { recursive: true, force: true }); });
|
||||
|
||||
describe("idle database backup persistence", () => {
|
||||
it("keeps a restart marker until a fresh valid backup is complete", async () => {
|
||||
const archive = path.join(directory, "backup.sql.gz");
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
await writeFile(archive, gzipSync("CREATE TABLE fixture (id integer);\n"));
|
||||
await verifyIdleDatabaseBackup(archive);
|
||||
// Verification on the sleep path must not remove the wake obligation.
|
||||
expect(await readFile(idleBackupWakeMarker(directory), "utf8")).toContain("restarts");
|
||||
await clearIdleBackupWakeRequired(directory);
|
||||
await expect(readFile(idleBackupWakeMarker(directory))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(await readFile(archive)).toEqual(gzipSync("CREATE TABLE fixture (id integer);\n"));
|
||||
});
|
||||
|
||||
it.each([
|
||||
["empty file", Buffer.alloc(0)], ["empty SQL", gzipSync("")],
|
||||
["truncated gzip", gzipSync("SELECT 1;").subarray(0, -4)],
|
||||
["plain text", Buffer.from("SELECT 1;")],
|
||||
])("rejects %s and retains the restart obligation", async (_label, contents) => {
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
const archive = path.join(directory, "invalid.sql.gz");
|
||||
await writeFile(archive, contents);
|
||||
await expect(verifyIdleDatabaseBackup(archive)).rejects.toThrow();
|
||||
expect(await readFile(idleBackupWakeMarker(directory), "utf8")).toContain("restarts");
|
||||
});
|
||||
|
||||
it("rejects a missing archive and symlinked archive or marker", async () => {
|
||||
const archive = path.join(directory, "backup.sql.gz");
|
||||
await expect(verifyIdleDatabaseBackup(archive)).rejects.toThrow();
|
||||
const target = path.join(directory, "target");
|
||||
await writeFile(target, gzipSync("SELECT 1;"));
|
||||
await symlink(target, archive);
|
||||
await expect(verifyIdleDatabaseBackup(archive)).rejects.toThrow();
|
||||
await symlink(target, idleBackupWakeMarker(directory));
|
||||
await expect(markIdleBackupWakeRequired(directory)).rejects.toThrow();
|
||||
expect(await readFile(target)).toEqual(gzipSync("SELECT 1;"));
|
||||
});
|
||||
});
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("idle backup restore", () => {
|
||||
it("restores the last pre-sleep state into a separate database", async () => {
|
||||
const source = await startEmbeddedPostgresTestDatabase("idle-backup-restore-");
|
||||
const sourceDb = createDb(source.connectionString);
|
||||
const targetUrl = new URL(source.connectionString);
|
||||
targetUrl.pathname = "/idle_restore";
|
||||
try {
|
||||
await sourceDb.execute(sql`CREATE TABLE backup_fixture (id integer PRIMARY KEY, value text NOT NULL)`);
|
||||
await sourceDb.execute(sql`INSERT INTO backup_fixture VALUES (1, 'last committed state')`);
|
||||
await markIdleBackupWakeRequired(directory);
|
||||
const result = await runDatabaseBackup({
|
||||
connectionString: source.connectionString, backupDir: directory,
|
||||
retention: { dailyDays: 14, weeklyWeeks: 4, monthlyMonths: 6 },
|
||||
verifyBeforePrune: verifyIdleDatabaseBackup,
|
||||
});
|
||||
await verifyIdleDatabaseBackup(result.backupFile);
|
||||
await ensurePostgresDatabase(source.connectionString, "idle_restore");
|
||||
await runDatabaseRestore({ connectionString: targetUrl.toString(), backupFile: result.backupFile });
|
||||
const target = createDb(targetUrl.toString());
|
||||
expect(await target.execute(sql`SELECT id, value FROM backup_fixture`)).toEqual([{ id: 1, value: "last committed state" }]);
|
||||
const checkpointBytes = await readFile(result.backupFile);
|
||||
expect(await readFile(idleBackupWakeMarker(directory), "utf8")).toContain("restarts");
|
||||
// A failed verification must not delete existing recovery points, even
|
||||
// when their age would make them eligible for the configured pruning.
|
||||
const history = path.join(directory, "paperclip-history.sql.gz");
|
||||
const historyBytes = gzipSync("SELECT 'historical recovery point';");
|
||||
await writeFile(history, historyBytes);
|
||||
await utimes(history, new Date(0), new Date(0));
|
||||
await expect(runDatabaseBackup({
|
||||
connectionString: source.connectionString, backupDir: directory,
|
||||
backupEngine: "javascript", retention: { dailyDays: 7, weeklyWeeks: 4, monthlyMonths: 1 },
|
||||
verifyBeforePrune: async () => { throw new Error("verification failed"); },
|
||||
})).rejects.toThrow("verification failed");
|
||||
expect(await readFile(history)).toEqual(historyBytes);
|
||||
expect(await readFile(result.backupFile)).toEqual(checkpointBytes);
|
||||
} finally {
|
||||
await source.cleanup();
|
||||
}
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -40,10 +40,12 @@ describe("idle cleanup inspection", () => {
|
||||
expect(await readIdleLocalWork()).toBe("none");
|
||||
markIdleStartupComplete({ scheduledBackups: true });
|
||||
expect(await readIdleLocalWork()).toBe("present");
|
||||
expect(await readIdleLocalWork({ backupCheckpoint: true })).toBe("none");
|
||||
markIdleStartupComplete({ scheduledBackups: false });
|
||||
await fs.mkdir(custom);
|
||||
await fs.writeFile(path.join(custom, "receipt.tmp"), "incomplete");
|
||||
expect(await readIdleLocalWork()).toBe("present");
|
||||
expect(await readIdleLocalWork({ backupCheckpoint: true })).toBe("present");
|
||||
await fs.rm(custom, { recursive: true });
|
||||
expect(await readIdleLocalWork()).toBe("none");
|
||||
});
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { beginIdleTrackedWork } from "../services/task-admission.js";
|
||||
import { beginIdleTrackedWork, idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
@@ -22,6 +22,73 @@ const ownerId = "d0b833f4-4098-42de-8420-1907f3aa4895";
|
||||
const owned = () => ({ ...held(), ownerId });
|
||||
const emptyLocal = async () => "none" as const;
|
||||
|
||||
describe("idle backup checkpoint", () => {
|
||||
const database = (blocked = false) => {
|
||||
const transaction = vi.fn(async (run: (tx: unknown) => Promise<unknown>) => run({ execute: async () => [{ blocked }] }));
|
||||
return { db: { transaction } as unknown as Db, transaction };
|
||||
};
|
||||
const backup = (work: () => Promise<boolean> = async () => true) => vi.fn(async () => {
|
||||
const finish = beginIdleTrackedWork();
|
||||
try { return await work(); } finally { finish(); }
|
||||
});
|
||||
it("backs up after durable inspection and rechecks local work before authorizing sleep", async () => {
|
||||
const { db, transaction } = database();
|
||||
const local = vi.fn(emptyLocal);
|
||||
const checkpoint = backup(async () => {
|
||||
expect(transaction).toHaveBeenCalledOnce();
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
return true;
|
||||
});
|
||||
expect(await readIdleSleepSafety(db, owned, () => now, ownerId, local, undefined, checkpoint)).toEqual(none);
|
||||
expect(checkpoint).toHaveBeenCalledOnce();
|
||||
expect(local).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
it.each(["present", "unknown"] as const)("avoids DB probes and backups for known local %s work", async state => {
|
||||
const { db, transaction } = database();
|
||||
const checkpoint = backup();
|
||||
const plugins = vi.fn();
|
||||
expect(await readIdleSleepSafety(db, owned, () => now, ownerId, async () => state, plugins, checkpoint))
|
||||
.toEqual({ version: 1, backgroundWork: state });
|
||||
expect(transaction).not.toHaveBeenCalled();
|
||||
expect(plugins).not.toHaveBeenCalled();
|
||||
expect(checkpoint).not.toHaveBeenCalled();
|
||||
});
|
||||
it("does not start a backup for durable work or an unowned hold", async () => {
|
||||
const checkpoint = backup();
|
||||
expect(await readIdleSleepSafety(database(true).db, owned, () => now, ownerId, emptyLocal, undefined, checkpoint)).toEqual(present);
|
||||
expect(await readIdleSleepSafety(database().db, owned, () => now, "stale", emptyLocal, undefined, checkpoint)).toEqual(unknown);
|
||||
expect(checkpoint).not.toHaveBeenCalled();
|
||||
});
|
||||
it.each(["failed", "threw", "untracked", "concurrent", "expired", "replaced"])("keeps the instance awake when the backup is %s", async kind => {
|
||||
let clock = now;
|
||||
let status = owned();
|
||||
const work = async () => {
|
||||
if (kind === "threw") throw new Error("private archive path");
|
||||
if (kind === "concurrent") { const finish = beginIdleTrackedWork(); finish(); }
|
||||
if (kind === "expired") clock += 60_000;
|
||||
if (kind === "replaced") status = { ...status, ownerId: "replacement" };
|
||||
return kind !== "failed";
|
||||
};
|
||||
const checkpoint = kind === "untracked" ? work : backup(work);
|
||||
expect(await readIdleSleepSafety(database().db, () => status, () => clock, ownerId, emptyLocal, undefined, checkpoint)).toEqual(unknown);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
it("retains actual backup work after the owning hold expires", async () => {
|
||||
let finishBackup!: () => void;
|
||||
const pending = new Promise<void>(resolve => { finishBackup = resolve; });
|
||||
const checkpoint = backup(async () => { await pending; return true; });
|
||||
let clock = now;
|
||||
const scan = readIdleSleepSafety(database().db, owned, () => clock, ownerId, emptyLocal, undefined, checkpoint);
|
||||
await vi.waitFor(() => expect(checkpoint).toHaveBeenCalledOnce());
|
||||
clock += 60_000;
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
expect(await readIdleSleepSafety(database().db, owned, () => clock, ownerId, emptyLocal)).toEqual(unknown);
|
||||
finishBackup();
|
||||
expect(await scan).toEqual(unknown);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("idle sleep safety failure boundaries", () => {
|
||||
it.each([
|
||||
@@ -36,7 +103,7 @@ describe("idle sleep safety failure boundaries", () => {
|
||||
|
||||
it("fails closed when the database is unavailable", async () => {
|
||||
const db = { transaction: vi.fn().mockRejectedValue(new Error("private connection detail")) } as unknown as Db;
|
||||
expect(await readIdleSleepSafety(db, held, () => now)).toEqual(unknown);
|
||||
expect(await readIdleSleepSafety(db, held, () => now, undefined, emptyLocal)).toEqual(unknown);
|
||||
});
|
||||
|
||||
it.each([
|
||||
@@ -46,14 +113,14 @@ describe("idle sleep safety failure boundaries", () => {
|
||||
const getStatus = vi.fn().mockReturnValueOnce(held()).mockReturnValue({ ...held(), ...change });
|
||||
const execute = vi.fn().mockResolvedValue([{ blocked: true }]);
|
||||
const transaction = vi.fn(async (fn: (tx: unknown) => Promise<unknown>) => fn({ execute }));
|
||||
expect(await readIdleSleepSafety({ transaction } as unknown as Db, getStatus, () => now)).toEqual(unknown);
|
||||
expect(await readIdleSleepSafety({ transaction } as unknown as Db, getStatus, () => now, undefined, emptyLocal)).toEqual(unknown);
|
||||
expect(transaction).toHaveBeenCalledWith(expect.any(Function), { isolationLevel: "repeatable read", accessMode: "read only" });
|
||||
});
|
||||
|
||||
it("reports persisted work when the admission hold remains unchanged", async () => {
|
||||
const execute = vi.fn().mockResolvedValue([{ blocked: true }]);
|
||||
const transaction = vi.fn(async (fn: (tx: unknown) => Promise<unknown>) => fn({ execute }));
|
||||
expect(await readIdleSleepSafety({ transaction } as unknown as Db, held, () => now)).toEqual(present);
|
||||
expect(await readIdleSleepSafety({ transaction } as unknown as Db, held, () => now, undefined, emptyLocal)).toEqual(present);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -77,7 +77,7 @@ describe("instance settings routes", () => {
|
||||
return { errorHandler, instanceSettingsRoutes };
|
||||
});
|
||||
|
||||
function createApp(actor: any, pluginWorkers?: PluginWorkerManager) {
|
||||
function createApp(actor: any, pluginWorkers?: PluginWorkerManager, prepareBackup?: () => Promise<boolean>) {
|
||||
const { errorHandler, instanceSettingsRoutes } = routeModules.value;
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
@@ -85,7 +85,7 @@ describe("instance settings routes", () => {
|
||||
req.actor = actor;
|
||||
next();
|
||||
});
|
||||
app.use("/api", instanceSettingsRoutes(mockDb as any, pluginWorkers));
|
||||
app.use("/api", instanceSettingsRoutes(mockDb as any, pluginWorkers, prepareBackup));
|
||||
app.use(errorHandler);
|
||||
return app;
|
||||
}
|
||||
@@ -973,7 +973,7 @@ describe("instance settings routes", () => {
|
||||
await request(createApp(adminActor)).post("/api/instance/task-drain").send({ purpose: "idle", ttlMs: 60_000 }).expect(200);
|
||||
expect(mockHeartbeatService.computeTaskDrain).toHaveBeenCalledWith({ purpose: "idle", ttlMs: 60_000 });
|
||||
await request(createApp(adminActor)).get(`/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`).expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, undefined, undefined);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined);
|
||||
});
|
||||
|
||||
it.each([{}, { ttlMs: null }, { ttlMs: 1 }, { ttlMs: 300_001 }])("rejects an unbounded or invalid idle TTL: %j", async body => {
|
||||
@@ -1000,7 +1000,7 @@ describe("instance settings routes", () => {
|
||||
const res = await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepSafety=1");
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toEqual({ ...idleStatus, idleSleepSafety: report });
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, undefined, undefined);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined);
|
||||
expect(mockReadIdleSleepSafety.mock.calls[0][1]()).toEqual(idleStatus);
|
||||
});
|
||||
|
||||
@@ -1028,6 +1028,21 @@ describe("instance settings routes", () => {
|
||||
expect(release).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("passes the operator backup hook only through the protected safety report", async () => {
|
||||
const backup = vi.fn(async () => true);
|
||||
mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus);
|
||||
const app = createApp(adminActor, undefined, backup);
|
||||
await request(app).get("/api/instance/task-drain").expect(200);
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
await request(app).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, backup);
|
||||
mockReadIdleSleepSafety.mockClear();
|
||||
await request(createApp(nonAdminActor, undefined, backup)).get("/api/instance/task-drain?idleSleepSafety=1").expect(403);
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
expect(backup).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["company member", nonAdminActor],
|
||||
["agent", { type: "agent", agentId: "agent-1", companyId: "company-1" }],
|
||||
|
||||
+2
-1
@@ -491,6 +491,7 @@ export async function createApp(
|
||||
}): Promise<unknown>;
|
||||
};
|
||||
databaseBackupService?: InstanceDatabaseBackupService;
|
||||
prepareIdleDatabaseBackup?: () => Promise<boolean>;
|
||||
databaseBackupHealth?: InspectDatabaseBackupHealthOptions;
|
||||
deploymentMode: DeploymentMode;
|
||||
deploymentExposure: DeploymentExposure;
|
||||
@@ -836,7 +837,7 @@ export async function createApp(
|
||||
api.use(resourceMembershipRoutes(db));
|
||||
api.use(primaryAgentRoutes(db));
|
||||
api.use(inboxDismissalRoutes(db));
|
||||
api.use(instanceSettingsRoutes(db, workerManager));
|
||||
api.use(instanceSettingsRoutes(db, workerManager, opts.prepareIdleDatabaseBackup));
|
||||
if (opts.databaseBackupService) {
|
||||
api.use(instanceDatabaseBackupRoutes(opts.databaseBackupService));
|
||||
}
|
||||
|
||||
+19
-1
@@ -1,5 +1,6 @@
|
||||
import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork } from "./services/task-admission.js";
|
||||
import { markIdleStartupComplete } from "./services/idle-local-work.js";
|
||||
import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "./services/idle-database-backup.js";
|
||||
import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js";
|
||||
import { createCloudWarmStandby } from "./services/cloud-warm-standby.js";
|
||||
import { subscribeAllCompanyLiveEvents } from "./services/live-events.js";
|
||||
@@ -819,6 +820,8 @@ async function startServerWithDatabaseTeardown(
|
||||
resolve(config.databaseBackupDir, "..", "db-backup-to-s3.failure"),
|
||||
];
|
||||
let databaseBackupInFlight = false;
|
||||
const idleDatabaseBackupEnabled = config.databaseBackupEnabled
|
||||
&& process.env.PAPERCLIP_DB_BACKUP_IDLE_CHECKPOINT_ENABLED === "1";
|
||||
const runServerDatabaseBackup = async (
|
||||
trigger: InstanceDatabaseBackupTrigger,
|
||||
): Promise<InstanceDatabaseBackupRunResult | null> => {
|
||||
@@ -835,19 +838,25 @@ async function startServerWithDatabaseTeardown(
|
||||
const finishIdleBackup = beginIdleTrackedWork();
|
||||
const startedAt = new Date();
|
||||
const startedAtMs = Date.now();
|
||||
const label = trigger === "scheduled" ? "Automatic" : "Manual";
|
||||
const label = trigger === "manual" ? "Manual" : trigger === "idle" ? "Pre-sleep" : "Automatic";
|
||||
try {
|
||||
if (trigger === "idle") await markIdleBackupWakeRequired(config.databaseBackupDir);
|
||||
logger.info({ backupDir: config.databaseBackupDir, trigger }, `${label} database backup starting`);
|
||||
// Read retention from Instance Settings (DB) so changes take effect without restart.
|
||||
const generalSettings = await backupSettingsSvc.getGeneral();
|
||||
const retention = generalSettings.backupRetention;
|
||||
const verifyCheckpoint = trigger === "idle" || existsSync(idleBackupWakeMarker(config.databaseBackupDir));
|
||||
|
||||
const result = await runDatabaseBackup({
|
||||
connectionString: activeDatabaseConnectionString,
|
||||
backupDir: config.databaseBackupDir,
|
||||
retention,
|
||||
filenamePrefix: "paperclip",
|
||||
verifyBeforePrune: verifyCheckpoint ? verifyIdleDatabaseBackup : undefined,
|
||||
});
|
||||
if (verifyCheckpoint) {
|
||||
if (trigger !== "idle") await clearIdleBackupWakeRequired(config.databaseBackupDir);
|
||||
}
|
||||
const finishedAt = new Date();
|
||||
const response: InstanceDatabaseBackupRunResult = {
|
||||
...result,
|
||||
@@ -892,6 +901,9 @@ async function startServerWithDatabaseTeardown(
|
||||
// self-hosted: createApp falls back to its built-in kubernetes-only default.
|
||||
const managedPluginAutoInstall = managedConfig?.plugins.autoInstall ?? null;
|
||||
const app = await createApp(db as any, {
|
||||
prepareIdleDatabaseBackup: idleDatabaseBackupEnabled
|
||||
? async () => (await runServerDatabaseBackup("idle")) !== null
|
||||
: undefined,
|
||||
cloudWarmStandby: isWarmStandby,
|
||||
uiMode,
|
||||
serverPort: listenPort,
|
||||
@@ -1872,6 +1884,12 @@ async function startServerWithDatabaseTeardown(
|
||||
},
|
||||
"Automatic database backups enabled",
|
||||
);
|
||||
// The final pre-sleep dump remains on the volume throughout sleep. Take
|
||||
// a fresh backup on restart before idle eligibility is enabled, including
|
||||
// after rollback of the opt-in flag. Failure retains the marker for retry.
|
||||
if (!isWarmStandby() && existsSync(idleBackupWakeMarker(config.databaseBackupDir))) {
|
||||
await runServerDatabaseBackup("scheduled").catch(() => {});
|
||||
}
|
||||
setInterval(() => {
|
||||
if (isWarmStandby() || isIdleTaskDrainActive()) return;
|
||||
void runServerDatabaseBackup("scheduled").catch(() => {
|
||||
|
||||
@@ -4,7 +4,7 @@ import { forbidden } from "../errors.js";
|
||||
import { isCloudManagedInstance } from "../services/cloud-instance.js";
|
||||
import { assertInstanceAdmin } from "./authz.js";
|
||||
|
||||
export type InstanceDatabaseBackupTrigger = "manual" | "scheduled";
|
||||
export type InstanceDatabaseBackupTrigger = "manual" | "scheduled" | "idle";
|
||||
|
||||
export type InstanceDatabaseBackupRunResult = RunDatabaseBackupResult & {
|
||||
trigger: InstanceDatabaseBackupTrigger;
|
||||
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
} from "../services/cloud-instance.js";
|
||||
import { getHiddenSettings } from "../services/settings-visibility.js";
|
||||
import { readIdleSleepSafety } from "../services/idle-sleep-safety.js";
|
||||
import { readIdleLocalWork } from "../services/idle-local-work.js";
|
||||
import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
|
||||
import { validate } from "../middleware/validate.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
@@ -120,7 +121,7 @@ function withTaskDrainTransition<T>(run: () => Promise<T>): Promise<T> {
|
||||
return turn;
|
||||
}
|
||||
|
||||
export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManager) {
|
||||
export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManager, prepareIdleDatabaseBackup?: () => Promise<boolean>) {
|
||||
const router = Router();
|
||||
const svc = instanceSettingsService(db);
|
||||
const environments = environmentService(db);
|
||||
@@ -305,8 +306,9 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
|
||||
// members may read process counters, but not instance-wide work state.
|
||||
assertCanManageInstanceSettings(req);
|
||||
const idleSleepSafety = await readIdleSleepSafety(db, () => heartbeat.getTaskDrainStatus(), Date.now,
|
||||
typeof req.query.ownerId === "string" ? req.query.ownerId : undefined, undefined,
|
||||
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers));
|
||||
typeof req.query.ownerId === "string" ? req.query.ownerId : undefined,
|
||||
() => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup) }),
|
||||
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup);
|
||||
res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety });
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { constants, createReadStream } from "node:fs";
|
||||
import { mkdir, open, unlink } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { Writable } from "node:stream";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { createGunzip } from "node:zlib";
|
||||
|
||||
// The marker survives process replacement on persistent instance storage.
|
||||
// It contains no credentials or tenant data, and never substitutes for a dump.
|
||||
export const idleBackupWakeMarker = (directory: string) => path.join(directory, ".idle-backup-wake-required");
|
||||
|
||||
async function syncDirectory(directory: string): Promise<void> {
|
||||
const handle = await open(directory, "r");
|
||||
try { await handle.sync(); } finally { await handle.close(); }
|
||||
}
|
||||
|
||||
export async function markIdleBackupWakeRequired(directory: string): Promise<void> {
|
||||
await mkdir(directory, { recursive: true });
|
||||
const marker = await open(idleBackupWakeMarker(directory), constants.O_WRONLY | constants.O_CREAT | constants.O_NOFOLLOW, 0o600);
|
||||
try {
|
||||
await marker.writeFile("Take a fresh database backup after this process restarts.\n");
|
||||
await marker.sync();
|
||||
} finally { await marker.close(); }
|
||||
await syncDirectory(directory);
|
||||
}
|
||||
|
||||
/** Read the entire archive to verify the gzip trailer, then persist its bytes
|
||||
* and directory entry before allowing compute to stop. The backup runner
|
||||
* retains its in-flight receipt until this function settles. */
|
||||
export async function verifyIdleDatabaseBackup(backupFile: string): Promise<void> {
|
||||
const handle = await open(backupFile, constants.O_RDONLY | constants.O_NOFOLLOW);
|
||||
try {
|
||||
const stat = await handle.stat();
|
||||
if (!stat.isFile() || stat.size === 0) throw new Error("Idle backup archive is empty or invalid");
|
||||
let bytes = 0;
|
||||
await pipeline(
|
||||
createReadStream(backupFile, { fd: handle.fd, autoClose: false }),
|
||||
createGunzip(),
|
||||
new Writable({ write(chunk, _encoding, done) { bytes += chunk.length; done(); } }),
|
||||
);
|
||||
if (bytes === 0) throw new Error("Idle backup archive contains no SQL");
|
||||
await handle.sync();
|
||||
} finally { await handle.close(); }
|
||||
await syncDirectory(path.dirname(backupFile));
|
||||
}
|
||||
|
||||
export async function clearIdleBackupWakeRequired(directory: string): Promise<void> {
|
||||
await unlink(idleBackupWakeMarker(directory)).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code !== "ENOENT") throw error;
|
||||
});
|
||||
await syncDirectory(directory);
|
||||
}
|
||||
@@ -33,9 +33,11 @@ export async function inspectIdleSpool(directory: string): Promise<IdleLocalWork
|
||||
}
|
||||
}
|
||||
|
||||
export async function readIdleLocalWork(): Promise<IdleLocalWork> {
|
||||
export async function readIdleLocalWork(options: { backupCheckpoint?: boolean } = {}): Promise<IdleLocalWork> {
|
||||
if (!startupComplete || !ingressTracked) return "unknown";
|
||||
if (scheduledBackups || idleWorkSnapshot().active !== 0) return "present";
|
||||
// This exemption belongs only to a scan which will finish a new backup
|
||||
// under the same admission hold before it can return an empty report.
|
||||
if ((scheduledBackups && !options.backupCheckpoint) || idleWorkSnapshot().active !== 0) return "present";
|
||||
const directories = new Set([...spoolDirectories, idleAccountingSpoolPath(), idleOrphanSpoolPath()]);
|
||||
const results = await Promise.all([...directories].map(inspectIdleSpool));
|
||||
if (results.includes("unknown")) return "unknown";
|
||||
|
||||
@@ -97,12 +97,17 @@ export async function readIdleSleepSafety(
|
||||
ownerId?: string,
|
||||
inspectLocalWork: () => Promise<IdleLocalWork> = readIdleLocalWork,
|
||||
inspectPlugins?: InspectIdlePlugins,
|
||||
prepareBackup?: () => Promise<boolean>,
|
||||
): Promise<IdleSleepSafety> {
|
||||
const unknown: IdleSleepSafety = { version: 1, backgroundWork: "unknown" };
|
||||
const before = getDrainStatus();
|
||||
const localBefore = idleWorkSnapshot();
|
||||
if (!sameQuietHold(before, before, now())) return unknown;
|
||||
try {
|
||||
// Reject known local blockers before querying (and potentially waking)
|
||||
// the tenant database. Keep the final inspection below as a second fence.
|
||||
const localFirst = await inspectLocalWork();
|
||||
if (localFirst !== "none") return { version: 1, backgroundWork: localFirst };
|
||||
let quietPlugins: string[] = [];
|
||||
if (inspectPlugins) {
|
||||
if (!ownerId || ownerId !== before.ownerId || before.expiresAt === null || localBefore.active !== 0) return unknown;
|
||||
@@ -124,11 +129,21 @@ export async function readIdleSleepSafety(
|
||||
if (blocked === undefined || !sameQuietHold(before, getDrainStatus(), now())) return unknown;
|
||||
if (blocked) return { version: 1, backgroundWork: "present" };
|
||||
if (!ownerId || ownerId !== before.ownerId || before.expiresAt === null || localBefore.active !== 0) return unknown;
|
||||
let expectedGeneration = localBefore.generation;
|
||||
if (prepareBackup) {
|
||||
if (!sameQuietHold(before, getDrainStatus(), now()) ||
|
||||
idleWorkSnapshot().generation !== expectedGeneration || idleWorkSnapshot().active !== 0) return unknown;
|
||||
// The runner owns exactly one tracked-work receipt through dump,
|
||||
// archive verification and fsync. Its start/finish are the only work
|
||||
// allowed during this scan. Any concurrent work invalidates the dump.
|
||||
if (!(await prepareBackup())) return unknown;
|
||||
expectedGeneration += 2;
|
||||
}
|
||||
const local = await inspectLocalWork();
|
||||
const after = getDrainStatus();
|
||||
const localAfter = idleWorkSnapshot();
|
||||
if (!sameQuietHold(before, after, now()) || localAfter.active !== 0 ||
|
||||
localBefore.generation !== localAfter.generation) return unknown;
|
||||
expectedGeneration !== localAfter.generation) return unknown;
|
||||
return { version: 1, backgroundWork: local };
|
||||
|
||||
} catch {
|
||||
|
||||
Reference in new issue
Block a user