s3: bound streaming remote-cache wait so large cold GetObject returns 503, not a hang (#9988)

* s3: bound streaming remote-cache wait so large cold GetObject returns 503, not a hang

A remote-only object whose download outlasts the 30s dedup attempt fell through
to cacheRemoteObjectForStreaming, which waited on the raw request context with no
bound. For multi-GB blobs the client read timeout fires first, so the request was
canceled and surfaced as InternalError rather than the 503 + Retry-After the
handler already emits. The filer keeps caching on a detached context, so the
retry would have streamed from the cached chunks.

Bound the wait under common client read timeouts: the 503 fires while the client
is still connected, and a retry picks up the finished cache.

* s3: make the streaming remote-cache timeout atomic

The bound is a package-level var so tests can shorten it. Hold it as an
atomic int64 so a test that stores a short value can never race the request
path that loads it.

* s3: assert the streaming-cache timeout test waits on the bound

The upper-bound-only check would also pass if the RPC returned immediately on a
setup error. Add a lower bound so the test fails unless the bounded wait fired.

weed/s3api/s3api_object_handlers.go

@@ -16,6 +16,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -3105,18 +3106,37 @@ func cachedEntryHasLocalData(entry *filer_pb.Entry) bool {
 	return entry != nil && (len(entry.GetChunks()) > 0 || len(entry.Content) > 0)
 }
 
+// remoteCacheStreamingTimeoutNS bounds (nanoseconds) how long the streaming read
+// path waits for the filer to finish caching a remote-only object. Without a
+// bound a multi-GB download outlasts the client's read timeout, so the request
+// is canceled before the caller can emit 503 + Retry-After. Kept under common S3
+// client read timeouts (~60s) given the 30s dedup attempt that precedes it.
+// Atomic so tests can shorten it without racing the read path.
+var remoteCacheStreamingTimeoutNS = int64(20 * time.Second)
+
 // cacheRemoteObjectForStreaming caches a remote-only object to the local cluster for streaming.
-// Uses the request context (no artificial timeout) so the caching can complete.
-// Returns the cached entry only when chunks are present; the streaming caller
-// is not wired to read inline Content from a cache result here.
+// 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
+// context, so a retry streams from the cached chunks. Returns the cached entry
+// only when chunks are present; the caller cannot read inline Content here.
 func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *filer_pb.Entry, bucket, object, versionId string) *filer_pb.Entry {
+	timeout := time.Duration(atomic.LoadInt64(&remoteCacheStreamingTimeoutNS))
+	cacheCtx, cancel := context.WithTimeout(r.Context(), timeout)
+	defer cancel()
+
 	dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId)
 
 	glog.V(1).Infof("cacheRemoteObjectForStreaming: caching %s/%s (remote size: %d, versionId: %s)", dir, name, entry.RemoteEntry.RemoteSize, versionId)
 
-	cachedEntry, err := s3a.doCacheRemoteObject(r.Context(), dir, name)
+	cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name)
 	if err != nil {
-		glog.Errorf("cacheRemoteObjectForStreaming: failed to cache %s/%s: %v", dir, name, err)
+		// A bounded-wait timeout (or client disconnect) is not a cache failure:
+		// the filer keeps downloading and the caller maps a nil return to 503.
+		if cacheCtx.Err() != nil {
+			glog.V(1).Infof("cacheRemoteObjectForStreaming: %s/%s not ready within %v (will retry)", dir, name, timeout)
+		} else {
+			glog.Errorf("cacheRemoteObjectForStreaming: failed to cache %s/%s: %v", dir, name, err)
+		}
 		return nil
 	}
 

weed/s3api/s3api_remote_storage_test.go

@@ -1,13 +1,24 @@
 package s3api
 
 import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
 	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 // TestIsInRemoteOnly tests the IsInRemoteOnly method on filer_pb.Entry
@@ -484,3 +495,93 @@ func TestCopyObjectRemoteOnlySourceDetection(t *testing.T) {
 		})
 	}
 }
