extproc: allow negative server window increments per spec

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019LK7JqPusvaySyrJFUPbsH

Author: nvxbug

internal/xds/httpfilter/extproc/ext_proc.go

@@ -2040,25 +2040,26 @@ func (cs *clientStream) waitForTrailerProcessing(recvErr error) error {
 
 // applyServerWindowUpdate adds a flow control window increment received from
 // the external processor to window, waking a blocked acquirer if the window
-// becomes positive. The increment is server-controlled, so it must be
-// non-negative and must not overflow the int64 window; an increment that
-// violates either is a protocol error and returns one instead of wrapping the
-// accounting, matching how HTTP/2 flow control rejects a window update that
-// exceeds the maximum. This recv loop is the only writer that grows the
-// window, so the compare-and-swap only contends with a concurrent deduction in
+// becomes positive. The increment may be negative: the ext_proc spec allows a
+// server to shrink the window, in which case senders block until later
+// updates bring it back above zero. The increment is server-controlled
+// though, so one that would wrap the int64 window in either direction is a
+// protocol error and returns one instead of corrupting the accounting. This
+// recv loop is the only writer that grows the window, so the compare-and-swap
+// only contends with a concurrent deduction in
 // acquire{Downstream,Upstream}ToSidestreamWindow.
 func applyServerWindowUpdate(window *atomic.Int64, positiveUpdateCh chan struct{}, delta int64) error {
 	if delta == 0 {
 		return nil
 	}
-	if delta < 0 {
-		return fmt.Errorf("negative window increment %d", delta)
-	}
 	for {
 		previousQuota := window.Load()
-		if previousQuota > math.MaxInt64-delta {
+		if delta > 0 && previousQuota > math.MaxInt64-delta {
 			return fmt.Errorf("window increment %d overflows the current window %d", delta, previousQuota)
 		}
+		if delta < 0 && previousQuota < math.MinInt64-delta {
+			return fmt.Errorf("window increment %d underflows the current window %d", delta, previousQuota)
+		}
 		newQuota := previousQuota + delta
 		if !window.CompareAndSwap(previousQuota, newQuota) {
 			continue

internal/xds/httpfilter/extproc/ext_proc_test.go

@@ -27,10 +27,11 @@ import (
 )
 
 // TestApplyServerWindowUpdate verifies that a flow control window increment
-// from the external processor is applied only when it is a non-negative value
-// that does not overflow the window. An out-of-range increment must be
-// rejected without mutating the window, so the accounting cannot wrap to a
-// value that permanently blocks acquireDownstreamToSidestreamWindow.
+// from the external processor is applied whenever it does not wrap the int64
+// window, including negative increments, which the ext_proc spec allows. An
+// increment that would overflow or underflow must be rejected without
+// mutating the window, so the accounting cannot wrap to a value that
+// permanently blocks acquireDownstreamToSidestreamWindow.
 func TestApplyServerWindowUpdate(t *testing.T) {
 	tests := []struct {
 		name       string
@@ -67,11 +68,23 @@ func TestApplyServerWindowUpdate(t *testing.T) {
 			wantWindow: iextproc.DefaultFlowControlWindowSize,
 		},
 		{
-			name:       "negative increment is rejected and leaves the window unchanged",
+			name:       "negative increment shrinks the window",
 			start:      iextproc.DefaultFlowControlWindowSize,
-			delta:      -1,
+			delta:      -1024,
+			wantWindow: iextproc.DefaultFlowControlWindowSize - 1024,
+		},
+		{
+			name:       "negative increment can drive the window below zero",
+			start:      1024,
+			delta:      -2048,
+			wantWindow: -1024,
+		},
+		{
+			name:       "underflowing increment is rejected and leaves the window unchanged",
+			start:      math.MinInt64 + 10,
+			delta:      -1024,
 			wantErr:    true,
-			wantWindow: iextproc.DefaultFlowControlWindowSize,
+			wantWindow: math.MinInt64 + 10,
 		},
 	}
 	for _, test := range tests {