Netcode

Testing Netcode: Lag Compensation, Rollback, and Network Chaos

HelpMeTest

HelpMeTest

— 8 min read
Testing Netcode: Lag Compensation, Rollback, and Network Chaos

Multiplayer game bugs are among the hardest to reproduce — they live in the intersection of network timing, client prediction, and server authority. Testing netcode rigorously requires understanding the three core problems: lag compensation (server rewinds time to validate hit registration), rollback (clients re-simulate when authoritative state arrives), and determinism (all clients must produce identical simulation given identical input). This guide covers how to test each one.

The Three Netcode Testing Domains

Domain What Breaks Testing Approach
Lag Compensation Hit registration unfair at high ping Simulate rewind, assert hit validity
Rollback Visual popping, desyncs after correction Inject state deltas, measure correction quality
Determinism Silent desync, clients diverge over time Hash game state each tick, compare across simulations

Testing Lag Compensation

Lag compensation rewinds the server's world state to the moment when a remote client fired their weapon, then checks if the hit was valid in that rewound state. The critical invariant: a shot that was clearly on-target at the client's view should register, even if the target has since moved.

The Rewind Buffer

# lag_compensation.py
from dataclasses import dataclass, field
from typing import List, Optional
import copy

@dataclass
class EntitySnapshot:
    entity_id: int
    x: float
    y: float
    timestamp: float  # server timestamp in ms

@dataclass
class WorldSnapshot:
    timestamp: float
    entities: List[EntitySnapshot] = field(default_factory=list)

class LagCompensationBuffer:
    def __init__(self, max_history_ms: float = 1000.0):
        self._snapshots: List[WorldSnapshot] = []
        self._max_history_ms = max_history_ms

    def record(self, snapshot: WorldSnapshot):
        self._snapshots.append(snapshot)
        cutoff = snapshot.timestamp - self._max_history_ms
        self._snapshots = [s for s in self._snapshots if s.timestamp >= cutoff]

    def rewind_to(self, target_timestamp: float) -> Optional[WorldSnapshot]:
        """Interpolate between snapshots to reconstruct world at target_timestamp."""
        if not self._snapshots:
            return None

        # Find bracketing snapshots
        before = None
        after = None
        for snap in self._snapshots:
            if snap.timestamp <= target_timestamp:
                before = snap
            elif after is None:
                after = snap
                break

        if before is None:
            return self._snapshots[0]
        if after is None:
            return before

        # Linear interpolation
        t = (target_timestamp - before.timestamp) / (after.timestamp - before.timestamp)
        interpolated = WorldSnapshot(timestamp=target_timestamp)
        entity_map_before = {e.entity_id: e for e in before.entities}
        entity_map_after  = {e.entity_id: e for e in after.entities}

        for eid, e_before in entity_map_before.items():
            if eid in entity_map_after:
                e_after = entity_map_after[eid]
                interpolated.entities.append(EntitySnapshot(
                    entity_id=eid,
                    x=e_before.x + (e_after.x - e_before.x) * t,
                    y=e_before.y + (e_after.y - e_before.y) * t,
                    timestamp=target_timestamp
                ))
        return interpolated

    def validate_hit(self, shooter_timestamp: float, target_id: int,
                     shot_x: float, shot_y: float, hitbox_radius: float) -> bool:
        rewound = self.rewind_to(shooter_timestamp)
        if not rewound:
            return False
        for entity in rewound.entities:
            if entity.entity_id == target_id:
                dx = entity.x - shot_x
                dy = entity.y - shot_y
                return (dx*dx + dy*dy) <= hitbox_radius * hitbox_radius
        return False
# test_lag_compensation.py
import pytest
from lag_compensation import LagCompensationBuffer, WorldSnapshot, EntitySnapshot

@pytest.fixture
def buffer():
    buf = LagCompensationBuffer(max_history_ms=500.0)
    # Enemy moves from x=100 to x=200 over 100ms
    buf.record(WorldSnapshot(timestamp=1000.0, entities=[
        EntitySnapshot(entity_id=1, x=100.0, y=50.0, timestamp=1000.0)
    ]))
    buf.record(WorldSnapshot(timestamp=1100.0, entities=[
        EntitySnapshot(entity_id=1, x=200.0, y=50.0, timestamp=1100.0)
    ]))
    return buf

