Merge v0.8.x branch into v0.9.x

CHANGELOG.md

@@ -2,6 +2,23 @@
 
 List of the most important changes for each release.
 
+## 0.8.15
+- Improves buffer serialization performance by performing bulk counter lookup
+- Corrects an initialization issue with `SessionContext` that leads to an incorrect value for `is_push`
+- Save calls to Morango models during a sync are now scoped to only the changed fields
+
+## 0.8.14
+- Adds utility for addressing immediate FK constraints caused by Django upgrade, automatically performed for morango models in a Django migration.
+- Adds retry behavior for low-level connection issues not handled by `urllib3` retries
+- Allows repeat pushes and pulls of buffers during transfer
+- Ignores HTTP 404 errors during sync or transfer session closure, which may occur if they're already closed
+
+## 0.8.13
+- Removes multiprocessing fallback for RSA key generation to avoid leaving zombie processes; key generation now stays in-process
+
+## 0.8.12
+- Fixes issue where dirty-bit isn't updated when calling save on a syncable model with `update_fields`
+
 ## 0.8.11
 - Adds additional `deserialization_exception` field to `Store` model to track the fully qualified exception path
 

morango/api/serializers.py

@@ -1,3 +1,5 @@
+from collections import defaultdict
+
 from rest_framework import exceptions
 from rest_framework import serializers
 from rest_framework.fields import ReadOnlyField
@@ -157,11 +159,40 @@ class Meta:
         read_only_fields = fields
 
 
+class BufferListSerializer(serializers.ListSerializer):
+    def to_representation(self, data):
+        buffers = list(data)
+        if buffers:
+            rmcb_map = defaultdict(list)
+            transfer_session_ids = {b.transfer_session_id for b in buffers}
+            # Morango implementation will only ever call this for one transfer session ID at a time,
+            # but a loop makes the code straightforward regardless and limits the quantity of
+            # SQLite variables in use
+            for transfer_session_id in transfer_session_ids:
+                # bulk-fetch all RMCB records needed for this batch of buffers in a
+                # single query, then cache the relevant subset on each buffer, instead
+                # of letting each buffer's nested rmcb_list serializer issue its own query
+                rmcb_queryset = RecordMaxCounterBuffer.objects.filter(
+                    transfer_session_id=transfer_session_id,
+                    model_uuid__in={
+                        b.model_uuid
+                        for b in buffers
+                        if b.transfer_session_id == transfer_session_id
+                    },
+                )
+                for rmcb in rmcb_queryset:
+                    rmcb_map[(transfer_session_id, rmcb.model_uuid)].append(rmcb)
+            for buffer in buffers:
+                buffer._rmcb_list = rmcb_map[(buffer.transfer_session_id, buffer.model_uuid)]
+        return super(BufferListSerializer, self).to_representation(buffers)
+
+
 class BufferSerializer(serializers.ModelSerializer):
     rmcb_list = RecordMaxCounterBufferSerializer(many=True)
 
     class Meta:
         model = Buffer
