mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-15 23:42:07 +00:00
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.
This commit is contained in:
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user