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

fix(multiplayer): complete live results atomically

Browse Source
This commit is contained in:
Josh Creek
2026-09-03 13:29:31 +01:00
parent 8c28374eb4
commit 2e9da3032c
6 changed files with 112 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/domain/result.go
+3
View File
@@ -102,6 +102,9 @@ func (s *ResultStore) Submit(resultID string, result MatchResult, binding Worklo
if resultID == "" || !sameBinding(s.expected, binding) {
return ResultReceipt{}, false, ErrResultBinding
}
if now.IsZero() {
return ResultReceipt{}, false, ErrResultInvalid
}
if err := validateResult(s.expected, result); err != nil {
return ResultReceipt{}, false, err
}
server/domain/result_test.go
+11
View File
@@ -36,6 +36,17 @@ func TestResultStoreBindsWorkloadAndMakesIdenticalDuplicateInert(t *testing.T) {
}
}
func TestResultStoreRejectsMissingAuthoritativeTime(t *testing.T) {
binding := testBinding()
store, err := NewResultStore(binding)
if err != nil {
t.Fatal(err)
}
if _, _, err := store.Submit("result-1", testResult(), binding, time.Time{}); !errors.Is(err, ErrResultInvalid) {
t.Fatalf("zero-time result error = %v", err)
}
}
func TestConflictingResultIsInertAndIntegritySuppressesRating(t *testing.T) {
now := time.Unix(1000, 0)
binding := testBinding()
server/store/postgres_integration_test.go
+8 -1
View File
@@ -1267,7 +1267,7 @@ func TestPostgreSQLConcurrentIdenticalResultSubmissionAppliesRatingsExactlyOnce(
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', 'RESULT_PENDING', 'NA', 1, 'result-race-server')`); err != nil {
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 {
@@ -1306,6 +1306,13 @@ func TestPostgreSQLConcurrentIdenticalResultSubmissionAppliesRatingsExactlyOnce(
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 {
server/store/result_sql.go
+71 -5
View File
@@ -33,6 +33,16 @@ FROM matches
WHERE match_id = $1 AND server_id = $2
FOR UPDATE`
const ResultMatchPendingSQL = `UPDATE matches
SET state = 'RESULT_PENDING', revision = revision + 1
WHERE match_id = $1 AND state = 'LIVE'`
const ResultTicketsPendingSQL = `UPDATE queue_tickets q
SET state = 'RESULT_PENDING', revision = revision + 1
FROM match_participants mp
WHERE mp.match_id = $1 AND q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id
AND mp.participation_active AND q.state = 'LIVE'`
const ResultMatchCompleteSQL = `UPDATE matches
SET state = 'COMPLETED', revision = revision + 1, completed_at = $2
WHERE match_id = $1 AND state = 'RESULT_PENDING'`
@@ -41,6 +51,14 @@ const ResultReceiptCommitSQL = `UPDATE result_receipts
SET committed_at = COALESCE(committed_at, $2)
WHERE match_id = $1`
const ResultTicketsCompleteSQL = `UPDATE queue_tickets q
SET state = 'COMPLETED', revision = revision + 1
FROM match_participants mp
WHERE mp.match_id = $1 AND q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id
AND mp.participation_active AND q.state = 'RESULT_PENDING'`
const ResultParticipantCountSQL = `SELECT count(*) FROM match_participants WHERE match_id = $1 AND participation_active`
const ResultOutboxSQL = `INSERT INTO outbox
(event_id, aggregate_type, aggregate_id, revision, event_type, payload)
VALUES ($1, 'match', $2, $3, 'match_completed', $4)`
@@ -51,11 +69,12 @@ WHERE player_id = ANY($1)
ORDER BY player_id
FOR UPDATE`
const MatchParticipantRatingsSQL = `SELECT mp.player_id, mp.team, r.rating, r.deviation,
const MatchParticipantRatingsSQL = `SELECT mp.player_id, mp.team, mp.abandoned_at, r.rating, r.deviation,
r.volatility, r.ranked_games, r.updated_at
FROM match_participants mp
JOIN ratings r ON r.player_id = mp.player_id
WHERE mp.match_id = $1
AND mp.participation_active
ORDER BY mp.player_id`
const RatingValuesSQL = `SELECT player_id, rating, deviation, volatility, ranked_games, updated_at
@@ -101,7 +120,7 @@ func CompleteResultWithResult(ctx context.Context, db *sql.DB, receipt domain.Re
}
func completeResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceipt, serverID, eventID string, payload []byte, now time.Time, result *domain.MatchResult) error {
if receipt.ResultID == "" || receipt.MatchID == "" || serverID == "" || eventID == "" || len(payload) == 0 {
if db == nil || receipt.ResultID == "" || receipt.MatchID == "" || len(receipt.ResultNonce) < 16 || len(receipt.ResultNonce) > 128 || receipt.ReceivedAt.IsZero() || now.IsZero() || serverID == "" || eventID == "" || len(payload) == 0 || (receipt.IntegrityState != domain.IntegrityCertified && receipt.IntegrityState != domain.IntegritySuppressed && receipt.IntegrityState != domain.IntegrityReview) {
return fmt.Errorf("invalid result transaction arguments")
}
return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
@@ -121,7 +140,7 @@ func completeResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceip
return fmt.Errorf("result receipt conflict: %w", err)
}
if priorID != receipt.ResultID || priorMatch != receipt.MatchID || priorNonce != receipt.ResultNonce || priorIntegrity != string(receipt.IntegrityState) || !bytes.Equal(priorDigest, receipt.PayloadDigest[:]) {
return fmt.Errorf("conflicting result receipt")
return fmt.Errorf("%w: durable receipt differs", domain.ErrResultConflict)
}
}
var lockedMatch, playlist, state string
@@ -133,9 +152,23 @@ func completeResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceip
_, err := tx.ExecContext(ctx, ResultReceiptCommitSQL, receipt.MatchID, now)
return err
}
if state == string(domain.Live) {
updated, err := tx.ExecContext(ctx, ResultMatchPendingSQL, receipt.MatchID)
if err != nil {
return err
}
if changed, err := updated.RowsAffected(); err != nil || changed != 1 {
return fmt.Errorf("result-pending transition lost race")
}
state = string(domain.ResultPending)
revision++
}
if state != "RESULT_PENDING" {
return fmt.Errorf("match is not result-pending: %s", state)
}
if _, err := tx.ExecContext(ctx, ResultTicketsPendingSQL, receipt.MatchID); err != nil {
return err
}
if result != nil && domain.RatingEligible(receipt) {
if err := applyResultRatings(ctx, tx, receipt.MatchID, domain.Playlist(playlist), *result, now); err != nil {
return err
@@ -152,6 +185,21 @@ func completeResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceip
if changed != 1 {
return fmt.Errorf("result completion lost race")
}
completedTickets, err := tx.ExecContext(ctx, ResultTicketsCompleteSQL, receipt.MatchID)
if err != nil {
return err
}
var participants int64
if err := tx.QueryRowContext(ctx, ResultParticipantCountSQL, receipt.MatchID).Scan(&participants); err != nil {
return err
}
completed, err := completedTickets.RowsAffected()
if err != nil {
return err
}
if completed != participants {
return fmt.Errorf("result ticket completion mismatch: completed=%d participants=%d", completed, participants)
}
if _, err := tx.ExecContext(ctx, ResultReceiptCommitSQL, receipt.MatchID, now); err != nil {
return err
}
@@ -165,6 +213,7 @@ type participantRating struct {
team int
rating domain.Rating
rankedGames int
abandoned bool
}
func applyResultRatings(ctx context.Context, tx *sql.Tx, matchID string, playlist domain.Playlist, result domain.MatchResult, now time.Time) error {
@@ -176,17 +225,29 @@ func applyResultRatings(ctx context.Context, tx *sql.Tx, matchID string, playlis
var players []participantRating
for rows.Next() {
var player participantRating
if err := rows.Scan(&player.playerID, &player.team, &player.rating.Value, &player.rating.RD, &player.rating.Volatility, &player.rankedGames, &player.rating.LastRatedAt); err != nil {
var abandonedAt sql.NullTime
if err := rows.Scan(&player.playerID, &player.team, &abandonedAt, &player.rating.Value, &player.rating.RD, &player.rating.Volatility, &player.rankedGames, &player.rating.LastRatedAt); err != nil {
return err
}
player.abandoned = abandonedAt.Valid
players = append(players, player)
}
if err := rows.Err(); err != nil {
return err
}
if err := rows.Close(); err != nil {
return err
}
if len(players) == 0 {
return nil
}
var participantCount int
if err := tx.QueryRowContext(ctx, ResultParticipantCountSQL, matchID).Scan(&participantCount); err != nil {
return err
}
if participantCount != len(players) {
return fmt.Errorf("result rating roster is incomplete")
}
ids := make([]string, len(players))
for i := range players {
ids[i] = players[i].playerID
@@ -239,7 +300,12 @@ func applyResultRatings(ctx context.Context, tx *sql.Tx, matchID string, playlis
if err := values.Close(); err != nil {
return err
}
outcome := domain.MatchOutcome{Team0Score: result.Team0Score, Team1Score: result.Team1Score}
outcome := domain.MatchOutcome{Team0Score: result.Team0Score, Team1Score: result.Team1Score, Abandoners: make(map[string]bool)}
for _, player := range players {
if player.abandoned {
outcome.Abandoners[player.playerID] = true
}
}
for _, player := range players {
current, ok := ratings[player.playerID]
if !ok {
server/store/result_sql_test.go
+18 -1
View File
@@ -2,6 +2,7 @@ package store
import (
"context"
"database/sql"
"testing"
"time"
@@ -13,11 +14,15 @@ func TestResultSQLPreservesReceiptConflictAndAtomicCommitBoundaries(t *testing.T
ResultReceiptInsertSQL: {"ON CONFLICT DO NOTHING", "payload_digest", "integrity_state"},
ResultReceiptSelectSQL: {"FOR UPDATE", "committed_at"},
ResultCommitLockSQL: {"server_id = $2", "FOR UPDATE"},
ResultMatchPendingSQL: {"state = 'RESULT_PENDING'", "state = 'LIVE'", "revision = revision + 1"},
ResultTicketsPendingSQL: {"queue_tickets", "match_participants", "participation_active", "state = 'LIVE'"},
ResultMatchCompleteSQL: {"state = 'RESULT_PENDING'", "revision = revision + 1"},
ResultTicketsCompleteSQL: {"state = 'COMPLETED'", "state = 'RESULT_PENDING'", "match_participants", "participation_active"},
ResultParticipantCountSQL: {"count(*)", "match_participants", "match_id = $1", "participation_active"},
ResultReceiptCommitSQL: {"COALESCE(committed_at", "committed_at"},
ResultOutboxSQL: {"match_completed", "aggregate_id", "revision"},
RatingLockSQL: {"ORDER BY player_id", "FOR UPDATE"},
MatchParticipantRatingsSQL: {"match_participants", "JOIN ratings", "ORDER BY mp.player_id"},
MatchParticipantRatingsSQL: {"match_participants", "abandoned_at", "JOIN ratings", "participation_active", "ORDER BY mp.player_id"},
RatingValuesSQL: {"player_id = ANY($1)", "ORDER BY player_id"},
RatingUpdateSQL: {"ranked_games = ranked_games + $5", "revision = revision + 1"},
}
@@ -46,6 +51,18 @@ func TestCompleteResultWithResultRejectsReceiptResultMismatchBeforeDatabaseUse(t
}
}
func TestCompleteResultRejectsIncompleteReceiptBeforeDatabaseUse(t *testing.T) {
now := time.Unix(100, 0).UTC()
receipt := domain.ResultReceipt{ResultID: "result", MatchID: "match", ResultNonce: "nonce-1234567890123456", IntegrityState: domain.IntegrityCertified, ReceivedAt: now}
if err := CompleteResult(context.Background(), nil, receipt, "server", "event", []byte("payload"), now); err == nil {
t.Fatal("nil database accepted")
}
receipt.ReceivedAt = time.Time{}
if err := CompleteResult(context.Background(), &sql.DB{}, receipt, "server", "event", []byte("payload"), now); err == nil {
t.Fatal("zero receipt time accepted")
}
}
func contains(value, fragment string) bool {
for i := 0; i+len(fragment) <= len(value); i++ {
if value[i:i+len(fragment)] == fragment {