shell: tier download replicates the downloaded volume before deleting the remote object (#12599)

* shell: replicate the downloaded volume before deleting its remote object

volume.tier.download downloaded every replica from the remote tier and let
the last one in the master's location list delete the shared object, so a
replica missing from that list kept reading a deleted object.

Download onto one replica, copy it over every other remote-backed replica,
re-read the topology so replicas that show up meanwhile are replaced too,
and only then delete the last remote-backed replica together with the
object before copying the local volume onto it.

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

* shell: test tier download replicates before deleting the remote object

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

* shell: copy the downloaded volume onto each replica's own disk type

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

* shell: test tier download keeps each replica's disk type

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

* shell: name the hdd disk type when copying a downloaded volume

VolumeCopy treats an empty disk type as the source's, so an HDD replica
copied from an SSD source was placed on SSD or failed.

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

* shell: block deletes on the tier download copy source

A readonly-can-delete source took deletes after VolumeCopy read its
index, so the replaced replica lost them and served the files again.
Hard-mark the source readonly while replicas are replaced, then restore
the writable mark or reload the persisted state.

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

* shell: keep the remote object while replicas are unlisted

Tier download deleted the shared object once the listed replicas were
local, even when the replica placement expected more copies than the
topology listed. Keep it in that case so a rerun can finish later.

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

* shell: fail tier download when no replica is listed

Also point the final-replica delete error at volume.fix.replication.

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

* 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>

* shell: restore the tier download source when its readonly mark fails

VolumeMarkReadonly blocks local writes and deletes before its second master
notification, so a failed notification returns an error with the source
already frozen. Restore it whenever the status was read.

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

* shell: report a failed tier download source restore

A restore that leaves the source unmounted or frozen was only printed, so the
command still returned success. Join the restore error into the result.

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

* shell: wrap the tier download restore error

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/shell/command_volume_tier_download.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -15,6 +16,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
 func init() {
@@ -43,7 +46,11 @@ func (c *commandVolumeTierDownload) Help() string {
 	volume.tier.download -volumeId=7
 	volume.tier.download -volumeId=7 -concurrency=1
 
-	This command will download the dat file of a volume from a remote tier to a volume server in local cluster.
+	This command downloads the dat file of a volume from a remote tier onto one replica, copies
+	that replica over every other replica, and only then deletes the remote object. The object is
+	kept while fewer replicas are listed than the replica placement expects; rerun the command once
+	they are back. A replica on an offline volume server that the placement does not account for
+	keeps referencing the deleted object: delete that replica and run volume.fix.replication.
 
 `
 }
@@ -125,12 +132,6 @@ func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, collectionPattern st
 }
 
 func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, topoInfo *master_pb.TopologyInfo, collectionPattern string, vid needle.VolumeId, concurrency int) (err error) {
-	// find volume location
-	locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(vid))
-	if !found {
-		return fmt.Errorf("volume %d not found", vid)
-	}
-
 	// The volume server verifies the request's collection against the volume's
 	// own, so the pattern must resolve to the actual collection here.
 	collection, found, err := findVolumeCollection(topoInfo, vid, collectionPattern)
@@ -140,28 +141,202 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, topoInfo *ma
 	if !found {
 		return fmt.Errorf("volume %d not found in collections matching %q", vid, collectionPattern)
 	}
+	readTopology := func() (*master_pb.TopologyInfo, error) {
+		topo, _, topoErr := collectTopologyInfo(commandEnv, 0)
+		return topo, topoErr
+	}
+	return downloadAndReplicateFromRemoteTier(commandEnv.option.GrpcDialOption, writer, readTopology, vid, collection, concurrency)
+}
 
-	// All replicas point at the same remote object; only the final download may delete
-	// it. Every earlier replica keeps it so the survivors are not left dangling.
-	// TODO parallelize this
-	for i, loc := range locations {
-		keepRemote := i < len(locations)-1
-		// copy the .dat file from remote tier to local
-		err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote, concurrency)
-		if err != nil {
-			// A replica already made local by a prior interrupted run is not a
-			// failure; skip it so the remaining remote replicas still download.
-			if strings.Contains(err.Error(), "already on local disk") {
-				fmt.Fprintf(writer, "volume %d on %s is already on local disk, skipping\n", vid, loc.Url)
-				continue
+// downloadAndReplicateFromRemoteTier downloads the shared remote object onto
+// one replica, copies that local volume over every other replica, and only
+// then deletes the remote object, so no replica is ever left referencing a
+// deleted object.
+func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, readTopology func() (*master_pb.TopologyInfo, error), vid needle.VolumeId, collection string, concurrency int) (err error) {
+	topoInfo, err := readTopology()
+	if err != nil {
+		return err
+	}
+	remote, local, copies := collectVolumeTierDownloadLocations(topoInfo, vid, collection)
+	listed := len(remote) + len(local)
+	if len(remote) == 0 {
+		if len(local) == 0 {
+			return fmt.Errorf("volume %d is not listed on any volume server", vid)
+		}
+		fmt.Fprintf(writer, "volume %d is already on local disk\n", vid)
+		return nil
+	}
+
+	var source pb.ServerAddress
+	fresh := true
+	if len(local) > 0 {
+		// a previous run already downloaded this replica
+		source = local[0].server
+	} else {
+		source, remote, fresh = remote[0].server, remote[1:], false
+		// a lone replica is the only reference, so it may delete the object itself
+		keepRemote := len(remote) > 0 || listed < copies
+		if err = downloadDatFromRemoteTier(grpcDialOption, writer, vid, collection, source, keepRemote, concurrency); err != nil {
+			if !strings.Contains(err.Error(), "already on local disk") {
+				return fmt.Errorf("download dat file for volume %d to %s: %v", vid, source, err)
 			}
-			return fmt.Errorf("download dat file for volume %d to %s: %v", vid, loc.Url, err)
+			fmt.Fprintf(writer, "volume %d on %s is already on local disk\n", vid, source)
+		}
+		if len(remote) == 0 {
+			if keepRemote {
+				fmt.Fprintf(writer, "volume %d: only %d of %d replicas are listed, the remote object is kept\n", vid, listed, copies)
+			}
+			return nil
 		}
 	}
 
+	restore, err := freezeTierDownloadSource(grpcDialOption, vid, source)
+	if restore != nil {
+		defer func() {
+			if restoreErr := restore(); restoreErr != nil {
+				err = errors.Join(err, fmt.Errorf("restore volume %d state on %s: %w", vid, source, restoreErr))
+			}
+		}()
+	}
+	if err != nil {
+		return fmt.Errorf("block deletes on volume %d at %s: %v", vid, source, err)
+	}
+
+	// Copy over all remote-backed replicas but one, re-reading the topology
+	// so a replica that was offline or created meanwhile is replaced too. The
+	// last one is deleted together with the remote object and then replaced.
+	replaced := map[pb.ServerAddress]bool{source: true}
+	for {
+		var pending []tierDownloadReplica
+		for _, replica := range remote {
+			if !replaced[replica.server] {
+				pending = append(pending, replica)
+			}
+		}
+		if len(pending) == 0 {
+			fmt.Fprintf(writer, "volume %d: no remote-backed replica is listed any more, the remote object is kept\n", vid)
+			return nil
+		}
+		if len(pending) == 1 && fresh {
+			if listed < copies {
+				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 {
+				return fmt.Errorf("delete volume %d with its remote object on %s: %v; if the delete was applied, run volume.fix.replication to restore the replica", vid, last.server, err)
+			}
+			if err = replicateTierDownloadedVolume(grpcDialOption, writer, vid, source, last); err != nil {
+				return fmt.Errorf("%v; run volume.fix.replication to restore the replica", err)
+			}
+			return nil
+		}
+		for _, target := range pending[:len(pending)-1] {
+			if err = replicateTierDownloadedVolume(grpcDialOption, writer, vid, source, target); err != nil {
+				return err
+			}
+			replaced[target.server] = true
+		}
+		if topoInfo, err = readTopology(); err != nil {
+			return err
+		}
+		remote, local, copies = collectVolumeTierDownloadLocations(topoInfo, vid, collection)
+		listed, fresh = len(remote)+len(local), true
+	}
+}
+
+// freezeTierDownloadSource blocks deletes on the copy source: a delete applied
+// after VolumeCopy read the source index would be lost on the replaced replica.
+// restore brings back the writable mark, or reloads the persisted state. It is
+// also returned when the mark fails, since the server may have applied it.
+func freezeTierDownloadSource(grpcDialOption grpc.DialOption, vid needle.VolumeId, source pb.ServerAddress) (restore func() error, err error) {
+	var wasWritable, statusRead bool
+	err = operation.WithVolumeServerClient(false, source, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
+		resp, statusErr := client.VolumeStatus(context.Background(), &volume_server_pb.VolumeStatusRequest{VolumeId: uint32(vid)})
+		if statusErr != nil {
+			return statusErr
+		}
+		wasWritable, statusRead = !resp.IsReadOnly, true
+		_, markErr := client.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid)})
+		return markErr
+	})
+	if !statusRead {
+		return nil, err
+	}
+	restore = func() error {
+		return operation.WithVolumeServerClient(false, source, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
+			if wasWritable {
+				_, markErr := client.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)})
+				return markErr
+			}
+			return remountVolume(client, vid)
+		})
+	}
+	return
+}
+
+func remountVolume(client volume_server_pb.VolumeServerClient, vid needle.VolumeId) error {
+	if _, err := client.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{VolumeId: uint32(vid)}); err != nil {
+		return err
+	}
+	_, err := client.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{VolumeId: uint32(vid)})
+	return err
+}
+
+func replicateTierDownloadedVolume(grpcDialOption grpc.DialOption, writer io.Writer, vid needle.VolumeId, source pb.ServerAddress, target tierDownloadReplica) error {
+	fmt.Fprintf(writer, "replicate downloaded volume %d from %s to %s\n", vid, source, target.server)
+	if _, err := copyVolume(context.Background(), grpcDialOption, writer, vid, source, target.server, types.ToDiskType(target.diskType).ReadableString(), 0, true); err != nil {
+		return fmt.Errorf("replicate volume %d from %s to %s: %v", vid, source, target.server, err)
+	}
 	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
+}
+
+// collectVolumeTierDownloadLocations splits the replicas of a volume into the
+// ones still reading the remote object and the ones already on local disk, and
+// reports how many replicas the replica placement expects.
+func collectVolumeTierDownloadLocations(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collection string) (remote, local []tierDownloadReplica, copies int) {
+	eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
+		for _, disk := range dn.DiskInfos {
+			for _, vi := range disk.VolumeInfos {
+				if needle.VolumeId(vi.Id) != vid || vi.Collection != collection {
+					continue
+				}
+				if rp, rpErr := super_block.NewReplicaPlacementFromByte(byte(vi.ReplicaPlacement)); rpErr == nil {
+					copies = max(copies, rp.GetCopyCount())
+				}
+				replica := tierDownloadReplica{server: pb.NewServerAddressFromDataNode(dn), diskType: vi.DiskType}
+				if vi.RemoteStorageName != "" {
+					remote = append(remote, replica)
+				} else {
+					local = append(local, replica)
+				}
+			}
+		}
+	})
+	return
+}
+
 func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool, concurrency int) error {
 
 	err := operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
@@ -193,21 +368,7 @@ func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer,
 			return downloadErr
 		}
 
-		_, unmountErr := volumeServerClient.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{
-			VolumeId: uint32(volumeId),
-		})
-		if unmountErr != nil {
-			return unmountErr
-		}
-
-		_, mountErr := volumeServerClient.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{
-			VolumeId: uint32(volumeId),
-		})
-		if mountErr != nil {
-			return mountErr
-		}
-
-		return nil
+		return remountVolume(volumeServerClient, volumeId)
 	})
 
 	return err

weed/shell/command_volume_tier_download_test.go

@@ -0,0 +1,430 @@
+package shell
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+const (
+	tierReplicaRemote = "remote"
+	tierReplicaLocal  = "local"
+)
+
+// tierDownloadCluster models the replicas of one tiered volume across fake
+// volume servers sharing one remote object.
+type tierDownloadCluster struct {
+	mu           sync.Mutex
+	servers      []pb.ServerAddress
+	ports        map[pb.ServerAddress]int
+	replicas     map[pb.ServerAddress]string
+	diskTypes    map[pb.ServerAddress]string
+	objectExists bool
+	downloads    int
+	dangling     []string
+	// hiddenOnRead hides servers from the n-th topology read (0-based)
+	hiddenOnRead map[int]map[pb.ServerAddress]bool
+	reads        int
+	frozen       map[pb.ServerAddress]bool
+	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
+	// markFails applies the readonly mark, then fails as a lost master notification does
+	markFails  map[pb.ServerAddress]bool
+	mountFails map[pb.ServerAddress]bool
+}
+
+type tierDownloadTestServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+	cluster *tierDownloadCluster
+	addr    pb.ServerAddress
+}
+
+func (c *tierDownloadCluster) deleteObjectLocked(deleter pb.ServerAddress) {
+	c.objectExists = false
+	for server, state := range c.replicas {
+		if server != deleter && state == tierReplicaRemote {
+			c.dangling = append(c.dangling, string(server))
+		}
+	}
+}
+
+func (s *tierDownloadTestServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.VolumeTierMoveDatFromRemoteRequest, _ volume_server_pb.VolumeServer_VolumeTierMoveDatFromRemoteServer) error {
+	c := s.cluster
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.replicas[s.addr] != tierReplicaRemote {
+		return fmt.Errorf("volume %d is already on local disk", req.VolumeId)
+	}
+	if !c.objectExists {
+		return fmt.Errorf("remote object missing")
+	}
+	c.downloads++
+	c.replicas[s.addr] = tierReplicaLocal
+	if !req.KeepRemoteDatFile {
+		c.deleteObjectLocked(s.addr)
+	}
+	return nil
+}
+
+func (s *tierDownloadTestServer) VolumeUnmount(context.Context, *volume_server_pb.VolumeUnmountRequest) (*volume_server_pb.VolumeUnmountResponse, error) {
+	return &volume_server_pb.VolumeUnmountResponse{}, nil
+}
+
+func (s *tierDownloadTestServer) VolumeMount(context.Context, *volume_server_pb.VolumeMountRequest) (*volume_server_pb.VolumeMountResponse, error) {
+	c := s.cluster
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.mountFails[s.addr] {
+		return nil, fmt.Errorf("mount volume: disk error")
+	}
+	c.frozen[s.addr] = false
+	c.remounts[s.addr]++
+	return &volume_server_pb.VolumeMountResponse{}, nil
+}
+
+func (s *tierDownloadTestServer) VolumeStatus(context.Context, *volume_server_pb.VolumeStatusRequest) (*volume_server_pb.VolumeStatusResponse, error) {
+	c := s.cluster
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return &volume_server_pb.VolumeStatusResponse{IsReadOnly: !c.writable[s.addr] || c.frozen[s.addr]}, nil
+}
+
+func (s *tierDownloadTestServer) VolumeMarkReadonly(_ context.Context, req *volume_server_pb.VolumeMarkReadonlyRequest) (*volume_server_pb.VolumeMarkReadonlyResponse, error) {
+	c := s.cluster
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.frozen[s.addr] = !req.CanDelete
+	if c.markFails[s.addr] {
+		return nil, fmt.Errorf("notify master: leader changed")
+	}
+	return &volume_server_pb.VolumeMarkReadonlyResponse{}, nil
+}
+
+func (s *tierDownloadTestServer) VolumeMarkWritable(context.Context, *volume_server_pb.VolumeMarkWritableRequest) (*volume_server_pb.VolumeMarkWritableResponse, error) {
+	c := s.cluster
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.frozen[s.addr] = false
+	c.writable[s.addr] = true
+	return &volume_server_pb.VolumeMarkWritableResponse{}, nil
+}
+
+func (s *tierDownloadTestServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, _ volume_server_pb.VolumeServer_VolumeCopyServer) error {
+	c := s.cluster
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	// an empty disk type means the source's, as on the real servers
+	want, got := types.ToDiskType(c.diskTypes[s.addr]), types.ToDiskType(c.diskTypes[pb.ServerAddress(req.SourceDataNode)])
+	if req.DiskType != "" {
+		got = types.ToDiskType(req.DiskType)
+	}
+	if got != want {
+		return fmt.Errorf("copy volume %d onto disk type %q, replica is on %q", req.VolumeId, got, want)
+	}
+	if !c.frozen[pb.ServerAddress(req.SourceDataNode)] {
+		return fmt.Errorf("volume %d on %s still accepts deletes during the copy", req.VolumeId, req.SourceDataNode)
+	}
+	switch c.replicas[pb.ServerAddress(req.SourceDataNode)] {
+	case tierReplicaLocal:
+		c.replicas[s.addr] = tierReplicaLocal
+	case tierReplicaRemote:
+		c.replicas[s.addr] = tierReplicaRemote
+	default:
+		return fmt.Errorf("volume %d not found on %s", req.VolumeId, req.SourceDataNode)
+	}
+	return nil
+}
+
+func (s *tierDownloadTestServer) VolumeDelete(_ context.Context, req *volume_server_pb.VolumeDeleteRequest) (*volume_server_pb.VolumeDeleteResponse, error) {
+	c := s.cluster
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.replicas[s.addr] == tierReplicaRemote && !req.KeepRemoteData {
+		c.deleteObjectLocked(s.addr)
+	}
+	delete(c.replicas, s.addr)
+	return &volume_server_pb.VolumeDeleteResponse{}, nil
+}
+
+func newTierDownloadCluster(t *testing.T, states ...string) *tierDownloadCluster {
+	t.Helper()
+	c := &tierDownloadCluster{
+		ports:        make(map[pb.ServerAddress]int),
+		replicas:     make(map[pb.ServerAddress]string),
+		diskTypes:    make(map[pb.ServerAddress]string),
+		objectExists: true,
+		hiddenOnRead: make(map[int]map[pb.ServerAddress]bool),
+		frozen:       make(map[pb.ServerAddress]bool),
+		writable:     make(map[pb.ServerAddress]bool),
+		remounts:     make(map[pb.ServerAddress]int),
+		staleTwin:    make(map[pb.ServerAddress]bool),
+		markFails:    make(map[pb.ServerAddress]bool),
+		mountFails:   make(map[pb.ServerAddress]bool),
+	}
+	for _, state := range states {
+		lis, err := net.Listen("tcp", "127.0.0.1:0")
+		if err != nil {
+			t.Fatalf("listen: %v", err)
+		}
+		port := lis.Addr().(*net.TCPAddr).Port
+		addr := pb.NewServerAddress("127.0.0.1", port, port)
+		srv := grpc.NewServer()
+		volume_server_pb.RegisterVolumeServerServer(srv, &tierDownloadTestServer{cluster: c, addr: addr})
+		go srv.Serve(lis)
+		t.Cleanup(srv.Stop)
+		c.servers = append(c.servers, addr)
+		c.ports[addr] = port
+		c.replicas[addr] = state
+		if len(c.servers)%2 == 0 {
+			c.diskTypes[addr] = "ssd"
+		}
+	}
+	return c
+}
+
+func (c *tierDownloadCluster) hide(read int, servers ...pb.ServerAddress) {
+	if c.hiddenOnRead[read] == nil {
+		c.hiddenOnRead[read] = make(map[pb.ServerAddress]bool)
+	}
+	for _, server := range servers {
+		c.hiddenOnRead[read][server] = true
+	}
+}
+
+func (c *tierDownloadCluster) readTopology() (*master_pb.TopologyInfo, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	hidden := c.hiddenOnRead[c.reads]
+	c.reads++
+	var dns []*master_pb.DataNodeInfo
+	for _, server := range c.servers {
+		state, found := c.replicas[server]
+		if !found || hidden[server] {
+			continue
+		}
+		vi := &master_pb.VolumeInformationMessage{Id: 7, Collection: "c", DiskType: c.diskTypes[server], ReplicaPlacement: c.placement}
+		if state == tierReplicaRemote {
+			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: disks,
+		})
+	}
+	return &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{
+		RackInfos: []*master_pb.RackInfo{{DataNodeInfos: dns}},
+	}}}, nil
+}
+
+func (c *tierDownloadCluster) run(t *testing.T) error {
+	t.Helper()
+	return downloadAndReplicateFromRemoteTier(grpc.WithTransportCredentials(insecure.NewCredentials()), io.Discard, c.readTopology, needle.VolumeId(7), "c", 0)
+}
+
+func (c *tierDownloadCluster) assertAllLocal(t *testing.T) {
+	t.Helper()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, server := range c.servers {
+		if state := c.replicas[server]; state != tierReplicaLocal {
+			t.Errorf("replica on %s: want local, got %q", server, state)
+		}
+	}
+	if len(c.dangling) > 0 {
+		t.Errorf("remote object deleted while still referenced by %s", strings.Join(c.dangling, ","))
+	}
+}
+
+func TestTierDownloadReplicatesBeforeDeletingRemoteObject(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote)
+	if err := c.run(t); err != nil {
+		t.Fatalf("tier download: %v", err)
+	}
+	c.assertAllLocal(t)
+	if c.objectExists {
+		t.Errorf("remote object should be deleted once every replica is local")
+	}
+	if c.downloads != 1 {
+		t.Errorf("want one download from the remote tier, got %d", c.downloads)
+	}
+}
+
+func TestTierDownloadReplacesReplicaMissingFromFirstTopology(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote)
+	c.hide(0, c.servers[2])
+	if err := c.run(t); err != nil {
+		t.Fatalf("tier download: %v", err)
+	}
+	c.assertAllLocal(t)
+	if c.objectExists {
+		t.Errorf("remote object should be deleted once every replica is local")
+	}
+}
+
+func TestTierDownloadKeepsRemoteObjectWhenLastReplicaVanishes(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote)
+	c.hide(1, c.servers[2])
+	if err := c.run(t); err != nil {
+		t.Fatalf("tier download: %v", err)
+	}
+	if !c.objectExists {
+		t.Fatalf("remote object must be kept while an unlisted replica may still reference it")
+	}
+	if len(c.dangling) > 0 {
+		t.Errorf("remote object deleted while still referenced by %v", c.dangling)
+	}
+}
+
+func TestTierDownloadSingleReplicaDeletesRemoteObject(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaRemote)
+	if err := c.run(t); err != nil {
+		t.Fatalf("tier download: %v", err)
+	}
+	c.assertAllLocal(t)
+	if c.objectExists {
+		t.Errorf("remote object should be deleted")
+	}
+}
+
+func TestTierDownloadRerunReusesLocalReplica(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote)
+	if err := c.run(t); err != nil {
+		t.Fatalf("tier download: %v", err)
+	}
+	c.assertAllLocal(t)
+	if c.objectExists {
+		t.Errorf("remote object should be deleted once every replica is local")
+	}
+	if c.downloads != 0 {
+		t.Errorf("want no download when a replica is already local, got %d", c.downloads)
+	}
+}
+
+func TestTierDownloadCopiesFromSsdSourceOntoHddReplica(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote)
+	c.hide(0, c.servers[0])
+	if err := c.run(t); err != nil {
+		t.Fatalf("tier download: %v", err)
+	}
+	c.assertAllLocal(t)
+}
+
+func TestTierDownloadRestoresSourceAfterBlockingDeletes(t *testing.T) {
+	for _, writable := range []bool{false, true} {
+		c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote)
+		source := c.servers[0]
+		c.writable[source] = writable
+		if err := c.run(t); err != nil {
+			t.Fatalf("tier download (writable=%v): %v", writable, err)
+		}
+		c.assertAllLocal(t)
+		if c.frozen[source] {
+			t.Errorf("writable=%v: source still blocks deletes after the command", writable)
+		}
+		if c.writable[source] != writable {
+			t.Errorf("writable=%v: source writable state changed to %v", writable, c.writable[source])
+		}
+		if !writable && c.remounts[source] != 1 {
+			t.Errorf("readonly source should reload its persisted state once, got %d remounts", c.remounts[source])
+		}
+	}
+}
+
+func TestTierDownloadKeepsRemoteObjectWhileReplicasAreUnlisted(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		states []string
+	}{
+		{"lone listed replica", []string{tierReplicaRemote, tierReplicaRemote}},
+		{"local source and one remote replica", []string{tierReplicaLocal, tierReplicaRemote, tierReplicaRemote}},
+		{"three remote replicas", []string{tierReplicaRemote, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote}},
+	} {
+		c := newTierDownloadCluster(t, tc.states...)
+		c.placement = uint32(len(tc.states) - 1) // "00n": n+1 copies, one more than listed
+		offline := c.servers[len(c.servers)-1]
+		for read := 0; read < 4; read++ {
+			c.hide(read, offline)
+		}
+		if err := c.run(t); err != nil {
+			t.Fatalf("%s: tier download: %v", tc.name, err)
+		}
+		if !c.objectExists {
+			t.Errorf("%s: remote object deleted while the replica on %s is unlisted", tc.name, offline)
+		}
+	}
+}
+
+func TestTierDownloadFailsWhenNoReplicaIsListed(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaRemote)
+	c.hide(0, c.servers[0])
+	if err := c.run(t); err == nil {
+		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])
+	}
+}
+
+func TestTierDownloadRestoresSourceWhenMarkFailsAfterApplying(t *testing.T) {
+	for _, writable := range []bool{false, true} {
+		c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote)
+		source := c.servers[0]
+		c.writable[source] = writable
+		c.markFails[source] = true
+		if err := c.run(t); err == nil {
+			t.Fatalf("writable=%v: tier download should fail when the readonly mark fails", writable)
+		}
+		if c.frozen[source] {
+			t.Errorf("writable=%v: source still blocks deletes after the failed mark", writable)
+		}
+		if c.writable[source] != writable {
+			t.Errorf("writable=%v: source writable state changed to %v", writable, c.writable[source])
+		}
+		if !c.objectExists {
+			t.Errorf("writable=%v: remote object deleted although no replica was replaced", writable)
+		}
+	}
+}
+
+func TestTierDownloadReportsSourceRestoreFailure(t *testing.T) {
+	c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote)
+	source := c.servers[0]
+	c.mountFails[source] = true
+	err := c.run(t)
+	if err == nil || !strings.Contains(err.Error(), "restore volume 7 state") {
+		t.Fatalf("tier download should report the failed source restore, got %v", err)
+	}
+	c.assertAllLocal(t)
+}