Merge pull request #29 from cloudblue/feature/lite-15627-message-retrying

LITE-15627 Message retrying

Author: maxipavlovic

dj_cqrs/constants.py

@@ -1,4 +1,4 @@
-#  Copyright © 2020 Ingram Micro Inc. All rights reserved.
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
 ALL_BASIC_FIELDS = '__all__'
 
@@ -20,3 +20,9 @@ class SignalType:
 
 
 NO_QUEUE = 'None'
+
+DEFAULT_DEAD_MESSAGE_TTL = 864000  # 10 days
+DEFAULT_DELAY_QUEUE_MAX_SIZE = None  # Infinite
+DEFAULT_CQRS_MESSAGE_TTL = 86400  # 1 day
+DEFAULT_CQRS_MAX_RETRIES = 30
+DEFAULT_CQRS_RETRY_DELAY = 2  # seconds

dj_cqrs/dataclasses.py

@@ -1,6 +1,10 @@
 #  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
+from dateutil.parser import parse as dateutil_parse
+from django.utils import timezone
+
 from dj_cqrs.correlation import get_correlation_id
+from dj_cqrs.utils import get_expires_datetime
 
 
 class TransportPayload:
@@ -21,17 +25,23 @@ class TransportPayload:
     :type previous_data: dict, optional
     :param correlation_id: Correlation ID of process, where this payload is used.
     :type correlation_id: str, optional
+    :param retries: Current number of message retries.
+    :type retries: int, optional
+    :param expires: Message expiration datetime, infinite if None
+    :type expires: datetime, optional
     """
 
     def __init__(
-            self,
-            signal_type,
-            cqrs_id,
-            instance_data,
-            instance_pk,
-            queue=None,
-            previous_data=None,
-            correlation_id=None,
+        self,
+        signal_type,
+        cqrs_id,
+        instance_data,
+        instance_pk,
+        queue=None,
+        previous_data=None,
+        correlation_id=None,
+        expires=None,
+        retries=0,
     ):
         self.__signal_type = signal_type
         self.__cqrs_id = cqrs_id
@@ -45,6 +55,37 @@ def __init__(
         else:
             self.__correlation_id = get_correlation_id(signal_type, cqrs_id, instance_pk, queue)
 
+        self.__expires = expires
+        self.__retries = retries
+
+    @classmethod
+    def from_message(cls, dct):
+        """Builds payload from message data.
+
+        :param dct: Deserialized message body data.
+        :type dct: dict
+        :return: TransportPayload instance.
+        :rtype: TransportPayload
+        """
+        if 'expires' in dct:
+            expires = dct['expires']
+            if dct['expires'] is not None:
+                expires = dateutil_parse(dct['expires'])
+        else:
+            # Backward compatibility for old messages otherwise they are infinite by default.
+            expires = get_expires_datetime()
+
+        return cls(
+            dct['signal_type'],
+            dct['cqrs_id'],
+            dct['instance_data'],
+            dct.get('instance_pk'),
+            previous_data=dct.get('previous_data'),
+            correlation_id=dct.get('correlation_id'),
+            expires=expires,
+            retries=dct.get('retries') or 0,
+        )
+
     @property
     def signal_type(self):
         return self.__signal_type
@@ -73,18 +114,47 @@ def previous_data(self):
     def correlation_id(self):
         return self.__correlation_id
 
+    @property
+    def expires(self):
+        return self.__expires
+
+    @property
+    def retries(self):
+        return self.__retries
+
+    @retries.setter
+    def retries(self, value):
+        assert value >= 0, "Payload retries field should be 0 or positive integer."
+        self.__retries = value
+
     def to_dict(self):
-        """
-        Return the payload as a dictionary.
+        """Return the payload as a dictionary.
 
         :return: This payload.
         :rtype: dict
         """
