Predict the entity you control

For Rust developers building a server-authoritative game who want their own character to respond the moment a key is pressed, even though the server is a round trip away

You will build a headless client and server in one program, joined by a simulated link with 60ms of latency in each direction. First the client draws only what the server has confirmed, so every press shows up 200ms late. Then PredictedPlayer moves the body the moment the input happens and reconciles each server update. Finally a server-side shove produces a correction that the client eases out over a few frames instead of snapping.

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 predict
cd predict

Add plaza for the server half and plaza_client_utils for the client half. The net-sim feature turns on the net_sim module, which holds the simulated link. Your [dependencies] section should look like this:

Cargo.toml
[dependencies]
plaza = "0"
plaza_client_utils = { version = "0", features = ["net-sim"] }

The program uses no async runtime and no socket. Both ends run in one loop on simulated time at 10ms per frame. The link delivers packets by that clock, so every run prints the same output.

The client and the server have to move the body with the same rule. Here that rule is apply_move: each input moves the body half a unit along x. Its signature, fn(&mut State, &Input, &Ctx), is the one PredictedPlayer takes later. The server calls the very same function. Ctx is () because this rule needs nothing from the world.

The server consumes one input per simulation step. It queues the inputs that arrive and applies one each step. ClientInputTracker from plaza's game_common::reconciliation module remembers the last input sequence it processed for each client. Every step the server sends back an AuthoritativeStateUpdate, which carries the body's state and that sequence number. The client sends each input wrapped in a SequencedClientInput. Both payload types come from the same module.

Both ends step in the same 50ms quantum. FixedTimestep turns the 10ms frames into whole 50ms steps and yields each step as a Duration. A real plaza server steps with TickDriver::run_fixed; here a FixedTimestep drives the server too, so the whole program runs on one simulated clock.

Each direction gets its own LatencyLink. send takes the current time, the packet, the latency, the jitter, the loss percentage and a seeded Rng. drain_due hands back every packet whose delivery time has come. This program uses 60ms each way with no jitter and no loss.

In this first version the client does no prediction. It draws whatever the server last confirmed.

Replace src/main.rs with this:

src/main.rs
use std::collections::VecDeque;

use plaza::game_common::reconciliation::{
    AuthoritativeStateUpdate, ClientInputTracker, SequencedClientInput,
};
use plaza_client_utils::FixedTimestep;
use plaza_client_utils::net_sim::{LatencyLink, Rng};

const PLAYER: u32 = 1;
const STEP_MS: u64 = 50;
const FRAME_MS: u64 = 10;
const ONE_WAY_MS: u64 = 60;
const RUN_MS: u64 = 400;

#[derive(Debug, Clone, Copy, Default, PartialEq)]
struct Body {
    x: f32,
}

#[derive(Debug, Clone, Copy)]
struct Move {
    dx: f32,
}

fn apply_move(body: &mut Body, mv: &Move, _ctx: &()) {
    body.x += mv.dx * 0.5;
}

struct Server {
    body: Body,
    queued: VecDeque<SequencedClientInput<Move>>,
    tracker: ClientInputTracker<u32>,
    ticker: FixedTimestep,
    now_ms: u64,
}

impl Server {
    fn new() -> Self {
        Self {
            body: Body::default(),
            queued: VecDeque::new(),
            tracker: ClientInputTracker::new(),
            ticker: FixedTimestep::from_step_ms(STEP_MS),
            now_ms: 0,
        }
    }

    fn receive(&mut self, input: SequencedClientInput<Move>) {
        self.queued.push_back(input);
    }

    fn frame(&mut self, elapsed_ms: u64) -> Vec<AuthoritativeStateUpdate<Body, u64>> {
        let mut updates = Vec::new();
        for step in self.ticker.advance(elapsed_ms) {
            self.now_ms += step.as_millis() as u64;
            if let Some(input) = self.queued.pop_front() {
                apply_move(&mut self.body, &input.input_data, &());
                self.tracker
                    .record_processed_input(PLAYER, input.sequence_number);
            }
            updates.push(AuthoritativeStateUpdate {
                last_processed_input_seq: self
                    .tracker
                    .get_last_processed_input_seq(&PLAYER)
                    .unwrap_or(0),
                authoritative_player_state: self.body,
                server_time_at_state: self.now_ms,
            });
        }
        updates
    }
}

