kvcoord: deflake TestTxnCoordSenderRetriesAcrossEndTxn

The first subtest asserted ExpectPusheeTxnRecordNotFound, which requires
the push to find no txn record at all. However, the heartbeat loop can
race and write a PENDING txn record during CommitInBatch execution,
causing the push to find that record and fail the assertion.

Replace with the new ExpectNoTxnRecovery expectation, which verifies the
actual invariant: no STAGING record was found and no transaction recovery
was performed. This tolerates a PENDING record from the heartbeat loop
while still checking that the txn was not implicitly committed.

Fixes: #170829
Release note: None

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pav-kv

pkg/kv/kvclient/kvcoord/dist_sender_server_test.go

@@ -3962,9 +3962,11 @@ func TestTxnCoordSenderRetriesAcrossEndTxn(t *testing.T) {
 			sideRejectedOnSecondAttempt: left,
 			// The first attempt of right side contains a parallel commit (i.e. an
 			// EndTxn), but fails. The 2nd attempt of the right side will no longer
-			// contain an EndTxn, as explained above. So we expect the txn record to
-			// not exist.
-			txnRecExpectation: kvclientutils.ExpectPusheeTxnRecordNotFound,
+			// contain an EndTxn, as explained above. So we expect no STAGING txn
+			// record / no transaction recovery. Note that a PENDING txn record may
+			// or may not exist depending on whether the heartbeat loop managed to
+			// fire before the CommitInBatch completed.
+			txnRecExpectation: kvclientutils.ExpectNoTxnRecovery,
 		},
 		{
 			// On the first attempt, the right side succeed in writing a STAGING txn

pkg/testutils/kvclientutils/txn_recovery.go

@@ -28,6 +28,11 @@ const (
 	// ExpectPusheeTxnRecordNotFound means we're expecting the push to not find the
 	// pushee txn record.
 	ExpectPusheeTxnRecordNotFound
+	// ExpectNoTxnRecovery means we're expecting the push to NOT perform
+	// transaction recovery. This is weaker than ExpectPusheeTxnRecordNotFound: it
+	// tolerates a PENDING txn record (e.g. written by the heartbeat loop) but
+	// verifies no STAGING record was found and recovered.
+	ExpectNoTxnRecovery
 	// DontExpectAnything means we're not going to check the state in which the
 	// pusher found the pushee's txn record.
 	DontExpectAnything
@@ -114,6 +119,13 @@ func CheckPushResult(
 				"push didn't run as expected (missing \"%s\"). recording: %s",
 				expMsg, recording)
 		}
+	case ExpectNoTxnRecovery:
+		unexpectedMsg := fmt.Sprintf("recovered txn %s", txn.ID.Short())
+		if _, ok := recording.FindLogMessage(unexpectedMsg); ok {
+			resolutionErr = errors.Errorf(
+				"push unexpectedly performed txn recovery (found \"%s\"). recording: %s",
+				unexpectedMsg, recording)
+		}
 	case DontExpectAnything:
 	}