fix: synchronize late-joining players

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
+			record.identity_fingerprint == str(data["sender_fingerprint"])
-			and 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([])
 

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 = {
+		submit_appearance_preview.rpc_id(1, authorization)
-			"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)
 	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():
+	if (
-		var signed := authorization.duplicate(true)
+		not authorization.is_empty()
-		signed["display_name"] = display_name
+		and not _session.verify_profile_authorization(
-		signed["appearance"] = appearance
+			peer_id,
-		var record := _session.get_peer_record(peer_id)
+			display_name,
-		if (
+			appearance,
-			record == null
+			authorization,
-			or record.identity_fingerprint
+		)
-				!= str(signed.get("sender_fingerprint", ""))
+	):
-			or not _session.verify_peer_action(
+		return
-				peer_id,
-				"profile_update",
-				NetworkProfileProtocol.signature_fields(signed),
-				signed.get("sender_signature", PackedByteArray()),
-			)
-		):
-			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))

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:
+func verify_profile_authorization(
-	var fingerprint := str(entry.get("identity_fingerprint", ""))
+	peer_id: int,
-	var public_pem := NetworkIdentityCrypto.normalize_public_pem(
+	display_name: String,
-		str(entry.get("identity_public_key", ""))
+	appearance: Dictionary,
-	)
+	authorization: Dictionary,
-	if (
+) -> bool:
-		NetworkIdentityCrypto.fingerprint_public_pem(public_pem) != fingerprint
+	var record := _registry.get_peer(peer_id)
-		or typeof(entry.get("profile_authorization")) != TYPE_DICTIONARY
+	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", "")),
+			profile_id,
-			str(entry.get("display_name", "")),
+			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["display_name"] = display_name
-		signed["appearance"] = entry.get("appearance", {})
+		signed["appearance"] = appearance
 		return NetworkIdentityCrypto.verify_fields(
 			NetworkIdentityCrypto.load_public_key(public_pem),
 			"profile_update",

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

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()

tests/late_join_multiplayer_validation.gd.uid

@@ -0,0 +1 @@
+uid://dnuwxkoeeykt3