mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-11 00:14:00 +00:00
feat: add deterministic allocation policy
This commit is contained in:
Josh Creek
+1
-1
@@ -1212,7 +1212,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, dynamic `SDR_LISTEN_PORT`/`SDR_IP` injection, explicit process-ready probing and Ready transition; direct mode bypasses Agones | `server/supervisor/` covers allocated/direct startup and dynamic endpoint/Ready ordering; Godot Agones adapter, metadata watch, Health/annotation/Shutdown 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; Godot readiness endpoint, detached-container and Health-reclaim integration remain |
|
||||
| 8.29 `[D:8.26,8.27]` | **IN PROGRESS.** Supervisor discovers 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 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]` | Atomic `GameServerAllocation` from Ready filtered by region/build/protocol/transport, attaching signed roster/non-secret config with bounded race retry | Duplicate commands yield one Allocated server; exhaustion or retry leaves no orphan; no client assignment is exposed merely because process is Ready |
|
||||
| 8.30 `[D:8.18,8.26,8.28,8.29]` | **IN PROGRESS.** Pure Go allocator filters Ready servers by region/build/protocol/transport and atomically claims one with idempotent allocation replay; assignment is not exposed from Ready state | `server/domain/allocator.go` covers deterministic compatible selection, exhaustion, conflicting/identical replay and invalid server input; Agones `GameServerAllocation`, signed roster metadata, bounded cross-replica retry and live integration remain |
|
||||
| 8.31 `[D:8.9,8.30]` | **Assignment-ready stage:** watch Allocated metadata, verify manifest/bindings, register hosted address, acknowledge backend; only then mint/expose client tickets | Modified/wrong manifest never reaches assignment-ready; clients cannot connect early; secrets never appear in metadata/args/logs |
|
||||
| 8.32 `[D:8.2,8.26,8.30]` | FleetAutoscaler with >=2 Ready processes across >=2 on-demand nodes/failure domains per queue-enabled region; pre-pull current/rollback; scale **Allocated** count to zero, never the Ready floor | Warm allocation meets p95 5 s/p99 10 s; disabled regions alone scale fully to zero; one-node loss retains certified Ready/headroom |
|
||||
| 8.33 `[D:8.26,8.32]` | On-demand-only live capacity and measured N+1: loss of largest node leaves two Ready slots plus headroom for surviving Allocated matches | Interruptible nodes cannot receive live matches; forced node loss neither overloads survivors nor prevents the next allocation |
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
package domain
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ServerLifecycle string
|
||||
|
||||
const (
|
||||
ServerReady ServerLifecycle = "READY"
|
||||
ServerAllocated ServerLifecycle = "ALLOCATED"
|
||||
)
|
||||
|
||||
type ReadyServer struct {
|
||||
ServerID string
|
||||
Region string
|
||||
Build string
|
||||
Protocol int
|
||||
Transport string
|
||||
State ServerLifecycle
|
||||
}
|
||||
|
||||
type AllocationRequest struct {
|
||||
AllocationID string
|
||||
MatchID string
|
||||
Region string
|
||||
Build string
|
||||
Protocol int
|
||||
Transport string
|
||||
}
|
||||
|
||||
type Allocation struct {
|
||||
AllocationID string
|
||||
MatchID string
|
||||
ServerID string
|
||||
State ServerLifecycle
|
||||
AllocatedAt time.Time
|
||||
}
|
||||
|
||||
type Allocator struct {
|
||||
mu sync.Mutex
|
||||
servers map[string]ReadyServer
|
||||
allocations map[string]Allocation
|
||||
requestHashes map[string][32]byte
|
||||
}
|
||||
|
||||
var (
|
||||
ErrNoCapacity = fmt.Errorf("no compatible ready server")
|
||||
ErrAllocationInput = fmt.Errorf("invalid allocation request")
|
||||
)
|
||||
|
||||
func NewAllocator(servers []ReadyServer) (*Allocator, error) {
|
||||
a := &Allocator{servers: make(map[string]ReadyServer, len(servers)), allocations: make(map[string]Allocation), requestHashes: make(map[string][32]byte)}
|
||||
for _, server := range servers {
|
||||
if server.ServerID == "" || server.Region == "" || server.Build == "" || server.Protocol <= 0 || (server.Transport != "enet" && server.Transport != "steam_sdr") || server.State != ServerReady {
|
||||
return nil, fmt.Errorf("%w: invalid ready server", ErrAllocationInput)
|
||||
}
|
||||
if _, exists := a.servers[server.ServerID]; exists {
|
||||
return nil, fmt.Errorf("%w: duplicate server", ErrAllocationInput)
|
||||
}
|
||||
a.servers[server.ServerID] = server
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// Allocate is the in-process equivalent of a GameServerAllocation. The mutex
|
||||
// represents the durable allocator transaction; the PostgreSQL/Agones adapter
|
||||
// must preserve this claim-before-assignment ordering across replicas.
|
||||
func (a *Allocator) Allocate(request AllocationRequest, now time.Time) (Allocation, error) {
|
||||
if err := validateAllocationRequest(request); err != nil {
|
||||
return Allocation{}, err
|
||||
}
|
||||
digest := allocationDigest(request)
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
if prior, ok := a.allocations[request.AllocationID]; ok {
|
||||
if a.requestHashes[request.AllocationID] != digest {
|
||||
return Allocation{}, ErrConflict
|
||||
}
|
||||
return prior, nil
|
||||
}
|
||||
ids := make([]string, 0)
|
||||
for id, server := range a.servers {
|
||||
if server.State == ServerReady && server.Region == request.Region && server.Build == request.Build && server.Protocol == request.Protocol && server.Transport == request.Transport {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return Allocation{}, ErrNoCapacity
|
||||
}
|
||||
sort.Strings(ids)
|
||||
server := a.servers[ids[0]]
|
||||
server.State = ServerAllocated
|
||||
a.servers[server.ServerID] = server
|
||||
allocation := Allocation{AllocationID: request.AllocationID, MatchID: request.MatchID, ServerID: server.ServerID, State: ServerAllocated, AllocatedAt: now}
|
||||
a.allocations[request.AllocationID] = allocation
|
||||
a.requestHashes[request.AllocationID] = digest
|
||||
return allocation, nil
|
||||
}
|
||||
|
||||
func validateAllocationRequest(request AllocationRequest) error {
|
||||
if request.AllocationID == "" || request.MatchID == "" || request.Region == "" || request.Build == "" || request.Protocol <= 0 || (request.Transport != "enet" && request.Transport != "steam_sdr") {
|
||||
return ErrAllocationInput
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func allocationDigest(request AllocationRequest) [32]byte {
|
||||
return sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%s\x00%s\x00%d\x00%s", request.AllocationID, request.MatchID, request.Region, request.Build, request.Protocol, request.Transport)))
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package domain
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestAllocatorFiltersAndAtomicallyClaimsCompatibleReadyServer(t *testing.T) {
|
||||
a, err := NewAllocator([]ReadyServer{
|
||||
{ServerID: "server-b", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: ServerReady},
|
||||
{ServerID: "server-a", Region: "NA", Build: "build-1", Protocol: 1, Transport: "enet", State: ServerReady},
|
||||
{ServerID: "server-c", Region: "EU", Build: "build-2", Protocol: 1, Transport: "enet", State: ServerReady},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
request := AllocationRequest{AllocationID: "allocation-1", MatchID: "match-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
|
||||
got, err := a.Allocate(request, time.Unix(1000, 0))
|
||||
if err != nil || got.ServerID != "server-b" || got.State != ServerAllocated {
|
||||
t.Fatalf("allocation = %+v err=%v", got, err)
|
||||
}
|
||||
if _, err := a.Allocate(AllocationRequest{AllocationID: "allocation-2", MatchID: "match-2", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}, time.Unix(1001, 0)); !errors.Is(err, ErrNoCapacity) {
|
||||
t.Fatalf("claimed server was reused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllocatorIsIdempotentAndRejectsConflictingReplay(t *testing.T) {
|
||||
a, _ := NewAllocator([]ReadyServer{{ServerID: "server-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "steam_sdr", State: ServerReady}})
|
||||
request := AllocationRequest{AllocationID: "allocation-1", MatchID: "match-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "steam_sdr"}
|
||||
first, err := a.Allocate(request, time.Unix(1000, 0))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
replay, err := a.Allocate(request, time.Unix(2000, 0))
|
||||
if err != nil || replay != first {
|
||||
t.Fatalf("replay = %+v err=%v", replay, err)
|
||||
}
|
||||
request.MatchID = "match-2"
|
||||
if _, err := a.Allocate(request, time.Unix(2000, 0)); !errors.Is(err, ErrConflict) {
|
||||
t.Fatalf("conflicting replay = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllocatorRejectsInvalidServerAndNoCompatibleCapacity(t *testing.T) {
|
||||
if _, err := NewAllocator([]ReadyServer{{ServerID: "server-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "udp", State: ServerReady}}); !errors.Is(err, ErrAllocationInput) {
|
||||
t.Fatalf("invalid server accepted: %v", err)
|
||||
}
|
||||
a, _ := NewAllocator(nil)
|
||||
if _, err := a.Allocate(AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 1, Transport: "enet"}, time.Unix(1000, 0)); !errors.Is(err, ErrNoCapacity) {
|
||||
t.Fatalf("empty allocator error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllocatorConcurrentClaimsCannotDoubleAllocateOneServer(t *testing.T) {
|
||||
a, _ := NewAllocator([]ReadyServer{{ServerID: "server-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: ServerReady}})
|
||||
requests := []AllocationRequest{
|
||||
{AllocationID: "allocation-a", MatchID: "match-a", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"},
|
||||
{AllocationID: "allocation-b", MatchID: "match-b", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"},
|
||||
}
|
||||
var wg sync.WaitGroup
|
||||
results := make(chan error, len(requests))
|
||||
for _, request := range requests {
|
||||
wg.Add(1)
|
||||
go func(request AllocationRequest) {
|
||||
defer wg.Done()
|
||||
_, err := a.Allocate(request, time.Unix(1000, 0))
|
||||
results <- err
|
||||
}(request)
|
||||
}
|
||||
wg.Wait()
|
||||
close(results)
|
||||
wins := 0
|
||||
for err := range results {
|
||||
if err == nil {
|
||||
wins++
|
||||
} else if !errors.Is(err, ErrNoCapacity) {
|
||||
t.Fatalf("unexpected concurrent claim error: %v", err)
|
||||
}
|
||||
}
|
||||
if wins != 1 {
|
||||
t.Fatalf("concurrent claims succeeded %d times", wins)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user