fix: flush activities and checkpoint heads in batches during onboarding (CM-1318)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/mailing_list_integration/src/crowdmail/settings.py

@@ -41,3 +41,8 @@ def load_env_var(key: str, required=True, default=None):
 STUCK_RECURRENT_LIST_TIMEOUT_HOURS = int(
     load_env_var("STUCK_RECURRENT_LIST_TIMEOUT_HOURS", default="4")
 )
+
+# Flush accumulated activities (and checkpoint processed heads) every this many
+# messages instead of buffering an entire list's history in memory before one
+# flush at the end — large lore archives can have 100k+ messages.
+ACTIVITY_FLUSH_BATCH_SIZE = int(load_env_var("ACTIVITY_FLUSH_BATCH_SIZE", default="500"))

services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py

@@ -21,6 +21,7 @@
 from crowdmail.services.parse.noteren import parse_email
 from crowdmail.services.queue import queue_service
 from crowdmail.settings import (
+    ACTIVITY_FLUSH_BATCH_SIZE,
     DEFAULT_TENANT_ID,
     WORKER_ERROR_BACKOFF_SEC,
     WORKER_POLLING_INTERVAL_SEC,
@@ -90,6 +91,7 @@ async def _process_single_list(self, mailing_list: MailingList):
                 shard = shard_index(shard_path)
                 commit_ids = await new_commits(shard_path, heads.get(shard))
                 for git_id in commit_ids:
+                    heads[shard] = git_id
                     try:
                         message, blob_id = read_email(shard_path, git_id)
                         parsed = parse_email(
@@ -134,8 +136,13 @@ async def _process_single_list(self, mailing_list: MailingList):
                             mailing_list.segment_id, mailing_list.integration_id, result_id
                         )
                     )
-                if commit_ids:
-                    heads[shard] = commit_ids[-1]
+
+                    if len(activities_db) >= ACTIVITY_FLUSH_BATCH_SIZE:
+                        await batch_insert_activities(activities_db)
+                        await queue_service.send_batch_activities(activities_kafka)
+                        await update_processed_heads(mailing_list.id, heads)
+                        activities_db = []
+                        activities_kafka = []
 
             if activities_db:
                 await batch_insert_activities(activities_db)