shell: keep the remote object when a server lists the volume twice

A stale twin on another disk of a replaced server still references the
object but is filtered out with its server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

weed/shell/command_volume_tier_download.go

@@ -215,6 +215,10 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i
 				fmt.Fprintf(writer, "volume %d: only %d of %d replicas are listed, the remote object is kept\n", vid, listed, copies)
 				return nil
 			}
+			if server, found := serverListedTwice(remote, local); found {
+				fmt.Fprintf(writer, "volume %d is listed twice on %s, the remote object is kept\n", vid, server)
+				return nil
+			}
 			last := pending[0]
 			fmt.Fprintf(writer, "delete remote-backed replica of volume %d on %s with its remote object\n", vid, last.server)
 			if err = deleteVolume(context.Background(), grpcDialOption, vid, last.server, false, false, false); err != nil {
@@ -284,6 +288,17 @@ func replicateTierDownloadedVolume(grpcDialOption grpc.DialOption, writer io.Wri
 	return nil
 }
 
+func serverListedTwice(remote, local []tierDownloadReplica) (pb.ServerAddress, bool) {
+	seen := make(map[pb.ServerAddress]bool)
+	for _, replica := range append(append([]tierDownloadReplica{}, remote...), local...) {
+		if seen[replica.server] {
+			return replica.server, true
+		}
+		seen[replica.server] = true
+	}
+	return "", false
+}
+
 type tierDownloadReplica struct {
 	server   pb.ServerAddress
 	diskType string

weed/shell/command_volume_tier_download_test.go

@@ -42,6 +42,8 @@ type tierDownloadCluster struct {
 	writable     map[pb.ServerAddress]bool
 	remounts     map[pb.ServerAddress]int
 	placement    uint32
+	// staleTwin lists a remote-backed twin of the volume on another disk
+	staleTwin map[pb.ServerAddress]bool
 }
 
 type tierDownloadTestServer struct {
@@ -162,6 +164,7 @@ func newTierDownloadCluster(t *testing.T, states ...string) *tierDownloadCluster
 		frozen:       make(map[pb.ServerAddress]bool),
 		writable:     make(map[pb.ServerAddress]bool),
 		remounts:     make(map[pb.ServerAddress]int),
+		staleTwin:    make(map[pb.ServerAddress]bool),
 	}
 	for _, state := range states {
 		lis, err := net.Listen("tcp", "127.0.0.1:0")
@@ -209,10 +212,16 @@ func (c *tierDownloadCluster) readTopology() (*master_pb.TopologyInfo, error) {
 			vi.RemoteStorageName = "s3.default"
 			vi.RemoteStorageKey = "key"
 		}
+		disks := map[string]*master_pb.DiskInfo{"": {VolumeInfos: []*master_pb.VolumeInformationMessage{vi}}}
+		if c.staleTwin[server] {
+			disks["twin"] = &master_pb.DiskInfo{VolumeInfos: []*master_pb.VolumeInformationMessage{{
+				Id: 7, Collection: "c", RemoteStorageName: "s3.default", RemoteStorageKey: "key", ReplicaPlacement: c.placement,
+			}}}
+		}
 		dns = append(dns, &master_pb.DataNodeInfo{
 			Id:        fmt.Sprintf("127.0.0.1:%d", c.ports[server]),
 			GrpcPort:  uint32(c.ports[server]),
-			DiskInfos: map[string]*master_pb.DiskInfo{"": {VolumeInfos: []*master_pb.VolumeInformationMessage{vi}}},
+			DiskInfos: disks,
 		})
 	}
 	return &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{
@@ -365,3 +374,14 @@ func TestTierDownloadFailsWhenNoReplicaIsListed(t *testing.T) {
 		t.Fatalf("tier download should fail when no replica is listed")
 	}
 }
+
+func TestTierDownloadKeepsRemoteObjectForStaleTwinOnSource(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote)
+	c.staleTwin[c.servers[0]] = true
+	if err := c.run(t); err != nil {
+		t.Fatalf("tier download: %v", err)
+	}
+	if !c.objectExists {
+		t.Errorf("remote object deleted while a twin on %s still references it", c.servers[0])
+	}
+}