# Test against a bad network

**For:** Rust developers serving a plaza app over WebSockets who want to see how it behaves on a slow or lossy link before a player on hotel wifi finds out.

You will serve a shared counter over WebSockets on localhost and then make the link to one client slow, lossy and jittery while the program runs, using the link conditioner built into `plaza_session`. The client times how long each change takes to come back and the server times its own latency probes, so you can see what each setting changes and what it leaves alone.

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 badlink
cd badlink
```

`plaza` holds the controller, `plaza_session` serves it over actix-web WebSockets and `plaza_ws` is the client. `plaza_client_utils` has the `RttEstimator` the client measures with. Your `Cargo.toml` should end like this:

```toml title="Cargo.toml"
[dependencies]
actix-web = "4"
async-trait = "0.1"
plaza = "0"
plaza_client_utils = "0"
plaza_session = "0"
plaza_wire = "0"
plaza_ws = { version = "0", features = ["native", "pump"] }
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }
```

The server and its client share one process. The server runs on its own thread inside an actix system and listens on `127.0.0.1:5780`. The client runs on the main thread and polls its socket every millisecond, the way a game polls once per frame. Nothing leaves your machine: the conditioner sits inside the session, on the server's side of each connection.

## Write the counter

This is the same counter as in the previous tutorial. Put it in `src/counter.rs`:

```rust title="src/counter.rs"
use async_trait::async_trait;
use plaza::agent::Agent;
use plaza::error::{SnapshotError, StateLogicError};
use plaza::snapshot::{SnapshotContext, SnapshotProvider};
use plaza::state_logic::{LogicInput, LogicOutput, SnapshotRequest, StateLogic};
use serde::{Deserialize, Serialize};

pub type UserId = u64;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum CounterOp {
    Increment(i64),
    Value(i64),
}

#[derive(Debug, Default)]
pub struct Counter {
    value: i64,
    here: Vec<Agent<UserId>>,
}

pub struct CounterLogic;

#[async_trait]
impl StateLogic<CounterOp, UserId, Counter> for CounterLogic {
    async fn process_input(
        &self,
        state: &mut Counter,
        input: LogicInput<CounterOp, UserId>,
    ) -> Result<LogicOutput<CounterOp, UserId>, StateLogicError> {
        match input {
            LogicInput::AgentJoined { agent } => state.here.push(agent),
            LogicInput::AgentLeft { agent_id } => state.here.retain(|a| a.id() != Some(&agent_id)),
            LogicInput::AgentOps { ops, .. } => {
                for op in ops {
                    if let CounterOp::Increment(by) = op {
                        state.value += by;
                    }
                }
                return Ok(
                    LogicOutput::none().and_snapshot(SnapshotRequest::to(state.here.clone()))
                );
            }
            LogicInput::TimeStep { .. } => {}
        }
        Ok(LogicOutput::none())
    }
}

pub struct CounterSnapshotter;

#[async_trait]
impl SnapshotProvider<UserId, Counter, CounterOp> for CounterSnapshotter {
    async fn create_snapshot(
        &self,
        state: &Counter,
        _target: Option<&Agent<UserId>>,
        _context: Option<SnapshotContext>,
    ) -> Result<Option<CounterOp>, SnapshotError<UserId>> {
        Ok(Some(CounterOp::Value(state.value)))
    }
}
```

Every batch of ops asks for a fresh snapshot for everyone connected, so a client that sends `Increment(1)` gets a `Value` back. The time between the two is what a player feels as the delay between doing something and seeing it.

## Measure a clean link

Replace `src/main.rs` with this:

```rust title="src/main.rs"
mod counter;

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};

use actix_web::{App, HttpRequest, HttpResponse, HttpServer, web};
use plaza::agent::Agent;
use plaza::controller::StateControllerBuilder;
use plaza_client_utils::RttEstimator;
use plaza_session::{ActixWsPlazaSession, SessionOptions};
use plaza_wire::{JsonCodec, WireCodec};
use plaza_ws::pump::{Arrival, FramePump};

use counter::{Counter, CounterLogic, CounterOp, CounterSnapshotter, UserId};

const ADDR: &str = "127.0.0.1:5780";
const URL: &str = "ws://127.0.0.1:5780/ws";

