From a616b7637eac5f3499a289dca000c7caa1cf8004 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Mon, 31 Aug 2026 20:36:43 +0100 Subject: [PATCH] feat: add serializable matchmaking store boundary --- multiplayer-todo.md | 2 +- server/store/serializable.go | 83 +++++++++++++++++++++++++++++++ server/store/serializable_test.go | 40 +++++++++++++++ 3 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 server/store/serializable.go create mode 100644 server/store/serializable_test.go diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 878738f8..3ca739b0 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1195,7 +1195,7 @@ the local/CI/community transport, not a silent production fallback. | 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 | `server/domain/probes.go` and adversarial fixtures cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine and five-clean release; Steam coordinator and regional probe adapters 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 | `server/domain/matcher.go`, `teams.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance and input-order independence; queue-backed candidate loading and full population fixtures 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 | `server/domain/proposal.go` and adversarial fixtures cover partial/unanimous response, expiry, replay/conflict and ranked six-player/cooldown rules; casual 6→2 composition, queue precedence and allocation integration remain | -| 8.18 `[D:8.5,8.14,8.17]` | Horizontally replicated matcher: Redis candidates, then PostgreSQL serializable proposal/participant fence, then cache cleanup/repair | Forced loss of the last acknowledged Redis write, retries, worker death and failover cannot claim a player into two proposals/matches | +| 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 | `server/store/serializable.go` and tests cover retry classification and claim-boundary invariants; live PostgreSQL adapter/row decoding, Redis candidate index/repair, worker-failure and concurrent two-matcher integration tests remain | | 8.19 `[D:8.18]` | Casual policy: proposal composition above; >=1 human/team, exhaustive rating-balanced teams, bots after 60 s, opt-in kickoff-only backfill, 30 s reconnect and defined backfill/casual penalties | Every 2–6-human shape is tested; no mid-play replacement; declined/backfill participant gets no excluded rating/cooldown; original leaver gets only documented outcome/cooldown | | 8.20 `[D:8.18]` | Ranked policy: exactly six verified solo humans, no bots/backfill, only `ArenaRegistry.random`; define initial no-show, proposal timeout and reconnect/abandon transitions | Ranked rejects parties/bots/backfill/elevated arenas; every pre-live failure returns five innocent players with original precedence and applies no rating | | 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, and deterministic opponent ordering | `server/domain/rating.go` has canonical/inactivity/weight/invalid-input fixtures; PostgreSQL snapshot locking, draws/OT/abandons, seasons and concurrent result transaction tests remain | diff --git a/server/store/serializable.go b/server/store/serializable.go new file mode 100644 index 00000000..e7e4981b --- /dev/null +++ b/server/store/serializable.go @@ -0,0 +1,83 @@ +// Package store contains PostgreSQL persistence boundaries for the control +// plane. Domain policy remains in package domain and is not duplicated here. +package store + +import ( + "context" + "database/sql" + "fmt" + "strings" + "time" +) + +const ( + DefaultSerializableAttempts = 3 + RetryBackoff = 10 * time.Millisecond +) + +// RunSerializable executes one logical mutation with PostgreSQL SERIALIZABLE +// isolation. Serialization failures and deadlocks retry the whole callback; +// partial work is never reused after rollback. +func RunSerializable(ctx context.Context, db *sql.DB, attempts int, fn func(context.Context, *sql.Tx) error) error { + if db == nil || fn == nil || attempts < 1 { + return fmt.Errorf("invalid serializable transaction arguments") + } + var last error + for attempt := 0; attempt < attempts; attempt++ { + tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable}) + if err != nil { + return err + } + err = fn(ctx, tx) + if err == nil { + err = tx.Commit() + } else { + _ = tx.Rollback() + } + if err == nil { + return nil + } + last = err + if !retryable(err) || attempt == attempts-1 { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(RetryBackoff * time.Duration(attempt+1)): + } + } + return last +} + +func retryable(err error) bool { + if err == nil { + return false + } + message := strings.ToLower(err.Error()) + return strings.Contains(message, "40001") || strings.Contains(message, "serialization failure") || strings.Contains(message, "40p01") || strings.Contains(message, "deadlock detected") +} + +var ( + // QueueTicketInsertSQL relies on the partial unique index in migration 0001 + // as the cross-replica one-active-ticket fence. + QueueTicketInsertSQL = `INSERT INTO queue_tickets + (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) + VALUES ($1, $2, $3, 'QUEUED', $4, $5, $6, $7)` + + // CandidateClaimSQL must run in the same serializable transaction as + // ProposalParticipantInsertSQL. SKIP LOCKED lets another matcher continue, + // while the participant unique/active indexes prevent a double claim. + CandidateClaimSQL = `SELECT ticket_id, player_id, playlist, client_build, protocol_version, enqueued_at, expires_at +FROM queue_tickets +WHERE state = 'QUEUED' AND expires_at > $1 +ORDER BY enqueued_at, ticket_id +LIMIT $2 +FOR UPDATE SKIP LOCKED` + + ProposalParticipantInsertSQL = `INSERT INTO proposal_participants (proposal_id, player_id, ticket_id, response) +VALUES ($1, $2, $3, 'PENDING')` + + QueueTicketProposeSQL = `UPDATE queue_tickets SET state = 'PROPOSED', revision = revision + 1 +WHERE ticket_id = $1 AND state = 'QUEUED' AND expires_at > $2` +) diff --git a/server/store/serializable_test.go b/server/store/serializable_test.go new file mode 100644 index 00000000..29e1c917 --- /dev/null +++ b/server/store/serializable_test.go @@ -0,0 +1,40 @@ +package store + +import ( + "errors" + "testing" +) + +func TestRetryableRecognisesPostgresSerializationAndDeadlockErrors(t *testing.T) { + for _, message := range []string{"pq: 40001 serialization_failure", "ERROR: deadlock detected (40P01)"} { + if !retryable(errors.New(message)) { + t.Fatalf("not retryable: %q", message) + } + } + for _, message := range []string{"duplicate key value violates unique constraint", "invalid input syntax"} { + if retryable(errors.New(message)) { + t.Fatalf("incorrectly retryable: %q", message) + } + } +} + +func TestClaimSQLContainsDurableOwnershipFences(t *testing.T) { + for _, fragment := range []string{"FOR UPDATE SKIP LOCKED", "state = 'QUEUED'", "proposal_participants", "revision = revision + 1"} { + if !containsAnySQL(fragment) { + t.Fatalf("claim boundary missing %q", fragment) + } + } +} + +func containsAnySQL(fragment string) bool { + return index(CandidateClaimSQL, fragment) >= 0 || index(ProposalParticipantInsertSQL, fragment) >= 0 || index(QueueTicketProposeSQL, fragment) >= 0 +} + +func index(s, fragment string) int { + for i := 0; i+len(fragment) <= len(s); i++ { + if s[i:i+len(fragment)] == fragment { + return i + } + } + return -1 +}