From 9633c2bdef5a5304262fe3cd0e755377f0aaaca4 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:20:13 -0700 Subject: [PATCH 01/12] Move deserialize FK helpers to module level for reuse by the sink Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 90 +++++++++++++++--------------- 1 file changed, 46 insertions(+), 44 deletions(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index ac8b4ce2..e032c3ff 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -157,6 +157,36 @@ def stream_for_filter( yield DeserializeTask(store_model, self.fk_cache) +def fk_references(app_model: SyncableModel) -> List[Tuple[ForeignKey, str]]: + """The FK fields on the app model that hold a value, paired with that value""" + references = [] + for field in app_model._meta.fields: + if not isinstance(field, ForeignKey): + continue + target_id = getattr(app_model, field.attname) + if target_id is not None: + references.append((field, target_id)) + return references + + +def failed_target_error(task: DeserializeTask, field: ForeignKey, target_id: str) -> Exception: + """The error for a task whose FK target failed to deserialize in the run""" + if field.attname == task.self_referential_fk(): + return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) + return exceptions.ValidationError( + "{} with id {} failed to deserialize".format( + field.related_model._meta.verbose_name, target_id + ) + ) + + +def deleted_target_error(field: ForeignKey, target_id: str) -> Exception: + """The error for a task whose FK target was deleted in the run""" + return exceptions.ValidationError( + "{} with id {} was deleted".format(field.related_model._meta.verbose_name, target_id) + ) + + class AppModelDeserialize(Transform[DeserializeTask]): """ Builds the unsaved app model from each task's serialized store data. Deleted store records @@ -216,87 +246,59 @@ def transform(self, task: DeserializeTask) -> DeserializeTask: self.deleted_ids[task.id] = task.store.hard_deleted return task - fk_references = self._fk_references(task.app_model) + references = fk_references(task.app_model) - for field, target_id in fk_references: + for field, target_id in references: if target_id in self.deleted_ids: if task.self_referential_fk() is not None: # the sink has not deleted the target's app row yet, so validation can't # report the missing target; fail as legacy does once the row is gone - self._fail(task, self._deleted_target_error(field, target_id)) + self._fail(task, deleted_target_error(field, target_id)) else: self._propagate_deletion(task, hard=self.deleted_ids[target_id]) return task - for field, target_id in fk_references: + for field, target_id in references: if target_id in self.failed_ids: - self._fail(task, self._failed_target_error(task, field, target_id)) + self._fail(task, failed_target_error(task, field, target_id)) return task try: task.app_model.cached_clean_fields(task.fk_cache, sync_filter=self.sync_filter) except (exceptions.ValidationError, exceptions.ObjectDoesNotExist, ValueError) as e: - self._classify_failure(task, fk_references, e) + self._classify_failure(task, references, e) return task - @staticmethod - def _fk_references(app_model: SyncableModel) -> List[Tuple[ForeignKey, str]]: - """The FK fields on the app model that hold a value, paired with that value""" - references = [] - for field in app_model._meta.fields: - if not isinstance(field, ForeignKey): - continue - target_id = getattr(app_model, field.attname) - if target_id is not None: - references.append((field, target_id)) - return references - - @staticmethod - def _failed_target_error(task: DeserializeTask, field: ForeignKey, target_id: str) -> Exception: - if field.attname == task.self_referential_fk(): - return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) - return exceptions.ValidationError( - "{} with id {} failed to deserialize".format( - field.related_model._meta.verbose_name, target_id - ) - ) - - @staticmethod - def _deleted_target_error(field: ForeignKey, target_id: str) -> Exception: - return exceptions.ValidationError( - "{} with id {} was deleted".format(field.related_model._meta.verbose_name, target_id) - ) - def _classify_failure( self, task: DeserializeTask, - fk_references: List[Tuple[ForeignKey, str]], + references: List[Tuple[ForeignKey, str]], error: Exception, ) -> None: """ Resolves a validation failure using the store records of the FK targets. Only runs on failure, so it costs at most one query per failed record and none per valid record. """ - targets = self._lookup_targets(fk_references) + targets = self._lookup_targets(references) # a reference to a deleted record propagates the deletion, instead of failing if task.self_referential_fk() is None or isinstance(error, exceptions.ObjectDoesNotExist): - for _, target_id in fk_references: + for _, target_id in references: target = targets.get(target_id) if target is not None and (target.deleted or target.hard_deleted): self._propagate_deletion(task, hard=target.hard_deleted) return - self._fail(task, self._failure_error(task, fk_references, targets, error)) + self._fail(task, self._failure_error(task, references, targets, error)) @staticmethod def _lookup_targets( - fk_references: List[Tuple[ForeignKey, str]], + references: List[Tuple[ForeignKey, str]], ) -> Dict[str, _TargetState]: """The store state of each FK target that has a store record, keyed by the target ID""" - if not fk_references: + if not references: return {} - target_ids = [target_id for _, target_id in fk_references] + target_ids = [target_id for _, target_id in references] return { target_id: _TargetState(deleted, hard_deleted, dirty_bit) for target_id, deleted, hard_deleted, dirty_bit in Store.objects.filter( @@ -307,7 +309,7 @@ def _lookup_targets( def _failure_error( self, task: DeserializeTask, - fk_references: List[Tuple[ForeignKey, str]], + references: List[Tuple[ForeignKey, str]], targets: Dict[str, _TargetState], error: Exception, ) -> Exception: @@ -316,7 +318,7 @@ def _failure_error( parent take precedence, then missing targets, otherwise the original validation error. """ self_ref_fk = task.self_referential_fk() - for field, target_id in fk_references: + for field, target_id in references: if field.attname != self_ref_fk: continue if target_id not in targets: @@ -326,7 +328,7 @@ def _failure_error( ): return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) - for field, target_id in fk_references: + for field, target_id in references: if target_id not in targets: return IntegrityError( "{from_model}.{from_field} references non-existent {to_model} instance with " From fc9ab54b40ce9b24a4042647bd7c205585a2feab Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:22:41 -0700 Subject: [PATCH 02/12] Add write helpers for deleting app models and tracking propagated deletions Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 56 ++++++++ .../tests/sync/stream/test_deserialize.py | 121 ++++++++++++++++++ 2 files changed, 177 insertions(+) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index e032c3ff..4be16139 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -10,18 +10,25 @@ from typing import Type from django.core import exceptions +from django.db import router from django.db.models import ForeignKey +from django.db.models import Model +from django.db.models import signals +from django.db.models.deletion import Collector from django.db.utils import IntegrityError from morango.errors import MorangoDirtyParent from morango.errors import MorangoMissingParent from morango.models.certificates import Filter +from morango.models.core import DeletedModels +from morango.models.core import HardDeletedModels from morango.models.core import Store from morango.models.core import SyncableModel from morango.registry import syncable_models from morango.sync.stream.core import Transform from morango.sync.stream.source import MorangoSource from morango.sync.stream.source import SourceTask +from morango.sync.utils import mute_signals DIRTY_PARENT_MESSAGE = "Parent is dirty; could not deserialize." MISSING_PARENT_MESSAGE = "Parent does not exist in Store; could not deserialize." @@ -187,6 +194,55 @@ def deleted_target_error(field: ForeignKey, target_id: str) -> Exception: ) +def row_key(model: Type[Model], pk) -> Tuple[str, str]: + """Identifies an app row across models, as the concrete model's label and the primary key""" + return model._meta.concrete_model._meta.label_lower, str(pk) + + +def delete_app_models(model: Type[SyncableModel], ids: List[str]) -> Set[Tuple[str, str]]: + """ + Deletes app rows without tracking them, and Django cascades to the rows that reference them. + + :return: The row keys of the cascaded rows, not including the rows in `ids` + """ + if not ids: + return set() + + with mute_signals(signals.post_delete): + # the collector checks for signal receivers when deciding to fast delete, so it must + # collect while `post_delete` is muted + collector = Collector(using=router.db_for_write(model)) + collector.collect(model.syncing_objects.filter(id__in=ids)) + + keys = set() + for related_model, instances in collector.data.items(): + keys.update(row_key(related_model, instance.pk) for instance in instances) + for qs in collector.fast_deletes: + keys.update(row_key(qs.model, pk) for pk in qs.values_list("pk", flat=True)) + + collector.delete() + + return keys - {row_key(model, pk) for pk in ids} + + +def record_propagated_deletions(tasks: List[DeserializeTask]) -> None: + """ + Adds the tasks to `DeletedModels`, and those with a propagated hard deletion to + `HardDeletedModels`, so the next serialization marks their store records deleted + """ + deleted = [] + hard_deleted = [] + for task in tasks: + deleted.append(DeletedModels(id=task.id, profile=task.store.profile)) + if task.outcome == DeserializeOutcome.PROPAGATE_HARD_DELETE: + hard_deleted.append(HardDeletedModels(id=task.id, profile=task.store.profile)) + + if deleted: + DeletedModels.objects.bulk_create(deleted, ignore_conflicts=True) + if hard_deleted: + HardDeletedModels.objects.bulk_create(hard_deleted, ignore_conflicts=True) + + class AppModelDeserialize(Transform[DeserializeTask]): """ Builds the unsaved app model from each task's serialized store data. Deleted store records diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 37b36e8c..80393cfe 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -5,12 +5,15 @@ from django.core.exceptions import ValidationError from django.core.serializers.json import DjangoJSONEncoder from django.db import connection +from django.db.models import SET_NULL +from django.db.models import signals from django.db.utils import IntegrityError from django.test import SimpleTestCase from django.test import TestCase from django.test.utils import CaptureQueriesContext from facility_profile.models import ConditionalLog from facility_profile.models import Facility +from facility_profile.models import InteractionLog from facility_profile.models import MyUser from facility_profile.models import SummaryLog @@ -25,8 +28,11 @@ from morango.sync.stream.core import Sink from morango.sync.stream.deserialize import AppModelDeserialize from morango.sync.stream.deserialize import AppModelValidate +from morango.sync.stream.deserialize import delete_app_models from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeTask +from morango.sync.stream.deserialize import record_propagated_deletions +from morango.sync.stream.deserialize import row_key from morango.sync.stream.deserialize import StoreModelSource from morango.utils import exception_path @@ -1153,3 +1159,118 @@ def test_parent_deleted_in_run__parent_not_loaded(self): grandchild = self._facility(parent=child, order=2) self._assert_parity(child.id, grandchild.id) + + +class DeleteAppModelsTestCase(TestCase): + def _facility(self, name, parent=None): + facility = Facility(name=name, parent=parent) + facility.save(update_dirty_bit_to=False) + return facility + + def _user(self, username="user"): + user = MyUser(username=username, password="password") + user.save(update_dirty_bit_to=False) + return user + + def test_self_ref_cascade(self): + root = self._facility("root") + child = self._facility("child", parent=root) + grandchild = self._facility("grandchild", parent=child) + + cascaded = delete_app_models(Facility, [root.id]) + + self.assertEqual(cascaded, {row_key(Facility, child.id), row_key(Facility, grandchild.id)}) + self.assertFalse( + Facility.objects.filter(id__in=[root.id, child.id, grandchild.id]).exists() + ) + self.assertFalse(DeletedModels.objects.exists()) + self.assertFalse(HardDeletedModels.objects.exists()) + + def test_cross_model_cascade__fast_delete(self): + """With `post_delete` muted, Django deletes the logs without loading them""" + user = self._user() + log = SummaryLog(user=user) + log.save(update_dirty_bit_to=False) + + cascaded = delete_app_models(MyUser, [user.id]) + + self.assertEqual(cascaded, {row_key(SummaryLog, log.id)}) + self.assertFalse(MyUser.objects.filter(id=user.id).exists()) + self.assertFalse(SummaryLog.objects.filter(id=log.id).exists()) + self.assertFalse(DeletedModels.objects.exists()) + + def test_set_null_not_cascaded(self): + user = self._user() + log = InteractionLog(user=user) + log.save(update_dirty_bit_to=False) + + remote_field = InteractionLog._meta.get_field("user").remote_field + with mock.patch.object(remote_field, "on_delete", SET_NULL): + cascaded = delete_app_models(MyUser, [user.id]) + + self.assertEqual(cascaded, set()) + self.assertIsNone(InteractionLog.objects.get(id=log.id).user_id) + + def test_post_delete_muted(self): + root = self._facility("root") + self._facility("child", parent=root) + receiver = mock.Mock() + signals.post_delete.connect(receiver) + self.addCleanup(signals.post_delete.disconnect, receiver) + + delete_app_models(Facility, [root.id]) + + receiver.assert_not_called() + + def test_missing_rows(self): + self.assertEqual(delete_app_models(Facility, [uuid.uuid4().hex]), set()) + + def test_empty_ids(self): + with self.assertNumQueries(0): + self.assertEqual(delete_app_models(Facility, []), set()) + + +class RecordPropagatedDeletionsTestCase(DeserializeFixturesMixin, TestCase): + def _task(self, hard=False): + task = DeserializeTask(self._user(), {}) + task.set_propagated_deletion(hard=hard) + return task + + def test_soft(self): + task = self._task() + + record_propagated_deletions([task]) + + self.assertEqual( + list(DeletedModels.objects.values_list("id", "profile")), [(task.id, self.profile)] + ) + self.assertFalse(HardDeletedModels.objects.exists()) + + def test_hard(self): + task = self._task(hard=True) + + record_propagated_deletions([task]) + + self.assertTrue(DeletedModels.objects.filter(id=task.id, profile=self.profile).exists()) + self.assertTrue(HardDeletedModels.objects.filter(id=task.id, profile=self.profile).exists()) + + def test_existing_row(self): + task = self._task() + DeletedModels.objects.create(id=task.id, profile=self.profile) + + record_propagated_deletions([task]) + + self.assertEqual(DeletedModels.objects.filter(id=task.id).count(), 1) + + def test_empty(self): + with self.assertNumQueries(0): + record_propagated_deletions([]) + + def test_full_chunk(self): + """Crosses SQLite's 499-row batch limit for `bulk_create`""" + tasks = [self._task(hard=True) for _ in range(500)] + + record_propagated_deletions(tasks) + + self.assertEqual(DeletedModels.objects.count(), 500) + self.assertEqual(HardDeletedModels.objects.count(), 500) From 0c64fa13e64f92e39605df7411fd72e4b56e8751 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:25:36 -0700 Subject: [PATCH 03/12] Add sink that writes deserialization outcomes per chunk Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 126 ++++++++++ .../tests/sync/stream/test_deserialize.py | 229 ++++++++++++++++++ 2 files changed, 355 insertions(+) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 4be16139..26264e39 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -1,4 +1,6 @@ import json +import logging +from collections import defaultdict from collections import namedtuple from enum import Enum from typing import Dict @@ -11,6 +13,7 @@ from django.core import exceptions from django.db import router +from django.db import transaction from django.db.models import ForeignKey from django.db.models import Model from django.db.models import signals @@ -25,11 +28,14 @@ from morango.models.core import Store from morango.models.core import SyncableModel from morango.registry import syncable_models +from morango.sync.stream.core import Sink from morango.sync.stream.core import Transform from morango.sync.stream.source import MorangoSource from morango.sync.stream.source import SourceTask from morango.sync.utils import mute_signals +logger = logging.getLogger(__name__) + DIRTY_PARENT_MESSAGE = "Parent is dirty; could not deserialize." MISSING_PARENT_MESSAGE = "Parent does not exist in Store; could not deserialize." @@ -419,3 +425,123 @@ def _propagate_deletion(self, task: DeserializeTask, hard: bool) -> None: def _fail(self, task: DeserializeTask, error: Exception) -> None: task.add_error(error) self.failed_ids.add(task.id) + + +class DeserializeSink(Sink[List[DeserializeTask]]): + """ + Writes each task's resolved outcome to the app tables, the store and the deletion tracking + tables. Consumes chunks of tasks, which the pipeline partitions by model. + + Must be instantiated per pipeline run, since it tracks the records that failed and the app rows + removed by cascading deletes. The buffer in front of the sink means a task is validated before + the sink writes the chunks ahead of it, so a save is checked against both before saving. + """ + + def __init__(self): + self.failed_ids: Set[str] = set() + self.cascaded_keys: Set[Tuple[str, str]] = set() + self.counts: Dict[DeserializeOutcome, int] = defaultdict(int) + + def consume(self, tasks: List[DeserializeTask]) -> None: + by_outcome = defaultdict(list) + for task in tasks: + by_outcome[task.outcome].append(task) + + clean = [] + errored = [] + + for task in by_outcome[DeserializeOutcome.ERROR]: + self._record_error(task, task.errors[0]) + errored.append(task) + + # deletes run before saves, so the saves are checked against the rows they cascade to + deletes = by_outcome[DeserializeOutcome.DELETE] + self._delete_all(deletes) + clean.extend(deletes) + + propagated = ( + by_outcome[DeserializeOutcome.PROPAGATE_DELETE] + + by_outcome[DeserializeOutcome.PROPAGATE_HARD_DELETE] + ) + record_propagated_deletions(propagated) + clean.extend(propagated) + + self._save_all(by_outcome[DeserializeOutcome.SAVE], clean, errored) + self._write_stores(clean, errored) + + def _delete_all(self, tasks: List[DeserializeTask]) -> None: + """Deletes the tasks' app rows with one delete per model, recording the cascaded rows""" + ids_by_model = defaultdict(list) + for task in tasks: + ids_by_model[task.model].append(task.id) + for model, ids in ids_by_model.items(): + self.cascaded_keys |= delete_app_models(model, ids) + + def _save_all( + self, + tasks: List[DeserializeTask], + clean: List[DeserializeTask], + errored: List[DeserializeTask], + ) -> None: + """Saves the tasks' app models in arrival order, adding each task to `clean` or `errored`""" + if not tasks: + return + with mute_signals(signals.pre_save, signals.post_save): + for task in tasks: + if self._save(task): + clean.append(task) + else: + errored.append(task) + + def _write_stores(self, clean: List[DeserializeTask], errored: List[DeserializeTask]) -> None: + """Clears the store records that were written, and saves the errors of those that weren't""" + if clean: + Store.objects.filter(id__in=[task.id for task in clean]).update( + dirty_bit=False, deserialization_error=None, deserialization_exception=None + ) + # errors are rare, and `bulk_update` is slower than saving each record + for task in errored: + task.store.save(update_fields=["deserialization_error", "deserialization_exception"]) + + for task in clean + errored: + self.counts[task.outcome] += 1 + + def summary(self) -> Dict[str, int]: + """The number of tasks consumed per outcome, counting failed saves as errors""" + return {outcome.value: count for outcome, count in self.counts.items()} + + def _save(self, task: DeserializeTask) -> bool: + """Saves the task's app model in its own savepoint, recording the error if it fails""" + try: + with transaction.atomic(): + task.app_model.save(update_dirty_bit_to=False) + except ( + exceptions.ValidationError, + exceptions.ObjectDoesNotExist, + ValueError, + IntegrityError, + ) as e: + self._record_error(task, e) + return False + return True + + def _record_error(self, task: DeserializeTask, error: Exception) -> None: + if error not in task.errors: + task.add_error(error) + task.store.set_deserialization_error(error) + self.failed_ids.add(task.id) + log_deserialization_error(task, error) + + +def log_deserialization_error(task: DeserializeTask, error: Exception) -> None: + """Logs a record that failed to deserialize, with the message used by legacy deserialization""" + logger.warning( + "Failed to deserialize Store record %s for %s: %s", + task.id, + task.model.__name__, + error, + ) + + +def store_model_partition_fn(task: DeserializeTask) -> str: + return task.store.model_name diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 80393cfe..bc255a3a 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -25,14 +25,17 @@ from morango.models.core import Store from morango.models.core import SyncableModel from morango.sync.operations import _deserialize_from_store +from morango.sync.stream.core import Buffer from morango.sync.stream.core import Sink from morango.sync.stream.deserialize import AppModelDeserialize from morango.sync.stream.deserialize import AppModelValidate from morango.sync.stream.deserialize import delete_app_models from morango.sync.stream.deserialize import DeserializeOutcome +from morango.sync.stream.deserialize import DeserializeSink from morango.sync.stream.deserialize import DeserializeTask from morango.sync.stream.deserialize import record_propagated_deletions from morango.sync.stream.deserialize import row_key +from morango.sync.stream.deserialize import store_model_partition_fn from morango.sync.stream.deserialize import StoreModelSource from morango.utils import exception_path @@ -1274,3 +1277,229 @@ def test_full_chunk(self): self.assertEqual(DeletedModels.objects.count(), 500) self.assertEqual(HardDeletedModels.objects.count(), 500) + + +class ListSink(Sink[DeserializeTask]): + """Collects the tasks in stream order""" + + def __init__(self): + self.tasks = [] + + def consume(self, task: DeserializeTask) -> None: + self.tasks.append(task) + + +class DeserializeSinkTestCase(DeserializeFixturesMixin, TestCase): + logger_name = "morango.sync.stream.deserialize" + + def _transform(self): + """The dirty store records as tasks with resolved outcomes, in stream order""" + sink = ListSink() + ( + StoreModelSource(self.profile) + .pipe(AppModelDeserialize()) + .pipe(AppModelValidate()) + .end(sink) + ) + return sink.tasks + + def _run(self, sink=None): + sink = sink or DeserializeSink() + ( + StoreModelSource(self.profile) + .pipe(AppModelDeserialize()) + .pipe(AppModelValidate()) + .pipe(Buffer(size=500, partition_fn=store_model_partition_fn)) + .end(sink) + ) + return sink + + def _assert_store(self, store_id, dirty_bit, exception=None): + store = Store.objects.get(id=store_id) + self.assertEqual(store.dirty_bit, dirty_bit) + self.assertEqual(store.deserialization_exception, exception) + if exception is None: + self.assertIsNone(store.deserialization_error) + else: + self.assertTrue(store.deserialization_error) + + def _app_user(self, username="user"): + user = MyUser(username=username, password="password") + user.save(update_dirty_bit_to=False) + self._store(user, dirty_bit=False) + return user + + def test_save(self): + user = self._user(deserialization_error="old", deserialization_exception="old.Error") + + self._run() + + self.assertFalse(MyUser.objects.get(id=user.id)._morango_dirty_bit) + self._assert_store(user.id, dirty_bit=False) + + def test_save__failure(self): + users = [ + self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + for _ in range(2) + ] + + with self.assertLogs(self.logger_name, "WARNING") as logs: + self._run() + + saved = MyUser.objects.get(username="same") + (failed,) = [user for user in users if user.id != saved.id] + self._assert_store(saved.id, dirty_bit=False) + self._assert_store(failed.id, dirty_bit=True, exception="django.db.utils.IntegrityError") + self.assertEqual(len(logs.output), 1) + self.assertIn(failed.id, logs.output[0]) + + def test_save__failure__task_errored(self): + self._app_user(username="same") + self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + (task,) = self._transform() + + with self.assertLogs(self.logger_name, "WARNING"): + DeserializeSink().consume([task]) + + self.assertEqual(task.outcome, DeserializeOutcome.ERROR) + self.assertEqual(len(task.errors), 1) + self.assertIsInstance(task.errors[0], IntegrityError) + + def test_save__signals_muted(self): + self._user() + receiver = mock.Mock() + signals.pre_save.connect(receiver) + signals.post_save.connect(receiver) + self.addCleanup(signals.pre_save.disconnect, receiver) + self.addCleanup(signals.post_save.disconnect, receiver) + + self._run() + + receiver.assert_not_called() + + def test_delete(self): + user = self._app_user() + Store.objects.filter(id=user.id).update(deleted=True, dirty_bit=True) + + self._run() + + self.assertFalse(MyUser.objects.filter(id=user.id).exists()) + self._assert_store(user.id, dirty_bit=False) + self.assertFalse(DeletedModels.objects.exists()) + + def test_delete__app_row_missing(self): + user = self._user(deleted=True) + + self._run() + + self._assert_store(user.id, dirty_bit=False) + + def test_propagate_delete(self): + user = self._user(dirty_bit=False, deleted=True) + log = self._summary_log(user.id) + + self._run() + + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store(log.id, dirty_bit=False) + self.assertEqual(list(DeletedModels.objects.values_list("id", flat=True)), [log.id]) + self.assertFalse(HardDeletedModels.objects.exists()) + + def test_propagate_hard_delete(self): + user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) + log = self._summary_log(user.id) + + self._run() + + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store(log.id, dirty_bit=False) + self.assertEqual(list(DeletedModels.objects.values_list("id", flat=True)), [log.id]) + self.assertEqual(list(HardDeletedModels.objects.values_list("id", flat=True)), [log.id]) + + def test_error(self): + facility = self._facility(serialized="{bad") + + with self.assertLogs(self.logger_name, "WARNING") as logs: + self._run() + + self.assertFalse(Facility.objects.exists()) + self._assert_store(facility.id, dirty_bit=True, exception="json.decoder.JSONDecodeError") + self.assertIn(facility.id, logs.output[0]) + + def test_error__single_error(self): + self._facility(serialized="{bad") + (task,) = self._transform() + + with self.assertLogs(self.logger_name, "WARNING"): + DeserializeSink().consume([task]) + + self.assertEqual(len(task.errors), 1) + + def test_delete__mixed_models(self): + """The pipeline partitions chunks by model, but each model's rows are deleted regardless""" + user = self._app_user() + facility = Facility(name="facility") + facility.save(update_dirty_bit_to=False) + self._store(facility, dirty_bit=False, _self_ref_fk="", _self_ref_order=0) + Store.objects.filter(id__in=[user.id, facility.id]).update(deleted=True, dirty_bit=True) + tasks = self._transform() + self.assertEqual({task.model for task in tasks}, {MyUser, Facility}) + + DeserializeSink().consume(tasks) + + self.assertFalse(MyUser.objects.exists()) + self.assertFalse(Facility.objects.exists()) + self._assert_store(user.id, dirty_bit=False) + self._assert_store(facility.id, dirty_bit=False) + + def test_summary(self): + self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + deleted_user = self._user(deleted=True) + self._summary_log(deleted_user.id) + self._facility(serialized="{bad") + + with self.assertLogs(self.logger_name, "WARNING"): + sink = self._run() + + self.assertEqual( + sink.summary(), + { + DeserializeOutcome.SAVE.value: 1, + DeserializeOutcome.ERROR.value: 2, + DeserializeOutcome.DELETE.value: 1, + DeserializeOutcome.PROPAGATE_DELETE.value: 1, + }, + ) + + def _count_queries(self, sink, tasks): + with CaptureQueriesContext(connection) as context: + sink.consume(tasks) + return len( + [ + query + for query in context.captured_queries + if not query["sql"].upper().startswith(("SAVEPOINT", "RELEASE SAVEPOINT")) + ] + ) + + def _batched_chunk(self, deleted_user_id, size): + """A chunk of one deleted log and `size` logs that propagate a deletion""" + self._summary_log(deleted_user_id, deleted=True) + for _ in range(size): + self._summary_log(deleted_user_id) + tasks = self._transform() + self.assertEqual(len(tasks), size + 1) + return tasks + + def test_consume__batched(self): + deleted_user = self._user(dirty_bit=False, deleted=True) + sink = DeserializeSink() + + small = self._count_queries(sink, self._batched_chunk(deleted_user.id, 10)) + large = self._count_queries(sink, self._batched_chunk(deleted_user.id, 500)) + + # SQLite splits the 500 `DeletedModels` inserts into batches of 499 + self.assertEqual(large - small, 1 if connection.vendor == "sqlite" else 0) + self.assertEqual(DeletedModels.objects.count(), 510) + self.assertFalse(Store.objects.filter(dirty_bit=True).exists()) From bdad4e644b1f706397eb5a98c712c5061ea27aef Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:26:37 -0700 Subject: [PATCH 04/12] Guard sink saves against FK targets that failed or were cascade-deleted FK constraints are deferred, so saving a record whose target the sink failed or removed earlier in the run would only fail at commit. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 19 ++- .../tests/sync/stream/test_deserialize.py | 108 +++++++++++++++++- 2 files changed, 125 insertions(+), 2 deletions(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 26264e39..cc6c7c45 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -488,11 +488,28 @@ def _save_all( return with mute_signals(signals.pre_save, signals.post_save): for task in tasks: - if self._save(task): + error = self._removed_target_error(task) + if error is not None: + self._record_error(task, error) + errored.append(task) + elif self._save(task): clean.append(task) else: errored.append(task) + def _removed_target_error(self, task: DeserializeTask) -> Optional[Exception]: + """ + The error for a task whose FK target was removed after the task was validated, because the + target failed in the sink or a delete cascaded to it. FK constraints are deferred, so saving + the task would only fail when the transaction commits. + """ + for field, target_id in fk_references(task.app_model): + if target_id in self.failed_ids: + return failed_target_error(task, field, target_id) + if row_key(field.related_model, target_id) in self.cascaded_keys: + return deleted_target_error(field, target_id) + return None + def _write_stores(self, clean: List[DeserializeTask], errored: List[DeserializeTask]) -> None: """Clears the store records that were written, and saves the errors of those that weren't""" if clean: diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index bc255a3a..1047cfe0 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -1289,7 +1289,9 @@ def consume(self, task: DeserializeTask) -> None: self.tasks.append(task) -class DeserializeSinkTestCase(DeserializeFixturesMixin, TestCase): +class DeserializeSinkFixturesMixin(DeserializeFixturesMixin): + """Runs the transforms into `DeserializeSink`, and checks what it wrote""" + logger_name = "morango.sync.stream.deserialize" def _transform(self): @@ -1329,6 +1331,8 @@ def _app_user(self, username="user"): self._store(user, dirty_bit=False) return user + +class DeserializeSinkTestCase(DeserializeSinkFixturesMixin, TestCase): def test_save(self): user = self._user(deserialization_error="old", deserialization_exception="old.Error") @@ -1503,3 +1507,105 @@ def test_consume__batched(self): self.assertEqual(large - small, 1 if connection.vendor == "sqlite" else 0) self.assertEqual(DeletedModels.objects.count(), 510) self.assertFalse(Store.objects.filter(dirty_bit=True).exists()) + + +class DeserializeSinkGuardsTestCase(DeserializeSinkFixturesMixin, TestCase): + """Saves whose FK target the sink failed or removed after the save was validated""" + + def _app_facility(self, name, parent=None): + facility = Facility(name=name, parent=parent) + facility.save(update_dirty_bit_to=False) + self._store( + facility, + dirty_bit=False, + _self_ref_fk=facility.parent_id or "", + _self_ref_order=0 if parent is None else 1, + ) + return facility + + def _error(self, store_id): + return Store.objects.get(id=store_id).deserialization_error + + def test_failed_target__self_ref(self): + parent = self._app_facility("parent") + child = self._facility(parent=parent, order=1) + tasks = self._transform() + sink = DeserializeSink() + sink.failed_ids.add(parent.id) + + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume(tasks) + + self.assertFalse(Facility.objects.filter(id=child.id).exists()) + self._assert_store(child.id, dirty_bit=True, exception=MorangoDirtyParent.path()) + self.assertIn(child.id, sink.failed_ids) + + def test_failed_target__cross_model(self): + user = self._app_user() + log = self._summary_log(user.id) + tasks = self._transform() + sink = DeserializeSink() + sink.failed_ids.add(user.id) + + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume(tasks) + + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store( + log.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertIn(user.id, self._error(log.id)) + self.assertIn(log.id, sink.failed_ids) + + def test_failed_save(self): + """The log was validated against the user before the user's save failed""" + self._app_user(username="same") + user = self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + log = self._summary_log(user.id) + + with self.assertLogs(self.logger_name, "WARNING"): + self._run() + + self._assert_store(user.id, dirty_bit=True, exception="django.db.utils.IntegrityError") + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store( + log.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertIn("failed to deserialize", self._error(log.id)) + + def _cascade_fixture(self): + """A deleted root, its clean child, and a dirty grandchild validated against the child""" + root = self._app_facility("root") + child = self._app_facility("child", parent=root) + Store.objects.filter(id=root.id).update(deleted=True, dirty_bit=True) + grandchild = self._facility(parent=child, order=2) + tasks = {task.id: task for task in self._transform()} + self.assertEqual(tasks[grandchild.id].outcome, DeserializeOutcome.SAVE) + return root, child, grandchild, tasks + + def _assert_cascaded(self, child, grandchild): + self.assertFalse(Facility.objects.exists()) + self._assert_store( + grandchild.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertIn("was deleted", self._error(grandchild.id)) + self.assertIn(child.id, self._error(grandchild.id)) + + def test_cascaded_target__same_chunk(self): + root, child, grandchild, tasks = self._cascade_fixture() + sink = DeserializeSink() + + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume([tasks[root.id], tasks[grandchild.id]]) + + self._assert_cascaded(child, grandchild) + + def test_cascaded_target__later_chunk(self): + root, child, grandchild, tasks = self._cascade_fixture() + sink = DeserializeSink() + + sink.consume([tasks[root.id]]) + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume([tasks[grandchild.id]]) + + self._assert_cascaded(child, grandchild) From 4eee0a87394f5e1d4b222b807d6ad2ab635ba97f Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:28:57 -0700 Subject: [PATCH 05/12] Add streaming deserialize_from_store pipeline entry point Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 30 ++++ .../tests/sync/stream/test_deserialize.py | 163 ++++++++++++++---- 2 files changed, 157 insertions(+), 36 deletions(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index cc6c7c45..84dedc98 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -28,6 +28,8 @@ from morango.models.core import Store from morango.models.core import SyncableModel from morango.registry import syncable_models +from morango.sync.db import begin_transaction +from morango.sync.stream.core import Buffer from morango.sync.stream.core import Sink from morango.sync.stream.core import Transform from morango.sync.stream.source import MorangoSource @@ -562,3 +564,31 @@ def log_deserialization_error(task: DeserializeTask, error: Exception) -> None: def store_model_partition_fn(task: DeserializeTask) -> str: return task.store.model_name + + +CHUNK_SIZE = 500 + + +def deserialize_from_store( + profile: str, + sync_filter: Optional[Filter] = None, + skip_errored: bool = False, +) -> None: + """ + Constructs and executes the deserialization pipeline, streaming dirty store records through + transforms that resolve each record's outcome, into a sink that writes them in chunks. + + :param profile: The Morango model profile + :param sync_filter: The Filter object for this sync + :param skip_errored: Whether to skip Store records with deserialization errors + """ + with begin_transaction(sync_filter, isolated=True): + sink = DeserializeSink() + ( + StoreModelSource(profile, sync_filter=sync_filter, skip_errored=skip_errored) + .pipe(AppModelDeserialize(sync_filter=sync_filter)) + .pipe(AppModelValidate(sync_filter=sync_filter)) + .pipe(Buffer(size=CHUNK_SIZE, partition_fn=store_model_partition_fn)) + .end(sink) + ) + logger.info("Deserialization done: %s", sink.summary()) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 1047cfe0..675e8b67 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -10,6 +10,7 @@ from django.db.utils import IntegrityError from django.test import SimpleTestCase from django.test import TestCase +from django.test import TransactionTestCase from django.test.utils import CaptureQueriesContext from facility_profile.models import ConditionalLog from facility_profile.models import Facility @@ -30,6 +31,7 @@ from morango.sync.stream.deserialize import AppModelDeserialize from morango.sync.stream.deserialize import AppModelValidate from morango.sync.stream.deserialize import delete_app_models +from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeSink from morango.sync.stream.deserialize import DeserializeTask @@ -1331,6 +1333,17 @@ def _app_user(self, username="user"): self._store(user, dirty_bit=False) return user + def _app_facility(self, name, parent=None): + facility = Facility(name=name, parent=parent) + facility.save(update_dirty_bit_to=False) + self._store( + facility, + dirty_bit=False, + _self_ref_fk=facility.parent_id or "", + _self_ref_order=0 if parent is None else 1, + ) + return facility + class DeserializeSinkTestCase(DeserializeSinkFixturesMixin, TestCase): def test_save(self): @@ -1512,17 +1525,6 @@ def test_consume__batched(self): class DeserializeSinkGuardsTestCase(DeserializeSinkFixturesMixin, TestCase): """Saves whose FK target the sink failed or removed after the save was validated""" - def _app_facility(self, name, parent=None): - facility = Facility(name=name, parent=parent) - facility.save(update_dirty_bit_to=False) - self._store( - facility, - dirty_bit=False, - _self_ref_fk=facility.parent_id or "", - _self_ref_order=0 if parent is None else 1, - ) - return facility - def _error(self, store_id): return Store.objects.get(id=store_id).deserialization_error @@ -1557,55 +1559,144 @@ def test_failed_target__cross_model(self): self.assertIn(user.id, self._error(log.id)) self.assertIn(log.id, sink.failed_ids) - def test_failed_save(self): - """The log was validated against the user before the user's save failed""" + +class DeserializeFromStoreTestCase(DeserializeSinkFixturesMixin, TransactionTestCase): + """ + Runs the whole pipeline in its own transaction, which commits at the end of each run. FK + constraints are deferred, so a save that breaks one would fail the commit. + """ + + def setUp(self): + patcher = mock.patch.object(Facility, "clean_dereferences_parent", False) + patcher.start() + self.addCleanup(patcher.stop) + + def _deserialize(self, **kwargs): + with self.assertLogs(self.logger_name, "INFO") as logs: + deserialize_from_store(self.profile, **kwargs) + return logs + + def test_failed_parent_save__cross_model(self): self._app_user(username="same") user = self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) log = self._summary_log(user.id) - with self.assertLogs(self.logger_name, "WARNING"): - self._run() + self._deserialize() self._assert_store(user.id, dirty_bit=True, exception="django.db.utils.IntegrityError") - self.assertFalse(SummaryLog.objects.exists()) self._assert_store( log.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" ) - self.assertIn("failed to deserialize", self._error(log.id)) + self.assertFalse(MyUser.objects.filter(id=user.id).exists()) + self.assertFalse(SummaryLog.objects.exists()) + + def test_failed_parent_save__self_ref(self): + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + save = Facility.save + + def failing_save(facility, *args, **kwargs): + if facility.id == root.id: + raise IntegrityError("root failed") + return save(facility, *args, **kwargs) + + with mock.patch.object(Facility, "save", failing_save): + self._deserialize() + + self.assertFalse(Facility.objects.exists()) + self._assert_store(root.id, dirty_bit=True, exception="django.db.utils.IntegrityError") + self._assert_store(child.id, dirty_bit=True, exception=MorangoDirtyParent.path()) + self._assert_store(grandchild.id, dirty_bit=True, exception=MorangoDirtyParent.path()) def _cascade_fixture(self): - """A deleted root, its clean child, and a dirty grandchild validated against the child""" + """A deleted root, its clean child, and a dirty grandchild""" root = self._app_facility("root") child = self._app_facility("child", parent=root) Store.objects.filter(id=root.id).update(deleted=True, dirty_bit=True) grandchild = self._facility(parent=child, order=2) - tasks = {task.id: task for task in self._transform()} - self.assertEqual(tasks[grandchild.id].outcome, DeserializeOutcome.SAVE) - return root, child, grandchild, tasks + return root, child, grandchild + + def _assert_cascade(self, root, grandchild): + self._deserialize() - def _assert_cascaded(self, child, grandchild): self.assertFalse(Facility.objects.exists()) + self._assert_store(root.id, dirty_bit=False) self._assert_store( grandchild.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" ) - self.assertIn("was deleted", self._error(grandchild.id)) - self.assertIn(child.id, self._error(grandchild.id)) + self.assertIn("was deleted", Store.objects.get(id=grandchild.id).deserialization_error) - def test_cascaded_target__same_chunk(self): - root, child, grandchild, tasks = self._cascade_fixture() - sink = DeserializeSink() + # the next run fails the grandchild in validation, since the child's row is gone + self._deserialize() - with self.assertLogs(self.logger_name, "WARNING"): - sink.consume([tasks[root.id], tasks[grandchild.id]]) + self._assert_store( + grandchild.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertNotIn("was deleted", Store.objects.get(id=grandchild.id).deserialization_error) - self._assert_cascaded(child, grandchild) + def test_cascaded_target__same_chunk(self): + root, _, grandchild = self._cascade_fixture() + self._assert_cascade(root, grandchild) + @mock.patch("morango.sync.stream.deserialize.CHUNK_SIZE", 1) def test_cascaded_target__later_chunk(self): - root, child, grandchild, tasks = self._cascade_fixture() - sink = DeserializeSink() + """The grandchild is validated before the root's chunk is written""" + root, _, grandchild = self._cascade_fixture() + self._assert_cascade(root, grandchild) - sink.consume([tasks[root.id]]) - with self.assertLogs(self.logger_name, "WARNING"): - sink.consume([tasks[grandchild.id]]) + def test_self_ref_chain(self): + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + self._deserialize() + + self.assertEqual(Facility.objects.count(), 3) + for facility in (root, child, grandchild): + self._assert_store(facility.id, dirty_bit=False) + + def test_self_ref_chain__dereferences_parent(self): + """ + A child can't load a parent that is in the same chunk, so each run saves one level. This + characterizes a known limitation of the sink lag, not a contract, so a fix may change it. + """ + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + with mock.patch.object(Facility, "clean_dereferences_parent", True): + for saved in ([root], [root, child], [root, child, grandchild]): + self._deserialize() + self.assertEqual( + set(Facility.objects.values_list("id", flat=True)), + {facility.id for facility in saved}, + ) + + self.assertFalse(Store.objects.filter(dirty_bit=True).exists()) + + def test_skip_errored(self): + user = self._user(deserialization_error="old", deserialization_exception="old.Error") + + self._deserialize(skip_errored=True) + + self._assert_store(user.id, dirty_bit=True, exception="old.Error") + self.assertFalse(MyUser.objects.exists()) + + def test_sync_filter(self): + included = self._user() + excluded = self._user() + + self._deserialize(sync_filter=Filter(included.partition)) + + self.assertEqual(list(MyUser.objects.values_list("id", flat=True)), [included.id]) + self._assert_store(excluded.id, dirty_bit=True) + + def test_summary_logged(self): + self._user() + self._facility(serialized="{bad") + + logs = self._deserialize() - self._assert_cascaded(child, grandchild) + self.assertIn("'save': 1", logs.output[-1]) + self.assertIn("'error': 1", logs.output[-1]) From aa9a6b718e3773bb16f8928574ce12f14ccb9ce1 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:29:29 -0700 Subject: [PATCH 06/12] Freeze legacy deserialization parity as literal expectations Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tests/sync/stream/test_deserialize.py | 102 ++++++++++-------- 1 file changed, 59 insertions(+), 43 deletions(-) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 675e8b67..f29a1b13 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -39,7 +39,6 @@ from morango.sync.stream.deserialize import row_key from morango.sync.stream.deserialize import store_model_partition_fn from morango.sync.stream.deserialize import StoreModelSource -from morango.utils import exception_path from ...helpers import StoreFactory @@ -1050,112 +1049,126 @@ def test_transforms_only_read(self): class LegacyParityTestCase(DeserializeFixturesMixin, TestCase): """ - Characterizes legacy `_deserialize_from_store` errors and propagated deletions against the - transforms, on the same store records. The transforms don't write, so both can run against the - same database state. Runs with the default `Facility.clean_dereferences_parent`, which is the - legacy test behavior, unless a test patches it. + Characterizes the end state that legacy `_deserialize_from_store` stores for errors and + propagated deletions. The expected values were captured from legacy, with the default + `Facility.clean_dereferences_parent` at the time patched in where a test relies on it. """ - def _legacy_result(self, store_id): - """The legacy result of a store record that did not save, in the transforms' terms""" - if HardDeletedModels.objects.filter(id=store_id).exists(): - return DeserializeOutcome.PROPAGATE_HARD_DELETE, None - if DeletedModels.objects.filter(id=store_id).exists(): - return DeserializeOutcome.PROPAGATE_DELETE, None - exception = Store.objects.get(id=store_id).deserialization_exception - self.assertIsNotNone(exception) - return DeserializeOutcome.ERROR, exception - - def _assert_parity(self, *store_ids): - results = self._run() + def _deserialize(self): _deserialize_from_store(self.profile) - for store_id in store_ids: - with self.subTest(store_id=store_id): - outcome, error_type = results[store_id] - error_path = exception_path(error_type) if error_type else None - self.assertEqual((outcome, error_path), self._legacy_result(store_id)) + def _assert_stored(self, store_id, dirty_bit, exception, deleted=False, hard_deleted=False): + store = Store.objects.get(id=store_id) + self.assertEqual( + ( + store.dirty_bit, + store.deserialization_exception, + DeletedModels.objects.filter(id=store_id).exists(), + HardDeletedModels.objects.filter(id=store_id).exists(), + ), + (dirty_bit, exception, deleted, hard_deleted), + ) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_failed(self): root = self._facility(serialized="{bad") child = self._facility(parent=root, order=1) grandchild = self._facility(parent=child, order=2) - self._assert_parity(root.id, child.id, grandchild.id) + self._deserialize() + + self._assert_stored(root.id, True, "json.decoder.JSONDecodeError") + self._assert_stored(child.id, True, MorangoDirtyParent.path()) + self._assert_stored(grandchild.id, True, MorangoDirtyParent.path()) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_missing(self): missing_parent = Facility(id=uuid.uuid4().hex, name="missing") child = self._facility(parent=missing_parent) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, True, MorangoMissingParent.path()) def test_target_missing(self): log = self._summary_log(uuid.uuid4().hex) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, "django.db.utils.IntegrityError") @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_own_error_with_parent_accepted(self): root = self._facility() child = self._facility(parent=root, order=1, name="x" * 200) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, True, "django.core.exceptions.ValidationError") def test_target_deleted_in_run(self): user = self._user(deleted=True) log = self._summary_log(user.id) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, None, deleted=True) def test_target_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True) log = self._summary_log(user.id) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, None, deleted=True) def test_target_hard_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) log = self._summary_log(user.id) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, None, deleted=True, hard_deleted=True) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_deleted_earlier(self): root = self._facility(dirty_bit=False, deleted=True) child = self._facility(parent=root, order=1) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, False, None, deleted=True) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_hard_deleted_earlier(self): root = self._facility(dirty_bit=False, deleted=True, hard_deleted=True) child = self._facility(parent=root, order=1) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, False, None, deleted=True, hard_deleted=True) @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_parent_deleted_earlier__parent_not_loaded(self): root = self._facility(dirty_bit=False, deleted=True) child = self._facility(parent=root, order=1) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, True, "django.core.exceptions.ValidationError") - def test_dirty_parent_not_in_run__differs(self): + def test_dirty_parent_not_in_run(self): """ - Characterizes an intentional difference: legacy never deserializes children of a parent - whose store record is dirty. The transforms save the child, which validates against the - parent's older app row. + Legacy never deserializes children of a parent whose store record is dirty, and records + `MorangoMissingParent` when the parent is left out of the run. """ parent = Facility.objects.create(name="parent") self._store(parent, _self_ref_order=0, deserialization_error="failed earlier") child = self._facility(parent=parent, order=1) - results = self._run(skip_errored=True) _deserialize_from_store(self.profile, skip_erroring=True) - self.assertEqual(results[child.id], (DeserializeOutcome.SAVE, None)) - self.assertEqual( - self._legacy_result(child.id), - (DeserializeOutcome.ERROR, exception_path(MorangoMissingParent)), - ) + self._assert_stored(child.id, True, MorangoMissingParent.path()) @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_parent_deleted_in_run__parent_not_loaded(self): @@ -1163,7 +1176,10 @@ def test_parent_deleted_in_run__parent_not_loaded(self): child = self._facility(parent=root, order=1) grandchild = self._facility(parent=child, order=2) - self._assert_parity(child.id, grandchild.id) + self._deserialize() + + self._assert_stored(child.id, True, "django.core.exceptions.ValidationError") + self._assert_stored(grandchild.id, True, MorangoDirtyParent.path()) class DeleteAppModelsTestCase(TestCase): From 07e82262da91c1024f58eb1c5768266737e7a6bf Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:35:06 -0700 Subject: [PATCH 07/12] Switch deserialization callers to the streaming pipeline Defaults the test Facility to not load its parent during validation, like models that don't load related objects. With sink lag, a dirty chain of models that do takes one run per level, so the test that deserializes a whole chain at once relies on the new default. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/controller.py | 6 ++-- morango/sync/operations.py | 5 ++-- tests/testapp/facility_profile/models.py | 5 ++-- .../tests/sync/stream/test_deserialize.py | 29 ++++++++++++------- tests/testapp/tests/sync/test_controller.py | 2 +- tests/testapp/tests/sync/test_operations.py | 20 ++++++------- 6 files changed, 39 insertions(+), 28 deletions(-) diff --git a/morango/sync/controller.py b/morango/sync/controller.py index 5fe180a8..f0e3f911 100644 --- a/morango/sync/controller.py +++ b/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 diff --git a/morango/sync/operations.py b/morango/sync/operations.py index f8452443..1d4a3e2b 100644 --- a/morango/sync/operations.py +++ b/morango/sync/operations.py @@ -56,6 +56,7 @@ from morango.sync.context import LocalSessionContext from morango.sync.context import NetworkSessionContext from morango.sync.db import begin_transaction +from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.stream.serialize import serialize_into_store from morango.sync.utils import mute_signals from morango.sync.utils import validate_and_create_buffer_data @@ -1166,8 +1167,8 @@ def handle(self, context): serialize_into_store( context.sync_session.profile, sync_filter=context.filter ) - _deserialize_from_store( - context.sync_session.profile, filter=context.filter + deserialize_from_store( + context.sync_session.profile, sync_filter=context.filter ) except OperationalError as e: # if we run into a transaction isolation error, we return a pending status to force diff --git a/tests/testapp/facility_profile/models.py b/tests/testapp/facility_profile/models.py index 37457471..706f58af 100644 --- a/tests/testapp/facility_profile/models.py +++ b/tests/testapp/facility_profile/models.py @@ -25,8 +25,9 @@ class Facility(FacilityDataSyncableModel): morango_model_name = "facility" # when True, clean_fields loads the parent object, which raises ObjectDoesNotExist if the - # parent row is not in the app table; used to test non-validation error handling - clean_dereferences_parent = True + # parent row is not in the app table; used to test non-validation error handling. Defaults to + # False like models that don't load related objects, and tests patch it to cover those that do + clean_dereferences_parent = False name = models.CharField(max_length=100) now_date = models.DateTimeField(default=timezone.now) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index f29a1b13..dfc9f96c 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -25,7 +25,6 @@ from morango.models.core import HardDeletedModels from morango.models.core import Store from morango.models.core import SyncableModel -from morango.sync.operations import _deserialize_from_store from morango.sync.stream.core import Buffer from morango.sync.stream.core import Sink from morango.sync.stream.deserialize import AppModelDeserialize @@ -1047,15 +1046,21 @@ def test_transforms_only_read(self): ) -class LegacyParityTestCase(DeserializeFixturesMixin, TestCase): +class DeserializeParityTestCase(DeserializeFixturesMixin, TestCase): """ - Characterizes the end state that legacy `_deserialize_from_store` stores for errors and - propagated deletions. The expected values were captured from legacy, with the default + Checks the end state that `deserialize_from_store` stores for errors and propagated deletions + against the values captured from legacy `_deserialize_from_store`, with the default `Facility.clean_dereferences_parent` at the time patched in where a test relies on it. + + The exceptions are: + - A record that propagates the deletion of its FK target, which legacy left dirty. It now ends + clean, since its deletion is tracked. + - A child of a parent that is dirty but left out of the run, which legacy failed with + `MorangoMissingParent`. It is now saved against the parent's older app row. """ def _deserialize(self): - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) def _assert_stored(self, store_id, dirty_bit, exception, deleted=False, hard_deleted=False): store = Store.objects.get(id=store_id) @@ -1112,7 +1117,7 @@ def test_target_deleted_in_run(self): self._deserialize() - self._assert_stored(log.id, True, None, deleted=True) + self._assert_stored(log.id, False, None, deleted=True) def test_target_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True) @@ -1120,7 +1125,7 @@ def test_target_deleted_earlier(self): self._deserialize() - self._assert_stored(log.id, True, None, deleted=True) + self._assert_stored(log.id, False, None, deleted=True) def test_target_hard_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) @@ -1128,7 +1133,7 @@ def test_target_hard_deleted_earlier(self): self._deserialize() - self._assert_stored(log.id, True, None, deleted=True, hard_deleted=True) + self._assert_stored(log.id, False, None, deleted=True, hard_deleted=True) @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_deleted_earlier(self): @@ -1160,15 +1165,17 @@ def test_parent_deleted_earlier__parent_not_loaded(self): def test_dirty_parent_not_in_run(self): """ Legacy never deserializes children of a parent whose store record is dirty, and records - `MorangoMissingParent` when the parent is left out of the run. + `MorangoMissingParent` when the parent is left out of the run. The child is now saved, + since it validates against the parent's older app row. """ parent = Facility.objects.create(name="parent") self._store(parent, _self_ref_order=0, deserialization_error="failed earlier") child = self._facility(parent=parent, order=1) - _deserialize_from_store(self.profile, skip_erroring=True) + deserialize_from_store(self.profile, skip_errored=True) - self._assert_stored(child.id, True, MorangoMissingParent.path()) + self.assertTrue(Facility.objects.filter(id=child.id).exists()) + self._assert_stored(child.id, False, None) @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_parent_deleted_in_run__parent_not_loaded(self): diff --git a/tests/testapp/tests/sync/test_controller.py b/tests/testapp/tests/sync/test_controller.py index 1de23018..f436a84b 100644 --- a/tests/testapp/tests/sync/test_controller.py +++ b/tests/testapp/tests/sync/test_controller.py @@ -614,7 +614,7 @@ 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: + with self.assertLogs("morango.sync.stream.deserialize", level="WARNING") as cm: self.mc.deserialize_from_store() unique_warnings = [ diff --git a/tests/testapp/tests/sync/test_operations.py b/tests/testapp/tests/sync/test_operations.py index 2eb264d0..979c32c2 100644 --- a/tests/testapp/tests/sync/test_operations.py +++ b/tests/testapp/tests/sync/test_operations.py @@ -30,7 +30,6 @@ from morango.sync.controller import MorangoProfileController from morango.sync.controller import SessionController from morango.sync.operations import _dequeue_into_store -from morango.sync.operations import _deserialize_from_store from morango.sync.operations import _queue_into_buffer_v1 from morango.sync.operations import _queue_into_buffer_v2 from morango.sync.operations import _update_legacy_self_ref_order @@ -41,6 +40,7 @@ from morango.sync.operations import ReceiverDequeueOperation from morango.sync.operations import ReceiverDeserializeOperation from morango.sync.operations import ReceiverQueueOperation +from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.syncsession import TransferClient from ..helpers import create_buffer_and_store_dummy_data @@ -1257,7 +1257,7 @@ def test_successful_deserialization(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization() @@ -1267,7 +1267,7 @@ def test_deserialization_with_missing_username(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization( user_deserialized=False, @@ -1282,7 +1282,7 @@ def test_deserialization_with_excessively_long_username(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization( user_deserialized=False, @@ -1297,7 +1297,7 @@ def test_deserialization_with_invalid_content_id(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization(log1_deserialized=False) @@ -1307,14 +1307,14 @@ def test_deserialization_with_invalid_log_user_id(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization(log1_deserialized=False) def test_deserialization__conditional__no_filter(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization() conditional_log = ConditionalLog.objects.get(pk=self.serialized_conditional["id"]) @@ -1323,7 +1323,7 @@ def test_deserialization__conditional__no_filter(self): def test_deserialization__conditional__with_filter(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile, filter=Filter(self.serialized_facility["id"])) + deserialize_from_store(self.profile, sync_filter=Filter(self.serialized_facility["id"])) self.assert_deserialization( user_deserialized=False, @@ -1341,7 +1341,7 @@ def test_deserialization__skip_erroring(self): ) conditional_log_store.save() - _deserialize_from_store(self.profile, skip_erroring=True) + deserialize_from_store(self.profile, skip_errored=True) self.assert_deserialization( conditional_deserialized=False, @@ -1353,6 +1353,6 @@ def test_deserialization__skip_erroring__empty_string(self): conditional_log_store.deserialization_error = "" conditional_log_store.save() - _deserialize_from_store(self.profile, skip_erroring=True) + deserialize_from_store(self.profile, skip_errored=True) self.assert_deserialization() From f2971c8ab85bf4055b978f05e2cc03d0bcf3778b Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:38:39 -0700 Subject: [PATCH 08/12] Remove legacy deserialization from operations Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/operations.py | 381 +------------------------------------ 1 file changed, 1 insertion(+), 380 deletions(-) diff --git a/morango/sync/operations.py b/morango/sync/operations.py index 1d4a3e2b..70da04d7 100644 --- a/morango/sync/operations.py +++ b/morango/sync/operations.py @@ -1,24 +1,13 @@ -import functools import itertools import json import logging import uuid -from collections import defaultdict -from django.core import exceptions from django.db import connection -from django.db import transaction -from django.db.models import CharField -from django.db.models import F from django.db.models import Exists from django.db.models import OuterRef -from django.db.models import Q from django.db.models import Subquery -from django.db.models import signals -from django.db.models import Value -from django.db.models.fields import BooleanField -from django.db.models.functions import NullIf, Cast -from django.db.utils import IntegrityError +from django.db.models.functions import Cast from django.db.utils import OperationalError from django.utils import timezone from rest_framework.exceptions import ValidationError @@ -29,17 +18,13 @@ from morango.constants.capabilities import ASYNC_OPERATIONS from morango.constants.capabilities import FSIC_V2_FORMAT from morango.constants.capabilities import SELF_REF_ORDER -from morango.errors import MorangoDirtyParent from morango.errors import MorangoInvalidFSICPartition from morango.errors import MorangoLimitExceeded -from morango.errors import MorangoMissingParent from morango.errors import MorangoResumeSyncError from morango.errors import MorangoSkipOperation from morango.models.certificates import Filter from morango.models.core import Buffer from morango.models.core import DatabaseMaxCounter -from morango.models.core import DeletedModels -from morango.models.core import HardDeletedModels from morango.models.core import InstanceIDModel from morango.models.core import RecordMaxCounter from morango.models.core import RecordMaxCounterBuffer @@ -52,17 +37,14 @@ from morango.models.fsic_utils import expand_fsic_for_use from morango.registry import syncable_models from morango.sync.backends.utils import load_backend -from morango.sync.backends.utils import TemporaryTable from morango.sync.context import LocalSessionContext from morango.sync.context import NetworkSessionContext from morango.sync.db import begin_transaction from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.stream.serialize import serialize_into_store -from morango.sync.utils import mute_signals from morango.sync.utils import validate_and_create_buffer_data from morango.utils import _assert from morango.utils import self_referential_fk -from morango.utils import exception_path from morango.utils import SETTINGS logger = logging.getLogger(__name__) @@ -92,367 +74,6 @@ def _join_with_logical_operator(lst, operator): return "(({items}))".format(items=op.join(lst)) -def _validate_missing_store_foreign_keys(from_model_name, to_model_name, temp_table): - """ - Performs validation on a bulk set of foreign keys (FKs), given a temp table with two columns, - `from_pk` and `to_pk`, the primary key (PK) pair to validate. Any store record matching - `from_pk`, while missing a store record matching `to_pk`, will be updated with a deserialization - error and its PK returned within a list. - - :param from_model_name: A str name of the model which has the FK, for logging purposes - :param to_model_name: A str name of the model referenced by the FK, for logging purposes - :param temp_table: A temp table object for querying against in the DB - :type temp_table: morango.sync.backends.utils.TemporaryTable - :return: A list of store PKs that have broken FKs - """ - invalid_pks = [] - select_sql = """ - SELECT t.from_field, t.from_pk, t.to_pk - FROM {temp_table} t - WHERE NOT EXISTS ( - SELECT 1 - FROM {store} s - WHERE s.{pk_field} = t.to_pk - ) - """.format( - temp_table=temp_table.sql_name, - store=Store._meta.db_table, - pk_field=Store._meta.pk.column, - ) - - store_deserialization_error_fields = [ - f for f in Store._meta.fields if f.name in ("deserialization_error", "deserialization_exception") - ] - store_update_fields = [Store._meta.pk, *store_deserialization_error_fields] - - from_pk_field = temp_table.get_field("from_pk") - to_pk_field = temp_table.get_field("to_pk") - update_values = [] - with connection.cursor() as c: - c.execute(select_sql) - for from_field, from_pk, to_pk in c.fetchall(): - err = IntegrityError( - "{from_model_name}.{from_field} references non-existent {to_model_name} instance with id '{to_pk}'".format( - from_model_name=from_model_name, - from_field=from_field, - to_model_name=to_model_name, - to_pk=to_pk_field.to_python(to_pk), - ) - ) - logger.warning( - "Error deserializing instance with id {from_pk}: {err}".format( - from_pk=from_pk, - err=str(err), - ) - ) - update_values.extend([from_pk, str(err), exception_path(err)]) - invalid_pks.append(from_pk_field.to_python(from_pk)) - if update_values: - # update Store with errors - DBBackend._bulk_update( - c, Store._meta.db_table, store_update_fields, update_values - ) - return invalid_pks - - -def _handle_deleted_store_foreign_keys(temp_table): - """ - Handles store records with foreign key (FK) references to other deleted store records to prevent - their deserialization and update the `DeletedModels` and `HardDeletedModels` tracking, given a - temp table with two columns, `from_pk` and `to_pk`, the primary key (PK) pair - :param temp_table: A temp table object for querying against in the DB - :type temp_table: morango.sync.backends.utils.TemporaryTable - :return: A list of store PKs that have FKs to deleted records - """ - select_sql = """ - SELECT t.from_pk, s.profile, s.deleted, s.hard_deleted - FROM {temp_table} t - INNER JOIN {store} s ON s.{pk_field} = t.to_pk - WHERE s.deleted OR s.hard_deleted - """.format( - temp_table=temp_table.sql_name, - store=Store._meta.db_table, - pk_field=Store._meta.pk.column, - ) - - from_pk_field = temp_table.get_field("from_pk") - deleted_pks = [] - deleted_values = [] - hard_deleted_values = [] - with connection.cursor() as c: - c.execute(select_sql) - # find all the store PKs which have FKs to deleted store PKs - for from_pk, profile, deleted, hard_deleted in c.fetchall(): - if deleted or hard_deleted: - deleted_values.extend([from_pk, profile]) - if hard_deleted: - hard_deleted_values.extend([from_pk, profile]) - deleted_pks.append(from_pk_field.to_python(from_pk)) - # update deleted tracking models - if deleted_values: - DBBackend._bulk_full_record_upsert( - c, - DeletedModels._meta.db_table, - DeletedModels._meta.fields, - deleted_values, - ) - if hard_deleted_values: - DBBackend._bulk_full_record_upsert( - c, - HardDeletedModels._meta.db_table, - HardDeletedModels._meta.fields, - hard_deleted_values, - ) - return deleted_pks - - -def _validate_store_foreign_keys(from_model_name, fk_references): - """ - Validates the foreign key (FK) references of a model whose name is `from_model_name`, through - bulk processing using a temporary table within the database for holding the FK references - - :param from_model_name: A str name of the model which has the FKs, for logging purposes - :param fk_references: A dictionary of lists containing `morango.models.core.ForeignKeyReference` - keyed by the - :return: A tuple of two lists containing store PKs with broken FKs and FKs to deleted records - """ - exclude_pks = [] - deleted_pks = [] - - for to_model_name, to_fk_references in fk_references.items(): - with TemporaryTable( - connection, - "fks", - from_field=CharField(max_length=255), - from_pk=UUIDField(), - to_pk=UUIDField(), - ) as temp_table: - # insert all the FK references into a temp table in the database - temp_table.bulk_insert([fks._asdict() for fks in to_fk_references]) - # now pass the temp table to validate against broken FKs - exclude_pks.extend( - _validate_missing_store_foreign_keys( - from_model_name, to_model_name, temp_table - ) - ) - # find any FKs referencing deleted records - deleted_pks.extend(_handle_deleted_store_foreign_keys(temp_table)) - - return exclude_pks, deleted_pks - - -def _save_deserialized_record(store_model, app_model, model_name, excluded_list=None): - """ - 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.unset_deserialization_error() - store_model.save( - update_fields=["dirty_bit", "deserialization_error", "deserialization_exception"] - ) - return True - except ( - exceptions.ValidationError, - exceptions.ObjectDoesNotExist, - ValueError, - IntegrityError, - ) as e: - if excluded_list is not None: - excluded_list.append(store_model.id) - store_model.set_deserialization_error(e) - store_model.save(update_fields=["deserialization_error", "deserialization_exception"]) - 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. 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. - """ - - fk_cache = {} - excluded_list = [] - deleted_list = [] - - with begin_transaction(filter, isolated=True): - # iterate through classes which are in foreign key dependency order - for model in syncable_models.get_models(profile): - deferred_fks = defaultdict(list) - store_models = Store.objects.filter(profile=profile) - - model_condition = Q(model_name=model.morango_model_name) - for klass in model.morango_model_dependencies: - model_condition |= Q(model_name=klass.morango_model_name) - - store_models = store_models.filter(model_condition) - - if filter: - # create Q objects for filtering by prefixes - prefix_condition = functools.reduce( - lambda x, y: x | y, - [Q(partition__startswith=prefix) for prefix in filter], - ) - store_models = store_models.filter(prefix_condition) - - # if requested, skip any records that previously errored, to be faster - if skip_erroring: - # previously, deserialization_error was not nullable, and set to empty string if - # there wasn't an error, so this nullifies empty strings to ensure the filter - # applies either way - store_models = ( - store_models - .annotate( - n_deserialization_error=NullIf( - F("deserialization_error"), - Value(""), - output_field=BooleanField() - ) - ) - .filter(n_deserialization_error__isnull=True) - ) - - # handle cases where a class has a single FK reference to itself - if self_referential_fk(model): - clean_parents = store_models.filter(dirty_bit=False).char_ids_list() - dirty_children = ( - store_models.filter(dirty_bit=True) - # handle parents or if the model has no parent - .filter(Q(_self_ref_fk__in=clean_parents) | Q(_self_ref_fk="")) - ) - - # keep iterating until size of dirty_children is 0 - while len(dirty_children) > 0: - for store_model in dirty_children: - try: - app_model, _ = store_model._deserialize_store_model( - fk_cache, sync_filter=filter - ) - except ( - exceptions.ValidationError, - exceptions.ObjectDoesNotExist, - ValueError, - ) as e: - excluded_list.append(store_model.id) - store_model.set_deserialization_error(e) - store_model.save(update_fields=["deserialization_error", "deserialization_exception"]) - continue - if app_model: - _save_deserialized_record( - store_model, app_model, model.__name__, excluded_list - ) - else: - store_model.dirty_bit = False - store_model.unset_deserialization_error() - store_model.save( - update_fields=["dirty_bit", "deserialization_error", "deserialization_exception"] - ) - - # update lists with new clean parents and dirty children - clean_parents = store_models.filter(dirty_bit=False).char_ids_list() - dirty_children = store_models.filter( - dirty_bit=True, _self_ref_fk__in=clean_parents - ).exclude(id__in=excluded_list) - - # A. Mark records that were skipped due to missing parents with error info - # A(i). The ones that have a parent Store entry but it's dirty - dirty_parents = store_models.filter(dirty_bit=True).char_ids_list() - store_models.filter( - dirty_bit=True, _self_ref_fk__in=dirty_parents - ).exclude(id__in=excluded_list).update( - deserialization_error="Parent is dirty; could not deserialize.", - deserialization_exception=MorangoDirtyParent.path(), - ) - # A(ii). The ones that don't even have Store entries for parent at all - all_parents = store_models.char_ids_list() - store_models.filter(dirty_bit=True).exclude( - _self_ref_fk__in=all_parents - ).exclude(id__in=excluded_list).update( - deserialization_error="Parent does not exist in Store; could not deserialize.", - deserialization_exception=MorangoMissingParent.path(), - ) - - else: - # collect all initially valid app models and validate their FKs - app_models = [] - for store_model in store_models.filter(dirty_bit=True): - try: - ( - app_model, - model_deferred_fks, - ) = store_model._deserialize_store_model( - fk_cache, - defer_fks=True, - sync_filter=filter, - ) - if 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 - # deserialize for reasons other than broken FKs at this point - for fk_ref in fk_refs: - if fk_ref.to_pk in excluded_list: - raise exceptions.ValidationError( - "{} with id {} failed to deserialize".format( - fk_model, fk_ref.to_pk - ) - ) - deferred_fks[fk_model].extend(fk_refs) - except ( - exceptions.ValidationError, - exceptions.ObjectDoesNotExist, - ValueError, - ) as e: - # if the app model did not validate, we leave the store dirty bit set - excluded_list.append(store_model.id) - store_model.set_deserialization_error(e) - store_model.save(update_fields=["deserialization_error", "deserialization_exception"]) - - # validate app model FKs - model_excluded_pks, model_deleted_pks = _validate_store_foreign_keys( - model.__name__, deferred_fks - ) - excluded_list.extend(model_excluded_pks) - deleted_list.extend(model_deleted_pks) - - # 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 in excluded_list - or app_model.pk in deleted_list - ): - continue - _save_deserialized_record( - store_model, app_model, model.__name__ - ) - - def _queue_into_buffer_v1(transfersession): """ Takes a chunk of data from the store to be put into the buffer to be sent to another morango instance. This is the legacy From 15d9b5d67f893f440078dea70c929eabc817a4b1 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:47:21 -0700 Subject: [PATCH 09/12] Replace Store._deserialize_store_model with deserialize_store_model Moves single-record deserialization to a standalone function on the streaming transforms and write helpers, so it can still be called outside of a pipeline run. It returns the validated app model, or None for a deleted record. The defer_fks option is dropped, since only the removed legacy bulk path used the deferred FK references. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/models/core.py | 79 ------------------- morango/sync/stream/deserialize.py | 34 ++++++++ .../tests/sync/stream/test_deserialize.py | 52 ++++++++++++ 3 files changed, 86 insertions(+), 79 deletions(-) diff --git a/morango/models/core.py b/morango/models/core.py index 76a46c77..319e242e 100644 --- a/morango/models/core.py +++ b/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): """ diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 84dedc98..c81f8896 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -566,6 +566,40 @@ def store_model_partition_fn(task: DeserializeTask) -> str: return task.store.model_name +def deserialize_store_model( + store: Store, fk_cache: Dict, sync_filter: Optional[Filter] = None +) -> Optional[SyncableModel]: + """ + Deserializes and validates a single store record outside of a pipeline run. Deleted records are + applied: a deleted record's app model is deleted, and a record whose FK target was deleted is + tracked as deleted. Neither the store record nor a valid app model is saved. + + :param store: The store record to deserialize + :param fk_cache: Dictionary cache for FK references, shared across calls + :param sync_filter: The Filter object for the current sync, if any + :return: The validated app model, or None if the record was deleted + :raises Exception: The error that the record failed with, after logging it + """ + task = DeserializeTask(store, fk_cache) + AppModelDeserialize(sync_filter=sync_filter).transform(task) + AppModelValidate(sync_filter=sync_filter).transform(task) + + outcome = task.outcome + if outcome == DeserializeOutcome.ERROR: + error = task.errors[0] + log_deserialization_error(task, error) + raise error + if outcome == DeserializeOutcome.DELETE: + # without sink lag, a later call validates against the app tables where the cascaded rows + # are already gone, so the cascaded keys aren't needed + delete_app_models(task.model, [task.id]) + return None + if outcome in (DeserializeOutcome.PROPAGATE_DELETE, DeserializeOutcome.PROPAGATE_HARD_DELETE): + record_propagated_deletions([task]) + return None + return task.app_model + + CHUNK_SIZE = 500 diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index dfc9f96c..2202725f 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -31,6 +31,7 @@ from morango.sync.stream.deserialize import AppModelValidate from morango.sync.stream.deserialize import delete_app_models from morango.sync.stream.deserialize import deserialize_from_store +from morango.sync.stream.deserialize import deserialize_store_model from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeSink from morango.sync.stream.deserialize import DeserializeTask @@ -1723,3 +1724,54 @@ def test_summary_logged(self): self.assertIn("'save': 1", logs.output[-1]) self.assertIn("'error': 1", logs.output[-1]) + + +class DeserializeStoreModelTestCase(DeserializeSinkFixturesMixin, TestCase): + """Deserializes a single store record by hand, outside of a pipeline run""" + + def test_valid(self): + store = self._user() + + app_model = deserialize_store_model(store, {}) + + self.assertIsInstance(app_model, MyUser) + self.assertEqual(app_model.id, store.id) + self.assertFalse(MyUser.objects.exists()) + self.assertTrue(Store.objects.get(id=store.id).dirty_bit) + + def test_deleted(self): + user = self._app_user() + store = Store.objects.get(id=user.id) + store.deleted = True + + self.assertIsNone(deserialize_store_model(store, {})) + self.assertFalse(MyUser.objects.exists()) + + def test_target_deleted(self): + user = self._user(dirty_bit=False, deleted=True) + store = self._summary_log(user.id) + + self.assertIsNone(deserialize_store_model(store, {})) + self.assertTrue(DeletedModels.objects.filter(id=store.id).exists()) + self.assertTrue(Store.objects.get(id=store.id).dirty_bit) + + def test_invalid(self): + user = MyUser(id=uuid.uuid4().hex, username="x" * 256, password="password") + store = self._store(user) + + with self.assertLogs(self.logger_name, "WARNING") as logs: + with self.assertRaises(ValidationError): + deserialize_store_model(store, {}) + + self.assertIn(store.id, logs.output[0]) + + def test_fk_cache(self): + """The user is validated but never saved, so the log only validates through the cache""" + fk_cache = {} + user = self._user() + log = self._summary_log(user.id) + + deserialize_store_model(user, fk_cache) + app_model = deserialize_store_model(log, fk_cache) + + self.assertEqual(app_model.id, log.id) From 3b0ed258f8798a5467355a2b2dc5cab3ddaf9dae Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Mon, 19 Oct 2026 13:32:41 -0700 Subject: [PATCH 10/12] Add 0.9.0 release notes for streaming deserialization Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 109a935f..37d25c6d 100644 --- a/CHANGELOG.md +++ b/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` From 5a6fb03d6547c03439f4a543c7fa043e54d520d4 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Mon, 19 Oct 2026 16:34:28 -0700 Subject: [PATCH 11/12] Assert only the done message in the deserialization summary log test The sink's summary test already checks the counts, so the log test doesn't need to depend on the formatting of the summary dict. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/testapp/tests/sync/stream/test_deserialize.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 2202725f..bafca894 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -1722,8 +1722,7 @@ def test_summary_logged(self): logs = self._deserialize() - self.assertIn("'save': 1", logs.output[-1]) - self.assertIn("'error': 1", logs.output[-1]) + self.assertIn("Deserialization done", logs.output[-1]) class DeserializeStoreModelTestCase(DeserializeSinkFixturesMixin, TestCase): From 2c7cc1b255d1aea55ec045ff711a68f83a1e5681 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 20 Oct 2026 13:43:40 -0700 Subject: [PATCH 12/12] Stream sync filter partitions in the order the caller gives The source streams records one partition prefix at a time, and sorted the prefixes first. With a filter where an FK target's partition sorts after the partition of a record that references it, the record was validated before its target was deserialized, and failed. Kolibri already orders its filters so that targets come first, so the source now keeps that order and drops the partition_order option. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 4 +- morango/sync/stream/source.py | 24 ++++-------- .../tests/sync/stream/test_deserialize.py | 39 +++++++++---------- 3 files changed, 28 insertions(+), 39 deletions(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index c81f8896..6aee7a21 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -133,7 +133,6 @@ def __init__( profile: str, sync_filter: Optional[Filter] = None, dirty_only: bool = True, - partition_order: str = "asc", fk_cache: Optional[Dict] = None, skip_errored: bool = False, ): @@ -141,11 +140,10 @@ def __init__( :param profile: The Morango model profile :param sync_filter: The Filter object for this sync :param dirty_only: Whether to filter on dirty records only - :param partition_order: Controls how the filter specificity is applied, "asc" or "desc" :param fk_cache: Dictionary cache for FK references :param skip_errored: Whether to skip Store records with deserialization errors """ - super().__init__(profile, sync_filter, dirty_only, partition_order) + super().__init__(profile, sync_filter, dirty_only) self.fk_cache = fk_cache if fk_cache is not None else {} self.skip_errored = skip_errored diff --git a/morango/sync/stream/source.py b/morango/sync/stream/source.py index 85940a58..2a0057f7 100644 --- a/morango/sync/stream/source.py +++ b/morango/sync/stream/source.py @@ -27,25 +27,22 @@ class MorangoSource(Source[T], abc.ABC): Common source functionality for Morango sources, such as SyncableModels and Store records. """ - __slots__ = ("profile", "sync_filter", "dirty_only", "partition_order", "_seen") + __slots__ = ("profile", "sync_filter", "dirty_only", "_seen") def __init__( self, profile: str, sync_filter: Optional[Filter] = None, dirty_only: bool = True, - partition_order: str = "asc", ): """ :param profile: The Morango model profile :param sync_filter: The Filter object for this sync :param dirty_only: Whether to filter on dirty records only - :param partition_order: Controls how the filter specificity is applied, "asc" or "desc" """ self.profile = profile self.sync_filter = sync_filter self.dirty_only = dirty_only - self.partition_order = partition_order self._seen: Optional[set] = None def begin(self) -> None: @@ -54,11 +51,12 @@ def begin(self) -> None: def prefix_conditions(self) -> Generator[Optional[str], None, None]: """ - Generates partition prefixes for queries based on the sync filter and partition order. + Generates partition prefixes for queries based on the sync filter. - This method outputs prefixes in sorted order according to the specified partition - order. If no sync filter is provided, it yields `None` to indicate a query - without filtering by partition. + This method outputs prefixes in the order of the sync filter. Records are streamed one + prefix at a time, so the filter must list the partitions of FK targets before the + partitions of the records that reference them. If no sync filter is provided, it yields + `None` to indicate a query without filtering by partition. :return: A generator yielding partition prefixes or `None` if no filtering is applied. """ @@ -66,14 +64,8 @@ def prefix_conditions(self) -> Generator[Optional[str], None, None]: # yield None once, so we do one query without a partition filter (everything) yield None else: - partitions_prefixes = [str(prefix) for prefix in self.sync_filter] - partition_iterator = sorted( - partitions_prefixes, - reverse=self.partition_order == "desc", - ) - - for prefix in partition_iterator: - yield prefix + for prefix in self.sync_filter: + yield str(prefix) def stream(self) -> Generator[T, None, None]: """ diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index bafca894..e43336ef 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -600,31 +600,16 @@ def test_success_queries_store_never(self): class StoreModelSourcePrefixConditionsTestCase(SimpleTestCase): - """`prefix_conditions` is pure ordering logic over the sync filter and touches no ORM""" + """`prefix_conditions` only reads the sync filter, and makes no queries""" def test_prefix_conditions__none(self): source = StoreModelSource(profile="test") self.assertEqual(list(source.prefix_conditions()), [None]) - def test_prefix_conditions__with_filter_asc(self): - source = StoreModelSource(profile="test", sync_filter=Filter("b\na"), partition_order="asc") - self.assertEqual(list(source.prefix_conditions()), ["a", "b"]) - - def test_prefix_conditions__with_filter_asc__realistic(self): - """ - Partitions often take the form of `{id}:{additional specificity}`, so the important aspect - of partition ordering is that the least specific filter is generally first - """ - source = StoreModelSource( - profile="test", sync_filter=Filter("a:test:z\na\na:initial"), partition_order="asc" - ) - self.assertEqual(list(source.prefix_conditions()), ["a", "a:initial", "a:test:z"]) - - def test_prefix_conditions__with_filter_desc(self): - source = StoreModelSource( - profile="test", sync_filter=Filter("a\nb"), partition_order="desc" - ) - self.assertEqual(list(source.prefix_conditions()), ["b", "a"]) + def test_prefix_conditions__with_filter(self): + """The prefixes keep the order of the sync filter, which the caller is responsible for""" + source = StoreModelSource(profile="test", sync_filter=Filter("b:user\na\na:initial")) + self.assertEqual(list(source.prefix_conditions()), ["b:user", "a", "a:initial"]) class StoreModelSourceBeginTestCase(SimpleTestCase): @@ -1716,6 +1701,20 @@ def test_sync_filter(self): self.assertEqual(list(MyUser.objects.values_list("id", flat=True)), [included.id]) self._assert_store(excluded.id, dirty_bit=True) + def test_sync_filter__partition_order(self): + """ + The source streams the partitions in the order of the sync filter, so a filter that lists + the user's partition first deserializes the user before a log in a later partition + """ + user = self._user(partition="b:user") + log = self._summary_log(user.id, partition="a:user:summary") + + self._deserialize(sync_filter=Filter("b:user\na:user:summary")) + + self._assert_store(user.id, dirty_bit=False) + self._assert_store(log.id, dirty_bit=False) + self.assertTrue(SummaryLog.objects.filter(id=log.id).exists()) + def test_summary_logged(self): self._user() self._facility(serialized="{bad")