Put a process-local gate before the existing retrieval worker. Run overlapping HTTP requests to observe an active request, a queued request, overflow refusal and waiting expiry. Then verify that a worker keeps its slot until it actually finishes. This is scripted source retrieval without a model call or generated answer.
When you need this. Use admission control when an application must limit simultaneous work instead of starting every request immediately. An active lease represents unfinished work. A waiting request holds a queue position and has not started retrieval.
Admit before creating retrieval tools
The gate belongs to one event loop in one Uvicorn process. A free slot admits a request immediately. Otherwise the request joins a bounded first-in, first-out (FIFO) queue. Full queues refuse the request before invoking retrieval.
The actual fixture permits 1 active request and 1 waiting request. Overflow returns 429 with queue_full. The driver observes active/queued server states before submitting the overflow request; client submission order alone does not establish that order.
The endpoint validates the bounded JSON body and strict request model before gate admission. After admission, asyncio.to_thread runs the existing retriever. That worker constructs its own SQLite tools and closes them at completion. Corpus loading at startup is outside these per-request slots.
Expire waiting work and promote in order
A queued request awaits a future with an event-loop monotonic deadline. Expiry removes its waiter and returns 503 with wait_timeout. It has not started the retriever. When active work releases a slot, the oldest waiter receives the lease.
Queue overflow. The unit test proves a denied lease never enters its body. Waiting deadline. The timeout test checks refusal and removal from the queue. Queued cancellation. Cancelling a waiting task removes it; if promotion wins that race, cancellation releases the transferred lease instead of leaking capacity.
A separate trusted fixture extends the waiting deadline so both requests complete after FIFO promotion. Its recorded statuses are completed and completed. The event log preserves transition order and active/waiting counts; it retains only its most recent bounded history.
Keep the lease until the worker ends
Python task cancellation cannot stop an already running synchronous worker thread. finish_worker shields its task and defers cancellation propagation until the thread finishes. The surrounding finally then releases the active lease. Releasing earlier would admit extra work while the original thread is still running.
Worker failure. The trusted factory raises during tool construction. The actual endpoint returns 500 and its active count returns to 0. The direct cancellation test blocks a real worker thread, cancels its awaiting task, verifies that the slot stays occupied, then releases the worker and verifies cleanup.
Client timeout. The HTTPX client records ReadTimeout. Server active work remains at 1 immediately afterward and eventually returns to 0. That observation is distinct from the direct cancellation test: a client read timeout does not establish that server work stopped.
Execute concurrent requests against one process
Download the limit concurrent rag requests lab. It includes the original Python-summary corpus, unchanged retrieval core, HTTP body contract, pinned dependencies and executable tests. Create a virtual environment and install requirements.lock before running the driver.
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.lock"""One-event-loop FIFO admission. A lease represents actual unfinished work."""
import asyncio
from collections import deque
from contextlib import asynccontextmanager
class Refused(Exception):pass
class Gate:
def __init__(self,max_active=1,max_waiting=1,wait_seconds=0.15):
if any(type(x) is not int or not 1<=x<=32 for x in (max_active,)) or type(max_waiting) is not int or not 0<=max_waiting<=32:raise ValueError('invalid_capacity')
if type(wait_seconds) not in (int,float) or not 0<wait_seconds<=30:raise ValueError('invalid_wait_deadline')
self.max_active=max_active;self.max_waiting=max_waiting;self.wait_seconds=wait_seconds
self.active=0;self.waiters=deque();self.events=deque(maxlen=128);self.sequence=0
def event(self,status):
self.sequence+=1;self.events.append({'order':self.sequence,'status':status,'active':self.active,'waiting':len(self.waiters)})
def snapshot(self):return {'active':self.active,'waiting':len(self.waiters),'events':list(self.events),'max_active':self.max_active,'max_waiting':self.max_waiting,'wait_seconds':self.wait_seconds}
async def acquire(self):
if self.active<self.max_active and not self.waiters:
self.active+=1;self.event('admitted');return
if len(self.waiters)>=self.max_waiting:
self.event('queue_full');raise Refused('queue_full')
future=asyncio.get_running_loop().create_future();self.waiters.append(future);self.event('queued')
try:await asyncio.wait_for(asyncio.shield(future),self.wait_seconds)
except (TimeoutError,asyncio.CancelledError) as error:
if future in self.waiters:self.waiters.remove(future)
elif future.done() and not future.cancelled():self.release()
self.event('wait_timeout' if isinstance(error,TimeoutError) else 'wait_cancelled')
if isinstance(error,TimeoutError):raise Refused('wait_timeout') from None
raise
def release(self):
if self.active<=0:raise RuntimeError('unbalanced_release')
self.active-=1;self.event('released')
if self.waiters:
future=self.waiters.popleft();self.active+=1;future.set_result(None);self.event('promoted')
@asynccontextmanager
async def lease(self):
await self.acquire()
try:yield
finally:self.release()
async def finish_worker(call,*args):
"""Cancellation cannot free capacity while an uncancellable thread still runs."""
task=asyncio.create_task(asyncio.to_thread(call,*args));cancelled=False
while not task.done():
try:await asyncio.shield(task)
except asyncio.CancelledError:cancelled=True
except Exception:break
if cancelled:
# Consume a possible worker exception before propagating cancellation.
if not task.cancelled():task.exception()
raise asyncio.CancelledError
return task.result()python run.py --output scratch-result.json
python -m unittest discover -v{
"execution_mode": "actual-concurrent-loopback-http-single-process",
"queue_overflow": {
"http_status": 429,
"status": "queue_full"
},
"wait_deadline": {
"http_status": 503,
"status": "wait_timeout"
},
"fifo_promotion": [
{
"http_status": 200,
"status": "completed"
},
{
"http_status": 200,
"status": "completed"
}
],
"server_failure": {
"http_status": 500,
"status": "internal_error"
},
"idle_active": 0
}The driver starts temporary loopback Uvicorn processes, performs real HTTPX requests and reaps each child. Trusted fixture factories control failure and overlap; public requests cannot choose a delay or bypass a guard. The small delay exposes queue states; it is not a latency or capacity benchmark.
Inspect the complete saved result
{
"execution_mode": "actual-concurrent-loopback-http-single-process",
"model_called": false,
"controlled_delay_seconds": 0.35,
"cases": {
"active_observed": {
"active": 1,
"waiting": 0,
"events": [
{
"order": 1,
"status": "admitted",
"active": 1,
"waiting": 0
}
],
"max_active": 1,
"max_waiting": 1,
"wait_seconds": 0.15
},
"queued_observed": {
"active": 1,
"waiting": 1,
"events": [
{
"order": 1,
"status": "admitted",
"active": 1,
"waiting": 0
},
{
"order": 2,
"status": "queued",
"active": 1,
"waiting": 1
}
],
"max_active": 1,
"max_waiting": 1,
"wait_seconds": 0.15
},
"queue_overflow": {
"http_status": 429,
"status": "queue_full"
},
"wait_deadline": {
"http_status": 503,
"status": "wait_timeout"
},
"first_completed": {
"http_status": 200,
"status": "completed"
},
"idle": {
"active": 0,
"waiting": 0,
"events": [
{
"order": 1,
"status": "admitted",
"active": 1,
"waiting": 0
},
{
"order": 2,
"status": "queued",
"active": 1,
"waiting": 1
},
{
"order": 3,
"status": "queue_full",
"active": 1,
"waiting": 1
},
{
"order": 4,
"status": "wait_timeout",
"active": 1,
"waiting": 0
},
{
"order": 5,
"status": "released",
"active": 0,
"waiting": 0
}
],
"max_active": 1,
"max_waiting": 1,
"wait_seconds": 0.15
},
"client_timeout": {
"client_exception": "ReadTimeout",
"server_after_timeout": {
"active": 1,
"waiting": 0,
"events": [
{
"order": 1,
"status": "admitted",
"active": 1,
"waiting": 0
},
{
"order": 2,
"status": "queued",
"active": 1,
"waiting": 1
},
{
"order": 3,
"status": "queue_full",
"active": 1,
"waiting": 1
},
{
"order": 4,
"status": "wait_timeout",
"active": 1,
"waiting": 0
},
{
"order": 5,
"status": "released",
"active": 0,
"waiting": 0
},
{
"order": 6,
"status": "admitted",
"active": 1,
"waiting": 0
}
],
"max_active": 1,
"max_waiting": 1,
"wait_seconds": 0.15
},
"server_after_completion": {
"active": 0,
"waiting": 0,
"events": [
{
"order": 1,
"status": "admitted",
"active": 1,
"waiting": 0
},
{
"order": 2,
"status": "queued",
"active": 1,
"waiting": 1
},
{
"order": 3,
"status": "queue_full",
"active": 1,
"waiting": 1
},
{
"order": 4,
"status": "wait_timeout",
"active": 1,
"waiting": 0
},
{
"order": 5,
"status": "released",
"active": 0,
"waiting": 0
},
{
"order": 6,
"status": "admitted",
"active": 1,
"waiting": 0
},
{
"order": 7,
"status": "released",
"active": 0,
"waiting": 0
}
],
"max_active": 1,
"max_waiting": 1,
"wait_seconds": 0.15
}
},
"fifo_promotion": {
"responses": [
{
"http_status": 200,
"status": "completed"
},
{
"http_status": 200,
"status": "completed"
}
],
"state": {
"active": 0,
"waiting": 0,
"events": [
{
"order": 1,
"status": "admitted",
"active": 1,
"waiting": 0
},
{
"order": 2,
"status": "queued",
"active": 1,
"waiting": 1
},
{
"order": 3,
"status": "released",
"active": 0,
"waiting": 1
},
{
"order": 4,
"status": "promoted",
"active": 1,
"waiting": 0
},
{
"order": 5,
"status": "released",
"active": 0,
"waiting": 0
}
],
"max_active": 1,
"max_waiting": 1,
"wait_seconds": 1
}
},
"server_constructor_failure": {
"response": {
"http_status": 500,
"status": "internal_error"
},
"state": {
"active": 0,
"waiting": 0,
"events": [
{
"order": 1,
"status": "admitted",
"active": 1,
"waiting": 0
},
{
"order": 2,
"status": "released",
"active": 0,
"waiting": 0
}
],
"max_active": 1,
"max_waiting": 1,
"wait_seconds": 0.15
}
}
},
"lifecycle": {
"command": [
"python3",
"serve.py",
"--fd",
"<inherited-loopback-socket>",
"--app",
"delay_fixture:app"
],
"startup_observed": true,
"shutdown_exit_status": -15,
"stdout": "",
"stderr": "INFO: Started server process [37400]\nINFO: Waiting for application startup.\nINFO: Application startup complete.\nINFO: Shutting down\nINFO: Waiting for application shutdown.\nINFO: Application shutdown complete.\nINFO: Finished server process [37400]\n",
"shutdown_complete_observed": true,
"server_process_reaped": true
},
"success_lifecycle": {
"command": [
"python3",
"serve.py",
"--fd",
"<inherited-loopback-socket>",
"--app",
"success_fixture:app"
],
"startup_observed": true,
"shutdown_exit_status": -15,
"stdout": "",
"stderr": "INFO: Started server process [37401]\nINFO: Waiting for application startup.\nINFO: Application startup complete.\nINFO: Shutting down\nINFO: Waiting for application shutdown.\nINFO: Application shutdown complete.\nINFO: Finished server process [37401]\n",
"shutdown_complete_observed": true,
"server_process_reaped": true
},
"fault_lifecycle": {
"command": [
"python3",
"serve.py",
"--fd",
"<inherited-loopback-socket>",
"--app",
"fault_fixture:app"
],
"startup_observed": true,
"shutdown_exit_status": -15,
"stdout": "",
"stderr": "INFO: Started server process [37403]\nINFO: Waiting for application startup.\nINFO: Application startup complete.\nINFO: Shutting down\nINFO: Waiting for application shutdown.\nINFO: Application shutdown complete.\nINFO: Finished server process [37403]\n",
"shutdown_complete_observed": true,
"server_process_reaped": true
},
"scope": "Trusted fixture delay exposes admission states; not a latency or capacity benchmark. Client timeout did not cancel server work. Direct task cancellation is tested separately. Corpus setup is outside gate; SQLite tools are created per worker after admission."
}Inspect a failure without overstating the result
Find the overflow and waiting-deadline responses, then compare their gate events with FIFO promotion. Inspect the client-timeout state: explain why the server still holds a lease. Run the blocked-worker cancellation test and identify the exact moment at which capacity becomes available.
This gate does not implement global rate limits, multiple-worker capacity, user fairness, authentication or a hard runtime deadline. A synchronous worker that never returns can hold its slot indefinitely; the lab does not claim to stop it.
Read the explanation
Client waiting, task cancellation and synchronous worker completion are distinct events.
What’s next
Continue with Run RAG Evals on an Updated Corpus.
References
The official asyncio documentation describes shielding, waiting and thread execution. HTTPX timeout documentation defines a client read timeout. Holding a lease until the underlying worker ends is this lab’s tested policy; these APIs do not terminate an arbitrary running thread.