diff --git a/Game/scripts/control_plane_client.gd b/Game/scripts/control_plane_client.gd index 0777d7df..cb6d45a8 100644 --- a/Game/scripts/control_plane_client.gd +++ b/Game/scripts/control_plane_client.gd @@ -6,6 +6,8 @@ extends Node signal request_succeeded(operation: String, payload: Dictionary) signal request_failed(operation: String, http_code: int, detail: String) signal session_expired() +signal probe_challenge_received(region: String, nonce_base64: String) +signal probe_recorded(region: String, server_rtt_ms: int) signal session_changed(player_id: String) signal websocket_event(event: Dictionary) signal websocket_status_changed(status: String) @@ -215,6 +217,50 @@ func queue_create(ticket_id: String, playlist: String, client_build: String, pro return err +# Regional latency probing. The backend issues a single-use nonce, the client +# echoes it back with its opaque platform location, and the backend derives the +# round trip from its own timestamps -- no client-measured latency is accepted. +# +# Until a ticket has RTT evidence for at least one region the matcher will not +# consider it (server/domain.validCandidate requires a non-empty map), so this +# has to complete before searching is meaningful. +const PROBE_REGIONS := ["EU", "NA"] + + +func request_probe_challenge(region: String) -> Error: + if not is_valid_probe_region(region): + return ERR_INVALID_PARAMETER + return _start_request("probe_challenge_" + region, HTTPClient.METHOD_POST, "/v1/probes/%s/challenge" % region, {}, "") + + +func submit_probe_answer(region: String, nonce_base64: String, opaque_location_base64: String) -> Error: + if not is_valid_probe_region(region) or nonce_base64.is_empty() or opaque_location_base64.is_empty(): + return ERR_INVALID_PARAMETER + return _start_request("probe_answer_" + region, HTTPClient.METHOD_POST, "/v1/probes/%s" % region, { + "nonce": nonce_base64, + "opaque_location": opaque_location_base64, + }, "") + + +static func is_valid_probe_region(region: String) -> bool: + return region == "EU" or region == "NA" + + +# The platform location is opaque to us by design: the backend treats it as a +# blob and never derives placement from anything the client measured. Without a +# Steam runtime there is nothing to report, so send a stable non-empty marker +# rather than failing the probe -- the RTT is what actually matters and that is +# measured by the backend either way. +static func opaque_location_payload() -> String: + if Engine.has_singleton("Steam"): + var steam := Engine.get_singleton("Steam") + if steam.has_method("getLocalPingLocation"): + var location = steam.call("getLocalPingLocation") + if location is String and not String(location).is_empty(): + return Marshalls.utf8_to_base64(String(location)) + return Marshalls.utf8_to_base64("no-platform-ping-location") + + func login_steam(web_api_ticket: String) -> Error: if not is_valid_web_api_ticket(web_api_ticket): return ERR_INVALID_PARAMETER @@ -592,6 +638,17 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head if not assignment.apply(payload, player_id): request_failed.emit(operation, response_code, assignment.error_message) return + elif operation.begins_with("probe_challenge_"): + # Answer immediately: the nonce is single-use and short-lived, and the + # interval to this answer is exactly what the backend measures. + var challenge_region := operation.trim_prefix("probe_challenge_") + var nonce := String(payload.get("nonce", "")) + if nonce.is_empty(): + request_failed.emit(operation, response_code, "probe challenge did not include a nonce") + return + probe_challenge_received.emit(challenge_region, nonce) + elif operation.begins_with("probe_answer_"): + probe_recorded.emit(operation.trim_prefix("probe_answer_"), int(payload.get("server_rtt_ms", -1))) request_succeeded.emit(operation, payload) if not _pending_resync_resource_id.is_empty(): call_deferred("_run_pending_resync") diff --git a/Game/scripts/matchmaking.gd b/Game/scripts/matchmaking.gd index eadefafe..af49007a 100644 --- a/Game/scripts/matchmaking.gd +++ b/Game/scripts/matchmaking.gd @@ -18,6 +18,12 @@ const RECOVERY_POLL_SECONDS := 2.0 var _elapsed_seconds := 0.0 var _heartbeat_seconds := 0.0 var _recovery_poll_seconds := 0.0 +# Regions still awaiting RTT evidence, and the queue request deferred until at +# least one lands. The matcher ignores a ticket with no predicted RTT, so +# queueing before probing produces a search that can never match. +var _pending_probe_regions: Array[String] = [] +var _probed_regions: Array[String] = [] +var _deferred_queue := {} func _ready() -> void: @@ -31,6 +37,8 @@ func _ready() -> void: ControlPlaneClient.request_failed.connect(_on_request_failed) ControlPlaneClient.request_succeeded.connect(_on_request_succeeded) ControlPlaneClient.session_expired.connect(_on_session_expired) + ControlPlaneClient.probe_challenge_received.connect(_on_probe_challenge_received) + ControlPlaneClient.probe_recorded.connect(_on_probe_recorded) _refresh_ranked_profile() _render(ControlPlaneClient.state.snapshot()) @@ -71,11 +79,73 @@ func _on_queue_pressed() -> void: _recovery_poll_seconds = 0.0 var playlist := String(playlist_dropdown.get_selected_metadata()) var ticket_id := "ticket-%s-%s" % [str(Time.get_ticks_usec()), str(randi())] + # A ticket with no regional RTT evidence is invisible to the matcher, so + # collect it first and queue once the first region reports. + if _probed_regions.is_empty(): + _deferred_queue = {"ticket_id": ticket_id, "playlist": playlist} + _start_probe_collection() + return var err := ControlPlaneClient.queue_create(ticket_id, playlist, CLIENT_BUILD, PROTOCOL_VERSION) if err != OK: _on_local_error("Could not start matchmaking: %s" % error_string(err)) +func _start_probe_collection() -> void: + _pending_probe_regions = [] + for region in ControlPlaneClient.PROBE_REGIONS: + _pending_probe_regions.append(String(region)) + ControlPlaneClient.state.set_notice("Measuring connection quality...") + _request_next_probe() + + +# One request at a time: the client serialises HTTP through a single +# HTTPRequest, so a second call would return ERR_BUSY. +func _request_next_probe() -> void: + if _pending_probe_regions.is_empty(): + _finish_probe_collection() + return + var region := _pending_probe_regions[0] + var err := ControlPlaneClient.request_probe_challenge(region) + if err != OK and err != ERR_BUSY: + # A region we cannot probe is not fatal; placement just uses the + # regions that did respond. + _pending_probe_regions.remove_at(0) + _request_next_probe() + + +func _on_probe_challenge_received(region: String, nonce_base64: String) -> void: + var err := ControlPlaneClient.submit_probe_answer(region, nonce_base64, ControlPlaneClient.opaque_location_payload()) + if err != OK: + _drop_pending_probe(region) + + +func _on_probe_recorded(region: String, _server_rtt_ms: int) -> void: + if not _probed_regions.has(region): + _probed_regions.append(region) + _drop_pending_probe(region) + + +func _drop_pending_probe(region: String) -> void: + var index := _pending_probe_regions.find(region) + if index >= 0: + _pending_probe_regions.remove_at(index) + _request_next_probe() + + +func _finish_probe_collection() -> void: + if _deferred_queue.is_empty(): + return + var queued := _deferred_queue + _deferred_queue = {} + if _probed_regions.is_empty(): + # Queueing now would create a ticket the matcher can never select. + _on_local_error("Could not measure connection quality to any region; matchmaking is unavailable") + return + var err := ControlPlaneClient.queue_create(String(queued["ticket_id"]), String(queued["playlist"]), CLIENT_BUILD, PROTOCOL_VERSION) + if err != OK: + _on_local_error("Could not start matchmaking: %s" % error_string(err)) + + func _on_cancel_pressed() -> void: if not ControlPlaneClient.state.can_cancel(): return diff --git a/Game/tests/cases/test_control_plane_client.gd b/Game/tests/cases/test_control_plane_client.gd index 4aef55ee..9865a92c 100644 --- a/Game/tests/cases/test_control_plane_client.gd +++ b/Game/tests/cases/test_control_plane_client.gd @@ -508,3 +508,52 @@ func test_ranked_profile_projects_and_bounds_season_countdown() -> void: assert_true(not profile.apply({"rating": 1500.0, "rd": 200.0, "volatility": 0.06, "ranked_games": 10, "tier": "GOLD", "provisional": false, "season_ends_at": 123}), "non-string season expiry is rejected") assert_true(not profile.apply({"rating": 1500.0, "rd": 200.0, "volatility": 0.06, "ranked_games": 10, "tier": "GOLD", "provisional": false, "season_id": "short"}), "short season identifier is rejected") assert_true(not profile.apply({"rating": 1500.0, "rd": 200.0, "volatility": 0.06, "ranked_games": 10, "tier": "GOLD", "provisional": false, "season_id": 123}), "non-string season identifier is rejected") + + +# The client had no probe support at all, so even with the backend wired a real +# player could never acquire the RTT evidence the matcher requires. +func test_probe_region_validation_rejects_unknown_regions() -> void: + assert_true(ControlPlaneClient.is_valid_probe_region("EU"), "EU is a placement region") + assert_true(ControlPlaneClient.is_valid_probe_region("NA"), "NA is a placement region") + for region in ["", "eu", "APAC", "EU/NA", "../EU"]: + assert_true(not ControlPlaneClient.is_valid_probe_region(region), "rejects %s" % region) + + +func test_probe_requests_require_a_session() -> void: + var client = ControlPlaneClient.new() + client.base_url = "http://127.0.0.1:8080" + client.access_token = "" + assert_eq(client.request_probe_challenge("EU"), ERR_UNAUTHORIZED, "probing without a session is refused") + assert_eq(client.submit_probe_answer("EU", "bm9uY2U=", "bG9j"), ERR_UNAUTHORIZED, "answering without a session is refused") + client.free() + + +func test_probe_answer_rejects_empty_nonce_or_location() -> void: + var client = ControlPlaneClient.new() + client.base_url = "http://127.0.0.1:8080" + client.access_token = "session-1234567890:token-1234567890" + assert_eq(client.submit_probe_answer("EU", "", "bG9j"), ERR_INVALID_PARAMETER, "an empty nonce is refused") + assert_eq(client.submit_probe_answer("EU", "bm9uY2U=", ""), ERR_INVALID_PARAMETER, "an empty location is refused") + assert_eq(client.request_probe_challenge("APAC"), ERR_INVALID_PARAMETER, "an unknown region is refused") + client.free() + + +func test_opaque_location_payload_is_never_empty() -> void: + # The backend rejects an empty opaque location, and without a Steam runtime + # there is nothing real to report -- but the RTT the backend measures is + # what actually drives placement, so the probe must still be answerable. + var payload := ControlPlaneClient.opaque_location_payload() + assert_true(not payload.is_empty(), "a probe answer always carries a location blob") + assert_true(not Marshalls.base64_to_raw(payload).is_empty(), "the location blob is valid base64") + + +func test_probe_challenge_response_without_a_nonce_is_a_failure() -> void: + var client = ControlPlaneClient.new() + client.base_url = "http://127.0.0.1:8080" + client.access_token = "session-1234567890:token-1234567890" + var failures: Array = [] + client.request_failed.connect(func(operation: String, _code: int, detail: String): failures.append([operation, detail])) + client._operation = "probe_challenge_EU" + client._on_request_completed(HTTPRequest.RESULT_SUCCESS, 201, PackedStringArray(), JSON.stringify({"region": "EU"}).to_utf8_buffer()) + assert_eq(failures.size(), 1, "a challenge with no nonce is reported as a failure") + client.free() diff --git a/server/allocator/allocator_integration_test.go b/server/allocator/allocator_integration_test.go index 1d795678..1adcfa89 100644 --- a/server/allocator/allocator_integration_test.go +++ b/server/allocator/allocator_integration_test.go @@ -35,7 +35,7 @@ func TestRealAllocatorWorkerReconcilesAgonesAllocationAndBindsMatch(t *testing.T if err := db.PingContext(ctx); err != nil { t.Fatal(err) } - if _, err := db.ExecContext(ctx, `DROP TABLE IF EXISTS schema_migrations, allocation_quotas, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil { + if _, err := db.ExecContext(ctx, `DROP SCHEMA public CASCADE; CREATE SCHEMA public`); err != nil { t.Fatal(err) } if err := migrations.Apply(ctx, db, filepath.Join("..", "migrations")); err != nil { @@ -135,7 +135,7 @@ func TestRealAllocatorWorkerPublishesSignedAssignmentRoster(t *testing.T) { } defer db.Close() ctx := context.Background() - if _, err := db.ExecContext(ctx, `DROP TABLE IF EXISTS schema_migrations, allocation_quotas, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil { + if _, err := db.ExecContext(ctx, `DROP SCHEMA public CASCADE; CREATE SCHEMA public`); err != nil { t.Fatal(err) } if err := migrations.Apply(ctx, db, filepath.Join("..", "migrations")); err != nil { diff --git a/server/api/service.go b/server/api/service.go index 87beb393..99e2056d 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -26,7 +26,14 @@ const maxBodyBytes = 8 << 10 type CandidateProvider func(playerID, ticketID string) (domain.Candidate, error) type CandidateProviderV2 func(playerID, ticketID string, spec domain.QueueSpec) (domain.Candidate, error) -type ProbeProvider func(playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) +// ProbeProvider validates a probe answer against the nonce the backend issued +// and returns evidence whose ServerRTT is derived from backend timestamps +// only. It takes a context because the issued nonce is durable: any replica +// may serve the submission for a challenge another replica issued. +type ProbeProvider func(ctx context.Context, playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) + +// ProbeChallengeIssuer mints the nonce a client must echo back. +type ProbeChallengeIssuer func(ctx context.Context, playerID, region string, now time.Time) ([]byte, error) type ProbeRecorder interface { RecordProbe(context.Context, string, string, time.Duration, time.Time) error } @@ -124,6 +131,10 @@ type Service struct { // row never receives the event. EventFanout func(ControlPlaneEvent) error Probe ProbeProvider + ProbeChallenger ProbeChallengeIssuer + // CandidateRefresh re-reads a player's durable queue candidate so the + // transient index can be corrected after its RTT changes. + CandidateRefresh func(context.Context, string, time.Time) (domain.Candidate, bool, error) ProbeRecorder ProbeRecorder WorkloadVerify WorkloadVerifier ResultSubmitter ResultSubmitter @@ -216,7 +227,7 @@ func (s *Service) Handler() http.Handler { mux.HandleFunc("/v1/proposals/", s.proposalMutation) mux.HandleFunc("/v1/assignments/", s.assignment) mux.HandleFunc("/v1/profile/ranked", s.rankedProfile) - mux.HandleFunc("/v1/probes/", s.probe) + mux.HandleFunc("/v1/probes/", s.probeRoute) mux.HandleFunc("/v1/events", s.controlPlaneEvent) mux.HandleFunc("/v1/servers/", s.serverMutation) // The public contract is served below /api/v1. Keep the original /v1 @@ -1147,7 +1158,50 @@ type probeRequest struct { Nonce []byte `json:"nonce"` } -func (s *Service) probe(w http.ResponseWriter, r *http.Request) { +// probeRoute splits /v1/probes/{region} from /v1/probes/{region}/challenge. +// The challenge must exist for the submission to mean anything: RTT is the +// interval between the backend issuing a nonce and receiving the answer, so +// without an issued nonce there is nothing to compare against and no +// backend-derived latency to record. +func (s *Service) probeRoute(w http.ResponseWriter, r *http.Request) { + path := strings.TrimPrefix(r.URL.Path, "/v1/probes/") + if strings.HasSuffix(path, "/challenge") { + s.probeChallenge(w, r, strings.TrimSuffix(path, "/challenge")) + return + } + s.probe(w, r, path) +} + +func (s *Service) probeChallenge(w http.ResponseWriter, r *http.Request, region string) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "method_not_allowed") + return + } + playerID, ok := s.authenticate(w, r) + if !ok { + return + } + if region != "EU" && region != "NA" { + writeError(w, http.StatusNotFound, "not_found") + return + } + if s.ProbeChallenger == nil { + writeError(w, http.StatusServiceUnavailable, "probe_unavailable") + return + } + now := s.now() + nonce, err := s.ProbeChallenger(r.Context(), playerID, region, now) + if err != nil || len(nonce) == 0 { + writeError(w, http.StatusServiceUnavailable, "probe_unavailable") + return + } + writeJSON(w, http.StatusCreated, map[string]any{ + "region": region, "nonce": nonce, + "expires_in_seconds": int(domain.ProbeFreshness.Seconds()), + }) +} + +func (s *Service) probe(w http.ResponseWriter, r *http.Request, region string) { if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "method_not_allowed") return @@ -1156,7 +1210,6 @@ func (s *Service) probe(w http.ResponseWriter, r *http.Request) { if !ok { return } - region := strings.TrimPrefix(r.URL.Path, "/v1/probes/") if (region != "EU" && region != "NA") || strings.Contains(region, "/") { writeError(w, http.StatusNotFound, "not_found") return @@ -1170,7 +1223,7 @@ func (s *Service) probe(w http.ResponseWriter, r *http.Request) { return } receivedAt := s.now() - evidence, expectedNonce, err := s.Probe(playerID, region, input.OpaqueLocation, input.Nonce, receivedAt) + evidence, expectedNonce, err := s.Probe(r.Context(), playerID, region, input.OpaqueLocation, input.Nonce, receivedAt) if err != nil { writeError(w, http.StatusUnprocessableEntity, "probe_unavailable") return @@ -1179,12 +1232,23 @@ func (s *Service) probe(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusUnprocessableEntity, "invalid_probe") return } - if s.ProbeRecorder != nil { - if err := s.ProbeRecorder.RecordProbe(r.Context(), playerID, region, evidence.ServerRTT, receivedAt); err != nil { - writeError(w, http.StatusServiceUnavailable, "probe_persistence_failed") - return - } + // Accepting a probe without persisting it used to look like success while + // leaving predicted_rtt empty, which silently keeps the ticket invisible + // to the matcher. A missing recorder is a misconfiguration, not a + // successful probe. + if s.ProbeRecorder == nil { + writeError(w, http.StatusServiceUnavailable, "probe_unavailable") + return } + if err := s.ProbeRecorder.RecordProbe(r.Context(), playerID, region, evidence.ServerRTT, receivedAt); err != nil { + writeError(w, http.StatusServiceUnavailable, "probe_persistence_failed") + return + } + // Refresh the transient projection. A candidate inserted at enqueue time + // carries an empty RTT map, and the Redis keyspace has its TTL + // continually refreshed, so without this the stale candidate need never + // repair itself and stays unmatchable despite a successful probe. + s.refreshCandidateAfterProbe(r.Context(), playerID, receivedAt) writeJSON(w, http.StatusAccepted, map[string]any{"region": region, "server_rtt_ms": evidence.ServerRTT.Milliseconds(), "status": "accepted"}) } @@ -1274,3 +1338,18 @@ func writeJSON(w http.ResponseWriter, status int, value any) { w.WriteHeader(status) _ = json.NewEncoder(w).Encode(value) } + +// refreshCandidateAfterProbe repairs the transient candidate index once a +// probe has changed the durable predicted RTT. It is best-effort: the index is +// an acceleration layer over PostgreSQL authority, and the probe itself has +// already committed. +func (s *Service) refreshCandidateAfterProbe(ctx context.Context, playerID string, now time.Time) { + if s.CandidateIndex == nil || s.CandidateRefresh == nil { + return + } + candidate, queued, err := s.CandidateRefresh(ctx, playerID, now) + if err != nil || !queued { + return + } + _ = s.CandidateIndex.Upsert(ctx, candidate) +} diff --git a/server/api/service_test.go b/server/api/service_test.go index 66001914..41744996 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -1759,7 +1759,10 @@ func TestProbeAPIUsesServerEvidenceAndRejectsClientRTTField(t *testing.T) { t.Fatal(err) } called := false - service := &Service{Sessions: sessions, Now: func() time.Time { return now }, Probe: func(playerID, region string, location, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) { + // A ProbeRecorder is required: accepting a probe without persisting it + // reports success while leaving predicted_rtt empty, which silently keeps + // the ticket invisible to the matcher. + service := &Service{Sessions: sessions, Now: func() time.Time { return now }, ProbeRecorder: &probeRecorderSpy{}, Probe: func(_ context.Context, playerID, region string, location, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) { called = true if playerID != "player-a" || region != "EU" || string(location) != "opaque" || string(nonce) != "nonce" || !receivedAt.Equal(now) { t.Fatalf("probe provider arguments = %q %s %q %q %v", playerID, region, location, nonce, receivedAt) @@ -1791,7 +1794,7 @@ func TestProbeAPIRecordsOnlyValidatedServerEvidence(t *testing.T) { sessions := domain.NewSessionStore() session, token, _ := sessions.Issue("player-a", time.Hour, now) recorder := &probeRecorderSpy{} - service := &Service{Sessions: sessions, Now: func() time.Time { return now }, ProbeRecorder: recorder, Probe: func(_ string, region string, location, nonce []byte, _ time.Time) (domain.ProbeEvidence, []byte, error) { + service := &Service{Sessions: sessions, Now: func() time.Time { return now }, ProbeRecorder: recorder, Probe: func(_ context.Context, _ string, region string, location, nonce []byte, _ time.Time) (domain.ProbeEvidence, []byte, error) { return domain.ProbeEvidence{OpaqueLocation: location, Nonce: nonce, IssuedAt: now, Region: region, ServerRTT: 37 * time.Millisecond}, nonce, nil }} server := httptest.NewServer(service.Handler()) diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index daf32a0d..5867463b 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -171,6 +171,21 @@ func newAPIService(db *sql.DB, workloadSecret string, indexes ...api.CandidateIn }, CandidateIndex: candidateIndex, ProbeRecorder: store.PostgresQueue{DB: db}, + // Regional latency placement. Without both of these the probe endpoint + // is unreachable, queue_tickets.predicted_rtt stays empty, and + // domain.validCandidate rejects every client-created ticket -- so the + // matcher can never form a match from real traffic. + ProbeChallenger: func(ctx context.Context, playerID, region string, now time.Time) ([]byte, error) { + return store.IssueProbeChallenge(ctx, db, playerID, region, now) + }, + Probe: func(ctx context.Context, playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) { + return store.ProbeEvidenceFromChallenge(ctx, db, playerID, region, opaqueLocation, nonce, receivedAt) + }, + // Repairs the transient index after a probe changes the durable RTT; + // the candidate inserted at enqueue time has an empty map. + CandidateRefresh: func(ctx context.Context, playerID string, now time.Time) (domain.Candidate, bool, error) { + return store.FindQueuedCandidateByPlayer(ctx, db, playerID, now) + }, WorkloadVerify: api.WorkloadVerifierFromSignedToken([]byte(workloadSecret), db), ReadinessCheck: db.PingContext, Now: func() time.Time { return time.Now().UTC() }, diff --git a/server/cmd/maintenance/main.go b/server/cmd/maintenance/main.go index 1d13e64c..65ce0172 100644 --- a/server/cmd/maintenance/main.go +++ b/server/cmd/maintenance/main.go @@ -89,6 +89,13 @@ func main() { if backlog > 0 { log.Printf("retention backlog is %d rows past their window", backlog) } + staleProbes, err := store.PurgeExpiredProbeChallenges(ctx, db, now) + if err != nil { + fatalf("probe challenge maintenance: %v", err) + } + if staleProbes > 0 { + log.Printf("purged %d unanswered probe challenges", staleProbes) + } deadLettered, err := store.CountDeadLetteredOutboxEvents(ctx, db) if err != nil { fatalf("dead-letter count: %v", err) diff --git a/server/contracts/v1/openapi.json b/server/contracts/v1/openapi.json index beddb30c..53c3d85f 100644 --- a/server/contracts/v1/openapi.json +++ b/server/contracts/v1/openapi.json @@ -5,104 +5,1212 @@ "version": "1.0.0", "description": "Versioned control-plane contract. Simulation traffic never uses this API." }, - "servers": [{"url": "https://matchmaking.invalid/api/v1"}], - "security": [{"bearerAuth": []}], + "servers": [ + { + "url": "https://matchmaking.invalid/api/v1" + } + ], + "security": [ + { + "bearerAuth": [] + } + ], "paths": { "/session/steam": { "post": { "security": [], "operationId": "createSteamSession", - "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/SteamLogin"}}}}, - "responses": {"200": {"description": "Session created", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Session"}}}}, "401": {"$ref": "#/components/responses/Unauthorized"}, "429": {"$ref": "#/components/responses/RateLimited"}} + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SteamLogin" + } + } + } + }, + "responses": { + "200": { + "description": "Session created", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Session" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "429": { + "$ref": "#/components/responses/RateLimited" + } + } } }, "/profile": { - "get": {"operationId": "getProfile", "responses": {"200": {"description": "Profile", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Profile"}}}}, "401": {"$ref": "#/components/responses/Unauthorized"}}} + "get": { + "operationId": "getProfile", + "responses": { + "200": { + "description": "Profile", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Profile" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + } + } + } }, "/profile/ranked": { - "get": {"operationId": "getRankedProfile", "responses": {"200": {"description": "Authoritative ranked profile", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/RankedProfile"}}}}, "401": {"$ref": "#/components/responses/Unauthorized"}, "404": {"$ref": "#/components/responses/NotFound"}, "503": {"$ref": "#/components/responses/Unavailable"}}} + "get": { + "operationId": "getRankedProfile", + "responses": { + "200": { + "description": "Authoritative ranked profile", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RankedProfile" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "404": { + "$ref": "#/components/responses/NotFound" + }, + "503": { + "$ref": "#/components/responses/Unavailable" + } + } + } }, "/queue/tickets": { "post": { "operationId": "createQueueTicket", - "parameters": [{"$ref": "#/components/parameters/IdempotencyKey"}], - "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/QueueCreate"}}}}, - "responses": {"201": {"description": "Ticket created", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/QueueTicket"}}}}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}} + "parameters": [ + { + "$ref": "#/components/parameters/IdempotencyKey" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/QueueCreate" + } + } + } + }, + "responses": { + "201": { + "description": "Ticket created", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/QueueTicket" + } + } + } + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "422": { + "$ref": "#/components/responses/Invalid" + } + } } }, "/queue/tickets/{ticketId}": { - "parameters": [{"$ref": "#/components/parameters/TicketId"}], - "get": {"operationId": "getQueueTicket", "responses": {"200": {"description": "Ticket", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/QueueTicket"}}}}, "404": {"$ref": "#/components/responses/NotFound"}}}, - "delete": {"operationId": "cancelQueueTicket", "parameters": [{"$ref": "#/components/parameters/IdempotencyKey"}, {"$ref": "#/components/parameters/ExpectedRevision"}], "responses": {"204": {"description": "Cancelled"}, "409": {"$ref": "#/components/responses/Conflict"}}} + "parameters": [ + { + "$ref": "#/components/parameters/TicketId" + } + ], + "get": { + "operationId": "getQueueTicket", + "responses": { + "200": { + "description": "Ticket", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/QueueTicket" + } + } + } + }, + "404": { + "$ref": "#/components/responses/NotFound" + } + } + }, + "delete": { + "operationId": "cancelQueueTicket", + "parameters": [ + { + "$ref": "#/components/parameters/IdempotencyKey" + }, + { + "$ref": "#/components/parameters/ExpectedRevision" + } + ], + "responses": { + "204": { + "description": "Cancelled" + }, + "409": { + "$ref": "#/components/responses/Conflict" + } + } + } }, "/queue/tickets/{ticketId}/heartbeat": { - "post": {"operationId": "heartbeatQueueTicket", "parameters": [{"$ref": "#/components/parameters/TicketId"}, {"$ref": "#/components/parameters/IdempotencyKey"}, {"$ref": "#/components/parameters/ExpectedRevision"}], "responses": {"200": {"description": "Ticket renewed", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/QueueTicket"}}}}, "409": {"$ref": "#/components/responses/Conflict"}}} + "post": { + "operationId": "heartbeatQueueTicket", + "parameters": [ + { + "$ref": "#/components/parameters/TicketId" + }, + { + "$ref": "#/components/parameters/IdempotencyKey" + }, + { + "$ref": "#/components/parameters/ExpectedRevision" + } + ], + "responses": { + "200": { + "description": "Ticket renewed", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/QueueTicket" + } + } + } + }, + "409": { + "$ref": "#/components/responses/Conflict" + } + } + } }, "/proposals/{proposalId}/accept": { - "post": {"operationId": "acceptProposal", "parameters": [{"$ref": "#/components/parameters/ProposalId"}, {"$ref": "#/components/parameters/IdempotencyKey"}, {"$ref": "#/components/parameters/ExpectedRevision"}], "responses": {"200": {"description": "Proposal updated", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Proposal"}}}}, "409": {"$ref": "#/components/responses/Conflict"}, "410": {"$ref": "#/components/responses/Expired"}}} + "post": { + "operationId": "acceptProposal", + "parameters": [ + { + "$ref": "#/components/parameters/ProposalId" + }, + { + "$ref": "#/components/parameters/IdempotencyKey" + }, + { + "$ref": "#/components/parameters/ExpectedRevision" + } + ], + "responses": { + "200": { + "description": "Proposal updated", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Proposal" + } + } + } + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "410": { + "$ref": "#/components/responses/Expired" + } + } + } }, "/proposals/{proposalId}/decline": { - "post": {"operationId": "declineProposal", "parameters": [{"$ref": "#/components/parameters/ProposalId"}, {"$ref": "#/components/parameters/IdempotencyKey"}, {"$ref": "#/components/parameters/ExpectedRevision"}], "responses": {"200": {"description": "Proposal declined", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Proposal"}}}}, "409": {"$ref": "#/components/responses/Conflict"}, "410": {"$ref": "#/components/responses/Expired"}}} + "post": { + "operationId": "declineProposal", + "parameters": [ + { + "$ref": "#/components/parameters/ProposalId" + }, + { + "$ref": "#/components/parameters/IdempotencyKey" + }, + { + "$ref": "#/components/parameters/ExpectedRevision" + } + ], + "responses": { + "200": { + "description": "Proposal declined", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Proposal" + } + } + } + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "410": { + "$ref": "#/components/responses/Expired" + } + } + } }, "/assignments/{matchId}": { - "get": {"operationId": "getAssignment", "parameters": [{"$ref": "#/components/parameters/MatchId"}], "responses": {"200": {"description": "Assignment", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Assignment"}}}}, "404": {"$ref": "#/components/responses/NotFound"}}} + "get": { + "operationId": "getAssignment", + "parameters": [ + { + "$ref": "#/components/parameters/MatchId" + } + ], + "responses": { + "200": { + "description": "Assignment", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Assignment" + } + } + } + }, + "404": { + "$ref": "#/components/responses/NotFound" + } + } + } }, "/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"}}} + "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": "claimPlayerConnection", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnectionClaim"}}}}, "responses": {"200": {"description": "Connection generation claimed", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnectionLease"}}}}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}, "503": {"$ref": "#/components/responses/Unavailable"}}} + "post": { + "security": [ + { + "serverCredential": [] + } + ], + "operationId": "claimPlayerConnection", + "parameters": [ + { + "$ref": "#/components/parameters/ServerId" + }, + { + "$ref": "#/components/parameters/IdempotencyKey" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ServerConnectionClaim" + } + } + } + }, + "responses": { + "200": { + "description": "Connection generation claimed", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ServerConnectionLease" + } + } + } + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "422": { + "$ref": "#/components/responses/Invalid" + }, + "503": { + "$ref": "#/components/responses/Unavailable" + } + } + } }, "/servers/{serverId}/disconnect": { - "post": {"security": [{"serverCredential": []}], "operationId": "recordPlayerDisconnected", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnectionDisconnect"}}}}, "responses": {"204": {"description": "Disconnection recorded"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}, "503": {"$ref": "#/components/responses/Unavailable"}}} + "post": { + "security": [ + { + "serverCredential": [] + } + ], + "operationId": "recordPlayerDisconnected", + "parameters": [ + { + "$ref": "#/components/parameters/ServerId" + }, + { + "$ref": "#/components/parameters/IdempotencyKey" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ServerConnectionDisconnect" + } + } + } + }, + "responses": { + "204": { + "description": "Disconnection 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"}}} + "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" + } + } + } }, "/servers/{serverId}/shutdown": { - "post": {"security": [{"serverCredential": []}], "operationId": "acknowledgeServerShutdown", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerShutdown"}}}}, "responses": {"204": {"description": "Shutdown acknowledged"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}}} + "post": { + "security": [ + { + "serverCredential": [] + } + ], + "operationId": "acknowledgeServerShutdown", + "parameters": [ + { + "$ref": "#/components/parameters/ServerId" + }, + { + "$ref": "#/components/parameters/IdempotencyKey" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ServerShutdown" + } + } + } + }, + "responses": { + "204": { + "description": "Shutdown acknowledged" + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "422": { + "$ref": "#/components/responses/Invalid" + } + } + } + }, + "/probes/{region}/challenge": { + "post": { + "operationId": "createProbeChallenge", + "summary": "Issue a single-use latency probe challenge for one region.", + "parameters": [ + { + "$ref": "#/components/parameters/Region" + } + ], + "responses": { + "201": { + "description": "Challenge issued", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProbeChallenge" + } + } + } + } + } + } + }, + "/probes/{region}": { + "post": { + "operationId": "submitProbeAnswer", + "summary": "Answer a probe challenge so the backend can record regional latency.", + "parameters": [ + { + "$ref": "#/components/parameters/Region" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProbeAnswer" + } + } + } + }, + "responses": { + "202": { + "description": "Probe accepted and recorded", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProbeAccepted" + } + } + } + } + } + } } }, "components": { "securitySchemes": { - "bearerAuth": {"type": "http", "scheme": "bearer"}, - "serverCredential": {"type": "http", "scheme": "bearer", "bearerFormat": "match-bound workload credential"} + "bearerAuth": { + "type": "http", + "scheme": "bearer" + }, + "serverCredential": { + "type": "http", + "scheme": "bearer", + "bearerFormat": "match-bound workload credential" + } }, "parameters": { - "IdempotencyKey": {"name": "Idempotency-Key", "in": "header", "required": true, "schema": {"type": "string", "minLength": 16, "maxLength": 128}}, - "ExpectedRevision": {"name": "If-Match-Revision", "in": "header", "required": true, "schema": {"type": "integer", "minimum": 0}}, - "TicketId": {"name": "ticketId", "in": "path", "required": true, "schema": {"$ref": "#/components/schemas/OpaqueId"}}, - "ProposalId": {"name": "proposalId", "in": "path", "required": true, "schema": {"$ref": "#/components/schemas/OpaqueId"}}, - "MatchId": {"name": "matchId", "in": "path", "required": true, "schema": {"$ref": "#/components/schemas/OpaqueId"}}, - "ServerId": {"name": "serverId", "in": "path", "required": true, "schema": {"$ref": "#/components/schemas/OpaqueId"}} + "IdempotencyKey": { + "name": "Idempotency-Key", + "in": "header", + "required": true, + "schema": { + "type": "string", + "minLength": 16, + "maxLength": 128 + } + }, + "ExpectedRevision": { + "name": "If-Match-Revision", + "in": "header", + "required": true, + "schema": { + "type": "integer", + "minimum": 0 + } + }, + "TicketId": { + "name": "ticketId", + "in": "path", + "required": true, + "schema": { + "$ref": "#/components/schemas/OpaqueId" + } + }, + "ProposalId": { + "name": "proposalId", + "in": "path", + "required": true, + "schema": { + "$ref": "#/components/schemas/OpaqueId" + } + }, + "MatchId": { + "name": "matchId", + "in": "path", + "required": true, + "schema": { + "$ref": "#/components/schemas/OpaqueId" + } + }, + "ServerId": { + "name": "serverId", + "in": "path", + "required": true, + "schema": { + "$ref": "#/components/schemas/OpaqueId" + } + }, + "Region": { + "name": "region", + "in": "path", + "required": true, + "description": "Placement region the probe measures.", + "schema": { + "type": "string", + "enum": [ + "EU", + "NA" + ] + } + } }, "responses": { - "Unauthorized": {"description": "Authentication failed"}, - "RateLimited": {"description": "Rate limit exceeded"}, - "Conflict": {"description": "Revision or idempotency conflict"}, - "Invalid": {"description": "Invalid state or schema"}, - "NotFound": {"description": "Resource not found"}, - "Unavailable": {"description": "Authoritative profile temporarily unavailable"}, - "Expired": {"description": "Resource expired"} + "Unauthorized": { + "description": "Authentication failed" + }, + "RateLimited": { + "description": "Rate limit exceeded" + }, + "Conflict": { + "description": "Revision or idempotency conflict" + }, + "Invalid": { + "description": "Invalid state or schema" + }, + "NotFound": { + "description": "Resource not found" + }, + "Unavailable": { + "description": "Authoritative profile temporarily unavailable" + }, + "Expired": { + "description": "Resource expired" + } }, "schemas": { - "OpaqueId": {"type": "string", "pattern": "^[A-Za-z0-9_-]{16,128}$"}, - "SteamLogin": {"type": "object", "required": ["web_api_ticket"], "additionalProperties": false, "properties": {"web_api_ticket": {"type": "string", "minLength": 1, "maxLength": 4096}}}, - "Session": {"type": "object", "required": ["player_id", "expires_at", "access_token"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "expires_at": {"type": "string", "format": "date-time"}, "access_token": {"type": "string"}}}, - "Profile": {"type": "object", "required": ["player_id", "rating", "rd", "provisional"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "rating": {"type": "number"}, "rd": {"type": "number"}, "provisional": {"type": "boolean"}}}, - "RankedProfile": {"type": "object", "required": ["rating", "rd", "volatility", "ranked_games", "tier", "provisional"], "additionalProperties": false, "properties": {"rating": {"type": "number", "minimum": 0}, "rd": {"type": "number", "minimum": 0}, "volatility": {"type": "number", "minimum": 0}, "ranked_games": {"type": "integer", "minimum": 0}, "tier": {"type": "string", "enum": ["PROVISIONAL", "BRONZE", "SILVER", "GOLD", "PLATINUM", "DIAMOND"]}, "provisional": {"type": "boolean"}, "season_id": {"$ref": "#/components/schemas/OpaqueId"}, "season_ends_at": {"type": "string", "format": "date-time"}}}, - "QueueCreate": {"type": "object", "required": ["playlist", "client_build", "protocol_version"], "additionalProperties": false, "properties": {"playlist": {"type": "string", "enum": ["casual", "ranked"]}, "client_build": {"type": "string", "minLength": 1, "maxLength": 128}, "protocol_version": {"type": "integer", "minimum": 1}}}, - "QueueTicket": {"type": "object", "required": ["ticket_id", "player_id", "playlist", "state", "revision", "enqueued_at", "expires_at"], "additionalProperties": false, "properties": {"ticket_id": {"$ref": "#/components/schemas/OpaqueId"}, "player_id": {"$ref": "#/components/schemas/OpaqueId"}, "proposal_id": {"$ref": "#/components/schemas/OpaqueId"}, "match_id": {"$ref": "#/components/schemas/OpaqueId"}, "playlist": {"type": "string", "enum": ["casual", "ranked"]}, "state": {"$ref": "#/components/schemas/QueueState"}, "revision": {"type": "integer", "minimum": 0}, "enqueued_at": {"type": "string", "format": "date-time"}, "expires_at": {"type": "string", "format": "date-time"}}}, - "QueueState": {"type": "string", "enum": ["QUEUED", "PROPOSED", "ACCEPTED", "ALLOCATING", "PROCESS_READY", "ASSIGNMENT_READY", "ASSIGNED", "CONNECTING", "LIVE", "RESULT_PENDING", "COMPLETED", "CANCELLED", "EXPIRED", "FAILED"]}, - "Proposal": {"type": "object", "required": ["proposal_id", "revision", "state", "expires_at", "participants"], "additionalProperties": false, "properties": {"proposal_id": {"$ref": "#/components/schemas/OpaqueId"}, "revision": {"type": "integer", "minimum": 0}, "state": {"type": "string", "enum": ["OPEN", "ACCEPTED", "DECLINED", "EXPIRED", "CANCELLED"]}, "expires_at": {"type": "string", "format": "date-time"}, "participants": {"type": "array", "minItems": 2, "maxItems": 6, "items": {"$ref": "#/components/schemas/ProposalParticipant"}}}}, - "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"}}}, - "ServerConnectionClaim": {"type": "object", "required": ["player_id", "expected_generation"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "expected_generation": {"type": "integer", "minimum": 0}}}, - "ServerConnectionDisconnect": {"type": "object", "required": ["player_id", "generation"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "generation": {"type": "integer", "minimum": 1}}}, - "ServerConnectionLease": {"type": "object", "required": ["generation"], "additionalProperties": false, "properties": {"generation": {"type": "integer", "minimum": 1}}}, - "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"]}}} + "OpaqueId": { + "type": "string", + "pattern": "^[A-Za-z0-9_-]{16,128}$" + }, + "SteamLogin": { + "type": "object", + "required": [ + "web_api_ticket" + ], + "additionalProperties": false, + "properties": { + "web_api_ticket": { + "type": "string", + "minLength": 1, + "maxLength": 4096 + } + } + }, + "Session": { + "type": "object", + "required": [ + "player_id", + "expires_at", + "access_token" + ], + "additionalProperties": false, + "properties": { + "player_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "expires_at": { + "type": "string", + "format": "date-time" + }, + "access_token": { + "type": "string" + } + } + }, + "Profile": { + "type": "object", + "required": [ + "player_id", + "rating", + "rd", + "provisional" + ], + "additionalProperties": false, + "properties": { + "player_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "rating": { + "type": "number" + }, + "rd": { + "type": "number" + }, + "provisional": { + "type": "boolean" + } + } + }, + "RankedProfile": { + "type": "object", + "required": [ + "rating", + "rd", + "volatility", + "ranked_games", + "tier", + "provisional" + ], + "additionalProperties": false, + "properties": { + "rating": { + "type": "number", + "minimum": 0 + }, + "rd": { + "type": "number", + "minimum": 0 + }, + "volatility": { + "type": "number", + "minimum": 0 + }, + "ranked_games": { + "type": "integer", + "minimum": 0 + }, + "tier": { + "type": "string", + "enum": [ + "PROVISIONAL", + "BRONZE", + "SILVER", + "GOLD", + "PLATINUM", + "DIAMOND" + ] + }, + "provisional": { + "type": "boolean" + }, + "season_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "season_ends_at": { + "type": "string", + "format": "date-time" + } + } + }, + "QueueCreate": { + "type": "object", + "required": [ + "playlist", + "client_build", + "protocol_version" + ], + "additionalProperties": false, + "properties": { + "playlist": { + "type": "string", + "enum": [ + "casual", + "ranked" + ] + }, + "client_build": { + "type": "string", + "minLength": 1, + "maxLength": 128 + }, + "protocol_version": { + "type": "integer", + "minimum": 1 + } + } + }, + "QueueTicket": { + "type": "object", + "required": [ + "ticket_id", + "player_id", + "playlist", + "state", + "revision", + "enqueued_at", + "expires_at" + ], + "additionalProperties": false, + "properties": { + "ticket_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "player_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "proposal_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "match_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "playlist": { + "type": "string", + "enum": [ + "casual", + "ranked" + ] + }, + "state": { + "$ref": "#/components/schemas/QueueState" + }, + "revision": { + "type": "integer", + "minimum": 0 + }, + "enqueued_at": { + "type": "string", + "format": "date-time" + }, + "expires_at": { + "type": "string", + "format": "date-time" + } + } + }, + "QueueState": { + "type": "string", + "enum": [ + "QUEUED", + "PROPOSED", + "ACCEPTED", + "ALLOCATING", + "PROCESS_READY", + "ASSIGNMENT_READY", + "ASSIGNED", + "CONNECTING", + "LIVE", + "RESULT_PENDING", + "COMPLETED", + "CANCELLED", + "EXPIRED", + "FAILED" + ] + }, + "Proposal": { + "type": "object", + "required": [ + "proposal_id", + "revision", + "state", + "expires_at", + "participants" + ], + "additionalProperties": false, + "properties": { + "proposal_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "revision": { + "type": "integer", + "minimum": 0 + }, + "state": { + "type": "string", + "enum": [ + "OPEN", + "ACCEPTED", + "DECLINED", + "EXPIRED", + "CANCELLED" + ] + }, + "expires_at": { + "type": "string", + "format": "date-time" + }, + "participants": { + "type": "array", + "minItems": 2, + "maxItems": 6, + "items": { + "$ref": "#/components/schemas/ProposalParticipant" + } + } + } + }, + "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" + } + } + }, + "ServerConnectionClaim": { + "type": "object", + "required": [ + "player_id", + "expected_generation" + ], + "additionalProperties": false, + "properties": { + "player_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "expected_generation": { + "type": "integer", + "minimum": 0 + } + } + }, + "ServerConnectionDisconnect": { + "type": "object", + "required": [ + "player_id", + "generation" + ], + "additionalProperties": false, + "properties": { + "player_id": { + "$ref": "#/components/schemas/OpaqueId" + }, + "generation": { + "type": "integer", + "minimum": 1 + } + } + }, + "ServerConnectionLease": { + "type": "object", + "required": [ + "generation" + ], + "additionalProperties": false, + "properties": { + "generation": { + "type": "integer", + "minimum": 1 + } + } + }, + "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" + ] + } + } + }, + "ProbeChallenge": { + "type": "object", + "required": [ + "region", + "nonce", + "expires_in_seconds" + ], + "additionalProperties": false, + "properties": { + "region": { + "type": "string", + "enum": [ + "EU", + "NA" + ] + }, + "nonce": { + "type": "string", + "format": "byte", + "description": "Single-use value the client must echo back with its probe answer." + }, + "expires_in_seconds": { + "type": "integer" + } + } + }, + "ProbeAnswer": { + "type": "object", + "required": [ + "opaque_location", + "nonce" + ], + "additionalProperties": false, + "description": "No client-measured latency is accepted: the backend derives RTT from the interval between issuing the challenge and receiving this answer.", + "properties": { + "opaque_location": { + "type": "string", + "format": "byte" + }, + "nonce": { + "type": "string", + "format": "byte" + } + } + }, + "ProbeAccepted": { + "type": "object", + "required": [ + "region", + "server_rtt_ms", + "status" + ], + "additionalProperties": false, + "properties": { + "region": { + "type": "string", + "enum": [ + "EU", + "NA" + ] + }, + "server_rtt_ms": { + "type": "integer", + "description": "Backend-computed round trip; never a client-reported value." + }, + "status": { + "type": "string", + "enum": [ + "accepted" + ] + } + } + } } } } diff --git a/server/contracts/v1/test_contracts.py b/server/contracts/v1/test_contracts.py index cb172160..e165d539 100644 --- a/server/contracts/v1/test_contracts.py +++ b/server/contracts/v1/test_contracts.py @@ -51,7 +51,12 @@ class ContractTest(unittest.TestCase): for method, operation in methods.items(): if method not in {"post", "delete", "put", "patch"} or "operationId" not in operation: continue - if operation["operationId"] == "createSteamSession": + # Exempt: these establish or consume a single-use credential + # rather than mutating a revisioned resource. A probe challenge + # is deliberately new on every call, and its answer is made + # single-use by consuming the nonce, so an idempotency key + # would be meaningless rather than protective. + if operation["operationId"] in {"createSteamSession", "createProbeChallenge", "submitProbeAnswer"}: continue refs = {item.get("$ref") for item in operation.get("parameters", [])} self.assertIn("#/components/parameters/IdempotencyKey", refs, path) diff --git a/server/migrations/0017_probe_challenges.sql b/server/migrations/0017_probe_challenges.sql new file mode 100644 index 00000000..a25e02f9 --- /dev/null +++ b/server/migrations/0017_probe_challenges.sql @@ -0,0 +1,23 @@ +-- Latency probes are nonce-bound: the backend issues a challenge, the client +-- echoes it back with its opaque Steam location, and the backend computes RTT +-- from its own send/receive timestamps rather than trusting a client-reported +-- number. +-- +-- Nothing issued that nonce before, so ProbeProvider had no expected value to +-- compare against and /v1/probes/{region} was unreachable in every real +-- binary. With no probe, queue_tickets.predicted_rtt stayed empty, and +-- domain.validCandidate hard-requires a non-empty map -- so no client-created +-- ticket could ever be selected by the matcher. +-- +-- The challenge is durable rather than per-process because any control-plane +-- replica may serve the follow-up submission. +CREATE TABLE probe_challenges ( + player_id TEXT NOT NULL REFERENCES identities(player_id) ON DELETE CASCADE, + region TEXT NOT NULL CHECK (region IN ('EU', 'NA')), + nonce BYTEA NOT NULL, + issued_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (player_id, region) +); + +-- Supports the expiry sweep; challenges are short-lived and single-use. +CREATE INDEX probe_challenges_issued_at ON probe_challenges (issued_at); diff --git a/server/migrations/down/0017_probe_challenges.sql b/server/migrations/down/0017_probe_challenges.sql new file mode 100644 index 00000000..30003026 --- /dev/null +++ b/server/migrations/down/0017_probe_challenges.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS probe_challenges; diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index cca3d90c..e6146e39 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -46,7 +46,7 @@ func openIntegrationPostgres(t *testing.T) *sql.DB { func applyIntegrationMigrations(t *testing.T, db *sql.DB) { t.Helper() - if _, err := db.ExecContext(context.Background(), `DROP TABLE IF EXISTS schema_migrations, allocation_quotas, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil { + if _, err := db.ExecContext(context.Background(), `DROP SCHEMA public CASCADE; CREATE SCHEMA public`); err != nil { t.Fatalf("reset PostgreSQL schema: %v", err) } if err := migrations.Apply(context.Background(), db, filepath.Join("..", "migrations")); err != nil { @@ -2184,3 +2184,96 @@ func countMigrationsAbove(t *testing.T, dir string, number int) int { } return count } + +// The second fatal blocker. domain.validCandidate hard-requires a non-empty +// PredictedRTT map, but CreateQueueTicket persisted an empty one and the only +// endpoint that could fill it returned 503 in every real binary because +// Service.Probe was never wired. No client-created ticket could ever be +// selected by the matcher. +// +// This drives the real enqueue and probe paths and then asks the actual +// matcher predicate, rather than hand-building a domain.Candidate the way the +// unit tests do -- which is precisely why they missed it. +func TestPostgreSQLProbedTicketBecomesSelectableByTheMatcher(t *testing.T) { + db := openIntegrationPostgres(t) + applyIntegrationMigrations(t, db) + + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Microsecond) + if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('probe-player', 'probe-steam')`); err != nil { + t.Fatal(err) + } + spec := domain.QueueSpec{Playlist: domain.Casual, ClientBuild: "probe-build", ProtocolVersion: 1} + if _, err := CreateQueueTicket(ctx, db, "probe-ticket", "probe-player", "probe-create-000001", spec, now); err != nil { + t.Fatalf("create queue ticket: %v", err) + } + + // Freshly queued: no RTT evidence yet, so the matcher must not consider it. + candidates, err := ListQueuedCandidates(ctx, db, domain.Casual, now, 100) + if err != nil { + t.Fatal(err) + } + if len(candidates) != 1 { + t.Fatalf("expected the queued ticket, got %d candidates", len(candidates)) + } + if len(candidates[0].PredictedRTT) != 0 { + t.Fatalf("a freshly queued ticket already has RTT evidence: %v", candidates[0].PredictedRTT) + } + if _, err := domain.SelectCandidates(candidates[0], candidates, 1, now); err == nil { + t.Fatal("a candidate with no RTT evidence was accepted by the matcher") + } + + // The real challenge/answer round trip: the backend issues the nonce and + // derives RTT from its own timestamps, never from a client-reported value. + nonce, err := IssueProbeChallenge(ctx, db, "probe-player", "EU", now) + if err != nil { + t.Fatalf("issue challenge: %v", err) + } + if len(nonce) != ProbeNonceBytes { + t.Fatalf("challenge nonce is %d bytes", len(nonce)) + } + answeredAt := now.Add(40 * time.Millisecond) + evidence, expectedNonce, err := ProbeEvidenceFromChallenge(ctx, db, "probe-player", "EU", []byte("opaque-location"), nonce, answeredAt) + if err != nil { + t.Fatalf("probe evidence: %v", err) + } + if err := domain.ValidateProbe(evidence, expectedNonce, answeredAt); err != nil { + t.Fatalf("backend-derived evidence failed its own validation: %v", err) + } + if evidence.ServerRTT != 40*time.Millisecond { + t.Fatalf("server-derived RTT = %v, want the 40ms round trip", evidence.ServerRTT) + } + + // A challenge is single-use, so a captured answer cannot be replayed to + // refresh a stale RTT. + if _, _, err := ProbeEvidenceFromChallenge(ctx, db, "probe-player", "EU", []byte("opaque-location"), nonce, answeredAt); err == nil { + t.Fatal("a probe challenge was answerable twice") + } + + if err := (PostgresQueue{DB: db}).RecordProbe(ctx, "probe-player", "EU", evidence.ServerRTT, answeredAt); err != nil { + t.Fatalf("record probe: %v", err) + } + + // Now the same ticket, read through the same production query, is + // selectable. + candidates, err = ListQueuedCandidates(ctx, db, domain.Casual, answeredAt, 100) + if err != nil { + t.Fatal(err) + } + if len(candidates) != 1 || candidates[0].PredictedRTT["EU"] == 0 { + t.Fatalf("probe did not reach the candidate projection: %+v", candidates) + } + if _, err := domain.SelectCandidates(candidates[0], candidates, 1, answeredAt); err != nil { + t.Fatalf("a probed, client-created ticket is still not selectable by the matcher: %v", err) + } + + // The same candidate must survive the Redis path, which is seeded from the + // per-player refresh the probe handler performs. + refreshed, queued, err := FindQueuedCandidateByPlayer(ctx, db, "probe-player", answeredAt) + if err != nil || !queued { + t.Fatalf("per-player candidate refresh: queued=%t err=%v", queued, err) + } + if refreshed.PredictedRTT["EU"] == 0 { + t.Fatal("the refreshed candidate still carries an empty RTT map, so Redis would keep a stale entry") + } +} diff --git a/server/store/probe_sql.go b/server/store/probe_sql.go new file mode 100644 index 00000000..3034c66f --- /dev/null +++ b/server/store/probe_sql.go @@ -0,0 +1,100 @@ +package store + +import ( + "context" + "crypto/rand" + "database/sql" + "fmt" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +// ProbeNonceBytes is the challenge size. It only needs to be unguessable +// within the freshness window, not long-lived key material. +const ProbeNonceBytes = 16 + +const ( + ProbeChallengeUpsertSQL = `INSERT INTO probe_challenges (player_id, region, nonce, issued_at) +VALUES ($1, $2, $3, $4) +ON CONFLICT (player_id, region) DO UPDATE +SET nonce = EXCLUDED.nonce, issued_at = EXCLUDED.issued_at` + // Consuming deletes in the same statement: a challenge is single-use, so a + // captured probe response cannot be replayed to refresh a stale RTT. + ProbeChallengeConsumeSQL = `DELETE FROM probe_challenges +WHERE player_id = $1 AND region = $2 +RETURNING nonce, issued_at` + ProbeChallengePurgeSQL = `DELETE FROM probe_challenges WHERE issued_at < $1` +) + +// IssueProbeChallenge mints and stores a fresh nonce for one player and +// region. It is durable rather than per-process because any control-plane +// replica may serve the follow-up submission. +func IssueProbeChallenge(ctx context.Context, db *sql.DB, playerID, region string, now time.Time) ([]byte, error) { + if db == nil || playerID == "" || (region != "EU" && region != "NA") || now.IsZero() { + return nil, fmt.Errorf("invalid probe challenge arguments") + } + nonce := make([]byte, ProbeNonceBytes) + if _, err := rand.Read(nonce); err != nil { + return nil, err + } + if _, err := db.ExecContext(ctx, ProbeChallengeUpsertSQL, playerID, region, nonce, now); err != nil { + return nil, err + } + return nonce, nil +} + +// ConsumeProbeChallenge returns the outstanding nonce and when it was issued, +// removing it so it cannot be reused. +func ConsumeProbeChallenge(ctx context.Context, db *sql.DB, playerID, region string) ([]byte, time.Time, error) { + if db == nil || playerID == "" || (region != "EU" && region != "NA") { + return nil, time.Time{}, fmt.Errorf("invalid probe challenge arguments") + } + var nonce []byte + var issuedAt time.Time + err := db.QueryRowContext(ctx, ProbeChallengeConsumeSQL, playerID, region).Scan(&nonce, &issuedAt) + if err == sql.ErrNoRows { + return nil, time.Time{}, domain.ErrInvalidProbe + } + if err != nil { + return nil, time.Time{}, err + } + return nonce, issuedAt, nil +} + +// PurgeExpiredProbeChallenges drops challenges that can no longer be answered +// within the freshness window, so an abandoned probe cannot accumulate. +func PurgeExpiredProbeChallenges(ctx context.Context, db *sql.DB, now time.Time) (int64, error) { + if db == nil || now.IsZero() { + return 0, fmt.Errorf("invalid probe challenge purge arguments") + } + result, err := db.ExecContext(ctx, ProbeChallengePurgeSQL, now.Add(-domain.ProbeFreshness)) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +// ProbeEvidenceFromChallenge is the production ProbeProvider. The RTT is +// derived entirely from backend timestamps -- the interval between issuing the +// challenge and receiving the answer -- so no client-reported latency +// influences placement, which is the property docs/MATCHMAKING.md ยง4 requires. +func ProbeEvidenceFromChallenge(ctx context.Context, db *sql.DB, playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) { + expectedNonce, issuedAt, err := ConsumeProbeChallenge(ctx, db, playerID, region) + if err != nil { + return domain.ProbeEvidence{}, nil, err + } + rtt := receivedAt.Sub(issuedAt) + if rtt < 0 { + // Clock skew between replicas; treat as immediate rather than letting + // a negative duration through to placement. + rtt = 0 + } + return domain.ProbeEvidence{ + OpaqueLocation: opaqueLocation, + Nonce: nonce, + IssuedAt: issuedAt, + Region: region, + ServerRTT: rtt, + }, expectedNonce, nil +} diff --git a/server/store/queue_sql.go b/server/store/queue_sql.go index 83d82a44..2e97937d 100644 --- a/server/store/queue_sql.go +++ b/server/store/queue_sql.go @@ -194,6 +194,14 @@ func CreateQueueTicket(ctx context.Context, db *sql.DB, ticketID, playerID, idem } return fmt.Errorf("%w until %s", domain.ErrPlayerCooldown, cooldownEndsAt.UTC().Format(time.RFC3339)) } + // A nil map marshals to JSON `null`, a JSONB scalar -- not an empty + // object. jsonb_set then fails with "cannot set path in scalar", so + // the first probe for this player could never be recorded even once + // the probe endpoint was wired. Persist an object from the start. + if candidate.PredictedRTT == nil { + candidate.PredictedRTT = map[string]float64{} + ticket.Candidate.PredictedRTT = candidate.PredictedRTT + } predictedRTT, err := json.Marshal(candidate.PredictedRTT) if err != nil { return err @@ -226,7 +234,13 @@ func (q PostgresQueue) RecordProviderAllocation(ctx context.Context, allocation } const QueueProbeRecordSQL = `UPDATE queue_tickets -SET predicted_rtt = jsonb_set(COALESCE(predicted_rtt, '{}'::jsonb), ARRAY[$2], to_jsonb($3::double precision), true) +-- COALESCE only guards SQL NULL. Rows written before the insert fix hold a +-- JSONB scalar null, which jsonb_set rejects outright, so normalise anything +-- that is not an object before setting the region key. +SET predicted_rtt = jsonb_set( + CASE WHEN jsonb_typeof(COALESCE(predicted_rtt, '{}'::jsonb)) = 'object' + THEN predicted_rtt ELSE '{}'::jsonb END, + ARRAY[$2], to_jsonb($3::double precision), true) WHERE player_id = $1 AND state IN ('QUEUED', 'PROPOSED') AND expires_at > $4` func (q PostgresQueue) RecordProbe(ctx context.Context, playerID, region string, rtt time.Duration, now time.Time) error { @@ -373,3 +387,37 @@ func queueTicketRecordToDomain(record queueTicketRecord) domain.QueueTicket { candidate := domain.Candidate{TicketID: record.TicketID, PlayerID: record.PlayerID, Playlist: domain.Playlist(record.Playlist), ClientBuild: record.ClientBuild, ProtocolVersion: record.ProtocolVersion, EnqueuedAt: record.EnqueuedAt, PredictedRTT: record.PredictedRTT} return domain.QueueTicket{TicketID: record.TicketID, PlayerID: record.PlayerID, ProposalID: record.ProposalID, MatchID: record.MatchID, Candidate: candidate, Playlist: domain.Playlist(record.Playlist), State: domain.State(record.State), Revision: record.Revision, EnqueuedAt: record.EnqueuedAt, ExpiresAt: record.ExpiresAt} } + +// QueueCandidateByPlayerSQL mirrors QueueCandidateProjectionSQL for a single +// player, so a probe can repair that player's transient index entry without +// re-reading the whole queue. +const QueueCandidateByPlayerSQL = `SELECT q.ticket_id, q.player_id, q.playlist, q.client_build, + q.protocol_version, q.enqueued_at, q.predicted_rtt, COALESCE(r.rating, $3) +FROM queue_tickets q +LEFT JOIN ratings r ON r.player_id = q.player_id +WHERE q.player_id = $1 AND q.state = 'QUEUED' AND q.expires_at > $2` + +// FindQueuedCandidateByPlayer returns the player's live queue candidate, if +// any. The second result reports whether the player is currently queued; a +// player who is not queued is not an error. +func FindQueuedCandidateByPlayer(ctx context.Context, db *sql.DB, playerID string, now time.Time) (domain.Candidate, bool, error) { + if db == nil || playerID == "" || now.IsZero() { + return domain.Candidate{}, false, fmt.Errorf("invalid queued candidate lookup") + } + var candidate domain.Candidate + var playlist string + var predictedRTT []byte + err := db.QueryRowContext(ctx, QueueCandidateByPlayerSQL, playerID, now, domain.GlickoInitialRating). + Scan(&candidate.TicketID, &candidate.PlayerID, &playlist, &candidate.ClientBuild, &candidate.ProtocolVersion, &candidate.EnqueuedAt, &predictedRTT, &candidate.Rating) + if err == sql.ErrNoRows { + return domain.Candidate{}, false, nil + } + if err != nil { + return domain.Candidate{}, false, err + } + if err := json.Unmarshal(predictedRTT, &candidate.PredictedRTT); err != nil { + return domain.Candidate{}, false, fmt.Errorf("decode candidate RTT: %w", err) + } + candidate.Playlist = domain.Playlist(playlist) + return candidate, true, nil +} diff --git a/server/supervisor/supervisor_integration_test.go b/server/supervisor/supervisor_integration_test.go index e43c82f6..31868d91 100644 --- a/server/supervisor/supervisor_integration_test.go +++ b/server/supervisor/supervisor_integration_test.go @@ -47,7 +47,7 @@ func TestRealSupervisorRegistersAllocatedServerThroughControlPlane(t *testing.T) if err := db.PingContext(ctx); err != nil { t.Fatal(err) } - if _, err := db.ExecContext(ctx, `DROP TABLE IF EXISTS schema_migrations, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil { + if _, err := db.ExecContext(ctx, `DROP SCHEMA public CASCADE; CREATE SCHEMA public`); err != nil { t.Fatal(err) } if err := migrations.Apply(ctx, db, filepath.Join("..", "migrations")); err != nil {