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
This commit is contained in:
ginuerzh
2026-07-28 23:14:27 +08:00
parent 30ba6d64a5
commit 50f4a6b096
4 changed files with 68 additions and 3 deletions
+14
View File
@@ -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.
+19 -2
View File
@@ -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
+13
View File
@@ -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)
}
+22 -1
View File
@@ -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) {