fix(multiplayer): server no longer rate-limits a backlog it caused itself

Closes task 5.10's three recording gaps, and the gap-closing found a real
input-loss bug.

Replay log: a failed write now ends the log permanently instead of
desyncing every later record's framing; close() is called from _exit_tree
with a summary, since the RefCounted destructor closes it implicitly but
never says whether the log is complete; rejected packets are recorded
with their reason in the kind byte (framing unchanged, FORMAT_VERSION 2
so "no rejects" differs from "this build never recorded them"). Recording
is capped at 8 per peer per window - uncapped, the diagnostic is a remote
disk-fill amplifier, since the attacker picks the packet rate. Uncapped
totals live on MatchSim and survive the peer's disconnect.

The bug: a 2s host stall has the client sending at 60Hz throughout, and
ENet delivers that whole backlog in the first window after resume - 70 of
an honest client's packets rejected as "rate limit exceeded". Redundancy
does not cover it, because the dropped packets are contiguous: 0 of 70
rescued, and 82 of 923 sequences (8.88%, ~1.4s of input) never reached
the server, against 0.00% with no stall. Every prediction gate passed.

Fixed by granting each already-tracked peer a capped, two-window packet
grace when the server detects its own wall-clock stall. Rate-limit
rejects 70 -> 0, sequences missing 8.88% -> 0.00%, seq-guard rejects
9 -> 0. Controls on the unfixed build lost 4.34/7.52/7.86%. All three
abuse roles still disconnect and no flood induced a stall, so the grace
cannot be farmed.

Also corrects an earlier wrong conclusion: the reviewer's free-flight
p95 0.688 is real and reproduces on two processes with 0.0% snapshot
loss. The plain --role=client drive fails the 0.5 free-flight bound in
3 of 8 runs because that drive is mostly a contact test - the harness
comment already said so - leaving a cohort as small as 12 samples.
Near-surface error is genuinely several times open-air error, so the
calibrated bound now belongs to --exercise-free-flight alone and the
plain role asserts the always-well-sampled all-cohort percentiles at
1.2/2.0, printing the free-flight numbers as reported-not-asserted.
6/6 plain runs pass where 3/7 failed; tightening to 0.3 still fails.

tools/replay_dump.gd reads a log back: counts by kind, plus how much of
the input sequence stream reached the server once redundancy is counted.

Game/scripts/match_sim.gd

@@ -21,6 +21,13 @@ const NetCodec = preload("res://scripts/net_codec.gd")
 
 signal match_config_received(arena_path: String, peer_ids: PackedInt32Array, teams: PackedInt32Array, spawn_indices: PackedInt32Array)
 signal input_received(peer_id: int, decoded: Dictionary)    # decoded: see NetCodec.unpack_input
+# Task 5.10. A packet this autoload dropped before it could ever reach a match,
+# with the verbatim bytes — the replay log's whole reason to exist is the field
+# report "my input did nothing", and an accepted-input-only log has thrown away
+# exactly the evidence that would explain it. `reason` is an InputRejectReason;
+# the transport layer deliberately does not know about the replay format's own
+# record kinds, so the mapping lives at the listener.
+signal input_rejected(peer_id: int, reason: int, bytes: PackedByteArray)
 signal snapshot_received(decoded: Dictionary)                # decoded: see NetCodec.unpack_snapshot
 signal score_update_received(score: Dictionary)
 signal state_change_received(state: int, at_tick: int)   # §6.1 MatchState.State
@@ -57,6 +64,40 @@ const RATE_LIMIT_WINDOW_MS := 1000
 const RATE_LIMIT_EXCESS_PACKETS_TO_DISCONNECT := RATE_LIMIT_PACKETS_PER_SEC * 3
 const RATE_LIMIT_EXCESS_BYTES_TO_DISCONNECT := RATE_LIMIT_BYTES_PER_SEC * 3
 const MALFORMED_LIMIT_TO_DISCONNECT := 20
