package store import ( "bytes" "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` // FindProviderAllocation verifies whether a recovered lease has already // crossed the durable provider boundary. A worker can then bind it without // issuing a second external allocation request after a crash. func FindProviderAllocation(ctx context.Context, db *sql.DB, request domain.AllocationRequest) (domain.Allocation, bool, error) { if db == nil || request.AllocationID == "" || request.MatchID == "" || request.Region == "" || request.Build == "" || request.Protocol <= 0 || (request.Transport != "enet" && request.Transport != "steam_sdr") { return domain.Allocation{}, false, fmt.Errorf("invalid provider allocation lookup") } var allocation domain.Allocation var digest []byte err := db.QueryRowContext(ctx, SelectAllocationSQL, request.AllocationID).Scan(&allocation.AllocationID, &allocation.MatchID, &allocation.ServerID, &allocation.Region, &allocation.Build, &allocation.Protocol, &allocation.Transport, &allocation.AllocatedAt, &digest) if err == sql.ErrNoRows { return domain.Allocation{}, false, nil } if err != nil { return domain.Allocation{}, false, err } want := allocationRequestDigest(request) if !bytes.Equal(digest, want[:]) || allocation.MatchID != request.MatchID || allocation.Region != request.Region || allocation.Build != request.Build || allocation.Protocol != request.Protocol || allocation.Transport != request.Transport { return domain.Allocation{}, false, domain.ErrConflict } allocation.State = domain.ServerAllocated return allocation, true, nil } // 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, ®ion, &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 }