mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-11 08:23:45 +00:00
feat(multiplayer): reclaim stalled allocations without penalising players
Closes the last named item on task 8.28 (Health-reclaim): nothing
currently detects or cleans up a match stuck in
ALLOCATING/PROCESS_READY/ASSIGNMENT_READY forever because its server
crashed or was reclaimed by Agones as unhealthy before ever
registering -- players would wait indefinitely for a match that was
never coming.
The design question this was blocked on -- does an abandoned match
auto-requeue its players, or fail and make them re-queue -- isn't
actually open: task 8.50's own stated acceptance criterion already
answers it ("infrastructure-caused cases cannot penalise affected
players"). A server-side crash/reclaim is exactly that, not player
behaviour, so store.ExpireStalledAllocations fails the match but
requeues every participant's ticket to QUEUED with a fresh expiry
(matching the ordinary 30s queue window), releases their
match_participants row (participation_active = false, so they're
matchable again immediately), all inside one FOR UPDATE SKIP LOCKED
pass so a second maintenance replica continues past whatever a
concurrent one is already reclaiming.
Wired into cmd/maintenance alongside the existing season-rollover
sweep: --stalled-allocation-deadline (default 2m) and
--stalled-allocation-batch (default 100).
Covered by a SQL-fragment test and a real PostgreSQL integration test:
two matches (one genuinely stalled, one recent), confirming the
deadline boundary is respected (recent match untouched), both
stranded participants' tickets requeue with a refreshed expiry, the
match_participants row releases, and a second pass doesn't reprocess
an already-FAILED match. Verified clean across 5 runs, plus the full
integration and unit suites.
This commit is contained in:
@@ -21,6 +21,8 @@ func main() {
|
||||
migrationDir := flag.String("migrations", "migrations", "directory containing numbered SQL migrations")
|
||||
interval := flag.Duration("interval", time.Minute, "maintenance poll interval")
|
||||
batch := flag.Int("batch", 100, "maximum player rollovers per pass")
|
||||
stalledAllocationDeadline := flag.Duration("stalled-allocation-deadline", 2*time.Minute, "reclaim a match stuck in ALLOCATING/PROCESS_READY/ASSIGNMENT_READY (server crashed or was reclaimed before registering) after this long, requeuing every participant without penalty")
|
||||
stalledAllocationBatch := flag.Int("stalled-allocation-batch", 100, "maximum stalled matches reclaimed per pass")
|
||||
flag.Parse()
|
||||
if *dsn == "" {
|
||||
fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required")
|
||||
@@ -28,6 +30,9 @@ func main() {
|
||||
if *interval <= 0 || *batch < 1 || *batch > 1000 {
|
||||
fatalf("invalid interval or batch")
|
||||
}
|
||||
if *stalledAllocationDeadline <= 0 || *stalledAllocationBatch < 1 || *stalledAllocationBatch > 1000 {
|
||||
fatalf("invalid stalled-allocation deadline or batch")
|
||||
}
|
||||
db, err := sql.Open("pgx", *dsn)
|
||||
if err != nil {
|
||||
fatalf("open PostgreSQL: %v", err)
|
||||
@@ -44,13 +49,21 @@ func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
for {
|
||||
count, err := store.RolloverDueSeasons(ctx, db, time.Now().UTC(), *batch)
|
||||
now := time.Now().UTC()
|
||||
count, err := store.RolloverDueSeasons(ctx, db, now, *batch)
|
||||
if err != nil {
|
||||
fatalf("season maintenance: %v", err)
|
||||
}
|
||||
if count > 0 {
|
||||
log.Printf("applied %d ranked season rollovers", count)
|
||||
}
|
||||
reclaimed, err := store.ExpireStalledAllocations(ctx, db, now, *stalledAllocationDeadline, *stalledAllocationBatch)
|
||||
if err != nil {
|
||||
fatalf("stalled-allocation maintenance: %v", err)
|
||||
}
|
||||
if reclaimed > 0 {
|
||||
log.Printf("reclaimed %d stalled allocations, requeuing their participants", reclaimed)
|
||||
}
|
||||
timer := time.NewTimer(*interval)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
|
||||
@@ -781,6 +781,109 @@ func TestPostgreSQLConcurrentIdenticalResultSubmissionAppliesRatingsExactlyOnce(
|
||||
}
|
||||
}
|
||||
|
||||
// TestPostgreSQLStalledAllocationsAreReclaimedWithoutPenalisingPlayers is the
|
||||
// live counterpart to the SQL fragment test: it proves the actual data
|
||||
// movement against a real database, not just that the right substrings are
|
||||
// present. Two matches: one genuinely stalled (old enough to reclaim), one
|
||||
// recent (must survive untouched) -- the deadline boundary and the
|
||||
// no-penalty requeue are both meaningless without a real row to check.
|
||||
func TestPostgreSQLStalledAllocationsAreReclaimedWithoutPenalisingPlayers(t *testing.T) {
|
||||
db := openIntegrationPostgres(t)
|
||||
applyIntegrationMigrations(t, db)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC().Truncate(time.Microsecond)
|
||||
stalledCreatedAt := now.Add(-10 * time.Minute)
|
||||
recentCreatedAt := now.Add(-5 * time.Second)
|
||||
|
||||
for _, player := range []string{"stall-player-a", "stall-player-b", "recent-player"} {
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
insertTicket := func(ticketID, playerID, state string, expiresAt time.Time) {
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', $3, 'integration-build', 1, $4, $5)`, ticketID, playerID, state, now, expiresAt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
insertTicket("stall-ticket-a", "stall-player-a", "PROCESS_READY", now.Add(time.Hour))
|
||||
insertTicket("stall-ticket-b", "stall-player-b", "PROCESS_READY", now.Add(time.Hour))
|
||||
insertTicket("recent-ticket", "recent-player", "ALLOCATING", now.Add(time.Hour))
|
||||
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, created_at) VALUES ('stalled-match', 'casual', 'PROCESS_READY', 'NA', 1, 'stalled-server', $1)`, stalledCreatedAt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, created_at) VALUES ('recent-match', 'casual', 'ALLOCATING', 'NA', 1, $1)`, recentCreatedAt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('stalled-match', 'stall-player-a', 'stall-ticket-a', 0, 0), ('stalled-match', 'stall-player-b', 'stall-ticket-b', 1, 1)`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('recent-match', 'recent-player', 'recent-ticket', 0, 0)`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
reclaimed, err := ExpireStalledAllocations(ctx, db, now, 2*time.Minute, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("expire stalled allocations: %v", err)
|
||||
}
|
||||
if reclaimed != 1 {
|
||||
t.Fatalf("reclaimed = %d, want exactly 1 (the recent match must survive)", reclaimed)
|
||||
}
|
||||
|
||||
var stalledState, recentState string
|
||||
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'stalled-match'`).Scan(&stalledState); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'recent-match'`).Scan(&recentState); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stalledState != "FAILED" {
|
||||
t.Fatalf("stalled match state = %s, want FAILED", stalledState)
|
||||
}
|
||||
if recentState != "ALLOCATING" {
|
||||
t.Fatalf("recent match state = %s, want untouched ALLOCATING", recentState)
|
||||
}
|
||||
|
||||
var ticketAState, ticketBState, recentTicketState string
|
||||
var ticketAExpiry time.Time
|
||||
if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'stall-ticket-a'`).Scan(&ticketAState, &ticketAExpiry); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'stall-ticket-b'`).Scan(&ticketBState); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'recent-ticket'`).Scan(&recentTicketState); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ticketAState != "QUEUED" || ticketBState != "QUEUED" {
|
||||
t.Fatalf("stalled participants' tickets = %s, %s -- want both requeued to QUEUED, not failed/left behind", ticketAState, ticketBState)
|
||||
}
|
||||
if !ticketAExpiry.After(now) {
|
||||
t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", ticketAExpiry, now)
|
||||
}
|
||||
if recentTicketState != "ALLOCATING" {
|
||||
t.Fatalf("recent match's ticket state = %s, want untouched ALLOCATING", recentTicketState)
|
||||
}
|
||||
|
||||
var activeParticipants int
|
||||
if err := db.QueryRow(`SELECT count(*) FROM match_participants WHERE match_id = 'stalled-match' AND participation_active`).Scan(&activeParticipants); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if activeParticipants != 0 {
|
||||
t.Fatalf("stalled match still has %d active participants, want 0 (so the player can be matched again)", activeParticipants)
|
||||
}
|
||||
|
||||
// Idempotent: the match is now FAILED, not one of the three reclaimable
|
||||
// states, so a second pass must not touch it again.
|
||||
reclaimedAgain, err := ExpireStalledAllocations(ctx, db, now.Add(time.Minute), 2*time.Minute, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("second expire pass: %v", err)
|
||||
}
|
||||
if reclaimedAgain != 0 {
|
||||
t.Fatalf("second pass reclaimed %d matches, want 0 (already-FAILED match must not be reprocessed)", reclaimedAgain)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgreSQLRankedSeasonRolloverIsExactlyOnce(t *testing.T) {
|
||||
db := openIntegrationPostgres(t)
|
||||
applyIntegrationMigrations(t, db)
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
||||
)
|
||||
|
||||
// ExpireStalledAllocationsSQL reclaims a match that has sat in
|
||||
// ALLOCATING/PROCESS_READY/ASSIGNMENT_READY past the deadline -- its server
|
||||
// crashed, was reclaimed by Agones as unhealthy, or otherwise never finished
|
||||
// registering. Requeues every participant instead of just failing the match:
|
||||
// task 8.50's own stated acceptance criterion is that "infrastructure-caused
|
||||
// cases cannot penalise affected players", and a server-side failure here is
|
||||
// exactly that, not player behaviour. FOR UPDATE SKIP LOCKED lets a second
|
||||
// maintenance replica continue past whatever a concurrent one is already
|
||||
// reclaiming rather than blocking on it.
|
||||
const ExpireStalledAllocationsSQL = `WITH stalled AS (
|
||||
SELECT match_id FROM matches
|
||||
WHERE state IN ('ALLOCATING', 'PROCESS_READY', 'ASSIGNMENT_READY') AND created_at <= $1
|
||||
ORDER BY created_at, match_id
|
||||
LIMIT $2
|
||||
FOR UPDATE SKIP LOCKED
|
||||
), failed AS (
|
||||
UPDATE matches SET state = 'FAILED', revision = revision + 1
|
||||
WHERE match_id IN (SELECT match_id FROM stalled)
|
||||
RETURNING match_id
|
||||
), released AS (
|
||||
UPDATE match_participants SET participation_active = FALSE
|
||||
WHERE match_id IN (SELECT match_id FROM failed) AND participation_active
|
||||
RETURNING ticket_id
|
||||
), requeued AS (
|
||||
UPDATE queue_tickets SET state = 'QUEUED', expires_at = $3, revision = revision + 1
|
||||
WHERE ticket_id IN (SELECT ticket_id FROM released)
|
||||
RETURNING ticket_id
|
||||
)
|
||||
SELECT (SELECT count(*) FROM failed), (SELECT count(*) FROM requeued)`
|
||||
|
||||
// ExpireStalledAllocations reclaims up to `limit` matches whose
|
||||
// created_at is at or before `now - deadline` and are still stuck in one of
|
||||
// the pre-live allocation states, failing the match and requeuing every
|
||||
// participant's ticket with a fresh expiry rather than penalising them. It
|
||||
// returns the number of matches reclaimed.
|
||||
func ExpireStalledAllocations(ctx context.Context, db *sql.DB, now time.Time, deadline time.Duration, limit int) (int, error) {
|
||||
if db == nil || now.IsZero() || deadline <= 0 || limit < 1 || limit > 1000 {
|
||||
return 0, fmt.Errorf("invalid stalled-allocation maintenance arguments")
|
||||
}
|
||||
var matches, requeued int
|
||||
err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
||||
return tx.QueryRowContext(ctx, ExpireStalledAllocationsSQL, now.Add(-deadline), limit, now.Add(domain.QueueExpiryWindow)).Scan(&matches, &requeued)
|
||||
})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return matches, nil
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestExpireStalledAllocationsSQLFencesAndRequeuesWithoutPenalty(t *testing.T) {
|
||||
for _, fragment := range []string{
|
||||
"ALLOCATING", "PROCESS_READY", "ASSIGNMENT_READY",
|
||||
"FOR UPDATE SKIP LOCKED",
|
||||
"SET state = 'FAILED'",
|
||||
"SET participation_active = FALSE",
|
||||
"SET state = 'QUEUED'",
|
||||
} {
|
||||
if !strings.Contains(ExpireStalledAllocationsSQL, fragment) {
|
||||
t.Fatalf("ExpireStalledAllocationsSQL missing fragment %q:\n%s", fragment, ExpireStalledAllocationsSQL)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpireStalledAllocationsRejectsInvalidArgumentsWithoutDatabase(t *testing.T) {
|
||||
now := time.Unix(1000, 0).UTC()
|
||||
if _, err := ExpireStalledAllocations(context.Background(), nil, now, time.Minute, 10); err == nil {
|
||||
t.Fatal("nil database accepted")
|
||||
}
|
||||
if _, err := ExpireStalledAllocations(context.Background(), nil, time.Time{}, time.Minute, 10); err == nil {
|
||||
t.Fatal("zero time accepted")
|
||||
}
|
||||
if _, err := ExpireStalledAllocations(context.Background(), nil, now, 0, 10); err == nil {
|
||||
t.Fatal("non-positive deadline accepted")
|
||||
}
|
||||
if _, err := ExpireStalledAllocations(context.Background(), nil, now, time.Minute, 0); err == nil {
|
||||
t.Fatal("zero limit accepted")
|
||||
}
|
||||
if _, err := ExpireStalledAllocations(context.Background(), nil, now, time.Minute, 1001); err == nil {
|
||||
t.Fatal("oversized limit accepted")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user