feat(multiplayer): bind admissions to durable leases

2026-09-13 11:22:04 +00:00 · 2026-09-03 13:45:39 +01:00
parent 3e0022ce9c
commit aac81c89b6
9 changed files with 301 additions and 64 deletions
@@ -0,0 +1,153 @@
+class_name ConnectionLeaseClient
+extends Node
+
+const AssignmentState = preload("res://scripts/assignment_state.gd")
+
+signal reconciliation_failed(reason: String)
+
+const CLAIMED := "claimed"
+const UNAVAILABLE := "unavailable"
+const REJECTED := "rejected"
+
+var _base_url := ""
+var _workload_token := ""
+var _match_id := ""
+var _server_id := ""
+var _pending: Array[Dictionary] = []
+var _processing := false
+
+
+func configure(base_url: String, workload_token: String, match_id: String, server_id: String) -> bool:
+	base_url = base_url.strip_edges().trim_suffix("/")
+	workload_token = workload_token.strip_edges()
+	if not valid_configuration(base_url, workload_token, match_id, server_id):
+		return false
+	_base_url = base_url
+	_workload_token = workload_token
+	_match_id = match_id
+	_server_id = server_id
+	return true
+
+
+# Admission awaits one bounded request only. If the control plane is down, the
+# same process may continue using its local generation and this exact event is
+# retained ahead of every later disconnect/reconnect for ordered reconciliation.
+func claim(player_id: String, expected_generation: int) -> Dictionary:
+	if not AssignmentState.is_valid_opaque_id(player_id) or expected_generation < 0:
+		return {"status": REJECTED}
+	var event := _connect_event(player_id, expected_generation)
+	if _processing or not _pending.is_empty():
+		if expected_generation == 0:
+			return {"status": REJECTED}
+		_pending.append(event)
+		_start_processing()
+		return {"status": UNAVAILABLE, "generation": expected_generation + 1}
+	var response := await _send(event, true)
+	if String(response.get("status", "")) == UNAVAILABLE:
+		if expected_generation == 0:
+			return {"status": REJECTED}
+		_pending.append(event)
+		_start_processing()
+		return {"status": UNAVAILABLE, "generation": expected_generation + 1}
+	return response
+
+
+func record_disconnect(player_id: String, generation: int) -> void:
+	if not AssignmentState.is_valid_opaque_id(player_id) or generation < 1:
+		return
+	_pending.append(_disconnect_event(player_id, generation))
+	_start_processing()
+
+
+func _start_processing() -> void:
+	if _processing or _pending.is_empty() or not is_inside_tree():
+		return
+	_process_pending()
+
+
+func _process_pending() -> void:
+	_processing = true
+	while not _pending.is_empty() and is_inside_tree():
+		var event := _pending[0]
+		var response := await _send(event)
+		var status := String(response.get("status", ""))
+		if status == CLAIMED:
+			_pending.pop_front()
+			continue
+		if status == REJECTED:
+			reconciliation_failed.emit("durable connection lease conflict")
+			_processing = false
+			return
+		await get_tree().create_timer(1.0).timeout
+	_processing = false
+
+
+func _send(event: Dictionary, allow_recovery := false) -> Dictionary:
+	var request := HTTPRequest.new()
+	request.timeout = 1.0
+	add_child(request)
+	var operation := String(event["operation"])
+	var endpoint := "%s/v1/servers/%s/%s" % [_base_url, _server_id.uri_encode(), operation]
+	var start_error := request.request(endpoint, [
+		"Authorization: Bearer " + _workload_token,
+		"Content-Type: application/json",
+		"Idempotency-Key: " + String(event["key"]),
+	], HTTPClient.METHOD_POST, JSON.stringify(event["payload"]))
+	if start_error != OK:
+		request.queue_free()
+		return {"status": UNAVAILABLE}
+	var raw: Array = await request.request_completed
+	request.queue_free()
+	return classify_response(operation, int(event["generation"]), int(raw[0]), int(raw[1]), raw[3], allow_recovery)
+
+
+func _connect_event(player_id: String, expected_generation: int) -> Dictionary:
+	return {
+		"operation": "connect",
+		"generation": expected_generation,
+		"key": event_key(_match_id, player_id, "connect", expected_generation),
+		"payload": {"player_id": player_id, "expected_generation": expected_generation},
+	}
+
+
+func _disconnect_event(player_id: String, generation: int) -> Dictionary:
+	return {
+		"operation": "disconnect",
+		"generation": generation,
+		"key": event_key(_match_id, player_id, "disconnect", generation),
+		"payload": {"player_id": player_id, "generation": generation},
+	}
+
+
+static func classify_response(operation: String, generation: int, request_result: int, response_code: int, body: PackedByteArray, allow_recovery := false) -> Dictionary:
+	if request_result != HTTPRequest.RESULT_SUCCESS or response_code == 0 or response_code == 429 or response_code >= 500:
+		return {"status": UNAVAILABLE}
+	if operation == "disconnect" and response_code == 204:
+		return {"status": CLAIMED, "generation": generation}
+	if operation == "connect" and response_code == 200:
+		var decoded = JSON.parse_string(body.get_string_from_utf8())
+		if decoded is Dictionary and _valid_generation(decoded.get("generation")):
+			var claimed_generation := int(decoded["generation"])
+			if claimed_generation == generation + 1 or (allow_recovery and generation == 0 and claimed_generation > 1):
+				return {"status": CLAIMED, "generation": claimed_generation}
+	return {"status": REJECTED}
+
+
+static func _valid_generation(value: Variant) -> bool:
+	if value is int:
+		return int(value) >= 1
+	if value is float:
+		return is_finite(float(value)) and float(value) >= 1.0 and float(value) == floor(float(value)) and float(value) <= 9007199254740991.0
+	return false
+
+
+static func event_key(match_id: String, player_id: String, operation: String, generation: int) -> String:
+	return "server-lease-" + (match_id + "\n" + player_id + "\n" + operation + "\n" + str(generation)).sha256_text()
+
+
+static func valid_configuration(base_url: String, workload_token: String, match_id: String, server_id: String) -> bool:
+	if not (base_url.begins_with("http://") or base_url.begins_with("https://")) or base_url.contains("\n") or base_url.contains("\r") or base_url.contains("?") or base_url.contains("#"):
+		return false
+	if workload_token.is_empty() or workload_token.contains("\n") or workload_token.contains("\r"):
+		return false
+	return AssignmentState.is_valid_opaque_id(match_id) and AssignmentState.is_valid_opaque_id(server_id)
@@ -66,6 +66,8 @@ var _active_join_peers: Dictionary = {} # opaque authorisation -> peer_id
 var _join_history: Dictionary = {} # token -> {generation, lost_at}
 var _join_authorisation_context: Dictionary = {}
 var _join_signing_key := PackedByteArray()
