Great work here! 🚀

Reminder that we could extend policy_refresher_test.go as well with a mocked out latency map. But let's do that in a followup PR.

Reviewed 4 of 11 files at r1, 15 of 15 files at r2, 11 of 11 files at r3.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @wenyihu6)

-- commits line 16 at r3:
nit: "by the time that all follower replicas have received this closed timestamp, it's greater than the present time, so they can serve reads without redirecting to the leaseholder".

-- commits line 20 at r3:
"we fail to bump closed timestamps" begs a few questions -- why would this happen? maybe we could just drop this line.

-- commits line 36 at r3:
nit: a more succinct way to put this would be "higher lead time leads to high commit wait times, which inturn impacts write latency"

-- commits line 39 at r3:
"followers not fully replicating the changes" sounds off. Should this instead say "a closed timestamp that lags present time, causing follower replicas to redirect..."

-- commits line 48 at r3:
Is this stale? I would imagine this bit would be subsumed by the PolicyRefresher bit in the design.

-- commits line 63 at r3:
"the policy refresher"

pkg/kv/kvserver/closedts/policy_calculation.go line 19 at r2 (raw file):

// network RTT. Returns default max RTT for no latency info, or midpoint RTT for
// known latency buckets. Panics on unknown policy.
func computeNetworkRTTBasedOnPolicy(policy ctpb.RangeClosedTimestampPolicy) time.Duration {

Should we add a test for this?

pkg/kv/kvserver/closedts/policy_calculation.go line 17 at r3 (raw file):

// FindBucketBasedOnNetworkRTT maps a network RTT to a closed timestamp policy bucket.
// It divides RTT by policy interval, adds 1 for zero-based indexing, and offsets by

nit: the details about the computation don't need to live on the function level comment; instead, they're better placed closed to the logic.

pkg/kv/kvserver/replica.go line 1342 at r3 (raw file):

// RefreshPolicy updates the replica's cached closed timestamp policy based on
// zone configurations and provided node latencies.

"span configurations" at this level.

pkg/kv/kvserver/replica.go line 1346 at r3 (raw file):

// For ranges serving global reads,
// 1. If the provided map is nil, the policy will be
// LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO. The latency will be hardcoded to

nit: at this layer, because we're just refreshing the policy, I think this commentary will be clearer if we didn't add words about what latency is inferred from the policy.

If you want to add it, maybe move that bit of the commentary inside the function where we're assigning policies instead

pkg/kv/kvserver/replica.go line 1376 at r3 (raw file):

			}
		}
		// FindBucketBasedOnNetworkRTT returns LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO

This commentary is misplaced -- should we move it to FindBucketBasedOnNetworkRTT instead?

pkg/kv/kvserver/closedts/policy.go line 17 at r2 (raw file):

const (
	defaultMaxNetworkRTT                 = 150 * time.Millisecond
	closedTimestampPolicyLatencyInterval = 20 * time.Millisecond

How do you feel about renaming this to "closedTimestampPolicyBucketWidth"?

pkg/kv/kvserver/replica_closedts.go line 25 at r3 (raw file):

// timestamp. Falls back to LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO if the
// range's policy is not found in targetByPolicy. This fallback handles race
// conditions where the policy refresher observes cluster version upgrades

Could you say more about this cluster version upgrades race?

pkg/kv/kvserver/replica_closedts.go line 35 at r3 (raw file):

		return policy, target
	}
	if policy >= ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_20MS &&

It may be worth a comment explaining how this happens.

pkg/kv/kvserver/closedts/policy_calculation_test.go line 20 at r3 (raw file):

// TestNetworkRTTAndPolicyCalculations tests the conversion between network
// RTT and closed timestamp policy.
func TestNetworkRTTAndPolicyCalculations(t *testing.T) {

Nice test!

pkg/kv/kvserver/closedts/policy_calculation_test.go line 61 at r3 (raw file):

		},
		{
			name:           "low-end bucket boundary",

Related to my comment on r1, let's add test cases for mid/low/high for each of the buckets. Mostly to serve as regression tests.

pkg/kv/kvserver/closedts/sidetransport/sender.go line 310 at r3 (raw file):

}

// If the cluster is not fully upgraded, only look at closed timestamps policies

nit: consider moving this commentary closer to the logic below.

For here, maybe something like: // maxClosedTimestampPolicy returns the distinct number of closed timestamp policies for which the side transport should send updates.

pkg/kv/kvserver/closedts/sidetransport/sender.go line 317 at r3 (raw file):

