mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
fix(allocator): publish signed assignment rosters before servers start
The root blocker (issue #14). The worker bound the provider allocation and stopped. Service.PublishRoster and store.SaveVerifiedAssignmentRoster both existed, fully tested, with zero non-test callers, and the production allocator configured neither a roster store nor a signing key. Nothing ever wrote the assignments table. The allocated supervisor fetches a non-empty roster before it launches the game child, so every real allocation failed at that fetch: no match could reach ASSIGNMENT_READY or accept a player. Existing tests seeded assignments directly, which is exactly why the missing hand-off went unnoticed. The worker now builds one join authorisation per durable participant, signs each with the active key, and publishes them. Participants are read through the same query SaveVerifiedAssignmentRoster re-validates against, so the allocator cannot construct a roster the persistence boundary would reject. The manifest commits to a digest over the whole roster, so a server cannot be handed a truncated roster whose surviving entries are each individually valid. Persist the provider endpoint on the allocation: it arrived on the provider response and was never stored, so a worker crashing between allocating and publishing had no endpoint to recover and would have stranded the match permanently. Republishing is idempotent, so that crash now simply retries. cmd/allocator refuses to start without key material rather than running an allocator that binds allocations and silently strands every match. The k8s allocator Deployment mounts the same key set the Fleet does, and both now take the JSON key map so a rotation can publish several. New integration test drives the real worker through to the supervisor's own roster read path without seeding the assignments table. Verified it fails with "assignments = 0, want 2" when the publish step is removed.
This commit is contained in:
Josh Creek
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 '';
|
||||
@@ -0,0 +1,2 @@
|
||||
ALTER TABLE allocations
|
||||
DROP COLUMN IF EXISTS endpoint;
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user