Merge branch 'origin/main' into 'next-release/main'

go.mod

@@ -116,7 +116,7 @@ require (
 	golang.org/x/term v0.47.0
 	golang.org/x/text v0.43.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc
-	google.golang.org/grpc v1.84.0
+	google.golang.org/grpc v1.85.0
 	google.golang.org/protobuf v1.36.12
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1

go.sum

@@ -1423,8 +1423,8 @@ google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3Iji
 google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.31.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0=
-google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0=
+google.golang.org/grpc v1.85.0 h1:BPBCfuVRQ1qvBhKDOq9uvQllMvO59rKsYhXpp7Dd8vw=
+google.golang.org/grpc v1.85.0/go.mod h1:0rQrOD58yjal/0EaVbGBANindixsCZNQPDV3VKtaKuQ=
 google.golang.org/grpc/examples v0.0.0-20211102180624-670c133e568e h1:m7aQHHqd0q89mRwhwS9Bx2rjyl/hsFAeta+uGrHsQaU=
 google.golang.org/grpc/examples v0.0.0-20211102180624-670c133e568e/go.mod h1:gID3PKrg7pWKntu9Ss6zTLJ0ttC0X9IHgREOCZwbCVU=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=

vendor/google.golang.org/grpc/CONTRIBUTING.md

@@ -23,6 +23,29 @@ wanted"](https://github.com/grpc/grpc-go/issues?q=sort%3Aupdated-desc%20is%3Aiss
 are especially nice for first-time contributors, as they should be well-defined
 problems that already have agreed-upon solutions.
 
+## Generative AI Policy
+
+AI tools have the ability to produce more code than is possible for the gRPC
+team to read, understand, review, and accept into the repository.  For this
+reason, we request that all contributions adhere to the following rules:
+
+1. **No AI-Generated Interactions:** All communication in the repo must be
+   authored by a human.  _Exception: AIs may be used for directed writing
+   assistance or translation._  Absolutely no automated agents are allowed to
+   directly publish to GitHub.
+
+2. **Author Ownership and Accountability:** Code contributions are expected to
+   be fully owned and understood by the human contributor.  If the code was
+   produced by generative AI, the author is expected to have reviewed and
+   understood it in its entirety before submitting it for review.  This includes
+   all content: production code, tests, examples, tools, etc.
+
+In addition to the above requirements, any AI-assisted contributions must also
+comply with the [Linux Foundation Generative AI
+Policy](https://www.linuxfoundation.org/legal/generative-ai).  This includes
+confirming that all contributions are legally allowed to be contributed to the
+gRPC project under the applicable license terms.
+
 ## Code Style
 
 We follow [Google's published Go style

vendor/google.golang.org/grpc/binarylog/grpc_binarylog_v1/binarylog.pb.go

@@ -18,7 +18,7 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.36.11
+// 	protoc-gen-go v1.36.12
 // 	protoc        v5.27.1
 // source: grpc/binlog/v1/binarylog.proto
 

vendor/google.golang.org/grpc/call.go

@@ -20,10 +20,6 @@ package grpc
 
 import (
 	"context"
-	"io"
-
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 )
 
 // Invoke sends the RPC request on the wire and returns after response is
@@ -71,27 +67,8 @@ func invoke(ctx context.Context, method string, req, reply any, cc *ClientConn,
 	if err != nil {
 		return err
 	}
-	// In case of nil or io.EOF error, call RecvMsg.
-	if err := cs.SendMsg(req); err != nil && err != io.EOF {
+	if err := cs.SendMsg(req); err != nil {
 		return err
 	}
-	// CloseSend is needed because in some scenarios (e.g., xDS), the same
-	// interceptors are used to process both unary and streaming RPCs. Calling
-	// CloseSend signals to those interceptors that no more messages are on the
-	// way.
-	if err := cs.CloseSend(); err != nil && err != io.EOF {
-		return err
-	}
-	if err := cs.RecvMsg(reply); err != nil {
-		return err
-	}
-	// Call RecvMsg again to get the trailers.
-	err = cs.RecvMsg(reply)
-	if err == io.EOF {
-		return nil
-	}
-	if err == nil {
-		return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
-	}
-	return err
+	return cs.RecvMsg(reply)
 }

vendor/google.golang.org/grpc/clientconn.go

@@ -553,7 +553,6 @@ func chainStreamClientInterceptors(cc *ClientConn) {
 	if cc.dopts.streamInt != nil {
 		interceptors = append([]StreamClientInterceptor{cc.dopts.streamInt}, interceptors...)
 	}
-	interceptors = append(interceptors, defaultStreamInterceptor)
 	var chainedInt StreamClientInterceptor
 	if len(interceptors) == 0 {
 		chainedInt = nil
@@ -1212,6 +1211,12 @@ func (cc *ClientConn) Close() error {
 	cc.mu.Unlock()
 
 	cc.resolverWrapper.close()
+	// Swap in a default ConfigSelector so the resolver-provided one (which may
+	// transitively retain resolver-owned state such as parsed service configs
+	// and cluster maps) is released once in-flight RPCs finish selecting on
+	// it. Done after the resolver has closed so no further UpdateState calls
+	// race with this swap.
+	cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{nil})
 	// The order of closing matters here since the balancer wrapper assumes the
 	// picker is closed before it is closed.
 	cc.pickerWrapper.close()

vendor/google.golang.org/grpc/health/grpc_health_v1/health.pb.go

@@ -17,7 +17,7 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.36.11
+// 	protoc-gen-go v1.36.12
 // 	protoc        v5.27.1
 // source: grpc/health/v1/health.proto
 

vendor/google.golang.org/grpc/health/server.go

@@ -107,6 +107,9 @@ func (s *Server) Watch(in *healthpb.HealthCheckRequest, stream healthgrpc.Health
 	defer func() {
 		s.mu.Lock()
 		delete(s.updates[service], stream)
+		if len(s.updates[service]) == 0 {
+			delete(s.updates, service)
+		}
 		s.mu.Unlock()
 	}()
 	s.mu.Unlock()

vendor/google.golang.org/grpc/internal/binarylog/method_logger.go

@@ -395,7 +395,7 @@ func (c *Cancel) toProto() *binlogpb.GrpcLogEntry {
 // omitted.
 func metadataKeyOmit(key string) bool {
 	switch key {
-	case "lb-token", ":path", ":authority", "content-encoding", "content-type", "user-agent", "te":
+	case "lb-token", ":path", ":authority", "content-encoding", "content-type", "user-agent", "te", "authorization":
 		return true
 	case "grpc-trace-bin": // grpc-trace-bin is special because it's visible to users.
 		return false

vendor/google.golang.org/grpc/internal/envconfig/envconfig.go

@@ -144,14 +144,14 @@ var (
 
 	// ControlBufferThrottleLimit is the maximum number of control frames that can
 	// be queued in the control buffer before throttling is applied. The value
-	// must be between 1 and 10,000, and is set to 100 by default.
+	// must be between 1 and 100,000, and is set to 10,000 by default.
 	//
 	// This environment variable serves as an escape hatch to increase the
 	// throttling limit if unforeseen issues arise, and it will be removed in a
 	// future release.
 	//
-	// TODO: Remove this env var once v1.83.0 is released.
-	ControlBufferThrottleLimit = uint64FromEnv("GRPC_GO_EXPERIMENTAL_CONTROL_BUFFER_THROTTLE_LIMIT", 100, 1, 10000)
+	// TODO: Remove this env var once v1.86.0 is released.
+	ControlBufferThrottleLimit = uint64FromEnv("GRPC_GO_EXPERIMENTAL_CONTROL_BUFFER_THROTTLE_LIMIT", 10000, 1, 100000)
 
 	// ALTSMaxFrameSize is the maximum frame size for ALTS in bytes.
 	// This can be overridden by setting the environment variable

vendor/google.golang.org/grpc/internal/transport/bdp_estimator.go

@@ -47,10 +47,9 @@ const (
 var bdpPing = &ping{data: [8]byte{2, 4, 16, 16, 9, 14, 7, 7}}
 
 type bdpEstimator struct {
+	mu sync.Mutex
 	// sentAt is the time when the ping was sent.
 	sentAt time.Time
-
-	mu sync.Mutex
 	// bdp is the current bdp estimate.
 	bdp uint32
 	// sample is the number of bytes received in one measurement cycle.
@@ -75,7 +74,9 @@ func (b *bdpEstimator) timesnap(d [8]byte) {
 	if bdpPing.data != d {
 		return
 	}
+	b.mu.Lock()
 	b.sentAt = time.Now()
+	b.mu.Unlock()
 }
 
 // add adds bytes to the current sample for calculating bdp.

vendor/google.golang.org/grpc/internal/transport/controlbuf.go

@@ -26,6 +26,7 @@ import (
 	"runtime"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
@@ -97,22 +98,35 @@ func (il *itemList) isEmpty() bool {
 	return il.head == nil
 }
 
-// maxQueuedControlBufferItems is the maximum number of frames (other than
-// HEADERS and DATA) that we will buffer before preventing new reads from
-// occurring on the transport.  These are control frames sent in response to
-// client requests, or frames that result in work being scheduled, such as
-// RST_STREAM due to bad headers or settings acks.
-var maxQueuedControlBufferItems = int(envconfig.ControlBufferThrottleLimit)
+// maxQueuedTransportResponseFrames is the maximum number of "transport
+// response" frames that we will buffer before preventing new reads from
+// occurring on the transport. See cbItem.isTransportResponseFrame for the
+// items that count towards this limit.
+var maxQueuedTransportResponseFrames = int(envconfig.ControlBufferThrottleLimit)
 
 type cbItem interface {
-	isThrottled() bool
+	// isTransportResponseFrame reports whether the item counts against the
+	// maxQueuedTransportResponseFrames limit, when queued in the
+	// controlBuffer of a transport on the given side.
+	//
+	// Items that the peer can make the transport queue in unbounded
+	// numbers, for example by sending PING or SETTINGS frames, must be
+	// counted, so that the transport stops reading from a peer that sends
+	// such frames without reading what the transport writes. Items whose
+	// number grows with the local application's load (for example, one
+	// per stream it creates) must not be counted. Otherwise, under high
+	// concurrency, both peers could stop reading at the same time and
+	// deadlock (see https://github.com/grpc/grpc-go/issues/9330).
+	isTransportResponseFrame(s side) bool
 }
 
-// throttledItem represents every item in the controlBuffer to which the overall
-// throttling limit applies, other than outgoing HEADERS and DATA frames.
-type throttledItem struct{}
+// nonTransportResponseFrame represents items in the controlBuffer that are not
+// queued in response to peer control/reset frames (for example,
+// locally-initiated or flow-controlled items). These items are not counted
+// against the maxQueuedTransportResponseFrames limit.
+type nonTransportResponseFrame struct{}
 
-func (throttledItem) isThrottled() bool { return true }
+func (nonTransportResponseFrame) isTransportResponseFrame(side) bool { return false }
 
 // The following defines various control items which could flow through
 // the control buffer of transport. They represent different aspects of
@@ -120,12 +134,13 @@ func (throttledItem) isThrottled() bool { return true }
 
 // registerStream is used to register an incoming stream with loopy writer.
 type registerStream struct {
-	throttledItem
+	nonTransportResponseFrame
 	streamID uint32
 	wq       *writeQuota
 }
 
 type clientHeaders struct {
+	nonTransportResponseFrame
 	streamID   uint32
 	hf         []hpack.HeaderField
 	initStream func(uint32) error
@@ -134,9 +149,8 @@ type clientHeaders struct {
 	onOrphaned func(error)
 }
 
-func (*clientHeaders) isThrottled() bool { return false }
-
 type serverHeaders struct {
+	nonTransportResponseFrame
 	streamID  uint32
 	hf        []hpack.HeaderField
 	endStream bool
@@ -144,24 +158,37 @@ type serverHeaders struct {
 	cleanup   *cleanupStream
 }
 
-func (h *serverHeaders) isThrottled() bool { return false }
-
 type cleanupStream struct {
-	throttledItem
 	streamID uint32
 	rst      bool
 	rstCode  http2.ErrCode
 	onWrite  func()
 }
 
+// isTransportResponseFrame returns true on the server since cleanupStream
+// either results in a RST_STREAM frame being sent on the wire or in
+// non-trivial work being performed when under a Rapid Reset attack.
+//
+// It returns false on the client, where a cleanupStream is queued at most once
+// per stream, and the number of streams is controlled by the client and not by
+// the server. Counting them on the client makes the client stop reading under
+// high concurrency, which deadlocks with a server that has also stopped
+// reading (see https://github.com/grpc/grpc-go/issues/9330).
+func (*cleanupStream) isTransportResponseFrame(s side) bool { return s == serverSide }
+
 type earlyAbortStream struct {
-	throttledItem
 	streamID uint32
 	rst      bool
 	hf       []hpack.HeaderField // Pre-built header fields
 }
 
+// isTransportResponseFrame returns true on the server since earlyAbortStream
+// results in a HEADERS frame being sent on the wire, and possibly a RST_STREAM
+// frame. It is never queued on the client.
+func (*earlyAbortStream) isTransportResponseFrame(s side) bool { return s == serverSide }
+
 type dataFrame struct {
+	nonTransportResponseFrame
 	streamID   uint32
 	endStream  bool
 	h          []byte
@@ -172,36 +199,40 @@ type dataFrame struct {
 	onEachWrite func()
 }
 
-func (*dataFrame) isThrottled() bool { return false }
-
 type incomingWindowUpdate struct {
-	throttledItem
 	streamID  uint32
 	increment uint32
 }
 
+// isTransportResponseFrame returns true to protect against a WINDOW_UPDATE
+// flood attack from the peer.
+func (*incomingWindowUpdate) isTransportResponseFrame(side) bool { return true }
+
 type outgoingWindowUpdate struct {
-	throttledItem
+	nonTransportResponseFrame
 	streamID  uint32
 	increment uint32
 }
 
 type incomingSettings struct {
-	throttledItem
 	ss []http2.Setting
 }
 
+// isTransportResponseFrame returns true since incomingSettings results in a
+// SETTINGS ACK frame being sent on the wire.
+func (*incomingSettings) isTransportResponseFrame(side) bool { return true }
+
 type outgoingSettings struct {
-	throttledItem
+	nonTransportResponseFrame
 	ss []http2.Setting
 }
 
 type incomingGoAway struct {
-	throttledItem
+	nonTransportResponseFrame
 }
 
 type goAway struct {
-	throttledItem
+	nonTransportResponseFrame
 	code      http2.ErrCode
 	debugData []byte
 	headsUp   bool
@@ -209,13 +240,23 @@ type goAway struct {
 }
 
 type ping struct {
-	throttledItem
 	ack  bool
 	data [8]byte
 }
 
+// isTransportResponseFrame returns true for all pings, and not only for ping
+// acks sent in response to pings from the peer. This is because a BDP ping is
+// queued in response to a DATA frame (of any size) from the peer, whenever
+// there is no BDP ping outstanding. And since the BDP ping's payload is fixed,
+// the peer can make it look like the BDP ping was acked at any time, by
+// sending a ping ack with that payload. A peer alternating between empty DATA
+// frames and such ping acks would otherwise be able to queue an unbounded
+// number of BDP pings. Pings initiated locally (keepalive and BDP) are rare
+// enough to not cause throttling of well-behaved peers.
+func (*ping) isTransportResponseFrame(side) bool { return true }
+
 type outFlowControlSizeRequest struct {
-	throttledItem
+	nonTransportResponseFrame
 	resp chan uint32
 }
 
@@ -225,7 +266,7 @@ type outFlowControlSizeRequest struct {
 // can inspect settled accounting values (such as bytesOutStanding) after loopyWriter
 // has run to completion without any data races.
 type outStreamRequestForTesting struct {
-	throttledItem
+	nonTransportResponseFrame
 	streamID uint32
 	resp     chan *outStream
 }
@@ -235,7 +276,7 @@ type outStreamRequestForTesting struct {
 // (by the client or server).  The transport itself will close after the reader
 // encounters the EOF caused by the connection closure.
 type closeConnection struct {
-	throttledItem
+	nonTransportResponseFrame
 }
 
 type outStreamState int
@@ -316,8 +357,9 @@ func (l *outStreamList) dequeue() *outStream {
 // shouldn't be confused with an HTTP2 frame, although some of the control
 // frames like dataFrame and headerFrame do go out on wire as HTTP2 frames.
 type controlBuffer struct {
-	wakeupCh chan struct{}   // Unblocks readers waiting for something to read.
+	wakeupCh chan struct{}   // Unblocks readers waiting for new items.
 	done     <-chan struct{} // Closed when the transport is done.
+	side     side            // Side of the transport owning this buffer.
 
 	// Mutex guards all the fields below, except trfChan which can be read
 	// atomically without holding mu.
@@ -326,26 +368,30 @@ type controlBuffer struct {
 	closed          bool      // True when the controlbuf is finished.
 	list            *itemList // List of queued control frames.
 
-	// transportResponseFrames counts the number of queued items that represent
-	// the response of an action initiated by the peer.  trfChan is created
-	// when transportResponseFrames >= maxQueuedTransportResponseFrames and is
-	// closed and nilled when transportResponseFrames drops below the
-	// threshold.  Both fields are protected by mu.
+	// transportResponseFrames counts the number of queued items that
+	// represent the response of an action initiated by the peer (as
+	// determined by cbItem.isTransportResponseFrame for this side).
+	// trfChan is created when transportResponseFrames >=
+	// maxQueuedTransportResponseFrames and is closed and nilled when
+	// transportResponseFrames drops below the threshold. Both fields are
+	// protected by mu.
 	transportResponseFrames int
 	trfChan                 atomic.Pointer[chan struct{}]
 }
 
-func newControlBuffer(done <-chan struct{}) *controlBuffer {
+func newControlBuffer(done <-chan struct{}, s side) *controlBuffer {
 	return &controlBuffer{
 		wakeupCh: make(chan struct{}, 1),
 		list:     &itemList{},
 		done:     done,
+		side:     s,
 	}
 }
 
-// throttle blocks if there are too many frames in the control buf that
-// represent the response of an action initiated by the peer, like
-// incomingSettings cleanupStreams etc.
+// throttle blocks if there are too many items in the control buffer that count
+// towards the maxQueuedTransportResponseFrames limit (see
+// cbItem.isTransportResponseFrame), until their number drops below the limit,
+// the control buffer is finished, or the transport is done.
 func (c *controlBuffer) throttle() {
 	if ch := c.trfChan.Load(); ch != nil {
 		select {
@@ -355,6 +401,25 @@ func (c *controlBuffer) throttle() {
 	}
 }
 
+// throttleWithTimeout is like throttle, but it stops waiting and returns an
+// error if the control buffer is still throttled after the given duration.
+func (c *controlBuffer) throttleWithTimeout(timeout time.Duration) error {
+	ch := c.trfChan.Load()
+	if ch == nil {
+		return nil
+	}
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	select {
+	case <-(*ch):
+		return nil
+	case <-c.done:
+		return nil
+	case <-timer.C:
+		return fmt.Errorf("still throttled after %v, with at least %d queued frames", timeout, maxQueuedTransportResponseFrames)
+	}
+}
+
 // put adds an item to the controlbuf.
 func (c *controlBuffer) put(it cbItem) error {
 	_, err := c.executeAndPut(nil, it)
@@ -390,9 +455,9 @@ func (c *controlBuffer) executeAndPut(f func() bool, it cbItem) (bool, error) {
 		c.consumerWaiting = false
 	}
 	c.list.enqueue(it)
-	if it.isThrottled() {
+	if it.isTransportResponseFrame(c.side) {
 		c.transportResponseFrames++
-		if c.transportResponseFrames == maxQueuedControlBufferItems {
+		if c.transportResponseFrames == maxQueuedTransportResponseFrames {
 			// We are adding the frame that puts us over the threshold; create
 			// a throttling channel.
 			ch := make(chan struct{})
@@ -447,8 +512,8 @@ func (c *controlBuffer) getOnceLocked() (any, error) {
 		return nil, nil
 	}
 	h := c.list.dequeue().(cbItem)
-	if h.isThrottled() {
-		if c.transportResponseFrames == maxQueuedControlBufferItems {
+	if h.isTransportResponseFrame(c.side) {
+		if c.transportResponseFrames == maxQueuedTransportResponseFrames {
 			// We are removing the frame that put us over the
 			// threshold; close and clear the throttling channel.
 			ch := c.trfChan.Swap(nil)
@@ -484,9 +549,10 @@ func (c *controlBuffer) finish() {
 		}
 	}
 
-	// In case throttle() is currently in flight, it needs to be unblocked.
-	// Otherwise, the transport may not close, since the transport is closed by
-	// the reader encountering the connection error.
+	// In case throttle() or throttleWithTimeout() is currently in flight,
+	// it needs to be unblocked. Otherwise, the transport may not close,
+	// since the transport is closed by the reader encountering the
+	// connection error.
 	ch := c.trfChan.Swap(nil)
 	if ch != nil {
 		close(*ch)

vendor/google.golang.org/grpc/internal/transport/handler_server.go

@@ -268,7 +268,7 @@ func (ht *serverHandlerTransport) writeStatus(s *ServerStream, st *status.Status
 				for _, v := range vv {
 					// http2 ResponseWriter mechanism to send undeclared Trailers after
 					// the headers have possibly been written.
-					h.Add(http2.TrailerPrefix+k, encodeMetadataHeader(k, v))
+					h.Add(http.TrailerPrefix+k, encodeMetadataHeader(k, v))
 				}
 			}
 		}
@@ -494,7 +494,8 @@ func mapRecvMsgError(err error) error {
 	if err == io.EOF || err == io.ErrUnexpectedEOF {
 		return err
 	}
-	if se, ok := err.(http2.StreamError); ok {
+	var se http2.StreamError
+	if errors.As(err, &se) {
 		if code, ok := http2ErrConvTab[se.Code]; ok {
 			return status.Error(code, se.Error())
 		}

vendor/google.golang.org/grpc/internal/transport/http2_client.go

@@ -65,6 +65,13 @@ var clientConnectionCounter uint64
 
 var goAwayLoopyWriterTimeout = 5 * time.Second
 
+// clientThrottleTimeout is the maximum amount of time for which the client
+// stops reading from the connection because too many frames queued in
+// response to the server (see maxQueuedTransportResponseFrames) are waiting
+// to be written. After that, the client closes the connection instead of
+// waiting forever, since the server is most likely not reading.
+var clientThrottleTimeout = 10 * time.Second
+
 var metadataFromOutgoingContextRaw = internal.FromOutgoingContextRaw.(func(context.Context) (metadata.MD, [][]string, bool))
 
 // http2Client implements the ClientTransport interface with HTTP2.
@@ -107,6 +114,12 @@ type http2Client struct {
 	kp               keepalive.ClientParameters
 	keepaliveEnabled bool
 
+	// throttleTimeout is the maximum amount of time for which reads can
+	// stay throttled before the connection is closed. It is copied from
+	// clientThrottleTimeout on creation, so that tests can override the
+	// latter without racing with the readers of other transports.
+	throttleTimeout time.Duration
+
 	statsHandler stats.Handler
 
 	initialWindowSize int32
@@ -359,6 +372,7 @@ func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 		streamQuota:           defaultMaxStreamsClient,
 		streamsQuotaAvailable: make(chan struct{}, 1),
 		keepaliveEnabled:      keepaliveEnabled,
+		throttleTimeout:       clientThrottleTimeout,
 		bufferPool:            opts.BufferPool,
 		onClose:               onClose,
 	}
@@ -386,7 +400,7 @@ func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	} else if md := imetadata.Get(addr); md != nil {
 		t.md = md
 	}
-	t.controlBuf = newControlBuffer(t.ctxDone)
+	t.controlBuf = newControlBuffer(t.ctxDone, clientSide)
 	if opts.InitialWindowSize >= defaultWindowSize {
 		t.initialWindowSize = opts.InitialWindowSize
 	}
@@ -952,6 +966,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr, handler s
 			LocalAddr:   t.localAddr,
 			Compression: callHdr.SendCompress,
 			Header:      header,
+			Authority:   callHdr.Host,
 		})
 	}
 	if transportDrainRequired {
@@ -1726,9 +1741,19 @@ func (t *http2Client) reader(errCh chan<- error) {
 		atomic.StoreInt64(&t.lastRead, time.Now().UnixNano())
 	}
 
-	// loop to keep reading incoming messages on this transport.
+	// Loop to keep reading incoming messages on this transport.
 	for {
-		t.controlBuf.throttle()
+		// Stop reading while too many items that the server can make
+		// the client queue (see cbItem.isTransportResponseFrame) are
+		// waiting to be written, to bound the memory that a server
+		// can make the client use by flooding it without reading. If
+		// reads stay throttled for t.throttleTimeout, close the
+		// connection instead of waiting forever.
+		if err := t.controlBuf.throttleWithTimeout(t.throttleTimeout); err != nil {
+			t.logger.Warningf("Closing connection since the server is not reading: %v", err)
+			errClose = connectionErrorf(true, err, "closing connection since the server is not reading: %v", err)
+			return
+		}
 		frame, err := t.framer.readFrame()
 		if t.keepaliveEnabled {
 			atomic.StoreInt64(&t.lastRead, time.Now().UnixNano())

vendor/google.golang.org/grpc/internal/transport/http2_server.go

@@ -297,7 +297,7 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport,
 	)
 	t.logger = prefixLoggerForServerTransport(t)
 
-	t.controlBuf = newControlBuffer(t.done)
+	t.controlBuf = newControlBuffer(t.done, serverSide)
 	if !config.StaticWindowSize {
 		t.bdpEst = &bdpEstimator{
 			bdp:               initialWindowSize,
@@ -380,6 +380,11 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 	defer t.maxStreamMu.Unlock()
 
 	streamID := frame.Header().StreamID
+	if streamID%2 != 1 || streamID <= t.maxStreamID {
+		// illegal gRPC stream id.
+		return fmt.Errorf("received an illegal stream id: %v. headers frame: %+v", streamID, frame)
+	}
+	t.maxStreamID = streamID
 
 	// frame.Truncated is set to true when framer detects that the current header
 	// list size hits MaxHeaderListSize limit.
@@ -393,12 +398,6 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 		return nil
 	}
 
-	if streamID%2 != 1 || streamID <= t.maxStreamID {
-		// illegal gRPC stream id.
-		return fmt.Errorf("received an illegal stream id: %v. headers frame: %+v", streamID, frame)
-	}
-	t.maxStreamID = streamID
-
 	s := &ServerStream{
 		Stream: Stream{
 			id: streamID,
@@ -1014,7 +1013,17 @@ func (t *http2Server) streamContextErr(s *ServerStream) error {
 		return ErrConnClosing
 	default:
 	}
-	return ContextErr(s.ctx.Err())
+	err := s.ctx.Err()
+	if err == nil {
+		// In closeStream and finishStream, the stream state is transitioned to
+		// streamDone before s.cancel() is invoked so that concurrent operations
+		// immediately observe the terminal state. A concurrent caller (such as
+		// write or writeHeader) can see streamDone and invoke streamContextErr
+		// before s.cancel() runs or propagates to s.ctx. Default to
+		// context.Canceled rather than passing nil to ContextErr.
+		err = context.Canceled
+	}
+	return ContextErr(err)
 }
 
 // WriteHeader sends the header metadata md back to the client.
@@ -1336,10 +1345,6 @@ func (t *http2Server) deleteStream(s *ServerStream, eosReceived bool) {
 
 // finishStream closes the stream and puts the trailing headerFrame into controlbuf.
 func (t *http2Server) finishStream(s *ServerStream, rst bool, rstCode http2.ErrCode, hdr *serverHeaders, eosReceived bool) {
-	// In case stream sending and receiving are invoked in separate
-	// goroutines (e.g., bi-directional streaming), cancel needs to be
-	// called to interrupt the potential blocking on other goroutines.
-	s.cancel()
 
 	oldState := s.swapState(streamDone)
 	if oldState == streamDone {
@@ -1347,6 +1352,11 @@ func (t *http2Server) finishStream(s *ServerStream, rst bool, rstCode http2.ErrC
 		return
 	}
 
+	// In case stream sending and receiving are invoked in separate
+	// goroutines (e.g., bi-directional streaming), cancel needs to be
+	// called to interrupt the potential blocking on other goroutines.
+	s.cancel()
+
 	hdr.cleanup = &cleanupStream{
 		streamID: s.id,
 		rst:      rst,
@@ -1360,15 +1370,16 @@ func (t *http2Server) finishStream(s *ServerStream, rst bool, rstCode http2.ErrC
 
 // closeStream clears the footprint of a stream when the stream is not needed any more.
 func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCode, eosReceived bool) {
+	// We can't return early even if the stream's state is "done" as the state
+	// might have been set by the `finishStream` method. Deleting the stream via
+	// `finishStream` can get blocked on flow control.
+	s.swapState(streamDone)
+
 	// In case stream sending and receiving are invoked in separate
 	// goroutines (e.g., bi-directional streaming), cancel needs to be
 	// called to interrupt the potential blocking on other goroutines.
 	s.cancel()
 
-	// We can't return early even if the stream's state is "done" as the state
-	// might have been set by the `finishStream` method. Deleting the stream via
-	// `finishStream` can get blocked on flow control.
-	s.swapState(streamDone)
 	t.deleteStream(s, eosReceived)
 
 	t.controlBuf.put(&cleanupStream{

vendor/google.golang.org/grpc/internal/transport/transport.go

@@ -121,7 +121,9 @@ func (b *recvBuffer) put(r recvMsg) {
 	if b.err != nil {
 		// drop the buffer on the floor. Since b.err is not nil, any subsequent reads
 		// will always return an error, making this buffer inaccessible.
-		r.buffer.Free()
+		if r.buffer != nil {
+			r.buffer.Free()
+		}
 		// An error had occurred earlier, don't accept more
 		// data or errors.
 		return

vendor/google.golang.org/grpc/metadata/metadata.go

@@ -360,6 +360,66 @@ func FromOutgoingContext(ctx context.Context) (MD, bool) {
 	return out, ok
 }
 
+// ValueFromOutgoingContext returns the metadata value corresponding to the
+// metadata key from the outgoing metadata if it exists. Keys are matched in a
+// case-insensitive manner.
+//
+// Unlike FromOutgoingContext, this does not copy the entire outgoing metadata.
+func ValueFromOutgoingContext(ctx context.Context, key string) []string {
+	raw, ok := ctx.Value(mdOutgoingKey{}).(rawMD)
+	if !ok {
+		return nil
+	}
+
+	key = strings.ToLower(key)
+	// Unlike ValueFromIncomingContext, this can't return as soon as raw.md is
+	// checked: raw.added still needs to be walked and accumulated regardless.
+	// The else below prevents that from overwriting an exact match with an
+	// unrelated case-insensitive one.
+	var matchedMD []string
+	if v, ok := raw.md[key]; ok {
+		matchedMD = v
+	} else {
+		for k, v := range raw.md {
+			// Case insensitive comparison: MD is a map, and there's no
+			// guarantee that the MD attached to the context is created using
+			// our helper functions.
+			if strings.EqualFold(k, key) {
+				matchedMD = v
+				break
+			}
+		}
+	}
+
+	// Defer allocating vals until raw.added actually has a match, so a match
+	// found only in raw.md doesn't pay for both a copyOf and a growing
+	// append.
+	var vals []string
+	for _, added := range raw.added {
+		if len(added)%2 == 1 {
+			panic(fmt.Sprintf("metadata: ValueFromOutgoingContext got an odd number of input pairs for metadata: %d", len(added)))
+		}
+
+		// Case insensitive, like FromOutgoingContext: added isn't guaranteed
+		// lowercase, though AppendToOutgoingContext already lowercases it in
+		// practice, so try the cheap exact match first.
+		for i := 0; i < len(added); i += 2 {
+			if added[i] == key || strings.EqualFold(added[i], key) {
+				if vals == nil {
+					vals = make([]string, 0, len(matchedMD)+1)
+					vals = append(vals, matchedMD...)
+				}
+				vals = append(vals, added[i+1])
+			}
+		}
+	}
+
+	if vals == nil && matchedMD != nil {
+		return copyOf(matchedMD)
+	}
+	return vals
+}
+
 type rawMD struct {
 	md    MD
 	added [][]string

vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection.pb.go

@@ -21,7 +21,7 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.36.11
+// 	protoc-gen-go v1.36.12
 // 	protoc        v5.27.1
 // source: grpc/reflection/v1/reflection.proto
 

vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection.pb.go

@@ -18,7 +18,7 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.36.11
+// 	protoc-gen-go v1.36.12
 // 	protoc        v5.27.1
 // grpc/reflection/v1alpha/reflection.proto is a deprecated file.
 

vendor/google.golang.org/grpc/rpc_util.go

@@ -179,7 +179,6 @@ type callInfo struct {
 	contentSubtype              string
 	codec                       baseCodec
 	maxRetryRPCBufferSize       int
-	onFinish                    []func(err error)
 	authority                   string
 	acceptedResponseCompressors []string
 }
@@ -394,8 +393,7 @@ type OnFinishCallOption struct {
 	OnFinish func(error)
 }
 
-func (o OnFinishCallOption) before(c *callInfo) error {
-	c.onFinish = append(c.onFinish, o.OnFinish)
+func (o OnFinishCallOption) before(*callInfo) error {
 	return nil
 }
 

vendor/google.golang.org/grpc/server.go

@@ -1400,7 +1400,7 @@ func (s *Server) processRPC(ctx context.Context, stream *transport.ServerStream,
 		ss.decompressorV1 = encoding.GetCompressor(rc)
 		if ss.decompressorV1 == nil {
 			st := status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", rc)
-			ss.writeStatus(st)
+			ss.s.WriteStatus(st)
 			return st.Err()
 		}
 	}
@@ -1491,7 +1491,7 @@ func (s *Server) processRPC(ctx context.Context, stream *transport.ServerStream,
 				binlog.Log(ctx, st)
 			}
 		}
-		if err := ss.writeStatus(appStatus); err != nil {
+		if err := ss.s.WriteStatus(appStatus); err != nil {
 			channelz.Warningf(logger, s.channelz, "grpc: Server.processRPC failed to write status: %v", err)
 		}
 		return appErr
@@ -1510,7 +1510,7 @@ func (s *Server) processRPC(ctx context.Context, stream *transport.ServerStream,
 			binlog.Log(ctx, st)
 		}
 	}
-	if err := ss.writeStatus(statusOK); err != nil {
+	if err := ss.s.WriteStatus(statusOK); err != nil {
 		channelz.Warningf(logger, s.channelz, "grpc: Server.processRPC failed to write status: %v", err)
 	}
 	return err
@@ -1719,7 +1719,13 @@ func (s *Server) stop(graceful bool) {
 	}
 
 	if graceful || s.opts.waitForHandlers {
+		// Release the lock while waiting for handlers to finish. Holding it
+		// here would deadlock a concurrent Stop() (the recommended way to abort
+		// a GracefulStop that is taking too long), because Stop() must acquire
+		// s.mu before it can force the server to shut down.
+		s.mu.Unlock()
 		s.handlersWG.Wait()
+		s.mu.Lock()
 	}
 
 	if s.events != nil {

vendor/google.golang.org/grpc/stats/stats.go

@@ -190,6 +190,9 @@ type OutHeader struct {
 	Compression string
 	// Header contains the header metadata sent.
 	Header metadata.MD
+	// Authority is the :authority pseudo-header sent for the RPC.
+	// It is valid only if Client is true.
+	Authority string
 
 	// The following fields are valid only if Client is true.
 	// FullMethod is the full RPC method string, i.e., /package.service/method.

vendor/google.golang.org/grpc/stream.go

@@ -201,9 +201,9 @@ func (w *clientStreamWrapper) SendMsg(m any) error {
 
 }
 
-// RecvMsg receives message m from the stream. For RPCs that call RecvMsg only
-// once i.e. only client streaming RPCs, it calls the underlying RecvMsg a
-// second time after receiving the first message to get the trailers.
+// RecvMsg receives message m from the stream. For non-server-streaming RPCs
+// (unary and client-streaming), it calls the underlying RecvMsg a second time
+// after receiving the first message to get the trailers.
 func (w *clientStreamWrapper) RecvMsg(m any) error {
 	err := w.ClientStream.RecvMsg(m)
 	if err != nil {
@@ -224,19 +224,6 @@ func (w *clientStreamWrapper) RecvMsg(m any) error {
 	return err
 }
 
-// defaultStreamInterceptor is a StreamClientInterceptor which wraps the
-// ClientStream and is always invoked as the first interceptor. It consolidates
-// behavior for different RPC types at the level closest to the application,
-// which simplifies the underlying stream implementation and other interceptors
-// by avoiding duplicate or scattered handling.
-func defaultStreamInterceptor(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error) {
-	cs, err := streamer(ctx, desc, cc, method, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return &clientStreamWrapper{ClientStream: cs, desc: desc}, nil
-}
-
 // NewStream creates a new Stream for the client side. This is typically
 // called by generated code. ctx is used for the lifetime of the stream.
 //
@@ -392,7 +379,11 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		}
 	}
 
-	return newStream(ctx, opts...)
+	cs, err := newStream(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return &clientStreamWrapper{ClientStream: cs, desc: desc}, nil
 }
 
 func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, mc *serviceconfig.MethodConfig, onCommit func(), nameResolutionDelayed bool, opts ...CallOption) (_ ClientStream, err error) {
@@ -632,7 +623,12 @@ func (a *csAttempt) getTransport() error {
 
 func (a *csAttempt) newStream() error {
 	cs := a.cs
-	cs.callHdr.PreviousAttempts = cs.numRetries
+	// The header is copied because the fields set below, notably the authority
+	// override taken from the pick result, describe the endpoint picked for
+	// this attempt only. Mutating the clientStream's header would carry them
+	// into a later attempt.
+	callHdr := *cs.callHdr
+	callHdr.PreviousAttempts = cs.numRetries
 
 	// Merge metadata stored in PickResult, if any, with existing call metadata.
 	// It is safe to overwrite the csAttempt's context here, since all state
@@ -659,11 +655,11 @@ func (a *csAttempt) newStream() error {
 		// apply it, as specified in gRFC A81.
 		if cs.callInfo.authority == "" {
 			if authMD := a.pickResult.Metadata.Get(":authority"); len(authMD) > 0 {
-				cs.callHdr.Authority = authMD[0]
+				callHdr.Authority = authMD[0]
 			}
 		}
 	}
-	s, err := a.transport.NewStream(a.ctx, cs.callHdr, a.statsHandler)
+	s, err := a.transport.NewStream(a.ctx, &callHdr, a.statsHandler)
 	if err != nil {
 		nse, ok := err.(*transport.NewStreamError)
 		if !ok {
@@ -698,10 +694,6 @@ type clientStream struct {
 
 	cancel context.CancelFunc // cancels all attempts
 
-	sentLast bool // sent an end stream
-
-	receivedFirstMsg bool // set after the first message is received
-
 	methodConfig *MethodConfig
 
 	ctx context.Context // the application's context, wrapped by stats/tracing
@@ -709,19 +701,10 @@ type clientStream struct {
 	retryThrottler *retryThrottler // The throttler active when the RPC began.
 
 	binlogs []binarylog.MethodLogger
-	// serverHeaderBinlogged is a boolean for whether server header has been
-	// logged. Server header will be logged when the first time one of those
-	// happens: stream.Header(), stream.Recv().
-	//
-	// It's only read and used by Recv() and Header(), so it doesn't need to be
-	// synchronized.
-	serverHeaderBinlogged bool
 
 	mu                      sync.Mutex
-	firstAttempt            bool // if true, transparent retry is valid
-	numRetries              int  // exclusive of transparent retry attempt(s)
-	numRetriesSincePushback int  // retries since pushback; to reset backoff
-	finished                bool // TODO: replace with atomic cmpxchg or sync.Once?
+	numRetries              int // exclusive of transparent retry attempt(s)
+	numRetriesSincePushback int // retries since pushback; to reset backoff
 	// attempt is the active client stream attempt.
 	// The only place where it is written is the newAttemptLocked method and this method never writes nil.
 	// So, attempt can be nil only inside newClientStream function when clientStream is first created.
@@ -731,13 +714,33 @@ type clientStream struct {
 	// place where we need to check if the attempt is nil.
 	attempt *csAttempt
 	// TODO(hedging): hedging will have multiple attempts simultaneously.
-	committed        bool // active attempt committed for retry?
 	onCommit         func()
 	replayBuffer     []replayOp // operations to replay on retry
 	replayBufferSize int        // current size of replayBuffer
+
+	// Bool fields are grouped at the tail to eliminate the alignment padding
+	// that would otherwise follow each bool when the next field is pointer- or
+	// int-sized. See https://github.com/grpc/grpc-go/issues/9280 for benchmarks.
+	// Add new bool fields here, not inline above.
+
+	// Not guarded by mu.
+	sentLast         bool // sent an end stream
+	receivedFirstMsg bool // set after the first message is received
+	// serverHeaderBinlogged is a boolean for whether server header has been
+	// logged. Server header will be logged when the first time one of those
+	// happens: stream.Header(), stream.Recv().
+	//
+	// It's only read and used by Recv() and Header(), so it doesn't need to be
+	// synchronized.
+	serverHeaderBinlogged bool
 	// nameResolutionDelay indicates if there was a delay in the name resolution.
 	// This field is only valid on client side, it's always false on server side.
 	nameResolutionDelay bool
+
+	// Guarded by mu.
+	firstAttempt bool // if true, transparent retry is valid
+	finished     bool // TODO: replace with atomic cmpxchg or sync.Once?
+	committed    bool // active attempt committed for retry?
 }
 
 type replayOp struct {
@@ -755,10 +758,8 @@ type csAttempt struct {
 	parser          parser
 	pickResult      balancer.PickResult
 
-	finished        bool
-	decompressorV0  Decompressor
-	decompressorV1  encoding.Compressor
-	decompressorSet bool
+	decompressorV0 Decompressor
+	decompressorV1 encoding.Compressor
 
 	mu sync.Mutex // guards trInfo.tr
 	// trInfo may be nil (if EnableTracing is false).
@@ -769,10 +770,18 @@ type csAttempt struct {
 	statsHandler stats.Handler
 	beginTime    time.Time
 
-	// set for newStream errors that may be transparently retried
-	allowTransparentRetry bool
-	// set for pick errors that are returned as a status
-	drop bool
+	// Bool fields are grouped at the tail to eliminate the alignment padding
+	// that would otherwise follow each bool when the next field is pointer- or
+	// int-sized. See https://github.com/grpc/grpc-go/issues/9347 for benchmarks.
+	// Add new bool fields here, not inline above.
+
+	// Not guarded by mu.
+	decompressorSet       bool
+	allowTransparentRetry bool // set for newStream errors that may be transparently retried
+	drop                  bool // set for pick errors that are returned as a status
+
+	// Guarded by mu.
+	finished bool
 }
 
 func (cs *clientStream) commitAttemptLocked() {
@@ -1500,20 +1509,28 @@ type addrConnStream struct {
 	callInfo         *callInfo
 	transport        transport.ClientTransport
 	ctx              context.Context
-	sentLast         bool
-	receivedFirstMsg bool
 	desc             *StreamDesc
 	codec            baseCodec
 	sendCompressorV0 Compressor
 	sendCompressorV1 encoding.Compressor
-	decompressorSet  bool
 	decompressorV0   Decompressor
 	decompressorV1   encoding.Compressor
-	parser           parser
 
 	// mu guards finished and is held for the entire finish method.
-	mu       sync.Mutex
-	finished bool
+	mu     sync.Mutex
+	parser parser
+
+	// Bool fields are grouped at the tail to eliminate the alignment padding
+	// that would otherwise follow each bool when the next field is pointer- or
+	// int-sized. See https://github.com/grpc/grpc-go/issues/9348 for benchmarks.
+	// Add new bool fields here, not inline above.
+
+	// Not guarded by mu.
+	sentLast         bool
+	receivedFirstMsg bool
+	decompressorSet  bool
+
+	finished bool // guarded by mu
 }
 
 func (as *addrConnStream) Header() (metadata.MD, error) {
@@ -1728,15 +1745,23 @@ type serverStream struct {
 
 	sendCompressorName string
 
-	recvFirstMsg bool // set after the first message is received
-
 	maxReceiveMessageSize int
 	maxSendMessageSize    int
-	trInfo                *traceInfo
+
+	// mu guards trInfo.tr after the service handler runs.
+	mu     sync.Mutex
+	trInfo *traceInfo
 
 	statsHandler stats.Handler
 
 	binlogs []binarylog.MethodLogger
+
+	// Bool fields are grouped at the tail to eliminate the alignment padding
+	// that would otherwise follow each bool when the next field is pointer- or
+	// int-sized. See https://github.com/grpc/grpc-go/issues/9349 for benchmarks.
+	// Add new bool fields here, not inline above.
+
+	recvFirstMsg bool // set after the first message is received
 	// serverHeaderBinlogged indicates whether server header has been logged. It
 	// will happen when one of the following two happens: stream.SendHeader(),
 	// stream.Send().
@@ -1744,10 +1769,6 @@ type serverStream struct {
 	// It's only checked in send and sendHeader, doesn't need to be
 	// synchronized.
 	serverHeaderBinlogged bool
-
-	statusWritten atomic.Bool // True if status has been written to the transport.
-
-	mu sync.Mutex // protects trInfo.tr after the service handler runs.
 }
 
 func (ss *serverStream) Context() context.Context {
@@ -1795,16 +1816,6 @@ func (ss *serverStream) SetTrailer(md metadata.MD) {
 	ss.s.SetTrailer(md)
 }
 
-// writeStatus sends the status of a stream to the client. It uses an atomic
-// CAS to guarantee that the status is written to the transport exactly once,
-// even if called concurrently.
-func (ss *serverStream) writeStatus(st *status.Status) error {
-	if !ss.statusWritten.CompareAndSwap(false, true) {
-		return nil
-	}
-	return ss.s.WriteStatus(st)
-}
-
 func (ss *serverStream) SendMsg(m any) (err error) {
 	defer func() {
 		if ss.trInfo != nil {
@@ -1821,7 +1832,7 @@ func (ss *serverStream) SendMsg(m any) (err error) {
 		}
 		if err != nil && err != io.EOF {
 			st, _ := status.FromError(toRPCErr(err))
-			ss.writeStatus(st)
+			ss.s.WriteStatus(st)
 			// Non-user specified status was sent out. This should be an error
 			// case (as a server side Cancel maybe).
 			//
@@ -1904,7 +1915,7 @@ func (ss *serverStream) RecvMsg(m any) (err error) {
 		}
 		if err != nil && err != io.EOF {
 			st, _ := status.FromError(toRPCErr(err))
-			ss.writeStatus(st)
+			ss.s.WriteStatus(st)
 			// Non-user specified status was sent out. This should be an error
 			// case (as a server side Cancel maybe).
 			//

vendor/google.golang.org/grpc/version.go

@@ -19,4 +19,4 @@
 package grpc
 
 // Version is the current grpc version.
-const Version = "1.84.0"
+const Version = "1.85.0"

vendor/modules.txt

@@ -2556,8 +2556,8 @@ google.golang.org/genproto/googleapis/api/httpbody
 ## explicit; go 1.26.0
 google.golang.org/genproto/googleapis/rpc/errdetails
 google.golang.org/genproto/googleapis/rpc/status
-# google.golang.org/grpc v1.84.0
-## explicit; go 1.25.0
+# google.golang.org/grpc v1.85.0
+## explicit; go 1.26.0
 google.golang.org/grpc
 google.golang.org/grpc/attributes
 google.golang.org/grpc/backoff