fn main() {
    let mut rng = Rng::new(7);
    let mut up: LatencyLink<SequencedClientInput<Move>> = LatencyLink::new();
    let mut down: LatencyLink<AuthoritativeStateUpdate<Body, u64>> = LatencyLink::new();
    let mut server = Server::new();

    let mut ticker = FixedTimestep::from_step_ms(STEP_MS);
    let mut next_seq = 0;
    let mut confirmed = Body::default();
    let mut acked = 0;

    let mut now_ms = 0;
    while now_ms < RUN_MS {
        now_ms += FRAME_MS;

        for input in up.drain_due(now_ms) {
            server.receive(input);
        }
        for update in server.frame(FRAME_MS) {
            down.send(now_ms, update, ONE_WAY_MS, 0, 0.0, &mut rng);
        }

        for update in down.drain_due(now_ms) {
            confirmed = update.authoritative_player_state;
            acked = update.last_processed_input_seq;
        }

        for _ in ticker.advance(FRAME_MS) {
            next_seq += 1;
            let input = SequencedClientInput {
                sequence_number: next_seq,
                input_data: Move { dx: 1.0 },
            };
            up.send(now_ms, input, ONE_WAY_MS, 0, 0.0, &mut rng);
            println!(
                "t={now_ms:>3}ms  pressed #{next_seq}  acked #{acked}  drawn x={:.2}",
                confirmed.x
            );
        }
    }
}

Run it:

$ cargo run
t= 50ms  pressed #1  acked #0  drawn x=0.00
t=100ms  pressed #2  acked #0  drawn x=0.00
t=150ms  pressed #3  acked #0  drawn x=0.00
t=200ms  pressed #4  acked #0  drawn x=0.00
t=250ms  pressed #5  acked #1  drawn x=0.50
t=300ms  pressed #6  acked #2  drawn x=1.00
t=350ms  pressed #7  acked #3  drawn x=1.50
t=400ms  pressed #8  acked #4  drawn x=2.00

Input #1 is pressed at 50ms. It reaches the server at 110ms, the server applies it on its 150ms step and the update arrives back at 210ms. The client draws the change on its next step at 250ms. For four steps the player is holding the key and nothing moves.

Predict your own movement#

PredictedPlayer is the bundle for a server that consumes one input per step. It applies each input locally as soon as it happens, keeps it in a buffer and hands back the sequence number to send. When an update arrives, reconcile resets the body to the server's state and replays every input the server had not acknowledged yet. If the prediction was right, the replay lands exactly where the body already was.

The plaza docs pair the two predictors with two server models. This server applies one input per step, so PredictedPlayer fits. A server that holds the latest input and integrates it every tick wants HeldInputPredictor instead. Replaying inputs against that kind of server double counts them.

PredictedPlayer::new takes the starting state, a PlayerConfig, the shared rule and a lerp function that blends two states. smoothing_secs: 0.0 turns smoothing off for now. Each frame calls advance with the frame time, render is what you draw and authoritative is the last state the server sent.

Replace src/main.rs with the predicting client. The server is unchanged:

src/main.rs
use std::collections::VecDeque;

use plaza::game_common::reconciliation::{
    AuthoritativeStateUpdate, ClientInputTracker, SequencedClientInput,
};
use plaza_client_utils::net_sim::{LatencyLink, Rng};
use plaza_client_utils::{FixedTimestep, PlayerConfig, PredictedPlayer};

const PLAYER: u32 = 1;
const STEP_MS: u64 = 50;
const FRAME_MS: u64 = 10;
const ONE_WAY_MS: u64 = 60;
const RUN_MS: u64 = 400;

