fix(relay): a lost database reply ends its connection within statement timeout + 10 s (#26829)

* fix(relay): a lost database reply ends its connection within statement timeout + 2 s

The request pool sets query_timeout to the statement timeout plus 2 s. A
read timeout poisons its transaction: later statements fail at once,
COMMIT never runs (even when the operation caught the error), no ROLLBACK
queues behind the dead statement, and the client is released with the
error so pg-pool destroys it. Autocommit statements destroy their client
too. The server bounds an orphaned idle backend with idle_session_timeout
30 s. A read timeout is answered 503 and never retried, and a renewal
batch that loses its reply fails whole instead of fanning out.

* fix(relay): read timeout margin 10 s, configurable, so routine 5-7 s database stalls never trip it

A whole-VM stall delays replies the server already timed. With a 2 s
margin the read timeout fired inside routine stalls, destroying healthy
clients and failing requests that succeed today. The margin is now 10 s
(ORCA_RELAY_POSTGRES_READ_TIMEOUT_MARGIN_MS), and a real-server test holds
a reply 6.5 s and requires it delivered on the same connection. The fix
level rises to 2 here, with the lost-reply fix.

* docs(relay): name the startup retry that can see a read timeout, and why it is safe

Boot open retries only an unreachable database. The director's startup
cell reconcile does retry a lost reply, which is safe: it re-runs upserts
and limit-change deletes that converge on the same rows.

* fix(relay): answer 503, not 500, when Postgres ends the session mid-request

Staging B1: pg_terminate_backend during /v1/assign returned 500 29 times.
57P01/57P02 and 08000/08003 join the transient SQLSTATEs; pg's
'Connection terminated' and ECONNRESET count only when the database layer
threw them. transaction() still never retries them.

* fix(relay): answer 503 when the pool acquire fails on a dead socket

Staging B1 re-test: a Cloud SQL restart failed PostgresDatabase.transaction's
acquire with 'write EPIPE', answered 500. An acquire failure with EPIPE,
ECONNRESET, ECONNREFUSED, ETIMEDOUT or EHOSTUNREACH is now a pool connect
failure (no statement ran), and acquire failures count as database-layer
errors. EPIPE mid-statement is treated like ECONNRESET.

cloud/apps/relay/src/assignment-store.ts

@@ -73,6 +73,7 @@ import {
 import type { RelayCellConfig } from './config.js'
 import type { CellLockHoldSite } from './cell-inventory-hold-samples.js'
 import { isPostgresPoolConnectFailure } from './postgres-pool-pressure.js'
+import { isPostgresReadTimeout } from './relay-database-rejection-fence.js'
 import type {
   RelayDatabase,
   RelayLockOptions,
@@ -4096,9 +4097,11 @@ export class RelayAssignmentStore {
           message: String((error as { message?: unknown }).message)
         })
       )
-      // No connection at all: per-host statements would only queue up behind the
-      // same saturated pool, multiplying the load that starved this flush.
-      if (isPostgresPoolConnectFailure(error)) return outcomes.fill('database_error')
+      // No connection at all, or a lost reply: per-host statements would only queue up
+      // behind the same saturated pool or stalled server, multiplying the load.
+      if (isPostgresPoolConnectFailure(error) || isPostgresReadTimeout(error)) {
+        return outcomes.fill('database_error')
+      }
       await Promise.all(
         ordered.map(async (row) => {
           try {

cloud/apps/relay/src/cell-admission-startup.ts

@@ -28,6 +28,8 @@ export async function reconcileCellAdmissionAtStartup(
   // Admission is operator/director state. A new worker must not enable itself
   // before its distinct candidate has passed production preflight.
   if (config.role === 'cell') return
+  // Retries every transient failure, a lost reply included: the reconcile is upserts and
+  // limit-change deletes that converge on the same rows if a lost COMMIT had landed.
   await retryTransientDatabaseStartup(
     async () => await assignments.reconcileCellsAtStartup(config.cells),
     STARTUP_RECONCILE_RETRY,

cloud/apps/relay/src/database-postgres-timeout.test.ts

@@ -46,6 +46,7 @@ vi.mock('pg', () => ({
 import {
   openRelayDatabase,
   POSTGRES_SCHEMA_MIGRATIONS,
+  relayPostgresReadTimeoutMarginMs,
   relayPostgresStatementTimeoutMs
 } from './database.js'
 import { applyPostgresSchema } from './postgres-schema-startup.js'
@@ -89,10 +90,15 @@ describe('PostgreSQL relay deadlines', () => {
         application_name: 'orca-relay/director/director',
         connectionTimeoutMillis: 2_000,
         statement_timeout: 5_000,
+        // A lost reply ends at the statement deadline plus 10 s, on this pool only.
+        query_timeout: 15_000,
         lock_timeout: 1_000,
-        idle_in_transaction_session_timeout: 5_000
+        idle_in_transaction_session_timeout: 5_000,
+        options: '-c idle_session_timeout=30000'
       })
     ])
+    expect(fakes.configs[0]).not.toHaveProperty('query_timeout')
+    expect(fakes.configs[0]).not.toHaveProperty('options')
     await database.close()
   })
 
@@ -160,7 +166,7 @@ describe('PostgreSQL relay deadlines', () => {
 
     expect(fakes.configs).toEqual([
       expect.objectContaining({ statement_timeout: 0 }),
-      expect.objectContaining({ statement_timeout: 2_500 })
+      expect.objectContaining({ statement_timeout: 2_500, query_timeout: 12_500 })
     ])
     await database.close()
   })
@@ -174,6 +180,18 @@ describe('PostgreSQL relay deadlines', () => {
     }
   )
 
+  it('takes the read timeout margin from the environment, defaulting to 10 s', () => {
+    expect(relayPostgresReadTimeoutMarginMs({})).toBe(10_000)
+    expect(
+      relayPostgresReadTimeoutMarginMs({ ORCA_RELAY_POSTGRES_READ_TIMEOUT_MARGIN_MS: '4000' })
+    ).toBe(4_000)
+    for (const value of ['0', '-1', '2.5', 'soon']) {
+      expect(() =>
+        relayPostgresReadTimeoutMarginMs({ ORCA_RELAY_POSTGRES_READ_TIMEOUT_MARGIN_MS: value })
+      ).toThrow('invalid_read_timeout_margin')
+    }
+  })
+
   it.each([undefined, ''])('defaults to 5s when the environment says %s', (value) => {
     expect(
       relayPostgresStatementTimeoutMs(

cloud/apps/relay/src/database-transient-error.test.ts

@@ -1,23 +1,56 @@
 import { describe, expect, it, vi } from 'vitest'
-import { isRelayDatabaseTransientError } from './database.js'
+import { isRelayDatabaseTransientError, PostgresDatabase } from './database.js'
 import { PostgresPoolPressure } from './postgres-pool-pressure.js'
 
 // The only way to mark an error as an acquire failure is to fail a real
 // acquire, so the gated cases go through the pressure wrapper the pool uses.
-async function failedAcquire(message: string): Promise<unknown> {
+async function failedAcquire(failure: string | Error): Promise<unknown> {
   const pool = {
     totalCount: 0,
     idleCount: 0,
     waitingCount: 0,
     connect: vi.fn(async () => {
-      throw new Error(message)
+      throw typeof failure === 'string' ? new Error(failure) : failure
     })
   }
   return await new PostgresPoolPressure(pool as never).connect().catch((error: unknown) => error)
 }
 
+async function failedQuery(failure: Error): Promise<unknown> {
+  const client = {
+    query: vi.fn(async () => {
+      throw failure
+    }),
+    release: vi.fn(),
+    on: vi.fn(),
+    removeListener: vi.fn()
+  }
+  const database = new PostgresDatabase({
+    totalCount: 1,
+    idleCount: 0,
+    waitingCount: 0,
+    connect: async () => client
+  } as never)
+  vi.spyOn(console, 'warn').mockImplementation(() => {})
+  return await database.query('SELECT 1').catch((error: unknown) => error)
+}
+
 describe('relay database transient errors', () => {
-  it.each(['40P01', '40001', '55P03', '57014', '53300', '57P03', '08001', '08006', '25P03'])(
+  it.each([
+    '40P01',
+    '40001',
+    '55P03',
+    '57014',
+    '53300',
+    '57P01',
+    '57P02',
+    '57P03',
+    '08000',
+    '08001',
+    '08003',
+    '08006',
+    '25P03'
+  ])(
     'classifies PostgreSQL code %s as retryable overload',
     (code) => {
       expect(isRelayDatabaseTransientError({ code })).toBe(true)
@@ -40,15 +73,50 @@ describe('relay database transient errors', () => {
     ).toBe(true)
   })
 
-  it('classifies an early-ended socket only when it ended during the acquire', async () => {
+  it('classifies an early-ended socket during the acquire', async () => {
     expect(
       isRelayDatabaseTransientError(await failedAcquire('Connection terminated unexpectedly'))
     ).toBe(true)
-    // The same message mid-statement leaves the commit outcome unknown, so it
-    // must stay a hard failure rather than invite a retry.
-    expect(
-      isRelayDatabaseTransientError(new Error('Connection terminated unexpectedly'))
-    ).toBe(false)
+  })
+
+  // Staging B1 re-test: a Cloud SQL restart failed the transaction's acquire with `write EPIPE`.
+  it.each(['EPIPE', 'ECONNRESET', 'ECONNREFUSED', 'ETIMEDOUT', 'EHOSTUNREACH'])(
+    'classifies an acquire that failed with %s, but not the bare errno',
+    async (code) => {
+      const failure = (): Error => Object.assign(new Error(`connect ${code}`), { code })
+      expect(isRelayDatabaseTransientError(await failedAcquire(failure()))).toBe(true)
+      expect(isRelayDatabaseTransientError(failure())).toBe(false)
+    }
+  )
+
+  it('classifies an EPIPE from the transaction acquire', async () => {
+    const database = new PostgresDatabase({
+      totalCount: 0,
+      idleCount: 0,
+      waitingCount: 0,
+      connect: async () => {
+        throw Object.assign(new Error('write EPIPE'), { code: 'EPIPE' })
+      }
+    } as never)
+    const error = await database.transaction(async () => 'unreached').catch((e: unknown) => e)
+    expect(error).toMatchObject({ code: 'EPIPE' })
+    expect(isRelayDatabaseTransientError(error)).toBe(true)
+  })
+
+  // Staging B1: pg_terminate_backend mid-request answered /v1/assign with 500. A dropped
+  // connection is 503 + Retry-After, as 08006 always was; transaction() still never retries it,
+  // since the commit outcome is unknown.
+  it('classifies a mid-statement drop only from the database layer', async () => {
+    const failures = [
+      (): Error => new Error('Connection terminated unexpectedly'),
+      (): Error => Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }),
+      (): Error => Object.assign(new Error('write EPIPE'), { code: 'EPIPE' })
+    ]
+    for (const failure of failures) {
+      expect(isRelayDatabaseTransientError(await failedQuery(failure()))).toBe(true)
+      // The same errno or message from any other socket is not a database answer.
+      expect(isRelayDatabaseTransientError(failure())).toBe(false)
+    }
   })
 
   it.each([null, undefined, 'a thrown string'])(

cloud/apps/relay/src/database.ts

@@ -14,7 +14,11 @@ import {
 import { applyPostgresSchema } from './postgres-schema-startup.js'
 import { POSTGRES_STATEMENT_STATS_MIGRATION } from './postgres-statement-stats.js'
 import { reportPostgresQueryFailure } from './postgres-query-failure.js'
-import { markRelayDatabaseError } from './relay-database-rejection-fence.js'
+import {
+  isPostgresReadTimeout,
+  isRelayDatabaseLayerError,
+  markRelayDatabaseError
+} from './relay-database-rejection-fence.js'
 import {
   CellInventoryHoldSamples,
   emptyCellInventoryHoldCounts,
@@ -890,15 +894,33 @@ class PostgresTransaction implements RelayDatabase {
   private lockUnavailable = 0
   private lockTimeouts = 0
   private state: 'open' | 'committed' | 'rolled-back' = 'open'
+  // A read timeout leaves the statement running with its reply lost: pg keeps it as the
+  // client's active query, so anything sent next queues behind it, and if it then lands, a
+  // later COMMIT would commit a statement its caller saw fail. The connection is finished.
+  private lostReply: Error | undefined
 
   constructor(protected readonly client: pg.PoolClient) {}
 
+  get poisonedBy(): Error | undefined {
+    return this.lostReply
+  }
+
+  private failIfPoisoned(): void {
+    if (this.lostReply) throw this.lostReply
+  }
+
+  private noteFailure(error: unknown): void {
+    markRelayDatabaseError(error)
+    if (isPostgresReadTimeout(error)) this.lostReply ??= error
+  }
+
   get open(): boolean {
     return this.state === 'open'
   }
 
   async commitWithFinal(sql: string, params: unknown[] = []): Promise<boolean> {
     this.assertNotCommitted()
+    this.failIfPoisoned()
     // Zero rows divides by zero, so the message stops before COMMIT exactly when the write missed.
     const message =
       `WITH final_write AS (${inlinePostgresParameters(sql, params, this.client)}) ` +
@@ -909,13 +931,13 @@ class PostgresTransaction implements RelayDatabase {
       // Simple query stops at the first error, so any server error means COMMIT never ran:
       // retryable codes take the caller's normal rollback-and-retry path. A lost connection
       // leaves the outcome unknown, and nothing retries it.
-      markRelayDatabaseError(error)
+      this.noteFailure(error)
       if (String((error as { code?: unknown }).code) !== '22012') {
         rememberPostgresTransactionPhase(error, sql)
         throw error
       }
       await this.client.query('ROLLBACK').catch((rollbackError: unknown) => {
-        markRelayDatabaseError(rollbackError)
+        this.noteFailure(rollbackError)
         throw rollbackError
       })
       this.state = 'rolled-back'
@@ -953,11 +975,12 @@ class PostgresTransaction implements RelayDatabase {
 
   async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
     this.assertNotCommitted()
+    this.failIfPoisoned()
     try {
       const result = await this.client.query(postgresSql(sql), params)
       return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
     } catch (error) {
-      markRelayDatabaseError(error)
+      this.noteFailure(error)
       rememberPostgresTransactionPhase(error, sql)
       throw error
     }
@@ -1021,6 +1044,11 @@ const POSTGRES_CONNECTION_TIMEOUT_MS = 2_000
 // leaves room for the connect timeout above.
 export const POSTGRES_STATEMENT_TIMEOUT_MS = 5_000
 const POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS = 5_000
+// Past statement_timeout because a whole-VM database stall delays replies the server has
+// already timed: 5-7 s stalls are routine (connect-burst RCA), and a read timeout inside one
+// would destroy healthy clients and fail requests that succeed today.
+export const POSTGRES_READ_TIMEOUT_MARGIN_MS = 10_000
+const POSTGRES_IDLE_SESSION_TIMEOUT_MS = 30_000
 
 export function relayPostgresStatementTimeoutMs(
   env: NodeJS.ProcessEnv = process.env
@@ -1036,6 +1064,18 @@ export function relayPostgresStatementTimeoutMs(
   return milliseconds
 }
 
+export function relayPostgresReadTimeoutMarginMs(
+  env: NodeJS.ProcessEnv = process.env
+): number {
+  const configured = env.ORCA_RELAY_POSTGRES_READ_TIMEOUT_MARGIN_MS
+  if (configured === undefined || configured === '') return POSTGRES_READ_TIMEOUT_MARGIN_MS
+  const milliseconds = Number(configured)
+  if (!Number.isInteger(milliseconds) || milliseconds < 1) {
+    throw new Error('invalid_read_timeout_margin')
+  }
+  return milliseconds
+}
+
 function retryablePostgresTransactionError(error: unknown): boolean {
   const code = String((error as { code?: unknown }).code)
   // 57014 is the pool statement_timeout firing. It aborts the transaction the
@@ -1056,14 +1096,33 @@ export function isRelayDatabaseTransientError(error: unknown): boolean {
       '57014',
       '53300',
       '57P03',
+      '08000',
       '08001',
+      '08003',
       '08006',
+      // The server ended the session: pg_terminate_backend or a fast/immediate shutdown.
+      '57P01',
+      '57P02',
       // The server ended a session idle in a transaction past its limit (a database stall).
       '25P03'
     ].includes(code)
   ) {
     return true
   }
+  // The connection dropped mid-statement: pg's own message, or the socket's reset. Only from
+  // the database layer, since any other socket can raise the same errno.
+  if (
+    isRelayDatabaseLayerError(error) &&
+    (code === 'ECONNRESET' ||
+      code === 'EPIPE' ||
+      String((error as { message?: unknown }).message).startsWith('Connection terminated'))
+  ) {
+    return true
+  }
+  // A lost reply, answered 503 like 08006. Its write may have landed, so transaction() never
+  // retries it. The one startup retry that can see it, the director's cell reconcile
+  // (cell-admission-startup.ts), re-runs only upserts that converge on the same rows.
+  if (isPostgresReadTimeout(error)) return true
   // A pool that cannot hand out a client reports no SQLSTATE at all, so the
   // acquire boundary owns that vocabulary.
   return isPostgresPoolConnectFailure(error)
@@ -1103,6 +1162,7 @@ export class PostgresDatabase implements RelayDatabase {
     const startedAt = performance.now()
     let phase: 'acquire' | 'execute' = 'acquire'
     let client: pg.PoolClient | undefined
+    let lostReply: Error | undefined
     try {
       client = await this.pressure.connect(lane)
       phase = 'execute'
@@ -1110,6 +1170,7 @@ export class PostgresDatabase implements RelayDatabase {
       return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
     } catch (error) {
       markRelayDatabaseError(error)
+      if (isPostgresReadTimeout(error)) lostReply = error
       reportPostgresQueryFailure({
         error,
         phase,
@@ -1122,7 +1183,8 @@ export class PostgresDatabase implements RelayDatabase {
       })
       throw error
     } finally {
-      client?.release()
+      // An error destroys the client rather than handing the next caller a dead connection.
+      client?.release(lostReply)
     }
   }
 
@@ -1154,9 +1216,12 @@ export class PostgresDatabase implements RelayDatabase {
     for (let attempt = 1; attempt <= POSTGRES_TRANSACTION_ATTEMPTS; attempt++) {
       const client = await this.pressure.connect()
       const transaction = new PostgresTransaction(client)
+      let lostReply: Error | undefined
       try {
         await client.query('BEGIN')
         const result = await operation(transaction)
+        // Even when the operation caught it: COMMIT must not run after a lost reply.
+        if (transaction.poisonedBy) throw transaction.poisonedBy
         if (transaction.open) await client.query('COMMIT')
         recordMeasuredHold(this.holds, transaction)
         this.holds.recordUnavailable(transaction.consumeLockUnavailable())
@@ -1164,13 +1229,17 @@ export class PostgresDatabase implements RelayDatabase {
         return result
       } catch (error) {
         markRelayDatabaseError(error)
+        // No ROLLBACK behind a lost reply: it would queue behind the dead statement. Destroying
+        // the client ends the session, and the server drops its locks.
+        lostReply = transaction.poisonedBy ?? (isPostgresReadTimeout(error) ? error : undefined)
         const open = transaction.open
-        if (open) await client.query('ROLLBACK').catch(() => undefined)
+        if (open && !lostReply) await client.query('ROLLBACK').catch(() => undefined)
         this.holds.recordUnavailable(transaction.consumeLockUnavailable())
         this.holds.recordLockTimeout(transaction.consumeLockTimeouts())
         // Once the fused commit has ended the transaction, a retry would apply the work twice.
         if (
           !open ||
+          lostReply ||
           !retryablePostgresTransactionError(error) ||
           attempt === POSTGRES_TRANSACTION_ATTEMPTS
         ) {
@@ -1197,7 +1266,7 @@ export class PostgresDatabase implements RelayDatabase {
           )
         }
       } finally {
-        client.release()
+        client.release(lostReply)
       }
       // A PostgreSQL transaction is unusable after an abort, so retry all work
       // on a fresh pooled client with a small full-jitter delay.
@@ -1260,6 +1329,27 @@ export function keepPostgresClientErrorsHandled(pool: Pick<pg.Pool, 'on'>): void
   })
 }
 
+// An `options=` in the database URL silently replaces the pool's, so the session reports what
+// it actually got. The URL itself is never read or logged.
+function reportPostgresIdleSessionTimeoutOnce(pool: Pick<pg.Pool, 'on'>): void {
+  let reported = false
+  pool.on('connect', (client) => {
+    if (reported) return
+    reported = true
+    client
+      .query("SELECT current_setting('idle_session_timeout') AS value")
+      .then((result: pg.QueryResult) => {
+        console.log(
+          JSON.stringify({
+            event: 'orca_relay_postgres_session_settings',
+            idleSessionTimeout: String(result.rows[0]?.value)
+          })
+        )
+      })
+      .catch(() => undefined)
+  })
+}
+
 export function absorbPostgresIdleClientErrors(pool: Pick<pg.Pool, 'on'>): void {
   pool.on('error', () => {
     // Why: node-postgres removes failed idle clients itself; leaving `error`
@@ -1336,6 +1426,7 @@ export type RelayDatabaseOpenInput = {
   poolMax?: number
   applicationName?: string
   statementTimeoutMs?: number
+  readTimeoutMarginMs?: number
   // Directors own the PostgreSQL schema. A cell skips it and never touches the database
   // at boot, so it starts listening while the database is down and stays unready until
   // its first successful query.
@@ -1349,17 +1440,26 @@ export async function openRelayDatabase(input: RelayDatabaseOpenInput): Promise<
     if (appliesPostgresSchema) {
       await applySchemaOnUntimedPool(input.databaseUrl, input.applicationName)
     }
+    const statementTimeoutMs = input.statementTimeoutMs ?? relayPostgresStatementTimeoutMs()
     const pool = new pg.Pool({
       connectionString: input.databaseUrl,
       max: input.poolMax ?? 10,
       application_name: input.applicationName,
       connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS,
-      statement_timeout: input.statementTimeoutMs ?? relayPostgresStatementTimeoutMs(),
+      statement_timeout: statementTimeoutMs,
+      // The server answers every statement within statement_timeout, as 57014 at worst, so
+      // this fires only on a lost reply, or a database or event-loop stall past the margin.
+      query_timeout:
+        statementTimeoutMs + (input.readTimeoutMarginMs ?? relayPostgresReadTimeoutMarginMs()),
       lock_timeout: POSTGRES_LOCK_TIMEOUT_MS,
-      idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS
+      idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS,
+      // Bounds a backend whose client was destroyed outside a transaction (after a lost
+      // reply on an autocommit write or COMMIT). pg-pool closes healthy idle clients at 10 s.
+      options: `-c idle_session_timeout=${POSTGRES_IDLE_SESSION_TIMEOUT_MS}`
     })
     absorbPostgresIdleClientErrors(pool)
     keepPostgresClientErrorsHandled(pool)
+    reportPostgresIdleSessionTimeoutOnce(pool)
     database = new PostgresDatabase(pool)
   } else {
     mkdirSync(input.dataDir, { recursive: true })

cloud/apps/relay/src/postgres-lost-reply-postgres.test.ts

@@ -0,0 +1,156 @@
+import { createServer, connect, type Server, type Socket } from 'node:net'
+import { afterEach, describe, expect, it } from 'vitest'
+import {
+  isRelayDatabaseTransientError,
+  openRelayDatabase,
+  POSTGRES_READ_TIMEOUT_MARGIN_MS,
+  type RelayDatabase
+} from './database.js'
+
+const databaseUrl = process.env.ORCA_RELAY_TEST_POSTGRES_URL
+const describePostgres = databaseUrl ? describe : describe.skip
+
+const STATEMENT_TIMEOUT_MS = 500
+// A short margin keeps the blackhole case fast; production's is 10 s.
+const READ_TIMEOUT_MARGIN_MS = 2_000
+const READ_TIMEOUT_MS = STATEMENT_TIMEOUT_MS + READ_TIMEOUT_MARGIN_MS
+const PROBE_TABLE = 'relay_lost_reply_probe'
+
+// Forwards each connection to PostgreSQL. Once a connection sends a statement naming the
+// trigger, the server's replies on that connection are dropped (a reply lost in the network),
+// or, with `delayMs`, held that long and then delivered (a whole-VM database stall).
+async function blackholeProxy(target: URL, trigger: string, delayMs?: number) {
+  const sockets = new Set<Socket>()
+  const afterTrigger: Buffer[] = []
+  const server: Server = createServer((client) => {
+    const upstream = connect(Number(target.port || 5432), target.hostname)
+    sockets.add(client)
+    sockets.add(upstream)
+    let swallowing = false
+    client.on('data', (chunk: Buffer) => {
+      if (swallowing) afterTrigger.push(chunk)
+      else if (chunk.includes(trigger)) swallowing = true
+      upstream.write(chunk)
+    })
+    upstream.on('data', (chunk: Buffer) => {
+      if (!swallowing) client.write(chunk)
+      else if (delayMs !== undefined) setTimeout(() => client.write(chunk), delayMs)
+    })
+    for (const socket of [client, upstream]) {
+      socket.on('error', () => undefined)
+      socket.on('close', () => {
+        client.destroy()
+        upstream.destroy()
+      })
+    }
+  })
+  await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve))
+  const address = server.address()
+  if (typeof address !== 'object' || address === null) throw new Error('proxy not listening')
+  const url = new URL(target)
+  url.hostname = '127.0.0.1'
+  url.port = String(address.port)
+  return {
+    url: url.toString(),
+    sentAfterTrigger: () => Buffer.concat(afterTrigger).toString('latin1'),
+    close: async () => {
+      for (const socket of sockets) socket.destroy()
+      await new Promise<void>((resolve) => server.close(() => resolve()))
+    }
+  }
+}
+
+describePostgres('a lost PostgreSQL reply against a real server', () => {
+  const cleanups: Array<() => Promise<void>> = []
+  afterEach(async () => {
+    for (const cleanup of cleanups.splice(0).reverse()) await cleanup()
+  })
+
+  async function open(
+    url: string,
+    options: { readTimeoutMarginMs?: number; poolMax?: number } = {}
+  ): Promise<RelayDatabase> {
+    const database = await openRelayDatabase({
+      databaseUrl: url,
+      dataDir: '',
+      appliesPostgresSchema: false,
+      statementTimeoutMs: STATEMENT_TIMEOUT_MS,
+      readTimeoutMarginMs: options.readTimeoutMarginMs ?? READ_TIMEOUT_MARGIN_MS,
+      poolMax: options.poolMax
+    })
+    cleanups.push(async () => await database.close())
+    return database
+  }
+
+  it('fails within the read timeout, sends nothing after the lost statement, and recovers', async () => {
+    const direct = await open(databaseUrl!)
+    await direct.query(`CREATE TABLE IF NOT EXISTS ${PROBE_TABLE} (id TEXT PRIMARY KEY)`)
+    await direct.query(`INSERT INTO ${PROBE_TABLE} (id) VALUES ('c1') ON CONFLICT DO NOTHING`)
+    cleanups.push(async () => {
+      await direct.query(`DROP TABLE IF EXISTS ${PROBE_TABLE}`)
+    })
+    const proxy = await blackholeProxy(new URL(databaseUrl!), `FROM ${PROBE_TABLE}`)
+    cleanups.push(proxy.close)
+    const database = await open(proxy.url)
+
+    const startedAt = performance.now()
+    const failure = await database
+      .transaction(
+        async (transaction) =>
+          await transaction.queryLocked(`SELECT id FROM ${PROBE_TABLE} WHERE id = ?`, ['c1'], {
+            lockTimeoutMs: 200
+          })
+      )
+      .catch((error: unknown) => error)
+    const elapsedMs = performance.now() - startedAt
+
+    expect(failure).toBeInstanceOf(Error)
+    expect(isRelayDatabaseTransientError(failure)).toBe(true)
+    expect(elapsedMs).toBeLessThan(READ_TIMEOUT_MS + 1_000)
+    // pg's Terminate ('X') may follow as the client is destroyed; no statement may.
+    const after = proxy.sentAfterTrigger()
+    for (const statement of ['lock_timeout', 'ROLLBACK', 'COMMIT', 'SELECT']) {
+      expect(after).not.toContain(statement)
+    }
+    // The next checkout is a fresh connection, so it is not stuck behind the lost reply.
+    await expect(database.query('SELECT 1 AS one')).resolves.toEqual([{ one: 1 }])
+    // The row lock went with the destroyed session.
+    await expect(
+      direct.transaction(
+        async (transaction) =>
+          await transaction.queryLocked(`SELECT id FROM ${PROBE_TABLE} WHERE id = ?`, ['c1'], {
+            failIfUnavailable: true
+          })
+      )
+    ).resolves.toEqual([{ id: 'c1' }])
+  }, 20_000)
+
+  // A reply the server sent after statement_timeout, held up by a whole-VM stall, is still a
+  // reply: the default margin must outlast the routine 5-7 s stalls.
+  it('keeps a reply delayed 6.5 s by a stall, on the same connection', async () => {
+    const proxy = await blackholeProxy(new URL(databaseUrl!), 'stalled_reply_probe', 6_500)
+    cleanups.push(proxy.close)
+    const database = await open(proxy.url, {
+      readTimeoutMarginMs: POSTGRES_READ_TIMEOUT_MARGIN_MS,
+      poolMax: 1
+    })
+    const pid = async () => (await database.query('SELECT pg_backend_pid() AS pid'))[0]?.pid
+    const before = await pid()
+    const startedAt = performance.now()
+    // Autocommit: inside a transaction the proxy's hold would trip the server's own 5 s
+    // idle-in-transaction limit, which a real VM stall pauses along with everything else.
+    await expect(database.query("SELECT 'stalled_reply_probe' AS probe")).resolves.toEqual([
+      { probe: 'stalled_reply_probe' }
+    ])
+    expect(performance.now() - startedAt).toBeGreaterThan(6_000)
+    // Not destroyed: the one pooled connection is still the same backend.
+    expect(await pid()).toBe(before)
+  }, 30_000)
+
+  it('runs request sessions with the server bounding idle backends at 30 s', async () => {
+    const database = await open(databaseUrl!)
+    await expect(
+      database.query("SELECT current_setting('idle_session_timeout') AS value")
+    ).resolves.toEqual([{ value: '30s' }])
+  })
+})

cloud/apps/relay/src/postgres-lost-reply.test.ts

@@ -0,0 +1,124 @@
+import { EventEmitter } from 'node:events'
+import { describe, expect, it, vi } from 'vitest'
+import { isRelayDatabaseTransientError, PostgresDatabase } from './database.js'
+import {
+  isFencedRelayDatabaseRejection,
+  POSTGRES_READ_TIMEOUT_MESSAGE
+} from './relay-database-rejection-fence.js'
+
+// pg's query_timeout firing: the statement's reply never arrived.
+const readTimeout = () => new Error(POSTGRES_READ_TIMEOUT_MESSAGE)
+
+class FakePoolClient extends EventEmitter {
+  readonly statements: string[] = []
+  readonly released: Array<Error | boolean | undefined> = []
+
+  constructor(private readonly timesOut: (sql: string) => boolean) {
+    super()
+  }
+
+  query = vi.fn(async (sql: string) => {
+    this.statements.push(sql)
+    if (this.timesOut(sql)) throw readTimeout()
+    return { rows: [{ one: 1 }], rowCount: 1 }
+  })
+
+  release = (error?: Error | boolean): void => {
+    this.released.push(error)
+  }
+}
+
+function databaseOver(timesOut: (sql: string) => boolean) {
+  const clients: FakePoolClient[] = []
+  const database = new PostgresDatabase({
+    totalCount: 1,
+    idleCount: 0,
+    waitingCount: 0,
+    connect: async () => {
+      const client = new FakePoolClient(timesOut)
+      clients.push(client)
+      return client
+    }
+  } as never)
+  return { database, clients }
+}
+
+describe('a lost PostgreSQL reply', () => {
+  it('ends a bounded locked statement with nothing else sent on its connection', async () => {
+    const { database, clients } = databaseOver((sql) => sql.includes('FOR UPDATE'))
+    const failure = await database
+      .transaction(
+        async (transaction) =>
+          await transaction.queryLocked('SELECT id FROM cells WHERE id = ?', ['c1'], {
+            lockTimeoutMs: 200
+          })
+      )
+      .catch((error: unknown) => error)
+    expect(failure).toEqual(readTimeout())
+    // Not retried, no lock_timeout restore, no ROLLBACK, no COMMIT behind the dead statement.
+    expect(clients).toHaveLength(1)
+    expect(clients[0]?.statements).toEqual([
+      'BEGIN',
+      "SET LOCAL lock_timeout = '200ms'",
+      'SELECT id FROM cells WHERE id = $1 FOR UPDATE'
+    ])
+    // Released with the error, so pg-pool destroys it instead of reusing it.
+    expect(clients[0]?.released).toEqual([failure])
+    // Answered 503 like a dropped connection, and survivable if nothing awaited it.
+    expect(isRelayDatabaseTransientError(failure)).toBe(true)
+    expect(isFencedRelayDatabaseRejection(failure)).toBe(true)
+  })
+
+  it('never commits after an operation that caught the timeout and carried on', async () => {
+    const { database, clients } = databaseOver((sql) => sql === 'UPDATE cells SET n = n + 1')
+    const failure = await database
+      .transaction(async (transaction) => {
+        await transaction.query('UPDATE cells SET n = n + 1').catch(() => undefined)
+        // Every later statement on the poisoned transaction fails at once, unsent.
+        await expect(transaction.query('SELECT 1')).rejects.toEqual(readTimeout())
+        return 'carried on'
+      })
+      .catch((error: unknown) => error)
+    expect(failure).toEqual(readTimeout())
+    expect(clients[0]?.statements).toEqual(['BEGIN', 'UPDATE cells SET n = n + 1'])
+    expect(clients[0]?.released).toEqual([failure])
+  })
+
+  it('treats a lost COMMIT reply as an unknown outcome: no ROLLBACK, no retry', async () => {
+    const { database, clients } = databaseOver((sql) => sql === 'COMMIT')
+    await expect(
+      database.transaction(async (transaction) => await transaction.query('SELECT 1'))
+    ).rejects.toEqual(readTimeout())
+    expect(clients).toHaveLength(1)
+    expect(clients[0]?.statements).toEqual(['BEGIN', 'SELECT 1', 'COMMIT'])
+    expect(clients[0]?.released[0]).toEqual(readTimeout())
+  })
+
+  it('destroys the client of a timed-out autocommit statement and serves the next one', async () => {
+    let first = true
+    const { database, clients } = databaseOver(() => {
+      const timesOut = first
+      first = false
+      return timesOut
+    })
+    vi.spyOn(console, 'warn').mockImplementation(() => {})
+    await expect(database.query('SELECT 1')).rejects.toEqual(readTimeout())
+    await expect(database.query('SELECT 1')).resolves.toEqual([{ one: 1 }])
+    expect(clients.map((client) => client.released)).toEqual([[readTimeout()], [undefined]])
+  })
+
+  it('still rolls back and retries an ordinary server error', async () => {
+    let failures = 1
+    const { database, clients } = databaseOver(() => false)
+    const result = await database.transaction(async (transaction) => {
+      if (failures-- > 0) throw Object.assign(new Error('deadlock detected'), { code: '40P01' })
+      return await transaction.query('SELECT 1')
+    })
+    expect(result).toEqual([{ one: 1 }])
+    expect(clients.map((client) => client.statements)).toEqual([
+      ['BEGIN', 'ROLLBACK'],
+      ['BEGIN', 'SELECT 1', 'COMMIT']
+    ])
+    expect(clients.map((client) => client.released)).toEqual([[undefined], [undefined]])
+  })
+})

cloud/apps/relay/src/postgres-pool-pressure.ts

@@ -21,6 +21,14 @@ const POOL_CONNECT_TIMEOUT_MESSAGES = [
 // pg raises this whenever a socket ends early, during the handshake and mid
 // statement alike, so only the acquire boundary can tell the two apart.
 const CONNECTION_TERMINATED_MESSAGE = 'Connection terminated unexpectedly'
+// The socket to Postgres (or its proxy) failed before a client existed, e.g. a Cloud SQL restart.
+const ACQUIRE_SOCKET_ERRNOS = new Set([
+  'EPIPE',
+  'ECONNRESET',
+  'ECONNREFUSED',
+  'ETIMEDOUT',
+  'EHOSTUNREACH'
+])
 
 // Membership is tracked beside the error rather than on it: an error object may
 // be frozen, and a mutated one would leak the marker into logs.
@@ -49,9 +57,10 @@ export function isPostgresPoolConnectTimeout(error: unknown): boolean {
 // outcome is unknown is not safe.
 export function isPostgresPoolConnectFailure(error: unknown): boolean {
   if (isPostgresPoolConnectTimeout(error)) return true
+  if (!isPostgresPoolAcquireFailure(error)) return false
   return (
-    errorMessage(error).includes(CONNECTION_TERMINATED_MESSAGE) &&
-    isPostgresPoolAcquireFailure(error)
+    errorMessage(error).includes(CONNECTION_TERMINATED_MESSAGE) ||
+    ACQUIRE_SOCKET_ERRNOS.has(errorCode(error))
   )
 }
 

cloud/apps/relay/src/postgres-query-failure.test.ts

@@ -72,12 +72,12 @@ describe('PostgreSQL query failure diagnostics', () => {
     expect(JSON.stringify(vi.mocked(console.warn).mock.calls)).not.toContain('private')
   })
 
-  // ECONNRESET carries no SQLSTATE the routes retry on, and it arrives after the
-  // statement went out, so it stays a hard failure. The pair pins that boundary.
+  // A reset from the database layer answers 503 like 08006 (staging B1); transaction() still
+  // never retries it, since the statement went out.
   it.each([
     ['57014', true],
     ['55P03', true],
-    ['ECONNRESET', false]
+    ['ECONNRESET', true]
   ] as const)(
     'identifies execute failure %s and releases its client',
     async (code, transient) => {

cloud/apps/relay/src/relay-database-rejection-fence.ts

@@ -25,6 +25,16 @@ const FENCED_SQLSTATE_CLASSES = new Set(['08', '23', '40', '53', '55', '57', '58
 // does to any backend caught idle mid-transaction at its onset.
 const FENCED_SQLSTATES = new Set(['25P03'])
 
+export function isPostgresReadTimeout(error: unknown): error is Error {
+  return error instanceof Error && error.message === POSTGRES_READ_TIMEOUT_MESSAGE
+}
+
+// The database layer threw or rethrew this; an errno or message alone could be any socket's.
+export function isRelayDatabaseLayerError(error: unknown): boolean {
+  if (typeof error !== 'object' || error === null) return false
+  return databaseLayerErrors.has(error) || isPostgresPoolAcquireFailure(error)
+}
+
 export function markRelayDatabaseError(error: unknown): void {
   if (typeof error === 'object' && error !== null) databaseLayerErrors.add(error)
 }

cloud/apps/relay/src/relay-fix-level.ts

@@ -3,4 +3,4 @@
 // the fixed floor `relay_cell_min_fix_level` (production.tfvars) for 6 hours; raise the floor
 // once every serving cell runs the new level. Images from before this constant report nothing,
 // which the alert reads as below.
-export const RELAY_FIX_LEVEL = 1
+export const RELAY_FIX_LEVEL = 2