add generic stateful input streaming

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,

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<Input> {
+  /** 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<Input>, right: Readonly<Input>): boolean;
+  neutralize(
+    lastInput: Readonly<Input>,
+    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<Input>,
+    context: InputStreamNeutralizeContext,
+  ): Input;
+}
+
+export interface InputStreamPolicy<Input> {
+  readonly heartbeatRateHz: number;
+  readonly heartbeatIntervalMs: number;
+  readonly timeoutTicks: number;
+  readonly timeoutMs: number;
+  inputsEqual(left: Readonly<Input>, right: Readonly<Input>): boolean;
+  cloneInput(input: Readonly<Input>): Input;
+  neutralize(
+    lastInput: Readonly<Input>,
+    context: InputStreamNeutralizeContext,
+  ): Input;
+  resume(
+    lastClientInput: Readonly<Input>,
+    context: InputStreamNeutralizeContext,
+  ): Input;
+}
+
+export interface InputStreamEmission<Input> {
+  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<Input> {
+  private current: Input | null = null;
+  private lastSent: Input | null = null;
+  private lastSentAt = Number.NEGATIVE_INFINITY;
+  private dirty = false;
+
+  constructor(readonly policy: InputStreamPolicy<Input>) {}
+
+  update(input: Readonly<Input>): void {
+    this.current = this.policy.cloneInput(input);
+    this.dirty =
+      this.lastSent === null ||
+      !this.policy.inputsEqual(this.current, this.lastSent);
+  }
+
+  reset(input?: Readonly<Input>): 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<Input> | 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<Input>(game: {
+  inputStream?: InputStreamPolicy<Input>;
+}): InputStateStream<Input> {
+  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<Input>;
+  createInputStream(): InputStateStream<Input>;
+};
+
+/** 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<Input>,
+): 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>): Input => {
+    const encoded = game.codecs.input.encode(input as Input);
+    return game.codecs.input.decode(encoded);
+  };
+  const policy: InputStreamPolicy<Input> = 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),
+  });
+}

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;
 

packages/engine/src/networked-server.ts

@@ -38,6 +38,20 @@ export interface NetworkedSimulationCheckpoint<
     playerId: PlayerId;
     packet: InputPacket<Input>;
   }>;
+  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<ClientState> {
   report: ClientStateReport<ClientState>;
 }
 
+interface InputStreamState<Input> {
+  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<PlayerId, number>();
   private readonly acknowledgedSequence = new Map<PlayerId, number>();
   private readonly history = new Map<number, AuthorityState>();
+  private readonly inputStreamStates = new Map<
+    PlayerId,
+    InputStreamState<Input>
+  >();
   private queuedInputs: QueuedInput<Input>[] = [];
   private queuedReports: QueuedReport<ClientState>[] = [];
   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<Input>): InputPacket<Input> {
-    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<Input> = {
+        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<Input> = {
+        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<ClientState>,

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<Event> 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<Input>;
   codecs: {
     input: BinaryCodec<Input>;
     state: BinaryCodec<ClientState>;

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],
   };
 }