type CounterSession = ActixWsPlazaSession<CounterOp, UserId>;

static NEXT_USER: AtomicU64 = AtomicU64::new(1);

async fn ws_route(
    req: HttpRequest,
    stream: web::Payload,
    session: web::Data<Arc<CounterSession>>,
) -> Result<HttpResponse, actix_web::Error> {
    let id = NEXT_USER.fetch_add(1, Ordering::Relaxed);
    session.handle_connection(&req, stream, Agent::new_human(id))
}

fn serve(ready: mpsc::Sender<Arc<CounterSession>>) {
    actix_web::rt::System::new().block_on(async move {
        let options = SessionOptions::default().probe_schedule(
            8,
            Duration::from_millis(125),
            Duration::from_millis(250),
        );
        let session = CounterSession::with_options(JsonCodec, options);
        let (_commands, controller) = StateControllerBuilder::new(
            Arc::new(CounterLogic),
            session.clone(),
            Arc::new(CounterSnapshotter),
            Counter::default(),
        )
        .build();
        tokio::spawn(controller.run());

        let data = web::Data::new(session.clone());
        let server = HttpServer::new(move || {
            App::new()
                .app_data(data.clone())
                .route("/ws", web::get().to(ws_route))
        })
        .workers(1)
        .bind(ADDR)
        .expect("bind");
        ready.send(session).expect("main is waiting");
        server.run().await.expect("serve");
    });
}

struct Client {
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
}

impl Client {
    fn connect() -> Self {
        let pump = FramePump::connect(URL, JsonCodec, 0).expect("connect");
        Client {
            pump,
            start: Instant::now(),
            values: VecDeque::new(),
        }
    }

    fn send(&mut self, op: CounterOp) {
        self.pump.send_op(&op);
    }

    fn poll(&mut self) {
        let mut arrivals = Vec::new();
        self.pump
            .poll(self.start.elapsed().as_millis() as u64, &mut arrivals);
        for arrival in arrivals {
            match arrival {
                Arrival::Ops(frame) => {
                    let ops: Vec<CounterOp> = JsonCodec.decode(frame.body()).expect("decode");
                    for op in ops {
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                        }
                    }
                }
                Arrival::Closed(closed) => panic!("closed: {closed}"),
                Arrival::Opened | Arrival::Mismatch { .. } => {}
            }
        }
    }

    fn idle(&mut self, time: Duration) {
        let until = Instant::now() + time;
        while Instant::now() < until {
            self.poll();
            thread::sleep(Duration::from_millis(1));
        }
    }

    fn next_value(&mut self) -> i64 {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(value) = self.values.pop_front() {
                return value;
            }
            thread::sleep(Duration::from_millis(1));
        }
        panic!("no value within 5s");
    }
}

fn floor_to(step: u64, ms: f32) -> u64 {
    ms as u64 / step * step
}

fn measure(label: &str, client: &mut Client, session: &CounterSession) {
    client.idle(Duration::from_millis(500));

    let mut ops = RttEstimator::default();
    let mut slowest = 0;
    for _ in 0..20 {
        let sent = Instant::now();
        client.send(CounterOp::Increment(1));
        client.next_value();
        let took = sent.elapsed().as_millis() as u64;
        ops.observe(took);
        slowest = slowest.max(took);
    }

    let manager = session.manager();
    let conn = manager.connections_of(&1)[0];
    let probe = manager.min_link_rtt(conn).expect("the session has probed");
    println!(
        "{label}: op round trip {} ms, probe round trip {} ms, slowest op {} ms",
        floor_to(10, ops.min_rtt().expect("twenty samples")),
        floor_to(10, probe.as_secs_f32() * 1000.0),
        floor_to(100, slowest as f32),
    );
}

