feat(core): add max overflow for buffer

Author: gusye1234

src/server/core/acontext_core/service/controller/message.py

@@ -15,10 +15,19 @@ async def process_session_pending_message(
     pending_message_ids = None
     try:
         async with DB_CLIENT.get_session_context() as session:
-            r = await MD.unpending_session_messages_to_running(session, session_id)
+            r = await MD.get_message_ids(
+                session,
+                session_id,
+                limit=project_config.project_session_message_buffer_max_overflow_turns,
+                asc=True,
+            )
             pending_message_ids, eil = r.unpack()
             if eil:
                 return
+            await MD.update_message_status_to(
+                session, pending_message_ids, TaskStatus.RUNNING
+            )
+        LOG.info(f"Unpending {len(pending_message_ids)} session messages to process")
 
         async with DB_CLIENT.get_session_context() as session:
             r = await MD.fetch_messages_data_by_ids(session, pending_message_ids)

src/server/core/acontext_core/service/data/message.py

@@ -181,7 +181,7 @@ async def get_message_ids(
 
 
 async def unpending_session_messages_to_running(
-    db_session: AsyncSession, session_id: asUUID
+    db_session: AsyncSession, session_id: asUUID, limit: int
 ) -> Result[List[asUUID]]:
     query = (
         update(Message)

src/server/core/acontext_core/service/session_message.py

@@ -98,7 +98,7 @@ async def insert_new_message(body: InsertNewMessage, message: Message):
     _l = await check_session_message_lock_or_set(str(body.session_id))
     if not _l:
         LOG.info(
-            f"Current Session is processing. "
+            f"Current Session is locked. "
             f"wait {DEFAULT_CORE_CONFIG.session_message_session_lock_wait_seconds} seconds for next resend. "
             f"Message {body.message_id}"
         )
@@ -113,6 +113,20 @@ async def insert_new_message(body: InsertNewMessage, message: Message):
         LOG.info(
             f"Session message buffer is full (size: {pending_message_length}), start process"
         )
+        if (
+            pending_message_length
+            > project_config.project_session_message_buffer_max_overflow_turns
+        ):
+            LOG.info(
+                f"Session message buffer is overflow "
+                f"(size: {pending_message_length} > {project_config.project_session_message_buffer_max_overflow_turns}), "
+                f"Truncate the buffer, the rest will be processed later in {DEFAULT_CORE_CONFIG.session_message_session_lock_wait_seconds} seconds"
+            )
+            await MQ_CLIENT.publish(
+                exchange_name=EX.session_message,
+                routing_key=RK.session_message_insert_retry,
+                body=body.model_dump_json(),
+            )
         await MC.process_session_pending_message(project_config, body.session_id)
     finally:
         await release_session_message_lock(str(body.session_id))
@@ -162,7 +176,7 @@ async def buffer_new_message(body: InsertNewMessage, message: Message):
     _l = await check_session_message_lock_or_set(str(body.session_id))
     if not _l:
         LOG.info(
-            f"Current Session is processing, resend Message {body.message_id} to insert queue."
+            f"Current Session is locked, resend Message {body.message_id} to insert queue."
         )
         await MQ_CLIENT.publish(
             exchange_name=EX.session_message,

src/server/core/acontext_core/telemetry/log.py

@@ -63,7 +63,7 @@ def format(self, record):
         base_msg = super().format(record)
 
         # Check if there are extra fields (beyond the standard ones)
-        printout_attrs = {"traceback"}
+        printout_attrs = {"traceback", "session_id", "message_id"}
 
         all_fields = record.__dict__
         extra_fields = []