Skip to content

Commit

Permalink
storage: fix quiescence when not leaseholder
Browse files Browse the repository at this point in the history
Fix check for whether we own a valid lease.

Fixes cockroachdb#17741
  • Loading branch information
petermattis authored and bdarnell committed Sep 2, 2017
1 parent 8633ca4 commit ddc67c1
Showing 1 changed file with 11 additions and 6 deletions.
17 changes: 11 additions & 6 deletions pkg/storage/replica.go
Original file line number Diff line number Diff line change
Expand Up @@ -1142,6 +1142,10 @@ func (r *Replica) getLeaseRLocked() (roachpb.Lease, *roachpb.Lease) {
func (r *Replica) OwnsValidLease(ts hlc.Timestamp) bool {
r.mu.RLock()
defer r.mu.RUnlock()
return r.ownsValidLeaseRLocked(ts)
}

func (r *Replica) ownsValidLeaseRLocked(ts hlc.Timestamp) bool {
return r.mu.state.Lease.OwnedBy(r.store.StoreID()) &&
r.leaseStatus(*r.mu.state.Lease, ts, r.mu.minLeaseProposedTS).state == leaseValid
}
Expand Down Expand Up @@ -3662,17 +3666,18 @@ func (r *Replica) maybeQuiesceLocked() bool {
// Only quiesce if this replica is the leaseholder as well;
// otherwise the replica which is the valid leaseholder may have
// pending commands which it's waiting on this leader to propose.
if l := *r.mu.state.Lease; !l.OwnedBy(r.store.StoreID()) &&
r.isLeaseValidRLocked(l, r.store.Clock().Now()) {
if now := r.store.Clock().Now(); !r.ownsValidLeaseRLocked(now) {
if log.V(4) {
log.Infof(ctx, "not quiescing: not leaseholder")
}
// Try to correct leader-not-leaseholder condition, if encountered,
// assuming the leaseholder is caught up to the commit index.
if pr, ok := status.Progress[uint64(l.Replica.ReplicaID)]; ok && pr.Match >= status.Commit {
log.VEventf(ctx, 1, "transferring raft leadership to replica ID %v", l.Replica.ReplicaID)
r.store.metrics.RangeRaftLeaderTransfers.Inc(1)
r.mu.internalRaftGroup.TransferLeader(uint64(l.Replica.ReplicaID))
if l := *r.mu.state.Lease; r.isLeaseValidRLocked(l, now) {
if pr, ok := status.Progress[uint64(l.Replica.ReplicaID)]; ok && pr.Match >= status.Commit {
log.VEventf(ctx, 1, "transferring raft leadership to replica ID %v", l.Replica.ReplicaID)
r.store.metrics.RangeRaftLeaderTransfers.Inc(1)
r.mu.internalRaftGroup.TransferLeader(uint64(l.Replica.ReplicaID))
}
}
return false
}
Expand Down

0 comments on commit ddc67c1

Please sign in to comment.