diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index d5c13c42..e30de833 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -570,6 +570,77 @@ func TestPostgreSQLProposalDeclineRequeuesEveryParticipant(t *testing.T) { } } +// TestPostgreSQLProposalTimeoutRequeuesEveryParticipant is the timeout +// sibling of the decline test above: a proposal that simply times out (no +// explicit decline, nobody ever responds) hits the exact same +// ProposalExpireSQL/ProposalParticipantExpireSQL path with the exact same +// gap -- neither ever touched queue_tickets, so this is the same severe +// stranding bug reached a different way. Uses GetProposal (the recovery/read +// path) rather than RespondToProposal, since a real client that just missed +// the expiry event and comes back later to check on it is exactly the +// scenario this path exists for. +func TestPostgreSQLProposalTimeoutRequeuesEveryParticipant(t *testing.T) { + db := openIntegrationPostgres(t) + applyIntegrationMigrations(t, db) + + now := time.Now().UTC().Truncate(time.Microsecond) + ctx := context.Background() + for _, player := range []string{"timeout-player-a", "timeout-player-b"} { + if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil { + t.Fatal(err) + } + } + for i, player := range []string{"timeout-player-a", "timeout-player-b"} { + 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', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("timeout-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + } + proposal, err := domain.NewProposal("timeout-proposal", domain.Casual, []string{"timeout-player-a", "timeout-player-b"}, now) + if err != nil { + t.Fatal(err) + } + if err := CreateProposal(ctx, db, proposal, map[string]string{"timeout-player-a": "timeout-ticket-0", "timeout-player-b": "timeout-ticket-1"}, now); err != nil { + t.Fatalf("create proposal: %v", err) + } + + // Nobody ever responds; recover the proposal well after its 10s window, + // exactly as a client reconnecting after missing the expiry event would. + afterExpiry := now.Add(domain.ProposalWindow + time.Second) + recovered, err := GetProposal(ctx, db, "timeout-player-a", proposal.ProposalID, afterExpiry) + if err != nil { + t.Fatalf("recover expired proposal: %v", err) + } + if recovered.State != domain.Expired { + t.Fatalf("proposal did not expire: %+v", recovered) + } + + var stateA, stateB string + var expiresB time.Time + if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'timeout-ticket-0'`).Scan(&stateA); err != nil { + t.Fatal(err) + } + if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'timeout-ticket-1'`).Scan(&stateB, &expiresB); err != nil { + t.Fatal(err) + } + if stateA != "QUEUED" || stateB != "QUEUED" { + t.Fatalf("timed-out participants left stranded: a=%s b=%s", stateA, stateB) + } + if !expiresB.After(afterExpiry) { + t.Fatalf("requeued ticket expiry %v was not refreshed forward from %v", expiresB, afterExpiry) + } + candidates, err := ListQueuedCandidates(ctx, db, domain.Casual, afterExpiry, 10) + if err != nil { + t.Fatalf("list queued candidates: %v", err) + } + found := map[string]bool{} + for _, candidate := range candidates { + found[candidate.PlayerID] = true + } + if !found["timeout-player-a"] || !found["timeout-player-b"] { + t.Fatalf("requeued players are not visible to the matcher: %+v", candidates) + } +} + func TestPostgreSQLProposalCreationRollsBackPartialClaims(t *testing.T) { db := openIntegrationPostgres(t) applyIntegrationMigrations(t, db) diff --git a/server/store/proposal_recovery_sql.go b/server/store/proposal_recovery_sql.go index d7510e5e..fb2b21dc 100644 --- a/server/store/proposal_recovery_sql.go +++ b/server/store/proposal_recovery_sql.go @@ -21,6 +21,22 @@ SET response = 'TIMED_OUT', responded_at = $2 WHERE proposal_id = $1 AND response = 'PENDING' AND EXISTS (SELECT 1 FROM proposals WHERE proposals.proposal_id = proposal_participants.proposal_id AND proposals.expires_at <= $2)` +// ProposalExpireRequeueSQL is the timeout sibling of +// ProposalDeclineRequeueSQL: a proposal that simply times out (no unanimous +// response inside the 10s window) leaves any participant still holding a +// PROPOSED ticket exactly as stranded as an explicit decline does, and for +// the identical reason -- nothing else ever moves a PROPOSED ticket back to +// QUEUED. The `state = 'EXPIRED'` guard makes this safe to call +// unconditionally right after ProposalExpireSQL: it's a no-op on a proposal +// that was already OPEN and stays OPEN (nothing to requeue) or one that was +// already EXPIRED on a prior pass (its participants' tickets, if any were +// still PROPOSED, were already requeued then). +const ProposalExpireRequeueSQL = `UPDATE queue_tickets q +SET state = 'QUEUED', expires_at = $2, revision = revision + 1 +FROM proposal_participants pp +WHERE pp.proposal_id = $1 AND q.ticket_id = pp.ticket_id AND q.player_id = pp.player_id AND q.state = 'PROPOSED' + AND EXISTS (SELECT 1 FROM proposals WHERE proposals.proposal_id = $1 AND proposals.state = 'EXPIRED')` + const ProposalRecoverySelectSQL = `SELECT proposal_id, playlist, state, revision, expires_at FROM proposals WHERE proposal_id = $1 @@ -108,6 +124,9 @@ func GetProposal(ctx context.Context, db *sql.DB, playerID, proposalID string, n if _, err := tx.ExecContext(ctx, ProposalParticipantExpireSQL, proposalID, now); err != nil { return domain.Proposal{}, err } + if _, err := tx.ExecContext(ctx, ProposalExpireRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow)); err != nil { + return domain.Proposal{}, err + } var proposal domain.Proposal var playlist, state string if err := tx.QueryRowContext(ctx, ProposalRecoverySelectSQL, proposalID, playerID).Scan(&proposal.ProposalID, &playlist, &state, &proposal.Revision, &proposal.ExpiresAt); err != nil { @@ -188,6 +207,9 @@ func RespondToProposal(ctx context.Context, db *sql.DB, playerID, proposalID, id if _, err := tx.ExecContext(ctx, ProposalParticipantExpireSQL, proposalID, now); err != nil { return err } + if _, err := tx.ExecContext(ctx, ProposalExpireRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow)); err != nil { + return err + } if !now.Before(expiresAt) { return domain.ErrProposalClosed } diff --git a/server/store/proposal_recovery_sql_test.go b/server/store/proposal_recovery_sql_test.go index d2566d32..981d9620 100644 --- a/server/store/proposal_recovery_sql_test.go +++ b/server/store/proposal_recovery_sql_test.go @@ -17,6 +17,7 @@ func TestProposalRecoverySQLBindsParticipantAndExpiresAtReadBoundary(t *testing. ProposalParticipantRespondSQL: {"response = 'PENDING'", "responded_at"}, ProposalRevisionBumpSQL: {"revision = revision + 1", "state = 'OPEN'"}, ProposalDeclineRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "proposal_participants"}, + ProposalExpireRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "state = 'EXPIRED'"}, } { for _, fragment := range fragments { if !contains(query, fragment) {