fix(multiplayer): report allocator readiness

server/allocator/health.go

@@ -0,0 +1,65 @@
+package allocator
+
+import (
+	"net/http"
+	"sync"
+	"time"
+)
+
+// Health records only complete successful cycles. Readiness ages out when
+// provider or durable-store work repeatedly fails or stalls, while liveness
+// remains independent so Kubernetes does not restart a healthy process for a
+// dependency outage.
+type Health struct {
+	mu                  sync.RWMutex
+	lastSuccessfulCycle time.Time
+}
+
+func (h *Health) ObserveSuccessfulCycle(at time.Time) {
+	if h == nil || at.IsZero() {
+		return
+	}
+	h.mu.Lock()
+	h.lastSuccessfulCycle = at
+	h.mu.Unlock()
+}
+
+func (h *Health) Ready(now time.Time, maxStale time.Duration) bool {
+	if h == nil || now.IsZero() || maxStale <= 0 {
+		return false
+	}
+	h.mu.RLock()
+	lastSuccess := h.lastSuccessfulCycle
+	h.mu.RUnlock()
+	return !lastSuccess.IsZero() && !now.Before(lastSuccess) && now.Sub(lastSuccess) <= maxStale
+}
+
+// RoleHandler exposes metrics plus distinct process-liveness and dependency-
+// progress readiness endpoints on the allocator's private listener.
+func RoleHandler(metrics *Metrics, health *Health, maxStale time.Duration, now func() time.Time) http.Handler {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", MetricsHandler(metrics))
+	mux.HandleFunc("/healthz", methodGet(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		_, _ = w.Write([]byte("ok\n"))
+	}))
+	mux.HandleFunc("/readyz", methodGet(func(w http.ResponseWriter, _ *http.Request) {
+		if now == nil || !health.Ready(now(), maxStale) {
+			http.Error(w, "allocator has no recent successful cycle", http.StatusServiceUnavailable)
+			return
+		}
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		_, _ = w.Write([]byte("ready\n"))
+	}))
+	return mux
+}
+
+func methodGet(next http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet {
+			w.WriteHeader(http.StatusMethodNotAllowed)
+			return
+		}
+		next(w, r)
+	}
+}

server/allocator/health_test.go

@@ -0,0 +1,53 @@
+package allocator
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestHealthRequiresRecentSuccessfulCycle(t *testing.T) {
+	health := &Health{}
+	now := time.Unix(1000, 0)
+	if health.Ready(now, 30*time.Second) {
+		t.Fatal("allocator was ready before a successful cycle")
+	}
+	health.ObserveSuccessfulCycle(now)
+	if !health.Ready(now.Add(30*time.Second), 30*time.Second) {
+		t.Fatal("allocator was not ready at the staleness boundary")
+	}
+	if health.Ready(now.Add(30*time.Second+time.Nanosecond), 30*time.Second) {
+		t.Fatal("stale allocator remained ready")
+	}
+	if health.Ready(now.Add(-time.Second), 30*time.Second) {
+		t.Fatal("clock reversal was accepted as ready")
+	}
+}
+
+func TestRoleHandlerSeparatesLivenessReadinessAndMetrics(t *testing.T) {
+	health := &Health{}
+	now := time.Unix(1000, 0)
+	handler := RoleHandler(NewMetrics(), health, 30*time.Second, func() time.Time { return now })
+	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 status = %d", got)
+	}
+	if got := status(http.MethodGet, "/readyz"); got != http.StatusServiceUnavailable {
+		t.Fatalf("startup readiness status = %d", got)
+	}
+	health.ObserveSuccessfulCycle(now)
+	if got := status(http.MethodGet, "/readyz"); got != http.StatusOK {
+		t.Fatalf("successful-cycle readiness status = %d", got)
+	}
+	if got := status(http.MethodGet, "/metrics"); got != http.StatusOK {
+		t.Fatalf("metrics status = %d", got)
+	}
+	if got := status(http.MethodPost, "/readyz"); got != http.StatusMethodNotAllowed {
+		t.Fatalf("readiness mutation status = %d", got)
+	}
+}

server/cmd/allocator/main.go

