From a1f86b57c499432b229c5ee75791965092c269ac Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Sat, 31 Oct 2026 06:30:58 -0700 Subject: [PATCH 1/4] List read-write partitions last in Scope filters Deserialization streams partitions in the order of the sync filter (#392), so a record whose FK target is in a later partition fails. Scope listed the read-write partitions first, which put a single-user scope's logs before the user they reference. The read filter now lists the read template's partitions, then the read-write template's, and the write filter does the same with the write template. Each template keeps its own order. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/models/certificates.py | 11 +++-- .../testapp/tests/models/test_certificates.py | 45 +++++++++++++++++++ 2 files changed, 52 insertions(+), 4 deletions(-) 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/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",)) From 8ca41376484660ef8921381fcafea530dac9ca28 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Sat, 31 Oct 2026 06:31:28 -0700 Subject: [PATCH 2/4] Test deserialization with a single-user Scope filter Builds a scope in the shape of Kolibri's single-user scope, with the user in a read-only partition and its log in the read-write partition, named so the read-write partition also sorts first. The scope's read filter deserializes both. A second test keeps the old read-write-first order as a counter-example, where the log fails on its missing user. Refs #398 Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/source.py | 5 ++- .../tests/sync/stream/test_deserialize.py | 44 +++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) 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/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") From 95df18dcc2c4ece13a7f5307df043c43acfab0c3 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Sat, 31 Oct 2026 06:33:37 -0700 Subject: [PATCH 3/4] Document the partition order of Scope filters Explains how a scope's filters order the partitions of its templates, and how implementers should order their templates so deserialization puts shared records before the records that reference them. The changelog notes that a server receiving a push from an older client still deserializes in the client's filter order. Closes #398 Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 1 + docs/architecture/index.rst | 2 ++ 2 files changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ece8bf65..170afded 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ 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 - 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. From aa614f2205003b37d339a871580e95b2e978d609 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Mon, 2 Nov 2026 07:31:05 -0800 Subject: [PATCH 4/4] Match sync filters in cleanupsyncs regardless of partition order cleanupsyncs matched --sync-filter against the stored filter string with startswith. Kolibri passes the full filter string of the current sync, so after Scope began listing read-write partitions last, the command no longer matched transfer sessions stored in the old order, and left their buffers behind. A transfer session now matches if its filter starts with the given filter, as before, or has the same partitions in any order. The match runs in Python, after the other options narrow the candidates in the database. The command now loads the transfer sessions to close in batches of 500 IDs, so no query binds more parameters than older versions of SQLite allow. Refs #398 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + morango/management/commands/cleanupsyncs.py | 68 +++++++++++++------ .../testapp/tests/test_management_commands.py | 55 ++++++++++++++- 3 files changed, 102 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 170afded..45a07f59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ List of the most important changes for each release. - 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/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/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)