server: make EtcdStartTimeout a no-progress timeout

server/etcd_start_test.go

@@ -0,0 +1,132 @@
+// Copyright 2026 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/pingcap/errors"
+
+	"github.com/tikv/pd/pkg/errs"
+)
+
+const (
+	testCheckInterval     = 5 * time.Millisecond
+	testNoProgressTimeout = 50 * time.Millisecond
+)
+
+// constApplied returns an appliedIndex getter that never advances, simulating a
+// hung etcd that makes no apply progress.
+func constApplied(v uint64) func() uint64 {
+	return func() uint64 { return v }
+}
+
+// advancingApplied returns an appliedIndex getter that advances on every call,
+// simulating a healthy etcd that keeps applying its raft log.
+func advancingApplied() func() uint64 {
+	var idx atomic.Uint64
+	return func() uint64 { return idx.Add(1) }
+}
+
+func TestWaitEtcdReadyProgress(t *testing.T) {
+	t.Run("ready immediately", func(t *testing.T) {
+		re := require.New(t)
+		ready := make(chan struct{})
+		close(ready)
+		err := waitEtcdReadyProgress(context.Background(), ready, nil, nil, constApplied(0),
+			testCheckInterval, testNoProgressTimeout)
+		re.NoError(err)
+	})
+
+	t.Run("fails fast on async etcd error", func(t *testing.T) {
+		re := require.New(t)
+		// etcd reports a fatal startup error before ever becoming ready.
+		errCh := make(chan error, 1)
+		errCh <- errors.New("boom")
+		start := time.Now()
+		err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), nil, errCh, constApplied(0),
+			testCheckInterval, testNoProgressTimeout)
+		re.Error(err)
+		// The real cause is wrapped in ErrStartEtcd, not swallowed by a timeout.
+		re.ErrorContains(err, "PD:etcd:ErrStartEtcd")
+		re.ErrorContains(err, "boom")
+		// It must not wait out the whole no-progress window.
+		re.Less(time.Since(start), testNoProgressTimeout)
+	})
+
+	t.Run("fails fast when etcd server stops before ready", func(t *testing.T) {
+		re := require.New(t)
+		// The server terminates before becoming ready without publishing an error
+		// on errCh (e.g. an internal EtcdServer.run failure).
+		stopped := make(chan struct{})
+		close(stopped)
+		start := time.Now()
+		err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), stopped, nil, constApplied(0),
+			testCheckInterval, testNoProgressTimeout)
+		re.Error(err)
+		re.ErrorContains(err, "PD:etcd:ErrStartEtcd")
+		re.Less(time.Since(start), testNoProgressTimeout)
+	})
+
+	t.Run("keeps waiting while applied index advances", func(t *testing.T) {
+		re := require.New(t)
+		// ready only fires well after the no-progress timeout would have elapsed;
+		// a fixed deadline would have killed this, but steady apply progress must
+		// keep the wait alive.
+		ready := make(chan struct{})
+		go func() {
+			time.Sleep(4 * testNoProgressTimeout)
+			close(ready)
+		}()
+		start := time.Now()
+		err := waitEtcdReadyProgress(context.Background(), ready, nil, nil, advancingApplied(),
+			testCheckInterval, testNoProgressTimeout)
+		re.NoError(err)
+		// It must have outlived the no-progress window instead of bailing at it.
+		re.GreaterOrEqual(time.Since(start), 3*testNoProgressTimeout)
+	})
+
+	t.Run("gives up when no apply progress", func(t *testing.T) {
+		re := require.New(t)
+		// ready never fires and applied index is stuck: a genuine hang.
+		start := time.Now()
+		err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), nil, nil, constApplied(7),
+			testCheckInterval, testNoProgressTimeout)
+		re.Error(err)
+		re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd))
+		// It should wait roughly the whole no-progress window before giving up.
+		re.GreaterOrEqual(time.Since(start), testNoProgressTimeout-testCheckInterval)
+	})
+
+	t.Run("honors ctx cancellation", func(t *testing.T) {
+		re := require.New(t)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		go func() {
+			time.Sleep(2 * testCheckInterval)
+			cancel()
+		}()
+		// applied index advances, so only ctx cancellation can end the wait.
+		err := waitEtcdReadyProgress(ctx, make(chan struct{}), nil, nil, advancingApplied(),
+			testCheckInterval, testNoProgressTimeout)
+		re.Error(err)
+		re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd))
+	})
+}

server/server.go

@@ -114,9 +114,21 @@ const (
 	lostPDLeaderReElectionFactor = 10
 )
 
-// EtcdStartTimeout the timeout of the startup etcd.
+// EtcdStartTimeout is the timeout for starting the embedded etcd. It is used as
+// a no-progress timeout rather than a fixed overall deadline: startEtcd only
+// gives up when etcd makes no apply progress for this duration, so a
+// slow-but-healthy startup that keeps catching up its raft log is not killed.
+// See waitEtcdReady.
 var EtcdStartTimeout = time.Minute * 5
 
