feat(s3api): full-chunk gzip pass-through skips volume-side decompress

Building on the io.Pipe streaming chunk copy: when a copy operation
covers an entire source chunk (the common case for Harbor's
part-size = chunk-size assemble pattern), ask the source volume for
compressed bytes via Accept-Encoding: gzip and forward them to the
destination as-is.

This trades a Range fetch (where the volume decompresses the chunk
internally to satisfy the byte range) for a full-chunk fetch that
returns whatever wire bytes the chunk is stored as. For gzipped
chunks the source volume avoids the decompression entirely; we never
allocate a chunk-sized decompress buffer.

Implementation: build the source GET directly instead of going
through ReadUrlAsStream, because that helper auto-decompresses gzip
responses (which would defeat the point). Trust the response's
Content-Encoding header over caller hints — for partial ranges the
volume always returns raw bytes regardless of how the chunk is
stored, so labeling those as gzip would corrupt subsequent reads.

End-to-end repro impact (512 MiB src, 6 parallel UploadPartCopy):
  + #9420/#9421/#9422       : 2236 MiB
  + io.Pipe streaming       : 1521 MiB
  + this commit             : 1149 MiB  (round 2 RSS, perfectly flat)

Round 3 now completes (was hitting volume-full before, since
chunks took up uncompressed space on disk; we now store the gzipped
chunks the volume gives us, which fit in the test's 8 GiB volume
budget).

Heap inuse_space (after force GC):
  before all: ~1.5 GiB
  this PR:    266 MiB

Volume-side bytes.Buffer.ReadFrom inuse:
  before:     611 MiB
  streaming:  571 MiB
  this PR:    297 MiB (now in destination-volume parseUpload's
                       size-hint decompression — separate
                       optimization opportunity for a hint header)

weed/s3api/s3api_object_handlers_copy.go

@@ -1053,7 +1053,20 @@ func (s3a *S3ApiServer) copySingleChunk(chunk *filer_pb.FileChunk, dstPath strin
 		return nil, err
 	}
 
