Server-Authoritative Tick System

Server-side tick system with lag compensation for fair multiplayer gameplay.

When to Use This Skill

• Building real-time multiplayer games

• Need to prevent cheating (speed hacks, teleports, aimbots)

• Want fair hit detection despite network latency

• Require deterministic physics simulation

Core Concepts

Client-authoritative multiplayer is trivially exploitable. Server-authoritative feels laggy without lag compensation. The solution:

• Fixed tick rate (60Hz) for deterministic physics

• Input validation with violation tracking and decay

• Position history for lag compensation (200ms window)

• Reduced broadcast rate (20Hz) to save bandwidth

Implementation

Python

from dataclasses import dataclass, field from collections import deque from typing import Dict, Optional, Tuple, Callable, Awaitable from enum import Enum import asyncio import time import math

class ViolationType(str, Enum): SPEED_HACK = "speed_hack" TELEPORT = "teleport" INVALID_ACTION = "invalid_action"

@dataclass(frozen=True) class TickConfig: rate_hz: int = 60 broadcast_divisor: int = 3 # 60/3 = 20Hz broadcast max_speed: float = 300.0 # Units per second teleport_threshold: float = 100.0 max_rewind_ms: int = 200 violation_threshold: int = 10 decay_per_second: float = 1.0

@property
def interval_ms(self) -> float:
    return 1000.0 / self.rate_hz

def max_distance_per_tick(self) -> float:
    return self.max_speed / self.rate_hz

@dataclass class PlayerInput: player_id: str tick: int dx: float = 0.0 dy: float = 0.0 timestamp_ms: float = 0.0

@dataclass class PositionSnapshot: x: float y: float tick: int timestamp_ms: float

@dataclass class PlayerState: player_id: str x: float = 0.0 y: float = 0.0 health: int = 100 last_valid_position: Tuple[float, float] = (0.0, 0.0) violations: float = 0.0 position_history: deque = field(default_factory=lambda: deque(maxlen=15))

def record_position(self, tick: int, timestamp_ms: float) -> None:
    self.position_history.append(PositionSnapshot(
        x=self.x, y=self.y, tick=tick, timestamp_ms=timestamp_ms
    ))

@dataclass class GameState: game_id: str tick: int = 0 players: Dict[str, PlayerState] = field(default_factory=dict) running: bool = False start_time_ms: float = 0.0

class InputValidator: """Validates player inputs for anti-cheat."""

def __init__(self, config: TickConfig):
    self._config = config

def validate_movement(
    self, player: PlayerState, input: PlayerInput
) -> Tuple[bool, Optional[ViolationType]]:
    distance = math.sqrt(input.dx ** 2 + input.dy ** 2)
    max_distance = self._config.max_distance_per_tick()
    
    # 50% tolerance for network jitter
    if distance > max_distance * 1.5:
        return False, ViolationType.SPEED_HACK
    
    # Check for teleport
    new_x = player.x + input.dx
    new_y = player.y + input.dy
    last_x, last_y = player.last_valid_position
    jump_distance = math.sqrt((new_x - last_x) ** 2 + (new_y - last_y) ** 2)
    
    if jump_distance > self._config.teleport_threshold:
        return False, ViolationType.TELEPORT
    
    return True, None

def apply_violation(self, player: PlayerState) -> Tuple[bool, bool]:
    player.violations += 1.0
    should_warn = player.violations >= self._config.violation_threshold * 0.5
    should_kick = player.violations >= self._config.violation_threshold
    return should_warn, should_kick

def decay_violations(self, player: PlayerState, delta_seconds: float) -> None:
    decay = self._config.decay_per_second * delta_seconds
    player.violations = max(0.0, player.violations - decay)

class LagCompensator: """Lag compensation for fair hit detection."""

def __init__(self, config: TickConfig):
    self._config = config

def get_position_at_time(
    self, player: PlayerState, target_time_ms: float, current_time_ms: float
) -> Tuple[float, float]:
    # Clamp rewind to max window
    rewind_ms = current_time_ms - target_time_ms
    if rewind_ms > self._config.max_rewind_ms:
        target_time_ms = current_time_ms - self._config.max_rewind_ms
    
    if not player.position_history:
        return player.x, player.y
    
    # Find surrounding snapshots
    before: Optional[PositionSnapshot] = None
    after: Optional[PositionSnapshot] = None
    
    for snapshot in player.position_history:
        if snapshot.timestamp_ms <= target_time_ms:
            before = snapshot
        elif after is None:
            after = snapshot
            break
    
    if before is None:
        oldest = player.position_history[0]
        return oldest.x, oldest.y
    
    if after is None:
        return before.x, before.y
    
    # Interpolate between snapshots
    time_range = after.timestamp_ms - before.timestamp_ms
    if time_range <= 0:
        return before.x, before.y
    
    t = max(0.0, min(1.0, (target_time_ms - before.timestamp_ms) / time_range))
    x = before.x + (after.x - before.x) * t
    y = before.y + (after.y - before.y) * t
    
    return x, y

def check_hit(
    self, shooter: PlayerState, target: PlayerState,
    shot_time_ms: float, current_time_ms: float, hit_radius: float = 20.0
) -> Tuple[bool, Tuple[float, float]]:
    target_x, target_y = self.get_position_at_time(target, shot_time_ms, current_time_ms)
    distance = math.sqrt((shooter.x - target_x) ** 2 + (shooter.y - target_y) ** 2)
    return distance <= hit_radius, (target_x, target_y)

class TickSystem: """Server-authoritative tick system."""

def __init__(self, config: TickConfig = None):
    self._config = config or TickConfig()
    self._games: Dict[str, GameState] = {}
    self._tasks: Dict[str, asyncio.Task] = {}
    self._validator = InputValidator(self._config)
    self._lag_comp = LagCompensator(self._config)
    self._broadcast_callback: Optional[Callable[[str, dict], Awaitable[None]]] = None
    self._kick_callback: Optional[Callable[[str, str, str], Awaitable[None]]] = None

def set_broadcast_callback(self, callback: Callable[[str, dict], Awaitable[None]]) -> None:
    self._broadcast_callback = callback

def set_kick_callback(self, callback: Callable[[str, str, str], Awaitable[None]]) -> None:
    self._kick_callback = callback

def create_game(
    self, game_id: str, player1_id: str, player2_id: str,
    spawn1: Tuple[float, float] = (100, 300),
    spawn2: Tuple[float, float] = (700, 300)
) -> GameState:
    game = GameState(game_id=game_id)
    game.players[player1_id] = PlayerState(
        player_id=player1_id, x=spawn1[0], y=spawn1[1], last_valid_position=spawn1
    )
    game.players[player2_id] = PlayerState(
        player_id=player2_id, x=spawn2[0], y=spawn2[1], last_valid_position=spawn2
    )
    self._games[game_id] = game
    return game

async def start_game(self, game_id: str) -> None:
    game = self._games.get(game_id)
    if not game:
        raise ValueError(f"Game {game_id} not found")
    
    game.running = True
    game.start_time_ms = time.time() * 1000
    self._tasks[game_id] = asyncio.create_task(self._tick_loop(game_id))

async def stop_game(self, game_id: str) -> None:
    game = self._games.get(game_id)
    if game:
        game.running = False
    
    task = self._tasks.pop(game_id, None)
    if task:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
    
    self._games.pop(game_id, None)

async def process_input(self, game_id: str, input: PlayerInput) -> bool:
    game = self._games.get(game_id)
    if not game or not game.running:
        return False
    
    player = game.players.get(input.player_id)
    if not player:
        return False
    
    valid, violation = self._validator.validate_movement(player, input)
    
    if not valid and violation:
        _, should_kick = self._validator.apply_violation(player)
        if should_kick and self._kick_callback:
            await self._kick_callback(game_id, input.player_id, f"Anti-cheat: {violation.value}")
        return False
    
    player.x += input.dx
    player.y += input.dy
    player.last_valid_position = (player.x, player.y)
    return True

async def _tick_loop(self, game_id: str) -> None:
    game = self._games.get(game_id)
    if not game:
        return
    
    interval = self._config.interval_ms / 1000.0
    last_time = time.time()
    
    while game.running:
        tick_start = time.time()
        delta = tick_start - last_time
        last_time = tick_start
        
        await self._tick(game, delta)
        
        if game.tick % self._config.broadcast_divisor == 0:
            await self._broadcast_state(game)
        
        game.tick += 1
        
        elapsed = time.time() - tick_start
        sleep_time = max(0, interval - elapsed)
        if sleep_time > 0:
            await asyncio.sleep(sleep_time)

async def _tick(self, game: GameState, delta: float) -> None:
    current_time_ms = time.time() * 1000
    for player in game.players.values():
        self._validator.decay_violations(player, delta)
        player.record_position(game.tick, current_time_ms)

async def _broadcast_state(self, game: GameState) -> None:
    if not self._broadcast_callback:
        return
    
    state = {
        "type": "game_state",
        "tick": game.tick,
        "players": {
            pid: {"x": p.x, "y": p.y, "health": p.health}
            for pid, p in game.players.items()
        }
    }
    await self._broadcast_callback(game.game_id, state)

TypeScript

enum ViolationType { SPEED_HACK = 'speed_hack', TELEPORT = 'teleport', INVALID_ACTION = 'invalid_action', }

