diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 371cb4ca..bfe3c2b4 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -346,6 +346,66 @@ func TestPostgreSQLQueueHeartbeatAndCancelAreRevisionFenced(t *testing.T) { } } +// TestPostgreSQLConcurrentQueueHeartbeatIsRevisionFencedUnderRealRace is the +// live counterpart to the sequential stale-heartbeat check above: calling the +// second heartbeat only after the first has already committed proves the SQL +// predicate is correct, but not that it actually fences two requests that +// genuinely overlap at the database. A client can legitimately double-send a +// heartbeat (a slow response triggering a client-side retry, or two tabs/ +// processes for the same player), and both requests can reach PostgreSQL +// truly concurrently -- this races that directly. +func TestPostgreSQLConcurrentQueueHeartbeatIsRevisionFencedUnderRealRace(t *testing.T) { + db := openIntegrationPostgres(t) + applyIntegrationMigrations(t, db) + now := time.Now().UTC().Truncate(time.Microsecond) + ctx := context.Background() + if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('race-heartbeat-player', 'race-heartbeat-steam')`); err != nil { + t.Fatal(err) + } + spec := domain.QueueSpec{Playlist: domain.Casual, ClientBuild: "integration-build", ProtocolVersion: 1} + if _, err := CreateQueueTicket(ctx, db, "race-heartbeat-ticket", "race-heartbeat-player", "race-heartbeat-create-01", spec, now); err != nil { + t.Fatal(err) + } + + const attempts = 5 + var wg sync.WaitGroup + tickets := make([]domain.QueueTicket, attempts) + errs := make([]error, attempts) + wg.Add(attempts) + for i := 0; i < attempts; i++ { + go func(i int) { + defer wg.Done() + tickets[i], errs[i] = HeartbeatQueueTicket(ctx, db, "race-heartbeat-player", "race-heartbeat-ticket", fmt.Sprintf("race-heartbeat-op-%08d", i), 0, now.Add(time.Duration(i)*time.Millisecond)) + }(i) + } + wg.Wait() + + won, lost := 0, 0 + for i, err := range errs { + if err == nil { + won++ + if tickets[i].Revision != 1 { + t.Fatalf("winning heartbeat %d landed at revision %d, want 1", i, tickets[i].Revision) + } + continue + } + lost++ + } + if won != 1 { + t.Fatalf("won=%d, want exactly 1 of %d concurrent heartbeats at the same expected revision to win", won, attempts) + } + if lost != attempts-1 { + t.Fatalf("lost=%d, want %d", lost, attempts-1) + } + var revision uint64 + if err := db.QueryRow(`SELECT revision FROM queue_tickets WHERE ticket_id = 'race-heartbeat-ticket'`).Scan(&revision); err != nil { + t.Fatal(err) + } + if revision != 1 { + t.Fatalf("durable revision = %d, want exactly 1 (a stale winner re-applying would leave it higher)", revision) + } +} + func TestPostgreSQLAssignmentPersistenceIsPlayerScopedAndExpiryBound(t *testing.T) { db := openIntegrationPostgres(t) applyIntegrationMigrations(t, db)