+        expires = self.__expires
+        if expires:
+            expires = expires.replace(microsecond=0).isoformat()
+
         return {
             'signal_type': self.__signal_type,
             'cqrs_id': self.__cqrs_id,
             'instance_data': self.__instance_data,
             'previous_data': self.__previous_data,
             'instance_pk': self.__instance_pk,
             'correlation_id': self.__correlation_id,
+            'retries': self.__retries,
+            'expires': expires,
         }
+
+    def is_expired(self):
+        """Checks if this payload is expired.
+
+        :return: True if payload is expired, False otherwise.
+        :rtype: bool
+        """
+        return (
+            self.__expires is not None
+            and self.__expires <= timezone.now()
+        )

dj_cqrs/delay.py

@@ -0,0 +1,81 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+from queue import PriorityQueue, Full
+
+from django.utils import timezone
+
+
+class DelayMessage:
+    """Delay message.
+
+    :param delivery_tag: The server-assigned and channel-specific delivery tag.
+    :type delivery_tag: int
+    :param payload: Transport payload.
+    :type payload: dj_cqrs.dataclasses.TransportPayload
+    :param eta: Time after which the message should be requeued.
+    :type eta: datetime.datetime
+    """
+
+    def __init__(self, delivery_tag, payload, eta):
+        self.delivery_tag = delivery_tag
+        self.payload = payload
+        self.eta = eta
+
+
+class DelayQueue:
+    """Queue for delay messages."""
+
+    def __init__(self, max_size=None):
+        if max_size is not None:
+            assert max_size > 0, "Delay queue max_size should be positive integer."
+
+        self._max_size = max_size
+        self._queue = PriorityQueue()
+
+    def get(self):
+        """
+        :rtype: DelayMessage
+        """
+        *_, delay_message = self._queue.get()
+        return delay_message
+
+    def get_ready(self):
+        """Returns messages with expired ETA.
+
+        :return: delayed messages generator
+        :rtype: typing.Generator[DelayMessage]
+        """
+        while self.qsize():
+            delay_message = self.get()
+            if delay_message.eta > timezone.now():
+                # Queue is ordered by message ETA.
+                # Remaining messages should wait longer, we don't check them.
+                self.put(delay_message)
+                break
+
+            yield delay_message
+
+    def put(self, delay_message):
+        """Adds message to queue.
+
+        :param delay_message: DelayMessage instance.
+        :type delay_message: DelayMessage
+        """
+        assert isinstance(delay_message, DelayMessage)
+        if self.full():
+            raise Full("Delay queue is full")
+
+        self._queue.put((
+            delay_message.eta.timestamp(),
+            delay_message.delivery_tag,
+            delay_message,
+        ))
+
+    def qsize(self):
+        return self._queue.qsize()
+
+    def full(self):
+        return (
+            self._max_size is not None
+            and self.qsize() >= self._max_size
+        )

dj_cqrs/mixins.py

@@ -1,5 +1,7 @@
 #  Copyright © 2021 Ingram Micro Inc. All rights reserved.
 
+import logging
+
 from django.conf import settings
 from django.db import router, transaction
 from django.db.models import DateField, DateTimeField, F, IntegerField, Manager, Model
@@ -8,6 +10,8 @@
 
 from dj_cqrs.constants import (
     ALL_BASIC_FIELDS,
+    DEFAULT_CQRS_MAX_RETRIES,
+    DEFAULT_CQRS_RETRY_DELAY,
     FIELDS_TRACKER_FIELD_NAME,
     TRACKED_FIELDS_ATTR_NAME,
 )
@@ -16,6 +20,9 @@
 from dj_cqrs.signals import MasterSignals, post_bulk_create, post_update
 
 
