From 181a928c874f424343e8a199666be079583eaeef Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 18:25:25 +0100 Subject: [PATCH] feat(multiplayer): add regional allocation budget --- multiplayer-next.md | 2 +- server/allocator/budget.go | 49 ++++++++++++++++++++++ server/allocator/budget_test.go | 72 +++++++++++++++++++++++++++++++++ server/allocator/service.go | 10 +++++ server/cmd/allocator/main.go | 14 +++++++ 5 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 server/allocator/budget.go create mode 100644 server/allocator/budget_test.go diff --git a/multiplayer-next.md b/multiplayer-next.md index 02dbdc78..f20379e3 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1251,7 +1251,7 @@ the local/CI/community transport, not a silent production fallback. | 8.49 `[D:8.25,8.26,8.28,8.29,8.30,8.31,8.35,8.36]` | **IN PROGRESS.** `scripts/verify_kind_agones.sh` creates a disposable kind cluster, installs pinned Agones, loads the real `game-server` image, applies the Fleet in an explicitly separate Agones-only supervisor/UDP readiness mode, and verifies readiness plus allocation of a dynamic UDP endpoint; `.github/workflows/agones-integration.yml` runs it for infrastructure changes and on demand | The cloud-free runner is committed and fails clearly when Docker/kind/Helm are unavailable. CI/live evidence for production control-plane registration, roster/no-show, both readiness stages, races, multi-match node, result-pending reconciliation, drain, and rollback remains open | | 8.50 `[D:8.25,8.37,8.43,8.49]` | Network/chaos suite: 100 ms RTT, jitter/loss, client/API/matcher restart, game-pod death, node drain, Redis failover and control-plane loss | System recovers to a defined state; infrastructure-caused cases cannot penalise affected players | | 8.51 `[D:8.17,8.18,8.30,8.31,8.45]` | **IN PROGRESS.** The opt-in `make verify-multiplayer-load` gate drives 10,000 real HTTP queue-create requests through the service with 256 in flight and records p95/p99; the handler and in-process ownership boundary are exercised without weakening normal tests | Local API load passes at p95 <250 ms in normal and race runs; PostgreSQL saturation, >=100 proposals/s, durable matcher fencing under load, forecast launch concurrency x2, and replica scaling remain live infrastructure gates | -| 8.52 `[D:8.32,8.34,8.45,8.51]` | Per-region cost model from measured density, warm capacity, bandwidth, DB/Redis and telemetry; add budgets and allocation quotas | Cost per completed match and forecast monthly bands are recorded; a denial-of-wallet test triggers limits/alerts before budget breach | +| 8.52 `[D:8.32,8.34,8.45,8.51]` | **IN PROGRESS.** Allocator now supports an opt-in, per-replica fixed-window allocation quota per EU/NA region (`--allocation-quota` / `--allocation-quota-window`), checked before any provider call and safe under concurrent attempts | Normal/race/vet tests cover quota exhaustion, window reset, region isolation, invalid input, and atomic concurrent consumption; measured regional cost model, shared/global quota, budget alerts, and denial-of-wallet production rehearsal remain | | 8.53 `[D:7.8,8.13,8.38,8.45,8.46,8.48,8.49,8.50,8.51,8.52]` | Progressive release: development → internal → casual canary → casual → provisional ranked → ranked | Each promotion requires SLO/security/cost gates, rollback rehearsal, EU+NA playtests and unchanged legacy gates; rollback criteria and owner are explicit | Implementation invariants for every task above: diff --git a/server/allocator/budget.go b/server/allocator/budget.go new file mode 100644 index 00000000..34dcc0bd --- /dev/null +++ b/server/allocator/budget.go @@ -0,0 +1,49 @@ +package allocator + +import ( + "fmt" + "sync" + "time" +) + +// ErrAllocationBudgetExceeded is deliberately generic: callers should not +// learn quota internals, and the allocator can safely retry the leased match +// after the current window expires. +var ErrAllocationBudgetExceeded = fmt.Errorf("allocation budget exceeded") + +// FixedWindowBudget is a process-local denial-of-wallet guard. It limits the +// number of provider allocation attempts per region in a time window. The +// production deployment must use the same policy behind a shared durable +// counter for a global quota; this type prevents one allocator replica from +// spending without bound and is useful in tests and single-replica setups. +type FixedWindowBudget struct { + mu sync.Mutex + limit int + window time.Duration + windowStart time.Time + counts map[string]int +} + +func NewFixedWindowBudget(limit int, window time.Duration) (*FixedWindowBudget, error) { + if limit < 1 || window <= 0 { + return nil, fmt.Errorf("invalid allocation budget") + } + return &FixedWindowBudget{limit: limit, window: window, counts: make(map[string]int)}, nil +} + +func (b *FixedWindowBudget) Allow(region string, now time.Time) error { + if b == nil || (region != "EU" && region != "NA") || now.IsZero() { + return fmt.Errorf("invalid allocation budget request") + } + b.mu.Lock() + defer b.mu.Unlock() + if b.windowStart.IsZero() || !now.Before(b.windowStart.Add(b.window)) { + b.windowStart = now + b.counts = make(map[string]int) + } + if b.counts[region] >= b.limit { + return ErrAllocationBudgetExceeded + } + b.counts[region]++ + return nil +} diff --git a/server/allocator/budget_test.go b/server/allocator/budget_test.go new file mode 100644 index 00000000..2a822ff7 --- /dev/null +++ b/server/allocator/budget_test.go @@ -0,0 +1,72 @@ +package allocator + +import ( + "errors" + "sync" + "testing" + "time" +) + +func TestFixedWindowBudgetLimitsEachRegionAndResets(t *testing.T) { + now := time.Unix(1000, 0).UTC() + budget, err := NewFixedWindowBudget(2, time.Minute) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if err := budget.Allow("EU", now); err != nil { + t.Fatalf("EU attempt %d: %v", i, err) + } + } + if err := budget.Allow("EU", now); !errors.Is(err, ErrAllocationBudgetExceeded) { + t.Fatalf("third EU attempt = %v, want budget error", err) + } + if err := budget.Allow("NA", now); err != nil { + t.Fatalf("NA should have an independent budget: %v", err) + } + if err := budget.Allow("EU", now.Add(time.Minute)); err != nil { + t.Fatalf("EU after window: %v", err) + } +} + +func TestFixedWindowBudgetIsAtomicUnderConcurrentAttempts(t *testing.T) { + budget, err := NewFixedWindowBudget(7, time.Minute) + if err != nil { + t.Fatal(err) + } + now := time.Unix(1000, 0).UTC() + var wg sync.WaitGroup + var mu sync.Mutex + allowed := 0 + for i := 0; i < 64; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if budget.Allow("EU", now) == nil { + mu.Lock() + allowed++ + mu.Unlock() + } + }() + } + wg.Wait() + if allowed != 7 { + t.Fatalf("allowed=%d, want exactly 7", allowed) + } +} + +func TestFixedWindowBudgetRejectsInvalidConfigurationAndInput(t *testing.T) { + if _, err := NewFixedWindowBudget(0, time.Minute); err == nil { + t.Fatal("zero limit accepted") + } + if _, err := NewFixedWindowBudget(1, 0); err == nil { + t.Fatal("zero window accepted") + } + budget, _ := NewFixedWindowBudget(1, time.Minute) + if err := budget.Allow("APAC", time.Unix(1000, 0)); err == nil { + t.Fatal("unknown region accepted") + } + if err := budget.Allow("EU", time.Time{}); err == nil { + t.Fatal("zero time accepted") + } +} diff --git a/server/allocator/service.go b/server/allocator/service.go index a81f7936..aa3a1d8b 100644 --- a/server/allocator/service.go +++ b/server/allocator/service.go @@ -26,10 +26,15 @@ type RosterPublisher interface { PublishRoster(context.Context, domain.Assignment, []domain.SignedJoinAuthorisation, func([]byte, []byte) bool) error } +type AllocationBudget interface { + Allow(region string, now time.Time) error +} + type Service struct { Provider Provider Durable Durable Roster RosterPublisher + Budget AllocationBudget Now func() time.Time } @@ -78,6 +83,11 @@ func (s Service) Allocate(ctx context.Context, request domain.AllocationRequest, return agones.AllocatedServer{}, errNotConfigured } now := s.Now() + if s.Budget != nil { + if err := s.Budget.Allow(request.Region, now); err != nil { + return agones.AllocatedServer{}, err + } + } result, err := s.Provider.Allocate(ctx, request, labels, now) if err != nil { return agones.AllocatedServer{}, err diff --git a/server/cmd/allocator/main.go b/server/cmd/allocator/main.go index ac93ab1a..380be26b 100644 --- a/server/cmd/allocator/main.go +++ b/server/cmd/allocator/main.go @@ -25,6 +25,8 @@ func main() { transport := flag.String("transport", envOrDefault("COSMIC_CLASH_TRANSPORT", "enet"), "game transport: enet or steam_sdr") interval := flag.Duration("interval", time.Second, "allocation poll interval") workloadSecret := flag.String("workload-secret", os.Getenv("COSMIC_CLASH_WORKLOAD_SECRET"), "HMAC secret for control-plane-issued workload tokens (see workload/signed_token.go); must match cmd/control-plane's own --workload-secret. Unset skips minting a cosmic-clash.io/workload-token annotation entirely") + allocationQuota := flag.Int("allocation-quota", 0, "optional per-replica allocation attempts per region per quota window; zero disables this local guard") + allocationQuotaWindow := flag.Duration("allocation-quota-window", time.Minute, "window for --allocation-quota") flag.Parse() if *dsn == "" || *agonesURL == "" { fatalf("--dsn/COSMIC_CLASH_POSTGRES_DSN and --agones-url/COSMIC_CLASH_AGONES_URL are required") @@ -32,6 +34,9 @@ func main() { if (*transport != "enet" && *transport != "steam_sdr") || *interval <= 0 { fatalf("--transport must be enet or steam_sdr and --interval must be positive") } + if *allocationQuota < 0 || *allocationQuotaWindow <= 0 { + fatalf("--allocation-quota must be non-negative and --allocation-quota-window must be positive") + } db, err := sql.Open("pgx", *dsn) if err != nil { fatalf("open PostgreSQL: %v", err) @@ -49,12 +54,21 @@ func main() { log.Printf("allocator: warning: --workload-secret / COSMIC_CLASH_WORKLOAD_SECRET is unset; allocated GameServers will receive no cosmic-clash.io/workload-token annotation, and control-plane registration will fail unless a --workload-token-path is separately configured on the supervisor") } now := func() time.Time { return time.Now().UTC() } + var budget allocator.AllocationBudget + if *allocationQuota > 0 { + budget, err = allocator.NewFixedWindowBudget(*allocationQuota, *allocationQuotaWindow) + if err != nil { + fatalf("allocation quota: %v", err) + } + log.Printf("allocator: enabled per-replica regional allocation quota=%d window=%s", *allocationQuota, *allocationQuotaWindow) + } client := agones.Client{BaseURL: *agonesURL, Namespace: *namespace, WorkloadSecret: []byte(*workloadSecret)} worker := allocator.Worker{ Claims: store.AllocatingMatchClaims{DB: db, Transport: *transport}, Service: allocator.Service{ Provider: client, Durable: store.AllocationRegistry{DB: db}, + Budget: budget, Now: now, }, Now: now,