Reorder filters produced by Scope

CHANGELOG.md

@@ -13,6 +13,8 @@ List of the most important changes for each release.
 - Logs deserialization warnings under `morango.sync.stream.deserialize` instead of `morango.sync.operations`
 - Removes `SyncableModel.deferred_clean_fields` and `morango.models.core.ForeignKeyReference`
 - Removes `morango.sync.backends.utils.TemporaryTable` and `morango.sync.backends.utils.get_pk_field`
+- `Scope.read_filter` and `Scope.write_filter` now list the read-write partitions last, so deserialization, which follows the filter order, puts shared records before the records that reference them. A server that receives a push from a client on an older version deserializes in the client's filter order
+- `cleanupsyncs --sync-filter` now also matches transfer sessions whose filter has the same partitions in a different order, so it still cleans up sessions stored before the partition order changed
 - Removes the unused bulk write helpers `_bulk_full_record_upsert`, `_bulk_update` and `_bulk_insert`, and the temporary table helper `_create_temporary_table`, from the sync backends
 
 ## 0.8.15

docs/architecture/index.rst

@@ -71,6 +71,8 @@ A **filter** is a set of `partition prefixes <#partitions>`__ represented as an
 
 When designing scopes – i.e. composing scopes from filters and partitions – care must be taken to ensure that foreign keys in synced models refer to other models that were also synced in the same scope. Otherwise, an alternative would be to ensure that the application can gracefully handle missing records when necessary because there would be no guarantee of coherence.
 
+The order of the partitions also matters. A scope's read filter lists the partitions of its read template, then those of its read-write template, and its write filter lists the partitions of its write template, then those of its read-write template. Deserialization processes the partitions in this order, so a record's foreign keys must refer to records in the same or an earlier partition. Within each template, list the partitions of shared records before the partitions of the records that refer to them. Read-write partitions usually hold a device's own records, such as a user's logs, which refer to shared records in the other partitions. Records in read-only or write-only partitions must not refer to records in read-write partitions. For example, the ``single-user`` scope below keeps the user in a read-only partition, and the user's logs in the read-write partition.
+
 As of this writing, there are currently two scope definitions defined in Kolibri for the ``facilitydata`` profile:
 
 - The ``full-facility`` scope provides full read and write access to all data related to a facility. This includes the facility model itself plus associated classes, lessons, users, groups, content interaction logs, and everything else related to running a typical Kolibri classroom server.

morango/management/commands/cleanupsyncs.py

@@ -8,9 +8,14 @@
 
 from morango.models import SyncSession
 from morango.models import TransferSession
+from morango.models.certificates import Filter
 
 logger = logging.getLogger(__name__)
 
+# the number of transfer sessions to query at once, which keeps each query below the limit of 999
+# bound parameters in older versions of SQLite
+TRANSFER_SESSION_BATCH_SIZE = 500
+
 
 class Command(BaseCommand):
     help = "Closes and cleans up the data for any incomplete sync sessions older than a certain number of hours."
@@ -45,7 +50,7 @@ def add_arguments(self, parser):
             "--sync-filter",
             type=str,
             default=None,
-            help="Filters the TransferSession models to those with 'filters' starting with 'sync_filter'",
+            help="Filters the TransferSession models to those with 'filters' starting with 'sync_filter', or with the same partitions in any order",
         )
         parser.add_argument(
             "--push",
@@ -60,6 +65,23 @@ def add_arguments(self, parser):
             help="Filters the TransferSession models to those with 'push' set to False",
         )
 
+    @staticmethod
+    def _matching_ids(transfer_sessions, sync_filter):
+        """
+        Returns the IDs of the transfer sessions whose filter starts with `sync_filter`, such as a
+        dataset ID, or has the same partitions in any order. The partition order of a filter can
+        change between versions, such as when `Scope` began listing read-write partitions last.
+        """
+        partitions = Filter(sync_filter)
+        return [
+            transfer_session_id
+            for transfer_session_id, transfer_session_filter in transfer_sessions.values_list(
+                "id", "filter"
+            )
+            if transfer_session_filter.startswith(sync_filter)
+            or Filter(transfer_session_filter) == partitions
+        ]
+
     def handle(self, *args, **options):
 
         # establish the cutoff time and date for stale sessions
