debug redis queue

Author: tangg555

src/memos/mem_scheduler/task_schedule_modules/dispatcher.py

@@ -14,7 +14,6 @@
 from memos.mem_scheduler.schemas.general_schemas import DEFAULT_STOP_WAIT
 from memos.mem_scheduler.schemas.message_schemas import ScheduleMessageItem
 from memos.mem_scheduler.schemas.task_schemas import RunningTaskItem
-from memos.mem_scheduler.task_schedule_modules.redis_queue import SchedulerRedisQueue
 from memos.mem_scheduler.utils.metrics import MetricsRegistry
 from memos.mem_scheduler.utils.misc_utils import group_messages_by_user_and_mem_cube
 
@@ -152,15 +151,15 @@ def wrapped_handler(messages: list[ScheduleMessageItem]):
 
                 # acknowledge redis messages
 
-                if (
-                    self.use_redis_queue
-                    and self.memos_message_queue is not None
-                    and isinstance(self.memos_message_queue, SchedulerRedisQueue)
-                ):
+                if self.use_redis_queue and self.memos_message_queue is not None:
                     for msg in messages:
                         redis_message_id = msg.redis_message_id
                         # Acknowledge message processing
-                        self.memos_message_queue.ack_message(redis_message_id=redis_message_id)
+                        self.memos_message_queue.ack_message(
+                            user_id=msg.user_id,
+                            mem_cube_id=msg.mem_cube_id,
+                            redis_message_id=redis_message_id,
+                        )
 
                 # Mark task as completed and remove from tracking
                 with self._task_lock:

src/memos/mem_scheduler/task_schedule_modules/redis_queue.py

@@ -151,7 +151,7 @@ def put(
             raise
 
     def ack_message(self, user_id, mem_cube_id, redis_message_id):
-        stream_key = f"{self.stream_key_prefix}:{user_id}:{mem_cube_id}"
+        stream_key = self.get_stream_key(user_id=user_id, mem_cube_id=mem_cube_id)
 
         self.redis.xack(stream_key, self.consumer_group, redis_message_id)
 

src/memos/mem_scheduler/task_schedule_modules/task_queue.py

@@ -85,9 +85,8 @@ def get_messages(self, batch_size: int) -> list[ScheduleMessageItem]:
         # Discover all active streams via queue API
         streams: list[tuple[str, str]] = []
 
-        keys = self.get_stream_keys()
-        for stream_key in keys:
-            # stream_key example: "{prefix}:{user_id}:{mem_cube_id}"
+        stream_keys = self.get_stream_keys()
+        for stream_key in stream_keys:
             try:
                 parts = stream_key.split(":")
                 if len(parts) >= 3: