From 5302e261d7ec8b7581148a59799597c9589aa335 Mon Sep 17 00:00:00 2001
From: Sirius DevOps <lance@siriusdevops.com>
Date: Fri, 28 Aug 2026 22:22:12 -0400
Subject: [PATCH] fix: chunk FIRMS bulk inserts under asyncpg 32767 bind cap

A ~90k-row INSERT ... ON CONFLICT dies with InterfaceError, so fire_loop
fails every poll and /api/health stays degraded (fires age > 30 min).
Chunk 2000 rows per statement, still one commit per poll.
---
 app/ingestor.py           | 22 +++++++++++------
 tests/test_fire_ingest.py | 52 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 67 insertions(+), 7 deletions(-)

diff --git a/app/ingestor.py b/app/ingestor.py
index 8d7eb97..ac0bcbd 100644
--- a/app/ingestor.py
+++ b/app/ingestor.py
@@ -20,6 +20,11 @@ from sources import event_dedup_key
 
 logger = logging.getLogger("osint.ingestor")
 
+# asyncpg rejects statements with >32767 bind params. A FIRMS poll is ~90k
+# rows × 14 columns. Chunk inserts; still one transaction / one commit.
+FIRE_ROW_BIND_PARAMS = 14
+FIRE_INSERT_CHUNK = 2000
+
 # NATS connection settings
 NATS_URLS = NATS_URL
 NATS_STREAM = "events"
@@ -126,15 +131,18 @@ async def ingest_fire_rows(msgs: list[dict]) -> int:
             rows.append(row)
     if not rows:
         return 0
+    inserted = 0
     async with async_session() as session:
-        stmt = (
-            pg_insert(fires_table)
-            .values(rows)
-            .on_conflict_do_nothing(constraint="pk_fires_natural_key")
-        )
-        result = await session.execute(stmt)
+        for i in range(0, len(rows), FIRE_INSERT_CHUNK):
+            chunk = rows[i:i + FIRE_INSERT_CHUNK]
+            stmt = (
+                pg_insert(fires_table)
+                .values(chunk)
+                .on_conflict_do_nothing(constraint="pk_fires_natural_key")
+            )
+            result = await session.execute(stmt)
+            inserted += int(result.rowcount or 0)
         await session.commit()
-    inserted = int(result.rowcount or 0)
     if inserted:
         logger.info("bulk ingested %d/%d FIRMS hotspots", inserted, len(rows))
         from live_layers import aircraft_last_known
diff --git a/tests/test_fire_ingest.py b/tests/test_fire_ingest.py
index 29a1950..51c41a4 100644
--- a/tests/test_fire_ingest.py
+++ b/tests/test_fire_ingest.py
@@ -213,3 +213,55 @@ def test_ingest_fire_rows_one_execute_one_commit(monkeypatch):
     assert inserted == 3
     assert session.executes == 1
     assert session.commits == 1
+
+
+def test_fire_insert_chunk_stays_under_asyncpg_bind_limit():
+    """asyncpg caps bind params at 32767 — a 93k-row INSERT dies."""
+    from ingestor import FIRE_INSERT_CHUNK, FIRE_ROW_BIND_PARAMS
+
+    assert FIRE_INSERT_CHUNK * FIRE_ROW_BIND_PARAMS < 32767
+    assert FIRE_INSERT_CHUNK >= 500
+
+
+def test_ingest_fire_rows_chunks_when_over_limit(monkeypatch):
+    from ingestor import ingest_fire_rows
+    import ingestor
+
+    class _Session:
+        def __init__(self):
+            self.executes = 0
+            self.commits = 0
+            self.rowcount = 0
+
+        async def execute(self, *a, **k):
+            self.executes += 1
+            self.rowcount = 2 if self.executes < 3 else 1
+            return self
+
+        async def commit(self):
+            self.commits += 1
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+    session = _Session()
+    monkeypatch.setattr(ingestor, "async_session", lambda: session)
+    monkeypatch.setattr(ingestor, "FIRE_INSERT_CHUNK", 2)
+
+    async def no_corr(*a, **k):
+        return []
+
+    monkeypatch.setattr("fire_aircraft.correlate_and_notify", no_corr)
+    monkeypatch.setattr("geofence.record_and_notify", no_corr)
+
+    msgs = [
+        make_fire_msg(latitude=39.1 + i * 0.01, longitude=-121.1)
+        for i in range(5)
+    ]
+    inserted = asyncio.run(ingest_fire_rows(msgs))
+    assert inserted == 5
+    assert session.executes == 3
+    assert session.commits == 1