GODRIVER-2516 NoWritesPerformed Operation Error Handling (#1052)

Co-authored-by: Benjamin Rewis <[email protected]>

Author: prestonvasquez

mongo/integration/mtest/mongotest.go

@@ -637,19 +637,30 @@ func (t *T) createTestClient() {
 	if clientOpts.Deployment == nil && t.clientType != Mock && clientOpts.ServerAPIOptions == nil && testContext.requireAPIVersion {
 		clientOpts.SetServerAPIOptions(options.ServerAPI(driver.TestServerAPIVersion))
 	}
-	// command monitor
+
+	// Setup command monitor
+	var customMonitor = clientOpts.Monitor
 	clientOpts.SetMonitor(&event.CommandMonitor{
 		Started: func(_ context.Context, cse *event.CommandStartedEvent) {
+			if customMonitor != nil && customMonitor.Started != nil {
+				customMonitor.Started(context.Background(), cse)
+			}
 			t.monitorLock.Lock()
 			defer t.monitorLock.Unlock()
 			t.started = append(t.started, cse)
 		},
 		Succeeded: func(_ context.Context, cse *event.CommandSucceededEvent) {
+			if customMonitor != nil && customMonitor.Succeeded != nil {
+				customMonitor.Succeeded(context.Background(), cse)
+			}
 			t.monitorLock.Lock()
 			defer t.monitorLock.Unlock()
 			t.succeeded = append(t.succeeded, cse)
 		},
 		Failed: func(_ context.Context, cfe *event.CommandFailedEvent) {
+			if customMonitor != nil && customMonitor.Failed != nil {
+				customMonitor.Failed(context.Background(), cfe)
+			}
 			t.monitorLock.Lock()
 			defer t.monitorLock.Unlock()
 			t.failed = append(t.failed, cfe)

mongo/integration/retryable_writes_prose_test.go

@@ -9,11 +9,14 @@ package integration
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/bsontype"
 	"go.mongodb.org/mongo-driver/event"
 	"go.mongodb.org/mongo-driver/internal/testutil/assert"
 	"go.mongodb.org/mongo-driver/internal/testutil/monitor"
@@ -214,4 +217,72 @@ func TestRetryableWritesProse(t *testing.T) {
 				"expected an insert event, got a(n) %v event", cmdEvt.CommandName)
 		}
 	})
+
+	mtNWPOpts := mtest.NewOptions().MinServerVersion("6.0").Topologies(mtest.ReplicaSet)
+	mt.RunOpts(fmt.Sprintf("%s label returns original error", driver.NoWritesPerformed), mtNWPOpts,
+		func(mt *mtest.T) {
+			const shutdownInProgressErrorCode int32 = 91
+			const notWritablePrimaryErrorCode int32 = 10107
+
+			monitor := new(event.CommandMonitor)
+			mt.ResetClient(options.Client().SetRetryWrites(true).SetMonitor(monitor))
+
+			// Configure a fail point for a "ShutdownInProgress" error.
+			mt.SetFailPoint(mtest.FailPoint{
+				ConfigureFailPoint: "failCommand",
+				Mode:               mtest.FailPointMode{Times: 1},
+				Data: mtest.FailPointData{
+					WriteConcernError: &mtest.WriteConcernErrorData{
+						Code: shutdownInProgressErrorCode,
+					},
+					FailCommands: []string{"insert"},
+				},
+			})
+
+			// secondFailPointConfigured is used to determine if the conditions from the
+			// shutdownInProgressErrorCode actually configures the "NoWritablePrimary" fail command.
+			var secondFailPointConfigured bool
+
+			//Set a command monitor on the client that configures a failpoint with a "NoWritesPerformed"
+			monitor.Succeeded = func(_ context.Context, evt *event.CommandSucceededEvent) {
+				var errorCode int32
+				if wce := evt.Reply.Lookup("writeConcernError"); wce.Type == bsontype.EmbeddedDocument {
+					var ok bool
+					errorCode, ok = wce.Document().Lookup("code").Int32OK()
+					if !ok {
+						t.Fatalf("expected code to be an int32, got %v",
+							wce.Document().Lookup("code").Type)
+						return
+					}
+				}
+
+				// Do not set a fail point if event was not a writeConcernError with an error code for
+				// "ShutdownInProgress".
+				if errorCode != shutdownInProgressErrorCode {
+					return
+				}
+
+				mt.SetFailPoint(mtest.FailPoint{
+					ConfigureFailPoint: "failCommand",
+					Mode:               mtest.FailPointMode{Times: 1},
+					Data: mtest.FailPointData{
+						ErrorCode: notWritablePrimaryErrorCode,
+						ErrorLabels: &[]string{
+							driver.NoWritesPerformed,
+							driver.RetryableWriteError,
+						},
+						FailCommands: []string{"insert"},
+					},
+				})
+				secondFailPointConfigured = true
+			}
+
+			// Attempt to insert a document.
+			_, err := mt.Coll.InsertOne(context.Background(), bson.D{{"x", 1}})
+
+			require.True(mt, secondFailPointConfigured)
+
+			// Assert that the "NotWritablePrimary" error is returned.
+			require.True(mt, err.(mongo.WriteException).HasErrorCode(int(shutdownInProgressErrorCode)))
+		})
 }

x/mongo/driver/errors.go

@@ -37,6 +37,8 @@ var (
 	NetworkError = "NetworkError"
 	// RetryableWriteError is an error lable for retryable write errors.
 	RetryableWriteError = "RetryableWriteError"
+	// NoWritesPerformed is an error label indicated that no writes were performed for an operation.
+	NoWritesPerformed = "NoWritesPerformed"
 	// ErrCursorNotFound is the cursor not found error for legacy find operations.
 	ErrCursorNotFound = errors.New("cursor not found")
 	// ErrUnacknowledgedWrite is returned from functions that have an unacknowledged
@@ -132,6 +134,18 @@ func (wce WriteCommandError) Retryable(wireVersion *description.VersionRange) bo
 	return (*wce.WriteConcernError).Retryable()
 }
 
+// HasErrorLabel returns true if the error contains the specified label.
+func (wce WriteCommandError) HasErrorLabel(label string) bool {
+	if wce.Labels != nil {
+		for _, l := range wce.Labels {
+			if l == label {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // WriteConcernError is a write concern failure that occurred as a result of a
 // write operation.
 type WriteConcernError struct {

x/mongo/driver/operation.go

@@ -59,6 +59,11 @@ type RetryablePoolError interface {
 	Retryable() bool
 }
 
+// LabeledError is an error that can have error labels added to it.
+type LabelledError interface {
+	HasErrorLabel(string) bool
+}
+
 // InvalidOperationError is returned from Validate and indicates that a required field is missing
 // from an instance of Operation.
 type InvalidOperationError struct{ MissingField string }
@@ -370,6 +375,7 @@ func (op Operation) Execute(ctx context.Context) error {
 	var res bsoncore.Document
 	var operationErr WriteCommandError
 	var prevErr error
+	var prevIndefiniteErr error
 	batching := op.Batches.Valid()
 	retryEnabled := op.RetryMode != nil && op.RetryMode.Enabled()
 	retrySupported := false
@@ -381,6 +387,16 @@ func (op Operation) Execute(ctx context.Context) error {
 	resetForRetry := func(err error) {
 		retries--
 		prevErr = err
+
+		// Set the previous indefinite error to be returned in any case where a retryable write error does not have a
+		// NoWritesPerfomed label (the definite case).
+		switch err := err.(type) {
+		case LabelledError:
+			if !err.HasErrorLabel(NoWritesPerformed) && err.HasErrorLabel(RetryableWriteError) {
+				prevIndefiniteErr = err.(error)
+			}
+		}
+
 		// If we got a connection, close it immediately to release pool resources for
 		// subsequent retries.
 		if conn != nil {
@@ -615,6 +631,12 @@ func (op Operation) Execute(ctx context.Context) error {
 				continue
 			}
 
+			// If the error is no longer retryable and has the NoWritesPerformed label, then we should
+			// return the previous indefinite error.
+			if tt.HasErrorLabel(NoWritesPerformed) {
+				return prevIndefiniteErr
+			}
+
 			// If the operation isn't being retried, process the response
 			if op.ProcessResponseFn != nil {
 				info := ResponseInfo{
@@ -702,6 +724,12 @@ func (op Operation) Execute(ctx context.Context) error {
 				continue
 			}
 
+			// If the error is no longer retryable and has the NoWritesPerformed label, then we should
+			// return the previous indefinite error.
+			if tt.HasErrorLabel(NoWritesPerformed) {
+				return prevIndefiniteErr
+			}
+
 			// If the operation isn't being retried, process the response
 			if op.ProcessResponseFn != nil {
 				info := ResponseInfo{