mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
feat(multiplayer): persist connection generation leases
This commit is contained in:
Josh Creek
+3
-3
@@ -1183,7 +1183,7 @@ production fallback.
|
||||
| 8.2 `[D:8.1]` | **DONE.** Encode the launch SLOs from `docs/MATCHMAKING.md`: RTT, allocation/connect latency, 99.9% allocation/result success, API latency and tick health | [`docs/MATCHMAKING-SLOs.md`](docs/MATCHMAKING-SLOs.md) defines each metric, denominator, percentile/window, owner, alert threshold and release evidence |
|
||||
| 8.3 `[D:8.1]` | **DONE.** Publish versioned OpenAPI + WebSocket contracts for Steam login/session, profile/rating, queue create/heartbeat/cancel/resume, proposal accept/decline, assignment/status, server registration/roster/result/shutdown | [`server/contracts/v1/`](server/contracts/v1/) contains machine-readable REST/events contracts and dependency-free structural tests; REST resync is specified by the contract; `server/api/service.go` also exposes the documented `/api/v1` route names (including server-assigned idempotent queue ticket IDs and DELETE cancellation) alongside the existing client `/v1` routes, covered by `TestDocumentedContractRoutesAdaptToServiceAPI` |
|
||||
| 8.4 `[D:8.3]` | **DONE.** Define opaque IDs, legal queue/match state transitions, revisions and idempotency keys | [`server/contracts/v1/state-transitions.json`](server/contracts/v1/state-transitions.json) locks terminal states, legal edges, stale-revision handling and same-key replay/conflict behavior; contract tests cover the invariants |
|
||||
| 8.5 `[D:8.4]` | **IN PROGRESS.** Initial PostgreSQL migration now defines durable idempotency keys, queue ownership/active-participation fencing, identities, sessions/revocations, ranked seasons, ratings/events, matches/participants, penalties, results, audits and outbox; follow-up migrations persist server-derived queue probe RTT metadata, the allocator GameServer/allocation registry, matcher-selected proposal region/protocol/team/slot plans, leased allocating-match claims, and optional shared regional allocation quotas | `server/migrations/0001_initial.sql`, `0003_queue_probe_metadata.sql`, `0004_allocator_registry.sql`, `0005_proposal_match_plans.sql`, `0006_match_allocation_claims.sql`, `0007_allocation_quotas.sql`, `migrations/runner.go`, `cmd/migrate` and static checks cover the durable tables, uniqueness/check constraints, Redis-as-cache boundary and serialized forward migration recording; opt-in `scripts/run_postgres_integration.sh` now runs the runner and real queue/assignment ownership, idempotency, revision and expiry checks through pgx. `migrations.Rollback` now reverses N most-applied migrations via `migrations/down/<version>.sql` files (one per existing migration, dropping in FK-safe reverse order), wired into `cmd/migrate --rollback=N`, verified live: roll back to empty and reapply reaches the same schema; remaining serializable adapters and cache-loss repair remain |
|
||||
| 8.5 `[D:8.4]` | **IN PROGRESS.** Initial PostgreSQL migration now defines durable idempotency keys, queue ownership/active-participation fencing, identities, sessions/revocations, ranked seasons, ratings/events, matches/participants, penalties, results, audits and outbox; follow-up migrations persist server-derived queue probe RTT metadata, the allocator GameServer/allocation registry, matcher-selected proposal region/protocol/team/slot plans, leased allocating-match claims, optional shared regional allocation quotas, initial-connect timing, and participant disconnect lease timestamps | `server/migrations/0001_initial.sql` through `0011_connection_leases.sql`, `migrations/runner.go`, `cmd/migrate` and static checks cover the durable tables, uniqueness/check constraints, Redis-as-cache boundary and serialized forward migration recording. Migration 0011 backfills legacy connected participants to generation one before enforcing its lease invariant, avoiding an upgrade-only failure on their next write. `migrations.Rollback` reverses N most-applied migrations via matching down files; prior live rollback/reapply verification remains valid, while 0011 awaits a live database rerun because local Docker storage is exhausted |
|
||||
| 8.6 `[D:8.3,8.4]` | **IN PROGRESS.** Add allocated-mode `ServerConfig` compatibility fields as opt-in defaults | `ServerConfig` now validates allocation mode, match/server IDs, playlist version, client build, future assignment expiry, image digest, transport and EU/NA region; `server_boot.gd` fails closed for the not-yet-wired Steam SDR transport, constrains allocated processes to one match, and emits allocation identity/transport in `server_started`; signed-authorisation admission, dynamic endpoint wiring and full manifest/runtime tests remain |
|
||||
|
||||
#### 8B — Authentication and secure control plane
|
||||
@@ -1192,7 +1192,7 @@ production fallback.
|
||||
|---|---|---|
|
||||
| 8.7 `[D:7.6,8.3]` | **IN PROGRESS.** Pure Go ticket policy binds the expected App ID and verified identity, rejects expiry/replay/wrong app/malformed tickets, and consumes each ticket once. `cmd/control-plane` now wires `SessionIssuer: store.PostgresSessions{DB: db}` (same discovery/fix pattern as §8.10's `ResultSubmitter`: the adapter already correctly implemented `Issue`, just wasn't wired, so `/v1/session/steam` 503'd even before considering whether `SteamLogin` — the real, still-correctly-unwired Steam blocker — was available) | `server/domain/auth.go` covers single-use and binding invariants; real `AuthenticateUserTicket` backend adapter, bans, publisher secret store and Steam verification remain. A separate test-only `server/cmd/testkit-api` binary (never referenced by any Dockerfile/K8s manifest) substitutes a fake Steam login accepting any non-empty ticket, enabling §8.40's real Go+Postgres+Godot integration test |
|
||||
| 8.8 `[D:8.7]` | **IN PROGRESS.** Pure Go session policy issues opaque short-lived tokens, stores only digests, authenticates by verified player identity and supports revocation; the production control-plane uses bounded atomic account+IP request limits | `server/domain/auth.go` covers wrong-token/expiry/revocation behavior; `server/store/session_sql.go` provides durable digest/revocation persistence and `server/api/rate_limit.go` plus `cmd/control-plane` provide per-replica request limiting; distributed revocation coordination and live Steam/session integration remain |
|
||||
| 8.9 `[D:8.4,8.7]` | **IN PROGRESS.** Pure Go join policy binds SteamID/player/match/server/team/slot/protocol/expiry, rejects duplicate roster slots, permits same-identity reclaim through backend loss and fences prior server-owned generations | `server/domain/reconnect.go` covers SteamID/server/slot binding, expiry, repeated reclaim, grace boundary and old-generation fencing; signed token issuance/verification, persistent leases, Godot `hello` transport and production integration remain |
|
||||
| 8.9 `[D:8.4,8.7]` | **IN PROGRESS.** Pure Go join policy binds SteamID/player/match/server/team/slot/protocol/expiry, rejects duplicate roster slots, permits same-identity reclaim through backend loss and fences prior server-owned generations. A workload-authenticated durable lease API now atomically claims the next exact generation and records exact-generation disconnects against the allocation/match/server/participant roster | `server/domain/reconnect.go`, `server/store/server_connection_sql.go`, migration 0011, `/servers/{serverId}/{connect|disconnect}`, and adversarial tests cover active duplicate claims, stale disconnect fencing, exact 60-second reclaim, wrong binding, initial assignment expiry, retry-safe receipts, and migration backfill. Godot still uses its local lease during admission; pre-admission durable claim/fallback reconciliation and cross-process runtime verification remain |
|
||||
| 8.10 `[D:8.5,8.31]` | **IN PROGRESS.** Dependency-free workload credential policy validates an adapter-verified signature, issuer/audience/time bounds, namespace/service account, pod UID, GameServer UID, allocation ID, match ID and server ID before result submission; `server/workload` now parses projected compact JWTs, rejects unsigned/malformed/ambiguous tokens, verifies signing input through an injected trust callback, and delegates exact binding to the domain policy; the API exposes the workload-authenticated server result route. **`WorkloadVerify` is now wired for real** -- the blocker this row previously named (`cmd/control-plane/main.go` never wiring it, so `/register` and `/result` 503'd on every real request) is closed by taking a different, equally-valid approach instead of the Kubernetes-JWT one: `serverMutation` only ever compares `WorkloadBinding.ServerID`/`.MatchID` (and `AdvanceServerRegistration` only additionally needs `.AllocationID`) -- nothing downstream requires Namespace/ServiceAcct/PodUID/GameServerUID, so a Kubernetes-issued token was never actually required, just the assumption that it was. `server/workload/signed_token.go` mints and verifies a short-lived HMAC-signed token with a secret only the control plane holds (the same trust model `domain.SessionStore` already uses for player sessions), needing no cluster JWKS/TokenReview to validate -- signature and expiry are fully self-contained. The token binds only `allocation_id` (deliberately, not `match_id`/`server_id` too: it must be requestable in the same Agones request that asks for a server, before Agones has picked one) -- `store.AllocationBindingByAllocationID` resolves `match_id`/`server_id` durably from the `allocations` table at verify time, so a token can never claim a pairing that wasn't actually, durably allocated. `api.WorkloadVerifierFromSignedToken` combines both and is wired into `cmd/control-plane` (`--workload-secret`/`COSMIC_CLASH_WORKLOAD_SECRET`; a startup warning fires and the route stays 503 if it's left unset) and `cmd/testkit-api` (fixed test secret). **The delivery channel is now wired too**: `agones.Client` gains `WorkloadSecret`/`WorkloadTokenTTL` -- when set, `Allocate` mints a token for the allocation and requests it as a third `cosmic-clash.io/workload-token` annotation alongside match-id/allocation-id, wired from `cmd/allocator`'s own `--workload-secret`/`COSMIC_CLASH_WORKLOAD_SECRET` (must match the control plane's); `supervisor.Supervisor.workloadToken()` resolves the bearer credential with an explicit `--workload-token-path` (kept for a possible future Kubernetes-JWT path) always winning, otherwise falling back to that same annotation -- the same fallback pattern `matchID()` already used for `cosmic-clash.io/match-id`. `WorkloadTokenPath` is accordingly no longer required at construction time | `server/domain/workload.go`, `server/workload/jwt.go`, `server/api/service.go` and adversarial tests reject every binding mutation, missing/unverified signature, `none`/malformed JWT, ambiguous audience, server/match mismatch and time boundary; `server/testkit/pipeline_test.go` carries allocation identity through the offline result path; `cmd/control-plane` now wires `ResultSubmitter: store.PostgresResults{DB: db}` (a ready-made adapter that had been referenced from nowhere at all, not even a test); `server/workload/signed_token_test.go` covers round-trip, tampered payload, wrong secret, malformed input and the exact expiry boundary; `server/api/workload_verifier_integration_test.go` (opt-in, real PostgreSQL) covers acceptance against a real allocation row, rejection of an unrecorded allocation, that two distinct real allocations each resolve to their own and only their own match/server pairing, and the previously-503 `Service.WorkloadVerify` field itself now succeeding -- all verified clean with `-race` across multiple runs; `cmd/control-plane/main_test.go`'s `TestServerRoutesRequireWorkloadVerifyToBeWired` now documents and pins the misconfigured-secret case specifically, not the "always unwired" case; `server/agones/allocation_test.go`'s `TestAllocateRequestsAWorkloadTokenAnnotationWhenConfigured` and `server/supervisor/supervisor_test.go`'s new annotation-fallback/fails-closed pair cover the delivery channel end to end (mint -> annotation -> supervisor read -> Authorization header) short of a live cluster. **What's still missing**: this has never run against a real Agones cluster (only HTTP-level fakes), so the exact `object_meta` JSON casing Agones actually returns remains unverified from this sandbox (see `GameServer.ObjectMeta`'s existing caveat); live duplicate/conflict alerting also remains |
|
||||
| 8.11 `[D:8.1]` | **DONE.** Write the threat model: forged clients/replay/queues/results, floods/bots, pod/insider compromise, gameplay and API DDoS, SDR signing-key theft, dependencies and denial-of-wallet | [`docs/THREAT-MODEL.md`](docs/THREAT-MODEL.md) records prevention, detection/response, owner and residual risk for every threat; it separates offline CA/online signer, client/game-server/PostgreSQL/Redis trust boundaries and recovery behavior |
|
||||
| 8.12 `[D:8.11]` | **IN PROGRESS.** Provider-neutral Kubernetes baseline enforces restricted namespace admission, non-root/read-only/no-capability workloads, separate service accounts, default-deny networking and explicit edge/data/DNS/Agones/metrics flows; application manifests consume externally populated Secret objects; the Go API and allocator now have bounded per-replica account+IP rate/quota boundaries and metrics endpoints, plus an atomic operator-controlled degraded-mode gate that sheds new login/queue/proposal work while preserving live-match paths | `deploy/k8s/base/` includes hardened control-plane, allocator, and game-server resources; the control-plane now has health probes, a disruption budget, explicit zero-unavailable/one-surge rolling updates with graceful termination, failure-domain spreading/anti-affinity, and Secret-backed `COSMIC_CLASH_POSTGRES_DSN`/`COSMIC_CLASH_WORKLOAD_SECRET` runtime wiring; the authenticated WebSocket now requires RFC 6455 version 13, enforces a bounded 64 KiB frame size, two-minute idle deadline, 120-message/minute inbound budget, and bounded per-player fan-out; `server/security/test_kubernetes_policies.py`, `server/api/rate_limit.go`, `server/api/admission.go`, allocator metrics tests and adversarial checks cover static hardening, secret-reference invariants, fixed-window limits, bounded labels/key memory, atomic account+IP charging, route classification, concurrent toggling and degraded responses; production `cmd/control-plane` now wires the bounded limiter by default with `--rate-limit`, `--rate-limit-window`, and `--rate-limit-max-keys`, and the base Deployment declares those defaults explicitly; start with control-plane `--degraded`, enable with SIGUSR1, disable with SIGUSR2; private-store provisioning, distributed/global quotas, edge DDoS/WAF/origin shielding, encrypted backups and live policy/load tests remain |
|
||||
@@ -1212,7 +1212,7 @@ production fallback.
|
||||
| 8.21 `[D:8.5,8.20]` | **IN PROGRESS.** Pure Go rating core implements canonical Glicko-2, daily inactivity, ranked 1/3 and casual 1/N human-opponent weights, deterministic opponent ordering, and authoritative draw/overtime/abandon scoring; certified result completion now applies per-player updates inside the durable transaction with lexical row locks and revision increments | `server/domain/rating.go`, `server/store/result_sql.go` and tests cover canonical/inactivity/weight/invalid-input, draw/OT/abandon, ordered participant snapshots, lock/value re-read and rating update SQL; live PostgreSQL rating and seasons execution now covered (§8.23), and concurrent result transaction cases are covered: `TestPostgreSQLConcurrentIdenticalResultSubmissionAppliesRatingsExactlyOnce` races 5 identical submissions and confirms one rating application, while `TestPostgreSQLConcurrentConflictingResultSubmissionsKeepOneReceipt` races different payloads and confirms exactly one winner, one conflict, one receipt and one completion event; live maintenance/DB execution remains |
|
||||
| 8.22 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked profile exposes the first ten games as provisional, derives tiers only through validated backend-owned rating bands, and keeps casual ratings outside the API; authenticated HTTP now returns the authoritative ranked view. **"Authoritative ranked view" was durable-adapter-shaped but had no durable adapter**: `rankedProfile`/`profile` only ever read an in-memory map, so every real `GET /v1/profile/ranked` 404'd regardless of a player's actual rating. `RankedProfileProvider` (interface) + `store.PostgresRankedProfiles` close it, preferred over the map when set so existing tests/literals are unaffected; `LastSeasonID`/`SeasonHistory` deliberately left unset (no season pointer on `ratings`, needs its own query/semantics) | `server/domain/rating.go`, `tier_test.go` and `server/api/service.go` cover provisional override, exact band boundaries, malformed policy rejection, session authentication and ranked-only response fields; `server/store/ranked_profile_sql.go`, verified against real PostgreSQL via curl (a fresh identity correctly 404s through the real adapter) and via §8.40's integration test (`control_plane_smoke.gd` now asserts this exact 404 round-trips before queueing); persisted tier policy, client UI and reconnect transport remain |
|
||||
| 8.23 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked-only season policy compresses 25% toward 1500, clamps RD to 200–350, preserves volatility/history, is idempotent by season ID, and defines exact 12-week windows/due detection; migration and Go store now persist a per-player/per-season marker and rating update atomically; `cmd/maintenance` runs bounded due-season batches with signal-bound shutdown | `server/domain/rating.go`, `season_test.go`, `server/migrations/0001_initial.sql` and `server/store/maintenance_sql.go` cover compression, floor/cap, duplicate replay, window boundary, completed-season idempotence, bounded enumeration, row locking and conflict-safe rollover markers; opt-in PostgreSQL execution now covers the durable rating update, marker creation and duplicate replay without a second compression — re-run live for the first time as part of the wider integration-suite verification below, after fixing a test setup gap (a missing `seasons` row tripped the `ranked_season_rollovers` foreign key before the rollover logic itself ran); live maintenance/DB execution remains |
|
||||
| 8.24 `[D:8.9,8.20,8.21]` | **IN PROGRESS.** Ranked connection policy binds match/server/player/Steam identity/team/global slot/protocol/expiry, permits a 60-second same-token reclaim with monotonically increasing server-owned generations, and applies the rolling 7-day 5 m/15 m/1 h/24 h abandon ladder. Admission now rejects an already-connected duplicate, zero-time operations, disconnect-before-admit, duplicate disconnect attempts that could extend grace, and clock-reversed disconnect/reclaim; Godot applies the same reversed-clock fence to its allowlisted signed-token reservation | Go/Godot adversarial fixtures cover signature tampering, every claim binding, active duplicate admission, repeated valid reclaim, old-generation fencing, exact grace boundary, expiry, zero/reversed clocks, and deterministic cooldown ordering. Persistent cross-process lease fencing and full match/result integration remain |
|
||||
| 8.24 `[D:8.9,8.20,8.21]` | **IN PROGRESS.** Ranked connection policy binds match/server/player/Steam identity/team/global slot/protocol/expiry, permits a 60-second same-token reclaim with monotonically increasing server-owned generations, and applies the rolling 7-day 5 m/15 m/1 h/24 h abandon ladder. Admission rejects active duplicates, zero/reversed clocks, disconnect-before-admit, and duplicate disconnects. PostgreSQL now persists generation/disconnect leases with serializable exact-generation CAS: a stale process cannot disconnect a newer generation, an active lease cannot be reclaimed, initial admission requires an unexpired assignment, and later reconnects use the durable grace boundary rather than the short publication expiry | Go/store/API/Godot adversarial fixtures cover signature tampering, every binding, replay/conflict semantics, active duplicate admission, repeated valid reclaim, stale-generation fencing, exact grace boundary, expiry, zero/reversed clocks, deterministic cooldown ordering, and legacy-row migration. The full local gate passes. Godot pre-admission use of the durable API, outage reconciliation, abandonment persistence, and live PostgreSQL/runtime execution remain |
|
||||
| 8.25 `[D:8.10,8.24]` | **IN PROGRESS.** Workload-bound result policy validates match/server identity, nonce, scores, integrity state, canonical digest, and idempotent receipt conflicts. Result ingestion now owns the complete serializable production transition: a `LIVE` match and its active tickets advance through `RESULT_PENDING`, certified ratings are computed from locked authoritative participant rows, then match/tickets become `COMPLETED`, the receipt is acknowledged, and one revisioned outbox event is inserted. Inactive pre-match no-shows are excluded; an active participant's durable `abandoned_at` forces loss scoring. Missing rating rows or ticket-count divergence fail the whole transaction. Zero-time/incomplete receipts fail before database use and conflicts wrap `ErrResultConflict` | Domain/store/API/outbox tests cover workload and digest binding, identical/conflicting concurrency, direct `LIVE` completion, active-ticket completion, abandonment rating input, incomplete roster failure, integrity suppression, ordered rating locks, receipt/outbox atomicity, fan-out retry/ack ordering, and delivery health. PostgreSQL-tagged regressions compile; prior live result/rating/race/fan-out runs remain valid, while the direct-live lifecycle change awaits a live database rerun. Production credentials, Agones annotation persistence/reconciliation, and integrity-evidence adapters remain |
|
||||
|
||||
#### 8D — Agones, allocation and regional scaling
|
||||
|
||||
+38
-10
@@ -44,7 +44,8 @@ type ServerShutdowner interface {
|
||||
ShutdownServer(context.Context, domain.WorkloadBinding, string, string, time.Time) error
|
||||
}
|
||||
type ServerConnectionRecorder interface {
|
||||
RecordPlayerConnected(context.Context, domain.WorkloadBinding, string, string, time.Time) error
|
||||
ClaimPlayerConnection(context.Context, domain.WorkloadBinding, string, uint64, string, time.Time) (uint64, error)
|
||||
RecordPlayerDisconnected(context.Context, domain.WorkloadBinding, string, uint64, string, time.Time) error
|
||||
}
|
||||
|
||||
type QueueBackend interface {
|
||||
@@ -559,7 +560,7 @@ func (s *Service) contractAssignment(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
func (s *Service) contractServerMutation(w http.ResponseWriter, r *http.Request) {
|
||||
// Unlike contractAssignment, the documented shape here is two segments
|
||||
// (/servers/{serverId}/{result|register|roster|connect|shutdown}) — rejecting
|
||||
// (/servers/{serverId}/{result|register|roster|connect|disconnect|shutdown}) — rejecting
|
||||
// any "/" would 404 every real call. Delegate shape validation to
|
||||
// serverMutation, which already enforces the exact operation allowlist.
|
||||
path := strings.TrimPrefix(r.URL.Path, "/api/v1/servers/")
|
||||
@@ -592,7 +593,7 @@ type serverRegistrationRequest struct {
|
||||
|
||||
func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) {
|
||||
parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/servers/"), "/")
|
||||
if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register" && parts[1] != "roster" && parts[1] != "shutdown" && parts[1] != "connect") {
|
||||
if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register" && parts[1] != "roster" && parts[1] != "shutdown" && parts[1] != "connect" && parts[1] != "disconnect") {
|
||||
writeError(w, http.StatusNotFound, "not_found")
|
||||
return
|
||||
}
|
||||
@@ -600,7 +601,7 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
||||
return
|
||||
}
|
||||
if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) || (parts[1] == "roster" && s.Roster == nil) || (parts[1] == "shutdown" && s.ServerShutdowner == nil) || (parts[1] == "connect" && s.ServerConnections == nil) {
|
||||
if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) || (parts[1] == "roster" && s.Roster == nil) || (parts[1] == "shutdown" && s.ServerShutdowner == nil) || ((parts[1] == "connect" || parts[1] == "disconnect") && s.ServerConnections == nil) {
|
||||
writeError(w, http.StatusServiceUnavailable, "server_unavailable")
|
||||
return
|
||||
}
|
||||
@@ -666,9 +667,11 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
if parts[1] == "connect" {
|
||||
if parts[1] == "connect" || parts[1] == "disconnect" {
|
||||
var input struct {
|
||||
PlayerID string `json:"player_id"`
|
||||
PlayerID string `json:"player_id"`
|
||||
Generation uint64 `json:"generation,omitempty"`
|
||||
ExpectedGeneration uint64 `json:"expected_generation,omitempty"`
|
||||
}
|
||||
if !decodeBody(w, r, &input) {
|
||||
return
|
||||
@@ -677,7 +680,23 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
||||
return
|
||||
}
|
||||
if err := s.ServerConnections.RecordPlayerConnected(r.Context(), binding, input.PlayerID, key, now); err != nil {
|
||||
var generation uint64
|
||||
var err error
|
||||
if parts[1] == "connect" {
|
||||
if input.Generation != 0 {
|
||||
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
||||
return
|
||||
}
|
||||
generation, err = s.ServerConnections.ClaimPlayerConnection(r.Context(), binding, input.PlayerID, input.ExpectedGeneration, key, now)
|
||||
} else {
|
||||
if input.Generation == 0 || input.ExpectedGeneration != 0 {
|
||||
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
||||
return
|
||||
}
|
||||
generation = input.Generation
|
||||
err = s.ServerConnections.RecordPlayerDisconnected(r.Context(), binding, input.PlayerID, input.Generation, key, now)
|
||||
}
|
||||
if err != nil {
|
||||
if errors.Is(err, domain.ErrConflict) {
|
||||
writeError(w, http.StatusConflict, "conflict")
|
||||
} else {
|
||||
@@ -686,11 +705,20 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) {
|
||||
// client fault; 503 keeps the game server's bounded retry alive.
|
||||
writeError(w, http.StatusServiceUnavailable, "server_unavailable")
|
||||
}
|
||||
s.logEvent(observability.Event{Event: "server_connect", MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now})
|
||||
s.logEvent(observability.Event{Event: "server_" + parts[1], MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now})
|
||||
return
|
||||
}
|
||||
s.logEvent(observability.Event{Event: "server_connect", MatchID: binding.MatchID, ServerID: parts[0], Stage: "connected", OccurredAt: now, Fields: map[string]any{"player_id": input.PlayerID}})
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
stage := "connected"
|
||||
if parts[1] == "disconnect" {
|
||||
stage = "disconnected"
|
||||
}
|
||||
s.logEvent(observability.Event{Event: "server_" + parts[1], MatchID: binding.MatchID, ServerID: parts[0], Stage: stage, OccurredAt: now, Fields: map[string]any{"player_id": input.PlayerID, "generation": generation}})
|
||||
if parts[1] == "disconnect" {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(map[string]uint64{"generation": generation})
|
||||
return
|
||||
}
|
||||
if parts[1] == "shutdown" {
|
||||
|
||||
+38
-19
@@ -71,16 +71,25 @@ type serverShutdownerSpy struct {
|
||||
}
|
||||
|
||||
type serverConnectionSpy struct {
|
||||
calls int
|
||||
binding domain.WorkloadBinding
|
||||
playerID string
|
||||
key string
|
||||
err error
|
||||
connectCalls int
|
||||
disconnectCalls int
|
||||
binding domain.WorkloadBinding
|
||||
playerID string
|
||||
key string
|
||||
expectedGeneration uint64
|
||||
generation uint64
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *serverConnectionSpy) RecordPlayerConnected(_ context.Context, binding domain.WorkloadBinding, playerID, key string, _ time.Time) error {
|
||||
s.calls++
|
||||
s.binding, s.playerID, s.key = binding, playerID, key
|
||||
func (s *serverConnectionSpy) ClaimPlayerConnection(_ context.Context, binding domain.WorkloadBinding, playerID string, expectedGeneration uint64, key string, _ time.Time) (uint64, error) {
|
||||
s.connectCalls++
|
||||
s.binding, s.playerID, s.expectedGeneration, s.key = binding, playerID, expectedGeneration, key
|
||||
return expectedGeneration + 1, s.err
|
||||
}
|
||||
|
||||
func (s *serverConnectionSpy) RecordPlayerDisconnected(_ context.Context, binding domain.WorkloadBinding, playerID string, generation uint64, key string, _ time.Time) error {
|
||||
s.disconnectCalls++
|
||||
s.binding, s.playerID, s.generation, s.key = binding, playerID, generation, key
|
||||
return s.err
|
||||
}
|
||||
|
||||
@@ -1485,35 +1494,45 @@ func TestServerConnectionAPIRequiresBoundWorkloadAndOpaqueAssignedPlayer(t *test
|
||||
server := httptest.NewServer(service.Handler())
|
||||
defer server.Close()
|
||||
|
||||
request := func(serverID, playerID, token, key string) int {
|
||||
body := fmt.Sprintf(`{"player_id":%q}`, playerID)
|
||||
req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/servers/"+serverID+"/connect", strings.NewReader(body))
|
||||
request := func(operation, serverID, playerID, token, key, bodySuffix string) (int, string) {
|
||||
body := fmt.Sprintf(`{"player_id":%q%s}`, playerID, bodySuffix)
|
||||
req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/servers/"+serverID+"/"+operation, strings.NewReader(body))
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
req.Header.Set("Idempotency-Key", key)
|
||||
response, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
responseBody, _ := io.ReadAll(response.Body)
|
||||
response.Body.Close()
|
||||
return response.StatusCode
|
||||
return response.StatusCode, string(responseBody)
|
||||
}
|
||||
if got := request(binding.ServerID, "player-123456789", "workload-token", "connect-player-123456789"); got != http.StatusNoContent {
|
||||
if got, body := request("connect", binding.ServerID, "player-123456789", "workload-token", "connect-player-123456789", `,"expected_generation":0`); got != http.StatusOK || !strings.Contains(body, `"generation":1`) {
|
||||
t.Fatalf("connection status = %d", got)
|
||||
}
|
||||
if recorder.calls != 1 || recorder.binding != binding || recorder.playerID != "player-123456789" || recorder.key != "connect-player-123456789" {
|
||||
if recorder.connectCalls != 1 || recorder.binding != binding || recorder.playerID != "player-123456789" || recorder.expectedGeneration != 0 || recorder.key != "connect-player-123456789" {
|
||||
t.Fatalf("connection receipt = %+v", recorder)
|
||||
}
|
||||
if got := request("server-000000000", "player-123456789", "workload-token", "connect-player-123456789"); got != http.StatusUnauthorized {
|
||||
if got, _ := request("connect", "server-000000000", "player-123456789", "workload-token", "connect-player-123456789", ""); got != http.StatusUnauthorized {
|
||||
t.Fatalf("wrong server status = %d", got)
|
||||
}
|
||||
if got := request(binding.ServerID, "short", "workload-token", "connect-player-short-123"); got != http.StatusUnprocessableEntity {
|
||||
if got, _ := request("connect", binding.ServerID, "short", "workload-token", "connect-player-short-123", ""); got != http.StatusUnprocessableEntity {
|
||||
t.Fatalf("short player status = %d", got)
|
||||
}
|
||||
if recorder.calls != 1 {
|
||||
t.Fatalf("invalid receipts reached backend: %d", recorder.calls)
|
||||
if recorder.connectCalls != 1 {
|
||||
t.Fatalf("invalid receipts reached backend: %d", recorder.connectCalls)
|
||||
}
|
||||
if got, _ := request("disconnect", binding.ServerID, "player-123456789", "workload-token", "disconnect-player-123456789", `,"generation":1`); got != http.StatusNoContent {
|
||||
t.Fatalf("disconnect status = %d", got)
|
||||
}
|
||||
if recorder.disconnectCalls != 1 || recorder.generation != 1 {
|
||||
t.Fatalf("disconnect receipt = %+v", recorder)
|
||||
}
|
||||
if got, _ := request("disconnect", binding.ServerID, "player-123456789", "workload-token", "disconnect-zero-123456", ""); got != http.StatusUnprocessableEntity || recorder.disconnectCalls != 1 {
|
||||
t.Fatalf("zero-generation disconnect status=%d calls=%d", got, recorder.disconnectCalls)
|
||||
}
|
||||
recorder.err = errors.New("database unavailable")
|
||||
if got := request(binding.ServerID, "player-123456789", "workload-token", "connect-player-retry-123"); got != http.StatusServiceUnavailable {
|
||||
if got, _ := request("connect", binding.ServerID, "player-123456789", "workload-token", "connect-player-retry-123", `,"expected_generation":1`); got != http.StatusServiceUnavailable {
|
||||
t.Fatalf("recorder outage status = %d, want retryable 503", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -89,8 +89,12 @@ func ServerShutdownerFromStore(db *sql.DB) ServerShutdowner {
|
||||
|
||||
type postgresServerConnections struct{ db *sql.DB }
|
||||
|
||||
func (p postgresServerConnections) RecordPlayerConnected(ctx context.Context, binding domain.WorkloadBinding, playerID, idempotencyKey string, now time.Time) error {
|
||||
return store.RecordPlayerConnected(ctx, p.db, binding, playerID, idempotencyKey, now)
|
||||
func (p postgresServerConnections) ClaimPlayerConnection(ctx context.Context, binding domain.WorkloadBinding, playerID string, expectedGeneration uint64, idempotencyKey string, now time.Time) (uint64, error) {
|
||||
return store.ClaimPlayerConnection(ctx, p.db, binding, playerID, expectedGeneration, idempotencyKey, now)
|
||||
}
|
||||
|
||||
func (p postgresServerConnections) RecordPlayerDisconnected(ctx context.Context, binding domain.WorkloadBinding, playerID string, generation uint64, idempotencyKey string, now time.Time) error {
|
||||
return store.RecordPlayerDisconnected(ctx, p.db, binding, playerID, generation, idempotencyKey, now)
|
||||
}
|
||||
|
||||
func ServerConnectionsFromStore(db *sql.DB) ServerConnectionRecorder {
|
||||
|
||||
@@ -51,7 +51,10 @@
|
||||
"post": {"security": [{"serverCredential": []}], "operationId": "registerServer", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerRegistration"}}}}, "responses": {"204": {"description": "Registered"}, "409": {"$ref": "#/components/responses/Conflict"}}}
|
||||
},
|
||||
"/servers/{serverId}/connect": {
|
||||
"post": {"security": [{"serverCredential": []}], "operationId": "recordPlayerConnected", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnection"}}}}, "responses": {"204": {"description": "Connection recorded"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}, "503": {"$ref": "#/components/responses/Unavailable"}}}
|
||||
"post": {"security": [{"serverCredential": []}], "operationId": "claimPlayerConnection", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnectionClaim"}}}}, "responses": {"200": {"description": "Connection generation claimed", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnectionLease"}}}}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}, "503": {"$ref": "#/components/responses/Unavailable"}}}
|
||||
},
|
||||
"/servers/{serverId}/disconnect": {
|
||||
"post": {"security": [{"serverCredential": []}], "operationId": "recordPlayerDisconnected", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerConnectionDisconnect"}}}}, "responses": {"204": {"description": "Disconnection recorded"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}, "503": {"$ref": "#/components/responses/Unavailable"}}}
|
||||
},
|
||||
"/servers/{serverId}/result": {
|
||||
"post": {"security": [{"serverCredential": []}], "operationId": "submitMatchResult", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/MatchResult"}}}}, "responses": {"202": {"description": "Result accepted"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}}}
|
||||
@@ -95,7 +98,9 @@
|
||||
"ProposalParticipant": {"type": "object", "required": ["player_id", "response", "team", "slot"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "response": {"type": "string", "enum": ["PENDING", "ACCEPTED", "DECLINED", "TIMED_OUT"]}, "team": {"type": "integer", "minimum": 0, "maximum": 1}, "slot": {"type": "integer", "minimum": 0, "maximum": 5}}},
|
||||
"Assignment": {"type": "object", "required": ["match_id", "server_id", "player_id", "slot", "expires_at", "protocol_version", "transport", "endpoint", "join_authorisation"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "server_id": {"$ref": "#/components/schemas/OpaqueId"}, "player_id": {"$ref": "#/components/schemas/OpaqueId"}, "slot": {"type": "integer", "minimum": 0, "maximum": 5}, "expires_at": {"type": "string", "format": "date-time"}, "protocol_version": {"type": "integer", "minimum": 1}, "transport": {"type": "string", "enum": ["steam_sdr", "enet"]}, "endpoint": {"type": "string", "minLength": 3, "maxLength": 256}, "join_authorisation": {"type": "string"}}},
|
||||
"ServerRegistration": {"type": "object", "required": ["match_id", "protocol_version", "image_digest", "assignment_ready"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "protocol_version": {"type": "integer", "minimum": 1}, "image_digest": {"type": "string", "pattern": "^sha256:[a-f0-9]{64}$"}, "assignment_ready": {"type": "boolean"}}},
|
||||
"ServerConnection": {"type": "object", "required": ["player_id"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}}},
|
||||
"ServerConnectionClaim": {"type": "object", "required": ["player_id", "expected_generation"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "expected_generation": {"type": "integer", "minimum": 0}}},
|
||||
"ServerConnectionDisconnect": {"type": "object", "required": ["player_id", "generation"], "additionalProperties": false, "properties": {"player_id": {"$ref": "#/components/schemas/OpaqueId"}, "generation": {"type": "integer", "minimum": 1}}},
|
||||
"ServerConnectionLease": {"type": "object", "required": ["generation"], "additionalProperties": false, "properties": {"generation": {"type": "integer", "minimum": 1}}},
|
||||
"ServerShutdown": {"type": "object", "required": ["reason"], "additionalProperties": false, "properties": {"reason": {"type": "string", "minLength": 1, "maxLength": 96}}},
|
||||
"MatchResult": {"type": "object", "required": ["match_id", "result_nonce", "score", "integrity_state"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "result_nonce": {"type": "string", "minLength": 16, "maxLength": 128}, "score": {"type": "object", "required": ["team_0", "team_1"], "additionalProperties": false, "properties": {"team_0": {"type": "integer", "minimum": 0}, "team_1": {"type": "integer", "minimum": 0}}}, "integrity_state": {"type": "string", "enum": ["CERTIFIED", "SUPPRESSED", "REVIEW"]}}}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
ALTER TABLE match_participants
|
||||
ADD COLUMN disconnected_at TIMESTAMPTZ;
|
||||
|
||||
-- The pre-lease connection receipt populated connected_at but did not advance
|
||||
-- the already-present generation column. Preserve those live admissions as
|
||||
-- generation one before enforcing the lease invariant.
|
||||
UPDATE match_participants
|
||||
SET connection_generation = 1
|
||||
WHERE connected_at IS NOT NULL AND connection_generation = 0;
|
||||
|
||||
ALTER TABLE match_participants
|
||||
ADD CONSTRAINT match_participants_connection_lease
|
||||
CHECK (
|
||||
(connection_generation = 0 AND connected_at IS NULL AND disconnected_at IS NULL)
|
||||
OR
|
||||
(connection_generation > 0 AND connected_at IS NOT NULL)
|
||||
) NOT VALID;
|
||||
@@ -0,0 +1,3 @@
|
||||
ALTER TABLE match_participants
|
||||
DROP CONSTRAINT IF EXISTS match_participants_connection_lease,
|
||||
DROP COLUMN IF EXISTS disconnected_at;
|
||||
@@ -606,13 +606,13 @@ func TestPostgreSQLConnectionReceiptsStartCompleteRelaxedCasualRoster(t *testing
|
||||
}
|
||||
}
|
||||
binding := domain.WorkloadBinding{AllocationID: "connect-allocation", MatchID: "connect-match", ServerID: "connect-server"}
|
||||
if err := RecordPlayerConnected(ctx, db, binding, "connect-player-0", "connect-receipt-key-0000", now); err != nil {
|
||||
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 0, "connect-receipt-key-0000", now); err != nil || generation != 1 {
|
||||
t.Fatalf("first receipt: %v", err)
|
||||
}
|
||||
if err := RecordPlayerConnected(ctx, db, domain.WorkloadBinding{AllocationID: "forged-allocation", MatchID: "connect-match", ServerID: "connect-server"}, "connect-player-1", "connect-receipt-key-forged", now); !errors.Is(err, domain.ErrConflict) {
|
||||
if _, err := ClaimPlayerConnection(ctx, db, domain.WorkloadBinding{AllocationID: "forged-allocation", MatchID: "connect-match", ServerID: "connect-server"}, "connect-player-1", 0, "connect-receipt-key-forged", now); !errors.Is(err, domain.ErrConflict) {
|
||||
t.Fatalf("forged binding err=%v, want conflict", err)
|
||||
}
|
||||
if err := RecordPlayerConnected(ctx, db, binding, "connect-player-1", "connect-receipt-key-0001", now); err != nil {
|
||||
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-1", 0, "connect-receipt-key-0001", now); err != nil || generation != 1 {
|
||||
t.Fatalf("second receipt: %v", err)
|
||||
}
|
||||
reconciled, err := ReconcileInitialConnect(ctx, db, now.Add(time.Second), 10)
|
||||
@@ -629,9 +629,24 @@ func TestPostgreSQLConnectionReceiptsStartCompleteRelaxedCasualRoster(t *testing
|
||||
}
|
||||
// A lost 204 can be retried after assignment expiry because the exact
|
||||
// durable receipt is replayed before checking the now-expired assignment.
|
||||
if err := RecordPlayerConnected(ctx, db, binding, "connect-player-0", "connect-receipt-key-0000", now.Add(2*time.Minute)); err != nil {
|
||||
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 0, "connect-receipt-key-0000", now.Add(2*time.Minute)); err != nil || generation != 1 {
|
||||
t.Fatalf("durable receipt replay: %v", err)
|
||||
}
|
||||
if _, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 1, "connect-active-duplicate", now.Add(2*time.Second)); !errors.Is(err, domain.ErrConflict) {
|
||||
t.Fatalf("active duplicate err=%v, want conflict", err)
|
||||
}
|
||||
if err := RecordPlayerDisconnected(ctx, db, binding, "connect-player-0", 1, "disconnect-receipt-0000", now.Add(3*time.Second)); err != nil {
|
||||
t.Fatalf("disconnect receipt: %v", err)
|
||||
}
|
||||
if _, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 0, "connect-receipt-key-0000", now.Add(4*time.Second)); !errors.Is(err, domain.ErrConflict) {
|
||||
t.Fatalf("stale connect replay err=%v, want conflict", err)
|
||||
}
|
||||
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 1, "reconnect-receipt-0000", now.Add(63*time.Second)); err != nil || generation != 2 {
|
||||
t.Fatalf("grace-boundary reconnect generation=%d err=%v", generation, err)
|
||||
}
|
||||
if err := RecordPlayerDisconnected(ctx, db, binding, "connect-player-0", 1, "stale-disconnect-0000", now.Add(64*time.Second)); !errors.Is(err, domain.ErrConflict) {
|
||||
t.Fatalf("stale disconnect err=%v, want conflict", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) {
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
@@ -18,61 +19,173 @@ const ServerConnectionIdempotencyInsertSQL = `INSERT INTO idempotency_keys
|
||||
(scope, idempotency_key, payload_digest, result)
|
||||
VALUES ($1, $2, $3, '{}'::jsonb) ON CONFLICT (scope, idempotency_key) DO NOTHING`
|
||||
|
||||
const ServerConnectionIdempotencySelectSQL = `SELECT payload_digest
|
||||
const ServerConnectionIdempotencySelectSQL = `SELECT payload_digest, result
|
||||
FROM idempotency_keys WHERE scope = $1 AND idempotency_key = $2 FOR UPDATE`
|
||||
|
||||
const ServerConnectionParticipantSQL = `UPDATE match_participants mp
|
||||
SET connected_at = COALESCE(mp.connected_at, $5)
|
||||
FROM matches m, allocations a, assignments assn
|
||||
const ServerConnectionLeaseSQL = `SELECT mp.connection_generation, mp.connected_at, mp.disconnected_at, assn.expires_at
|
||||
FROM match_participants mp
|
||||
JOIN matches m ON m.match_id = mp.match_id
|
||||
JOIN allocations a ON a.allocation_id = $3 AND a.match_id = m.match_id AND a.server_id = m.server_id
|
||||
JOIN assignments assn ON assn.match_id = mp.match_id AND assn.player_id = mp.player_id
|
||||
AND assn.allocation_id = a.allocation_id AND assn.server_id = m.server_id
|
||||
WHERE mp.match_id = $1 AND mp.player_id = $4 AND mp.participation_active
|
||||
AND m.match_id = mp.match_id AND m.server_id = $2
|
||||
AND m.state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING', 'LIVE')
|
||||
AND a.allocation_id = $3 AND a.match_id = m.match_id AND a.server_id = m.server_id
|
||||
AND m.server_id = $2 AND m.state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING', 'LIVE')
|
||||
AND a.state = 'ALLOCATED'
|
||||
AND assn.match_id = mp.match_id AND assn.player_id = mp.player_id
|
||||
AND assn.allocation_id = a.allocation_id AND assn.server_id = m.server_id
|
||||
AND assn.expires_at > $5
|
||||
RETURNING mp.connected_at`
|
||||
FOR UPDATE OF mp`
|
||||
|
||||
// RecordPlayerConnected persists authoritative admission observed by the
|
||||
// allocated game server. The workload allocation, match/server binding,
|
||||
// active participant, and still-live assignment must all agree.
|
||||
func RecordPlayerConnected(ctx context.Context, db *sql.DB, binding domain.WorkloadBinding, playerID, idempotencyKey string, now time.Time) error {
|
||||
if db == nil || binding.AllocationID == "" || binding.MatchID == "" || binding.ServerID == "" || playerID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() {
|
||||
return fmt.Errorf("invalid server connection receipt")
|
||||
const ServerConnectionAdmitSQL = `UPDATE match_participants
|
||||
SET connection_generation = $3, connected_at = COALESCE(connected_at, $4), disconnected_at = NULL
|
||||
WHERE match_id = $1 AND player_id = $2 AND connection_generation = $5
|
||||
RETURNING connection_generation`
|
||||
|
||||
const ServerConnectionDisconnectSQL = `UPDATE match_participants
|
||||
SET disconnected_at = $4
|
||||
WHERE match_id = $1 AND player_id = $2 AND connection_generation = $3
|
||||
AND connected_at IS NOT NULL AND disconnected_at IS NULL
|
||||
RETURNING connection_generation`
|
||||
|
||||
type connectionReceipt struct {
|
||||
Generation uint64 `json:"generation"`
|
||||
}
|
||||
|
||||
// ClaimPlayerConnection atomically acquires the next durable connection
|
||||
// generation. expectedGeneration is server-owned state, never a client claim.
|
||||
// A reconnect is legal only after the exact previous generation was durably
|
||||
// disconnected and while its 60-second grace period remains open.
|
||||
func ClaimPlayerConnection(ctx context.Context, db *sql.DB, binding domain.WorkloadBinding, playerID string, expectedGeneration uint64, idempotencyKey string, now time.Time) (uint64, error) {
|
||||
if err := validateConnectionMutation(db, binding, playerID, idempotencyKey, now); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
digest := sha256.Sum256([]byte(binding.AllocationID + "\x00" + binding.MatchID + "\x00" + binding.ServerID + "\x00" + playerID))
|
||||
return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
||||
inserted, err := tx.ExecContext(ctx, ServerConnectionIdempotencyInsertSQL, ServerConnectionIdempotencyScope, idempotencyKey, digest[:])
|
||||
digest := connectionDigest("connect", binding, playerID, expectedGeneration)
|
||||
var claimed uint64
|
||||
err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
||||
replay, generation, err := beginConnectionMutation(ctx, tx, idempotencyKey, digest)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
changed, err := inserted.RowsAffected()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if changed == 0 {
|
||||
var prior []byte
|
||||
if err := tx.QueryRowContext(ctx, ServerConnectionIdempotencySelectSQL, ServerConnectionIdempotencyScope, idempotencyKey).Scan(&prior); err != nil {
|
||||
if replay {
|
||||
current, connectedAt, disconnectedAt, _, err := lockConnectionLease(ctx, tx, binding, playerID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !bytes.Equal(prior, digest[:]) {
|
||||
if current != generation || !connectedAt.Valid || disconnectedAt.Valid {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
claimed = generation
|
||||
return nil
|
||||
}
|
||||
var connectedAt time.Time
|
||||
if err := tx.QueryRowContext(ctx, ServerConnectionParticipantSQL, binding.MatchID, binding.ServerID, binding.AllocationID, playerID, now).Scan(&connectedAt); err != nil {
|
||||
current, connectedAt, disconnectedAt, assignmentExpiry, err := lockConnectionLease(ctx, tx, binding, playerID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if current != expectedGeneration || expectedGeneration == ^uint64(0) {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
if current == 0 {
|
||||
if connectedAt.Valid || disconnectedAt.Valid || !now.Before(assignmentExpiry) {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
} else if !connectedAt.Valid || !disconnectedAt.Valid || now.Before(disconnectedAt.Time) || now.Sub(disconnectedAt.Time) > domain.RankedReconnectGrace {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
claimed = current + 1
|
||||
if err := tx.QueryRowContext(ctx, ServerConnectionAdmitSQL, binding.MatchID, playerID, claimed, now, current).Scan(&claimed); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
return err
|
||||
}
|
||||
result, err := json.Marshal(map[string]any{"match_id": binding.MatchID, "player_id": playerID, "connected_at": connectedAt})
|
||||
return finishConnectionMutation(ctx, tx, idempotencyKey, claimed)
|
||||
})
|
||||
return claimed, err
|
||||
}
|
||||
|
||||
// RecordPlayerDisconnected closes exactly one active generation. A delayed
|
||||
// disconnect from an older peer can therefore never evict a reclaimed lease.
|
||||
func RecordPlayerDisconnected(ctx context.Context, db *sql.DB, binding domain.WorkloadBinding, playerID string, generation uint64, idempotencyKey string, now time.Time) error {
|
||||
if err := validateConnectionMutation(db, binding, playerID, idempotencyKey, now); err != nil {
|
||||
return err
|
||||
}
|
||||
if generation == 0 {
|
||||
return fmt.Errorf("invalid server disconnect receipt")
|
||||
}
|
||||
digest := connectionDigest("disconnect", binding, playerID, generation)
|
||||
return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
||||
replay, _, err := beginConnectionMutation(ctx, tx, idempotencyKey, digest)
|
||||
if err != nil || replay {
|
||||
return err
|
||||
}
|
||||
current, connectedAt, disconnectedAt, _, err := lockConnectionLease(ctx, tx, binding, playerID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, ServerConnectionIdempotencyScope, idempotencyKey, result)
|
||||
return err
|
||||
if current != generation || !connectedAt.Valid || disconnectedAt.Valid || now.Before(connectedAt.Time) {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
var recorded uint64
|
||||
if err := tx.QueryRowContext(ctx, ServerConnectionDisconnectSQL, binding.MatchID, playerID, generation, now).Scan(&recorded); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
return err
|
||||
}
|
||||
return finishConnectionMutation(ctx, tx, idempotencyKey, recorded)
|
||||
})
|
||||
}
|
||||
|
||||
func validateConnectionMutation(db *sql.DB, binding domain.WorkloadBinding, playerID, key string, now time.Time) error {
|
||||
if db == nil || binding.AllocationID == "" || binding.MatchID == "" || binding.ServerID == "" || playerID == "" || len(key) < 16 || len(key) > 128 || now.IsZero() {
|
||||
return fmt.Errorf("invalid server connection receipt")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func connectionDigest(operation string, binding domain.WorkloadBinding, playerID string, generation uint64) [sha256.Size]byte {
|
||||
payload := []byte(operation + "\x00" + binding.AllocationID + "\x00" + binding.MatchID + "\x00" + binding.ServerID + "\x00" + playerID + "\x00")
|
||||
encoded := make([]byte, 8)
|
||||
binary.BigEndian.PutUint64(encoded, generation)
|
||||
return sha256.Sum256(append(payload, encoded...))
|
||||
}
|
||||
|
||||
func beginConnectionMutation(ctx context.Context, tx *sql.Tx, key string, digest [sha256.Size]byte) (bool, uint64, error) {
|
||||
inserted, err := tx.ExecContext(ctx, ServerConnectionIdempotencyInsertSQL, ServerConnectionIdempotencyScope, key, digest[:])
|
||||
if err != nil {
|
||||
return false, 0, err
|
||||
}
|
||||
changed, err := inserted.RowsAffected()
|
||||
if err != nil || changed != 0 {
|
||||
return false, 0, err
|
||||
}
|
||||
var prior, result []byte
|
||||
if err := tx.QueryRowContext(ctx, ServerConnectionIdempotencySelectSQL, ServerConnectionIdempotencyScope, key).Scan(&prior, &result); err != nil {
|
||||
return false, 0, err
|
||||
}
|
||||
if !bytes.Equal(prior, digest[:]) {
|
||||
return false, 0, domain.ErrConflict
|
||||
}
|
||||
var receipt connectionReceipt
|
||||
if err := json.Unmarshal(result, &receipt); err != nil || receipt.Generation == 0 {
|
||||
return false, 0, domain.ErrConflict
|
||||
}
|
||||
return true, receipt.Generation, nil
|
||||
}
|
||||
|
||||
func lockConnectionLease(ctx context.Context, tx *sql.Tx, binding domain.WorkloadBinding, playerID string) (uint64, sql.NullTime, sql.NullTime, time.Time, error) {
|
||||
var generation uint64
|
||||
var connectedAt, disconnectedAt sql.NullTime
|
||||
var assignmentExpiry time.Time
|
||||
err := tx.QueryRowContext(ctx, ServerConnectionLeaseSQL, binding.MatchID, binding.ServerID, binding.AllocationID, playerID).Scan(&generation, &connectedAt, &disconnectedAt, &assignmentExpiry)
|
||||
if err == sql.ErrNoRows {
|
||||
err = domain.ErrConflict
|
||||
}
|
||||
return generation, connectedAt, disconnectedAt, assignmentExpiry, err
|
||||
}
|
||||
|
||||
func finishConnectionMutation(ctx context.Context, tx *sql.Tx, key string, generation uint64) error {
|
||||
result, err := json.Marshal(connectionReceipt{Generation: generation})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, ServerConnectionIdempotencyScope, key, result)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -12,22 +12,24 @@ import (
|
||||
|
||||
func TestServerConnectionSQLBindsWorkloadParticipantAndLiveAssignment(t *testing.T) {
|
||||
for _, fragment := range []string{
|
||||
"connected_at = COALESCE", "mp.participation_active", "m.server_id = $2",
|
||||
"a.allocation_id = $3", "a.state = 'ALLOCATED'", "assn.player_id = mp.player_id",
|
||||
"assn.expires_at > $5", "RETURNING mp.connected_at",
|
||||
"connection_generation", "mp.disconnected_at", "mp.participation_active", "m.server_id = $2",
|
||||
"a.allocation_id = $3", "a.state = 'ALLOCATED'", "assn.player_id = mp.player_id", "FOR UPDATE OF mp",
|
||||
} {
|
||||
if !strings.Contains(ServerConnectionParticipantSQL, fragment) {
|
||||
t.Fatalf("connection SQL missing %q: %s", fragment, ServerConnectionParticipantSQL)
|
||||
if !strings.Contains(ServerConnectionLeaseSQL, fragment) {
|
||||
t.Fatalf("connection SQL missing %q: %s", fragment, ServerConnectionLeaseSQL)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordPlayerConnectedRejectsInvalidArguments(t *testing.T) {
|
||||
binding := domain.WorkloadBinding{AllocationID: "allocation-1", MatchID: "match-1", ServerID: "server-1"}
|
||||
if err := RecordPlayerConnected(context.Background(), (*sql.DB)(nil), binding, "player-1", "connection-key-123456", time.Unix(1000, 0)); err == nil {
|
||||
if _, err := ClaimPlayerConnection(context.Background(), (*sql.DB)(nil), binding, "player-1", 0, "connection-key-123456", time.Unix(1000, 0)); err == nil {
|
||||
t.Fatal("nil database accepted")
|
||||
}
|
||||
if err := RecordPlayerConnected(context.Background(), &sql.DB{}, domain.WorkloadBinding{}, "player-1", "connection-key-123456", time.Unix(1000, 0)); err == nil {
|
||||
if _, err := ClaimPlayerConnection(context.Background(), &sql.DB{}, domain.WorkloadBinding{}, "player-1", 0, "connection-key-123456", time.Unix(1000, 0)); err == nil {
|
||||
t.Fatal("empty workload binding accepted")
|
||||
}
|
||||
if err := RecordPlayerDisconnected(context.Background(), &sql.DB{}, binding, "player-1", 0, "disconnect-key-123456", time.Unix(1000, 0)); err == nil {
|
||||
t.Fatal("zero generation disconnect accepted")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user