feat(backup): anchor incremental chain on VM active_checkpoint_id

Per @abh1sar's review at NASBackupProvider.java:251, the "find the latest
backup with a bitmap" heuristic in decideChain breaks after a restore — the
restored disk image has no QEMU dirty-bitmap, so taking the last-backed-up
row as parent for the next incremental produces a chain whose first link
references blocks that no longer exist.

Track the active checkpoint per-VM in vm_instance_details under the new
nas.active_checkpoint_id key:

* takeBackup writes it after every successful backup (cleared on the
  stopped-VM offline path where no bitmap is created).
* restoreVMFromBackup / restoreBackupToVM clear it on success, so the next
  backup is automatically a full and starts a fresh chain.
* decideChain reads it first: null => full; non-null => find the matching
  bitmap_name on a BackedUp backup row; if none, fall back to full.

The new VM_ACTIVE_CHECKPOINT_ID key lives next to the existing chain keys
in NASBackupChainKeys (it's a vm_instance_details detail, not a backup
detail, so no schema migration is needed).

Author: jmsperu

plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupChainKeys.java

@@ -45,6 +45,15 @@ public final class NASBackupChainKeys {
     /** Set to the bitmap name when this incremental had to recreate its parent bitmap on the host (informational; this incremental is larger than usual). */
     public static final String BITMAP_RECREATED = "nas.bitmap_recreated";
 
+    /**
+     * VM-scoped detail (stored in {@code vm_instance_details}) holding the QEMU dirty-bitmap
+     * name that currently exists on the running VM and is therefore the only valid parent
+     * for the next incremental backup. Written by {@link #BITMAP_NAME} on each successful
+     * backup; cleared on restore (the restored disk image has no bitmap, so the next backup
+     * must be a fresh full). When the VM has no detail, {@code decideChain} forces full.
+     */
+    public static final String VM_ACTIVE_CHECKPOINT_ID = "nas.active_checkpoint_id";
+
     private NASBackupChainKeys() {
     }
 }

plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java

@@ -38,8 +38,10 @@
 import com.cloud.utils.Pair;
 import com.cloud.utils.component.AdapterBase;
 import com.cloud.utils.exception.CloudRuntimeException;
+import com.cloud.vm.VMInstanceDetailVO;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.dao.VMInstanceDao;
+import com.cloud.vm.dao.VMInstanceDetailsDao;
 import com.cloud.vm.snapshot.VMSnapshot;
 import com.cloud.vm.snapshot.VMSnapshotDetailsVO;
 import com.cloud.vm.snapshot.VMSnapshotVO;
@@ -117,6 +119,9 @@ public class NASBackupProvider extends AdapterBase implements BackupProvider, Co
     @Inject
     private VMInstanceDao vmInstanceDao;
 
+    @Inject
+    private VMInstanceDetailsDao vmInstanceDetailsDao;
+
     @Inject
     private PrimaryDataStoreDao primaryDataStoreDao;
 
