Say no to a connection: admission, kicks and timeouts

For Rust developers running a plaza server who need to refuse, remove and time out connections

You will build a small chat lounge on plaza_session's TCP transport on localhost and give its server the right to say no. A door reads each client's credential before the connection registers, so a banned name, an unreadable credential and a duplicate login are all settled before the lounge hears of anyone. Inside, the host can kick a guest with a reason that the guest reads before its socket closes. The lounge can also tell that kick apart from a dropped link. An OpGuard refuses ops before the rules see them. Finally, an idle rule and a session deadline close connections on a timer.

Before you start

You need Rust with Cargo. plaza comes from crates.io:

plaza = "0"

Set up the project#

Start a new binary crate.

cargo new lounge
cd lounge

plaza holds the controller and your rules. plaza_session puts a real transport under them; this tutorial only needs its TCP transport and JSON, so the default WebSocket features are turned off. plaza_wire is the wire format, which the test client uses to write frames by hand. Your [dependencies] section should look like this:

Cargo.toml
[dependencies]
async-trait = "0.1"
parking_lot = "0.12"
plaza = "0"
plaza_session = { version = "0", default-features = false, features = ["tcp", "json"] }
plaza_wire = "0"
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }

The project has four source files. src/lounge.rs holds the rules, src/door.rs decides who gets in, src/client.rs is a small test client and src/main.rs runs each scenario. The server and its clients all live in one process, but every client talks to the server over a real TCP socket on 127.0.0.1.

Open the lounge behind a door#

Start with the rules. A guest who arrives is welcomed by name and anything a guest says is repeated to everyone. Write src/lounge.rs:

src/lounge.rs
use std::sync::Arc;

use async_trait::async_trait;
use plaza::session::TargetedOp;
use plaza::state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError};
use serde::{Deserialize, Serialize};

use crate::door::Door;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum LoungeOp {
    Say(String),
    Welcome { name: String },
    Said { who: String, text: String },
}

#[derive(Debug, Default)]
pub struct LoungeState {
    pub guests: Vec<u64>,
}

pub struct Lounge {
    pub door: Arc<Door>,
}

#[async_trait]
impl StateLogic<LoungeOp, u64, LoungeState> for Lounge {
    async fn process_input(
        &self,
        state: &mut LoungeState,
        input: LogicInput<LoungeOp, u64>,
    ) -> Result<LogicOutput<LoungeOp, u64>, StateLogicError> {
        match input {
            LogicInput::AgentJoined { agent } => {
                let Some(key) = agent.id_cloned() else {
                    return Ok(LogicOutput::none());
                };
                state.guests.push(key);
                let name = self.door.name_of(key);
                let welcome = LoungeOp::Welcome { name };
                Ok(vec![TargetedOp::new_system_to(key, vec![welcome])].into())
            }
            LogicInput::AgentOps { source, ops } => {
                let Some(key) = source.id_cloned() else {
                    return Ok(LogicOutput::none());
                };
                let who = self.door.name_of(key);
                let mut out = Vec::new();
                for op in ops {
                    if let LoungeOp::Say(text) = op {
                        let said = LoungeOp::Said {
                            who: who.clone(),
                            text,
                        };
                        out.push(TargetedOp::new_system_all(vec![said]));
                    }
                }
                Ok(out.into())
            }
            LogicInput::AgentLeft { agent_id } => {
                state.guests.retain(|key| *key != agent_id);
                self.door.left(agent_id);
                Ok(LogicOutput::none())
            }
            _ => Ok(LogicOutput::none()),
        }
    }
}

The lounge never sees a socket. It gets AgentJoined, AgentOps and AgentLeft from the controller and returns the ops to send.

The door is a ConnectionAdmitter. A client sends a Kind::Credential frame after it connects, the session holds the socket unregistered until the admitter answers and only an Admitted answer registers it. Here the credential is simply the guest's name. Plaza verifies nothing itself, so a real door would check a token here instead. Write src/door.rs:

src/door.rs
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};

use async_trait::async_trait;
use parking_lot::Mutex;
use plaza::agent::Agent;
use plaza_session::{ConnectionAdmission, ConnectionAdmitter, Farewell, Peer};

pub const UNREADABLE: u16 = 4400;

#[derive(Default)]
pub struct Door {
    names: Mutex<HashMap<u64, String>>,
    next_key: AtomicU64,
}

impl Door {
    pub fn name_of(&self, key: u64) -> String {
        self.names.lock().get(&key).cloned().unwrap_or_default()
    }

    pub fn left(&self, key: u64) {
        self.names.lock().remove(&key);
    }
}

