From 96b73922cce4c27212e10663b01b03f97c897b98 Mon Sep 17 00:00:00 2001 From: Syncer Deploy Date: Fri, 28 Aug 2026 11:27:53 -0300 Subject: [PATCH] add generic stateful input streaming --- README.md | 43 +++++ apps/web/src/useFluxClient.ts | 33 +++- apps/web/src/useGameClient.ts | 54 +++---- apps/web/src/useRoyaleClient.ts | 55 +++---- packages/engine/src/index.ts | 10 ++ packages/engine/src/input-stream.ts | 194 +++++++++++++++++++++++ packages/engine/src/network-clock.ts | 5 +- packages/engine/src/networked-server.ts | 202 +++++++++++++++++++++++- packages/engine/src/networked-types.ts | 3 + packages/engine/src/time-travel.ts | 10 ++ packages/engine/test/engine.test.mjs | 154 ++++++++++++++++++ packages/shared/src/flux-game.ts | 10 +- packages/shared/src/royale-game.ts | 32 +++- packages/shared/src/shooter-game.ts | 37 ++++- packages/shared/test/flux.test.mjs | 6 +- 15 files changed, 767 insertions(+), 81 deletions(-) create mode 100644 packages/engine/src/input-stream.ts diff --git a/README.md b/README.md index 1fc4f5f..e6cd152 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,48 @@ const protocol = game.protocol; Flux Relay in `packages/shared/src/flux-game.ts` is the compact reference implementation. Both games are mounted by the same `hostNetworkedGame()` transport adapter, demonstrating that the server loop has no shooter knowledge. +## Stream stateful input safely + +`withInputStream()` turns edge-only input delivery into an acknowledged +latest-state stream. Changes receive a new sequence immediately; unchanged +heartbeats resend the exact same packet, so they refresh the authority timeout +without executing an action twice. If the stream goes silent, the server +applies one developer-defined neutral input instead of allowing stuck movement: + +```ts +const streamedGame = withInputStream(game, { + heartbeatRateHz: 20, + timeoutMs: 400, + inputsEqual: (left, right) => + left.forward === right.forward && + left.fire === right.fire && + left.reload === right.reload, + neutralize: (lastInput) => ({ + ...lastInput, + forward: 0, + fire: false, + reload: false, + }), + // A heartbeat after a timeout may restore persistent state, but must strip + // edge actions that are not allowed to execute again. + resume: (lastClientInput) => ({ + ...lastClientInput, + reload: false, + }), +}); + +const inputStream = createInputStateStream(streamedGame); +inputStream.update(currentInput); +const emission = inputStream.consume(performance.now()); +``` + +`emission.kind === "state"` receives a new `engine.createInput()` packet. +`"heartbeat"` resends the previously encoded packet verbatim. This preserves +exactly-once sequence semantics while the generic authority owns timeout and +resume behavior. Stream state is part of deterministic checkpoints, so time +travel and branching reproduce the same dead-man-switch decisions. Apply this +HOF before authority wrappers such as lag compensation and time travel. + ## Scale a game with spatial replication `withSpatialReplication()` wraps any networked game with a deterministic grid index, per-viewer interest selection, priorities, and an explicit byte budget. The ordinary visibility callback still runs first, so this layer can reduce an already-safe projection but can never reveal private authority state: @@ -292,6 +334,7 @@ Replication is perception-aware. A client snapshot includes itself, public picku - Private authoritative snapshot history and developer-defined client-state validation. - Client prediction and reconciliation from server snapshots. - Ping/pong RTT, jitter, clock-offset estimation, and adaptive input lead. +- Latest-state input heartbeats, duplicate-safe sequence reuse, and deterministic authority timeouts. - Per-viewer state projection and internal-event-to-perception filtering. - Generic binary message framing around developer-provided input, visible-state, and perception codecs. - Optional visibility grouping for sharing encoded snapshots without weakening privacy. diff --git a/apps/web/src/useFluxClient.ts b/apps/web/src/useFluxClient.ts index da73b70..cfb21cc 100644 --- a/apps/web/src/useFluxClient.ts +++ b/apps/web/src/useFluxClient.ts @@ -1,5 +1,9 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; -import { FixedStepClock, type NetworkStats } from "@syncer/engine"; +import { + createInputStateStream, + FixedStepClock, + type NetworkStats, +} from "@syncer/engine"; import { FLUX_SOCKET_PATH, fluxGame, @@ -61,6 +65,8 @@ export function useFluxClient(): FluxClientView { let connection: ConnectionStatus = "connecting"; let validation: ValidationStatus = "waiting"; let input: FluxInput = { thrust: false }; + const inputStream = createInputStateStream(fluxGame); + let lastInputFrame: ArrayBuffer | null = null; const publish = () => { if (!active) return; @@ -79,9 +85,18 @@ export function useFluxClient(): FluxClientView { if (socket?.readyState === WebSocket.OPEN) socket.send(frame); }; - const sendInput = () => { + const sendInput = (force = false) => { if (!engine.initialized) return; - send(protocol.encodeClient({ kind: "input", packet: engine.createInput(input) })); + inputStream.update(input); + const emission = inputStream.consume(performance.now(), force); + if (!emission) return; + if (emission.kind === "state" || !lastInputFrame) { + lastInputFrame = protocol.encodeClient({ + kind: "input", + packet: engine.createInput(emission.input), + }); + } + send(lastInputFrame); }; const updateThrust = (thrust: boolean) => { @@ -111,8 +126,10 @@ export function useFluxClient(): FluxClientView { engine.initialize(message.playerId, message.snapshot); clock.reset(performance.now()); input = { thrust: false }; + inputStream.reset(input); + lastInputFrame = null; validation = "waiting"; - sendInput(); + sendInput(true); break; case "snapshot": engine.reconcile(message.snapshot); @@ -128,7 +145,10 @@ export function useFluxClient(): FluxClientView { break; case "reject-input": engine.reject(message.sequence); + inputStream.invalidate(); + lastInputFrame = null; validation = "invalid"; + sendInput(true); break; case "event": engine.receiveEvent(message.event, message.tick); @@ -174,6 +194,10 @@ export function useFluxClient(): FluxClientView { ping: engine.networkClock.createPing(performance.now()), })); }, 1_000); + const inputTimer = window.setInterval( + () => sendInput(), + inputStream.policy.heartbeatIntervalMs, + ); const validationTimer = window.setInterval(() => { if (engine.initialized) { send(protocol.encodeClient({ @@ -195,6 +219,7 @@ export function useFluxClient(): FluxClientView { setThrustRef.current = () => undefined; window.clearTimeout(retryTimer); window.clearInterval(pingTimer); + window.clearInterval(inputTimer); window.clearInterval(validationTimer); if (animationFrame !== undefined) window.cancelAnimationFrame(animationFrame); window.removeEventListener("keydown", handleKeyDown); diff --git a/apps/web/src/useGameClient.ts b/apps/web/src/useGameClient.ts index 7bf9046..f45a493 100644 --- a/apps/web/src/useGameClient.ts +++ b/apps/web/src/useGameClient.ts @@ -1,12 +1,12 @@ import { useEffect, useMemo, useState } from "react"; import { + createInputStateStream, FixedStepClock, type NetworkStats, type ProjectedReplayFrame, } from "@syncer/engine"; import { GAME_SOCKET_PATH, - WEAPONS, Weapon, shooterGame, type ShooterInput, @@ -110,8 +110,8 @@ export function useGameClient(): GameClientView { let validation: ValidationStatus = "waiting"; const pressedKeys = new Set(); let input: ShooterInput = { ...neutralInput }; - let lastSentInput: ShooterInput | null = null; - let inputDirty = true; + const inputStream = createInputStateStream(shooterGame); + let lastInputFrame: ArrayBuffer | null = null; const incomingReplays = new Map(); let activeReplay: ActiveReplay | null = null; @@ -178,15 +178,15 @@ export function useGameClient(): GameClientView { }; const sendInput = (force = false) => { - if (!engine.initialized || (!force && !inputDirty)) return; - if (!force && lastSentInput && inputsEqual(input, lastSentInput)) { - inputDirty = false; - return; + if (!engine.initialized) return; + inputStream.update(input); + const emission = inputStream.consume(performance.now(), force); + if (!emission) return; + if (emission.kind === "state" || !lastInputFrame) { + const packet = engine.createInput(emission.input); + lastInputFrame = protocol.encodeClient({ kind: "input", packet }); } - const packet = engine.createInput(input); - send(protocol.encodeClient({ kind: "input", packet })); - lastSentInput = { ...input }; - inputDirty = false; + send(lastInputFrame); }; const updateMovement = () => { @@ -196,7 +196,6 @@ export function useGameClient(): GameClientView { forward: Number(pressedKeys.has("w")) - Number(pressedKeys.has("s")), sprint: pressedKeys.has("shift"), }; - inputDirty = true; sendInput(); }; @@ -229,8 +228,8 @@ export function useGameClient(): GameClientView { pitch: local?.pitch ?? 0, weapon: local?.weapon ?? Weapon.PulseRifle, }; - lastSentInput = null; - inputDirty = true; + inputStream.reset(input); + lastInputFrame = null; validation = "waiting"; sendInput(true); break; @@ -249,7 +248,10 @@ export function useGameClient(): GameClientView { break; case "reject-input": engine.reject(message.sequence); + inputStream.invalidate(); + lastInputFrame = null; validation = "invalid"; + sendInput(true); break; case "event": engine.receiveEvent(message.event, message.tick); @@ -334,7 +336,6 @@ export function useGameClient(): GameClientView { updateMovement(); } else if (key === "r") { input = { ...input, reload: true }; - inputDirty = true; sendInput(); } else if (key === "1" || key === "2" || key === "3") { input = { @@ -346,7 +347,6 @@ export function useGameClient(): GameClientView { ? Weapon.Scattergun : Weapon.RailRifle, }; - inputDirty = true; sendInput(); } }; @@ -359,7 +359,6 @@ export function useGameClient(): GameClientView { updateMovement(); } else if (key === "r") { input = { ...input, reload: false }; - inputDirty = true; sendInput(); } }; @@ -371,13 +370,11 @@ export function useGameClient(): GameClientView { yaw: normalizeAngle(input.yaw + event.movementX * 0.00225), pitch: clamp(input.pitch - event.movementY * 0.0019, -1.25, 1.25), }; - inputDirty = true; }; const handleMouseDown = (event: MouseEvent) => { if (activeReplay || event.button !== 0 || !document.pointerLockElement) return; input = { ...input, fire: true }; - inputDirty = true; sendInput(); }; @@ -385,14 +382,12 @@ export function useGameClient(): GameClientView { if (event.button !== 0) return; if (activeReplay) return; input = { ...input, fire: false }; - inputDirty = true; sendInput(); }; const releaseControls = () => { pressedKeys.clear(); input = { ...input, strafe: 0, forward: 0, fire: false, sprint: false, reload: false }; - inputDirty = true; sendInput(true); }; @@ -409,8 +404,8 @@ export function useGameClient(): GameClientView { document.addEventListener("pointerlockchange", handlePointerLockChange); const inputTimer = window.setInterval( - () => sendInput(input.fire && WEAPONS[input.weapon].automatic), - 1_000 / 30, + () => sendInput(), + inputStream.policy.heartbeatIntervalMs, ); const pingTimer = window.setInterval(() => { send(protocol.encodeClient({ kind: "ping", ping: engine.networkClock.createPing(performance.now()) })); @@ -445,19 +440,6 @@ export function useGameClient(): GameClientView { return view; } -function inputsEqual(left: ShooterInput, right: ShooterInput): boolean { - return ( - left.strafe === right.strafe && - left.forward === right.forward && - left.yaw === right.yaw && - left.pitch === right.pitch && - left.fire === right.fire && - left.sprint === right.sprint && - left.reload === right.reload && - left.weapon === right.weapon - ); -} - function normalizeAngle(value: number): number { return Math.atan2(Math.sin(value), Math.cos(value)); } diff --git a/apps/web/src/useRoyaleClient.ts b/apps/web/src/useRoyaleClient.ts index ba2dbdf..3fc8e9a 100644 --- a/apps/web/src/useRoyaleClient.ts +++ b/apps/web/src/useRoyaleClient.ts @@ -1,5 +1,9 @@ import { useEffect, useMemo, useState } from "react"; -import { FixedStepClock, type NetworkStats } from "@syncer/engine"; +import { + createInputStateStream, + FixedStepClock, + type NetworkStats, +} from "@syncer/engine"; import { ROYALE_SOCKET_PATH, royaleGame, @@ -72,8 +76,8 @@ export function useRoyaleClient(): RoyaleClientView { let connection: ConnectionStatus = "connecting"; let validation: ValidationStatus = "waiting"; let input: RoyaleInput = { ...neutralInput }; - let lastSent: RoyaleInput | null = null; - let inputDirty = false; + const inputStream = createInputStateStream(royaleGame); + let lastInputFrame: ArrayBuffer | null = null; let socketUrlIndex = 0; const connectionUrls = socketUrls(); const pressed = new Set(); @@ -96,15 +100,14 @@ export function useRoyaleClient(): RoyaleClientView { }; const sendInput = (force = false) => { if (!engine.initialized) return; - if (!force && !inputDirty) return; - if (!force && lastSent && inputsEqual(lastSent, input)) { - inputDirty = false; - return; + inputStream.update(input); + const emission = inputStream.consume(performance.now(), force); + if (!emission) return; + if (emission.kind === "state" || !lastInputFrame) { + const packet = engine.createInput(emission.input); + lastInputFrame = protocol.encodeClient({ kind: "input", packet }); } - const packet = engine.createInput(input); - send(protocol.encodeClient({ kind: "input", packet })); - lastSent = { ...input }; - inputDirty = false; + send(lastInputFrame); }; const updateMovement = () => { input = { @@ -113,7 +116,6 @@ export function useRoyaleClient(): RoyaleClientView { strafe: Number(pressed.has("d")) - Number(pressed.has("a")), sprint: pressed.has("shift"), }; - inputDirty = true; sendInput(); }; @@ -145,8 +147,8 @@ export function useRoyaleClient(): RoyaleClientView { yaw: local?.yaw ?? 0, pitch: local?.pitch ?? 0, }; - lastSent = null; - inputDirty = true; + inputStream.reset(input); + lastInputFrame = null; validation = "waiting"; sendInput(true); break; @@ -165,7 +167,10 @@ export function useRoyaleClient(): RoyaleClientView { break; case "reject-input": engine.reject(message.sequence); + inputStream.invalidate(); + lastInputFrame = null; validation = "invalid"; + sendInput(true); break; case "event": engine.receiveEvent(message.event, message.tick); @@ -202,7 +207,6 @@ export function useRoyaleClient(): RoyaleClientView { updateMovement(); } else if (key === "r") { input = { ...input, reload: true }; - inputDirty = true; sendInput(true); } }; @@ -213,7 +217,6 @@ export function useRoyaleClient(): RoyaleClientView { updateMovement(); } else if (key === "r") { input = { ...input, reload: false }; - inputDirty = true; sendInput(true); } }; @@ -224,24 +227,20 @@ export function useRoyaleClient(): RoyaleClientView { yaw: normalizeAngle(input.yaw - event.movementX * 0.0027), pitch: clamp(input.pitch - event.movementY * 0.0022, -1.25, 1.25), }; - inputDirty = true; }; const mouseDown = (event: MouseEvent) => { if (event.button !== 0 || !document.pointerLockElement) return; input = { ...input, fire: true }; - inputDirty = true; sendInput(true); }; const mouseUp = (event: MouseEvent) => { if (event.button !== 0) return; input = { ...input, fire: false }; - inputDirty = true; sendInput(true); }; const release = () => { pressed.clear(); input = { ...input, forward: 0, strafe: 0, sprint: false, fire: false, reload: false }; - inputDirty = true; sendInput(true); }; const pointerLockChange = () => { @@ -257,8 +256,8 @@ export function useRoyaleClient(): RoyaleClientView { document.addEventListener("pointerlockchange", pointerLockChange); const inputTimer = window.setInterval(() => { - sendInput(input.fire); - }, 1_000 / royaleGame.tickRateHz); + sendInput(); + }, inputStream.policy.heartbeatIntervalMs); const pingTimer = window.setInterval(() => { send(protocol.encodeClient({ kind: "ping", @@ -302,18 +301,6 @@ export function useRoyaleClient(): RoyaleClientView { return view; } -function inputsEqual(left: RoyaleInput, right: RoyaleInput): boolean { - return ( - left.forward === right.forward && - left.strafe === right.strafe && - left.yaw === right.yaw && - left.pitch === right.pitch && - left.fire === right.fire && - left.sprint === right.sprint && - left.reload === right.reload - ); -} - function normalizeAngle(value: number): number { return Math.atan2(Math.sin(value), Math.cos(value)); } diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 2a6eb3f..e74309c 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -27,6 +27,16 @@ export { type LagCompensationPolicies, type LagCompensationResolutionContext, } from "./lag-compensation.js"; +export { + createInputStateStream, + InputStateStream, + withInputStream, + type InputStreamDefinition, + type InputStreamEmission, + type InputStreamNetworkedGame, + type InputStreamNeutralizeContext, + type InputStreamPolicy, +} from "./input-stream.js"; export { NetworkedPredictedEngine } from "./networked-client.js"; export { NetworkedAuthoritativeEngine, diff --git a/packages/engine/src/input-stream.ts b/packages/engine/src/input-stream.ts new file mode 100644 index 0000000..ee3015b --- /dev/null +++ b/packages/engine/src/input-stream.ts @@ -0,0 +1,194 @@ +import { + defineNetworkedGame, + type DefinedNetworkedGame, +} from "./define-networked-game.js"; +import type { PlayerId } from "./types.js"; + +export interface InputStreamNeutralizeContext { + playerId: PlayerId; + tick: number; + sequence: number; +} + +export interface InputStreamDefinition { + /** Stateful input refresh rate. Changes are still sent immediately. */ + heartbeatRateHz?: number; + /** Authority fails safe to a neutral command after this silence window. */ + timeoutMs?: number; + inputsEqual(left: Readonly, right: Readonly): boolean; + neutralize( + lastInput: Readonly, + context: InputStreamNeutralizeContext, + ): Input; + /** + * Rebuilds persistent state after a timed-out connection resumes. Clear + * edge-triggered actions here so a heartbeat cannot execute them twice. + */ + resume?( + lastClientInput: Readonly, + context: InputStreamNeutralizeContext, + ): Input; +} + +export interface InputStreamPolicy { + readonly heartbeatRateHz: number; + readonly heartbeatIntervalMs: number; + readonly timeoutTicks: number; + readonly timeoutMs: number; + inputsEqual(left: Readonly, right: Readonly): boolean; + cloneInput(input: Readonly): Input; + neutralize( + lastInput: Readonly, + context: InputStreamNeutralizeContext, + ): Input; + resume( + lastClientInput: Readonly, + context: InputStreamNeutralizeContext, + ): Input; +} + +export interface InputStreamEmission { + kind: "state" | "heartbeat"; + input: Input; +} + +/** + * Transport-independent latest-state sender. A changed command is available + * immediately; an unchanged command is made available only when its heartbeat + * is due. + */ +export class InputStateStream { + private current: Input | null = null; + private lastSent: Input | null = null; + private lastSentAt = Number.NEGATIVE_INFINITY; + private dirty = false; + + constructor(readonly policy: InputStreamPolicy) {} + + update(input: Readonly): void { + this.current = this.policy.cloneInput(input); + this.dirty = + this.lastSent === null || + !this.policy.inputsEqual(this.current, this.lastSent); + } + + reset(input?: Readonly): void { + this.current = input === undefined ? null : this.policy.cloneInput(input); + this.lastSent = null; + this.lastSentAt = Number.NEGATIVE_INFINITY; + this.dirty = input !== undefined; + } + + /** Forces the current value to receive a new sequence on its next send. */ + invalidate(): void { + this.lastSent = null; + this.lastSentAt = Number.NEGATIVE_INFINITY; + this.dirty = this.current !== null; + } + + /** Returns the next state to transmit, or null when no send is due. */ + consume(now: number, force = false): InputStreamEmission | null { + if (!Number.isFinite(now)) { + throw new RangeError("input stream time must be finite"); + } + if (this.current === null) return null; + + const heartbeatDue = + now - this.lastSentAt >= this.policy.heartbeatIntervalMs; + if (!force && !this.dirty && !heartbeatDue) return null; + + const kind = this.dirty || this.lastSent === null ? "state" : "heartbeat"; + const outgoing = this.policy.cloneInput(this.current); + this.lastSent = this.policy.cloneInput(this.current); + this.lastSentAt = now; + this.dirty = false; + return { kind, input: outgoing }; + } +} + +export function createInputStateStream(game: { + inputStream?: InputStreamPolicy; +}): InputStateStream { + if (!game.inputStream) { + throw new Error("Game is not configured with withInputStream()"); + } + return new InputStateStream(game.inputStream); +} + +export type InputStreamNetworkedGame< + AuthorityState, + ClientState, + Input, + AuthorityEvent, + PerceptionEvent, +> = DefinedNetworkedGame< + AuthorityState, + ClientState, + Input, + AuthorityEvent, + PerceptionEvent +> & { + readonly inputStream: InputStreamPolicy; + createInputStream(): InputStateStream; +}; + +/** Adds reliable latest-state delivery and a server-side dead-man timeout. */ +export function withInputStream< + AuthorityState, + ClientState, + Input, + AuthorityEvent, + PerceptionEvent, +>( + game: DefinedNetworkedGame< + AuthorityState, + ClientState, + Input, + AuthorityEvent, + PerceptionEvent + >, + definition: InputStreamDefinition, +): InputStreamNetworkedGame< + AuthorityState, + ClientState, + Input, + AuthorityEvent, + PerceptionEvent +> { + const heartbeatRateHz = definition.heartbeatRateHz ?? Math.min(20, game.tickRateHz); + const timeoutMs = definition.timeoutMs ?? 400; + if (!Number.isFinite(heartbeatRateHz) || heartbeatRateHz <= 0) { + throw new RangeError("heartbeatRateHz must be positive"); + } + if (heartbeatRateHz > game.tickRateHz) { + throw new RangeError("heartbeatRateHz cannot exceed the simulation tick rate"); + } + if (!Number.isFinite(timeoutMs) || timeoutMs <= 1_000 / heartbeatRateHz) { + throw new RangeError("timeoutMs must exceed one heartbeat interval"); + } + + const cloneInput = (input: Readonly): Input => { + const encoded = game.codecs.input.encode(input as Input); + return game.codecs.input.decode(encoded); + }; + const policy: InputStreamPolicy = Object.freeze({ + heartbeatRateHz, + heartbeatIntervalMs: 1_000 / heartbeatRateHz, + timeoutTicks: Math.max(1, Math.ceil((timeoutMs / 1_000) * game.tickRateHz)), + timeoutMs, + inputsEqual: definition.inputsEqual, + cloneInput, + neutralize: definition.neutralize, + resume: definition.resume ?? ((input) => cloneInput(input)), + }); + const wrapped = defineNetworkedGame({ + ...game, + inputStream: policy, + }); + + return Object.freeze({ + ...wrapped, + inputStream: policy, + createInputStream: () => new InputStateStream(policy), + }); +} diff --git a/packages/engine/src/network-clock.ts b/packages/engine/src/network-clock.ts index c8db19c..f08623d 100644 --- a/packages/engine/src/network-clock.ts +++ b/packages/engine/src/network-clock.ts @@ -71,7 +71,10 @@ export class NetworkClock { return clientNow + this.offset; } - recommendedInputLeadTicks(tickRateHz: number, maximum = 8): number { + recommendedInputLeadTicks( + tickRateHz: number, + maximum = Math.max(8, Math.ceil(tickRateHz / 2)), + ): number { const tickMilliseconds = 1_000 / tickRateHz; const latencyBudget = this.smoothedRtt / 2 + this.smoothedJitter * 2; diff --git a/packages/engine/src/networked-server.ts b/packages/engine/src/networked-server.ts index 12aae26..78775ea 100644 --- a/packages/engine/src/networked-server.ts +++ b/packages/engine/src/networked-server.ts @@ -38,6 +38,20 @@ export interface NetworkedSimulationCheckpoint< playerId: PlayerId; packet: InputPacket; }>; + inputStreamStates: Array<{ + playerId: PlayerId; + lastReceivedTick: number; + clientSequence: number; + clientTargetTick: number; + clientObservedTick: number; + clientInput: Input; + lastAppliedTick: number; + appliedSequence: number; + appliedInput: Input; + applied: boolean; + neutralized: boolean; + resumePending: boolean; + }>; pendingEvents: AuthorityEvent[]; } @@ -51,6 +65,20 @@ interface QueuedReport { report: ClientStateReport; } +interface InputStreamState { + lastReceivedTick: number; + clientSequence: number; + clientTargetTick: number; + clientObservedTick: number; + clientInput: Input; + lastAppliedTick: number; + appliedSequence: number; + appliedInput: Input; + applied: boolean; + neutralized: boolean; + resumePending: boolean; +} + export class NetworkedAuthoritativeEngine< AuthorityState, ClientState, @@ -73,6 +101,10 @@ export class NetworkedAuthoritativeEngine< private readonly lastReceivedSequence = new Map(); private readonly acknowledgedSequence = new Map(); private readonly history = new Map(); + private readonly inputStreamStates = new Map< + PlayerId, + InputStreamState + >(); private queuedInputs: QueuedInput[] = []; private queuedReports: QueuedReport[] = []; private pendingEvents: AuthorityEvent[] = []; @@ -142,6 +174,7 @@ export class NetworkedAuthoritativeEngine< }); this.lastReceivedSequence.delete(playerId); this.acknowledgedSequence.delete(playerId); + this.inputStreamStates.delete(playerId); this.queuedInputs = this.queuedInputs.filter( (queued) => queued.playerId !== playerId, ); @@ -158,6 +191,27 @@ export class NetworkedAuthoritativeEngine< } const previousSequence = this.lastReceivedSequence.get(playerId) ?? 0; + if (packet.sequence === previousSequence && this.game.inputStream) { + const stream = this.inputStreamStates.get(playerId); + if ( + !stream || + stream.clientSequence !== packet.sequence || + stream.clientTargetTick !== packet.targetTick || + stream.clientObservedTick !== (packet.observedTick ?? packet.targetTick) || + !this.game.inputStream.inputsEqual(stream.clientInput, packet.input) + ) { + return { accepted: false, reason: "duplicate" }; + } + stream.lastReceivedTick = this.currentTick; + if ( + stream.applied && + stream.neutralized && + stream.appliedSequence === stream.clientSequence + ) { + stream.resumePending = true; + } + return { accepted: true }; + } if (packet.sequence <= previousSequence) { return { accepted: false, reason: "duplicate" }; } @@ -174,6 +228,23 @@ export class NetworkedAuthoritativeEngine< } this.lastReceivedSequence.set(playerId, packet.sequence); + if (this.game.inputStream) { + const previous = this.inputStreamStates.get(playerId); + this.inputStreamStates.set(playerId, { + lastReceivedTick: this.currentTick, + clientSequence: packet.sequence, + clientTargetTick: packet.targetTick, + clientObservedTick: packet.observedTick ?? packet.targetTick, + clientInput: this.game.inputStream.cloneInput(packet.input), + lastAppliedTick: previous?.lastAppliedTick ?? this.currentTick, + appliedSequence: previous?.appliedSequence ?? packet.sequence, + appliedInput: previous?.appliedInput ?? + this.game.inputStream.cloneInput(packet.input), + applied: previous?.applied ?? false, + neutralized: previous?.neutralized ?? false, + resumePending: false, + }); + } this.queuedInputs.push({ playerId, packet }); return { accepted: true }; } @@ -234,6 +305,18 @@ export class NetworkedAuthoritativeEngine< emit, }); + if (this.game.inputStream) { + const stream = this.inputStreamStates.get(playerId); + if (stream) { + stream.lastAppliedTick = this.currentTick; + stream.appliedSequence = packet.sequence; + stream.appliedInput = this.game.inputStream.cloneInput(packet.input); + stream.applied = true; + stream.neutralized = false; + stream.resumePending = false; + } + } + const previousAck = this.acknowledgedSequence.get(playerId) ?? 0; if (packet.sequence > previousAck) { this.acknowledgedSequence.set(playerId, packet.sequence); @@ -241,6 +324,9 @@ export class NetworkedAuthoritativeEngine< } } + this.applyInputResumes(emit); + this.applyInputTimeouts(emit); + this.game.server.step(this.state, { tick: this.currentTick, deltaSeconds, @@ -344,6 +430,22 @@ export class NetworkedAuthoritativeEngine< playerId, packet: this.cloneInputPacket(packet), })), + inputStreamStates: [...this.inputStreamStates].map( + ([playerId, stream]) => ({ + playerId, + lastReceivedTick: stream.lastReceivedTick, + clientSequence: stream.clientSequence, + clientTargetTick: stream.clientTargetTick, + clientObservedTick: stream.clientObservedTick, + clientInput: this.cloneInput(stream.clientInput), + lastAppliedTick: stream.lastAppliedTick, + appliedSequence: stream.appliedSequence, + appliedInput: this.cloneInput(stream.appliedInput), + applied: stream.applied, + neutralized: stream.neutralized, + resumePending: stream.resumePending, + }), + ), pendingEvents: [...this.pendingEvents], }; } @@ -379,6 +481,22 @@ export class NetworkedAuthoritativeEngine< playerId, packet: this.cloneInputPacket(packet), })); + this.inputStreamStates.clear(); + for (const stream of checkpoint.inputStreamStates) { + this.inputStreamStates.set(stream.playerId, { + lastReceivedTick: stream.lastReceivedTick, + clientSequence: stream.clientSequence, + clientTargetTick: stream.clientTargetTick, + clientObservedTick: stream.clientObservedTick, + clientInput: this.cloneInput(stream.clientInput), + lastAppliedTick: stream.lastAppliedTick, + appliedSequence: stream.appliedSequence, + appliedInput: this.cloneInput(stream.appliedInput), + applied: stream.applied, + neutralized: stream.neutralized, + resumePending: stream.resumePending, + }); + } this.queuedReports = []; this.pendingEvents = [...checkpoint.pendingEvents]; this.history.clear(); @@ -401,15 +519,95 @@ export class NetworkedAuthoritativeEngine< } private cloneInputPacket(packet: InputPacket): InputPacket { - const encoded = this.game.codecs.input.encode(packet.input); return { sequence: packet.sequence, targetTick: packet.targetTick, observedTick: packet.observedTick ?? packet.targetTick, - input: this.game.codecs.input.decode(encoded), + input: this.cloneInput(packet.input), }; } + private cloneInput(input: Input): Input { + const encoded = this.game.codecs.input.encode(input); + return this.game.codecs.input.decode(encoded); + } + + private applyInputTimeouts(emit: (event: AuthorityEvent) => void): void { + const policy = this.game.inputStream; + if (!policy) return; + + for (const [playerId, stream] of this.inputStreamStates) { + if ( + !stream.applied || + stream.neutralized || + this.currentTick - stream.lastReceivedTick < policy.timeoutTicks + ) { + continue; + } + + const input = policy.neutralize(policy.cloneInput(stream.appliedInput), { + playerId, + tick: this.currentTick, + sequence: stream.appliedSequence, + }); + const packet: InputPacket = { + sequence: stream.appliedSequence, + targetTick: this.currentTick, + observedTick: this.currentTick, + input, + }; + if (!this.game.validateInput(input, this.inputContext(playerId, packet))) { + throw new Error("Input stream neutralizer produced an invalid input"); + } + this.game.server.applyInput(this.state, input, { + ...this.inputContext(playerId, packet), + emit, + }); + stream.appliedInput = policy.cloneInput(input); + stream.neutralized = true; + stream.resumePending = false; + } + } + + private applyInputResumes(emit: (event: AuthorityEvent) => void): void { + const policy = this.game.inputStream; + if (!policy) return; + + for (const [playerId, stream] of this.inputStreamStates) { + if ( + !stream.applied || + !stream.neutralized || + !stream.resumePending || + stream.appliedSequence !== stream.clientSequence + ) { + continue; + } + + const input = policy.resume(policy.cloneInput(stream.clientInput), { + playerId, + tick: this.currentTick, + sequence: stream.clientSequence, + }); + const packet: InputPacket = { + sequence: stream.clientSequence, + targetTick: this.currentTick, + observedTick: this.currentTick, + input, + }; + if (!this.game.validateInput(input, this.inputContext(playerId, packet))) { + throw new Error("Input stream resume produced an invalid input"); + } + this.game.server.applyInput(this.state, input, { + ...this.inputContext(playerId, packet), + emit, + }); + stream.lastAppliedTick = this.currentTick; + stream.appliedInput = policy.cloneInput(input); + stream.neutralized = false; + stream.resumePending = false; + } + } + private validateReport( playerId: PlayerId, report: ClientStateReport, diff --git a/packages/engine/src/networked-types.ts b/packages/engine/src/networked-types.ts index c1176f2..e9603f5 100644 --- a/packages/engine/src/networked-types.ts +++ b/packages/engine/src/networked-types.ts @@ -6,6 +6,7 @@ import type { PlayerId, TickContext, } from "./types.js"; +import type { InputStreamPolicy } from "./input-stream.js"; export interface EmittingTickContext extends TickContext { emit(event: Event): void; @@ -112,6 +113,8 @@ export interface NetworkedGameDefinition< PerceptionEvent >; validateInput(input: Input, context: InputContext): boolean; + /** Present when the game is wrapped with withInputStream(). */ + inputStream?: InputStreamPolicy; codecs: { input: BinaryCodec; state: BinaryCodec; diff --git a/packages/engine/src/time-travel.ts b/packages/engine/src/time-travel.ts index 46c8d70..b8fb05e 100644 --- a/packages/engine/src/time-travel.ts +++ b/packages/engine/src/time-travel.ts @@ -877,6 +877,7 @@ function createSeededEngine< client: game.client, replication: game.replication, validateInput: game.validateInput, + ...(game.inputStream ? { inputStream: game.inputStream } : {}), codecs: game.codecs, }; return new NetworkedAuthoritativeEngine(definition, serverOptions); @@ -969,6 +970,15 @@ function cloneSimulationCheckpoint< playerId, packet: cloneInputPacket(game, packet), })), + inputStreamStates: checkpoint.inputStreamStates.map((stream) => ({ + ...stream, + clientInput: game.codecs.input.decode( + game.codecs.input.encode(stream.clientInput), + ), + appliedInput: game.codecs.input.decode( + game.codecs.input.encode(stream.appliedInput), + ), + })), pendingEvents: [...checkpoint.pendingEvents], }; } diff --git a/packages/engine/test/engine.test.mjs b/packages/engine/test/engine.test.mjs index abf4c2e..67904d5 100644 --- a/packages/engine/test/engine.test.mjs +++ b/packages/engine/test/engine.test.mjs @@ -6,12 +6,14 @@ import { ReplayDivergenceError, SpatialGridIndex, createBinaryProtocol, + createInputStateStream, createJsonCodec, deterministicHash, defineGame, defineMultiplayerGame, defineNetworkedGame, withLagCompensation, + withInputStream, withReplayTransport, withSpatialReplication, withTimeTravel, @@ -270,6 +272,158 @@ test("network clock estimates RTT, offset, and input lead", () => { assert.equal(stats.clockOffset, 10); assert.equal(clock.toServerTime(300), 310); assert.equal(clock.recommendedInputLeadTicks(60), 4); + + const distantClock = new NetworkClock(); + const distantPing = distantClock.createPing(0); + distantClock.receivePong( + { + ...distantPing, + serverReceivedAt: 1_000, + serverSentAt: 1_000, + }, + 2_000, + ); + assert.equal(distantClock.recommendedInputLeadTicks(60), 30); +}); + +test("input-stream HOF heartbeats state, executes actions once, and fails safe", () => { + const codec = createJsonCodec(); + const base = defineMultiplayerGame({ + clock: { ticksPerSecond: 10, snapshotsPerSecond: 5 }, + authority: { + createInitialState: () => ({ moving: false, actionHeld: false, actions: 0 }), + cloneState: (state) => ({ ...state }), + applyInput(state, input) { + state.moving = input.move; + if (input.action && !state.actionHeld) state.actions += 1; + state.actionHeld = input.action; + }, + step() {}, + }, + prediction: { + createInitialState: () => ({ moving: false, actionHeld: false, actions: 0 }), + cloneState: (state) => ({ ...state }), + applyInput(state, input) { + state.moving = input.move; + state.actionHeld = input.action; + }, + step() {}, + }, + visibility: { + createSnapshot: (authority) => ({ ...authority }), + }, + input: { + validate: (input) => + typeof input?.move === "boolean" && typeof input.action === "boolean", + }, + encoding: { input: codec, clientState: codec }, + }); + const game = withInputStream(base, { + heartbeatRateHz: 5, + timeoutMs: 300, + inputsEqual: (left, right) => + left.move === right.move && left.action === right.action, + neutralize: () => ({ move: false, action: false }), + resume: (lastClientInput) => ({ + move: lastClientInput.move, + action: false, + }), + }); + + assert.equal(game.inputStream.timeoutTicks, 3); + const stream = createInputStateStream(game); + stream.update({ move: true, action: true }); + assert.deepEqual(stream.consume(0), { + kind: "state", + input: { move: true, action: true }, + }); + assert.equal(stream.consume(199), null); + assert.deepEqual(stream.consume(200), { + kind: "heartbeat", + input: { move: true, action: true }, + }); + stream.update({ move: false, action: false }); + assert.deepEqual(stream.consume(201), { + kind: "state", + input: { move: false, action: false }, + }); + + const server = game.createServer(); + server.addPlayer(1); + const packet = { + sequence: 1, + targetTick: 1, + observedTick: 0, + input: { move: true, action: true }, + }; + assert.deepEqual(server.submitInput(1, packet), { accepted: true }); + assert.deepEqual(server.step().acknowledgements, [ + { playerId: 1, sequence: 1 }, + ]); + assert.deepEqual(server.currentState, { + moving: true, + actionHeld: true, + actions: 1, + }); + + // The exact same sequence is a keepalive, not another action. + assert.deepEqual(server.submitInput(1, packet), { accepted: true }); + assert.deepEqual( + server.submitInput(1, { + ...packet, + input: { move: false, action: true }, + }), + { accepted: false, reason: "duplicate" }, + ); + assert.deepEqual( + server.submitInput(1, { ...packet, observedTick: 1 }), + { accepted: false, reason: "duplicate" }, + ); + server.step(); + assert.equal(server.currentState.actions, 1); + + // Silence trips the dead-man switch exactly once. + server.step(); + server.step(); + assert.deepEqual(server.currentState, { + moving: false, + actionHeld: false, + actions: 1, + }); + + // A delayed heartbeat restores state through resume(), with the action + // stripped so it still has exactly-once semantics. + assert.deepEqual(server.submitInput(1, packet), { accepted: true }); + server.step(); + assert.deepEqual(server.currentState, { + moving: true, + actionHeld: false, + actions: 1, + }); + + const checkpoint = server.createSimulationCheckpoint(); + const restored = game.createServer(); + restored.restoreSimulationCheckpoint(checkpoint); + assert.deepEqual(restored.createSimulationCheckpoint(), checkpoint); + + const recordedGame = withTimeTravel(game, { + createSeed: () => 0, + hashState: deterministicHash, + checkpointIntervalTicks: 2, + verificationIntervalTicks: 1, + }); + const recordedServer = recordedGame.createServer({ replaySeed: 0 }); + recordedServer.addPlayer(1); + recordedServer.submitInput(1, packet); + recordedServer.step(); + recordedServer.submitInput(1, packet); + recordedServer.step(); + recordedServer.step(); + recordedServer.step(); + recordedServer.submitInput(1, packet); + recordedServer.step(); + const replay = recordedGame.createReplay(recordedServer.exportRecording()); + assert.deepEqual(replay.seek(recordedServer.tick).state, recordedServer.currentState); }); test("binary protocol round-trips generic inputs, state, and control frames", () => { diff --git a/packages/shared/src/flux-game.ts b/packages/shared/src/flux-game.ts index b055e44..3223d55 100644 --- a/packages/shared/src/flux-game.ts +++ b/packages/shared/src/flux-game.ts @@ -1,6 +1,7 @@ import { createJsonCodec, defineMultiplayerGame, + withInputStream, } from "@syncer/engine"; import type { FluxAuthorityEvent, @@ -36,7 +37,7 @@ const perceptionCodec = createJsonCodec(); * a team tug-of-war: hold thrust, manage private energy, and pull the shared * core through your team's gate. */ -export const fluxGame = defineMultiplayerGame({ +const baseFluxGame = defineMultiplayerGame({ clock: { ticksPerSecond: FLUX_TICK_RATE, snapshotsPerSecond: FLUX_SNAPSHOT_RATE, @@ -220,6 +221,13 @@ export const fluxGame = defineMultiplayerGame({ }, }); +export const fluxGame = withInputStream(baseFluxGame, { + heartbeatRateHz: 20, + timeoutMs: 400, + inputsEqual: (left, right) => left.thrust === right.thrust, + neutralize: () => ({ thrust: false }), +}); + function createAuthorityState(): FluxAuthorityState { return { core: 0, diff --git a/packages/shared/src/royale-game.ts b/packages/shared/src/royale-game.ts index 1d68522..ab2a728 100644 --- a/packages/shared/src/royale-game.ts +++ b/packages/shared/src/royale-game.ts @@ -1,6 +1,7 @@ import { createJsonCodec, defineMultiplayerGame, + withInputStream, withSpatialReplication, } from "@syncer/engine"; import { @@ -290,7 +291,36 @@ const baseRoyaleGame = defineMultiplayerGame({ }, }); -export const royaleGame = withSpatialReplication(baseRoyaleGame, { +const streamedRoyaleGame = withInputStream(baseRoyaleGame, { + heartbeatRateHz: 20, + timeoutMs: 400, + inputsEqual(left, right) { + return ( + left.forward === right.forward && + left.strafe === right.strafe && + left.yaw === right.yaw && + left.pitch === right.pitch && + left.fire === right.fire && + left.sprint === right.sprint && + left.reload === right.reload + ); + }, + neutralize(lastInput) { + return { + ...lastInput, + forward: 0, + strafe: 0, + fire: false, + sprint: false, + reload: false, + }; + }, + resume(lastInput) { + return { ...lastInput, reload: false }; + }, +}); + +export const royaleGame = withSpatialReplication(streamedRoyaleGame, { cellSize: ROYALE_CHUNK_SIZE, bandwidthBudgetBytesPerSecond: 30_000, reservedBytesPerSnapshot: 300, diff --git a/packages/shared/src/shooter-game.ts b/packages/shared/src/shooter-game.ts index 1a3014f..7b33254 100644 --- a/packages/shared/src/shooter-game.ts +++ b/packages/shared/src/shooter-game.ts @@ -1,6 +1,7 @@ import { defineNetworkedGame, deterministicHash, + withInputStream, withLagCompensation, withReplayTransport, withTimeTravel, @@ -242,7 +243,41 @@ const shooterDefinition = defineNetworkedGame< }, }); -const lagCompensatedShooterDefinition = withLagCompensation(shooterDefinition, { +const streamedShooterDefinition = withInputStream(shooterDefinition, { + heartbeatRateHz: 20, + timeoutMs: 400, + inputsEqual(left, right) { + return ( + left.strafe === right.strafe && + left.forward === right.forward && + left.yaw === right.yaw && + left.pitch === right.pitch && + left.fire === right.fire && + left.sprint === right.sprint && + left.reload === right.reload && + left.weapon === right.weapon + ); + }, + neutralize(lastInput) { + return { + ...lastInput, + strafe: 0, + forward: 0, + fire: false, + sprint: false, + reload: false, + }; + }, + resume(lastInput) { + return { + ...lastInput, + fire: lastInput.fire && WEAPONS[lastInput.weapon].automatic, + reload: false, + }; + }, +}); + +const lagCompensatedShooterDefinition = withLagCompensation(streamedShooterDefinition, { historySeconds: 0.5, captureState(state) { return { diff --git a/packages/shared/test/flux.test.mjs b/packages/shared/test/flux.test.mjs index ccc2d07..4e84868 100644 --- a/packages/shared/test/flux.test.mjs +++ b/packages/shared/test/flux.test.mjs @@ -22,12 +22,16 @@ test("Flux Relay is a complete game defined through the public authoring API", ( const server = fluxGame.createServer(); server.addPlayer(1); - assert.deepEqual(server.submitInput(1, createThrustPacket(1, 1)), { + const thrustPacket = createThrustPacket(1, 1); + assert.deepEqual(server.submitInput(1, thrustPacket), { accepted: true, }); const events = []; for (let tick = 0; tick < 180; tick += 1) { + if (tick > 0 && tick % 6 === 0) { + assert.deepEqual(server.submitInput(1, thrustPacket), { accepted: true }); + } events.push(...server.step().events); }