cmd/cdc/server: return code 0 on graceful shutdown (pingcap#4564)

close pingcap#4563

Author: pingyu

cmd/cdc/server/server.go

@@ -147,7 +147,8 @@ func (o *options) run(cmd *cobra.Command) error {
 	util.InitSignalHandling(shutdown, cancel)
 
 	err = svr.Run(ctx)
-	if err != nil && !errors.Is(errors.Cause(err), context.Canceled) {
+	isNormalExit := isNormalServerShutdown(err, ctx)
+	if !isNormalExit {
 		log.Error("cdc server exits with error", zap.Error(err))
 	} else {
 		log.Info("cdc server exits normally")
@@ -157,17 +158,33 @@ func (o *options) run(cmd *cobra.Command) error {
 	ticker := time.NewTicker(server.GracefulShutdownTimeout)
 	defer ticker.Stop()
 	go func() {
-		svr.Close(ctx)
+		svr.Close()
 		close(ch)
 	}()
 	select {
 	case <-ch:
 	case <-ticker.C:
 		log.Warn("graceful shutdown timeout, exit server")
+		if isNormalExit {
+			return errors.New("graceful shutdown timeout")
+		}
+	}
+	if isNormalExit {
+		return nil
 	}
 	return err
 }
 
+func isNormalServerShutdown(err error, ctx context.Context) bool {
+	if err == nil {
+		return true
+	}
+	// Treat cancellation as a normal exit only when the top-level context was
+	// explicitly canceled by shutdown/signal. This avoids masking internal module
+	// failures that also surface as context.Canceled via errgroup cancellation.
+	return errors.Is(err, context.Canceled) && ctx.Err() == context.Canceled
+}
+
 // complete adapts from the command line args and config file to the data required.
 func (o *options) complete(command *cobra.Command) error {
 	cfg := config.GetDefaultServerConfig()

cmd/cdc/server/server_test.go

@@ -14,9 +14,11 @@
 package server
 
 import (
+	"context"
 	"strings"
 	"testing"
 
+	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -118,3 +120,58 @@ func TestNewOptionsDefaultSecurity(t *testing.T) {
 	require.Empty(t, o.serverConfig.Security.CertPath)
 	require.Empty(t, o.serverConfig.Security.KeyPath)
 }
+
+func TestIsNormalServerShutdown(t *testing.T) {
+	testCases := []struct {
+		name      string
+		err       error
+		cancelCtx bool
+		expected  bool
+	}{
+		{
+			name:     "nil error",
+			err:      nil,
+			expected: true,
+		},
+		{
+			name:      "context canceled by shutdown",
+			err:       context.Canceled,
+			cancelCtx: true,
+			expected:  true,
+		},
+		{
+			name:      "wrapped context canceled by shutdown",
+			err:       cerror.Trace(context.Canceled),
+			cancelCtx: true,
+			expected:  true,
+		},
+		{
+			name:     "context canceled without shutdown",
+			err:      context.Canceled,
+			expected: false,
+		},
+		{
+			name:     "wrapped context canceled without shutdown",
+			err:      cerror.Trace(context.Canceled),
+			expected: false,
+		},
+		{
+			name:     "other error",
+			err:      cerror.New("boom"),
+			expected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			if tc.cancelCtx {
+				cancel()
+			} else {
+				defer cancel()
+			}
+
+			require.Equal(t, tc.expected, isNormalServerShutdown(tc.err, ctx))
+		})
+	}
+}

pkg/server/server.go

@@ -27,7 +27,7 @@ import (
 // information in etcd and schedules Task on it.
 type Server interface {
 	Run(ctx context.Context) error
-	Close(ctx context.Context)
+	Close()
 
 	SelfInfo() (*node.Info, error)
 	Liveness() api.Liveness

server/server.go

@@ -415,7 +415,7 @@ func (c *server) GetCoordinator() (tiserver.Coordinator, error) {
 // Close closes the server by deregister it from etcd,
 // it also closes the coordinator and processorManager
 // Note: this function should be reentrant
-func (c *server) Close(ctx context.Context) {
+func (c *server) Close() {
 	if !c.closed.CompareAndSwap(false, true) {
 		return
 	}
@@ -435,11 +435,14 @@ func (c *server) Close(ctx context.Context) {
 		c.closePreServices()
 	}()
 
+	closeCtx, closeCancel := context.WithTimeout(context.Background(), GracefulShutdownTimeout)
+	defer closeCancel()
+
 	// There are also some dependencies inside subModules,
 	// so we close subModules in reverse order of their startup.
 	for i := len(c.subModules) - 1; i >= 0; i-- {
 		m := c.subModules[i]
-		if err := m.Close(ctx); err != nil {
+		if err := m.Close(closeCtx); err != nil {
 			log.Warn("failed to close sub module",
 				zap.String("module", m.Name()),
 				zap.Error(err))
@@ -448,7 +451,7 @@ func (c *server) Close(ctx context.Context) {
 	}
 
 	for _, m := range c.nodeModules {
-		if err := m.Close(ctx); err != nil {
+		if err := m.Close(closeCtx); err != nil {
 			log.Warn("failed to close sub common module",
 				zap.String("module", m.Name()),
 				zap.Error(err))
@@ -457,7 +460,7 @@ func (c *server) Close(ctx context.Context) {
 	}
 
 	for _, nm := range c.networkModules {
-		if err := nm.Close(ctx); err != nil {
+		if err := nm.Close(closeCtx); err != nil {
 			log.Warn("failed to close sub base module",
 				zap.String("module", nm.Name()),
 				zap.Error(err))
@@ -466,8 +469,8 @@ func (c *server) Close(ctx context.Context) {
 	}
 
 	// delete server info from etcd
-	timeoutCtx, cancel := context.WithTimeout(context.Background(), cleanMetaDuration)
-	defer cancel()
+	timeoutCtx, timeoutCancel := context.WithTimeout(closeCtx, cleanMetaDuration)
+	defer timeoutCancel()
 	if err := c.EtcdClient.DeleteCaptureInfo(timeoutCtx, string(c.info.ID)); err != nil {
 		log.Warn("failed to delete server info when server exited",
 			zap.String("captureID", string(c.info.ID)),