@@ -26,6 +26,7 @@ func main() {
 	kubernetesTokenPath := flag.String("kubernetes-token-path", envOrDefault("COSMIC_CLASH_KUBERNETES_TOKEN_PATH", "/var/run/secrets/kubernetes.io/serviceaccount/token"), "rotating Kubernetes service-account bearer token")
 	kubernetesCAPath := flag.String("kubernetes-ca-path", envOrDefault("COSMIC_CLASH_KUBERNETES_CA_PATH", "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"), "Kubernetes API cluster CA bundle")
 	providerTimeout := flag.Duration("provider-timeout", 10*time.Second, "timeout for each Kubernetes/Agones API request")
+	readinessMaxStale := flag.Duration("readiness-max-stale", 30*time.Second, "maximum age of the last fully successful allocator cycle")
 	transport := flag.String("transport", envOrDefault("COSMIC_CLASH_TRANSPORT", "enet"), "game transport: enet or steam_sdr")
 	interval := flag.Duration("interval", time.Second, "allocation poll interval")
 	workloadSecret := flag.String("workload-secret", os.Getenv("COSMIC_CLASH_WORKLOAD_SECRET"), "HMAC secret for control-plane-issued workload tokens (see workload/signed_token.go); must match cmd/control-plane's own --workload-secret. Unset skips minting a cosmic-clash.io/workload-token annotation entirely")
@@ -36,8 +37,11 @@ func main() {
 	if *dsn == "" || *agonesURL == "" {
 		fatalf("--dsn/COSMIC_CLASH_POSTGRES_DSN and --agones-url/COSMIC_CLASH_AGONES_URL are required")
 	}
-	if (*transport != "enet" && *transport != "steam_sdr") || *interval <= 0 || *providerTimeout <= 0 {
-		fatalf("--transport must be enet or steam_sdr and --interval/--provider-timeout must be positive")
+	if (*transport != "enet" && *transport != "steam_sdr") || *interval <= 0 || *providerTimeout <= 0 || *readinessMaxStale <= 0 {
+		fatalf("--transport must be enet or steam_sdr and --interval/--provider-timeout/--readiness-max-stale must be positive")
+	}
+	if *readinessMaxStale < *interval+*providerTimeout {
+		fatalf("--readiness-max-stale must be at least --interval plus --provider-timeout")
 	}
 	if *allocationQuota < 0 || *allocationQuotaWindow <= 0 {
 		fatalf("--allocation-quota must be non-negative and --allocation-quota-window must be positive")
@@ -68,6 +72,7 @@ func main() {
 		log.Printf("allocator: enabled per-replica regional allocation quota=%d window=%s", *allocationQuota, *allocationQuotaWindow)
 	}
 	metrics := allocator.NewMetrics()
+	health := &allocator.Health{}
 	providerHTTP, err := agones.NewKubernetesHTTPClient(*agonesURL, *kubernetesTokenPath, *kubernetesCAPath, *providerTimeout)
 	if err != nil {
 		fatalf("configure Kubernetes API client: %v", err)
@@ -89,7 +94,7 @@ func main() {
 	defer stop()
 	var metricsServer *http.Server
 	if *metricsAddr != "" {
-		metricsServer = &http.Server{Addr: *metricsAddr, Handler: allocator.MetricsHandler(metrics)}
+		metricsServer = &http.Server{Addr: *metricsAddr, Handler: allocator.RoleHandler(metrics, health, *readinessMaxStale, now), ReadHeaderTimeout: 5 * time.Second}
 		go func() {
 			if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 				log.Printf("allocator: metrics server: %v", err)
@@ -107,19 +112,26 @@ func main() {
 	ticker := time.NewTicker(*interval)
 	defer ticker.Stop()
 	for {
+		cycleHealthy := true
 		servers, err := client.ListReadyServers(ctx)
 		if err != nil && ctx.Err() == nil {
+			cycleHealthy = false
 			log.Printf("allocator: list Ready GameServers: %v", err)
 		} else {
 			for _, server := range servers {
 				if err := store.RegisterReadyServer(ctx, db, server, now()); err != nil && ctx.Err() == nil {
+					cycleHealthy = false
 					log.Printf("allocator: register Ready GameServer %s: %v", server.ServerID, err)
 				}
 			}
 		}
 		if _, err := worker.RunOnce(ctx); err != nil && ctx.Err() == nil {
+			cycleHealthy = false
 			log.Printf("allocator: run once: %v", err)
 		}
+		if cycleHealthy && ctx.Err() == nil {
+			health.ObserveSuccessfulCycle(now())
+		}
 		select {
 		case <-ctx.Done():
 			return

server/security/test_kubernetes_policies.py

@@ -39,6 +39,7 @@ class KubernetesPolicyTest(unittest.TestCase):
             "--metrics-addr=:9091", "containerPort: 9091",
             "key: dsn", "key: secret", "automountServiceAccountToken: true",
             "--agones-url=https://kubernetes.default.svc", "--provider-timeout=10s",
+            "--readiness-max-stale=30s",
         ):
             self.assertIn(required, deployment)
         self.assertNotRegex(deployment, r"(?im)^\s*(password|token|private.?key):\s*[^\n]+$")
@@ -72,7 +73,7 @@ class KubernetesPolicyTest(unittest.TestCase):
             "type: RollingUpdate", "maxUnavailable: 0", "maxSurge: 1",
             "terminationGracePeriodSeconds: 10",
             "readinessProbe:", "livenessProbe:",
-            "path: /metrics", "port: metrics",
+            "path: /readyz", "path: /healthz", "port: metrics",
         ):
             self.assertIn(required, deployment)