kvserver: replace roachpb. with ctpb.RangeClosedTimestampPolicy

wenyihu6 · wenyihu6 · commit 2a2f28a4f4c9 · 2025-03-27T21:57:07.000-04:00
This commit replaces roachpb.RangeClosedTimestampPolicy with
ctpb.LatencyBasedRangeClosedTimestampPolicy for kvserver sidetransport side.
Currently, it maintains the existing behavior. Future commits will introduce
additional enum values to add locality information into the side-transport
closed timestamp estimation.

Epic: none
Release note: none
diff --git a/pkg/kv/BUILD.bazel b/pkg/kv/BUILD.bazel
@@ -20,6 +20,7 @@ go_library(
         "//pkg/keys",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/closedts",
+        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/kv/kvserver/concurrency/isolation",
         "//pkg/roachpb",
         "//pkg/settings",
diff --git a/pkg/kv/kvserver/BUILD.bazel b/pkg/kv/kvserver/BUILD.bazel
@@ -140,6 +140,7 @@ go_library(
         "//pkg/kv/kvserver/batcheval/result",
         "//pkg/kv/kvserver/benignerror",
         "//pkg/kv/kvserver/closedts",
+        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/kv/kvserver/closedts/sidetransport",
         "//pkg/kv/kvserver/closedts/tracker",
         "//pkg/kv/kvserver/concurrency",
@@ -426,6 +427,7 @@ go_test(
         "//pkg/kv/kvserver/batcheval/result",
         "//pkg/kv/kvserver/benignerror",
         "//pkg/kv/kvserver/closedts",
+        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/kv/kvserver/closedts/tracker",
         "//pkg/kv/kvserver/concurrency",
         "//pkg/kv/kvserver/concurrency/isolation",
diff --git a/pkg/kv/kvserver/closedts/BUILD.bazel b/pkg/kv/kvserver/closedts/BUILD.bazel
@@ -9,7 +9,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts",
     visibility = ["//visibility:public"],
     deps = [
-        "//pkg/roachpb",
+        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/settings",
         "//pkg/util/hlc",
     ],
@@ -20,7 +20,7 @@ go_test(
     srcs = ["policy_test.go"],
     embed = [":closedts"],
     deps = [
-        "//pkg/roachpb",
+        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
diff --git a/pkg/kv/kvserver/closedts/ctpb/BUILD.bazel b/pkg/kv/kvserver/closedts/ctpb/BUILD.bazel
@@ -21,7 +21,6 @@ proto_library(
     strip_import_prefix = "/pkg",
     visibility = ["//visibility:public"],
     deps = [
-        "//pkg/roachpb:roachpb_proto",
         "//pkg/util/hlc:hlc_proto",
         "@com_github_gogo_protobuf//gogoproto:gogo_proto",
     ],
@@ -34,7 +33,6 @@ go_proto_library(
     proto = ":ctpb_proto",
     visibility = ["//visibility:public"],
     deps = [
-        "//pkg/roachpb",
         "//pkg/util/hlc",
         "@com_github_gogo_protobuf//gogoproto",
     ],
diff --git a/pkg/kv/kvserver/closedts/ctpb/service.proto b/pkg/kv/kvserver/closedts/ctpb/service.proto
@@ -7,11 +7,41 @@ syntax = "proto3";
 package cockroach.kv.kvserver.ctupdate;
 option go_package = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/ctpb";
 
-import "roachpb/data.proto";
 import "util/hlc/timestamp.proto";
 
 import "gogoproto/gogo.proto";
 
+// RangeClosedTimestampPolicy defines a range's closed timestamp policy. All
+// ranges in a cluster are placed in a closed timestamp policy bucket, which is
+// decided by the leaseholder replica of the range. The side transport sender
+// periodically transmits closed timestamp updates for ranges it's responsible
+// for, closing timestamps on a per-bucket basis.
+//
+// NB: The policy is only used inside kvserver, and shouldn't leak to the
+// client. See roachpb.ClosedTimestampPolicy and toClientClosedTsPolicy for the
+// client side equivalent.
+//
+// Closed timestamp policies on the server can be dynamically adjusted based on
+// observed network latencies between the leaseholder and its farthest follower.
+// Each range is bucketed based on observed network latencies, and buckets may
+// change as network latencies change, or the range's SpanConfig changes.
+enum RangeClosedTimestampPolicy {
+  option (gogoproto.goproto_enum_prefix) = false;
+  // LAG_BY_CLUSTER_SETTING indicates that the range's closed timestamp is
+  // configured to lag behind present time by the value configured for the
+  // `kv.closed_timestamp.target_duration` cluster setting.
+  LAG_BY_CLUSTER_SETTING = 0;
+  // The following policies correspond to roachpb.LEAD_FOR_GLOBAL_READS policy.
+  // LEAD_FOR_GLOBAL_READS indicates that the range's closed timestamp is
+  // 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_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO = 1;
+  // Sentinel value for slice sizing.
+  MAX_CLOSED_TIMESTAMP_POLICY = 2;
+}
+
 // Update contains information about (the advancement of) closed timestamps for
 // ranges with leases on the sender node. Updates are of two types: snapshots
 // and incrementals. Snapshots are stand-alone messages, explicitly containing
@@ -57,7 +87,7 @@ message Update {
   // and the regular Raft transport are possible, as are races between two
   // side-transport streams for an outgoing and incoming leaseholder.
   message GroupUpdate {
-    roachpb.RangeClosedTimestampPolicy policy = 1;
+    RangeClosedTimestampPolicy policy = 1;
     util.hlc.Timestamp closed_timestamp = 2 [(gogoproto.nullable) = false];
   }
   repeated GroupUpdate closed_timestamps = 4 [(gogoproto.nullable) = false];
@@ -80,7 +110,7 @@ message Update {
   message RangeUpdate {
     uint64 range_id = 1 [(gogoproto.customname) = "RangeID", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/roachpb.RangeID"];
     uint64 lai = 2 [(gogoproto.customname) = "LAI", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/kv/kvpb.LeaseAppliedIndex"];
-    roachpb.RangeClosedTimestampPolicy policy = 3;
+    RangeClosedTimestampPolicy policy = 3;
   }
   repeated RangeUpdate added_or_updated = 6 [(gogoproto.nullable) = false];
 }
diff --git a/pkg/kv/kvserver/closedts/policy.go b/pkg/kv/kvserver/closedts/policy.go
@@ -8,7 +8,7 @@ package closedts
 import (
 	"time"
 
-	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/ctpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 )
 
@@ -104,14 +104,14 @@ func TargetForPolicy(
 	lagTargetDuration time.Duration,
 	leadTargetOverride time.Duration,
 	sideTransportCloseInterval time.Duration,
-	policy roachpb.RangeClosedTimestampPolicy,
+	policy ctpb.RangeClosedTimestampPolicy,
 ) hlc.Timestamp {
 	var targetOffsetTime time.Duration
 	switch policy {
-	case roachpb.LAG_BY_CLUSTER_SETTING:
+	case ctpb.LAG_BY_CLUSTER_SETTING:
 		// Simple calculation: lag now by desired duration.
 		targetOffsetTime = -lagTargetDuration
-	case roachpb.LEAD_FOR_GLOBAL_READS:
+	case ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO:
 		// Override entirely with cluster setting, if necessary.
 		if leadTargetOverride != 0 {
 			targetOffsetTime = leadTargetOverride
diff --git a/pkg/kv/kvserver/closedts/policy_test.go b/pkg/kv/kvserver/closedts/policy_test.go
@@ -9,7 +9,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/ctpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -31,30 +31,30 @@ func TestTargetForPolicy(t *testing.T) {
 		lagTargetNanos             time.Duration
 		leadTargetOverride         time.Duration
 		sideTransportCloseInterval time.Duration
-		rangePolicy                roachpb.RangeClosedTimestampPolicy
+		rangePolicy                ctpb.RangeClosedTimestampPolicy
 		expClosedTSTarget          hlc.Timestamp
 	}{
 		{
 			lagTargetNanos:    secs(3),
-			rangePolicy:       roachpb.LAG_BY_CLUSTER_SETTING,
+			rangePolicy:       ctpb.LAG_BY_CLUSTER_SETTING,
 			expClosedTSTarget: now.Add(-secs(3).Nanoseconds(), 0),
 		},
 		{
 			lagTargetNanos:    secs(1),
-			rangePolicy:       roachpb.LAG_BY_CLUSTER_SETTING,
+			rangePolicy:       ctpb.LAG_BY_CLUSTER_SETTING,
 			expClosedTSTarget: now.Add(-secs(1).Nanoseconds(), 0),
 		},
 		{
 			sideTransportCloseInterval: millis(200),
-			rangePolicy:                roachpb.LEAD_FOR_GLOBAL_READS,
+			rangePolicy:                ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
 			expClosedTSTarget: now.
 				Add((maxClockOffset +
 					millis(275) /* sideTransportPropTime */ +
 					millis(25) /* bufferTime */).Nanoseconds(), 0),
 		},
 		{
 			sideTransportCloseInterval: millis(50),
-			rangePolicy:                roachpb.LEAD_FOR_GLOBAL_READS,
+			rangePolicy:                ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
 			expClosedTSTarget: now.
 				Add((maxClockOffset +
 					millis(245) /* raftTransportPropTime */ +
@@ -63,7 +63,7 @@ func TestTargetForPolicy(t *testing.T) {
 		{
 			leadTargetOverride:         millis(1234),
 			sideTransportCloseInterval: millis(200),
-			rangePolicy:                roachpb.LEAD_FOR_GLOBAL_READS,
+			rangePolicy:                ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
 			expClosedTSTarget:          now.Add(millis(1234).Nanoseconds(), 0),
 		},
 	} {
diff --git a/pkg/kv/kvserver/closedts/sidetransport/receiver_test.go b/pkg/kv/kvserver/closedts/sidetransport/receiver_test.go
@@ -86,13 +86,13 @@ func TestIncomingStreamProcessUpdateBasic(t *testing.T) {
 		SeqNum:   1,
 		Snapshot: true,
 		ClosedTimestamps: []ctpb.Update_GroupUpdate{
-			{Policy: roachpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: ts10},
-			{Policy: roachpb.LEAD_FOR_GLOBAL_READS, ClosedTimestamp: ts20},
+			{Policy: ctpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: ts10},
+			{Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO, ClosedTimestamp: ts20},
 		},
 		AddedOrUpdated: []ctpb.Update_RangeUpdate{
-			{RangeID: 1, LAI: lai100, Policy: roachpb.LAG_BY_CLUSTER_SETTING},
-			{RangeID: 2, LAI: lai101, Policy: roachpb.LAG_BY_CLUSTER_SETTING},
-			{RangeID: 3, LAI: lai102, Policy: roachpb.LEAD_FOR_GLOBAL_READS},
+			{RangeID: 1, LAI: lai100, Policy: ctpb.LAG_BY_CLUSTER_SETTING},
+			{RangeID: 2, LAI: lai101, Policy: ctpb.LAG_BY_CLUSTER_SETTING},
+			{RangeID: 3, LAI: lai102, Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO},
 		},
 		Removed: nil,
 	}
@@ -114,11 +114,11 @@ func TestIncomingStreamProcessUpdateBasic(t *testing.T) {
 		SeqNum:   2,
 		Snapshot: false,
 		ClosedTimestamps: []ctpb.Update_GroupUpdate{
-			{Policy: roachpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: ts11},
-			{Policy: roachpb.LEAD_FOR_GLOBAL_READS, ClosedTimestamp: ts21},
+			{Policy: ctpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: ts11},
+			{Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO, ClosedTimestamp: ts21},
 		},
 		AddedOrUpdated: []ctpb.Update_RangeUpdate{
-			{RangeID: 3, LAI: lai103, Policy: roachpb.LEAD_FOR_GLOBAL_READS},
+			{RangeID: 3, LAI: lai103, Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO},
 		},
 		Removed: []roachpb.RangeID{1},
 	}
@@ -140,12 +140,12 @@ func TestIncomingStreamProcessUpdateBasic(t *testing.T) {
 		SeqNum:   3,
 		Snapshot: true,
 		ClosedTimestamps: []ctpb.Update_GroupUpdate{
-			{Policy: roachpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: ts12},
-			{Policy: roachpb.LEAD_FOR_GLOBAL_READS, ClosedTimestamp: ts22},
+			{Policy: ctpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: ts12},
+			{Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO, ClosedTimestamp: ts22},
 		},
 		AddedOrUpdated: []ctpb.Update_RangeUpdate{
-			{RangeID: 3, LAI: lai102, Policy: roachpb.LEAD_FOR_GLOBAL_READS},
-			{RangeID: 4, LAI: lai100, Policy: roachpb.LAG_BY_CLUSTER_SETTING},
+			{RangeID: 3, LAI: lai102, Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO},
+			{RangeID: 4, LAI: lai100, Policy: ctpb.LAG_BY_CLUSTER_SETTING},
 		},
 		Removed: nil,
 	}
@@ -186,10 +186,10 @@ func TestIncomingStreamCallsIntoStoresDontHoldLock(t *testing.T) {
 	msg := &ctpb.Update{
 		NodeID: 1, SeqNum: 1, Snapshot: true,
 		ClosedTimestamps: []ctpb.Update_GroupUpdate{
-			{Policy: roachpb.LEAD_FOR_GLOBAL_READS, ClosedTimestamp: ts10},
+			{Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO, ClosedTimestamp: ts10},
 		},
 		AddedOrUpdated: []ctpb.Update_RangeUpdate{
-			{RangeID: 1, LAI: lai100, Policy: roachpb.LEAD_FOR_GLOBAL_READS},
+			{RangeID: 1, LAI: lai100, Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO},
 		},
 		Removed: nil,
 	}
diff --git a/pkg/kv/kvserver/closedts/sidetransport/sender.go b/pkg/kv/kvserver/closedts/sidetransport/sender.go
@@ -116,7 +116,7 @@ type connTestingKnobs struct {
 // about a particular range.
 type trackedRange struct {
 	lai    kvpb.LeaseAppliedIndex
-	policy roachpb.RangeClosedTimestampPolicy
+	policy ctpb.RangeClosedTimestampPolicy
 }
 
 // leaseholder represents a leaseholder replicas that has been registered with
@@ -170,7 +170,7 @@ type BumpSideTransportClosedResult struct {
 	// The range's current LAI, to be associated with the closed timestamp.
 	LAI kvpb.LeaseAppliedIndex
 	// The range's current policy.
-	Policy roachpb.RangeClosedTimestampPolicy
+	Policy ctpb.RangeClosedTimestampPolicy
 }
 
 // CantCloseReason enumerates the reasons why BunpSideTransportClosed might fail
@@ -327,7 +327,7 @@ func (s *Sender) publish(ctx context.Context) hlc.ClockTimestamp {
 	leadTargetOverride := closedts.LeadForGlobalReadsOverride.Get(&s.st.SV)
 	sideTransportCloseInterval := closedts.SideTransportCloseInterval.Get(&s.st.SV)
 	for i := range s.trackedMu.lastClosed {
-		pol := roachpb.RangeClosedTimestampPolicy(i)
+		pol := ctpb.RangeClosedTimestampPolicy(i)
 		target := closedts.TargetForPolicy(
 			now,
 			maxClockOffset,
@@ -487,7 +487,7 @@ func (s *Sender) GetSnapshot() *ctpb.Update {
 	}
 	for pol, ts := range s.trackedMu.lastClosed {
 		msg.ClosedTimestamps[pol] = ctpb.Update_GroupUpdate{
-			Policy:          roachpb.RangeClosedTimestampPolicy(pol),
+			Policy:          ctpb.RangeClosedTimestampPolicy(pol),
 			ClosedTimestamp: ts,
 		}
 	}
@@ -895,7 +895,7 @@ func (s streamState) String() string {
 		id roachpb.RangeID
 		trackedRange
 	}
-	rangesByPolicy := make(map[roachpb.RangeClosedTimestampPolicy][]rangeInfo)
+	rangesByPolicy := make(map[ctpb.RangeClosedTimestampPolicy][]rangeInfo)
 	for rid, info := range s.tracked {
 		rangesByPolicy[info.policy] = append(rangesByPolicy[info.policy], rangeInfo{id: rid, trackedRange: info})
 	}
diff --git a/pkg/kv/kvserver/closedts/sidetransport/sender_test.go b/pkg/kv/kvserver/closedts/sidetransport/sender_test.go
@@ -43,7 +43,7 @@ type mockReplica struct {
 	canBump        bool
 	cantBumpReason CantCloseReason
 	lai            kvpb.LeaseAppliedIndex
-	policy         roachpb.RangeClosedTimestampPolicy
+	policy         ctpb.RangeClosedTimestampPolicy
 }
 
 var _ Replica = &mockReplica{}
@@ -120,7 +120,7 @@ func newMockReplica(id roachpb.RangeID, nodes ...roachpb.NodeID) *mockReplica {
 		rangeID: id,
 		canBump: true,
 		lai:     5,
-		policy:  roachpb.LAG_BY_CLUSTER_SETTING,
+		policy:   ctpb.LAG_BY_CLUSTER_SETTING,
 	}
 	r.mu.desc = desc
 	return r
@@ -137,14 +137,14 @@ func newMockReplicaEx(id roachpb.RangeID, replicas ...roachpb.ReplicationTarget)
 		rangeID: id,
 		canBump: true,
 		lai:     5,
-		policy:  roachpb.LAG_BY_CLUSTER_SETTING,
+		policy:   ctpb.LAG_BY_CLUSTER_SETTING,
 	}
 	r.mu.desc = desc
 	return r
 }
 
 func expGroupUpdates(s *Sender, now hlc.ClockTimestamp) []ctpb.Update_GroupUpdate {
-	targetForPolicy := func(pol roachpb.RangeClosedTimestampPolicy) hlc.Timestamp {
+	targetForPolicy := func(pol ctpb.RangeClosedTimestampPolicy) hlc.Timestamp {
 		return closedts.TargetForPolicy(
 			now,
 			s.clock.MaxOffset(),
@@ -155,8 +155,8 @@ func expGroupUpdates(s *Sender, now hlc.ClockTimestamp) []ctpb.Update_GroupUpdat
 		)
 	}
 	return []ctpb.Update_GroupUpdate{
-		{Policy: roachpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: targetForPolicy(roachpb.LAG_BY_CLUSTER_SETTING)},
-		{Policy: roachpb.LEAD_FOR_GLOBAL_READS, ClosedTimestamp: targetForPolicy(roachpb.LEAD_FOR_GLOBAL_READS)},
+		{Policy: ctpb.LAG_BY_CLUSTER_SETTING, ClosedTimestamp: targetForPolicy(ctpb.LAG_BY_CLUSTER_SETTING)},
+		{Policy: ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO, ClosedTimestamp: targetForPolicy(ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO)},
 	}
 }
 
@@ -190,7 +190,7 @@ func TestSenderBasic(t *testing.T) {
 	now = s.publish(ctx)
 	require.Len(t, s.trackedMu.tracked, 1)
 	require.Equal(t, map[roachpb.RangeID]trackedRange{
-		15: {lai: 5, policy: roachpb.LAG_BY_CLUSTER_SETTING},
+		15: {lai: 5, policy: ctpb.LAG_BY_CLUSTER_SETTING},
 	}, s.trackedMu.tracked)
 	require.Len(t, s.leaseholdersMu.leaseholders, 1)
 	require.Len(t, s.connsMu.conns, 2)
@@ -204,7 +204,7 @@ func TestSenderBasic(t *testing.T) {
 	require.Equal(t, expGroupUpdates(s, now), up.ClosedTimestamps)
 	require.Nil(t, up.Removed)
 	require.Equal(t, []ctpb.Update_RangeUpdate{
-		{RangeID: 15, LAI: 5, Policy: roachpb.LAG_BY_CLUSTER_SETTING},
+		{RangeID: 15, LAI: 5, Policy: ctpb.LAG_BY_CLUSTER_SETTING},
 	}, up.AddedOrUpdated)
 
 	c2, ok := s.connsMu.conns[2]
@@ -287,7 +287,7 @@ func TestSenderColocateReplicasOnSameNode(t *testing.T) {
 	now := s.publish(ctx)
 	require.Len(t, s.trackedMu.tracked, 1)
 	require.Equal(t, map[roachpb.RangeID]trackedRange{
-		15: {lai: 5, policy: roachpb.LAG_BY_CLUSTER_SETTING},
+		15: {lai: 5, policy: ctpb.LAG_BY_CLUSTER_SETTING},
 	}, s.trackedMu.tracked)
 	require.Len(t, s.leaseholdersMu.leaseholders, 1)
 	// Ensure that we have two connections, one for remote node and one for local.
@@ -304,7 +304,7 @@ func TestSenderColocateReplicasOnSameNode(t *testing.T) {
 	require.Equal(t, expGroupUpdates(s, now), up.ClosedTimestamps)
 	require.Nil(t, up.Removed)
 	require.Equal(t, []ctpb.Update_RangeUpdate{
-		{RangeID: 15, LAI: 5, Policy: roachpb.LAG_BY_CLUSTER_SETTING},
+		{RangeID: 15, LAI: 5, Policy: ctpb.LAG_BY_CLUSTER_SETTING},
 	}, up.AddedOrUpdated)
 }
 
diff --git a/pkg/kv/kvserver/helpers_test.go b/pkg/kv/kvserver/helpers_test.go
@@ -599,7 +599,7 @@ func (r *Replica) ReadCachedProtectedTS() (readAt, earliestProtectionTimestamp h
 func (r *Replica) ClosedTimestampPolicy() roachpb.RangeClosedTimestampPolicy {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
-	return r.closedTimestampPolicyRLocked()
+	return toClientClosedTsPolicy(r.closedTimestampPolicyRLocked())
 }
 
 // TripBreaker synchronously trips the breaker.
diff --git a/pkg/kv/kvserver/replica.go b/pkg/kv/kvserver/replica.go
diff --git a/pkg/kv/kvserver/replica_proposal_buf_test.go b/pkg/kv/kvserver/replica_proposal_buf_test.go
diff --git a/pkg/kv/txn.go b/pkg/kv/txn.go

Original file line number	Diff line number	Diff line change
`@@ -116,7 +116,7 @@ type connTestingKnobs struct {`
`116`	`116`	`// about a particular range.`
`117`	`117`	`type trackedRange struct {`
`118`	`118`	`lai kvpb.LeaseAppliedIndex`
`119`		`- policy roachpb.RangeClosedTimestampPolicy`
	`119`	`+ policy ctpb.RangeClosedTimestampPolicy`
`120`	`120`	`}`
`121`	`121`
`122`	`122`	`// leaseholder represents a leaseholder replicas that has been registered with`
`@@ -170,7 +170,7 @@ type BumpSideTransportClosedResult struct {`
`170`	`170`	`// The range's current LAI, to be associated with the closed timestamp.`
`171`	`171`	`LAI kvpb.LeaseAppliedIndex`
`172`	`172`	`// The range's current policy.`
`173`		`- Policy roachpb.RangeClosedTimestampPolicy`
	`173`	`+ Policy ctpb.RangeClosedTimestampPolicy`
`174`	`174`	`}`
`175`	`175`
`176`	`176`	`// CantCloseReason enumerates the reasons why BunpSideTransportClosed might fail`
`@@ -327,7 +327,7 @@ func (s *Sender) publish(ctx context.Context) hlc.ClockTimestamp {`
`327`	`327`	`leadTargetOverride := closedts.LeadForGlobalReadsOverride.Get(&s.st.SV)`
`328`	`328`	`sideTransportCloseInterval := closedts.SideTransportCloseInterval.Get(&s.st.SV)`
`329`	`329`	`for i := range s.trackedMu.lastClosed {`
`330`		`- pol := roachpb.RangeClosedTimestampPolicy(i)`
	`330`	`+ pol := ctpb.RangeClosedTimestampPolicy(i)`
`331`	`331`	`target := closedts.TargetForPolicy(`
`332`	`332`	`now,`
`333`	`333`	`maxClockOffset,`
`@@ -487,7 +487,7 @@ func (s Sender) GetSnapshot() ctpb.Update {`
`487`	`487`	`}`
`488`	`488`	`for pol, ts := range s.trackedMu.lastClosed {`
`489`	`489`	`msg.ClosedTimestamps[pol] = ctpb.Update_GroupUpdate{`
`490`		`- Policy: roachpb.RangeClosedTimestampPolicy(pol),`
	`490`	`+ Policy: ctpb.RangeClosedTimestampPolicy(pol),`
`491`	`491`	`ClosedTimestamp: ts,`
`492`	`492`	`}`
`493`	`493`	`}`
`@@ -895,7 +895,7 @@ func (s streamState) String() string {`
`895`	`895`	`id roachpb.RangeID`
`896`	`896`	`trackedRange`
`897`	`897`	`}`
`898`		`- rangesByPolicy := make(map[roachpb.RangeClosedTimestampPolicy][]rangeInfo)`
	`898`	`+ rangesByPolicy := make(map[ctpb.RangeClosedTimestampPolicy][]rangeInfo)`
`899`	`899`	`for rid, info := range s.tracked {`
`900`	`900`	`rangesByPolicy[info.policy] = append(rangesByPolicy[info.policy], rangeInfo{id: rid, trackedRange: info})`
`901`	`901`	`}`