kvserver: deflake TestTxnReadWithinUncertaintyIntervalAfterLeaseTransfer

The test failed in the past, and it seems that there is a test race condition. This
commit adds more logging and increases the verbosity of the test so hopefully we
understand better what happens next time.

One failure that was captures was that clocks didn't become in sync at the
beginning of the test, probably due to a message between the nodes affecting
the sync. This commit also keeps trying to sync the clocks at the beginning of the
test.

Fixes: #151859

Release note: None

Author: iskettaneh

pkg/kv/kvserver/client_replica_test.go

@@ -777,6 +777,8 @@ func TestTxnReadWithinUncertaintyIntervalAfterLeaseTransfer(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	// Increase the verbosity of the test to help investigate failures.
+	require.NoError(t, log.SetVModule("replica_range_lease=3,raft=4,txn=3,txn_coord_sender=3"))
 	const numNodes = 2
 	var manuals []*hlc.HybridManualClock
 	var clocks []*hlc.Clock
@@ -804,42 +806,69 @@ func TestTxnReadWithinUncertaintyIntervalAfterLeaseTransfer(t *testing.T) {
 	keyA, keyB := roachpb.Key("a"), roachpb.Key("b")
 	tc.SplitRangeOrFatal(t, keyA)
 	keyADesc, keyBDesc := tc.SplitRangeOrFatal(t, keyB)
+	t.Logf("split range at keyB=%s, got descriptors: keyADesc=%+v, keyBDesc=%+v",
+		keyB, keyADesc, keyBDesc)
+
 	// Place key A's sole replica on node 1 and key B's sole replica on node 2.
 	tc.AddVotersOrFatal(t, keyB, tc.Target(1))
+	t.Logf("added voter for keyB=%s to node 2", keyB)
 	tc.TransferRangeLeaseOrFatal(t, keyBDesc, tc.Target(1))
+	t.Logf("transferred lease for keyB range to node 2")
 	tc.RemoveVotersOrFatal(t, keyB, tc.Target(0))
+	t.Logf("removed voter for keyB=%s from node 1", keyB)
 
 	// Pause the servers' clocks going forward.
-	var maxNanos int64
 	for i, m := range manuals {
 		m.Pause()
-		if cur := m.UnixNano(); cur > maxNanos {
-			maxNanos = cur
-		}
 		clocks = append(clocks, tc.Servers[i].Clock())
 	}
-	// After doing so, perfectly synchronize them.
-	for _, m := range manuals {
-		m.Increment(maxNanos - m.UnixNano())
-	}
+
+	// Synchronize the clocks. We wrap this in a SucceedsSoon to avoid messages
+	// between the nodes to cause them to remain out of sync.
+	var nowN1, nowN2 hlc.Timestamp
+	testutils.SucceedsSoon(t, func() error {
+		// Find the maximum clock value.
+		maxNanos := manuals[0].UnixNano()
+		for _, m := range manuals[1:] {
+			maxNanos = max(maxNanos, m.UnixNano())
+		}
+		// After doing so, perfectly synchronize them.
+		for _, m := range manuals {
+			m.Increment(maxNanos - m.UnixNano())
+		}
+
+		nowN1, nowN2 = clocks[0].Now(), clocks[1].Now()
+		if nowN1.WallTime != nowN2.WallTime {
+			return errors.Errorf("clocks are not synchronized: n1's clock: %+v, n2's clock: %+v",
+				nowN1, nowN2)
+		}
+		return nil
+	})
+
+	t.Logf("all clocks synchronized: n1's clock: %+v, n2's clock: %+v", nowN1, nowN2)
 
 	// Create a new transaction using the second node as the gateway.
-	now := clocks[1].Now()
 	maxOffset := clocks[1].MaxOffset().Nanoseconds()
 	require.NotZero(t, maxOffset)
-	txn := roachpb.MakeTransaction("test", keyB, isolation.Serializable, 1, now, maxOffset, int32(tc.Servers[1].SQLInstanceID()), 0, false /* omitInRangefeeds */)
+	txn := roachpb.MakeTransaction("test", keyB, isolation.Serializable, 1, nowN2, maxOffset, int32(tc.Servers[1].SQLInstanceID()), 0, false /* omitInRangefeeds */)
+	t.Logf("created transaction: %+v", txn)
 	require.True(t, txn.ReadTimestamp.Less(txn.GlobalUncertaintyLimit))
 	require.Len(t, txn.ObservedTimestamps, 0)
 
 	// Collect an observed timestamp in that transaction from node 2.
+	t.Logf("collecting observed timestamp from node 2 for keyB=%s", keyB)
 	getB := getArgs(keyB)
 	resp, pErr := kv.SendWrappedWith(ctx, tc.Servers[1].DistSenderI().(kv.Sender), kvpb.Header{Txn: &txn}, getB)
 	require.Nil(t, pErr)
+	t.Logf("successfully read keyB=%s, response: %+v", keyB, resp)
 	txn.Update(resp.Header().Txn)
 	require.Len(t, txn.ObservedTimestamps, 1)
+	t.Logf("updated transaction with observed timestamp, now has %+v observed timestamps", txn.ObservedTimestamps)
 
 	// Advance the clock on the first node.
 	manuals[0].Increment(100)
+	nowN1, nowN2 = clocks[0].Now(), clocks[1].Now()
+	t.Logf("advanced n1's clock by 100ns: n1's clock: %+v, n2's clock: %+v", nowN1, nowN2)
 
 	// Perform a non-txn write on node 1. This will grab a timestamp from node 1's
 	// clock, which leads the clock on node 2.
@@ -856,15 +885,20 @@ func TestTxnReadWithinUncertaintyIntervalAfterLeaseTransfer(t *testing.T) {
 	// flakiness. For now, we just re-order the operations and assert that we
 	// receive an uncertainty error even though its absence would not be a true
 	// stale read.
+	t.Logf("Performing non-txn write on node 0 for keyA=%s", keyA)
 	ba := &kvpb.BatchRequest{}
 	ba.Add(putArgs(keyA, []byte("val")))
 	br, pErr := tc.Servers[0].DistSenderI().(kv.Sender).Send(ctx, ba)
 	require.Nil(t, pErr)
 	writeTs := br.Timestamp
+	t.Logf("Successfully wrote keyA=%s, write timestamp: %s", keyA, writeTs)
 
 	// The transaction has a read timestamp beneath the write's commit timestamp
 	// but a global uncertainty limit above the write's commit timestamp. The
 	// observed timestamp collected is also beneath the write's commit timestamp.
+	nowN1, nowN2 = clocks[0].Now(), clocks[1].Now()
+	t.Logf("validating the clocks before test assertions: n1's clock: %+v, n2's clock: %+v",
+		nowN1, nowN2)
 	assert.True(t, txn.ReadTimestamp.Less(writeTs))
 	assert.True(t, writeTs.Less(txn.GlobalUncertaintyLimit))
 	assert.True(t, txn.ObservedTimestamps[0].Timestamp.ToTimestamp().Less(writeTs))