perf(relay): stop indexing the column every control renewal writes (#21286)

* perf(relay): stop indexing the column every control renewal writes

relay_assignment_activity_expiry indexes expires_at on
relay_assignment_activity_leases, and expires_at is what every control
renewal updates: ~471 calls/s, all of them non-HOT because a changed
indexed column forbids HOT. The index has one reader, the 30s expiry
sweep, which seq-scans the whole 14.8k-row table in under a millisecond.

Drop it, and set fillfactor to 70 so a renewal has room for a second
row version on its own page. Measured on postgres:16-alpine over 14.8k
rows, WAL bytes per renewal and HOT ratio:

  index, fillfactor 100 (today)     0% HOT   371 B
  index, fillfactor 70              0% HOT   246 B
  no index, fillfactor 100        0.5% HOT   298 B
  no index, fillfactor 70         100% HOT    80 B

Both are needed: the index makes HOT illegal, and the default fillfactor
leaves no page space to make it possible.

Neither statement can use the catalog pre-check as it stood. DROP INDEX
IF EXISTS resolves the name before it locks, so once the index is gone
it costs a catalog miss and takes no lock on the table - pinned in the
lock-target census as the one exempt statement. ALTER TABLE SET does
take a lock, so it gets a new 'reloption' target kind that asks
pg_class.reloptions for the name=value pair, keeping the invariant that
no lock-taking statement reaches a warm boot unchecked.

* fix(relay): pre-check the activity-expiry drop and let it defer on a lock timeout

The drop had no catalog pre-check, so it was sent on every boot, and a
55P03 from it was fatal: apply-postgres-schema throws on a lock timeout
with no retry. On the migration boot that combination is a crash loop.
All 28 directors reach the same DROP INDEX at once, it needs ACCESS
EXCLUSIVE on a table written ~475/s with lock_timeout at 1s, and a boot
that fails restarts the instance to re-queue the same DDL behind the
same writers.

Two changes:

- A new 'index-by-name' lock target. A DROP INDEX names no table, so the
  existing index check could not serve it; this one resolves by name
  through the search_path with relkind = 'i', which is how the DROP
  itself resolves, and skips when absent. DROP INDEX now counts as
  lock-taking in the census, so it is covered rather than exempt, and
  IF EXISTS is required the way it is on DROP CONSTRAINT.

- A 'schema-deferrable' marker, read from a statement's leading comment.
  A 55P03 on a marked statement logs
  orca_relay_postgres_schema_object_deferred and leaves the statement
  unapplied instead of failing the boot; the next boot re-sends it. Both
  activity-lease migrations carry it. Everything else keeps the old
  contract and still fails loudly. SchemaApplySummary gains a deferred
  count so a boot that skipped work is distinguishable from one with
  nothing to do.

Verified against a real server: with the index present and the table
held in ACCESS EXCLUSIVE by another session, both statements defer, the
boot completes, nothing is half-applied, and the next boot finishes the
job. A warm boot now sends neither statement at all.

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

@@ -134,7 +134,7 @@ describe('PostgreSQL relay deadlines', () => {
       statements.every(
         (statement) =>
           statement === POSTGRES_STATEMENT_STATS_MIGRATION.trim() ||
-          /^(?:CREATE|ALTER TABLE)\b/i.test(body(statement))
+          /^(?:CREATE|ALTER TABLE|DROP INDEX)\b/i.test(body(statement))
       )
     ).toBe(true)
     // The backfill is DML, so it stays on the deadline-bearing serving pool.

cloud/apps/relay/src/database.ts

@@ -524,8 +524,9 @@ CREATE TABLE IF NOT EXISTS relay_assignment_activity_leases (
   updated_at BIGINT NOT NULL,
   PRIMARY KEY (user_id, relay_host_id, activity_id)
 );
-CREATE INDEX IF NOT EXISTS relay_assignment_activity_expiry
-  ON relay_assignment_activity_leases(expires_at);
+-- expires_at is deliberately unindexed: every control renewal writes it (~471/s), so an index on
+-- it makes each renewal a non-HOT update that rewrites index entries. Its only reader is the 30s
+-- expiry sweep, which seq-scans 14.8k rows / 7MB in a few milliseconds.
 
 CREATE TABLE IF NOT EXISTS relay_control_connection_reservations (
   reservation_id TEXT PRIMARY KEY,
@@ -645,7 +646,24 @@ export const POSTGRES_SCHEMA_MIGRATIONS = [
      ADD COLUMN IF NOT EXISTS host_cooldown_ms BIGINT NOT NULL
      DEFAULT ${REGIONAL_REHOME_DEFAULT_HOST_COOLDOWN_MS}`,
   `ALTER TABLE relay_control_capabilities ADD COLUMN IF NOT EXISTS idle_regional_rehome BIGINT NOT NULL DEFAULT 0`,
-  `ALTER TABLE relay_region_rehome_attempts ADD COLUMN IF NOT EXISTS source_generation BIGINT NOT NULL DEFAULT 0`
+  `ALTER TABLE relay_region_rehome_attempts ADD COLUMN IF NOT EXISTS source_generation BIGINT NOT NULL DEFAULT 0`,
+  // Dropped, not created: see the comment on relay_assignment_activity_leases. Deferrable because
+  // this is the one boot where it has to take ACCESS EXCLUSIVE on a table under continuous write,
+  // and all 28 directors reach it at once; a lock timeout here must not restart the instance, which
+  // would only re-queue the same DDL behind the same writers. Once it wins, the pre-check answers
+  // absent and no later boot sends it at all.
+  `-- schema-deferrable: one boot has to win ACCESS EXCLUSIVE on a table written ~475/s
+   DROP INDEX IF EXISTS relay_assignment_activity_expiry`,
+  // The drop is what makes HOT legal; this is what makes it possible. A renewal can only reuse the
+  // row's own page when that page has room for a second version, and at the default fillfactor of
+  // 100 a freshly filled page has none - measured at 0.5% HOT with the index gone and the default,
+  // against 100% at 70. Takes SHARE UPDATE EXCLUSIVE, which blocks vacuum and DDL but no reader or
+  // writer, and only for the catalog write. Applies to pages as they refill, so the table converges
+  // over its own renewal cycle rather than at boot.
+  // Deferrable for the same reason, though SHARE UPDATE EXCLUSIVE blocks only vacuum and DDL: it
+  // buys nothing until the drop lands, so a boot that deferred the drop should defer this too.
+  `-- schema-deferrable: buys nothing until the drop above lands
+   ALTER TABLE relay_assignment_activity_leases SET (fillfactor = 70)`
 ]
 
 // The exact statement list a Postgres boot applies, in order, so the lock-target census can read

cloud/apps/relay/src/relay-schema-catalog-precheck-postgres.test.ts

@@ -1,5 +1,5 @@
 import pg from 'pg'
-import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   applyPostgresSchema,
   catalogObjectPresence,
@@ -190,6 +190,79 @@ describePostgres('relay boot-time schema against PostgreSQL', () => {
     }
   })
 
+  it('defers the activity-lease migrations and still boots while their table is locked', async () => {
+    // The migration boot, reproduced: the index is there, the table is locked by someone else, and
+    // all the drop can do is time out. It has to leave the statement for the next boot rather than
+    // fail, or 28 directors crash-loop through a stall on a table written ~475/s.
+    const cold = await openRelayDatabase({ databaseUrl: url, dataDir: '' })
+    opened.push(cold)
+    // Put the database back in its pre-migration shape, which is what makes the drop lock-taking.
+    await pool.query(
+      `CREATE INDEX relay_assignment_activity_expiry
+         ON ${schema}.relay_assignment_activity_leases(expires_at)`
+    )
+    await pool.query(`ALTER TABLE ${schema}.relay_assignment_activity_leases RESET (fillfactor)`)
+
+    const warned: string[] = []
+    const warn = vi.spyOn(console, 'warn').mockImplementation((line: string) => {
+      warned.push(line)
+    })
+    const holder = new pg.Client({ connectionString: url })
+    await holder.connect()
+    await holder.query('BEGIN')
+    await holder.query(
+      `LOCK TABLE ${schema}.relay_assignment_activity_leases IN ACCESS EXCLUSIVE MODE`
+    )
+    let summary: Awaited<ReturnType<typeof applyPostgresSchema>>
+    try {
+      summary = await applyPostgresSchema(
+        relayPostgresSchemaStatements(),
+        (statement) => pool.query(statement),
+        { catalogQuery: async (sql, params) => (await pool.query(sql, params)).rows }
+      )
+    } finally {
+      await holder.query('ROLLBACK')
+      await holder.end()
+      warn.mockRestore()
+    }
+
+    // Both statements deferred, and the boot still applied everything else.
+    expect(summary.deferred).toBe(2)
+    expect(summary.ran).toBeGreaterThan(0)
+    const deferred = warned
+      .map((line) => JSON.parse(line) as { event?: string; name?: string })
+      .filter((event) => event.event === 'orca_relay_postgres_schema_object_deferred')
+    expect(deferred.map((event) => event.name)).toEqual([
+      'relay_assignment_activity_expiry',
+      'fillfactor=70'
+    ])
+    // Nothing was applied, so the next boot has the same work to do, not half of it.
+    const stillThere = await pool.query(
+      `SELECT 1 FROM pg_indexes WHERE schemaname = $1 AND indexname = $2`,
+      [schema, 'relay_assignment_activity_expiry']
+    )
+    expect(stillThere.rowCount).toBe(1)
+
+    // And the next boot, with the lock gone, finishes the job.
+    const retry = await applyPostgresSchema(
+      relayPostgresSchemaStatements(),
+      (statement) => pool.query(statement),
+      { catalogQuery: async (sql, params) => (await pool.query(sql, params)).rows }
+    )
+    expect(retry.deferred).toBe(0)
+    const gone = await pool.query(
+      `SELECT 1 FROM pg_indexes WHERE schemaname = $1 AND indexname = $2`,
+      [schema, 'relay_assignment_activity_expiry']
+    )
+    expect(gone.rowCount).toBe(0)
+    const options = await pool.query(
+      `SELECT reloptions FROM pg_class WHERE oid = to_regclass($1)`,
+      [`${schema}.relay_assignment_activity_leases`]
+    )
+    expect(options.rows[0]?.reloptions).toEqual(['fillfactor=70'])
+    await pool.end()
+  })
+
   it('fails that same boot with 55P03 when the pre-check is not wired in', async () => {
     // Keeps the test above from passing vacuously: the lock really does block relay's DDL.
     const cold = await openRelayDatabase({ databaseUrl: url, dataDir: '' })

cloud/apps/relay/src/relay-schema-lock-targets.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from 'vitest'
 import {
   requireSchemaLockTarget,
+  schemaDeferrable,
   schemaLockTarget,
   sqlWithoutComments,
   takesRelationLock,
@@ -72,12 +73,6 @@ const GOLDEN_LOCK_TAKING: SchemaLockTarget[] = [
     name: 'relay_cell_drain_attempt_states_cell',
     skipWhen: 'present'
   },
-  {
-    kind: 'index',
-    table: 'relay_assignment_activity_leases',
-    name: 'relay_assignment_activity_expiry',
-    skipWhen: 'present'
-  },
   {
     kind: 'index',
     table: 'relay_control_connection_reservations',
@@ -116,7 +111,14 @@ const GOLDEN_LOCK_TAKING: SchemaLockTarget[] = [
   },
   { kind: 'column', table: 'relay_region_rehome_control', name: 'host_cooldown_ms', skipWhen: 'present' },
   { kind: 'column', table: 'relay_control_capabilities', name: 'idle_regional_rehome', skipWhen: 'present' },
-  { kind: 'column', table: 'relay_region_rehome_attempts', name: 'source_generation', skipWhen: 'present' }
+  { kind: 'column', table: 'relay_region_rehome_attempts', name: 'source_generation', skipWhen: 'present' },
+  { kind: 'index-by-name', name: 'relay_assignment_activity_expiry', skipWhen: 'absent' },
+  {
+    kind: 'reloption',
+    table: 'relay_assignment_activity_leases',
+    name: 'fillfactor=70',
+    skipWhen: 'present'
+  }
 ]
 
 const INDEX_OR_ADD_COLUMN = /^(?:CREATE\s+(?:UNIQUE\s+)?INDEX|ALTER\s+TABLE\s+[^\s]+\s+ADD\s+COLUMN)/i
@@ -149,8 +151,12 @@ describe('relay boot-time lock targets', () => {
     for (const statement of relayPostgresSchemaStatements()) {
       const target = schemaLockTarget(statement)
       if (!target) continue
-      expect(target.name).toMatch(/^[a-z_][a-z0-9_]*$/)
-      expect(target.table).toMatch(/^[a-z_][a-z0-9_]*$/)
+      // A reloption is the one target whose name is a pair rather than an identifier, because
+      // pg_class stores reloptions as `name=value` text and the value is half the question.
+      const shape = target.kind === 'reloption' ? /^[a-z_][a-z0-9_]*=[A-Za-z0-9_.]+$/ : /^[a-z_][a-z0-9_]*$/
+      expect(target.name).toMatch(shape)
+      // A DROP INDEX names no table, so there is none to check.
+      if (target.kind !== 'index-by-name') expect(target.table).toMatch(/^[a-z_][a-z0-9_]*$/)
     }
   })
 
@@ -194,7 +200,48 @@ describe('relay boot-time lock targets', () => {
 
   it('leaves every statement classifiable once its leading comments are stripped', () => {
     for (const statement of relayPostgresSchemaStatements()) {
-      expect(sqlWithoutComments(statement)).toMatch(/^(?:CREATE|ALTER|DO)\s/i)
+      expect(sqlWithoutComments(statement)).toMatch(/^(?:CREATE|ALTER|DROP|DO)\s/i)
     }
   })
+
+  it('pre-checks the activity-expiry drop by name, and skips it once the index is gone', () => {
+    // A DROP INDEX takes ACCESS EXCLUSIVE on the index's table for as long as the index is there,
+    // so it is in the census like any other lock-taking statement. Its target resolves by name
+    // alone, because the statement names no table and needs none.
+    const drops = relayPostgresSchemaStatements().filter((statement) =>
+      /^DROP\s/i.test(sqlWithoutComments(statement))
+    )
+    expect(drops.map(sqlWithoutComments)).toEqual([
+      'DROP INDEX IF EXISTS relay_assignment_activity_expiry'
+    ])
+    for (const statement of drops) {
+      expect(takesRelationLock(statement)).toBe(true)
+      expect(schemaLockTarget(statement)).toEqual({
+        kind: 'index-by-name',
+        name: 'relay_assignment_activity_expiry',
+        skipWhen: 'absent'
+      })
+    }
+  })
+
+  it('marks both activity-lease migrations deferrable, and nothing else', () => {
+    // The two statements a lock timeout must not turn into a crash loop, and the only two: every
+    // other statement still fails the boot loudly, which is what keeps the marker meaningful.
+    const deferrable = relayPostgresSchemaStatements().filter(schemaDeferrable)
+    expect(deferrable.map(sqlWithoutComments)).toEqual([
+      'DROP INDEX IF EXISTS relay_assignment_activity_expiry',
+      'ALTER TABLE relay_assignment_activity_leases SET (fillfactor = 70)'
+    ])
+  })
+
+  it('no longer creates an index on the column every control renewal writes', () => {
+    // The regression this drop exists to prevent: re-adding it would make ~471 renewals/s non-HOT
+    // again. A CREATE anywhere in the schema naming that index fails here.
+    const creates = relayPostgresSchemaStatements().filter((statement) =>
+      /relay_assignment_activity_expiry/i.test(sqlWithoutComments(statement))
+    )
+    expect(creates.map(sqlWithoutComments)).toEqual([
+      'DROP INDEX IF EXISTS relay_assignment_activity_expiry'
+    ])
+  })
 })

cloud/packages/postgres-schema/src/apply-postgres-schema.test.ts

@@ -84,7 +84,7 @@ describe('applyPostgresSchema classification', () => {
       wait: async () => undefined
     })
     expect(query).toHaveBeenCalledTimes(3)
-    expect(summary).toEqual({ ran: 1, skipped: 0 })
+    expect(summary).toEqual({ ran: 1, skipped: 0, deferred: 0 })
   })
 
   it('treats an already-applied constraint as skipped rather than an error', async () => {
@@ -94,7 +94,110 @@ describe('applyPostgresSchema classification', () => {
       throw postgresError('42710')
     })
     const summary = await applyPostgresSchema(['ALTER TABLE t ADD CONSTRAINT c CHECK (x > 0)'], query)
-    expect(summary).toEqual({ ran: 0, skipped: 1 })
+    expect(summary).toEqual({ ran: 0, skipped: 1, deferred: 0 })
+  })
+
+  it('treats an index another director already dropped as skipped rather than an error', async () => {
+    // Every director boots at once on a deploy and all of them send the same DROP INDEX IF EXISTS.
+    // Only one can win; the losers must not fail their boot over a drop that already happened.
+    const query = vi.fn(async () => {
+      throw postgresError('42704')
+    })
+    const summary = await applyPostgresSchema(['DROP INDEX IF EXISTS i'], query)
+    expect(summary).toEqual({ ran: 0, skipped: 1, deferred: 0 })
+    expect(query).toHaveBeenCalledTimes(1)
+  })
+
+  it('still propagates 42704 from a statement that is not a DROP IF EXISTS', async () => {
+    // Keeps the case above narrow: an undefined object anywhere else is a real boot failure.
+    const query = vi.fn(async () => {
+      throw postgresError('42704')
+    })
+    await expect(
+      applyPostgresSchema(['ALTER TABLE t ADD COLUMN IF NOT EXISTS c BIGINT'], query)
+    ).rejects.toThrow(/42704/)
+  })
+
+  it('leaves a deferrable statement unapplied on a lock timeout instead of failing the boot', async () => {
+    // The crash loop this prevents: 28 directors reach the same DROP INDEX at once on a table
+    // under continuous write, all of them time out, and every one restarts to re-queue the same
+    // DDL behind the same writers.
+    const warned: string[] = []
+    vi.spyOn(console, 'warn').mockImplementation((line: string) => {
+      warned.push(line)
+    })
+    const query = vi.fn(async () => {
+      throw postgresError('55P03')
+    })
+    const summary = await applyPostgresSchema(
+      ['-- schema-deferrable: reason\nDROP INDEX IF EXISTS i'],
+      query
+    )
+    expect(summary).toEqual({ ran: 0, skipped: 0, deferred: 1 })
+    expect(query).toHaveBeenCalledTimes(1)
+    const event = JSON.parse(warned[warned.length - 1] ?? '{}')
+    expect(event.event).toBe('orca_relay_postgres_schema_object_deferred')
+    expect(event.code).toBe('55P03')
+    expect(event.name).toBe('i')
+  })
+
+  it('runs the statements after a deferral, rather than abandoning the boot at that point', async () => {
+    // A deferral is not a failure, so nothing behind it may be skipped: the schema still has
+    // tables to create, and a boot that stopped here would come up against a partial schema.
+    const sent: string[] = []
+    const query = vi.fn(async (statement: string) => {
+      sent.push(statement)
+      if (statement.includes('DROP INDEX')) throw postgresError('55P03')
+      return undefined
+    })
+    vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const summary = await applyPostgresSchema(
+      ['-- schema-deferrable: reason\nDROP INDEX IF EXISTS i', 'CREATE TABLE IF NOT EXISTS t (id TEXT)'],
+      query
+    )
+    expect(summary).toEqual({ ran: 1, skipped: 0, deferred: 1 })
+    expect(sent).toHaveLength(2)
+  })
+
+  it('still fails the boot on a lock timeout for a statement that is not marked deferrable', async () => {
+    // Keeps the marker meaningful. An unmarked statement retains the old contract: fail once and
+    // loudly, because retrying parks every writer behind the same queue again.
+    vi.spyOn(console, 'error').mockImplementation(() => undefined)
+    const query = vi.fn(async () => {
+      throw postgresError('55P03')
+    })
+    await expect(applyPostgresSchema(['DROP INDEX IF EXISTS i'], query)).rejects.toMatchObject({
+      code: '55P03'
+    })
+  })
+
+  it('defers only on a lock timeout, not on any other error from a deferrable statement', async () => {
+    // A deferrable statement is not a statement whose failures stop mattering. A permission error
+    // is still a boot failure.
+    const query = vi.fn(async () => {
+      throw postgresError('42501')
+    })
+    await expect(
+      applyPostgresSchema(['-- schema-deferrable: reason\nDROP INDEX IF EXISTS i'], query)
+    ).rejects.toThrow(/42501/)
+  })
+
+  it('asks the catalog for a dropped index by name and skips the DROP once it is gone', async () => {
+    const query = vi.fn(async (_statement: string) => undefined)
+    const { catalogQuery, asked } = catalogAnswers([])
+    const summary = await applyPostgresSchema(['DROP INDEX IF EXISTS i'], query, { catalogQuery })
+    // One parameter, the index name: the statement names no table, and the query references no $2.
+    expect(asked).toEqual([[expect.stringContaining("relkind = 'i'"), 'i']])
+    expect(query).not.toHaveBeenCalled()
+    expect(summary).toEqual({ ran: 0, skipped: 1, deferred: 0 })
+  })
+
+  it('sends the DROP while the index is still there, which is the boot that has to win', async () => {
+    const query = vi.fn(async (_statement: string) => undefined)
+    const { catalogQuery } = catalogAnswers([{}])
+    const summary = await applyPostgresSchema(['DROP INDEX IF EXISTS i'], query, { catalogQuery })
+    expect(query).toHaveBeenCalledTimes(1)
+    expect(summary).toEqual({ ran: 1, skipped: 0, deferred: 0 })
   })
 
   it('propagates an unrelated error without retrying', async () => {
@@ -168,7 +271,7 @@ describe('applyPostgresSchema catalog pre-check', () => {
     expect(asked).toEqual([
       [expect.stringContaining('pg_catalog.pg_index'), 'relay_connection_bases', 'relay_bases_active']
     ])
-    expect(summary).toEqual({ ran: 1, skipped: 1 })
+    expect(summary).toEqual({ ran: 1, skipped: 1, deferred: 0 })
   })
 
   it('skips an index the catalog reports as invalid rather than rebuilding it', async () => {
@@ -206,7 +309,8 @@ describe('applyPostgresSchema catalog pre-check', () => {
     expect(JSON.parse(logged[logged.length - 1] ?? '{}')).toEqual({
       event: 'orca_push_postgres_schema_applied',
       ran: 1,
-      skipped: 1
+      skipped: 1,
+      deferred: 0
     })
   })
 
@@ -219,7 +323,7 @@ describe('applyPostgresSchema catalog pre-check', () => {
       [expect.stringContaining('pg_catalog.pg_attribute'), 'relay_control_capabilities', 'idle']
     ])
     expect(query.mock.calls.map(([sql]) => sql)).toEqual([statement])
-    expect(summary).toEqual({ ran: 1, skipped: 0 })
+    expect(summary).toEqual({ ran: 1, skipped: 0, deferred: 0 })
   })
 
   it('never probes the catalog for a statement that takes no relation lock', async () => {
@@ -247,7 +351,7 @@ describe('applyPostgresSchema catalog pre-check', () => {
       ]
     ])
     expect(query).not.toHaveBeenCalled()
-    expect(summary).toEqual({ ran: 0, skipped: 1 })
+    expect(summary).toEqual({ ran: 0, skipped: 1, deferred: 0 })
   })
 
   it('skips a DROP CONSTRAINT IF EXISTS when the constraint is already gone', async () => {
@@ -265,7 +369,7 @@ describe('applyPostgresSchema catalog pre-check', () => {
       { catalogQuery }
     )
     expect(query).not.toHaveBeenCalled()
-    expect(summary).toEqual({ ran: 0, skipped: 1 })
+    expect(summary).toEqual({ ran: 0, skipped: 1, deferred: 0 })
     expect(logged).toContainEqual({
       event: 'orca_relay_postgres_schema_object_absent',
       kind: 'constraint',
@@ -281,7 +385,7 @@ describe('applyPostgresSchema catalog pre-check', () => {
     const statement = 'ALTER TABLE t DROP CONSTRAINT IF EXISTS region_check'
     const summary = await applyPostgresSchema([statement], query, { catalogQuery })
     expect(query.mock.calls.map(([sql]) => sql)).toEqual([statement])
-    expect(summary).toEqual({ ran: 1, skipped: 0 })
+    expect(summary).toEqual({ ran: 1, skipped: 0, deferred: 0 })
   })
 
   it('sends an ADD CONSTRAINT the catalog does not name yet', async () => {
@@ -290,14 +394,14 @@ describe('applyPostgresSchema catalog pre-check', () => {
     const statement = 'ALTER TABLE t ADD CONSTRAINT region_valid CHECK (r IN (1))'
     const summary = await applyPostgresSchema([statement], query, { catalogQuery })
     expect(query.mock.calls.map(([sql]) => sql)).toEqual([statement])
-    expect(summary).toEqual({ ran: 1, skipped: 0 })
+    expect(summary).toEqual({ ran: 1, skipped: 0, deferred: 0 })
   })
 
   it('sends every statement when no catalog query is supplied', async () => {
     const query = vi.fn(async (_statement: string) => undefined)
     const summary = await applyPostgresSchema([COMMENTED_TABLE, COMMENTED_INDEX], query)
     expect(query).toHaveBeenCalledTimes(2)
-    expect(summary).toEqual({ ran: 2, skipped: 0 })
+    expect(summary).toEqual({ ran: 2, skipped: 0, deferred: 0 })
   })
 })
 
@@ -316,7 +420,7 @@ describe('applyPostgresSchema concurrent creates', () => {
     })
     expect(query).toHaveBeenCalledTimes(1)
     expect(asked).toHaveLength(2)
-    expect(summary).toEqual({ ran: 0, skipped: 1 })
+    expect(summary).toEqual({ ran: 0, skipped: 1, deferred: 0 })
   })
 
   it('still retries when the catalog says the object is not there after all', async () => {
@@ -332,7 +436,7 @@ describe('applyPostgresSchema concurrent creates', () => {
       wait: async () => undefined
     })
     expect(query).toHaveBeenCalledTimes(2)
-    expect(summary).toEqual({ ran: 1, skipped: 0 })
+    expect(summary).toEqual({ ran: 1, skipped: 0, deferred: 0 })
   })
 
   it('retries a CREATE TABLE collision without a catalog re-ask, having no target to ask about', async () => {

cloud/packages/postgres-schema/src/apply-postgres-schema.ts

@@ -25,7 +25,7 @@ export type SchemaStartupOptions = {
   wait?: (delayMs: number) => Promise<void>
 }
 
-export type SchemaApplySummary = { ran: number; skipped: number }
+export type SchemaApplySummary = { ran: number; skipped: number; deferred: number }
 
 function retryDelayMs(attempt: number, random: () => number): number {
   const ceiling = Math.min(RETRY_BASE_DELAY_MS * 2 ** (attempt - 1), RETRY_MAX_DELAY_MS)
@@ -39,6 +39,17 @@ function wait(delayMs: number): Promise<void> {
 const CREATE_TABLE_IF_NOT_EXISTS = /^CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\b/i
 const CREATE_INDEX_IF_NOT_EXISTS = /^CREATE\s+(?:UNIQUE\s+)?INDEX\s+IF\s+NOT\s+EXISTS\b/i
 const ALTER_TABLE_ADD_CONSTRAINT = /^ALTER\s+TABLE\s+\S+\s+ADD\s+CONSTRAINT\b/i
+const DROP_INDEX_IF_EXISTS = /^DROP\s+INDEX\s+(?:CONCURRENTLY\s+)?IF\s+EXISTS\b/i
+
+// Marked in the schema text, beside the SQL it applies to, and read from the raw statement because
+// classification strips comments. Says: this boot may leave the statement unapplied rather than
+// fail. Only sound for a statement that is idempotent AND that nothing this boot goes on to do
+// depends on, because the database is then simply as it was and the next boot re-sends it.
+const DEFERRABLE = /^\s*--[^\n]*\bschema-deferrable\b/
+
+export function schemaDeferrable(statement: string): boolean {
+  return DEFERRABLE.test(statement)
+}
 
 // `IF NOT EXISTS` only checks the name before the catalog inserts, so the loser of a concurrent
 // CREATE can fail on the catalog unique index (23505) or, when the winner has already committed by
@@ -70,6 +81,14 @@ function constraintAlreadyApplied(error: unknown, sql: string): boolean {
   )
 }
 
+// `IF EXISTS` resolves the name, then locks; between those two steps another director's drop can
+// commit and the loser raises 42704 instead of the notice it would have got a moment later. Every
+// director boots at once on a deploy, so without this the losers fail their boot over a drop that
+// already happened.
+function dropAlreadyApplied(error: unknown, sql: string): boolean {
+  return DROP_INDEX_IF_EXISTS.test(sql) && (error as { code?: unknown } | null)?.code === '42704'
+}
+
 function retryableSchemaError(error: unknown, sql: string): boolean {
   const value = (error as { code?: unknown; constraint?: unknown } | null) ?? {}
   return RETRYABLE_SCHEMA_CODES.has(String(value.code)) || concurrentCreateCollision(value, sql)
@@ -90,7 +109,7 @@ async function nothingToDo(
     JSON.stringify({
       event: `${eventPrefix}_object_${target.skipWhen}`,
       kind: target.kind,
-      table: target.table,
+      table: target.kind === 'index-by-name' ? undefined : target.table,
       name: target.name,
       indisvalid: presence.indisvalid
     })
@@ -108,7 +127,7 @@ export async function applyPostgresSchema(
   const random = options.random ?? Math.random
   const pause = options.wait ?? wait
   const deadlineAt = now() + (options.retryDeadlineMs ?? DEFAULT_RETRY_DEADLINE_MS)
-  const summary: SchemaApplySummary = { ran: 0, skipped: 0 }
+  const summary: SchemaApplySummary = { ran: 0, skipped: 0, deferred: 0 }
 
   for (const statement of statements) {
     // Throws when an index or column statement's target cannot be read, rather than sending it
@@ -126,7 +145,7 @@ export async function applyPostgresSchema(
         summary.ran += 1
         break
       } catch (error) {
-        if (constraintAlreadyApplied(error, sql)) {
+        if (constraintAlreadyApplied(error, sql) || dropAlreadyApplied(error, sql)) {
           summary.skipped += 1
           break
         }
@@ -135,6 +154,24 @@ export async function applyPostgresSchema(
         // this boot lost the queue. Relation locks are granted in queue order, so each retry parks
         // every writer behind it again for another timeout. Fail once, loudly.
         if (code === LOCK_NOT_AVAILABLE && !options.retryLockTimeout) {
+          // A deferrable statement yields the queue instead of crash-looping the instance. Every
+          // director boots at once on a migration, so a table under continuous write can hand the
+          // whole fleet a lock timeout on the one statement that has to win once; failing the boot
+          // for it restarts the instance, which re-queues the same DDL behind the same writers.
+          if (schemaDeferrable(statement)) {
+            console.warn(
+              JSON.stringify({
+                event: `${eventPrefix}_object_deferred`,
+                code,
+                kind: target?.kind,
+                name: target?.name,
+                statement: sql.split('\n')[0],
+                detail: 'could not take its lock; left unapplied for the next boot to retry'
+              })
+            )
+            summary.deferred += 1
+            break
+          }
           console.error(
             JSON.stringify({
               event: `${eventPrefix}_lock_timeout`,

cloud/packages/postgres-schema/src/catalog-object-precheck.ts

@@ -26,10 +26,26 @@ WHERE attrelid = to_regclass($1) AND attname = $2 AND attnum > 0 AND NOT attisdr
 const CONSTRAINT_PRESENT = `SELECT 1 FROM pg_catalog.pg_constraint
 WHERE conrelid = to_regclass($1) AND conname = $2`
 
+// By name through the search_path, with no table condition, because a DROP INDEX has no table to
+// condition on and does not need one: a name that resolves to no visible index is nothing to drop.
+// `relkind = 'i'` keeps a same-named table or view from answering for an index. Partitioned indexes
+// are 'I', which this deliberately does not match - relay has none, and dropping one is not a
+// boot-time operation.
+const INDEX_BY_NAME_PRESENT = `SELECT 1 FROM pg_catalog.pg_class c
+WHERE c.relname = $1 AND c.relkind = 'i' AND pg_catalog.pg_table_is_visible(c.oid)`
+
+// reloptions is a text[] of `name=value` pairs, absent entirely while the option is at its
+// default. Comparing the whole pair is what makes a changed value re-run: `@>` on a different
+// value answers no, and the statement runs and overwrites it.
+const RELOPTION_PRESENT = `SELECT 1 FROM pg_catalog.pg_class
+WHERE oid = to_regclass($1) AND reloptions @> ARRAY[$2]`
+
 const PRESENCE_SQL = {
   index: INDEX_PRESENT,
   column: COLUMN_PRESENT,
-  constraint: CONSTRAINT_PRESENT
+  constraint: CONSTRAINT_PRESENT,
+  reloption: RELOPTION_PRESENT,
+  'index-by-name': INDEX_BY_NAME_PRESENT
 } as const
 
 export type SchemaCatalogPresence = { present: boolean; indisvalid: unknown }
@@ -42,7 +58,10 @@ export async function catalogObjectPresence(
   target: SchemaLockTarget
 ): Promise<SchemaCatalogPresence> {
   const sql = PRESENCE_SQL[target.kind]
-  const rows = await query(sql, [target.table, target.name])
+  // The name-only lookup binds one parameter; every other shape binds the table first. Passing a
+  // parameter the SQL never references is a bind error, not a harmless extra.
+  const params = target.kind === 'index-by-name' ? [target.name] : [target.table, target.name]
+  const rows = await query(sql, params)
   const row = rows[0]
   return row ? { present: true, indisvalid: row.indisvalid } : { present: false, indisvalid: undefined }
 }

cloud/packages/postgres-schema/src/index.ts

@@ -1,5 +1,6 @@
 export {
   applyPostgresSchema,
+  schemaDeferrable,
   type SchemaApplySummary,
   type SchemaStartupOptions
 } from './apply-postgres-schema.js'

cloud/packages/postgres-schema/src/schema-lock-target.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from 'vitest'
+import { schemaDeferrable } from './apply-postgres-schema.js'
 import {
   requireSchemaLockTarget,
   schemaLockTarget,
@@ -417,3 +418,105 @@ describe('dollar-quoted bodies', () => {
     )
   })
 })
+
+describe('schemaLockTarget storage parameters', () => {
+  it('reads a storage parameter as a name=value target the catalog can be asked about', () => {
+    expect(schemaLockTarget('ALTER TABLE t SET (fillfactor = 70)')).toEqual({
+      kind: 'reloption',
+      table: 't',
+      name: 'fillfactor=70',
+      skipWhen: 'present'
+    })
+  })
+
+  it('folds the option name but keeps the value as written, the way pg_class stores the pair', () => {
+    expect(schemaLockTarget('ALTER TABLE t SET (FillFactor=70)')?.name).toBe('fillfactor=70')
+  })
+
+  it('makes a changed value a different target, so it re-runs instead of skipping', () => {
+    // The failure this prevents: matching on the option name alone would read `fillfactor=100` as
+    // already satisfying `fillfactor = 70` and skip the statement for the life of the database.
+    const seventy = schemaLockTarget('ALTER TABLE t SET (fillfactor = 70)')
+    const eighty = schemaLockTarget('ALTER TABLE t SET (fillfactor = 80)')
+    expect(seventy?.name).not.toBe(eighty?.name)
+  })
+
+  it('refuses a multi-option SET rather than skipping on only the first option', () => {
+    // Same reason a multi-action ALTER TABLE is refused: skipping on one option would silently
+    // drop the others for good.
+    expect(() =>
+      requireSchemaLockTarget('ALTER TABLE t SET (fillfactor = 70, autovacuum_enabled = false)')
+    ).toThrow(/unparsed_schema_lock_target/)
+  })
+
+  it('fails the boot on a SET whose shape it cannot read, rather than sending it unchecked', () => {
+    // A storage parameter takes a relation lock, so no target means the lock is taken on every
+    // boot. RESET has no value to compare and is not supported.
+    expect(() => requireSchemaLockTarget('ALTER TABLE t RESET (fillfactor)')).not.toThrow()
+    expect(() => requireSchemaLockTarget('ALTER TABLE t SET (fillfactor)')).toThrow(
+      /unparsed_schema_lock_target/
+    )
+  })
+
+  it('takes a relation lock, so the census requires it to carry a target', () => {
+    expect(takesRelationLock('ALTER TABLE t SET (fillfactor = 70)')).toBe(true)
+  })
+})
+
+describe('schemaLockTarget dropped indexes', () => {
+  it('resolves a dropped index by name, with no table to name', () => {
+    expect(schemaLockTarget('DROP INDEX IF EXISTS i')).toEqual({
+      kind: 'index-by-name',
+      name: 'i',
+      skipWhen: 'absent'
+    })
+  })
+
+  it('reads CONCURRENTLY as a modifier rather than the index name', () => {
+    expect(schemaLockTarget('DROP INDEX CONCURRENTLY IF EXISTS i')?.name).toBe('i')
+  })
+
+  it('folds an unquoted name and keeps a quoted one, the way relname stores it', () => {
+    expect(schemaLockTarget('DROP INDEX IF EXISTS MyIndex')?.name).toBe('myindex')
+    expect(schemaLockTarget('DROP INDEX IF EXISTS "MyIndex"')?.name).toBe('MyIndex')
+  })
+
+  it('takes a relation lock, because the index is there on the boot that has to drop it', () => {
+    expect(takesRelationLock('DROP INDEX IF EXISTS i')).toBe(true)
+  })
+
+  it('requires IF EXISTS, so a bare DROP fails the boot instead of running unchecked', () => {
+    // Same contract as DROP CONSTRAINT: a bare DROP on a missing index is an error the server is
+    // supposed to raise, and a pre-check that skipped it would swallow that.
+    expect(() => requireSchemaLockTarget('DROP INDEX i')).toThrow(/unparsed_schema_lock_target/)
+  })
+
+  it('refuses a multi-index DROP rather than pre-checking only the first name', () => {
+    // Skipping on one name would leave the other index in place for the life of the database.
+    expect(() => requireSchemaLockTarget('DROP INDEX IF EXISTS a, b')).toThrow(
+      /unparsed_schema_lock_target/
+    )
+  })
+
+  it('derives the target through a leading deferrable marker', () => {
+    // The real shape in relay's schema: the marker is a comment, so classification must see past
+    // it or the statement would reach the server with no pre-check at all.
+    const statement = '-- schema-deferrable: reason\nDROP INDEX IF EXISTS i'
+    expect(sqlWithoutComments(statement)).toBe('DROP INDEX IF EXISTS i')
+    expect(schemaLockTarget(statement)?.name).toBe('i')
+  })
+})
+
+describe('schemaDeferrable', () => {
+  it('reads the marker only from a leading comment, never from the SQL body', () => {
+    // A name or a string containing the word must not make a statement deferrable.
+    expect(schemaDeferrable('-- schema-deferrable: reason\nDROP INDEX IF EXISTS i')).toBe(true)
+    expect(schemaDeferrable('DROP INDEX IF EXISTS schema_deferrable')).toBe(false)
+    expect(schemaDeferrable("CREATE TABLE t (c TEXT DEFAULT 'schema-deferrable')")).toBe(false)
+  })
+
+  it('treats an unmarked statement as fatal on a lock timeout, which is the default', () => {
+    expect(schemaDeferrable('DROP INDEX IF EXISTS i')).toBe(false)
+    expect(schemaDeferrable('ALTER TABLE t SET (fillfactor = 70)')).toBe(false)
+  })
+})

cloud/packages/postgres-schema/src/schema-lock-target.ts

@@ -2,14 +2,20 @@
 // already exists before the statement joins the lock queue. `table` is kept exactly as the
 // statement wrote it, schema qualification and quoting included, because it is fed to
 // `to_regclass`; `name` is the bare identifier the catalog stores in `relname`/`attname`.
-export type SchemaLockTarget = {
-  kind: 'index' | 'column' | 'constraint'
-  table: string
-  name: string
-  // The catalog answer that means this statement has nothing left to do. Creating statements skip
-  // on present; `DROP CONSTRAINT IF EXISTS` is the inverse, because nothing to drop is done.
-  skipWhen: 'present' | 'absent'
-}
+export type SchemaLockTarget =
+  | {
+      kind: 'index' | 'column' | 'constraint' | 'reloption'
+      table: string
+      name: string
+      // The catalog answer that means this statement has nothing left to do. Creating statements
+      // skip on present; `DROP CONSTRAINT IF EXISTS` is the inverse, because nothing to drop is
+      // done.
+      skipWhen: 'present' | 'absent'
+    }
+  // A `DROP INDEX` names no table, and needs none: an index name that resolves to nothing is
+  // nothing to drop, whatever table it used to belong to. Resolution is by name through the
+  // search_path, which is how the DROP itself would resolve it.
+  | { kind: 'index-by-name'; name: string; skipWhen: 'absent' }
 
 // Keywords that sit in an identifier position when the optional clause before them is absent.
 // Without this, `CREATE UNIQUE INDEX CONCURRENTLY ON t(c)` reads CONCURRENTLY as the index name and
@@ -119,10 +125,27 @@ const DROP_CONSTRAINT = new RegExp(
   'i'
 )
 
+// `IF EXISTS` is required for the same reason it is on DROP CONSTRAINT: a bare `DROP INDEX` on a
+// missing index is an error the server is supposed to raise. Without a target the statement throws
+// at boot instead, which tells the author to write `IF EXISTS`.
+const DROP_INDEX = new RegExp(`^DROP\\s+INDEX\\s+(?:CONCURRENTLY\\s+)?IF\\s+EXISTS\\s+${QUALIFIED}\\s*$`, 'i')
+
+// One option per statement, and a literal value: the catalog stores reloptions as `name=value`
+// text, so the pre-check compares the written pair against that array verbatim. A list of options
+// is refused by `hasTopLevelComma` before it reaches here, the same as a multi-action ALTER TABLE.
+const SET_RELOPTION = new RegExp(
+  `^ALTER\\s+TABLE\\s+(?:IF\\s+EXISTS\\s+)?(?:ONLY\\s+)?${QUALIFIED}\\s+` +
+    `SET\\s+\\(\\s*([A-Za-z_][A-Za-z0-9_]*)\\s*=\\s*([A-Za-z0-9_.]+)\\s*\\)\\s*$`,
+  'i'
+)
+
 // Every statement shape that takes a relation lock before Postgres evaluates its existence test.
 // `CREATE TABLE IF NOT EXISTS` is absent on purpose: it resolves a name against the schema and
 // takes no lock on an existing table.
-const TAKES_RELATION_LOCK = /^(?:CREATE\s+(?:UNIQUE\s+)?INDEX|ALTER\s+TABLE)\b/i
+// `DROP INDEX` is here because it takes ACCESS EXCLUSIVE on the index's table whenever the index is
+// actually there, which is every boot until the first one wins. That it takes no lock once the
+// index is gone is what the pre-check turns into the steady state, not a reason to omit it.
+const TAKES_RELATION_LOCK = /^(?:CREATE\s+(?:UNIQUE\s+)?INDEX|DROP\s+INDEX|ALTER\s+TABLE)\b/i
 
 export function takesRelationLock(statement: string): boolean {
   return TAKES_RELATION_LOCK.test(sqlWithoutComments(statement))
@@ -177,7 +200,9 @@ const MUST_PARSE = [
   /^CREATE\s+(?:UNIQUE\s+)?INDEX\b/i,
   /^ALTER\s+TABLE\b[\s\S]*\bADD\s+COLUMN\b/i,
   /^ALTER\s+TABLE\b[\s\S]*\bADD\s+CONSTRAINT\b/i,
-  /^ALTER\s+TABLE\b[\s\S]*\bDROP\s+CONSTRAINT\b/i
+  /^ALTER\s+TABLE\b[\s\S]*\bDROP\s+CONSTRAINT\b/i,
+  /^ALTER\s+TABLE\b[\s\S]*\bSET\s+\(/i,
+  /^DROP\s+INDEX\b/i
 ]
 
 // Derived from the statement itself so a renamed index cannot drift away from its pre-check.
@@ -209,6 +234,22 @@ export function schemaLockTarget(statement: string): SchemaLockTarget | undefine
       skipWhen: 'absent'
     }
   }
+  const droppedIndex = DROP_INDEX.exec(sql)
+  if (droppedIndex?.[1]) {
+    return { kind: 'index-by-name', name: catalogName(droppedIndex[1]), skipWhen: 'absent' }
+  }
+  const option = SET_RELOPTION.exec(sql)
+  if (option?.[1] && option[2] && option[3]) {
+    // Option names are always folded, but the value is stored as written, so only the name goes
+    // through catalogName. `fillfactor=70` and `fillfactor=80` are different targets, which is
+    // what makes a changed value re-run rather than skip.
+    return {
+      kind: 'reloption',
+      table: option[1],
+      name: `${catalogName(option[2])}=${option[3]}`,
+      skipWhen: 'present'
+    }
+  }
   return undefined
 }