Add queue timeout for stuck pre-running submissions

apps/challenges/migrations/0128_challenge_submission_queue_time_limit.py

@@ -0,0 +1,26 @@
+# Generated by Django 2.2.20 on 2026-07-09
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("challenges", "0127_alter_challenge_worker_python_version_choices"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="challenge",
+            name="submission_queue_time_limit",
+            field=models.PositiveIntegerField(
+                default=7200,
+                help_text=(
+                    "Maximum seconds a submission may remain in a pre-running "
+                    "queue state (submitted, submitting, queued, resuming) "
+                    "before it is automatically marked failed. Set to 0 to "
+                    "disable."
+                ),
+            ),
+        ),
+    ]

apps/challenges/models.py

@@ -94,6 +94,14 @@ def __init__(self, *args, **kwargs):
         default=False, verbose_name="Publicly Available", db_index=True
     )
     submission_time_limit = models.PositiveIntegerField(default=86400)
+    submission_queue_time_limit = models.PositiveIntegerField(
+        default=7200,
+        help_text=(
+            "Maximum seconds a submission may remain in a pre-running queue "
+            "state (submitted, submitting, queued, resuming) before it is "
+            "automatically marked failed. Set to 0 to disable."
+        ),
+    )
     is_registration_open = models.BooleanField(default=True)
     enable_forum = models.BooleanField(default=True)
     forum_url = models.URLField(max_length=100, blank=True, null=True)

apps/challenges/serializers.py

@@ -96,6 +96,7 @@ class Meta:
             "has_sponsors",
             "published",
             "submission_time_limit",
+            "submission_queue_time_limit",
             "is_registration_open",
             "enable_forum",
             "anonymous_leaderboard",
@@ -356,6 +357,7 @@ class Meta:
             "published",
             "challenge_usage_type",
             "submission_time_limit",
+            "submission_queue_time_limit",
             "is_registration_open",
             "enable_forum",
             "anonymous_leaderboard",

apps/challenges/views.py

@@ -5566,6 +5566,7 @@ def update_challenge_attributes(request):
         "start_date",
         "max_concurrent_submission_evaluation",
         "sqs_retention_period",
+        "submission_queue_time_limit",
     }
 
     for key, value in request.data.items():

apps/jobs/management/commands/fail_stuck_submissions.py

@@ -0,0 +1,44 @@
+from django.core.management.base import BaseCommand
+
+from jobs.queue_timeout import fail_stuck_submissions
+
+
+class Command(BaseCommand):
+    help = (
+        "Fail submissions that have remained in a pre-running queue state "
+        "(submitted, submitting, queued, resuming) longer than the "
+        "challenge's submission_queue_time_limit."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--challenge-pk",
+            type=int,
+            default=None,
+            help="Only process submissions for this challenge.",
+        )
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Report submissions that would be failed without updating them.",
+        )
+
+    def handle(self, *args, **options):
+        failed_count = fail_stuck_submissions(
+            challenge_pk=options["challenge_pk"],
+            dry_run=options["dry_run"],
+        )
+        if options["dry_run"]:
+            self.stdout.write(
+                self.style.WARNING(
+                    "Dry run: {} submission(s) would be failed.".format(
+                        failed_count
+                    )
+                )
+            )
+        else:
+            self.stdout.write(
+                self.style.SUCCESS(
+                    "Failed {} queue-stuck submission(s).".format(failed_count)
+                )
+            )

apps/jobs/queue_timeout.py

