Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/notifications/notifications.py (Python)

Code before:
            logger.error(f"Error processing notification: {e}")
            return False

    def run_service(self):
        logger.info(f"[{self.service_name}] Started notification service")

        # Set up queue consumers
        channel = self.run_and_wait(self.rabbit.get_channel())

        immediate_queue = self.run_and_wait(
            channel.get_queue("immediate_notifications")
        )

        while self.running:
            try:
                # Process immediate notifications
                try:
                    message = self.run_and_wait(immediate_queue.get())

                    if message:
                        success = self.run_and_wait(
                            self._process_immediate(message.body.decode())
                        )
                        if success:
                            self.run_and_wait(message.ack())
                        else:
                            self.run_and_wait(message.reject(requeue=True))
                except QueueEmpty:
                    logger.debug("Immediate queue empty")

                time.sleep(0.1)

            except QueueEmpty as e:
                logger.debug(f"Queue empty: {e}")
            except Exception as e:
                logger.error(f"Error in notification service loop: {e}")

    def cleanup(self):
        """Cleanup service resources"""
        self.running = False
        super().cleanup()

    # ------- UTILITIES ------- #


@thread_cached
def get_db_client() -> "DatabaseManager":
    from backend.executor import DatabaseManager

    return get_service_client(DatabaseManager)


Review comment (bug):
Could this not cause an infinite loop, of requeueing failed messages, which are then reprocessed, which fails and requeue's them again. 

A mechanism to avoid this inf loop needs to be added.

Code after:
            return False

    def run_service(self):
        logger.info(f"[{self.service_name}] Started notification service")

        # Set up queue consumers
        channel = self.run_and_wait(self.rabbit.get_channel())

        immediate_queue = self.run_and_wait(
            channel.get_queue("immediate_notifications")
        )

        while self.running:
            try:
                # Process immediate notifications
                try:
                    message = self.run_and_wait(immediate_queue.get())

                    if message:
                        success = self.run_and_wait(
                            self._process_immediate(message.body.decode())
                        )
                        if success:
                            self.run_and_wait(message.ack())
                        else:
                            self.run_and_wait(message.reject(requeue=True))
                except QueueEmpty:
                    logger.debug("Immediate queue empty")

                time.sleep(0.1)

            except QueueEmpty as e:
                logger.debug(f"Queue empty: {e}")
            except Exception as e:
                logger.error(f"Error in notification service loop: {e}")

    def cleanup(self):
        """Cleanup service resources"""
        self.running = False
        super().cleanup()

    # ------- UTILITIES ------- #


@thread_cached
def get_db_client() -> "DatabaseManager":
    from backend.executor import DatabaseManager

    return get_service_client(DatabaseManager)