# API Reference: `plaza_session`
`plaza_session` implements `plaza::session::Session` over real network transports: actix-web WebSockets and length-delimited TCP.
## Contents
- [Feature Flags](#feature-flags)
- [1. Module `codec`](#1-module-codec)
- [Trait `WireCodec`](#trait-wirecodec)
- [Struct `JsonCodec`](#struct-jsoncodec)
- [2. Module `manager`](#2-module-manager)
- [Struct `ConnectionManager<ID: AgentId>`](#struct-connectionmanagerid-agentid)
- [Struct `TransportSession<Op, ID, C: WireCodec>`](#struct-transportsessionop-id-c-wirecodec)
- [Type `SessionClock` and struct `SessionOptions`](#type-sessionclock-and-struct-sessionoptions)
- [3. Module `conditioner`](#3-module-conditioner)
- [Type Alias `LinkSink` and struct `LinkPublisher`](#type-alias-linksink-and-struct-linkpublisher)
- [Function `target_matches`](#function-target_matches)
- [Type Alias `OutboundFrame`](#type-alias-outboundframe)
- [Enum `ConnectionOrder`](#enum-connectionorder)
- [Struct `IncomingFrame<ID>`](#struct-incomingframeid)
- [Structs `Queues` and `Limits`](#structs-queues-and-limits)
- [Struct `Overflow`](#struct-overflow)
- [Constants](#constants)
- [4. Module `workload`](#4-module-workload)
- [5. Module `actix_ws` (feature `actix_ws`)](#5-module-actix_ws-feature-actix_ws)
- [Struct `ActixWsPlazaSession<Op, ID, C = JsonCodec>`](#struct-actixwsplazasessionop-id-c--jsoncodec)
- [6. Module `tcp` (feature `tcp`)](#6-module-tcp-feature-tcp)
- [Struct `TcpPlazaSession<Op, ID, C = JsonCodec>`](#struct-tcpplazasessionop-id-c--jsoncodec)
- [Type Alias `AgentFactory<ID>`](#type-alias-agentfactoryid)
- [7. Module `host` (feature `actix_host`)](#7-module-host-feature-actix_host)
- [Struct `Host`](#struct-host)
- [Struct `SimHost` and struct `SimWiring`](#struct-simhost-and-struct-simwiring)
- [Function `lan_address() -> Option<String>`](#function-lan_address---optionstring)
- [Function `init_logging()`](#function-init_logging)
- [8. Module `stats`](#8-module-stats)
- [Struct `TransportStats`](#struct-transportstats)
- [9. The control plane in pieces](#9-the-control-plane-in-pieces)
- [Struct `LinkDriver<ID, C>`](#struct-linkdriverid-c)
- [Enum `Inbound`](#enum-inbound)
- [Struct `LinkHandle`](#struct-linkhandle)
- [Struct `ProbeState` and function `make_probe`](#struct-probestate-and-function-make_probe)
- [Function `earliest` and constant `DOWN_SEED_FLIP`](#function-earliest-and-constant-down_seed_flip)
- [10. Module `gate`](#10-module-gate)
- [11. Module `budget`](#11-module-budget)
- [12. Module `admission`](#12-module-admission)
- [Struct `Farewell`](#struct-farewell)
- [Trait `ConnectionAdmitter<ID>` and enum `ConnectionAdmission<ID>`](#trait-connectionadmitterid-and-enum-connectionadmissionid)
- [Struct `Peer`](#struct-peer)
- [Struct `Pending` and enum `Presented`](#struct-pending-and-enum-presented)
- [13. Error Handling](#13-error-handling)
- [Enum `SessionLayerError`](#enum-sessionlayererror)
## Feature Flags
| Feature | Default | Enables |
|---|---|---|
| `actix_ws` | yes | [`ActixWsPlazaSession`](#struct-actixwsplazasessionop-id-c--jsoncodec): actix-web WebSockets. |
| `tcp` | yes | [`TcpPlazaSession`](#struct-tcpplazasessionop-id-c--jsoncodec): length-delimited TCP. |
| `actix_host` | no | [`host::Host`](#7-module-host-feature-actix_host): the listen-server HTTP layer. Implies `actix_ws`. |
| `json` | yes | `JsonCodec` and the codec a session type falls back to when it names none. Enables `plaza_wire/json`, which is what pulls in `serde_json`. |
| `msgpack` | no | `MsgPackCodec` (compact) and `MsgPackNamedCodec` (struct field names kept, for a peer that decodes by name). Enables `plaza_wire/msgpack`. |
[`manager`](#2-module-manager), [`codec`](#1-module-codec) and [`error`](#13-error-handling) compile unconditionally.
**Dropping `serde_json`.** Turn off `json` and the crate no longer builds it: `plaza_session = { version = "0", default-features = false, features = ["tcp", "msgpack"] }`. Nothing else has to change, because `plaza` and `plaza_wire` are depended on with `default-features = false` here and neither `plaza` nor `plaza_lobby` names a codec at all, so no internal dependency forces the choice back on. Bring your own codec and you can drop `msgpack` too, leaving no built-in format compiled.
## 1. Module `codec`
### Trait `WireCodec`
```rust,ignore
pub trait WireCodec: Clone + Send + Sync + 'static {
fn name(&self) -> &'static str;
fn encode<T: Serialize>(&self, value: &T)
-> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>>;
fn decode<T: DeserializeOwned>(&self, bytes: &[u8])
-> Result<T, Box<dyn std::error::Error + Send + Sync>>;
}
```
Implementations must be stateless and cheap to clone, one lives inside every session and is shared across all of its connections.
### Struct `JsonCodec`
The default. Human-readable, so a browser console or `websocat` can inspect traffic. `Debug + Clone + Copy + Default`.
## 2. Module `manager`
Shared machinery. Most applications use it only through a transport, but it is public so new transports can be written outside this crate.
### Struct `ConnectionManager<ID: AgentId>`
The connection registry plus the notification channels a `StateController` consumes. Connections are held by id and indexed by agent. An agent may hold several at once, so `set_agent_link_profile`, `record_protocol` and `agent_link_dropped` act on each, while `agent_rtt`, `agent_link_rtt` and `protocol` answer from the first registered.
* **`new(transport: &'static str, capacity: usize) -> Self`**
* **`with_hello(transport: &'static str, capacity: usize, hello: Option<OutboundFrame>) -> Self`**: the same, plus a pre-encoded frame pushed to every connection the moment it registers. `TransportSession::with_protocol` builds one `Kind::Hello` frame at construction and hands it here, so the handshake costs one encode for the process rather than one per client.
* **`with_hello_and_clock(transport, capacity, hello: Option<OutboundFrame>, clock: Option<SessionClock>) -> Self`**: the same again, plus the clock a `Pong` is stamped with.
* **`with_options(transport: &'static str, hello: Option<OutboundFrame>, options: &SessionOptions) -> Self`**: takes the clock, queue depths and limits from `options`. `hello` stays a separate argument because the manager holds the encoded frame while `options` holds the version it was built from. The `capacity` constructors above are this one with `inbound`, `decoded` and `presence` all set to that number.
* **`queues(&self) -> &Queues`**, **`limits(&self) -> &Limits`**: what this manager was built with. A transport adapter reads `queues().outbound` when it creates a connection's outbound queue, `queues().conditioner` for its delay queues, `limits().probe_slots` for the probe table and whichever byte cap its own framing enforces.
* **`async register(&self, agent: Agent<ID>, to_client_tx: SessionSender<OutboundFrame>) -> ConnectionId`**: records a connected client, sends the hello frame if there is one and announces the join. Build the queue with `plaza::session::session_channel`, so a transport never names the channel crate. `async` because announcing may wait under `PresenceOverflow::Backpressure`.
* **`async deregister(&self, conn_id: ConnectionId)`**: removes it and announces the departure. **Bookkeeping only**: the socket belongs to the connection task and dropping the outbound sender does not wake it. To end a session, use `close_connection`; the task then closes the socket and deregisters itself.
* **`connections_of(&self, id: &ID) -> Vec<ConnectionId>`**: the live connections an agent holds. It maps an agent to its connections: a decoded op names an agent, while a close, a deadline or a per-connection reader needs a connection. `PresenceEvent` carries the same id at join and leave.
* **`close_connection(&self, conn_id: ConnectionId, farewell: Farewell) -> bool`**: orders the connection's task to flush what is queued, write the farewell's `Goodbye` and close the socket with its code. Sync (`try_send` under the registry's read guard), so it is callable from inside `StateLogic`. Returns whether a live connection took the order. The departure then arrives as an ordinary `Left`: a forced disconnect and a cable pull look the same to the controller, on purpose. The application supplies the code and whatever detail it wants heard; see [`Farewell`](#struct-farewell).
* **`idle_for(&self, conn_id) -> Option<Duration>`** / **`agent_idle_for(&self, id: &ID) -> Option<Duration>`**: how long a connection has been silent, counted from its last data frame or from `register` if it never sent one. **Probes do not count** and that is only implementable here: the control plane answers a `Ping` invisibly, so an AFK rule written against decoded ops is right by accident and one written against frames would never fire. The agent form takes the shortest across its connections. No timers and no timeout policy live here; read it from your own tick and apply your own number.
* **`connection_inbound(&self, conn_id) -> Option<InboundVolume>`** / **`agent_inbound(&self, id: &ID) -> InboundVolume`**: monotonic per-connection inbound counters (`frames`, `bytes`, `shed`), counting what the connection sent rather than what survived the queues. [`TransportStats`](#struct-transportstats) counts the session; this answers "who", which the session-wide numbers cannot. Windowing and thresholds are the application's: keep the last reading and diff or feed a `plaza_client_utils::RateMeter`.
* **`connection_outbound(&self, conn_id) -> Option<OutboundVolume>`** / **`agent_outbound(&self, id: &ID) -> OutboundVolume`**: the same shape pointing the other way (`frames`, `bytes`, `withheld`), counting what `broadcast` handed to the connection's queue rather than what the socket managed to write. A frame the fan-out dropped was never that connection's traffic and is not counted. `withheld` counts the times `connection_owed` / `agent_owed` answered no.
* **`set_outbound_budget(&self, conn_id, Option<OutboundBudget>) -> bool`** / **`set_agent_outbound_budget(&self, id: &ID, Option<OutboundBudget>) -> usize`** / **`outbound_budget(&self, conn_id) -> Option<OutboundBudget>`**: what a connection may be sent, see [module `budget`](#11-module-budget). A new budget starts with a full burst; `None` takes it away, debt included. `Limits::outbound_budget` is the one every connection starts with.
* **`connection_owed(&self, conn_id) -> bool`** / **`agent_owed(&self, id: &ID) -> bool`**: whether a recipient is owed a frame now: whether its budget has credit left or always if it has none. What a snapshot pass asks before it builds for a recipient, answering `Ok(None)` from its provider when the answer is no. A connection that is gone is owed nothing and an agent is owed one when any of its connections is. Every refusal is counted as `withheld`.
* **`record_inbound_activity(&self, conn_id, bytes: usize) -> Verdict`**: what the control plane calls for each inbound data frame; a custom transport that bypasses `handle_inbound` calls it itself. It both counts the frame and judges it against [`Limits::inbound_rate`](#10-module-gate) and the return is `#[must_use]`: a frame it refuses must not be forwarded. Without a rate configured, always `Verdict::Admit`.
* **`set_deadline(&self, conn_id, after: Option<Duration>, farewell: Farewell) -> bool`**: arms, moves or (with `None`) clears a deadline the connection task enforces; expiry goes through the same flush-then-goodbye close. Setting again replaces the deadline, which is how a renewal extends a session (an arcade credit, an auth token's expiry). No timer exists outside the connection task's own loop; the application decides what stamps, renews or revokes it.
* **`deregister_agent(&self, id: &ID, farewell: Farewell) -> usize`**: `connections_of` then `close_connection` on each; how many took the order.
* **`disconnect_all(&self, farewell: Farewell) -> usize`**: the same close as `deregister_agent`, applied to every live connection: each gets the goodbye and is then closed.
* **`begin_pending(&self) -> Option<PendingSlot>`**: takes one of the places [`Limits::pending_connections`](#structs-queues-and-limits) allows for a socket awaiting its credential or `None` when they are all held, which a transport answers by refusing the socket before its upgrade. The slot frees itself when dropped; `admitted()`, `refused()` and `timed_out()` consume it and count the outcome on [`TransportStats`](#struct-transportstats). The shipped transports call this; a custom transport with an admitter does the same.
* **`take_orders(&self, conn_id: ConnectionId) -> Option<SessionReceiver<ConnectionOrder>>`**: for transport authors. The order stream a connection task selects on beside its outbound queue; it must be its own `select!` arm, because the outbound arm is disabled the moment `deregister` drops the sender, which is exactly when a close must still work. Single-consumer, like the `take_*` streams.
* **`async forward_incoming(&self, from: Agent<ID>, frame: Bytes)`**: publishes one client frame toward the controller, still encoded. Drops under load by default; waits under `InboundOverflow::Backpressure`, which stops that client's socket being read. Takes anything that converts into a [`Frame`](#type-alias-outboundframe), so a transport handing over a buffer it already owns reaches the deserialize bridge without a copy.
* **`broadcast(&self, target: &MessageTarget<ID>, frame: OutboundFrame) -> Result<Vec<ConnectionId>, SessionLayerError>`**: fans one already-encoded frame out to the matching connections. It takes bytes, not a message, because a `SessionMessage` is encoded **once** by [`encode_message`](#struct-transportsessionop-id-c-wirecodec) and the same buffer is shared with every recipient, at the cost of a refcount bump each. `Agent` and `Agents` cost a lookup per named agent rather than a pass over the registry, which matters because per-recipient snapshots address one agent at a time: a scan there made a snapshot pass quadratic in connections. Measured in `benches/broadcast.rs`, addressing one agent is flat at ~37ns from 8 connections to 4096, against a scan that reaches 10µs at the top of that range; below roughly a dozen connections the scan is the cheaper of the two, by about 9ns.
The connections returned are those that could not take the frame, populated only under `OutboundOverflow::Disconnect`: under `Drop` every recipient may be full at once and naming them would allocate on the fan-out path for a list nobody reads. It stays sync and releases the read guard before returning, because `deregister` takes the write guard and would deadlock against it on the same thread.
* **`async disconnect_overflowed(&self, overflowed: Vec<ConnectionId>)`**: ends those connections in the order `broadcast` reported them. `TransportSession::send_message` calls it for you. The departures are announced **without waiting**, even under `PresenceOverflow::Backpressure`: a send that disconnects a client must not block on the controller hearing about it, since a controller behind on presence is what filled the queue. `LossFree` derives `Disconnect` and `Backpressure` together, so this is the combination it ships with.
* **`overflow(&self) -> Overflow`**: what this manager does when a queue is full.
`Agents` still delivers once to an agent named twice. The variants carrying a list of ids test that list directly rather than hashing it into a set, which the same bench settled: for a `u32` id, building the set costs more than the comparisons it saves at every list length up to 128.
* **`take_raw_incoming(&self) -> SessionReceiver<IncomingFrame<ID>>`**: the inbound stream, whose payloads are still encoded bytes for the deserialize bridge to decode.
* **`take_presence(&self) -> SessionReceiver<PresenceEvent<ID>>`**
* **`agent_rtt(&self, id: &ID) -> Option<(Duration, u64)>`**: the measured round trip for an agent and how many samples it rests on. Keyed by agent because that is what an application holds: it knows who joined, not which socket they arrived on and a reconnecting player is a new connection but the same agent.
* **`rtt(conn_id)`** / **`min_rtt(conn_id)`** / **`rtt_samples(conn_id)`**: the same, per connection.
* **`record_rtt(conn_id, Duration)`**: what a transport calls when it has timed one.
* **`agent_link_rtt(&self, id: &ID)`** / **`link_rtt(conn_id)`** / **`min_link_rtt(conn_id)`** / **`link_rtt_samples(conn_id)`** / **`record_link_rtt(conn_id, Duration)`**: the same family for the round trip measured over the frame path, probe frame out and `Pong` back, through whatever impairment the link carries. The transport family above measures the socket underneath all of that; the difference between the two is what plaza and the configured link cost. A transport with no ping frame of its own reports only this one.
* **`set_link_profile(conn_id, LinkProfile)`** / **`set_agent_link_profile(&ID, LinkProfile)`** / **`set_all_link_profiles(LinkProfile)`** / **`link_profile(conn_id)`**: the delay, jitter and loss frames ride through. See [`conditioner`](#3-module-conditioner). The agent and all-connection forms also **clear the link readings** they touch, because `agent_link_rtt` reports a minimum and a minimum taken under the old link would outlive it. The sample count and the transport-plane numbers are left alone, since those describe the socket rather than the link. `set_link_profile`, the single-connection form, does not clear them.
* **`link_handle(&self, conn_id) -> Option<Arc<LinkHandle>>`**: the shared profile cell, taken once by a connection task. See [`LinkHandle`](#struct-linkhandle).
* **`record_link_drop(conn_id)`** / **`link_dropped(conn_id)`** / **`agent_link_dropped(&ID)`** / **`total_link_dropped()`**: how many frames the link threw away, which only a `Delivery::Datagram` profile ever does. The transports call the first whenever `Conditioner::push` refuses a frame. The application cannot count these itself, because what the link loses never reaches it.
* **`clock(&self) -> Option<&SessionClock>`**: the clock a `Pong` is stamped with, if one was installed.
* **`record_protocol(&self, agent: &Agent<ID>, version: ProtocolVersion)`** / **`protocol(&self, id: &ID) -> Option<ProtocolVersion>`**: what a peer declared in its `Hello`, kept per agent. The deserialize bridge records it; an application reads it to decide what a given client can be sent or to tell it to reload.
* **`connection_count(&self) -> usize`**
The `take_*` methods hand out single-consumer streams; calling one twice panics.
### Struct `TransportSession<Op, ID, C: WireCodec>`
A complete `Session` implementation over any byte transport. Both shipped adapters wrap one and delegate to it.
* **`with_protocol(transport: &'static str, codec: C, capacity: usize, protocol: ProtocolVersion) -> Arc<Self>`**: spawns the deserialize bridge and declares what this build speaks (from [`plaza_wire::build`](../wire/API_REFERENCE.md#6-module-build-feature-build)). It encodes one `Kind::Hello` frame up front for `ConnectionManager::with_hello` to send on every connection and tells the bridge what to compare an inbound `Hello` against.
* **`new(transport: &'static str, codec: C, capacity: usize) -> Arc<Self>`**: the same with `ProtocolVersion::UNKNOWN`, which declares nothing and so disables the check rather than failing it.
* **`with_options(transport: &'static str, codec: C, options: SessionOptions) -> Arc<Self>`**: what the two above delegate to. It takes no `capacity`, because `options.queues` carries every depth including the bridge's own output queue.
* **`manager(&self) -> &Arc<ConnectionManager<ID>>`**
* **`codec(&self) -> &C`**
* **`encode_message(&self, msg: SessionMessage<Op, ID>) -> Result<OutboundFrame, SessionLayerError>`**: writes `[Kind::Ops][encoded ops]` into one buffer, in a single pass and takes the `Vec` whole into `Bytes` with no copy. **`msg.from` is not sent**: the wire is the tag and the ops and the sender is bookkeeping the receiving transport attaches from the connection it read. Hand the frame to [`broadcast`](#struct-connectionmanagerid-agentid); recipients share the buffer rather than each getting a re-encode or a copy.
The buffer is **sized from the last frames rather than reused**, because `Bytes::from` takes the allocation with it: the buffer leaves as the frame, so there is nothing to hand back. The speedup comes from sizing. Measured in `benches/encode.rs`, a `Vec` starting empty reallocates and copies four or five times before a one-op frame is finished and starting it at size is 2.7x faster on JSON, 3.0x on MessagePack, falling to a few percent on a large snapshot where the encode dominates. The hint is an untracked `Relaxed` atomic that decays toward smaller sizes, so one fat snapshot does not oversize every op batch after it.
**The deserialize bridge dispatches on the tag before it decodes anything**, which is what the framing byte is for. `Kind::Ops` decodes as `Vec<Op>` and becomes a `SessionMessage`; `Kind::Hello` decodes as a `ProtocolVersion` and is recorded against the agent for the application to read back; an unknown tag is skipped with a `trace!` and the connection carries on. A malformed body of any kind is a per-message problem: it is logged and dropped, never a disconnect.
`Kind::Ping` and `Kind::Pong` never reach the bridge: they are answered and timed on the connection task, which is the only place that knows which socket to reply on and holds the timer that sent the probe.
### Type `SessionClock` and struct `SessionOptions`
```rust
pub type SessionClock = Arc<dyn Fn() -> u64 + Send + Sync>;
pub struct SessionOptions {
pub protocol: ProtocolVersion,
pub clock: Option<SessionClock>,
pub queues: Queues,
pub limits: Limits,
}
```
What a session declares, what it can answer with and how much it will hold or accept. Passed to `TransportSession::with_options` and each adapter's `with_options` / `bind_with_options`. `SessionOptions::with_protocol(v).clock(f)` builds one and every field has a one-call builder:
```rust
SessionOptions::with_protocol(ProtocolVersion(PROTOCOL))
.outbound_capacity(512)
.max_frame_bytes(256 * 1024)
```
* **`queues(Queues)`**, **`limits(Limits)`**: replace a whole group.
* **`inbound_capacity`**, **`decoded_capacity`**, **`presence_capacity`**, **`outbound_capacity`**, **`conditioner_capacity`**: one queue depth each.
* **`max_frame_bytes`**, **`max_message_bytes`**: one limit each.
* **`rate_limit_inbound(Rate)`**: caps how fast one connection may send. See [module `gate`](#10-module-gate).
* **`budget_outbound(OutboundBudget)`**: what every connection may be sent, judged at the snapshot pass. See [module `budget`](#11-module-budget).
* **`probes`**, **`without_probes`**, **`probe_schedule`**, **`probe_slots`**: the link plane or none of it.
The clock is read when answering a latency probe and its reading becomes `Pong.responder`. It is called on a connection task, so a clock that lives on the simulation loop has to be shared: store the tick into an `AtomicU64` and close over it. **The unit is the application's**; nothing here reads the value as a quantity, converts it or has a default for it. Without a clock, `Pong.responder` is `None` and a client can still measure a round trip but cannot estimate the offset between the two clocks.
## 3. Module `conditioner`
```rust
pub enum Delivery { Reliable, Datagram } // Reliable is the default
pub struct DirectionProfile { pub delay: Duration, pub jitter: Duration, pub loss: f32, pub delivery: Delivery }
pub struct LinkProfile { pub up: DirectionProfile, pub down: DirectionProfile }
```
Impairment applied where the link is. `up` is what the client sends, `down` what the server sends, so a symmetric 100ms round trip is 50ms each way; `LinkProfile::symmetric(p)` builds one. The default is passthrough and costs nothing: no queue, no allocation, no deadline arithmetic on a frame that crosses an unimpaired connection.
**Order is preserved.** Release times are made monotone as frames are queued, so a delayed frame holds up everything behind it and a jitter spike arrives as a stall followed by a burst. That is what jitter does to a reliable stream and letting frames overtake each other would model a datagram link neither transport provides.
**`loss` and `delivery`.** `loss` is the probability a frame is lost and `delivery` decides what happens to it. Under `Delivery::Reliable`, the default and what both transports actually do, the segment is retransmitted: the frame arrives `RETRANSMIT_PENALTY` (200ms, TCP's minimum RTO) late and everything behind it waits. Nothing is deleted, because on a reliable stream a lost segment never reaches the application as a missing message. Under `Delivery::Datagram` the frame is gone and the two ends reconcile, which over a WebSocket is a deliberate simulation of a transport plaza does not yet have, useful for exercising an application's recovery before the channel it is for exists.
**No frame kind is exempt under either model.** Under `Reliable` nothing is lost. Under `Datagram` a lost probe costs one sample of the several the session keeps in flight and a lost `Hello` reads as a peer that declared nothing, the case that handshake already survives. The queue bound is the one place a frame is discarded outright and it stands for a socket buffer running out rather than for anything the network did; control frames are still admitted there.
**The jitter draw is reproducible**: an inline xorshift seeded from the connection id, not from the clock, so an impaired session re-runs the same way.
**`DirectionProfile::is_passthrough()`** is the question the fast path asks and **`LinkProfile::is_passthrough()`** is true only when both directions are.
A `Conditioner` is one direction's queue, driven through three calls: **`next_release() -> Option<Instant>`**, the release time of the head frame and so what to sleep until; **`pop_ready(now) -> Option<Frame>`**, the next frame that has come due; **`is_empty()`**. A frame may take the fast path only when the profile is passthrough **and** the queue is empty: a connection set back to passthrough can still hold delayed frames and releasing a new one past them reorders the stream.
**`Conditioner::drain()`** empties the queue in order with release times ignored: what a close uses to flush what was queued rather than wait out a simulated delay.
### Type Alias `LinkSink` and struct `LinkPublisher`
```rust
pub type LinkSink = Arc<dyn Fn(LinkProfile) + Send + Sync>;
```
Where an application publishes a profile to the transport that owns the links: usually `move |profile| session.set_all_link_profiles(profile)`. A closure rather than a session handle, so the logic that decides what the link should be stays testable without a socket.
`LinkPublisher` is a `LinkSink` with a latch that skips an unchanged setting: **`new(sink)`**, **`publish(profile) -> bool`** (the profile reaches the sink when it differs from the last one published and not otherwise).
A mismatch is **recorded and left connected**, at `debug!`. This layer does nothing more: a version is a build hash, so a peer that merely recompiled is indistinguishable here from one whose shapes changed. Refusing would drop clients that are fine. Whether the mismatch is fatal, cosmetic or worth telling the client to reload is the application's call, made by reading [`ConnectionManager::protocol`](#struct-connectionmanagerid-agentid) and answering in its own ops. It is not logged at `warn!` either, since on a fleet mid-rollout that would be one warning per connection for a harmless rebuild. Whether a mismatch deserves a warning is also the application's call.
* Implements `Session`. Joins are transport-implicit: a client joins by connecting and the adapter calls `register`; a server-side disconnect is [`ConnectionManager::close_connection`](#struct-connectionmanagerid-agentid).
### Function `target_matches`
```rust,ignore
pub fn target_matches<ID: AgentId>(target: &MessageTarget<ID>, agent: &Agent<ID>) -> bool
```
The targeting rules in scanning form, for a transport that keeps its own registry. Agents without an ID (the system agent) are never a delivery target.
[`ConnectionManager`](#struct-connectionmanagerid-agentid) does not use it: it holds an agent-to-connection index and resolves `Agent`/`Agents` by lookup rather than by walking the registry, which is what keeps a per-recipient snapshot pass from costing a scan per recipient. A unit test pins the two to the same answer over every target variant, including the cases where they could plausibly differ (a repeated id in `Agents`, an id nobody holds, the system agent).
### Type Alias `OutboundFrame`
A newtype over a refcounted buffer: one fully-encoded message, kind tag then body, in either direction. **Cloning shares rather than copies**, which is what makes a broadcast to N recipients cost a refcount bump each instead of N allocations, inside `broadcast`'s read guard. It is a newtype rather than an alias so that guarantee belongs to this crate rather than to whichever buffer type it holds.
Get one with `Frame::from(Vec<u8>)` or `Frame::from(bytes::Bytes)`; read it through `AsRef<[u8]>` or `Deref<Target = [u8]>`; ask directly with **`len()`** and **`is_empty()`**; hand the shared buffer to a writer that wants one with `into_bytes()`. A transport that already speaks `Bytes`, which both shipped ones and most WebSocket and QUIC crates do, converts for free in either direction and one that reads into a `Vec<u8>` never needs that crate at all. `OutboundFrame` is an alias for it, under the name a transport adapter meets it by.
### Enum `ConnectionOrder`
What `close_connection` and `set_deadline` send and a connection task receives through `take_orders`. It rides its own channel rather than the outbound queue because the queue's receive arm is disabled the moment `deregister` drops the sender, which is exactly when a close must still work.
* **`Close { farewell: Farewell }`**: flush what is queued, write the farewell's `Goodbye` frame, then close the socket with its code where the transport has one.
* **`Deadline { after: Option<Duration>, farewell: Farewell }`**: arm, replace or clear the task-local deadline; expiry performs the same close. The deadline is task state delivered by order rather than a shared cell, so a custom transport gets it through the same stream it already selects on.
### Struct `IncomingFrame<ID>`
`{ from: Agent<ID>, frame: Frame }`: one inbound frame exactly as it arrived (kind tag, then body), with the `Agent` the transport attached. Refcounted rather than copied out of whatever the socket produced. It survives on the **inbound** path only, where the deserialize bridge decodes a client's raw ops before handing them to the controller; the outbound path encodes once to an [`OutboundFrame`](#type-alias-outboundframe) and never builds one of these.
### Structs `Queues` and `Limits`
```rust
pub struct Queues {
pub inbound: usize, // encoded frames waiting for the deserialize bridge
pub decoded: usize, // decoded messages waiting for the controller
pub presence: usize, // joins and leaves waiting for the controller
pub outbound: usize, // frames waiting to be written to one client
pub conditioner: usize, // frames held per direction per connection, while a LinkProfile is set
}
pub struct Limits {
pub max_frame_bytes: usize, // largest inbound length-delimited frame, TCP only
pub max_message_bytes: usize, // largest inbound message once continuations are joined, WebSocket only
pub inbound_rate: Option<Rate>, // how fast one connection may send; None admits whatever arrives
pub outbound_budget: Option<OutboundBudget>, // what every connection may be sent; None owes every frame
pub credential_timeout: Duration, // how long an admitter's socket may wait for its credential
pub pending_connections: Option<usize>, // sockets waiting for a credential at once; None for no cap
pub max_credential_bytes: usize, // largest credential frame body; larger closes with 1009
}
pub struct Probes {
pub enabled: bool, // false stops this side originating probes
pub slots: usize, // in flight before the oldest is abandoned
pub fast_pings: u32, // sent at fast_interval before settling
pub fast_interval: Duration,
pub idle_interval: Duration,
}
```
`Probes` is a separate struct from `Limits` because it holds a schedule rather than caps. `Probes::off()` and `SessionOptions::without_probes()` stop this session measuring; an inbound `Ping` is still answered, since refusing it would break a peer measuring its own side and `agent_link_rtt` then stays `None`. `ConnectionManager::probes()` reads it back, which is where a transport adapter sets up its timer. Defaults: enabled, 16 slots, 8 probes at 125ms, then one every 5s.
Both are plain structs with `Default`, reachable through [`SessionOptions`](#type-sessionclock-and-struct-sessionoptions) and readable off a manager with `ConnectionManager::queues()` / `limits()`, which is where a transport adapter picks them up. `Queues::for_workload` and `Limits::for_workload` derive them from a [`Workload`](#4-module-workload).
The two byte caps are separate because they bound different mechanisms and each defaults to what its transport enforced before it was nameable. A build serving both transports that wants one number sets both. They bound one frame and `inbound_rate` bounds the stream of them. Together they give a byte ceiling, which is why the [`gate`](#10-module-gate) counts frames rather than bytes. `Limits::for_workload` leaves `inbound_rate` at `None` while deriving both caps. A badly sized cap only wastes memory, while a rate set too low drops a player's input, so applying [`Rate::for_workload`](#4-module-workload) stays a line the application writes.
### Struct `Overflow`
```rust
pub struct Overflow {
pub outbound: OutboundOverflow, // Drop | Disconnect
pub inbound: InboundOverflow, // Drop | Backpressure
pub presence: PresenceOverflow, // Drop | Backpressure
}
```
What each queue does when it is full. `Default` is `Drop` on all three, which is what shipped before the policy existed. Reach it through `SessionOptions::overflow(..)`, the one-call builders `disconnect_slow_clients` / `backpressure_inbound` / `backpressure_presence` or `Overflow::for_workload`. `ConnectionManager::overflow()` reads it back.
`Overflow::drop_everywhere()` is `Default` under a name. `Overflow::block_where_possible()` waits at the two queues that have an arm to wait on and leaves `outbound` on `Drop`.
Three types rather than one shared enum, because the coherent arms differ and a shared enum would make `presence: Disconnect` typeable. `Disconnect` belongs to `outbound` alone: an inbound queue fills because the controller is behind, which names nothing a particular client did and a lost presence event is a bookkeeping failure that disconnecting would compound. There is no `block_everywhere` for the same reason in reverse: `broadcast` fans out under the registry's read guard and has no arm that can wait, so `Overflow::block_where_possible()` leaves `outbound` on `Drop` and says so.
`PresenceOverflow::Backpressure` wedges every connection at registration if the session starts before its controller and the queue fills. `InboundOverflow::Backpressure` is meant to apply TCP backpressure to one client, but a slow controller applies it to all of them at once.
Both are reachable only because `forward_incoming`, `register`, `deregister` and `broadcast`'s disconnect path became `async`; there is no blocking arm without one.
### Constants
* `DEFAULT_BROADCAST_CAPACITY: usize = 256`: `Queues::inbound`, `decoded` and `presence`.
* `DEFAULT_CLIENT_QUEUE_CAPACITY: usize = 64`: `Queues::outbound`.
* `DEFAULT_CONDITIONER_CAPACITY: usize = 1024`: `Queues::conditioner`.
* `DEFAULT_MAX_FRAME_BYTES: usize = 8 * 1024 * 1024`: `Limits::max_frame_bytes`, which is what `LengthDelimitedCodec` enforces without being asked.
* `DEFAULT_MAX_MESSAGE_BYTES: usize = 1024 * 1024`: `Limits::max_message_bytes`.
* `DEFAULT_CREDENTIAL_TIMEOUT: Duration = 5s`: `Limits::credential_timeout`. The timer starts after the handshake, so it only fires for a socket that never presents.
* `DEFAULT_PENDING_CONNECTIONS: usize = 1024`: `Limits::pending_connections`. Over it a socket is refused before the upgrade.
* `DEFAULT_MAX_CREDENTIAL_BYTES: usize = 4096`: `Limits::max_credential_bytes`.
* `DEFAULT_PROBE_SLOTS: usize = 16`: `Probes::slots`, about two seconds of the probe's fast phase.
* `DEFAULT_PROBE_FAST_PINGS: u32 = 8`, `DEFAULT_PROBE_FAST_INTERVAL: Duration = 125ms`, `DEFAULT_PROBE_IDLE_INTERVAL: Duration = 5s`: the schedule.
## 4. Module `workload`
```rust
pub struct Workload {
pub tick_rate: u32,
pub peak_players: usize,
pub ops_per_player_per_tick: u32,
pub stall_tolerance: Duration,
pub join_burst: usize,
pub tick_budget: Option<Duration>,
pub max_payload: usize,
pub memory_budget: Option<usize>,
pub priority: Priority, // LossFree | LatencyFirst
pub socket_buffer_bytes: usize,
}
```
What an application does, in terms its author already knows and the input `Queues::for_workload` and `Limits::for_workload` derive depths from. `SessionOptions::workload(&w)` applies both.
Seven presets, each a `Workload` literal rather than an opaque table, so changing one field of one is the same mechanism as writing your own: `action`, `horde`, `turn_based`, `social_relay`, `spectator`, `lobby`, `local`. A test asserts no two derive the same shape, since a preset that computes the same values as another would only be a second name for it.
Three constants are judgement rather than measurement and say so in their own docs:
* `DEFAULT_SOCKET_BUFFER_BYTES: usize = 540 * 1024`: what the host buffers under the outbound queue, measured on macOS loopback in `benches/saturation.rs` and constant across an 80x range of frame sizes. It is a property of the host, which is why `Workload::socket_buffer_bytes` overrides it.
* `MIN_OUTBOUND_CAPACITY: usize = 4`: floor under a derived per-connection depth, because the derivation subtracts a socket estimate this crate cannot verify and an over-large one concludes the queue is unnecessary. A `memory_budget` that cannot afford the floor still wins, so a build that cannot pay four frames per connection learns it rather than being quietly given them.
* `MIN_CONTROLLER_CAPACITY: usize = 8`: floor under the controller-facing depths. `ops_per_player_per_tick: 0` is a claim about intent and not a guarantee, since a client can always send.
`Priority::headroom` doubles derived depths under `LossFree`: an under-provisioned queue costs correctness there and one superseded frame otherwise.
`conditioner` and `probe_slots` are not derived. What the conditioner holds follows from the `LinkProfile` set at runtime and the probe table from the round trip and the probe schedule; neither is something a workload describes.
**`Rate::for_workload(&w)`** derives a per-connection inbound ceiling from the same numbers: `tick_rate * ops_per_player_per_tick` is what one client's frames per second come to if it does what the workload says, times `RATE_HEADROOM: f64 = 4.0`, floored at `MIN_INBOUND_RATE: f64 = 5.0`, with a burst of one second's worth. Both constants are judgement rather than measurement. The headroom is deliberately large because the input is a claim about intent and the errors are not symmetrical: too high still bounds a flood to a small multiple of honest traffic, too low silently discards ops a client believes arrived. The floor exists because `ops_per_player_per_tick: 0` describes an audience rather than promising silence and a spectator still says hello and asks for a resync.
**`SessionOptions::workload` does not apply it**, unlike every other derivation here. Enforcing a rate is the one that refuses traffic, so it stays a line the application writes: `.rate_limit_inbound(Rate::for_workload(&w))`.
## 5. Module `actix_ws` (feature `actix_ws`)
### Struct `ActixWsPlazaSession<Op, ID, C = JsonCodec>`
* **`new() -> Arc<Self>`**: JSON on the wire, the usual choice for browsers. Only on `C = JsonCodec`.
* **`with_codec(codec: C) -> Arc<Self>`**: declares nothing, so no `Hello` is sent and the version check is off.
* **`with_protocol(codec: C, protocol: ProtocolVersion) -> Arc<Self>`**: announces `protocol` (from [`plaza_wire::build`](../wire/API_REFERENCE.md#6-module-build-feature-build)) as a `Hello` before anything else, so a client learns about a skew on connect rather than by mis-decoding an op. This is the only way a WebSocket session declares a version: without it the handshake is unreachable for exactly the browser and mobile clients it exists for and an installed app cannot be forced to reload the way a page can. `ProtocolVersion::UNKNOWN` is what `with_codec` passes.
* **`with_options(codec: C, options: SessionOptions) -> Arc<Self>`**: `with_protocol` plus a [`SessionClock`](#type-sessionclock-and-struct-sessionoptions) for stamping `Pong.responder`. The other constructors delegate to it with no clock.
* **`handle_connection(&self, req: &HttpRequest, stream: web::Payload, agent: Agent<ID>) -> Result<HttpResponse, actix_web::Error>`** Completes the handshake, registers the connection and spawns its pump. Return the `HttpResponse` from your route. `agent` identifies the client: derive it from an auth token or a query string or mint a fresh id for anonymous play.
* **`admit_connection(&self, req: &HttpRequest, stream: web::Payload, admitter: Arc<dyn ConnectionAdmitter<ID>>) -> Result<HttpResponse, actix_web::Error>`**: the door for a client that cannot present identity before the upgrade, which is every browser and mobile client since neither can set a header. Completes the handshake and holds the socket on its own task, unregistered, until a `Kind::Credential` arrives and `admitter` answers; `Admitted` registers it as that agent and the join fires from there, `Refused` writes the goodbye and closes with its code. See [module `admission`](#12-module-admission) for what may cross the socket meanwhile and the timer that ends a socket presenting nothing. Over [`Limits::pending_connections`](#structs-queues-and-limits) the route answers 503 with no socket created, which a client sees as a failed connect.
* **`connection_rtt(conn_id) -> Option<(Duration, Duration, u64)>`** (smoothed, minimum, samples), **`agent_rtt(id) -> Option<(Duration, u64)>`**, **`connection_link_rtt(conn_id)`**, **`agent_link_rtt(id)`**, **`set_agent_link_profile(id, LinkProfile)`**, **`set_all_link_profiles(LinkProfile)`**, **`link_dropped()`**, **`agent_link_dropped(id)`**, **`stats() -> Arc<TransportStats>`**: available whatever the codec. They read the manager underneath, so a session built `with_codec` gets the same measurements as a JSON one.
* **`protocol(&self, id: &ID) -> Option<ProtocolVersion>`**: what that agent declared in its `Hello` and where an application decides what to do about it. `None` means the peer declared nothing, which is not a mismatch.
* **`manager(&self) -> &Arc<ConnectionManager<ID>>`**: the registry itself, for everything keyed on a connection rather than an agent: `connections_of`, `close_connection` and the per-connection readers.
* **`encode_message(&self, msg) -> Result<OutboundFrame, SessionLayerError>`**: this session's codec, for frames that bypass the targeting path.
* Implements `Session`.
Usage is a five-line route:
```rust,ignore
async fn ws_route(
req: HttpRequest,
stream: web::Payload,
session: web::Data<Arc<ActixWsPlazaSession<Op, PlayerId>>>,
) -> Result<HttpResponse, actix_web::Error> {
let id = Uuid::new_v4();
session.handle_connection(&req, stream, Agent::new_human(id))
}
```
Inbound, text and binary frames are both accepted. Outbound, the frame type follows the codec: [`WireCodec::is_text()`](../wire/API_REFERENCE.md#method-is_text) decides, so a `JsonCodec` sends **text** frames a browser or `websocat` can read and a binary codec sends binary. WebSocket pings are answered automatically, as are `Kind::Ping` frames and the connection deregisters itself when the pump exits.
## 6. Module `tcp` (feature `tcp`)
### Struct `TcpPlazaSession<Op, ID, C = JsonCodec>`
Length-delimited framing over TCP (`tokio_util::codec::LengthDelimitedCodec`). Each frame carries one encoded value.
* **`async bind(addr: impl Into<String>, agent_factory: AgentFactory<ID>) -> Result<Arc<Self>, SessionLayerError>`** JSON on the wire.
* **`async bind_with_codec(addr, agent_factory, codec: C) -> Result<Arc<Self>, SessionLayerError>`**
* **`async bind_with_protocol(addr, agent_factory, codec: C, protocol: ProtocolVersion) -> Result<Arc<Self>, SessionLayerError>`**: announces `protocol` as a `Hello` on every connection.
* **`async bind_with_options(addr, agent_factory, codec: C, options: SessionOptions) -> Result<Arc<Self>, SessionLayerError>`**: `bind_with_protocol` plus a [`SessionClock`](#type-sessionclock-and-struct-sessionoptions) for stamping `Pong.responder`. The other constructors delegate to it.
* **`async bind_with_admitter(addr, admitter: Arc<dyn ConnectionAdmitter<ID>>, codec: C, options: SessionOptions) -> Result<Arc<Self>, SessionLayerError>`**: the factory's counterpart for identity that arrives after the connection: every accepted socket waits, unregistered, for a `Kind::Credential` and registers as whoever `admitter` says; see [module `admission`](#12-module-admission). Over `Limits::pending_connections` a socket is accepted and closed at once.
* **`local_addr(&self) -> SocketAddr`**: resolves `:0` to the assigned port.
* **`manager(&self) -> &Arc<ConnectionManager<ID>>`**, **`encode_message(&self, msg) -> Result<OutboundFrame, SessionLayerError>`**: as on `ActixWsPlazaSession`.
* Implements `Session`. `Drop` aborts the accept loop.
Binding happens **before** the accept loop is spawned, so an address already in use surfaces as `SessionLayerError::Bind` rather than silently killing a detached task. `Kind::Ping` frames are answered on the connection task, so a TCP client is measured without the application doing anything; there is no transport-plane ping here, so the link plane is the only round trip TCP has.
### Type Alias `AgentFactory<ID>`
```rust,ignore
pub type AgentFactory<ID> = Arc<dyn Fn(SocketAddr) -> Result<Agent<ID>, Farewell> + Send + Sync>;
```
Builds the `Agent` for each accepted connection or turns the socket away with a [`Farewell`](#struct-farewell), whose `Goodbye` is the only thing the socket hears. For reconnection support, derive a stable ID here rather than generating a fresh one per connection.
A refusal happens **before** `register`: nothing is allocated, announced or snapshotted for the socket and no presence event fires. The only rules that can be judged here are ones keyed on what a socket shows (its address); a rule keyed on identity belongs in a [`ConnectionAdmitter`](#12-module-admission) behind `bind_with_admitter`. The WS transport needs no factory because the application owns the HTTP route and can refuse before calling `handle_connection`.
Each refusal is counted on [`TransportStats::refused`](#struct-transportstats).
## 7. Module `host` (feature `actix_host`)
The HTTP half of a listen server: bind a port, serve a browser client from it and put the WebSocket route on the same origin so the page connects back to whoever served it. It owns what is the same in every such application and easy to get subtly wrong; the application supplies its own routes, so none of this knows anything about the state being shared.
### Struct `Host`
* **`new(bind: impl Into<String>)`**: the bind address.
* **`serve_dir(Option<String>)`**: the static directory. Preflighted at startup rather than per request, because a missing bundle should fail where you can see it.
* **`cache_bust(asset)`**: stamp this asset's modification time into the dynamically served `index.html`, read per request.
* **`ws_path(path)`**: what the banner prints. **`announce(bool)`**: whether to print it.
* **`run(|cfg| { .. })`**: register your routes and serve. Signal handling is left to the process.
**Cache busting.** It is required. A browser client is a build product that does **not** rebuild when the server does, so a page built before a wire change still loads, still appears to run and only the messages whose shape changed are rejected. This looks like a netcode bug even though the cause is a stale page; it cost two rounds of diagnosis. Two parts are needed together: the stamp and `no-cache` on static assets. Without `no-cache` a cached page keeps quoting the old stamp and cache busting appears not to work. A third part is [`plaza_wire::build`](../wire/API_REFERENCE.md), which gives a client a protocol version to announce so it can be told to reload.
**Signals stay with the process.** Actix catching Ctrl-C for a graceful shutdown while a game window keeps running is why a windowed host could not be killed and why the controller then logged queue-full errors for dead links.
### Struct `SimHost` and struct `SimWiring`
A [`Host`](#struct-host) with the simulation stack behind it: everything needed to take a `StateLogic` to a listening server. It is built from public blocks. Each choice it makes can be undone by using those blocks directly:
* **Joiners get no snapshot** (`snapshot_context_on_join(None)` over `plaza::NoSnapshots`). This stack is for worlds streamed as deltas on a cadence, where a joiner is caught up by the stream itself. A world that catches joiners up with state wants `StateControllerBuilder` and its join snapshot instead.
* **Connections are numbered and the number is the agent id** (`u64`, assigned at accept, never client-supplied), on a `/ws` route it registers itself. An application with identity has its own id type and registers its own route on a plain `Host`.
* **The driver is `run_fixed` by default**: delivering measured elapsed time would make the simulation's rate a property of the host's scheduler, which no predicting, replaying or rolling-back client can reproduce. **`SimHost::measured(bind, tick_hz)`** is the deliberate exception for logic that only integrates over elapsed time (corrections-based prediction is the standing example, `blackhole_playground`); there the driver is `run` at `tick_hz` and `delta_time` is what the scheduler delivered.
```rust,ignore
SimHost::new(bind, Duration::from_millis(SIM_STEP_MS))
.serve_dir(static_dir)
.cache_bust("my_game.wasm")
.run(MsgPackCodec, PROTOCOL, Arena::new(initial), |wiring| {
ArenaLogic::new(controls, view)
.with_link(wiring.link_sink())
.with_clock(wiring.sim_clock.clone())
})
.await
```
* **`new(bind, step)`**: `step` is the simulation's step, the unit its ticks are counted in. **`measured(bind, tick_hz)`**: the measured-time variant above. **`serve_dir`**, **`cache_bust`**, **`announce`** forward to `Host`. **`wake_hz(hz)`** (default `DEFAULT_WAKE_HZ` = 120 under `new`, `tick_hz` under `measured`: waking more often than you step keeps the phase error small) and **`command_buffer(n)`** (default `DEFAULT_COMMAND_BUFFER` = 256).
* **`run(codec, protocol, initial_state, logic_for)`**: builds the session with the given protocol version and a simulation clock for its pongs, the controller, the fixed-step driver and the route, then serves until the process ends. `logic_for` receives a **`SimWiring`** (`session: Arc<ActixWsPlazaSession<Op, u64, C>>`, `sim_clock: Arc<AtomicU64>` and **`link_sink()`**, the usual destination for a panel's impairment sliders) so measurement sources and sinks can be wired into the logic it returns. Store the simulation's clock into `sim_clock` each tick and clients synchronise against simulation time rather than wall time.
### Function `lan_address() -> Option<String>`
A local address somebody else could actually reach. No dependency and no packets: connecting a UDP socket only picks a route, so the kernel fills in the source address it would use. Printing it matters because knowing a server is running does not tell you what address to send a friend.
### Function `init_logging()`
Turns on a console subscriber, once. `plaza` and `plaza_session` are instrumented throughout, but `tracing` is silent without a subscriber and a server that logs nothing is indistinguishable from a server that is not running. A convenience for binaries: it is a no-op after the first call and after any other global subscriber is installed and `RUST_LOG` overrides the default. A library or an application with its own subscriber should not call it.
**What is deliberately not here.** Deciding what a process *is* (headless, observer, host, joiner) and parsing that off a command line. The browser client needs the same vocabulary, a wasm bundle must not inherit an HTTP server to learn the name of its own role and every real application already parses its own arguments. That lives in `examples/playground_common/` as shared scaffolding rather than in a library crate.
## 8. Module `stats`
### Struct `TransportStats`
Live counters for one transport, from `ActixWsPlazaSession::stats` or `ConnectionManager::stats`.
* **`inbound()`** / **`inbound_dropped()`** / **`inbound_shed()`**, **`outbound()`** / **`outbound_bytes()`** / **`outbound_dropped()`** / **`outbound_withheld()`**, **`presence_dropped()`**, **`refused()`**, **`pending()`** / **`admitted()`** / **`timed_out()`** / **`over_cap()`**.
* `inbound_dropped` and `inbound_shed` differ in who is at fault and who pays. A drop means the controller fell behind, names nothing a client did and costs whoever happened to be sending; a shed names one connection that exceeded its [rate](#10-module-gate) and costs only that connection. A server seeing both should read the shed first.
* `outbound_bytes` counts the frame once per recipient it was queued for, because that is what the sockets will carry; per-connection figures are `ConnectionManager::connection_outbound` / `agent_outbound`.
* `outbound_withheld` counts the times a budgeted connection was asked for and not owed a frame (see [module `budget`](#11-module-budget)). It is the budget's cost in frames not built and it climbs only when a snapshot pass asks.
* **`record_refused()`**: what a transport calls when it turns a socket away before `register`. Nothing else here can see such a socket. A pending socket closed with a goodbye lands here too, whether the admitter refused its credential or it sent ops before one and got 4401.
* `pending` is a gauge: sockets waiting on a credential right now. `admitted`, `timed_out` and `over_cap` count how pending phases ended: registered, closed with 4408 for presenting nothing or turned away before the upgrade because [`Limits::pending_connections`](#structs-queues-and-limits) was full. Together with `refused` they are the evidence the cap's default rests on.
The fan-out uses `try_send` by default: a wedged client must not stall the controller. The drop used to be announced only with `warn!`, which a human reads afterwards and a server cannot read at all, so the events are countable and an application can shed load deliberately instead of degrading quietly. What the default does when a queue fills is now [`Overflow`](#struct-overflow)'s to say and the counters read the same whichever arm is chosen.
**Totals are kept beside the drops** because a drop count alone cannot tell "nothing is being dropped" from "nothing is being sent".
**The three are separate on purpose.** An outbound drop is usually benign for a stream of absolute state, since the next frame supersedes it. An inbound drop is ops a client already sent and believes arrived and nothing upstream will retry, so it is lost player input. A presence drop is a correctness failure from a single occurrence: a lost join leaves the controller with a client it never heard of, a lost leave leaves it holding a seat forever. A single combined number would hide presence drops behind the benign outbound ones.
## 9. The control plane in pieces
The pieces [`LinkDriver`](#struct-linkdriverid-c) assembles, each usable alone by an adapter that needs a different arrangement.
### Struct `LinkDriver<ID, C>`
One connection's link plane: impairment both ways, probes and their deadlines. It holds a [`LinkHandle`](#struct-linkhandle) rather than reading the profile off the registry.
* **`new(manager, conn_id, codec) -> Option<Self>`**: `None` when `conn_id` is not registered, which means a transport called this before `register` or after `deregister`.
* **`inbound(frame, now) -> Inbound`**: answers probes, times the ones this side sent and applies upstream impairment.
* **`outbound(frame, now) -> Option<Frame>`**: `None` when the frame was queued behind downstream impairment or thrown away by a `Datagram` profile's loss and the frame itself when it may go straight out.
* **`deadline() -> Option<Instant>`**: the earliest of the next probe and the two conditioners' heads, which is what the connection task sleeps until.
* **`due(now) -> Vec<Frame>`**: everything owed to the socket now, in order: released downstream frames, replies to probes that came out of the upstream queue and a fresh probe if one fell due.
* **`take_forwarded() -> Vec<Frame>`**: inbound frames whose delay has expired, for `forward_incoming`. Separate from `due`'s return because these go to the application rather than the socket and that call is `async` while `due` is not.
* **`ejected() -> bool`**: whether a frame the upstream queue released exceeded a rate that ends connections. A held frame is judged when the link releases it and `due` hands back what the socket is owed, so a close cannot travel that way. The direct path needs no flag: `inbound` returns `Inbound::Eject` to its caller.
### Enum `Inbound`
What `LinkDriver::inbound` decided: **`Forward(Frame)`** (not ours, hand it to the application), **`Reply(Frame)`** (ours and it wants an answer written back to the peer), **`Consumed`** (ours and finished with), **`Shed`** (theirs, over the [rate](#10-module-gate), dropped before the shared queue), **`Eject`** (the same, on a rate that ends connections).
`control::route_inbound` folds the same decision into **`Routed::{Nothing, Reply(Frame), Eject}`** for a transport that wants delivery done for it: `Nothing` covers delivered, answered internally and shed alike, because all three leave the socket with nothing to write.
### Struct `LinkHandle`
One connection's impairment, taken from [`ConnectionManager::link_handle`](#struct-connectionmanagerid-agentid) and held by the connection task for its lifetime.
The profile is 80 bytes and cannot be an atomic; the question the frame path asks is one bit. **`impaired() -> bool`** is that bit, one relaxed load. **`read() -> LinkProfile`** takes the lock, for the frames that need the numbers. The flag and the profile are not written atomically together, so a profile installed between one frame and the next may miss that frame. That is deliberate: the alternative is a lock on the path this type exists to keep clear.
**`generation() -> u64`** counts settings of the profile, so a connection task can tell the profile moved without holding a copy of it. A probe sent under one generation whose `Pong` arrives under another measured a path that existed for neither leg's whole journey and recording it would leave a wrong minimum for the life of the connection.
### Struct `ProbeState` and function `make_probe`
The probe table and its schedule, one per connection.
* **`new(&Probes) -> Self`**: from the schedule on [`Probes`](#structs-queues-and-limits).
* **`interval() -> Option<Duration>`**: until the next probe: `fast_interval` for the first `fast_pings`, then `idle_interval`. `None` when this session does not probe.
* **`first_due(now) -> Option<Instant>`**: when the first one is due.
* **`forget_in_flight()`**: discards every outstanding probe without touching the sequence number, so their pongs answer no open probe and are dropped. Call it when `generation()` moves.
* **`make_probe(&codec, &mut ProbeState, now) -> Frame`**: builds the next probe. Its origin is a sequence number, not a clock reading: the session times the round trip with an `Instant` it keeps, so its own probes need no clock and no unit.
### Function `earliest` and constant `DOWN_SEED_FLIP`
**`earliest(a: Option<Instant>, b: Option<Instant>) -> Option<Instant>`**: the sooner of two optional deadlines or whichever one exists.
**`DOWN_SEED_FLIP: u64`**: xor a connection id with this to seed the downstream conditioner. Seeding both directions from the connection id makes an impaired session re-run identically. The flip keeps the two directions from drawing the same jitter sequence.
## 10. Module `gate`
How fast one connection may speak, judged on the connection task before the queue every connection shares.
```rust
pub struct Rate {
pub per_sec: f64, // sustained frames per second
pub burst: u32, // frames admissible at once from a full bucket
pub over: Over, // Shed | Close
}
pub enum Verdict { Admit, Shed, Close }
```
A token bucket, reached through [`SessionOptions::rate_limit_inbound`](#type-sessionclock-and-struct-sessionoptions) or by setting [`Limits::inbound_rate`](#structs-queues-and-limits). Both numbers matter: a sustained rate alone punishes an honest client whose packets arrived in a clump after a hiccup and a burst alone is a flood budget that refills the moment it is spent. A bucket starts full and never holds more than `burst`, so a quiet period earns no extra allowance.
* **`Rate::per_second(f64)`**: a sustained rate with a burst of one second's worth, shedding what it cannot pay for. **`burst(u32)`** and **`disconnecting()`** adjust it. **`Rate::for_workload`** is in [module `workload`](#4-module-workload).
* **`Over::Shed`** discards the frame and keeps the connection, which is right where the traffic is an eager client rather than an attack. **These are ops the client believes arrived**, so a stream where each op matters exactly once wants `Close` or a rate it will not hit.
* **`Over::Close`** ends the connection, for a rate no honest client of the build can reach. The WebSocket transport closes with `1008 Policy Violation`; TCP has no code and drops the connection.
**It runs before the shared queue.** A flood dropped at the shared inbound queue has already cost every other client the frames behind it, so the judgement is one step earlier, per connection, before anything shared is touched. **It reads no content**: at that point a frame is still encoded bytes and opening them would put a decode on the path a flood is trying to saturate.
**It counts frames rather than bytes**, because a frame's size is already bounded by `max_frame_bytes` / `max_message_bytes`: frames per second times the largest one is a byte ceiling without a second number to keep.
**There is no default.** A session with no rate admits everything, exactly as it did before this module existed. Only the application knows how fast is too fast and whether that is worth ending a connection over, the same division [`Overflow`](#struct-overflow) and the AFK rule already draw.
Shed frames are counted in three places: `TransportStats::inbound_shed` for the session and `InboundVolume::shed` per connection and per agent, which is what an application escalates on. A shed frame still counts as a frame and still stamps activity, since it did arrive and a flooding client is not idle.
## 11. Module `budget`
How much one connection may be sent, judged where the frames are decided on rather than in the transport.
```rust
pub struct OutboundBudget {
pub bytes_per_sec: Option<f64>, // sustained bytes a second, None for unbounded
pub frames_per_sec: Option<f64>, // sustained frames a second, None for unbounded
pub burst: Duration, // unspent credit a connection may hold, as time at the rate; one second by default
}
```
The outbound twin of the [`gate`](#10-module-gate). A client on a slow link is sent what a client on fibre is sent and the only thing between it and its link is its bounded outbound queue, so its failure is binary: keep up or lose frames, be disconnected or stall the controller, whichever [`Overflow`](#struct-overflow) says. A budget is the middle where it is sent *fewer, complete* frames.
**The transport never withholds.** `broadcast` still queues every frame it is handed and charges each recipient's credit for it, ops and events as well as snapshots, because the link carries all of it. What the budget adds is an answer to the question a snapshot pass asks before it builds for a recipient: `ConnectionManager::connection_owed` / `agent_owed`. A `SnapshotProvider` that returns `Ok(None)` for a recipient not owed a frame has skipped it and a skip there costs latency and never correctness, since a delta stream's baseline is what was acknowledged and the next frame carries everything since. A skip in the transport would have to drop a frame already built, which is the failure the budget replaces.
Credit runs negative: a frame larger than what is left still goes and the connection is owed nothing until the debt refills at its rate. That is what turns a budget into a cadence, at 10 KiB a second and 2 KiB a frame, five whole frames a second. A new budget starts with a full burst and an idle connection banks no more than one.
* **`OutboundBudget::DEFAULT_BURST: Duration = 1s`**: the `burst` a new budget starts with.
* **`OutboundBudget::bytes_per_second(f64)`**, **`OutboundBudget::frames_per_second(f64)`**: one bound, the other unbounded. **`and_bytes_per_second`**, **`and_frames_per_second`** add the other; both then have to hold. **`burst(Duration)`** sets how far ahead a connection may run; it must be positive.
* Reached through [`SessionOptions::budget_outbound`](#type-sessionclock-and-struct-sessionoptions) or `Limits::outbound_budget` for every connection at registration or per connection with `ConnectionManager::set_outbound_budget` / `set_agent_outbound_budget`, read back with `outbound_budget`. A client declaring what its link can carry and the server clamping the declaration are both the application's lines to write.
* Refusals are counted in `OutboundVolume::withheld` per connection and per agent and in `TransportStats::outbound_withheld` for the session: frames a snapshot pass did not build, which the sent counts cannot show.
**There is no default.** A connection with no budget is always owed a frame, as every connection was before this module existed.
## 12. Module `admission`
Admitting a connection on what it presents and how every close says why.
A route that resolves identity before the upgrade hands the transport an `Agent` and the connection registers at once. A route that cannot hands it a `ConnectionAdmitter` instead, through [`admit_connection`](#struct-actixwsplazasessionop-id-c--jsoncodec) or [`bind_with_admitter`](#struct-tcpplazasessionop-id-c--jsoncodec). The socket then waits on its own task, unregistered, until a `Kind::Credential` arrives, the admitter answers and the connection registers as whoever it said. Nothing crosses an unadmitted socket in either direction: inbound, a `Hello` is kept and recorded on admission, probes are neither answered nor forwarded, an unknown kind is skipped and an `Ops` frame closes the socket with `Goodbye::CREDENTIAL_EXPECTED` (4401); outbound, a refusal is a `Goodbye` and nothing else. A socket that presents nothing within [`Limits::credential_timeout`](#structs-queues-and-limits) is closed with `Goodbye::CREDENTIAL_TIMEOUT` (4408) and a credential larger than `Limits::max_credential_bytes` with 1009.
Holding the socket on its task rather than in the registry was measured: a parked socket costs about 23 KiB and 45 us where a registered one costs about 48 KiB and 215 us and keeps costing probe CPU while idle, so a flood of sockets that never present pays for nothing but the socket.
### Struct `Farewell`
```rust
pub struct Farewell { pub code: u16, pub detail: Option<Vec<u8>> }
impl Farewell {
pub fn new(code: u16) -> Self;
pub fn with_detail(self, detail: impl Into<Vec<u8>>) -> Self;
pub fn credential_expected() -> Self; // 4401
pub fn credential_timeout() -> Self; // 4408
pub fn encode<C: WireCodec>(&self, codec: &C) -> OutboundFrame;
}
```
The one vocabulary for every server-initiated close: a refusal, the pending timer, `close_connection`, `set_deadline`, `deregister_agent`, `disconnect_all` and the factory's refusal all take one. Each transport writes its `Goodbye` frame (`encode`) last, after flushing what was queued; on WebSocket it then closes with the same `code` and on TCP the frame is all there is. `code` is a WebSocket close code whatever the transport. RFC 6455 gives 4000 to 4999 to the application and the application supplies the code on every close it orders, with no default; the two the session sends by itself are the constructors above and a connection ejected for exceeding its inbound rate hears 1008 (`POLICY_VIOLATION`). `detail` is whatever is worth saying beside the code, encoded however the application chooses; the session never reads it.
The client libraries turn the goodbye into their disconnected event and stop reconnecting on a 4xxx by default, because a server that refused a credential would refuse it again. The code says what class of close this is and the detail says what to do next; neither should say why in enough detail to serve as an oracle.
### Trait `ConnectionAdmitter<ID>` and enum `ConnectionAdmission<ID>`
```rust
#[async_trait]
pub trait ConnectionAdmitter<ID: AgentId>: Send + Sync {
async fn admit(&self, credential: &[u8], peer: &Peer) -> ConnectionAdmission<ID>;
}
pub enum ConnectionAdmission<ID: AgentId> {
Admitted(Agent<ID>),
Refused(Farewell),
}
```
The gate a socket passes to enter the session and as whom. `credential` is the body of the client's `Kind::Credential` frame exactly as it arrived; plaza verifies nothing itself, so what it is and what it proves are the application's. Async because most identity lives in a store and one boxed future per connection open is nothing next to the handshake before it. `Admitted` registers the connection as that agent and the join is announced from there; `Refused` writes the goodbye and closes; nothing was registered or announced.
An admitter that holds the `ConnectionManager` can settle a duplicate login itself: under kick-the-oldest it calls `deregister_agent` on the older id before returning `Admitted`, which works because the older connection is registered and the newcomer is not yet. `examples/door_policy` is the worked example: ban, per-address cap, capacity and duplicate login all judged before anything registers.
### Struct `Peer`
```rust
#[non_exhaustive]
pub struct Peer { pub addr: Option<SocketAddr> }
impl Peer { pub fn new(addr: Option<SocketAddr>) -> Self; }
```
What the transport knows about a socket before anything has been presented. The address is there for rate-limiting bad attempts per IP and for audit. Anything from the upgrade request itself, a forwarded-for header behind a proxy or a room in the URL, is the route's to capture: build the admitter per request and close over it, since one `Arc` per connect costs nothing.
### Struct `Pending` and enum `Presented`
```rust
pub struct Pending { .. }
impl Pending {
pub fn new(max_credential_bytes: usize) -> Self;
pub fn on_frame<C: WireCodec>(&mut self, codec: &C, bytes: &[u8]) -> Presented;
pub fn declared(&self) -> Option<ProtocolVersion>;
}
pub enum Presented {
Nothing,
Credential(Vec<u8>),
Refuse(Farewell),
}
```
The rules for a socket that has not been admitted, shared by both shipped transports and available to a custom one: feed it every inbound frame until it yields the credential or a refusal, run the timer beside it, then hand the socket to the ordinary pump and record `declared()` against the agent once there is one. `MESSAGE_TOO_BIG` (1009) and `POLICY_VIOLATION` (1008) are the two standard codes the module names.
## 13. Error Handling
### Enum `SessionLayerError`
Transport failures. Deliberately **non-generic**: these concern sockets and wire formats, not application agent IDs.
* **Variants**:
* `Bind { addr: String, source: std::io::Error }`
* `Serialization { transport: &'static str, context: &'static str, source: Box<dyn Error + Send + Sync> }`
* `Deserialization { transport, context, source }`
* `ClientSendFailed { transport, conn_id: ConnectionId, reason: &'static str }`
* **Conversion**: `impl<ID: AgentId> From<SessionLayerError> for PlazaError<ID>`. Serialization and deserialization map onto the matching `PlazaError` variants; everything else becomes `PlazaError::Session(TransportError(..))` using `Display`, so the `#[source]` chain stays readable.