perf(s3): route GET through ChunkReadAt + per-request ReaderCache (#9068)

perf(s3): route GET through ChunkReadAt + shared ReaderCache

The S3 GET path previously used filer.PrepareStreamContentWithPrefetch,
which hands chunk bytes from the volume-server fetch goroutine to the
consumer through an io.Pipe. io.Pipe is a synchronous rendezvous, so
the prefetch=4 window only overlapped HTTP connection setup — the
actual data bytes still flowed one pipe at a time.

Switch to the same path WebDAV uses (server/webdav_server.go): build
a filer.ChunkReadAt backed by a server-wide filer.ReaderCache.
ReaderCache prefetches whole chunks into []byte buffers, so the
prefetch window translates into real in-flight bytes and the consumer
copies them out as memcpys.

The ReaderCache is server-wide (not per-request) for two reasons:

1. ChunkReadAt.Close() destroys the ReaderCache's downloader map.
   With a per-request cache, the defer on the handler would wait for
   background chunk downloads that run on context.Background() — so
   a client disconnect would block handler cleanup on downloads that
   the client no longer wants, tying up goroutines and memory.

2. Concurrent requests for the same object can share in-flight
   downloads through the shared downloader map.

No persistent ChunkCache is added in this commit — the ReaderCache is
constructed with a nil *chunk_cache.TieredChunkCache (all its methods
are nil-receiver safe). A follow-up PR wires in an in-memory chunk
cache for cross-request warm hits.

JWT for volume-server requests is generated internally by
util_http.RetriedFetchChunkData from jwtSigningReadKey, so the new
path remains compatible with JWT-protected clusters — this is the
same mechanism the WebDAV and mount read paths have been using.

Measured on weed mini + 1 GiB random object over loopback, cold
cache, single-stream curl on a presigned URL:

    before (io.Pipe):        2100-2200 MB/s
    after  (ChunkReadAt):    2900-3800 MB/s

weed/s3api/s3api_object_handlers.go

@@ -1026,14 +1026,15 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		}
 	}
 
-	// CRITICAL: Resolve chunks and prepare stream BEFORE WriteHeader
-	// This ensures we can write proper error responses if these operations fail
+	// CRITICAL: Resolve chunks and prepare reader BEFORE WriteHeader so failures
+	// can still return a proper S3 error response.
 	ctx := r.Context()
 	lookupFileIdFn := s3a.createLookupFileIdFunction()
 
-	// Resolve chunk manifests with the requested range
+	// Resolve chunk manifests into visible intervals for the requested range.
+	// NonOverlappingVisibleIntervals internally calls ResolveChunkManifest.
 	tChunkResolve := time.Now()
-	resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, chunks, offset, offset+size)
+	visibleIntervals, err := filer.NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, offset, offset+size)
 	chunkResolveTime = time.Since(tChunkResolve)
 	if err != nil {
 		if isCanceledStreamingError(err) {
@@ -1050,34 +1051,23 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		return newStreamErrorWithResponse(fmt.Errorf("failed to resolve chunks: %v", err))
 	}
 
-	// Prepare streaming function with simple master client wrapper
+	// Build a ChunkReadAt backed by the server-wide ReaderCache. This mirrors
+	// the WebDAV read path (server/webdav_server.go) and outperforms the
+	// io.Pipe-based streamChunksPrefetched path: ReaderCache prefetches whole
+	// chunks into memory buffers that the consumer can memcpy out of, so
+	// prefetchCount translates into actual in-flight bytes rather than just
+	// parallel TCP handshakes.
+	//
+	// We do NOT call reader.Close() here — ChunkReadAt.Close() would destroy
+	// the ReaderCache's downloaders map, which is shared across concurrent
+	// requests. Eviction is handled by the ReaderCache's own downloader
+	// limit. JWT for volume-server requests is generated internally by
+	// util_http.RetriedFetchChunkData from jwtSigningReadKey, matching the
+	// WebDAV and mount read paths.
 	tStreamPrep := time.Now()
-	// Use filerClient directly (not wrapped) so it can support cache invalidation
-	streamFn, err := filer.PrepareStreamContentWithPrefetch(
-		ctx,
-		s3a.filerClient,
-		filer.JwtForVolumeServer, // Use filer's JWT function (loads config once, generates JWT locally)
-		resolvedChunks,
-		offset,
-		size,
-		0, // no throttling
-		4, // prefetch 4 chunks ahead for overlapped fetching
-	)
+	chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, offset, size)
+	reader := filer.NewChunkReaderAtFromClient(ctx, s3a.readerCache, chunkViews, totalSize, filer.DefaultPrefetchCount)
 	streamPrepTime = time.Since(tStreamPrep)
