From 9dc1cc2d6f322a1132cfbdc3191576412c50d86e Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:44:43 +0100 Subject: [PATCH] feat(multiplayer): refresh allocator ready servers --- multiplayer-next.md | 3 +- multiplayer-todo.md | 2 +- server/agones/allocation.go | 62 +++++++++++++++++++++++ server/agones/allocation_test.go | 24 +++++++++ server/cmd/allocator/main.go | 13 ++++- server/store/allocator_sql.go | 3 +- server/store/allocator_sql_test.go | 2 +- server/store/postgres_integration_test.go | 7 +++ 8 files changed, 111 insertions(+), 5 deletions(-) diff --git a/multiplayer-next.md b/multiplayer-next.md index efff359a..bae3ebb1 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -66,7 +66,8 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md). calls; allocator-facing roster publication now requires an allocated endpoint and verifies canonical join-authorisation signatures before exposing player rows; `cmd/allocator` - now polls these claims and drives provider allocation/reconciliation/binding; + now refreshes strictly labelled Ready GameServers into the durable registry, + then polls claims and drives provider allocation/reconciliation/binding; binding atomically advances every participant ticket to `ALLOCATING`, and durable post-provider recovery avoids a second allocation after a bind crash; unknown provider-outcome reconciliation and live Agones integration remain. diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 5bc17471..2b9406c6 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1213,7 +1213,7 @@ the local/CI/community transport, not a silent production fallback. | 8.27 `[D:8.26]` | **IN PROGRESS.** Go supervisor package provides local-safe Agones REST discovery, validates assigned address/port data, injects dynamic `SDR_LISTEN_PORT`/`SDR_IP`, performs explicit process-ready probing and Ready transition; direct mode bypasses Agones | `server/supervisor/` covers allocated/direct startup, invalid endpoint rejection, dynamic endpoint/Ready ordering and authenticated drain; allocated Godot now supplies a loopback readiness/drain control surface and `agones_sdk.gd` supplies sidecar Health/Ready/Shutdown/annotation REST operations; metadata watch, real Agones annotation/shutdown confirmation and emulator integration remain | | 8.28 `[D:8.6,8.27]` | **IN PROGRESS.** Supervisor separates explicit process-ready from Agones Ready and never scrapes stdout; allocated mode refuses to mark Ready without a configured readiness probe | `server/supervisor/` tests prove Ready follows the probe and direct mode remains functional; `server_control.gd`, `agones_sdk.gd` and process-level smokes prove loopback `/ready`, `/health`, bearer-protected `/drain`, sidecar-shaped Health/Ready calls and drain admission fencing; detached-container and Health-reclaim integration remain | | 8.29 `[D:8.26,8.27]` | **IN PROGRESS.** Supervisor discovers and validates the Agones endpoint, propagates the actual dynamic `--port`, and exports `SDR_LISTEN_PORT`/`SDR_IP` only for Hosted-SDR while preserving an isolated ENet path | `server/supervisor/` tests cover invalid address/port rejection, dynamic port argument/env propagation and SDR-vs-ENet separation; real Agones dynamic/passthrough mapping, POP/cert/firewall/NAT and multi-match fixture remain | -| 8.30 `[D:8.18,8.26,8.28,8.29]` | **IN PROGRESS.** Pure Go allocator filters Ready servers by region/build/protocol/transport, atomically claims one with idempotent allocation replay, verifies unanimous accepted-proposal/playlist/participant invariants before provider invocation, and now owns the assignment-publication boundary; PostgreSQL adds durable GameServer registration and compatible `SKIP LOCKED` claims with request-digest fencing, plus a leased `ALLOCATING`-match claim that derives the immutable compatibility tuple and fences bind/release by deterministic allocation ID; bind atomically attaches only a recorded provider allocation and advances every accepted participant ticket to `ALLOCATING`; `server/agones` submits and validates namespaced `GameServerAllocation` responses and dynamic endpoints; `server/allocator` reconciles provider success into durable state before exposing the endpoint; `cmd/allocator` is a signal-bound polling role that drives the lease → provider → durable-record → match-bind sequence and rebinds a recovered durable provider allocation without a second provider call | `server/domain/allocator.go`, `server/store/allocator_sql.go`, `server/store/allocation_match_sql.go`, `server/store/allocation_match_adapter.go`, `server/agones/allocation.go`, `server/allocator/service.go`, `server/allocator/worker.go`, `server/cmd/allocator`, `server/migrations/0004_allocator_registry.sql`, `0006_match_allocation_claims.sql` and tests cover deterministic compatible selection, exhaustion, conflicting/identical allocation replay, allocation-match lease recovery/bind fencing, atomic participant lifecycle transition, recovered-allocation binding without provider recall, immutable Fleet selector labels, provider ambiguity lease retention, unknown allocations, SQL claim ordering, invalid input, provider error/malformed response/IPv6 endpoint handling, accepted-proposal gating, durable-reconciliation failure isolation and assignment replay/conflict; `TestPostgreSQLAllocatorClaimReplayAndCapacityFence` and `TestPostgreSQLAllocationMatchClaimLeaseAndBindFence` cover the live database paths when the disposable database gate is run; unknown provider-outcome reconciliation, signed roster metadata, bounded cross-replica retry and live Agones integration remain | +| 8.30 `[D:8.18,8.26,8.28,8.29]` | **IN PROGRESS.** Pure Go allocator filters Ready servers by region/build/protocol/transport, atomically claims one with idempotent allocation replay, verifies unanimous accepted-proposal/playlist/participant invariants before provider invocation, and now owns the assignment-publication boundary; PostgreSQL adds durable GameServer registration and compatible `SKIP LOCKED` claims with request-digest fencing, plus a leased `ALLOCATING`-match claim that derives the immutable compatibility tuple and fences bind/release by deterministic allocation ID; bind atomically attaches only a recorded provider allocation and advances every accepted participant ticket to `ALLOCATING`; `server/agones` strictly projects Ready GameServers from Fleet compatibility labels, submits/validates namespaced `GameServerAllocation` responses and dynamic endpoints; `server/allocator` reconciles provider success into durable state before exposing the endpoint; `cmd/allocator` refreshes that Ready projection before driving the lease → provider → durable-record → match-bind sequence and rebinds a recovered durable provider allocation without a second provider call | `server/domain/allocator.go`, `server/store/allocator_sql.go`, `server/store/allocation_match_sql.go`, `server/store/allocation_match_adapter.go`, `server/agones/allocation.go`, `server/allocator/service.go`, `server/allocator/worker.go`, `server/cmd/allocator`, `server/migrations/0004_allocator_registry.sql`, `0006_match_allocation_claims.sql` and tests cover strict Ready-server projection, stale-projection protection for allocated rows, deterministic compatible selection, exhaustion, conflicting/identical allocation replay, allocation-match lease recovery/bind fencing, atomic participant lifecycle transition, recovered-allocation binding without provider recall, immutable Fleet selector labels, provider ambiguity lease retention, unknown allocations, SQL claim ordering, invalid input, provider error/malformed response/IPv6 endpoint handling, accepted-proposal gating, durable-reconciliation failure isolation and assignment replay/conflict; `TestPostgreSQLAllocatorClaimReplayAndCapacityFence` and `TestPostgreSQLAllocationMatchClaimLeaseAndBindFence` cover the live database paths when the disposable database gate is run; unknown provider-outcome reconciliation, signed roster metadata, bounded cross-replica retry and live Agones integration remain | | 8.31 `[D:8.9,8.30]` | **IN PROGRESS.** Pure Go assignment gate requires Allocated state, exact allocation ID/match/server/region/build/protocol/transport compatibility, non-empty hosted endpoint and verified manifest signature before exposure; allocator publication cannot expose Ready state; durable roster persistence now verifies each canonical join-authorisation signature before publishing player rows; allocator service gates roster publication on allocated state and endpoint presence | `server/domain/assignment.go`, `allocator.go`, `store/assignment_sql.go` plus adversarial fixtures cover early-connect, tampered signature/manifest, wrong compatibility, empty endpoint, unknown allocation, forged roster signature, valid signature, premature publication and post-publication mutation rejection; Agones metadata watch, hosted-address registration, production signer and client-ticket publication remain | | 8.32 `[D:8.2,8.26,8.30]` | **IN PROGRESS.** Provider-neutral FleetAutoscaler baseline preserves a two-process Ready buffer, caps warm capacity, and leaves Allocated scale-down independent of the Ready floor; Fleet image references remain digest-pinned for current/rollback pre-pull | `deploy/k8s/base/fleet-autoscaler.yaml` and manifest tests cover Fleet ownership, Buffer policy and floor/cap invariants; regional on-demand node pools/failure domains, pre-pull rollout, warm-allocation p95/p99 and N+1 certification remain | | 8.33 `[D:8.26,8.32]` | **IN PROGRESS.** Fleet scheduling now requires on-demand capacity and spreads Ready processes across zones with skew 1; the autoscaler preserves the two-process Ready floor | `deploy/k8s/base/fleet.yaml` and manifest tests reject interruptible placement and single-zone concentration structurally; regional node pools, forced node-loss testing and measured N+1 headroom remain | diff --git a/server/agones/allocation.go b/server/agones/allocation.go index 7c63870d..13027bb9 100644 --- a/server/agones/allocation.go +++ b/server/agones/allocation.go @@ -52,6 +52,68 @@ type allocationResponse struct { } `json:"status"` } +type gameServerListResponse struct { + Items []struct { + Metadata struct { + Name string `json:"name"` + Labels map[string]string `json:"labels"` + } `json:"metadata"` + Status struct { + State string `json:"state"` + } `json:"status"` + } `json:"items"` +} + +// ListReadyServers projects only Agones Ready GameServers into the durable +// allocator registry. Compatibility fields must be present as Fleet labels; +// malformed Ready objects fail closed instead of creating selectable capacity. +func (c Client) ListReadyServers(ctx context.Context) ([]domain.ReadyServer, error) { + if c.HTTP == nil { + c.HTTP = http.DefaultClient + } + base, err := c.endpoint() + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/apis/agones.dev/v1/namespaces/"+url.PathEscape(c.Namespace)+"/gameservers", nil) + if err != nil { + return nil, err + } + response, err := c.HTTP.Do(req) + if err != nil { + return nil, err + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + return nil, fmt.Errorf("Agones GameServer list returned %s", response.Status) + } + var decoded gameServerListResponse + if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&decoded); err != nil { + return nil, fmt.Errorf("decode Agones GameServer list: %w", err) + } + ready := make([]domain.ReadyServer, 0, len(decoded.Items)) + for _, item := range decoded.Items { + if item.Status.State != "Ready" { + continue + } + server, err := readyServerFromGameServer(item.Metadata.Name, item.Metadata.Labels) + if err != nil { + return nil, err + } + ready = append(ready, server) + } + return ready, nil +} + +func readyServerFromGameServer(name string, labels map[string]string) (domain.ReadyServer, error) { + protocol, err := strconv.Atoi(labels["cosmic-clash.io/protocol"]) + server := domain.ReadyServer{ServerID: name, Region: labels["cosmic-clash.io/region"], Build: labels["cosmic-clash.io/build"], Protocol: protocol, Transport: labels["cosmic-clash.io/transport"], State: domain.ServerReady} + if err != nil || server.ServerID == "" || (server.Region != "EU" && server.Region != "NA") || server.Build == "" || server.Protocol < 1 || (server.Transport != "enet" && server.Transport != "steam_sdr") { + return domain.ReadyServer{}, fmt.Errorf("invalid Ready GameServer compatibility labels") + } + return server, nil +} + func (c Client) Allocate(ctx context.Context, request domain.AllocationRequest, labels map[string]string, now time.Time) (AllocatedServer, error) { if c.HTTP == nil { c.HTTP = http.DefaultClient diff --git a/server/agones/allocation_test.go b/server/agones/allocation_test.go index e08125a4..59126e83 100644 --- a/server/agones/allocation_test.go +++ b/server/agones/allocation_test.go @@ -70,3 +70,27 @@ func TestAllocateRejectsUnsafeConfigurationAndProviderFailure(t *testing.T) { t.Fatalf("provider failure err=%v", err) } } + +func TestListReadyServersProjectsOnlyStrictReadyFleetMembers(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.URL.Path != "/apis/agones.dev/v1/namespaces/games/gameservers" { + t.Fatalf("request=%s %s", r.Method, r.URL.Path) + } + _, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"ready-a","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"}},"status":{"state":"Ready"}},{"metadata":{"name":"allocated-a","labels":{}},"status":{"state":"Allocated"}}]}`)) + })) + defer server.Close() + ready, err := (Client{BaseURL: server.URL, Namespace: "games", HTTP: server.Client()}).ListReadyServers(context.Background()) + if err != nil || len(ready) != 1 || ready[0] != (domain.ReadyServer{ServerID: "ready-a", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}) { + t.Fatalf("ready=%+v err=%v", ready, err) + } +} + +func TestListReadyServersFailsClosedOnInvalidReadyCompatibility(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"ready-a","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"bad","cosmic-clash.io/transport":"enet"}},"status":{"state":"Ready"}}]}`)) + })) + defer server.Close() + if _, err := (Client{BaseURL: server.URL, Namespace: "games", HTTP: server.Client()}).ListReadyServers(context.Background()); err == nil { + t.Fatal("invalid Ready GameServer accepted") + } +} diff --git a/server/cmd/allocator/main.go b/server/cmd/allocator/main.go index 3d20e7a9..11298bea 100644 --- a/server/cmd/allocator/main.go +++ b/server/cmd/allocator/main.go @@ -45,10 +45,11 @@ func main() { fatalf("apply migrations: %v", err) } now := func() time.Time { return time.Now().UTC() } + client := agones.Client{BaseURL: *agonesURL, Namespace: *namespace} worker := allocator.Worker{ Claims: store.AllocatingMatchClaims{DB: db, Transport: *transport}, Service: allocator.Service{ - Provider: agones.Client{BaseURL: *agonesURL, Namespace: *namespace}, + Provider: client, Durable: store.AllocationRegistry{DB: db}, Now: now, }, @@ -59,6 +60,16 @@ func main() { ticker := time.NewTicker(*interval) defer ticker.Stop() for { + servers, err := client.ListReadyServers(ctx) + if err != nil && ctx.Err() == nil { + log.Printf("allocator: list Ready GameServers: %v", err) + } else { + for _, server := range servers { + if err := store.RegisterReadyServer(ctx, db, server, now()); err != nil && ctx.Err() == nil { + log.Printf("allocator: register Ready GameServer %s: %v", server.ServerID, err) + } + } + } if _, err := worker.RunOnce(ctx); err != nil && ctx.Err() == nil { log.Printf("allocator: run once: %v", err) } diff --git a/server/store/allocator_sql.go b/server/store/allocator_sql.go index a4a93696..46af3281 100644 --- a/server/store/allocator_sql.go +++ b/server/store/allocator_sql.go @@ -16,7 +16,8 @@ const RegisterReadyServerSQL = `INSERT INTO game_servers VALUES ($1, $2, $3, $4, $5, 'READY', $6) ON CONFLICT (server_id) DO UPDATE SET region = EXCLUDED.region, build = EXCLUDED.build, protocol_version = EXCLUDED.protocol_version, - transport = EXCLUDED.transport, state = 'READY', updated_at = EXCLUDED.updated_at` + transport = EXCLUDED.transport, updated_at = EXCLUDED.updated_at +WHERE game_servers.state = 'READY'` const ClaimReadyServerSQL = `UPDATE game_servers SET state = 'ALLOCATED', updated_at = $5 WHERE server_id = ( diff --git a/server/store/allocator_sql_test.go b/server/store/allocator_sql_test.go index da7f54e1..9860756b 100644 --- a/server/store/allocator_sql_test.go +++ b/server/store/allocator_sql_test.go @@ -9,7 +9,7 @@ import ( func TestAllocatorSQLClaimsAndAuditsCompatibleReadyServers(t *testing.T) { for query, fragments := range map[string][]string{ - RegisterReadyServerSQL: {"game_servers", "ON CONFLICT", "state = 'READY'"}, + RegisterReadyServerSQL: {"game_servers", "ON CONFLICT", "WHERE game_servers.state = 'READY'"}, ClaimReadyServerSQL: {"state = 'READY'", "region = $1", "protocol_version = $3", "FOR UPDATE SKIP LOCKED", "ORDER BY server_id"}, InsertAllocationSQL: {"allocations", "request_digest", "state", "ALLOCATED"}, ProviderServerClaimSQL: {"state = 'READY'", "region = $2", "protocol_version = $4", "RETURNING"}, diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 2a5699ee..9b2f3fdf 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -71,6 +71,13 @@ func TestPostgreSQLAllocatorClaimReplayAndCapacityFence(t *testing.T) { if allocation.ServerID != "allocator-server-a" || allocation.State != domain.ServerAllocated { t.Fatalf("allocation=%+v", allocation) } + if err := RegisterReadyServer(ctx, db, servers[1], now.Add(time.Second)); err != nil { + t.Fatalf("stale Ready projection: %v", err) + } + var lifecycle string + if err := db.QueryRowContext(ctx, `SELECT state FROM game_servers WHERE server_id = 'allocator-server-a'`).Scan(&lifecycle); err != nil || lifecycle != "ALLOCATED" { + t.Fatalf("stale Ready projection reopened allocation state=%q err=%v", lifecycle, err) + } replay, err := ClaimAllocation(ctx, db, request, now.Add(time.Second)) if err != nil || replay.ServerID != allocation.ServerID || !replay.AllocatedAt.Equal(allocation.AllocatedAt) { t.Fatalf("replay=%+v err=%v", replay, err)