kvserver: fix a reproposal check

In some situations, we repropose a command with a new LAI. When doing
so, we need to make sure that the closed timestamp has not advanced past
the command's write timestamp since the original proposal. Except we
were checking the command's read timestamp, not write timestamp.
Also changes a Less to a LessEq when comparing with the closed
timestamp, which I think is how that comparison should be.

Release note (bug fix): Fixed a very rare chance of inconsistent
follower reads.

pkg/kv/kvserver/replica_application_result.go

@@ -205,8 +205,9 @@ func (r *Replica) tryReproposeWithNewLeaseIndex(
 
 	minTS, untrack := r.store.cfg.ClosedTimestamp.Tracker.Track(ctx)
 	defer untrack(ctx, 0, 0, 0) // covers all error paths below
-	// NB: p.Request.Timestamp reflects the action of ba.SetActiveTimestamp.
-	if p.Request.Timestamp.Less(minTS) {
+	// The ConsultsTimestampCache condition matches the similar logic for caring
+	// about the closed timestamp cache in applyTimestampCache().
+	if p.Request.ConsultsTimestampCache() && p.Request.WriteTimestamp().LessEq(minTS) {
 		// The tracker wants us to forward the request timestamp, but we can't
 		// do that without re-evaluating, so give up. The error returned here
 		// will go to back to DistSender, so send something it can digest.

pkg/roachpb/api.go

@@ -166,7 +166,7 @@ func IsRange(args Request) bool {
 	return (args.flags() & isRange) != 0
 }
 
-// ConsultsTimestampCache returns whether the command must consult
+// ConsultsTimestampCache returns whether the request must consult
 // the timestamp cache to determine whether a mutation is safe at
 // a proposed timestamp or needs to move to a higher timestamp to
 // avoid re-writing history.

pkg/roachpb/batch.go

@@ -74,6 +74,19 @@ func (ba BatchRequest) EarliestActiveTimestamp() hlc.Timestamp {
 	return ts
 }
 
+// WriteTimestamp returns the timestamps at which this request is writing. For
+// non-transactional requests, this is the same as the read timestamp. For
+// transactional requests, the write timestamp can be higher until commit time.
+//
+// This should only be called after SetActiveTimestamp().
+func (ba *BatchRequest) WriteTimestamp() hlc.Timestamp {
+	ts := ba.Timestamp
+	if ba.Txn != nil {
+		ts.Forward(ba.Txn.WriteTimestamp)
+	}
+	return ts
+}
+
 // UpdateTxn updates the batch transaction from the supplied one in
 // a copy-on-write fashion, i.e. without mutating an existing
 // Transaction struct.
@@ -156,6 +169,14 @@ func (ba *BatchRequest) IsUnsplittable() bool {
 	return ba.hasFlag(isUnsplittable)
 }
 
+// ConsultsTimestampCache returns whether the request must consult
+// the timestamp cache to determine whether a mutation is safe at
+// a proposed timestamp or needs to move to a higher timestamp to
+// avoid re-writing history.
+func (ba *BatchRequest) ConsultsTimestampCache() bool {
+	return ba.hasFlag(consultsTSCache)
+}
+
 // IsSingleRequest returns true iff the BatchRequest contains a single request.
 func (ba *BatchRequest) IsSingleRequest() bool {
 	return len(ba.Requests) == 1