Be kinder about streaming volume in memory

app/repository/raw_packets.py

@@ -95,13 +95,22 @@ class RawPacketRepository:
         return row["oldest"] if row and row["oldest"] is not None else None
 
     @staticmethod
-    async def get_all_undecrypted() -> list[tuple[int, bytes, int]]:
-        """Get all undecrypted packets as (id, data, timestamp) tuples."""
+    async def stream_all_undecrypted(
+        batch_size: int = UNDECRYPTED_PACKET_BATCH_SIZE,
+    ) -> AsyncIterator[tuple[int, bytes, int]]:
+        """Yield all undecrypted packets as (id, data, timestamp) in bounded batches."""
         cursor = await db.conn.execute(
             "SELECT id, data, timestamp FROM raw_packets WHERE message_id IS NULL ORDER BY timestamp ASC"
         )
-        rows = await cursor.fetchall()
-        return [(row["id"], bytes(row["data"]), row["timestamp"]) for row in rows]
+        try:
+            while True:
+                rows = await cursor.fetchmany(batch_size)
+                if not rows:
+                    break
+                for row in rows:
+                    yield (row["id"], bytes(row["data"]), row["timestamp"])
+        finally:
+            await cursor.close()
 
     @staticmethod
     async def stream_undecrypted_text_messages(

app/routers/channels.py

@@ -122,8 +122,7 @@ def _normalize_bulk_hashtag_name(name: str) -> str | None:
 async def _run_historical_channel_decryption_for_channels(
     channels: list[tuple[bytes, str, str]],
 ) -> None:
-    packets = await RawPacketRepository.get_all_undecrypted()
-    total = len(packets)
+    total = await RawPacketRepository.get_undecrypted_count()
     decrypted_count = 0
     matched_channel_names: set[str] = set()
 
@@ -137,7 +136,11 @@ async def _run_historical_channel_decryption_for_channels(
         len(channels),
     )
 
-    for packet_id, packet_data, packet_timestamp in packets:
+    async for (
+        packet_id,
+        packet_data,
+        packet_timestamp,
+    ) in RawPacketRepository.stream_all_undecrypted():
         packet_info = parse_packet(packet_data)
         path_hex = packet_info.path.hex() if packet_info else None
         path_len = packet_info.path_length if packet_info else None

app/routers/packets.py

@@ -49,8 +49,7 @@ async def _run_historical_channel_decryption(
     channel_key_bytes: bytes, channel_key_hex: str, display_name: str | None = None
 ) -> None:
     """Background task to decrypt historical packets with a channel key."""
-    packets = await RawPacketRepository.get_all_undecrypted()
-    total = len(packets)
+    total = await RawPacketRepository.get_undecrypted_count()
     decrypted_count = 0
 
     if total == 0:
@@ -59,7 +58,11 @@ async def _run_historical_channel_decryption(
 
     logger.info("Starting historical channel decryption of %d packets", total)
 
-    for packet_id, packet_data, packet_timestamp in packets:
+    async for (
+        packet_id,
+        packet_data,
+        packet_timestamp,
+    ) in RawPacketRepository.stream_all_undecrypted():
         result = try_decrypt_packet_with_channel_key(packet_data, channel_key_bytes)
 
         if result is not None: