Run game time with the scheduler

For Rust developers who have built a plaza controller and now need things to happen later: timeouts, effects that wear off and countdowns

You will build three small programs on plaza's optional scheduler modules. A typing indicator clears itself after three seconds of silence and every keystroke pushes that back. A slow effect wears off after three ticks and a second cast extends it instead of being cut short by the first. A lobby counts down to the start of a match and drops the countdown when a player backs out. None of them uses a wall-clock timer. Time only moves when the controller handles a time step, so every program runs the same way every time.

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

Your [dependencies] section should look like this:

Cargo.toml

[dependencies]
plaza = "0"
tokio = { version = "1", features = ["macros", "rt"] }
async-trait = "0.1"

The schedulers are part of the plaza crate and need no feature flag. The programs use Tokio's single-threaded current_thread runtime. They also send each client op straight to the controller with ControllerCommand::SubmitAgentOps instead of through the session. That puts ops and time steps on the same queue, so each op lands at a known point in game time and the output is the same on every run. A real client sends its ops through the session and the rules handle them identically.

Time out a typing indicator#

The controller does not know what time it is. A TickDriver sends it ProcessTimeStep commands and the rules receive each one as LogicInput::TimeStep { delta_time }. The rules add each step to a clock they keep in the state. That clock is the only time the schedulers ever see.

TimeEventScheduler<E> is a scheduler on that game-time axis. schedule_after(now, delay, event) stores an event of your own type and returns a ScheduledEventId. tick(now) hands back every event that has come due. The scheduler lives in the state, so it is owned by the controller like everything else and needs no lock.

For a typing indicator, each keystroke schedules a StopTyping event three seconds out. If the user was already typing, the rules cancel the previous event by its id, so only the latest keystroke's timeout is ever pending.

Replace src/main.rs with this:

src/main.rs

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

use async_trait::async_trait;
use plaza::{
    agent::Agent,
    common::scheduler::{ScheduledEventId, TimeEventScheduler},
    controller::{CommandSender, ControllerCommand, StateControllerBuilder, query_with},
    session::{InProcessSession, TargetedOp, in_process::ClientInbox},
    state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError},
    tick_driver::TickDriver,
};

type UserId = u32;

const TYPING_TIMEOUT: Duration = Duration::from_secs(3);

#[derive(Clone, Debug)]
enum RoomEvent {
    StopTyping(UserId),
}

#[derive(Debug, Default)]
struct Room {
    now: Duration,
    typing: HashMap<UserId, ScheduledEventId>,
    timers: TimeEventScheduler<RoomEvent>,
}

#[derive(Clone, Debug)]
enum ChatOp {
    Keystroke,
    Typing { user: UserId, typing: bool },
}

struct RoomRules;

#[async_trait]
impl StateLogic<ChatOp, UserId, Room> for RoomRules {
    async fn process_input(
        &self,
        state: &mut Room,
        input: LogicInput<ChatOp, UserId>,
    ) -> Result<LogicOutput<ChatOp, UserId>, StateLogicError> {
        let mut out = Vec::new();
        match input {
            LogicInput::AgentOps { source, ops } => {
                let Some(user) = source.id_cloned() else {
                    return Ok(LogicOutput::none());
                };
                for op in ops {
                    let ChatOp::Keystroke = op else { continue };
                    let timer = state.timers.schedule_after(
                        state.now,
                        TYPING_TIMEOUT,
                        RoomEvent::StopTyping(user),
                    );
                    match state.typing.insert(user, timer) {
                        Some(previous) => {
                            state.timers.cancel(previous);
                        }
                        None => out.push(TargetedOp::new_system_all(vec![ChatOp::Typing {
                            user,
                            typing: true,
                        }])),
                    }
                }
            }
            LogicInput::TimeStep { delta_time } => {
                state.now += delta_time;
                for event in state.timers.tick(state.now) {
                    let RoomEvent::StopTyping(user) = event;
                    state.typing.remove(&user);
                    out.push(TargetedOp::new_system_all(vec![ChatOp::Typing {
                        user,
                        typing: false,
                    }]));
                }
            }
            _ => {}
        }
        Ok(out.into())
    }
}

