Merge #42939

42939: storage: respect closed timestamp in tryReproposeWithNewLeaseIndex r=nvanbenschoten a=tbg

Prior to this commit, tryReproposeWithNewLeaseIndex would not consult
with the closed timestamp tracker. This meant that a proposal may end up
with a new lease index at a timestamp that consumers of closed timestamp
updates had already considered immutable. In other words, this broke
closed timestamp guarantees and would by extension also have the
potential to break CDC's.

This path is fairly rare and it isn't one likely to cause problems
in practice, though it certainly could have.

See #42821 for details.

Fixes #42821

Release note (bug fix): fix a bug which could lead to follower reads or
CDC updates that did not reflect the full set of data at the timestamp.
This bug was never observed in practice and should be rare to cause
issues, one of the necessary ingredients being an aggressive closed
timestamp interval.

Co-authored-by: Tobias Schottdorf <tobias.schottdorf@gmail.com>

pkg/storage/replica_application_result.go

@@ -14,6 +14,7 @@ import (
 	"context"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/storage/closedts/ctpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/storage/storagebase"
 	"github.com/cockroachdb/cockroach/pkg/storage/storagepb"
@@ -149,6 +150,7 @@ func (r *Replica) prepareLocalResult(ctx context.Context, cmd *replicatedCmd) {
 			// the proposal would be a user-visible error.
 			pErr = r.tryReproposeWithNewLeaseIndex(ctx, cmd)
 			if pErr != nil {
+				log.Warningf(ctx, "failed to repropose with new lease index: %s", pErr)
 				cmd.response.Err = pErr
 			} else {
 				// Unbind the entry's local proposal because we just succeeded
@@ -207,13 +209,33 @@ func (r *Replica) tryReproposeWithNewLeaseIndex(
 		// succeeding in the Raft log for a given command.
 		return nil
 	}
+
+	minTS, untrack := r.store.cfg.ClosedTimestamp.Tracker.Track(ctx)
+	defer untrack(ctx, 0, 0, 0) // covers all error paths below
+	// NB: p.Request.Timestamp reflects the action of ba.SetActiveTimestamp.
+	if p.Request.Timestamp.Less(minTS) {
+		// The tracker wants us to forward the request timestamp, but we can't
+		// do that without re-evaluating, so give up. The error returned here
+		// will go to back to DistSender, so send something it can digest.
+		lhErr := roachpb.NewError(newNotLeaseHolderError(
+			r.mu.state.Lease,
+			r.store.StoreID(),
+			r.mu.state.Desc,
+		))
+
+		return lhErr
+	}
 	// Some tests check for this log message in the trace.
 	log.VEventf(ctx, 2, "retry: proposalIllegalLeaseIndex")
+
 	maxLeaseIndex, pErr := r.propose(ctx, p)
 	if pErr != nil {
-		log.Warningf(ctx, "failed to repropose with new lease index: %s", pErr)
 		return pErr
 	}
+	// NB: The caller already promises that the lease check succeeded, meaning
+	// the sequence numbers match, implying that the lease epoch hasn't changed
+	// from what it was under the proposal-time lease.
+	untrack(ctx, ctpb.Epoch(r.mu.state.Lease.Epoch), r.RangeID, ctpb.LAI(maxLeaseIndex))
 	log.VEventf(ctx, 2, "reproposed command %x at maxLeaseIndex=%d", cmd.idKey, maxLeaseIndex)
 	return nil
 }