From 72e8d276337ee49018ed20800a958d68b0f9e254 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 18:38:06 +0100 Subject: [PATCH] feat(multiplayer): add shared allocation quota --- multiplayer-next.md | 2 +- server/allocator/service.go | 15 ++++ server/allocator/service_test.go | 35 +++++++++ server/cmd/allocator/main.go | 1 + server/migrations/0007_allocation_quotas.sql | 10 +++ .../down/0007_allocation_quotas.sql | 1 + server/migrations/test_migration.py | 6 ++ server/store/allocation_quota_sql.go | 78 +++++++++++++++++++ server/store/allocator_sql.go | 3 + server/store/allocator_sql_test.go | 23 ++++++ server/store/postgres_integration_test.go | 41 +++++++++- 11 files changed, 211 insertions(+), 4 deletions(-) create mode 100644 server/migrations/0007_allocation_quotas.sql create mode 100644 server/migrations/down/0007_allocation_quotas.sql create mode 100644 server/store/allocation_quota_sql.go diff --git a/multiplayer-next.md b/multiplayer-next.md index f3ffe05e..807a5f3f 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]` | **IN PROGRESS.** `make verify-chaos-recovery` provides a disposable PostgreSQL + real testkit API + real maintenance flow: it restarts the API, injects a stale allocation, and verifies no-penalty requeue plus a durable participant-targeted lifecycle event | The API-restart/stalled-allocation slice is implemented and documented; 100 ms RTT/jitter/loss, matcher/client restart, game-pod death, node drain, Redis failover, control-plane loss, and live chaos evidence remain | | 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, plus 100 concurrent proposal formations through the real matcher/domain path; 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, and the matcher forms 100 unique proposals; PostgreSQL saturation, 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]` | **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.52 `[D:8.32,8.34,8.45,8.51]` | **IN PROGRESS.** Allocator supports both an opt-in per-replica fixed-window quota (`--allocation-quota` / `--allocation-quota-window`) and an optional PostgreSQL-backed EU/NA quota table consumed inside the serializable allocation transaction before any provider call; idempotent replays do not double-count | Normal/race/vet tests cover local quota exhaustion, window reset, region isolation, invalid input, and atomic concurrent consumption; migration/SQL coverage defines the shared quota boundary; measured regional cost model, 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]` | **IN PROGRESS.** `scripts/verify_release_gate.py` provides a fail-closed promotion check for the ordered development → internal → casual canary → casual → provisional ranked → ranked stages, requiring an evidence report for SLO, security, cost, rollback, EU+NA playtests, and both legacy gates | Validator and adversarial tests cover skipped stages, unknown stages, missing gates, non-boolean gate values, and blank release IDs; the actual reports, production rollback rehearsal, regional playtests, and live promotion remain open | Implementation invariants for every task above: diff --git a/server/allocator/service.go b/server/allocator/service.go index aa3a1d8b..9bfccf87 100644 --- a/server/allocator/service.go +++ b/server/allocator/service.go @@ -30,11 +30,16 @@ type AllocationBudget interface { Allow(region string, now time.Time) error } +type SharedAllocationQuota interface { + Consume(context.Context, string, time.Time) error +} + type Service struct { Provider Provider Durable Durable Roster RosterPublisher Budget AllocationBudget + Quota SharedAllocationQuota Now func() time.Time } @@ -88,6 +93,11 @@ func (s Service) Allocate(ctx context.Context, request domain.AllocationRequest, return agones.AllocatedServer{}, err } } + if s.Quota != nil { + if err := s.Quota.Consume(ctx, 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 @@ -102,6 +112,11 @@ func (s Service) RecordProviderAllocation(ctx context.Context, result agones.All if s.Durable == nil || result.Allocation.State != domain.ServerAllocated || result.Endpoint == "" { return domain.Allocation{}, domain.ErrAllocationInput } + if s.Quota != nil { + if err := s.Quota.Consume(ctx, result.Allocation.Region, now); err != nil { + return domain.Allocation{}, err + } + } return s.Durable.RecordProviderAllocation(ctx, result.Allocation, now) } diff --git a/server/allocator/service_test.go b/server/allocator/service_test.go index b0194206..116b4ce6 100644 --- a/server/allocator/service_test.go +++ b/server/allocator/service_test.go @@ -32,6 +32,16 @@ type rosterSpy struct { err error } +type quotaSpy struct { + calls int + err error +} + +func (q *quotaSpy) Consume(context.Context, string, time.Time) error { + q.calls++ + return q.err +} + func (r *rosterSpy) PublishRoster(_ context.Context, _ domain.Assignment, _ []domain.SignedJoinAuthorisation, _ func([]byte, []byte) bool) error { r.calls++ return r.err @@ -63,6 +73,31 @@ func TestServiceDoesNotReturnProviderResultAfterDurableFailure(t *testing.T) { } } +func TestServiceConsumesSharedQuotaBeforeFreshProviderCall(t *testing.T) { + provider := &providerSpy{result: agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: "a", State: domain.ServerAllocated}, Endpoint: "127.0.0.1:7777"}} + quota := "aSpy{err: errors.New("quota exhausted")} + service := Service{Provider: provider, Durable: &durableSpy{}, Quota: quota, Now: func() time.Time { return time.Unix(1000, 0) }} + if _, err := service.Allocate(context.Background(), domain.AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 1, Transport: "enet"}, nil); err == nil { + t.Fatal("quota rejection was ignored") + } + if quota.calls != 1 || provider.calls != 0 { + t.Fatalf("quota/provider calls = %d/%d, want 1/0", quota.calls, provider.calls) + } +} + +func TestServiceConsumesSharedQuotaOnceWhenReconcilingProviderResult(t *testing.T) { + quota := "aSpy{} + durable := &durableSpy{} + service := Service{Durable: durable, Quota: quota, Now: func() time.Time { return time.Unix(1000, 0) }} + result := agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: "a", MatchID: "m", Region: "EU", State: domain.ServerAllocated}, Endpoint: "127.0.0.1:7777"} + if _, err := service.RecordProviderAllocation(context.Background(), result, time.Unix(1000, 0)); err != nil { + t.Fatalf("reconciliation failed: %v", err) + } + if quota.calls != 1 || durable.calls != 1 { + t.Fatalf("quota/durable calls = %d/%d, want 1/1", quota.calls, durable.calls) + } +} + func TestServiceAllocatesOnlyUnanimouslyAcceptedMatchingProposal(t *testing.T) { proposal := domain.Proposal{ ProposalID: "proposal-1", Playlist: domain.Casual, State: domain.Accepted, diff --git a/server/cmd/allocator/main.go b/server/cmd/allocator/main.go index 380be26b..8d1047ff 100644 --- a/server/cmd/allocator/main.go +++ b/server/cmd/allocator/main.go @@ -68,6 +68,7 @@ func main() { Service: allocator.Service{ Provider: client, Durable: store.AllocationRegistry{DB: db}, + Quota: store.AllocationQuota{DB: db}, Budget: budget, Now: now, }, diff --git a/server/migrations/0007_allocation_quotas.sql b/server/migrations/0007_allocation_quotas.sql new file mode 100644 index 00000000..8f7091e0 --- /dev/null +++ b/server/migrations/0007_allocation_quotas.sql @@ -0,0 +1,10 @@ +-- Optional operator-configured regional spend guard. A missing row means +-- unlimited, preserving existing deployments until they opt into a quota. +CREATE TABLE allocation_quotas ( + region TEXT PRIMARY KEY CHECK (region IN ('EU', 'NA')), + window_started_at TIMESTAMPTZ NOT NULL, + window_seconds INTEGER NOT NULL CHECK (window_seconds > 0), + used_allocations INTEGER NOT NULL DEFAULT 0 CHECK (used_allocations >= 0), + max_allocations INTEGER NOT NULL CHECK (max_allocations > 0), + updated_at TIMESTAMPTZ NOT NULL +); diff --git a/server/migrations/down/0007_allocation_quotas.sql b/server/migrations/down/0007_allocation_quotas.sql new file mode 100644 index 00000000..c53d4b99 --- /dev/null +++ b/server/migrations/down/0007_allocation_quotas.sql @@ -0,0 +1 @@ +DROP TABLE allocation_quotas; diff --git a/server/migrations/test_migration.py b/server/migrations/test_migration.py index e2d075e8..664c2cb1 100644 --- a/server/migrations/test_migration.py +++ b/server/migrations/test_migration.py @@ -6,6 +6,7 @@ import unittest SQL = (Path(__file__).parent / "0001_initial.sql").read_text() ASSIGNMENTS_SQL = (Path(__file__).parent / "0002_assignments.sql").read_text() +QUOTAS_SQL = (Path(__file__).parent / "0007_allocation_quotas.sql").read_text() class MigrationTest(unittest.TestCase): @@ -53,6 +54,11 @@ class MigrationTest(unittest.TestCase): ): self.assertIn(fragment, ASSIGNMENTS_SQL) + def test_allocation_quotas_are_optional_and_region_bound(self): + for fragment in ("CREATE TABLE allocation_quotas", "region TEXT PRIMARY KEY", "window_seconds", "max_allocations"): + self.assertIn(fragment, QUOTAS_SQL) + self.assertIn("region IN ('EU', 'NA')", QUOTAS_SQL) + if __name__ == "__main__": unittest.main() diff --git a/server/store/allocation_quota_sql.go b/server/store/allocation_quota_sql.go new file mode 100644 index 00000000..93ac2dc3 --- /dev/null +++ b/server/store/allocation_quota_sql.go @@ -0,0 +1,78 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +var ErrAllocationQuotaExceeded = errors.New("allocation quota exceeded") + +type AllocationQuota struct { + DB *sql.DB +} + +const allocationQuotaSelectSQL = `SELECT window_started_at, window_seconds, used_allocations, max_allocations +FROM allocation_quotas WHERE region = $1 FOR UPDATE` + +const allocationQuotaResetSQL = `UPDATE allocation_quotas +SET window_started_at = $2, used_allocations = 1, updated_at = $2 WHERE region = $1` + +const allocationQuotaIncrementSQL = `UPDATE allocation_quotas +SET used_allocations = used_allocations + 1, updated_at = $2 WHERE region = $1` + +const SetAllocationQuotaSQL = `INSERT INTO allocation_quotas + (region, window_started_at, window_seconds, used_allocations, max_allocations, updated_at) +VALUES ($1, $2, $3, 0, $4, $2) +ON CONFLICT (region) DO UPDATE SET window_started_at = EXCLUDED.window_started_at, + window_seconds = EXCLUDED.window_seconds, used_allocations = 0, + max_allocations = EXCLUDED.max_allocations, updated_at = EXCLUDED.updated_at` + +// SetAllocationQuota configures the optional shared regional quota. It is +// intended for operator provisioning, not for a request path. +func SetAllocationQuota(ctx context.Context, db *sql.DB, region string, maxAllocations int, window time.Duration, now time.Time) error { + if db == nil || (region != "EU" && region != "NA") || maxAllocations < 1 || window <= 0 || window > 365*24*time.Hour || now.IsZero() { + return fmt.Errorf("invalid allocation quota") + } + seconds := int(window / time.Second) + if seconds < 1 { + return fmt.Errorf("allocation quota window is too small") + } + _, err := db.ExecContext(ctx, SetAllocationQuotaSQL, region, now, seconds, maxAllocations) + return err +} + +func (q AllocationQuota) Consume(ctx context.Context, region string, now time.Time) error { + if q.DB == nil || (region != "EU" && region != "NA") || now.IsZero() { + return fmt.Errorf("invalid allocation quota request") + } + return RunSerializable(ctx, q.DB, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { + return consumeAllocationQuotaTx(ctx, tx, region, now) + }) +} + +// consumeAllocationQuotaTx consumes one unit when a quota row exists. The +// caller must already be inside the serializable allocation transaction; the +// row lock makes this global across allocator replicas sharing PostgreSQL. +func consumeAllocationQuotaTx(ctx context.Context, tx *sql.Tx, region string, now time.Time) error { + var started time.Time + var seconds, used, maximum int + err := tx.QueryRowContext(ctx, allocationQuotaSelectSQL, region).Scan(&started, &seconds, &used, &maximum) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return err + } + if !now.Before(started.Add(time.Duration(seconds) * time.Second)) { + _, err = tx.ExecContext(ctx, allocationQuotaResetSQL, region, now) + return err + } + if used >= maximum { + return ErrAllocationQuotaExceeded + } + _, err = tx.ExecContext(ctx, allocationQuotaIncrementSQL, region, now) + return err +} diff --git a/server/store/allocator_sql.go b/server/store/allocator_sql.go index 46af3281..c439a7bb 100644 --- a/server/store/allocator_sql.go +++ b/server/store/allocator_sql.go @@ -75,6 +75,9 @@ func ClaimAllocation(ctx context.Context, db *sql.DB, request domain.AllocationR if err != sql.ErrNoRows { return err } + if err := consumeAllocationQuotaTx(ctx, tx, request.Region, now); err != nil { + return err + } var serverID string if err := tx.QueryRowContext(ctx, ClaimReadyServerSQL, request.Region, request.Build, request.Protocol, request.Transport, now).Scan(&serverID); err != nil { if err == sql.ErrNoRows { diff --git a/server/store/allocator_sql_test.go b/server/store/allocator_sql_test.go index 9860756b..bb2cf304 100644 --- a/server/store/allocator_sql_test.go +++ b/server/store/allocator_sql_test.go @@ -21,6 +21,29 @@ func TestAllocatorSQLClaimsAndAuditsCompatibleReadyServers(t *testing.T) { } } } + for _, fragment := range []string{"allocation_quotas", "ON CONFLICT (region)", "used_allocations"} { + if !contains(SetAllocationQuotaSQL, fragment) { + t.Fatalf("quota query missing %q", fragment) + } + } +} + +func TestSetAllocationQuotaRejectsInvalidArgumentsWithoutDatabase(t *testing.T) { + if err := SetAllocationQuota(nil, nil, "EU", 1, time.Minute, time.Unix(1000, 0)); err == nil { + t.Fatal("nil database accepted") + } + if err := SetAllocationQuota(nil, nil, "APAC", 1, time.Minute, time.Unix(1000, 0)); err == nil { + t.Fatal("unknown region accepted") + } + if err := SetAllocationQuota(nil, nil, "EU", 0, time.Minute, time.Unix(1000, 0)); err == nil { + t.Fatal("zero limit accepted") + } +} + +func TestAllocationQuotaConsumeRejectsInvalidArgumentsWithoutDatabase(t *testing.T) { + if err := (AllocationQuota{}).Consume(nil, "EU", time.Unix(1000, 0)); err == nil { + t.Fatal("nil database accepted") + } } func TestClaimAllocationRejectsInvalidRequestsWithoutDatabase(t *testing.T) { diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 6bde46b8..aff4eb61 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -44,7 +44,7 @@ func openIntegrationPostgres(t *testing.T) *sql.DB { func applyIntegrationMigrations(t *testing.T, db *sql.DB) { t.Helper() - if _, err := db.ExecContext(context.Background(), `DROP TABLE IF EXISTS schema_migrations, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil { + if _, err := db.ExecContext(context.Background(), `DROP TABLE IF EXISTS schema_migrations, allocation_quotas, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil { t.Fatalf("reset PostgreSQL schema: %v", err) } if err := migrations.Apply(context.Background(), db, filepath.Join("..", "migrations")); err != nil { @@ -104,6 +104,41 @@ func TestPostgreSQLAllocatorClaimReplayAndCapacityFence(t *testing.T) { } } +func TestPostgreSQLSharedAllocationQuotaFencesClaimsAndReplays(t *testing.T) { + db := openIntegrationPostgres(t) + applyIntegrationMigrations(t, db) + now := time.Now().UTC().Truncate(time.Microsecond) + ctx := context.Background() + if err := SetAllocationQuota(ctx, db, "EU", 1, time.Minute, now); err != nil { + t.Fatalf("set quota: %v", err) + } + for _, server := range []domain.ReadyServer{ + {ServerID: "quota-server-a", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}, + {ServerID: "quota-server-b", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}, + } { + if err := RegisterReadyServer(ctx, db, server, now); err != nil { + t.Fatal(err) + } + } + first := domain.AllocationRequest{AllocationID: "quota-allocation-1", MatchID: "quota-match-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"} + if _, err := ClaimAllocation(ctx, db, first, now); err != nil { + t.Fatalf("first claim: %v", err) + } + if _, err := ClaimAllocation(ctx, db, first, now.Add(time.Second)); err != nil { + t.Fatalf("idempotent replay was fenced: %v", err) + } + second := domain.AllocationRequest{AllocationID: "quota-allocation-2", MatchID: "quota-match-2", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"} + if _, err := ClaimAllocation(ctx, db, second, now.Add(2*time.Second)); !errors.Is(err, ErrAllocationQuotaExceeded) { + t.Fatalf("second claim err=%v, want shared quota fence", err) + } + if err := SetAllocationQuota(ctx, db, "EU", 1, time.Minute, now.Add(time.Minute)); err != nil { + t.Fatalf("reset quota: %v", err) + } + if _, err := ClaimAllocation(ctx, db, second, now.Add(time.Minute)); err != nil { + t.Fatalf("claim after quota window reset: %v", err) + } +} + // TestPostgreSQLConcurrentAllocationClaimNeverDoubleBooksAReadyServer is the // live counterpart to TestPostgreSQLAllocatorClaimReplayAndCapacityFence: that // test claims strictly one request at a time, so it cannot show what happens @@ -1249,8 +1284,8 @@ func TestPostgreSQLMigrationsRollBackAndReapplyCleanly(t *testing.T) { // Roll back every migration one at a time, in reverse, checking each // down file actually undoes what its forward file created — not just // that Rollback returns nil. - if err := migrations.Rollback(context.Background(), db, dir, 1); err != nil { - t.Fatalf("rollback 0006: %v", err) + if err := migrations.Rollback(context.Background(), db, dir, 2); err != nil { + t.Fatalf("rollback 0007 and 0006: %v", err) } var hasAllocationClaimColumn bool if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'allocation_id'`).Scan(&hasAllocationClaimColumn); err != nil {