feat: wire persistent queue backend into API

This commit is contained in:
Josh Creek
2026-08-31 22:17:32 +01:00
parent b2d68d93cf
commit 59cf29949f
4 changed files with 91 additions and 6 deletions
+1 -1
View File
@@ -1191,7 +1191,7 @@ the local/CI/community transport, not a silent production fallback.
| # | Task | Acceptance | | # | 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, 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; live PostgreSQL row execution, real 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` 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; live PostgreSQL row execution, real Redis index/TTLs and restart/failover 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.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.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 | | 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 |
+35 -3
View File
@@ -4,6 +4,7 @@
package api package api
import ( import (
"context"
"encoding/json" "encoding/json"
"errors" "errors"
"io" "io"
@@ -22,11 +23,19 @@ type CandidateProvider func(playerID, ticketID string) (domain.Candidate, error)
type CandidateProviderV2 func(playerID, ticketID string, spec domain.QueueSpec) (domain.Candidate, error) type CandidateProviderV2 func(playerID, ticketID string, spec domain.QueueSpec) (domain.Candidate, error)
type ProbeProvider func(playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) type ProbeProvider func(playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error)
type QueueBackend interface {
Create(context.Context, string, string, string, domain.QueueSpec, time.Time) (domain.QueueTicket, error)
Heartbeat(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error)
Cancel(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error)
Get(context.Context, string, string, time.Time) (domain.QueueTicket, error)
}
type Service struct { type Service struct {
Sessions *domain.SessionStore Sessions *domain.SessionStore
Queue *domain.Queue Queue *domain.Queue
Candidate CandidateProvider Candidate CandidateProvider
CandidateV2 CandidateProviderV2 CandidateV2 CandidateProviderV2
QueueBackend QueueBackend
Probe ProbeProvider Probe ProbeProvider
Now func() time.Time Now func() time.Time
Proposals map[string]*domain.Proposal Proposals map[string]*domain.Proposal
@@ -75,7 +84,7 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) {
if !ok { if !ok {
return return
} }
if s.Queue == nil || (s.Candidate == nil && s.CandidateV2 == nil) { if (s.Queue == nil && s.QueueBackend == nil) || (s.QueueBackend == nil && s.Candidate == nil && s.CandidateV2 == nil) {
writeError(w, http.StatusServiceUnavailable, "queue_unavailable") writeError(w, http.StatusServiceUnavailable, "queue_unavailable")
return return
} }
@@ -94,6 +103,15 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) {
} }
now := s.now() now := s.now()
spec := domain.QueueSpec{Playlist: domain.Playlist(input.Playlist), ClientBuild: input.ClientBuild, ProtocolVersion: input.ProtocolVersion} spec := domain.QueueSpec{Playlist: domain.Playlist(input.Playlist), ClientBuild: input.ClientBuild, ProtocolVersion: input.ProtocolVersion}
if s.QueueBackend != nil {
ticket, err := s.QueueBackend.Create(r.Context(), playerID, input.TicketID, key, spec, now)
if err != nil {
writeDomainError(w, err)
return
}
writeJSON(w, http.StatusCreated, toQueueResponse(ticket))
return
}
var candidate domain.Candidate var candidate domain.Candidate
var err error var err error
if s.CandidateV2 != nil { if s.CandidateV2 != nil {
@@ -131,7 +149,7 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) {
if !ok { if !ok {
return return
} }
if s.Queue == nil { if s.Queue == nil && s.QueueBackend == nil {
writeError(w, http.StatusServiceUnavailable, "queue_unavailable") writeError(w, http.StatusServiceUnavailable, "queue_unavailable")
return return
} }
@@ -141,7 +159,13 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusNotFound, "not_found") writeError(w, http.StatusNotFound, "not_found")
return return
} }
ticket, err := s.Queue.Get(playerID, parts[0], s.now()) var ticket domain.QueueTicket
var err error
if s.QueueBackend != nil {
ticket, err = s.QueueBackend.Get(r.Context(), playerID, parts[0], s.now())
} else {
ticket, err = s.Queue.Get(playerID, parts[0], s.now())
}
if err != nil { if err != nil {
writeDomainError(w, err) writeDomainError(w, err)
return return
@@ -166,10 +190,18 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) {
now := s.now() now := s.now()
var ticket domain.QueueTicket var ticket domain.QueueTicket
if parts[1] == "heartbeat" { if parts[1] == "heartbeat" {
if s.QueueBackend != nil {
ticket, err = s.QueueBackend.Heartbeat(r.Context(), playerID, ticketID, key, revision, now)
} else {
ticket, err = s.Queue.Heartbeat(playerID, ticketID, key, revision, now) ticket, err = s.Queue.Heartbeat(playerID, ticketID, key, revision, now)
}
} else {
if s.QueueBackend != nil {
ticket, err = s.QueueBackend.Cancel(r.Context(), playerID, ticketID, key, revision, now)
} else { } else {
ticket, err = s.Queue.Cancel(playerID, ticketID, key, revision, now) ticket, err = s.Queue.Cancel(playerID, ticketID, key, revision, now)
} }
}
if err != nil { if err != nil {
writeDomainError(w, err) writeDomainError(w, err)
return return
+38
View File
@@ -1,6 +1,7 @@
package api package api
import ( import (
"context"
"encoding/json" "encoding/json"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
@@ -11,6 +12,22 @@ import (
"github.com/cosmic-clash/cosmic-clash/server/domain" "github.com/cosmic-clash/cosmic-clash/server/domain"
) )
type queueBackendSpy struct{ createCalls int }
func (b *queueBackendSpy) Create(_ context.Context, playerID, ticketID, _ string, spec domain.QueueSpec, now time.Time) (domain.QueueTicket, error) {
b.createCalls++
return domain.QueueTicket{TicketID: ticketID, PlayerID: playerID, Playlist: spec.Playlist, State: domain.Queued, EnqueuedAt: now, ExpiresAt: now.Add(domain.QueueExpiryWindow)}, nil
}
func (*queueBackendSpy) Heartbeat(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error) {
return domain.QueueTicket{}, nil
}
func (*queueBackendSpy) Cancel(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error) {
return domain.QueueTicket{}, nil
}
func (*queueBackendSpy) Get(context.Context, string, string, time.Time) (domain.QueueTicket, error) {
return domain.QueueTicket{}, nil
}
func TestAuthenticatedQueueAPIUsesServerCandidateAndRevisionedMutations(t *testing.T) { func TestAuthenticatedQueueAPIUsesServerCandidateAndRevisionedMutations(t *testing.T) {
now := time.Unix(1000, 0).UTC() now := time.Unix(1000, 0).UTC()
sessions := domain.NewSessionStore() sessions := domain.NewSessionStore()
@@ -184,6 +201,27 @@ func TestQueueCreateRejectsCandidateMetadataMismatch(t *testing.T) {
} }
} }
func TestQueueAPIUsesInjectedPersistentBackendWithoutCandidateProvider(t *testing.T) {
now := time.Unix(1000, 0).UTC()
sessions := domain.NewSessionStore()
session, token, _ := sessions.Issue("player-1", time.Hour, now)
backend := &queueBackendSpy{}
service := &Service{Sessions: sessions, QueueBackend: backend, Now: func() time.Time { return now }}
server := httptest.NewServer(service.Handler())
defer server.Close()
req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/queue", strings.NewReader(`{"ticket_id":"ticket-1","playlist":"ranked","client_build":"build-1","protocol_version":1}`))
req.Header.Set("Authorization", "Bearer "+session.SessionID+":"+token)
req.Header.Set("Idempotency-Key", "create-key-123456")
response, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusCreated || backend.createCalls != 1 {
t.Fatalf("status=%d backend_calls=%d", response.StatusCode, backend.createCalls)
}
}
func TestQueueRecoveryAPIIsAuthenticatedOwnerOnlyAndExpiresStaleTickets(t *testing.T) { func TestQueueRecoveryAPIIsAuthenticatedOwnerOnlyAndExpiresStaleTickets(t *testing.T) {
now := time.Unix(1000, 0).UTC() now := time.Unix(1000, 0).UTC()
sessions := domain.NewSessionStore() sessions := domain.NewSessionStore()
+15
View File
@@ -91,6 +91,21 @@ type queueTicketRecord struct {
Revision uint64 `json:"revision"` Revision uint64 `json:"revision"`
} }
type PostgresQueue struct{ DB *sql.DB }
func (q PostgresQueue) Create(ctx context.Context, playerID, ticketID, idempotencyKey string, spec domain.QueueSpec, now time.Time) (domain.QueueTicket, error) {
return CreateQueueTicket(ctx, q.DB, ticketID, playerID, idempotencyKey, spec, now)
}
func (q PostgresQueue) Heartbeat(ctx context.Context, playerID, ticketID, idempotencyKey string, revision uint64, now time.Time) (domain.QueueTicket, error) {
return HeartbeatQueueTicket(ctx, q.DB, playerID, ticketID, idempotencyKey, revision, now)
}
func (q PostgresQueue) Cancel(ctx context.Context, playerID, ticketID, idempotencyKey string, revision uint64, now time.Time) (domain.QueueTicket, error) {
return CancelQueueTicket(ctx, q.DB, playerID, ticketID, idempotencyKey, revision, now)
}
func (q PostgresQueue) Get(ctx context.Context, playerID, ticketID string, now time.Time) (domain.QueueTicket, error) {
return GetQueueTicket(ctx, q.DB, playerID, ticketID, now)
}
func GetQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID string, now time.Time) (domain.QueueTicket, error) { func GetQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID string, now time.Time) (domain.QueueTicket, error) {
if db == nil || playerID == "" || ticketID == "" || now.IsZero() { if db == nil || playerID == "" || ticketID == "" || now.IsZero() {
return domain.QueueTicket{}, fmt.Errorf("invalid queue recovery arguments") return domain.QueueTicket{}, fmt.Errorf("invalid queue recovery arguments")