#[async_trait]
impl ConnectionAdmitter<u64> for Door {
    async fn admit(&self, credential: &[u8], _peer: &Peer) -> ConnectionAdmission<u64> {
        let Ok(name) = std::str::from_utf8(credential) else {
            return refuse(UNREADABLE, "unreadable credential");
        };
        let key = self.next_key.fetch_add(1, Ordering::Relaxed) + 1;
        self.names.lock().insert(key, name.to_owned());
        ConnectionAdmission::Admitted(Agent::new_human(key))
    }
}

fn refuse(code: u16, why: &str) -> ConnectionAdmission<u64> {
    ConnectionAdmission::Refused(Farewell::new(code).with_detail(why.as_bytes()))
}

The door mints a fresh u64 key for every connection it admits and keeps the name beside it. That key is the agent id the controller sees. Giving each connection its own id matters later: when the same name logs in twice, the two connections need two ids to act on. Every AgentLeft then names exactly one of them.

The client speaks the wire by hand. On TCP each frame is a 4-byte length followed by the frame itself. Every frame starts with a kind byte. The client sends a Credential frame and Ops frames, answers the session's latency probes with answer_ping and keeps the Goodbye frame the server writes before it closes a connection. Each thing it hears goes on a channel and hear prints the next one. Write src/client.rs:

src/client.rs
use std::fmt;
use std::sync::Arc;

use plaza_wire::frame::{self, Goodbye, Kind};
use plaza_wire::framing::{LengthDelimited, delimit};
use plaza_wire::{JsonCodec, WireCodec};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::net::tcp::OwnedWriteHalf;
use tokio::sync::{Mutex, mpsc};

use crate::lounge::LoungeOp;

const MAX_FRAME_BYTES: usize = 64 * 1024;

enum Heard {
    Ops(Vec<LoungeOp>),
    Goodbye(Goodbye),
    Closed,
}

impl fmt::Display for Heard {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Heard::Ops(ops) => write!(f, "{ops:?}"),
            Heard::Goodbye(bye) => {
                let detail = bye
                    .detail
                    .as_deref()
                    .map(String::from_utf8_lossy)
                    .unwrap_or_default();
                write!(f, "goodbye {} {detail:?}", bye.code)
            }
            Heard::Closed => write!(f, "socket closed"),
        }
    }
}

pub struct Client {
    name: &'static str,
    writer: Arc<Mutex<OwnedWriteHalf>>,
    heard: mpsc::UnboundedReceiver<Heard>,
}

impl Client {
    pub async fn connect(addr: &str, name: &'static str) -> std::io::Result<Self> {
        let (mut read_half, write_half) = TcpStream::connect(addr).await?.into_split();
        let writer = Arc::new(Mutex::new(write_half));
        let (tx, heard) = mpsc::unbounded_channel();

        let pong_writer = Arc::downgrade(&writer);
        tokio::spawn(async move {
            let mut framing = LengthDelimited::new(MAX_FRAME_BYTES);
            let mut chunk = [0u8; 8192];
            loop {
                while let Ok(Some(bytes)) = framing.next_frame() {
                    let Some((tag, body)) = frame::split(&bytes) else {
                        continue;
                    };
                    match Kind::from_byte(tag) {
                        Some(Kind::Ops) => {
                            if let Ok(ops) = JsonCodec.decode::<Vec<LoungeOp>>(body) {
                                let _ = tx.send(Heard::Ops(ops));
                            }
                        }
                        Some(Kind::Ping) => {
                            let pong = frame::answer_ping(&JsonCodec, body, None);
                            if let (Some(pong), Some(writer)) = (pong, pong_writer.upgrade()) {
                                let mut wire = Vec::new();
                                delimit(&pong, &mut wire);
                                let _ = writer.lock().await.write_all(&wire).await;
                            }
                        }
                        Some(Kind::Goodbye) => {
                            if let Some(bye) = frame::decode_goodbye(&JsonCodec, &bytes) {
                                let _ = tx.send(Heard::Goodbye(bye));
                            }
                        }
                        _ => {}
                    }
                }
                match read_half.read(&mut chunk).await {
                    Ok(0) | Err(_) => {
                        let _ = tx.send(Heard::Closed);
                        return;
                    }
                    Ok(n) => framing.feed(&chunk[..n]),
                }
            }
        });

        Ok(Self {
            name,
            writer,
            heard,
        })
    }

    pub async fn present(&self, credential: &[u8]) -> std::io::Result<()> {
        let mut body = Vec::new();
        frame::begin(Kind::Credential, &mut body);
        body.extend_from_slice(credential);
        self.write_frame(&body).await
    }

    pub async fn say(&self, op: LoungeOp) -> std::io::Result<()> {
        let mut body = Vec::new();
        frame::begin(Kind::Ops, &mut body);
        JsonCodec
            .encode_into(&vec![op], &mut body)
            .expect("ops encode");
        self.write_frame(&body).await
    }

    pub async fn hear(&mut self) {
        let heard = self.heard.recv().await.unwrap_or(Heard::Closed);
        println!("{}: {heard}", self.name);
    }

