diff --git a/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java b/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java
index a72920fbae88..60c9983f016e 100644
--- a/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java
+++ b/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java
@@ -717,15 +717,21 @@ public boolean restoreVMFromBackup(VirtualMachine vm, Backup backup) {
     }
 
     private Pair<Boolean, String> restoreVMBackup(VirtualMachine vm, Backup backup) {
-        List<Backup.VolumeInfo> backedVolumes = backup.getBackedUpVolumes();
-        List<String> backedVolumesUUIDs = backedVolumes.stream()
+        // Sort once and derive every per-volume list from that same ordering. The UUID list and the
+        // backup file list are consumed index-by-index on the agent side, so they have to agree;
+        // sorting only one of them leaves the two out of step whenever the stored order of the
+        // backed up volumes differs from their device id order.
+        List<Backup.VolumeInfo> backedVolumes = backup.getBackedUpVolumes().stream()
                 .sorted(Comparator.comparingLong(Backup.VolumeInfo::getDeviceId))
+                .collect(Collectors.toList());
+        List<String> backedVolumesUUIDs = backedVolumes.stream()
                 .map(Backup.VolumeInfo::getUuid)
                 .collect(Collectors.toList());
 
-        List<VolumeVO> restoreVolumes = volumeDao.findByInstance(vm.getId()).stream()
+        List<VolumeVO> vmVolumes = volumeDao.findByInstance(vm.getId()).stream()
                 .sorted(Comparator.comparingLong(VolumeVO::getDeviceId))
                 .collect(Collectors.toList());
+        List<VolumeVO> restoreVolumes = getVolumesToRestoreInto(backedVolumes, vmVolumes);
 
         LOG.debug("Restoring vm {} from backup {} on the NAS Backup Provider", vm, backup);
         BackupRepository backupRepository = getBackupRepository(backup);
@@ -765,6 +771,42 @@ private Pair<Boolean, String> restoreVMBackup(VirtualMachine vm, Backup backup)
         return new Pair<>(answer.getResult(), answer.getDetails());
     }
 
