Realtime Shards

​

Set up

rustup target add wasm32-unknown-unknown

[package]
name = "arena-shard"
version = "0.1.0"
edition = "2021"

[lib]
crate-type = ["cdylib"]

[dependencies]
pylon-shard-guest = "0.13"
serde = { version = "1", features = ["derive"] }

[profile.release]
opt-level = "s"
lto = true

​

Write the simulation

use std::time::Duration;

use pylon_shard_guest::{export_shard, Shard};
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct Params {
    width: f32,
    height: f32,
}

#[derive(Deserialize)]
#[serde(rename_all = "snake_case")]
enum Input {
    MoveTo { x: f32, y: f32 },
}

#[derive(Serialize, Clone)]
struct Player {
    id: String,
    x: f32,
    y: f32,
}

#[derive(Serialize)]
struct Snapshot {
    players: Vec<Player>,
}

struct Arena {
    width: f32,
    height: f32,
    players: Vec<Player>,
}

impl Shard for Arena {
    type Params = Params;
    type Input = Input;
    type Snapshot = Snapshot;

    fn init(_shard_id: &str, p: Params) -> Result<Self, String> {
        Ok(Arena { width: p.width, height: p.height, players: Vec::new() })
    }

    fn apply_input(&mut self, sid: &str, input: Input) -> Result<(), String> {
        let Input::MoveTo { x, y } = input;
        if x < 0.0 || x > self.width || y < 0.0 || y > self.height {
            return Err("outside the arena".into());
        }
        match self.players.iter_mut().find(|p| p.id == sid) {
            Some(p) => (p.x, p.y) = (x, y),
            None => self.players.push(Player { id: sid.into(), x, y }),
        }
        Ok(())
    }

    fn tick(&mut self, _dt: Duration) {}

    fn snapshot(&self) -> Snapshot {
        Snapshot { players: self.players.clone() }
    }
}

export_shard!(Arena);

​

Send each player only what is near them

use pylon_shard_guest::{EntityPos, InterestArea, InterestConfig, View};

impl Shard for Zone {
    // ...
    fn interest(&self) -> Option<InterestConfig> {
        Some(InterestConfig { cell_size: 150.0, margin: 10.0, shared_snapshots: true })
    }

    fn entities(&self, out: &mut Vec<EntityPos>) {
        out.extend(self.units.iter().map(|u| EntityPos { id: u.id, x: u.x, y: u.y }));
    }

    fn interest_area(&self, sid: &str) -> Option<InterestArea> {
        let me = self.player_unit(sid)?;
        Some(InterestArea { x: me.x, y: me.y, radius: 150.0 })
    }

    fn can_see(&self, sid: &str, entity: u64) -> bool {
        !self.is_stealthed(entity) || self.can_detect(sid)
    }

    fn snapshot_visible(&self, _sid: &str, view: &View) -> Option<Snapshot> {
        Some(Snapshot { units: view.visible.iter().map(|id| self.unit(*id)).collect() })
    }
}

• An entity leaves view only after it is margin past the radius, so an entity at the edge does not flicker in and out.
• can_see hides an entity in range, for stealth or fog of war.
• view.entered and view.left list the ids that came into and went out of view since the last tick.
• With shared_snapshots: true, subscribers that see the same entities get one snapshot, encoded once. Use it only when snapshot_visible reads nothing but view.visible.
• Positions are 2D. A 3D world passes its ground plane.
• For 300 subscribers and 1,000 moving entities, the grid work takes about 0.2 ms per tick when players are spread out, and about 1.5 ms when they crowd together.

​

Replicate entities instead of sending snapshots

use pylon_shard_guest::{Replicated, ReplicationConfig};

const HP: u8 = 1; // component ids are yours to choose

impl Shard for Zone {
    // ...
    fn tick(&mut self, dt: Duration) {
        for unit in &mut self.units {
            unit.step(dt);
            self.store.set_pos(unit.id, [unit.x, unit.y, 0.0]);
        }
    }

    fn replicated(&mut self) -> Option<&mut Replicated> {
        Some(&mut self.store)
    }

    fn replication_config(&self) -> ReplicationConfig {
        ReplicationConfig { precision: 0.01, max_bytes_per_tick: 900, y_up: false }
    }
}

• despawn for entities that left its view or the store,
• spawn with the full state for entities that came into view,
• update for entities in view that changed: only the axes whose position moved (as a difference) and the components that changed or were removed.

​

Client: a render loop

import { connectShardGame } from "@pylonsync/realtime";

const game = connectShardGame<Input>(shardId, { subscriberId, ticket, tickRate: 20 });