    async fn write_frame(&self, body: &[u8]) -> std::io::Result<()> {
        let mut wire = Vec::new();
        delimit(body, &mut wire);
        self.writer.lock().await.write_all(&wire).await
    }
}

The reader task only holds a weak handle on the writer. Dropping a Client therefore drops the write half, which closes the socket without any goodbye. A later step uses that to fake a dropped link.

Now src/main.rs. bind_with_admitter binds the listener with the door in front of it. The controller uses without_snapshots because a guest needs no catch-up when it joins:

src/main.rs
mod client;
mod door;
mod lounge;

use std::sync::Arc;

use plaza::controller::StateControllerBuilder;
use plaza_session::SessionOptions;
use plaza_session::codec::JsonCodec;
use plaza_session::tcp::TcpPlazaSession;

use crate::client::Client;
use crate::door::Door;
use crate::lounge::{Lounge, LoungeOp, LoungeState};

const ADDR: &str = "127.0.0.1:5800";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let door = Arc::new(Door::default());
    let session = TcpPlazaSession::<LoungeOp, u64>::bind_with_admitter(
        ADDR,
        door.clone(),
        JsonCodec,
        SessionOptions::default(),
    )
    .await?;

    let lounge = Arc::new(Lounge { door: door.clone() });
    let (_commands, controller) =
        StateControllerBuilder::without_snapshots(lounge, session.clone(), LoungeState::default())
            .build();
    tokio::spawn(controller.run());

    let mut alice = Client::connect(ADDR, "alice").await?;
    alice.present(b"alice").await?;
    alice.hear().await;

    let mut bob = Client::connect(ADDR, "bob").await?;
    bob.present(b"bob").await?;
    bob.hear().await;

    alice.say(LoungeOp::Say("hi bob".into())).await?;
    alice.hear().await;
    bob.hear().await;

    let stats = session.manager().stats();
    println!("admitted {}, refused {}", stats.admitted(), stats.refused());
    Ok(())
}

Every hear waits for the next thing that client receives, so the program prints in the same order on every run. Run it:

$ cargo run
alice: [Welcome { name: "alice" }]
bob: [Welcome { name: "bob" }]
alice: [Said { who: "alice", text: "hi bob" }]
bob: [Said { who: "alice", text: "hi bob" }]
admitted 2, refused 0

admitted and refused come from the transport's own counters. Both guests went through the door and nobody was refused yet.

Refuse a bad credential at the door#

The door can refuse as well as admit. A refusal is a ConnectionAdmission::Refused carrying a Farewell: a close code plus detail bytes the session writes into a Goodbye frame before closing the socket. Codes 4000 to 4999 belong to the application. Add a ban list to src/door.rs:

src/door.rs
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};

use async_trait::async_trait;
use parking_lot::Mutex;
use plaza::agent::Agent;
use plaza_session::{ConnectionAdmission, ConnectionAdmitter, Farewell, Peer};

pub const UNREADABLE: u16 = 4400;
pub const BANNED: u16 = 4403;

#[derive(Default)]
pub struct Door {
    names: Mutex<HashMap<u64, String>>,
    banned: Mutex<Vec<String>>,
    next_key: AtomicU64,
}

impl Door {
    pub fn ban(&self, name: &str) {
        self.banned.lock().push(name.to_owned());
    }

    pub fn name_of(&self, key: u64) -> String {
        self.names.lock().get(&key).cloned().unwrap_or_default()
    }

    pub fn left(&self, key: u64) {
        self.names.lock().remove(&key);
    }
}

#[async_trait]
impl ConnectionAdmitter<u64> for Door {
    async fn admit(&self, credential: &[u8], _peer: &Peer) -> ConnectionAdmission<u64> {
        let Ok(name) = std::str::from_utf8(credential) else {
            return refuse(UNREADABLE, "unreadable credential");
        };
        if self.banned.lock().iter().any(|banned| banned == name) {
            return refuse(BANNED, "banned");
        }
        let key = self.next_key.fetch_add(1, Ordering::Relaxed) + 1;
        self.names.lock().insert(key, name.to_owned());
        ConnectionAdmission::Admitted(Agent::new_human(key))
    }
}

fn refuse(code: u16, why: &str) -> ConnectionAdmission<u64> {
    ConnectionAdmission::Refused(Farewell::new(code).with_detail(why.as_bytes()))
}

The session also refuses two kinds of socket on its own. A socket that sends an Ops frame before its credential is closed with 4401. A socket that presents nothing within Limits::credential_timeout is closed with 4408. The default timeout is five seconds, which this program shortens to 300 milliseconds so it does not have to wait. Replace src/main.rs:

src/main.rs
mod client;
mod door;
mod lounge;

use std::sync::Arc;
use std::time::Duration;

use plaza::controller::StateControllerBuilder;
use plaza_session::SessionOptions;
use plaza_session::codec::JsonCodec;
use plaza_session::tcp::TcpPlazaSession;