#[derive(Debug, Clone, Copy, Default, PartialEq)]
struct Body {
    x: f32,
}

#[derive(Debug, Clone, Copy)]
struct Move {
    dx: f32,
}

fn apply_move(body: &mut Body, mv: &Move, _ctx: &()) {
    body.x += mv.dx * 0.5;
}

fn lerp_body(a: &Body, b: &Body, t: f32) -> Body {
    Body {
        x: a.x + (b.x - a.x) * t,
    }
}

struct Server {
    body: Body,
    queued: VecDeque<SequencedClientInput<Move>>,
    tracker: ClientInputTracker<u32>,
    ticker: FixedTimestep,
    now_ms: u64,
}

impl Server {
    fn new() -> Self {
        Self {
            body: Body::default(),
            queued: VecDeque::new(),
            tracker: ClientInputTracker::new(),
            ticker: FixedTimestep::from_step_ms(STEP_MS),
            now_ms: 0,
        }
    }

    fn receive(&mut self, input: SequencedClientInput<Move>) {
        self.queued.push_back(input);
    }

    fn frame(&mut self, elapsed_ms: u64) -> Vec<AuthoritativeStateUpdate<Body, u64>> {
        let mut updates = Vec::new();
        for step in self.ticker.advance(elapsed_ms) {
            self.now_ms += step.as_millis() as u64;
            if let Some(input) = self.queued.pop_front() {
                apply_move(&mut self.body, &input.input_data, &());
                self.tracker
                    .record_processed_input(PLAYER, input.sequence_number);
            }
            updates.push(AuthoritativeStateUpdate {
                last_processed_input_seq: self
                    .tracker
                    .get_last_processed_input_seq(&PLAYER)
                    .unwrap_or(0),
                authoritative_player_state: self.body,
                server_time_at_state: self.now_ms,
            });
        }
        updates
    }
}

fn main() {
    let mut rng = Rng::new(7);
    let mut up: LatencyLink<SequencedClientInput<Move>> = LatencyLink::new();
    let mut down: LatencyLink<AuthoritativeStateUpdate<Body, u64>> = LatencyLink::new();
    let mut server = Server::new();

    let mut ticker = FixedTimestep::from_step_ms(STEP_MS);
    let config = PlayerConfig {
        smoothing_secs: 0.0,
        ..PlayerConfig::default()
    };
    let mut me = PredictedPlayer::new(Body::default(), config, apply_move, lerp_body);

    let mut now_ms = 0;
    while now_ms < RUN_MS {
        now_ms += FRAME_MS;

        for input in up.drain_due(now_ms) {
            server.receive(input);
        }
        for update in server.frame(FRAME_MS) {
            down.send(now_ms, update, ONE_WAY_MS, 0, 0.0, &mut rng);
        }

        for update in down.drain_due(now_ms) {
            me.reconcile(
                update.authoritative_player_state,
                update.last_processed_input_seq,
            );
        }

        for _ in ticker.advance(FRAME_MS) {
            let mv = Move { dx: 1.0 };
            let seq = me.input(mv);
            let input = SequencedClientInput {
                sequence_number: seq,
                input_data: mv,
            };
            up.send(now_ms, input, ONE_WAY_MS, 0, 0.0, &mut rng);
            println!(
                "t={now_ms:>3}ms  pressed #{seq}  acked #{}  drawn x={:.2}  server x={:.2}  unacked {}",
                me.acked_seq(),
                me.render().x,
                me.authoritative().x,
                me.unacked_count(),
            );
        }
        me.advance(FRAME_MS as f32 / 1000.0);
    }
}

Run it:

$ cargo run
t= 50ms  pressed #1  acked #0  drawn x=0.50  server x=0.00  unacked 1
t=100ms  pressed #2  acked #0  drawn x=1.00  server x=0.00  unacked 2
t=150ms  pressed #3  acked #0  drawn x=1.50  server x=0.00  unacked 3
t=200ms  pressed #4  acked #0  drawn x=2.00  server x=0.00  unacked 4
t=250ms  pressed #5  acked #1  drawn x=2.50  server x=0.50  unacked 4
t=300ms  pressed #6  acked #2  drawn x=3.00  server x=1.00  unacked 4
t=350ms  pressed #7  acked #3  drawn x=3.50  server x=1.50  unacked 4
t=400ms  pressed #8  acked #4  drawn x=4.00  server x=2.00  unacked 4