@@ -85,33 +107,39 @@ def handle(self, *args, **options):
             last_activity_timestamp__lt=cutoff,
         )
 
-        if options["sync_filter"]:
-            transfer_sessions = transfer_sessions.filter(filter__startswith=options["sync_filter"])
-
         if options["push"] and not options["pull"]:
             transfer_sessions = transfer_sessions.filter(push=True)
 
         if options["pull"] and not options["push"]:
             transfer_sessions = transfer_sessions.filter(push=False)
 
-        transfer_count = transfer_sessions.count()
-
-        # loop over the stale sessions one by one to close them out
-        for i, transfer_session in enumerate(transfer_sessions):
-            logger.info(
-                "TransferSession {} of {}: deleting {} Buffers and {} RMC Buffers...".format(
-                    i + 1,
-                    transfer_count,
-                    transfer_session.buffer_set.all().count(),
-                    transfer_session.recordmaxcounterbuffer_set.all().count(),
+        # ordering this last to keep manual filtering to a minimum
+        if options["sync_filter"]:
+            transfer_session_ids = self._matching_ids(transfer_sessions, options["sync_filter"])
+        else:
+            transfer_session_ids = list(transfer_sessions.values_list("id", flat=True))
+
+        transfer_count = len(transfer_session_ids)
+
+        # loop over the stale sessions one by one to close them out, querying them in batches
+        for offset in range(0, transfer_count, TRANSFER_SESSION_BATCH_SIZE):
+            batch_ids = transfer_session_ids[offset : offset + TRANSFER_SESSION_BATCH_SIZE]
+            batch = TransferSession.objects.filter(id__in=batch_ids)
+            for i, transfer_session in enumerate(batch, start=offset):
+                logger.info(
+                    "TransferSession {} of {}: deleting {} Buffers and {} RMC Buffers...".format(
+                        i + 1,
+                        transfer_count,
+                        transfer_session.buffer_set.all().count(),
+                        transfer_session.recordmaxcounterbuffer_set.all().count(),
+                    )
                 )
-            )
 
-            # delete buffer data and mark as inactive
-            with transaction.atomic():
-                transfer_session.delete_buffers()
-                transfer_session.active = False
-                transfer_session.save()
+                # delete buffer data and mark as inactive
+                with transaction.atomic():
+                    transfer_session.delete_buffers()
+                    transfer_session.active = False
+                    transfer_session.save()
 
         # in order to close a sync session, it must have no active transfer sessions
         # and must have no activity since the cutoff

morango/models/certificates.py

@@ -483,12 +483,15 @@ def from_template(cls, template, params=None):
 
 
 class Scope(object):
+    """The read and write filters of a scope definition, with its params applied"""
+
     def __init__(self, definition, params):
-        # turn the scope definition filter templates into Filter objects
+        # turn the scope definition filter templates into Filter objects, with the read-write
+        # partitions last, since their records usually reference records in the others
         rw_filter = Filter.from_template(definition.read_write_filter_template, params)
-        self.read_filter = rw_filter + Filter.from_template(definition.read_filter_template, params)
-        self.write_filter = rw_filter + Filter.from_template(
-            definition.write_filter_template, params
+        self.read_filter = Filter.from_template(definition.read_filter_template, params) + rw_filter
+        self.write_filter = (
+            Filter.from_template(definition.write_filter_template, params) + rw_filter
         )
 
     def is_subset_of(self, other):

morango/sync/stream/source.py

@@ -55,8 +55,9 @@ def prefix_conditions(self) -> Generator[Optional[str], None, None]:
 
         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.
+        partitions of the records that reference them. `Scope` filters follow this by listing the
+        read-write partitions last (see `docs/architecture/index.rst`). 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.
         """

tests/testapp/tests/models/test_certificates.py

@@ -11,6 +11,7 @@
 from morango.models.certificates import Certificate
 from morango.models.certificates import Filter
 from morango.models.certificates import Key
+from morango.models.certificates import Scope
 from morango.models.certificates import ScopeDefinition
 
 
@@ -271,3 +272,47 @@ def test_from_template(self):
     def test_from_template__with_json_params(self):
         f = Filter.from_template("test:${param}", '{"param": "value"}')
         self.assertEqual(f._filter_tuple, ("test:value",))
+
+
+class ScopeTestCase(SimpleTestCase):
+    params = {"user_id": "U"}
+
+    def _scope(self, read="", write="", read_write=""):
+        definition = ScopeDefinition(
+            read_filter_template=read,
+            write_filter_template=write,
+            read_write_filter_template=read_write,
+        )
+        return Scope(definition, self.params)
+
+    def test_read_filter__read_write_last(self):
+        scope = self._scope(
+            read="a:allusers-ro\na:user-ro:${user_id}",
+            read_write="a:user-rw:${user_id}\na:rw2",
+        )
+        self.assertEqual(
+            tuple(scope.read_filter),
+            ("a:allusers-ro", "a:user-ro:U", "a:user-rw:U", "a:rw2"),
+        )
+
+    def test_write_filter__read_write_last(self):
+        scope = self._scope(
+            write="a:user-wo:${user_id}\na:other-wo",
+            read_write="a:user-rw:${user_id}\na:rw2",
+        )
+        self.assertEqual(
+            tuple(scope.write_filter),
+            ("a:user-wo:U", "a:other-wo", "a:user-rw:U", "a:rw2"),
+        )
+
+    def test_filters__duplicate_partition(self):
+        scope = self._scope(
+            read="a:shared\na:user-ro:${user_id}",
+            read_write="a:user-rw:${user_id}\na:shared",
+        )
+        self.assertEqual(tuple(scope.read_filter), ("a:shared", "a:user-ro:U", "a:user-rw:U"))
+
+    def test_filters__empty_read_write_template(self):
+        scope = self._scope(read="a:ro\nb:ro", write="a:wo")
+        self.assertEqual(tuple(scope.read_filter), ("a:ro", "b:ro"))
+        self.assertEqual(tuple(scope.write_filter), ("a:wo",))

tests/testapp/tests/sync/stream/test_deserialize.py

@@ -21,6 +21,8 @@
 from morango.errors import MorangoDirtyParent
 from morango.errors import MorangoMissingParent
 from morango.models.certificates import Filter
+from morango.models.certificates import Scope
+from morango.models.certificates import ScopeDefinition
 from morango.models.core import DeletedModels
 from morango.models.core import HardDeletedModels
 from morango.models.core import Store
@@ -1715,6 +1717,48 @@ def test_sync_filter__partition_order(self):
         self._assert_store(log.id, dirty_bit=False)
         self.assertTrue(SummaryLog.objects.filter(id=log.id).exists())
 
+    def _single_user_scope(self):
+        """
+        A scope in the shape of Kolibri's `single-user` scope, with the user in a read-only
+        partition and its logs in the read-write partition, which sorts first
+        """
+        definition = ScopeDefinition(
+            read_filter_template="u:b-ro:${user_id}",
+            write_filter_template="",
+            read_write_filter_template="u:a-rw:${user_id}",
+        )
+        return Scope(definition, {"user_id": "U"})
+
+    def test_sync_filter__scope_read_filter(self):
+        """
+        A `single-user` scope keeps the user in a read-only partition and its logs in the
+        read-write partition, and its read filter deserializes the user before the logs
+        """
+        user = self._user(partition="u:b-ro:U")
+        log = self._summary_log(user.id, partition="u:a-rw:U")
+
+        self._deserialize(sync_filter=self._single_user_scope().read_filter)
+
+        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_sync_filter__read_write_partition_first(self):
+        """
+        A filter that lists the read-write partition first, the order `Scope` used before,
+        deserializes the log before its user, so the log fails
+        """
+        user = self._user(partition="u:b-ro:U")
+        log = self._summary_log(user.id, partition="u:a-rw:U")
+
+        self._deserialize(sync_filter=Filter("u:a-rw:U\nu:b-ro:U"))
+
+        self._assert_store(user.id, dirty_bit=False)
+        self._assert_store(
+            log.id, dirty_bit=True, exception="django.core.exceptions.ValidationError"
+        )
+        self.assertFalse(SummaryLog.objects.exists())
+
     def test_summary_logged(self):
         self._user()
         self._facility(serialized="{bad")

tests/testapp/tests/test_management_commands.py

@@ -1,6 +1,7 @@
 import datetime
 import uuid
 
+import mock
 from django.core.management import call_command
 from django.test import TestCase
 from django.utils import timezone
@@ -11,7 +12,7 @@
 from .helpers import create_buffer_and_store_dummy_data
 
 
-def _create_sessions(last_activity_offset=0, sync_session=None, push=True):
+def _create_sessions(last_activity_offset=0, sync_session=None, push=True, filter="1:2\n"):
 
     last_activity_timestamp = timezone.now() - datetime.timedelta(hours=last_activity_offset)
 
@@ -29,7 +30,7 @@ def _create_sessions(last_activity_offset=0, sync_session=None, push=True):
         sync_session=sync_session,
         push=push,
         last_activity_timestamp=last_activity_timestamp,
-        filter="1:2\n",
+        filter=filter,
     )
 
     return sync_session, transfer_session
@@ -127,6 +128,56 @@ def test_filtering_sessions_by_sync_filter_cleared(self):
         self.assertTransferSessionIsCleared(self.transfersession_new)
         self.assertSyncSessionIsNotActive(self.syncsession_new)
 
+    def test_filtering_sessions_by_sync_filter__prefix(self):
+        """A prefix of the stored filters, such as a dataset ID, matches them"""
+        call_command("cleanupsyncs", sync_filter="1", expiration=0)
+        self.assertTransferSessionIsCleared(self.transfersession_old)
+        self.assertTransferSessionIsCleared(self.transfersession_new)
+
+    def test_filtering_sessions_by_sync_filter__partition_order(self):
+        """
+        A filter with the same partitions in a different order matches, so sessions stored before
+        `Scope` changed its partition order are still cleaned up
+        """
+        _, reordered = _create_sessions(24, filter="1:2:rw\n1:2:ro")
+        create_buffer_and_store_dummy_data(reordered.id)
+
+        call_command("cleanupsyncs", sync_filter="1:2:ro\n1:2:rw", expiration=0)
+
+        self.assertTransferSessionIsCleared(reordered)
+        self.assertTransferSessionIsNotCleared(self.transfersession_old)
+        self.assertTransferSessionIsNotCleared(self.transfersession_new)
+
+    def test_filtering_sessions_by_sync_filter__different_partitions(self):
+        """A filter with other partitions does not match"""
+        _, other = _create_sessions(24, filter="1:2:rw\n1:2:ro")
+        create_buffer_and_store_dummy_data(other.id)
+
+        call_command("cleanupsyncs", sync_filter="1:2:ro\n1:2:wo", expiration=0)
+
+        self.assertTransferSessionIsNotCleared(other)
+
+    @mock.patch("morango.management.commands.cleanupsyncs.TRANSFER_SESSION_BATCH_SIZE", 2)
+    def test_sessions_cleared_in_batches(self):
+        """Sessions in every batch are cleared, including a partial last batch"""
+        reordered = []
+        for _ in range(3):
+            _, transfer_session = _create_sessions(24, filter="1:2:rw\n1:2:ro")
+            create_buffer_and_store_dummy_data(transfer_session.id)
+            reordered.append(transfer_session)
+
+        call_command("cleanupsyncs", sync_filter="1:2:ro\n1:2:rw", expiration=0)
+
+        for transfer_session in reordered:
+            self.assertTransferSessionIsCleared(transfer_session)
+        self.assertTransferSessionIsNotCleared(self.transfersession_old)
+
+    @mock.patch("morango.management.commands.cleanupsyncs.TRANSFER_SESSION_BATCH_SIZE", 1)
+    def test_sessions_cleared_in_batches__no_sync_filter(self):
+        call_command("cleanupsyncs", expiration=1)
+        self.assertTransferSessionIsCleared(self.transfersession_old)
+        self.assertTransferSessionIsCleared(self.transfersession_new)
+
     def test_filtering_sessions_by_push_cleared(self):
         call_command("cleanupsyncs", push=self.transfersession_old.push, expiration=0)
         self.assertTransferSessionIsCleared(self.transfersession_old)