From d8ded0c1469894cee88eeb06b948196a0014fccd Mon Sep 17 00:00:00 2001 From: tongjian <1045931706@qq.com> Date: Wed, 15 Jul 2026 22:53:19 +0800 Subject: [PATCH 1/4] server: make EtcdStartTimeout a no-progress timeout startEtcd previously wrapped the whole "wait until etcd is ready" in a fixed context deadline of EtcdStartTimeout (5m). A slow-but-healthy startup that keeps catching up its raft log would be killed once the deadline elapsed, and on process restart it would loop forever. Wait for readiness using the etcd applied index as a liveness signal instead: keep waiting as long as the applied index advances, and only give up when there is no apply progress for EtcdStartTimeout. This tells a slow restore apart from a genuine hang (e.g. a removed member that can never rejoin the quorum), which still fails fast. Parent ctx cancellation is still honored for normal shutdown. Co-Authored-By: Claude Opus 4.8 Signed-off-by: tongjian <1045931706@qq.com> --- server/server.go | 67 +++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 58 insertions(+), 9 deletions(-) diff --git a/server/server.go b/server/server.go index 58e3caa31e..4d60d50d8d 100644 --- a/server/server.go +++ b/server/server.go @@ -115,8 +115,16 @@ const ( ) // EtcdStartTimeout the timeout of the startup 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 + var ( // WithLabelValues is a heavy operation, define variable to avoid call it every time. etcdTermGauge = etcdStateGauge.WithLabelValues("term") @@ -345,9 +353,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,11 +400,11 @@ 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 = s.waitEtcdReady(ctx, etcd); err != nil { + return err } // Start the etcd and HTTP clients, then init the member. @@ -407,7 +412,9 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { 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 +423,48 @@ 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 honors ctx cancellation for normal shutdown. +func (s *Server) waitEtcdReady(ctx context.Context, etcd *embed.Etcd) error { + ticker := time.NewTicker(etcdStartProgressCheckInterval) + defer ticker.Stop() + + lastApplied := etcd.Server.AppliedIndex() + lastProgress := time.Now() + for { + select { + case <-etcd.Server.ReadyNotify(): + return nil + case <-ctx.Done(): + return errs.ErrCancelStartEtcd.FastGenByArgs() + case now := <-ticker.C: + applied := etcd.Server.AppliedIndex() + if applied > lastApplied { + log.Info("etcd is not ready yet but still making apply progress, keep waiting", + zap.Uint64("last-applied-index", lastApplied), + zap.Uint64("applied-index", applied)) + lastApplied = applied + lastProgress = now + continue + } + if stalled := now.Sub(lastProgress); stalled >= EtcdStartTimeout { + 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", EtcdStartTimeout)) + return errs.ErrCancelStartEtcd.FastGenByArgs() + } + } + } +} + func (s *Server) initGRPCServiceLabels() { for name, serviceInfo := range s.grpcServer.GetServiceInfo() { if name == gRPCServiceName { From fc2e8be75dabe5ff12fff6ed9e3e9561b43ea894 Mon Sep 17 00:00:00 2001 From: tongjian <1045931706@qq.com> Date: Thu, 16 Jul 2026 15:39:42 +0800 Subject: [PATCH 2/4] server: add unit tests for etcd start no-progress timeout Extract the wait loop of waitEtcdReady into waitEtcdReadyProgress, which depends only on a ready channel, an applied-index getter and the two durations, so it can be driven deterministically without a real embedded etcd. Cover the four behaviors: ready immediately, keep waiting while the applied index advances (past what a fixed deadline would allow), give up after no apply progress, and honor ctx cancellation. Co-Authored-By: Claude Opus 4.8 Signed-off-by: tongjian <1045931706@qq.com> --- server/etcd_start_test.go | 104 ++++++++++++++++++++++++++++++++++++++ server/server.go | 26 +++++++--- 2 files changed, 124 insertions(+), 6 deletions(-) create mode 100644 server/etcd_start_test.go diff --git a/server/etcd_start_test.go b/server/etcd_start_test.go new file mode 100644 index 0000000000..a27124340f --- /dev/null +++ b/server/etcd_start_test.go @@ -0,0 +1,104 @@ +// 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, constApplied(0), + testCheckInterval, testNoProgressTimeout) + re.NoError(err) + }) + + 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, 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{}), 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()) + go func() { + time.Sleep(2 * testCheckInterval) + cancel() + }() + start := time.Now() + // applied index advances, so only ctx cancellation can end the wait. + err := waitEtcdReadyProgress(ctx, make(chan struct{}), advancingApplied(), + testCheckInterval, testNoProgressTimeout) + re.Error(err) + re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd)) + // Cancellation returns before the no-progress timeout would trigger. + re.Less(time.Since(start), testNoProgressTimeout) + }) +} diff --git a/server/server.go b/server/server.go index 4d60d50d8d..ae13867766 100644 --- a/server/server.go +++ b/server/server.go @@ -433,19 +433,33 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { // from a genuine hang, such as a removed member that can never rejoin the // quorum. It also honors ctx cancellation for normal shutdown. func (s *Server) waitEtcdReady(ctx context.Context, etcd *embed.Etcd) error { - ticker := time.NewTicker(etcdStartProgressCheckInterval) + return waitEtcdReadyProgress(ctx, etcd.Server.ReadyNotify(), etcd.Server.AppliedIndex, + etcdStartProgressCheckInterval, EtcdStartTimeout) +} + +// waitEtcdReadyProgress is the testable core of waitEtcdReady. It blocks until +// ready is signaled, treating appliedIndex as a liveness signal: it keeps +// waiting while appliedIndex advances, and only returns an error when there is +// no apply progress for noProgressTimeout, or when ctx is canceled. +func waitEtcdReadyProgress( + ctx context.Context, + ready <-chan struct{}, + appliedIndex func() uint64, + checkInterval, noProgressTimeout time.Duration, +) error { + ticker := time.NewTicker(checkInterval) defer ticker.Stop() - lastApplied := etcd.Server.AppliedIndex() + lastApplied := appliedIndex() lastProgress := time.Now() for { select { - case <-etcd.Server.ReadyNotify(): + case <-ready: return nil case <-ctx.Done(): return errs.ErrCancelStartEtcd.FastGenByArgs() case now := <-ticker.C: - applied := etcd.Server.AppliedIndex() + applied := appliedIndex() if applied > lastApplied { log.Info("etcd is not ready yet but still making apply progress, keep waiting", zap.Uint64("last-applied-index", lastApplied), @@ -454,11 +468,11 @@ func (s *Server) waitEtcdReady(ctx context.Context, etcd *embed.Etcd) error { lastProgress = now continue } - if stalled := now.Sub(lastProgress); stalled >= EtcdStartTimeout { + 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", EtcdStartTimeout)) + zap.Duration("timeout", noProgressTimeout)) return errs.ErrCancelStartEtcd.FastGenByArgs() } } From 3b75894174593627dfc405ab9c142a7e4a1c8420 Mon Sep 17 00:00:00 2001 From: tongjian <1045931706@qq.com> Date: Thu, 16 Jul 2026 15:48:03 +0800 Subject: [PATCH 3/4] server: fail fast on async etcd startup error; fix godoc Address review feedback on the etcd start no-progress timeout: - Monitor etcd.Err() while waiting for readiness. If the embedded etcd hits a fatal error in the background before ReadyNotify() closes, the error is now returned (wrapped in ErrStartEtcd) immediately instead of stalling for the whole no-progress timeout and returning a generic error. waitEtcdReadyProgress takes the error channel as a parameter so it stays unit-testable. - Start the EtcdStartTimeout GoDoc with the identifier name. - Add a unit test for the async-error path. Co-Authored-By: Claude Opus 4.8 Signed-off-by: tongjian <1045931706@qq.com> --- server/etcd_start_test.go | 24 ++++++++++++++++++++---- server/server.go | 19 ++++++++++++------- 2 files changed, 32 insertions(+), 11 deletions(-) diff --git a/server/etcd_start_test.go b/server/etcd_start_test.go index a27124340f..40896af3a2 100644 --- a/server/etcd_start_test.go +++ b/server/etcd_start_test.go @@ -50,11 +50,27 @@ func TestWaitEtcdReadyProgress(t *testing.T) { re := require.New(t) ready := make(chan struct{}) close(ready) - err := waitEtcdReadyProgress(context.Background(), ready, constApplied(0), + err := waitEtcdReadyProgress(context.Background(), ready, 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{}), 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("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; @@ -66,7 +82,7 @@ func TestWaitEtcdReadyProgress(t *testing.T) { close(ready) }() start := time.Now() - err := waitEtcdReadyProgress(context.Background(), ready, advancingApplied(), + err := waitEtcdReadyProgress(context.Background(), ready, nil, advancingApplied(), testCheckInterval, testNoProgressTimeout) re.NoError(err) // It must have outlived the no-progress window instead of bailing at it. @@ -77,7 +93,7 @@ func TestWaitEtcdReadyProgress(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{}), constApplied(7), + err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), nil, constApplied(7), testCheckInterval, testNoProgressTimeout) re.Error(err) re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd)) @@ -94,7 +110,7 @@ func TestWaitEtcdReadyProgress(t *testing.T) { }() start := time.Now() // applied index advances, so only ctx cancellation can end the wait. - err := waitEtcdReadyProgress(ctx, make(chan struct{}), advancingApplied(), + err := waitEtcdReadyProgress(ctx, make(chan struct{}), nil, advancingApplied(), testCheckInterval, testNoProgressTimeout) re.Error(err) re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd)) diff --git a/server/server.go b/server/server.go index ae13867766..90643ee027 100644 --- a/server/server.go +++ b/server/server.go @@ -114,11 +114,11 @@ const ( lostPDLeaderReElectionFactor = 10 ) -// EtcdStartTimeout the timeout of the startup 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. +// 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 @@ -433,17 +433,20 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { // from a genuine hang, such as a removed member that can never rejoin the // quorum. It also honors ctx cancellation for normal shutdown. func (s *Server) waitEtcdReady(ctx context.Context, etcd *embed.Etcd) error { - return waitEtcdReadyProgress(ctx, etcd.Server.ReadyNotify(), etcd.Server.AppliedIndex, + return waitEtcdReadyProgress(ctx, etcd.Server.ReadyNotify(), etcd.Err(), etcd.Server.AppliedIndex, etcdStartProgressCheckInterval, EtcdStartTimeout) } // waitEtcdReadyProgress is the testable core of waitEtcdReady. It blocks until // ready is signaled, treating appliedIndex as a liveness signal: it keeps // waiting while appliedIndex advances, and only returns an error when there is -// no apply progress for noProgressTimeout, or when ctx is canceled. +// no apply progress for noProgressTimeout, or when ctx is canceled. It also +// fails fast if etcd reports an asynchronous startup error on errCh, which would +// otherwise leave ready unsignaled and stall the wait for the whole timeout. func waitEtcdReadyProgress( ctx context.Context, ready <-chan struct{}, + errCh <-chan error, appliedIndex func() uint64, checkInterval, noProgressTimeout time.Duration, ) error { @@ -456,6 +459,8 @@ func waitEtcdReadyProgress( select { case <-ready: return nil + case err := <-errCh: + return errs.ErrStartEtcd.Wrap(err).GenWithStackByCause() case <-ctx.Done(): return errs.ErrCancelStartEtcd.FastGenByArgs() case now := <-ticker.C: From ab2cee00c779601958c81c74bb78f65c4844a926 Mon Sep 17 00:00:00 2001 From: tongjian <1045931706@qq.com> Date: Thu, 16 Jul 2026 17:52:12 +0800 Subject: [PATCH 4/4] server: address etcd start wait review feedback Follow-up on reviewer feedback for the etcd start no-progress timeout: - Drop the unused *Server receiver on waitEtcdReady (revive unused-receiver, which was failing the static check) by making it a package-level function. - Also monitor etcd.Server.StopNotify(): an internal EtcdServer.run failure can stop the server without publishing on etcd.Err(), which previously got misreported as a no-progress timeout. On stop, preserve the underlying etcd.Err() cause when available. - Sample time.Now() when reading the applied index instead of reusing the ticker's scheduled timestamp, so a tick buffered during a scheduling/GC pause cannot trip a false no-progress timeout. - Rate-limit the progress log to at most once per 30s (a slow recovery can run for many minutes), instead of logging on every poll. - Test: drop the flaky timing upper-bound assertion in the cancellation case, add defer cancel(), and cover the server-stopped path. Co-Authored-By: Claude Opus 4.8 Signed-off-by: tongjian <1045931706@qq.com> --- server/etcd_start_test.go | 28 ++++++++++++----- server/server.go | 64 ++++++++++++++++++++++++++++----------- 2 files changed, 66 insertions(+), 26 deletions(-) diff --git a/server/etcd_start_test.go b/server/etcd_start_test.go index 40896af3a2..bbace96f37 100644 --- a/server/etcd_start_test.go +++ b/server/etcd_start_test.go @@ -50,7 +50,7 @@ func TestWaitEtcdReadyProgress(t *testing.T) { re := require.New(t) ready := make(chan struct{}) close(ready) - err := waitEtcdReadyProgress(context.Background(), ready, nil, constApplied(0), + err := waitEtcdReadyProgress(context.Background(), ready, nil, nil, constApplied(0), testCheckInterval, testNoProgressTimeout) re.NoError(err) }) @@ -61,7 +61,7 @@ func TestWaitEtcdReadyProgress(t *testing.T) { errCh := make(chan error, 1) errCh <- errors.New("boom") start := time.Now() - err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), errCh, constApplied(0), + 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. @@ -71,6 +71,20 @@ func TestWaitEtcdReadyProgress(t *testing.T) { 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; @@ -82,7 +96,7 @@ func TestWaitEtcdReadyProgress(t *testing.T) { close(ready) }() start := time.Now() - err := waitEtcdReadyProgress(context.Background(), ready, nil, advancingApplied(), + 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. @@ -93,7 +107,7 @@ func TestWaitEtcdReadyProgress(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, constApplied(7), + err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), nil, nil, constApplied(7), testCheckInterval, testNoProgressTimeout) re.Error(err) re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd)) @@ -104,17 +118,15 @@ func TestWaitEtcdReadyProgress(t *testing.T) { 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() }() - start := time.Now() // applied index advances, so only ctx cancellation can end the wait. - err := waitEtcdReadyProgress(ctx, make(chan struct{}), nil, advancingApplied(), + err := waitEtcdReadyProgress(ctx, make(chan struct{}), nil, nil, advancingApplied(), testCheckInterval, testNoProgressTimeout) re.Error(err) re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd)) - // Cancellation returns before the no-progress timeout would trigger. - re.Less(time.Since(start), testNoProgressTimeout) }) } diff --git a/server/server.go b/server/server.go index 90643ee027..7850927827 100644 --- a/server/server.go +++ b/server/server.go @@ -125,6 +125,10 @@ var EtcdStartTimeout = time.Minute * 5 // 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") @@ -403,7 +407,7 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { // 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 = s.waitEtcdReady(ctx, etcd); err != nil { + if err = waitEtcdReady(ctx, etcd); err != nil { return err } @@ -431,21 +435,26 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { // 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 honors ctx cancellation for normal shutdown. -func (s *Server) waitEtcdReady(ctx context.Context, etcd *embed.Etcd) error { - return waitEtcdReadyProgress(ctx, etcd.Server.ReadyNotify(), etcd.Err(), etcd.Server.AppliedIndex, - etcdStartProgressCheckInterval, EtcdStartTimeout) -} - -// waitEtcdReadyProgress is the testable core of waitEtcdReady. It blocks until -// ready is signaled, treating appliedIndex as a liveness signal: it keeps -// waiting while appliedIndex advances, and only returns an error when there is -// no apply progress for noProgressTimeout, or when ctx is canceled. It also -// fails fast if etcd reports an asynchronous startup error on errCh, which would -// otherwise leave ready unsignaled and stall the wait for the whole timeout. +// 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 <-chan struct{}, + ready, stopped <-chan struct{}, errCh <-chan error, appliedIndex func() uint64, checkInterval, noProgressTimeout time.Duration, @@ -455,22 +464,41 @@ func waitEtcdReadyProgress( 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 now := <-ticker.C: + 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 { - log.Info("etcd is not ready yet but still making apply progress, keep waiting", - zap.Uint64("last-applied-index", lastApplied), - zap.Uint64("applied-index", applied)) 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 {