diff --git a/multiplayer-next.md b/multiplayer-next.md index 36d41120..e5e71b10 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -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/.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 diff --git a/server/api/service.go b/server/api/service.go index 08e82459..1dc1e72e 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -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" { diff --git a/server/api/service_test.go b/server/api/service_test.go index 8a082500..075c3b77 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -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) } } diff --git a/server/api/store_adapters.go b/server/api/store_adapters.go index 68405f05..3914fa03 100644 --- a/server/api/store_adapters.go +++ b/server/api/store_adapters.go @@ -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 { diff --git a/server/contracts/v1/openapi.json b/server/contracts/v1/openapi.json index efc0fb9e..beddb30c 100644 --- a/server/contracts/v1/openapi.json +++ b/server/contracts/v1/openapi.json @@ -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"]}}} } diff --git a/server/migrations/0011_connection_leases.sql b/server/migrations/0011_connection_leases.sql new file mode 100644 index 00000000..1965c5d8 --- /dev/null +++ b/server/migrations/0011_connection_leases.sql @@ -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; diff --git a/server/migrations/down/0011_connection_leases.sql b/server/migrations/down/0011_connection_leases.sql new file mode 100644 index 00000000..fc333250 --- /dev/null +++ b/server/migrations/down/0011_connection_leases.sql @@ -0,0 +1,3 @@ +ALTER TABLE match_participants + DROP CONSTRAINT IF EXISTS match_participants_connection_lease, + DROP COLUMN IF EXISTS disconnected_at; diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index cb4a991b..9afe2c8d 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -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) { diff --git a/server/store/server_connection_sql.go b/server/store/server_connection_sql.go index cfcf0626..a0dad604 100644 --- a/server/store/server_connection_sql.go +++ b/server/store/server_connection_sql.go @@ -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 +} diff --git a/server/store/server_connection_sql_test.go b/server/store/server_connection_sql_test.go index bc5e0690..a71e0c74 100644 --- a/server/store/server_connection_sql_test.go +++ b/server/store/server_connection_sql_test.go @@ -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") + } }