feat(backups/delta writers): split run method in prepare/transfer/cleanup

Fixes xoa-support#3523

This avoids starting the transfer before the writers are ready, which caused it to failed with `deleteFirst` when deletion was so long that the transfer stalled.

@xen-orchestra/backups/_ContinuousReplicationWriter.js

@@ -14,7 +14,7 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
     this._settings = settings
     this._sr = sr
 
-    this.run = Task.wrapFn(
+    this.transfer = Task.wrapFn(
       {
         name: 'export',
         data: ({ deltaExport }) => ({
@@ -23,8 +23,10 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
           type: 'SR',
         }),
       },
-      this.run
+      this.transfer
     )
+
+    this[settings.deleteFirst ? 'prepare' : 'cleanup'] = this._deleteOld
   }
 
   async checkBaseVdis(baseUuidToSrcVdi, baseVm) {
@@ -51,9 +53,17 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
     }
   }
 
-  async run({ timestamp, deltaExport, sizeContainers }) {
+  async _deleteOld() {
+    const { uuid: srUuid, $xapi: xapi } = this._sr
+    const { scheduleId, vm } = this._backup
+
+    const oldVms = getOldEntries(this._settings.copyRetention - 1, listReplicatedVms(xapi, scheduleId, srUuid, vm.uuid))
+
+    return asyncMapSettled(oldVms, vm => xapi.VM_destroy(vm.$ref))
+  }
+
+  async transfer({ timestamp, deltaExport, sizeContainers }) {
     const sr = this._sr
-    const settings = this._settings
     const { job, scheduleId, vm } = this._backup
 
     const { uuid: srUuid, $xapi: xapi } = sr
@@ -63,14 +73,6 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
       asyncMapSettled(listReplicatedVms(xapi, scheduleId, undefined, vm.uuid), vm => xapi.VM_destroy(vm.$ref))
     )
 
-    const oldVms = getOldEntries(settings.copyRetention - 1, listReplicatedVms(xapi, scheduleId, srUuid, vm.uuid))
-
-    const deleteOldBackups = () => asyncMapSettled(oldVms, vm => xapi.VM_destroy(vm.$ref))
-    const { deleteFirst } = settings
-    if (deleteFirst) {
-      await deleteOldBackups()
-    }
-
     let targetVmRef
     await Task.run({ name: 'transfer' }, async () => {
       targetVmRef = await importDeltaVm(
@@ -108,9 +110,5 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
         'xo:backup:vm': vm.uuid,
       }),
     ])
-
-    if (!deleteFirst) {
-      await deleteOldBackups()
-    }
   }
 }

@xen-orchestra/backups/_DeltaBackupWriter.js

@@ -22,7 +22,7 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
     this._backup = backup
     this._settings = settings
 
-    this.run = Task.wrapFn(
+    this.transfer = Task.wrapFn(
       {
         name: 'export',
         data: ({ deltaExport }) => ({
@@ -31,8 +31,10 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
           type: 'remote',
         }),
       },
-      this.run
+      this.transfer
     )
+
+    this[settings.deleteFirst ? 'prepare' : 'cleanup'] = this._deleteOld
   }
 
   async checkBaseVdis(baseUuidToSrcVdi) {
@@ -70,21 +72,12 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
     })
   }
 
-  async run({ timestamp, deltaExport, sizeContainers }) {
+  async _deleteOld() {
     const adapter = this._adapter
-    const backup = this._backup
-    const settings = this._settings
-
-    const { job, scheduleId, vm } = backup
-
-    const jobId = job.id
-    const handler = adapter.handler
-    const backupDir = getVmBackupDir(vm.uuid)
-
-    // TODO: clean VM backup directory
+    const { scheduleId, vm } = this._backup
 
     const oldBackups = getOldEntries(
-      settings.exportRetention - 1,
+      this._settings.exportRetention - 1,
       await adapter.listVmBackups(vm.uuid, _ => _.mode === 'delta' && _.scheduleId === scheduleId)
     )
 
@@ -102,8 +95,7 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
       oldBackups.length = maxMergedDeltasPerRun
     }
 
-    const deleteOldBackups = () =>
-      Task.run({ name: 'merge' }, async () => {
+    return Task.run({ name: 'merge' }, async () => {
       let size = 0
       // delete sequentially from newest to oldest to avoid unnecessary merges
       for (let i = oldBackups.length; i-- > 0; ) {
@@ -113,6 +105,19 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
         size,
       }
     })
+  }
+
+  async transfer({ timestamp, deltaExport, sizeContainers }) {
+    const adapter = this._adapter
+    const backup = this._backup
+
+    const { job, scheduleId, vm } = backup
+
+    const jobId = job.id
+    const handler = adapter.handler
+    const backupDir = getVmBackupDir(vm.uuid)
+
+    // TODO: clean VM backup directory
 
     const basename = formatFilenameDate(timestamp)
     const vhds = mapValues(
@@ -142,11 +147,6 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
       vmSnapshot: this._backup.exportedVm,
     }
 
-    const { deleteFirst } = settings
-    if (deleteFirst) {
-      await deleteOldBackups()
-    }
-
     const { size } = await Task.run({ name: 'transfer' }, async () => {
       await Promise.all(
         map(deltaExport.vdis, async (vdi, id) => {
@@ -201,10 +201,6 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
       dirMode: backup.config.dirMode,
     })
 
-    if (!deleteFirst) {
-      await deleteOldBackups()
-    }
-
     // TODO: run cleanup?
   }
 }

@xen-orchestra/backups/_VmBackup.js

@@ -147,6 +147,8 @@ exports.VmBackup = class VmBackup {
     const { exportedVm } = this
     const baseVm = this._baseVm
 
+    await asyncMap(this._writers, writer => writer.prepare && writer.prepare())
+
     const deltaExport = await exportDeltaVm(exportedVm, baseVm, {
       fullVdisRequired: this._fullVdisRequired,
     })
@@ -156,7 +158,7 @@ exports.VmBackup = class VmBackup {
 
     await asyncMap(this._writers, async writer => {
       try {
-        await writer.run({
+        await writer.transfer({
           deltaExport: forkDeltaExport(deltaExport),
           sizeContainers,
           timestamp,
@@ -192,6 +194,8 @@ exports.VmBackup = class VmBackup {
       speed: duration !== 0 ? (size * 1e3) / 1024 / 1024 / duration : 0,
       size,
     })
+
+    await asyncMap(this._writers, writer => writer.cleanup && writer.cleanup())
   }
 
   async _copyFull() {