From eadaa7b54e0dba77e836d0e0ee9eba97b75b335a Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 09:01:32 +0100 Subject: [PATCH] feat: add rebuildable Redis candidate index --- multiplayer-next.md | 4 +- multiplayer-todo.md | 2 +- server/go.mod | 9 +- server/go.sum | 14 +++ server/store/redis_candidates.go | 161 ++++++++++++++++++++++++++ server/store/redis_candidates_test.go | 72 ++++++++++++ 6 files changed, 259 insertions(+), 3 deletions(-) create mode 100644 server/store/redis_candidates.go create mode 100644 server/store/redis_candidates_test.go diff --git a/multiplayer-next.md b/multiplayer-next.md index 5bc076d7..bf47a21c 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -41,7 +41,9 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md). and serializable store boundaries are implemented, including durable queue create/heartbeat/cancel/recovery adapters; an opt-in pgx/Docker harness now executes the migrations and real queue create/idempotency/ownership/recovery - path; proposal/result transactions and live Redis/cache-repair gates remain. + path, and a TTL-bound Redis candidate index now supports atomic rebuild, + snapshot and removal; proposal/result transactions and live Redis + restart/failover gates remain. - [ ] **IN PROGRESS:** Define assignment compatibility and opt-in `ServerConfig` flags whose defaults reproduce the community-server path. Allocation manifest validation now covers client build and future expiry; signed admission remains. diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 54ed03d1..ecbfa40c 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1192,7 +1192,7 @@ the local/CI/community transport, not a silent production fallback. | # | Task | Acceptance | |---|---|---| -| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue domain enforces one active ticket per verified player under concurrent mutation, 10 s heartbeat/30 s expiry, retry-safe create/heartbeat/cancel, owner-only recovery reads and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary and authenticated HTTP queue adapter with playlist/build/protocol compatibility metadata | `server/domain/queue.go`, `server/store/candidates.go`, `server/store/queue_sql.go` and `server/api/service.go` cover ownership/expiry/idempotency, concurrent create fencing, candidate/player ownership binding, owner/revision-scoped SQL heartbeat/cancel/recovery, injectable PostgreSQL queue backend selected by the HTTP service, authoritative queue-to-cache rebuild, expired recovery as a terminal error, owner-scoped SQL recovery with authoritative expiry handling, server-owned candidate resolution, strict compatibility metadata and cache loss/atomic rebuild; opt-in real PostgreSQL execution now covers create/replay/active-player fencing, owner recovery, revision-fenced heartbeat/cancel and expiry; Redis index/TTLs and restart/failover integration remain | +| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue domain enforces one active ticket per verified player under concurrent mutation, 10 s heartbeat/30 s expiry, retry-safe create/heartbeat/cancel, owner-only recovery reads and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary and authenticated HTTP queue adapter with playlist/build/protocol compatibility metadata | `server/domain/queue.go`, `server/store/candidates.go`, `server/store/queue_sql.go`, `server/store/redis_candidates.go` and `server/api/service.go` cover ownership/expiry/idempotency, concurrent create fencing, candidate/player ownership binding, owner/revision-scoped SQL heartbeat/cancel/recovery, injectable PostgreSQL queue backend selected by the HTTP service, authoritative queue-to-cache rebuild, expired recovery as a terminal error, owner-scoped SQL recovery with authoritative expiry handling, server-owned candidate resolution, strict compatibility metadata, TTL-bound Redis upsert/remove/snapshot and atomic rebuild; opt-in real PostgreSQL execution now covers create/replay/active-player fencing, owner recovery, revision-fenced heartbeat/cancel and expiry, while miniredis covers Redis behavior; live Redis restart/failover and worker integration remain | | 8.15 `[D:7.8,8.3]` | **IN PROGRESS.** Pure Go probe validation treats Steam location as opaque, requires nonce/freshness/region and server-computed RTT, and implements discrepancy quarantine/release; authenticated HTTP now accepts only opaque location/nonce input through a server-owned probe provider | `server/domain/probes.go`, adversarial fixtures and `server/api/service.go` cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine, five-clean release, authenticated provider arguments and rejection of client RTT fields; Steam coordinator and regional probe adapters remain | | 8.16 `[D:8.14,8.15]` | **IN PROGRESS.** Pure Go candidate/team selection implements the <=100 ms ceiling, pairwise widening tolerance, anchor inclusion, deterministic set/region scoring and balanced team partitioning; queue-backed formation now consumes the server-owned projection, fences duplicate player identities and rejects playlist/build/protocol mixing | `server/domain/matcher.go`, `teams.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance, malformed candidates, duplicate identities, compatibility mismatches and queue-backed oldest-anchor formation; full population fixtures and durable matcher claim integration remain | | 8.17 `[D:8.14,8.16]` | **IN PROGRESS.** Pure Go proposal policy sends a 10-second response window to every selected human, requires unanimous acceptance, applies exact decline/timeout cooldowns and ranked escalation; authenticated API exposes revisioned accept/decline mutations; formed matches now pass through a playlist-aware proposal boundary | `server/domain/proposal.go`, `formation.go` and `server/api/service.go` plus adversarial fixtures cover partial/unanimous response, expiry, replay/conflict, stale API revision, casual lineup preparation and ranked metadata validation; queue precedence and allocation integration remain | diff --git a/server/go.mod b/server/go.mod index 8a1285d9..37a36490 100644 --- a/server/go.mod +++ b/server/go.mod @@ -2,12 +2,19 @@ module github.com/cosmic-clash/cosmic-clash/server go 1.23 -require github.com/jackc/pgx/v5 v5.7.4 +require ( + github.com/alicebob/miniredis/v2 v2.38.0 + github.com/jackc/pgx/v5 v5.7.4 + github.com/redis/go-redis/v9 v9.7.0 +) require ( + github.com/cespare/xxhash/v2 v2.2.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect golang.org/x/crypto v0.31.0 // indirect golang.org/x/sync v0.10.0 // indirect golang.org/x/text v0.21.0 // indirect diff --git a/server/go.sum b/server/go.sum index fa0f7db5..aa5eb99c 100644 --- a/server/go.sum +++ b/server/go.sum @@ -1,6 +1,16 @@ +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= +github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= @@ -11,11 +21,15 @@ github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E= +github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ= diff --git a/server/store/redis_candidates.go b/server/store/redis_candidates.go new file mode 100644 index 00000000..5848d6fd --- /dev/null +++ b/server/store/redis_candidates.go @@ -0,0 +1,161 @@ +package store + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" + "github.com/redis/go-redis/v9" +) + +// RedisCandidateIndex is a rebuildable acceleration index. It never decides +// ownership or claims a match; callers must source candidates from the +// durable queue projection before rebuilding it. +type RedisCandidateIndex struct { + Client *redis.Client + Prefix string + TTL time.Duration +} + +func (r RedisCandidateIndex) keys() (string, string) { + prefix := r.Prefix + if prefix == "" { + prefix = "cosmic-clash" + } + return prefix + ":queue:candidates:data", prefix + ":queue:candidates:order" +} + +func (r RedisCandidateIndex) validate() error { + if r.Client == nil || r.TTL <= 0 { + return fmt.Errorf("invalid Redis candidate index") + } + return nil +} + +func validateRedisCandidate(candidate domain.Candidate) error { + if candidate.TicketID == "" || candidate.PlayerID == "" || candidate.EnqueuedAt.IsZero() { + return fmt.Errorf("invalid candidate") + } + return nil +} + +// Upsert stores the candidate payload and its deterministic enqueue ordering. +// Both keys receive a TTL so a Redis restart or abandoned index cannot become +// a permanent source of stale presence. +func (r RedisCandidateIndex) Upsert(ctx context.Context, candidate domain.Candidate) error { + if err := r.validate(); err != nil { + return err + } + if err := validateRedisCandidate(candidate); err != nil { + return err + } + payload, err := json.Marshal(candidate) + if err != nil { + return err + } + dataKey, orderKey := r.keys() + 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}) + pipe.Expire(ctx, dataKey, r.TTL) + pipe.Expire(ctx, orderKey, r.TTL) + _, err = pipe.Exec(ctx) + return err +} + +func (r RedisCandidateIndex) Remove(ctx context.Context, ticketID string) error { + if err := r.validate(); err != nil { + return err + } + if ticketID == "" { + return fmt.Errorf("ticket ID is required") + } + dataKey, orderKey := r.keys() + pipe := r.Client.TxPipeline() + pipe.HDel(ctx, dataKey, ticketID) + pipe.ZRem(ctx, orderKey, ticketID) + _, err := pipe.Exec(ctx) + return err +} + +// 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) { + if err := r.validate(); err != nil { + return nil, err + } + if now.IsZero() { + return nil, fmt.Errorf("authoritative time is required") + } + dataKey, orderKey := r.keys() + tickets, err := r.Client.ZRangeByScore(ctx, orderKey, &redis.ZRangeBy{ + Min: "-inf", Max: fmt.Sprint(now.UnixNano()), + }).Result() + if err != nil { + return nil, err + } + if len(tickets) == 0 { + return []domain.Candidate{}, nil + } + payloads, err := r.Client.HMGet(ctx, dataKey, tickets...).Result() + if err != nil { + return nil, err + } + result := make([]domain.Candidate, 0, len(payloads)) + for _, raw := range payloads { + text, ok := raw.(string) + if !ok { + continue + } + var candidate domain.Candidate + if err := json.Unmarshal([]byte(text), &candidate); err != nil { + continue + } + if err := validateRedisCandidate(candidate); err != nil || candidate.EnqueuedAt.After(now) { + continue + } + result = append(result, candidate) + } + return result, nil +} + +// 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 { + if err := r.validate(); 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)) + for _, candidate := range candidates { + if err := validateRedisCandidate(candidate); err != nil { + return err + } + if _, exists := seen[candidate.TicketID]; exists { + return fmt.Errorf("duplicate candidate in rebuild") + } + seen[candidate.TicketID] = struct{}{} + payload, err := json.Marshal(candidate) + if err != nil { + return err + } + values = append(values, candidate.TicketID, payload) + scores = append(scores, redis.Z{Score: float64(candidate.EnqueuedAt.UnixNano()), Member: candidate.TicketID}) + } + dataKey, orderKey := r.keys() + pipe := r.Client.TxPipeline() + pipe.Del(ctx, dataKey, orderKey) + if len(values) > 0 { + pipe.HSet(ctx, dataKey, values...) + pipe.ZAdd(ctx, orderKey, scores...) + } + pipe.Expire(ctx, dataKey, r.TTL) + pipe.Expire(ctx, orderKey, r.TTL) + _, err := pipe.Exec(ctx) + return err +} diff --git a/server/store/redis_candidates_test.go b/server/store/redis_candidates_test.go new file mode 100644 index 00000000..da31a2a1 --- /dev/null +++ b/server/store/redis_candidates_test.go @@ -0,0 +1,72 @@ +package store + +import ( + "context" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/cosmic-clash/cosmic-clash/server/domain" + "github.com/redis/go-redis/v9" +) + +func TestRedisCandidateIndexRebuildSnapshotAndRemove(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: "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}, + } + if err := index.Rebuild(context.Background(), candidates); err != nil { + t.Fatal(err) + } + got, err := index.Snapshot(context.Background(), now) + 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 { + t.Fatal(err) + } + got, err = index.Snapshot(context.Background(), now.Add(2*time.Second)) + 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 { + 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 { + t.Fatal("nil Redis client accepted") + } + 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.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 { + t.Fatal("duplicate candidate accepted") + } + if err := index.Upsert(context.Background(), domain.Candidate{TicketID: "", PlayerID: "player-a", EnqueuedAt: candidate.EnqueuedAt}); err == nil { + t.Fatal("invalid candidate accepted") + } +}