From e1331e031d21d55708dd1b0689201636634dd612 Mon Sep 17 00:00:00 2001 From: Raphael Fakhri <153192858+RaphaelFakhri@users.noreply.github.com> Date: Wed, 4 Nov 2026 07:44:14 +0000 Subject: [PATCH 1/4] pkg/storage: fix goroutine leak when chunked release decoding fails decodeRelease streams chunk data to the gzip and JSON decoders through an io.Pipe, but never closes the read side. If decoding fails before all of the chunk data is read, for example because of a corrupt or truncated chunk, the writer goroutine blocks on pw.Write forever and keeps the secret data in memory. Each failed Get, List or Query leaks one goroutine, so an operator that repeatedly reconciles such a release grows without bound. Close the pipe reader when decodeRelease returns so the pending write fails and the goroutine exits. --- pkg/storage/chunked.go | 3 +++ pkg/storage/chunked_test.go | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/pkg/storage/chunked.go b/pkg/storage/chunked.go index be411e66..29036d97 100644 --- a/pkg/storage/chunked.go +++ b/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"] diff --git a/pkg/storage/chunked_test.go b/pkg/storage/chunked_test.go index d8bd0c12..3d9fba91 100644 --- a/pkg/storage/chunked_test.go +++ b/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) } From 7d52c3329a76efd4eea5ed6e6669df5c1ce4d97c Mon Sep 17 00:00:00 2001 From: Raphael Fakhri <153192858+RaphaelFakhri@users.noreply.github.com> Date: Fri, 6 Nov 2026 20:41:06 +0000 Subject: [PATCH 2/4] pkg/storage: check the goroutine count instead of stack traces Count all goroutines with runtime.NumGoroutine, so the leak test does not depend on the stack trace format or on function names. Move the deferred pr.Close next to the gzip reader that consumes pr. --- pkg/storage/chunked.go | 6 +++--- pkg/storage/chunked_test.go | 15 ++------------- 2 files changed, 5 insertions(+), 16 deletions(-) diff --git a/pkg/storage/chunked.go b/pkg/storage/chunked.go index 29036d97..2dcff5dc 100644 --- a/pkg/storage/chunked.go +++ b/pkg/storage/chunked.go @@ -398,9 +398,6 @@ 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"] @@ -430,6 +427,9 @@ func (c *chunkedSecrets) decodeRelease(ctx context.Context, indexSecret *corev1. } }() + // 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() gzr, err := gzip.NewReader(pr) if err != nil { return nil, fmt.Errorf("failed to create gzip reader: %w", err) diff --git a/pkg/storage/chunked_test.go b/pkg/storage/chunked_test.go index 3d9fba91..e99901c6 100644 --- a/pkg/storage/chunked_test.go +++ b/pkg/storage/chunked_test.go @@ -6,7 +6,6 @@ import ( "fmt" "maps" "runtime" - "strings" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -111,11 +110,12 @@ var _ = Describe("chunkedSecrets", func() { _, err := secretInterface.Create(context.Background(), corruptIndex, metav1.CreateOptions{}) Expect(err).ToNot(HaveOccurred()) + goroutinesBefore := runtime.NumGoroutine() for range 5 { _, err := chunkedDriver.Get(corruptIndex.Name) Expect(err).To(MatchError(ContainSubstring("failed to create gzip reader"))) } - Eventually(countDecodeReleaseGoroutines).Should(BeZero()) + Eventually(runtime.NumGoroutine).Should(BeNumerically("<=", goroutinesBefore)) }) }) @@ -349,17 +349,6 @@ 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) } From a16cf243efed31fc9e497de27e3b10ce202d117c Mon Sep 17 00:00:00 2001 From: RaphaelFakhri <153192858+RaphaelFakhri@users.noreply.github.com> Date: Mon, 9 Nov 2026 20:36:35 +0000 Subject: [PATCH 3/4] pkg/storage: wait for the chunk writer goroutines in the leak test Track the goroutines started by decodeRelease with a WaitGroup and have the test wait on it, instead of comparing runtime.NumGoroutine, so unrelated goroutines starting or exiting cannot affect the result. --- pkg/storage/chunked.go | 6 ++++++ pkg/storage/chunked_test.go | 12 +++++++++--- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/pkg/storage/chunked.go b/pkg/storage/chunked.go index 2dcff5dc..11c0827a 100644 --- a/pkg/storage/chunked.go +++ b/pkg/storage/chunked.go @@ -57,6 +57,10 @@ type chunkedSecrets struct { hashMu sync.Mutex hash hash.Hash64 hashEncoding *base32.Encoding + + // chunkWriters tracks the goroutines started by decodeRelease that feed + // chunk data into the decoder, so tests can check that they all exit. + chunkWriters sync.WaitGroup } func (c *chunkedSecrets) Create(key string, rls *release.Release) error { @@ -398,7 +402,9 @@ func (c *chunkedSecrets) decodeRelease(ctx context.Context, indexSecret *corev1. } pr, pw := io.Pipe() + c.chunkWriters.Add(1) go func() { + defer c.chunkWriters.Done() defer pw.Close() firstChunkData, ok := indexSecret.Data["chunk"] if !ok { diff --git a/pkg/storage/chunked_test.go b/pkg/storage/chunked_test.go index e99901c6..9112baf2 100644 --- a/pkg/storage/chunked_test.go +++ b/pkg/storage/chunked_test.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "maps" - "runtime" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -110,12 +109,19 @@ var _ = Describe("chunkedSecrets", func() { _, err := secretInterface.Create(context.Background(), corruptIndex, metav1.CreateOptions{}) Expect(err).ToNot(HaveOccurred()) - goroutinesBefore := runtime.NumGoroutine() for range 5 { _, err := chunkedDriver.Get(corruptIndex.Name) Expect(err).To(MatchError(ContainSubstring("failed to create gzip reader"))) } - Eventually(runtime.NumGoroutine).Should(BeNumerically("<=", goroutinesBefore)) + + // Wait only for the chunk writer goroutines started by this + // driver, so unrelated goroutines cannot affect the result. + writersDone := make(chan struct{}) + go func() { + chunkedDriver.(*chunkedSecrets).chunkWriters.Wait() + close(writersDone) + }() + Eventually(writersDone).Should(BeClosed()) }) }) From 633f0a59560099bb0f6286e28a4dfc8e4fb20522 Mon Sep 17 00:00:00 2001 From: RaphaelFakhri <153192858+RaphaelFakhri@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:31:29 +0000 Subject: [PATCH 4/4] pkg/storage: wait for the chunk writer in decodeRelease and cancel its requests Wait for the chunk writer goroutine before decodeRelease returns instead of tracking it in a struct field for the test. To keep that wait bounded, the read side of the pipe is closed and the context used for the chunk Get calls is cancelled first, so the writer is unblocked whether it is in pw.Write or in an in-flight Get. The tests run Get in a separate goroutine with a timeout, so a regression fails the test instead of hanging it, and a new test checks that a blocked chunk Get is cancelled. Signed-off-by: RaphaelFakhri <153192858+RaphaelFakhri@users.noreply.github.com> --- pkg/storage/chunked.go | 23 +++++++------- pkg/storage/chunked_test.go | 63 ++++++++++++++++++++++++++++++------- 2 files changed, 63 insertions(+), 23 deletions(-) diff --git a/pkg/storage/chunked.go b/pkg/storage/chunked.go index 11c0827a..0eaaef44 100644 --- a/pkg/storage/chunked.go +++ b/pkg/storage/chunked.go @@ -57,10 +57,6 @@ type chunkedSecrets struct { hashMu sync.Mutex hash hash.Hash64 hashEncoding *base32.Encoding - - // chunkWriters tracks the goroutines started by decodeRelease that feed - // chunk data into the decoder, so tests can check that they all exit. - chunkWriters sync.WaitGroup } func (c *chunkedSecrets) Create(key string, rls *release.Release) error { @@ -401,10 +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() - c.chunkWriters.Add(1) - go func() { - defer c.chunkWriters.Done() + var chunkWriter sync.WaitGroup + chunkWriter.Go(func() { defer pw.Close() firstChunkData, ok := indexSecret.Data["chunk"] if !ok { @@ -431,10 +427,15 @@ func (c *chunkedSecrets) decodeRelease(ctx context.Context, indexSecret *corev1. return } } - }() - - // 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. + }) + + // 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 { diff --git a/pkg/storage/chunked_test.go b/pkg/storage/chunked_test.go index 9112baf2..f96c9d8b 100644 --- a/pkg/storage/chunked_test.go +++ b/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,7 +96,7 @@ 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() { + 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, @@ -109,19 +110,30 @@ var _ = Describe("chunkedSecrets", func() { _, 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"))) + // 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()) - // Wait only for the chunk writer goroutines started by this - // driver, so unrelated goroutines cannot affect the result. - writersDone := make(chan struct{}) - go func() { - chunkedDriver.(*chunkedSecrets).chunkWriters.Wait() - close(writersDone) - }() - Eventually(writersDone).Should(BeClosed()) + 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"))) }) }) @@ -374,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) +}