fix: thread lifecycle context to avoid goroutine leaks on shutdown

Pass the application context from main() through wireApp to all NATS
reconnect subscribers (caches and audit publisher). When the context is
cancelled on shutdown, subscriber channels are closed and watcher
goroutines exit cleanly.

Signed-off-by: Miguel Martinez Trivino <miguel@chainloop.dev>

Author: migmartri

app/controlplane/cmd/main.go

@@ -144,7 +144,7 @@ func main() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	app, cleanup, err := wireApp(&bc, credsWriter, logger, availablePlugins)
+	app, cleanup, err := wireApp(ctx, &bc, credsWriter, logger, availablePlugins)
 	if err != nil {
 		panic(err)
 	}

app/controlplane/cmd/wire.go

@@ -45,7 +45,7 @@ import (
 	"github.com/google/wire"
 )
 
-func wireApp(*conf.Bootstrap, credentials.ReaderWriter, log.Logger, sdk.AvailablePlugins) (*app, func(), error) {
+func wireApp(context.Context, *conf.Bootstrap, credentials.ReaderWriter, log.Logger, sdk.AvailablePlugins) (*app, func(), error) {
 	panic(
 		wire.Build(
 			wire.Bind(new(credentials.Reader), new(credentials.ReaderWriter)),
@@ -143,40 +143,40 @@ var cacheProviderSet = wire.NewSet(
 	newPolicyEvalBundleCache,
 )
 
-func newClaimsCache(rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*jwt.MapClaims], error) {
+func newClaimsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*jwt.MapClaims], error) {
 	l := log.NewHelper(logger)
 	backend := "memory"
 	opts := []cache.Option{cache.WithTTL(10 * time.Second), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for JWT claims")}
 	if rc != nil {
 		backend = "nats"
 		opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-jwt-claims"))
-		opts = append(opts, cache.WithReconnect(rc.Subscribe(context.Background())))
+		opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx)))
 	}
 	l.Infow("msg", "cache initialized", "bucket", "chainloop-jwt-claims", "backend", backend, "ttl", "10s")
 	return cache.New[*jwt.MapClaims](opts...)
 }
 
-func newMembershipsCache(rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*entities.Membership], error) {
+func newMembershipsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*entities.Membership], error) {
 	l := log.NewHelper(logger)
 	backend := "memory"
 	opts := []cache.Option{cache.WithTTL(time.Second), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for org memberships")}
 	if rc != nil {
 		backend = "nats"
 		opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-memberships"))
-		opts = append(opts, cache.WithReconnect(rc.Subscribe(context.Background())))
+		opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx)))
 	}
 	l.Infow("msg", "cache initialized", "bucket", "chainloop-memberships", "backend", backend, "ttl", "1s")
 	return cache.New[*entities.Membership](opts...)
 }
 
-func newPolicyEvalBundleCache(rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[[]byte], error) {
+func newPolicyEvalBundleCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[[]byte], error) {
 	l := log.NewHelper(logger)
 	backend := "memory"
 	opts := []cache.Option{cache.WithTTL(24 * time.Hour), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for policy evaluation bundles from CAS")}
 	if rc != nil {
 		backend = "nats"
 		opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-policy-eval-bundles"))
-		opts = append(opts, cache.WithReconnect(rc.Subscribe(context.Background())))
+		opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx)))
 	}
 	l.Infow("msg", "cache initialized", "bucket", "chainloop-policy-eval-bundles", "backend", backend, "ttl", "24h")
 	return cache.New[[]byte](opts...)

app/controlplane/cmd/wire_gen.go

@@ -41,7 +41,7 @@ type AuditLogPublisher struct {
 	logger *log.Helper
 }
 
-func NewAuditLogPublisher(rc *natsconn.ReloadableConnection, logger log.Logger) (*AuditLogPublisher, error) {
+func NewAuditLogPublisher(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*AuditLogPublisher, error) {
 	l := log.NewHelper(log.With(logger, "component", "natsAuditLogPublisher"))
 	if rc == nil {
 		l.Infow("msg", "NATS connection not set, audit log publisher disabled")
@@ -54,7 +54,7 @@ func NewAuditLogPublisher(rc *natsconn.ReloadableConnection, logger log.Logger)
 		return nil, err
 	}
 
-	go p.watchReconnect(rc.Subscribe(context.Background()))
+	go p.watchReconnect(rc.Subscribe(ctx))
 
 	return p, nil
 }

app/controlplane/pkg/auditor/nats.go

@@ -1,5 +1,5 @@
 //
-// Copyright 2024-2025 The Chainloop Authors.
+// Copyright 2024-2026 The Chainloop Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -146,7 +146,7 @@ func NewTestingUseCases(t *testing.T, opts ...NewTestingUCOpt) *TestingUseCases
 
 	db := NewTestDatabase(t)
 	log := log.NewStdLogger(io.Discard)
-	testData, _, err := WireTestData(db, t, log, newArgs.credsReaderWriter, &robotaccount.Builder{}, &conf.Auth{
+	testData, _, err := WireTestData(t.Context(), db, t, log, newArgs.credsReaderWriter, &robotaccount.Builder{}, &conf.Auth{
 		GeneratedJwsHmacSecret:        "test",
 		CasRobotAccountPrivateKeyPath: "./testdata/test-key.ec.pem",
 	}, &conf.Bootstrap{}, newArgs.onboardingConfiguration, newArgs.integrations, newArgs.providers)

app/controlplane/pkg/biz/testhelpers/database.go

@@ -21,6 +21,7 @@
 package testhelpers
 
 import (
+	"context"
 	"testing"
 
 	conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1"
@@ -41,7 +42,7 @@ import (
 )
 
 // wireTestData init testing data
-func WireTestData(*TestDatabase, *testing.T, log.Logger, credentials.ReaderWriter, *robotaccount.Builder, *conf.Auth, *conf.Bootstrap, []*config.OnboardingSpec, sdk.AvailablePlugins, backends.Providers) (*TestingUseCases, func(), error) {
+func WireTestData(context.Context, *TestDatabase, *testing.T, log.Logger, credentials.ReaderWriter, *robotaccount.Builder, *conf.Auth, *conf.Bootstrap, []*config.OnboardingSpec, sdk.AvailablePlugins, backends.Providers) (*TestingUseCases, func(), error) {
 	panic(
 		wire.Build(
 			data.ProviderSet,