kv: ship timestamp cache summary during lease transfers and range merges

Fixes cockroachdb#57688.
Fixes cockroachdb#59679.
Fixes cockroachdb#60520.

This commit introduces new logic to ship summaries of a leaseholders
timestamp cache through lease transfers and range merges. For lease
transfers, the read summary is sent from the outgoing leaseholder to the
incoming leaseholder. For range merges, the read summary is sent from
the right-hand side leaseholder (through the SubsumeResponse), to the
left-hand side leaseholder (through the MergeTrigger).

The read summaries perform the role of the lease start time and merge
freeze time used to play for lease transfers and range merges,
respectively - the summaries instruct the post-operation leaseholder on
how to update its timestamp cache to ensure that no future writes are
allowed to invalidate prior reads.

Read summaries have two distinct advantages over the old approach:
1. they can transfer a higher-resolution snapshot of the reads on the
    range through a lease transfer, to make the lease transfers less
    disruptive to writes because the timestamp cache won't be bumped as
    high. This avoids transaction aborts and retries after lease
    transfers and merges.
2. they can transfer information about reads with synthetic timestamps,
    which are not otherwise captured by the new lease's start time.
    Because of this, they are needed for correctness on `global_read`
    ranges, which can serve reads in the future.

This commit does not realize the first benefit, because it uses very
low-resolution read summaries. However, it sets up the infrastructure
that will allow us to realize the benefit in the future by capturing and
shipping higher-resolution read summaries. The commit does realize the
second benefit, as it fixes correctness issues around future time reads.

----

The commit also fixes a related bug that was revealed during the
development of this patch. As explained in cockroachdb#60520, it was possible for a
range merge to be applied to the leaseholder of the LHS of the merge
through a Raft snapshot. In such cases, we were not properly updating
the leaseholder's timestamp cache to reflect the reads served on the RHS
range. This could allow the post-merged range to invalidate reads served
by the pre-merge RHS range.

This commit fixes this bug using the new read summary infrastructure.
Merge triggers now write to the left-hand side's prior read summary with
a read summary gathered from the right-hand side during subsumption.
Later, upon ingesting a Raft snapshot, we check if we subsumed any
replicas and if we are the leaseholder. If both of those conditions are
true, we forward the replica's timestamp cache to the read summary on
the range. Since this read summary must have been updated by the merge
trigger, it will include all reads served on the pre-merge RHS range.

----

Release note (bug fix): Fixes a very rare, possible impossible in
practice, bug where a range merge that applied through a Raft snapshot
on the left-hand side range's leaseholder could allow that leaseholder
to serve writes that invalidated reads from before the merge on the
right-hand side.

docs/generated/settings/settings.html

@@ -100,6 +100,6 @@
 <tr><td><code>trace.debug.enable</code></td><td>boolean</td><td><code>false</code></td><td>if set, traces for recent requests can be seen at https://<ui>/debug/requests</td></tr>
 <tr><td><code>trace.lightstep.token</code></td><td>string</td><td><code></code></td><td>if set, traces go to Lightstep using this token</td></tr>
 <tr><td><code>trace.zipkin.collector</code></td><td>string</td><td><code></code></td><td>if set, traces go to the given Zipkin instance (example: '127.0.0.1:9411'); ignored if trace.lightstep.token is set</td></tr>
-<tr><td><code>version</code></td><td>version</td><td><code>20.2-24</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
+<tr><td><code>version</code></td><td>version</td><td><code>20.2-26</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
 </tbody>
 </table>

pkg/ccl/kvccl/kvfollowerreadsccl/followerreads.go

@@ -53,6 +53,8 @@ var followerReadMultiple = settings.RegisterFloatSetting(
 // the offset from now to request a follower read. The same value less the clock
 // uncertainty, then is used to determine at the kv layer if a query can use a
 // follower read.
+// TODO(nvanbenschoten): don't route to followers if closedts.TargetDuration is
+// 0, as this disables closed timestamps.
 func getFollowerReadDuration(st *cluster.Settings) time.Duration {
 	targetMultiple := followerReadMultiple.Get(&st.SV)
 	targetDuration := closedts.TargetDuration.Get(&st.SV)

pkg/clusterversion/cockroach_versions.go

@@ -235,6 +235,10 @@ const (
 	// using the replicated legacy TruncatedState. It's also used in asserting
 	// that no replicated truncated state representation is found.
 	PostTruncatedAndRangeAppliedStateMigration
+	// PriorReadSummaries introduces support for the use of read summary objects
+	// ship information about reads on a range through lease changes and range
+	// merges.
+	PriorReadSummaries
 
 	// Step (1): Add new versions here.
 )
@@ -389,6 +393,10 @@ var versionsSingleton = keyedVersions([]keyedVersion{
 		Key:     PostTruncatedAndRangeAppliedStateMigration,
 		Version: roachpb.Version{Major: 20, Minor: 2, Internal: 24},
 	},
+	{
+		Key:     PriorReadSummaries,
+		Version: roachpb.Version{Major: 20, Minor: 2, Internal: 26},
+	},
 	// Step (2): Add new versions here.
 })
 

pkg/keys/constants.go

@@ -93,6 +93,9 @@ var (
 	// LocalLeaseAppliedIndexLegacySuffix is the suffix for the applied lease
 	// index.
 	LocalLeaseAppliedIndexLegacySuffix = []byte("rlla")
+	// LocalRangePriorReadSummarySuffix is the suffix for a range's prior read
+	// summary.
+	LocalRangePriorReadSummarySuffix = []byte("rprs")
 	// LocalRangeVersionSuffix is the suffix for the range version.
 	LocalRangeVersionSuffix = []byte("rver")
 	// LocalRangeStatsLegacySuffix is the suffix for range statistics.

pkg/keys/doc.go

@@ -192,6 +192,7 @@ var _ = [...]interface{}{
 	RaftTruncatedStateLegacyKey, // "rftt"
 	RangeLeaseKey,               // "rll-"
 	LeaseAppliedIndexLegacyKey,  // "rlla"
+	RangePriorReadSummaryKey,    // "rprs"
 	RangeVersionKey,             // "rver"
 	RangeStatsLegacyKey,         // "stat"
 

pkg/keys/keys.go

@@ -271,6 +271,12 @@ func RangeLeaseKey(rangeID roachpb.RangeID) roachpb.Key {
 	return MakeRangeIDPrefixBuf(rangeID).RangeLeaseKey()
 }
 
+// RangePriorReadSummaryKey returns a system-local key for a range's prior read
+// summary.
+func RangePriorReadSummaryKey(rangeID roachpb.RangeID) roachpb.Key {
+	return MakeRangeIDPrefixBuf(rangeID).RangePriorReadSummaryKey()
+}
+
 // RangeStatsLegacyKey returns the key for accessing the MVCCStats struct for
 // the specified Range ID. The key is no longer written to. Its responsibility
 // has been subsumed by the RangeAppliedStateKey.
@@ -951,6 +957,12 @@ func (b RangeIDPrefixBuf) RangeLeaseKey() roachpb.Key {
 	return append(b.replicatedPrefix(), LocalRangeLeaseSuffix...)
 }
 
+// RangePriorReadSummaryKey returns a system-local key for a range's prior read
+// summary.
+func (b RangeIDPrefixBuf) RangePriorReadSummaryKey() roachpb.Key {
+	return append(b.replicatedPrefix(), LocalRangePriorReadSummarySuffix...)
+}
+
 // RangeStatsLegacyKey returns the key for accessing the MVCCStats struct
 // for the specified Range ID.
 // See comment on RangeStatsLegacyKey function.

pkg/keys/printer.go

@@ -169,6 +169,7 @@ var (
 		{name: "RaftTruncatedState", suffix: LocalRaftTruncatedStateLegacySuffix},
 		{name: "RangeLastReplicaGCTimestamp", suffix: LocalRangeLastReplicaGCTimestampSuffix},
 		{name: "RangeLease", suffix: LocalRangeLeaseSuffix},
+		{name: "RangePriorReadSummary", suffix: LocalRangePriorReadSummarySuffix},
 		{name: "RangeStats", suffix: LocalRangeStatsLegacySuffix},
 		{name: "RangeLastGC", suffix: LocalRangeLastGCSuffix},
 		{name: "RangeVersion", suffix: LocalRangeVersionSuffix},

pkg/keys/printer_test.go

@@ -74,6 +74,7 @@ func TestPrettyPrint(t *testing.T) {
 		{keys.RaftTruncatedStateLegacyKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RaftTruncatedState", revertSupportUnknown},
 		{keys.RaftTruncatedStateKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RaftTruncatedState", revertSupportUnknown},
 		{keys.RangeLeaseKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeLease", revertSupportUnknown},
+		{keys.RangePriorReadSummaryKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangePriorReadSummary", revertSupportUnknown},
 		{keys.RangeStatsLegacyKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeStats", revertSupportUnknown},
 		{keys.RangeLastGCKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeLastGC", revertSupportUnknown},
 		{keys.RangeVersionKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeVersion", revertSupportUnknown},

pkg/kv/kvserver/BUILD.bazel

@@ -132,6 +132,8 @@ go_library(
         "//pkg/kv/kvserver/raftentry",
         "//pkg/kv/kvserver/rangefeed",
         "//pkg/kv/kvserver/rditer",
+        "//pkg/kv/kvserver/readsummary",
+        "//pkg/kv/kvserver/readsummary/rspb",
         "//pkg/kv/kvserver/spanset",
         "//pkg/kv/kvserver/split",
         "//pkg/kv/kvserver/stateloader",

pkg/kv/kvserver/batcheval/BUILD.bazel

@@ -60,6 +60,8 @@ go_library(
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/kv/kvserver/kvserverpb",
         "//pkg/kv/kvserver/rditer",
+        "//pkg/kv/kvserver/readsummary",
+        "//pkg/kv/kvserver/readsummary/rspb",
         "//pkg/kv/kvserver/spanset",
         "//pkg/kv/kvserver/stateloader",
         "//pkg/kv/kvserver/txnwait",
@@ -106,6 +108,8 @@ go_test(
         "//pkg/kv",
         "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/abortspan",
+        "//pkg/kv/kvserver/readsummary",
+        "//pkg/kv/kvserver/readsummary/rspb",
         "//pkg/kv/kvserver/spanset",
         "//pkg/roachpb",
         "//pkg/security",

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/rditer"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/stateloader"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -159,6 +160,12 @@ func declareKeysEndTxn(
 					Key:    keys.MakeRangeIDReplicatedPrefix(mt.RightDesc.RangeID),
 					EndKey: keys.MakeRangeIDReplicatedPrefix(mt.RightDesc.RangeID).PrefixEnd(),
 				})
+				// Merges the prior read summary from the RHS over to the LHS,
+				// which ensures that the current and all future leaseholders
+				// on the joint range respect reads served on the RHS.
+				latchSpans.AddNonMVCC(spanset.SpanReadWrite, roachpb.Span{
+					Key: keys.RangePriorReadSummaryKey(mt.LeftDesc.RangeID),
+				})
 			}
 		}
 	}
@@ -1103,6 +1110,30 @@ func mergeTrigger(
 		return result.Result{}, err
 	}
 
+	// If we collected a read summary from the right-hand side when freezing it,
+	// merge that summary into the left-hand side's prior read summary. In the
+	// usual case, the RightReadSummary in the MergeTrigger will be used to
+	// update the left-hand side's leaseholder's timestamp cache when applying
+	// the merge trigger's Raft log entry. However, if the left-hand side's
+	// leaseholder hears about the merge through a Raft snapshot, the merge
+	// trigger will not be available, so it will need to use the range's prior
+	// read summary to update its timestamp cache to ensure that it does not
+	// serve any writes that invalidate previous reads served on the right-hand
+	// side range. See TestStoreRangeMergeTimestampCache for an example of where
+	// this behavior is necessary.
+	readSumActive := rec.ClusterSettings().Version.IsActive(ctx, clusterversion.PriorReadSummaries)
+	if merge.RightReadSummary != nil && readSumActive {
+		mergedSum := merge.RightReadSummary.Clone()
+		if priorSum, err := readsummary.Load(ctx, batch, rec.GetRangeID()); err != nil {
+			return result.Result{}, err
+		} else if priorSum != nil {
+			mergedSum.Merge(*priorSum)
+		}
+		if err := readsummary.Set(ctx, batch, rec.GetRangeID(), ms, mergedSum); err != nil {
+			return result.Result{}, err
+		}
+	}
+
 	// The stats for the merged range are the sum of the LHS and RHS stats, less
 	// the RHS's replicated range ID stats. The only replicated range ID keys we
 	// copy from the RHS are the keys in the abort span, and we've already

pkg/kv/kvserver/batcheval/cmd_lease.go

@@ -14,8 +14,11 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/batcheval/result"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary/rspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
@@ -52,6 +55,7 @@ func evalNewLease(
 	ms *enginepb.MVCCStats,
 	lease roachpb.Lease,
 	prevLease roachpb.Lease,
+	priorReadSum *rspb.ReadSummary,
 	isExtension bool,
 	isTransfer bool,
 ) (result.Result, error) {
@@ -125,6 +129,15 @@ func evalNewLease(
 	}
 	pd.Replicated.PrevLeaseProposal = prevLease.ProposedTS
 
+	// If we're setting a new prior read summary, store it to disk & in-memory.
+	readSumActive := rec.ClusterSettings().Version.IsActive(ctx, clusterversion.PriorReadSummaries)
+	if priorReadSum != nil && readSumActive {
+		if err := readsummary.Set(ctx, readWriter, rec.GetRangeID(), ms, priorReadSum); err != nil {
+			return newFailedLeaseTrigger(isTransfer), err
+		}
+		pd.Replicated.PriorReadSummary = priorReadSum
+	}
+
 	pd.Local.Metrics = new(result.Metrics)
 	if isTransfer {
 		pd.Local.Metrics.LeaseTransferSuccess = 1

pkg/kv/kvserver/batcheval/cmd_lease_request.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/batcheval/result"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary/rspb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
@@ -33,6 +34,7 @@ func declareKeysRequestLease(
 	// (see concurrency.shouldAcquireLatches). However, we continue to
 	// declare the keys in order to appease SpanSet assertions under race.
 	latchSpans.AddNonMVCC(spanset.SpanReadWrite, roachpb.Span{Key: keys.RangeLeaseKey(rs.GetRangeID())})
+	latchSpans.AddNonMVCC(spanset.SpanReadWrite, roachpb.Span{Key: keys.RangePriorReadSummaryKey(rs.GetRangeID())})
 	latchSpans.AddNonMVCC(spanset.SpanReadOnly, roachpb.Span{Key: keys.RangeDescriptorKey(rs.GetStartKey())})
 }
 
@@ -134,6 +136,22 @@ func RequestLease(
 		return newFailedLeaseTrigger(false /* isTransfer */), rErr
 	}
 	newLease.Start = effectiveStart
+
+	var priorReadSum *rspb.ReadSummary
+	if !prevLease.Equivalent(newLease) {
+		// If the new lease is not equivalent to the old lease (i.e. either the
+		// lease is changing hands or the leaseholder restarted), construct a
+		// read summary to instruct the new leaseholder on how to update its
+		// timestamp cache. Since we are not the leaseholder ourselves, we must
+		// pessimistically assume that prior leaseholders served reads all the
+		// way up to the start of the new lease.
+		//
+		// NB: this is equivalent to the leaseChangingHands condition in
+		// leasePostApplyLocked.
+		worstCaseSum := rspb.FromTimestamp(newLease.Start.ToTimestamp())
+		priorReadSum = &worstCaseSum
+	}
+
 	return evalNewLease(ctx, cArgs.EvalCtx, readWriter, cArgs.Stats,
-		newLease, prevLease, isExtension, false /* isTransfer */)
+		newLease, prevLease, priorReadSum, isExtension, false /* isTransfer */)
 }

pkg/kv/kvserver/batcheval/cmd_lease_test.go

@@ -16,6 +16,8 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary/rspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -171,53 +173,78 @@ func TestLeaseTransferForwardsStartTime(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	testutils.RunTrueAndFalse(t, "epoch", func(t *testing.T, epoch bool) {
-		ctx := context.Background()
-		db := storage.NewDefaultInMem()
-		defer db.Close()
-		batch := db.NewBatch()
-		defer batch.Close()
+		testutils.RunTrueAndFalse(t, "served-future-reads", func(t *testing.T, servedFutureReads bool) {
+			ctx := context.Background()
+			db := storage.NewDefaultInMem()
+			defer db.Close()
+			batch := db.NewBatch()
+			defer batch.Close()
 
-		replicas := []roachpb.ReplicaDescriptor{
-			{NodeID: 1, StoreID: 1, Type: roachpb.ReplicaTypeVoterFull(), ReplicaID: 1},
-			{NodeID: 2, StoreID: 2, Type: roachpb.ReplicaTypeVoterFull(), ReplicaID: 2},
-		}
-		desc := roachpb.RangeDescriptor{}
-		desc.SetReplicas(roachpb.MakeReplicaSet(replicas))
-		manual := hlc.NewManualClock(123)
-		clock := hlc.NewClock(manual.UnixNano, time.Nanosecond)
+			replicas := []roachpb.ReplicaDescriptor{
+				{NodeID: 1, StoreID: 1, Type: roachpb.ReplicaTypeVoterFull(), ReplicaID: 1},
+				{NodeID: 2, StoreID: 2, Type: roachpb.ReplicaTypeVoterFull(), ReplicaID: 2},
+			}
+			desc := roachpb.RangeDescriptor{}
+			desc.SetReplicas(roachpb.MakeReplicaSet(replicas))
+			manual := hlc.NewManualClock(123)
+			clock := hlc.NewClock(manual.UnixNano, time.Nanosecond)
 
-		nextLease := roachpb.Lease{
-			Replica: replicas[1],
-			Start:   clock.NowAsClockTimestamp(),
-		}
-		if epoch {
-			nextLease.Epoch = 1
-		} else {
-			exp := nextLease.Start.ToTimestamp().Add(9*time.Second.Nanoseconds(), 0)
-			nextLease.Expiration = &exp
-		}
-		cArgs := CommandArgs{
-			EvalCtx: (&MockEvalCtx{
-				StoreID: 1,
-				Desc:    &desc,
-				Clock:   clock,
-			}).EvalContext(),
-			Args: &roachpb.TransferLeaseRequest{
-				Lease: nextLease,
-			},
-		}
+			nextLease := roachpb.Lease{
+				Replica: replicas[1],
+				Start:   clock.NowAsClockTimestamp(),
+			}
+			if epoch {
+				nextLease.Epoch = 1
+			} else {
+				exp := nextLease.Start.ToTimestamp().Add(9*time.Second.Nanoseconds(), 0)
+				nextLease.Expiration = &exp
+			}
+
+			var maxPriorReadTS hlc.Timestamp
+			if servedFutureReads {
+				maxPriorReadTS = nextLease.Start.ToTimestamp().Add(1*time.Second.Nanoseconds(), 0)
+			} else {
+				maxPriorReadTS = nextLease.Start.ToTimestamp().Add(-2*time.Second.Nanoseconds(), 0)
+			}
+			currentReadSummary := rspb.FromTimestamp(maxPriorReadTS)
+
+			cArgs := CommandArgs{
+				EvalCtx: (&MockEvalCtx{
+					StoreID:            1,
+					Desc:               &desc,
+					Clock:              clock,
+					CurrentReadSummary: currentReadSummary,
+				}).EvalContext(),
+				Args: &roachpb.TransferLeaseRequest{
+					Lease: nextLease,
+				},
+			}
+
+			manual.Increment(1000)
+			beforeEval := clock.NowAsClockTimestamp()
 
-		manual.Increment(1000)
-		beforeEval := clock.NowAsClockTimestamp()
+			res, err := TransferLease(ctx, batch, cArgs, nil)
+			require.NoError(t, err)
 
-		res, err := TransferLease(ctx, batch, cArgs, nil)
-		require.NoError(t, err)
+			// The proposed lease start time should be assigned at eval time.
+			propLease := res.Replicated.State.Lease
+			require.NotNil(t, propLease)
+			require.True(t, nextLease.Start.Less(propLease.Start))
+			require.True(t, beforeEval.Less(propLease.Start))
 
-		// The proposed lease start time should be assigned at eval time.
-		propLease := res.Replicated.State.Lease
-		require.NotNil(t, propLease)
-		require.True(t, nextLease.Start.Less(propLease.Start))
-		require.True(t, beforeEval.Less(propLease.Start))
+			// The prior read summary should reflect the maximum read times
+			// served under the current leaseholder.
+			propReadSum, err := readsummary.Load(ctx, batch, desc.RangeID)
+			require.NoError(t, err)
+			require.NotNil(t, propReadSum, "should write prior read summary")
+			if servedFutureReads {
+				require.Equal(t, maxPriorReadTS, propReadSum.Local.LowWater)
+				require.Equal(t, maxPriorReadTS, propReadSum.Global.LowWater)
+			} else {
+				require.Equal(t, propLease.Start.ToTimestamp(), propReadSum.Local.LowWater)
+				require.Equal(t, propLease.Start.ToTimestamp(), propReadSum.Global.LowWater)
+			}
+		})
 	})
 }