From 8643df8e429a556e22d5183a69bbcc95800ee8e0 Mon Sep 17 00:00:00 2001
From: Rene Peinthor <rene.peinthor@linbit.com>
Date: Wed, 29 Jul 2026 15:01:33 +0200
Subject: [PATCH 1/6] linstor: incremental snapshots on secondary storage via
 qemu-img rebase diff

Linstor volume snapshots were always backed up to NFS secondary storage as
full qcow2 copies, so every snapshot consumed full-volume space. The existing
KVM incremental snapshot feature (kvm.incremental.snapshot) is limited to
file-based primary storage because it relies on libvirt checkpoints backed by
qcow2-persistent dirty bitmaps, which do not exist for raw DRBD devices.

Add incremental snapshots for Linstor through a content diff instead: the
agent creates an empty qcow2 overlay backed by the raw snapshot device (with
an explicit virtual size, clipping the trailing DRBD metadata), then
safe-rebases it onto the previous snapshot qcow2 on secondary storage. qemu
writes only the clusters in which the two backing files differ, yielding a
delta qcow2 chained to its parent by a relative backing name (portable across
mount points). No bitmaps, no libvirt version gates, no LINSTOR snapshot
retained on primary, and it works for running and stopped VMs alike. Because
the diff is content-based, any ready previous snapshot is a correct diff base,
which makes fallback-to-full trivial.

The feature is gated by the existing cluster-scoped kvm.incremental.snapshot
setting plus the pool being Linstor; chain length is capped by
snapshot.delta.max and the end_of_chain rotation. Encrypted volumes stay
full-only (a rebase would need the LUKS secret for delta and backing file).

- SnapshotDataStoreDaoImpl.findParent: parent selection for Linstor volumes
  (non-encrypted) uses the plain search instead of requiring
  kvm_checkpoint_path, so chains can start; ObjectInDataStoreManagerImpl then
  links parent_snapshot_id on the image store ref as usual.
- SnapshotObject.getCorrectIncrementalParent: fall back to the plain parent
  instead of throwing when no ref carries a checkpoint path. The TO
  constructor still early-returns for such parents, keeping parents[] null so
  the file-based agent paths are unaffected, while getParent() correctly
  drives convert-before-use and chain-aware deletion.
- SnapshotManagerImpl: generalize the three checkpoint-keyed predicates
  (extract URL flattening, endLastChainIfNeeded, copySnapshotChainToZone)
  through isIncrementalChainRef(), which also recognizes parent-linked KVM
  chains; XenServer per-member chain copy semantics are preserved.
- Linstor driver: copySnapshot passes fullSnapshot=false and the parent
  install path when the dest ref has a ready, same-sized parent on the same
  store; the parent link is cleared whenever a full copy was made (encrypted,
  parent missing/unready/resized or agent-side fallback) so the DB chain
  always matches what is on disk.
- LinstorBackupSnapshotCommandWrapper: new incremental branch (qemu-img
  create -F raw -b <snapdev> + safe rebase onto the parent + rebase -u to the
  relative backing name); falls back to the full convert path if the parent
  file is missing. The answer TO reports parentSnapshotPath so
  SnapshotObject.processEvent keeps the parent link only for real deltas.

Restore, volume/template-from-snapshot, extract and cross-zone copy all reuse
the existing machinery: qemu-img follows the relative backing chain on the
same mount, and chain members are flattened via ConvertSnapshotCommand before
standalone use. Delete ordering is handled by the generic parent-link logic
(children hide their parents, leaf-first physical deletion).
---
 .../datastore/db/SnapshotDataStoreDao.java    |   7 ++
 .../db/SnapshotDataStoreDaoImpl.java          |  36 ++++++-
 .../db/SnapshotDataStoreDaoImplTest.java      |  61 +++++++++++
 .../storage/snapshot/SnapshotObject.java      |  11 +-
 .../storage/snapshot/SnapshotObjectTest.java  | 101 ++++++++++++++++++
 .../ObjectInDataStoreManagerImpl.java         |   5 +-
 plugins/storage/volume/linstor/CHANGELOG.md   |   6 ++
 .../storage/LinstorBackupSnapshotCommand.java |   7 ++
 .../LinstorBackupSnapshotCommandWrapper.java  |  74 ++++++++++++-
 .../LinstorPrimaryDataStoreDriverImpl.java    |  97 ++++++++++++++---
 ...LinstorPrimaryDataStoreDriverImplTest.java |  93 ++++++++++++++++
 .../storage/snapshot/SnapshotManagerImpl.java |  16 ++-
 .../snapshot/SnapshotManagerImplTest.java     |  30 ++++++
 13 files changed, 520 insertions(+), 24 deletions(-)
 create mode 100644 engine/storage/snapshot/src/test/java/org/apache/cloudstack/storage/snapshot/SnapshotObjectTest.java

diff --git a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java
index 3329983d711e..e81bf7305fd6 100644
--- a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java
+++ b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java
@@ -49,6 +49,13 @@ public interface SnapshotDataStoreDao extends GenericDao<SnapshotDataStoreVO, Lo
 
     SnapshotDataStoreVO findParent(DataStoreRole role, Long storeId, Long zoneId, Long volumeId, boolean kvmIncrementalSnapshot, Hypervisor.HypervisorType hypervisorType);
 
+    /**
+     * Whether snapshots of this volume are chained on secondary storage through a content diff against
+     * the parent snapshot file (Linstor) instead of qemu checkpoints. Such chains live purely on the
+     * image store, so checkpoint-oriented parent handling must not be applied to them.
+     */
+    boolean usesContentBasedChain(long volumeId);
+
     SnapshotDataStoreVO findBySnapshotIdAndDataStoreRoleAndState(long snapshotId, DataStoreRole role, ObjectInDataStoreStateMachine.State state);
 
     List<SnapshotDataStoreVO> listBySnapshotIdAndDataStoreRoleAndStateIn(long snapshotId, DataStoreRole role, ObjectInDataStoreStateMachine.State... state);
diff --git a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
index 8b7a2b78de7e..924805017862 100644
--- a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
+++ b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
@@ -19,8 +19,11 @@
 import com.cloud.hypervisor.Hypervisor;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.SnapshotVO;
+import com.cloud.storage.Storage;
 import com.cloud.storage.VMTemplateStorageResourceAssoc;
+import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.SnapshotDao;
+import com.cloud.storage.dao.VolumeDao;
 import com.cloud.utils.db.DB;
 import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.GenericDaoBase;
@@ -88,6 +91,12 @@ public class SnapshotDataStoreDaoImpl extends GenericDaoBase<SnapshotDataStoreVO
     @Inject
     protected ImageStoreDao imageStoreDao;
 
+    @Inject
+    protected VolumeDao volumeDao;
+
+    @Inject
+    protected PrimaryDataStoreDao storagePoolDao;
+
     private static final String FIND_OLDEST_OR_LATEST_SNAPSHOT = "select store_id, store_role, snapshot_id from cloud.snapshot_store_ref where " +
             " store_role = ? and volume_id = ? and state = 'Ready'" +
             " order by created %s " +
@@ -352,8 +361,15 @@ public SnapshotDataStoreVO findParent(DataStoreRole role, Long storeId, Long zon
             return null;
         }
 
+        boolean contentBasedChain = kvmIncrementalSnapshot && Hypervisor.HypervisorType.KVM.equals(hypervisorType) && usesContentBasedChain(volumeId);
+        if (contentBasedChain && (role == null || !role.isImageStore())) {
+            logger.trace("Content-based snapshot chains only exist on the image store. Returning null as parent for volume [{}] and role [{}].", volumeId, role);
+            return null;
+        }
+        boolean checkpointBasedChain = kvmIncrementalSnapshot && Hypervisor.HypervisorType.KVM.equals(hypervisorType) && !contentBasedChain;
+
         SearchCriteria<SnapshotDataStoreVO> sc;
-        if (kvmIncrementalSnapshot && Hypervisor.HypervisorType.KVM.equals(hypervisorType)) {
+        if (checkpointBasedChain) {
             sc = searchFilteringStoreIdInVolumeIdEqStoreRoleEqStateEqKVMCheckpointNotNull.create();
         } else {
             sc = searchFilteringStoreIdInVolumeIdEqStoreRoleEqStateEq.create();
@@ -379,13 +395,29 @@ public SnapshotDataStoreVO findParent(DataStoreRole role, Long storeId, Long zon
 
         SnapshotDataStoreVO parent = snapshotList.get(0);
 
-        if (kvmIncrementalSnapshot && parent.getKvmCheckpointPath() == null && Hypervisor.HypervisorType.KVM.equals(hypervisorType)) {
+        if (checkpointBasedChain && parent.getKvmCheckpointPath() == null) {
             return null;
         }
 
         return parent;
     }
 
+    /**
+     * Volumes on Linstor primary storage chain incremental snapshots on secondary storage through a
+     * content diff (qemu-img rebase) against the parent snapshot file instead of qemu checkpoints, so
+     * parent selection must not require a checkpoint path. Encrypted volumes are excluded as they are
+     * always backed up as full copies (a rebase would need the LUKS secret for delta and backing file).
+     */
+    @Override
+    public boolean usesContentBasedChain(long volumeId) {
+        VolumeVO volume = volumeDao.findByIdIncludingRemoved(volumeId);
+        if (volume == null || volume.getPoolId() == null || volume.getPassphraseId() != null) {
+            return false;
+        }
+        StoragePoolVO pool = storagePoolDao.findById(volume.getPoolId());
+        return pool != null && Storage.StoragePoolType.Linstor.equals(pool.getPoolType());
+    }
+
     @Override
     public SnapshotDataStoreVO findBySnapshotIdAndDataStoreRoleAndState(long snapshotId, DataStoreRole role, State state) {
         SearchCriteria<SnapshotDataStoreVO> sc = createSearchCriteriaBySnapshotIdAndStoreRole(snapshotId, role);
diff --git a/engine/schema/src/test/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImplTest.java b/engine/schema/src/test/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImplTest.java
index 85240ab4a058..2146ac6cc770 100644
--- a/engine/schema/src/test/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImplTest.java
+++ b/engine/schema/src/test/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImplTest.java
@@ -20,13 +20,18 @@
 import java.util.List;
 
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.Spy;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.mockito.stubbing.Answer;
 
+import com.cloud.storage.Storage;
+import com.cloud.storage.VolumeVO;
+import com.cloud.storage.dao.VolumeDao;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 
@@ -36,6 +41,18 @@ public class SnapshotDataStoreDaoImplTest {
     @Spy
     SnapshotDataStoreDaoImpl snapshotDataStoreDaoImplSpy;
 
+    @Mock
+    VolumeDao volumeDaoMock;
+
+    @Mock
+    PrimaryDataStoreDao storagePoolDaoMock;
+
+    @Before
+    public void setUp() {
+        snapshotDataStoreDaoImplSpy.volumeDao = volumeDaoMock;
+        snapshotDataStoreDaoImplSpy.storagePoolDao = storagePoolDaoMock;
+    }
+
     @Test
     public void testExpungeByVmListNoVms() {
         Assert.assertEquals(0, snapshotDataStoreDaoImplSpy.expungeBySnapshotList(
@@ -64,4 +81,48 @@ public void testExpungeByVmList() {
         Mockito.verify(snapshotDataStoreDaoImplSpy, Mockito.times(1))
                 .batchExpunge(sc, batchSize);
     }
+
+    private VolumeVO mockVolume(Long poolId, Long passphraseId) {
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        Mockito.when(volume.getPoolId()).thenReturn(poolId);
+        Mockito.lenient().when(volume.getPassphraseId()).thenReturn(passphraseId);
+        Mockito.when(volumeDaoMock.findByIdIncludingRemoved(1L)).thenReturn(volume);
+        return volume;
+    }
+
+    @Test
+    public void testUsesContentBasedChainVolumeNotFound() {
+        Mockito.when(volumeDaoMock.findByIdIncludingRemoved(1L)).thenReturn(null);
+        Assert.assertFalse(snapshotDataStoreDaoImplSpy.usesContentBasedChain(1L));
+    }
+
+    @Test
+    public void testUsesContentBasedChainNoPool() {
+        mockVolume(null, null);
+        Assert.assertFalse(snapshotDataStoreDaoImplSpy.usesContentBasedChain(1L));
+    }
+
+    @Test
+    public void testUsesContentBasedChainEncryptedVolume() {
+        mockVolume(3L, 7L);
+        Assert.assertFalse(snapshotDataStoreDaoImplSpy.usesContentBasedChain(1L));
+    }
+
+    @Test
+    public void testUsesContentBasedChainLinstorPool() {
+        mockVolume(3L, null);
+        StoragePoolVO pool = Mockito.mock(StoragePoolVO.class);
+        Mockito.when(pool.getPoolType()).thenReturn(Storage.StoragePoolType.Linstor);
+        Mockito.when(storagePoolDaoMock.findById(3L)).thenReturn(pool);
+        Assert.assertTrue(snapshotDataStoreDaoImplSpy.usesContentBasedChain(1L));
+    }
+
+    @Test
+    public void testUsesContentBasedChainNonLinstorPool() {
+        mockVolume(3L, null);
+        StoragePoolVO pool = Mockito.mock(StoragePoolVO.class);
+        Mockito.when(pool.getPoolType()).thenReturn(Storage.StoragePoolType.NetworkFilesystem);
+        Mockito.when(storagePoolDaoMock.findById(3L)).thenReturn(pool);
+        Assert.assertFalse(snapshotDataStoreDaoImplSpy.usesContentBasedChain(1L));
+    }
 }
diff --git a/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/SnapshotObject.java b/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/SnapshotObject.java
index 6a8bbd93ca4e..213014db88f5 100644
--- a/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/SnapshotObject.java
+++ b/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/SnapshotObject.java
@@ -129,7 +129,9 @@ public SnapshotInfo getParent() {
     }
 
     /**
-     * Returns the snapshotInfo of the passed snapshot parentId. Will search for the snapshot reference which has a checkpoint path. If none is found, throws an exception.
+     * Returns the snapshotInfo of the passed snapshot parentId. Will search for the snapshot reference which has a checkpoint path.
+     * If none is found, returns the plain parent on this snapshot's store: KVM snapshots may also be chained without checkpoints,
+     * e.g. Linstor chains deltas through a content diff against the parent snapshot file on secondary storage.
      * */
     protected SnapshotInfo getCorrectIncrementalParent(long parentId) {
         List<SnapshotDataStoreVO> parentSnapshotDatastoreVos = snapshotStoreDao.findBySnapshotId(parentId);
@@ -141,8 +143,11 @@ protected SnapshotInfo getCorrectIncrementalParent(long parentId) {
         logger.debug("Found parent snapshot references {}, will filter to just one.", parentSnapshotDatastoreVos);
 
         SnapshotDataStoreVO parent = parentSnapshotDatastoreVos.stream().filter(snapshotDataStoreVO -> snapshotDataStoreVO.getKvmCheckpointPath() != null)
-                .findFirst().
-                orElseThrow(() -> new CloudRuntimeException(String.format("Could not find snapshot parent with id [%s]. None of the records have a checkpoint path.", parentId)));
+                .findFirst().orElse(null);
+
+        if (parent == null) {
+            return snapshotFactory.getSnapshot(parentId, store);
+        }
 
         SnapshotInfo snapshotInfo = snapshotFactory.getSnapshot(parentId, parent.getDataStoreId(), parent.getRole());
         snapshotInfo.setKvmIncrementalSnapshot(parent.getKvmCheckpointPath() != null);
diff --git a/engine/storage/snapshot/src/test/java/org/apache/cloudstack/storage/snapshot/SnapshotObjectTest.java b/engine/storage/snapshot/src/test/java/org/apache/cloudstack/storage/snapshot/SnapshotObjectTest.java
new file mode 100644
index 000000000000..8b4e6b07dc16
--- /dev/null
+++ b/engine/storage/snapshot/src/test/java/org/apache/cloudstack/storage/snapshot/SnapshotObjectTest.java
@@ -0,0 +1,101 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.cloudstack.storage.snapshot;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
+import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotDataFactory;
+import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import com.cloud.storage.DataStoreRole;
+import com.cloud.storage.SnapshotVO;
+
+@RunWith(MockitoJUnitRunner.class)
+public class SnapshotObjectTest {
+
+    private static final long PARENT_SNAPSHOT_ID = 2L;
+
+    @Mock
+    SnapshotDataStoreDao snapshotStoreDao;
+
+    @Mock
+    SnapshotDataFactory snapshotFactory;
+
+    @Mock
+    DataStore store;
+
+    @Mock
+    SnapshotVO snapshotVO;
+
+    SnapshotObject snapshotObject;
+
+    @Before
+    public void setUp() {
+        snapshotObject = new SnapshotObject();
+        snapshotObject.configure(snapshotVO, store);
+        snapshotObject.snapshotStoreDao = snapshotStoreDao;
+        snapshotObject.snapshotFactory = snapshotFactory;
+    }
+
+    @Test
+    public void testGetCorrectIncrementalParentNoRefsReturnsNull() {
+        Mockito.when(snapshotStoreDao.findBySnapshotId(PARENT_SNAPSHOT_ID)).thenReturn(Collections.emptyList());
+
+        Assert.assertNull(snapshotObject.getCorrectIncrementalParent(PARENT_SNAPSHOT_ID));
+    }
+
+    @Test
+    public void testGetCorrectIncrementalParentPrefersCheckpointBearingRef() {
+        SnapshotDataStoreVO refWithoutCheckpoint = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(refWithoutCheckpoint.getKvmCheckpointPath()).thenReturn(null);
+        SnapshotDataStoreVO refWithCheckpoint = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(refWithCheckpoint.getKvmCheckpointPath()).thenReturn("checkpoints/2/5/uuid");
+        Mockito.when(refWithCheckpoint.getDataStoreId()).thenReturn(5L);
+        Mockito.when(refWithCheckpoint.getRole()).thenReturn(DataStoreRole.Image);
+        Mockito.when(snapshotStoreDao.findBySnapshotId(PARENT_SNAPSHOT_ID)).thenReturn(List.of(refWithoutCheckpoint, refWithCheckpoint));
+
+        SnapshotInfo parentInfo = Mockito.mock(SnapshotInfo.class);
+        Mockito.when(snapshotFactory.getSnapshot(PARENT_SNAPSHOT_ID, 5L, DataStoreRole.Image)).thenReturn(parentInfo);
+
+        Assert.assertEquals(parentInfo, snapshotObject.getCorrectIncrementalParent(PARENT_SNAPSHOT_ID));
+        Mockito.verify(parentInfo).setKvmIncrementalSnapshot(true);
+    }
+
+    @Test
+    public void testGetCorrectIncrementalParentFallsBackToPlainParentWithoutCheckpointRefs() {
+        SnapshotDataStoreVO refWithoutCheckpoint = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(refWithoutCheckpoint.getKvmCheckpointPath()).thenReturn(null);
+        Mockito.when(snapshotStoreDao.findBySnapshotId(PARENT_SNAPSHOT_ID)).thenReturn(List.of(refWithoutCheckpoint));
+
+        SnapshotInfo parentInfo = Mockito.mock(SnapshotInfo.class);
+        Mockito.when(snapshotFactory.getSnapshot(PARENT_SNAPSHOT_ID, store)).thenReturn(parentInfo);
+
+        Assert.assertEquals(parentInfo, snapshotObject.getCorrectIncrementalParent(PARENT_SNAPSHOT_ID));
+        Mockito.verify(parentInfo, Mockito.never()).setKvmIncrementalSnapshot(Mockito.anyBoolean());
+    }
+}
diff --git a/engine/storage/src/main/java/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java b/engine/storage/src/main/java/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java
index d03be9c4d294..dbf1f9de0bd3 100644
--- a/engine/storage/src/main/java/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java
+++ b/engine/storage/src/main/java/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java
@@ -222,7 +222,10 @@ public DataObject create(DataObject obj, DataStore dataStore) {
     private SnapshotDataStoreVO findParent(DataStore dataStore, Long clusterId, SnapshotInfo snapshotInfo) {
         boolean kvmIncrementalSnapshot = SnapshotManager.kvmIncrementalSnapshot.valueIn(clusterId);
         SnapshotDataStoreVO snapshotDataStoreVO;
-        if (Hypervisor.HypervisorType.KVM.equals(snapshotInfo.getHypervisorType()) && kvmIncrementalSnapshot) {
+        // content-based chains (Linstor) live purely on the image store; the cross-role checkpoint
+        // handling below (with its end-of-chain marking) must not be applied to their parents
+        if (Hypervisor.HypervisorType.KVM.equals(snapshotInfo.getHypervisorType()) && kvmIncrementalSnapshot
+                && !snapshotDataStoreDao.usesContentBasedChain(snapshotInfo.getVolumeId())) {
             snapshotDataStoreVO = snapshotDataStoreDao.findParent(null, null, null, snapshotInfo.getVolumeId(),
                     kvmIncrementalSnapshot, snapshotInfo.getHypervisorType());
             snapshotDataStoreVO = returnNullIfNotOnSameTypeOfStoreRole(snapshotInfo, snapshotDataStoreVO);
diff --git a/plugins/storage/volume/linstor/CHANGELOG.md b/plugins/storage/volume/linstor/CHANGELOG.md
index a6ab050b090e..43f13a1af52f 100644
--- a/plugins/storage/volume/linstor/CHANGELOG.md
+++ b/plugins/storage/volume/linstor/CHANGELOG.md
@@ -24,6 +24,12 @@ All notable changes to Linstor CloudStack plugin will be documented in this file
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2026-07-30]
+
+### Added
+
+- Support for incremental snapshots on secondary storage backuped snapshots
+
 ## [2026-06-24]
 
 ### Fixed
diff --git a/plugins/storage/volume/linstor/src/main/java/com/cloud/api/storage/LinstorBackupSnapshotCommand.java b/plugins/storage/volume/linstor/src/main/java/com/cloud/api/storage/LinstorBackupSnapshotCommand.java
index 8d887dbba21a..069368bd7640 100644
--- a/plugins/storage/volume/linstor/src/main/java/com/cloud/api/storage/LinstorBackupSnapshotCommand.java
+++ b/plugins/storage/volume/linstor/src/main/java/com/cloud/api/storage/LinstorBackupSnapshotCommand.java
@@ -21,6 +21,13 @@
 
 public class LinstorBackupSnapshotCommand extends CopyCommand
 {
+    /**
+     * Option holding the secondary storage install path of the parent snapshot qcow2. When set (and
+     * fullSnapshot=false), the agent writes an incremental backup: a qcow2 containing only the blocks
+     * that differ from the parent, with the parent as its backing file.
+     */
+    public static final String OPTION_PARENT_PATH = "parentPath";
+
     public LinstorBackupSnapshotCommand(DataTO srcData, DataTO destData, int timeout, boolean executeInSequence)
     {
         super(srcData, destData, timeout, executeInSequence);
diff --git a/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java b/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java
index c111d320cb4e..b56d5ebe315a 100644
--- a/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java
+++ b/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java
@@ -121,6 +121,49 @@ private String convertImageToQCow2(
         return dstPath;
     }
 
+    /**
+     * Writes an incremental backup: a qcow2 on secondary storage containing only the blocks of the
+     * snapshot device that differ from the parent snapshot qcow2, with the parent as backing file.
+     * The overlay starts out backed by the raw snapshot device itself; the safe-mode rebase onto the
+     * parent then copies every cluster in which the two backing files differ into the overlay. The
+     * explicit virtual size clips the DRBD metadata trailing the storage snapshot device.
+     */
+    private String createIncrementalQCow2(
+        final String srcPath,
+        final SnapshotObjectTO dst,
+        final KVMStoragePool secondaryPool,
+        final File parentFile,
+        final long netSize,
+        int waitMilliSeconds)
+        throws LibvirtException, QemuImgException, IOException
+    {
+        final String dstDir = secondaryPool.getLocalPath() + File.separator + dst.getPath();
+        FileUtils.forceMkdir(new File(dstDir));
+        final String dstPath = dstDir + File.separator + dst.getName();
+
+        final Script createOverlay = new Script("qemu-img", Duration.millis(waitMilliSeconds));
+        createOverlay.add("create", "-f", "qcow2", "-F", "raw", "-b", srcPath, dstPath, String.valueOf(netSize));
+        final String createResult = createOverlay.execute();
+        if (createResult != null) {
+            throw new QemuImgException("Unable to create qcow2 overlay of " + srcPath + ": " + createResult);
+        }
+
+        try {
+            final QemuImg qemu = new QemuImg(waitMilliSeconds);
+            final QemuImgFile dstFile = new QemuImgFile(dstPath, QemuImg.PhysicalDiskFormat.QCOW2);
+            qemu.rebase(dstFile, new QemuImgFile(parentFile.getAbsolutePath(), QemuImg.PhysicalDiskFormat.QCOW2),
+                QemuImg.PhysicalDiskFormat.QCOW2.toString(), true);
+            // metadata-only rewrite to a relative backing name, so the chain stays valid on any mount point
+            qemu.rebase(dstFile, new QemuImgFile(parentFile.getName(), QemuImg.PhysicalDiskFormat.QCOW2),
+                QemuImg.PhysicalDiskFormat.QCOW2.toString(), false);
+        } catch (final QemuImgException | LibvirtException e) {
+            FileUtils.deleteQuietly(new File(dstPath));
+            throw e;
+        }
+        LOGGER.info("Incremental backup snapshot '{}' to '{}' (parent '{}')", srcPath, dstPath, parentFile.getName());
+        return dstPath;
+    }
+
     private SnapshotObjectTO setCorrectSnapshotSize(final SnapshotObjectTO dst, final String dstPath) {
         final File snapFile = new File(dstPath);
         long size;
@@ -176,12 +219,33 @@ public CopyCmdAnswer execute(LinstorBackupSnapshotCommand cmd, LibvirtComputingR
             final byte[] passphrase = src.getVolume() != null ? src.getVolume().getPassphrase() : null;
             final boolean encrypted = passphrase != null && passphrase.length > 0;
 
-            String dstPath = convertImageToQCow2(srcPath, dst, secondaryPool, passphrase, cmd.getWaitInMillSeconds());
+            final Map<String, String> options = cmd.getOptions();
+            final String parentInstallPath = options != null ?
+                options.get(LinstorBackupSnapshotCommand.OPTION_PARENT_PATH) : null;
+
+            boolean incremental = false;
+            String dstPath = null;
+            if (!encrypted && parentInstallPath != null && src.getVolume() != null) {
+                final File parentFile = new File(secondaryPool.getLocalPath() + File.separator + parentInstallPath);
+                if (parentFile.isFile()) {
+                    dstPath = createIncrementalQCow2(
+                        srcPath, dst, secondaryPool, parentFile, src.getVolume().getSize(), cmd.getWaitInMillSeconds());
+                    incremental = true;
+                } else {
+                    LOGGER.warn("Parent snapshot file '{}' missing on secondary storage, taking a full backup instead",
+                        parentFile.getAbsolutePath());
+                }
+            }
+
+            if (dstPath == null) {
+                dstPath = convertImageToQCow2(srcPath, dst, secondaryPool, passphrase, cmd.getWaitInMillSeconds());
+            }
 
-            if (!encrypted) {
+            if (!encrypted && !incremental) {
                 // resize to real volume size, cutting of drbd metadata
                 // For encrypted volumes the source is the decrypted DRBD device (already net-sized,
                 // no drbd metadata to cut); shrinking an encrypted qcow2 would also need the secret.
+                // Incremental backups are created with the net size already, nothing to cut there.
                 String result = qemuShrink(dstPath, src.getVolume().getSize(), cmd.getWaitInMillSeconds());
                 if (result != null) {
                     return new CopyCmdAnswer("qemu-img shrink failed: " + result);
@@ -190,6 +254,12 @@ public CopyCmdAnswer execute(LinstorBackupSnapshotCommand cmd, LibvirtComputingR
             }
 
             SnapshotObjectTO snapshot = setCorrectSnapshotSize(dst, dstPath);
+            // tells the management server whether the file is a delta qcow2 backed by the parent
+            // snapshot or a standalone full copy, so it can fix up the chain bookkeeping:
+            // SnapshotObject.processEvent drops the store ref's parent link when
+            // parentSnapshotPath is null, and the driver clears it when the answer is not incremental
+            snapshot.setKvmIncrementalSnapshot(incremental);
+            snapshot.setParentSnapshotPath(incremental ? parentInstallPath : null);
             LOGGER.info("Actual file size for '{}' is {}", dstPath, snapshot.getPhysicalSize());
             return new CopyCmdAnswer(snapshot);
         } catch (final Exception e) {
diff --git a/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java b/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java
index c3b4e73ead03..2d12744024c9 100644
--- a/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java
+++ b/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java
@@ -103,6 +103,8 @@
 import org.apache.cloudstack.storage.command.CopyCommand;
 import org.apache.cloudstack.storage.command.CreateObjectAnswer;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.datastore.util.LinstorConfigurationManager;
 import org.apache.cloudstack.storage.datastore.util.LinstorUtil;
@@ -124,6 +126,7 @@ public class LinstorPrimaryDataStoreDriverImpl implements PrimaryDataStoreDriver
     @Inject private VMTemplatePoolDao _vmTemplatePoolDao;
     @Inject private SnapshotDao _snapshotDao;
     @Inject private SnapshotDetailsDao _snapshotDetailsDao;
+    @Inject private SnapshotDataStoreDao _snapshotStoreDao;
     @Inject private StorageManager _storageMgr;
     @Inject
     ConfigurationDao _configDao;
@@ -1071,15 +1074,30 @@ protected Answer copySnapshot(DataObject srcData, DataObject destData) {
             value, Integer.parseInt(Config.BackupSnapshotWait.getDefaultValue()));
 
         SnapshotObject snapshotObject = (SnapshotObject)srcData;
-        Boolean snapshotFullBackup = snapshotObject.getFullBackup();
         final StoragePoolVO pool = _storagePoolDao.findById(srcData.getDataStore().getId());
         final DevelopersApi api = getLinstorAPI(pool);
-        boolean fullSnapshot = true;
-        if (snapshotFullBackup != null) {
-            fullSnapshot = snapshotFullBackup;
-        }
+
+        // For encrypted volumes Linstor adds a LUKS layer (DRBD -> LUKS -> STORAGE). The storage
+        // layer snapshot device (getSnapshotPath) therefore only exposes the raw LUKS ciphertext,
+        // while restore writes onto the decrypted DRBD device (/dev/drbd/by-res/.../0). Backing up
+        // the ciphertext and writing it back to the decrypted layer corrupts the volume (and the
+        // shrink to the net volume size would even truncate the ciphertext). So for encrypted
+        // volumes we never read the storage snapshot directly: restore the snapshot into a temporary
+        // resource and back up its decrypted DRBD device instead, symmetric to the restore path.
+        final boolean encrypted = snapshotObject.getBaseVolume().getPassphraseId() != null;
+
+        SnapshotDataStoreVO destRef = _snapshotStoreDao.findByStoreSnapshot(
+            destData.getDataStore().getRole(), destData.getDataStore().getId(), destData.getId());
+        // encrypted volumes are always backed up as full copies: an incremental rebase would need
+        // the LUKS secret for both the delta and the backing file
+        String parentPath = encrypted ? null : getIncrementalParentPath(destRef);
+        boolean fullSnapshot = parentPath == null;
+
         Map<String, String> options = new HashMap<>();
         options.put("fullSnapshot", fullSnapshot + "");
+        if (parentPath != null) {
+            options.put(LinstorBackupSnapshotCommand.OPTION_PARENT_PATH, parentPath);
+        }
         options.put(SnapshotInfo.BackupSnapshotAfterTakingSnapshot.key(),
             String.valueOf(SnapshotInfo.BackupSnapshotAfterTakingSnapshot.value()));
         options.put("volumeSize", snapshotObject.getBaseVolume().getSize() + "");
@@ -1095,14 +1113,6 @@ protected Answer copySnapshot(DataObject srcData, DataObject destData) {
                 VirtualMachineManager.ExecuteInSequence.value());
             cmd.setOptions(options);
 
-            // For encrypted volumes Linstor adds a LUKS layer (DRBD -> LUKS -> STORAGE). The storage
-            // layer snapshot device (getSnapshotPath) therefore only exposes the raw LUKS ciphertext,
-            // while restore writes onto the decrypted DRBD device (/dev/drbd/by-res/.../0). Backing up
-            // the ciphertext and writing it back to the decrypted layer corrupts the volume (and the
-            // shrink to the net volume size would even truncate the ciphertext). So for encrypted
-            // volumes we never read the storage snapshot directly: restore the snapshot into a temporary
-            // resource and back up its decrypted DRBD device instead, symmetric to the restore path.
-            final boolean encrypted = snapshotObject.getBaseVolume().getPassphraseId() != null;
             Optional<RemoteHostEndPoint> optEP = encrypted ?
                 Optional.empty() : getDiskfullEP(api, pool, rscName);
             Answer answer;
@@ -1113,6 +1123,9 @@ protected Answer copySnapshot(DataObject srcData, DataObject destData) {
                     encrypted);
                 answer = copyFromTemporaryResource(api, pool, rscName, snapshotName, snapshotObject, cmd);
             }
+            if (answer != null && answer.getResult()) {
+                clearChainParentIfFullCopy(destRef, answer);
+            }
             return answer;
         } catch (Exception e) {
             logger.debug("copy snapshot failed, please cleanup snapshot manually: ", e);
@@ -1121,6 +1134,64 @@ protected Answer copySnapshot(DataObject srcData, DataObject destData) {
 
     }
 
+    /**
+     * Returns the secondary storage install path of the parent snapshot to build an incremental
+     * (content-diff) backup against, or null if a full backup has to be taken. The parent link on the
+     * destination store ref was set at allocation time (honoring end_of_chain and kvm.incremental.snapshot).
+     */
+    protected String getIncrementalParentPath(SnapshotDataStoreVO destRef) {
+        if (destRef == null || destRef.getParentSnapshotId() <= 0) {
+            return null;
+        }
+        if (!LinstorConfigurationManager.BackupSnapshots.value()) {
+            // snapshots-kept-on-primary mode: copies to secondary are only temporary (template from
+            // snapshot, extract, cross-zone copy) and their refs get expunged after use. A delta
+            // against such a parent would be left with a dangling backing file, so always copy full.
+            logger.debug("{} is disabled, taking full backup of snapshot {}",
+                LinstorConfigurationManager.BackupSnapshots.key(), destRef.getSnapshotId());
+            return null;
+        }
+        SnapshotDataStoreVO parentRef = _snapshotStoreDao.findByStoreSnapshot(
+            destRef.getRole(), destRef.getDataStoreId(), destRef.getParentSnapshotId());
+        if (parentRef == null || parentRef.getInstallPath() == null
+                || parentRef.getState() != ObjectInDataStoreStateMachine.State.Ready) {
+            logger.debug("Parent snapshot {} of snapshot {} not ready on store, taking full backup instead",
+                destRef.getParentSnapshotId(), destRef.getSnapshotId());
+            return null;
+        }
+        if (parentRef.getSize() != destRef.getSize()) {
+            // volume was resized between the two snapshots; the delta would need to cover the size
+            // change through beyond-EOF backing semantics, take a full backup instead
+            logger.debug("Parent snapshot {} has a different volume size than snapshot {} ({} != {}), taking full backup instead",
+                destRef.getParentSnapshotId(), destRef.getSnapshotId(), parentRef.getSize(), destRef.getSize());
+            return null;
+        }
+        return parentRef.getInstallPath();
+    }
+
+    /**
+     * The parent link is only valid if the backup really was written as a delta of the parent file.
+     * If a full copy was made instead (encrypted volume, missing/unready parent or agent-side
+     * fallback), drop the link so chain bookkeeping (delete ordering, chain length, flattening)
+     * matches what is on disk.
+     */
+    protected void clearChainParentIfFullCopy(SnapshotDataStoreVO destRef, Answer answer) {
+        if (destRef == null || destRef.getParentSnapshotId() <= 0) {
+            return;
+        }
+        boolean deltaWritten = false;
+        if (answer instanceof CopyCmdAnswer) {
+            DataTO newData = ((CopyCmdAnswer) answer).getNewData();
+            deltaWritten = newData instanceof SnapshotObjectTO && ((SnapshotObjectTO) newData).isKvmIncrementalSnapshot();
+        }
+        if (!deltaWritten) {
+            logger.debug("Snapshot {} was backed up as a full copy, clearing its chain parent {}",
+                destRef.getSnapshotId(), destRef.getParentSnapshotId());
+            destRef.setParentSnapshotId(0);
+            _snapshotStoreDao.update(destRef.getId(), destRef);
+        }
+    }
+
     @Override
     public void copyAsync(DataObject srcData, DataObject destData, Host destHost, AsyncCompletionCallback<CopyCommandResult> callback)
     {
diff --git a/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImplTest.java b/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImplTest.java
index 4653cfa358b0..69a6e90128f7 100644
--- a/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImplTest.java
+++ b/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImplTest.java
@@ -26,14 +26,23 @@
 import java.util.Collections;
 import java.util.List;
 
+import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreStateMachine;
+import org.apache.cloudstack.storage.command.CopyCmdAnswer;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
 import org.apache.cloudstack.storage.datastore.util.LinstorUtil;
+import org.apache.cloudstack.storage.to.SnapshotObjectTO;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
+import com.cloud.storage.DataStoreRole;
+
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -42,6 +51,9 @@ public class LinstorPrimaryDataStoreDriverImplTest {
 
     private DevelopersApi api;
 
+    @Mock
+    private SnapshotDataStoreDao snapshotStoreDao;
+
     @InjectMocks
     private LinstorPrimaryDataStoreDriverImpl linstorPrimaryDataStoreDriver;
 
@@ -50,6 +62,87 @@ public void setUp() {
         api = mock(DevelopersApi.class);
     }
 
+    private SnapshotDataStoreVO mockDestRef(long parentSnapshotId, long size) {
+        SnapshotDataStoreVO destRef = mock(SnapshotDataStoreVO.class);
+        when(destRef.getParentSnapshotId()).thenReturn(parentSnapshotId);
+        if (parentSnapshotId > 0) {
+            Mockito.lenient().when(destRef.getRole()).thenReturn(DataStoreRole.Image);
+            Mockito.lenient().when(destRef.getDataStoreId()).thenReturn(10L);
+            Mockito.lenient().when(destRef.getSize()).thenReturn(size);
+        }
+        return destRef;
+    }
+
+    private SnapshotDataStoreVO mockParentRef(String installPath, ObjectInDataStoreStateMachine.State state, long size) {
+        SnapshotDataStoreVO parentRef = mock(SnapshotDataStoreVO.class);
+        Mockito.lenient().when(parentRef.getInstallPath()).thenReturn(installPath);
+        Mockito.lenient().when(parentRef.getState()).thenReturn(state);
+        Mockito.lenient().when(parentRef.getSize()).thenReturn(size);
+        when(snapshotStoreDao.findByStoreSnapshot(DataStoreRole.Image, 10L, 2L)).thenReturn(parentRef);
+        return parentRef;
+    }
+
+    @Test
+    public void testGetIncrementalParentPathNoDestRef() {
+        Assert.assertNull(linstorPrimaryDataStoreDriver.getIncrementalParentPath(null));
+    }
+
+    @Test
+    public void testGetIncrementalParentPathNoParentLink() {
+        Assert.assertNull(linstorPrimaryDataStoreDriver.getIncrementalParentPath(mockDestRef(0, 100L)));
+    }
+
+    @Test
+    public void testGetIncrementalParentPathParentRefMissing() {
+        when(snapshotStoreDao.findByStoreSnapshot(DataStoreRole.Image, 10L, 2L)).thenReturn(null);
+        Assert.assertNull(linstorPrimaryDataStoreDriver.getIncrementalParentPath(mockDestRef(2L, 100L)));
+    }
+
+    @Test
+    public void testGetIncrementalParentPathParentNotReady() {
+        mockParentRef("snapshots/2/5/parent", ObjectInDataStoreStateMachine.State.Destroyed, 100L);
+        Assert.assertNull(linstorPrimaryDataStoreDriver.getIncrementalParentPath(mockDestRef(2L, 100L)));
+    }
+
+    @Test
+    public void testGetIncrementalParentPathVolumeResized() {
+        mockParentRef("snapshots/2/5/parent", ObjectInDataStoreStateMachine.State.Ready, 50L);
+        Assert.assertNull(linstorPrimaryDataStoreDriver.getIncrementalParentPath(mockDestRef(2L, 100L)));
+    }
+
+    @Test
+    public void testGetIncrementalParentPathReadyParent() {
+        mockParentRef("snapshots/2/5/parent", ObjectInDataStoreStateMachine.State.Ready, 100L);
+        Assert.assertEquals("snapshots/2/5/parent",
+            linstorPrimaryDataStoreDriver.getIncrementalParentPath(mockDestRef(2L, 100L)));
+    }
+
+    @Test
+    public void testClearChainParentIfFullCopyClearsOnFullBackup() {
+        SnapshotDataStoreVO destRef = mock(SnapshotDataStoreVO.class);
+        when(destRef.getParentSnapshotId()).thenReturn(2L);
+
+        SnapshotObjectTO to = new SnapshotObjectTO();
+        to.setKvmIncrementalSnapshot(false);
+        linstorPrimaryDataStoreDriver.clearChainParentIfFullCopy(destRef, new CopyCmdAnswer(to));
+
+        Mockito.verify(destRef).setParentSnapshotId(0);
+        Mockito.verify(snapshotStoreDao).update(Mockito.anyLong(), Mockito.eq(destRef));
+    }
+
+    @Test
+    public void testClearChainParentIfFullCopyKeepsLinkOnIncremental() {
+        SnapshotDataStoreVO destRef = mock(SnapshotDataStoreVO.class);
+        when(destRef.getParentSnapshotId()).thenReturn(2L);
+
+        SnapshotObjectTO to = new SnapshotObjectTO();
+        to.setKvmIncrementalSnapshot(true);
+        linstorPrimaryDataStoreDriver.clearChainParentIfFullCopy(destRef, new CopyCmdAnswer(to));
+
+        Mockito.verify(destRef, Mockito.never()).setParentSnapshotId(Mockito.anyLong());
+        Mockito.verify(snapshotStoreDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any());
+    }
+
     @Test
     public void testGetEncryptedLayerList() throws ApiException  {
         ResourceGroup dfltRscGrp = new ResourceGroup();
diff --git a/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java b/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
index d60bff095406..4372507c1623 100755
--- a/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
+++ b/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
@@ -597,7 +597,7 @@ public String extractSnapshot(ExtractSnapshotCmd cmd) {
 
         SnapshotInfo snapshotObject = snapshotFactory.getSnapshot(snapshotId, chosenStore);
 
-        if (snapshotDataStoreReference.getKvmCheckpointPath() != null) {
+        if (isIncrementalChainRef(snapshotDataStoreReference, snapshot.getHypervisorType())) {
             snapshotSrv.convertSnapshot(snapshotObject);
         }
 
@@ -814,7 +814,7 @@ protected void endLastChainIfNeeded(long clusterId, long volumeId) {
         SnapshotDataStoreVO snapshotDataStoreVO;
         for (int i = 1; i < volumeSnapshots.size(); i++) {
             snapshotDataStoreVO = volumeSnapshots.get(i);
-            if (snapshotDataStoreVO.getKvmCheckpointPath() != null) {
+            if (isIncrementalChainRef(snapshotDataStoreVO, HypervisorType.KVM)) {
                 if (!snapshotDataStoreVO.isEndOfChain()) {
                     logger.debug("Found snapshot reference [{}] that used to belong to a now dead snapshot chain. Will mark it as end of chain.", snapshotDataStoreVO);
                     snapshotDataStoreVO.setEndOfChain(true);
@@ -825,6 +825,16 @@ protected void endLastChainIfNeeded(long clusterId, long volumeId) {
         }
     }
 
+    /**
+     * Whether this snapshot store reference is part of a KVM incremental snapshot chain on secondary
+     * storage. File-based storage chains carry a qemu checkpoint path; Linstor chains only link deltas
+     * through the parent snapshot id (content-diff qcow2 chain). Members of either chain kind must be
+     * flattened (converted) before being used standalone.
+     */
+    protected boolean isIncrementalChainRef(SnapshotDataStoreVO ref, HypervisorType hypervisorType) {
+        return HypervisorType.KVM.equals(hypervisorType) && (ref.getKvmCheckpointPath() != null || ref.getParentSnapshotId() > 0);
+    }
+
     private void postCreateRecurringSnapshotForPolicy(long userId, long volumeId, long snapshotId, long policyId) {
         // Use count query
         SnapshotVO spstVO = _snapshotDao.findById(snapshotId);
@@ -2182,7 +2192,7 @@ private boolean copySnapshotChainToZone(SnapshotVO snapshotVO, DataStore srcSecS
         List<SnapshotDataStoreVO> snapshotChain = new ArrayList<>();
         long size = 0L;
         DataStore dstSecStore = null;
-        boolean kvmIncrementalSnapshot = currentSnap.getKvmCheckpointPath() != null;
+        boolean kvmIncrementalSnapshot = isIncrementalChainRef(currentSnap, snapshotVO.getHypervisorType());
         do {
             dstSecStore = getSnapshotZoneImageStore(currentSnap.getSnapshotId(), destZone.getId());
             if (dstSecStore != null) {
diff --git a/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java b/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java
index ff0888c184c0..0216a18929d9 100644
--- a/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java
+++ b/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java
@@ -20,6 +20,7 @@
 import com.cloud.dc.DataCenterVO;
 import com.cloud.dc.dao.DataCenterDao;
 import com.cloud.event.ActionEventUtils;
+import com.cloud.hypervisor.Hypervisor;
 import com.cloud.exception.InvalidParameterValueException;
 import com.cloud.exception.PermissionDeniedException;
 import com.cloud.exception.ResourceUnavailableException;
@@ -610,4 +611,33 @@ public void testDeleteSnapshotPoliciesManualPolicyId() {
 
         snapshotManager.deleteSnapshotPolicies(cmd);
     }
+
+    @Test
+    public void testIsIncrementalChainRefNonKvmHypervisor() {
+        SnapshotDataStoreVO ref = Mockito.mock(SnapshotDataStoreVO.class);
+        Assert.assertFalse(snapshotManager.isIncrementalChainRef(ref, Hypervisor.HypervisorType.XenServer));
+    }
+
+    @Test
+    public void testIsIncrementalChainRefKvmWithCheckpointPath() {
+        SnapshotDataStoreVO ref = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(ref.getKvmCheckpointPath()).thenReturn("checkpoints/2/5/uuid");
+        Assert.assertTrue(snapshotManager.isIncrementalChainRef(ref, Hypervisor.HypervisorType.KVM));
+    }
+
+    @Test
+    public void testIsIncrementalChainRefKvmWithParentLinkOnly() {
+        SnapshotDataStoreVO ref = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(ref.getKvmCheckpointPath()).thenReturn(null);
+        Mockito.when(ref.getParentSnapshotId()).thenReturn(2L);
+        Assert.assertTrue(snapshotManager.isIncrementalChainRef(ref, Hypervisor.HypervisorType.KVM));
+    }
+
+    @Test
+    public void testIsIncrementalChainRefKvmStandalone() {
+        SnapshotDataStoreVO ref = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(ref.getKvmCheckpointPath()).thenReturn(null);
+        Mockito.when(ref.getParentSnapshotId()).thenReturn(0L);
+        Assert.assertFalse(snapshotManager.isIncrementalChainRef(ref, Hypervisor.HypervisorType.KVM));
+    }
 }

From a11de687563394bb584d91661a0af0e335ba8a1b Mon Sep 17 00:00:00 2001
From: Rene Peinthor <rene.peinthor@linbit.com>
Date: Thu, 6 Aug 2026 09:28:52 +0200
Subject: [PATCH 2/6] linstor: address review comments on incremental snapshot
 backup

- LinstorBackupSnapshotCommandWrapper: follow coding conventions for the
  createIncrementalQCow2 signature, use KVMStoragePool.getLocalPathFor()
  instead of manual path concatenation, and create the qcow2 overlay via
  QemuImg.create() instead of a hand-rolled qemu-img Script call.
- QemuImg.create: pass the created file's format to -f instead of the
  backing file's format, so an overlay format can differ from its backing
  format (-f qcow2 -F raw). All existing callers use matching formats and
  are unaffected.
- CHANGELOG.md: fix wording of the incremental snapshot entry.
---
 .../apache/cloudstack/utils/qemu/QemuImg.java |  4 +--
 plugins/storage/volume/linstor/CHANGELOG.md   |  2 +-
 .../LinstorBackupSnapshotCommandWrapper.java  | 29 +++++--------------
 3 files changed, 10 insertions(+), 25 deletions(-)

diff --git a/plugins/hypervisors/kvm/src/main/java/org/apache/cloudstack/utils/qemu/QemuImg.java b/plugins/hypervisors/kvm/src/main/java/org/apache/cloudstack/utils/qemu/QemuImg.java
index e51c80e521c7..94978e53e12c 100644
--- a/plugins/hypervisors/kvm/src/main/java/org/apache/cloudstack/utils/qemu/QemuImg.java
+++ b/plugins/hypervisors/kvm/src/main/java/org/apache/cloudstack/utils/qemu/QemuImg.java
@@ -252,14 +252,12 @@ public void create(final QemuImgFile file, final QemuImgFile backingFile, final
             Shouldn't this be -o backing_file=filename instead?
          */
         s.add("-f");
+        s.add(file.getFormat().toString());
         if (backingFile != null) {
-            s.add(backingFile.getFormat().toString());
             s.add("-F");
             s.add(backingFile.getFormat().toString());
             s.add("-b");
             s.add(backingFile.getFileName());
-        } else {
-            s.add(file.getFormat().toString());
         }
 
         s.add(file.getFileName());
diff --git a/plugins/storage/volume/linstor/CHANGELOG.md b/plugins/storage/volume/linstor/CHANGELOG.md
index 43f13a1af52f..3f0ef3fc417d 100644
--- a/plugins/storage/volume/linstor/CHANGELOG.md
+++ b/plugins/storage/volume/linstor/CHANGELOG.md
@@ -28,7 +28,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Support for incremental snapshots on secondary storage backuped snapshots
+- Support for incremental snapshot backups on secondary storage
 
 ## [2026-06-24]
 
diff --git a/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java b/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java
index b56d5ebe315a..1c48eb2b22f4 100644
--- a/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java
+++ b/plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LinstorBackupSnapshotCommandWrapper.java
@@ -128,35 +128,22 @@ private String convertImageToQCow2(
      * parent then copies every cluster in which the two backing files differ into the overlay. The
      * explicit virtual size clips the DRBD metadata trailing the storage snapshot device.
      */
-    private String createIncrementalQCow2(
-        final String srcPath,
-        final SnapshotObjectTO dst,
-        final KVMStoragePool secondaryPool,
-        final File parentFile,
-        final long netSize,
-        int waitMilliSeconds)
-        throws LibvirtException, QemuImgException, IOException
-    {
-        final String dstDir = secondaryPool.getLocalPath() + File.separator + dst.getPath();
+    private String createIncrementalQCow2(final String srcPath, final SnapshotObjectTO dst, final KVMStoragePool secondaryPool,
+        final File parentFile, final long netSize, int waitMilliSeconds) throws LibvirtException, QemuImgException, IOException {
+        final String dstDir = secondaryPool.getLocalPathFor(dst.getPath());
         FileUtils.forceMkdir(new File(dstDir));
         final String dstPath = dstDir + File.separator + dst.getName();
 
-        final Script createOverlay = new Script("qemu-img", Duration.millis(waitMilliSeconds));
-        createOverlay.add("create", "-f", "qcow2", "-F", "raw", "-b", srcPath, dstPath, String.valueOf(netSize));
-        final String createResult = createOverlay.execute();
-        if (createResult != null) {
-            throw new QemuImgException("Unable to create qcow2 overlay of " + srcPath + ": " + createResult);
-        }
-
+        final QemuImg qemu = new QemuImg(waitMilliSeconds);
+        final QemuImgFile dstFile = new QemuImgFile(dstPath, netSize, QemuImg.PhysicalDiskFormat.QCOW2);
         try {
-            final QemuImg qemu = new QemuImg(waitMilliSeconds);
-            final QemuImgFile dstFile = new QemuImgFile(dstPath, QemuImg.PhysicalDiskFormat.QCOW2);
+            qemu.create(dstFile, new QemuImgFile(srcPath, QemuImg.PhysicalDiskFormat.RAW));
             qemu.rebase(dstFile, new QemuImgFile(parentFile.getAbsolutePath(), QemuImg.PhysicalDiskFormat.QCOW2),
                 QemuImg.PhysicalDiskFormat.QCOW2.toString(), true);
             // metadata-only rewrite to a relative backing name, so the chain stays valid on any mount point
             qemu.rebase(dstFile, new QemuImgFile(parentFile.getName(), QemuImg.PhysicalDiskFormat.QCOW2),
                 QemuImg.PhysicalDiskFormat.QCOW2.toString(), false);
-        } catch (final QemuImgException | LibvirtException e) {
+        } catch (final QemuImgException e) {
             FileUtils.deleteQuietly(new File(dstPath));
             throw e;
         }
@@ -226,7 +213,7 @@ public CopyCmdAnswer execute(LinstorBackupSnapshotCommand cmd, LibvirtComputingR
             boolean incremental = false;
             String dstPath = null;
             if (!encrypted && parentInstallPath != null && src.getVolume() != null) {
-                final File parentFile = new File(secondaryPool.getLocalPath() + File.separator + parentInstallPath);
+                final File parentFile = new File(secondaryPool.getLocalPathFor(parentInstallPath));
                 if (parentFile.isFile()) {
                     dstPath = createIncrementalQCow2(
                         srcPath, dst, secondaryPool, parentFile, src.getVolume().getSize(), cmd.getWaitInMillSeconds());

From 2bb786e00a44bcf703139a9fba5aff22105bfe93 Mon Sep 17 00:00:00 2001
From: Rene Peinthor <rene.peinthor@linbit.com>
Date: Thu, 6 Aug 2026 09:36:50 +0200
Subject: [PATCH 3/6] linstor: consider all diskful copies when picking a host

getDiskfulStoragePools only ever returned the storage pool of a single
node, so an endpoint was only looked for on that one node - if its
cloudstack host was down or disabled the operation fell back to a
temporary resource or a diskless attach although other diskful copies
were usable.

Return all diskful storage pools ordered by how suited the copy is (in
use, then active, then inactive) and walk that order when picking a
host. The order matters: thick LVM snapshots on shared storage pools
must be read on the node the volume is active on, so the host selection
no longer shuffles candidates.
---
 .../LinstorPrimaryDataStoreDriverImpl.java    | 14 ++--
 .../storage/datastore/util/LinstorUtil.java   | 75 +++++++++++++------
 .../datastore/util/LinstorUtilTest.java       | 61 +++++++++++++++
 3 files changed, 120 insertions(+), 30 deletions(-)

diff --git a/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java b/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java
index 2d12744024c9..00383c249f88 100644
--- a/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java
+++ b/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java
@@ -860,15 +860,17 @@ private Optional<RemoteHostEndPoint> getLinstorEP(DevelopersApi api, StoragePool
 
     private Optional<RemoteHostEndPoint> getDiskfullEP(DevelopersApi api, StoragePool storagePool, String rscName)
             throws ApiException {
-        List<com.linbit.linstor.api.model.StoragePool> linSPs = LinstorUtil.getDiskfulStoragePools(api, rscName);
-        if (linSPs != null) {
-            List<String> linstorNodeNames = linSPs.stream()
-                    .map(com.linbit.linstor.api.model.StoragePool::getNodeName)
-                    .collect(Collectors.toList());
-            Host host = getEnabledClusterHost(storagePool, linstorNodeNames);
+        // Take the first diskful copy whose host can be used: the copies are ordered by how
+        // suited they are (in use, then active, then inactive), and for shared storage pools
+        // only the active node may be read, so the order must not be broken here.
+        for (com.linbit.linstor.api.model.StoragePool linSP
+                : LinstorUtil.getDiskfulStoragePoolsByPreference(api, rscName)) {
+            Host host = getEnabledClusterHost(storagePool, Collections.singletonList(linSP.getNodeName()));
             if (host != null) {
                 return Optional.of(RemoteHostEndPoint.getHypervisorHostEndPoint(host));
             }
+            logger.debug("Linstor: no usable cloudstack host for diskful node {}, trying next copy",
+                    linSP.getNodeName());
         }
         logger.error("Linstor: No diskfull host found.");
         return Optional.empty();
diff --git a/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/util/LinstorUtil.java b/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/util/LinstorUtil.java
index 67c070f84eb0..434f6ca0162a 100644
--- a/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/util/LinstorUtil.java
+++ b/plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/util/LinstorUtil.java
@@ -41,9 +41,11 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -148,8 +150,26 @@ public static List<String> getLinstorNodeNames(@Nonnull DevelopersApi api) throw
         return nodes.stream().map(Node::getName).collect(Collectors.toList());
     }
 
+    /**
+     * How suitable a diskful resource is for reading its data (snapshots, volume copies):
+     * the node it is in use on is best, then nodes with an active resource, then inactive ones.
+     * Thick LVM snapshots on shared storage pools (dm-snapshot) are not cluster aware, so they
+     * must be read on the node the origin volume is active on - callers must honor this order.
+     */
+    private static int diskfulPreference(ResourceWithVolumes rwv) {
+        if (rwv.getState() != null && Boolean.TRUE.equals(rwv.getState().isInUse())) {
+            return 2;
+        }
+        return rwv.getFlags() == null || !rwv.getFlags().contains(ApiConsts.FLAG_RSC_INACTIVE) ? 1 : 0;
+    }
+
+    /**
+     * All storage pools holding a diskful copy of the resource, best suited first
+     * (see {@link #diskfulPreference}). Callers that need a single pool should take the first
+     * one, callers picking a host should walk the list in order and use the first usable one.
+     */
     public static List<com.linbit.linstor.api.model.StoragePool>
-    getDiskfulStoragePools(@Nonnull DevelopersApi api, @Nonnull String rscName) throws ApiException
+    getDiskfulStoragePoolsByPreference(@Nonnull DevelopersApi api, @Nonnull String rscName) throws ApiException
     {
         List<ResourceWithVolumes> resources = api.viewResources(
                 Collections.emptyList(),
@@ -159,43 +179,50 @@ public static List<String> getLinstorNodeNames(@Nonnull DevelopersApi api) throw
                 null,
                 null);
 
-        String nodeName = null;
-        String storagePoolName = null;
-        for (ResourceWithVolumes rwv : resources) {
-            if (rwv.getVolumes().isEmpty()) {
-                continue;
-            }
-            Volume vol = rwv.getVolumes().get(0);
-            if (vol.getProviderKind() != ProviderKind.DISKLESS) {
-                nodeName = rwv.getNodeName();
-                storagePoolName = vol.getStoragePoolName();
-                break;
-            }
-        }
+        // node name -> storage pool name, ordered by how suited the copy is
+        List<Pair<String, String>> candidates = resources.stream()
+                .filter(rwv -> !rwv.getVolumes().isEmpty()
+                        && rwv.getVolumes().get(0).getProviderKind() != ProviderKind.DISKLESS)
+                .sorted(Comparator.comparingInt(LinstorUtil::diskfulPreference).reversed())
+                .map(rwv -> new Pair<>(rwv.getNodeName(), rwv.getVolumes().get(0).getStoragePoolName()))
+                .collect(Collectors.toList());
 
-        if (nodeName == null) {
-            return null;
+        if (candidates.isEmpty()) {
+            return Collections.emptyList();
         }
 
         List<com.linbit.linstor.api.model.StoragePool> sps = api.viewStoragePools(
-                Collections.singletonList(nodeName),
-                Collections.singletonList(storagePoolName),
+                candidates.stream().map(Pair::first).distinct().collect(Collectors.toList()),
+                candidates.stream().map(Pair::second).distinct().collect(Collectors.toList()),
                 Collections.emptyList(),
                 null,
                 null,
                 true
         );
-        return sps != null ? sps : Collections.emptyList();
+        if (sps == null) {
+            return Collections.emptyList();
+        }
+
+        // viewStoragePools does not keep our ordering, so re-apply it
+        List<com.linbit.linstor.api.model.StoragePool> ordered = new ArrayList<>();
+        for (Pair<String, String> candidate : candidates) {
+            sps.stream()
+                    .filter(sp -> candidate.first().equals(sp.getNodeName())
+                            && candidate.second().equals(sp.getStoragePoolName()))
+                    .findFirst()
+                    .ifPresent(ordered::add);
+        }
+        return ordered;
     }
 
+    /**
+     * The storage pool of the best suited diskful copy, null if the resource has none.
+     */
     public static com.linbit.linstor.api.model.StoragePool
     getDiskfulStoragePool(@Nonnull DevelopersApi api, @Nonnull String rscName) throws ApiException
     {
-        List<com.linbit.linstor.api.model.StoragePool> sps = getDiskfulStoragePools(api, rscName);
-        if (sps != null) {
-            return !sps.isEmpty() ? sps.get(0) : null;
-        }
-        return null;
+        List<com.linbit.linstor.api.model.StoragePool> sps = getDiskfulStoragePoolsByPreference(api, rscName);
+        return !sps.isEmpty() ? sps.get(0) : null;
     }
 
     public static String getSnapshotPath(com.linbit.linstor.api.model.StoragePool sp, String rscName, String snapshotName) {
diff --git a/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/util/LinstorUtilTest.java b/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/util/LinstorUtilTest.java
index 55f0c6ebe6dc..b3a6f127f5a2 100644
--- a/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/util/LinstorUtilTest.java
+++ b/plugins/storage/volume/linstor/src/test/java/org/apache/cloudstack/storage/datastore/util/LinstorUtilTest.java
@@ -16,6 +16,7 @@
 // under the License.
 package org.apache.cloudstack.storage.datastore.util;
 
+import com.linbit.linstor.api.ApiConsts;
 import com.linbit.linstor.api.ApiException;
 import com.linbit.linstor.api.DevelopersApi;
 import com.linbit.linstor.api.model.AutoSelectFilter;
@@ -23,7 +24,10 @@
 import com.linbit.linstor.api.model.Properties;
 import com.linbit.linstor.api.model.ProviderKind;
 import com.linbit.linstor.api.model.ResourceGroup;
+import com.linbit.linstor.api.model.ResourceState;
+import com.linbit.linstor.api.model.ResourceWithVolumes;
 import com.linbit.linstor.api.model.StoragePool;
+import com.linbit.linstor.api.model.Volume;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -115,6 +119,63 @@ public void testGetSnapshotPath() {
         }
     }
 
+    private ResourceWithVolumes mockResource(String node, String pool, boolean inUse, boolean inactive) {
+        ResourceWithVolumes rwv = new ResourceWithVolumes();
+        rwv.setName("cs-test");
+        rwv.setNodeName(node);
+        Volume vol = new Volume();
+        vol.setProviderKind(ProviderKind.LVM);
+        vol.setStoragePoolName(pool);
+        rwv.setVolumes(Collections.singletonList(vol));
+        ResourceState state = new ResourceState();
+        state.setInUse(inUse);
+        rwv.setState(state);
+        if (inactive) {
+            rwv.setFlags(Collections.singletonList(ApiConsts.FLAG_RSC_INACTIVE));
+        }
+        return rwv;
+    }
+
+    @Test
+    public void testDiskfulStoragePoolsOrderedByPreference() throws ApiException {
+        // inactive copy first, in-use copy last: the result must be in-use, active, inactive
+        when(api.viewResources(Collections.emptyList(), Collections.singletonList("cs-test"),
+                Collections.emptyList(), Collections.emptyList(), null, null))
+                .thenReturn(Arrays.asList(
+                        mockResource("nodeC", "poolC", false, true),
+                        mockResource("nodeB", "poolB", false, false),
+                        mockResource("nodeA", "poolA", true, false)));
+        // the pools are queried in preference order (in use, active, inactive)
+        when(api.viewStoragePools(Arrays.asList("nodeA", "nodeB", "nodeC"),
+                Arrays.asList("poolA", "poolB", "poolC"), Collections.emptyList(), null, null, true))
+                .thenReturn(Arrays.asList(
+                        mockStoragePool("poolB", "nodeB", ProviderKind.LVM),
+                        mockStoragePool("poolA", "nodeA", ProviderKind.LVM),
+                        mockStoragePool("poolC", "nodeC", ProviderKind.LVM)));
+
+        List<StoragePool> pools = LinstorUtil.getDiskfulStoragePoolsByPreference(api, "cs-test");
+        Assert.assertEquals(Arrays.asList("nodeA", "nodeB", "nodeC"),
+                pools.stream().map(StoragePool::getNodeName).collect(Collectors.toList()));
+        // the single-pool accessor keeps returning the best suited copy
+        Assert.assertEquals("nodeA", LinstorUtil.getDiskfulStoragePool(api, "cs-test").getNodeName());
+    }
+
+    @Test
+    public void testDiskfulStoragePoolsIgnoresDiskless() throws ApiException {
+        ResourceWithVolumes diskless = new ResourceWithVolumes();
+        diskless.setName("cs-test");
+        diskless.setNodeName("nodeD");
+        Volume dlVol = new Volume();
+        dlVol.setProviderKind(ProviderKind.DISKLESS);
+        diskless.setVolumes(Collections.singletonList(dlVol));
+        when(api.viewResources(Collections.emptyList(), Collections.singletonList("cs-test"),
+                Collections.emptyList(), Collections.emptyList(), null, null))
+                .thenReturn(Collections.singletonList(diskless));
+
+        Assert.assertTrue(LinstorUtil.getDiskfulStoragePoolsByPreference(api, "cs-test").isEmpty());
+        Assert.assertNull(LinstorUtil.getDiskfulStoragePool(api, "cs-test"));
+    }
+
     @Test
     public void testGetRscGroupStoragePools() throws ApiException {
         List<StoragePool> storagePools = LinstorUtil.getRscGroupStoragePools(api, "cloudstack");

From 26f95a889cc39ef748af308606869638e7f0c14b Mon Sep 17 00:00:00 2001
From: Rene Peinthor <rene.peinthor@linbit.com>
Date: Thu, 6 Aug 2026 10:40:55 +0200
Subject: [PATCH 4/6] snapshot: enforce snapshot.delta.max on content-based
 (Linstor) chains

endCurrentChainIfNeeded walks the snapshot chain through
findOneBySnapshotId, whose FIND_SNAPSHOT_IN_ZONE query returns an
arbitrary store reference of the snapshot. Content-based chains only
link their image store reference - the primary one stays unlinked - so
for Linstor volumes the walk always saw a chain length of one and
snapshot.delta.max was never enforced: deltas were chained forever and
no full snapshot was ever taken again. Checkpoint-based chains were
unaffected because they carry the parent link on both references.

Order references carrying a chain link (parent or checkpoint) first in
FIND_SNAPSHOT_IN_ZONE, and stop the walk gracefully if a parent has no
references left instead of throwing an NPE.
---
 .../db/SnapshotDataStoreDaoImpl.java          |  6 ++-
 .../storage/snapshot/SnapshotManagerImpl.java |  4 +-
 .../snapshot/SnapshotManagerImplTest.java     | 50 +++++++++++++++++++
 3 files changed, 57 insertions(+), 3 deletions(-)

diff --git a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
index 924805017862..fbb2767060c6 100644
--- a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
+++ b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
@@ -102,9 +102,13 @@ public class SnapshotDataStoreDaoImpl extends GenericDaoBase<SnapshotDataStoreVO
             " order by created %s " +
             " limit 1";
 
+    // prefer the reference carrying the incremental chain link: content-based (Linstor) chains only
+    // link their image store reference, the primary one stays unlinked, and chain walks (e.g.
+    // snapshot.delta.max enforcement) must not end up on the unlinked copy
     private static final String FIND_SNAPSHOT_IN_ZONE = "SELECT ssr.* FROM " +
             "snapshot_store_ref ssr, snapshots s " +
-            "WHERE ssr.snapshot_id=? AND ssr.snapshot_id = s.id AND s.data_center_id=?;";
+            "WHERE ssr.snapshot_id=? AND ssr.snapshot_id = s.id AND s.data_center_id=? " +
+            "ORDER BY (ssr.parent_snapshot_id > 0) DESC, (ssr.kvm_checkpoint_path IS NOT NULL) DESC LIMIT 1;";
 
     private static final String GET_PHYSICAL_SIZE_OF_SNAPSHOTS_ON_PRIMARY_BY_ACCOUNT = "SELECT SUM(s.physical_size) " +
             "FROM cloud.snapshot_store_ref s " +
diff --git a/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java b/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
index 4372507c1623..3fbe33aa6d89 100755
--- a/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
+++ b/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
@@ -779,10 +779,10 @@ private void postCreateSnapshot(Long volumeId, Long snapshotId, Long policyId, L
     /**
      * Will mark this snapshot as the end of the chain if it has reached the value of snapshot.delta.max.
      * */
-    private void endCurrentChainIfNeeded(Long snapshotId, Long zoneId) {
+    protected void endCurrentChainIfNeeded(Long snapshotId, Long zoneId) {
         SnapshotDataStoreVO snapshotDataStoreVo = _snapshotStoreDao.findOneBySnapshotId(snapshotId, zoneId);
         int chainSize = 1;
-        while (snapshotDataStoreVo.getParentSnapshotId() > 0) {
+        while (snapshotDataStoreVo != null && snapshotDataStoreVo.getParentSnapshotId() > 0) {
             snapshotDataStoreVo = _snapshotStoreDao.findOneBySnapshotId(snapshotDataStoreVo.getParentSnapshotId(), zoneId);
             chainSize++;
         }
diff --git a/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java b/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java
index 0216a18929d9..e283fd88ed8c 100644
--- a/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java
+++ b/server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java
@@ -640,4 +640,54 @@ public void testIsIncrementalChainRefKvmStandalone() {
         Mockito.when(ref.getParentSnapshotId()).thenReturn(0L);
         Assert.assertFalse(snapshotManager.isIncrementalChainRef(ref, Hypervisor.HypervisorType.KVM));
     }
+
+    private void mockParentLinkedChain(long chainLength, long zoneId) {
+        // snapshot i is a delta on snapshot i - 1, snapshot 1 is the full chain start (parent 0)
+        for (long i = 1; i <= chainLength; i++) {
+            SnapshotDataStoreVO ref = Mockito.mock(SnapshotDataStoreVO.class);
+            Mockito.when(ref.getParentSnapshotId()).thenReturn(i - 1);
+            Mockito.when(snapshotStoreDao.findOneBySnapshotId(i, zoneId)).thenReturn(ref);
+        }
+    }
+
+    @Test
+    public void testEndCurrentChainIfNeededMarksEndOfChainAtDeltaMax() {
+        final long zoneId = 1L;
+        final long deltaMax = SnapshotManager.snapshotDeltaMax.value();
+        mockParentLinkedChain(deltaMax, zoneId);
+        SnapshotDataStoreVO leafPrimaryRef = Mockito.mock(SnapshotDataStoreVO.class);
+        SnapshotDataStoreVO leafImageRef = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(snapshotStoreDao.listBySnapshotId(deltaMax)).thenReturn(List.of(leafPrimaryRef, leafImageRef));
+
+        snapshotManager.endCurrentChainIfNeeded(deltaMax, zoneId);
+
+        Mockito.verify(leafPrimaryRef).setEndOfChain(true);
+        Mockito.verify(leafImageRef).setEndOfChain(true);
+        Mockito.verify(snapshotStoreDao, Mockito.times(2)).update(Mockito.anyLong(), Mockito.any(SnapshotDataStoreVO.class));
+    }
+
+    @Test
+    public void testEndCurrentChainIfNeededKeepsChainBelowDeltaMax() {
+        final long zoneId = 1L;
+        final long chainLength = SnapshotManager.snapshotDeltaMax.value() - 1;
+        mockParentLinkedChain(chainLength, zoneId);
+
+        snapshotManager.endCurrentChainIfNeeded(chainLength, zoneId);
+
+        Mockito.verify(snapshotStoreDao, Mockito.never()).listBySnapshotId(Mockito.anyLong());
+        Mockito.verify(snapshotStoreDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any(SnapshotDataStoreVO.class));
+    }
+
+    @Test
+    public void testEndCurrentChainIfNeededStopsOnMissingParentRef() {
+        final long zoneId = 1L;
+        SnapshotDataStoreVO leafRef = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(leafRef.getParentSnapshotId()).thenReturn(5L);
+        Mockito.when(snapshotStoreDao.findOneBySnapshotId(6L, zoneId)).thenReturn(leafRef);
+        Mockito.when(snapshotStoreDao.findOneBySnapshotId(5L, zoneId)).thenReturn(null);
+
+        snapshotManager.endCurrentChainIfNeeded(6L, zoneId);
+
+        Mockito.verify(snapshotStoreDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any(SnapshotDataStoreVO.class));
+    }
 }

From 22c192daed214ad156c2de1fad271ed92456b9fc Mon Sep 17 00:00:00 2001
From: Rene Peinthor <rene.peinthor@linbit.com>
Date: Thu, 6 Aug 2026 12:58:08 +0200
Subject: [PATCH 5/6] linstor: marvin tests for incremental snapshot backups

Cover the incremental (content-diff) snapshot feature end to end:
full -> delta chaining with parent links, snapshot.delta.max chain
rotation, revert to a mid-chain delta, template from a delta (chain
flattening), fallback to a full backup when the parent file is missing
on secondary storage, mid-chain deletion, and that encrypted volumes
are always backed up as full copies.

Chain bookkeeping (parent_snapshot_id, end_of_chain) is not exposed via
the API, so the tests verify it in the DB when reachable and otherwise
fall back to physical-size comparisons against the chain's full backup.
Tests needing host SSH (secondary storage manipulation/inspection)
self-skip without credentials.

The ServiceReady helper the Linstor test modules used to copy around
moves to a shared linstor_test_utils module instead of gaining a third
copy.
---
 test/integration/plugins/linstor/README.md    |  30 +
 .../plugins/linstor/linstor_test_utils.py     |  54 ++
 .../test_linstor_encrypted_snapshots.py       |  28 +-
 .../test_linstor_incremental_snapshots.py     | 639 ++++++++++++++++++
 .../plugins/linstor/test_linstor_volumes.py   |  42 +-
 5 files changed, 727 insertions(+), 66 deletions(-)
 create mode 100644 test/integration/plugins/linstor/linstor_test_utils.py
 create mode 100644 test/integration/plugins/linstor/test_linstor_incremental_snapshots.py

diff --git a/test/integration/plugins/linstor/README.md b/test/integration/plugins/linstor/README.md
index 4971c9506b5b..f3c486de03c2 100644
--- a/test/integration/plugins/linstor/README.md
+++ b/test/integration/plugins/linstor/README.md
@@ -66,3 +66,33 @@ Extra prerequisites:
 ```
 nosetests --with-marvin --marvin-config=<marvin-cfg-file> <cloudstack-dir>/test/integration/plugins/linstor/test_linstor_encrypted_snapshots.py --zone=<zone> --hypervisor=kvm
 ```
+
+## Incremental snapshot tests
+
+`test_linstor_incremental_snapshots.py` covers incremental (content-diff) snapshot backups on
+NFS secondary storage: full -> delta chaining, `snapshot.delta.max` chain rotation, revert to a
+mid-chain delta, template creation from a delta (chain flattening), the fallback to a full backup
+when the parent file is missing on secondary storage, mid-chain deletion, and that encrypted
+volumes are always backed up as full copies.
+
+Extra prerequisites:
+
+* NFS secondary storage (incremental snapshots are only supported there).
+* `kvm.incremental.snapshot`, `kvm.snapshot.enabled` and `lin.backup.snapshots` are set by the
+  tests themselves (and restored afterwards). Snapshots are taken while the VMs are running
+  (crash-consistent storage snapshots; markers are synced before each snapshot); VMs are only
+  stopped where CloudStack requires it (revert).
+* The fallback and qcow2-inspection checks need host SSH credentials, either in the marvin config
+  (zones->pods->clusters->hosts) or via the `HOST_SSH_USER` / `HOST_SSH_PASSWORD` env vars; those
+  tests self-skip without them.
+* The parent-link assertions read `snapshot_store_ref` directly and therefore need the DB
+  connection of the marvin config to work; without it the tests fall back to comparing
+  physical sizes.
+
+```
+nosetests --with-marvin --marvin-config=<marvin-cfg-file> <cloudstack-dir>/test/integration/plugins/linstor/test_linstor_incremental_snapshots.py --zone=<zone> --hypervisor=kvm
+```
+
+Note: select single tests with `... test_linstor_incremental_snapshots.py:TestLinstorIncrementalSnapshots -m <test-name-pattern>`
+(class selection plus a method filter); the `file.py:Class.test_method` form bypasses the marvin
+plugin's test client injection and fails with `'NoneType' object has no attribute 'getApiClient'`.
diff --git a/test/integration/plugins/linstor/linstor_test_utils.py b/test/integration/plugins/linstor/linstor_test_utils.py
new file mode 100644
index 000000000000..c048f93951f1
--- /dev/null
+++ b/test/integration/plugins/linstor/linstor_test_utils.py
@@ -0,0 +1,54 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Helpers shared by the Linstor plugin integration tests."""
+
+import socket
+import time
+
+
+class ServiceReady:
+    @classmethod
+    def ready(cls, hostname: str, port: int) -> bool:
+        try:
+            s = socket.create_connection((hostname, port), timeout=1)
+            s.close()
+            return True
+        except (ConnectionRefusedError, socket.timeout, OSError):
+            return False
+
+    @classmethod
+    def wait(cls, hostname: str, port: int, wait_interval: float = 5, timeout: int = 120,
+             service_name: str = 'ssh') -> bool:
+        """
+        Wait until the given service can be reached, raise RuntimeError on timeout.
+        :param hostname: host to connect to
+        :param port: port of the application
+        :param wait_interval: seconds between connection attempts
+        :param timeout: seconds to wait before raising
+        :param service_name: name of the service waited for (used in the error message)
+        """
+        starttime = int(round(time.time() * 1000))
+        while not cls.ready(hostname, port):
+            if starttime + timeout * 1000 < int(round(time.time() * 1000)):
+                raise RuntimeError("{s} {h} cannot be reached.".format(s=service_name, h=hostname))
+            time.sleep(wait_interval)
+        return True
+
+    @classmethod
+    def wait_ssh_ready(cls, hostname: str, wait_interval: float = 2, timeout: int = 120) -> bool:
+        return cls.wait(hostname, 22, wait_interval, timeout, "ssh")
diff --git a/test/integration/plugins/linstor/test_linstor_encrypted_snapshots.py b/test/integration/plugins/linstor/test_linstor_encrypted_snapshots.py
index 5f440309bb34..97902942f412 100644
--- a/test/integration/plugins/linstor/test_linstor_encrypted_snapshots.py
+++ b/test/integration/plugins/linstor/test_linstor_encrypted_snapshots.py
@@ -19,8 +19,6 @@
 import logging
 import os
 import random
-import socket
-import time
 
 # All tests inherit from cloudstackTestCase
 from marvin.cloudstackTestCase import cloudstackTestCase
@@ -35,6 +33,8 @@
 from marvin.sshClient import SshClient
 from nose.plugins.attrib import attr
 
+from linstor_test_utils import ServiceReady
+
 # Prerequisites:
 #  Only one zone / pod / cluster
 #  Only KVM hypervisor (Linstor only supports KVM)
@@ -119,30 +119,6 @@ def __init__(self, linstor_controller_url):
         }
 
 
-class ServiceReady:
-    @classmethod
-    def ready(cls, hostname, port):
-        try:
-            s = socket.create_connection((hostname, port), timeout=1)
-            s.close()
-            return True
-        except (ConnectionRefusedError, socket.timeout, OSError):
-            return False
-
-    @classmethod
-    def wait(cls, hostname, port, wait_interval=5, timeout=120, service_name='ssh'):
-        starttime = int(round(time.time() * 1000))
-        while not cls.ready(hostname, port):
-            if starttime + timeout * 1000 < int(round(time.time() * 1000)):
-                raise RuntimeError("{s} {h} cannot be reached.".format(s=service_name, h=hostname))
-            time.sleep(wait_interval)
-        return True
-
-    @classmethod
-    def wait_ssh_ready(cls, hostname, wait_interval=2, timeout=120):
-        return cls.wait(hostname, 22, wait_interval, timeout, "ssh")
-
-
 class TestLinstorEncryptedSnapshots(cloudstackTestCase):
 
     @classmethod
diff --git a/test/integration/plugins/linstor/test_linstor_incremental_snapshots.py b/test/integration/plugins/linstor/test_linstor_incremental_snapshots.py
new file mode 100644
index 000000000000..340d90d4d163
--- /dev/null
+++ b/test/integration/plugins/linstor/test_linstor_incremental_snapshots.py
@@ -0,0 +1,639 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import json
+import logging
+import os
+import random
+
+# All tests inherit from cloudstackTestCase
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+# Import Integration Libraries
+from marvin.lib.base import Account, Configurations, ServiceOffering, \
+    Snapshot, StoragePool, Template, User, VirtualMachine, Volume
+from marvin.lib.common import get_domain, get_template, get_zone, list_hosts, list_virtual_machines, list_volumes
+from marvin.lib.utils import cleanup_resources
+from marvin.sshClient import SshClient
+from nose.plugins.attrib import attr
+
+from linstor_test_utils import ServiceReady
+
+# Prerequisites:
+#  Only one zone / pod / cluster
+#  Only KVM hypervisor (Linstor only supports KVM)
+#  One Linstor storage pool (resource group without a LUKS layer; plain volumes)
+#  NFS secondary storage (incremental snapshots are only supported on NFS secondary)
+#  'lin.backup.snapshots' enabled (default true) so snapshots are backed up to secondary storage.
+#  'kvm.snapshot.enabled' enabled (default true): snapshots are taken on running VMs (the tests
+#    set it). Markers are synced before each snapshot, so the crash-consistent images contain them.
+#
+# What this exercises (the Linstor incremental snapshot feature):
+#  Volumes on Linstor primary storage back their snapshots up to secondary storage as qcow2 files.
+#  With kvm.incremental.snapshot enabled, every snapshot after a full one is stored as a delta
+#  qcow2 (a content diff produced by qemu-img rebase) whose backing file is the previous snapshot,
+#  so a chain member only consumes the space of the blocks that changed since its parent.
+#
+#  * chaining:  full -> delta -> delta, parent links recorded on the image store refs
+#  * rotation:  snapshot.delta.max caps the chain length, then a new full is taken
+#  * restore:   reverting to a chain member follows the backing chain
+#  * flatten:   templates/volumes created from a mid-chain delta get standalone content
+#  * fallback:  a missing parent file on secondary storage degrades to a full backup
+#  * exclusion: encrypted volumes are always backed up as full copies
+#
+# Note on verification: the parent/child relationship of the backups is bookkeeping internal to the
+# management server (snapshot_store_ref.parent_snapshot_id), not exposed through the API, so these
+# tests read it from the DB where a DB connection is available and otherwise fall back to comparing
+# physical sizes (a delta of a small change is orders of magnitude smaller than a full).
+
+MARKER_PATH = "/root/cs_incr_marker.txt"
+
+
+class TestData:
+    account = "account"
+    computeOffering = "computeoffering"
+    domainId = "domainId"
+    hypervisor = "hypervisor"
+    provider = "provider"
+    scope = "scope"
+    storageTag = "linstor"
+    tags = "tags"
+    user = "user"
+    virtualMachine = "virtualmachine"
+    zoneId = "zoneId"
+
+    def __init__(self, linstor_controller_url):
+        self.testdata = {
+            TestData.account: {
+                "email": "test-incr@test.com",
+                "firstname": "John",
+                "lastname": "Doe",
+                "username": "test-incr",
+                "password": "test"
+            },
+            TestData.user: {
+                "email": "user-incr@test.com",
+                "firstname": "Jane",
+                "lastname": "Doe",
+                "username": "test-incr-user",
+                "password": "password"
+            },
+            "primarystorage": {
+                "name": "LinstorIncrPool-%d" % random.randint(0, 100000),
+                TestData.scope: "ZONE",
+                "url": linstor_controller_url,
+                TestData.provider: "Linstor",
+                TestData.tags: TestData.storageTag,
+                TestData.hypervisor: "KVM",
+                "details": {
+                    "resourceGroup": "acs-basic"
+                }
+            },
+            TestData.virtualMachine: {
+                "name": "TestIncrVM",
+                "displayname": "Test Incremental VM"
+            },
+            TestData.computeOffering: {
+                "name": "Linstor_Compute_Incr",
+                "displaytext": "Linstor_Compute_Incr",
+                "cpunumber": 1,
+                "cpuspeed": 500,
+                "memory": 512,
+                "storagetype": "shared",
+                TestData.tags: TestData.storageTag
+            },
+            TestData.zoneId: 1,
+            TestData.domainId: 1,
+        }
+
+
+class TestLinstorIncrementalSnapshots(cloudstackTestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        testclient = super(TestLinstorIncrementalSnapshots, cls).getClsTestClient()
+
+        cls.apiClient = testclient.getApiClient()
+        cls.dbConnection = testclient.getDbConnection()
+
+        cls._cleanup = []
+        cls.skip_reason = None
+        cls.original_config = {}
+
+        # The first host runs the Linstor controller (per the Linstor test prerequisites).
+        first_host = list_hosts(cls.apiClient)[0]
+        cls.testdata = TestData(first_host.ipaddress).testdata
+
+        cls.zone = get_zone(cls.apiClient, zone_id=cls.testdata[TestData.zoneId])
+        cls.domain = get_domain(cls.apiClient, cls.testdata[TestData.domainId])
+        cls.template = get_template(cls.apiClient, cls.zone.id, hypervisor="KVM")
+
+        # Host SSH credentials, needed by the tests that inspect or manipulate the backed-up qcow2
+        # files on secondary storage. A full marvin config carries these under
+        # zones->pods->clusters->hosts; a lightweight config may omit them - fall back to
+        # HOST_SSH_USER / HOST_SSH_PASSWORD env vars, and skip those tests if neither is present.
+        cls.hostConfig = None
+        try:
+            cls.hostConfig = cls.config.__dict__["zones"][0].__dict__["pods"][0].__dict__["clusters"][0] \
+                .__dict__["hosts"][0].__dict__
+        except (KeyError, IndexError, AttributeError, TypeError):
+            host_user = os.environ.get("HOST_SSH_USER")
+            host_pass = os.environ.get("HOST_SSH_PASSWORD")
+            if host_user and host_pass:
+                cls.hostConfig = {"username": host_user, "password": host_pass}
+
+        # The feature under test needs snapshot backups on secondary storage and the (cluster-scoped,
+        # here globally set) incremental snapshot switch. Save the original values and restore them in
+        # tearDownClass so the run leaves no configuration behind.
+        cls._set_config("lin.backup.snapshots", "true")
+        cls._set_config("kvm.incremental.snapshot", "true")
+        # all snapshots in these tests are taken while the VM is running (crash-consistent
+        # storage snapshots; also exercises the running-VM path of the feature)
+        cls._set_config("kvm.snapshot.enabled", "true")
+
+        primarystorage = cls.testdata["primarystorage"]
+        api_token = os.environ.get("LINSTOR_API_TOKEN")
+        if api_token:
+            primarystorage["details"]["lin.auth.apitoken"] = api_token
+
+        try:
+            cls.primary_storage = StoragePool.create(
+                cls.apiClient,
+                primarystorage,
+                scope=primarystorage[TestData.scope],
+                zoneid=cls.zone.id,
+                provider=primarystorage[TestData.provider],
+                tags=primarystorage[TestData.tags],
+                hypervisor=primarystorage[TestData.hypervisor]
+            )
+        except Exception as e:
+            cls.skip_reason = (
+                "Could not register the Linstor primary storage pool (%s). If the Linstor controller "
+                "requires authentication, set the LINSTOR_API_TOKEN env var to a valid controller API "
+                "token before running these tests." % e)
+            return
+
+        cls.compute_offering = ServiceOffering.create(
+            cls.apiClient,
+            cls.testdata[TestData.computeOffering]
+        )
+
+        cls.account = Account.create(cls.apiClient, cls.testdata[TestData.account], admin=1)
+        cls.user = User.create(
+            cls.apiClient, cls.testdata[TestData.user],
+            account=cls.account.name, domainid=cls.domain.id)
+
+        cls._cleanup = [
+            cls.compute_offering,
+            cls.user,
+            cls.account,
+        ]
+
+    @classmethod
+    def tearDownClass(cls):
+        try:
+            cleanup_resources(cls.apiClient, cls._cleanup)
+            if getattr(cls, "primary_storage", None) is not None:
+                cls.primary_storage.delete(cls.apiClient)
+        except Exception as e:
+            logging.debug("Exception in tearDownClass: %s" % e)
+        finally:
+            cls._restore_configs()
+
+    def setUp(self):
+        if self.skip_reason:
+            self.skipTest(self.skip_reason)
+        self.cleanup = []
+
+    def tearDown(self):
+        cleanup_resources(self.apiClient, self.cleanup)
+
+    # --------------------------------------------------------------------- #
+    # Tests
+    # --------------------------------------------------------------------- #
+
+    @attr(tags=['basic'], required_hardware=True)
+    def test_01_second_snapshot_is_incremental(self):
+        """The first snapshot is a full backup, later ones are deltas chained to their parent."""
+        vm = self._deploy_vm("TestIncrVM-chain")
+
+        self._write_marker(vm, "incr-chain-v1")
+        snap_full = self._snapshot_root_volume(vm)
+        self._assert_full(snap_full)
+
+        self._write_marker(vm, "incr-chain-v2")
+        snap_delta1 = self._snapshot_root_volume(vm)
+        self._assert_delta(snap_delta1, parent=snap_full)
+
+        self._write_marker(vm, "incr-chain-v3")
+        snap_delta2 = self._snapshot_root_volume(vm)
+        self._assert_delta(snap_delta2, parent=snap_delta1, full=snap_full)
+
+        # a delta of a one-line change must be far smaller than the full backup
+        self.assertLess(
+            int(snap_delta1.physicalsize) * 4, int(snap_full.physicalsize),
+            "Delta snapshot (%s bytes) is not substantially smaller than the full one (%s bytes)"
+            % (snap_delta1.physicalsize, snap_full.physicalsize))
+
+        # if we can reach the files on secondary storage, the delta must be a qcow2 whose
+        # backing file is its parent, referenced by a relative name (portable across mounts)
+        info = self._qemu_img_info_of_backed_up_snapshot(snap_delta1)
+        if info is not None:
+            backing = info.get("backing-filename")
+            self.assertIsNotNone(backing, "Delta snapshot qcow2 has no backing file: %s" % json.dumps(info))
+            self.assertNotIn("/", backing, "Backing file %r is not a relative name" % backing)
+
+    @attr(tags=['basic'], required_hardware=True)
+    def test_02_revert_to_delta_snapshot(self):
+        """Reverting to a mid-chain delta restores exactly that snapshot's content."""
+        vm = self._deploy_vm("TestIncrVM-revert")
+
+        self._write_marker(vm, "incr-revert-v1")
+        snap1 = self._snapshot_root_volume(vm)
+
+        self._write_marker(vm, "incr-revert-v2")
+        snap2 = self._snapshot_root_volume(vm)
+        self._assert_delta(snap2, parent=snap1)
+
+        # change the data once more so a successful revert is detectable
+        self._write_marker(vm, "incr-revert-v3-CHANGED")
+        vm.stop(self.apiClient)
+
+        # revert to the delta: the restore has to follow the backing chain (full + delta)
+        Volume.revertToSnapshot(self.apiClient, snap2.id)
+        self._start_vm(vm)
+        self.assertEqual("incr-revert-v2", self._read_marker(vm),
+                         "Revert to the delta snapshot did not restore its content")
+
+        # revert to the chain-starting full as well
+        vm.stop(self.apiClient)
+        Volume.revertToSnapshot(self.apiClient, snap1.id)
+        self._start_vm(vm)
+        self.assertEqual("incr-revert-v1", self._read_marker(vm),
+                         "Revert to the full snapshot did not restore its content")
+
+    @attr(tags=['basic'], required_hardware=True)
+    def test_03_delta_max_rotates_chain(self):
+        """After snapshot.delta.max chain members the chain is ended and a new full is taken."""
+        original = Configurations.list(self.apiClient, name="snapshot.delta.max")[0].value
+        Configurations.update(self.apiClient, name="snapshot.delta.max", value="2")
+        try:
+            vm = self._deploy_vm("TestIncrVM-rotate")
+
+            snap1 = self._snapshot_root_volume(vm)   # chain member 1: full
+            snap2 = self._snapshot_root_volume(vm)   # chain member 2: delta, reaches the cap
+            snap3 = self._snapshot_root_volume(vm)   # must start a new chain: full
+
+            self._assert_full(snap1)
+            self._assert_delta(snap2, parent=snap1)
+            self._assert_full(snap3, full=snap1)
+
+            ref2 = self._image_store_ref(snap2)
+            if ref2 is not None:
+                self.assertTrue(ref2["end_of_chain"],
+                                "Snapshot that reached snapshot.delta.max was not marked end of chain")
+        finally:
+            Configurations.update(self.apiClient, name="snapshot.delta.max", value=original)
+
+    @attr(tags=['basic'], required_hardware=True)
+    def test_04_template_from_delta_snapshot(self):
+        """A template created from a mid-chain delta must contain the full (flattened) content."""
+        vm = self._deploy_vm("TestIncrVM-tmpl")
+
+        self._write_marker(vm, "incr-tmpl-v1")
+        snap1 = self._snapshot_root_volume(vm)
+
+        self._write_marker(vm, "incr-tmpl-v2")
+        snap2 = self._snapshot_root_volume(vm)
+        self._assert_delta(snap2, parent=snap1)
+
+        template = Template.create_from_snapshot(
+            self.apiClient, snap2,
+            {
+                "name": "incr-tmpl",
+                "displaytext": "template from delta snapshot",
+                "ostypeid": self.template.ostypeid,
+                "ispublic": False,
+            })
+        self.cleanup.insert(0, template)
+
+        # a VM deployed from that template must carry the delta's content: the template was
+        # created from a delta qcow2 and only works if the chain got flattened along the way
+        vm_from_template = VirtualMachine.create(
+            self.apiClient,
+            {"name": "TestIncrVM-fromtmpl", "displayname": "TestIncrVM-fromtmpl"},
+            accountid=self.account.name,
+            zoneid=self.zone.id,
+            serviceofferingid=self.compute_offering.id,
+            templateid=template.id,
+            domainid=self.domain.id,
+            startvm=True,
+            mode='basic',
+        )
+        self.cleanup.insert(0, vm_from_template)
+        # the template holds a crash-consistent image (snapshot of a running VM), so the first
+        # boot replays the filesystem journal / fscks and can take considerably longer
+        self._start_vm(vm_from_template, ssh_timeout=600)
+        self.assertEqual("incr-tmpl-v2", self._read_marker(vm_from_template),
+                         "VM from the delta-snapshot template does not contain the snapshot's content")
+
+    @attr(tags=['basic'], required_hardware=True)
+    def test_05_full_backup_when_parent_file_missing(self):
+        """If the parent file vanished from secondary storage, the next snapshot degrades to a full."""
+        if not self.hostConfig:
+            self.skipTest("No host SSH credentials available (set HOST_SSH_USER/HOST_SSH_PASSWORD or "
+                          "provide them in the marvin config) - cannot manipulate secondary storage")
+
+        vm = self._deploy_vm("TestIncrVM-fallback")
+
+        snap1 = self._snapshot_root_volume(vm, add_to_cleanup=False)
+        self._assert_full(snap1)
+
+        # remove the would-be parent behind CloudStack's back
+        if not self._delete_backed_up_snapshot_file(snap1):
+            self.skipTest("Could not remove the backed-up snapshot from secondary storage")
+
+        # the driver still offers snap1 as parent (its ref is Ready), but the agent must
+        # notice the missing file, fall back to a full copy and report it as such, upon
+        # which the management server drops the parent link again
+        snap2 = self._snapshot_root_volume(vm)
+        self._assert_full(snap2, full=snap1)
+
+        # snap1's backing file is gone; delete it via the API and tolerate the missing file
+        try:
+            Snapshot.delete(snap1, self.apiClient)
+        except Exception as e:
+            logging.debug("Deleting the sabotaged snapshot failed (tolerated): %s" % e)
+
+    @attr(tags=['basic'], required_hardware=True)
+    def test_06_delete_mid_chain_member(self):
+        """Deleting a mid-chain delta keeps its children restorable (leaf-first physical deletion)."""
+        vm = self._deploy_vm("TestIncrVM-del")
+
+        self._write_marker(vm, "incr-del-v1")
+        snap1 = self._snapshot_root_volume(vm)
+
+        self._write_marker(vm, "incr-del-v2")
+        snap2 = self._snapshot_root_volume(vm, add_to_cleanup=False)
+        self._assert_delta(snap2, parent=snap1)
+
+        self._write_marker(vm, "incr-del-v3")
+        snap3 = self._snapshot_root_volume(vm)
+        self._assert_delta(snap3, parent=snap2, full=snap1)
+
+        # snap3's delta is backed by snap2's file: deleting snap2 must not break snap3
+        Snapshot.delete(snap2, self.apiClient)
+
+        vm.stop(self.apiClient)
+
+        Volume.revertToSnapshot(self.apiClient, snap3.id)
+        self._start_vm(vm)
+        self.assertEqual("incr-del-v3", self._read_marker(vm),
+                         "Revert to a delta broke after its parent snapshot was deleted")
+
+    @attr(tags=['basic'], required_hardware=True)
+    def test_07_encrypted_volumes_stay_full(self):
+        """Snapshots of encrypted volumes are never incremental (a delta would need the LUKS secret)."""
+        if not self._encryption_capable_host_exists():
+            self.skipTest("No KVM host with volume-encryption support found")
+
+        offering_data = dict(self.testdata[TestData.computeOffering])
+        offering_data["name"] = offering_data["displaytext"] = "Linstor_Compute_Incr_Enc"
+        offering = ServiceOffering.create(self.apiClient, offering_data, encryptroot=True)
+        self.cleanup.append(offering)
+
+        vm = VirtualMachine.create(
+            self.apiClient,
+            {"name": "TestIncrVM-enc", "displayname": "TestIncrVM-enc"},
+            accountid=self.account.name,
+            zoneid=self.zone.id,
+            serviceofferingid=offering.id,
+            templateid=self.template.id,
+            domainid=self.domain.id,
+            startvm=False,
+            mode='basic',
+        )
+        self.cleanup.insert(0, vm)
+        # the root volume is only provisioned on first start; a never-started
+        # VM's volume stays Allocated and cannot be snapshotted
+        self._start_vm(vm)
+
+        snap1 = self._snapshot_root_volume(vm)
+        snap2 = self._snapshot_root_volume(vm)
+
+        self._assert_full(snap1)
+        self._assert_full(snap2, full=snap1)
+
+    # --------------------------------------------------------------------- #
+    # Helpers
+    # --------------------------------------------------------------------- #
+
+    @classmethod
+    def _set_config(cls, name, value):
+        """Set a global configuration, remembering the original value for tearDownClass."""
+        if name not in cls.original_config:
+            cls.original_config[name] = Configurations.list(cls.apiClient, name=name)[0].value
+        Configurations.update(cls.apiClient, name=name, value=value)
+
+    @classmethod
+    def _restore_configs(cls):
+        for name, value in cls.original_config.items():
+            try:
+                Configurations.update(cls.apiClient, name=name, value=value)
+            except Exception as e:
+                logging.debug("Could not restore configuration %s=%s: %s" % (name, value, e))
+
+    def _deploy_vm(self, name):
+        vm = VirtualMachine.create(
+            self.apiClient,
+            {"name": name, "displayname": name},
+            accountid=self.account.name,
+            zoneid=self.zone.id,
+            serviceofferingid=self.compute_offering.id,
+            templateid=self.template.id,
+            domainid=self.domain.id,
+            startvm=False,
+            mode='basic',
+        )
+        self.cleanup.insert(0, vm)
+        self._start_vm(vm)
+        return vm
+
+    def _snapshot_root_volume(self, vm, add_to_cleanup=True):
+        root = list_volumes(self.apiClient, virtualmachineid=vm.id, type="ROOT", listall=True)[0]
+        snapshot = Snapshot.create(
+            self.apiClient,
+            volume_id=root.id,
+            account=self.account.name,
+            domainid=self.domain.id,
+        )
+        self.assertIsNotNone(snapshot, "Could not create snapshot of root volume")
+        if add_to_cleanup:
+            self.cleanup.insert(0, snapshot)
+        return snapshot
+
+    def _image_store_ref(self, snapshot):
+        """The snapshot's image-store reference bookkeeping, or None if the DB is not reachable.
+
+        The parent link and end-of-chain flag are internal to the management server and not part of
+        any API response, so they can only be checked directly in the database.
+        """
+        try:
+            rows = self.dbConnection.execute(
+                "SELECT ss.parent_snapshot_id, ss.end_of_chain, ss.physical_size, ss.install_path "
+                "FROM snapshot_store_ref ss JOIN snapshots s ON s.id = ss.snapshot_id "
+                "WHERE s.uuid = '%s' AND ss.store_role = 'Image' AND ss.state != 'Destroyed'"
+                % snapshot.id)
+        except Exception as e:
+            logging.debug("DB lookup of the image store ref failed: %s" % e)
+            return None
+        if not rows:
+            return None
+        return {
+            "parent_snapshot_id": rows[0][0],
+            "end_of_chain": bool(rows[0][1]),
+            "physical_size": rows[0][2],
+            "install_path": rows[0][3],
+        }
+
+    def _db_snapshot_id(self, snapshot):
+        rows = self.dbConnection.execute("SELECT id FROM snapshots WHERE uuid = '%s'" % snapshot.id)
+        return rows[0][0] if rows else None
+
+    def _assert_full(self, snapshot, full=None):
+        """Assert the snapshot was backed up as a standalone full copy.
+
+        Without DB access the parent link cannot be checked; if a reference full backup of the
+        same volume is given, fall back to comparing physical sizes (a full is in the order of
+        the volume's used space, a delta of a small change is orders of magnitude smaller).
+        """
+        ref = self._image_store_ref(snapshot)
+        if ref is not None:
+            self.assertEqual(
+                0, ref["parent_snapshot_id"],
+                "Snapshot %s should be a full backup but has parent %s" % (snapshot.name, ref["parent_snapshot_id"]))
+        elif full is not None:
+            self.assertGreater(
+                int(snapshot.physicalsize) * 2, int(full.physicalsize),
+                "Snapshot %s (%s bytes) does not look like a full backup (reference full %s is %s bytes)"
+                % (snapshot.name, snapshot.physicalsize, full.name, full.physicalsize))
+        else:
+            logging.debug("No DB access - cannot verify that snapshot %s is a full backup" % snapshot.name)
+
+    def _assert_delta(self, snapshot, parent, full=None):
+        """Assert the snapshot was backed up as a delta of the given parent.
+
+        Without DB access the parent link cannot be checked; the fallback compares the size
+        against the chain's full backup (never against the parent - that may itself be a
+        similarly sized delta).
+        """
+        ref = self._image_store_ref(snapshot)
+        if ref is not None:
+            self.assertEqual(
+                self._db_snapshot_id(parent), ref["parent_snapshot_id"],
+                "Snapshot %s should be an incremental backup of %s" % (snapshot.name, parent.name))
+        else:
+            if full is None:
+                full = parent
+            self.assertLess(
+                int(snapshot.physicalsize) * 4, int(full.physicalsize),
+                "Snapshot %s (%s bytes) does not look like a delta (full backup %s is %s bytes)"
+                % (snapshot.name, snapshot.physicalsize, full.name, full.physicalsize))
+
+    def _vm_ssh(self, vm):
+        # The VM is deployed stopped, so its instance has no ssh_ip yet; the IP may also change across
+        # stop/start cycles. Always pass the current address from a fresh lookup.
+        ipaddress = self._get_vm(vm.id).ipaddress
+        return vm.get_ssh_client(ipaddress=ipaddress, reconnect=True, retries=5)
+
+    def _write_marker(self, vm, content):
+        ssh = self._vm_ssh(vm)
+        ssh.execute("echo '%s' > %s" % (content, MARKER_PATH))
+        ssh.execute("sync")
+
+    def _read_marker(self, vm):
+        ssh = self._vm_ssh(vm)
+        result = ssh.execute("cat %s" % MARKER_PATH)
+        return result[0].strip() if result else None
+
+    @classmethod
+    def _encryption_capable_host_exists(cls):
+        hosts = list_hosts(cls.apiClient, zoneid=cls.zone.id, type='Routing', hypervisor='KVM', state='Up')
+        return any(getattr(h, "encryptionsupported", False) for h in (hosts or []))
+
+    @classmethod
+    def _get_vm(cls, vm_id):
+        return list_virtual_machines(cls.apiClient, id=vm_id)[0]
+
+    @classmethod
+    def _start_vm(cls, vm, ssh_timeout=120):
+        vm_for_check = cls._get_vm(vm.id)
+        if vm_for_check.state == VirtualMachine.STOPPED:
+            vm.start(cls.apiClient)
+            vm_for_check = cls._get_vm(vm.id)
+        ServiceReady.wait_ssh_ready(vm_for_check.ipaddress, timeout=ssh_timeout)
+        return vm_for_check
+
+    def _host_ssh(self):
+        host = list_hosts(self.apiClient, type='Routing', hypervisor='KVM', state='Up')[0]
+        return SshClient(
+            host=host.ipaddress, port=22,
+            user=self.hostConfig['username'], passwd=self.hostConfig['password'])
+
+    def _on_mounted_secondary(self, snapshot, action):
+        """Self-mount the secondary NFS export on a host and run action(ssh, snapshot_path) on it."""
+        ref = self._image_store_ref(snapshot)
+        if ref is None or not ref["install_path"]:
+            return None
+        try:
+            store = self.dbConnection.execute(
+                "SELECT url FROM image_store WHERE role = 'Image' AND removed IS NULL LIMIT 1")
+        except Exception as e:
+            logging.debug("DB lookup of the image store url failed: %s" % e)
+            return None
+        if not store or not store[0][0] or not store[0][0].startswith("nfs://"):
+            return None
+        server, export = store[0][0][len("nfs://"):].split("/", 1)
+
+        ssh = self._host_ssh()
+        mount_point = "/tmp/cs_sectest_%d" % random.randint(0, 100000)
+        try:
+            ssh.execute("mkdir -p %s" % mount_point)
+            ssh.execute("mount -t nfs %s:/%s %s" % (server, export, mount_point))
+            return action(ssh, "%s/%s" % (mount_point, ref["install_path"]))
+        except Exception as e:
+            logging.debug("Action on mounted secondary storage failed: %s" % e)
+            return None
+        finally:
+            ssh.execute("umount %s 2>/dev/null; rmdir %s 2>/dev/null" % (mount_point, mount_point))
+
+    def _qemu_img_info_of_backed_up_snapshot(self, snapshot):
+        if not self.hostConfig:
+            return None
+
+        def qemu_img_info(ssh, path):
+            out = ssh.execute("qemu-img info --output=json %s" % path)
+            return json.loads("".join(out)) if out else None
+
+        return self._on_mounted_secondary(snapshot, qemu_img_info)
+
+    def _delete_backed_up_snapshot_file(self, snapshot):
+        def delete_file(ssh, path):
+            ssh.execute("rm -f %s" % path)
+            return True
+
+        return bool(self._on_mounted_secondary(snapshot, delete_file))
diff --git a/test/integration/plugins/linstor/test_linstor_volumes.py b/test/integration/plugins/linstor/test_linstor_volumes.py
index c2c220bfe9e9..a6883afe0333 100644
--- a/test/integration/plugins/linstor/test_linstor_volumes.py
+++ b/test/integration/plugins/linstor/test_linstor_volumes.py
@@ -19,7 +19,6 @@
 import os
 import random
 import time
-import socket
 
 # All tests inherit from cloudstackTestCase
 from marvin.cloudstackTestCase import cloudstackTestCase
@@ -38,6 +37,8 @@
 from marvin.codes import PASS
 from nose.plugins.attrib import attr
 
+from linstor_test_utils import ServiceReady
+
 # Prerequisites:
 #  Only one zone
 #  Only one pod
@@ -226,45 +227,6 @@ def __init__(self, linstor_controller_url):
             },
         }
 
-class ServiceReady:
-    @classmethod
-    def ready(cls, hostname: str, port: int) -> bool:
-        try:
-            s = socket.create_connection((hostname, port), timeout=1)
-            s.close()
-            return True
-        except (ConnectionRefusedError, socket.timeout, OSError):
-            return False
-
-    @classmethod
-    def wait(
-            cls,
-            hostname,
-            port,
-            wait_interval = 5,
-            timeout = 90,
-            service_name = 'ssh') -> bool:
-        """
-        Wait until the controller can be reached.
-        :param hostname:
-        :param port: port of the application
-        :param wait_interval:
-        :param timeout: time to wait until exit with False
-        :param service_name: name of the service to wait
-        :return:
-        """
-        starttime = int(round(time.time() * 1000))
-        while not cls.ready(hostname, port):
-            if starttime + timeout * 1000 < int(round(time.time() * 1000)):
-                raise RuntimeError("{s} {h} cannot be reached.".format(s=service_name, h=hostname))
-            time.sleep(wait_interval)
-        return True
-
-    @classmethod
-    def wait_ssh_ready(cls, hostname, wait_interval = 1, timeout = 90):
-        return cls.wait(hostname, 22, wait_interval, timeout, "ssh")
-
-
 class TestLinstorVolumes(cloudstackTestCase):
     _volume_vm_id_and_vm_id_do_not_match_err_msg = "The volume's VM ID and the VM's ID do not match."
     _vm_not_in_running_state_err_msg = "The VM is not in the 'Running' state."

From 9f97aeeef1c2d55281fb5de0167c42f725dd94d7 Mon Sep 17 00:00:00 2001
From: Rene Peinthor <rene.peinthor@linbit.com>
Date: Fri, 7 Aug 2026 11:49:20 +0200
Subject: [PATCH 6/6] address review comments round 2

- SnapshotDataStoreDaoImpl: use an explicit JOIN instead of the implicit
  comma join in FIND_SNAPSHOT_IN_ZONE.
- QemuImgTest: cover creating an image whose format differs from its
  backing file's format (qcow2 overlay on a raw backing file), the case
  enabled by passing the created file's format to -f.
---
 .../db/SnapshotDataStoreDaoImpl.java          |  6 +++---
 .../cloudstack/utils/qemu/QemuImgTest.java    | 19 +++++++++++++++++++
 2 files changed, 22 insertions(+), 3 deletions(-)

diff --git a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
index fbb2767060c6..8a044cd8211c 100644
--- a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
+++ b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDaoImpl.java
@@ -105,9 +105,9 @@ public class SnapshotDataStoreDaoImpl extends GenericDaoBase<SnapshotDataStoreVO
     // prefer the reference carrying the incremental chain link: content-based (Linstor) chains only
     // link their image store reference, the primary one stays unlinked, and chain walks (e.g.
     // snapshot.delta.max enforcement) must not end up on the unlinked copy
-    private static final String FIND_SNAPSHOT_IN_ZONE = "SELECT ssr.* FROM " +
-            "snapshot_store_ref ssr, snapshots s " +
-            "WHERE ssr.snapshot_id=? AND ssr.snapshot_id = s.id AND s.data_center_id=? " +
+    private static final String FIND_SNAPSHOT_IN_ZONE = "SELECT ssr.* FROM snapshot_store_ref ssr " +
+            "JOIN snapshots s ON ssr.snapshot_id = s.id " +
+            "WHERE ssr.snapshot_id = ? AND s.data_center_id = ? " +
             "ORDER BY (ssr.parent_snapshot_id > 0) DESC, (ssr.kvm_checkpoint_path IS NOT NULL) DESC LIMIT 1;";
 
     private static final String GET_PHYSICAL_SIZE_OF_SNAPSHOTS_ON_PRIMARY_BY_ACCOUNT = "SELECT SUM(s.physical_size) " +
diff --git a/plugins/hypervisors/kvm/src/test/java/org/apache/cloudstack/utils/qemu/QemuImgTest.java b/plugins/hypervisors/kvm/src/test/java/org/apache/cloudstack/utils/qemu/QemuImgTest.java
index 15f6785c1fd4..017ca7391d15 100644
--- a/plugins/hypervisors/kvm/src/test/java/org/apache/cloudstack/utils/qemu/QemuImgTest.java
+++ b/plugins/hypervisors/kvm/src/test/java/org/apache/cloudstack/utils/qemu/QemuImgTest.java
@@ -316,6 +316,25 @@ public void testCreateWithBackingFile() throws QemuImgException, LibvirtExceptio
         }
     }
 
+    @Test
+    public void testCreateWithBackingFileOfOtherFormat() throws QemuImgException, LibvirtException {
+        String backingFileName = "/tmp/" + UUID.randomUUID() + ".raw";
+        String overlayFileName = "/tmp/" + UUID.randomUUID() + ".qcow2";
+
+        QemuImgFile backingFile = new QemuImgFile(backingFileName, 20480, PhysicalDiskFormat.RAW);
+        QemuImgFile overlayFile = new QemuImgFile(overlayFileName, 20480, PhysicalDiskFormat.QCOW2);
+
+        QemuImg qemu = new QemuImg(0);
+        qemu.create(backingFile);
+        qemu.create(overlayFile, backingFile);
+
+        // the created file keeps its own format, the backing file's format is only passed as -F
+        Map<String, String> info = qemu.info(overlayFile);
+        assertEquals(PhysicalDiskFormat.QCOW2.toString(), info.get(QemuImg.FILE_FORMAT));
+        assertEquals(backingFileName, info.get(QemuImg.BACKING_FILE));
+        assertEquals(PhysicalDiskFormat.RAW.toString(), info.get(QemuImg.BACKING_FILE_FORMAT));
+    }
+
     @Test
     public void testConvertBasic() throws QemuImgException, LibvirtException {
         long srcSize = 20480;