diff --git a/CHANGELOG.md b/CHANGELOG.md index ece8bf65..45a07f59 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/architecture/index.rst b/docs/architecture/index.rst index 13723d9e..23c4ee2d 100644 --- a/docs/architecture/index.rst +++ b/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. diff --git a/morango/management/commands/cleanupsyncs.py b/morango/management/commands/cleanupsyncs.py index 326edfa8..3f0d605c 100644 --- a/morango/management/commands/cleanupsyncs.py +++ b/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 diff --git a/morango/models/certificates.py b/morango/models/certificates.py index 341af0bd..6ce30591 100644 --- a/morango/models/certificates.py +++ b/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): diff --git a/morango/sync/stream/source.py b/morango/sync/stream/source.py index 2a0057f7..da08467a 100644 --- a/morango/sync/stream/source.py +++ b/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. """ diff --git a/tests/testapp/tests/models/test_certificates.py b/tests/testapp/tests/models/test_certificates.py index a7b3289c..02092562 100644 --- a/tests/testapp/tests/models/test_certificates.py +++ b/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",)) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index e43336ef..8f527b9d 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/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") diff --git a/tests/testapp/tests/test_management_commands.py b/tests/testapp/tests/test_management_commands.py index 85b48b8b..f11d9614 100644 --- a/tests/testapp/tests/test_management_commands.py +++ b/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)