x: close forwarder hop probe goroutines on service shutdown

Same leak class as the previous chainGroup fix: a forwarder built from
a hopGroup or inline nodes starts probe goroutines at parse time, but
nothing closed them — every SIGHUP reload leaked them permanently.

- parseForwarder returns an io.Closer for service-owned hops (hop
  group or inline nodes); registry-owned named hops return nil since
  the registry closes them on unregister
- ParseService registers the closer via ClosersOption and closes it
  when construction fails partway
- hopGroup.Close no longer closes the underlying hops: they come from
  the HopRegistry and are shared with other referents; closing them
  here would kill shared state
This commit is contained in:
ginuerzh
2026-07-28 23:45:40 +08:00
parent 291fa48fae
commit 3cd4fad4bf
2 changed files with 38 additions and 15 deletions
+35 -9
View File
@@ -2,6 +2,7 @@ package service
import (
"fmt"
"io"
"runtime"
"strings"
"time"
@@ -209,9 +210,11 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
}
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.
// The routers may hold a chainGroup running probe goroutines, and the
// forwarder hop may run probe goroutines of its own; make sure they are
// closed if service construction fails after this point.
var hRouter *xchain.Router
var fwdCloser io.Closer
success := false
defer func() {
if !success {
@@ -219,6 +222,9 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
if hRouter != nil {
hRouter.Close()
}
if fwdCloser != nil {
fwdCloser.Close()
}
}
}()
@@ -422,10 +428,11 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
}
if forwarder, ok := h.(handler.Forwarder); ok {
hop, err := parseForwarder(cfg.Forwarder, log)
hop, closer, err := parseForwarder(cfg.Forwarder, log)
if err != nil {
return nil, err
}
fwdCloser = closer
forwarder.Forward(hop)
}
@@ -438,6 +445,10 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
return nil, err
}
closers := []io.Closer{lnRouter, hRouter}
if fwdCloser != nil {
closers = append(closers, fwdCloser)
}
s := xservice.NewService(cfg.Name, ln, h,
xservice.AdmissionOption(xadmission.AdmissionGroup(admissions...)),
xservice.PreUpOption(preUp),
@@ -450,7 +461,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
xservice.ObserverPeriodOption(observerPeriod),
xservice.LoggerOption(serviceLogger),
xservice.LabelsOption(labels),
xservice.ClosersOption(lnRouter, hRouter),
xservice.ClosersOption(closers...),
)
success = true
@@ -458,14 +469,23 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
return s, nil
}
func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, error) {
// parseForwarder builds the forwarder hop for a service. The returned
// io.Closer is non-nil only when the hop is owned by the service (inline
// nodes or a hop group) and must be closed on service shutdown; hops from
// the HopRegistry are closed by the registry on unregister.
func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, io.Closer, error) {
if cfg == nil {
return nil, nil
return nil, nil, nil
}
// HopGroup takes precedence over Hop / Name / Nodes.
if cfg.HopGroup != nil {
return parseHopGroup(cfg.HopGroup, log)
group, err := parseHopGroup(cfg.HopGroup, log)
if err != nil || group == nil {
return nil, nil, err
}
closer, _ := group.(io.Closer)
return group, closer, nil
}
hopName := cfg.Hop
@@ -473,7 +493,7 @@ func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, er
hopName = cfg.Name
}
if hopName != "" {
return registry.HopRegistry().Get(hopName), nil
return registry.HopRegistry().Get(hopName), nil, nil
}
hc := config.HopConfig{
@@ -519,7 +539,13 @@ func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, er
Metadata: node.Metadata,
})
}
return hop_parser.ParseHop(&hc, log)
// Inline nodes build a per-service hop, owned (and closed) by the service.
h, err := hop_parser.ParseHop(&hc, log)
if err != nil || h == nil {
return nil, nil, err
}
closer, _ := h.(io.Closer)
return h, closer, nil
}
func parseHopGroup(cfg *config.ForwardHopGroupConfig, log logger.Logger) (hop.Hop, error) {
+3 -6
View File
@@ -187,14 +187,11 @@ func (g *hopGroup) Nodes() []*chain.Node {
return nodes
}
// Close stops all probe goroutines and closes each hop entry.
// Close stops all probe goroutines. It does NOT close the underlying hops:
// they come from the HopRegistry and are shared — the registry closes them
// on unregister.
func (g *hopGroup) Close() error {
g.cancelFunc()
for _, e := range g.entries {
if closer, ok := e.hop.(interface{ Close() error }); ok {
closer.Close()
}
}
return nil
}