mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
84 lines
2.7 KiB
Go
84 lines
2.7 KiB
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"
|
|
"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 player_id = $2 AND playlist = $3 AND state = 'QUEUED' AND expires_at > $4`
|
|
)
|