-	if err != nil {
-		if isCanceledStreamingError(err) {
-			glog.V(3).Infof("streamFromVolumeServers: request canceled while preparing stream: %v", err)
-			return err
-		}
-		if errors.Is(err, context.DeadlineExceeded) {
-			glog.Warningf("streamFromVolumeServers: request deadline exceeded while preparing stream: %v", err)
-		} else {
-			glog.Errorf("streamFromVolumeServers: failed to prepare stream: %v", err)
-		}
-		// Write S3-compliant XML error response
-		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
-		return newStreamErrorWithResponse(fmt.Errorf("failed to prepare stream: %v", err))
-	}
 
 	// All validation and preparation successful - NOW set headers and write status
 	tHeaderSet := time.Now()
@@ -1102,11 +1092,22 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 	// Track time to first byte metric
 	TimeToFirstByte(r.Method, t0, r)
 
-	// Stream directly to response with counting wrapper
+	// Stream directly to response with counting wrapper.
+	// ChunkReadAt's ReadAt is backed by in-memory prefetched chunk buffers, so
+	// io.CopyBuffer drains them as fast memcpys.
 	tStreamExec := time.Now()
-	glog.V(4).Infof("streamFromVolumeServers: starting streamFn, offset=%d, size=%d", offset, size)
+	glog.V(4).Infof("streamFromVolumeServers: starting chunk reader, offset=%d, size=%d", offset, size)
 	cw := &countingWriter{w: w}
-	err = streamFn(cw)
+	// Cap the copy buffer to the response size so small-object GETs (common
+	// for thumbnails, config files, etc.) don't allocate a 256 KiB scratch
+	// buffer per request.
+	const maxCopyBuf = 256 * 1024
+	copyBufSize := int64(maxCopyBuf)
+	if size > 0 && size < copyBufSize {
+		copyBufSize = size
+	}
+	copyBuf := make([]byte, copyBufSize)
+	_, err = io.CopyBuffer(cw, io.NewSectionReader(reader, offset, size), copyBuf)
 	streamExecTime = time.Since(tStreamExec)
 	// Track traffic even on partial writes for accurate egress accounting
 	if cw.written > 0 {

weed/s3api/s3api_server.go

@@ -31,6 +31,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
 	"github.com/seaweedfs/seaweedfs/weed/util/grace"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
@@ -84,6 +85,12 @@ type S3ApiServer struct {
 	stsHandlers           *STSHandlers    // STS HTTP handlers for AssumeRoleWithWebIdentity
 	cipher                bool            // encrypt data on volume servers
 	newObjectWriteLock    func(bucket, object string) objectWriteLock
+	// Shared ReaderCache used by the S3 GET streaming path. It lives for the
+	// lifetime of the server so that concurrent and repeat reads share a
+	// single in-flight download per chunk, and so that no per-request
+	// teardown waits on context.Background() fetches. The chunkCache field
+	// is nil in this commit; a follow-up wires in an in-memory chunk cache.
+	readerCache *filer.ReaderCache
 }
 
 type objectWriteLock interface {
@@ -178,6 +185,22 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 		}
 	}()
 
+	// Shared ReaderCache for the S3 GET streaming path. The chunk cache is
+	// nil for now — all TieredChunkCache receiver methods are nil-safe. A
+	// follow-up adds an in-memory chunk cache on top. Keeping this shared
+	// (rather than per-request) avoids the per-request Close(), which would
+	// otherwise wait for background chunk downloads that run on
+	// context.Background() even after the client disconnects.
+	//
+	// Downloader slots: each slot holds one in-flight / recently-completed
+	// chunk buffer (~4 MiB by default), so this caps both peak memory for
+	// in-flight chunks (s3ReaderCacheDownloaderLimit × chunkSize) and the
+	// global fetch concurrency across all S3 GET requests. WebDAV uses 32
+	// because it typically has a handful of clients; S3 serves many
+	// concurrent readers, so we pick a more generous default here.
+	const s3ReaderCacheDownloaderLimit = 256
+	readerCache := filer.NewReaderCache(s3ReaderCacheDownloaderLimit, (*chunk_cache.TieredChunkCache)(nil), filerClient.GetLookupFileIdFunction())
+
 	s3ApiServer = &S3ApiServer{
 		option:                option,
 		iam:                   iam,
@@ -190,6 +213,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 		policyEngine:          policyEngine,                           // Initialize bucket policy engine
 		inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
 		cipher:                option.Cipher,
+		readerCache:           readerCache,
 	}
 
 	if len(option.Filers) > 0 {