From 50f4a6b096e73dd4f1bf654930e2eff345c71ba0 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Tue, 28 Jul 2026 23:14:27 +0800 Subject: [PATCH] x: close chainGroup probe goroutines on service shutdown chainGroup.Close() existed but was never called: the group is wrapped in a chain.Router held by the listener/handler, and service.Close() only closed the handler, observer, and listener. Every SIGHUP reload leaked the old config's probe goroutines permanently. Wire the close path end to end: - Router.Close() closes its chain when it implements io.Closer - service ClosersOption registers io.Closers called on shutdown - ParseService passes both routers via ClosersOption, and closes them on construction failure so probes started early don't leak --- chain/router.go | 14 ++++++++++++++ config/parsing/service/parse.go | 21 +++++++++++++++++++-- service/service.go | 13 +++++++++++++ service/service_test.go | 23 ++++++++++++++++++++++- 4 files changed, 68 insertions(+), 3 deletions(-) diff --git a/chain/router.go b/chain/router.go index 5edfe56..b641802 100644 --- a/chain/router.go +++ b/chain/router.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "io" "net" "strings" "time" @@ -49,6 +50,19 @@ func (r *Router) Options() *chain.RouterOptions { return &r.options } +// Close releases resources held by the router's chain. A chain that owns +// background goroutines (e.g. a chainGroup running liveness probes) is closed +// here so its goroutines stop when the service shuts down or is reloaded. +func (r *Router) Close() error { + if r == nil { + return nil + } + if c, ok := r.options.Chain.(io.Closer); ok { + return c.Close() + } + return nil +} + // Dial establishes a connection to the target address through the configured // chain. It resolves the address, records the dial event, retries up to // Retries+1 times, and returns the resulting connection. diff --git a/config/parsing/service/parse.go b/config/parsing/service/parse.go index 96ae761..6ae9eec 100644 --- a/config/parsing/service/parse.go +++ b/config/parsing/service/parse.go @@ -207,10 +207,24 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { chain.ChainRouterOption(chainGroup(cfg.Listener.Chain, cfg.Listener.ChainGroup, log)), ) } + lnRouter := xchain.NewRouter(routerOpts...) + + // The routers may hold a chainGroup running probe goroutines; make sure + // they are closed if service construction fails after this point. + var hRouter *xchain.Router + success := false + defer func() { + if !success { + lnRouter.Close() + if hRouter != nil { + hRouter.Close() + } + } + }() listenOpts := []listener.Option{ listener.AddrOption(cfg.Addr), - listener.RouterOption(xchain.NewRouter(routerOpts...)), + listener.RouterOption(lnRouter), listener.AutherOption(auther), listener.AuthOption(auth_parser.Info(cfg.Listener.Auth)), listener.TLSConfigOption(tlsConfig), @@ -386,8 +400,9 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { var h handler.Handler if rf := registry.HandlerRegistry().Get(cfg.Handler.Type); rf != nil { + hRouter = xchain.NewRouter(routerOpts...) h = rf( - handler.RouterOption(xchain.NewRouter(routerOpts...)), + handler.RouterOption(hRouter), handler.AutherOption(auther), handler.AuthOption(auth_parser.Info(cfg.Handler.Auth)), handler.BypassOption(xbypass.BypassGroup(bypass_parser.List(cfg.Bypass, cfg.Bypasses...)...)), @@ -435,7 +450,9 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { xservice.ObserverPeriodOption(observerPeriod), xservice.LoggerOption(serviceLogger), xservice.LabelsOption(labels), + xservice.ClosersOption(lnRouter, hRouter), ) + success = true serviceLogger.Infof("listening on %s/%s", s.Addr().String(), s.Addr().Network()) return s, nil diff --git a/service/service.go b/service/service.go index 44ca936..e907948 100644 --- a/service/service.go +++ b/service/service.go @@ -44,6 +44,7 @@ type options struct { observerPeriod time.Duration logger logger.Logger labels map[string]string + closers []io.Closer } // Option is a functional option for configuring a service. @@ -122,6 +123,13 @@ func LabelsOption(labels map[string]string) Option { } } +// ClosersOption registers io.Closers that are called on service shutdown. +func ClosersOption(closers ...io.Closer) Option { + return func(opts *options) { + opts.closers = append(opts.closers, closers...) + } +} + // LoggerOption sets the logger for the service. If not provided, a no-op // logger is used. func LoggerOption(logger logger.Logger) Option { @@ -364,6 +372,11 @@ func (s *defaultService) Close() error { } } } + for _, c := range s.options.closers { + if err := c.Close(); err != nil { + errs = append(errs, err) + } + } if err := s.listener.Close(); err != nil { errs = append(errs, err) } diff --git a/service/service_test.go b/service/service_test.go index 6db0fc5..6b328af 100644 --- a/service/service_test.go +++ b/service/service_test.go @@ -585,13 +585,34 @@ func TestCloseJoinsErrors(t *testing.T) { type closeErrHandler struct { err error } - func (h *closeErrHandler) Init(metadata.Metadata) error { return nil } func (h *closeErrHandler) Handle(context.Context, net.Conn, ...handler.HandleOption) error { return nil } func (h *closeErrHandler) Close() error { return h.err } +type mockCloser struct { + closed atomic.Bool +} + +func (c *mockCloser) Close() error { + c.closed.Store(true) + return nil +} + +func TestCloseClosesClosers(t *testing.T) { + ln := newMockListener() + c1, c2 := &mockCloser{}, &mockCloser{} + svc := NewService("test", ln, newMockHandler(nil), ClosersOption(c1, c2)).(*defaultService) + + if err := svc.Close(); err != nil { + t.Errorf("Close() = %v", err) + } + if !c1.closed.Load() || !c2.closed.Load() { + t.Error("closers were not closed") + } +} + // --- Status tests --- func TestStatusReturnsStatus(t *testing.T) {