storage: allow HeartbeatTxn and EndTxn requests to create txn records

Based on #33257.
Informs #25437.
Informs #32971.

This is the first part of addressing #32971. Part two will update
concurrent txns to not immediately abort missing txn records and
part three will update the txn client to stop sending BeginTxn
records.

The change closely resembles what was laid out in the corresponding
RFC sections. `HeartbeatTxn` and `EndTxn` are both updated to permit
the creation of transaction records when they finds that one is missing.

To prevent replays from causing issues, they *both* check the write
timestamp cache when creating a transaction record. This is one area
where the change diverges from the RFC. Instead of having `EndTxn`
always check the write timestamp cache, it only does so if it is
creating a txn record from scratch. To make this safe, `HeartbeatTxn`
also checks the write timestamp cache if it is creating a txn record
from scratch. This prevents a long running transaction from increasingly
running the risk of being aborted by a lease transfer as its `EndTxn`
continues to be delayed. Instead, a lease transfer will only abort a
transaction if it comes before the transaction record creation, which
should be within a second of the transaction starting.

The change pulls out a new `CanCreateTxnRecord` method, which has the
potential of being useful for detecting eagerly GCed transaction records
and solving the major problem from the RFC without an extra QueryIntent.
This is what Andrei was pushing for before. More details are included
in TODOs within the PR.

_### Migration Safety

Most of the changes to the transaction state machine are straightforward
to validate. Transaction records can still never be created without
checking for replays and only an EndTxn from the client's sync path
can GC an ABORTED txn record. This means that it is still impossible for
a transaction to move between the two finalized statuses (at some point
it would be worth it to draw this state machine).

The one tricky part is ensuring that the changes in this PR are safe
when run in mixed-version clusters. The two areas of concern are:
1. lease transfers between a new and an old cluster on a range that
should/does contain a transaction record.
2. an old txn client that is not expecting HeartbeatTxn and EndTxn
requests to create txn records if they are found to be missing.
3. an old txn client may not expect a HeartbeatTxn to hit the
write timestamp cache if run concurrently with an EndTxn.

The first concern is protected by the write timestamp cache. Regardless
of which replica creates that transaction record from a BeginTxn req or
a HeartbeatTxn req (on the new replica), it will first need to check
the timestamp cache. This prevents against any kind of replay that could
create a transaction record that the old replica is not prepared to
handle.

We can break the second concern into two parts. First, an old txn client
will never send an EndTxn without having sent a successful BeginTxn, so
the new behavior there is not an issue. Second, an old txn client may
send a HeartbeatTxn concurrently with a BeginTxn. If the heartbeat wins,
it will create a txn record on a new replica. If the BeginTxn evaluates
on a new replica, it will be a no-op. If it evaluates on an old replica,
it will result in a retryable error.

The third concern is specific to the implementation of the heartbeat loop
itself. If a HeartbeatTxn loses a race with an EndTxn, it may get an
aborted error after checking the timestamp cache. This is desirable from
the replica-side, but we'd need to ensure that the client will handle it
correctly. This PR adds some protection (see `sendingEndTxn`) to the txn
client to prevent this case from causing weirdness on the client, but I
don't think this could actually cause issues even without this protection.
The reason is that the txn client might mark itself as aborted due to the
heartbeat, but this will be overwritten when the EndTxn returns and the sql
client will still hear back from the successful EndTxn. This must have
actually always been an issue because it was always possible for a committed
txn record to be GCed and then written as aborted later, at which point a
concurrent heartbeat could have observed it.

I'd like Andrei to sign off on this last hazard, as he's thought about
this kind of thing more than anybody.

Release note: None

pkg/kv/integration_test.go

