mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
feat(multiplayer): run leased allocator worker
This commit is contained in:
Josh Creek
+8
-5
@@ -65,8 +65,9 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md).
|
||||
persisted compatibility tuple, to fence allocator replicas before provider
|
||||
calls; allocator-facing roster
|
||||
publication now requires an allocated endpoint and verifies canonical
|
||||
join-authorisation signatures before exposing player rows; allocator runtime
|
||||
wiring and live Agones integration remain.
|
||||
join-authorisation signatures before exposing player rows; `cmd/allocator`
|
||||
now polls these claims and drives provider allocation/reconciliation/binding;
|
||||
interrupted-provider reconciliation and live Agones integration remain.
|
||||
- [ ] **IN PROGRESS:** Run the Go control plane against PostgreSQL/Redis with
|
||||
independently runnable API, matcher, allocator and maintenance roles. The
|
||||
`cmd/control-plane` API role now opens PostgreSQL, applies migrations, wires
|
||||
@@ -74,10 +75,12 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md).
|
||||
down gracefully; optional `--redis-addr` publishes queue mutations to a
|
||||
TTL-bound best-effort candidate projection without making Redis authoritative;
|
||||
a runnable casual `cmd/matcher` role now polls PostgreSQL and delegates
|
||||
proposal claims to the durable transaction; `cmd/maintenance` now runs
|
||||
proposal claims to the durable transaction; `cmd/allocator` now polls leased
|
||||
allocating matches and invokes Agones through the durable allocator boundary;
|
||||
`cmd/maintenance` now runs
|
||||
bounded ranked-season rollover batches with signal-bound shutdown;
|
||||
provider-backed allocation, allocator runtime wiring, Redis failover and live
|
||||
service checks remain.
|
||||
interrupted-provider reconciliation, Redis failover and live service checks
|
||||
remain.
|
||||
- [ ] **IN PROGRESS:** Define assignment compatibility and opt-in `ServerConfig`
|
||||
flags whose defaults reproduce the community-server path. Allocation manifest
|
||||
validation now covers client build and future expiry; allocated servers now
|
||||
|
||||
+1
-1
@@ -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; `server/agones` submits and validates namespaced `GameServerAllocation` responses and dynamic endpoints; `server/allocator` reconciles provider success into durable state before exposing the endpoint | `server/domain/allocator.go`, `server/store/allocator_sql.go`, `server/store/allocation_match_sql.go`, `server/agones/allocation.go`, `server/allocator/service.go`, `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, 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; allocator runtime wiring, 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; `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 | `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, 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; interrupted-provider 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 |
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
package allocator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
||||
)
|
||||
|
||||
// MatchClaimSource is the durable allocator work queue. Implementations must
|
||||
// lease a match before returning it and fence binding by allocation ID.
|
||||
type MatchClaimSource interface {
|
||||
ClaimAllocatingMatch(context.Context, time.Time) (domain.AllocationRequest, bool, error)
|
||||
BindAllocatedMatch(context.Context, domain.Allocation) error
|
||||
}
|
||||
|
||||
// Worker consumes one leased match at a time. Provider failures deliberately
|
||||
// retain the lease: an HTTP/provider failure can be ambiguous after an external
|
||||
// allocation, so releasing it could allocate two GameServers for one match.
|
||||
type Worker struct {
|
||||
Claims MatchClaimSource
|
||||
Service Service
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// RunOnce returns whether it found a claimed match. It never exposes an
|
||||
// endpoint itself; Service first records the provider allocation durably and
|
||||
// BindAllocatedMatch then attaches that already-recorded allocation to the
|
||||
// fenced match claim.
|
||||
func (w Worker) RunOnce(ctx context.Context) (bool, error) {
|
||||
if w.Claims == nil || w.Now == nil {
|
||||
return false, errNotConfigured
|
||||
}
|
||||
request, found, err := w.Claims.ClaimAllocatingMatch(ctx, w.Now())
|
||||
if err != nil || !found {
|
||||
return found, err
|
||||
}
|
||||
result, err := w.Service.Allocate(ctx, request, AllocationLabels(request))
|
||||
if err != nil {
|
||||
return true, fmt.Errorf("allocate claimed match %s: %w", request.MatchID, err)
|
||||
}
|
||||
if err := w.Claims.BindAllocatedMatch(ctx, result.Allocation); err != nil {
|
||||
return true, fmt.Errorf("bind allocated match %s: %w", request.MatchID, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// AllocationLabels are the compatibility selectors shared with the Fleet
|
||||
// template. They are derived only from the durable match plan, never client
|
||||
// input or mutable worker configuration.
|
||||
func AllocationLabels(request domain.AllocationRequest) map[string]string {
|
||||
return map[string]string{
|
||||
"cosmic-clash.io/region": request.Region,
|
||||
"cosmic-clash.io/build": request.Build,
|
||||
"cosmic-clash.io/protocol": strconv.Itoa(request.Protocol),
|
||||
"cosmic-clash.io/transport": request.Transport,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package allocator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/cosmic-clash/cosmic-clash/server/agones"
|
||||
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
||||
)
|
||||
|
||||
type matchClaimSpy struct {
|
||||
request domain.AllocationRequest
|
||||
found bool
|
||||
err error
|
||||
bound domain.Allocation
|
||||
bindErr error
|
||||
}
|
||||
|
||||
func (s *matchClaimSpy) ClaimAllocatingMatch(_ context.Context, _ time.Time) (domain.AllocationRequest, bool, error) {
|
||||
return s.request, s.found, s.err
|
||||
}
|
||||
|
||||
func (s *matchClaimSpy) BindAllocatedMatch(_ context.Context, allocation domain.Allocation) error {
|
||||
s.bound = allocation
|
||||
return s.bindErr
|
||||
}
|
||||
|
||||
func TestWorkerClaimsAllocatesAndBindsDurably(t *testing.T) {
|
||||
request := domain.AllocationRequest{AllocationID: "allocation-1", MatchID: "match-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
|
||||
claims := &matchClaimSpy{request: request, found: true}
|
||||
provider := &providerSpy{result: agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: request.AllocationID, MatchID: request.MatchID, ServerID: "server-1", Region: request.Region, Build: request.Build, Protocol: request.Protocol, Transport: request.Transport, State: domain.ServerAllocated}, Endpoint: "127.0.0.1:7777"}}
|
||||
durable := &durableSpy{}
|
||||
worker := Worker{Claims: claims, Service: Service{Provider: provider, Durable: durable, Now: func() time.Time { return time.Unix(1_000, 0) }}, Now: func() time.Time { return time.Unix(1_000, 0) }}
|
||||
processed, err := worker.RunOnce(context.Background())
|
||||
if err != nil || !processed || provider.calls != 1 || durable.calls != 1 || claims.bound.ServerID != "server-1" {
|
||||
t.Fatalf("processed=%t err=%v provider=%d durable=%d bound=%+v", processed, err, provider.calls, durable.calls, claims.bound)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkerRetainsClaimWhenProviderOutcomeIsAmbiguous(t *testing.T) {
|
||||
request := domain.AllocationRequest{AllocationID: "allocation-1", MatchID: "match-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
|
||||
claims := &matchClaimSpy{request: request, found: true}
|
||||
provider := &providerSpy{err: errors.New("provider timeout")}
|
||||
worker := Worker{Claims: claims, Service: Service{Provider: provider, Durable: &durableSpy{}, Now: func() time.Time { return time.Unix(1_000, 0) }}, Now: func() time.Time { return time.Unix(1_000, 0) }}
|
||||
processed, err := worker.RunOnce(context.Background())
|
||||
if err == nil || !processed || claims.bound != (domain.Allocation{}) {
|
||||
t.Fatalf("processed=%t err=%v bound=%+v", processed, err, claims.bound)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkerDoesNothingWhenNoDurableMatchIsAvailable(t *testing.T) {
|
||||
claims := &matchClaimSpy{}
|
||||
worker := Worker{Claims: claims, Now: func() time.Time { return time.Unix(1_000, 0) }}
|
||||
processed, err := worker.RunOnce(context.Background())
|
||||
if err != nil || processed {
|
||||
t.Fatalf("processed=%t err=%v", processed, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllocationLabelsMirrorFleetCompatibilityTuple(t *testing.T) {
|
||||
got := AllocationLabels(domain.AllocationRequest{Region: "NA", Build: "build-4", Protocol: 12, Transport: "steam_sdr"})
|
||||
want := map[string]string{"cosmic-clash.io/region": "NA", "cosmic-clash.io/build": "build-4", "cosmic-clash.io/protocol": "12", "cosmic-clash.io/transport": "steam_sdr"}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("labels=%v want=%v", got, want)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"flag"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/cosmic-clash/cosmic-clash/server/agones"
|
||||
"github.com/cosmic-clash/cosmic-clash/server/allocator"
|
||||
"github.com/cosmic-clash/cosmic-clash/server/migrations"
|
||||
"github.com/cosmic-clash/cosmic-clash/server/store"
|
||||
_ "github.com/jackc/pgx/v5/stdlib"
|
||||
)
|
||||
|
||||
func main() {
|
||||
dsn := flag.String("dsn", os.Getenv("COSMIC_CLASH_POSTGRES_DSN"), "PostgreSQL connection string")
|
||||
migrationDir := flag.String("migrations", "migrations", "directory containing numbered SQL migrations")
|
||||
agonesURL := flag.String("agones-url", os.Getenv("COSMIC_CLASH_AGONES_URL"), "Agones allocation API base URL")
|
||||
namespace := flag.String("agones-namespace", envOrDefault("COSMIC_CLASH_AGONES_NAMESPACE", "default"), "Agones namespace")
|
||||
transport := flag.String("transport", envOrDefault("COSMIC_CLASH_TRANSPORT", "enet"), "game transport: enet or steam_sdr")
|
||||
interval := flag.Duration("interval", time.Second, "allocation poll interval")
|
||||
flag.Parse()
|
||||
if *dsn == "" || *agonesURL == "" {
|
||||
fatalf("--dsn/COSMIC_CLASH_POSTGRES_DSN and --agones-url/COSMIC_CLASH_AGONES_URL are required")
|
||||
}
|
||||
if (*transport != "enet" && *transport != "steam_sdr") || *interval <= 0 {
|
||||
fatalf("--transport must be enet or steam_sdr and --interval must be positive")
|
||||
}
|
||||
db, err := sql.Open("pgx", *dsn)
|
||||
if err != nil {
|
||||
fatalf("open PostgreSQL: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
startupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
if err := db.PingContext(startupCtx); err != nil {
|
||||
fatalf("ping PostgreSQL: %v", err)
|
||||
}
|
||||
if err := migrations.Apply(startupCtx, db, *migrationDir); err != nil {
|
||||
fatalf("apply migrations: %v", err)
|
||||
}
|
||||
now := func() time.Time { return time.Now().UTC() }
|
||||
worker := allocator.Worker{
|
||||
Claims: store.AllocatingMatchClaims{DB: db, Transport: *transport},
|
||||
Service: allocator.Service{
|
||||
Provider: agones.Client{BaseURL: *agonesURL, Namespace: *namespace},
|
||||
Durable: store.AllocationRegistry{DB: db},
|
||||
Now: now,
|
||||
},
|
||||
Now: now,
|
||||
}
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
ticker := time.NewTicker(*interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if _, err := worker.RunOnce(ctx); err != nil && ctx.Err() == nil {
|
||||
log.Printf("allocator: run once: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func envOrDefault(name, fallback string) string {
|
||||
if value := os.Getenv(name); value != "" {
|
||||
return value
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
func fatalf(format string, args ...any) {
|
||||
log.Printf("allocator: "+format, args...)
|
||||
os.Exit(1)
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"time"
|
||||
|
||||
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
||||
)
|
||||
|
||||
// AllocatingMatchClaims adapts the PostgreSQL lease boundary for allocator
|
||||
// workers without making the allocator package depend on the store package.
|
||||
type AllocatingMatchClaims struct {
|
||||
DB *sql.DB
|
||||
Transport string
|
||||
}
|
||||
|
||||
func (s AllocatingMatchClaims) ClaimAllocatingMatch(ctx context.Context, now time.Time) (domain.AllocationRequest, bool, error) {
|
||||
item, found, err := ClaimAllocatingMatch(ctx, s.DB, s.Transport, now)
|
||||
return item.Request, found, err
|
||||
}
|
||||
|
||||
func (s AllocatingMatchClaims) BindAllocatedMatch(ctx context.Context, allocation domain.Allocation) error {
|
||||
return BindAllocatedMatch(ctx, s.DB, allocation)
|
||||
}
|
||||
|
||||
// AllocationRegistry adapts provider-allocation reconciliation for allocator
|
||||
// workers. A successful provider response is not publishable until this store
|
||||
// boundary records the same compatibility tuple and GameServer identity.
|
||||
type AllocationRegistry struct{ DB *sql.DB }
|
||||
|
||||
func (s AllocationRegistry) RecordProviderAllocation(ctx context.Context, allocation domain.Allocation, now time.Time) (domain.Allocation, error) {
|
||||
return RecordProviderAllocation(ctx, s.DB, allocation, now)
|
||||
}
|
||||
Reference in New Issue
Block a user