Merge #143010

143010: kvserver: replace roachpb. with ctpb.RangeClosedTimestampPolicy r=arulajmani a=wenyihu6


**kvserver: replace roachpb. with ctpb.RangeClosedTimestampPolicy**

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

---

**kvserver/closedts: change lastClosed to a map**

This commit changes streamState shared between sender and
receiver from a fixed size slice to a map, similar to
trackedRange. Currently, it does not change any existing
behavior. But future commits use this to make sure only
lastClosed will be populated for enums with a corresponding
roachpb.RangeClosedTimestampPolicy until the cluster has
been fully upgraded to v25.2.

Part of: #59680
Release note: none



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

Author: craig[bot]

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",

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",

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",

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",
     ],

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];
 }

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"
 )
 
@@ -105,14 +105,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

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),
 		},
 	} {

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

@@ -246,8 +246,11 @@ func (r *incomingStream) processUpdate(ctx context.Context, msg *ctpb.Update) {
 			if !ok {
 				log.Fatalf(ctx, "attempting to unregister a missing range: r%d", rangeID)
 			}
-			r.stores.ForwardSideTransportClosedTimestampForRange(
-				ctx, rangeID, r.mu.lastClosed[info.policy], info.lai)
+			ts, ok := r.mu.lastClosed[info.policy]
+			if !ok {
+				log.Fatalf(ctx, "missing closed timestamp policy %v for range r%d", info.policy, rangeID)
+			}
+			r.stores.ForwardSideTransportClosedTimestampForRange(ctx, rangeID, ts, info.lai)
 		}
 		r.mu.RUnlock()
 	}
@@ -258,9 +261,7 @@ func (r *incomingStream) processUpdate(ctx context.Context, msg *ctpb.Update) {
 
 	// Reset all the state on snapshots.
 	if msg.Snapshot {
-		for i := range r.mu.lastClosed {
-			r.mu.lastClosed[i] = hlc.Timestamp{}
-		}
+		r.mu.lastClosed = make(map[ctpb.RangeClosedTimestampPolicy]hlc.Timestamp, len(r.mu.lastClosed))
 		r.mu.tracked = make(map[roachpb.RangeID]trackedRange, len(r.mu.tracked))
 	} else if msg.SeqNum != r.mu.lastSeqNum+1 {
 		log.Fatalf(ctx, "expected closed timestamp side-transport message with sequence number "+

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,
 	}