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.

weed/server/master_grpc_server.go

@@ -247,7 +247,15 @@ 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 remote/local classification flipped on tier
+				// transition. Routed the same way as newVolumes so the client
+				// receives the updated DataInRemote through NewVids/RemoteVids.
+				if v.IsRemote() {
+					message.RemoteVids = append(message.RemoteVids, uint32(v.Id))
+				} else {
+					message.NewVids = append(message.NewVids, uint32(v.Id))
+				}
 			}
 		}
 

weed/wdclient/vid_map.go

@@ -245,6 +245,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 existing entry is replaced
+// in place 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 {
@@ -259,8 +267,12 @@ 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
+			}
+			entry.locations[i] = location
 			return
 		}
 	}

weed/wdclient/vid_map_remote_transition_test.go

@@ -0,0 +1,92 @@
+package wdclient
+
+import (
+	"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)
+	}
+}