diff --git a/Game/scripts/server_boot.gd b/Game/scripts/server_boot.gd index b7f54759..55c371ee 100644 --- a/Game/scripts/server_boot.gd +++ b/Game/scripts/server_boot.gd @@ -3,6 +3,7 @@ extends Node 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") # Headless dedicated server entry point (task 1.6). Parses CLI args, hosts # via NetworkManager, logs structured lines, and watches for physics-tick @@ -29,8 +30,11 @@ var _last_physics_frame := 0 var config: ServerConfig = null var _watchdog_armed := false # skip the first _process(): engine startup scheduling can batch several physics frames before the first idle frame runs, which isn't a real overrun var _control: ServerControl = null +var _match_loop: ServerMatchLoop = null var _agones = null var _drain_requested := false +var _connection_reports_inflight: Dictionary = {} +var _connection_reports_complete: Dictionary = {} func _ready() -> void: @@ -89,6 +93,7 @@ func _ready() -> void: config.values["min-players"] = required_min_players(true, roster_tokens.size(), int(config.get_value("min-players"))) _control.name = "ServerControl" _control.drain_requested.connect(_on_drain_requested) + _control.initial_connect_ready.connect(_on_initial_connect_ready) get_tree().root.add_child.call_deferred(_control) var control_err := _control.start(int(config.get_value("readiness-port")), OS.get_environment(String(config.get_value("drain-token-env")))) if control_err != OK: @@ -134,6 +139,7 @@ func _ready() -> void: # it started. Same constraint the smoke-test hooks document. func _install_match_loop() -> void: var loop := ServerMatchLoop.new() + _match_loop = loop loop.name = "ServerMatchLoop" loop.min_players = int(config.get_value("min-players")) loop.start_countdown_seconds = float(config.get_value("start-countdown")) @@ -143,6 +149,10 @@ func _install_match_loop() -> void: loop.allocated_playlist = String(config.get_value("playlist")) loop.allocated_roster_size = MatchNet.assigned_player_slots().size() if loop.allocated_mode else 0 loop.allocated_arena_path = String(config.get_value("arena-path")) + # The backend's fair timeout starts only after durable assignment-ready. + # When a control plane is present, the supervisor arms this loop through + # the authenticated local control endpoint after that transition commits. + loop.allocated_admission_armed = not loop.allocated_mode or OS.get_environment("COSMIC_CLASH_INITIAL_CONNECT_SIGNAL_REQUIRED") != "1" get_tree().root.add_child.call_deferred(loop) @@ -178,6 +188,8 @@ 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: @@ -191,6 +203,60 @@ func _on_drain_requested() -> void: ServerLog.info("server_draining", {"reason": "control_request"}) +func _on_initial_connect_ready() -> void: + if _match_loop != null and is_instance_valid(_match_loop): + _match_loop.arm_allocated_admission() + 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}) + + +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) + + static func required_min_players(allocated: bool, roster_size: int, configured: int) -> int: if allocated and roster_size > 0: return roster_size diff --git a/Game/scripts/server_control.gd b/Game/scripts/server_control.gd index a4401c6e..532c9a17 100644 --- a/Game/scripts/server_control.gd +++ b/Game/scripts/server_control.gd @@ -6,6 +6,7 @@ extends Node # controlled termination. Direct/community servers do not start this node. signal drain_requested +signal initial_connect_ready var _listener := TCPServer.new() var _peers: Array = [] @@ -87,6 +88,18 @@ func _respond(peer: StreamPeerTCP, request: String) -> void: drain_requested.emit() status = 202 reason = "Accepted" + elif method == "POST" and path == "/initial-connect-ready": + var supplied := "" + for line in lines: + if line.begins_with("Authorization: Bearer "): + supplied = line.substr("Authorization: Bearer ".length()) + if _drain_token.is_empty() or not _constant_time_equal(supplied, _drain_token): + status = 401 + reason = "Unauthorized" + else: + initial_connect_ready.emit() + status = 202 + reason = "Accepted" else: status = 405 if method in ["GET", "POST"] else 400 reason = "Method Not Allowed" if status == 405 else "Bad Request" diff --git a/Game/scripts/server_match_loop.gd b/Game/scripts/server_match_loop.gd index 55ac39b4..8b9d0c34 100644 --- a/Game/scripts/server_match_loop.gd +++ b/Game/scripts/server_match_loop.gd @@ -48,6 +48,7 @@ var allocated_mode := false var allocated_playlist := "" var allocated_roster_size := 0 var allocated_arena_path := "" +var allocated_admission_armed := true var matches_completed := 0 var _countdown_started_ms := -1 @@ -74,6 +75,8 @@ func _process(_delta: float) -> void: func _poll_allocated_match_start(now: int) -> void: + if not allocated_admission_armed: + return if _allocated_connect_started_ms < 0: _allocated_connect_started_ms = now var connected := MatchNet.roster.size() @@ -95,14 +98,27 @@ func _poll_allocated_match_start(now: int) -> void: _poll_match_start(now) +func arm_allocated_admission() -> void: + allocated_admission_armed = true + _allocated_connect_started_ms = -1 + + static func allocated_initial_connect_action(playlist: String, elapsed_ms: int, connected: int, expected: int, has_team_zero: bool, has_team_one: bool) -> String: if elapsed_ms < 0 or connected < 0 or expected < 1: return ALLOCATED_CANCEL - if connected >= expected: - return ALLOCATED_READY if playlist == "ranked": + if expected != 6: + return ALLOCATED_CANCEL + if connected >= expected: + return ALLOCATED_READY return ALLOCATED_CANCEL if elapsed_ms >= 30000 else ALLOCATED_WAIT if playlist == "casual": + if expected < 2 or expected > 6: + return ALLOCATED_CANCEL + if connected >= expected: + if expected == 6: + return ALLOCATED_READY + return ALLOCATED_START_WITH_BOTS if has_team_zero and has_team_one else ALLOCATED_CANCEL if elapsed_ms < 60000: return ALLOCATED_WAIT return ALLOCATED_START_WITH_BOTS if connected >= 2 and has_team_zero and has_team_one else ALLOCATED_CANCEL diff --git a/Game/tests/cases/test_server_config.gd b/Game/tests/cases/test_server_config.gd index 85c797c3..3aa7d5b5 100644 --- a/Game/tests/cases/test_server_config.gd +++ b/Game/tests/cases/test_server_config.gd @@ -173,3 +173,12 @@ func test_allocated_start_floor_is_the_verified_roster_size() -> void: assert_eq(boot.required_min_players(true, 6, 1), 6, "allocated six-player roster cannot start with one player") assert_eq(boot.required_min_players(true, 2, 6), 2, "allocated casual roster uses its complete size") assert_eq(boot.required_min_players(false, 1, 1), 1, "direct server keeps its configured floor") + + +func test_connection_reporting_requires_safe_workload_configuration() -> void: + var boot = preload("res://scripts/server_boot.gd") + assert_true(boot.valid_connection_report_configuration("http://control-plane:8080", "signed-token", "match-1234567890", "server-123456789", "player-123456789"), "allocated workload configuration is accepted") + assert_true(not boot.valid_connection_report_configuration("http://control-plane:8080?token=leak", "signed-token", "match-1234567890", "server-123456789", "player-123456789"), "query-bearing control-plane URL is rejected") + assert_true(not boot.valid_connection_report_configuration("http://control-plane:8080", "token\nforged", "match-1234567890", "server-123456789", "player-123456789"), "header injection token is rejected") + assert_true(not boot.valid_connection_report_configuration("http://control-plane:8080", "signed-token", "short", "server-123456789", "player-123456789"), "non-opaque match identity is rejected") + assert_true(not boot.valid_connection_report_configuration("http://control-plane:8080", "signed-token", "match-1234567890", "server-123456789", "short"), "non-opaque player identity is rejected") diff --git a/Game/tests/cases/test_server_match_loop.gd b/Game/tests/cases/test_server_match_loop.gd index fcb7ac31..840e0334 100644 --- a/Game/tests/cases/test_server_match_loop.gd +++ b/Game/tests/cases/test_server_match_loop.gd @@ -6,6 +6,14 @@ func test_allocated_initial_connect_policy_has_explicit_boundaries() -> void: assert_eq(loop.allocated_initial_connect_action("ranked", 30000, 5, 6, true, true), loop.ALLOCATED_CANCEL, "ranked cancels at 30 seconds") assert_eq(loop.allocated_initial_connect_action("casual", 59999, 2, 6, true, true), loop.ALLOCATED_WAIT, "casual waits before 60 seconds") assert_eq(loop.allocated_initial_connect_action("casual", 60000, 2, 6, true, true), loop.ALLOCATED_START_WITH_BOTS, "casual starts with bots when both teams are represented") + assert_eq(loop.allocated_initial_connect_action("casual", 1000, 2, 2, true, true), loop.ALLOCATED_START_WITH_BOTS, "complete relaxed casual roster starts with disclosed bots immediately") + assert_eq(loop.allocated_initial_connect_action("casual", 1000, 2, 2, true, false), loop.ALLOCATED_CANCEL, "malformed relaxed casual roster fails closed") assert_eq(loop.allocated_initial_connect_action("casual", 60000, 2, 6, true, false), loop.ALLOCATED_CANCEL, "casual cancels when one team is empty") assert_eq(loop.allocated_initial_connect_action("casual", 1000, 6, 6, true, true), loop.ALLOCATED_READY, "complete roster is ready immediately") + assert_eq(loop.allocated_initial_connect_action("ranked", 1000, 5, 5, true, true), loop.ALLOCATED_CANCEL, "ranked cannot shrink its expected roster") assert_eq(loop.allocated_initial_connect_action("other", 0, 1, 6, true, true), loop.ALLOCATED_CANCEL, "unknown allocated playlist fails closed") + var instance = loop.new() + instance.allocated_admission_armed = false + instance.arm_allocated_admission() + assert_true(instance.allocated_admission_armed, "durable readiness signal arms the local timeout") + instance.free() diff --git a/Game/tests/server_control_smoke.gd b/Game/tests/server_control_smoke.gd index 294138e9..7088ba2d 100644 --- a/Game/tests/server_control_smoke.gd +++ b/Game/tests/server_control_smoke.gd @@ -12,6 +12,8 @@ func _init() -> void: quit(1) return control.set_process_ready(true) + control.set_meta("admission_armed", false) + control.initial_connect_ready.connect(func() -> void: control.set_meta("admission_armed", true)) await process_frame var ready_response := await _request("GET", "/ready", []) if ready_response != 200: @@ -23,6 +25,16 @@ func _init() -> void: printerr("unauthorized drain response was %d" % unauthorized) quit(1) return + var unauthorized_admission := await _request("POST", "/initial-connect-ready", ["Authorization: Bearer wrong"]) + if unauthorized_admission != 401 or bool(control.get_meta("admission_armed")): + printerr("unauthorized initial-connect response/state was %d/%s" % [unauthorized_admission, control.get_meta("admission_armed")]) + quit(1) + return + var admitted := await _request("POST", "/initial-connect-ready", ["Authorization: Bearer drain-secret"]) + if admitted != 202 or not bool(control.get_meta("admission_armed")): + printerr("authorized initial-connect response/state was %d/%s" % [admitted, control.get_meta("admission_armed")]) + quit(1) + return var drained := await _request("POST", "/drain", ["Authorization: Bearer drain-secret"]) if drained != 202 or not control.is_draining(): printerr("authorized drain response/state was %d/%s" % [drained, control.is_draining()]) diff --git a/deploy/k8s/base/fleet.yaml b/deploy/k8s/base/fleet.yaml index 9f17f6ae..c3192497 100644 --- a/deploy/k8s/base/fleet.yaml +++ b/deploy/k8s/base/fleet.yaml @@ -58,6 +58,7 @@ spec: - --sdk-base-url=http://127.0.0.1:9357 - --ready-url=http://127.0.0.1:7780/ready - --drain-url=http://127.0.0.1:7780/drain + - --initial-connect-ready-url=http://127.0.0.1:7780/initial-connect-ready - --drain-token-env=COSMIC_CLASH_DRAIN_TOKEN - --control-plane-url=http://control-plane.cosmic-clash.svc.cluster.local:8080 - --server-id-env=COSMIC_CLASH_SERVER_ID diff --git a/docs/MATCHMAKING.md b/docs/MATCHMAKING.md index f17a305c..da470ce6 100644 --- a/docs/MATCHMAKING.md +++ b/docs/MATCHMAKING.md @@ -208,6 +208,15 @@ allocation fails. Ordering ties use ticket ID. A ranked initial-connect no-show after accepting uses the ranked abandon cooldown ladder but never a rating loss because no rated match began. +The initial-connect clock begins only after the server's durable +`ASSIGNMENT_READY` transition. Player assignments are not exposed before that +gate. Each allocated server reports a successfully verified signed-roster +admission through the workload-authenticated, idempotent +`POST /servers/{serverId}/connect` boundary; PostgreSQL `connected_at` values, +not client claims, drive no-show reconciliation. The supervisor arms the game +process's matching local timeout through an authenticated loopback control call +only after the same transition commits. + ### Casual - Target six humans in 3v3. diff --git a/multiplayer-next.md b/multiplayer-next.md index e1f3decd..7344b346 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1228,7 +1228,7 @@ production fallback. | 8.32 `[D:8.2,8.26,8.30]` | **IN PROGRESS.** Provider-neutral FleetAutoscaler baseline preserves a two-process Ready buffer, caps warm capacity, and leaves Allocated scale-down independent of the Ready floor; Fleet image references remain digest-pinned for current/rollback pre-pull | `deploy/k8s/base/fleet-autoscaler.yaml` and manifest tests cover Fleet ownership, Buffer policy and floor/cap invariants; regional on-demand node pools/failure domains, pre-pull rollout, warm-allocation p95/p99 and N+1 certification remain | | 8.33 `[D:8.26,8.32]` | **IN PROGRESS.** Fleet scheduling now requires on-demand capacity and spreads Ready processes across zones with skew 1; the autoscaler preserves the two-process Ready floor | `deploy/k8s/base/fleet.yaml` and manifest tests reject interruptible placement and single-zone concentration structurally; regional node pools, forced node-loss testing and measured N+1 headroom remain | | 8.34 `[D:8.28,8.29]` | Native x86_64 benchmark of boot-to-process-ready and assignment-ready, p99 CPU/RSS/network and 60 Hz ticks; limits/node cap with 30% headroom | Measurements replace old estimates and certify density with no tick backlog | -| 8.35 `[D:8.17,8.19,8.20,8.30,8.31]` | **IN PROGRESS.** Pure Go initial-connect policy decides ranked 30 s no-show cancellation with abandon ladder and casual 60 s bot start only when each team has a human; empty-team casual allocations cancel | `server/domain/noshow.go` covers wait/deadline boundaries, deterministic no-show/innocent ordering, ranked cooldown history and no pre-live rating action; persistent ticket restoration, allocation shutdown, bot spawn and live integration remain | +| 8.35 `[D:8.17,8.19,8.20,8.30,8.31]` | **IN PROGRESS.** Allocated Godot servers now report each signed-roster admission through a match-bound workload-authenticated/idempotent API; PostgreSQL persists `connected_at`, starts the fair deadline at durable `ASSIGNMENT_READY`, and atomically starts complete rosters, applies ranked 30 s no-show cancellation/abandon ladders, or applies casual bot/cancel outcomes after 60 s. The maintenance role evaluates this path every second. Godot's local clock is armed only after the same durable readiness transition and applies the same complete/partial roster policy | Domain/store/API/supervisor/Godot tests cover forged workload/allocation/player bindings, replay after response loss, malformed rosters, complete ranked/casual starts, relaxed 2–5-human bot starts, canonical team/global-slot preservation, empty-team cancellation, stale-snapshot races, retryable datastore outages, and readiness-clock ordering. Migration `0010_initial_connect_ready_at.sql` gives deployed in-flight matches a fresh window. Live PostgreSQL execution, allocated process termination evidence, and real Agones multi-client verification remain | | 8.36 `[D:8.10,8.25,8.28,8.30]` | **IN PROGRESS.** Supervisor exposes an authenticated loopback-only drain request boundary that never places the token in command arguments/logs and rejects remote/partial/query-bearing configurations; the base now includes a two-Ready PodDisruptionBudget; `Supervisor.Run` and `cmd/game-server-supervisor` now orchestrate signal-bound drain-before-kill with a bounded grace deadline | `server/supervisor/`, `server/cmd/game-server-supervisor/`, `server_control.gd` and `deploy/k8s/base/game-server-pdb.yaml` cover bearer-token enforcement, loopback URL validation, secret-safe configuration, missing drain credentials, readiness transitions, Ready-floor disruption protection, graceful child exit after drain and force-kill of an unresponsive child; live 300 s/285 s lifecycle, PDB/Fleet drain and infrastructure-abort classification remain | | 8.37 `[D:8.5,8.10,8.25,8.26,8.31]` | Horizontally scaled primary control plane + warm standby, EU/NA fleets, RPO <=5 m/RTO <=30 m; signed result annotation/reconciliation preserves delivery outages | Restore/failover meet targets; live simulation continues; valid delayed result commits exactly once after recovery | | 8.38 `[D:7.7,8.26,8.36,8.37]` | Provider migration after Valve approves both providers' EU/NA POPs/certs and public UDP: restore, validate coordinator trust, switch allocations, drain old | Both geographies complete Hosted-SDR matches on new provider and no old-provider live match is terminated | @@ -1649,3 +1649,7 @@ Allocator probes now distinguish process liveness from useful progress. `/health The timeout boundary is enforced inside both network adapters as well as in the production allocator wiring: an `agones.Client` or game-server `Supervisor` constructed without an injected HTTP client now receives a ten-second client rather than Go's unbounded `http.DefaultClient`. This prevents alternate binaries, tests, and future callers from restoring an infinite GameServer, roster, registration, or SDK wait by omission. Control-plane probes now separate liveness from datastore readiness too. `/healthz` proves the process can serve without restarting it during a PostgreSQL outage; `/readyz` runs a one-second-bounded `PingContext` and the Deployment routes traffic only to replicas whose core durable store responds. Probe and metrics routes bypass the player request limiter, so operator-selected low limits cannot make Kubernetes evict a healthy replica. Missing checks, datastore errors, non-GET methods, and successful recovery are covered by API tests. + +The task 8.35 adversarial pass closed the previously disconnected initial-connect implementations. An accepted signed player now produces a workload-authenticated `POST /servers/{serverId}/connect` receipt bound to the exact allocation, match, server, participant, and unexpired assignment; durable replay survives a lost response and keys include the match so a later match cannot conflict. Unknown datastore failures return retryable 503 responses. Player assignment reads are hidden until the match has durably reached `ASSIGNMENT_READY`, and the supervisor now fails closed if that transition never commits. + +Initial-connect timing and topology now agree across every layer. Migration 0010 records `initial_connect_ready_at` at the assignment-ready transition instead of using match creation time; maintenance polls that path independently every second; and an authenticated loopback signal arms Godot's local timeout only after the durable transition. Complete rosters enter `LIVE` immediately, relaxed two-to-five-human casual rosters immediately fill their disclosed vacant slots with bots, and six-human casual no-shows use the 60-second policy. Casual lineup, reconnect, signed-roster, matcher, store, and Godot validation all use canonical global slots 0–2 for team 0 and 3–5 for team 1; the earlier alternating-slot bot layout has been removed. Focused Go tests, contract/migration/manifest checks, and the 204-test Godot harness pass; the committed PostgreSQL integration assertion remains unexecuted locally while Docker storage is exhausted. diff --git a/server/api/service.go b/server/api/service.go index e80153c2..08e82459 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -43,6 +43,9 @@ type ServerRegistrar interface { type ServerShutdowner interface { ShutdownServer(context.Context, domain.WorkloadBinding, string, string, time.Time) error } +type ServerConnectionRecorder interface { + RecordPlayerConnected(context.Context, domain.WorkloadBinding, string, string, time.Time) error +} type QueueBackend interface { Create(context.Context, string, string, string, domain.QueueSpec, time.Time) (domain.QueueTicket, error) @@ -118,6 +121,7 @@ type Service struct { ResultSubmitter ResultSubmitter ServerRegistrar ServerRegistrar ServerShutdowner ServerShutdowner + ServerConnections ServerConnectionRecorder Assignment AssignmentProvider Roster RosterProvider Now func() time.Time @@ -555,10 +559,9 @@ func (s *Service) contractAssignment(w http.ResponseWriter, r *http.Request) { func (s *Service) contractServerMutation(w http.ResponseWriter, r *http.Request) { // Unlike contractAssignment, the documented shape here is two segments - // (/servers/{serverId}/result, /servers/{serverId}/register, or - // /servers/{serverId}/shutdown) — rejecting + // (/servers/{serverId}/{result|register|roster|connect|shutdown}) — rejecting // any "/" would 404 every real call. Delegate shape validation to - // serverMutation, which already enforces exactly {id}/{result|register}. + // serverMutation, which already enforces the exact operation allowlist. path := strings.TrimPrefix(r.URL.Path, "/api/v1/servers/") parts := strings.Split(path, "/") if path == "" || len(parts) < 2 || !controlPlaneResourceIDRE.MatchString(parts[0]) { @@ -589,7 +592,7 @@ type serverRegistrationRequest struct { func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/servers/"), "/") - if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register" && parts[1] != "roster" && parts[1] != "shutdown") { + if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register" && parts[1] != "roster" && parts[1] != "shutdown" && parts[1] != "connect") { writeError(w, http.StatusNotFound, "not_found") return } @@ -597,7 +600,7 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusMethodNotAllowed, "method_not_allowed") return } - if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) || (parts[1] == "roster" && s.Roster == nil) || (parts[1] == "shutdown" && s.ServerShutdowner == nil) { + if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) || (parts[1] == "roster" && s.Roster == nil) || (parts[1] == "shutdown" && s.ServerShutdowner == nil) || (parts[1] == "connect" && s.ServerConnections == nil) { writeError(w, http.StatusServiceUnavailable, "server_unavailable") return } @@ -663,6 +666,33 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) return } + if parts[1] == "connect" { + var input struct { + PlayerID string `json:"player_id"` + } + if !decodeBody(w, r, &input) { + return + } + if !controlPlaneResourceIDRE.MatchString(input.PlayerID) { + writeError(w, http.StatusUnprocessableEntity, "invalid_request") + return + } + if err := s.ServerConnections.RecordPlayerConnected(r.Context(), binding, input.PlayerID, key, now); err != nil { + if errors.Is(err, domain.ErrConflict) { + writeError(w, http.StatusConflict, "conflict") + } else { + // The request has already passed schema and workload checks. An + // unknown recorder error is infrastructure failure, not a terminal + // client fault; 503 keeps the game server's bounded retry alive. + writeError(w, http.StatusServiceUnavailable, "server_unavailable") + } + s.logEvent(observability.Event{Event: "server_connect", MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now}) + return + } + s.logEvent(observability.Event{Event: "server_connect", MatchID: binding.MatchID, ServerID: parts[0], Stage: "connected", OccurredAt: now, Fields: map[string]any{"player_id": input.PlayerID}}) + w.WriteHeader(http.StatusNoContent) + return + } if parts[1] == "shutdown" { var input struct { Reason string `json:"reason"` diff --git a/server/api/service_test.go b/server/api/service_test.go index 92836c63..8a082500 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -70,6 +70,20 @@ type serverShutdownerSpy struct { err error } +type serverConnectionSpy struct { + calls int + binding domain.WorkloadBinding + playerID string + key string + err error +} + +func (s *serverConnectionSpy) RecordPlayerConnected(_ context.Context, binding domain.WorkloadBinding, playerID, key string, _ time.Time) error { + s.calls++ + s.binding, s.playerID, s.key = binding, playerID, key + return s.err +} + func (s *serverShutdownerSpy) ShutdownServer(_ context.Context, binding domain.WorkloadBinding, reason, key string, _ time.Time) error { s.calls++ s.binding, s.reason, s.key = binding, reason, key @@ -1458,6 +1472,52 @@ func TestServerShutdownAPIRequiresBoundWorkloadAndDelegatesAcknowledgement(t *te response.Body.Close() } +func TestServerConnectionAPIRequiresBoundWorkloadAndOpaqueAssignedPlayer(t *testing.T) { + now := time.Unix(1000, 0).UTC() + binding := domain.WorkloadBinding{AllocationID: "allocation-123456", MatchID: "match-1234567890", ServerID: "server-123456789"} + recorder := &serverConnectionSpy{} + service := &Service{Now: func() time.Time { return now }, WorkloadVerify: func(token string, _ time.Time) (domain.WorkloadBinding, error) { + if token != "workload-token" { + return domain.WorkloadBinding{}, errors.New("bad token") + } + return binding, nil + }, ServerConnections: recorder} + server := httptest.NewServer(service.Handler()) + defer server.Close() + + request := func(serverID, playerID, token, key string) int { + body := fmt.Sprintf(`{"player_id":%q}`, playerID) + req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/servers/"+serverID+"/connect", strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Idempotency-Key", key) + response, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + return response.StatusCode + } + if got := request(binding.ServerID, "player-123456789", "workload-token", "connect-player-123456789"); got != http.StatusNoContent { + t.Fatalf("connection status = %d", got) + } + if recorder.calls != 1 || recorder.binding != binding || recorder.playerID != "player-123456789" || recorder.key != "connect-player-123456789" { + t.Fatalf("connection receipt = %+v", recorder) + } + if got := request("server-000000000", "player-123456789", "workload-token", "connect-player-123456789"); got != http.StatusUnauthorized { + t.Fatalf("wrong server status = %d", got) + } + if got := request(binding.ServerID, "short", "workload-token", "connect-player-short-123"); got != http.StatusUnprocessableEntity { + t.Fatalf("short player status = %d", got) + } + if recorder.calls != 1 { + t.Fatalf("invalid receipts reached backend: %d", recorder.calls) + } + recorder.err = errors.New("database unavailable") + if got := request(binding.ServerID, "player-123456789", "workload-token", "connect-player-retry-123"); got != http.StatusServiceUnavailable { + t.Fatalf("recorder outage status = %d, want retryable 503", got) + } +} + func TestMetricsEndpointExportsBoundedAPILatencyAndSkipsItsOwnScrape(t *testing.T) { metrics := observability.NewMetrics() service := &Service{Metrics: metrics, Now: time.Now} diff --git a/server/api/store_adapters.go b/server/api/store_adapters.go index 51052e82..68405f05 100644 --- a/server/api/store_adapters.go +++ b/server/api/store_adapters.go @@ -87,6 +87,19 @@ func ServerShutdownerFromStore(db *sql.DB) ServerShutdowner { return postgresServerShutdowner{db: db} } +type postgresServerConnections struct{ db *sql.DB } + +func (p postgresServerConnections) RecordPlayerConnected(ctx context.Context, binding domain.WorkloadBinding, playerID, idempotencyKey string, now time.Time) error { + return store.RecordPlayerConnected(ctx, p.db, binding, playerID, idempotencyKey, now) +} + +func ServerConnectionsFromStore(db *sql.DB) ServerConnectionRecorder { + if db == nil { + return nil + } + return postgresServerConnections{db: db} +} + // WorkloadVerifierFromSignedToken builds WorkloadVerify from a control-plane // -owned signed token instead of a Kubernetes-projected JWT (see // workload/signed_token.go for why: it needs no live cluster to verify). diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index 893b4151..3f44eb37 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -134,6 +134,7 @@ func newAPIService(db *sql.DB, workloadSecret string, indexes ...api.CandidateIn ProposalPromoter: api.ProposalPromoterFromStore(db), ServerRegistrar: api.ServerRegistrarFromStore(db), ServerShutdowner: api.ServerShutdownerFromStore(db), + ServerConnections: api.ServerConnectionsFromStore(db), ResultSubmitter: store.PostgresResults{DB: db}, RankedProfileProvider: store.PostgresRankedProfiles{DB: db}, TierPolicy: domain.DefaultTierPolicy(), diff --git a/server/cmd/game-server-supervisor/main.go b/server/cmd/game-server-supervisor/main.go index d3aaefd8..06661cf1 100644 --- a/server/cmd/game-server-supervisor/main.go +++ b/server/cmd/game-server-supervisor/main.go @@ -39,6 +39,7 @@ func main() { sdkBaseURL := options.String("sdk-base-url", "", "Agones SDK REST base URL; empty enables direct mode") readyURL := options.String("ready-url", "", "explicit process-ready probe URL") drainURL := options.String("drain-url", "", "loopback drain URL") + admissionURL := options.String("initial-connect-ready-url", "", "authenticated loopback URL that starts the initial-connect clock after durable assignment readiness") drainTokenEnv := options.String("drain-token-env", "COSMIC_CLASH_DRAIN_TOKEN", "environment variable containing the drain bearer token") transport := options.String("transport", "enet", "enet or steam_sdr") grace := options.Duration("drain-grace", supervisor.DefaultDrainGrace, "maximum graceful drain duration") @@ -64,6 +65,7 @@ func main() { SDKBaseURL: *sdkBaseURL, ReadyURL: *readyURL, DrainURL: *drainURL, + AdmissionURL: *admissionURL, DrainToken: token, Transport: *transport, ReadyTimeout: 30 * time.Second, diff --git a/server/cmd/maintenance/main.go b/server/cmd/maintenance/main.go index b46ff1eb..44960970 100644 --- a/server/cmd/maintenance/main.go +++ b/server/cmd/maintenance/main.go @@ -20,6 +20,7 @@ func main() { dsn := flag.String("dsn", os.Getenv("COSMIC_CLASH_POSTGRES_DSN"), "PostgreSQL connection string") migrationDir := flag.String("migrations", "migrations", "directory containing numbered SQL migrations") interval := flag.Duration("interval", time.Minute, "maintenance poll interval") + initialConnectInterval := flag.Duration("initial-connect-interval", time.Second, "initial-connect reconciliation poll interval") batch := flag.Int("batch", 100, "maximum player rollovers per pass") stalledAllocationDeadline := flag.Duration("stalled-allocation-deadline", 2*time.Minute, "reclaim a match stuck in ALLOCATING/PROCESS_READY/ASSIGNMENT_READY (server crashed or was reclaimed before registering) after this long, requeuing every participant without penalty") stalledAllocationBatch := flag.Int("stalled-allocation-batch", 100, "maximum stalled matches reclaimed per pass") @@ -28,7 +29,7 @@ func main() { if *dsn == "" { fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required") } - if *interval <= 0 || *batch < 1 || *batch > 1000 { + if *interval <= 0 || *initialConnectInterval <= 0 || *batch < 1 || *batch > 1000 { fatalf("invalid interval or batch") } if *stalledAllocationDeadline <= 0 || *stalledAllocationBatch < 1 || *stalledAllocationBatch > 1000 { @@ -52,8 +53,7 @@ func main() { } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - for { - now := time.Now().UTC() + runGeneral := func(now time.Time) { count, err := store.RolloverDueSeasons(ctx, db, now, *batch) if err != nil { fatalf("season maintenance: %v", err) @@ -68,6 +68,8 @@ func main() { if reclaimed > 0 { log.Printf("reclaimed %d stalled allocations, requeuing their participants", reclaimed) } + } + runInitialConnect := func(now time.Time) { reconciled, err := store.ReconcileInitialConnect(ctx, db, now, *initialConnectBatch) if err != nil { fatalf("initial-connect maintenance: %v", err) @@ -75,12 +77,22 @@ func main() { if reconciled > 0 { log.Printf("reconciled %d initial-connect outcomes", reconciled) } - timer := time.NewTimer(*interval) + } + + runGeneral(time.Now().UTC()) + runInitialConnect(time.Now().UTC()) + generalTicker := time.NewTicker(*interval) + initialConnectTicker := time.NewTicker(*initialConnectInterval) + defer generalTicker.Stop() + defer initialConnectTicker.Stop() + for { select { case <-ctx.Done(): - timer.Stop() return - case <-timer.C: + case now := <-generalTicker.C: + runGeneral(now.UTC()) + case now := <-initialConnectTicker.C: + runInitialConnect(now.UTC()) } } } diff --git a/server/cmd/testkit-api/main.go b/server/cmd/testkit-api/main.go index 5557a0b1..6b5f7147 100644 --- a/server/cmd/testkit-api/main.go +++ b/server/cmd/testkit-api/main.go @@ -64,6 +64,7 @@ func main() { ProposalPromoter: api.ProposalPromoterFromStore(db), ServerRegistrar: api.ServerRegistrarFromStore(db), ServerShutdowner: api.ServerShutdownerFromStore(db), + ServerConnections: api.ServerConnectionsFromStore(db), ResultSubmitter: store.PostgresResults{DB: db}, RankedProfileProvider: store.PostgresRankedProfiles{DB: db}, TierPolicy: domain.DefaultTierPolicy(), diff --git a/server/contracts/v1/openapi.json b/server/contracts/v1/openapi.json index 4b5bd971..efc0fb9e 100644 --- a/server/contracts/v1/openapi.json +++ b/server/contracts/v1/openapi.json @@ -50,6 +50,9 @@ "/servers/{serverId}/register": { "post": {"security": [{"serverCredential": []}], "operationId": "registerServer", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerRegistration"}}}}, "responses": {"204": {"description": "Registered"}, "409": {"$ref": "#/components/responses/Conflict"}}} }, + "/servers/{serverId}/connect": { + "post": {"security": [{"serverCredential": []}], "operationId": "recordPlayerConnected", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnection"}}}}, "responses": {"204": {"description": "Connection recorded"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}, "503": {"$ref": "#/components/responses/Unavailable"}}} + }, "/servers/{serverId}/result": { "post": {"security": [{"serverCredential": []}], "operationId": "submitMatchResult", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/MatchResult"}}}}, "responses": {"202": {"description": "Result accepted"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}}} }, @@ -92,6 +95,7 @@ "ProposalParticipant": {"type": "object", "required": ["player_id", "response", "team", "slot"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "response": {"type": "string", "enum": ["PENDING", "ACCEPTED", "DECLINED", "TIMED_OUT"]}, "team": {"type": "integer", "minimum": 0, "maximum": 1}, "slot": {"type": "integer", "minimum": 0, "maximum": 5}}}, "Assignment": {"type": "object", "required": ["match_id", "server_id", "player_id", "slot", "expires_at", "protocol_version", "transport", "endpoint", "join_authorisation"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "server_id": {"$ref": "#/components/schemas/OpaqueId"}, "player_id": {"$ref": "#/components/schemas/OpaqueId"}, "slot": {"type": "integer", "minimum": 0, "maximum": 5}, "expires_at": {"type": "string", "format": "date-time"}, "protocol_version": {"type": "integer", "minimum": 1}, "transport": {"type": "string", "enum": ["steam_sdr", "enet"]}, "endpoint": {"type": "string", "minLength": 3, "maxLength": 256}, "join_authorisation": {"type": "string"}}}, "ServerRegistration": {"type": "object", "required": ["match_id", "protocol_version", "image_digest", "assignment_ready"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "protocol_version": {"type": "integer", "minimum": 1}, "image_digest": {"type": "string", "pattern": "^sha256:[a-f0-9]{64}$"}, "assignment_ready": {"type": "boolean"}}}, + "ServerConnection": {"type": "object", "required": ["player_id"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}}}, "ServerShutdown": {"type": "object", "required": ["reason"], "additionalProperties": false, "properties": {"reason": {"type": "string", "minLength": 1, "maxLength": 96}}}, "MatchResult": {"type": "object", "required": ["match_id", "result_nonce", "score", "integrity_state"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "result_nonce": {"type": "string", "minLength": 16, "maxLength": 128}, "score": {"type": "object", "required": ["team_0", "team_1"], "additionalProperties": false, "properties": {"team_0": {"type": "integer", "minimum": 0}, "team_1": {"type": "integer", "minimum": 0}}}, "integrity_state": {"type": "string", "enum": ["CERTIFIED", "SUPPRESSED", "REVIEW"]}}} } diff --git a/server/contracts/v1/test_contracts.py b/server/contracts/v1/test_contracts.py index d3248f9e..fbfbe9be 100644 --- a/server/contracts/v1/test_contracts.py +++ b/server/contracts/v1/test_contracts.py @@ -27,7 +27,7 @@ class ContractTest(unittest.TestCase): "createSteamSession", "getProfile", "createQueueTicket", "heartbeatQueueTicket", "cancelQueueTicket", "acceptProposal", "declineProposal", "getAssignment", "registerServer", - "submitMatchResult", "getRankedProfile", + "recordPlayerConnected", "submitMatchResult", "getRankedProfile", } <= operations) def test_ranked_profile_contract_is_authoritative_and_optional_season_metadata(self): diff --git a/server/domain/casual.go b/server/domain/casual.go index eaf9d8c1..903df7de 100644 --- a/server/domain/casual.go +++ b/server/domain/casual.go @@ -30,21 +30,21 @@ func BuildCasualLineup(participants []ConnectParticipant) ([]CasualSlot, error) teamHuman := map[int]bool{} lineup := make([]CasualSlot, 6) usedSlots := make(map[int]bool) - for i, participant := range participants { - if participant.PlayerID == "" || participant.Team < 0 || participant.Team > 1 || seen[participant.PlayerID] || usedSlots[i] { + for _, participant := range participants { + if participant.PlayerID == "" || participant.Team < 0 || participant.Team > 1 || participant.Slot < 0 || participant.Slot > 5 || participant.Slot/3 != participant.Team || seen[participant.PlayerID] || usedSlots[participant.Slot] { return nil, fmt.Errorf("invalid casual participant") } seen[participant.PlayerID] = true - usedSlots[i] = true + usedSlots[participant.Slot] = true teamHuman[participant.Team] = true - lineup[i] = CasualSlot{Slot: i, Team: participant.Team, PlayerID: participant.PlayerID} + lineup[participant.Slot] = CasualSlot{Slot: participant.Slot, Team: participant.Team, PlayerID: participant.PlayerID} } if !teamHuman[0] || !teamHuman[1] { return nil, fmt.Errorf("casual lineup requires one human on each team") } for i := range lineup { if lineup[i].PlayerID == "" { - lineup[i] = CasualSlot{Slot: i, Team: i % 2, PlayerID: fmt.Sprintf("bot-slot-%d", i), IsBot: true} + lineup[i] = CasualSlot{Slot: i, Team: i / 3, PlayerID: fmt.Sprintf("bot-slot-%d", i), IsBot: true} } } return lineup, nil diff --git a/server/domain/casual_test.go b/server/domain/casual_test.go index 64505aa3..91326379 100644 --- a/server/domain/casual_test.go +++ b/server/domain/casual_test.go @@ -3,11 +3,11 @@ package domain import "testing" func TestCasualLineupUsesBotsOnlyForMissingSlotsAndRequiresBothTeams(t *testing.T) { - lineup, err := BuildCasualLineup([]ConnectParticipant{{PlayerID: "p2", Team: 1}, {PlayerID: "p1", Team: 0}}) - if err != nil || len(lineup) != 6 || lineup[0].IsBot || lineup[1].IsBot || !lineup[2].IsBot || lineup[2].Team != 0 { + lineup, err := BuildCasualLineup([]ConnectParticipant{{PlayerID: "p2", Team: 1, Slot: 3}, {PlayerID: "p1", Team: 0, Slot: 0}}) + if err != nil || len(lineup) != 6 || lineup[0].IsBot || lineup[3].IsBot || !lineup[2].IsBot || lineup[2].Team != 0 || !lineup[5].IsBot || lineup[5].Team != 1 { t.Fatalf("casual lineup = %+v err=%v", lineup, err) } - if _, err := BuildCasualLineup([]ConnectParticipant{{PlayerID: "p1", Team: 0}, {PlayerID: "p2", Team: 0}}); err == nil { + if _, err := BuildCasualLineup([]ConnectParticipant{{PlayerID: "p1", Team: 0, Slot: 0}, {PlayerID: "p2", Team: 0, Slot: 1}}); err == nil { t.Fatal("lineup without a human on team 1 was accepted") } } diff --git a/server/domain/formation.go b/server/domain/formation.go index a18d1319..e9525dce 100644 --- a/server/domain/formation.go +++ b/server/domain/formation.go @@ -32,11 +32,11 @@ func PrepareProposal(id string, playlist Playlist, formation MatchFormation, ran switch playlist { case Casual: participants := make([]ConnectParticipant, 0, len(formation.Selection.Players)) - for _, player := range formation.Teams.Team0 { - participants = append(participants, ConnectParticipant{PlayerID: player.PlayerID, Team: 0}) + for index, player := range formation.Teams.Team0 { + participants = append(participants, ConnectParticipant{PlayerID: player.PlayerID, Team: 0, Slot: index}) } - for _, player := range formation.Teams.Team1 { - participants = append(participants, ConnectParticipant{PlayerID: player.PlayerID, Team: 1}) + for index, player := range formation.Teams.Team1 { + participants = append(participants, ConnectParticipant{PlayerID: player.PlayerID, Team: 1, Slot: 3 + index}) } var err error lineup, err = BuildCasualLineup(participants) diff --git a/server/domain/noshow.go b/server/domain/noshow.go index afdbf934..27c9e5df 100644 --- a/server/domain/noshow.go +++ b/server/domain/noshow.go @@ -15,6 +15,7 @@ const ( type ConnectParticipant struct { PlayerID string Team int + Slot int Connected bool } @@ -22,6 +23,7 @@ type InitialConnectAction string const ( InitialConnectWait InitialConnectAction = "WAIT" + InitialConnectStart InitialConnectAction = "START" InitialConnectCancel InitialConnectAction = "CANCEL" InitialConnectStartWithBot InitialConnectAction = "START_WITH_BOTS" ) @@ -57,6 +59,9 @@ func PlanInitialConnect(playlist Playlist, readyAt, now time.Time, participants switch decision.Action { case InitialConnectWait: return plan, nil + case InitialConnectStart: + plan.MatchState = Live + return plan, nil case InitialConnectCancel: plan.MatchState = Cancelled return plan, nil @@ -86,16 +91,18 @@ func EvaluateInitialConnect(playlist Playlist, readyAt, now time.Time, participa if playlist != Ranked && playlist != Casual || readyAt.IsZero() || len(participants) == 0 { return InitialConnectDecision{}, fmt.Errorf("invalid initial-connect policy input") } - if now.Before(readyAt.Add(InitialConnectWindow)) { - return InitialConnectDecision{Action: InitialConnectWait}, nil + if playlist == Ranked && len(participants) != 6 || playlist == Casual && (len(participants) < 2 || len(participants) > 6) { + return InitialConnectDecision{}, fmt.Errorf("invalid initial-connect roster size") } missing := make([]ConnectParticipant, 0) connected := make([]string, 0) teamConnected := map[int]bool{} + seen := make(map[string]bool, len(participants)) for _, participant := range participants { - if participant.PlayerID == "" || participant.Team < 0 { + if participant.PlayerID == "" || participant.Team < 0 || participant.Team > 1 || participant.Slot < 0 || participant.Slot > 5 || participant.Slot/3 != participant.Team || seen[participant.PlayerID] { return InitialConnectDecision{}, fmt.Errorf("invalid participant") } + seen[participant.PlayerID] = true if participant.Connected { connected = append(connected, participant.PlayerID) teamConnected[participant.Team] = true @@ -103,10 +110,19 @@ func EvaluateInitialConnect(playlist Playlist, readyAt, now time.Time, participa missing = append(missing, participant) } } - if playlist == Ranked { - if len(participants) != 6 { - return InitialConnectDecision{}, fmt.Errorf("ranked requires six participants") + if len(missing) == 0 { + if playlist == Casual && len(participants) < 6 { + if !teamConnected[0] || !teamConnected[1] { + return InitialConnectDecision{}, fmt.Errorf("casual bot roster requires a human on each team") + } + return InitialConnectDecision{Action: InitialConnectStartWithBot, Innocent: sortedIDs(connected)}, nil } + return InitialConnectDecision{Action: InitialConnectStart, Innocent: sortedIDs(connected)}, nil + } + if now.Before(readyAt.Add(InitialConnectWindow)) { + return InitialConnectDecision{Action: InitialConnectWait}, nil + } + if playlist == Ranked { return InitialConnectDecision{Action: InitialConnectCancel, NoShows: rankedNoShows(missing, now, priorAbandons), Innocent: sortedIDs(connected)}, nil } if now.Before(readyAt.Add(CasualBotStartAfter)) { diff --git a/server/domain/noshow_test.go b/server/domain/noshow_test.go index 92c019f7..a60e6de0 100644 --- a/server/domain/noshow_test.go +++ b/server/domain/noshow_test.go @@ -12,7 +12,7 @@ func sixConnectParticipants(connected ...int) []ConnectParticipant { } result := make([]ConnectParticipant, 6) for i := range result { - result[i] = ConnectParticipant{PlayerID: string(rune('a' + i)), Team: i % 2, Connected: set[i]} + result[i] = ConnectParticipant{PlayerID: string(rune('a' + i)), Team: i / 3, Slot: i, Connected: set[i]} } return result } @@ -25,9 +25,41 @@ func TestRankedInitialNoShowCancelsWithoutRatingPenalty(t *testing.T) { } } +func TestCompleteRosterStartsImmediatelyForEitherPlaylist(t *testing.T) { + readyAt := time.Unix(1000, 0) + participants := sixConnectParticipants(0, 1, 2, 3, 4, 5) + for _, playlist := range []Playlist{Ranked, Casual} { + plan, err := PlanInitialConnect(playlist, readyAt, readyAt.Add(time.Second), participants, nil) + if err != nil || plan.Action != InitialConnectStart || plan.MatchState != Live || len(plan.Connected) != 6 || len(plan.NoShows) != 0 || len(plan.CasualLineup) != 0 { + t.Fatalf("%s complete-roster plan = %+v err=%v", playlist, plan, err) + } + } +} + +func TestCompleteRelaxedCasualRosterStartsImmediatelyWithBots(t *testing.T) { + readyAt := time.Unix(1000, 0) + participants := []ConnectParticipant{{PlayerID: "a", Team: 0, Slot: 0, Connected: true}, {PlayerID: "d", Team: 1, Slot: 3, Connected: true}} + plan, err := PlanInitialConnect(Casual, readyAt, readyAt.Add(time.Second), participants, nil) + if err != nil || plan.Action != InitialConnectStartWithBot || plan.MatchState != Live || len(plan.Connected) != 2 || len(plan.NoShows) != 0 || len(plan.CasualLineup) != 6 { + t.Fatalf("relaxed casual plan = %+v err=%v", plan, err) + } +} + +func TestInitialConnectRejectsMalformedRosterBeforeStarting(t *testing.T) { + readyAt := time.Unix(1000, 0) + if _, err := EvaluateInitialConnect(Ranked, readyAt, readyAt, sixConnectParticipants(0, 1, 2, 3, 4)[:5], nil); err == nil { + t.Fatal("five-player ranked roster accepted") + } + duplicate := sixConnectParticipants(0, 1, 2, 3, 4, 5) + duplicate[5].PlayerID = duplicate[0].PlayerID + if _, err := EvaluateInitialConnect(Casual, readyAt, readyAt, duplicate, nil); err == nil { + t.Fatal("duplicate player accepted") + } +} + func TestCasualWaitsThenStartsWithBotsOnlyWithHumanOnEachTeam(t *testing.T) { readyAt := time.Unix(1000, 0) - participants := sixConnectParticipants(0, 1) + participants := sixConnectParticipants(0, 3) if decision, err := EvaluateInitialConnect(Casual, readyAt, readyAt.Add(45*time.Second), participants, nil); err != nil || decision.Action != InitialConnectWait { t.Fatalf("casual early decision = %+v err=%v", decision, err) } @@ -44,7 +76,7 @@ func TestCasualWaitsThenStartsWithBotsOnlyWithHumanOnEachTeam(t *testing.T) { func TestPlanInitialConnectMakesLifecycleActionExplicit(t *testing.T) { readyAt := time.Unix(1000, 0) - participants := sixConnectParticipants(0, 1) + participants := sixConnectParticipants(0, 3) plan, err := PlanInitialConnect(Casual, readyAt, readyAt.Add(CasualBotStartAfter), participants, nil) if err != nil || plan.Action != InitialConnectStartWithBot || plan.MatchState != Live || len(plan.CasualLineup) != 6 || len(plan.NoShows) != 4 { t.Fatalf("casual initial-connect plan = %+v err=%v", plan, err) diff --git a/server/domain/reconnect.go b/server/domain/reconnect.go index c2579e59..d91f3403 100644 --- a/server/domain/reconnect.go +++ b/server/domain/reconnect.go @@ -78,7 +78,7 @@ func NewRankedConnections(matchID, serverID, protocol string, players []JoinAuth } func (r *RankedConnections) validate(auth JoinAuthorisation, now time.Time) error { - if auth.MatchID != r.MatchID || auth.ServerID != r.ServerID || auth.Protocol != r.Protocol || auth.PlayerID == "" || auth.SteamID == "" || auth.Slot < 0 || auth.Team < 0 || auth.ExpiresAt.IsZero() { + if auth.MatchID != r.MatchID || auth.ServerID != r.ServerID || auth.Protocol != r.Protocol || auth.PlayerID == "" || auth.SteamID == "" || auth.Slot < 0 || auth.Slot > 5 || auth.Team < 0 || auth.Team > 1 || auth.Slot/3 != auth.Team || auth.ExpiresAt.IsZero() { return ErrJoinAuthorisation } if !now.IsZero() && !now.Before(auth.ExpiresAt) { diff --git a/server/domain/reconnect_test.go b/server/domain/reconnect_test.go index 3abf8644..1409eeb7 100644 --- a/server/domain/reconnect_test.go +++ b/server/domain/reconnect_test.go @@ -11,7 +11,7 @@ import ( func testRoster(now time.Time) []JoinAuthorisation { roster := make([]JoinAuthorisation, 6) for i := range roster { - roster[i] = JoinAuthorisation{MatchID: "match-1", ServerID: "server-1", Protocol: "v1", PlayerID: string(rune('a' + i)), SteamID: string(rune('A' + i)), Slot: i, Team: i % 2, Generation: 1, ExpiresAt: now.Add(time.Hour)} + roster[i] = JoinAuthorisation{MatchID: "match-1", ServerID: "server-1", Protocol: "v1", PlayerID: string(rune('a' + i)), SteamID: string(rune('A' + i)), Slot: i, Team: i / 3, Generation: 1, ExpiresAt: now.Add(time.Hour)} } return roster } @@ -76,6 +76,15 @@ func TestRankedRosterRejectsDuplicateSlots(t *testing.T) { } } +func TestRankedRosterRejectsTeamSlotMismatch(t *testing.T) { + now := time.Unix(1000, 0) + roster := testRoster(now) + roster[3].Team = 0 + if _, err := NewRankedConnections("match-1", "server-1", "v1", roster); !errors.Is(err, ErrJoinAuthorisation) { + t.Fatalf("team/slot mismatch accepted: %v", err) + } +} + func TestRankedAbandonCooldownUsesRollingSevenDayLadder(t *testing.T) { now := time.Unix(1000, 0) r, err := NewRankedConnections("match-1", "server-1", "v1", testRoster(now)) diff --git a/server/migrations/0010_initial_connect_ready_at.sql b/server/migrations/0010_initial_connect_ready_at.sql new file mode 100644 index 00000000..16ec7091 --- /dev/null +++ b/server/migrations/0010_initial_connect_ready_at.sql @@ -0,0 +1,13 @@ +ALTER TABLE matches + ADD COLUMN initial_connect_ready_at TIMESTAMPTZ; + +-- Existing in-flight matches receive a fresh, fair connection window when +-- this migration is deployed. Future rows are stamped by the transition to +-- ASSIGNMENT_READY, not by match creation or provider allocation. +UPDATE matches +SET initial_connect_ready_at = now() +WHERE state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING'); + +ALTER TABLE matches + ADD CONSTRAINT matches_initial_connect_ready_at + CHECK (state NOT IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING') OR initial_connect_ready_at IS NOT NULL) NOT VALID; diff --git a/server/migrations/down/0010_initial_connect_ready_at.sql b/server/migrations/down/0010_initial_connect_ready_at.sql new file mode 100644 index 00000000..af0d6180 --- /dev/null +++ b/server/migrations/down/0010_initial_connect_ready_at.sql @@ -0,0 +1,3 @@ +ALTER TABLE matches + DROP CONSTRAINT IF EXISTS matches_initial_connect_ready_at, + DROP COLUMN IF EXISTS initial_connect_ready_at; diff --git a/server/migrations/test_migration.py b/server/migrations/test_migration.py index a85c0805..b083adaa 100644 --- a/server/migrations/test_migration.py +++ b/server/migrations/test_migration.py @@ -9,6 +9,7 @@ ASSIGNMENTS_SQL = (Path(__file__).parent / "0002_assignments.sql").read_text() QUOTAS_SQL = (Path(__file__).parent / "0007_allocation_quotas.sql").read_text() ARENAS_SQL = (Path(__file__).parent / "0008_match_arena_paths.sql").read_text() ALLOCATION_ARENAS_SQL = (Path(__file__).parent / "0009_allocation_arena_paths.sql").read_text() +INITIAL_CONNECT_READY_SQL = (Path(__file__).parent / "0010_initial_connect_ready_at.sql").read_text() class MigrationTest(unittest.TestCase): @@ -73,6 +74,10 @@ class MigrationTest(unittest.TestCase): self.assertIn("ALTER TABLE allocations", ALLOCATION_ARENAS_SQL) self.assertIn("ADD COLUMN arena_path TEXT", ALLOCATION_ARENAS_SQL) + def test_initial_connect_window_starts_at_assignment_readiness(self): + self.assertIn("ADD COLUMN initial_connect_ready_at TIMESTAMPTZ", INITIAL_CONNECT_READY_SQL) + self.assertIn("state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING')", INITIAL_CONNECT_READY_SQL) + if __name__ == "__main__": unittest.main() diff --git a/server/store/allocation_match_sql.go b/server/store/allocation_match_sql.go index 7cbb0a27..b16e28d9 100644 --- a/server/store/allocation_match_sql.go +++ b/server/store/allocation_match_sql.go @@ -66,7 +66,9 @@ WHERE match_id = $1 AND state = 'ALLOCATING' AND allocation_id = $2 AND server_i const AdvanceServerRegistrationSQL = `WITH matched AS ( UPDATE matches - SET state = $4, revision = revision + 1 + SET state = $4, + initial_connect_ready_at = CASE WHEN $4 = 'ASSIGNMENT_READY' THEN $6 ELSE initial_connect_ready_at END, + revision = revision + 1 WHERE match_id = $1 AND server_id = $2 AND state = $3 AND protocol_version = $7 AND EXISTS (SELECT 1 FROM allocations WHERE match_id = $1 AND server_id = $2 AND allocation_id = $5 AND protocol_version = $7 AND state = 'ALLOCATED') AND ($4 <> 'ASSIGNMENT_READY' OR (SELECT count(*) FROM assignments WHERE match_id = $1 AND expires_at > $6) = (SELECT count(*) FROM match_participants WHERE match_id = $1)) diff --git a/server/store/allocation_match_sql_test.go b/server/store/allocation_match_sql_test.go index 8942f981..01735620 100644 --- a/server/store/allocation_match_sql_test.go +++ b/server/store/allocation_match_sql_test.go @@ -13,7 +13,7 @@ func TestAllocationMatchClaimSQLFencesConcurrentWorkers(t *testing.T) { AllocatingMatchBuildSQL: {"match_participants", "queue_tickets", "ORDER BY q.client_build"}, BindAllocatedMatchParticipantsSQL: {"allocation_id = $2", "server_id IS NULL", "SET server_id = $3", "FROM allocations", "state = 'ALLOCATING'", "revision = revision + 1", "SELECT revision FROM bound"}, ReleaseAllocatedMatchClaimSQL: {"allocation_id = $2", "allocation_id = NULL", "allocation_claimed_at = NULL"}, - AdvanceServerRegistrationSQL: {"state = $4", "protocol_version = $7", "ASSIGNMENT_READY", "revision = revision + 1"}, + AdvanceServerRegistrationSQL: {"state = $4", "initial_connect_ready_at", "$6", "protocol_version = $7", "ASSIGNMENT_READY", "revision = revision + 1"}, ServerRegistrationIdempotencyInsertSQL: {"idempotency_keys", "ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"}, ServerRegistrationIdempotencySelectSQL: {"scope = $1", "idempotency_key = $2", "FOR UPDATE"}, } diff --git a/server/store/assignment_sql.go b/server/store/assignment_sql.go index 7965e613..7a93e8d8 100644 --- a/server/store/assignment_sql.go +++ b/server/store/assignment_sql.go @@ -63,11 +63,13 @@ WHERE assignments.allocation_id = EXCLUDED.allocation_id AND assignments.expires_at = EXCLUDED.expires_at AND assignments.revision = EXCLUDED.revision` -const AssignmentSelectSQL = `SELECT match_id, player_id, allocation_id, server_id, - slot, region, client_build, protocol_version, transport, endpoint, - join_authorisation, manifest_digest, expires_at, revision -FROM assignments -WHERE match_id = $1 AND player_id = $2 AND expires_at > $3` +const AssignmentSelectSQL = `SELECT a.match_id, a.player_id, a.allocation_id, a.server_id, + a.slot, a.region, a.client_build, a.protocol_version, a.transport, a.endpoint, + a.join_authorisation, a.manifest_digest, a.expires_at, a.revision +FROM assignments a +JOIN matches m ON m.match_id = a.match_id AND m.server_id = a.server_id +WHERE a.match_id = $1 AND a.player_id = $2 AND a.expires_at > $3 + AND m.state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING', 'LIVE')` const AssignmentRosterSelectSQL = `SELECT allocation_id, server_id, join_authorisation FROM assignments diff --git a/server/store/assignment_sql_test.go b/server/store/assignment_sql_test.go index 211158f3..c166bb13 100644 --- a/server/store/assignment_sql_test.go +++ b/server/store/assignment_sql_test.go @@ -10,7 +10,7 @@ import ( func TestAssignmentSQLBindsPlayerAndPreservesIdenticalReplay(t *testing.T) { for query, fragments := range map[string][]string{ AssignmentUpsertSQL: {"ON CONFLICT (match_id, player_id)", "WHERE assignments.allocation_id = EXCLUDED.allocation_id", "join_authorisation", "manifest_digest"}, - AssignmentSelectSQL: {"match_id = $1", "player_id = $2", "expires_at > $3"}, + AssignmentSelectSQL: {"a.match_id = $1", "a.player_id = $2", "a.expires_at > $3", "JOIN matches", "ASSIGNMENT_READY", "m.server_id = a.server_id"}, } { for _, fragment := range fragments { if !contains(query, fragment) { diff --git a/server/store/initial_connect_maintenance.go b/server/store/initial_connect_maintenance.go index 2d08f45b..ff55aa35 100644 --- a/server/store/initial_connect_maintenance.go +++ b/server/store/initial_connect_maintenance.go @@ -3,16 +3,18 @@ package store import ( "context" "database/sql" + "errors" "fmt" "time" "github.com/cosmic-clash/cosmic-clash/server/domain" ) -const initialConnectCandidatesSQL = `SELECT match_id, playlist, created_at +const initialConnectCandidatesSQL = `SELECT match_id, playlist, initial_connect_ready_at FROM matches WHERE state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING') -ORDER BY created_at, match_id + AND initial_connect_ready_at IS NOT NULL +ORDER BY initial_connect_ready_at, match_id LIMIT $1` const initialConnectHistorySQL = `SELECT starts_at @@ -71,6 +73,12 @@ func ReconcileInitialConnect(ctx context.Context, db *sql.DB, now time.Time, lim continue } if err := ApplyInitialConnectPlan(ctx, db, candidate.matchID, "initial-connect:"+candidate.matchID, plan, now); err != nil { + // A connection receipt or another maintenance replica may have + // changed the locked roster/state after our snapshot. Re-evaluate on + // the next bounded pass instead of killing the maintenance process. + if errors.Is(err, domain.ErrConflict) { + continue + } return count, err } count++ @@ -79,7 +87,7 @@ func ReconcileInitialConnect(ctx context.Context, db *sql.DB, now time.Time, lim } func loadInitialConnectSnapshot(ctx context.Context, db *sql.DB, matchID string) ([]domain.ConnectParticipant, error) { - rows, err := db.QueryContext(ctx, `SELECT player_id, team, connected_at + rows, err := db.QueryContext(ctx, `SELECT player_id, team, slot, connected_at FROM match_participants WHERE match_id = $1 AND participation_active ORDER BY player_id`, matchID) if err != nil { return nil, err @@ -88,12 +96,12 @@ FROM match_participants WHERE match_id = $1 AND participation_active ORDER BY pl var participants []domain.ConnectParticipant for rows.Next() { var playerID string - var team int + var team, slot int var connectedAt sql.NullTime - if err := rows.Scan(&playerID, &team, &connectedAt); err != nil { + if err := rows.Scan(&playerID, &team, &slot, &connectedAt); err != nil { return nil, err } - participants = append(participants, domain.ConnectParticipant{PlayerID: playerID, Team: team, Connected: connectedAt.Valid}) + participants = append(participants, domain.ConnectParticipant{PlayerID: playerID, Team: team, Slot: slot, Connected: connectedAt.Valid}) } return participants, rows.Err() } diff --git a/server/store/initial_connect_sql.go b/server/store/initial_connect_sql.go index 38557e3f..d63c3c40 100644 --- a/server/store/initial_connect_sql.go +++ b/server/store/initial_connect_sql.go @@ -18,7 +18,7 @@ const InitialConnectIdempotencyScope = "match.initial_connect" const initialConnectMatchLockSQL = `SELECT playlist, state, revision FROM matches WHERE match_id = $1 FOR UPDATE` -const initialConnectParticipantsSQL = `SELECT player_id, ticket_id, team, connected_at, +const initialConnectParticipantsSQL = `SELECT player_id, ticket_id, team, slot, connected_at, participation_active FROM match_participants WHERE match_id = $1 ORDER BY player_id FOR UPDATE` @@ -76,6 +76,7 @@ type initialConnectParticipant struct { PlayerID string TicketID string Team int + Slot int ConnectedAt sql.NullTime Active bool } @@ -84,7 +85,8 @@ type initialConnectParticipant struct { // It is deliberately a store operation: no-show penalties and innocent-ticket // requeue must commit with the match transition or neither may commit. func ApplyInitialConnectPlan(ctx context.Context, db *sql.DB, matchID, idempotencyKey string, plan domain.InitialConnectPlan, now time.Time) error { - if db == nil || matchID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() || plan.Action == domain.InitialConnectWait || (plan.Action != domain.InitialConnectCancel && plan.Action != domain.InitialConnectStartWithBot) || plan.MatchState == domain.Live && plan.Action != domain.InitialConnectStartWithBot || plan.MatchState == domain.Cancelled && plan.Action != domain.InitialConnectCancel { + validActionState := (plan.Action == domain.InitialConnectStart || plan.Action == domain.InitialConnectStartWithBot) && plan.MatchState == domain.Live || plan.Action == domain.InitialConnectCancel && plan.MatchState == domain.Cancelled + if db == nil || matchID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() || !validActionState { return fmt.Errorf("invalid initial-connect transaction arguments") } digest, err := initialConnectDigest(matchID, plan) @@ -107,7 +109,7 @@ func ApplyInitialConnectPlan(ctx context.Context, db *sql.DB, matchID, idempoten return err } if !bytes.Equal(prior, digest[:]) { - return fmt.Errorf("conflicting initial-connect request") + return fmt.Errorf("%w: conflicting initial-connect request", domain.ErrConflict) } return nil } @@ -117,14 +119,14 @@ func ApplyInitialConnectPlan(ctx context.Context, db *sql.DB, matchID, idempoten return err } if state != string(domain.AssignmentReady) && state != string(domain.Assigned) && state != string(domain.Connecting) { - return fmt.Errorf("match is not awaiting initial connect: %s", state) + return fmt.Errorf("%w: match is not awaiting initial connect: %s", domain.ErrConflict, state) } participants, err := loadInitialConnectParticipants(ctx, tx, matchID) if err != nil { return err } if err := validateInitialConnectPlan(plan, participants, domain.Playlist(playlist)); err != nil { - return err + return fmt.Errorf("%w: %v", domain.ErrConflict, err) } if plan.Action == domain.InitialConnectCancel { if _, err := tx.ExecContext(ctx, initialConnectReleaseAllSQL, matchID); err != nil { @@ -195,7 +197,7 @@ func loadInitialConnectParticipants(ctx context.Context, tx *sql.Tx, matchID str var result []initialConnectParticipant for rows.Next() { var p initialConnectParticipant - if err := rows.Scan(&p.PlayerID, &p.TicketID, &p.Team, &p.ConnectedAt, &p.Active); err != nil { + if err := rows.Scan(&p.PlayerID, &p.TicketID, &p.Team, &p.Slot, &p.ConnectedAt, &p.Active); err != nil { return nil, err } result = append(result, p) @@ -204,15 +206,17 @@ func loadInitialConnectParticipants(ctx context.Context, tx *sql.Tx, matchID str } func validateInitialConnectPlan(plan domain.InitialConnectPlan, participants []initialConnectParticipant, playlist domain.Playlist) error { - if len(participants) == 0 || (plan.Action == domain.InitialConnectStartWithBot && playlist != domain.Casual) || (plan.Action == domain.InitialConnectCancel && plan.MatchState != domain.Cancelled) { + if len(participants) == 0 || (plan.Action == domain.InitialConnectStartWithBot && playlist != domain.Casual) || (plan.Action == domain.InitialConnectCancel && plan.MatchState != domain.Cancelled) || (plan.Action == domain.InitialConnectStart && (plan.MatchState != domain.Live || len(plan.NoShows) != 0 || len(plan.CasualLineup) != 0)) { return fmt.Errorf("invalid initial-connect plan") } known, connected, missing := map[string]bool{}, map[string]bool{}, map[string]bool{} + stored := make(map[string]initialConnectParticipant, len(participants)) for _, p := range participants { - if p.PlayerID == "" || !p.Active || known[p.PlayerID] { + if p.PlayerID == "" || !p.Active || p.Team < 0 || p.Team > 1 || p.Slot < 0 || p.Slot > 5 || p.Slot/3 != p.Team || known[p.PlayerID] { return fmt.Errorf("invalid stored participant roster") } known[p.PlayerID] = true + stored[p.PlayerID] = p if p.ConnectedAt.Valid { connected[p.PlayerID] = true } @@ -232,6 +236,9 @@ func validateInitialConnectPlan(plan domain.InitialConnectPlan, participants []i if len(missing) != len(known) { return fmt.Errorf("initial-connect plan does not cover roster") } + if plan.Action == domain.InitialConnectStart && len(connected) != len(known) { + return fmt.Errorf("initial-connect start requires complete connected roster") + } if plan.Action == domain.InitialConnectStartWithBot { if len(plan.CasualLineup) != 6 { return fmt.Errorf("casual bot lineup must contain six players") @@ -239,7 +246,7 @@ func validateInitialConnectPlan(plan domain.InitialConnectPlan, participants []i lineupSlots := make(map[int]bool, 6) lineupPlayers := make(map[string]bool, 6) for _, slot := range plan.CasualLineup { - if slot.Slot < 0 || slot.Slot > 5 || slot.Team != slot.Slot%2 || lineupSlots[slot.Slot] || slot.PlayerID == "" || lineupPlayers[slot.PlayerID] { + if slot.Slot < 0 || slot.Slot > 5 || slot.Team != slot.Slot/3 || lineupSlots[slot.Slot] || slot.PlayerID == "" || lineupPlayers[slot.PlayerID] { return fmt.Errorf("invalid casual bot lineup") } lineupSlots[slot.Slot] = true @@ -250,6 +257,10 @@ func validateInitialConnectPlan(plan domain.InitialConnectPlan, participants []i if !connected[slot.PlayerID] { return fmt.Errorf("lineup contains non-connected human") } + participant := stored[slot.PlayerID] + if participant.Slot != slot.Slot || participant.Team != slot.Team { + return fmt.Errorf("lineup moves connected human from assigned slot") + } } for id := range connected { if !lineupPlayers[id] { diff --git a/server/store/initial_connect_sql_test.go b/server/store/initial_connect_sql_test.go index 911d60fd..833e6de4 100644 --- a/server/store/initial_connect_sql_test.go +++ b/server/store/initial_connect_sql_test.go @@ -9,7 +9,7 @@ import ( ) func TestInitialConnectSQLPreservesAtomicNoShowReconciliation(t *testing.T) { - if !contains(initialConnectCandidatesSQL, "ASSIGNMENT_READY") || !contains(initialConnectCandidatesSQL, "LIMIT $1") { + if !contains(initialConnectCandidatesSQL, "ASSIGNMENT_READY") || !contains(initialConnectCandidatesSQL, "initial_connect_ready_at") || !contains(initialConnectCandidatesSQL, "LIMIT $1") { t.Fatal("initial-connect sweep is not bounded to pre-live matches") } for query, fragments := range map[string][]string{ @@ -31,28 +31,45 @@ func TestInitialConnectSQLPreservesAtomicNoShowReconciliation(t *testing.T) { func TestInitialConnectPlanValidationRejectsIncompleteOrForgedPlans(t *testing.T) { participants := []initialConnectParticipant{ - {PlayerID: "p0", TicketID: "t0", Team: 0, ConnectedAt: validTime(100), Active: true}, - {PlayerID: "p1", TicketID: "t1", Team: 1, Active: true}, + {PlayerID: "p0", TicketID: "t0", Team: 0, Slot: 0, ConnectedAt: validTime(100), Active: true}, + {PlayerID: "p1", TicketID: "t1", Team: 1, Slot: 3, Active: true}, } plan := domain.InitialConnectPlan{ Action: domain.InitialConnectStartWithBot, MatchState: domain.Live, Connected: []string{"p0"}, NoShows: []domain.Abandonment{{PlayerID: "p1", Cooldown: time.Minute, AbandonedAt: time.Unix(100, 0)}}, CasualLineup: []domain.CasualSlot{ - {Slot: 0, Team: 0, PlayerID: "p0"}, {Slot: 1, Team: 1, PlayerID: "bot-1", IsBot: true}, + {Slot: 0, Team: 0, PlayerID: "p0"}, {Slot: 1, Team: 0, PlayerID: "bot-1", IsBot: true}, {Slot: 2, Team: 0, PlayerID: "bot-2", IsBot: true}, {Slot: 3, Team: 1, PlayerID: "bot-3", IsBot: true}, - {Slot: 4, Team: 0, PlayerID: "bot-4", IsBot: true}, {Slot: 5, Team: 1, PlayerID: "bot-5", IsBot: true}, + {Slot: 4, Team: 1, PlayerID: "bot-4", IsBot: true}, {Slot: 5, Team: 1, PlayerID: "bot-5", IsBot: true}, }, } if err := validateInitialConnectPlan(plan, participants, domain.Casual); err != nil { t.Fatalf("valid plan rejected: %v", err) } - plan.CasualLineup[1].Team = 0 + plan.CasualLineup[1].Team = 1 if err := validateInitialConnectPlan(plan, participants, domain.Casual); err == nil { t.Fatal("team-swapped lineup accepted") } } +func TestInitialConnectPlanValidationRequiresCompleteRosterToStart(t *testing.T) { + participants := []initialConnectParticipant{ + {PlayerID: "p0", TicketID: "t0", Team: 0, Slot: 0, ConnectedAt: validTime(100), Active: true}, + {PlayerID: "p1", TicketID: "t1", Team: 1, Slot: 3, ConnectedAt: validTime(100), Active: true}, + } + plan := domain.InitialConnectPlan{ + Action: domain.InitialConnectStart, MatchState: domain.Live, Connected: []string{"p0", "p1"}, + } + if err := validateInitialConnectPlan(plan, participants, domain.Ranked); err != nil { + t.Fatalf("valid complete start rejected: %v", err) + } + participants[1].ConnectedAt = sql.NullTime{} + if err := validateInitialConnectPlan(plan, participants, domain.Ranked); err == nil { + t.Fatal("start with disconnected participant accepted") + } +} + func validTime(unix int64) (result sql.NullTime) { result.Time = time.Unix(unix, 0) result.Valid = true diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index c1f5cbdf..be0dc8a0 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -468,7 +468,7 @@ func TestPostgreSQLAssignmentPersistenceIsPlayerScopedAndExpiryBound(t *testing. if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ('assignment-ticket', 'assignment-player', 'casual', 'ASSIGNED', 'integration-build', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil { t.Fatal(err) } - if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id) VALUES ('assignment-match', 'casual', 'ASSIGNED', 'EU', 1, 'assignment-server')`); err != nil { + if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, initial_connect_ready_at) VALUES ('assignment-match', 'casual', 'ASSIGNED', 'EU', 1, 'assignment-server', $1)`, now); err != nil { t.Fatal(err) } if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('assignment-match', 'assignment-player', 'assignment-ticket', 0, 0)`); err != nil { @@ -493,6 +493,71 @@ func TestPostgreSQLAssignmentPersistenceIsPlayerScopedAndExpiryBound(t *testing. } } +func TestPostgreSQLConnectionReceiptsStartCompleteRelaxedCasualRoster(t *testing.T) { + db := openIntegrationPostgres(t) + applyIntegrationMigrations(t, db) + now := time.Now().UTC().Truncate(time.Microsecond) + ctx := context.Background() + + for i := 0; i < 2; i++ { + playerID := fmt.Sprintf("connect-player-%d", i) + ticketID := fmt.Sprintf("connect-ticket-%d", i) + if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, playerID); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'ASSIGNMENT_READY', 'integration-build', 1, $3, $4)`, ticketID, playerID, now, now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + } + if _, err := db.ExecContext(ctx, `INSERT INTO game_servers (server_id, region, build, protocol_version, transport, state) VALUES ('connect-server', 'EU', 'integration-build', 1, 'enet', 'ALLOCATED')`); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, allocation_id, initial_connect_ready_at) VALUES ('connect-match', 'casual', 'ASSIGNMENT_READY', 'EU', 1, 'connect-server', 'connect-allocation', $1)`, now); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO allocations (allocation_id, match_id, server_id, region, build, protocol_version, transport, request_digest, state, allocated_at) VALUES ('connect-allocation', 'connect-match', 'connect-server', 'EU', 'integration-build', 1, 'enet', $1, 'ALLOCATED', $2)`, []byte("request"), now); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + playerID := fmt.Sprintf("connect-player-%d", i) + ticketID := fmt.Sprintf("connect-ticket-%d", i) + slot := i * 3 + if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('connect-match', $1, $2, $3, $4)`, playerID, ticketID, slot, i); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO assignments (match_id, player_id, allocation_id, server_id, slot, region, client_build, protocol_version, transport, endpoint, join_authorisation, manifest_digest, expires_at) VALUES ('connect-match', $1, 'connect-allocation', 'connect-server', $2, 'EU', 'integration-build', 1, 'enet', '127.0.0.1:7777', 'join-token', $3, $4)`, playerID, slot, []byte("manifest"), now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + } + binding := domain.WorkloadBinding{AllocationID: "connect-allocation", MatchID: "connect-match", ServerID: "connect-server"} + if err := RecordPlayerConnected(ctx, db, binding, "connect-player-0", "connect-receipt-key-0000", now); err != nil { + t.Fatalf("first receipt: %v", err) + } + if err := RecordPlayerConnected(ctx, db, domain.WorkloadBinding{AllocationID: "forged-allocation", MatchID: "connect-match", ServerID: "connect-server"}, "connect-player-1", "connect-receipt-key-forged", now); !errors.Is(err, domain.ErrConflict) { + t.Fatalf("forged binding err=%v, want conflict", err) + } + if err := RecordPlayerConnected(ctx, db, binding, "connect-player-1", "connect-receipt-key-0001", now); err != nil { + t.Fatalf("second receipt: %v", err) + } + reconciled, err := ReconcileInitialConnect(ctx, db, now.Add(time.Second), 10) + if err != nil || reconciled != 1 { + t.Fatalf("reconcile count=%d err=%v", reconciled, err) + } + var state string + if err := db.QueryRowContext(ctx, `SELECT state FROM matches WHERE match_id = 'connect-match'`).Scan(&state); err != nil || state != string(domain.Live) { + t.Fatalf("match state=%q err=%v", state, err) + } + var liveTickets int + if err := db.QueryRowContext(ctx, `SELECT count(*) FROM queue_tickets WHERE state = 'LIVE'`).Scan(&liveTickets); err != nil || liveTickets != 2 { + t.Fatalf("live tickets=%d err=%v", liveTickets, err) + } + // A lost 204 can be retried after assignment expiry because the exact + // durable receipt is replayed before checking the now-expired assignment. + if err := RecordPlayerConnected(ctx, db, binding, "connect-player-0", "connect-receipt-key-0000", now.Add(2*time.Minute)); err != nil { + t.Fatalf("durable receipt replay: %v", err) + } +} + func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) { db := openIntegrationPostgres(t) applyIntegrationMigrations(t, db) @@ -1333,8 +1398,19 @@ func TestPostgreSQLMigrationsRollBackAndReapplyCleanly(t *testing.T) { // Roll back every migration one at a time, in reverse, checking each // down file actually undoes what its forward file created — not just // that Rollback returns nil. - if err := migrations.Rollback(context.Background(), db, dir, 2); err != nil { - t.Fatalf("rollback 0007 and 0006: %v", err) + if err := migrations.Rollback(context.Background(), db, dir, 4); err != nil { + t.Fatalf("rollback 0010 through 0007: %v", err) + } + var hasInitialConnectReadyColumn bool + if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'initial_connect_ready_at'`).Scan(&hasInitialConnectReadyColumn); err != nil { + t.Fatal(err) + } + if hasInitialConnectReadyColumn { + t.Fatal("0010 rollback did not drop matches.initial_connect_ready_at") + } + + if err := migrations.Rollback(context.Background(), db, dir, 1); err != nil { + t.Fatalf("rollback 0006: %v", err) } var hasAllocationClaimColumn bool if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'allocation_id'`).Scan(&hasAllocationClaimColumn); err != nil { @@ -1345,7 +1421,7 @@ func TestPostgreSQLMigrationsRollBackAndReapplyCleanly(t *testing.T) { } if err := migrations.Rollback(context.Background(), db, dir, 4); err != nil { - t.Fatalf("rollback remaining down to 0001: %v", err) + t.Fatalf("rollback 0005 through 0002: %v", err) } if tableExists("assignments") || tableExists("allocations") || tableExists("game_servers") { t.Fatal("rollback left later-migration tables behind") diff --git a/server/store/server_connection_sql.go b/server/store/server_connection_sql.go new file mode 100644 index 00000000..cfcf0626 --- /dev/null +++ b/server/store/server_connection_sql.go @@ -0,0 +1,78 @@ +package store + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/json" + "fmt" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +const ServerConnectionIdempotencyScope = "server.connection" + +const ServerConnectionIdempotencyInsertSQL = `INSERT INTO idempotency_keys + (scope, idempotency_key, payload_digest, result) +VALUES ($1, $2, $3, '{}'::jsonb) ON CONFLICT (scope, idempotency_key) DO NOTHING` + +const ServerConnectionIdempotencySelectSQL = `SELECT payload_digest +FROM idempotency_keys WHERE scope = $1 AND idempotency_key = $2 FOR UPDATE` + +const ServerConnectionParticipantSQL = `UPDATE match_participants mp +SET connected_at = COALESCE(mp.connected_at, $5) +FROM matches m, allocations a, assignments assn +WHERE mp.match_id = $1 AND mp.player_id = $4 AND mp.participation_active + AND m.match_id = mp.match_id AND m.server_id = $2 + AND m.state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING', 'LIVE') + AND a.allocation_id = $3 AND a.match_id = m.match_id AND a.server_id = m.server_id + AND a.state = 'ALLOCATED' + AND assn.match_id = mp.match_id AND assn.player_id = mp.player_id + AND assn.allocation_id = a.allocation_id AND assn.server_id = m.server_id + AND assn.expires_at > $5 +RETURNING mp.connected_at` + +// RecordPlayerConnected persists authoritative admission observed by the +// allocated game server. The workload allocation, match/server binding, +// active participant, and still-live assignment must all agree. +func RecordPlayerConnected(ctx context.Context, db *sql.DB, binding domain.WorkloadBinding, playerID, idempotencyKey string, now time.Time) error { + if db == nil || binding.AllocationID == "" || binding.MatchID == "" || binding.ServerID == "" || playerID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() { + return fmt.Errorf("invalid server connection receipt") + } + digest := sha256.Sum256([]byte(binding.AllocationID + "\x00" + binding.MatchID + "\x00" + binding.ServerID + "\x00" + playerID)) + return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { + inserted, err := tx.ExecContext(ctx, ServerConnectionIdempotencyInsertSQL, ServerConnectionIdempotencyScope, idempotencyKey, digest[:]) + if err != nil { + return err + } + changed, err := inserted.RowsAffected() + if err != nil { + return err + } + if changed == 0 { + var prior []byte + if err := tx.QueryRowContext(ctx, ServerConnectionIdempotencySelectSQL, ServerConnectionIdempotencyScope, idempotencyKey).Scan(&prior); err != nil { + return err + } + if !bytes.Equal(prior, digest[:]) { + return domain.ErrConflict + } + return nil + } + var connectedAt time.Time + if err := tx.QueryRowContext(ctx, ServerConnectionParticipantSQL, binding.MatchID, binding.ServerID, binding.AllocationID, playerID, now).Scan(&connectedAt); err != nil { + if err == sql.ErrNoRows { + return domain.ErrConflict + } + return err + } + result, err := json.Marshal(map[string]any{"match_id": binding.MatchID, "player_id": playerID, "connected_at": connectedAt}) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, ServerConnectionIdempotencyScope, idempotencyKey, result) + return err + }) +} diff --git a/server/store/server_connection_sql_test.go b/server/store/server_connection_sql_test.go new file mode 100644 index 00000000..bc5e0690 --- /dev/null +++ b/server/store/server_connection_sql_test.go @@ -0,0 +1,33 @@ +package store + +import ( + "context" + "database/sql" + "strings" + "testing" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +func TestServerConnectionSQLBindsWorkloadParticipantAndLiveAssignment(t *testing.T) { + for _, fragment := range []string{ + "connected_at = COALESCE", "mp.participation_active", "m.server_id = $2", + "a.allocation_id = $3", "a.state = 'ALLOCATED'", "assn.player_id = mp.player_id", + "assn.expires_at > $5", "RETURNING mp.connected_at", + } { + if !strings.Contains(ServerConnectionParticipantSQL, fragment) { + t.Fatalf("connection SQL missing %q: %s", fragment, ServerConnectionParticipantSQL) + } + } +} + +func TestRecordPlayerConnectedRejectsInvalidArguments(t *testing.T) { + binding := domain.WorkloadBinding{AllocationID: "allocation-1", MatchID: "match-1", ServerID: "server-1"} + if err := RecordPlayerConnected(context.Background(), (*sql.DB)(nil), binding, "player-1", "connection-key-123456", time.Unix(1000, 0)); err == nil { + t.Fatal("nil database accepted") + } + if err := RecordPlayerConnected(context.Background(), &sql.DB{}, domain.WorkloadBinding{}, "player-1", "connection-key-123456", time.Unix(1000, 0)); err == nil { + t.Fatal("empty workload binding accepted") + } +} diff --git a/server/supervisor/supervisor.go b/server/supervisor/supervisor.go index 952c658e..7a4d1979 100644 --- a/server/supervisor/supervisor.go +++ b/server/supervisor/supervisor.go @@ -51,6 +51,7 @@ type Config struct { ReadyURL string Transport string DrainURL string + AdmissionURL string DrainToken string ReadyTimeout time.Duration PollInterval time.Duration @@ -87,9 +88,8 @@ type Config struct { // holding a live, unexpired assignment -- see // AdvanceServerRegistrationSQL) may not be satisfied on the very first // attempt if the signed roster is still propagating, and that is - // expected, not fatal: unlike a process-ready registration failure, this - // does not kill the child, since the process is already legitimately - // listening and usable either way. Default 5 attempts, 2s apart. + // expected during the bounded retries. Exhaustion is fatal because player + // assignments remain hidden until this transition. Default 5 attempts, 2s apart. AssignmentReadyAttempts int AssignmentReadyBackoff time.Duration // RosterPath is an operator-mounted writable path where the supervisor @@ -106,6 +106,12 @@ type Supervisor struct { lastGameServer GameServer } +const ( + ChildControlPlaneURLEnv = "COSMIC_CLASH_CONTROL_PLANE_URL" + ChildWorkloadTokenEnv = "COSMIC_CLASH_WORKLOAD_TOKEN" + ChildAdmissionSignalEnv = "COSMIC_CLASH_INITIAL_CONNECT_SIGNAL_REQUIRED" +) + const ( DefaultDrainGrace = 285 * time.Second DefaultHTTPTimeout = 10 * time.Second @@ -144,9 +150,23 @@ func New(config Config) (*Supervisor, error) { return nil, err } } + if config.AdmissionURL != "" { + if config.DrainToken == "" { + return nil, fmt.Errorf("initial-connect admission URL requires a control token") + } + if err := validateLocalDrainURL(config.AdmissionURL); err != nil { + return nil, fmt.Errorf("invalid initial-connect admission URL: %w", err) + } + } if config.ControlPlaneURL != "" && (config.ServerID == "" || config.ProtocolVersion < 1 || config.ImageDigest == "") { return nil, fmt.Errorf("control-plane registration requires a server ID, protocol version and image digest") } + if config.ControlPlaneURL != "" { + parsed, err := url.Parse(config.ControlPlaneURL) + if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || parsed.Path != "" { + return nil, fmt.Errorf("control-plane URL must be an HTTP(S) origin") + } + } if config.RosterPath != "" && config.ControlPlaneURL == "" { return nil, fmt.Errorf("roster path requires control-plane URL") } @@ -190,6 +210,11 @@ func (s *Supervisor) Start(ctx context.Context) error { if err != nil { return err } + childControlPlaneEnv, err := s.controlPlaneChildEnvironment() + if err != nil { + return err + } + env = append(env, childControlPlaneEnv...) command := withAllocatedConfig(s.config.Command, s.matchID(), s.config.ServerID, s.config.ImageDigest, rosterExpiry) command, err = withAllocatedCompatibility(command, s.lastGameServer) if err != nil { @@ -224,10 +249,61 @@ func (s *Supervisor) Start(ctx context.Context) error { _ = s.cmd.Process.Kill() return err } - s.reportAssignmentReady(ctx) + if err := s.reportAssignmentReady(ctx); err != nil { + // Player assignments remain deliberately hidden until this durable + // transition succeeds. Do not leave an Agones-Ready process accepting + // connections for a match the control plane cannot expose. + _ = s.cmd.Process.Kill() + return err + } + if err := s.signalInitialConnectReady(ctx); err != nil { + _ = s.cmd.Process.Kill() + return err + } return nil } +func (s *Supervisor) signalInitialConnectReady(ctx context.Context) error { + if s.config.AdmissionURL == "" { + return nil + } + request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.config.AdmissionURL, nil) + if err != nil { + return err + } + request.Header.Set("Authorization", "Bearer "+s.config.DrainToken) + response, err := s.client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode/100 != 2 { + return fmt.Errorf("initial-connect admission control returned %s", response.Status) + } + return nil +} + +func (s *Supervisor) controlPlaneChildEnvironment() ([]string, error) { + if s.config.ControlPlaneURL == "" { + return nil, nil + } + token, err := s.workloadToken() + if err != nil { + return nil, err + } + if strings.ContainsRune(token, '\x00') { + return nil, fmt.Errorf("workload token contains an invalid environment byte") + } + environment := []string{ + ChildControlPlaneURLEnv + "=" + strings.TrimRight(s.config.ControlPlaneURL, "/"), + ChildWorkloadTokenEnv + "=" + token, + } + if s.config.AdmissionURL != "" { + environment = append(environment, ChildAdmissionSignalEnv+"=1") + } + return environment, nil +} + func (s *Supervisor) fetchRoster(ctx context.Context) (time.Time, error) { if s.config.RosterPath == "" { return time.Time{}, nil @@ -387,29 +463,27 @@ func withAllocatedValues(command []string, values map[string]string) []string { return result } -// reportAssignmentReady is best-effort: process-ready has already succeeded, -// so the process is legitimately usable either way. A persistent failure is -// written to stderr rather than returned, since treating it as fatal would -// kill a perfectly healthy process over what is usually just the signed -// roster's durable rows not having propagated yet. -func (s *Supervisor) reportAssignmentReady(ctx context.Context) { +// reportAssignmentReady retries the durable gate that makes player +// assignments visible. A process without this transition is not usable even +// when Agones and the local readiness probe consider it healthy. +func (s *Supervisor) reportAssignmentReady(ctx context.Context) error { if s.config.ControlPlaneURL == "" { - return + return nil } var lastErr error for attempt := 0; attempt < s.config.AssignmentReadyAttempts; attempt++ { if attempt > 0 { select { case <-ctx.Done(): - return + return ctx.Err() case <-time.After(s.config.AssignmentReadyBackoff): } } if lastErr = s.registerControlPlane(ctx, true); lastErr == nil { - return + return nil } } - fmt.Fprintf(os.Stderr, "game-server-supervisor: assignment-ready registration did not succeed after %d attempts: %v\n", s.config.AssignmentReadyAttempts, lastErr) + return fmt.Errorf("assignment-ready registration did not succeed after %d attempts: %w", s.config.AssignmentReadyAttempts, lastErr) } // registerControlPlane reports the allocated process's readiness to the diff --git a/server/supervisor/supervisor_test.go b/server/supervisor/supervisor_test.go index b06401e2..2ada2f29 100644 --- a/server/supervisor/supervisor_test.go +++ b/server/supervisor/supervisor_test.go @@ -23,6 +23,41 @@ func TestSupervisorDefaultHTTPClientHasRequestDeadline(t *testing.T) { } } +func TestAllocatedChildReceivesConnectionReportingEnvironmentWithoutCommandSecrets(t *testing.T) { + s, err := New(Config{ + Command: []string{"game-server"}, ControlPlaneURL: "https://control.example", + ServerID: "server-1", ProtocolVersion: 1, ImageDigest: "sha256:" + strings.Repeat("a", 64), + }) + if err != nil { + t.Fatal(err) + } + s.lastGameServer.ObjectMeta.Annotations = map[string]string{"cosmic-clash.io/workload-token": "signed-workload-token"} + environment, err := s.controlPlaneChildEnvironment() + if err != nil { + t.Fatal(err) + } + joined := strings.Join(environment, "\n") + if !strings.Contains(joined, ChildControlPlaneURLEnv+"=https://control.example") || !strings.Contains(joined, ChildWorkloadTokenEnv+"=signed-workload-token") || strings.Contains(joined, ChildAdmissionSignalEnv) { + t.Fatalf("child connection-reporting environment = %v", environment) + } + if strings.Contains(strings.Join(s.config.Command, " "), "signed-workload-token") { + t.Fatal("workload token leaked into child command arguments") + } + s.config.AdmissionURL = "http://127.0.0.1:7780/initial-connect-ready" + environment, err = s.controlPlaneChildEnvironment() + if err != nil || !strings.Contains(strings.Join(environment, "\n"), ChildAdmissionSignalEnv+"=1") { + t.Fatalf("child admission signal environment = %v err=%v", environment, err) + } +} + +func TestSupervisorRejectsUnsafeControlPlaneOrigins(t *testing.T) { + for _, raw := range []string{"control.example", "https://user:secret@control.example", "https://control.example/path", "https://control.example?token=secret"} { + if _, err := New(Config{Command: []string{"game-server"}, ControlPlaneURL: raw, ServerID: "server-1", ProtocolVersion: 1, ImageDigest: "sha256:" + strings.Repeat("a", 64)}); err == nil { + t.Fatalf("unsafe control-plane URL accepted: %q", raw) + } + } +} + func TestWithAllocatedConfigOverridesAuthoritativeChildFlags(t *testing.T) { command := []string{ "game-server", "--", "--allocated-mode", "--match-id=stale-match", @@ -265,6 +300,7 @@ func TestControlPlaneRegistrationReportsProcessReadyThenAssignmentReady(t *testi func TestAssignmentReadyRegistrationRetriesUntilTheControlPlaneCatchesUp(t *testing.T) { var mu sync.Mutex assignmentReadyAttempts := 0 + admissionCalled := false server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case r.URL.Path == "/gameserver": @@ -288,6 +324,15 @@ func TestAssignmentReadyRegistrationRetriesUntilTheControlPlaneCatchesUp(t *test return } w.WriteHeader(http.StatusNoContent) + case r.URL.Path == "/initial-connect-ready": + mu.Lock() + defer mu.Unlock() + if assignmentReadyAttempts != 3 || r.Header.Get("Authorization") != "Bearer control-token-123456" { + w.WriteHeader(http.StatusConflict) + return + } + admissionCalled = true + w.WriteHeader(http.StatusAccepted) default: w.WriteHeader(http.StatusNotFound) } @@ -301,13 +346,14 @@ func TestAssignmentReadyRegistrationRetriesUntilTheControlPlaneCatchesUp(t *test s, err := New(Config{ Command: []string{"/bin/sh", "-c", "true"}, SDKBaseURL: server.URL, ReadyURL: server.URL + "/ready-probe", ReadyTimeout: time.Second, PollInterval: time.Millisecond, ControlPlaneURL: server.URL, WorkloadTokenPath: tokenPath, ServerID: "server-1", MatchID: "match-1", ProtocolVersion: 1, ImageDigest: "sha256:aa", + DrainURL: server.URL + "/drain", AdmissionURL: server.URL + "/initial-connect-ready", DrainToken: "control-token-123456", AssignmentReadyAttempts: 5, AssignmentReadyBackoff: time.Millisecond, }) if err != nil { t.Fatal(err) } - // Start must still succeed -- a slow-to-propagate assignment-ready must - // never be treated as a Start() failure (which would kill the child). + // A transient conflict is retried inside Start; the assignment only becomes + // visible after the durable transition eventually succeeds. if err := s.Start(context.Background()); err != nil { t.Fatalf("Start failed despite assignment-ready eventually succeeding: %v", err) } @@ -319,6 +365,42 @@ func TestAssignmentReadyRegistrationRetriesUntilTheControlPlaneCatchesUp(t *test if assignmentReadyAttempts != 3 { t.Fatalf("assignment-ready attempts = %d, want exactly 3 (2 conflicts then success)", assignmentReadyAttempts) } + if !admissionCalled { + t.Fatal("initial-connect clock was not armed after durable assignment readiness") + } +} + +func TestPersistentAssignmentReadyFailureFailsClosed(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/gameserver": + _, _ = w.Write([]byte(`{"object_meta":{"annotations":{"cosmic-clash.io/match-id":"match-1","cosmic-clash.io/workload-token":"workload-token"}},"status":{"address":"127.0.0.1","ports":[{"name":"game","port":31001}]}}`)) + case "/ready-probe", "/ready": + w.WriteHeader(http.StatusOK) + case "/v1/servers/server-1/register": + body, _ := io.ReadAll(r.Body) + if strings.Contains(string(body), `"assignment_ready":true`) { + w.WriteHeader(http.StatusConflict) + return + } + w.WriteHeader(http.StatusNoContent) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + s, err := New(Config{ + Command: []string{"/bin/sh", "-c", "sleep 30"}, SDKBaseURL: server.URL, ReadyURL: server.URL + "/ready-probe", + ControlPlaneURL: server.URL, ServerID: "server-1", ProtocolVersion: 1, ImageDigest: "sha256:aa", + ReadyTimeout: time.Second, PollInterval: time.Millisecond, AssignmentReadyAttempts: 2, AssignmentReadyBackoff: time.Millisecond, + }) + if err != nil { + t.Fatal(err) + } + if err := s.Start(context.Background()); err == nil || !strings.Contains(err.Error(), "assignment-ready registration did not succeed") { + t.Fatalf("persistent assignment-ready failure did not fail closed: %v", err) + } + _ = s.Wait() } func TestControlPlaneRegistrationFallsBackToGameServerAnnotationForMatchID(t *testing.T) {