diff --git a/seaweed-volume/proto/remote.proto b/seaweed-volume/proto/remote.proto
index d79a79df9..c5d81eda5 100644
--- a/seaweed-volume/proto/remote.proto
+++ b/seaweed-volume/proto/remote.proto
@@ -75,4 +75,6 @@ message RemoteStorageLocation {
   string name = 1;
   string bucket = 2;
   string path = 3;
+  int32 listing_cache_ttl_seconds = 4; // 0 = disabled; >0 enables on-demand directory listing with this TTL in seconds
+  optional int32 cache_wait_ms = 5; // unset = size based default; 0 = read straight from the remote without caching
 }
diff --git a/test/s3/remote_cache/Makefile b/test/s3/remote_cache/Makefile
index 762db65c5..bd6d693bd 100644
--- a/test/s3/remote_cache/Makefile
+++ b/test/s3/remote_cache/Makefile
@@ -128,6 +128,8 @@ start-primary: check-deps
 		-s3.allowDeleteBucketNotEmpty=true \
 		-s3.config=s3_config.json \
 		-volume.allowUntrustedRemoteEndpoints \
+		-filer.allowUntrustedRemoteEndpoints \
+		-s3.allowUntrustedRemoteEndpoints \
 		-dir=$(PRIMARY_DIR) \
 		-ip=127.0.0.1 \
 		-ip.bind=127.0.0.1 \
diff --git a/test/s3/remote_cache/README.md b/test/s3/remote_cache/README.md
index 516320a2c..7a54032ab 100644
--- a/test/s3/remote_cache/README.md
+++ b/test/s3/remote_cache/README.md
@@ -42,7 +42,7 @@ This tests the full remote caching workflow including singleflight deduplication
 
 | Test File | Commands Tested | Test Count | Description |
 |-----------|----------------|------------|-------------|
-| `remote_cache_test.go` | Basic caching | 5 tests | Original caching workflow and singleflight tests |
+| `remote_cache_test.go` | Basic caching | 6 tests | Original caching workflow and singleflight tests, plus a mount with `-cacheWait=0` reading straight from the remote |
 | `remote_cache_copy_test.go` | S3 CopyObject / UploadPartCopy from a remote-only source | 2 tests | Source object lives only in remote storage; CopyObject and UploadPartCopy must cache it locally before persisting the destination so the result is readable |
 | `command_remote_configure_test.go` | `remote.configure` | 6 tests | Configuration management |
 | `command_remote_mount_test.go` | `remote.mount`, `remote.unmount`, `remote.mount.buckets` | 10 tests | Mount operations |
@@ -51,7 +51,7 @@ This tests the full remote caching workflow including singleflight deduplication
 | `command_remote_meta_sync_test.go` | `remote.meta.sync` | 8 tests | Metadata synchronization |
 | `command_edge_cases_test.go` | All commands | 11 tests | Edge cases and stress tests |
 
-**Total: 67 test cases covering 8 weed shell commands and the S3 copy paths for remote-only sources**
+**Total: 68 test cases covering 8 weed shell commands and the S3 copy paths for remote-only sources**
 
 ### Commands Tested
 
@@ -223,6 +223,11 @@ Tests HTTP range requests work correctly after caching.
 ### TestRemoteCacheNotFound
 Tests proper error handling for non-existent objects.
 
+### TestRemoteCacheWaitZero
+Remounts with `remote.mount -cacheWait=0` and checks that a read of a remote-only
+object is served from the remote and leaves the object uncached, while the default
+size-based wait caches it.
+
 ## Troubleshooting
 
 ### View logs
diff --git a/test/s3/remote_cache/remote_cache_test.go b/test/s3/remote_cache/remote_cache_test.go
index 535c4ba89..4da4c024e 100644
--- a/test/s3/remote_cache/remote_cache_test.go
+++ b/test/s3/remote_cache/remote_cache_test.go
@@ -40,6 +40,9 @@ const (
 	// Bucket name - mounted on primary as remote storage
 	testBucket = "remotemounted"
 
+	// Bucket name on the remote that testBucket is mounted on
+	remoteBucket = "remotesourcebucket"
+
 	// Path to weed binary
 	weedBinary = "../../../weed/weed_binary"
 )
@@ -388,6 +391,66 @@ func TestRemoteCacheRangeRequest(t *testing.T) {
 	t.Log("Range request test passed")
 }
 