type Tx = CommandSender<ChatOp, UserId, Room>;

async fn keystroke(tx: &Tx, user: &Agent<UserId>) -> Result<(), Box<dyn std::error::Error>> {
    tx.send(ControllerCommand::SubmitAgentOps {
        agent: user.clone(),
        ops: vec![ChatOp::Keystroke],
    })
    .await?;
    Ok(())
}

async fn look(
    tx: &Tx,
    inbox: &mut ClientInbox<ChatOp, UserId>,
) -> Result<(), Box<dyn std::error::Error>> {
    let now = query_with(tx, |room| room.now).await?;
    let mut seen = false;
    while let Ok(msg) = inbox.try_recv() {
        for op in msg.ops {
            if let ChatOp::Typing { user, typing } = op {
                let doing = if typing {
                    "is typing"
                } else {
                    "stopped typing"
                };
                println!("t={}s  bob sees user {user} {doing}", now.as_secs());
                seen = true;
            }
        }
    }
    if !seen {
        println!("t={}s  nothing new", now.as_secs());
    }
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let session = InProcessSession::<ChatOp, UserId>::new();
    let (tx, controller) = StateControllerBuilder::without_snapshots(
        Arc::new(RoomRules),
        session.clone(),
        Room::default(),
    )
    .build();
    tokio::spawn(controller.run());

    let alice = Agent::new_human(1);
    let (_, mut bob_inbox) = session.connect(Agent::new_human(2)).await?;
    let second = Duration::from_secs(1);

    keystroke(&tx, &alice).await?;
    look(&tx, &mut bob_inbox).await?;

    TickDriver::run_virtual(&tx, second, 2).await;
    keystroke(&tx, &alice).await?;
    look(&tx, &mut bob_inbox).await?;

    TickDriver::run_virtual(&tx, second, 2).await;
    look(&tx, &mut bob_inbox).await?;

    TickDriver::run_virtual(&tx, second, 1).await;
    look(&tx, &mut bob_inbox).await?;

    Ok(())
}

The builder uses without_snapshots because a client joining this room needs no catch-up. TickDriver::run_virtual(&tx, step, n) sends n time steps of step each without waiting for real time to pass, which is how a test or a demo skips ahead. On a server you would spawn TickDriver::from_hz(10).run(tx.clone()) instead and let real time drive it.

look asks the controller for the time with query_with and then prints whatever Bob's inbox holds. The query travels the same command queue as the ops and the time steps, so by the time it answers, everything sent before it has been handled and its broadcasts are already in the inbox. try_recv then drains the inbox without waiting.

Run it:

$ cargo run
t=0s  bob sees user 1 is typing
t=2s  nothing new
t=4s  nothing new
t=5s  bob sees user 1 stopped typing

Alice's first keystroke at 0s would have expired at 3s. Her second keystroke at 2s cancelled it and scheduled a new one for 5s, so Bob sees nothing at 4s and the indicator clears at 5s.

Undo an effect with a timer#

The same pattern undoes an effect. A Slow op marks the target as slowed and schedules a SlowEnds event. When the event fires, the rules clear the flag and tell everyone.

This program counts in ticks instead of seconds. TickEventScheduler<E> takes a u64 instead of a Duration and the rules add one to tick on every time step, whatever its length. Pick ticks when the game thinks in turns or frames and game time when it thinks in seconds.

Replace src/main.rs with this:

src/main.rs

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

use async_trait::async_trait;
use plaza::{
    agent::Agent,
    common::scheduler::TickEventScheduler,
    controller::{CommandSender, ControllerCommand, StateControllerBuilder, query_with},
    session::{InProcessSession, TargetedOp, in_process::ClientInbox},
    state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError},
    tick_driver::TickDriver,
};

type PlayerId = u32;

const SLOW_TICKS: u64 = 3;
const ALICE: PlayerId = 1;
const BOB: PlayerId = 2;

#[derive(Clone, Debug)]
enum ArenaEvent {
    SlowEnds(PlayerId),
}

