storage: export CanForwardCommitTimestampWithoutRefresh

We can then use this to centralize its logic and avoid leaking it
around 1PC transaction handling.

Release note: None

pkg/kv/dist_sender_server_test.go

@@ -2464,9 +2464,6 @@ func TestTxnCoordSenderRetries(t *testing.T) {
 				b.Put("c", "put")
 				return txn.CommitInBatch(ctx, b)
 			},
-			// Parallel commits do not support the canForwardSerializableTimestamp
-			// optimization. That's ok because we need to removed that optimization
-			// anyway. See #36431.
 			txnCoordRetry: true,
 		},
 		{
@@ -2549,9 +2546,6 @@ func TestTxnCoordSenderRetries(t *testing.T) {
 				b.Put("c", "put")
 				return txn.CommitInBatch(ctx, b) // both puts will succeed, et will retry
 			},
-			// Parallel commits do not support the canForwardSerializableTimestamp
-			// optimization. That's ok because we need to removed that optimization
-			// anyway. See #36431.
 			txnCoordRetry: true,
 		},
 		{

pkg/storage/batcheval/cmd_end_transaction.go

@@ -402,7 +402,7 @@ func IsEndTransactionTriggeringRetryError(
 	}
 
 	// A transaction can still avoid a retry under certain conditions.
-	if retry && canForwardSerializableTimestamp(txn, args.NoRefreshSpans) {
+	if retry && CanForwardCommitTimestampWithoutRefresh(txn, args) {
 		retry, reason = false, 0
 	}
 
@@ -419,14 +419,17 @@ func IsEndTransactionTriggeringRetryError(
 	return retry, reason, extraMsg
 }
 
-// canForwardSerializableTimestamp returns whether a serializable txn can
-// be safely committed with a forwarded timestamp. This requires that
+// CanForwardCommitTimestampWithoutRefresh returns whether a txn can be
+// safely committed with a timestamp above its read timestamp without
+// requiring a read refresh (see txnSpanRefresher). This requires that
 // the transaction's timestamp has not leaked and that the transaction
 // has encountered no spans which require refreshing at the forwarded
 // timestamp. If either of those conditions are true, a client-side
 // retry is required.
-func canForwardSerializableTimestamp(txn *roachpb.Transaction, noRefreshSpans bool) bool {
-	return !txn.OrigTimestampWasObserved && noRefreshSpans
+func CanForwardCommitTimestampWithoutRefresh(
+	txn *roachpb.Transaction, args *roachpb.EndTransactionRequest,
+) bool {
+	return !txn.OrigTimestampWasObserved && args.NoRefreshSpans
 }
 
 const intentResolutionBatchSize = 500

pkg/storage/replica_write.go

@@ -277,18 +277,18 @@ func (r *Replica) evaluateWriteBatch(
 			strippedBa.Requests = ba.Requests[:len(ba.Requests)-1] // strip end txn req
 		}
 
-		// If there were no refreshable spans earlier in the txn
-		// (e.g. earlier gets or scans), then the batch can be retried
-		// locally in the event of write too old errors.
-		retryLocally := etArg.NoRefreshSpans && !ba.Txn.OrigTimestampWasObserved
+		// Is the transaction allowed to retry locally in the event of
+		// write too old errors? This is only allowed if it is able to
+		// forward its commit timestamp without a read refresh.
+		canForwardTimestamp := batcheval.CanForwardCommitTimestampWithoutRefresh(ba.Txn, etArg)
 
 		// If all writes occurred at the intended timestamp, we've succeeded on the fast path.
 		rec := NewReplicaEvalContext(r, spans)
 		batch, br, res, pErr := r.evaluateWriteBatchWithLocalRetries(
-			ctx, idKey, rec, &ms, &strippedBa, spans, retryLocally,
+			ctx, idKey, rec, &ms, &strippedBa, spans, canForwardTimestamp,
 		)
 		if pErr == nil && (ba.Timestamp == br.Timestamp ||
-			(retryLocally && !batcheval.IsEndTransactionExceedingDeadline(br.Timestamp, etArg))) {
+			(canForwardTimestamp && !batcheval.IsEndTransactionExceedingDeadline(br.Timestamp, etArg))) {
 			clonedTxn := ba.Txn.Clone()
 			clonedTxn.Status = roachpb.COMMITTED
 			// Make sure the returned txn has the actual commit