closedts: add the pacing to the closed timestamp lead policy calculation

This commit adds the pacing interval to the closed timestamp lead policy
calculation, this is important because we pace side transport messages,
and that means that some nodes might not receive the closed timestamp
update by an extra duration (the pacing interval).

Fixes: #148211

Release note: None

Author: iskettaneh

pkg/kv/kvserver/closedts/policy.go

@@ -22,7 +22,10 @@ const (
 // global reads. It accounts for network latency, clock offset, and both Raft
 // and side-transport propagation delays.
 func computeLeadTimeForGlobalReads(
-	networkRTT time.Duration, maxClockOffset time.Duration, sideTransportCloseInterval time.Duration,
+	networkRTT time.Duration,
+	maxClockOffset time.Duration,
+	sideTransportCloseInterval time.Duration,
+	sideTransportPacingInterval time.Duration,
 ) time.Duration {
 	// The LEAD_FOR_GLOBAL_READS calculation is more complex. Instead of the
 	// policy defining an offset from the publisher's perspective, the
@@ -61,10 +64,13 @@ func computeLeadTimeForGlobalReads(
 	//  # the worst-case.
 	//  side_propagation_time = max_network_rtt * 0.5 + side_transport_close_interval
 	//
+	//  # We pace sending side-transport updates up to side_transport_pacing_refresh_interval.
+	//  # We have to also include it into our calculation.
+	//
 	//  # Combine, we get the following result
 	//  closed_ts_at_sender = now + max_offset + max(
 	//    max_network_rtt * 1.5 + raft_overhead,
-	//    max_network_rtt * 0.5 + side_transport_close_interval,
+	//    max_network_rtt * 0.5 + side_transport_close_interval + side_transport_pacing_refresh_interval,
 	//  )
 	//
 	// By default, this leads to a closed timestamp target that leads the
@@ -86,7 +92,7 @@ func computeLeadTimeForGlobalReads(
 	raftTransportPropTime := (networkRTT*3)/2 + raftTransportOverhead
 
 	// See side_propagation_time.
-	sideTransportPropTime := networkRTT/2 + sideTransportCloseInterval
+	sideTransportPropTime := networkRTT/2 + sideTransportCloseInterval + sideTransportPacingInterval
 
 	// See propagation_time.
 	maxTransportPropTime := max(sideTransportPropTime, raftTransportPropTime)
@@ -106,6 +112,7 @@ func TargetForPolicy(
 	lagTargetDuration time.Duration,
 	leadTargetOverride time.Duration,
 	sideTransportCloseInterval time.Duration,
+	sideTransportPacingInterval time.Duration,
 	policy ctpb.RangeClosedTimestampPolicy,
 ) hlc.Timestamp {
 	var targetOffsetTime time.Duration
@@ -121,7 +128,7 @@ func TargetForPolicy(
 			break
 		}
 		targetOffsetTime = computeLeadTimeForGlobalReads(computeNetworkRTTBasedOnPolicy(policy),
-			maxClockOffset, sideTransportCloseInterval)
+			maxClockOffset, sideTransportCloseInterval, sideTransportPacingInterval)
 	default:
 		panic("unexpected RangeClosedTimestampPolicy")
 	}

pkg/kv/kvserver/closedts/policy_test.go

@@ -28,11 +28,12 @@ func TestTargetForPolicy(t *testing.T) {
 	maxClockOffset := millis(500)
 
 	for _, tc := range []struct {
-		lagTargetNanos             time.Duration
-		leadTargetOverride         time.Duration
-		sideTransportCloseInterval time.Duration
-		rangePolicy                ctpb.RangeClosedTimestampPolicy
-		expClosedTSTarget          hlc.Timestamp
+		lagTargetNanos              time.Duration
+		leadTargetOverride          time.Duration
+		sideTransportCloseInterval  time.Duration
+		sideTransportPacingInterval time.Duration
+		rangePolicy                 ctpb.RangeClosedTimestampPolicy
+		expClosedTSTarget           hlc.Timestamp
 	}{
 		{
 			lagTargetNanos:    secs(3),
@@ -52,6 +53,16 @@ func TestTargetForPolicy(t *testing.T) {
 					millis(275) /* sideTransportPropTime */ +
 					millis(25) /* bufferTime */).Nanoseconds(), 0),
 		},
+		{
+			sideTransportCloseInterval:  millis(200),
+			sideTransportPacingInterval: millis(10),
+			rangePolicy:                 ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
+			expClosedTSTarget: now.
+				Add((maxClockOffset +
+					millis(275) /* sideTransportPropTime */ +
+					millis(10) /* sideTransportPacing */ +
+					millis(25) /* bufferTime */).Nanoseconds(), 0),
+		},
 		{
 			sideTransportCloseInterval: millis(50),
 			rangePolicy:                ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
@@ -61,10 +72,11 @@ func TestTargetForPolicy(t *testing.T) {
 					millis(25) /* bufferTime */).Nanoseconds(), 0),
 		},
 		{
-			leadTargetOverride:         millis(1234),
-			sideTransportCloseInterval: millis(200),
-			rangePolicy:                ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
-			expClosedTSTarget:          now.Add(millis(1234).Nanoseconds(), 0),
+			leadTargetOverride:          millis(1234),
+			sideTransportCloseInterval:  millis(200),
+			sideTransportPacingInterval: millis(10),
+			rangePolicy:                 ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
+			expClosedTSTarget:           now.Add(millis(1234).Nanoseconds(), 0),
 		},
 	} {
 		t.Run("", func(t *testing.T) {
@@ -74,6 +86,7 @@ func TestTargetForPolicy(t *testing.T) {
 				tc.lagTargetNanos,
 				tc.leadTargetOverride,
 				tc.sideTransportCloseInterval,
+				tc.sideTransportPacingInterval,
 				tc.rangePolicy,
 			)
 			require.Equal(t, tc.expClosedTSTarget, target)

pkg/kv/kvserver/closedts/sidetransport/sender.go

@@ -368,13 +368,15 @@ func (s *Sender) publish(ctx context.Context) hlc.ClockTimestamp {
 	lagTargetDuration := closedts.TargetDuration.Get(&s.st.SV)
 	leadTargetOverride := closedts.LeadForGlobalReadsOverride.Get(&s.st.SV)
 	sideTransportCloseInterval := closedts.SideTransportCloseInterval.Get(&s.st.SV)
+	sideTransportPacingInterval := closedts.SideTransportPacingRefreshInterval.Get(&s.st.SV)
 	for pol := ctpb.RangeClosedTimestampPolicy(0); pol < ctpb.RangeClosedTimestampPolicy(numPolicies); pol++ {
 		target := closedts.TargetForPolicy(
 			now,
 			maxClockOffset,
 			lagTargetDuration,
 			leadTargetOverride,
 			sideTransportCloseInterval,
+			sideTransportPacingInterval,
 			pol,
 		)
 		s.trackedMu.lastClosed[pol] = target

pkg/kv/kvserver/closedts/sidetransport/sender_test.go

@@ -190,6 +190,7 @@ func expGroupUpdates(s *Sender, now hlc.ClockTimestamp) []ctpb.Update_GroupUpdat
 			closedts.TargetDuration.Get(&s.st.SV),
 			closedts.LeadForGlobalReadsOverride.Get(&s.st.SV),
 			closedts.SideTransportCloseInterval.Get(&s.st.SV),
+			closedts.SideTransportPacingRefreshInterval.Get(&s.st.SV),
 			pol,
 		)
 	}

pkg/kv/kvserver/replica_closedts.go

@@ -155,6 +155,7 @@ func (r *Replica) closedTimestampTargetRLocked() hlc.Timestamp {
 		closedts.TargetDuration.Get(&r.ClusterSettings().SV),
 		closedts.LeadForGlobalReadsOverride.Get(&r.ClusterSettings().SV),
 		closedts.SideTransportCloseInterval.Get(&r.ClusterSettings().SV),
+		closedts.SideTransportPacingRefreshInterval.Get(&r.ClusterSettings().SV),
 		closedTimestampPolicy(r.descRLocked(), *r.cachedClosedTimestampPolicy.Load()),
 	)
 }

pkg/kv/kvserver/replica_proposal_buf_test.go

@@ -174,6 +174,7 @@ func (t *testProposer) closedTimestampTarget() hlc.Timestamp {
 		1*time.Second,
 		0,
 		200*time.Millisecond,
+		10*time.Millisecond,
 		t.rangePolicy,
 	)
 }
@@ -921,6 +922,7 @@ func TestProposalBufferClosedTimestamp(t *testing.T) {
 	nowMinusTwiceClosedLag := nowTS.Add(-2*closedts.TargetDuration.Get(&st.SV).Nanoseconds(), 0)
 	nowPlusGlobalReadLead := nowTS.Add((maxOffset +
 		275*time.Millisecond /* sideTransportPropTime */ +
+		10*time.Millisecond /* sideTransportPacing */ +
 		25*time.Millisecond /* bufferTime */).Nanoseconds(), 0)
 	expiredLeaseTimestamp := nowTS.Add(-1000, 0)
 	someClosedTS := nowTS.Add(-2000, 0)

pkg/kv/txn.go

@@ -969,12 +969,14 @@ func (txn *Txn) DeadlineLikelySufficient() bool {
 		lagTargetDuration := closedts.TargetDuration.Get(sv)
 		leadTargetOverride := closedts.LeadForGlobalReadsOverride.Get(sv)
 		sideTransportCloseInterval := closedts.SideTransportCloseInterval.Get(sv)
+		sideTransportPacingInterval := closedts.SideTransportPacingRefreshInterval.Get(sv)
+
 		// Pass the DefaultMaxNetworkRTT regardless of leadTargetAutoTune because we
 		// don't have a good way to estimate the network RTT here. We choose to be
 		// more conservative as this is just for an optimization if the deadline is
 		// far in the future. Missing the optimization is not a big deal.
 		return closedts.TargetForPolicy(now, maxClockOffset,
-			lagTargetDuration, leadTargetOverride, sideTransportCloseInterval,
+			lagTargetDuration, leadTargetOverride, sideTransportCloseInterval, sideTransportPacingInterval,
 			ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO).Add(int64(time.Second), 0)
 	}