From 18538e833bcbb8e26cc89003faada5b2bb401c99 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 09:28:49 +0100 Subject: [PATCH] feat: wire API queue projection to Redis --- multiplayer-next.md | 5 +-- multiplayer-todo.md | 2 +- server/api/service.go | 27 ++++++++++++++ server/api/service_test.go | 60 ++++++++++++++++++++++++++++++++ server/cmd/control-plane/main.go | 30 ++++++++++++++-- 5 files changed, 119 insertions(+), 5 deletions(-) diff --git a/multiplayer-next.md b/multiplayer-next.md index 28b363f9..596b0074 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -50,8 +50,9 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md). independently runnable API, matcher, allocator and maintenance roles. The `cmd/control-plane` API role now opens PostgreSQL, applies migrations, wires authenticated durable queue/proposal/assignment/session adapters, and shuts - down gracefully; worker roles, Redis worker wiring and live service checks - remain. + down gracefully; optional `--redis-addr` publishes queue mutations to a + TTL-bound best-effort candidate projection without making Redis authoritative; + worker roles, Redis worker wiring and live service checks 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; allocated servers now diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 143db1b9..6d6ad130 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`, `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 durable-source repair on partial/malformed cache state; 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 and repair-source failure; live Redis restart/failover and worker 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, API create/heartbeat/cancel projection hooks, optional control-plane Redis configuration and atomic durable-source repair on partial/malformed cache state; 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 and repair-source failure; 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/api/service.go b/server/api/service.go index 7cc8f6b7..f4e22146 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -34,6 +34,13 @@ type QueueBackend interface { Get(context.Context, string, string, time.Time) (domain.QueueTicket, error) } +// CandidateIndex is a transient projection of durable queue ownership. Index +// 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 +} + type SessionBackend interface { Authenticate(context.Context, string, string, time.Time) (domain.Session, error) } @@ -77,6 +84,7 @@ type Service struct { Candidate CandidateProvider CandidateV2 CandidateProviderV2 QueueBackend QueueBackend + CandidateIndex CandidateIndex Probe ProbeProvider Assignment AssignmentProvider Now func() time.Time @@ -220,6 +228,7 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) { writeDomainError(w, err) return } + s.projectCandidate(r.Context(), ticket) s.publishTicketEvent(ticket, now) writeJSON(w, http.StatusCreated, toQueueResponse(ticket)) return @@ -249,10 +258,23 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) { writeDomainError(w, err) return } + s.projectCandidate(r.Context(), ticket) s.publishTicketEvent(ticket, now) writeJSON(w, http.StatusCreated, toQueueResponse(ticket)) } +func (s *Service) projectCandidate(ctx context.Context, ticket domain.QueueTicket) { + if s.CandidateIndex != nil { + _ = s.CandidateIndex.Upsert(ctx, ticket.Candidate) + } +} + +func (s *Service) removeCandidate(ctx context.Context, ticketID string) { + if s.CandidateIndex != nil { + _ = s.CandidateIndex.Remove(ctx, ticketID) + } +} + func (s *Service) contractQueueCreate(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { s.queueCreate(w, r) @@ -391,6 +413,11 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) { writeDomainError(w, err) return } + if ticket.State == domain.Cancelled { + s.removeCandidate(r.Context(), ticket.TicketID) + } else { + s.projectCandidate(r.Context(), ticket) + } s.publishTicketEvent(ticket, now) if r.Header.Get("X-Contract-Delete") == "1" { w.WriteHeader(http.StatusNoContent) diff --git a/server/api/service_test.go b/server/api/service_test.go index 20a6452a..288ad784 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -19,6 +19,23 @@ import ( type queueBackendSpy struct{ createCalls, heartbeatCalls, cancelCalls, getCalls int } +type candidateIndexSpy struct { + upsertCalls, removeCalls int + upsertErr, removeErr error + last domain.Candidate +} + +func (i *candidateIndexSpy) Upsert(_ context.Context, candidate domain.Candidate) error { + i.upsertCalls++ + i.last = candidate + return i.upsertErr +} + +func (i *candidateIndexSpy) Remove(_ context.Context, _ string) error { + i.removeCalls++ + return i.removeErr +} + type sessionBackendSpy struct{ calls int } type proposalBackendSpy struct { @@ -612,6 +629,49 @@ func TestQueueAPIUsesInjectedPersistentBackendWithoutCandidateProvider(t *testin } } +func TestQueueAPIProjectsSuccessfulMutationsWithoutMakingRedisRequired(t *testing.T) { + now := time.Unix(1000, 0).UTC() + sessions := domain.NewSessionStore() + session, token, _ := sessions.Issue("player-1", time.Hour, now) + backend := &queueBackendSpy{} + index := &candidateIndexSpy{upsertErr: errors.New("redis unavailable"), removeErr: errors.New("redis unavailable")} + service := &Service{Sessions: sessions, QueueBackend: backend, CandidateIndex: index, Now: func() time.Time { return now }} + server := httptest.NewServer(service.Handler()) + defer server.Close() + auth := "Bearer " + session.SessionID + ":" + token + request := func(method, path, key, revision string) *http.Response { + req, _ := http.NewRequest(method, server.URL+path, strings.NewReader(`{"ticket_id":"ticket-1","playlist":"ranked","client_build":"build-1","protocol_version":1}`)) + req.Header.Set("Authorization", auth) + if key != "" { + req.Header.Set("Idempotency-Key", key) + } + if revision != "" { + req.Header.Set("If-Match-Revision", revision) + } + response, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + return response + } + response := request(http.MethodPost, "/v1/queue", "create-key-123456", "") + if response.StatusCode != http.StatusCreated { + t.Fatalf("create status=%d", response.StatusCode) + } + response.Body.Close() + if index.upsertCalls != 1 { + t.Fatalf("upsert calls=%d", index.upsertCalls) + } + response = request(http.MethodPost, "/v1/queue/ticket-1/cancel", "cancel-key-123456", "0") + if response.StatusCode != http.StatusOK { + t.Fatalf("cancel status=%d", response.StatusCode) + } + response.Body.Close() + if index.removeCalls != 1 { + t.Fatalf("remove calls=%d", index.removeCalls) + } +} + func TestQueueAPIDelegatesAllMutationsAndRecoveryToBackend(t *testing.T) { now := time.Unix(1000, 0).UTC() sessions := domain.NewSessionStore() diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index 20ea0684..fcd5f34c 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -15,6 +15,7 @@ import ( "github.com/cosmic-clash/cosmic-clash/server/migrations" "github.com/cosmic-clash/cosmic-clash/server/store" _ "github.com/jackc/pgx/v5/stdlib" + "github.com/redis/go-redis/v9" ) func main() { @@ -22,6 +23,9 @@ func main() { role := flag.String("role", "api", "control-plane role; currently api") dsn := flag.String("dsn", os.Getenv("COSMIC_CLASH_POSTGRES_DSN"), "PostgreSQL connection string") migrationDir := flag.String("migrations", "migrations", "directory containing numbered SQL migrations") + redisAddr := flag.String("redis-addr", os.Getenv("COSMIC_CLASH_REDIS_ADDR"), "optional Redis address for the candidate projection") + redisPrefix := flag.String("redis-prefix", envOrDefault("COSMIC_CLASH_REDIS_PREFIX", "cosmic-clash"), "Redis key prefix") + redisTTL := flag.Duration("redis-ttl", 60*time.Second, "TTL for transient candidate projection entries") flag.Parse() if *role != "api" { fatalf("unsupported role %q (only api is implemented)", *role) @@ -29,6 +33,9 @@ func main() { if *dsn == "" { fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required") } + if *redisTTL <= 0 { + fatalf("--redis-ttl must be positive") + } db, err := sql.Open("pgx", *dsn) if err != nil { fatalf("open PostgreSQL: %v", err) @@ -42,7 +49,14 @@ func main() { if err := migrations.Apply(startupCtx, db, *migrationDir); err != nil { fatalf("apply migrations: %v", err) } - server := &http.Server{Addr: *listen, Handler: newAPIHandler(db), ReadHeaderTimeout: 5 * time.Second} + var candidateIndex api.CandidateIndex + var redisClient *redis.Client + if *redisAddr != "" { + redisClient = redis.NewClient(&redis.Options{Addr: *redisAddr}) + defer redisClient.Close() + candidateIndex = store.RedisCandidateIndex{Client: redisClient, Prefix: *redisPrefix, TTL: *redisTTL} + } + server := &http.Server{Addr: *listen, Handler: newAPIHandler(db, candidateIndex), ReadHeaderTimeout: 5 * time.Second} serveErr := make(chan error, 1) go func() { serveErr <- server.ListenAndServe() }() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) @@ -61,16 +75,28 @@ func main() { } } -func newAPIHandler(db *sql.DB) http.Handler { +func newAPIHandler(db *sql.DB, indexes ...api.CandidateIndex) http.Handler { + var candidateIndex api.CandidateIndex + if len(indexes) > 0 { + candidateIndex = indexes[0] + } return (&api.Service{ SessionBackend: store.PostgresSessions{DB: db}, QueueBackend: store.PostgresQueue{DB: db}, ProposalBackend: api.ProposalProviderFromStore(db), Assignment: api.AssignmentProviderFromStore(db), + CandidateIndex: candidateIndex, Now: func() time.Time { return time.Now().UTC() }, }).Handler() } +func envOrDefault(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +} + func fatalf(format string, args ...any) { fmt.Fprintf(os.Stderr, "control-plane: "+format+"\n", args...) os.Exit(1)