From 7807b9706b56b1853756dffa6662ac2e8e8ec00e Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:21:41 +0100 Subject: [PATCH] feat: promote accepted proposals into matches --- multiplayer-next.md | 4 +- multiplayer-todo.md | 2 +- server/store/match_sql.go | 227 ++++++++++++++++++++++ server/store/match_sql_test.go | 74 +++++++ server/store/postgres_integration_test.go | 49 +++++ 5 files changed, 354 insertions(+), 2 deletions(-) create mode 100644 server/store/match_sql.go create mode 100644 server/store/match_sql_test.go diff --git a/multiplayer-next.md b/multiplayer-next.md index 8214fdfd..05675f65 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -138,7 +138,9 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md). - [ ] **IN PROGRESS:** Fence proposals/participants in a PostgreSQL serializable transaction; prove loss of an acknowledged Redis write cannot split players. The Go store adapter now performs proposal insertion, participant insertion, and every - queue-ticket promotion in one rollback-safe SERIALIZABLE callback; the runnable + queue-ticket promotion in one rollback-safe SERIALIZABLE callback; accepted + proposals now atomically promote their exact team/slot map and tickets into an + `ALLOCATING` match; the runnable casual matcher can optionally use a Redis candidate projection and repairs an empty/lost index from PostgreSQL before claiming durably; live DB/Redis failover testing remains. diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 999d0562..506896c7 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1196,7 +1196,7 @@ the local/CI/community transport, not a silent production fallback. | 8.15 `[D:7.8,8.3]` | **IN PROGRESS.** Pure Go probe validation treats Steam location as opaque, requires nonce/freshness/region and server-computed RTT, and implements discrepancy quarantine/release; authenticated HTTP now accepts only opaque location/nonce input through a server-owned probe provider and records validated RTT into the active player's durable queue ticket; durable queue projections have a server-derived RTT JSON field for matcher reads | `server/domain/probes.go`, `server/migrations/0003_queue_probe_metadata.sql`, adversarial fixtures and `server/api/service.go`/`store/queue_sql.go` cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine, five-clean release, authenticated provider arguments, rejection of client RTT fields, player/ticket/expiry binding, persistence failure, playlist-scoped candidate reads and bounded metadata decoding; Steam coordinator, regional probe adapters and multi-region probe population remain | | 8.16 `[D:8.14,8.15]` | **IN PROGRESS.** Pure Go candidate/team selection implements the <=100 ms ceiling, pairwise widening tolerance, anchor inclusion, deterministic set/region scoring and balanced team partitioning; queue-backed formation now consumes the server-owned projection, fences duplicate player identities and rejects playlist/build/protocol mixing; the matcher worker performs bounded formation and delegates the final claim to the durable proposal transaction; the runnable matcher now supports explicitly enabled ranked six-player polling with durable verified-Steam identity lookup | `server/domain/matcher.go`, `teams.go`, `server/matcher/worker.go`, `server/store/queue_sql.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance, malformed candidates, duplicate identities, compatibility mismatches, incomplete batches, source failure, durable claim failure, queue-backed oldest-anchor formation and incomplete ranked identity metadata; arena selection and long-running worker integration remain | | 8.17 `[D:8.14,8.16]` | **IN PROGRESS.** Pure Go proposal policy sends a 10-second response window to every selected human, requires unanimous acceptance, applies exact decline/timeout cooldowns and ranked escalation; authenticated API exposes revisioned accept/decline mutations; formed matches now pass through a playlist-aware proposal boundary | `server/domain/proposal.go`, `formation.go` and `server/api/service.go` plus adversarial fixtures cover partial/unanimous response, expiry, replay/conflict, stale API revision, casual lineup preparation and ranked metadata validation; queue precedence and allocation integration remain | -| 8.18 `[D:8.5,8.14,8.17]` | **IN PROGRESS.** Go store layer defines PostgreSQL SERIALIZABLE whole-transaction retries and queue candidate/proposal claim SQL using `FOR UPDATE SKIP LOCKED` plus durable uniqueness/revision fences; proposal creation now inserts proposal/participants and promotes every ticket in one rollback-safe transaction with player- and playlist-bound claim predicates, queue creation has a durable idempotency/owner-read adapter, participant-scoped proposal recovery now expires OPEN proposals and pending participants transactionally at read time, and proposal accept/decline now uses participant/proposal locks, revision fencing and durable idempotency; response attempts also advance expired proposals and pending participants before returning closed; runnable casual matcher polling now supports an optional Redis candidate projection that repairs empty/lost cache state from authoritative PostgreSQL before the durable final claim | `server/store/serializable.go`, `queue_sql.go`, `proposal_sql.go`, `proposal_recovery_sql.go`, `redis_candidates.go`, `server/matcher/worker.go` and tests cover retry classification, claim-boundary invariants, player/ticket/playlist mapping, durable queue replay/conflict, owner-scoped queue/proposal recovery, expiry at read and mutation boundaries, response replay/conflict, stale revisions, zero-row claim aborts, atomic statement ordering, incomplete matcher batches, source failures and empty-index Redis repair; opt-in PostgreSQL execution now covers queue create/replay/fencing, assignment persistence, proposal claim/promotion, participant recovery, unanimous response and rollback of partial claims; ranked provider, worker-failure and concurrent two-matcher integration tests remain | +| 8.18 `[D:8.5,8.14,8.17]` | **IN PROGRESS.** Go store layer defines PostgreSQL SERIALIZABLE whole-transaction retries and queue candidate/proposal claim SQL using `FOR UPDATE SKIP LOCKED` plus durable uniqueness/revision fences; proposal creation now inserts proposal/participants and promotes every ticket in one rollback-safe transaction with player- and playlist-bound claim predicates, queue creation has a durable idempotency/owner-read adapter, participant-scoped proposal recovery now expires OPEN proposals and pending participants transactionally at read time, and proposal accept/decline now uses participant/proposal locks, revision fencing and durable idempotency; an accepted proposal can now atomically create the exact `ALLOCATING` match/team/slot topology and promote all claimed tickets to `ACCEPTED`; runnable matcher polling supports an optional Redis candidate projection that repairs empty/lost cache state from authoritative PostgreSQL before the durable final claim | `server/store/serializable.go`, `queue_sql.go`, `proposal_sql.go`, `proposal_recovery_sql.go`, `match_sql.go`, `redis_candidates.go`, `server/matcher/worker.go` and tests cover retry classification, claim-boundary invariants, player/ticket/playlist mapping, durable queue replay/conflict, owner-scoped queue/proposal recovery, expiry at read and mutation boundaries, response replay/conflict, exact match-promotion replay/conflict, fixed team/slot persistence, zero-row claim aborts, atomic statement ordering, incomplete matcher batches, source failures and empty-index Redis repair; opt-in PostgreSQL execution now covers queue create/replay/fencing, assignment persistence, proposal claim/promotion, accepted-proposal match promotion, participant recovery, unanimous response and rollback of partial claims; worker invocation, allocation runtime and concurrent two-matcher integration tests remain | | 8.19 `[D:8.18]` | **IN PROGRESS.** Pure Go casual lineup requires 2–6 humans with at least one per team, fills missing slots with explicit bots, permits kickoff-only bot-slot backfill and assigns no backfill penalty/rating update; proposal preparation now derives the lineup from formed teams | `server/domain/casual.go`, `formation.go` cover both-team minimum, bot shape, live-play rejection, zero-penalty backfill and casual proposal composition; queue candidate selection, opt-in 10 s backfill proposals, reconnect/leave penalties and live integration remain | | 8.20 `[D:8.18]` | **IN PROGRESS.** Pure Go ranked admission requires six unique verified solo humans, rejects bots/backfill/parties, and allows only random-enabled non-elevated arenas; proposal preparation requires matching metadata for every formed player | `server/domain/ranked.go`, `formation.go` cover count, identity, party, bot/backfill, arena eligibility and formed-player metadata rejection; `ArenaRegistry` integration, allocation wiring and innocent-ticket restoration remain | | 8.21 `[D:8.5,8.20]` | **IN PROGRESS.** Pure Go rating core implements canonical Glicko-2, daily inactivity, ranked 1/3 and casual 1/N human-opponent weights, deterministic opponent ordering, and authoritative draw/overtime/abandon scoring; certified result completion now applies per-player updates inside the durable transaction with lexical row locks and revision increments | `server/domain/rating.go`, `server/store/result_sql.go` and tests cover canonical/inactivity/weight/invalid-input, draw/OT/abandon, ordered participant snapshots, lock/value re-read and rating update SQL; live PostgreSQL rating, seasons and concurrent result transaction tests remain | diff --git a/server/store/match_sql.go b/server/store/match_sql.go new file mode 100644 index 00000000..0b520e9d --- /dev/null +++ b/server/store/match_sql.go @@ -0,0 +1,227 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "sort" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +// AcceptedMatchPlan is the durable hand-off from an accepted proposal to +// allocation. Team and slot originate from the matcher formation and are +// persisted before allocation so later roster issuance cannot re-partition a +// match after players have accepted it. +type AcceptedMatchPlan struct { + MatchID string + ProposalID string + Region string + Protocol int + Players []MatchPlayer +} + +type MatchPlayer struct { + PlayerID string + Team int + Slot int +} + +const AcceptedProposalLockSQL = `SELECT playlist, state +FROM proposals +WHERE proposal_id = $1 +FOR UPDATE` + +const AcceptedProposalParticipantsSQL = `SELECT player_id, ticket_id, response +FROM proposal_participants +WHERE proposal_id = $1 +ORDER BY player_id +FOR UPDATE` + +const AcceptedMatchInsertSQL = `INSERT INTO matches + (match_id, playlist, state, region, protocol_version) +VALUES ($1, $2, 'ALLOCATING', $3, $4) +ON CONFLICT (match_id) DO NOTHING` + +const AcceptedMatchSelectSQL = `SELECT playlist, state, region, protocol_version, server_id +FROM matches +WHERE match_id = $1 +FOR UPDATE` + +const AcceptedMatchParticipantsSQL = `SELECT player_id, ticket_id, slot, team +FROM match_participants +WHERE match_id = $1 +ORDER BY player_id` + +const AcceptedTicketSQL = `UPDATE queue_tickets +SET state = 'ACCEPTED', revision = revision + 1 +WHERE ticket_id = $1 AND player_id = $2 AND state = 'PROPOSED' +RETURNING protocol_version` + +const AcceptedMatchParticipantInsertSQL = `INSERT INTO match_participants + (match_id, player_id, ticket_id, slot, team) +VALUES ($1, $2, $3, $4, $5)` + +// CreateMatchFromAcceptedProposal atomically promotes the exact accepted +// roster into an ALLOCATING match. An existing match ID is an idempotent retry +// only if every durable field and participant assignment matches the request. +func CreateMatchFromAcceptedProposal(ctx context.Context, db *sql.DB, plan AcceptedMatchPlan, now time.Time) error { + if db == nil || now.IsZero() || !validAcceptedMatchPlan(plan) { + return fmt.Errorf("invalid accepted match plan") + } + return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { + var playlist, proposalState string + if err := tx.QueryRowContext(ctx, AcceptedProposalLockSQL, plan.ProposalID).Scan(&playlist, &proposalState); err != nil { + return err + } + if proposalState != string(domain.Accepted) { + return fmt.Errorf("proposal is not accepted") + } + participants, err := acceptedProposalParticipants(ctx, tx, plan) + if err != nil { + return err + } + inserted, err := tx.ExecContext(ctx, AcceptedMatchInsertSQL, plan.MatchID, playlist, plan.Region, plan.Protocol) + if err != nil { + return err + } + changed, err := inserted.RowsAffected() + if err != nil { + return err + } + if changed == 0 { + return verifyAcceptedMatchReplay(ctx, tx, plan, domain.Playlist(playlist), participants) + } + for _, player := range plan.Players { + ticketID := participants[player.PlayerID] + var protocol int + if err := tx.QueryRowContext(ctx, AcceptedTicketSQL, ticketID, player.PlayerID).Scan(&protocol); err != nil { + return fmt.Errorf("accepted ticket transition: %w", err) + } + if protocol != plan.Protocol { + return fmt.Errorf("accepted ticket protocol mismatch") + } + if _, err := tx.ExecContext(ctx, AcceptedMatchParticipantInsertSQL, plan.MatchID, player.PlayerID, ticketID, player.Slot, player.Team); err != nil { + return err + } + } + return nil + }) +} + +func validAcceptedMatchPlan(plan AcceptedMatchPlan) bool { + if plan.MatchID == "" || plan.ProposalID == "" || (plan.Region != "EU" && plan.Region != "NA") || plan.Protocol < 1 || len(plan.Players) < 2 || len(plan.Players) > 6 { + return false + } + players := make(map[string]struct{}, len(plan.Players)) + slots := make(map[int]struct{}, len(plan.Players)) + teams := [2]int{} + for _, player := range plan.Players { + if player.PlayerID == "" || player.Team < 0 || player.Team > 1 || player.Slot < 0 || player.Slot > 5 { + return false + } + if _, exists := players[player.PlayerID]; exists { + return false + } + if _, exists := slots[player.Slot]; exists { + return false + } + players[player.PlayerID] = struct{}{} + slots[player.Slot] = struct{}{} + teams[player.Team]++ + } + return teams[0] > 0 && teams[1] > 0 +} + +func acceptedProposalParticipants(ctx context.Context, tx *sql.Tx, plan AcceptedMatchPlan) (map[string]string, error) { + rows, err := tx.QueryContext(ctx, AcceptedProposalParticipantsSQL, plan.ProposalID) + if err != nil { + return nil, err + } + defer rows.Close() + participants := make(map[string]string, len(plan.Players)) + for rows.Next() { + var playerID, ticketID, response string + if err := rows.Scan(&playerID, &ticketID, &response); err != nil { + return nil, err + } + if response != string(domain.AcceptedResponse) { + return nil, fmt.Errorf("proposal participant has not accepted") + } + participants[playerID] = ticketID + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(participants) != len(plan.Players) { + return nil, fmt.Errorf("proposal participants do not match accepted plan") + } + for _, player := range plan.Players { + if participants[player.PlayerID] == "" { + return nil, fmt.Errorf("accepted plan includes non-participant") + } + } + return participants, nil +} + +func verifyAcceptedMatchReplay(ctx context.Context, tx *sql.Tx, plan AcceptedMatchPlan, playlist domain.Playlist, tickets map[string]string) error { + var existingPlaylist, state, region string + var protocol int + var serverID sql.NullString + if err := tx.QueryRowContext(ctx, AcceptedMatchSelectSQL, plan.MatchID).Scan(&existingPlaylist, &state, ®ion, &protocol, &serverID); err != nil { + return err + } + if existingPlaylist != string(playlist) || state != string(domain.Allocating) || region != plan.Region || protocol != plan.Protocol || serverID.Valid { + return domain.ErrConflict + } + rows, err := tx.QueryContext(ctx, AcceptedMatchParticipantsSQL, plan.MatchID) + if err != nil { + return err + } + defer rows.Close() + existing := make(map[string]MatchPlayer, len(plan.Players)) + for rows.Next() { + var player MatchPlayer + var ticketID string + if err := rows.Scan(&player.PlayerID, &ticketID, &player.Slot, &player.Team); err != nil { + return err + } + if tickets[player.PlayerID] != ticketID { + return domain.ErrConflict + } + existing[player.PlayerID] = player + } + if err := rows.Err(); err != nil { + return err + } + if len(existing) != len(plan.Players) { + return domain.ErrConflict + } + for _, player := range plan.Players { + if existing[player.PlayerID] != player { + return domain.ErrConflict + } + } + return nil +} + +// MatchPlayersFromTeams turns the deterministic matcher partition into the +// persisted six-slot topology. Each team is sorted by player ID first, so slot +// assignment does not depend on cache/database row order. +func MatchPlayersFromTeams(teams domain.Teams) ([]MatchPlayer, error) { + if len(teams.Team0) == 0 || len(teams.Team1) == 0 || len(teams.Team0)+len(teams.Team1) > 6 { + return nil, fmt.Errorf("invalid match teams") + } + result := make([]MatchPlayer, 0, len(teams.Team0)+len(teams.Team1)) + add := func(team int, players []domain.Candidate) { + ordered := append([]domain.Candidate(nil), players...) + sort.Slice(ordered, func(i, j int) bool { return ordered[i].PlayerID < ordered[j].PlayerID }) + for index, player := range ordered { + result = append(result, MatchPlayer{PlayerID: player.PlayerID, Team: team, Slot: team*3 + index}) + } + } + add(0, teams.Team0) + add(1, teams.Team1) + return result, nil +} diff --git a/server/store/match_sql_test.go b/server/store/match_sql_test.go new file mode 100644 index 00000000..a0cf10d4 --- /dev/null +++ b/server/store/match_sql_test.go @@ -0,0 +1,74 @@ +package store + +import ( + "testing" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +func TestAcceptedMatchSQLPreservesAtomicProposalToMatchBoundary(t *testing.T) { + checks := map[string][]string{ + AcceptedProposalLockSQL: {"FOR UPDATE", "proposal_id = $1"}, + AcceptedProposalParticipantsSQL: {"response", "ORDER BY player_id", "FOR UPDATE"}, + AcceptedMatchInsertSQL: {"'ALLOCATING'", "ON CONFLICT (match_id) DO NOTHING"}, + AcceptedTicketSQL: {"state = 'ACCEPTED'", "state = 'PROPOSED'", "revision = revision + 1"}, + AcceptedMatchParticipantInsertSQL: {"match_participants", "slot", "team"}, + } + for query, fragments := range checks { + for _, fragment := range fragments { + if !contains(query, fragment) { + t.Fatalf("query %q missing %q", query, fragment) + } + } + } +} + +func TestAcceptedMatchPlanRejectsInvalidPlansBeforeDatabaseUse(t *testing.T) { + valid := AcceptedMatchPlan{ + MatchID: "match-1", ProposalID: "proposal-1", Region: "EU", Protocol: 1, + Players: []MatchPlayer{{PlayerID: "player-a", Team: 0, Slot: 0}, {PlayerID: "player-b", Team: 1, Slot: 3}}, + } + if !validAcceptedMatchPlan(valid) { + t.Fatal("valid accepted match plan rejected") + } + for name, mutate := range map[string]func(*AcceptedMatchPlan){ + "no second team": func(p *AcceptedMatchPlan) { p.Players[1].Team = 0 }, + "duplicate slot": func(p *AcceptedMatchPlan) { p.Players[1].Slot = 0 }, + "duplicate player": func(p *AcceptedMatchPlan) { p.Players[1].PlayerID = "player-a" }, + "bad region": func(p *AcceptedMatchPlan) { p.Region = "AP" }, + } { + plan := valid + plan.Players = append([]MatchPlayer(nil), valid.Players...) + mutate(&plan) + if validAcceptedMatchPlan(plan) { + t.Fatalf("%s plan accepted", name) + } + } + if err := CreateMatchFromAcceptedProposal(nil, nil, valid, time.Now()); err == nil { + t.Fatal("nil database accepted") + } +} + +func TestMatchPlayersFromTeamsUsesDeterministicTeamSlots(t *testing.T) { + teams := domain.Teams{ + Team0: []domain.Candidate{{PlayerID: "bravo"}, {PlayerID: "alpha"}}, + Team1: []domain.Candidate{{PlayerID: "delta"}, {PlayerID: "charlie"}}, + } + players, err := MatchPlayersFromTeams(teams) + if err != nil { + t.Fatal(err) + } + want := []MatchPlayer{ + {PlayerID: "alpha", Team: 0, Slot: 0}, {PlayerID: "bravo", Team: 0, Slot: 1}, + {PlayerID: "charlie", Team: 1, Slot: 3}, {PlayerID: "delta", Team: 1, Slot: 4}, + } + if len(players) != len(want) { + t.Fatalf("players = %+v", players) + } + for index := range want { + if players[index] != want[index] { + t.Fatalf("player %d = %+v, want %+v", index, players[index], want[index]) + } + } +} diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 770f6be1..10a37949 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -94,6 +94,55 @@ func TestPostgreSQLAllocatorClaimReplayAndCapacityFence(t *testing.T) { } } +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) + } + conflict := plan + conflict.Players = append([]MatchPlayer(nil), plan.Players...) + conflict.Players[1].Slot = 4 + if err := CreateMatchFromAcceptedProposal(ctx, db, conflict, now.Add(2*time.Second)); err == nil { + t.Fatal("conflicting match promotion replay was accepted") + } +} + func TestPostgreSQLQueueAdapterAgainstRealDatabase(t *testing.T) { db := openIntegrationPostgres(t) applyIntegrationMigrations(t, db)