# flare.quic.server

> A module of flare_mlake's API.

``flare.quic.server`` -- QUIC server reactor.

Wraps the sans-I/O QUIC connection state machine
(:class:`flare.quic.state.Connection`) in a UDP listener +
per-connection dispatcher. The per-datagram dispatch loop
threads bytes through :class:`OpenSslQuicCrypto` packet
protection, the transport-frame parser, and
:meth:`Connection.handle_frame` to produce
:class:`ConnectionEvents` for the H3 driver above; idle
timers sit on the shared TimerWheel.

## What ships here

- :class:`QuicServerConfig` -- bind configuration: host, port,
  rustls config carrier, idle timeout, max packet size.
- :class:`QuicListener` -- factory that owns a bound UDP
  socket plus the per-connection dispatch table. The :meth:`bind`
  factory opens the socket, :meth:`tick` drains one datagram
  (unit-testable), :meth:`run` is the blocking event loop, and
  :meth:`shutdown` requests a clean exit.
- :class:`QuicConnection` -- per-connection driver that
  composes the existing :class:`flare.quic.state.Connection`
  state machine with the rustls QUIC session.
- :class:`ConnectionIdTable` -- per-listener routing table from
  Connection ID to connection slot, used by the dispatch loop
  to route inbound datagrams via the Destination Connection ID
  in the packet header (RFC 9000 §5.1).

References:
- RFC 9000 §5 "Connections" -- Connection ID routing.
- RFC 9000 §10 "Connection Termination" -- idle / draining.
- RFC 9000 §17 "Packet Formats" -- long / short header parse
  for the dispatch path.

## Structs

### `QuicListener`

```mojo
struct QuicListener
```

UDP listener + per-connection dispatcher.

Long-lived. One instance per QUIC server bind. Construct via
:meth:`bind`; the constructor opens the UDP socket and binds
it. The reactor drives the listener via :meth:`run` (blocks
until :meth:`shutdown`) or :meth:`tick` (single iteration --
used by tests + by callers that want to multiplex the event
loop with other work).

Implements: `Movable`

Fields:

