fix(multiplayer): fence client queue cancellation states

server/domain/queue.go

@@ -131,6 +131,9 @@ func (q *Queue) Cancel(playerID, ticketID, idempotencyKey string, expectedRevisi
 	if ticket.Revision != expectedRevision {
 		return QueueTicket{}, ErrStaleRevision
 	}
+	if ticket.State != Queued && ticket.State != Proposed {
+		return QueueTicket{}, fmt.Errorf("%w: cancel in %s", ErrConflict, ticket.State)
+	}
 	if idempotencyKey == "" {
 		return QueueTicket{}, fmt.Errorf("%w: empty cancel key", ErrConflict)
 	}

server/domain/queue_test.go

@@ -50,6 +50,24 @@ func TestQueueHeartbeatExtendsExpiryExactlyAndRejectsStaleReplay(t *testing.T) {
 	}
 }
 
+func TestQueueCancelCannotOverrideMatchOwnedLifecycle(t *testing.T) {
+	q := NewQueue()
+	now := time.Unix(1000, 0)
+	candidate := Candidate{TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: now}
+	if _, err := q.Create("player-a", "ticket-a", "create-key-123456", candidate, now); err != nil {
+		t.Fatal(err)
+	}
+	ticket := q.tickets["ticket-a"]
+	ticket.State = Live
+	q.tickets["ticket-a"] = ticket
+	if _, err := q.Cancel("player-a", "ticket-a", "cancel-key-123456", 0, now.Add(time.Second)); !errors.Is(err, ErrConflict) {
+		t.Fatalf("live ticket cancellation error = %v, want conflict", err)
+	}
+	if got := q.tickets["ticket-a"].State; got != Live {
+		t.Fatalf("live ticket state = %s after cancellation attempt", got)
+	}
+}
+
 func TestQueueExpiryReleasesOwnershipAndDoesNotReturnExpiredCandidates(t *testing.T) {
 	q := NewQueue()
 	now := time.Unix(1000, 0)

server/store/postgres_integration_test.go

@@ -388,8 +388,8 @@ func TestPostgreSQLQueueHeartbeatAndCancelAreRevisionFenced(t *testing.T) {
 	if heartbeat.Revision != 1 || !heartbeat.ExpiresAt.Equal(now.Add(35*time.Second)) {
 		t.Fatalf("unexpected heartbeat result: %+v", heartbeat)
 	}
-	if _, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000002", 0, now.Add(6*time.Second)); err == nil {
-		t.Fatal("stale heartbeat revision was accepted")
+	if _, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000002", 0, now.Add(6*time.Second)); !errors.Is(err, domain.ErrStaleRevision) {
+		t.Fatalf("stale heartbeat error = %v, want ErrStaleRevision", err)
 	}
 	cancelled, err := CancelQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000003", 1, now.Add(7*time.Second))
 	if err != nil {
@@ -398,6 +398,22 @@ func TestPostgreSQLQueueHeartbeatAndCancelAreRevisionFenced(t *testing.T) {
 	if cancelled.State != domain.Cancelled || cancelled.Revision != 2 {
 		t.Fatalf("unexpected cancellation result: %+v", cancelled)
 	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('live-cancel-player', 'live-cancel-steam')`); 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 ('live-cancel-ticket', 'live-cancel-player', 'ranked', 'LIVE', 'integration-build', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := CancelQueueTicket(ctx, db, "live-cancel-player", "live-cancel-ticket", "live-cancel-op-0001", 0, now.Add(8*time.Second)); !errors.Is(err, domain.ErrConflict) {
+		t.Fatalf("live cancellation error = %v, want ErrConflict", err)
+	}
+	var liveState string
+	if err := db.QueryRow(`SELECT state FROM queue_tickets WHERE ticket_id = 'live-cancel-ticket'`).Scan(&liveState); err != nil {
+		t.Fatal(err)
+	}
+	if liveState != "LIVE" {
+		t.Fatalf("live ticket state = %s after cancellation attempt", liveState)
+	}
 }
 
 // TestPostgreSQLConcurrentQueueHeartbeatIsRevisionFencedUnderRealRace is the

server/store/queue_sql.go

@@ -6,6 +6,7 @@ import (
 	"crypto/sha256"
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
@@ -42,8 +43,12 @@ RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version,
 	QueueTicketCancelSQL = `UPDATE queue_tickets SET state = 'CANCELLED',
     revision = revision + 1, expires_at = $4
 WHERE ticket_id = $1 AND player_id = $2 AND revision = $3
-  AND state NOT IN ('COMPLETED', 'CANCELLED', 'EXPIRED', 'FAILED')
+  AND state IN ('QUEUED', 'PROPOSED')
 RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at, revision, predicted_rtt`
+	QueueMutationFailureSQL = `SELECT player_id, state, revision, expires_at
+FROM queue_tickets
+WHERE ticket_id = $1
+FOR UPDATE`
 	QueueCooldownSelectSQL = `SELECT ends_at
 FROM penalties
 WHERE player_id = $1 AND playlist = $2
@@ -294,7 +299,10 @@ func mutateQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idem
 		var record queueTicketRecord
 		var predictedRTT []byte
 		if err := tx.QueryRowContext(ctx, mutationSQL, ticketID, playerID, expectedRevision, now).Scan(&record.TicketID, &record.PlayerID, &record.Playlist, &record.State, &record.ClientBuild, &record.ProtocolVersion, &record.EnqueuedAt, &record.ExpiresAt, &record.Revision, &predictedRTT); err != nil {
-			return fmt.Errorf("queue mutation rejected: %w", err)
+			if !errors.Is(err, sql.ErrNoRows) {
+				return fmt.Errorf("queue mutation rejected: %w", err)
+			}
+			return classifyQueueMutationFailure(ctx, tx, playerID, ticketID, expectedRevision, now)
 		}
 		if err := json.Unmarshal(predictedRTT, &record.PredictedRTT); err != nil {
 			return fmt.Errorf("decode queue RTT: %w", err)
@@ -315,6 +323,29 @@ func mutateQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idem
 	return ticket, err
 }
 
+func classifyQueueMutationFailure(ctx context.Context, tx *sql.Tx, playerID, ticketID string, expectedRevision uint64, now time.Time) error {
+	var owner, state string
+	var revision uint64
+	var expiresAt time.Time
+	err := tx.QueryRowContext(ctx, QueueMutationFailureSQL, ticketID).Scan(&owner, &state, &revision, &expiresAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return domain.ErrTicketNotFound
+	}
+	if err != nil {
+		return err
+	}
+	if owner != playerID {
+		return domain.ErrNotTicketOwner
+	}
+	if (state == string(domain.Queued) || state == string(domain.Proposed)) && !now.Before(expiresAt) {
+		return domain.ErrTicketExpired
+	}
+	if revision != expectedRevision {
+		return domain.ErrStaleRevision
+	}
+	return fmt.Errorf("%w: %s in %s", domain.ErrConflict, "queue mutation", state)
+}
+
 func queueTicketRecordFromDomain(ticket domain.QueueTicket) queueTicketRecord {
 	return queueTicketRecord{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, ProposalID: ticket.ProposalID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), ClientBuild: ticket.Candidate.ClientBuild, ProtocolVersion: ticket.Candidate.ProtocolVersion, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt, Revision: ticket.Revision, PredictedRTT: ticket.Candidate.PredictedRTT}
 }

server/store/queue_sql_test.go

@@ -13,7 +13,8 @@ func TestQueueSQLUsesDurableIdempotencyAndOwnerScopedRecovery(t *testing.T) {
 		QueueTicketSelectSQL:         {"q.ticket_id = $1", "q.player_id = $2", "proposal_participants", "p.state = 'OPEN'", "match_participants", "participation_active"},
 		QueueTicketInsertSQL:         {"player_id", "playlist", "client_build", "protocol_version"},
 		QueueTicketHeartbeatSQL:      {"player_id = $2", "revision = $3", "expires_at > $4", "RETURNING"},
-		QueueTicketCancelSQL:         {"player_id = $2", "revision = $3", "state NOT IN", "RETURNING"},
+		QueueTicketCancelSQL:         {"player_id = $2", "revision = $3", "state IN ('QUEUED', 'PROPOSED')", "RETURNING"},
+		QueueMutationFailureSQL:      {"ticket_id = $1", "state", "revision", "expires_at", "FOR UPDATE"},
 		QueueCandidateProjectionSQL:  {"playlist = $1", "predicted_rtt", "expires_at > $2", "LIMIT $3"},
 		RankedParticipantSQL:         {"steam_id", "player_id = ANY($1)", "ORDER BY player_id"},
 		ProposalInsertSQL:            {"match_region", "match_protocol", "NULLIF($4, '')"},