test: qualify assembled maintenance boot and lifecycle

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-30 19:26:52 -05:00
1 parent 5fb3731640
commit d26c7b09ad
8 files changed
+798

No files matched your search

@@ -0,0 +1,116 @@
# Aggregate maintenance interval integration
## Delivery boundary
Remove the catch-all heartbeat maintenance interval only after its domain duties
have replacement owners. Keep existing routine, agent-timer and issue-monitor
scheduling, feature toggles, plugin readiness, environment cleanup, idle tracking,
hot-restart shutdown, and conservative sleep-safety checks. This work does not
qualify actual database or host suspension, Cloud alarms, or cost savings.
The dependency base currently retains the interval while residual run recovery
is being completed. The assembled acceptance harness is implemented first so
that its observation boundary has a positive control before interval removal.
## Duty ownership
| Former interval duty | Owning replacement |
| --- | --- |
| Attention retention | Attention eligibility, committed source notifications and actual archive deadlines |
| Run/task recovery | Run dispatch, issue recovery, watchdogs, locks, residual live/orphan obligations |
| Routine execution projection | Routine execution-status notifications and projection debt |
| GitHub discovery/credentials | Durable supported event ingress, enrollment and credential continuity |
| Merged PR confirmations | Actual unresolved cards, provider ingress and scoped fallback |
| External objects | Supported source/configuration producers, exact refresh and retry deadlines |
| Status cards | Feature-owned scheduled/reactive evaluation, claims and assignment sequencing |
| Native backup expiration | Existing backup/session/file records and source-owned retention deadlines |
| Ephemeral runtime status | Expiry on read, explicit terminal clear and time-gated cleanup on output writes |
The in-memory runtime-status store performs at most one expired-entry scan per
strictly more than 90 seconds of continuing writes. It has no maintenance timer.
An otherwise inactive process can retain its last unread abandoned cohort until
its next read/write or process exit; those entries are never returned after TTL.
## Assembled acceptance boundary
`server/src/__tests__/maintenance-server-routes.test.ts` starts the real exported
`startServer` in a child with its real app, services, coordinator and PostgreSQL.
The suite prepares shared/SDK/Runner TypeScript artifacts, uses the existing
owned embedded PostgreSQL helper by default, and accepts validated disposable
external database overrides for constrained hosts. The file name places this
expensive suite in the existing serialized route lane. It does not skip merely
because an external database environment variable is absent.
The test-only PostgreSQL proxy observes all frontend simple Query messages and
extended Parse/Bind/Execute messages. It counts prepared executions that reuse
an earlier parse, raw tagged SQL, transaction commands and rollback. SQL
parameters and result values are not written to measurement receipts. Receipts
contain fingerprints, static table names and separate protocol frame counts.
The observer issues no database queries and has a real-socket calibration test.
Setup and independent result inspection use another bounded client outside the
application's measured connection boundary.
Child-local diagnostics observe outbound HTTP requests and non-database TCP
connections/bytes. A controlled GitHub HTTP endpoint is explicitly simulated;
its request log independently records provider activity. The Cloud standby case
uses actual managed bootstrap validation, an S3 HTTP guard and generated
Ed25519 signing keys, without a deployed Cloud service or real bucket.
The absent-feature fixture disables telemetry, announcements, scheduled database
backups and UI serving. It supplies no provider secrets or configured external
channels. Other maintenance registrations remain on their production paths.
Those exclusions are fixture conditions, not a claim that configured periodic
backups, live operation keepalives or authorized external reconciliation disappear.
## Clock and settlement
One positive control uses a real 31-second observation after actual startup
settlement. Longer windows use only child-local Date/setTimeout/setInterval
control. Boot and ordinary HTTP producer phases advance at real speed. The
clock freezes after readiness and real socket settlement, then advances at most
30 seconds per step, settling real I/O between steps. Parent deadlines,
setImmediate and database sockets remain usable. Receipts distinguish real
elapsed time from simulated elapsed time. PostgreSQL's server clock is not
advanced; this window does not replace owner-specific durable deadline tests.
The final enabled-mode claim still requires a separate real 90-second quiet
window after interval removal. A fake clock alone is insufficient.
## Evidence and remaining gates
Before removal, empty enabled boot issued 325 startup SQL statements and another
26 executions in the next real 31-second window: 24 extended executions and two
simple queries, covering residual heartbeat/native/resource/wake recovery.
This establishes that the observer sees the legacy interval's actual effect.
The obsolete-positive assertion will be replaced when the interval is removed;
the raw/prepared real-PostgreSQL calibration remains.
Current assembled cases cover:
- Disabled-heartbeat empty boot across two simulated hourly windows.
- Ordinary HTTP company/settings/completed-task creation, independent durable
inspection, drain/resume and two additional hourly windows without owner work.
- An ordinary HTTP issue insertion held by an isolated PostgreSQL lock while
idle drain starts, followed by its actual commit and external-object source
ingestion. Provider work waits during drain, then resume fetches the simulated
merged PR, persists terminal state and settles for two hourly windows.
- A real HTTP 503 from the controlled provider, durable retry deadline, no
SQL/provider retry during drain, successful resume and quiet after settlement.
- Actual Cloud warm standby, signed identity claim and durable canonical identity.
Warm standby retains exactly one identified startup intent: the committed
`companyTransferRunService.recoverStrandedApplyingRuns` notification to import
spool cleanup. The Cloud coordinator deliberately disables both dispatch and
reconciliation before claim, so the hold remains even with zero transfer rows.
The test qualifies SQL/network quiet with that explicit deferred intent, then
requires the ordinary signed claim to release it to zero active work at the
existing bounded 30-second retry deadline. It does not call standby sleep-eligible
while this hold exists. Temporary diagnostic begin/end tracing used to identify
this owner is not part of the production change.
Remaining acceptance before a removal PR: final recovery integration, all owner
first-producer paths through assembled wiring, enabled empty/configured/history
quiet including the real 90-second window, scheduling suppression, startup and
shutdown races, relevant contracts across the combined tree, current-head
Greptile 5/5 and required CI. No aggregate-removal acceptance or deployment claim
is made by the preliminary harness checkpoint.
@@ -0,0 +1,103 @@
import { createHash } from "node:crypto";
import { createConnection, createServer, type Socket } from "node:net";
export type ObservedQuery = { protocol: "simple" | "extended"; sql: string; fingerprint: string };
export function createPostgresProtocolObserver(onQuery: (query: ObservedQuery) => void, onFrame: (type: string) => void = () => {}) {
let frontend = Buffer.alloc(0), backend = Buffer.alloc(0), startup = true, pending = 0;
const statements = new Map<string, string>(), portals = new Map<string, string>();
function cstring(body: Buffer, offset = 0): [string, number] {
const end = body.indexOf(0, offset);
if (end < 0) throw new Error("PostgreSQL protocol string is incomplete");
return [body.toString("utf8", offset, end), end + 1];
}
function observe(protocol: ObservedQuery["protocol"], sql: string) {
onQuery({ protocol, sql, fingerprint: createHash("sha256").update(sql).digest("hex").slice(0, 16) });
}
return {
get pending() { return pending; },
frontend(chunk: Buffer) {
frontend = Buffer.concat([frontend, chunk]);
while (frontend.length >= (startup ? 8 : 5)) {
const length = frontend.readInt32BE(startup ? 0 : 1);
if (length < (startup ? 8 : 4) || length > 64 * 1024 * 1024) throw new Error("Invalid PostgreSQL frame length");
const total = length + (startup ? 0 : 1);
if (frontend.length < total) return;
const frame = frontend.subarray(0, total);
frontend = frontend.subarray(total);
if (startup) {
const protocol = frame.readInt32BE(4);
if (protocol === 196608) startup = false;
else if (protocol !== 80877103 && protocol !== 80877104 && protocol !== 80877102) throw new Error("Unsupported PostgreSQL startup protocol");
continue;
}
const type = String.fromCharCode(frame[0]!), body = frame.subarray(5);
onFrame(type);
if (type === "Q") { pending++; observe("simple", cstring(body)[0]); }
else if (type === "P") {
const [name, offset] = cstring(body);
statements.set(name, cstring(body, offset)[0]);
} else if (type === "B") {
const [portal, offset] = cstring(body);
portals.set(portal, cstring(body, offset)[0]);
} else if (type === "E") {
const portal = cstring(body)[0], statement = portals.get(portal);
if (statement === undefined || !statements.has(statement)) throw new Error("Executed PostgreSQL statement was not observed");
observe("extended", statements.get(statement)!);
} else if (type === "S") pending++;
else if (type === "C") {
const name = cstring(body, 1)[0];
if (body[0] === 83) statements.delete(name);
else portals.delete(name);
}
}
},
backend(chunk: Buffer) {
backend = Buffer.concat([backend, chunk]);
if (startup && backend.length === 1 && backend[0] === 78) { backend = Buffer.alloc(0); return; }
while (backend.length >= 5) {
const length = backend.readInt32BE(1);
if (length < 4 || length > 64 * 1024 * 1024) throw new Error("Unsupported encrypted or invalid PostgreSQL response");
if (backend.length < length + 1) return;
if (backend[0] === 90) pending = Math.max(0, pending - 1);
backend = backend.subarray(length + 1);
}
},
};
}
export async function startMaintenancePostgresProxy(databaseUrl: string) {
const upstream = new URL(databaseUrl), queries: ObservedQuery[] = [], errors: Error[] = [];
const frames = { Query: 0, Parse: 0, Bind: 0, Execute: 0, Sync: 0 };
const sockets = new Set<Socket>(), observers = new Set<ReturnType<typeof createPostgresProtocolObserver>>();
const server = createServer(client => {
const target = createConnection({ host: upstream.hostname, port: Number(upstream.port) || 5432 });
const observer = createPostgresProtocolObserver(query => queries.push(query), type => {
if (type === "Q") frames.Query++; else if (type === "P") frames.Parse++; else if (type === "B") frames.Bind++;
else if (type === "E") frames.Execute++; else if (type === "S") frames.Sync++;
});
observers.add(observer);
for (const socket of [client, target]) {
sockets.add(socket);
socket.on("error", error => { errors.push(error); client.destroy(); target.destroy(); });
socket.on("close", () => { sockets.delete(socket); if (client.destroyed && target.destroyed) observers.delete(observer); });
}
client.on("data", chunk => { try { observer.frontend(chunk); } catch (error) { errors.push(error as Error); client.destroy(); target.destroy(); } });
target.on("data", chunk => { try { observer.backend(chunk); } catch (error) { errors.push(error as Error); client.destroy(); target.destroy(); } });
client.pipe(target); target.pipe(client);
client.on("end", () => target.end()); target.on("end", () => client.end());
});
await new Promise<void>((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); });
const address = server.address();
if (!address || typeof address === "string") throw new Error("PostgreSQL observer failed to bind");
const url = new URL(databaseUrl); url.hostname = "127.0.0.1"; url.port = String(address.port); url.searchParams.set("sslmode", "disable");
return {
url: url.toString(), queries, errors,
get frames() { return { ...frames }; },
get pending() { return [...observers].reduce((sum, observer) => sum + observer.pending, 0); },
async stop() {
for (const socket of sockets) socket.destroy();
await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
},
};
}
@@ -0,0 +1,21 @@
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { fileURLToPath } from "node:url";
let ready: Promise<void> | undefined;
export function prepareMaintenanceRuntime() {
ready ??= (async () => {
const script = process.env.npm_execpath;
const javascript = script && /\.[cm]?js$/i.test(script);
const command = javascript ? process.execPath : script ?? "pnpm";
const prefix = javascript ? [script!] : [];
for (const args of [
["--filter", "@paperclipai/plugin-sdk", "ensure-build-deps"],
["--filter", "@paperclipai/paperclip-runner", "build:typescript"],
]) await promisify(execFile)(command, [...prefix, ...args], {
cwd: fileURLToPath(new URL("../../../../..", import.meta.url)), env: process.env,
timeout: 90_000, maxBuffer: 8 * 1024 * 1024,
});
})();
return ready;
}
@@ -0,0 +1,75 @@
import { channel } from "node:diagnostics_channel";
import { mock } from "node:test";
import type { Socket } from "node:net";
const realTimeout = globalThis.setTimeout, realInterval = globalThis.setInterval, clearRealInterval = globalThis.clearInterval, realNow = Date.now;
const fakeClock = process.env.PAPERCLIP_MAINTENANCE_FAKE_CLOCK === "true";
let bootClock: ReturnType<typeof setInterval> | undefined;
function runClock() {
if (bootClock) clearRealInterval(bootClock);
let previous = realNow();
bootClock = realInterval(() => { const now = realNow(); mock.timers.tick(Math.max(0, now - previous)); previous = now; }, 10);
}
if (fakeClock) {
mock.timers.enable({ apis: ["Date", "setTimeout", "setInterval"], now: realNow() });
runClock();
}
if (process.env.PAPERCLIP_MAINTENANCE_GITHUB_ORIGIN) {
const fetch = globalThis.fetch;
globalThis.fetch = (input, init) => {
const url = new URL(input instanceof Request ? input.url : input.toString());
return fetch(url.origin === "https://api.github.com"
? new URL(url.pathname + url.search, process.env.PAPERCLIP_MAINTENANCE_GITHUB_ORIGIN) : input, init);
};
}
let outgoingHttpRequests = 0;
const sockets: Array<{ socket: Socket; database: boolean }> = [];
channel("net.client.socket").subscribe(message => {
const socket = (message as { socket: Socket }).socket;
socket.once("connect", () => sockets.push({ socket, database: socket.remotePort === Number(new URL(process.env.DATABASE_URL!).port) }));
});
function network() {
return { nonDatabaseConnections: sockets.filter(entry => !entry.database).length,
databaseConnections: sockets.filter(entry => entry.database).length,
nonDatabaseBytesWritten: sockets.filter(entry => !entry.database).reduce((sum, entry) => sum + entry.socket.bytesWritten, 0) };
}
const observeRequest = () => { outgoingHttpRequests++; };
channel("undici:request:create").subscribe(observeRequest);
channel("http.client.request.start").subscribe(observeRequest);
try {
const { startServer } = await import("../../../index.js");
const { idleWorkSnapshot, startTaskDrain, stopTaskDrain } = await import("../../../services/task-admission.js");
const { getStartupRecoveryState } = await import("../../../startup-recovery-state.js");
const server = await startServer();
process.send?.({ kind: "ready", port: server.listenPort });
process.on("message", async (message: { id: number; command: string; ms?: number }) => {
try {
if (message.command === "snapshot" || message.command === "advance" || message.command === "drain" || message.command === "resume" || message.command === "freeze" || message.command === "realtime") {
if (message.command === "freeze" && bootClock) { clearRealInterval(bootClock); bootClock = realInterval(() => mock.timers.tick(0), 10); }
if (message.command === "realtime" && fakeClock) runClock();
if (message.command === "advance") {
if (!fakeClock || typeof message.ms !== "number" || !Number.isFinite(message.ms) || message.ms < 0 || message.ms > 30_000) throw new Error("Invalid maintenance clock advance");
mock.timers.tick(message.ms);
}
if (message.command === "drain") startTaskDrain({ purpose: "idle" });
if (message.command === "resume") stopTaskDrain();
if (fakeClock) mock.timers.tick(0);
process.send?.({ id: message.id, kind: "snapshot", idle: idleWorkSnapshot(), startup: getStartupRecoveryState(), outgoingHttpRequests, network: network(), now: Date.now() });
} else if (message.command === "shutdown") {
if (bootClock) { clearRealInterval(bootClock); bootClock = undefined; }
await server.shutdown();
process.send?.({ id: message.id, kind: "stopped", outgoingHttpRequests });
if (fakeClock) mock.timers.reset();
await new Promise(resolve => realTimeout(resolve, 20));
process.exit(0);
} else throw new Error(`Unknown maintenance harness command: ${message.command}`);
} catch (error) { process.send?.({ id: message.id, kind: "error", error: error instanceof Error ? error.stack : String(error) }); }
});
} catch (error) {
if (bootClock) clearRealInterval(bootClock);
process.send?.({ kind: "error", error: error instanceof Error ? error.stack : String(error) });
process.exitCode = 1;
}
@@ -0,0 +1,138 @@
import { fork } from "node:child_process";
import { randomBytes, randomUUID } from "node:crypto";
import { createWriteStream } from "node:fs";
import { mkdtemp, writeFile } from "node:fs/promises";
import { createServer } from "node:net";
import { tmpdir } from "node:os";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { setTimeout as delay } from "node:timers/promises";
import { startMaintenancePostgresProxy } from "./postgres-proxy.js";
async function withDeadline<T>(promise: Promise<T>, ms: number, message: () => string): Promise<T> {
const controller = new AbortController();
try {
return await Promise.race([promise, delay(ms, undefined, { signal: controller.signal }).then(() => { throw new Error(message()); })]);
} finally { controller.abort(); }
}
type Snapshot = { idle: { active: number; generation: number }; startup: { phase: string }; outgoingHttpRequests: number; network: { nonDatabaseConnections: number; databaseConnections: number; nonDatabaseBytesWritten: number }; now: number };
export async function startMaintenanceServer(databaseUrl: string, options: { enabled?: boolean; clock?: boolean; suppressed?: boolean; githubOrigin?: string; cloud?: { jwks: string; s3Origin: string; standby: boolean } } = {}) {
const proxy = await startMaintenancePostgresProxy(databaseUrl);
const directory = await mkdtemp(path.join(tmpdir(), "paperclip-maintenance-"));
const reservation = createServer();
await new Promise<void>(resolve => reservation.listen(0, "127.0.0.1", resolve));
const address = reservation.address();
if (!address || typeof address === "string") throw new Error("Maintenance HTTP reservation failed");
const configPath = path.join(directory, "config.json");
await writeFile(configPath, JSON.stringify({
$meta: { version: 1, updatedAt: new Date().toISOString(), source: "onboard" },
database: { mode: "postgres", connectionString: proxy.url, backup: { enabled: false, dir: path.join(directory, "backups") } },
logging: { mode: "file", logDir: path.join(directory, "logs") },
server: { deploymentMode: "local_trusted", exposure: "private", bind: "loopback", host: "127.0.0.1", port: address.port, serveUi: false },
storage: options.cloud ? { provider: "s3", s3: { bucket: "maintenance-test", region: "us-east-1", endpoint: options.cloud.s3Origin, forcePathStyle: true } }
: { provider: "local_disk", localDisk: { baseDir: path.join(directory, "storage") } },
secrets: { provider: "local_encrypted", strictMode: false, localEncrypted: { keyFilePath: path.join(directory, "master.key") } },
telemetry: { enabled: false }, updates: { checkEnabled: false },
}), { mode: 0o600 });
await new Promise<void>(resolve => reservation.close(() => resolve()));
const child = fork(fileURLToPath(new URL("./server-child.ts", import.meta.url)), [], {
cwd: fileURLToPath(new URL("../../../../..", import.meta.url)),
execArgv: ["--import", fileURLToPath(new URL("../../../../node_modules/tsx/dist/loader.mjs", import.meta.url))],
env: {
PATH: process.env.PATH, TMPDIR: process.env.TMPDIR, NODE_ENV: "production",
PAPERCLIP_HOME: directory, PAPERCLIP_CONFIG: configPath, PAPERCLIP_INSTANCE_ID: "maintenance-test",
DATABASE_URL: proxy.url, DATABASE_POOL_MAX: "6", DATABASE_APPLICATION_NAME: "maintenance-integration-test",
PAPERCLIP_AGENT_JWT_SECRET: randomUUID(), PAPERCLIP_DECISION_SIGNING_SECRET: randomUUID(),
PAPERCLIP_TOOL_ACTION_SIGNING_SECRET: randomUUID(), BETTER_AUTH_SECRET: randomUUID(),
CODEX_HOME: path.join(directory, "codex"), PAPERCLIP_ANNOUNCEMENTS_ENABLED: "false",
PAPERCLIP_DB_BACKUP_ENABLED: "false", PAPERCLIP_UI_DEV_MIDDLEWARE: "false",
HEARTBEAT_SCHEDULER_ENABLED: String(options.enabled ?? true),
PAPERCLIP_MAINTENANCE_FAKE_CLOCK: String(options.clock ?? false),
...(options.suppressed ? { PAPERCLIP_IN_WORKTREE: "true" } : {}),
...(options.githubOrigin ? { PAPERCLIP_MAINTENANCE_GITHUB_ORIGIN: options.githubOrigin } : {}),
...(options.cloud ? {
PAPERCLIP_MANAGED_CONFIG: JSON.stringify({ v: 1, mode: "cloud", catalogVersion: "2026.720.0", features: {}, plugins: { autoInstall: [] } }),
PAPERCLIP_SECRETS_MASTER_KEY: randomBytes(32).toString("hex"),
PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "maintenance-tenant-token", PAPERCLIP_CLOUD_STACK_ID: "maintenance-stack",
PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: options.cloud.jwks, PAPERCLIP_CLOUD_WARM_STANDBY: options.cloud.standby ? "1" : "0",
PAPERCLIP_PUBLIC_URL: "https://pool-maintenance.staging.paperclip.app", PAPERCLIP_PRIMARY_HOST: "pool-maintenance.staging.paperclip.app",
AWS_ACCESS_KEY_ID: "maintenance-fixture", AWS_SECRET_ACCESS_KEY: "maintenance-fixture", AWS_EC2_METADATA_DISABLED: "true",
} : {}),
PAPERCLIP_MIGRATION_PROMPT: "never", PAPERCLIP_MIGRATION_AUTO_APPLY: "true", PAPERCLIP_MANAGED_RUNTIME_HTTPS: "off",
},
stdio: ["ignore", "pipe", "pipe", "ipc"],
});
let output = "", requestId = 0;
const log = createWriteStream(path.join(directory, "server.log"));
child.stdout?.on("data", chunk => { output += String(chunk); log.write(chunk); });
child.stderr?.on("data", chunk => { output += String(chunk); log.write(chunk); });
const requests = new Map<number, { resolve: (value: any) => void; reject: (error: Error) => void }>();
let readyResolve: (port: number) => void, readyReject: (error: Error) => void;
const ready = new Promise<number>((resolve, reject) => { readyResolve = resolve; readyReject = reject; });
const exited = new Promise<void>(resolve => child.once("exit", () => resolve()));
child.on("message", (message: any) => {
if (message.kind === "ready") readyResolve(message.port);
else if (message.id) {
const request = requests.get(message.id); requests.delete(message.id);
if (message.kind === "error") request?.reject(new Error(message.error)); else request?.resolve(message);
} else if (message.kind === "error") readyReject(new Error(message.error));
});
child.once("error", error => readyReject(error));
child.once("exit", (code, signal) => {
const error = new Error(`Maintenance child exited (${code}/${signal}): ${output.slice(-12_000)}`);
readyReject(error);
for (const request of requests.values()) request.reject(error);
requests.clear();
});
async function request<T>(command: string, payload: { ms?: number } = {}): Promise<T> {
const id = ++requestId;
const result = new Promise<T>((resolve, reject) => requests.set(id, { resolve, reject }));
child.send({ id, command, ...payload });
try { return await withDeadline(result, 30_000, () => `Maintenance child ${command} timed out: ${output.slice(-8_000)}`); }
finally { requests.delete(id); }
}
async function stop() {
try {
if (child.exitCode === null && child.signalCode === null && child.connected) await request("shutdown");
await withDeadline(exited, 30_000, () => "Maintenance child did not exit after shutdown");
} finally {
if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
await proxy.stop();
await new Promise<void>(resolve => log.end(resolve));
}
}
try {
const port = await withDeadline(ready, 90_000, () => `Maintenance boot timed out: ${output.slice(-12_000)}`);
return {
proxy, directory, port, stop,
async api<T>(method: string, route: string, body?: unknown): Promise<T> {
if (options.clock) await request("realtime");
try {
const response = await fetch(`http://127.0.0.1:${port}/api${route}`, { method, signal: AbortSignal.timeout(30_000),
headers: { "Content-Type": "application/json" }, ...(body === undefined ? {} : { body: JSON.stringify(body) }) });
const value = await response.json();
if (!response.ok) throw new Error(`Maintenance API ${method} ${route}: ${response.status} ${JSON.stringify(value)}`);
return value as T;
} finally { if (options.clock) await request("freeze"); }
},
snapshot: () => request<Snapshot>("snapshot"),
drain: () => request<Snapshot>("drain"), resume: () => request<Snapshot>("resume"),
advance: (ms: number) => request<Snapshot>("advance", { ms }), freeze: () => request<Snapshot>("freeze"),
get output() { return output; },
async settled(quietMs = 500, deferredStartupHolds = 0) {
let previous = -1, quietSince = Date.now();
for (const deadline = Date.now() + 30_000; Date.now() < deadline;) {
const snapshot = await request<Snapshot>("snapshot");
if (proxy.errors.length) throw new AggregateError(proxy.errors, "PostgreSQL observer failed");
if (snapshot.idle.active !== deferredStartupHolds || proxy.pending !== 0 || proxy.queries.length !== previous) quietSince = Date.now();
previous = proxy.queries.length;
if (snapshot.idle.active === deferredStartupHolds && proxy.pending === 0 && Date.now() - quietSince >= quietMs) return snapshot;
await delay(25);
}
throw new Error(`Maintenance did not settle: ${JSON.stringify({ queries: proxy.queries.length, pending: proxy.pending, snapshot: await request("snapshot") })}\n${output.slice(-12_000)}`);
},
};
} catch (error) { await stop().catch(() => {}); throw error; }
}
@@ -0,0 +1,8 @@
{
"extends": "../../../../../tsconfig.base.json",
"compilerOptions": {
"noEmit": true,
"rootDir": "../../../../.."
},
"include": ["*.ts", "../../maintenance-*.test.ts", "../../../types/express.d.ts"]
}
@@ -0,0 +1,75 @@
import { createDb } from "@paperclipai/db";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { createPostgresProtocolObserver, startMaintenancePostgresProxy, type ObservedQuery } from "./helpers/maintenance/postgres-proxy.js";
const startup = Buffer.from([0, 0, 0, 8, 0, 3, 0, 0]);
function frame(type: string, body: Buffer | string = "") {
const payload = Buffer.from(body), header = Buffer.alloc(5);
header[0] = type.charCodeAt(0); header.writeInt32BE(payload.length + 4, 1);
return Buffer.concat([header, payload]);
}
describe("passive maintenance PostgreSQL observer", () => {
it("counts fragmented simple and reused prepared executions without counting parses or parameter values", () => {
const queries: ObservedQuery[] = [];
const observer = createPostgresProtocolObserver(query => queries.push(query));
const bytes = Buffer.concat([startup,
frame("Q", "select 1\0"),
frame("P", "statement\0select $1::text\0"),
frame("B", "portal\0statement\0parameter-value-sentinel"),
frame("E", Buffer.concat([Buffer.from("portal\0"), Buffer.alloc(4)])), frame("S"),
frame("B", "portal2\0statement\0other-value-sentinel"),
frame("E", Buffer.concat([Buffer.from("portal2\0"), Buffer.alloc(4)])), frame("S"),
]);
for (let index = 0; index < bytes.length; index += 3) observer.frontend(bytes.subarray(index, index + 3));
expect(queries.map(query => [query.protocol, query.sql])).toEqual([
["simple", "select 1"], ["extended", "select $1::text"], ["extended", "select $1::text"],
]);
expect(queries[1]!.fingerprint).toBe(queries[2]!.fingerprint);
expect(JSON.stringify(queries)).not.toContain("value-sentinel");
expect(observer.pending).toBe(3);
const ready = frame("Z", "I");
for (const byte of Buffer.concat([ready, ready, ready])) observer.backend(Buffer.from([byte]));
expect(observer.pending).toBe(0);
});
it("rejects unobserved prepared executions instead of silently claiming quiet", () => {
const observer = createPostgresProtocolObserver(() => {});
observer.frontend(startup);
expect(() => observer.frontend(frame("E", Buffer.from("missing\0\0\0\0\0")))).toThrow("not observed");
});
});
let databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_MAINTENANCE_TEST_DATABASE_URL");
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
describe("maintenance observer against PostgreSQL", () => {
beforeAll(async () => {
if (!databaseUrl) { temporary = await startEmbeddedPostgresTestDatabase("maintenance-observer-"); databaseUrl = temporary.connectionString; }
}, 90_000);
afterAll(async () => { await temporary?.cleanup(); });
it("observes raw tagged SQL, simple SQL, rollback and reused prepared executions with real sockets", async () => {
const proxy = await startMaintenancePostgresProxy(databaseUrl!);
const db = createDb(proxy.url, { prepare: true, maxConnections: 1 });
try {
const tagged = (value: number) => db.$client`select ${value}::integer as value`;
expect((await tagged(1))[0]!.value).toBe(1);
const first = proxy.queries.length, frames = proxy.frames;
expect((await tagged(2))[0]!.value).toBe(2);
expect(proxy.queries.length).toBe(first + 1);
expect(proxy.frames.Parse).toBe(frames.Parse);
expect(proxy.frames.Execute).toBe(frames.Execute + 1);
expect(proxy.queries.at(-1)!.fingerprint).toBe(proxy.queries.at(-2)!.fingerprint);
await db.$client.unsafe("select 3 as value").simple();
await expect(db.$client.begin(async tx => { await tx`select 4`; throw new Error("rollback control"); })).rejects.toThrow("rollback control");
expect(proxy.queries.some(query => query.protocol === "simple" && query.sql.includes("select 3"))).toBe(true);
expect(proxy.queries.some(query => /rollback/i.test(query.sql))).toBe(true);
expect(proxy.pending).toBe(0);
expect(proxy.errors).toEqual([]);
const before = proxy.queries.length;
await new Promise(resolve => setTimeout(resolve, 100));
expect(proxy.queries.length).toBe(before);
} finally { await db.$client.end(); await proxy.stop(); }
});
});
@@ -0,0 +1,262 @@
import { generateKeyPairSync, sign } from "node:crypto";
import { createServer } from "node:http";
import { createDb } from "@paperclipai/db";
import { writeFile } from "node:fs/promises";
import path from "node:path";
import { setTimeout as delay } from "node:timers/promises";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { CLOUD_RUNTIME_IDENTITY_AUDIENCE, CLOUD_RUNTIME_IDENTITY_ISSUER, CLOUD_RUNTIME_IDENTITY_JWS_TYPE } from "../services/cloud-runtime-identity.js";
import { prepareMaintenanceRuntime } from "./helpers/maintenance/runtime-build.js";
import { startMaintenanceServer } from "./helpers/maintenance/server-harness.js";
let databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_MAINTENANCE_BOOT_TEST_DATABASE_URL");
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
async function clearDatabase() {
if (!temporary && !new URL(databaseUrl!).pathname.endsWith("_test")) throw new Error("Maintenance boot requires an owned disposable _test database");
const fixture = createDb(databaseUrl!, { maxConnections: 1 });
try {
const tables = await fixture.$client<{ tablename: string }[]>`select tablename from pg_tables where schemaname='public'`;
if (tables.length) await fixture.$client.unsafe(`truncate ${tables.map(table => `"${table.tablename.replaceAll('"', '""')}"`).join(",")} cascade`);
} finally { await fixture.$client.end(); }
}
function receipt(queries: Array<{ protocol: string; sql: string; fingerprint: string }>) {
const groups = new Map<string, { fingerprint: string; protocol: string; calls: number; tables: string[] }>();
for (const query of queries) {
const key = `${query.protocol}:${query.fingerprint}`;
const entry = groups.get(key) ?? { fingerprint: query.fingerprint, protocol: query.protocol, calls: 0,
tables: [...new Set([...query.sql.matchAll(/(?:from|join|update|into)\s+"?([a-z_][a-z_0-9]*)"?/gi)].map(match => match[1]!))] };
entry.calls++; groups.set(key, entry);
}
return [...groups.values()];
}
describe("assembled maintenance server production wiring", () => {
beforeAll(async () => {
await prepareMaintenanceRuntime();
if (!databaseUrl) { temporary = await startEmbeddedPostgresTestDatabase("maintenance-server-"); databaseUrl = temporary.connectionString; }
}, 150_000);
afterAll(async () => { await temporary?.cleanup(); });
it("detects legacy aggregate recovery SQL after an empty real server boot settles", async () => {
await clearDatabase();
const server = await startMaintenanceServer(databaseUrl!);
try {
const settled = await server.settled();
expect(settled.startup.phase).toBe("ready");
expect(settled.outgoingHttpRequests).toBe(0);
const baseline = server.proxy.queries.length, initialFrames = server.proxy.frames;
await delay(31_000);
await server.settled();
const repeated = server.proxy.queries.slice(baseline);
expect(repeated.length).toBeGreaterThan(0);
expect(repeated.some(query => query.sql.includes('"heartbeat_runs"'))).toBe(true);
expect(server.proxy.errors).toEqual([]);
expect((await server.snapshot()).outgoingHttpRequests).toBe(0);
await writeFile(path.join(server.directory, "measurement.json"), JSON.stringify({ phase: "legacy-positive-control",
observationMs: 31_000, startupQueries: baseline, startup: receipt(server.proxy.queries.slice(0, baseline)),
repeatedQueries: repeated.length, repeated: receipt(repeated), initialFrames, finalFrames: server.proxy.frames, outgoingHttpRequests: 0 }, null, 2));
} finally { await server.stop(); }
}, 150_000);
it("keeps the disabled-heartbeat boot quiet across two former hourly windows with a qualified child clock", async () => {
await clearDatabase();
const server = await startMaintenanceServer(databaseUrl!, { enabled: false, clock: true });
try {
await server.settled();
await server.freeze();
const baseline = server.proxy.queries.length, start = (await server.snapshot()).now;
for (let step = 0; step < 240; step++) {
await server.advance(30_000);
await server.settled(10);
}
const after = await server.snapshot();
expect(after.now - start).toBe(2 * 60 * 60_000);
expect(server.proxy.queries.length).toBe(baseline);
expect(after.outgoingHttpRequests).toBe(0);
expect(after.network.nonDatabaseConnections).toBe(0);
expect(server.proxy.errors).toEqual([]);
await writeFile(path.join(server.directory, "measurement.json"), JSON.stringify({ phase: "disabled-empty-boot-clock",
clockAdvanceMs: after.now - start, stepMs: 30_000, ioSettledBetweenSteps: true,
startupQueries: baseline, startup: receipt(server.proxy.queries), frames: server.proxy.frames, network: after.network, measuredQueries: 0, outgoingHttpRequests: 0 }, null, 2));
const company = await server.api<{ id: string }>("POST", "/companies", { name: "Configured idle maintenance fixture" });
await server.api("PATCH", "/instance/settings/experimental", { enableExternalObjects: true, enableStatusCards: true });
const history = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { title: "Retained completed history", status: "done" });
await server.advance(30_000);
await server.settled();
const persisted = createDb(databaseUrl!, { maxConnections: 1 });
try {
const [saved] = await persisted.$client`select company_id, status from issues where id=${history.id}`;
expect(saved).toMatchObject({ company_id: company.id, status: "done" });
} finally { await persisted.$client.end(); }
const historyBaseline = server.proxy.queries.length;
await server.drain();
for (let step = 0; step < 3; step++) { await server.advance(30_000); await server.settled(10); }
await server.resume(); await server.settled();
const resumed = server.proxy.queries.length;
for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10); }
expect(server.proxy.queries.length).toBe(resumed);
expect((await server.snapshot()).outgoingHttpRequests).toBe(0);
await writeFile(path.join(server.directory, "history-measurement.json"), JSON.stringify({ phase: "disabled-configured-history-drain",
companyId: company.id, issueId: history.id, historyBaseline, resumedQueries: resumed,
postResumeClockMs: 2 * 60 * 60_000, measuredQueries: server.proxy.queries.length - resumed,
frames: server.proxy.frames, outgoingHttpRequests: 0 }, null, 2));
} finally { await server.stop(); }
}, 150_000);
it("processes the first ordinary external-object producer committed during idle drain and becomes quiet again", async () => {
await clearDatabase();
const providerRequests: string[] = [];
const provider = createServer((req, res) => {
providerRequests.push(req.url!);
res.setHeader("Content-Type", "application/json");
if (!/^\/repos\/maintenance\/example\/pulls\/[12]$/.test(req.url!)) { res.statusCode = 500; res.end(JSON.stringify({ error: "Unexpected simulated GitHub request" })); return; }
if (req.url!.endsWith("/2") && providerRequests.filter(url => url.endsWith("/2")).length === 1) {
res.statusCode = 503; res.setHeader("Retry-After", "2"); res.end(JSON.stringify({ message: "Controlled provider outage" })); return;
}
const number = Number(req.url!.split("/").at(-1));
res.end(JSON.stringify({ id: number, number, state: "closed", merged: true, merged_at: new Date().toISOString(),
title: "Controlled merged pull request", html_url: `https://github.com/maintenance/example/pull/${number}`, user: { login: "fixture" } }));
});
await new Promise<void>(resolve => provider.listen(0, "127.0.0.1", resolve));
const providerAddress = provider.address();
if (!providerAddress || typeof providerAddress === "string") throw new Error("Simulated GitHub failed to bind");
const server = await startMaintenanceServer(databaseUrl!, { enabled: false, clock: true,
githubOrigin: `http://127.0.0.1:${providerAddress.port}` });
const fixture = createDb(databaseUrl!, { maxConnections: 1 });
let reserved: Awaited<ReturnType<typeof fixture.$client.reserve>> | undefined;
let pendingIssue: Promise<{ id: string }> | undefined;
try {
await server.settled();
const company = await server.api<{ id: string }>("POST", "/companies", { name: "Drain activation" });
await server.api("PATCH", "/instance/settings/experimental", { enableExternalObjects: true });
await server.advance(30_000); await server.settled();
await fixture.$client.unsafe(`create or replace function maintenance_hold_issue() returns trigger language plpgsql as $$begin
if new.title = 'First write during idle drain' then perform pg_advisory_xact_lock(913701); end if;
return new; end$$`);
await fixture.$client.unsafe(`create trigger maintenance_hold_issue before insert on issues for each row execute function maintenance_hold_issue()`);
reserved = await fixture.$client.reserve();
await reserved`select pg_advisory_lock(913701)`;
const before = server.proxy.queries.length;
pendingIssue = server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, {
title: "First write during idle drain", description: "https://github.com/maintenance/example/pull/1", status: "done",
});
void pendingIssue.catch(() => {});
await expect.poll(() => server.proxy.queries.slice(before).some(query => /insert into "issues"/.test(query.sql)), { timeout: 10_000 }).toBe(true);
await server.drain();
await reserved`select pg_advisory_unlock(913701)`; reserved.release(); reserved = undefined;
const issue = await pendingIssue;
await server.settled();
expect(providerRequests).toEqual([]);
const [waiting] = await fixture.$client`select eo.company_id, eo.is_terminal, eo.last_resolved_at from external_objects eo
join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${issue.id}`;
expect(waiting).toMatchObject({ company_id: company.id, is_terminal: false, last_resolved_at: null });
const draining = server.proxy.queries.length;
for (let step = 0; step < 4; step++) { await server.advance(30_000); await server.settled(10); }
expect(server.proxy.queries.length).toBe(draining);
await server.resume(); await server.advance(30_000); await server.settled();
const [resolved] = await fixture.$client`select eo.company_id, eo.is_terminal, eo.status_key, floor(extract(epoch from eo.last_resolved_at) * 1000)::float8 as resolved_ms from external_objects eo
join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${issue.id}`;
expect(resolved).toMatchObject({ company_id: company.id, is_terminal: true, status_key: "merged", resolved_ms: expect.any(Number) });
expect(resolved!.resolved_ms).toBeGreaterThan(0);
expect(resolved!.resolved_ms).toBeLessThanOrEqual((await server.snapshot()).now);
expect(providerRequests).toEqual(["/repos/maintenance/example/pulls/1"]);
const settledQueries = server.proxy.queries.length, network = await server.snapshot();
for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10); }
expect(server.proxy.queries.length).toBe(settledQueries);
expect(providerRequests).toHaveLength(1);
expect((await server.snapshot()).outgoingHttpRequests).toBe(network.outgoingHttpRequests);
await writeFile(path.join(server.directory, "external-object-measurement.json"), JSON.stringify({
phase: "ordinary-issue-producer-commit-during-drain", providerBoundary: "simulated GitHub HTTP; real application/database/transactions",
companyId: company.id, issueId: issue.id, resolved, drainedClockMs: 120_000, drainedQueries: 0,
postResolveClockMs: 7_200_000, postResolveQueries: 0, providerRequests, frames: server.proxy.frames, network,
}, null, 2));
const retryIssue = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, {
title: "Retry after idle drain", description: "https://github.com/maintenance/example/pull/2", status: "done",
});
await server.advance(1_000); await server.settled(100);
const [failed] = await fixture.$client`select eo.last_error_code, floor(extract(epoch from eo.next_refresh_at) * 1000)::float8 as next_ms
from external_objects eo join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${retryIssue.id}`;
expect(failed!.last_error_code).toBe("github_unreachable");
expect(failed!.next_ms).toBeGreaterThan((await server.snapshot()).now);
await server.drain(); await server.settled();
const retryDrainQueries = server.proxy.queries.length;
for (let step = 0; step < 4; step++) { await server.advance(30_000); await server.settled(10); }
expect(server.proxy.queries.length).toBe(retryDrainQueries);
expect(providerRequests.filter(url => url.endsWith("/2"))).toHaveLength(1);
await server.resume(); await server.advance(30_000); await server.settled();
const [retried] = await fixture.$client`select eo.is_terminal, eo.status_key, eo.last_error_code from external_objects eo
join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${retryIssue.id}`;
expect(retried).toMatchObject({ is_terminal: true, status_key: "merged", last_error_code: null });
expect(providerRequests.filter(url => url.endsWith("/2"))).toHaveLength(2);
const retrySettled = server.proxy.queries.length;
for (let step = 0; step < 20; step++) { await server.advance(30_000); await server.settled(10); }
expect(server.proxy.queries.length).toBe(retrySettled);
await writeFile(path.join(server.directory, "retry-measurement.json"), JSON.stringify({ phase: "provider-failure-retry-across-drain",
providerBoundary: "simulated GitHub HTTP 503 then merged", retryIssueId: retryIssue.id, failed, retried,
drainedClockMs: 120_000, drainedQueries: 0, quietAfterRetryMs: 600_000, quietAfterRetryQueries: 0,
providerRequests, frames: server.proxy.frames }, null, 2));
} finally {
if (reserved) { await reserved`select pg_advisory_unlock(913701)`.catch(() => {}); reserved.release(); }
await pendingIssue?.catch(() => {});
await fixture.$client.unsafe("drop trigger if exists maintenance_hold_issue on issues").catch(() => {});
await fixture.$client.unsafe("drop function if exists maintenance_hold_issue()").catch(() => {});
await fixture.$client.end(); await server.stop();
provider.closeAllConnections(); await new Promise<void>(resolve => provider.close(() => resolve()));
}
}, 150_000);
it("keeps an actual Cloud warm standby quiet and activates after a normally signed durable claim", async () => {
await clearDatabase();
const pair = generateKeyPairSync("ed25519");
const publicJwk = { ...pair.publicKey.export({ format: "jwk" }), kid: "maintenance-test-key", use: "sig", alg: "EdDSA" };
const storageRequests: string[] = [];
const storage = createServer((req, res) => { storageRequests.push(req.url!); res.statusCode = 500; res.end("Unexpected S3 request in empty standby fixture"); });
await new Promise<void>(resolve => storage.listen(0, "127.0.0.1", resolve));
const address = storage.address();
if (!address || typeof address === "string") throw new Error("Simulated S3 failed to bind");
const server = await startMaintenanceServer(databaseUrl!, { clock: true,
cloud: { jwks: JSON.stringify({ keys: [publicJwk] }), s3Origin: `http://127.0.0.1:${address.port}`, standby: true } })
.catch(async error => { storage.closeAllConnections(); await new Promise<void>(resolve => storage.close(() => resolve())); throw error; });
try {
const standby = await server.settled(500, 1); await server.freeze();
expect(standby.idle.active).toBe(1);
const before = server.proxy.queries.length;
for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10, 1); }
const health = await fetch(`http://127.0.0.1:${server.port}/api/health`, { signal: AbortSignal.timeout(10_000) });
expect(health.status).toBe(200); expect(await health.json()).toMatchObject({ warmStandby: true });
expect(server.proxy.queries.length).toBe(before);
expect(storageRequests).toEqual([]);
expect((await server.snapshot()).network.nonDatabaseConnections).toBe(0);
const now = Math.floor((await server.snapshot()).now / 1000);
const header = Buffer.from(JSON.stringify({ alg: "EdDSA", typ: CLOUD_RUNTIME_IDENTITY_JWS_TYPE, kid: publicJwk.kid })).toString("base64url");
const payload = Buffer.from(JSON.stringify({ v: 1, iss: CLOUD_RUNTIME_IDENTITY_ISSUER, aud: CLOUD_RUNTIME_IDENTITY_AUDIENCE,
sub: "maintenance-stack", claimId: "maintenance-claim", previousOrigin: "https://pool-maintenance.staging.paperclip.app",
canonicalOrigin: "https://claimed-maintenance.staging.paperclip.app", stackSlug: "claimed-maintenance", iat: now, exp: now + 300 })).toString("base64url");
const assertion = `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`, "ascii"), pair.privateKey).toString("base64url")}`;
const claimed = await fetch(`http://127.0.0.1:${server.port}/api/health`, {
headers: { "x-paperclip-cloud-runtime-identity": assertion }, signal: AbortSignal.timeout(10_000),
});
expect(claimed.status).toBe(200); expect((await claimed.json()).warmStandby).toBeUndefined();
await server.advance(30_000); const afterClaim = await server.settled();
expect(afterClaim.idle.active).toBe(0);
expect(server.proxy.queries.length).toBeGreaterThan(before);
const fixture = createDb(databaseUrl!, { maxConnections: 1 });
try {
const [transfers] = await fixture.$client`select count(*)::integer as count from company_transfer_runs`;
expect(transfers!.count).toBe(0);
const [identity] = await fixture.$client`select general from instance_settings where singleton_key='cloud-runtime-identity/v1'`;
expect(identity!.general).toMatchObject({ stackId: "maintenance-stack", claimId: "maintenance-claim", canonicalOrigin: "https://claimed-maintenance.staging.paperclip.app" });
} finally { await fixture.$client.end(); }
await writeFile(path.join(server.directory, "standby-measurement.json"), JSON.stringify({ phase: "real-cloud-standby-signed-claim",
cloudBoundary: "actual managed configuration, local S3 HTTP guard, generated Ed25519 trust root; no production Cloud claim",
clockMs: 7_200_000, standbyQueries: 0,
deferredStartupIntent: { count: 1, queue: "import-spool-cleanup", producer: "companyTransferRunService.recoverStrandedApplyingRuns",
reason: "Cloud standby disables both reconciliation and dispatch", releasedAfterSignedClaim: true, boundedResumeMs: 30_000, sleepEligibleWhileHeld: false }, storageRequests, beforeClaimQueries: before, afterClaimQueries: server.proxy.queries.length,
frames: server.proxy.frames, network: await server.snapshot() }, null, 2));
} finally {
await server.stop(); storage.closeAllConnections(); await new Promise<void>(resolve => storage.close(() => resolve()));
}
}, 150_000);
});