Service.serve: a shutdown that is bounded and returns

Write the shutdown path — the part of a service that is never tested and always matters.

type State = Literal["idle", "running", "draining", "stopped"]

def describe(state: State) -> str

class Service[T]:
    def __init__(self, *, drain_timeout: float) -> None
    def state(self) -> State
    def request_stop(self) -> None
    async def serve(self, work: AsyncIterator[T],
                    handle: Callable[[T], Awaitable[None]]) -> None

serve pulls items from work, runs handle(item) as a task, and keeps going until the source is exhausted or a stop is requested. Then it drains:

1. wait drain_timeout for the in-flight handlers,
2. cancel what is left and wait drain_timeout again,
3. cancel again and wait a final drain_timeout, then abandon.

Three bounded rounds, so total shutdown time is bounded and you can size it against your orchestrator’s grace period.

serve must return normally — never by raising CancelledError — and no task it created may still be pending when it returns. A handler that raises is recorded and ignored: a service does not die because one item failed, but the exception must still be retrieved, or it resurfaces later as “Task exception was never retrieved”.

request_stop() is synchronous, idempotent, and honoured even when called before serve() — without consuming and dropping an item from the source.

Why request_stop is synchronous

loop.add_signal_handler(signal.SIGTERM, service.request_stop) takes a Callable[[], object]. You cannot pass a coroutine function: doing so creates a coroutine object and immediately discards it, silently doing nothing. So the handler’s entire job is to flip a flag and let code that can await react. That constraint is the reason for the shape, and the reason it must be safe to call twice.

What the report proves

• produced — how many items the source yielded. In stop_before it must be 0: a stop requested before serve() must not pull an item and drop it.
• started / handled — in-flight handlers complete rather than being abandoned mid-write.
• cancelled — which handlers observed a CancelledError. In the stubborn scenario the handler swallows the first one and honours the second, so it appears twice. That is the assertion that your escalation ladder exists: one cancel()-then-await waits forever, because asyncio’s cancellation is edge-triggered and delivers exactly one CancelledError.
• returned_normally / outcome — serve() returns. A shutdown that propagates CancelledError gives you a non-zero exit code and an orchestrator that records a failed termination for what was a clean deploy.
• final_state / description — the state machine ends at stopped, and describe covers every state.
• pending — 0.

Where the type system earns its keep

assert_never in describe makes the match exhaustive at type-check time. Add a fifth state and every unhandled match becomes an error pointing at the exact line; without it, the unhandled state falls through and returns None, and the bug surfaces somewhere else entirely. In a shutdown state machine the states encode which operations are legal, so a missing case is a legality check that silently stopped running.

Service[T] is generic in the item type, and handle: Callable[[T], Awaitable[None]] says the handler’s return value goes nowhere — which is the honest description of a worker and forces you to notice if you meant to collect results.