From 3f17d6b9af4086151f511ffce84a3740153128b8 Mon Sep 17 00:00:00 2001
From: philipp <klein.philipp@gmail.com>
Date: Tue, 16 Jun 2026 17:21:44 +0200
Subject: [PATCH] fix: add lock_timeout to schema index creation, scope
 inventory by time
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two fixes for history-writer stability:

1. ensureHistorySchema / ensureSchemaIndex: CREATE INDEX IF NOT EXISTS
   now uses a 5s lock_timeout via a dedicated pool client. When multiple
   pods restart in a crash loop, each queued a CREATE INDEX that blocked
   on the previous pod's lock, creating a cascade of 19+ stuck queries
   (observed blocking for 3000+ seconds). With lock_timeout, the pod
   skips the index creation (it already exists) and proceeds to startup.

2. refreshQuoteOutcomes: inventory query is now scoped to the time
   window of the current submission batch ± 15min buffer instead of a
   blind LIMIT 50000. This drops memory from ~442MB to ~3MB for a
   typical batch, eliminating OOM kills on the 1280Mi pod.

Proof: history-writer OOM + startup lock cascade from ensureHistorySchema
Assumptions: indexes already exist in steady state; 15min buffer covers attribution window
Still fake: heuristic gap outcomes may attribute trades imprecisely
---
 src/lib/postgres.mjs | 50 +++++++++++++++++++++++++++++---------------
 1 file changed, 33 insertions(+), 17 deletions(-)

diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs
index 2b26a3d..8f14711 100644
--- a/src/lib/postgres.mjs
+++ b/src/lib/postgres.mjs
@@ -126,22 +126,32 @@ export async function ensureHistorySchema(pool) {
         raw JSONB
       )
     `);
-    await pool.query(`
-      CREATE INDEX IF NOT EXISTS ${table}_quote_id_idx
-      ON ${table} (quote_id)
-    `);
-    await pool.query(`
-      CREATE INDEX IF NOT EXISTS ${table}_decision_key_idx
-      ON ${table} (decision_key)
-    `);
-    await pool.query(`
-      CREATE INDEX IF NOT EXISTS ${table}_ingested_at_idx
-      ON ${table} (ingested_at DESC)
-    `);
-    await pool.query(`
-      CREATE INDEX IF NOT EXISTS ${table}_coalesced_at_idx
-      ON ${table} (COALESCE(observed_at, ingested_at) DESC)
-    `);
+    // Use a short lock_timeout so CREATE INDEX IF NOT EXISTS doesn't block
+    // startup indefinitely when another session holds a conflicting lock.
+    // The indexes already exist in steady state, so a timeout is harmless.
+    for (const indexSql of [
+      `CREATE INDEX IF NOT EXISTS ${table}_quote_id_idx ON ${table} (quote_id)`,
+      `CREATE INDEX IF NOT EXISTS ${table}_decision_key_idx ON ${table} (decision_key)`,
+      `CREATE INDEX IF NOT EXISTS ${table}_ingested_at_idx ON ${table} (ingested_at DESC)`,
+      `CREATE INDEX IF NOT EXISTS ${table}_coalesced_at_idx ON ${table} (COALESCE(observed_at, ingested_at) DESC)`,
+    ]) {
+      const client = await pool.connect();
+      try {
+        await client.query(`SET lock_timeout = '5s'`);
+        await client.query(indexSql);
+      } catch (error) {
+        if (error?.code === '55P03') {
+          // lock_not_available — another session holds the lock, skip
+          continue;
+        }
+        if (isConcurrentSchemaIndexExistsError(error)) continue;
+        throw error;
+      } finally {
+        // Reset lock_timeout before returning client to pool
+        try { await client.query(`SET lock_timeout = '0'`); } catch { /* ignore */ }
+        client.release();
+      }
+    }
   }
 
   await ensureExpressionIndex(pool, {
@@ -6448,11 +6458,17 @@ async function ensureExpressionIndex(pool, { name, table, expression }) {
 }
 
 export async function ensureSchemaIndex(pool, sql) {
+  const client = await pool.connect();
   try {
-    await pool.query(sql);
+    await client.query(`SET lock_timeout = '5s'`);
+    await client.query(sql);
   } catch (error) {
     if (isConcurrentSchemaIndexExistsError(error)) return;
+    if (error?.code === '55P03') return; // lock_not_available — skip gracefully
     throw error;
+  } finally {
+    try { await client.query(`SET lock_timeout = '0'`); } catch { /* ignore */ }
+    client.release();
   }
 }