mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-14 04:02:03 +00:00
fix(multiplayer): gate API readiness on database
This commit is contained in:
@@ -61,7 +61,7 @@ spec:
|
|||||||
containerPort: 8080
|
containerPort: 8080
|
||||||
readinessProbe:
|
readinessProbe:
|
||||||
httpGet:
|
httpGet:
|
||||||
path: /healthz
|
path: /readyz
|
||||||
port: http
|
port: http
|
||||||
periodSeconds: 5
|
periodSeconds: 5
|
||||||
timeoutSeconds: 2
|
timeoutSeconds: 2
|
||||||
|
|||||||
@@ -1646,3 +1646,5 @@ Per-IP API limiting now resolves the client behind the edge gateway instead of c
|
|||||||
Allocator probes now distinguish process liveness from useful progress. `/healthz` remains live during dependency outages, while `/readyz` starts unavailable and requires a fully successful provider-list, Ready-registration, and worker cycle within `--readiness-max-stale` (30 seconds in the base deployment). The Kubernetes/Agones HTTP path is bounded by `--provider-timeout=10s`, so an unavailable provider cannot leave readiness green indefinitely; startup rejects a freshness window shorter than the poll interval plus provider timeout, and the probe listener has its own header-read deadline. Boundary and HTTP tests cover startup, exact staleness, clock reversal, recovery, method rejection, and metrics coexistence.
|
Allocator probes now distinguish process liveness from useful progress. `/healthz` remains live during dependency outages, while `/readyz` starts unavailable and requires a fully successful provider-list, Ready-registration, and worker cycle within `--readiness-max-stale` (30 seconds in the base deployment). The Kubernetes/Agones HTTP path is bounded by `--provider-timeout=10s`, so an unavailable provider cannot leave readiness green indefinitely; startup rejects a freshness window shorter than the poll interval plus provider timeout, and the probe listener has its own header-read deadline. Boundary and HTTP tests cover startup, exact staleness, clock reversal, recovery, method rejection, and metrics coexistence.
|
||||||
|
|
||||||
The timeout boundary is enforced inside both network adapters as well as in the production allocator wiring: an `agones.Client` or game-server `Supervisor` constructed without an injected HTTP client now receives a ten-second client rather than Go's unbounded `http.DefaultClient`. This prevents alternate binaries, tests, and future callers from restoring an infinite GameServer, roster, registration, or SDK wait by omission.
|
The timeout boundary is enforced inside both network adapters as well as in the production allocator wiring: an `agones.Client` or game-server `Supervisor` constructed without an injected HTTP client now receives a ten-second client rather than Go's unbounded `http.DefaultClient`. This prevents alternate binaries, tests, and future callers from restoring an infinite GameServer, roster, registration, or SDK wait by omission.
|
||||||
|
|
||||||
|
Control-plane probes now separate liveness from datastore readiness too. `/healthz` proves the process can serve without restarting it during a PostgreSQL outage; `/readyz` runs a one-second-bounded `PingContext` and the Deployment routes traffic only to replicas whose core durable store responds. Probe and metrics routes bypass the player request limiter, so operator-selected low limits cannot make Kubernetes evict a healthy replica. Missing checks, datastore errors, non-GET methods, and successful recovery are covered by API tests.
|
||||||
|
|||||||
@@ -56,7 +56,7 @@ func TestRateLimitedHTTPBoundaryReturnsGeneric429(t *testing.T) {
|
|||||||
service := &Service{RateLimiter: limiter, Now: func() time.Time { return time.Unix(1000, 0) }}
|
service := &Service{RateLimiter: limiter, Now: func() time.Time { return time.Unix(1000, 0) }}
|
||||||
server := httptest.NewServer(service.Handler())
|
server := httptest.NewServer(service.Handler())
|
||||||
defer server.Close()
|
defer server.Close()
|
||||||
request, err := http.NewRequest(http.MethodGet, server.URL+"/healthz", nil)
|
request, err := http.NewRequest(http.MethodGet, server.URL+"/unknown", nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -66,10 +66,10 @@ func TestRateLimitedHTTPBoundaryReturnsGeneric429(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
_ = response.Body.Close()
|
_ = response.Body.Close()
|
||||||
if response.StatusCode != http.StatusOK {
|
if response.StatusCode != http.StatusNotFound {
|
||||||
t.Fatalf("first request status = %d", response.StatusCode)
|
t.Fatalf("first request status = %d", response.StatusCode)
|
||||||
}
|
}
|
||||||
request, _ = http.NewRequest(http.MethodGet, server.URL+"/healthz", strings.NewReader(""))
|
request, _ = http.NewRequest(http.MethodGet, server.URL+"/unknown", strings.NewReader(""))
|
||||||
request.Header.Set("Authorization", "Bearer secret-session:secret-token")
|
request.Header.Set("Authorization", "Bearer secret-session:secret-token")
|
||||||
response, err = server.Client().Do(request)
|
response, err = server.Client().Do(request)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -128,7 +128,7 @@ func TestRateLimiterSeparatesClientsBehindTrustedGateway(t *testing.T) {
|
|||||||
server := httptest.NewServer(service.Handler())
|
server := httptest.NewServer(service.Handler())
|
||||||
defer server.Close()
|
defer server.Close()
|
||||||
request := func(forwarded string) int {
|
request := func(forwarded string) int {
|
||||||
req, requestErr := http.NewRequest(http.MethodGet, server.URL+"/healthz", nil)
|
req, requestErr := http.NewRequest(http.MethodGet, server.URL+"/unknown", nil)
|
||||||
if requestErr != nil {
|
if requestErr != nil {
|
||||||
t.Fatal(requestErr)
|
t.Fatal(requestErr)
|
||||||
}
|
}
|
||||||
@@ -140,10 +140,10 @@ func TestRateLimiterSeparatesClientsBehindTrustedGateway(t *testing.T) {
|
|||||||
response.Body.Close()
|
response.Body.Close()
|
||||||
return response.StatusCode
|
return response.StatusCode
|
||||||
}
|
}
|
||||||
if got := request("198.51.100.1"); got != http.StatusOK {
|
if got := request("198.51.100.1"); got != http.StatusNotFound {
|
||||||
t.Fatalf("first client status = %d", got)
|
t.Fatalf("first client status = %d", got)
|
||||||
}
|
}
|
||||||
if got := request("198.51.100.2"); got != http.StatusOK {
|
if got := request("198.51.100.2"); got != http.StatusNotFound {
|
||||||
t.Fatalf("second client behind gateway status = %d", got)
|
t.Fatalf("second client behind gateway status = %d", got)
|
||||||
}
|
}
|
||||||
if got := request("198.51.100.1"); got != http.StatusTooManyRequests {
|
if got := request("198.51.100.1"); got != http.StatusTooManyRequests {
|
||||||
|
|||||||
+31
-2
@@ -100,6 +100,7 @@ type AssignmentView struct {
|
|||||||
|
|
||||||
type AssignmentProvider func(context.Context, string, string, time.Time) (AssignmentView, error)
|
type AssignmentProvider func(context.Context, string, string, time.Time) (AssignmentView, error)
|
||||||
type RosterProvider func(context.Context, domain.WorkloadBinding, time.Time) ([][]byte, error)
|
type RosterProvider func(context.Context, domain.WorkloadBinding, time.Time) ([][]byte, error)
|
||||||
|
type ReadinessCheck func(context.Context) error
|
||||||
|
|
||||||
type Service struct {
|
type Service struct {
|
||||||
Sessions *domain.SessionStore
|
Sessions *domain.SessionStore
|
||||||
@@ -129,6 +130,7 @@ type Service struct {
|
|||||||
RateLimiter *RateLimiter
|
RateLimiter *RateLimiter
|
||||||
ClientIPs *ClientIPResolver
|
ClientIPs *ClientIPResolver
|
||||||
Admission AdmissionController
|
Admission AdmissionController
|
||||||
|
ReadinessCheck ReadinessCheck
|
||||||
// Log receives a credential-safe structured event for lifecycle-relevant
|
// Log receives a credential-safe structured event for lifecycle-relevant
|
||||||
// reads and mutations. Nil
|
// reads and mutations. Nil
|
||||||
// is a valid, silent no-op -- every call site must stay optional so
|
// is a valid, silent no-op -- every call site must stay optional so
|
||||||
@@ -185,6 +187,7 @@ func (s *Service) logProposalOutcome(proposalID string, proposal domain.Proposal
|
|||||||
func (s *Service) Handler() http.Handler {
|
func (s *Service) Handler() http.Handler {
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
mux.HandleFunc("/healthz", s.health)
|
mux.HandleFunc("/healthz", s.health)
|
||||||
|
mux.HandleFunc("/readyz", s.ready)
|
||||||
mux.HandleFunc("/metrics", s.metrics)
|
mux.HandleFunc("/metrics", s.metrics)
|
||||||
mux.HandleFunc("/v1/session/steam", s.steamSession)
|
mux.HandleFunc("/v1/session/steam", s.steamSession)
|
||||||
mux.HandleFunc("/v1/queue", s.queueCreate)
|
mux.HandleFunc("/v1/queue", s.queueCreate)
|
||||||
@@ -208,6 +211,10 @@ func (s *Service) Handler() http.Handler {
|
|||||||
var handler http.Handler = mux
|
var handler http.Handler = mux
|
||||||
if s.RateLimiter != nil {
|
if s.RateLimiter != nil {
|
||||||
handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if r.URL.Path == "/healthz" || r.URL.Path == "/readyz" || r.URL.Path == "/metrics" {
|
||||||
|
mux.ServeHTTP(w, r)
|
||||||
|
return
|
||||||
|
}
|
||||||
if !s.RateLimiter.AllowKeys(requestRateKeys(r, s.ClientIPs), s.now()) {
|
if !s.RateLimiter.AllowKeys(requestRateKeys(r, s.ClientIPs), s.now()) {
|
||||||
writeError(w, http.StatusTooManyRequests, "rate_limited")
|
writeError(w, http.StatusTooManyRequests, "rate_limited")
|
||||||
return
|
return
|
||||||
@@ -229,7 +236,7 @@ func (s *Service) Handler() http.Handler {
|
|||||||
return handler
|
return handler
|
||||||
}
|
}
|
||||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
if r.URL.Path == "/metrics" || r.URL.Path == "/healthz" || strings.HasSuffix(r.URL.Path, "/events") {
|
if r.URL.Path == "/metrics" || r.URL.Path == "/healthz" || r.URL.Path == "/readyz" || strings.HasSuffix(r.URL.Path, "/events") {
|
||||||
handler.ServeHTTP(w, r)
|
handler.ServeHTTP(w, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -337,10 +344,32 @@ func (s *Service) steamSession(w http.ResponseWriter, r *http.Request) {
|
|||||||
writeJSON(w, http.StatusOK, map[string]any{"player_id": session.PlayerID, "expires_at": session.ExpiresAt, "access_token": session.SessionID + ":" + token})
|
writeJSON(w, http.StatusOK, map[string]any{"player_id": session.PlayerID, "expires_at": session.ExpiresAt, "access_token": session.SessionID + ":" + token})
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Service) health(w http.ResponseWriter, _ *http.Request) {
|
func (s *Service) health(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if r.Method != http.MethodGet {
|
||||||
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
||||||
|
return
|
||||||
|
}
|
||||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Service) ready(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if r.Method != http.MethodGet {
|
||||||
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if s.ReadinessCheck == nil {
|
||||||
|
writeError(w, http.StatusServiceUnavailable, "not_ready")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
ctx, cancel := context.WithTimeout(r.Context(), time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if err := s.ReadinessCheck(ctx); err != nil {
|
||||||
|
writeError(w, http.StatusServiceUnavailable, "not_ready")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
|
||||||
|
}
|
||||||
|
|
||||||
type queueCreateRequest struct {
|
type queueCreateRequest struct {
|
||||||
TicketID string `json:"ticket_id"`
|
TicketID string `json:"ticket_id"`
|
||||||
Playlist string `json:"playlist"`
|
Playlist string `json:"playlist"`
|
||||||
|
|||||||
@@ -1487,6 +1487,59 @@ func TestMetricsEndpointExportsBoundedAPILatencyAndSkipsItsOwnScrape(t *testing.
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestControlPlaneLivenessAndDatastoreReadinessAreIndependent(t *testing.T) {
|
||||||
|
ready := false
|
||||||
|
checks := 0
|
||||||
|
limiter, err := NewRateLimiter(1, time.Minute, 8)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
service := &Service{
|
||||||
|
RateLimiter: limiter,
|
||||||
|
Now: func() time.Time { return time.Unix(1000, 0) },
|
||||||
|
ReadinessCheck: func(context.Context) error {
|
||||||
|
checks++
|
||||||
|
if !ready {
|
||||||
|
return errors.New("database unavailable")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
}
|
||||||
|
handler := service.Handler()
|
||||||
|
status := func(method, path string) int {
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(recorder, httptest.NewRequest(method, path, nil))
|
||||||
|
return recorder.Code
|
||||||
|
}
|
||||||
|
if got := status(http.MethodGet, "/healthz"); got != http.StatusOK {
|
||||||
|
t.Fatalf("liveness during datastore outage = %d", got)
|
||||||
|
}
|
||||||
|
if got := status(http.MethodGet, "/readyz"); got != http.StatusServiceUnavailable {
|
||||||
|
t.Fatalf("readiness during datastore outage = %d", got)
|
||||||
|
}
|
||||||
|
ready = true
|
||||||
|
if got := status(http.MethodGet, "/readyz"); got != http.StatusOK {
|
||||||
|
t.Fatalf("recovered readiness = %d", got)
|
||||||
|
}
|
||||||
|
if got := status(http.MethodGet, "/healthz"); got != http.StatusOK {
|
||||||
|
t.Fatalf("repeated probe was incorrectly rate limited: %d", got)
|
||||||
|
}
|
||||||
|
if checks != 2 {
|
||||||
|
t.Fatalf("readiness checks = %d", checks)
|
||||||
|
}
|
||||||
|
if got := status(http.MethodPost, "/readyz"); got != http.StatusMethodNotAllowed {
|
||||||
|
t.Fatalf("readiness mutation status = %d", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestControlPlaneReadinessFailsClosedWithoutCheck(t *testing.T) {
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
(&Service{}).Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil))
|
||||||
|
if recorder.Code != http.StatusServiceUnavailable {
|
||||||
|
t.Fatalf("unconfigured readiness status = %d", recorder.Code)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestProbeAPIUsesServerEvidenceAndRejectsClientRTTField(t *testing.T) {
|
func TestProbeAPIUsesServerEvidenceAndRejectsClientRTTField(t *testing.T) {
|
||||||
now := time.Unix(1000, 0).UTC()
|
now := time.Unix(1000, 0).UTC()
|
||||||
sessions := domain.NewSessionStore()
|
sessions := domain.NewSessionStore()
|
||||||
|
|||||||
@@ -144,6 +144,7 @@ func newAPIService(db *sql.DB, workloadSecret string, indexes ...api.CandidateIn
|
|||||||
CandidateIndex: candidateIndex,
|
CandidateIndex: candidateIndex,
|
||||||
ProbeRecorder: store.PostgresQueue{DB: db},
|
ProbeRecorder: store.PostgresQueue{DB: db},
|
||||||
WorkloadVerify: api.WorkloadVerifierFromSignedToken([]byte(workloadSecret), db),
|
WorkloadVerify: api.WorkloadVerifierFromSignedToken([]byte(workloadSecret), db),
|
||||||
|
ReadinessCheck: db.PingContext,
|
||||||
Now: func() time.Time { return time.Now().UTC() },
|
Now: func() time.Time { return time.Now().UTC() },
|
||||||
Log: logEvent,
|
Log: logEvent,
|
||||||
Metrics: observability.NewMetrics(),
|
Metrics: observability.NewMetrics(),
|
||||||
|
|||||||
@@ -47,7 +47,7 @@ class KubernetesPolicyTest(unittest.TestCase):
|
|||||||
def test_control_plane_has_health_rollout_and_failure_domain_guards(self):
|
def test_control_plane_has_health_rollout_and_failure_domain_guards(self):
|
||||||
deployment = self.read("control-plane-deployment.yaml")
|
deployment = self.read("control-plane-deployment.yaml")
|
||||||
for required in (
|
for required in (
|
||||||
"readinessProbe:", "livenessProbe:", "path: /healthz", "port: http",
|
"readinessProbe:", "livenessProbe:", "path: /readyz", "path: /healthz", "port: http",
|
||||||
"type: RollingUpdate", "maxUnavailable: 0", "maxSurge: 1",
|
"type: RollingUpdate", "maxUnavailable: 0", "maxSurge: 1",
|
||||||
"terminationGracePeriodSeconds: 10",
|
"terminationGracePeriodSeconds: 10",
|
||||||
"topologySpreadConstraints:", "maxSkew: 1",
|
"topologySpreadConstraints:", "maxSkew: 1",
|
||||||
|
|||||||
Reference in New Issue
Block a user