Merge #142721

142721: kvserver/closedts: auto-tune closed ts r=arulajmani a=wenyihu6


**kvserver/closedts: add latency based closed timestamp policies**

This patch adds latency based closed timestamp policies to
ctpb.LatencyBasedRangeClosedTimestampPolicy.

Part of: #59680
Release note: none

---

**kvserver/closedts: auto-tune closed ts**

For global table ranges, we aim to ensure follower replicas can serve
present-time reads by calculating a target closed timestamp that leads the
current time (using the `LEAD_FOR_GLOBAL_READS` policy). The goal of this
estimation is to be confident that by the time that all follower have received
this closed timestamp, it's greater than the present time, so they can serve
reads without redirecting to the leaseholder. Note that this is an estimated
goal, not the exact lead time. It is still possible that some follower nodes are
really behind.

Calculation:

```
raft_propagation_time = max_network_rtt*1.5 + raft_overhead(20ms)
side_propagation_time = max_network_rtt*0.5 + side_transport_close_interval(200ms by default)
closed_ts_propagation = max(raft_propagation_time,side_propagation_time)
closed_ts_at_sender = now(sender’s current clock) + maxClockOffset(500ms by
default) + buffer (25ms) + closed_ts_propagation
```

Currently, `max_network_rtt` is hardcoded at 150ms, leading to a default 800ms
lead.

Problems with the current implementation:

Tuning trade-offs: A higher lead timestamp leads to high commit wait times,
which inturn impacts write latency. A lower lead timestamp may result in a
closed timestamp that lags present time, causing follower replicas to redirect,
which inturn impacts read latency.

Cluster wide setting: kv.closed_timestamp.lead_for_global_reads_override only
applies to the entire cluster, not per range. For geographically distant
replicas, a short lead time may not be enough for full application. However,
increasing the lead time cluster wide can harm write performance for replicas
that are close together and don’t require as much time to replicate.

This commit adds newly added latency based policies to the side transport
senders.

For side transport senders, nodes create a map of closed timestamps for each
policy without consulting leaseholders. Senders pass this map to leaseholders,
who decide whether to "bump" their closed timestamp and which policy to use.
Senders then send this information to receivers to opt in ranges and pick the
closed timestamps based on the policies ranges opt in.

Currently, side transport senders create only two policies, offering limited
control over closed timestamps based on range specific latencies. The new design
introduces 16 additional policies, each representing a different network latency
bucket ranges fall under. Instead of relying on a hardcoded 150ms latency for
closed timestamp calculations, ranges will now use the appropriate bucket
based on its observed network conditions.

Periodically, the policy refresher iterates over all leaseholders on a node,
updating cached closed timestamp policies based on the observed latency between
the leaseholder replica and the furthest replica. This policy is then used by
side transport and by raft side closed timestamp computation.

Instead of using the hardcoded 150ms network roundtrip latency, a more
dynamic latency would be used based on the latency based closed
timestamp policy the range holds.

kv.closed_timestamp.lead_for_global_reads_auto_tune.enabled can now be used
to auto-tune the lead time that global table ranges use to publish
close timestamps. The kv.closed_timestamp.lead_for_global_reads_override
cluster setting takes precedence over this one. If auto-tuning is disabled or
no data is observed, the system falls back to a hardcoded computed lead time.

Resolves: #59680
Release note: none


Co-authored-by: wenyihu6 <[email protected]>

Author: craig[bot]

pkg/ccl/multiregionccl/BUILD.bazel

@@ -61,6 +61,8 @@ go_test(
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/allocator/allocatorimpl",
+        "//pkg/kv/kvserver/closedts",
+        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/roachpb",
         "//pkg/rpc",
         "//pkg/security/securityassets",

pkg/ccl/multiregionccl/multiregion_test.go

@@ -6,14 +6,23 @@
 package multiregionccl_test
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/multiregionccl/multiregionccltestutils"
+	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/ctpb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -150,3 +159,104 @@ func TestGlobalReadsAfterEnterpriseDisabled(t *testing.T) {
 	_, err = sqlDB.Exec(`ALTER TABLE t2 CONFIGURE ZONE USING global_reads = false`)
 	require.NoError(t, err)
 }
+
+// This test verifies closed timestamp policy behavior with and without the
+// auto-tuning. With the auto-tuning enabled, we expect latency-based policies
+// on global tables. Without it, no latency-based policies should exist.
+func TestReplicaClosedTSPolicyWithPolicyRefresher(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	st := cluster.MakeTestingClusterSettings()
+
+	// Helper function to check if a policy is a newly introduced latency-based policy.
+	isLatencyBasedPolicy := func(policy ctpb.RangeClosedTimestampPolicy) bool {
+		return policy >= ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_20MS &&
+			policy <= ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_EQUAL_OR_GREATER_THAN_300MS
+	}
+
+	// Set small intervals for faster testing.
+	closedts.LeadForGlobalReadsAutoTuneEnabled.Override(ctx, &st.SV, true)
+	closedts.RangeClosedTimestampPolicyRefreshInterval.Override(ctx, &st.SV, 5*time.Millisecond)
+	closedts.RangeClosedTimestampPolicyLatencyRefreshInterval.Override(ctx, &st.SV, 5*time.Millisecond)
+
+	// Create a multi-region cluster with manual replication.
+	numServers := 3
+	tc, sqlDB, cleanup := multiregionccltestutils.TestingCreateMultiRegionCluster(
+		t, numServers, base.TestingKnobs{}, multiregionccltestutils.WithReplicationMode(base.ReplicationManual),
+		multiregionccltestutils.WithSettings(st),
+	)
+	defer cleanup()
+
+	// Create a multi-region database and global table.
+	_, err := sqlDB.Exec(`CREATE DATABASE t PRIMARY REGION "us-east1" REGIONS "us-east2", "us-east3"`)
+	require.NoError(t, err)
+	_, err = sqlDB.Exec(`CREATE TABLE t.test_table (k INT PRIMARY KEY) LOCALITY GLOBAL`)
+	require.NoError(t, err)
+
+	// Look up the table ID and get its key prefix.
+	var tableID uint32
+	err = sqlDB.QueryRow(`SELECT id from system.namespace WHERE name='test_table'`).Scan(&tableID)
+	require.NoError(t, err)
+	tablePrefix := keys.MustAddr(keys.SystemSQLCodec.TablePrefix(tableID))
+	// Split the range at the table prefix and replicate it across all nodes.
+	tc.SplitRangeOrFatal(t, tablePrefix.AsRawKey())
+	tc.AddVotersOrFatal(t, tablePrefix.AsRawKey(), tc.Target(1), tc.Target(2))
+
+	// Get the store and replica for testing.
+	store := tc.GetFirstStoreFromServer(t, 0)
+	replica := store.LookupReplica(roachpb.RKey(tablePrefix.AsRawKey()))
+	require.NotNil(t, replica)
+
+	// Wait for the closed timestamp policies to stabilize and verify the
+	// expected behavior.
+	testutils.SucceedsSoon(t, func() error {
+		snapshot := store.GetStoreConfig().ClosedTimestampSender.GetSnapshot()
+		if len(snapshot.ClosedTimestamps) != int(ctpb.MAX_CLOSED_TIMESTAMP_POLICY) {
+			return errors.Errorf("expected %d closed timestamps, got %d",
+				ctpb.MAX_CLOSED_TIMESTAMP_POLICY, len(snapshot.ClosedTimestamps))
+		}
+
+		// Ensure there are ranges to check.
+		if len(snapshot.AddedOrUpdated) == 0 {
+			return errors.Errorf("no ranges")
+		}
+
+		// With policy refresher enabled: expect to find latency-based policy.
+		for _, policy := range snapshot.AddedOrUpdated {
+			if policy.RangeID != replica.GetRangeID() {
+				continue
+			}
+			if isLatencyBasedPolicy(policy.Policy) {
+				return nil
+			}
+		}
+		return errors.Errorf("no ranges with latency based policy")
+	})
+
+	// Without policy refresher: expect to NOT find latency-based policy.
+	_, err = sqlDB.Exec(`SET CLUSTER SETTING kv.closed_timestamp.lead_for_global_reads_auto_tune.enabled = false`)
+	require.NoError(t, err)
+
+	// Wait for policy refresher to disable and verify no latency-based policies exist.
+	testutils.SucceedsSoon(t, func() error {
+		snapshot := store.GetStoreConfig().ClosedTimestampSender.GetSnapshot()
+		if len(snapshot.ClosedTimestamps) != int(ctpb.MAX_CLOSED_TIMESTAMP_POLICY) {
+			return errors.Errorf("expected %d closed timestamps, got %d",
+				ctpb.MAX_CLOSED_TIMESTAMP_POLICY, len(snapshot.ClosedTimestamps))
+		}
+
+		// Ensure there are ranges to check.
+		if len(snapshot.AddedOrUpdated) == 0 {
+			return errors.Errorf("no ranges")
+		}
+
+		// Verify no latency-based policies remain.
+		for _, policy := range snapshot.AddedOrUpdated {
+			if isLatencyBasedPolicy(policy.Policy) {
+				return errors.Errorf("range %d has latency based policy %s", policy.RangeID, policy.Policy)
+			}
+		}
+		return nil
+	})
+}

pkg/kv/kvserver/closedts/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "closedts",
     srcs = [
         "policy.go",
+        "policy_calculation.go",
         "setting.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts",
@@ -17,7 +18,10 @@ go_library(
 
 go_test(
     name = "closedts_test",
-    srcs = ["policy_test.go"],
+    srcs = [
+        "policy_calculation_test.go",
+        "policy_test.go",
+    ],
     embed = [":closedts"],
     deps = [
         "//pkg/kv/kvserver/closedts/ctpb",

pkg/kv/kvserver/closedts/ctpb/service.proto

@@ -36,10 +36,29 @@ enum RangeClosedTimestampPolicy {
   // configured to lead present time such that all followers of the range are
   // able to serve consistent, present time reads.
   //
-  // Lead policy for global reads with no locality information.
+  // Lead policy for global reads with no locality information. It uses a
+  // hardcoded network latency 150ms by default.
   LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO = 1;
+  // Lead policy for ranges with global reads policy. The following policies are
+  // selected based on max leaseholder-to-follower latency.
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_20MS = 2;  // [0,20)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_40MS = 3;  // [20,40)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_60MS = 4;  // [40,60)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_80MS = 5;  // [60,80)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_100MS = 6;  // [80,100)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_120MS = 7;  // [100,120)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_140MS = 8;  // [120,140)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_160MS = 9;  // [140,160)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_180MS = 10; // [160,180)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_200MS = 11; // [180,200)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_220MS = 12; // [200,220)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_240MS = 13; // [220,240)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_260MS = 14; // [240,260)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_280MS = 15; // [260,280)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_300MS = 16; // [280,300)ms
+  LEAD_FOR_GLOBAL_READS_LATENCY_EQUAL_OR_GREATER_THAN_300MS = 17; // >=300ms
   // Sentinel value for slice sizing.
-  MAX_CLOSED_TIMESTAMP_POLICY = 2;
+  MAX_CLOSED_TIMESTAMP_POLICY = 18;
 }
 
 // Update contains information about (the advancement of) closed timestamps for

pkg/kv/kvserver/closedts/policy.go

@@ -13,7 +13,8 @@ import (
 )
 
 const (
-	DefaultMaxNetworkRTT = 150 * time.Millisecond
+	DefaultMaxNetworkRTT             = 150 * time.Millisecond
+	closedTimestampPolicyBucketWidth = 20 * time.Millisecond
 )
 
 // computeLeadTimeForGlobalReads calculates how far ahead of the current time a
@@ -108,17 +109,18 @@ func TargetForPolicy(
 	policy ctpb.RangeClosedTimestampPolicy,
 ) hlc.Timestamp {
 	var targetOffsetTime time.Duration
-	switch policy {
-	case ctpb.LAG_BY_CLUSTER_SETTING:
+	switch {
+	case policy == ctpb.LAG_BY_CLUSTER_SETTING:
 		// Simple calculation: lag now by desired duration.
 		targetOffsetTime = -lagTargetDuration
-	case ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO:
+	case policy >= ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO &&
+		policy <= ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_EQUAL_OR_GREATER_THAN_300MS:
 		// Override entirely with cluster setting, if necessary.
 		if leadTargetOverride != 0 {
 			targetOffsetTime = leadTargetOverride
 			break
 		}
-		targetOffsetTime = computeLeadTimeForGlobalReads(DefaultMaxNetworkRTT,
+		targetOffsetTime = computeLeadTimeForGlobalReads(computeNetworkRTTBasedOnPolicy(policy),
 			maxClockOffset, sideTransportCloseInterval)
 	default:
 		panic("unexpected RangeClosedTimestampPolicy")

pkg/kv/kvserver/closedts/policy_calculation.go

@@ -0,0 +1,49 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package closedts
+
+import (
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/ctpb"
+)
+
+// FindBucketBasedOnNetworkRTT maps a network RTT to a closed timestamp policy
+// bucket.
+func FindBucketBasedOnNetworkRTT(networkRTT time.Duration) ctpb.RangeClosedTimestampPolicy {
+	// If maxLatency is negative (i.e. no peer latency is provided), return
+	// LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO
+	if networkRTT < 0 {
+		return ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO
+	}
+	if networkRTT >= 300*time.Millisecond {
+		return ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_EQUAL_OR_GREATER_THAN_300MS
+	}
+	// Divide RTT by policy interval, add 1 for zero-based indexing, and offset by
+	// the base policy enum value.
+	bucketNum := int32(math.Floor(float64(networkRTT)/float64(closedTimestampPolicyBucketWidth))) + 1
+	return ctpb.RangeClosedTimestampPolicy(bucketNum + int32(ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO))
+}
+
+// computeNetworkRTTBasedOnPolicy converts a closed timestamp policy to an estimated
+// network RTT.
+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))
+	}
+}