feat: Improve transaction handling to prevent concurrent processing and add timeout mechanism (Get Submission)

- Implemented locking to ensure submissions are processed by a single xqueue watcher.
- Added timeout mechanism for submissions stuck in 'pulled' state.
- Updated tests to cover new error scenarios and timeout handling.

Author: gabrielC1409

docs/source/decisions/0004-transaction-handling-in-get_submission

@@ -0,0 +1,33 @@
+Improving Transaction Handling in Get Submission
+================================================
+
+Status
+------
+
+**Provisional** *2025-03-19*
+
+Context
+-------
+
+We identified a concurrency issue in the submission processing flow. Multiple instances of the xqueue watcher were processing the same transaction simultaneously, causing duplicate handling of submissions. Additionally, submissions remained stuck in the "pulled" state indefinitely when errors or unexpected interruptions occurred during processing.
+
+Decision
+--------
+
+We introduced a robust locking mechanism at the database level to prevent concurrent processing of the same submission. Specifically, we employed database row-level locking (`select_for_update(nowait=True)`) on submissions to ensure only one xqueue watcher instance can process a transaction at a time. Additionally, we implemented a timeout-based fallback, marking submissions in the "pulled" state as available again if they exceed a defined timeout period (5 minutes).
+
+Consequences
+------------
+
+Positive
+~~~~~~~~
+
+- Ensures data consistency by preventing duplicate transaction processing.
+- Implements automatic recovery for submissions stuck in the "pulled" state, improving overall system reliability.
+- Minimal impact on system performance due to efficient database-level locking.
+
+Negative
+~~~~~~~~
+
+- Slightly increased complexity in transaction handling logic.
+- Potential minor latency increase when contention occurs due to database locks.

submissions/tests/test_viewsets.py

@@ -7,9 +7,9 @@
 
 from django.contrib.auth import get_user_model
 from django.core.files.base import ContentFile
+from django.db import DatabaseError
 from django.test import override_settings
 from django.urls import reverse
-from django.utils import timezone
 from rest_framework import status
 from rest_framework.permissions import AllowAny, IsAuthenticated
 from rest_framework.test import APITestCase
@@ -71,35 +71,66 @@ def test_get_submission_queue_empty(self):
             self.viewset.compose_reply(False, f"Queue '{queue_name}' is empty")
         )
 
-    @patch('submissions.views.xqueue.timezone.now', return_value=timezone.now())
-    @patch('submissions.views.xqueue.uuid.uuid4', return_value=str(uuid.uuid4()))
-    def test_get_submission_success(self, mock_uuid, _):
-        """Test successfully retrieving a submission from the queue."""
-        queue_name = 'prueba'
+    def test_get_submission_successful(self):
+        """Test retrieving a valid submission from the queue."""
         self.client.login(username='testuser', password='testpass')
-        new_submission = SubmissionFactory()
-        submission_queue_record = ExternalGraderDetailFactory(
+
+        queue_name = 'valid_queue'
+
+        submission = SubmissionFactory()
+        submission_record = ExternalGraderDetailFactory(
+            submission=submission,
             queue_name=queue_name,
-            status='pending',
-            submission=new_submission
+            status='pending'
         )
 
         response = self.client.get(self.get_submission_url, {'queue_name': queue_name})
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        content = json.loads(response.data['content'])
-        self.assertEqual(response.data['return_code'], 0)
+        submission_record.refresh_from_db()
+        self.assertEqual(submission_record.status, 'pulled')
+        self.assertIsNotNone(submission_record.pullkey)
+
+    def test_get_submission_error_updating_status(self):
+        """Test error handling when updating the submission status fails."""
+        self.client.login(username='testuser', password='testpass')
+        new_submission = SubmissionFactory()
+
+        queue_name = 'test_queue'
+        submission_record = ExternalGraderDetailFactory(
+            submission=new_submission,  # Se usa una nueva Submission
+            queue_name=queue_name,
+            status='pending'
+        )
+        with patch(
+            'submissions.views.xqueue.ExternalGraderDetail.update_status',
+            side_effect=ValueError('Invalid transition')
+        ):
+            response = self.client.get(self.get_submission_url, {'queue_name': queue_name})
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data,
+            self.viewset.compose_reply(False, "Error processing submission: Invalid transition")
+        )
+        submission_record.refresh_from_db()
+        self.assertEqual(submission_record.status, 'pending')
 