function frame(now: number) {
  game.frame(now);
  for (const id of game.left) removeMesh(id);
  for (const id of game.entered) addMesh(id);
  for (const e of game.entities.values()) {
    moveMesh(e.id, e.x, e.y, e.z, e.components.get(HP));
  }
  requestAnimationFrame(frame);
}
requestAnimationFrame(frame);

• Interpolation. Entities are drawn interpolationDelayMs (default 100) behind the shard’s current tick, between the two frames around that time. Spawns, despawns, and component changes show at the tick they happened. e.from, e.to, and e.t give the samples and the fraction, to interpolate your own data in components (a heading). An id in both left and entered is a new entity that reused the id.
• The clock. game.clock estimates the shard’s current tick from when frames arrive, anchored on the least-delayed frame. Pass tickRate when you know it; otherwise it is measured.
• Prediction. game.predict(step) returns a Predictor that records each input you send. After each frame, reconcile(serverState, ack) starts from the shard’s state for your entity and applies the inputs it has not processed. Refused inputs drop out, and a reconnect clears it.
• Round trip. game.rttMs is the smoothed time from an input to the frame that acknowledges it.
• View tick. game.send stamps each input with the tick the last frame drew, for a shard with lag compensation.
• Reconnects get a full frame. The entities it lacks leave.

​

WebTransport

const game = connectShardGame(shardId, { subscriberId, ticket, tickRate: 20 });
game.connection.transport; // "webtransport" or "websocket"

• "auto" (the default for connectShardGame): WebTransport when the browser has it and the app serves it, else a WebSocket. When a session does not open (UDP blocked, webTransportTimeoutMs, default 3000), the client uses WebSockets from then on.
• "websocket" (the default for connectShard and useShard).
• "webtransport": WebTransport or an error.

​

Lag compensation

