Merge #148364

148364: kvserver,storage: add ExpectExclusionSince to Get r=arulajmani a=stevendanna

This adds ExpectExclusionSince to Get requests. This option semantically works the same as the ExpectExclusionSince option on Put and Del requests. The implementation, however, is slightly different. It adjusts the read timestamp used by storage.MVCCGet and then configures the scanner to return an ExclusionViolationError rather than a WriteTooOld error.

Informs #142977

Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvpb/api.go

@@ -2530,6 +2530,13 @@ func (r *ConditionalPutRequest) Validate(_ Header) error {
 	return nil
 }
 
+func (r *GetRequest) Validate(_ Header) error {
+	if !r.ExpectExclusionSince.IsEmpty() && r.KeyLockingStrength == lock.None {
+		return errors.AssertionFailedf("invalid GetRequest: ExpectExclusionSince is non-empty for non-locking request")
+	}
+	return nil
+}
+
 func (r *PutRequest) Validate(bh Header) error {
 	if err := validateExclusionTimestampForBatch(r.ExpectExclusionSince, bh); err != nil {
 		return errors.NewAssertionErrorWithWrappedErrf(err, "invalid PutRequest")

pkg/kv/kvpb/api.proto

@@ -223,6 +223,15 @@ message GetRequest {
   // LockNonExisting indicates whether the Get request should acquire a lock
   // on keys that don't exist.
   bool lock_non_existing = 5;
+
+  // ExpectExclusionSince, if set, indicates that the request should return an
+  // error if this key has been written to at a timestamp at or after the given
+  // timestamp. This allows the request to verify that a previously acquired
+  // unreplicated lock has provided the required protection. Note that it does
+  // not check for the existence of the lock. It checks for a write that the
+  // lock should have prevented. If such a write is found an
+  // ExclusionViolationError is returned.
+  util.hlc.Timestamp expect_exclusion_since = 6 [(gogoproto.nullable) = false];
 }
 
 // A GetResponse is the return value from the Get() method.
@@ -2965,7 +2974,7 @@ message Header {
 
   // HasBufferedAllPrecedingWrites, if set, indicates that the batch belongs to
   // a transaction that has buffered all of its writes (from preceding batches)
-  // on the client. 
+  // on the client.
   //
   // The server may use this field to omit checking the AbortSpan. Transactions
   // use the AbortSpan to check whether they've been aborted or not. If they
@@ -2980,7 +2989,7 @@ message Header {
   // read-your-own-writes. As such, they can eschew checking the AbortSpan on
   // the server.
   bool has_buffered_all_preceding_writes = 37;
-  
+
   reserved 7, 10, 12, 14, 20;
 
   // Next ID: 38

pkg/kv/kvpb/api_test.go

@@ -370,6 +370,20 @@ func TestFlagCombinations(t *testing.T) {
 	}
 }
 
+func TestGetValidate(t *testing.T) {
+	t.Run("ExpectExclusionSinceOnNonLockingGet", func(t *testing.T) {
+		getReq := &GetRequest{ExpectExclusionSince: hlc.Timestamp{WallTime: 1}}
+		require.Error(t, getReq.Validate(Header{}))
+	})
+	t.Run("ExpectExclusionSinceOnLockingGet", func(t *testing.T) {
+		getReq := &GetRequest{
+			ExpectExclusionSince: hlc.Timestamp{WallTime: 1},
+			KeyLockingStrength:   lock.Exclusive,
+		}
+		require.NoError(t, getReq.Validate(Header{}))
+	})
+}
+
 func TestRequestHeaderRoundTrip(t *testing.T) {
 	var seq kvnemesisutil.Container
 	seq.Set(123)

pkg/kv/kvpb/batch.go

@@ -108,6 +108,15 @@ func (ba *BatchRequest) EarliestActiveTimestamp() hlc.Timestamp {
 				// NB: StartTime.Next() because StartTime is exclusive.
 				ts.Backward(t.StartTime.Next())
 			}
+		case *GetRequest:
+			// A GetRequest with ExpectExclusionSince set need to be able to observe
+			// MVCC versions from the specified time to correctly detect isolation
+			// violations.
+			//
+			// See the example in RefreshRequest for more details.
+			if !t.ExpectExclusionSince.IsEmpty() {
+				ts.Backward(t.ExpectExclusionSince.Next())
+			}
 		case *PutRequest:
 			// A PutRequest with ExpectExclusionSince set need to be able to observe MVCC
 			// versions from the specified time to correctly detect isolation

pkg/kv/kvpb/data.go

@@ -112,13 +112,6 @@ func PrepareTransactionForRetry(
 		return roachpb.Transaction{}, errors.AssertionFailedf(
 			"unexpected intent missing error (%T); should be transformed into retry error", pErr.GetDetail())
 	case *ExclusionViolationError:
-		// With a fixed read snapshot, we should never see this error as WriteTooOld will take priority.
-		if !txn.IsoLevel.PerStatementReadSnapshot() {
-			return roachpb.Transaction{}, errors.AssertionFailedf(
-				"unexpected exclusion violation error in %s transaction: %s should be transformed into retry error",
-				txn.IsoLevel,
-				pErr)
-		}
 		// An exclusion violation error always requires a restart.
 		txn.WriteTimestamp.Forward(tErr.RetryTimestamp())
 		errorAlwaysRequireRestart = true

pkg/kv/kvserver/batcheval/cmd_get.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/fs"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
 )
 
 func init() {
@@ -29,6 +30,10 @@ func Get(
 	h := cArgs.Header
 	reply := resp.(*kvpb.GetResponse)
 
+	if err := args.Validate(h); err != nil {
+		return result.Result{}, err
+	}
+
 	var lockTableForSkipLocked storage.LockTableView
 	if h.WaitPolicy == lock.WaitPolicy_SkipLocked {
 		lockTableForSkipLocked = newRequestBoundLockTableView(
@@ -37,11 +42,24 @@ func Get(
 		defer lockTableForSkipLocked.Close()
 	}
 
-	getRes, err := storage.MVCCGet(ctx, readWriter, args.Key, h.Timestamp, storage.MVCCGetOptions{
+	readTimestamp := h.Timestamp
+	failOnMoreRecent := args.KeyLockingStrength != lock.None
+	expectExclusionSince := !args.ExpectExclusionSince.IsEmpty()
+
+	// If ExpectExclusionSince is set, use it as the read timestamp and set the
+	// MoreRecentPolicy to ExclusionViolationErrorOnMoreRecent to ensure that an
+	// exclusion violation error is returned if a write has occurred since the
+	// exclusion timestamp.
+	if expectExclusionSince {
+		readTimestamp = args.ExpectExclusionSince
+		failOnMoreRecent = true
+	}
+
+	getRes, err := storage.MVCCGet(ctx, readWriter, args.Key, readTimestamp, storage.MVCCGetOptions{
 		Inconsistent:          h.ReadConsistency != kvpb.CONSISTENT,
 		SkipLocked:            h.WaitPolicy == lock.WaitPolicy_SkipLocked,
 		Txn:                   h.Txn,
-		FailOnMoreRecent:      args.KeyLockingStrength != lock.None,
+		FailOnMoreRecent:      failOnMoreRecent,
 		ScanStats:             cArgs.ScanStats,
 		Uncertainty:           cArgs.Uncertainty,
 		MemoryAccount:         cArgs.EvalCtx.GetResponseMemoryAccount(),
@@ -54,6 +72,17 @@ func Get(
 		ReturnRawMVCCValues:   args.ReturnRawMVCCValues,
 	})
 	if err != nil {
+		// If the user has set ExpectExclusionSince, transform any WriteTooOld error
+		// into an ExclusionViolationError.
+		if expectExclusionSince {
+			if wtoErr := (*kvpb.WriteTooOldError)(nil); errors.As(err, &wtoErr) {
+				err = kvpb.NewExclusionViolationError(
+					readTimestamp,
+					wtoErr.ActualTimestamp.Prev(),
+					args.Key,
+				)
+			}
+		}
 		return result.Result{}, err
 	}
 	reply.ResumeSpan = getRes.ResumeSpan