- `config: QuicServerConfig`
- `cid_table: ConnectionIdTable`
- `connections: List[QuicConnection]`: Connection slab. Per-connection state lives here; the :class:`ConnectionIdTable` maps each Connection ID to the slot index. Slots are append-only; the closed-slot sweeper runs at the end of every :meth:`advance_timers` call so idle-timed-out connections get reaped against the same monotonic clock the wheel uses.
- `tls_acceptor: RustlsQuicAcceptor`: Long-lived rustls QUIC acceptor built once at :meth:`bind` from :attr:`QuicServerConfig.rustls_config`. Owns the rustls ``ServerConfig`` (cert + key + ALPN list) that every per-connection session derives keys from. When the caller passes the default (empty PEM) :class:`RustlsQuicConfig`, the FFI returns a NULL handle and the dispatch path routes all CRYPTO frames through the silent-drop branch -- existing tests that bind a listener purely to exercise the UDP / routing / timer surfaces continue to work without supplying real PEM material.
- `free_slots: List[Int]`: Reclaimed slots ready for the next accepted connection; see :meth:`_reclaim_slot`.
- `slot_free: List[Bool]`: Per-slot: on :attr:`free_slots`, so never reclaimed twice.
- `tls_sessions: List[_SessionSlot]`: Parallel slab to :attr:`connections`: one rustls QUIC session carrier per :class:`QuicConnection` slot. Each :class:`_SessionSlot` holds the raw rustls ``Box<Session>*`` (or 0 for the NULL-PEM sentinel) plus the current outbound encryption level. The slab owns the handles; :meth:`__deinit__` walks every non-zero handle through :func:`flare.tls._rustls_quic_ffi._do_session_free` exactly once at listener teardown. The shared FFI library handle for every per-slot call is borrowed from :attr:`tls_acceptor._lib` so the .so stays mapped across every feed-crypto / take-crypto roundtrip.
- `tls_egress_queues: List[List[UInt8]]`: Per-slot outbound CRYPTO byte queue at the INITIAL level. Each successful :meth:`feed_crypto` is followed by :meth:`take_crypto` and the resulting bytes append here; :meth:`_drain_and_send` wraps them in a CRYPTO frame inside an Initial-level packet, AEAD-protects with :func:`protect_initial_packet`, and emits via :meth:`send_to`. Cleared after every successful drain.
- `tls_handshake_egress_queues: List[List[UInt8]]`: Per-slot outbound CRYPTO byte queue at the HANDSHAKE level. Populated by :meth:`_dispatch_crypto_frames` after rustls emits ``KeyChange::Handshake``. Drained by the egress builder which wraps each batch in a CRYPTO frame inside a Handshake-level packet, AEAD-protects via :class:`RustlsQuicSession.packet_encrypt` at level 2, and emits via :meth:`send_to`.
- `crypto_reasm: List[_CryptoReasm]`: Per-slot inbound CRYPTO reassembly state (one per encryption level). Parallel slab to :attr:`connections`; reorders + coalesces inbound CRYPTO fragments so rustls's in-order ``read_hs`` sees a contiguous handshake stream.
- `tls_1rtt_egress_queues: List[List[UInt8]]`: Per-slot outbound CRYPTO byte queue at the 1-RTT (APPLICATION) level. Populated by :meth:`_dispatch_crypto_frames` after rustls emits ``KeyChange::OneRtt``. Drained by the egress builder which encodes each batch into a 1-RTT short-header packet, AEAD-protects via :class:`RustlsQuicSession.packet_encrypt` at level 3, and emits via :meth:`send_to`.
- `peer_addrs: List[SocketAddr]`: Per-slot peer UDP address. Parallel slab to :attr:`connections` -- captured in :meth:`_accept_initial` from the inbound datagram's sender. The egress path reads this to call :meth:`send_to(slot, ...)` without re-parsing the inbound datagram.
- `migration_probe: List[MigrationProbe]`: Per-slot connection-migration path-validation state (RFC 9000 sec 8 / sec 9): the candidate address under validation plus the anti-amplification byte counters. Default (no migration) is an idle probe that costs nothing.
- `loss: List[LossRecovery]`: Per-slot RFC 9002 loss-recovery state. Parallel slab to :attr:`connections`, grown in :meth:`_accept_initial` alongside the others. ``LossRecovery`` is Movable but not Copyable, so this slab is read through ``self.loss[slot].method()`` or a ``ref`` binding -- never the ``.copy()`` idiom the Copyable slabs use.
- `loss_init: List[LossRecovery]`: Per-slot loss recovery for the CRYPTO packets sent in the Initial space (RFC 9002 sec 6, separate packet-number space). A lost server flight used to be recovered only by the idle timeout; these and :attr:`loss_hs` re-send it on loss or PTO.
- `loss_hs: List[LossRecovery]`: The same for the Handshake space.
- `rx_1rtt_ranges: List[List[UInt64]]`: Per-slot received 1-RTT packet numbers, stored as disjoint ranges (flat [low, high] pairs, descending by high). The ACK frame is built from these so we acknowledge only packets we actually received: the peer skips packet numbers to detect optimistic-ACK attacks (RFC 9000 sec 21.4), so acking a contiguous span would acknowledge a never-sent number and the peer would close with PROTOCOL_VIOLATION (sec 13.1).
- `rx_1rtt_ack_pending: List[Bool]`: Per-slot flag: an ack-eliciting 1-RTT packet arrived and has not yet been acknowledged. Cleared once the egress path emits the ACK frame.
- `handshake_done_sent: List[Bool]`: Per-slot flag: the server has emitted HANDSHAKE_DONE (RFC 9000 sec 19.20) once the 1-RTT keys installed. Confirms the handshake so the peer stops retransmitting its Finished and discards Handshake keys.
- `next_local_cid_seq: List[UInt64]`: Per-slot next NEW_CONNECTION_ID sequence number to issue (RFC 9000 sec 19.15). The handshake CID is sequence 0; the server issues sequence 1 with the first 1-RTT flight so a migrating client has a spare Destination CID to switch to (sec 9.5 requires a fresh CID per path).
- `pending_migration_tx: List[List[UInt8]]`: Per-slot buffer of already-encoded 1-RTT migration frames (PATH_RESPONSE echoes for inbound PATH_CHALLENGEs) awaiting egress. Stashed in :meth:`_process_one_packet` from the surfaced :class:`ConnectionEvents` and flushed by the 1-RTT coalesced drain.
- `rx_stream_bytes: List[UInt64]`: Per-slot cumulative count of inbound stream payload bytes. Feeds the connection-level MAX_DATA the egress path advertises so the peer's flow-control window never closes (RFC 9000 sec 19.9).
- `rx_bidi_stream_count: List[UInt64]`: Per-slot count of client-initiated bidi streams observed (largest ``(stream_id >> 2) + 1``). Feeds the MAX_STREAMS_BIDI grant (RFC 9000 sec 19.11) so HTTP/3, which opens one bidi stream per request, can keep opening streams past the initial transport-parameter limit.
- `http3_connections: List[Http3Connection]`: Per-slot HTTP/3 connection driver. Parallel slab to :attr:`connections` -- one :class:`flare.http3.Http3Connection` instance per QUIC connection. Allocated in :meth:`_accept_initial` so every accepted connection has its H3 driver ready; STREAM frames from the post-handshake 1-RTT payload route through :meth:`_route_http3_stream_chunks` into the matching slot.
- `early_strike: EarlyDataStrikeSet`: Listener-level cross-connection 0-RTT replay defense (RFC 9001 sec 9.2). Keyed on each accepted connection's original DCID with a time window from :attr:`QuicServerConfig.early_data_strike_window_ms`. Consulted once per connection on its first 0-RTT packet so a captured first flight replayed as a fresh accept is refused. Idle (never populated) unless 0-RTT is enabled.
- `timer_wheel: TimerWheel`: Per-listener :class:`flare.runtime.timer_wheel.TimerWheel` driving idle timeouts. Each scheduled timer's token is :func:`flare.quic.timers.encode_timer_token(kind, slot)`; :meth:`advance_timers` dispatches each fired token to the matching :class:`QuicConnection` callback.