// kv.closed_timestamp.policy_refresh_interval is 0.
func (s *Sender) maxClosedTimestampPolicy() int {
	// Even if no auto-tuning is disabled, still send them. No ranges would use

nit: if you take my comment above, this could be just deleted, as it's subsumed by the commentary above.

pkg/kv/kvserver/closedts/sidetransport/sender.go line 441 at r3 (raw file):

			// LAI.
			needExplicit = true
		} else if lastMsg.policy != closeRes.Policy {

Around version upgrades, is it possible that the policy has been refreshed but by the range but we're not sending an update for latency buckets (presumably because maxClosedTimestampPolicy above was called when the cluster version had not been bumped)?

At the least, it would be worth trying to add a test that reproduces this race to see how things behave.

pkg/kv/kvserver/closedts/ctpb/service.proto line 42 at r2 (raw file):

  // hardcoded network latency 150ms by default.
  LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO = 1;
  // Lead policy for ranges with global reads policy. Following policies are

nit: "The following policies"

pkg/kv/kvserver/closedts/sidetransport/sender_test.go line 392 at r3 (raw file):

// TestSenderWithLatencyTracker verifies that the sender correctly updates
// closed timestamp policies based on network latency between nodes.
func TestSenderWithLatencyTracker(t *testing.T) {

Nice test! 💯

pkg/ccl/multiregionccl/multiregion_test.go line 164 at r3 (raw file):

// This test verifies closed timestamp policy behavior with and without the
// policy refresher. With the refresher enabled, we expect latency-based

Do you mean autotuning? The policy refresher should be running regardless, right?

Sounds good, I added a new item in #143888 to track this. Thanks for the nice rewording as always!

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @arulajmani)

-- commits line 16 at r3:

Done.

-- commits line 20 at r3:

Removed for clarity. This may happen due to any of the reasons here - https://github.com/cockroachdb/cockroach/blob/872849938a5ab0c96d39bdd9274ea4897b2947e2/pkg/kv/kvserver/closedts/sidetransport/sender.go#L188-L198. In those cases, those ranges would be removed from the side transport ranges map.

-- commits line 36 at r3:

Done.

-- commits line 39 at r3:

Done.

-- commits line 48 at r3:

Yes, corrected now.

-- commits line 63 at r3:

Done.

pkg/kv/kvserver/replica.go line 1342 at r3 (raw file):

Done.

pkg/kv/kvserver/replica.go line 1346 at r3 (raw file):

Moved the comment closer to the logic below.

pkg/kv/kvserver/replica.go line 1376 at r3 (raw file):

Moved to FindBucketBasedOnNetworkRTT instead.

pkg/kv/kvserver/replica_closedts.go line 25 at r3 (raw file):

I added more details to the race condition I was worried about. I think it is exactly what you described here - https://reviewable.io/reviews/cockroachdb/cockroach/142721#-ONKsU7WCKoy8IrrX8hq.

pkg/kv/kvserver/replica_closedts.go line 35 at r3 (raw file):

Done.

pkg/kv/kvserver/closedts/policy.go line 17 at r2 (raw file):

I like it, done.

pkg/kv/kvserver/closedts/policy_calculation.go line 19 at r2 (raw file):

I initially added tests for them in the second commit. I've now moved the computeNetworkRTTBasedOnPolicy test to the first commit and rebased.

pkg/kv/kvserver/closedts/policy_calculation.go line 17 at r3 (raw file):

Done.

pkg/kv/kvserver/closedts/ctpb/service.proto line 42 at r2 (raw file):

Done.

pkg/kv/kvserver/closedts/sidetransport/sender.go line 310 at r3 (raw file):

Done.

pkg/kv/kvserver/closedts/sidetransport/sender.go line 317 at r3 (raw file):

Removed.

pkg/kv/kvserver/closedts/sidetransport/sender.go line 441 at r3 (raw file):

Sg, added another item for testing in #143888. I think it is possible, and this is what I was trying to guard against in

pkg/ccl/multiregionccl/multiregion_test.go line 164 at r3 (raw file):

Done.

pkg/kv/kvserver/closedts/policy_calculation_test.go line 61 at r3 (raw file):

Done.

pkg/kv/kvserver/closedts/sidetransport/sender_test.go line 392 at r3 (raw file):

Thanks 😃

modulo some nits and one comment about how we want to handle the case where latency information isn't available for a peer. We should add a test for that case in this patch before merging, but this is very close.

Reviewed 2 of 13 files at r5, 1 of 11 files at r6, 10 of 10 files at r9, 10 of 10 files at r10, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @wenyihu6)

pkg/kv/kvserver/closedts/policy_calculation.go line 19 at r2 (raw file):

Great, thanks!

pkg/kv/kvserver/closedts/policy_calculation.go line 17 at r3 (raw file):

nit: Consider moving everything after the first sentence inline. So this would look something like:

// computeNetworkRTTBasedOnPolicy returns the network RTT based on the closed
// timestamp policy.
func computeNetworkRTTBasedOnPolicy(policy ctpb.RangeClosedTimestampPolicy) time.Duration {
	switch {
	case policy == ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO:
		// If no latency info is available, return the default max RTT.
		return defaultMaxNetworkRTT
	case policy >= ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_20MS &&
		policy <= ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_EQUAL_OR_GREATER_THAN_300MS:
		// For known latency buckets, we return the midpoint RTT for the bucket.
		// The midpointRTT for bucket N is (N+0.5)*interval.
		bucket := int(policy) - int(ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO) - 1
		return time.Duration((float64(bucket) + 0.5) * float64(closedTimestampPolicyBucketWidth.Nanoseconds()))
	default:
		panic(fmt.Sprintf("unknown closed timestamp policy: %s", policy))
	}
}