use crate::client::Client;
use crate::door::Door;
use crate::lounge::{Lounge, LoungeOp, LoungeState};

const ADDR: &str = "127.0.0.1:5801";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let door = Arc::new(Door::default());
    door.ban("mallory");

    let mut options = SessionOptions::default();
    options.limits.credential_timeout = Duration::from_millis(300);
    let session = TcpPlazaSession::<LoungeOp, u64>::bind_with_admitter(
        ADDR,
        door.clone(),
        JsonCodec,
        options,
    )
    .await?;

    let lounge = Arc::new(Lounge { door: door.clone() });
    let (_commands, controller) =
        StateControllerBuilder::without_snapshots(lounge, session.clone(), LoungeState::default())
            .build();
    tokio::spawn(controller.run());

    let mut alice = Client::connect(ADDR, "alice").await?;
    alice.present(b"alice").await?;
    alice.hear().await;

    let mut mallory = Client::connect(ADDR, "mallory").await?;
    mallory.present(b"mallory").await?;
    mallory.hear().await;
    mallory.hear().await;

    let mut garbled = Client::connect(ADDR, "garbled").await?;
    garbled.present(&[0xff, 0xfe]).await?;
    garbled.hear().await;
    garbled.hear().await;

    let mut eager = Client::connect(ADDR, "eager").await?;
    eager.say(LoungeOp::Say("let me in".into())).await?;
    eager.hear().await;
    eager.hear().await;

    let mut silent = Client::connect(ADDR, "silent").await?;
    silent.hear().await;
    silent.hear().await;

    let stats = session.manager().stats();
    println!(
        "admitted {}, refused {}, timed out {}",
        stats.admitted(),
        stats.refused(),
        stats.timed_out()
    );
    Ok(())
}

Run it:

$ cargo run
alice: [Welcome { name: "alice" }]
mallory: goodbye 4403 "banned"
mallory: socket closed
garbled: goodbye 4400 "unreadable credential"
garbled: socket closed
eager: goodbye 4401 ""
eager: socket closed
silent: goodbye 4408 ""
silent: socket closed
admitted 1, refused 3, timed out 1

Every refused socket hears its goodbye first and is closed after it. None of them registered, so the lounge never saw a join or a departure for them and no other guest heard anything. refused counts three: the ban, the unreadable credential and the ops frame sent too early. The silent socket is counted as timed out instead.

Settle a duplicate login#

When a name that is already inside logs in again, someone has to lose. Plaza leaves the choice to you. To refuse the newcomer, return a Refused with a code of your own. This door does the other thing and closes the older connection.

To close a connection the door needs the session's ConnectionManager. The door has to exist before the session it admits into, so it takes the manager afterwards through a OnceLock. Replace src/door.rs:

src/door.rs
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};

use async_trait::async_trait;
use parking_lot::Mutex;
use plaza::agent::Agent;
use plaza_session::{ConnectionAdmission, ConnectionAdmitter, ConnectionManager, Farewell, Peer};

pub const UNREADABLE: u16 = 4400;
pub const BANNED: u16 = 4403;
pub const SIGNED_IN_ELSEWHERE: u16 = 4409;

#[derive(Default)]
pub struct Door {
    names: Mutex<HashMap<u64, String>>,
    banned: Mutex<Vec<String>>,
    next_key: AtomicU64,
    manager: OnceLock<Arc<ConnectionManager<u64>>>,
}

impl Door {
    pub fn attach(&self, manager: Arc<ConnectionManager<u64>>) {
        let _ = self.manager.set(manager);
    }

    pub fn ban(&self, name: &str) {
        self.banned.lock().push(name.to_owned());
    }

    pub fn name_of(&self, key: u64) -> String {
        self.names.lock().get(&key).cloned().unwrap_or_default()
    }

    pub fn keys_of(&self, name: &str) -> Vec<u64> {
        self.names
            .lock()
            .iter()
            .filter(|(_, held)| *held == name)
            .map(|(key, _)| *key)
            .collect()
    }

    pub fn left(&self, key: u64) {
        self.names.lock().remove(&key);
    }
}

#[async_trait]
impl ConnectionAdmitter<u64> for Door {
    async fn admit(&self, credential: &[u8], _peer: &Peer) -> ConnectionAdmission<u64> {
        let Ok(name) = std::str::from_utf8(credential) else {
            return refuse(UNREADABLE, "unreadable credential");
        };
        if self.banned.lock().iter().any(|banned| banned == name) {
            return refuse(BANNED, "banned");
        }
        if let Some(manager) = self.manager.get() {
            for key in self.keys_of(name) {
                let farewell = Farewell::new(SIGNED_IN_ELSEWHERE)
                    .with_detail("signed in from somewhere else".as_bytes());
                manager.deregister_agent(&key, farewell);
            }
        }
        let key = self.next_key.fetch_add(1, Ordering::Relaxed) + 1;
        self.names.lock().insert(key, name.to_owned());
        ConnectionAdmission::Admitted(Agent::new_human(key))
    }
}

