Per-operator concentration cap + reputation hardening (URL granularity, rate cooldown, HC dedup, WS rebind)

cmd/main.go

@@ -330,6 +330,11 @@ func main() {
 		QoSLevelReporters:     qosLevelReporters,
 	}
 
+	// Admin handler to reset per-service chain state (perceived block height) via
+	// POST /admin/chain-state/clear/{serviceId}. Recovers a stuck/too-high perceived
+	// height that the max-only consensus and external floor cannot self-correct.
+	chainStateAdmin := gateway.NewChainStateAdmin(qosInstances)
+
 	// Initialize the API router to serve requests to the PATH API.
 	apiRouter := router.NewRouter(
 		logger,
@@ -338,6 +343,8 @@ func main() {
 		healthChecker,
 		config.GetRouterConfig(),
 		gtw.DomainCircuitBreaker,
+		chainStateAdmin,
+		unifiedServicesConfig,
 	)
 
 	// -------------------- Start PATH API Router --------------------

cmd/qos.go

@@ -109,6 +109,25 @@ func getServiceQoSInstances(
 			noopQoS := noop.NewNoOpQoSService(qosLogger, serviceID)
 			qosServices[serviceID] = noopQoS
 		}
+
+		// Per-operator concentration cap: bounds any single operator's (eTLD+1) share of
+		// endpoint selection (config-driven, shipped ON by default). Applied uniformly to
+		// every QoS type that supports it — EVM/Cosmos/Solana/NoOp — covering both HTTP and
+		// WebSocket initial selection via requestContext.Select. Disabled (>= 1 or <= 0)
+		// leaves selection as a flat random pick.
+		if unifiedConfig != nil {
+			if setter, ok := qosServices[serviceID].(interface{ SetMaxOperatorShare(float64) }); ok {
+				maxOperatorShare := unifiedConfig.GetMaxOperatorShareForService(serviceID)
+				setter.SetMaxOperatorShare(maxOperatorShare)
+				if maxOperatorShare > 0 && maxOperatorShare < 1 {
+					svcLogger.Info().Float64("max_operator_share", maxOperatorShare).Msg("✅ QoS: per-operator concentration cap ENABLED")
+				}
+			} else {
+				// Every current QoS type implements SetMaxOperatorShare; a new type that
+				// forgets it would silently run uncapped despite the shipped-on default.
+				svcLogger.Warn().Msg("⚠️ QoS type does not support the per-operator concentration cap; selection runs uncapped")
+			}
+		}
 	}
 
 	hydratedLogger.Info().Msgf("Initialized %d QoS service instances", len(qosServices))

config/config.schema.yaml

@@ -626,6 +626,9 @@ properties:
                   enum: ["json_rpc", "rest", "comet_bft", "websocket", "grpc"]
             health_checks:
               $ref: "#/definitions/service_health_check_override"
+            max_operator_share:
+              description: "Per-operator (registrable domain / eTLD+1) concentration cap for endpoint selection. Caps the fraction of this service's selections any single operator may receive; excess is spread across the other valid operators. Bounds the blast radius of one operator failing when it holds most of the valid pool. Range (0, 1); a value <= 0 or >= 1 disables the cap. If unset, the global default (0.65) applies."
+              type: number
 
   # Logger Configuration (optional)
   logger_config:

config/examples/config.shannon_example.yaml

@@ -259,6 +259,8 @@ gateway_config:
   # - fallback: Fallback endpoints (no defaults - must be explicitly configured)
   # - health_checks: Per-service health check rules
   # - external_block_sources: External RPC endpoints for ground-truth block height validation
+  # - static_routes: Fixed responses served by the gateway for a given path, without
+  #   relaying to a backend (e.g. a small service metadata/version endpoint)
 
   services:
     # -------------------------------------------------------------------------
@@ -277,6 +279,20 @@ gateway_config:
       retry_config:
         max_retries: 2
         hedge_delay: 300ms
+      # Per-operator concentration cap (optional). Bounds the fraction of this service's
+      # endpoint selections that any single operator (registrable domain / eTLD+1) may
+      # receive; the excess is spread across the other valid operators. This limits the
+      # blast radius when one operator holds most of the valid pool and then degrades.
+      # Ships ON by default at 0.65 for every service — set a value here to override, or
+      # 1.0 (or 0) to disable it for this service.
+      # max_operator_share: 0.6
+      # WebSocket rebind strategy (optional). When true (the default), a session-rollover
+      # rebind spreads the target uniformly across operators (each provider equally likely),
+      # then picks an endpoint within it — maximizing per-provider spread. When false, the
+      # rebind uses the endpoint-count-weighted max_operator_share cap instead. Note:
+      # operator-uniform gives a single-endpoint operator the same share as a large one, so
+      # disable it per-service if that overloads a small provider.
+      # websocket_rebind_operator_uniform: true
       # External block sources for ground-truth validation (optional)
       # Multiple sources provide redundancy — the max height across all is used.
       # If all session endpoints are behind the real chain tip, these correct it.
