rebase: bump google.golang.org/grpc from 1.42.0 to 1.43.0

Bumps [google.golang.org/grpc](https://github.com/grpc/grpc-go) from 1.42.0 to 1.43.0.
- [Release notes](https://github.com/grpc/grpc-go/releases)
- [Commits](https://github.com/grpc/grpc-go/compare/v1.42.0...v1.43.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/channelz/funcs.go

@@ -204,9 +204,9 @@ func RegisterChannel(c Channel, pid int64, ref string) int64 {
 		trace:       &channelTrace{createdTime: time.Now(), events: make([]*TraceEvent, 0, getMaxTraceEntry())},
 	}
 	if pid == 0 {
-		db.get().addChannel(id, cn, true, pid, ref)
+		db.get().addChannel(id, cn, true, pid)
 	} else {
-		db.get().addChannel(id, cn, false, pid, ref)
+		db.get().addChannel(id, cn, false, pid)
 	}
 	return id
 }
@@ -228,7 +228,7 @@ func RegisterSubChannel(c Channel, pid int64, ref string) int64 {
 		pid:     pid,
 		trace:   &channelTrace{createdTime: time.Now(), events: make([]*TraceEvent, 0, getMaxTraceEntry())},
 	}
-	db.get().addSubChannel(id, sc, pid, ref)
+	db.get().addSubChannel(id, sc, pid)
 	return id
 }
 
@@ -258,7 +258,7 @@ func RegisterListenSocket(s Socket, pid int64, ref string) int64 {
 	}
 	id := idGen.genID()
 	ls := &listenSocket{refName: ref, s: s, id: id, pid: pid}
-	db.get().addListenSocket(id, ls, pid, ref)
+	db.get().addListenSocket(id, ls, pid)
 	return id
 }
 
@@ -273,11 +273,11 @@ func RegisterNormalSocket(s Socket, pid int64, ref string) int64 {
 	}
 	id := idGen.genID()
 	ns := &normalSocket{refName: ref, s: s, id: id, pid: pid}
-	db.get().addNormalSocket(id, ns, pid, ref)
+	db.get().addNormalSocket(id, ns, pid)
 	return id
 }
 
