fix: decrypt a submission once however many decryption tasks run

docs/data.rst

@@ -963,7 +963,7 @@ Example
 
 Filter submissions using the decryption_status field.
 
-Use ``pending`` to get submissions that are pending decryption, ``success`` to get submissions that were successfully decrypted, ``failed`` to get submissions that failed decryption.
+Use ``pending`` to get submissions that are waiting to be decrypted, ``in_progress`` to get submissions that are being decrypted, ``success`` to get submissions that were successfully decrypted, ``failed`` to get submissions that failed decryption.
 
 Example
 ^^^^^^^^^
@@ -974,10 +974,13 @@ Example
 
 For submissions that have a status of ``failed``, there exists a ``_decryption_error`` field that contains an error name and message. The following are the possible errors:
 
-- ``MAX_RETRIES_EXCEEDED``: System was unable to decrypt the submission after multiple attempts.
 - ``KMS_KEY_DISABLED``: Encryption key is disabled.
 - ``KMS_KEY_NOT_FOUND``: Encryption key used for encryption not found.
 - ``INVALID_SUBMISSION``: Data is corrupted.
+
+A submission with a status of ``pending`` can be decrypted again. Another attempt is made when the submission is saved with all its media files received, such as when its last media file arrives. If an earlier attempt to decrypt it did not succeed, its ``_decryption_error`` field contains the reason. The following are the possible errors:
+
+- ``MAX_RETRIES_EXCEEDED``: System was unable to decrypt the submission after multiple attempts.
 - ``NOT_ALL_MEDIA_RECEIVED``: Not all media files have been received.
 
 Tag a submission data point

onadata/apps/api/tests/viewsets/test_xform_submission_viewset.py

@@ -2211,6 +2211,9 @@ def test_edit_managed_submission(self, mock_decrypt_async):
         self.assertTrue(edited_instance.is_encrypted)
         self.assertEqual(edited_instance.xml, self._enc_instance_manifest_xml())
         self.assertTrue(edited_instance.media_all_received)
+        self.assertEqual(
+            edited_instance.decryption_status, Instance.DecryptionStatus.PENDING
+        )
         history = InstanceHistory.objects.get(xform_instance=edited_instance)
         self.assertEqual(history.uuid, original_uuid)
         self.assertEqual(history.xml, original_instance.xml)

onadata/apps/logger/migrations/0042_decrypt_swept_encrypted_submissions.py

@@ -37,6 +37,7 @@ def decrypt_swept_encrypted_submissions(apps, schema_editor):
             continue
 
         instance.attachments.filter(deleted_at__isnull=False).update(deleted_at=None)
+        Instance.objects.filter(pk=instance.pk).update(decryption_status="pending")
 
         try:
             decrypt_instance(LiveInstance.objects.get(pk=instance.pk))

onadata/apps/logger/migrations/0048_alter_instance_decryption_status.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.17 on 2026-10-12 09:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('logger', '0047_instance_user_date_created_idx'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='instance',
+            name='decryption_status',
+            field=models.CharField(choices=[('unmanaged', 'Does not use managed keys'), ('pending', 'Pending decryption'), ('in_progress', 'Decryption in progress'), ('success', 'Decryption successful'), ('failed', 'Decryption failed')], default='unmanaged', max_length=20),
+        ),
+    ]

onadata/apps/logger/migrations/0049_retry_failed_decryptions.py

