atenet-router: raise actor-cluster circuit breakers

cmd/atenet/internal/router/cmd.go

@@ -75,7 +75,7 @@ func NewRouterCmd() *cobra.Command {
 	cmd.Flags().DurationVar(&cfg.ParkedRequest.RetryInterval, "parked-request-retry-interval", defaultParkedRequestRetryInterval, "Delay before a parked request's first resume retry")
 	cmd.Flags().Float64Var(&cfg.ParkedRequest.RetryFactor, "parked-request-retry-factor", defaultParkedRequestRetryFactor, "Multiplier applied to the retry delay after each attempt; must be >= 1")
 	cmd.Flags().Float64Var(&cfg.ParkedRequest.RetryJitter, "parked-request-retry-jitter", defaultParkedRequestRetryJitter, "Random fraction in [0, 1) added to each retry delay to de-synchronize parked requests")
-	cmd.Flags().IntVar(&cfg.ExtProcMaxRequests, "extproc-max-requests", 0, "Circuit-breaker max_requests for Envoy's ext_proc cluster; 0 (the default) derives it as twice --parked-request-max (minimum 1024). Explicit values must be >= --parked-request-max: every parked request holds one slot for its full wait, and the excess is fast-path headroom")
+	cmd.Flags().IntVar(&cfg.ExtProcMaxRequests, "extproc-max-requests", 0, "Circuit-breaker ceiling for Envoy's ext_proc cluster, applied to max_requests and max_pending_requests alike; 0 (the default) derives it as twice --parked-request-max (minimum 1024). Explicit values must be >= --parked-request-max: every parked request holds one slot for its full wait, and the excess is fast-path headroom")
 	// Graceful shutdown knobs. The router sits behind a Service, so
 	// route-drain window is needed: after SIGTERM the readiness flip
 	// must propagate to the Service endpoints before the drain starts.

cmd/atenet/internal/router/xds.go

@@ -89,10 +89,11 @@ const (
 // otherwise Envoy abandons a parked request (500) long before the router does.
 const defaultExtProcMessageTimeout = 5 * time.Second
 
-// defaultExtProcMaxRequests is the circuit-breaker max_requests set on the
-// ext_proc cluster: defaultParkedRequestMax plus equal fast-path headroom, so a
-// full parking lot cannot starve the millisecond-scale header exchanges of
-// requests to already-running actors. See buildCluster.
+// defaultExtProcMaxRequests is the circuit-breaker ceiling set on the ext_proc
+// cluster, applied to both max_requests and max_pending_requests:
+// defaultParkedRequestMax plus equal fast-path headroom, so a full parking lot
+// cannot starve the millisecond-scale header exchanges of requests to
+// already-running actors. See buildCluster.
 const defaultExtProcMaxRequests = 2048
 
 // defaultRouteTimeout is Envoy's end-to-end route timeout for workload traffic:
@@ -121,6 +122,12 @@ const defaultRouteTimeout = 10 * time.Second
 // routeIdleTimeout.
 const envoyDefaultStreamIdleTimeout = 5 * time.Minute
 
+// actorClusterMaxConcurrency replaces Envoy's 1024 default, which is far below
+// what the router can carry. Kept under the ~28,232 ephemeral-port budget
+// (each in-flight HTTP/1.1 request holds one source port) so the breaker's
+// counted overflow trips before the kernel's opaque EADDRNOTAVAIL.
+const actorClusterMaxConcurrency = 20000
+
 // XdsServer implements an aggregated discovery service server for dynamic Envoy router nodes.
 type XdsServer struct {
 	xdsPort      int
@@ -160,10 +167,11 @@ type XdsServer struct {
 	// response. Must be >= the parking budget so parked requests aren't cut short.
 	extProcMessageTimeout time.Duration
 
-	// extProcMaxRequests is the circuit-breaker max_requests on the ext_proc
-	// cluster — the hard ceiling on concurrent requests held open against the
-	// router's processing server, parked requests included. Must be >= the
-	// parking lot size (enforced at startup in Run).
+	// extProcMaxRequests is the circuit-breaker ceiling on the ext_proc cluster
+	// — the hard limit on concurrent requests held open against the router's
+	// processing server, parked requests included. Applied to max_requests and
+	// max_pending_requests alike. Must be >= the parking lot size (enforced at
+	// startup in Run).
 	extProcMaxRequests uint32
 
 	// routeTimeout is Envoy's end-to-end timeout on the workload route. Actors
@@ -209,9 +217,10 @@ func (x *XdsServer) SetExtProcMessageTimeout(d time.Duration) {
 	}
 }
 
-// SetExtProcMaxRequests sets the circuit-breaker max_requests on the ext_proc
-// cluster. Size it to the parking lot plus fast-path headroom (validated in
-// Run()); a non-positive value leaves the default unchanged.
+// SetExtProcMaxRequests sets the circuit-breaker ceiling on the ext_proc
+// cluster, for max_requests and max_pending_requests together. Size it to the
+// parking lot plus fast-path headroom (validated in Run()); a non-positive
+// value leaves the default unchanged.
 func (x *XdsServer) SetExtProcMaxRequests(n int) {
 	x.mu.Lock()
 	defer x.mu.Unlock()
@@ -475,10 +484,15 @@ func (x *XdsServer) buildCluster() *clusterv3.Cluster {
 			Type: clusterv3.Cluster_STATIC,
 		},
 		LbPolicy: clusterv3.Cluster_ROUND_ROBIN,
+		// max_pending_requests rises with max_requests: a request is pending
+		// until the pool hands it a stream, and a shallower pending queue just
+		// moves the rejection. max_connections and max_retries keep their
+		// defaults — one connection per worker thread, and no retry policy.
 		CircuitBreakers: &clusterv3.CircuitBreakers{
 			Thresholds: []*clusterv3.CircuitBreakers_Thresholds{{
-				Priority:    corev3.RoutingPriority_DEFAULT,
-				MaxRequests: wrapperspb.UInt32(x.extProcMaxRequests),
+				Priority:           corev3.RoutingPriority_DEFAULT,
+				MaxRequests:        wrapperspb.UInt32(x.extProcMaxRequests),
+				MaxPendingRequests: wrapperspb.UInt32(x.extProcMaxRequests),
 			}},
 		},
 		LoadAssignment: &endpointv3.ClusterLoadAssignment{
@@ -639,6 +653,18 @@ func (x *XdsServer) buildOriginalDstCluster() *clusterv3.Cluster {
 				HttpHeaderName: OriginalDstHeader,
 			},
 		},
+		// Connections, pending and requests are lifted together: the upstream
+		// hop is HTTP/1.1, so capping any one below the others just moves
+		// where the queue forms. max_retries keeps Envoy's default — no route
+		// to this cluster sets a retry policy.
+		CircuitBreakers: &clusterv3.CircuitBreakers{
+			Thresholds: []*clusterv3.CircuitBreakers_Thresholds{{
+				Priority:           corev3.RoutingPriority_DEFAULT,
+				MaxConnections:     wrapperspb.UInt32(actorClusterMaxConcurrency),
+				MaxPendingRequests: wrapperspb.UInt32(actorClusterMaxConcurrency),
+				MaxRequests:        wrapperspb.UInt32(actorClusterMaxConcurrency),
+			}},
+		},
 	}
 
 	if ts := x.buildUpstreamTransportSocket(); ts != nil {

cmd/atenet/internal/router/xds_test.go

@@ -562,6 +562,59 @@ func TestXdsServer_ExtProcCircuitBreaker(t *testing.T) {
 			t.Errorf("max_requests after SetExtProcMaxRequests(0) = %d, want default %d", got, defaultExtProcMaxRequests)
 		}
 	})
+
+	// A request is pending until the pool hands it a stream, so the shallower
+	// of the two breakers is the one a burst actually meets.
+	t.Run("PendingTracksMaxRequests", func(t *testing.T) {
+		for _, set := range []int{0, 4096, 20000} {
+			x := NewXdsServer(0)
+			x.SetExtProcMaxRequests(set)
+			th := x.buildCluster().GetCircuitBreakers().GetThresholds()[0]
+			req, pend := th.GetMaxRequests().GetValue(), th.GetMaxPendingRequests().GetValue()
+			if pend != req {
+				t.Errorf("SetExtProcMaxRequests(%d): max_pending_requests = %d, max_requests = %d; the shallower one is the real ceiling",
+					set, pend, req)
+			}
+		}
+	})
+}
+
+func TestXdsServer_ActorClusterCircuitBreaker(t *testing.T) {
+	thresholds := NewXdsServer(0).buildOriginalDstCluster().GetCircuitBreakers().GetThresholds()
+	if len(thresholds) != 1 {
+		t.Fatalf("got %d thresholds, want 1 (on the default priority)", len(thresholds))
+	}
+	th := thresholds[0]
+	if got := th.GetPriority(); got != corev3.RoutingPriority_DEFAULT {
+		t.Errorf("priority = %v, want DEFAULT: actor traffic carries no priority header, so a HIGH threshold would never apply", got)
+	}
+
+	// Connections, pending and requests rise together; any one left at the
+	// 1024 default just relocates the queue. max_retries stays unset: no
+	// route to this cluster retries, and a 20k retry budget would be noise.
+	for _, tc := range []struct {
+		name string
+		got  uint32
+	}{
+		{"max_connections", th.GetMaxConnections().GetValue()},
+		{"max_pending_requests", th.GetMaxPendingRequests().GetValue()},
+		{"max_requests", th.GetMaxRequests().GetValue()},
+	} {
+		if tc.got != uint32(actorClusterMaxConcurrency) {
+			t.Errorf("%s = %d, want %d", tc.name, tc.got, actorClusterMaxConcurrency)
+		}
+	}
+	if th.GetMaxRetries() != nil {
+		t.Errorf("max_retries = %d, want unset (Envoy default): nothing on this cluster retries", th.GetMaxRetries().GetValue())
+	}
+
+	// Each in-flight HTTP/1.1 request holds one ephemeral port; a breaker
+	// above the port budget would let the kernel fail first, uncounted.
+	const defaultEphemeralPorts = 60999 - 32768 + 1
+	if actorClusterMaxConcurrency >= defaultEphemeralPorts {
+		t.Errorf("actorClusterMaxConcurrency (%d) >= the default ephemeral port range (%d): the kernel, not the breaker, becomes the binding limit",
+			actorClusterMaxConcurrency, defaultEphemeralPorts)
+	}
 }
 
 func TestXdsServer_RouteTimeout(t *testing.T) {

manifests/ate-install/atenet-router.yaml

@@ -60,6 +60,11 @@ data:
           address: 0.0.0.0
           port_value: 9901
 
+    # Per-worker event-loop duration histograms; the documented overhead is
+    # accepted for the observability.
+    enable_dispatcher_stats: true
+
+
     node:
       id: substrate-envoy-node
       cluster: substrate-router-cluster
@@ -158,6 +163,15 @@ spec:
         # turns to survive a shutdown, raise --drain-timeout and
         # terminationGracePeriodSeconds alongside it.
         # - "--route-timeout=5m"
+        resources:
+          # Small requests so the pod schedules anywhere (including CI kind
+          # nodes); the limits are the real cap.
+          requests:
+            cpu: 250m
+            memory: 256Mi
+          limits:
+            cpu: "8"
+            memory: 2Gi
         env:
         - name: POD_NAME
           valueFrom:
@@ -217,13 +231,22 @@ spec:
         - name: "drain-signal"
           mountPath: "/var/run/atenet"
       - name: envoy
+        # Not v1.39+: it measured 20-38% lower sustained throughput on this
+        # workload.
         image: envoyproxy/envoy:v1.30-latest
         command:
         - "/usr/local/bin/envoy"
         - "-c"
         - "/etc/envoy/envoy.yaml"
         - "--component-log-level"
         - "upstream:debug,router:debug,ext_proc:debug"
+        # Envoy sizes worker threads from the node's CPU count unless told
+        # otherwise; on a large node that is dozens of event loops.
+        - "--concurrency"
+        - "8"
+        # Observability: mutex-contention counters on the admin /contention
+        # endpoint. Costs one atomic per contended acquisition.
+        - "--enable-mutex-tracing"
         # Prevents Envoy from fast-exiting on SIGTERM before atenet-router finishes
         # its drain sequence. Polls for the drain-complete marker written by the
         # router on the shared emptyDir, terminating Envoy as soon as the drain
@@ -232,6 +255,17 @@ spec:
           preStop:
             exec:
               command: ["sh", "-c", "while [ ! -f /var/run/atenet/drain-complete ]; do sleep 0.5; done"]
+        resources:
+          # Small requests, real cap in limits (see the sidecar's note); keep
+          # the CPU limit in step with --concurrency above. Do not raise it
+          # past 8 cores: measured capacity per core falls steeply beyond
+          # that — scale by adding router replicas instead.
+          requests:
+            cpu: 250m
+            memory: 256Mi
+          limits:
+            cpu: "8"
+            memory: 4Gi
         ports:
         - name: http
           containerPort: 8080