refactor: revise _submit_web_logs to address log missing issue

Author: tangg555

src/memos/mem_scheduler/base_scheduler.py

@@ -838,27 +838,26 @@ def _submit_web_logs(
         Args:
             messages: Single log message or list of log messages
         """
-        messages_list = [messages] if isinstance(messages, ScheduleLogForWebItem) else messages
-        for message in messages_list:
+        if isinstance(messages, ScheduleLogForWebItem):
+            messages = [messages]  # transform single message to list
+
+        for message in messages:
             logger.info(
                 f"[DIAGNOSTIC] base_scheduler._submit_web_logs called. Message to publish: {message.model_dump_json(indent=2)}"
             )
+
         if self.rabbitmq_config is None:
             logger.info(
                 "[DIAGNOSTIC] base_scheduler._submit_web_logs: RabbitMQ config not loaded; skipping publish."
             )
             return
 
-        if isinstance(messages, ScheduleLogForWebItem):
-            messages = [messages]  # transform single message to list
-
         for message in messages:
-            if not isinstance(message, ScheduleLogForWebItem):
-                error_msg = f"Invalid message type: {type(message)}, expected ScheduleLogForWebItem"
-                logger.error(error_msg)
-                raise TypeError(error_msg)
+            try:
+                self._web_log_message_queue.put(message)
+            except Exception as e:
+                logger.warning(f"Failed to put message to web log queue: {e}", stack_info=True)
 
-            self._web_log_message_queue.put(message)
             message_info = message.debug_info()
             logger.debug(f"Submitted Scheduling log for web: {message_info}")
 

src/memos/mem_scheduler/general_modules/misc.py

@@ -217,19 +217,20 @@ def put(self, item: T, block: bool = False, timeout: float | None = None) -> Non
             block: Ignored (kept for compatibility with Queue interface)
             timeout: Ignored (kept for compatibility with Queue interface)
         """
-        try:
-            # First try non-blocking put
-            super().put(item, block=block, timeout=timeout)
-        except Full:
-            # Remove the oldest item and mark it done to avoid leaking unfinished_tasks
-            with suppress(Empty):
-                _ = self.get_nowait()
-                # If the removed item had previously incremented unfinished_tasks,
-                # we must decrement here since it will never be processed.
-                with suppress(ValueError):
-                    self.task_done()
-            # Retry putting the new item
-            super().put(item, block=block, timeout=timeout)
+        while True:
+            try:
+                # First try non-blocking put
+                super().put(item, block=block, timeout=timeout)
+                return
+            except Full:
+                # Remove the oldest item and mark it done to avoid leaking unfinished_tasks
+                with suppress(Empty):
+                    _ = self.get_nowait()
+                    # If the removed item had previously incremented unfinished_tasks,
+                    # we must decrement here since it will never be processed.
+                    with suppress(ValueError):
+                        self.task_done()
+                # Continue loop to retry putting the item
 
     def get(
         self, block: bool = True, timeout: float | None = None, batch_size: int | None = None