mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-19 21:56:54 +00:00
fix(s3api): stream multipart SSE-S3 chunks lazily to avoid truncated GETs (#8908)
buildMultipartSSES3Reader opened a volume-server HTTP response for EVERY chunk upfront, then walked them with io.MultiReader. For a multipart SSE-S3 object with N internal chunks (e.g. a 200MB Docker Registry blob with 25+ chunks), N volume-server bodies sat live at once; chunks 1..N-1 were idle while io.MultiReader drained chunk 0. Under concurrent load the volume server's keep-alive logic closed those idle responses mid-flight, and the S3 client saw `unexpected EOF` partway through the GET. Truncated bytes hash to the wrong SHA-256, which is exactly the "Digest did not match" symptom Docker Registry reports in #8908 (and which persisted even after the per-chunk metadata fix in #9211 and the completion backfill in #9224). Introduce lazyMultipartChunkReader + preparedMultipartChunk{chunk, wrap}: a generic lazy chunk streamer with a per-chunk wrap closure for the SSE-specific decryption setup. Per-chunk metadata is still validated UPFRONT so a malformed chunk fails fast without opening any HTTP connection -- the eager validation contract callers and tests rely on is preserved. The volume-server GET and the SSE-specific decrypt wrap, however, fire LAZILY: at most one chunk body is live at any time, regardless of object size. This commit applies the new pattern to buildMultipartSSES3Reader only; the SSE-KMS and SSE-C multipart readers retain their eager form for now and will be migrated in follow-up commits, since the same shape exists there too. Tests: - TestBuildMultipartSSES3Reader_LazyChunkFetch pins the new contract: zero chunks opened at construction, peak liveness == 1, all closed after drain. - TestBuildMultipartSSES3Reader_RejectsBadChunkBeforeAnyFetch (replaces ClosesAppendedOnError) asserts a malformed chunk in position N causes zero fetches for chunks 0..N -- the previous test pinned a weaker contract (cleanup after eager open). - TestBuildMultipartSSES3Reader_InvalidIVLength updated for the same reason: the fetch callback must NOT be invoked at all on a bad-IV chunk. - TestMultipartSSES3RealisticEndToEnd round-trips multiple parts encrypted the way putToFiler writes them (shared DEK + baseIV, partOffset=0, post-completion global offsets) and walks them through buildMultipartSSES3Reader.
This commit is contained in:
Chris Lu
@@ -489,9 +489,10 @@ func TestBuildMultipartSSES3Reader_InvalidIVLength(t *testing.T) {
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
closed := false
|
||||
fetchCalled := false
|
||||
fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) {
|
||||
return &closeTrackingReadCloser{Reader: bytes.NewReader([]byte("whatever")), closed: &closed}, nil
|
||||
fetchCalled = true
|
||||
return io.NopCloser(bytes.NewReader([]byte("whatever"))), nil
|
||||
}
|
||||
|
||||
chunks := []*filer_pb.FileChunk{
|
||||
@@ -511,24 +512,29 @@ func TestBuildMultipartSSES3Reader_InvalidIVLength(t *testing.T) {
|
||||
if !strings.Contains(err.Error(), "invalid IV length") {
|
||||
t.Errorf("expected 'invalid IV length' in error, got: %v", err)
|
||||
}
|
||||
if !closed {
|
||||
t.Error("chunk reader for the bad chunk was not closed on error")
|
||||
// Validation runs upfront before any chunk fetch, so no volume-server
|
||||
// HTTP connection should have been opened on the failure path.
|
||||
if fetchCalled {
|
||||
t.Error("fetchChunk was called for an invalid-IV chunk; metadata validation should fail before any fetch")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildMultipartSSES3Reader_ClosesAppendedOnError verifies that when a
|
||||
// later chunk fails (e.g., malformed metadata), readers already appended for
|
||||
// earlier valid chunks are closed so volume-server HTTP connections do not leak.
|
||||
func TestBuildMultipartSSES3Reader_ClosesAppendedOnError(t *testing.T) {
|
||||
// TestBuildMultipartSSES3Reader_RejectsBadChunkBeforeAnyFetch verifies that
|
||||
// when any chunk's metadata is malformed, the helper returns an error WITHOUT
|
||||
// having opened a volume-server HTTP connection for any chunk. Per-chunk
|
||||
// metadata is validated upfront precisely so a bad chunk in position N does
|
||||
// not leak open HTTP responses for chunks 0..N-1 (the original eager
|
||||
// implementation depended on a closeAppendedReaders cleanup path; this test
|
||||
// pins the stronger contract: nothing is opened in the first place).
|
||||
func TestBuildMultipartSSES3Reader_RejectsBadChunkBeforeAnyFetch(t *testing.T) {
|
||||
keyManager := initSSES3KeyManagerForTest(t)
|
||||
|
||||
// First chunk: valid SSE-S3 chunk.
|
||||
cipher1, meta1 := encryptSSES3Part(t, []byte("first chunk plaintext"))
|
||||
|
||||
// Second chunk: missing per-chunk metadata, triggers error after first is
|
||||
// already appended.
|
||||
// Second chunk: missing per-chunk metadata, triggers error.
|
||||
chunks := []*filer_pb.FileChunk{
|
||||
{
|
||||
FileId: "1,good",
|
||||
@@ -546,36 +552,20 @@ func TestBuildMultipartSSES3Reader_ClosesAppendedOnError(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
firstClosed := false
|
||||
secondClosed := false
|
||||
fetched := map[string]int{}
|
||||
fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) {
|
||||
switch c.GetFileIdString() {
|
||||
case "1,good":
|
||||
return &closeTrackingReadCloser{Reader: bytes.NewReader(cipher1), closed: &firstClosed}, nil
|
||||
case "2,bad":
|
||||
return &closeTrackingReadCloser{Reader: bytes.NewReader([]byte("x")), closed: &secondClosed}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("unexpected chunk %s", c.GetFileIdString())
|
||||
fetched[c.GetFileIdString()]++
|
||||
return io.NopCloser(bytes.NewReader([]byte("x"))), nil
|
||||
}
|
||||
|
||||
_, err := buildMultipartSSES3Reader(chunks, keyManager, fetch)
|
||||
if err == nil {
|
||||
t.Fatal("expected error from missing chunk metadata, got nil")
|
||||
}
|
||||
if !firstClosed {
|
||||
t.Error("previously appended chunk reader was not closed on error")
|
||||
if !strings.Contains(err.Error(), "missing per-chunk metadata") {
|
||||
t.Errorf("expected 'missing per-chunk metadata' in error, got: %v", err)
|
||||
}
|
||||
if !secondClosed {
|
||||
t.Error("chunk reader for the failing chunk was not closed on error")
|
||||
if len(fetched) != 0 {
|
||||
t.Errorf("expected no chunks fetched on validation failure, got %v", fetched)
|
||||
}
|
||||
}
|
||||
|
||||
type closeTrackingReadCloser struct {
|
||||
io.Reader
|
||||
closed *bool
|
||||
}
|
||||
|
||||
func (r *closeTrackingReadCloser) Close() error {
|
||||
*r.closed = true
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -2786,6 +2786,15 @@ func (s3a *S3ApiServer) createMultipartSSES3DecryptedReaderDirect(ctx context.Co
|
||||
// SSE-S3 chunks. Chunks are fetched via fetchChunk and decrypted using their
|
||||
// per-chunk metadata (each multipart part has its own DEK and IV). Exposed as a
|
||||
// standalone helper so tests can inject a mock chunk fetcher.
|
||||
//
|
||||
// All per-chunk metadata is validated upfront so a malformed chunk fails fast
|
||||
// without opening any HTTP connections to volume servers. The actual chunk
|
||||
// fetch and decryption happens LAZILY as the returned reader is read: at most
|
||||
// one chunk's HTTP connection is open at a time. Eagerly opening every chunk's
|
||||
// HTTP response (the previous behavior) caused later chunks' connections to
|
||||
// sit idle while earlier chunks were still being consumed, which under load
|
||||
// could trip volume-server idle/keepalive limits and yield truncated reads
|
||||
// (issue #8908).
|
||||
func buildMultipartSSES3Reader(chunks []*filer_pb.FileChunk, keyManager *SSES3KeyManager, fetchChunk func(*filer_pb.FileChunk) (io.ReadCloser, error)) (io.Reader, error) {
|
||||
// Sort a copy of the slice so callers do not observe their input chunks
|
||||
// reordered (the backing array is shared with entry.Chunks, which other
|
||||
@@ -2796,82 +2805,138 @@ func buildMultipartSSES3Reader(chunks []*filer_pb.FileChunk, keyManager *SSES3Ke
|
||||
return sortedChunks[i].GetOffset() < sortedChunks[j].GetOffset()
|
||||
})
|
||||
|
||||
// Create readers for each chunk, decrypting them independently
|
||||
readers := make([]io.Reader, 0, len(sortedChunks))
|
||||
|
||||
// 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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Validate every chunk's SSE-S3 metadata before returning a reader. This
|
||||
// keeps the eager-validation contract that callers and tests rely on
|
||||
// (malformed metadata fails immediately), without holding open any
|
||||
// volume-server HTTP connections.
|
||||
preparedChunks := make([]preparedMultipartChunk, 0, len(sortedChunks))
|
||||
for _, chunk := range sortedChunks {
|
||||
// Get this chunk's encrypted data
|
||||
chunkReader, err := fetchChunk(chunk)
|
||||
if chunk.GetSseType() != filer_pb.SSEType_SSE_S3 {
|
||||
preparedChunks = append(preparedChunks, preparedMultipartChunk{chunk: chunk})
|
||||
continue
|
||||
}
|
||||
if len(chunk.GetSseMetadata()) == 0 {
|
||||
return nil, fmt.Errorf("SSE-S3 chunk %s missing per-chunk metadata", chunk.GetFileIdString())
|
||||
}
|
||||
meta, err := DeserializeSSES3Metadata(chunk.GetSseMetadata(), keyManager)
|
||||
if err != nil {
|
||||
closeAppendedReaders()
|
||||
return nil, fmt.Errorf("failed to create chunk reader: %v", err)
|
||||
return nil, fmt.Errorf("failed to deserialize SSE-S3 metadata for chunk %s: %v", chunk.GetFileIdString(), err)
|
||||
}
|
||||
|
||||
// Handle based on chunk's encryption type
|
||||
if chunk.GetSseType() == filer_pb.SSEType_SSE_S3 {
|
||||
// Check if this chunk has per-chunk SSE-S3 metadata
|
||||
if len(chunk.GetSseMetadata()) == 0 {
|
||||
chunkReader.Close()
|
||||
closeAppendedReaders()
|
||||
return nil, fmt.Errorf("SSE-S3 chunk %s missing per-chunk metadata", chunk.GetFileIdString())
|
||||
}
|
||||
|
||||
// Deserialize the per-chunk SSE-S3 metadata to get the IV
|
||||
chunkSSES3Metadata, err := DeserializeSSES3Metadata(chunk.GetSseMetadata(), keyManager)
|
||||
if err != nil {
|
||||
chunkReader.Close()
|
||||
closeAppendedReaders()
|
||||
return nil, fmt.Errorf("failed to deserialize SSE-S3 metadata for chunk %s: %v", chunk.GetFileIdString(), err)
|
||||
}
|
||||
|
||||
// Use the IV from the chunk metadata. DeserializeSSES3Metadata does
|
||||
// not require an IV, so validate the length here before it reaches
|
||||
// cipher.NewCTR, which would otherwise panic on a nil or short IV.
|
||||
iv := chunkSSES3Metadata.IV
|
||||
if len(iv) != s3_constants.AESBlockSize {
|
||||
chunkReader.Close()
|
||||
closeAppendedReaders()
|
||||
return nil, fmt.Errorf("SSE-S3 chunk %s has invalid IV length %d (expected %d)",
|
||||
chunk.GetFileIdString(), len(iv), s3_constants.AESBlockSize)
|
||||
}
|
||||
glog.V(4).Infof("Decrypting SSE-S3 chunk %s with KeyID=%s, IV length=%d",
|
||||
chunk.GetFileIdString(), chunkSSES3Metadata.KeyID, len(iv))
|
||||
|
||||
// Create decrypted reader for this chunk
|
||||
decryptedChunkReader, decErr := CreateSSES3DecryptedReader(chunkReader, chunkSSES3Metadata, iv)
|
||||
if decErr != nil {
|
||||
chunkReader.Close()
|
||||
closeAppendedReaders()
|
||||
return nil, fmt.Errorf("failed to decrypt SSE-S3 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-S3 chunk %s", chunk.GetFileIdString())
|
||||
} else {
|
||||
// Non-SSE-S3 chunk, use as-is
|
||||
readers = append(readers, chunkReader)
|
||||
glog.V(4).Infof("Added non-encrypted reader for chunk %s", chunk.GetFileIdString())
|
||||
// DeserializeSSES3Metadata does not require an IV, so validate the
|
||||
// length here before it reaches cipher.NewCTR, which would otherwise
|
||||
// panic on a nil or short IV.
|
||||
if len(meta.IV) != s3_constants.AESBlockSize {
|
||||
return nil, fmt.Errorf("SSE-S3 chunk %s has invalid IV length %d (expected %d)",
|
||||
chunk.GetFileIdString(), len(meta.IV), s3_constants.AESBlockSize)
|
||||
}
|
||||
// Capture meta and chunk by-value into the wrap closure so each
|
||||
// prepared entry decrypts with its own per-chunk key + IV.
|
||||
fileId := chunk.GetFileIdString()
|
||||
preparedChunks = append(preparedChunks, preparedMultipartChunk{
|
||||
chunk: chunk,
|
||||
wrap: func(raw io.ReadCloser) (io.Reader, error) {
|
||||
glog.V(4).Infof("Decrypting SSE-S3 chunk %s with KeyID=%s, IV length=%d",
|
||||
fileId, meta.KeyID, len(meta.IV))
|
||||
dec, err := CreateSSES3DecryptedReader(raw, meta, meta.IV)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to decrypt SSE-S3 chunk: %v", err)
|
||||
}
|
||||
return dec, nil
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
return NewMultipartSSEReader(readers), nil
|
||||
return &lazyMultipartChunkReader{
|
||||
chunks: preparedChunks,
|
||||
fetch: fetchChunk,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// preparedMultipartChunk pairs a chunk with the per-SSE wrapping logic the
|
||||
// lazy reader applies to its raw HTTP body. wrap is nil for chunks that
|
||||
// stream as-is (no SSE on the chunk, even though the object is multipart-SSE);
|
||||
// otherwise wrap is the SSE-specific decryption setup, which receives the
|
||||
// already-opened raw chunk body and returns the plaintext reader.
|
||||
type preparedMultipartChunk struct {
|
||||
chunk *filer_pb.FileChunk
|
||||
wrap func(raw io.ReadCloser) (io.Reader, error)
|
||||
}
|
||||
|
||||
// lazyMultipartChunkReader streams a sequence of multipart chunks one at a
|
||||
// time. It opens each chunk's underlying HTTP fetch (and applies the
|
||||
// SSE-specific decryption wrapper) only when the previous chunk has been
|
||||
// fully consumed, so volume-server connections do not pile up for large
|
||||
// objects. This is the same shape used by all three SSE multipart read
|
||||
// paths (SSE-S3, SSE-KMS, SSE-C); only the per-chunk wrap closure differs.
|
||||
type lazyMultipartChunkReader struct {
|
||||
chunks []preparedMultipartChunk
|
||||
fetch func(*filer_pb.FileChunk) (io.ReadCloser, error)
|
||||
idx int
|
||||
current io.Reader // current chunk's plaintext reader (or raw reader for non-SSE chunks)
|
||||
closer io.Closer // current chunk's underlying HTTP body, to close on advance/Close
|
||||
finished bool
|
||||
}
|
||||
|
||||
func (l *lazyMultipartChunkReader) Read(p []byte) (int, error) {
|
||||
for {
|
||||
if l.finished {
|
||||
return 0, io.EOF
|
||||
}
|
||||
if l.current == nil {
|
||||
if l.idx >= len(l.chunks) {
|
||||
l.finished = true
|
||||
return 0, io.EOF
|
||||
}
|
||||
pc := l.chunks[l.idx]
|
||||
l.idx++
|
||||
chunkReader, err := l.fetch(pc.chunk)
|
||||
if err != nil {
|
||||
l.finished = true
|
||||
return 0, fmt.Errorf("failed to create chunk reader: %v", err)
|
||||
}
|
||||
if pc.wrap == nil {
|
||||
// Non-SSE chunk in an otherwise SSE-multipart object: stream
|
||||
// raw bytes through.
|
||||
l.current = chunkReader
|
||||
l.closer = chunkReader
|
||||
glog.V(4).Infof("Streaming non-encrypted chunk %s", pc.chunk.GetFileIdString())
|
||||
} else {
|
||||
wrapped, wrapErr := pc.wrap(chunkReader)
|
||||
if wrapErr != nil {
|
||||
chunkReader.Close()
|
||||
l.finished = true
|
||||
return 0, wrapErr
|
||||
}
|
||||
l.current = wrapped
|
||||
l.closer = chunkReader
|
||||
}
|
||||
}
|
||||
n, err := l.current.Read(p)
|
||||
if err == io.EOF {
|
||||
closeErr := l.closer.Close()
|
||||
l.current = nil
|
||||
l.closer = nil
|
||||
if n > 0 {
|
||||
return n, nil
|
||||
}
|
||||
if closeErr != nil {
|
||||
glog.V(2).Infof("Error closing chunk reader: %v", closeErr)
|
||||
}
|
||||
continue
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
}
|
||||
|
||||
func (l *lazyMultipartChunkReader) Close() error {
|
||||
l.finished = true
|
||||
if l.closer != nil {
|
||||
err := l.closer.Close()
|
||||
l.current = nil
|
||||
l.closer = nil
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// createEncryptedChunkReader creates a reader for a single encrypted chunk
|
||||
|
||||
@@ -0,0 +1,280 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/rand"
|
||||
"fmt"
|
||||
"io"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
)
|
||||
|
||||
// TestMultipartSSES3RealisticEndToEnd reproduces the production multipart SSE-S3
|
||||
// flow where ALL parts share the same DEK and baseIV (the upload-init key/IV),
|
||||
// and each part is encrypted with partOffset=0. Each part is then chunked at 8MB
|
||||
// boundaries the way UploadReaderInChunks does. After completion, the chunks
|
||||
// have global offsets but per-chunk stored IVs derived from part-local offsets.
|
||||
//
|
||||
// buildMultipartSSES3Reader is then run on the assembled chunks; the
|
||||
// concatenated decrypted output must equal the concatenation of the part
|
||||
// plaintexts. This is the round trip that fails in #8908 if anything in the
|
||||
// encrypt or decrypt path is inconsistent.
|
||||
func TestMultipartSSES3RealisticEndToEnd(t *testing.T) {
|
||||
keyManager := initSSES3KeyManagerForTest(t)
|
||||
|
||||
// One DEK and one baseIV, shared by all parts (the upload-init values).
|
||||
key, err := GenerateSSES3Key()
|
||||
if err != nil {
|
||||
t.Fatalf("GenerateSSES3Key: %v", err)
|
||||
}
|
||||
baseIV := make([]byte, s3_constants.AESBlockSize)
|
||||
if _, err := rand.Read(baseIV); err != nil {
|
||||
t.Fatalf("rand.Read baseIV: %v", err)
|
||||
}
|
||||
|
||||
const chunkSize = int64(8 * 1024 * 1024)
|
||||
|
||||
// Realistic mix of part sizes: small (one chunk), exact 8MB, >8MB (two
|
||||
// chunks), much larger (multiple chunks).
|
||||
partSizes := []int{
|
||||
5 * 1024 * 1024, // 5MB (single chunk)
|
||||
8 * 1024 * 1024, // 8MB exactly (single chunk, full)
|
||||
8*1024*1024 + 123, // crosses chunk boundary (two chunks)
|
||||
17 * 1024 * 1024, // three chunks
|
||||
1234, // tiny
|
||||
}
|
||||
|
||||
parts := make([][]byte, len(partSizes))
|
||||
for i, n := range partSizes {
|
||||
parts[i] = makeRandomPlaintext(t, n)
|
||||
}
|
||||
|
||||
// Build the chunks list the way completion would produce it: encrypt each
|
||||
// part with partOffset=0, slice the ciphertext at chunkSize boundaries,
|
||||
// store per-chunk metadata IV = calculateIVWithOffset(baseIV, partLocalOff),
|
||||
// then assign GLOBAL offsets to the FileChunk.
|
||||
type chunkBlob struct {
|
||||
fid string
|
||||
ciphertext []byte
|
||||
}
|
||||
var chunks []*filer_pb.FileChunk
|
||||
chunkData := map[string][]byte{}
|
||||
var globalOffset int64
|
||||
for partIdx, partPlaintext := range parts {
|
||||
encReader, _, err := CreateSSES3EncryptedReaderWithBaseIV(bytes.NewReader(partPlaintext), key, baseIV, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("CreateSSES3EncryptedReaderWithBaseIV(part %d): %v", partIdx, err)
|
||||
}
|
||||
ciphertext, err := io.ReadAll(encReader)
|
||||
if err != nil {
|
||||
t.Fatalf("read encrypted part %d: %v", partIdx, err)
|
||||
}
|
||||
|
||||
for partLocalOff := int64(0); partLocalOff < int64(len(ciphertext)); partLocalOff += chunkSize {
|
||||
end := partLocalOff + chunkSize
|
||||
if end > int64(len(ciphertext)) {
|
||||
end = int64(len(ciphertext))
|
||||
}
|
||||
cipherSlice := ciphertext[partLocalOff:end]
|
||||
|
||||
chunkIV, _ := calculateIVWithOffset(baseIV, partLocalOff)
|
||||
chunkKey := &SSES3Key{
|
||||
Key: key.Key,
|
||||
KeyID: key.KeyID,
|
||||
Algorithm: key.Algorithm,
|
||||
IV: chunkIV,
|
||||
}
|
||||
meta, err := SerializeSSES3Metadata(chunkKey)
|
||||
if err != nil {
|
||||
t.Fatalf("SerializeSSES3Metadata(part %d off %d): %v", partIdx, partLocalOff, err)
|
||||
}
|
||||
|
||||
fid := fmt.Sprintf("%d,%d", partIdx+1, partLocalOff)
|
||||
chunks = append(chunks, &filer_pb.FileChunk{
|
||||
FileId: fid,
|
||||
Offset: globalOffset, // global offset assigned at completion
|
||||
Size: uint64(end - partLocalOff),
|
||||
SseType: filer_pb.SSEType_SSE_S3,
|
||||
SseMetadata: meta,
|
||||
})
|
||||
chunkData[fid] = cipherSlice
|
||||
globalOffset += end - partLocalOff
|
||||
}
|
||||
}
|
||||
|
||||
fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) {
|
||||
data, ok := chunkData[c.GetFileIdString()]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unexpected chunk %s", c.GetFileIdString())
|
||||
}
|
||||
return io.NopCloser(bytes.NewReader(data)), nil
|
||||
}
|
||||
|
||||
reader, err := buildMultipartSSES3Reader(chunks, keyManager, fetch)
|
||||
if err != nil {
|
||||
t.Fatalf("buildMultipartSSES3Reader: %v", err)
|
||||
}
|
||||
got, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadAll decrypted output: %v", err)
|
||||
}
|
||||
|
||||
want := bytes.Join(parts, nil)
|
||||
if !bytes.Equal(got, want) {
|
||||
idx := firstMismatch(got, want)
|
||||
end := idx + 32
|
||||
if end > len(got) {
|
||||
end = len(got)
|
||||
}
|
||||
if end > len(want) {
|
||||
end = len(want)
|
||||
}
|
||||
t.Fatalf("decrypted output mismatch at byte %d (total len got=%d want=%d)\n got: %x\nwant: %x",
|
||||
idx, len(got), len(want), got[idx:end], want[idx:end])
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildMultipartSSES3Reader_LazyChunkFetch pins the lazy behavior of
|
||||
// buildMultipartSSES3Reader: chunk N's HTTP fetch only happens after chunk
|
||||
// N-1 has been fully consumed. The original eager loop opened every chunk's
|
||||
// HTTP response upfront and held them open while io.MultiReader walked
|
||||
// through readers[0]; for objects with many chunks (e.g. a 200MB Docker image
|
||||
// blob), this could trip volume-server idle/keepalive limits and produce
|
||||
// truncated reads at the client (issue #8908).
|
||||
//
|
||||
// The test installs a fetch hook that tracks how many chunks have been
|
||||
// opened and when each one is closed, and verifies:
|
||||
// - At any point during streaming, at most one chunk's reader is open.
|
||||
// - The number of opened chunks grows as bytes are read out, not upfront.
|
||||
// - All chunks are closed when the outer reader is fully drained.
|
||||
func TestBuildMultipartSSES3Reader_LazyChunkFetch(t *testing.T) {
|
||||
keyManager := initSSES3KeyManagerForTest(t)
|
||||
|
||||
key, err := GenerateSSES3Key()
|
||||
if err != nil {
|
||||
t.Fatalf("GenerateSSES3Key: %v", err)
|
||||
}
|
||||
baseIV := make([]byte, s3_constants.AESBlockSize)
|
||||
if _, err := rand.Read(baseIV); err != nil {
|
||||
t.Fatalf("rand.Read baseIV: %v", err)
|
||||
}
|
||||
|
||||
// Many small chunks (mirrors many-part Docker Registry uploads).
|
||||
const numChunks = 8
|
||||
const chunkPayload = 1024
|
||||
plaintexts := make([][]byte, numChunks)
|
||||
chunkData := map[string][]byte{}
|
||||
chunks := make([]*filer_pb.FileChunk, 0, numChunks)
|
||||
for i := 0; i < numChunks; i++ {
|
||||
plaintexts[i] = makeRandomPlaintext(t, chunkPayload)
|
||||
|
||||
// Encrypt as a fresh "part" with partOffset=0 (matching putToFiler).
|
||||
encReader, _, err := CreateSSES3EncryptedReaderWithBaseIV(bytes.NewReader(plaintexts[i]), key, baseIV, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("encrypt chunk %d: %v", i, err)
|
||||
}
|
||||
ciphertext, err := io.ReadAll(encReader)
|
||||
if err != nil {
|
||||
t.Fatalf("read ciphertext %d: %v", i, err)
|
||||
}
|
||||
|
||||
chunkIV, _ := calculateIVWithOffset(baseIV, 0)
|
||||
chunkKey := &SSES3Key{
|
||||
Key: key.Key,
|
||||
KeyID: key.KeyID,
|
||||
Algorithm: key.Algorithm,
|
||||
IV: chunkIV,
|
||||
}
|
||||
meta, err := SerializeSSES3Metadata(chunkKey)
|
||||
if err != nil {
|
||||
t.Fatalf("serialize meta %d: %v", i, err)
|
||||
}
|
||||
|
||||
fid := fmt.Sprintf("vol,c%d", i)
|
||||
chunks = append(chunks, &filer_pb.FileChunk{
|
||||
FileId: fid,
|
||||
Offset: int64(i) * chunkPayload,
|
||||
Size: uint64(chunkPayload),
|
||||
SseType: filer_pb.SSEType_SSE_S3,
|
||||
SseMetadata: meta,
|
||||
})
|
||||
chunkData[fid] = ciphertext
|
||||
}
|
||||
|
||||
var openCount int64 // total opens
|
||||
var liveCount int64 // currently open
|
||||
var maxLive int64
|
||||
|
||||
fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) {
|
||||
atomic.AddInt64(&openCount, 1)
|
||||
if live := atomic.AddInt64(&liveCount, 1); live > atomic.LoadInt64(&maxLive) {
|
||||
atomic.StoreInt64(&maxLive, live)
|
||||
}
|
||||
data, ok := chunkData[c.GetFileIdString()]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unexpected chunk %s", c.GetFileIdString())
|
||||
}
|
||||
return &liveTrackingReadCloser{Reader: bytes.NewReader(data), live: &liveCount}, nil
|
||||
}
|
||||
|
||||
reader, err := buildMultipartSSES3Reader(chunks, keyManager, fetch)
|
||||
if err != nil {
|
||||
t.Fatalf("buildMultipartSSES3Reader: %v", err)
|
||||
}
|
||||
|
||||
// Construction alone must not have opened any chunk reader.
|
||||
if got := atomic.LoadInt64(&openCount); got != 0 {
|
||||
t.Fatalf("expected no chunks opened before any Read, got %d", got)
|
||||
}
|
||||
|
||||
// Read first byte: should open chunk 0 only.
|
||||
one := make([]byte, 1)
|
||||
if n, err := reader.Read(one); n != 1 || err != nil {
|
||||
t.Fatalf("first Read: n=%d err=%v", n, err)
|
||||
}
|
||||
if got := atomic.LoadInt64(&openCount); got != 1 {
|
||||
t.Errorf("after first byte, expected 1 chunk opened, got %d", got)
|
||||
}
|
||||
if got := atomic.LoadInt64(&liveCount); got != 1 {
|
||||
t.Errorf("after first byte, expected 1 chunk live, got %d", got)
|
||||
}
|
||||
|
||||
// Drain the rest.
|
||||
rest, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
t.Fatalf("drain: %v", err)
|
||||
}
|
||||
got := append(one, rest...)
|
||||
want := bytes.Join(plaintexts, nil)
|
||||
if !bytes.Equal(got, want) {
|
||||
idx := firstMismatch(got, want)
|
||||
t.Fatalf("decrypted output mismatch at byte %d (got len %d, want len %d)", idx, len(got), len(want))
|
||||
}
|
||||
if got := atomic.LoadInt64(&openCount); got != int64(numChunks) {
|
||||
t.Errorf("expected exactly %d chunk opens after drain, got %d", numChunks, got)
|
||||
}
|
||||
if got := atomic.LoadInt64(&maxLive); got > 1 {
|
||||
t.Errorf("expected at most 1 chunk reader live at a time (lazy), saw peak of %d", got)
|
||||
}
|
||||
if got := atomic.LoadInt64(&liveCount); got != 0 {
|
||||
t.Errorf("expected all chunks closed after drain, %d still live", got)
|
||||
}
|
||||
}
|
||||
|
||||
type liveTrackingReadCloser struct {
|
||||
io.Reader
|
||||
live *int64
|
||||
once bool
|
||||
}
|
||||
|
||||
func (r *liveTrackingReadCloser) Close() error {
|
||||
if r.once {
|
||||
return nil
|
||||
}
|
||||
r.once = true
|
||||
atomic.AddInt64(r.live, -1)
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user