+# Task 5.10: how many rejected packets per peer per rate-limit window are
+# forwarded to `input_rejected`. Sized so an honest client — whose rejects are
+# occasional by definition, since a client rejected every tick is a bug the log
+# is meant to catch — is never sampled away, while a flood cannot turn the log
+# into unbounded attacker-controlled disk writes.
+const REJECTS_RECORDED_PER_WINDOW := 8
+
+# Server-stall grace (found by task 5.10's own reject recording, which is the
+# only reason it was visible at all).
+#
+# When the server stalls — a 2s SIGSTOP stands in for a GC/IO/scheduler hitch —
+# the client keeps sending at 60Hz throughout, and ENet delivers that entire
+# backlog in the first window after resume. Measured: 70 of an HONEST client's
+# input packets rejected as "rate limit exceeded", against a limit the client
+# never came close to violating on its own. Redundancy does not cover it: the
+# dropped packets are CONTIGUOUS, so each one's redundancy window falls inside
+# the same dropped run — 0 of 70 were rescued, and 82 of 923 sequences (8.88%,
+# ~1.4s of that player's input) never reached the server at all, versus 0.00%
+# missing on an otherwise identical run with no stall. Every prediction gate
+# still passed, which is exactly why this needed the log to find.
+#
+# So: don't rate-limit a backlog the server itself caused. The grace is capped,
+# expires after two windows, and is granted only to peers already being
+# tracked, so it cannot be farmed by a peer that connects during the stall. An
+# attacker who can induce server stalls to earn budget already has a strictly
+# worse capability than sending extra input packets.
+const STALL_DETECT_MS := 250
+const MAX_STALL_GRACE_PACKETS := SimConstants.TICK_HZ * 4   # 4s of a 60Hz client's backlog
+const STALL_GRACE_WINDOWS := 2
+
+enum InputRejectReason {
+	MALFORMED = 0,
+	RATE_LIMIT = 1,
+}
 
 
 class _PeerInputState:
@@ -72,9 +113,29 @@ class _PeerInputState:
 	var excess_packets := 0.0
 	var excess_bytes := 0.0
 	var malformed_count := 0
+	# Reject-recording budget for the current window. Without it the diagnostic
+	# is a remote disk-fill amplifier: the attacker chooses the flood rate, and
+	# every dropped packet would otherwise become a disk write. Capped per
+	# window, reset with the window itself, so an honest client's occasional
+	# reject is always captured while a flood contributes a bounded sample.
+	var rejects_recorded_this_window := 0
+	# Extra packets this peer may send before the limiter treats it as abuse,
+	# granted when the SERVER stalls and expiring shortly after.
+	var grace_packets := 0
+	var grace_windows_left := 0
 
 
 var _peer_input_state: Dictionary = {}   # peer_id -> _PeerInputState, server only
+# Uncapped lifetime reject totals, so the sampled log can be read against the
+# true figure — "8 rate-limit rejects recorded" means nothing on its own when
+# the recorder itself stops at 8 per window. Deliberately NOT part of
+# _PeerInputState, which is erased the moment a peer disconnects: a departed
+# peer's reject history is exactly what the post-mortem wants, and the first
+# version of this lost it (every summary printed an empty dictionary, because
+# the client had always disconnected by the time the server tore the match
+# down). peer_id -> {"malformed": int, "rate_limit": int}.
+var _reject_totals: Dictionary = {}
+var _last_physics_ms := 0
 
 # Bandwidth (task 3.7's debug overlay): only the two 60Hz hot-path channels
 # (input, snapshot) — match_config/score_update are low-frequency control
@@ -114,6 +175,32 @@ func get_bytes_received_per_sec() -> float:
 
 func _ready() -> void:
 	NetworkManager.client_disconnected.connect(func(peer_id: int) -> void: _peer_input_state.erase(peer_id))
