Release the gRPC connections and SDK clients the factories own (#11438)

## What changed

`RPCFactory` and the SDK client factory close the gRPC connections they
own on shutdown, as fx stop hooks. Removes the four gRPC connection
ignores from the leak test.

Also bumps `auto-scaled-workers` to pick up
temporalio/temporal-auto-scaled-workers#108, without which the SDK
connection stays open.

## Why?

Nothing released these connections, so every connection's goroutines and
the membership resolver watching for changes outlived the cluster.
Clients from `NewClient` share the system client's ref-counted
connection, so the SDK closes it only once the last one is closed.

## How did you test it?
- [x] built
- [x] covered by existing tests

`make leak-test` at the CI settings reports `no unexpected goroutines`.
Each change was verified to fail the leak test when reverted.
This commit is contained in:
Prathyush PV
2026-08-11 17:58:27 -07:00
committed by GitHub
parent 264a483cd5
commit 9f85d70a21
6 changed files with 58 additions and 20 deletions

View File

@@ -385,6 +385,7 @@ func ArchiverProviderProvider(
}
func SdkClientFactoryProvider(
lc fx.Lifecycle,
cfg *config.Config,
tlsConfigProvider encryption.TLSConfigProvider,
metricsHandler metrics.Handler,
@@ -396,13 +397,15 @@ func SdkClientFactoryProvider(
if err != nil {
return nil, err
}
return sdk.NewClientFactory(
factory := sdk.NewClientFactory(
frontendURL,
frontendTLSConfig,
metricsHandler,
logger,
dynamicconfig.WorkerStickyCacheSize.Get(dc),
), nil
)
lc.Append(fx.StopHook(factory.Close))
return factory, nil
}
func DCRedirectionPolicyProvider(cfg *config.Config) config.DCRedirectionPolicy {
@@ -421,6 +424,7 @@ func PerServiceDialOptionsProvider(
}
func RPCFactoryProvider(
lc fx.Lifecycle,
cfg *config.Config,
svcName primitives.ServiceName,
logger log.Logger,
@@ -462,6 +466,7 @@ func RPCFactoryProvider(
factory.EnableInternodeServerKeepalive = enableServerKeepalive
factory.EnableInternodeClientKeepalive = enableClientKeepalive
logger.Debug(fmt.Sprintf("RPC factory created. enableServerKeepalive: %v, enableClientKeepalive: %v", enableServerKeepalive, enableClientKeepalive))
lc.Append(fx.StopHook(factory.Close))
return factory, nil
}

View File

@@ -386,6 +386,26 @@ func (d *RPCFactory) dial(hostName string, tlsClientConfig *tls.Config, dialOpti
return connection
}
func (d *RPCFactory) Close() {
d.internodeConnCleanupTicker.Stop()
d.internodeGRPCConnections.Lock()
for _, conn := range d.internodeGRPCConnections.conns {
_ = conn.Close()
}
clear(d.internodeGRPCConnections.conns)
d.internodeGRPCConnections.Unlock()
d.remoteFrontendGRPCConns.Range(func(_, v any) bool {
if conn, ok := v.(*grpc.ClientConn); ok {
_ = conn.Close()
}
return true
})
_ = d.localFrontendGRPCConn().Close()
}
func (d *RPCFactory) getClientKeepAliveConfig(serviceName primitives.ServiceName) grpc.DialOption {
// default keepalive settings for clients
params := keepalive.ClientParameters{

View File

@@ -42,6 +42,9 @@ type (
systemSdkClient sdkclient.Client
stickyCacheSize dynamicconfig.IntPropertyFn
once sync.Once
clientsLock sync.Mutex
closed bool
}
)
@@ -90,15 +93,16 @@ func (f *clientFactory) NewClient(options sdkclient.Options) sdkclient.Client {
func (f *clientFactory) GetSystemClient() sdkclient.Client {
f.once.Do(func() {
var sdkClient sdkclient.Client
err := backoff.ThrottleRetry(func() error {
sdkClient, err := sdkclient.Dial(f.options(sdkclient.Options{
var err error
sdkClient, err = sdkclient.Dial(f.options(sdkclient.Options{
Namespace: primitives.SystemLocalNamespace,
}))
if err != nil {
f.logger.Warn("error creating sdk client", tag.Error(err))
return err
}
f.systemSdkClient = sdkClient
return nil
}, common.CreateSdkClientFactoryRetryPolicy(), func(err error) bool {
// note err is wrapped by sdk
@@ -113,6 +117,14 @@ func (f *clientFactory) GetSystemClient() sdkclient.Client {
f.logger.Info("setting sticky workflow cache size", tag.Int("size", size))
sdkworker.SetStickyWorkflowCacheSize(size)
}
f.clientsLock.Lock()
defer f.clientsLock.Unlock()
f.systemSdkClient = sdkClient
if f.closed {
sdkClient.Close()
}
})
return f.systemSdkClient
}
@@ -125,6 +137,20 @@ func (f *clientFactory) NewWorker(
return sdkworker.New(client, taskQueue, options)
}
func (f *clientFactory) Close() {
f.clientsLock.Lock()
defer f.clientsLock.Unlock()
if f.closed {
return
}
f.closed = true
if f.systemSdkClient != nil {
f.systemSdkClient.Close()
}
}
// Overwrite the 'client-name' and 'client-version' headers on gRPC requests sent using the Go SDK
// so they clearly indicate that the request is coming from the Temporal server.
func sdkClientNameHeadersInjectorInterceptor() grpc.UnaryClientInterceptor {

2
go.mod
View File

@@ -67,7 +67,7 @@ require (
go.opentelemetry.io/otel/sdk/metric v1.43.0
go.opentelemetry.io/otel/trace v1.44.0
go.temporal.io/api v1.63.5
go.temporal.io/auto-scaled-workers v0.0.0-20260728174133-979694be4176
go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab
go.temporal.io/sdk v1.44.0
go.uber.org/fx v1.24.0
go.uber.org/goleak v1.3.0

4
go.sum
View File

@@ -481,8 +481,8 @@ go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN
go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4=
go.temporal.io/api v1.63.5 h1:c11+kPYHkXXL3UiShPdbMD+xtvqGsbTibUA9ypmiCa4=
go.temporal.io/api v1.63.5/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ=
go.temporal.io/auto-scaled-workers v0.0.0-20260728174133-979694be4176 h1:6AUglT8D3HsbOmV2zwwPpQOmxFXZN4NrMe6Z69n3fSc=
go.temporal.io/auto-scaled-workers v0.0.0-20260728174133-979694be4176/go.mod h1:hhHijO9XRPIkAflLJJHix61M9FzbRPqk8fSydkcLkqw=
go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab h1:99wXW0317BBi49d6xgMdA0EZtvA+xbBUWV4HsTEGEcg=
go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab/go.mod h1:hhHijO9XRPIkAflLJJHix61M9FzbRPqk8fSydkcLkqw=
go.temporal.io/sdk v1.44.0 h1:suitPDukX74rW3/N1FqvEbZTZVJJsxMKhv0KMa/j7pU=
go.temporal.io/sdk v1.44.0/go.mod h1:vkApR12F9/Y8OR+hkxe7WyXQFuCX6clhzqnAk6rzDAM=
go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=

View File

@@ -24,19 +24,6 @@ import (
var goleakOpts = []goleak.Option{
// By design: sqlite keeps one *sql.DB per file DSN for the process lifetime.
goleak.IgnoreTopFunction("database/sql.(*DB).connectionOpener"),
// TODO: gRPC connection goroutines leaked because history/matching
// connection pools are not closed on cluster shutdown.
//
// IgnoreAnyFunction (rather than IgnoreTopFunction) for addrConn's
// reconnect loop: a goroutine caught mid-reconnect can have any of
// several functions (fmt/channelz logging, context.Err, ...) at the
// top of the stack depending on exactly when it was snapshotted, but
// resetTransportAndUnlock is always present as a caller.
goleak.IgnoreTopFunction("google.golang.org/grpc/internal/grpcsync.(*CallbackSerializer).run"),
goleak.IgnoreAnyFunction("google.golang.org/grpc.(*addrConn).resetTransportAndUnlock"),
goleak.IgnoreTopFunction("google.golang.org/grpc/internal/balancer/gracefulswitch.(*Balancer).updateSubConnState"),
goleak.IgnoreTopFunction("go.temporal.io/server/common/membership.(*grpcResolver).listen"),
}
var objectLeakOpts = []objectleak.Option{