-	// Download and upload the chunk
+	// Stream the chunk through io.Pipe when no in-transit transformation is
+	// required; this holds only ~32 KiB per copy in flight, vs. the
+	// chunk-sized buffers the buffered path needs. isFullChunk=true asks
+	// the source volume for compressed bytes, so a gzipped chunk is
+	// forwarded to the destination without anyone having to decompress.
+	if canStreamCopyChunk(chunk) {
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), true /*isFullChunk*/, assignResult); err != nil {
+			return nil, fmt.Errorf("stream chunk: %w", err)
+		}
+		return dstChunk, nil
+	}
+
+	// SSE / per-chunk-cipher: bytes need to be transformed in transit, so
+	// download into a buffer first.
 	chunkData, err := s3a.downloadChunkData(srcUrl, fileId, 0, int64(chunk.Size), chunk.CipherKey)
 	if err != nil {
 		return nil, fmt.Errorf("download chunk data: %w", err)
@@ -1088,6 +1101,25 @@ func (s3a *S3ApiServer) copySingleChunkForRange(originalChunk, rangeChunk *filer
 	overlapStart := max(rangeStart, chunkStart)
 	offsetInChunk := overlapStart - chunkStart
 
+	// Stream the byte range through io.Pipe when there's no in-transit
+	// transformation required (see canStreamCopyChunk for the eligibility
+	// rules); this is the dominant path for Harbor-style multipart
+	// assemble loads, which use UploadPartCopy with a CopySourceRange.
+	//
+	// When the requested range happens to cover the entire source chunk
+	// exactly, switch to the full-chunk fetch mode: that asks the source
+	// volume for compressed bytes (Accept-Encoding: gzip) and forwards
+	// them as-is, avoiding the volume-side decompression that a Range
+	// fetch on a gzipped chunk would otherwise pay. Harbor's typical
+	// part-size = chunk-size assemble pattern hits this branch.
+	if canStreamCopyChunk(originalChunk) {
+		isFullChunk := offsetInChunk == 0 && rangeChunk.Size == originalChunk.Size
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), isFullChunk, assignResult); err != nil {
+			return nil, fmt.Errorf("stream chunk range: %w", err)
+		}
+		return dstChunk, nil
+	}
+
 	// Download and upload the chunk portion
 	chunkData, err := s3a.downloadChunkData(srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), originalChunk.CipherKey)
 	if err != nil {
@@ -1379,16 +1411,7 @@ func (s3a *S3ApiServer) prepareChunkCopy(sourceFileId, dstPath string, expectedD
 // uploadChunkData uploads chunk data to the destination using common upload logic
 // isCompressed indicates if the data is already compressed and should not be compressed again
 func (s3a *S3ApiServer) uploadChunkData(chunkData []byte, assignResult *filer_pb.AssignVolumeResponse, isCompressed bool) error {
-	dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId)
-
-	uploadOption := &operation.UploadOption{
-		UploadUrl:         dstUrl,
-		Cipher:            false, // Data is already encrypted if source had CipherKey; don't re-encrypt
-		IsInputCompressed: isCompressed,
-		MimeType:          "",
-		PairMap:           nil,
-		Jwt:               security.EncodedJwt(assignResult.Auth),
-	}
+	uploadOption := newChunkUploadOption(chunkData, assignResult, isCompressed)
 	uploader, err := operation.NewUploader()
 	if err != nil {
 		return fmt.Errorf("create uploader: %w", err)
@@ -1401,6 +1424,33 @@ func (s3a *S3ApiServer) uploadChunkData(chunkData []byte, assignResult *filer_pb
 	return nil
 }
 
+// multipartFramingOverhead reserves space for the multipart wrapper
+// upload_content writes around chunkData (boundary + Content-Disposition +
+// optional Content-Type/Content-Encoding/Content-MD5 headers + trailing
+// boundary). Real-world overhead is a few hundred bytes; rounding to 1 KiB
+// avoids a single grow on the buffer we hand to the multipart writer.
+const multipartFramingOverhead = 1024
+
+// newChunkUploadOption builds the operation.UploadOption used by every
+// chunk-copy upload. It always sets BytesBuffer to a fresh, per-call buffer
+// so upload_content does not fall back to the package-global
+// valyala/bytebufferpool — that pool retains every high-water buffer for the
+// process's lifetime, and under concurrent UploadPartCopy load it hoarded
+// one chunk-sized buffer per concurrent upload (see #6541). The per-call
+// buffer is GC'd as soon as the upload returns.
+func newChunkUploadOption(chunkData []byte, assignResult *filer_pb.AssignVolumeResponse, isCompressed bool) *operation.UploadOption {
+	dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId)
+	return &operation.UploadOption{
+		UploadUrl:         dstUrl,
+		Cipher:            false, // Data is already encrypted if source had CipherKey; don't re-encrypt
+		IsInputCompressed: isCompressed,
+		MimeType:          "",
+		PairMap:           nil,
+		Jwt:               security.EncodedJwt(assignResult.Auth),
+		BytesBuffer:       bytes.NewBuffer(make([]byte, 0, len(chunkData)+multipartFramingOverhead)),
+	}
+}
+
 // downloadChunkData downloads chunk data from the source URL
 func (s3a *S3ApiServer) downloadChunkData(srcUrl, fileId string, offset, size int64, cipherKey []byte) ([]byte, error) {
 	jwt := filer.JwtForVolumeServer(fileId)

weed/s3api/s3api_object_handlers_copy_bench_test.go

@@ -0,0 +1,186 @@
+package s3api
+
+import (
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+)
+
+// benchEnv stands up a fake source volume (returns N bytes on GET) and a
+// fake destination volume (parses an incoming multipart POST and discards
+// the body, asserting the data arrived intact via SHA-256). It returns the
+// URLs and an AssignVolumeResponse pointing at the destination.
+type benchEnv struct {
+	srcSrv  *httptest.Server
+	dstSrv  *httptest.Server
+	payload []byte
+	dstHash [32]byte
+	assign  *filer_pb.AssignVolumeResponse
+}
+
+func newBenchEnv(b *testing.B, payloadSize int) *benchEnv {
+	b.Helper()
+	payload := make([]byte, payloadSize)
+	for i := range payload {
+		payload[i] = byte(i*31 + 7) // arbitrary repeatable pattern
+	}
+	wantHash := sha256.Sum256(payload)
+
+	srcSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Honor Range header if present (UploadPartCopy passes one).
+		start, end := int64(0), int64(len(payload)-1)
+		if rng := r.Header.Get("Range"); rng != "" {
+			var s, e int64
+			if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &s, &e); err == nil {
+				start, end = s, e
+			}
+		}
+		w.Header().Set("Content-Length", strconv.FormatInt(end-start+1, 10))
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write(payload[start : end+1])
+	}))
+
+	dstSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mr, err := r.MultipartReader()
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		part, err := mr.NextPart()
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		h := sha256.New()
+		if _, err := io.Copy(h, part); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		var got [32]byte
+		copy(got[:], h.Sum(nil))
+		if got != wantHash {
+			http.Error(w, "hash mismatch", http.StatusBadRequest)
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+		_, _ = io.WriteString(w, `{"size":`+strconv.Itoa(payloadSize)+`}`)
+	}))
+
+	dstURL, _ := neturl(dstSrv.URL)
+	assign := &filer_pb.AssignVolumeResponse{
+		FileId: "1,benchmark",
+		Location: &filer_pb.Location{
+			Url:       dstURL,
+			PublicUrl: dstURL,
+		},
+		Auth: "",
+	}
+
+	return &benchEnv{
+		srcSrv:  srcSrv,
+		dstSrv:  dstSrv,
+		payload: payload,
+		dstHash: wantHash,
+		assign:  assign,
+	}
+}
+
+func (e *benchEnv) close() {
+	e.srcSrv.Close()
+	e.dstSrv.Close()
+}
+
+// neturl extracts host:port from "http://host:port".
+func neturl(rawURL string) (string, error) {
+	const prefix = "http://"
+	if len(rawURL) < len(prefix) || rawURL[:len(prefix)] != prefix {
+		return "", fmt.Errorf("unexpected URL: %q", rawURL)
+	}
+	host := rawURL[len(prefix):]
+	if _, _, err := net.SplitHostPort(host); err != nil {
+		return "", err
+	}
+	return host, nil
+}
+
+// BenchmarkCopyChunk_Buffered measures the cost of the existing buffered
+// chunk-copy path (downloadChunkData + uploadChunkData). The path
+// allocates one chunk-sized []byte and one multipart-encoded buffer per
+// call.
+func BenchmarkCopyChunk_Buffered(b *testing.B) {
+	for _, size := range []int{1 << 20, 8 << 20, 64 << 20} {
+		b.Run(humanByteName(size), func(b *testing.B) {
+			util_http.InitGlobalHttpClient()
+			env := newBenchEnv(b, size)
+			defer env.close()
+
+			s3a := &S3ApiServer{}
+			b.ResetTimer()
+			b.SetBytes(int64(size))
+			b.ReportAllocs()
+
+			for i := 0; i < b.N; i++ {
+				data, err := s3a.downloadChunkData(env.srcSrv.URL, env.assign.FileId, 0, int64(size), nil)
+				if err != nil {
+					b.Fatalf("download: %v", err)
+				}
+				if err := s3a.uploadChunkData(data, env.assign, false); err != nil {
+					b.Fatalf("upload: %v", err)
+				}
+			}
+		})
+	}
+}
+
+// BenchmarkCopyChunk_Streamed measures the cost of the io.Pipe streaming
+// chunk-copy path. The path holds only the pipe's hand-off buffer (~32
+// KiB) plus http transport buffers per call, regardless of chunk size.
+func BenchmarkCopyChunk_Streamed(b *testing.B) {
+	for _, size := range []int{1 << 20, 8 << 20, 64 << 20} {
+		b.Run(humanByteName(size), func(b *testing.B) {
+			util_http.InitGlobalHttpClient()
+			env := newBenchEnv(b, size)
+			defer env.close()
+
+			s3a := &S3ApiServer{}
+			b.ResetTimer()
+			b.SetBytes(int64(size))
+			b.ReportAllocs()
+
+			for i := 0; i < b.N; i++ {
+				if err := s3a.streamCopyChunkRange(context.Background(),
+					env.srcSrv.URL, env.assign.FileId, 0, int64(size),
+					true /*isFullChunk*/, env.assign); err != nil {
+					b.Fatalf("stream: %v", err)
+				}
+			}
+		})
+	}
+}
+
+func humanByteName(n int) string {
+	switch {
+	case n >= 1<<30:
+		return strconv.Itoa(n>>30) + "GiB"
+	case n >= 1<<20:
+		return strconv.Itoa(n>>20) + "MiB"
+	case n >= 1<<10:
+		return strconv.Itoa(n>>10) + "KiB"
+	default:
+		return strconv.Itoa(n) + "B"
+	}
+}
+
+// Compile-time sanity: ensure the multipart library version we depend on
+// is the one whose framing we mirror in streamCopyChunkRange.
+var _ = multipart.NewWriter

