Ingester client: close conn and cancel stream ctx when Run() fails

CHANGELOG.md

@@ -92,6 +92,7 @@
 * [BUGFIX] Parquet Converter: Fix `auto_forget_delay` having no effect. The ring lifecycler was created without the auto-forget delegate, so unhealthy instances were never automatically removed from the ring. #7752
 * [BUGFIX] Alertmanager: Reject the global `mattermost_webhook_url_file` setting in per-tenant configs, consistent with every other global `*_file` setting. #7768
 * [BUGFIX] Alertmanager: Tighten per-tenant config validation to reject additional file-based settings. #7767
+* [BUGFIX] Ingester Client: Fix `grpc.ClientConn` and goroutine leak in `MakeIngesterClient` when `-distributor.use-stream-push=true` and starting the stream-push workers fails (e.g. the ingester address is in the ring but unreachable). The connection is now closed and the stream context cancelled on that error path. #7759
 
 ## 1.21.1 2026-06-04
 

pkg/ingester/client/client.go

@@ -168,6 +168,11 @@ func MakeIngesterClient(addr string, cfg Config, useStreamConnection bool) (Heal
 		streamCtx, streamCancel := context.WithCancel(context.Background())
 		err = c.Run(make(chan *streamWriteJob, INGESTER_CLIENT_STREAM_WORKER_COUNT), streamCtx, streamCancel)
 		if err != nil {
+			// Cancelling the stream context unblocks the job-processing goroutines
+			// started by any workers that did succeed, and closing conn tears down
+			// the underlying ClientConn. Without this, both are leaked forever.
+			streamCancel()
+			_ = conn.Close()
 			return nil, err
 		}
 	}
@@ -210,7 +215,7 @@ func (c *closableHealthAndIngesterClient) Run(streamPushChan chan *streamWriteJo
 	c.streamCtx = streamCtx
 	c.streamCancel = streamCancel
 
-	var workerErr error
+	workerErr := atomic.NewError(nil)
 	var wg sync.WaitGroup
 	// Sanitize addr: colons (from host:port) are not allowed in tenant IDs.
 	sanitizedAddr := strings.ReplaceAll(c.addr, ":", "-")
@@ -220,12 +225,12 @@ func (c *closableHealthAndIngesterClient) Run(streamPushChan chan *streamWriteJo
 			workerCtx := user.InjectOrgID(streamCtx, workerName)
 			err := c.worker(workerCtx)
 			if err != nil {
-				workerErr = err
+				workerErr.Store(err)
 			}
 		})
 	}
 	wg.Wait()
-	return workerErr
+	return workerErr.Load()
 }
 
 func (c *closableHealthAndIngesterClient) worker(ctx context.Context) error {

pkg/ingester/client/client_test.go

@@ -2,8 +2,11 @@ package client
 
 import (
 	"context"
+	"flag"
 	"fmt"
+	"net"
 	"net/http/httptest"
+	"runtime"
 	"strconv"
 	"testing"
 	"time"
@@ -17,6 +20,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/test"
 )
 
 // TestMarshall is useful to try out various optimisation on the unmarshalling code.
@@ -393,3 +397,36 @@ func TestClosableHealthAndIngesterClient_ShouldNotPanicWhenClose(t *testing.T) {
 
 	time.Sleep(100 * time.Millisecond)
 }
+
+// TestMakeIngesterClient_CleansUpOnRunFailure is a regression test for a
+// leaked grpc.ClientConn (and its background reconnect goroutines) when
+// Run() fails to start the stream-push workers. Using an address nothing
+// listens on reproduces the reported failure mode: PushStream is created
+// without WaitForReady, so it fails fast once the addrConn reaches
+// TRANSIENT_FAILURE, and every worker fails the same way.
+func TestMakeIngesterClient_CleansUpOnRunFailure(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	addr := ln.Addr().String()
+	require.NoError(t, ln.Close())
+
+	cfg := Config{}
+	cfg.RegisterFlags(flag.NewFlagSet("test", flag.ContinueOnError))
+
+	// Let goroutines from earlier tests settle before taking the baseline.
+	test.Poll(t, time.Second, true, func() any {
+		before := runtime.NumGoroutine()
+		time.Sleep(10 * time.Millisecond)
+		return runtime.NumGoroutine() == before
+	})
+	baseline := runtime.NumGoroutine()
+
+	_, err = MakeIngesterClient(addr, cfg, true)
+	require.Error(t, err)
+
+	// Without the fix, the ClientConn's addrConn keeps retrying in the
+	// background and its goroutines never return to baseline.
+	test.Poll(t, time.Second, true, func() any {
+		return runtime.NumGoroutine() <= baseline
+	})
+}