storage: require QueryTxn requests set a sufficient timestamp

docs/generated/settings/settings.html

@@ -121,6 +121,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>19.1-1</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>19.1-2</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

@@ -69,6 +69,7 @@ const (
 	VersionSnapshotsWithoutLog
 	Version19_1
 	VersionStart19_2
+	VersionQueryTxnTimestamp
 
 	// Add new versions here (step one of two).
 
@@ -450,15 +451,22 @@ var versionsSingleton = keyedVersions([]keyedVersion{
 		// VersionSnapshotsWithoutLog is https://github.com/cockroachdb/cockroach/pull/36714.
 		Key:     VersionSnapshotsWithoutLog,
 		Version: roachpb.Version{Major: 2, Minor: 1, Unstable: 11},
-	}, {
+	},
+	{
 		// Version19_1 is CockroachDB v19.1. It's used for all v19.1.x patch releases.
 		Key:     Version19_1,
 		Version: roachpb.Version{Major: 19, Minor: 1},
-	}, {
+	},
+	{
 		// Version19_2_Start demarcates work towards CockroachDB v19.2.
 		Key:     VersionStart19_2,
 		Version: roachpb.Version{Major: 19, Minor: 1, Unstable: 1},
 	},
+	{
+		// VersionQueryTxnTimestamp is https://github.com/cockroachdb/cockroach/pull/36307.
+		Key:     VersionQueryTxnTimestamp,
+		Version: roachpb.Version{Major: 19, Minor: 1, Unstable: 2},
+	},
 
 	// 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/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

@@ -825,6 +825,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,