feat: add serializable matchmaking store boundary

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`
+)

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
+}