fn refuse(code: u16, why: &str) -> ConnectionAdmission<u64> {
    ConnectionAdmission::Refused(Farewell::new(code).with_detail(why.as_bytes()))
}

deregister_agent closes every connection the key holds and sends each one the farewell first. The older login can be closed from inside admit because it is registered and the newcomer is not yet.

In src/main.rs, call attach once the session exists. Alice logs in from her laptop and then from her phone:

src/main.rs
mod client;
mod door;
mod lounge;

use std::sync::Arc;

use plaza::controller::StateControllerBuilder;
use plaza_session::SessionOptions;
use plaza_session::codec::JsonCodec;
use plaza_session::tcp::TcpPlazaSession;

use crate::client::Client;
use crate::door::Door;
use crate::lounge::{Lounge, LoungeOp, LoungeState};

const ADDR: &str = "127.0.0.1:5802";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let door = Arc::new(Door::default());
    door.ban("mallory");
    let session = TcpPlazaSession::<LoungeOp, u64>::bind_with_admitter(
        ADDR,
        door.clone(),
        JsonCodec,
        SessionOptions::default(),
    )
    .await?;
    door.attach(session.manager().clone());

    let lounge = Arc::new(Lounge { door: door.clone() });
    let (_commands, controller) =
        StateControllerBuilder::without_snapshots(lounge, session.clone(), LoungeState::default())
            .build();
    tokio::spawn(controller.run());

    let mut laptop = Client::connect(ADDR, "laptop").await?;
    laptop.present(b"alice").await?;
    laptop.hear().await;

    let mut phone = Client::connect(ADDR, "phone").await?;
    phone.present(b"alice").await?;
    laptop.hear().await;
    laptop.hear().await;
    phone.hear().await;

    phone
        .say(LoungeOp::Say("switched to my phone".into()))
        .await?;
    phone.hear().await;
    Ok(())
}

Run it:

$ cargo run
laptop: [Welcome { name: "alice" }]
laptop: goodbye 4409 "signed in from somewhere else"
laptop: socket closed
phone: [Welcome { name: "alice" }]
phone: [Said { who: "alice", text: "switched to my phone" }]

The laptop hears why it was closed before its socket goes. The phone is admitted as a new connection with its own key and carries on as alice.

Kick a guest with a reason#

A kick is the same close as the duplicate login, ordered by the lounge instead of the door. The lounge needs the manager for that, so it gets a manager field.

The harder part is what happens afterwards. Every departure reaches the rules as the same AgentLeft, whether the host kicked the guest or the guest's network went away. Only the code that ordered the close knows why. ClosureLog is where that code writes it down. order records a reason and returns true the first time for a key, so a guest is sent one goodbye even when two rules decide to close it. departed reads the reason back on AgentLeft and answers Netdrop when nobody ordered anything.

Replace src/lounge.rs:

src/lounge.rs
use std::sync::Arc;

use async_trait::async_trait;
use plaza::common::closure::{ClosureLog, Departed};
use plaza::session::TargetedOp;
use plaza::state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError};
use plaza_session::{ConnectionManager, Farewell};
use serde::{Deserialize, Serialize};

use crate::door::Door;

pub const KICKED: u16 = 4001;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum LoungeOp {
    Say(String),
    Kick { who: String, reason: String },
    Welcome { name: String },
    Said { who: String, text: String },
    Gone { who: String, why: String },
}

#[derive(Debug, Default)]
pub struct LoungeState {
    pub guests: Vec<u64>,
    pub partings: ClosureLog<u64, String>,
}

pub struct Lounge {
    pub door: Arc<Door>,
    pub manager: Arc<ConnectionManager<u64>>,
}

