s3sink: upload through a concurrent-safe ReaderAt so parts do not thrash

s3manager splits a seekable body into parallel section readers, and every
part's ReadAt landed on the single chunk buffer in ChunkStreamReader: parts
at different offsets kept evicting each other's chunk and refetching it
whole, which produced massive read amplification against the volume
servers and collapsed throughput.

Build the upload body from filer.ChunkReadAt instead, the same
concurrent-safe reader the mount and query paths use: each part's range is
served from a shared reader cache with parallel chunk fetches and
range-reads for clipped views, so every byte is fetched once per pass.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/replication/sink/s3sink/s3_sink.go

@@ -1,8 +1,12 @@
 package S3Sink
 
 import (
+	"bytes"
+	"context"
 	"encoding/base64"
 	"fmt"
+	"io"
+	"math"
 	"net/url"
 	"strconv"
 	"strings"
@@ -22,6 +26,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
 )
 
 type S3Sink struct {
@@ -164,15 +169,21 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
 		return nil
 	}
 
-	reader := filer.NewFileReader(s3sink.filerSource, entry)
+	reader, closeReader, err := s3sink.entryReader(entry)
+	if err != nil {
+		return err
+	}
 
 	// Decrypt SSE-encrypted objects so the destination receives plaintext
 	decryptedReader, err := repl_util.MaybeDecryptReader(reader, entry)
 	if err != nil {
-		repl_util.CloseReader(reader)
+		closeReader()
 		return fmt.Errorf("decrypt SSE object: %w", err)
 	}
-	defer repl_util.CloseMaybeDecryptedReader(reader, decryptedReader)
+	defer func() {
+		repl_util.CloseMaybeDecryptedReader(reader, decryptedReader)
+		closeReader()
+	}()
 
 	// Create an uploader with the session and custom options
 	uploader := s3manager.NewUploaderWithClient(s3sink.conn, func(u *s3manager.Uploader) {
@@ -220,16 +231,35 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
 		uploadInput.ContentMD5 = aws.String(base64.StdEncoding.EncodeToString([]byte(entry.Attributes.Md5)))
 	}
 	_, err = uploader.Upload(&uploadInput)
-	if err != nil {
-		if sourceErr := filer.ReaderSourceError(reader); sourceErr != nil {
-			return fmt.Errorf("read source %s: %w", key, sourceErr)
-		}
-	}
 
 	return err
 
 }
 
+// entryReader returns a reader over the entry's content for the uploader.
+// Chunked entries are read through a concurrent-safe ReaderAt with its own
+// chunk cache, so parallel part readers each fetch the ranges they need; a
+// single stream reader buffers only one chunk and concurrent parts would keep
+// evicting it and refetching from the volume servers.
+func (s3sink *S3Sink) entryReader(entry *filer_pb.Entry) (reader io.Reader, closeFn func(), err error) {
+	if len(entry.Content) > 0 {
+		return bytes.NewReader(entry.Content), func() {}, nil
+	}
+
+	lookupFileIdFn := filer.LookupFn(s3sink.filerSource)
+	fileSize := int64(filer.FileSize(entry))
+	chunkViews, err := filer.ViewFromChunks(context.Background(), lookupFileIdFn, entry.GetChunks(), 0, math.MaxInt64)
+	if err != nil {
+		return nil, nil, fmt.Errorf("resolve chunks: %w", err)
+	}
+
+	concurrency := max(s3sink.uploaderConcurrency, 1) + filer.DefaultPrefetchCount
+	readerAt := filer.NewChunkReaderAtFromClient(context.Background(),
+		filer.NewReaderCache(concurrency, chunk_cache.NewChunkCacheInMemory(int64(concurrency)), lookupFileIdFn, nil),
+		chunkViews, fileSize, filer.DefaultPrefetchCount)
+	return io.NewSectionReader(readerAt, 0, fileSize), func() { readerAt.Close() }, nil
+}
+
 func (s3sink *S3Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
 	key = cleanKey(key)
 	return true, s3sink.CreateEntry(key, newEntry, signatures)

weed/replication/sink/s3sink/s3_sink_test.go

@@ -94,6 +94,48 @@ func newCapturingS3Sink(t *testing.T) (*S3Sink, *capturePutRoundTripper) {
 	return s, rt
 }
 
+func TestS3SinkEntryReaderSupportsConcurrentParts(t *testing.T) {
+	sink := &S3Sink{uploaderConcurrency: 8}
+	entry := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+		Chunks: []*filer_pb.FileChunk{
+			{FileId: "1,01637037d6", Size: 100},
+		},
+	}
+
+	reader, closeFn, err := sink.entryReader(entry)
+	require.NoError(t, err)
+	defer closeFn()
+
+	// s3manager only splits a body into parallel part readers when it is both
+	// an io.ReaderAt and an io.Seeker; the reader must be safe for concurrent
+	// ReadAt so parts do not evict one another's buffered chunk.
+	_, isReaderAt := reader.(io.ReaderAt)
+	seeker, isSeeker := reader.(io.Seeker)
+	require.True(t, isReaderAt, "upload body should support concurrent ReadAt")
+	require.True(t, isSeeker, "upload body should be seekable")
+
+	size, err := seeker.Seek(0, io.SeekEnd)
+	require.NoError(t, err)
+	require.Equal(t, int64(100), size)
+}
+
+func TestS3SinkEntryReaderForInlineContent(t *testing.T) {
+	sink := &S3Sink{uploaderConcurrency: 8}
+	entry := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{FileSize: 4},
+		Content:    []byte("data"),
+	}
+
+	reader, closeFn, err := sink.entryReader(entry)
+	require.NoError(t, err)
+	defer closeFn()
+
+	content, err := io.ReadAll(reader)
+	require.NoError(t, err)
+	require.Equal(t, []byte("data"), content)
+}
+
 func TestS3SinkCreateEntryPassesMimeAsContentType(t *testing.T) {
 	sink, rt := newCapturingS3Sink(t)
 	entry := &filer_pb.Entry{