interface TickConfig { rateHz: number; broadcastDivisor: number; maxSpeed: number; teleportThreshold: number; maxRewindMs: number; violationThreshold: number; decayPerSecond: number; }

const DEFAULT_CONFIG: TickConfig = { rateHz: 60, broadcastDivisor: 3, maxSpeed: 300, teleportThreshold: 100, maxRewindMs: 200, violationThreshold: 10, decayPerSecond: 1.0, };

interface PositionSnapshot { x: number; y: number; tick: number; timestampMs: number; }

interface PlayerState { playerId: string; x: number; y: number; health: number; lastValidPosition: [number, number]; violations: number; positionHistory: PositionSnapshot[]; }

class TickSystem { private config: TickConfig; private games = new Map<string, GameState>(); private intervals = new Map<string, NodeJS.Timeout>();

constructor(config: Partial<TickConfig> = {}) { this.config = { ...DEFAULT_CONFIG, ...config }; }

createGame(gameId: string, player1Id: string, player2Id: string): GameState { const game: GameState = { gameId, tick: 0, players: new Map([ [player1Id, this.createPlayer(player1Id, 100, 300)], [player2Id, this.createPlayer(player2Id, 700, 300)], ]), running: false, }; this.games.set(gameId, game); return game; }

private createPlayer(id: string, x: number, y: number): PlayerState { return { playerId: id, x, y, health: 100, lastValidPosition: [x, y], violations: 0, positionHistory: [], }; }

processInput(gameId: string, input: PlayerInput): boolean { const game = this.games.get(gameId); if (!game?.running) return false;

const player = game.players.get(input.playerId);
if (!player) return false;

const distance = Math.sqrt(input.dx ** 2 + input.dy ** 2);
const maxDistance = this.config.maxSpeed / this.config.rateHz;

if (distance > maxDistance * 1.5) {
  player.violations += 1;
  return false;
}

player.x += input.dx;
player.y += input.dy;
player.lastValidPosition = [player.x, player.y];
return true;

}

getPositionAtTime(player: PlayerState, targetTimeMs: number, currentTimeMs: number): [number, number] { const rewindMs = Math.min(currentTimeMs - targetTimeMs, this.config.maxRewindMs); const clampedTargetTime = currentTimeMs - rewindMs;

if (player.positionHistory.length === 0) {
  return [player.x, player.y];
}

let before: PositionSnapshot | null = null;
let after: PositionSnapshot | null = null;

for (const snapshot of player.positionHistory) {
  if (snapshot.timestampMs <= clampedTargetTime) {
    before = snapshot;
  } else if (!after) {
    after = snapshot;
    break;
  }
}

if (!before) return [player.positionHistory[0].x, player.positionHistory[0].y];
if (!after) return [before.x, before.y];

const t = Math.max(0, Math.min(1, 
  (clampedTargetTime - before.timestampMs) / (after.timestampMs - before.timestampMs)
));

return [
  before.x + (after.x - before.x) * t,
  before.y + (after.y - before.y) * t,
];

} }

Usage Examples

Game Setup

tick_system = TickSystem()

Create game with two players

game = tick_system.create_game("match-123", "player-1", "player-2")

Set up callbacks

tick_system.set_broadcast_callback(broadcast_to_players) tick_system.set_kick_callback(kick_player)

Start the game loop

await tick_system.start_game("match-123")

Processing Player Input

When receiving input from client

input = PlayerInput( player_id="player-1", tick=game.tick, dx=5.0, dy=2.0, timestamp_ms=client_timestamp )

valid = await tick_system.process_input("match-123", input) if not valid: # Input was rejected (possible cheat attempt) pass

Lag-Compensated Hit Detection

When player fires a shot

shooter = game.players["player-1"] target = game.players["player-2"]

hit, target_pos = tick_system._lag_comp.check_hit( shooter=shooter, target=target, shot_time_ms=client_shot_timestamp, current_time_ms=time.time() * 1000, hit_radius=20.0 )

if hit: target.health -= 25

Best Practices

• Tune tick rate for your game - 30-60Hz is typical for action games

• Set broadcast rate lower than tick rate to save bandwidth

• Keep lag compensation window reasonable (100-200ms)

• Use violation decay to forgive occasional network hiccups

• Log all kicks with full context for review

• Test with simulated high latency

Common Mistakes

• Running physics at variable rate (causes non-determinism)

• No tolerance for network jitter (false positives)

• Lag compensation window too large (feels unfair to targets)

• Not decaying violations (kicks legitimate players)

• Broadcasting every tick (wastes bandwidth)

• Trusting client timestamps without bounds checking

Related Patterns

• websocket-management - Connection handling for game clients

• atomic-matchmaking - Creating matches before game starts

• graceful-shutdown - Draining games on server shutdown