From 320ec46ba2c902f18b01622e46b169af35ba7f10 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Sat, 5 Sep 2026 10:23:52 +0100 Subject: [PATCH] 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. --- server/api/service.go | 10 +- server/api/service_test.go | 2 +- server/cmd/matcher/main.go | 24 ++-- server/store/candidate_projection_test.go | 28 ++--- server/store/redis_candidates.go | 78 +++++++++---- server/store/redis_candidates_test.go | 128 ++++++++++++++++++++-- server/store/redis_integration_test.go | 20 ++-- 7 files changed, 215 insertions(+), 75 deletions(-) diff --git a/server/api/service.go b/server/api/service.go index 361a459c..cb08ffc1 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -59,7 +59,9 @@ type QueueBackend interface { // failures must never change the result of an already successful mutation. type CandidateIndex interface { Upsert(context.Context, domain.Candidate) error - Remove(context.Context, string) error + // Remove is playlist-scoped because the projection is partitioned per + // playlist; a ticket ID alone does not identify its namespace. + Remove(context.Context, domain.Playlist, string) error } type SessionBackend interface { @@ -486,9 +488,9 @@ func (s *Service) projectCandidate(ctx context.Context, ticket domain.QueueTicke } } -func (s *Service) removeCandidate(ctx context.Context, ticketID string) { +func (s *Service) removeCandidate(ctx context.Context, playlist domain.Playlist, ticketID string) { if s.CandidateIndex != nil { - _ = s.CandidateIndex.Remove(ctx, ticketID) + _ = s.CandidateIndex.Remove(ctx, playlist, ticketID) } } @@ -894,7 +896,7 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) { } s.logQueueOutcome(eventName, ticketID, ticket, nil, now) if ticket.State == domain.Cancelled { - s.removeCandidate(r.Context(), ticket.TicketID) + s.removeCandidate(r.Context(), ticket.Playlist, ticket.TicketID) } else { s.projectCandidate(r.Context(), ticket) } diff --git a/server/api/service_test.go b/server/api/service_test.go index f92c837d..66001914 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -135,7 +135,7 @@ func (i *candidateIndexSpy) Upsert(_ context.Context, candidate domain.Candidate return i.upsertErr } -func (i *candidateIndexSpy) Remove(_ context.Context, _ string) error { +func (i *candidateIndexSpy) Remove(_ context.Context, _ domain.Playlist, _ string) error { i.removeCalls++ return i.removeErr } diff --git a/server/cmd/matcher/main.go b/server/cmd/matcher/main.go index d8034dc3..668ce7ff 100644 --- a/server/cmd/matcher/main.go +++ b/server/cmd/matcher/main.go @@ -66,29 +66,21 @@ func main() { defer redisClient.Close() candidateProjection := store.CandidateProjection{ Index: store.RedisCandidateIndex{Client: redisClient, Prefix: *redisPrefix, TTL: *redisTTL}, - Source: func(ctx context.Context, at time.Time) ([]domain.Candidate, error) { - return store.ListQueuedCandidates(ctx, db, selectedPlaylist, at, 1000) + Source: func(ctx context.Context, playlist domain.Playlist, at time.Time, limit int) ([]domain.Candidate, error) { + return store.ListQueuedCandidates(ctx, db, playlist, at, limit) }, } projection = &candidateProjection } worker := matcher.Worker{ + // Both branches are now playlist-filtered and limit-bounded at the + // source. The Redis branch previously read the whole shared queue, + // truncated it to limit, and only then filtered by playlist -- so a + // large casual prefix could leave the ranked worker with zero + // candidates indefinitely even while ranked tickets were queued. Source: func(ctx context.Context, at time.Time, playlist domain.Playlist, limit int) ([]domain.Candidate, error) { if projection != nil { - candidates, err := projection.Snapshot(ctx, at) - if err != nil { - return nil, err - } - if len(candidates) > limit { - candidates = candidates[:limit] - } - filtered := make([]domain.Candidate, 0, len(candidates)) - for _, candidate := range candidates { - if candidate.Playlist == playlist { - filtered = append(filtered, candidate) - } - } - return filtered, nil + return projection.Snapshot(ctx, playlist, at, limit) } return store.ListQueuedCandidates(ctx, db, playlist, at, limit) }, diff --git a/server/store/candidate_projection_test.go b/server/store/candidate_projection_test.go index 6e35e4a8..b822f6c6 100644 --- a/server/store/candidate_projection_test.go +++ b/server/store/candidate_projection_test.go @@ -19,16 +19,16 @@ func TestCandidateProjectionRepairsPartialRedisStateFromDurableSource(t *testing client := redis.NewClient(&redis.Options{Addr: mini.Addr()}) defer client.Close() now := time.Unix(1000, 0).UTC() - candidate := domain.Candidate{TicketID: "repair-ticket", PlayerID: "repair-player", EnqueuedAt: now} + candidate := domain.Candidate{Playlist: domain.Casual, TicketID: "repair-ticket", PlayerID: "repair-player", EnqueuedAt: now} index := RedisCandidateIndex{Client: client, Prefix: "repair", TTL: time.Minute} - _, orderKey := index.keys() + _, orderKey := index.keys(domain.Casual) if err := client.ZAdd(context.Background(), orderKey, redis.Z{Score: float64(now.UnixNano()), Member: candidate.TicketID}).Err(); err != nil { t.Fatal(err) } - projection := CandidateProjection{Index: index, Source: func(context.Context, time.Time) ([]domain.Candidate, error) { + projection := CandidateProjection{Index: index, Source: func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) { return []domain.Candidate{candidate}, nil }} - got, err := projection.Snapshot(context.Background(), now) + got, err := projection.Snapshot(context.Background(), domain.Casual, now, 1000) if err != nil { t.Fatal(err) } @@ -39,10 +39,10 @@ func TestCandidateProjectionRepairsPartialRedisStateFromDurableSource(t *testing func TestCandidateProjectionDoesNotReturnCacheWhenRepairSourceFails(t *testing.T) { index := RedisCandidateIndex{TTL: time.Minute} - projection := CandidateProjection{Index: index, Source: func(context.Context, time.Time) ([]domain.Candidate, error) { + projection := CandidateProjection{Index: index, Source: func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) { return nil, context.DeadlineExceeded }} - if _, err := projection.Snapshot(context.Background(), time.Unix(1000, 0)); err == nil { + if _, err := projection.Snapshot(context.Background(), domain.Casual, time.Unix(1000, 0), 1000); err == nil { t.Fatal("cache projection succeeded without a usable Redis/index source") } } @@ -67,16 +67,16 @@ func TestCandidateProjectionFallsBackToSourceWhenRedisIsEntirelyUnreachable(t *t mini.Close() // Redis is now entirely unreachable, not merely empty or stale. now := time.Unix(1000, 0).UTC() - candidate := domain.Candidate{TicketID: "down-ticket", PlayerID: "down-player", EnqueuedAt: now} + candidate := domain.Candidate{Playlist: domain.Casual, TicketID: "down-ticket", PlayerID: "down-player", EnqueuedAt: now} sourceCalls := 0 projection := CandidateProjection{ Index: RedisCandidateIndex{Client: client, Prefix: "down", TTL: time.Minute}, - Source: func(context.Context, time.Time) ([]domain.Candidate, error) { + Source: func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) { sourceCalls++ return []domain.Candidate{candidate}, nil }, } - got, err := projection.Snapshot(context.Background(), now) + got, err := projection.Snapshot(context.Background(), domain.Casual, now, 1000) if err != nil { t.Fatalf("Snapshot failed while Redis was down, even though Source (PostgreSQL) was healthy: %v", err) } @@ -102,11 +102,11 @@ func TestCandidateProjectionStillFailsWhenBothRedisAndSourceAreDown(t *testing.T projection := CandidateProjection{ Index: RedisCandidateIndex{Client: client, Prefix: "down", TTL: time.Minute}, - Source: func(context.Context, time.Time) ([]domain.Candidate, error) { + Source: func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) { return nil, context.DeadlineExceeded }, } - if _, err := projection.Snapshot(context.Background(), time.Unix(1000, 0)); err == nil { + if _, err := projection.Snapshot(context.Background(), domain.Casual, time.Unix(1000, 0), 1000); err == nil { t.Fatal("Snapshot succeeded with both Redis and the durable source unavailable") } } @@ -120,14 +120,14 @@ func TestCandidateProjectionRepairsEmptyIndexFromDurableSource(t *testing.T) { client := redis.NewClient(&redis.Options{Addr: mini.Addr()}) defer client.Close() now := time.Unix(1000, 0).UTC() - candidate := domain.Candidate{TicketID: "miss-ticket", PlayerID: "miss-player", EnqueuedAt: now} + candidate := domain.Candidate{Playlist: domain.Casual, TicketID: "miss-ticket", PlayerID: "miss-player", EnqueuedAt: now} projection := CandidateProjection{ Index: RedisCandidateIndex{Client: client, Prefix: "miss", TTL: time.Minute}, - Source: func(context.Context, time.Time) ([]domain.Candidate, error) { + Source: func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) { return []domain.Candidate{candidate}, nil }, } - got, err := projection.Snapshot(context.Background(), now) + got, err := projection.Snapshot(context.Background(), domain.Casual, now, 1000) if err != nil { t.Fatal(err) } diff --git a/server/store/redis_candidates.go b/server/store/redis_candidates.go index f99eb5db..22094f94 100644 --- a/server/store/redis_candidates.go +++ b/server/store/redis_candidates.go @@ -21,8 +21,9 @@ type RedisCandidateIndex struct { // DurableCandidateSource is the authoritative queue projection used to // repair Redis. Implementations must apply queue state and expiry rules before -// returning candidates. -type DurableCandidateSource func(context.Context, time.Time) ([]domain.Candidate, error) +// returning candidates. It is playlist- and limit-scoped so a repair reads +// only the namespace it is about to rebuild, and never an unbounded queue. +type DurableCandidateSource func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) // CandidateProjection couples the transient index to its durable repair // source. A cache miss, partial write, malformed payload, or Redis restart is @@ -32,15 +33,15 @@ type CandidateProjection struct { Source DurableCandidateSource } -func (p CandidateProjection) Repair(ctx context.Context, now time.Time) error { +func (p CandidateProjection) Repair(ctx context.Context, playlist domain.Playlist, now time.Time, limit int) error { if p.Source == nil || now.IsZero() { return fmt.Errorf("invalid candidate repair source") } - candidates, err := p.Source(ctx, now) + candidates, err := p.Source(ctx, playlist, now, limit) if err != nil { return err } - return p.Index.Rebuild(ctx, candidates) + return p.Index.Rebuild(ctx, playlist, candidates) } // Snapshot never fails just because Redis specifically is unreachable. @@ -58,31 +59,47 @@ func (p CandidateProjection) Repair(ctx context.Context, now time.Time) error { // directly, and only attempt to repopulate Redis on a best-effort basis -- // its outcome is deliberately ignored, since a caller must never be denied // service just because the rebuild's own Redis write also failed. -func (p CandidateProjection) Snapshot(ctx context.Context, now time.Time) ([]domain.Candidate, error) { +func (p CandidateProjection) Snapshot(ctx context.Context, playlist domain.Playlist, now time.Time, limit int) ([]domain.Candidate, error) { if p.Source == nil { return nil, fmt.Errorf("invalid candidate repair source") } - candidates, err := p.Index.Snapshot(ctx, now) + candidates, err := p.Index.Snapshot(ctx, playlist, now, limit) if err == nil && len(candidates) > 0 { return candidates, nil } // Either the index errored outright, or came back empty -- indistinguishable // from a Redis restart or a lost keyspace. Consult PostgreSQL, the // authoritative source, either way. - source, sourceErr := p.Source(ctx, now) + source, sourceErr := p.Source(ctx, playlist, now, limit) if sourceErr != nil { return nil, sourceErr } - _ = p.Index.Rebuild(ctx, source) + // Rebuild only this playlist's namespace. When the keys were shared, a + // casual repair replaced the keys ranked candidates lived in and vice + // versa, so each worker could erase the other's projection. + _ = p.Index.Rebuild(ctx, playlist, source) return source, nil } -func (r RedisCandidateIndex) keys() (string, string) { +// keys are namespaced per playlist. They used to be shared, which caused two +// independent failures: the matcher truncated a mixed snapshot to its +// candidate limit before filtering by playlist, so a large casual backlog +// could starve the ranked worker indefinitely; and Rebuild replaced the shared +// keys, so one playlist's repair erased the other's projection. +func (r RedisCandidateIndex) keys(playlist domain.Playlist) (string, string) { prefix := r.Prefix if prefix == "" { prefix = "cosmic-clash" } - return prefix + ":queue:candidates:data", prefix + ":queue:candidates:order" + base := prefix + ":queue:candidates:" + string(playlist) + return base + ":data", base + ":order" +} + +func validRedisPlaylist(playlist domain.Playlist) error { + if playlist != domain.Casual && playlist != domain.Ranked { + return fmt.Errorf("invalid candidate playlist %q", playlist) + } + return nil } func (r RedisCandidateIndex) validate() error { @@ -109,11 +126,14 @@ func (r RedisCandidateIndex) Upsert(ctx context.Context, candidate domain.Candid if err := validateRedisCandidate(candidate); err != nil { return err } + if err := validRedisPlaylist(candidate.Playlist); err != nil { + return err + } payload, err := json.Marshal(candidate) if err != nil { return err } - dataKey, orderKey := r.keys() + dataKey, orderKey := r.keys(candidate.Playlist) pipe := r.Client.TxPipeline() pipe.HSet(ctx, dataKey, candidate.TicketID, payload) pipe.ZAdd(ctx, orderKey, redis.Z{Score: float64(candidate.EnqueuedAt.UnixNano()), Member: candidate.TicketID}) @@ -123,14 +143,17 @@ func (r RedisCandidateIndex) Upsert(ctx context.Context, candidate domain.Candid return err } -func (r RedisCandidateIndex) Remove(ctx context.Context, ticketID string) error { +func (r RedisCandidateIndex) Remove(ctx context.Context, playlist domain.Playlist, ticketID string) error { if err := r.validate(); err != nil { return err } + if err := validRedisPlaylist(playlist); err != nil { + return err + } if ticketID == "" { return fmt.Errorf("ticket ID is required") } - dataKey, orderKey := r.keys() + dataKey, orderKey := r.keys(playlist) pipe := r.Client.TxPipeline() pipe.HDel(ctx, dataKey, ticketID) pipe.ZRem(ctx, orderKey, ticketID) @@ -141,16 +164,25 @@ func (r RedisCandidateIndex) Remove(ctx context.Context, ticketID string) error // Snapshot reads only candidates whose enqueue timestamp is not in the // future. Missing payloads are ignored; the durable rebuild path repairs such // partial cache state without allowing it to affect ownership. -func (r RedisCandidateIndex) Snapshot(ctx context.Context, now time.Time) ([]domain.Candidate, error) { +func (r RedisCandidateIndex) Snapshot(ctx context.Context, playlist domain.Playlist, now time.Time, limit int) ([]domain.Candidate, error) { if err := r.validate(); err != nil { return nil, err } + if err := validRedisPlaylist(playlist); err != nil { + return nil, err + } if now.IsZero() { return nil, fmt.Errorf("authoritative time is required") } - dataKey, orderKey := r.keys() + if limit < 1 || limit > 1000 { + return nil, fmt.Errorf("invalid candidate snapshot limit") + } + dataKey, orderKey := r.keys(playlist) + // The limit is applied by Redis (LIMIT 0 N), not after transfer. The + // unbounded range and HMGET decoded the entire queue on every one-second + // poll, allocating and transferring in proportion to total queue depth. tickets, err := r.Client.ZRangeByScore(ctx, orderKey, &redis.ZRangeBy{ - Min: "-inf", Max: fmt.Sprint(now.UnixNano()), + Min: "-inf", Max: fmt.Sprint(now.UnixNano()), Offset: 0, Count: int64(limit), }).Result() if err != nil { return nil, err @@ -191,10 +223,13 @@ func (r RedisCandidateIndex) Snapshot(ctx context.Context, now time.Time) ([]dom // Rebuild atomically replaces both Redis keys from the authoritative queue // projection. It is the required path after Redis restart/failover or cache // loss, and rejects duplicate ticket IDs before touching Redis. -func (r RedisCandidateIndex) Rebuild(ctx context.Context, candidates []domain.Candidate) error { +func (r RedisCandidateIndex) Rebuild(ctx context.Context, playlist domain.Playlist, candidates []domain.Candidate) error { if err := r.validate(); err != nil { return err } + if err := validRedisPlaylist(playlist); err != nil { + return err + } seen := make(map[string]struct{}, len(candidates)) values := make([]interface{}, 0, len(candidates)*2) scores := make([]redis.Z, 0, len(candidates)) @@ -202,6 +237,11 @@ func (r RedisCandidateIndex) Rebuild(ctx context.Context, candidates []domain.Ca if err := validateRedisCandidate(candidate); err != nil { return err } + if candidate.Playlist != playlist { + // A rebuild that mixed playlists would write foreign candidates + // into this namespace, reintroducing the starvation it fixes. + return fmt.Errorf("rebuild candidate %s is %q, not %q", candidate.TicketID, candidate.Playlist, playlist) + } if _, exists := seen[candidate.TicketID]; exists { return fmt.Errorf("duplicate candidate in rebuild") } @@ -213,7 +253,7 @@ func (r RedisCandidateIndex) Rebuild(ctx context.Context, candidates []domain.Ca values = append(values, candidate.TicketID, payload) scores = append(scores, redis.Z{Score: float64(candidate.EnqueuedAt.UnixNano()), Member: candidate.TicketID}) } - dataKey, orderKey := r.keys() + dataKey, orderKey := r.keys(playlist) pipe := r.Client.TxPipeline() pipe.Del(ctx, dataKey, orderKey) if len(values) > 0 { diff --git a/server/store/redis_candidates_test.go b/server/store/redis_candidates_test.go index da31a2a1..8fa19ae5 100644 --- a/server/store/redis_candidates_test.go +++ b/server/store/redis_candidates_test.go @@ -2,6 +2,7 @@ package store import ( "context" + "fmt" "testing" "time" @@ -21,37 +22,37 @@ func TestRedisCandidateIndexRebuildSnapshotAndRemove(t *testing.T) { index := RedisCandidateIndex{Client: client, Prefix: "integration", TTL: time.Minute} now := time.Unix(1000, 0).UTC() candidates := []domain.Candidate{ - {TicketID: "ticket-b", PlayerID: "player-b", EnqueuedAt: now.Add(time.Second)}, - {TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: now}, + {Playlist: domain.Casual, TicketID: "ticket-b", PlayerID: "player-b", EnqueuedAt: now.Add(time.Second)}, + {Playlist: domain.Casual, TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: now}, } - if err := index.Rebuild(context.Background(), candidates); err != nil { + if err := index.Rebuild(context.Background(), domain.Casual, candidates); err != nil { t.Fatal(err) } - got, err := index.Snapshot(context.Background(), now) + got, err := index.Snapshot(context.Background(), domain.Casual, now, 1000) if err != nil { t.Fatal(err) } if len(got) != 1 || got[0].TicketID != "ticket-a" { t.Fatalf("snapshot before future candidate = %+v", got) } - if err := index.Remove(context.Background(), "ticket-a"); err != nil { + if err := index.Remove(context.Background(), domain.Casual, "ticket-a"); err != nil { t.Fatal(err) } - got, err = index.Snapshot(context.Background(), now.Add(2*time.Second)) + got, err = index.Snapshot(context.Background(), domain.Casual, now.Add(2*time.Second), 1000) if err != nil { t.Fatal(err) } if len(got) != 1 || got[0].TicketID != "ticket-b" { t.Fatalf("snapshot after remove = %+v", got) } - if ttl, err := client.TTL(context.Background(), "integration:queue:candidates:data").Result(); err != nil || ttl <= 0 { + if ttl, err := client.TTL(context.Background(), "integration:queue:candidates:casual:data").Result(); err != nil || ttl <= 0 { t.Fatalf("candidate data TTL = %v, err = %v", ttl, err) } } func TestRedisCandidateIndexRejectsInvalidAndDuplicateRebuilds(t *testing.T) { index := RedisCandidateIndex{TTL: time.Minute} - if err := index.Rebuild(context.Background(), nil); err == nil { + if err := index.Rebuild(context.Background(), domain.Casual, nil); err == nil { t.Fatal("nil Redis client accepted") } mini, err := miniredis.Run() @@ -62,11 +63,116 @@ func TestRedisCandidateIndexRejectsInvalidAndDuplicateRebuilds(t *testing.T) { client := redis.NewClient(&redis.Options{Addr: mini.Addr()}) defer client.Close() index.Client = client - candidate := domain.Candidate{TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: time.Unix(1000, 0)} - if err := index.Rebuild(context.Background(), []domain.Candidate{candidate, candidate}); err == nil { + candidate := domain.Candidate{Playlist: domain.Casual, TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: time.Unix(1000, 0)} + if err := index.Rebuild(context.Background(), domain.Casual, []domain.Candidate{candidate, candidate}); err == nil { t.Fatal("duplicate candidate accepted") } - if err := index.Upsert(context.Background(), domain.Candidate{TicketID: "", PlayerID: "player-a", EnqueuedAt: candidate.EnqueuedAt}); err == nil { + if err := index.Upsert(context.Background(), domain.Candidate{Playlist: domain.Casual, TicketID: "", PlayerID: "player-a", EnqueuedAt: candidate.EnqueuedAt}); err == nil { t.Fatal("invalid candidate accepted") } } + +// Both playlists used to share one hash and sorted set. Two failures followed: +// the matcher truncated a mixed snapshot to its candidate limit before +// filtering by playlist, so a large casual prefix could leave the ranked +// worker with zero candidates indefinitely; and Rebuild replaced the shared +// keys, so a casual repair erased ranked projections and vice versa. +func TestRedisCandidateIndexIsolatesPlaylists(t *testing.T) { + mini, err := miniredis.Run() + if err != nil { + t.Fatal(err) + } + defer mini.Close() + client := redis.NewClient(&redis.Options{Addr: mini.Addr()}) + defer client.Close() + index := RedisCandidateIndex{Client: client, Prefix: "isolation", TTL: time.Minute} + ctx := context.Background() + now := time.Unix(1000, 0).UTC() + + // A large casual backlog enqueued strictly before the ranked tickets. With + // shared keys this prefix is exactly what starved the ranked worker. + casual := make([]domain.Candidate, 0, 300) + for i := 0; i < 300; i++ { + casual = append(casual, domain.Candidate{ + Playlist: domain.Casual, TicketID: fmt.Sprintf("casual-%03d", i), + PlayerID: fmt.Sprintf("casual-player-%03d", i), EnqueuedAt: now.Add(time.Duration(i) * time.Millisecond), + }) + } + ranked := []domain.Candidate{ + {Playlist: domain.Ranked, TicketID: "ranked-a", PlayerID: "ranked-player-a", EnqueuedAt: now.Add(time.Second)}, + {Playlist: domain.Ranked, TicketID: "ranked-b", PlayerID: "ranked-player-b", EnqueuedAt: now.Add(2 * time.Second)}, + } + if err := index.Rebuild(ctx, domain.Casual, casual); err != nil { + t.Fatalf("casual rebuild: %v", err) + } + if err := index.Rebuild(ctx, domain.Ranked, ranked); err != nil { + t.Fatalf("ranked rebuild: %v", err) + } + + // The casual rebuild must not have erased the ranked projection. + at := now.Add(time.Hour) + gotRanked, err := index.Snapshot(ctx, domain.Ranked, at, 50) + if err != nil { + t.Fatalf("ranked snapshot: %v", err) + } + if len(gotRanked) != 2 { + t.Fatalf("ranked worker saw %d candidates behind a 300-deep casual backlog, want 2", len(gotRanked)) + } + for _, candidate := range gotRanked { + if candidate.Playlist != domain.Ranked { + t.Fatalf("ranked snapshot leaked a %q candidate: %s", candidate.Playlist, candidate.TicketID) + } + } + + // A ranked repair must likewise leave casual alone. + if err := index.Rebuild(ctx, domain.Ranked, ranked[:1]); err != nil { + t.Fatalf("ranked re-repair: %v", err) + } + gotCasual, err := index.Snapshot(ctx, domain.Casual, at, 1000) + if err != nil { + t.Fatalf("casual snapshot: %v", err) + } + if len(gotCasual) != 300 { + t.Fatalf("ranked rebuild erased casual projection: %d remain", len(gotCasual)) + } +} + +// The limit must be applied by Redis, not after transfer: the old unbounded +// ZRANGEBYSCORE plus HMGET decoded the entire queue on every one-second poll. +func TestRedisCandidateIndexSnapshotIsBoundedByRedis(t *testing.T) { + mini, err := miniredis.Run() + if err != nil { + t.Fatal(err) + } + defer mini.Close() + client := redis.NewClient(&redis.Options{Addr: mini.Addr()}) + defer client.Close() + index := RedisCandidateIndex{Client: client, Prefix: "bounded", TTL: time.Minute} + ctx := context.Background() + now := time.Unix(1000, 0).UTC() + + candidates := make([]domain.Candidate, 0, 500) + for i := 0; i < 500; i++ { + candidates = append(candidates, domain.Candidate{ + Playlist: domain.Casual, TicketID: fmt.Sprintf("bulk-%03d", i), + PlayerID: fmt.Sprintf("bulk-player-%03d", i), EnqueuedAt: now.Add(time.Duration(i) * time.Millisecond), + }) + } + if err := index.Rebuild(ctx, domain.Casual, candidates); err != nil { + t.Fatal(err) + } + got, err := index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 10) + if err != nil { + t.Fatal(err) + } + if len(got) != 10 { + t.Fatalf("snapshot returned %d candidates for a limit of 10", len(got)) + } + // Oldest-first ordering must survive the bound. + if got[0].TicketID != "bulk-000" || got[9].TicketID != "bulk-009" { + t.Fatalf("bounded snapshot lost enqueue ordering: %s..%s", got[0].TicketID, got[9].TicketID) + } + if _, err := index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 0); err == nil { + t.Fatal("unbounded snapshot accepted") + } +} diff --git a/server/store/redis_integration_test.go b/server/store/redis_integration_test.go index fa3f783d..3a863148 100644 --- a/server/store/redis_integration_test.go +++ b/server/store/redis_integration_test.go @@ -56,7 +56,7 @@ func TestRealRedisCandidateIndexUpsertSnapshotRemove(t *testing.T) { if err := index.Upsert(ctx, b); err != nil { t.Fatalf("upsert b: %v", err) } - got, err := index.Snapshot(ctx, now.Add(time.Hour)) + got, err := index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000) if err != nil { t.Fatalf("snapshot: %v", err) } @@ -64,10 +64,10 @@ func TestRealRedisCandidateIndexUpsertSnapshotRemove(t *testing.T) { t.Fatalf("snapshot after upsert = %+v", got) } - if err := index.Remove(ctx, "real-ticket-a"); err != nil { + if err := index.Remove(ctx, domain.Casual, "real-ticket-a"); err != nil { t.Fatalf("remove: %v", err) } - got, err = index.Snapshot(ctx, now.Add(time.Hour)) + got, err = index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000) if err != nil { t.Fatalf("snapshot after remove: %v", err) } @@ -77,11 +77,11 @@ func TestRealRedisCandidateIndexUpsertSnapshotRemove(t *testing.T) { // 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{TicketID: "real-ticket-ttl", PlayerID: "real-player-ttl", EnqueuedAt: now}); err != nil { + 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, now.Add(time.Hour)) + got, err = shortLived.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000) if err != nil { t.Fatalf("snapshot after real TTL expiry: %v", err) } @@ -100,11 +100,11 @@ func TestRealRedisCandidateProjectionRepairsAfterFlush(t *testing.T) { now := time.Now().UTC().Truncate(time.Microsecond) durable := []domain.Candidate{ - {TicketID: "repair-ticket-a", PlayerID: "repair-player-a", EnqueuedAt: now}, - {TicketID: "repair-ticket-b", PlayerID: "repair-player-b", EnqueuedAt: now.Add(time.Second)}, + {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, time.Time) ([]domain.Candidate, error) { + projection := CandidateProjection{Index: index, Source: func(context.Context, domain.Playlist, time.Time, int) ([]domain.Candidate, error) { sourceCalls++ return durable, nil }} @@ -120,7 +120,7 @@ func TestRealRedisCandidateProjectionRepairsAfterFlush(t *testing.T) { t.Fatalf("flush: %v", err) } - got, err := projection.Snapshot(ctx, now.Add(time.Hour)) + got, err := projection.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 1000) if err != nil { t.Fatalf("snapshot after flush: %v", err) } @@ -134,7 +134,7 @@ func TestRealRedisCandidateProjectionRepairsAfterFlush(t *testing.T) { // 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, now.Add(time.Hour)) + 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) }