diff --git a/src/live_cycle.py b/src/live_cycle.py index 7c1aacd..9371111 100644 --- a/src/live_cycle.py +++ b/src/live_cycle.py @@ -12,7 +12,36 @@ from .pending_pool import build_ordered_candidates from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source, token_probe_status from .ledger import load_baked, load_persisted, persist, rebuild +REJECT_CODES = { + 60006: 'CANNOT_CREATE_CONTEST_TASK_BECAUSE_NOT_IN_WHITE_LIST', + 60007: 'MODEL_VERIFY_TASK_NUM_LIMIT', + 60008: 'CREATE_CONTEST_TASK_REQUEST_EMPTY', + 60009: 'CREATE_CONTEST_TASK_CONTRIBUTORS_EMPTY', + 60010: 'CREATE_CONTEST_TASK_SUBMISSION_CONFIG_EMPTY', + 60011: 'CREATE_CONTEST_TASK_GPU_TYPE_EMPTY', + 60012: 'CREATE_CONTEST_TASK_MULTI_GPU_UNSUPPORTED', + 60013: 'CREATE_CONTEST_TASK_MODEL_ID_EMPTY', + 60014: 'CREATE_CONTEST_TASK_SCRIPT_SUBMIT_DISABLED', + 60015: 'CREATE_CONTEST_TASK_TASK_TYPE_EMPTY', + 60016: 'CREATE_CONTEST_TASK_FRAMEWORK_EMPTY', + 60017: 'CREATE_CONTEST_TASK_DRIVER_IMAGE_NOT_FOUND', + 60018: 'CREATE_CONTEST_TASK_CONTRIBUTOR_NOT_FOUND', + 60019: 'CREATE_CONTEST_TASK_SUBMISSION_GPU_TYPE_MISMATCH', + 60020: 'CREATE_CONTEST_TASK_MODEL_NOT_DOWNLOADED', + 60021: 'CREATE_CONTEST_TASK_FRAMEWORK_UNRESOLVED', + 60022: 'CREATE_CONTEST_TASK_GPU_CONFIG_UNSUPPORTED', + 60023: 'CREATE_CONTEST_TASK_YAML_CONFIG_EMPTY', + 60024: 'CREATE_CONTEST_TASK_YAML_CONFIG_INVALID', + 60025: 'CREATE_CONTEST_TASK_GPU_TYPE_UNSUPPORTED', + 60026: 'CREATE_CONTEST_TASK_MODEL_NOT_FOUND', + 60027: 'CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED', + 60028: 'CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING', + 60029: 'CREATE_CONTEST_TASK_TRANSFORMERS_REQUIRES_FAILED_BASELINE', + 60030: 'CREATE_CONTEST_TASK_API_VERSION_UNSUPPORTED', +} +_DONE_CODES = {60027, 60028} _submitted = set() +_rejected = set() _cycle_lock = threading.Lock() @@ -50,6 +79,12 @@ def _run_cycle(candidates, client, queue_reader, now): except ValueError: emit('hold', 'invalid_cycle_cap') return decisions + try: + attempt_cap = int(os.environ.get('LIVE_MAX_ATTEMPTS_PER_CYCLE', '15')) + if attempt_cap < 1: + raise ValueError + except ValueError: + attempt_cap = 15 if candidates is None: try: if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'): @@ -67,13 +102,14 @@ def _run_cycle(candidates, client, queue_reader, now): try: persisted = load_persisted(state) | _submitted baked, blocked = load_baked() - known = persisted | baked + known = persisted | baked | _rejected except (OSError, ValueError, KeyError, TypeError): emit('hold', 'state_read_failed') return decisions reader = queue_reader or read_queue client = client if client is not None else TaskAddClient() attempts = 0 + counted = 0 for candidate in candidates: model = candidate.get('model_address') or candidate.get('modelAddress') or '' rule = banned_rule(model + ' ' + str(candidate.get('model_name', candidate.get('modelName', '')))) @@ -90,7 +126,7 @@ def _run_cycle(candidates, client, queue_reader, now): if candidate.get('skip_reason'): emit('skip', 'pending_filter', model=model) continue - if attempts >= cap: + if counted >= cap: emit('hold', 'cycle_cap') break choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')] @@ -142,6 +178,29 @@ def _run_cycle(candidates, client, queue_reader, now): reason = 'task_add_response' if status is not None else 'transport_error' except Exception: status, code, reason = None, None, 'submit_error' + if code is not None and code != 0: + emit('rejected', REJECT_CODES.get(code, 'unknown'), model=model, + gpu=GPU_ENUMS[gpu], code=code, http=status, queue_before=count) + if code not in _DONE_CODES: + _rejected.add(pair) + persisted.discard(pair) + _submitted.discard(pair) + try: + persist(state, persisted) + except OSError: + emit('hold', 'state_write_failed') + break + if code == 60007: + emit('hold', 'task_num_limit') + break + if attempts >= attempt_cap: + emit('hold', 'attempt_cap') + break + continue + counted += 1 emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count, http=status, code=code, **meta) + if attempts >= attempt_cap: + emit('hold', 'attempt_cap') + break return decisions diff --git a/tests/test_live_cycle.py b/tests/test_live_cycle.py index 984c72a..c60863d 100644 --- a/tests/test_live_cycle.py +++ b/tests/test_live_cycle.py @@ -1,9 +1,11 @@ import io +import json +from pathlib import Path import os import tempfile import unittest from unittest.mock import Mock, patch -from src.live_cycle import run_cycle, _submitted +from src.live_cycle import run_cycle, _submitted, _rejected def candidate(name, section='B', foothold='unknown'): @@ -18,6 +20,7 @@ class TestLiveCycle(unittest.TestCase): self.env.start() self.addCleanup(self.env.stop) _submitted.clear() + _rejected.clear() self.client = Mock() self.client.submit.return_value = dict(http_status=200, platform_code=0) network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden')) @@ -31,6 +34,80 @@ class TestLiveCycle(unittest.TestCase): reader = Mock(side_effect=[(n, 'platform_error' if n is None else 'ok', set()) for n in counts]) return run_cycle(rows or [candidate('a')], self.client, reader) + def state_pairs(self): + return json.loads(Path(os.environ['LIVE_STATE_FILE']).read_text()) + + def test_done_rejection_keeps_reservation_and_continues(self): + for code in (60027, 60028): + with self.subTest(code=code): + self.client.submit.side_effect = [ + dict(http_status=200, platform_code=code), + dict(http_status=200, platform_code=0)] + with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}): + decisions = self.cycle([candidate(str(code)), candidate('ok' + str(code))], (1, 1)) + self.assertEqual([d['action'] for d in decisions], ['rejected', 'submitted']) + self.assertIn('org/' + str(code) + '|Biren_166m', self.state_pairs()) + self.assertIn('rejected', self.logs.getvalue()) + self.assertIn('CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING', self.logs.getvalue()) + + def test_rejection_removed_but_not_retried_in_process(self): + self.client.submit.return_value = dict(http_status=200, platform_code=60020) + self.cycle() + self.assertEqual(self.state_pairs(), []) + self.cycle(counts=()) + self.client.submit.assert_called_once() + + def test_unknown_rejection(self): + self.client.submit.return_value = dict(http_status=200, platform_code=99999) + decisions = self.cycle() + self.assertEqual(decisions[0]['action'], 'rejected') + self.assertEqual(decisions[0]['reason'], 'unknown') + self.assertEqual(self.state_pairs(), []) + + def test_attempt_cap_and_invalid_fallback(self): + for value, expected in ((None, 15), ('bad', 15), ('0', 15), ('-1', 15), ('2', 2)): + with self.subTest(value=value): + _rejected.clear() + self.client.reset_mock() + self.client.submit.return_value = dict(http_status=200, platform_code=60020) + with patch.dict(os.environ, {} if value is None else {'LIVE_MAX_ATTEMPTS_PER_CYCLE': value}): + decisions = self.cycle([candidate(str(i)) for i in range(expected)], (1,) * expected) + self.assertEqual(self.client.submit.call_count, expected) + self.assertEqual(decisions[-1]['reason'], 'attempt_cap') + + def test_submit_cap_counts_successes_after_rejections(self): + self.client.submit.side_effect = [dict(http_status=200, platform_code=c) for c in (60020, 0, 60028, 0, 0)] + decisions = self.cycle([candidate(str(i)) for i in range(7)], (1,) * 5) + self.assertEqual(self.client.submit.call_count, 5) + self.assertEqual(sum(d['action'] == 'submitted' for d in decisions), 3) + self.assertEqual(decisions[-1]['reason'], 'cycle_cap') + + def test_task_limit_stops_cycle(self): + self.client.submit.return_value = dict(http_status=200, platform_code=60007) + decisions = self.cycle([candidate('a'), candidate('b')]) + self.assertEqual([d['action'] for d in decisions], ['rejected', 'hold']) + self.assertEqual(decisions[-1]['reason'], 'task_num_limit') + self.assertEqual(self.state_pairs(), []) + self.client.submit.assert_called_once() + + def test_unknown_outcomes_count_and_reserve_before_submit(self): + for result in (dict(http_status=200, platform_code='0'), dict(http_status=None, platform_code=None), OSError()): + with self.subTest(result=result): + _submitted.clear() + Path(os.environ['LIVE_STATE_FILE']).unlink(missing_ok=True) + self.client.reset_mock() + def submit(payload): + self.assertIn('org/a|Biren_166m', self.state_pairs()) + if isinstance(result, Exception): + raise result + return result + self.client.submit.side_effect = submit + with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}): + decisions = self.cycle([candidate('a'), candidate('b')]) + self.assertEqual(decisions[0]['action'], 'submitted') + self.assertEqual(decisions[-1]['reason'], 'cycle_cap') + self.client.submit.assert_called_once() + def test_high_and_failed_queue(self): for n in (95, None): decisions = self.cycle(counts=(n,))