Merge #148534

148534: kv: add more tracing on the commit path  r=arulajmani a=arulajmani

See individual commits for details.

`@tbg` I don't imagine any of this to have made a big difference in the investigation FWIW. I went through the entire KV commit path and added tracing for bits that may be interesting -- happy to take things out if any of them seem overkill or useless.

Co-authored-by: Arul Ajmani <[email protected]>

Author: craig[bot]

pkg/kv/kvclient/kvcoord/dist_sender.go

@@ -1472,6 +1472,7 @@ func (ds *DistSender) divideAndSendParallelCommit(
 	}
 	if swapIdx == -1 {
 		// No pre-commit QueryIntents. Nothing to split.
+		log.VEvent(ctx, 3, "no pre-commit QueryIntents found, sending batch as-is")
 		return ds.divideAndSendBatchToRanges(ctx, ba, rs, isReverse, true /* withCommit */, batchIdx)
 	}
 
@@ -1504,6 +1505,7 @@ func (ds *DistSender) divideAndSendParallelCommit(
 	qiResponseCh := make(chan response, 1)
 
 	sendPreCommit := func(ctx context.Context) {
+		log.VEvent(ctx, 3, "sending split out pre-commit QueryIntent batch")
 		// Map response index to the original un-swapped batch index.
 		// Remember that we moved the last QueryIntent in this batch
 		// from swapIdx to the end.
@@ -1562,7 +1564,9 @@ func (ds *DistSender) divideAndSendParallelCommit(
 
 	// Wait for the QueryIntent-only batch to complete and stitch
 	// the responses together.
+	log.VEvent(ctx, 3, "waiting for pre-commit QueryIntent batch response")
 	qiReply := <-qiResponseCh
+	log.VEventf(ctx, 3, "received pre-commit QueryIntent batch response")
 
 	// Handle error conditions.
 	if pErr != nil {

pkg/kv/kvclient/kvcoord/txn_interceptor_committer.go

@@ -195,6 +195,7 @@ func (tc *txnCommitter) SendLocked(
 		// so interceptors above the txnCommitter in the stack don't need to be
 		// made aware that the record is staging.
 		pErr = maybeRemoveStagingStatusInErr(pErr)
+		log.VEventf(ctx, 2, "batch with EndTxn(commit=true) failed: %v", pErr)
 		return nil, pErr
 	}
 
@@ -216,6 +217,7 @@ func (tc *txnCommitter) SendLocked(
 		// the EndTxn request, either because canCommitInParallel returned false
 		// or because there were no unproven in-flight writes (see txnPipeliner)
 		// and there were no writes in the batch request.
+		log.VEventf(ctx, 2, "parallel commit attempt for transaction %s resulted in explicit commit", br.Txn)
 		return br, nil
 	default:
 		return nil, kvpb.NewErrorf("unexpected response status without error: %v", br.Txn)
@@ -296,6 +298,8 @@ func (tc *txnCommitter) validateEndTxnBatch(ba *kvpb.BatchRequest) error {
 func (tc *txnCommitter) sendLockedWithElidedEndTxn(
 	ctx context.Context, ba *kvpb.BatchRequest, et *kvpb.EndTxnRequest,
 ) (br *kvpb.BatchResponse, pErr *kvpb.Error) {
+	log.VEventf(ctx, 2, "eliding EndTxn request for read-only, non-locking transaction")
+
 	// Send the batch without its final request, which we know to be the EndTxn
 	// request that we're eliding. If this would result in us sending an empty
 	// batch, mock out a reply instead of sending anything.

pkg/kv/kvpb/api.proto

@@ -1027,18 +1027,24 @@ message EndTxnResponse {
   // The commit timestamp of the STAGING transaction record written
   // by the request. Only set if the transaction record was staged.
   util.hlc.Timestamp staging_timestamp = 5 [(gogoproto.nullable) = false];
-  // ReplicatedLocksReleasedOnCommit, if non-empty, indicate that replicated
-  // locks with strength Shared or Exclusive were released in the referenced key
-  // spans when committing this transaction. Notably, this field is left unset
-  // if only write intents were resolved. The field is also left unset for
-  // transactions that aborted.
+  // ReplicatedLocalLocksReleasedOnCommit, if non-empty, indicate that
+  // replicated locks with strength Shared or Exclusive were released in the
+  // referenced key spans when committing this transaction. These locks are
+  // local to the range on which the EndTxn request evaluated. Notably, this
+  // field is left unset if only write intents were resolved. The field is only
+  // set when transactions are explicitly marked as committed.
   //
   // The caller must bump the timestamp cache across these spans to the
-  // transaction's commit timestamp. Doing so ensures that the released locks
-  // (acquired by the now committed transaction) continue to provide protection
-  // against other writers up to the commit timestamp, even after the locks have
-  // been released.
-  repeated Span replicated_locks_released_on_commit = 6 [(gogoproto.nullable) = false];
+  // transaction's commit timestamp. Doing so ensures that the released local[1]
+  // locks (acquired by the now committed transaction) continue to provide
+  // protection against other writers up to the commit timestamp, even after the
+  // locks have been released.
+  //
+  // [1] Non-local replicated locks provide the same protection, however, the
+  // mechanism of bumping the timestamp cache is different there. See the
+  // ReplicatedLocksReleasedCommitTimestamp field on
+  // ResolveIntent{,Range}Response.
+  repeated Span replicated_local_locks_released_on_commit = 6 [(gogoproto.nullable) = false];
 }
 
 // An AdminSplitRequest is the argument to the AdminSplit() method. The

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -273,6 +273,9 @@ func EndTxn(
 
 	// Fetch existing transaction.
 	var existingTxn roachpb.Transaction
+	log.VEventf(
+		ctx, 2, "checking to see if transaction record already exists for txn: %s", h.Txn,
+	)
 	recordAlreadyExisted, err := storage.MVCCGetProto(
 		ctx, readWriter, key, hlc.Timestamp{}, &existingTxn, storage.MVCCGetOptions{
 			ReadCategory: fs.BatchEvalReadCategory,
@@ -281,6 +284,7 @@ func EndTxn(
 	if err != nil {
 		return result.Result{}, err
 	} else if !recordAlreadyExisted {
+		log.VEvent(ctx, 2, "no existing txn record found")
 		// No existing transaction record was found - create one by writing it
 		// below in updateFinalizedTxn.
 		reply.Txn = h.Txn.Clone()
@@ -290,10 +294,12 @@ func EndTxn(
 		// an aborted txn record.
 		if args.Commit {
 			if err := CanCreateTxnRecord(ctx, cArgs.EvalCtx, reply.Txn); err != nil {
+				log.VEventf(ctx, 2, "cannot create transaction record: %v", err)
 				return result.Result{}, err
 			}
 		}
 	} else {
+		log.VEventf(ctx, 2, "existing transaction record found: %s", existingTxn)
 		// We're using existingTxn on the reply, although it can be stale
 		// compared to the Transaction in the request (e.g. the Sequence,
 		// and various timestamps). We must be careful to update it with the
@@ -317,8 +323,11 @@ func EndTxn(
 				"already committed")
 
 		case roachpb.ABORTED:
+			// The transaction has already been aborted by someone else.
+			log.VEventf(
+				ctx, 2, "transaction %s found to have be already aborted (by someone else)", reply.Txn,
+			)
 			if !args.Commit {
-				// The transaction has already been aborted by other.
 				// Do not return TransactionAbortedError since the client anyway
 				// wanted to abort the transaction.
 				resolvedLocks, _, externalLocks, err := resolveLocalLocks(ctx, readWriter, cArgs.EvalCtx, ms, args, reply.Txn)
@@ -379,6 +388,7 @@ func EndTxn(
 				// not consider the transaction to be performing a parallel commit and
 				// potentially already implicitly committed because we know that the
 				// transaction restarted since entering the STAGING state.
+				log.VEventf(ctx, 2, "request with newer epoch %d than STAGING txn record; parallel commit must have failed", h.Txn.Epoch)
 				reply.Txn.Status = roachpb.PENDING
 			default:
 				panic("unreachable")
@@ -543,13 +553,18 @@ func EndTxn(
 	txnResult.Local.ResolvedLocks = resolvedLocks
 
 	if reply.Txn.Status == roachpb.COMMITTED {
-		// Return whether replicated {shared, exclusive} locks were released by
-		// the committing transaction. If such locks were released, we still
-		// need to make sure other transactions can't write underneath the
-		// transaction's commit timestamp to the key spans previously protected
-		// by the locks. We return the spans on the response and update the
-		// timestamp cache a few layers above to ensure this.
-		reply.ReplicatedLocksReleasedOnCommit = releasedReplLocks
+		if len(releasedReplLocks) != 0 {
+			// Return that local replicated {shared, exclusive} locks were released by
+			// the committing transaction. If such locks were released, we still need
+			// to make sure other transactions can't write underneath the
+			// transaction's commit timestamp to the key spans previously protected by
+			// the locks. We return the spans on the response and update the timestamp
+			// cache a few layers above to ensure this.
+			reply.ReplicatedLocalLocksReleasedOnCommit = releasedReplLocks
+			log.VEventf(
+				ctx, 2, "committed transaction released local replicated shared/exclusive locks",
+			)
+		}
 
 		// Run the commit triggers if successfully committed.
 		triggerResult, err := RunCommitTrigger(

pkg/kv/kvserver/replica_tscache.go

@@ -170,12 +170,12 @@ func (r *Replica) updateTimestampCache(
 			// transaction's MinTimestamp, which is consulted in CanCreateTxnRecord.
 			key := transactionTombstoneMarker(start, txnID)
 			addToTSCache(key, nil, ts, txnID)
-			// Additionally, EndTxn requests that release replicated locks for
-			// committed transactions bump the timestamp cache over those lock
-			// spans to the commit timestamp of the transaction to ensure that
-			// the released locks continue to provide protection against writes
-			// underneath the transaction's commit timestamp.
-			for _, sp := range resp.(*kvpb.EndTxnResponse).ReplicatedLocksReleasedOnCommit {
+			// Additionally, EndTxn requests that release local replicated locks for
+			// committed transactions bump the timestamp cache over those lock spans
+			// to the commit timestamp of the transaction to ensure that the released
+			// locks continue to provide protection against writes underneath the
+			// transaction's commit timestamp.
+			for _, sp := range resp.(*kvpb.EndTxnResponse).ReplicatedLocalLocksReleasedOnCommit {
 				addToTSCache(sp.Key, sp.EndKey, br.Txn.WriteTimestamp, txnID)
 			}
 		case *kvpb.HeartbeatTxnRequest: