diff --git a/.github/workflows/release-artifacts.yml b/.github/workflows/release-artifacts.yml index 75e0f2179..fe93ff69a 100644 --- a/.github/workflows/release-artifacts.yml +++ b/.github/workflows/release-artifacts.yml @@ -22,7 +22,10 @@ jobs: GITHUB_TOKEN: ${{ github.token }} if: startsWith(github.ref, 'refs/tags/v') steps: - - uses: rymndhng/release-on-push-action@master + # Pinned to an immutable commit SHA (v0.28.0) rather than a mutable branch. + # A moving @master ref means a compromised upstream repo would get arbitrary + # code execution in CI with GITHUB_TOKEN. + - uses: rymndhng/release-on-push-action@aebba2bbce07a9474bf95e8710e5ee8a9e922fe2 # v0.28.0 with: bump_version_scheme: patch release_body: ":rocket: Release Notes !:fireworks: " diff --git a/cmd/main.go b/cmd/main.go index 54d996bf6..681158b2d 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -302,6 +302,7 @@ func main() { WebsocketMessageBufferSize: config.GetRouterConfig().WebsocketMessageBufferSize, ObservationQueue: observationQueue, DomainCircuitBreaker: domainCircuitBreaker, + WebsocketConnectionLimiter: gateway.NewWebsocketConnectionLimiter(config.GetRouterConfig().MaxConcurrentWebsocketConnections), } // Until all components are ready, the `/healthz` endpoint will return a 503 Service diff --git a/cmd/metrics.go b/cmd/metrics.go index 7b3227f6a..c81580eb1 100644 --- a/cmd/metrics.go +++ b/cmd/metrics.go @@ -14,6 +14,10 @@ func setupMetricsServer(logger polylog.Logger, addr string) (*metrics.Prometheus Logger: logger, } + // Wire the logger used for the one-time WARN a cardinality guard emits when + // it first saturates (otherwise a capped metric silently goes incomplete). + metrics.SetCardinalityGuardLogger(logger) + if err := pmr.ServeMetrics(addr); err != nil { return nil, err } diff --git a/config/config.schema.yaml b/config/config.schema.yaml index 010d0a68d..3f657908b 100644 --- a/config/config.schema.yaml +++ b/config/config.schema.yaml @@ -663,6 +663,12 @@ properties: websocket_message_buffer_size: description: "Buffer size for websocket messages." type: integer + max_concurrent_websocket_connections: + description: "Maximum number of concurrent live websocket connections per gateway pod (defense-in-depth against goroutine/FD exhaustion). Defaults to 10000; set to a negative value to disable the limit." + type: integer + max_request_body_bytes: + description: "Maximum size in bytes of an HTTP request body PATH will read into memory (prevents OOM from oversized bodies). Defaults to 78643200 (75MB); set to a negative value to disable the limit." + type: integer # Concurrency Configuration (optional) concurrency_config: diff --git a/config/config_test.go b/config/config_test.go index e064f4058..b07be29b7 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -93,13 +93,15 @@ func Test_LoadGatewayConfigFromYAML(t *testing.T) { }, }, Router: RouterConfig{ - Port: defaultPort, - MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, - ReadTimeout: 30 * time.Second, // Matches example config - WriteTimeout: 30 * time.Second, // Matches example config - IdleTimeout: 120 * time.Second, // Matches example config - SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, - WebsocketMessageBufferSize: 8192, // Matches example config + Port: defaultPort, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: 30 * time.Second, // Matches example config + WriteTimeout: 30 * time.Second, // Matches example config + IdleTimeout: 120 * time.Second, // Matches example config + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: 8192, // Matches example config + MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections, + MaxRequestBodyBytes: defaultMaxRequestBodyBytes, }, Logger: LoggerConfig{ Level: "info", // Matches example config @@ -191,13 +193,15 @@ logger_config: }, }, Router: RouterConfig{ - Port: defaultPort, - MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, - ReadTimeout: defaultHTTPServerReadTimeout, - WriteTimeout: defaultHTTPServerWriteTimeout, - IdleTimeout: defaultHTTPServerIdleTimeout, - SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, - WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + Port: defaultPort, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: defaultHTTPServerReadTimeout, + WriteTimeout: defaultHTTPServerWriteTimeout, + IdleTimeout: defaultHTTPServerIdleTimeout, + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections, + MaxRequestBodyBytes: defaultMaxRequestBodyBytes, }, Logger: LoggerConfig{ Level: "debug", @@ -404,13 +408,15 @@ logger_config: }, }, Router: RouterConfig{ - Port: defaultPort, - MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, - ReadTimeout: defaultHTTPServerReadTimeout, - WriteTimeout: defaultHTTPServerWriteTimeout, - IdleTimeout: defaultHTTPServerIdleTimeout, - SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, - WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + Port: defaultPort, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: defaultHTTPServerReadTimeout, + WriteTimeout: defaultHTTPServerWriteTimeout, + IdleTimeout: defaultHTTPServerIdleTimeout, + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections, + MaxRequestBodyBytes: defaultMaxRequestBodyBytes, }, Logger: LoggerConfig{ Level: "info", diff --git a/config/router.go b/config/router.go index 313789948..484b08c14 100644 --- a/config/router.go +++ b/config/router.go @@ -31,6 +31,25 @@ const ( // Reduced from 1000 to prevent OOM. At 100: 100 × ~3KB × 100 connections = ~30MB. // Can be tuned based on expected concurrent websocket connections and message frequency. defaultWebsocketMessageBufferSize = 100 + + // defaultMaxRequestBodyBytes caps the size of an HTTP request body PATH will + // read into memory. Without a limit, a single request with a multi-GB body + // OOMs the process (the body is buffered whole, and observation/detection + // paths read it before any downstream limit applies). 75MB leaves generous + // headroom for the largest legitimate payloads (big batches, large + // eth_getLogs/debug traces, contract deploys) while still bounding a single + // request's allocation. Reads past the limit fail (request rejected) rather + // than growing memory without bound. + defaultMaxRequestBodyBytes = 75 * 1024 * 1024 + + // defaultMaxConcurrentWebsocketConnections caps concurrent live websocket + // connections per gateway pod (defense-in-depth against goroutine/FD + // exhaustion). Each connection costs ~5-7 goroutines + 2 sockets + buffers + // (~50-100KB), so 10000 ≈ 0.5-1GB worst case — a generous ceiling well above + // normal load that still bounds catastrophic runaway. Tune to observed + // concurrent websocket load (Polygon is the primary websocket service). + // Set to a negative value in config to disable the limit entirely. + defaultMaxConcurrentWebsocketConnections = 10000 ) /* --------------------------------- Router Config Struct -------------------------------- */ @@ -48,6 +67,13 @@ type RouterConfig struct { // Larger values use more memory but can handle higher message throughput. // Default: 50 (prevents OOM while maintaining reasonable throughput) WebsocketMessageBufferSize int `yaml:"websocket_message_buffer_size"` + // MaxConcurrentWebsocketConnections caps the number of concurrent live + // websocket connections per gateway pod. Default: 10000. A negative value + // disables the limit. + MaxConcurrentWebsocketConnections int `yaml:"max_concurrent_websocket_connections"` + // MaxRequestBodyBytes caps the size (in bytes) of an HTTP request body PATH + // will read into memory. Default: 10MB. A negative value disables the limit. + MaxRequestBodyBytes int64 `yaml:"max_request_body_bytes"` } /* --------------------------------- Router Config Private Helpers -------------------------------- */ @@ -76,6 +102,14 @@ func (c *RouterConfig) hydrateRouterDefaults() error { if c.WebsocketMessageBufferSize == 0 { c.WebsocketMessageBufferSize = defaultWebsocketMessageBufferSize } + // Only an unset (zero) value takes the default; a negative value is preserved + // so operators can explicitly disable the limit. + if c.MaxConcurrentWebsocketConnections == 0 { + c.MaxConcurrentWebsocketConnections = defaultMaxConcurrentWebsocketConnections + } + if c.MaxRequestBodyBytes == 0 { + c.MaxRequestBodyBytes = defaultMaxRequestBodyBytes + } if c.SystemOverheadAllowanceDuration >= c.ReadTimeout || c.SystemOverheadAllowanceDuration >= c.WriteTimeout { return fmt.Errorf("system overhead allowance duration %v must be less than read timeout %v and write timeout %v", c.SystemOverheadAllowanceDuration, c.ReadTimeout, c.WriteTimeout) } diff --git a/config/router_test.go b/config/router_test.go index 800342369..c8e4d2caa 100644 --- a/config/router_test.go +++ b/config/router_test.go @@ -61,13 +61,15 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) { name: "should set all defaults", cfg: RouterConfig{}, want: RouterConfig{ - Port: defaultPort, - MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, - ReadTimeout: defaultHTTPServerReadTimeout, - WriteTimeout: defaultHTTPServerWriteTimeout, - IdleTimeout: defaultHTTPServerIdleTimeout, - SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, - WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + Port: defaultPort, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: defaultHTTPServerReadTimeout, + WriteTimeout: defaultHTTPServerWriteTimeout, + IdleTimeout: defaultHTTPServerIdleTimeout, + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections, + MaxRequestBodyBytes: defaultMaxRequestBodyBytes, }, wantErr: false, }, @@ -77,13 +79,15 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) { Port: 8080, }, want: RouterConfig{ - Port: 8080, - MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, - ReadTimeout: defaultHTTPServerReadTimeout, - WriteTimeout: defaultHTTPServerWriteTimeout, - IdleTimeout: defaultHTTPServerIdleTimeout, - SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, - WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + Port: 8080, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: defaultHTTPServerReadTimeout, + WriteTimeout: defaultHTTPServerWriteTimeout, + IdleTimeout: defaultHTTPServerIdleTimeout, + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections, + MaxRequestBodyBytes: defaultMaxRequestBodyBytes, }, wantErr: false, }, @@ -111,13 +115,51 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) { WebsocketMessageBufferSize: 500, }, want: RouterConfig{ - Port: defaultPort, - MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, - ReadTimeout: defaultHTTPServerReadTimeout, - WriteTimeout: defaultHTTPServerWriteTimeout, - IdleTimeout: defaultHTTPServerIdleTimeout, - SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, - WebsocketMessageBufferSize: 500, // Custom value should be preserved + Port: defaultPort, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: defaultHTTPServerReadTimeout, + WriteTimeout: defaultHTTPServerWriteTimeout, + IdleTimeout: defaultHTTPServerIdleTimeout, + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: 500, // Custom value should be preserved + MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections, + MaxRequestBodyBytes: defaultMaxRequestBodyBytes, + }, + wantErr: false, + }, + { + name: "should preserve a negative max websocket connections (limit disabled)", + cfg: RouterConfig{ + MaxConcurrentWebsocketConnections: -1, + }, + want: RouterConfig{ + Port: defaultPort, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: defaultHTTPServerReadTimeout, + WriteTimeout: defaultHTTPServerWriteTimeout, + IdleTimeout: defaultHTTPServerIdleTimeout, + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + MaxConcurrentWebsocketConnections: -1, // Negative preserved: limit disabled + MaxRequestBodyBytes: defaultMaxRequestBodyBytes, + }, + wantErr: false, + }, + { + name: "should preserve a negative max request body bytes (limit disabled)", + cfg: RouterConfig{ + MaxRequestBodyBytes: -1, + }, + want: RouterConfig{ + Port: defaultPort, + MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes, + ReadTimeout: defaultHTTPServerReadTimeout, + WriteTimeout: defaultHTTPServerWriteTimeout, + IdleTimeout: defaultHTTPServerIdleTimeout, + SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration, + WebsocketMessageBufferSize: defaultWebsocketMessageBufferSize, + MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections, + MaxRequestBodyBytes: -1, // Negative preserved: limit disabled }, wantErr: false, }, diff --git a/gateway/gateway.go b/gateway/gateway.go index c3c1dd76b..134f0938b 100644 --- a/gateway/gateway.go +++ b/gateway/gateway.go @@ -73,6 +73,10 @@ type Gateway struct { // on initial attempts for a TTL window, avoiding wasted relay attempts. // Optional - if nil, no cross-pod domain circuit breaking occurs. DomainCircuitBreaker *DomainCircuitBreaker + + // WebsocketConnectionLimiter bounds the number of concurrent live websocket + // connections held open by this gateway. Optional - if nil, no limit is applied. + WebsocketConnectionLimiter *WebsocketConnectionLimiter } // HandleServiceRequest implements PATH gateway's service request processing. @@ -207,6 +211,26 @@ func (g Gateway) handleWebSocketRequest( ) { logger := g.Logger.With("method", "handleWebSocketRequest") + // Bound the number of concurrent live websocket connections (defense-in-depth + // against goroutine/FD exhaustion). Reserve a slot up front; release it either + // immediately if setup fails, or when the connection terminates if it succeeds. + if !g.WebsocketConnectionLimiter.Acquire() { + logger.Warn(). + Int64("active_websocket_connections", g.WebsocketConnectionLimiter.Active()). + Msg("⚠️ rejecting websocket connection: concurrent connection limit reached") + http.Error(w, "too many concurrent websocket connections", http.StatusServiceUnavailable) + return + } + // established gates the slot release: while false, any early return releases the + // slot immediately (connection never went live); once true, the slot is released + // asynchronously when the connection terminates (see the goroutine below). + established := false + defer func() { + if !established { + g.WebsocketConnectionLimiter.Release() + } + }() + // Use a cancellable background context for the long-lived Websocket connection lifecycle. // Unlike HTTP requests, Websocket connections are long-lived and should not be tied to the HTTP request context. // The HTTP request context gets canceled when the HTTP handler returns, which would stop the observation listener. @@ -233,6 +257,9 @@ func (g Gateway) handleWebSocketRequest( // Note: We do NOT close messageObservationsChan here because Websocket connections // outlive the HTTP handler. The channel will be closed when the Websocket actually disconnects. messageObservationsChan: make(chan *observation.RequestResponseObservations, websocketBufferSize), + // ready is closed once the protocol context is assigned; message + // processors block on it to avoid the setup-window race. + ready: make(chan struct{}), } // Initialize the websocket request context using the HTTP request. @@ -268,10 +295,21 @@ func (g Gateway) handleWebSocketRequest( return } - // At this point, the Websocket connection has terminated and the bridge has shut down. - // The defer block above will now execute and broadcast connection observations with: - // - Complete connection duration (from establishment to termination) - // - Final connection status and termination reason - // This ensures we send only ONE connection observation per Websocket connection. - logger.Debug().Msg("Websocket connection and bridge shutdown complete, ready to broadcast final observations") + // The connection is now live and owned by background bridge/listener goroutines. + // Hand the concurrency slot off to the connection's lifecycle: websocketCtx is + // canceled when the connection fully terminates (websocketRequestContext.cancelCtx), + // so release the slot then. established=true disarms the immediate-release defer, + // guaranteeing exactly one Release per successful Acquire. + established = true + go func() { + <-websocketCtx.Done() + g.WebsocketConnectionLimiter.Release() + }() + + // The connection is now live and self-managed by the background bridge and + // listener goroutines; this handler returns immediately (the upgraded conn is + // detached from the HTTP request lifecycle via websocketCtx). Connection + // establishment/closure observations are emitted by those goroutines over the + // lifetime of the connection, not here. + logger.Debug().Msg("Websocket connection established; bridge running in background") } diff --git a/gateway/http_request_context_handle_request.go b/gateway/http_request_context_handle_request.go index dd6a247ad..3003d7daf 100644 --- a/gateway/http_request_context_handle_request.go +++ b/gateway/http_request_context_handle_request.go @@ -919,6 +919,33 @@ func (rc *requestContext) handleBatchRelayRequest(payloads []protocol.Payload) e startTime := time.Now() rpcType := payloads[0].RPCType + // Resolve the concurrency limits. max_batch_payloads supports a per-service + // override; max_concurrent_relays is global only. + maxBatchPayloads := rc.protocol.GetConcurrencyConfig().MaxBatchPayloads + maxConcurrentRelays := rc.protocol.GetConcurrencyConfig().MaxConcurrentRelays + if cc := rc.getConcurrencyConfigForService(); cc != nil && cc.MaxBatchPayloads != nil { + maxBatchPayloads = *cc.MaxBatchPayloads + } + + // Enforce the batch cap BEFORE spawning any goroutines. Previously the cap was + // only checked deep inside each spawned goroutine, so an oversized batch spawned + // one goroutine per payload (each fanning out further via retry/hedge) before any + // rejection — a goroutine bomb. Reject early instead. + if maxBatchPayloads > 0 && len(payloads) > maxBatchPayloads { + err := fmt.Errorf("batch size %d exceeds the configured max_batch_payloads %d", len(payloads), maxBatchPayloads) + logger.Error().Err(err).Msg("❌ rejecting oversized batch request") + return err + } + + // Bound the number of payload goroutines alive at once so the batch fan-out + // (amplified by per-payload retry/hedge) cannot exhaust goroutines/FDs. The + // slot is acquired BEFORE spawning, so at most maxConcurrentRelays payload + // goroutines exist concurrently. nil = unbounded (limit disabled). + var sem chan struct{} + if maxConcurrentRelays > 0 { + sem = make(chan struct{}, maxConcurrentRelays) + } + // Process each payload in parallel resultChan := make(chan batchPayloadResult, len(payloads)) var wg sync.WaitGroup @@ -935,9 +962,15 @@ func (rc *requestContext) handleBatchRelayRequest(payloads []protocol.Payload) e }()). Msg("Processing batch payload") + if sem != nil { + sem <- struct{}{} // blocks here (not in a spawned goroutine) once at capacity + } wg.Add(1) go func(index int, p protocol.Payload) { defer wg.Done() + if sem != nil { + defer func() { <-sem }() + } response, err := rc.processSinglePayloadWithRetry(p, index, rpcType, logger) resultChan <- batchPayloadResult{ index: index, diff --git a/gateway/unified_service_config.go b/gateway/unified_service_config.go index a6a1e4d12..6b485b789 100644 --- a/gateway/unified_service_config.go +++ b/gateway/unified_service_config.go @@ -6,6 +6,7 @@ package gateway import ( "fmt" + "sync" "time" "github.com/pokt-network/poktroll/pkg/polylog" @@ -204,12 +205,12 @@ type ServiceHealthCheckOverride struct { // ensuring behind-on-block suppliers are correctly filtered even when // all session endpoints are equally stale. type ExternalBlockSource struct { - URL string `yaml:"url"` // RPC endpoint URL - Type string `yaml:"type,omitempty"` // "jsonrpc" (default) or "rest". REST uses GET, JSON-RPC uses POST. - Method string `yaml:"method,omitempty"` // JSON-RPC method. Default: "eth_blockNumber". Use "status" for Cosmos, "getBlockHeight" for Solana. - Path string `yaml:"path,omitempty"` // Request path appended to URL. Default: "/" - Interval time.Duration `yaml:"interval,omitempty"` // Poll interval. Default: 30s - Timeout time.Duration `yaml:"timeout,omitempty"` // HTTP timeout. Default: 5s + URL string `yaml:"url"` // RPC endpoint URL + Type string `yaml:"type,omitempty"` // "jsonrpc" (default) or "rest". REST uses GET, JSON-RPC uses POST. + Method string `yaml:"method,omitempty"` // JSON-RPC method. Default: "eth_blockNumber". Use "status" for Cosmos, "getBlockHeight" for Solana. + Path string `yaml:"path,omitempty"` // Request path appended to URL. Default: "/" + Interval time.Duration `yaml:"interval,omitempty"` // Poll interval. Default: 30s + Timeout time.Duration `yaml:"timeout,omitempty"` // HTTP timeout. Default: 5s GracePeriod time.Duration `yaml:"grace_period,omitempty"` // Wait after startup before applying external floor. Default: 60s } @@ -222,36 +223,36 @@ type ServiceFallbackConfig struct { // ServiceDefaults contains default settings inherited by all services. type ServiceDefaults struct { - Type ServiceType `yaml:"type,omitempty"` - RPCTypes []string `yaml:"rpc_types,omitempty"` - LatencyProfile string `yaml:"latency_profile,omitempty"` - ReputationConfig ServiceReputationConfig `yaml:"reputation_config,omitempty"` - Latency ServiceLatencyConfig `yaml:"latency,omitempty"` - TieredSelection ServiceTieredSelectionConfig `yaml:"tiered_selection,omitempty"` - Probation ServiceProbationConfig `yaml:"probation,omitempty"` - RetryConfig ServiceRetryConfig `yaml:"retry_config,omitempty"` - ObservationPipeline ServiceObservationConfig `yaml:"observation_pipeline,omitempty"` - ConcurrencyConfig ServiceConcurrencyConfig `yaml:"concurrency_config,omitempty"` - TimeoutConfig ServiceTimeoutConfig `yaml:"timeout_config,omitempty"` - ActiveHealthChecks ServiceHealthCheckOverride `yaml:"active_health_checks,omitempty"` - ExternalBlockSources []ExternalBlockSource `yaml:"external_block_sources,omitempty"` + Type ServiceType `yaml:"type,omitempty"` + RPCTypes []string `yaml:"rpc_types,omitempty"` + LatencyProfile string `yaml:"latency_profile,omitempty"` + ReputationConfig ServiceReputationConfig `yaml:"reputation_config,omitempty"` + Latency ServiceLatencyConfig `yaml:"latency,omitempty"` + TieredSelection ServiceTieredSelectionConfig `yaml:"tiered_selection,omitempty"` + Probation ServiceProbationConfig `yaml:"probation,omitempty"` + RetryConfig ServiceRetryConfig `yaml:"retry_config,omitempty"` + ObservationPipeline ServiceObservationConfig `yaml:"observation_pipeline,omitempty"` + ConcurrencyConfig ServiceConcurrencyConfig `yaml:"concurrency_config,omitempty"` + TimeoutConfig ServiceTimeoutConfig `yaml:"timeout_config,omitempty"` + ActiveHealthChecks ServiceHealthCheckOverride `yaml:"active_health_checks,omitempty"` + ExternalBlockSources []ExternalBlockSource `yaml:"external_block_sources,omitempty"` } // ServiceConfig defines configuration for a single service. type ServiceConfig struct { - ID protocol.ServiceID `yaml:"id"` - Type ServiceType `yaml:"type,omitempty"` - RPCTypes []string `yaml:"rpc_types,omitempty"` - RPCTypeFallbacks map[string]string `yaml:"rpc_type_fallbacks,omitempty"` - LatencyProfile string `yaml:"latency_profile,omitempty"` - ReputationConfig *ServiceReputationConfig `yaml:"reputation_config,omitempty"` - Latency *ServiceLatencyConfig `yaml:"latency,omitempty"` - TieredSelection *ServiceTieredSelectionConfig `yaml:"tiered_selection,omitempty"` - Probation *ServiceProbationConfig `yaml:"probation,omitempty"` - RetryConfig *ServiceRetryConfig `yaml:"retry_config,omitempty"` - ObservationPipeline *ServiceObservationConfig `yaml:"observation_pipeline,omitempty"` - ConcurrencyConfig *ServiceConcurrencyConfig `yaml:"concurrency_config,omitempty"` - TimeoutConfig *ServiceTimeoutConfig `yaml:"timeout_config,omitempty"` + ID protocol.ServiceID `yaml:"id"` + Type ServiceType `yaml:"type,omitempty"` + RPCTypes []string `yaml:"rpc_types,omitempty"` + RPCTypeFallbacks map[string]string `yaml:"rpc_type_fallbacks,omitempty"` + LatencyProfile string `yaml:"latency_profile,omitempty"` + ReputationConfig *ServiceReputationConfig `yaml:"reputation_config,omitempty"` + Latency *ServiceLatencyConfig `yaml:"latency,omitempty"` + TieredSelection *ServiceTieredSelectionConfig `yaml:"tiered_selection,omitempty"` + Probation *ServiceProbationConfig `yaml:"probation,omitempty"` + RetryConfig *ServiceRetryConfig `yaml:"retry_config,omitempty"` + ObservationPipeline *ServiceObservationConfig `yaml:"observation_pipeline,omitempty"` + ConcurrencyConfig *ServiceConcurrencyConfig `yaml:"concurrency_config,omitempty"` + TimeoutConfig *ServiceTimeoutConfig `yaml:"timeout_config,omitempty"` Fallback *ServiceFallbackConfig `yaml:"fallback,omitempty"` HealthChecks *ServiceHealthCheckOverride `yaml:"health_checks,omitempty"` ExternalBlockSources []ExternalBlockSource `yaml:"external_block_sources,omitempty"` @@ -269,6 +270,41 @@ type UnifiedServicesConfig struct { LatencyProfiles map[string]LatencyProfileConfig `yaml:"latency_profiles,omitempty"` Defaults ServiceDefaults `yaml:"defaults,omitempty"` Services []ServiceConfig `yaml:"services,omitempty"` + + // servicesMu guards concurrent access to Services. The external health-check + // refresh mutates Services at runtime (SetServiceSyncAllowance) while every + // request reads it (GetServiceConfig etc.); without synchronization the + // append path races a concurrent slice-header read → torn header → panic. + // It is a pointer (not a value sync.RWMutex) so this struct — which is + // returned by value from LoadGatewayConfigFromYAML during load — stays + // copylock-safe. Initialized in HydrateDefaults (called once at startup + // before any concurrency); the lock helpers no-op if it is nil so + // directly-constructed configs in tests remain usable. + servicesMu *sync.RWMutex +} + +// rlock/runlock/lock/unlock guard Services access. They are nil-safe: a config +// built without HydrateDefaults (e.g. in tests) is used single-threaded, so +// skipping the lock there is safe. +func (c *UnifiedServicesConfig) rlock() { + if c.servicesMu != nil { + c.servicesMu.RLock() + } +} +func (c *UnifiedServicesConfig) runlock() { + if c.servicesMu != nil { + c.servicesMu.RUnlock() + } +} +func (c *UnifiedServicesConfig) lock() { + if c.servicesMu != nil { + c.servicesMu.Lock() + } +} +func (c *UnifiedServicesConfig) unlock() { + if c.servicesMu != nil { + c.servicesMu.Unlock() + } } // Validate validates the UnifiedServicesConfig. @@ -341,6 +377,11 @@ func IsBuiltInLatencyProfile(name string) bool { // HydrateDefaults applies default values to UnifiedServicesConfig. func (c *UnifiedServicesConfig) HydrateDefaults() { + // Initialize the Services mutex once, before the config is shared with the + // request path and the health-check refresh goroutine. + if c.servicesMu == nil { + c.servicesMu = &sync.RWMutex{} + } if c.Defaults.Type == "" { c.Defaults.Type = ServiceTypePassthrough } @@ -506,9 +547,16 @@ func (c *UnifiedServicesConfig) HydrateDefaults() { // GetServiceConfig returns the configuration for a specific service. func (c *UnifiedServicesConfig) GetServiceConfig(serviceID protocol.ServiceID) *ServiceConfig { + c.rlock() + defer c.runlock() for i := range c.Services { if c.Services[i].ID == serviceID { - return &c.Services[i] + // Return a copy, not &c.Services[i]: a pointer into the slice would + // let the caller read the element after the lock is released, racing + // a concurrent SetServiceSyncAllowance. The writer copy-on-writes the + // HealthChecks pointer, so this shallow copy stays stable. + cfg := c.Services[i] + return &cfg } } return nil @@ -536,6 +584,8 @@ func (c *UnifiedServicesConfig) GetServiceRPCTypes(serviceID protocol.ServiceID) // GetConfiguredServiceIDs returns a list of all configured service IDs. func (c *UnifiedServicesConfig) GetConfiguredServiceIDs() []protocol.ServiceID { + c.rlock() + defer c.runlock() ids := make([]protocol.ServiceID, len(c.Services)) for i, svc := range c.Services { ids[i] = svc.ID @@ -803,12 +853,21 @@ func (c *UnifiedServicesConfig) GetSyncAllowanceForService(serviceID protocol.Se // This is called when external health check rules are loaded to propagate // the sync_allowance to the unified config for use by the QoS layer. func (c *UnifiedServicesConfig) SetServiceSyncAllowance(serviceID protocol.ServiceID, syncAllowance int) { + c.lock() + defer c.unlock() for i := range c.Services { if c.Services[i].ID == serviceID { - if c.Services[i].HealthChecks == nil { - c.Services[i].HealthChecks = &ServiceHealthCheckOverride{} + // Copy-on-write the HealthChecks pointer rather than mutating the + // existing one in place: a concurrent reader may hold a copy of this + // ServiceConfig (with the old pointer), and mutating that shared + // override in place would race its read. Publishing a fresh pointer + // leaves the old one immutable. + hc := ServiceHealthCheckOverride{} + if c.Services[i].HealthChecks != nil { + hc = *c.Services[i].HealthChecks } - c.Services[i].HealthChecks.SyncAllowance = &syncAllowance + hc.SyncAllowance = &syncAllowance + c.Services[i].HealthChecks = &hc return } } diff --git a/gateway/websocket_limiter.go b/gateway/websocket_limiter.go new file mode 100644 index 000000000..05a68bb82 --- /dev/null +++ b/gateway/websocket_limiter.go @@ -0,0 +1,69 @@ +package gateway + +import "sync/atomic" + +// WebsocketConnectionLimiter bounds the number of concurrent live websocket +// connections a gateway will hold open. +// +// Each websocket connection is long-lived and costs ~5-7 goroutines plus two +// TCP sockets and read/write buffers (~50-100KB total). Without a ceiling, a +// flood of upgrades — or a slow drain of never-closing connections — grows +// goroutines and file descriptors without bound and can exhaust the pod. Envoy +// rate-limits connection *attempts* at the edge, but nothing else bounds the +// count of *live* connections inside PATH; this is the internal backstop. +// +// A nil *WebsocketConnectionLimiter means limiting is disabled: all methods are +// nil-safe and Acquire always succeeds. This keeps the limiter optional without +// forcing every caller/test to construct one. +type WebsocketConnectionLimiter struct { + max int64 + active atomic.Int64 +} + +// NewWebsocketConnectionLimiter returns a limiter capping concurrent websocket +// connections at max. A max <= 0 returns nil, which disables limiting. +func NewWebsocketConnectionLimiter(max int) *WebsocketConnectionLimiter { + if max <= 0 { + return nil + } + return &WebsocketConnectionLimiter{max: int64(max)} +} + +// Acquire reserves a connection slot. It returns true if a slot was reserved +// (the caller must later call Release) and false if the limiter is already at +// capacity (the caller must reject the connection and must NOT call Release). +// A nil limiter always returns true (limiting disabled). +func (l *WebsocketConnectionLimiter) Acquire() bool { + if l == nil { + return true + } + // Lock-free CAS loop: only increment when strictly below the cap. Unlike an + // optimistic add-then-rollback, this never lets the counter overshoot the cap + // even transiently, so a concurrent Active() read can never exceed max. + for { + cur := l.active.Load() + if cur >= l.max { + return false + } + if l.active.CompareAndSwap(cur, cur+1) { + return true + } + } +} + +// Release frees a slot previously reserved by a successful Acquire. It must be +// called exactly once per successful Acquire. A nil limiter is a no-op. +func (l *WebsocketConnectionLimiter) Release() { + if l == nil { + return + } + l.active.Add(-1) +} + +// Active returns the number of currently reserved slots. A nil limiter reports 0. +func (l *WebsocketConnectionLimiter) Active() int64 { + if l == nil { + return 0 + } + return l.active.Load() +} diff --git a/gateway/websocket_limiter_test.go b/gateway/websocket_limiter_test.go new file mode 100644 index 000000000..c97698c7f --- /dev/null +++ b/gateway/websocket_limiter_test.go @@ -0,0 +1,71 @@ +package gateway + +import ( + "sync" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_WebsocketConnectionLimiter_NilIsDisabled(t *testing.T) { + c := require.New(t) + + // A max <= 0 yields a nil limiter (limiting disabled). + var l *WebsocketConnectionLimiter = NewWebsocketConnectionLimiter(0) + c.Nil(l) + c.Nil(NewWebsocketConnectionLimiter(-5)) + + // All methods are nil-safe: Acquire always succeeds, Release/Active are no-ops. + c.True(l.Acquire()) + c.True(l.Acquire()) + l.Release() + c.Equal(int64(0), l.Active()) +} + +func Test_WebsocketConnectionLimiter_CapAndRelease(t *testing.T) { + c := require.New(t) + + l := NewWebsocketConnectionLimiter(2) + c.NotNil(l) + + c.True(l.Acquire(), "first slot should be acquired") + c.True(l.Acquire(), "second slot should be acquired") + c.Equal(int64(2), l.Active()) + + // At capacity: further acquires are rejected and must not change the count. + c.False(l.Acquire(), "third slot should be rejected at capacity") + c.Equal(int64(2), l.Active(), "rejected acquire must not leak a slot") + + // Releasing frees a slot so the next acquire succeeds. + l.Release() + c.Equal(int64(1), l.Active()) + c.True(l.Acquire(), "slot should be available again after release") + c.Equal(int64(2), l.Active()) +} + +func Test_WebsocketConnectionLimiter_ConcurrentNeverExceedsCap(t *testing.T) { + c := require.New(t) + + const max = 8 + l := NewWebsocketConnectionLimiter(max) + + // Hammer Acquire from many goroutines; the number that succeed must never + // exceed the cap, and Active must never read above the cap. + var wg sync.WaitGroup + var granted atomic.Int64 + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if l.Acquire() { + granted.Add(1) + } + c.LessOrEqual(l.Active(), int64(max)) + }() + } + wg.Wait() + + c.Equal(int64(max), granted.Load(), "exactly cap slots should be granted") + c.Equal(int64(max), l.Active()) +} diff --git a/gateway/websocket_request_context.go b/gateway/websocket_request_context.go index 8e699650f..371ff8d57 100644 --- a/gateway/websocket_request_context.go +++ b/gateway/websocket_request_context.go @@ -70,6 +70,16 @@ type websocketRequestContext struct { protocolCtxMu sync.RWMutex protocolCtx ProtocolRequestContextWebsocket + // ready is closed once protocolCtx has been assigned. Message processors + // block on it before touching protocolCtx: the bridge can deliver a frame + // (e.g. an endpoint-speaks-first backend pushing on connect) before + // BuildWebsocketRequestContextForEndpoint returns and protocolCtx is set. + // Gating here — rather than rejecting a nil protocolCtx — prevents that + // setup-window frame from failing message processing and tearing the whole + // connection down (close 1011). The wait is bounded by the connection + // context, which is canceled on every failure path. + ready chan struct{} + // gatewayObservations stores gateway related observations. gatewayObservations *observation.GatewayObservations @@ -232,6 +242,11 @@ func (wrc *websocketRequestContext) buildProtocolContextAndStartBridge( wrc.protocolCtxMu.Lock() wrc.protocolCtx = protocolCtx wrc.protocolCtxMu.Unlock() + // Unblock any message processors waiting on setup. Closing happens-after the + // assignment above, so a processor that observes the close is guaranteed to + // read the assigned protocolCtx. Only ever closed here (the single success + // path), so no double-close: failure paths cancel the context instead. + close(wrc.ready) logger.Debug().Msgf("Successfully built protocol context and started bridge for websocket endpoint: %s", selectedEndpoint) return connectionObservationChan, nil } @@ -251,17 +266,28 @@ func (wrc *websocketRequestContext) ProcessClientWebsocketMessage(msgData []byte logger.Debug().Msgf("received message from client: %s", string(msgData)) - // Acquire read lock to safely check protocolCtx. The bridge goroutine can - // deliver messages before BuildWebsocketRequestContextForEndpoint returns and - // assigns protocolCtx. Without this lock, a concurrent unsynchronized read of - // the interface field can observe a partially-written value (type pointer set, - // data pointer nil) causing a nil-receiver panic in the Shannon layer. + // Wait for protocol context setup to complete before processing. The bridge + // can deliver a message before BuildWebsocketRequestContextForEndpoint returns + // and protocolCtx is assigned; gating here (rather than rejecting) keeps a + // message that arrives in the setup window from tearing down the connection. + // The wait is bounded by the connection context, canceled on every failure path. + select { + case <-wrc.ready: + // Setup complete: protocolCtx is assigned. + case <-wrc.context.Done(): + logger.Error().Msg("❌ websocket context canceled before setup completed") + return nil, errWebsocketProtocolContextNotReady + } + + // Acquire read lock to safely read protocolCtx. The lock still guards against + // a torn read of the interface value; the ready gate above guarantees the + // value has been assigned by the time we get here. wrc.protocolCtxMu.RLock() pCtx := wrc.protocolCtx wrc.protocolCtxMu.RUnlock() if pCtx == nil { - logger.Error().Msg("❌ protocol context is nil — message arrived before websocket setup completed") + logger.Error().Msg("❌ SHOULD NEVER HAPPEN: protocol context is nil after ready signal") return nil, errWebsocketProtocolContextNotReady } @@ -281,13 +307,26 @@ func (wrc *websocketRequestContext) ProcessClientWebsocketMessage(msgData []byte func (wrc *websocketRequestContext) ProcessEndpointWebsocketMessage(msgData []byte) ([]byte, *observation.RequestResponseObservations, error) { logger := wrc.logger.With("method", "ProcessEndpointWebsocketMessage") - // Acquire read lock — same race condition guard as ProcessClientWebsocketMessage. + // Wait for protocol context setup to complete — same setup-window guard as + // ProcessClientWebsocketMessage. This is the frame that actually triggered the + // original defect: an endpoint-speaks-first backend pushes a frame on connect, + // which lands here before protocolCtx is assigned. + select { + case <-wrc.ready: + // Setup complete: protocolCtx is assigned. + case <-wrc.context.Done(): + logger.Error().Msg("❌ websocket context canceled before setup completed") + return nil, nil, errWebsocketProtocolContextNotReady + } + + // Acquire read lock to safely read protocolCtx (torn-read guard; the ready + // gate above guarantees the value has been assigned). wrc.protocolCtxMu.RLock() pCtx := wrc.protocolCtx wrc.protocolCtxMu.RUnlock() if pCtx == nil { - logger.Error().Msg("❌ protocol context is nil — message arrived before websocket setup completed") + logger.Error().Msg("❌ SHOULD NEVER HAPPEN: protocol context is nil after ready signal") return nil, nil, errWebsocketProtocolContextNotReady } @@ -436,6 +475,15 @@ func (wrc *websocketRequestContext) handleConnectionObservation(protocolObs *pro // BroadcastMessageObservations delivers the collected details regarding all aspects // of the websocket message to all the interested parties. +// +// This runs synchronously in the caller's goroutine (listenForMessageNotifications), +// which already processes messageObservationsChan sequentially — one per connection. +// It is deliberately NOT wrapped in a per-message goroutine: doing so spawned an +// unbounded number of goroutines under a high-frequency subscription (e.g. a +// Polygon logs/newHeads firehose), each racing on the shared observation state. +// Running inline bounds concurrency to one per connection and lets the buffered +// messageObservationsChan (plus the bridge's non-blocking, drop-on-full send) +// absorb bursts. func (wrc *websocketRequestContext) BroadcastMessageObservations( messageObservations *observation.RequestResponseObservations, ) { @@ -445,32 +493,29 @@ func (wrc *websocketRequestContext) BroadcastMessageObservations( return } - // observation-related tasks are called in Goroutines to avoid potentially blocking the handler. - go func() { - if protocolObservations := messageObservations.GetProtocol(); protocolObservations != nil { - err := wrc.protocol.ApplyWebSocketObservations(protocolObservations) - if err != nil { - wrc.logger.Warn().Err(err).Msg("error applying protocol observations for websocket.") - } + if protocolObservations := messageObservations.GetProtocol(); protocolObservations != nil { + err := wrc.protocol.ApplyWebSocketObservations(protocolObservations) + if err != nil { + wrc.logger.Warn().Err(err).Msg("error applying protocol observations for websocket.") } + } - // Apply QoS observations - if qosObservations := messageObservations.GetQos(); qosObservations != nil { - if err := wrc.serviceQoS.ApplyObservations(qosObservations); err != nil { - wrc.logger.Warn().Err(err).Msg("error applying QoS observations for websocket.") - } + // Apply QoS observations + if qosObservations := messageObservations.GetQos(); qosObservations != nil { + if err := wrc.serviceQoS.ApplyObservations(qosObservations); err != nil { + wrc.logger.Warn().Err(err).Msg("error applying QoS observations for websocket.") } + } - // Prepare and publish observations to both the metrics and data reporters. - observations := &observation.RequestResponseObservations{ - Gateway: wrc.gatewayObservations, - Protocol: messageObservations.Protocol, - Qos: messageObservations.Qos, - } - if wrc.metricsReporter != nil { - wrc.metricsReporter.Publish(observations) - } - }() + // Prepare and publish observations to both the metrics and data reporters. + observations := &observation.RequestResponseObservations{ + Gateway: wrc.gatewayObservations, + Protocol: messageObservations.Protocol, + Qos: messageObservations.Qos, + } + if wrc.metricsReporter != nil { + wrc.metricsReporter.Publish(observations) + } } // initializeMessageObservations creates a copy of observations. diff --git a/metrics/cardinality_guard.go b/metrics/cardinality_guard.go index b6ca4cf2c..e15749f8b 100644 --- a/metrics/cardinality_guard.go +++ b/metrics/cardinality_guard.go @@ -5,6 +5,7 @@ import ( "sync" "sync/atomic" + "github.com/pokt-network/poktroll/pkg/polylog" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" ) @@ -42,6 +43,17 @@ var MetricsLabelDropped = promauto.NewCounterVec( []string{"metric"}, ) +// guardLogger emits a one-time WARN when a cardinality guard first trips. Set +// once at startup via SetCardinalityGuardLogger, before any traffic; read-only +// afterwards, so a plain package var is safe (no concurrent writer). Nil until +// set — guards simply skip the WARN in that window (they cannot trip before a +// pod serves traffic anyway). +var guardLogger polylog.Logger + +// SetCardinalityGuardLogger wires the logger used for the one-time +// guard-tripped WARN. Call once during metrics startup. +func SetCardinalityGuardLogger(l polylog.Logger) { guardLogger = l } + // cardinalityGuard is a per-metric label-tuple counter. allow() reports // whether a given label tuple should be admitted to the underlying metric. // Tuples seen before are always admitted via a lock-free fast path; novel @@ -50,11 +62,12 @@ var MetricsLabelDropped = promauto.NewCounterVec( // The slow path is rare in steady state — once a workload's active suppliers // are all in the seen-set, every call returns from the fast path. type cardinalityGuard struct { - name string - limit int64 - seen sync.Map - count atomic.Int64 - addMu sync.Mutex + name string + limit int64 + seen sync.Map + count atomic.Int64 + addMu sync.Mutex + warned bool // guarded by addMu; ensures the trip WARN fires exactly once } func newCardinalityGuard(name string, limit int64) *cardinalityGuard { @@ -80,6 +93,20 @@ func (g *cardinalityGuard) allow(labelValues ...string) bool { return true } if g.count.Load() >= g.limit { + // Fire a single WARN the first time this guard saturates. Past this + // point the metric silently stops counting novel tuples and looks + // identical to a healthy one on the wire; the log (and the + // path_metrics_label_dropped_total counter) are the only signals that + // it has gone incomplete. + if !g.warned { + g.warned = true + if guardLogger != nil { + guardLogger.Warn(). + Str("metric", g.name). + Int64("series_limit", g.limit). + Msg("cardinality guard tripped: metric is now incomplete, novel label tuples are being dropped. Watch path_metrics_label_dropped_total{metric} for the drop rate.") + } + } MetricsLabelDropped.WithLabelValues(g.name).Inc() return false } diff --git a/metrics/metrics.go b/metrics/metrics.go index d96dfc813..90d4c26d8 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -34,6 +34,7 @@ const ( LabelBatchCount = "batch_count" LabelSupplier = "supplier" LabelSignalType = "signal_type" + LabelSeverity = "severity" // --- Latency signal values @@ -288,6 +289,28 @@ var ResponseBytesSent = promauto.NewCounterVec( []string{LabelRPCType, LabelServiceID}, ) +// RequestBodySizeBytes is the distribution of request body sizes, complementing +// the request_bytes_received_total counter (which only yields totals/averages). +// It exists to make p50/p99/max request sizes observable — e.g. to size the +// max_request_body_bytes limit against real traffic. +// +// Cardinality is bounded: it reuses the same fixed rpc_type/service_id labels as +// the counter (no new label dimensions) and a small, fixed bucket set. Buckets +// span from ~256B to 128MB so the common case (single-KB), the tail (single-digit +// MB), and anything approaching the request-size limit are all visible. +var RequestBodySizeBytes = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: MetricPrefix + "request_body_size_bytes", + Help: "Distribution of HTTP request body sizes in bytes by rpc_type and service_id.", + Buckets: []float64{ + 256, 1024, 4096, 16384, 65536, // 256B .. 64KB + 262144, 1048576, 4194304, 16777216, // 256KB .. 16MB + 67108864, 134217728, // 64MB, 128MB + }, + }, + []string{LabelRPCType, LabelServiceID}, +) + // ============================================================================= // Probation Events (Counter) // Labels: domain, rpc_type, service_id, event @@ -323,6 +346,7 @@ var DomainCircuitBreakerState = promauto.NewGaugeVec( // SetCircuitBreakerState sets the per-domain circuit-breaker gauge. // - broken=true → 1 (domain is currently locked out) // - broken=false → 0 (domain is healthy / has been recovered) +// // Skipped silently when domain is empty (which would happen for endpoints whose // URL parse fails and only happens in error paths anyway). func SetCircuitBreakerState(serviceID, domain string, broken bool) { @@ -352,8 +376,8 @@ func SetCircuitBreakerState(serviceID, domain string, broken bool) { // ============================================================================= const ( - LabelReasonCategory = "reason_category" - LabelCircuitBreakerEvent = "event" + LabelReasonCategory = "reason_category" + LabelCircuitBreakerEvent = "event" CircuitBreakerEventBroken = "broken" CircuitBreakerEventRecovered = "recovered" ) @@ -363,12 +387,12 @@ const ( // how variable the underlying reason strings are. Add new categories here // when MarkBroken call sites grow new reason prefixes. const ( - CircuitBreakReasonRetry = "retry" // failure during retry path - CircuitBreakReasonBatchTransport = "batch_transport" // batch path: transport-level error - CircuitBreakReasonBatchHeuristic = "batch_heuristic" // batch path: heuristic flagged response - CircuitBreakReasonParallelRetry = "parallel_retry" // parallel-retry path failure - CircuitBreakReasonHeuristic = "heuristic" // top-level heuristic break - CircuitBreakReasonUnknown = "unknown" // anything we can't classify + CircuitBreakReasonRetry = "retry" // failure during retry path + CircuitBreakReasonBatchTransport = "batch_transport" // batch path: transport-level error + CircuitBreakReasonBatchHeuristic = "batch_heuristic" // batch path: heuristic flagged response + CircuitBreakReasonParallelRetry = "parallel_retry" // parallel-retry path failure + CircuitBreakReasonHeuristic = "heuristic" // top-level heuristic break + CircuitBreakReasonUnknown = "unknown" // anything we can't classify ) var DomainCircuitBreakerEventsTotal = promauto.NewCounterVec( @@ -588,14 +612,45 @@ var SupplierReputationScore = promauto.NewGaugeVec( []string{LabelSupplier, LabelServiceID}, ) +// Severity classes for supplier_signal_total. The reputation layer emits 8 +// distinct signal-type strings; carrying all 8 as a label multiplies this +// counter's cardinality 8× on top of the (supplier × service_id) base — the +// base already accumulates toward the full network supplier set as sessions +// rotate, so the extra 8× is what pushes the metric into the 25K guard within +// ~15 min of pod start. Collapsing to 3 severity classes cuts the fan to 3× +// while preserving the only distinction this per-supplier view needs: is the +// supplier working, degraded-but-serving, or failing. Full error taxonomy +// (5xx vs timeout vs config) remains available per-domain on relays_total +// (status_code + reputation_signal). +const ( + SupplierSeverityOK = "ok" // success, recovery_success + SupplierSeveritySlow = "slow" // slow_response, very_slow_response + SupplierSeverityError = "error" // minor/major/critical/fatal error +) + var SupplierSignalTotal = promauto.NewCounterVec( prometheus.CounterOpts{ Name: MetricPrefix + "supplier_signal_total", - Help: "Reputation signals emitted by supplier, service_id, and signal_type (success/minor_error/major_error/critical_error/fatal_error/recovery_success/slow_response/very_slow_response).", + Help: "Reputation signals emitted by supplier and service_id, collapsed to a severity class (ok/slow/error). Full error taxonomy is available per-domain on relays_total.", }, - []string{LabelSupplier, LabelServiceID, LabelSignalType}, + []string{LabelSupplier, LabelServiceID, LabelSeverity}, ) +// supplierSignalSeverity collapses a reputation signal-type string (the 8 +// reputation.SignalType wire values) into one of three severity classes. +// Unknown/new signal types fall through to "error" so a mis-added type shows +// up loudly rather than silently vanishing, and cardinality stays bounded. +func supplierSignalSeverity(signalType string) string { + switch signalType { + case "success", "recovery_success": + return SupplierSeverityOK + case "slow_response", "very_slow_response": + return SupplierSeveritySlow + default: // minor_error, major_error, critical_error, fatal_error, unknown + return SupplierSeverityError + } +} + // ============================================================================= // Relays (Counter + Histogram) // Labels: domain, rpc_type, service_id, status_code, reputation_signal, request_type @@ -671,9 +726,12 @@ var WebsocketConnectionEventsTotal = promauto.NewCounterVec( var WebsocketConnectionDuration = promauto.NewHistogramVec( prometheus.HistogramOpts{ - Name: MetricPrefix + "websocket_connection_duration_seconds", - Help: "WebSocket connection duration in seconds by domain and service_id.", - Buckets: []float64{1, 5, 10, 30, 60, 300, 600, 1800, 3600}, // 1s to 1h + Name: MetricPrefix + "websocket_connection_duration_seconds", + Help: "WebSocket connection duration in seconds by domain and service_id.", + // Sub-second buckets (0.1/0.25/0.5) make immediate supplier resets — a + // connection that establishes then dies in well under a second — visible + // and distinguishable from healthy long-lived subscriptions. + Buckets: []float64{0.1, 0.25, 0.5, 1, 5, 10, 30, 60, 300, 600, 1800, 3600}, // 100ms to 1h }, []string{LabelDomain, LabelServiceID}, ) @@ -755,17 +813,21 @@ var HedgeWinningLatency = promauto.NewHistogramVec( ) // ============================================================================= -// Hedge per-supplier outcome (Histogram) -// Labels: supplier, role (winner|loser) -// Purpose: Answers "am I losing races, and by how much?" — a question the -// existing path_hedge_* metrics cannot answer because they don't carry a -// supplier label. Use win-rate = count{role=winner} / count{role=*}. +// Hedge per-supplier outcome (Counter) + role latency (Histogram) +// Purpose: Answers "am I losing races, and by how much?". // -// Cardinality: deliberately omits service_id. With service_id, observed -// production cardinality was ~79K unique tuples × 12 histogram series each -// = 945K series for this one metric (audit 2026-04-28). Per-supplier latency -// across all services answers the operator question; service-level breakdown -// is available without supplier on path_hedge_winning_latency_seconds. +// Split into two metrics to bound cardinality. The per-supplier signal is a +// win-rate — a ratio of counts — which needs no buckets, so it lives on a plain +// COUNTER (supplier × role = 2 series/supplier, ~10K series network-wide, +// well under the 25K guard → complete, not first-seen-biased). The "by how +// much" is a latency distribution that does NOT need per-supplier resolution, +// so it lives on a HISTOGRAM keyed by role only (~2×12 series total). +// +// This replaces the earlier single per-supplier HistogramVec, which multiplied +// ~8K supplier×role tuples by ~12 bucket series each = the largest single +// metric family in the gateway (~585K series, ~39% of all gateway cardinality; +// with service_id it was 945K — audit 2026-04-28). Win-rate = winner / +// (winner + loser) on the counter; loser-vs-winner latency gap on the histogram. // ============================================================================= const ( @@ -773,13 +835,26 @@ const ( HedgeRoleLoser = "loser" ) -var HedgeSupplierLatency = promauto.NewHistogramVec( +// HedgeSupplierOutcomeTotal is the per-supplier win/loss counter. Bounded and +// complete: supplier × {winner,loser} stays well under the guard cap. +var HedgeSupplierOutcomeTotal = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: MetricPrefix + "hedge_supplier_outcome_total", + Help: "Per-supplier hedge race outcomes. role=winner|loser. Win-rate = winner/(winner+loser).", + }, + []string{LabelSupplier, "role"}, +) + +// HedgeRoleLatency is the hedge outcome latency distribution by role, aggregated +// across all suppliers (no supplier label → tiny, fixed cardinality). Pairs +// with HedgeSupplierOutcomeTotal for the per-supplier win-rate. +var HedgeRoleLatency = promauto.NewHistogramVec( prometheus.HistogramOpts{ - Name: MetricPrefix + "hedge_supplier_latency_seconds", - Help: "Per-supplier hedge race outcome latency. role=winner|loser. service_id intentionally omitted to bound cardinality.", + Name: MetricPrefix + "hedge_role_latency_seconds", + Help: "Hedge race outcome latency by role (winner|loser), aggregated across suppliers. Pairs with hedge_supplier_outcome_total for per-supplier win-rate.", Buckets: []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}, }, - []string{LabelSupplier, "role"}, + []string{"role"}, ) // RecordHedgeRequest records a hedge request outcome @@ -796,17 +871,23 @@ func RecordHedgeLatencySavings(rpcType, serviceID string, savingsSeconds float64 HedgeLatencySavings.WithLabelValues(rpcType, serviceID).Observe(savingsSeconds) } -// RecordHedgeSupplierOutcome records a per-supplier hedge race outcome. -// role must be HedgeRoleWinner or HedgeRoleLoser. Skipped when supplier is -// empty or when the cardinality guard has tripped for this metric. +// RecordHedgeSupplierOutcome records a hedge race outcome. role must be +// HedgeRoleWinner or HedgeRoleLoser. The latency distribution is recorded per +// role (no supplier label, always). The per-supplier win/loss count is recorded +// only when supplier is known and the cardinality guard admits it. func RecordHedgeSupplierOutcome(supplier, role string, latencySeconds float64) { + // Role-only latency distribution: fixed, tiny cardinality — always recorded. + HedgeRoleLatency.WithLabelValues(role).Observe(latencySeconds) + + // Per-supplier win/loss counter: 2 series/supplier, still guarded as a + // backstop against a supplier-address label leak. if supplier == "" { return } if !hedgeSupplierGuard.allow(supplier, role) { return } - HedgeSupplierLatency.WithLabelValues(supplier, role).Observe(latencySeconds) + HedgeSupplierOutcomeTotal.WithLabelValues(supplier, role).Inc() } // RecordBatchSize records a batch request with latency. @@ -843,6 +924,13 @@ func batchCountBucket(batchCount int) string { func RecordRequestSize(rpcType, serviceID string, bytesReceived, bytesSent int64) { RequestBytesReceived.WithLabelValues(rpcType, serviceID).Add(float64(bytesReceived)) ResponseBytesSent.WithLabelValues(rpcType, serviceID).Add(float64(bytesSent)) + + // Also record the per-request size distribution (for p99/max visibility). + // Only observe real request bodies; skip zero to avoid diluting the histogram + // with sizeless requests that carry no signal. + if bytesReceived > 0 { + RequestBodySizeBytes.WithLabelValues(rpcType, serviceID).Observe(float64(bytesReceived)) + } } // RecordProbationEvent records a probation event (entered, exited, or routed) @@ -898,17 +986,19 @@ func SetSupplierReputationScore(supplier, serviceID string, score float64) { SupplierReputationScore.WithLabelValues(supplier, serviceID).Set(score) } -// RecordSupplierSignal increments the per-supplier signal counter. -// Skipped silently when supplier is empty (e.g., per-domain reputation key) -// or when the cardinality guard has tripped for this metric. +// RecordSupplierSignal increments the per-supplier signal counter, collapsing +// the reputation signal type to a severity class (ok/slow/error) to bound +// cardinality. Skipped silently when supplier is empty (e.g., per-domain +// reputation key) or when the cardinality guard has tripped for this metric. func RecordSupplierSignal(supplier, serviceID, signalType string) { if supplier == "" { return } - if !supplierSignalGuard.allow(supplier, serviceID, signalType) { + severity := supplierSignalSeverity(signalType) + if !supplierSignalGuard.allow(supplier, serviceID, severity) { return } - SupplierSignalTotal.WithLabelValues(supplier, serviceID, signalType).Inc() + SupplierSignalTotal.WithLabelValues(supplier, serviceID, severity).Inc() } // RecordRelay records an outgoing relay to a supplier endpoint with latency diff --git a/metrics/server.go b/metrics/server.go index a98126e64..56762b740 100644 --- a/metrics/server.go +++ b/metrics/server.go @@ -10,11 +10,23 @@ const endpointMetrics = "/metrics" // Starts a metrics server on the given address. func (pmr *PrometheusMetricsReporter) ServeMetrics(addr string) error { + // Use a dedicated ServeMux rather than http.DefaultServeMux. Importing + // net/http/pprof (in pprof.go) registers the /debug/pprof/* handlers on + // DefaultServeMux via its init(); serving DefaultServeMux here would expose + // those profiling endpoints (heap dumps, goroutine stacks, CPU profiles) on + // the public metrics port. A private mux keeps this server to /metrics only. + mux := http.NewServeMux() + mux.Handle(endpointMetrics, promhttp.Handler()) + + server := &http.Server{ + Addr: addr, + Handler: mux, + } + // Start the server in a new goroutine go func() { pmr.Logger.Info().Str("endpoint_addr", addr).Msg("starting Prometheus reporter to serve metrics asynchronously.") - http.Handle(endpointMetrics, promhttp.Handler()) - if err := http.ListenAndServe(addr, nil); err != nil { + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { pmr.Logger.Error().Err(err).Msg("prometheus metrics reporter failed starting server") return } diff --git a/metrics/supplier_hedge_cardinality_test.go b/metrics/supplier_hedge_cardinality_test.go new file mode 100644 index 000000000..671561522 --- /dev/null +++ b/metrics/supplier_hedge_cardinality_test.go @@ -0,0 +1,72 @@ +package metrics + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +// Test_supplierSignalSeverity locks the 8-signal-type → 3-severity-class +// collapse that bounds supplier_signal_total cardinality. The exact strings are +// the reputation.SignalType wire values (metrics cannot import reputation — +// import cycle — so they are asserted literally here as the contract). +func Test_supplierSignalSeverity(t *testing.T) { + cases := map[string]string{ + "success": SupplierSeverityOK, + "recovery_success": SupplierSeverityOK, + "slow_response": SupplierSeveritySlow, + "very_slow_response": SupplierSeveritySlow, + "minor_error": SupplierSeverityError, + "major_error": SupplierSeverityError, + "critical_error": SupplierSeverityError, + "fatal_error": SupplierSeverityError, + // Unknown / newly-added types must fall through to error, never leak a + // new label value (keeps cardinality bounded + surfaces the omission). + "some_future_signal": SupplierSeverityError, + "": SupplierSeverityError, + } + for in, want := range cases { + require.Equalf(t, want, supplierSignalSeverity(in), "signal %q", in) + } + + // Only three severity values may ever be emitted. + seen := map[string]struct{}{} + for in := range cases { + seen[supplierSignalSeverity(in)] = struct{}{} + } + require.LessOrEqual(t, len(seen), 3, "severity label must have at most 3 values") +} + +// Test_RecordSupplierSignal_EmptySupplierDropped guards the empty-supplier skip +// (per-domain reputation keys carry no supplier). +func Test_RecordSupplierSignal_EmptySupplierDropped(t *testing.T) { + before := testutil.CollectAndCount(SupplierSignalTotal) + RecordSupplierSignal("", "eth", "success") + require.Equal(t, before, testutil.CollectAndCount(SupplierSignalTotal), + "empty supplier must not create a series") +} + +// Test_RecordHedgeSupplierOutcome_Split guards the histogram→(counter+role +// histogram) split: the role latency histogram is always recorded, but the +// per-supplier counter is skipped when supplier is empty. +func Test_RecordHedgeSupplierOutcome_Split(t *testing.T) { + // Empty supplier: role histogram still observes, per-supplier counter does not. + histBefore := testutil.CollectAndCount(HedgeRoleLatency) + RecordHedgeSupplierOutcome("", HedgeRoleWinner, 0.12) + require.Greater(t, testutil.CollectAndCount(HedgeRoleLatency), histBefore-1, + "role latency histogram must record even without a supplier") + + // Known supplier: per-supplier counter increments for the right role. + const supplier = "pokt1testsupplierhedge" + winBefore := testutil.ToFloat64(HedgeSupplierOutcomeTotal.WithLabelValues(supplier, HedgeRoleWinner)) + RecordHedgeSupplierOutcome(supplier, HedgeRoleWinner, 0.2) + winAfter := testutil.ToFloat64(HedgeSupplierOutcomeTotal.WithLabelValues(supplier, HedgeRoleWinner)) + require.Equal(t, winBefore+1, winAfter, "winner count must increment for known supplier") + + // Loser role is a distinct series. + loseBefore := testutil.ToFloat64(HedgeSupplierOutcomeTotal.WithLabelValues(supplier, HedgeRoleLoser)) + RecordHedgeSupplierOutcome(supplier, HedgeRoleLoser, 0.5) + require.Equal(t, loseBefore+1, + testutil.ToFloat64(HedgeSupplierOutcomeTotal.WithLabelValues(supplier, HedgeRoleLoser))) +} diff --git a/pnf_path_rules.yaml b/pnf_path_rules.yaml index 38b0c42ce..121e4ff14 100644 --- a/pnf_path_rules.yaml +++ b/pnf_path_rules.yaml @@ -2,7 +2,7 @@ # Pocket Network Health Check Configuration # ========================================== # Generated for all supported chains at api.pocket.network -# Last updated: January 30, 2026 +# Last updated: July 13, 2026 # # Service IDs match the subdomain of each endpoint URL # Template format: gateway/health_check_config.go @@ -56,8 +56,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x1"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - name: eth_syncing type: json_rpc method: POST @@ -104,8 +108,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x89"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - name: eth_syncing type: json_rpc method: POST @@ -150,8 +158,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x38"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - name: eth_syncing type: json_rpc method: POST @@ -198,8 +210,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xa86a"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - queries balance at historical block 5,000,000 # Address: 0x9f8c163cBA728e99993ABe7495F06c0A3c8Ac8b9 (from E2E tests) # Block: 0x4c4b40 (5,000,000) @@ -263,8 +279,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x138de"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - queries balance at historical block 2,000,000 # Address: 0x6969696969696969696969696969696969696969 (from E2E tests) # Block: 0x1e8480 (2,000,000) @@ -302,8 +322,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x92"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - queries balance at historical block 10,769,279 # Address: 0xfc00face00000000000000000000000000000000 (from E2E tests) # Block: 0xa4537f (10,769,279) @@ -341,8 +365,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xdef1"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - queries balance at historical block 4,500,000 # Address: 0x4200000000000000000000000000000000000006 (from E2E tests) # Block: 0x44aa20 (4,500,000) @@ -380,8 +408,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x504"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 677,000 # Address: 0xf89d7b9c864f589bbf53a82105107622b35eaa40 (from E2E tests) # Block: 0xa5488 (677,000) @@ -418,8 +450,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x505"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: gnosis # Block time: ~5s | Source: gnosisscan.io, gnosis chain docs @@ -443,8 +479,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x64"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 20,000,000 # Address: 0xe91d153e0b41518a2ce8dd3d7944fa863463a97d (from E2E tests) # Block: 0x1312d00 (20,000,000) @@ -481,8 +521,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xa4ec"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 20,000,000 # Address: 0xf89d7b9c864f589bbF53a82105107622B35EaA40 (from E2E tests) # Block: 0x1312d00 (20,000,000) @@ -519,8 +563,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xfa"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 110,633,000 # Address: 0xaabf86ab3646a7064aa2f61e5959e39129ca46b6 (from E2E tests) # Block: 0x6982028 (110,633,000) @@ -582,8 +630,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x7a"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: iotex # Block time: ~5s | Source: chainspect.app, iotexscan.io @@ -607,8 +659,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x1251"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: oasys # Block time: ~15s | Source: docs.oasys.games (Hub-Layer same as Ethereum) @@ -632,8 +688,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xf8"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 424,300 # Address: 0xf89d7b9c864f589bbF53a82105107622B35EaA40 (from E2E tests) # Block: 0x6796c (424,300) @@ -670,8 +730,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x2019"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 170,000,000 # Address: 0x0051ef9259c7ec0644a80e866ab748a2f30841b3 (from E2E tests) # Block: 0xa21fe80 (170,000,000) @@ -708,8 +772,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x15f900"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 1,002,479 # Address: 0x7C21a90E3eCD3215d16c3BBe76a491f8f792d4Bf (from E2E tests) # Block: 0xf4bef (1,002,479) @@ -748,8 +816,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x3e7"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # ========================================== # LAYER 2 CHAINS @@ -777,8 +849,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xa4b1"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - name: eth_syncing type: json_rpc method: POST @@ -823,8 +899,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xa"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - name: eth_syncing type: json_rpc method: POST @@ -869,8 +949,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x2105"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - name: eth_syncing type: json_rpc method: POST @@ -915,8 +999,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x44d"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: zksync-era # Block time: ~1s | Source: era.zksync.network explorer @@ -940,8 +1028,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x144"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: zklink-nova # Block time: ~2s | Source: zkLink Nova explorer @@ -965,8 +1057,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xc5cc4"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: scroll # Block time: ~3s | Source: scrollscan.com @@ -990,8 +1086,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x82750"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 5,000,000 # Address: 0x5300000000000000000000000000000000000004 (from E2E tests) # Block: 0x4c4b40 (5,000,000) @@ -1028,8 +1128,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xe708"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 10,000,000 # Address: 0xf89d7b9c864f589bbf53a82105107622b35eaa40 (from E2E tests) # Block: 0x989680 (10,000,000) @@ -1067,8 +1171,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x1388"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: blast # Block time: ~2s | Source: blastscan.io (OP Stack default) @@ -1092,8 +1200,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x13e31"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 1,000,000 # Address: 0x4300000000000000000000000000000000000004 (from E2E tests) # Block: 0xf4240 (1,000,000) @@ -1168,8 +1280,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x440"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 15,000,000 # Address: 0xfad31cd4d45Ac7C4B5aC6A0044AA05Ca7C017e62 (from E2E tests) # Block: 0xe4e1c0 (15,000,000) @@ -1206,8 +1322,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x28c58"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 170,163 # Address: 0x1670000000000000000000000000000000000001 (from E2E tests) # Block: 0x298b3 (170,163) @@ -1244,8 +1364,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x82"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: opbnb # Block time: ~1s | Source: opbnbscan.com @@ -1269,8 +1393,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xcc"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # Archival check - block 20,000,000 # Address: 0x001ceb373c83ae75b9f5cf78fc2aba3e185d09e2 (from E2E tests) # Block: 0x1312d00 (20,000,000) @@ -1307,8 +1435,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xfc"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # ========================================== # TESTNET CHAINS @@ -1336,8 +1468,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xaa36a7"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # DEPRECATED from public portal but retained for operators - service_id: eth-holesky-testnet @@ -1387,8 +1523,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x13882"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: arb-sepolia-testnet # Block time: ~0.25s | Source: Same as Arbitrum mainnet @@ -1412,8 +1552,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x66eee"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: op-sepolia-testnet # Block time: ~2s | Source: Same as Optimism mainnet @@ -1437,8 +1581,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xaa37dc"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: base-sepolia-testnet # Block time: ~2s | Source: Same as Base mainnet @@ -1462,8 +1610,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x14a34"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: xrplevm-testnet # Block time: ~4s | Source: Same as XRPL EVM mainnet @@ -1487,8 +1639,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x161c28"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: giwa-sepolia-testnet # Block time: ~12s | Source: Ethereum L2 testnet @@ -1512,8 +1668,12 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0x164ce"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error # ========================================== # NON-EVM CHAINS @@ -1644,6 +1804,49 @@ reputation_signal: critical_error sync_check: true + - name: chainId + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' + expected_status_code: 200 + expected_response_contains: '0x2b6653dc' + timeout: 5s + reputation_signal: critical_error + +- service_id: manta + # Manta Pacific. Previously had NO health check rules at all: all 50 endpoints + # were unreachable by any check and stayed at score 100 permanently. + # Block time: ~8s (measured live 2026-07-12, small sample) | Formula: 300s / 8s ~= 37 + # Rounded up to 50 to avoid false-flagging healthy endpoints as stale. + sync_allowance: 50 + # 30s (not 10s): health checks are billable synthetic relays, so probe volume has + # a direct POKT cost. 30s keeps detection well inside the sync window at 1/3 the cost. + check_interval: 30s + enabled: true + checks: + - name: eth_blockNumber + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"eth_blockNumber","params":[]}' + expected_status_code: 200 + timeout: 5s + reputation_signal: critical_error + sync_check: true + - name: eth_chainId + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' + expected_status_code: 200 + # Chain-ID assertion: guards against a supplier serving a DIFFERENT chain + # under this service ID (see tron/Base incident). Trailing quote anchors the + # match so a short id like 0x1 cannot substring-match 0x1388 etc. + expected_response_contains: '0xa9"' + timeout: 5s + reputation_signal: critical_error + # ========================================== # COSMOS SDK CHAINS # ========================================== @@ -2081,9 +2284,11 @@ path: / body: '{"jsonrpc":"2.0","id":1,"method":"eth_chainId","params":[]}' expected_status_code: 200 - expected_response_contains: '"0x' + # Was '"0x' — which matched ANY hex chain ID and so asserted nothing. + # Sei mainnet is 1329 = 0x531 (verified against evm-rpc.sei-apis.com). + expected_response_contains: '0x531"' timeout: 5s - reputation_signal: major_error + reputation_signal: critical_error - service_id: shentu # Block time: ~6s | Source: Cosmos SDK default @@ -2151,4 +2356,133 @@ expected_status_code: 200 expected_response_contains: '"syncing"' timeout: 5s - reputation_signal: major_error \ No newline at end of file + reputation_signal: major_error + +# ========================================== +# PREVIOUSLY UNCHECKED SERVICES +# Added 2026-07-12. Each of these had NO health check rules at all: no check +# could reach their endpoints, so every endpoint sat at score 100 permanently +# and could never be penalized, no matter how it behaved. +# All paths/assertions below were verified against live suppliers before landing. +# ========================================== + +- service_id: cosmoshub + # Block time: ~6s (measured live 2026-07-12) | Formula: 300s / 6s = 50 + sync_allowance: 50 + check_interval: 30s + enabled: true + checks: + - name: health + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"health"}' + expected_status_code: 200 + timeout: 5s + reputation_signal: minor_error + - name: status + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"status"}' + expected_status_code: 200 + # Chain-identity assertion: CometBFT `status` reports the network it is on. + # This is the Cosmos analogue of the eth_chainId assertion, and guards against + # a supplier serving a DIFFERENT Cosmos chain under this service ID. + expected_response_contains: '"network":"cosmoshub-4"' + timeout: 5s + reputation_signal: critical_error + sync_check: true + - name: syncing + type: rest + method: GET + path: /cosmos/base/tendermint/v1beta1/syncing + expected_status_code: 200 + expected_response_contains: '"syncing"' + timeout: 5s + reputation_signal: major_error + +- service_id: provenance + # Block time: ~6s (measured live 2026-07-12) | Formula: 300s / 6s = 50 + sync_allowance: 50 + check_interval: 30s + enabled: true + checks: + - name: health + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"health"}' + expected_status_code: 200 + timeout: 5s + reputation_signal: minor_error + - name: status + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"status"}' + expected_status_code: 200 + # Chain-identity assertion. Provenance mainnet network id is pio-mainnet-1. + expected_response_contains: '"network":"pio-mainnet-1"' + timeout: 5s + reputation_signal: critical_error + sync_check: true + - name: syncing + type: rest + method: GET + path: /cosmos/base/tendermint/v1beta1/syncing + expected_status_code: 200 + expected_response_contains: '"syncing"' + timeout: 5s + reputation_signal: major_error + +- service_id: eth-beacon + # Ethereum Beacon Chain (consensus layer) - Beacon REST API, not EVM JSON-RPC. + # Slot time is a fixed 12s protocol constant | Formula: 300s / 12s = 25 slots. + # NOTE: head_slot is not an EVM block height, so sync-allowance filtering may be + # inert here even with sync_check set. The checks still gate liveness/reputation. + sync_allowance: 25 + check_interval: 30s + enabled: true + checks: + - name: node_syncing + type: rest + method: GET + path: /eth/v1/node/syncing + expected_status_code: 200 + expected_response_contains: '"is_syncing"' + timeout: 5s + reputation_signal: critical_error + sync_check: true + - name: node_version + type: rest + method: GET + path: /eth/v1/node/version + expected_status_code: 200 + expected_response_contains: '"version"' + timeout: 5s + reputation_signal: major_error + +- service_id: taiko-hekla-testnet + # WARNING: this service looks DEFUNCT as of 2026-07-12. All 4 session endpoints + # return: service "taiko-hekla-testnet" not configured: service endpoint not + # handled by relayer proxy -- the relay miner has no config for it. Public Taiko + # Hekla RPCs were also unreachable, so the testnet itself may be retired. + # Recommend `enabled: false` or deregistering the service entirely. + # These checks exist so the dead endpoints drop from score 100 instead of being + # advertised as healthy and handed live traffic. + # Deliberately NO chain-ID assertion: the canonical chain ID could not be verified + # against any live source, and asserting an unverified value would be worse than + # asserting nothing. + sync_allowance: 0 + check_interval: 30s + enabled: true + checks: + - name: eth_blockNumber + type: json_rpc + method: POST + path: / + body: '{"jsonrpc":"2.0","id":1,"method":"eth_blockNumber","params":[]}' + expected_status_code: 200 + timeout: 5s + reputation_signal: critical_error diff --git a/protocol/shannon/context.go b/protocol/shannon/context.go index a3828bb14..2fd03e6c9 100644 --- a/protocol/shannon/context.go +++ b/protocol/shannon/context.go @@ -9,6 +9,7 @@ import ( "net/http" "strconv" "sync" + "sync/atomic" "time" "github.com/alitto/pond/v2" @@ -77,16 +78,24 @@ type requestContext struct { // currentRPCType: // - Tracks the RPC type of the current relay being processed. // - Set during relay execution and used when building observations. - currentRPCType sharedtypes.RPCType + // - Atomic: written per-relay (RPC-type fallback) and read to build + // observations concurrently across the parallel-batch relay goroutines, + // which all target the single selected endpoint. Stored as its int32 + // underlying value. + currentRPCType atomic.Int32 // heuristicRPCType preserves the original detected RPC type before any // fallback was applied. Used for heuristic response analysis so that // CometBFT requests routed through JSON_RPC are still analyzed correctly. + // Set once before parallelization (happens-before the relay goroutines), so + // it is only read — never written — during parallel execution. heuristicRPCType sharedtypes.RPCType // requestErrorObservation: // - Tracks any errors encountered during request processing. - requestErrorObservation *protocolobservations.ShannonRequestError + // - Atomic: handleInternalError may write it from a parallel relay goroutine + // while it is read to build observations. + requestErrorObservation atomic.Pointer[protocolobservations.ShannonRequestError] // endpointObservations: // - Captures observations about endpoints used during request handling. @@ -101,7 +110,10 @@ type requestContext struct { // currentRelayMinerError: // - Tracks RelayMinerError data from the current relay response for reporting. // - Set by trackRelayMinerError method and used when building observations. - currentRelayMinerError *protocolobservations.ShannonRelayMinerError + // - Atomic: written per-relay (trackRelayMinerError) and read to build + // observations / detect over-servicing concurrently across parallel-batch + // relay goroutines. A plain pointer field allowed a torn read → crash. + currentRelayMinerError atomic.Pointer[protocolobservations.ShannonRelayMinerError] // HTTP client used for sending relay requests to endpoints while also capturing various debug metrics httpClient *pathhttp.HTTPClientWithDebugMetrics @@ -165,6 +177,14 @@ type requestContext struct { // - Handles both single requests and JSON-RPC batch requests concurrently when beneficial. // - Returns responses as an array to match interface, but gateway currently expects single response. // - Captures RelayMinerError data when available for reporting purposes. +// +// getCurrentRPCType returns the current relay's RPC type via an atomic read. +// currentRPCType is written per-relay and read concurrently across the parallel +// batch relay goroutines, so all access goes through the atomic. +func (rc *requestContext) getCurrentRPCType() sharedtypes.RPCType { + return sharedtypes.RPCType(rc.currentRPCType.Load()) +} + func (rc *requestContext) HandleServiceRequest(payloads []protocol.Payload) ([]protocol.Response, error) { // Internal error: No endpoint selected. if rc.getSelectedEndpoint() == nil { @@ -184,7 +204,7 @@ func (rc *requestContext) HandleServiceRequest(payloads []protocol.Payload) ([]p // Only override if payload has an explicit RPC type (non-zero value). // This preserves the default set in BuildHTTPRequestContextForEndpoint for health checks. if payloads[0].RPCType != sharedtypes.RPCType_UNKNOWN_RPC { - rc.currentRPCType = payloads[0].RPCType + rc.currentRPCType.Store(int32(payloads[0].RPCType)) rc.heuristicRPCType = payloads[0].EffectiveRPCType() } @@ -423,7 +443,7 @@ func (rc *requestContext) GetObservations() protocolobservations.Observations { Observations: []*protocolobservations.ShannonRequestObservations{ { ServiceId: string(rc.serviceID), - RequestError: rc.requestErrorObservation, + RequestError: rc.requestErrorObservation.Load(), ObservationData: &protocolobservations.ShannonRequestObservations_HttpObservations{ HttpObservations: &protocolobservations.ShannonHTTPEndpointObservations{ EndpointObservations: rc.endpointObservations, @@ -609,7 +629,7 @@ func (rc *requestContext) sendProtocolRelay(payload protocol.Payload) (protocol. } // Default: use the RPC-type-specific URL from the selected endpoint default: - targetServerURL = selectedEndpoint.GetURL(rc.currentRPCType) + targetServerURL = selectedEndpoint.GetURL(rc.getCurrentRPCType()) // FALLBACK HANDLING: If the URL is empty, it means the endpoint doesn't support // the originally requested RPC type. This happens when RPC type fallback occurred @@ -628,11 +648,11 @@ func (rc *requestContext) sendProtocolRelay(payload protocol.Payload) (protocol. if url := selectedEndpoint.GetURL(rpcType); url != "" { targetServerURL = url rc.logger.Debug(). - Str("requested_rpc_type", rc.currentRPCType.String()). + Str("requested_rpc_type", rc.getCurrentRPCType().String()). Str("actual_rpc_type", rpcType.String()). Str("endpoint", string(selectedEndpoint.Addr())). Msg("Endpoint doesn't support requested RPC type, using supported type from endpoint") - rc.currentRPCType = rpcType // Update to the actual RPC type + rc.currentRPCType.Store(int32(rpcType)) // Update to the actual RPC type break } } @@ -1085,11 +1105,11 @@ func (rc *requestContext) trackRelayMinerError(relayResponse *servicetypes.Relay ).Debug().Msg("RelayMiner returned an error in RelayResponse (captured for reporting)") // Store RelayMinerError data in request context for use in observations - rc.currentRelayMinerError = &protocolobservations.ShannonRelayMinerError{ + rc.currentRelayMinerError.Store(&protocolobservations.ShannonRelayMinerError{ Codespace: relayMinerErr.Codespace, Code: relayMinerErr.Code, Message: relayMinerErr.Message, - } + }) } // handleInternalError: @@ -1104,7 +1124,7 @@ func (rc *requestContext) handleInternalError(internalErr error) (protocol.Respo rc.logger.Error().Err(internalErr).Msg("Internal error occurred. This should be investigated as a bug.") // Set request processing error for generating observations. - rc.requestErrorObservation = buildInternalRequestProcessingErrorObservation(internalErr) + rc.requestErrorObservation.Store(buildInternalRequestProcessingErrorObservation(internalErr)) return protocol.Response{}, internalErr } @@ -1137,8 +1157,9 @@ func (rc *requestContext) handleEndpointError( // (service, session, supplier) tuple so endpoint selection skips it for // the rest of the session, and let the classifier (which now returns // SuccessSignal for over-serviced payloads) handle the no-penalty path. - overServiced := rc.currentRelayMinerError != nil && - heuristic.IsOverServicedRelayMinerError(rc.currentRelayMinerError.Codespace, rc.currentRelayMinerError.Code) + relayMinerErr := rc.currentRelayMinerError.Load() + overServiced := relayMinerErr != nil && + heuristic.IsOverServicedRelayMinerError(relayMinerErr.Codespace, relayMinerErr.Code) if !overServiced && endpointErr != nil && heuristic.IsOverServicedError(endpointErr.Error()) { overServiced = true } @@ -1179,8 +1200,8 @@ func (rc *requestContext) handleEndpointError( time.Now(), // Timestamp: endpoint query completed. endpointErrorType, fmt.Sprintf("relay error: %v", endpointErr), - rc.currentRelayMinerError, // Use RelayMinerError data from request context - rc.currentRPCType, // Use RPC type from request context + rc.currentRelayMinerError.Load(), // Use RelayMinerError data from request context + rc.getCurrentRPCType(), // Use RPC type from request context ) // Track endpoint error observation for metrics @@ -1202,7 +1223,7 @@ func (rc *requestContext) handleEndpointError( if rc.reputationService != nil && !isBlacklisted { keyBuilder := rc.reputationService.KeyBuilderForService(rc.serviceID) - endpointKey := keyBuilder.BuildKey(rc.serviceID, selectedEndpointAddr, rc.currentRPCType) + endpointKey := keyBuilder.BuildKey(rc.serviceID, selectedEndpointAddr, rc.getCurrentRPCType()) // Fire-and-forget: don't block request on reputation recording if err := rc.reputationService.RecordSignal(rc.context, endpointKey, signal); err != nil { @@ -1219,7 +1240,7 @@ func (rc *requestContext) handleEndpointError( if domainErr != nil { domain = shannonmetrics.ErrDomain } - rpcTypeStr := metrics.NormalizeRPCType(rc.currentRPCType.String()) + rpcTypeStr := metrics.NormalizeRPCType(rc.getCurrentRPCType().String()) reputationSignal := mapSignalTypeToMetricSignal(signal.Type) // Extract status code from error if possible, otherwise use "error" @@ -1271,8 +1292,8 @@ func (rc *requestContext) handleEndpointSuccess( endpointQueryTime, time.Now(), // Timestamp: endpoint query completed. endpointResponse, - rc.currentRelayMinerError, // Use RelayMinerError data from request context - rc.currentRPCType, // Use RPC type from request context + rc.currentRelayMinerError.Load(), // Use RelayMinerError data from request context + rc.getCurrentRPCType(), // Use RPC type from request context ) // Track endpoint success observation for metrics @@ -1296,7 +1317,7 @@ func (rc *requestContext) handleEndpointSuccess( if rc.reputationService != nil { keyBuilder := rc.reputationService.KeyBuilderForService(rc.serviceID) - endpointKey := keyBuilder.BuildKey(rc.serviceID, selectedEndpointAddr, rc.currentRPCType) + endpointKey := keyBuilder.BuildKey(rc.serviceID, selectedEndpointAddr, rc.getCurrentRPCType()) var signal reputation.Signal @@ -1329,7 +1350,7 @@ func (rc *requestContext) handleEndpointSuccess( if probationDomain == "" { probationDomain = shannonmetrics.ErrDomain } - metrics.RecordProbationEvent(probationDomain, metrics.NormalizeRPCType(rc.currentRPCType.String()), string(rc.serviceID), metrics.ProbationEventRouted) + metrics.RecordProbationEvent(probationDomain, metrics.NormalizeRPCType(rc.getCurrentRPCType().String()), string(rc.serviceID), metrics.ProbationEventRouted) rc.logger.Debug(). Str("endpoint", string(selectedEndpointAddr)). @@ -1375,7 +1396,7 @@ func (rc *requestContext) handleEndpointSuccess( domain = shannonmetrics.ErrDomain } statusCodeStr := metrics.GetStatusCodeCategory(endpointResponse.HTTPStatusCode) - rpcTypeStr := metrics.NormalizeRPCType(rc.currentRPCType.String()) + rpcTypeStr := metrics.NormalizeRPCType(rc.getCurrentRPCType().String()) metrics.RecordRelay(domain, rpcTypeStr, string(rc.serviceID), statusCodeStr, reputationSignal, relayType, latency.Seconds()) // Return relay response received from endpoint. diff --git a/protocol/shannon/observation_websocket.go b/protocol/shannon/observation_websocket.go index a77e506e6..6be624559 100644 --- a/protocol/shannon/observation_websocket.go +++ b/protocol/shannon/observation_websocket.go @@ -89,6 +89,7 @@ func getWebsocketConnectionEstablishedObservation( logger polylog.Logger, serviceID protocol.ServiceID, selectedEndpoint endpoint, + establishedAt time.Time, ) *protocolobservations.Observations { return &protocolobservations.Observations{ Shannon: &protocolobservations.ShannonObservationsList{ @@ -100,6 +101,8 @@ func getWebsocketConnectionEstablishedObservation( logger, selectedEndpoint, protocolobservations.ShannonWebsocketConnectionObservation_CONNECTION_ESTABLISHED, + establishedAt, + time.Time{}, // no close timestamp on establishment ), }, }, @@ -112,6 +115,8 @@ func getWebsocketConnectionClosedObservation( logger polylog.Logger, serviceID protocol.ServiceID, selectedEndpoint endpoint, + establishedAt time.Time, + closedAt time.Time, ) *protocolobservations.Observations { return &protocolobservations.Observations{ Shannon: &protocolobservations.ShannonObservationsList{ @@ -123,6 +128,8 @@ func getWebsocketConnectionClosedObservation( logger, selectedEndpoint, protocolobservations.ShannonWebsocketConnectionObservation_CONNECTION_CLOSED, + establishedAt, + closedAt, ), }, }, @@ -234,15 +241,24 @@ func buildWebsocketMessageErrorObservation( // buildWebsocketConnectionObservation creates a Shannon websocket connection observation for connection lifecycle events. // It includes endpoint details and session information for connection-level tracking. // Used when websocket connection setup succeeds or when connection closes. +// +// establishedAt is the real connection-establishment time and is carried on BOTH +// the ESTABLISHED and CLOSED events. closedAt is set only on the CLOSED event +// (zero otherwise); the gateway computes connection duration as closedAt-establishedAt, +// so the CLOSED observation must carry both. Previously this stamped +// ConnectionEstablishedTimestamp=now on every event and never set +// ConnectionClosedTimestamp, so the duration metric always recorded 0. func buildWebsocketConnectionObservation( _ polylog.Logger, endpoint endpoint, eventType protocolobservations.ShannonWebsocketConnectionObservation_ConnectionEventType, + establishedAt time.Time, + closedAt time.Time, ) *protocolobservations.ShannonWebsocketConnectionObservation { session := *endpoint.Session() sessionHeader := session.GetHeader() - return &protocolobservations.ShannonWebsocketConnectionObservation{ + obs := &protocolobservations.ShannonWebsocketConnectionObservation{ // Endpoint information Supplier: endpoint.Supplier(), EndpointUrl: endpoint.PublicURL(), @@ -256,9 +272,18 @@ func buildWebsocketConnectionObservation( SessionEndHeight: sessionHeader.SessionEndBlockHeight, // Connection lifecycle - ConnectionEstablishedTimestamp: timestamppb.New(time.Now()), + ConnectionEstablishedTimestamp: timestamppb.New(establishedAt), EventType: eventType, } + + // Only the CLOSED event carries a close timestamp; its presence (together with + // the established timestamp above) is what lets the gateway record a non-zero + // connection duration. + if !closedAt.IsZero() { + obs.ConnectionClosedTimestamp = timestamppb.New(closedAt) + } + + return obs } // buildWebsocketConnectionErrorObservation creates a Shannon websocket connection observation for failed connection events. diff --git a/protocol/shannon/observation_websocket_test.go b/protocol/shannon/observation_websocket_test.go new file mode 100644 index 000000000..a5a7eaabb --- /dev/null +++ b/protocol/shannon/observation_websocket_test.go @@ -0,0 +1,44 @@ +package shannon + +import ( + "testing" + "time" + + "github.com/pokt-network/poktroll/pkg/polylog/polyzero" + "github.com/stretchr/testify/require" + + "github.com/pokt-network/path/protocol" +) + +// Test_WebsocketConnectionObservation_Duration guards the fix for the always-zero +// WebSocket connection duration metric. The closure observation must carry BOTH +// the real establishment timestamp and a close timestamp so the gateway can +// compute a non-zero duration; the establishment observation must carry only the +// establishment timestamp. +func Test_WebsocketConnectionObservation_Duration(t *testing.T) { + logger := polyzero.NewLogger() + ep := &mockEndpoint{addr: "supplier1-https://relay.example.com:443"} + svc := protocol.ServiceID("eth") + + established := time.Now() + closed := established.Add(42 * time.Second) + + // ESTABLISHED: established timestamp set, close timestamp absent. + estObs := getWebsocketConnectionEstablishedObservation(logger, svc, ep, established) + estConn := estObs.GetShannon().GetObservations()[0].GetWebsocketConnectionObservation() + require.NotNil(t, estConn.GetConnectionEstablishedTimestamp()) + require.Nil(t, estConn.GetConnectionClosedTimestamp(), "establishment obs must not carry a close timestamp") + + // CLOSED: both timestamps set, and their delta is the real connection duration. + // Before the fix, ConnectionClosedTimestamp was never set and the established + // timestamp was re-stamped to now, so the computed duration was always 0. + clsObs := getWebsocketConnectionClosedObservation(logger, svc, ep, established, closed) + clsConn := clsObs.GetShannon().GetObservations()[0].GetWebsocketConnectionObservation() + require.NotNil(t, clsConn.GetConnectionEstablishedTimestamp()) + require.NotNil(t, clsConn.GetConnectionClosedTimestamp(), "closure obs must carry a close timestamp") + + gotDur := clsConn.GetConnectionClosedTimestamp().AsTime(). + Sub(clsConn.GetConnectionEstablishedTimestamp().AsTime()) + require.Greater(t, gotDur.Seconds(), 0.0, "duration must be non-zero (regression guard for the always-0 bug)") + require.InDelta(t, (42 * time.Second).Seconds(), gotDur.Seconds(), 0.001) +} diff --git a/protocol/shannon/protocol.go b/protocol/shannon/protocol.go index e577870fc..1cf0cf8e9 100644 --- a/protocol/shannon/protocol.go +++ b/protocol/shannon/protocol.go @@ -806,7 +806,7 @@ func (p *Protocol) BuildHTTPRequestContextForEndpoint( } // Return new request context for the pre-selected endpoint - return &requestContext{ + rc := &requestContext{ logger: fullyHydratedLogger, // Use fully-hydrated logger context: ctx, fullNode: p.FullNode, @@ -823,8 +823,11 @@ func (p *Protocol) BuildHTTPRequestContextForEndpoint( tieredSelector: tieredSelector, supplierBlacklist: p.supplierBlacklist, sessionExhaustion: p.sessionExhaustion, - currentRPCType: actualRPCType, // Use actual RPC type after fallback (may differ from requested) - }, protocolobservations.Observations{}, nil + } + // currentRPCType is atomic; set it after construction. + // Use actual RPC type after fallback (may differ from requested). + rc.currentRPCType.Store(int32(actualRPCType)) + return rc, protocolobservations.Observations{}, nil } // ApplyHTTPObservations updates protocol instance state based on endpoint observations. diff --git a/protocol/shannon/session_exhaustion.go b/protocol/shannon/session_exhaustion.go index 00c0bb556..ba1111549 100644 --- a/protocol/shannon/session_exhaustion.go +++ b/protocol/shannon/session_exhaustion.go @@ -7,9 +7,13 @@ import ( ) // sessionExhaustionTTL bounds how long an entry stays in the tracker before -// being lazily reaped on the next Mark or IsExhausted call. Sessions on -// Shannon are short (~10 blocks ≈ 10 min); 30 min covers ~3 session -// lifecycles which is enough to absorb clock skew and session-boundary races +// being lazily reaped on the next Mark or IsExhausted call. Entries are keyed +// by (service, session, supplier), so a new session never reuses an old key — +// the TTL only bounds memory growth, it does not affect correctness (a stale +// entry from a past session is dead weight until reaped, never a false +// exhaustion). It is set comfortably longer than a single Shannon session +// (session length is a chain param) so a flagged supplier stays skipped for the +// whole remaining session plus slack for clock skew and session-boundary races, // while preventing the map from growing unbounded if the reaping path is // somehow not triggered. const sessionExhaustionTTL = 30 * time.Minute diff --git a/protocol/shannon/websocket_context.go b/protocol/shannon/websocket_context.go index 50d317842..f9f2d0a84 100644 --- a/protocol/shannon/websocket_context.go +++ b/protocol/shannon/websocket_context.go @@ -414,15 +414,20 @@ func (wrc *websocketRequestContext) startWebSocketBridge( go func() { defer close(connectionObservationChan) + // Capture the establishment time once and carry it through to the closure + // observation so the gateway can compute a real connection duration + // (closedAt - establishedAt). The bridge is live at this point. + establishedAt := time.Now() + // Send establishment observation immediately (buffered channel ensures it's captured) wrc.logger.Debug().Msg("Websocket bridge started successfully, sending establishment observation") - connectionObservationChan <- getWebsocketConnectionEstablishedObservation(wrc.logger, wrc.serviceID, wrc.selectedEndpoint) + connectionObservationChan <- getWebsocketConnectionEstablishedObservation(wrc.logger, wrc.serviceID, wrc.selectedEndpoint, establishedAt) // Wait for the bridge to complete (blocks until Websocket connection terminates) <-bridgeCompletionChan - // Send closure observation + // Send closure observation, stamping the actual close time. wrc.logger.Debug().Msg("Websocket connection closed, sending closure observation") - connectionObservationChan <- getWebsocketConnectionClosedObservation(wrc.logger, wrc.serviceID, wrc.selectedEndpoint) + connectionObservationChan <- getWebsocketConnectionClosedObservation(wrc.logger, wrc.serviceID, wrc.selectedEndpoint, establishedAt, time.Now()) }() return nil diff --git a/qos/block_height_plausibility.go b/qos/block_height_plausibility.go new file mode 100644 index 000000000..79a6b770e --- /dev/null +++ b/qos/block_height_plausibility.go @@ -0,0 +1,47 @@ +package qos + +// Block-height plausibility guards. +// +// Several QoS types derive a "perceived" chain height by taking the maximum +// height any endpoint reports. That makes a single endpoint able to set the +// perceived height to an arbitrary value: report a height far above the real +// tip and every honest endpoint falls outside the sync-allowance window and is +// filtered out, handing 100% of the service's traffic to the liar. +// +// IsPlausibleBlockHeight is a cheap defense-in-depth guard applied where the +// perceived height is updated. It rejects absurd absolute values and +// implausibly large single-update jumps, which blunts the total-hijack case +// (e.g. a report of MaxUint64). It does NOT defend against small over-reports +// (a value just past the sync allowance) — that requires outlier-resistant, +// median-anchored consensus (as EVM has and the other QoS types still need). +const ( + // MaxPlausibleBlockHeight is an absolute sanity ceiling on any reported + // block height. No real chain is within many orders of magnitude of 10^12 + // blocks (at 200ms/block that is ~6000 years), so a reported height above + // this is malicious or garbage and must never become the perceived height. + MaxPlausibleBlockHeight uint64 = 1_000_000_000_000 // 1e12 + + // MaxBlockHeightJump bounds how far above the current perceived height a + // single update may move it. Real chains advance a bounded number of blocks + // between observations; a jump of 10M blocks (days of the fastest chains) + // is implausible in steady state. Only applied once a perceived height is + // established (current > 0); the absolute ceiling covers cold start. + MaxBlockHeightJump uint64 = 10_000_000 // 1e7 +) + +// IsPlausibleBlockHeight reports whether a newly-reported block height is a +// plausible update to the current perceived height. Callers should only adopt +// `candidate` as the new perceived height when this returns true. +func IsPlausibleBlockHeight(candidate, current uint64) bool { + // Absolute sanity ceiling (also the only guard at cold start, current == 0). + if candidate > MaxPlausibleBlockHeight { + return false + } + // Reject implausibly large jumps above an established perceived height. + // current <= MaxPlausibleBlockHeight here, so current + MaxBlockHeightJump + // cannot overflow. + if current > 0 && candidate > current+MaxBlockHeightJump { + return false + } + return true +} diff --git a/qos/block_height_plausibility_test.go b/qos/block_height_plausibility_test.go new file mode 100644 index 000000000..97d155a96 --- /dev/null +++ b/qos/block_height_plausibility_test.go @@ -0,0 +1,78 @@ +package qos + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_IsPlausibleBlockHeight(t *testing.T) { + tests := []struct { + name string + candidate uint64 + current uint64 + want bool + }{ + { + name: "normal small advance is plausible", + candidate: 1_000_100, + current: 1_000_000, + want: true, + }, + { + name: "cold start: reasonable first value is plausible", + candidate: 20_000_000, + current: 0, + want: true, + }, + { + name: "cold start: MaxUint64 is rejected by the absolute ceiling", + candidate: math.MaxUint64, + current: 0, + want: false, + }, + { + name: "steady state: MaxUint64 is rejected", + candidate: math.MaxUint64, + current: 1_000_000, + want: false, + }, + { + name: "above the absolute ceiling is rejected", + candidate: MaxPlausibleBlockHeight + 1, + current: 1_000_000, + want: false, + }, + { + name: "at the absolute ceiling with no established perceived is allowed", + candidate: MaxPlausibleBlockHeight, + current: 0, + want: true, + }, + { + name: "implausibly large jump above established perceived is rejected", + candidate: 1_000_000 + MaxBlockHeightJump + 1, + current: 1_000_000, + want: false, + }, + { + name: "jump right at the limit is allowed", + candidate: 1_000_000 + MaxBlockHeightJump, + current: 1_000_000, + want: true, + }, + { + name: "equal values are plausible", + candidate: 500, + current: 500, + want: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.want, IsPlausibleBlockHeight(test.candidate, test.current)) + }) + } +} diff --git a/qos/cosmos/service_state.go b/qos/cosmos/service_state.go index ad0e5e37b..95295721b 100644 --- a/qos/cosmos/service_state.go +++ b/qos/cosmos/service_state.go @@ -9,6 +9,7 @@ import ( "github.com/pokt-network/path/metrics/devtools" qosobservations "github.com/pokt-network/path/observation/qos" "github.com/pokt-network/path/protocol" + "github.com/pokt-network/path/qos" ) var _ protocol.EndpointSelector = &serviceState{} @@ -97,8 +98,15 @@ func (ss *serviceState) updateFromEndpoints(updatedEndpoints map[protocol.Endpoi // Per perceivedBlockNumber field documentation, it should be "the maximum of block height reported by any endpoint" // but code was incorrectly overwriting with each endpoint, causing validation failures. if blockNumber > ss.perceivedBlockNumber { - logger.Debug().Msgf("Updating perceived block number from %d to %d", ss.perceivedBlockNumber, blockNumber) - ss.perceivedBlockNumber = blockNumber + if !qos.IsPlausibleBlockHeight(blockNumber, ss.perceivedBlockNumber) { + logger.Warn(). + Uint64("reported_block", blockNumber). + Uint64("perceived_block", ss.perceivedBlockNumber). + Msg("⚠️ ignoring implausible block height (possible poisoning attempt)") + } else { + logger.Debug().Msgf("Updating perceived block number from %d to %d", ss.perceivedBlockNumber, blockNumber) + ss.perceivedBlockNumber = blockNumber + } } } @@ -113,8 +121,23 @@ func (ss *serviceState) getDisqualifiedEndpointsResponse(serviceID protocol.Serv DisqualifiedEndpoints: make(map[protocol.EndpointAddr]devtools.QoSDisqualifiedEndpoint), } - // Populate the data response object using the endpoints in the endpoint store. - for endpointAddr, endpoint := range ss.endpointStore.endpoints { + // Snapshot the endpoints under the store lock, then validate outside it. + // Ranging ss.endpointStore.endpoints directly (as this did) without holding + // endpointsMu races with the writers that update the map on every observation + // — Go's runtime turns a concurrent iterate+write into an unrecoverable + // `fatal error: concurrent map iteration and map write` that crashes the pod + // (reachable via GET /disqualified_endpoints). basicEndpointValidation takes + // serviceStateLock, so we copy first and release endpointsMu before calling it + // to avoid holding two locks across the call. + ss.endpointStore.endpointsMu.RLock() + endpointsSnapshot := make(map[protocol.EndpointAddr]endpoint, len(ss.endpointStore.endpoints)) + for addr, ep := range ss.endpointStore.endpoints { + endpointsSnapshot[addr] = ep + } + ss.endpointStore.endpointsMu.RUnlock() + + // Populate the data response object using the snapshot. + for endpointAddr, endpoint := range endpointsSnapshot { if err := ss.basicEndpointValidation(endpoint); err != nil { qosLevelDataResponse.DisqualifiedEndpoints[endpointAddr] = devtools.QoSDisqualifiedEndpoint{ EndpointAddr: endpointAddr, diff --git a/qos/cosmos/service_state_endpoint_validation.go b/qos/cosmos/service_state_endpoint_validation.go index b8b964f72..0435cc29e 100644 --- a/qos/cosmos/service_state_endpoint_validation.go +++ b/qos/cosmos/service_state_endpoint_validation.go @@ -5,6 +5,7 @@ import ( "fmt" "time" + "github.com/pokt-network/path/qos" sharedtypes "github.com/pokt-network/poktroll/x/shared/types" ) @@ -246,7 +247,7 @@ func (ss *serviceState) validateBlockHeightSyncAllowance(latestBlockHeight uint6 Uint64("endpoint_block", latestBlockHeight). Msg("🔍 Sync allowance check ENABLED") - minAllowedBlockNumber := ss.perceivedBlockNumber - syncAllowance + minAllowedBlockNumber := qos.MinAllowedBlockNumber(ss.perceivedBlockNumber, syncAllowance) if latestBlockHeight < minAllowedBlockNumber { ss.logger.Warn(). Str("service_id", string(ss.serviceQoSConfig.GetServiceID())). diff --git a/qos/evm/endpoint_selection.go b/qos/evm/endpoint_selection.go index 007529c24..f6b7fca36 100644 --- a/qos/evm/endpoint_selection.go +++ b/qos/evm/endpoint_selection.go @@ -14,6 +14,7 @@ import ( "github.com/pokt-network/path/metrics" qosobservations "github.com/pokt-network/path/observation/qos" "github.com/pokt-network/path/protocol" + "github.com/pokt-network/path/qos" "github.com/pokt-network/path/qos/selector" "github.com/pokt-network/path/reputation" ) @@ -24,8 +25,10 @@ var ( errEmptyEndpointListObs = errors.New("received empty list of endpoints to select from") ) -// TODO_UPNEXT(@adshmh): make the invalid response timeout duration configurable -// It is set to 5 minutes because that is the session time as of #321. +// TODO_UPNEXT(@adshmh): make the invalid response timeout duration configurable. +// Fixed cooldown applied after an endpoint returns an empty/invalid response. +// Intentionally kept shorter than a Shannon session so an endpoint can recover +// within the same session; it is NOT derived from session length. const invalidResponseTimeout = 5 * time.Minute // EndpointSelectionResult contains endpoint selection results and metadata. @@ -305,7 +308,7 @@ func (ss *serviceState) filterValidEndpointsWithDetails(availableEndpoints proto if syncAllowance > 0 && perceivedBlock > 0 { if url, err := data.addr.GetURL(); err == nil { if urlBlock, ok := urlBlockHeights[url]; ok { - minAllowed := perceivedBlock - syncAllowance + minAllowed := qos.MinAllowedBlockNumber(perceivedBlock, syncAllowance) if urlBlock < minAllowed { blocksBehind := int64(perceivedBlock) - int64(urlBlock) invalidCount++ @@ -453,7 +456,7 @@ func (ss *serviceState) categorizeValidationFailure(err error) qosobservations.E // // It returns an error if: // - The endpoint has returned an empty response in the past. -// - The endpoint has returned an invalid response within the last 30 minutes. +// - The endpoint has returned an invalid response within the invalidResponseTimeout window. // - The endpoint's response to an `eth_chainId` request is not the expected chain ID. // - The endpoint's response to an `eth_blockNumber` request is greater than the perceived block number. // - The endpoint's archival check is invalid, if requiresArchival is true and archival checks are enabled. @@ -602,7 +605,7 @@ func (ss *serviceState) isBlockNumberValid(check endpointCheckBlockNumber) error // If the endpoint's block height is less than the perceived block height minus the sync allowance, // then the endpoint is behind the chain and should be filtered out. - minAllowedBlockNumber := perceivedBlock - syncAllowance + minAllowedBlockNumber := qos.MinAllowedBlockNumber(perceivedBlock, syncAllowance) if parsedBlockNumber < minAllowedBlockNumber { blocksBehind := int64(perceivedBlock) - int64(parsedBlockNumber) @@ -792,7 +795,7 @@ func (ss *serviceState) filterStaleURLEndpoints(endpoints protocol.EndpointAddrL Msg("filterStaleURLEndpoints: perceived_block is 0, skipping stale URL filtering") return endpoints } - minAllowed := perceivedBlock - syncAllowance + minAllowed := qos.MinAllowedBlockNumber(perceivedBlock, syncAllowance) urlBlockHeights := ss.buildURLBlockHeightMap() diff --git a/qos/evm/service_state.go b/qos/evm/service_state.go index 135849858..35a8daaef 100644 --- a/qos/evm/service_state.go +++ b/qos/evm/service_state.go @@ -210,9 +210,24 @@ func (ss *serviceState) getDisqualifiedEndpointsResponse(serviceID protocol.Serv DisqualifiedEndpoints: make(map[protocol.EndpointAddr]devtools.QoSDisqualifiedEndpoint), } - // Populate the data response object using the endpoints in the endpoint store. + // Snapshot the endpoints under the store lock, then validate outside it. + // Ranging ss.endpointStore.endpoints directly (as this did) without holding + // endpointsMu races with the writers that update the map on every observation + // — Go's runtime turns a concurrent iterate+write into an unrecoverable + // `fatal error: concurrent map iteration and map write` that crashes the pod + // (reachable via GET /disqualified_endpoints). basicEndpointValidation takes + // its own lock, so we copy first and release before calling it to avoid + // holding endpointsMu across it. + ss.endpointStore.endpointsMu.RLock() + endpointsSnapshot := make(map[protocol.EndpointAddr]endpoint, len(ss.endpointStore.endpoints)) + for addr, ep := range ss.endpointStore.endpoints { + endpointsSnapshot[addr] = ep + } + ss.endpointStore.endpointsMu.RUnlock() + + // Populate the data response object using the snapshot. // Use requiresArchival=true to check full validation including archival (most conservative) - for endpointAddr, endpoint := range ss.endpointStore.endpoints { + for endpointAddr, endpoint := range endpointsSnapshot { if err := ss.basicEndpointValidation(endpointAddr, endpoint, true); err != nil { qosLevelDataResponse.DisqualifiedEndpoints[endpointAddr] = devtools.QoSDisqualifiedEndpoint{ EndpointAddr: endpointAddr, diff --git a/qos/noop/noop.go b/qos/noop/noop.go index fccc00db4..d30acb0f0 100644 --- a/qos/noop/noop.go +++ b/qos/noop/noop.go @@ -18,6 +18,7 @@ import ( "github.com/pokt-network/path/metrics/devtools" qosobservations "github.com/pokt-network/path/observation/qos" "github.com/pokt-network/path/protocol" + "github.com/pokt-network/path/qos" qostypes "github.com/pokt-network/path/qos/types" "github.com/pokt-network/path/reputation" ) @@ -142,7 +143,16 @@ func (n *NoOpQoS) UpdateFromExtractedData(endpointAddr protocol.EndpointAddr, da n.serviceStateMu.Lock() defer n.serviceStateMu.Unlock() - if blockHeight > n.perceivedBlockHeight { + // Guard against poisoning: ignore an implausibly high report so one endpoint + // cannot set the perceived height arbitrarily high and filter out all honest + // endpoints. The per-endpoint Redis write below still runs regardless. + if blockHeight > n.perceivedBlockHeight && !qos.IsPlausibleBlockHeight(blockHeight, n.perceivedBlockHeight) { + n.logger.Warn(). + Str("endpoint", string(endpointAddr)). + Uint64("reported_block", blockHeight). + Uint64("perceived_block", n.perceivedBlockHeight). + Msg("⚠️ ignoring implausible block height (possible poisoning attempt)") + } else if blockHeight > n.perceivedBlockHeight { n.logger.Debug(). Str("endpoint", string(endpointAddr)). Uint64("old_block", n.perceivedBlockHeight). diff --git a/qos/noop/selector.go b/qos/noop/selector.go index 4c18ba2fd..388e76a2d 100644 --- a/qos/noop/selector.go +++ b/qos/noop/selector.go @@ -7,6 +7,7 @@ import ( "github.com/pokt-network/poktroll/pkg/polylog" "github.com/pokt-network/path/protocol" + "github.com/pokt-network/path/qos" "github.com/pokt-network/path/qos/selector" ) @@ -126,7 +127,7 @@ func (fs *filteringSelector) filterValidEndpoints(endpoints protocol.EndpointAdd return endpoints } - minAllowedBlock := perceivedBlock - syncAllowance + minAllowedBlock := qos.MinAllowedBlockNumber(perceivedBlock, syncAllowance) fs.endpointStore.mu.RLock() diff --git a/qos/solana/solana.go b/qos/solana/solana.go index 9f28a5cb6..5b42f45ee 100644 --- a/qos/solana/solana.go +++ b/qos/solana/solana.go @@ -13,6 +13,7 @@ import ( "github.com/pokt-network/path/metrics/devtools" qosobservations "github.com/pokt-network/path/observation/qos" "github.com/pokt-network/path/protocol" + "github.com/pokt-network/path/qos" qostypes "github.com/pokt-network/path/qos/types" "github.com/pokt-network/path/reputation" ) @@ -142,8 +143,17 @@ func (q *QoS) UpdateFromExtractedData(endpointAddr protocol.EndpointAddr, data * q.serviceStateLock.Lock() defer q.serviceStateLock.Unlock() - // Update perceived block height to maximum across all endpoints - if blockHeight > q.perceivedBlockHeight { + // Update perceived block height to maximum across all endpoints. + // Guard against poisoning: ignore an implausibly high report so one endpoint + // cannot set the perceived height arbitrarily high and filter out all honest + // endpoints. The per-endpoint Redis write below still runs regardless. + if blockHeight > q.perceivedBlockHeight && !qos.IsPlausibleBlockHeight(blockHeight, q.perceivedBlockHeight) { + q.logger.Warn(). + Str("endpoint", string(endpointAddr)). + Uint64("reported_block", blockHeight). + Uint64("perceived_block", q.perceivedBlockHeight). + Msg("⚠️ ignoring implausible block height (possible poisoning attempt)") + } else if blockHeight > q.perceivedBlockHeight { q.logger.Debug(). Str("endpoint", string(endpointAddr)). Uint64("old_block", q.perceivedBlockHeight). diff --git a/qos/solana/state.go b/qos/solana/state.go index 9a04dee6a..cb0e3bb7d 100644 --- a/qos/solana/state.go +++ b/qos/solana/state.go @@ -7,6 +7,7 @@ import ( "github.com/pokt-network/poktroll/pkg/polylog" "github.com/pokt-network/path/protocol" + "github.com/pokt-network/path/qos" ) // ServiceState keeps the expected current state of the Solana blockchain @@ -71,9 +72,18 @@ func (s *ServiceState) UpdateFromEndpoints(updatedEndpoints map[protocol.Endpoin continue } - // TODO_TECHDEBT: use a more resilient method for updating block height. - // e.g. one endpoint returning a very large number as block height should - // not result in all other endpoints being marked as invalid. + // Defense-in-depth against block-height poisoning: ignore implausibly high + // reports so one endpoint cannot set the perceived height arbitrarily high + // and filter out every honest endpoint. (Small over-reports still require + // median-anchored consensus — see qos.IsPlausibleBlockHeight.) + if !qos.IsPlausibleBlockHeight(endpoint.BlockHeight, s.perceivedBlockHeight) { + s.logger.Warn(). + Uint64("reported_block", endpoint.BlockHeight). + Uint64("perceived_block", s.perceivedBlockHeight). + Msg("⚠️ ignoring implausible block height (possible poisoning attempt)") + continue + } + s.perceivedEpoch = endpoint.Epoch s.perceivedBlockHeight = endpoint.BlockHeight diff --git a/qos/sync_allowance.go b/qos/sync_allowance.go new file mode 100644 index 000000000..a3fd925a7 --- /dev/null +++ b/qos/sync_allowance.go @@ -0,0 +1,19 @@ +package qos + +// MinAllowedBlockNumber returns the lowest block number an endpoint may report +// while still being considered in sync with the chain: perceivedBlock minus the +// sync allowance, floored at 0. +// +// Both inputs are uint64, so a naive perceivedBlock - syncAllowance underflows +// to a value near MaxUint64 whenever the sync allowance exceeds the perceived +// block height (e.g. very early in a chain's life, or a misconfigured allowance). +// That underflow would make the "endpoint block >= minAllowed" check fail for +// every endpoint and reject the entire endpoint set. Saturating the subtraction +// at 0 instead means "no floor" — every endpoint passes — which is the correct +// behavior when the allowance is larger than the chain height so far. +func MinAllowedBlockNumber(perceivedBlock, syncAllowance uint64) uint64 { + if perceivedBlock <= syncAllowance { + return 0 + } + return perceivedBlock - syncAllowance +} diff --git a/qos/sync_allowance_test.go b/qos/sync_allowance_test.go new file mode 100644 index 000000000..1f8df908d --- /dev/null +++ b/qos/sync_allowance_test.go @@ -0,0 +1,60 @@ +package qos + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_MinAllowedBlockNumber(t *testing.T) { + tests := []struct { + name string + perceivedBlock uint64 + syncAllowance uint64 + want uint64 + }{ + { + name: "normal: perceived well above allowance", + perceivedBlock: 1_000_000, + syncAllowance: 100, + want: 999_900, + }, + { + name: "allowance greater than perceived clamps to 0 (no underflow)", + perceivedBlock: 5, + syncAllowance: 100, + want: 0, + }, + { + name: "allowance equal to perceived clamps to 0", + perceivedBlock: 50, + syncAllowance: 50, + want: 0, + }, + { + name: "zero perceived block clamps to 0", + perceivedBlock: 0, + syncAllowance: 10, + want: 0, + }, + { + name: "zero allowance returns perceived unchanged", + perceivedBlock: 12345, + syncAllowance: 0, + want: 12345, + }, + { + name: "large allowance near MaxUint64 does not wrap", + perceivedBlock: 100, + syncAllowance: math.MaxUint64, + want: 0, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.want, MinAllowedBlockNumber(test.perceivedBlock, test.syncAllowance)) + }) + } +} diff --git a/reputation/storage/redis.go b/reputation/storage/redis.go index 4fd513bfc..e39e159d4 100644 --- a/reputation/storage/redis.go +++ b/reputation/storage/redis.go @@ -47,6 +47,15 @@ const ( fieldArchivalExpires = "archival_expires_at" ) +// perceivedBlockTTL bounds how long a perceived block-height entry lives in +// Redis without being superseded by a higher write. A normally-advancing chain +// refreshes this every write cycle (seconds), so it never expires under normal +// operation; a stuck or poisoned value that no honest write can beat receives no +// further writes and expires, self-healing. 1h is far above the write cadence +// and ≈3 Shannon sessions (a session is ~20 blocks / ~20 min), so a perceived +// height stuck this long is unambiguously stale. +const perceivedBlockTTL = 1 * time.Hour + // NewRedisStorage creates a new Redis-backed storage. // It validates the connection by sending a PING command. func NewRedisStorage(ctx context.Context, config reputation.RedisConfig, ttl time.Duration) (*RedisStorage, error) { @@ -392,18 +401,30 @@ func (s *RedisStorage) SetPerceivedBlockNumber(ctx context.Context, serviceID pr key := s.perceivedBlockKey(serviceID) // Use Lua script for atomic compare-and-set (only update if higher) - // This ensures that across all replicas, the max value always wins + // This ensures that across all replicas, the max value always wins. + // + // The SET carries a TTL (ARGV[2] seconds) so a stuck perceived height + // self-heals instead of persisting forever. A poisoned/too-high value would + // otherwise never recover: max-wins means honest lower writes are dropped + // (return 0) and every replica re-reads the bad value from Redis on restart, + // needing a manual DEL. Crucially the TTL is refreshed ONLY on the update + // path (return 1): a normally-advancing chain writes a higher value each + // cycle and keeps the key alive, while a value that no honest write can beat + // (because it is above the real tip) receives no further writes and expires. + // Refreshing the TTL on the no-op path would keep a poison alive forever. script := redis.NewScript(` local current = redis.call('GET', KEYS[1]) local newVal = tonumber(ARGV[1]) + local ttl = tonumber(ARGV[2]) if current == false or tonumber(current) < newVal then - redis.call('SET', KEYS[1], ARGV[1]) + redis.call('SET', KEYS[1], ARGV[1], 'EX', ttl) return 1 end return 0 `) - _, err := script.Run(ctx, s.client, []string{key}, blockNumber).Result() + ttlSeconds := int(perceivedBlockTTL.Seconds()) + _, err := script.Run(ctx, s.client, []string{key}, blockNumber, ttlSeconds).Result() if err != nil { return fmt.Errorf("failed to set perceived block number: %w", err) } diff --git a/router/router.go b/router/router.go index e8f64f307..cd9dd6ecb 100644 --- a/router/router.go +++ b/router/router.go @@ -230,6 +230,15 @@ func (r *router) removeGrovePortalPrefixMiddleware(next http.HandlerFunc) http.H // 2. Prevents empty responses on long operations // 3. Forwards request to gateway handler func (r *router) handleServiceRequest(w http.ResponseWriter, req *http.Request) { + // Bound the request body size to prevent a single oversized body from OOMing + // the process. This wraps the underlying body before any handler reads it, so + // every downstream read (RPC-type detection, observation capture, relay) draws + // from a capped source; a read past the limit fails rather than allocating + // without bound. Harmless for websocket upgrades, which do not read req.Body. + if r.config.MaxRequestBodyBytes > 0 && req.Body != nil { + req.Body = http.MaxBytesReader(w, req.Body, r.config.MaxRequestBodyBytes) + } + // Reserve time for system overhead and apply it to the business logic operations. processingTimeout := r.config.WriteTimeout - r.config.SystemOverheadAllowanceDuration diff --git a/websockets/bridge.go b/websockets/bridge.go index d592fdbfd..2a9bea3bb 100644 --- a/websockets/bridge.go +++ b/websockets/bridge.go @@ -105,10 +105,19 @@ func StartBridge( endpointConn, err := ConnectWebsocketEndpoint(logger, websocketURL, headers) if err != nil { logger.Error().Err(err).Msg("❌ error connecting to websocket endpoint") - // Clean up the client connection that was successfully upgraded - if closeErr := clientConn.Close(); closeErr != nil { - logger.Warn().Err(closeErr).Msg("error closing client connection after endpoint connection failure") - } + // The client upgrade already succeeded, so the client is a live Websocket + // peer. Send a legible close frame BEFORE closing: without one, gorilla + // drops the TCP connection with no close handshake, which clients surface + // as an abnormal closure (1006) or a protocol error (1002) rather than a + // meaningful reason. 1013 (Try Again Later) tells the client to reconnect + // — a fresh connection typically draws a different supplier — which is the + // correct guidance when a randomly selected upstream endpoint is down. + closeClientConnWithReason( + logger, + clientConn, + websocket.CloseTryAgainLater, + "upstream endpoint unavailable, please reconnect", + ) cancelCtx() // Clean up context on error return nil, fmt.Errorf("createWebsocketBridge: %w", err) } @@ -163,6 +172,26 @@ func StartBridge( return completionChan, nil } +// closeClientConnWithReason sends a best-effort Websocket close frame carrying a +// legible status code to an already-upgraded client connection, then closes it. +// +// Used on connection-establishment failures that occur AFTER the client upgrade +// succeeds but BEFORE the bridge exists (so the normal shutdown() close-frame +// path is not yet available). Without an explicit close frame, gorilla's Close() +// tears down the TCP connection with no close handshake, which clients report as +// an abnormal closure (1006) or a protocol error (1002) — misleading, since the +// real cause is an unavailable upstream endpoint, not a client protocol fault. +func closeClientConnWithReason(logger polylog.Logger, conn *websocket.Conn, closeCode int, reason string) { + deadline := time.Now().Add(1 * time.Second) + closeMsg := websocket.FormatCloseMessage(closeCode, reason) + if err := conn.WriteControl(websocket.CloseMessage, closeMsg, deadline); err != nil { + logger.Warn().Err(err).Msg("⚠️ could not write close frame to client connection after endpoint connection failure") + } + if err := conn.Close(); err != nil { + logger.Warn().Err(err).Msg("error closing client connection after endpoint connection failure") + } +} + // validateComponents ensures the Bridge is not created with nil components. // This is done to avoid panics and to make the Bridge's behavior more predictable. func (b *bridge) validateComponents() error { @@ -229,8 +258,13 @@ func (b *bridge) shutdown(err error) { // from sending on msgChan after it's no longer being read. b.cancelCtx() - // Determine appropriate close code and message for client reconnection guidance + // Determine appropriate close code and message for client reconnection guidance. + // Sanitize at this single emit choke point so a reserved code (e.g. a 1006 + // propagated from an abnormal peer disconnect) never reaches the wire, where + // the relay miner rejects it ("websocket: bad close code 1006") and clients + // see a malformed frame. closeCode, errMsg := b.determineCloseCodeAndMessage(err) + closeCode = sanitizeCloseCode(closeCode) closeMsg := websocket.FormatCloseMessage(closeCode, errMsg) // Write close messages with timeout to prevent hanging on broken connections @@ -262,6 +296,25 @@ func (b *bridge) shutdown(err error) { }) } +// sanitizeCloseCode maps RFC 6455 §7.4.1 reserved status codes — which are for +// internal endpoint use and MUST NOT appear in a close frame on the wire — to a +// valid code (1011 Internal Error). gorilla/websocket synthesizes a +// *CloseError{Code: 1006} when a peer drops the TCP connection without a close +// handshake; extractCloseInfo caches that code and determineCloseCodeAndMessage +// would otherwise propagate it verbatim to the other peer. Relay miners reject a +// 1006 close frame ("websocket: bad close code 1006") and clients see a malformed +// frame. Reserved: 1005 (no status), 1006 (abnormal closure), 1015 (TLS handshake). +func sanitizeCloseCode(code int) int { + switch code { + case websocket.CloseNoStatusReceived, // 1005 + websocket.CloseAbnormalClosure, // 1006 + websocket.CloseTLSHandshake: // 1015 + return websocket.CloseInternalServerErr // 1011 + default: + return code + } +} + // determineCloseCodeAndMessage determines the appropriate Websocket close code and message // based on the error that caused the bridge shutdown. This guides client reconnection behavior. // @@ -351,7 +404,7 @@ func (b *bridge) handleClientMessage(msg message) { b.logger.Debug().Msgf("🔗 client message successfully processed, sending message to endpoint: %s", string(processedData)) // Send the processed message to the endpoint - if err := b.endpointConn.WriteMessage(msg.messageType, processedData); err != nil { + if err := b.endpointConn.writeMessage(msg.messageType, processedData); err != nil { b.logger.Error().Err(err).Msg("❌ error writing client message to endpoint, shutting down bridge") b.shutdown(fmt.Errorf("%w: failed to write client message to endpoint: %w", ErrBridgeConnectionFailed, err)) @@ -390,7 +443,7 @@ func (b *bridge) handleEndpointMessage(msg message) { // Send the processed message to the client // NOTE: On session rollover, the Endpoint will disconnect the Endpoint connection, which will trigger this // error. This is expected and the Client is expected to handle the reconnection in their connection logic. - if err := b.clientConn.WriteMessage(msg.messageType, processedData); err != nil { + if err := b.clientConn.writeMessage(msg.messageType, processedData); err != nil { b.logger.Error().Err(err).Msg("❌ error writing endpoint message to client, shutting down bridge") b.shutdown(fmt.Errorf("%w: failed to write endpoint message to client: %w", ErrBridgeConnectionFailed, err)) return diff --git a/websockets/bridge_test.go b/websockets/bridge_test.go index 8945508cc..6ab16c377 100644 --- a/websockets/bridge_test.go +++ b/websockets/bridge_test.go @@ -208,6 +208,115 @@ func Test_Bridge_NoPanicOnProcessingError(t *testing.T) { } } +// Test_Bridge_ClientGetsCloseFrameOnEndpointConnectFailure verifies that when the +// upstream endpoint connection fails AFTER the client upgrade has succeeded, the +// client receives a legible Websocket close frame (1013 Try Again Later) instead +// of an abnormal/protocol-error closure. This mirrors production: a randomly +// selected supplier resets the bridge connection, and the client should be told +// to reconnect (which draws a different supplier) rather than see a 1002/1006. +func Test_Bridge_ClientGetsCloseFrameOnEndpointConnectFailure(t *testing.T) { + c := require.New(t) + + // Stand up an endpoint server, capture its address, then close it so the + // bridge's upstream dial is refused — simulating a supplier that refuses/resets + // the connection right after the client upgrade succeeds. + endpointServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + endpointURL := "ws" + strings.TrimPrefix(endpointServer.URL, "http") + endpointServer.Close() // dials to this address now fail (connection refused) + + messageProcessor := &mockWebsocketMessageProcessor{} + observationsChan := make(chan *observation.RequestResponseObservations, 10) + + clientServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // StartBridge upgrades the client, then fails to reach the (dead) upstream. + _, _ = StartBridge( + context.Background(), + polyzero.NewLogger(), + r, + w, + endpointURL, + http.Header{}, + messageProcessor, + observationsChan, + ) + })) + defer clientServer.Close() + + clientURL := "ws" + strings.TrimPrefix(clientServer.URL, "http") + clientConn, _, err := websocket.DefaultDialer.Dial(clientURL, nil) + c.NoError(err, "client upgrade should succeed before the upstream failure") + defer clientConn.Close() + + // The client must receive a well-formed close frame with code 1013, not an + // abnormal closure (1006) or protocol error (1002). + _ = clientConn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, _, readErr := clientConn.ReadMessage() + c.Error(readErr, "expected a close frame from the gateway") + c.True( + websocket.IsCloseError(readErr, websocket.CloseTryAgainLater), + "client should receive a 1013 (Try Again Later) close frame, got: %v", readErr, + ) +} + +// Test_Bridge_SanitizesReservedCloseCode verifies that when the endpoint drops its +// TCP connection with no close handshake — gorilla surfaces this as a reserved +// *CloseError{Code: 1006} — PATH does NOT forward the reserved 1006 to the client. +// Instead the client receives a valid 1011 (Internal Error) close frame. Without +// sanitization the relay miner rejects the forwarded frame ("bad close code 1006") +// and clients see a malformed close. +func Test_Bridge_SanitizesReservedCloseCode(t *testing.T) { + c := require.New(t) + + // Endpoint upgrades, then abruptly closes the underlying TCP connection with no + // close frame — the exact condition that makes gorilla synthesize a 1006. + endpointServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + up := websocket.Upgrader{} + conn, err := up.Upgrade(w, r, nil) + if err != nil { + return + } + conn.UnderlyingConn().Close() + })) + defer endpointServer.Close() + + messageProcessor := &mockWebsocketMessageProcessor{} + observationsChan := make(chan *observation.RequestResponseObservations, 10) + endpointURL := "ws" + strings.TrimPrefix(endpointServer.URL, "http") + + clientServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = StartBridge( + context.Background(), + polyzero.NewLogger(), + r, + w, + endpointURL, + http.Header{}, + messageProcessor, + observationsChan, + ) + })) + defer clientServer.Close() + + clientURL := "ws" + strings.TrimPrefix(clientServer.URL, "http") + clientConn, _, err := websocket.DefaultDialer.Dial(clientURL, nil) + c.NoError(err) + defer clientConn.Close() + + // The client must receive a valid (non-reserved) close code. Before the fix it + // would receive the reserved 1006 propagated straight from the endpoint. + _ = clientConn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, _, readErr := clientConn.ReadMessage() + c.Error(readErr, "expected a close frame from the gateway") + c.False( + websocket.IsCloseError(readErr, websocket.CloseAbnormalClosure), + "client must NOT receive a reserved 1006 close code, got: %v", readErr, + ) + c.True( + websocket.IsCloseError(readErr, websocket.CloseInternalServerErr), + "reserved endpoint code should be sanitized to 1011, got: %v", readErr, + ) +} + // Mock implementations for testing type mockWebsocketMessageProcessor struct{} diff --git a/websockets/closeinfo_race_test.go b/websockets/closeinfo_race_test.go new file mode 100644 index 000000000..f74d68a42 --- /dev/null +++ b/websockets/closeinfo_race_test.go @@ -0,0 +1,55 @@ +package websockets + +import ( + "context" + "sync" + "testing" + + "github.com/gorilla/websocket" + "github.com/pokt-network/poktroll/pkg/polylog" +) + +// Test_CloseInfo_DataRace proves that lastCloseCode/lastCloseText are accessed +// without synchronization: handleDisconnect (called concurrently from connLoop +// and pingLoop of the SAME connection) writes them, while GetCloseInfo (called +// from the bridge shutdown path) reads them. Run with -race. +func Test_CloseInfo_DataRace(t *testing.T) { + _, cancel := context.WithCancel(context.Background()) + defer cancel() + + c := &websocketConnection{ + logger: polylog.Ctx(context.Background()), + cancelCtx: cancel, + source: messageSourceEndpoint, + } + + // A close error so handleDisconnect takes the write path (closeCode != 0). + closeErr := &websocket.CloseError{Code: websocket.CloseNormalClosure, Text: "bye"} + + var wg sync.WaitGroup + // Writer 1: simulates connLoop hitting a read error. + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 1000; i++ { + c.handleDisconnect(closeErr) + } + }() + // Writer 2: simulates pingLoop hitting a ping-write failure. + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 1000; i++ { + c.handleDisconnect(closeErr) + } + }() + // Reader: simulates bridge.shutdown → determineCloseCodeAndMessage → GetCloseInfo. + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 1000; i++ { + _, _ = c.GetCloseInfo() + } + }() + wg.Wait() +} diff --git a/websockets/connection.go b/websockets/connection.go index 9865e8d93..4931e7c48 100644 --- a/websockets/connection.go +++ b/websockets/connection.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "net/url" + "sync" "time" "github.com/gorilla/websocket" @@ -25,6 +26,20 @@ const ( // Send pings to peer with this period over the websocket connection // Must be greater than pongWaitDuration pingPeriodDuration = (pongWaitDuration * 9) / 10 + + // maxMessageBytes caps the size of a single websocket frame read from either + // peer. Without a limit, gorilla/websocket buffers an entire frame in memory, + // so one oversized frame — from a malicious client or a compromised/malicious + // endpoint — can OOM the process. 15MB matches go-ethereum's WS message size + // limit (wsMessageSizeLimit): large enough for legitimate EVM responses (e.g. + // a big eth_getLogs) carried over a subscription connection, but bounded. + // + // Polygon is the primary WS service on PATH; its bor client is a go-ethereum + // fork that inherits the same 15MB wsMessageSizeLimit, so a poly endpoint will + // never send a frame larger than this — the cap cannot break legitimate poly + // traffic. When exceeded, ReadMessage returns an error and the connection is + // torn down via the existing handleDisconnect path. + maxMessageBytes = 15 * 1024 * 1024 ) // messageSource is used to identify the source of a message in a bidirectional websocket connection. @@ -67,6 +82,13 @@ type websocketConnection struct { source messageSource msgChan chan<- message + // closeInfoMu guards lastCloseCode/lastCloseText. Both connLoop and pingLoop + // can call handleDisconnect concurrently on a broken socket (write/write), and + // the bridge shutdown path reads them via GetCloseInfo while the other + // connection's loops may still be writing (read/write). Without this lock those + // accesses race — a torn read of the lastCloseText string header is undefined + // behavior, and CI's -race build flags it. + closeInfoMu sync.Mutex // lastCloseCode stores the close code from the last disconnect. // This is used to propagate close codes from endpoint to client. lastCloseCode int @@ -151,12 +173,35 @@ func newConnection( msgChan: msgChan, } + // Bound the size of a single inbound frame to prevent a single oversized + // message from OOMing the process. Applied to both client and endpoint + // connections since either peer can send an abusive frame. + conn.SetReadLimit(maxMessageBytes) + go c.connLoop() go c.pingLoop() return c } +// writeMessage sets a write deadline and then writes a data message to the peer. +// +// Without a deadline, gorilla's WriteMessage blocks indefinitely when the peer +// stops reading (its TCP receive buffer fills). That would wedge the bridge's +// single message-processing goroutine — which does not select on ctx during the +// write — so a slow or stalled reader could hang the connection and hold all of +// its resources until the OS-level TCP timeout. The deadline bounds that wait so +// the write fails fast and the bridge shuts the connection down. +// +// Note: gorilla's control-frame writes (ping/close) already pass their own +// deadline; this covers the data-frame path used by the bridge. +func (c *websocketConnection) writeMessage(messageType int, data []byte) error { + if err := c.SetWriteDeadline(time.Now().Add(writeWaitDuration)); err != nil { + return err + } + return c.WriteMessage(messageType, data) +} + // connLoop reads messages from the websocket connection and sends them to the bridge's msgChan. // Network-level read errors trigger handleDisconnect() for coordinated bridge shutdown. func (c *websocketConnection) connLoop() { @@ -202,8 +247,10 @@ func (c *websocketConnection) handleDisconnect(err error) { closeCode, closeText := extractCloseInfo(err) if closeCode != 0 { // Store close code for propagation to the other side of the bridge + c.closeInfoMu.Lock() c.lastCloseCode = closeCode c.lastCloseText = closeText + c.closeInfoMu.Unlock() c.logger.Info(). Int("close_code", closeCode). Str("close_text", closeText). @@ -218,6 +265,8 @@ func (c *websocketConnection) handleDisconnect(err error) { // GetCloseInfo returns the close code and text from the last disconnect. // Returns 0 and empty string if no close code was received. func (c *websocketConnection) GetCloseInfo() (int, string) { + c.closeInfoMu.Lock() + defer c.closeInfoMu.Unlock() return c.lastCloseCode, c.lastCloseText } diff --git a/websockets/connection_test.go b/websockets/connection_test.go index 3e0d39e1a..8f5bf0ec3 100644 --- a/websockets/connection_test.go +++ b/websockets/connection_test.go @@ -146,6 +146,51 @@ func Test_Connection_HandleDisconnect(t *testing.T) { c.NotNil(connection) } +// Test_Connection_ReadLimit verifies that a single frame larger than +// maxMessageBytes does not get buffered/forwarded: the read fails, the +// connection is torn down (context canceled) and no message reaches msgChan. +// This guards against the unbounded-read OOM (a peer sending a huge frame). +func Test_Connection_ReadLimit(t *testing.T) { + c := require.New(t) + + // Server sends one frame that exceeds the read limit. + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upgrader := websocket.Upgrader{} + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Error("Error during connection upgrade:", err) + return + } + // Server has no read limit set, so it can send an oversized frame. + oversized := make([]byte, maxMessageBytes+100) + _ = conn.WriteMessage(websocket.BinaryMessage, oversized) + // Keep the handler open briefly so the client can process the frame. + time.Sleep(200 * time.Millisecond) + })) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + c.NoError(err) + + msgChan := make(chan message, 1) + ctx, cancelCtx := context.WithCancel(context.Background()) + defer cancelCtx() + + newConnection(ctx, cancelCtx, polyzero.NewLogger(), conn, messageSourceEndpoint, msgChan) + + // The oversized frame must NOT be forwarded, and the read failure must + // tear the connection down via context cancellation. + select { + case msg := <-msgChan: + t.Fatalf("oversized frame should have been rejected, but %d bytes were forwarded", len(msg.data)) + case <-ctx.Done(): + // Expected: connLoop hit the read-limit error and called handleDisconnect → cancelCtx. + case <-time.After(2 * time.Second): + t.Fatal("timeout: connection was not torn down after oversized frame") + } +} + // createTestConnection creates a websocket connection for testing func Test_ConnectWebsocketEndpoint(t *testing.T) { tests := []struct {