diff --git a/Game/scripts/control_plane_client.gd b/Game/scripts/control_plane_client.gd index f812f68d..f33cb73c 100644 --- a/Game/scripts/control_plane_client.gd +++ b/Game/scripts/control_plane_client.gd @@ -14,6 +14,7 @@ signal assignment_connection_failed(detail: String) const DEFAULT_BASE_URL := "http://127.0.0.1:8080" const PERSIST_PATH := "user://matchmaking_state.cfg" +const AUTHORITATIVE_RECOVERY_INTERVAL_SECONDS := 5.0 var base_url := DEFAULT_BASE_URL var access_token := "" @@ -33,6 +34,8 @@ var _websocket: WebSocketPeer var _websocket_status := "DISCONNECTED" var _websocket_retry_seconds := 0.0 var _websocket_backoff := 1.0 +var _authoritative_recovery_seconds := AUTHORITATIVE_RECOVERY_INTERVAL_SECONDS +var _pending_proposal_id := "" var _pending_assignment_match_id := "" var _pending_resync_resource_id := "" @@ -74,10 +77,15 @@ func _process(_delta: float) -> void: _websocket_retry_seconds = _websocket_backoff _websocket_backoff = minf(_websocket_backoff * 2.0, 30.0) connect_event_stream() - if not _pending_assignment_match_id.is_empty() and _operation.is_empty() and not player_id.is_empty(): + if not _pending_proposal_id.is_empty() and _operation.is_empty() and not player_id.is_empty(): + var proposal_id := _pending_proposal_id + _pending_proposal_id = "" + recover_proposal(proposal_id) + elif not _pending_assignment_match_id.is_empty() and _operation.is_empty() and not player_id.is_empty(): var match_id := _pending_assignment_match_id _pending_assignment_match_id = "" fetch_assignment(match_id) + _poll_authoritative_recovery(_delta) func configure(url: String, token: String) -> bool: @@ -417,8 +425,10 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head _last_mutation_retryable = _last_mutation.get("operation", "") == operation if operation == "ranked_profile": ranked_profile.set_error("Ranked profile request failed") - elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover": + elif operation == "queue_create": state.fail("Control-plane request failed") + elif operation == "queue_recover" or operation == "proposal_recover": + state.set_notice("Could not refresh matchmaking state; retrying") else: state.set_notice("Control-plane request failed; retrying is safe") request_failed.emit(operation, response_code, "network error") @@ -428,8 +438,10 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head _last_mutation_retryable = _last_mutation.get("operation", "") == operation if operation == "ranked_profile": ranked_profile.set_error("Ranked profile returned invalid JSON") - elif operation == "queue_create" or operation == "queue_recover" or operation == "proposal_recover": + elif operation == "queue_create": state.fail("Control-plane returned invalid JSON") + elif operation == "queue_recover" or operation == "proposal_recover": + state.set_notice("Could not refresh matchmaking state; retrying") else: state.set_notice("Control-plane returned invalid JSON; retrying is safe") request_failed.emit(operation, response_code, "invalid JSON") @@ -494,6 +506,7 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head return if operation.begins_with("queue_"): if state.apply_ticket_update(normalize_ticket(payload), operation == "queue_recover"): + _queue_proposal_if_ready(payload) _queue_assignment_if_ready(payload) elif operation.begins_with("proposal_"): state.apply_proposal_update(normalize_proposal(payload)) @@ -538,7 +551,8 @@ func _handle_websocket_packet(packet: PackedByteArray) -> void: elif event_name == "proposal_changed": var proposal_update := event.duplicate(true) proposal_update["proposal_id"] = String(event["resource_id"]) - state.apply_proposal_update(proposal_update) + if state.prepare_proposal_recovery(String(proposal_update["proposal_id"])): + state.apply_proposal_update(proposal_update) elif event_name == "assignment_changed": state.mark_assignment_ready() _pending_assignment_match_id = String(event["match_id"]) @@ -599,6 +613,13 @@ static func _valid_queue_response(payload: Dictionary) -> bool: return false if String(payload["state"]) not in ["ALLOCATING", "PROCESS_READY", "ASSIGNMENT_READY", "ASSIGNED", "CONNECTING", "LIVE", "RESULT_PENDING", "COMPLETED", "FAILED", "CANCELLED"]: return false + if payload.has("proposal_id"): + if not payload["proposal_id"] is String or not is_valid_resource_id(String(payload["proposal_id"])): + return false + if String(payload["state"]) != "PROPOSED": + return false + if payload.has("match_id") and payload.has("proposal_id"): + return false if not _valid_revision(payload["revision"]): return false return payload["enqueued_at"] is String and is_valid_rfc3339_timestamp(String(payload["enqueued_at"])) and payload["expires_at"] is String and is_valid_rfc3339_timestamp(String(payload["expires_at"])) @@ -612,6 +633,26 @@ func _queue_assignment_if_ready(payload: Dictionary) -> void: _pending_assignment_match_id = match_id +func _queue_proposal_if_ready(payload: Dictionary) -> void: + if String(payload.get("state", "")) != "PROPOSED": + return + var proposal_id := String(payload.get("proposal_id", "")) + if is_valid_resource_id(proposal_id) and state.prepare_proposal_recovery(proposal_id): + _pending_proposal_id = proposal_id + + +func _poll_authoritative_recovery(delta: float) -> void: + if auth_expired or not is_valid_access_token(access_token) or state.ticket_id.is_empty() or state.phase in [MatchmakingState.CANCELLED, MatchmakingState.EXPIRED, MatchmakingState.FAILED, MatchmakingState.COMPLETED]: + _authoritative_recovery_seconds = AUTHORITATIVE_RECOVERY_INTERVAL_SECONDS + return + _authoritative_recovery_seconds -= maxf(0.0, delta) + if _authoritative_recovery_seconds > 0.0 or not _operation.is_empty(): + return + _authoritative_recovery_seconds = AUTHORITATIVE_RECOVERY_INTERVAL_SECONDS + var resource_id := state.proposal_id if state.has_open_proposal() else state.ticket_id + _run_resync(resource_id) + + static func _valid_proposal_response(payload: Dictionary) -> bool: if not _valid_response_opaque_id(payload, "proposal_id") or not payload.has("expires_at") or not payload["expires_at"] is String or not is_valid_rfc3339_timestamp(String(payload["expires_at"])) or not payload.has("participants") or not payload["participants"] is Array: return false diff --git a/Game/scripts/matchmaking_state.gd b/Game/scripts/matchmaking_state.gd index 2868647b..694f800f 100644 --- a/Game/scripts/matchmaking_state.gd +++ b/Game/scripts/matchmaking_state.gd @@ -161,6 +161,19 @@ func apply_proposal_update(update: Dictionary) -> bool: return true +func prepare_proposal_recovery(new_proposal_id: String) -> bool: + if not _valid_opaque_id(new_proposal_id): + return false + if proposal_id == new_proposal_id: + return true + if proposal_state not in ["", "DECLINED", "EXPIRED", "CANCELLED"]: + return false + proposal_id = new_proposal_id + proposal_revision = 0 + proposal_state = "" + return true + + func mark_assignment_ready() -> void: phase = ASSIGNMENT_READY message = "Match server is ready" diff --git a/Game/tests/cases/test_control_plane_client.gd b/Game/tests/cases/test_control_plane_client.gd index 9513429b..c4167d8e 100644 --- a/Game/tests/cases/test_control_plane_client.gd +++ b/Game/tests/cases/test_control_plane_client.gd @@ -153,6 +153,20 @@ func test_websocket_reconnect_defers_recovery_while_http_mutation_is_in_flight() client.free() +func test_transient_rest_recovery_failure_does_not_end_matchmaking() -> void: + var client := ControlPlaneClient.new() + client._ready() + client.state.begin_queue("ticket_recovery_123", "casual") + client._operation = "queue_recover" + client._on_request_completed(HTTPRequest.RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray()) + assert_eq(client.state.phase, MatchmakingState.QUEUED, "network failure during recovery keeps the active search") + assert_true(client.state.message.contains("retrying"), "recovery failure remains visible and retryable") + client._operation = "proposal_recover" + client._on_request_completed(HTTPRequest.RESULT_SUCCESS, 200, PackedStringArray(), "[]".to_utf8_buffer()) + assert_eq(client.state.phase, MatchmakingState.QUEUED, "malformed transient recovery response does not become terminal") + client.free() + + func test_resync_of_terminal_proposal_recovers_the_ticket() -> void: assert_eq(ControlPlaneClient.resync_target("proposal-terminal-resync", "ticket-terminal-resync", "proposal-terminal-resync", false), "ticket-terminal-resync", "terminal proposal resync targets the requeued ticket") assert_eq(ControlPlaneClient.resync_target("proposal-terminal-resync", "ticket-terminal-resync", "proposal-terminal-resync", true), "proposal-terminal-resync", "open proposal resync retains the proposal target") @@ -228,6 +242,17 @@ func test_queue_response_requires_the_complete_contract_shape() -> void: assert_true(not ControlPlaneClient._valid_queue_response(premature_match), "pre-match ticket cannot smuggle a match identity") assigned["match_id"] = "match/unsafe" assert_true(not ControlPlaneClient._valid_queue_response(assigned), "unsafe recovered match identity is rejected") + var proposed := valid.duplicate() + proposed["state"] = "PROPOSED" + proposed["proposal_id"] = "proposal_12345678" + assert_true(ControlPlaneClient._valid_queue_response(proposed), "recovered proposed ticket carries its proposal lookup identity") + var client := ControlPlaneClient.new() + client._ready() + client.state.begin_queue("ticket_1234567890", "casual") + client._queue_proposal_if_ready(proposed) + assert_eq(client._pending_proposal_id, "proposal_12345678", "recovered proposal is queued for authoritative fetch") + assert_eq(client.state.proposal_id, "proposal_12345678", "recovered proposal identity becomes the active projection") + client.free() func test_proposal_response_requires_structured_unique_participants() -> void: diff --git a/Game/tests/cases/test_matchmaking_state.gd b/Game/tests/cases/test_matchmaking_state.gd index 9c755561..e017923e 100644 --- a/Game/tests/cases/test_matchmaking_state.gd +++ b/Game/tests/cases/test_matchmaking_state.gd @@ -156,6 +156,10 @@ func test_terminal_proposal_is_not_an_active_recovery_target() -> void: assert_true(state.has_open_proposal(), "open proposal is an active recovery target") assert_true(state.apply_proposal_update({"proposal_id": "proposal-terminal", "revision": 2, "state": "EXPIRED"}), "proposal expires") assert_true(not state.has_open_proposal(), "terminal proposal uses ticket recovery instead") + assert_true(state.prepare_proposal_recovery("proposal_second_123"), "a later proposal can replace a terminal proposal identity") + assert_true(state.apply_proposal_update({"proposal_id": "proposal_second_123", "revision": 4, "state": "OPEN"}), "recovered later proposal accepts its authoritative revision") + assert_true(state.has_open_proposal(), "later proposal becomes the active recovery target") + assert_true(not state.prepare_proposal_recovery("proposal_third_1234"), "an open proposal cannot be replaced by another identity") func test_assignment_lifecycle_has_explicit_connecting_and_live_states() -> void: diff --git a/multiplayer-next.md b/multiplayer-next.md index e2f04a26..af847673 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1632,3 +1632,5 @@ The audio TODO now has a runtime foundation: `AudioManager` generates bounded pl The video-settings TODO is likewise locally implemented: presets, vsync, refresh-derived FPS caps, and resolution scaling are wired through `VideoSettings` and the settings menu. The remaining acceptance work is low/mid-tier hardware frame-time and image-quality profiling, which cannot be certified from this workspace. Assignment handoff now has a non-circular recovery path. Match-scoped lifecycle events are no longer misapplied as queue-ticket resources: they trigger owner-scoped ticket recovery, and recovered active tickets include their durable `match_id`. An `ASSIGNMENT_READY` event or recovered ticket can therefore drive `GET /assignments/{matchId}` without already having fetched that assignment. Owner-scoped REST ticket snapshots may cross missed revisions only along a reachable forward lifecycle path, while incremental WebSocket updates remain strictly contiguous and neither path can rewind state. The OpenAPI queue projection includes the optional active match identity, Go tests cover the store/API projection, and the 199-test Godot harness covers match-resource separation, malformed identities, missed-revision recovery, illegal rewinds, and assignment-fetch scheduling. The real PostgreSQL assertion is committed with the store integration suite; rerunning it in this workspace is temporarily blocked by Docker storage exhaustion (`initdb` cannot create `pg_wal`), so live SQL evidence remains open rather than being claimed from the static/unit gates. + +Replica-independent client convergence now supersedes the earlier "at-least-once WebSocket delivery" wording in tasks 8.25/8.40 and the allocation-outbox progress notes. The database outbox guarantees ordered, replayable invocation of a replica's transient publication adapter, not receipt by a socket that may be absent or attached to another replica. Active clients now perform bounded five-second owner-scoped REST recovery; ticket recovery exposes the active `proposal_id` or `match_id`, so a missed proposal, allocation, assignment, or result notification cannot strand the client without the next resource key. A terminal proposal can be replaced by a later recovered proposal identity, while an open proposal cannot be overwritten. Network and malformed-JSON failures during recovery remain visible and retryable instead of falsely terminating matchmaking. WebSocket events remain the low-latency path; REST snapshots are the correctness path. Store/API and the 200-test Godot harness cover projection, transient failure, replacement, and hostile identity/state combinations, with live PostgreSQL execution still subject to the Docker storage gate recorded above. diff --git a/server/api/outbox.go b/server/api/outbox.go index 9f231238..012dabe6 100644 --- a/server/api/outbox.go +++ b/server/api/outbox.go @@ -13,8 +13,10 @@ import ( // RunProposalOutboxDispatcher delivers committed proposal changes to the // authenticated WebSocket subscribers. It only reads proposal_changed rows; // result and other outbox event types remain owned by their own consumers. -// Delivery is at-least-once because the row is acknowledged only after every -// participant publication succeeds. +// The outbox guarantees after-commit publication into this replica's bounded +// transient hub; WebSocket receipt is deliberately best-effort. Clients use +// owner-scoped periodic REST recovery for correctness across disconnects and +// replicas, so a socket notification is only a latency optimization. func RunProposalOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service) { if db == nil || service == nil { return diff --git a/server/api/service.go b/server/api/service.go index 500c43e0..3b4d42ba 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -349,6 +349,7 @@ type queueCreateRequest struct { type queueResponse struct { TicketID string `json:"ticket_id"` PlayerID string `json:"player_id"` + ProposalID string `json:"proposal_id,omitempty"` MatchID string `json:"match_id,omitempty"` State string `json:"state"` Revision uint64 `json:"revision"` @@ -1117,7 +1118,7 @@ func decodeBody(w http.ResponseWriter, r *http.Request, target any) bool { } func toQueueResponse(ticket domain.QueueTicket) queueResponse { - return queueResponse{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), Revision: ticket.Revision, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt} + return queueResponse{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, ProposalID: ticket.ProposalID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), Revision: ticket.Revision, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt} } func toProposalResponse(proposal domain.Proposal) proposalResponse { diff --git a/server/api/service_test.go b/server/api/service_test.go index 885f0e21..756e299b 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -23,7 +23,10 @@ import ( type queueBackendSpy struct{ createCalls, heartbeatCalls, cancelCalls, getCalls int } func TestQueueResponseCarriesRecoveredMatchIdentity(t *testing.T) { - response := toQueueResponse(domain.QueueTicket{TicketID: "ticket-1234567890", PlayerID: "player-1234567890", MatchID: "match-1234567890", State: domain.AssignmentReady}) + response := toQueueResponse(domain.QueueTicket{TicketID: "ticket-1234567890", PlayerID: "player-1234567890", ProposalID: "proposal-1234567890", MatchID: "match-1234567890", State: domain.AssignmentReady}) + if response.ProposalID != "proposal-1234567890" { + t.Fatalf("queue response proposal ID = %q", response.ProposalID) + } if response.MatchID != "match-1234567890" { t.Fatalf("queue response match ID = %q", response.MatchID) } diff --git a/server/contracts/v1/openapi.json b/server/contracts/v1/openapi.json index fc2350a9..4b5bd971 100644 --- a/server/contracts/v1/openapi.json +++ b/server/contracts/v1/openapi.json @@ -86,7 +86,7 @@ "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"}, "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"}}}, + "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}}}, diff --git a/server/domain/queue.go b/server/domain/queue.go index 3c9a5107..24d8c575 100644 --- a/server/domain/queue.go +++ b/server/domain/queue.go @@ -26,6 +26,7 @@ var ( type QueueTicket struct { TicketID string PlayerID string + ProposalID string MatchID string Candidate Candidate Playlist Playlist diff --git a/server/store/outbox.go b/server/store/outbox.go index eff11c1a..96c16f14 100644 --- a/server/store/outbox.go +++ b/server/store/outbox.go @@ -8,8 +8,9 @@ import ( ) // OutboxEvent is the durable hand-off between a committed domain mutation and -// transient WebSocket delivery. Consumers must make delivery idempotent by -// event ID and only acknowledge after successful fan-out. +// transient WebSocket publication. Consumers must make publication idempotent +// by event ID and only acknowledge after the local adapter accepts the event. +// Subscriber receipt is not durable; clients converge through REST recovery. type OutboxEvent struct { EventID string AggregateType string @@ -64,8 +65,9 @@ type OutboxDelivery func(context.Context, OutboxEvent) error // OutboxDispatcher is the durable-to-transient bridge. Read and Ack are // injectable so ordering can be tested without a live PostgreSQL instance. -// Delivery is at-least-once: a crash after delivery and before acknowledgement -// leaves the event replayable, while a delivery failure stops the batch. +// Adapter invocation is at-least-once: a crash after invocation and before +// acknowledgement leaves the event replayable, while an adapter failure stops +// the batch. This does not imply that a transient subscriber received it. type OutboxDispatcher struct { Read func(context.Context, int) ([]OutboxEvent, error) Ack func(context.Context, string, time.Time) error diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 9ab7feee..c1f5cbdf 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -520,6 +520,10 @@ func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) { if err := db.QueryRow(`SELECT count(*) FROM queue_tickets WHERE state = 'PROPOSED'`).Scan(&proposed); err != nil || proposed != 2 { t.Fatalf("proposed queue tickets = %d, err = %v", proposed, err) } + recoveredTicket, err := GetQueueTicket(ctx, db, "proposal-player-a", "proposal-ticket-0", now) + if err != nil || recoveredTicket.ProposalID != proposal.ProposalID { + t.Fatalf("recovered ticket proposal=%q err=%v", recoveredTicket.ProposalID, err) + } recovered, err := GetProposal(ctx, db, "proposal-player-a", proposal.ProposalID, now) if err != nil { t.Fatalf("recover proposal: %v", err) diff --git a/server/store/queue_sql.go b/server/store/queue_sql.go index e28252e0..57e61101 100644 --- a/server/store/queue_sql.go +++ b/server/store/queue_sql.go @@ -23,6 +23,11 @@ WHERE scope = $1 AND idempotency_key = $2 FOR UPDATE` QueueTicketSelectSQL = `SELECT q.ticket_id, q.player_id, q.playlist, q.state, q.client_build, q.protocol_version, q.enqueued_at, q.expires_at, q.revision, q.predicted_rtt, + COALESCE((SELECT pp.proposal_id FROM proposal_participants pp + JOIN proposals p ON p.proposal_id = pp.proposal_id + WHERE pp.ticket_id = q.ticket_id AND pp.player_id = q.player_id + AND p.state = 'OPEN' + LIMIT 1), ''), COALESCE((SELECT mp.match_id FROM match_participants mp WHERE mp.ticket_id = q.ticket_id AND mp.player_id = q.player_id AND mp.participation_active @@ -177,6 +182,7 @@ func CreateQueueTicket(ctx context.Context, db *sql.DB, ticketID, playerID, idem type queueTicketRecord struct { TicketID string `json:"ticket_id"` PlayerID string `json:"player_id"` + ProposalID string `json:"proposal_id,omitempty"` MatchID string `json:"match_id,omitempty"` Playlist string `json:"playlist"` State string `json:"state"` @@ -235,7 +241,7 @@ func GetQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID string, } var record queueTicketRecord var predictedRTT []byte - if err := db.QueryRowContext(ctx, QueueTicketSelectSQL, ticketID, playerID).Scan(&record.TicketID, &record.PlayerID, &record.Playlist, &record.State, &record.ClientBuild, &record.ProtocolVersion, &record.EnqueuedAt, &record.ExpiresAt, &record.Revision, &predictedRTT, &record.MatchID); err != nil { + if err := db.QueryRowContext(ctx, QueueTicketSelectSQL, ticketID, playerID).Scan(&record.TicketID, &record.PlayerID, &record.Playlist, &record.State, &record.ClientBuild, &record.ProtocolVersion, &record.EnqueuedAt, &record.ExpiresAt, &record.Revision, &predictedRTT, &record.ProposalID, &record.MatchID); err != nil { return domain.QueueTicket{}, err } if err := json.Unmarshal(predictedRTT, &record.PredictedRTT); err != nil { @@ -310,9 +316,9 @@ func mutateQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idem } func queueTicketRecordFromDomain(ticket domain.QueueTicket) queueTicketRecord { - return queueTicketRecord{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), ClientBuild: ticket.Candidate.ClientBuild, ProtocolVersion: ticket.Candidate.ProtocolVersion, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt, Revision: ticket.Revision, PredictedRTT: ticket.Candidate.PredictedRTT} + return queueTicketRecord{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, ProposalID: ticket.ProposalID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), ClientBuild: ticket.Candidate.ClientBuild, ProtocolVersion: ticket.Candidate.ProtocolVersion, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt, Revision: ticket.Revision, PredictedRTT: ticket.Candidate.PredictedRTT} } 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, MatchID: record.MatchID, Candidate: candidate, Playlist: domain.Playlist(record.Playlist), State: domain.State(record.State), Revision: record.Revision, EnqueuedAt: record.EnqueuedAt, ExpiresAt: record.ExpiresAt} + 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} } diff --git a/server/store/queue_sql_test.go b/server/store/queue_sql_test.go index ece663e8..28bcdcdc 100644 --- a/server/store/queue_sql_test.go +++ b/server/store/queue_sql_test.go @@ -10,7 +10,7 @@ func TestQueueSQLUsesDurableIdempotencyAndOwnerScopedRecovery(t *testing.T) { for query, fragments := range map[string][]string{ QueueIdempotencyInsertSQL: {"idempotency_keys", "ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"}, QueueIdempotencySelectSQL: {"scope = $1", "idempotency_key = $2", "FOR UPDATE"}, - QueueTicketSelectSQL: {"q.ticket_id = $1", "q.player_id = $2", "match_participants", "participation_active"}, + QueueTicketSelectSQL: {"q.ticket_id = $1", "q.player_id = $2", "proposal_participants", "p.state = 'OPEN'", "match_participants", "participation_active"}, QueueTicketInsertSQL: {"player_id", "playlist", "client_build", "protocol_version"}, QueueTicketHeartbeatSQL: {"player_id = $2", "revision = $3", "expires_at > $4", "RETURNING"}, QueueTicketCancelSQL: {"player_id = $2", "revision = $3", "state NOT IN", "RETURNING"}, @@ -28,13 +28,19 @@ func TestQueueSQLUsesDurableIdempotencyAndOwnerScopedRecovery(t *testing.T) { } func TestQueueTicketRecordPreservesRecoveredMatchIdentity(t *testing.T) { - ticket := queueTicketRecordToDomain(queueTicketRecord{TicketID: "ticket-1", PlayerID: "player-1", MatchID: "match-1", Playlist: string(domain.Casual), State: string(domain.AssignmentReady)}) + ticket := queueTicketRecordToDomain(queueTicketRecord{TicketID: "ticket-1", PlayerID: "player-1", ProposalID: "proposal-1", MatchID: "match-1", Playlist: string(domain.Casual), State: string(domain.AssignmentReady)}) + if ticket.ProposalID != "proposal-1" { + t.Fatalf("recovered proposal ID = %q", ticket.ProposalID) + } if ticket.MatchID != "match-1" { t.Fatalf("recovered match ID = %q", ticket.MatchID) } if got := queueTicketRecordFromDomain(ticket).MatchID; got != "match-1" { t.Fatalf("stored match ID = %q", got) } + if got := queueTicketRecordFromDomain(ticket).ProposalID; got != "proposal-1" { + t.Fatalf("stored proposal ID = %q", got) + } } func TestLoadRankedParticipantsRejectsNonSixPlayerLookupsWithoutDatabase(t *testing.T) {