diff --git a/server/cmd/maintenance/main.go b/server/cmd/maintenance/main.go index 8fcaf861..e533d82d 100644 --- a/server/cmd/maintenance/main.go +++ b/server/cmd/maintenance/main.go @@ -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(): diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index bfe3c2b4..bc015da9 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -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) diff --git a/server/store/stalled_allocation_sql.go b/server/store/stalled_allocation_sql.go new file mode 100644 index 00000000..0e9aecfc --- /dev/null +++ b/server/store/stalled_allocation_sql.go @@ -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 +} diff --git a/server/store/stalled_allocation_sql_test.go b/server/store/stalled_allocation_sql_test.go new file mode 100644 index 00000000..6360d811 --- /dev/null +++ b/server/store/stalled_allocation_sql_test.go @@ -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") + } +}