From 5e0e8162ab5a5de6da52ac2b41453d8d34b65154 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 02:48:15 -0500 Subject: [PATCH] Keep virtual maintenance time behind database startup Count accepted database sockets as pending through initial ReadyForQuery and bound virtual clock advancement on startup readiness. Qualify the held actual driver write, disconnect and cancellation boundaries. Retain fail-closed header diagnostics; the earlier malformed-frame cause remains unknown. Co-Authored-By: Paperclip --- .../helpers/maintenance/postgres-proxy.ts | 8 ++-- .../helpers/maintenance/server-harness.ts | 12 ++++- .../maintenance-postgres-proxy.test.ts | 34 +++++++++++++- .../maintenance-server-routes.test.ts | 47 +++++++++++++++++++ 4 files changed, 96 insertions(+), 5 deletions(-) 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 });