feat(multiplayer): add idempotent client mutation retries

This commit is contained in:
Josh Creek
2026-09-01 17:02:55 +01:00
parent 9ae01ecc5a
commit 6ef55affcb
3 changed files with 35 additions and 0 deletions
+24
View File
@@ -27,6 +27,8 @@ var assignment: AssignmentState
var _request: HTTPRequest var _request: HTTPRequest
var _operation := "" var _operation := ""
var _last_queue_create: Dictionary = {} var _last_queue_create: Dictionary = {}
var _last_mutation: Dictionary = {}
var _last_mutation_retryable := false
var _websocket: WebSocketPeer var _websocket: WebSocketPeer
var _websocket_status := "DISCONNECTED" var _websocket_status := "DISCONNECTED"
var _websocket_retry_seconds := 0.0 var _websocket_retry_seconds := 0.0
@@ -159,6 +161,17 @@ func can_retry_queue_create() -> bool:
return not _last_queue_create.is_empty() and state.phase == MatchmakingState.FAILED and String(_last_queue_create.get("ticket_id", "")) == state.ticket_id return not _last_queue_create.is_empty() and state.phase == MatchmakingState.FAILED and String(_last_queue_create.get("ticket_id", "")) == state.ticket_id
func retry_last_mutation() -> Error:
if not can_retry_last_mutation():
return ERR_INVALID_DATA
var request := _last_mutation.duplicate(true)
return _start_request(String(request["operation"]), int(request["method"]), String(request["path"]), request["payload"], String(request["key"]), int(request["expected_revision"]))
func can_retry_last_mutation() -> bool:
return _last_mutation_retryable and not _last_mutation.is_empty() and _operation.is_empty() and not auth_expired and is_valid_access_token(access_token)
func recover_queue(ticket_id: String) -> Error: func recover_queue(ticket_id: String) -> Error:
if ticket_id.is_empty(): if ticket_id.is_empty():
return ERR_INVALID_PARAMETER return ERR_INVALID_PARAMETER
@@ -256,6 +269,10 @@ static func is_valid_access_token(token: String) -> bool:
return separator > 0 and separator < token.length() - 1 and token.length() <= 4096 and not token.contains("\r") and not token.contains("\n") return separator > 0 and separator < token.length() - 1 and token.length() <= 4096 and not token.contains("\r") and not token.contains("\n")
static func is_retryable_mutation_response(response_code: int) -> bool:
return response_code == 0 or response_code == HTTPClient.RESPONSE_REQUEST_TIMEOUT or response_code == HTTPClient.RESPONSE_TOO_MANY_REQUESTS or response_code >= 500
static func normalize_ticket(payload: Dictionary) -> Dictionary: static func normalize_ticket(payload: Dictionary) -> Dictionary:
var result := payload.duplicate(true) var result := payload.duplicate(true)
if result.has("expires_at") and result["expires_at"] is String: if result.has("expires_at") and result["expires_at"] is String:
@@ -281,6 +298,9 @@ func _start_request(operation: String, method: HTTPClient.Method, path: String,
if err != OK: if err != OK:
_operation = "" _operation = ""
return err return err
if not idempotency_key.is_empty():
_last_mutation = {"operation": operation, "method": method, "path": path, "payload": payload.duplicate(true), "key": idempotency_key, "expected_revision": expected_revision}
_last_mutation_retryable = false
return OK return OK
@@ -288,6 +308,7 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head
var operation := _operation var operation := _operation
_operation = "" _operation = ""
if result != HTTPRequest.RESULT_SUCCESS: if result != HTTPRequest.RESULT_SUCCESS:
_last_mutation_retryable = _last_mutation.get("operation", "") == operation
if operation == "ranked_profile": if operation == "ranked_profile":
ranked_profile.set_error("Ranked profile request failed") ranked_profile.set_error("Ranked profile request failed")
elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover": elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover":
@@ -298,6 +319,7 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head
return return
var parsed = JSON.parse_string(body.get_string_from_utf8()) var parsed = JSON.parse_string(body.get_string_from_utf8())
if not parsed is Dictionary: if not parsed is Dictionary:
_last_mutation_retryable = _last_mutation.get("operation", "") == operation
if operation == "ranked_profile": if operation == "ranked_profile":
ranked_profile.set_error("Ranked profile returned invalid JSON") ranked_profile.set_error("Ranked profile returned invalid JSON")
elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover": elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover":
@@ -307,6 +329,7 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head
request_failed.emit(operation, response_code, "invalid JSON") request_failed.emit(operation, response_code, "invalid JSON")
return return
if response_code < 200 or response_code >= 300: if response_code < 200 or response_code >= 300:
_last_mutation_retryable = _last_mutation.get("operation", "") == operation and is_retryable_mutation_response(response_code)
var detail := String(parsed.get("error", "request rejected")) var detail := String(parsed.get("error", "request rejected"))
var recover_proposal_after_conflict := response_code == HTTPClient.RESPONSE_CONFLICT and (operation == "proposal_accept" or operation == "proposal_decline") and not state.proposal_id.is_empty() var recover_proposal_after_conflict := response_code == HTTPClient.RESPONSE_CONFLICT and (operation == "proposal_accept" or operation == "proposal_decline") and not state.proposal_id.is_empty()
if response_code == HTTPClient.RESPONSE_UNAUTHORIZED: if response_code == HTTPClient.RESPONSE_UNAUTHORIZED:
@@ -334,6 +357,7 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head
call_deferred("_run_pending_resync") call_deferred("_run_pending_resync")
return return
var payload: Dictionary = parsed var payload: Dictionary = parsed
_last_mutation_retryable = false
if operation == "steam_session": if operation == "steam_session":
var returned_token := String(payload.get("access_token", "")) var returned_token := String(payload.get("access_token", ""))
var returned_player_id := String(payload.get("player_id", "")) var returned_player_id := String(payload.get("player_id", ""))
@@ -50,6 +50,15 @@ func test_websocket_event_validation_requires_contract_specific_fields() -> void
assert_true(not ControlPlaneClient._valid_websocket_event(assignment), "incomplete assignment event is rejected") assert_true(not ControlPlaneClient._valid_websocket_event(assignment), "incomplete assignment event is rejected")
func test_retryable_mutation_policy_only_retries_safe_failures() -> void:
assert_true(ControlPlaneClient.is_retryable_mutation_response(0), "transport failure is retryable")
assert_true(ControlPlaneClient.is_retryable_mutation_response(408), "request timeout is retryable")
assert_true(ControlPlaneClient.is_retryable_mutation_response(429), "rate limit is retryable")
assert_true(ControlPlaneClient.is_retryable_mutation_response(503), "server failure is retryable")
assert_true(not ControlPlaneClient.is_retryable_mutation_response(401), "authentication failure is not blindly replayed")
assert_true(not ControlPlaneClient.is_retryable_mutation_response(409), "revision/idempotency conflict is not blindly replayed")
func test_assignment_endpoint_split_never_accepts_url_or_bad_port() -> void: func test_assignment_endpoint_split_never_accepts_url_or_bad_port() -> void:
var endpoint := ControlPlaneClient._split_assignment_endpoint("127.0.0.1:31001") var endpoint := ControlPlaneClient._split_assignment_endpoint("127.0.0.1:31001")
assert_eq(endpoint["host"], "127.0.0.1", "assignment host is separated from the port") assert_eq(endpoint["host"], "127.0.0.1", "assignment host is separated from the port")
+2
View File
@@ -1438,3 +1438,5 @@ The allocated supervisor now calls that shutdown acknowledgment during signal-bo
Allocator-selected region, build, protocol, and transport now travel with the allocation as Agones annotations and override stale child launch flags immediately before an allocated process starts. The overlay rejects control characters and preserves direct-server command behavior; focused supervisor/allocator tests cover precedence and annotation payloads, while live Agones passthrough remains an infrastructure gate. Allocator-selected region, build, protocol, and transport now travel with the allocation as Agones annotations and override stale child launch flags immediately before an allocated process starts. The overlay rejects control characters and preserves direct-server command behavior; focused supervisor/allocator tests cover precedence and annotation payloads, while live Agones passthrough remains an infrastructure gate.
The same allocation path now carries the matcher-selected playlist, preventing a ranked match from inheriting the Fleets casual default. Durable allocation claims return the playlist, the worker includes it in Fleet selection metadata, Agones copies it to the allocated GameServer, and the supervisor overrides `--playlist` before launch; the existing compatibility tests remain green. The same allocation path now carries the matcher-selected playlist, preventing a ranked match from inheriting the Fleets casual default. Durable allocation claims return the playlist, the worker includes it in Fleet selection metadata, Agones copies it to the allocated GameServer, and the supervisor overrides `--playlist` before launch; the existing compatibility tests remain green.
The Godot control-plane client now retains the exact last idempotent mutation and exposes `retry_last_mutation()` for transport, timeout, rate-limit, and 5xx failures. Retries reuse the original idempotency key and expected revision, while 401 and 409 responses remain non-retryable; the harness covers the policy boundary. This closes the local duplicate-action recovery mechanism for heartbeat/cancel/proposal calls, with broader live UI retry verification still remaining.