kvcoord: treat non-transactional writes as non-idempotent

Previously, non-transactional requests were treated as idempotent by the
`DistSender`, meaning they were retried in the presence of RPC errors.
This could lead to double-evaluating these requests, including writes.
This can result in, for example, an increment applying twice, or more
subtle problems like a blind write evaluating twice, overwriting
another unrelated write that fell in-between. This issue is known and
documented in the `DistSender` code. It is not a production correctness
issue since SQL always uses the transactional KV API.

This commit treats non-transactional write batches the same way as
transactional batches that contain a commit: if they experience an RPC
error, they should not be retried, and should return an ambiguous
error. The new logic is currently behind an off-by-default cluster
setting as it will require deflaking many tests. The plan is to enable
this in some kvnemesis tests first.

Informs: #154389
Informs: #114814

Release note: None

Author: miraradeva

pkg/kv/kvclient/kvcoord/dist_sender.go

@@ -419,6 +419,20 @@ var ProxyBatchRequest = settings.RegisterBoolSetting(
 	true,
 )
 
+// NonTransactionalWritesNotIdempotent controls whether non-transactional writes
+// are considered idempotent or not. When this setting is true, a
+// non-transactional write that experiences an RPC error is not retried, and
+// returns an ambiguous error. This is the same behavior as commit batches (or
+// batched issued concurrently with a commit batch). This is arguably the
+// correct behavior for non-transactional writes, but it's behind a default-off
+// cluster setting to get some kvnemesis mileage first.
+var NonTransactionalWritesNotIdempotent = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"kv.dist_sender.non_transactional_writes_not_idempotent.enabled",
+	"when true, non-transactional writes are not retried and may return an ambiguous error",
+	false,
+)
+
 // DistSenderMetrics is the set of metrics for a given distributed sender.
 type DistSenderMetrics struct {
 	BatchCount                         *metric.Counter
@@ -2541,7 +2555,13 @@ const slowDistSenderReplicaThreshold = 10 * time.Second
 func (ds *DistSender) sendToReplicas(
 	ctx context.Context, ba *kvpb.BatchRequest, routing rangecache.EvictionToken, withCommit bool,
 ) (*kvpb.BatchResponse, error) {
-
+	// In addition to batches where withCommit is true, non-transactional write
+	// batches are also not safe to be retried as they are not guaranteed to be
+	// idempotent. Returning ambiguous errors for those batches is controlled by a
+	// cluster setting for now.
+	nonIdempotentWrite :=
+		ba.Txn == nil && ba.IsWrite() && NonTransactionalWritesNotIdempotent.Get(&ds.st.SV)
+	nonIdempotent := withCommit || nonIdempotentWrite
 	// If this request can be sent to a follower to perform a consistent follower
 	// read under the closed timestamp, promote its routing policy to NEAREST.
 	// If we don't know the closed timestamp policy, we ought to optimistically
@@ -2759,7 +2779,9 @@ func (ds *DistSender) sendToReplicas(
 		}
 
 		tBegin := crtime.NowMono() // for slow log message
-		sendCtx, cbToken, cbErr := ds.circuitBreakers.ForReplica(desc, &curReplica).Track(ctx, ba, withCommit, tBegin)
+		sendCtx, cbToken, cbErr := ds.circuitBreakers.ForReplica(desc, &curReplica).Track(
+			ctx, ba, nonIdempotent, tBegin,
+		)
 		if cbErr != nil {
 			// Circuit breaker is tripped. err will be handled below.
 			err = cbErr
@@ -2861,12 +2883,12 @@ func (ds *DistSender) sendToReplicas(
 			// prevents them from double evaluation. This can result in, for example,
 			// an increment applying twice, or more subtle problems like a blind write
 			// evaluating twice, overwriting another unrelated write that fell
-			// in-between.
+			// in-between. This is fixed under the cluster setting
+			// NonTransactionalWritesNotIdempotent. Consider enabling it by default.
 			//
-			// NB: If this partial batch does not contain the EndTxn request but the
-			// batch contains a commit, the ambiguous error should be caught on
-			// retrying the writes, should it need to be propagated.
-			if withCommit && !grpcutil.RequestDidNotStart(err) {
+			// NB: If this partial batch is not idempotent, the ambiguous error should
+			// be caught on retrying the writes, should it need to be propagated.
+			if nonIdempotent && !grpcutil.RequestDidNotStart(err) {
 				ambiguousError = err
 			}
 			// If we get a gRPC error against the leaseholder, we don't want to
@@ -2987,9 +3009,9 @@ func (ds *DistSender) sendToReplicas(
 					// return it if all other replicas fail (regardless of error).
 					replicaUnavailableError = br.Error.GoError()
 				}
-				// The circuit breaker may have tripped while a commit proposal was in
-				// flight, so we have to mark it as ambiguous as well.
-				if withCommit && ambiguousError == nil {
+				// The circuit breaker may have tripped while a non-idempotent request
+				// was in flight, so we have to mark it as ambiguous as well.
+				if nonIdempotent && ambiguousError == nil {
 					ambiguousError = br.Error.GoError()
 				}
 			case *kvpb.NotLeaseHolderError:

pkg/kv/kvclient/kvcoord/dist_sender_test.go

@@ -3664,6 +3664,7 @@ func TestReplicaErrorsMerged(t *testing.T) {
 
 	notLeaseHolderErr := kvpb.NewError(kvpb.NewNotLeaseHolderError(lease3, 0, &descriptor2, ""))
 	startedRequestError := errors.New("request might have started")
+	notStartedRequestError := grpcstatus.Errorf(codes.PermissionDenied, "request did not start")
 	unavailableError1 := kvpb.NewError(kvpb.NewReplicaUnavailableError(errors.New("unavailable"), &initDescriptor, initDescriptor.InternalReplicas[0]))
 	unavailableError2 := kvpb.NewError(kvpb.NewReplicaUnavailableError(errors.New("unavailable"), &initDescriptor, initDescriptor.InternalReplicas[1]))
 
@@ -3675,52 +3676,101 @@ func TestReplicaErrorsMerged(t *testing.T) {
 	// See https://cockroachlabs.com/blog/demonic-nondeterminism/#appendix for
 	// the gory details.
 	testCases := []struct {
+		transactional      bool
 		withCommit         bool
 		sendErr1, sendErr2 error
 		err1, err2         *kvpb.Error
 		expErr             string
 	}{
 		// The ambiguous error is returned with higher priority for withCommit.
 		{
-			withCommit: true,
-			sendErr1:   startedRequestError,
-			err2:       notLeaseHolderErr,
-			expErr:     "result is ambiguous",
+			transactional: true,
+			withCommit:    true,
+			sendErr1:      startedRequestError,
+			err2:          notLeaseHolderErr,
+			expErr:        "result is ambiguous",
 		},
 		// The not leaseholder errors is the last error.
 		{
-			withCommit: false,
-			sendErr1:   startedRequestError,
-			err2:       notLeaseHolderErr,
-			expErr:     "leaseholder not found in transport",
+			transactional: true,
+			withCommit:    false,
+			sendErr1:      startedRequestError,
+			err2:          notLeaseHolderErr,
+			expErr:        "leaseholder not found in transport",
 		},
 		// The ambiguous error is returned with higher priority for withCommit.
 		{
-			withCommit: true,
-			sendErr1:   startedRequestError,
-			err2:       unavailableError2,
-			expErr:     "result is ambiguous",
+			transactional: true,
+			withCommit:    true,
+			sendErr1:      startedRequestError,
+			err2:          unavailableError2,
+			expErr:        "result is ambiguous",
 		},
 		// The unavailable error is the last error.
 		{
-			withCommit: false,
-			sendErr1:   startedRequestError,
-			err2:       unavailableError2,
-			expErr:     "unavailable",
+			transactional: true,
+			withCommit:    false,
+			sendErr1:      startedRequestError,
+			err2:          unavailableError2,
+			expErr:        "unavailable",
+		},
+		// The ambiguous error is returned with higher priority for
+		// non-transactional batches (next 2 test cases). This is the case only
+		// because the test sets NonTransactionalWritesNotIdempotent = true.
+		// Otherwise, the non-transactional requests would be treated like they are
+		// idempotent and the NLHE/RUE would be returned as the last error.
+		{
+			transactional: false,
+			withCommit:    false,
+			sendErr1:      startedRequestError,
+			err2:          notLeaseHolderErr,
+			expErr:        "result is ambiguous",
 		},
-		// The unavailable error is returned with higher priority regardless of withCommit.
 		{
-			withCommit: true,
-			err1:       unavailableError1,
-			err2:       notLeaseHolderErr,
-			expErr:     "unavailable",
+			transactional: false,
+			withCommit:    false,
+			sendErr1:      startedRequestError,
+			err2:          unavailableError2,
+			expErr:        "result is ambiguous",
+		},
+		// If we know the request did not start, do not return an ambiguous error
+		// (next 2 test cases).
+		{
+			transactional: false,
+			withCommit:    false,
+			sendErr1:      notStartedRequestError,
+			err2:          notLeaseHolderErr,
+			expErr:        "leaseholder not found in transport",
+		},
+		{
+			transactional: false,
+			withCommit:    false,
+			sendErr1:      notStartedRequestError,
+			err2:          unavailableError2,
+			expErr:        "unavailable",
+		},
+		// The unavailable error is returned with higher priority regardless of
+		// withCommit and transactional (next 3 test cases).
+		{
+			transactional: true,
+			withCommit:    true,
+			err1:          unavailableError1,
+			err2:          notLeaseHolderErr,
+			expErr:        "unavailable",
 		},
-		// The unavailable error is returned with higher priority regardless of withCommit.
 		{
-			withCommit: false,
-			err1:       unavailableError1,
-			err2:       notLeaseHolderErr,
-			expErr:     "unavailable",
+			transactional: true,
+			withCommit:    false,
+			err1:          unavailableError1,
+			err2:          notLeaseHolderErr,
+			expErr:        "unavailable",
+		},
+		{
+			transactional: false,
+			withCommit:    false,
+			err1:          unavailableError1,
+			err2:          notLeaseHolderErr,
+			expErr:        "unavailable",
 		},
 	}
 	clock := hlc.NewClockForTesting(nil)
@@ -3744,6 +3794,7 @@ func TestReplicaErrorsMerged(t *testing.T) {
 				stopper := stop.NewStopper()
 				defer stopper.Stop(ctx)
 				st := cluster.MakeTestingClusterSettings()
+				NonTransactionalWritesNotIdempotent.Override(ctx, &st.SV, true)
 				rc := rangecache.NewRangeCache(st, nil /* db */, func() int64 { return 100 }, stopper)
 				rc.Insert(ctx, roachpb.RangeInfo{
 					Desc:  initDescriptor,
@@ -3786,12 +3837,16 @@ func TestReplicaErrorsMerged(t *testing.T) {
 						return nil, nil, errors.New("range desc db unexpectedly used")
 					}),
 					TransportFactory: adaptSimpleTransport(transportFn),
-					Settings:         cluster.MakeTestingClusterSettings(),
+					Settings:         st,
 				}
 				ds := NewDistSender(cfg)
 
 				ba := &kvpb.BatchRequest{}
 				ba.Add(kvpb.NewGet(roachpb.Key("a")))
+				ba.Add(kvpb.NewPut(roachpb.Key("b"), roachpb.MakeValueFromString("value")))
+				if tc.transactional {
+					ba.Txn = &roachpb.Transaction{Name: "test"}
+				}
 				tok, err := rc.LookupWithEvictionToken(ctx, roachpb.RKeyMin, rangecache.EvictionToken{}, false)
 				require.NoError(t, err)
 				br, err := ds.sendToReplicas(ctx, ba, tok, tc.withCommit)