+
+// fakeCacheFiler is a minimal SeaweedFiler gRPC server whose
+// CacheRemoteObjectToLocalCluster behavior is driven by a callback, so tests can
+// model a slow (still-caching) filer or one that returns cached chunks.
+type fakeCacheFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	cache func(context.Context, *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error)
+}
+
+func (f *fakeCacheFiler) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
+	return f.cache(ctx, req)
+}
+
+// startFakeCacheFiler serves impl on a random localhost port and returns the
+// S3-style filer address whose ToGrpcAddress resolves back to that port.
+func startFakeCacheFiler(t *testing.T, impl *fakeCacheFiler) pb.ServerAddress {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	srv := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(srv, impl)
+	go srv.Serve(lis)
+	t.Cleanup(srv.Stop)
+	port := lis.Addr().(*net.TCPAddr).Port
+	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", port))
+}
+
+func newRemoteCacheTestServer(filerAddr pb.ServerAddress) *S3ApiServer {
+	return &S3ApiServer{
+		option: &S3ApiServerOption{
+			Filers:         []pb.ServerAddress{filerAddr},
+			BucketsPath:    "/buckets",
+			GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		},
+	}
+}
+
+func remoteOnlyEntry(name string, size int64) *filer_pb.Entry {
+	return &filer_pb.Entry{Name: name, RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: size}}
+}
+
+// TestCacheRemoteObjectForStreamingTimeout pins the fix for large-file cold
+// GetObject: when the filer is still caching, the streaming path returns nil
+// within its bound (so the handler emits 503 + Retry-After) instead of blocking
+// on the raw request context until the client gives up.
+func TestCacheRemoteObjectForStreamingTimeout(t *testing.T) {
+	filerAddr := startFakeCacheFiler(t, &fakeCacheFiler{
+		cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
+			<-ctx.Done() // never finishes within the bound
+			return nil, ctx.Err()
+		},
+	})
+	s3a := newRemoteCacheTestServer(filerAddr)
+
+	const shortTimeout = 200 * time.Millisecond
+	defer func(prev int64) { atomic.StoreInt64(&remoteCacheStreamingTimeoutNS, prev) }(atomic.LoadInt64(&remoteCacheStreamingTimeoutNS))
+	atomic.StoreInt64(&remoteCacheStreamingTimeoutNS, int64(shortTimeout))
+
+	r := httptest.NewRequest(http.MethodGet, "/mybucket/large.bin", nil)
+	start := time.Now()
+	got := s3a.cacheRemoteObjectForStreaming(r, remoteOnlyEntry("large.bin", 1<<30), "mybucket", "large.bin", "")
+	elapsed := time.Since(start)
+
+	assert.Nil(t, got, "uncached object must return nil so the caller maps to 503")
+	assert.GreaterOrEqual(t, elapsed, shortTimeout/2, "must wait on the bounded timeout, not return early on a setup error")
+	assert.Less(t, elapsed, 5*time.Second, "must not block on the full download")
+	assert.NoError(t, r.Context().Err(), "request context stays alive so the caller chooses 503, not cancellation")
+}
+
+// TestCacheRemoteObjectForStreamingCached confirms that once the filer reports
+// local chunks, the streaming path returns the cached entry to stream from.
+func TestCacheRemoteObjectForStreamingCached(t *testing.T) {
+	filerAddr := startFakeCacheFiler(t, &fakeCacheFiler{
+		cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
+			return &filer_pb.CacheRemoteObjectToLocalClusterResponse{
+				Entry: &filer_pb.Entry{
+					Name:   req.Name,
+					Chunks: []*filer_pb.FileChunk{{FileId: "1,abc", Size: 100, Offset: 0}},
+				},
+			}, nil
+		},
+	})
+	s3a := newRemoteCacheTestServer(filerAddr)
+
+	r := httptest.NewRequest(http.MethodGet, "/mybucket/obj.bin", nil)
+	got := s3a.cacheRemoteObjectForStreaming(r, remoteOnlyEntry("obj.bin", 100), "mybucket", "obj.bin", "")
+
+	require.NotNil(t, got)
+	assert.Len(t, got.GetChunks(), 1)
+}