#[derive(Debug, Default)]
struct Player {
    slowed: bool,
}

#[derive(Debug, Default)]
struct Arena {
    tick: u64,
    players: HashMap<PlayerId, Player>,
    timers: TickEventScheduler<ArenaEvent>,
}

#[derive(Clone, Debug)]
enum ArenaOp {
    Slow { target: PlayerId },
    Slowed { target: PlayerId },
    Recovered { target: PlayerId },
}

fn to_all(op: ArenaOp) -> TargetedOp<ArenaOp, PlayerId> {
    TargetedOp::new_system_all(vec![op])
}

struct ArenaRules;

#[async_trait]
impl StateLogic<ArenaOp, PlayerId, Arena> for ArenaRules {
    async fn process_input(
        &self,
        state: &mut Arena,
        input: LogicInput<ArenaOp, PlayerId>,
    ) -> Result<LogicOutput<ArenaOp, PlayerId>, StateLogicError> {
        let mut out = Vec::new();
        match input {
            LogicInput::AgentOps { ops, .. } => {
                for op in ops {
                    let ArenaOp::Slow { target } = op else {
                        continue;
                    };
                    state.players.entry(target).or_default().slowed = true;
                    state.timers.schedule_after(
                        state.tick,
                        SLOW_TICKS,
                        ArenaEvent::SlowEnds(target),
                    );
                    out.push(to_all(ArenaOp::Slowed { target }));
                }
            }
            LogicInput::TimeStep { .. } => {
                state.tick += 1;
                for event in state.timers.tick(state.tick) {
                    let ArenaEvent::SlowEnds(target) = event;
                    if let Some(player) = state.players.get_mut(&target) {
                        player.slowed = false;
                        out.push(to_all(ArenaOp::Recovered { target }));
                    }
                }
            }
            _ => {}
        }
        Ok(out.into())
    }
}

type Tx = CommandSender<ArenaOp, PlayerId, Arena>;

async fn slow(
    tx: &Tx,
    caster: &Agent<PlayerId>,
    target: PlayerId,
) -> Result<(), Box<dyn std::error::Error>> {
    tx.send(ControllerCommand::SubmitAgentOps {
        agent: caster.clone(),
        ops: vec![ArenaOp::Slow { target }],
    })
    .await?;
    Ok(())
}

async fn look(
    tx: &Tx,
    inbox: &mut ClientInbox<ArenaOp, PlayerId>,
) -> Result<(), Box<dyn std::error::Error>> {
    let (tick, slowed) = query_with(tx, |arena| {
        (
            arena.tick,
            arena.players.get(&ALICE).is_some_and(|p| p.slowed),
        )
    })
    .await?;
    let mut news = Vec::new();
    while let Ok(msg) = inbox.try_recv() {
        for op in msg.ops {
            match op {
                ArenaOp::Slowed { target } => news.push(format!("player {target} slowed")),
                ArenaOp::Recovered { target } => news.push(format!("player {target} recovered")),
                ArenaOp::Slow { .. } => {}
            }
        }
    }
    let news = if news.is_empty() {
        "nothing new".to_string()
    } else {
        news.join(", ")
    };
    println!("tick {tick}  alice sees {news:<20} slowed: {slowed}");
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let session = InProcessSession::<ArenaOp, PlayerId>::new();
    let (tx, controller) = StateControllerBuilder::without_snapshots(
        Arc::new(ArenaRules),
        session.clone(),
        Arena::default(),
    )
    .build();
    tokio::spawn(controller.run());

    let (_, mut alice_inbox) = session.connect(Agent::new_human(ALICE)).await?;
    let bob = Agent::new_human(BOB);
    let tick = Duration::from_millis(100);

    slow(&tx, &bob, ALICE).await?;
    look(&tx, &mut alice_inbox).await?;
    for _ in 0..4 {
        TickDriver::run_virtual(&tx, tick, 1).await;
        look(&tx, &mut alice_inbox).await?;
    }

    Ok(())
}

look prints what Alice's inbox received and whether the controller says she is slowed.

Run it:

