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>

weed/shell/command_volume_tier_download.go

@@ -179,6 +179,12 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i
 		}
 	}
 
+	restore, err := freezeTierDownloadSource(grpcDialOption, vid, source)
+	if err != nil {
+		return fmt.Errorf("block deletes on volume %d at %s: %v", vid, source, err)
+	}
+	defer restore(writer)
+
 	// 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.
@@ -219,6 +225,43 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i
 	}
 }
 
+// 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.
+func freezeTierDownloadSource(grpcDialOption grpc.DialOption, vid needle.VolumeId, source pb.ServerAddress) (restore func(io.Writer), err error) {
+	var wasWritable 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 = !resp.IsReadOnly
+		_, markErr := client.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid)})
+		return markErr
+	})
+	restore = func(writer io.Writer) {
+		restoreErr := 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)
+		})
+		if restoreErr != nil {
+			fmt.Fprintf(writer, "restore volume %d state on %s: %v\n", vid, source, restoreErr)
+		}
+	}
+	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 {
@@ -284,21 +327,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

@@ -38,6 +38,9 @@ type tierDownloadCluster struct {
 	// 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
 }
 
 type tierDownloadTestServer struct {
@@ -78,11 +81,36 @@ func (s *tierDownloadTestServer) VolumeUnmount(context.Context, *volume_server_p
 }
 
 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()
+	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) {
-	return &volume_server_pb.VolumeStatusResponse{IsReadOnly: true}, nil
+	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
+	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 {
@@ -97,6 +125,9 @@ func (s *tierDownloadTestServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequ
 	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
@@ -127,6 +158,9 @@ func newTierDownloadCluster(t *testing.T, states ...string) *tierDownloadCluster
 		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),
 	}
 	for _, state := range states {
 		lis, err := net.Listen("tcp", "127.0.0.1:0")
@@ -277,3 +311,24 @@ func TestTierDownloadCopiesFromSsdSourceOntoHddReplica(t *testing.T) {
 	}
 	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])
+		}
+	}
+}