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
Files
f6a87463c5163ca3cd45f973c36a70f81a188942
CosmicClash/server/store/postgres_integration_test.go
T
Josh Creek f6a87463c5 fix(server): load authoritative ratings into ranked candidates 
The candidate projection selected only from queue_tickets and its scan
never set Candidate.Rating, so every PostgreSQL-sourced ranked candidate
arrived with Go's zero value. Rating tolerance, selection scoring and
team partitioning all read that field, so ranked matchmaking treated a
900-rated player as identical to a 2100-rated one. Unit tests missed it
because they construct candidates with ratings already populated.

Join the ratings table, defaulting to domain.GlickoInitialRating for a
player with no ratings row yet -- a genuinely new profile, matching the
column default.

Fix the same defect on the Redis path too, which is reached differently:
the projection is seeded from the candidate CreateQueueTicket builds,
not from the candidate query, and that candidate also left Rating unset.
Resolve the rating inside the enqueue transaction so both projections
agree on one authoritative value. The rating is never client-supplied.

Add a store-backed test with deliberately distant ratings (900 vs 2100)
plus an unrated player, asserting both projections and that the spread
survives. Verified it fails without the fix.
2026-09-05 10:19:23 +01:00

1919 lines
95 KiB
Go
Raw Blame History

//go:build integration
package store
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"github.com/cosmic-clash/cosmic-clash/server/domain"
"github.com/cosmic-clash/cosmic-clash/server/migrations"
_ "github.com/jackc/pgx/v5/stdlib"
)
// This binary is deliberately opt-in. It requires a disposable PostgreSQL
// instance supplied by scripts/run_postgres_integration.sh.
func openIntegrationPostgres(t *testing.T) *sql.DB {
t.Helper()
dsn := os.Getenv("COSMIC_CLASH_POSTGRES_DSN")
if dsn == "" {
t.Skip("COSMIC_CLASH_POSTGRES_DSN is not set")
}
db, err := sql.Open("pgx", dsn)
if err != nil {
t.Fatalf("open PostgreSQL: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := db.PingContext(ctx); err != nil {
db.Close()
t.Fatalf("ping PostgreSQL: %v", err)
}
t.Cleanup(func() { db.Close() })
return db
}
func applyIntegrationMigrations(t *testing.T, db *sql.DB) {
t.Helper()
if _, err := db.ExecContext(context.Background(), `DROP TABLE IF EXISTS schema_migrations, allocation_quotas, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil {
t.Fatalf("reset PostgreSQL schema: %v", err)
}
if err := migrations.Apply(context.Background(), db, filepath.Join("..", "migrations")); err != nil {
t.Fatalf("apply migrations: %v", err)
}
}
func TestPostgreSQLAllocatorClaimReplayAndCapacityFence(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
servers := []domain.ReadyServer{
{ServerID: "allocator-server-b", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady},
{ServerID: "allocator-server-a", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady},
}
for _, server := range servers {
if err := RegisterReadyServer(ctx, db, server, now); err != nil {
t.Fatal(err)
}
}
request := domain.AllocationRequest{AllocationID: "allocation-integration-1", MatchID: "match-integration-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
allocation, err := ClaimAllocation(ctx, db, request, now)
if err != nil {
t.Fatalf("claim: %v", err)
}
if allocation.ServerID != "allocator-server-a" || allocation.State != domain.ServerAllocated {
t.Fatalf("allocation=%+v", allocation)
}
if err := RegisterReadyServer(ctx, db, servers[1], now.Add(time.Second)); err != nil {
t.Fatalf("stale Ready projection: %v", err)
}
var lifecycle string
if err := db.QueryRowContext(ctx, `SELECT state FROM game_servers WHERE server_id = 'allocator-server-a'`).Scan(&lifecycle); err != nil || lifecycle != "ALLOCATED" {
t.Fatalf("stale Ready projection reopened allocation state=%q err=%v", lifecycle, err)
}
replay, err := ClaimAllocation(ctx, db, request, now.Add(time.Second))
if err != nil || replay.ServerID != allocation.ServerID || !replay.AllocatedAt.Equal(allocation.AllocatedAt) {
t.Fatalf("replay=%+v err=%v", replay, err)
}
conflict := request
conflict.MatchID = "match-integration-other"
if _, err := ClaimAllocation(ctx, db, conflict, now); err != domain.ErrConflict {
t.Fatalf("conflicting replay err=%v", err)
}
second := request
second.AllocationID = "allocation-integration-2"
second.MatchID = "match-integration-2"
if _, err := ClaimAllocation(ctx, db, second, now); err != nil {
t.Fatalf("second claim: %v", err)
}
third := second
third.AllocationID = "allocation-integration-3"
third.MatchID = "match-integration-3"
if _, err := ClaimAllocation(ctx, db, third, now); err != domain.ErrNoCapacity {
t.Fatalf("capacity err=%v", err)
}
}
func TestPostgreSQLSharedAllocationQuotaFencesClaimsAndReplays(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if err := SetAllocationQuota(ctx, db, "EU", 1, time.Minute, now); err != nil {
t.Fatalf("set quota: %v", err)
}
for _, server := range []domain.ReadyServer{
{ServerID: "quota-server-a", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady},
{ServerID: "quota-server-b", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady},
} {
if err := RegisterReadyServer(ctx, db, server, now); err != nil {
t.Fatal(err)
}
}
first := domain.AllocationRequest{AllocationID: "quota-allocation-1", MatchID: "quota-match-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
if _, err := ClaimAllocation(ctx, db, first, now); err != nil {
t.Fatalf("first claim: %v", err)
}
if _, err := ClaimAllocation(ctx, db, first, now.Add(time.Second)); err != nil {
t.Fatalf("idempotent replay was fenced: %v", err)
}
second := domain.AllocationRequest{AllocationID: "quota-allocation-2", MatchID: "quota-match-2", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
if _, err := ClaimAllocation(ctx, db, second, now.Add(2*time.Second)); !errors.Is(err, ErrAllocationQuotaExceeded) {
t.Fatalf("second claim err=%v, want shared quota fence", err)
}
if err := SetAllocationQuota(ctx, db, "EU", 1, time.Minute, now.Add(time.Minute)); err != nil {
t.Fatalf("reset quota: %v", err)
}
if _, err := ClaimAllocation(ctx, db, second, now.Add(time.Minute)); err != nil {
t.Fatalf("claim after quota window reset: %v", err)
}
}
// TestPostgreSQLConcurrentAllocationClaimNeverDoubleBooksAReadyServer is the
// live counterpart to TestPostgreSQLAllocatorClaimReplayAndCapacityFence: that
// test claims strictly one request at a time, so it cannot show what happens
// when two allocator replicas race for the same compatible capacity, which is
// exactly the scenario 8.30's "bounded cross-replica retry" is about. Register
// fewer Ready servers than concurrent requests and fire them all at once;
// exactly as many must win as there was capacity, each winner must get a
// distinct server, and every loser must fail with ErrNoCapacity rather than a
// raw serialization error, a duplicate claim, or a hang.
func TestPostgreSQLConcurrentAllocationClaimNeverDoubleBooksAReadyServer(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
const capacity = 3
const contenders = 6
for i := 0; i < capacity; i++ {
server := domain.ReadyServer{ServerID: fmt.Sprintf("race-server-%d", i), Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}
if err := RegisterReadyServer(ctx, db, server, now); err != nil {
t.Fatalf("register %s: %v", server.ServerID, err)
}
}
var wg sync.WaitGroup
allocations := make([]domain.Allocation, contenders)
errs := make([]error, contenders)
wg.Add(contenders)
for i := 0; i < contenders; i++ {
go func(i int) {
defer wg.Done()
request := domain.AllocationRequest{AllocationID: fmt.Sprintf("race-allocation-%d", i), MatchID: fmt.Sprintf("race-match-%d", i), Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
allocations[i], errs[i] = ClaimAllocation(ctx, db, request, now)
}(i)
}
wg.Wait()
wonServers := map[string]int{}
won, lost := 0, 0
for i, err := range errs {
switch {
case err == nil:
won++
if allocations[i].ServerID == "" {
t.Fatalf("claim %d succeeded with no server", i)
}
wonServers[allocations[i].ServerID]++
case errors.Is(err, domain.ErrNoCapacity):
lost++
default:
t.Fatalf("claim %d failed with unexpected error: %v", i, err)
}
}
if won != capacity || lost != contenders-capacity {
t.Fatalf("won=%d lost=%d, want won=%d lost=%d", won, lost, capacity, contenders-capacity)
}
if len(wonServers) != capacity {
t.Fatalf("expected %d distinct servers claimed, got %d: %v", capacity, len(wonServers), wonServers)
}
for server, count := range wonServers {
if count != 1 {
t.Fatalf("server %s was claimed %d times", server, count)
}
}
var allocatedCount int
if err := db.QueryRowContext(ctx, `SELECT count(*) FROM game_servers WHERE state = 'ALLOCATED'`).Scan(&allocatedCount); err != nil {
t.Fatal(err)
}
if allocatedCount != capacity {
t.Fatalf("durable ALLOCATED server count = %d, want %d", allocatedCount, capacity)
}
}
func TestPostgreSQLAcceptedProposalPromotesOneAtomicAllocatingMatch(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"promote-a", "promote-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
for index, player := range []string{"promote-a", "promote-b"} {
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', 'PROPOSED', 'build-1', 1, $3, $4)`, fmt.Sprintf("promote-ticket-%d", index), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
if _, err := db.ExecContext(ctx, `INSERT INTO proposals (proposal_id, playlist, state, expires_at, revision) VALUES ('promote-proposal', 'casual', 'ACCEPTED', $1, 2)`, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
for index, player := range []string{"promote-a", "promote-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO proposal_participants (proposal_id, player_id, ticket_id, response) VALUES ('promote-proposal', $1, $2, 'ACCEPTED')`, player, fmt.Sprintf("promote-ticket-%d", index)); err != nil {
t.Fatal(err)
}
}
plan := AcceptedMatchPlan{MatchID: "promote-match", ProposalID: "promote-proposal", Region: "EU", Protocol: 1, Players: []MatchPlayer{{PlayerID: "promote-a", Team: 0, Slot: 0}, {PlayerID: "promote-b", Team: 1, Slot: 3}}}
if err := CreateMatchFromAcceptedProposal(ctx, db, plan, now); err != nil {
t.Fatalf("promote accepted proposal: %v", err)
}
var state string
if err := db.QueryRowContext(ctx, `SELECT state FROM matches WHERE match_id = 'promote-match'`).Scan(&state); err != nil || state != "ALLOCATING" {
t.Fatalf("match state=%q err=%v", state, err)
}
var acceptedTickets, participantCount int
if err := db.QueryRowContext(ctx, `SELECT count(*) FROM queue_tickets WHERE ticket_id LIKE 'promote-ticket-%' AND state = 'ACCEPTED'`).Scan(&acceptedTickets); err != nil || acceptedTickets != 2 {
t.Fatalf("accepted tickets=%d err=%v", acceptedTickets, err)
}
if err := db.QueryRowContext(ctx, `SELECT count(*) FROM match_participants WHERE match_id = 'promote-match'`).Scan(&participantCount); err != nil || participantCount != 2 {
t.Fatalf("participants=%d err=%v", participantCount, err)
}
if err := CreateMatchFromAcceptedProposal(ctx, db, plan, now.Add(time.Second)); err != nil {
t.Fatalf("identical match promotion replay: %v", err)
}
if _, err := db.ExecContext(ctx, `UPDATE matches SET state = 'LIVE' WHERE match_id = 'promote-match'`); err != nil {
t.Fatal(err)
}
if err := CreateMatchFromAcceptedProposal(ctx, db, plan, now.Add(2*time.Second)); err != nil {
t.Fatalf("promotion replay after match lifecycle advanced: %v", err)
}
conflict := plan
conflict.Players = append([]MatchPlayer(nil), plan.Players...)
conflict.Players[1].Slot = 4
if err := CreateMatchFromAcceptedProposal(ctx, db, conflict, now.Add(3*time.Second)); err == nil {
t.Fatal("conflicting match promotion replay was accepted")
}
}
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", Playlist: domain.Casual, 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)
}
recoveredTicket, err := GetQueueTicket(ctx, db, "allocation-match-a", "allocation-match-ticket-0", now.Add(2*time.Second))
if err != nil || recoveredTicket.MatchID != "allocation-match" {
t.Fatalf("recovered ticket match=%q err=%v", recoveredTicket.MatchID, err)
}
var allocatingTickets int
if err := db.QueryRowContext(ctx, `SELECT count(*) FROM queue_tickets WHERE ticket_id LIKE 'allocation-match-ticket-%' AND state = 'ALLOCATING'`).Scan(&allocatingTickets); err != nil || allocatingTickets != 2 {
t.Fatalf("allocating tickets=%d err=%v", allocatingTickets, err)
}
var eventType string
var eventPayload []byte
if err := db.QueryRowContext(ctx, `SELECT event_type, payload FROM outbox WHERE aggregate_id = 'allocation-match' AND event_type = 'state_changed'`).Scan(&eventType, &eventPayload); err != nil {
t.Fatalf("allocation outbox event: %v", err)
}
var event struct {
State string `json:"state"`
PlayerIDs []string `json:"player_ids"`
}
if err := json.Unmarshal(eventPayload, &event); err != nil {
t.Fatalf("decode allocation outbox event: %v", err)
}
players := make(map[string]bool, len(event.PlayerIDs))
for _, playerID := range event.PlayerIDs {
players[playerID] = true
}
if eventType != "state_changed" || event.State != "ALLOCATING" || !players["allocation-match-a"] || !players["allocation-match-b"] {
t.Fatalf("allocation outbox event = %s", eventPayload)
}
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)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('integration-player', 'integration-steam')`); err != nil {
t.Fatal(err)
}
spec := domain.QueueSpec{Playlist: domain.Casual, ClientBuild: "integration-build", ProtocolVersion: 1}
ticket, err := CreateQueueTicket(ctx, db, "integration-ticket", "integration-player", "integration-create-0001", spec, now)
if err != nil {
t.Fatalf("create queue ticket: %v", err)
}
if ticket.State != domain.Queued || ticket.Revision != 0 {
t.Fatalf("unexpected ticket: %+v", ticket)
}
replay, err := CreateQueueTicket(ctx, db, "integration-ticket", "integration-player", "integration-create-0001", spec, now.Add(time.Second))
if err != nil {
t.Fatalf("idempotent queue replay: %v", err)
}
if replay.TicketID != ticket.TicketID || !replay.ExpiresAt.Equal(ticket.ExpiresAt) {
t.Fatalf("replay changed durable result: %+v vs %+v", replay, ticket)
}
if _, err := CreateQueueTicket(ctx, db, "integration-ticket-2", "integration-player", "integration-create-0002", spec, now); err == nil {
t.Fatal("second active player ticket was accepted")
}
if _, err := GetQueueTicket(ctx, db, "integration-player", "integration-ticket", now); err != nil {
t.Fatalf("owner recovery: %v", err)
}
if _, err := GetQueueTicket(ctx, db, "other-player", "integration-ticket", now); err == nil {
t.Fatal("non-owner recovered queue ticket")
}
}
func TestPostgreSQLQueueHeartbeatAndCancelAreRevisionFenced(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('heartbeat-player', 'heartbeat-steam')`); err != nil {
t.Fatal(err)
}
spec := domain.QueueSpec{Playlist: domain.Ranked, ClientBuild: "integration-build", ProtocolVersion: 1}
if _, err := CreateQueueTicket(ctx, db, "heartbeat-ticket", "heartbeat-player", "heartbeat-create-0001", spec, now); err != nil {
t.Fatal(err)
}
heartbeat, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000001", 0, now.Add(5*time.Second))
if err != nil {
t.Fatalf("heartbeat: %v", err)
}
if heartbeat.Revision != 1 || !heartbeat.ExpiresAt.Equal(now.Add(35*time.Second)) {
t.Fatalf("unexpected heartbeat result: %+v", heartbeat)
}
if _, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000002", 0, now.Add(6*time.Second)); !errors.Is(err, domain.ErrStaleRevision) {
t.Fatalf("stale heartbeat error = %v, want ErrStaleRevision", err)
}
cancelled, err := CancelQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000003", 1, now.Add(7*time.Second))
if err != nil {
t.Fatalf("cancel: %v", err)
}
if cancelled.State != domain.Cancelled || cancelled.Revision != 2 {
t.Fatalf("unexpected cancellation result: %+v", cancelled)
}
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('live-cancel-player', 'live-cancel-steam')`); 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 ('live-cancel-ticket', 'live-cancel-player', 'ranked', 'LIVE', 'integration-build', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
if _, err := CancelQueueTicket(ctx, db, "live-cancel-player", "live-cancel-ticket", "live-cancel-op-0001", 0, now.Add(8*time.Second)); !errors.Is(err, domain.ErrConflict) {
t.Fatalf("live cancellation error = %v, want ErrConflict", err)
}
var liveState string
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'live-cancel-ticket'`).Scan(&liveState); err != nil {
t.Fatal(err)
}
if liveState != "LIVE" {
t.Fatalf("live ticket state = %s after cancellation attempt", liveState)
}
}
// TestPostgreSQLConcurrentQueueHeartbeatIsRevisionFencedUnderRealRace is the
// live counterpart to the sequential stale-heartbeat check above: calling the
// second heartbeat only after the first has already committed proves the SQL
// predicate is correct, but not that it actually fences two requests that
// genuinely overlap at the database. A client can legitimately double-send a
// heartbeat (a slow response triggering a client-side retry, or two tabs/
// processes for the same player), and both requests can reach PostgreSQL
// truly concurrently -- this races that directly.
func TestPostgreSQLConcurrentQueueHeartbeatIsRevisionFencedUnderRealRace(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('race-heartbeat-player', 'race-heartbeat-steam')`); err != nil {
t.Fatal(err)
}
spec := domain.QueueSpec{Playlist: domain.Casual, ClientBuild: "integration-build", ProtocolVersion: 1}
if _, err := CreateQueueTicket(ctx, db, "race-heartbeat-ticket", "race-heartbeat-player", "race-heartbeat-create-01", spec, now); err != nil {
t.Fatal(err)
}
const attempts = 5
var wg sync.WaitGroup
tickets := make([]domain.QueueTicket, attempts)
errs := make([]error, attempts)
wg.Add(attempts)
for i := 0; i < attempts; i++ {
go func(i int) {
defer wg.Done()
tickets[i], errs[i] = HeartbeatQueueTicket(ctx, db, "race-heartbeat-player", "race-heartbeat-ticket", fmt.Sprintf("race-heartbeat-op-%08d", i), 0, now.Add(time.Duration(i)*time.Millisecond))
}(i)
}
wg.Wait()
won, lost := 0, 0
for i, err := range errs {
if err == nil {
won++
if tickets[i].Revision != 1 {
t.Fatalf("winning heartbeat %d landed at revision %d, want 1", i, tickets[i].Revision)
}
continue
}
lost++
}
if won != 1 {
t.Fatalf("won=%d, want exactly 1 of %d concurrent heartbeats at the same expected revision to win", won, attempts)
}
if lost != attempts-1 {
t.Fatalf("lost=%d, want %d", lost, attempts-1)
}
var revision uint64
if err := db.QueryRow(`SELECT revision FROM queue_tickets WHERE ticket_id = 'race-heartbeat-ticket'`).Scan(&revision); err != nil {
t.Fatal(err)
}
if revision != 1 {
t.Fatalf("durable revision = %d, want exactly 1 (a stale winner re-applying would leave it higher)", revision)
}
}
func TestPostgreSQLAssignmentPersistenceIsPlayerScopedAndExpiryBound(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"assignment-player", "assignment-other"} {
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 ('assignment-ticket', 'assignment-player', 'casual', 'ASSIGNED', 'integration-build', 1, $1, $2)`, 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, server_id, initial_connect_ready_at) VALUES ('assignment-match', 'casual', 'ASSIGNED', 'EU', 1, 'assignment-server', $1)`, now); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('assignment-match', 'assignment-player', 'assignment-ticket', 0, 0)`); err != nil {
t.Fatal(err)
}
assignment := DurableAssignment{MatchID: "assignment-match", PlayerID: "assignment-player", AllocationID: "allocation-1", ServerID: "assignment-server", Slot: 0, Region: "EU", ClientBuild: "integration-build", ProtocolVersion: 1, Transport: "enet", Endpoint: "127.0.0.1:7777", JoinAuthorisation: "join-token", ManifestDigest: []byte("manifest"), ExpiresAt: now.Add(time.Minute), Revision: 1}
if err := SaveAssignment(ctx, db, assignment); err != nil {
t.Fatalf("save assignment: %v", err)
}
got, err := GetAssignment(ctx, db, assignment.PlayerID, assignment.MatchID, now)
if err != nil {
t.Fatalf("recover assignment: %v", err)
}
if got.JoinAuthorisation != assignment.JoinAuthorisation || got.Slot != assignment.Slot {
t.Fatalf("assignment changed on round trip: %+v", got)
}
if _, err := GetAssignment(ctx, db, "assignment-other", assignment.MatchID, now); err == nil {
t.Fatal("non-owner recovered assignment")
}
if _, err := GetAssignment(ctx, db, assignment.PlayerID, assignment.MatchID, now.Add(2*time.Minute)); err == nil {
t.Fatal("expired assignment was recovered")
}
}
func TestPostgreSQLVerifiedAssignmentRosterMustMatchDurableParticipants(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for index, player := range []string{"roster-player-a", "roster-player-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $2)`, player, "steam-"+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', 'ALLOCATING', 'build-1', 1, $3, $4)`, fmt.Sprintf("roster-ticket-%d", index), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
if _, err := db.ExecContext(ctx, `INSERT INTO game_servers (server_id, region, build, protocol_version, transport, state, updated_at) VALUES ('roster-server', 'EU', 'build-1', 1, 'enet', 'ALLOCATED', $1)`, now); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO allocations (allocation_id, match_id, server_id, region, build, protocol_version, transport, request_digest, state, allocated_at) VALUES ('roster-allocation', 'roster-match', 'roster-server', 'EU', 'build-1', 1, 'enet', 'digest', 'ALLOCATED', $1)`, now); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, allocation_id, allocation_claimed_at) VALUES ('roster-match', 'casual', 'ALLOCATING', 'EU', 1, 'roster-server', 'roster-allocation', $1)`, now); err != nil {
t.Fatal(err)
}
for index, player := range []string{"roster-player-a", "roster-player-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('roster-match', $1, $2, $3, $4)`, player, fmt.Sprintf("roster-ticket-%d", index), index*3, index); err != nil {
t.Fatal(err)
}
}
allocation := domain.Allocation{AllocationID: "roster-allocation", MatchID: "roster-match", ServerID: "roster-server", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerAllocated, AllocatedAt: now}
assignment := domain.Assignment{Allocation: allocation, Manifest: domain.AllocationManifest{AllocationID: allocation.AllocationID, MatchID: allocation.MatchID, ServerID: allocation.ServerID, Region: allocation.Region, Build: allocation.Build, Protocol: allocation.Protocol, Transport: allocation.Transport, RosterDigest: "roster-digest"}, Endpoint: "127.0.0.1:7777"}
roster := []domain.SignedJoinAuthorisation{
{Authorisation: domain.JoinAuthorisation{MatchID: "roster-match", ServerID: "roster-server", PlayerID: "roster-player-a", SteamID: "steam-roster-player-a", Slot: 0, Team: 0, Protocol: "1", Generation: 1, ExpiresAt: now.Add(time.Minute)}, Signature: []byte("sig-a")},
{Authorisation: domain.JoinAuthorisation{MatchID: "roster-match", ServerID: "roster-server", PlayerID: "roster-player-b", SteamID: "steam-roster-player-b", Slot: 3, Team: 1, Protocol: "1", Generation: 1, ExpiresAt: now.Add(time.Minute)}, Signature: []byte("sig-b")},
}
verify := func([]byte, []byte) bool { return true }
if err := SaveVerifiedAssignmentRoster(ctx, db, assignment, roster[:1], verify); err == nil {
t.Fatal("partial signed roster was accepted")
}
var count int
if err := db.QueryRow(`SELECT count(*) FROM assignments WHERE match_id = 'roster-match'`).Scan(&count); err != nil || count != 0 {
t.Fatalf("partial roster persisted rows=%d err=%v", count, err)
}
if err := SaveVerifiedAssignmentRoster(ctx, db, assignment, roster, verify); err != nil {
t.Fatalf("complete durable roster rejected: %v", err)
}
if err := db.QueryRow(`SELECT count(*) FROM assignments WHERE match_id = 'roster-match'`).Scan(&count); err != nil || count != 2 {
t.Fatalf("complete roster rows=%d err=%v", count, err)
}
forged := append([]domain.SignedJoinAuthorisation(nil), roster...)
forged[1].Authorisation.SteamID = "steam-other"
if err := SaveVerifiedAssignmentRoster(ctx, db, assignment, forged, verify); err == nil {
t.Fatal("signed roster with wrong durable Steam identity was accepted")
}
}
func TestPostgreSQLConnectionReceiptsStartCompleteRelaxedCasualRoster(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for i := 0; i < 2; i++ {
playerID := fmt.Sprintf("connect-player-%d", i)
ticketID := fmt.Sprintf("connect-ticket-%d", i)
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, playerID); 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', 'ASSIGNMENT_READY', 'integration-build', 1, $3, $4)`, ticketID, playerID, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
if _, err := db.ExecContext(ctx, `INSERT INTO game_servers (server_id, region, build, protocol_version, transport, state) VALUES ('connect-server', 'EU', 'integration-build', 1, 'enet', 'ALLOCATED')`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version) VALUES ('connect-match', 'casual', 'ALLOCATING', 'EU', 1)`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO allocations (allocation_id, match_id, server_id, region, build, protocol_version, transport, request_digest, state, allocated_at) VALUES ('connect-allocation', 'connect-match', 'connect-server', 'EU', 'integration-build', 1, 'enet', $1, 'ALLOCATED', $2)`, []byte("request"), now); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `UPDATE matches SET state = 'ASSIGNMENT_READY', server_id = 'connect-server', allocation_id = 'connect-allocation', allocation_claimed_at = $1, initial_connect_ready_at = $1 WHERE match_id = 'connect-match'`, now); err != nil {
t.Fatal(err)
}
for i := 0; i < 2; i++ {
playerID := fmt.Sprintf("connect-player-%d", i)
ticketID := fmt.Sprintf("connect-ticket-%d", i)
slot := i * 3
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('connect-match', $1, $2, $3, $4)`, playerID, ticketID, slot, i); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO assignments (match_id, player_id, allocation_id, server_id, slot, region, client_build, protocol_version, transport, endpoint, join_authorisation, manifest_digest, expires_at) VALUES ('connect-match', $1, 'connect-allocation', 'connect-server', $2, 'EU', 'integration-build', 1, 'enet', '127.0.0.1:7777', 'join-token', $3, $4)`, playerID, slot, []byte("manifest"), now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
binding := domain.WorkloadBinding{AllocationID: "connect-allocation", MatchID: "connect-match", ServerID: "connect-server"}
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 0, "connect-receipt-key-0000", now); err != nil || generation != 1 {
t.Fatalf("first receipt: %v", err)
}
if _, err := ClaimPlayerConnection(ctx, db, domain.WorkloadBinding{AllocationID: "forged-allocation", MatchID: "connect-match", ServerID: "connect-server"}, "connect-player-1", 0, "connect-receipt-key-forged", now); !errors.Is(err, domain.ErrConflict) {
t.Fatalf("forged binding err=%v, want conflict", err)
}
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-1", 0, "connect-receipt-key-0001", now); err != nil || generation != 1 {
t.Fatalf("second receipt: %v", err)
}
reconciled, err := ReconcileInitialConnect(ctx, db, now.Add(time.Second), 10)
if err != nil || reconciled != 1 {
t.Fatalf("reconcile count=%d err=%v", reconciled, err)
}
var state string
if err := db.QueryRowContext(ctx, `SELECT state FROM matches WHERE match_id = 'connect-match'`).Scan(&state); err != nil || state != string(domain.Live) {
t.Fatalf("match state=%q err=%v", state, err)
}
var liveTickets int
if err := db.QueryRowContext(ctx, `SELECT count(*) FROM queue_tickets WHERE state = 'LIVE'`).Scan(&liveTickets); err != nil || liveTickets != 2 {
t.Fatalf("live tickets=%d err=%v", liveTickets, err)
}
// A lost 204 can be retried after assignment expiry because the exact
// durable receipt is replayed before checking the now-expired assignment.
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 0, "connect-receipt-key-0000", now.Add(2*time.Minute)); err != nil || generation != 1 {
t.Fatalf("durable receipt replay: %v", err)
}
if _, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 1, "connect-active-duplicate", now.Add(2*time.Second)); !errors.Is(err, domain.ErrConflict) {
t.Fatalf("active duplicate err=%v, want conflict", err)
}
if err := RecordPlayerDisconnected(ctx, db, binding, "connect-player-0", 1, "disconnect-receipt-0000", now.Add(3*time.Second)); err != nil {
t.Fatalf("disconnect receipt: %v", err)
}
if _, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 0, "connect-receipt-key-0000", now.Add(4*time.Second)); !errors.Is(err, domain.ErrConflict) {
t.Fatalf("stale connect replay err=%v, want conflict", err)
}
if generation, err := ClaimPlayerConnection(ctx, db, binding, "connect-player-0", 0, "reconnect-receipt-0000", now.Add(63*time.Second)); err != nil || generation != 2 {
t.Fatalf("grace-boundary reconnect generation=%d err=%v", generation, err)
}
if err := RecordPlayerDisconnected(ctx, db, binding, "connect-player-0", 1, "stale-disconnect-0000", now.Add(64*time.Second)); !errors.Is(err, domain.ErrConflict) {
t.Fatalf("stale disconnect err=%v, want conflict", err)
}
}
func TestPostgreSQLLiveReconnectGraceExpiryPersistsAbandonmentWithoutReleasingResultRoster(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Microsecond)
for _, playerID := range []string{"live-abandon-player", "live-present-player"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, playerID); 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, 'ranked', 'LIVE', 'integration-build', 1, $3, $4)`, "live-abandon-ticket-"+playerID, playerID, now, now.Add(time.Hour)); err != nil {
t.Fatal(err)
}
}
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, arena_path) VALUES ('live-abandon-match', 'ranked', 'LIVE', 'EU', 1, 'live-abandon-server', 'res://scenes/arena_01.tscn')`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team, connection_generation, connected_at, disconnected_at) VALUES
('live-abandon-match', 'live-abandon-player', 'live-abandon-ticket-live-abandon-player', 0, 0, 1, $1, $2),
('live-abandon-match', 'live-present-player', 'live-abandon-ticket-live-present-player', 3, 1, 1, $1, NULL)`, now.Add(-2*time.Minute), now.Add(-domain.RankedReconnectGrace-time.Nanosecond)); err != nil {
t.Fatal(err)
}
reconciled, err := ReconcileLiveAbandonments(ctx, db, now, 10)
if err != nil || reconciled != 1 {
t.Fatalf("reconciled=%d err=%v", reconciled, err)
}
var active bool
var abandonedAt sql.NullTime
if err := db.QueryRowContext(ctx, `SELECT participation_active, abandoned_at FROM match_participants WHERE match_id = 'live-abandon-match' AND player_id = 'live-abandon-player'`).Scan(&active, &abandonedAt); err != nil || !active || !abandonedAt.Valid || !abandonedAt.Time.Equal(now) {
t.Fatalf("participant active=%t abandoned=%v err=%v", active, abandonedAt, err)
}
var ticketState string
if err := db.QueryRowContext(ctx, `SELECT state FROM queue_tickets WHERE ticket_id = 'live-abandon-ticket-live-abandon-player'`).Scan(&ticketState); err != nil || ticketState != "LIVE" {
t.Fatalf("ticket state=%q err=%v", ticketState, err)
}
var endsAt time.Time
if err := db.QueryRowContext(ctx, `SELECT ends_at FROM penalties WHERE player_id = 'live-abandon-player' AND kind = 'MATCH_ABANDONED'`).Scan(&endsAt); err != nil || !endsAt.Equal(now.Add(5*time.Minute)) {
t.Fatalf("penalty ends=%v err=%v", endsAt, err)
}
var outboxCount, revision int
if err := db.QueryRowContext(ctx, `SELECT count(*) FROM outbox WHERE aggregate_id = 'live-abandon-match' AND event_type = 'state_changed'`).Scan(&outboxCount); err != nil || outboxCount != 1 {
t.Fatalf("outbox=%d err=%v", outboxCount, err)
}
if err := db.QueryRowContext(ctx, `SELECT revision FROM matches WHERE match_id = 'live-abandon-match'`).Scan(&revision); err != nil || revision != 1 {
t.Fatalf("revision=%d err=%v", revision, err)
}
if reconciled, err = ReconcileLiveAbandonments(ctx, db, now.Add(time.Minute), 10); err != nil || reconciled != 0 {
t.Fatalf("replay reconciled=%d err=%v", reconciled, err)
}
}
func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"proposal-player-a", "proposal-player-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
for i, player := range []string{"proposal-player-a", "proposal-player-b"} {
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', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("proposal-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
proposal, err := domain.NewProposal("proposal-integration", domain.Casual, []string{"proposal-player-a", "proposal-player-b"}, now)
if err != nil {
t.Fatal(err)
}
proposal.Region = "EU"
proposal.Protocol = 1
for index := range proposal.Participants {
if proposal.Participants[index].PlayerID == "proposal-player-a" {
proposal.Participants[index].Team = 0
proposal.Participants[index].Slot = 0
} else {
proposal.Participants[index].Team = 1
proposal.Participants[index].Slot = 3
}
}
if err := CreateProposal(ctx, db, proposal, map[string]string{"proposal-player-a": "proposal-ticket-0", "proposal-player-b": "proposal-ticket-1"}, now); err != nil {
t.Fatalf("create proposal: %v", err)
}
var proposed int
if err := db.QueryRow(`SELECT count(*) FROM queue_tickets WHERE state = 'PROPOSED'`).Scan(&proposed); err != nil || proposed != 2 {
t.Fatalf("proposed queue tickets = %d, err = %v", proposed, err)
}
recoveredTicket, err := GetQueueTicket(ctx, db, "proposal-player-a", "proposal-ticket-0", now)
if err != nil || recoveredTicket.ProposalID != proposal.ProposalID {
t.Fatalf("recovered ticket proposal=%q err=%v", recoveredTicket.ProposalID, err)
}
recovered, err := GetProposal(ctx, db, "proposal-player-a", proposal.ProposalID, now)
if err != nil {
t.Fatalf("recover proposal: %v", err)
}
if len(recovered.Participants) != 2 || recovered.Revision != 0 {
t.Fatalf("unexpected recovered proposal: %+v", recovered)
}
accepted, err := RespondToProposal(ctx, db, "proposal-player-a", proposal.ProposalID, "proposal-response-a-0001", true, 0, now)
if err != nil {
t.Fatalf("first proposal acceptance: %v", err)
}
if accepted.Revision != 1 || accepted.State != domain.Open {
t.Fatalf("unexpected first acceptance: %+v", accepted)
}
accepted, err = RespondToProposal(ctx, db, "proposal-player-b", proposal.ProposalID, "proposal-response-b-0001", true, 1, now)
if err != nil {
t.Fatalf("second proposal acceptance: %v", err)
}
if accepted.State != domain.Accepted || accepted.Revision != 2 {
t.Fatalf("proposal did not close after unanimous acceptance: %+v", accepted)
}
// The API's post-commit recovery promoter must accept the nullable casual
// arena path left by the atomic response transaction and converge on the
// already-created match.
if err := PromoteStoredAcceptedProposal(ctx, db, proposal.ProposalID, now.Add(time.Second)); err != nil {
t.Fatalf("replay persisted casual promotion: %v", err)
}
var matchState string
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'match-proposal-integration'`).Scan(&matchState); err != nil {
t.Fatalf("atomic accepted match: %v", err)
}
var acceptedTickets, matchPlayers int
if err := db.QueryRow(`SELECT count(*) FROM queue_tickets WHERE ticket_id LIKE 'proposal-ticket-%' AND state = 'ACCEPTED'`).Scan(&acceptedTickets); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM match_participants WHERE match_id = 'match-proposal-integration'`).Scan(&matchPlayers); err != nil {
t.Fatal(err)
}
if matchState != "ALLOCATING" || acceptedTickets != 2 || matchPlayers != 2 {
t.Fatalf("acceptance did not atomically materialize match: state=%s tickets=%d players=%d", matchState, acceptedTickets, matchPlayers)
}
}
// TestPostgreSQLProposalDeclineCancelsOffenderAndRequeuesInnocent protects the
// durable decline boundary: the offender's ticket becomes terminal while
// every innocent ticket keeps its original queue precedence.
func TestPostgreSQLProposalDeclineCancelsOffenderAndRequeuesInnocent(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"decline-player-a", "decline-player-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
for i, player := range []string{"decline-player-a", "decline-player-b"} {
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', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("decline-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
proposal, err := domain.NewProposal("decline-proposal", domain.Casual, []string{"decline-player-a", "decline-player-b"}, now)
if err != nil {
t.Fatal(err)
}
if err := CreateProposal(ctx, db, proposal, map[string]string{"decline-player-a": "decline-ticket-0", "decline-player-b": "decline-ticket-1"}, now); err != nil {
t.Fatalf("create proposal: %v", err)
}
// player-a declines; player-b never responded at all -- the bug affects
// even a participant who was never asked to do anything wrong.
declined, err := RespondToProposal(ctx, db, "decline-player-a", proposal.ProposalID, "decline-response-a-0001", false, 0, now)
if err != nil {
t.Fatalf("decline: %v", err)
}
if declined.State != domain.Declined {
t.Fatalf("proposal did not close on decline: %+v", declined)
}
var stateA, stateB string
var expiresB time.Time
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'decline-ticket-0'`).Scan(&stateA); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'decline-ticket-1'`).Scan(&stateB, &expiresB); err != nil {
t.Fatal(err)
}
if stateA != "CANCELLED" {
t.Fatalf("decliner's own ticket state = %s, want CANCELLED", stateA)
}
if stateB != "QUEUED" {
t.Fatalf("uninvolved participant's ticket state = %s, want QUEUED -- they must not be stranded by someone else's decline", stateB)
}
if !expiresB.After(now) {
t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", expiresB, now)
}
// Only the innocent player can be selected again. A durable cooldown also
// rejects a new ticket from the decliner until the policy window ends.
candidates, err := ListQueuedCandidates(ctx, db, domain.Casual, now, 10)
if err != nil {
t.Fatalf("list queued candidates: %v", err)
}
found := map[string]bool{}
for _, candidate := range candidates {
found[candidate.PlayerID] = true
}
if found["decline-player-a"] || !found["decline-player-b"] {
t.Fatalf("matcher did not isolate offender from innocent: %+v", candidates)
}
var cooldownEnd time.Time
if err := db.QueryRow(`SELECT ends_at FROM penalties WHERE player_id = 'decline-player-a' AND kind = 'PROPOSAL_DECLINED'`).Scan(&cooldownEnd); err != nil {
t.Fatal(err)
}
if want := now.Add(30 * time.Second); !cooldownEnd.Equal(want) {
t.Fatalf("decline cooldown end = %v, want %v", cooldownEnd, want)
}
// Recovering the closed proposal after its old deadline must not convert
// the innocent participant's PENDING response into a timeout penalty.
if _, err := GetProposal(ctx, db, "decline-player-b", proposal.ProposalID, now.Add(domain.ProposalWindow+time.Second)); err != nil {
t.Fatalf("recover declined proposal: %v", err)
}
var innocentTimeouts int
if err := db.QueryRow(`SELECT count(*) FROM penalties WHERE player_id = 'decline-player-b' AND kind = 'PROPOSAL_TIMEOUT'`).Scan(&innocentTimeouts); err != nil {
t.Fatal(err)
}
if innocentTimeouts != 0 {
t.Fatalf("innocent participant received %d timeout penalties after decline", innocentTimeouts)
}
}
// TestPostgreSQLProposalTimeoutExpiresOffenderAndRequeuesAccepted protects the
// timeout sibling: accepted participants retain precedence, while no-shows
// receive a terminal ticket and cooldown.
func TestPostgreSQLProposalTimeoutExpiresOffenderAndRequeuesAccepted(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"timeout-player-a", "timeout-player-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
for i, player := range []string{"timeout-player-a", "timeout-player-b"} {
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', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("timeout-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
proposal, err := domain.NewProposal("timeout-proposal", domain.Casual, []string{"timeout-player-a", "timeout-player-b"}, now)
if err != nil {
t.Fatal(err)
}
if err := CreateProposal(ctx, db, proposal, map[string]string{"timeout-player-a": "timeout-ticket-0", "timeout-player-b": "timeout-ticket-1"}, now); err != nil {
t.Fatalf("create proposal: %v", err)
}
if _, err := RespondToProposal(ctx, db, "timeout-player-a", proposal.ProposalID, "timeout-accept-a-0001", true, 0, now.Add(time.Second)); err != nil {
t.Fatalf("accept proposal: %v", err)
}
// player-b never responds; recover well after the response window.
afterExpiry := now.Add(domain.ProposalWindow + time.Second)
recovered, err := GetProposal(ctx, db, "timeout-player-a", proposal.ProposalID, afterExpiry)
if err != nil {
t.Fatalf("recover expired proposal: %v", err)
}
if recovered.State != domain.Expired {
t.Fatalf("proposal did not expire: %+v", recovered)
}
var stateA, stateB string
var expiresA time.Time
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'timeout-ticket-0'`).Scan(&stateA, &expiresA); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'timeout-ticket-1'`).Scan(&stateB); err != nil {
t.Fatal(err)
}
if stateA != "QUEUED" || stateB != "EXPIRED" {
t.Fatalf("timeout did not split accepted and offender tickets: a=%s b=%s", stateA, stateB)
}
if !expiresA.After(afterExpiry) {
t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", expiresA, afterExpiry)
}
candidates, err := ListQueuedCandidates(ctx, db, domain.Casual, afterExpiry, 10)
if err != nil {
t.Fatalf("list queued candidates: %v", err)
}
found := map[string]bool{}
for _, candidate := range candidates {
found[candidate.PlayerID] = true
}
if !found["timeout-player-a"] || found["timeout-player-b"] {
t.Fatalf("matcher did not isolate timeout offender: %+v", candidates)
}
var cooldownEnd time.Time
if err := db.QueryRow(`SELECT ends_at FROM penalties WHERE player_id = 'timeout-player-b' AND kind = 'PROPOSAL_TIMEOUT'`).Scan(&cooldownEnd); err != nil {
t.Fatal(err)
}
if want := afterExpiry.Add(60 * time.Second); !cooldownEnd.Equal(want) {
t.Fatalf("timeout cooldown end = %v, want %v", cooldownEnd, want)
}
}
// A late response must report a closed proposal only after committing the
// expiry recovery. Returning that domain error from inside RunSerializable
// used to roll every recovery write back.
func TestPostgreSQLLateProposalResponseCommitsExpiryRecovery(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"late-player-a", "late-player-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
for i, player := range []string{"late-player-a", "late-player-b"} {
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', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("late-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
proposal, err := domain.NewProposal("late-proposal", domain.Casual, []string{"late-player-a", "late-player-b"}, now)
if err != nil {
t.Fatal(err)
}
if err := CreateProposal(ctx, db, proposal, map[string]string{"late-player-a": "late-ticket-0", "late-player-b": "late-ticket-1"}, now); err != nil {
t.Fatalf("create proposal: %v", err)
}
late := now.Add(domain.ProposalWindow + time.Second)
_, err = RespondToProposal(ctx, db, "late-player-a", proposal.ProposalID, "late-response-a-0001", true, 0, late)
if !errors.Is(err, domain.ErrProposalClosed) {
t.Fatalf("late response error = %v, want ErrProposalClosed", err)
}
var proposalState, ticketA, ticketB string
if err := db.QueryRow(`SELECT state FROM proposals WHERE proposal_id = 'late-proposal'`).Scan(&proposalState); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'late-ticket-0'`).Scan(&ticketA); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'late-ticket-1'`).Scan(&ticketB); err != nil {
t.Fatal(err)
}
if proposalState != "EXPIRED" || ticketA != "EXPIRED" || ticketB != "EXPIRED" {
t.Fatalf("late recovery was not committed: proposal=%s tickets=%s,%s", proposalState, ticketA, ticketB)
}
var penalties, idempotencyRows int
if err := db.QueryRow(`SELECT count(*) FROM penalties WHERE kind = 'PROPOSAL_TIMEOUT' AND player_id IN ('late-player-a', 'late-player-b')`).Scan(&penalties); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM idempotency_keys WHERE scope = $1 AND idempotency_key = 'late-response-a-0001'`, ProposalResponseIdempotencyScope).Scan(&idempotencyRows); err != nil {
t.Fatal(err)
}
if penalties != 2 || idempotencyRows != 0 {
t.Fatalf("late recovery side effects: penalties=%d idempotency_rows=%d", penalties, idempotencyRows)
}
}
// TestPostgreSQLConcurrentProposalExpiryRecoveryAppliesCooldownsExactlyOnce
// covers the race §8.46 flagged as still open: multiple concurrent recovery
// paths (a read-side GetProposal from each participant polling for an
// update, and a RespondToProposal arriving right at the same boundary) can
// all observe the same past-expiry proposal simultaneously. Every one of
// them runs the identical expiry-advance SQL in its own transaction, so this
// proves that racing recovery does not multiply the durable side effects: a
// PROPOSAL_TIMEOUT cooldown must land exactly once per offending player, not
// once per racing transaction that happened to perform the PENDING ->
// TIMED_OUT flip. The design's own defense is that ProposalParticipantExpireSQL
// only ever flips a still-PENDING row once, and recordProposalTimeoutCooldowns
// only cooldowns participants whose responded_at equals this transaction's
// own `now` -- so a loser transaction's `now` simply matches nothing.
func TestPostgreSQLConcurrentProposalExpiryRecoveryAppliesCooldownsExactlyOnce(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"race-expiry-a", "race-expiry-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
for i, player := range []string{"race-expiry-a", "race-expiry-b"} {
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', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("race-expiry-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
proposal, err := domain.NewProposal("race-expiry-proposal", domain.Casual, []string{"race-expiry-a", "race-expiry-b"}, now)
if err != nil {
t.Fatal(err)
}
if err := CreateProposal(ctx, db, proposal, map[string]string{"race-expiry-a": "race-expiry-ticket-0", "race-expiry-b": "race-expiry-ticket-1"}, now); err != nil {
t.Fatalf("create proposal: %v", err)
}
late := now.Add(domain.ProposalWindow + time.Second)
const racers = 8
var wg sync.WaitGroup
errs := make([]error, racers)
wg.Add(racers)
for i := 0; i < racers; i++ {
go func(i int) {
defer wg.Done()
// Each racer's `now` is distinct (and every one is past expiry), so a
// real implementation bug would show up as several of them believing
// they were the one that performed the PENDING -> TIMED_OUT flip.
racerNow := late.Add(time.Duration(i) * time.Millisecond)
switch i % 3 {
case 0:
_, errs[i] = GetProposal(ctx, db, "race-expiry-a", proposal.ProposalID, racerNow)
case 1:
_, errs[i] = GetProposal(ctx, db, "race-expiry-b", proposal.ProposalID, racerNow)
default:
_, errs[i] = RespondToProposal(ctx, db, "race-expiry-a", proposal.ProposalID, fmt.Sprintf("race-expiry-key-%04d", i), true, 0, racerNow)
}
}(i)
}
wg.Wait()
for i, err := range errs {
// GetProposal never errors on an already-expired proposal (it's a pure
// read-with-recovery); RespondToProposal on an already-closed proposal
// must report exactly ErrProposalClosed, nothing else.
if err != nil && !errors.Is(err, domain.ErrProposalClosed) {
t.Fatalf("racer %d: unexpected error %v", i, err)
}
}
var proposalState string
if err := db.QueryRow(`SELECT state FROM proposals WHERE proposal_id = 'race-expiry-proposal'`).Scan(&proposalState); err != nil {
t.Fatal(err)
}
if proposalState != "EXPIRED" {
t.Fatalf("proposal state = %s, want EXPIRED", proposalState)
}
var ticketA, ticketB string
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'race-expiry-ticket-0'`).Scan(&ticketA); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'race-expiry-ticket-1'`).Scan(&ticketB); err != nil {
t.Fatal(err)
}
if ticketA != "EXPIRED" || ticketB != "EXPIRED" {
t.Fatalf("tickets not expired exactly once: a=%s b=%s", ticketA, ticketB)
}
// The crux of the race: exactly one PROPOSAL_TIMEOUT penalty per player,
// however many transactions raced to observe the expiry.
var penaltiesA, penaltiesB int
if err := db.QueryRow(`SELECT count(*) FROM penalties WHERE kind = 'PROPOSAL_TIMEOUT' AND player_id = 'race-expiry-a'`).Scan(&penaltiesA); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM penalties WHERE kind = 'PROPOSAL_TIMEOUT' AND player_id = 'race-expiry-b'`).Scan(&penaltiesB); err != nil {
t.Fatal(err)
}
if penaltiesA != 1 || penaltiesB != 1 {
t.Fatalf("cooldown was not applied exactly once per player: a=%d b=%d", penaltiesA, penaltiesB)
}
var idempotencyRows int
if err := db.QueryRow(`SELECT count(*) FROM idempotency_keys WHERE scope = $1`, ProposalResponseIdempotencyScope).Scan(&idempotencyRows); err != nil {
t.Fatal(err)
}
if idempotencyRows != 0 {
t.Fatalf("closed-proposal responses left stray idempotency rows: %d", idempotencyRows)
}
}
// TestPostgreSQLCancellingAProposedTicketImmediatelyRequeuesTheOtherParticipant
// covers the responsiveness gap the decline/timeout fixes above left bounded
// but not closed: cancelling a ticket that's part of an OPEN proposal used
// to leave the OTHER participant waiting out the full 10s window for
// something the system already knew couldn't happen (their proposal partner
// just walked away). CascadeCancelToOpenProposal declines and requeues that
// proposal in the same transaction as the cancel itself.
func TestPostgreSQLCancellingAProposedTicketImmediatelyRequeuesTheOtherParticipant(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"cancel-cascade-a", "cancel-cascade-b"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
for i, player := range []string{"cancel-cascade-a", "cancel-cascade-b"} {
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', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("cancel-cascade-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
proposal, err := domain.NewProposal("cancel-cascade-proposal", domain.Casual, []string{"cancel-cascade-a", "cancel-cascade-b"}, now)
if err != nil {
t.Fatal(err)
}
if err := CreateProposal(ctx, db, proposal, map[string]string{"cancel-cascade-a": "cancel-cascade-ticket-0", "cancel-cascade-b": "cancel-cascade-ticket-1"}, now); err != nil {
t.Fatalf("create proposal: %v", err)
}
// player-a cancels their own ticket directly, well within the response
// window -- not a decline, not a timeout, just abandoning the queue.
// CreateProposal's own QueueTicketProposeSQL already bumped the ticket's
// revision from 0 to 1, so the cancel's expected revision is 1, not 0.
cancelled, err := CancelQueueTicket(ctx, db, "cancel-cascade-a", "cancel-cascade-ticket-0", "cancel-cascade-key-0001", 1, now.Add(time.Second))
if err != nil {
t.Fatalf("cancel: %v", err)
}
if cancelled.State != domain.Cancelled {
t.Fatalf("ticket did not cancel: %+v", cancelled)
}
var proposalState string
if err := db.QueryRow(`SELECT state FROM proposals WHERE proposal_id = 'cancel-cascade-proposal'`).Scan(&proposalState); err != nil {
t.Fatal(err)
}
if proposalState != "DECLINED" {
t.Fatalf("proposal state = %s, want DECLINED immediately, not left OPEN to time out", proposalState)
}
var stateB string
var expiresB time.Time
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'cancel-cascade-ticket-1'`).Scan(&stateB, &expiresB); err != nil {
t.Fatal(err)
}
if stateB != "QUEUED" {
t.Fatalf("other participant's ticket state = %s, want QUEUED immediately", stateB)
}
if !expiresB.After(now.Add(time.Second)) {
t.Fatalf("requeued ticket expiry %v was not refreshed forward", expiresB)
}
// The cancelling player's own ticket must stay CANCELLED, not get swept
// back up into the requeue meant for the other participant.
var stateA string
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'cancel-cascade-ticket-0'`).Scan(&stateA); err != nil {
t.Fatal(err)
}
if stateA != "CANCELLED" {
t.Fatalf("cancelling player's own ticket state = %s, want it to stay CANCELLED", stateA)
}
}
func TestPostgreSQLProposalCreationRollsBackPartialClaims(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('rollback-player', 'rollback-steam')`); 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 ('rollback-ticket', 'rollback-player', 'casual', 'QUEUED', 'integration-build', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
proposal, err := domain.NewProposal("rollback-proposal", domain.Casual, []string{"rollback-player", "missing-player"}, now)
if err != nil {
t.Fatal(err)
}
if err := CreateProposal(ctx, db, proposal, map[string]string{"rollback-player": "rollback-ticket"}, now); err == nil {
t.Fatal("proposal with missing ticket mapping was accepted")
}
var proposals, participants, proposed int
if err := db.QueryRow(`SELECT count(*) FROM proposals WHERE proposal_id = 'rollback-proposal'`).Scan(&proposals); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM proposal_participants WHERE proposal_id = 'rollback-proposal'`).Scan(&participants); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM queue_tickets WHERE ticket_id = 'rollback-ticket' AND state = 'PROPOSED'`).Scan(&proposed); err != nil {
t.Fatal(err)
}
if proposals != 0 || participants != 0 || proposed != 0 {
t.Fatalf("partial proposal claim was not rolled back: proposals=%d participants=%d proposed=%d", proposals, participants, proposed)
}
}
// TestPostgreSQLConcurrentProposalCreationClaimsContestedTicketOnce is the
// live counterpart to TestPostgreSQLProposalCreationRollsBackPartialClaims:
// every other proposal test in this file (and the whole matcher/allocator
// suite) runs its transactions strictly one at a time, so none of them can
// actually exercise the SERIALIZABLE retry-and-fence path CreateProposal
// relies on -- only two goroutines racing a real connection pool can. Two
// matchers independently form a proposal that both include the same waiting
// player's ticket (a real scenario: nothing stops two matcher replicas from
// reading the same QUEUED ticket in the same poll window); exactly one
// CreateProposal must win, the other must fail with its whole transaction
// rolled back, not a database/sql panic, deadlock, or a half-inserted row.
func TestPostgreSQLConcurrentProposalCreationClaimsContestedTicketOnce(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"race-player-a", "race-player-b", "race-player-c"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
tickets := map[string]string{"race-player-a": "race-ticket-a", "race-player-b": "race-ticket-b", "race-player-c": "race-ticket-c"}
for player, ticket := range tickets {
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', 'QUEUED', 'integration-build', 1, $3, $4)`, ticket, player, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
}
proposalA, err := domain.NewProposal("race-proposal-a", domain.Casual, []string{"race-player-a", "race-player-b"}, now)
if err != nil {
t.Fatal(err)
}
proposalB, err := domain.NewProposal("race-proposal-b", domain.Casual, []string{"race-player-b", "race-player-c"}, now)
if err != nil {
t.Fatal(err)
}
var wg sync.WaitGroup
errs := make([]error, 2)
wg.Add(2)
go func() {
defer wg.Done()
errs[0] = CreateProposal(ctx, db, proposalA, map[string]string{"race-player-a": tickets["race-player-a"], "race-player-b": tickets["race-player-b"]}, now)
}()
go func() {
defer wg.Done()
errs[1] = CreateProposal(ctx, db, proposalB, map[string]string{"race-player-b": tickets["race-player-b"], "race-player-c": tickets["race-player-c"]}, now)
}()
wg.Wait()
succeeded := errs[0] == nil
if succeeded == (errs[1] == nil) {
t.Fatalf("exactly one contested proposal must win, got errA=%v errB=%v", errs[0], errs[1])
}
winner, loser := "race-proposal-a", "race-proposal-b"
if !succeeded {
winner, loser = "race-proposal-b", "race-proposal-a"
}
var winnerRows, loserRows, loserParticipants int
if err := db.QueryRow(`SELECT count(*) FROM proposals WHERE proposal_id = $1`, winner).Scan(&winnerRows); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM proposals WHERE proposal_id = $1`, loser).Scan(&loserRows); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM proposal_participants WHERE proposal_id = $1`, loser).Scan(&loserParticipants); err != nil {
t.Fatal(err)
}
if winnerRows != 1 {
t.Fatalf("winning proposal %s was not persisted", winner)
}
if loserRows != 0 || loserParticipants != 0 {
t.Fatalf("losing proposal %s was not fully rolled back: proposals=%d participants=%d", loser, loserRows, loserParticipants)
}
var contestedState string
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = $1`, tickets["race-player-b"]).Scan(&contestedState); err != nil {
t.Fatal(err)
}
if contestedState != "PROPOSED" {
t.Fatalf("contested ticket should be claimed by the winner, got state=%s", contestedState)
}
// The loser's OWN uncontested ticket (a or c) must have rolled back to
// QUEUED too -- CreateProposal is one transaction per proposal, so a
// contested loss on one participant must not leave another participant's
// ticket stranded as PROPOSED with no surviving proposal to reference it.
// A (player-a + contested player-b) won iff succeeded, in which case B's
// own uncontested ticket (player-c) is the one that must have rolled
// back; if A lost, it's A's own uncontested ticket (player-a) instead.
loserOnlyTicket := tickets["race-player-c"]
if !succeeded {
loserOnlyTicket = tickets["race-player-a"]
}
var loserOnlyState string
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = $1`, loserOnlyTicket).Scan(&loserOnlyState); err != nil {
t.Fatal(err)
}
if loserOnlyState != "QUEUED" {
t.Fatalf("loser's uncontested ticket %s should have rolled back to QUEUED, got %s", loserOnlyTicket, loserOnlyState)
}
}
func TestPostgreSQLResultCompletionAndOutboxAreAtomicAndReplayable(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id) VALUES ('result-match', 'casual', 'RESULT_PENDING', 'NA', 1, 'result-server')`); err != nil {
t.Fatal(err)
}
payload := []byte(`{"match_id":"result-match","team0_score":2,"team1_score":1}`)
digest := sha256.Sum256(payload)
receipt := domain.ResultReceipt{ResultID: "result-receipt", MatchID: "result-match", ResultNonce: "result-nonce-123456", PayloadDigest: digest, IntegrityState: domain.IntegrityCertified, ReceivedAt: now}
if err := CompleteResult(ctx, db, receipt, "result-server", "result-event", payload, now); err != nil {
t.Fatalf("complete result: %v", err)
}
var state string
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'result-match'`).Scan(&state); err != nil {
t.Fatal(err)
}
if state != "COMPLETED" {
t.Fatalf("result match state = %s", state)
}
events, err := ReadUnpublishedOutbox(ctx, db, 10)
if err != nil || len(events) != 1 || events[0].EventID != "result-event" {
t.Fatalf("unpublished result events = %+v, err = %v", events, err)
}
if err := MarkOutboxPublished(ctx, db, events[0].EventID, now.Add(time.Second)); err != nil {
t.Fatalf("ack result event: %v", err)
}
if remaining, err := ReadUnpublishedOutbox(ctx, db, 10); err != nil || len(remaining) != 0 {
t.Fatalf("outbox after ack = %+v, err = %v", remaining, err)
}
if err := CompleteResult(ctx, db, receipt, "result-server", "result-event-retry", payload, now.Add(time.Second)); err != nil {
t.Fatalf("identical completed result replay: %v", err)
}
conflict := receipt
conflict.ResultID = "different-result"
if err := CompleteResult(ctx, db, conflict, "result-server", "different-event", []byte(`{"conflict":true}`), now.Add(2*time.Second)); err == nil {
t.Fatal("conflicting completed result was accepted")
}
}
// TestPostgreSQLConcurrentConflictingResultSubmissionsKeepOneReceipt
// exercises the other side of the result race: retries with different payloads
// must not let the winner's durable receipt be overwritten or create a second
// completion event.
func TestPostgreSQLConcurrentConflictingResultSubmissionsKeepOneReceipt(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"result-conflict-a", "result-conflict-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 matches (match_id, playlist, state, region, protocol_version, server_id) VALUES ('result-conflict-match', 'casual', 'RESULT_PENDING', 'NA', 1, 'result-conflict-server')`); 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 ('result-conflict-ticket-a', 'result-conflict-a', 'casual', 'LIVE', 'build-1', 1, $1, $2), ('result-conflict-ticket-b', 'result-conflict-b', 'casual', 'LIVE', 'build-1', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('result-conflict-match', 'result-conflict-a', 'result-conflict-ticket-a', 0, 0), ('result-conflict-match', 'result-conflict-b', 'result-conflict-ticket-b', 1, 1)`); err != nil {
t.Fatal(err)
}
base := domain.MatchResult{MatchID: "result-conflict-match", ServerID: "result-conflict-server", IntegrityState: domain.IntegritySuppressed}
results := []domain.MatchResult{
{MatchID: base.MatchID, ServerID: base.ServerID, ResultNonce: "result-conflict-nonce-a-123456", Team0Score: 2, Team1Score: 1, IntegrityState: base.IntegrityState},
{MatchID: base.MatchID, ServerID: base.ServerID, ResultNonce: "result-conflict-nonce-b-123456", Team0Score: 1, Team1Score: 2, IntegrityState: base.IntegrityState},
}
errs := make([]error, 2)
var wg sync.WaitGroup
wg.Add(2)
for i := range results {
go func(i int) {
defer wg.Done()
digest := domain.ResultDigest(results[i])
receipt := domain.ResultReceipt{ResultID: fmt.Sprintf("result-conflict-receipt-%d", i), MatchID: results[i].MatchID, ResultNonce: results[i].ResultNonce, PayloadDigest: digest, IntegrityState: results[i].IntegrityState, ReceivedAt: now}
errs[i] = CompleteResult(ctx, db, receipt, results[i].ServerID, fmt.Sprintf("result-conflict-event-%d", i), []byte(fmt.Sprintf(`{"nonce":%q}`, results[i].ResultNonce)), now)
}(i)
}
wg.Wait()
wins := 0
for _, err := range errs {
if err == nil {
wins++
} else if !strings.Contains(err.Error(), "conflict") {
t.Fatalf("non-conflict error in conflicting race: %v", err)
}
}
if wins != 1 {
t.Fatalf("successful conflicting submissions = %d, want exactly one; errors=%v", wins, errs)
}
var receipts, events int
if err := db.QueryRow(`SELECT count(*) FROM result_receipts WHERE match_id = 'result-conflict-match'`).Scan(&receipts); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM outbox WHERE aggregate_id = 'result-conflict-match' AND event_type = 'match_completed'`).Scan(&events); err != nil {
t.Fatal(err)
}
if receipts != 1 || events != 1 {
t.Fatalf("durable conflict race left receipts=%d events=%d, want one of each", receipts, events)
}
}
// TestPostgreSQLConcurrentIdenticalResultSubmissionAppliesRatingsExactlyOnce
// races real concurrent duplicate result submissions -- the scenario behind
// task 8.25's "identical duplicates idempotent" claim, which every other
// result test in this file (and the mocked-driver unit tests) only exercises
// sequentially. A game server can legitimately retry an unacknowledged
// result POST, and two such retries can land at PostgreSQL genuinely
// concurrently; every one of them must succeed (this is the identical-replay
// path, not a conflict), the match must complete exactly once, and -- the
// part that matters -- the rating update inside applyResultRatings must not
// run twice just because it raced.
func TestPostgreSQLConcurrentIdenticalResultSubmissionAppliesRatingsExactlyOnce(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
for _, player := range []string{"result-race-winner", "result-race-loser"} {
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 ratings (player_id, rating, deviation, volatility, ranked_games) VALUES ($1, 1500, 350, 0.06, 0)`, player); err != nil {
t.Fatal(err)
}
}
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id) VALUES ('result-race-match', 'casual', 'LIVE', 'NA', 1, 'result-race-server')`); 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 ('result-race-ticket-w', 'result-race-winner', 'casual', 'LIVE', 'build-1', 1, $1, $2), ('result-race-ticket-l', 'result-race-loser', 'casual', 'LIVE', 'build-1', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('result-race-match', 'result-race-winner', 'result-race-ticket-w', 0, 0), ('result-race-match', 'result-race-loser', 'result-race-ticket-l', 1, 1)`); err != nil {
t.Fatal(err)
}
result := domain.MatchResult{MatchID: "result-race-match", ServerID: "result-race-server", ResultNonce: "result-race-nonce-123456", Team0Score: 3, Team1Score: 1, IntegrityState: domain.IntegrityCertified}
digest := domain.ResultDigest(result)
receipt := domain.ResultReceipt{ResultID: "result-race-receipt", MatchID: result.MatchID, ResultNonce: result.ResultNonce, PayloadDigest: digest, IntegrityState: result.IntegrityState, ReceivedAt: now}
payload := []byte(`{"match_id":"result-race-match"}`)
const attempts = 5
var wg sync.WaitGroup
errs := make([]error, attempts)
wg.Add(attempts)
for i := 0; i < attempts; i++ {
go func(i int) {
defer wg.Done()
errs[i] = CompleteResultWithResult(ctx, db, receipt, result.ServerID, fmt.Sprintf("result-race-event-%d", i), payload, result, now)
}(i)
}
wg.Wait()
for i, err := range errs {
if err != nil {
t.Fatalf("identical concurrent submission %d failed: %v", i, err)
}
}
var state string
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'result-race-match'`).Scan(&state); err != nil {
t.Fatal(err)
}
if state != "COMPLETED" {
t.Fatalf("match state = %s, want COMPLETED", state)
}
var completedTickets int
if err := db.QueryRow(`SELECT count(*) FROM queue_tickets WHERE ticket_id IN ('result-race-ticket-w', 'result-race-ticket-l') AND state = 'COMPLETED'`).Scan(&completedTickets); err != nil {
t.Fatal(err)
}
if completedTickets != 2 {
t.Fatalf("completed tickets = %d, want 2", completedTickets)
}
var winnerGames, loserGames int
var winnerRating, loserRating float64
if err := db.QueryRow(`SELECT ranked_games, rating FROM ratings WHERE player_id = 'result-race-winner'`).Scan(&winnerGames, &winnerRating); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT ranked_games, rating FROM ratings WHERE player_id = 'result-race-loser'`).Scan(&loserGames, &loserRating); err != nil {
t.Fatal(err)
}
// Casual results never increment ranked_games by design (rankedIncrement
// is unconditionally 0 for domain.Casual in applyResultRatings) -- that's
// not what this test is verifying. What proves "applied exactly once, not
// N times under the race" is the rating VALUE: a second application would
// recompute from the already-updated current rating and compound further
// away from 1500, so an exact match against a single, independently
// computed application is the assertion that actually falsifies a double
// application (unlike an inequality check, which a doubled update would
// still satisfy).
if winnerGames != 0 || loserGames != 0 {
t.Fatalf("casual result should never touch ranked_games: winner=%d loser=%d", winnerGames, loserGames)
}
baseline := domain.Rating{Value: 1500, RD: 350, Volatility: 0.06}
winnerOpponents, err := domain.CasualOpponents([]domain.Opponent{{PlayerID: "result-race-loser", Rating: baseline, Score: 1}})
if err != nil {
t.Fatal(err)
}
wantWinner, err := domain.UpdateRating(baseline, winnerOpponents, now)
if err != nil {
t.Fatal(err)
}
loserOpponents, err := domain.CasualOpponents([]domain.Opponent{{PlayerID: "result-race-winner", Rating: baseline, Score: 0}})
if err != nil {
t.Fatal(err)
}
wantLoser, err := domain.UpdateRating(baseline, loserOpponents, now)
if err != nil {
t.Fatal(err)
}
if winnerRating != wantWinner.Value {
t.Fatalf("winner rating = %v, want exactly %v (a value between these would indicate a partial/compounded update)", winnerRating, wantWinner.Value)
}
if loserRating != wantLoser.Value {
t.Fatalf("loser rating = %v, want exactly %v", loserRating, wantLoser.Value)
}
if winnerRating <= loserRating {
t.Fatalf("winner rating %v should exceed loser rating %v after a certified result", winnerRating, loserRating)
}
}
// TestPostgreSQLStalledAllocationsAreReclaimedWithoutPenalisingPlayers is the
// live counterpart to the SQL fragment test: it proves the actual data
// movement against a real database, not just that the right substrings are
// present. Two matches: one genuinely stalled (old enough to reclaim), one
// recent (must survive untouched) -- the deadline boundary and the
// no-penalty requeue are both meaningless without a real row to check.
func TestPostgreSQLStalledAllocationsAreReclaimedWithoutPenalisingPlayers(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Microsecond)
stalledCreatedAt := now.Add(-10 * time.Minute)
recentCreatedAt := now.Add(-5 * time.Second)
for _, player := range []string{"stall-player-a", "stall-player-b", "recent-player"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
t.Fatal(err)
}
}
insertTicket := func(ticketID, playerID, state string, expiresAt time.Time) {
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', $3, 'integration-build', 1, $4, $5)`, ticketID, playerID, state, now, expiresAt); err != nil {
t.Fatal(err)
}
}
insertTicket("stall-ticket-a", "stall-player-a", "PROCESS_READY", now.Add(time.Hour))
insertTicket("stall-ticket-b", "stall-player-b", "PROCESS_READY", now.Add(time.Hour))
insertTicket("recent-ticket", "recent-player", "ALLOCATING", now.Add(time.Hour))
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, created_at) VALUES ('stalled-match', 'casual', 'PROCESS_READY', 'NA', 1, 'stalled-server', $1)`, stalledCreatedAt); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, created_at) VALUES ('recent-match', 'casual', 'ALLOCATING', 'NA', 1, $1)`, recentCreatedAt); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('stalled-match', 'stall-player-a', 'stall-ticket-a', 0, 0), ('stalled-match', 'stall-player-b', 'stall-ticket-b', 1, 1)`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('recent-match', 'recent-player', 'recent-ticket', 0, 0)`); err != nil {
t.Fatal(err)
}
reclaimed, err := ExpireStalledAllocations(ctx, db, now, 2*time.Minute, 10)
if err != nil {
t.Fatalf("expire stalled allocations: %v", err)
}
if reclaimed != 1 {
t.Fatalf("reclaimed = %d, want exactly 1 (the recent match must survive)", reclaimed)
}
var stalledState, recentState string
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'stalled-match'`).Scan(&stalledState); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'recent-match'`).Scan(&recentState); err != nil {
t.Fatal(err)
}
if stalledState != "FAILED" {
t.Fatalf("stalled match state = %s, want FAILED", stalledState)
}
if recentState != "ALLOCATING" {
t.Fatalf("recent match state = %s, want untouched ALLOCATING", recentState)
}
var ticketAState, ticketBState, recentTicketState string
var ticketAExpiry time.Time
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'stall-ticket-a'`).Scan(&ticketAState, &ticketAExpiry); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'stall-ticket-b'`).Scan(&ticketBState); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'recent-ticket'`).Scan(&recentTicketState); err != nil {
t.Fatal(err)
}
if ticketAState != "QUEUED" || ticketBState != "QUEUED" {
t.Fatalf("stalled participants' tickets = %s, %s -- want both requeued to QUEUED, not failed/left behind", ticketAState, ticketBState)
}
if !ticketAExpiry.After(now) {
t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", ticketAExpiry, now)
}
if recentTicketState != "ALLOCATING" {
t.Fatalf("recent match's ticket state = %s, want untouched ALLOCATING", recentTicketState)
}
var activeParticipants int
if err := db.QueryRow(`SELECT count(*) FROM match_participants WHERE match_id = 'stalled-match' AND participation_active`).Scan(&activeParticipants); err != nil {
t.Fatal(err)
}
if activeParticipants != 0 {
t.Fatalf("stalled match still has %d active participants, want 0 (so the player can be matched again)", activeParticipants)
}
var eventType string
var eventPayload []byte
if err := db.QueryRow(`SELECT event_type, payload FROM outbox WHERE event_id = 'stalled-allocation:stalled-match:1'`).Scan(&eventType, &eventPayload); err != nil {
t.Fatalf("stalled allocation state event missing: %v", err)
}
var event struct {
State string `json:"state"`
PlayerIDs []string `json:"player_ids"`
}
if err := json.Unmarshal(eventPayload, &event); err != nil {
t.Fatalf("decode stalled allocation outbox event: %v", err)
}
players := make(map[string]bool, len(event.PlayerIDs))
for _, playerID := range event.PlayerIDs {
players[playerID] = true
}
if eventType != "state_changed" || event.State != "FAILED" || !players["stall-player-a"] {
t.Fatalf("stalled allocation event = %s %s, want FAILED state and affected player IDs", eventType, eventPayload)
}
// Idempotent: the match is now FAILED, not one of the three reclaimable
// states, so a second pass must not touch it again.
reclaimedAgain, err := ExpireStalledAllocations(ctx, db, now.Add(time.Minute), 2*time.Minute, 10)
if err != nil {
t.Fatalf("second expire pass: %v", err)
}
if reclaimedAgain != 0 {
t.Fatalf("second pass reclaimed %d matches, want 0 (already-FAILED match must not be reprocessed)", reclaimedAgain)
}
}
func TestPostgreSQLRankedSeasonRolloverIsExactlyOnce(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('season-player', 'season-steam')`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO ratings (player_id, rating, deviation, volatility, ranked_games) VALUES ('season-player', 2000, 100, 0.12, 25)`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO seasons (season_id, playlist, starts_at, ends_at) VALUES ('season-1', 'ranked', $1, $2)`, now.Add(-12*7*24*time.Hour), now); err != nil {
t.Fatal(err)
}
profile := domain.RankedProfile{Rating: domain.Rating{Value: 1900, RD: 100, Volatility: 0.12}, RankedGames: 25}
updated, applied, err := ApplyRankedSeasonRollover(ctx, db, "season-player", "season-1", profile, now)
if err != nil || !applied {
t.Fatalf("first season rollover = %+v applied=%v err=%v", updated, applied, err)
}
if updated.Value != 1875 || updated.RD != 200 {
t.Fatalf("unexpected rolled rating: %+v", updated)
}
var rating float64
var markers int
if err := db.QueryRow(`SELECT rating FROM ratings WHERE player_id = 'season-player'`).Scan(&rating); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT count(*) FROM ranked_season_rollovers WHERE player_id = 'season-player' AND season_id = 'season-1'`).Scan(&markers); err != nil {
t.Fatal(err)
}
if rating != 1875 || markers != 1 {
t.Fatalf("durable rollover state rating=%v markers=%d", rating, markers)
}
duplicate, applied, err := ApplyRankedSeasonRollover(ctx, db, "season-player", "season-1", profile, now.Add(time.Second))
if err != nil || applied || duplicate.Value != 1875 {
t.Fatalf("duplicate rollover = %+v applied=%v err=%v", duplicate, applied, err)
}
if err := db.QueryRow(`SELECT rating FROM ratings WHERE player_id = 'season-player'`).Scan(&rating); err != nil {
t.Fatal(err)
}
if rating != 1875 {
t.Fatalf("duplicate rollover changed rating to %v", rating)
}
}
func TestPostgreSQLEmptyRankedSeasonIsMarkedRolledOver(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO seasons (season_id, playlist, starts_at, ends_at) VALUES ('empty-season', 'ranked', $1, $2)`, now.Add(-12*7*24*time.Hour), now); err != nil {
t.Fatal(err)
}
if count, err := RolloverDueSeasons(ctx, db, now, 100); err != nil || count != 0 {
t.Fatalf("empty-season maintenance count=%d err=%v", count, err)
}
var rolledAt sql.NullTime
if err := db.QueryRowContext(ctx, `SELECT rolled_over_at FROM seasons WHERE season_id = 'empty-season'`).Scan(&rolledAt); err != nil {
t.Fatal(err)
}
if !rolledAt.Valid {
t.Fatal("empty ranked season was not marked rolled over")
}
}
func TestPostgreSQLRankedProfileProjectsActiveSeason(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('profile-season-player', 'profile-season-steam')`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO ratings (player_id, rating, deviation, volatility, ranked_games) VALUES ('profile-season-player', 1600, 200, 0.06, 10)`); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO seasons (season_id, playlist, starts_at, ends_at) VALUES ('profile-season-current', 'ranked', $1, $2)`, now.Add(-time.Hour), now.Add(time.Hour)); err != nil {
t.Fatal(err)
}
profile, found, err := (PostgresRankedProfiles{DB: db}).Get(ctx, "profile-season-player")
if err != nil || !found || profile.CurrentSeasonID != "profile-season-current" || !profile.CurrentSeasonEndsAt.Equal(now.Add(time.Hour)) {
t.Fatalf("profile=%+v found=%t err=%v", profile, found, err)
}
}
func TestPostgreSQLMigrationsAreForwardExecutable(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
var tableCount int
if err := db.QueryRow(`SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'assignments'`).Scan(&tableCount); err != nil {
t.Fatal(err)
}
if tableCount != 1 {
t.Fatal("assignments migration did not create its table")
}
}
func TestPostgreSQLMigrationsRollBackAndReapplyCleanly(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
dir := filepath.Join("..", "migrations")
tableExists := func(table string) bool {
var count int
if err := db.QueryRow(`SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1`, table).Scan(&count); err != nil {
t.Fatal(err)
}
return count == 1
}
if !tableExists("assignments") || !tableExists("allocations") {
t.Fatal("expected forward-applied schema before rollback")
}
// Roll back every migration one at a time, in reverse, checking each
// down file actually undoes what its forward file created — not just
// that Rollback returns nil.
// This count is the number of migrations above 0006, so it must grow with
// every new migration; otherwise the later fixed-count rollbacks below
// silently target the wrong files.
if err := migrations.Rollback(context.Background(), db, dir, 8); err != nil {
t.Fatalf("rollback 0014 through 0007: %v", err)
}
var hasInitialConnectReadyColumn bool
if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'initial_connect_ready_at'`).Scan(&hasInitialConnectReadyColumn); err != nil {
t.Fatal(err)
}
if hasInitialConnectReadyColumn {
t.Fatal("0010 rollback did not drop matches.initial_connect_ready_at")
}
if err := migrations.Rollback(context.Background(), db, dir, 1); err != nil {
t.Fatalf("rollback 0006: %v", err)
}
var hasAllocationClaimColumn bool
if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'allocation_id'`).Scan(&hasAllocationClaimColumn); err != nil {
t.Fatal(err)
}
if hasAllocationClaimColumn {
t.Fatal("0006 rollback did not drop matches.allocation_id")
}
if err := migrations.Rollback(context.Background(), db, dir, 4); err != nil {
t.Fatalf("rollback 0005 through 0002: %v", err)
}
if tableExists("assignments") || tableExists("allocations") || tableExists("game_servers") {
t.Fatal("rollback left later-migration tables behind")
}
if err := migrations.Rollback(context.Background(), db, dir, 1); err != nil {
t.Fatalf("rollback 0001: %v", err)
}
if tableExists("identities") || tableExists("matches") {
t.Fatal("0001 rollback did not drop its own tables")
}
var remaining int
if err := db.QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&remaining); err != nil {
t.Fatal(err)
}
if remaining != 0 {
t.Fatalf("expected schema_migrations empty after full rollback, got %d rows", remaining)
}
// Reapplying from a fully rolled-back state must reach the same schema,
// proving down files don't leave orphaned state that trips a forward
// re-run (e.g. a constraint or index Apply then tries to recreate).
if err := migrations.Apply(context.Background(), db, dir); err != nil {
t.Fatalf("reapply after full rollback: %v", err)
}
if !tableExists("assignments") || !tableExists("allocations") {
t.Fatal("reapply after rollback did not recreate the schema")
}
}
// Ranked matchmaking reads Candidate.Rating for tolerance, selection scoring
// and team partitioning. The candidate projection did not join the ratings
// table and its scan never set the field, so every PostgreSQL-sourced ranked
// candidate arrived with Go's zero value and the matcher treated a 900-rated
// player as identical to a 2100-rated one. Unit tests missed this because they
// construct candidates with ratings already populated.
func TestPostgreSQLQueuedCandidatesCarryAuthoritativeRatings(t *testing.T) {
db := openIntegrationPostgres(t)
applyIntegrationMigrations(t, db)
now := time.Now().UTC().Truncate(time.Microsecond)
ctx := context.Background()
// "rated-low" and "rated-high" are deliberately far apart; "unrated" has no
// ratings row at all and must fall back to the new-profile default.
seeded := map[string]float64{"rated-low": 900, "rated-high": 2100}
for _, playerID := range []string{"rated-low", "rated-high", "unrated"} {
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $2)`, playerID, "steam-"+playerID); err != nil {
t.Fatal(err)
}
if rating, ok := seeded[playerID]; ok {
if _, err := db.ExecContext(ctx, `INSERT INTO ratings (player_id, rating) VALUES ($1, $2)`, playerID, rating); err != nil {
t.Fatal(err)
}
}
spec := domain.QueueSpec{Playlist: domain.Ranked, ClientBuild: "rating-build", ProtocolVersion: 1}
ticket, err := CreateQueueTicket(ctx, db, "ticket-"+playerID, playerID, "rating-create-"+playerID+"-01", spec, now)
if err != nil {
t.Fatalf("create queue ticket for %s: %v", playerID, err)
}
// The Redis projection is seeded from this candidate rather than from
// the query below, so it must carry the same rating or the two
// projections disagree about who is comparable to whom.
want := domain.GlickoInitialRating
if rating, ok := seeded[playerID]; ok {
want = rating
}
if ticket.Candidate.Rating != want {
t.Fatalf("%s enqueue candidate rating = %v, want %v", playerID, ticket.Candidate.Rating, want)
}
}
candidates, err := ListQueuedCandidates(ctx, db, domain.Ranked, now, 100)
if err != nil {
t.Fatalf("list queued candidates: %v", err)
}
if len(candidates) != 3 {
t.Fatalf("expected 3 candidates, got %d", len(candidates))
}
got := make(map[string]float64, len(candidates))
for _, candidate := range candidates {
got[candidate.PlayerID] = candidate.Rating
}
for playerID, want := range map[string]float64{
"rated-low": 900, "rated-high": 2100, "unrated": domain.GlickoInitialRating,
} {
if got[playerID] != want {
t.Fatalf("%s durable candidate rating = %v, want %v", playerID, got[playerID], want)
}
}
// The whole point of loading the rating is that the matcher can tell these
// players apart. Assert the spread survives into team partitioning rather
// than only that the field is non-zero.
if got["rated-high"]-got["rated-low"] != 1200 {
t.Fatalf("rating spread collapsed: %v", got)
}
}
Reference in New Issue View Git Blame Copy Permalink