@@ -288,6 +304,19 @@ gateway_config:
       #   - url: "https://rpc.ankr.com/eth"
       #     interval: 30s
       #     timeout: 5s
+      # Static routes (optional): serve a fixed response for a path directly from the
+      # gateway, bypassing endpoint selection and the relay. Matching is exact on `path`
+      # (evaluated after the /v1 and portal-app-id prefixes are stripped) and, when set,
+      # on the request method. A per-service route overrides a global default on the same
+      # path. `path: "/"` is rejected (it would shadow all relay traffic).
+      # static_routes:
+      #   - path: /example
+      #     # methods: ["GET"]              # optional; empty = any method
+      #     # status_code: 200              # optional; default 200
+      #     # content_type: text/plain      # optional; default text/plain
+      #     body: "<value-served-at-this-path>"
+      #     # headers:                       # optional extra response headers
+      #     #   Cache-Control: "public, max-age=3600"
 
     - id: poly
       type: "evm"
@@ -340,10 +369,14 @@ gateway_config:
       latency_profile: "standard"
       timeout_config:
         relay_timeout: 15s
-      # Solana external block source — uses "getBlockHeight" method (numeric result)
+      # Solana external block source — use "getEpochInfo" and read its blockHeight field.
+      # Do NOT use "getBlockHeight": it returns a bare number that some providers mislabel
+      # with the SLOT (~5% higher than block height), which poisons the max-based perceived
+      # height and makes honest endpoints look "behind". getEpochInfo.blockHeight is explicit
+      # and unambiguous. (The external floor is also plausibility-guarded as defense in depth.)
       # external_block_sources:
-      #   - url: "https://api.mainnet-beta.solana.com"
-      #     method: "getBlockHeight"
+      #   - url: "https://solana-rpc.publicnode.com"
+      #     method: "getEpochInfo"
       #     interval: 15s
       #     timeout: 5s
 

gateway/chain_state_admin.go

@@ -0,0 +1,50 @@
+package gateway
+
+import (
+	"context"
+
+	"github.com/pokt-network/path/protocol"
+)
+
+// perceivedBlockResetter is implemented by QoS instances that track a perceived block
+// height and can reset it (in-memory + Redis). EVM, Cosmos, Solana, and NoOp QoS
+// implement it; QoS types without a perceived height (e.g. generic JSON-RPC) do not.
+type perceivedBlockResetter interface {
+	ResetPerceivedBlockHeight(ctx context.Context) error
+}
+
+// ChainStateAdmin resets per-service chain state (perceived block height) via admin
+// endpoints. It backs POST /admin/chain-state/clear/{serviceId}.
+//
+// Perceived block height is a monotonic max floor held in per-pod memory and mirrored to
+// Redis; neither the max-based consensus nor the external floor can LOWER it, so a
+// too-high value (e.g. a slot-poisoned Solana height, or a value raised by a mislabeled
+// external source) cannot self-correct. This admin path deletes it so it rebuilds from
+// fresh endpoint observations. Like the circuit-breaker clear, it must be called on each
+// pod because the in-memory floor is per-pod.
+type ChainStateAdmin struct {
+	qosInstances map[protocol.ServiceID]QoSService
+}
+
+// NewChainStateAdmin builds a ChainStateAdmin over the given QoS instances.
+func NewChainStateAdmin(qosInstances map[protocol.ServiceID]QoSService) *ChainStateAdmin {
+	return &ChainStateAdmin{qosInstances: qosInstances}
+}
+
+// ResetChainState clears the perceived block height (in-memory + Redis) for a service.
+// Returns found=false if there is no such service or its QoS does not track a perceived
+// block height; err is non-nil only if the reset itself failed.
+func (a *ChainStateAdmin) ResetChainState(ctx context.Context, serviceID string) (found bool, err error) {
+	qos, ok := a.qosInstances[protocol.ServiceID(serviceID)]
+	if !ok {
+		return false, nil
+	}
+	resetter, ok := qos.(perceivedBlockResetter)
+	if !ok {
+		return false, nil
+	}
+	if err := resetter.ResetPerceivedBlockHeight(ctx); err != nil {
+		return true, err
+	}
+	return true, nil
+}

gateway/concentration_cap_config_test.go

