From 2ba0b8bd39fb059ef3bf831fd214a25bd1a095ab Mon Sep 17 00:00:00 2001 From: Vladimir Eremin Date: Sun, 11 Oct 2026 21:43:53 +0300 Subject: [PATCH] fix(readiness): accept concurrent browser asset probes --- assets/composition-readiness/adapter.py | 1 + assets/composition-readiness/test_adapter.py | 20 ++++++++++++++++++++ 2 files changed, 21 insertions(+) diff --git a/assets/composition-readiness/adapter.py b/assets/composition-readiness/adapter.py index 1909b8f..db0d684 100644 --- a/assets/composition-readiness/adapter.py +++ b/assets/composition-readiness/adapter.py @@ -16,6 +16,7 @@ class ReadinessServer(http.server.ThreadingHTTPServer): daemon_threads = True block_on_close = False allow_reuse_address = True + request_queue_size = 64 def __init__(self, address, *, token, path, command, timeout=1.5, slots=4, single_flight=False): if not isinstance(token, str) or not re.fullmatch(r'[!-~]{16,512}', token): diff --git a/assets/composition-readiness/test_adapter.py b/assets/composition-readiness/test_adapter.py index aa46e34..555bda4 100644 --- a/assets/composition-readiness/test_adapter.py +++ b/assets/composition-readiness/test_adapter.py @@ -62,6 +62,26 @@ class AdapterTests(unittest.TestCase): def one(): barrier.wait(timeout=2);start=time.monotonic();result=self.request(s);return result,time.monotonic()-start with concurrent.futures.ThreadPoolExecutor(max_workers=count) as pool:return list(pool.map(lambda _:one(),range(count))) + def test_browser_asset_burst_connects_and_shares_one_slow_probe(self): + code=f"import pathlib,time;p=pathlib.Path({str(self.marker)!r});f=p.open('a');f.write('probe\\n');f.close();time.sleep(.8)" + s=self.server(code,timeout=3,slots=16,single_flight=True) + process_request=s.process_request + def delayed_accept(request,address): + time.sleep(.01) + process_request(request,address) + s.process_request=delayed_accept + barrier=threading.Barrier(32) + def one(_): + barrier.wait(timeout=2) + c=http.client.HTTPConnection(*s.server_address,timeout=.5) + try: + c.connect();c.sock.settimeout(3) + c.request('GET','/.well-known/apex-ready',headers={'X-APEX-Ready-Token':'private-test-token'}) + r=c.getresponse();return r.status,r.read() + finally:c.close() + with concurrent.futures.ThreadPoolExecutor(max_workers=32) as pool:results=list(pool.map(one,range(32))) + self.assertEqual(results,[(204,b'')]*32) + self.assertEqual(self.marker.read_text().splitlines(),['probe']) def test_single_flight_shares_success_and_sequential_request_reprobes(self): code=f"import pathlib,time;p=pathlib.Path({str(self.marker)!r});f=p.open('a');f.write('probe\\n');f.close();time.sleep(.2)" s=self.server(code,timeout=1,slots=1,single_flight=True)