+var _connection_lease_claim := Callable()
+var _connection_lease_disconnect := Callable()

 # Test hook (tests/match_net_smoke.gd): set false before connecting to
 # suppress the automatic real hello, so a test can send a deliberately
@@ -105,6 +107,8 @@ func _on_shutting_down() -> void:
 	_join_history.clear()
 	_join_authorisation_context.clear()
 	_join_signing_key = PackedByteArray()
+	_connection_lease_claim = Callable()
+	_connection_lease_disconnect = Callable()
 	require_join_authorisation = false
 	admissions_open = true

@@ -184,6 +188,8 @@ func _remove_player(peer_id: int) -> void:
 			var history: Dictionary = _join_history.get(token, {})
 			history["lost_at"] = Time.get_unix_time_from_system()
 			_join_history[token] = history
+			if _connection_lease_disconnect.is_valid():
+				_connection_lease_disconnect.call(_join_identity(token), int(history.get("generation", 0)))
 			break
 	if not roster.has(peer_id):
 		return
@@ -286,9 +292,9 @@ func _hello(protocol_version: int, tick_hz: int, player_name: String, supplied_j
 		return
 	var join_generation := 1
 	if require_join_authorisation:
-		join_generation = _reserve_join_authorisation(supplied_join_authorisation, peer_id)
+		join_generation = await _claim_join_authorisation(supplied_join_authorisation, peer_id)
 		if join_generation < 0:
-			await _reject(peer_id, "join authorisation reclaim expired")
+			await _reject(peer_id, "join authorisation lease rejected")
 			return
 	if player_name.length() > MAX_INPUT_LENGTH:
 		await _reject(peer_id, "player name too long")
@@ -432,16 +438,68 @@ func is_join_authorisation_active(token: String) -> bool:
 	return not token.is_empty() and _active_join_peers.has(token)


-func _reserve_join_authorisation(token: String, peer_id: int) -> int:
+func configure_connection_lease_callbacks(claim: Callable, disconnect: Callable) -> void:
+	_connection_lease_claim = claim
+	_connection_lease_disconnect = disconnect
+
+
+func _claim_join_authorisation(token: String, peer_id: int) -> int:
+	var expected_generation := _available_join_generation(token)
+	if expected_generation < 0:
+		return -1
+	var generation := expected_generation + 1
+	if _connection_lease_claim.is_valid():
+		var response = await _connection_lease_claim.call(_join_identity(token), expected_generation)
+		generation = lease_claim_generation(response, expected_generation)
+		if generation < 0:
+			return -1
+	# The await above deliberately allows one bounded control-plane request.
+	# Re-evaluate every local fact that can change during that suspension before
+	# publishing the reservation. If a durable claim succeeded, close it again.
+	# A concurrent same-token hello can receive the same idempotent claim; its
+	# loser must not close the generation now owned by the local winner.
+	if _active_join_peers.has(token):
+		return -1
+	if not admissions_open or not _valid_join_authorisation(token) or peer_id not in multiplayer.get_peers():
+		if _connection_lease_disconnect.is_valid():
+			_connection_lease_disconnect.call(_join_identity(token), generation)
+		return -1
+	_join_history[token] = {"generation": generation, "lost_at": 0.0}
+	_active_join_peers[token] = peer_id
+	return generation
+
+
+func _available_join_generation(token: String) -> int:
 	if token.is_empty() or _active_join_peers.has(token):
 		return -1
 	var now := Time.get_unix_time_from_system()
 	var history: Dictionary = _join_history.get(token, {})
 	var lost_at := float(history.get("lost_at", 0.0))
-	if lost_at > 0.0:
-		if now < lost_at or now - lost_at > RECONNECT_GRACE_SECONDS:
-			return -1
-	var generation := int(history.get("generation", 0)) + 1
+	if lost_at > 0.0 and (now < lost_at or now - lost_at > RECONNECT_GRACE_SECONDS):
+		return -1
+	return int(history.get("generation", 0))
+
+
+static func lease_claim_generation(response, expected_generation: int) -> int:
+	if not response is Dictionary or expected_generation < 0:
+		return -1
+	var status := String(response.get("status", ""))
+	if status not in ["claimed", "unavailable"] or not response.get("generation") is int:
+		return -1
+	var generation := int(response["generation"])
+	if status == "unavailable":
+		return generation if expected_generation > 0 and generation == expected_generation + 1 else -1
+	# A durable backend may return a later generation only to a fresh process
+	# recovering an already-disconnected lease. Locally known generations never
+	# skip, and outage fallback never invents a jump.
+	return generation if generation == expected_generation + 1 or (expected_generation == 0 and generation > 1) else -1
+
+
+func _reserve_join_authorisation(token: String, peer_id: int) -> int:
+	var expected_generation := _available_join_generation(token)
+	if expected_generation < 0:
+		return -1
+	var generation := expected_generation + 1
 	_join_history[token] = {"generation": generation, "lost_at": 0.0}
 	_active_join_peers[token] = peer_id
 	return generation
@@ -4,6 +4,7 @@ const NetCodec = preload("res://scripts/net_codec.gd")
 const ServerControlScript = preload("res://scripts/server_control.gd")
 const AgonesSDKScript = preload("res://scripts/agones_sdk.gd")
 const AssignmentState = preload("res://scripts/assignment_state.gd")
+const ConnectionLeaseClientScript = preload("res://scripts/connection_lease_client.gd")

 # Headless dedicated server entry point (task 1.6). Parses CLI args, hosts
 # via NetworkManager, logs structured lines, and watches for physics-tick
@@ -32,9 +33,8 @@ var _watchdog_armed := false   # skip the first _process(): engine startup sched
 var _control: ServerControl = null
 var _match_loop: ServerMatchLoop = null
 var _agones = null
+var _connection_leases = null
 var _drain_requested := false
-var _connection_reports_inflight: Dictionary = {}
-var _connection_reports_complete: Dictionary = {}


 func _ready() -> void:
@@ -105,6 +105,18 @@ func _ready() -> void:
 		get_tree().root.add_child.call_deferred(_agones)
 		if _agones.configure_from_environment():
 			_agones.start_health()
+		_connection_leases = ConnectionLeaseClientScript.new()
+		_connection_leases.name = "ConnectionLeases"
+		var lease_url := OS.get_environment("COSMIC_CLASH_CONTROL_PLANE_URL")
+		var lease_token := OS.get_environment("COSMIC_CLASH_WORKLOAD_TOKEN")
+		if _connection_leases.configure(lease_url, lease_token, String(config.get_value("match-id")), String(config.get_value("server-id"))):
+			_connection_leases.reconciliation_failed.connect(_on_connection_lease_reconciliation_failed)
+			get_tree().root.add_child.call_deferred(_connection_leases)
+			MatchNet.configure_connection_lease_callbacks(_connection_leases.claim, _connection_leases.record_disconnect)
+		else:
+			_connection_leases.queue_free()
+			_connection_leases = null
+			ServerLog.warn("connection_lease_backend_unavailable", {"reason": "invalid_or_missing_configuration"})

 	NetworkManager.client_connected.connect(_on_client_connected)
 	NetworkManager.client_disconnected.connect(_on_client_disconnected)
@@ -188,8 +200,6 @@ func _on_client_disconnected(peer_id: int) -> void:

 func _on_player_joined(peer_id: int, player_name: String) -> void:
 	ServerLog.info("player_joined", {"peer_id": peer_id, "name": player_name, "roster": MatchNet.roster.size()})
-	if config != null and bool(config.get_value("allocated-mode")):
-		_report_player_connected(MatchNet.player_identity(peer_id))


 func _on_player_left(peer_id: int) -> void:
@@ -209,52 +219,16 @@ func _on_initial_connect_ready() -> void:
 		ServerLog.info("initial_connect_window_started", {"match_id": String(config.get_value("match-id"))})


-func _report_player_connected(player_id: String) -> void:
-	var base_url := OS.get_environment("COSMIC_CLASH_CONTROL_PLANE_URL").strip_edges().trim_suffix("/")
-	var workload_token := OS.get_environment("COSMIC_CLASH_WORKLOAD_TOKEN").strip_edges()
-	var server_id := String(config.get_value("server-id"))
-	var match_id := String(config.get_value("match-id"))
-	if not valid_connection_report_configuration(base_url, workload_token, match_id, server_id, player_id) or _connection_reports_inflight.has(player_id) or _connection_reports_complete.has(player_id):
-		return
-	_connection_reports_inflight[player_id] = true
-	var endpoint := "%s/v1/servers/%s/connect" % [base_url, server_id.uri_encode()]
-	# A player can join many matches. Scope the durable key to this match so a
-	# later valid report cannot conflict with an earlier match's stored digest.
-	var idempotency_key := "server-connect-" + (match_id + "\n" + player_id).sha256_text()
-	var payload := JSON.stringify({"player_id": player_id})
-	for attempt in range(5):
-		var request := HTTPRequest.new()
-		request.timeout = 5.0
-		add_child(request)
-		var start_error := request.request(endpoint, [
-			"Authorization: Bearer " + workload_token,
-			"Content-Type: application/json",
-			"Idempotency-Key: " + idempotency_key,
-		], HTTPClient.METHOD_POST, payload)
-		var response_code := 0
-		if start_error == OK:
-			var response: Array = await request.request_completed
-			response_code = int(response[1])
-		request.queue_free()
-		if response_code == 204:
-			_connection_reports_complete[player_id] = true
-			_connection_reports_inflight.erase(player_id)
-			ServerLog.debug("player_connection_recorded", {"player_id": player_id})
-			return
-		if response_code in [400, 401, 404, 409, 422]:
-			break
-		if attempt < 4 and is_inside_tree():
-			await get_tree().create_timer(1.0).timeout
-	_connection_reports_inflight.erase(player_id)
-	ServerLog.warn("player_connection_report_failed", {"player_id": player_id})
+func _on_connection_lease_reconciliation_failed(reason: String) -> void:
+	# A durable/local divergence means this process can no longer prove that a
+	# future generation is globally current. Preserve the live match but close
+	# admission so it cannot mint additional ambiguous leases.
+	MatchNet.admissions_open = false
+	ServerLog.error("connection_lease_reconciliation_failed", {"reason": reason})


 static func valid_connection_report_configuration(base_url: String, workload_token: String, match_id: String, server_id: String, player_id: String) -> bool:
-	if not (base_url.begins_with("http://") or base_url.begins_with("https://")) or base_url.contains("\n") or base_url.contains("\r") or base_url.contains("?") or base_url.contains("#"):
-		return false
-	if workload_token.is_empty() or workload_token.contains("\n") or workload_token.contains("\r"):
-		return false
-	return AssignmentState.is_valid_opaque_id(match_id) and AssignmentState.is_valid_opaque_id(server_id) and AssignmentState.is_valid_opaque_id(player_id)
+	return ConnectionLeaseClientScript.valid_configuration(base_url, workload_token, match_id, server_id) and AssignmentState.is_valid_opaque_id(player_id)


 static func required_min_players(allocated: bool, roster_size: int, configured: int) -> int: