kvnemesis: properly skip validation for rolled-back locking reads in …

…weak-isolation transactions

Previously, #126417 attempted to skip validation for rolled-back locking
reads that tolerate write skew. However, it depended on the validator's
`observeLocking` filter in `checkAtomicCommitted` to detect if a read
tolerates write skew. However, that filter is set correctly only while
processing ops.

This patch adds a field to `observedRead` to indicate that the read
should be skipped from validation if rolled back by a savepoint. The
flag is set in `processOp` and used later in `checkAtomicCommitted`.

Fixes: #128373

Release note: None

pkg/kv/kvnemesis/validator.go

@@ -228,6 +228,22 @@ type observedRead struct {
 	SkipLocked bool
 	Value      roachpb.Value
 	ValidTimes disjointTimeSpans
+	// AlwaysValid indicates whether a read should be considered valid in the
+	// entire interval [hlc.MinTimestamp, hlc.MaxTimestamp].
+	//
+	// A lock acquired by a locking read is kept until commit time, unless it's
+	// rolled back by a savepoint; in that case, the lock is not released eagerly,
+	// but it's also not guaranteed to be held (if the transaction is pushed, the
+	// lock may be released). Serializable transactions refresh all reads at
+	// commit time, so their reads should be validated even if they were rolled
+	// back. Weaker-isolation transaction don't refresh their reads at commit
+	// time, so if a locking read from such a transaction is rolled back, we
+	// don't need to validate the read; it's valid at all times.
+	AlwaysValid bool
+	// DoNotObserveOnSavepointRollback indicates whether a read should be skipped
+	// from validation if it's rolled back by a savepoint. It's set to true for
+	// all locking reads from weak-isolation transactions.
+	DoNotObserveOnSavepointRollback bool
 }
 
 func (*observedRead) observedMarker() {}
@@ -436,6 +452,7 @@ func (v *validator) processOp(op Operation) {
 			// otherwise no lock would have been acquired on the non-existent key.
 			// Gets do not acquire gap locks.
 			observe = t.GuaranteedDurability && read.Value.IsPresent()
+			read.DoNotObserveOnSavepointRollback = true
 		default:
 			panic("unexpected")
 		}
@@ -752,9 +769,10 @@ func (v *validator) processOp(op Operation) {
 			if t.GuaranteedDurability {
 				for _, kv := range t.Result.Values {
 					read := &observedRead{
-						Key:        kv.Key,
-						SkipLocked: t.SkipLocked,
-						Value:      roachpb.Value{RawBytes: kv.Value},
+						Key:                             kv.Key,
+						SkipLocked:                      t.SkipLocked,
+						Value:                           roachpb.Value{RawBytes: kv.Value},
+						DoNotObserveOnSavepointRollback: true,
 					}
 					v.curObservations = append(v.curObservations, read)
 				}
@@ -1115,14 +1133,10 @@ func (v *validator) checkAtomicCommitted(
 				}
 			}
 		case *observedRead:
-			// If v.observationFilter == observeLocking, this must be a locking read
-			// from a weak-isolation transaction. If this locking read is rolled back
-			// by a savepoint, there is no guarantee that the lock will be present
-			// after the savepoint rollback. Since the read is not part of a
-			// serializatble transaction and is no longer locking, it's valid at all
-			// times. Alternatively, we can remove the read from txnObservations.
-			if rollbackSp != nil && v.observationFilter == observeLocking {
-				o.ValidTimes = disjointTimeSpans{{Start: hlc.MinTimestamp, End: hlc.MaxTimestamp}}
+			// If this read should not be observed on savepoint rollback, and there is
+			// a savepoint being rolled back, mark the read as valid at all times.
+			if rollbackSp != nil && o.DoNotObserveOnSavepointRollback {
+				o.AlwaysValid = true
 			}
 		case *observedSavepoint:
 			switch o.Type {
@@ -1196,9 +1210,9 @@ func (v *validator) checkAtomicCommitted(
 				}
 			}
 		case *observedRead:
-			// For rolled back locking reads from weak-isolation transactions, the
-			// ValidTimes are already set above.
-			if len(o.ValidTimes) == 0 {
+			if o.AlwaysValid {
+				o.ValidTimes = disjointTimeSpans{{Start: hlc.MinTimestamp, End: hlc.MaxTimestamp}}
+			} else {
 				o.ValidTimes = validReadTimes(batch, o.Key, o.Value.RawBytes, o.SkipLocked /* missingKeyValid */)
 			}
 		case *observedScan: