Lab: build a bounded inference gateway

Assignment

Create a local API backed by a fake model client that waits 100 ms, returns a JSON response, or raises a timeout. No paid API is needed.

Build sequence

1. Define request and response schemas and reject blank or oversized input.
2. Limit in-flight calls to three; submit twenty concurrent requests and record queue time.
3. Persist an operation ID and status in SQLite; simulate a retry with the same ID.
4. Add a deadline and test cancellation while queued and while executing.

Acceptance checks

• At most three fake calls run simultaneously.
• Expired requests do not create fresh work; retries cannot insert duplicate operations.
• The test report distinguishes validation failures, overload, timeouts, and successful responses.

Evidence to keep

Submit the working artifact or decision document described above, the inputs used, and the observed results for every acceptance check. Include one failed attempt and the change you made after investigating it. Label fixture-based outcomes separately from live-system measurements. These artifacts become part of your final portfolio review.

Guided workshop: exercise deadlines without a paid API

Start with a fake asynchronous provider so you can force success, timeout, and invalid output reliably. The gateway owns the overall deadline; a provider adapter should also use transport timeouts. Limiting concurrency is separate from limiting queued requests.

Copy code# academy-check
import asyncio

async def provider(question):
    if question == "slow":
        await asyncio.sleep(1)
    return {"answer": "fixture answer"}

async def gateway(question, call=provider):
    if not isinstance(question, str) or not question.strip():
        return {"status": "invalid_input"}
    try:
        result = await asyncio.wait_for(call(question), timeout=.02)
    except asyncio.TimeoutError:
        return {"status": "timeout"}
    if not isinstance(result, dict) or not isinstance(result.get("answer"), str):
        return {"status": "invalid_output"}
    return {"status": "ok", "answer": result["answer"]}

async def check():
    assert (await gateway("hello"))["status"] == "ok"
    assert (await gateway("slow"))["status"] == "timeout"
    assert (await gateway(""))["status"] == "invalid_input"
    async def broken(_):
        return []
    assert (await gateway("hello", broken))["status"] == "invalid_output"
asyncio.run(check())

The fake provider is cooperative with cancellation. Blocking synchronous work inside an async function would prevent timely cancellation; a timeout is not a mechanism for killing arbitrary code. Record this limitation in your design.

Build steps: wrap the function with the HTTP boundary from the foundation module. Authenticate first, validate body size and question length, then acquire capacity. Reject excess admission rather than allowing an unbounded waiting list. Give each request a trace ID and log status, elapsed time, and provider version without logging raw private prompts.

Challenge: add one retry for a transient provider failure while keeping a single overall deadline. Compute the remaining time before each attempt. Do not retry invalid input or invalid schema indefinitely. A fallback model can change answer quality and output compatibility, so validate its output through the same contract.

Expected evidence: valid input returns a validated answer; the slow fixture returns timeout; malformed provider output returns invalid_output; cancellation cleans up capacity; a load test never exceeds the configured active-request limit. Include failed attempts in the latency and cost table.

Review answer: the fake proves runtime behavior, not model quality. Replacing it with a real adapter requires a separately versioned live evaluation, provider error mapping, explicit usage accounting, and credentials supplied at runtime. Keep this deterministic suite so provider variability cannot obscure a regression in your gateway.