jcreek/CosmicClash
1
0
Fork 0
mirror of https://github.com/jcreek/CosmicClash.git synced 2026-09-10 16:04:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(multiplayer): recover ambiguous agones allocations

Browse Source
This commit is contained in:
Josh Creek
2026-09-01 18:05:43 +01:00
parent b72a7cf843
commit bf396afcaf
6 changed files with 161 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
multiplayer-next.md
+1 -1
View File
File diff suppressed because one or more lines are too long
server/agones/allocation.go
+68 -3
View File
@@ -85,15 +85,80 @@ type allocationResponse struct {
type gameServerListResponse struct {
Items []struct {
Metadata struct {
Name string `json:"name"`
Labels map[string]string `json:"labels"`
Name string `json:"name"`
Labels map[string]string `json:"labels"`
Annotations map[string]string `json:"annotations"`
} `json:"metadata"`
Status struct {
State string `json:"state"`
State string `json:"state"`
Address string `json:"address"`
Ports []struct {
Name string `json:"name"`
Port int `json:"port"`
} `json:"ports"`
} `json:"status"`
} `json:"items"`
}
// RecoverAllocation finds a provider-side allocation that may have completed
// before the durable allocation record was written. The allocation ID and
// compatibility tuple are checked together so a stale or forged provider
// object cannot be rebound to another match.
func (c Client) RecoverAllocation(ctx context.Context, request domain.AllocationRequest, now time.Time) (AllocatedServer, bool, error) {
if request.AllocationID == "" || request.MatchID == "" || now.IsZero() {
return AllocatedServer{}, false, domain.ErrAllocationInput
}
if c.HTTP == nil {
c.HTTP = http.DefaultClient
}
base, err := c.endpoint()
if err != nil {
return AllocatedServer{}, false, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/apis/agones.dev/v1/namespaces/"+url.PathEscape(c.Namespace)+"/gameservers", nil)
if err != nil {
return AllocatedServer{}, false, err
}
response, err := c.HTTP.Do(req)
if err != nil {
return AllocatedServer{}, false, err
}
defer response.Body.Close()
if response.StatusCode < 200 || response.StatusCode >= 300 {
return AllocatedServer{}, false, fmt.Errorf("Agones GameServer recovery returned %s", response.Status)
}
var decoded gameServerListResponse
if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&decoded); err != nil {
return AllocatedServer{}, false, fmt.Errorf("decode Agones recovery list: %w", err)
}
found := false
var recovered AllocatedServer
for _, item := range decoded.Items {
if item.Status.State != "Allocated" || item.Metadata.Annotations["cosmic-clash.io/allocation-id"] != request.AllocationID {
continue
}
if found {
return AllocatedServer{}, false, domain.ErrConflict
}
if item.Metadata.Name == "" || item.Status.Address == "" || strings.ContainsAny(item.Status.Address, " \t\r\n") {
return AllocatedServer{}, false, fmt.Errorf("Agones recovered GameServer has invalid identity or address")
}
if item.Metadata.Annotations["cosmic-clash.io/match-id"] != request.MatchID {
return AllocatedServer{}, false, domain.ErrConflict
}
if item.Metadata.Labels["cosmic-clash.io/region"] != request.Region || item.Metadata.Labels["cosmic-clash.io/build"] != request.Build || item.Metadata.Labels["cosmic-clash.io/protocol"] != strconv.Itoa(request.Protocol) || item.Metadata.Labels["cosmic-clash.io/transport"] != request.Transport {
return AllocatedServer{}, false, domain.ErrConflict
}
port, err := selectPort(item.Status.Ports)
if err != nil {
return AllocatedServer{}, false, err
}
recovered = AllocatedServer{Allocation: domain.Allocation{AllocationID: request.AllocationID, MatchID: request.MatchID, ServerID: item.Metadata.Name, Region: request.Region, Build: request.Build, Protocol: request.Protocol, Transport: request.Transport, State: domain.ServerAllocated, AllocatedAt: now}, Endpoint: net.JoinHostPort(item.Status.Address, strconv.Itoa(port)), GameServer: item.Metadata.Name}
found = true
}
return recovered, found, nil
}
// 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.
server/agones/allocation_test.go
+34
View File
@@ -154,3 +154,37 @@ func TestListReadyServersFailsClosedOnInvalidReadyCompatibility(t *testing.T) {
t.Fatal("invalid Ready GameServer accepted")
}
}
func TestRecoverAllocationFindsMatchingAllocatedGameServer(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet || !strings.HasSuffix(r.URL.Path, "/gameservers") {
t.Fatalf("request=%s %s", r.Method, r.URL.Path)
}
_, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"gs-recovered","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"},"annotations":{"cosmic-clash.io/allocation-id":"allocation-1","cosmic-clash.io/match-id":"match-1"}},"status":{"state":"Allocated","address":"127.0.0.1","ports":[{"name":"default","port":31001}]}},{"metadata":{"name":"gs-other","annotations":{"cosmic-clash.io/allocation-id":"other"},"status":{"state":"Allocated"}}}]}`))
}))
defer server.Close()
recovered, found, err := (Client{BaseURL: server.URL, Namespace: "games", HTTP: server.Client()}).RecoverAllocation(context.Background(), request(), time.Unix(1000, 0))
if err != nil || !found || recovered.GameServer != "gs-recovered" || recovered.Endpoint != "127.0.0.1:31001" || recovered.Allocation.ServerID != "gs-recovered" {
t.Fatalf("recovered=%+v found=%t err=%v", recovered, found, err)
}
}
func TestRecoverAllocationRejectsMismatchedBinding(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"gs-forged","labels":{"cosmic-clash.io/region":"NA","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"},"annotations":{"cosmic-clash.io/allocation-id":"allocation-1","cosmic-clash.io/match-id":"other-match"}},"status":{"state":"Allocated","address":"127.0.0.1","ports":[{"name":"default","port":31001}]}}]}`))
}))
defer server.Close()
if _, found, err := (Client{BaseURL: server.URL, Namespace: "games", HTTP: server.Client()}).RecoverAllocation(context.Background(), request(), time.Unix(1000, 0)); err == nil || found {
t.Fatalf("mismatched recovery accepted: found=%t err=%v", found, err)
}
}
func TestRecoverAllocationRejectsDuplicateProviderMatches(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"gs-one","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"},"annotations":{"cosmic-clash.io/allocation-id":"allocation-1","cosmic-clash.io/match-id":"match-1"}},"status":{"state":"Allocated","address":"127.0.0.1","ports":[{"name":"default","port":31001}]}},{"metadata":{"name":"gs-two","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"},"annotations":{"cosmic-clash.io/allocation-id":"allocation-1","cosmic-clash.io/match-id":"match-1"}},"status":{"state":"Allocated","address":"127.0.0.1","ports":[{"name":"default","port":31002}]}}]}`))
}))
defer server.Close()
if _, found, err := (Client{BaseURL: server.URL, Namespace: "games", HTTP: server.Client()}).RecoverAllocation(context.Background(), request(), time.Unix(1000, 0)); err == nil || found {
t.Fatalf("duplicate recovery accepted: found=%t err=%v", found, err)
}
}
server/allocator/service.go
+11
View File
@@ -14,6 +14,10 @@ type Provider interface {
Allocate(context.Context, domain.AllocationRequest, map[string]string, time.Time) (agones.AllocatedServer, error)
}
type ProviderRecoverer interface {
RecoverAllocation(context.Context, domain.AllocationRequest, time.Time) (agones.AllocatedServer, bool, error)
}
type Durable interface {
RecordProviderAllocation(context.Context, domain.Allocation, time.Time) (domain.Allocation, error)
}
@@ -84,6 +88,13 @@ func (s Service) Allocate(ctx context.Context, request domain.AllocationRequest,
return result, nil
}
func (s Service) RecordProviderAllocation(ctx context.Context, result agones.AllocatedServer, now time.Time) (domain.Allocation, error) {
if s.Durable == nil || result.Allocation.State != domain.ServerAllocated || result.Endpoint == "" {
return domain.Allocation{}, domain.ErrAllocationInput
}
return s.Durable.RecordProviderAllocation(ctx, result.Allocation, now)
}
var errNotConfigured = &configurationError{}
type configurationError struct{}
server/allocator/worker.go
+23 -4
View File
@@ -43,11 +43,30 @@ func (w Worker) RunOnce(ctx context.Context) (bool, error) {
return true, fmt.Errorf("recover allocation for match %s: %w", request.MatchID, err)
}
if !recorded {
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 recoverer, ok := w.Service.Provider.(ProviderRecoverer); ok {
recovered, found, err := recoverer.RecoverAllocation(ctx, request, w.Now())
if err != nil {
return true, fmt.Errorf("recover provider allocation for match %s: %w", request.MatchID, err)
}
if found {
if _, err := w.Service.RecordProviderAllocation(ctx, recovered, w.Now()); err != nil {
return true, fmt.Errorf("record recovered allocation for match %s: %w", request.MatchID, err)
}
allocation = recovered.Allocation
} else {
result, err := w.Service.Allocate(ctx, request, AllocationLabels(request))
if err != nil {
return true, fmt.Errorf("allocate claimed match %s: %w", request.MatchID, err)
}
allocation = result.Allocation
}
} else {
result, err := w.Service.Allocate(ctx, request, AllocationLabels(request))
if err != nil {
return true, fmt.Errorf("allocate claimed match %s: %w", request.MatchID, err)
}
allocation = result.Allocation
}
allocation = result.Allocation
}
if err := w.Claims.BindAllocatedMatch(ctx, allocation); err != nil {
return true, fmt.Errorf("bind allocated match %s: %w", request.MatchID, err)
server/allocator/worker_test.go
+24
View File
@@ -21,6 +21,17 @@ type matchClaimSpy struct {
bindErr error
}
type recoverableProviderSpy struct {
providerSpy
recovered agones.AllocatedServer
found bool
recoverErr error
}
func (p *recoverableProviderSpy) RecoverAllocation(_ context.Context, _ domain.AllocationRequest, _ time.Time) (agones.AllocatedServer, bool, error) {
return p.recovered, p.found, p.recoverErr
}
func (s *matchClaimSpy) FindProviderAllocation(_ context.Context, _ domain.AllocationRequest) (domain.Allocation, bool, error) {
return s.recorded, s.recorded.AllocationID != "", s.recordErr
}
@@ -69,6 +80,19 @@ func TestWorkerRecoversDurableProviderAllocationWithoutCallingProvider(t *testin
}
}
func TestWorkerRecoversProviderAllocationBeforeIssuingSecondAllocation(t *testing.T) {
request := domain.AllocationRequest{AllocationID: "allocation-1", MatchID: "match-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
recovered := agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: request.AllocationID, MatchID: request.MatchID, ServerID: "server-recovered", Region: request.Region, Build: request.Build, Protocol: request.Protocol, Transport: request.Transport, State: domain.ServerAllocated}, Endpoint: "127.0.0.1:31001"}
claims := &matchClaimSpy{request: request, found: true}
provider := &recoverableProviderSpy{recovered: recovered, found: true}
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 != 0 || durable.calls != 1 || claims.bound.ServerID != "server-recovered" {
t.Fatalf("processed=%t err=%v provider_calls=%d durable_calls=%d bound=%+v", processed, err, provider.calls, durable.calls, claims.bound)
}
}
func TestWorkerDoesNothingWhenNoDurableMatchIsAvailable(t *testing.T) {
claims := &matchClaimSpy{}
worker := Worker{Claims: claims, Now: func() time.Time { return time.Unix(1_000, 0) }}