Merge pull request #7 from maxipavlovic/feature/LITE-16458

LITE-16458 Now CQRS won't send duplicate messages for the same instance in transaction

Author: d3rky

README.md

@@ -109,24 +109,10 @@ CQRS = {
 Notes
 -----
 
-When there are master models with related entities in CQRS_SERIALIZER, it's important to have operations within atomic transactions.
-CQRS sync will happen on transaction commit. Please, avoid saving master model within transaction more then once to reduce syncing and potential racing on replica side.
-Updating of related model won't trigger CQRS automatic synchronization for master model. This needs to be done manually.
-
-Example:
-```python
-with transaction.atomic():
-    publisher = models.Publisher.objects.create(id=1, name='publisher')
-    author = models.Author.objects.create(id=1, name='author', publisher=publisher)
-
-with transaction.atomic():
-    publisher.name = 'new'
-    publisher.save()
-
-    author.save()
-```
-
-When only needed instances need to be synchronized, there is a method `is_sync_instance` to set filtering rule. 
+* When there are master models with related entities in CQRS_SERIALIZER, it's important to have operations within atomic transactions. CQRS sync will happen on transaction commit. 
+* Please, avoid saving different instances of the same entity within transaction to reduce syncing and potential racing on replica side.
+* Updating of related model won't trigger CQRS automatic synchronization for master model. This needs to be done manually.
+* When only needed instances need to be synchronized, there is a method `is_sync_instance` to set filtering rule. 
 It's important to understand, that CQRS counting works even without syncing and rule is applied every time model is updated.
 
 Example:

dj_cqrs/mixins.py

@@ -1,5 +1,6 @@
 #  Copyright © 2020 Ingram Micro Inc. All rights reserved.
 
+from django.db import router, transaction
 from django.db.models import DateField, DateTimeField, F, IntegerField, Manager, Model
 from django.db.models.expressions import CombinedExpression
 from django.utils.module_loading import import_string
@@ -64,10 +65,39 @@ class directly.**"""
     class Meta:
         abstract = True
 
+    @property
+    def cqrs_saves_count(self):
+        """Shows how many times this instance has been saved within the transaction."""
+        return getattr(self, '_cqrs_saves_count', 0)
+
+    @property
+    def is_initial_cqrs_save(self):
+        """This flag is used to check if instance has already been registered for CQRS update."""
+        return self.cqrs_saves_count < 2
+
+    def reset_cqrs_saves_count(self):
+        """This method is used to automatically reset instance CQRS counters on transaction commit.
+        But this can also be used to control custom behaviour within transaction
+        or in case of rollback,
+        when several sequential transactions are used to change the same instance.
+        """
+        if hasattr(self, '_cqrs_saves_count'):
+            self._cqrs_saves_count = 0
+
     def save(self, *args, **kwargs):
-        if not self._state.adding:
+        using = kwargs.get('using') or router.db_for_write(self.__class__, instance=self)
+        connection = transaction.get_connection(using)
+        if connection.in_atomic_block:
+            _cqrs_saves_count = self.cqrs_saves_count
+            self._cqrs_saves_count = _cqrs_saves_count + 1
+        else:
+            self.reset_cqrs_saves_count()
+
+        if self.is_initial_cqrs_save and (not self._state.adding):
             self.cqrs_revision = F('cqrs_revision') + 1
+
         self._save_tracked_fields()
+
         return super(RawMasterMixin, self).save(*args, **kwargs)
 
     def _save_tracked_fields(self):

dj_cqrs/signals.py

@@ -56,7 +56,9 @@ def post_save(cls, sender, **kwargs):
         sync = kwargs.get('sync', False)
         queue = kwargs.get('queue', None)
 
-        if not transaction.get_connection(using).in_atomic_block:
+        connection = transaction.get_connection(using)
+        if not connection.in_atomic_block:
+            instance.reset_cqrs_saves_count()
             instance_data = instance.to_cqrs_dict(using)
             previous_data = instance.get_tracked_fields_data()
             signal_type = SignalType.SYNC if sync else SignalType.SAVE
@@ -66,7 +68,7 @@ def post_save(cls, sender, **kwargs):
             )
             producer.produce(payload)
 
-        else:
+        elif instance.is_initial_cqrs_save:
             transaction.on_commit(
                 lambda: MasterSignals.post_save(
                     sender, instance=instance, using=using, sync=sync, queue=queue,

tests/dj_master/models.py

@@ -135,7 +135,7 @@ class TrackedFieldsChildModel(MasterMixin):
     CQRS_TRACKED_FIELDS = ('char_field', 'parent')
 
     char_field = models.CharField(max_length=10)
-    parent = models.ForeignKey(TrackedFieldsParentModel, on_delete=models.CASCADE)
+    parent = models.ForeignKey(TrackedFieldsParentModel, on_delete=models.CASCADE, null=True)
 
 
 class TrackedFieldsAllWithChildModel(MasterMixin):

tests/test_master/test_mixin.py

@@ -10,6 +10,7 @@
 
 from dj_cqrs.constants import SignalType, FIELDS_TRACKER_FIELD_NAME
 from dj_cqrs.metas import MasterMeta
+
 from tests.dj_master import models
 from tests.dj_master.serializers import AuthorSerializer
 from tests.utils import (
@@ -268,7 +269,7 @@ def test_create():
         assert m.cqrs_updated is not None
 
 
-@pytest.mark.django_db
+@pytest.mark.django_db(transaction=True)
 def test_update():
     m = models.AutoFieldsModel.objects.create()
     cqrs_updated = m.cqrs_updated
@@ -668,3 +669,126 @@ def test_m2m_not_supported():
 
     assert 'm2m_field' not in cqrs_data
     assert 'char_field' in cqrs_data
+
+
+@pytest.mark.django_db(transaction=True)
+def test_transaction_instance_saved_once_simple_case(mocker):
+    publisher_mock = mocker.patch('dj_cqrs.controller.producer.produce')
+
+    i0 = models.TrackedFieldsChildModel.objects.create(char_field='old')
+    with transaction.atomic():
+        i1 = models.TrackedFieldsParentModel.objects.create(char_field='1')
+        i1.char_field = '2'
+        i1.save()
+
+        i2 = models.TrackedFieldsParentModel(char_field='a')
+        i2.save()
+
+        i3 = models.TrackedFieldsChildModel.objects.create(char_field='.')
+
+        i0.char_field = 'new'
+        i0.save()
+
+    assert publisher_mock.call_count == 5
+
+    for i in [i0, i1, i2, i3]:
+        i.refresh_from_db()
+    assert i0.cqrs_revision == 1
+    assert i1.cqrs_revision == 0
+    assert i2.cqrs_revision == 0
+    assert i3.cqrs_revision == 0
+
+    mapper = (
+        (i0.pk, 0, 'old', None),
+        (i1.pk, 0, '2', '1'),
+        (i2.pk, 0, 'a', None),
+        (i3.pk, 0, '.', None),
+        (i0.pk, 1, 'new', 'old'),
+    )
+    for index, call in enumerate(publisher_mock.call_args_list):
+        payload = call[0][0]
+        expected_data = mapper[index]
+
+        assert payload.pk == expected_data[0]
+        assert payload.instance_data['cqrs_revision'] == expected_data[1]
+        assert payload.instance_data['char_field'] == expected_data[2]
+        assert payload.previous_data['char_field'] == expected_data[3]
+
+
+@pytest.mark.django_db(transaction=True)
+def test_cqrs_saves_count_lifecycle():
+    instance = models.TrackedFieldsParentModel(char_field='1')
+    instance.reset_cqrs_saves_count()
+    assert instance.cqrs_saves_count == 0
+    assert instance.is_initial_cqrs_save
+
+    instance.save()
+    assert instance.cqrs_saves_count == 0
+    assert instance.is_initial_cqrs_save
+
+    instance.save()
+    assert instance.cqrs_saves_count == 0
+    assert instance.is_initial_cqrs_save
+
+    instance.refresh_from_db()
+    assert instance.cqrs_saves_count == 0
+    assert instance.is_initial_cqrs_save
+
+    with transaction.atomic():
+        instance.save()
+        assert instance.cqrs_saves_count == 1
+        assert instance.is_initial_cqrs_save
+
+        instance.save()
+        assert instance.cqrs_saves_count == 2
+        assert not instance.is_initial_cqrs_save
+
+        instance.refresh_from_db()
+        assert instance.cqrs_saves_count == 2
+        assert not instance.is_initial_cqrs_save
+
+        same_db_object_other_instance = models.TrackedFieldsParentModel.objects.first()
+        assert same_db_object_other_instance.pk == instance.pk
+        assert same_db_object_other_instance.cqrs_saves_count == 0
+        assert same_db_object_other_instance.is_initial_cqrs_save
+
+        same_db_object_other_instance.save()
+        assert same_db_object_other_instance.cqrs_saves_count == 1
+        assert same_db_object_other_instance.is_initial_cqrs_save
+
+        same_db_object_other_instance.reset_cqrs_saves_count()
+        assert same_db_object_other_instance.cqrs_saves_count == 0
+        assert same_db_object_other_instance.is_initial_cqrs_save
+
+        same_db_object_other_instance.save()
+        assert same_db_object_other_instance.cqrs_saves_count == 1
+        assert same_db_object_other_instance.is_initial_cqrs_save
+
+    assert instance.cqrs_saves_count == 0
+    assert same_db_object_other_instance.cqrs_saves_count == 0
+
+
+@pytest.mark.django_db(transaction=True)
+def test_sequential_transactions(mocker):
+    publisher_mock = mocker.patch('dj_cqrs.controller.producer.produce')
+
+    with transaction.atomic():
+        instance = models.TrackedFieldsParentModel.objects.create(char_field='1')
+
+    with transaction.atomic():
+        instance.char_field = '3'
+        instance.save()
+
+        transaction.set_rollback(True)
+        instance.reset_cqrs_saves_count()
+
+    with transaction.atomic():
+        instance.char_field = '2'
+        instance.save()
+
+    instance.refresh_from_db()
+
+    assert publisher_mock.call_count == 2
+    assert instance.cqrs_revision == 1
+    assert publisher_mock.call_args_list[0][0][0].instance_data['char_field'] == '1'
+    assert publisher_mock.call_args_list[1][0][0].instance_data['char_field'] == '2'