fix: cap pool retention so chunk-copy buffers don't hoard memory (#9422)

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")
+	}
+}

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)
+}