weed/s3api/s3api_object_handlers_copy_chunk_upload_test.go

@@ -0,0 +1,77 @@
+package s3api
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+// TestNewChunkUploadOption_AvoidsBytePool is a regression guard for the
+// s3-side retention amplifier in https://github.com/seaweedfs/seaweedfs/issues/6541.
+//
+// operation.upload_content falls back to the package-global
+// valyala/bytebufferpool (which retains high-water buffers for the
+// process's lifetime) whenever UploadOption.BytesBuffer is nil. Concurrent
+// UploadPartCopy calls then hoard one chunk-sized buffer per concurrent
+// upload in that pool, and RSS never recedes. The chunk-copy path must
+// therefore always provide its own buffer; this test pins that contract.
+func TestNewChunkUploadOption_AvoidsBytePool(t *testing.T) {
+	cases := []struct {
+		name     string
+		chunkLen int
+	}{
+		{"empty chunk", 0},
+		{"small chunk", 1024},
+		{"typical 8 MiB chunk", 8 * 1024 * 1024},
+		{"large 64 MiB chunk", 64 * 1024 * 1024},
+	}
+
+	assign := &filer_pb.AssignVolumeResponse{
+		FileId: "1,foo",
+		Location: &filer_pb.Location{
+			Url:       "127.0.0.1:8080",
+			PublicUrl: "127.0.0.1:8080",
+		},
+		Auth: "",
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			data := make([]byte, c.chunkLen)
+			opt := newChunkUploadOption(data, assign, false)
+
+			if opt.BytesBuffer == nil {
+				t.Fatalf("BytesBuffer must be non-nil so chunk uploads bypass " +
+					"the package-global bytebufferpool (regression: #6541)")
+			}
+			if got, want := opt.BytesBuffer.Cap(), c.chunkLen+multipartFramingOverhead; got < want {
+				t.Errorf("BytesBuffer cap=%d, want >=%d (one alloc-and-grow "+
+					"per upload defeats the purpose of pre-sizing)", got, want)
+			}
+			if opt.BytesBuffer.Len() != 0 {
+				t.Errorf("BytesBuffer not empty: len=%d", opt.BytesBuffer.Len())
+			}
+			if opt.Cipher {
+				t.Errorf("Cipher must be false: chunk-copy data is already " +
+					"encrypted if the source had a CipherKey")
+			}
+		})
+	}
+}
+
+// TestNewChunkUploadOption_PerCallIsolation verifies each call returns a
+// distinct buffer. Sharing one buffer across concurrent uploads would
+// corrupt the multipart bodies under concurrent UploadPartCopy.
+func TestNewChunkUploadOption_PerCallIsolation(t *testing.T) {
+	assign := &filer_pb.AssignVolumeResponse{
+		FileId:   "1,foo",
+		Location: &filer_pb.Location{Url: "127.0.0.1:8080"},
+	}
+
+	o1 := newChunkUploadOption(nil, assign, false)
+	o2 := newChunkUploadOption(nil, assign, false)
+	if o1.BytesBuffer == o2.BytesBuffer {
+		t.Fatalf("newChunkUploadOption must hand each caller a distinct buffer; " +
+			"sharing one across concurrent uploads would corrupt the multipart bodies")
+	}
+}

weed/s3api/s3api_object_handlers_copy_stream.go

@@ -0,0 +1,214 @@
+package s3api
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"math"
+	"mime"
+	"mime/multipart"
+	"net/http"
+	"net/textproto"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+)
+
+// canStreamCopyChunk reports whether the chunk-copy fast path can use the
+// io.Pipe streaming variant. The streaming path bypasses both the s3-side
+// download buffer and the s3-side multipart-encode buffer, holding only the
+// io.Pipe internal buffer (~32 KiB) per copy in flight. It does not work
+// when bytes need to be transformed in transit:
+//
+//   - SSE-C / SSE-KMS / SSE-S3: bytes need to be re-encrypted with a
+//     different key on the destination, so they have to land in memory.
+//   - Per-chunk CipherKey (per-chunk encryption from the source): the
+//     source bytes need to be decrypted before being re-uploaded.
+//
+// In both cases the caller falls back to the buffered copySingleChunk path,
+// which already handles those transformations correctly. The streaming
+// path is still safe for IsCompressed chunks — gzip is end-to-end and we
+// just forward the wire bytes (with the destination Content-Encoding
+// header set to whatever the source actually returned).
+func canStreamCopyChunk(chunk *filer_pb.FileChunk) bool {
+	if len(chunk.CipherKey) > 0 {
+		return false
+	}
+	if chunk.GetSseType() != filer_pb.SSEType_NONE {
+		return false
+	}
+	return true
+}
+
+// streamCopyChunkRange copies size bytes starting at offset from srcUrl to
+// the destination volume identified by assignResult, using io.Pipe to avoid
+// buffering the chunk in memory. The destination volume server expects the
+// same multipart wire format that operation.upload_content writes — see
+// upload_content.go for the canonical version; we mirror its part headers
+// here so the volume's needle.parseUpload accepts the request.
+//
+// isFullChunk is the load-bearing distinction:
+//
+//   - false (partial range): a Range header is sent to the source volume,
+//     which serves the requested byte range. For compressed chunks the
+//     volume internally decompresses to satisfy the range, then returns
+//     raw bytes (no Content-Encoding); we forward those raw bytes.
+//   - true (full chunk): we ask the source volume to send compressed
+//     bytes via Accept-Encoding: gzip, bypass Go's auto-decompression on
+//     resp.Body, and forward whatever wire bytes arrived to the
+//     destination. When the chunk is stored gzipped that means we never
+//     decompress on either the source-volume or s3-server side; the
+//     destination receives gzipped bytes and stores them as-is.
+//
+// For the Harbor-style assemble repro (where each UploadPartCopy range
+// matches a source chunk boundary), the full-chunk path eliminates the
+// "bytes.growSlice from util.DecompressData" cost that dominates the
+// volume-side heap profile after the s3-side fixes.
+//
+// The destination's Content-Encoding header is set from the source's
+// response Content-Encoding only — never from a caller-supplied hint.
+// Trusting a hint that disagrees with the wire bytes (e.g. caller says
+// "compressed" but the volume returned raw bytes for a partial range)
+// would label raw bytes as gzip and corrupt reads on the destination.
+func (s3a *S3ApiServer) streamCopyChunkRange(
+	ctx context.Context,
+	srcUrl, srcFileId string,
+	offset, size int64,
+	isFullChunk bool,
+	assignResult *filer_pb.AssignVolumeResponse,
+) error {
+	// Range header takes int64, but mirror the int32 bound from
+	// downloadChunkData so the two paths reject the same inputs.
+	if size > int64(math.MaxInt32) {
+		return fmt.Errorf("chunk size %d exceeds maximum int32 size", size)
+	}
+	// Child context so a terminal error here unblocks both legs
+	// immediately. Without this, a failed POST closes pipeReader
+	// (which only fails the producer's writes), but the source GET's
+	// read loop would keep draining srcResp.Body in the background
+	// until EOF — wasting source-volume bandwidth and CPU on a copy
+	// that's already failed. Cancelling streamCtx tears down both the
+	// source GET and the destination POST when this function returns.
+	streamCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId)
+	dstJwt := security.EncodedJwt(assignResult.Auth)
+	srcJwt := filer.JwtForVolumeServer(srcFileId)
+
+	// Source GET: built directly (not through ReadUrlAsStream) because that
+	// helper auto-decompresses gzipped responses, which would defeat the
+	// whole-chunk pass-through optimization. We want the wire bytes.
+	srcReq, err := http.NewRequestWithContext(streamCtx, http.MethodGet, srcUrl, nil)
+	if err != nil {
+		return fmt.Errorf("create source GET: %w", err)
+	}
+	if srcJwt != "" {
+		srcReq.Header.Set("Authorization", "BEARER "+srcJwt)
+	}
+	if isFullChunk {
+		// Manually setting Accept-Encoding tells Go's http.Transport that
+		// the user wants raw bytes back — Go's automatic decompression is
+		// only applied when it adds the header itself. The server may or
+		// may not actually gzip; the response Content-Encoding header is
+		// authoritative.
+		srcReq.Header.Set("Accept-Encoding", "gzip")
+	} else {
+		srcReq.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+size-1))
+	}
+
+	srcResp, err := util_http.GetGlobalHttpClient().Do(srcReq)
+	if err != nil {
+		return fmt.Errorf("source GET: %w", err)
+	}
+	// CloseResponse drains and closes resp.Body for keepalive.
+	defer util_http.CloseResponse(srcResp)
+
+	if srcResp.StatusCode >= 400 {
+		return fmt.Errorf("source GET %s: %s", srcUrl, srcResp.Status)
+	}
+
+	// What's actually on the wire — trust this over caller hints. For range
+	// fetches the volume always returns raw bytes (no Content-Encoding); for
+	// full-chunk fetches it depends on whether the chunk is stored gzipped.
+	bodyIsGzipped := srcResp.Header.Get("Content-Encoding") == "gzip"
+
+	// io.Pipe gives us a synchronous handoff: the producer goroutine builds
+	// the multipart body, the consumer (HTTP transport) reads it. Writes
+	// block until reads consume them, so the in-flight footprint is the
+	// pipe's hand-off buffer — not the chunk size.
+	pipeReader, pipeWriter := io.Pipe()
+	mw := multipart.NewWriter(pipeWriter)
+	contentType := mw.FormDataContentType()
+
+	go func() {
+		// CloseWithError on the writer end propagates the failure to the
+		// HTTP transport reading from pipeReader, which then aborts the
+		// POST. Plain Close (with err==nil) sends EOF normally.
+		var producerErr error
+		defer func() {
+			pipeWriter.CloseWithError(producerErr)
+		}()
+
+		h := make(textproto.MIMEHeader)
+		h.Set("Content-Disposition", mime.FormatMediaType("form-data",
+			map[string]string{"name": "file", "filename": ""}))
+		h.Set("Idempotency-Key", dstUrl)
+		if bodyIsGzipped {
+			h.Set("Content-Encoding", "gzip")
+		}
+
+		fw, err := mw.CreatePart(h)
+		if err != nil {
+			producerErr = fmt.Errorf("multipart create part: %w", err)
+			return
+		}
+
+		if _, err := io.Copy(fw, srcResp.Body); err != nil {
+			if shouldLogStreamError(err) {
+				glog.V(2).Infof("stream copy %s offset=%d size=%d: %v",
+					srcUrl, offset, size, err)
+			}
+			producerErr = fmt.Errorf("stream copy: %w", err)
+			return
+		}
+
+		if err := mw.Close(); err != nil {
+			producerErr = fmt.Errorf("multipart close: %w", err)
+			return
+		}
+	}()
+
+	req, err := http.NewRequestWithContext(streamCtx, http.MethodPost, dstUrl, pipeReader)
+	if err != nil {
+		pipeReader.CloseWithError(err)
+		return fmt.Errorf("create POST request: %w", err)
+	}
+	req.Header.Set("Content-Type", contentType)
+	if dstJwt != "" {
+		req.Header.Set("Authorization", "BEARER "+string(dstJwt))
+	}
+
+	resp, err := util_http.GetGlobalHttpClient().Do(req)
+	if err != nil {
+		// Closing the reader unblocks the producer if it was still mid-write;
+		// the deferred cancel above also stops any in-flight source read.
+		pipeReader.CloseWithError(err)
+		return fmt.Errorf("POST: %w", err)
+	}
+	defer util_http.CloseResponse(resp)
+
+	if resp.StatusCode >= 400 {
+		return fmt.Errorf("POST %s: %s", dstUrl, resp.Status)
+	}
+	return nil
+}
+
+func shouldLogStreamError(err error) bool {
+	// io.ErrClosedPipe shows up when the HTTP transport closed pipeReader
+	// after the POST failed; the underlying error is reported via the POST
+	// path, no need to log twice.
+	return err != nil && err != io.ErrClosedPipe
+}