#[async_trait]
impl StateLogic<LoungeOp, u64, LoungeState> for Lounge {
    async fn process_input(
        &self,
        state: &mut LoungeState,
        input: LogicInput<LoungeOp, u64>,
    ) -> Result<LogicOutput<LoungeOp, u64>, StateLogicError> {
        match input {
            LogicInput::AgentJoined { agent } => {
                let Some(key) = agent.id_cloned() else {
                    return Ok(LogicOutput::none());
                };
                state.guests.push(key);
                let name = self.door.name_of(key);
                let welcome = LoungeOp::Welcome { name };
                Ok(vec![TargetedOp::new_system_to(key, vec![welcome])].into())
            }
            LogicInput::AgentOps { source, ops } => {
                let Some(key) = source.id_cloned() else {
                    return Ok(LogicOutput::none());
                };
                let who = self.door.name_of(key);
                let mut out = Vec::new();
                for op in ops {
                    match op {
                        LoungeOp::Say(text) => {
                            let said = LoungeOp::Said {
                                who: who.clone(),
                                text,
                            };
                            out.push(TargetedOp::new_system_all(vec![said]));
                        }
                        LoungeOp::Kick { who: name, reason } => {
                            let why = format!("kicked: {reason}");
                            for target in self.door.keys_of(&name) {
                                if state.partings.order(target, why.clone()) {
                                    let farewell =
                                        Farewell::new(KICKED).with_detail(why.as_bytes());
                                    self.manager.deregister_agent(&target, farewell);
                                }
                            }
                        }
                        _ => {}
                    }
                }
                Ok(out.into())
            }
            LogicInput::AgentLeft { agent_id } => {
                state.guests.retain(|key| *key != agent_id);
                let who = self.door.name_of(agent_id);
                self.door.left(agent_id);
                let why = match state.partings.departed(&agent_id) {
                    Departed::Ordered(why) => why,
                    Departed::Netdrop => "netdrop".to_owned(),
                };
                let gone = LoungeOp::Gone { who, why };
                Ok(vec![TargetedOp::new_system_all(vec![gone])].into())
            }
            _ => Ok(LogicOutput::none()),
        }
    }
}

deregister_agent is synchronous, so it can be called from inside the rules. The lounge then tells everyone who is left why the guest went.

In src/main.rs, alice kicks bob and carol's link drops:

src/main.rs
mod client;
mod door;
mod lounge;

use std::sync::Arc;

use plaza::controller::StateControllerBuilder;
use plaza_session::SessionOptions;
use plaza_session::codec::JsonCodec;
use plaza_session::tcp::TcpPlazaSession;

use crate::client::Client;
use crate::door::Door;
use crate::lounge::{Lounge, LoungeOp, LoungeState};

const ADDR: &str = "127.0.0.1:5803";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let door = Arc::new(Door::default());
    door.ban("mallory");
    let session = TcpPlazaSession::<LoungeOp, u64>::bind_with_admitter(
        ADDR,
        door.clone(),
        JsonCodec,
        SessionOptions::default(),
    )
    .await?;
    door.attach(session.manager().clone());

    let lounge = Arc::new(Lounge {
        door: door.clone(),
        manager: session.manager().clone(),
    });
    let (_commands, controller) =
        StateControllerBuilder::without_snapshots(lounge, session.clone(), LoungeState::default())
            .build();
    tokio::spawn(controller.run());

    let mut alice = Client::connect(ADDR, "alice").await?;
    alice.present(b"alice").await?;
    alice.hear().await;
    let mut bob = Client::connect(ADDR, "bob").await?;
    bob.present(b"bob").await?;
    bob.hear().await;
    let mut carol = Client::connect(ADDR, "carol").await?;
    carol.present(b"carol").await?;
    carol.hear().await;

    alice
        .say(LoungeOp::Kick {
            who: "bob".into(),
            reason: "spamming".into(),
        })
        .await?;
    bob.hear().await;
    bob.hear().await;
    alice.hear().await;
    carol.hear().await;

    drop(carol);
    alice.hear().await;
    Ok(())
}

drop(carol) closes carol's socket from the client side without a goodbye, which is what a pulled cable looks like to the server. Run it:

$ cargo run
alice: [Welcome { name: "alice" }]
bob: [Welcome { name: "bob" }]
carol: [Welcome { name: "carol" }]
bob: goodbye 4001 "kicked: spamming"
bob: socket closed
alice: [Gone { who: "bob", why: "kicked: spamming" }]
carol: [Gone { who: "bob", why: "kicked: spamming" }]
alice: [Gone { who: "carol", why: "netdrop" }]

Bob reads the reason before his socket closes. The code 4001 tells his client it was removed rather than dropped. The lounge reports bob as kicked and carol as a netdrop, from the same AgentLeft event.

Refuse an op before the rules see it#

Right now any guest can kick any other. Whether an agent may submit an op at all is an authorization question. An OpGuard answers it in one place. The controller runs the guard on every op a client sends, before process_input, with the state read-only. A refused op never reaches the rules.

The guard needs to know who the host is. The first guest to arrive becomes the host. In src/lounge.rs, add use plaza::agent::Agent; and use plaza::op_guard::OpClearance; to the imports. Add a Refused op and a host field, then write the guard as a plain function:

src/lounge.rs
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum LoungeOp {
    Say(String),
    Kick { who: String, reason: String },
    Welcome { name: String },
    Said { who: String, text: String },
    Gone { who: String, why: String },
    Refused(String),
}

#[derive(Debug, Default)]
pub struct LoungeState {
    pub guests: Vec<u64>,
    pub host: Option<u64>,
    pub partings: ClosureLog<u64, String>,
}

