implement server compute order during the dequeue stage

Author: ozer550

morango/constants/capabilities.py

@@ -2,3 +2,4 @@
 ALLOW_CERTIFICATE_PUSHING = "ALLOW_CERTIFICATE_PUSHING"
 ASYNC_OPERATIONS = "ASYNC_OPERATIONS"
 FSIC_V2_FORMAT = "FSIC_V2_FORMAT"
+SELF_REF_ORDER = "SELF_REF_ORDER"

morango/sync/operations.py

@@ -19,6 +19,7 @@
 from morango.constants import transfer_statuses
 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 MorangoInvalidFSICPartition
 from morango.errors import MorangoLimitExceeded
 from morango.errors import MorangoResumeSyncError
@@ -679,7 +680,52 @@ def _queue_into_buffer_v2(transfersession, chunk_size=200):
             )
 
 
-def _dequeue_into_store(transfer_session, fsic, v2_format=False):
+def _update_legacy_self_ref_order(transfer_session):
+    transferred_store_records = Store.objects.filter(
+        last_transfer_session_id=transfer_session.id
+    )
+    records_by_id = {record.id: record for record in transferred_store_records}
+    records_by_model = defaultdict(list)
+
+    for record in records_by_id.values():
+        try:
+            Model = syncable_models.get_model(record.profile, record.model_name)
+        except KeyError:
+            continue
+
+        if self_referential_fk(Model):
+            records_by_model[(record.profile, record.model_name)].append(record)
+        elif record._self_ref_order is not None:
+            record._self_ref_order = None
+            record.save(update_fields=["_self_ref_order"])
+
+    for records in records_by_model.values():
+        pending_record_ids = set(record.id for record in records)
+        while pending_record_ids:
+            updated = False
+            for record_id in list(pending_record_ids):
+                record = records_by_id[record_id]
+                if not record._self_ref_fk:
+                    order = 0
+                else:
+                    parent = records_by_id.get(record._self_ref_fk)
+                    if parent is not None and parent.id in pending_record_ids:
+                        continue
+                    order = Store.objects.filter(id=record._self_ref_fk).values_list(
+                        "_self_ref_order", flat=True
+                    ).first()
+
+                if record._self_ref_order != order:
+                    record._self_ref_order = order
+                    record.save(update_fields=["_self_ref_order"])
+                pending_record_ids.remove(record_id)
+                updated = True
+
+            if not updated:
+                break
+
+
+def _dequeue_into_store(transfer_session, fsic, v2_format=False, self_ref_order=True):
     """
     Takes data from the buffers and merges into the store and record max counters.
 
@@ -703,6 +749,8 @@ def _dequeue_into_store(transfer_session, fsic, v2_format=False):
             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)
+            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)
 
@@ -1041,6 +1089,7 @@ def handle(self, context):
                 context.transfer_session,
                 fsic,
                 v2_format=FSIC_V2_FORMAT in context.capabilities,
+                self_ref_order=SELF_REF_ORDER in context.capabilities,
             )
 
         return transfer_statuses.COMPLETED

morango/utils.py

@@ -9,6 +9,7 @@
     ASYNC_OPERATIONS,
     FSIC_V2_FORMAT,
     GZIP_BUFFER_POST,
+    SELF_REF_ORDER,
 )
 
 
@@ -61,6 +62,8 @@ def get_capabilities():
     if not SETTINGS.MORANGO_DISALLOW_ASYNC_OPERATIONS:
         capabilities.add(ASYNC_OPERATIONS)
 
+    capabilities.add(SELF_REF_ORDER)
+
     return capabilities
 
 

tests/testapp/tests/sync/test_operations.py

@@ -8,7 +8,7 @@
 from facility_profile.models import ConditionalLog, Facility, MyUser, SummaryLog
 
 from morango.constants import transfer_statuses
-from morango.constants.capabilities import FSIC_V2_FORMAT
+from morango.constants.capabilities import FSIC_V2_FORMAT, SELF_REF_ORDER
 from morango.errors import MorangoLimitExceeded
 from morango.models.certificates import Filter
 from morango.models.core import (
@@ -991,6 +991,24 @@ def test_dequeue_into_store(self):
             ).exists()
         )
 
+    def test_dequeue_into_store__self_ref_order_fallback_for_missing_capability(self):
+        Buffer.objects.filter(model_uuid=self.data["model3"]).update(
+            _self_ref_fk="", _self_ref_order=None
+        )
+        Buffer.objects.filter(model_uuid=self.data["model4"]).update(
+            _self_ref_fk=self.data["model3"], _self_ref_order=None
+        )
+
+        _dequeue_into_store(
+            self.transfer_session,
+            self.transfer_session.client_fsic,
+            v2_format=False,
+            self_ref_order=False,
+        )
+
+        self.assertEqual(Store.objects.get(id=self.data["model3"])._self_ref_order, 0)
+        self.assertEqual(Store.objects.get(id=self.data["model4"])._self_ref_order, 0)
+
     def test_local_dequeue_operation(self):
         self.transfer_session.records_transferred = 1
         self.context.filter = [self.transfer_session.filter]
@@ -1000,6 +1018,19 @@ def test_local_dequeue_operation(self):
             Buffer.objects.filter(transfer_session_id=self.transfer_session.id).exists()
         )
 
+    @mock.patch("morango.sync.operations._dequeue_into_store")
+    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}
+        operation = ReceiverDequeueOperation()
+        self.assertEqual(transfer_statuses.COMPLETED, operation.handle(self.context))
+        mock_dequeue.assert_called_once_with(
+            self.transfer_session,
+            self.transfer_session.client_fsic,
+            v2_format=False,
+            self_ref_order=True,
+        )
+
     @mock.patch("morango.sync.operations._dequeue_into_store")
     def test_local_dequeue_operation__noop(self, mock_dequeue):
         self.context.is_server = False

tests/testapp/tests/test_utils.py

@@ -12,6 +12,7 @@
     ALLOW_CERTIFICATE_PUSHING,
     ASYNC_OPERATIONS,
     FSIC_V2_FORMAT,
+    SELF_REF_ORDER,
 )
 from morango.utils import (
     CAPABILITIES_CLIENT_HEADER,
@@ -73,6 +74,9 @@ def test_get_capabilities__fsic_v2_format(self):
         with self.settings(MORANGO_DISABLE_FSIC_V2_FORMAT=True):
             self.assertNotIn(FSIC_V2_FORMAT, get_capabilities())
 
+    def test_get_capabilities__self_ref_order(self):
+        self.assertIn(SELF_REF_ORDER, get_capabilities())
+
     @mock.patch("morango.utils.CAPABILITIES", ("TEST", "SERIALIZE"))
     def test_serialize(self):
         req = Request()