app/ingestor.py

@@ -20,11 +20,6 @@ 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"
@@ -131,18 +126,15 @@ 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:
-        for i in range(0, len(rows), FIRE_INSERT_CHUNK):
-            chunk = rows[i:i + FIRE_INSERT_CHUNK]
         stmt = (
             pg_insert(fires_table)
-                .values(chunk)
+            .values(rows)
             .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

tests/test_fire_ingest.py

@@ -213,55 +213,3 @@ 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