test_t3_stress.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Stress tier tests — T3 concurrent submissions, queue concurrency bound.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import concurrent.futures |
| 6 | import threading |
| 7 | import unittest |
| 8 | |
| 9 | from scooling_lab_helpers import valid_payload |
| 10 | |
| 11 | from scooling_lab.dataset_review import DatasetStore |
| 12 | from scooling_lab.errors import ApiError, ErrorCode |
| 13 | from scooling_lab.service import TrainingApiService |
| 14 | from scooling_lab.store import TrainingJobStore |
| 15 | |
| 16 | |
| 17 | def _make_service( |
| 18 | *, |
| 19 | queue_limit: int = 256, |
| 20 | max_concurrent_running: int = 1, |
| 21 | auto_run: bool = True, |
| 22 | dataset_store: DatasetStore | None = None, |
| 23 | ) -> TrainingApiService: |
| 24 | store = TrainingJobStore( |
| 25 | queue_limit=queue_limit, |
| 26 | max_concurrent_running=max_concurrent_running, |
| 27 | ) |
| 28 | return TrainingApiService( |
| 29 | store, auto_run_worker=auto_run, dataset_store=dataset_store |
| 30 | ) |
| 31 | |
| 32 | |
| 33 | class T3StressConcurrencyBoundTests(unittest.TestCase): |
| 34 | """Stress tests verifying the running-job concurrency bound holds under load.""" |
| 35 | |
| 36 | def test_stress_t3_concurrent_submissions_never_exceed_running_bound(self) -> None: |
| 37 | """At most max_concurrent_running jobs are in running state at once. |
| 38 | |
| 39 | We track the peak concurrent running count by hooking into the store |
| 40 | check; since the fake worker is synchronous the semaphore ensures the |
| 41 | bound is never exceeded. |
| 42 | """ |
| 43 | |
| 44 | max_running = 2 |
| 45 | service = _make_service( |
| 46 | max_concurrent_running=max_running, queue_limit=64, auto_run=True |
| 47 | ) |
| 48 | observed_running: list[int] = [] |
| 49 | lock = threading.Lock() |
| 50 | original_run_job = service._worker.run_job |
| 51 | |
| 52 | def instrumented_run(job_id: str) -> object: # type: ignore[misc] |
| 53 | snapshot = service._store.running_count() |
| 54 | with lock: |
| 55 | observed_running.append(snapshot) |
| 56 | return original_run_job(job_id) |
| 57 | |
| 58 | service._worker.run_job = instrumented_run # type: ignore[method-assign] |
| 59 | |
| 60 | with concurrent.futures.ThreadPoolExecutor(max_workers=12) as pool: |
| 61 | list( |
| 62 | pool.map( |
| 63 | lambda i: service.create_training_job(valid_payload(f"conc-{i}")), |
| 64 | range(24), |
| 65 | ) |
| 66 | ) |
| 67 | |
| 68 | self.assertTrue(len(observed_running) > 0) |
| 69 | peak = max(observed_running) |
| 70 | self.assertLessEqual( |
| 71 | peak, |
| 72 | max_running, |
| 73 | msg=f"Peak concurrent running {peak} exceeded bound {max_running}", |
| 74 | ) |
| 75 | |
| 76 | def test_stress_t3_concurrent_dataset_registrations_are_safe(self) -> None: |
| 77 | """Concurrent registrations for distinct datasets are all stored correctly.""" |
| 78 | |
| 79 | ds_store = DatasetStore() |
| 80 | dataset_ids = [f"stress-ds-{i:04d}" for i in range(50)] |
| 81 | |
| 82 | with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool: |
| 83 | results = list( |
| 84 | pool.map(lambda did: ds_store.register(did), dataset_ids) |
| 85 | ) |
| 86 | |
| 87 | self.assertEqual(len(results), 50) |
| 88 | for did in dataset_ids: |
| 89 | self.assertEqual(ds_store.get(did).dataset_id, did) |
| 90 | |
| 91 | def test_stress_t3_concurrent_review_decisions_are_stable(self) -> None: |
| 92 | """Only the first review decision for a dataset wins; retries are idempotent.""" |
| 93 | |
| 94 | ds_store = DatasetStore() |
| 95 | ds_store.register("concurrent-review-ds") |
| 96 | ds_store.submit_for_review("concurrent-review-ds") |
| 97 | |
| 98 | errors: list[ApiError] = [] |
| 99 | successes: list[str] = [] |
| 100 | lock = threading.Lock() |
| 101 | |
| 102 | def try_approve() -> None: |
| 103 | try: |
| 104 | record = ds_store.approve("concurrent-review-ds") |
| 105 | with lock: |
| 106 | successes.append(record.status.value) |
| 107 | except ApiError as exc: |
| 108 | with lock: |
| 109 | errors.append(exc) |
| 110 | |
| 111 | with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: |
| 112 | list(pool.map(lambda _: try_approve(), range(16))) |
| 113 | |
| 114 | final = ds_store.get("concurrent-review-ds") |
| 115 | self.assertEqual(final.status.value, "approved") |
| 116 | |
| 117 | def test_stress_t3_queue_limit_still_enforced_under_concurrent_load(self) -> None: |
| 118 | """QUEUE_LIMIT_EXCEEDED is returned reliably under concurrent pressure.""" |
| 119 | |
| 120 | service = _make_service( |
| 121 | queue_limit=3, max_concurrent_running=3, auto_run=False |
| 122 | ) |
| 123 | service.create_training_job(valid_payload("ql-a")) |
| 124 | service.create_training_job(valid_payload("ql-b")) |
| 125 | service.create_training_job(valid_payload("ql-c")) |
| 126 | |
| 127 | limit_hits: list[bool] = [] |
| 128 | lock = threading.Lock() |
| 129 | |
| 130 | def try_create(suffix: str) -> None: |
| 131 | try: |
| 132 | service.create_training_job(valid_payload(f"ql-overflow-{suffix}")) |
| 133 | except ApiError as exc: |
| 134 | if exc.code == ErrorCode.QUEUE_LIMIT_EXCEEDED: |
| 135 | with lock: |
| 136 | limit_hits.append(True) |
| 137 | |
| 138 | with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: |
| 139 | list(pool.map(try_create, [str(i) for i in range(16)])) |
| 140 | |
| 141 | self.assertGreater(len(limit_hits), 0) |
| 142 | |
| 143 | |
| 144 | if __name__ == "__main__": |
| 145 | unittest.main() |