#### `__init__`

```mojo
def __init__(out self, config: QuicServerConfig, var sock: UdpSocket, addr: SocketAddr, var tls_acceptor: RustlsQuicAcceptor)
```

Wrap an already-bound :class:`UdpSocket`. Internal -- callers use :meth:`bind`.

#### `__deinit__`

```mojo
def __deinit__(deinit self)
```

Drop the listener: release every rustls session in the slab exactly once.

Each :class:`_SessionSlot` is a non-owning carrier; the
slab itself is the unique owner of the underlying
``Box<Session>*`` allocations rustls produced via
:func:`_do_accept`. The free routes through
:meth:`RustlsQuicAcceptor.free_session` so Mojo's
``deinit`` rule (no sub-field access during ``deinit``)
is respected -- the acceptor borrows ``self`` (and its
``_lib``) by reference rather than via a sub-field of
the listener.

#### `bind`

```mojo
def bind(config: QuicServerConfig) -> Self
```

Open the UDP socket and bind it to ``config.host`` / ``config.port``. Returns a ready-to-run listener.

If ``config.port == 0`` the kernel picks an ephemeral
port; read it back via :meth:`local_addr`.

Constructs the per-listener rustls QUIC acceptor from
``config.rustls_config`` at bind time so each accepted
connection's TLS session is materialized against the
same long-lived ``ServerConfig``. An empty / malformed
PEM does not raise here; the acceptor
surfaces a NULL handle and CRYPTO bytes route through
the silent-drop branch.

#### `local_addr`

```mojo
def local_addr(self) -> SocketAddr
```

Return the address the UDP socket is actually bound to. After :meth:`bind` with ``config.port == 0`` this reports the kernel-chosen ephemeral port.

#### `bound`

```mojo
def bound(self) -> Bool
```

Whether the UDP socket is bound. True for every listener returned by :meth:`bind`.

#### `connection_count`

