From 2aeaf1d44d6ea1afbccb28f1749b5f86836b1310 Mon Sep 17 00:00:00 2001 From: Voyager Date: Sun, 23 Aug 2026 01:23:00 -0400 Subject: [PATCH] fix: synchronize late-joining players --- network/network_chat_service.gd | 50 ++++++- network/network_profile_service.gd | 63 ++++---- network/network_session.gd | 95 +++++++++--- scripts/run_validations.sh | 74 +++++++++- tests/late_join_multiplayer_validation.gd | 137 ++++++++++++++++++ tests/late_join_multiplayer_validation.gd.uid | 1 + 6 files changed, 371 insertions(+), 49 deletions(-) create mode 100644 tests/late_join_multiplayer_validation.gd create mode 100644 tests/late_join_multiplayer_validation.gd.uid diff --git a/network/network_chat_service.gd b/network/network_chat_service.gd index b24d558..55ba768 100644 --- a/network/network_chat_service.gd +++ b/network/network_chat_service.gd @@ -5,6 +5,7 @@ const BURST_COUNT: int = 3 const WINDOW_COUNT: int = 5 const WINDOW_SECONDS: float = 10.0 const CALL_COOLDOWN_MILLISECONDS: int = 90 +const MAX_PENDING_MESSAGES_PER_PEER: int = 8 const DEDICATED_CHAT_WARNING: String = ( "Privacy notice: This dedicated server records and retains chat messages." ) @@ -40,6 +41,7 @@ var _last_call_msec: Dictionary[int, int] = {} var _call_variant_indices: Dictionary[int, int] = {} var _sequence: int = 0 var _peer_names: Dictionary[int, String] = {} +var _pending_peer_messages: Dictionary[int, Array] = {} var _relationships: PlayerRelationshipStore var _dedicated_history_enabled: bool = false var _chat_log_path: String = "" @@ -418,9 +420,11 @@ func _apply_message( else: var sender_id := int(data["sender_peer_id"]) var record := _session.get_peer_record(sender_id) + if record == null: + _queue_pending_peer_message(sender_id, data, emit_live_signals) + return valid_signature = ( - record != null - and record.identity_fingerprint == str(data["sender_fingerprint"]) + record.identity_fingerprint == str(data["sender_fingerprint"]) and _session.verify_peer_action( sender_id, "chat_send", @@ -466,7 +470,47 @@ func receive_chat_rejection(message: String) -> void: send_rejected.emit(message.left(80)) +func _queue_pending_peer_message( + peer_id: int, + message: Dictionary, + emit_live_signals: bool, +) -> void: + if peer_id <= 0: + return + var pending: Array = _pending_peer_messages.get(peer_id, []) + var message_id := str(message.get("message_id", "")) + for value: Variant in pending: + if ( + typeof(value) == TYPE_DICTIONARY + and str(value.get("message", {}).get("message_id", "")) == message_id + ): + return + pending.append({ + "message": message.duplicate(true), + "emit_live_signals": emit_live_signals, + }) + while pending.size() > MAX_PENDING_MESSAGES_PER_PEER: + pending.pop_front() + _pending_peer_messages[peer_id] = pending + + +func _flush_pending_peer_messages(peer_id: int) -> void: + var pending: Array = _pending_peer_messages.get(peer_id, []) + _pending_peer_messages.erase(peer_id) + for value: Variant in pending: + if typeof(value) != TYPE_DICTIONARY: + continue + var queued: Dictionary = value + if typeof(queued.get("message")) != TYPE_DICTIONARY: + continue + _apply_message( + queued["message"], + bool(queued.get("emit_live_signals", true)), + ) + + func _on_peer_authenticated(peer_id: int, display_name: String) -> void: + _flush_pending_peer_messages(peer_id) if not _session.is_host(): return _peer_names[peer_id] = display_name @@ -500,6 +544,7 @@ func _on_peer_removed(peer_id: int) -> void: _rate_times.erase(peer_id) _last_call_msec.erase(peer_id) _call_variant_indices.erase(peer_id) + _pending_peer_messages.erase(peer_id) if not _session.is_host(): return var display_name: String = _peer_names.get(peer_id, "Player") @@ -601,6 +646,7 @@ func _on_session_state_changed(state: NetworkSession.State) -> void: _last_call_msec.clear() _call_variant_indices.clear() _peer_names.clear() + _pending_peer_messages.clear() _sequence = 0 history_replaced.emit([]) diff --git a/network/network_profile_service.gd b/network/network_profile_service.gd index da26d86..e4732a9 100644 --- a/network/network_profile_service.gd +++ b/network/network_profile_service.gd @@ -97,6 +97,10 @@ func preview_appearance(appearance: Dictionary) -> bool: _preferences.display_name, snapshot, ) + var authorization := _make_appearance_preview_authorization(snapshot) + var local_record := _session.get_peer_record(local_peer_id) + if local_record != null: + local_record.profile_authorization = authorization.duplicate(true) if _session.is_host(): _broadcast_appearance_preview_to_supported_peers( local_peer_id, @@ -105,20 +109,27 @@ func preview_appearance(appearance: Dictionary) -> bool: elif _session.supports_server_capability( NetworkProtocol.APPEARANCE_PREVIEW_CAPABILITY ): - var request: Dictionary = { - "request_id": _new_id(), - "session_id": _session.get_session_id(), - "appearance": snapshot.duplicate(true), - "sender_fingerprint": _session.get_local_identity_fingerprint(), - } - request["sender_signature"] = _session.sign_local_action( - "appearance_preview", - NetworkProfileProtocol.appearance_preview_signature_fields(request), - ) - submit_appearance_preview.rpc_id(1, request) + submit_appearance_preview.rpc_id(1, authorization) return true +func _make_appearance_preview_authorization( + appearance: Dictionary, +) -> Dictionary: + var request: Dictionary = { + "domain": "appearance_preview", + "request_id": _new_id(), + "session_id": _session.get_session_id(), + "appearance": appearance.duplicate(true), + "sender_fingerprint": _session.get_local_identity_fingerprint(), + } + request["sender_signature"] = _session.sign_local_action( + "appearance_preview", + NetworkProfileProtocol.appearance_preview_signature_fields(request), + ) + return request + + func restore_persisted_appearance() -> void: preview_appearance(_appearance_store.get_snapshot()) @@ -274,6 +285,9 @@ func submit_appearance_preview(data: Dictionary) -> void: var appearance := CharacterCustomizationCatalog.sanitized_snapshot( data["appearance"] ) + var authorization := data.duplicate(true) + authorization["domain"] = "appearance_preview" + record.profile_authorization = authorization _session.apply_canonical_profile( sender_id, record.display_name, @@ -482,23 +496,16 @@ func broadcast_profile_snapshot( or not CharacterCustomizationCatalog.validate_snapshot(appearance) ): return - if not authorization.is_empty(): - var signed := authorization.duplicate(true) - signed["display_name"] = display_name - signed["appearance"] = appearance - var record := _session.get_peer_record(peer_id) - if ( - record == null - or record.identity_fingerprint - != str(signed.get("sender_fingerprint", "")) - or not _session.verify_peer_action( - peer_id, - "profile_update", - NetworkProfileProtocol.signature_fields(signed), - signed.get("sender_signature", PackedByteArray()), - ) - ): - return + if ( + not authorization.is_empty() + and not _session.verify_profile_authorization( + peer_id, + display_name, + appearance, + authorization, + ) + ): + return _session.apply_canonical_profile(peer_id, display_name, appearance) _apply_to_avatar(peer_id, appearance) profile_snapshot_changed.emit(peer_id, display_name, appearance.duplicate(true)) diff --git a/network/network_session.gd b/network/network_session.gd index 21d400c..4d38629 100644 --- a/network/network_session.gd +++ b/network/network_session.gd @@ -1559,8 +1559,9 @@ func _apply_spawn_entry(entry: Dictionary) -> void: own_snapshot["visual_yaw"] = float(entry["yaw"]) own_avatar.apply_network_teleport(own_snapshot) return + var peer_was_added: bool = false if not _registry.has_peer(peer_id): - _registry.add_peer( + if not _registry.add_peer( peer_id, entry["profile_id"], entry["display_name"], @@ -1568,8 +1569,14 @@ func _apply_spawn_entry(entry: Dictionary) -> void: str(entry.get("identity_fingerprint", "")), str(entry.get("identity_public_key", "")), _sanitized_capabilities(entry.get("capability_flags", [])), - ) + ): + return + peer_was_added = true var added_record := _registry.get_peer(peer_id) + if added_record != null: + added_record.profile_authorization = Dictionary( + entry.get("profile_authorization", {}) + ).duplicate(true) if added_record != null and added_record.identity_authenticated: _archive_authenticated_identity(added_record) var status := _known_players.observe( @@ -1596,6 +1603,8 @@ func _apply_spawn_entry(entry: Dictionary) -> void: var record := _registry.get_peer(peer_id) if avatar != null and record != null: avatar.apply_appearance_snapshot(record.appearance_snapshot) + if peer_was_added and record != null: + peer_authenticated.emit(peer_id, record.display_name) func _build_spawn_list() -> Array[Dictionary]: @@ -1695,25 +1704,62 @@ func _update_local_operator() -> void: _local_operator = bool(_operator_peer_ids.get(local_peer_id, false)) -func _verify_spawn_identity(entry: Dictionary) -> bool: - var fingerprint := str(entry.get("identity_fingerprint", "")) - var public_pem := NetworkIdentityCrypto.normalize_public_pem( - str(entry.get("identity_public_key", "")) - ) - if ( - NetworkIdentityCrypto.fingerprint_public_pem(public_pem) != fingerprint - or typeof(entry.get("profile_authorization")) != TYPE_DICTIONARY - ): +func verify_profile_authorization( + peer_id: int, + display_name: String, + appearance: Dictionary, + authorization: Dictionary, +) -> bool: + var record := _registry.get_peer(peer_id) + if record == null: + return false + return _verify_profile_authorization_fields( + peer_id, + record.profile_id, + display_name, + appearance, + record.identity_fingerprint, + record.identity_public_key, + authorization, + ) + + +func _verify_spawn_identity(entry: Dictionary) -> bool: + if typeof(entry.get("profile_authorization")) != TYPE_DICTIONARY: + return false + return _verify_profile_authorization_fields( + int(entry.get("peer_id", 0)), + str(entry.get("profile_id", "")), + str(entry.get("display_name", "")), + Dictionary(entry.get("appearance", {})), + str(entry.get("identity_fingerprint", "")), + str(entry.get("identity_public_key", "")), + Dictionary(entry.get("profile_authorization", {})), + ) + + +func _verify_profile_authorization_fields( + peer_id: int, + profile_id: String, + display_name: String, + appearance: Dictionary, + fingerprint: String, + identity_public_key: String, + authorization: Dictionary, +) -> bool: + var public_pem := NetworkIdentityCrypto.normalize_public_pem( + identity_public_key + ) + if NetworkIdentityCrypto.fingerprint_public_pem(public_pem) != fingerprint: return false - var authorization: Dictionary = entry["profile_authorization"] if authorization.is_empty(): - return int(entry.get("peer_id", 0)) == 1 + return peer_id == 1 if authorization.get("domain") == "handshake_client_profile": var hello := NetworkProtocol.make_client_hello( - str(entry.get("profile_id", "")), - str(entry.get("display_name", "")), + profile_id, + display_name, str(authorization.get("client_nonce", "")), - Dictionary(entry.get("appearance", {})), + appearance, fingerprint, authorization.get("signature", PackedByteArray()), ) @@ -1723,10 +1769,23 @@ func _verify_spawn_identity(entry: Dictionary) -> bool: NetworkProtocol.client_profile_fields(hello), hello["identity_signature"], ) + if authorization.get("domain") == "appearance_preview": + if str(authorization.get("sender_fingerprint", "")) != fingerprint: + return false + var preview := authorization.duplicate(true) + preview["appearance"] = appearance + return NetworkIdentityCrypto.verify_fields( + NetworkIdentityCrypto.load_public_key(public_pem), + "appearance_preview", + NetworkProfileProtocol.appearance_preview_signature_fields(preview), + preview.get("sender_signature", PackedByteArray()), + ) if authorization.has("sender_signature"): + if str(authorization.get("sender_fingerprint", "")) != fingerprint: + return false var signed := authorization.duplicate(true) - signed["display_name"] = entry.get("display_name", "") - signed["appearance"] = entry.get("appearance", {}) + signed["display_name"] = display_name + signed["appearance"] = appearance return NetworkIdentityCrypto.verify_fields( NetworkIdentityCrypto.load_public_key(public_pem), "profile_update", diff --git a/scripts/run_validations.sh b/scripts/run_validations.sh index 64fbfc3..1368afc 100755 --- a/scripts/run_validations.sh +++ b/scripts/run_validations.sh @@ -96,6 +96,7 @@ readonly -a NETWORK_TESTS=( ) readonly SESSION_SWITCH_NETWORK_TEST="tests/session_switch_multiplayer_validation.gd" +readonly LATE_JOIN_NETWORK_TEST="tests/late_join_multiplayer_validation.gd" cleanup() { rm -rf -- "${RUN_ROOT}" @@ -104,7 +105,7 @@ trap cleanup EXIT INT TERM usage() { printf \ - 'Usage: %s {quick|full|host|network|session-switch|all|--list}\n' \ + 'Usage: %s {quick|full|host|network|session-switch|late-join|all|--list}\n' \ "$0" } @@ -290,6 +291,71 @@ run_session_switch_network_test() { fi } + +run_late_join_network_test() { + local script="${LATE_JOIN_NETWORK_TEST}" + local name="${script#tests/}" + name="${name%.gd}" + local host_root first_root late_root + local host_pid first_pid late_pid host_status first_status late_status + host_root="$(prepare_root "${name}-host")" + first_root="$(prepare_root "${name}-first")" + late_root="$(prepare_root "${name}-late")" + printf '\n==> %s (host + first client + late client)\n' "${script}" + XDG_DATA_HOME="${host_root}/data" \ + XDG_CONFIG_HOME="${host_root}/config" \ + XDG_CACHE_HOME="${host_root}/cache" \ + timeout "${TEST_TIMEOUT_SECONDS}s" "${GODOT_BIN}" \ + --headless --path "${PROJECT_ROOT}" --script "${script}" -- host \ + >"${host_root}/output.log" 2>&1 & + host_pid=$! + if ! wait_for_network_host "${script}" "${host_pid}"; then + set +e + wait "${host_pid}" + host_status=$? + set -e + printf '%s\n' '-- host output --' + sed -n '1,240p' "${host_root}/output.log" + printf 'error: host did not become ready (exit %d)\n' \ + "${host_status}" >&2 + return 1 + fi + XDG_DATA_HOME="${first_root}/data" \ + XDG_CONFIG_HOME="${first_root}/config" \ + XDG_CACHE_HOME="${first_root}/cache" \ + timeout "${TEST_TIMEOUT_SECONDS}s" "${GODOT_BIN}" \ + --headless --path "${PROJECT_ROOT}" --script "${script}" -- first \ + >"${first_root}/output.log" 2>&1 & + first_pid=$! + sleep 4 + XDG_DATA_HOME="${late_root}/data" \ + XDG_CONFIG_HOME="${late_root}/config" \ + XDG_CACHE_HOME="${late_root}/cache" \ + timeout "${TEST_TIMEOUT_SECONDS}s" "${GODOT_BIN}" \ + --headless --path "${PROJECT_ROOT}" --script "${script}" -- late \ + >"${late_root}/output.log" 2>&1 & + late_pid=$! + set +e + wait "${late_pid}" + late_status=$? + wait "${first_pid}" + first_status=$? + wait "${host_pid}" + host_status=$? + set -e + printf '%s\n' '-- host output --' + sed -n '1,240p' "${host_root}/output.log" + printf '%s\n' '-- first client output --' + sed -n '1,240p' "${first_root}/output.log" + printf '%s\n' '-- late client output --' + sed -n '1,240p' "${late_root}/output.log" + if ((host_status != 0 || first_status != 0 || late_status != 0)); then + printf 'error: host exited %d; clients exited %d/%d\n' \ + "${host_status}" "${first_status}" "${late_status}" >&2 + return 1 + fi +} + run_quick() { local test_script for test_script in "${QUICK_TESTS[@]}"; do @@ -317,6 +383,7 @@ run_network() { for test_script in "${NETWORK_TESTS[@]}"; do run_network_test "${test_script}" done + run_late_join_network_test run_session_switch_network_test } @@ -331,6 +398,8 @@ list_tests() { printf ' %s\n' "${NETWORK_TESTS[@]}" printf '%s\n' 'Two-host session-switch network test:' printf ' %s\n' "${SESSION_SWITCH_NETWORK_TEST}" + printf '%s\n' 'Three-peer late-join network test:' + printf ' %s\n' "${LATE_JOIN_NETWORK_TEST}" } case "${1:-}" in @@ -349,6 +418,9 @@ case "${1:-}" in session-switch) run_session_switch_network_test ;; + late-join) + run_late_join_network_test + ;; all) run_full run_host diff --git a/tests/late_join_multiplayer_validation.gd b/tests/late_join_multiplayer_validation.gd new file mode 100644 index 0000000..f726dbc --- /dev/null +++ b/tests/late_join_multiplayer_validation.gd @@ -0,0 +1,137 @@ +extends SceneTree + +const MainScene: PackedScene = preload("res://main/main.tscn") +const TEST_PORT: int = 18194 +const LATE_MESSAGE: String = "late join visibility check" + + +func _initialize() -> void: + call_deferred("_run") + + +func _run() -> void: + var arguments: PackedStringArray = OS.get_cmdline_user_args() + if arguments.has("host"): + await _run_host() + elif arguments.has("first"): + await _run_first_client() + elif arguments.has("late"): + await _run_late_client() + else: + push_error("Late-join validation needs host, first, or late mode.") + quit(1) + + +func _run_host() -> void: + var main: Node = await _create_initialized_main() + var session := main.get_node("%NetworkSession") as NetworkSession + var save_manager := main.get("_save_manager") as PlayerSaveManager + assert(session.start_private_host(TEST_PORT)) + assert(save_manager.initialize_new_game()) + main.call("_enter_gameplay") + for _frame: int in 4: + await physics_frame + assert(session.set_host_open(true)) + assert(await _wait_for_registry_size(session, 3)) + var spawn := main.get_node("%PlayerSpawnService") as PlayerSpawnService + assert(spawn.get_peer_ids().size() == 3) + await create_timer(5.0).timeout + print("Late-join multiplayer host validation: PASS") + await _cleanup(main, session) + + +func _run_first_client() -> void: + var main: Node = await _create_initialized_main() + var session := await _join(main) + var profile := main.get_node("%NetworkProfileService") as NetworkProfileService + var preview: Dictionary = profile.get_persisted_appearance().duplicate(true) + preview["scale"] = 0.9 + assert(profile.preview_appearance(preview)) + await create_timer(2.0).timeout + var chat := main.get_node("%NetworkChatService") as NetworkChatService + var deadline: int = Time.get_ticks_msec() + 15000 + while Time.get_ticks_msec() < deadline: + await process_frame + if ( + session.get_authenticated_peer_ids().size() == 3 + and _history_contains(chat, LATE_MESSAGE) + ): + break + assert(session.get_authenticated_peer_ids().size() == 3) + assert(_history_contains(chat, LATE_MESSAGE)) + var spawn := main.get_node("%PlayerSpawnService") as PlayerSpawnService + assert(spawn.get_peer_ids().size() == 3) + print("Late-join first-client validation: PASS") + await _cleanup(main, session) + + +func _run_late_client() -> void: + var main: Node = await _create_initialized_main() + var session := await _join(main) + var deadline: int = Time.get_ticks_msec() + 8000 + while ( + Time.get_ticks_msec() < deadline + and session.get_authenticated_peer_ids().size() < 3 + ): + await process_frame + assert(session.get_authenticated_peer_ids().size() == 3) + var spawn := main.get_node("%PlayerSpawnService") as PlayerSpawnService + assert(spawn.get_peer_ids().size() == 3) + var chat := main.get_node("%NetworkChatService") as NetworkChatService + assert(chat.send_local_message(LATE_MESSAGE)) + await create_timer(1.0).timeout + print("Late-join late-client validation: PASS") + await _cleanup(main, session) + + +func _join(main: Node) -> NetworkSession: + main.call("_on_title_join_game_requested", "127.0.0.1:%d" % TEST_PORT) + var session := main.get_node("%NetworkSession") as NetworkSession + var deadline: int = Time.get_ticks_msec() + 20000 + while Time.get_ticks_msec() < deadline: + await process_frame + if session.state == NetworkSession.State.VERIFYING_SERVER_IDENTITY: + main.call("_confirm_server_trust") + if session.is_joined_client() and bool(main.get("_gameplay_started")): + return session + assert(false, "Client did not join in time.") + return session + + +func _wait_for_registry_size(session: NetworkSession, expected: int) -> bool: + var deadline: int = Time.get_ticks_msec() + 25000 + while Time.get_ticks_msec() < deadline: + await process_frame + if session.get_authenticated_peer_ids().size() == expected: + return true + return false + + +func _history_contains(service: NetworkChatService, body: String) -> bool: + return service.get_history().any( + func(message: Dictionary) -> bool: + return str(message.get("body", "")) == body + ) + + +func _create_initialized_main() -> Node: + root.size = Vector2i(1280, 720) + var main: Node = MainScene.instantiate() + root.add_child(main) + for _frame: int in 4: + await process_frame + if not bool(main.get("_application_initialized")): + main.call("_activate_selected_data_path", "", true) + for _frame: int in 8: + await process_frame + assert(bool(main.get("_application_initialized"))) + return main + + +func _cleanup(main: Node, session: NetworkSession) -> void: + session.disconnect_session("") + main.queue_free() + for _frame: int in 4: + await process_frame + await create_timer(0.1).timeout + quit() diff --git a/tests/late_join_multiplayer_validation.gd.uid b/tests/late_join_multiplayer_validation.gd.uid new file mode 100644 index 0000000..5d33548 --- /dev/null +++ b/tests/late_join_multiplayer_validation.gd.uid @@ -0,0 +1 @@ +uid://dnuwxkoeeykt3