GODRIVER-2577 WIP

Author: qingyang-hu

x/mongo/driver/topology/server.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -554,7 +555,9 @@ func (s *Server) update() {
 			defer s.processErrorLock.Unlock()
 
 			s.updateDescription(desc)
-			if lastError := desc.LastError; lastError != nil {
+			lastError := desc.LastError
+			log.Println("lastError", lastError)
+			if lastError != nil {
 				// Retry after the first timeout in case of a FAAS pause as described in GODRIVER-2577.
 				if err := unwrapConnectionError(lastError); err != nil {
 					err, ok := err.(net.Error)

x/mongo/driver/topology/server_test.go

@@ -11,8 +11,14 @@ package topology
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
+	"fmt"
+	"io/ioutil"
+	"log"
 	"net"
+	"os"
 	"runtime"
 	"sync"
 	"sync/atomic"
@@ -54,14 +60,14 @@ type errorQueue struct {
 	mutex  sync.Mutex
 }
 
-func (eq *errorQueue) head() error {
-	eq.mutex.Lock()
-	defer eq.mutex.Unlock()
-	if len(eq.errors) > 0 {
-		return eq.errors[0]
-	}
-	return nil
-}
+// func (eq *errorQueue) head() error {
+// 	eq.mutex.Lock()
+// 	defer eq.mutex.Unlock()
+// 	if len(eq.errors) > 0 {
+// 		return eq.errors[0]
+// 	}
+// 	return nil
+// }
 
 func (eq *errorQueue) dequeue() bool {
 	eq.mutex.Lock()
@@ -73,78 +79,95 @@ func (eq *errorQueue) dequeue() bool {
 	return false
 }
 
+/*
 type timeoutConn struct {
 	net.Conn
 	errors *errorQueue
 }
 
 func (c *timeoutConn) Read(b []byte) (int, error) {
-	n, err := 0, c.errors.head()
-	if err == nil {
-		n, err = c.Conn.Read(b)
-	}
+	//n, err := 0, c.errors.head()
+	//if err == nil {
+	n, err := c.Conn.Read(b)
+	//}
+	log.Println("read", n, err)
 	return n, err
 }
 
 func (c *timeoutConn) Write(b []byte) (int, error) {
-	n, err := 0, c.errors.head()
-	if err == nil {
-		n, err = c.Conn.Write(b)
-	}
+	//n, err := 0, c.errors.head()
+	//if err == nil {
+	n, err := c.Conn.Write(b)
+	//}
+	log.Println("write", n, err)
 	return n, err
 }
+*/
 
 type timeoutDialer struct {
 	Dialer
 	errors *errorQueue
 }
 
 func (d *timeoutDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
-	var dialer net.Dialer
-	c, e := dialer.DialContext(ctx, network, address)
-	return &timeoutConn{c, d.errors}, e
-}
+	c, e := d.Dialer.DialContext(ctx, network, address)
+
+	caFile := os.Getenv("MONGO_GO_DRIVER_CA_FILE")
+	log.Println("dial", caFile)
+	if len(caFile) > 0 {
+		pem, err := ioutil.ReadFile(caFile)
+		if err != nil {
+			return nil, err
+		}
 
-type timeoutErr struct {
-	net.UnknownNetworkError
-}
+		ca := x509.NewCertPool()
+		if !ca.AppendCertsFromPEM(pem) {
+			return nil, fmt.Errorf("unable to load CA file")
+		}
 
-func (e *timeoutErr) Timeout() bool {
-	return true
+		config := &tls.Config{
+			InsecureSkipVerify: true,
+			RootCAs:            ca,
+		}
+		c = tls.Client(c, config)
+	}
+	return c, e
 }
 
-var timeout = &timeoutErr{"test timeout"}
-
 // TestServerHeartbeatTimeout tests timeout retry for GODRIVER-2577.
 func TestServerHeartbeatTimeout(t *testing.T) {
+	networkTimeoutError := &net.DNSError{
+		IsTimeout: true,
+	}
+
 	testCases := []struct {
 		desc              string
 		ioErrors          []error
 		expectPoolCleared bool
 	}{
 		{
 			desc:              "one single timeout should not clear the pool",
-			ioErrors:          []error{nil, timeout, nil, timeout, nil},
+			ioErrors:          []error{nil, networkTimeoutError, nil, networkTimeoutError, nil},
 			expectPoolCleared: false,
 		},
-		{
-			desc:              "continuous timeouts should clear the pool",
-			ioErrors:          []error{nil, timeout, timeout, nil},
-			expectPoolCleared: true,
-		},
+		// {
+		// 	desc:              "continuous timeouts should clear the pool",
+		// 	ioErrors:          []error{nil, networkTimeoutError, networkTimeoutError, nil},
+		// 	expectPoolCleared: true,
+		// },
 	}
 	for _, tc := range testCases {
 		tc := tc
 		t.Run(tc.desc, func(t *testing.T) {
-			t.Parallel()
+			//t.Parallel()
 
 			var wg sync.WaitGroup
 			wg.Add(1)
 
 			errors := &errorQueue{errors: tc.ioErrors}
 			tpm := monitor.NewTestPoolMonitor()
 			server := NewServer(
-				address.Address("localhost:27017"),
+				address.Address("localhost"),
 				primitive.NewObjectID(),
 				WithConnectionPoolMonitor(func(*event.PoolMonitor) *event.PoolMonitor {
 					return tpm.PoolMonitor
@@ -165,6 +188,9 @@ func TestServerHeartbeatTimeout(t *testing.T) {
 						},
 					}
 				}),
+				WithHeartbeatInterval(func(time.Duration) time.Duration {
+					return 3 * time.Second
+				}),
 			)
 			require.NoError(t, server.Connect(nil))
 			wg.Wait()