feat(multiplayer): lease allocating match claims

server/store/allocation_match_sql.go

@@ -0,0 +1,134 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+const AllocationClaimLease = time.Minute
+
+type PendingAllocation struct {
+	Request domain.AllocationRequest
+}
+
+const ClaimAllocatingMatchSQL = `WITH candidate AS (
+    SELECT match_id FROM matches
+    WHERE state = 'ALLOCATING' AND server_id IS NULL
+      AND (allocation_id IS NULL OR allocation_claimed_at <= $1)
+    ORDER BY created_at, match_id
+    LIMIT 1
+    FOR UPDATE SKIP LOCKED
+)
+UPDATE matches m
+SET allocation_id = 'allocation-' || candidate.match_id, allocation_claimed_at = $2
+FROM candidate
+WHERE m.match_id = candidate.match_id
+RETURNING m.match_id, m.region, m.protocol_version, m.allocation_id`
+
+const AllocatingMatchBuildSQL = `SELECT client_build
+FROM queue_tickets q
+JOIN match_participants mp ON mp.ticket_id = q.ticket_id AND mp.player_id = q.player_id
+WHERE mp.match_id = $1
+ORDER BY q.client_build`
+
+const BindAllocatedMatchSQL = `UPDATE matches
+SET server_id = $3
+WHERE match_id = $1 AND state = 'ALLOCATING' AND allocation_id = $2 AND server_id IS NULL
+  AND EXISTS (
+      SELECT 1 FROM allocations
+      WHERE allocation_id = $2 AND match_id = $1 AND server_id = $3 AND state = 'ALLOCATED'
+  )`
+
+const ReleaseAllocatedMatchClaimSQL = `UPDATE matches
+SET allocation_id = NULL, allocation_claimed_at = NULL
+WHERE match_id = $1 AND state = 'ALLOCATING' AND allocation_id = $2 AND server_id IS NULL`
+
+// ClaimAllocatingMatch returns one durable provider work item. The fixed
+// allocation ID is retained across a lease recovery, allowing every later
+// reconciliation step to reject a different server for the same match.
+func ClaimAllocatingMatch(ctx context.Context, db *sql.DB, transport string, now time.Time) (PendingAllocation, bool, error) {
+	if db == nil || (transport != "enet" && transport != "steam_sdr") || now.IsZero() {
+		return PendingAllocation{}, false, fmt.Errorf("invalid allocation claim arguments")
+	}
+	var item PendingAllocation
+	found := false
+	err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
+		var matchID, region string
+		var protocol int
+		var claimedID string
+		err := tx.QueryRowContext(ctx, ClaimAllocatingMatchSQL, now.Add(-AllocationClaimLease), now).Scan(&matchID, &region, &protocol, &claimedID)
+		if err == sql.ErrNoRows {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		rows, err := tx.QueryContext(ctx, AllocatingMatchBuildSQL, matchID)
+		if err != nil {
+			return err
+		}
+		defer rows.Close()
+		build := ""
+		for rows.Next() {
+			var candidate string
+			if err := rows.Scan(&candidate); err != nil {
+				return err
+			}
+			if build == "" {
+				build = candidate
+			} else if build != candidate {
+				return fmt.Errorf("allocating match has mixed client builds")
+			}
+		}
+		if err := rows.Err(); err != nil {
+			return err
+		}
+		if build == "" {
+			return fmt.Errorf("allocating match has no participants")
+		}
+		item.Request = domain.AllocationRequest{AllocationID: claimedID, MatchID: matchID, Region: region, Build: build, Protocol: protocol, Transport: transport}
+		found = true
+		return nil
+	})
+	return item, found, err
+}
+
+func BindAllocatedMatch(ctx context.Context, db *sql.DB, allocation domain.Allocation) error {
+	if db == nil || allocation.MatchID == "" || allocation.AllocationID == "" || allocation.ServerID == "" || allocation.State != domain.ServerAllocated {
+		return fmt.Errorf("invalid allocated match binding")
+	}
+	result, err := db.ExecContext(ctx, BindAllocatedMatchSQL, allocation.MatchID, allocation.AllocationID, allocation.ServerID)
+	if err != nil {
+		return err
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if changed != 1 {
+		return domain.ErrConflict
+	}
+	return nil
+}
+
+func ReleaseAllocatedMatchClaim(ctx context.Context, db *sql.DB, matchID, allocationID string) error {
+	if db == nil || matchID == "" || allocationID == "" {
+		return fmt.Errorf("invalid allocated match claim release")
+	}
+	result, err := db.ExecContext(ctx, ReleaseAllocatedMatchClaimSQL, matchID, allocationID)
+	if err != nil {
+		return err
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if changed != 1 {
+		return domain.ErrConflict
+	}
+	return nil
+}

server/store/allocation_match_sql_test.go

@@ -0,0 +1,44 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+func TestAllocationMatchClaimSQLFencesConcurrentWorkers(t *testing.T) {
+	checks := map[string][]string{
+		ClaimAllocatingMatchSQL:       {"FOR UPDATE SKIP LOCKED", "allocation_id = 'allocation-' || candidate.match_id", "allocation_claimed_at <= $1", "ORDER BY created_at, match_id"},
+		AllocatingMatchBuildSQL:       {"match_participants", "queue_tickets", "ORDER BY q.client_build"},
+		BindAllocatedMatchSQL:         {"allocation_id = $2", "server_id IS NULL", "SET server_id = $3", "FROM allocations"},
+		ReleaseAllocatedMatchClaimSQL: {"allocation_id = $2", "allocation_id = NULL", "allocation_claimed_at = NULL"},
+	}
+	for query, fragments := range checks {
+		for _, fragment := range fragments {
+			if !contains(query, fragment) {
+				t.Fatalf("query missing %q", fragment)
+			}
+		}
+	}
+}
+
+func TestAllocationMatchClaimRejectsInvalidArgumentsWithoutDatabase(t *testing.T) {
+	now := time.Unix(1_000, 0)
+	if _, _, err := ClaimAllocatingMatch(nil, nil, "enet", now); err == nil {
+		t.Fatal("nil database accepted")
+	}
+	if _, _, err := ClaimAllocatingMatch(nil, nil, "udp", now); err == nil {
+		t.Fatal("invalid transport accepted")
+	}
+	if _, _, err := ClaimAllocatingMatch(nil, nil, "enet", time.Time{}); err == nil {
+		t.Fatal("zero claim time accepted")
+	}
+	allocated := domain.Allocation{AllocationID: "allocation-match-1", MatchID: "match-1", ServerID: "server-1", State: domain.ServerAllocated}
+	if err := BindAllocatedMatch(nil, nil, allocated); err == nil {
+		t.Fatal("nil database accepted for bind")
+	}
+	if err := ReleaseAllocatedMatchClaim(nil, nil, "match-1", "allocation-match-1"); err == nil {
+		t.Fatal("nil database accepted for release")
+	}
+}

server/store/postgres_integration_test.go

@@ -143,6 +143,57 @@ func TestPostgreSQLAcceptedProposalPromotesOneAtomicAllocatingMatch(t *testing.T
 	}
 }
 
+func TestPostgreSQLAllocationMatchClaimLeaseAndBindFence(t *testing.T) {
+	db := openIntegrationPostgres(t)
+	applyIntegrationMigrations(t, db)
+	now := time.Now().UTC().Truncate(time.Microsecond)
+	ctx := context.Background()
+	for index, player := range []string{"allocation-match-a", "allocation-match-b"} {
+		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'ACCEPTED', 'build-1', 1, $3, $4)`, fmt.Sprintf("allocation-match-ticket-%d", index), player, now, now.Add(time.Minute)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version) VALUES ('allocation-match', 'casual', 'ALLOCATING', 'EU', 1)`); err != nil {
+		t.Fatal(err)
+	}
+	for index, player := range []string{"allocation-match-a", "allocation-match-b"} {
+		if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('allocation-match', $1, $2, $3, $4)`, player, fmt.Sprintf("allocation-match-ticket-%d", index), index*3, index); err != nil {
+			t.Fatal(err)
+		}
+	}
+	claim, found, err := ClaimAllocatingMatch(ctx, db, "enet", now)
+	if err != nil || !found || claim.Request != (domain.AllocationRequest{AllocationID: "allocation-allocation-match", MatchID: "allocation-match", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}) {
+		t.Fatalf("claim=%+v found=%t err=%v", claim, found, err)
+	}
+	if err := ReleaseAllocatedMatchClaim(ctx, db, claim.Request.MatchID, "different-allocation"); err != domain.ErrConflict {
+		t.Fatalf("wrong-claim release err=%v", err)
+	}
+	if err := ReleaseAllocatedMatchClaim(ctx, db, claim.Request.MatchID, claim.Request.AllocationID); err != nil {
+		t.Fatalf("release claim: %v", err)
+	}
+	reclaimed, found, err := ClaimAllocatingMatch(ctx, db, "enet", now.Add(time.Second))
+	if err != nil || !found || reclaimed.Request.AllocationID != claim.Request.AllocationID {
+		t.Fatalf("reclaimed=%+v found=%t err=%v", reclaimed, found, err)
+	}
+	server := domain.ReadyServer{ServerID: "allocation-server", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}
+	if err := RegisterReadyServer(ctx, db, server, now); err != nil {
+		t.Fatalf("register allocation server: %v", err)
+	}
+	allocation, err := ClaimAllocation(ctx, db, reclaimed.Request, now.Add(time.Second))
+	if err != nil {
+		t.Fatalf("record provider allocation: %v", err)
+	}
+	if err := BindAllocatedMatch(ctx, db, allocation); err != nil {
+		t.Fatalf("bind allocation: %v", err)
+	}
+	if _, found, err := ClaimAllocatingMatch(ctx, db, "enet", now.Add(2*time.Second)); err != nil || found {
+		t.Fatalf("bound match re-claimed found=%t err=%v", found, err)
+	}
+}
+
 func TestPostgreSQLQueueAdapterAgainstRealDatabase(t *testing.T) {
 	db := openIntegrationPostgres(t)
 	applyIntegrationMigrations(t, db)