-// RemoveEntry removes an entry with unique channelz trakcing id to be id from
+// RemoveEntry removes an entry with unique channelz tracking id to be id from
 // channelz database.
 func RemoveEntry(id int64) {
 	db.get().removeEntry(id)
@@ -333,7 +333,7 @@ func (c *channelMap) addServer(id int64, s *server) {
 	c.mu.Unlock()
 }
 
-func (c *channelMap) addChannel(id int64, cn *channel, isTopChannel bool, pid int64, ref string) {
+func (c *channelMap) addChannel(id int64, cn *channel, isTopChannel bool, pid int64) {
 	c.mu.Lock()
 	cn.cm = c
 	cn.trace.cm = c
@@ -346,7 +346,7 @@ func (c *channelMap) addChannel(id int64, cn *channel, isTopChannel bool, pid in
 	c.mu.Unlock()
 }
 
-func (c *channelMap) addSubChannel(id int64, sc *subChannel, pid int64, ref string) {
+func (c *channelMap) addSubChannel(id int64, sc *subChannel, pid int64) {
 	c.mu.Lock()
 	sc.cm = c
 	sc.trace.cm = c
@@ -355,7 +355,7 @@ func (c *channelMap) addSubChannel(id int64, sc *subChannel, pid int64, ref stri
 	c.mu.Unlock()
 }
 
-func (c *channelMap) addListenSocket(id int64, ls *listenSocket, pid int64, ref string) {
+func (c *channelMap) addListenSocket(id int64, ls *listenSocket, pid int64) {
 	c.mu.Lock()
 	ls.cm = c
 	c.listenSockets[id] = ls
@@ -363,7 +363,7 @@ func (c *channelMap) addListenSocket(id int64, ls *listenSocket, pid int64, ref
 	c.mu.Unlock()
 }
 
-func (c *channelMap) addNormalSocket(id int64, ns *normalSocket, pid int64, ref string) {
+func (c *channelMap) addNormalSocket(id int64, ns *normalSocket, pid int64) {
 	c.mu.Lock()
 	ns.cm = c
 	c.normalSockets[id] = ns

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

@@ -22,20 +22,14 @@ package envconfig
 import (
 	"os"
 	"strings"
-
-	xdsenv "google.golang.org/grpc/internal/xds/env"
 )
 
 const (
 	prefix          = "GRPC_GO_"
-	retryStr        = prefix + "RETRY"
 	txtErrIgnoreStr = prefix + "IGNORE_TXT_ERRORS"
 )
 
 var (
-	// Retry is enabled unless explicitly disabled via "GRPC_GO_RETRY=off" or
-	// if XDS retry support is explicitly disabled.
-	Retry = !strings.EqualFold(os.Getenv(retryStr), "off") && xdsenv.RetrySupport
 	// TXTErrIgnore is set if TXT errors should be ignored ("GRPC_GO_IGNORE_TXT_ERRORS" is not "false").
 	TXTErrIgnore = !strings.EqualFold(os.Getenv(txtErrIgnoreStr), "false")
 )

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

@@ -16,9 +16,7 @@
  *
  */
 
-// Package env acts a single source of definition for all environment variables
-// related to the xDS implementation in gRPC.
-package env
+package envconfig
 
 import (
 	"os"
@@ -26,72 +24,67 @@ import (
 )
 
 const (
-	// BootstrapFileNameEnv is the env variable to set bootstrap file name.
+	// XDSBootstrapFileNameEnv is the env variable to set bootstrap file name.
 	// Do not use this and read from env directly. Its value is read and kept in
 	// variable BootstrapFileName.
 	//
 	// When both bootstrap FileName and FileContent are set, FileName is used.
-	BootstrapFileNameEnv = "GRPC_XDS_BOOTSTRAP"
-	// BootstrapFileContentEnv is the env variable to set bootstrapp file
+	XDSBootstrapFileNameEnv = "GRPC_XDS_BOOTSTRAP"
+	// XDSBootstrapFileContentEnv is the env variable to set bootstrapp file
 	// content. Do not use this and read from env directly. Its value is read
 	// and kept in variable BootstrapFileName.
 	//
 	// When both bootstrap FileName and FileContent are set, FileName is used.
-	BootstrapFileContentEnv = "GRPC_XDS_BOOTSTRAP_CONFIG"
+	XDSBootstrapFileContentEnv = "GRPC_XDS_BOOTSTRAP_CONFIG"
 
 	ringHashSupportEnv           = "GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH"
 	clientSideSecuritySupportEnv = "GRPC_XDS_EXPERIMENTAL_SECURITY_SUPPORT"
 	aggregateAndDNSSupportEnv    = "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER"
-	retrySupportEnv              = "GRPC_XDS_EXPERIMENTAL_ENABLE_RETRY"
 	rbacSupportEnv               = "GRPC_XDS_EXPERIMENTAL_RBAC"
+	federationEnv                = "GRPC_EXPERIMENTAL_XDS_FEDERATION"
 
-	c2pResolverSupportEnv                    = "GRPC_EXPERIMENTAL_GOOGLE_C2P_RESOLVER"
 	c2pResolverTestOnlyTrafficDirectorURIEnv = "GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI"
 )
 
 var (
-	// BootstrapFileName holds the name of the file which contains xDS bootstrap
-	// configuration. Users can specify the location of the bootstrap file by
-	// setting the environment variable "GRPC_XDS_BOOTSTRAP".
+	// XDSBootstrapFileName holds the name of the file which contains xDS
+	// bootstrap configuration. Users can specify the location of the bootstrap
+	// file by setting the environment variable "GRPC_XDS_BOOTSTRAP".
 	//
 	// When both bootstrap FileName and FileContent are set, FileName is used.
-	BootstrapFileName = os.Getenv(BootstrapFileNameEnv)
-	// BootstrapFileContent holds the content of the xDS bootstrap
-	// configuration. Users can specify the bootstrap config by
-	// setting the environment variable "GRPC_XDS_BOOTSTRAP_CONFIG".
+	XDSBootstrapFileName = os.Getenv(XDSBootstrapFileNameEnv)
+	// XDSBootstrapFileContent holds the content of the xDS bootstrap
+	// configuration. Users can specify the bootstrap config by setting the
+	// environment variable "GRPC_XDS_BOOTSTRAP_CONFIG".
 	//
 	// When both bootstrap FileName and FileContent are set, FileName is used.
-	BootstrapFileContent = os.Getenv(BootstrapFileContentEnv)
-	// RingHashSupport indicates whether ring hash support is enabled, which can
-	// be disabled by setting the environment variable
+	XDSBootstrapFileContent = os.Getenv(XDSBootstrapFileContentEnv)
+	// XDSRingHash indicates whether ring hash support is enabled, which can be
+	// disabled by setting the environment variable
 	// "GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH" to "false".
-	RingHashSupport = !strings.EqualFold(os.Getenv(ringHashSupportEnv), "false")
-	// ClientSideSecuritySupport is used to control processing of security
+	XDSRingHash = !strings.EqualFold(os.Getenv(ringHashSupportEnv), "false")
+	// XDSClientSideSecurity is used to control processing of security
 	// configuration on the client-side.
 	//
 	// Note that there is no env var protection for the server-side because we
 	// have a brand new API on the server-side and users explicitly need to use
 	// the new API to get security integration on the server.
-	ClientSideSecuritySupport = !strings.EqualFold(os.Getenv(clientSideSecuritySupportEnv), "false")
-	// AggregateAndDNSSupportEnv indicates whether processing of aggregated
-	// cluster and DNS cluster is enabled, which can be enabled by setting the
+	XDSClientSideSecurity = !strings.EqualFold(os.Getenv(clientSideSecuritySupportEnv), "false")
+	// XDSAggregateAndDNS indicates whether processing of aggregated cluster
+	// and DNS cluster is enabled, which can be enabled by setting the
 	// environment variable
 	// "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER" to
 	// "true".
-	AggregateAndDNSSupportEnv = strings.EqualFold(os.Getenv(aggregateAndDNSSupportEnv), "true")
+	XDSAggregateAndDNS = strings.EqualFold(os.Getenv(aggregateAndDNSSupportEnv), "true")
 
-	// RetrySupport indicates whether xDS retry is enabled.
-	RetrySupport = !strings.EqualFold(os.Getenv(retrySupportEnv), "false")
-
-	// RBACSupport indicates whether xDS configured RBAC HTTP Filter is enabled,
+	// XDSRBAC indicates whether xDS configured RBAC HTTP Filter is enabled,
 	// which can be disabled by setting the environment variable
 	// "GRPC_XDS_EXPERIMENTAL_RBAC" to "false".
-	RBACSupport = !strings.EqualFold(os.Getenv(rbacSupportEnv), "false")
+	XDSRBAC = !strings.EqualFold(os.Getenv(rbacSupportEnv), "false")
+
+	// XDSFederation indicates whether federation support is enabled.
+	XDSFederation = strings.EqualFold(os.Getenv(federationEnv), "true")
 
-	// C2PResolverSupport indicates whether support for C2P resolver is enabled.
-	// This can be enabled by setting the environment variable
-	// "GRPC_EXPERIMENTAL_GOOGLE_C2P_RESOLVER" to "true".
-	C2PResolverSupport = strings.EqualFold(os.Getenv(c2pResolverSupportEnv), "true")
 	// C2PResolverTestOnlyTrafficDirectorURI is the TD URI for testing.
 	C2PResolverTestOnlyTrafficDirectorURI = os.Getenv(c2pResolverTestOnlyTrafficDirectorURIEnv)
 )

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

@@ -136,12 +136,10 @@ type inFlow struct {
 
 // newLimit updates the inflow window to a new value n.
 // It assumes that n is always greater than the old limit.
-func (f *inFlow) newLimit(n uint32) uint32 {
+func (f *inFlow) newLimit(n uint32) {
 	f.mu.Lock()
-	d := n - f.limit
 	f.limit = n
 	f.mu.Unlock()
-	return d
 }
 
 func (f *inFlow) maybeAdjust(n uint32) uint32 {

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

@@ -201,6 +201,12 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 		}
 	}()
 
+	// gRPC, resolver, balancer etc. can specify arbitrary data in the
+	// Attributes field of resolver.Address, which is shoved into connectCtx
+	// and passed to the dialer and credential handshaker. This makes it possible for
+	// address specific arbitrary data to reach custom dialers and credential handshakers.
+	connectCtx = icredentials.NewClientHandshakeInfoContext(connectCtx, credentials.ClientHandshakeInfo{Attributes: addr.Attributes})
+
 	conn, err := dial(connectCtx, opts.Dialer, addr, opts.UseProxy, opts.UserAgent)
 	if err != nil {
 		if opts.FailOnNonTempDialError {
@@ -245,11 +251,6 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 		}
 	}
 	if transportCreds != nil {
-		// gRPC, resolver, balancer etc. can specify arbitrary data in the
-		// Attributes field of resolver.Address, which is shoved into connectCtx
-		// and passed to the credential handshaker. This makes it possible for
-		// address specific arbitrary data to reach the credential handshaker.
-		connectCtx = icredentials.NewClientHandshakeInfoContext(connectCtx, credentials.ClientHandshakeInfo{Attributes: addr.Attributes})
 		rawConn := conn
 		// Pull the deadline from the connectCtx, which will be used for
 		// timeouts in the authentication protocol handshake. Can ignore the
@@ -587,7 +588,7 @@ func (t *http2Client) getTrAuthData(ctx context.Context, audience string) (map[s
 				return nil, err
 			}
 
-			return nil, status.Errorf(codes.Unauthenticated, "transport: %v", err)
+			return nil, status.Errorf(codes.Unauthenticated, "transport: per-RPC creds failed due to error: %v", err)
 		}
 		for k, v := range data {
 			// Capital header names are illegal in HTTP/2.
@@ -1556,7 +1557,7 @@ func minTime(a, b time.Duration) time.Duration {
 	return b
 }
 
-// keepalive running in a separate goroutune makes sure the connection is alive by sending pings.
+// keepalive running in a separate goroutine makes sure the connection is alive by sending pings.
 func (t *http2Client) keepalive() {
 	p := &ping{data: [8]byte{}}
 	// True iff a ping has been sent, and no data has been received since then.

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

@@ -73,7 +73,6 @@ type http2Server struct {
 	writerDone  chan struct{} // sync point to enable testing.
 	remoteAddr  net.Addr
 	localAddr   net.Addr
-	maxStreamID uint32               // max stream ID ever seen
 	authInfo    credentials.AuthInfo // auth info about the connection
 	inTapHandle tap.ServerInHandle
 	framer      *framer
@@ -123,6 +122,11 @@ type http2Server struct {
 	bufferPool *bufferPool
 
 	connectionID uint64
+
+	// maxStreamMu guards the maximum stream ID
+	// This lock may not be taken if mu is already held.
+	maxStreamMu sync.Mutex
+	maxStreamID uint32 // max stream ID ever seen
 }
 
 // NewServerTransport creates a http2 transport with conn and configuration
@@ -334,6 +338,10 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport,
 
 // operateHeader takes action on the decoded headers.
 func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (fatal bool) {
+	// Acquire max stream ID lock for entire duration
+	t.maxStreamMu.Lock()
+	defer t.maxStreamMu.Unlock()
+
 	streamID := frame.Header().StreamID
 
 	// frame.Truncated is set to true when framer detects that the current header
@@ -348,6 +356,15 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		return false
 	}
 
+	if streamID%2 != 1 || streamID <= t.maxStreamID {
+		// illegal gRPC stream id.
+		if logger.V(logLevel) {
+			logger.Errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", streamID)
+		}
+		return true
+	}
+	t.maxStreamID = streamID
+
 	buf := newRecvBuffer()
 	s := &Stream{
 		id:  streamID,
@@ -355,7 +372,6 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		buf: buf,
 		fc:  &inFlow{limit: uint32(t.initialWindowSize)},
 	}
-
 	var (
 		// If a gRPC Response-Headers has already been received, then it means
 		// that the peer is speaking gRPC and we are in gRPC mode.
@@ -498,16 +514,6 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		s.cancel()
 		return false
 	}
-	if streamID%2 != 1 || streamID <= t.maxStreamID {
-		t.mu.Unlock()
-		// illegal gRPC stream id.
-		if logger.V(logLevel) {
-			logger.Errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", streamID)
-		}
-		s.cancel()
-		return true
-	}
-	t.maxStreamID = streamID
 	if httpMethod != http.MethodPost {
 		t.mu.Unlock()
 		if logger.V(logLevel) {
@@ -1293,20 +1299,23 @@ var goAwayPing = &ping{data: [8]byte{1, 6, 1, 8, 0, 3, 3, 9}}
 // Handles outgoing GoAway and returns true if loopy needs to put itself
 // in draining mode.
 func (t *http2Server) outgoingGoAwayHandler(g *goAway) (bool, error) {
+	t.maxStreamMu.Lock()
 	t.mu.Lock()
 	if t.state == closing { // TODO(mmukhi): This seems unnecessary.
 		t.mu.Unlock()
+		t.maxStreamMu.Unlock()
 		// The transport is closing.
 		return false, ErrConnClosing
 	}
-	sid := t.maxStreamID
 	if !g.headsUp {
 		// Stop accepting more streams now.
 		t.state = draining
+		sid := t.maxStreamID
 		if len(t.activeStreams) == 0 {
 			g.closeConn = true
 		}
 		t.mu.Unlock()
+		t.maxStreamMu.Unlock()
 		if err := t.framer.fr.WriteGoAway(sid, g.code, g.debugData); err != nil {
 			return false, err
 		}
@@ -1319,6 +1328,7 @@ func (t *http2Server) outgoingGoAwayHandler(g *goAway) (bool, error) {
 		return true, nil
 	}
 	t.mu.Unlock()
+	t.maxStreamMu.Unlock()
 	// For a graceful close, send out a GoAway with stream ID of MaxUInt32,
 	// Follow that with a ping and wait for the ack to come back or a timer
 	// to expire. During this time accept new streams since they might have

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

@@ -37,7 +37,7 @@ var (
 	httpProxyFromEnvironment = http.ProxyFromEnvironment
 )
 
-func mapAddress(ctx context.Context, address string) (*url.URL, error) {
+func mapAddress(address string) (*url.URL, error) {
 	req := &http.Request{
 		URL: &url.URL{
 			Scheme: "https",
@@ -114,7 +114,7 @@ func doHTTPConnectHandshake(ctx context.Context, conn net.Conn, backendAddr stri
 // connection.
 func proxyDial(ctx context.Context, addr string, grpcUA string) (conn net.Conn, err error) {
 	newAddr := addr
-	proxyURL, err := mapAddress(ctx, addr)
+	proxyURL, err := mapAddress(addr)
 	if err != nil {
 		return nil, err
 	}