storage: require QueryTxn requests set a sufficient timestamp

This commit adds a check that QueryTxn requests set sufficiently high
timestamps on their request header. This is in the spirit of cockroachdb#35297.

Even though QueryTxn will never change the timestamp cache, I think
there is a very rare hazard if it doesn't provide a timestamp at
least as large as the transaction's timestamp in its batch header.
If it doesn't do this, it seems possible that it could evaluate
concurrently with a lease transfer (the lease transfer would have
to start right after). The timestamp cache could then rotate pages
a number of times until its low water mark is above the lease
transfer timestamp (which is crazy as this would need to take at
least 10s), and the QueryTxn could then consider a transaction ABORTED
while evaluating because of the timestamp cache low water mark. The
new leaseholder would have a lower timestamp cache low water mark
and the transaction might still be able to create its transaction
record.

I don't think we would have ever seen this in the wild because of
the 10s stall and the repeat timestamp cache rotations rotations,
but this seems like a good change to make anyway.

Release note: None

docs/generated/settings/settings.html

@@ -112,6 +112,6 @@
 <tr><td><code>trace.debug.enable</code></td><td>boolean</td><td><code>false</code></td><td>if set, traces for recent requests can be seen in the /debug page</td></tr>
 <tr><td><code>trace.lightstep.token</code></td><td>string</td><td><code></code></td><td>if set, traces go to Lightstep using this token</td></tr>
 <tr><td><code>trace.zipkin.collector</code></td><td>string</td><td><code></code></td><td>if set, traces go to the given Zipkin instance (example: '127.0.0.1:9411'); ignored if trace.lightstep.token is set</td></tr>
-<tr><td><code>version</code></td><td>custom validation</td><td><code>2.1-10</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
+<tr><td><code>version</code></td><td>custom validation</td><td><code>2.1-11</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
 </tbody>
 </table>

pkg/kv/txn_coord_sender_test.go

@@ -404,6 +404,7 @@ func getTxn(ctx context.Context, txn *client.Txn) (*roachpb.Transaction, *roachp
 	}
 
 	ba := roachpb.BatchRequest{}
+	ba.Timestamp = txnMeta.Timestamp
 	ba.Add(qt)
 
 	db := txn.DB()

pkg/settings/cluster/cockroach_versions.go

@@ -66,6 +66,7 @@ const (
 	VersionDirectImport
 	VersionSideloadedStorageNoReplicaID // see versionsSingleton for details
 	VersionPushTxnToInclusive
+	VersionQueryTxnTimestamp
 
 	// Add new versions here (step one of two).
 
@@ -443,6 +444,11 @@ var versionsSingleton = keyedVersions([]keyedVersion{
 		Key:     VersionPushTxnToInclusive,
 		Version: roachpb.Version{Major: 2, Minor: 1, Unstable: 10},
 	},
+	{
+		// VersionQueryTxnTimestamp is https://github.com/cockroachdb/cockroach/pull/36307.
+		Key:     VersionQueryTxnTimestamp,
+		Version: roachpb.Version{Major: 2, Minor: 1, Unstable: 11},
+	},
 
 	// Add new versions here (step two of two).
 

pkg/storage/batcheval/cmd_push_txn.go

@@ -115,18 +115,20 @@ func PushTxn(
 	if h.Txn != nil {
 		return result.Result{}, ErrTransactionUnsupported
 	}
-
-	// Verify that the PushTxn's timestamp is not less than the timestamp that
-	// the request intends to push the transaction to. Transactions should not
-	// be pushed into the future or their effect may not be fully reflected in
-	// a future leaseholder's timestamp cache. This is analogous to how reads
-	// should not be performed at a timestamp in the future.
 	if h.Timestamp.Less(args.PushTo) {
-		return result.Result{}, errors.Errorf("PushTo %s larger than PushRequest header timestamp %s", args.PushTo, h.Timestamp)
+		// Verify that the PushTxn's timestamp is not less than the timestamp that
+		// the request intends to push the transaction to. Transactions should not
+		// be pushed into the future or their effect may not be fully reflected in
+		// a future leaseholder's timestamp cache. This is analogous to how reads
+		// should not be performed at a timestamp in the future.
+		return result.Result{}, errors.Errorf("request timestamp %s less than PushTo timestamp %s", h.Timestamp, args.PushTo)
+	}
+	if h.Timestamp.Less(args.PusheeTxn.Timestamp) {
+		// This condition must hold for the timestamp cache access/update to be safe.
+		return result.Result{}, errors.Errorf("request timestamp %s less than pushee txn timestamp %s", h.Timestamp, args.PusheeTxn.Timestamp)
 	}
-
 	if !bytes.Equal(args.Key, args.PusheeTxn.Key) {
-		return result.Result{}, errors.Errorf("request key %s should match pushee's txn key %s", args.Key, args.PusheeTxn.Key)
+		return result.Result{}, errors.Errorf("request key %s should match pushee txn key %s", args.Key, args.PusheeTxn.Key)
 	}
 	key := keys.TransactionKey(args.PusheeTxn.Key, args.PusheeTxn.ID)
 

pkg/storage/batcheval/cmd_query_txn.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/storage/batcheval/result"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine"
 	"github.com/cockroachdb/cockroach/pkg/storage/spanset"
@@ -49,11 +50,20 @@ func QueryTxn(
 	ctx context.Context, batch engine.ReadWriter, cArgs CommandArgs, resp roachpb.Response,
 ) (result.Result, error) {
 	args := cArgs.Args.(*roachpb.QueryTxnRequest)
+	h := cArgs.Header
 	reply := resp.(*roachpb.QueryTxnResponse)
 
-	if cArgs.Header.Txn != nil {
+	if h.Txn != nil {
 		return result.Result{}, ErrTransactionUnsupported
 	}
+	// TODO(nvanbenschoten): old clusters didn't attach header timestamps to
+	// QueryTxn requests, so only perform this check for clusters that will
+	// always attach a valid timestamps.
+	checkHeaderTS := cArgs.EvalCtx.ClusterSettings().Version.IsActive(cluster.VersionQueryTxnTimestamp)
+	if h.Timestamp.Less(args.Txn.Timestamp) && checkHeaderTS {
+		// This condition must hold for the timestamp cache access to be safe.
+		return result.Result{}, errors.Errorf("request timestamp %s less than txn timestamp %s", h.Timestamp, args.Txn.Timestamp)
+	}
 	if !bytes.Equal(args.Key, args.Txn.Key) {
 		return result.Result{}, errors.Errorf("request key %s does not match txn key %s", args.Key, args.Txn.Key)
 	}

pkg/storage/batcheval/cmd_recover_txn.go

@@ -64,8 +64,8 @@ func RecoverTxn(
 		return result.Result{}, errors.Errorf("request key %s does not match txn key %s", args.Key, args.Txn.Key)
 	}
 	if h.Timestamp.Less(args.Txn.Timestamp) {
-		// This condition must hold for the timestamp cache update to be safe.
-		return result.Result{}, errors.Errorf("request timestamp %v less than txn timestamp %v", h.Timestamp, args.Txn.Timestamp)
+		// This condition must hold for the timestamp cache access/update to be safe.
+		return result.Result{}, errors.Errorf("request timestamp %s less than txn timestamp %s", h.Timestamp, args.Txn.Timestamp)
 	}
 	key := keys.TransactionKey(args.Txn.Key, args.Txn.ID)
 

pkg/storage/replica_command.go

@@ -374,6 +374,9 @@ func (r *Replica) AdminMerge(
 	// descriptor end key. We look up the descriptor here only to get
 	// the new end key and then repeat the lookup inside the
 	// transaction.
+	// TODO(nvanbenschoten): Why can't we do this in the transaction?
+	// Performing an initial read won't affect the transaction record
+	// placement.
 	{
 		var rightDesc roachpb.RangeDescriptor
 		if err := r.store.DB().GetProto(ctx, rightDescKey, &rightDesc); err != nil {

pkg/storage/txnrecovery/manager.go

@@ -252,6 +252,7 @@ func (m *manager) resolveIndeterminateCommitForTxnProbe(
 	// write is prevented, or we run out of in-flight writes to query.
 	for len(queryIntentReqs) > 0 {
 		var b client.Batch
+		b.Header.Timestamp = m.clock.Now()
 		b.AddRawRequest(&queryTxnReq)
 		for i := 0; i < defaultBatchSize && len(queryIntentReqs) > 0; i++ {
 			b.AddRawRequest(&queryIntentReqs[0])

pkg/storage/txnwait/txnqueue.go

@@ -821,6 +821,7 @@ func (q *Queue) queryTxnStatus(
 	now hlc.Timestamp,
 ) (*roachpb.Transaction, []uuid.UUID, *roachpb.Error) {
 	b := &client.Batch{}
+	b.Header.Timestamp = q.store.Clock().Now()
 	b.AddRawRequest(&roachpb.QueryTxnRequest{
 		RequestHeader: roachpb.RequestHeader{
 			Key: txnMeta.Key,