The body moves on the same step the key is pressed. The server's copy trails four inputs behind, because a press takes 200ms to come back confirmed and the client presses once every 50ms. unacked_count holds steady at 4 for the same reason. Every update reconciles the body, but the server agrees with the prediction, so the replay puts the body back where it was and nothing visible happens.

Smooth a correction#

A prediction is wrong whenever the server knows something the client does not. In this step the server shoves the body back two units on its 300ms step, standing in for a collision with another player that the client could not see. The shove is server-only on purpose. It is not part of apply_move, because the client has no way to know it happened.

reconcile returns a Correction with the state that was seen before the update and the state it settled on after the replay. When the two differ, the client mispredicted.

Snapping the drawn body two units in one frame looks like a glitch. PlayerConfig::smoothing_secs sets the ErrorSmoother inside PredictedPlayer, which eases what you draw toward the corrected state over that many seconds while the logical state jumps at once. The client crate's docs say to keep the ease shorter than the interval between server updates, otherwise a new correction arrives before the last one has finished easing. Updates come every 50ms here, so the ease is 0.04 seconds.

Replace src/main.rs. The server gains the shove and the client prints each frame around the correction:

src/main.rs
use std::collections::VecDeque;

use plaza::game_common::reconciliation::{
    AuthoritativeStateUpdate, ClientInputTracker, SequencedClientInput,
};
use plaza_client_utils::net_sim::{LatencyLink, Rng};
use plaza_client_utils::{FixedTimestep, PlayerConfig, PredictedPlayer};

const PLAYER: u32 = 1;
const STEP_MS: u64 = 50;
const FRAME_MS: u64 = 10;
const ONE_WAY_MS: u64 = 60;
const RUN_MS: u64 = 420;
const SHOVE_AT_MS: u64 = 300;

#[derive(Debug, Clone, Copy, Default, PartialEq)]
struct Body {
    x: f32,
}

#[derive(Debug, Clone, Copy)]
struct Move {
    dx: f32,
}

fn apply_move(body: &mut Body, mv: &Move, _ctx: &()) {
    body.x += mv.dx * 0.5;
}

fn lerp_body(a: &Body, b: &Body, t: f32) -> Body {
    Body {
        x: a.x + (b.x - a.x) * t,
    }
}

struct Server {
    body: Body,
    queued: VecDeque<SequencedClientInput<Move>>,
    tracker: ClientInputTracker<u32>,
    ticker: FixedTimestep,
    now_ms: u64,
}

impl Server {
    fn new() -> Self {
        Self {
            body: Body::default(),
            queued: VecDeque::new(),
            tracker: ClientInputTracker::new(),
            ticker: FixedTimestep::from_step_ms(STEP_MS),
            now_ms: 0,
        }
    }

    fn receive(&mut self, input: SequencedClientInput<Move>) {
        self.queued.push_back(input);
    }

    fn frame(&mut self, elapsed_ms: u64) -> Vec<AuthoritativeStateUpdate<Body, u64>> {
        let mut updates = Vec::new();
        for step in self.ticker.advance(elapsed_ms) {
            self.now_ms += step.as_millis() as u64;
            if self.now_ms == SHOVE_AT_MS {
                self.body.x -= 2.0;
            }
            if let Some(input) = self.queued.pop_front() {
                apply_move(&mut self.body, &input.input_data, &());
                self.tracker
                    .record_processed_input(PLAYER, input.sequence_number);
            }
            updates.push(AuthoritativeStateUpdate {
                last_processed_input_seq: self
                    .tracker
                    .get_last_processed_input_seq(&PLAYER)
                    .unwrap_or(0),
                authoritative_player_state: self.body,
                server_time_at_state: self.now_ms,
            });
        }
        updates
    }
}

