From badffb132d91a4716188540c28a78a38c208cd02 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 26 Apr 2026 13:25:12 -0700 Subject: [PATCH] fix(s3api): stream multipart SSE-KMS chunks lazily Apply the same fix as the previous commit to createMultipartSSEKMSDecryptedReaderDirect: per-chunk SSE-KMS metadata is validated upfront, but volume-server GETs fire lazily through lazyMultipartChunkReader. At most one chunk body is live at any time. This is the same eager-open-all-chunks shape that produced #8908's truncated GETs for SSE-S3; SSE-KMS multipart objects with many chunks were exposed to the same idle-keepalive failure mode under concurrent load. The wire format on disk is unchanged (same per-chunk metadata, same encrypted bytes, same object Extended attributes). Existing SSE-KMS multipart objects read back identically -- only when the volume-server GETs fire changes. --- weed/s3api/s3api_object_handlers.go | 98 +++++++++++------------------ 1 file changed, 35 insertions(+), 63 deletions(-) diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 511321c28..f8932e132 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -2682,10 +2682,13 @@ func (s3a *S3ApiServer) createMultipartSSECDecryptedReaderDirect(ctx context.Con // createMultipartSSEKMSDecryptedReaderDirect creates a reader that decrypts each chunk independently for multipart SSE-KMS objects (direct volume path) // Note: encryptedStream parameter is unused (always nil) as this function fetches chunks directly to avoid double I/O. // It's kept in the signature for API consistency with non-Direct versions. +// +// Per-chunk metadata is validated upfront (so a malformed chunk fails fast +// without opening any HTTP connections); chunk fetches happen LAZILY through +// lazyMultipartChunkReader, so at most one volume-server connection is open +// at a time. See buildMultipartSSES3Reader for the rationale (issue #8908). func (s3a *S3ApiServer) createMultipartSSEKMSDecryptedReaderDirect(ctx context.Context, encryptedStream io.ReadCloser, entry *filer_pb.Entry) (io.Reader, error) { // Close the original encrypted stream since chunks are fetched individually. - // Defer so the stream is closed on every return path (including error - // returns from inside the per-chunk loop), matching the SSE-S3 helper. if encryptedStream != nil { defer encryptedStream.Close() } @@ -2700,72 +2703,41 @@ func (s3a *S3ApiServer) createMultipartSSEKMSDecryptedReaderDirect(ctx context.C return chunks[i].GetOffset() < chunks[j].GetOffset() }) - // Create readers for each chunk, decrypting them independently - readers := make([]io.Reader, 0, len(chunks)) - - // Close any readers already appended to `readers` on error paths, to avoid - // leaking volume-server HTTP connections. - closeAppendedReaders := func() { - for _, r := range readers { - if closer, ok := r.(io.Closer); ok { - closer.Close() - } - } - } - + preparedChunks := make([]preparedMultipartChunk, 0, len(chunks)) for _, chunk := range chunks { - // Get this chunk's encrypted data - chunkReader, err := s3a.createEncryptedChunkReader(ctx, chunk) + if chunk.GetSseType() != filer_pb.SSEType_SSE_KMS { + preparedChunks = append(preparedChunks, preparedMultipartChunk{chunk: chunk}) + continue + } + if len(chunk.GetSseMetadata()) == 0 { + return nil, fmt.Errorf("SSE-KMS chunk %s missing per-chunk metadata", chunk.GetFileIdString()) + } + kmsKey, err := DeserializeSSEKMSMetadata(chunk.GetSseMetadata()) if err != nil { - closeAppendedReaders() - return nil, fmt.Errorf("failed to create chunk reader: %v", err) - } - - // Handle based on chunk's encryption type - if chunk.GetSseType() == filer_pb.SSEType_SSE_KMS { - // Check if this chunk has per-chunk SSE-KMS metadata - if len(chunk.GetSseMetadata()) == 0 { - chunkReader.Close() - closeAppendedReaders() - return nil, fmt.Errorf("SSE-KMS chunk %s missing per-chunk metadata", chunk.GetFileIdString()) - } - - // Use the per-chunk SSE-KMS metadata - kmsKey, err := DeserializeSSEKMSMetadata(chunk.GetSseMetadata()) - if err != nil { - chunkReader.Close() - closeAppendedReaders() - return nil, fmt.Errorf("failed to deserialize SSE-KMS metadata for chunk %s: %v", chunk.GetFileIdString(), err) - } - - glog.V(4).Infof("Decrypting SSE-KMS chunk %s with KeyID=%s", - chunk.GetFileIdString(), kmsKey.KeyID) - - // Create decrypted reader for this chunk - decryptedChunkReader, decErr := CreateSSEKMSDecryptedReader(chunkReader, kmsKey) - if decErr != nil { - chunkReader.Close() - closeAppendedReaders() - return nil, fmt.Errorf("failed to decrypt chunk: %v", decErr) - } - - // Use the streaming decrypted reader directly - readers = append(readers, struct { - io.Reader - io.Closer - }{ - Reader: decryptedChunkReader, - Closer: chunkReader, - }) - glog.V(4).Infof("Added streaming decrypted reader for SSE-KMS chunk %s", chunk.GetFileIdString()) - } else { - // Non-SSE-KMS chunk, use as-is - readers = append(readers, chunkReader) - glog.V(4).Infof("Added non-encrypted reader for chunk %s", chunk.GetFileIdString()) + return nil, fmt.Errorf("failed to deserialize SSE-KMS metadata for chunk %s: %v", chunk.GetFileIdString(), err) } + // Capture kmsKey and chunk into the wrap closure so each prepared + // entry decrypts with its own per-chunk SSE-KMS key. + fileId := chunk.GetFileIdString() + preparedChunks = append(preparedChunks, preparedMultipartChunk{ + chunk: chunk, + wrap: func(raw io.ReadCloser) (io.Reader, error) { + glog.V(4).Infof("Decrypting SSE-KMS chunk %s with KeyID=%s", fileId, kmsKey.KeyID) + dec, decErr := CreateSSEKMSDecryptedReader(raw, kmsKey) + if decErr != nil { + return nil, fmt.Errorf("failed to decrypt chunk: %v", decErr) + } + return dec, nil + }, + }) } - return NewMultipartSSEReader(readers), nil + return &lazyMultipartChunkReader{ + chunks: preparedChunks, + fetch: func(c *filer_pb.FileChunk) (io.ReadCloser, error) { + return s3a.createEncryptedChunkReader(ctx, c) + }, + }, nil } // createMultipartSSES3DecryptedReaderDirect creates a reader that decrypts each chunk independently for multipart SSE-S3 objects (direct volume path)