Health-check heuristic fix, operator drain/block controls, and websocket close-code correctness

cmd/main.go

@@ -38,6 +38,22 @@ var (
 // the executable to get the full path to the config file.
 const defaultConfigPath = "config/.config.yaml"
 
+// websocketShutdownTimeout bounds the close-handshake sweep over live websocket
+// connections at termination.
+//
+// The sweep runs the closes concurrently, so this is a ceiling on the slowest peer rather
+// than a per-connection cost — five seconds is far past the one-second write deadline on
+// each close frame, and comfortably inside the pod's 30s termination grace period. Being
+// polite must never be the reason a pod gets SIGKILLed, which would produce exactly the
+// abrupt teardown the sweep exists to prevent.
+const websocketShutdownTimeout = 5 * time.Second
+
+// websocketShutdowner is implemented by protocols that hold live websocket bridges and
+// must close them explicitly at termination.
+type websocketShutdowner interface {
+	ShutdownWebsockets(ctx context.Context, reason string) int
+}
+
 func main() {
 	log.Printf(`{"level":"info","message":"PATH 🌿 gateway starting..."}`)
 
@@ -347,6 +363,14 @@ func main() {
 	// failing startup.
 	websocketAdmin, _ := protocol.(router.WebsocketAdmin)
 
+	// Admin handler to temporarily bench one operator's endpoints for a service via
+	// POST /admin/reputation/drain/{serviceId}. nil when reputation is disabled, which
+	// leaves the endpoint reporting 503 rather than failing startup.
+	var reputationAdmin router.ReputationAdmin
+	if reputationSvc := protocol.GetReputationService(); reputationSvc != nil {
+		reputationAdmin = reputationSvc
+	}
+
 	// Initialize the API router to serve requests to the PATH API.
 	apiRouter := router.NewRouter(
 		logger,
@@ -357,6 +381,7 @@ func main() {
 		gtw.DomainCircuitBreaker,
 		chainStateAdmin,
 		websocketAdmin,
+		reputationAdmin,
 		unifiedServicesConfig,
 	)
 
@@ -433,6 +458,30 @@ func main() {
 
 	logger.Info().Msg("Shutting down PATH...")
 
+	// Close live websocket connections FIRST, and with a close handshake.
+	//
+	// server.Shutdown below cannot do this: it explicitly does not close hijacked
+	// connections, and every websocket is hijacked. Without this the process just exits,
+	// every socket dies with its TCP connection, and both peers see an abnormal closure
+	// (1006) — indistinguishable from a crash for the client, and logged as their own
+	// fault by the endpoint. Fleetwide that made every rollout emit a burst of 1006s
+	// across all services within the same second.
+	//
+	// Before backgroundCancel() so the bridges are torn down deliberately rather than
+	// racing a context cancellation that would reach them as a generic failure.
+	//
+	// Its own short budget, well inside the pod's termination grace period: the close
+	// frames are best-effort and a peer that will not answer must not delay exiting.
+	//
+	// Behind an assertion rather than a method on gateway.Protocol: holding live
+	// websocket bridges is a property of the protocol implementation, not of the
+	// interface, and a protocol that holds none needs nothing here.
+	if wsShutdowner, ok := protocol.(websocketShutdowner); ok {
+		wsCtx, wsCancel := context.WithTimeout(context.Background(), websocketShutdownTimeout)
+		wsShutdowner.ShutdownWebsockets(wsCtx, "gateway shutting down")
+		wsCancel()
+	}
+
 	// Cancel background context to stop all background services (pprof, health checks)
 	backgroundCancel()
 

config/config.schema.yaml

@@ -394,6 +394,19 @@ properties:
             description: "Enable/disable health checks."
             type: boolean
             default: false
+          sync_allowance:
+            description: "Default blocks behind the latest an endpoint may be before it counts as out of sync. Per-service entries in local[] override it. 0 disables the sync check."
+            type: integer
+            minimum: 0
+            default: 0
+          max_workers:
+            description: "Maximum concurrent health check workers. Auto-sized from service and check counts when unset."
+            type: integer
+            minimum: 1
+          backend_dedup:
+            description: "Fire ONE relay per unique backend URL each cycle (a rotating representative supplier) and fan the result to the other suppliers sharing that URL. Cuts redundant health-check relay volume by the redundancy factor. WebSocket checks are never deduped. Default: true."
+            type: boolean
+            default: true
           coordination:
             description: "Leader election configuration for multi-instance deployments."
             type: object
@@ -447,6 +460,10 @@ properties:
                   description: "Enable/disable health checks for this service."
                   type: boolean
                   default: true
+                sync_allowance:
+                  description: "Blocks behind the latest an endpoint may be before it counts as out of sync. Overrides the global default for this service. 0 disables the sync check."
+                  type: integer
+                  minimum: 0
                 checks:
                   description: "List of health check configurations."
                   type: array
@@ -461,8 +478,17 @@ properties:
                         description: "Name of the health check."
                         type: string
                       type:
-                        description: "Type of health check (e.g., 'jsonrpc')."
+                        description: "RPC protocol type (must match service's rpc_types): json_rpc, rest, comet_bft, websocket, grpc."
                         type: string
+                        enum: ["json_rpc", "rest", "comet_bft", "websocket", "grpc"]
+                      enabled:
+                        description: "Enable/disable this individual check."
+                        type: boolean
+                      headers:
+                        description: "Request headers."
+                        type: object
+                        additionalProperties:
+                          type: string
                       method:
                         description: "HTTP method (GET, POST)."
                         type: string
@@ -481,10 +507,17 @@ properties:
                       timeout:
                         description: "Request timeout."
                         type: string
+                      archival:
+                        description: "Archival-specific check: only runs on endpoints marked archival."
+                        type: boolean
+                      sync_check:
+                        description: "Enable sync check validation. When true, extracts block height from the response and validates against the service's sync_allowance. Requires sync_allowance > 0."
+                        type: boolean
+                        default: false
                       reputation_signal:
                         description: "Reputation signal to emit on failure."
                         type: string
-                        enum: ["minor_error", "major_error", "critical_error"]
+                        enum: ["minor_error", "major_error", "critical_error", "fatal_error"]
 
       # ====================================
       # UNIFIED SERVICE CONFIGURATION

gateway/health_check_executor.go

@@ -902,7 +902,7 @@ func (e *HealthCheckExecutor) recordCheckResult(
 	// Over-servicing rejections — same no-penalty rule as the request path.
 	// Without this, every health-check probe to an exhausted supplier records a
 	// MajorError signal and pins their reputation at 0 even after the request
-	// path has stopped penalizing them. Production canary observed easy2stake's
+	// path has stopped penalizing them. Production canary observed operator-zeta's
 	// BSC supplier set stuck at score=0 with success-only request-path signals
 	// because the health-check executor was draining them in parallel.
 	if heuristic.IsOverServicedError(checkErr.Error()) {
@@ -1245,8 +1245,21 @@ func (e *HealthCheckExecutor) ExecuteCheckViaProtocol(
 	}
 
 	// Heuristic analysis - detect bad gateway, empty responses, and other error patterns
-	// This runs BEFORE QoS validation to catch issues the basic validation might miss
-	heuristicResult := heuristic.Analyze(responseBody, httpStatusCode, servicePayload.EffectiveRPCType(), "")
+	// This runs BEFORE QoS validation to catch issues the basic validation might miss.
+	//
+	// The method must be threaded through here exactly as the other four call sites do it
+	// (protocol/shannon/context.go, gateway/hedge.go, http_request_context_handle_request.go).
+	// This call passed a hardcoded "" - so even with JSONRPCMethod populated on the payload,
+	// every method-aware rule was blind at this site, and this is the site that decides the
+	// health check's outcome. A CometBFT `health` response
+	// ({"jsonrpc":"2.0","id":1,"result":{}}) is flagged jsonrpc_empty_object_result at
+	// confidence 0.95 unless the method is known, which maps to SignalMajorError below.
+	jsonrpcMethod := servicePayload.JSONRPCMethod
+	if jsonrpcMethod == "" {
+		// REST checks carry no JSON-RPC method; path-aware rules key off Path instead.
+		jsonrpcMethod = servicePayload.Path
+	}
+	heuristicResult := heuristic.Analyze(responseBody, httpStatusCode, servicePayload.EffectiveRPCType(), jsonrpcMethod)
 	if heuristicResult.ShouldRetry {
 		heuristicErr := fmt.Errorf("heuristic detected error: %s - %s", heuristicResult.Reason, heuristicResult.Details)
 		e.logger.Debug().
@@ -1628,12 +1641,25 @@ func (e *HealthCheckExecutor) buildServicePayload(check HealthCheckConfig) proto
 		rpcType = sharedtypes.RPCType_UNKNOWN_RPC
 	}
 
+	// JSONRPCMethod drives the method-aware heuristic checks. User traffic gets it
+	// from QoS parsing; a health check never goes through QoS, so without this it
+	// stayed empty and the heuristic fell back to Path — which for a JSON-RPC check
+	// is "/", matching no method at all.
+	//
+	// That broke the CometBFT carve-out in analyzeJSONRPC: `health` legitimately
+	// answers {"jsonrpc":"2.0","id":1,"result":{}}, and the empty-object rule is
+	// skipped only when rpcType is COMET_BFT or isCometBFTMethod(method) holds.
+	// Cosmos services declare the check as type json_rpc, so neither guard fired and
+	// every healthy node was scored jsonrpc_empty_object_result → retry + minor_error.
+	// Measured 2026-08-06: 16 cosmos services failing this check 100%, ~19% of all
+	// failing health-check signals fleet-wide.
 	return protocol.Payload{
-		Method:  check.Method,
-		Path:    check.Path,
-		Data:    check.Body,
-		Headers: headers,
-		RPCType: rpcType, // Set from aligned health check type
+		Method:        check.Method,
+		Path:          check.Path,
+		Data:          check.Body,
+		Headers:       headers,
+		RPCType:       rpcType, // Set from aligned health check type
+		JSONRPCMethod: extractJSONRPCMethod([]byte(check.Body)),
 	}
 }