diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index e30de833..2c7cfa92 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -641,6 +641,78 @@ func TestPostgreSQLProposalTimeoutRequeuesEveryParticipant(t *testing.T) { } } +// TestPostgreSQLCancellingAProposedTicketImmediatelyRequeuesTheOtherParticipant +// covers the responsiveness gap the decline/timeout fixes above left bounded +// but not closed: cancelling a ticket that's part of an OPEN proposal used +// to leave the OTHER participant waiting out the full 10s window for +// something the system already knew couldn't happen (their proposal partner +// just walked away). CascadeCancelToOpenProposal declines and requeues that +// proposal in the same transaction as the cancel itself. +func TestPostgreSQLCancellingAProposedTicketImmediatelyRequeuesTheOtherParticipant(t *testing.T) { + db := openIntegrationPostgres(t) + applyIntegrationMigrations(t, db) + + now := time.Now().UTC().Truncate(time.Microsecond) + ctx := context.Background() + for _, player := range []string{"cancel-cascade-a", "cancel-cascade-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{"cancel-cascade-a", "cancel-cascade-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("cancel-cascade-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + } + proposal, err := domain.NewProposal("cancel-cascade-proposal", domain.Casual, []string{"cancel-cascade-a", "cancel-cascade-b"}, now) + if err != nil { + t.Fatal(err) + } + if err := CreateProposal(ctx, db, proposal, map[string]string{"cancel-cascade-a": "cancel-cascade-ticket-0", "cancel-cascade-b": "cancel-cascade-ticket-1"}, now); err != nil { + t.Fatalf("create proposal: %v", err) + } + + // player-a cancels their own ticket directly, well within the response + // window -- not a decline, not a timeout, just abandoning the queue. + // CreateProposal's own QueueTicketProposeSQL already bumped the ticket's + // revision from 0 to 1, so the cancel's expected revision is 1, not 0. + cancelled, err := CancelQueueTicket(ctx, db, "cancel-cascade-a", "cancel-cascade-ticket-0", "cancel-cascade-key-0001", 1, now.Add(time.Second)) + if err != nil { + t.Fatalf("cancel: %v", err) + } + if cancelled.State != domain.Cancelled { + t.Fatalf("ticket did not cancel: %+v", cancelled) + } + + var proposalState string + if err := db.QueryRow(`SELECT state FROM proposals WHERE proposal_id = 'cancel-cascade-proposal'`).Scan(&proposalState); err != nil { + t.Fatal(err) + } + if proposalState != "DECLINED" { + t.Fatalf("proposal state = %s, want DECLINED immediately, not left OPEN to time out", proposalState) + } + var stateB string + var expiresB time.Time + if err := db.QueryRow(`SELECT state, expires_at FROM queue_tickets WHERE ticket_id = 'cancel-cascade-ticket-1'`).Scan(&stateB, &expiresB); err != nil { + t.Fatal(err) + } + if stateB != "QUEUED" { + t.Fatalf("other participant's ticket state = %s, want QUEUED immediately", stateB) + } + if !expiresB.After(now.Add(time.Second)) { + t.Fatalf("requeued ticket expiry %v was not refreshed forward", expiresB) + } + // The cancelling player's own ticket must stay CANCELLED, not get swept + // back up into the requeue meant for the other participant. + var stateA string + if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'cancel-cascade-ticket-0'`).Scan(&stateA); err != nil { + t.Fatal(err) + } + if stateA != "CANCELLED" { + t.Fatalf("cancelling player's own ticket state = %s, want it to stay CANCELLED", stateA) + } +} + 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 fb2b21dc..7103f446 100644 --- a/server/store/proposal_recovery_sql.go +++ b/server/store/proposal_recovery_sql.go @@ -37,6 +37,35 @@ 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 OpenProposalForCancelledTicketSQL = `SELECT pp.proposal_id +FROM proposal_participants pp +JOIN proposals p ON p.proposal_id = pp.proposal_id +WHERE pp.ticket_id = $1 AND pp.player_id = $2 AND p.state = 'OPEN'` + +// CascadeCancelToOpenProposal declines and requeues an OPEN proposal +// immediately when one of its participants cancels their own queue ticket +// directly, rather than leaving every other participant to wait out the +// full response window for something the system already knows can't happen +// -- ProposalExpireRequeueSQL would eventually rescue them anyway, but not +// for up to ProposalWindow's full duration for no reason. Must run inside +// the same transaction as the ticket cancel itself; a no-op if the ticket +// wasn't part of any currently-OPEN proposal. +func CascadeCancelToOpenProposal(ctx context.Context, tx *sql.Tx, ticketID, playerID string, now time.Time) error { + var proposalID string + err := tx.QueryRowContext(ctx, OpenProposalForCancelledTicketSQL, ticketID, playerID).Scan(&proposalID) + if err == sql.ErrNoRows { + return nil + } + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, ProposalDeclineSQL, proposalID); err != nil { + return err + } + _, err = tx.ExecContext(ctx, ProposalDeclineRequeueSQL, proposalID, now.Add(domain.QueueExpiryWindow)) + return err +} + const ProposalRecoverySelectSQL = `SELECT proposal_id, playlist, state, revision, expires_at FROM proposals WHERE proposal_id = $1 diff --git a/server/store/proposal_recovery_sql_test.go b/server/store/proposal_recovery_sql_test.go index 981d9620..b0b9028c 100644 --- a/server/store/proposal_recovery_sql_test.go +++ b/server/store/proposal_recovery_sql_test.go @@ -18,6 +18,7 @@ func TestProposalRecoverySQLBindsParticipantAndExpiresAtReadBoundary(t *testing. ProposalRevisionBumpSQL: {"revision = revision + 1", "state = 'OPEN'"}, ProposalDeclineRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "proposal_participants"}, ProposalExpireRequeueSQL: {"SET state = 'QUEUED'", "state = 'PROPOSED'", "state = 'EXPIRED'"}, + OpenProposalForCancelledTicketSQL: {"proposal_participants", "state = 'OPEN'"}, } { for _, fragment := range fragments { if !contains(query, fragment) { diff --git a/server/store/queue_sql.go b/server/store/queue_sql.go index dbc0152c..116a2ec7 100644 --- a/server/store/queue_sql.go +++ b/server/store/queue_sql.go @@ -275,6 +275,11 @@ func mutateQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idem return fmt.Errorf("decode queue RTT: %w", err) } ticket = queueTicketRecordToDomain(record) + if operation == "cancel" { + if err := CascadeCancelToOpenProposal(ctx, tx, ticketID, playerID, now); err != nil { + return err + } + } stored, err := json.Marshal(record) if err != nil { return err