Implement PROTOCOL_VIOLATION idioms

[lilalinda]

We now have a few spots were PROTOCOL VIOLATION could occur. Handle in code accordingly.

[lilalinda]

We need to tighten the contextmanager pattern(s) for QuicConnection to allow "cancellation from inside" when protocol gets violated or the user initiates closing of the connection with explicit calls...

Here is what ChatGPT offers although I think as_safe_channel decorator might be more succinct?

The neat Trio way is: wrap the caller’s block inside a CancelScope owned by your context manager, and have a background “state watcher” cancel that scope when the connection reaches DRAINING. That immediately unwinds the user’s async with ... block and lets your context manager run its cleanup.

Here’s the minimal pattern:

from contextlib import asynccontextmanager
import trio

# Assume your connection object exposes:
#   - state (enum), with .DRAINING and .CLOSED
#   - draining_event: trio.Event() fired when state becomes DRAINING
#   - send_loop() / recv_loop() coroutines

@asynccontextmanager
async def connect(...):
    conn = await _initiate_handshake(...)
    # Everything inside this scope (including the caller's with-block) can be cancelled
    with trio.CancelScope() as lifetime:
        async with trio.open_nursery() as nursery:

            async def watch_state():
                # Preferred: wait on an event your connection sets on DRAINING
                await conn.draining_event.wait()
                # Optionally: sleep until the end of the drain period, or cancel immediately
                lifetime.cancel()                    # ← exits caller's `async with` block

            nursery.start_soon(conn.send_loop)
            nursery.start_soon(conn.recv_loop)
            nursery.start_soon(watch_state)

            try:
                # The caller’s code runs *inside* lifetime CancelScope
                yield conn
            finally:
                # We’re exiting (either caller finished, or we cancelled them from watch_state)
                nursery.cancel_scope.cancel()
                # Optional: grace period to let draining finish, e.g.:
                # with trio.move_on_after(conn.drain_timeout):
                #     await conn.finish_draining()
                await conn.close_transport()

Caller:

async with client.connect(...) as conn:
    # do work; if conn enters DRAINING, watch_state() cancels `lifetime`,
    # this block is cancelled, and control jumps to the context manager’s `finally`.
    await conn.send(...)

Notes & variants

• If you don’t have a draining_event, you can poll:

  async def watch_state():
      while conn.state not in (State.DRAINING, State.CLOSED):
          await trio.sleep(0.05)
      lifetime.cancel()

• For a server-side handle_connection(), use the same pattern: wrap the yield (or the block serving the connection) in a CancelScope and cancel it when the protocol transitions to DRAINING.

• If you need a graceful drain before tearing down sockets, do it in the context manager’s finally after cancelling the nursery (e.g., wait up to your drain timeout with trio.move_on_after(...), then close).

This pattern keeps the user-facing API simple (async with ...) and gives you precise control to exit early when the protocol enters DRAINING.

[lilalinda]

Still need to write and debug 3 tests (test_endpoints.py):

1. idle_timeout = 5s (or similar) at client and client (and server?) connection gets properly closed
2. endpoint turns off DATAGRAM after connection established:
   a) client turns off and receives DATAGRAM (echo server) -> client closes with PROTOCOL_VIOLATION
   b) server turns off and receives DATAGRAM -> server closes with PV but does not exit (still available for other connections)