mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
fix(server): schedule automatic backups through shared lifecycle (#17836)
<!-- entire-trail-link-start --> https://entire.com/gh/paperclipai/-/trails/39/changes/paperclip/39 <!-- entire-trail-link-end --> ## Thinking Path > - Paperclip manages agent work and its durable state. > - Automatic database backups protect that state on a configured schedule. > - Their independent interval is outside the shared deadline and shutdown lifecycle. > - An enabled backup schedule owes real work even when no queue changed. > - The existing coordinator can own this deadline without another table or notification protocol. > - This change preserves backup policy and moves scheduling into that shared lifecycle. ## Linked Issues or Issue Description Refs: #17835. Related work: #11756 (abortable backups), #3549 (stale-backup catch-up), #2120 (interval unref), #16364 (bounded shutdown steps). This slice changes scheduling; it does not implement those separate features. **What existing behavior does this improve?** Automatic database backup scheduling and shutdown ownership. **Current behavior** A standalone interval invokes backups, skips drain and warm standby, and is not registered with the shared work scheduler. Its future deadline is invisible there. The interval is not cleared or joined by the delivery coordinator during shutdown. **Proposed behavior** Register the configured backup deadline with the shared coordinator. Call the existing backup owner only when due. Keep the startup-relative phase, skip missed slots, retain the next interval after failure or a busy-owner skip, and join active scheduled work on shutdown. **Reason and benefit** One scheduling lifecycle owns overlap, admission, deadlines and shutdown. Waiting does not query the database. This is a true periodic obligation, so enabled backups still recur; it is not an empty-queue polling removal or a claim of measured savings. This existing slice now follows the merged import-spool parent #17835 and targets master. Merge authorization and exact-head CI/review gates are tracked separately. ## What Changed - Add a small backup scheduling owner using the existing coordinator and scheduler. - Replace the automatic-backup interval in startup. Register only when backups are enabled, independently of heartbeat scheduling. - Preserve the existing backup executor, manual serialization, retention reads, pre-sleep verification, restart marker and host-sleep gates. - Test deadlines, drain/standby, failures, busy-owner skips, slow attempts, restart and shutdown. Update startup-marker tests to exercise the registered deadline. - Record the audit and update inventory item 15. No schema or backup format changes. ## Verification - Current source acceptance: **130 passing cases in four complete files**, no skips: backup worker, startup checkpoints, scoped deadline work and delivery coordinator. Database build, production server typechecking/build, module boundaries, target migration order and Runner profile identity pass. - Runtime and all four selected test sources are byte-identical to accepted source `11775eee7ed795b989cbb31052e69643c468c3cf` after normal parent reconciliation. The parent carries the tool-access test-order repair; it changes no backup behavior. - Historical broader checks and the real JavaScript backup/restore probe, including their intermediate failures and limits, remain in [the backup audit](doc/plans/2026-10-26-database-backup-sleep-schedule.md). They are not claimed as repeated current-head acceptance. - No full local repository-suite, deployed suspend/wake, pg_dump, live provider or cost-savings claim. Final-head CI, security, Greptile 5/5 and any actionable Ponytail findings remain separate merge gates. ## Risks - Enabled backups still require periodic database and disk work. This change centralizes the deadline; it does not reduce backup frequency or establish deployed cost savings. - The backup engine has no AbortSignal contract. Coordinator stop joins an active scheduled backup, so a stalled backup can delay graceful shutdown. Bounded backup cancellation remains separate work. - Restart still starts a new ordinary interval; it does not replay historical slots. The existing pre-sleep marker still requests an immediate verified restart backup and persists on failure. - A due backup waits during drain or standby and retries on the next coordinator attempt after admission reopens, within one configured interval. - The internal deadline is not a persisted Cloud alarm. Existing scheduled-backup and checkpoint sleep gates remain unchanged. The real probe qualifies the JavaScript engine only, not pg_dump or deployed shutdown latency. ## Model Used OpenAI Codex (GPT-6-based coding agent). Repository inspection, ordinary Git merges, source-composition proofs and local lifecycle tests were used for this reconciliation. No subagents were used in the current merge campaign. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes:` / `Closes:` / `Refs:` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
7 files changed
+362
-15
No files matched your search
@@ -0,0 +1,155 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { registerDatabaseBackupWork } from "../services/database-backup-work.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
|
||||
afterEach(async () => {
|
||||
for (const coordinator of coordinators.splice(0)) await coordinator.stop();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
function setup() {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-10-26T12:00:00Z"));
|
||||
const execute = vi.fn(async () => []);
|
||||
const owner = installDatabaseWorkSignals({ execute, transaction: vi.fn() });
|
||||
const canRun = vi.fn(() => true), canReconcile = vi.fn(() => true), onError = vi.fn();
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner, canRun, canReconcile, onError });
|
||||
coordinators.push(coordinator);
|
||||
const run = vi.fn(async (): Promise<unknown> => ({}));
|
||||
return { coordinator, execute, canRun, canReconcile, onError, run };
|
||||
}
|
||||
const intervalMs = 60_000;
|
||||
|
||||
describe("automatic database backup scheduling", () => {
|
||||
it("waits for the actual backup deadline without querying or holding idle work", async () => {
|
||||
const s = setup(), now = Date.now();
|
||||
const worker = registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run });
|
||||
await worker.ready;
|
||||
expect(s.coordinator.nextWakeAt()).toBe(now + intervalMs);
|
||||
await vi.advanceTimersByTimeAsync(intervalMs - 1);
|
||||
expect(s.run).not.toHaveBeenCalled();
|
||||
expect(s.execute).not.toHaveBeenCalled();
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
expect(s.coordinator.nextWakeAt()).toBe(now + 2 * intervalMs);
|
||||
worker.wake();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
expect(s.execute).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each(["drain", "standby"])("retains a due backup through %s without passive SQL", async mode => {
|
||||
const s = setup();
|
||||
const worker = registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run });
|
||||
await worker.ready;
|
||||
s.canRun.mockReturnValue(false);
|
||||
s.canReconcile.mockReturnValue(mode === "drain");
|
||||
await vi.advanceTimersByTimeAsync(3 * intervalMs);
|
||||
expect(s.run).not.toHaveBeenCalled();
|
||||
expect(s.execute).not.toHaveBeenCalled();
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
s.canRun.mockReturnValue(true);
|
||||
s.canReconcile.mockReturnValue(true);
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
expect(s.coordinator.nextWakeAt()).toBe(Date.now() + intervalMs);
|
||||
});
|
||||
|
||||
it("startup while draining preserves the first due backup", async () => {
|
||||
const s = setup();
|
||||
s.canRun.mockReturnValue(false);
|
||||
await registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run }).ready;
|
||||
await vi.advanceTimersByTimeAsync(2 * intervalMs);
|
||||
expect(s.run).not.toHaveBeenCalled();
|
||||
s.canRun.mockReturnValue(true);
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
expect(s.execute).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reports a failed attempt and retries at the next scheduled slot", async () => {
|
||||
const s = setup(), now = Date.now();
|
||||
s.run.mockRejectedValueOnce(new Error("backup storage unavailable"));
|
||||
await registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run }).ready;
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
expect(s.onError).toHaveBeenCalledOnce();
|
||||
expect(s.coordinator.nextWakeAt()).toBe(now + 2 * intervalMs);
|
||||
s.canRun.mockReturnValue(false);
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
expect(s.execute).not.toHaveBeenCalled();
|
||||
s.canRun.mockReturnValue(true);
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
expect(s.run).toHaveBeenCalledTimes(2);
|
||||
expect(s.coordinator.nextWakeAt()).toBe(now + 4 * intervalMs);
|
||||
});
|
||||
|
||||
it("keeps the schedule after the backup owner skips an overlapping manual backup", async () => {
|
||||
const s = setup();
|
||||
s.run.mockResolvedValueOnce(null);
|
||||
await registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run }).ready;
|
||||
await vi.advanceTimersByTimeAsync(2 * intervalMs);
|
||||
expect(s.run).toHaveBeenCalledTimes(2);
|
||||
expect(s.onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("coalesces wakes during a slow backup and skips missed slots", async () => {
|
||||
const s = setup(), now = Date.now();
|
||||
let finish!: () => void;
|
||||
s.run.mockImplementationOnce(() => new Promise<void>(resolve => { finish = resolve; }));
|
||||
const worker = registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run });
|
||||
await worker.ready;
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
expect(idleWorkSnapshot().active).toBe(1);
|
||||
worker.wake(); worker.wake();
|
||||
await vi.advanceTimersByTimeAsync(2 * intervalMs + 100);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
finish();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
expect(s.coordinator.nextWakeAt()).toBe(now + 4 * intervalMs);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
await vi.advanceTimersByTimeAsync(intervalMs - 101);
|
||||
expect(s.run).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("shutdown waits for the active backup and cannot start another", async () => {
|
||||
const s = setup();
|
||||
let finish!: () => void;
|
||||
s.run.mockImplementationOnce(() => new Promise<void>(resolve => { finish = resolve; }));
|
||||
const worker = registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run });
|
||||
await worker.ready;
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
let stopped = false;
|
||||
const stopping = s.coordinator.stop().then(() => { stopped = true; });
|
||||
worker.wake();
|
||||
await vi.advanceTimersByTimeAsync(3 * intervalMs);
|
||||
expect(stopped).toBe(false);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
finish();
|
||||
await stopping;
|
||||
expect(s.coordinator.nextWakeAt()).toBeNull();
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
await vi.advanceTimersByTimeAsync(3 * intervalMs);
|
||||
expect(s.run).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("restart establishes a fresh interval without replaying historical slots", async () => {
|
||||
const s = setup();
|
||||
await registerDatabaseBackupWork(s.coordinator, { intervalMs, run: s.run }).ready;
|
||||
await vi.advanceTimersByTimeAsync(intervalMs / 2);
|
||||
await s.coordinator.stop();
|
||||
await vi.advanceTimersByTimeAsync(4 * intervalMs);
|
||||
const next = setup();
|
||||
const now = Date.now();
|
||||
await registerDatabaseBackupWork(next.coordinator, { intervalMs, run: next.run }).ready;
|
||||
expect(next.coordinator.nextWakeAt()).toBe(now + intervalMs);
|
||||
expect(next.run).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(intervalMs);
|
||||
expect(next.run).toHaveBeenCalledOnce();
|
||||
expect(s.run).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -62,6 +62,7 @@ const {
|
||||
const deliveryWorkErrors: Array<{ queue: string; error: unknown }> = [];
|
||||
const deliveryWorkRegisterMock = vi.fn((queue: string, task: {
|
||||
retryMs?: number;
|
||||
nextRunAt?: () => number | null;
|
||||
run: (signal?: AbortSignal) => Promise<unknown>;
|
||||
hasPending?: () => Promise<boolean>;
|
||||
}) => ({ ready: task.run(new AbortController().signal).catch(error => { deliveryWorkErrors.push({ queue, error }); }), wake: vi.fn(), stop: vi.fn(async () => {}) }));
|
||||
@@ -800,6 +801,19 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(caseHandoffDeliveryServiceMock.hasPending).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it.each([false, true])("registers automatic backups independently of heartbeat scheduling when enabled=%s", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ databaseBackupEnabled: enabled, heartbeatSchedulerEnabled: false }));
|
||||
await startServer();
|
||||
const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.databaseBackups);
|
||||
expect(Boolean(registration)).toBe(enabled);
|
||||
expect(runDatabaseBackup).not.toHaveBeenCalled();
|
||||
if (registration) {
|
||||
expect(registration[1].nextRunAt!()).toBeGreaterThan(Date.now());
|
||||
await registration[1].run(new AbortController().signal);
|
||||
expect(runDatabaseBackup).not.toHaveBeenCalled();
|
||||
}
|
||||
});
|
||||
|
||||
it.each([false, true].flatMap(checkpointMode => [false, true].map(fails => ({ checkpointMode, fails }))))
|
||||
("finishes restart backup before idle eligibility (checkpoint=$checkpointMode, failure=$fails)", async ({ checkpointMode, fails }) => {
|
||||
const directory = mkdtempSync(path.join(tmpdir(), "startup-idle-backup-"));
|
||||
@@ -807,11 +821,6 @@ describe("startServer feedback export wiring", () => {
|
||||
process.env.PAPERCLIP_DB_BACKUP_IDLE_CHECKPOINT_ENABLED = checkpointMode ? "1" : "0";
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ databaseBackupEnabled: true, databaseBackupDir: directory }));
|
||||
const eligible = vi.spyOn(idleLocalWork, "markIdleStartupComplete");
|
||||
let retry: (() => void) | undefined;
|
||||
const timers = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, ms: number) => {
|
||||
if (ms === 60 * 60 * 1000) retry = callback;
|
||||
return 1 as unknown as ReturnType<typeof setInterval>;
|
||||
}) as typeof setInterval);
|
||||
let finish!: () => void;
|
||||
const pending = new Promise<void>(resolve => { finish = resolve; });
|
||||
const writeBackup = async (options: Parameters<typeof runDatabaseBackup>[0]) => {
|
||||
@@ -843,9 +852,12 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(fails);
|
||||
if (fails) {
|
||||
vi.mocked(runDatabaseBackup).mockImplementationOnce(writeBackup);
|
||||
expect(retry).toEqual(expect.any(Function));
|
||||
const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.databaseBackups);
|
||||
expect(registration).toBeDefined();
|
||||
const activeBeforeRetry = idleWorkSnapshot().active;
|
||||
retry!();
|
||||
const clock = vi.spyOn(Date, "now").mockReturnValue(registration![1].nextRunAt!()!);
|
||||
try { await registration![1].run(new AbortController().signal); }
|
||||
finally { clock.mockRestore(); }
|
||||
await vi.waitFor(() => {
|
||||
expect(existsSync(idleBackupWakeMarker(directory))).toBe(false);
|
||||
// Marker removal precedes directory fsync and release of the work receipt.
|
||||
@@ -856,7 +868,7 @@ describe("startServer feedback export wiring", () => {
|
||||
} finally {
|
||||
finish();
|
||||
await starting;
|
||||
timers.mockRestore(); eligible.mockRestore();
|
||||
eligible.mockRestore();
|
||||
if (previousFlag === undefined) delete process.env.PAPERCLIP_DB_BACKUP_IDLE_CHECKPOINT_ENABLED;
|
||||
else process.env.PAPERCLIP_DB_BACKUP_IDLE_CHECKPOINT_ENABLED = previousFlag;
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
|
||||
+5
-6
@@ -1,3 +1,4 @@
|
||||
import { registerDatabaseBackupWork } from "./services/database-backup-work.js";
|
||||
import { registerToolConnectionHealthWork } from "./services/tool-connection-health-work.js";
|
||||
import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "./services/approval-queue-seeding-work.js";
|
||||
import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "./services/join-request-retention.js";
|
||||
@@ -2127,12 +2128,10 @@ async function startServerWithDatabaseTeardown(
|
||||
if (!isWarmStandby() && existsSync(idleBackupWakeMarker(config.databaseBackupDir))) {
|
||||
await runServerDatabaseBackup("scheduled").catch(() => {});
|
||||
}
|
||||
setInterval(() => {
|
||||
if (isWarmStandby() || isIdleTaskDrainActive()) return;
|
||||
void runServerDatabaseBackup("scheduled").catch(() => {
|
||||
// runServerDatabaseBackup already logs the failure with context.
|
||||
});
|
||||
}, backupIntervalMs);
|
||||
await registerDatabaseBackupWork(deliveryWork, {
|
||||
intervalMs: backupIntervalMs,
|
||||
run: () => runServerDatabaseBackup("scheduled"),
|
||||
}).ready;
|
||||
}
|
||||
|
||||
// Wait for external adapters to finish loading before accepting requests.
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
|
||||
|
||||
export function registerDatabaseBackupWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
options: { intervalMs: number; run: () => Promise<unknown> },
|
||||
) {
|
||||
const { intervalMs } = options;
|
||||
|
||||
|
||||
let nextAt = Date.now() + intervalMs;
|
||||
return coordinator.register(DELIVERY_QUEUES.databaseBackups, {
|
||||
retryMs: intervalMs,
|
||||
pausePassiveMaintenance: true,
|
||||
hasPending: async () => false,
|
||||
nextRunAt: () => nextAt,
|
||||
run: async () => {
|
||||
if (Date.now() < nextAt) return;
|
||||
try {
|
||||
await options.run();
|
||||
} finally {
|
||||
|
||||
|
||||
nextAt += Math.max(1, Math.floor((Date.now() - nextAt) / intervalMs) + 1) * intervalMs;
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
|
||||
/** Topics for existing durable queues; register intent before their writes. */
|
||||
export const DELIVERY_QUEUES = {
|
||||
databaseBackups: "database-backups",
|
||||
importSpoolCleanup: "import-spool-cleanup",
|
||||
externalConversationSettlement: "external-conversation-settlement",
|
||||
nativeRecoveryNotices: "native-recovery-notices",
|
||||
|
||||
Reference in new issue
Block a user