$ cargo run
tick 0  alice sees player 1 slowed      slowed: true
tick 1  alice sees nothing new          slowed: true
tick 2  alice sees nothing new          slowed: true
tick 3  alice sees player 1 recovered   slowed: false
tick 4  alice sees nothing new          slowed: false

Bob slows Alice at tick 0 and the effect ends at tick 3, three ticks later.

Drop stale scheduled work with a Mark#

Now Bob casts the slow a second time at tick 2, while Alice is still slowed. The second cast should run for three ticks from tick 2 and end at tick 5. Replace main with this:

src/main.rs

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let session = InProcessSession::<ArenaOp, PlayerId>::new();
    let (tx, controller) = StateControllerBuilder::without_snapshots(
        Arc::new(ArenaRules),
        session.clone(),
        Arena::default(),
    )
    .build();
    tokio::spawn(controller.run());

    let (_, mut alice_inbox) = session.connect(Agent::new_human(ALICE)).await?;
    let bob = Agent::new_human(BOB);
    let tick = Duration::from_millis(100);

    slow(&tx, &bob, ALICE).await?;
    look(&tx, &mut alice_inbox).await?;
    for n in 1..=6 {
        TickDriver::run_virtual(&tx, tick, 1).await;
        if n == 2 {
            slow(&tx, &bob, ALICE).await?;
        }
        look(&tx, &mut alice_inbox).await?;
    }

    Ok(())
}

Run it:

$ cargo run
tick 0  alice sees player 1 slowed      slowed: true
tick 1  alice sees nothing new          slowed: true
tick 2  alice sees player 1 slowed      slowed: true
tick 3  alice sees player 1 recovered   slowed: false
tick 4  alice sees nothing new          slowed: false
tick 5  alice sees player 1 recovered   slowed: false
tick 6  alice sees nothing new          slowed: false

The first cast's SlowEnds still fired at tick 3 and ended the second cast two ticks early. By the time scheduled work fires, the game may have moved on, so the work has to check whether the moment it was scheduled for still stands.

The typing indicator handled this by cancelling the old event. The other way is a Situation, a counter for one recurring decision, here how long Alice stays slowed. Call advance whenever that decision changes, stamp the scheduled event with mark() and check holds(mark) when it fires. A mark taken before the last advance no longer holds.

Add the import game_common::flow_control::{Mark, Situation} to the plaza list. Each player gets their own Situation and the event carries a Mark:

src/main.rs

#[derive(Clone, Debug)]
enum ArenaEvent {
    SlowEnds { target: PlayerId, mark: Mark },
}

#[derive(Debug, Default)]
struct Player {
    slowed: bool,
    slow: Situation,
}

A cast advances the target's situation before scheduling:

src/main.rs

let player = state.players.entry(target).or_default();
player.slowed = true;
player.slow.advance();
let mark = player.slow.mark();
state.timers.schedule_after(
    state.tick,
    SLOW_TICKS,
    ArenaEvent::SlowEnds { target, mark },
);
out.push(to_all(ArenaOp::Slowed { target }));

When the event fires, it does nothing unless its mark still holds:

src/main.rs

for event in state.timers.tick(state.tick) {
    let ArenaEvent::SlowEnds { target, mark } = event;
    let Some(player) = state.players.get_mut(&target) else {
        continue;
    };
    if !player.slow.holds(mark) {
        continue;
    }
    player.slowed = false;
    out.push(to_all(ArenaOp::Recovered { target }));
}

Run it:

$ cargo run
tick 0  alice sees player 1 slowed      slowed: true
tick 1  alice sees nothing new          slowed: true
tick 2  alice sees player 1 slowed      slowed: true
tick 3  alice sees nothing new          slowed: true
tick 4  alice sees nothing new          slowed: true
tick 5  alice sees player 1 recovered   slowed: false
tick 6  alice sees nothing new          slowed: false

The tick 3 event carried the first cast's mark, which stopped holding when the second cast advanced the situation, so it was skipped. Alice stays slowed until tick 5. Mark is Copy and holds one u64, so an event carries it by value.