+	# Seeded here, not left at 0, so the first physics frame measures a frame
+	# gap rather than the whole process uptime.
+	_last_physics_ms = Time.get_ticks_msec()
+
+
+# Server-side stall watchdog. A SIGSTOPped or hitching process doesn't run this
+# either, so the first physics frame after the stall is the one that sees the
+# whole wall-clock gap — which is precisely the size of the client backlog
+# about to arrive. Grace is handed only to peers ALREADY sending input, so a
+# peer that connects during the stall gets none of it.
+func _physics_process(_delta: float) -> void:
+	var now := Time.get_ticks_msec()
+	var gap := now - _last_physics_ms
+	_last_physics_ms = now
+	if not multiplayer.is_server() or _peer_input_state.is_empty():
+		return
+	if gap < STALL_DETECT_MS:
+		return
+	var credit: int = mini(int(float(gap) * SimConstants.TICK_HZ / 1000.0), MAX_STALL_GRACE_PACKETS)
+	for peer_id in _peer_input_state:
+		var state: _PeerInputState = _peer_input_state[peer_id]
+		state.grace_packets = mini(state.grace_packets + credit, MAX_STALL_GRACE_PACKETS)
+		state.grace_windows_left = STALL_GRACE_WINDOWS
+	push_warning("MatchSim: server stalled %dms — granting %d packets of rate-limit grace to %d peer(s)" % [
+		gap, credit, _peer_input_state.size()
+	])
 
 
 func _track_sent(n: int) -> void:
@@ -262,19 +349,35 @@ func _recv_input(bytes: PackedByteArray) -> void:
 	# when nothing is arriving anyway.
 	var now_ms := Time.get_ticks_msec()
 	if now_ms - state.window_start_ms >= RATE_LIMIT_WINDOW_MS:
-		state.excess_packets = maxf(0.0, state.excess_packets + float(state.packets_this_window) - float(RATE_LIMIT_PACKETS_PER_SEC))
-		state.excess_bytes = maxf(0.0, state.excess_bytes + float(state.bytes_this_window) - float(RATE_LIMIT_BYTES_PER_SEC))
+		# The leaky bucket drains against the SAME budget the window itself was
+		# policed with, grace included — otherwise a server stall would still
+		# accumulate excess toward a disconnect for traffic the server just
+		# explicitly allowed.
+		state.excess_packets = maxf(0.0, state.excess_packets + float(state.packets_this_window) - float(_packet_budget(state)))
+		state.excess_bytes = maxf(0.0, state.excess_bytes + float(state.bytes_this_window) - float(_byte_budget(state)))
 		state.window_start_ms = now_ms
 		state.packets_this_window = 0
 		state.bytes_this_window = 0
+		state.rejects_recorded_this_window = 0
+		if state.grace_windows_left > 0:
+			state.grace_windows_left -= 1
+			if state.grace_windows_left == 0:
+				state.grace_packets = 0
 		if state.excess_packets > RATE_LIMIT_EXCESS_PACKETS_TO_DISCONNECT or state.excess_bytes > RATE_LIMIT_EXCESS_BYTES_TO_DISCONNECT:
+			# Record before disconnecting, same reasoning as _count_malformed:
+			# the log should contain the packet that ended the connection, not
+			# stop one short of it.
+			_emit_reject(peer_id, state, InputRejectReason.RATE_LIMIT, bytes)
 			_disconnect_abusive_peer(peer_id, "input rate limit exceeded (excess_packets=%.0f excess_bytes=%.0f)" % [state.excess_packets, state.excess_bytes])
 			return
 
 	state.packets_this_window += 1
 	state.bytes_this_window += bytes.size()
-	if state.packets_this_window > RATE_LIMIT_PACKETS_PER_SEC or state.bytes_this_window > RATE_LIMIT_BYTES_PER_SEC:
-		return   # over budget for the current window — drop, counted above at the next window roll
+	if state.packets_this_window > _packet_budget(state) or state.bytes_this_window > _byte_budget(state):
+		# Over budget for the current window — drop, counted above at the next
+		# window roll.
+		_emit_reject(peer_id, state, InputRejectReason.RATE_LIMIT, bytes)
+		return
 
 	# Framing (§3.1 step 3), validated before decoding — unpack_input can't
 	# be trusted to catch this itself: StreamPeerBuffer silently zero-fills