```mojo
def connection_count(self) -> Int
```

Number of connection slots currently allocated.

#### `dispatch_datagram`

```mojo
def dispatch_datagram(mut self, datagram: Span[UInt8], peer: SocketAddr) -> Int
```

Route a single UDP datagram to the right connection slot.

Parses the first byte to pick long vs short header,
extracts the Destination Connection ID, looks it up in
:attr:`cid_table`, and either:

* Routes the datagram to the existing slot via
  :meth:`_handle_inbound`. Decryption failures are caught
  and converted to silent drops so a single bad sender
  can't poison the listener.
* Allocates a new slot for an Initial packet with an
  unknown DCID (the QUIC accept path -- RFC 9000 §7),
  then drives the same inbound path on the new connection
  so the first Initial advances the state machine.
* Returns ``-1`` to drop short-header packets with
  unknown DCIDs (those datagrams are silently
  discarded).

#### `take_http3_completed_streams`

```mojo
def take_http3_completed_streams(self, slot: Int) -> List[Int]
```

Return the stream ids ready for handler dispatch on ``slot``. Delegates to :meth:`flare.http3.Http3Connection.take_completed_streams`. Empty list if the slot is out of range or has no H3 driver attached.

#### `take_http3_request`

```mojo
def take_http3_request(mut self, slot: Int, stream_id: Int) -> Request
```

Materialize the :class:`flare.http.Request` for ``(slot, stream_id)``. The dispatch caller invokes a Handler with this Request, then feeds the Response back through :meth:`emit_http3_response`.

Raises if the slot is out of range or the H3 driver
does not track the stream (see
:meth:`flare.http3.Http3Connection.take_request` for the
underlying gating).

#### `emit_http3_response`

```mojo
def emit_http3_response(mut self, slot: Int, stream_id: Int, var response: ResponseImpl)
```

Encode ``response`` and queue it on the stream's egress for the 1-RTT drain (:meth:`_drain_1rtt_coalesced`), which frames it as far as flow control and the congestion window allow once the slot's 1-RTT keys are installed.

A streaming response (one carrying a ``body_stream`` chunk
source) is handled incrementally: its HEADERS (and any inline
``body`` prefix) are queued now, the source is boxed on the
stream's record, and :meth:`pump_http3_streams` pulls its body
one chunk per tick. The FIN waits until the source is exhausted.

#### `take_http3_response_egress`

```mojo
def take_http3_response_egress(mut self, slot: Int, stream_id: Int) -> List[UInt8]
```

Take the response bytes queued on ``stream_id`` and not yet framed, bypassing the 1-RTT drain. Returns an empty list when nothing is queued. A finished buffered response's record is dropped; a streaming one keeps its source.

#### `open_http3_tunnel`

```mojo
def open_http3_tunnel(mut self, slot: Int, stream_id: Int) -> Http3Tunnel
```

Accept the extended CONNECT request on ``stream_id`` (taken with :meth:`take_http3_request`): queue a 200 HEADERS and keep the stream open both ways.

#### `is_http3_tunnel_request`

```mojo
def is_http3_tunnel_request(self, slot: Int, stream_id: Int) -> Bool
```

Whether the request on ``stream_id`` is an extended CONNECT (RFC 9220), ready before the client ends the stream.

#### `http3_tunnel_open`

```mojo
def http3_tunnel_open(self, t: Http3Tunnel) -> Bool
```

Whether ``t`` still carries bytes both ways: neither side has ended or reset it and its connection is up.

#### `send_http3_tunnel`

```mojo
def send_http3_tunnel(mut self, t: Http3Tunnel, data: Span[UInt8]) -> Bool
```

Queue ``data`` as one DATA frame on ``t``; False once our side has ended or been reset (the peer having ended its side does not stop ours). The drain sends it as flow control allows.

#### `take_http3_tunnel`

```mojo
def take_http3_tunnel(mut self, t: Http3Tunnel) -> List[UInt8]
```

The bytes the peer sent on ``t`` since the last call. They count as consumed, so connection credit keeps up with a long-lived tunnel (RFC 9000 sec 4.2).

