expose whether a volume replica is backed by remote storage, and prefer local replicas (#11105)

* expose whether a volume replica is backed by remote storage

Volume locations returned by lookups do not indicate whether a replica
has been tiered to remote storage. Readers cannot distinguish a local
replica from a remote-backed one, so they may hit a remote-backed
replica first even when a local replica is available.

Add DataInRemote to the lookup location message, populate it from the
master's volume info, and carry it through the wdclient vid map so
clients can prefer local replicas when resolving chunk locations.

* wdclient: prefer local volume replicas over remote-tier replicas on lookup

LookupFileIdWithFallback (and the publicUrl variant in FilerClient)
didn't honor the DataInRemote flag when shuffling URLs, so the
DataInRemote patch only took effect in LookupVolumeServerUrl. Apply
the same ReorderToFront(localUrls) to sameDcUrls/otherDcUrls so
non-remote replicas stay at the front, matching the existing vidMap
convention.

* wdclient: propagate DataInRemote across tier transitions on existing replicas

When a volume is tiered to remote storage or a remote-backed replica is
restored locally, the cached DataInRemote on the same volume-server URL
stayed at its old value because two pieces of state never updated:

* master_grpc_server.go only split newVolumes and (already-tracked) volumes
  into NewVids vs RemoteVids. ChangedVolumes went straight to NewVids, so
  the broadcast announced the re-classified volume as a fresh arrival and
  the client had no way to tell whether its existing cache was stale.

* vid_map.addLocationToMap early-returned when an entry already had the
  same URL. A tier transition reports the same URL with DataInRemote
  flipped, so the cached entry stayed at the old classification.

Wire both sides together: ChangedVolumes now go through the same IsRemote
split as newVolumes, and addLocationToMap replaces the existing entry in
place when the URL matches but DataInRemote has changed. The server
reference key only depends on URL/grpc port, so the refcount does not
move across the flip.

Adds vid_map_remote_transition_test.go covering the local->remote and
remote->local paths so the in-place update and the cache-key stability
are pinned by tests.

* wdclient: prefer local replicas across data-center boundaries

The previous local-first ordering hoisted local URLs to the front of each
data-center bucket separately, then concatenated same-DC before other-DC.
That meant a same-DC remote replica could still be tried before an
other-DC local replica even though the local one would answer cheaply.

Reorder once across the full candidate list: concatenate same-DC and
other-DC first, then ReorderToFront pulls every local replica to the very
front while preserving the DC preference inside each tier. Apply the same
ordering in all four lookup paths so the cached vidMap, the
LookupFileIdWithFallback provider path, FilerClient.GetLookupFileIdFunction
(PublicUrl-preferred variant), and the deprecated filer.LookupFn all agree:
- weed/wdclient/vid_map.go (LookupVolumeServerUrl)
- weed/wdclient/vidmap_client.go (LookupFileIdWithFallback)
- weed/wdclient/filer_client.go (LookupFileId)
- weed/filer/reader_at.go (LookupFn)

Strengthen the existing local-first tests: vidmap_client_localfirst_test
now asserts both endpoints are present (not just the local one is first),
and slice_test asserts an exact match instead of accepting two orderings.

Add TestLookupFileIdWithFallbackGlobalLocalFirst to pin the cross-DC
ordering invariant: any local replica (same or other DC) precedes every
remote-tier replica; within each tier DC1 precedes DC2.

Add docstrings to ToVolumeLocations, ReorderToFront, LookupVolumeServerUrl,
LookupFileId, GetVidLocations, GetLocations, LookupFileIdWithFallback, and
updateVidMap so the touched lookup paths are described in one place.

* topology: broadcast tier transitions on existing replicas

When a volume replica is tiered to remote storage or restored locally, the
wdclient's cached DataInRemote went stale: every connected client kept
preferring a remote-backed replica over a freshly restored local one, or
demoted a freshly tiered remote replica. The fix in commit 116982595 routed
ChangedVolumes to NewVids/RemoteVids on the master, but ApplyVolumeChanges
returned only fresh arrivals and previously servable replicas. An existing
replica whose IsRemote() classification flipped was neither, so it never
reached the broadcast loop and the wdclient never learned.

Make Disk.doAddOrUpdateVolume return a third signal -- tierTransition --
true exactly when an existing replica's IsRemote() flips. ApplyVolumeChanges
treats that as an arrival so the existing SendHeartbeat routing loop now
sees it. Add a master-side end-to-end test covering local->remote,
remote->local, no-op re-reports, and a mixed heartbeat that only announces
the tier transition.

Also add docstrings to LookupFileId, wdclientLocationsToPb, and
LookupVolume where the prior change touched their bodies.

* topology: broadcast tier transitions received through full reconciliation

The previous commit added tier-transition routing on the ChangedVolumes
delta path, but that is not the only way a re-tiered replica reaches the
master. After a digest mismatch the volume server resends a full Volumes
list, and SyncDataNodeRegistration applies the new IsRemote() classification
silently -- the changedVolumes return value was being thrown away. The
master therefore never broadcast NewVids/RemoteVids, and a wdclient connected
during the recovery kept the stale DataInRemote until it lost contact with
the master.

Surface the changed set through UpdateVolumes.changedVolumes (now covering
both ReadOnly flips and tier flips) and SyncDataNodeRegistration, then route
it through NewVids/RemoteVids in SendHeartbeat the same way the delta path
already does. Add an end-to-end test for the full reconciliation path.

* master: keep an EC volume's locations in the volume lookup

The nodes that answer for an EC volume hold shards, not a volume record,
so asking them for one fails. Dropping the location on that failure
emptied the result and turned every EC read through the master's HTTP
lookup and fid redirect into a 404.

Treat an absent volume record as a local read and keep the node in the
answer. The per-node conversion moves into topologyLocation so the EC
case is covered by a test.

Claude-Session: https://claude.ai/code/session_01FcSp6quCaxQ1cb3Vx7o9fW

* wdclient: replace a tier-flipped location without writing under a reader

GetLocations hands back the entry's own slice and the caller walks it
after the read lock is dropped, which is why every other mutation here
builds a new slice. Writing the flipped replica into the array in place
raced LookupVolumeServerUrl, reported by -race.

Copy the slice, swap the one element, and publish it.

Claude-Session: https://claude.ai/code/session_01FcSp6quCaxQ1cb3Vx7o9fW

* master: keep a remote volume on NewVids for older clients

Moving remote-tier volumes out of NewVids and into RemoteVids alone is a
wire break in the wrong direction. A master upgraded ahead of its filers
and mounts -- the usual order -- announces a tiered volume only on a
field the older client ignores, so the volume drops out of that client's
vid map entirely and reads for it fail.

Announce every volume on NewVids and repeat the remote-tier subset on
RemoteVids, so a new client still learns the tier and an old one keeps
the location. The routing moves into announceVolume, which the heartbeat
paths and their tests now share instead of each restating it.

On the client, RemoteVids no longer needs a second write per volume: the
tier is settled before anything is added.

Claude-Session: https://claude.ai/code/session_01FcSp6quCaxQ1cb3Vx7o9fW

* topology: split the volume snapshot by tier without copying the records

ToVolumeLocations runs on every KeepConnected, so a filer or mount
connecting made the master allocate a full VolumeInfo per volume per node
just to read four bytes of id off each one. AppendVolumeIds exists to
avoid exactly that.

Extend it to fill the remote-tier list alongside the full one, and use it
again in the snapshot.

Claude-Session: https://claude.ai/code/session_01FcSp6quCaxQ1cb3Vx7o9fW

* wdclient: keep the data-center preference ahead of the local-first ordering

Hoisting every local replica to the very front puts an other-DC local
read ahead of a same-DC remote one. When the remote tier sits in the same
region as the replicas -- the common arrangement -- that trades an
in-region GET for a WAN round trip and costs more than the remote read it
avoids.

Reorder inside each data-center bucket instead, so local still wins among
equals and the data-center preference still wins overall.

Claude-Session: https://claude.ai/code/session_01FcSp6quCaxQ1cb3Vx7o9fW

* operation: pick the read replica from one list

The local-preferring lookup built a list of local URLs and then branched
on whether it was empty, duplicating the random pick. Fall back by
filling the same list with every replica instead.

Claude-Session: https://claude.ai/code/session_01FcSp6quCaxQ1cb3Vx7o9fW

---------

Co-authored-by: Bruce Zou <gift_secondst@msn.com>
Co-authored-by: bruce-zzz <bruce.zou@hhy-data.com>

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

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 {

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
 	}

weed/operation/lookup.go

@@ -15,10 +15,11 @@ import (
 )
 
 type Location struct {
-	Url        string `json:"url,omitempty"`
+	Url          string `json:"url,omitempty"`
-	PublicUrl  string `json:"publicUrl,omitempty"`
+	PublicUrl    string `json:"publicUrl,omitempty"`
-	DataCenter string `json:"dataCenter,omitempty"`
+	DataCenter   string `json:"dataCenter,omitempty"`
-	GrpcPort   int    `json:"grpcPort,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,
+					Url:          loc.Url,
-					PublicUrl:  loc.PublicUrl,
+					PublicUrl:    loc.PublicUrl,
-					DataCenter: loc.DataCenter,
+					DataCenter:   loc.DataCenter,
-					GrpcPort:   int(loc.GrpcPort),
+					GrpcPort:     int(loc.GrpcPort),
+					DataInRemote: loc.DataInRemote,
 				})
 			}
 			if vidLocations.Error == "" {

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;

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" +

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:])

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 {

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" +

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,
+			Url:          loc.Url,
-			PublicUrl:  loc.PublicUrl,
+			PublicUrl:    loc.PublicUrl,
-			GrpcPort:   uint32(loc.GrpcPort),
+			GrpcPort:     uint32(loc.GrpcPort),
-			DataCenter: loc.DataCenter,
+			DataCenter:   loc.DataCenter,
+			DataInRemote: loc.DataInRemote,
 		})
 	}
 	return locs

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())

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
 		}},
 	} {

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
+}

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,
+					Url:          loc.Url,
-					PublicUrl:  loc.PublicUrl,
+					PublicUrl:    loc.PublicUrl,
-					DataCenter: loc.DataCenter,
+					DataCenter:   loc.DataCenter,
-					GrpcPort:   uint32(loc.GrpcPort),
+					GrpcPort:     uint32(loc.GrpcPort),
+					DataInRemote: loc.DataInRemote,
 				})
 			}
 			var auth string

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)) {

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{
+				locations = append(locations, topologyLocation(loc, volumeId))
-					Url:        loc.Url(),
-					PublicUrl:  loc.PublicUrl,
-					DataCenter: loc.GetDataCenterId(),
-					GrpcPort:   loc.GrpcPort,
-				})
 			}
 		}
 	} else {
 		machines, getVidLocationsErr := ms.MasterClient.GetVidLocations(vid)
 		for _, loc := range machines {
 			locations = append(locations, operation.Location{
-				Url:        loc.Url,
+				Url:          loc.Url,
-				PublicUrl:  loc.PublicUrl,
+				PublicUrl:    loc.PublicUrl,
-				DataCenter: loc.DataCenter,
+				DataCenter:   loc.DataCenter,
-				GrpcPort:   loc.GrpcPort,
+				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()

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)
+	}
+}

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
+// AppendVolumeIds appends the ids of this node's volumes to all, and repeats
-// copying the volume records to read them.
+// the remote-tier ones on remote, without copying the volume records to read
-func (dn *DataNode) AppendVolumeIds(dst []uint32) []uint32 {
+// 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) {

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
+// AppendVolumeIds appends the ids of the disk's volumes to all, and repeats
-// only need to name volumes use this rather than AppendVolumes, which copies
+// the remote-tier ones on remote. Callers that only need to name volumes use
-// a whole record per volume to be read for four bytes of it.
+// this rather than AppendVolumes, which copies a whole record per volume to
-func (d *Disk) AppendVolumeIds(dst []uint32) []uint32 {
+// 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 {
+	for id, v := range d.volumes {
-		dst = append(dst, uint32(id))
+		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

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)

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.

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...)
+}

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)
+	}
+}

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
 	}

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,
+							Url:          masterLoc.Url,
-							PublicUrl:  masterLoc.PublicUrl,
+							PublicUrl:    masterLoc.PublicUrl,
-							GrpcPort:   int(masterLoc.GrpcPort),
+							GrpcPort:     int(masterLoc.GrpcPort),
-							DataCenter: masterLoc.DataCenter,
+							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.NewVids)-len(resp.VolumeLocation.RemoteVids),
-		len(resp.VolumeLocation.NewEcVids), len(resp.VolumeLocation.DeletedEcVids))
+		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 {

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"`
+	Url          string `json:"url,omitempty"`
-	PublicUrl  string `json:"publicUrl,omitempty"`
+	PublicUrl    string `json:"publicUrl,omitempty"`
-	DataCenter string `json:"dataCenter,omitempty"`
+	DataCenter   string `json:"dataCenter,omitempty"`
-	GrpcPort   int    `json:"grpcPort,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
 		}
 	}

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()
+}

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
 }

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)
+		}
+	}
+}