diff --git a/server/src/__tests__/helpers/maintenance/postgres-proxy.ts b/server/src/__tests__/helpers/maintenance/postgres-proxy.ts index 5c96d0a3afc..d80aa2fc689 100644 --- a/server/src/__tests__/helpers/maintenance/postgres-proxy.ts +++ b/server/src/__tests__/helpers/maintenance/postgres-proxy.ts @@ -4,7 +4,7 @@ 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; + let frontend = Buffer.alloc(0), backend = Buffer.alloc(0), startup = true, ready = false, pending = 1; const statements = new Map(), portals = new Map(); function cstring(body: Buffer, offset = 0): [string, number] { const end = body.indexOf(0, offset); @@ -16,11 +16,12 @@ export function createPostgresProtocolObserver(onQuery: (query: ObservedQuery) = } return { get pending() { return pending; }, + get ready() { return ready; }, 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"); + if (length < (startup ? 8 : 4) || length > 64 * 1024 * 1024) throw new Error(`Invalid PostgreSQL frame length: ${JSON.stringify({ length, startup, bufferedBytes: frontend.length, header: frontend.subarray(0, startup ? 8 : 5).toString("hex") })}`); const total = length + (startup ? 0 : 1); if (frontend.length < total) return; const frame = frontend.subarray(0, total); @@ -59,7 +60,7 @@ export function createPostgresProtocolObserver(onQuery: (query: ObservedQuery) = 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); + if (backend[0] === 90) { ready = true; pending = Math.max(0, pending - 1); } backend = backend.subarray(length + 1); } }, @@ -99,6 +100,7 @@ export async function startMaintenancePostgresProxy(databaseUrl: string) { get frames() { return { ...frames }; }, get transport() { return { ...transport }; }, get pending() { return [...observers].reduce((sum, observer) => sum + observer.pending, 0); }, + get starting() { return [...observers].filter(observer => !observer.ready).length; }, async stop() { for (const socket of sockets) socket.destroy(); await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); diff --git a/server/src/__tests__/helpers/maintenance/server-harness.ts b/server/src/__tests__/helpers/maintenance/server-harness.ts index a7197cd50a3..606c449dd7b 100644 --- a/server/src/__tests__/helpers/maintenance/server-harness.ts +++ b/server/src/__tests__/helpers/maintenance/server-harness.ts @@ -159,7 +159,17 @@ export async function startMaintenanceServer(databaseUrl: string, options: { art snapshot: () => request("snapshot"), drain: () => request("drain"), resume: () => request("resume"), realtime: () => request("realtime"), - advance: (ms: number) => request("advance", { ms }), freeze: () => request("freeze"), + async advance(ms: number) { + const deadline = performance.now() + 30_000; + while (proxy.starting > 0) { + if (proxy.errors.length) throw new AggregateError(proxy.errors, "PostgreSQL observer failed"); + if (performance.now() >= deadline) throw new Error("Maintenance database startup did not finish before clock advance"); + await delay(5); + } + if (proxy.errors.length) throw new AggregateError(proxy.errors, "PostgreSQL observer failed"); + return request("advance", { ms }); + }, + freeze: () => request("freeze"), get output() { return output; }, async settled(quietMs = 500, deferredStartupHolds = 0) { let previous = -1, quietSince = Date.now(); diff --git a/server/src/__tests__/maintenance-postgres-proxy.test.ts b/server/src/__tests__/maintenance-postgres-proxy.test.ts index caa3fdf0dd9..c09e99b0a5c 100644 --- a/server/src/__tests__/maintenance-postgres-proxy.test.ts +++ b/server/src/__tests__/maintenance-postgres-proxy.test.ts @@ -1,4 +1,6 @@ import { createDb } from "@paperclipai/db"; +import { createConnection } from "node:net"; +import { once } from "node:events"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; @@ -16,7 +18,18 @@ 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, + expect(observer.pending).toBe(1); + expect(observer.ready).toBe(false); + for (const byte of startup) observer.frontend(Buffer.from([byte])); + expect(observer.pending).toBe(1); + const initialReady = frame("Z", "I"); + observer.backend(initialReady.subarray(0, 3)); + expect(observer.pending).toBe(1); + expect(observer.ready).toBe(false); + observer.backend(initialReady.subarray(3)); + expect(observer.pending).toBe(0); + expect(observer.ready).toBe(true); + const bytes = Buffer.concat([ frame("Q", "select 1\0"), frame("P", "statement\0select $1::text\0"), frame("B", "portal\0statement\0parameter-value-sentinel"), @@ -78,6 +91,25 @@ describeDatabase("maintenance observer against PostgreSQL", () => { await new Promise(resolve => setTimeout(resolve, 100)); expect(proxy.queries.length).toBe(before); expect(proxy.transport).toEqual(transport); + expect(proxy.starting).toBe(0); + const address = new URL(proxy.url); + for (const cancel of [false, true]) { + const socket = createConnection({ host: address.hostname, port: Number(address.port) }); + try { + await once(socket, "connect"); + await expect.poll(() => proxy.starting).toBe(1); + const closed = once(socket, "close"); + if (cancel) { + const request = Buffer.alloc(16); + request.writeInt32BE(16); request.writeInt32BE(80877102, 4); + socket.write(request); + } else socket.end(); + await closed; + await expect.poll(() => proxy.starting).toBe(0); + expect(proxy.pending).toBe(0); + expect(proxy.errors).toEqual([]); + } finally { socket.destroy(); } + } } finally { await db.$client.end(); await proxy.stop(); } }); }); diff --git a/server/src/__tests__/maintenance-server-routes.test.ts b/server/src/__tests__/maintenance-server-routes.test.ts index 14c0324b7fb..8ac64ce28ea 100644 --- a/server/src/__tests__/maintenance-server-routes.test.ts +++ b/server/src/__tests__/maintenance-server-routes.test.ts @@ -1,5 +1,7 @@ import { createCipheriv, createHash, diffieHellman, generateKeyPairSync, hkdfSync, randomBytes, randomUUID, sign, verify } from "node:crypto"; import { createServer } from "node:http"; +import { Socket } from "node:net"; +import { once } from "node:events"; import { agents, agentApiKeys, agentSessionGoalActions, agentTaskSessions, authUsers, companies, companyMemberships, connectionGrants, createDb, heartbeatRuns, issues, issueThreadInteractions, toolApplications, toolConnections } from "@paperclipai/db"; import { access, mkdir, readdir, utimes, writeFile } from "node:fs/promises"; import path from "node:path"; @@ -294,6 +296,51 @@ describeDatabase("assembled maintenance server production wiring", () => { } }, 150_000); + it("does not advance the child clock over an actual driver socket whose startup write is held", async () => { + await clearDatabase(); + const server = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true }); + const client = createDb(server.proxy.url, { maxConnections: 1, idleTimeoutSeconds: 0, connectTimeoutSeconds: 30 }); + const address = new URL(server.proxy.url); + const socket = new Socket(); + let release = () => {}; + let query: Promise | undefined; + let advancing: ReturnType | undefined; + try { + await settleClock(server); await server.freeze(); + socket.connect({ host: address.hostname, port: Number(address.port) }); + await once(socket, "connect"); + const write = socket.write; + let entered!: () => void; + const startupHeld = new Promise(resolve => { entered = resolve; }); + socket.write = ((...args: Parameters) => { + socket.write = write; + expect(Buffer.isBuffer(args[0]) && args[0].readInt32BE(4)).toBe(196608); + let released = false; + release = () => { if (!released) { released = true; Reflect.apply(write, socket, args); } }; + entered(); + return true; + }) as typeof socket.write; + Object.assign(client.$client.options, { socket: () => socket }); + query = client.$client`select 1 as startup_control`.then(rows => rows); + await startupHeld; + const before = await server.snapshot(); + advancing = server.advance(30_000); + const during = await server.snapshot(); + expect(during.now).toBe(before.now); + release(); + expect(await query).toMatchObject([{ startup_control: 1 }]); + expect((await advancing).now).toBe(before.now + 30_000); + await server.settled(); + await writeFile(path.join(server.directory, "startup-clock-measurement.json"), JSON.stringify({ phase: "held-real-driver-startup-clock-barrier", + before, during, after: await server.snapshot(), boundary: "actual postgres.js StartupMessage write held before wire flush; child clock IPC stays behind startup, independent of the unexplained invalid-frame failure", + }, null, 2)); + } finally { + release(); + await Promise.allSettled([query, advancing]); + await client.$client.end(); socket.destroy(); 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 });