mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
fix(multiplayer): terminate proposal offenders atomically
This commit is contained in:
+8
-7
@@ -1205,7 +1205,7 @@ production fallback.
|
||||
| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue domain enforces one active ticket per verified player under concurrent mutation, 10 s heartbeat/30 s expiry, retry-safe create/heartbeat/cancel, owner-only recovery reads and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary and authenticated HTTP queue adapter with playlist/build/protocol compatibility metadata | `server/domain/queue.go`, `server/store/candidates.go`, `server/store/queue_sql.go`, `server/store/redis_candidates.go` and `server/api/service.go` cover ownership/expiry/idempotency, concurrent create fencing, candidate/player ownership binding, owner/revision-scoped SQL heartbeat/cancel/recovery, injectable PostgreSQL queue backend selected by the HTTP service, authoritative queue-to-cache rebuild, expired recovery as a terminal error, owner-scoped SQL recovery with authoritative expiry handling, server-owned candidate resolution, strict compatibility metadata, TTL-bound Redis upsert/remove/snapshot, API create/heartbeat/cancel projection hooks, optional control-plane Redis configuration and atomic durable-source repair on partial/malformed cache state; opt-in real PostgreSQL execution now covers create/replay/active-player fencing, owner recovery, revision-fenced heartbeat/cancel and expiry, while miniredis covers Redis behavior and repair-source failure — this suite had never actually been run clean against a live database before: doing so once found `CreateQueueTicket` passing one extra unbound argument to its insert, which failed every real ticket creation with a param-count mismatch (fixed, re-verified against a real `postgres:17-alpine` container). A separate opt-in real-Redis suite (`server/store/redis_integration_test.go`, `scripts/run_redis_integration.sh`, `COSMIC_CLASH_REDIS_ADDR`-gated) now covers upsert/snapshot/remove, a real TTL actually waited out, and the "lost keyspace" repair path against a genuine `FLUSHALL` — including that the repair persists back to Redis, not just returned an in-memory answer. `TestPostgreSQLConcurrentQueueHeartbeatIsRevisionFencedUnderRealRace` races 5 concurrent same-revision heartbeats against real PostgreSQL: exactly one wins, the durable revision lands at exactly 1; 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 decline/timeout cooldowns and ranked escalation; authenticated API exposes revisioned accept/decline mutations; formed matches now pass through a playlist-aware proposal boundary. **Fixed a real severe bug**: declining a proposal never requeued anyone's ticket — every participant, decliner included, was left stranded at `PROPOSED` (invisible to the matcher, still blocking a fresh `queue_create`, renewable forever by an ordinary heartbeat) with no path back into matchmaking. `ProposalDeclineRequeueSQL` now requeues every participant to `QUEUED` with a fresh expiry on decline; the not-yet-built decline cooldown mentioned here can later exempt the decliner specifically, but leaving anyone stuck today wasn't that cooldown, it was just broken. **The same bug's timeout sibling is fixed too**: a proposal that simply expires (no unanimous response inside the window) hit the identical gap in `ProposalExpireSQL`/`ProposalParticipantExpireSQL`, reached from both `GetProposal` (a client recovering after missing the expiry event) and `RespondToProposal` (a response arriving after the window); `ProposalExpireRequeueSQL` mirrors the decline fix, guarded on `state = 'EXPIRED'` so it's safe to call unconditionally. **Closed the remaining responsiveness gap too**: cancelling a queue ticket directly while it's part of an OPEN proposal used to leave the other participant waiting out the full window instead of being told immediately; `CascadeCancelToOpenProposal` now declines and requeues the proposal in the same transaction as the cancel | `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; real PostgreSQL integration tests confirm the decline, timeout, and cancel-cascade paths all requeue every participant (decliner/uninvolved participant/cancelling player's partner alike) to `QUEUED` with a refreshed expiry, visible again to `ListQueuedCandidates` (the matcher's own read), and that a cancelling player's own ticket correctly stays `CANCELLED` rather than being swept back up; clean across 5 runs each; queue precedence, allocation integration and the decline cooldown itself 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 |
|
||||
| 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 persists its matcher-selected region/protocol/team/slot plan alongside proposal/participant claims, 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; final unanimous API acceptance invokes replay-safe promotion of the exact stored `ALLOCATING` match/team/slot topology and 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`, `server/api/service.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, final-response promotion retry, 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 — actually running this suite live (it had not been before) found `ProposalParticipantExpireSQL` had no expiry-time condition at all, so every call timed out every pending participant on the spot; the very first accept on any proposal then failed with a false conflict. Fixed with the same `expires_at <=` gate `ProposalExpireSQL` already used, re-verified live. A real concurrent-goroutine test now covers the two-matcher race this was missing: two proposals sharing one contested ticket, racing two real Postgres connections under `-race`, exactly-one-wins/loser-fully-rolls-back including the loser's own uncontested ticket, stable across 8 runs; allocation runtime 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 |
|
||||
@@ -1489,12 +1489,13 @@ becoming an unbounded account-level resource cost. Over-limit attempts fail
|
||||
before upgrade with `429 websocket_connection_limited`, rather than becoming
|
||||
ambiguous post-upgrade disconnects.
|
||||
|
||||
Proposal explicit-decline cooldowns are now durable: the declining player is
|
||||
requeued for recovery, but a subsequent queue create is rejected until the
|
||||
playlist-specific cooldown computed by `domain.CooldownUntil` expires. The
|
||||
operation is idempotent and does not affect the other participants' requeue;
|
||||
timeout-derived cooldown recording now uses the same durable penalty path,
|
||||
with deterministic per-proposal/player IDs for replay safety.
|
||||
Proposal decline and timeout cooldowns are now durable and matchable-state
|
||||
safe: an offender's existing ticket becomes terminal (`CANCELLED` for decline,
|
||||
`EXPIRED` for timeout), while innocent or already-accepted participants retain
|
||||
their original queue precedence. Late response recovery commits before the API
|
||||
returns `ErrProposalClosed`; deterministic penalty IDs preserve replay safety,
|
||||
future/corrupt cooldown events are ignored, and reopening an old declined
|
||||
proposal cannot create false timeout penalties for its innocent participants.
|
||||
|
||||
### Current local completion index (2026-09-02)
|
||||
|
||||
|
||||
@@ -182,7 +182,8 @@ func CooldownUntil(events []CooldownEvent, playlist Playlist, now time.Time) tim
|
||||
cutoff := now.Add(-window)
|
||||
filtered := make([]CooldownEvent, 0, len(events))
|
||||
for _, event := range events {
|
||||
if event.Playlist == playlist && !event.At.Before(cutoff) {
|
||||
validKind := event.Kind == DeclinedResponse || event.Kind == TimedOutResponse
|
||||
if event.Playlist == playlist && validKind && !event.At.Before(cutoff) && !event.At.After(now) {
|
||||
filtered = append(filtered, event)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,3 +94,17 @@ func TestRankedProposalAndCooldownEscalation(t *testing.T) {
|
||||
t.Fatalf("ranked escalation cooldown = %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCooldownIgnoresFutureForeignAndInvalidEvents(t *testing.T) {
|
||||
now := time.Unix(10_000, 0)
|
||||
events := []CooldownEvent{
|
||||
{At: now.Add(-time.Minute), Playlist: Casual, Kind: DeclinedResponse},
|
||||
{At: now.Add(time.Hour), Playlist: Ranked, Kind: TimedOutResponse},
|
||||
{At: now, Playlist: Casual, Kind: TimedOutResponse},
|
||||
{At: now, Playlist: Ranked, Kind: AcceptedResponse},
|
||||
{At: now.Add(-31 * time.Minute), Playlist: Ranked, Kind: DeclinedResponse},
|
||||
}
|
||||
if got := CooldownUntil(events, Ranked, now); !got.IsZero() {
|
||||
t.Fatalf("untrusted cooldown events produced %v, want zero", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -612,12 +612,10 @@ func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestPostgreSQLProposalDeclineRequeuesEveryParticipant protects the durable
|
||||
// decline boundary: every ticket returns to QUEUED, while the decliner's
|
||||
// separate penalty prevents an immediate replacement queue ticket. Without
|
||||
// the requeue, tickets are invisible to the matcher and remain trapped in
|
||||
// PROPOSED despite the proposal having closed.
|
||||
func TestPostgreSQLProposalDeclineRequeuesEveryParticipant(t *testing.T) {
|
||||
// TestPostgreSQLProposalDeclineCancelsOffenderAndRequeuesInnocent protects the
|
||||
// durable decline boundary: the offender's ticket becomes terminal while
|
||||
// every innocent ticket keeps its original queue precedence.
|
||||
func TestPostgreSQLProposalDeclineCancelsOffenderAndRequeuesInnocent(t *testing.T) {
|
||||
db := openIntegrationPostgres(t)
|
||||
applyIntegrationMigrations(t, db)
|
||||
|
||||
@@ -659,8 +657,8 @@ func TestPostgreSQLProposalDeclineRequeuesEveryParticipant(t *testing.T) {
|
||||
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'decline-ticket-1'`).Scan(&stateB, &expiresB); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stateA != "QUEUED" {
|
||||
t.Fatalf("decliner's own ticket state = %s, want QUEUED while cooldown is recorded separately", stateA)
|
||||
if stateA != "CANCELLED" {
|
||||
t.Fatalf("decliner's own ticket state = %s, want CANCELLED", stateA)
|
||||
}
|
||||
if stateB != "QUEUED" {
|
||||
t.Fatalf("uninvolved participant's ticket state = %s, want QUEUED -- they must not be stranded by someone else's decline", stateB)
|
||||
@@ -669,8 +667,8 @@ func TestPostgreSQLProposalDeclineRequeuesEveryParticipant(t *testing.T) {
|
||||
t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", expiresB, now)
|
||||
}
|
||||
|
||||
// The real, end-to-end regression: both players can be proposed a NEW
|
||||
// match instead of ListQueuedCandidates silently never seeing them again.
|
||||
// Only the innocent player can be selected again. A durable cooldown also
|
||||
// rejects a new ticket from the decliner until the policy window ends.
|
||||
candidates, err := ListQueuedCandidates(ctx, db, domain.Casual, now, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("list queued candidates: %v", err)
|
||||
@@ -679,17 +677,34 @@ func TestPostgreSQLProposalDeclineRequeuesEveryParticipant(t *testing.T) {
|
||||
for _, candidate := range candidates {
|
||||
found[candidate.PlayerID] = true
|
||||
}
|
||||
if !found["decline-player-a"] || !found["decline-player-b"] {
|
||||
t.Fatalf("requeued players are not visible to the matcher: %+v", candidates)
|
||||
if found["decline-player-a"] || !found["decline-player-b"] {
|
||||
t.Fatalf("matcher did not isolate offender from innocent: %+v", candidates)
|
||||
}
|
||||
var cooldownEnd time.Time
|
||||
if err := db.QueryRow(`SELECT ends_at FROM penalties WHERE player_id = 'decline-player-a' AND kind = 'PROPOSAL_DECLINED'`).Scan(&cooldownEnd); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := now.Add(30 * time.Second); !cooldownEnd.Equal(want) {
|
||||
t.Fatalf("decline cooldown end = %v, want %v", cooldownEnd, want)
|
||||
}
|
||||
// Recovering the closed proposal after its old deadline must not convert
|
||||
// the innocent participant's PENDING response into a timeout penalty.
|
||||
if _, err := GetProposal(ctx, db, "decline-player-b", proposal.ProposalID, now.Add(domain.ProposalWindow+time.Second)); err != nil {
|
||||
t.Fatalf("recover declined proposal: %v", err)
|
||||
}
|
||||
var innocentTimeouts int
|
||||
if err := db.QueryRow(`SELECT count(*) FROM penalties WHERE player_id = 'decline-player-b' AND kind = 'PROPOSAL_TIMEOUT'`).Scan(&innocentTimeouts); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if innocentTimeouts != 0 {
|
||||
t.Fatalf("innocent participant received %d timeout penalties after decline", innocentTimeouts)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPostgreSQLProposalTimeoutRequeuesEveryParticipant protects the timeout
|
||||
// sibling of the decline path: expiry must requeue every ticket and record a
|
||||
// timeout cooldown for each participant who failed to respond. It uses
|
||||
// GetProposal, the recovery/read path, to exercise a client returning after
|
||||
// it missed the expiry event.
|
||||
func TestPostgreSQLProposalTimeoutRequeuesEveryParticipant(t *testing.T) {
|
||||
// TestPostgreSQLProposalTimeoutExpiresOffenderAndRequeuesAccepted protects the
|
||||
// timeout sibling: accepted participants retain precedence, while no-shows
|
||||
// receive a terminal ticket and cooldown.
|
||||
func TestPostgreSQLProposalTimeoutExpiresOffenderAndRequeuesAccepted(t *testing.T) {
|
||||
db := openIntegrationPostgres(t)
|
||||
applyIntegrationMigrations(t, db)
|
||||
|
||||
@@ -712,9 +727,11 @@ func TestPostgreSQLProposalTimeoutRequeuesEveryParticipant(t *testing.T) {
|
||||
if err := CreateProposal(ctx, db, proposal, map[string]string{"timeout-player-a": "timeout-ticket-0", "timeout-player-b": "timeout-ticket-1"}, now); err != nil {
|
||||
t.Fatalf("create proposal: %v", err)
|
||||
}
|
||||
if _, err := RespondToProposal(ctx, db, "timeout-player-a", proposal.ProposalID, "timeout-accept-a-0001", true, 0, now.Add(time.Second)); err != nil {
|
||||
t.Fatalf("accept proposal: %v", err)
|
||||
}
|
||||
|
||||
// Nobody ever responds; recover the proposal well after its 10s window,
|
||||
// exactly as a client reconnecting after missing the expiry event would.
|
||||
// player-b never responds; recover well after the response window.
|
||||
afterExpiry := now.Add(domain.ProposalWindow + time.Second)
|
||||
recovered, err := GetProposal(ctx, db, "timeout-player-a", proposal.ProposalID, afterExpiry)
|
||||
if err != nil {
|
||||
@@ -725,18 +742,18 @@ func TestPostgreSQLProposalTimeoutRequeuesEveryParticipant(t *testing.T) {
|
||||
}
|
||||
|
||||
var stateA, stateB string
|
||||
var expiresB time.Time
|
||||
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'timeout-ticket-0'`).Scan(&stateA); err != nil {
|
||||
var expiresA time.Time
|
||||
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'timeout-ticket-0'`).Scan(&stateA, &expiresA); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'timeout-ticket-1'`).Scan(&stateB, &expiresB); err != nil {
|
||||
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'timeout-ticket-1'`).Scan(&stateB); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stateA != "QUEUED" || stateB != "QUEUED" {
|
||||
t.Fatalf("timed-out participants left stranded: a=%s b=%s", stateA, stateB)
|
||||
if stateA != "QUEUED" || stateB != "EXPIRED" {
|
||||
t.Fatalf("timeout did not split accepted and offender tickets: a=%s b=%s", stateA, stateB)
|
||||
}
|
||||
if !expiresB.After(afterExpiry) {
|
||||
t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", expiresB, afterExpiry)
|
||||
if !expiresA.After(afterExpiry) {
|
||||
t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", expiresA, afterExpiry)
|
||||
}
|
||||
candidates, err := ListQueuedCandidates(ctx, db, domain.Casual, afterExpiry, 10)
|
||||
if err != nil {
|
||||
@@ -746,8 +763,72 @@ func TestPostgreSQLProposalTimeoutRequeuesEveryParticipant(t *testing.T) {
|
||||
for _, candidate := range candidates {
|
||||
found[candidate.PlayerID] = true
|
||||
}
|
||||
if !found["timeout-player-a"] || !found["timeout-player-b"] {
|
||||
t.Fatalf("requeued players are not visible to the matcher: %+v", candidates)
|
||||
if !found["timeout-player-a"] || found["timeout-player-b"] {
|
||||
t.Fatalf("matcher did not isolate timeout offender: %+v", candidates)
|
||||
}
|
||||
var cooldownEnd time.Time
|
||||
if err := db.QueryRow(`SELECT ends_at FROM penalties WHERE player_id = 'timeout-player-b' AND kind = 'PROPOSAL_TIMEOUT'`).Scan(&cooldownEnd); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := afterExpiry.Add(60 * time.Second); !cooldownEnd.Equal(want) {
|
||||
t.Fatalf("timeout cooldown end = %v, want %v", cooldownEnd, want)
|
||||
}
|
||||
}
|
||||
|
||||
// A late response must report a closed proposal only after committing the
|
||||
// expiry recovery. Returning that domain error from inside RunSerializable
|
||||
// used to roll every recovery write back.
|
||||
func TestPostgreSQLLateProposalResponseCommitsExpiryRecovery(t *testing.T) {
|
||||
db := openIntegrationPostgres(t)
|
||||
applyIntegrationMigrations(t, db)
|
||||
|
||||
now := time.Now().UTC().Truncate(time.Microsecond)
|
||||
ctx := context.Background()
|
||||
for _, player := range []string{"late-player-a", "late-player-b"} {
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for i, player := range []string{"late-player-a", "late-player-b"} {
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("late-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
proposal, err := domain.NewProposal("late-proposal", domain.Casual, []string{"late-player-a", "late-player-b"}, now)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := CreateProposal(ctx, db, proposal, map[string]string{"late-player-a": "late-ticket-0", "late-player-b": "late-ticket-1"}, now); err != nil {
|
||||
t.Fatalf("create proposal: %v", err)
|
||||
}
|
||||
|
||||
late := now.Add(domain.ProposalWindow + time.Second)
|
||||
_, err = RespondToProposal(ctx, db, "late-player-a", proposal.ProposalID, "late-response-a-0001", true, 0, late)
|
||||
if !errors.Is(err, domain.ErrProposalClosed) {
|
||||
t.Fatalf("late response error = %v, want ErrProposalClosed", err)
|
||||
}
|
||||
var proposalState, ticketA, ticketB string
|
||||
if err := db.QueryRow(`SELECT state FROM proposals WHERE proposal_id = 'late-proposal'`).Scan(&proposalState); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'late-ticket-0'`).Scan(&ticketA); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'late-ticket-1'`).Scan(&ticketB); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if proposalState != "EXPIRED" || ticketA != "EXPIRED" || ticketB != "EXPIRED" {
|
||||
t.Fatalf("late recovery was not committed: proposal=%s tickets=%s,%s", proposalState, ticketA, ticketB)
|
||||
}
|
||||
var penalties, idempotencyRows int
|
||||
if err := db.QueryRow(`SELECT count(*) FROM penalties WHERE kind = 'PROPOSAL_TIMEOUT' AND player_id IN ('late-player-a', 'late-player-b')`).Scan(&penalties); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.QueryRow(`SELECT count(*) FROM idempotency_keys WHERE scope = $1 AND idempotency_key = 'late-response-a-0001'`, ProposalResponseIdempotencyScope).Scan(&idempotencyRows); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if penalties != 2 || idempotencyRows != 0 {
|
||||
t.Fatalf("late recovery side effects: penalties=%d idempotency_rows=%d", penalties, idempotencyRows)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -20,22 +20,24 @@ WHERE proposal_id = $1 AND state = 'OPEN' AND expires_at <= $2`
|
||||
const ProposalParticipantExpireSQL = `UPDATE proposal_participants
|
||||
SET response = 'TIMED_OUT', responded_at = $2
|
||||
WHERE proposal_id = $1 AND response = 'PENDING'
|
||||
AND EXISTS (SELECT 1 FROM proposals WHERE proposals.proposal_id = proposal_participants.proposal_id AND proposals.expires_at <= $2)`
|
||||
AND EXISTS (SELECT 1 FROM proposals WHERE proposals.proposal_id = proposal_participants.proposal_id
|
||||
AND proposals.state = 'EXPIRED' AND proposals.expires_at <= $2)`
|
||||
|
||||
// ProposalExpireRequeueSQL is the timeout sibling of
|
||||
// ProposalDeclineRequeueSQL: a proposal that simply times out (no unanimous
|
||||
// response inside the 10s window) leaves any participant still holding a
|
||||
// PROPOSED ticket exactly as stranded as an explicit decline does, and for
|
||||
// the identical reason -- nothing else ever moves a PROPOSED ticket back to
|
||||
// QUEUED. The `state = 'EXPIRED'` guard makes this safe to call
|
||||
// unconditionally right after ProposalExpireSQL: it's a no-op on a proposal
|
||||
// that was already OPEN and stays OPEN (nothing to requeue) or one that was
|
||||
// already EXPIRED on a prior pass (its participants' tickets, if any were
|
||||
// still PROPOSED, were already requeued then).
|
||||
// ProposalExpireRequeueSQL preserves queue precedence only for participants
|
||||
// who accepted. Participants who did not respond are offenders and their
|
||||
// tickets are terminated separately by ProposalTimeoutTicketExpireSQL.
|
||||
const ProposalExpireRequeueSQL = `UPDATE queue_tickets q
|
||||
SET state = 'QUEUED', expires_at = $2, revision = revision + 1
|
||||
FROM proposal_participants pp
|
||||
WHERE pp.proposal_id = $1 AND q.ticket_id = pp.ticket_id AND q.player_id = pp.player_id AND q.state = 'PROPOSED'
|
||||
AND pp.response = 'ACCEPTED'
|
||||
AND EXISTS (SELECT 1 FROM proposals WHERE proposals.proposal_id = $1 AND proposals.state = 'EXPIRED')`
|
||||
|
||||
const ProposalTimeoutTicketExpireSQL = `UPDATE queue_tickets q
|
||||
SET state = 'EXPIRED', revision = revision + 1
|
||||
FROM proposal_participants pp
|
||||
WHERE pp.proposal_id = $1 AND q.ticket_id = pp.ticket_id AND q.player_id = pp.player_id AND q.state = 'PROPOSED'
|
||||
AND pp.response = 'TIMED_OUT'
|
||||
AND EXISTS (SELECT 1 FROM proposals WHERE proposals.proposal_id = $1 AND proposals.state = 'EXPIRED')`
|
||||
|
||||
const OpenProposalForCancelledTicketSQL = `SELECT pp.proposal_id
|
||||
@@ -47,8 +49,8 @@ WHERE pp.ticket_id = $1 AND pp.player_id = $2 AND p.state = 'OPEN'`
|
||||
// immediately when one of its participants cancels their own queue ticket
|
||||
// directly, rather than leaving every other participant to wait out the
|
||||
// full response window for something the system already knows can't happen
|
||||
// -- ProposalExpireRequeueSQL would eventually rescue them anyway, but not
|
||||
// for up to ProposalWindow's full duration for no reason. Must run inside
|
||||
// -- expiry recovery would eventually release them anyway, but not for up to
|
||||
// ProposalWindow's full duration for no reason. Must run inside
|
||||
// the same transaction as the ticket cancel itself; a no-op if the ticket
|
||||
// wasn't part of any currently-OPEN proposal.
|
||||
func CascadeCancelToOpenProposal(ctx context.Context, tx *sql.Tx, ticketID, playerID string, now time.Time) error {
|
||||
@@ -63,7 +65,7 @@ func CascadeCancelToOpenProposal(ctx context.Context, tx *sql.Tx, ticketID, play
|
||||
if _, err := tx.ExecContext(ctx, ProposalDeclineSQL, proposalID); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, ProposalDeclineRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow))
|
||||
_, err = tx.ExecContext(ctx, ProposalAbortRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow))
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -89,6 +91,9 @@ FROM idempotency_keys
|
||||
WHERE scope = $1 AND idempotency_key = $2
|
||||
FOR UPDATE`
|
||||
|
||||
const ProposalResponseIdempotencyDeleteSQL = `DELETE FROM idempotency_keys
|
||||
WHERE scope = $1 AND idempotency_key = $2`
|
||||
|
||||
const ProposalLockSQL = `SELECT playlist, state, revision, expires_at
|
||||
FROM proposals
|
||||
WHERE proposal_id = $1
|
||||
@@ -115,21 +120,33 @@ const ProposalDeclineSQL = `UPDATE proposals
|
||||
SET state = 'DECLINED', revision = revision + 1
|
||||
WHERE proposal_id = $1 AND state = 'OPEN'`
|
||||
|
||||
// ProposalDeclineRequeueSQL requeues every participant's ticket, including
|
||||
// the decliner's own. The durable decline penalty separately prevents that
|
||||
// player from immediately creating a replacement ticket; leaving this ticket
|
||||
// at PROPOSED would not implement a cooldown, it would strand the player and
|
||||
// hide the ticket from the matcher.
|
||||
const ProposalDeclineActorCancelSQL = `UPDATE queue_tickets q
|
||||
SET state = 'CANCELLED', revision = revision + 1
|
||||
FROM proposal_participants pp
|
||||
WHERE pp.proposal_id = $1 AND pp.player_id = $2
|
||||
AND q.ticket_id = pp.ticket_id AND q.player_id = pp.player_id AND q.state = 'PROPOSED'`
|
||||
|
||||
// ProposalDeclineRequeueSQL preserves the original queue precedence of every
|
||||
// innocent participant while terminating the declining player's ticket.
|
||||
const ProposalDeclineRequeueSQL = `UPDATE queue_tickets q
|
||||
SET state = 'QUEUED', expires_at = $2, revision = revision + 1
|
||||
FROM proposal_participants pp
|
||||
WHERE pp.proposal_id = $1 AND pp.player_id <> $3
|
||||
AND q.ticket_id = pp.ticket_id AND q.player_id = pp.player_id AND q.state = 'PROPOSED'`
|
||||
|
||||
// ProposalAbortRequeueSQL is used when a participant has already cancelled
|
||||
// their own ticket. It requeues every remaining PROPOSED ticket; the cancelled
|
||||
// ticket cannot be selected by the state predicate.
|
||||
const ProposalAbortRequeueSQL = `UPDATE queue_tickets q
|
||||
SET state = 'QUEUED', expires_at = $2, revision = revision + 1
|
||||
FROM proposal_participants pp
|
||||
WHERE pp.proposal_id = $1 AND q.ticket_id = pp.ticket_id AND q.player_id = pp.player_id AND q.state = 'PROPOSED'`
|
||||
|
||||
const ProposalCooldownEventsSQL = `SELECT kind, starts_at
|
||||
FROM penalties
|
||||
WHERE player_id = $1 AND playlist = $2
|
||||
AND kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT')
|
||||
AND starts_at >= $3
|
||||
AND starts_at >= $3 AND starts_at <= $4
|
||||
ORDER BY starts_at`
|
||||
|
||||
const ProposalCooldownInsertSQL = `INSERT INTO penalties
|
||||
@@ -143,7 +160,7 @@ WHERE proposal_id = $1 AND response = 'TIMED_OUT' AND responded_at = $2
|
||||
ORDER BY player_id`
|
||||
|
||||
func recordProposalCooldown(ctx context.Context, tx *sql.Tx, playerID string, playlist domain.Playlist, proposalID, kind string, response domain.Response, now time.Time) error {
|
||||
rows, err := tx.QueryContext(ctx, ProposalCooldownEventsSQL, playerID, string(playlist), now.Add(-30*time.Minute))
|
||||
rows, err := tx.QueryContext(ctx, ProposalCooldownEventsSQL, playerID, string(playlist), now.Add(-30*time.Minute), now)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -162,6 +179,10 @@ func recordProposalCooldown(ctx context.Context, tx *sql.Tx, playerID string, pl
|
||||
events = append(events, domain.CooldownEvent{At: at, Playlist: playlist, Kind: response})
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
events = append(events, domain.CooldownEvent{At: now, Playlist: playlist, Kind: response})
|
||||
@@ -235,6 +256,9 @@ func GetProposal(ctx context.Context, db *sql.DB, playerID, proposalID string, n
|
||||
return domain.Proposal{}, err
|
||||
}
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, ProposalTimeoutTicketExpireSQL, proposalID); err != nil {
|
||||
return domain.Proposal{}, err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, ProposalExpireRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow)); err != nil {
|
||||
return domain.Proposal{}, err
|
||||
}
|
||||
@@ -278,7 +302,9 @@ func RespondToProposal(ctx context.Context, db *sql.DB, playerID, proposalID, id
|
||||
}
|
||||
digest := sha256.Sum256([]byte(fmt.Sprintf("%s|%s|%t|%d", playerID, proposalID, accept, expectedRevision)))
|
||||
var proposal domain.Proposal
|
||||
closed := false
|
||||
err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
||||
closed = false
|
||||
result, err := tx.ExecContext(ctx, ProposalResponseIdempotencyInsertSQL, ProposalResponseIdempotencyScope, idempotencyKey, digest[:], []byte("{}"))
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -321,14 +347,20 @@ func RespondToProposal(ctx context.Context, db *sql.DB, playerID, proposalID, id
|
||||
if err := recordProposalTimeoutCooldowns(ctx, tx, proposalID, domain.Playlist(playlist), now); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, ProposalTimeoutTicketExpireSQL, proposalID); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, ProposalExpireRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow)); err != nil {
|
||||
return err
|
||||
}
|
||||
if !now.Before(expiresAt) {
|
||||
return domain.ErrProposalClosed
|
||||
}
|
||||
if state != string(domain.Open) || !now.Before(expiresAt) {
|
||||
return domain.ErrProposalClosed
|
||||
// Commit any expiry recovery above, but do not retain a placeholder
|
||||
// idempotency result for a mutation that was rejected as closed.
|
||||
if _, err := tx.ExecContext(ctx, ProposalResponseIdempotencyDeleteSQL, ProposalResponseIdempotencyScope, idempotencyKey); err != nil {
|
||||
return err
|
||||
}
|
||||
closed = true
|
||||
return nil
|
||||
}
|
||||
if revision != expectedRevision {
|
||||
return domain.ErrStaleRevision
|
||||
@@ -374,7 +406,10 @@ func RespondToProposal(ctx context.Context, db *sql.DB, playerID, proposalID, id
|
||||
if err := recordProposalDeclineCooldown(ctx, tx, playerID, domain.Playlist(playlist), proposalID, now); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, ProposalDeclineRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow))
|
||||
if _, err = tx.ExecContext(ctx, ProposalDeclineActorCancelSQL, proposalID, playerID); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, ProposalDeclineRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow), playerID)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -411,5 +446,8 @@ func RespondToProposal(ctx context.Context, db *sql.DB, playerID, proposalID, id
|
||||
_, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, ProposalResponseIdempotencyScope, idempotencyKey, stored)
|
||||
return err
|
||||
})
|
||||
if err == nil && closed {
|
||||
return domain.Proposal{}, domain.ErrProposalClosed
|
||||
}
|
||||
return proposal, err
|
||||
}
|
||||
|
||||
@@ -8,17 +8,21 @@ import (
|
||||
func TestProposalRecoverySQLBindsParticipantAndExpiresAtReadBoundary(t *testing.T) {
|
||||
for query, fragments := range map[string][]string{
|
||||
ProposalExpireSQL: {"state = 'OPEN'", "expires_at <= $2", "revision = revision + 1"},
|
||||
ProposalParticipantExpireSQL: {"response = 'PENDING'", "response = 'TIMED_OUT'", "proposals.expires_at <= $2"},
|
||||
ProposalParticipantExpireSQL: {"response = 'PENDING'", "response = 'TIMED_OUT'", "proposals.state = 'EXPIRED'", "proposals.expires_at <= $2"},
|
||||
ProposalRecoverySelectSQL: {"proposal_id = $1", "player_id = $2", "EXISTS"},
|
||||
ProposalParticipantsSelectSQL: {"proposal_id = $1", "ORDER BY player_id"},
|
||||
ProposalResponseIdempotencyInsertSQL: {"ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"},
|
||||
ProposalResponseIdempotencyDeleteSQL: {"DELETE FROM idempotency_keys", "scope = $1", "idempotency_key = $2"},
|
||||
ProposalLockSQL: {"proposal_id = $1", "FOR UPDATE"},
|
||||
ProposalParticipantLockSQL: {"proposal_id = $1", "player_id = $2", "FOR UPDATE"},
|
||||
ProposalParticipantRespondSQL: {"response = 'PENDING'", "responded_at"},
|
||||
ProposalRevisionBumpSQL: {"revision = revision + 1", "state = 'OPEN'"},
|
||||
ProposalDeclineRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "proposal_participants"},
|
||||
ProposalExpireRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "state = 'EXPIRED'"},
|
||||
ProposalCooldownEventsSQL: {"kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT')", "starts_at >= $3", "ORDER BY starts_at"},
|
||||
ProposalDeclineActorCancelSQL: {"SET state = 'CANCELLED'", "player_id = $2", "state = 'PROPOSED'"},
|
||||
ProposalDeclineRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "player_id <> $3"},
|
||||
ProposalAbortRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "proposal_participants"},
|
||||
ProposalExpireRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "response = 'ACCEPTED'", "state = 'EXPIRED'"},
|
||||
ProposalTimeoutTicketExpireSQL: {"SET state = 'EXPIRED'", "response = 'TIMED_OUT'", "state = 'PROPOSED'"},
|
||||
ProposalCooldownEventsSQL: {"kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT')", "starts_at >= $3", "starts_at <= $4", "ORDER BY starts_at"},
|
||||
ProposalCooldownInsertSQL: {"INSERT INTO penalties", "starts_at", "ends_at", "ON CONFLICT (penalty_id) DO NOTHING"},
|
||||
ProposalTimedOutParticipantsSQL: {"response = 'TIMED_OUT'", "responded_at = $2", "ORDER BY player_id"},
|
||||
OpenProposalForCancelledTicketSQL: {"proposal_participants", "state = 'OPEN'"},
|
||||
|
||||
Reference in New Issue
Block a user