Streaming deserialization part 3: sink

CHANGELOG.md

@@ -2,6 +2,16 @@
 
 List of the most important changes for each release.
 
+## 0.9.0
+- Deserializes `Store` records as a stream with chunked writes, so memory use no longer grows with the number of dirty records
+- Replaces `morango.sync.operations._deserialize_from_store` with `morango.sync.stream.deserialize.deserialize_from_store`. The `filter` argument is renamed to `sync_filter`, and `skip_erroring` to `skip_errored`
+- Replaces `Store._deserialize_store_model` with `morango.sync.stream.deserialize.deserialize_store_model`, which returns the validated app model, or `None` for a deleted record. The `defer_fks` option is removed
+- Records that reference a record which failed to save or was removed by a cascading delete now get a deserialization error, instead of failing the deserialization transaction
+- Records that propagate the deletion of a referenced record are now marked clean after deserialization
+- Children of a dirty parent that is left out of the deserialization run are now deserialized against the parent's existing app row
+- For models whose `clean_fields` loads related objects, a dirty self-referential chain is now deserialized one level per run
+- Logs deserialization warnings under `morango.sync.stream.deserialize` instead of `morango.sync.operations`
+
 ## 0.8.15
 - Improves buffer serialization performance by performing bulk counter lookup
 - Corrects an initialization issue with `SessionContext` that leads to an incorrect value for `is_push`

morango/models/core.py

@@ -16,12 +16,10 @@
 from django.db.models import Func
 from django.db.models import Max
 from django.db.models import Q
-from django.db.models import signals
 from django.db.models import TextField
 from django.db.models import Value
 from django.db.models.deletion import Collector
 from django.db.models.expressions import CombinedExpression
-from django.db.models.fields.related import ForeignKey
 from django.db.models.functions import Cast
 from django.db.models.functions import NullIf
 from django.utils import timezone
@@ -41,7 +39,6 @@
 from morango.models.utils import get_0_4_system_parameters
 from morango.models.utils import get_0_5_mac_address
 from morango.models.utils import get_0_5_system_id
-from morango.registry import syncable_models
 from morango.utils import _assert
 from morango.utils import exception_path
 from morango.utils import SETTINGS
@@ -501,82 +498,6 @@ def unset_deserialization_error(self):
         self.deserialization_error = None
         self.deserialization_exception = None
 
-    def _deserialize_store_model(self, fk_cache, defer_fks=False, sync_filter=None):  # noqa: C901
-        """
-        When deserializing a store model, we look at the deleted flags to know if we should delete the app model.
-        Upon loading the app model in memory we validate the app models fields, if any errors occurs we follow
-        foreign key relationships to see if the related model has been deleted to propagate that deletion to the target app model.
-        We return:
-            None => if the model was deleted successfully
-            model => if the model validates successfully
-
-        :param fk_cache: A cache for foreign key lookups
-        :type fk_cache: dict
-        :param defer_fks: Whether to defer foreign key lookups
-        :type defer_fks: bool
-        :param sync_filter: The current sync's filter, if any
-        :type sync_filter: Filter|None
-        """
-        deferred_fks = {}
-        klass_model = syncable_models.get_model(self.profile, self.model_name)
-        # if store model marked as deleted, attempt to delete in app layer
-        if self.deleted:
-            # Don't differentiate between deletion and hard deletion here,
-            # as we don't want to add additional tracking for models in either case,
-            # just to actually delete them.
-            # Import here to avoid circular import, as the utils module
-            # imports core models.
-            from morango.sync.utils import mute_signals
-
-            with mute_signals(signals.post_delete):
-                klass_model.syncing_objects.filter(id=self.id).delete()
-            return None, deferred_fks
-        else:
-            # load model into memory
-            app_model = klass_model.deserialize(
-                json.loads(self.serialized), sync_filter=sync_filter
-            )
-            app_model._morango_source_id = self.source_id
-            app_model._morango_partition = self.partition
-            app_model._morango_dirty_bit = False
-
-            try:
-                # validate and return the model
-                if defer_fks:
-                    deferred_fks = app_model.deferred_clean_fields(sync_filter=sync_filter)
-                else:
-                    app_model.cached_clean_fields(fk_cache, sync_filter=sync_filter)
-                return app_model, deferred_fks
-
-            except (exceptions.ValidationError, exceptions.ObjectDoesNotExist) as e:
-                logger.warning(
-                    "Error deserializing instance of {model} with id {id}: {error}".format(
-                        model=klass_model.__name__, id=app_model.id, error=e
-                    )
-                )
-
-                if not defer_fks and isinstance(e, exceptions.ObjectDoesNotExist):
-                    # check FKs in store to see if any of those models were deleted or hard_deleted to propagate to this model
-                    fk_ids = [
-                        getattr(app_model, field.attname)
-                        for field in app_model._meta.fields
-                        if isinstance(field, ForeignKey)
-                    ]
-                    for fk_id in fk_ids:
-                        try:
-                            st_model = Store.objects.get(id=fk_id)
-                            if st_model.deleted:
-                                # if hard deleted, propagate to store model
-                                if st_model.hard_deleted:
-                                    app_model._update_hard_deleted_models()
-                                app_model._update_deleted_models()
-                                return None, {}
-                        except Store.DoesNotExist:
-                            pass
-
-                # if we got here, it means the validation error wasn't handled by propagating deletion, so re-raise it
-                raise e
-
 
 class Buffer(AbstractStore):
     """

morango/sync/controller.py

@@ -5,8 +5,8 @@
 from morango.constants import transfer_stages
 from morango.constants import transfer_statuses
 from morango.registry import session_middleware
-from morango.sync.operations import _deserialize_from_store
 from morango.sync.operations import OperationLogger
+from morango.sync.stream.deserialize import deserialize_from_store
 from morango.sync.stream.serialize import serialize_into_store
 from morango.sync.utils import SyncSignalGroup
 from morango.utils import _assert
@@ -33,7 +33,9 @@ def deserialize_from_store(self, skip_erroring=False, sync_filter=None):
         with OperationLogger("Deserializing records", "Deserialization complete"):
             # we first serialize to avoid deserialization merge conflicts
             serialize_into_store(self.profile, sync_filter=sync_filter)
-            _deserialize_from_store(self.profile, filter=sync_filter, skip_erroring=skip_erroring)
+            deserialize_from_store(
+                self.profile, sync_filter=sync_filter, skip_errored=skip_erroring
+            )
 
     def create_network_connection(self, base_url, **kwargs):
         from morango.sync.syncsession import NetworkSyncConnection