fix: prevent goroutine leak in downstreamer on context cancellation (#21062)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: xbglowx

pkg/querier/queryrange/downstreamer.go

@@ -195,9 +195,16 @@ func (in instance) For(
 			return nil
 		})
 		if err != nil {
-			ch <- logql.Resp{
+			// Send the error unless the context has been canceled and the
+			// reader has already exited. Without this select, a canceled
+			// context causes this goroutine to block on the channel send
+			// forever, leaking the goroutine.
+			select {
+			case ch <- logql.Resp{
 				I:   -1,
 				Err: err,
+			}:
+			case <-ctx.Done():
 			}
 		}
 		close(ch)

pkg/querier/queryrange/downstreamer_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -587,6 +588,74 @@ func TestCancelWhileWaitingResponse(t *testing.T) {
 			"but the For method did not return as expected.")
 }
 
+func TestCancelDoesNotLeakGoroutines(t *testing.T) {
+	mkIn := func() *instance {
+		return DownstreamHandler{
+			limits: fakeLimits{},
+			next:   nil,
+		}.Downstreamer(context.Background()).(*instance)
+	}
+	in := mkIn()
+
+