-        xqueue_header = json.loads(content['xqueue_header'])
-        xqueue_body = json.loads(content['xqueue_body'])
-        self.assertEqual(xqueue_header['submission_id'], new_submission.id)
-        self.assertEqual(xqueue_header['submission_key'], str(mock_uuid.return_value))
-        self.assertEqual(xqueue_body['student_response'], new_submission.answer)
-        self.assertEqual(content['xqueue_files'], '{}')
+    def test_get_submission_db_error(self):
+        """Test DatabaseError handling when retrieving a submission."""
+        self.client.login(username='testuser', password='testpass')
+        queue_name = 'test_queue'
 
-        submission_queue_record.refresh_from_db()
-        self.assertEqual(submission_queue_record.status, 'pulled')
-        self.assertEqual(submission_queue_record.pullkey, str(mock_uuid.return_value))
-        self.assertIsNotNone(submission_queue_record.status_time)
+        with patch(
+            'submissions.views.xqueue.ExternalGraderDetail.objects.select_for_update',
+            side_effect=DatabaseError("Simulated DB Error")
+        ):
+            response = self.client.get(self.get_submission_url, {'queue_name': queue_name})
+
+        self.assertEqual(response.status_code, status.HTTP_409_CONFLICT)
+        self.assertEqual(
+            response.data,
+            self.viewset.compose_reply(False, "Submission already in process")
+        )
 
     @patch('submissions.views.xqueue.ExternalGraderDetail.update_status',
            side_effect=ValueError('Invalid transition'))
@@ -283,7 +314,7 @@ def test_put_result_auto_retire(self):
     @patch('submissions.views.xqueue.log')
     def test_put_result_logging(self, mock_log):
         """
-        Test that appropriate logging occurs in various scenarios.
+        Test that appropriate logging occurs in various escenarios.
         """
         self.submission.queue_record.status = 'pulled'
         self.submission.queue_record.save()
@@ -305,14 +336,8 @@ def test_put_result_logging(self, mock_log):
             mock_set_score.return_value = True
             self.client.post(self.url, payload, format='json')
 
-        mock_log.info.assert_any_call(
-            "Score event sent to bus successfully <====="
-        )
-
-        mock_log.info.assert_any_call(
-            "Successfully updated submission score for submission %s",
-            self.submission.id
-        )
+        # Verificar que se llamó el log con el mensaje correcto
+        mock_log.info.assert_any_call('=====> Saving score {"score": 8}')
 
     @patch('submissions.views.xqueue.log')
     def test_put_result_success(self, mock_log):

submissions/views/xqueue.py

@@ -3,9 +3,11 @@
 import json
 import logging
 import uuid
+from datetime import timedelta
 
 from django.contrib.auth import authenticate, login, logout
-from django.db import transaction
+from django.db import DatabaseError, transaction
+from django.db.models import Q
 from django.http import HttpResponse
 from django.utils import timezone
 from openedx_events.content_authoring.data import ExternalGraderScoreData
