CosmicClash/server/store/serializable.go

// 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"
	"math/rand/v2"
	"strings"
	"time"
)

const (
	// DefaultSerializableAttempts is the retry budget for one logical
	// mutation. Contention here is expected rather than exceptional: several
	// game servers can submit results, and several matchers can claim
	// candidates, against the same rows at once. Three attempts was too tight
	// for even five-way contention on identical rows.
	DefaultSerializableAttempts = 5
	// RetryBackoff is the base delay. The actual wait is jittered -- see
	// retryDelay -- because an unjittered backoff makes every contending
	// transaction wake at the same instants and collide again.
	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(retryDelay(attempt)):
		}
	}
	return last
}

// retryDelay applies full jitter to a linearly growing ceiling, so contending
// transactions do not all wake at the same instant and collide again.
//
// Measured honestly: jitter is the smaller half of this fix. Against the
// five-way contention in TestPostgreSQLConcurrentIdenticalResultSubmission,
// jitter alone moved the failure rate from 4/20 to 3/20, while raising the
// attempt budget from 3 to 5 took it to 0/20 on its own. The budget was the
// real constraint. Jitter is kept because it costs nothing and its benefit
// grows with the number of contending writers, which in production is not
// capped at five -- but it should not be mistaken for the reason this got
// better.
func retryDelay(attempt int) time.Duration {
	ceiling := RetryBackoff * time.Duration(attempt+1)
	if ceiling <= 0 {
		return 0
	}
	// math/rand/v2's top-level functions are safe for concurrent use, which
	// matters because every contending goroutine calls this.
	return time.Duration(rand.Int64N(int64(ceiling)))
}

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, predicted_rtt)
	VALUES ($1, $2, $3, 'QUEUED', $4, $5, $6, $7, $8)`

	// 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, team, slot)
VALUES ($1, $2, $3, 'PENDING', $4, $5)`

	QueueTicketProposeSQL = `UPDATE queue_tickets SET state = 'PROPOSED', revision = revision + 1
WHERE ticket_id = $1 AND player_id = $2 AND playlist = $3 AND state = 'QUEUED' AND expires_at > $4`
)