kv: don't serve non-locking, read-write requests on followers

Discovered while investigating a test failure in cockroachdb#59566.

In 278a21b, we shifted from talking about read and write requests to
locking and non-locking requests when deciding whether a request could
be served on a follower. This prevented locking scans and gets from
being served on followers. However, it began letting lone HeartbeatTxn
and EndTxn requests past the old `!IsReadOnly()` check. Luckily, these
were still prevented from being served on followers because they are
only sent in read-write transactions, which were also prevented from
performing follower reads.

Yesterday, in 0ac8ab9, we lifted this second limitation, allowing
read-write transactions to perform follower reads for non-locking
batches. However, this no longer prevented HeartbeatTxn and EndTxn
requests from being routed and served on follower replicas. This
resulted in a pretty disastrous situation where in very rare cases, a
follower was proposing a write under a lease that it did not own.
Luckily, new assertions added in cockroachdb#59566 caught this.

This commit fixes this oversight be re-introducing "read-only" as a
condition for serving follower reads.

Release note: None

pkg/ccl/kvccl/kvfollowerreadsccl/followerreads.go

@@ -92,12 +92,6 @@ func evalFollowerReadOffset(clusterID uuid.UUID, st *cluster.Settings) (time.Dur
  return getFollowerReadLag(st), nil
 }
 
-// batchCanBeEvaluatedOnFollower determines if a batch consists exclusively of
-// requests that can be evaluated on a follower replica.
-func batchCanBeEvaluatedOnFollower(ba roachpb.BatchRequest) bool {
- return ba.Txn != nil && !ba.IsLocking() && ba.IsAllTransactional()
-}
-
 // closedTimestampLikelySufficient determines if a request with a given required
 // frontier timestamp is likely to be below a follower's closed timestamp and
 // serviceable as a follower read were the request to be sent to a follower
@@ -131,7 +125,7 @@ func canSendToFollower(
  ba roachpb.BatchRequest,
 ) bool {
  return checkFollowerReadsEnabled(clusterID, st) &&
- batchCanBeEvaluatedOnFollower(ba) &&
+ kvserver.BatchCanBeEvaluatedOnFollower(ba) &&
  closedTimestampLikelySufficient(st, clock, ctPolicy, ba.Txn.RequiredFrontier())
 }
 

pkg/ccl/kvccl/kvfollowerreadsccl/followerreads_test.go

@@ -121,6 +121,16 @@ func TestCanSendToFollower(t *testing.T) {
  ba: batch(txn(stale), &roachpb.PutRequest{}),
  exp: false,
  },
+ {
+ name: "stale heartbeat txn",
+ ba: batch(txn(stale), &roachpb.HeartbeatTxnRequest{}),
+ exp: false,
+ },
+ {
+ name: "stale end txn",
+ ba: batch(txn(stale), &roachpb.EndTxnRequest{}),
+ exp: false,
+ },
  {
  name: "stale non-txn request",
  ba: batch(txn(stale), &roachpb.QueryTxnRequest{}),
@@ -170,6 +180,18 @@ func TestCanSendToFollower(t *testing.T) {
  ctPolicy: roachpb.LEAD_FOR_GLOBAL_READS,
  exp: false,
  },
+ {
+ name: "stale heartbeat txn, global reads policy",
+ ba: batch(txn(stale), &roachpb.HeartbeatTxnRequest{}),
+ ctPolicy: roachpb.LEAD_FOR_GLOBAL_READS,
+ exp: false,
+ },
+ {
+ name: "stale end txn, global reads policy",
+ ba: batch(txn(stale), &roachpb.EndTxnRequest{}),
+ ctPolicy: roachpb.LEAD_FOR_GLOBAL_READS,
+ exp: false,
+ },
  {
  name: "stale non-txn request, global reads policy",
  ba: batch(txn(stale), &roachpb.QueryTxnRequest{}),

pkg/kv/kvserver/replica.go

@@ -1243,6 +1243,7 @@ func (r *Replica) checkExecutionCanProceed(
  if !r.canServeFollowerReadRLocked(ctx, ba, err) {
  return st, err
  }
+ err = nil // ignoring error
  }
  }
 

pkg/kv/kvserver/replica_follower_read.go

@@ -33,6 +33,25 @@ var FollowerReadsEnabled = settings.RegisterBoolSetting(
  true,
 ).WithPublic()
 
+// BatchCanBeEvaluatedOnFollower determines if a batch consists exclusively of
+// requests that can be evaluated on a follower replica, given a sufficiently
+// advanced closed timestamp.
+func BatchCanBeEvaluatedOnFollower(ba roachpb.BatchRequest) bool {
+ // Explanation of conditions:
+ // 1. the batch needs to be part of a transaction, because non-transactional
+ // batches often rely on the server setting their timestamp. If a follower
+ // with a lagging clock sets their timestamp then they might miss past
+ // writes served at higher timestamps.
+ // 2. each request in the batch needs to be "transactional", because those are
+ // the only ones that have clearly defined semantics when served under the
+ // closed timestamp.
+ // 3. the batch needs to be read-only, because a follower replica cannot
+ // propose writes to Raft.
+ // 4. the batch needs to be non-locking, because unreplicated locks are only
+ // held on the leaseholder.
+ return ba.Txn != nil && ba.IsAllTransactional() && ba.IsReadOnly() && !ba.IsLocking()
+}
+
 // canServeFollowerReadRLocked tests, when a range lease could not be acquired,
 // whether the batch can be served as a follower read despite the error. Only
 // non-locking, read-only requests can be served as follower reads. The batch
@@ -44,7 +63,7 @@ func (r *Replica) canServeFollowerReadRLocked(
  var lErr *roachpb.NotLeaseHolderError
  eligible := errors.As(err, &lErr) &&
  lErr.LeaseHolder != nil && lErr.Lease.Type() == roachpb.LeaseEpoch &&
- (ba.Txn != nil && !ba.IsLocking() && ba.IsAllTransactional()) && // followerreadsccl.batchCanBeEvaluatedOnFollower
+ BatchCanBeEvaluatedOnFollower(*ba) &&
  FollowerReadsEnabled.Get(&r.store.cfg.Settings.SV)
 
  if !eligible {

pkg/kv/kvserver/replica_raft.go

@@ -151,6 +151,10 @@ func (r *Replica) evalAndPropose(
 
  // Attach information about the proposer to the command.
  proposal.command.ProposerLeaseSequence = st.Lease.Sequence
+ // Perform a sanity check that the lease is owned by this replica.
+ if !st.Lease.OwnedBy(r.store.StoreID()) && !ba.IsLeaseRequest() {
+ log.Fatalf(ctx, "cannot propose %s on follower with remotely owned lease %s", ba, st.Lease)
+ }
 
  // Once a command is written to the raft log, it must be loaded into memory
  // and replayed on all replicas. If a command is too big, stop it here. If