@@ -283,11 +386,11 @@ func _recv_input(bytes: PackedByteArray) -> void:
 	# payload would otherwise decode "successfully" into garbage actions
 	# instead of being rejected.
 	if bytes.size() < NetCodec.INPUT_HEADER_SIZE:
-		_count_malformed(peer_id, state)
+		_count_malformed(peer_id, state, bytes)
 		return
 	var count: int = bytes[5]   # type_version(1) + seq(4) precede count — see pack_input's own layout
 	if count == 0 or count > NetCodec.MAX_REDUNDANCY or bytes.size() != NetCodec.INPUT_HEADER_SIZE + count * NetCodec.INPUT_ENTRY_SIZE:
-		_count_malformed(peer_id, state)
+		_count_malformed(peer_id, state, bytes)
 		return
 
 	var decoded := NetCodec.unpack_input(bytes)
@@ -299,12 +402,45 @@ func _recv_input(bytes: PackedByteArray) -> void:
 	input_received.emit(peer_id, decoded)
 
 
-func _count_malformed(peer_id: int, state: _PeerInputState) -> void:
+# The budget a peer is actually policed against right now: the standing limit
+# plus any outstanding server-stall grace. Bytes scale with packets by the same
+# worst-case-packet factor RATE_LIMIT_BYTES_PER_SEC itself is derived from, so
+# the two budgets can never drift apart by hand.
+func _packet_budget(state: _PeerInputState) -> int:
+	return RATE_LIMIT_PACKETS_PER_SEC + state.grace_packets
+
+
+func _byte_budget(state: _PeerInputState) -> int:
+	return _packet_budget(state) * (NetCodec.INPUT_HEADER_SIZE + NetCodec.MAX_REDUNDANCY * NetCodec.INPUT_ENTRY_SIZE)
+
+
+func _count_malformed(peer_id: int, state: _PeerInputState, bytes: PackedByteArray) -> void:
 	state.malformed_count += 1
+	# Emitted before the disconnect check so the packet that finally crossed
+	# the limit is itself in the log, not just the 19 before it.
+	_emit_reject(peer_id, state, InputRejectReason.MALFORMED, bytes)
 	if state.malformed_count >= MALFORMED_LIMIT_TO_DISCONNECT:
 		_disconnect_abusive_peer(peer_id, "too many malformed input packets (%d)" % state.malformed_count)
 
 
+func _emit_reject(peer_id: int, state: _PeerInputState, reason: int, bytes: PackedByteArray) -> void:
+	var totals: Dictionary = _reject_totals.get(peer_id, {"malformed": 0, "rate_limit": 0})
+	var key := "rate_limit" if reason == InputRejectReason.RATE_LIMIT else "malformed"
+	totals[key] = int(totals[key]) + 1
+	_reject_totals[peer_id] = totals
+	if state.rejects_recorded_this_window >= REJECTS_RECORDED_PER_WINDOW:
+		return
+	state.rejects_recorded_this_window += 1
+	input_rejected.emit(peer_id, reason, bytes)
+
+
+# Server-side, diagnostic. peer_id -> {"malformed": int, "rate_limit": int},
+# uncapped and surviving the peer's disconnect. Peers with no rejects at all
+# never appear, so an empty dictionary means a clean session.
+func get_reject_totals() -> Dictionary:
+	return _reject_totals.duplicate(true)
+
+
 func _disconnect_abusive_peer(peer_id: int, reason: String) -> void:
 	push_warning("MatchSim: disconnecting peer %d for abuse: %s" % [peer_id, reason])
 	_peer_input_state.erase(peer_id)