@@ -0,0 +1,71 @@
+"""Decrypt submissions that failed decryption for a temporary reason."""
+
+from django.db import migrations
+
+# Live models are used for decrypt_instance. This is needed because
+# apps.get_model("logger", "Instance") returns a frozen version of
+# Instance that has only the fields known at this migration
+from onadata.apps.logger.models import Instance as LiveInstance
+from onadata.libs.kms.tools import decrypt_instance
+
+
+# pylint: disable=unused-argument
+def retry_failed_decryptions(apps, schema_editor):
+    """Decrypt submissions that failed decryption for a temporary reason.
+
+    Decryption only picks up pending submissions, so these are set to pending
+    first, otherwise they would never be decrypted.
+    """
+    # pylint: disable=invalid-name
+    Instance = apps.get_model("logger", "Instance")
+    candidates = Instance.objects.filter(
+        deleted_at__isnull=True,
+        is_encrypted=True,
+        decryption_status="failed",
+    ).only("json")
+    processed_count = 0
+    decrypted_count = 0
+    failed_count = 0
+    skipped_count = 0
+
+    for instance in candidates.iterator(chunk_size=100):
+        processed_count += 1
+        print(f"Processed {processed_count} submissions")
+
+        if (instance.json or {}).get("_decryption_error") not in (
+            "NOT_ALL_MEDIA_RECEIVED",
+            "MAX_RETRIES_EXCEEDED",
+        ):
+            skipped_count += 1
+            continue
+
+        Instance.objects.filter(pk=instance.pk).update(decryption_status="pending")
+
+        try:
+            decrypt_instance(LiveInstance.objects.get(pk=instance.pk))
+            decrypted_count += 1
+            print(f"Decrypted Instance {instance.id}")
+
+        # pylint: disable=broad-exception-caught
+        except Exception as e:
+            # Best-effort; decryption records why the submission failed
+            failed_count += 1
+            print(f"Decrypting Instance {instance.id} failed: {e}")
+
+    print(
+        f"Decrypted {decrypted_count} submissions,"
+        f" {failed_count} failed, {skipped_count} skipped"
+    )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("logger", "0048_alter_instance_decryption_status"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            retry_failed_decryptions,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

onadata/apps/logger/models/instance.py

@@ -382,7 +382,12 @@ def save_full_json(instance, include_related=True):
 
     # Preserve dynamic metadata fields if present
     if (
-        instance.decryption_status == Instance.DecryptionStatus.FAILED
+        instance.decryption_status
+        in (
+            Instance.DecryptionStatus.PENDING,
+            Instance.DecryptionStatus.IN_PROGRESS,
+            Instance.DecryptionStatus.FAILED,
+        )
         and instance.json.get(DECRYPTION_ERROR) is not None
     ):
         json[DECRYPTION_ERROR] = instance.json[DECRYPTION_ERROR]
@@ -654,6 +659,7 @@ class Instance(models.Model, InstanceBaseClass):
     class DecryptionStatus(models.TextChoices):  # pylint: disable=too-many-ancestors
         UNMANAGED = "unmanaged", _("Does not use managed keys")
         PENDING = "pending", _("Pending decryption")
+        IN_PROGRESS = "in_progress", _("Decryption in progress")
         SUCCESS = "success", _("Decryption successful")
         FAILED = "failed", _("Decryption failed")
 
@@ -795,7 +801,7 @@ def _set_encryption_status(self):
         if self.check_encrypted():
             self.is_encrypted = True
 
-            if self.xform.is_was_managed:
+            if self._state.adding and self.xform.is_was_managed:
                 self.decryption_status = Instance.DecryptionStatus.PENDING
         else:
             self.is_encrypted = False

onadata/apps/logger/tasks.py

@@ -201,7 +201,11 @@ def on_failure(self, exc, task_id, args, kwargs, einfo):
         instance = self.get_instance_from_args(args, kwargs)
 
         if instance is not None and isinstance(exc, MaxRetriesExceededError):
-            save_decryption_error(instance, DECRYPTION_FAILURE_MAX_RETRIES)
+            save_decryption_error(
+                instance,
+                DECRYPTION_FAILURE_MAX_RETRIES,
+                Instance.DecryptionStatus.PENDING,
+            )
 
         super().on_failure(exc, task_id, args, kwargs, einfo)
 

onadata/apps/logger/tests/models/test_instance.py

@@ -1440,6 +1440,36 @@ def test_decryption_pending_encryption_disabled(self):
         self.assertTrue(instance.is_encrypted)
         self.assertEqual(instance.decryption_status, Instance.DecryptionStatus.PENDING)
 
+    def test_save_keeps_decryption_error_of_in_progress(self):
+        """Saving a submission being decrypted keeps its last decryption error."""
+        self._publish_managed_form()
+        instance = self._submit_encrypted_instance()
+        Instance.objects.filter(pk=instance.pk).update(
+            decryption_status=Instance.DecryptionStatus.IN_PROGRESS,
+            json={**instance.json, "_decryption_error": "MAX_RETRIES_EXCEEDED"},
+        )
+        instance.refresh_from_db()
+
+        instance.save()
+
+        instance.refresh_from_db()
+        self.assertEqual(instance.json.get("_decryption_error"), "MAX_RETRIES_EXCEEDED")
+
+    def test_save_keeps_decryption_error_of_pending(self):
+        """Saving a submission pending decryption keeps its last decryption error."""
+        self._publish_managed_form()
+        instance = self._submit_encrypted_instance()
+        Instance.objects.filter(pk=instance.pk).update(
+            json={**instance.json, "_decryption_error": "MAX_RETRIES_EXCEEDED"}
+        )
+        instance.refresh_from_db()
+
+        instance.save()
+
+        instance.refresh_from_db()
+        self.assertEqual(instance.decryption_status, Instance.DecryptionStatus.PENDING)
+        self.assertEqual(instance.json.get("_decryption_error"), "MAX_RETRIES_EXCEEDED")
+
     @patch(
         "onadata.apps.logger.tasks.adjust_xform_num_of_decrypted_submissions_async.delay"
     )

onadata/apps/logger/tests/test_tasks.py

@@ -15,7 +15,7 @@
 from celery.exceptions import MaxRetriesExceededError
 from valigetta.exceptions import ConnectionException as ValigettaConnectionException
 
-from onadata.apps.logger.models import EntityList
+from onadata.apps.logger.models import EntityList, Instance
 from onadata.apps.logger.tasks import (
     adjust_xform_num_of_decrypted_submissions_async,
     apply_project_date_modified_async,
@@ -425,10 +425,13 @@ def test_invalid_pk(self, mock_logger, mock_decrypt):
         mock_decrypt.assert_not_called()
         mock_logger.assert_called_once()
 
-    @patch("onadata.apps.logger.tasks.save_decryption_error")
     # pylint: disable=unused-argument
-    def test_max_retries_exceeded(self, mock_save_decryption_error, mock_decrypt):
-        """Instance is tagged as failed decryption if max retries exceeded"""
+    def test_max_retries_exceeded(self, mock_decrypt):
+        """Instance waits for decryption with an error if max retries exceeded"""
+        Instance.objects.filter(pk=self.instance.pk).update(
+            decryption_status=Instance.DecryptionStatus.PENDING
+        )
+
         decrypt_instance_async.on_failure(
             MaxRetriesExceededError(),
             "task_id",
@@ -437,8 +440,12 @@ def test_max_retries_exceeded(self, mock_save_decryption_error, mock_decrypt):
             None,
         )
 
-        mock_save_decryption_error.assert_called_once_with(
-            self.instance, "MAX_RETRIES_EXCEEDED"
+        self.instance.refresh_from_db()
+        self.assertEqual(
+            self.instance.decryption_status, Instance.DecryptionStatus.PENDING
+        )
+        self.assertEqual(
+            self.instance.json.get("_decryption_error"), "MAX_RETRIES_EXCEEDED"
         )
 
 

onadata/libs/kms/tools.py

@@ -57,7 +57,6 @@
 from onadata.libs.utils.common_tags import (
     DECRYPTION_ERROR,
     DECRYPTION_FAILURE_ENCRYPTION_UNMANAGED,
-    DECRYPTION_FAILURE_INSTANCE_NOT_ENCRYPTED,
     DECRYPTION_FAILURE_INVALID_SUBMISSION,
     DECRYPTION_FAILURE_KEY_DISABLED,
     DECRYPTION_FAILURE_KEY_NOT_FOUND,
@@ -70,7 +69,6 @@
     adjust_counter,
     commit_cached_counters,
     queryset_iterator,
-    update_fields_directly,
 )
 
 logger = logging.getLogger(__name__)
@@ -454,25 +452,63 @@ def encrypt_xform(xform, encrypted_by=None, override_encryption=False) -> None:
     _encrypt_xform(xform=xform, kms_key=kms_key, encrypted_by=encrypted_by)
 
 
-def save_decryption_error(instance: Instance, error_name: str):
+def save_decryption_error(
+    instance: Instance,
+    error_name: str,
+    status: str = Instance.DecryptionStatus.FAILED,
+):
     """Add decryption error metadata to Instance json.
 
+    The error is saved only if the Instance is unchanged since it was loaded,
+    so an edit saved meanwhile is not overwritten.
+
     :param instance: Instance
     :param error_name: Error code
+    :param status: Decryption status to set
     """
     # Create a copy that we'll mutate
     json = dict(instance.json or {})
     json[DECRYPTION_ERROR] = error_name
-    update_fields_directly(
-        instance,
-        json=json,
-        decryption_status=Instance.DecryptionStatus.FAILED,
-    )
+    Instance.objects.filter(
+        pk=instance.pk,
+        checksum=instance.checksum,
+        decryption_status=instance.decryption_status,
+    ).update(json=json, decryption_status=status)
 
 
-# pylint: disable=too-many-locals,too-many-statements
 def decrypt_instance(instance: Instance) -> None:
-    """Decrypt encrypted Instance
+    """Decrypt encrypted Instance if it is pending decryption
+
+    :param instance: Instance to be decrypted
+    """
+    claimed = Instance.objects.filter(
+        pk=instance.pk,
+        checksum=instance.checksum,
+        decryption_status=Instance.DecryptionStatus.PENDING,
+    ).update(decryption_status=Instance.DecryptionStatus.IN_PROGRESS)
+
+    if not claimed:
+        logger.info("Instance %s is not pending decryption", instance.pk)
+        return
+
+    instance.decryption_status = Instance.DecryptionStatus.IN_PROGRESS
+
+    try:
+        _decrypt_instance(instance)
+
+    except Exception:
+        Instance.objects.filter(
+            pk=instance.pk, decryption_status=Instance.DecryptionStatus.IN_PROGRESS
+        ).update(decryption_status=Instance.DecryptionStatus.PENDING)
+        raise
+
+
+# pylint: disable=too-many-locals,too-many-statements
+def _decrypt_instance(instance: Instance) -> None:
+    """Decrypt an Instance the caller has claimed for decryption.
+
+    Callers must first claim the Instance by moving its decryption status from
+    pending to in progress, as decrypt_instance does.
 
     :param instance: Instance to be decrypted
     """
@@ -492,11 +528,6 @@ def get_encrypted_files(attachment_qs):
 
         return enc_files
 
-    if not instance.check_encrypted():
-        raise DecryptionError(
-            DECRYPTION_FAILURE_MESSAGES[DECRYPTION_FAILURE_INSTANCE_NOT_ENCRYPTED]
-        )
-
     if not instance.xform.is_was_managed:
         raise DecryptionError(
             DECRYPTION_FAILURE_MESSAGES[DECRYPTION_FAILURE_ENCRYPTION_UNMANAGED]
@@ -519,7 +550,11 @@ def get_encrypted_files(attachment_qs):
         )
 
     if not instance.media_all_received:
-        save_decryption_error(instance, DECRYPTION_FAILURE_NOT_ALL_MEDIA_RECEIVED)
+        save_decryption_error(
+            instance,
+            DECRYPTION_FAILURE_NOT_ALL_MEDIA_RECEIVED,
+            Instance.DecryptionStatus.PENDING,
+        )
         raise NotAllMediaReceivedError(
             DECRYPTION_FAILURE_MESSAGES[DECRYPTION_FAILURE_NOT_ALL_MEDIA_RECEIVED]
         )
@@ -558,6 +593,23 @@ def get_encrypted_files(attachment_qs):
 
     try:
         with transaction.atomic():
+            # Lock the submission so an edit saved meanwhile is not overwritten
+            unchanged = (
+                Instance.objects.select_for_update()
+                .filter(
+                    pk=instance.pk,
+                    checksum=instance.checksum,
+                    decryption_status=Instance.DecryptionStatus.IN_PROGRESS,
+                )
+                .only("pk")
+            )
+
+            if not unchanged:
+                # An edit or another decryption changed the submission while
+                # it was being decrypted, so this result is stale
+                logger.info("Instance %s changed while being decrypted", instance.pk)
+                return
+
             for original_name, decrypted_file in decrypted_files:
                 if original_name.lower() == "submission.xml":
                     # Replace submission with decrypted submission