feat(s3api): stream chunk copy via io.Pipe to cut peak working set (#9424)

* 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.

* feat(s3api): stream chunk copy via io.Pipe to cut peak working set

Final piece for #6541. The buffered chunk-copy path holds two
chunk-sized buffers per copy in flight (download buffer + multipart-
encoded upload buffer). Under concurrent UploadPartCopy that put a
floor on RSS at concurrency × 2 × chunk_size — about 768 MiB for the
6-way / 64 MiB Harbor-style assemble repro, even after the previous
pool/retention fixes.

Replace the buffered path with an io.Pipe between the source GET and
the destination POST: ReadUrlAsStream pumps data into the pipe via a
multipart.Writer, the http.Client reads from the pipe end and POSTs
the body. In-flight per copy is now ~32 KiB (pipe hand-off + http
buffers), regardless of chunk size.

The streaming path is gated by canStreamCopyChunk: only used when no
in-transit transformation is needed (no per-chunk CipherKey, no SSE).
SSE-C / SSE-KMS / SSE-S3 paths still go through the buffered path,
which already handles re-encryption correctly.

Benchmarks (Apple M4, httptest source/dest, B/op = bytes per copy):

  Buffered  1 MiB:   6.0 MB B/op,  443 MB/s
  Streamed  1 MiB:   374 KB B/op,  727 MB/s
  Buffered  8 MiB:    56 MB B/op,  559 MB/s
  Streamed  8 MiB:   379 KB B/op, 1138 MB/s
  Buffered 64 MiB:   455 MB B/op,  718 MB/s
  Streamed 64 MiB:   304 KB B/op, 1387 MB/s

End-to-end repro (512 MiB src, 6 parallel UploadPartCopy):
  pre-#9420 RSS round 2: 3134 MiB
  + #9420/#9421/#9422  : 2236 MiB
  + this PR            : 1521 MiB
  heap inuse_space     :  350 MiB (was 1422 / 1187 MiB)
  HeapSys (MemStats)   : 1.74 GiB (was 2.49 GiB)

* review: surface shouldRetry, add int32 guard, drop redundant drains

Address review on PR 9424:

* coderabbit (HIGH, line 122): ReadUrlAsStream can set shouldRetry=true
  with readErr=nil. Before this fix, that fell through to mw.Close()
  and the destination POST succeeded against a possibly-truncated
  multipart body. Mirror downloadChunkData's explicit check and
  surface shouldRetry as a producer error so the dst POST aborts.
* gemini (line 98): chunk size is int64 but ReadUrlAsStream takes int.
  Reject sizes above MaxInt32 up front so the int(size) cast can't
  truncate negative on 32-bit platforms — same guard downloadChunkData
  uses.
* gemini (line 151): util_http.CloseResponse already drains the body
  (io.Copy(io.Discard, ...) inside the helper) before closing, so the
  manual io.Copy drains we added are redundant. Drop them.

* review: cancel source GET when destination POST fails

Address coderabbit review (line 165 / second pass on PR 9424): when
the POST leg fails or returns an error status, closing pipeReader
only fails the producer's *writes*. ReadUrlAsStream's own read loop
runs under the parent ctx, so it keeps draining the source body in
the background until EOF — wasting source-volume bandwidth and CPU
on a copy that's already failed.

Wrap streamCopyChunkRange in a child context cancelled on return.
ReadUrlAsStream checks ctx.Done() per 256 KiB tick, so the in-flight
read aborts on the next iteration once the function returns. The POST
also moves to streamCtx so the in-flight request can be cancelled the
same way if the producer fails first.

Defer-cancel runs after both legs return, so the success path still
sends EOF cleanly through pipeWriter.Close before cancellation.

weed/s3api/s3api_object_handlers_copy.go

@@ -1053,7 +1053,18 @@ 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.
+	if canStreamCopyChunk(chunk) {
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), assignResult, chunk.IsCompressed); 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 +1099,17 @@ 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.
+	if canStreamCopyChunk(originalChunk) {
+		if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), assignResult, originalChunk.IsCompressed); 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 {

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),
+					env.assign, false); 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_stream.go

@@ -0,0 +1,177 @@
+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 compressed bytes with the Content-Encoding header.
+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.
+func (s3a *S3ApiServer) streamCopyChunkRange(
+	ctx context.Context,
+	srcUrl, srcFileId string,
+	offset, size int64,
+	assignResult *filer_pb.AssignVolumeResponse,
+	isCompressed bool,
+) error {
+	// ReadUrlAsStream takes int for size; reject anything that would
+	// truncate negative on 32-bit. Mirrors the guard in downloadChunkData.
+	if size > int64(math.MaxInt32) {
+		return fmt.Errorf("chunk size %d exceeds maximum int32 size", size)
+	}
+	// Child context so a terminal error here unblocks the producer
+	// goroutine immediately. Without this, a failed POST closes
+	// pipeReader (which only fails the producer's writes), but
+	// ReadUrlAsStream can keep draining the source body in its read
+	// loop until EOF before noticing — wasting source-volume bandwidth
+	// and CPU. Cancelling streamCtx makes ReadUrlAsStream's per-tick
+	// ctx.Done() check return on the next iteration.
+	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)
+
+	// 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 internal 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 isCompressed {
+			h.Set("Content-Encoding", "gzip")
+		}
+
+		fw, err := mw.CreatePart(h)
+		if err != nil {
+			producerErr = fmt.Errorf("multipart create part: %w", err)
+			return
+		}
+
+		// ReadUrlAsStream reads in 256 KiB ticks and hands each tick to
+		// the callback. Forwarding directly into fw means each tick is
+		// flushed through the multipart writer into the pipe, where the
+		// HTTP transport picks it up — no per-chunk buffering on either
+		// side of the pipe.
+		var writeErr error
+		shouldRetry, readErr := util_http.ReadUrlAsStream(streamCtx, srcUrl, srcJwt, nil, false, false, offset, int(size), func(data []byte) {
+			if writeErr != nil {
+				return
+			}
+			if _, err := fw.Write(data); err != nil {
+				writeErr = err
+			}
+		})
+		if writeErr != nil {
+			producerErr = fmt.Errorf("stream write: %w", writeErr)
+			return
+		}
+		if readErr != nil {
+			if shouldRetry {
+				glog.V(2).Infof("stream copy %s offset=%d size=%d: retryable error: %v",
+					srcUrl, offset, size, readErr)
+			}
+			producerErr = fmt.Errorf("stream read: %w", readErr)
+			return
+		}
+		// shouldRetry can be set without an error (e.g. ReadUrlAsStream
+		// surfacing a partial-read condition that the buffered path
+		// re-fetches). Treat it as a failed copy here too — otherwise we
+		// would close the multipart cleanly and let the destination POST
+		// succeed against a possibly-truncated body. Mirrors the explicit
+		// check downloadChunkData makes after ReadUrlAsStream returns.
+		if shouldRetry {
+			producerErr = fmt.Errorf("stream read %s offset=%d size=%d: retry needed", srcUrl, offset, size)
+			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 {
+		// Drain the pipe so the producer goroutine doesn't leak waiting
+		// on a never-read writer.
+		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)
+	}
+	// CloseResponse drains and closes resp.Body for us; no manual io.Copy
+	// drain needed for keepalive.
+	defer util_http.CloseResponse(resp)
+
+	if resp.StatusCode >= 400 {
+		return fmt.Errorf("POST %s: %s", dstUrl, resp.Status)
+	}
+	return nil
+}