diff --git a/Game/scripts/control_plane_client.gd b/Game/scripts/control_plane_client.gd index 438b3ac4..fd9f05f8 100644 --- a/Game/scripts/control_plane_client.gd +++ b/Game/scripts/control_plane_client.gd @@ -32,6 +32,7 @@ var _websocket_status := "DISCONNECTED" var _websocket_retry_seconds := 0.0 var _websocket_backoff := 1.0 var _pending_assignment_match_id := "" +var _pending_resync_resource_id := "" func _ready() -> void: @@ -356,6 +357,8 @@ func _on_request_completed(result: HTTPRequest.Result, response_code: int, _head request_failed.emit(operation, response_code, assignment.error_message) return request_succeeded.emit(operation, payload) + if not _pending_resync_resource_id.is_empty(): + call_deferred("_run_pending_resync") func _handle_websocket_packet(packet: PackedByteArray) -> void: @@ -405,6 +408,21 @@ static func _valid_websocket_event(event: Dictionary) -> bool: func _on_resync_required(resource_id: String) -> void: + if not _operation.is_empty(): + _pending_resync_resource_id = resource_id + return + _run_resync(resource_id) + + +func _run_pending_resync() -> void: + if not _operation.is_empty() or _pending_resync_resource_id.is_empty(): + return + var resource_id := _pending_resync_resource_id + _pending_resync_resource_id = "" + _run_resync(resource_id) + + +func _run_resync(resource_id: String) -> void: if resource_id == state.ticket_id and not state.ticket_id.is_empty(): recover_queue(state.ticket_id) elif resource_id == state.proposal_id and not state.proposal_id.is_empty(): diff --git a/Game/tests/control_plane_proposal_smoke.gd b/Game/tests/control_plane_proposal_smoke.gd new file mode 100644 index 00000000..ea58de4a --- /dev/null +++ b/Game/tests/control_plane_proposal_smoke.gd @@ -0,0 +1,141 @@ +extends Node + +# Two-process real end-to-end proposal smoke test: extends +# control_plane_smoke.gd's single-player login/queue/heartbeat/cancel +# coverage to the matcher path -- two real Godot clients, two real queued +# tickets, a real running server/cmd/matcher pairing them, both clients +# observing the resulting proposal over the real WebSocket event stream and +# accepting it for real. multiplayer-next.md 8.40 names this "a two-player +# proposal round trip" as the next scoped extension to this harness. +# +# ControlPlaneClient is a singleton autoload, so one process can only ever be +# one player -- this mirrors net_smoke.gd's own host/client two-process +# pattern rather than trying to simulate two players in one process: +# +# godot --headless --path Game res://tests/control_plane_proposal_smoke.tscn -- \ +# --control-plane-url=http://127.0.0.1:PORT --role=player-a +# godot --headless --path Game res://tests/control_plane_proposal_smoke.tscn -- \ +# --control-plane-url=http://127.0.0.1:PORT --role=player-b +# +# Prints one "SMOKE PASS/FAIL: ..." line and exits 0/1. + +const TIMEOUT_SECONDS := 20.0 + +var _role := "" +var _finished := false +var _ticket_id := "" +var _accept_sent := false + + +func _ready() -> void: + var control_plane_url := "" + for arg in OS.get_cmdline_user_args(): + if arg.begins_with("--control-plane-url="): + control_plane_url = arg.substr("--control-plane-url=".length()) + elif arg.begins_with("--role="): + _role = arg.substr("--role=".length()) + if control_plane_url.is_empty() or (_role != "player-a" and _role != "player-b"): + _finish(false, "missing --control-plane-url or --role=player-a|player-b") + return + if not ControlPlaneClient.configure(control_plane_url, "0:0"): + _finish(false, "configure() rejected a valid-looking base URL") + return + + _ticket_id = "proposal-smoke-%s-%d" % [_role, Time.get_unix_time_from_system()] + ControlPlaneClient.request_succeeded.connect(_on_request_succeeded) + ControlPlaneClient.request_failed.connect(_on_request_failed) + ControlPlaneClient.state.changed.connect(_on_state_changed) + + var web_api_ticket := "proposal-smoke-web-api-ticket-%s-%d" % [_role, Time.get_ticks_usec()] + var err := ControlPlaneClient.login_steam(web_api_ticket) + if err != OK: + _finish(false, "login_steam() failed to start: %s" % error_string(err)) + return + print("SMOKE[%s]: logging in against %s..." % [_role, control_plane_url]) + + var timer := Timer.new() + timer.wait_time = TIMEOUT_SECONDS + timer.one_shot = true + timer.timeout.connect(func(): _finish(false, "timed out after %.1fs waiting for a proposal" % TIMEOUT_SECONDS)) + add_child(timer) + timer.start() + + +func _on_request_succeeded(operation: String, payload: Dictionary) -> void: + if _finished: + return + if operation == "steam_session": + print("SMOKE[%s]: logged in as %s, queueing for a 2-player casual match..." % [_role, ControlPlaneClient.player_id]) + var err := ControlPlaneClient.queue_create(_ticket_id, "casual", "smoke-build", 1) + if err != OK: + _finish(false, "queue_create() failed to start: %s" % error_string(err)) + return + if operation == "queue_create": + print("SMOKE[%s]: ticket %s QUEUED, waiting for the matcher to propose a match..." % [_role, _ticket_id]) + return + if operation.begins_with("proposal_"): + if payload.get("state", "") != "ACCEPTED" and payload.get("state", "") != "OPEN": + _finish(false, "unexpected proposal response after accept: %s" % payload) + return + if payload.get("state", "") == "ACCEPTED": + _finish(true, "both players queued, the real matcher formed a proposal, and this client's accept was recorded") + else: + print("SMOKE[%s]: accepted; waiting for the other player..." % _role) + var recovery_timer := get_tree().create_timer(1.0) + recovery_timer.timeout.connect(_recover_after_accept) + + +func _recover_after_accept() -> void: + if _finished or ControlPlaneClient._operation != "" or ControlPlaneClient.state.proposal_id.is_empty(): + return + var err := ControlPlaneClient.recover_proposal(ControlPlaneClient.state.proposal_id) + if err != OK and err != ERR_BUSY: + _finish(false, "proposal recovery after accept failed to start: %s" % error_string(err)) + + +func _on_state_changed(snapshot: Dictionary) -> void: + if _finished or _accept_sent or String(snapshot.get("proposal_state", "")) != "OPEN": + return + if ControlPlaneClient._operation != "": + return # Already mid-request (e.g. the accept itself); avoid double-sending. + print("SMOKE[%s]: proposal %s is OPEN at revision %d, accepting..." % [_role, ControlPlaneClient.state.proposal_id, ControlPlaneClient.state.proposal_revision]) + _accept_sent = true + if _role == "player-b": + var delay := get_tree().create_timer(0.5) + delay.timeout.connect(_send_accept) + return + _send_accept() + + +func _send_accept() -> void: + if _finished: + return + var err := ControlPlaneClient.respond_to_proposal(ControlPlaneClient.state.proposal_id, true, ControlPlaneClient.state.proposal_revision) + if err != OK and err != ERR_BUSY: + _accept_sent = false + _finish(false, "respond_to_proposal() failed to start: %s" % error_string(err)) + + +func _on_request_failed(operation: String, http_code: int, detail: String) -> void: + if _finished: + return + if operation == "proposal_accept" and http_code == 409 and not ControlPlaneClient.state.proposal_id.is_empty(): + print("SMOKE[%s]: concurrent accept conflicted at an old revision; recovering the authoritative proposal..." % _role) + _accept_sent = false + var err := ControlPlaneClient.recover_proposal(ControlPlaneClient.state.proposal_id) + if err != OK and err != ERR_BUSY: + _finish(false, "proposal recovery after concurrent accept failed to start: %s" % error_string(err)) + return + _finish(false, "%s failed: http=%d detail=%s" % [operation, http_code, detail]) + + +func _finish(passed: bool, detail: String) -> void: + if _finished: + return + _finished = true + if passed: + print("SMOKE PASS: [%s] %s" % [_role, detail]) + get_tree().quit(0) + else: + print("SMOKE FAIL: [%s] %s" % [_role, detail]) + get_tree().quit(1) diff --git a/Game/tests/control_plane_proposal_smoke.tscn b/Game/tests/control_plane_proposal_smoke.tscn new file mode 100644 index 00000000..3f846fae --- /dev/null +++ b/Game/tests/control_plane_proposal_smoke.tscn @@ -0,0 +1,6 @@ +[gd_scene load_steps=2 format=3] + +[ext_resource type="Script" path="res://tests/control_plane_proposal_smoke.gd" id="1_cpps"] + +[node name="ControlPlaneProposalSmoke" type="Node"] +script = ExtResource("1_cpps") diff --git a/multiplayer-next.md b/multiplayer-next.md index 7910d864..c20ff55a 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1234,7 +1234,7 @@ the local/CI/community transport, not a silent production fallback. | # | Task | Acceptance | |---|---|---| | 8.39 `[D:8.3,8.14,8.17]` | **IN PROGRESS.** `MatchmakingState` now projects queue → proposal → allocation/process-ready/assignment-ready/connect/live plus terminal failure states; autoload `ControlPlaneClient` provides authenticated queue create/recovery/heartbeat/cancel and proposal response requests with idempotency/revision headers; `matchmaking.tscn`/`matchmaking.gd` expose the state and authoritative actions from the main menu; authenticated proposal recovery now reconciles missed proposal events and expires them at read time; the Go API publishes targeted authenticated revisioned queue/proposal/assignment events and the Godot client consumes state/proposal/assignment events, fetching the authoritative assignment after assignment readiness; the API can now use the durable participant-scoped proposal provider for both recovery and accept/decline mutations | `test_matchmaking_state.gd`, `test_control_plane_client.gd`, `test_matchmaking_ui.gd`, `TestProposalRecoveryIsParticipantScopedAndExpiresAtReadBoundary`, `TestAuthenticatedWebSocketDeliversOnlyTargetedRevisionedEvents`, `TestStateChangingAPIActionsPublishTargetedEvents`, `TestProposalRecoveryUsesDurableBackendAndRemainsParticipantScoped` and store proposal SQL tests reject stale/gapped/conflicting updates, validate endpoint/token/payload normalization, preserve idempotent duplicates, and guarantee visible phase/terminal copy; wiring durable allocation events into the outbox, wait/latency explanations and Godot runtime verification remain | -| 8.40 `[D:8.3,8.14]` | **IN PROGRESS.** Pure Go revisioned replica reducer rejects gaps for REST resync, makes duplicate/out-of-order events idempotent, and resumes from the authoritative snapshot revision; authenticated queue-ticket recovery now has an owner-checked REST read; Godot client projection persists non-secret ticket/proposal state, forces authoritative recovery after restart, and can replay a lost queue-create response with the original ticket/idempotency key; the Go API now exposes an authenticated `/v1/events` WebSocket with bounded per-player queues, strict upgrade/vocabulary validation and REST-resync-safe slow-client failure; Godot `ControlPlaneClient` can connect to the stream using Godot 4.7's handshake-header API, consume validated events, automatically reconnect with bounded backoff, and trigger REST recovery on projection gaps and stream return; `OutboxDispatcher` now provides ordered at-least-once delivery after durable commit; assignment recovery events carry the durable assignment revision without leaking it into REST JSON; prediction startup now distinguishes the sequence-0 warm-up acknowledgement from genuine missing/evicted history, preventing a false hard-snap during a live match | `server/domain/sync.go`, `server/api/events.go`, `server/api/service.go`, `server/store/outbox.go`, `service_test.go`, `outbox_test.go`, `matchmaking_state.gd`, `control_plane_client.gd`, `local_prediction_history.gd`, `net_ship_predictor.gd` and tests cover gap, snapshot, replay, same-revision conflict, owner-only ticket recovery, expired-ticket terminal handling, malformed restart snapshots, API-level duplicate-create replay/conflict, authenticated handshake/key rejection, targeted event delivery, exactly-once slow-subscriber closure, invalid-event rejection, delivery-before-ack failure ordering, assignment event/response revision separation and prediction warm-up/hard-resync separation; Godot 4.7.1 headless project parse and 143-test unit harness pass with compatibility rendering, and a two-process ENet authoritative match smoke passes spawn, movement, prediction, 0% snapshot loss and 0 hard snaps; `cmd/control-plane` provides a signal-bound API role, `cmd/matcher` now supports casual and explicitly enabled ranked roles with durable identity lookup, and `cmd/maintenance` provides bounded season maintenance. **Live multi-process control-plane/game verification now exists**: `scripts/verify_control_plane_integration.sh` runs a real `postgres:17-alpine`, the real `api.Service` (via the new test-only `server/cmd/testkit-api`, wired identically to `cmd/control-plane` except for a fake Steam login — see §8.7), and a real headless Godot client (`control_plane_smoke.gd`) round-tripping login → `fetch_ranked_profile` (expect 404, §8.22) → `queue_create` → heartbeat → `cancel_queue` over an actual network connection — the first time this boundary was exercised end to end rather than against a mock on either side. It immediately found a real bug: `matchmaking_state.gd`'s `apply_ticket_update` treated every same-revision confirmation right after `begin_queue()` as a conflict (comparing `expires_at_unix`, a field the client can't know in advance), so a real client would loop on `recover_queue` forever instead of ever settling into `QUEUED` — fixed and re-verified stable across 3 consecutive full runs, three times now (once per coverage addition). A two-player proposal round trip (needs a running matcher, not yet wired into `testkit-api`), allocator, and Redis fan-out live verification remain | +| 8.40 `[D:8.3,8.14]` | **IN PROGRESS.** Pure Go revisioned replica reducer rejects gaps for REST resync, makes duplicate/out-of-order events idempotent, and resumes from the authoritative snapshot revision; authenticated queue-ticket recovery now has an owner-checked REST read; Godot client projection persists non-secret ticket/proposal state, forces authoritative recovery after restart, and can replay a lost queue-create response with the original ticket/idempotency key; the Go API now exposes an authenticated `/v1/events` WebSocket with bounded per-player queues, strict upgrade/vocabulary validation and REST-resync-safe slow-client failure; Godot `ControlPlaneClient` can connect to the stream using Godot 4.7's handshake-header API, consume validated events, automatically reconnect with bounded backoff, and trigger REST recovery on projection gaps and stream return, including a deferred recovery when an event arrives during an in-flight HTTP mutation; `OutboxDispatcher` now provides ordered at-least-once delivery after durable commit; assignment recovery events carry the durable assignment revision without leaking it into REST JSON; prediction startup now distinguishes the sequence-0 warm-up acknowledgement from genuine missing/evicted history, preventing a false hard-snap during a live match; proposal creation now writes one durable targeted `proposal_changed` outbox event per proposal aggregate revision, and the test-only API harness dispatches it to authenticated participants | `server/domain/sync.go`, `server/api/events.go`, `server/api/service.go`, `server/store/outbox.go`, `server/store/proposal_sql.go`, `server/cmd/testkit-api`, `service_test.go`, `outbox_test.go`, `matchmaking_state.gd`, `control_plane_client.gd`, `local_prediction_history.gd`, `net_ship_predictor.gd` and tests cover gap, snapshot, replay, same-revision conflict, owner-only ticket recovery, expired-ticket terminal handling, malformed restart snapshots, API-level duplicate-create replay/conflict, authenticated handshake/key rejection, targeted event delivery, exactly-once slow-subscriber closure, invalid-event rejection, delivery-before-ack failure ordering, assignment event/response revision separation, prediction warm-up/hard-resync separation and deferred proposal recovery; Godot 4.7.1 headless project parse and 143-test unit harness pass with compatibility rendering, and a two-process ENet authoritative match smoke passes spawn, movement, prediction, 0% snapshot loss and 0 hard snaps; `cmd/control-plane` provides a signal-bound API role, `cmd/matcher` now supports casual and explicitly enabled ranked roles with durable identity lookup, and `cmd/maintenance` provides bounded season maintenance. **Live multi-process control-plane/game verification now exists**: `scripts/verify_control_plane_integration.sh` runs a real `postgres:17-alpine`, the real `api.Service` (via the new test-only `server/cmd/testkit-api`, wired identically to `cmd/control-plane` except for a fake Steam login — see §8.7), and a real headless Godot client (`control_plane_smoke.gd`) round-tripping login → `fetch_ranked_profile` (expect 404, §8.22) → `queue_create` → heartbeat → `cancel_queue` over an actual network connection — the first time this boundary was exercised end to end rather than against a mock on either side. **The two-player proposal extension now also passes**: `scripts/verify_control_plane_proposal_integration.sh` starts isolated PostgreSQL, the real matcher and testkit API, launches two headless Godot clients, seeds deterministic RTT projections, observes the real WebSocket `OPEN` proposal, accepts from both clients, and verifies both durable tickets plus the proposal reach `ACCEPTED` (including authoritative recovery when the concurrent revision advances). Allocator, production outbox wiring and Redis fan-out live verification remain | | 8.41 `[D:7.8,8.9,8.31,8.40]` | **IN PROGRESS.** Authenticated `GET /v1/assignments/{matchId}` now exposes only a validated, player-scoped assignment view including the hosted endpoint; Godot `AssignmentState`/`ControlPlaneClient.fetch_assignment()` bind the response to the authenticated player, recheck expiry, validate and retain the endpoint, and `connect_to_assignment()` now starts only the validated ENet/Steam transport after assignment readiness; the opaque join authorisation is carried in the MatchNet hello payload rather than the endpoint URL; allocated Godot servers now fail closed unless an operator-mounted JSON roster of control-plane signed envelopes and an HMAC-SHA256 key are present, and MatchNet verifies the canonical Go claim bytes, checks exact token membership plus match/server/protocol/slot/expiry claims before admitting a peer, rejects concurrent reuse of an active token, tracks server-owned reconnect generations across the 60-second reclaim window, fences expired reclaims and invalidates old peers; migration 0002 and the Go store adapter now persist/recover the complete player-scoped assignment projection with conflict-safe identical replay; `AssignmentProviderFromStore` wires that durable projection into the API injection point; `SaveAssignments` publishes a complete signed roster atomically instead of allowing partial player visibility; `SaveVerifiedAssignmentRoster` rechecks signed claims before deriving player rows | `server/api/service.go`, `store_adapters.go`, `service_test.go`, `assignment_state.gd`, `control_plane_client.gd`, `match_net.gd`, `server_boot.gd`, `server_config.gd`, `test_assignment_state.gd`, `test_control_plane_client.gd`, `test_match_net.gd`, `server/contracts/v1/openapi.json`, `server/migrations/0002_assignments.sql` and `server/store/assignment_sql.go` cover participant/identity/expiry/shape/transport/endpoint boundaries, player-scoped schema keys, expiry-filtered reads, assignment upsert conflict handling, atomic batch validation, signed-claim binding, strict endpoint splitting, allowlisted claim rejection, canonical HMAC interoperability, forged-signature rejection, duplicate active-token rejection, reclaim generation/expiry behavior and 146-test Godot compatibility coverage; the three-process disconnect/reclaim smoke passes ship retention, replacement drive and old-peer invalidation; SDR relay-ticket installation, live allocated-token process integration and live Godot/PostgreSQL verification remain | | 8.42 `[D:8.22,8.23,8.24,8.40]` | **IN PROGRESS.** `RankedProfileState` and `ControlPlaneClient.fetch_ranked_profile()` expose the backend-authoritative rating/RD/volatility/games/tier/provisional/season view; matchmaking UI displays provisional/tier status without client-side rating math | `test_control_plane_client.gd` validates profile shape, numeric safety and provisional display; ranked profile fetch/display, committed revision after reconnect, abandon status and season countdown remain dependent on live auth/backend events and Godot runtime verification | | 8.43 `[D:8.39,8.40,8.41]` | **IN PROGRESS.** Matchmaking client now distinguishes expired queue recovery, session expiry, missing records and retryable control-plane outages; a 401 clears the in-memory token, emits `session_expired` and disables retry until a new session is configured; terminal messages remain visible and active searches are not falsely failed on transient errors | `MatchmakingState` and `ControlPlaneClient` tests cover explicit expiry and the existing terminal/retry-safe state paths; decline, version mismatch, regional outage retry UI, failed reconnect, duplicate-action recovery and live Godot verification remain | diff --git a/scripts/verify_control_plane_proposal_integration.sh b/scripts/verify_control_plane_proposal_integration.sh new file mode 100755 index 00000000..7a9626c0 --- /dev/null +++ b/scripts/verify_control_plane_proposal_integration.sh @@ -0,0 +1,136 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Two-player extension of verify_control_plane_integration.sh: a real +# PostgreSQL instance, the real api.Service (via testkit-api, see that +# script's header for why), a real server/cmd/matcher (the actual production +# binary -- it needs no fake, it only ever touches queue_tickets/proposals), +# and two real headless Godot clients each playing one player through +# login -> queue_create -> (real matcher pairs them) -> proposal accept. +# multiplayer-next.md 8.40 names this as the next scoped extension to the +# single-player harness. + +root_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$root_dir" + +godot_bin="${GODOT_BIN:-godot}" +container_name="cosmic-clash-control-plane-proposal-integration" +database="cosmic_clash_test" +user="cosmic_clash_test" +password="cosmic_clash_test" +pg_port="55435" +api_port="18100" +logs_dir="$(mktemp -d "${TMPDIR:-/tmp}/cosmic-clash-control-plane-proposal.XXXXXX")" + +testkit_pid="" +matcher_pid="" +cleanup() { + local status=$? + if (( status != 0 )); then + for log_file in "$logs_dir"/*.log; do + [[ -f "$log_file" ]] || continue + echo "--- $log_file" >&2 + cat "$log_file" >&2 + done + fi + [[ -n "$testkit_pid" ]] && kill "$testkit_pid" 2>/dev/null || true + [[ -n "$matcher_pid" ]] && kill "$matcher_pid" 2>/dev/null || true + lsof -ti "tcp:${api_port}" 2>/dev/null | xargs -r kill -9 2>/dev/null || true + docker rm -f "$container_name" >/dev/null 2>&1 || true + echo "Control-plane proposal integration logs: $logs_dir" +} +trap cleanup EXIT + +docker rm -f "$container_name" >/dev/null 2>&1 || true +docker run --rm -d --name "$container_name" \ + -e POSTGRES_DB="$database" \ + -e POSTGRES_USER="$user" \ + -e POSTGRES_PASSWORD="$password" \ + -p "${pg_port}:5432" postgres:17-alpine >/dev/null + +for attempt in $(seq 1 30); do + if docker exec "$container_name" pg_isready -U "$user" -d "$database" >/dev/null 2>&1; then + break + fi + if [ "$attempt" = 30 ]; then + echo "PostgreSQL did not become ready" >&2 + exit 1 + fi + sleep 1 +done + +dsn="postgres://${user}:${password}@127.0.0.1:${pg_port}/${database}?sslmode=disable" + +go -C server build -o "$logs_dir/testkit-api" ./cmd/testkit-api +go -C server build -o "$logs_dir/matcher" ./cmd/matcher + +COSMIC_CLASH_POSTGRES_DSN="$dsn" "$logs_dir/testkit-api" --listen="127.0.0.1:${api_port}" --migrations="$root_dir/server/migrations" \ + >"$logs_dir/testkit-api.log" 2>&1 & +testkit_pid=$! + +for attempt in $(seq 1 30); do + if curl -sSf "http://127.0.0.1:${api_port}/healthz" >/dev/null 2>&1; then + break + fi + if [ "$attempt" = 30 ]; then + echo "testkit-api did not become ready" >&2 + exit 1 + fi + sleep 1 +done + +# --interval=250ms: this is the whole test's own latency budget, not a +# production setting -- fast polling here just keeps the smoke test quick. +COSMIC_CLASH_POSTGRES_DSN="$dsn" "$logs_dir/matcher" --playlist=casual --size=2 --interval=250ms --migrations="$root_dir/server/migrations" \ + >"$logs_dir/matcher.log" 2>&1 & +matcher_pid=$! + +"$godot_bin" --headless --path Game res://tests/control_plane_proposal_smoke.tscn -- \ + --control-plane-url="http://127.0.0.1:${api_port}" --role=player-a \ + >"$logs_dir/godot-player-a.log" 2>&1 & +player_a_pid=$! +"$godot_bin" --headless --path Game res://tests/control_plane_proposal_smoke.tscn -- \ + --control-plane-url="http://127.0.0.1:${api_port}" --role=player-b \ + >"$logs_dir/godot-player-b.log" 2>&1 & +player_b_pid=$! + +# The matcher only forms a match from candidates that share a verified +# region (server/domain/matcher.go's commonRegions, over each candidate's +# queue_tickets.predicted_rtt) -- populated for real only via the +# authenticated Steam-relay probe flow (task 8.15/8.16), which this harness +# has no real Steam access to drive. Seed it directly in the same database +# instead of building another fake auth boundary just for this: both real +# clients still go through the real queue/matcher/proposal path end to end, +# only the region-probe INPUT is synthetic, exactly like testkit-api's fake +# Steam login already is for identity. Bounded to the same overall timeout +# the Godot clients themselves use. +seed_deadline=$(( $(date +%s) + 20 )) +while [ "$(date +%s)" -lt "$seed_deadline" ]; do + if ! kill -0 "$player_a_pid" 2>/dev/null && ! kill -0 "$player_b_pid" 2>/dev/null; then + break + fi + docker exec "$container_name" psql -U "$user" -d "$database" -c \ + "UPDATE queue_tickets SET predicted_rtt = '{\"EU\": 20}'::jsonb WHERE state = 'QUEUED' AND client_build = 'smoke-build'" \ + >/dev/null 2>&1 || true + sleep 0.2 +done + +status_a=0 +status_b=0 +wait "$player_a_pid" || status_a=$? +wait "$player_b_pid" || status_b=$? + +if [ "$status_a" -ne 0 ] || [ "$status_b" -ne 0 ] \ + || ! grep -q "^SMOKE PASS:" "$logs_dir/godot-player-a.log" \ + || ! grep -q "^SMOKE PASS:" "$logs_dir/godot-player-b.log"; then + echo "Control-plane proposal integration FAILED (player-a=$status_a player-b=$status_b)" >&2 + docker exec "$container_name" psql -U "$user" -d "$database" -c \ + "SELECT state, client_build, predicted_rtt, count(*) FROM queue_tickets GROUP BY state, client_build, predicted_rtt ORDER BY state" >&2 || true + docker exec "$container_name" psql -U "$user" -d "$database" -c \ + "SELECT proposal_id, state, revision, count(*) AS participants FROM proposals LEFT JOIN proposal_participants USING (proposal_id) GROUP BY proposal_id, state, revision" >&2 || true + cat "$logs_dir/godot-player-a.log" >&2 + cat "$logs_dir/godot-player-b.log" >&2 + exit 1 +fi + +echo "Control-plane proposal integration PASS" diff --git a/server/cmd/matcher/main.go b/server/cmd/matcher/main.go index bb20a3de..a7398ecc 100644 --- a/server/cmd/matcher/main.go +++ b/server/cmd/matcher/main.go @@ -112,6 +112,7 @@ func main() { } return domain.PrepareProposal(id, playlist, formation, nil, domain.RankedArena{}, at) }, + OnError: func(err error) { log.Printf("matcher pass: %v", err) }, } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() diff --git a/server/cmd/testkit-api/main.go b/server/cmd/testkit-api/main.go index 123f13f8..42487511 100644 --- a/server/cmd/testkit-api/main.go +++ b/server/cmd/testkit-api/main.go @@ -16,6 +16,7 @@ import ( "crypto/sha256" "database/sql" "encoding/hex" + "encoding/json" "flag" "fmt" "net" @@ -54,7 +55,7 @@ func main() { if err := migrations.Apply(startupCtx, db, *migrationDir); err != nil { fatalf("apply migrations: %v", err) } - handler := (&api.Service{ + service := &api.Service{ SessionBackend: store.PostgresSessions{DB: db}, SessionIssuer: store.PostgresSessions{DB: db}, SteamLogin: fakeSteamLogin{db: db}, @@ -68,7 +69,8 @@ func main() { ProbeRecorder: store.PostgresQueue{DB: db}, WorkloadVerify: api.WorkloadVerifierFromSignedToken([]byte(*workloadSecret), db), Now: func() time.Time { return time.Now().UTC() }, - }).Handler() + } + handler := service.Handler() listener, err := net.Listen("tcp", *listen) if err != nil { fatalf("listen: %v", err) @@ -79,6 +81,7 @@ func main() { go func() { serveErr <- server.Serve(listener) }() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + go dispatchProposalOutbox(ctx, db, service) select { case err := <-serveErr: if err != nil && err != http.ErrServerClosed { @@ -91,6 +94,46 @@ func main() { } } +func dispatchProposalOutbox(ctx context.Context, db *sql.DB, service *api.Service) { + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + events, err := store.ReadUnpublishedOutbox(ctx, db, 100) + if err != nil { + continue + } + for _, event := range events { + if event.EventType != "proposal_changed" { + continue + } + var controlEvent api.ControlPlaneEvent + var envelope struct { + Event string `json:"event"` + Revision uint64 `json:"revision"` + ResourceID string `json:"resource_id"` + OccurredAt time.Time `json:"occurred_at"` + State string `json:"state"` + PlayerIDs []string `json:"player_ids"` + } + if err := json.Unmarshal(event.Payload, &envelope); err != nil { + continue + } + for _, playerID := range envelope.PlayerIDs { + controlEvent = api.ControlPlaneEvent{Event: envelope.Event, Revision: envelope.Revision, ResourceID: envelope.ResourceID, OccurredAt: envelope.OccurredAt, State: envelope.State, PlayerID: playerID} + if err := service.PublishControlPlaneEvent(controlEvent); err != nil { + continue + } + } + _ = store.MarkOutboxPublished(ctx, db, event.EventID, time.Now().UTC()) + } + } + } +} + // fakeSteamLogin derives a deterministic identity from the ticket string // itself (never a real Steam Web API ticket in this binary) and ensures its // identities row exists so session issuance's foreign key is satisfied. diff --git a/server/matcher/worker.go b/server/matcher/worker.go index 13d037d1..fcb495e1 100644 --- a/server/matcher/worker.go +++ b/server/matcher/worker.go @@ -47,6 +47,7 @@ type Worker struct { Now func() time.Time NextID func() string Prepare PrepareFunc + OnError func(error) } // Run polls until cancellation. A failed attempt is returned so a supervisor @@ -57,6 +58,9 @@ func (w Worker) Run(ctx context.Context, interval time.Duration) error { } for { if _, err := w.RunOnce(ctx); err != nil { + if w.OnError != nil { + w.OnError(err) + } if errors.Is(err, ErrWorkerNotConfigured) || errors.Is(err, ErrUnsupportedPlaylist) || errors.Is(err, ErrInvalidMatcherSize) { return err } diff --git a/server/store/proposal_sql.go b/server/store/proposal_sql.go index 716f78e9..48ce2f8a 100644 --- a/server/store/proposal_sql.go +++ b/server/store/proposal_sql.go @@ -3,6 +3,7 @@ package store import ( "context" "database/sql" + "encoding/json" "fmt" "time" @@ -13,6 +14,10 @@ const ProposalInsertSQL = `INSERT INTO proposals (proposal_id, playlist, state, expires_at, revision, match_region, match_protocol) VALUES ($1, $2, 'OPEN', $3, 0, NULLIF($4, ''), NULLIF($5, 0))` +const ProposalOutboxInsertSQL = `INSERT INTO outbox + (event_id, aggregate_type, aggregate_id, revision, event_type, payload) +VALUES ($1, 'proposal', $2, 0, 'proposal_changed', $3)` + // CreateProposal atomically claims the queue tickets and creates the proposal. // Every statement runs inside the same SERIALIZABLE retry callback; callers // must never publish a proposal from a cache-only candidate list. @@ -27,6 +32,7 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t if _, err := tx.ExecContext(ctx, ProposalInsertSQL, proposal.ProposalID, proposal.Playlist, proposal.ExpiresAt, proposal.Region, proposal.Protocol); err != nil { return err } + players := make([]string, 0, len(proposal.Participants)) for _, participant := range proposal.Participants { ticketID := ticketIDs[participant.PlayerID] if participant.PlayerID == "" || ticketID == "" { @@ -46,6 +52,17 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t if changed != 1 { return fmt.Errorf("queue ticket claim lost") } + players = append(players, participant.PlayerID) + } + payload, err := json.Marshal(map[string]any{ + "event": "proposal_changed", "revision": uint64(0), "resource_id": proposal.ProposalID, + "occurred_at": now, "state": string(proposal.State), "player_ids": players, + }) + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, ProposalOutboxInsertSQL, proposal.ProposalID, proposal.ProposalID, payload); err != nil { + return err } return nil })