mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-19 21:56:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(s3api): stream multipart SSE-KMS chunks lazily

Browse Source 
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.
This commit is contained in:
Chris Lu
2026-04-26 13:25:12 -07:00
parent 7bc87bec07
commit badffb132d
1 changed files with 35 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/s3api/s3api_object_handlers.go
+35 -63
View File
@@ -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)