def test_hit_valid_at_rewound_position(buffer):
    """Shot at x=100 at t=1000 should hit even though enemy is now at x=200."""
    # Client fired at t=1000, server receives at t=1100 (100ms lag)
    hit = buffer.validate_hit(
        shooter_timestamp=1000.0,
        target_id=1,
        shot_x=100.0, shot_y=50.0,
        hitbox_radius=20.0
    )
    assert hit, "Shot that was on-target at fire time should register"

def test_miss_at_rewound_position(buffer):
    """Shot clearly off target at fire time should not register."""
    hit = buffer.validate_hit(
        shooter_timestamp=1000.0,
        target_id=1,
        shot_x=300.0, shot_y=50.0,
        hitbox_radius=20.0
    )
    assert not hit

def test_interpolated_position_at_midpoint(buffer):
    """At t=1050, enemy should be at x=150."""
    hit = buffer.validate_hit(
        shooter_timestamp=1050.0,
        target_id=1,
        shot_x=150.0, shot_y=50.0,
        hitbox_radius=5.0
    )
    assert hit

def test_history_pruned_beyond_max(buffer):
    """Snapshots older than max_history_ms should be pruned."""
    buffer.record(WorldSnapshot(timestamp=2000.0, entities=[]))
    result = buffer.rewind_to(500.0)  # before all our history
    assert result is not None  # returns oldest available, not None

Testing Rollback Netcode

GGPO-style rollback means: predict locally, receive authoritative state from server, if mismatch then rollback to last confirmed state and re-simulate forward. The invariants to test:

  1. After rollback and re-simulation, client state matches server state
  2. Re-simulation produces the same output as original simulation for unchanged inputs
  3. Visual correction magnitude is bounded (large corrections indicate a bug)
# rollback_sim.py
from dataclasses import dataclass, field
from typing import Dict, List, Tuple
import copy

@dataclass
class GameState:
    tick: int
    player_x: float
    player_y: float
    player_vx: float = 0.0
    player_vy: float = 0.0

    def __eq__(self, other):
        return (self.player_x == other.player_x and
                self.player_y == other.player_y)

@dataclass
class Input:
    tick: int
    dx: float  # -1, 0, 1
    dy: float
    jump: bool = False

MOVE_SPEED = 5.0
JUMP_VELOCITY = -15.0
GRAVITY = 0.5

def simulate_tick(state: GameState, inp: Input) -> GameState:
    new = copy.copy(state)
    new.tick = state.tick + 1
    new.player_x += inp.dx * MOVE_SPEED
    new.player_vy += GRAVITY
    if inp.jump and state.player_y >= 0:
        new.player_vy = JUMP_VELOCITY
    new.player_y = max(0.0, state.player_y + new.player_vy)
    if new.player_y == 0:
        new.player_vy = 0
    return new

class RollbackClient:
    def __init__(self, initial_state: GameState):
        self._confirmed_state: GameState = initial_state
        self._confirmed_tick: int = initial_state.tick
        self._predicted_states: Dict[int, GameState] = {}
        self._inputs: Dict[int, Input] = {}

    def predict(self, inp: Input) -> GameState:
        prev = self._predicted_states.get(inp.tick - 1, self._confirmed_state)
        predicted = simulate_tick(prev, inp)
        self._predicted_states[inp.tick] = predicted
        self._inputs[inp.tick] = inp
        return predicted

    def receive_authoritative(self, auth_state: GameState) -> Tuple[GameState, float]:
        """Returns (final_state, correction_distance)."""
        predicted = self._predicted_states.get(auth_state.tick)
        correction = 0.0
        if predicted:
            dx = auth_state.player_x - predicted.player_x
            dy = auth_state.player_y - predicted.player_y
            correction = (dx*dx + dy*dy) ** 0.5

        # Rollback: re-simulate from confirmed state
        self._confirmed_state = auth_state
        self._confirmed_tick = auth_state.tick

        state = auth_state
        current_tick = auth_state.tick + 1
        max_tick = max(self._inputs.keys()) if self._inputs else auth_state.tick

        while current_tick <= max_tick:
            inp = self._inputs.get(current_tick)
            if inp:
                state = simulate_tick(state, inp)
                self._predicted_states[current_tick] = state
            current_tick += 1

        return state, correction
# test_rollback.py
import pytest
from rollback_sim import GameState, Input, RollbackClient, simulate_tick

