From a15368ed29002511730d3d195a040872a1c6a25f Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Thu, 3 Sep 2026 00:15:41 +0100 Subject: [PATCH] fix(multiplayer): fence client queue cancellation states --- multiplayer-next.md | 2 +- server/domain/queue.go | 3 ++ server/domain/queue_test.go | 18 ++++++++++++ server/store/postgres_integration_test.go | 20 +++++++++++-- server/store/queue_sql.go | 35 +++++++++++++++++++++-- server/store/queue_sql_test.go | 3 +- 6 files changed, 75 insertions(+), 6 deletions(-) diff --git a/multiplayer-next.md b/multiplayer-next.md index fc84c55f..81525e83 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 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.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.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 | diff --git a/server/domain/queue.go b/server/domain/queue.go index 24d8c575..a45a59f5 100644 --- a/server/domain/queue.go +++ b/server/domain/queue.go @@ -131,6 +131,9 @@ func (q *Queue) Cancel(playerID, ticketID, idempotencyKey string, expectedRevisi if ticket.Revision != expectedRevision { return QueueTicket{}, ErrStaleRevision } + if ticket.State != Queued && ticket.State != Proposed { + return QueueTicket{}, fmt.Errorf("%w: cancel in %s", ErrConflict, ticket.State) + } if idempotencyKey == "" { return QueueTicket{}, fmt.Errorf("%w: empty cancel key", ErrConflict) } diff --git a/server/domain/queue_test.go b/server/domain/queue_test.go index aab86d40..06436c0a 100644 --- a/server/domain/queue_test.go +++ b/server/domain/queue_test.go @@ -50,6 +50,24 @@ func TestQueueHeartbeatExtendsExpiryExactlyAndRejectsStaleReplay(t *testing.T) { } } +func TestQueueCancelCannotOverrideMatchOwnedLifecycle(t *testing.T) { + q := NewQueue() + now := time.Unix(1000, 0) + candidate := Candidate{TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: now} + if _, err := q.Create("player-a", "ticket-a", "create-key-123456", candidate, now); err != nil { + t.Fatal(err) + } + ticket := q.tickets["ticket-a"] + ticket.State = Live + q.tickets["ticket-a"] = ticket + if _, err := q.Cancel("player-a", "ticket-a", "cancel-key-123456", 0, now.Add(time.Second)); !errors.Is(err, ErrConflict) { + t.Fatalf("live ticket cancellation error = %v, want conflict", err) + } + if got := q.tickets["ticket-a"].State; got != Live { + t.Fatalf("live ticket state = %s after cancellation attempt", got) + } +} + func TestQueueExpiryReleasesOwnershipAndDoesNotReturnExpiredCandidates(t *testing.T) { q := NewQueue() now := time.Unix(1000, 0) diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index c5b2137b..6cc6f3f7 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -388,8 +388,8 @@ func TestPostgreSQLQueueHeartbeatAndCancelAreRevisionFenced(t *testing.T) { if heartbeat.Revision != 1 || !heartbeat.ExpiresAt.Equal(now.Add(35*time.Second)) { t.Fatalf("unexpected heartbeat result: %+v", heartbeat) } - if _, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000002", 0, now.Add(6*time.Second)); err == nil { - t.Fatal("stale heartbeat revision was accepted") + if _, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000002", 0, now.Add(6*time.Second)); !errors.Is(err, domain.ErrStaleRevision) { + t.Fatalf("stale heartbeat error = %v, want ErrStaleRevision", err) } cancelled, err := CancelQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000003", 1, now.Add(7*time.Second)) if err != nil { @@ -398,6 +398,22 @@ func TestPostgreSQLQueueHeartbeatAndCancelAreRevisionFenced(t *testing.T) { if cancelled.State != domain.Cancelled || cancelled.Revision != 2 { t.Fatalf("unexpected cancellation result: %+v", cancelled) } + if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('live-cancel-player', 'live-cancel-steam')`); 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 ('live-cancel-ticket', 'live-cancel-player', 'ranked', 'LIVE', 'integration-build', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + if _, err := CancelQueueTicket(ctx, db, "live-cancel-player", "live-cancel-ticket", "live-cancel-op-0001", 0, now.Add(8*time.Second)); !errors.Is(err, domain.ErrConflict) { + t.Fatalf("live cancellation error = %v, want ErrConflict", err) + } + var liveState string + if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'live-cancel-ticket'`).Scan(&liveState); err != nil { + t.Fatal(err) + } + if liveState != "LIVE" { + t.Fatalf("live ticket state = %s after cancellation attempt", liveState) + } } // TestPostgreSQLConcurrentQueueHeartbeatIsRevisionFencedUnderRealRace is the diff --git a/server/store/queue_sql.go b/server/store/queue_sql.go index 57e61101..08bf2d45 100644 --- a/server/store/queue_sql.go +++ b/server/store/queue_sql.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "database/sql" "encoding/json" + "errors" "fmt" "time" @@ -42,8 +43,12 @@ RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, QueueTicketCancelSQL = `UPDATE queue_tickets SET state = 'CANCELLED', revision = revision + 1, expires_at = $4 WHERE ticket_id = $1 AND player_id = $2 AND revision = $3 - AND state NOT IN ('COMPLETED', 'CANCELLED', 'EXPIRED', 'FAILED') + AND state IN ('QUEUED', 'PROPOSED') RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at, revision, predicted_rtt` + QueueMutationFailureSQL = `SELECT player_id, state, revision, expires_at +FROM queue_tickets +WHERE ticket_id = $1 +FOR UPDATE` QueueCooldownSelectSQL = `SELECT ends_at FROM penalties WHERE player_id = $1 AND playlist = $2 @@ -294,7 +299,10 @@ func mutateQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idem var record queueTicketRecord var predictedRTT []byte if err := tx.QueryRowContext(ctx, mutationSQL, ticketID, playerID, expectedRevision, now).Scan(&record.TicketID, &record.PlayerID, &record.Playlist, &record.State, &record.ClientBuild, &record.ProtocolVersion, &record.EnqueuedAt, &record.ExpiresAt, &record.Revision, &predictedRTT); err != nil { - return fmt.Errorf("queue mutation rejected: %w", err) + if !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("queue mutation rejected: %w", err) + } + return classifyQueueMutationFailure(ctx, tx, playerID, ticketID, expectedRevision, now) } if err := json.Unmarshal(predictedRTT, &record.PredictedRTT); err != nil { return fmt.Errorf("decode queue RTT: %w", err) @@ -315,6 +323,29 @@ func mutateQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idem return ticket, err } +func classifyQueueMutationFailure(ctx context.Context, tx *sql.Tx, playerID, ticketID string, expectedRevision uint64, now time.Time) error { + var owner, state string + var revision uint64 + var expiresAt time.Time + err := tx.QueryRowContext(ctx, QueueMutationFailureSQL, ticketID).Scan(&owner, &state, &revision, &expiresAt) + if errors.Is(err, sql.ErrNoRows) { + return domain.ErrTicketNotFound + } + if err != nil { + return err + } + if owner != playerID { + return domain.ErrNotTicketOwner + } + if (state == string(domain.Queued) || state == string(domain.Proposed)) && !now.Before(expiresAt) { + return domain.ErrTicketExpired + } + if revision != expectedRevision { + return domain.ErrStaleRevision + } + return fmt.Errorf("%w: %s in %s", domain.ErrConflict, "queue mutation", state) +} + func queueTicketRecordFromDomain(ticket domain.QueueTicket) queueTicketRecord { return queueTicketRecord{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, ProposalID: ticket.ProposalID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), ClientBuild: ticket.Candidate.ClientBuild, ProtocolVersion: ticket.Candidate.ProtocolVersion, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt, Revision: ticket.Revision, PredictedRTT: ticket.Candidate.PredictedRTT} } diff --git a/server/store/queue_sql_test.go b/server/store/queue_sql_test.go index 28bcdcdc..6605ddb2 100644 --- a/server/store/queue_sql_test.go +++ b/server/store/queue_sql_test.go @@ -13,7 +13,8 @@ func TestQueueSQLUsesDurableIdempotencyAndOwnerScopedRecovery(t *testing.T) { QueueTicketSelectSQL: {"q.ticket_id = $1", "q.player_id = $2", "proposal_participants", "p.state = 'OPEN'", "match_participants", "participation_active"}, QueueTicketInsertSQL: {"player_id", "playlist", "client_build", "protocol_version"}, QueueTicketHeartbeatSQL: {"player_id = $2", "revision = $3", "expires_at > $4", "RETURNING"}, - QueueTicketCancelSQL: {"player_id = $2", "revision = $3", "state NOT IN", "RETURNING"}, + QueueTicketCancelSQL: {"player_id = $2", "revision = $3", "state IN ('QUEUED', 'PROPOSED')", "RETURNING"}, + QueueMutationFailureSQL: {"ticket_id = $1", "state", "revision", "expires_at", "FOR UPDATE"}, QueueCandidateProjectionSQL: {"playlist = $1", "predicted_rtt", "expires_at > $2", "LIMIT $3"}, RankedParticipantSQL: {"steam_id", "player_id = ANY($1)", "ORDER BY player_id"}, ProposalInsertSQL: {"match_region", "match_protocol", "NULLIF($4, '')"},