The idea being that a casual reader doesn't need to know about these intricate details.

-- commits line 48 at r10:
"adds newly added" seems redundant. Should we just say "adds"?

pkg/kv/kvserver/closedts/policy_calculation.go line 19 at r10 (raw file):

// bucket.
func FindBucketBasedOnNetworkRTT(networkRTT time.Duration) ctpb.RangeClosedTimestampPolicy {
	// If maxLatency is negative (i.e. no peer latency is provided), It returns

nit: s/It returns/return/g

pkg/kv/kvserver/closedts/policy_calculation.go line 27 at r10 (raw file):

		return ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_EQUAL_OR_GREATER_THAN_300MS
	}
	// It divides RTT by policy interval, adds 1 for zero-based indexing, and

nit: let's reword this comment as well, similar to other suggestions, now that we've inlined it, so as to not use "It" at the start.

pkg/kv/kvserver/replica_closedts.go line 32 at r10 (raw file):

		return policy, target
	}
	// If the range's policy is not found in targetByPolicy, it falls back to

nit: s/it falls back/fall back/g

pkg/kv/kvserver/replica_closedts.go line 37 at r10 (raw file):

	// This fallback handles race conditions between the policy refresher and side
	// transport during cluster version upgrades. The race can occur in this
	// sequence: 1. Side transport prepares a policy map before cluster upgrade is

nit: should we have these bullets be on separate lines?

pkg/kv/kvserver/replica_closedts.go line 44 at r10 (raw file):

	//
	// In this case, we temporarily fall back to the no-latency policy until the
	// side transport catches up with the new cluster version.

Would you mind adding a TODO to remove this once version compatibility with 25.1 is no longer a concern? At that point, maybe we could refactor some of this because we could expand the assertion below to panic if ok == false in all cases.

pkg/kv/kvserver/replica.go line 1364 at r10 (raw file):

		}

		// For ranges serving global reads, it determines the maximum latency

nit: "it determines" sounds off when inlined. Should we just say "For ranges serving global reads, determine the maximum leaseholder-to-peer replica using the provided map and set an appropriate policy bucket. This then controls how far in the future timestamps will be closed for the range."

pkg/kv/kvserver/replica.go line 1371 at r10 (raw file):

		maxLatency := time.Duration(-1)
		for _, peer := range r.shMu.state.Desc.InternalReplicas {
			if peerLatency, ok := latencies[peer.NodeID]; ok {

If a peer isn't in the map, should we be instead using defaultMaxNetworkRTT instead of skipping it? Separately, should we log (maybe a log.Every) if that's the case?

Addressed your comment and added a new test - TestRefreshPolicyWithVariousLatencies.

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @arulajmani)

-- commits line 48 at r10:

Done.

pkg/kv/kvserver/replica.go line 1364 at r10 (raw file):

Done.

pkg/kv/kvserver/replica.go line 1371 at r10 (raw file):

Using defaultMaxNetworkRTT for peer replica latency if it is missing now. I made a note about the logs in #143890 which I plan to address as we add more metrics for this work.

pkg/kv/kvserver/replica_closedts.go line 32 at r10 (raw file):

Done.

pkg/kv/kvserver/replica_closedts.go line 37 at r10 (raw file):

Done.

pkg/kv/kvserver/replica_closedts.go line 44 at r10 (raw file):

Done.

pkg/kv/kvserver/closedts/policy_calculation.go line 17 at r3 (raw file):

Done.

pkg/kv/kvserver/closedts/policy_calculation.go line 19 at r10 (raw file):

Done.

pkg/kv/kvserver/closedts/policy_calculation.go line 27 at r10 (raw file):

Done.

Let's ship it

Reviewed 15 of 15 files at r11, 1 of 1 files at r12, 6 of 6 files at r13, 12 of 12 files at r14, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @wenyihu6)

pkg/kv/kvserver/replica.go line 1371 at r10 (raw file):

Adding the logs later sounds good!

pkg/kv/kvserver/replica_closedts_test.go line 1157 at r14 (raw file):

			latencies: map[roachpb.NodeID]time.Duration{
				tc.Target(0).NodeID: 10 * time.Millisecond,
				// tc.Target(1,2).NodeID is missing, simulating nil.

Im not sure what you mean by "simulating nil". Would it be clearer if you just said // tc.Target(1, 2).NodeID are missing.

Here, and elsehwere.

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @arulajmani)

pkg/kv/kvserver/replica_closedts_test.go line 1157 at r14 (raw file):

Removed.