+logger = logging.getLogger('django-cqrs')
+
+
 class RawMasterMixin(Model):
 
     """Base class for MasterMixin. **Users shouldn't use this
@@ -404,3 +411,55 @@ def cqrs_delete(cls, master_data):
             raise NotImplementedError
 
         return cls.cqrs.delete_instance(master_data)
+
+    @staticmethod
+    def should_retry_cqrs(current_retry, exception=None):
+        """Checks if we should retry the message after current attempt.
+
+        :param current_retry: Current number of message retries.
+        :type current_retry: int
+        :param exception: Exception instance raised during message consume.
+        :type exception: Exception, optional
+        :return: True if message should be retried, False otherwise.
+        :rtype: bool
+        """
+        replica_settings = settings.CQRS.get('replica', {})
+        if 'CQRS_MAX_RETRIES' in replica_settings and replica_settings['CQRS_MAX_RETRIES'] is None:
+            # Infinite
+            return True
+
+        min_value = 0
+        max_retries = replica_settings.get('CQRS_MAX_RETRIES', DEFAULT_CQRS_MAX_RETRIES)
+        if not isinstance(max_retries, int) or max_retries < min_value:
+            logger.warning(
+                "Replica setting CQRS_MAX_RETRIES=%s is invalid, using default %s",
+                max_retries, DEFAULT_CQRS_MAX_RETRIES,
+            )
+            max_retries = DEFAULT_CQRS_MAX_RETRIES
+
+        return current_retry < max_retries
+
+    @staticmethod
+    def get_cqrs_retry_delay(current_retry):
+        """Returns number of seconds to wait before requeuing the message.
+
+        :param current_retry: Current number of message retries.
+        :type current_retry: int
+        :return: Delay in seconds.
+        :rtype: int
+        """
+        retry_delay = (
+            settings.CQRS
+            .get('replica', {})
+            .get('CQRS_RETRY_DELAY', DEFAULT_CQRS_RETRY_DELAY)
+        )
+
+        min_value = 1
+        if not isinstance(retry_delay, int) or retry_delay < min_value:
+            logger.warning(
+                "Replica setting CQRS_RETRY_DELAY=%s is invalid, using default %s",
+                retry_delay, DEFAULT_CQRS_RETRY_DELAY,
+            )
+            retry_delay = DEFAULT_CQRS_RETRY_DELAY
+
+        return retry_delay

dj_cqrs/signals.py

@@ -7,7 +7,7 @@
 from dj_cqrs.controller import producer
 from dj_cqrs.constants import SignalType
 from dj_cqrs.dataclasses import TransportPayload
-
+from dj_cqrs.utils import get_expires_datetime
 
 post_bulk_create = Signal(providing_args=['instances', 'using'])
 """
@@ -73,6 +73,7 @@ def post_save(cls, sender, **kwargs):
                 instance.pk,
                 queue,
                 previous_data,
+                expires=get_expires_datetime(),
             )
             producer.produce(payload)
 
@@ -107,7 +108,13 @@ def post_delete(cls, sender, **kwargs):
 
         signal_type = SignalType.DELETE
 
-        payload = TransportPayload(signal_type, sender.CQRS_ID, instance_data, instance.pk)
+        payload = TransportPayload(
+            signal_type,
+            sender.CQRS_ID,
+            instance_data,
+            instance.pk,
+            expires=get_expires_datetime(),
+        )
         # Delete is always in transaction!
         transaction.on_commit(lambda: producer.produce(payload))
 

dj_cqrs/transport/kombu.py

@@ -111,18 +111,19 @@ def produce(cls, payload):
     def _consume_message(cls, body, message):
         try:
             dct = ujson.loads(body)
-            for key in ('signal_type', 'cqrs_id', 'instance_data'):
-                if key not in dct:
-                    raise ValueError
-
-            if 'instance_pk' not in dct:
-                logger.warning('CQRS deprecated package structure.')
-
         except ValueError:
             logger.error("CQRS couldn't be parsed: {}.".format(body))
             message.reject()
             return
 
+        required_keys = {'instance_pk', 'signal_type', 'cqrs_id', 'instance_data'}
+        for key in required_keys:
+            if key not in dct:
+                msg = "CQRS couldn't proceed, %s isn't found in body: %s."
+                logger.error(msg, key, body)
+                message.reject()
+                return
+
         payload = TransportPayload(
             dct['signal_type'],
             dct['cqrs_id'],