fix: retention issues with replication

@xen-orchestra/backups/_otherConfig.mjs

@@ -84,26 +84,43 @@ export async function getVmDeltaChainLength(xapi, vmRef) {
  *
  * Reset the other_config field related to backups of a VM and its VDIs
  *
- *
  * @param {Xapi} xapi
  * @param {String} vmRef
+ * @param {Object} [options]
+ * @param {Boolean} [options.isReplicationTarget] - keep replica specific keys on the VM itself
+ *
+ * The live target VM of an incremental replication is not a restore point — its snapshots
+ * are — so it must lose the content and chain markers, which would otherwise make it a
+ * candidate base for the next run. It must however stay findable by `listReplicatedVms`:
+ * when a run cannot chain onto it and imports into a new VM instead, an untagged target
+ * VM becomes an orphan that retention can never collect, and replicas pile up on the SR
+ * whatever the retention is set to.
+ *
+ * Its VDIs are still reset entirely: only the snapshot VDIs, taken before this call, need
+ * to carry the markers `checkBaseVdis` looks for.
+ *
  * @returns {Promise}
  */
-export function resetVmOtherConfig(xapi, vmRef) {
+export function resetVmOtherConfig(xapi, vmRef, { isReplicationTarget = false } = {}) {
   return applyToVmAndVdis(xapi, vmRef, (type, ref) => {
-    return xapi.setFieldEntries(type, ref, 'other_config', {
+    const otherConfig = {
       [CONTENT_KEY]: null,
       [COPY_OF]: null,
-      [DATETIME]: null,
       [DELTA_CHAIN_LENGTH]: null,
       [EXPORTED_SUCCESSFULLY]: null,
       [INCLUDE_NON_NBD_QCOW2_FIX]: null,
-      [JOB_ID]: null,
-      [SCHEDULE_ID]: null,
-      [VM_UUID]: null,
 
       // REPLICATED_TO_SR_UUID is not reset since we can replicate a replication
-    })
+    }
+
+    if (!(isReplicationTarget && type === 'VM')) {
+      otherConfig[DATETIME] = null
+      otherConfig[JOB_ID] = null
+      otherConfig[SCHEDULE_ID] = null
+      otherConfig[VM_UUID] = null
+    }
+
+    return xapi.setFieldEntries(type, ref, 'other_config', otherConfig)
   })
 }
 

@xen-orchestra/backups/_otherConfig.test.mjs

@@ -0,0 +1,97 @@
+import { describe, it } from 'node:test'
+import assert from 'node:assert/strict'
+
+import {
+  CONTENT_KEY,
+  COPY_OF,
+  DATETIME,
+  DELTA_CHAIN_LENGTH,
+  EXPORTED_SUCCESSFULLY,
+  INCLUDE_NON_NBD_QCOW2_FIX,
+  JOB_ID,
+  REPLICATED_TO_SR_UUID,
+  SCHEDULE_ID,
+  VM_UUID,
+  resetVmOtherConfig,
+} from './_otherConfig.mjs'
+
+// Records the other_config entries written per object, which is all resetVmOtherConfig does.
+const makeXapi = vdiRefs => {
+  const writes = {}
+  return {
+    writes,
+    VM_getDisks: async () => vdiRefs,
+    setFieldEntries: async (type, ref, field, entries) => {
+      assert.equal(field, 'other_config')
+      writes[ref] = { type, entries }
+    },
+  }
+}
+
+const REPLICA_IDENTITY_KEYS = [DATETIME, JOB_ID, SCHEDULE_ID, VM_UUID]
+const CHAIN_MARKER_KEYS = [CONTENT_KEY, COPY_OF, DELTA_CHAIN_LENGTH, EXPORTED_SUCCESSFULLY, INCLUDE_NON_NBD_QCOW2_FIX]
+
+describe('resetVmOtherConfig()', () => {
+  it('clears every backup key on the VM and its VDIs by default', async () => {
+    const xapi = makeXapi(['vdi-1', 'vdi-2'])
+
+    await resetVmOtherConfig(xapi, 'vm-ref')
+
+    for (const ref of ['vm-ref', 'vdi-1', 'vdi-2']) {
+      const { entries } = xapi.writes[ref]
+      for (const key of [...REPLICA_IDENTITY_KEYS, ...CHAIN_MARKER_KEYS]) {
+        assert.equal(entries[key], null, `${key} should be cleared on ${ref}`)
+      }
+    }
+  })
+
+  it('never clears REPLICATED_TO_SR_UUID, so a replication can be replicated', async () => {
+    const xapi = makeXapi(['vdi-1'])
+
+    await resetVmOtherConfig(xapi, 'vm-ref', { isReplicationTarget: true })
+
+    for (const ref of ['vm-ref', 'vdi-1']) {
+      assert.equal(REPLICATED_TO_SR_UUID in xapi.writes[ref].entries, false)
+    }
+  })
+})
+
+describe('regression: a replication target VM loses the keys retention finds it by', () => {
+  it('keeps the replica identity keys on the VM when isReplicationTarget is set', async () => {
+    const xapi = makeXapi(['vdi-1'])
+
+    await resetVmOtherConfig(xapi, 'vm-ref', { isReplicationTarget: true })
+
+    const { entries } = xapi.writes['vm-ref']
+    for (const key of REPLICA_IDENTITY_KEYS) {
+      // clearing these made the target VM invisible to listReplicatedVms(), so a target
+      // abandoned by a later run could never be collected and replicas piled up
+      assert.equal(key in entries, false, `${key} should be left untouched on a replication target VM`)
+    }
+  })
+
+  it('still clears the chain markers on the VM, so it cannot be picked as a base', async () => {
+    const xapi = makeXapi(['vdi-1'])
+
+    await resetVmOtherConfig(xapi, 'vm-ref', { isReplicationTarget: true })
+
+    const { entries } = xapi.writes['vm-ref']
+    for (const key of CHAIN_MARKER_KEYS) {
+      assert.equal(entries[key], null, `${key} should be cleared even on a replication target VM`)
+    }
+  })
+
+  it('still clears everything on the VDIs, whose markers live on the snapshots', async () => {
+    const xapi = makeXapi(['vdi-1', 'vdi-2'])
+
+    await resetVmOtherConfig(xapi, 'vm-ref', { isReplicationTarget: true })
+
+    for (const ref of ['vdi-1', 'vdi-2']) {
+      const { type, entries } = xapi.writes[ref]
+      assert.equal(type, 'VDI')
+      for (const key of [...REPLICA_IDENTITY_KEYS, ...CHAIN_MARKER_KEYS]) {
+        assert.equal(entries[key], null, `${key} should be cleared on ${ref}`)
+      }
+    }
+  })
+})

@xen-orchestra/backups/_runners/_writers/AggregatedIncrementalXapiWriter.mjs

@@ -6,13 +6,14 @@ export class AggregatedIncrementalXapiWriter extends AbstractAggregatedXapiWrite
   /**
    *
    * @param {Map<string,string>} baseUuidToSrcVdi
+   * @param {Map<string,string>} contentKeys
    */
-  async checkBaseVdis(baseUuidToSrcVdi) {
+  async checkBaseVdis(baseUuidToSrcVdi, contentKeys) {
     debug('checkBaseVdis', { baseUuidToSrcVdi })
     let selectedCopy = new Map()
     for (const writer of this.writers) {
       const copy = new Map(baseUuidToSrcVdi)
-      await writer.checkBaseVdis(copy)
+      await writer.checkBaseVdis(copy, contentKeys)
       if (copy.size > 0) {
         debug('checkBaseVdis found a mainwriter candidate ', writer._sr.name_label)
         // there can be multiple candidates

@xen-orchestra/backups/_runners/_writers/AggregatedIncrementalXapiWriter.test.mjs

@@ -0,0 +1,85 @@
+import { describe, it } from 'node:test'
+import assert from 'node:assert/strict'
+
+import { AggregatedIncrementalXapiWriter } from './AggregatedIncrementalXapiWriter.mjs'
+
+const SRS = [
+  { uuid: 'sr-1', name_label: 'SR 1' },
+  { uuid: 'sr-2', name_label: 'SR 2' },
+]
+
+// Stands in for IncrementalXapiWriter: only reproduces what checkBaseVdis() does with its
+// two arguments, including the `contentKeys.get()` that used to throw.
+const makeFakeWriterClass = ({ srWithABase } = {}) => {
+  const calls = []
+  class FakeReplicationWriter {
+    constructor({ sr }) {
+      this._sr = sr
+    }
+
+    async checkBaseVdis(baseUuidToSrcVdi, contentKeys) {
+      calls.push({ srUuid: this._sr.uuid, baseUuidToSrcVdi, contentKeys })
+      for (const baseUuid of [...baseUuidToSrcVdi.keys()]) {
+        // the real writer looks the base up by content key first
+        const hasBase = contentKeys.get(baseUuid) !== undefined && this._sr.uuid === srWithABase
+        if (!hasBase) {
+          baseUuidToSrcVdi.delete(baseUuid)
+        }
+      }
+    }
+  }
+  return { FakeReplicationWriter, calls }
+}
+
+const makeWriter = ReplicationWriter =>
+  new AggregatedIncrementalXapiWriter({
+    srs: SRS,
+    ReplicationWriter,
+    settings: { copyRetention: 1 },
+    job: { id: 'job-1', settings: {} },
+    scheduleId: 'schedule-1',
+    vmUuid: 'vm-1',
+  })
+
+describe('regression: AggregatedIncrementalXapiWriter.checkBaseVdis() drops the content keys', () => {
+  it('forwards the content keys to every underlying writer', async () => {
+    const { FakeReplicationWriter, calls } = makeFakeWriterClass({ srWithABase: 'sr-2' })
+    const writer = makeWriter(FakeReplicationWriter)
+    await writer.setupWriters()
+
+    const contentKeys = new Map([['base-1', 'content-key-1']])
+
+    // used to reject with "Cannot read properties of undefined (reading 'get')", which
+    // removed the writer from the whole job execution and silently skipped replication
+    await writer.checkBaseVdis(new Map([['base-1', 'src-vdi-1']]), contentKeys)
+
+    assert.equal(calls.length, SRS.length, 'every underlying writer should be asked')
+    for (const call of calls) {
+      assert.equal(call.contentKeys, contentKeys, `writer for ${call.srUuid} should receive the content keys map`)
+    }
+  })
+
+  it('selects as main writer the SR that has a usable base', async () => {
+    const { FakeReplicationWriter } = makeFakeWriterClass({ srWithABase: 'sr-2' })
+    const writer = makeWriter(FakeReplicationWriter)
+    await writer.setupWriters()
+
+    const baseUuidToSrcVdi = new Map([['base-1', 'src-vdi-1']])
+    await writer.checkBaseVdis(baseUuidToSrcVdi, new Map([['base-1', 'content-key-1']]))
+
+    assert.equal(writer.mainWriter?._sr.uuid, 'sr-2')
+    assert.deepEqual([...baseUuidToSrcVdi], [['base-1', 'src-vdi-1']], 'the base found on sr-2 should be kept')
+  })
+
+  it('keeps no base, and no main writer, when no SR has one', async () => {
+    const { FakeReplicationWriter } = makeFakeWriterClass({ srWithABase: undefined })
+    const writer = makeWriter(FakeReplicationWriter)
+    await writer.setupWriters()
+
+    const baseUuidToSrcVdi = new Map([['base-1', 'src-vdi-1']])
+    await writer.checkBaseVdis(baseUuidToSrcVdi, new Map([['base-1', 'content-key-1']]))
+
+    assert.equal(writer.mainWriter, undefined, 'prepare() should then fall back to a new SR')
+    assert.equal(baseUuidToSrcVdi.size, 0, 'the transfer must be a full')
+  })
+})

@xen-orchestra/backups/_runners/_writers/IncrementalXapiWriter.mjs

@@ -10,7 +10,7 @@ import { importIncrementalVm } from '../../_incrementalVm.mjs'
 
 import { AbstractIncrementalWriter } from './_AbstractIncrementalWriter.mjs'
 import { MixinXapiWriter } from './_MixinXapiWriter.mjs'
-import { compareReplicatedVmDatetime, listReplicatedVms } from './_listReplicatedVms.mjs'
+import { compareReplicatedVmDatetime, filterRetentionEntries, listReplicatedVms } from './_listReplicatedVms.mjs'
 import {
   COPY_OF,
   setVmOtherConfig,
@@ -344,16 +344,13 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
     const scheduleId = this._schedule.id
 
     // delete previous interrupted copies
-    ignoreErrors.call(asyncMapSettled(listReplicatedVms(xapi, scheduleId, undefined, vmUuid), vm => vm.$destroy))
+    ignoreErrors.call(
+      asyncMapSettled(listReplicatedVms(xapi, scheduleId, undefined, vmUuid), vm =>
+        vm.$destroy({ bypassBlockedOperation: true })
+      )
+    )
 
-    const allEntries = listReplicatedVms(xapi, scheduleId, srUuid, vmUuid)
-
-    // In the snapshot-based flow a non-snapshot VM (the live target) coexists with its
-    // snapshots (one per transfer). That VM must not be subject to retention — only its
-    // snapshots are. Build the set of VM refs that already have snapshots in the list so
-    // we can exclude them, while keeping old-style non-snapshot VMs (no snapshots).
-    const vmRefsWithSnapshots = new Set(allEntries.filter(e => e.is_a_snapshot).map(e => e.snapshot_of))
-    const retentionEntries = allEntries.filter(e => e.is_a_snapshot || !vmRefsWithSnapshots.has(e.$ref))
+    const retentionEntries = filterRetentionEntries(listReplicatedVms(xapi, scheduleId, srUuid, vmUuid))
     retentionEntries.sort(compareReplicatedVmDatetime)
     this._oldEntries = getOldEntries(settings.copyRetention - 1, retentionEntries)
 
@@ -471,7 +468,9 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
         await xapi.VM_snapshot(targetVmRef, {
           name_label: `${vm.name_label} - ${job.name} / ${schedule.name} ${formatFilenameDate(timestamp)}`,
         })
-        await resetVmOtherConfig(xapi, targetVmRef)
+        // keep the job/schedule/VM/datetime tags on the target VM: they are what makes it
+        // collectable by retention if a later run has to import into a new VM instead
+        await resetVmOtherConfig(xapi, targetVmRef, { isReplicationTarget: true })
       })
 
       return {

@xen-orchestra/backups/_runners/_writers/_AbstractAggregatedXapiWriter.mjs

@@ -1,6 +1,6 @@
 import { getOldEntries } from '../../_getOldEntries.mjs'
 import { createLogger } from '@xen-orchestra/log'
-import { compareReplicatedVmDatetime, listReplicatedVms } from './_listReplicatedVms.mjs'
+import { compareReplicatedVmDatetime, filterRetentionEntries, listReplicatedVms } from './_listReplicatedVms.mjs'
 import { asyncMapSettled } from '@xen-orchestra/async-map'
 
 const { debug } = createLogger('xo:backups:AbstractAggregatedXapiWriter')
@@ -67,7 +67,9 @@ export class AbstractAggregatedXapiWriter {
 
     const replicatedVms = this.#storageRepositories
       .map(sr => {
-        return listReplicatedVms(sr.$xapi, scheduleId, sr.uuid, vmUuid)
+        // per SR: a live target VM must only be spared by the snapshots living on that
+        // same SR, so the filtering cannot be done on the concatenated list
+        return filterRetentionEntries(listReplicatedVms(sr.$xapi, scheduleId, sr.uuid, vmUuid))
       })
       .flat(1)
       .filter(_ => !!_)

@xen-orchestra/backups/_runners/_writers/_listReplicatedVms.mjs

@@ -37,3 +37,23 @@ export function listReplicatedVms(xapi, scheduleOrJobId, srUuid, vmUuid) {
 
   return Object.values(vms).sort(compareReplicatedVmDatetime)
 }
+
+/**
+ * Keep, out of the entries returned by {@link listReplicatedVms}, the ones retention must
+ * consider.
+ *
+ * In the snapshot-based flow a live target VM coexists with its snapshots, one per
+ * transfer. Destroying that VM would destroy the snapshots it holds, so it is spared as
+ * long as it still has any. It becomes an ordinary entry once its last snapshot has been
+ * reaped, which is how a target VM abandoned by a chain reset is eventually collected.
+ *
+ * Old-style entries (one non-snapshot VM per transfer, hence no snapshots) are always
+ * kept in the list.
+ *
+ * @param {Array<XoVm>} entries
+ * @returns {Array<XoVm>}
+ */
+export function filterRetentionEntries(entries) {
+  const vmRefsWithSnapshots = new Set(entries.filter(_ => _.is_a_snapshot).map(_ => _.snapshot_of))
+  return entries.filter(_ => _.is_a_snapshot || !vmRefsWithSnapshots.has(_.$ref))
+}

@xen-orchestra/backups/_runners/_writers/_listReplicatedVms.test.mjs

@@ -0,0 +1,113 @@
+import { describe, it } from 'node:test'
+import assert from 'node:assert/strict'
+
+import { formatDateTime } from '@xen-orchestra/xapi'
+
+import { compareReplicatedVmDatetime, filterRetentionEntries, listReplicatedVms } from './_listReplicatedVms.mjs'
+import { getOldEntries } from '../../_getOldEntries.mjs'
+import { DATETIME, JOB_ID, REPLICATED_TO_SR_UUID, SCHEDULE_ID, VM_UUID } from '../../_otherConfig.mjs'
+
+const JOB = 'job-1'
+const SCHEDULE = 'schedule-1'
+const SR = 'sr-1'
+const VM = 'vm-1'
+
+// A replication target VM: non-snapshot, start blocked, tagged by the last successful run.
+const makeTargetVm = (ref, timestamp) => ({
+  $type: 'VM',
+  $id: ref,
+  $ref: ref,
+  name_label: `[XO Backup job] ${VM}`,
+  is_a_template: false,
+  is_a_snapshot: false,
+  blocked_operations: { start: 'blocked' },
+  other_config: {
+    [JOB_ID]: JOB,
+    [SCHEDULE_ID]: SCHEDULE,
+    [VM_UUID]: VM,
+    [REPLICATED_TO_SR_UUID]: SR,
+    [DATETIME]: formatDateTime(timestamp),
+  },
+})
+
+// A restore point: a snapshot of the target VM, one per successful transfer.
+const makeTargetSnapshot = (ref, targetRef, timestamp) => ({
+  ...makeTargetVm(ref, timestamp),
+  $id: ref,
+  $ref: ref,
+  is_a_snapshot: true,
+  snapshot_of: targetRef,
+})
+
+const makeXapi = objects => ({
+  objects: { indexes: { type: { VM: Object.fromEntries(objects.map(_ => [_.$ref, _])) } } },
+})
+
+describe('filterRetentionEntries()', () => {
+  it('spares a target VM as long as it holds snapshots', () => {
+    const target = makeTargetVm('target-a', 1)
+    const snapshot = makeTargetSnapshot('snapshot-a1', 'target-a', 1)
+
+    const kept = filterRetentionEntries([target, snapshot])
+
+    assert.deepEqual(
+      kept.map(_ => _.$ref),
+      ['snapshot-a1'],
+      'destroying the target VM would destroy the restore points it holds'
+    )
+  })
+
+  it('exposes a target VM that no longer holds any snapshot', () => {
+    // this is the abandoned target of a run that could not chain onto it: once retention
+    // has reaped its last snapshot, nothing else references it and it must be collectable
+    const abandoned = makeTargetVm('target-a', 1)
+    const current = makeTargetVm('target-b', 2)
+    const snapshot = makeTargetSnapshot('snapshot-b1', 'target-b', 2)
+
+    const kept = filterRetentionEntries([abandoned, current, snapshot])
+
+    assert.deepEqual(kept.map(_ => _.$ref).sort(), ['snapshot-b1', 'target-a'])
+  })
+
+  it('keeps old-style entries, which are non-snapshot VMs without snapshots', () => {
+    const entries = [makeTargetVm('old-1', 1), makeTargetVm('old-2', 2)]
+
+    assert.deepEqual(
+      filterRetentionEntries(entries).map(_ => _.$ref),
+      ['old-1', 'old-2']
+    )
+  })
+})
+
+describe('regression: replicated VMs pile up beyond the retention', () => {
+  // reproduces the state left by a run that could not reuse the previous target VM:
+  // target-a was abandoned, target-b is the new one, and retention is 1
+  const abandoned = makeTargetVm('target-a', 1)
+  const current = makeTargetVm('target-b', 2)
+  const currentSnapshot = makeTargetSnapshot('snapshot-b1', 'target-b', 2)
+
+  it('still finds an abandoned target VM after its other_config has been reset', () => {
+    // resetVmOtherConfig() used to strip JOB_ID/SCHEDULE_ID/VM_UUID/DATETIME from the
+    // target VM, making it invisible here and therefore impossible to ever collect
+    const xapi = makeXapi([abandoned, current, currentSnapshot])
+
+    assert.deepEqual(
+      listReplicatedVms(xapi, SCHEDULE, SR, VM).map(_ => _.$ref),
+      ['target-a', 'target-b', 'snapshot-b1']
+    )
+  })
+
+  it('collects the abandoned target VM, and only it, with a retention of 1', () => {
+    const xapi = makeXapi([abandoned, current, currentSnapshot])
+
+    const retentionEntries = filterRetentionEntries(listReplicatedVms(xapi, SCHEDULE, SR, VM))
+    retentionEntries.sort(compareReplicatedVmDatetime)
+    const oldEntries = getOldEntries(/* copyRetention - 1 */ 0, retentionEntries)
+
+    assert.deepEqual(
+      oldEntries.map(_ => _.$ref),
+      ['target-a', 'snapshot-b1'],
+      'the abandoned target VM must be collected; the current target VM must never be'
+    )
+  })
+})

CHANGELOG.unreleased.md

@@ -28,7 +28,7 @@
 - [Servers] fix a mishandling in the grace period before marking a pool disconnected, this will keep the ui in sync AND not redownload all the xapi object for a transient issue (PR [#10355](https://github.com/vatesfr/xen-orchestra/pull/10355))
 - [Rolling pool update/reboot] VMs are brought back to the host they were running on more reliably, and a VM that cannot be moved back no longer fails the whole operation (PR [#10295](https://github.com/vatesfr/xen-orchestra/pull/10295))
 - [XO server] Fix current_operations format on host and pool objects (PR [#10283](https://github.com/vatesfr/xen-orchestra/pull/10283))
-
+- [Backup/Replication] Fix incremental replication silently stopping after the first run when "Distribute replications across the storage repositories" is enabled, with the warning `the writer AggregatedIncrementalXapiWriter has failed the step writer.checkBaseVdis()` (PR [#10375](https://github.com/vatesfr/xen-orchestra/pull/10375))
 
 ### Packages to release