Add actionable error for retryable writes on unsupported storage engine

GODRIVER-1219

Change-Id: I530785bce7116c30eec96f4c728130d5f9da330c

Author: eppingere

mongo/client_internal_test.go

@@ -13,6 +13,7 @@ import (
 	"os"
 	"path"
 	"reflect"
+	"strconv"
 	"testing"
 	"time"
 
@@ -27,7 +28,13 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/writeconcern"
 	"go.mongodb.org/mongo-driver/tag"
 	"go.mongodb.org/mongo-driver/x/bsonx"
+	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
+	"go.mongodb.org/mongo-driver/x/mongo/driver"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/address"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/connstring"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/description"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/drivertest"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/operation"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/session"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/uuid"
 )
@@ -283,6 +290,161 @@ func TestClient_ReplaceTopologyError(t *testing.T) {
 
 }
 
+type retryableSSD struct {
+	C driver.Connection
+}
+
+var _ driver.Deployment = retryableSSD{}
+var _ driver.Server = retryableSSD{}
+
+func (rssd retryableSSD) SelectServer(context.Context, description.ServerSelector) (driver.Server, error) {
+	return rssd, nil
+}
+
+func (rssd retryableSSD) Kind() description.TopologyKind {
+	return description.Single
+}
+
+func (rssd retryableSSD) Connection(context.Context) (driver.Connection, error) {
+	return rssd.C, nil
+}
+
+func (rssd retryableSSD) SupportsRetryWrites() bool {
+	return true
+}
+
+func TestRetryWritesError20Wrapped(t *testing.T) {
+	serverVersion, err := getServerVersion(createTestDatabase(t, nil))
+	require.NoError(t, err)
+
+	if compareVersions(t, serverVersion, "3.6") < 0 {
+		t.Skip()
+	}
+
+	idx, writeError := bsoncore.AppendDocumentStart(nil)
+	writeError = bsoncore.AppendInt32Element(writeError, "ok", 1)
+	elemIdx, elem := bsoncore.AppendDocumentStart(nil)
+	elem = bsoncore.AppendInt32Element(elem, "index", 0)
+	elem = bsoncore.AppendStringElement(elem, "errmsg", "Transaction numbers")
+	elem = bsoncore.AppendInt32Element(elem, "code", 20)
+	elem, _ = bsoncore.AppendDocumentEnd(elem, elemIdx)
+	writeErrorsIdx, writeErrors := bsoncore.AppendArrayStart(nil)
+	writeErrors = bsoncore.AppendDocumentElement(writeErrors, strconv.Itoa(0), elem)
+	writeErrors, _ = bsoncore.AppendArrayEnd(writeErrors, writeErrorsIdx)
+	writeError = bsoncore.AppendArrayElement(writeError, "writeErrors", writeErrors)
+	writeError, _ = bsoncore.AppendDocumentEnd(writeError, idx)
+
+	idx, writeErrorNot20 := bsoncore.AppendDocumentStart(nil)
+	writeErrorNot20 = bsoncore.AppendInt32Element(writeErrorNot20, "ok", 1)
+	elemIdx, elem = bsoncore.AppendDocumentStart(nil)
+	elem = bsoncore.AppendInt32Element(elem, "index", 0)
+	elem = bsoncore.AppendStringElement(elem, "errmsg", "Transaction numbers")
+	elem = bsoncore.AppendInt32Element(elem, "code", 19)
+	elem, _ = bsoncore.AppendDocumentEnd(elem, elemIdx)
+	writeErrorsIdx, writeErrors = bsoncore.AppendArrayStart(nil)
+	writeErrors = bsoncore.AppendDocumentElement(writeErrors, strconv.Itoa(0), elem)
+	writeErrors, _ = bsoncore.AppendArrayEnd(writeErrors, writeErrorsIdx)
+	writeErrorNot20 = bsoncore.AppendArrayElement(writeErrorNot20, "writeErrors", writeErrors)
+	writeErrorNot20, _ = bsoncore.AppendDocumentEnd(writeErrorNot20, idx)
+
+	idx, writeErrorOnly20 := bsoncore.AppendDocumentStart(nil)
+	writeErrorOnly20 = bsoncore.AppendInt32Element(writeErrorOnly20, "ok", 1)
+	elemIdx, elem = bsoncore.AppendDocumentStart(nil)
+	elem = bsoncore.AppendInt32Element(elem, "index", 0)
+	elem = bsoncore.AppendStringElement(elem, "errmsg", "something other than transaction numbers")
+	elem = bsoncore.AppendInt32Element(elem, "code", 20)
+	elem, _ = bsoncore.AppendDocumentEnd(elem, elemIdx)
+	writeErrorsIdx, writeErrors = bsoncore.AppendArrayStart(nil)
+	writeErrors = bsoncore.AppendDocumentElement(writeErrors, strconv.Itoa(0), elem)
+	writeErrors, _ = bsoncore.AppendArrayEnd(writeErrors, writeErrorsIdx)
+	writeErrorOnly20 = bsoncore.AppendArrayElement(writeErrorOnly20, "writeErrors", writeErrors)
+	writeErrorOnly20, _ = bsoncore.AppendDocumentEnd(writeErrorOnly20, idx)
+
+	idx, notOk := bsoncore.AppendDocumentStart(nil)
+	notOk = bsoncore.AppendInt64Element(notOk, "ok", 0)
+	notOk = bsoncore.AppendStringElement(notOk, "errmsg", "Transaction numbers")
+	notOk = bsoncore.AppendInt32Element(notOk, "code", 20)
+	notOk, _ = bsoncore.AppendDocumentEnd(notOk, idx)
+
+	idx, not20notOK := bsoncore.AppendDocumentStart(nil)
+	not20notOK = bsoncore.AppendInt64Element(not20notOK, "ok", 0)
+	not20notOK = bsoncore.AppendStringElement(not20notOK, "errmsg", "Transaction numbers")
+	not20notOK = bsoncore.AppendInt32Element(not20notOK, "code", 19)
+	not20notOK, _ = bsoncore.AppendDocumentEnd(not20notOK, idx)
+
+	idx, only20NotOK := bsoncore.AppendDocumentStart(nil)
+	only20NotOK = bsoncore.AppendInt64Element(only20NotOK, "ok", 0)
+	only20NotOK = bsoncore.AppendStringElement(only20NotOK, "errmsg", "something other than transaction numbers")
+	only20NotOK = bsoncore.AppendInt32Element(only20NotOK, "code", 20)
+	only20NotOK, _ = bsoncore.AppendDocumentEnd(only20NotOK, idx)
+
+	tests := []struct {
+		name                 string
+		wireMessage          []byte // bsoncore byte slice
+		shouldError          bool
+		expectedErrorMessage string
+	}{
+		{"writeError", writeError, true, driver.ErrUnsupportedStorageEngine.Error()},
+		{"writeError with only err code 20 and wrong err message", writeErrorOnly20, true, "write command error: [{write errors: [{something other than transaction numbers}]}, {<nil>}]"},
+		{"writeError with only err code 19 and right err message", writeErrorNot20, true, "write command error: [{write errors: [{Transaction numbers}]}, {<nil>}]"},
+		{"NotOkError", notOk, true, driver.ErrUnsupportedStorageEngine.Error()},
+		{"NotOkError with err code 20 and wrong err message", only20NotOK, true, "something other than transaction numbers"},
+		{"NotOkError with err code 19 and right err message", not20notOK, true, "Transaction numbers"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			conn := &drivertest.ChannelConn{
+				Written: make(chan []byte, 1),
+				Desc: description.Server{
+					CanonicalAddr:         address.Address("localhost:27017"),
+					MaxDocumentSize:       16777216,
+					MaxMessageSize:        48000000,
+					MaxBatchCount:         100000,
+					SessionTimeoutMinutes: 30,
+					Kind:                  description.RSPrimary,
+					WireVersion: &description.VersionRange{
+						Max: 8,
+					},
+				},
+				ReadResp: make(chan []byte, 1),
+			}
+
+			conn.ReadResp <- drivertest.MakeReply(test.wireMessage)
+
+			deployment := retryableSSD{C: conn}
+
+			client := createTestClient(t)
+			coll := client.Database("test").Collection("test")
+
+			sess, err := client.StartSession()
+			defer sess.EndSession(context.Background())
+			noerr(t, err)
+
+			idx, writeError = bsoncore.AppendDocumentStart(nil)
+			writeError = bsoncore.AppendStringElement(writeError, "_id", "1")
+			writeError, _ = bsoncore.AppendDocumentEnd(writeError, idx)
+
+			op := operation.NewInsert(writeError).CommandMonitor(coll.client.monitor).ClusterClock(coll.client.clock).
+				Database(coll.db.name).Collection(coll.name).
+				Deployment(coll.client.topology).Deployment(deployment).Retry(driver.RetryOnce).Session(sess.(*sessionImpl).clientSession)
+
+			err = op.Execute(context.Background())
+			if test.shouldError {
+				if err == nil || err.Error() != test.expectedErrorMessage {
+					t.Fatalf("unexpected error occured, wanted: %v got: %v", test.expectedErrorMessage, err)
+				}
+				return
+			}
+
+			if err != nil {
+				t.Fatalf("did not expect an error, instead recieved: %v", err)
+			}
+
+		})
+	}
+}
+
 func TestClient_ListDatabases_noFilter(t *testing.T) {
 	t.Parallel()
 

mongo/retryable_writes_test.go

@@ -12,13 +12,10 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
-	"testing"
-
 	"strings"
-
-	"time"
-
 	"sync"
+	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	"go.mongodb.org/mongo-driver/bson"
@@ -29,6 +26,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/readpref"
 	"go.mongodb.org/mongo-driver/mongo/writeconcern"
 	"go.mongodb.org/mongo-driver/x/bsonx"
+	"go.mongodb.org/mongo-driver/x/mongo/driver"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/connstring"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/session"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/topology"
@@ -147,6 +145,38 @@ func TestTxnNumberIncluded(t *testing.T) {
 	}
 }
 