@@ -0,0 +1,116 @@
+import logging
+from datetime import timedelta
+
+from challenges.aws_utils import trigger_eks_node_autoscale
+from django.core.files.base import ContentFile
+from django.db.models.functions import Coalesce
+from django.utils import timezone
+
+from .models import Submission
+
+logger = logging.getLogger(__name__)
+
+STUCK_QUEUE_STATUSES = (
+    Submission.SUBMITTED,
+    Submission.SUBMITTING,
+    Submission.QUEUED,
+    Submission.RESUMING,
+)
+
+QUEUE_TIMEOUT_STDERR = (
+    "Submission exceeded the queue time limit and was not picked up for "
+    "evaluation."
+)
+
+
+def get_stuck_submission_queryset(challenge_pk=None):
+    """Return submissions in pre-running states that may be queue-stuck."""
+    queryset = (
+        Submission.objects.filter(
+            status__in=STUCK_QUEUE_STATUSES,
+            ignore_submission=False,
+            challenge_phase__challenge__submission_queue_time_limit__gt=0,
+        )
+        .select_related("challenge_phase__challenge")
+        .annotate(queue_started_at=Coalesce("rerun_resumed_at", "submitted_at"))
+    )
+    if challenge_pk is not None:
+        queryset = queryset.filter(challenge_phase__challenge__pk=challenge_pk)
+    return queryset
+
+
+def is_submission_queue_timed_out(submission, now=None):
+    """Return True when a submission has exceeded its challenge queue limit."""
+    now = now or timezone.now()
+    challenge = submission.challenge_phase.challenge
+    queue_time_limit = challenge.submission_queue_time_limit
+    if not queue_time_limit:
+        return False
+
+    queue_started_at = submission.rerun_resumed_at or submission.submitted_at
+    return queue_started_at <= now - timedelta(seconds=queue_time_limit)
+
+
+def fail_submission_due_to_queue_timeout(submission, dry_run=False):
+    """Mark a queue-stuck submission as failed."""
+    if dry_run:
+        return True
+
+    previous_status = submission.status
+    submission.status = Submission.FAILED
+    submission.completed_at = timezone.now()
+    submission.stderr_file.save(
+        "stderr.txt",
+        ContentFile(QUEUE_TIMEOUT_STDERR.encode("utf-8")),
+    )
+    submission.save()
+
+    challenge = submission.challenge_phase.challenge
+    trigger_eks_node_autoscale(
+        challenge.pk,
+        trigger_source="submission_queue_timeout",
+        submission_pk=submission.pk,
+        submission_status=Submission.FAILED,
+        previous_submission_status=previous_status,
+    )
+    return True
+
+
+def fail_stuck_submissions(challenge_pk=None, dry_run=False):
+    """
+    Fail submissions that exceeded their challenge queue time limit.
+
+    Returns the number of submissions that were (or would be) failed.
+    """
+    now = timezone.now()
+    failed_count = 0
+
+    for submission in get_stuck_submission_queryset(challenge_pk):
+        if not is_submission_queue_timed_out(submission, now=now):
+            continue
+
+        previous_status = submission.status
+        queue_started_at = submission.rerun_resumed_at or submission.submitted_at
+        challenge = submission.challenge_phase.challenge
+
+        if fail_submission_due_to_queue_timeout(submission, dry_run=dry_run):
+            failed_count += 1
+            action = "Would fail" if dry_run else "Failed"
+            logger.warning(
+                "%s queue-stuck submission pk=%s challenge_pk=%s status=%s "
+                "queue_started_at=%s queue_time_limit=%ss",
+                action,
+                submission.pk,
+                challenge.pk,
+                previous_status,
+                queue_started_at,
+                challenge.submission_queue_time_limit,
+            )
+
+    if failed_count:
+        logger.info(
+            "fail_stuck_submissions: %s %d submission(s).",
+            "would fail" if dry_run else "failed",
+            failed_count,
+        )
+    return failed_count

apps/jobs/tasks.py

@@ -125,3 +125,11 @@ def tag_submission_artifact_retention_tags(submission_pk, artifact_paths):
     except Submission.DoesNotExist:
         return
     tag_submission_artifacts_for_retention(submission, artifact_paths)
+
+
+@app.task
+def fail_stuck_submissions_task():
+    """Periodic task to fail submissions stuck in pre-running queue states."""
+    from .queue_timeout import fail_stuck_submissions
+
+    return fail_stuck_submissions()

settings/common.py

@@ -257,6 +257,10 @@
         # to "America/Los_Angeles" if year-round 10:30 Pacific is required.
         "schedule": crontab(hour=18, minute=30),
     },
+    "fail-stuck-submissions": {
+        "task": "jobs.tasks.fail_stuck_submissions_task",
+        "schedule": datetime.timedelta(minutes=30),
+    },
 }
 
 # CORS Settings