1. connectShardGame (TypeScript) and ShardGame (C#) stamp each input with view_tick: the tick the last frame drew.
2. The shard keeps what it sent each subscriber for history_ticks ticks. This is what the client drew, including updates the byte budget deferred and only the entities interest management sent.
3. The shard clamps the view tick and calls apply_input_at with a Rewind, which places each entity where that subscriber drew it then. The clients interpolate the same way, so positions match exactly.

use pylon_shard_guest::{LagCompensation, Rewind, Shard};

impl Shard for Range {
    // ...

    // Keep 10 ticks: 500 ms at 20 Hz.
    fn lag_compensation(&self) -> Option<LagCompensation> {
        Some(LagCompensation::new(10))
    }

    // An input with a view tick comes here.
    fn apply_input_at(&mut self, sid: &str, input: Input, rewind: &Rewind) -> Result<(), String> {
        match input {
            // A hitscan shot: the first target on the line, as the shooter saw it.
            Input::Shoot { aim } => {
                let shooter = self.position_of(sid)?;
                let hit = rewind
                    .entities_near(shooter, RANGE)
                    .into_iter()
                    .filter(|(id, p)| *id != self.entity_of(sid) && on_line(shooter, aim, *p, HIT_RADIUS))
                    .min_by(|a, b| dist(shooter, a.1).total_cmp(&dist(shooter, b.1)));
                if let Some((target, _)) = hit {
                    self.damage(target, 25);
                }
                Ok(())
            }
            // A projectile: spawn it where it is now on the shooter's
            // screen, then advance it to the present. The tick loop moves
            // it from there.
            Input::Throw { dir } => {
                let start = self.position_of(sid)?;
                let flown = rewind.ticks_behind() * TICK_SECS * SPEAR_SPEED;
                self.spawn_spear(add(start, scale(dir, flown)), dir);
                Ok(())
            }
        }
    }

    // Inputs without a view tick (and every input with history off).
    fn apply_input(&mut self, sid: &str, input: Input) -> Result<(), String> {
        // ... check against the current positions.
        Ok(())
    }
}

• rewind.position_at(id) gives where the subscriber drew one entity, or None when it did not draw it. rewind.entities_near(center, radius) gives every entity it drew within the radius; narrow a cone or a swing arc from those.
• rewind.tick() is the clamped view tick, and rewind.ticks_behind() is how far the shooter’s picture was behind the shard.
• The shard clamps the view tick so a client cannot shoot further into the past than its connection explains. The tick is never after the newest tick sent. It is never before the newest tick minus history_ticks. Once the connection’s round trip is known (from WebSocket pings, or QUIC on WebTransport), it is never before the round trip plus max_view_delay_ms (default 250 ms).
• An input without a view tick goes to apply_input. So does every input when lag_compensation returns None.
• In a test outside the server, build a Rewind with Rewind::with_positions(view_tick, now_tick, positions).
• A native Rust shard implements SimState::lag_compensation and SimState::apply_input_at, which gets Option<&Rewind>.
• The rewind interpolates as the default client options do: holdWhenQuiet on and no snapDistance.

pylon bench shard --join joinRange --join-args '{"history":10}' --bots 20 --rate 5 \
  --transport webtransport --net-delay 50 --net-loss 2 \
  --aim 1 --input '{"shoot":["$aim:x","$aim:y"]}'

​

Declare the shard kind

import { buildManifest, shard } from "@pylonsync/sdk";

export default buildManifest({
  // ...
  shards: [
    shard({
      name: "arena",
      wasm: "shards/arena.wasm",
      crate: "shards/arena",
      codec: "msgpack",
      tickRate: 20,
    }),
  ],
});

​

Run a function when a shard ends

• "stopped": ctx.shards.stop or the admin API stopped it.
• "idle": it had no subscribers for idleShutdownSecs.
• "finished": the module ended it.
• "failed": the module trapped, panicked, or went over its budget.

shard({ name: "arena", wasm: "shards/arena.wasm", onStop: "raceEnded" });

// functions/raceEnded.ts
export default mutation({
  internal: true,
  handler: async (ctx, { shardId, reason }) => {
    const race = await ctx.db.lookup("Race", "shardId", shardId);
    if (race) await ctx.db.update("Race", race.id, { endedAt: new Date().toISOString(), endReason: reason });
  },
});

​

Build the module

pylon shards build

​

Start a shard and join it

import { action } from "@pylonsync/functions";

export default action({
  args: {},
  async handler(ctx) {
    const id = "arena-main";
    if (!(await ctx.shards.get(id))) {
      await ctx.shards.create("arena", id, { width: 800, height: 500 });
    }
    const ticket = await ctx.shards.ticket(id);
    return { shardId: id, subscriberId: ctx.auth.userId, ticket };
  },
});

import { useShard } from "@pylonsync/react";

const { snapshot, send, lastRejection } = useShard<Snapshot, Input>(shardId, {
  subscriberId,
  ticket,
});

send({ move_to: { x: 120, y: 80 } });

ticket: async () => (await callFn<{ ticket: string }>("joinArena", {})).ticket,

​

Move a player to another shard

fn transfer_out(&mut self, subscriber: &str) -> Result<Option<Vec<u8>>, String> {
    match self.players.remove(subscriber) {
        Some(p) => serde_json::to_vec(&p).map(Some).map_err(|e| e.to_string()),
        None => Ok(None), // no entity here
    }
}

fn transfer_in(
    &mut self,
    subscriber: &str,
    state: &[u8],
    auth: &Auth,      // the player's user and its ticket for this shard
    returning: bool,  // this shard's own player, back from a failed move
) -> Result<(), String> {
    // Always take a returning player back. Refusing keeps the player in
    // the source.
    if !returning && self.full() {
        return Err("the zone is full".into());
    }
    if !returning && auth.claim("party").is_none() {
        return Err("join a party first".into());
    }
    let p: Player = serde_json::from_slice(state).map_err(|e| e.to_string())?;
    self.players.insert(subscriber.to_string(), p);
    Ok(())
}

export default action({
  args: { from: v.string(), to: v.string() },
  async handler(ctx, args) {
    return ctx.shards.transfer(args.from, ctx.auth.userId!, args.to, {
      claims: { party: "north" },
    });
  },
});

fn transfer_requests(&mut self) -> Vec<(String, String)> {
    // (subscriber, target shard) for each player past the zone line
    std::mem::take(&mut self.leaving)
}

1. The source’s transfer_out removes the entity. The source refuses the player’s inputs until the move ends.
2. The target’s transfer_in adds it. When the target refuses (or is not running), the source’s transfer_in gets the state back, and ctx.shards.transfer throws SHARD_TRANSFER_REFUSED (or SHARD_NOT_FOUND). The player stays in the source, still connected.
3. Each of the player’s connections to the source gets a transfer frame: the target shard and a ticket for it (valid 10 minutes). A connection that comes back to the source within that time gets the frame again, unless it brings a ticket issued after the move (a new join). connectShard, connectShardGame, useShard, and the Swift ShardClient reconnect to the target at once, and onTransfer (or transfers() in Swift) reports it. The target sends a baseline.

​

Messages between shards

fn groups(&self) -> Vec<String> {
    vec![format!("realm:{}", self.realm)]
}

fn outbox(&mut self) -> Vec<Outgoing> {
    // Filled during the tick, taken after it.
    std::mem::take(&mut self.to_send)
}

fn on_message(&mut self, from: &str, topic: &str, data: &[u8]) {
    if topic == "relic_taken" {
        self.banner = Some(String::from_utf8_lossy(data).into_owned());
    }
}

self.to_send.push(Outgoing {
    to: Target::Group("realm:midgard".into()),
    topic: "relic_taken".into(),
    data: b"the relic of Midgard is taken".to_vec(),
});

if (!ctx.auth.isAdmin) throw ctx.error("FORBIDDEN", "GMs only");
await ctx.shards.publish("all", "gm", { command: "restart", inSecs: 300 });

​

The app’s data

​

Call a function

fn calls(&mut self) -> Vec<Call> {
    // Every call still waiting, every tick. The host skips a key that is
    // in flight, so a call is sent again only after a crash.
    self.grants
        .iter()
        .map(|(key, g)| Call {
            key: key.clone(),
            function: "grantItem".into(),
            args: serde_json::json!({ "characterId": g.character, "item": g.item }),
        })
        .collect()
}

fn on_call_result(&mut self, key: &str, result: Result<serde_json::Value, String>) {
    let Some(grant) = self.grants.remove(key) else { return };
    match result {
        // The mutation committed: show the item now.
        Ok(_) => self.give(&grant),
        // The function refused, and rolled back: no item exists. The text
        // is "CODE: message".
        Err(e) => log(Level::Warn, &format!("grant {key} failed: {e}")),
    }
}

• Save the calls in flight with the shard’s state (save). A shard that restarts from its last save sends them again under the same keys.
• Apply the effect (the item in the player’s inventory) only in on_call_result. Before that, the mutation may not have committed.
• Load the characters again after a restore. A grant can commit after the last save, and a zone restored from that save does not know it yet. Keep items by row id (the mutation returns it), so an item that both the reload and a late result bring shows once.
• Make the key unique for good. A stored result is kept for 7 days. The example keeps the next grant number in the character row and uses loot:<character>:<n>. On KEY_REUSED (a restored zone counted from an older row), it sends the grant again under the next number.

​

Write entity fields

fn writes(&mut self) -> Vec<Write> {
    std::mem::take(&mut self.moved)
        .into_iter()
        .map(|id| Write {
            entity: "Character".into(),
            id: self.players[&id].character.clone(),
            set: serde_json::Map::from_iter([("x".into(), self.players[&id].x.into())]),
        })
        .collect()
}

​

Send an input from a function

Input::Heal { who, hp } => {
    if !subscriber.is_empty() {
        return Err("from the server only".into());
    }
    self.players.get_mut(&who).ok_or("no such player")?.hp += hp;
}

export default mutation({
  args: { zone: v.string(), userId: v.string(), hp: v.number() },
  async handler(ctx, args) {
    if (!ctx.auth.isAdmin) throw ctx.error("FORBIDDEN", "GMs only");
    await ctx.db.insert("GmAction", { zone: args.zone, userId: args.userId, action: "heal" });
    await ctx.shards.send(args.zone, { heal: { who: args.userId, hp: args.hp } });
  },
});

​

Load test

pylon bench shard --join joinArena --bots 300 --duration 30 --rate 10 \
  --input '"join"' \
  --input '{"move_to":{"x":"$rand:0:800","y":"$rand:0:500"}}'

• --join <function> signs each bot in as a guest and calls the function, which returns { shardId, subscriberId, ticket }.
• --shard <id> joins a running shard with tickets the command mints. Run it from the app directory, so it uses the same key as pylon dev, or set PYLON_SHARD_TICKET_SECRET to the server’s value.
• --input is a JSON template, and repeated templates are sent in turn. The strings "$rand:A:B", "$int:A:B", "$bot", and "$sid" are replaced for each input.
• --url defaults to http://localhost:4321. An https:// origin connects over wss://.
• --transport webtransport connects over WebTransport, from the URL and hashes at /_pylon/shard/webtransport. --drop-datagrams <percent> drops that share of the datagrams each bot receives, to test loss.
• --net-delay <ms> and --net-loss <percent> (with WebTransport) send every packet through a local UDP proxy that delays and drops packets each way.
• --aim <entity> makes each bot draw entities like a client (--interp-delay, default 100 ms, behind the shard’s tick at --tick-rate, default 20). In an input template, the strings "$aim:x", "$aim:y", and "$aim:z" become where the bot draws the entity, and the input carries its view tick. The report counts the aimed shots the shard refused.

pylon bench shard ... --bots 300 --bot-offset 0   --report a.json   # machine 1
pylon bench shard ... --bots 300 --bot-offset 300 --report b.json   # machine 2
pylon bench merge a.json b.json

​

Metrics and operations

• GET /api/shards and GET /api/shards/<id> (the detail adds subscriber_ids),
• POST /api/shards/<id>/kick with { "subscriber": "<id>" } closes that subscriber’s connections,
• POST /api/shards/<id>/stop stops the shard and closes every connection.

​

Shards across machines

• Placement. ctx.shards.create runs the shard on the live machine with the most free capacity (PYLON_SHARD_CAPACITY shards a machine, default 100), or on the machine you name: ctx.shards.create("arena", id, params, { machine }). maxInstances counts every machine. ctx.shards.get and list report each shard’s machine.
• Routing. A client can connect to any machine. The machine that gets a connection for a shard it does not run sends it on. Between Fly machines (Stack0 Cloud), it answers with fly-replay: instance=<machine>, and Fly’s proxy connects the client to that machine directly, with no extra hop. Elsewhere, it proxies the connection over the machines’ private addresses. The WebSocket (on /shard and on the shard port), the SSE stream, and HTTP inputs all route this way.
• Failover. A machine with no heartbeat for 10 seconds is dead. Each of its shards moves to one live machine (the same choice on every machine) and starts again: from its saved state when the module saves state (below), or from init with its create params when it does not. Clients reconnect on their own (the client libraries retry, and a ticket function gets a new ticket).
• Deploys. A machine that gets SIGTERM hands each shard to another live machine before it exits, and its players stay connected (see Deploys). A shard it cannot hand over is saved and starts on another machine once this one leaves the directory; its clients reconnect after the gap.
• Restarts. A machine that comes back under the same id (a crash and restart, or a deploy on Fly, where the machine id stays) waits until the old process has left (a SIGTERM) or has been silent for 10 seconds, so two processes never hold one id. Then it gets a new lease, and the old process’s shards move from their saved state, possibly back to it.
• Fencing. A machine may run shards for 5 seconds after it sent the last heartbeat that reached the database. When no heartbeat lands in that time (the database is down, slow, or unreachable, even if calls hang instead of failing), a separate thread stops its shards and closes their connections, and the module refuses every call and interrupts a running one. The others start them only when the heartbeat is 10 seconds old. Two copies of a shard never run at once, and a machine cut off from the database cannot overwrite the state of the copy that replaced it.
• A shard that cannot start (its module refuses the saved state) keeps its state, and ctx.shards.get reports why. Stopping it forgets it.

fn save(&self) -> Option<Vec<u8>> {
    serde_json::to_vec(&self.players).ok()
}

fn restore(shard_id: &str, params: Params, state: &[u8]) -> Result<Self, String> {
    let mut arena = Self::init(shard_id, params)?;
    arena.players = serde_json::from_slice(state).map_err(|e| e.to_string())?;
    Ok(arena)
}

​

Deploys

1. Stops ticking the shard. Connections stay open, and the shard’s buffered entity writes are written.
2. Saves the shard’s state.
3. Asks another live machine to take the shard. That machine moves the shard’s place in the directory to itself and starts it from the saved state.
4. Sends each connection a transfer frame for the same shard, with a new ticket, and closes it. The client libraries (JavaScript, React, Swift, and pylon bench shard) reconnect at once. A same-shard transfer does not call onTransfer.

​

Limits and failures

• A tick whose module code uses more CPU time than tickBudgetMs stops the shard.
• A module that grows past memoryMb stops the shard.
• A panic stops the shard. The server log shows the panic message.
• ctx.shards.get(id) returns the reason in error until the host removes the stopped shard (within a few seconds).
• A module may import only pylon.log. The server refuses a module built for WASI or with any other import, and the boot fails.
• A shard’s state is in memory. A module that implements save and restore starts again from its last save after a restart, a deploy, or a crash (saves run every 5 seconds, PYLON_SHARD_SAVE_SECS, and when a machine stops), on SQLite and on Postgres. A player moving between shards when the machine crashed ends up in exactly one of them. A module with no save, or an app with no directory (an in-memory database, or PYLON_SHARD_DIRECTORY=off), loses the shard at a restart.

​

Hosted on Stack0 Cloud

• An open shard connection keeps the app’s machine running.
• A machine with autostop on can stop when no client is connected. Its shards end, like on any restart.
• WebTransport is off by default. Turn it on in the app’s Scaling settings. The app gets a dedicated IPv4 address for UDP (Fly bills it monthly) and must run one machine per region.
• A machine gets 45 seconds to stop, so a deploy saves its shards (and, on an app with several machines, hands them over) before it exits. On one machine, the new one starts them from that save.