make the remote-mount cache wait configurable per mount (#11168)

* add a per-mount cache_wait_ms to the remote storage mount mapping

A read of an uncached remote-only object waits on a hardcoded size tier
before it can fall back to the origin, so every ranged read of a large
remote-only object pays that wait. Carry the wait in the mount mapping so
it can be tuned, or set to zero, per mount.

* resolve the cache wait of an uncached remote-only read from its mount

The wait came only from the object size, so an operator could not trade
cache hits for time to first byte. Both read paths now resolve the mount
covering the object and let its cache_wait_ms replace the size tiers.

* read straight from the remote when a mount waits zero for its cache

A mount used as a streaming source pays the cache wait on every ranged
read of an object too large to finish caching, and the caching itself is
wasted work. A zero wait now skips the cache call, so both read paths go
to the origin immediately.

* let remote.mount set the cache wait of a mount

remote.mount -cacheWait=0 turns a mount into a streaming source, and any
other duration trades cache hits against time to first byte.

* keep the size based wait for a version-specific read

A read pinned to a version cannot fall back to the origin, since the
mounted remote only holds the current key, so a mount that opts out of
caching would leave it on the 503 retry loop forever.

* let the operator allow a remote-only read to dial an internal endpoint

The remote-mount read paths in the filer and the S3 gateway always refused
an endpoint resolving to a loopback or private host, so a mount backed by
an internal S3 could never be read from its origin, only through the local
cache. Both now take the allowance the volume server already has, still
off by default.

* skip the background cache of a mount that waits zero for its cache

GetObjectHandler kicks off caching for every remote-only read, so a mount
serving as a streaming source kept downloading whole objects even though no
read ever waited for them.

* cover a zero cache wait end to end

The read has to reach a real origin, so the harness also opts the filer and
the S3 gateway into dialing the loopback remote it already allows for the
volume server.

* resolve the S3 cache wait once so the background cache follows it too

The background cache that GetObjectHandler starts read the mount on its
own, so it skipped a version-specific read that the foreground path still
waits for. Both now ask the same resolver.

* answer 404 when the origin of a zero-wait read is gone

Metadata can outlive the object it points at, and with no cache to fill
the read would sit on the 503 retry path forever. The remote backends
already report a missing object as ErrRemoteObjectNotFound.

* open the origin at write time for a multipart range

Every part of a multipart Range is prepared before any is written, so
opening eagerly would hold one origin connection per part and leak the
ones already opened when a later part fails to open.

* reject a cache wait shorter than a millisecond

The mapping stores milliseconds, so -cacheWait=500us truncated to zero
and silently turned caching off instead of waiting.

* restore the doc comment of cacheRemoteObjectForStreamingWithShortTimeout

Extracting the wait resolver left its comment on the new function.

* stat the origin before committing a multipart range

Opening at write time keeps no connection through the preparation, but it
also moved a failure past the point where the multipart body picks the
response status, so a gone origin truncated a 206 instead of answering
404. One stat up front puts the status back.

* stat the origin once per request

Every part of a multipart Range is prepared on its own, so the preflight
ran once per range instead of once per read.

* map Azure and GCS stream not-found to ErrRemoteObjectNotFound

ReadFileAsStream on Azure and GCS returned provider-specific not-found
errors instead of ErrRemoteObjectNotFound, so a zero-wait read of a
deleted object was misclassified as a transient cache failure and
retried indefinitely. Map BlobNotFound and ErrObjectNotExist the same
way StatFile already does.

* Update weed/remote_storage/gcs/gcs_storage_client.go

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/s3api/s3api_object_handlers.go

@@ -3294,17 +3294,10 @@ func (s3a *S3ApiServer) buildRemoteObjectPath(bucket, object string) (dir, name
 	return dir, name
 }
 
-// openRemoteStream opens a ranged read of a remote-only object straight from
-// its mounted origin, resolving the mount and storage conf from the filer.
-// cached, when set, is the remote generation the local copy was made from: the
-// stream is refused unless the remote still matches it.
-func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object string, offset, size int64, cached *filer_pb.RemoteEntry) (io.ReadCloser, error) {
-	dir, name := s3a.buildRemoteObjectPath(bucket, object)
-
-	var storageConf *remote_pb.RemoteConf
-	var localMountedDir string
-	var mountedLocation *remote_pb.RemoteStorageLocation
-	err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+// findMountedRemoteMapping resolves the remote mount covering dir from the
+// mount mapping kept in the filer.
+func (s3a *S3ApiServer) findMountedRemoteMapping(ctx context.Context, dir string) (localMountedDir string, mountedLocation *remote_pb.RemoteStorageLocation, err error) {
+	err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		mappingContent, readErr := filer.ReadInsideFiler(ctx, client, filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE)
 		if readErr != nil {
 			return readErr
@@ -3315,9 +3308,25 @@ func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object str
 		}
 		var findErr error
 		localMountedDir, mountedLocation, findErr = filer.FindMountedRemoteMapping(mappings, dir)
-		if findErr != nil {
-			return findErr
-		}
+		return findErr
+	})
+	return localMountedDir, mountedLocation, err
+}
+
+// openRemoteStream opens a ranged read of a remote-only object straight from
+// its mounted origin, resolving the mount and storage conf from the filer.
+// cached, when set, is the remote generation the local copy was made from: the
+// stream is refused unless the remote still matches it.
+func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object string, offset, size int64, cached *filer_pb.RemoteEntry) (io.ReadCloser, error) {
+	dir, name := s3a.buildRemoteObjectPath(bucket, object)
+
+	localMountedDir, mountedLocation, err := s3a.findMountedRemoteMapping(ctx, dir)
+	if err != nil {
+		return nil, err
+	}
+
+	var storageConf *remote_pb.RemoteConf
+	err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		confContent, readErr := filer.ReadInsideFiler(ctx, client, filer.DirectoryEtcRemote, mountedLocation.Name+filer.REMOTE_STORAGE_CONF_SUFFIX)
 		if readErr != nil {
 			return readErr
@@ -3329,7 +3338,7 @@ func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object str
 		return nil, err
 	}
 
-	client, err := weed_server.BuildGuardedRemoteStorageClient(ctx, storageConf, false)
+	client, err := weed_server.BuildGuardedRemoteStorageClient(ctx, storageConf, s3a.option.AllowUntrustedRemoteEndpoints)
 	if err != nil {
 		return nil, err
 	}
@@ -3396,19 +3405,41 @@ func cachedEntryHasLocalData(entry *filer_pb.Entry) bool {
 // Atomic so tests can shorten it without racing the read path.
 var remoteCacheStreamingTimeoutNS = int64(20 * time.Second)
 
+// remoteCacheWait resolves how long a read of dir may wait for the local cache.
+// Zero means the mount opted out of caching, so the read is served from the
+// origin instead -- except for a version-specific read, which has no origin key
+// to fall back to and therefore keeps the size tiers.
+func (s3a *S3ApiServer) remoteCacheWait(ctx context.Context, dir string, remoteSize int64, versionId string) time.Duration {
+	_, mountedLocation, mountErr := s3a.findMountedRemoteMapping(ctx, dir)
+	if mountErr != nil {
+		glog.V(2).Infof("remoteCacheWait: find mount for %s: %v", dir, mountErr)
+	}
+	wait := remote_storage.CacheWaitTimeout(remoteSize, mountedLocation)
+	if wait <= 0 && versionId != "" && versionId != "null" {
+		return remote_storage.CacheWaitTimeout(remoteSize, nil)
+	}
+	return wait
+}
+
 // 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.
+// Timeout comes from the file size or the mount's cache_wait_ms: 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) {
-	pollTimeout := remote_storage.CacheWaitTimeout(entry.GetRemoteEntry().GetRemoteSize())
+	dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId)
+
+	pollTimeout := s3a.remoteCacheWait(r.Context(), dir, entry.GetRemoteEntry().GetRemoteSize(), versionId)
+	if pollTimeout <= 0 {
+		// The mount opted out of caching: report it uncached so the caller serves the origin.
+		glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: caching disabled for %s/%s", dir, name)
+		return nil, nil
+	}
 
 	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)
@@ -3514,6 +3545,10 @@ func (s3a *S3ApiServer) startBackgroundRemoteCache(bucket, object, versionId str
 		// 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()
+		if s3a.remoteCacheWait(bgCtx, dir, entry.GetRemoteEntry().GetRemoteSize(), versionId) <= 0 {
+			glog.V(2).Infof("startBackgroundRemoteCache: caching disabled for %s/%s", dir, name)
+			return
+		}
 		_, err := s3a.doCacheRemoteObject(bgCtx, dir, name)
 		if err != nil {
 			glog.V(2).Infof("startBackgroundRemoteCache: cache failed for %s/%s: %v", bucket, object, err)

weed/s3api/s3api_remote_storage_test.go

@@ -637,10 +637,14 @@ func (m *fakeStreamRemoteMaker) HasBucket() bool { return true }
 
 // startStreamThroughFiler serves a filer whose cache RPC always fails with
 // cacheErr and whose /etc/remote mounts /buckets/mybucket on a fake origin.
-func startStreamThroughFiler(t *testing.T, remoteName string, cacheErr error) pb.ServerAddress {
+func startStreamThroughFiler(t *testing.T, remoteName string, cacheErr error, mountOpts ...func(*remote_pb.RemoteStorageLocation)) pb.ServerAddress {
+	mount := &remote_pb.RemoteStorageLocation{Name: remoteName, Bucket: "origin-bucket", Path: "/data"}
+	for _, opt := range mountOpts {
+		opt(mount)
+	}
 	mappingBytes, err := proto.Marshal(&remote_pb.RemoteStorageMapping{
 		Mappings: map[string]*remote_pb.RemoteStorageLocation{
-			"/buckets/mybucket": {Name: remoteName, Bucket: "origin-bucket", Path: "/data"},
+			"/buckets/mybucket": mount,
 		},
 	})
 	require.NoError(t, err)
@@ -767,6 +771,41 @@ func TestS3ColdReadStreamsFromOrigin(t *testing.T) {
 		assert.Equal(t, content, w.Body.Bytes())
 	})
 
+	t.Run("mount with a zero cache wait never asks the cache", func(t *testing.T) {
+		originClient := &fakeStreamRemoteClient{data: content}
+		remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: originClient}
+		defer func() {
+			remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client}
+		}()
+		// a consulted cache would 404 this read, so a 200 proves it was skipped
+		s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-nowait", status.Error(codes.NotFound, "entry vanished"), func(loc *remote_pb.RemoteStorageLocation) {
+			loc.CacheWaitMs = proto.Int32(0)
+		}))
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+
+		err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "")
+
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusOK, w.Code)
+		assert.Equal(t, content, w.Body.Bytes())
+		require.NotNil(t, originClient.gotLoc, "must read from the origin")
+	})
+
+	t.Run("zero cache wait still caches a version-specific read", func(t *testing.T) {
+		// the version has no origin key, so the cache stays its only source
+		s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-nowait-versioned", status.Error(codes.NotFound, "entry vanished"), func(loc *remote_pb.RemoteStorageLocation) {
+			loc.CacheWaitMs = proto.Int32(0)
+		}))
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin?versionId=v123", nil)
+
+		err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "v123")
+
+		require.Error(t, err)
+		assert.Equal(t, http.StatusNotFound, w.Code)
+	})
+
 	t.Run("entry not found stays 404", func(t *testing.T) {
 		notFoundForms := map[string]error{
 			"canonical status": status.Error(codes.NotFound, "entry vanished"),

weed/s3api/s3api_server.go

@@ -69,6 +69,9 @@ type S3ApiServerOption struct {
 	DefaultFileMode           uint32 // default file permission mode for S3 uploads (e.g. 0660, 0644)
 	CacheSizeMB               int64  // in-memory chunk cache capacity in MB for the shared ReaderCache; 0 disables
 	MaxMB                     int32  // filer's -maxMB, read from the filer configuration at startup
+	// AllowUntrustedRemoteEndpoints lets a read of a remote-only object dial a
+	// mounted endpoint that resolves to a loopback / private / metadata host.
+	AllowUntrustedRemoteEndpoints bool
 }
 
 // s3ChunkCacheChunkSizeMB is the assumed chunk size (in MiB) used to convert