def test_rollback_converges_to_server_state():
    """After receiving authoritative state, client should match server."""
    initial = GameState(tick=0, player_x=0.0, player_y=0.0)
    client = RollbackClient(initial)

    # Client predicts ticks 1-5
    for tick in range(1, 6):
        client.predict(Input(tick=tick, dx=1.0, dy=0.0))

    # Server corrects tick 3 (slightly different due to lag)
    server_state_at_3 = GameState(tick=3, player_x=14.9, player_y=0.0)  # tiny drift
    final_state, correction = client.receive_authoritative(server_state_at_3)

    # Final state should be re-simulated from server's tick 3
    assert final_state.tick == 5
    # x should be server_x + 2 more ticks of movement (14.9 + 5 + 5 = 24.9)
    assert abs(final_state.player_x - 24.9) < 0.001

def test_correction_magnitude_bounded():
    """Correction should be small for minor prediction errors."""
    initial = GameState(tick=0, player_x=0.0, player_y=0.0)
    client = RollbackClient(initial)

    client.predict(Input(tick=1, dx=1.0, dy=0.0))

    # Tiny server correction
    server_state = GameState(tick=1, player_x=5.01, player_y=0.0)
    _, correction = client.receive_authoritative(server_state)

    assert correction < 1.0, f"Correction {correction} too large for minor prediction error"

def test_determinism_identical_inputs():
    """Same initial state + same inputs = identical result, always."""
    initial = GameState(tick=0, player_x=50.0, player_y=0.0)
    inputs = [Input(tick=i, dx=1.0 if i % 2 == 0 else -1.0, dy=0.0) for i in range(1, 21)]

    state_a = initial
    for inp in inputs:
        state_a = simulate_tick(state_a, inp)

    state_b = initial
    for inp in inputs:
        state_b = simulate_tick(state_b, inp)

    assert state_a == state_b, "Simulation must be deterministic"

Network Chaos Injection

Real network problems — packet loss, reordering, duplication, jitter — surface bugs that clean-network tests miss. Build a chaos proxy into your test environment:

# network_chaos.py
import random
import time
from dataclasses import dataclass, field
from typing import List, Callable, Any

@dataclass
class ChaosConfig:
    packet_loss_rate: float = 0.0      # 0.0 - 1.0
    min_latency_ms: float = 0.0
    max_latency_ms: float = 0.0
    jitter_ms: float = 0.0
    duplicate_rate: float = 0.0
    reorder_rate: float = 0.0

@dataclass
class PendingPacket:
    data: Any
    deliver_at: float

class NetworkChaosProxy:
    def __init__(self, config: ChaosConfig, deliver_fn: Callable):
        self._config = config
        self._deliver = deliver_fn
        self._pending: List[PendingPacket] = []

    def send(self, data: Any):
        if random.random() < self._config.packet_loss_rate:
            return  # dropped

        delay = (self._config.min_latency_ms +
                 random.uniform(0, self._config.jitter_ms)) / 1000.0
        deliver_at = time.monotonic() + delay

        if random.random() < self._config.reorder_rate and self._pending:
            # Swap with a random pending packet
            idx = random.randrange(len(self._pending))
            deliver_at = self._pending[idx].deliver_at

        self._pending.append(PendingPacket(data=data, deliver_at=deliver_at))

        if random.random() < self._config.duplicate_rate:
            extra_delay = delay + random.uniform(0.01, 0.1)
            self._pending.append(PendingPacket(
                data=data,
                deliver_at=time.monotonic() + extra_delay
            ))

    def tick(self):
        now = time.monotonic()
        delivered = [p for p in self._pending if p.deliver_at <= now]
        self._pending = [p for p in self._pending if p.deliver_at > now]
        for packet in delivered:
            self._deliver(packet.data)
# test_network_chaos.py
import pytest
import time
from network_chaos import NetworkChaosProxy, ChaosConfig

def test_30_percent_packet_loss():
    received = []
    config = ChaosConfig(packet_loss_rate=0.30)
    proxy = NetworkChaosProxy(config, received.append)

    for i in range(1000):
        proxy.send(i)
        proxy.tick()

    # With 30% loss, expect 60-80% delivery (statistical)
    delivery_rate = len(received) / 1000
    assert 0.60 <= delivery_rate <= 0.80, \
        f"Expected ~70% delivery with 30% loss, got {delivery_rate:.1%}"

