feat(xen-api/Record#resolve_): resolve props named as types

Introduced by be9b5332d9

.eslintrc.js

@@ -1,13 +1,5 @@
 module.exports = {
-  extends: [
-    'plugin:eslint-comments/recommended',
-
-    'standard',
-    'standard-jsx',
-    'prettier',
-    'prettier/standard',
-    'prettier/react',
-  ],
+  extends: ['plugin:eslint-comments/recommended', 'standard', 'standard-jsx', 'prettier'],
   globals: {
     __DEV__: true,
     $Dict: true,

.gitignore

@@ -4,6 +4,8 @@
 /lerna-debug.log
 /lerna-debug.log.*
 
+/@vates/*/dist/
+/@vates/*/node_modules/
 /@xen-orchestra/*/dist/
 /@xen-orchestra/*/node_modules/
 /packages/*/dist/

.travis.yml

@@ -1,6 +1,6 @@
 language: node_js
 node_js:
-  - 12
+  - 14
 
 # Use containers.
 # http://docs.travis-ci.com/user/workers/container-based-infrastructure/

@vates/coalesce-calls/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/compose/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/compose/README.md

@@ -50,7 +50,7 @@ Functions may also be passed in an array:
 const f = compose([add2, mul3])
 ```
 
-Options can be passed as first parameters:
+Options can be passed as first parameter:
 
 ```js
 const f = compose(

@vates/compose/USAGE.md

@@ -32,7 +32,7 @@ Functions may also be passed in an array:
 const f = compose([add2, mul3])
 ```
 
-Options can be passed as first parameters:
+Options can be passed as first parameter:
 
 ```js
 const f = compose(

@vates/decorate-with/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/disposable/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/disposable/deduped.js

@@ -17,15 +17,15 @@ exports.deduped = (factory, keyFn = (...args) => args) =>
       if (state === undefined) {
         const result = factory.apply(this, arguments)
 
-        const createFactory = ({ value, dispose }) => {
+        const createFactory = disposable => {
           const wrapper = {
             dispose() {
               if (--state.users === 0) {
                 states.delete(keys)
-                return dispose()
+                return disposable.dispose()
               }
             },
-            value,
+            value: disposable.value,
           }
 
           return () => {

@vates/disposable/package.json

@@ -14,7 +14,7 @@
     "url": "https://vates.fr"
   },
   "license": "ISC",
-  "version": "0.1.0",
+  "version": "0.1.1",
   "engines": {
     "node": ">=8.10"
   },

@vates/multi-key-map/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/parse-duration/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/read-chunk/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/read-chunk/index.js

@@ -1,27 +1,30 @@
-exports.readChunk = (stream, size) =>
-  new Promise((resolve, reject) => {
-    function onEnd() {
-      resolve(null)
-      removeListeners()
-    }
-    function onError(error) {
-      reject(error)
-      removeListeners()
-    }
-    function onReadable() {
-      const data = stream.read(size)
-      if (data !== null) {
-        resolve(data)
-        removeListeners()
-      }
-    }
-    function removeListeners() {
-      stream.removeListener('end', onEnd)
-      stream.removeListener('error', onError)
-      stream.removeListener('readable', onReadable)
-    }
-    stream.on('end', onEnd)
-    stream.on('error', onError)
-    stream.on('readable', onReadable)
-    onReadable()
-  })
+const readChunk = (stream, size) =>
+  size === 0
+    ? Promise.resolve(Buffer.alloc(0))
+    : new Promise((resolve, reject) => {
+        function onEnd() {
+          resolve(null)
+          removeListeners()
+        }
+        function onError(error) {
+          reject(error)
+          removeListeners()
+        }
+        function onReadable() {
+          const data = stream.read(size)
+          if (data !== null) {
+            resolve(data)
+            removeListeners()
+          }
+        }
+        function removeListeners() {
+          stream.removeListener('end', onEnd)
+          stream.removeListener('error', onError)
+          stream.removeListener('readable', onReadable)
+        }
+        stream.on('end', onEnd)
+        stream.on('error', onError)
+        stream.on('readable', onReadable)
+        onReadable()
+      })
+exports.readChunk = readChunk

@vates/read-chunk/index.spec.js

@@ -0,0 +1,43 @@
+/* eslint-env jest */
+
+const { Readable } = require('stream')
+
+const { readChunk } = require('./')
+
+const makeStream = it => Readable.from(it, { objectMode: false })
+makeStream.obj = Readable.from
+
+describe('readChunk', () => {
+  it('returns null if stream is empty', async () => {
+    expect(await readChunk(makeStream([]))).toBe(null)
+  })
+
+  describe('with binary stream', () => {
+    it('returns the first chunk of data', async () => {
+      expect(await readChunk(makeStream(['foo', 'bar']))).toEqual(Buffer.from('foo'))
+    })
+
+    it('returns a chunk of the specified size (smaller than first)', async () => {
+      expect(await readChunk(makeStream(['foo', 'bar']), 2)).toEqual(Buffer.from('fo'))
+    })
+
+    it('returns a chunk of the specified size (larger than first)', async () => {
+      expect(await readChunk(makeStream(['foo', 'bar']), 4)).toEqual(Buffer.from('foob'))
+    })
+
+    it('returns less data if stream ends', async () => {
+      expect(await readChunk(makeStream(['foo', 'bar']), 10)).toEqual(Buffer.from('foobar'))
+    })
+
+    it('returns an empty buffer if the specified size is 0', async () => {
+      expect(await readChunk(makeStream(['foo', 'bar']), 0)).toEqual(Buffer.alloc(0))
+    })
+  })
+
+  describe('with object stream', () => {
+    it('returns the first chunk of data verbatim', async () => {
+      const chunks = [{}, {}]
+      expect(await readChunk(makeStream.obj(chunks))).toBe(chunks[0])
+    })
+  })
+})

@vates/read-chunk/package.json

@@ -19,10 +19,13 @@
     "type": "git",
     "url": "https://github.com/vatesfr/xen-orchestra.git"
   },
-  "version": "0.1.1",
+  "version": "0.1.2",
   "engines": {
     "node": ">=8.10"
   },
+  "files": [
+    "index.js"
+  ],
   "scripts": {
     "postversion": "npm publish --access public"
   },

@vates/toggle-scripts/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/async-map/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/async-map/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/async-map/index.js

@@ -12,7 +12,7 @@ const wrapCall = (fn, arg, thisArg) => {
  * WARNING: Does not handle plain objects
  *
  * @template Item,This
- * @param {Iterable<Item>} arrayLike
+ * @param {Iterable<Item>} iterable
  * @param {(this: This, item: Item) => (Item | PromiseLike<Item>)} mapFn
  * @param {This} [thisArg]
  * @returns {Promise<Item[]>}

@xen-orchestra/audit-core/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/audit-core/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/audit-core/src/index.js

@@ -2,9 +2,9 @@
 import 'core-js/features/symbol/async-iterator'
 
 import assert from 'assert'
-import createLogger from '@xen-orchestra/log'
 import defer from 'golike-defer'
 import hash from 'object-hash'
+import { createLogger } from '@xen-orchestra/log'
 
 const log = createLogger('xo:audit-core')
 

@xen-orchestra/babel-config/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/backups-cli/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/backups-cli/commands/clean-vms.js

@@ -1,312 +1,16 @@
 #!/usr/bin/env node
 
-// assigned when options are parsed by the main function
-let merge, remove
-
 // -----------------------------------------------------------------------------
 
-const assert = require('assert')
 const asyncMap = require('lodash/curryRight')(require('@xen-orchestra/async-map').asyncMap)
-const flatten = require('lodash/flatten')
 const getopts = require('getopts')
-const limitConcurrency = require('limit-concurrency-decorator').default
-const lockfile = require('proper-lockfile')
-const pipe = require('promise-toolbox/pipe')
-const { default: Vhd, mergeVhd } = require('vhd-lib')
-const { dirname, resolve } = require('path')
-const { DISK_TYPE_DIFFERENCING } = require('vhd-lib/dist/_constants')
-const { isValidXva } = require('@xen-orchestra/backups/isValidXva')
+const { RemoteAdapter } = require('@xen-orchestra/backups/RemoteAdapter')
+const { resolve } = require('path')
 
-const fs = require('../_fs')
-
-const handler = require('@xen-orchestra/fs').getHandler({ url: 'file://' })
-
-// -----------------------------------------------------------------------------
-
-// chain is an array of VHDs from child to parent
-//
-// the whole chain will be merged into parent, parent will be renamed to child
-// and all the others will deleted
-const mergeVhdChain = limitConcurrency(1)(async function mergeVhdChain(chain) {
-  assert(chain.length >= 2)
-
-  let child = chain[0]
-  const parent = chain[chain.length - 1]
-  const children = chain.slice(0, -1).reverse()
-
-  console.warn('Unused parents of VHD', child)
-  chain
-    .slice(1)
-    .reverse()
-    .forEach(parent => {
-      console.warn('  ', parent)
-    })
-  merge && console.warn('  merging…')
-  console.warn('')
-  if (merge) {
-    // `mergeVhd` does not work with a stream, either
-    // - make it accept a stream
-    // - or create synthetic VHD which is not a stream
-    if (children.length !== 1) {
-      console.warn('TODO: implement merging multiple children')
-      children.length = 1
-      child = children[0]
-    }
-
-    let done, total
-    const handle = setInterval(() => {
-      if (done !== undefined) {
-        console.log('merging %s: %s/%s', child, done, total)
-      }
-    }, 10e3)
-
-    await mergeVhd(
-      handler,
-      parent,
-      handler,
-      child,
-      // children.length === 1
-      //   ? child
-      //   : await createSyntheticStream(handler, children),
-      {
-        onProgress({ done: d, total: t }) {
-          done = d
-          total = t
-        },
-      }
-    )
-
-    clearInterval(handle)
-  }
-
-  await Promise.all([
-    remove && fs.rename(parent, child),
-    asyncMap(children.slice(0, -1), child => {
-      console.warn('Unused VHD', child)
-      remove && console.warn('  deleting…')
-      console.warn('')
-      return remove && handler.unlink(child)
-    }),
-  ])
-})
-
-const listVhds = pipe([
-  vmDir => vmDir + '/vdis',
-  fs.readdir2,
-  asyncMap(fs.readdir2),
-  flatten,
-  asyncMap(fs.readdir2),
-  flatten,
-  _ => _.filter(_ => _.endsWith('.vhd')),
-])
-
-async function handleVm(vmDir) {
-  const vhds = new Set()
-  const vhdParents = { __proto__: null }
-  const vhdChildren = { __proto__: null }
-
-  // remove broken VHDs
-  await asyncMap(await listVhds(vmDir), async path => {
-    try {
-      const vhd = new Vhd(handler, path)
-      await vhd.readHeaderAndFooter()
-      vhds.add(path)
-      if (vhd.footer.diskType === DISK_TYPE_DIFFERENCING) {
-        const parent = resolve(dirname(path), vhd.header.parentUnicodeName)
-        vhdParents[path] = parent
-        if (parent in vhdChildren) {
-          const error = new Error('this script does not support multiple VHD children')
-          error.parent = parent
-          error.child1 = vhdChildren[parent]
-          error.child2 = path
-          throw error // should we throw?
-        }
-        vhdChildren[parent] = path
-      }
-    } catch (error) {
-      console.warn('Error while checking VHD', path)
-      console.warn('  ', error)
-      if (error != null && error.code === 'ERR_ASSERTION') {
-        remove && console.warn('  deleting…')
-        console.warn('')
-        remove && (await handler.unlink(path))
-      }
-    }
-  })
-
-  // remove VHDs with missing ancestors
-  {
-    const deletions = []
-
-    // return true if the VHD has been deleted or is missing
-    const deleteIfOrphan = vhd => {
-      const parent = vhdParents[vhd]
-      if (parent === undefined) {
-        return
-      }
-
-      // no longer needs to be checked
-      delete vhdParents[vhd]
-
-      deleteIfOrphan(parent)
-
-      if (!vhds.has(parent)) {
-        vhds.delete(vhd)
-
-        console.warn('Error while checking VHD', vhd)
-        console.warn('  missing parent', parent)
-        remove && console.warn('  deleting…')
-        console.warn('')
-        remove && deletions.push(handler.unlink(vhd))
-      }
-    }
-
-    // > A property that is deleted before it has been visited will not be
-    // > visited later.
-    // >
-    // > -- https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for...in#Deleted_added_or_modified_properties
-    for (const child in vhdParents) {
-      deleteIfOrphan(child)
-    }
-
-    await Promise.all(deletions)
-  }
-
-  const [jsons, xvas, xvaSums] = await fs
-    .readdir2(vmDir)
-    .then(entries => [
-      entries.filter(_ => _.endsWith('.json')),
-      new Set(entries.filter(_ => _.endsWith('.xva'))),
-      entries.filter(_ => _.endsWith('.xva.cheksum')),
-    ])
-
-  await asyncMap(xvas, async path => {
-    // check is not good enough to delete the file, the best we can do is report
-    // it
-    if (!(await isValidXva(path))) {
-      console.warn('Potential broken XVA', path)
-      console.warn('')
-    }
-  })
-
-  const unusedVhds = new Set(vhds)
-  const unusedXvas = new Set(xvas)
-
-  // compile the list of unused XVAs and VHDs, and remove backup metadata which
-  // reference a missing XVA/VHD
-  await asyncMap(jsons, async json => {
-    const metadata = JSON.parse(await fs.readFile(json))
-    const { mode } = metadata
-    if (mode === 'full') {
-      const linkedXva = resolve(vmDir, metadata.xva)
-
-      if (xvas.has(linkedXva)) {
-        unusedXvas.delete(linkedXva)
-      } else {
-        console.warn('Error while checking backup', json)
-        console.warn('  missing file', linkedXva)
-        remove && console.warn('  deleting…')
-        console.warn('')
-        remove && (await handler.unlink(json))
-      }
-    } else if (mode === 'delta') {
-      const linkedVhds = (() => {
-        const { vhds } = metadata
-        return Object.keys(vhds).map(key => resolve(vmDir, vhds[key]))
-      })()
-
-      // FIXME: find better approach by keeping as much of the backup as
-      // possible (existing disks) even if one disk is missing
-      if (linkedVhds.every(_ => vhds.has(_))) {
-        linkedVhds.forEach(_ => unusedVhds.delete(_))
-      } else {
-        console.warn('Error while checking backup', json)
-        const missingVhds = linkedVhds.filter(_ => !vhds.has(_))
-        console.warn('  %i/%i missing VHDs', missingVhds.length, linkedVhds.length)
-        missingVhds.forEach(vhd => {
-          console.warn('  ', vhd)
-        })
-        remove && console.warn('  deleting…')
-        console.warn('')
-        remove && (await handler.unlink(json))
-      }
-    }
-  })
-
-  // TODO: parallelize by vm/job/vdi
-  const unusedVhdsDeletion = []
-  {
-    // VHD chains (as list from child to ancestor) to merge indexed by last
-    // ancestor
-    const vhdChainsToMerge = { __proto__: null }
-
-    const toCheck = new Set(unusedVhds)
-
-    const getUsedChildChainOrDelete = vhd => {
-      if (vhd in vhdChainsToMerge) {
-        const chain = vhdChainsToMerge[vhd]
-        delete vhdChainsToMerge[vhd]
-        return chain
-      }
-
-      if (!unusedVhds.has(vhd)) {
-        return [vhd]
-      }
-
-      // no longer needs to be checked
-      toCheck.delete(vhd)
-
-      const child = vhdChildren[vhd]
-      if (child !== undefined) {
-        const chain = getUsedChildChainOrDelete(child)
-        if (chain !== undefined) {
-          chain.push(vhd)
-          return chain
-        }
-      }
-
-      console.warn('Unused VHD', vhd)
-      remove && console.warn('  deleting…')
-      console.warn('')
-      remove && unusedVhdsDeletion.push(handler.unlink(vhd))
-    }
-
-    toCheck.forEach(vhd => {
-      vhdChainsToMerge[vhd] = getUsedChildChainOrDelete(vhd)
-    })
-
-    Object.keys(vhdChainsToMerge).forEach(key => {
-      const chain = vhdChainsToMerge[key]
-      if (chain !== undefined) {
-        unusedVhdsDeletion.push(mergeVhdChain(chain))
-      }
-    })
-  }
-
-  await Promise.all([
-    unusedVhdsDeletion,
-    asyncMap(unusedXvas, path => {
-      console.warn('Unused XVA', path)
-      remove && console.warn('  deleting…')
-      console.warn('')
-      return remove && handler.unlink(path)
-    }),
-    asyncMap(xvaSums, path => {
-      // no need to handle checksums for XVAs deleted by the script, they will be handled by `unlink()`
-      if (!xvas.has(path.slice(0, -'.checksum'.length))) {
-        console.warn('Unused XVA checksum', path)
-        remove && console.warn('  deleting…')
-        console.warn('')
-        return remove && handler.unlink(path)
-      }
-    }),
-  ])
-}
-
-// -----------------------------------------------------------------------------
+const adapter = new RemoteAdapter(require('@xen-orchestra/fs').getHandler({ url: 'file://' }))
 
 module.exports = async function main(args) {
-  const opts = getopts(args, {
+  const { _, remove, merge } = getopts(args, {
     alias: {
       remove: 'r',
       merge: 'm',
@@ -318,19 +22,12 @@ module.exports = async function main(args) {
     },
   })
 
-  ;({ remove, merge } = opts)
-  await asyncMap(opts._, async vmDir => {
+  await asyncMap(_, async vmDir => {
     vmDir = resolve(vmDir)
-
-    // TODO: implement this in `xo-server`, not easy because not compatible with
-    // `@xen-orchestra/fs`.
-    const release = await lockfile.lock(vmDir)
     try {
-      await handleVm(vmDir)
+      await adapter.cleanVm(vmDir, { remove, merge, onLog: log => console.warn(log) })
     } catch (error) {
-      console.error('handleVm', vmDir, error)
-    } finally {
-      await release()
+      console.error('adapter.cleanVm', vmDir, error)
     }
   })
 }

@xen-orchestra/backups-cli/package.json

@@ -7,14 +7,12 @@
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "dependencies": {
     "@xen-orchestra/async-map": "^0.1.2",
-    "@xen-orchestra/backups": "^0.7.0",
-    "@xen-orchestra/fs": "^0.13.0",
+    "@xen-orchestra/backups": "^0.9.1",
+    "@xen-orchestra/fs": "^0.14.0",
     "filenamify": "^4.1.0",
     "getopts": "^2.2.5",
-    "limit-concurrency-decorator": "^0.4.0",
     "lodash": "^4.17.15",
-    "promise-toolbox": "^0.17.0",
-    "proper-lockfile": "^4.1.1",
+    "promise-toolbox": "^0.18.0",
     "vhd-lib": "^1.0.0"
   },
   "engines": {
@@ -34,7 +32,7 @@
   "scripts": {
     "postversion": "npm publish --access public"
   },
-  "version": "0.4.0",
+  "version": "0.5.0",
   "license": "AGPL-3.0-or-later",
   "author": {
     "name": "Vates SAS",

@xen-orchestra/backups/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/backups/Backup.js

@@ -2,7 +2,6 @@ const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
 const Disposable = require('promise-toolbox/Disposable')
 const ignoreErrors = require('promise-toolbox/ignoreErrors')
 const limitConcurrency = require('limit-concurrency-decorator').default
-const using = require('promise-toolbox/using')
 const { compileTemplate } = require('@xen-orchestra/template')
 
 const { extractIdsFromSimplePattern } = require('./_extractIdsFromSimplePattern')
@@ -87,7 +86,7 @@ exports.Backup = class Backup {
       throw new Error('no retentions corresponding to the metadata modes found')
     }
 
-    await using(
+    await Disposable.use(
       Disposable.all(
         poolIds.map(id =>
           this._getRecord('pool', id).catch(error => {
@@ -196,7 +195,7 @@ exports.Backup = class Backup {
       ...settings[schedule.id],
     }
 
-    await using(
+    await Disposable.use(
       Disposable.all(
         extractIdsFromSimplePattern(job.srs).map(id =>
           this._getRecord('SR', id).catch(error => {
@@ -242,7 +241,7 @@ exports.Backup = class Backup {
 
         const handleVm = vmUuid =>
           runTask({ name: 'backup VM', data: { type: 'VM', id: vmUuid } }, () =>
-            using(this._getRecord('VM', vmUuid), vm =>
+            Disposable.use(this._getRecord('VM', vmUuid), vm =>
               new VmBackup({
                 config,
                 getSnapshotNameLabel,

@xen-orchestra/backups/ImportVmBackup.js

@@ -5,8 +5,9 @@ const { importDeltaVm } = require('./_deltaVm')
 const { Task } = require('./Task')
 
 exports.ImportVmBackup = class ImportVmBackup {
-  constructor({ adapter, metadata, srUuid, xapi }) {
+  constructor({ adapter, metadata, srUuid, xapi, settings: { newMacAddresses } = {} }) {
     this._adapter = adapter
+    this._importDeltaVmSettings = { newMacAddresses }
     this._metadata = metadata
     this._srUuid = srUuid
     this._xapi = xapi
@@ -37,6 +38,7 @@ exports.ImportVmBackup = class ImportVmBackup {
         const vmRef = isFull
           ? await xapi.VM_import(backup, srRef)
           : await importDeltaVm(backup, await xapi.getRecord('SR', srRef), {
+              ...this._importDeltaVmSettings,
               detectBase: false,
             })
 
@@ -54,6 +56,6 @@ exports.ImportVmBackup = class ImportVmBackup {
           id: await xapi.getField('VM', vmRef, 'uuid'),
         }
       }
-    ).catch(() => {}) // errors are handled by logs
+    )
   }
 }

@xen-orchestra/backups/RemoteAdapter.js

@@ -4,7 +4,6 @@ const fromCallback = require('promise-toolbox/fromCallback')
 const fromEvent = require('promise-toolbox/fromEvent')
 const pDefer = require('promise-toolbox/defer')
 const pump = require('pump')
-const using = require('promise-toolbox/using')
 const { basename, dirname, join, normalize, resolve } = require('path')
 const { createLogger } = require('@xen-orchestra/log')
 const { createSyntheticStream, mergeVhd, default: Vhd } = require('vhd-lib')
@@ -14,7 +13,9 @@ const { readdir, stat } = require('fs-extra')
 const { ZipFile } = require('yazl')
 
 const { BACKUP_DIR } = require('./_getVmBackupDir')
+const { cleanVm } = require('./_cleanVm')
 const { getTmpDir } = require('./_getTmpDir')
+const { isMetadataFile, isVhdFile } = require('./_backupType')
 const { listPartitions, LVM_PARTITION_TYPE } = require('./_listPartitions')
 const { lvs, pvs } = require('./_lvm')
 
@@ -24,13 +25,10 @@ exports.DIR_XO_CONFIG_BACKUPS = DIR_XO_CONFIG_BACKUPS
 const DIR_XO_POOL_METADATA_BACKUPS = 'xo-pool-metadata-backups'
 exports.DIR_XO_POOL_METADATA_BACKUPS = DIR_XO_POOL_METADATA_BACKUPS
 
-const { warn } = createLogger('xo:proxy:backups:RemoteAdapter')
+const { warn } = createLogger('xo:backups:RemoteAdapter')
 
 const compareTimestamp = (a, b) => a.timestamp - b.timestamp
 
-const isMetadataFile = filename => filename.endsWith('.json')
-const isVhdFile = filename => filename.endsWith('.vhd')
-
 const noop = Function.prototype
 
 const resolveRelativeFromFile = (file, path) => resolve('/', dirname(file), path).slice(1)
@@ -67,8 +65,8 @@ const debounceResourceFactory = factory =>
     return this._debounceResource(factory.apply(this, arguments))
   }
 
-exports.RemoteAdapter = class RemoteAdapter {
-  constructor(handler, { debounceResource, dirMode }) {
+class RemoteAdapter {
+  constructor(handler, { debounceResource = res => res, dirMode } = {}) {
     this._debounceResource = debounceResource
     this._dirMode = dirMode
     this._handler = handler
@@ -204,7 +202,7 @@ exports.RemoteAdapter = class RemoteAdapter {
   }
 
   _listLvmLogicalVolumes(devicePath, partition, results = []) {
-    return using(this._getLvmPhysicalVolume(devicePath, partition), async path => {
+    return Disposable.use(this._getLvmPhysicalVolume(devicePath, partition), async path => {
       const lvs = await pvs(['lv_name', 'lv_path', 'lv_size', 'vg_name'], path)
       const partitionId = partition !== undefined ? partition.id : ''
       lvs.forEach((lv, i) => {
@@ -235,7 +233,7 @@ exports.RemoteAdapter = class RemoteAdapter {
 
   fetchPartitionFiles(diskId, partitionId, paths) {
     const { promise, reject, resolve } = pDefer()
-    using(
+    Disposable.use(
       async function* () {
         const files = yield this._usePartitionFiles(diskId, partitionId, paths)
         const zip = new ZipFile()
@@ -375,7 +373,7 @@ exports.RemoteAdapter = class RemoteAdapter {
   }
 
   listPartitionFiles(diskId, partitionId, path) {
-    return using(this.getPartition(diskId, partitionId), async rootPath => {
+    return Disposable.use(this.getPartition(diskId, partitionId), async rootPath => {
       path = resolveSubpath(rootPath, path)
 
       const entriesMap = {}
@@ -395,7 +393,7 @@ exports.RemoteAdapter = class RemoteAdapter {
   }
 
   listPartitions(diskId) {
-    return using(this.getDisk(diskId), async devicePath => {
+    return Disposable.use(this.getDisk(diskId), async devicePath => {
       const partitions = await listPartitions(devicePath)
 
       if (partitions.length === 0) {
@@ -552,3 +550,9 @@ exports.RemoteAdapter = class RemoteAdapter {
     return Object.defineProperty(JSON.parse(await this._handler.readFile(path)), '_filename', { value: path })
   }
 }
+
+RemoteAdapter.prototype.cleanVm = function (vmDir) {
+  return Disposable.use(this._handler.lock(vmDir), () => cleanVm.apply(this, arguments))
+}
+
+exports.RemoteAdapter = RemoteAdapter

@xen-orchestra/backups/RestoreMetadataBackup.js

@@ -15,7 +15,7 @@ exports.RestoreMetadataBackup = class RestoreMetadataBackup {
 
     if (backupId.split('/')[0] === DIR_XO_POOL_METADATA_BACKUPS) {
       return xapi.putResource(await handler.createReadStream(`${backupId}/data`), PATH_DB_DUMP, {
-        task: xapi.createTask('Import pool metadata'),
+        task: xapi.task_create('Import pool metadata'),
       })
     } else {
       return String(await handler.readFile(`${backupId}/data.json`))

@xen-orchestra/backups/Task.js

@@ -164,7 +164,14 @@ const Task = {
     }
   },
 
-  wrapFn({ name, data, onLog }, fn) {
+  wrapFn(opts, fn) {
+    // compatibility with @decorateWith
+    if (typeof fn !== 'function') {
+      ;[fn, opts] = [opts, fn]
+    }
+
+    const { name, data, onLog } = opts
+
     return function () {
       const evaluate = v => (typeof v === 'function' ? v.apply(this, arguments) : v)
       return Task.run({ name: evaluate(name), data: evaluate(data), onLog }, () => fn.apply(this, arguments))

@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,7 +23,7 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
           type: 'SR',
         }),
       },
-      this.run
+      this.transfer
     )
   }
 
@@ -51,26 +51,33 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
     }
   }
 
-  async run({ timestamp, deltaExport, sizeContainers }) {
-    const sr = this._sr
+  async prepare() {
     const settings = this._settings
+    const { uuid: srUuid, $xapi: xapi } = this._sr
+    const { scheduleId, vm } = this._backup
+
+    // delete previous interrupted copies
+    ignoreErrors.call(asyncMapSettled(listReplicatedVms(xapi, scheduleId, undefined, vm.uuid), vm => vm.$destroy))
+
+    this._oldEntries = getOldEntries(settings.copyRetention - 1, listReplicatedVms(xapi, scheduleId, srUuid, vm.uuid))
+
+    if (settings.deleteFirst) {
+      await this._deleteOldEntries()
+    } else {
+      this.cleanup = this._deleteOldEntries
+    }
+  }
+
+  async _deleteOldEntries() {
+    return asyncMapSettled(this._oldEntries, vm => vm.$destroy())
+  }
+
+  async transfer({ timestamp, deltaExport, sizeContainers }) {
+    const sr = this._sr
     const { job, scheduleId, vm } = this._backup
 
     const { uuid: srUuid, $xapi: xapi } = sr
 
-    // delete previous interrupted copies
-    ignoreErrors.call(
-      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 +115,5 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
         'xo:backup:vm': vm.uuid,
       }),
     ])
-
-    if (!deleteFirst) {
-      await deleteOldBackups()
-    }
   }
 }

@xen-orchestra/backups/_DeltaBackupWriter.js

@@ -14,7 +14,7 @@ const { getVmBackupDir } = require('./_getVmBackupDir')
 const { packUuid } = require('./_packUuid')
 const { Task } = require('./Task')
 
-const { warn } = createLogger('xo:proxy:backups:DeltaBackupWriter')
+const { warn } = createLogger('xo:backups:DeltaBackupWriter')
 
 exports.DeltaBackupWriter = class DeltaBackupWriter {
   constructor(backup, remoteId, settings) {
@@ -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._deleteOldEntries
   }
 
   async checkBaseVdis(baseUuidToSrcVdi) {
@@ -70,23 +72,16 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
     })
   }
 
-  async run({ timestamp, deltaExport, sizeContainers }) {
+  async prepare() {
     const adapter = this._adapter
-    const backup = this._backup
     const settings = this._settings
+    const { scheduleId, vm } = 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 oldBackups = getOldEntries(
+    const oldEntries = getOldEntries(
       settings.exportRetention - 1,
       await adapter.listVmBackups(vm.uuid, _ => _.mode === 'delta' && _.scheduleId === scheduleId)
     )
+    this._oldEntries = oldEntries
 
     // FIXME: implement optimized multiple VHDs merging with synthetic
     // delta
@@ -98,21 +93,44 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
     // The old backups will be eventually merged in future runs of the
     // job.
     const { maxMergedDeltasPerRun } = this._settings
-    if (oldBackups.length > maxMergedDeltasPerRun) {
-      oldBackups.length = maxMergedDeltasPerRun
+    if (oldEntries.length > maxMergedDeltasPerRun) {
+      oldEntries.length = maxMergedDeltasPerRun
     }
 
-    const deleteOldBackups = () =>
-      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; ) {
-          size += await adapter.deleteDeltaVmBackups([oldBackups[i]])
-        }
-        return {
-          size,
-        }
-      })
+    if (settings.deleteFirst) {
+      await this._deleteOldEntries()
+    } else {
+      this.cleanup = this._deleteOldEntries
+    }
+  }
+
+  async _deleteOldEntries() {
+    return Task.run({ name: 'merge' }, async () => {
+      const adapter = this._adapter
+      const oldEntries = this._oldEntries
+
+      let size = 0
+      // delete sequentially from newest to oldest to avoid unnecessary merges
+      for (let i = oldEntries.length; i-- > 0; ) {
+        size += await adapter.deleteDeltaVmBackups([oldEntries[i]])
+      }
+      return {
+        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 +160,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 +214,6 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
       dirMode: backup.config.dirMode,
     })
 
-    if (!deleteFirst) {
-      await deleteOldBackups()
-    }
-
     // TODO: run cleanup?
   }
 }

@xen-orchestra/backups/_PoolMetadataBackup.js

@@ -21,7 +21,7 @@ exports.PoolMetadataBackup = class PoolMetadataBackup {
   _exportPoolMetadata() {
     const xapi = this._pool.$xapi
     return xapi.getResource(PATH_DB_DUMP, {
-      task: xapi.createTask('Export pool metadata'),
+      task: xapi.task_create('Export pool metadata'),
     })
   }
 

@xen-orchestra/backups/_VmBackup.js

@@ -16,7 +16,7 @@ const { getOldEntries } = require('./_getOldEntries')
 const { Task } = require('./Task')
 const { watchStreamSize } = require('./_watchStreamSize')
 
-const { debug, warn } = createLogger('xo:proxy:backups:VmBackup')
+const { debug, warn } = createLogger('xo:backups:VmBackup')
 
 const forkDeltaExport = deltaExport =>
   Object.create(deltaExport, {
@@ -121,9 +121,9 @@ exports.VmBackup = class VmBackup {
           await vm.$assertHealthyVdiChains()
         }
 
-        const snapshotRef = await vm[settings.checkpointSnapshot ? '$checkpoint' : '$snapshot'](
-          this._getSnapshotNameLabel(vm)
-        )
+        const snapshotRef = await vm[settings.checkpointSnapshot ? '$checkpoint' : '$snapshot']({
+          name_label: this._getSnapshotNameLabel(vm),
+        })
         this.timestamp = Date.now()
 
         await xapi.setFieldEntries('VM', snapshotRef, 'other_config', {
@@ -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() {

@xen-orchestra/backups/_backupType.js

@@ -0,0 +1,4 @@
+exports.isMetadataFile = filename => filename.endsWith('.json')
+exports.isVhdFile = filename => filename.endsWith('.vhd')
+exports.isXvaFile = filename => filename.endsWith('.xva')
+exports.isXvaSumFile = filename => filename.endsWith('.xva.cheksum')

@xen-orchestra/backups/_backupWorker.js

@@ -0,0 +1,151 @@
+const Disposable = require('promise-toolbox/Disposable')
+const ignoreErrors = require('promise-toolbox/ignoreErrors')
+const { compose } = require('@vates/compose')
+const { createDebounceResource } = require('@vates/disposable/debounceResource')
+const { deduped } = require('@vates/disposable/deduped')
+const { getHandler } = require('@xen-orchestra/fs')
+const { parseDuration } = require('@vates/parse-duration')
+const { Xapi } = require('@xen-orchestra/xapi')
+
+const { Backup } = require('./Backup')
+const { RemoteAdapter } = require('./RemoteAdapter')
+const { Task } = require('./Task')
+
+class BackupWorker {
+  #config
+  #job
+  #recordToXapi
+  #remoteOptions
+  #remotes
+  #schedule
+  #xapiOptions
+  #xapis
+
+  constructor({ config, job, recordToXapi, remoteOptions, remotes, resourceCacheDelay, schedule, xapiOptions, xapis }) {
+    this.#config = config
+    this.#job = job
+    this.#recordToXapi = recordToXapi
+    this.#remoteOptions = remoteOptions
+    this.#remotes = remotes
+    this.#schedule = schedule
+    this.#xapiOptions = xapiOptions
+    this.#xapis = xapis
+
+    const debounceResource = createDebounceResource()
+    debounceResource.defaultDelay = parseDuration(resourceCacheDelay)
+    this.debounceResource = debounceResource
+  }
+
+  run() {
+    return new Backup({
+      config: this.#config,
+      getAdapter: remoteId => this.getAdapter(this.#remotes[remoteId]),
+      getConnectedRecord: Disposable.factory(async function* getConnectedRecord(type, uuid) {
+        const xapiId = this.#recordToXapi[uuid]
+        if (xapiId === undefined) {
+          throw new Error('no XAPI associated to ' + uuid)
+        }
+
+        const xapi = yield this.getXapi(this.#xapis[xapiId])
+        return xapi.getRecordByUuid(type, uuid)
+      }).bind(this),
+      job: this.#job,
+      schedule: this.#schedule,
+    }).run()
+  }
+
+  getAdapter = Disposable.factory(this.getAdapter)
+  getAdapter = deduped(this.getAdapter, remote => [remote.url])
+  getAdapter = compose(this.getAdapter, function (resource) {
+    return this.debounceResource(resource)
+  })
+  async *getAdapter(remote) {
+    const handler = getHandler(remote, this.#remoteOptions)
+    await handler.sync()
+    try {
+      yield new RemoteAdapter(handler, {
+        debounceResource: this.debounceResource,
+        dirMode: this.#config.dirMode,
+      })
+    } finally {
+      await handler.forget()
+    }
+  }
+
+  getXapi = Disposable.factory(this.getXapi)
+  getXapi = deduped(this.getXapi, ({ url }) => [url])
+  getXapi = compose(this.getXapi, function (resource) {
+    return this.debounceResource(resource)
+  })
+  async *getXapi({ credentials: { username: user, password }, ...opts }) {
+    const xapi = new Xapi({
+      ...this.#xapiOptions,
+      ...opts,
+      auth: {
+        user,
+        password,
+      },
+    })
+
+    await xapi.connect()
+    try {
+      await xapi.objectsFetched
+
+      yield xapi
+    } finally {
+      await xapi.disconnect()
+    }
+  }
+}
+
+// Received message:
+//
+// Message {
+//   action: 'run'
+//   data: object
+//   runWithLogs: boolean
+// }
+//
+// Sent message:
+//
+// Message {
+//   type: 'log' | 'result'
+//   data?: object
+//   status?: 'success' | 'failure'
+//   result?: any
+// }
+process.on('message', async message => {
+  if (message.action === 'run') {
+    const backupWorker = new BackupWorker(message.data)
+    try {
+      const result = message.runWithLogs
+        ? await Task.run(
+            {
+              name: 'backup run',
+              onLog: data =>
+                process.send({
+                  data,
+                  type: 'log',
+                }),
+            },
+            () => backupWorker.run()
+          )
+        : await backupWorker.run()
+
+      process.send({
+        type: 'result',
+        result,
+        status: 'success',
+      })
+    } catch (error) {
+      process.send({
+        type: 'result',
+        result: error,
+        status: 'failure',
+      })
+    } finally {
+      await ignoreErrors.call(backupWorker.debounceResource.flushAll())
+      process.disconnect()
+    }
+  }
+})

@xen-orchestra/backups/_cleanVm.js

@@ -0,0 +1,277 @@
+const assert = require('assert')
+const limitConcurrency = require('limit-concurrency-decorator').default
+const { asyncMap } = require('@xen-orchestra/async-map')
+const { default: Vhd, mergeVhd } = require('vhd-lib')
+const { dirname, resolve } = require('path')
+const { DISK_TYPE_DIFFERENCING } = require('vhd-lib/dist/_constants')
+const { isMetadataFile, isVhdFile, isXvaFile, isXvaSumFile } = require('./_backupType')
+const { isValidXva } = require('./isValidXva')
+
+// chain is an array of VHDs from child to parent
+//
+// the whole chain will be merged into parent, parent will be renamed to child
+// and all the others will deleted
+const mergeVhdChain = limitConcurrency(1)(async function mergeVhdChain(chain, { handler, onLog, remove, merge }) {
+  assert(chain.length >= 2)
+
+  let child = chain[0]
+  const parent = chain[chain.length - 1]
+  const children = chain.slice(0, -1).reverse()
+
+  chain
+    .slice(1)
+    .reverse()
+    .forEach(parent => {
+      onLog(`the parent ${parent} of the child ${child} is unused`)
+    })
+
+  if (merge) {
+    // `mergeVhd` does not work with a stream, either
+    // - make it accept a stream
+    // - or create synthetic VHD which is not a stream
+    if (children.length !== 1) {
+      // TODO: implement merging multiple children
+      children.length = 1
+      child = children[0]
+    }
+
+    let done, total
+    const handle = setInterval(() => {
+      if (done !== undefined) {
+        onLog(`merging ${child}: ${done}/${total}`)
+      }
+    }, 10e3)
+
+    await mergeVhd(
+      handler,
+      parent,
+      handler,
+      child,
+      // children.length === 1
+      //   ? child
+      //   : await createSyntheticStream(handler, children),
+      {
+        onProgress({ done: d, total: t }) {
+          done = d
+          total = t
+        },
+      }
+    )
+
+    clearInterval(handle)
+  }
+
+  await Promise.all([
+    remove && handler.rename(parent, child),
+    asyncMap(children.slice(0, -1), child => {
+      onLog(`the VHD ${child} is unused`)
+      return remove && handler.unlink(child)
+    }),
+  ])
+})
+
+const noop = Function.prototype
+
+exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop }) {
+  const handler = this._handler
+
+  const vhds = new Set()
+  const vhdParents = { __proto__: null }
+  const vhdChildren = { __proto__: null }
+
+  // remove broken VHDs
+  await asyncMap(
+    await handler.list(`${vmDir}/vdis`, {
+      filter: isVhdFile,
+      prependDir: true,
+    }),
+    async path => {
+      try {
+        const vhd = new Vhd(handler, path)
+        await vhd.readHeaderAndFooter()
+        vhds.add(path)
+        if (vhd.footer.diskType === DISK_TYPE_DIFFERENCING) {
+          const parent = resolve(dirname(path), vhd.header.parentUnicodeName)
+          vhdParents[path] = parent
+          if (parent in vhdChildren) {
+            const error = new Error('this script does not support multiple VHD children')
+            error.parent = parent
+            error.child1 = vhdChildren[parent]
+            error.child2 = path
+            throw error // should we throw?
+          }
+          vhdChildren[parent] = path
+        }
+      } catch (error) {
+        onLog(`error while checking the VHD with path ${path}`)
+        if (error?.code === 'ERR_ASSERTION' && remove) {
+          await handler.unlink(path)
+        }
+      }
+    }
+  )
+
+  // remove VHDs with missing ancestors
+  {
+    const deletions = []
+
+    // return true if the VHD has been deleted or is missing
+    const deleteIfOrphan = vhd => {
+      const parent = vhdParents[vhd]
+      if (parent === undefined) {
+        return
+      }
+
+      // no longer needs to be checked
+      delete vhdParents[vhd]
+
+      deleteIfOrphan(parent)
+
+      if (!vhds.has(parent)) {
+        vhds.delete(vhd)
+
+        onLog(`the parent ${parent} of the VHD ${vhd} is missing`)
+        if (remove) {
+          deletions.push(handler.unlink(vhd))
+        }
+      }
+    }
+
+    // > A property that is deleted before it has been visited will not be
+    // > visited later.
+    // >
+    // > -- https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for...in#Deleted_added_or_modified_properties
+    for (const child in vhdParents) {
+      deleteIfOrphan(child)
+    }
+
+    await Promise.all(deletions)
+  }
+
+  const jsons = []
+  const xvas = new Set()
+  const xvaSums = []
+  const entries = await handler.list(vmDir, {
+    prependDir: true,
+  })
+  entries.forEach(path => {
+    if (isMetadataFile(path)) {
+      jsons.push(path)
+    } else if (isXvaFile(path)) {
+      xvas.add(path)
+    } else if (isXvaSumFile(path)) {
+      xvaSums.push(path)
+    }
+  })
+
+  await asyncMap(xvas, async path => {
+    // check is not good enough to delete the file, the best we can do is report
+    // it
+    if (!(await isValidXva(path))) {
+      onLog(`the XVA with path ${path} is potentially broken`)
+    }
+  })
+
+  const unusedVhds = new Set(vhds)
+  const unusedXvas = new Set(xvas)
+
+  // compile the list of unused XVAs and VHDs, and remove backup metadata which
+  // reference a missing XVA/VHD
+  await asyncMap(jsons, async json => {
+    const metadata = JSON.parse(await handler.readFile(json))
+    const { mode } = metadata
+    if (mode === 'full') {
+      const linkedXva = resolve(vmDir, metadata.xva)
+
+      if (xvas.has(linkedXva)) {
+        unusedXvas.delete(linkedXva)
+      } else {
+        onLog(`the XVA linked to the metadata ${json} is missing`)
+        if (remove) {
+          await handler.unlink(json)
+        }
+      }
+    } else if (mode === 'delta') {
+      const linkedVhds = (() => {
+        const { vhds } = metadata
+        return Object.keys(vhds).map(key => resolve(vmDir, vhds[key]))
+      })()
+
+      // FIXME: find better approach by keeping as much of the backup as
+      // possible (existing disks) even if one disk is missing
+      if (linkedVhds.every(_ => vhds.has(_))) {
+        linkedVhds.forEach(_ => unusedVhds.delete(_))
+      } else {
+        onLog(`Some VHDs linked to the metadata ${json} are missing`)
+        if (remove) {
+          await handler.unlink(json)
+        }
+      }
+    }
+  })
+
+  // TODO: parallelize by vm/job/vdi
+  const unusedVhdsDeletion = []
+  {
+    // VHD chains (as list from child to ancestor) to merge indexed by last
+    // ancestor
+    const vhdChainsToMerge = { __proto__: null }
+
+    const toCheck = new Set(unusedVhds)
+
+    const getUsedChildChainOrDelete = vhd => {
+      if (vhd in vhdChainsToMerge) {
+        const chain = vhdChainsToMerge[vhd]
+        delete vhdChainsToMerge[vhd]
+        return chain
+      }
+
+      if (!unusedVhds.has(vhd)) {
+        return [vhd]
+      }
+
+      // no longer needs to be checked
+      toCheck.delete(vhd)
+
+      const child = vhdChildren[vhd]
+      if (child !== undefined) {
+        const chain = getUsedChildChainOrDelete(child)
+        if (chain !== undefined) {
+          chain.push(vhd)
+          return chain
+        }
+      }
+
+      onLog(`the VHD ${vhd} is unused`)
+      if (remove) {
+        unusedVhdsDeletion.push(handler.unlink(vhd))
+      }
+    }
+
+    toCheck.forEach(vhd => {
+      vhdChainsToMerge[vhd] = getUsedChildChainOrDelete(vhd)
+    })
+
+    Object.keys(vhdChainsToMerge).forEach(key => {
+      const chain = vhdChainsToMerge[key]
+      if (chain !== undefined) {
+        unusedVhdsDeletion.push(mergeVhdChain(chain, { onLog, remove, merge }))
+      }
+    })
+  }
+
+  await Promise.all([
+    unusedVhdsDeletion,
+    asyncMap(unusedXvas, path => {
+      onLog(`the XVA ${path} is unused`)
+      return remove && handler.unlink(path)
+    }),
+    asyncMap(xvaSums, path => {
+      // no need to handle checksums for XVAs deleted by the script, they will be handled by `unlink()`
+      if (!xvas.has(path.slice(0, -'.checksum'.length))) {
+        onLog(`the XVA checksum ${path} is unused`)
+        return remove && handler.unlink(path)
+      }
+    }),
+  ])
+}

@xen-orchestra/backups/_deltaVm.js

@@ -143,7 +143,7 @@ exports.importDeltaVm = defer(async function importDeltaVm(
   $defer,
   deltaVm,
   sr,
-  { cancelToken = CancelToken.none, detectBase = true, mapVdisSrs = {} } = {}
+  { cancelToken = CancelToken.none, detectBase = true, mapVdisSrs = {}, newMacAddresses = false } = {}
 ) {
   const { version } = deltaVm
   if (compareVersions(version, '1.0.0') < 0) {
@@ -213,6 +213,7 @@ exports.importDeltaVm = defer(async function importDeltaVm(
     },
     {
       bios_strings: vmRecord.bios_strings,
+      generateMacSeed: newMacAddresses,
       suspend_VDI: suspendVdi?.$ref,
     }
   )
@@ -325,11 +326,16 @@ exports.importDeltaVm = defer(async function importDeltaVm(
       }
 
       if (network) {
-        return xapi.VIF_create({
-          ...vif,
-          network: network.$ref,
-          VM: vmRef,
-        })
+        return xapi.VIF_create(
+          {
+            ...vif,
+            network: network.$ref,
+            VM: vmRef,
+          },
+          {
+            MAC: newMacAddresses ? undefined : vif.MAC,
+          }
+        )
       }
     }),
   ])

@xen-orchestra/backups/_listPartitions.js

@@ -3,7 +3,7 @@ const { createLogger } = require('@xen-orchestra/log')
 const { createParser } = require('parse-pairs')
 const { execFile } = require('child_process')
 
-const { debug } = createLogger('xo:proxy:api')
+const { debug } = createLogger('xo:backups:listPartitions')
 
 const IGNORED_PARTITION_TYPES = {
   // https://github.com/jhermsmeier/node-mbr/blob/master/lib/partition.js#L38

@xen-orchestra/backups/package.json

@@ -8,7 +8,7 @@
     "type": "git",
     "url": "https://github.com/vatesfr/xen-orchestra.git"
   },
-  "version": "0.7.0",
+  "version": "0.9.1",
   "engines": {
     "node": ">=14.5"
   },
@@ -16,9 +16,12 @@
     "postversion": "npm publish --access public"
   },
   "dependencies": {
-    "@vates/disposable": "^0.1.0",
+    "@vates/compose": "^2.0.0",
+    "@vates/disposable": "^0.1.1",
     "@vates/multi-key-map": "^0.1.0",
+    "@vates/parse-duration": "^0.1.0",
     "@xen-orchestra/async-map": "^0.1.2",
+    "@xen-orchestra/fs": "^0.14.0",
     "@xen-orchestra/log": "^0.2.0",
     "@xen-orchestra/template": "^0.1.0",
     "compare-versions": "^3.6.0",
@@ -31,12 +34,12 @@
     "lodash": "^4.17.20",
     "node-zone": "^0.4.0",
     "parse-pairs": "^1.1.0",
-    "promise-toolbox": "^0.17.0",
+    "promise-toolbox": "^0.18.0",
     "vhd-lib": "^1.0.0",
     "yazl": "^2.5.1"
   },
   "peerDependencies": {
-    "@xen-orchestra/xapi": "^0.4.4"
+    "@xen-orchestra/xapi": "^0.6.0"
   },
   "license": "AGPL-3.0-or-later",
   "author": {

@xen-orchestra/backups/runBackupWorker.js

@@ -0,0 +1,38 @@
+const path = require('path')
+const { createLogger } = require('@xen-orchestra/log')
+const { fork } = require('child_process')
+
+const { warn } = createLogger('xo:backups:backupWorker')
+
+const PATH = path.resolve(__dirname, '_backupWorker.js')
+
+exports.runBackupWorker = function runBackupWorker(params, onLog) {
+  return new Promise((resolve, reject) => {
+    const worker = fork(PATH)
+
+    worker.on('exit', code => reject(new Error(`worker exited with code ${code}`)))
+    worker.on('error', reject)
+
+    worker.on('message', message => {
+      try {
+        if (message.type === 'result') {
+          if (message.status === 'success') {
+            resolve(message.result)
+          } else {
+            reject(message.result)
+          }
+        } else if (message.type === 'log') {
+          onLog(message.data)
+        }
+      } catch (error) {
+        warn(error)
+      }
+    })
+
+    worker.send({
+      action: 'run',
+      data: params,
+      runWithLogs: onLog !== undefined,
+    })
+  })
+}

@xen-orchestra/cr-seed-cli/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/cr-seed-cli/package.json

@@ -18,7 +18,7 @@
   "preferGlobal": true,
   "dependencies": {
     "golike-defer": "^0.5.1",
-    "xen-api": "^0.30.0"
+    "xen-api": "^0.31.0"
   },
   "scripts": {
     "postversion": "npm publish"

@xen-orchestra/cron/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/cron/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/defined/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/defined/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/defined/README.md

@@ -4,6 +4,8 @@
 
 [![Package Version](https://badgen.net/npm/v/@xen-orchestra/defined)](https://npmjs.org/package/@xen-orchestra/defined) ![License](https://badgen.net/npm/license/@xen-orchestra/defined) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@xen-orchestra/defined)](https://bundlephobia.com/result?p=@xen-orchestra/defined) [![Node compatibility](https://badgen.net/npm/node/@xen-orchestra/defined)](https://npmjs.org/package/@xen-orchestra/defined)
 
+> Utilities to help handling (possibly) undefined values
+
 ## Install
 
 Installation of the [npm package](https://npmjs.org/package/@xen-orchestra/defined):

@xen-orchestra/defined/package.json

@@ -3,8 +3,7 @@
   "name": "@xen-orchestra/defined",
   "version": "0.0.0",
   "license": "ISC",
-  "description": "",
-  "keywords": [],
+  "description": "Utilities to help handling (possibly) undefined values",
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/defined",
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "repository": {
@@ -27,13 +26,10 @@
   "engines": {
     "node": ">=6"
   },
-  "dependencies": {},
   "devDependencies": {
     "@babel/cli": "^7.0.0",
     "@babel/core": "^7.0.0",
     "@babel/preset-env": "^7.0.0",
-    "@babel/preset-flow": "^7.0.0",
-    "babel-plugin-lodash": "^3.3.2",
     "cross-env": "^7.0.2",
     "rimraf": "^3.0.0"
   },

@xen-orchestra/defined/src/index.js

@@ -1,5 +1,3 @@
-// @flow
-
 // Usage:
 //
 // ```js
@@ -41,7 +39,7 @@ export default function defined() {
 // const getFriendName = _ => _.friends[0].name
 // const friendName = get(getFriendName, props.user)
 // ```
-export const get = (accessor: (input: ?any) => any, arg: ?any) => {
+export const get = (accessor, arg) => {
   try {
     return accessor(arg)
   } catch (error) {
@@ -60,4 +58,4 @@ export const get = (accessor: (input: ?any) => any, arg: ?any) => {
 //   _ => new ProxyAgent(_)
 // )
 // ```
-export const ifDef = (value: ?any, thenFn: (value: any) => any) => (value !== undefined ? thenFn(value) : value)
+export const ifDef = (value, thenFn) => (value !== undefined ? thenFn(value) : value)

@xen-orchestra/emit-async/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/emit-async/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/emit-async/README.md

@@ -4,6 +4,8 @@
 
 [![Package Version](https://badgen.net/npm/v/@xen-orchestra/emit-async)](https://npmjs.org/package/@xen-orchestra/emit-async) ![License](https://badgen.net/npm/license/@xen-orchestra/emit-async) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@xen-orchestra/emit-async)](https://bundlephobia.com/result?p=@xen-orchestra/emit-async) [![Node compatibility](https://badgen.net/npm/node/@xen-orchestra/emit-async)](https://npmjs.org/package/@xen-orchestra/emit-async)
 
+> Emit an event for async listeners to settle
+
 ## Install
 
 Installation of the [npm package](https://npmjs.org/package/@xen-orchestra/emit-async):

@xen-orchestra/emit-async/package.json

@@ -3,8 +3,7 @@
   "name": "@xen-orchestra/emit-async",
   "version": "0.0.0",
   "license": "ISC",
-  "description": "",
-  "keywords": [],
+  "description": "Emit an event for async listeners to settle",
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/emit-async",
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "repository": {
@@ -27,12 +26,10 @@
   "engines": {
     "node": ">=6"
   },
-  "dependencies": {},
   "devDependencies": {
     "@babel/cli": "^7.0.0",
     "@babel/core": "^7.0.0",
     "@babel/preset-env": "^7.0.0",
-    "babel-plugin-lodash": "^3.3.2",
     "cross-env": "^7.0.2",
     "rimraf": "^3.0.0"
   },

@xen-orchestra/fs/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/fs/package.json

@@ -1,10 +1,9 @@
 {
   "private": false,
   "name": "@xen-orchestra/fs",
-  "version": "0.13.0",
+  "version": "0.14.0",
   "license": "AGPL-3.0-or-later",
   "description": "The File System for Xen Orchestra backups.",
-  "keywords": [],
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/fs",
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "repository": {
@@ -32,7 +31,8 @@
     "get-stream": "^6.0.0",
     "limit-concurrency-decorator": "^0.4.0",
     "lodash": "^4.17.4",
-    "promise-toolbox": "^0.17.0",
+    "promise-toolbox": "^0.18.0",
+    "proper-lockfile": "^4.1.2",
     "readable-stream": "^3.0.6",
     "through2": "^4.0.2",
     "tmp": "^0.2.1",

@xen-orchestra/fs/src/abstract.js

@@ -18,6 +18,7 @@ import { createChecksumStream, validChecksumOfReadStream } from './checksum'
 const { dirname } = path.posix
 
 type Data = Buffer | Readable | string
+type Disposable<T> = {| dispose: () => void | Promise<void>, value?: T |}
 type FileDescriptor = {| fd: mixed, path: string |}
 type LaxReadable = Readable & Object
 type LaxWritable = Writable & Object
@@ -72,6 +73,7 @@ class PrefixWrapper {
 }
 
 export default class RemoteHandlerAbstract {
+  _highWaterMark: number
   _remote: Object
   _timeout: number
 
@@ -84,7 +86,7 @@ export default class RemoteHandlerAbstract {
         throw new Error('Incorrect remote type')
       }
     }
-    ;({ timeout: this._timeout = DEFAULT_TIMEOUT } = options)
+    ;({ highWaterMark: this._highWaterMark, timeout: this._timeout = DEFAULT_TIMEOUT } = options)
 
     const sharedLimit = limit(options.maxParallelOperations ?? DEFAULT_MAX_PARALLEL_OPERATIONS)
     this.closeFile = sharedLimit(this.closeFile)
@@ -163,24 +165,26 @@ export default class RemoteHandlerAbstract {
       file = normalizePath(file)
     }
     const path = typeof file === 'string' ? file : file.path
-    const streamP = timeout.call(this._createReadStream(file, options), this._timeout).then(stream => {
-      // detect early errors
-      let promise = fromEvent(stream, 'readable')
+    const streamP = timeout
+      .call(this._createReadStream(file, { ...options, highWaterMark: this._highWaterMark }), this._timeout)
+      .then(stream => {
+        // detect early errors
+        let promise = fromEvent(stream, 'readable')
 
-      // try to add the length prop if missing and not a range stream
-      if (stream.length === undefined && options.end === undefined && options.start === undefined) {
-        promise = Promise.all([
-          promise,
-          ignoreErrors.call(
-            this._getSize(file).then(size => {
-              stream.length = size
-            })
-          ),
-        ])
-      }
+        // try to add the length prop if missing and not a range stream
+        if (stream.length === undefined && options.end === undefined && options.start === undefined) {
+          promise = Promise.all([
+            promise,
+            ignoreErrors.call(
+              this._getSize(file).then(size => {
+                stream.length = size
+              })
+            ),
+          ])
+        }
 
-      return promise.then(() => stream)
-    })
+        return promise.then(() => stream)
+      })
 
     if (!checksum) {
       return streamP
@@ -213,7 +217,6 @@ export default class RemoteHandlerAbstract {
     input: Readable | Promise<Readable>,
     { checksum = true, dirMode }: { checksum?: boolean, dirMode?: number } = {}
   ): Promise<void> {
-    path = normalizePath(path)
     return this._outputStream(normalizePath(path), await input, {
       checksum,
       dirMode,
@@ -260,6 +263,11 @@ export default class RemoteHandlerAbstract {
     return entries
   }
 
+  async lock(path: string): Promise<Disposable> {
+    path = normalizePath(path)
+    return { dispose: await this._lock(path) }
+  }
+
   async mkdir(dir: string, { mode }: { mode?: number } = {}): Promise<void> {
     await this.__mkdir(normalizePath(dir), { mode })
   }
@@ -409,7 +417,7 @@ export default class RemoteHandlerAbstract {
 
   async _createOutputStream(file: File, { dirMode, ...options }: Object = {}): Promise<LaxWritable> {
     try {
-      return await this._createWriteStream(file, options)
+      return await this._createWriteStream(file, { ...options, highWaterMark: this._highWaterMark })
     } catch (error) {
       if (typeof file !== 'string' || error.code !== 'ENOENT') {
         throw error
@@ -424,6 +432,8 @@ export default class RemoteHandlerAbstract {
     throw new Error('Not implemented')
   }
 
+  // createWriteStream takes highWaterMark as option even if it's not documented.
+  // Source: https://stackoverflow.com/questions/55026306/how-to-set-writeable-highwatermark
   async _createWriteStream(file: File, options: Object): Promise<LaxWritable> {
     throw new Error('Not implemented')
   }
@@ -435,6 +445,10 @@ export default class RemoteHandlerAbstract {
     return {}
   }
 
+  async _lock(path: string): Promise<Function> {
+    return () => Promise.resolve()
+  }
+
   async _getSize(file: File): Promise<number> {
     throw new Error('Not implemented')
   }
@@ -501,7 +515,7 @@ export default class RemoteHandlerAbstract {
   }
 
   _readFile(file: string, options?: Object): Promise<Buffer> {
-    return this._createReadStream(file, options).then(getStream.buffer)
+    return this._createReadStream(file, { ...options, highWaterMark: this._highWaterMark }).then(getStream.buffer)
   }
 
   async _rename(oldPath: string, newPath: string) {

@xen-orchestra/fs/src/local.js

@@ -1,10 +1,22 @@
 import df from '@sindresorhus/df'
 import fs from 'fs-extra'
-import { fromEvent } from 'promise-toolbox'
+import lockfile from 'proper-lockfile'
+import { fromEvent, retry } from 'promise-toolbox'
 
 import RemoteHandlerAbstract from './abstract'
 
 export default class LocalHandler extends RemoteHandlerAbstract {
+  constructor(remote, opts = {}) {
+    super(remote)
+    this._retriesOnEagain = {
+      delay: 1e3,
+      retries: 9,
+      ...opts.retriesOnEagain,
+      when: {
+        code: 'EAGAIN',
+      },
+    }
+  }
   get type() {
     return 'file'
   }
@@ -71,6 +83,10 @@ export default class LocalHandler extends RemoteHandlerAbstract {
     return fs.readdir(this._getFilePath(dir))
   }
 
+  _lock(path) {
+    return lockfile.lock(path)
+  }
+
   _mkdir(dir, { mode }) {
     return fs.mkdir(this._getFilePath(dir), { mode })
   }
@@ -92,7 +108,8 @@ export default class LocalHandler extends RemoteHandlerAbstract {
   }
 
   async _readFile(file, options) {
-    return fs.readFile(this._getFilePath(file), options)
+    const filePath = this._getFilePath(file)
+    return await retry(() => fs.readFile(filePath, options), this._retriesOnEagain)
   }
 
   async _rename(oldPath, newPath) {
@@ -114,7 +131,8 @@ export default class LocalHandler extends RemoteHandlerAbstract {
   }
 
   async _unlink(file) {
-    return fs.unlink(this._getFilePath(file))
+    const filePath = this._getFilePath(file)
+    return await retry(() => fs.unlink(filePath), this._retriesOnEagain)
   }
 
   _writeFd(file, buffer, position) {

@xen-orchestra/log/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/log/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/log/README.md

@@ -4,6 +4,8 @@
 
 [![Package Version](https://badgen.net/npm/v/@xen-orchestra/log)](https://npmjs.org/package/@xen-orchestra/log) ![License](https://badgen.net/npm/license/@xen-orchestra/log) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@xen-orchestra/log)](https://bundlephobia.com/result?p=@xen-orchestra/log) [![Node compatibility](https://badgen.net/npm/node/@xen-orchestra/log)](https://npmjs.org/package/@xen-orchestra/log)
 
+> Logging system with decoupled producers/consumer
+
 ## Install
 
 Installation of the [npm package](https://npmjs.org/package/@xen-orchestra/log):

@xen-orchestra/log/package.json

@@ -3,8 +3,7 @@
   "name": "@xen-orchestra/log",
   "version": "0.2.0",
   "license": "ISC",
-  "description": "",
-  "keywords": [],
+  "description": "Logging system with decoupled producers/consumer",
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/log",
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "repository": {
@@ -31,7 +30,7 @@
   },
   "dependencies": {
     "lodash": "^4.17.4",
-    "promise-toolbox": "^0.17.0"
+    "promise-toolbox": "^0.18.0"
   },
   "devDependencies": {
     "@babel/cli": "^7.0.0",

@xen-orchestra/log/src/transports/console.js

@@ -57,15 +57,6 @@ if (process.stdout !== undefined && process.stdout.isTTY && process.stderr !== u
     for (let i = 0, n = namespace.length; i < n; ++i) {
       hash = ((hash << 5) - hash + namespace.charCodeAt(i)) | 0
     }
-    // // select a hue (HSV)
-    // const h = (Math.abs(hash) % 20) * 18
-    // // convert to RGB
-    // const f = (n, k = (n + h / 60) % 6) =>
-    //   Math.round(255 * (1 - Math.max(Math.min(k, 4 - k, 1), 0)))
-    // const r = f(5)
-    // const g = f(3)
-    // const b = f(1)
-    // return ansi(`38;2;${r};${g};${b}`, namespace)
     return ansi(`1;38;5;${NAMESPACE_COLORS[Math.abs(hash) % NAMESPACE_COLORS.length]}`, namespace)
   }
 } else {

@xen-orchestra/mixin/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/mixin/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/mixin/package.json

@@ -3,8 +3,6 @@
   "name": "@xen-orchestra/mixin",
   "version": "0.0.0",
   "license": "ISC",
-  "description": "",
-  "keywords": [],
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/mixin",
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "repository": {
@@ -35,7 +33,6 @@
     "@babel/core": "^7.0.0",
     "@babel/preset-env": "^7.0.0",
     "babel-plugin-dev": "^1.0.0",
-    "babel-plugin-lodash": "^3.3.2",
     "cross-env": "^7.0.2",
     "rimraf": "^3.0.0"
   },

@xen-orchestra/openflow/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/openflow/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/openflow/package.json

@@ -30,7 +30,7 @@
     "rimraf": "^3.0.0"
   },
   "dependencies": {
-    "@vates/read-chunk": "^0.1.0"
+    "@vates/read-chunk": "^0.1.2"
   },
   "author": {
     "name": "Vates SAS",

@xen-orchestra/proxy-cli/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/proxy-cli/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/proxy-cli/README.md

@@ -4,6 +4,8 @@
 
 [![Package Version](https://badgen.net/npm/v/@xen-orchestra/proxy-cli)](https://npmjs.org/package/@xen-orchestra/proxy-cli) ![License](https://badgen.net/npm/license/@xen-orchestra/proxy-cli) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@xen-orchestra/proxy-cli)](https://bundlephobia.com/result?p=@xen-orchestra/proxy-cli) [![Node compatibility](https://badgen.net/npm/node/@xen-orchestra/proxy-cli)](https://npmjs.org/package/@xen-orchestra/proxy-cli)
 
+> CLI for @xen-orchestra/proxy
+
 ## Install
 
 Installation of the [npm package](https://npmjs.org/package/@xen-orchestra/proxy-cli):

@xen-orchestra/proxy-cli/package.json

@@ -3,7 +3,7 @@
   "name": "@xen-orchestra/proxy-cli",
   "version": "0.2.0",
   "license": "AGPL-3.0-or-later",
-  "description": "",
+  "description": "CLI for @xen-orchestra/proxy",
   "keywords": [
     "backup",
     "proxy",
@@ -30,7 +30,7 @@
   },
   "dependencies": {
     "@iarna/toml": "^2.2.0",
-    "@vates/read-chunk": "^0.1.1",
+    "@vates/read-chunk": "^0.1.2",
     "ansi-colors": "^4.1.1",
     "app-conf": "^0.9.0",
     "content-type": "^1.0.4",
@@ -38,7 +38,7 @@
     "getopts": "^2.2.3",
     "http-request-plus": "^0.9.1",
     "json-rpc-protocol": "^0.13.1",
-    "promise-toolbox": "^0.15.1",
+    "promise-toolbox": "^0.18.1",
     "pump": "^3.0.0",
     "pumpify": "^2.0.1",
     "split2": "^3.1.1"
@@ -49,7 +49,6 @@
     "@babel/plugin-proposal-nullish-coalescing-operator": "^7.7.4",
     "@babel/plugin-proposal-optional-chaining": "^7.0.0",
     "@babel/preset-env": "^7.0.0",
-    "babel-plugin-lodash": "^3.3.2",
     "cross-env": "^7.0.2",
     "rimraf": "^3.0.0"
   },

@xen-orchestra/proxy/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/proxy/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/proxy/README.md

@@ -2,6 +2,8 @@
 
 # @xen-orchestra/proxy
 
+> XO Proxy used to remotely execute backup jobs
+
 ## Contributions
 
 Contributions are _very_ welcomed, either on the documentation or on

@xen-orchestra/proxy/docs/api.md

@@ -140,6 +140,7 @@ declare namespace backup {
   function listPoolMetadataBackups(_: {
     remotes: { [id: string]: Remote }
   }): { [remoteId: string]: { [poolUuid: string]: object[] } }
+
   function listVmBackups(_: {
     remotes: { [remoteId: string]: Remote }
   }): { [remoteId: string]: { [vmUuid: string]: object[] } }

@xen-orchestra/proxy/package.json

@@ -1,9 +1,9 @@
 {
   "private": true,
   "name": "@xen-orchestra/proxy",
-  "version": "0.11.5",
+  "version": "0.12.0",
   "license": "AGPL-3.0-or-later",
-  "description": "",
+  "description": "XO Proxy used to remotely execute backup jobs",
   "keywords": [
     "backup",
     "proxy",
@@ -32,41 +32,40 @@
   },
   "dependencies": {
     "@iarna/toml": "^2.2.0",
+    "@koa/router": "^10.0.0",
     "@vates/compose": "^2.0.0",
     "@vates/decorate-with": "^0.0.1",
-    "@vates/disposable": "^0.1.0",
+    "@vates/disposable": "^0.1.1",
     "@vates/parse-duration": "^0.1.0",
-    "@xen-orchestra/backups": "^0.7.0",
+    "@xen-orchestra/backups": "^0.9.1",
     "@xen-orchestra/emit-async": "^0.0.0",
-    "@xen-orchestra/fs": "^0.13.0",
+    "@xen-orchestra/fs": "^0.14.0",
     "@xen-orchestra/log": "^0.2.0",
     "@xen-orchestra/self-signed": "^0.1.0",
-    "@xen-orchestra/xapi": "^0.4.4",
-    "ajv": "^6.10.0",
+    "@xen-orchestra/xapi": "^0.6.0",
+    "ajv": "^8.0.3",
     "app-conf": "^0.9.0",
     "async-iterator-to-stream": "^1.1.0",
-    "compare-versions": "^3.4.0",
-    "fs-extra": "^8.1.0",
-    "get-stream": "^5.1.0",
+    "fs-extra": "^9.1.0",
+    "get-stream": "^6.0.0",
     "getopts": "^2.2.3",
     "golike-defer": "^0.5.1",
     "http-server-plus": "^0.11.0",
     "json-rpc-protocol": "^0.13.1",
     "jsonrpc-websocket-client": "^0.5.0",
     "koa": "^2.5.1",
-    "koa-compress": "^3.0.0",
+    "koa-compress": "^4.0.0",
     "koa-helmet": "^5.1.0",
-    "koa-router": "^7.4.0",
     "lodash": "^4.17.10",
     "ms": "^2.1.2",
     "node-zone": "^0.4.0",
     "parse-pairs": "^1.0.0",
-    "promise-toolbox": "^0.17.0",
+    "promise-toolbox": "^0.18.0",
     "source-map-support": "^0.5.16",
     "stoppable": "^1.0.6",
     "xdg-basedir": "^4.0.0",
-    "xen-api": "^0.30.0",
-    "xo-common": "^0.6.0"
+    "xen-api": "^0.31.0",
+    "xo-common": "^0.7.0"
   },
   "devDependencies": {
     "@babel/cli": "^7.0.0",

@xen-orchestra/proxy/src/app/mixins/api.js

@@ -8,7 +8,7 @@ import getStream from 'get-stream'
 import helmet from 'koa-helmet'
 import Koa from 'koa'
 import once from 'lodash/once'
-import Router from 'koa-router'
+import Router from '@koa/router'
 import Zone from 'node-zone'
 import { createLogger } from '@xen-orchestra/log'
 
@@ -19,7 +19,11 @@ const { debug, warn } = createLogger('xo:proxy:api')
 const ndJsonStream = asyncIteratorToStream(async function* (responseId, iterable) {
   yield format.response(responseId, { $responseType: 'ndjson' }) + '\n'
   for await (const data of iterable) {
-    yield JSON.stringify(data) + '\n'
+    try {
+      yield JSON.stringify(data) + '\n'
+    } catch (error) {
+      warn('ndJsonStream', { error })
+    }
   }
 })
 

@xen-orchestra/proxy/src/app/mixins/appliance.js

@@ -3,7 +3,6 @@ import fromCallback from 'promise-toolbox/fromCallback'
 import fromEvent from 'promise-toolbox/fromEvent'
 import JsonRpcWebsocketClient from 'jsonrpc-websocket-client'
 import parsePairs from 'parse-pairs'
-import using from 'promise-toolbox/using'
 import { createLogger } from '@xen-orchestra/log/dist'
 import { deduped } from '@vates/disposable/deduped'
 import { execFile, spawn } from 'child_process'
@@ -20,7 +19,7 @@ const getUpdater = deduped(async function () {
 })
 
 const callUpdate = params =>
-  using(
+  Disposable.use(
     getUpdater(),
     updater =>
       new Promise((resolve, reject) => {
@@ -144,7 +143,7 @@ export default class Appliance {
           ],
         },
         updater: {
-          getLocalManifest: () => using(getUpdater(), _ => _.call('getLocalManifest')),
+          getLocalManifest: () => Disposable.use(getUpdater(), _ => _.call('getLocalManifest')),
           getState: () => callUpdate(),
           upgrade: () => callUpdate({ upgrade: true }),
         },
@@ -154,6 +153,6 @@ export default class Appliance {
 
   // A proxy can be bound to a unique license
   getSelfLicense() {
-    return using(getUpdater(), _ => _.call('getSelfLicenses').then(licenses => licenses[0]))
+    return Disposable.use(getUpdater(), _ => _.call('getSelfLicenses').then(licenses => licenses[0]))
   }
 }

@xen-orchestra/proxy/src/app/mixins/backups/index.js

@@ -1,7 +1,6 @@
 import defer from 'golike-defer'
 import Disposable from 'promise-toolbox/Disposable'
 import fromCallback from 'promise-toolbox/fromCallback'
-import using from 'promise-toolbox/using'
 import { asyncMap } from '@xen-orchestra/async-map'
 import { Backup } from '@xen-orchestra/backups/Backup'
 import { compose } from '@vates/compose'
@@ -15,6 +14,7 @@ import { ImportVmBackup } from '@xen-orchestra/backups/ImportVmBackup'
 import { Readable } from 'stream'
 import { RemoteAdapter } from '@xen-orchestra/backups/RemoteAdapter'
 import { RestoreMetadataBackup } from '@xen-orchestra/backups/RestoreMetadataBackup'
+import { runBackupWorker } from '@xen-orchestra/backups/runBackupWorker'
 import { Task } from '@xen-orchestra/backups/Task'
 import { Xapi } from '@xen-orchestra/xapi'
 
@@ -46,26 +46,42 @@ export default class Backups {
       await fromCallback(execFile, 'pvscan', ['--cache'])
     })
 
-    let run = ({ recordToXapi, remotes, xapis, ...rest }) =>
-      new Backup({
-        ...rest,
+    let run = (params, onLog) => {
+      // don't change config during backup execution
+      const config = app.config.get('backups')
+      if (config.disableWorkers) {
+        const { recordToXapi, remotes, xapis, ...rest } = params
+        return new Backup({
+          ...rest,
 
-        // don't change config during backup execution
-        config: app.config.get('backups'),
+          config,
 
-        // pass getAdapter in order to mutualize the adapter resources usage
-        getAdapter: remoteId => this.getAdapter(remotes[remoteId]),
+          // pass getAdapter in order to mutualize the adapter resources usage
+          getAdapter: remoteId => this.getAdapter(remotes[remoteId]),
 
-        getConnectedRecord: Disposable.factory(async function* getConnectedRecord(type, uuid) {
-          const xapiId = recordToXapi[uuid]
-          if (xapiId === undefined) {
-            throw new Error('no XAPI associated to ' + uuid)
-          }
+          getConnectedRecord: Disposable.factory(async function* getConnectedRecord(type, uuid) {
+            const xapiId = recordToXapi[uuid]
+            if (xapiId === undefined) {
+              throw new Error('no XAPI associated to ' + uuid)
+            }
 
-          const xapi = yield this.getXapi(xapis[xapiId])
-          return xapi.getRecordByUuid(type, uuid)
-        }).bind(this),
-      }).run()
+            const xapi = yield this.getXapi(xapis[xapiId])
+            return xapi.getRecordByUuid(type, uuid)
+          }).bind(this),
+        }).run()
+      } else {
+        return runBackupWorker(
+          {
+            config,
+            remoteOptions: app.config.get('remoteOptions'),
+            resourceCacheDelay: app.config.getDuration('resourceCacheDelay'),
+            xapiOptions: app.config.get('xapiOptions'),
+            ...params,
+          },
+          onLog
+        )
+      }
+    }
 
     const runningJobs = { __proto__: null }
     run = (run => {
@@ -98,8 +114,8 @@ export default class Backups {
         return run.apply(this, arguments)
       })(run)
     run = (run => async (params, onLog) => {
-      if (onLog === undefined) {
-        return run(params)
+      if (onLog === undefined || !app.config.get('backups').disableWorkers) {
+        return run(params, onLog)
       }
 
       const { job, schedule } = params
@@ -126,7 +142,8 @@ export default class Backups {
     app.api.addMethods({
       backup: {
         deleteMetadataBackup: [
-          ({ backupId, remote }) => using(this.getAdapter(remote), adapter => adapter.deleteMetadataBackup(backupId)),
+          ({ backupId, remote }) =>
+            Disposable.use(this.getAdapter(remote), adapter => adapter.deleteMetadataBackup(backupId)),
           {
             description: 'delete Metadata backup',
             params: {
@@ -136,7 +153,8 @@ export default class Backups {
           },
         ],
         deleteVmBackup: [
-          ({ filename, remote }) => using(this.getAdapter(remote), adapter => adapter.deleteVmBackup(filename)),
+          ({ filename, remote }) =>
+            Disposable.use(this.getAdapter(remote), adapter => adapter.deleteVmBackup(filename)),
           {
             description: 'delete VM backup',
             params: {
@@ -147,7 +165,7 @@ export default class Backups {
         ],
         fetchPartitionFiles: [
           ({ disk: diskId, remote, partition: partitionId, paths }) =>
-            using(this.getAdapter(remote), adapter => adapter.fetchPartitionFiles(diskId, partitionId, paths)),
+            Disposable.use(this.getAdapter(remote), adapter => adapter.fetchPartitionFiles(diskId, partitionId, paths)),
           {
             description: 'fetch files from partition',
             params: {
@@ -159,10 +177,10 @@ export default class Backups {
           },
         ],
         importVmBackup: [
-          defer(($defer, { backupId, remote, srUuid, streamLogs = false, xapi: xapiOpts }) =>
-            using(this.getAdapter(remote), this.getXapi(xapiOpts), async (adapter, xapi) => {
+          defer(($defer, { backupId, remote, srUuid, settings, streamLogs = false, xapi: xapiOpts }) =>
+            Disposable.use(this.getAdapter(remote), this.getXapi(xapiOpts), async (adapter, xapi) => {
               const metadata = await adapter.readVmBackupMetadata(backupId)
-              const run = () => new ImportVmBackup({ adapter, metadata, srUuid, xapi }).run()
+              const run = () => new ImportVmBackup({ adapter, metadata, settings, srUuid, xapi }).run()
               return streamLogs
                 ? runWithLogs(
                     async (args, onLog) =>
@@ -187,6 +205,7 @@ export default class Backups {
             params: {
               backupId: { type: 'string' },
               remote: { type: 'object' },
+              settings: { type: 'object', optional: true },
               srUuid: { type: 'string' },
               streamLogs: { type: 'boolean', optional: true },
               xapi: { type: 'object' },
@@ -194,7 +213,8 @@ export default class Backups {
           },
         ],
         listDiskPartitions: [
-          ({ disk: diskId, remote }) => using(this.getAdapter(remote), adapter => adapter.listPartitions(diskId)),
+          ({ disk: diskId, remote }) =>
+            Disposable.use(this.getAdapter(remote), adapter => adapter.listPartitions(diskId)),
           {
             description: 'list disk partitions',
             params: {
@@ -205,7 +225,7 @@ export default class Backups {
         ],
         listPartitionFiles: [
           ({ disk: diskId, remote, partition: partitionId, path }) =>
-            using(this.getAdapter(remote), adapter => adapter.listPartitionFiles(diskId, partitionId, path)),
+            Disposable.use(this.getAdapter(remote), adapter => adapter.listPartitionFiles(diskId, partitionId, path)),
           {
             description: 'list partition files',
             params: {
@@ -221,7 +241,7 @@ export default class Backups {
             const backupsByRemote = {}
             await asyncMap(Object.entries(remotes), async ([remoteId, remote]) => {
               try {
-                await using(this.getAdapter(remote), async adapter => {
+                await Disposable.use(this.getAdapter(remote), async adapter => {
                   backupsByRemote[remoteId] = await adapter.listPoolMetadataBackups()
                 })
               } catch (error) {
@@ -245,7 +265,7 @@ export default class Backups {
             const backups = {}
             await asyncMap(Object.keys(remotes), async remoteId => {
               try {
-                await using(this.getAdapter(remotes[remoteId]), async adapter => {
+                await Disposable.use(this.getAdapter(remotes[remoteId]), async adapter => {
                   backups[remoteId] = formatVmBackups(await adapter.listAllVmBackups())
                 })
               } catch (error) {
@@ -275,7 +295,7 @@ export default class Backups {
             const backupsByRemote = {}
             await asyncMap(Object.entries(remotes), async ([remoteId, remote]) => {
               try {
-                await using(this.getAdapter(remote), async adapter => {
+                await Disposable.use(this.getAdapter(remote), async adapter => {
                   backupsByRemote[remoteId] = await adapter.listXoMetadataBackups()
                 })
               } catch (error) {
@@ -296,7 +316,7 @@ export default class Backups {
         ],
         restoreMetadataBackup: [
           ({ backupId, remote, xapi: xapiOptions }) =>
-            using(app.remotes.getHandler(remote), xapiOptions && this.getXapi(xapiOptions), (handler, xapi) =>
+            Disposable.use(app.remotes.getHandler(remote), xapiOptions && this.getXapi(xapiOptions), (handler, xapi) =>
               runWithLogs(
                 async (args, onLog) =>
                   Task.run(
@@ -347,7 +367,7 @@ export default class Backups {
       backup: {
         mountPartition: [
           async ({ disk, partition, remote }) =>
-            using(this.getAdapter(remote), adapter => durablePartition.mount(adapter, disk, partition)),
+            Disposable.use(this.getAdapter(remote), adapter => durablePartition.mount(adapter, disk, partition)),
           {
             description: 'mount a partition',
             params: {

@xen-orchestra/proxy/src/app/mixins/remotes.js

@@ -1,5 +1,4 @@
 import Disposable from 'promise-toolbox/Disposable'
-import using from 'promise-toolbox/using'
 import { compose } from '@vates/compose'
 import { decorateWith } from '@vates/decorate-with'
 import { deduped } from '@vates/disposable/deduped'
@@ -12,7 +11,7 @@ export default class Remotes {
     app.api.addMethods({
       remote: {
         getInfo: [
-          ({ remote }) => using(this.getHandler(remote), handler => handler.getInfo()),
+          ({ remote }) => Disposable.use(this.getHandler(remote), handler => handler.getInfo()),
           {
             params: {
               remote: { type: 'object' },
@@ -22,7 +21,7 @@ export default class Remotes {
 
         test: [
           ({ remote }) =>
-            using(this.getHandler(remote), handler => handler.test()).catch(error => ({
+            Disposable.use(this.getHandler(remote), handler => handler.test()).catch(error => ({
               success: false,
               error: error.message ?? String(error),
             })),

@xen-orchestra/proxy/src/index.js

@@ -43,10 +43,9 @@ ${name} v${version}
   })
 
   let httpServer = new (require('http-server-plus'))({
-    createSecureServer:
-      require('compare-versions')(process.version, '10.10.0') >= 0
-        ? (({ createSecureServer }) => opts => createSecureServer({ ...opts, allowHTTP1: true }))(require('http2'))
-        : undefined,
+    createSecureServer: (({ createSecureServer }) => opts => createSecureServer({ ...opts, allowHTTP1: true }))(
+      require('http2')
+    ),
   })
 
   const { readFileSync, outputFileSync, unlinkSync } = require('fs-extra')

@xen-orchestra/self-signed/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/template/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/upload-ova/.npmignore

@@ -1,24 +0,0 @@
-/benchmark/
-/benchmarks/
-*.bench.js
-*.bench.js.map
-
-/examples/
-example.js
-example.js.map
-*.example.js
-*.example.js.map
-
-/fixture/
-/fixtures/
-*.fixture.js
-*.fixture.js.map
-*.fixtures.js
-*.fixtures.js.map
-
-/test/
-/tests/
-*.spec.js
-*.spec.js.map
-
-__snapshots__/

@xen-orchestra/upload-ova/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/upload-ova/package.json

@@ -35,7 +35,7 @@
   "dependencies": {
     "chalk": "^4.1.0",
     "exec-promise": "^0.7.0",
-    "form-data": "^3.0.0",
+    "form-data": "^4.0.0",
     "fs-extra": "^9.0.0",
     "fs-promise": "^2.0.3",
     "get-stream": "^6.0.0",

@xen-orchestra/xapi/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@xen-orchestra/xapi/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@xen-orchestra/xapi",
-  "version": "0.4.4",
+  "version": "0.6.0",
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/xapi",
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "repository": {
@@ -12,6 +12,9 @@
   "files": [
     "dist/"
   ],
+  "bin": {
+    "xo-xapi": "./dist/cli.js"
+  },
   "engines": {
     "node": ">=8.10"
   },
@@ -21,10 +24,11 @@
     "@babel/plugin-proposal-decorators": "^7.3.0",
     "@babel/preset-env": "^7.3.1",
     "cross-env": "^7.0.2",
-    "rimraf": "^3.0.0"
+    "rimraf": "^3.0.0",
+    "xo-common": "^0.7.0"
   },
   "peerDependencies": {
-    "xen-api": "^0.30.0"
+    "xen-api": "^0.31.0"
   },
   "scripts": {
     "build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
@@ -39,11 +43,11 @@
   "dependencies": {
     "@xen-orchestra/async-map": "^0.1.2",
     "@xen-orchestra/log": "^0.2.0",
-    "d3-time-format": "^2.2.3",
+    "d3-time-format": "^3.0.0",
     "golike-defer": "^0.5.1",
     "lodash": "^4.17.15",
     "make-error": "^1.3.5",
-    "promise-toolbox": "^0.17.0"
+    "promise-toolbox": "^0.18.0"
   },
   "private": false,
   "license": "AGPL-3.0-or-later",

@xen-orchestra/xapi/src/cli.js

@@ -0,0 +1,6 @@
+#!/usr/bin/env node
+
+const { Xapi } = require('./')
+require('xen-api/dist/cli')
+  .default(opts => new Xapi({ ignoreNobakVdis: true, ...opts }))
+  .catch(console.error.bind(console, 'FATAL'))

@xen-orchestra/xapi/src/index.js

@@ -1,8 +1,11 @@
 const assert = require('assert')
 const defer = require('promise-toolbox/defer')
+const pRetry = require('promise-toolbox/retry')
 const { utcFormat, utcParse } = require('d3-time-format')
 const { Xapi: Base } = require('xen-api')
 
+exports.isDefaultTemplate = require('./isDefaultTemplate')
+
 // VDI formats. (Raw is not available for delta vdi.)
 exports.VDI_FORMAT_RAW = 'raw'
 exports.VDI_FORMAT_VHD = 'vhd'
@@ -32,16 +35,28 @@ const hasProps = o => {
 }
 
 class Xapi extends Base {
-  constructor({ ignoreNobakVdis, maxUncoalescedVdis, vdiDestroyRetryWhenInUse, ...opts }) {
+  constructor({
+    callRetryWhenTooManyPendingTasks,
+    ignoreNobakVdis,
+    maxUncoalescedVdis,
+    vdiDestroyRetryWhenInUse,
+    ...opts
+  }) {
     assert.notStrictEqual(ignoreNobakVdis, undefined)
 
     super(opts)
+    this._callRetryWhenTooManyPendingTasks = {
+      delay: 5e3,
+      tries: 10,
+      ...callRetryWhenTooManyPendingTasks,
+      when: { code: 'TOO_MANY_PENDING_TASKS' },
+    }
     this._ignoreNobakVdis = ignoreNobakVdis
     this._maxUncoalescedVdis = maxUncoalescedVdis
     this._vdiDestroyRetryWhenInUse = {
-      ...vdiDestroyRetryWhenInUse,
       delay: 5e3,
       retries: 10,
+      ...vdiDestroyRetryWhenInUse,
     }
 
     const genericWatchers = (this._genericWatchers = new Set())
@@ -122,3 +137,17 @@ mixin({
   VM: require('./vm'),
 })
 exports.Xapi = Xapi
+
+// TODO: remove once using next promise-toolbox
+function pRetryWrap(fn, options) {
+  const getOptions = typeof options !== 'function' ? () => options : options
+  return function () {
+    return pRetry(() => fn.apply(this, arguments), getOptions.apply(this, arguments))
+  }
+}
+
+function getCallRetryOpts() {
+  return this._callRetryWhenTooManyPendingTasks
+}
+Xapi.prototype.call = pRetryWrap(Xapi.prototype.call, getCallRetryOpts)
+Xapi.prototype.callAsync = pRetryWrap(Xapi.prototype.callAsync, getCallRetryOpts)

@xen-orchestra/xapi/src/isDefaultTemplate.js

@@ -0,0 +1 @@
+module.exports = vmTpl => vmTpl.is_default_template || vmTpl.other_config.default_template === 'true'

@xen-orchestra/xapi/src/vdi.js

@@ -1,19 +1,27 @@
 const CancelToken = require('promise-toolbox/CancelToken')
+const pCatch = require('promise-toolbox/catch')
 const pRetry = require('promise-toolbox/retry')
 
 const extractOpaqueRef = require('./_extractOpaqueRef')
 
+const noop = Function.prototype
+
 module.exports = class Vdi {
   async clone(vdiRef) {
     return extractOpaqueRef(await this.callAsync('VDI.clone', vdiRef))
   }
 
   async destroy(vdiRef) {
-    // work around a race condition in XCP-ng/XenServer where the disk is not fully unmounted yet
-    await pRetry(() => this.callAsync('VDI.destroy', vdiRef), {
-      ...this._vdiDestroyRetry,
-      when: { code: 'VDI_IN_USE' },
-    })
+    await pCatch.call(
+      // work around a race condition in XCP-ng/XenServer where the disk is not fully unmounted yet
+      pRetry(() => this.callAsync('VDI.destroy', vdiRef), {
+        ...this._vdiDestroyRetry,
+        when: { code: 'VDI_IN_USE' },
+      }),
+      // if this VDI is not found, consider it destroyed
+      { code: 'HANDLE_INVALID' },
+      noop
+    )
   }
 
   async create(

@xen-orchestra/xapi/src/vif.js

@@ -1,20 +1,26 @@
 const isVmRunning = require('./_isVmRunning')
 
 module.exports = class Vif {
-  async create({
-    currently_attached = true,
-    device,
-    ipv4_allowed,
-    ipv6_allowed,
-    locking_mode,
-    MAC,
-    MTU,
-    network,
-    other_config = {},
-    qos_algorithm_params = {},
-    qos_algorithm_type = '',
-    VM,
-  }) {
+  async create(
+    {
+      currently_attached = true,
+      device,
+      ipv4_allowed,
+      ipv6_allowed,
+      locking_mode,
+      MTU,
+      network,
+      other_config = {},
+      qos_algorithm_params = {},
+      qos_algorithm_type = '',
+      VM,
+    },
+    {
+      // duplicated MAC addresses can lead to issues,
+      // therefore it should be passed explicitely
+      MAC = '',
+    } = {}
+  ) {
     const [powerState, ...rest] = await Promise.all([
       this.getField('VM', VM, 'power_state'),
       device ?? (await this.call('VM.get_allowed_VIF_devices', VM))[0],

@xen-orchestra/xapi/src/vm.js

@@ -1,15 +1,18 @@
-const cancelable = require('promise-toolbox/cancelable')
+const CancelToken = require('promise-toolbox/CancelToken')
 const defer = require('golike-defer').default
 const groupBy = require('lodash/groupBy')
 const pickBy = require('lodash/pickBy')
 const ignoreErrors = require('promise-toolbox/ignoreErrors')
+const omit = require('lodash/omit')
 const pCatch = require('promise-toolbox/catch')
 const pRetry = require('promise-toolbox/retry')
 const { asyncMap } = require('@xen-orchestra/async-map')
 const { createLogger } = require('@xen-orchestra/log')
+const { incorrectState } = require('xo-common/api-errors')
 const { Ref } = require('xen-api')
 
 const extractOpaqueRef = require('./_extractOpaqueRef')
+const isDefaultTemplate = require('./isDefaultTemplate')
 const isVmRunning = require('./_isVmRunning')
 
 const { warn } = createLogger('xo:xapi:vm')
@@ -122,16 +125,15 @@ module.exports = class Vm {
     }
   }
 
-  @cancelable
-  async checkpoint($cancelToken, vmRef, nameLabel) {
-    if (nameLabel === undefined) {
-      nameLabel = await this.getField('VM', vmRef, 'name_label')
+  async checkpoint(vmRef, { cancelToken = CancelToken.none, name_label } = {}) {
+    if (name_label === undefined) {
+      name_label = await this.getField('VM', vmRef, 'name_label')
     }
     try {
-      return await this.callAsync($cancelToken, 'VM.checkpoint', vmRef, nameLabel).then(extractOpaqueRef)
+      return await this.callAsync(cancelToken, 'VM.checkpoint', vmRef, name_label).then(extractOpaqueRef)
     } catch (error) {
       if (error.code === 'VM_BAD_POWER_STATE') {
-        return this.VM_snapshot($cancelToken, vmRef, nameLabel)
+        return this.VM_snapshot(vmRef, { cancelToken, name_label })
       }
       throw error
     }
@@ -195,6 +197,17 @@ module.exports = class Vm {
       // not supported by `VM.create`, therefore it should be passed explicitly
       bios_strings,
 
+      // The field `other_config.mac_seed` is used (in conjunction with VIFs'
+      // devices) to generate MAC addresses of VIFs for this VM.
+      //
+      // It's automatically generated by VM.create if missing.
+      //
+      // If this is true, it will be filtered out by this method to ensure a
+      // new one is generated.
+      //
+      // See https://github.com/xapi-project/xen-api/blob/0a6d6de0704ca2cc439326c35af7cf45128a17d5/ocaml/xapi/xapi_vm.ml#L628
+      generateMacSeed = true,
+
       // if set, will create the VM in Suspended power_state with this VDI
       //
       // it's a separate param because it's not supported for all versions of
@@ -214,7 +227,7 @@ module.exports = class Vm {
       memory_dynamic_min,
       memory_static_max,
       memory_static_min,
-      other_config,
+      other_config: generateMacSeed ? omit(other_config, 'mac_seed') : other_config,
       PCI_bus,
       platform,
       PV_args,
@@ -261,21 +274,30 @@ module.exports = class Vm {
 
     bios_strings = cleanBiosStrings(bios_strings)
     if (bios_strings !== undefined) {
-      await this.call('VM.set_bios_strings', ref, bios_strings)
+      // Only available on XS >= 7.3
+      await pCatch.call(this.call('VM.set_bios_strings', ref, bios_strings), { code: 'MESSAGE_METHOD_UNKNOWN' }, noop)
     }
 
     return ref
   }
 
-  async destroy(vmRef, { deleteDisks = true, force = false, forceDeleteDefaultTemplate = false } = {}) {
+  async destroy(
+    vmRef,
+    { deleteDisks = true, force = false, bypassBlockedOperation = force, forceDeleteDefaultTemplate = force } = {}
+  ) {
     const vm = await this.getRecord('VM', vmRef)
 
-    if (!force && 'destroy' in vm.blocked_operations) {
+    if (!bypassBlockedOperation && 'destroy' in vm.blocked_operations) {
       throw new Error('destroy is blocked')
     }
 
-    if (!forceDeleteDefaultTemplate && vm.other_config.default_template === 'true') {
-      throw new Error('VM is default template')
+    if (!forceDeleteDefaultTemplate && isDefaultTemplate(vm)) {
+      throw incorrectState({
+        actual: true,
+        expected: false,
+        object: vm.$id,
+        property: 'isDefaultTemplate',
+      })
     }
 
     // It is necessary for suspended VMs to be shut down
@@ -285,9 +307,12 @@ module.exports = class Vm {
     }
 
     await Promise.all([
+      forceDeleteDefaultTemplate &&
+        // Only available on XS >= 7.2
+        pCatch.call(vm.set_is_default_template(false), { code: 'MESSAGE_METHOD_UNKNOWN' }, noop),
+      forceDeleteDefaultTemplate && vm.update_other_config('default_template', null),
       vm.set_is_a_template(false),
-      vm.update_blocked_operations('destroy', null),
-      vm.update_other_config('default_template', null),
+      bypassBlockedOperation && vm.update_blocked_operations('destroy', null),
     ])
 
     // must be done before destroying the VM
@@ -316,26 +341,36 @@ module.exports = class Vm {
     ])
   }
 
-  @cancelable
   @defer
-  async export($defer, $cancelToken, vmRef, { compress = false, useSnapshot } = {}) {
+  async export($defer, vmRef, { cancelToken = CancelToken.none, compress = false, useSnapshot } = {}) {
     const vm = await this.getRecord('VM', vmRef)
     const taskRef = await this.task_create('VM export', vm.name_label)
     $defer.onFailure.call(this, 'task_destroy', taskRef)
     if (useSnapshot === undefined) {
       useSnapshot = isVmRunning(vm)
     }
-    const exportedVmRef = useSnapshot
-      ? await this.VM_snapshot($cancelToken, vmRef, `[XO Export] ${vm.name_label}`)
-      : vmRef
+    let exportedVmRef, destroySnapshot
+    if (useSnapshot) {
+      exportedVmRef = await this.VM_snapshot(vmRef, { cancelToken, name_label: `[XO Export] ${vm.name_label}` })
+      destroySnapshot = () => ignoreErrors.call(this.VM_destroy(exportedVmRef))
+      $defer.onFailure(destroySnapshot)
+    } else {
+      exportedVmRef = vmRef
+    }
     try {
-      return await this.getResource($cancelToken, '/export/', {
+      const stream = await this.getResource(cancelToken, '/export/', {
         query: {
           ref: exportedVmRef,
           use_compression: compress === 'zstd' ? 'zstd' : compress === true || compress === 'gzip' ? 'true' : 'false',
         },
         task: taskRef,
       })
+
+      if (useSnapshot) {
+        stream.once('end', destroySnapshot).once('error', destroySnapshot)
+      }
+
+      return stream
     } catch (error) {
       // augment the error with as much relevant info as possible
       const [poolMaster, exportedVm] = await Promise.all([
@@ -345,14 +380,7 @@ module.exports = class Vm {
       error.pool_master = poolMaster
       error.VM = exportedVm
       throw error
-    } finally {
     }
-    // if (useSnapshot) {
-    //   const destroySnapshot = () => this.deleteVm(exportedVm)::ignoreErrors()
-    //   promise.then(_ => _.task::pFinally(destroySnapshot), destroySnapshot)
-    // }
-    //
-    // return promise
   }
 
   async getDisks(vmRef, vbdRefs) {
@@ -400,8 +428,7 @@ module.exports = class Vm {
   }
 
   @defer
-  @cancelable
-  async snapshot($cancelToken, $defer, vmRef, nameLabel) {
+  async snapshot($defer, vmRef, { cancelToken = CancelToken.none, name_label } = {}) {
     const vm = await this.getRecord('VM', vmRef)
     // cannot unplug VBDs on Running, Paused and Suspended VMs
     if (vm.power_state === 'Halted' && this._ignoreNobakVdis) {
@@ -418,31 +445,32 @@ module.exports = class Vm {
       })
     }
 
-    if (nameLabel === undefined) {
-      nameLabel = vm.name_label
+    if (name_label === undefined) {
+      name_label = vm.name_label
     }
     let ref
     do {
       if (!vm.tags.includes('xo-disable-quiesce')) {
         try {
+          let { snapshots } = vm
           ref = await pRetry(
-            async bail => {
+            async () => {
               try {
-                return await this.callAsync($cancelToken, 'VM.snapshot_with_quiesce', vmRef, nameLabel)
+                return await this.callAsync(cancelToken, 'VM.snapshot_with_quiesce', vmRef, name_label)
               } catch (error) {
                 if (error == null || error.code !== 'VM_SNAPSHOT_WITH_QUIESCE_FAILED') {
-                  throw bail(error)
+                  throw pRetry.bail(error)
                 }
                 // detect and remove new broken snapshots
                 //
                 // see https://github.com/vatesfr/xen-orchestra/issues/3936
-                const prevSnapshotRefs = new Set(vm.snapshots)
+                const prevSnapshotRefs = new Set(snapshots)
                 const snapshotNameLabelPrefix = `Snapshot of ${vm.uuid} [`
-                await vm.refresh_snapshots()
+                snapshots = await this.getField('VM', vm.$ref, 'snapshots')
                 const createdSnapshots = (
                   await this.getRecords(
                     'VM',
-                    vm.snapshots.filter(_ => !prevSnapshotRefs.has(_))
+                    snapshots.filter(_ => !prevSnapshotRefs.has(_))
                   )
                 ).filter(_ => _.name_label.startsWith(snapshotNameLabelPrefix))
                 // be safe: only delete if there was a single match
@@ -475,7 +503,7 @@ module.exports = class Vm {
           }
         }
       }
-      ref = await this.callAsync($cancelToken, 'VM.snapshot', vmRef, nameLabel).then(extractOpaqueRef)
+      ref = await this.callAsync(cancelToken, 'VM.snapshot', vmRef, name_label).then(extractOpaqueRef)
     } while (false)
 
     // VM snapshots are marked as templates, unfortunately it does not play well with XVA export/import