kvserver: add test for transaction unexpectedly committed

This adds a unit test to reproduce the behavior described in cockroachdb#103817 and
seen in cockroachdb#67765, which currently is a bug in our implementation of the
parallel commit protocol that results in the assertion failure known as
`transaction unexpectedly committed`. The test currently validates the
incorrect behavior of the known issue, though it is inded to be used to
validate the potential fixes as proposed in cockroachdb#103817.

Release note: None

Part of: cockroachdb#103817

pkg/kv/kvpb/batch.go

@@ -258,6 +258,12 @@ func (ba *BatchRequest) IsSinglePushTxnRequest() bool {
 	return ba.isSingleRequestWithMethod(PushTxn)
 }
 
+// IsSingleRecoverTxnRequest returns true iff the batch contains a single request,
+// and that request is a RecoverTxnRequest.
+func (ba *BatchRequest) IsSingleRecoverTxnRequest() bool {
+	return ba.isSingleRequestWithMethod(RecoverTxn)
+}
+
 // IsSingleHeartbeatTxnRequest returns true iff the batch contains a single
 // request, and that request is a HeartbeatTxn.
 func (ba *BatchRequest) IsSingleHeartbeatTxnRequest() bool {
@@ -827,6 +833,15 @@ func (ba BatchRequest) SafeFormat(s redact.SafePrinter, _ rune) {
 				s.Printf(" %s", et.InternalCommitTrigger.Kind())
 			}
 			s.Printf(") [%s]", h.Key)
+		} else if rt, ok := req.(*RecoverTxnRequest); ok {
+			h := req.Header()
+			s.Printf("%s(%s, ", req.Method(), rt.Txn.Short())
+			if rt.ImplicitlyCommitted {
+				s.Printf("commit")
+			} else {
+				s.Printf("abort")
+			}
+			s.Printf(") [%s]", h.Key)
 		} else {
 			h := req.Header()
 			if req.Method() == PushTxn {

pkg/kv/kvserver/BUILD.bazel

@@ -275,6 +275,7 @@ go_test(
         "client_store_test.go",
         "client_tenant_test.go",
         "client_test.go",
+        "client_unexpected_commit_test.go",
         "closed_timestamp_test.go",
         "consistency_queue_test.go",
         "debug_print_test.go",
@@ -521,7 +522,9 @@ go_test(
         "@io_etcd_go_raft_v3//raftpb",
         "@io_etcd_go_raft_v3//tracker",
         "@org_golang_google_grpc//:go_default_library",
+        "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//metadata",
+        "@org_golang_google_grpc//status",
         "@org_golang_x_sync//errgroup",
         "@org_golang_x_sync//syncmap",
         "@org_golang_x_time//rate",

pkg/kv/kvserver/batcheval/cmd_lease_request.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 )
 
 func init() {
@@ -159,6 +160,7 @@ func RequestLease(
 		priorReadSum = &worstCaseSum
 	}
 
+	log.VEventf(ctx, 2, "lease request: prev lease: %+v, new lease: %+v", prevLease, newLease)
 	return evalNewLease(ctx, cArgs.EvalCtx, readWriter, cArgs.Stats,
 		newLease, prevLease, priorReadSum, isExtension, false /* isTransfer */)
 }