fix: cap pool retention so chunk-copy buffers don't hoard memory

Two pool-retention sites kept the runaway-RSS pattern in #6541 visible
even after #9420 and #9421:

* weed/util/buffer_pool: SyncPoolPutBuffer dropped a buffer back into
  sync.Pool regardless of how big it had grown. After a 64 MiB chunk
  upload through volume.PostHandler -> needle.ParseUpload, the pool
  hoarded a 64 MiB byte array per cached entry for the rest of the
  process's lifetime. Cap retention at 4 MiB; oversized buffers are
  dropped so GC can reclaim the backing array.

* weed/s3api/...copy.go: uploadChunkData left UploadOption.BytesBuffer
  unset, so operation.upload_content fell back to the package-global
  valyala/bytebufferpool. That pool also retains high-water buffers
  forever, and concurrent UploadPartCopy filled it with one chunk-sized
  buffer per concurrent upload. Provide a fresh per-call bytes.Buffer
  pre-sized to chunk + multipart framing; it's GC'd as soon as the
  upload returns.

Tests:
- weed/util/buffer_pool/sync_pool_test.go: pin the cap (oversized
  buffers don't round-trip), the inverse (right-sized buffers do), and
  nil-safety.
- weed/s3api/...copy_chunk_upload_test.go: extract newChunkUploadOption
  and pin that BytesBuffer is always non-nil and pre-sized, and that
  each call gets a distinct buffer.

weed/s3api/s3api_object_handlers_copy.go

@@ -1379,16 +1379,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 +1392,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_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")
+	}
+}