storage: perform SpanSet assertions on read-only requests

Before this change, we only performed SpanSet assertions on read-write
requests. This was a pretty huge oversight and is what allowed #23944 to
slip in. This change fixes this by adding a `engine.ReadWriter` spanset
variant and using it for read-only requests while in race mode.

It is currently blocked on #23996 and the revival of #18579.

Release note: None

pkg/storage/batcheval/cmd_range_lookup.go

@@ -50,18 +50,20 @@ func declareKeysDeprecatedRangeLookup(
 	// Both forward and reverse RangeLookupRequests scan forward initially. We are
 	// unable to bound this initial scan any further than from the lookup key to
 	// the end of the current descriptor.
-	scanBounds, err := rangeLookupScanBounds(&desc, key, false /* reverse */)
-	if err != nil {
-		// Errors will be caught during evaluation.
-		return
+	if !lookupReq.Reverse || key.Less(roachpb.RKey(keys.Meta2KeyMax)) {
+		scanBounds, err := rangeLookupScanBounds(&desc, key, false /* reverse */)
+		if err != nil {
+			// Errors will be caught during evaluation.
+			return
+		}
+		spans.Add(spanset.SpanReadOnly, roachpb.Span{
+			Key:    scanBounds.Key.AsRawKey(),
+			EndKey: scanBounds.EndKey.AsRawKey(),
+		})
 	}
-	spans.Add(spanset.SpanReadOnly, roachpb.Span{
-		Key:    scanBounds.Key.AsRawKey(),
-		EndKey: scanBounds.EndKey.AsRawKey(),
-	})
 
+	// A reverse RangeLookupRequest also scans backwards.
 	if lookupReq.Reverse {
-		// A reverse RangeLookupRequest also scans backwards.
 		revScanBounds, err := rangeLookupScanBounds(&desc, key, true /* reverse */)
 		if err != nil {
 			// Errors will be caught during evaluation.

pkg/storage/replica.go

@@ -2544,6 +2544,9 @@ func (r *Replica) executeReadOnlyBatch(
 	var result result.Result
 	rec := NewReplicaEvalContext(r, spans)
 	readOnly := r.store.Engine().NewReadOnly()
+	if util.RaceEnabled {
+		readOnly = spanset.NewReadWriter(readOnly, spans)
+	}
 	defer readOnly.Close()
 	br, result, pErr = evaluateBatch(ctx, storagebase.CmdIDKey(""), readOnly, rec, nil, ba)
 
@@ -5272,7 +5275,7 @@ func (r *Replica) evaluateWriteBatchWithLocalRetries(
 			batch.Close()
 		}
 		batch = r.store.Engine().NewBatch()
-		if util.RaceEnabled && spans != nil {
+		if util.RaceEnabled {
 			batch = spanset.NewBatch(batch, spans)
 		}
 		br, res, pErr = evaluateBatch(ctx, idKey, batch, rec, ms, ba)

pkg/storage/spanset/batch.go

@@ -307,6 +307,12 @@ func makeSpanSetReadWriter(rw engine.ReadWriter, spans *SpanSet) spanSetReadWrit
 	}
 }
 
+// NewReadWriter returns an engine.ReadWriter that asserts access of the
+// underlying ReadWriter against the given SpanSet.
+func NewReadWriter(rw engine.ReadWriter, spans *SpanSet) engine.ReadWriter {
+	return makeSpanSetReadWriter(rw, spans)
+}
+
 type spanSetBatch struct {
 	spanSetReadWriter
 	b     engine.Batch

pkg/storage/spanset/spanset.go

@@ -114,9 +114,6 @@ func (ss *SpanSet) CheckAllowed(access SpanAccess, span roachpb.Span) error {
 	}
 	for ac := access; ac < NumSpanAccess; ac++ {
 		for _, s := range ss.spans[ac][scope] {
-			if s.Key.Equal(keys.LocalMax) && s.EndKey.Equal(roachpb.KeyMax) {
-				continue
-			}
 			if s.Contains(span) {
 				return nil
 			}