+func TestRetryableWritesErrorOnMMAPV1(t *testing.T) {
+	name := "test"
+	version, err := getServerVersion(createTestDatabase(t, &name))
+	require.NoError(t, err)
+
+	if shouldSkipRetryTest(t, version) {
+		t.Skip("only run on 3.6.x and not on standalone")
+	}
+
+	client := createTestClient(t)
+	require.NoError(t, err)
+	db := client.Database("test")
+	defer func() { _ = db.Drop(context.Background()) }()
+	coll := client.Database("test").Collection("test")
+	defer func() { _ = coll.Drop(context.Background()) }()
+
+	res := db.RunCommand(context.Background(), bson.D{
+		{"serverStatus", 1},
+	})
+	noerr(t, res.Err())
+
+	storageEngine, ok := res.rdr.Lookup("storageEngine", "name").StringValueOK()
+	if !ok || storageEngine != "mmapv1" {
+		t.Skip("only run on mmapv1")
+	}
+
+	_, err = coll.InsertOne(context.Background(), bson.D{
+		{"_id", 1},
+	})
+	require.Equal(t, driver.ErrUnsupportedStorageEngine, err)
+}
+
 // test case for all RetryableWritesSpec tests
 func TestRetryableWritesSpec(t *testing.T) {
 	for _, file := range testhelpers.FindJSONFilesInDir(t, retryWritesDir) {

x/mongo/driver/errors.go

@@ -28,6 +28,9 @@ var (
 	// ErrUnacknowledgedWrite is returned from functions that have an unacknowledged
 	// write concern.
 	ErrUnacknowledgedWrite = errors.New("unacknowledged write")
+	// ErrUnsupportedStorageEngine is returned when a retryable write is attempted against a server
+	// that uses a storage engine that does not support retryable writes
+	ErrUnsupportedStorageEngine = errors.New("this MongoDB deployment does not support retryable writes. Please add retryWrites=false to your connection string")
 )
 
 // QueryFailureError is an error representing a command failure as a document.
@@ -66,6 +69,17 @@ type WriteCommandError struct {
 	WriteErrors       WriteErrors
 }
 
+// UnsupportedStorageEngine returns whether or not the WriteCommandError comes from a retryable write being attempted
+// against a server that has a storage engine where they are not supported
+func (wce WriteCommandError) UnsupportedStorageEngine() bool {
+	for _, writeError := range wce.WriteErrors {
+		if writeError.Code == 20 && strings.HasPrefix(strings.ToLower(writeError.Message), "transaction numbers") {
+			return true
+		}
+	}
+	return false
+}
+
 func (wce WriteCommandError) Error() string {
 	var buf bytes.Buffer
 	fmt.Fprint(&buf, "write command error: [")
@@ -167,6 +181,11 @@ type Error struct {
 	Name    string
 }
 
+// UnsupportedStorageEngine returns whether e came as a result of an unsupported storage engine
+func (e Error) UnsupportedStorageEngine() bool {
+	return e.Code == 20 && strings.HasPrefix(strings.ToLower(e.Message), "transaction numbers")
+}
+
 // Error implements the error interface.
 func (e Error) Error() string {
 	if e.Name != "" {

x/mongo/driver/operation.go

@@ -361,6 +361,9 @@ func (op Operation) Execute(ctx context.Context, scratch []byte) error {
 		}
 		switch tt := err.(type) {
 		case WriteCommandError:
+			if e := err.(WriteCommandError); retryable && op.Type == Write && e.UnsupportedStorageEngine() {
+				return ErrUnsupportedStorageEngine
+			}
 			if retryable && tt.Retryable() && retries != 0 {
 				retries--
 				original, err = err, nil
@@ -407,6 +410,9 @@ func (op Operation) Execute(ctx context.Context, scratch []byte) error {
 			if tt.HasErrorLabel(TransientTransactionError) || tt.HasErrorLabel(UnknownTransactionCommitResult) {
 				op.Client.ClearPinnedServer()
 			}
+			if e := err.(Error); retryable && op.Type == Write && e.UnsupportedStorageEngine() {
+				return ErrUnsupportedStorageEngine
+			}
 			if retryable && tt.Retryable() && retries != 0 {
 				retries--
 				original, err = err, nil
@@ -830,7 +836,7 @@ func (op Operation) addSession(dst []byte, desc description.SelectedServer) ([]b
 	dst = bsoncore.AppendDocumentElement(dst, "lsid", lsid)
 
 	var addedTxnNumber bool
-	if op.Type == Write && client != nil && client.RetryWrite {
+	if op.Type == Write && client.RetryWrite {
 		addedTxnNumber = true
 		dst = bsoncore.AppendInt64Element(dst, "txnNumber", op.Client.TxnNumber)
 	}