pub fn screen(state: &LoungeState, source: &Agent<u64>, op: &LoungeOp) -> OpClearance<LoungeOp> {
    match op {
        LoungeOp::Say(_) => OpClearance::Cleared,
        LoungeOp::Kick { .. } if source.id() == state.host.as_ref() => OpClearance::Cleared,
        LoungeOp::Kick { .. } => OpClearance::Refused {
            reply: Some(LoungeOp::Refused("only the host can kick".into())),
        },
        _ => OpClearance::Refused { reply: None },
    }
}

Then record the host when a guest joins, right after the guest is added to guests:

src/lounge.rs
state.host.get_or_insert(key);

Say is cleared for everyone and Kick only for the host. A refused kick gets a Refused reply so the client knows what happened. Every other op is one the server sends, such as Said or Welcome, so a client that sends one is trying to forge it. That op is dropped with no reply.

GuardFn turns the function into a guard and StateControllerBuilder::guard installs it. controller.stats() hands back the controller's counters, which include ops_refused. Replace src/main.rs:

src/main.rs
mod client;
mod door;
mod lounge;

use std::sync::Arc;

use plaza::controller::StateControllerBuilder;
use plaza::op_guard::GuardFn;
use plaza_session::SessionOptions;
use plaza_session::codec::JsonCodec;
use plaza_session::tcp::TcpPlazaSession;

use crate::client::Client;
use crate::door::Door;
use crate::lounge::{Lounge, LoungeOp, LoungeState, screen};

const ADDR: &str = "127.0.0.1:5804";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let door = Arc::new(Door::default());
    door.ban("mallory");
    let session = TcpPlazaSession::<LoungeOp, u64>::bind_with_admitter(
        ADDR,
        door.clone(),
        JsonCodec,
        SessionOptions::default(),
    )
    .await?;
    door.attach(session.manager().clone());

    let lounge = Arc::new(Lounge {
        door: door.clone(),
        manager: session.manager().clone(),
    });
    let (_commands, controller) =
        StateControllerBuilder::without_snapshots(lounge, session.clone(), LoungeState::default())
            .guard(Arc::new(GuardFn(screen)))
            .build();
    let stats = controller.stats();
    tokio::spawn(controller.run());

    let mut alice = Client::connect(ADDR, "alice").await?;
    alice.present(b"alice").await?;
    alice.hear().await;
    let mut bob = Client::connect(ADDR, "bob").await?;
    bob.present(b"bob").await?;
    bob.hear().await;

    bob.say(LoungeOp::Kick {
        who: "alice".into(),
        reason: "mutiny".into(),
    })
    .await?;
    bob.hear().await;

    bob.say(LoungeOp::Said {
        who: "alice".into(),
        text: "I resign".into(),
    })
    .await?;
    bob.say(LoungeOp::Say("fine".into())).await?;
    alice.hear().await;
    bob.hear().await;

    println!("ops refused by the guard: {}", stats.ops_refused());
    Ok(())
}

Bob tries to kick alice, then tries to forge a message in alice's name, then says something real. Run it:

$ cargo run
alice: [Welcome { name: "alice" }]
bob: [Welcome { name: "bob" }]
bob: [Refused("only the host can kick")]
alice: [Said { who: "bob", text: "fine" }]
bob: [Said { who: "bob", text: "fine" }]
ops refused by the guard: 2

Bob's kick was answered with a refusal and alice heard nothing about it. The forged Said vanished and alice only heard what bob really said. Both refusals are counted. A client's ops are handled in the order it sent them, so by the time bob's real message arrives both refusals have happened.

Close idle and expired sessions#

The session offers two ways to close a connection on a timer and decides no timeout itself.

agent_idle_for reports how long a connection has been silent, counted from its last data frame. The session answers latency probes itself, so probe traffic never counts as activity. This is why an away-from-keyboard rule has to read the session's number: a rule written against the frames on the socket would see the probes and never fire. The session only reports the number. Your code checks it on its own schedule and decides what is too long.

set_deadline limits how long a connection may stay open and the connection's own task enforces it. Expiry goes through the same flush-then-goodbye close as a kick. Setting it again moves the deadline, which is how a renewal works. Setting it to None clears it.

In src/lounge.rs, add use std::time::Duration; to the imports. Add two codes next to KICKED and three settings to Lounge:

src/lounge.rs
pub const KICKED: u16 = 4001;
pub const TRIAL_OVER: u16 = 4002;
pub const AWAY: u16 = 4010;
src/lounge.rs
pub struct Lounge {
    pub door: Arc<Door>,
    pub manager: Arc<ConnectionManager<u64>>,
    pub away_after: Duration,
    pub trial_guests: Vec<String>,
    pub trial_length: Duration,
}

A guest on the trial list gets a deadline when it joins. The deadline needs a connection id rather than an agent id, so connections_of resolves the key. Add this to the AgentJoined arm, after the name lookup:

