Merge pull request #76 from cloudblue/fix/LITE-23142_transaction-commit-error

LITE-23142 catch database layer errors on transaction commit

Author: maxipavlovic

dj_cqrs/controller/consumer.py

@@ -7,7 +7,7 @@
 from dj_cqrs.constants import SignalType
 from dj_cqrs.registries import ReplicaRegistry
 
-from django.db import close_old_connections, transaction
+from django.db import Error, close_old_connections, transaction
 
 
 logger = logging.getLogger('django-cqrs')
@@ -42,16 +42,26 @@ def route_signal_to_replica_model(signal_type, cqrs_id, instance_data, previous_
         if db_is_needed:
             close_old_connections()
 
-        with transaction.atomic(savepoint=False) if db_is_needed else ExitStack():
-            if signal_type == SignalType.DELETE:
-                return model_cls.cqrs_delete(instance_data)
+        try:
+            with transaction.atomic(savepoint=False) if db_is_needed else ExitStack():
+                if signal_type == SignalType.DELETE:
+                    return model_cls.cqrs_delete(instance_data)
 
-            elif signal_type == SignalType.SAVE:
-                return model_cls.cqrs_save(instance_data, previous_data=previous_data)
+                elif signal_type == SignalType.SAVE:
+                    return model_cls.cqrs_save(instance_data, previous_data=previous_data)
 
-            elif signal_type == SignalType.SYNC:
-                return model_cls.cqrs_save(
-                    instance_data,
-                    previous_data=previous_data,
-                    sync=True,
-                )
+                elif signal_type == SignalType.SYNC:
+                    return model_cls.cqrs_save(
+                        instance_data,
+                        previous_data=previous_data,
+                        sync=True,
+                    )
+        except Error as e:
+            pk_value = instance_data.get(model_cls._meta.pk.name)
+            cqrs_revision = instance_data.get('cqrs_revision')
+
+            logger.error(
+                '{0}\nCQRS {1} error: pk = {2}, cqrs_revision = {3} ({4}).'.format(
+                    str(e), signal_type, pk_value, cqrs_revision, model_cls.CQRS_ID,
+                ),
+            )

dj_cqrs/managers.py

@@ -5,7 +5,7 @@
 from dj_cqrs.constants import FIELDS_TRACKER_FIELD_NAME, TRACKED_FIELDS_ATTR_NAME
 
 from django.core.exceptions import ValidationError
-from django.db import Error, IntegrityError, transaction
+from django.db import Error, transaction
 from django.db.models import F, Manager
 from django.utils import timezone
 
@@ -121,13 +121,6 @@ def create_instance(self, mapped_data, previous_data=None, sync=False):
             )
         except (Error, ValidationError) as e:
             pk_value = mapped_data[self._get_model_pk_name()]
-            if isinstance(e, IntegrityError):
-                logger.warning(
-                    'Potentially wrong CQRS sync order: '
-                    'pk = {0}, cqrs_revision = {1} ({2}).'.format(
-                        pk_value, mapped_data['cqrs_revision'], self.model.CQRS_ID,
-                    ),
-                )
 
             logger.error(
                 '{0}\nCQRS create error: pk = {1} ({2}).'.format(

tests/dj_replica/models.py

@@ -121,6 +121,24 @@ class Book(models.Model):
     author = models.ForeignKey(AuthorRef, on_delete=models.CASCADE)
 
 
+class Article(ReplicaMixin, models.Model):
+    CQRS_ID = 'article'
+
+    id = models.IntegerField(primary_key=True)
+
+    author = models.ForeignKey(AuthorRef, on_delete=models.CASCADE)
+
+    @classmethod
+    def cqrs_create(cls, sync, mapped_data, previous_data=None):
+        data = {
+            'id': mapped_data['id'],
+            'author_id': mapped_data['author']['id'],
+            'cqrs_revision': mapped_data['cqrs_revision'],
+            'cqrs_updated': mapped_data['cqrs_updated'],
+        }
+        return super().cqrs_create(sync, data, previous_data)
+
+
 class FailModel(ReplicaMixin, models.Model):
     CQRS_ID = 'fail'
 

tests/test_controller.py

@@ -6,6 +6,7 @@
 from dj_cqrs.dataclasses import TransportPayload
 
 from django.conf import settings
+from django.utils.timezone import now
 
 import pytest
 
@@ -66,6 +67,30 @@ def test_route_signal_to_replica_model_with_db(django_assert_num_queries):
         route_signal_to_replica_model(SignalType.SAVE, 'lock', {})
 
 
+@pytest.mark.django_db(transaction=True)
+def test_route_signal_to_replica_model_integrity_error(caplog):
+    instance_data = {
+        'id': 10,
+        'author': {
+            'id': 100,
+        },
+        'cqrs_revision': 0,
+        'cqrs_updated': now(),
+    }
+    instance = route_signal_to_replica_model(SignalType.SAVE, 'article', instance_data)
+    assert not instance
+
+    errors = {
+        'sqlite': 'FOREIGN KEY constraint failed',
+        'postgres': (
+            'insert or update on table "dj_replica_article" violates foreign key constraint'
+        ),
+        'mysql': 'Cannot add or update a child row: a foreign key constraint',
+    }
+
+    assert errors[settings.DB_ENGINE] in caplog.text
+
+
 def test_route_signal_to_replica_model_without_db():
     with pytest.raises(NotImplementedError):
         route_signal_to_replica_model(SignalType.SAVE, 'no_db', {})

tests/test_replica/test_mixin.py

@@ -2,6 +2,7 @@
 
 from dj_cqrs.metas import ReplicaMeta
 
+from django.conf import settings
 from django.db.models import CharField, IntegerField, QuerySet
 from django.utils.timezone import now
 
@@ -331,8 +332,24 @@ def test_update_before_create_is_over(caplog):
     assert updated_instance.cqrs_revision == 1
     assert updated_instance.char_field == 'new_text'
     assert not created_instance
-    assert 'Potentially wrong CQRS sync order: pk = 1, cqrs_revision = 0 (basic).' in caplog.text
-    assert 'CQRS create error: pk = 1 (basic).' in caplog.text
+
+    errors = {
+        'sqlite': (
+            'UNIQUE constraint failed: dj_replica_basicfieldsmodelref.int_field\n'
+            'CQRS create error: pk = 1 (basic).\n'
+        ),
+        'postgres': (
+            'duplicate key value violates unique constraint "dj_replica_basicfieldsmodelref_pkey"\n'
+            'DETAIL:  Key (int_field)=(1) already exists.\n\n'
+            'CQRS create error: pk = 1 (basic).\n'
+        ),
+        'mysql': (
+            '(1062, "Duplicate entry \'1\' for key \'dj_replica_basicfieldsmodelref.PRIMARY\'")\n'
+            'CQRS create error: pk = 1 (basic).\n'
+        ),
+    }
+
+    assert errors[settings.DB_ENGINE] in caplog.text
 
 
 @pytest.mark.django_db(transaction=True)