+// mountRemote remounts the test bucket, keeping the metadata already pulled,
+// with the given extra remote.mount flags.
+func mountRemote(t *testing.T, extraFlags string) {
+	cmd := fmt.Sprintf("remote.mount -dir=/buckets/%s -remote=seaweedremote/%s -nonempty -metadataStrategy=lazy %s", testBucket, remoteBucket, extraFlags)
+	output, err := runWeedShell(t, cmd)
+	require.NoErrorf(t, err, "remount failed: %s", output)
+	time.Sleep(time.Second)
+}
+
+// copyLocalToRemote pushes matching local objects to the remote, which is what
+// lets remote.uncache drop their local chunks afterwards.
+func copyLocalToRemote(t *testing.T, pattern string) {
+	output, err := runWeedShell(t, fmt.Sprintf("remote.copy.local -dir=/buckets/%s -include=%s", testBucket, pattern))
+	require.NoErrorf(t, err, "remote.copy.local failed: %s", output)
+}
+
+// localChunkCount reports how many local chunks an entry has, so a test can
+// tell a cached read from one served straight out of the remote.
+func localChunkCount(t *testing.T, key string) string {
+	meta, err := runWeedShell(t, fmt.Sprintf("fs.meta.cat /buckets/%s/%s", testBucket, key))
+	require.NoError(t, err)
+	// fs.meta.cat closes with "chunks N meta size: ..."
+	summary := strings.LastIndex(meta, "chunks ")
+	require.GreaterOrEqualf(t, summary, 0, "no chunk count in %s", meta)
+	return strings.Fields(meta[summary+len("chunks "):])[0]
+}
+
+// TestRemoteCacheWaitZero tests that a mount with -cacheWait=0 serves a read of
+// an uncached object from the remote without caching it locally.
+func TestRemoteCacheWaitZero(t *testing.T) {
+	checkServersRunning(t)
+
+	testKey := fmt.Sprintf("test-cachewait-%d.bin", time.Now().UnixNano())
+	testData := make([]byte, 1024*1024)
+	for i := range testData {
+		testData[i] = byte(i % 256)
+	}
+
+	t.Log("Step 1: Writing 1MB object, pushing it to the remote and dropping the local chunks...")
+	uploadToPrimary(t, testKey, testData)
+	copyLocalToRemote(t, testKey)
+	uncacheLocal(t, testKey)
+	require.Equal(t, "0", localChunkCount(t, testKey), "the object must start out remote-only")
+
+	t.Log("Step 2: Remounting with -cacheWait=0...")
+	mountRemote(t, "-cacheWait=0")
+	defer mountRemote(t, "")
+
+	t.Log("Step 3: Reading the object (should stream from remote)...")
+	assert.Equal(t, testData, getFromPrimary(t, testKey), "data mismatch reading from remote")
+	assert.Equal(t, "0", localChunkCount(t, testKey), "the read must not cache chunks locally")
+
+	t.Log("Step 4: Reading with the size based wait (should cache locally)...")
+	mountRemote(t, "")
+	assert.Equal(t, testData, getFromPrimary(t, testKey), "data mismatch reading through the cache")
+	assert.NotEqual(t, "0", localChunkCount(t, testKey), "the read must cache chunks locally")
+
+	t.Log("Zero cache wait test passed")
+}
+
 // TestRemoteCacheNotFound tests that non-existent objects return proper errors
 func TestRemoteCacheNotFound(t *testing.T) {
 	checkServersRunning(t)
diff --git a/weed/command/filer.go b/weed/command/filer.go
index 6b3503aac..0fa629996 100644
--- a/weed/command/filer.go
+++ b/weed/command/filer.go
@@ -46,6 +46,10 @@ var (
 	filerSftpOptions   SftpOptions
 )
 
+// allowUntrustedRemoteEndpointsUsage documents the flag shared by the filer and
+// S3 gateway, whose remote-mount read paths dial the mounted endpoint directly.
+const allowUntrustedRemoteEndpointsUsage = "if true, a read of a remote-only entry accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints."
+
 type FilerOptions struct {
 	masters                   *pb.ServerDiscovery
 	mastersString             *string
@@ -83,6 +87,8 @@ type FilerOptions struct {
 	tusMaxSizeMB              *int
 	tusSessionExpiry          *time.Duration
 	s3ConfigFile              *string // optional path to static S3 identity config
+
+	allowUntrustedRemoteEndpoints *bool
 	// shutdownCtx, when non-nil, tells startFiler to gracefully shut down its
 	// HTTP/gRPC servers once the ctx is cancelled. Used by integration tests
 	// and by weed mini; nil for standalone weed filer.
@@ -127,6 +133,7 @@ func init() {
 	f.tusBasePath = cmdFiler.Flag.String("tusBasePath", "/.tus", "TUS resumable upload endpoint base path (e.g., /.tus)")
 	f.tusMaxSizeMB = cmdFiler.Flag.Int("tusMaxSizeMB", 5*1024, "maximum TUS upload size in MB")
 	f.tusSessionExpiry = cmdFiler.Flag.Duration("tusSessionExpiry", 24*time.Hour, "incomplete TUS upload sessions are cleaned up after this duration, e.g. \"48h\", \"7h30m\"")
+	f.allowUntrustedRemoteEndpoints = cmdFiler.Flag.Bool("allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
 
 	// start s3 on filer
 	filerStartS3 = cmdFiler.Flag.Bool("s3", false, "whether to start S3 gateway")
@@ -161,6 +168,7 @@ func init() {
 	filerS3Options.externalUrl = cmdFiler.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 	filerS3Options.defaultFileMode = cmdFiler.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666")
 	filerS3Options.cacheSizeMB = cmdFiler.Flag.Int64("s3.cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)")
+	filerS3Options.allowUntrustedRemoteEndpoints = cmdFiler.Flag.Bool("s3.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
 
 	// start webdav on filer
 	filerStartWebDav = cmdFiler.Flag.Bool("webdav", false, "whether to start webdav gateway")
@@ -396,6 +404,8 @@ func (fo *FilerOptions) startFiler() {
 		TusMaxSize:                int64(*fo.tusMaxSizeMB) * 1024 * 1024,
 		TusSessionExpiry:          *fo.tusSessionExpiry,
 		CredentialManager:         credentialManager,
+
+		AllowUntrustedRemoteEndpoints: *fo.allowUntrustedRemoteEndpoints,
 	})
 	if nfs_err != nil {
 		glog.Fatalf("Filer startup error: %v", nfs_err)
diff --git a/weed/command/mini.go b/weed/command/mini.go
index a70eff9b1..29cb36e58 100644
--- a/weed/command/mini.go
+++ b/weed/command/mini.go
@@ -464,6 +464,7 @@ func initMiniFilerFlags() {
 	miniFilerOptions.tusBasePath = cmdMini.Flag.String("filer.tusBasePath", "/.tus", "TUS resumable upload endpoint base path")
 	miniFilerOptions.tusMaxSizeMB = cmdMini.Flag.Int("filer.tusMaxSizeMB", 5*1024, "maximum TUS upload size in MB")
 	miniFilerOptions.tusSessionExpiry = cmdMini.Flag.Duration("filer.tusSessionExpiry", 24*time.Hour, "incomplete TUS upload sessions are cleaned up after this duration")
+	miniFilerOptions.allowUntrustedRemoteEndpoints = cmdMini.Flag.Bool("filer.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
 }
 
 // initMiniVolumeFlags initializes Volume server flag options
@@ -530,6 +531,7 @@ func initMiniS3Flags() {
 	miniS3Options.externalUrl = cmdMini.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 	miniS3Options.defaultFileMode = cmdMini.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666")
 	miniS3Options.cacheSizeMB = cmdMini.Flag.Int64("s3.cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)")
+	miniS3Options.allowUntrustedRemoteEndpoints = cmdMini.Flag.Bool("s3.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
 	// In mini mode, S3 uses the shared debug server started at line 681, not its own separate debug server
 	miniS3Options.debug = new(bool) // explicitly false
 	miniS3Options.debugPort = cmdMini.Flag.Int("s3.debug.port", 6060, "http port for debugging (unused in mini mode)")
diff --git a/weed/command/s3.go b/weed/command/s3.go
index e7051401d..6ca45f057 100644
--- a/weed/command/s3.go
+++ b/weed/command/s3.go
@@ -80,6 +80,8 @@ type S3Options struct {
 	externalUrl               *string
 	defaultFileMode           *string
 	cacheSizeMB               *int64
+
+	allowUntrustedRemoteEndpoints *bool
 	// shutdownCtx, when non-nil, tells startS3Server/startIcebergServer to
 	// gracefully shut down their HTTP/gRPC servers once the ctx is cancelled.
 	// Used by weed mini to orchestrate an ordered shutdown; nil for standalone
@@ -127,6 +129,7 @@ func init() {
 	s3StandaloneOptions.externalUrl = cmdS3.Flag.String("externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 	s3StandaloneOptions.defaultFileMode = cmdS3.Flag.String("defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666")
 	s3StandaloneOptions.cacheSizeMB = cmdS3.Flag.Int64("cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)")
+	s3StandaloneOptions.allowUntrustedRemoteEndpoints = cmdS3.Flag.Bool("allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
 }
 
 var cmdS3 = &Command{
@@ -378,6 +381,8 @@ func (s3opt *S3Options) startS3Server() bool {
 		DefaultFileMode:           defaultFileMode,
 		CacheSizeMB:               *s3opt.cacheSizeMB,
 		MaxMB:                     filerMaxMB,
+
+		AllowUntrustedRemoteEndpoints: *s3opt.allowUntrustedRemoteEndpoints,
 	})
 	if s3ApiServer_err != nil {
 		glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err)
diff --git a/weed/command/server.go b/weed/command/server.go
index 732121cd4..72b2e7a2d 100644
--- a/weed/command/server.go
+++ b/weed/command/server.go
@@ -132,6 +132,7 @@ func init() {
 	filerOptions.tusBasePath = cmdServer.Flag.String("filer.tusBasePath", "/.tus", "TUS resumable upload endpoint base path (e.g., /.tus)")
 	filerOptions.tusMaxSizeMB = cmdServer.Flag.Int("filer.tusMaxSizeMB", 5*1024, "maximum TUS upload size in MB")
 	filerOptions.tusSessionExpiry = cmdServer.Flag.Duration("filer.tusSessionExpiry", 24*time.Hour, "incomplete TUS upload sessions are cleaned up after this duration, e.g. \"48h\", \"7h30m\"")
+	filerOptions.allowUntrustedRemoteEndpoints = cmdServer.Flag.Bool("filer.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
 
 	serverOptions.v.port = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port")
 	serverOptions.v.portGrpc = cmdServer.Flag.Int("volume.port.grpc", 0, "volume server grpc listen port")
@@ -190,6 +191,7 @@ func init() {
 	s3Options.externalUrl = cmdServer.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
 	s3Options.defaultFileMode = cmdServer.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666")
 	s3Options.cacheSizeMB = cmdServer.Flag.Int64("s3.cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)")
+	s3Options.allowUntrustedRemoteEndpoints = cmdServer.Flag.Bool("s3.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
 
 	sftpOptions.port = cmdServer.Flag.Int("sftp.port", 2022, "SFTP server listen port")
 	sftpOptions.sshPrivateKey = cmdServer.Flag.String("sftp.sshPrivateKey", "", "path to the SSH private key file for host authentication")
diff --git a/weed/pb/remote.proto b/weed/pb/remote.proto
index 9654719de..c5d81eda5 100644
--- a/weed/pb/remote.proto
+++ b/weed/pb/remote.proto
@@ -76,4 +76,5 @@ message RemoteStorageLocation {
   string bucket = 2;
   string path = 3;
   int32 listing_cache_ttl_seconds = 4; // 0 = disabled; >0 enables on-demand directory listing with this TTL in seconds
+  optional int32 cache_wait_ms = 5; // unset = size based default; 0 = read straight from the remote without caching
 }
diff --git a/weed/pb/remote_pb/remote.pb.go b/weed/pb/remote_pb/remote.pb.go
index 611819c08..e246d2744 100644
--- a/weed/pb/remote_pb/remote.pb.go
+++ b/weed/pb/remote_pb/remote.pb.go
@@ -478,6 +478,7 @@ type RemoteStorageLocation struct {
 	Bucket                 string                 `protobuf:"bytes,2,opt,name=bucket,proto3" json:"bucket,omitempty"`
 	Path                   string                 `protobuf:"bytes,3,opt,name=path,proto3" json:"path,omitempty"`
 	ListingCacheTtlSeconds int32                  `protobuf:"varint,4,opt,name=listing_cache_ttl_seconds,json=listingCacheTtlSeconds,proto3" json:"listing_cache_ttl_seconds,omitempty"` // 0 = disabled; >0 enables on-demand directory listing with this TTL in seconds
+	CacheWaitMs            *int32                 `protobuf:"varint,5,opt,name=cache_wait_ms,json=cacheWaitMs,proto3,oneof" json:"cache_wait_ms,omitempty"`                              // unset = size based default; 0 = read straight from the remote without caching
 	unknownFields          protoimpl.UnknownFields
 	sizeCache              protoimpl.SizeCache
 }
@@ -540,6 +541,13 @@ func (x *RemoteStorageLocation) GetListingCacheTtlSeconds() int32 {
 	return 0
 }
 
+func (x *RemoteStorageLocation) GetCacheWaitMs() int32 {
+	if x != nil && x.CacheWaitMs != nil {
+		return *x.CacheWaitMs
+	}
+	return 0
+}
+
 var File_remote_proto protoreflect.FileDescriptor
 
 const file_remote_proto_rawDesc = "" +
@@ -599,12 +607,14 @@ const file_remote_proto_rawDesc = "" +
 	"\x1bprimary_bucket_storage_name\x18\x02 \x01(\tR\x18primaryBucketStorageName\x1a]\n" +
 	"\rMappingsEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x126\n" +
-	"\x05value\x18\x02 \x01(\v2 .remote_pb.RemoteStorageLocationR\x05value:\x028\x01\"\x92\x01\n" +
+	"\x05value\x18\x02 \x01(\v2 .remote_pb.RemoteStorageLocationR\x05value:\x028\x01\"\xcd\x01\n" +
 	"\x15RemoteStorageLocation\x12\x12\n" +
 	"\x04name\x18\x01 \x01(\tR\x04name\x12\x16\n" +
 	"\x06bucket\x18\x02 \x01(\tR\x06bucket\x12\x12\n" +
 	"\x04path\x18\x03 \x01(\tR\x04path\x129\n" +
-	"\x19listing_cache_ttl_seconds\x18\x04 \x01(\x05R\x16listingCacheTtlSecondsBP\n" +
+	"\x19listing_cache_ttl_seconds\x18\x04 \x01(\x05R\x16listingCacheTtlSeconds\x12'\n" +
+	"\rcache_wait_ms\x18\x05 \x01(\x05H\x00R\vcacheWaitMs\x88\x01\x01B\x10\n" +
+	"\x0e_cache_wait_msBP\n" +
 	"\x10seaweedfs.clientB\n" +
 	"FilerProtoZ0github.com/seaweedfs/seaweedfs/weed/pb/remote_pbb\x06proto3"
 
@@ -642,6 +652,7 @@ func file_remote_proto_init() {
 	if File_remote_proto != nil {
 		return
 	}
+	file_remote_proto_msgTypes[2].OneofWrappers = []any{}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
diff --git a/weed/remote_storage/azure/azure_storage_client.go b/weed/remote_storage/azure/azure_storage_client.go
index 18c7ed47d..f72ffc8c2 100644
--- a/weed/remote_storage/azure/azure_storage_client.go
+++ b/weed/remote_storage/azure/azure_storage_client.go
@@ -382,6 +382,9 @@ func (az *azureRemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *r
 		},
 	})
 	if err != nil {
+		if bloberror.HasCode(err, bloberror.BlobNotFound) {
+			return nil, remote_storage.ErrRemoteObjectNotFound
+		}
 		return nil, fmt.Errorf("failed to open stream for %s%s: %v", loc.Bucket, loc.Path, err)
 	}
 	return downloadResponse.Body, nil
diff --git a/weed/remote_storage/gcs/gcs_storage_client.go b/weed/remote_storage/gcs/gcs_storage_client.go
index 12219064a..374fa3a44 100644
--- a/weed/remote_storage/gcs/gcs_storage_client.go
+++ b/weed/remote_storage/gcs/gcs_storage_client.go
@@ -265,7 +265,14 @@ func (gcs *gcsRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation
 
 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).ReadCompressed(true).NewRangeReader(ctx, offset, size)
+	reader, err = gcs.client.Bucket(loc.Bucket).Object(key).ReadCompressed(true).NewRangeReader(ctx, offset, size)
+	if err != nil {
+		if errors.Is(err, storage.ErrObjectNotExist) {
+			return nil, remote_storage.ErrRemoteObjectNotFound
+		}
+		return nil, fmt.Errorf("failed to open stream for %s%s: %w", loc.Bucket, loc.Path, err)
+	}
+	return reader, nil
 }
 
 func (gcs *gcsRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) {
diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go
index e23ffbedd..e66cdb947 100644
--- a/weed/remote_storage/remote_storage.go
+++ b/weed/remote_storage/remote_storage.go
@@ -112,8 +112,12 @@ type RemoteStorageStreamReader interface {
 
 // CacheWaitTimeout is how long a read of an uncached remote-only object waits
 // for the local cache before serving another way: small files wait longer since
-// their cache completes quickly, large files fail fast for better TTFB.
-func CacheWaitTimeout(remoteSize int64) time.Duration {
+// their cache completes quickly, large files fail fast for better TTFB. A mount
+// carrying cache_wait_ms replaces the size tiers, and 0 means never wait.
+func CacheWaitTimeout(remoteSize int64, mountedLocation *remote_pb.RemoteStorageLocation) time.Duration {
+	if mountedLocation != nil && mountedLocation.CacheWaitMs != nil {
+		return max(0, time.Duration(*mountedLocation.CacheWaitMs)*time.Millisecond)
+	}
 	switch {
 	case remoteSize > 500*1024*1024:
 		return 2 * time.Second
diff --git a/weed/remote_storage/remote_storage_test.go b/weed/remote_storage/remote_storage_test.go
new file mode 100644
index 000000000..6417cc65d
--- /dev/null
+++ b/weed/remote_storage/remote_storage_test.go
@@ -0,0 +1,54 @@
+package remote_storage
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestCacheWaitTimeout(t *testing.T) {
+	tests := []struct {
+		name            string
+		remoteSize      int64
+		mountedLocation *remote_pb.RemoteStorageLocation
+		want            time.Duration
+	}{
+		{name: "large file fails fast", remoteSize: 600 * 1024 * 1024, want: 2 * time.Second},
+		{name: "small file waits longer", remoteSize: 1024, want: 10 * time.Second},
+		{name: "medium file", remoteSize: 100 * 1024 * 1024, want: 5 * time.Second},
+		{name: "unknown size", want: 5 * time.Second},
+		{
+			name:            "mount replaces the size tier",
+			remoteSize:      600 * 1024 * 1024,
+			mountedLocation: &remote_pb.RemoteStorageLocation{CacheWaitMs: proto.Int32(30000)},
+			want:            30 * time.Second,
+		},
+		{
+			name:            "mount opts out of caching",
+			remoteSize:      1024,
+			mountedLocation: &remote_pb.RemoteStorageLocation{CacheWaitMs: proto.Int32(0)},
+			want:            0,
+		},
+		{
+			name:            "negative wait never waits",
+			remoteSize:      1024,
+			mountedLocation: &remote_pb.RemoteStorageLocation{CacheWaitMs: proto.Int32(-1)},
+			want:            0,
+		},
+		{
+			name:            "unset wait keeps the size tier",
+			remoteSize:      600 * 1024 * 1024,
+			mountedLocation: &remote_pb.RemoteStorageLocation{},
+			want:            2 * time.Second,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := CacheWaitTimeout(tt.remoteSize, tt.mountedLocation); got != tt.want {
+				t.Errorf("CacheWaitTimeout(%d) = %v, want %v", tt.remoteSize, got, tt.want)
+			}
+		})
+	}
+}
diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go
index f4563681d..58d0eb3e4 100644
--- a/weed/s3api/s3api_object_handlers.go
+++ b/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)
diff --git a/weed/s3api/s3api_remote_storage_test.go b/weed/s3api/s3api_remote_storage_test.go
index 92c910e0c..2edc86400 100644
--- a/weed/s3api/s3api_remote_storage_test.go
+++ b/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"),
diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go
index f9fd535f6..34780547c 100644
--- a/weed/s3api/s3api_server.go
+++ b/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
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index c5e323897..d1abf8e56 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -88,6 +88,9 @@ type FilerOption struct {
 	TusSessionExpiry          time.Duration
 	S3ConfigFile              string // optional path to static S3 identity config file
 	CredentialManager         *credential.CredentialManager
+	// AllowUntrustedRemoteEndpoints lets a read of a remote-only entry dial a
+	// mounted endpoint that resolves to a loopback / private / metadata host.
+	AllowUntrustedRemoteEndpoints bool
 }
 
 type FilerServer struct {
diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go
index c3c597b6e..ff7e1d09b 100644
--- a/weed/server/filer_server_handlers_read.go
+++ b/weed/server/filer_server_handlers_read.go
@@ -11,11 +11,13 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"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/pb/remote_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/security"
@@ -202,6 +204,18 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		stats.RecordRemoteCacheRead(stats.RemoteCacheSourceFiler, fs.filer.DetectBucket(entry.FullPath), hit)
 	}
 
+	// Every part of a multipart Range is prepared on its own, so the origin
+	// preflight below is memoized to one stat per request.
+	statOrigin := sync.OnceValue(func() error {
+		dir, name := entry.FullPath.DirAndName()
+		client, remoteLocation, err := fs.mountedRemoteClient(ctx, dir, name)
+		if err != nil {
+			return err
+		}
+		_, err = client.StatFile(remoteLocation)
+		return err
+	})
+
 	ProcessRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
 		if offset+size <= int64(len(entry.Content)) {
 			return func(writer io.Writer) error {
@@ -216,10 +230,18 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		chunks := entry.GetChunks()
 		if entry.IsInRemoteOnly() {
 			dir, name := entry.FullPath.DirAndName()
+			var mountedLocation *remote_pb.RemoteStorageLocation
+			if fs.filer.RemoteStorage != nil {
+				_, mountedLocation = fs.filer.RemoteStorage.FindMountDirectory(entry.FullPath)
+			}
+			cacheWait := remote_storage.CacheWaitTimeout(entry.Remote.RemoteSize, mountedLocation)
+			if cacheWait <= 0 {
+				return fs.streamFromRemoteOnly(ctx, r, dir, name, offset, size, statOrigin)
+			}
 			// Bounded wait: a large download outlasts any client timeout, so
 			// serve straight from the origin once the wait expires while the
 			// detached cache keeps filling for later reads.
-			cacheCtx, cancelCache := context.WithTimeout(ctx, remote_storage.CacheWaitTimeout(entry.Remote.RemoteSize))
+			cacheCtx, cancelCache := context.WithTimeout(ctx, cacheWait)
 			resp, err := fs.CacheRemoteObjectToLocalCluster(cacheCtx, &filer_pb.CacheRemoteObjectToLocalClusterRequest{
 				Directory: dir,
 				Name:      name,
@@ -279,22 +301,71 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 	})
 }
 
+// streamFromRemoteOnly serves a byte range of an entry whose mount opted out of
+// caching, leaving the origin as its only source. A multipart Range prepares
+// every part before writing any, so those open the origin at write time instead
+// of holding one connection per part through the whole preparation.
+func (fs *FilerServer) streamFromRemoteOnly(ctx context.Context, r *http.Request, dir, name string, offset, size int64, statOrigin func() error) (filer.DoStreamContent, error) {
+	fullPath := util.FullPath(dir).Child(name)
+	if strings.Contains(r.Header.Get("Range"), ",") {
+		// Stat first so a missing or unreachable origin still picks the response
+		// status, which the multipart body would otherwise have committed.
+		if err := statOrigin(); err != nil {
+			return nil, fs.remoteReadError(ctx, fullPath, err)
+		}
+		return func(writer io.Writer) error {
+			streamFn, remoteErr := fs.streamFromRemote(ctx, dir, name, offset, size)
+			if remoteErr != nil {
+				stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
+				return remoteErr
+			}
+			return streamFn(writer)
+		}, nil
+	}
+	streamFn, remoteErr := fs.streamFromRemote(ctx, dir, name, offset, size)
+	if remoteErr != nil {
+		return nil, fs.remoteReadError(ctx, fullPath, remoteErr)
+	}
+	return streamFn, nil
+}
+
+// remoteReadError maps a failed origin read of an uncachable entry onto the
+// sentinel the caller turns into a response: a vanished object is final, since
+// no cache can resurrect it, while anything else may pass.
+func (fs *FilerServer) remoteReadError(ctx context.Context, fullPath util.FullPath, err error) error {
+	stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
+	glog.WarningfCtx(ctx, "read %s from remote: %v", fullPath, err)
+	if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) {
+		return filer_pb.ErrNotFound
+	}
+	return fmt.Errorf("read %s: %w", fullPath, ErrCacheNotReady)
+}
+
+// mountedRemoteClient resolves the origin of dir/name into a client that can
+// stream it.
+func (fs *FilerServer) mountedRemoteClient(ctx context.Context, dir, name string) (remote_storage.RemoteStorageClient, *remote_pb.RemoteStorageLocation, error) {
+	storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, dir, name)
+	if err != nil {
+		return nil, nil, err
+	}
+	client, err := BuildGuardedRemoteStorageClient(ctx, storageConf, fs.option.AllowUntrustedRemoteEndpoints)
+	if err != nil {
+		return nil, nil, err
+	}
+	if _, ok := client.(remote_storage.RemoteStorageStreamReader); !ok {
+		return nil, nil, fmt.Errorf("remote storage type %s does not support streaming reads", storageConf.Type)
+	}
+	return client, remoteLocation, nil
+}
+
 // streamFromRemote serves a byte range of a remote-only entry straight from the
 // mounted origin, so a first read is not blocked by the full local caching.
 func (fs *FilerServer) streamFromRemote(ctx context.Context, dir, name string, offset, size int64) (filer.DoStreamContent, error) {
-	storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, dir, name)
+	client, remoteLocation, err := fs.mountedRemoteClient(ctx, dir, name)
 	if err != nil {
 		return nil, err
 	}
-	client, err := BuildGuardedRemoteStorageClient(ctx, storageConf, false)
-	if err != nil {
-		return nil, err
-	}
-	streamer, ok := client.(remote_storage.RemoteStorageStreamReader)
-	if !ok {
-		return nil, fmt.Errorf("remote storage type %s does not support streaming reads", storageConf.Type)
-	}
-	reader, err := streamer.ReadFileAsStream(ctx, remoteLocation, offset, size)
+	reader, err := client.(remote_storage.RemoteStorageStreamReader).ReadFileAsStream(ctx, remoteLocation, offset, size)
 	if err != nil {
 		return nil, err
 	}
diff --git a/weed/shell/command_remote_mount.go b/weed/shell/command_remote_mount.go
index 60c1c93e0..1eeb43bf5 100644
--- a/weed/shell/command_remote_mount.go
+++ b/weed/shell/command_remote_mount.go
@@ -6,6 +6,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"strings"
 	"time"
@@ -50,6 +51,8 @@ func (c *commandRemoteMount) Help() string {
 	remote.mount -dir=/xxx -remote=cloud1/bucket/dir1
 	# mount with on-demand directory listing cached for 5 minutes
 	remote.mount -dir=/xxx -remote=cloud1/bucket -listingCacheTTL=300
+	# mount as a streaming source: reads go to the remote instead of waiting for the local cache
+	remote.mount -dir=/xxx -remote=cloud1/bucket -cacheWait=0
 
 	# after mount, start a separate process to write updates to remote storage
 	weed filer.remote.sync -filer=<filerHost>:<filerPort> -dir=/xxx
@@ -70,6 +73,7 @@ func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io
 	metadataStrategy := remoteMountCommand.String("metadataStrategy", string(MetadataCacheEager), "lazy: skip upfront metadata pull; eager: full metadata pull (default)")
 	remote := remoteMountCommand.String("remote", "", "a directory in remote storage, ex. <storageName>/<bucket>/path/to/dir")
 	listingCacheTTL := remoteMountCommand.Int("listingCacheTTL", 0, "seconds to cache remote directory listings (0 = disabled)")
+	cacheWait := remoteMountCommand.Duration("cacheWait", -1, "how long a read of an uncached object waits for the local cache, ex. 0 or 500ms (default: by object size)")
 
 	if err = remoteMountCommand.Parse(args); err != nil {
 		return nil
@@ -91,6 +95,17 @@ func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io
 		return err
 	}
 	remoteStorageLocation.ListingCacheTtlSeconds = int32(*listingCacheTTL)
+	if *cacheWait >= 0 {
+		waitMs := cacheWait.Milliseconds()
+		if waitMs > math.MaxInt32 {
+			return fmt.Errorf("cacheWait %v is too long", *cacheWait)
+		}
+		// truncating to 0 would read as "never wait" instead of the asked-for wait
+		if waitMs == 0 && *cacheWait > 0 {
+			return fmt.Errorf("cacheWait %v is shorter than 1ms", *cacheWait)
+		}
+		remoteStorageLocation.CacheWaitMs = proto.Int32(int32(waitMs))
+	}
 
 	strategy := MetadataCacheStrategy(strings.ToLower(*metadataStrategy))
 	if strategy != MetadataCacheLazy && strategy != MetadataCacheEager {