flare_mlake · API · quic

flare.quic.server

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

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__

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__

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

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

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

def bound(self) -> Bool

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

connection_count

def connection_count(self) -> Int

Number of connection slots currently allocated.

dispatch_datagram

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

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

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

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

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

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

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

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

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

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

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

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

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.
  2. 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

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

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

def ready_slots(self) -> List[Int]

:meth:take_ready_slots without consuming them.

take_ready_slots

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

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

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

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

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

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

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

run

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

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

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

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

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

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

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.