fn main() {
    let (ready, started) = mpsc::channel();
    thread::spawn(move || serve(ready));
    let session = started.recv().expect("the server started");

    let mut client = Client::connect();
    client.next_value();

    measure("clean", &mut client, &session);
}
```

The session measures every connection itself. It sends a `Ping` frame down the connection, the client's `FramePump` answers with a `Pong` and the server times the round trip. By default it sends eight probes at 125 ms intervals and then one every five seconds. Setting a link profile clears the readings for that connection, so `probe_schedule` here drops the idle interval to 250 ms to get new readings soon after each change.

`measure` takes two readings. On the client it sends twenty increments one at a time and feeds how long each took into a fresh `RttEstimator`. On the server it reads `min_link_rtt` for the client's connection, the smallest probe round trip since the readings were last cleared. Both report the minimum, because jitter only ever adds delay and the smallest sample is the best estimate of the link itself. The program rounds the minimums down to tens of milliseconds and the slowest op down to hundreds, since the last few milliseconds depend on scheduling.

`FramePump` keeps an `RttEstimator` of its own for the probes it sends, which you can read as `pump.timeline().rtt` or `pump.rtt_ms()`. The client cannot tell when the server changes its link, so that estimator keeps its old minimum across a change. That is why `measure` starts a new one each time.

Run it:

```shell-session
$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
```

On localhost both round trips take a millisecond or two.

## Add delay while the program runs

A `LinkProfile` has an `up` direction for what the client sends and a `down` direction for what the server sends. Each is a `DirectionProfile` with a delay, a jitter, a loss probability and what a loss means. `LinkProfile::symmetric` uses the same profile both ways. Change the `plaza_session` import to bring in the two types:

```rust title="src/main.rs"
use plaza_session::{ActixWsPlazaSession, DirectionProfile, LinkProfile, SessionOptions};
```

Then add this to the end of `main`:

```rust title="src/main.rs"
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile::delayed(Duration::from_millis(100))),
    );
    measure("delay", &mut client, &session);
```

The client is agent 1, the first connection the route numbered. `set_agent_link_profile` applies to every connection that agent holds and takes effect at once on a live connection. `set_all_link_profiles` does the same for every connection the session has, which is what a room-wide latency slider wants.

Run it:

```shell-session
$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms
```

A hundred milliseconds each way is a 200 ms round trip, which is what both measurements show. The op takes the same time as the probe because both cross the same conditioner: the increment waits 100 ms on the way up and the snapshot waits 100 ms on the way down. Everything on the connection goes through it, your ops, snapshots and the probes alike. Without rounding, the minimums land a few milliseconds above 200.

## Add loss

Add this to the end of `main`:

```rust title="src/main.rs"
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile {
            delay: Duration::from_millis(100),
            loss: 0.05,
            ..DirectionProfile::default()
        }),
    );
    measure("loss", &mut client, &session);
    println!(
        "server: the link dropped {} frames",
        session.agent_link_dropped(&1)
    );
```

Each frame now has a 5% chance of being lost in each direction. What a loss means is the profile's `delivery`, which defaults to `Delivery::Reliable`.

Run it:

```shell-session
$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms
loss: op round trip 200 ms, probe round trip 200 ms, slowest op 400 ms
server: the link dropped 0 frames
```

TCP and WebSockets are reliable streams, so a lost segment is retransmitted and never reaches the application as a missing message. `Delivery::Reliable` models that. A lost frame arrives `RETRANSMIT_PENALTY` (200 ms, TCP's minimum retransmission timeout) late and everything behind it on the connection waits, so the link dropped nothing. The fastest op still took 200 ms. The slowest took about 400, the delay plus one retransmission. Nothing went missing: `measure` waits for each increment's snapshot and all twenty came back.

The jitter and loss draws are seeded from the connection id instead of the clock, so the same program on the same machine loses the same frames from run to run. That is why the slowest op here came out the same in every run. With different traffic the loss lands on different frames.

## Add jitter

Jitter is the part of the delay that changes from frame to frame. A delay round trip is a single number, but under jitter every op takes a different time. The program prints a property that holds whatever the draws are instead of a number. Add this function above `main`:

```rust title="src/main.rs"
fn jittery(client: &mut Client) {
    client.idle(Duration::from_millis(500));

    let mut inside = 0;
    for _ in 0..20 {
        let sent = Instant::now();
        client.send(CounterOp::Increment(1));
        client.next_value();
        let took = sent.elapsed();
        if took >= Duration::from_millis(200) && took < Duration::from_millis(300) {
            inside += 1;
        }
    }
    println!("jitter: {inside} of 20 op round trips took 200 to 300 ms");
}
```

Then add this to the end of `main`:

```rust title="src/main.rs"
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile {
            delay: Duration::from_millis(100),
            jitter: Duration::from_millis(40),
            ..DirectionProfile::default()
        }),
    );
    jittery(&mut client);
