diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto
index b8beefd71..ca1c0114c 100644
--- a/other/java/client/src/main/proto/filer.proto
+++ b/other/java/client/src/main/proto/filer.proto
@@ -576,6 +576,7 @@ message Location {
     string public_url = 2;
     uint32 grpc_port = 3;
     string data_center = 4;
+    bool data_in_remote = 5;
 }
 message LookupVolumeResponse {
     map<string, Locations> locations_map = 1;
@@ -662,6 +663,7 @@ message SubscribeMetadataResponse {
     int64 ts_ns = 3;
     repeated SubscribeMetadataResponse events = 4; // batch of additional events (backlog catch-up)
     repeated LogFileChunkRef log_file_refs = 5;    // log file chunk refs for client direct-read
+    int64 flushed_ts_ns = 6; // local log-buffer flush watermark: everything at or below it is on disk
 }
 message ListMetadataSubscribersRequest {
     repeated string client_types = 1; // optional filter by client type, e.g. "mount"; empty = all
diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto
index 2dfd9caa0..eec385e4b 100644
--- a/seaweed-volume/proto/master.proto
+++ b/seaweed-volume/proto/master.proto
@@ -222,6 +222,7 @@ message VolumeLocation {
   uint32 grpc_port = 7;
   repeated uint32 new_ec_vids = 8;
   repeated uint32 deleted_ec_vids = 9;
+  repeated uint32 remote_vids = 10;
 }
 
 message ClusterNodeUpdate {
@@ -267,6 +268,7 @@ message Location {
   string public_url = 2;
   uint32 grpc_port = 3;
   string data_center = 4;
+  bool data_in_remote = 5;
 }
 
 message AssignRequest {
diff --git a/weed/filer/reader_at.go b/weed/filer/reader_at.go
index 55231a193..8ba08d489 100644
--- a/weed/filer/reader_at.go
+++ b/weed/filer/reader_at.go
@@ -117,9 +117,15 @@ func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionTyp
 
 		fcDataCenter := filerClient.GetDataCenter()
 		var sameDcTargetUrls, otherTargetUrls []string
+		localUrls := make(map[string]bool)
 		for _, loc := range locations.Locations {
 			volumeServerAddress := filerClient.AdjustedUrl(loc)
 			targetUrl := fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
+			glog.V(4).Infof("lookup %s => %s, data in remote storage tier: %v", fileId, targetUrl, loc.DataInRemote)
+
+			if !loc.DataInRemote {
+				localUrls[targetUrl] = true
+			}
 			if fcDataCenter == "" || fcDataCenter != loc.DataCenter {
 				otherTargetUrls = append(otherTargetUrls, targetUrl)
 			} else {
@@ -132,7 +138,13 @@ func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionTyp
 		rand.Shuffle(len(otherTargetUrls), func(i, j int) {
 			otherTargetUrls[i], otherTargetUrls[j] = otherTargetUrls[j], otherTargetUrls[i]
 		})
-		// Prefer same data center
+		// Local replicas go first inside each data center, but never ahead of
+		// the data-center preference itself. Matches the wdclient lookup paths
+		// so deprecated callers pick cheap reads first too.
+		if len(localUrls) > 0 {
+			sameDcTargetUrls = util.ReorderToFront(localUrls, sameDcTargetUrls)
+			otherTargetUrls = util.ReorderToFront(localUrls, otherTargetUrls)
+		}
 		targetUrls = append(sameDcTargetUrls, otherTargetUrls...)
 		return
 	}
diff --git a/weed/operation/lookup.go b/weed/operation/lookup.go
index eabea475e..5b00725f6 100644
--- a/weed/operation/lookup.go
+++ b/weed/operation/lookup.go
@@ -15,10 +15,11 @@ import (
 )
 
 type Location struct {
-	Url        string `json:"url,omitempty"`
-	PublicUrl  string `json:"publicUrl,omitempty"`
-	DataCenter string `json:"dataCenter,omitempty"`
-	GrpcPort   int    `json:"grpcPort,omitempty"`
+	Url          string `json:"url,omitempty"`
+	PublicUrl    string `json:"publicUrl,omitempty"`
+	DataCenter   string `json:"dataCenter,omitempty"`
+	GrpcPort     int    `json:"grpcPort,omitempty"`
+	DataInRemote bool   `json:"dataInRemote,omitempty"`
 }
 
 func (l *Location) ServerAddress() pb.ServerAddress {
@@ -41,6 +42,12 @@ var (
 	vc VidCache // caching of volume locations, re-check if after 10 minutes
 )
 
+// LookupFileId resolves a "<vid>,<cookie>" file id to one HTTP read URL,
+// preferring a volume server whose replica holds the data locally over one
+// backed by remote-tier storage. If no local replica is known the function
+// falls back to a random remote replica. The returned jwt is the read
+// authorization the master stamped on the volume; pass it through to the
+// volume server on the read request.
 func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrl string, jwt string, err error) {
 	parts := strings.Split(fileId, ",")
 	if len(parts) != 2 {
@@ -53,7 +60,20 @@ func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId s
 	if len(lookup.Locations) == 0 {
 		return "", jwt, errors.New("File Not Found")
 	}
-	return "http://" + lookup.Locations[rand.IntN(len(lookup.Locations))].Url + "/" + fileId, lookup.Jwt, nil
+
+	localUrls := make([]string, 0, len(lookup.Locations))
+	for _, loc := range lookup.Locations {
+		if !loc.DataInRemote {
+			localUrls = append(localUrls, loc.Url)
+		}
+	}
+	if len(localUrls) == 0 {
+		for _, loc := range lookup.Locations {
+			localUrls = append(localUrls, loc.Url)
+		}
+	}
+
+	return "http://" + localUrls[rand.IntN(len(localUrls))] + "/" + fileId, lookup.Jwt, nil
 }
 
 func LookupVolumeId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid string) (*LookupResult, error) {
@@ -102,10 +122,11 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 			var locations []Location
 			for _, loc := range vidLocations.Locations {
 				locations = append(locations, Location{
-					Url:        loc.Url,
-					PublicUrl:  loc.PublicUrl,
-					DataCenter: loc.DataCenter,
-					GrpcPort:   int(loc.GrpcPort),
+					Url:          loc.Url,
+					PublicUrl:    loc.PublicUrl,
+					DataCenter:   loc.DataCenter,
+					GrpcPort:     int(loc.GrpcPort),
+					DataInRemote: loc.DataInRemote,
 				})
 			}
 			if vidLocations.Error == "" {
diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto
index adef26743..dbd18dfaf 100644
--- a/weed/pb/filer.proto
+++ b/weed/pb/filer.proto
@@ -579,6 +579,7 @@ message Location {
     string public_url = 2;
     uint32 grpc_port = 3;
     string data_center = 4;
+    bool data_in_remote = 5;
 }
 message LookupVolumeResponse {
     map<string, Locations> locations_map = 1;
diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go
index b95a2cfe5..14e532d49 100644
--- a/weed/pb/filer_pb/filer.pb.go
+++ b/weed/pb/filer_pb/filer.pb.go
@@ -3433,6 +3433,7 @@ type Location struct {
 	PublicUrl     string                 `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
 	GrpcPort      uint32                 `protobuf:"varint,3,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
 	DataCenter    string                 `protobuf:"bytes,4,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
+	DataInRemote  bool                   `protobuf:"varint,5,opt,name=data_in_remote,json=dataInRemote,proto3" json:"data_in_remote,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -3495,6 +3496,13 @@ func (x *Location) GetDataCenter() string {
 	return ""
 }
 
+func (x *Location) GetDataInRemote() bool {
+	if x != nil {
+		return x.DataInRemote
+	}
+	return false
+}
+
 type LookupVolumeResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	LocationsMap  map[string]*Locations  `protobuf:"bytes,1,rep,name=locations_map,json=locationsMap,proto3" json:"locations_map,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
@@ -7287,14 +7295,15 @@ const file_filer_proto_rawDesc = "" +
 	"\n" +
 	"volume_ids\x18\x01 \x03(\tR\tvolumeIds\"=\n" +
 	"\tLocations\x120\n" +
-	"\tlocations\x18\x01 \x03(\v2\x12.filer_pb.LocationR\tlocations\"y\n" +
+	"\tlocations\x18\x01 \x03(\v2\x12.filer_pb.LocationR\tlocations\"\x9f\x01\n" +
 	"\bLocation\x12\x10\n" +
 	"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
 	"\n" +
 	"public_url\x18\x02 \x01(\tR\tpublicUrl\x12\x1b\n" +
 	"\tgrpc_port\x18\x03 \x01(\rR\bgrpcPort\x12\x1f\n" +
 	"\vdata_center\x18\x04 \x01(\tR\n" +
-	"dataCenter\"\xc3\x01\n" +
+	"dataCenter\x12$\n" +
+	"\x0edata_in_remote\x18\x05 \x01(\bR\fdataInRemote\"\xc3\x01\n" +
 	"\x14LookupVolumeResponse\x12U\n" +
 	"\rlocations_map\x18\x01 \x03(\v20.filer_pb.LookupVolumeResponse.LocationsMapEntryR\flocationsMap\x1aT\n" +
 	"\x11LocationsMapEntry\x12\x10\n" +
diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go
index 8ba5613e0..6d944c6b2 100644
--- a/weed/pb/filer_pb/filer_vtproto.pb.go
+++ b/weed/pb/filer_pb/filer_vtproto.pb.go
@@ -3136,6 +3136,16 @@ func (m *Location) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.DataInRemote {
+		i--
+		if m.DataInRemote {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x28
+	}
 	if len(m.DataCenter) > 0 {
 		i -= len(m.DataCenter)
 		copy(dAtA[i:], m.DataCenter)
@@ -7494,6 +7504,9 @@ func (m *Location) SizeVT() (n int) {
 	if l > 0 {
 		n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
 	}
+	if m.DataInRemote {
+		n += 2
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -17370,6 +17383,26 @@ func (m *Location) UnmarshalVT(dAtA []byte) error {
 			}
 			m.DataCenter = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
+		case 5:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field DataInRemote", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.DataInRemote = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])
diff --git a/weed/pb/master.proto b/weed/pb/master.proto
index 2dfd9caa0..eec385e4b 100644
--- a/weed/pb/master.proto
+++ b/weed/pb/master.proto
@@ -222,6 +222,7 @@ message VolumeLocation {
   uint32 grpc_port = 7;
   repeated uint32 new_ec_vids = 8;
   repeated uint32 deleted_ec_vids = 9;
+  repeated uint32 remote_vids = 10;
 }
 
 message ClusterNodeUpdate {
@@ -267,6 +268,7 @@ message Location {
   string public_url = 2;
   uint32 grpc_port = 3;
   string data_center = 4;
+  bool data_in_remote = 5;
 }
 
 message AssignRequest {
diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go
index c64f8931a..95e34060f 100644
--- a/weed/pb/master_pb/master.pb.go
+++ b/weed/pb/master_pb/master.pb.go
@@ -1061,6 +1061,7 @@ type VolumeLocation struct {
 	GrpcPort      uint32                 `protobuf:"varint,7,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
 	NewEcVids     []uint32               `protobuf:"varint,8,rep,packed,name=new_ec_vids,json=newEcVids,proto3" json:"new_ec_vids,omitempty"`
 	DeletedEcVids []uint32               `protobuf:"varint,9,rep,packed,name=deleted_ec_vids,json=deletedEcVids,proto3" json:"deleted_ec_vids,omitempty"`
+	RemoteVids    []uint32               `protobuf:"varint,10,rep,packed,name=remote_vids,json=remoteVids,proto3" json:"remote_vids,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1158,6 +1159,13 @@ func (x *VolumeLocation) GetDeletedEcVids() []uint32 {
 	return nil
 }
 
+func (x *VolumeLocation) GetRemoteVids() []uint32 {
+	if x != nil {
+		return x.RemoteVids
+	}
+	return nil
+}
+
 type ClusterNodeUpdate struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	NodeType      string                 `protobuf:"bytes,1,opt,name=node_type,json=nodeType,proto3" json:"node_type,omitempty"`
@@ -1460,6 +1468,7 @@ type Location struct {
 	PublicUrl     string                 `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
 	GrpcPort      uint32                 `protobuf:"varint,3,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
 	DataCenter    string                 `protobuf:"bytes,4,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
+	DataInRemote  bool                   `protobuf:"varint,5,opt,name=data_in_remote,json=dataInRemote,proto3" json:"data_in_remote,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1522,6 +1531,13 @@ func (x *Location) GetDataCenter() string {
 	return ""
 }
 
+func (x *Location) GetDataInRemote() bool {
+	if x != nil {
+		return x.DataInRemote
+	}
+	return false
+}
+
 type AssignRequest struct {
 	state               protoimpl.MessageState `protogen:"open.v1"`
 	Count               uint64                 `protobuf:"varint,1,opt,name=count,proto3" json:"count,omitempty"`
@@ -5072,7 +5088,7 @@ const file_master_proto_rawDesc = "" +
 	"filerGroup\x12\x1f\n" +
 	"\vdata_center\x18\x06 \x01(\tR\n" +
 	"dataCenter\x12\x12\n" +
-	"\x04rack\x18\a \x01(\tR\x04rack\"\x9d\x02\n" +
+	"\x04rack\x18\a \x01(\tR\x04rack\"\xbe\x02\n" +
 	"\x0eVolumeLocation\x12\x10\n" +
 	"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
 	"\n" +
@@ -5084,7 +5100,10 @@ const file_master_proto_rawDesc = "" +
 	"dataCenter\x12\x1b\n" +
 	"\tgrpc_port\x18\a \x01(\rR\bgrpcPort\x12\x1e\n" +
 	"\vnew_ec_vids\x18\b \x03(\rR\tnewEcVids\x12&\n" +
-	"\x0fdeleted_ec_vids\x18\t \x03(\rR\rdeletedEcVids\"\xa6\x01\n" +
+	"\x0fdeleted_ec_vids\x18\t \x03(\rR\rdeletedEcVids\x12\x1f\n" +
+	"\vremote_vids\x18\n" +
+	" \x03(\rR\n" +
+	"remoteVids\"\xa6\x01\n" +
 	"\x11ClusterNodeUpdate\x12\x1b\n" +
 	"\tnode_type\x18\x01 \x01(\tR\bnodeType\x12\x18\n" +
 	"\aaddress\x18\x02 \x01(\tR\aaddress\x12\x15\n" +
@@ -5112,14 +5131,15 @@ const file_master_proto_rawDesc = "" +
 	"\x11volume_or_file_id\x18\x01 \x01(\tR\x0evolumeOrFileId\x121\n" +
 	"\tlocations\x18\x02 \x03(\v2\x13.master_pb.LocationR\tlocations\x12\x14\n" +
 	"\x05error\x18\x03 \x01(\tR\x05error\x12\x12\n" +
-	"\x04auth\x18\x04 \x01(\tR\x04auth\"y\n" +
+	"\x04auth\x18\x04 \x01(\tR\x04auth\"\x9f\x01\n" +
 	"\bLocation\x12\x10\n" +
 	"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
 	"\n" +
 	"public_url\x18\x02 \x01(\tR\tpublicUrl\x12\x1b\n" +
 	"\tgrpc_port\x18\x03 \x01(\rR\bgrpcPort\x12\x1f\n" +
 	"\vdata_center\x18\x04 \x01(\tR\n" +
-	"dataCenter\"\xfe\x02\n" +
+	"dataCenter\x12$\n" +
+	"\x0edata_in_remote\x18\x05 \x01(\bR\fdataInRemote\"\xfe\x02\n" +
 	"\rAssignRequest\x12\x14\n" +
 	"\x05count\x18\x01 \x01(\x04R\x05count\x12 \n" +
 	"\vreplication\x18\x02 \x01(\tR\vreplication\x12\x1e\n" +
diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go
index 1679b1b4e..0168a9279 100644
--- a/weed/server/filer_grpc_server.go
+++ b/weed/server/filer_grpc_server.go
@@ -158,14 +158,19 @@ func (fs *FilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVol
 	return resp, err
 }
 
+// wdclientLocationsToPb converts the wdclient's internal Location entries
+// (carrying DataInRemote and grpc-port metadata) to the protobuf form served
+// by the filer gRPC API, preserving the fields the lookup client uses to
+// prefer a local replica over a remote-tiered one.
 func wdclientLocationsToPb(locations []wdclient.Location) []*filer_pb.Location {
 	locs := make([]*filer_pb.Location, 0, len(locations))
 	for _, loc := range locations {
 		locs = append(locs, &filer_pb.Location{
-			Url:        loc.Url,
-			PublicUrl:  loc.PublicUrl,
-			GrpcPort:   uint32(loc.GrpcPort),
-			DataCenter: loc.DataCenter,
+			Url:          loc.Url,
+			PublicUrl:    loc.PublicUrl,
+			GrpcPort:     uint32(loc.GrpcPort),
+			DataCenter:   loc.DataCenter,
+			DataInRemote: loc.DataInRemote,
 		})
 	}
 	return locs
diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go
index 15cac59f6..ac98b3359 100644
--- a/weed/server/master_grpc_server.go
+++ b/weed/server/master_grpc_server.go
@@ -12,6 +12,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster/maintenance"
+
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
@@ -83,6 +84,17 @@ func (ms *MasterServer) UnRegisterUuids(ip string, port int) {
 	glog.V(0).Infof("remove volume server %v, online volume server: %v", key, ms.Topo.UuidMap)
 }
 
+// announceVolume records vid on the broadcast message. A remote-tier volume
+// goes on both lists: RemoteVids carries the classification, and NewVids keeps
+// a client too old to read RemoteVids from losing the volume altogether during
+// a rolling upgrade.
+func announceVolume(message *master_pb.VolumeLocation, vid uint32, isRemote bool) {
+	message.NewVids = append(message.NewVids, vid)
+	if isRemote {
+		message.RemoteVids = append(message.RemoteVids, vid)
+	}
+}
+
 func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
 	var dn *topology.DataNode
 
@@ -233,7 +245,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 
 			// process delta volume ids if exists for fast volume id updates
 			for _, volInfo := range heartbeat.NewVolumes {
-				message.NewVids = append(message.NewVids, volInfo.Id)
+				// The short form carries no remote-storage name, so the volume
+				// reads as local until a changed or full report names its tier.
+				announceVolume(message, volInfo.Id, false)
 			}
 			for _, volInfo := range heartbeat.DeletedVolumes {
 				if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(volInfo.Id)) {
@@ -246,7 +260,10 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 		if len(heartbeat.ChangedVolumes) > 0 {
 			stats.MasterReceivedHeartbeatCounter.WithLabelValues("changedVolumes").Inc()
 			for _, v := range ms.Topo.ApplyVolumeChanges(heartbeat.ChangedVolumes, dn) {
-				message.NewVids = append(message.NewVids, uint32(v.Id))
+				// Changed volumes include both newly-added replicas and existing
+				// replicas whose tier classification flipped, which the client
+				// has to be told about to refresh its replica priority.
+				announceVolume(message, uint32(v.Id), v.IsRemote())
 			}
 		}
 
@@ -258,11 +275,19 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 
 			// process heartbeat.Volumes
 			stats.MasterReceivedHeartbeatCounter.WithLabelValues("Volumes").Inc()
-			newVolumes, deletedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)
+			newVolumes, deletedVolumes, changedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)
 
 			for _, v := range newVolumes {
 				glog.V(1).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
-				message.NewVids = append(message.NewVids, uint32(v.Id))
+				announceVolume(message, uint32(v.Id), v.IsRemote())
+			}
+			// A full reconciliation is the digest mismatch recovery path, and
+			// the only way a re-tiered replica reaches the master without a
+			// separate ChangedVolumes heartbeat. Announcing the changed set
+			// too is what stops the client keeping the old classification.
+			for _, v := range changedVolumes {
+				glog.V(1).Infof("master see tier/readonly change on volume %d from %s", uint32(v.Id), dn.Url())
+				announceVolume(message, uint32(v.Id), v.IsRemote())
 			}
 			for _, v := range deletedVolumes {
 				glog.V(1).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
diff --git a/weed/server/master_grpc_server_changed_volumes_test.go b/weed/server/master_grpc_server_changed_volumes_test.go
index 5ece14aa7..5436418fb 100644
--- a/weed/server/master_grpc_server_changed_volumes_test.go
+++ b/weed/server/master_grpc_server_changed_volumes_test.go
@@ -118,7 +118,7 @@ func TestRepairedLookupEntryIsAnnounced(t *testing.T) {
 			return topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{v}, dn)
 		}},
 		{"ViaFullList", func(topo *topology.Topology, dn *topology.DataNode, v *master_pb.VolumeInformationMessage) []storage.VolumeInfo {
-			announced, _ := topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, dn)
+			announced, _, _ := topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, dn)
 			return announced
 		}},
 	} {
diff --git a/weed/server/master_grpc_server_changed_volumes_tier_test.go b/weed/server/master_grpc_server_changed_volumes_tier_test.go
new file mode 100644
index 000000000..b411752a1
--- /dev/null
+++ b/weed/server/master_grpc_server_changed_volumes_tier_test.go
@@ -0,0 +1,157 @@
+package weed_server
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+// changedTierVolume returns a VolumeInformationMessage that mirrors
+// changedTestVolume plus a remote-storage name, so a heartbeat can flip a
+// replica's tier classification.
+func changedTierVolume(id uint32, size uint64, remoteStorageName string) *master_pb.VolumeInformationMessage {
+	v := changedTestVolume(id, size)
+	v.RemoteStorageName = remoteStorageName
+	return v
+}
+
+// announceChangedVolumes drives the message SendHeartbeat builds on a
+// ChangedVolumes heartbeat through announceVolume, the routing the server
+// itself uses, so the tests below assert what would really be broadcast
+// without standing up a gRPC stream.
+func announceChangedVolumes(topo *topology.Topology, dn *topology.DataNode, changed []*master_pb.VolumeInformationMessage) (newVids, remoteVids []uint32) {
+	message := &master_pb.VolumeLocation{}
+	for _, v := range topo.ApplyVolumeChanges(changed, dn) {
+		announceVolume(message, uint32(v.Id), v.IsRemote())
+	}
+	return message.NewVids, message.RemoteVids
+}
+
+// A replica that the heartbeat says has just been tiered to remote storage is
+// still servable from the same node, but every connected client is holding
+// stale DataInRemote=false and would prefer it over a real local replica. The
+// master must rebroadcast it on RemoteVids so the wdclient replaces the entry.
+func TestChangedVolumesAnnounceLocalToRemoteTierTransition(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024),
+	}, dn)
+
+	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
+		changedTierVolume(1, 1024, "s3-bucket"),
+	})
+	if !containsUint32(remoteVids, 1) {
+		t.Errorf("a local-to-remote transition was not announced as RemoteVids: %v", remoteVids)
+	}
+	if !containsUint32(newVids, 1) {
+		t.Errorf("a remote volume must stay on NewVids for clients that cannot read RemoteVids: %v", newVids)
+	}
+}
+
+// A replica restored from remote storage back to a local disk flips the other
+// way. The wdclient is currently demoting it behind same-DC remote replicas;
+// the master must announce it on NewVids so the wdclient hoists it back to
+// the front of the read order.
+func TestChangedVolumesAnnounceRemoteToLocalTierTransition(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTierVolume(1, 1024, "s3-bucket"),
+	}, dn)
+
+	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024),
+	})
+	if !containsUint32(newVids, 1) {
+		t.Errorf("a remote-to-local transition was not announced as NewVids: %v", newVids)
+	}
+	if containsUint32(remoteVids, 1) {
+		t.Errorf("a remote-to-local transition was routed as RemoteVids: %v", remoteVids)
+	}
+}
+
+// A heartbeat that re-reports an existing replica with the same tier
+// classification is not a change worth broadcasting. Volumes grow constantly
+// and a steady stream of those would flood every client's bounded queue.
+func TestChangedVolumesSuppressNoOpTierReport(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024),
+	}, dn)
+
+	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 4096),
+	})
+	if containsUint32(newVids, 1) || containsUint32(remoteVids, 1) {
+		t.Errorf("a no-op tier report was broadcast: newVids=%v remoteVids=%v", newVids, remoteVids)
+	}
+}
+
+// A heartbeat that mixes a pure growth with a tier transition only announces
+// the tier-transitioned replica: a growth is local state, not a re-route, and
+// must not push other topology updates out of a bounded client queue.
+func TestChangedVolumesAnnounceOnlyTierTransitions(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024),
+		changedTestVolume(2, 1024),
+	}, dn)
+
+	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 4096),              // pure growth
+		changedTierVolume(2, 1024, "s3-bucket"), // tier transition
+	})
+	if containsUint32(newVids, 1) || containsUint32(remoteVids, 1) {
+		t.Errorf("a volume that only grew was broadcast: newVids=%v remoteVids=%v", newVids, remoteVids)
+	}
+	if !containsUint32(remoteVids, 2) {
+		t.Errorf("the tier-transitioned replica was missing from RemoteVids: newVids=%v remoteVids=%v", newVids, remoteVids)
+	}
+	if _, err := dn.GetVolumesById(needle.VolumeId(1)); err != nil {
+		t.Errorf("a pure-growth heartbeat should not have removed the volume: %v", err)
+	}
+}
+
+func containsUint32(haystack []uint32, needle uint32) bool {
+	for _, v := range haystack {
+		if v == needle {
+			return true
+		}
+	}
+	return false
+}
+
+// A full Volumes reconciliation (the digest mismatch recovery path) is the
+// only way a re-tiered replica reaches the master without a separate
+// ChangedVolumes heartbeat. The routing in master_grpc_server.go has to
+// re-announce it on NewVids/RemoteVids or the wdclient keeps the stale
+// DataInRemote classification forever.
+func TestFullReconciliationAnnouncesTierTransition(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024),
+	}, dn)
+
+	newVids, remoteVids := announceFullReconciliation(topo, dn, []*master_pb.VolumeInformationMessage{
+		changedTierVolume(1, 1024, "s3-bucket"),
+	})
+	if !containsUint32(remoteVids, 1) {
+		t.Errorf("a tier transition reported in a full reconciliation was not announced: newVids=%v remoteVids=%v", newVids, remoteVids)
+	}
+	if !containsUint32(newVids, 1) {
+		t.Errorf("a remote volume must stay on NewVids for clients that cannot read RemoteVids: %v", newVids)
+	}
+}
+
+// announceFullReconciliation runs the same routing loop master_grpc_server's
+// SendHeartbeat does on a full Volumes heartbeat, including the changed-set
+// re-route added so digest-mismatch recovery propagates tier transitions.
+func announceFullReconciliation(topo *topology.Topology, dn *topology.DataNode, volumes []*master_pb.VolumeInformationMessage) (newVids, remoteVids []uint32) {
+	message := &master_pb.VolumeLocation{}
+	newOnes, _, changedOnes := topo.SyncDataNodeRegistration(volumes, dn)
+	for _, v := range append(newOnes, changedOnes...) {
+		announceVolume(message, uint32(v.Id), v.IsRemote())
+	}
+	return message.NewVids, message.RemoteVids
+}
diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go
index ef09c8f05..386b4dae4 100644
--- a/weed/server/master_grpc_server_volume.go
+++ b/weed/server/master_grpc_server_volume.go
@@ -151,6 +151,11 @@ func (ms *MasterServer) ProcessGrowRequest() {
 	}()
 }
 
+// LookupVolume resolves one or more volume ids (or "<vid>,<cookie>" file ids)
+// to their current replica locations on the volume servers. Each returned
+// entry carries DataInRemote per replica so the caller can prefer a local
+// replica; entries carrying a file id also receive a freshly generated read
+// jwt the volume server will accept.
 func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
 
 	resp := &master_pb.LookupVolumeResponse{}
@@ -167,10 +172,11 @@ func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupV
 			var locations []*master_pb.Location
 			for _, loc := range result.Locations {
 				locations = append(locations, &master_pb.Location{
-					Url:        loc.Url,
-					PublicUrl:  loc.PublicUrl,
-					DataCenter: loc.DataCenter,
-					GrpcPort:   uint32(loc.GrpcPort),
+					Url:          loc.Url,
+					PublicUrl:    loc.PublicUrl,
+					DataCenter:   loc.DataCenter,
+					GrpcPort:     uint32(loc.GrpcPort),
+					DataInRemote: loc.DataInRemote,
 				})
 			}
 			var auth string
diff --git a/weed/server/master_grpc_server_volume_move_test.go b/weed/server/master_grpc_server_volume_move_test.go
index 46ae5bbbd..5feb8b5bf 100644
--- a/weed/server/master_grpc_server_volume_move_test.go
+++ b/weed/server/master_grpc_server_volume_move_test.go
@@ -29,7 +29,7 @@ func TestVolumeMovedBetweenDisksIsNotBroadcastAsRemoved(t *testing.T) {
 	topo, dn := moveTestNode(t)
 	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moveTestVolume("", 0)}, dn)
 
-	_, deleted := topo.SyncDataNodeRegistration(
+	_, deleted, _ := topo.SyncDataNodeRegistration(
 		[]*master_pb.VolumeInformationMessage{moveTestVolume("ssd", 1)}, dn)
 
 	if len(deleted) != 1 {
@@ -44,7 +44,7 @@ func TestVolumeGoneFromTheNodeIsBroadcastAsRemoved(t *testing.T) {
 	topo, dn := moveTestNode(t)
 	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moveTestVolume("", 0)}, dn)
 
-	if _, deleted := topo.SyncDataNodeRegistration(nil, dn); len(deleted) != 1 {
+	if _, deleted, _ := topo.SyncDataNodeRegistration(nil, dn); len(deleted) != 1 {
 		t.Fatalf("expected the volume to be removed, got %d removals", len(deleted))
 	}
 	if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(1)) {
@@ -89,7 +89,7 @@ func TestVolumeReplacedByEcShardsIsBroadcastAsRemoved(t *testing.T) {
 		{Id: 1, Collection: "c", EcIndexBits: 0x3fff},
 	}, dn)
 
-	if _, deleted := topo.SyncDataNodeRegistration(nil, dn); len(deleted) != 1 {
+	if _, deleted, _ := topo.SyncDataNodeRegistration(nil, dn); len(deleted) != 1 {
 		t.Fatalf("expected the normal volume to be removed, got %d removals", len(deleted))
 	}
 	if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(1)) {
diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go
index f028a5a04..827f27e14 100644
--- a/weed/server/master_server_handlers.go
+++ b/weed/server/master_server_handlers.go
@@ -81,26 +81,22 @@ func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.Loo
 	if ms.Topo.IsLeader() {
 		volumeId, newVolumeIdErr := needle.NewVolumeId(vid)
 		if newVolumeIdErr != nil {
-			err = fmt.Errorf("Unknown volume id %s", vid)
+			err = fmt.Errorf("unknown volume id %s", vid)
 		} else {
 			machines := ms.Topo.Lookup(collection, volumeId)
 			for _, loc := range machines {
-				locations = append(locations, operation.Location{
-					Url:        loc.Url(),
-					PublicUrl:  loc.PublicUrl,
-					DataCenter: loc.GetDataCenterId(),
-					GrpcPort:   loc.GrpcPort,
-				})
+				locations = append(locations, topologyLocation(loc, volumeId))
 			}
 		}
 	} else {
 		machines, getVidLocationsErr := ms.MasterClient.GetVidLocations(vid)
 		for _, loc := range machines {
 			locations = append(locations, operation.Location{
-				Url:        loc.Url,
-				PublicUrl:  loc.PublicUrl,
-				DataCenter: loc.DataCenter,
-				GrpcPort:   loc.GrpcPort,
+				Url:          loc.Url,
+				PublicUrl:    loc.PublicUrl,
+				DataCenter:   loc.DataCenter,
+				GrpcPort:     loc.GrpcPort,
+				DataInRemote: loc.DataInRemote,
 			})
 		}
 		err = getVidLocationsErr
@@ -121,6 +117,23 @@ func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.Loo
 	return ret
 }
 
+// topologyLocation describes one node holding vid. A node that answers for an
+// EC volume holds shards rather than a volume record, so an absent record means
+// the read is local, never that the node should be left out of the answer.
+func topologyLocation(dn *topology.DataNode, vid needle.VolumeId) operation.Location {
+	dataInRemote := false
+	if volInfo, lookupErr := dn.GetVolumesById(vid); lookupErr == nil {
+		dataInRemote = volInfo.IsRemote()
+	}
+	return operation.Location{
+		Url:          dn.Url(),
+		PublicUrl:    dn.PublicUrl,
+		DataCenter:   dn.GetDataCenterId(),
+		GrpcPort:     dn.GrpcPort,
+		DataInRemote: dataInRemote,
+	}
+}
+
 func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) {
 	if ms.Topo.IsLeader() && ms.Topo.IsWarmingUp() {
 		remaining := ms.Topo.RemainingWarmupDuration()
diff --git a/weed/server/master_server_handlers_lookup_test.go b/weed/server/master_server_handlers_lookup_test.go
new file mode 100644
index 000000000..4ae0a910d
--- /dev/null
+++ b/weed/server/master_server_handlers_lookup_test.go
@@ -0,0 +1,48 @@
+package weed_server
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// The nodes answering for an EC volume hold shards, not a volume record, so
+// asking them for one fails. Dropping them would empty the lookup and turn
+// every EC read through the master into a 404.
+func TestEcVolumeLocationsSurviveTheLookup(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		{Id: 7, Collection: "c", EcIndexBits: 0x3fff},
+	}, dn)
+
+	machines := topo.Lookup("c", needle.VolumeId(7))
+	if len(machines) == 0 {
+		t.Fatalf("topology lost the EC volume")
+	}
+	for _, node := range machines {
+		loc := topologyLocation(node, needle.VolumeId(7))
+		if loc.Url == "" {
+			t.Errorf("EC location came back empty: %+v", loc)
+		}
+		if loc.DataInRemote {
+			t.Errorf("an EC shard holder was reported as remote-tier: %+v", loc)
+		}
+	}
+}
+
+// A volume the node really does hold keeps its tier classification.
+func TestVolumeLocationCarriesTheRemoteTier(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024),
+		changedTierVolume(2, 1024, "s3-bucket"),
+	}, dn)
+
+	if loc := topologyLocation(dn, needle.VolumeId(1)); loc.DataInRemote {
+		t.Errorf("a local volume was reported as remote-tier: %+v", loc)
+	}
+	if loc := topologyLocation(dn, needle.VolumeId(2)); !loc.DataInRemote {
+		t.Errorf("a tiered volume was reported as local: %+v", loc)
+	}
+}
diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go
index 78daf2276..e9f698f56 100644
--- a/weed/topology/data_node.go
+++ b/weed/topology/data_node.go
@@ -58,7 +58,7 @@ func (dn *DataNode) String() string {
 	return fmt.Sprintf("Node:%s, Ip:%s, Port:%d, PublicUrl:%s", dn.NodeImpl.String(), dn.Ip, dn.Port, dn.PublicUrl)
 }
 
-func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChangedRO bool) {
+func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChangedRO, tierTransition bool) {
 	dn.Lock()
 	defer dn.Unlock()
 	return dn.doAddOrUpdateVolume(v)
@@ -74,14 +74,14 @@ func (dn *DataNode) getOrCreateDisk(diskType string) *Disk {
 	return disk
 }
 
-func (dn *DataNode) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
+func (dn *DataNode) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged, tierTransition bool) {
 	disk := dn.getOrCreateDisk(v.DiskType)
 	return disk.AddOrUpdateVolume(v)
 }
 
 // AddProvisionalVolume records a volume the master registered on its own,
 // ahead of any server report naming it. See Disk.AddProvisionalVolume.
-func (dn *DataNode) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
+func (dn *DataNode) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged, tierTransition bool) {
 	dn.Lock()
 	defer dn.Unlock()
 	disk := dn.getOrCreateDisk(v.DiskType)
@@ -90,6 +90,14 @@ func (dn *DataNode) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged
 
 // UpdateVolumes detects new/deleted/changed volumes on a volume server
 // used in master to notify master clients of these changes.
+//
+// changedVolumes covers every replica the disk already held whose
+// classification the new report altered in a way clients must learn about:
+// the ReadOnly flag flipped, or IsRemote() flipped on tier transition. The
+// latter is what lets the wdclient refresh DataInRemote after the digest
+// mismatch recovery path resends a full Volumes list -- that path is the
+// only way a re-tiered replica reaches the master without a separate
+// ChangedVolumes heartbeat.
 func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolumes, deletedVolumes, changedVolumes []storage.VolumeInfo) {
 
 	reported := newReportedVolumes(len(actualVolumes))
@@ -133,11 +141,11 @@ func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolume
 		newVolumes = make([]storage.VolumeInfo, 0, addedCount)
 	}
 	for _, v := range actualVolumes {
-		isNew, isChanged := dn.doAddOrUpdateVolume(v)
+		isNew, isChanged, tierTransition := dn.doAddOrUpdateVolume(v)
 		if isNew {
 			newVolumes = append(newVolumes, v)
 		}
-		if isChanged {
+		if isChanged || tierTransition {
 			changedVolumes = append(changedVolumes, v)
 		}
 	}
@@ -219,15 +227,16 @@ func (dn *DataNode) AdjustDiskUsageBytes(diskTotalBytes, diskFreeBytes map[strin
 	}
 }
 
-// AppendVolumeIds appends the ids of this node's volumes to dst, without
-// copying the volume records to read them.
-func (dn *DataNode) AppendVolumeIds(dst []uint32) []uint32 {
+// AppendVolumeIds appends the ids of this node's volumes to all, and repeats
+// the remote-tier ones on remote, without copying the volume records to read
+// them.
+func (dn *DataNode) AppendVolumeIds(all, remote []uint32) ([]uint32, []uint32) {
 	dn.RLock()
 	defer dn.RUnlock()
 	for _, c := range dn.children {
-		dst = c.(*Disk).AppendVolumeIds(dst)
+		all, remote = c.(*Disk).AppendVolumeIds(all, remote)
 	}
-	return dst
+	return all, remote
 }
 
 func (dn *DataNode) GetVolumes() (ret []storage.VolumeInfo) {
diff --git a/weed/topology/disk.go b/weed/topology/disk.go
index 08ab99f63..a34697191 100644
--- a/weed/topology/disk.go
+++ b/weed/topology/disk.go
@@ -209,7 +209,7 @@ func (d *Disk) String() string {
 	return fmt.Sprintf("Disk:%s, volumes:%v, ecShards:%v", d.NodeImpl.String(), d.volumes, d.ecShards)
 }
 
-func (d *Disk) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
+func (d *Disk) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged, tierTransition bool) {
 	d.Lock()
 	defer d.Unlock()
 	return d.doAddOrUpdateVolume(v, true)
@@ -218,13 +218,26 @@ func (d *Disk) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
 // AddProvisionalVolume records a volume the master registered on its own --
 // volume growth -- before any server report has named it. Until one does, the
 // volume is protected from removal by a report that raced its creation.
-func (d *Disk) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
+func (d *Disk) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged, tierTransition bool) {
 	d.Lock()
 	defer d.Unlock()
 	return d.doAddOrUpdateVolume(v, false)
 }
 
-func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo, fromReport bool) (isNew, isChanged bool) {
+// doAddOrUpdateVolume returns three signals about how v was installed against
+// any existing record:
+//
+//   - isNew: no record was held before; the volume arrived.
+//   - isChanged: the ReadOnly flag flipped (the only field isChanged currently
+//     tracks). Other fields changing without ReadOnly flipping leaves this
+//     false.
+//   - tierTransition: v's IsRemote() classification differs from the previous
+//     record. The volume was already known and remains servable, but every
+//     connected client needs to refresh its replica priority -- a remote-tier
+//     replica restored locally must jump the read order, and one newly tiered
+//     to remote storage must give way. Callers that broadcast volume changes
+//     to clients must include tier-transitioned volumes alongside arrivals.
+func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo, fromReport bool) (isNew, isChanged, tierTransition bool) {
 	deltaDiskUsage := &DiskUsageCounts{}
 	if oldV, ok := d.volumes[v.Id]; !ok {
 		stored := v
@@ -253,7 +266,8 @@ func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo, fromReport bool) (isNew
 			// server keeps reporting.
 			v.DiskId = oldV.DiskId
 		}
-		if oldV.IsRemote() != v.IsRemote() {
+		tierTransition = oldV.IsRemote() != v.IsRemote()
+		if tierTransition {
 			if v.IsRemote() {
 				deltaDiskUsage.remoteVolumeCount = 1
 			}
@@ -291,16 +305,20 @@ func (d *Disk) GetVolumes() []storage.VolumeInfo {
 	return d.AppendVolumes(make([]storage.VolumeInfo, 0, d.VolumeCount()))
 }
 
-// AppendVolumeIds appends the ids of the disk's volumes to dst. Callers that
-// only need to name volumes use this rather than AppendVolumes, which copies
-// a whole record per volume to be read for four bytes of it.
-func (d *Disk) AppendVolumeIds(dst []uint32) []uint32 {
+// AppendVolumeIds appends the ids of the disk's volumes to all, and repeats
+// the remote-tier ones on remote. Callers that only need to name volumes use
+// this rather than AppendVolumes, which copies a whole record per volume to
+// be read for four bytes of it.
+func (d *Disk) AppendVolumeIds(all, remote []uint32) ([]uint32, []uint32) {
 	d.RLock()
 	defer d.RUnlock()
-	for id := range d.volumes {
-		dst = append(dst, uint32(id))
+	for id, v := range d.volumes {
+		all = append(all, uint32(id))
+		if v.IsRemote() {
+			remote = append(remote, uint32(id))
+		}
 	}
-	return dst
+	return all, remote
 }
 
 // AppendVolumes appends the disk's volumes to dst, so a caller gathering
diff --git a/weed/topology/topology.go b/weed/topology/topology.go
index 5155eaab0..f1c56a05e 100644
--- a/weed/topology/topology.go
+++ b/weed/topology/topology.go
@@ -630,7 +630,7 @@ func (t *Topology) ListDCAndRacks() (dcs map[NodeId][]NodeId) {
 	return dcs
 }
 
-func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes, deletedVolumes []storage.VolumeInfo) {
+func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes, deletedVolumes, changedVolumes []storage.VolumeInfo) {
 	// convert into in memory struct storage.VolumeInfo
 	volumeInfos := make([]storage.VolumeInfo, 0, len(volumes))
 	for _, v := range volumes {
@@ -641,7 +641,7 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 		}
 	}
 	// find out the delta volumes
-	newVolumes, deletedVolumes, _ = dn.UpdateVolumes(volumeInfos)
+	newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos)
 	for _, v := range newVolumes {
 		t.RegisterVolumeLayout(v, dn)
 	}
@@ -722,7 +722,10 @@ func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolume
 //
 // Most changes are a volume growing, which moves no location, so returning
 // only the arrivals keeps a busy cluster from telling every client about
-// volumes they can already reach.
+// volumes they can already reach. The arrival set also includes replicas whose
+// IsRemote() classification flipped on tier transition: the volume is still
+// servable from the same node, but every connected client has stale replica
+// priority and must be told to refresh.
 func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes []storage.VolumeInfo) {
 	volumeInfos := make([]storage.VolumeInfo, 0, len(changed))
 	for _, v := range changed {
@@ -735,7 +738,7 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 	}
 
 	for _, vi := range volumeInfos {
-		isNew, _ := dn.AddOrUpdateVolume(vi)
+		isNew, _, tierTransition := dn.AddOrUpdateVolume(vi)
 		if vi.ReplicaPlacement == nil {
 			if isNew {
 				newVolumes = append(newVolumes, vi)
@@ -751,7 +754,7 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 			// Dropped with its collection; the next lookup creates a fresh one.
 			vl = t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType))
 		}
-		if isNew || becameServable {
+		if isNew || becameServable || tierTransition {
 			newVolumes = append(newVolumes, vi)
 		}
 		vl.UpdateOversizedState(&vi, dn)
diff --git a/weed/topology/topology_info.go b/weed/topology/topology_info.go
index f8df207b0..c766f5655 100644
--- a/weed/topology/topology_info.go
+++ b/weed/topology/topology_info.go
@@ -87,6 +87,12 @@ func (t *Topology) ToVolumeMap() interface{} {
 	return m
 }
 
+// ToVolumeLocations snapshots every data node's volume set into a list of
+// per-node VolumeLocation messages. NewVids carries every volume; RemoteVids
+// repeats the remote-tier subset so clients can route reads to a local
+// replica first when one exists on any node. EC shards are flattened into
+// NewEcVids so each vid is reported once even when its shards live on
+// multiple disks of the same node.
 func (t *Topology) ToVolumeLocations() (volumeLocations []*master_pb.VolumeLocation) {
 	for _, c := range t.Children() {
 		dc := c.(*DataCenter)
@@ -100,7 +106,9 @@ func (t *Topology) ToVolumeLocations() (volumeLocations []*master_pb.VolumeLocat
 					DataCenter: dn.GetDataCenterId(),
 					GrpcPort:   uint32(dn.GrpcPort),
 				}
-				volumeLocation.NewVids = dn.AppendVolumeIds(nil)
+
+				volumeLocation.NewVids, volumeLocation.RemoteVids = dn.AppendVolumeIds(nil, nil)
+
 				// A single EC volume's shards can live on multiple disks of
 				// one DataNode, so GetEcShards returns per-(vid,disk) entries.
 				// Dedupe so the snapshot carries each vid once.
diff --git a/weed/util/slice.go b/weed/util/slice.go
index 6b60beec5..978992c71 100644
--- a/weed/util/slice.go
+++ b/weed/util/slice.go
@@ -14,3 +14,23 @@ func DrainChannel[T any](ch chan T, first T) []T {
 		}
 	}
 }
+
+// ReorderToFront returns a new slice with every element present in frontMap
+// pulled to the front while keeping the relative order seen in inputSlice
+// within each partition. Items not in frontMap keep their relative order
+// behind the moved-up items. Useful for prioritizing a subset of candidates
+// (e.g. local replicas) without disturbing the shuffle order of the rest.
+func ReorderToFront[T comparable](frontMap map[T]bool, inputSlice []T) []T {
+	var prioritized []T
+	var remaining []T
+
+	for _, item := range inputSlice {
+		if frontMap[item] {
+			prioritized = append(prioritized, item)
+		} else {
+			remaining = append(remaining, item)
+		}
+	}
+
+	return append(prioritized, remaining...)
+}
diff --git a/weed/util/slice_test.go b/weed/util/slice_test.go
new file mode 100644
index 000000000..d8338d809
--- /dev/null
+++ b/weed/util/slice_test.go
@@ -0,0 +1,33 @@
+package util
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestReorderToFront_StringSlice(t *testing.T) {
+	localUrls := map[string]bool{
+		"http://local1": true,
+		"http://local2": true,
+	}
+
+	sameDcTargetUrls := []string{
+		"http://remote1",
+		"http://local1",
+		"http://remote2",
+		"http://local2",
+	}
+
+	expected := []string{
+		"http://local1",
+		"http://local2",
+		"http://remote1",
+		"http://remote2",
+	}
+
+	result := ReorderToFront(localUrls, sameDcTargetUrls)
+
+	if !reflect.DeepEqual(result, expected) {
+		t.Errorf("ReorderToFront failed for strings. Got: %v, Expected: %v", result, expected)
+	}
+}
diff --git a/weed/wdclient/filer_client.go b/weed/wdclient/filer_client.go
index 65237fb3c..d4f150ddc 100644
--- a/weed/wdclient/filer_client.go
+++ b/weed/wdclient/filer_client.go
@@ -468,6 +468,7 @@ func (fc *FilerClient) GetLookupFileIdFunction() LookupFileIdFunctionType {
 
 		// Build URLs with publicUrl preference, and also prefer same DC
 		var sameDcUrls, otherDcUrls []string
+		localUrls := make(map[string]bool)
 		dataCenter := fc.GetDataCenter()
 		for _, loc := range locations {
 			url := loc.PublicUrl
@@ -475,6 +476,10 @@ func (fc *FilerClient) GetLookupFileIdFunction() LookupFileIdFunctionType {
 				url = loc.Url
 			}
 			httpUrl := "http://" + url + "/" + fileId
+			glog.V(4).Infof("lookup %s => %s, data in remote storage tier: %v", fileId, url, loc.DataInRemote)
+			if !loc.DataInRemote {
+				localUrls[httpUrl] = true
+			}
 			if dataCenter != "" && dataCenter == loc.DataCenter {
 				sameDcUrls = append(sameDcUrls, httpUrl)
 			} else {
@@ -484,7 +489,13 @@ func (fc *FilerClient) GetLookupFileIdFunction() LookupFileIdFunctionType {
 		// Shuffle to distribute load across volume servers
 		rand.Shuffle(len(sameDcUrls), func(i, j int) { sameDcUrls[i], sameDcUrls[j] = sameDcUrls[j], sameDcUrls[i] })
 		rand.Shuffle(len(otherDcUrls), func(i, j int) { otherDcUrls[i], otherDcUrls[j] = otherDcUrls[j], otherDcUrls[i] })
-		// Prefer same data center
+		// Local replicas go first inside each data center, but never ahead of
+		// the data-center preference itself. Mirrors
+		// vidMap.LookupVolumeServerUrl so all client lookup paths agree.
+		if len(localUrls) > 0 {
+			sameDcUrls = util.ReorderToFront(localUrls, sameDcUrls)
+			otherDcUrls = util.ReorderToFront(localUrls, otherDcUrls)
+		}
 		fullUrls = append(sameDcUrls, otherDcUrls...)
 		return fullUrls, nil
 	}
diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go
index 4fe8a1e2a..c7d0de00b 100644
--- a/weed/wdclient/masterclient.go
+++ b/weed/wdclient/masterclient.go
@@ -105,10 +105,11 @@ func (p *masterVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds []
 					var locations []Location
 					for _, masterLoc := range vidLoc.Locations {
 						loc := Location{
-							Url:        masterLoc.Url,
-							PublicUrl:  masterLoc.PublicUrl,
-							GrpcPort:   int(masterLoc.GrpcPort),
-							DataCenter: masterLoc.DataCenter,
+							Url:          masterLoc.Url,
+							PublicUrl:    masterLoc.PublicUrl,
+							GrpcPort:     int(masterLoc.GrpcPort),
+							DataCenter:   masterLoc.DataCenter,
+							DataInRemote: masterLoc.DataInRemote,
 						}
 						// Update cache with the location
 						p.masterClient.addLocation(uint32(vid), loc)
@@ -367,6 +368,12 @@ func addedVids(added, removed []uint32) map[uint32]struct{} {
 	return index
 }
 
+// updateVidMap applies a KeepConnectedResponse volume-location message to the
+// local vidMap. NewVids adds the entries; RemoteVids names the subset of them
+// backed by remote storage, which is added with DataInRemote so read paths can
+// prefer the cheap local replica. DeletedVids drops the named entry unless the
+// same message also added it back (volume moved between this server's disks).
+// EC vid changes go through the parallel addEcLocation / deleteEcLocation pair.
 func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
 	if resp.VolumeLocation.IsEmptyUrl() {
 		glog.V(0).Infof("updateVidMap ignore short heartbeat: %+v", resp)
@@ -380,10 +387,28 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
 		GrpcPort:   int(resp.VolumeLocation.GrpcPort),
 	}
 	stillOnServer := addedVids(resp.VolumeLocation.NewVids, resp.VolumeLocation.DeletedVids)
+	// RemoteVids repeats ids NewVids already carries, so the tier is settled
+	// before anything is written rather than adding each one twice.
+	var remoteVids map[uint32]struct{}
+	if len(resp.VolumeLocation.RemoteVids) > 0 {
+		remoteVids = make(map[uint32]struct{}, len(resp.VolumeLocation.RemoteVids))
+		for _, vid := range resp.VolumeLocation.RemoteVids {
+			remoteVids[vid] = struct{}{}
+		}
+	}
 	for _, newVid := range resp.VolumeLocation.NewVids {
+		if _, isRemote := remoteVids[newVid]; isRemote {
+			continue
+		}
 		glog.V(2).Infof("%s.%s: %s masterClient adds volume %d", mc.FilerGroup, mc.clientType, loc.Url, newVid)
 		mc.addLocation(newVid, loc)
 	}
+	for _, remoteVid := range resp.VolumeLocation.RemoteVids {
+		remoteLoc := loc
+		remoteLoc.DataInRemote = true
+		glog.V(2).Infof("%s.%s: %s masterClient adds remote volume %d", mc.FilerGroup, mc.clientType, remoteLoc.Url, remoteVid)
+		mc.addLocation(remoteVid, remoteLoc)
+	}
 	for _, deletedVid := range resp.VolumeLocation.DeletedVids {
 		if _, moved := stillOnServer[deletedVid]; moved {
 			continue
@@ -403,10 +428,12 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
 		glog.V(2).Infof("%s.%s: %s masterClient removes ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedEcVid)
 		mc.deleteEcLocation(deletedEcVid, loc)
 	}
-	glog.V(1).Infof("updateVidMap(%s) %s.%s: %s volume add: %d, del: %d, add ec: %d del ec: %d",
+	glog.V(1).Infof("updateVidMap(%s) %s.%s: %s volume add local: %d, remote: %d, del: %d, add ec: %d del ec: %d",
 		resp.VolumeLocation.DataCenter, mc.FilerGroup, mc.clientType, loc.Url,
-		len(resp.VolumeLocation.NewVids), len(resp.VolumeLocation.DeletedVids),
-		len(resp.VolumeLocation.NewEcVids), len(resp.VolumeLocation.DeletedEcVids))
+		len(resp.VolumeLocation.NewVids)-len(resp.VolumeLocation.RemoteVids),
+		len(resp.VolumeLocation.RemoteVids),
+		len(resp.VolumeLocation.DeletedVids), len(resp.VolumeLocation.NewEcVids),
+		len(resp.VolumeLocation.DeletedEcVids))
 }
 
 func (mc *MasterClient) WithClient(ctx context.Context, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
diff --git a/weed/wdclient/vid_map.go b/weed/wdclient/vid_map.go
index fdae7dfca..b2c2926b2 100644
--- a/weed/wdclient/vid_map.go
+++ b/weed/wdclient/vid_map.go
@@ -9,9 +9,9 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/seaweedfs/seaweedfs/weed/pb"
-
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 type HasLookupFileIdFunction interface {
@@ -21,10 +21,11 @@ type HasLookupFileIdFunction interface {
 type LookupFileIdFunctionType func(ctx context.Context, fileId string) (targetUrls []string, err error)
 
 type Location struct {
-	Url        string `json:"url,omitempty"`
-	PublicUrl  string `json:"publicUrl,omitempty"`
-	DataCenter string `json:"dataCenter,omitempty"`
-	GrpcPort   int    `json:"grpcPort,omitempty"`
+	Url          string `json:"url,omitempty"`
+	PublicUrl    string `json:"publicUrl,omitempty"`
+	DataCenter   string `json:"dataCenter,omitempty"`
+	GrpcPort     int    `json:"grpcPort,omitempty"`
+	DataInRemote bool   `json:"dataInRemote,omitempty"`
 }
 
 func (l Location) ServerAddress() pb.ServerAddress {
@@ -87,6 +88,10 @@ func (vc *vidMap) isSameDataCenter(loc *Location) bool {
 	return true
 }
 
+// LookupVolumeServerUrl returns the cached volume-server URLs for vid in
+// preference order: same-DC local, same-DC remote-tier, then the other data
+// centers on the same footing. Within each group the order is randomized so
+// load spreads across equivalent servers.
 func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrls []string, err error) {
 	id, err := strconv.Atoi(vid)
 	if err != nil {
@@ -99,7 +104,15 @@ func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrls []string, err er
 		return nil, fmt.Errorf("volume %d not found", id)
 	}
 	var sameDcServers, otherDcServers []string
+	localUrls := make(map[string]bool)
+
 	for _, loc := range locations {
+		glog.V(4).Infof("lookup %s => %s, data in remote storage tier: %v", vid, loc.Url, loc.DataInRemote)
+
+		if !loc.DataInRemote {
+			localUrls[loc.Url] = true
+		}
+
 		if vc.isSameDataCenter(&loc) {
 			sameDcServers = append(sameDcServers, loc.Url)
 		} else {
@@ -112,11 +125,20 @@ func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrls []string, err er
 	rand.Shuffle(len(otherDcServers), func(i, j int) {
 		otherDcServers[i], otherDcServers[j] = otherDcServers[j], otherDcServers[i]
 	})
-	// Prefer same data center
+	// Local replicas go first inside each data center, but never ahead of the
+	// data-center preference itself: a remote tier is often in the same region
+	// as the local replicas, so crossing a DC boundary to avoid it can cost
+	// more than the remote read it saves.
+	if len(localUrls) > 0 {
+		sameDcServers = util.ReorderToFront(localUrls, sameDcServers)
+		otherDcServers = util.ReorderToFront(localUrls, otherDcServers)
+	}
 	serverUrls = append(sameDcServers, otherDcServers...)
 	return
 }
 
+// LookupFileId resolves a "<vid>,<cookie>" file id to a list of HTTP read
+// URLs using the same DC-then-local ordering as LookupVolumeServerUrl.
 func (vc *vidMap) LookupFileId(ctx context.Context, fileId string) (fullUrls []string, err error) {
 	parts := strings.Split(fileId, ",")
 	if len(parts) != 2 {
@@ -132,6 +154,9 @@ func (vc *vidMap) LookupFileId(ctx context.Context, fileId string) (fullUrls []s
 	return
 }
 
+// GetVidLocations returns the cached Location entries for vid as a string,
+// for callers that need richer per-server fields than raw URLs (e.g.
+// DataInRemote, PublicUrl).
 func (vc *vidMap) GetVidLocations(vid string) (locations []Location, err error) {
 	id, err := strconv.Atoi(vid)
 	if err != nil {
@@ -145,6 +170,11 @@ func (vc *vidMap) GetVidLocations(vid string) (locations []Location, err error)
 	return nil, fmt.Errorf("volume id %s not found", vid)
 }
 
+// GetLocations returns the cached Location entries for vid as a uint32.
+// When both regular and EC entries are present, whichever was learned last
+// wins so a volume that switched between regular and EC encoding stops
+// answering from the stale copy. Returns found=false when nothing remains,
+// including when only an older-generation entry would otherwise apply.
 func (vc *vidMap) GetLocations(vid uint32) (locations []Location, found bool) {
 	vc.RLock()
 	defer vc.RUnlock()
@@ -232,6 +262,14 @@ func (vc *vidMap) addEcLocation(vid uint32, location Location) {
 // replaces what an earlier one held instead of merging with it: after a reset
 // the new master is the authority, so a volume that moved must not keep
 // answering with the server it moved off. Callers must hold the write lock.
+//
+// If the URL is already present and the remote/local classification matches,
+// the entry is left untouched (same replica, same view). When the
+// classification flips -- e.g. a volume tiered to remote storage, or a
+// remote-backed replica restored locally -- the entry is rebuilt so
+// subsequent lookups pick up the new DataInRemote. The server reference key
+// only depends on the URL/grpc port, so it stays stable across the flip and
+// the refcount does not need to move.
 func (vc *vidMap) addLocationToMap(vid2Locations map[uint32]*locationsEntry, vid uint32, location Location) {
 	entry, found := vid2Locations[vid]
 	if !found || entry.generation != vc.generation {
@@ -246,8 +284,18 @@ func (vc *vidMap) addLocationToMap(vid2Locations map[uint32]*locationsEntry, vid
 		return
 	}
 
-	for _, loc := range entry.locations {
+	for i, loc := range entry.locations {
 		if loc.Url == location.Url {
+			if loc.DataInRemote == location.DataInRemote {
+				return
+			}
+			// A reader holds the slice GetLocations handed it after the lock
+			// was dropped, so the replacement is copied rather than written
+			// into the array underneath it.
+			updated := make([]Location, len(entry.locations))
+			copy(updated, entry.locations)
+			updated[i] = location
+			entry.locations = updated
 			return
 		}
 	}
diff --git a/weed/wdclient/vid_map_remote_transition_test.go b/weed/wdclient/vid_map_remote_transition_test.go
new file mode 100644
index 000000000..f270b5b09
--- /dev/null
+++ b/weed/wdclient/vid_map_remote_transition_test.go
@@ -0,0 +1,122 @@
+package wdclient
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// When the master reports a tier transition (local ↔ remote) for an existing
+// replica on the same server URL, the cached DataInRemote must update. Otherwise
+// reads keep preferring a remote-backed replica or skip a newly-restored local
+// one. The server refcount must stay stable because the server key only depends
+// on the URL/grpc port.
+func TestAddLocationUpdatesDataInRemoteOnTransition(t *testing.T) {
+	vm := newVidMap("", DefaultVidMapCacheSize)
+	vid := uint32(11)
+	server := Location{Url: "10.0.0.1:8080", DataCenter: "dc1", GrpcPort: 18080}
+
+	// First seen as local.
+	vm.addLocation(vid, server)
+	locs, found := vm.GetLocations(vid)
+	if !found || len(locs) != 1 || locs[0].DataInRemote {
+		t.Fatalf("expected single local replica, got %+v", locs)
+	}
+
+	// Same URL flips to remote.
+	tiered := server
+	tiered.DataInRemote = true
+	vm.addLocation(vid, tiered)
+
+	locs, found = vm.GetLocations(vid)
+	if !found {
+		t.Fatalf("tier transition dropped the replica")
+	}
+	if len(locs) != 1 {
+		t.Fatalf("tier transition should replace in place, got %d entries: %+v", len(locs), locs)
+	}
+	if !locs[0].DataInRemote {
+		t.Errorf("DataInRemote did not flip to remote on tier-out: %+v", locs[0])
+	}
+	if locs[0].Url != server.Url || locs[0].DataCenter != server.DataCenter {
+		t.Errorf("replaced entry lost non-DataInRemote fields: %+v", locs[0])
+	}
+	if !vm.hasVolumeServer(pb.ServerAddress(server.Url)) {
+		t.Errorf("server ref should remain stable across DataInRemote flip")
+	}
+
+	// Restored locally: same URL, DataInRemote back to false.
+	restored := server
+	vm.addLocation(vid, restored)
+
+	locs, found = vm.GetLocations(vid)
+	if !found {
+		t.Fatalf("restore transition dropped the replica")
+	}
+	if len(locs) != 1 {
+		t.Fatalf("restore should replace in place, got %d entries: %+v", len(locs), locs)
+	}
+	if locs[0].DataInRemote {
+		t.Errorf("DataInRemote did not flip back to local on restore: %+v", locs[0])
+	}
+}
+
+// LookupVolumeServerUrl must reflect the latest DataInRemote so the local-first
+// ordering picks up newly-restored local replicas on the next read.
+func TestLookupVolumeServerUrlReflectsRemoteTransition(t *testing.T) {
+	vm := newVidMap("dc1", DefaultVidMapCacheSize)
+	vid := uint32(12)
+
+	remote := Location{Url: "10.0.0.1:8080", DataCenter: "dc1", DataInRemote: true}
+	vm.addLocation(vid, remote)
+
+	urls, err := vm.LookupVolumeServerUrl("12")
+	if err != nil {
+		t.Fatalf("lookup failed: %v", err)
+	}
+	if len(urls) != 1 || urls[0] != "10.0.0.1:8080" {
+		t.Fatalf("expected only the remote replica, got %v", urls)
+	}
+
+	// Tier restored: same URL, DataInRemote=false.
+	local := Location{Url: "10.0.0.1:8080", DataCenter: "dc1"}
+	vm.addLocation(vid, local)
+
+	urls, err = vm.LookupVolumeServerUrl("12")
+	if err != nil {
+		t.Fatalf("lookup after restore failed: %v", err)
+	}
+	if len(urls) != 1 || urls[0] != "10.0.0.1:8080" {
+		t.Fatalf("expected only the restored replica, got %v", urls)
+	}
+}
+
+// A tier flip must not write into the slice a concurrent lookup is still
+// walking: GetLocations hands out the entry's own slice and the caller reads
+// it after the lock is dropped.
+func TestTierFlipDoesNotRaceWithLookup(t *testing.T) {
+	vm := newVidMap("dc1", DefaultVidMapCacheSize)
+	vid := uint32(13)
+	vm.addLocation(vid, Location{Url: "10.0.0.1:8080", DataCenter: "dc1"})
+	vm.addLocation(vid, Location{Url: "10.0.0.2:8080", DataCenter: "dc1"})
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 2000; i++ {
+			vm.addLocation(vid, Location{Url: "10.0.0.1:8080", DataCenter: "dc1", DataInRemote: i%2 == 0})
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 2000; i++ {
+			if _, err := vm.LookupVolumeServerUrl("13"); err != nil {
+				t.Errorf("lookup failed mid-flip: %v", err)
+				return
+			}
+		}
+	}()
+	wg.Wait()
+}
diff --git a/weed/wdclient/vidmap_client.go b/weed/wdclient/vidmap_client.go
index f2fc9add9..631adf5d4 100644
--- a/weed/wdclient/vidmap_client.go
+++ b/weed/wdclient/vidmap_client.go
@@ -13,6 +13,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 // VolumeLocationProvider is the interface for looking up volume locations
@@ -51,7 +52,13 @@ func (vc *vidMapClient) GetLookupFileIdFunction() LookupFileIdFunctionType {
 	return vc.LookupFileIdWithFallback
 }
 
-// LookupFileIdWithFallback looks up a file ID, checking cache first, then using provider
+// LookupFileIdWithFallback resolves a "<vid>,<cookie>" file id to a list of
+// HTTP read URLs, using the cached vidMap when populated and falling back to
+// the provider for a fresh lookup on miss. URLs are returned in preference
+// order: same-DC local, same-DC remote-tier, then the other data centers on
+// the same footing -- mirroring the cached vidMap path so both routes agree.
+// Concurrent misses for the same vid are coalesced via the singleflight group
+// on LookupVolumeIdsWithFallback.
 func (vc *vidMapClient) LookupFileIdWithFallback(ctx context.Context, fileId string) (fullUrls []string, err error) {
 	// Try cache first
 	dataCenter := vc.vidMap.DataCenter
@@ -90,8 +97,13 @@ func (vc *vidMapClient) LookupFileIdWithFallback(ctx context.Context, fileId str
 
 	// Build HTTP URLs from locations, preferring same data center
 	var sameDcUrls, otherDcUrls []string
+	localUrls := make(map[string]bool)
 	for _, loc := range locations {
 		httpUrl := "http://" + loc.Url + "/" + fileId
+		glog.V(4).Infof("lookup %s => %s, data in remote storage tier: %v", fileId, loc.Url, loc.DataInRemote)
+		if !loc.DataInRemote {
+			localUrls[httpUrl] = true
+		}
 		if dataCenter != "" && dataCenter == loc.DataCenter {
 			sameDcUrls = append(sameDcUrls, httpUrl)
 		} else {
@@ -103,7 +115,13 @@ func (vc *vidMapClient) LookupFileIdWithFallback(ctx context.Context, fileId str
 	rand.Shuffle(len(sameDcUrls), func(i, j int) { sameDcUrls[i], sameDcUrls[j] = sameDcUrls[j], sameDcUrls[i] })
 	rand.Shuffle(len(otherDcUrls), func(i, j int) { otherDcUrls[i], otherDcUrls[j] = otherDcUrls[j], otherDcUrls[i] })
 
-	// Prefer same data center
+	// Local replicas go first inside each data center, but never ahead of the
+	// data-center preference itself. Mirrors vidMap.LookupVolumeServerUrl so
+	// all client lookup paths agree.
+	if len(localUrls) > 0 {
+		sameDcUrls = util.ReorderToFront(localUrls, sameDcUrls)
+		otherDcUrls = util.ReorderToFront(localUrls, otherDcUrls)
+	}
 	fullUrls = append(sameDcUrls, otherDcUrls...)
 	return fullUrls, nil
 }
diff --git a/weed/wdclient/vidmap_client_localfirst_test.go b/weed/wdclient/vidmap_client_localfirst_test.go
new file mode 100644
index 000000000..931fe5e4b
--- /dev/null
+++ b/weed/wdclient/vidmap_client_localfirst_test.go
@@ -0,0 +1,113 @@
+package wdclient
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+type testLocationProvider struct {
+	locations map[string][]Location
+}
+
+func (p *testLocationProvider) LookupVolumeIds(ctx context.Context, volumeIds []string) (map[string][]Location, error) {
+	result := make(map[string][]Location)
+	for _, vid := range volumeIds {
+		if locs, found := p.locations[vid]; found {
+			result[vid] = locs
+		}
+	}
+	return result, nil
+}
+
+// TestLookupFileIdWithFallbackLocalFirst ensures volumes whose data is still
+// local are tried before remote-tier replicas, on the provider (cache miss) path.
+func TestLookupFileIdWithFallbackLocalFirst(t *testing.T) {
+	vc := newVidMapClient(&testLocationProvider{
+		locations: map[string][]Location{
+			"5": {
+				{Url: "10.0.0.1:8080", DataInRemote: true},
+				{Url: "10.0.0.2:8080", DataInRemote: false},
+			},
+		},
+	}, "", 5)
+
+	urls, err := vc.LookupFileIdWithFallback(context.Background(), "5,abcdef0123456789")
+	if err != nil {
+		t.Fatalf("lookup failed: %v", err)
+	}
+	if len(urls) != 2 {
+		t.Fatalf("expected 2 urls, got %v", urls)
+	}
+	hasLocal, hasRemote := false, false
+	for _, u := range urls {
+		if strings.Contains(u, "10.0.0.2:8080") {
+			hasLocal = true
+		}
+		if strings.Contains(u, "10.0.0.1:8080") {
+			hasRemote = true
+		}
+	}
+	if !hasLocal {
+		t.Errorf("local replica missing from result: %v", urls)
+	}
+	if !hasRemote {
+		t.Errorf("remote replica missing from result: %v", urls)
+	}
+	if !strings.Contains(urls[0], "10.0.0.2:8080") {
+		t.Errorf("expected local replica first, got %v", urls)
+	}
+}
+
+// TestLookupFileIdWithFallbackAllRemote keeps shuffled order when every replica is remote.
+func TestLookupFileIdWithFallbackAllRemote(t *testing.T) {
+	vc := newVidMapClient(&testLocationProvider{
+		locations: map[string][]Location{
+			"6": {
+				{Url: "10.0.0.1:8080", DataInRemote: true},
+				{Url: "10.0.0.2:8080", DataInRemote: true},
+			},
+		},
+	}, "", 5)
+
+	urls, err := vc.LookupFileIdWithFallback(context.Background(), "6,abcdef0123456789")
+	if err != nil {
+		t.Fatalf("lookup failed: %v", err)
+	}
+	if len(urls) != 2 {
+		t.Fatalf("expected 2 urls, got %v", urls)
+	}
+}
+
+// TestLookupFileIdWithFallbackKeepsDataCenterFirst verifies that the local-first
+// ordering applies inside each data center and does not override the data-center
+// preference: a same-DC remote replica still beats an other-DC local one, because
+// the remote tier is usually nearer than another data center.
+func TestLookupFileIdWithFallbackKeepsDataCenterFirst(t *testing.T) {
+	vc := newVidMapClient(&testLocationProvider{
+		locations: map[string][]Location{
+			"7": {
+				{Url: "10.0.0.1:8080", DataCenter: "dc1", DataInRemote: true},
+				{Url: "10.0.0.2:8080", DataCenter: "dc1", DataInRemote: false},
+				{Url: "10.0.0.3:8080", DataCenter: "dc2", DataInRemote: false},
+				{Url: "10.0.0.4:8080", DataCenter: "dc2", DataInRemote: true},
+			},
+		},
+	}, "dc1", 5)
+
+	urls, err := vc.LookupFileIdWithFallback(context.Background(), "7,abcdef0123456789")
+	if err != nil {
+		t.Fatalf("lookup failed: %v", err)
+	}
+	if len(urls) != 4 {
+		t.Fatalf("expected 4 urls, got %v", urls)
+	}
+
+	// dc1 local, dc1 remote, dc2 local, dc2 remote.
+	want := []string{"10.0.0.2:8080", "10.0.0.1:8080", "10.0.0.3:8080", "10.0.0.4:8080"}
+	for i, host := range want {
+		if !strings.Contains(urls[i], host) {
+			t.Fatalf("position %d should be %s, got %v", i, host, urls)
+		}
+	}
+}