Fix silent deserialization failure with unique together constraint

morango/sync/operations.py

@@ -13,6 +13,7 @@
 from django.db.models import CharField
 from django.db.models import Q
 from django.db.models import signals
+from django.db.utils import IntegrityError
 from django.db.utils import OperationalError
 from django.utils import timezone
 from rest_framework.exceptions import ValidationError
@@ -431,14 +432,55 @@ def _validate_store_foreign_keys(from_model_name, fk_references):
     return exclude_pks, deleted_pks
 
 
+def _save_deserialized_record(store_model, app_model, model_name, excluded_list):
+    """
+    Attempt to save one deserialized app model into the app table.
+
+    On success: clears dirty_bit and deserialization_error on the Store record.
+    On failure: records the error on the Store record, keeps dirty_bit True,
+    and logs a warning.
+
+    :returns: True if save succeeded, False otherwise
+    """
+
+    try:
+        with transaction.atomic():
+            with mute_signals(signals.pre_save, signals.post_save):
+                app_model.save(update_dirty_bit_to=False)
+        store_model.dirty_bit = False
+        store_model.deserialization_error = ""
+        store_model.save(
+            update_fields=["dirty_bit", "deserialization_error"]
+        )
+        return True
+    except (
+        exceptions.ValidationError,
+        exceptions.ObjectDoesNotExist,
+        ValueError,
+        IntegrityError,
+    ) as e:
+        excluded_list.append(store_model.id)
+        store_model.deserialization_error = str(e)
+        store_model.save(update_fields=["deserialization_error"])
+        logger.warning(
+            "Failed to deserialize Store record %s for %s: %s",
+            store_model.id,
+            model_name,
+            e,
+        )
+        return False
+
+
 def _deserialize_from_store(profile, skip_erroring=False, filter=None):
     """
     Takes data from the store and integrates into the application.
 
     ALGORITHM: On a per syncable model basis, we iterate through each class model and we go through 2 possible cases:
 
     1. For class models that have a self referential foreign key, we iterate down the dependency tree deserializing model by model.
-    2. On a per app model basis, we append the field values to a single list, and do a single bulk insert/replace query.
+    2. For other models, we deserialize and validate each record, then save individually so that DB-level errors
+       (e.g. unique constraint violations) are caught and recorded per-record rather than silently lost or crashing
+       the entire deserialization.
 
     If a model fails to deserialize/validate, we exclude it from being marked as clean in the store.
     """
@@ -487,24 +529,25 @@ def _deserialize_from_store(profile, skip_erroring=False, filter=None):
                             app_model, _ = store_model._deserialize_store_model(
                                 fk_cache, sync_filter=filter
                             )
-                            if app_model:
-                                with mute_signals(signals.pre_save, signals.post_save):
-                                    app_model.save(update_dirty_bit_to=False)
-                            # we update a store model after we have deserialized it to be able to mark it as a clean parent
-                            store_model.dirty_bit = False
-                            store_model.deserialization_error = ""
-                            store_model.save(
-                                update_fields=["dirty_bit", "deserialization_error"]
-                            )
                         except (
                             exceptions.ValidationError,
                             exceptions.ObjectDoesNotExist,
                             ValueError,
                         ) as e:
                             excluded_list.append(store_model.id)
-                            # if the app model did not validate, we leave the store dirty bit set, but mark the error
                             store_model.deserialization_error = str(e)
                             store_model.save(update_fields=["deserialization_error"])
+                            continue
+                        if app_model:
+                            _save_deserialized_record(
+                                store_model, app_model, model.__name__, excluded_list
+                            )
+                        else:
+                            store_model.dirty_bit = False
+                            store_model.deserialization_error = ""
+                            store_model.save(
+                                update_fields=["dirty_bit", "deserialization_error"]
+                            )
 
                     # update lists with new clean parents and dirty children
                     clean_parents = store_models.filter(dirty_bit=False).char_ids_list()
@@ -529,9 +572,8 @@ def _deserialize_from_store(profile, skip_erroring=False, filter=None):
                 )
 
             else:
-                # collect all initially valid app models
+                # collect all initially valid app models and validate their FKs
                 app_models = []
-                fields = model._meta.fields
                 for store_model in store_models.filter(dirty_bit=True):
                     try:
                         (
@@ -541,7 +583,7 @@ def _deserialize_from_store(profile, skip_erroring=False, filter=None):
                             fk_cache, defer_fks=True, sync_filter=filter,
                         )
                         if app_model:
-                            app_models.append(app_model)
+                            app_models.append((store_model, app_model))
                         for fk_model, fk_refs in model_deferred_fks.items():
                             # validate that the FK references aren't to anything already in the
                             # excluded list, which should only contain models which failed to
@@ -571,40 +613,17 @@ def _deserialize_from_store(profile, skip_erroring=False, filter=None):
                 excluded_list.extend(model_excluded_pks)
                 deleted_list.extend(model_deleted_pks)
 
-                # array for holding db values from the fields of each model for this class
-                db_values = []
-                for app_model in app_models:
+                # save each app model individually so we can catch per-record
+                # DB-level errors (e.g. unique constraint violations)
+                for store_model, app_model in app_models:
                     if (
-                        app_model.pk not in excluded_list
-                        and app_model.pk not in deleted_list
+                        app_model.pk in excluded_list
+                        or app_model.pk in deleted_list
                     ):
