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 <noreply@paperclip.ing>

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<string, string>(), portals = new Map<string, string>();
   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<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()));

server/src/__tests__/helpers/maintenance/server-harness.ts

@@ -159,7 +159,17 @@ export async function startMaintenanceServer(databaseUrl: string, options: { art
       snapshot: () => request<Snapshot>("snapshot"),
       drain: () => request<Snapshot>("drain"), resume: () => request<Snapshot>("resume"),
       realtime: () => request<Snapshot>("realtime"),
-      advance: (ms: number) => request<Snapshot>("advance", { ms }), freeze: () => request<Snapshot>("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<Snapshot>("advance", { ms });
+      },
+      freeze: () => request<Snapshot>("freeze"),
       get output() { return output; },
       async settled(quietMs = 500, deferredStartupHolds = 0) {
         let previous = -1, quietSince = Date.now();

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(); }
   });
 });

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<unknown> | undefined;
+    let advancing: ReturnType<typeof server.advance> | 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<void>(resolve => { entered = resolve; });
+      socket.write = ((...args: Parameters<typeof socket.write>) => {
+        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 });