```

Run it:

```shell-session
$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms
loss: op round trip 200 ms, probe round trip 200 ms, slowest op 400 ms
server: the link dropped 0 frames
jitter: 20 of 20 op round trips took 200 to 300 ms
```

Each frame gets the delay plus an extra draw between zero and the jitter, so each direction takes 100 to 140 ms and an op round trip falls between 200 and 280 ms plus a few for scheduling. The conditioner keeps frames in order, as a stream does. A frame that drew a short delay still waits behind one that drew a long one, so a jitter spike arrives as a stall and then a burst. The individual round trips differ from run to run, but every one stays inside that range.

## Drop frames for real

`Delivery::Datagram` makes a lost frame disappear, which is what a UDP-style transport does. Over a WebSocket it simulates a datagram link, which is useful for testing how your client recovers. The client now needs to remember the last value it saw, since some increments and some snapshots will never arrive. Add a field to `Client`:

```rust title="src/main.rs"
struct Client {
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
    last: i64,
}
```

Set it to `last: 0,` in `Client::connect` and record each value in `Client::poll`:

```rust title="src/main.rs"
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                            self.last = value;
                        }
```

Add `Delivery` to the `plaza_session` import:

```rust title="src/main.rs"
use plaza_session::{ActixWsPlazaSession, Delivery, DirectionProfile, LinkProfile, SessionOptions};
```

Then add this to the end of `main`:

```rust title="src/main.rs"
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile {
            loss: 0.2,
            delivery: Delivery::Datagram,
            ..DirectionProfile::default()
        }),
    );
    client.idle(Duration::from_millis(500));
    let before = client.last;
    for _ in 0..50 {
        client.send(CounterOp::Increment(1));
    }
    client.idle(Duration::from_secs(1));
    println!(
        "datagram: sent 50 increments, the client saw all 50: {}",
        client.last - before == 50
    );
    println!(
        "server: the link dropped frames: {}",
        session.agent_link_dropped(&1) > 0
    );

    session.set_agent_link_profile(&1, LinkProfile::default());
    measure("passthrough", &mut client, &session);
```

The client fires fifty increments without waiting for each answer, because under this profile an answer may never come. The last three lines set the link back to passthrough with `LinkProfile::default()` and measure it again.

Run it:

```shell-session
$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms
loss: op round trip 200 ms, probe round trip 200 ms, slowest op 400 ms
server: the link dropped 0 frames
jitter: 20 of 20 op round trips took 200 to 300 ms
datagram: sent 50 increments, the client saw all 50: false
server: the link dropped frames: true
passthrough: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
```

With 20% loss each way, some increments never reach the controller and some snapshots never reach the client. No frame kind is exempt: probes and the `Hello` can be lost too. `agent_link_dropped` counts what the link threw away. Your application cannot count that itself, because a frame the link lost never reached your code. How many go missing depends on how the frames interleave, so the program prints only that some did.

A passthrough profile costs nothing: no queue, no allocation and no deadline arithmetic. After it the connection measures as clean as it did at the start.

## Where to go next

Profiles are plain data, so a debug panel only has to send new ones. `LinkPublisher` wraps a closure such as `move |profile| session.set_all_link_profiles(profile)` and skips a setting that did not change, which suits a slider that fires on every frame. For tests with no session in the middle, `plaza_client_utils` has `LatencyLink` behind its `net-sim` feature, a deterministic latency, jitter and loss queue that keeps order by default, as the conditioner does.

Guide chapter [31, Faking a bad network](https://github.com/excsn/plaza/blob/main/docs/guide/31-faking-a-bad-network.md) explains the conditioner and the two measurement planes. Its lab points at the playground examples, whose sliders drive the same profiles from a browser.

The [plaza_session usage guide](https://github.com/excsn/plaza/blob/main/session/README.USAGE.md) covers impairing a link, the probe schedule and the counters a session keeps. The [plaza_ws usage guide](https://github.com/excsn/plaza/blob/main/ws_client/README.USAGE.md) covers reading the link from the pump.

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