src/lounge.rs
if self.trial_guests.contains(&name) {
    let farewell =
        Farewell::new(TRIAL_OVER).with_detail("your trial is over".as_bytes());
    for conn in self.manager.connections_of(&key) {
        self.manager
            .set_deadline(conn, Some(self.trial_length), farewell.clone());
    }
    state.partings.order(key, "trial over".to_owned());
}

Arming the deadline is the lounge ordering a close, so it records the reason in the ClosureLog right away. That also stops the idle rule from sending the same guest a second goodbye.

The idle rule runs on every time step. Add a TimeStep arm before the AgentLeft arm and delete the _ => Ok(LogicOutput::none()) arm, since the four arms now cover every input:

src/lounge.rs
LogicInput::TimeStep { .. } => {
    for &key in &state.guests {
        let idle = self.manager.agent_idle_for(&key).unwrap_or_default();
        if idle >= self.away_after && state.partings.order(key, "away".to_owned()) {
            let farewell =
                Farewell::new(AWAY).with_detail("away from the lounge".as_bytes());
            self.manager.deregister_agent(&key, farewell);
        }
    }
    Ok(LogicOutput::none())
}

Time steps come from a TickDriver, which sends the controller one every 100 milliseconds here. Replace src/main.rs:

src/main.rs
mod client;
mod door;
mod lounge;

use std::sync::Arc;
use std::time::Duration;

use plaza::controller::StateControllerBuilder;
use plaza::op_guard::GuardFn;
use plaza::tick_driver::TickDriver;
use plaza_session::SessionOptions;
use plaza_session::codec::JsonCodec;
use plaza_session::tcp::TcpPlazaSession;

use crate::client::Client;
use crate::door::Door;
use crate::lounge::{Lounge, LoungeOp, LoungeState, screen};

const ADDR: &str = "127.0.0.1:5805";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let door = Arc::new(Door::default());
    door.ban("mallory");
    let session = TcpPlazaSession::<LoungeOp, u64>::bind_with_admitter(
        ADDR,
        door.clone(),
        JsonCodec,
        SessionOptions::default(),
    )
    .await?;
    door.attach(session.manager().clone());

    let lounge = Arc::new(Lounge {
        door: door.clone(),
        manager: session.manager().clone(),
        away_after: Duration::from_millis(600),
        trial_guests: vec!["frank".to_owned()],
        trial_length: Duration::from_millis(300),
    });
    let (commands, controller) =
        StateControllerBuilder::without_snapshots(lounge, session.clone(), LoungeState::default())
            .guard(Arc::new(GuardFn(screen)))
            .build();
    tokio::spawn(controller.run());
    tokio::spawn(TickDriver::new(Duration::from_millis(100)).run(commands));

    let mut erin = Client::connect(ADDR, "erin").await?;
    erin.present(b"erin").await?;
    erin.hear().await;
    let mut frank = Client::connect(ADDR, "frank").await?;
    frank.present(b"frank").await?;
    frank.hear().await;

    erin.say(LoungeOp::Say("back in a minute".into())).await?;
    erin.hear().await;
    frank.hear().await;

    frank.hear().await;
    frank.hear().await;
    erin.hear().await;
    erin.hear().await;
    erin.hear().await;
    Ok(())
}

Erin says one thing and then goes quiet. Frank is on a 300 millisecond trial and the lounge removes anyone silent for 600 milliseconds. Run it:

$ cargo run
erin: [Welcome { name: "erin" }]
frank: [Welcome { name: "frank" }]
erin: [Said { who: "erin", text: "back in a minute" }]
frank: [Said { who: "erin", text: "back in a minute" }]
frank: goodbye 4002 "your trial is over"
frank: socket closed
erin: [Gone { who: "frank", why: "trial over" }]
erin: goodbye 4010 "away from the lounge"
erin: socket closed

Frank's deadline fires first and erin is told why he left. Erin is removed for being away, even though her client answered every latency probe the session sent her the whole time. The exact moments both closes happen depend on the 100 millisecond tick and the scheduler, but the trial always ends before the idle rule fires, so the order of the lines stays the same.

Where to go next#

This tutorial left out two of the session's tools for saying no. rate_limit_inbound sets a per-connection Rate that the session enforces before a frame reaches the queue every connection shares, so a flood costs only the flooder. disconnect_all sends every connection the same farewell and closes them all, which is a graceful shutdown in one call.

Guide chapter 40, The right to say no explains where each mechanism sits and why plaza ships no policy defaults. Its two labs are worth reading next to this tutorial: door_policy adds a per-address cap, a capacity rule and a credit that buys time through set_deadline. table_manners adds flood shedding, a drain and seats that survive a netdrop but not a kick.

The plaza_session usage guide on GitHub covers admission, close codes, deadlines and the per-connection readers. The plaza usage guide covers OpGuard and the controller.

The full API references are on docs.rs: plaza_session, plaza and plaza_wire.