support linearizable renew lease

etcd-io · Feb 12, 2022 · a641fde · a641fde
1 parent bba3937
commit a641fde
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 4 deletions.
diff --git a/server/etcdserver/v3_server.go b/server/etcdserver/v3_server.go
@@ -270,6 +270,16 @@ func (s *EtcdServer) LeaseGrant(ctx context.Context, r *pb.LeaseGrantRequest) (*
 	return resp.(*pb.LeaseGrantResponse), nil
 }
 
+func (s *EtcdServer) waitApplyPendingCommits() error {
+	select {
+	case <-s.ApplyWait():
+	case <-time.After(leasehttp.ApplyTimeout):
+		return ErrTimeout
+	}
+
+	return nil
+}
+
 func (s *EtcdServer) LeaseRevoke(ctx context.Context, r *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error) {
 	resp, err := s.raftRequestOnce(ctx, pb.InternalRaftRequest{LeaseRevoke: r})
 	if err != nil {
@@ -279,6 +289,12 @@ func (s *EtcdServer) LeaseRevoke(ctx context.Context, r *pb.LeaseRevokeRequest)
 }
 
 func (s *EtcdServer) LeaseRenew(ctx context.Context, id lease.LeaseID) (int64, error) {
+	if s.isLeader() {
+		if err := s.waitApplyPendingCommits(); err != nil {
+			return 0, err
+		}
+	}
+
 	ttl, err := s.lessor.Renew(id)
 	if err == nil { // already requested to primary lessor(leader)
 		return ttl, nil
@@ -314,7 +330,10 @@ func (s *EtcdServer) LeaseRenew(ctx context.Context, id lease.LeaseID) (int64, e
 }
 
 func (s *EtcdServer) LeaseTimeToLive(ctx context.Context, r *pb.LeaseTimeToLiveRequest) (*pb.LeaseTimeToLiveResponse, error) {
-	if s.Leader() == s.ID() {
+	if s.isLeader() {
+		if err := s.waitApplyPendingCommits(); err != nil {
+			return nil, err
+		}
 		// primary; timetolive directly from leader
 		le := s.lessor.Lookup(lease.LeaseID(r.ID))
 		if le == nil {

diff --git a/server/lease/leasehttp/http.go b/server/lease/leasehttp/http.go
@@ -32,7 +32,7 @@ import (
 var (
 	LeasePrefix         = "/leases"
 	LeaseInternalPrefix = "/leases/internal"
-	applyTimeout        = time.Second
+	ApplyTimeout        = time.Second
 	ErrLeaseHTTPTimeout = errors.New("waiting for node to catch up its applied index has timed out")
 )
 
@@ -69,7 +69,7 @@ func (h *leaseHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		select {
 		case <-h.waitch():
-		case <-time.After(applyTimeout):
+		case <-time.After(ApplyTimeout):
 			http.Error(w, ErrLeaseHTTPTimeout.Error(), http.StatusRequestTimeout)
 			return
 		}
@@ -99,7 +99,7 @@ func (h *leaseHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		select {
 		case <-h.waitch():
-		case <-time.After(applyTimeout):
+		case <-time.After(ApplyTimeout):
 			http.Error(w, ErrLeaseHTTPTimeout.Error(), http.StatusRequestTimeout)
 			return
 		}