A state with more than one kind of decision can give each Situation a marker type, as in Situation<Ask> and Situation<March>. Checking a Mark<Ask> against a Situation<March> is then a compile error. A state with one kind uses the default and writes plain Situation and Mark.

Drop scheduled work when the phase moves on#

Some work belongs to a phase of the game rather than to one decision. A lobby that counts down to the start of a match is one. If a player backs out during the countdown, the lobby goes back to waiting and the pending start has to be dropped. If everyone readies up again, a new countdown starts from the beginning.

Phased<P> holds the current phase. The only way to change it is transition_to, which also queues a PhaseChangedNoticePayload for every client, so a client can never miss a phase change. Each change also starts a new occupancy of the phase, identified by an Epoch. PhasedScheduler<E> is a tick scheduler that records the epoch when you schedule and whose due(now, &phase) returns only events whose occupancy is still current. It drops the rest.

transition_to takes a context to push its notice into and a function that wraps the notice in your op type. OpsQueue is the minimal context. ctx.into_ops() turns what it collected into the rules' output.

Replace src/main.rs with this:

src/main.rs

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

use async_trait::async_trait;
use plaza::{
    agent::Agent,
    common::fsm::OpsQueue,
    controller::{CommandSender, ControllerCommand, StateControllerBuilder, query_with},
    game_common::flow_control::{
        Phased, PhasedScheduler, phases::op_payloads::PhaseChangedNoticePayload,
    },
    session::{InProcessSession, in_process::ClientInbox},
    state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError},
    tick_driver::TickDriver,
};

type PlayerId = u32;

const ALICE: PlayerId = 1;
const BOB: PlayerId = 2;
const PLAYERS: usize = 2;
const COUNTDOWN_TICKS: u64 = 3;

#[derive(Clone, Debug, PartialEq)]
enum Stage {
    Waiting,
    Countdown,
    Playing,
}

#[derive(Clone, Debug)]
enum LobbyEvent {
    Start,
}

#[derive(Debug)]
struct Lobby {
    tick: u64,
    stage: Phased<Stage>,
    ready: BTreeSet<PlayerId>,
    timers: PhasedScheduler<LobbyEvent>,
}

impl Default for Lobby {
    fn default() -> Self {
        Self {
            tick: 0,
            stage: Phased::new(Stage::Waiting),
            ready: BTreeSet::new(),
            timers: PhasedScheduler::new(),
        }
    }
}

#[derive(Clone, Debug)]
enum LobbyOp {
    Ready,
    Unready,
    StageChanged(PhaseChangedNoticePayload<Stage>),
}

struct LobbyRules;

#[async_trait]
impl StateLogic<LobbyOp, PlayerId, Lobby> for LobbyRules {
    async fn process_input(
        &self,
        state: &mut Lobby,
        input: LogicInput<LobbyOp, PlayerId>,
    ) -> Result<LogicOutput<LobbyOp, PlayerId>, StateLogicError> {
        let mut ctx = OpsQueue::new();
        match input {
            LogicInput::AgentOps { source, ops } => {
                let Some(player) = source.id_cloned() else {
                    return Ok(LogicOutput::none());
                };
                for op in ops {
                    match op {
                        LobbyOp::Ready => state.ready.insert(player),
                        LobbyOp::Unready => state.ready.remove(&player),
                        LobbyOp::StageChanged(_) => continue,
                    };
                    let everyone_ready = state.ready.len() == PLAYERS;
                    match (state.stage.current(), everyone_ready) {
                        (Stage::Waiting, true) => {
                            state.stage.transition_to(
                                Stage::Countdown,
                                &mut ctx,
                                LobbyOp::StageChanged,
                            );
                            state.timers.schedule_after(
                                state.tick,
                                COUNTDOWN_TICKS,
                                &state.stage,
                                LobbyEvent::Start,
                            );
                        }
                        (Stage::Countdown, false) => {
                            state.stage.transition_to(
                                Stage::Waiting,
                                &mut ctx,
                                LobbyOp::StageChanged,
                            );
                        }
                        _ => {}
                    }
                }
            }
            LogicInput::TimeStep { .. } => {
                state.tick += 1;
                for event in state.timers.due(state.tick, &state.stage) {
                    match event {
                        LobbyEvent::Start => {
                            state.stage.transition_to(
                                Stage::Playing,
                                &mut ctx,
                                LobbyOp::StageChanged,
                            );
                        }
                    }
                }
            }
            _ => {}
        }
        Ok(ctx.into_ops().into())
    }
}