#### `close_http3_tunnel`

```mojo
def close_http3_tunnel(mut self, t: Http3Tunnel)
```

End our side of ``t``: FIN after the queued bytes, then the stream is reclaimed.

#### `pump_http3_streams`

```mojo
def pump_http3_streams(mut self, cancel: Cancel) -> Int
```

Advance every active HTTP/3 streaming response by one chunk.

For each stream whose source is still open, pulls a single chunk
(mirroring the H1/H2 one-chunk-per-edge yield) and queues it as a
DATA frame on the stream's egress. On end-of-stream the source is
freed, the trailers are queued, and the record is marked ``done``
so the drain sends the FIN after the last byte. Returns the
number of streams advanced (produced a chunk or reached EOS).

Backpressure: a stream with :data:`_STREAM_PULL_HIGH_WATER` bytes
still unsent is skipped. The drain only frames what the peer's
MAX_STREAM_DATA / MAX_DATA and the congestion window allow, so a
fast source cannot outrun a slow reader.

#### `tick`

```mojo
def tick(mut self, timeout_ms: Int = Int(100)) -> Bool
```

Drain a burst of inbound datagrams and pump egress.

Reactor I/O loop step:

1. One ``recvmmsg`` blocks (``SO_RCVTIMEO``) until a datagram
   arrives or the wait times out cleanly, and takes the rest of
   the burst, up to :data:`_RX_BATCH` datagrams, with each
   one's ECN codepoint. Without ``recvmmsg``, a blocking
   ``recv_from`` wakes the tick and ``try_recv_from`` drains
   the rest. Either way the per-tick H3 pump / timer / egress
   bookkeeping in the serve loop is amortized over the whole
   burst instead of paid per datagram.
3. Each datagram routes through :meth:`dispatch_datagram`
   by DCID; the matched slot's :meth:`_drain_and_send`
   flushes its ACK / response.

Returns ``True`` if at least one datagram was dispatched,
``False`` if the initial ``recv_from`` timed out.

#### `poll_timeout_ms`

```mojo
def poll_timeout_ms(self, max_ms: Int = Int(100)) -> Int
```

How long :meth:`tick` may block: until the next timer (PTO, loss time, idle, closing) is due, at most ``max_ms`` and at least 1 ms (``SO_RCVTIMEO`` 0 would block forever). A fixed 100 ms wait delayed every timer on a quiet connection by up to that long.

#### `send_to`

```mojo
def send_to(mut self, datagram: Span[UInt8], addr: SocketAddr) -> Int
```

Emit a single fully-protected QUIC datagram on the listener's UDP socket, ECN-marked when the 1-RTT packet just built asked for it (``_tx_ecn``, consumed here). Returns the byte count actually written (UDP either sends the whole datagram or raises).

#### `ready_slots`

```mojo
def ready_slots(self) -> List[Int]
```

:meth:`take_ready_slots` without consuming them.

#### `take_ready_slots`

```mojo
def take_ready_slots(mut self) -> List[Int]
```

The slots that received request-stream data since the last call; only these can have new completed requests.

#### `drain_all_egress`

```mojo
def drain_all_egress(mut self) -> Int
```

Drain the egress of every slot something was queued on since the last call (a packet, a response, a tunnel or datagram send, a streaming chunk, GOAWAY). Returns the number of slots that emitted. Timers drain their own slot.

#### `send_datagram`

```mojo
def send_datagram(mut self, slot: Int, payload: Span[UInt8]) -> Bool
```

Queue ``payload`` as an unreliable DATAGRAM (RFC 9221) for the next drain. False, and nothing queued, when the client takes no DATAGRAMs, the frame would exceed its ``max_datagram_frame_size`` or one packet, or the queue is full.

#### `send_http3_datagram`

```mojo
def send_http3_datagram(mut self, t: Http3Tunnel, payload: Span[UInt8]) -> Bool
```

Send ``payload`` as an HTTP Datagram (RFC 9297) on tunnel ``t``. False, and nothing sent, unless both ends sent SETTINGS_H3_DATAGRAM, the tunnel is open, and :meth:`send_datagram` takes it.

