GODRIVER-1581 Transform network errors into context.DeadlineExceeded (#430)

If a context is translated to a network read/write deadline and the
read/write call returns a timeout error, this commit will swallow
that error and replace it with context.DeadlineExceeded so the
resulting error will return true for
errors.Is(err, context.DeadlineExceeded).

Author: Divjot Arora

mongo/errors.go

@@ -47,7 +47,10 @@ func replaceErrors(err error) error {
 	}
 	if qe, ok := err.(driver.QueryFailureError); ok {
 		// qe.Message is "command failure"
-		ce := CommandError{Name: qe.Message}
+		ce := CommandError{
+			Name:    qe.Message,
+			Wrapped: qe.Wrapped,
+		}
 
 		dollarErr, err := qe.Response.LookupErr("$err")
 		if err == nil {

mongo/integration/errors_test.go

@@ -12,8 +12,11 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net"
 	"testing"
+	"time"
 
+	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/internal/testutil/assert"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/integration/mtest"
@@ -57,4 +60,49 @@ func TestErrors(t *testing.T) {
 			assert.True(mt, errors.Is(err, io.EOF), "expected error %v, got %v", io.EOF, err)
 		})
 	})
+
+	mt.RunOpts("network timeouts", noClientOpts, func(mt *mtest.T) {
+		mt.Run("context timeouts return DeadlineExceeded", func(mt *mtest.T) {
+			_, err := mt.Coll.InsertOne(mtest.Background, bson.D{{"x", 1}})
+			assert.Nil(mt, err, "InsertOne error: %v", err)
+
+			mt.ClearEvents()
+			filter := bson.M{
+				"$where": "function() { sleep(1000); return false; }",
+			}
+			timeoutCtx, cancel := context.WithTimeout(mtest.Background, 100*time.Millisecond)
+			defer cancel()
+			_, err = mt.Coll.Find(timeoutCtx, filter)
+
+			evt := mt.GetStartedEvent()
+			assert.Equal(mt, "find", evt.CommandName, "expected command 'find', got %q", evt.CommandName)
+			assert.True(mt, errors.Is(err, context.DeadlineExceeded),
+				"errors.Is failure: expected error %v to be %v", err, context.DeadlineExceeded)
+		})
+
+		socketTimeoutOpts := options.Client().
+			SetSocketTimeout(100 * time.Millisecond)
+		socketTimeoutMtOpts := mtest.NewOptions().
+			ClientOptions(socketTimeoutOpts)
+		mt.RunOpts("socketTimeoutMS timeouts return network errors", socketTimeoutMtOpts, func(mt *mtest.T) {
+			_, err := mt.Coll.InsertOne(mtest.Background, bson.D{{"x", 1}})
+			assert.Nil(mt, err, "InsertOne error: %v", err)
+
+			mt.ClearEvents()
+			filter := bson.M{
+				"$where": "function() { sleep(1000); return false; }",
+			}
+			_, err = mt.Coll.Find(mtest.Background, filter)
+
+			evt := mt.GetStartedEvent()
+			assert.Equal(mt, "find", evt.CommandName, "expected command 'find', got %q", evt.CommandName)
+
+			assert.False(mt, errors.Is(err, context.DeadlineExceeded),
+				"errors.Is failure: expected error %v to not be %v", err, context.DeadlineExceeded)
+			var netErr net.Error
+			ok := errors.As(err, &netErr)
+			assert.True(mt, ok, "errors.As failure: expected error %v to be a net.Error", err)
+			assert.True(mt, netErr.Timeout(), "expected error %v to be a network timeout", err)
+		})
+	})
 }

x/mongo/driver/errors.go