type Tx = CommandSender<LobbyOp, PlayerId, Lobby>;

async fn send(tx: &Tx, player: PlayerId, op: LobbyOp) -> Result<(), Box<dyn std::error::Error>> {
    tx.send(ControllerCommand::SubmitAgentOps {
        agent: Agent::new_human(player),
        ops: vec![op],
    })
    .await?;
    Ok(())
}

async fn look(
    tx: &Tx,
    inbox: &mut ClientInbox<LobbyOp, PlayerId>,
) -> Result<(), Box<dyn std::error::Error>> {
    let tick = query_with(tx, |lobby| lobby.tick).await?;
    let mut news = Vec::new();
    while let Ok(msg) = inbox.try_recv() {
        for op in msg.ops {
            if let LobbyOp::StageChanged(notice) = op {
                news.push(format!("stage {:?}", notice.new_phase));
            }
        }
    }
    let news = if news.is_empty() {
        "nothing new".to_string()
    } else {
        news.join(", ")
    };
    println!("tick {tick}  alice sees {news}");
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let session = InProcessSession::<LobbyOp, PlayerId>::new();
    let (tx, controller) = StateControllerBuilder::without_snapshots(
        Arc::new(LobbyRules),
        session.clone(),
        Lobby::default(),
    )
    .build();
    tokio::spawn(controller.run());

    let (_, mut alice_inbox) = session.connect(Agent::new_human(ALICE)).await?;
    let tick = Duration::from_millis(100);

    send(&tx, ALICE, LobbyOp::Ready).await?;
    send(&tx, BOB, LobbyOp::Ready).await?;
    look(&tx, &mut alice_inbox).await?;

    for n in 1..=5 {
        TickDriver::run_virtual(&tx, tick, 1).await;
        match n {
            1 => send(&tx, BOB, LobbyOp::Unready).await?,
            2 => send(&tx, BOB, LobbyOp::Ready).await?,
            _ => {}
        }
        look(&tx, &mut alice_inbox).await?;
    }

    Ok(())
}

Phased::new announces nothing. The opening phase is part of the state, not a change. The start is scheduled after the transition to Countdown, so it belongs to that occupancy. Scheduling it before the transition would stamp it with the occupancy that just ended.

Run it:

$ cargo run
tick 0  alice sees stage Countdown
tick 1  alice sees stage Waiting
tick 2  alice sees stage Countdown
tick 3  alice sees nothing new
tick 4  alice sees nothing new
tick 5  alice sees stage Playing

Both players ready up at tick 0 and the start is due at tick 3. Bob backs out at tick 1 and readies again at tick 2, so the second countdown's start is due at tick 5. At tick 3 the first start came due, but its occupancy of Countdown had ended, so due dropped it and the match started at tick 5 as it should. Returning to Countdown was a new occupancy even though the phase has the same value.

PhasedScheduler checks only the epoch. When the phase stays the same and something smaller moves on, such as a turn passing, pair it with a Mark as in the previous section. PhasedScheduler has no cancel, so a Mark is the way to retire one of its events early.

Where to go next#

There are two scheduler shapes and each runs on either axis. EventScheduler returns your events for the rules to match on, as all three programs here do. CallbackScheduler runs closures against the state instead and comes as TickCallbackScheduler and TimeCallbackScheduler. Repeating schedules (schedule_repeating_after) skip ahead after a stall rather than firing every missed interval. common::reconnect::ReconnectTracker uses the same no-timers approach for disconnect grace periods.

The plaza guide's One loop, one truth explains why time is an input to the controller. The typing_indicator and timed_debuff examples are fuller versions of the first two programs.

The plaza usage guide on GitHub has the section on dropping stale scheduled work. The full API reference is on docs.rs.