Merge pull request #151956 from cockroachdb/blathers/backport-release-25.3-151923

release-25.3: kvserver: deflake TestTxnReadWithinUncertaintyIntervalAfterLeaseTransfer

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))