Merge cockroachdb#40329

40329: storage: return LeaseRejectedError when learner requests lease r=danhhz a=tbg

We would previously return an unstructured error which would leak to the
client.
The LeaseRejectedError gets turned into a NotLeaseholderError instead,
and DistSender will try another replica.

Fixes cockroachdb#40323.

Note that DistSender doesn't send to learners in the first place, so
it's unclear why we asked a learner to request a lease, but it's likely
that we upgraded the learner to a voter while it was down (the test
below restarts nodes) and it received a request while it was already
a voter but not aware of it.

Release note: None

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

pkg/storage/batcheval/cmd_lease.go

@@ -58,7 +58,15 @@ func checkCanReceiveLease(rec EvalContext) error {
 		//
 		// Since the leaseholder can't remove itself and is a VOTER_FULL, we
 		// also know that in any configuration there's at least one VOTER_FULL.
-		return errors.Errorf(`cannot transfer lease to replica of type %s`, t)
+		//
+		// TODO(tbg): if this code path is hit during a lease transfer (we check
+		// upstream of raft, but this check has false negatives) then we are in
+		// a situation where the leaseholder is a node that has set its
+		// minProposedTS and won't be using its lease any more. Either the setting
+		// of minProposedTS needs to be "reversible" (tricky) or we make the
+		// lease evaluation succeed, though with a lease that's "invalid" so that
+		// a new lease can be requested right after.
+		return errors.Errorf(`replica of type %s cannot hold lease`, t)
 	}
 	return nil
 }

pkg/storage/batcheval/cmd_lease_request.go

@@ -36,6 +36,12 @@ func RequestLease(
 	// newFailedLeaseTrigger() to satisfy stats.
 	args := cArgs.Args.(*roachpb.RequestLeaseRequest)
 
+	prevLease, _ := cArgs.EvalCtx.GetLease()
+	rErr := &roachpb.LeaseRejectedError{
+		Existing:  prevLease,
+		Requested: args.Lease,
+	}
+
 	// For now, don't allow replicas of type LEARNER to be leaseholders. There's
 	// no reason this wouldn't work in principle, but it seems inadvisable. In
 	// particular, learners can't become raft leaders, so we wouldn't be able to
@@ -48,14 +54,8 @@ func RequestLease(
 	// If this check is removed at some point, the filtering of learners on the
 	// sending side would have to be removed as well.
 	if err := checkCanReceiveLease(cArgs.EvalCtx); err != nil {
-		return newFailedLeaseTrigger(false /* isTransfer */), err
-	}
-
-	prevLease, _ := cArgs.EvalCtx.GetLease()
-
-	rErr := &roachpb.LeaseRejectedError{
-		Existing:  prevLease,
-		Requested: args.Lease,
+		rErr.Message = err.Error()
+		return newFailedLeaseTrigger(false /* isTransfer */), rErr
 	}
 
 	// MIGRATION(tschottdorf): needed to apply Raft commands which got proposed

pkg/storage/batcheval/cmd_lease_test.go

@@ -129,9 +129,13 @@ func TestLeaseCommandLearnerReplica(t *testing.T) {
 	// Learners are not allowed to become leaseholders for now, see the comments
 	// in TransferLease and RequestLease.
 	_, err := TransferLease(ctx, nil, cArgs, nil)
-	require.EqualError(t, err, `cannot transfer lease to replica of type LEARNER`)
+	require.EqualError(t, err, `replica of type LEARNER cannot hold lease`)
 
 	cArgs.Args = &roachpb.RequestLeaseRequest{}
 	_, err = RequestLease(ctx, nil, cArgs, nil)
-	require.EqualError(t, err, `cannot transfer lease to replica of type LEARNER`)
+
+	const exp = `cannot replace lease repl=(n0,s0):? seq=0 start=0.000000000,0 exp=<nil> ` +
+		`with repl=(n0,s0):? seq=0 start=0.000000000,0 exp=<nil>: ` +
+		`replica of type LEARNER cannot hold lease`
+	require.EqualError(t, err, exp)
 }

pkg/storage/batcheval/cmd_resolve_intent_test.go

@@ -42,6 +42,7 @@ type mockEvalCtx struct {
 	gcThreshold      hlc.Timestamp
 	term, firstIndex uint64
 	canCreateTxnFn   func() (bool, hlc.Timestamp, roachpb.TransactionAbortedReason)
+	lease            roachpb.Lease
 }
 
 func (m *mockEvalCtx) String() string {
@@ -119,7 +120,7 @@ func (m *mockEvalCtx) GetLastReplicaGCTimestamp(context.Context) (hlc.Timestamp,
 	panic("unimplemented")
 }
 func (m *mockEvalCtx) GetLease() (roachpb.Lease, roachpb.Lease) {
-	panic("unimplemented")
+	return m.lease, roachpb.Lease{}
 }
 
 func TestDeclareKeysResolveIntent(t *testing.T) {