From 253ba8f2308ebb7045d73823e59ac0fe0c280ed0 Mon Sep 17 00:00:00 2001 From: P SRIKAR NAIDU Date: Wed, 11 Nov 2026 19:44:35 +0530 Subject: [PATCH] Dequeue transferred records through an explicit database handle Pass the context database through dequeue SQL, counter updates, and legacy ordering. Restore routed default handles after context pickling. Keep FSIC helper routing for Task 6 (#367), as specified by Task 5. --- morango/sync/context.py | 8 ++ morango/sync/operations.py | 54 ++++---- tests/testapp/tests/sync/test_context.py | 19 ++- tests/testapp/tests/sync/test_operations.py | 145 +++++++++++++++++++- 4 files changed, 198 insertions(+), 28 deletions(-) diff --git a/morango/sync/context.py b/morango/sync/context.py index 3c4d72d3..4319cbd5 100644 --- a/morango/sync/context.py +++ b/morango/sync/context.py @@ -4,6 +4,7 @@ from morango.models.certificates import Filter from morango.models.core import SyncSession from morango.models.core import TransferSession +from morango.sync.db import SyncDatabase from morango.utils import CAPABILITIES from morango.utils import parse_capabilities_from_server_request @@ -20,6 +21,7 @@ class SessionContext(object): "is_push", "capabilities", "error", + "db", ) max_backoff_interval = 5 @@ -30,6 +32,7 @@ def __init__( sync_filter=None, is_push=None, capabilities=None, + db=None, ): """ :param sync_session: The sync session instance @@ -42,6 +45,8 @@ def __init__( :type is_push: bool :param capabilities: Capabilities set that is combined (intersected) with our own capabilities :type capabilities: set|None + :param db: Database handle for local sync operations, defaulting to routed access + :type db: SyncDatabase|None """ self.sync_session = sync_session self.transfer_session = transfer_session @@ -49,6 +54,7 @@ def __init__( self.is_push = is_push self.capabilities = set(capabilities or []) & CAPABILITIES self.error = None + self.db = db if db is not None else SyncDatabase.default() if self.transfer_session: self.sync_session = transfer_session.sync_session or self.sync_session @@ -179,6 +185,7 @@ def __getstate__(self): def __setstate__(self, state): """Re-apply dict state after serialization""" + self.db = SyncDatabase.default() sync_session_id = state.get("sync_session_id", None) if sync_session_id is not None: self.sync_session = SyncSession.objects.get(pk=sync_session_id) @@ -498,6 +505,7 @@ def __getstate__(self): def __setstate__(self, state): """Re-apply dict state after serialization""" + self.db = SyncDatabase.default() self.children = state.get("children", []) self._counter = state.get("counter", 0) self._stage = state.get("stage", None) diff --git a/morango/sync/operations.py b/morango/sync/operations.py index 65059c1b..86e78b4e 100644 --- a/morango/sync/operations.py +++ b/morango/sync/operations.py @@ -25,10 +25,10 @@ from morango.models.certificates import Filter from morango.models.core import Buffer from morango.models.core import DatabaseMaxCounter -from morango.models.core import InstanceIDModel from morango.models.core import RecordMaxCounter from morango.models.core import RecordMaxCounterBuffer from morango.models.core import Store +from morango.models.core import SyncSession from morango.models.core import TransferSession from morango.models.core import UUIDField from morango.models.fsic_utils import calculate_directional_fsic_diff @@ -332,13 +332,13 @@ def _queue_into_buffer_v2(transfersession, chunk_size=200): ) -def _update_legacy_self_ref_order_for_model(queryset): +def _update_legacy_self_ref_order_for_model(queryset, db): # root nodes set the _self_ref_order to 0 queryset.filter(_self_ref_fk="").exclude(_self_ref_order=0).update(_self_ref_order=0) # reset the _self_ref_order to None for all records that have a parent queryset.exclude(_self_ref_fk="").exclude(_self_ref_order=None).update(_self_ref_order=None) - parent = Store.objects.filter( + parent = db.manager(Store).filter( id=Cast(OuterRef("_self_ref_fk"), UUIDField()), _self_ref_order__isnull=False, ) @@ -349,9 +349,9 @@ def _update_legacy_self_ref_order_for_model(queryset): pass -def _update_legacy_self_ref_order(transfer_session): - profile = transfer_session.sync_session.profile - transferred_store_records = Store.objects.filter( +def _update_legacy_self_ref_order(transfer_session, db): + profile = db.manager(SyncSession).get(pk=transfer_session.sync_session_id).profile + transferred_store_records = db.manager(Store).filter( last_transfer_session_id=transfer_session.id, profile=profile, ) @@ -359,35 +359,42 @@ def _update_legacy_self_ref_order(transfer_session): for Model in syncable_models.get_models(profile): queryset = transferred_store_records.filter(model_name=Model.morango_model_name) if self_referential_fk(Model): - _update_legacy_self_ref_order_for_model(queryset) + _update_legacy_self_ref_order_for_model(queryset, db) else: queryset.exclude(_self_ref_order=None).update(_self_ref_order=None) -def _dequeue_into_store(transfer_session, fsic, v2_format=False, self_ref_order=True): +def _dequeue_into_store(transfer_session, fsic, v2_format, self_ref_order, db): """ Takes data from the buffers and merges into the store and record max counters. + :param transfer_session: Transfer session whose buffered records will be merged + :param fsic: JSON-encoded filter-specific instance counters from the sender + :param v2_format: Whether the counters use the v2 format + :param self_ref_order: Whether the sender supplied self-referential ordering + :param db: Database handle for the dequeue transaction, SQL, and instance counter + :type db: morango.sync.db.SyncDatabase + ALGORITHM: Incrementally insert and delete on a case by case basis to ensure subsequent cases are not affected by previous cases. """ - with begin_transaction(Filter(transfer_session.filter)): - with connection.cursor() as cursor: - DBBackend._dequeuing_delete_rmcb_records(cursor, transfer_session.id) - DBBackend._dequeuing_delete_buffered_records(cursor, transfer_session.id) - current_id = InstanceIDModel.get_current_instance_and_increment_counter() - DBBackend._dequeuing_merge_conflict_buffer(cursor, current_id, transfer_session.id) - DBBackend._dequeuing_merge_conflict_rmcb(cursor, transfer_session.id) - DBBackend._dequeuing_update_rmcs_last_saved_by(cursor, current_id, transfer_session.id) - DBBackend._dequeuing_delete_mc_rmcb(cursor, transfer_session.id) - DBBackend._dequeuing_delete_mc_buffer(cursor, transfer_session.id) - DBBackend._dequeuing_insert_remaining_buffer(cursor, transfer_session.id) - DBBackend._dequeuing_insert_remaining_rmcb(cursor, transfer_session.id) + with db.begin_transaction(Filter(transfer_session.filter)): + with db.cursor() as cursor: + db.backend._dequeuing_delete_rmcb_records(cursor, transfer_session.id) + db.backend._dequeuing_delete_buffered_records(cursor, transfer_session.id) + current_id = db.current_instance_and_increment_counter() + db.backend._dequeuing_merge_conflict_buffer(cursor, current_id, transfer_session.id) + db.backend._dequeuing_merge_conflict_rmcb(cursor, transfer_session.id) + db.backend._dequeuing_update_rmcs_last_saved_by(cursor, current_id, transfer_session.id) + db.backend._dequeuing_delete_mc_rmcb(cursor, transfer_session.id) + db.backend._dequeuing_delete_mc_buffer(cursor, transfer_session.id) + db.backend._dequeuing_insert_remaining_buffer(cursor, transfer_session.id) + db.backend._dequeuing_insert_remaining_rmcb(cursor, transfer_session.id) if not self_ref_order: - _update_legacy_self_ref_order(transfer_session) - DBBackend._dequeuing_delete_remaining_rmcb(cursor, transfer_session.id) - DBBackend._dequeuing_delete_remaining_buffer(cursor, transfer_session.id) + _update_legacy_self_ref_order(transfer_session, db) + db.backend._dequeuing_delete_remaining_rmcb(cursor, transfer_session.id) + db.backend._dequeuing_delete_remaining_buffer(cursor, transfer_session.id) DatabaseMaxCounter.update_fsics( json.loads(fsic), @@ -712,6 +719,7 @@ def handle(self, context): fsic, v2_format=FSIC_V2_FORMAT in context.capabilities, self_ref_order=SELF_REF_ORDER in context.capabilities, + db=context.db, ) return transfer_statuses.COMPLETED diff --git a/tests/testapp/tests/sync/test_context.py b/tests/testapp/tests/sync/test_context.py index f9941a19..fb708170 100644 --- a/tests/testapp/tests/sync/test_context.py +++ b/tests/testapp/tests/sync/test_context.py @@ -15,12 +15,18 @@ from morango.sync.context import NetworkSessionContext from morango.sync.context import SessionContext from morango.sync.controller import SessionController +from morango.sync.db import SyncDatabase from ..helpers import create_dummy_store_data from ..helpers import TestSessionContext class SessionContextTestCase(SimpleTestCase): + def test_database_handle(self): + self.assertIsNone(TestSessionContext().db.alias) + db = SyncDatabase("default2") + self.assertIs(TestSessionContext(db=db).db, db) + def test_init__nothing(self): context = TestSessionContext() context.update_state( @@ -259,6 +265,15 @@ def test_init(self): class ContextPicklingTestCase(TestCase): + def test_local_restores_default_database_handle(self): + context = LocalSessionContext(db=SyncDatabase("default2")) + self.assertNotIn("db", context.__getstate__()) + + restored = pickle.loads(pickle.dumps(context)) + + self.assertIsInstance(restored.db, SyncDatabase) + self.assertIsNone(restored.db.alias) + def test_basic(self): data = create_dummy_store_data() transfer_session = data["tx"] @@ -319,7 +334,7 @@ def test_composite(self): ) network = NetworkSessionContext(conn) - composite = CompositeSessionContext([local, network]) + composite = CompositeSessionContext([local, network], db=SyncDatabase("default2")) for _ in range(2): composite.update_state( stage=transfer_stages.INITIALIZING, stage_status=transfer_statuses.COMPLETED @@ -333,8 +348,10 @@ def test_composite(self): self.assertEqual(composite._counter, unpickled_context._counter) self.assertEqual(composite.stage, unpickled_context.stage) self.assertEqual(composite.stage_status, unpickled_context.stage_status) + self.assertIsNone(unpickled_context.db.alias) for i, context_type in enumerate([LocalSessionContext, NetworkSessionContext]): self.assertIsInstance(unpickled_context.children[i], context_type) + self.assertIsNone(unpickled_context.children[i].db.alias) class CompositeSessionContextTestCase(SimpleTestCase): diff --git a/tests/testapp/tests/sync/test_operations.py b/tests/testapp/tests/sync/test_operations.py index 979c32c2..f7d472f2 100644 --- a/tests/testapp/tests/sync/test_operations.py +++ b/tests/testapp/tests/sync/test_operations.py @@ -2,7 +2,10 @@ import uuid import mock +import pytest +from django.conf import settings from django.db import connection +from django.db import transaction from django.test import override_settings from django.test import TestCase from django.utils import timezone @@ -29,6 +32,7 @@ from morango.sync.context import LocalSessionContext from morango.sync.controller import MorangoProfileController from morango.sync.controller import SessionController +from morango.sync.db import SyncDatabase from morango.sync.operations import _dequeue_into_store from morango.sync.operations import _queue_into_buffer_v1 from morango.sync.operations import _queue_into_buffer_v2 @@ -43,6 +47,7 @@ from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.syncsession import TransferClient +from ..helpers import assert_no_default_db_queries from ..helpers import create_buffer_and_store_dummy_data from ..helpers import create_dummy_store_data @@ -673,6 +678,7 @@ def setUp(self): sync_session=self.transfer_session.sync_session, is_server=True, capabilities=[], + db=SyncDatabase.default(), ) def assert_store_records_tagged_with_last_session(self, store_ids): @@ -707,7 +713,11 @@ def test_dequeuing_sets_last_session(self): store_ids = [self.data[key] for key in ["model2", "model3", "model4", "model5", "model7"]] self.assert_store_records_not_tagged_with_last_session(store_ids) _dequeue_into_store( - self.transfer_session, self.transfer_session.client_fsic, v2_format=False + self.transfer_session, + self.transfer_session.client_fsic, + v2_format=False, + self_ref_order=True, + db=SyncDatabase.default(), ) # this one is a reverse fast forward, so it doesn't modify the store record and shouldn't be tagged self.assert_store_records_not_tagged_with_last_session([self.data["model1"]]) @@ -939,7 +949,11 @@ def test_dequeuing_delete_remaining_buffer(self): def test_dequeue_into_store(self): _dequeue_into_store( - self.transfer_session, self.transfer_session.client_fsic, v2_format=False + self.transfer_session, + self.transfer_session.client_fsic, + v2_format=False, + self_ref_order=True, + db=SyncDatabase.default(), ) # ensure a record with different transfer session id is not affected self.assertTrue(Buffer.objects.filter(transfer_session_id=self.data["tfs_id"]).exists()) @@ -1023,6 +1037,7 @@ def test_dequeue_into_store__self_ref_order_fallback_for_missing_capability(self self.transfer_session.client_fsic, v2_format=False, self_ref_order=False, + db=SyncDatabase.default(), ) self.assertEqual(Store.objects.get(id=self.data["model3"])._self_ref_order, 0) @@ -1034,7 +1049,7 @@ def test_update_legacy_self_ref_order_nulls_non_self_ref_models(self): _self_ref_order=3, ) - _update_legacy_self_ref_order(self.transfer_session) + _update_legacy_self_ref_order(self.transfer_session, SyncDatabase.default()) store.refresh_from_db() self.assertIsNone(store._self_ref_order) @@ -1050,7 +1065,7 @@ def test_update_legacy_self_ref_order_handles_deeper_self_ref_chains(self): _self_ref_order=None, ) - _update_legacy_self_ref_order(self.transfer_session) + _update_legacy_self_ref_order(self.transfer_session, SyncDatabase.default()) root.refresh_from_db() child.refresh_from_db() @@ -1072,6 +1087,7 @@ def test_local_dequeue_operation(self): def test_local_dequeue_operation__passes_self_ref_order_capability(self, mock_dequeue): self.transfer_session.records_transferred = 1 self.context.capabilities = {SELF_REF_ORDER} + self.context.db = SyncDatabase("default2") operation = ReceiverDequeueOperation() self.assertEqual(transfer_statuses.COMPLETED, operation.handle(self.context)) mock_dequeue.assert_called_once_with( @@ -1079,6 +1095,7 @@ def test_local_dequeue_operation__passes_self_ref_order_capability(self, mock_de self.transfer_session.client_fsic, v2_format=False, self_ref_order=True, + db=self.context.db, ) @mock.patch("morango.sync.operations._dequeue_into_store") @@ -1126,6 +1143,126 @@ def test_local_cleanup(self): ) +@pytest.mark.skipif("default2" not in settings.DATABASES, reason="requires a second database") +@override_settings(DATABASE_ROUTERS=[]) +class PinnedDequeueTestCase(TestCase): + databases = {"default", "default2"} + + def setUp(self): + self.db = SyncDatabase("default2") + self.host_instance, _ = InstanceIDModel.get_or_create_current_instance(clear_cache=True) + self.host_counter = self.host_instance.counter + database = DatabaseIDModel(id=uuid.uuid4().hex) + self.db.manager(DatabaseIDModel).bulk_create([database]) + self.instance = InstanceIDModel( + id=uuid.uuid4().hex, database_id=database.id, counter=7, current=True + ) + self.db.manager(InstanceIDModel).bulk_create([self.instance]) + session = self.db.manager(SyncSession).create( + id=uuid.uuid4().hex, profile="facilitydata", last_activity_timestamp=timezone.now() + ) + self.transfer_session = self.db.manager(TransferSession).create( + id=uuid.uuid4().hex, + sync_session_id=session.id, + last_activity_timestamp=timezone.now(), + filter="partition", + push=True, + ) + self.root_id = uuid.uuid4().hex + self.child_id = uuid.uuid4().hex + self.conflict_id = uuid.uuid4().hex + self.sender_id = uuid.uuid4().hex + for record_id, parent_id in ( + (self.root_id, ""), + (self.child_id, self.root_id), + (self.conflict_id, ""), + ): + self.db.manager(Buffer).create( + transfer_session_id=self.transfer_session.id, + model_uuid=record_id, + profile="facilitydata", + model_name="facility", + partition="partition", + serialized="buffer", + last_saved_instance=self.sender_id, + last_saved_counter=2, + _self_ref_fk=parent_id, + _self_ref_order=0 if not parent_id else 1, + ) + self.db.manager(RecordMaxCounterBuffer).create( + transfer_session_id=self.transfer_session.id, + model_uuid=record_id, + instance_id=self.sender_id, + counter=2, + ) + self.db.manager(Store).create( + id=self.conflict_id, + profile="facilitydata", + model_name="facility", + partition="partition", + serialized="local", + last_saved_instance=self.instance.id, + last_saved_counter=7, + ) + self.db.manager(RecordMaxCounter).create( + store_model_id=self.conflict_id, instance_id=self.instance.id, counter=7 + ) + + @mock.patch.object(DatabaseMaxCounter, "update_fsics") + def test_pinned_dequeue(self, update_fsics): + # FSIC helper routing is introduced by Task 6 (#367), after this dequeue refactor. + for self_ref_order in (True, False): + with self.subTest(self_ref_order=self_ref_order): + with self.db.atomic(): + with assert_no_default_db_queries(): + if not self_ref_order: + self.db.manager(Buffer).update(_self_ref_order=None) + _dequeue_into_store( + self.transfer_session, "{}", False, self_ref_order, self.db + ) + stores = self.db.manager(Store) + self.assertEqual(stores.get(pk=self.root_id).serialized, "buffer") + self.assertEqual(stores.get(pk=self.root_id)._self_ref_order, 0) + self.assertEqual(stores.get(pk=self.child_id)._self_ref_order, 1) + conflict = stores.get(pk=self.conflict_id) + self.assertEqual(conflict.last_saved_instance, self.instance.id) + self.assertEqual(conflict.last_saved_counter, 8) + self.assertEqual(conflict.serialized, "local") + self.assertEqual(conflict.conflicting_serialized_data, "buffer\n") + self.assertEqual( + self.db.manager(InstanceIDModel).get(pk=self.instance.pk).counter, 8 + ) + self.assertEqual( + self.db.manager(RecordMaxCounter) + .filter(store_model_id__in=[self.root_id, self.child_id]) + .count(), + 2, + ) + self.assertFalse(self.db.manager(Buffer).exists()) + self.assertFalse(self.db.manager(RecordMaxCounterBuffer).exists()) + # Roll back the subtest's writes so each capability path starts with buffers. + transaction.set_rollback(True, using=self.db.alias) + + update_fsics.assert_has_calls([mock.call({}, Filter("partition"), v2_format=False)] * 2) + self.host_instance.refresh_from_db() + self.assertEqual(self.host_instance.counter, self.host_counter) + self.assertFalse(Store.objects.filter(pk__in=[self.root_id, self.child_id]).exists()) + + @mock.patch.object(DatabaseMaxCounter, "update_fsics") + def test_pinned_dequeue_rolls_back_on_failure(self, update_fsics): + with mock.patch.object( + self.db.backend, "_dequeuing_insert_remaining_rmcb", side_effect=RuntimeError("stop") + ): + with self.assertRaisesRegex(RuntimeError, "stop"), assert_no_default_db_queries(): + _dequeue_into_store(self.transfer_session, "{}", False, True, self.db) + + self.assertEqual(self.db.manager(Buffer).count(), 3) + self.assertFalse(self.db.manager(Store).filter(pk=self.root_id).exists()) + self.assertEqual(self.db.manager(Store).get(pk=self.conflict_id).last_saved_counter, 7) + self.assertEqual(self.db.manager(InstanceIDModel).get(pk=self.instance.pk).counter, 7) + update_fsics.assert_not_called() + + class DeserializationTestCases(TestCase): def setUp(self): self.profile = "facilitydata"