Close the QUIC datagram transports on shutdown

[CoryLowe5]

worker_serve keeps its TCP servers in servers and closes them in
its finally, but the transport returned by create_datagram_endpoint
is discarded, so shutdown cannot close what it never kept. Each
QUIC-serving worker leaks its datagram transport, visible as a
GC-time unclosed transport ResourceWarning under python -W always
(we found it via tracemalloc pointing into hypercorn/asyncio/run.py
from a test suite that serves QUIC in-process many times per run).

The fix keeps the datagram transports and closes them after the
graceful drain, alongside the existing server close. After the drain
rather than before it, deliberately: unlike a TCP Server — where
close() only stops new connections — the datagram transport is both
the listener and the data path for established QUIC connections, so
closing it earlier would break in-flight traffic during
graceful_timeout.

Includes a test (tests/asyncio/test_run.py) that serves one QUIC
socket through the real worker_serve and asserts every transport
created by create_datagram_endpoint is closing after it returns; it
skips when aioquic is not installed, matching the h3 extra. A
self-contained reproduction script is attached below in a comment for
convenience.

(The trio worker was inspected and left alone: it hands its socket to
trio.socket.from_stdlib_socket inside the UDPServer under the
nursery — different ownership shape, not measured, not touched.)

[CoryLowe5]

Reproduction (self-contained — generates a throwaway self-signed certificate; run with python -W always repro_datagram_leak.py). On 0.18.0 it prints LEAK: 1 'unclosed transport' ResourceWarning(s); with this branch it prints CLEAN.

"""Reproduce: hypercorn's asyncio worker leaks its QUIC datagram transport.

`worker_serve` keeps its TCP servers in `servers` and closes them in its
finally, but the transport returned by `create_datagram_endpoint` is
discarded into `_`, so shutdown cannot close what it never kept. The
leaked transport surfaces as a GC-time ResourceWarning.

Run against hypercorn 0.18.0 with the h3 extra:
    python -W always repro_datagram_leak.py
Expected output on 0.18.0:  LEAK: 1 'unclosed transport' ResourceWarning(s)
With the fix:               CLEAN: no unclosed-transport ResourceWarning

Self-contained: generates a throwaway self-signed certificate.
"""
import asyncio
import datetime
import gc
import tempfile
import warnings
from pathlib import Path

from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import NameOID

from hypercorn.app_wrappers import ASGIWrapper
from hypercorn.asyncio.run import worker_serve
from hypercorn.config import Config


async def app(scope, receive, send):
    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                await send({"type": "lifespan.shutdown.complete"})
                return


def write_self_signed(directory: Path) -> tuple[Path, Path]:
    key = ec.generate_private_key(ec.SECP256R1())
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
    now = datetime.datetime.now(datetime.timezone.utc)
    cert = (
        x509.CertificateBuilder()
        .subject_name(name)
        .issuer_name(name)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now)
        .not_valid_after(now + datetime.timedelta(days=1))
        .sign(key, hashes.SHA256())
    )
    cert_path = directory / "cert.pem"
    key_path = directory / "key.pem"
    cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
    key_path.write_bytes(
        key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.TraditionalOpenSSL,
            serialization.NoEncryption(),
        )
    )
    return cert_path, key_path


async def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        cert, key = write_self_signed(Path(tmp))
        config = Config()
        config.bind = []
        config.quic_bind = ["127.0.0.1:0"]
        config.certfile = str(cert)
        config.keyfile = str(key)
        config.graceful_timeout = 0.1

        async def shutdown_now() -> None:
            return None

        await worker_serve(
            ASGIWrapper(app), config, shutdown_trigger=shutdown_now
        )


with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    asyncio.run(main())
    gc.collect()

leaks = [w for w in caught
         if issubclass(w.category, ResourceWarning)
         and "unclosed transport" in str(w.message)]
if leaks:
    print(f"LEAK: {len(leaks)} 'unclosed transport' ResourceWarning(s)")
    for w in leaks:
        print(f"  {w.message}")
else:
    print("CLEAN: no unclosed-transport ResourceWarning")

[noqt]

One cleanup edge remains here: if a later QUIC endpoint fails to start, control never reaches the try below, so transports already appended to datagram_transports stay open.

I reproduced this on dca1b5f with two fake QUIC sockets: the first create_datagram_endpoint() returned a tracked transport, the second raised RuntimeError, and after worker_serve() exited the first transport still reported is_closing() == False. The new test covers normal shutdown only. Moving endpoint setup under the cleanup try (and closing partially created transports on startup failure) would cover this path too.

Disclosure: I ran the changed worker through Lumi Trace at NOQT for review context, then verified this behavior with the focused runtime repro above.

[CoryLowe5]

Thanks for checking this. Agreed that a failure while starting a later QUIC endpoint leaves earlier transports open — but that is the existing shape of worker_serve rather than something this change adds: the TCP servers list a few lines up has the same gap, since setup for both runs before the cleanup try. I kept this PR to the one observable defect (transports discarded on the normal shutdown path). Happy to do the startup-failure cleanup for both the TCP servers and the datagram transports as a follow-up if @pgjones would like it.