s3: improve TTFB for large remote objects (#10010)

* s3: add streaming reader interface for remote storage

Add RemoteStorageStreamReader optional interface to support efficient
streaming of large remote objects without buffering entire file in memory.
This enables future stream-through caching where data can be served to
clients while simultaneously writing to volume servers.

Implement ReadFileAsStream() for S3, GCS, and Azure backends using their
native streaming APIs. This provides the foundation for improving TTFB
on large remote file access by serving data directly from remote storage
while background cache operation populates local chunks.

The streaming interface allows remote storage backends to return io.ReadCloser,
enabling efficient memory usage for multi-GB objects compared to the
current ReadFile() approach which buffers entire ranges in memory.

* s3: adaptive timeout for remote object caching to improve TTFB

Use size-aware cache polling timeout to balance cache-hit rate against
time-to-first-byte:

- Small files (<50MB): 10s timeout - more likely to complete caching
  before timeout, improving subsequent request performance
- Medium files (50-500MB): 5s timeout - default balance
- Large files (>500MB): 2s timeout - fail-fast to improve initial TTFB
  for very large downloads

This reduces waiting time for large remote files while maintaining
high cache-hit rate for smaller files that cache quickly.

* s3: address code review feedback for stream-through cache

- Move startBackgroundRemoteCache call after policy recheck to avoid
  cache side effects for denied requests (authorization first)
- Make startBackgroundRemoteCache version-aware by accepting versionId
  parameter and using buildVersionedRemoteObjectPath
- Add timeout (5 minutes) to background cache context to prevent
  goroutine pile-up if RPC stalls under load
- Update cacheRemoteObjectForStreamingWithShortTimeout to return both
  entry and error, allowing callers to distinguish transient errors
  (timeout/cancellation) from permanent errors (not found, denied)
- Update streamFromVolumeServers to handle permanent cache errors with
  appropriate HTTP status codes (404 for not found, 503 for transient)

weed/remote_storage/azure/azure_storage_client.go

@@ -344,6 +344,22 @@ func (az *azureRemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.Remot
 	return data, nil
 }
 
+func (az *azureRemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) {
+	key := loc.Path[1:]
+	blobClient := az.client.ServiceClient().NewContainerClient(loc.Bucket).NewBlockBlobClient(key)
+
+	downloadResponse, err := blobClient.DownloadStream(ctx, &blob.DownloadStreamOptions{
+		Range: blob.HTTPRange{
+			Offset: offset,
+			Count:  size,
+		},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to open stream for %s%s: %v", loc.Bucket, loc.Path, err)
+	}
+	return downloadResponse.Body, nil
+}
+
 func (az *azureRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) {
 	return nil
 }

weed/remote_storage/gcs/gcs_storage_client.go

@@ -212,6 +212,11 @@ func (gcs *gcsRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation
 	return
 }
 
+func (gcs *gcsRemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) {
+	key := loc.Path[1:]
+	return gcs.client.Bucket(loc.Bucket).Object(key).NewRangeReader(ctx, offset, size)
+}
+
 func (gcs *gcsRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) {
 	return nil
 }

weed/remote_storage/remote_storage.go

@@ -95,6 +95,12 @@ type RemoteStorageConcurrentReader interface {
 	ReadFileWithConcurrency(loc *remote_pb.RemoteStorageLocation, offset int64, size int64, concurrency int) (data []byte, err error)
 }
 
+// RemoteStorageStreamReader is an optional interface for remote storage clients
+// that support streaming reads with io.Reader for efficient memory usage.
+type RemoteStorageStreamReader interface {
+	ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error)
+}
+
 type RemoteStorageClientMaker interface {
 	Make(remoteConf *remote_pb.RemoteConf) (RemoteStorageClient, error)
 	HasBucket() bool

weed/remote_storage/s3/s3_storage_client.go

@@ -261,6 +261,21 @@ func (s *s3RemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.RemoteSto
 	return writerAt.Bytes(), nil
 }
 
+func (s *s3RemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) {
+	output, err := s.conn.GetObjectWithContext(ctx, &s3.GetObjectInput{
+		Bucket: aws.String(loc.Bucket),
+		Key:    aws.String(loc.Path[1:]),
+		Range:  aws.String(fmt.Sprintf("bytes=%d-%d", offset, offset+size-1)),
+	})
+	if err != nil {
+		if aerr, ok := err.(awserr.Error); ok && aerr.Code() == s3.ErrCodeNoSuchKey {
+			return nil, remote_storage.ErrRemoteObjectNotFound
+		}
+		return nil, fmt.Errorf("failed to open stream for %s%s: %v", loc.Bucket, loc.Path, err)
+	}
+	return output.Body, nil
+}
+
 func (s *s3RemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) {
 	return nil
 }

weed/s3api/s3api_object_handlers.go

@@ -776,19 +776,24 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	// Handle remote storage objects: cache to local cluster if object is remote-only
-	// This uses singleflight to deduplicate concurrent caching requests for the same object
-	// On cache error, gracefully falls back to streaming from remote
-	if objectEntryForSSE.IsInRemoteOnly() {
-		objectEntryForSSE = s3a.cacheRemoteObjectWithDedup(r.Context(), bucket, object, objectEntryForSSE)
-	}
-
 	// Re-check bucket policy with object entry for tag-based conditions (e.g., s3:ExistingObjectTag)
 	if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, string(s3_constants.ACTION_READ), objectEntryForSSE.Extended, "GetObjectHandler"); errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
+	// Handle remote storage objects: initiate background caching without blocking
+	// This implements stream-through caching: serve first byte immediately while
+	// caching happens in the background, improving TTFB for large files
+	if objectEntryForSSE.IsInRemoteOnly() {
+		// Start background cache without waiting (non-blocking)
+		// Only after authorization passes to avoid cache side effects for denied requests
+		versionId := r.URL.Query().Get("versionId")
+		cacheVersionId := resolvedSourceVersionId(versionId, objectEntryForSSE)
+		s3a.startBackgroundRemoteCache(bucket, object, cacheVersionId, objectEntryForSSE)
+		// Continue with streaming immediately - will serve from remote or cached chunks
+	}
+
 	// Check if PartNumber query parameter is present (for multipart GET requests)
 	partNumberStr := r.URL.Query().Get("partNumber")
 	if partNumberStr == "" {
@@ -978,28 +983,31 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		len(chunks), totalSize, isRangeRequest, offset, size)
 
 	if len(chunks) == 0 {
-		// Check if this is a remote-only entry that needs caching
-		// This handles the case where initial caching attempt timed out or failed
+		// Check if this is a remote-only entry
 		if entry.IsInRemoteOnly() {
-			glog.V(1).Infof("streamFromVolumeServers: entry is remote-only, attempting to cache before streaming")
-			// Latest-version reads carry an empty query versionId even when the
-			// entry lives at .versions/v_<id>; resolve from the entry itself.
+			glog.V(1).Infof("streamFromVolumeServers: entry is remote-only, attempting stream-through cache")
 			cacheVersionId := resolvedSourceVersionId(versionId, entry)
-			cachedEntry := s3a.cacheRemoteObjectForStreaming(r, entry, bucket, object, cacheVersionId)
-			if cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 {
+			cachedEntry, cacheErr := s3a.cacheRemoteObjectForStreamingWithShortTimeout(r, entry, bucket, object, cacheVersionId)
+			if cacheErr == nil && cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 {
+				// Cache completed, use cached chunks
 				chunks = cachedEntry.GetChunks()
 				entry = cachedEntry
 				glog.V(1).Infof("streamFromVolumeServers: successfully cached remote object, got %d chunks", len(chunks))
-			} else {
-				// Client disconnected: report cancellation, not 503.
-				if ctxErr := r.Context().Err(); ctxErr != nil {
-					return ctxErr
+			} else if cacheErr != nil && !errors.Is(cacheErr, context.DeadlineExceeded) && !errors.Is(cacheErr, context.Canceled) && status.Code(cacheErr) != codes.DeadlineExceeded && status.Code(cacheErr) != codes.Canceled {
+				// Permanent error (e.g. not found, permission denied) - return final status
+				glog.Errorf("streamFromVolumeServers: permanent cache error for %s/%s: %v", bucket, object, cacheErr)
+				if status.Code(cacheErr) == codes.NotFound {
+					s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+				} else {
+					s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 				}
-				// Cache still filling: 503 + Retry-After so SDKs back off and retry.
+				return newStreamErrorWithResponse(cacheErr)
+			} else {
+				// Cache not ready yet; return 503 with Retry-After for client backoff
 				glog.V(1).Infof("streamFromVolumeServers: remote object %s/%s not cached yet, returning 503 for retry", bucket, object)
-				w.Header().Set("Retry-After", "5")
+				w.Header().Set("Retry-After", "2")
 				s3err.WriteErrorResponse(w, r, s3err.ErrServiceUnavailable)
-				return newStreamErrorWithResponse(fmt.Errorf("remote object not cached yet"))
+				return newStreamErrorWithResponse(fmt.Errorf("remote object not cached yet, will be available on retry"))
 			}
 		} else if totalSize > 0 && len(entry.Content) == 0 {
 			// Not a remote entry but has size without content - this is a data integrity issue
@@ -3114,6 +3122,51 @@ func cachedEntryHasLocalData(entry *filer_pb.Entry) bool {
 // Atomic so tests can shorten it without racing the read path.
 var remoteCacheStreamingTimeoutNS = int64(20 * time.Second)
 
+// cacheRemoteObjectForStreamingWithShortTimeout polls for cache completion with an adaptive timeout.
+// Timeout is based on file size: small files wait longer to maximize cache hits, large files
+// fail-fast to improve TTFB. Returns the cached entry and error to allow callers to distinguish
+// between transient errors (timeout) and permanent errors (not found, permission denied).
+// The filer continues caching on detached context, so retry finds cached chunks.
+func (s3a *S3ApiServer) cacheRemoteObjectForStreamingWithShortTimeout(r *http.Request, entry *filer_pb.Entry, bucket, object, versionId string) (*filer_pb.Entry, error) {
+	// Adaptive timeout: smaller files can afford to wait longer since cache completes faster
+	pollTimeout := 5 * time.Second
+	if entry.RemoteEntry != nil && entry.RemoteEntry.RemoteSize > 0 {
+		// For very large files (>500MB), use shorter timeout to improve TTFB
+		// For smaller files, allow longer to increase cache hit rate
+		remoteSize := entry.RemoteEntry.RemoteSize
+		if remoteSize > 500*1024*1024 {
+			pollTimeout = 2 * time.Second
+		} else if remoteSize < 50*1024*1024 {
+			pollTimeout = 10 * time.Second
+		}
+	}
+
+	cacheCtx, cancel := context.WithTimeout(r.Context(), pollTimeout)
+	defer cancel()
+
+	dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId)
+
+	glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: polling cache status for %s/%s (timeout=%v)", dir, name, pollTimeout)
+
+	cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name)
+	if err != nil {
+		// Distinguish transient errors (timeout/cancellation) from permanent errors
+		if cacheCtx.Err() != nil {
+			glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: %s/%s not cached within %v", dir, name, pollTimeout)
+			return nil, cacheCtx.Err()
+		}
+		glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: cache error for %s/%s: %v", dir, name, err)
+		return nil, err
+	}
+
+	if cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 {
+		glog.V(1).Infof("cacheRemoteObjectForStreamingWithShortTimeout: successfully cached %s/%s (%d chunks)", dir, name, len(cachedEntry.GetChunks()))
+		return cachedEntry, nil
+	}
+
+	return nil, nil
+}
+
 // cacheRemoteObjectForStreaming caches a remote-only object to the local cluster for streaming.
 // Bounded so a slow large-file download returns to the caller (which emits 503 +
 // Retry-After) before the client gives up; the filer keeps caching on a detached
@@ -3179,6 +3232,34 @@ func (s3a *S3ApiServer) cacheRemoteObjectForCopy(ctx context.Context, bucket, ob
 	return nil
 }
 
+// startBackgroundRemoteCache initiates caching without blocking the request.
+// This enables fast TTFB: the client's request returns immediately (via 503 if not
+// cached), while a background task fills the cache. Subsequent requests will find
+// cached chunks via singleflight deduplication in the filer's CacheRemoteObjectToLocalCluster.
+// Uses detached context with reasonable timeout to prevent goroutine pile-up.
+func (s3a *S3ApiServer) startBackgroundRemoteCache(bucket, object, versionId string, entry *filer_pb.Entry) {
+	if !entry.IsInRemoteOnly() {
+		return
+	}
+
+	dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId)
+
+	// Start background cache without blocking. The filer's CacheRemoteObjectToLocalCluster
+	// uses singleflight internally, so concurrent requests will all benefit from the same
+	// cache operation.
+	go func() {
+		// Use timeout to bound goroutine and prevent pile-up if RPC stalls under load
+		bgCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		defer cancel()
+		_, err := s3a.doCacheRemoteObject(bgCtx, dir, name)
+		if err != nil {
+			glog.V(2).Infof("startBackgroundRemoteCache: cache failed for %s/%s: %v", bucket, object, err)
+		} else {
+			glog.V(2).Infof("startBackgroundRemoteCache: cached %s/%s in background", bucket, object)
+		}
+	}()
+}
+
 // resolvedSourceVersionId falls back to the version recorded on the entry
 // when the request didn't carry one — necessary for latest-version reads
 // in versioning-enabled buckets, where the entry lives at .versions/v_<id>.