pkg/storage: fix goroutine leak when chunked release decoding fails

pkg/storage/chunked.go

@@ -398,6 +398,9 @@ func (c *chunkedSecrets) decodeRelease(ctx context.Context, indexSecret *corev1.
 	}
 
 	pr, pw := io.Pipe()
+	// Close the read side when decoding returns so that the writer goroutine
+	// does not block forever if decoding stops before all chunk data is read.
+	defer pr.Close()
 	go func() {
 		defer pw.Close()
 		firstChunkData, ok := indexSecret.Data["chunk"]

pkg/storage/chunked_test.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"maps"
+	"runtime"
+	"strings"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -95,6 +97,26 @@ var _ = Describe("chunkedSecrets", func() {
 			_, err := maxReadDriver.Get(releaseKey(rel))
 			Expect(err).To(MatchError(ContainSubstring("release too large")))
 		})
+		It("should not leak goroutines if the release fails to decode", func() {
+			corruptIndex := &corev1.Secret{
+				ObjectMeta: metav1.ObjectMeta{Name: "corrupt-release.v1"},
+				Type:       SecretTypeChunkedIndex,
+				Data: map[string][]byte{
+					"extraChunks": []byte("[]"),
+					// Larger than the gzip reader's read buffer, so the
+					// reader stops before consuming all of the chunk data.
+					"chunk": make([]byte, 10*chunkSize),
+				},
+			}
+			_, err := secretInterface.Create(context.Background(), corruptIndex, metav1.CreateOptions{})
+			Expect(err).ToNot(HaveOccurred())
+
+			for range 5 {
+				_, err := chunkedDriver.Get(corruptIndex.Name)
+				Expect(err).To(MatchError(ContainSubstring("failed to create gzip reader")))
+			}
+			Eventually(countDecodeReleaseGoroutines).Should(BeZero())
+		})
 	})
 
 	var _ = Describe("Update", func() {
@@ -327,6 +349,17 @@ func verifySecrets(secretInterface clientcorev1.SecretInterface, expected int) {
 	Expect(actualExtraChunkNames).To(ConsistOf(expectedExtraChunkNames))
 }
 
+func countDecodeReleaseGoroutines() int {
+	buf := make([]byte, 1<<20)
+	for {
+		n := runtime.Stack(buf, true)
+		if n < len(buf) {
+			return strings.Count(string(buf[:n]), "(*chunkedSecrets).decodeRelease.func")
+		}
+		buf = make([]byte, 2*len(buf))
+	}
+}
+
 func releaseKey(rel *release.Release) string {
 	return fmt.Sprintf("%s.v%d", rel.Name, rel.Version)
 }