feat(multiplayer): persist live reconnect abandonments

server/cmd/maintenance/main.go

@@ -25,6 +25,7 @@ func main() {
 	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")
 	initialConnectBatch := flag.Int("initial-connect-batch", 100, "maximum pre-live matches evaluated per pass")
+	liveAbandonmentBatch := flag.Int("live-abandonment-batch", 100, "maximum live ranked matches evaluated for expired reconnect leases per pass")
 	flag.Parse()
 	if *dsn == "" {
 		fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required")
@@ -35,8 +36,8 @@ func main() {
 	if *stalledAllocationDeadline <= 0 || *stalledAllocationBatch < 1 || *stalledAllocationBatch > 1000 {
 		fatalf("invalid stalled-allocation deadline or batch")
 	}
-	if *initialConnectBatch < 1 || *initialConnectBatch > 1000 {
-		fatalf("invalid initial-connect batch")
+	if *initialConnectBatch < 1 || *initialConnectBatch > 1000 || *liveAbandonmentBatch < 1 || *liveAbandonmentBatch > 1000 {
+		fatalf("invalid initial-connect or live-abandonment batch")
 	}
 	db, err := sql.Open("pgx", *dsn)
 	if err != nil {
@@ -77,6 +78,13 @@ func main() {
 		if reconciled > 0 {
 			log.Printf("reconciled %d initial-connect outcomes", reconciled)
 		}
+		abandoned, err := store.ReconcileLiveAbandonments(ctx, db, now, *liveAbandonmentBatch)
+		if err != nil {
+			fatalf("live-abandonment maintenance: %v", err)
+		}
+		if abandoned > 0 {
+			log.Printf("recorded expired reconnect leases in %d live matches", abandoned)
+		}
 	}
 
 	runGeneral(time.Now().UTC())

server/domain/reconnect.go

@@ -156,6 +156,38 @@ type Abandonment struct {
 	AbandonedAt time.Time
 }
 
+// ReconnectParticipant is the durable subset needed to evaluate an expired
+// live reconnect lease. Connected players are deliberately absent: only a
+// persisted disconnect can start a player-caused abandon clock.
+type ReconnectParticipant struct {
+	PlayerID       string
+	DisconnectedAt time.Time
+}
+
+// PlanRankedAbandonments turns expired durable reconnect leases into the
+// same rolling cooldown ladder used by pre-live ranked no-shows. Future
+// disconnect timestamps are ignored rather than penalised: they can only be
+// an infrastructure clock anomaly, not a player abandonment.
+func PlanRankedAbandonments(now time.Time, participants []ReconnectParticipant, priorAbandons map[string][]time.Time) ([]Abandonment, error) {
+	if now.IsZero() {
+		return nil, fmt.Errorf("invalid reconnect-abandonment time")
+	}
+	seen := make(map[string]bool, len(participants))
+	result := make([]Abandonment, 0, len(participants))
+	for _, participant := range participants {
+		if participant.PlayerID == "" || participant.DisconnectedAt.IsZero() || seen[participant.PlayerID] {
+			return nil, fmt.Errorf("invalid reconnect participant")
+		}
+		seen[participant.PlayerID] = true
+		if now.Before(participant.DisconnectedAt) || now.Sub(participant.DisconnectedAt) <= RankedReconnectGrace {
+			continue
+		}
+		result = append(result, Abandonment{PlayerID: participant.PlayerID, Cooldown: abandonCooldown(priorAbandons[participant.PlayerID], now), AbandonedAt: now})
+	}
+	sort.Slice(result, func(i, j int) bool { return result[i].PlayerID < result[j].PlayerID })
+	return result, nil
+}
+
 // ExpireGrace marks every disconnected player whose 60-second reclaim window
 // has elapsed. The returned list is lexical for stable audit/event ordering.
 func (r *RankedConnections) ExpireGrace(now time.Time, priorAbandons map[string][]time.Time) []Abandonment {

server/domain/reconnect_test.go

@@ -140,6 +140,21 @@ func TestRankedAbandonCooldownUsesRollingSevenDayLadder(t *testing.T) {
 	}
 }
 
+func TestPlanRankedAbandonmentsFencesGraceAndClockAnomalies(t *testing.T) {
+	now := time.Unix(1000, 0).UTC()
+	planned, err := PlanRankedAbandonments(now, []ReconnectParticipant{
+		{PlayerID: "within", DisconnectedAt: now.Add(-RankedReconnectGrace)},
+		{PlayerID: "future", DisconnectedAt: now.Add(time.Second)},
+		{PlayerID: "expired", DisconnectedAt: now.Add(-RankedReconnectGrace - time.Nanosecond)},
+	}, map[string][]time.Time{"expired": {now.Add(-time.Hour)}})
+	if err != nil || len(planned) != 1 || planned[0].PlayerID != "expired" || planned[0].Cooldown != 15*time.Minute || !planned[0].AbandonedAt.Equal(now) {
+		t.Fatalf("planned=%+v err=%v", planned, err)
+	}
+	if _, err := PlanRankedAbandonments(now, []ReconnectParticipant{{PlayerID: "duplicate", DisconnectedAt: now}, {PlayerID: "duplicate", DisconnectedAt: now}}, nil); err == nil {
+		t.Fatal("duplicate reconnect participant accepted")
+	}
+}
+
 func TestSignedJoinAuthorisationBindsEveryClaimBeforeReclaim(t *testing.T) {
 	now := time.Unix(1000, 0).UTC()
 	r, err := NewRankedConnections("match-1", "server-1", "v1", testRoster(now))

server/store/initial_connect_maintenance.go

@@ -19,7 +19,7 @@ LIMIT $1`
 
 const initialConnectHistorySQL = `SELECT starts_at
 FROM penalties
-WHERE player_id = $1 AND kind = 'INITIAL_CONNECT_NO_SHOW'
+WHERE player_id = $1 AND kind IN ('INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED')
 ORDER BY starts_at`
 
 // ReconcileInitialConnect evaluates a bounded set of matches and applies only

server/store/initial_connect_sql_test.go

@@ -27,6 +27,14 @@ func TestInitialConnectSQLPreservesAtomicNoShowReconciliation(t *testing.T) {
 			}
 		}
 	}
+	for _, fragment := range []string{"INITIAL_CONNECT_NO_SHOW", "MATCH_ABANDONED"} {
+		if !contains(initialConnectHistorySQL, fragment) {
+			t.Fatalf("initial-connect abandon history missing %q", fragment)
+		}
+		if !contains(QueueCooldownSelectSQL, fragment) {
+			t.Fatalf("queue cooldown fence missing %q", fragment)
+		}
+	}
 }
 
 func TestInitialConnectPlanValidationRejectsIncompleteOrForgedPlans(t *testing.T) {

server/store/live_abandonment_sql.go

@@ -0,0 +1,213 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+const liveAbandonmentCandidatesSQL = `SELECT m.match_id
+FROM matches m
+WHERE m.playlist = 'ranked' AND m.state = 'LIVE'
+  AND EXISTS (
+    SELECT 1 FROM match_participants mp
+    WHERE mp.match_id = m.match_id AND mp.participation_active
+      AND mp.abandoned_at IS NULL AND mp.disconnected_at IS NOT NULL
+      AND mp.disconnected_at < $1
+  )
+ORDER BY m.match_id
+LIMIT $2`
+
+const liveAbandonmentMatchLockSQL = `SELECT playlist, state
+FROM matches WHERE match_id = $1 FOR UPDATE`
+
+const liveAbandonmentParticipantsSQL = `SELECT player_id, disconnected_at
+FROM match_participants
+WHERE match_id = $1 AND participation_active
+  AND abandoned_at IS NULL AND disconnected_at IS NOT NULL
+ORDER BY player_id
+FOR UPDATE`
+
+const liveAbandonmentHistorySQL = `SELECT starts_at
+FROM penalties
+WHERE player_id = $1 AND kind IN ('INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED')
+ORDER BY starts_at`
+
+const liveAbandonmentParticipantSQL = `UPDATE match_participants
+SET abandoned_at = $3
+WHERE match_id = $1 AND player_id = $2 AND participation_active
+  AND abandoned_at IS NULL AND disconnected_at IS NOT NULL
+RETURNING player_id`
+
+const liveAbandonmentPenaltySQL = `INSERT INTO penalties
+    (penalty_id, player_id, match_id, playlist, kind, starts_at, ends_at)
+VALUES ($1, $2, $3, 'ranked', 'MATCH_ABANDONED', $4, $5)
+ON CONFLICT (penalty_id) DO NOTHING`
+
+const liveAbandonmentRevisionSQL = `UPDATE matches
+SET revision = revision + 1
+WHERE match_id = $1 AND state = 'LIVE'
+RETURNING revision`
+
+const liveAbandonmentOutboxSQL = `INSERT INTO outbox
+    (event_id, aggregate_type, aggregate_id, revision, event_type, payload)
+VALUES ($1, 'match', $2, $3, 'participant_abandoned', $4)`
+
+// ReconcileLiveAbandonments applies a bounded, durable reconnect-grace sweep.
+// It does not deactivate participants or alter LIVE tickets: an abandonment
+// must remain in the authoritative result roster so rating correctly scores a
+// loss if the match later completes.
+func ReconcileLiveAbandonments(ctx context.Context, db *sql.DB, now time.Time, limit int) (int, error) {
+	if db == nil || now.IsZero() || limit < 1 || limit > 1000 {
+		return 0, fmt.Errorf("invalid live-abandonment maintenance arguments")
+	}
+	rows, err := db.QueryContext(ctx, liveAbandonmentCandidatesSQL, now.Add(-domain.RankedReconnectGrace), limit)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	var matchIDs []string
+	for rows.Next() {
+		var matchID string
+		if err := rows.Scan(&matchID); err != nil {
+			return 0, err
+		}
+		matchIDs = append(matchIDs, matchID)
+	}
+	if err := rows.Err(); err != nil {
+		return 0, err
+	}
+	// Do not hold the candidate cursor while opening serializable per-match
+	// transactions. A deliberately small production pool (including size one)
+	// would otherwise wait on its own still-open read connection.
+	if err := rows.Close(); err != nil {
+		return 0, err
+	}
+	count := 0
+	for _, matchID := range matchIDs {
+		changed, err := ApplyLiveAbandonments(ctx, db, matchID, now)
+		if err != nil {
+			return count, err
+		}
+		if changed > 0 {
+			count++
+		}
+	}
+	return count, nil
+}
+
+// ApplyLiveAbandonments is independently serializable so concurrent
+// maintenance replicas or a result submission cannot double-penalise a
+// player. It returns the number of participants newly abandoned.
+func ApplyLiveAbandonments(ctx context.Context, db *sql.DB, matchID string, now time.Time) (int, error) {
+	if db == nil || matchID == "" || now.IsZero() {
+		return 0, fmt.Errorf("invalid live-abandonment transaction arguments")
+	}
+	changed := 0
+	err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
+		var playlist, state string
+		if err := tx.QueryRowContext(ctx, liveAbandonmentMatchLockSQL, matchID).Scan(&playlist, &state); err != nil {
+			return err
+		}
+		if playlist != string(domain.Ranked) || state != string(domain.Live) {
+			return nil
+		}
+		participants, err := loadLiveReconnectParticipants(ctx, tx, matchID)
+		if err != nil {
+			return err
+		}
+		history, err := loadLiveAbandonmentHistory(ctx, tx, participants)
+		if err != nil {
+			return err
+		}
+		planned, err := domain.PlanRankedAbandonments(now, participants, history)
+		if err != nil {
+			return err
+		}
+		if len(planned) == 0 {
+			return nil
+		}
+		for _, abandonment := range planned {
+			var playerID string
+			if err := tx.QueryRowContext(ctx, liveAbandonmentParticipantSQL, matchID, abandonment.PlayerID, abandonment.AbandonedAt).Scan(&playerID); err != nil {
+				if err == sql.ErrNoRows {
+					return fmt.Errorf("%w: reconnect participant changed", domain.ErrConflict)
+				}
+				return err
+			}
+			penaltyID := "live-abandon:" + matchID + ":" + abandonment.PlayerID
+			if _, err := tx.ExecContext(ctx, liveAbandonmentPenaltySQL, penaltyID, abandonment.PlayerID, matchID, abandonment.AbandonedAt, abandonment.AbandonedAt.Add(abandonment.Cooldown)); err != nil {
+				return err
+			}
+		}
+		var revision uint64
+		if err := tx.QueryRowContext(ctx, liveAbandonmentRevisionSQL, matchID).Scan(&revision); err != nil {
+			return err
+		}
+		payload, err := json.Marshal(map[string]any{"match_id": matchID, "abandoned_player_ids": abandonmentIDs(planned)})
+		if err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, liveAbandonmentOutboxSQL, fmt.Sprintf("live-abandon:%s:%d", matchID, revision), matchID, revision, payload); err != nil {
+			return err
+		}
+		changed = len(planned)
+		return nil
+	})
+	return changed, err
+}
+
+func loadLiveReconnectParticipants(ctx context.Context, tx *sql.Tx, matchID string) ([]domain.ReconnectParticipant, error) {
+	rows, err := tx.QueryContext(ctx, liveAbandonmentParticipantsSQL, matchID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	participants := make([]domain.ReconnectParticipant, 0)
+	for rows.Next() {
+		var participant domain.ReconnectParticipant
+		if err := rows.Scan(&participant.PlayerID, &participant.DisconnectedAt); err != nil {
+			return nil, err
+		}
+		participants = append(participants, participant)
+	}
+	return participants, rows.Err()
+}
+
+func loadLiveAbandonmentHistory(ctx context.Context, tx *sql.Tx, participants []domain.ReconnectParticipant) (map[string][]time.Time, error) {
+	history := make(map[string][]time.Time, len(participants))
+	for _, participant := range participants {
+		rows, err := tx.QueryContext(ctx, liveAbandonmentHistorySQL, participant.PlayerID)
+		if err != nil {
+			return nil, err
+		}
+		for rows.Next() {
+			var started time.Time
+			if err := rows.Scan(&started); err != nil {
+				rows.Close()
+				return nil, err
+			}
+			history[participant.PlayerID] = append(history[participant.PlayerID], started)
+		}
+		if err := rows.Err(); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		if err := rows.Close(); err != nil {
+			return nil, err
+		}
+	}
+	return history, nil
+}
+
+func abandonmentIDs(abandonments []domain.Abandonment) []string {
+	ids := make([]string, len(abandonments))
+	for i := range abandonments {
+		ids[i] = abandonments[i].PlayerID
+	}
+	return ids
+}

server/store/live_abandonment_sql_test.go

@@ -0,0 +1,27 @@
+package store
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestLiveAbandonmentSQLPreservesResultRosterAndReconnectFences(t *testing.T) {
+	for query, fragments := range map[string][]string{
+		liveAbandonmentCandidatesSQL:   {"playlist = 'ranked'", "state = 'LIVE'", "abandoned_at IS NULL", "disconnected_at < $1", "LIMIT $2"},
+		liveAbandonmentMatchLockSQL:    {"FOR UPDATE", "match_id = $1"},
+		liveAbandonmentParticipantsSQL: {"participation_active", "abandoned_at IS NULL", "disconnected_at IS NOT NULL", "FOR UPDATE"},
+		liveAbandonmentParticipantSQL:  {"SET abandoned_at", "participation_active", "abandoned_at IS NULL", "RETURNING"},
+		liveAbandonmentPenaltySQL:      {"MATCH_ABANDONED", "ON CONFLICT"},
+		liveAbandonmentRevisionSQL:     {"state = 'LIVE'", "revision = revision + 1"},
+		liveAbandonmentOutboxSQL:       {"participant_abandoned", "revision"},
+	} {
+		for _, fragment := range fragments {
+			if !strings.Contains(query, fragment) {
+				t.Fatalf("query missing %q: %s", fragment, query)
+			}
+		}
+	}
+	if strings.Contains(liveAbandonmentParticipantSQL, "participation_active = FALSE") || strings.Contains(liveAbandonmentParticipantSQL, "queue_tickets") {
+		t.Fatal("live abandonment must retain participant and ticket for the result transaction")
+	}
+}

server/store/postgres_integration_test.go

@@ -649,6 +649,57 @@ func TestPostgreSQLConnectionReceiptsStartCompleteRelaxedCasualRoster(t *testing
 	}
 }
 
+func TestPostgreSQLLiveReconnectGraceExpiryPersistsAbandonmentWithoutReleasingResultRoster(t *testing.T) {
+	db := openIntegrationPostgres(t)
+	applyIntegrationMigrations(t, db)
+	ctx := context.Background()
+	now := time.Now().UTC().Truncate(time.Microsecond)
+	for _, playerID := range []string{"live-abandon-player", "live-present-player"} {
+		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, playerID); err != nil {
+			t.Fatal(err)
+		}
+		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, 'ranked', 'LIVE', 'integration-build', 1, $3, $4)`, "live-abandon-ticket-"+playerID, playerID, now, now.Add(time.Hour)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id) VALUES ('live-abandon-match', 'ranked', 'LIVE', 'EU', 1, 'live-abandon-server')`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team, connection_generation, connected_at, disconnected_at) VALUES
+('live-abandon-match', 'live-abandon-player', 'live-abandon-ticket-live-abandon-player', 0, 0, 1, $1, $2),
+('live-abandon-match', 'live-present-player', 'live-abandon-ticket-live-present-player', 3, 1, 1, $1, NULL)`, now.Add(-2*time.Minute), now.Add(-domain.RankedReconnectGrace-time.Nanosecond)); err != nil {
+		t.Fatal(err)
+	}
+
+	reconciled, err := ReconcileLiveAbandonments(ctx, db, now, 10)
+	if err != nil || reconciled != 1 {
+		t.Fatalf("reconciled=%d err=%v", reconciled, err)
+	}
+	var active bool
+	var abandonedAt sql.NullTime
+	if err := db.QueryRowContext(ctx, `SELECT participation_active, abandoned_at FROM match_participants WHERE match_id = 'live-abandon-match' AND player_id = 'live-abandon-player'`).Scan(&active, &abandonedAt); err != nil || !active || !abandonedAt.Valid || !abandonedAt.Time.Equal(now) {
+		t.Fatalf("participant active=%t abandoned=%v err=%v", active, abandonedAt, err)
+	}
+	var ticketState string
+	if err := db.QueryRowContext(ctx, `SELECT state FROM queue_tickets WHERE ticket_id = 'live-abandon-ticket-live-abandon-player'`).Scan(&ticketState); err != nil || ticketState != "LIVE" {
+		t.Fatalf("ticket state=%q err=%v", ticketState, err)
+	}
+	var endsAt time.Time
+	if err := db.QueryRowContext(ctx, `SELECT ends_at FROM penalties WHERE player_id = 'live-abandon-player' AND kind = 'MATCH_ABANDONED'`).Scan(&endsAt); err != nil || !endsAt.Equal(now.Add(5*time.Minute)) {
+		t.Fatalf("penalty ends=%v err=%v", endsAt, err)
+	}
+	var outboxCount, revision int
+	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM outbox WHERE aggregate_id = 'live-abandon-match' AND event_type = 'participant_abandoned'`).Scan(&outboxCount); err != nil || outboxCount != 1 {
+		t.Fatalf("outbox=%d err=%v", outboxCount, err)
+	}
+	if err := db.QueryRowContext(ctx, `SELECT revision FROM matches WHERE match_id = 'live-abandon-match'`).Scan(&revision); err != nil || revision != 1 {
+		t.Fatalf("revision=%d err=%v", revision, err)
+	}
+	if reconciled, err = ReconcileLiveAbandonments(ctx, db, now.Add(time.Minute), 10); err != nil || reconciled != 0 {
+		t.Fatalf("replay reconciled=%d err=%v", reconciled, err)
+	}
+}
+
 func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) {
 	db := openIntegrationPostgres(t)
 	applyIntegrationMigrations(t, db)

server/store/queue_sql.go

@@ -52,7 +52,7 @@ FOR UPDATE`
 	QueueCooldownSelectSQL = `SELECT ends_at
 FROM penalties
 WHERE player_id = $1 AND playlist = $2
-  AND kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT')
+  AND kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT', 'INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED')
   AND ends_at > $3
 ORDER BY ends_at DESC
 LIMIT 1`