weed/util/buffer_pool/sync_pool.go

@@ -5,6 +5,16 @@ import (
 	"sync"
 )
 
+// maxRetainedBufferCap caps the capacity of buffers we hand back to the
+// sync.Pool. Buffers grown past this (e.g. by a 64 MiB chunk upload through
+// volume.PostHandler -> needle.ParseUpload -> bytes.Buffer.ReadFrom) are
+// dropped instead of pooled, so the underlying byte array becomes garbage
+// and is collected. Without this cap the pool effectively hoards every
+// high-water buffer for the process's lifetime — see #6541, where Harbor's
+// concurrent UploadPartCopy filled the pool with 64 MiB buffers and RSS
+// never receded.
+const maxRetainedBufferCap = 4 * 1024 * 1024
+
 var syncPool = sync.Pool{
 	New: func() interface{} {
 		return new(bytes.Buffer)
@@ -16,5 +26,13 @@ func SyncPoolGetBuffer() *bytes.Buffer {
 }
 
 func SyncPoolPutBuffer(buffer *bytes.Buffer) {
+	if buffer == nil {
+		return
+	}
+	if buffer.Cap() > maxRetainedBufferCap {
+		// Drop the buffer; let GC reclaim the oversized backing array.
+		return
+	}
+	buffer.Reset()
 	syncPool.Put(buffer)
 }

weed/util/buffer_pool/sync_pool_test.go

@@ -0,0 +1,92 @@
+package buffer_pool
+
+import (
+	"bytes"
+	"testing"
+)
+
+// TestSyncPoolPutBuffer_DropsOversized is a regression guard for the volume-side
+// retention amplifier in https://github.com/seaweedfs/seaweedfs/issues/6541.
+//
+// volume.PostHandler -> needle.ParseUpload -> bytes.Buffer.ReadFrom grows the
+// buffer to chunk size. If we Put that buffer back as-is, sync.Pool keeps it
+// at the grown capacity for the rest of the process's lifetime. With
+// concurrent UploadPartCopy load that fills the pool with N × chunk-size
+// backing arrays that never shrink — exactly the "RSS never recedes" pattern
+// reported in the issue.
+//
+// We verify a Put + Get round trip can never round-trip a buffer larger than
+// maxRetainedBufferCap, regardless of how big it grew while in use.
+func TestSyncPoolPutBuffer_DropsOversized(t *testing.T) {
+	// Drain the pool so we start from a deterministic point. sync.Pool may
+	// still hold cached entries on other Ps, but for the small/big-cap
+	// distinction below that doesn't matter — we assert an invariant on
+	// every Get, not the identity of a specific buffer.
+	for i := 0; i < 64; i++ {
+		_ = SyncPoolGetBuffer()
+	}
+
+	big := &bytes.Buffer{}
+	big.Grow(maxRetainedBufferCap * 4) // simulate a large upload buffer
+	if got := big.Cap(); got <= maxRetainedBufferCap {
+		t.Fatalf("test setup: big.Cap=%d should exceed threshold %d",
+			got, maxRetainedBufferCap)
+	}
+	SyncPoolPutBuffer(big)
+
+	// Any number of Gets must not return a buffer with cap > threshold.
+	// (If big had been retained, sync.Pool's per-P cache would surface it
+	// on the very next Get on this goroutine.)
+	for i := 0; i < 16; i++ {
+		got := SyncPoolGetBuffer()
+		if cap := got.Cap(); cap > maxRetainedBufferCap {
+			t.Fatalf("Get %d returned buffer with cap=%d, exceeds threshold %d "+
+				"(regression: oversized buffers retained in pool?)",
+				i, cap, maxRetainedBufferCap)
+		}
+	}
+}
+
+// TestSyncPoolPutBuffer_KeepsRightSized verifies the cap is one-sided: we
+// still pool reasonably-sized buffers so the common case (small uploads,
+// header parsing) doesn't pay an alloc per request.
+func TestSyncPoolPutBuffer_KeepsRightSized(t *testing.T) {
+	for i := 0; i < 64; i++ {
+		_ = SyncPoolGetBuffer()
+	}
+
+	small := &bytes.Buffer{}
+	small.Grow(maxRetainedBufferCap / 2)
+	smallCap := small.Cap()
+	SyncPoolPutBuffer(small)
+
+	// We don't assert pointer identity (sync.Pool can hand back any cached
+	// buffer), but the previously-Put buffer should appear among Gets on
+	// the same goroutine in the absence of a GC. If our cap-policy ever
+	// regresses to dropping right-sized buffers, this test starts seeing
+	// only fresh (cap=0) buffers and fails.
+	sawPooled := false
+	for i := 0; i < 8; i++ {
+		got := SyncPoolGetBuffer()
+		if got.Cap() == smallCap {
+			sawPooled = true
+			break
+		}
+	}
+	if !sawPooled {
+		t.Fatalf("right-sized buffer (cap=%d) never came back from the pool "+
+			"(regression: cap policy too aggressive?)", smallCap)
+	}
+}
+
+// TestSyncPoolPutBuffer_NilSafe documents that Put tolerates a nil buffer.
+// The volume server defers Put on a buffer obtained via Get, but defensive
+// callers in other paths may Put(nil); we should not panic.
+func TestSyncPoolPutBuffer_NilSafe(t *testing.T) {
+	defer func() {
+		if r := recover(); r != nil {
+			t.Fatalf("SyncPoolPutBuffer(nil) panicked: %v", r)
+		}
+	}()
+	SyncPoolPutBuffer(nil)
+}