Klima-Cold-Cell-Fetch nicht mehr innerhalb offener DB-Transaktion (#428)

open-meteo-climate.ts hielt den 30s-Archiv-Fetch bislang innerhalb einer
pg_advisory_xact_lock-Transaktion offen. Dadurch verlor der Fetch häufig
das Rennen gegen seinen eigenen Timeout, sobald die Verbindung mit anderer
Pool-Last konkurrierte — betraf vor allem DE, das mit Abstand die meisten
distinkten Klimazellen braucht (Coverage dort bei ~18% statt ~97-100% wie
in jedem anderen Land). Lock läuft jetzt session-scoped (pg_advisory_lock/
unlock) auf einem eigenen Client, der Fetch selbst läuft außerhalb jeder
offenen Transaktion.

Author: haexhub

server/utils/external-data/open-meteo-climate.test.ts

@@ -133,10 +133,10 @@ describe('aggregate', () => {
 })
 
 describe('createOpenMeteoClimateNormalsEnhancer / readClimateNormals', () => {
-  // Models the same-cell serialization a real pg_advisory_xact_lock gives: a
+  // Models the same-cell serialization a real pg_advisory_lock gives: a
   // second lock request for a key already held waits for the holder's
-  // COMMIT/ROLLBACK before it proceeds. A successful write updates `row` so
-  // the loser's re-read (inside the lock) sees the winner's cached result.
+  // pg_advisory_unlock before it proceeds. A successful write updates `row`
+  // so the loser's re-read (inside the lock) sees the winner's cached result.
   function fakePool(existingRow: Record<string, unknown> | null = null) {
     const inserted: unknown[][] = []
     let row = existingRow
@@ -154,13 +154,12 @@ describe('createOpenMeteoClimateNormalsEnhancer / readClimateNormals', () => {
       return vi.fn(async (queryArg: unknown, params: unknown[] = []) => {
         const text = typeof queryArg === 'string' ? queryArg : (queryArg as { text: string }).text
         const n = text.replace(/\s+/g, ' ').trim().toLowerCase()
-        if (n === 'begin') return { rows: [], rowCount: 0 }
-        if (n === 'commit' || n === 'rollback') {
+        if (n.includes('pg_advisory_unlock')) {
           releaseLock?.()
           releaseLock = null
           return { rows: [], rowCount: 0 }
         }
-        if (n.includes('pg_advisory_xact_lock')) {
+        if (n.includes('pg_advisory_lock')) {
           releaseLock = await acquireLock(String(params[0]))
           return { rows: [], rowCount: 0 }
         }
@@ -184,16 +183,13 @@ describe('createOpenMeteoClimateNormalsEnhancer / readClimateNormals', () => {
       })
     }
 
-    // drizzle's transaction() (withCellLock) only checks out its own
-    // connection when the object it wraps looks like a `pg.Pool` — it tests
-    // `instanceof Pool` or a constructor name containing "Pool" — so this
-    // needs a named constructor to take that branch and give each concurrent
-    // lock attempt its own `makeQuery()` closure, matching a real per-session
+    // withCellLock calls pool.connect() itself (no drizzle transaction
+    // dispatch involved), so each concurrent lock attempt just needs its own
+    // `makeQuery()` closure from `connect`, matching a real per-session
     // connection.
-    function MockPool() {}
     const query = makeQuery()
     const connect = vi.fn(async () => ({ query: makeQuery(), release: vi.fn() }))
-    const pool = Object.assign(new (MockPool as unknown as new () => object)(), { query, connect, inserted })
+    const pool = { query, connect, inserted }
     return pool as unknown as Pool & { inserted: unknown[][] }
   }
 

server/utils/external-data/open-meteo-climate.ts

@@ -13,7 +13,7 @@
 // Climate normals don't go stale (WP-7 doc: "einmal geholt, nie wieder"), so
 // a cell is fetched at most once, ever, keyed by its rounded coordinates.
 
-import type { Pool } from 'pg'
+import type { Pool, PoolClient } from 'pg'
 import { sql } from 'drizzle-orm'
 import { drizzle, type NodePgDatabase } from 'drizzle-orm/node-postgres'
 import type {
@@ -27,8 +27,8 @@ import { EXTERNAL_DATA_SOURCES } from './sources'
 import { fetchOpenMeteo } from './open-meteo-rate-limit'
 
 /** `readCachedCell`/`writeCachedCell` run both outside any lock (against the
- *  pool directly) and inside `withCellLock`'s transaction (against `tx`) — this
- *  is the common surface both a `NodePgDatabase` and its transaction share. */
+ *  pool directly) and inside `withCellLock`'s locked client, wrapped in its
+ *  own `drizzle()` instance — this is the common surface both share. */
 type Queryable = Pick<NodePgDatabase, 'execute'>
 
 export interface OpenMeteoClimateOptions {
@@ -92,31 +92,45 @@ export async function readClimateNormals(
   if (cached) return toLocationClimateNormals(cached, options.checkedAt)
 
   // Cold cell: two concurrent requests can both observe the miss above
-  // before either has written the row. A per-cell advisory lock (held for
-  // the transaction, scoped to the checked-out client) serializes them, and
-  // the re-read after acquiring it lets the loser of the race serve the
-  // winner's freshly-cached row instead of hitting Open-Meteo again.
-  return withCellLock(db, cell, async (tx) => {
-    const recached = await readCachedCell(tx, cell)
+  // before either has written the row. A per-cell *session* advisory lock
+  // (pg_advisory_lock/unlock on a dedicated checked-out client, not
+  // pg_advisory_xact_lock inside a BEGIN/COMMIT) serializes them without
+  // pinning that connection inside an open transaction for the whole
+  // archive fetch below — doing that previously left the fetch racing its
+  // own abort timeout against everything else contending for the
+  // connection pool, so cold cells routinely lost that race and never got
+  // cached at all (observed in prod: DE, with by far the most distinct
+  // cells to fetch, stuck at ~18% cached vs. 97-100% for every other,
+  // smaller country). The re-read after acquiring the lock lets the loser
+  // of the race serve the winner's freshly-cached row instead of hitting
+  // Open-Meteo again.
+  return withCellLock(options.db, cell, async (client) => {
+    const clientDb = drizzle(client)
+    const recached = await readCachedCell(clientDb, cell)
     if (recached) return toLocationClimateNormals(recached, options.checkedAt)
 
     const daily = await fetchDailySeries(cell, options)
     if (!daily) return null
     const data = aggregate(daily)
-    await writeCachedCell(tx, cell, data)
+    await writeCachedCell(clientDb, cell, data)
     return toLocationClimateNormals(data, options.checkedAt)
   })
 }
 
 async function withCellLock<T>(
-  db: NodePgDatabase,
+  pool: Pool,
   cell: { lat: number; lon: number },
-  fn: (tx: Queryable) => Promise<T>,
+  fn: (client: PoolClient) => Promise<T>,
 ): Promise<T> {
-  return db.transaction(async (tx) => {
-    await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext(${cellLockKey(cell)})::bigint)`)
-    return fn(tx)
-  })
+  const client = await pool.connect()
+  const key = cellLockKey(cell)
+  try {
+    await client.query('SELECT pg_advisory_lock(hashtext($1)::bigint)', [key])
+    return await fn(client)
+  } finally {
+    await client.query('SELECT pg_advisory_unlock(hashtext($1)::bigint)', [key]).catch(() => {})
+    client.release()
+  }
 }
 
 function cellLockKey(cell: { lat: number; lon: number }): string {