kv: bump timestamp cache to MinTimestamp on PUSH_ABORT

Fixes #60779.

We were only checking that the batch header timestamp was equal to or
greater than this pushee's min timestamp, so this is as far as we can
bump the timestamp cache.

pkg/kv/kvserver/replica_tscache.go

@@ -136,16 +136,21 @@ func (r *Replica) updateTimestampCache(
 				addToTSCache(key, nil, ts, recovered.ID)
 			}
 		case *roachpb.PushTxnRequest:
-			// A successful PushTxn request bumps the timestamp cache for
-			// the pushee's transaction key. The pushee will consult the
-			// timestamp cache when creating its record. If the push left
-			// the transaction in a PENDING state (PUSH_TIMESTAMP) then we
-			// update the timestamp cache. This will cause the creator
-			// of the transaction record to forward its provisional commit
-			// timestamp to honor the result of this push. If the push left
-			// the transaction in an ABORTED state (PUSH_ABORT) then we
-			// update the a special record in the timestamp cache. This will prevent
-			// the creation of the transaction record entirely.
+			// A successful PushTxn request bumps the timestamp cache for the
+			// pushee's transaction key. The pushee will consult the timestamp
+			// cache when creating its record - see CanCreateTxnRecord.
+			//
+			// If the push left the transaction in a PENDING state
+			// (PUSH_TIMESTAMP) then we add a "push" marker to the timestamp
+			// cache with the push time. This will cause the creator of the
+			// transaction record to forward its provisional commit timestamp to
+			// honor the result of this push, preventing it from committing at
+			// any prior time.
+			//
+			// If the push left the transaction in an ABORTED state (PUSH_ABORT)
+			// then we add a "tombstone" marker to the timestamp cache wth the
+			// pushee's minimum timestamp. This will prevent the creation of the
+			// transaction record entirely.
 			pushee := br.Responses[i].GetInner().(*roachpb.PushTxnResponse).PusheeTxn
 
 			var tombstone bool
@@ -165,12 +170,25 @@ func (r *Replica) updateTimestampCache(
 			}
 
 			var key roachpb.Key
+			var pushTS hlc.Timestamp
 			if tombstone {
+				// NOTE: we only push to the pushee's MinTimestamp and not to
+				// its WriteTimestamp here because we don't want to add an entry
+				// to the timestamp cache at a higher time than the request's
+				// timestamp. The request's PushTo field is validated to be less
+				// than or equal to its timestamp during evaluation. However, a
+				// PUSH_ABORT may not contain a PushTo timestamp or may contain
+				// a PushTo timestamp that lags the pushee's WriteTimestamp. So
+				// if we were to bump the timestamp cache to the WriteTimestamp,
+				// we could risk adding an entry at a time in advance of the
+				// local clock.
 				key = transactionTombstoneMarker(start, pushee.ID)
+				pushTS = pushee.MinTimestamp
 			} else {
 				key = transactionPushMarker(start, pushee.ID)
+				pushTS = pushee.WriteTimestamp
 			}
-			addToTSCache(key, nil, pushee.WriteTimestamp, t.PusherTxn.ID)
+			addToTSCache(key, nil, pushTS, t.PusherTxn.ID)
 		case *roachpb.ConditionalPutRequest:
 			// ConditionalPut only updates on ConditionFailedErrors. On other
 			// errors, no information is returned. On successful writes, the

pkg/kv/kvserver/txnwait/queue.go

@@ -919,6 +919,7 @@ func (q *Queue) forcePushAbort(
 	forcePush.PushType = roachpb.PUSH_ABORT
 	b := &kv.Batch{}
 	b.Header.Timestamp = q.cfg.Clock.Now()
+	b.Header.Timestamp.Forward(req.PushTo)
 	b.AddRawRequest(&forcePush)
 	if err := q.cfg.DB.Run(ctx, b); err != nil {
 		return nil, b.MustPErr()