+        list_serializer_class = BufferListSerializer
         fields = (
             "serialized",
             "deleted",

morango/deferrable_foreign_keys.py

@@ -0,0 +1,137 @@
+"""
+Helpers for ensuring that existing SQLite foreign key constraints are
+``DEFERRABLE INITIALLY DEFERRED``.
+
+Prior to Django 3.1, SQLite tables were created with immediate foreign key constraints, e.g.::
+
+    "store_model_id" char(32) NOT NULL REFERENCES "morango_store" ("id")
+
+Since Django 3.1 the same column is created as::
+
+    "store_model_id" char(32) NOT NULL REFERENCES "morango_store" ("id") DEFERRABLE INITIALLY DEFERRED
+
+Django relies on deferred constraint checking for correct cascade-deletion ordering. Databases
+created before Morango 0.8 (with Django 1.11) therefore retain immediate constraints, which can
+raise ``IntegrityError: FOREIGN KEY constraint failed`` during operations such as sync
+deserialization once foreign key enforcement is enabled at the database level (the default since
+Django 3.2).
+
+The migration framework does not regenerate these constraints on its own because the deferrable
+clause is not part of the field definition that migrations track. This module rebuilds the affected
+tables so that a migrated database ends up with the same schema as a freshly created one.
+"""
+
+import logging
+from typing import List
+from typing import Optional
+
+from django.apps import apps as global_apps
+from django.apps.registry import Apps
+from django.db import connection
+from django.db import DatabaseError
+from django.db import router
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+from django.db.backends.sqlite3.schema import DatabaseSchemaEditor
+
+logger = logging.getLogger(__name__)
+
+
+def _get_table_sql(conn, table_name: str) -> Optional[str]:
+    """
+    Return the ``CREATE TABLE`` statement stored by SQLite for ``table_name``, or ``None`` if the
+    table does not exist.
+    """
+    with conn.cursor() as cursor:
+        cursor.execute(
+            "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = %s",
+            [table_name],
+        )
+        row = cursor.fetchone()
+    return row[0] if row else None
+
+
+def _table_has_immediate_foreign_key(conn, table_name: str) -> bool:
+    """
+    Determine whether ``table_name`` has at least one foreign key constraint that is not deferrable.
+    Tables are created with all constraints in the same style, so the presence of a ``REFERENCES``
+    clause without ``DEFERRABLE`` indicates an immediate constraint that needs to be rebuilt.
+    """
+    sql = _get_table_sql(conn, table_name)
+    if not sql:
+        return False
+    references_count = sql.count("REFERENCES")
+    if references_count == 0:
+        return False
+    # make sure that all FK columns (identified by references) have a deferrable constraint
+    return references_count != sql.count("DEFERRABLE")
+
+
+class MakeForeignKeysDeferrable:
+    """
+    Invokable utility class that rebuilds of any table belonging to one of ``*app_labels`` whose
+    foreign key constraints are still immediate, so that they become
+    ``DEFERRABLE INITIALLY DEFERRED``.
+
+    This can be utilized by passing this class to ``RunPython`` in a Django migration, or by
+    manually invoking its ``run`` method.
+
+    This is a no-op on non-SQLite backends (PostgreSQL already creates deferrable
+    constraints) and on tables that already use deferrable constraints, so it is
+    safe to apply to any database.
+    """
+
+    def __init__(
+        self,
+        include_app_labels: Optional[List[str]] = None,
+        exclude_app_labels: Optional[List[str]] = None,
+    ):
+        self.include_app_labels = include_app_labels
+        self.exclude_app_labels = exclude_app_labels
+
+    def _iter_apps(self, apps: Apps):
+        for app_config in apps.get_app_configs():
+            if self.exclude_app_labels is not None and app_config.label in self.exclude_app_labels:
+                continue
+            if self.include_app_labels is None or app_config.label in self.include_app_labels:
+                yield app_config
+
+    def __call__(self, apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
+        """
+        Loops through all relevant django apps and their models, checking if the model's table has
+        FKs, and if so, runs the schema editor's utility that synchronizes the schema to the model,
+        which will make FKs deferrable
+        """
+        conn = schema_editor.connection
+        # these should be in sync, but since we call `_remake_table`, specific to this schema editor
+        # class, we ensure we have the correct instance to begin with
+        if conn.vendor != "sqlite" or not isinstance(schema_editor, DatabaseSchemaEditor):
+            return
+
+        for app_config in self._iter_apps(apps):
+            for model in app_config.get_models(include_auto_created=True):
+                if not router.allow_migrate_model(conn.alias, model):
+                    continue
+                table_name = model._meta.db_table
+                if not _table_has_immediate_foreign_key(conn, table_name):
+                    continue
+                logger.info(
+                    "Rebuilding table %s to make foreign key constraints deferrable",
+                    table_name,
+                )
+                try:
+                    # Rebuilding the table from the current model state regenerates the column
+                    # definitions, which SQLite always emits with the deferrable clause (see
+                    # DatabaseSchemaEditor.sql_create_inline_fk). Only tested with Django 3.2
+                    schema_editor._remake_table(model)
+                except DatabaseError as e:
+                    logger.error("Failed to rebuild table %s", table_name)
+                    logger.exception(e)
+
+    def run(self):
+        """
+        Invoke this utility outside of Django migrations, manually passing in Django's app registry
+        and schema editor.
+        """
+        # since this method is meant for manual invocation, we let the editor create a transaction
+        with connection.schema_editor(atomic=True) as schema_editor:
+            self.__call__(global_apps, schema_editor)

morango/migrations/0004_deferrable_foreign_keys.py

@@ -0,0 +1,24 @@
+"""
+Make existing foreign key constraints ``DEFERRABLE INITIALLY DEFERRED``.
+
+Databases created before Morango 0.8 (with Django 1.11) have immediate SQLite foreign key
+constraints, which can break cascade deletions (e.g. during sync) now that foreign keys are enforced
+at the database level. See https://github.com/learningequality/kolibri/issues/14884.
+"""
+
+from django.db import migrations
+
+from morango.deferrable_foreign_keys import MakeForeignKeysDeferrable
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("morango", "0003_store_deserialization_errors"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            MakeForeignKeysDeferrable(include_app_labels=["morango"]),
+            migrations.RunPython.noop,
+        ),
+    ]

morango/migrations/0004_self_ref_order.py → morango/migrations/0005_self_ref_order.py

@@ -7,7 +7,7 @@
 
 class Migration(migrations.Migration):
     dependencies = [
-        ("morango", "0003_store_deserialization_errors"),
+        ("morango", "0004_deferrable_foreign_keys"),
     ]
 
     operations = [

morango/models/core.py

@@ -322,6 +322,7 @@ def update_state(self, stage=None, stage_status=None):
         :type stage: morango.constants.transfer_stages.*|None
         :type stage_status: morango.constants.transfer_statuses.*|None
         """
+        update_fields = []
         if stage is not None:
             if self.transfer_stage and transfer_stages.stage(
                 self.transfer_stage
@@ -332,13 +333,18 @@ def update_state(self, stage=None, stage_status=None):
                     )
                 )
             self.transfer_stage = stage
+            update_fields.append("transfer_stage")
+
         if stage_status is not None:
             self.transfer_stage_status = stage_status
-        if stage is not None or stage_status is not None:
+            update_fields.append("transfer_stage_status")
+
+        if update_fields:
             self.last_activity_timestamp = timezone.now()
-            self.save()
+            update_fields.append("last_activity_timestamp")
+            self.save(update_fields=update_fields)
             self.sync_session.last_activity_timestamp = timezone.now()
-            self.sync_session.save()
+            self.sync_session.save(update_fields=["last_activity_timestamp"])
 
     def delete_buffers(self):
         """
@@ -565,6 +571,11 @@ class Meta:
         unique_together = ("transfer_session", "model_uuid")
 
     def rmcb_list(self):
+        # allow callers (e.g. BufferListSerializer) to batch-fetch RMCB records
+        # for many buffers at once and cache them here, to avoid an N+1 query
+        # pattern when serializing a large number of buffers
+        if hasattr(self, "_rmcb_list"):
+            return self._rmcb_list
         return RecordMaxCounterBuffer.objects.filter(
             model_uuid=self.model_uuid, transfer_session_id=self.transfer_session_id
         )
@@ -844,6 +855,11 @@ def save(self, update_dirty_bit_to=True, *args, **kwargs):
             self._morango_dirty_bit = True
         elif not update_dirty_bit_to:
             self._morango_dirty_bit = False
+
+        # ensure the dirty bit field is in the fields to update if present, to keep it in sync
+        if update_dirty_bit_to is not None and kwargs.get("update_fields") is not None:
+            kwargs["update_fields"] = set(kwargs["update_fields"]) | {"_morango_dirty_bit"}
+
         super(SyncableModel, self).save(*args, **kwargs)
 
     def delete(self, using=None, keep_parents=False, hard_delete=False, *args, **kwargs):

morango/models/fields/crypto.py

@@ -178,10 +178,7 @@ class PythonRSAKey(BaseKey):
     _private_key = None
 
     def generate_new_key(self, keysize=2048):
-        try:
-            self._public_key, self._private_key = PYRSA.newkeys(keysize, poolsize=4)
-        except:  # noqa: E722
-            self._public_key, self._private_key = PYRSA.newkeys(keysize)
+        self._public_key, self._private_key = PYRSA.newkeys(keysize)
 
     def _sign(self, message):
 

morango/sync/context.py

@@ -52,7 +52,7 @@ def __init__(
 
         if self.transfer_session:
             self.sync_session = transfer_session.sync_session or self.sync_session
-            self.is_push = transfer_session.push or self.is_push
+            self.is_push = transfer_session.push
             if transfer_session.filter:
                 self.filter = transfer_session.get_filter()
 

morango/sync/controller.py

@@ -254,7 +254,7 @@ def _invoke_middleware(self, context, middleware):
             return context.stage_status
         except Exception as e:
             # always log the error itself
-            logger.error(e)
+            logger.exception(e)
             context.update(stage_status=transfer_statuses.ERRORED, error=e)
             # fire completed signal, after context update. handlers can use context to detect error
             signal.completed.fire(context=prepared_context or context)