@@ -117,6 +119,7 @@ def get_submission(self, request):
         - Submission data with pull information if successful
         - Error message if queue is empty or invalid
         """
+
         queue_name = request.query_params.get('queue_name')
 
         if not queue_name:
@@ -125,69 +128,70 @@ def get_submission(self, request):
                 status=status.HTTP_400_BAD_REQUEST
             )
 
-        # if queue_name not in settings.XQUEUES: TODO: Define how to set this variable,
-        #  maybe as a tutor config or hardcoded en edx platform
-        # if queue_name not in {'my_course_queue': 'http://172.16.0.220:8125', 'test-pull': None}:
-        #    return Response(
-        #        {'success': False, 'message': f"Queue '{queue_name}' not found"},
-        #        status=status.HTTP_404_NOT_FOUND
-        #    )
-
-        submission_record = ExternalGraderDetail.objects.filter(
-            queue_name=queue_name,
-            status__in=['pending']
-        ).select_related('submission').order_by('status_time').first()
-
-        if not submission_record:
+        timeout_threshold = timezone.now() - timedelta(minutes=5)
+        try:
+            submission_record = (
+                ExternalGraderDetail.objects
+                .select_for_update(nowait=True)
+                .filter(
+                    Q(queue_name=queue_name, status='pending') |
+                    Q(queue_name=queue_name, status='pulled', status_time__lt=timeout_threshold)
+                )
+                .select_related('submission')
+                .order_by('status_time')
+                .first()
+            )
+        except DatabaseError:
             return Response(
-                self.compose_reply(False, f"Queue '{queue_name}' is empty"),
-                status=status.HTTP_404_NOT_FOUND
+                self.compose_reply(False, "Submission already in process"),
+                status=status.HTTP_409_CONFLICT
             )
 
-        try:
-            pull_time = timezone.now()
-            pullkey = str(uuid.uuid4())
-            # grader_id = request.META.get('REMOTE_ADDR', '')
-            submission_record.update_status('pulled')
-            submission_record.pullkey = pullkey
-            submission_record.status_time = pull_time
-            submission_record.save(update_fields=['pullkey', 'status_time'])
-
-            ext_header = {
-                'submission_id': submission_record.submission.id,
-                'submission_key': pullkey
-            }
-            answer = submission_record.submission.answer
-            submission_data = {
-                "grader_payload": json.dumps({
-                    "grader": submission_record.grader_file_name
-                }),
-                "student_info": json.dumps({
-                    "anonymous_student_id": str(submission_record.submission.student_item.student_id,),
-                    "submission_time": str(int(submission_record.created_at.timestamp())),
-                    "random_seed": 1
-                }),
-                "student_response": answer
-            }
-
-            file_manager = SubmissionFileManager(submission_record)
-
-            payload = {
-                'xqueue_header': json.dumps(ext_header),
-                'xqueue_body': json.dumps(submission_data),
-                'xqueue_files': json.dumps(file_manager.get_files_for_grader())
-            }
+        if submission_record:
+            try:
+                pull_time = timezone.now()
+                pullkey = str(uuid.uuid4())
+                submission_record.update_status('pulled')
+                submission_record.pullkey = pullkey
+                submission_record.status_time = pull_time
+                submission_record.save(update_fields=['pullkey', 'status_time'])
+
+                submission_data = {
+                    "grader_payload": json.dumps({"grader": submission_record.grader_file_name}),
+                    "student_info": json.dumps({
+                        "anonymous_student_id": str(submission_record.submission.student_item.student_id),
+                        "submission_time": str(int(submission_record.created_at.timestamp())),
+                        "random_seed": 1
+                    }),
+                    "student_response": submission_record.submission.answer
+                }
+
+                file_manager = SubmissionFileManager(submission_record)
+
+                payload = {
+                    'xqueue_header': json.dumps({
+                        'submission_id': submission_record.submission.id,
+                        'submission_key': pullkey
+                    }),
+                    'xqueue_body': json.dumps(submission_data),
+                    'xqueue_files': json.dumps(file_manager.get_files_for_grader())
+                }
+
+                return Response(
+                    self.compose_reply(True, content=json.dumps(payload)),
+                    status=status.HTTP_200_OK
+                )
 
-            return Response(
-                self.compose_reply(True, content=json.dumps(payload)),
-                status=status.HTTP_200_OK
-            )
+            except ValueError as e:
+                return Response(
+                    self.compose_reply(False, f"Error processing submission: {str(e)}"),
+                    status=status.HTTP_400_BAD_REQUEST
+                )
 
-        except ValueError as e:
-            return Response(
-                self.compose_reply(False, f"Error processing submission: {str(e)}"),
-                status=status.HTTP_400_BAD_REQUEST
-            )
+        return Response(
+            self.compose_reply(False, f"Queue '{queue_name}' is empty"),
+            status=status.HTTP_404_NOT_FOUND
+        )
 
     @action(detail=False, methods=['post'], url_name='put_result')
     @transaction.atomic