@@ -226,6 +231,12 @@ boolean isIncremental() {
      * appended to the existing chain. Stopped VMs are always full (libvirt {@code backup-begin}
      * requires a running QEMU process). The {@code nas.backup.full.every} ConfigKey controls
      * how many backups (full + incrementals) form one chain before a new full is forced.
+     *
+     * <p>The decision is anchored on the VM's {@code nas.active_checkpoint_id} detail, which
+     * records the bitmap that currently exists on the running QEMU. After a restore that
+     * detail is cleared, so the next backup is automatically full — even though there may be
+     * a more recent "last backup taken" row in the database. This matches the prescription in
+     * the PR review (avoid relying on "last backup" because that breaks after restore).</p>
      */
     protected ChainDecision decideChain(VirtualMachine vm) {
         final String newBitmap = "backup-" + System.currentTimeMillis() / 1000L;
@@ -242,38 +253,29 @@ protected ChainDecision decideChain(VirtualMachine vm) {
             return ChainDecision.fullStart(newBitmap);
         }
 
-        // Walk this VM's backups newest→oldest, find the most recent BackedUp backup that has a
-        // bitmap stored. If we don't find one, this is the first backup in a chain — start full.
-        List<Backup> history = backupDao.listByVmId(vm.getDataCenterId(), vm.getId());
-        if (history == null || history.isEmpty()) {
+        // 1. If the VM has no active_checkpoint_id, there is no bitmap on the host to use as
+        //    a parent. This is the case after restore (we clear it), after VM was just assigned
+        //    to the offering, or on the very first backup.
+        String activeCheckpoint = readVmActiveCheckpoint(vm.getId());
+        if (activeCheckpoint == null) {
             return ChainDecision.fullStart(newBitmap);
         }
-        history.sort(Comparator.comparing(Backup::getDate).reversed());
 
-        Backup parent = null;
-        String parentBitmap = null;
-        String parentChainId = null;
-        int parentChainPosition = -1;
-        for (Backup b : history) {
-            if (!Backup.Status.BackedUp.equals(b.getStatus())) {
-                continue;
-            }
-            String bm = readDetail(b, NASBackupChainKeys.BITMAP_NAME);
-            if (bm == null) {
-                continue;
-            }
-            parent = b;
-            parentBitmap = bm;
-            parentChainId = readDetail(b, NASBackupChainKeys.CHAIN_ID);
-            String posStr = readDetail(b, NASBackupChainKeys.CHAIN_POSITION);
-            try {
-                parentChainPosition = posStr == null ? 0 : Integer.parseInt(posStr);
-            } catch (NumberFormatException e) {
-                parentChainPosition = 0;
-            }
-            break;
+        // 2. Find the latest BackedUp backup of this VM whose BITMAP_NAME matches the VM's
+        //    active_checkpoint_id. Only that backup is a safe parent — any earlier backup
+        //    would have a bitmap that QEMU may have rotated out. Per the review:
+        //      "The latest backup should have the bitmap_name equal to the VM's
+        //       active_checkpoint_id which will become the parent backup. If not, force full."
+        Backup parent = findLatestBackedUpBackupWithBitmap(vm.getId(), activeCheckpoint);
+        if (parent == null) {
+            LOG.debug("VM {} has active_checkpoint_id={} but no matching backup found — forcing full",
+                    vm.getInstanceName(), activeCheckpoint);
+            return ChainDecision.fullStart(newBitmap);
         }
-        if (parent == null || parentBitmap == null || parentChainId == null) {
+
+        String parentChainId = readDetail(parent, NASBackupChainKeys.CHAIN_ID);
+        int parentChainPosition = chainPosition(parent);
+        if (parentChainId == null || parentChainPosition == Integer.MAX_VALUE) {
             return ChainDecision.fullStart(newBitmap);
         }
 
@@ -286,10 +288,44 @@ protected ChainDecision decideChain(VirtualMachine vm) {
         // qcow2 onto it. The path is stored relative to the NAS mount point — the script
         // resolves it inside its mount session.
         String parentPath = composeParentBackupPath(parent);
-        return ChainDecision.incremental(newBitmap, parentBitmap, parentPath,
+        return ChainDecision.incremental(newBitmap, activeCheckpoint, parentPath,
                 parentChainId, parentChainPosition + 1);
     }
 
+    /**
+     * Read the {@code nas.active_checkpoint_id} VM detail. Returns {@code null} when no detail
+     * exists (post-restore, first backup, or after explicit reset).
+     */
+    private String readVmActiveCheckpoint(long vmId) {
+        VMInstanceDetailVO d = vmInstanceDetailsDao.findDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID);
+        if (d == null) {
+            return null;
+        }
+        String v = d.getValue();
+        return (v == null || v.isEmpty()) ? null : v;
+    }
+
+    /**
+     * Locate the most-recent {@code BackedUp} backup for {@code vmId} whose bitmap name equals
+     * {@code bitmapName}. Used by {@link #decideChain} to anchor the next incremental.
+     */
+    private Backup findLatestBackedUpBackupWithBitmap(long vmId, String bitmapName) {
+        List<Backup> history = backupDao.listByVmId(null, vmId);
+        if (history == null || history.isEmpty()) {
+            return null;
+        }
+        history.sort(Comparator.comparing(Backup::getDate).reversed());
+        for (Backup b : history) {
+            if (!Backup.Status.BackedUp.equals(b.getStatus())) {
+                continue;
+            }
+            if (bitmapName.equals(readDetail(b, NASBackupChainKeys.BITMAP_NAME))) {
+                return b;
+            }
+        }
+        return null;
+    }
+
     private String readDetail(Backup backup, String key) {
         BackupDetailVO d = backupDetailsDao.findDetail(backup.getId(), key);
         return d == null ? null : d.getValue();
@@ -331,6 +367,34 @@ private void persistChainMetadata(Backup backup, ChainDecision decision, String
         }
     }
 
+    /**
+     * Upsert the VM's {@code nas.active_checkpoint_id} detail to {@code bitmapName}. Called
+     * after every successful backup so the next backup's parent-bitmap decision is anchored
+     * on what actually exists on QEMU, not on "last backup taken".
+     */
+    private void upsertVmActiveCheckpoint(long vmId, String bitmapName) {
+        VMInstanceDetailVO existing = vmInstanceDetailsDao.findDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID);
+        if (existing == null) {
+            vmInstanceDetailsDao.addDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID, bitmapName, false);
+            return;
+        }
+        existing.setValue(bitmapName);
+        vmInstanceDetailsDao.update(existing.getId(), existing);
+    }
+
+    /**
+     * Remove the VM's {@code nas.active_checkpoint_id} detail. Called from the restore paths:
+     * after restore the disk image has no QEMU bitmap attached, so any future incremental
+     * would be based on stale state. Clearing forces the next backup to be a fresh full.
+     */
+    private void clearVmActiveCheckpoint(long vmId) {
+        VMInstanceDetailVO existing = vmInstanceDetailsDao.findDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID);
+        if (existing != null) {
+            vmInstanceDetailsDao.removeDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID);
+            LOG.debug("Cleared nas.active_checkpoint_id for VM id={} (was {})", vmId, existing.getValue());
+        }
+    }
+
     private String lookupParentBackupUuid(long vmId, String parentBitmap) {
         if (parentBitmap == null) {
             return null;
@@ -440,6 +504,18 @@ public Pair<Boolean, Backup> takeBackup(final VirtualMachine vm, Boolean quiesce
                     logger.info("NAS incremental for VM {} recreated parent bitmap {} (likely VM was restarted since last backup)",
                             vm.getInstanceName(), answer.getBitmapRecreated());
                 }
+                // Pin the VM's active_checkpoint_id to whichever bitmap the agent actually
+                // created. This is the only valid parent for the next incremental (see
+                // decideChain). For the stopped-VM offline path BITMAP_CREATED is null —
+                // no bitmap exists on the host, so we also clear any stale detail from a
+                // prior online backup. Either way, after this step the detail accurately
+                // reflects what's on the running QEMU (or absence thereof).
+                String confirmedBitmap = answer.getBitmapCreated();
+                if (confirmedBitmap != null) {
+                    upsertVmActiveCheckpoint(vm.getId(), confirmedBitmap);
+                } else {
+                    clearVmActiveCheckpoint(vm.getId());
+                }
                 return new Pair<>(true, backupVO);
             } else {
                 throw new CloudRuntimeException("Failed to update backup");
@@ -531,6 +607,11 @@ private Pair<Boolean, String> restoreVMBackup(VirtualMachine vm, Backup backup)
         } catch (OperationTimedoutException e) {
             throw new CloudRuntimeException("Operation to restore backup timed out, please try again");
         }
+        // After a restore the QEMU dirty-bitmap chain is gone — clear active_checkpoint_id so
+        // the next backup is taken as a fresh full and starts a new chain. See decideChain.
+        if (answer != null && answer.getResult()) {
+            clearVmActiveCheckpoint(vm.getId());
+        }
         return new Pair<>(answer.getResult(), answer.getDetails());
     }
 

plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java

@@ -48,8 +48,10 @@
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.VolumeDao;
 import com.cloud.utils.Pair;
+import com.cloud.vm.VMInstanceDetailVO;
 import com.cloud.vm.VMInstanceVO;
 import com.cloud.vm.dao.VMInstanceDao;
+import com.cloud.vm.dao.VMInstanceDetailsDao;
 
 import org.apache.cloudstack.backup.dao.BackupDao;
 import org.apache.cloudstack.backup.dao.BackupDetailsDao;
@@ -100,6 +102,9 @@ public class NASBackupProviderTest {
     @Mock
     private BackupDetailsDao backupDetailsDao;
 
+    @Mock
+    private VMInstanceDetailsDao vmInstanceDetailsDao;
+
     @Test
     public void testDeleteBackup() throws OperationTimedoutException, AgentUnavailableException {
         Long hostId = 1L;
@@ -357,4 +362,79 @@ public void testGetVMHypervisorHostFallbackToZoneWideKVMHost() {
         Mockito.verify(hostDao).findHypervisorHostInCluster(clusterId);
         Mockito.verify(resourceManager).findOneRandomRunningHostByHypervisor(Hypervisor.HypervisorType.KVM, zoneId);
     }
+
+    // -- decideChain anchored on VM's active_checkpoint_id -------------------------------
+
+    /**
+     * No active_checkpoint_id on the VM (post-restore, first-ever backup, or detail purged) =>
+     * decideChain must return a fresh full. This is the regression abh1sar called out at
+     * NASBackupProvider.java:251 ("This logic breaks after restore. We can't use the last
+     * backup taken as parent in that case.").
+     */
+    @Test
+    public void decideChainReturnsFullWhenVmHasNoActiveCheckpoint() {
+        Long zoneId = 1L;
+        Long vmId = 42L;
+        VMInstanceVO vm = mock(VMInstanceVO.class);
+        Mockito.when(vm.getId()).thenReturn(vmId);
+        Mockito.when(vm.getDataCenterId()).thenReturn(zoneId);
+        Mockito.when(vm.getState()).thenReturn(VMInstanceVO.State.Running);
+
+        Mockito.when(vmInstanceDetailsDao.findDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID)).thenReturn(null);
+
+        NASBackupProvider.ChainDecision decision = nasBackupProvider.decideChain(vm);
+        Assert.assertNotNull(decision);
+        Assert.assertEquals(NASBackupChainKeys.TYPE_FULL, decision.mode);
+        Assert.assertNull(decision.bitmapParent);
+        Assert.assertEquals(0, decision.chainPosition);
+    }
+
+    /**
+     * After a successful restoreVMFromBackup, decideChain on the next backup must produce
+     * a full. We verify by checking that vmInstanceDetailsDao.removeDetail is called with
+     * the active_checkpoint_id key.
+     */
+    @Test
+    public void restoreClearsActiveCheckpointDetail() throws AgentUnavailableException, OperationTimedoutException {
+        Long vmId = 7L;
+        Long hostId = 8L;
+        Long backupOfferingId = 9L;
+
+        VMInstanceVO vm = mock(VMInstanceVO.class);
+        Mockito.when(vm.getId()).thenReturn(vmId);
+        Mockito.when(vm.getLastHostId()).thenReturn(hostId);
+        Mockito.when(vm.getRemoved()).thenReturn(null);
+        Mockito.when(vm.getName()).thenReturn("vm7");
+
+        HostVO host = mock(HostVO.class);
+        Mockito.when(host.getStatus()).thenReturn(Status.Up);
+        Mockito.when(host.getId()).thenReturn(hostId);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(host);
+
+        BackupVO backup = new BackupVO();
+        backup.setVmId(vmId);
+        backup.setBackupOfferingId(backupOfferingId);
+        backup.setExternalId("i-2-7-VM/2026.05.16.10.00.00");
+        ReflectionTestUtils.setField(backup, "id", 100L);
+        // backedUpVolumes defaults to null => BackupVO.getBackedUpVolumes returns emptyList().
+
+        BackupRepositoryVO repo = new BackupRepositoryVO(1L, "nas", "test-repo",
+                "nfs", "address", "sync", 1024L, null);
+        Mockito.when(backupRepositoryDao.findByBackupOfferingId(backupOfferingId)).thenReturn(repo);
+
+        Mockito.when(volumeDao.findByInstance(vmId)).thenReturn(Collections.emptyList());
+
+        BackupAnswer answer = mock(BackupAnswer.class);
+        Mockito.when(answer.getResult()).thenReturn(true);
+        Mockito.when(agentManager.send(Mockito.anyLong(), Mockito.any(RestoreBackupCommand.class))).thenReturn(answer);
+
+        // Pre-existing checkpoint detail so removeDetail has something to "clear".
+        VMInstanceDetailVO existing = mock(VMInstanceDetailVO.class);
+        Mockito.when(existing.getValue()).thenReturn("backup-1715000000");
+        Mockito.when(vmInstanceDetailsDao.findDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID)).thenReturn(existing);
+
+        boolean ok = nasBackupProvider.restoreVMFromBackup(vm, backup);
+        Assert.assertTrue(ok);
+        Mockito.verify(vmInstanceDetailsDao).removeDetail(vmId, NASBackupChainKeys.VM_ACTIVE_CHECKPOINT_ID);
+    }
 }