chore: address feedback on PR

Author: Taimoor Ahmed

submissions/migrations/0006_externalgraderdetail_queue_key_and_more.py

@@ -0,0 +1,23 @@
+# Generated by Django 4.2.25 on 2025-11-09 10:02
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('submissions', '0007_merge_20251020_2045'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='externalgraderdetail',
+            name='queue_key',
+            field=models.CharField(max_length=128, null=True),
+        ),
+        migrations.AlterField(
+            model_name='externalgraderdetail',
+            name='status',
+            field=models.CharField(choices=[('pending', 'Pending'), ('pulled', 'Pulled'), ('retired', 'Retired'), ('failed', 'Failed'), ('retry', 'Retry')], default='pending', max_length=20),
+        ),
+    ]

submissions/migrations/0008_externalgraderdetail_queue_key_and_more.py

@@ -226,7 +226,17 @@ class Meta:
 class SubmissionListSerializer(serializers.ModelSerializer):
     class Meta:
         model = Submission
-        fields = '__all__'
+        fields = (
+            'id',
+            'uuid',
+            'student_item',
+            'attempt_number',
+            'submitted_at',
+            'created_at',
+            'answer',
+            'status',
+            'team_submission'
+        )
 
 
 class ExternalGraderDetailSerializer(serializers.ModelSerializer):
@@ -235,4 +245,17 @@ class ExternalGraderDetailSerializer(serializers.ModelSerializer):
 
     class Meta:
         model = ExternalGraderDetail
-        fields = '__all__'
+        fields = (
+            'id',
+            'submission',
+            'queue_name',
+            'queue_key',
+            'grader_file_name',
+            'points_possible',
+            'status',
+            'pullkey',
+            'grader_reply',
+            'status_time',
+            'created_at',
+            'num_failures'
+        )

submissions/serializers.py

@@ -5,7 +5,7 @@
 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
@@ -24,6 +24,9 @@
 log = logging.getLogger(__name__)
 
 
