Repository: grpc/grpc-go

Issue:
Flaky test package: test/xds
Alongside #6913 and #6912, I have ran the test/xds suite on master since I added tests to it for my xDS Server fix #6889. I have encountered numerous flakes on g3, particularly those outlined in custom lb tests for distribution #6601. However, I have encountered almost every client and server side xDS test flake with a context timeout for a RPC expected to proceed. Each has different logs/events preceeding it's timeout, but every test seems susceptible to timeout. The flakes are generally rare, but due to the number of tests in the test suite you can successfully trigger by running the full test suite enough times. My initial inkling tells me there's some synchronization needed or something gets stuck in the management server/testing xDS Client flow. This also manifests in rare flakes for my xDS Server fix, where I expect something like an err that represents Accept and Close, and I get a context timeout instead.

Discussion:
@arvindbr8: another one for  TestServerSideXDS_WithValidAndInvalidSecurityConfiguration: https://github.com/grpc/grpc-go/actions/runs/7480796959/job/20361025267?pr=6916@zasweq: https://github.com/grpc/grpc-go/actions/runs/7618526481/job/20749870802?pr=6933@arvindbr8: https://github.com/grpc/grpc-go/actions/runs/7716359885/job/21032969470?pr=6949@arvindbr8: https://github.com/grpc/grpc-go/actions/runs/7791646589/job/21248143511?pr=6965@zasweq: https://github.com/grpc/grpc-go/actions/runs/8546803299/job/23417838293?pr=7085@arjan-bal: https://github.com/grpc/grpc-go/actions/runs/9790522733/job/27032314481?pr=7390@arjan-bal: @zasweq I investigated this and the problem seems to be due to the xDS management server getting stuck while writing to this buffered channel
https://github.com/grpc/grpc-go/blob/d27ddb5eb5940c949f88bc2cb21eed9254f8be75/test/xds/xds_server_certificate_providers_test.go#L249

In the logs of failing runs for `TestServerSideXDS_WithValidAndInvalidSecurityConfiguration`, I noticed that the resource snapshot update request is sent to the xds management server before the xds client is able to connect to the xds server. This somehow results in more than 1 Listener requests being sent to the xds server which get stuck waiting to write to the buffered channel.

This seems to be a problem with the test and not the implementation. Adding a 50 millis sleep after starting both the servers did get rid of the flakiness in `TestServerSideXDS_WithValidAndInvalidSecurityConfiguration`.@zasweq: Ah nice thank you for figuring this out!@arvindbr8: https://github.com/grpc/grpc-go/actions/runs/9996172593/job/27629932247?pr=7397@zasweq: You mentioned this solved the test, but not the flakes in the full package. This was my flaky test in this PR so thanks for fixing this: https://github.com/grpc/grpc-go/actions/runs/10050840269/job/27779434995?pr=7434 :).@arjan-bal: Closing this issue as the root cause of all flaking tests seems to be different. There are individual issues open for tests that have flaked recently.

Solution patch:
diff --git a/clientconn.go b/clientconn.go
index 423be7b4..c3cafbc1 100644
--- a/clientconn.go
+++ b/clientconn.go
@@ -918,9 +918,8 @@ func (ac *addrConn) connect() error {
 		ac.mu.Unlock()
 		return nil
 	}
-	ac.mu.Unlock()
 
-	ac.resetTransport()
+	ac.resetTransportAndUnlock()
 	return nil
 }
 
@@ -992,11 +991,9 @@ func (ac *addrConn) updateAddrs(addrs []resolver.Address) {
 		ac.updateConnectivityState(connectivity.Idle, nil)
 	}
 
-	ac.mu.Unlock()
-
 	// Since we were connecting/connected, we should start a new connection
 	// attempt.
-	go ac.resetTransport()
+	go ac.resetTransportAndUnlock()
 }
 
 // getServerName determines the serverName to be used in the connection
@@ -1231,8 +1228,10 @@ func (ac *addrConn) adjustParams(r transport.GoAwayReason) {
 	}
 }
 
-func (ac *addrConn) resetTransport() {
-	ac.mu.Lock()
+// resetTransportAndUnlock unconditionally connects the addrConn.
+//
+// ac.mu must be held by the caller, and this function will guarantee it is released.
+func (ac *addrConn) resetTransportAndUnlock() {
 	acCtx := ac.ctx
 	if acCtx.Err() != nil {
 		ac.mu.Unlock()



Test patch:
diff --git a/test/channelz_test.go b/test/channelz_test.go
index 7d8ea136..5bfa0f57 100644
--- a/test/channelz_test.go
+++ b/test/channelz_test.go
@@ -1086,7 +1086,7 @@ func (s) TestCZClientAndServerSocketMetricsStreamsCountFlowControlRSTStream(t *t
 	go func() {
 		payload := make([]byte, 16384)
 		for i := 0; i < 6; i++ {
-			dw.getRawConnWrapper().writeRawFrame(http2.FrameData, 0, tc.getCurrentStreamID(), payload)
+			dw.getRawConnWrapper().writeDataFrame(tc.getCurrentStreamID(), payload)
 		}
 	}()
 	if _, err := stream.Recv(); status.Code(err) != codes.ResourceExhausted {
diff --git a/test/rawConnWrapper.go b/test/rawConnWrapper.go
index 8814810a..4928056e 100644
--- a/test/rawConnWrapper.go
+++ b/test/rawConnWrapper.go
@@ -239,8 +239,8 @@ func (rcw *rawConnWrapper) writeGoAway(maxStreamID uint32, code http2.ErrCode, d
 	return nil
 }
 
-func (rcw *rawConnWrapper) writeRawFrame(t http2.FrameType, flags http2.Flags, streamID uint32, payload []byte) error {
-	if err := rcw.fr.WriteRawFrame(t, flags, streamID, payload); err != nil {
+func (rcw *rawConnWrapper) writeDataFrame(streamID uint32, payload []byte) error {
+	if err := rcw.fr.WriteData(streamID, false, payload); err != nil {
 		return fmt.Errorf("error writing Raw Frame: %v", err)
 	}
 	return nil