Files
CosmicClash/server/store/redis_integration_test.go
Josh Creek 320ec46ba2 fix(server): partition and bound the Redis candidate projection
Both playlists shared one hash and sorted set, causing two independent
failures.

Starvation: Snapshot performed an unbounded ZRANGEBYSCORE and HMGET,
decoded the whole queue, and the matcher then truncated to its candidate
limit *before* filtering by playlist. A large casual prefix could
therefore leave the ranked worker with zero candidates indefinitely even
while ranked tickets were queued further down the set.

Mutual erasure: each matcher captured only its own playlist as the
durable source, but Rebuild replaced the shared keys, so a casual repair
wiped ranked projections and vice versa.

Namespace the keys per playlist, push the limit into Redis (LIMIT 0 N)
so reads no longer scale with total queue depth, and scope Rebuild to
one namespace. Rebuild now rejects a candidate whose playlist does not
match the namespace, which would reintroduce the starvation. Upsert
derives the namespace from the candidate; Remove takes the playlist,
since a ticket ID alone no longer identifies its namespace.

Add tests for a 300-deep casual backlog not starving ranked, for neither
playlist's rebuild erasing the other, and for the limit being applied
without losing enqueue ordering.
2026-09-05 10:23:52 +01:00

148 lines
5.7 KiB
Go

//go:build integration
package store
import (
"context"
"os"
"testing"
"time"
"github.com/cosmic-clash/cosmic-clash/server/domain"
"github.com/redis/go-redis/v9"
)
// This binary is deliberately opt-in, mirroring postgres_integration_test.go:
// it requires a disposable real Redis supplied by
// scripts/run_redis_integration.sh, as distinct from the miniredis-backed
// unit tests in redis_candidates_test.go and candidate_projection_test.go.
// miniredis is a from-scratch Go reimplementation of the Redis command set --
// it does not run real Redis's own float64 score encoding, real TTL/expiry,
// or real RESP wire behavior, so it cannot by itself prove this code works
// against the real thing, only that it works against a same-language model of
// it.
func openIntegrationRedis(t *testing.T) *redis.Client {
t.Helper()
addr := os.Getenv("COSMIC_CLASH_REDIS_ADDR")
if addr == "" {
t.Skip("COSMIC_CLASH_REDIS_ADDR is not set")
}
client := redis.NewClient(&redis.Options{Addr: addr})
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := client.Ping(ctx).Err(); err != nil {
client.Close()
t.Fatalf("ping Redis: %v", err)
}
if err := client.FlushAll(ctx).Err(); err != nil {
client.Close()
t.Fatalf("reset Redis: %v", err)
}
t.Cleanup(func() { client.Close() })
return client
}
func TestRealRedisCandidateIndexUpsertSnapshotRemove(t *testing.T) {
client := openIntegrationRedis(t)
ctx := context.Background()
index := RedisCandidateIndex{Client: client, Prefix: "integration-real", TTL: time.Minute}
now := time.Now().UTC().Truncate(time.Microsecond)
a := domain.Candidate{TicketID: "real-ticket-a", PlayerID: "real-player-a", Playlist: domain.Casual, ClientBuild: "build-1", ProtocolVersion: 1, EnqueuedAt: now}
b := domain.Candidate{TicketID: "real-ticket-b", PlayerID: "real-player-b", Playlist: domain.Casual, ClientBuild: "build-1", ProtocolVersion: 1, EnqueuedAt: now.Add(time.Second)}
if err := index.Upsert(ctx, a); err != nil {
t.Fatalf("upsert a: %v", err)
}
if err := index.Upsert(ctx, b); err != nil {
t.Fatalf("upsert b: %v", err)
}
got, err := index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000)
if err != nil {
t.Fatalf("snapshot: %v", err)
}
if len(got) != 2 || got[0].TicketID != "real-ticket-a" || got[1].TicketID != "real-ticket-b" {
t.Fatalf("snapshot after upsert = %+v", got)
}
if err := index.Remove(ctx, domain.Casual, "real-ticket-a"); err != nil {
t.Fatalf("remove: %v", err)
}
got, err = index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000)
if err != nil {
t.Fatalf("snapshot after remove: %v", err)
}
if len(got) != 1 || got[0].TicketID != "real-ticket-b" {
t.Fatalf("snapshot after remove = %+v", got)
}
// A real TTL, actually waited out, not miniredis's manual FastForward.
shortLived := RedisCandidateIndex{Client: client, Prefix: "integration-real-ttl", TTL: 1500 * time.Millisecond}
if err := shortLived.Upsert(ctx, domain.Candidate{Playlist: domain.Casual, TicketID: "real-ticket-ttl", PlayerID: "real-player-ttl", EnqueuedAt: now}); err != nil {
t.Fatalf("upsert ttl candidate: %v", err)
}
time.Sleep(2 * time.Second)
got, err = shortLived.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000)
if err != nil {
t.Fatalf("snapshot after real TTL expiry: %v", err)
}
if len(got) != 0 {
t.Fatalf("candidate survived its real TTL: %+v", got)
}
}
// TestRealRedisCandidateProjectionRepairsAfterFlush proves the documented
// "Redis restart or lost keyspace" repair path against an actual data loss
// event on a real server -- FLUSHALL -- not a simulated empty map.
func TestRealRedisCandidateProjectionRepairsAfterFlush(t *testing.T) {
client := openIntegrationRedis(t)
ctx := context.Background()
index := RedisCandidateIndex{Client: client, Prefix: "integration-real-repair", TTL: time.Minute}
now := time.Now().UTC().Truncate(time.Microsecond)
durable := []domain.Candidate{
{Playlist: domain.Casual, TicketID: "repair-ticket-a", PlayerID: "repair-player-a", EnqueuedAt: now},
{Playlist: domain.Casual, TicketID: "repair-ticket-b", PlayerID: "repair-player-b", EnqueuedAt: now.Add(time.Second)},
}
sourceCalls := 0
projection := CandidateProjection{Index: index, Source: func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) {
sourceCalls++
return durable, nil
}}
if err := index.Upsert(ctx, durable[0]); err != nil {
t.Fatalf("seed upsert: %v", err)
}
// Simulate the actual failure mode this path exists for: the whole Redis
// instance loses its data (restart without persistence, failover to an
// empty replica, an operator FLUSHALL) mid-operation, not just "this one
// key expired".
if err := client.FlushAll(ctx).Err(); err != nil {
t.Fatalf("flush: %v", err)
}
got, err := projection.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000)
if err != nil {
t.Fatalf("snapshot after flush: %v", err)
}
if sourceCalls != 1 {
t.Fatalf("expected exactly one durable repair call, got %d", sourceCalls)
}
if len(got) != 2 || got[0].TicketID != "repair-ticket-a" || got[1].TicketID != "repair-ticket-b" {
t.Fatalf("snapshot after repair = %+v", got)
}
// The repair must actually have written back to Redis, not just returned
// the durable source's answer in memory -- confirm a second snapshot
// (Redis not flushed again) reads it back without a second Source call.
got, err = index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000)
if err != nil {
t.Fatalf("snapshot directly against Redis after repair: %v", err)
}
if len(got) != 2 {
t.Fatalf("repaired data was not actually persisted to Redis: %+v", got)
}
if sourceCalls != 1 {
t.Fatalf("expected repair to persist so a second read needs no further Source call, got %d calls", sourceCalls)
}
}