#### `take_http3_datagrams`

```mojo
def take_http3_datagrams(mut self, t: Http3Tunnel) -> List[List[UInt8]]
```

The HTTP Datagram payloads received on tunnel ``t`` since the last call, Quarter Stream ID removed.

#### `take_datagrams`

```mojo
def take_datagrams(mut self, slot: Int) -> List[List[UInt8]]
```

The DATAGRAM payloads received on ``slot`` since the last call, in arrival order.

#### `run`

```mojo
def run(mut self)
```

Run the listener's event loop. Blocks until :meth:`shutdown` flips the stop flag.

The loop drives the full I/O cycle
``recv -> dispatch -> drain -> protect -> sendto ->
advance_timers``:

1. ``tick(100)`` blocks up to 100 ms in ``recv_from``;
   on a datagram it runs the dispatch + handle + drain
   chain.
2. On a 100 ms timeout it still calls
   :meth:`drain_all_egress` so any session that started
   handshaking can flush its first response.
3. :meth:`advance_timers` then runs the wheel against
   the current monotonic clock so the idle-timeout
   callbacks fire on time.

The 100 ms recv timeout caps the worst-case timer slop
and the shutdown-flag polling interval.

#### `begin_drain`

```mojo
def begin_drain(mut self, timeout_ms: Int)
```

Start a graceful shutdown (RFC 9114 sec 5.2): every live connection is sent GOAWAY naming the next request stream it may not open, new connections are refused, and :meth:`drain_step` closes each connection once its requests are answered or ``timeout_ms`` has passed.

#### `drain_step`

```mojo
def drain_step(mut self) -> Bool
```

Advance a drain started by :meth:`begin_drain`: close (H3_NO_ERROR, RFC 9114 sec 5.2) each connection with no request left in flight, and every connection once the deadline passes. Returns whether no live connection remains; False when no drain is running.

#### `shutdown`

```mojo
def shutdown(mut self)
```

Request the event loop to exit. Idempotent; safe to call from a signal handler (sets a single Bool flag).

#### `schedule_idle_timeout`

```mojo
def schedule_idle_timeout(mut self, slot: Int) -> UInt64
```

Arm the idle timer for ``slot`` from the current wheel tick.

The delay is the effective idle timeout of RFC 9000 sec 10.1:
the minimum of ``config.max_idle_timeout_ms`` and the client's
``max_idle_timeout`` (when non-zero), at least three PTOs. When
both are 0 the timeout is disabled: nothing is armed and the
returned id is 0. Cancels the slot's previous idle timer if any
so every ``handle_packet`` only ever has one idle timer in
flight per connection.

Returns the new timer id (stored back into the slot's
:attr:`QuicConnection.idle_timer_id`).

#### `advance_timers`

```mojo
def advance_timers(mut self, now_ms: UInt64) -> Int
```

Advance the wheel to ``now_ms`` and dispatch every fired timer token to the matching action (per :mod:`flare.quic.timers`). Returns the number of tokens dispatched.

Handles all three registered kinds:

* ``TIMER_KIND_IDLE`` -- RFC 9000 sec 10.1 idle timeout:
  flips the connection to closed via ``on_idle_expired``.
* ``TIMER_KIND_ACK_DELAY`` -- RFC 9000 sec 13.2.1 deferred
  ACK: flushes any owed 1-RTT ACK through the coalesced
  egress path so a quiescent peer still learns what arrived
  even when no further inbound packet drives a drain.
* ``TIMER_KIND_PTO`` -- RFC 9002 sec 6.2 probe timeout:
  dispatched to :meth:`_on_pto_expired`.

Also sweeps the CID table for any connections whose
``alive`` flag flipped False during dispatch (idle
timeout, draining); their CIDs are retired so a stray
retransmit doesn't route to a dead slot.

#### `close`

```mojo
def close(mut self)
```

Close the underlying UDP socket. Idempotent. The :class:`UdpSocket` destructor also closes the fd, so explicit calls are only required when callers want to free the port before the listener goes out of scope.