+// etcdStartProgressCheckInterval is how often startEtcd polls the embedded
+// etcd applied index to tell a slow-but-progressing startup apart from a hang.
+const etcdStartProgressCheckInterval = time.Second
+
+// etcdStartProgressLogInterval rate-limits the "still making progress" log, so a
+// recovery that runs for many minutes does not emit one line per poll.
+const etcdStartProgressLogInterval = 30 * time.Second
+
 var (
 	// WithLabelValues is a heavy operation, define variable to avoid call it every time.
 	etcdTermGauge           = etcdStateGauge.WithLabelValues("term")
@@ -345,9 +357,6 @@ func CreateServer(ctx context.Context, cfg *config.Config, services []string, le
 }
 
 func (s *Server) startEtcd(ctx context.Context) (retErr error) {
-	newCtx, cancel := context.WithTimeout(ctx, EtcdStartTimeout)
-	defer cancel()
-
 	etcd, err := embed.StartEtcd(s.etcdCfg)
 	if err != nil {
 		return errs.ErrStartEtcd.Wrap(err).GenWithStackByCause()
@@ -395,19 +404,21 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) {
 		return err
 	}
 
-	select {
-	// Wait etcd until it is ready to use
-	case <-etcd.Server.ReadyNotify():
-	case <-newCtx.Done():
-		return errs.ErrCancelStartEtcd.FastGenByArgs()
+	// Wait until etcd is ready to use. Instead of a fixed overall deadline, only
+	// give up when etcd makes no apply progress for EtcdStartTimeout, so a slow
+	// but healthy startup that is still catching up its raft log is not killed.
+	if err = waitEtcdReady(ctx, etcd); err != nil {
+		return err
 	}
 
 	// Start the etcd and HTTP clients, then init the member.
 	err = s.startClient()
 	if err != nil {
 		return err
 	}
-	err = s.initMember(newCtx, etcd)
+	initCtx, cancel := context.WithTimeout(ctx, EtcdStartTimeout)
+	defer cancel()
+	err = s.initMember(initCtx, etcd)
 	if err != nil {
 		return err
 	}
@@ -416,6 +427,91 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) {
 	return nil
 }
 
+// waitEtcdReady blocks until the embedded etcd is ready to serve requests.
+//
+// Rather than bounding the wait by a fixed deadline, it treats the etcd applied
+// index as a liveness signal: as long as the applied index keeps advancing, the
+// startup is still making progress (e.g. applying its raft log while catching
+// up) and we keep waiting. Only when there is no apply progress for
+// EtcdStartTimeout do we give up. This distinguishes a slow-but-healthy startup
+// from a genuine hang, such as a removed member that can never rejoin the
+// quorum. It also fails fast when etcd reports a startup error or stops before
+// becoming ready, and honors ctx cancellation for normal shutdown.
+func waitEtcdReady(ctx context.Context, etcd *embed.Etcd) error {
+	return waitEtcdReadyProgress(ctx, etcd.Server.ReadyNotify(), etcd.Server.StopNotify(),
+		etcd.Err(), etcd.Server.AppliedIndex, etcdStartProgressCheckInterval, EtcdStartTimeout)
+}
+
+// waitEtcdReadyProgress is the testable core of waitEtcdReady, decoupled from
+// *embed.Etcd so it can be driven without a real server. It keeps waiting while
+// appliedIndex advances and reports a no-progress failure only when appliedIndex
+// does not advance for noProgressTimeout. It fails fast when:
+//   - etcd reports an asynchronous startup error on errCh (e.g. listener or
+//     serving failures);
+//   - the etcd server stops before becoming ready (stopped), e.g. an internal
+//     EtcdServer.run failure that may not surface on errCh — the underlying
+//     error is preserved when etcd published one;
+//   - ctx is canceled (normal shutdown).
+func waitEtcdReadyProgress(
+	ctx context.Context,
+	ready, stopped <-chan struct{},
+	errCh <-chan error,
+	appliedIndex func() uint64,
+	checkInterval, noProgressTimeout time.Duration,
+) error {
+	ticker := time.NewTicker(checkInterval)
+	defer ticker.Stop()
+
+	lastApplied := appliedIndex()
+	lastProgress := time.Now()
+	lastLog := lastProgress
+	for {
+		select {
+		case <-ready:
+			return nil
+		case err := <-errCh:
+			return errs.ErrStartEtcd.Wrap(err).GenWithStackByCause()
+		case <-stopped:
+			// The server terminated before becoming ready; surface the real
+			// cause if etcd published one, otherwise report the stop itself.
+			select {
+			case err := <-errCh:
+				return errs.ErrStartEtcd.Wrap(err).GenWithStackByCause()
+			default:
+				return errs.ErrStartEtcd.GenWithStackByArgs()
+			}
+		case <-ctx.Done():
+			return errs.ErrCancelStartEtcd.FastGenByArgs()
+		case <-ticker.C:
+			applied := appliedIndex()
+			// Sample the wall clock here rather than reusing the ticker's
+			// scheduled time: a tick buffered during a scheduling or GC pause
+			// carries a stale timestamp that could otherwise trip a false
+			// no-progress timeout on the following tick.
+			now := time.Now()
+			if applied > lastApplied {
+				lastApplied = applied
+				lastProgress = now
+				// Rate-limit: this path can run for many minutes during a slow
+				// recovery, so avoid logging on every poll.
+				if now.Sub(lastLog) >= etcdStartProgressLogInterval {
+					log.Info("etcd is not ready yet but still making apply progress, keep waiting",
+						zap.Uint64("applied-index", applied))
+					lastLog = now
+				}
+				continue
+			}
+			if stalled := now.Sub(lastProgress); stalled >= noProgressTimeout {
+				log.Warn("etcd is not ready and made no apply progress, stop waiting",
+					zap.Uint64("applied-index", applied),
+					zap.Duration("no-progress-duration", stalled),
+					zap.Duration("timeout", noProgressTimeout))
+				return errs.ErrCancelStartEtcd.FastGenByArgs()
+			}
+		}
+	}
+}
+
 func (s *Server) initGRPCServiceLabels() {
 	for name, serviceInfo := range s.grpcServer.GetServiceInfo() {
 		if name == gRPCServiceName {