fn main() {
    let mut rng = Rng::new(7);
    let mut up: LatencyLink<SequencedClientInput<Move>> = LatencyLink::new();
    let mut down: LatencyLink<AuthoritativeStateUpdate<Body, u64>> = LatencyLink::new();
    let mut server = Server::new();

    let mut ticker = FixedTimestep::from_step_ms(STEP_MS);
    let config = PlayerConfig {
        smoothing_secs: 0.04,
        ..PlayerConfig::default()
    };
    let mut me = PredictedPlayer::new(Body::default(), config, apply_move, lerp_body);

    let mut now_ms = 0;
    while now_ms < RUN_MS {
        now_ms += FRAME_MS;

        for input in up.drain_due(now_ms) {
            server.receive(input);
        }
        for update in server.frame(FRAME_MS) {
            down.send(now_ms, update, ONE_WAY_MS, 0, 0.0, &mut rng);
        }

        for update in down.drain_due(now_ms) {
            let correction = me.reconcile(
                update.authoritative_player_state,
                update.last_processed_input_seq,
            );
            if correction.seen != correction.settled {
                println!(
                    "t={now_ms:>3}ms  correction after #{}: x={:.2} -> x={:.2}",
                    me.acked_seq(),
                    correction.seen.x,
                    correction.settled.x,
                );
            }
        }

        for _ in ticker.advance(FRAME_MS) {
            let mv = Move { dx: 1.0 };
            let seq = me.input(mv);
            let input = SequencedClientInput {
                sequence_number: seq,
                input_data: mv,
            };
            up.send(now_ms, input, ONE_WAY_MS, 0, 0.0, &mut rng);
        }
        me.advance(FRAME_MS as f32 / 1000.0);
        if (340..=420).contains(&now_ms) {
            println!(
                "t={now_ms:>3}ms  drawn x={:.2}  logical x={:.2}",
                me.render().x,
                me.logical().x
            );
        }
    }
}

Run it:

$ cargo run
t=340ms  drawn x=3.00  logical x=3.00
t=350ms  drawn x=3.50  logical x=3.50
t=360ms  correction after #4: x=3.50 -> x=1.50
t=360ms  drawn x=3.00  logical x=1.50
t=370ms  drawn x=2.50  logical x=1.50
t=380ms  drawn x=2.00  logical x=1.50
t=390ms  drawn x=1.50  logical x=1.50
t=400ms  drawn x=2.00  logical x=2.00
t=410ms  drawn x=2.00  logical x=2.00
t=420ms  drawn x=2.00  logical x=2.00

The update for input #4 left the server right after the shove. When it arrives at 360ms the logical body jumps from 3.50 to 1.50. The drawn body covers the same two units over four frames: 3.00, 2.50, 2.00 and then 1.50, where it meets the logical state. From 400ms on the two agree again and the next input moves both.

render is only for drawing. A rule that both sides run must read logical() or the authoritative state, never the eased value. For a real discontinuity, such as a respawn or a teleport, call teleport instead, which snaps without easing.

Where to go next#

The shove here is a stand-in for any server decision the client cannot predict. To see a correction caused by the network instead, give the up link some loss. A lost input is never applied on the server, so the next update disagrees with the prediction. In a two-second run with no shove and 10% loss on the up link, this program reports two corrections. It reports the same two on every run, because the Rng is seeded. CorrectionMonitor takes the distance of each Correction and learns what a normal one looks like, so it can flag the abnormal ones without a hand-tuned threshold. AdaptiveDecay clears large errors faster than small ones.

Chapter 20 of the plaza guide explains the predictor choice, the two clocks and the smoothing measurements. The csp_net_example runs the same loop against a full plaza server over channels, built from the lower-level PredictedEntity and ClientInputBuffer.

The plaza_client_utils usage guide on GitHub covers the predictors, the render clock, corrections and the simulated link. The API references are on docs.rs for plaza_client_utils and plaza.