def test_high_latency_delays_delivery():
    received = []
    config = ChaosConfig(min_latency_ms=200, max_latency_ms=200)
    proxy = NetworkChaosProxy(config, received.append)

    proxy.send("message")
    proxy.tick()
    assert len(received) == 0, "Message should not arrive immediately"

    time.sleep(0.25)
    proxy.tick()
    assert len(received) == 1, "Message should arrive after 200ms"

def test_game_state_eventually_consistent_under_chaos():
    """Even with 20% packet loss, game state should eventually converge."""
    from rollback_sim import GameState, Input, RollbackClient, simulate_tick

    initial = GameState(tick=0, player_x=0.0, player_y=0.0)
    client = RollbackClient(initial)
    server_state = initial

    server_messages = []
    config = ChaosConfig(packet_loss_rate=0.20, min_latency_ms=50, max_latency_ms=150)
    proxy = NetworkChaosProxy(config, server_messages.append)

    # Simulate 20 ticks
    for tick in range(1, 21):
        inp = Input(tick=tick, dx=1.0, dy=0.0)
        client.predict(inp)
        server_state = simulate_tick(server_state, inp)
        proxy.send(server_state)
        proxy.tick()

    # Flush all pending
    time.sleep(0.2)
    proxy.tick()

    # Apply all received server states
    final_state = None
    for msg in server_messages:
        final_state, _ = client.receive_authoritative(msg)

    if final_state:
        # Final state should be within reasonable bounds of server state
        assert abs(final_state.player_x - server_state.player_x) < 10.0

Determinism Hash Testing

The most critical invariant for any networked game: given the same inputs, two independent simulations must produce bit-identical state. Test this continuously:

# test_determinism.py
import hashlib, json, copy
from rollback_sim import GameState, Input, simulate_tick

def state_hash(state: GameState) -> str:
    data = json.dumps({
        "tick": state.tick,
        "x": round(state.player_x, 6),
        "y": round(state.player_y, 6),
        "vx": round(state.player_vx, 6),
        "vy": round(state.player_vy, 6),
    }, sort_keys=True)
    return hashlib.sha256(data.encode()).hexdigest()[:16]

def test_two_clients_produce_identical_hashes():
    initial = GameState(tick=0, player_x=100.0, player_y=0.0)
    inputs = [
        Input(tick=1, dx=1.0, dy=0.0),
        Input(tick=2, dx=1.0, dy=0.0, jump=True),
        Input(tick=3, dx=0.0, dy=0.0),
        Input(tick=4, dx=-1.0, dy=0.0),
        Input(tick=5, dx=-1.0, dy=0.0),
    ]

    # Simulate on "client A"
    state_a = copy.copy(initial)
    hashes_a = []
    for inp in inputs:
        state_a = simulate_tick(state_a, inp)
        hashes_a.append(state_hash(state_a))

    # Simulate on "client B" (independently)
    state_b = copy.copy(initial)
    hashes_b = []
    for inp in inputs:
        state_b = simulate_tick(state_b, inp)
        hashes_b.append(state_hash(state_b))

    for tick, (ha, hb) in enumerate(zip(hashes_a, hashes_b), 1):
        assert ha == hb, f"Desync at tick {tick}: A={ha} B={hb}"

CI Integration

Run all netcode tests in a short-lived CI job with controlled random seeds for reproducibility:

name: Netcode Tests
on: [push, pull_request]

jobs:
  netcode:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with: { python-version: '3.11' }
      - run: pip install pytest pytest-cov
      - name: Run netcode tests with fixed seed
        run: |
          PYTHONHASHSEED=42 pytest tests/netcode/ \
            --cov=src/netcode \
            --cov-report=xml \
            -v
      - uses: codecov/codecov-action@v3

Setting PYTHONHASHSEED=42 (or equivalent in your language) ensures chaos tests with random elements are reproducible across CI runs, while still exercising probabilistic code paths.

Conclusion

Testing netcode means testing time. Lag compensation tests validate that historical state can be reconstructed accurately. Rollback tests verify that prediction errors are corrected without corrupting ongoing simulation. Determinism tests are your canary — a single hash mismatch at any tick means your simulation is broken in a way that will eventually cause a desync in production. Build these three test layers before any networked feature ships, and you'll catch the class of bugs that otherwise only appear at 150ms ping in a tournament final.

Read more