feat: add durable allocator claim boundary

server/migrations/0004_allocator_registry.sql

@@ -0,0 +1,29 @@
+-- Durable allocator registry. Agones remains the provider-facing lifecycle
+-- authority; these rows are the control-plane's auditable claim projection.
+CREATE TABLE game_servers (
+    server_id TEXT PRIMARY KEY,
+    region TEXT NOT NULL CHECK (region IN ('EU', 'NA')),
+    build TEXT NOT NULL,
+    protocol_version INTEGER NOT NULL CHECK (protocol_version > 0),
+    transport TEXT NOT NULL CHECK (transport IN ('enet', 'steam_sdr')),
+    state TEXT NOT NULL CHECK (state IN ('READY', 'ALLOCATED')),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE allocations (
+    allocation_id TEXT PRIMARY KEY,
+    match_id TEXT NOT NULL UNIQUE,
+    server_id TEXT NOT NULL REFERENCES game_servers(server_id),
+    region TEXT NOT NULL CHECK (region IN ('EU', 'NA')),
+    build TEXT NOT NULL,
+    protocol_version INTEGER NOT NULL CHECK (protocol_version > 0),
+    transport TEXT NOT NULL CHECK (transport IN ('enet', 'steam_sdr')),
+    request_digest BYTEA NOT NULL,
+    state TEXT NOT NULL CHECK (state = 'ALLOCATED'),
+    allocated_at TIMESTAMPTZ NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX game_servers_ready_compatibility
+    ON game_servers (region, build, protocol_version, transport, server_id)
+    WHERE state = 'READY';

server/store/allocator_sql.go

@@ -0,0 +1,85 @@
+package store
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+const RegisterReadyServerSQL = `INSERT INTO game_servers
+    (server_id, region, build, protocol_version, transport, state, updated_at)
+VALUES ($1, $2, $3, $4, $5, 'READY', $6)
+ON CONFLICT (server_id) DO UPDATE SET region = EXCLUDED.region,
+    build = EXCLUDED.build, protocol_version = EXCLUDED.protocol_version,
+    transport = EXCLUDED.transport, state = 'READY', updated_at = EXCLUDED.updated_at`
+
+const ClaimReadyServerSQL = `UPDATE game_servers SET state = 'ALLOCATED', updated_at = $5
+WHERE server_id = (
+    SELECT server_id FROM game_servers
+    WHERE state = 'READY' AND region = $1 AND build = $2
+      AND protocol_version = $3 AND transport = $4
+    ORDER BY server_id
+    LIMIT 1
+    FOR UPDATE SKIP LOCKED
+)
+RETURNING server_id`
+
+const InsertAllocationSQL = `INSERT INTO allocations
+    (allocation_id, match_id, server_id, region, build, protocol_version, transport, request_digest, state, allocated_at)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'ALLOCATED', $9)`
+
+const SelectAllocationSQL = `SELECT allocation_id, match_id, server_id, region, build,
+    protocol_version, transport, allocated_at, request_digest
+FROM allocations WHERE allocation_id = $1`
+
+func RegisterReadyServer(ctx context.Context, db *sql.DB, server domain.ReadyServer, now time.Time) error {
+	if db == nil || server.ServerID == "" || (server.Region != "EU" && server.Region != "NA") || server.Build == "" || server.Protocol <= 0 || (server.Transport != "enet" && server.Transport != "steam_sdr") || server.State != domain.ServerReady || now.IsZero() {
+		return fmt.Errorf("invalid ready server registration")
+	}
+	_, err := db.ExecContext(ctx, RegisterReadyServerSQL, server.ServerID, server.Region, server.Build, server.Protocol, server.Transport, now)
+	return err
+}
+
+func ClaimAllocation(ctx context.Context, db *sql.DB, request domain.AllocationRequest, now time.Time) (domain.Allocation, error) {
+	if db == nil || request.AllocationID == "" || request.MatchID == "" || (request.Region != "EU" && request.Region != "NA") || request.Build == "" || request.Protocol <= 0 || (request.Transport != "enet" && request.Transport != "steam_sdr") || now.IsZero() {
+		return domain.Allocation{}, domain.ErrAllocationInput
+	}
+	digest := allocationRequestDigest(request)
+	var allocation domain.Allocation
+	err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
+		var prior domain.Allocation
+		var priorDigest []byte
+		err := tx.QueryRowContext(ctx, SelectAllocationSQL, request.AllocationID).Scan(&prior.AllocationID, &prior.MatchID, &prior.ServerID, &prior.Region, &prior.Build, &prior.Protocol, &prior.Transport, &prior.AllocatedAt, &priorDigest)
+		if err == nil {
+			if !bytes.Equal(priorDigest, digest[:]) {
+				return domain.ErrConflict
+			}
+			allocation = prior
+			allocation.State = domain.ServerAllocated
+			return nil
+		}
+		if err != sql.ErrNoRows {
+			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 {
+				return domain.ErrNoCapacity
+			}
+			return err
+		}
+		allocation = domain.Allocation{AllocationID: request.AllocationID, MatchID: request.MatchID, ServerID: serverID, Region: request.Region, Build: request.Build, Protocol: request.Protocol, Transport: request.Transport, State: domain.ServerAllocated, AllocatedAt: now}
+		_, err = tx.ExecContext(ctx, InsertAllocationSQL, request.AllocationID, request.MatchID, serverID, request.Region, request.Build, request.Protocol, request.Transport, digest[:], now)
+		return err
+	})
+	return allocation, err
+}
+
+func allocationRequestDigest(request domain.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)))
+}

server/store/allocator_sql_test.go

@@ -0,0 +1,32 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+func TestAllocatorSQLClaimsAndAuditsCompatibleReadyServers(t *testing.T) {
+	for query, fragments := range map[string][]string{
+		RegisterReadyServerSQL: {"game_servers", "ON CONFLICT", "state = 'READY'"},
+		ClaimReadyServerSQL:    {"state = 'READY'", "region = $1", "protocol_version = $3", "FOR UPDATE SKIP LOCKED", "ORDER BY server_id"},
+		InsertAllocationSQL:    {"allocations", "request_digest", "state", "ALLOCATED"},
+	} {
+		for _, fragment := range fragments {
+			if !contains(query, fragment) {
+				t.Fatalf("query missing %q", fragment)
+			}
+		}
+	}
+}
+
+func TestClaimAllocationRejectsInvalidRequestsWithoutDatabase(t *testing.T) {
+	_, err := ClaimAllocation(nil, nil, domain.AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 1, Transport: "enet"}, time.Unix(1000, 0))
+	if err == nil {
+		t.Fatal("nil database accepted")
+	}
+	if _, err := ClaimAllocation(nil, nil, domain.AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 0, Transport: "enet"}, time.Unix(1000, 0)); err != domain.ErrAllocationInput {
+		t.Fatalf("invalid request err=%v", err)
+	}
+}