diff --git a/pkg/storage/chunked.go b/pkg/storage/chunked.go index be411e66..0eaaef44 100644 --- a/pkg/storage/chunked.go +++ b/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) diff --git a/pkg/storage/chunked_test.go b/pkg/storage/chunked_test.go index d8bd0c12..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,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) +}