kvm: fix delete of failed Instance Snapshot on stopped VM with RBD storage

StorageVMSnapshotStrategy.canHandle() required either State.Allocated or
a persisted vm_snapshot_details/STORAGE_SNAPSHOT row before it would
claim a VM snapshot. A snapshot that fails before any underlying disk
snapshot is taken (e.g. VM freeze fails because the QEMU guest agent is
not connected) transitions to State.Error without ever persisting such
a row, so canHandle() returned CANT_HANDLE and deletion fell back to
DefaultVMSnapshotStrategy. That strategy sends a DeleteVMSnapshotCommand
to the KVM agent, which requires a live libvirt domain - failing with
'Domain not found' once the VM is stopped, and blocking both snapshot
deletion and VM destruction.

Let StorageVMSnapshotStrategy also claim ownership of snapshots in
State.Error even when no disk-snapshot details were persisted: there is
nothing on the hypervisor to clean up in that case, and
deleteVMSnapshot()/deleteDiskSnapshot() are already safe no-ops when the
details list is empty - no libvirt domain is required.

Fixes #13972

Author: weizhouapache

engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/vmsnapshot/StorageVMSnapshotStrategy.java

@@ -338,7 +338,9 @@ public boolean revertVMSnapshot(VMSnapshot vmSnapshot) {
     @Override
     public StrategyPriority canHandle(VMSnapshot vmSnapshot) {
        UserVmVO userVm = userVmDao.findById(vmSnapshot.getVmId());
-       if (!VMSnapshot.State.Allocated.equals(vmSnapshot.getState())) {
+       // A snapshot that failed before any disk snapshot was taken (e.g. freeze failed) is in Error state
+       // without STORAGE_SNAPSHOT details. Still handle it, as the default strategy needs a libvirt domain.
+       if (!VMSnapshot.State.Allocated.equals(vmSnapshot.getState()) && !VMSnapshot.State.Error.equals(vmSnapshot.getState())) {
            List<VMSnapshotDetailsVO> vmSnapshotDetails = vmSnapshotDetailsDao.findDetails(vmSnapshot.getId(), STORAGE_SNAPSHOT);
            if (CollectionUtils.isEmpty(vmSnapshotDetails)) {
                return StrategyPriority.CANT_HANDLE;

engine/storage/snapshot/src/test/java/org/apache/cloudstack/storage/vmsnapshot/VMSnapshotStrategyKVMTest.java

@@ -37,6 +37,7 @@
 import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotStrategy;
 import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotStrategy.SnapshotOperation;
 import org.apache.cloudstack.engine.subsystem.api.storage.StorageStrategyFactory;
+import org.apache.cloudstack.engine.subsystem.api.storage.StrategyPriority;
 import org.apache.cloudstack.engine.subsystem.api.storage.VMSnapshotOptions;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
@@ -297,6 +298,45 @@ public void testDeleteVMsnapshot() throws AgentUnavailableException, OperationTi
         assertTrue(result);
     }
 
+    @Test
+    public void testCanHandleErrorStateWithNoDiskSnapshots() {
+        // Simulates a failed Instance Snapshot (e.g. freeze failed before any disk snapshot was taken):
+        // no vm_snapshot_details/STORAGE_SNAPSHOT rows exist, but the strategy must still claim it so
+        // delete/cleanup does not fall back to DefaultVMSnapshotStrategy (which needs a live libvirt domain).
+        VMSnapshotVO vmSnapshot = Mockito.mock(VMSnapshotVO.class);
+        Long vmId = 1L;
+        UserVmVO userVmVO = Mockito.mock(UserVmVO.class);
+        Mockito.when(vmSnapshot.getVmId()).thenReturn(vmId);
+        Mockito.when(vmSnapshot.getId()).thenReturn(1L);
+        Mockito.when(vmSnapshot.getState()).thenReturn(VMSnapshot.State.Error);
+        Mockito.when(vmSnapshot.getType()).thenReturn(VMSnapshot.Type.Disk);
+        Mockito.when(userVmDao.findById(vmId)).thenReturn(userVmVO);
+        Mockito.when(userVmVO.getHypervisorType()).thenReturn(HypervisorType.KVM);
+        Mockito.when(vmSnapshotDetailsDao.findDetails(ArgumentMatchers.anyLong(), ArgumentMatchers.anyString()))
+                .thenReturn(new ArrayList<>());
+
+        StrategyPriority priority = vmStrategy.canHandle(vmSnapshot);
+        assertEquals(StrategyPriority.HYPERVISOR, priority);
+    }
+
+    @Test
+    public void testCanHandleReadyStateWithNoDiskSnapshotsCantHandle() {
+        // A Ready snapshot with no persisted disk snapshots is inconsistent state, unrelated to the
+        // Error-before-any-snapshot scenario above, so the existing CANT_HANDLE behaviour is preserved.
+        VMSnapshotVO vmSnapshot = Mockito.mock(VMSnapshotVO.class);
+        Long vmId = 1L;
+        UserVmVO userVmVO = Mockito.mock(UserVmVO.class);
+        Mockito.when(vmSnapshot.getVmId()).thenReturn(vmId);
+        Mockito.when(vmSnapshot.getId()).thenReturn(1L);
+        Mockito.when(vmSnapshot.getState()).thenReturn(VMSnapshot.State.Ready);
+        Mockito.when(userVmDao.findById(vmId)).thenReturn(userVmVO);
+        Mockito.when(vmSnapshotDetailsDao.findDetails(ArgumentMatchers.anyLong(), ArgumentMatchers.anyString()))
+                .thenReturn(new ArrayList<>());
+
+        StrategyPriority priority = vmStrategy.canHandle(vmSnapshot);
+        assertEquals(StrategyPriority.CANT_HANDLE, priority);
+    }
+
     @SuppressWarnings("unchecked")
     private SnapshotVO testFindSnapshotByName(String snapshotName) {
         SearchBuilder<SnapshotVO> sb = Mockito.mock(SearchBuilder.class);