+MAX_SCORE_UPDATE_RETRIES = 5
+
+
 class XQueueViewSet(viewsets.ViewSet):
     """
     A collection of services for xqueue-watcher interactions and authentication.
@@ -151,9 +154,23 @@ def get_submission(self, request):
         )
 
         if external_grader:
-            try:
-                if external_grader.status != 'pulled':
+            submission_id = external_grader.submission.id
+            if external_grader.status != 'pulled':
+                try:
                     external_grader.update_status("pulled")
+                except ValueError as error:
+                    log.error(
+                        "Invalid status transition for submission %s: current=%s target='pulled'",
+                        submission_id,
+                        external_grader.status,
+                        exc_info=error
+                    )
+                    return Response(
+                        self.compose_reply(False, 'Unable to transition submission for grading'),
+                        status=status.HTTP_409_CONFLICT
+                    )
+
+            try:
                 submission_data = {
                     "grader_payload": json.dumps({"grader": external_grader.grader_file_name}),
                     "student_info": json.dumps({
@@ -166,24 +183,29 @@ def get_submission(self, request):
 
                 payload = {
                     'xqueue_header': json.dumps({
-                        'submission_id': external_grader.submission.id,
+                        'submission_id': submission_id,
                         'submission_key': external_grader.pullkey
                     }),
                     'xqueue_body': json.dumps(submission_data),
                     # Xqueue watcher expects this to be a JSON string
                     'xqueue_files': json.dumps(get_files_for_grader(external_grader))
                 }
-
-                return Response(
-                    self.compose_reply(True, content=json.dumps(payload)),
-                    status=status.HTTP_200_OK
+            except (TypeError, ValueError) as error:
+                log.error(
+                    "Unable to serialize submission payload for submission %s: %s",
+                    submission_id,
+                    error
                 )
-            except ValueError as e:
                 return Response(
-                    self.compose_reply(False, f"Error processing submission: {str(e)}"),
-                    status=status.HTTP_400_BAD_REQUEST
+                    self.compose_reply(False, 'Unable to serialize submission payload'),
+                    status=status.HTTP_500_INTERNAL_SERVER_ERROR
                 )
 
+            return Response(
+                self.compose_reply(True, content=json.dumps(payload)),
+                status=status.HTTP_200_OK
+            )
+
         return Response(
             self.compose_reply(False, f"Queue '{queue_name}' is empty"),
             status=status.HTTP_200_OK
@@ -207,8 +229,14 @@ def put_result(self, request):
             )
 
         try:
-            external_grader = ExternalGraderDetail.objects.select_for_update(
-                                                                        nowait=True).get(submission__id=submission_id)
+            # Lock the row to prevent duplicate grading attempts from racing and emitting
+            # duplicate score events for the same submission.
+            external_grader = (
+                ExternalGraderDetail.objects
+                .select_for_update(nowait=True)
+                .select_related('submission__student_item')
+                .get(submission__id=submission_id)
+            )
         except ExternalGraderDetail.DoesNotExist:
             log.error(
                 "Grader submission_id refers to nonexistent entry in Submission DB: "
@@ -221,6 +249,15 @@ def put_result(self, request):
                 self.compose_reply(False, 'Submission does not exist'),
                 status=status.HTTP_404_NOT_FOUND
             )
+        except DatabaseError:
+            log.warning(
+                "Concurrent grade update detected for submission %s; rejecting duplicate request",
+                submission_id
+            )
+            return Response(
+                self.compose_reply(False, 'Submission is currently being processed'),
+                status=status.HTTP_409_CONFLICT
+            )
 
         if not external_grader.pullkey or submission_key != external_grader.pullkey:
             log.error(
@@ -235,39 +272,59 @@ def put_result(self, request):
             )
 
         # pylint: disable=broad-exception-caught
+        submission_context = {
+            'submission_id': submission_id,
+            'course_id': external_grader.submission.student_item.course_id,
+            'user_id': external_grader.submission.student_item.student_id,
+            'item_id': external_grader.submission.student_item.item_id,
+            'queue_name': external_grader.queue_name,
+            'queue_key': external_grader.queue_key,
+        }
+
         try:
-            log.info("Attempting to set_score...")
-            set_score(str(external_grader.submission.uuid),
-                      points_earned,
-                      external_grader.points_possible
-                      )
+            log.info(
+                "Attempting to record score for submission %(submission_id)s (course=%(course_id)s, item=%(item_id)s, user=%(user_id)s)",
+                submission_context
+            )
+            set_score(
+                str(external_grader.submission.uuid),
+                points_earned,
+                external_grader.points_possible
+            )
             external_grader.grader_reply = score_msg
-            external_grader.save()
+            external_grader.save(update_fields=['grader_reply'])
             external_grader.update_status('retired')
-            log.info("Successfully updated submission score for submission %s", submission_id)
+            log.info(
+                "Successfully updated submission %(submission_id)s for user %(user_id)s",
+                submission_context
+            )
 
-            # Modify the event emission in put_result
             EXTERNAL_GRADER_SCORE_SUBMITTED.send_event(
                 send_robust=False,
                 score=ExternalGraderScoreData(
                     points_possible=external_grader.points_possible,
                     points_earned=points_earned,
-                    course_id=external_grader.submission.student_item.course_id,
+                    course_id=submission_context['course_id'],
                     score_msg=score_msg,
                     submission_id=submission_id,
-                    # Extract these from the submission
-                    user_id=external_grader.submission.student_item.student_id,
-                    module_id=external_grader.submission.student_item.item_id,
-                    queue_key=external_grader.queue_key,
-                    queue_name=external_grader.queue_name
+                    user_id=submission_context['user_id'],
+                    module_id=submission_context['item_id'],
+                    queue_key=submission_context['queue_key'],
+                    queue_name=submission_context['queue_name']
                 )
             )
-            log.info("Score event sent to bus successfully")
+            log.info(
+                "External grader score event emitted for submission %(submission_id)s (course=%(course_id)s, item=%(item_id)s, user=%(user_id)s, queue=%(queue_name)s, queue_key=%(queue_key)s)",
+                submission_context
+            )
 
         except Exception:
-            log.exception("Error when execute set_score")
+            log.exception(
+                "Error recording score for submission %(submission_id)s (course=%(course_id)s, item=%(item_id)s, user=%(user_id)s, queue=%(queue_name)s)",
+                submission_context
+            )
             # Keep track of how many times we've failed to set_score a grade for this submission
-            if external_grader.num_failures >= 30:
+            if external_grader.num_failures + 1 >= MAX_SCORE_UPDATE_RETRIES:
                 external_grader.update_status('failed')
             else:
                 external_grader.update_status('retry')