@@ -51,6 +51,7 @@ import (
 // 3) Another txn runs into the intents on r2, tries to push, and succeeds
 // because the txn record doesn't exist yet. It writes the txn record as
 // Aborted.
+// TODO(nvanbenschoten): This test will change when #25437 is fully addressed.
 // 4) If the Begin were to execute now, it'd discover the Aborted txn and return
 // a retryable TxnAbortedError. But it doesn't execute now; it's still delayed
 // somehow.
@@ -67,6 +68,10 @@ import (
 // result of this convoluted scenario.
 func TestDelayedBeginRetryable(t *testing.T) {
 	defer leaktest.AfterTest(t)()
+	t.Skip(`WIP: @andrei what should we do here? Now that the timestamp
+	cache check is below the abort span, we're hitting ABORT_REASON_ABORT_SPAN
+	instead. Can we rely on the abort span or do we need to clear it and test
+	that we still hit the timestamp cache without its protection?`)
 
 	// Here's how this test is gonna go:
 	// - We're going to send a BeginTxn+Put(a)+Put(c). The first two will be split
@@ -183,7 +188,7 @@ func TestDelayedBeginRetryable(t *testing.T) {
 	if _, ok := pErr.GetDetail().(*roachpb.HandledRetryableTxnError); !ok {
 		t.Fatalf("expected HandledRetryableTxnError, got: %v", pErr)
 	}
-	exp := "TransactionAbortedError(ABORT_REASON_ALREADY_COMMITTED_OR_ROLLED_BACK_POSSIBLE_REPLAY)"
+	exp := "TransactionAbortedError(ABORT_REASON_ABORT_SPAN)"
 	if !testutils.IsPError(pErr, regexp.QuoteMeta(exp)) {
 		t.Fatalf("expected %s, got: %s", exp, pErr)
 	}

pkg/kv/txn_interceptor_heartbeat.go

@@ -97,6 +97,16 @@ type txnHeartbeat struct {
 		// EndTransaction can be elided - we want to allow multiple rollback attempts
 		// to be sent and the first one stops the heartbeat loop.
 		everSentBeginTxn bool
+
+		// sendingEndTxn is set when an EndTransactionRequest is in flight to the
+		// server. It is used to signify to the heartbeat loop that it should ignore
+		// the results of any heartbeat request.
+		//
+		// The field is not needed for client-perceived correctness, but prevents the
+		// coordinator's local transaction status from flipping to ABORTED after a
+		// heartbeat that observes the effect of a concurrent EndTransactionRequest
+		// and then over to COMMITTED after the EndTransactionRequest returns.
+		sendingEndTxn bool
 	}
 }
 
@@ -203,6 +213,7 @@ func (h *txnHeartbeat) SendLocked(
 	var elideEndTxn bool
 	var commitTurnedToRollback bool
 	if haveEndTxn {
+		h.mu.sendingEndTxn = true
 		// Are we writing now or have we written in the past?
 		elideEndTxn = !h.mu.everSentBeginTxn
 		if elideEndTxn {
@@ -252,6 +263,10 @@ func (h *txnHeartbeat) SendLocked(
 		}
 	}
 
+	if haveEndTxn {
+		h.mu.sendingEndTxn = false
+	}
+
 	if pErr != nil {
 		return nil, pErr
 	}
@@ -451,6 +466,13 @@ func (h *txnHeartbeat) heartbeat(ctx context.Context) bool {
 	log.VEvent(ctx, 2, "heartbeat")
 	br, pErr := h.gatekeeper.SendLocked(ctx, ba)
 
+	// If there is an in-flight EndTransaction request, avoid updating our local
+	// state based on the result of heartbeat requests. Instead, wait for the
+	// EndTransaction to return.
+	if h.mu.sendingEndTxn {
+		return true
+	}
+
 	var respTxn *roachpb.Transaction
 	if pErr != nil {
 		log.VEventf(ctx, 2, "heartbeat failed: %s", pErr)
@@ -459,20 +481,14 @@ func (h *txnHeartbeat) heartbeat(ctx context.Context) bool {
 		// then we ignore the error. This is possible if the heartbeat loop was
 		// started before a BeginTxn request succeeds because of ambiguity in the
 		// first write request's response.
+		//
+		// TODO(nvanbenschoten): Remove this in 2.3.
 		if tse, ok := pErr.GetDetail().(*roachpb.TransactionStatusError); ok &&
 			tse.Reason == roachpb.TransactionStatusError_REASON_TXN_NOT_FOUND {
 			return true
 		}
 
-		if pErr.GetTxn() != nil {
-			// It is not expected for a 2.1 node to return an error with a transaction
-			// in it. For one, heartbeats are not supposed to return
-			// TransactionAbortedErrors.
-			// TODO(andrei): Remove this in 2.2.
-			respTxn = pErr.GetTxn()
-		} else {
-			return true
-		}
+		respTxn = pErr.GetTxn()
 	} else {
 		respTxn = br.Responses[0].GetInner().(*roachpb.HeartbeatTxnResponse).Txn
 	}
@@ -494,12 +510,6 @@ func (h *txnHeartbeat) heartbeat(ctx context.Context) bool {
 // abortTxnAsyncLocked send an EndTransaction(commmit=false) asynchronously.
 // The asyncAbortCallbackLocked callback is also called.
 func (h *txnHeartbeat) abortTxnAsyncLocked(ctx context.Context) {
-	// Stop the heartbeat loop if it is still running.
-	if h.mu.txnEnd != nil {
-		close(h.mu.txnEnd)
-		h.mu.txnEnd = nil
-	}
-
 	if h.mu.txn.Status != roachpb.ABORTED {
 		log.Fatalf(ctx, "abortTxnAsyncLocked called for non-aborted txn: %s", h.mu.txn)
 	}

pkg/roachpb/errors.go

@@ -407,15 +407,6 @@ func NewTransactionStatusError(msg string) *TransactionStatusError {
 	}
 }
 
-// NewTransactionNotFoundStatusError initializes a new TransactionStatusError with
-// a REASON_TXN_NOT_FOUND reason.
-func NewTransactionNotFoundStatusError() *TransactionStatusError {
-	return &TransactionStatusError{
-		Msg:    "txn record not found",
-		Reason: TransactionStatusError_REASON_TXN_NOT_FOUND,
-	}
-}
-
 // NewTransactionCommittedStatusError initializes a new TransactionStatusError
 // with a REASON_TXN_COMMITTED.
 func NewTransactionCommittedStatusError() *TransactionStatusError {