diff --git a/apps/challenges/migrations/0128_challenge_submission_queue_time_limit.py b/apps/challenges/migrations/0128_challenge_submission_queue_time_limit.py new file mode 100644 index 0000000000..8eeb955da2 --- /dev/null +++ b/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." + ), + ), + ), + ] diff --git a/apps/challenges/models.py b/apps/challenges/models.py index 1b0cb0656a..16a310e51f 100644 --- a/apps/challenges/models.py +++ b/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) diff --git a/apps/challenges/serializers.py b/apps/challenges/serializers.py index 90fbf3bcba..1c543c6641 100644 --- a/apps/challenges/serializers.py +++ b/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", diff --git a/apps/challenges/views.py b/apps/challenges/views.py index 3392d35900..a9707e4dfa 100644 --- a/apps/challenges/views.py +++ b/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(): diff --git a/apps/jobs/management/commands/fail_stuck_submissions.py b/apps/jobs/management/commands/fail_stuck_submissions.py new file mode 100644 index 0000000000..98a21281c9 --- /dev/null +++ b/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) + ) + ) diff --git a/apps/jobs/queue_timeout.py b/apps/jobs/queue_timeout.py new file mode 100644 index 0000000000..cf453c2d0e --- /dev/null +++ b/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 diff --git a/apps/jobs/tasks.py b/apps/jobs/tasks.py index caaedb2412..138708ef86 100644 --- a/apps/jobs/tasks.py +++ b/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() diff --git a/settings/common.py b/settings/common.py index 62ea1987b3..fb0ddf7ad0 100755 --- a/settings/common.py +++ b/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 diff --git a/tests/unit/jobs/test_queue_timeout.py b/tests/unit/jobs/test_queue_timeout.py new file mode 100644 index 0000000000..ca186b4184 --- /dev/null +++ b/tests/unit/jobs/test_queue_timeout.py @@ -0,0 +1,234 @@ +import os +import shutil +from datetime import timedelta +from unittest.mock import patch + +from challenges.models import Challenge, ChallengePhase +from django.contrib.auth.models import User +from django.core.files.uploadedfile import SimpleUploadedFile +from django.test import TestCase +from django.utils import timezone +from hosts.models import ChallengeHostTeam +from jobs.models import Submission +from jobs.queue_timeout import ( + fail_stuck_submissions, + get_stuck_submission_queryset, + is_submission_queue_timed_out, +) +from jobs.tasks import fail_stuck_submissions_task +from participants.models import ParticipantTeam + + +class QueueTimeoutBaseTestCase(TestCase): + def setUp(self): + self.user = User.objects.create( + username="user", email="user@test.com", password="password" + ) + self.participant_team = ParticipantTeam.objects.create( + team_name="Participant Team", created_by=self.user + ) + self.challenge_host_team = ChallengeHostTeam.objects.create( + team_name="Test Challenge Host Team", created_by=self.user + ) + self.challenge = Challenge.objects.create( + title="Test Challenge", + description="Description for test challenge", + terms_and_conditions="Terms and conditions for test challenge", + submission_guidelines="Submission guidelines for test challenge", + creator=self.challenge_host_team, + start_date=timezone.now() - timedelta(days=2), + end_date=timezone.now() + timedelta(days=1), + published=False, + enable_forum=True, + anonymous_leaderboard=False, + submission_queue_time_limit=7200, + ) + + try: + os.makedirs("/tmp/evalai") + except OSError: + pass + + with self.settings(MEDIA_ROOT="/tmp/evalai"): + self.challenge_phase = ChallengePhase.objects.create( + name="Challenge Phase", + description="Description for Challenge Phase", + leaderboard_public=False, + is_public=True, + start_date=timezone.now() - timedelta(days=2), + end_date=timezone.now() + timedelta(days=1), + challenge=self.challenge, + test_annotation=SimpleUploadedFile( + "test_sample_file.txt", + b"Dummy file content", + content_type="text/plain", + ), + ) + + def tearDown(self): + shutil.rmtree("/tmp/evalai", ignore_errors=True) + + def _create_submission(self, status=Submission.SUBMITTED, submitted_at=None): + with self.settings(MEDIA_ROOT="/tmp/evalai"): + submission = Submission.objects.create( + participant_team=self.participant_team, + challenge_phase=self.challenge_phase, + created_by=self.user, + status=status, + input_file=self.challenge_phase.test_annotation, + is_public=True, + ) + if submitted_at is not None: + Submission.objects.filter(pk=submission.pk).update( + submitted_at=submitted_at + ) + submission.refresh_from_db() + return submission + + +class TestIsSubmissionQueueTimedOut(QueueTimeoutBaseTestCase): + def test_returns_false_when_within_limit(self): + submission = self._create_submission( + submitted_at=timezone.now() - timedelta(hours=1) + ) + + self.assertFalse(is_submission_queue_timed_out(submission)) + + def test_returns_true_when_past_limit(self): + submission = self._create_submission( + submitted_at=timezone.now() - timedelta(hours=3) + ) + + self.assertTrue(is_submission_queue_timed_out(submission)) + + def test_returns_false_when_disabled_for_challenge(self): + self.challenge.submission_queue_time_limit = 0 + self.challenge.save() + submission = self._create_submission( + submitted_at=timezone.now() - timedelta(hours=10) + ) + + self.assertFalse(is_submission_queue_timed_out(submission)) + + def test_uses_rerun_resumed_at_when_present(self): + submission = self._create_submission( + submitted_at=timezone.now() - timedelta(hours=10) + ) + resumed_at = timezone.now() - timedelta(minutes=30) + Submission.objects.filter(pk=submission.pk).update( + rerun_resumed_at=resumed_at + ) + submission.refresh_from_db() + + self.assertFalse(is_submission_queue_timed_out(submission)) + + +class TestFailStuckSubmissions(QueueTimeoutBaseTestCase): + @patch("jobs.queue_timeout.trigger_eks_node_autoscale") + def test_fails_old_submitted_submission(self, mock_autoscale): + submission = self._create_submission( + submitted_at=timezone.now() - timedelta(hours=3) + ) + + with self.settings(MEDIA_ROOT="/tmp/evalai"): + failed_count = fail_stuck_submissions() + + self.assertEqual(failed_count, 1) + submission.refresh_from_db() + self.assertEqual(submission.status, Submission.FAILED) + self.assertIsNotNone(submission.completed_at) + self.assertTrue(submission.stderr_file) + mock_autoscale.assert_called_once() + + @patch("jobs.queue_timeout.trigger_eks_node_autoscale") + def test_skips_recent_submission(self, mock_autoscale): + self._create_submission(submitted_at=timezone.now() - timedelta(minutes=30)) + + failed_count = fail_stuck_submissions() + + self.assertEqual(failed_count, 0) + mock_autoscale.assert_not_called() + + @patch("jobs.queue_timeout.trigger_eks_node_autoscale") + def test_dry_run_does_not_update_submission(self, mock_autoscale): + submission = self._create_submission( + submitted_at=timezone.now() - timedelta(hours=3) + ) + + failed_count = fail_stuck_submissions(dry_run=True) + + self.assertEqual(failed_count, 1) + submission.refresh_from_db() + self.assertEqual(submission.status, Submission.SUBMITTED) + mock_autoscale.assert_not_called() + + @patch("jobs.queue_timeout.trigger_eks_node_autoscale") + def test_filters_by_challenge_pk(self, mock_autoscale): + other_challenge = Challenge.objects.create( + title="Other Challenge", + description="Description", + terms_and_conditions="Terms", + submission_guidelines="Guidelines", + creator=self.challenge_host_team, + start_date=timezone.now() - timedelta(days=2), + end_date=timezone.now() + timedelta(days=1), + submission_queue_time_limit=7200, + ) + with self.settings(MEDIA_ROOT="/tmp/evalai"): + other_phase = ChallengePhase.objects.create( + name="Other Phase", + description="Description", + leaderboard_public=False, + is_public=True, + start_date=timezone.now() - timedelta(days=2), + end_date=timezone.now() + timedelta(days=1), + challenge=other_challenge, + test_annotation=SimpleUploadedFile( + "other.txt", b"content", content_type="text/plain" + ), + ) + other_submission = Submission.objects.create( + participant_team=self.participant_team, + challenge_phase=other_phase, + created_by=self.user, + status=Submission.SUBMITTED, + input_file=other_phase.test_annotation, + ) + Submission.objects.filter(pk=other_submission.pk).update( + submitted_at=timezone.now() - timedelta(hours=3) + ) + + stuck_submission = self._create_submission( + submitted_at=timezone.now() - timedelta(hours=3) + ) + + with self.settings(MEDIA_ROOT="/tmp/evalai"): + failed_count = fail_stuck_submissions(challenge_pk=self.challenge.pk) + + self.assertEqual(failed_count, 1) + stuck_submission.refresh_from_db() + other_submission.refresh_from_db() + self.assertEqual(stuck_submission.status, Submission.FAILED) + self.assertEqual(other_submission.status, Submission.SUBMITTED) + + def test_get_stuck_submission_queryset_includes_queue_states(self): + submitted = self._create_submission(status=Submission.SUBMITTED) + queued = self._create_submission(status=Submission.QUEUED) + running = self._create_submission(status=Submission.RUNNING) + + stuck_pks = set( + get_stuck_submission_queryset().values_list("pk", flat=True) + ) + + self.assertIn(submitted.pk, stuck_pks) + self.assertIn(queued.pk, stuck_pks) + self.assertNotIn(running.pk, stuck_pks) + + +class TestFailStuckSubmissionsTask(QueueTimeoutBaseTestCase): + @patch("jobs.queue_timeout.fail_stuck_submissions", return_value=2) + def test_task_delegates_to_helper(self, mock_fail): + result = fail_stuck_submissions_task() + + self.assertEqual(result, 2) + mock_fail.assert_called_once_with()