diff --git a/Game/scripts/input_jitter_buffer.gd b/Game/scripts/input_jitter_buffer.gd new file mode 100644 index 00000000..9a0ab9be --- /dev/null +++ b/Game/scripts/input_jitter_buffer.gd @@ -0,0 +1,113 @@ +class_name InputJitterBuffer +extends RefCounted + +# Per-player server-side input state (multiplayer-todo.md §3, task 3.2). +# Deliberately a standalone RefCounted with no scene/RPC dependency — same +# reason net_codec.gd and net_interpolator.gd are pure classes — so task +# 3.5's unit tests can drive it with scripted arrival traces with no live +# match. NetworkedMatch owns one instance per connected slot and is the only +# thing that talks to the network layer; this class only knows about +# sequence numbers and ShipActions. +# +# Ring is fixed-size and slot-tagged (§3.1 step 5's "a client can never make +# the server allocate"): ingest() writes seq % RING_SIZE regardless of how +# large or malicious seq is, and consume() only ever trusts a slot whose +# stored seq exactly matches the one it expects — a stale or wrapped-around +# entry is indistinguishable from an empty one. Range/rate validation of seq +# against the current server tick is the CALLER's job (task 3.4), not this +# class's, since only the caller knows the current server tick. + +const RING_SIZE := 32 +# 500ms at 60Hz (multiplayer-todo.md §3.2's own numbers) — a duration, not a +# tick-rate-derived constant, so left as a literal rather than pulling in +# SimConstants for one number. +const STARVE_ZERO_TICKS := 30 + +var last_applied_seq := -1 # -1: consume() has never been called yet +var last_action := ShipAction.new() +var starved_ticks := 0 +var stalled := false + +var _ring_action: Array = [] +var _ring_seq: PackedInt32Array = PackedInt32Array() +# True once ingest() has ever been called for real. Consumption is a no-op +# (no starvation counted, no advancement) until then — see ingest()'s own +# comment for why an un-seeded buffer would otherwise never converge with +# what the client is actually sending. +var _seeded := false + + +func _init() -> void: + _ring_action.resize(RING_SIZE) + _ring_seq.resize(RING_SIZE) + for i in RING_SIZE: + _ring_seq[i] = -1 + + +# newest_seq/actions match NetCodec.unpack_input's own "seq"/"actions" +# fields directly: actions[i] is the action for sequence (newest_seq - i), +# newest-first. Already-consumed or stale entries are silently discarded +# (§3.1 step 5) — this is what makes redundant re-delivery of an already- +# applied tick harmless. +func ingest(newest_seq: int, actions: Array) -> void: + if not _seeded: + # The server starts calling consume() every tick the instant this + # slot exists — well before this player's first packet has had time + # to arrive (connection handshake, arena/ship spawn, first + # _physics_process tick on the client all take real time first). An + # un-seeded last_applied_seq of -1 would have consume() "expecting" + # sequence 0, 1, 2, ... via pure starvation the whole time, racing + # arbitrarily far ahead of whatever the client's own from-1 + # numbering has actually reached by the time real packets show up — + # and since both sides only ever advance monotonically with no + # resync mechanism, that gap would never close. Seed to align + # "expected" with reality the moment real data first exists. + last_applied_seq = newest_seq - actions.size() + _seeded = true + for i in actions.size(): + var seq: int = newest_seq - i + if seq <= last_applied_seq: + continue + var idx := seq % RING_SIZE + _ring_seq[idx] = seq + _ring_action[idx] = actions[i] + + +# Contiguous run of not-yet-applied entries starting right after +# last_applied_seq — reported as input_buffer_depth in every snapshot +# (§3.3) and consumed client-side by the input_lead control loop (task 3.3). +func depth() -> int: + if not _seeded or last_applied_seq < 0: + return 0 + var d := 0 + var seq := last_applied_seq + 1 + while d < RING_SIZE and _ring_seq[seq % RING_SIZE] == seq: + d += 1 + seq += 1 + return d + + +# Called once per server physics tick, before the step (§3.2). A no-op +# (returns the zero-initialized last_action, no starvation counted) until +# this player's first real packet has ever arrived — see ingest()'s comment. +func consume() -> ShipAction: + if not _seeded: + return last_action + var expected := last_applied_seq + 1 + var idx := expected % RING_SIZE + if _ring_seq[idx] == expected: + last_action = _ring_action[idx] + starved_ticks = 0 + stalled = false + else: + # Repeat-last, not zero: inputs are heavily autocorrelated at 60Hz, + # and the client already predicted with the real input either way, + # so repeating minimises expected divergence (§3.2). Only zero after + # a sustained stall, so a disconnecting player's ship doesn't fly + # into a wall at full throttle forever. + starved_ticks += 1 + if starved_ticks > STARVE_ZERO_TICKS: + last_action = ShipAction.new() + stalled = true + last_applied_seq = expected + return last_action diff --git a/Game/scripts/networked_match.gd b/Game/scripts/networked_match.gd index a744769a..24971fb0 100644 --- a/Game/scripts/networked_match.gd +++ b/Game/scripts/networked_match.gd @@ -27,6 +27,7 @@ signal score_changed(score: Dictionary) const NetCodec = preload("res://scripts/net_codec.gd") const NetBodyState = preload("res://scripts/net_body_state.gd") const NetInterpolator = preload("res://scripts/net_interpolator.gd") +const InputJitterBuffer = preload("res://scripts/input_jitter_buffer.gd") const HUD_SCENE = preload("res://scenes/HUD.tscn") # Minimum plausible interpolation delay even on a same-machine/LAN link — @@ -72,6 +73,8 @@ class SlotInfo: var spawn_index: int var ship: Ship var controller: RLShipController # server only + var jitter_buffer := InputJitterBuffer.new() # server only (§3.2) + var last_client_send_ms := 0 # server only: echoed back per-peer next snapshot (§2.4) var interpolator := NetInterpolator.new() # client only @@ -80,6 +83,11 @@ var _my_slot: SlotInfo = null # client only var _ball_interpolator := NetInterpolator.new() # client only var _local_input_sampler := PlayerShipController.new() # client only: reads local input each tick to forward; never added to a Ship, never in the tree — get_action() only touches the global Input singleton var _input_seq := 0 # client only +# Redundancy (§3.1): newest-first, capped at NetCodec.MAX_REDUNDANCY, so a +# 3-packet burst loss still recovers every tick's action via a later +# packet's history. Client only. +var _input_history: Array[ShipAction] = [] +var _last_received_snapshot_tick := 0 # client only: echoed back as ack_snapshot_tick var _reset_gen := 0 # server only: bumped on every kickoff/goal reset so the client hard-snaps instead of interpolating across the teleport # Server only. _on_goal_scored's reset_ball()/reset_ships() only QUEUE # teleports (task 0.15's queue_teleport — applied on each body's next @@ -196,11 +204,8 @@ func _start_server() -> void: func _on_input_received(peer_id: int, decoded: Dictionary) -> void: for slot in _slots: if slot.peer_id == peer_id: - var actions: Array = decoded["actions"] - # Newest-first; no redundancy handling yet (task 3.x) — just take - # the newest one every time a packet arrives. - if not actions.is_empty(): - slot.controller.action = actions[0] + slot.jitter_buffer.ingest(decoded["seq"], decoded["actions"]) + slot.last_client_send_ms = decoded["client_send_ms"] return @@ -231,12 +236,15 @@ func _broadcast_snapshot() -> void: if is_instance_valid(ball): bodies.append(_ball_to_net_body_state(ball)) var segment := NetCodec.pack_snapshot_body_segment(server_tick, 0, _reset_gen, bodies) - # Per-client header fields (last_input_seq/input_buffer_depth/echo) aren't - # tracked yet — that's the jitter-buffer work in Phase 3 (tasks 3.1-3.2). # Building the shared body segment once and reusing it per peer (rather # than re-encoding per client) is the whole reason §2.4 splits the wire # format into a per-client header + a shared body segment in the first - # place — see pack_snapshot_body_segment's own doc comment. + # place — see pack_snapshot_body_segment's own doc comment. The per- + # client header (last_input_seq/input_buffer_depth/echo_client_send_ms) + # is genuinely per-peer, built fresh below from each slot's own + # InputJitterBuffer (§3.2) — last_applied_seq of -1 (nothing consumed + # yet) encodes as 0 on the wire, which is safe: the client's own seq + # numbering starts at 1, so 0 never collides with a real seq. # "No ship is ever despawned" (§6.4) means _slots outlives a disconnect — # a real one will be handled by Phase 5's reconnect/controller-swap # logic, but sending an RPC to a peer_id ENet no longer knows about @@ -246,7 +254,9 @@ func _broadcast_snapshot() -> void: var connected_peers := multiplayer.get_peers() for slot in _slots: if connected_peers.has(slot.peer_id): - MatchSim.send_snapshot(slot.peer_id, NetCodec.pack_snapshot(0, 0, 0, segment)) + var last_input_seq := maxi(slot.jitter_buffer.last_applied_seq, 0) + var bytes := NetCodec.pack_snapshot(last_input_seq, slot.jitter_buffer.depth(), slot.last_client_send_ms, segment) + MatchSim.send_snapshot(slot.peer_id, bytes) func _ship_to_net_body_state(ship: Ship) -> NetBodyState: @@ -341,7 +351,16 @@ func _send_local_input() -> void: return # match_config hasn't arrived yet var action := _local_input_sampler.get_action().copy() _input_seq += 1 - var bytes := NetCodec.pack_input(_input_seq, 0, Time.get_ticks_msec(), [action]) + # Redundancy (§3.1): carry the last MAX_REDUNDANCY ticks' actions, + # newest-first, so a burst of up to (MAX_REDUNDANCY - 1) consecutive + # packet losses still lets the server recover every dropped tick's + # action from a later packet — InputJitterBuffer.ingest() discards + # whichever of these the server already applied, so re-sending old + # ticks every packet is harmless, not just tolerated. + _input_history.push_front(action) + if _input_history.size() > NetCodec.MAX_REDUNDANCY: + _input_history.resize(NetCodec.MAX_REDUNDANCY) + var bytes := NetCodec.pack_input(_input_seq, _last_received_snapshot_tick, Time.get_ticks_msec(), _input_history) MatchSim.send_input(bytes) @@ -349,6 +368,7 @@ func _on_snapshot_received(decoded: Dictionary) -> void: var server_tick: int = decoded["server_tick"] var reset_gen: int = decoded["reset_gen"] var bodies: Array = decoded["bodies"] + _last_received_snapshot_tick = server_tick _update_tick_bias(server_tick) for i in _slots.size(): if i < bodies.size(): @@ -418,6 +438,16 @@ func _physics_process(_delta: float) -> void: if _owns_world_simulation(): _respawn_escaped_bodies() if multiplayer.is_server(): + # Once per tick, before the step (§3.2) — RLShipController reads + # .action lazily in the ship's own _integrate_forces, which for this + # tick already ran (physics step precedes _physics_process, §9 + # gotcha 34), so this actually takes effect on the NEXT tick's step. + # That's the same one-tick input latency Phase 2 already had; this + # just replaces "read the newest packet naively" with a real + # sequence-tracked ring buffer that survives redundant/reordered/ + # lost packets. + for slot in _slots: + slot.controller.action = slot.jitter_buffer.consume() if _pending_reset_gen_bump and Engine.get_physics_frames() > _pending_reset_gen_bump_tick: _reset_gen = (_reset_gen + 1) % 256 _pending_reset_gen_bump = false diff --git a/Game/tests/cases/test_input_jitter_buffer.gd b/Game/tests/cases/test_input_jitter_buffer.gd new file mode 100644 index 00000000..79af3a4a --- /dev/null +++ b/Game/tests/cases/test_input_jitter_buffer.gd @@ -0,0 +1,112 @@ +extends "res://tests/test_case.gd" + +const InputJitterBuffer = preload("res://scripts/input_jitter_buffer.gd") +const ShipAction = preload("res://scripts/ship_action.gd") + + +func _action(thrust_z: float) -> ShipAction: + var a := ShipAction.new() + a.thrust = Vector3(0.0, 0.0, thrust_z) + return a + + +func test_sequential_ingest_and_consume() -> void: + var buf := InputJitterBuffer.new() + buf.ingest(0, [_action(0.1)]) + assert_almost_eq(buf.consume().thrust.z, 0.1, 0.0001, "tick 0") + buf.ingest(1, [_action(0.2)]) + assert_almost_eq(buf.consume().thrust.z, 0.2, 0.0001, "tick 1") + assert_eq(buf.last_applied_seq, 1, "last_applied_seq after 2 ticks") + assert_eq(buf.starved_ticks, 0, "no starvation on a clean sequential stream") + + +# §3.1's own acceptance criterion: "a 3-packet burst loss produces no +# starvation." Redundancy-4 means a single surviving packet after 3 losses +# still carries all 4 of the most recent ticks' actions. +func test_redundancy_survives_3_packet_burst_loss() -> void: + var buf := InputJitterBuffer.new() + buf.ingest(0, [_action(0.0)]) + assert_almost_eq(buf.consume().thrust.z, 0.0, 0.0001, "seq 0") + + # Packets for seq 1, 2, 3 are "lost" (never ingested individually) — only + # the seq=4 packet, carrying seq 4,3,2,1 (newest-first, redundancy 4), + # actually arrives. + buf.ingest(4, [_action(0.4), _action(0.3), _action(0.2), _action(0.1)]) + + assert_almost_eq(buf.consume().thrust.z, 0.1, 0.0001, "seq 1 recovered from redundancy") + assert_eq(buf.starved_ticks, 0, "seq 1 was not a starve") + assert_almost_eq(buf.consume().thrust.z, 0.2, 0.0001, "seq 2 recovered from redundancy") + assert_almost_eq(buf.consume().thrust.z, 0.3, 0.0001, "seq 3 recovered from redundancy") + assert_almost_eq(buf.consume().thrust.z, 0.4, 0.0001, "seq 4 recovered from redundancy") + assert_eq(buf.starved_ticks, 0, "no starvation anywhere across the whole burst-loss window") + + +func test_starvation_repeats_last_action_then_zeroes_after_500ms() -> void: + var buf := InputJitterBuffer.new() + buf.ingest(0, [_action(0.7)]) + buf.consume() + + # Nothing else ever arrives — every consume() from here on starves. + for i in InputJitterBuffer.STARVE_ZERO_TICKS: + var a := buf.consume() + assert_almost_eq(a.thrust.z, 0.7, 0.0001, "repeat-last during starve, tick %d" % i) + assert_true(not buf.stalled, "not yet stalled at tick %d" % i) + + # One more tick past STARVE_ZERO_TICKS (30 = 500ms at 60Hz) crosses the + # "> 30" threshold and zeroes rather than keeps repeating forever. + var stalled_action := buf.consume() + assert_almost_eq(stalled_action.thrust.z, 0.0, 0.0001, "zeroed after sustained stall") + assert_true(buf.stalled, "stalled flag set after 500ms of starvation") + + +func test_late_stale_packet_is_discarded_harmlessly() -> void: + var buf := InputJitterBuffer.new() + buf.ingest(5, [_action(0.5)]) + buf.consume() # seeded to 4 by ingest() (newest_seq - 1 action), one consume reaches 5 + assert_eq(buf.last_applied_seq, 5, "consumed up through seq 5") + + # A reordered/duplicated packet for an already-consumed seq arrives late. + buf.ingest(3, [_action(0.3)]) + assert_eq(buf.depth(), 0, "a stale packet below last_applied_seq must not appear as buffered depth") + + buf.ingest(6, [_action(0.6)]) + assert_almost_eq(buf.consume().thrust.z, 0.6, 0.0001, "the genuinely-next seq still consumes correctly") + + +func test_depth_reports_contiguous_buffered_run() -> void: + var buf := InputJitterBuffer.new() + buf.ingest(0, [_action(0.0)]) + buf.consume() # last_applied_seq = 0 + + assert_eq(buf.depth(), 0, "nothing buffered ahead yet") + buf.ingest(3, [_action(0.3), _action(0.2), _action(0.1)]) + assert_eq(buf.depth(), 3, "seq 1,2,3 all buffered and contiguous with last_applied_seq") + + # A gap (seq 5 arrives but seq 4 never does) caps depth at the gap, not + # the highest seq seen. + buf.ingest(5, [_action(0.5)]) + assert_eq(buf.depth(), 3, "seq 5 sits past a gap at seq 4, so it doesn't extend the contiguous run") + + +func test_ring_wraparound_does_not_confuse_a_stale_slot_with_a_fresh_one() -> void: + var buf := InputJitterBuffer.new() + buf.ingest(0, [_action(0.0)]) + buf.consume() + + # Advance last_applied_seq well past one full lap of the ring (32 + # entries) purely via starvation, with nothing re-ingested — every + # ring slot's stored seq is now far behind "expected" at each step, so + # none of them should ever be misread as valid. + for i in InputJitterBuffer.RING_SIZE * 2: + buf.consume() + assert_eq(buf.last_applied_seq, InputJitterBuffer.RING_SIZE * 2, "advanced purely by starvation") + assert_true(buf.stalled, "long starvation run ends stalled") + + # Now a fresh packet lands at the seq the ring slot for "expected" was + # LAST used for, one full lap ago — if slot-tagging didn't work, this + # would be misread as already-fresh data from the stale write. + var expected := buf.last_applied_seq + 1 + buf.ingest(expected, [_action(0.9)]) + var a := buf.consume() + assert_almost_eq(a.thrust.z, 0.9, 0.0001, "correctly reads the fresh same-slot-index seq, not a stale wraparound ghost") + assert_eq(buf.starved_ticks, 0, "starvation clears once fresh data resumes")