-                        # handle any errors that might come from `get_db_prep_value`
-                        try:
-                            new_db_values = []
-                            for f in fields:
-                                value = getattr(app_model, f.attname)
-                                db_value = f.get_db_prep_value(value, connection)
-                                new_db_values.append(db_value)
-                            db_values += new_db_values
-                        except ValueError as e:
-                            excluded_list.append(app_model.pk)
-                            store_model = store_models.get(pk=app_model.pk)
-                            store_model.deserialization_error = str(e)
-                            store_model.save(update_fields=["deserialization_error"])
-
-                if db_values:
-                    with connection.cursor() as cursor:
-                        DBBackend._bulk_full_record_upsert(
-                            cursor,
-                            model._meta.db_table,
-                            fields,
-                            db_values,
-                        )
-
-                # clear dirty bit for all store records for this model/profile except for rows that did not validate
-                store_models.exclude(id__in=excluded_list).filter(
-                    dirty_bit=True
-                ).update(dirty_bit=False)
+                        continue
+                    _save_deserialized_record(
+                        store_model, app_model, model.__name__, excluded_list
+                    )
 
 
 def _queue_into_buffer_v1(transfersession):

tests/testapp/tests/sync/test_controller.py

@@ -535,6 +535,125 @@ def test_filtered_deserialization(self):
         self.assertFalse(MyUser.objects.filter(username="changed2").exists())
 
 
+class UniqueConstraintDeserializationTestCase(TestCase):
+    """
+    Regression tests for silent deserialization failure when two Store records
+    have different Morango IDs but violate a unique constraint on the app model table.
+
+    On SQLite, ``REPLACE INTO`` silently deletes the conflicting row and inserts the
+    new one, so no IntegrityError is raised. On PostgreSQL, the ``INSERT`` without an
+    ``ON CONFLICT`` clause for non-PK constraints would raise IntegrityError.
+
+    In both cases, the expected behavior is:
+    - The conflicting Store record should have ``deserialization_error`` populated
+    - Its ``dirty_bit`` should remain True
+    - A warning should be logged
+    """
+
+    def setUp(self):
+        InstanceIDModel.get_or_create_current_instance()
+        self.mc = MorangoProfileController("facilitydata")
+
+    def _create_conflicting_user_store_records(self):
+        """Create two Store records for MyUser with different IDs but the same unique username."""
+        self.user1_id = uuid.uuid4().hex
+        self.user2_id = uuid.uuid4().hex
+
+        user1 = MyUser(id=self.user1_id, username="duplicate", password="password")
+        user2 = MyUser(id=self.user2_id, username="duplicate", password="password")
+
+        self.store1 = StoreModelFacilityFactory(
+            id=self.user1_id,
+            serialized=json.dumps(user1.serialize()),
+            model_name="user",
+        )
+        self.store2 = StoreModelFacilityFactory(
+            id=self.user2_id,
+            serialized=json.dumps(user2.serialize()),
+            model_name="user",
+        )
+
+    def test_unique_violation_sets_deserialization_error(self):
+        """At least one conflicting record should have deserialization_error set."""
+        self._create_conflicting_user_store_records()
+
+        self.mc.deserialize_from_store()
+
+        # At most one app model should exist due to the unique constraint
+        self.assertLessEqual(MyUser.objects.filter(username="duplicate").count(), 1)
+
+        # Both Store records should still exist
+        self.assertEqual(
+            Store.objects.filter(id__in=[self.user1_id, self.user2_id]).count(), 2
+        )
+
+        # At least one Store record should have deserialization_error set
+        errored_stores = Store.objects.filter(
+            id__in=[self.user1_id, self.user2_id],
+        ).exclude(deserialization_error="")
+        self.assertGreater(
+            errored_stores.count(),
+            0,
+            "Expected at least one Store record to have deserialization_error "
+            "set for unique constraint violation, but none did.",
+        )
+
+    def test_unique_violation_leaves_dirty_bit(self):
+        """The conflicting record's dirty_bit should remain True."""
+        self._create_conflicting_user_store_records()
+
+        self.mc.deserialize_from_store()
+
+        # At least one Store record should still have dirty_bit=True
+        dirty_stores = Store.objects.filter(
+            id__in=[self.user1_id, self.user2_id],
+            dirty_bit=True,
+        )
+        self.assertGreater(
+            dirty_stores.count(),
+            0,
+            "Expected at least one Store record to still have dirty_bit=True "
+            "for a unique constraint violation, but all were cleared.",
+        )
+
+    def test_unique_violation_logs_warning(self):
+        """A warning should be logged when deserialization fails due to a unique constraint."""
+        self._create_conflicting_user_store_records()
+
+        with self.assertLogs("morango.sync.operations", level="WARNING") as cm:
+            self.mc.deserialize_from_store()
+
+        unique_warnings = [
+            msg for msg in cm.output if "unique" in msg.lower() or "integrity" in msg.lower()
+        ]
+        self.assertGreater(
+            len(unique_warnings),
+            0,
+            "Expected a warning log about unique constraint violation during "
+            "deserialization, but none was found. Logs: {}".format(cm.output),
+        )
+
+    def test_non_conflicting_records_still_deserialize(self):
+        """Records that don't conflict should still be deserialized even when others conflict."""
+        self._create_conflicting_user_store_records()
+
+        ok_user_id = uuid.uuid4().hex
+        ok_user = MyUser(id=ok_user_id, username="noconflict", password="password")
+        StoreModelFacilityFactory(
+            id=ok_user_id,
+            serialized=json.dumps(ok_user.serialize()),
+            model_name="user",
+        )
+
+        self.mc.deserialize_from_store()
+
+        self.assertTrue(MyUser.objects.filter(id=ok_user_id, username="noconflict").exists())
+
+        ok_store = Store.objects.get(id=ok_user_id)
+        self.assertFalse(ok_store.dirty_bit)
+        self.assertEqual(ok_store.deserialization_error, "")
+
+
 class SelfReferentialFKDeserializationTestCase(TestCase):
     def setUp(self):
         (self.current_id, _) = InstanceIDModel.get_or_create_current_instance()