rebase: Bump google.golang.org/grpc from 1.53.0 to 1.54.0

Bumps [google.golang.org/grpc](https://github.com/grpc/grpc-go) from 1.53.0 to 1.54.0.
- [Release notes](https://github.com/grpc/grpc-go/releases)
- [Commits](https://github.com/grpc/grpc-go/compare/v1.53.0...v1.54.0)

---
updated-dependencies:
- dependency-name: google.golang.org/grpc
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

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

@@ -22,6 +22,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"net"
 	"runtime"
 	"strconv"
 	"sync"
@@ -486,12 +487,13 @@ type loopyWriter struct {
 	hEnc          *hpack.Encoder // HPACK encoder.
 	bdpEst        *bdpEstimator
 	draining      bool
+	conn          net.Conn
 
 	// Side-specific handlers
 	ssGoAwayHandler func(*goAway) (bool, error)
 }
 
-func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimator) *loopyWriter {
+func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimator, conn net.Conn) *loopyWriter {
 	var buf bytes.Buffer
 	l := &loopyWriter{
 		side:          s,
@@ -504,6 +506,7 @@ func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimato
 		hBuf:          &buf,
 		hEnc:          hpack.NewEncoder(&buf),
 		bdpEst:        bdpEst,
+		conn:          conn,
 	}
 	return l
 }
@@ -521,15 +524,27 @@ const minBatchSize = 1000
 // 2. Stream level flow control quota available.
 //
 // In each iteration of run loop, other than processing the incoming control
-// frame, loopy calls processData, which processes one node from the activeStreams linked-list.
-// This results in writing of HTTP2 frames into an underlying write buffer.
-// When there's no more control frames to read from controlBuf, loopy flushes the write buffer.
-// As an optimization, to increase the batch size for each flush, loopy yields the processor, once
-// if the batch size is too low to give stream goroutines a chance to fill it up.
+// frame, loopy calls processData, which processes one node from the
+// activeStreams linked-list.  This results in writing of HTTP2 frames into an
+// underlying write buffer.  When there's no more control frames to read from
+// controlBuf, loopy flushes the write buffer.  As an optimization, to increase
+// the batch size for each flush, loopy yields the processor, once if the batch
+// size is too low to give stream goroutines a chance to fill it up.
+//
+// Upon exiting, if the error causing the exit is not an I/O error, run()
+// flushes and closes the underlying connection.  Otherwise, the connection is
+// left open to allow the I/O error to be encountered by the reader instead.
 func (l *loopyWriter) run() (err error) {
-	// Always flush the writer before exiting in case there are pending frames
-	// to be sent.
-	defer l.framer.writer.Flush()
+	defer func() {
+		if logger.V(logLevel) {
+			logger.Infof("transport: loopyWriter exiting with error: %v", err)
+		}
+		if !isIOError(err) {
+			l.framer.writer.Flush()
+			l.conn.Close()
+		}
+		l.cbuf.finish()
+	}()
 	for {
 		it, err := l.cbuf.get(true)
 		if err != nil {
@@ -581,11 +596,11 @@ func (l *loopyWriter) outgoingWindowUpdateHandler(w *outgoingWindowUpdate) error
 	return l.framer.fr.WriteWindowUpdate(w.streamID, w.increment)
 }
 
-func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) error {
+func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) {
 	// Otherwise update the quota.
 	if w.streamID == 0 {
 		l.sendQuota += w.increment
-		return nil
+		return
 	}
 	// Find the stream and update it.
 	if str, ok := l.estdStreams[w.streamID]; ok {
@@ -593,10 +608,9 @@ func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) error
 		if strQuota := int(l.oiws) - str.bytesOutStanding; strQuota > 0 && str.state == waitingOnStreamQuota {
 			str.state = active
 			l.activeStreams.enqueue(str)
-			return nil
+			return
 		}
 	}
-	return nil
 }
 
 func (l *loopyWriter) outgoingSettingsHandler(s *outgoingSettings) error {
@@ -604,13 +618,11 @@ func (l *loopyWriter) outgoingSettingsHandler(s *outgoingSettings) error {
 }
 
 func (l *loopyWriter) incomingSettingsHandler(s *incomingSettings) error {
-	if err := l.applySettings(s.ss); err != nil {
-		return err
-	}
+	l.applySettings(s.ss)
 	return l.framer.fr.WriteSettingsAck()
 }
 
-func (l *loopyWriter) registerStreamHandler(h *registerStream) error {
+func (l *loopyWriter) registerStreamHandler(h *registerStream) {
 	str := &outStream{
 		id:    h.streamID,
 		state: empty,
@@ -618,7 +630,6 @@ func (l *loopyWriter) registerStreamHandler(h *registerStream) error {
 		wq:    h.wq,
 	}
 	l.estdStreams[h.streamID] = str
-	return nil
 }
 
 func (l *loopyWriter) headerHandler(h *headerFrame) error {
@@ -720,10 +731,10 @@ func (l *loopyWriter) writeHeader(streamID uint32, endStream bool, hf []hpack.He
 	return nil
 }
 
-func (l *loopyWriter) preprocessData(df *dataFrame) error {
+func (l *loopyWriter) preprocessData(df *dataFrame) {
 	str, ok := l.estdStreams[df.streamID]
 	if !ok {
-		return nil
+		return
 	}
 	// If we got data for a stream it means that
 	// stream was originated and the headers were sent out.
@@ -732,7 +743,6 @@ func (l *loopyWriter) preprocessData(df *dataFrame) error {
 		str.state = active
 		l.activeStreams.enqueue(str)
 	}
-	return nil
 }
 
 func (l *loopyWriter) pingHandler(p *ping) error {
@@ -743,9 +753,8 @@ func (l *loopyWriter) pingHandler(p *ping) error {
 
 }
 
-func (l *loopyWriter) outFlowControlSizeRequestHandler(o *outFlowControlSizeRequest) error {
+func (l *loopyWriter) outFlowControlSizeRequestHandler(o *outFlowControlSizeRequest) {
 	o.resp <- l.sendQuota
-	return nil
 }
 
 func (l *loopyWriter) cleanupStreamHandler(c *cleanupStream) error {
@@ -763,6 +772,7 @@ func (l *loopyWriter) cleanupStreamHandler(c *cleanupStream) error {
 		}
 	}
 	if l.draining && len(l.estdStreams) == 0 {
+		// Flush and close the connection; we are done with it.
 		return errors.New("finished processing active streams while in draining mode")
 	}
 	return nil
@@ -798,6 +808,7 @@ func (l *loopyWriter) incomingGoAwayHandler(*incomingGoAway) error {
 	if l.side == clientSide {
 		l.draining = true
 		if len(l.estdStreams) == 0 {
+			// Flush and close the connection; we are done with it.
 			return errors.New("received GOAWAY with no active streams")
 		}
 	}
@@ -816,17 +827,10 @@ func (l *loopyWriter) goAwayHandler(g *goAway) error {
 	return nil
 }
 
-func (l *loopyWriter) closeConnectionHandler() error {
-	// Exit loopyWriter entirely by returning an error here.  This will lead to
-	// the transport closing the connection, and, ultimately, transport
-	// closure.
-	return ErrConnClosing
-}
-
 func (l *loopyWriter) handle(i interface{}) error {
 	switch i := i.(type) {
 	case *incomingWindowUpdate:
-		return l.incomingWindowUpdateHandler(i)
+		l.incomingWindowUpdateHandler(i)
 	case *outgoingWindowUpdate:
 		return l.outgoingWindowUpdateHandler(i)
 	case *incomingSettings:
@@ -836,7 +840,7 @@ func (l *loopyWriter) handle(i interface{}) error {
 	case *headerFrame:
 		return l.headerHandler(i)
 	case *registerStream:
-		return l.registerStreamHandler(i)
+		l.registerStreamHandler(i)
 	case *cleanupStream:
 		return l.cleanupStreamHandler(i)
 	case *earlyAbortStream:
@@ -844,21 +848,24 @@ func (l *loopyWriter) handle(i interface{}) error {
 	case *incomingGoAway:
 		return l.incomingGoAwayHandler(i)
 	case *dataFrame:
-		return l.preprocessData(i)
+		l.preprocessData(i)
 	case *ping:
 		return l.pingHandler(i)
 	case *goAway:
 		return l.goAwayHandler(i)
 	case *outFlowControlSizeRequest:
-		return l.outFlowControlSizeRequestHandler(i)
+		l.outFlowControlSizeRequestHandler(i)
 	case closeConnection:
-		return l.closeConnectionHandler()
+		// Just return a non-I/O error and run() will flush and close the
+		// connection.
+		return ErrConnClosing
 	default:
 		return fmt.Errorf("transport: unknown control message type %T", i)
 	}
+	return nil
 }
 
-func (l *loopyWriter) applySettings(ss []http2.Setting) error {
+func (l *loopyWriter) applySettings(ss []http2.Setting) {
 	for _, s := range ss {
 		switch s.ID {
 		case http2.SettingInitialWindowSize:
@@ -877,7 +884,6 @@ func (l *loopyWriter) applySettings(ss []http2.Setting) error {
 			updateHeaderTblSize(l.hEnc, s.Val)
 		}
 	}
-	return nil
 }
 
 // processData removes the first stream from active streams, writes out at most 16KB
@@ -911,7 +917,7 @@ func (l *loopyWriter) processData() (bool, error) {
 				return false, err
 			}
 			if err := l.cleanupStreamHandler(trailer.cleanup); err != nil {
-				return false, nil
+				return false, err
 			}
 		} else {
 			l.activeStreams.enqueue(str)

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

@@ -444,15 +444,8 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 		return nil, err
 	}
 	go func() {
-		t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst)
-		err := t.loopy.run()
-		if logger.V(logLevel) {
-			logger.Infof("transport: loopyWriter exited. Closing connection. Err: %v", err)
-		}
-		// Do not close the transport.  Let reader goroutine handle it since
-		// there might be data in the buffers.
-		t.conn.Close()
-		t.controlBuf.finish()
+		t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst, t.conn)
+		t.loopy.run()
 		close(t.writerDone)
 	}()
 	return t, nil
@@ -1264,10 +1257,12 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
 		t.mu.Unlock()
 		return
 	}
-	if f.ErrCode == http2.ErrCodeEnhanceYourCalm {
-		if logger.V(logLevel) {
-			logger.Infof("Client received GoAway with http2.ErrCodeEnhanceYourCalm.")
-		}
+	if f.ErrCode == http2.ErrCodeEnhanceYourCalm && string(f.DebugData()) == "too_many_pings" {
+		// When a client receives a GOAWAY with error code ENHANCE_YOUR_CALM and debug
+		// data equal to ASCII "too_many_pings", it should log the occurrence at a log level that is
+		// enabled by default and double the configure KEEPALIVE_TIME used for new connections
+		// on that channel.
+		logger.Errorf("Client received GoAway with error code ENHANCE_YOUR_CALM and debug data equal to ASCII \"too_many_pings\".")
 	}
 	id := f.LastStreamID
 	if id > 0 && id%2 == 0 {

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

@@ -331,14 +331,9 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport,
 	t.handleSettings(sf)
 
 	go func() {
-		t.loopy = newLoopyWriter(serverSide, t.framer, t.controlBuf, t.bdpEst)
+		t.loopy = newLoopyWriter(serverSide, t.framer, t.controlBuf, t.bdpEst, t.conn)
 		t.loopy.ssGoAwayHandler = t.outgoingGoAwayHandler
-		err := t.loopy.run()
-		if logger.V(logLevel) {
-			logger.Infof("transport: loopyWriter exited. Closing connection. Err: %v", err)
-		}
-		t.conn.Close()
-		t.controlBuf.finish()
+		t.loopy.run()
 		close(t.writerDone)
 	}()
 	go t.keepalive()
@@ -383,7 +378,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		// if false, content-type was missing or invalid
 		isGRPC      = false
 		contentType = ""
-		mdata       = make(map[string][]string)
+		mdata       = make(metadata.MD, len(frame.Fields))
 		httpMethod  string
 		// these are set if an error is encountered while parsing the headers
 		protocolError bool
@@ -404,6 +399,17 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 			mdata[hf.Name] = append(mdata[hf.Name], hf.Value)
 			s.contentSubtype = contentSubtype
 			isGRPC = true
+
+		case "grpc-accept-encoding":
+			mdata[hf.Name] = append(mdata[hf.Name], hf.Value)
+			if hf.Value == "" {
+				continue
+			}
+			compressors := hf.Value
+			if s.clientAdvertisedCompressors != "" {
+				compressors = s.clientAdvertisedCompressors + "," + compressors
+			}
+			s.clientAdvertisedCompressors = compressors
 		case "grpc-encoding":
 			s.recvCompress = hf.Value
 		case ":method":
@@ -595,7 +601,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 			LocalAddr:   t.localAddr,
 			Compression: s.recvCompress,
 			WireLength:  int(frame.Header().Length),
-			Header:      metadata.MD(mdata).Copy(),
+			Header:      mdata.Copy(),
 		}
 		sh.HandleRPC(s.ctx, inHeader)
 	}
@@ -1344,9 +1350,6 @@ func (t *http2Server) outgoingGoAwayHandler(g *goAway) (bool, error) {
 			return false, err
 		}
 		if retErr != nil {
-			// Abruptly close the connection following the GoAway (via
-			// loopywriter).  But flush out what's inside the buffer first.
-			t.framer.writer.Flush()
 			return false, retErr
 		}
 		return true, nil

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

@@ -21,6 +21,7 @@ package transport
 import (
 	"bufio"
 	"encoding/base64"
+	"errors"
 	"fmt"
 	"io"
 	"math"
@@ -330,7 +331,8 @@ func (w *bufWriter) Write(b []byte) (n int, err error) {
 		return 0, w.err
 	}
 	if w.batchSize == 0 { // Buffer has been disabled.
-		return w.conn.Write(b)
+		n, err = w.conn.Write(b)
+		return n, toIOError(err)
 	}
 	for len(b) > 0 {
 		nn := copy(w.buf[w.offset:], b)
@@ -352,10 +354,30 @@ func (w *bufWriter) Flush() error {
 		return nil
 	}
 	_, w.err = w.conn.Write(w.buf[:w.offset])
+	w.err = toIOError(w.err)
 	w.offset = 0
 	return w.err
 }
 
+type ioError struct {
+	error
+}
+
+func (i ioError) Unwrap() error {
+	return i.error
+}
+
+func isIOError(err error) bool {
+	return errors.As(err, &ioError{})
+}
+
+func toIOError(err error) error {
+	if err == nil {
+		return nil
+	}
+	return ioError{error: err}
+}
+
 type framer struct {
 	writer *bufWriter
 	fr     *http2.Framer

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

@@ -257,6 +257,9 @@ type Stream struct {
 	fc           *inFlow
 	wq           *writeQuota
 
+	// Holds compressor names passed in grpc-accept-encoding metadata from the
+	// client. This is empty for the client side stream.
+	clientAdvertisedCompressors string
 	// Callback to state application's intentions to read data. This
 	// is used to adjust flow control, if needed.
 	requestRead func(int)
@@ -345,8 +348,24 @@ func (s *Stream) RecvCompress() string {
 }
 
 // SetSendCompress sets the compression algorithm to the stream.
-func (s *Stream) SetSendCompress(str string) {
-	s.sendCompress = str
+func (s *Stream) SetSendCompress(name string) error {
+	if s.isHeaderSent() || s.getState() == streamDone {
+		return errors.New("transport: set send compressor called after headers sent or stream done")
+	}
+
+	s.sendCompress = name
+	return nil
+}
+
+// SendCompress returns the send compressor name.
+func (s *Stream) SendCompress() string {
+	return s.sendCompress
+}
+
+// ClientAdvertisedCompressors returns the compressor names advertised by the
+// client via grpc-accept-encoding header.
+func (s *Stream) ClientAdvertisedCompressors() string {
+	return s.clientAdvertisedCompressors
 }
 
 // Done returns a channel which is closed when it receives the final status