# 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 as `plaza = "0"`.

## Set up the project

Start a new binary crate.

```sh
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:

```toml title="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`:

```rust title="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`:

```rust title="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`:

```rust title="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:

```rust title="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:

```shell-session
$ 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`:

```rust title="src/door.rs" {10,15,20-22,39-41}
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`:

```rust title="src/main.rs" {6,17,22,24-25,30,44-47,49-52,54-57,59-61,64-69}
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:

```shell-session
$ 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`:

```rust title="src/door.rs" {3,8,12,19,23-25,33,35-41,58-64}
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:

```rust title="src/main.rs" {16,26,29,37-39,41-45,47-50}
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:

```shell-session
$ 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`:

```rust title="src/lounge.rs" {4,7,12,17,20,26,31,58-76,83,85-90}
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:

```rust title="src/main.rs" {16,31-34,40-48,50-59,61-62}
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:

```shell-session
$ 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:

```rust title="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`:

```rust title="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`:

```rust title="src/main.rs" {8,15,17,38,40,50-54,57-63,66}
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:

```shell-session
$ 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`:

```rust title="src/lounge.rs"
pub const KICKED: u16 = 4001;
pub const TRIAL_OVER: u16 = 4002;
pub const AWAY: u16 = 4010;
```

```rust title="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:

```rust title="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:

```rust title="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`:

```rust title="src/main.rs" {6,10,19,37-39,41,46,48-53,55-57,59-63}
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:

```shell-session
$ 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](https://github.com/excsn/plaza/blob/main/docs/guide/40-the-right-to-say-no.md) explains where each mechanism sits and why plaza ships no policy defaults. Its two labs are worth reading next to this tutorial: [door_policy](https://github.com/excsn/plaza/tree/main/examples/door_policy) adds a per-address cap, a capacity rule and a credit that buys time through `set_deadline`. [table_manners](https://github.com/excsn/plaza/tree/main/examples/table_manners) adds flood shedding, a drain and seats that survive a netdrop but not a kick.

The [plaza_session usage guide](https://github.com/excsn/plaza/blob/main/session/README.USAGE.md) on GitHub covers admission, close codes, deadlines and the per-connection readers. The [plaza usage guide](https://github.com/excsn/plaza/blob/main/core/README.USAGE.md) covers `OpGuard` and the controller.

The full API references are on docs.rs: [plaza_session](https://docs.rs/plaza_session), [plaza](https://docs.rs/plaza) and [plaza_wire](https://docs.rs/plaza_wire).