@@ -0,0 +1,80 @@
+package gateway
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/pokt-network/path/protocol"
+)
+
+func TestGetMaxOperatorShareForService(t *testing.T) {
+	f := func(v float64) *float64 { return &v }
+
+	tests := []struct {
+		name    string
+		cfg     *UnifiedServicesConfig
+		service protocol.ServiceID
+		want    float64
+	}{
+		{
+			name:    "unset anywhere → global default",
+			cfg:     &UnifiedServicesConfig{Services: []ServiceConfig{{ID: "eth"}}},
+			service: "eth",
+			want:    DefaultMaxOperatorShare,
+		},
+		{
+			name:    "unknown service → global default",
+			cfg:     &UnifiedServicesConfig{},
+			service: "missing",
+			want:    DefaultMaxOperatorShare,
+		},
+		{
+			name: "defaults set, no per-service → defaults",
+			cfg: &UnifiedServicesConfig{
+				Defaults: ServiceDefaults{MaxOperatorShare: f(0.6)},
+				Services: []ServiceConfig{{ID: "eth"}},
+			},
+			service: "eth",
+			want:    0.6,
+		},
+		{
+			name: "per-service overrides defaults",
+			cfg: &UnifiedServicesConfig{
+				Defaults: ServiceDefaults{MaxOperatorShare: f(0.6)},
+				Services: []ServiceConfig{{ID: "eth", MaxOperatorShare: f(0.5)}},
+			},
+			service: "eth",
+			want:    0.5,
+		},
+		{
+			name: "per-service disable (1.0) is honored",
+			cfg: &UnifiedServicesConfig{
+				Services: []ServiceConfig{{ID: "eth", MaxOperatorShare: f(1.0)}},
+			},
+			service: "eth",
+			want:    1.0,
+		},
+		{
+			name: "per-service disable (0) is honored",
+			cfg: &UnifiedServicesConfig{
+				Services: []ServiceConfig{{ID: "eth", MaxOperatorShare: f(0)}},
+			},
+			service: "eth",
+			want:    0,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.want, tc.cfg.GetMaxOperatorShareForService(tc.service))
+		})
+	}
+}
+
+// TestDefaultMaxOperatorShare_ShippedOn documents that the cap is ON by default and set to
+// a sane in-range value (a change here is a deliberate rollout decision).
+func TestDefaultMaxOperatorShare_ShippedOn(t *testing.T) {
+	require.Greater(t, DefaultMaxOperatorShare, 0.0, "default must be enabled (> 0)")
+	require.Less(t, DefaultMaxOperatorShare, 1.0, "default must be an actual cap (< 1)")
+}

gateway/extract_block_height_test.go

@@ -0,0 +1,59 @@
+package gateway
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestExtractBlockHeight_GetEpochInfo verifies the external block fetcher can parse a
+// Solana getEpochInfo response by its explicit blockHeight field, and that it NEVER
+// substitutes absoluteSlot (the slot, ~5% higher, which poisons the max-based perceived
+// height). getEpochInfo is the preferred Solana external-source method precisely because
+// getBlockHeight returns a bare number some providers mislabel with the slot.
+func TestExtractBlockHeight_GetEpochInfo(t *testing.T) {
+	tests := []struct {
+		name     string
+		body     string
+		want     int64
+		wantErr  bool
+	}{
+		{
+			name: "getEpochInfo uses blockHeight, not absoluteSlot",
+			body: `{"jsonrpc":"2.0","id":1,"result":{"absoluteSlot":434344678,"blockHeight":412405238,"epoch":1005}}`,
+			want: 412405238,
+		},
+		{
+			name:    "getEpochInfo with only absoluteSlot (no blockHeight) must error, not return the slot",
+			body:    `{"jsonrpc":"2.0","id":1,"result":{"absoluteSlot":434344678,"epoch":1005}}`,
+			wantErr: true,
+		},
+		{
+			name: "getBlockHeight bare number still works",
+			body: `{"jsonrpc":"2.0","id":1,"result":412405238}`,
+			want: 412405238,
+		},
+		{
+			name: "EVM hex still works",
+			body: `{"jsonrpc":"2.0","id":1,"result":"0x1940c6f5"}`,
+			want: 0x1940c6f5,
+		},
+		{
+			name: "Cosmos sync_info still works",
+			body: `{"jsonrpc":"2.0","id":1,"result":{"sync_info":{"latest_block_height":"12345"}}}`,
+			want: 12345,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := extractBlockHeight([]byte(tc.body))
+			if tc.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.want, got)
+		})
+	}
+}

gateway/health_check_config.go

@@ -271,6 +271,14 @@ type (
 		// Local defines health checks in the local config.
 		// These override any checks from External with the same service_id + name.
 		Local []ServiceHealthCheckConfig `yaml:"local,omitempty"`
+		// BackendDedup enables backend-URL deduplication of health check relays.
+		// Many staked suppliers front the SAME backend URL (measured: 2-4x redundancy
+		// per service, up to 10 suppliers per URL). When enabled, each cycle fires ONE
+		// relay per unique backend URL (a rotating representative supplier) and fans the
+		// backend-derived result (reputation signal, block height, archival status) to the
+		// other suppliers on that URL — cutting HC relay volume by the redundancy factor.
+		// WebSocket checks are never deduped (connectivity is per-endpoint). Default: true.
+		BackendDedup *bool `yaml:"backend_dedup,omitempty"`
 	}
 
 	// RetryConfig configures automatic retry behavior for failed requests.
@@ -345,6 +353,12 @@ func (hc *ActiveHealthChecksConfig) HydrateDefaults(hasRedis bool) {
 		hc.SyncAllowance = DefaultSyncAllowance
 	}
 
+	// Backend-URL dedup defaults to enabled.
+	if hc.BackendDedup == nil {
+		enabled := true
+		hc.BackendDedup = &enabled
+	}
+
 	// Hydrate coordination defaults
 	hc.Coordination.HydrateDefaults()