+    /**
+     * Returns, for each backed up volume in the given order, the instance's volume its backup has to be
+     * written into.
+     * <p>
+     * A backed up volume goes into the instance's volume with the same uuid. The uuid is compared, not the
+     * volume's path: the path is the volume's name on the storage, and it differs from the uuid once the
+     * volume was migrated to another pool, or when it was created from a snapshot or an upload.
+     * <p>
+     * Creating an instance from a backup gives it brand new volumes, so none of the uuids can match. Only then
+     * are the instance's volumes taken in device id order. When some of them match, a missing one is a volume
+     * that is no longer attached, and the restore is refused before anything is written, instead of writing
+     * that backup into another volume.
+     */
+    protected List<VolumeVO> getVolumesToRestoreInto(List<Backup.VolumeInfo> backedVolumes, List<VolumeVO> vmVolumes) {
+        Map<String, VolumeVO> vmVolumeByUuid = vmVolumes.stream().collect(Collectors.toMap(VolumeVO::getUuid, volume -> volume));
+        boolean restoringIntoNewVolumes = backedVolumes.stream().noneMatch(backedVolume -> vmVolumeByUuid.containsKey(backedVolume.getUuid()));
+        if (restoringIntoNewVolumes) {
+            if (backedVolumes.size() != vmVolumes.size()) {
+                throw new CloudRuntimeException(String.format("Unable to restore backup: it holds %d volumes but the instance has %d.",
+                        backedVolumes.size(), vmVolumes.size()));
+            }
+            LOG.debug("None of the backed up volumes belong to this instance; restoring into its volumes in device id order.");
+            return vmVolumes;
+        }
+        List<VolumeVO> restoreVolumes = new ArrayList<>();
+        for (Backup.VolumeInfo backedVolume : backedVolumes) {
+            VolumeVO volume = vmVolumeByUuid.get(backedVolume.getUuid());
+            if (volume == null) {
+                throw new CloudRuntimeException(String.format("Unable to restore backup: volume [%s] recorded in the backup"
+                        + " is not attached to the instance any more.", backedVolume.getUuid()));
+            }
+            restoreVolumes.add(volume);
+        }
+        return restoreVolumes;
+    }
+
     private List<String> getBackupFiles(List<Backup.VolumeInfo> backedVolumes) {
         List<String> backupFiles = new ArrayList<>();
         for (Backup.VolumeInfo backedVolume : backedVolumes) {
diff --git a/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java b/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java
index 787d3ff94567..fbd681e726f8 100644
--- a/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java
+++ b/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java
@@ -56,6 +56,7 @@
 import com.cloud.user.ResourceLimitService;
 import com.cloud.utils.Pair;
 import com.cloud.utils.db.GlobalLock;
+import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.vm.VMInstanceDetailVO;
 import com.cloud.vm.VMInstanceVO;
 import com.cloud.vm.VirtualMachine;
@@ -442,6 +443,71 @@ public void testGetVMHypervisorHostFallbackToZoneWideKVMHost() {
         Mockito.verify(resourceManager).findOneRandomRunningHostByHypervisor(Hypervisor.HypervisorType.KVM, zoneId);
     }
 
+    private VolumeVO volume(String uuid, String path, long deviceId) {
+        VolumeVO volume = mock(VolumeVO.class);
+        Mockito.lenient().when(volume.getUuid()).thenReturn(uuid);
+        Mockito.lenient().when(volume.getPath()).thenReturn(path);
+        Mockito.lenient().when(volume.getDeviceId()).thenReturn(deviceId);
+        return volume;
+    }
+
+    private Backup.VolumeInfo backedVolume(String uuid, long deviceId) {
+        return new Backup.VolumeInfo(uuid, uuid, deviceId == 0 ? Volume.Type.ROOT : Volume.Type.DATADISK, 1024L, deviceId, null, null, null);
+    }
+
+    @Test
+    public void getVolumesToRestoreIntoMatchesByUuidNotByPathOrDeviceId() {
+        // data-a was migrated (its path is no longer its uuid) and the data disks swapped device ids
+        VolumeVO root = volume("root", "root", 0L);
+        VolumeVO dataB = volume("data-b", "data-b", 1L);
+        VolumeVO dataA = volume("data-a", "migrated-path-of-a", 2L);
+        List<Backup.VolumeInfo> backed = List.of(backedVolume("root", 0L), backedVolume("data-a", 1L), backedVolume("data-b", 2L));
+
+        List<VolumeVO> result = nasBackupProvider.getVolumesToRestoreInto(backed, List.of(root, dataB, dataA));
+
+        Assert.assertEquals(List.of(root, dataA, dataB), result);
+    }
+
+    @Test
+    public void getVolumesToRestoreIntoRefusesWhenABackedUpVolumeIsNoLongerAttached() {
+        List<Backup.VolumeInfo> backed = List.of(backedVolume("root", 0L), backedVolume("data-z", 1L));
+        List<VolumeVO> vmVolumes = List.of(volume("root", "root", 0L), volume("data-a", "data-a", 1L));
+
+        CloudRuntimeException e = Assert.assertThrows(CloudRuntimeException.class,
+                () -> nasBackupProvider.getVolumesToRestoreInto(backed, vmVolumes));
+        Assert.assertTrue(e.getMessage().contains("data-z"));
+    }
+
+    @Test
+    public void getVolumesToRestoreIntoUsesDeviceIdOrderForBrandNewVolumes() {
+        // an instance created from the backup: none of its volumes has a uuid recorded in the backup
+        VolumeVO newRoot = volume("new-root", "new-root", 0L);
+        VolumeVO newData = volume("new-data", "new-data", 1L);
+        List<Backup.VolumeInfo> backed = List.of(backedVolume("root", 0L), backedVolume("data-a", 1L));
+
+        Assert.assertEquals(List.of(newRoot, newData), nasBackupProvider.getVolumesToRestoreInto(backed, List.of(newRoot, newData)));
+    }
+
+    @Test(expected = CloudRuntimeException.class)
+    public void getVolumesToRestoreIntoRefusesBrandNewVolumesOfADifferentCount() {
+        nasBackupProvider.getVolumesToRestoreInto(List.of(backedVolume("root", 0L), backedVolume("data-a", 1L)),
+                List.of(volume("new-root", "new-root", 0L)));
+    }
+
+    @Test
+    public void restoreVMFromBackupSendsNothingWhenABackedUpVolumeIsNoLongerAttached() throws AgentUnavailableException, OperationTimedoutException {
+        VMInstanceVO vm = mock(VMInstanceVO.class);
+        Mockito.when(vm.getId()).thenReturn(1L);
+        BackupVO backup = mock(BackupVO.class);
+        Mockito.when(backup.getBackedUpVolumes()).thenReturn(List.of(backedVolume("root", 0L), backedVolume("data-z", 1L)));
+        VolumeVO root = volume("root", "root", 0L);
+        VolumeVO dataA = volume("data-a", "data-a", 1L);
+        Mockito.when(volumeDao.findByInstance(1L)).thenReturn(List.of(root, dataA));
+
+        Assert.assertThrows(CloudRuntimeException.class, () -> nasBackupProvider.restoreVMFromBackup(vm, backup));
+        Mockito.verify(agentManager, Mockito.never()).send(anyLong(), Mockito.any(com.cloud.agent.api.Command.class));
+    }
+
     // -- nas.backup.incremental.enabled master switch ------------------------------------
 
     /**
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapper.java
index 92de04b1b95a..d067a29d31a1 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapper.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapper.java
@@ -132,6 +132,9 @@ private void restoreVolumesOfExistingVM(KVMStoragePoolManager storagePoolMgr, Li
                                             String backupPath, List<String> backupFiles, String mountDirectory, int timeout, Integer mountTimeout) {
         String diskType = "root";
         try {
+            // The management server sends the instance's volumes in the order of the backed up volumes,
+            // each one being the volume that backup has to be written into (it matches them by uuid, see
+            // NASBackupProvider#getVolumesToRestoreInto), so they are paired by position here.
             for (int idx = 0; idx < restoreVolumePaths.size(); idx++) {
                 PrimaryDataStoreTO restoreVolumePool = restoreVolumePools.get(idx);
                 String restoreVolumePath = restoreVolumePaths.get(idx);
diff --git a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapperTest.java b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapperTest.java
index 3bdc23d27a15..138b57aef0ac 100644
--- a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapperTest.java
+++ b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRestoreBackupCommandWrapperTest.java
@@ -16,6 +16,7 @@
 // under the License.
 package com.cloud.hypervisor.kvm.resource.wrapper;
 
+import static org.mockito.AdditionalMatchers.aryEq;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -587,6 +588,64 @@ public void testExecuteWithMultipleVolumes() throws Exception {
         }
     }
 
+    /**
+     * The management server sends the volumes to restore into in the order of the backed up volumes (it
+     * matches them by uuid). Their paths do not have to contain the backed up uuid: a volume migrated to
+     * another pool, or created from a snapshot, has a path that differs from its uuid. Each backup file
+     * has to be written into the volume at the same position.
+     */
+    @Test
+    public void testRestoreOfExistingVmWritesEachBackupIntoTheVolumeAtTheSamePosition() throws Exception {
+        when(command.getVmName()).thenReturn("test-vm");
+        when(command.getBackupPath()).thenReturn("backup/path");
+        when(command.getBackupRepoAddress()).thenReturn("192.168.1.100:/backup");
+        when(command.getBackupRepoType()).thenReturn("nfs");
+        when(command.getMountOptions()).thenReturn("rw");
+        when(command.isVmExists()).thenReturn(true);
+        when(command.getDiskType()).thenReturn("root");
+        PrimaryDataStoreTO pool = Mockito.mock(PrimaryDataStoreTO.class);
+        lenient().when(pool.getPoolType()).thenReturn(Storage.StoragePoolType.NetworkFilesystem);
+        when(command.getRestoreVolumePools()).thenReturn(Arrays.asList(pool, pool, pool));
+        // data-vol-a was migrated, so its path is a different name than its uuid
+        when(command.getRestoreVolumePaths()).thenReturn(Arrays.asList(
+                "/var/lib/libvirt/images/root-vol",
+                "/var/lib/libvirt/images/migrated-path-of-a",
+                "/var/lib/libvirt/images/data-vol-b"
+        ));
+        when(command.getBackupVolumesUUIDs()).thenReturn(Arrays.asList("root-vol", "data-vol-a", "data-vol-b"));
+        when(command.getBackupFiles()).thenReturn(Arrays.asList("root-vol", "data-vol-a", "data-vol-b"));
+        when(command.getMountTimeout()).thenReturn(30);
+
+        try (MockedStatic<Files> filesMock = mockStatic(Files.class)) {
+            Path tempPath = Mockito.mock(Path.class);
+            when(tempPath.toString()).thenReturn("/tmp/csbackup.abc123");
+            filesMock.when(() -> Files.createTempDirectory(anyString())).thenReturn(tempPath);
+
+            try (MockedStatic<Script> scriptMock = mockStatic(Script.class)) {
+                scriptMock.when(() -> Script.getExecutableAbsolutePath(anyString()))
+                        .thenAnswer(invocation -> invocation.getArgument(0));
+                scriptMock.when(() -> Script.executeCommandForExitValue(anyLong(), any(String[].class))).thenReturn(0);
+                scriptMock.when(() -> Script.runSimpleBashScriptForExitValue(anyString()))
+                        .thenAnswer(invocation -> {
+                            String command = invocation.getArgument(0);
+                            // No backing chain, so the restore takes the rsync path this test asserts on.
+                            return command.contains("backing-filename") ? 1 : 0;
+                        });
+                filesMock.when(() -> Files.deleteIfExists(any(Path.class))).thenReturn(true);
+
+                Answer result = wrapper.execute(command, libvirtComputingResource);
+
+                Assert.assertTrue(((BackupAnswer) result).getResult());
+                scriptMock.verify(() -> Script.executeCommandForExitValue(anyLong(), aryEq(new String[] {"rsync", "-az",
+                        "/tmp/csbackup.abc123/backup/path/root.root-vol.qcow2", "/var/lib/libvirt/images/root-vol"})));
+                scriptMock.verify(() -> Script.executeCommandForExitValue(anyLong(), aryEq(new String[] {"rsync", "-az",
+                        "/tmp/csbackup.abc123/backup/path/datadisk.data-vol-a.qcow2", "/var/lib/libvirt/images/migrated-path-of-a"})));
+                scriptMock.verify(() -> Script.executeCommandForExitValue(anyLong(), aryEq(new String[] {"rsync", "-az",
+                        "/tmp/csbackup.abc123/backup/path/datadisk.data-vol-b.qcow2", "/var/lib/libvirt/images/data-vol-b"})));
+            }
+        }
+    }
+
     private String invokeGetDeviceToAttachDisk(String vmName) throws Exception {
         Method method = LibvirtRestoreBackupCommandWrapper.class.getDeclaredMethod("getDeviceToAttachDisk", String.class);
         method.setAccessible(true);
diff --git a/server/src/main/java/com/cloud/hypervisor/KVMGuru.java b/server/src/main/java/com/cloud/hypervisor/KVMGuru.java
index afb695d78574..7c58097c3eb8 100644
--- a/server/src/main/java/com/cloud/hypervisor/KVMGuru.java
+++ b/server/src/main/java/com/cloud/hypervisor/KVMGuru.java
@@ -369,9 +369,18 @@ public VirtualMachine importVirtualMachineFromBackup(long zoneId, long domainId,
                    _volumeDao.update(volume.getId(), volume);
                    _volumeDao.attachVolume(volume.getId(), vm.getId(), 0L);
                } else if (VMVolToRestore.getType() == Volume.Type.DATADISK) {
-                   List<VolumeVO> vmVolumes = _volumeDao.findByInstance(vm.getId());
                    _volumeDao.update(volume.getId(), volume);
-                   _volumeDao.attachVolume(volume.getId(), vm.getId(), getNextAvailableDeviceId(vmVolumes));
+                   // Reinstate the device id the volume had when the backup was taken. The volumes are
+                   // not detached before this loop, so their current device ids still count as "in use"
+                   // and getNextAvailableDeviceId() can never return the id a volume already holds. Using
+                   // it here shifts every data disk by one slot on every restore, which then misaligns the
+                   // device-id ordering that the restore itself uses to map backup files onto volumes.
+                   Long deviceId = VMVolToRestore.getDeviceId();
+                   if (deviceId == null) {
+                       // Backups taken before the device id was recorded in the backup metadata.
+                       deviceId = getNextAvailableDeviceId(_volumeDao.findByInstance(vm.getId()));
+                   }
+                   _volumeDao.attachVolume(volume.getId(), vm.getId(), deviceId);
                }
                UsageEventUtils.publishUsageEvent(EventTypes.EVENT_VOLUME_ATTACH, volume.getAccountId(), volume.getDataCenterId(), volume.getId(), volume.getName(),
                        volume.getDiskOfferingId(), volume.getTemplateId(), volume.getSize(), Volume.class.getName(), volume.getUuid(), vm.getId(), volume.isDisplay());
diff --git a/server/src/test/java/com/cloud/hypervisor/KVMGuruTest.java b/server/src/test/java/com/cloud/hypervisor/KVMGuruTest.java
index d94f9db0c99c..1e63dcbcf65a 100644
--- a/server/src/test/java/com/cloud/hypervisor/KVMGuruTest.java
+++ b/server/src/test/java/com/cloud/hypervisor/KVMGuruTest.java
@@ -18,10 +18,12 @@
 
 import java.io.UnsupportedEncodingException;
 import java.util.Arrays;
+import java.util.List;
 import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.cloudstack.api.ApiConstants;
+import org.apache.cloudstack.backup.Backup;
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.utils.bytescale.ByteScaleUtils;
 import org.junit.After;
@@ -31,6 +33,7 @@
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
+import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 import org.mockito.Spy;
 import org.mockito.junit.MockitoJUnitRunner;
@@ -38,6 +41,7 @@
 import com.cloud.agent.api.to.NicTO;
 import com.cloud.agent.api.to.VirtualMachineTO;
 import com.cloud.configuration.ConfigurationManagerImpl;
+import com.cloud.event.UsageEventUtils;
 import com.cloud.host.HostVO;
 import com.cloud.host.dao.HostDao;
 import com.cloud.hypervisor.kvm.dpdk.DpdkHelper;
@@ -48,11 +52,16 @@
 import com.cloud.service.dao.ServiceOfferingDetailsDao;
 import com.cloud.storage.GuestOSHypervisorVO;
 import com.cloud.storage.GuestOSVO;
+import com.cloud.storage.Volume;
+import com.cloud.storage.VolumeVO;
+import com.cloud.storage.dao.VolumeDao;
 import com.cloud.storage.dao.GuestOSDao;
 import com.cloud.storage.dao.GuestOSHypervisorDao;
 import com.cloud.utils.Pair;
+import com.cloud.vm.VMInstanceVO;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachineProfile;
+import com.cloud.vm.dao.VMInstanceDao;
 
 @RunWith(MockitoJUnitRunner.class)
 public class KVMGuruTest {
@@ -61,6 +70,10 @@ public class KVMGuruTest {
     HostDao hostDao;
     @Mock
     ServiceOfferingDetailsDao serviceOfferingDetailsDao;
+    @Mock
+    VMInstanceDao _instanceDao;
+    @Mock
+    VolumeDao _volumeDao;
 
     @Spy
     @InjectMocks
@@ -506,4 +519,39 @@ public void testGetNullWhenVMThereIsNoInformationOfUsedHosts() {
 
         Assert.assertNull(clusterId);
     }
+
+    /**
+     * The restore maps backup files onto volumes by their device id ordering, so re-importing the
+     * instance has to put every data disk back on the device id recorded in the backup. Handing out
+     * the next free id instead shifts the disks on every restore and misaligns that mapping.
+     */
+    @Test
+    public void testImportVirtualMachineFromBackupReinstatesRecordedDeviceIds() {
+        String vmInternalName = "i-2-42-VM";
+        VMInstanceVO vmInstance = Mockito.mock(VMInstanceVO.class);
+        Mockito.when(vmInstance.getId()).thenReturn(42L);
+        Mockito.when(vmInstance.getRemoved()).thenReturn(null);
+        Mockito.when(_instanceDao.findVMByInstanceNameIncludingRemoved(vmInternalName)).thenReturn(vmInstance);
+
+        Backup.VolumeInfo rootVolume = new Backup.VolumeInfo("root-uuid", "root-uuid", Volume.Type.ROOT, 1L, 0L, null, null, null);
+        Backup.VolumeInfo dataVolume = new Backup.VolumeInfo("data-uuid", "data-uuid", Volume.Type.DATADISK, 1L, 5L, null, null, null);
+        Mockito.when(vmInstance.getBackupVolumeList()).thenReturn(List.of(rootVolume, dataVolume));
+
+        VolumeVO rootVolumeVO = Mockito.mock(VolumeVO.class);
+        Mockito.when(rootVolumeVO.getId()).thenReturn(10L);
+        VolumeVO dataVolumeVO = Mockito.mock(VolumeVO.class);
+        Mockito.when(dataVolumeVO.getId()).thenReturn(11L);
+        Mockito.when(_volumeDao.findByUuidIncludingRemoved("root-uuid")).thenReturn(rootVolumeVO);
+        Mockito.when(_volumeDao.findByUuidIncludingRemoved("data-uuid")).thenReturn(dataVolumeVO);
+
+        Backup backup = Mockito.mock(Backup.class);
+
+        try (MockedStatic<UsageEventUtils> ignored = Mockito.mockStatic(UsageEventUtils.class)) {
+            guru.importVirtualMachineFromBackup(1L, 1L, 1L, 1L, vmInternalName, backup);
+        }
+
+        Mockito.verify(_volumeDao).attachVolume(10L, 42L, 0L);
+        Mockito.verify(_volumeDao).attachVolume(11L, 42L, 5L);
+        Mockito.verify(guru, Mockito.never()).getNextAvailableDeviceId(Mockito.anyList());
+    }
 }