⚡ Bolt: [performance improvement] optimize sequence number queries for job events

.jules/bolt.md

@@ -0,0 +1,3 @@
+## 2024-10-24 - Efficient Next Sequence Calculation for Job Events
+**Learning:** The codebase previously fetched all `JobEvent` rows into memory (`list_for_job(limit=1000)[-1]`) just to determine the latest `sequence_no` when inserting a new event. This acts like an N+1 inefficiency during bulk event insertions or long-running jobs.
+**Action:** Use an optimized SQL aggregation method `get_max_sequence_no` (e.g. using `select(func.coalesce(func.max(Model.sequence_no), 0))`) in the repository and query for only the scalar max instead of loading list instances.

apps/api/src/cortex_api/services/jobs.py

@@ -110,8 +110,8 @@ async def cancel_job(uow: CortexUnitOfWork, job_id: str) -> JobStatusDetail:
     if updated is None:  # pragma: no cover - defensive
         raise NotFoundError(f"Job `{job_id}` was not found.")
 
-    existing_events = await uow.job_events.list_for_job(job_id, limit=1000)
-    next_sequence = existing_events[-1].sequence_no + 1 if existing_events else 1
+    max_seq = await uow.job_events.get_max_sequence_no(job_id)
+    next_sequence = max_seq + 1
     await uow.job_events.add(
         JobEventRecord(
             job_id=job_id,

packages/db/src/cortex_db/repositories.py

@@ -49,7 +49,7 @@
     SynthesisType,
     TenantRecord,
 )
-from sqlalchemy import desc, or_, select, update
+from sqlalchemy import desc, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from .models import (
@@ -1651,6 +1651,15 @@ async def list_for_job(
         )
         return [_job_event_from_model(model) for model in result.scalars().all()]
 
+    async def get_max_sequence_no(self, job_id: str) -> int:
+        # Optimized: fetching max sequence directly via SQL to prevent loading 1000 rows into memory
+        result = await self._session.execute(
+            select(func.coalesce(func.max(JobEventModel.sequence_no), 0)).where(
+                JobEventModel.job_id == job_id
+            )
+        )
+        return result.scalar_one()
+
 
 class ParseRunRepository:
     def __init__(self, session: AsyncSession) -> None:

packages/evaluation/src/cortex_evaluation/jobs.py

@@ -561,8 +561,8 @@ async def _append_event(
         message: str,
         details: dict[str, Any] | None = None,
     ) -> None:
-        existing_events = await uow.job_events.list_for_job(job.job_id, limit=1000)
-        next_sequence = existing_events[-1].sequence_no + 1 if existing_events else 1
+        max_seq = await uow.job_events.get_max_sequence_no(job.job_id)
+        next_sequence = max_seq + 1
         await uow.job_events.add(
             JobEventRecord(
                 job_id=job.job_id,

packages/knowledge/src/cortex_knowledge/jobs.py

@@ -480,8 +480,8 @@ async def _append_event(
         message: str,
         details: dict[str, Any] | None = None,
     ) -> None:
-        existing_events = await uow.job_events.list_for_job(job.job_id, limit=1000)
-        next_sequence = existing_events[-1].sequence_no + 1 if existing_events else 1
+        max_seq = await uow.job_events.get_max_sequence_no(job.job_id)
+        next_sequence = max_seq + 1
         await uow.job_events.add(
             JobEventRecord(
                 job_id=job.job_id,

packages/parse/src/cortex_parse/jobs.py

@@ -412,8 +412,8 @@ async def _append_event(
         message: str,
         details: dict[str, Any] | None = None,
     ) -> None:
-        existing_events = await uow.job_events.list_for_job(job.job_id, limit=1000)
-        next_sequence = existing_events[-1].sequence_no + 1 if existing_events else 1
+        max_seq = await uow.job_events.get_max_sequence_no(job.job_id)
+        next_sequence = max_seq + 1
         await uow.job_events.add(
             JobEventRecord(
                 job_id=job.job_id,

packages/synthesis/src/cortex_synthesis/jobs.py

@@ -401,8 +401,8 @@ async def _append_event(
         message: str,
         details: dict[str, Any] | None = None,
     ) -> None:
-        existing_events = await uow.job_events.list_for_job(job.job_id, limit=1000)
-        next_sequence = existing_events[-1].sequence_no + 1 if existing_events else 1
+        max_seq = await uow.job_events.get_max_sequence_no(job.job_id)
+        next_sequence = max_seq + 1
         await uow.job_events.add(
             JobEventRecord(
                 job_id=job.job_id,

workers/evaluation-worker/src/cortex_worker_evaluation/artifacts.py

@@ -2,6 +2,8 @@
 
 from cortex_evaluation import (
     EvaluationStorageCaller as WorkerStorageCaller,
+)
+from cortex_evaluation import (
     persist_evaluation_report,
 )