@@ -44,13 +44,19 @@ var (
 type QueryFailureError struct {
 	Message  string
 	Response bsoncore.Document
+	Wrapped  error
 }
 
 // Error implements the error interface.
 func (e QueryFailureError) Error() string {
 	return fmt.Sprintf("%s: %v", e.Message, e.Response)
 }
 
+// Unwrap returns the underlying error.
+func (e QueryFailureError) Unwrap() error {
+	return e.Wrapped
+}
+
 // ResponseError is an error parsing the response to a command.
 type ResponseError struct {
 	Message string

x/mongo/driver/operation_legacy.go

@@ -648,12 +648,12 @@ func (op Operation) roundTripLegacyCursor(ctx context.Context, wm []byte, srvr S
 func (op Operation) roundTripLegacy(ctx context.Context, conn Connection, wm []byte) ([]byte, error) {
 	err := conn.WriteWireMessage(ctx, wm)
 	if err != nil {
-		return nil, Error{Message: err.Error(), Labels: []string{TransientTransactionError, NetworkError}}
+		return nil, Error{Message: err.Error(), Labels: []string{TransientTransactionError, NetworkError}, Wrapped: err}
 	}
 
 	wm, err = conn.ReadWireMessage(ctx, wm[:0])
 	if err != nil {
-		err = Error{Message: err.Error(), Labels: []string{TransientTransactionError, NetworkError}}
+		err = Error{Message: err.Error(), Labels: []string{TransientTransactionError, NetworkError}, Wrapped: err}
 	}
 	return wm, err
 }

x/mongo/driver/topology/connection.go

@@ -198,6 +198,20 @@ func (c *connection) closeConnectContext() {
 	c.cancelConnectContext()
 }
 
+func transformNetworkError(originalError error, contextDeadlineUsed bool) error {
+	if originalError == nil {
+		return nil
+	}
+	if !contextDeadlineUsed {
+		return originalError
+	}
+
+	if netErr, ok := originalError.(net.Error); ok && netErr.Timeout() {
+		return context.DeadlineExceeded
+	}
+	return originalError
+}
+
 func (c *connection) writeWireMessage(ctx context.Context, wm []byte) error {
 	var err error
 	if atomic.LoadInt32(&c.connected) != connected {
@@ -214,7 +228,9 @@ func (c *connection) writeWireMessage(ctx context.Context, wm []byte) error {
 		deadline = time.Now().Add(c.writeTimeout)
 	}
 
+	var contextDeadlineUsed bool
 	if dl, ok := ctx.Deadline(); ok && (deadline.IsZero() || dl.Before(deadline)) {
+		contextDeadlineUsed = true
 		deadline = dl
 	}
 
@@ -225,7 +241,11 @@ func (c *connection) writeWireMessage(ctx context.Context, wm []byte) error {
 	_, err = c.nc.Write(wm)
 	if err != nil {
 		c.close()
-		return ConnectionError{ConnectionID: c.id, Wrapped: err, message: "unable to write wire message to network"}
+		return ConnectionError{
+			ConnectionID: c.id,
+			Wrapped:      transformNetworkError(err, contextDeadlineUsed),
+			message:      "unable to write wire message to network",
+		}
 	}
 
 	c.bumpIdleDeadline()
@@ -251,7 +271,9 @@ func (c *connection) readWireMessage(ctx context.Context, dst []byte) ([]byte, e
 		deadline = time.Now().Add(c.readTimeout)
 	}
 
+	var contextDeadlineUsed bool
 	if dl, ok := ctx.Deadline(); ok && (deadline.IsZero() || dl.Before(deadline)) {
+		contextDeadlineUsed = true
 		deadline = dl
 	}
 
@@ -270,7 +292,11 @@ func (c *connection) readWireMessage(ctx context.Context, dst []byte) ([]byte, e
 	if err != nil {
 		// We closeConnection the connection because we don't know if there are other bytes left to read.
 		c.close()
-		return nil, ConnectionError{ConnectionID: c.id, Wrapped: err, message: "incomplete read of message header"}
+		return nil, ConnectionError{
+			ConnectionID: c.id,
+			Wrapped:      transformNetworkError(err, contextDeadlineUsed),
+			message:      "incomplete read of message header",
+		}
 	}
 
 	// read the length as an int32
@@ -289,7 +315,11 @@ func (c *connection) readWireMessage(ctx context.Context, dst []byte) ([]byte, e
 	if err != nil {
 		// We closeConnection the connection because we don't know if there are other bytes left to read.
 		c.close()
-		return nil, ConnectionError{ConnectionID: c.id, Wrapped: err, message: "incomplete read of full message"}
+		return nil, ConnectionError{
+			ConnectionID: c.id,
+			Wrapped:      transformNetworkError(err, contextDeadlineUsed),
+			message:      "incomplete read of full message",
+		}
 	}
 
 	c.bumpIdleDeadline()