From 2e9da3032cd2d4856faa7dd642e0219498aa8b27 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Thu, 3 Sep 2026 13:29:31 +0100 Subject: [PATCH] fix(multiplayer): complete live results atomically --- multiplayer-next.md | 2 +- server/domain/result.go | 3 + server/domain/result_test.go | 11 ++++ server/store/postgres_integration_test.go | 9 ++- server/store/result_sql.go | 76 +++++++++++++++++++++-- server/store/result_sql_test.go | 19 +++++- 6 files changed, 112 insertions(+), 8 deletions(-) diff --git a/multiplayer-next.md b/multiplayer-next.md index 2560350d..36d41120 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1213,7 +1213,7 @@ production fallback. | 8.22 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked profile exposes the first ten games as provisional, derives tiers only through validated backend-owned rating bands, and keeps casual ratings outside the API; authenticated HTTP now returns the authoritative ranked view. **"Authoritative ranked view" was durable-adapter-shaped but had no durable adapter**: `rankedProfile`/`profile` only ever read an in-memory map, so every real `GET /v1/profile/ranked` 404'd regardless of a player's actual rating. `RankedProfileProvider` (interface) + `store.PostgresRankedProfiles` close it, preferred over the map when set so existing tests/literals are unaffected; `LastSeasonID`/`SeasonHistory` deliberately left unset (no season pointer on `ratings`, needs its own query/semantics) | `server/domain/rating.go`, `tier_test.go` and `server/api/service.go` cover provisional override, exact band boundaries, malformed policy rejection, session authentication and ranked-only response fields; `server/store/ranked_profile_sql.go`, verified against real PostgreSQL via curl (a fresh identity correctly 404s through the real adapter) and via §8.40's integration test (`control_plane_smoke.gd` now asserts this exact 404 round-trips before queueing); persisted tier policy, client UI and reconnect transport remain | | 8.23 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked-only season policy compresses 25% toward 1500, clamps RD to 200–350, preserves volatility/history, is idempotent by season ID, and defines exact 12-week windows/due detection; migration and Go store now persist a per-player/per-season marker and rating update atomically; `cmd/maintenance` runs bounded due-season batches with signal-bound shutdown | `server/domain/rating.go`, `season_test.go`, `server/migrations/0001_initial.sql` and `server/store/maintenance_sql.go` cover compression, floor/cap, duplicate replay, window boundary, completed-season idempotence, bounded enumeration, row locking and conflict-safe rollover markers; opt-in PostgreSQL execution now covers the durable rating update, marker creation and duplicate replay without a second compression — re-run live for the first time as part of the wider integration-suite verification below, after fixing a test setup gap (a missing `seasons` row tripped the `ranked_season_rollovers` foreign key before the rollover logic itself ran); live maintenance/DB execution remains | | 8.24 `[D:8.9,8.20,8.21]` | **IN PROGRESS.** Ranked connection policy binds match/server/player/Steam identity/team/global slot/protocol/expiry, permits a 60-second same-token reclaim with monotonically increasing server-owned generations, and applies the rolling 7-day 5 m/15 m/1 h/24 h abandon ladder. Admission now rejects an already-connected duplicate, zero-time operations, disconnect-before-admit, duplicate disconnect attempts that could extend grace, and clock-reversed disconnect/reclaim; Godot applies the same reversed-clock fence to its allowlisted signed-token reservation | Go/Godot adversarial fixtures cover signature tampering, every claim binding, active duplicate admission, repeated valid reclaim, old-generation fencing, exact grace boundary, expiry, zero/reversed clocks, and deterministic cooldown ordering. Persistent cross-process lease fencing and full match/result integration remain | -| 8.25 `[D:8.10,8.24]` | **IN PROGRESS.** Pure Go result policy binds match/server/workload identity, hashes canonical payloads, makes identical retries idempotent, leaves conflicts inert, separates integrity eligibility, classifies roster/simulation/result/fairness evidence, validates annotation signatures/digests, and exposes 5 m alert/30 m review delivery thresholds; Go store SQL now executes receipt → match lock → certified rating updates → completion → receipt acknowledgment → outbox atomically, and exposes bounded ordered outbox reads plus publish acknowledgements for replayable fan-out; `OutboxDispatcher` now delivers in order and acknowledges only after successful fan-out; the API validates workload-bound server result submissions and the PostgreSQL adapter repeats domain validation before invoking this durable boundary; production now also runs a filtered `match_completed` dispatcher that turns each committed result into targeted `COMPLETED` state events for every durable participant, without acknowledging proposal rows | `server/domain/result.go`, `workload.go`, `server/workload/jwt.go`, `server/api/service.go`, `server/api/outbox.go`, `server/store/result_sql.go` and `outbox.go` plus adversarial fixtures cover credential binding, duplicate/conflict, annotation forgery, server/match mismatch, invalid direct-adapter payloads, delivery-outage-versus-integrity classification, commit ordering, idempotent SQL reconciliation, ordered rating locks, certified-update gating, unpublished-event replay/ack boundaries, event-type isolation and delivery-before-ack failure ordering; opt-in PostgreSQL execution now covers result-pending completion, durable receipt/outbox publication, ack removal, identical replay and conflicting replay rejection, and real concurrent identical/conflicting submissions; `scripts/run_result_fanout_integration.sh` now drives real PostgreSQL → API WebSocket delivery for an authenticated participant; production credential verification, Agones annotation persistence/reconciliation and integrity evidence adapters remain | +| 8.25 `[D:8.10,8.24]` | **IN PROGRESS.** Workload-bound result policy validates match/server identity, nonce, scores, integrity state, canonical digest, and idempotent receipt conflicts. Result ingestion now owns the complete serializable production transition: a `LIVE` match and its active tickets advance through `RESULT_PENDING`, certified ratings are computed from locked authoritative participant rows, then match/tickets become `COMPLETED`, the receipt is acknowledged, and one revisioned outbox event is inserted. Inactive pre-match no-shows are excluded; an active participant's durable `abandoned_at` forces loss scoring. Missing rating rows or ticket-count divergence fail the whole transaction. Zero-time/incomplete receipts fail before database use and conflicts wrap `ErrResultConflict` | Domain/store/API/outbox tests cover workload and digest binding, identical/conflicting concurrency, direct `LIVE` completion, active-ticket completion, abandonment rating input, incomplete roster failure, integrity suppression, ordered rating locks, receipt/outbox atomicity, fan-out retry/ack ordering, and delivery health. PostgreSQL-tagged regressions compile; prior live result/rating/race/fan-out runs remain valid, while the direct-live lifecycle change awaits a live database rerun. Production credentials, Agones annotation persistence/reconciliation, and integrity-evidence adapters remain | #### 8D — Agones, allocation and regional scaling diff --git a/server/domain/result.go b/server/domain/result.go index b8c9c689..6c11e484 100644 --- a/server/domain/result.go +++ b/server/domain/result.go @@ -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 } diff --git a/server/domain/result_test.go b/server/domain/result_test.go index bb070e0b..06c3571d 100644 --- a/server/domain/result_test.go +++ b/server/domain/result_test.go @@ -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() diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 3c0ec839..cb4a991b 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -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 { diff --git a/server/store/result_sql.go b/server/store/result_sql.go index 81389a6a..5b6df773 100644 --- a/server/store/result_sql.go +++ b/server/store/result_sql.go @@ -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 { diff --git a/server/store/result_sql_test.go b/server/store/result_sql_test.go index 93612932..7eed594a 100644 --- a/server/store/result_sql_test.go +++ b/server/store/result_sql_test.go @@ -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 {