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

pkg/storage/chunked.go

@@ -397,8 +397,10 @@ func (c *chunkedSecrets) decodeRelease(ctx context.Context, indexSecret *corev1.
 		return nil, fmt.Errorf("release too large: %q consists of %d chunks, which exceeds the maximum of %d", indexSecret.Name, 1+len(extraChunkNames), c.MaxReadChunks)
 	}
 
+	ctx, cancel := context.WithCancel(ctx)
 	pr, pw := io.Pipe()
-	go func() {
+	var chunkWriter sync.WaitGroup
+	chunkWriter.Go(func() {
 		defer pw.Close()
 		firstChunkData, ok := indexSecret.Data["chunk"]
 		if !ok {
@@ -425,8 +427,16 @@ func (c *chunkedSecrets) decodeRelease(ctx context.Context, indexSecret *corev1.
 				return
 			}
 		}
-	}()
-
+	})
+
+	// Make sure the writer goroutine has exited before returning. If decoding
+	// stops before all chunk data is read, the writer is unblocked by closing
+	// the read side (for a pending pw.Write) and by cancelling ctx (for an
+	// in-flight chunk Get). Deferred calls run in reverse order, so both
+	// happen before the Wait.
+	defer chunkWriter.Wait()
+	defer cancel()
+	defer pr.Close()
 	gzr, err := gzip.NewReader(pr)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create gzip reader: %w", err)

pkg/storage/chunked_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"maps"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -95,6 +96,45 @@ var _ = Describe("chunkedSecrets", func() {
 			_, err := maxReadDriver.Get(releaseKey(rel))
 			Expect(err).To(MatchError(ContainSubstring("release too large")))
 		})
+		It("should return once the chunk writer has stopped 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())
+
+			// Get waits for the chunk writer goroutine, so it would block
+			// forever if the writer stayed stuck on the pipe.
+			Expect(getWithTimeout(chunkedDriver, corruptIndex.Name)).To(MatchError(ContainSubstring("failed to create gzip reader")))
+		})
+		It("should cancel an in-flight chunk request 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(`["corrupt-release.v1.2"]`),
+					"chunk":       []byte("not gzip data"),
+				},
+			}
+			_, err := secretInterface.Create(context.Background(), corruptIndex, metav1.CreateOptions{})
+			Expect(err).ToNot(HaveOccurred())
+
+			blockingDriver := NewChunkedSecrets(&blockingChunkGetter{
+				SecretInterface: secretInterface,
+				blockName:       "corrupt-release.v1.2",
+			}, "test-owner", ChunkedSecretsConfig{ChunkSize: chunkSize})
+
+			// Without cancellation, the writer would stay in the blocking Get
+			// and Get would never return.
+			Expect(getWithTimeout(blockingDriver, corruptIndex.Name)).To(MatchError(ContainSubstring("failed to create gzip reader")))
+		})
 	})
 
 	var _ = Describe("Update", func() {
@@ -346,3 +386,30 @@ func genRelease(name string, version int, status release.Status, extraLabels map
 		Labels: lbls,
 	}
 }
+
+// getWithTimeout calls drv.Get in a separate goroutine and fails the test if it
+// does not return in time, so a hang shows up as a test failure.
+func getWithTimeout(drv driver.Driver, key string) error {
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := drv.Get(key)
+		errCh <- err
+	}()
+	var err error
+	EventuallyWithOffset(1, errCh).WithTimeout(10 * time.Second).Should(Receive(&err))
+	return err
+}
+
+// blockingChunkGetter blocks Get for blockName until the context is cancelled.
+type blockingChunkGetter struct {
+	clientcorev1.SecretInterface
+	blockName string
+}
+
+func (b *blockingChunkGetter) Get(ctx context.Context, name string, opts metav1.GetOptions) (*corev1.Secret, error) {
+	if name == b.blockName {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	return b.SecretInterface.Get(ctx, name, opts)
+}