diff --git a/multiplayer-next.md b/multiplayer-next.md index a751608e..03e839eb 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1202,7 +1202,7 @@ production fallback. | # | Task | Acceptance | |---|---|---| -| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue policy and PostgreSQL enforce one active ticket per verified player, 10 s heartbeat/30 s expiry, retry-safe owner/revision-scoped create/heartbeat/cancel, and deterministic candidate projection. Client cancellation is now limited to `QUEUED`/`PROPOSED`; it cannot overwrite match-owned `ACCEPTED` through `LIVE` lifecycle states. A locked rejection classifier maps missing ticket, wrong owner, expiry, stale revision, and invalid state to distinct domain/API outcomes without weakening the atomic mutation predicate. Redis is an optional rebuildable projection over authoritative PostgreSQL | Domain/store/API tests cover ownership, expiry, idempotency, candidate binding, exact mutation-state fences, live-ticket cancellation rejection, stale revision classification, concurrent create/heartbeat races, durable-source cache repair, Redis TTL/lost-keyspace behavior, and playlist/build/protocol compatibility. PostgreSQL-tagged lifecycle regressions compile and prior live runs cover the queue races; this state-fence change awaits a live database rerun. Live Redis failover-under-load and worker integration remain | +| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue policy and PostgreSQL enforce one active ticket per verified player, 10 s heartbeat/30 s expiry, retry-safe owner/revision-scoped create/heartbeat/cancel, and deterministic candidate projection. Client cancellation is limited to `QUEUED`/`PROPOSED`; it cannot overwrite match-owned `ACCEPTED` through `LIVE` lifecycle states. A locked rejection classifier maps missing ticket, wrong owner, expiry, stale revision, and invalid state to distinct domain/API outcomes without weakening the atomic mutation predicate. Queue admission also honors both pre-live and live ranked abandonment penalties, so an expired reconnect cannot immediately requeue after result completion. Redis is an optional rebuildable projection over authoritative PostgreSQL | Domain/store/API tests cover ownership, expiry, idempotency, candidate binding, exact mutation-state fences, live-ticket cancellation rejection, stale revision classification, abandonment cooldown selection, concurrent create/heartbeat races, durable-source cache repair, Redis TTL/lost-keyspace behavior, and playlist/build/protocol compatibility. PostgreSQL-tagged lifecycle regressions compile and prior live runs cover the queue races; live database reruns remain blocked by Docker storage. Live Redis failover-under-load and worker integration remain | | 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. **Fixed a real crash-loop**: `Worker.Run` treated every `RunOnce` error as fatal to the whole loop, including "no compatible candidates" (`FormFromQueue`'s completely routine answer when currently-queued players share no verified region) — found building a live two-player integration attempt (see below): two real players with no common region crashed the entire matcher process, taking matchmaking down for every other player in the playlist, and would crash-loop again on restart since the same incompatible candidates stay queued. Now only genuine static misconfiguration (`ErrWorkerNotConfigured`/`ErrUnsupportedPlaylist`/`ErrInvalidMatcherSize`) stops the loop; everything else retries next interval | `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; two new tests cover `Run` (not just `RunOnce`) surviving a per-pass error via a real concurrent goroutine, and still stopping immediately on a real configuration error, both clean across repeated `-race` runs. **A live two-player Godot proposal integration attempt is on disk but not committed**: `Game/tests/control_plane_proposal_smoke.gd`/`.tscn` and `scripts/verify_control_plane_proposal_integration.sh` exist and found the bug above, but the session paused running further concurrent headless Godot processes after discovering they'd been causing native engine crashes (macOS crash reporter, `EXC_BAD_ACCESS`/`SIGBUS`) intermittently all session, confirmed by the user; the two-player script was never itself verified to a clean pass. 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 casual/ranked decline and timeout cooldowns with ranked escalation, and exposes revisioned idempotent responses through the authenticated API. Proposal closure now atomically separates offenders from innocents: a decliner's ticket is `CANCELLED`; a timed-out player's ticket is `EXPIRED`; accepted or otherwise innocent participants return to `QUEUED` with their original `enqueued_at` and refreshed expiry. Direct queue cancellation closes the open proposal and requeues remaining participants immediately. Late API responses commit expiry, timeout penalties, and ticket release before returning `ErrProposalClosed`; recovery of an old declined proposal cannot misclassify its pending innocents as timeouts. Cooldown history rejects future, foreign-playlist, and invalid-kind events, and database rows are closed before penalty writes | Domain/store/API fixtures cover partial/unanimous response, expiry, replay/conflict, stale revision, exact cooldown windows/escalation, corrupt history filtering, offender ticket termination, innocent precedence preservation, direct-cancel cascade, and the former late-response rollback. PostgreSQL-tagged regressions compile and assert the durable split and penalty rows; the full local Go suite passes. Live PostgreSQL execution and allocation integration remain | @@ -1212,7 +1212,7 @@ production fallback. | 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 and seasons execution now covered (§8.23), and concurrent result transaction cases are covered: `TestPostgreSQLConcurrentIdenticalResultSubmissionAppliesRatingsExactlyOnce` races 5 identical submissions and confirms one rating application, while `TestPostgreSQLConcurrentConflictingResultSubmissionsKeepOneReceipt` races different payloads and confirms exactly one winner, one conflict, one receipt and one completion event; live maintenance/DB execution remains | | 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 rejects active duplicates, zero/reversed clocks, disconnect-before-admit, and duplicate disconnects. PostgreSQL persists generation/disconnect leases with serializable CAS: stale disconnects cannot evict newer generations, active leases cannot be reclaimed, initial admission requires an unexpired assignment, and later reconnects use the durable grace boundary. Godot now consumes that lease before admission, reconciles known-generation outage events in order, closes future admissions on reconciliation divergence, and rejects unsafe generation-zero outage fallback | Go/store/API/Godot adversarial fixtures cover signature tampering, every binding, replay/conflict semantics, active duplicate admission, repeated valid reclaim, stale-generation fencing, exact grace boundary, process recovery, expiry, zero/reversed clocks, malformed JSON generations, deterministic cooldown ordering, legacy-row migration, and rolling-upgrade 204 compatibility. The 207-test Godot harness and focused Go suites pass. Abandonment persistence and live PostgreSQL/process-restart/outage execution remain | +| 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 rejects active duplicates, zero/reversed clocks, disconnect-before-admit, and duplicate disconnects. PostgreSQL persists generation/disconnect leases with serializable CAS: stale disconnects cannot evict newer generations, active leases cannot be reclaimed, initial admission requires an unexpired assignment, and later reconnects use the durable grace boundary. Godot consumes that lease before admission, reconciles known-generation outage events in order, closes future admissions on reconciliation divergence, and rejects unsafe generation-zero outage fallback. Maintenance now turns a ranked `LIVE` lease expired beyond 60 seconds into `abandoned_at`, a durable `MATCH_ABANDONED` cooldown, and a revisioned outbox event without releasing the participant or `LIVE` ticket needed by the result transaction | Go/store/API/Godot adversarial fixtures cover signature tampering, every binding, replay/conflict semantics, active duplicate admission, repeated valid reclaim, stale-generation fencing, grace boundaries, process recovery, expiry, zero/reversed clocks, malformed JSON generations, deterministic cooldown ordering, legacy-row migration, rolling-upgrade 204 compatibility, result-roster retention, and cursor-pool safety. The 207-test Godot harness, focused Go suites, and PostgreSQL-tagged abandonment regression compile. Live PostgreSQL/process-restart/outage execution remains blocked by Docker storage | | 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/cmd/maintenance/main.go b/server/cmd/maintenance/main.go index 44960970..ee14706b 100644 --- a/server/cmd/maintenance/main.go +++ b/server/cmd/maintenance/main.go @@ -25,6 +25,7 @@ func main() { stalledAllocationDeadline := flag.Duration("stalled-allocation-deadline", 2*time.Minute, "reclaim a match stuck in ALLOCATING/PROCESS_READY/ASSIGNMENT_READY (server crashed or was reclaimed before registering) after this long, requeuing every participant without penalty") stalledAllocationBatch := flag.Int("stalled-allocation-batch", 100, "maximum stalled matches reclaimed per pass") initialConnectBatch := flag.Int("initial-connect-batch", 100, "maximum pre-live matches evaluated per pass") + liveAbandonmentBatch := flag.Int("live-abandonment-batch", 100, "maximum live ranked matches evaluated for expired reconnect leases per pass") flag.Parse() if *dsn == "" { fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required") @@ -35,8 +36,8 @@ func main() { if *stalledAllocationDeadline <= 0 || *stalledAllocationBatch < 1 || *stalledAllocationBatch > 1000 { fatalf("invalid stalled-allocation deadline or batch") } - if *initialConnectBatch < 1 || *initialConnectBatch > 1000 { - fatalf("invalid initial-connect batch") + if *initialConnectBatch < 1 || *initialConnectBatch > 1000 || *liveAbandonmentBatch < 1 || *liveAbandonmentBatch > 1000 { + fatalf("invalid initial-connect or live-abandonment batch") } db, err := sql.Open("pgx", *dsn) if err != nil { @@ -77,6 +78,13 @@ func main() { if reconciled > 0 { log.Printf("reconciled %d initial-connect outcomes", reconciled) } + abandoned, err := store.ReconcileLiveAbandonments(ctx, db, now, *liveAbandonmentBatch) + if err != nil { + fatalf("live-abandonment maintenance: %v", err) + } + if abandoned > 0 { + log.Printf("recorded expired reconnect leases in %d live matches", abandoned) + } } runGeneral(time.Now().UTC()) diff --git a/server/domain/reconnect.go b/server/domain/reconnect.go index 8a6b1857..771b8f74 100644 --- a/server/domain/reconnect.go +++ b/server/domain/reconnect.go @@ -156,6 +156,38 @@ type Abandonment struct { AbandonedAt time.Time } +// ReconnectParticipant is the durable subset needed to evaluate an expired +// live reconnect lease. Connected players are deliberately absent: only a +// persisted disconnect can start a player-caused abandon clock. +type ReconnectParticipant struct { + PlayerID string + DisconnectedAt time.Time +} + +// PlanRankedAbandonments turns expired durable reconnect leases into the +// same rolling cooldown ladder used by pre-live ranked no-shows. Future +// disconnect timestamps are ignored rather than penalised: they can only be +// an infrastructure clock anomaly, not a player abandonment. +func PlanRankedAbandonments(now time.Time, participants []ReconnectParticipant, priorAbandons map[string][]time.Time) ([]Abandonment, error) { + if now.IsZero() { + return nil, fmt.Errorf("invalid reconnect-abandonment time") + } + seen := make(map[string]bool, len(participants)) + result := make([]Abandonment, 0, len(participants)) + for _, participant := range participants { + if participant.PlayerID == "" || participant.DisconnectedAt.IsZero() || seen[participant.PlayerID] { + return nil, fmt.Errorf("invalid reconnect participant") + } + seen[participant.PlayerID] = true + if now.Before(participant.DisconnectedAt) || now.Sub(participant.DisconnectedAt) <= RankedReconnectGrace { + continue + } + result = append(result, Abandonment{PlayerID: participant.PlayerID, Cooldown: abandonCooldown(priorAbandons[participant.PlayerID], now), AbandonedAt: now}) + } + sort.Slice(result, func(i, j int) bool { return result[i].PlayerID < result[j].PlayerID }) + return result, nil +} + // ExpireGrace marks every disconnected player whose 60-second reclaim window // has elapsed. The returned list is lexical for stable audit/event ordering. func (r *RankedConnections) ExpireGrace(now time.Time, priorAbandons map[string][]time.Time) []Abandonment { diff --git a/server/domain/reconnect_test.go b/server/domain/reconnect_test.go index f6a3a716..77f5a2e0 100644 --- a/server/domain/reconnect_test.go +++ b/server/domain/reconnect_test.go @@ -140,6 +140,21 @@ func TestRankedAbandonCooldownUsesRollingSevenDayLadder(t *testing.T) { } } +func TestPlanRankedAbandonmentsFencesGraceAndClockAnomalies(t *testing.T) { + now := time.Unix(1000, 0).UTC() + planned, err := PlanRankedAbandonments(now, []ReconnectParticipant{ + {PlayerID: "within", DisconnectedAt: now.Add(-RankedReconnectGrace)}, + {PlayerID: "future", DisconnectedAt: now.Add(time.Second)}, + {PlayerID: "expired", DisconnectedAt: now.Add(-RankedReconnectGrace - time.Nanosecond)}, + }, map[string][]time.Time{"expired": {now.Add(-time.Hour)}}) + if err != nil || len(planned) != 1 || planned[0].PlayerID != "expired" || planned[0].Cooldown != 15*time.Minute || !planned[0].AbandonedAt.Equal(now) { + t.Fatalf("planned=%+v err=%v", planned, err) + } + if _, err := PlanRankedAbandonments(now, []ReconnectParticipant{{PlayerID: "duplicate", DisconnectedAt: now}, {PlayerID: "duplicate", DisconnectedAt: now}}, nil); err == nil { + t.Fatal("duplicate reconnect participant accepted") + } +} + func TestSignedJoinAuthorisationBindsEveryClaimBeforeReclaim(t *testing.T) { now := time.Unix(1000, 0).UTC() r, err := NewRankedConnections("match-1", "server-1", "v1", testRoster(now)) diff --git a/server/store/initial_connect_maintenance.go b/server/store/initial_connect_maintenance.go index ff55aa35..a94df98e 100644 --- a/server/store/initial_connect_maintenance.go +++ b/server/store/initial_connect_maintenance.go @@ -19,7 +19,7 @@ LIMIT $1` const initialConnectHistorySQL = `SELECT starts_at FROM penalties -WHERE player_id = $1 AND kind = 'INITIAL_CONNECT_NO_SHOW' +WHERE player_id = $1 AND kind IN ('INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED') ORDER BY starts_at` // ReconcileInitialConnect evaluates a bounded set of matches and applies only diff --git a/server/store/initial_connect_sql_test.go b/server/store/initial_connect_sql_test.go index 833e6de4..f938f2a5 100644 --- a/server/store/initial_connect_sql_test.go +++ b/server/store/initial_connect_sql_test.go @@ -27,6 +27,14 @@ func TestInitialConnectSQLPreservesAtomicNoShowReconciliation(t *testing.T) { } } } + for _, fragment := range []string{"INITIAL_CONNECT_NO_SHOW", "MATCH_ABANDONED"} { + if !contains(initialConnectHistorySQL, fragment) { + t.Fatalf("initial-connect abandon history missing %q", fragment) + } + if !contains(QueueCooldownSelectSQL, fragment) { + t.Fatalf("queue cooldown fence missing %q", fragment) + } + } } func TestInitialConnectPlanValidationRejectsIncompleteOrForgedPlans(t *testing.T) { diff --git a/server/store/live_abandonment_sql.go b/server/store/live_abandonment_sql.go new file mode 100644 index 00000000..21680564 --- /dev/null +++ b/server/store/live_abandonment_sql.go @@ -0,0 +1,213 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +const liveAbandonmentCandidatesSQL = `SELECT m.match_id +FROM matches m +WHERE m.playlist = 'ranked' AND m.state = 'LIVE' + AND EXISTS ( + SELECT 1 FROM match_participants mp + WHERE mp.match_id = m.match_id AND mp.participation_active + AND mp.abandoned_at IS NULL AND mp.disconnected_at IS NOT NULL + AND mp.disconnected_at < $1 + ) +ORDER BY m.match_id +LIMIT $2` + +const liveAbandonmentMatchLockSQL = `SELECT playlist, state +FROM matches WHERE match_id = $1 FOR UPDATE` + +const liveAbandonmentParticipantsSQL = `SELECT player_id, disconnected_at +FROM match_participants +WHERE match_id = $1 AND participation_active + AND abandoned_at IS NULL AND disconnected_at IS NOT NULL +ORDER BY player_id +FOR UPDATE` + +const liveAbandonmentHistorySQL = `SELECT starts_at +FROM penalties +WHERE player_id = $1 AND kind IN ('INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED') +ORDER BY starts_at` + +const liveAbandonmentParticipantSQL = `UPDATE match_participants +SET abandoned_at = $3 +WHERE match_id = $1 AND player_id = $2 AND participation_active + AND abandoned_at IS NULL AND disconnected_at IS NOT NULL +RETURNING player_id` + +const liveAbandonmentPenaltySQL = `INSERT INTO penalties + (penalty_id, player_id, match_id, playlist, kind, starts_at, ends_at) +VALUES ($1, $2, $3, 'ranked', 'MATCH_ABANDONED', $4, $5) +ON CONFLICT (penalty_id) DO NOTHING` + +const liveAbandonmentRevisionSQL = `UPDATE matches +SET revision = revision + 1 +WHERE match_id = $1 AND state = 'LIVE' +RETURNING revision` + +const liveAbandonmentOutboxSQL = `INSERT INTO outbox + (event_id, aggregate_type, aggregate_id, revision, event_type, payload) +VALUES ($1, 'match', $2, $3, 'participant_abandoned', $4)` + +// ReconcileLiveAbandonments applies a bounded, durable reconnect-grace sweep. +// It does not deactivate participants or alter LIVE tickets: an abandonment +// must remain in the authoritative result roster so rating correctly scores a +// loss if the match later completes. +func ReconcileLiveAbandonments(ctx context.Context, db *sql.DB, now time.Time, limit int) (int, error) { + if db == nil || now.IsZero() || limit < 1 || limit > 1000 { + return 0, fmt.Errorf("invalid live-abandonment maintenance arguments") + } + rows, err := db.QueryContext(ctx, liveAbandonmentCandidatesSQL, now.Add(-domain.RankedReconnectGrace), limit) + if err != nil { + return 0, err + } + defer rows.Close() + var matchIDs []string + for rows.Next() { + var matchID string + if err := rows.Scan(&matchID); err != nil { + return 0, err + } + matchIDs = append(matchIDs, matchID) + } + if err := rows.Err(); err != nil { + return 0, err + } + // Do not hold the candidate cursor while opening serializable per-match + // transactions. A deliberately small production pool (including size one) + // would otherwise wait on its own still-open read connection. + if err := rows.Close(); err != nil { + return 0, err + } + count := 0 + for _, matchID := range matchIDs { + changed, err := ApplyLiveAbandonments(ctx, db, matchID, now) + if err != nil { + return count, err + } + if changed > 0 { + count++ + } + } + return count, nil +} + +// ApplyLiveAbandonments is independently serializable so concurrent +// maintenance replicas or a result submission cannot double-penalise a +// player. It returns the number of participants newly abandoned. +func ApplyLiveAbandonments(ctx context.Context, db *sql.DB, matchID string, now time.Time) (int, error) { + if db == nil || matchID == "" || now.IsZero() { + return 0, fmt.Errorf("invalid live-abandonment transaction arguments") + } + changed := 0 + err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { + var playlist, state string + if err := tx.QueryRowContext(ctx, liveAbandonmentMatchLockSQL, matchID).Scan(&playlist, &state); err != nil { + return err + } + if playlist != string(domain.Ranked) || state != string(domain.Live) { + return nil + } + participants, err := loadLiveReconnectParticipants(ctx, tx, matchID) + if err != nil { + return err + } + history, err := loadLiveAbandonmentHistory(ctx, tx, participants) + if err != nil { + return err + } + planned, err := domain.PlanRankedAbandonments(now, participants, history) + if err != nil { + return err + } + if len(planned) == 0 { + return nil + } + for _, abandonment := range planned { + var playerID string + if err := tx.QueryRowContext(ctx, liveAbandonmentParticipantSQL, matchID, abandonment.PlayerID, abandonment.AbandonedAt).Scan(&playerID); err != nil { + if err == sql.ErrNoRows { + return fmt.Errorf("%w: reconnect participant changed", domain.ErrConflict) + } + return err + } + penaltyID := "live-abandon:" + matchID + ":" + abandonment.PlayerID + if _, err := tx.ExecContext(ctx, liveAbandonmentPenaltySQL, penaltyID, abandonment.PlayerID, matchID, abandonment.AbandonedAt, abandonment.AbandonedAt.Add(abandonment.Cooldown)); err != nil { + return err + } + } + var revision uint64 + if err := tx.QueryRowContext(ctx, liveAbandonmentRevisionSQL, matchID).Scan(&revision); err != nil { + return err + } + payload, err := json.Marshal(map[string]any{"match_id": matchID, "abandoned_player_ids": abandonmentIDs(planned)}) + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, liveAbandonmentOutboxSQL, fmt.Sprintf("live-abandon:%s:%d", matchID, revision), matchID, revision, payload); err != nil { + return err + } + changed = len(planned) + return nil + }) + return changed, err +} + +func loadLiveReconnectParticipants(ctx context.Context, tx *sql.Tx, matchID string) ([]domain.ReconnectParticipant, error) { + rows, err := tx.QueryContext(ctx, liveAbandonmentParticipantsSQL, matchID) + if err != nil { + return nil, err + } + defer rows.Close() + participants := make([]domain.ReconnectParticipant, 0) + for rows.Next() { + var participant domain.ReconnectParticipant + if err := rows.Scan(&participant.PlayerID, &participant.DisconnectedAt); err != nil { + return nil, err + } + participants = append(participants, participant) + } + return participants, rows.Err() +} + +func loadLiveAbandonmentHistory(ctx context.Context, tx *sql.Tx, participants []domain.ReconnectParticipant) (map[string][]time.Time, error) { + history := make(map[string][]time.Time, len(participants)) + for _, participant := range participants { + rows, err := tx.QueryContext(ctx, liveAbandonmentHistorySQL, participant.PlayerID) + if err != nil { + return nil, err + } + for rows.Next() { + var started time.Time + if err := rows.Scan(&started); err != nil { + rows.Close() + return nil, err + } + history[participant.PlayerID] = append(history[participant.PlayerID], started) + } + if err := rows.Err(); err != nil { + rows.Close() + return nil, err + } + if err := rows.Close(); err != nil { + return nil, err + } + } + return history, nil +} + +func abandonmentIDs(abandonments []domain.Abandonment) []string { + ids := make([]string, len(abandonments)) + for i := range abandonments { + ids[i] = abandonments[i].PlayerID + } + return ids +} diff --git a/server/store/live_abandonment_sql_test.go b/server/store/live_abandonment_sql_test.go new file mode 100644 index 00000000..b40f3f80 --- /dev/null +++ b/server/store/live_abandonment_sql_test.go @@ -0,0 +1,27 @@ +package store + +import ( + "strings" + "testing" +) + +func TestLiveAbandonmentSQLPreservesResultRosterAndReconnectFences(t *testing.T) { + for query, fragments := range map[string][]string{ + liveAbandonmentCandidatesSQL: {"playlist = 'ranked'", "state = 'LIVE'", "abandoned_at IS NULL", "disconnected_at < $1", "LIMIT $2"}, + liveAbandonmentMatchLockSQL: {"FOR UPDATE", "match_id = $1"}, + liveAbandonmentParticipantsSQL: {"participation_active", "abandoned_at IS NULL", "disconnected_at IS NOT NULL", "FOR UPDATE"}, + liveAbandonmentParticipantSQL: {"SET abandoned_at", "participation_active", "abandoned_at IS NULL", "RETURNING"}, + liveAbandonmentPenaltySQL: {"MATCH_ABANDONED", "ON CONFLICT"}, + liveAbandonmentRevisionSQL: {"state = 'LIVE'", "revision = revision + 1"}, + liveAbandonmentOutboxSQL: {"participant_abandoned", "revision"}, + } { + for _, fragment := range fragments { + if !strings.Contains(query, fragment) { + t.Fatalf("query missing %q: %s", fragment, query) + } + } + } + if strings.Contains(liveAbandonmentParticipantSQL, "participation_active = FALSE") || strings.Contains(liveAbandonmentParticipantSQL, "queue_tickets") { + t.Fatal("live abandonment must retain participant and ticket for the result transaction") + } +} diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index f9f05083..b07935e5 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -649,6 +649,57 @@ func TestPostgreSQLConnectionReceiptsStartCompleteRelaxedCasualRoster(t *testing } } +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) VALUES ('live-abandon-match', 'ranked', 'LIVE', 'EU', 1, 'live-abandon-server')`); 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 = 'participant_abandoned'`).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) diff --git a/server/store/queue_sql.go b/server/store/queue_sql.go index 08bf2d45..4d2c2213 100644 --- a/server/store/queue_sql.go +++ b/server/store/queue_sql.go @@ -52,7 +52,7 @@ FOR UPDATE` QueueCooldownSelectSQL = `SELECT ends_at FROM penalties WHERE player_id = $1 AND playlist = $2 - AND kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT') + AND kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT', 'INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED') AND ends_at > $3 ORDER BY ends_at DESC LIMIT 1`