diff --git a/compose.allocated-smoke.yml b/compose.allocated-smoke.yml index dba4c8ef..065d698c 100644 --- a/compose.allocated-smoke.yml +++ b/compose.allocated-smoke.yml @@ -101,7 +101,7 @@ services: - --transport=enet - --region=EU - --join-authorisations-file=/run/cosmic-clash/join-roster.json - - --join-authorisations-key-file=/run/secrets/cosmic-clash/join-signing-key + - --join-authorisations-key-file=/run/secrets/cosmic-clash/join-signing-keys.json - --readiness-port=7780 environment: COSMIC_CLASH_DRAIN_TOKEN: compose-drain-token @@ -109,4 +109,4 @@ services: COSMIC_CLASH_WORKLOAD_TOKEN: ${COSMIC_CLASH_COMPOSE_WORKLOAD_TOKEN:?allocated smoke workload token is required} volumes: - ${COMPOSE_SMOKE_DIR:-/tmp/cosmic-clash-allocated-smoke}/join-roster.json:/run/cosmic-clash/join-roster.json:ro - - ${COMPOSE_SMOKE_DIR:-/tmp/cosmic-clash-allocated-smoke}/join-signing-key:/run/secrets/cosmic-clash/join-signing-key:ro + - ${COMPOSE_SMOKE_DIR:-/tmp/cosmic-clash-allocated-smoke}/join-signing-keys.json:/run/secrets/cosmic-clash/join-signing-keys.json:ro diff --git a/deploy/k8s/base/allocator-deployment.yaml b/deploy/k8s/base/allocator-deployment.yaml index 92128abe..a5696c90 100644 --- a/deploy/k8s/base/allocator-deployment.yaml +++ b/deploy/k8s/base/allocator-deployment.yaml @@ -60,9 +60,18 @@ spec: - --readiness-max-stale=30s - --workload-token-ttl=2h - --metrics-addr=:9091 + # Without these the allocator binds allocations but never publishes + # an assignment roster, and no allocated match can become joinable. + # The same key material is mounted into game servers by fleet.yaml. + - --join-authorisations-key-file=/run/secrets/cosmic-clash/join-signing-keys.json + - --join-authorisations-key-id=$(COSMIC_CLASH_JOIN_SIGNING_KEY_ID) ports: - name: metrics containerPort: 9091 + volumeMounts: + - name: join-signing-keys + mountPath: /run/secrets/cosmic-clash + readOnly: true readinessProbe: httpGet: path: /readyz @@ -102,3 +111,19 @@ spec: secretKeyRef: name: cosmic-clash-workload key: secret + # Rotation: publish the new key in the Secret everywhere first, + # then move this ID to it, then drop the retired key once no live + # match can still reference it. + - name: COSMIC_CLASH_JOIN_SIGNING_KEY_ID + valueFrom: + secretKeyRef: + name: cosmic-clash-game-server + key: join-signing-key-id + volumes: + - name: join-signing-keys + secret: + secretName: cosmic-clash-game-server + defaultMode: 0400 + items: + - key: join-signing-keys.json + path: join-signing-keys.json diff --git a/deploy/k8s/base/fleet.yaml b/deploy/k8s/base/fleet.yaml index c3192497..ca92eda6 100644 --- a/deploy/k8s/base/fleet.yaml +++ b/deploy/k8s/base/fleet.yaml @@ -80,7 +80,10 @@ spec: - --transport=enet - --region=EU - --join-authorisations-file=/run/cosmic-clash/join-roster.json - - --join-authorisations-key-file=/run/secrets/cosmic-clash/join-signing-key + # The key SET, not one key: an allocated server must accept + # authorisations signed with any currently-valid key so a + # rotation does not break matches already in flight. + - --join-authorisations-key-file=/run/secrets/cosmic-clash/join-signing-keys.json - --readiness-port=7780 env: - name: COSMIC_CLASH_SERVER_ID @@ -121,5 +124,5 @@ spec: secret: secretName: cosmic-clash-game-server items: - - key: join-signing-key - path: join-signing-key + - key: join-signing-keys.json + path: join-signing-keys.json diff --git a/review-findings.md b/review-findings.md new file mode 100644 index 00000000..4eb5c35c --- /dev/null +++ b/review-findings.md @@ -0,0 +1,280 @@ +# Branch review findings + +Review scope: `feat/multiplayer` at `089c127c`, compared with merge-base +`3aa0f5b9` (`origin/master`). This is a second, stricter adversarial pass over +the complete branch. + +## [P0] Ship runnable control-plane and matcher workloads + +**Location:** `Dockerfile:51-100`, `deploy/k8s/base/kustomization.yaml:3-18`, +`deploy/k8s/base/control-plane-deployment.yaml:48-50` + +The Kubernetes base deploys a `control-plane` image, but the Dockerfile neither +builds `cmd/control-plane` nor defines a `control-plane` target. Conversely, the +Dockerfile does build a matcher image, but the Kubernetes base contains no +matcher Deployment at all. Applying the checked-in base therefore cannot +produce the advertised production topology: there is no repository-defined +artifact for one required workload, and no running process that consumes +queued tickets for the other. Tickets can be created but can never become +proposals. + +Add a production control-plane image target (not the fake-login `testkit-api` +target), add separately configured casual and ranked matcher Deployments plus +their network policies/health checks, and make the release pipeline build and +pin every referenced target. Add a rendered-manifest test that asserts every +required role is present and every image maps to a real Docker target. + +## [P0] Wire production Steam authentication and the client sign-in flow + +**Location:** `server/cmd/control-plane/main.go:129-157`, +`server/api/service.go:321-340`, `Game/scripts/control_plane_client.gd:15-23`, +`Game/scripts/control_plane_client.gd:157-168`, +`Game/scripts/control_plane_client.gd:218-221`, +`Game/scripts/main_menu.gd:194-195` + +`newAPIService` never supplies `SteamLogin`, so the production +`POST /v1/session/steam` handler always returns `503 auth_unavailable`. On the +other side, the game starts with an empty token and a localhost base URL; it +has `configure` and `login_steam` methods, but no production code calls either +one and the menu enters matchmaking directly. All matchmaking HTTP operations +then fail locally with `ERR_UNAUTHORIZED`. Only `cmd/testkit-api` supplies an +authentication provider, so the passing integration path is not a deployable +or secure player path. + +Implement and configure the real Steam ticket adapter, expose explicit +control-plane endpoint configuration for release builds, obtain a Steam Web +API ticket through the platform integration, and complete login before +enabling Find Match. Add an end-to-end test using the production binary wiring +(with the external Steam boundary stubbed), rather than the testkit service. + +## [P0] Populate server-derived RTT or every queued candidate is invalid + +**Location:** `server/cmd/control-plane/main.go:134-154`, +`server/api/service.go:1143-1168`, `server/store/queue_sql.go:134-181`, +`server/store/queue_sql.go:208-227`, `server/domain/matcher.go:159-168`, +`Game/scripts/control_plane_client.gd:205-221`, +`server/api/service.go:1175-1181` + +Queue creation persists an empty `predicted_rtt` map, while `validCandidate` +rejects every candidate whose map remains empty. The production control plane +sets `ProbeRecorder` but never sets the `Probe` provider, so the probe endpoint +always returns `503 probe_unavailable`; the Godot client also implements no +probe request at all. As a result, even if a matcher Deployment is added, no +real client-created ticket can participate in a formation. There is a second +cache-coherency failure behind that blocker: a successful probe updates only +PostgreSQL and does not refresh `CandidateIndex`, leaving a previously inserted +Redis candidate with its empty RTT map. In a busy shared keyspace whose TTL is +continually refreshed, that stale candidate need not repair itself. + +Wire regional probe adapters into the production service and have the client +complete authenticated probe collection for supported regions after queuing +(or before making a candidate visible to the matcher), and update/invalidate +the Redis projection after probe persistence. Add a full production-wiring +test proving a newly logged-in client can acquire RTT evidence and be selected +through both the PostgreSQL and Redis paths without direct database seeding. + +## [P0] Publish signed assignment rosters before starting allocated servers + +**Location:** `server/allocator/worker.go:34-79`, +`server/cmd/allocator/main.go:81-93`, `server/allocator/service.go:84-91`, +`server/store/assignment_sql.go:178-331`, +`server/supervisor/supervisor.go:198-224`, +`server/supervisor/supervisor.go:313-388` + +The worker stops after binding the provider allocation. Although +`Service.PublishRoster` and `SaveVerifiedAssignmentRoster` exist, the +production allocator configures no roster store/signing key and never calls +them. The allocated supervisor fetches a non-empty roster before it launches +the game child, so every real allocation fails at that fetch and can never +reach assignment-ready or accept a player. Existing tests seed assignments +directly and therefore bypass the missing production hand-off. + +Define the signing-key ownership and rotation model, build one signed join +authorisation per participant, persist the assignment and roster atomically +with the allocation transition, and make retries idempotent. Exercise the real +allocator worker through supervisor startup without fixture-seeding the +assignment tables. + +## [P0] Allow both game traffic and workload callbacks through NetworkPolicy + +**Location:** `deploy/k8s/base/network-policies.yaml:1-92`, +`deploy/k8s/base/fleet.yaml:54-83` + +The namespace-wide policy selects every pod and denies ingress and egress. No +ingress policy allows UDP/7777 to `game-server` pods, so public players cannot +reach an allocated ENet server. Independently, game-server egress permits TCP +8080 to the control plane, but control-plane ingress permits only pods labelled +`edge-gateway`; the game-server source is not allowed. Consequently roster +fetch, registration, connection receipts, shutdown, and result submission are +all blocked even inside the cluster. + +Add narrowly scoped game-server UDP ingress for the chosen Agones/public relay +source and control-plane TCP ingress from the game-server pod selector. Keep +the default deny and add policy tests for both directions, including a real +NetworkPolicy-enforcing cluster smoke test. + +## [P1] Emit a valid initial-connect outbox envelope so one row cannot poison the queue + +**Location:** `server/store/initial_connect_sql.go:155-164`, +`server/api/outbox.go:95-106`, `server/api/outbox.go:168-195`, +`server/store/outbox.go:46-51` + +`ApplyInitialConnectPlan` writes `state_changed` payloads containing only +`match_id`, `state`, and `action`. The state dispatcher requires `event`, +`revision`, `resource_id`, `occurred_at`, and a non-empty `player_ids` list, so +delivery always rejects that row. Dispatch stops on the first error and the row +is never acknowledged; because reads are ordered oldest-first, the malformed +row is retried forever and can prevent all later state events in the batch from +being delivered. + +Construct the same complete envelope used by the other lifecycle writers (or +centralize envelope creation), include the authoritative participant list, and +add a store-to-dispatch integration test for both LIVE and CANCELLED initial- +connect outcomes. Also isolate/dead-letter permanently invalid rows so one bad +event cannot globally head-of-line block publication. + +## [P1] Load authoritative ratings into ranked matcher candidates + +**Location:** `server/store/queue_sql.go:61-66`, +`server/store/queue_sql.go:105-131`, `server/domain/matcher.go:171-186`, +`server/domain/matcher.go:220-239`, `server/domain/teams.go:59-93` + +The production candidate query does not join or otherwise read the `ratings` +table, and its scan never sets `domain.Candidate.Rating`. All PostgreSQL- +sourced ranked candidates therefore have the Go zero value. Rating tolerance, +selection scoring, and team partitioning all consume that field, so ranked +matchmaking treats every player as identically rated regardless of their +authoritative profile. Unit tests mask the defect by constructing candidates +with ratings directly. + +Populate ranked candidates from the authoritative rating row (with an explicit +default for a genuinely new profile), carry it through Redis, and add store- +backed matcher tests with deliberately distant ratings and a team-balancing +assertion. Never accept a client-supplied rating. + +## [P1] Partition and bound Redis snapshots before filtering by playlist + +**Location:** `server/store/redis_candidates.go:80-85`, +`server/store/redis_candidates.go:141-188`, +`server/cmd/matcher/main.go:67-93` + +Both playlists share one Redis hash/sorted set. `Snapshot` performs an +unbounded `ZRANGEBYSCORE` and `HMGET`, materializes and decodes the whole queue, +then the matcher truncates to its candidate limit *before* filtering by +playlist. A large casual prefix can therefore make the ranked worker see zero +candidates indefinitely even when ranked tickets exist later in the set. A +repair is worse: each matcher captures only its selected playlist as the +durable source, but `Rebuild` replaces the shared keys, so a casual repair can +erase ranked projections and vice versa. The unbounded read also makes each +one-second poll allocate and transfer data proportional to total queue depth. + +Use playlist-specific keys and make the snapshot API accept a hard limit that +is applied by Redis (`LIMIT 0 N`) before transfer. Rebuild only the matching +playlist namespace. Add mixed-playlist and large-backlog tests proving neither +worker can erase/starve the other and that Redis never receives an unbounded +range/HMGET. + +## [P1] Enforce durable identity bans during session issuance and authentication + +**Location:** `server/migrations/0001_initial.sql:5-10`, +`server/store/session_sql.go:16-22`, `server/store/session_sql.go:49-63`, +`server/domain/auth.go:166-197` + +The durable schema has `banned_until` and `ban_reason`, but production session +authentication reads only the `sessions` row and no production store code +reads either ban column. The only ban check is an in-memory `TicketVerifier` +used by domain tests. Once real Steam login is wired, a banned identity can +continue using every existing session until expiry and, unless the future +adapter independently duplicates this policy, can receive new sessions too. +This defeats the server-authoritative anti-abuse boundary. + +Make ban state part of the durable authentication transaction: refuse session +issuance for an active ban and join/check identities on every authenticated +request (or revoke all sessions atomically when applying a ban). Add tests for +immediate enforcement across two control-plane replicas and for expiry/unban +semantics. + +## [P1] Fan out outbox events to every control-plane replica + +**Location:** `deploy/k8s/base/control-plane-deployment.yaml:8-14`, +`server/api/events.go:55-117`, `server/api/events.go:217-230`, +`server/api/outbox.go:69-90`, `server/store/outbox.go:46-60` + +The Deployment runs two replicas, but WebSocket subscribers live only in each +process's in-memory hub. Every replica races to read the same global unpublished +outbox rows, and publishing succeeds even when the winning replica has no +matching local subscriber; that replica then sets the single global +`published_at`. A client connected to the other replica never receives the +event. The REST recovery polls eventually converge, but WebSocket delivery +degrades as replicas are added and short-lived proposal transitions can be +observed late. + +Publish committed events through a shared fan-out transport, or maintain a +durable per-replica/consumer-group cursor so every connection-owning replica +sees them. Do not globally acknowledge merely because a local hub accepted an +event for zero subscribers. Add a two-replica integration test with the client +connected to the non-consuming replica. + +## [P1] Add retention for high-volume idempotency and outbox records + +**Location:** `server/migrations/0001_initial.sql:13-29`, +`server/migrations/0001_initial.sql:147-177`, +`Game/scripts/matchmaking.gd:38-52`, +`Game/scripts/control_plane_client.gd:794-795`, +`server/store/queue_sql.go:262-320`, `server/cmd/maintenance/main.go:57-104` + +Each ten-second queue heartbeat gets a fresh idempotency key and permanently +inserts a new row. Published outbox rows and expired/revoked sessions are also +never purged; the maintenance role performs lifecycle reconciliation only. +At 10,000 queued players, heartbeats alone add roughly 60,000 durable rows per +minute, causing unbounded table/index growth, vacuum pressure, backup growth, +and progressively slower recovery on a service intended to scale horizontally. + +Define retention windows longer than every supported retry/recovery horizon, +index cleanup predicates, and delete/archive in bounded `SKIP LOCKED` batches. +Expose deletion lag/row-count metrics and load-test sustained heartbeat volume +to verify that steady-state storage remains bounded. + +## [P2] Make the observability verifier test reach its intended assertion + +**Location:** `server/security/test_observability_manifests.py:20-32`, +`scripts/verify_observability_manifests.py:16-22` + +`test_checker_rejects_wrong_namespace_and_broad_scrape` copies only the +control-plane ServiceMonitor and rules into its temporary directory. The +verifier first requires `kustomization.yaml` and the allocator ServiceMonitor, +so the test fails on a missing file before it examines the mutated namespace +or scrape path. The security suite is red and the stated regression case is +not covered. + +Copy the complete minimum fixture (including kustomization and allocator +ServiceMonitor), then assert the namespace and `/metrics` mutations separately +so either defect produces the intended diagnostic. + +## [P2] Synchronize the contract test with the renamed connection operation + +**Location:** `server/contracts/v1/test_contracts.py:21-28`, +`server/contracts/v1/openapi.json:54` + +The OpenAPI document calls the endpoint `claimPlayerConnection`, while the +structural test still requires `recordPlayerConnected`. The checked-in +contract suite therefore fails despite the endpoint being present, making the +gate noisy and capable of obscuring real compatibility regressions. + +Choose the intended public operation ID and update the test or document. If +the rename is intentional, document the generated-client compatibility impact +and assert `claimPlayerConnection` consistently. + +## Verification notes + +- `go test ./...`: passed. +- `go test -race ./...`: passed. +- `go vet ./...`: passed. +- Godot unit suite: 220 tests passed with the project-compatible headless + renderer flags. +- Training unit suite: 16 focused generation/evaluation tests passed in + `training/.venv`; the reviewed training changes keep new distributions and + team reward sharing opt-in, so no training-regression finding was raised. +- Contract suite: one failure, recorded above. +- Security manifest suite: one failure, recorded above. +- Script verifier unit suite: 10 tests passed. diff --git a/scripts/verify_allocated_compose.sh b/scripts/verify_allocated_compose.sh index 0574192f..b1e9d626 100755 --- a/scripts/verify_allocated_compose.sh +++ b/scripts/verify_allocated_compose.sh @@ -40,7 +40,7 @@ canonical = b"\0".join(field.encode() for field in fields) signature = base64.urlsafe_b64encode(hmac.new(key, canonical, hashlib.sha256).digest()).rstrip(b"=").decode() envelope = {"Authorisation": {"MatchID": fields[0], "ServerID": fields[1], "PlayerID": fields[2], "SteamID": fields[3], "Slot": 0, "Team": 0, "Protocol": fields[6], "Generation": 1, "ExpiresAt": expires, "KeyID": key_id}, "Signature": signature} # The key file maps key ID -> base64 key so a rotation can publish several. -(directory / "join-signing-key").write_text(json.dumps({key_id: base64.b64encode(key).decode()}) + "\n") +(directory / "join-signing-keys.json").write_text(json.dumps({key_id: base64.b64encode(key).decode()}) + "\n") (directory / "join-roster.json").write_text(json.dumps([base64.urlsafe_b64encode(json.dumps(envelope, separators=(",", ":")).encode()).rstrip(b"=").decode()]) + "\n") PY diff --git a/scripts/verify_kind_agones.sh b/scripts/verify_kind_agones.sh index f547f9c4..19eb0a50 100755 --- a/scripts/verify_kind_agones.sh +++ b/scripts/verify_kind_agones.sh @@ -85,7 +85,8 @@ sed -e "s|ghcr.io/cosmic-clash/game-server@sha256:${zero_digest}|$game_server_im kubectl apply -f deploy/k8s/base/namespace.yaml kubectl -n cosmic-clash create secret generic cosmic-clash-game-server \ --from-literal=drain-token=kind-smoke-drain-token \ - --from-literal=join-signing-key=kind-smoke-signing-key \ + --from-literal=join-signing-keys.json='{"kind-smoke-key":"a2luZC1zbW9rZS1zaWduaW5nLWtleQ=="}' \ + --from-literal=join-signing-key-id=kind-smoke-key \ --dry-run=client -o yaml | kubectl apply -f - kubectl apply -f deploy/k8s/base/service-accounts.yaml kubectl apply -f "$work_dir/fleet.yaml" diff --git a/server/allocator/allocator_integration_test.go b/server/allocator/allocator_integration_test.go index a33acb2f..1d795678 100644 --- a/server/allocator/allocator_integration_test.go +++ b/server/allocator/allocator_integration_test.go @@ -15,6 +15,7 @@ import ( "time" "github.com/cosmic-clash/cosmic-clash/server/agones" + "github.com/cosmic-clash/cosmic-clash/server/domain" "github.com/cosmic-clash/cosmic-clash/server/migrations" "github.com/cosmic-clash/cosmic-clash/server/store" _ "github.com/jackc/pgx/v5/stdlib" @@ -34,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, 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 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 { t.Fatal(err) } if err := migrations.Apply(ctx, db, filepath.Join("..", "migrations")); err != nil { @@ -112,3 +113,149 @@ func TestRealAllocatorWorkerReconcilesAgonesAllocationAndBindsMatch(t *testing.T t.Fatalf("recorded allocations=%d err=%v", recorded, err) } } + +// The root blocker: the worker bound the provider allocation and stopped. +// Service.PublishRoster and store.SaveVerifiedAssignmentRoster both existed but +// had no non-test callers, so nothing in production ever wrote the assignments +// table. The allocated supervisor fetches a non-empty roster before launching +// the game child, so every real allocation died at that fetch and no match +// could reach ASSIGNMENT_READY or accept a player. +// +// This drives the real worker and asserts against the durable tables. It never +// seeds the assignments table, which is exactly how the existing tests missed +// the missing hand-off. +func TestRealAllocatorWorkerPublishesSignedAssignmentRoster(t *testing.T) { + dsn := os.Getenv("COSMIC_CLASH_POSTGRES_DSN") + if dsn == "" { + t.Skip("COSMIC_CLASH_POSTGRES_DSN is not set") + } + db, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + 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 { + t.Fatal(err) + } + if err := migrations.Apply(ctx, db, filepath.Join("..", "migrations")); err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Microsecond) + players := []string{"roster-worker-a", "roster-worker-b"} + for index, player := range players { + if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $2)`, player, "steam-"+player); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'ACCEPTED', 'build-1', 1, $3, $4)`, fmt.Sprintf("roster-worker-ticket-%d", index), player, now, now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + } + if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version) VALUES ('roster-worker-match', 'casual', 'ALLOCATING', 'EU', 1)`); err != nil { + t.Fatal(err) + } + for index, player := range players { + if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('roster-worker-match', $1, $2, $3, $4)`, player, fmt.Sprintf("roster-worker-ticket-%d", index), index*3, index); err != nil { + t.Fatal(err) + } + } + + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + _, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"roster-ready-1","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"}},"status":{"state":"Ready"}}]}`)) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"status":{"state":"Allocated","gameServerName":"roster-ready-1","address":"127.0.0.1","ports":[{"name":"default","port":7777}]}}`)) + })) + defer provider.Close() + + agonesClient := agones.Client{BaseURL: provider.URL, Namespace: "games", HTTP: provider.Client()} + ready, err := agonesClient.ListReadyServers(ctx) + if err != nil || len(ready) != 1 { + t.Fatalf("ready projection = %+v err=%v", ready, err) + } + if err := store.RegisterReadyServer(ctx, db, ready[0], now); err != nil { + t.Fatal(err) + } + + // Two keys, signing with the newer: proves the rotation set is threaded + // through signing and the persistence boundary's re-verification. + keys := JoinSigningKeys{ + ActiveKeyID: "key-new", + Keys: map[string][]byte{"key-old": []byte("retired-key"), "key-new": []byte("active-key")}, + } + worker := Worker{ + Claims: store.AllocatingMatchClaims{DB: db, Transport: "enet"}, + Service: Service{Provider: agonesClient, Durable: store.AllocationRegistry{DB: db}, Roster: store.PostgresRosterStore{DB: db}, Now: func() time.Time { return now }}, + Now: func() time.Time { return now }, + Roster: store.AssignmentRosters{DB: db}, + Keys: keys, + } + processed, err := worker.RunOnce(ctx) + if err != nil || !processed { + t.Fatalf("worker processed=%t err=%v", processed, err) + } + + // One assignment row per participant, which is precisely what the + // ASSIGNMENT_READY transition and the supervisor's roster fetch require. + var assignments int + if err := db.QueryRowContext(ctx, `SELECT count(*) FROM assignments WHERE match_id = 'roster-worker-match'`).Scan(&assignments); err != nil { + t.Fatal(err) + } + if assignments != len(players) { + t.Fatalf("assignments = %d, want %d; the allocator did not publish the roster", assignments, len(players)) + } + + // The supervisor's own read path must return a usable roster. + roster, err := store.GetAssignmentRoster(ctx, db, "roster-worker-match", "roster-ready-1", now) + if err != nil { + t.Fatalf("supervisor roster fetch: %v", err) + } + if len(roster) != len(players) { + t.Fatalf("supervisor roster has %d entries, want %d", len(roster), len(players)) + } + verify := domain.VerifyJoinAuthorisationHMAC(keys.Keys) + seenSlots := map[int]bool{} + for _, encoded := range roster { + var signed domain.SignedJoinAuthorisation + if err := json.Unmarshal(encoded, &signed); err != nil { + t.Fatalf("decode roster entry: %v", err) + } + if signed.Authorisation.KeyID != "key-new" { + t.Fatalf("entry signed with %q, want the active key", signed.Authorisation.KeyID) + } + if !verify(domain.JoinAuthorisationBytes(signed.Authorisation), signed.Signature) { + t.Fatalf("roster entry for %s does not verify", signed.Authorisation.PlayerID) + } + if signed.Authorisation.MatchID != "roster-worker-match" || signed.Authorisation.ServerID != "roster-ready-1" { + t.Fatalf("roster entry bound to the wrong match/server: %+v", signed.Authorisation) + } + seenSlots[signed.Authorisation.Slot] = true + } + if len(seenSlots) != len(players) { + t.Fatalf("roster slots collided: %v", seenSlots) + } + + // Republishing must be idempotent: a worker that crashed after binding but + // before publishing retries this same path. + allocation, recorded, err := store.AllocatingMatchClaims{DB: db, Transport: "enet"}.FindProviderAllocation(ctx, domain.AllocationRequest{ + AllocationID: "allocation-roster-worker-match", MatchID: "roster-worker-match", + Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", + }) + if err != nil || !recorded { + t.Fatalf("recover allocation: recorded=%t err=%v", recorded, err) + } + if allocation.Endpoint == "" { + t.Fatal("the recovered allocation lost its endpoint, so a crashed worker could never republish") + } + if err := worker.publishAssignmentRoster(ctx, allocation); err != nil { + t.Fatalf("republish: %v", err) + } + if err := db.QueryRowContext(ctx, `SELECT count(*) FROM assignments WHERE match_id = 'roster-worker-match'`).Scan(&assignments); err != nil { + t.Fatal(err) + } + if assignments != len(players) { + t.Fatalf("republish duplicated assignments: %d", assignments) + } +} diff --git a/server/allocator/roster.go b/server/allocator/roster.go new file mode 100644 index 00000000..69944d2b --- /dev/null +++ b/server/allocator/roster.go @@ -0,0 +1,99 @@ +package allocator + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +// JoinAuthorisationLifetime bounds how long an issued authorisation may be +// replayed. It must outlive the initial-connect window (a player still loading +// must be able to join) without leaving a usable credential lying around after +// the match it belongs to is over. +const JoinAuthorisationLifetime = 30 * time.Minute + +// AssignmentRosterSource reads the authoritative participants of an allocated +// match. It is deliberately the same query the persistence boundary +// re-validates against, so the allocator cannot construct a roster that +// disagrees with the durable match_participants rows. +type AssignmentRosterSource interface { + LoadAssignmentParticipants(context.Context, domain.Allocation) ([]domain.AssignmentParticipant, error) +} + +// JoinSigningKeys is the allocator's key material. ActiveKeyID names the key +// new authorisations are signed with; Keys holds every currently-valid key so +// verification (including the re-check at the persistence boundary) still +// accepts authorisations issued before a rotation. +type JoinSigningKeys struct { + ActiveKeyID string + Keys map[string][]byte +} + +func (k JoinSigningKeys) validate() error { + if k.ActiveKeyID == "" || len(k.Keys) == 0 { + return fmt.Errorf("join signing keys are not configured") + } + if len(k.Keys[k.ActiveKeyID]) == 0 { + return fmt.Errorf("active join signing key %q is not present in the key set", k.ActiveKeyID) + } + return nil +} + +// BuildSignedRoster turns the durable participants into one signed join +// authorisation each, plus the manifest that commits to the whole set. +// +// Signing each entry proves each individual claim; the manifest's roster +// digest additionally commits to the set, so a server cannot be handed a +// truncated roster whose surviving entries are each individually valid. +func BuildSignedRoster(allocation domain.Allocation, participants []domain.AssignmentParticipant, keys JoinSigningKeys, now time.Time) (domain.Assignment, []domain.SignedJoinAuthorisation, error) { + if err := keys.validate(); err != nil { + return domain.Assignment{}, nil, err + } + if allocation.State != domain.ServerAllocated || allocation.Endpoint == "" || len(participants) == 0 || now.IsZero() { + return domain.Assignment{}, nil, domain.ErrManifestRejected + } + active := keys.Keys[keys.ActiveKeyID] + roster := make([]domain.SignedJoinAuthorisation, 0, len(participants)) + for _, participant := range participants { + signed, err := domain.SignJoinAuthorisationHMAC(domain.JoinAuthorisation{ + MatchID: allocation.MatchID, + ServerID: allocation.ServerID, + PlayerID: participant.PlayerID, + SteamID: participant.SteamID, + Slot: participant.Slot, + Team: participant.Team, + Protocol: strconv.Itoa(allocation.Protocol), + // Generation 1 is the first connection lease. Reconnects fence by + // advancing the durable generation, not by reissuing this token. + Generation: 1, + ExpiresAt: now.Add(JoinAuthorisationLifetime).UTC(), + KeyID: keys.ActiveKeyID, + }, active) + if err != nil { + return domain.Assignment{}, nil, fmt.Errorf("sign join authorisation for %s: %w", participant.PlayerID, err) + } + roster = append(roster, signed) + } + rosterDigest, err := domain.AssignmentRosterDigest(roster) + if err != nil { + return domain.Assignment{}, nil, err + } + assignment := domain.Assignment{ + Allocation: allocation, + Endpoint: allocation.Endpoint, + Manifest: domain.AllocationManifest{ + AllocationID: allocation.AllocationID, + MatchID: allocation.MatchID, + ServerID: allocation.ServerID, + Region: allocation.Region, + Build: allocation.Build, + Protocol: allocation.Protocol, + Transport: allocation.Transport, + RosterDigest: rosterDigest, + }, + } + return assignment, roster, nil +} diff --git a/server/allocator/service.go b/server/allocator/service.go index aa88ee9a..45c3538f 100644 --- a/server/allocator/service.go +++ b/server/allocator/service.go @@ -128,6 +128,11 @@ func (s Service) Allocate(ctx context.Context, request domain.AllocationRequest, } return agones.AllocatedServer{}, err } + // The client-facing endpoint arrives on the provider result, not on the + // allocation. Carry it onto the record so publishing the assignment roster + // -- and recovering after a crash between allocating and publishing -- has + // an endpoint to work from. + result.Allocation.Endpoint = result.Endpoint recorded, err := s.Durable.RecordProviderAllocation(ctx, result.Allocation, now) if err != nil { if s.Metrics != nil { @@ -149,6 +154,7 @@ func (s Service) RecordProviderAllocation(ctx context.Context, result agones.All // Quota is consumed by Allocate before a fresh provider request. This // method only reconciles an already-issued provider result after an // ambiguous write, so consuming here would charge one allocation twice. + result.Allocation.Endpoint = result.Endpoint allocation, err := s.Durable.RecordProviderAllocation(ctx, result.Allocation, now) if s.Metrics != nil { if err != nil { diff --git a/server/allocator/worker.go b/server/allocator/worker.go index ee490448..d6ddbd05 100644 --- a/server/allocator/worker.go +++ b/server/allocator/worker.go @@ -25,6 +25,13 @@ type Worker struct { Claims MatchClaimSource Service Service Now func() time.Time + // Roster and Keys wire the assignment hand-off. Without them the worker + // binds an allocation and stops, nothing ever writes the assignments + // table, and the allocated supervisor's roster fetch fails -- so every + // real allocation dies before the game process launches. They are optional + // only so existing allocation-only tests need no key material. + Roster AssignmentRosterSource + Keys JoinSigningKeys } // RunOnce returns whether it found a claimed match. It never exposes an @@ -76,9 +83,41 @@ func (w Worker) RunOnce(ctx context.Context) (bool, error) { if err := w.Claims.BindAllocatedMatch(ctx, allocation); err != nil { return true, fmt.Errorf("bind allocated match %s: %w", request.MatchID, err) } + if err := w.publishAssignmentRoster(ctx, allocation); err != nil { + return true, fmt.Errorf("publish assignment roster for match %s: %w", request.MatchID, err) + } return true, nil } +// publishAssignmentRoster completes the hand-off from allocation to a joinable +// match. The supervisor fetches a non-empty roster before it launches the game +// child, so skipping this leaves the match stuck short of ASSIGNMENT_READY +// forever. +// +// It is safe to retry: SaveVerifiedAssignmentRoster upserts by (match, player) +// and re-validates every claim against the durable participants, so a worker +// that crashed after binding but before publishing simply republishes on the +// next pass. +func (w Worker) publishAssignmentRoster(ctx context.Context, allocation domain.Allocation) error { + if w.Roster == nil { + // Allocation-only deployments (and the allocation-focused tests) leave + // this unset deliberately. + return nil + } + if err := w.Keys.validate(); err != nil { + return err + } + participants, err := w.Roster.LoadAssignmentParticipants(ctx, allocation) + if err != nil { + return err + } + assignment, roster, err := BuildSignedRoster(allocation, participants, w.Keys, w.Now()) + if err != nil { + return err + } + return w.Service.PublishRoster(ctx, assignment, roster, domain.VerifyJoinAuthorisationHMAC(w.Keys.Keys)) +} + func validateProviderAllocation(request domain.AllocationRequest, result agones.AllocatedServer) error { allocation := result.Allocation if result.Endpoint == "" || allocation.State != domain.ServerAllocated || allocation.AllocationID != request.AllocationID || allocation.MatchID != request.MatchID || allocation.ServerID == "" || allocation.Region != request.Region || allocation.Build != request.Build || allocation.Protocol != request.Protocol || allocation.Transport != request.Transport || allocation.ArenaPath != request.ArenaPath { diff --git a/server/cmd/allocator/main.go b/server/cmd/allocator/main.go index 82a080f0..cb71e441 100644 --- a/server/cmd/allocator/main.go +++ b/server/cmd/allocator/main.go @@ -3,7 +3,10 @@ package main import ( "context" "database/sql" + "encoding/base64" + "encoding/json" "flag" + "fmt" "log" "net/http" "os" @@ -34,6 +37,8 @@ func main() { allocationQuota := flag.Int("allocation-quota", 0, "optional per-replica allocation attempts per region per quota window; zero disables this local guard") allocationQuotaWindow := flag.Duration("allocation-quota-window", time.Minute, "window for --allocation-quota") metricsAddr := flag.String("metrics-addr", envOrDefault("COSMIC_CLASH_ALLOCATOR_METRICS_ADDR", ":9091"), "allocator Prometheus metrics address; empty disables metrics") + joinKeyFile := flag.String("join-authorisations-key-file", os.Getenv("COSMIC_CLASH_JOIN_SIGNING_KEY_FILE"), "JSON file mapping join-signing key ID to base64 key; the same material allocated game servers mount. Required: without it no assignment roster is published and no allocated match can start") + joinKeyID := flag.String("join-authorisations-key-id", os.Getenv("COSMIC_CLASH_JOIN_SIGNING_KEY_ID"), "which key in --join-authorisations-key-file signs new authorisations; other keys stay valid for verification so a rotation does not break in-flight matches") flag.Parse() if *dsn == "" || *agonesURL == "" { fatalf("--dsn/COSMIC_CLASH_POSTGRES_DSN and --agones-url/COSMIC_CLASH_AGONES_URL are required") @@ -47,6 +52,16 @@ func main() { if *allocationQuota < 0 || *allocationQuotaWindow <= 0 || *workloadTokenTTL <= 0 { fatalf("--allocation-quota must be non-negative and --allocation-quota-window/--workload-token-ttl must be positive") } + // Refuse to start without signing material rather than running an + // allocator that binds allocations and silently never publishes a roster, + // which strands every match short of ASSIGNMENT_READY. + if *joinKeyFile == "" || *joinKeyID == "" { + fatalf("--join-authorisations-key-file/COSMIC_CLASH_JOIN_SIGNING_KEY_FILE and --join-authorisations-key-id/COSMIC_CLASH_JOIN_SIGNING_KEY_ID are required; without them allocated matches can never become joinable") + } + joinKeys, err := loadJoinSigningKeys(*joinKeyFile, *joinKeyID) + if err != nil { + fatalf("load join signing keys: %v", err) + } db, err := sql.Open("pgx", *dsn) if err != nil { fatalf("open PostgreSQL: %v", err) @@ -89,7 +104,9 @@ func main() { Metrics: metrics, Now: now, }, - Now: now, + Now: now, + Roster: store.AssignmentRosters{DB: db}, + Keys: joinKeys, } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() @@ -152,3 +169,32 @@ func fatalf(format string, args ...any) { log.Printf("allocator: "+format, args...) os.Exit(1) } + +// loadJoinSigningKeys reads the key ID to base64 key map shared with allocated +// game servers. Every key in the file stays valid for verification; only the +// named one signs, so rotation is: publish the new key everywhere, then point +// --join-authorisations-key-id at it, then drop the old key once no live match +// can still reference it. +func loadJoinSigningKeys(path, activeKeyID string) (allocator.JoinSigningKeys, error) { + raw, err := os.ReadFile(path) + if err != nil { + return allocator.JoinSigningKeys{}, err + } + var encoded map[string]string + if err := json.Unmarshal(raw, &encoded); err != nil { + return allocator.JoinSigningKeys{}, fmt.Errorf("expected a JSON object of key ID to base64 key: %w", err) + } + keys := make(map[string][]byte, len(encoded)) + for keyID, value := range encoded { + key, err := base64.StdEncoding.DecodeString(value) + if err != nil || len(key) == 0 { + return allocator.JoinSigningKeys{}, fmt.Errorf("join signing key %q is not valid base64", keyID) + } + keys[keyID] = key + } + result := allocator.JoinSigningKeys{ActiveKeyID: activeKeyID, Keys: keys} + if len(keys[activeKeyID]) == 0 { + return allocator.JoinSigningKeys{}, fmt.Errorf("active key ID %q is not present in %s", activeKeyID, path) + } + return result, nil +} diff --git a/server/domain/allocator.go b/server/domain/allocator.go index 0f9313d3..75f23a89 100644 --- a/server/domain/allocator.go +++ b/server/domain/allocator.go @@ -46,6 +46,10 @@ type Allocation struct { Transport string State ServerLifecycle AllocatedAt time.Time + // Endpoint is the client-facing address the provider returned. It is + // persisted so a worker that crashes between allocating and publishing the + // assignment roster can recover it instead of stranding the match. + Endpoint string } type Allocator struct { diff --git a/server/domain/assignment.go b/server/domain/assignment.go index 62e4aad9..c56951f8 100644 --- a/server/domain/assignment.go +++ b/server/domain/assignment.go @@ -48,3 +48,14 @@ func manifestBytes(manifest AllocationManifest) []byte { func ManifestDigest(manifest AllocationManifest) [32]byte { return sha256.Sum256(manifestBytes(manifest)) } + +// AssignmentParticipant is the durable roster row the allocator turns into one +// signed join authorisation. It lives here rather than in the store so the +// allocator can consume it through an interface without depending on the +// persistence package. +type AssignmentParticipant struct { + PlayerID string + SteamID string + Slot int + Team int +} diff --git a/server/domain/join_auth.go b/server/domain/join_auth.go index 65c2e1b4..763da837 100644 --- a/server/domain/join_auth.go +++ b/server/domain/join_auth.go @@ -1,9 +1,12 @@ package domain import ( + "bytes" "crypto/hmac" "crypto/sha256" + "encoding/hex" "fmt" + "sort" "time" ) @@ -56,3 +59,54 @@ func (r *RankedConnections) AdmitSigned(signed SignedJoinAuthorisation, verify f } return r.Admit(signed.Authorisation, now) } + +// AssignmentRosterDigest binds a manifest to the exact roster it was issued +// with. Signing each authorisation individually proves each claim, but the +// manifest also has to commit to the set, so a server cannot be handed a +// truncated roster whose entries are each individually valid. +// +// Entries are hashed in slot order so the digest is independent of the order +// the caller happened to build them in. +func AssignmentRosterDigest(roster []SignedJoinAuthorisation) (string, error) { + if len(roster) == 0 { + return "", ErrJoinAuthorisation + } + ordered := make([]SignedJoinAuthorisation, len(roster)) + copy(ordered, roster) + sort.Slice(ordered, func(i, j int) bool { + return ordered[i].Authorisation.Slot < ordered[j].Authorisation.Slot + }) + digest := sha256.New() + for _, signed := range ordered { + if signed.Authorisation.PlayerID == "" { + return "", ErrJoinAuthorisation + } + digest.Write(JoinAuthorisationBytes(signed.Authorisation)) + digest.Write([]byte{0}) + } + return hex.EncodeToString(digest.Sum(nil)), nil +} + +// VerifyJoinAuthorisationHMAC builds the verifier the persistence boundary +// re-checks each signature with, selecting the key named by the claim. Keys is +// key ID to raw key; an unknown ID verifies as false rather than falling back +// to any other key. +func VerifyJoinAuthorisationHMAC(keys map[string][]byte) func([]byte, []byte) bool { + return func(claims, signature []byte) bool { + if len(keys) == 0 || len(claims) == 0 || len(signature) == 0 { + return false + } + // The key ID is the last NUL-separated field of the canonical bytes. + separator := bytes.LastIndexByte(claims, 0) + if separator < 0 { + return false + } + key, known := keys[string(claims[separator+1:])] + if !known || len(key) == 0 { + return false + } + mac := hmac.New(sha256.New, key) + mac.Write(claims) + return hmac.Equal(mac.Sum(nil), signature) + } +} diff --git a/server/migrations/0016_allocation_endpoints.sql b/server/migrations/0016_allocation_endpoints.sql new file mode 100644 index 00000000..7c87b75c --- /dev/null +++ b/server/migrations/0016_allocation_endpoints.sql @@ -0,0 +1,8 @@ +-- The allocator learns the server's client-facing endpoint from the provider +-- allocation response, but nothing persisted it. Publishing the assignment +-- roster needs that endpoint, and a worker that crashed between allocating and +-- publishing had no way to recover it -- FindProviderAllocation would report +-- the allocation as already recorded while the endpoint was gone, leaving the +-- match permanently unable to reach ASSIGNMENT_READY. +ALTER TABLE allocations + ADD COLUMN endpoint TEXT NOT NULL DEFAULT ''; diff --git a/server/migrations/down/0016_allocation_endpoints.sql b/server/migrations/down/0016_allocation_endpoints.sql new file mode 100644 index 00000000..b3ea65af --- /dev/null +++ b/server/migrations/down/0016_allocation_endpoints.sql @@ -0,0 +1,2 @@ +ALTER TABLE allocations + DROP COLUMN IF EXISTS endpoint; diff --git a/server/store/allocation_match_sql.go b/server/store/allocation_match_sql.go index c734ef51..f50eea55 100644 --- a/server/store/allocation_match_sql.go +++ b/server/store/allocation_match_sql.go @@ -173,7 +173,7 @@ func FindProviderAllocation(ctx context.Context, db *sql.DB, request domain.Allo } var allocation domain.Allocation var digest []byte - err := db.QueryRowContext(ctx, SelectAllocationSQL, request.AllocationID).Scan(&allocation.AllocationID, &allocation.MatchID, &allocation.ServerID, &allocation.Region, &allocation.Build, &allocation.Protocol, &allocation.ArenaPath, &allocation.Transport, &allocation.AllocatedAt, &digest) + err := db.QueryRowContext(ctx, SelectAllocationSQL, request.AllocationID).Scan(&allocation.AllocationID, &allocation.MatchID, &allocation.ServerID, &allocation.Region, &allocation.Build, &allocation.Protocol, &allocation.ArenaPath, &allocation.Transport, &allocation.AllocatedAt, &digest, &allocation.Endpoint) if err == sql.ErrNoRows { return domain.Allocation{}, false, nil } diff --git a/server/store/allocator_sql.go b/server/store/allocator_sql.go index 36dfa63a..1f66d4c6 100644 --- a/server/store/allocator_sql.go +++ b/server/store/allocator_sql.go @@ -31,11 +31,11 @@ WHERE server_id = ( RETURNING server_id` const InsertAllocationSQL = `INSERT INTO allocations - (allocation_id, match_id, server_id, region, build, protocol_version, arena_path, transport, request_digest, state, allocated_at) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'ALLOCATED', $10)` + (allocation_id, match_id, server_id, region, build, protocol_version, arena_path, transport, request_digest, state, allocated_at, endpoint) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'ALLOCATED', $10, $11)` const SelectAllocationSQL = `SELECT allocation_id, match_id, server_id, region, build, - protocol_version, arena_path, transport, allocated_at, request_digest + protocol_version, arena_path, transport, allocated_at, request_digest, endpoint FROM allocations WHERE allocation_id = $1` const ProviderServerClaimSQL = `UPDATE game_servers SET state = 'ALLOCATED', updated_at = $6 @@ -63,7 +63,7 @@ func ClaimAllocation(ctx context.Context, db *sql.DB, request domain.AllocationR err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { var prior domain.Allocation var priorDigest []byte - err := tx.QueryRowContext(ctx, SelectAllocationSQL, request.AllocationID).Scan(&prior.AllocationID, &prior.MatchID, &prior.ServerID, &prior.Region, &prior.Build, &prior.Protocol, &prior.ArenaPath, &prior.Transport, &prior.AllocatedAt, &priorDigest) + err := tx.QueryRowContext(ctx, SelectAllocationSQL, request.AllocationID).Scan(&prior.AllocationID, &prior.MatchID, &prior.ServerID, &prior.Region, &prior.Build, &prior.Protocol, &prior.ArenaPath, &prior.Transport, &prior.AllocatedAt, &priorDigest, &prior.Endpoint) if err == nil { if !bytes.Equal(priorDigest, digest[:]) { return domain.ErrConflict @@ -86,7 +86,7 @@ func ClaimAllocation(ctx context.Context, db *sql.DB, request domain.AllocationR return err } allocation = domain.Allocation{AllocationID: request.AllocationID, MatchID: request.MatchID, ServerID: serverID, Region: request.Region, Build: request.Build, Protocol: request.Protocol, ArenaPath: request.ArenaPath, Transport: request.Transport, State: domain.ServerAllocated, AllocatedAt: now} - _, err = tx.ExecContext(ctx, InsertAllocationSQL, request.AllocationID, request.MatchID, serverID, request.Region, request.Build, request.Protocol, request.ArenaPath, request.Transport, digest[:], now) + _, err = tx.ExecContext(ctx, InsertAllocationSQL, request.AllocationID, request.MatchID, serverID, request.Region, request.Build, request.Protocol, request.ArenaPath, request.Transport, digest[:], now, "") return err }) return allocation, err @@ -106,7 +106,7 @@ func RecordProviderAllocation(ctx context.Context, db *sql.DB, allocation domain err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { var prior domain.Allocation var priorDigest []byte - err := tx.QueryRowContext(ctx, SelectAllocationSQL, allocation.AllocationID).Scan(&prior.AllocationID, &prior.MatchID, &prior.ServerID, &prior.Region, &prior.Build, &prior.Protocol, &prior.ArenaPath, &prior.Transport, &prior.AllocatedAt, &priorDigest) + err := tx.QueryRowContext(ctx, SelectAllocationSQL, allocation.AllocationID).Scan(&prior.AllocationID, &prior.MatchID, &prior.ServerID, &prior.Region, &prior.Build, &prior.Protocol, &prior.ArenaPath, &prior.Transport, &prior.AllocatedAt, &priorDigest, &prior.Endpoint) if err == nil { if !bytes.Equal(priorDigest, digest[:]) || prior.ServerID != allocation.ServerID { return domain.ErrConflict @@ -133,7 +133,7 @@ func RecordProviderAllocation(ctx context.Context, db *sql.DB, allocation domain } recorded = allocation recorded.AllocatedAt = now - _, err = tx.ExecContext(ctx, InsertAllocationSQL, allocation.AllocationID, allocation.MatchID, serverID, allocation.Region, allocation.Build, allocation.Protocol, allocation.ArenaPath, allocation.Transport, digest[:], now) + _, err = tx.ExecContext(ctx, InsertAllocationSQL, allocation.AllocationID, allocation.MatchID, serverID, allocation.Region, allocation.Build, allocation.Protocol, allocation.ArenaPath, allocation.Transport, digest[:], now, allocation.Endpoint) return err }) return recorded, err diff --git a/server/store/assignment_sql.go b/server/store/assignment_sql.go index 2f560357..f7a449c1 100644 --- a/server/store/assignment_sql.go +++ b/server/store/assignment_sql.go @@ -329,3 +329,47 @@ func GetAssignmentRoster(ctx context.Context, db *sql.DB, matchID, serverID stri } return roster, nil } + +// LoadAssignmentParticipants reads the authoritative roster for an allocated +// match. It reuses AssignmentExpectedRosterSQL -- the same query +// SaveVerifiedAssignmentRoster re-validates against -- so the allocator cannot +// build a roster the persistence boundary would then reject for disagreeing +// with the durable participants. +func LoadAssignmentParticipants(ctx context.Context, db *sql.DB, allocation domain.Allocation) ([]domain.AssignmentParticipant, error) { + if db == nil || allocation.MatchID == "" || allocation.AllocationID == "" || allocation.ServerID == "" { + return nil, fmt.Errorf("invalid assignment participant arguments") + } + rows, err := db.QueryContext(ctx, AssignmentExpectedRosterSQL, + allocation.MatchID, allocation.AllocationID, allocation.ServerID, + allocation.Region, allocation.Protocol, allocation.Build, allocation.Transport) + if err != nil { + return nil, err + } + defer rows.Close() + var participants []domain.AssignmentParticipant + for rows.Next() { + var participant domain.AssignmentParticipant + if err := rows.Scan(&participant.PlayerID, &participant.SteamID, &participant.Slot, &participant.Team); err != nil { + return nil, err + } + if participant.PlayerID == "" || participant.SteamID == "" || participant.Slot < 0 || participant.Slot > 5 || participant.Team < 0 || participant.Team > 1 || participant.Slot/3 != participant.Team { + return nil, fmt.Errorf("assignment participant %q has an invalid slot/team", participant.PlayerID) + } + participants = append(participants, participant) + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(participants) == 0 { + return nil, fmt.Errorf("allocated match %s has no durable participants", allocation.MatchID) + } + return participants, nil +} + +// AssignmentRosters adapts the participant loader to the allocator's +// AssignmentRosterSource interface. +type AssignmentRosters struct{ DB *sql.DB } + +func (a AssignmentRosters) LoadAssignmentParticipants(ctx context.Context, allocation domain.Allocation) ([]domain.AssignmentParticipant, error) { + return LoadAssignmentParticipants(ctx, a.DB, allocation) +}