WiP

Fixes #5633

.eslintrc.js

@@ -13,19 +13,13 @@ module.exports = {
 
   overrides: [
     {
-      files: ['cli.js', '*-cli.js', '**/*cli*/**/*.js'],
+      files: ['cli.{,c,m}js', '*-cli.{,c,m}js', '**/*cli*/**/*.{,c,m}js'],
       rules: {
         'no-console': 'off',
       },
     },
   ],
 
-  parser: 'babel-eslint',
-  parserOptions: {
-    ecmaFeatures: {
-      legacyDecorators: true,
-    },
-  },
   rules: {
     // disabled because XAPI objects are using camel case
     camelcase: ['off'],

.gitignore

@@ -11,7 +11,7 @@
 /packages/*/dist/
 /packages/*/node_modules/
 
-/@xen-orchestra/proxy/src/app/mixins/index.js
+/@xen-orchestra/proxy/src/app/mixins/index.mjs
 
 /packages/vhd-cli/src/commands/index.js
 
@@ -19,9 +19,9 @@
 /packages/xen-api/plot.dat
 
 /packages/xo-server/.xo-server.*
-/packages/xo-server/src/api/index.js
-/packages/xo-server/src/xapi/mixins/index.js
-/packages/xo-server/src/xo-mixins/index.js
+/packages/xo-server/src/api/index.mjs
+/packages/xo-server/src/xapi/mixins/index.mjs
+/packages/xo-server/src/xo-mixins/index.mjs
 
 /packages/xo-server-auth-ldap/ldap.cache.conf
 

@vates/coalesce-calls/package.json

@@ -20,9 +20,6 @@
     "type": "git",
     "url": "https://github.com/vatesfr/xen-orchestra.git"
   },
-  "files": [
-    "index.js"
-  ],
   "author": {
     "name": "Vates SAS",
     "url": "https://vates.fr"

@vates/disposable/README.md

@@ -48,7 +48,7 @@ import { createDebounceResource } from '@vates/disposable/debounceResource'
 const debounceResource = createDebounceResource()
 
 // it will wait for 10 seconds before calling the disposer
-using(debounceResource(getConnection(host), 10e3), connection => {})
+Disposable.use(debounceResource(getConnection(host), 10e3), connection => {})
 ```
 
 ### `debounceResource.flushAll()`

@vates/disposable/USAGE.md

@@ -30,7 +30,7 @@ import { createDebounceResource } from '@vates/disposable/debounceResource'
 const debounceResource = createDebounceResource()
 
 // it will wait for 10 seconds before calling the disposer
-using(debounceResource(getConnection(host), 10e3), connection => {})
+Disposable.use(debounceResource(getConnection(host), 10e3), connection => {})
 ```
 
 ### `debounceResource.flushAll()`

@vates/disposable/package.json

@@ -23,6 +23,7 @@
   },
   "dependencies": {
     "@vates/multi-key-map": "^0.1.0",
+    "@xen-orchestra/async-map": "^0.1.2",
     "@xen-orchestra/log": "^0.2.0",
     "ensure-array": "^1.0.0"
   }

@vates/parse-duration/README.md

@@ -44,4 +44,4 @@ You may:
 
 ## License
 
-[AGPL-3.0-or-later](https://spdx.org/licenses/AGPL-3.0-or-later) © [Vates SAS](https://vates.fr)
+[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

@vates/parse-duration/index.js

@@ -6,7 +6,7 @@ exports.parseDuration = value => {
   }
   const duration = ms(value)
   if (duration === undefined) {
-    throw new TypeError(`not a valid duration: ${duration}`)
+    throw new TypeError(`not a valid duration: ${value}`)
   }
   return duration
 }

@vates/parse-duration/package.json

@@ -18,8 +18,8 @@
     "name": "Vates SAS",
     "url": "https://vates.fr"
   },
-  "license": "AGPL-3.0-or-later",
-  "version": "0.1.0",
+  "license": "ISC",
+  "version": "0.1.1",
   "engines": {
     "node": ">=8.10"
   },

@vates/read-chunk/package.json

@@ -23,9 +23,6 @@
   "engines": {
     "node": ">=8.10"
   },
-  "files": [
-    "index.js"
-  ],
   "scripts": {
     "postversion": "npm publish --access public"
   },

@vates/toggle-scripts/package.json

@@ -31,9 +31,6 @@
   "engines": {
     "node": ">=6"
   },
-  "files": [
-    "index.js"
-  ],
   "bin": "./index.js",
   "scripts": {
     "postversion": "npm publish --access public"

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

@@ -18,6 +18,17 @@ const wrapCall = (fn, arg, thisArg) => {
  * @returns {Promise<Item[]>}
  */
 exports.asyncMap = function asyncMap(iterable, mapFn, thisArg = iterable) {
+  let onError
+  if (onError !== undefined) {
+    const original = mapFn
+    mapFn = async function () {
+      try {
+        return await original.apply(this, arguments)
+      } catch (error) {
+        return onError.call(this, error, ...arguments)
+      }
+    }
+  }
   return Promise.all(Array.from(iterable, mapFn, thisArg))
 }
 

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

@@ -24,10 +24,6 @@
     "url": "https://vates.fr"
   },
   "preferGlobal": false,
-  "files": [
-    "index.js",
-    "legacy.js"
-  ],
   "engines": {
     "node": ">=6"
   },

@xen-orchestra/audit-core/.eslintrc.js

@@ -0,0 +1 @@
+../../scripts/babel-eslintrc.js

@xen-orchestra/audit-core/package.json

@@ -26,14 +26,14 @@
     "@babel/plugin-proposal-decorators": "^7.8.0",
     "@babel/plugin-proposal-nullish-coalescing-operator": "^7.8.0",
     "@babel/preset-env": "^7.7.4",
-    "cross": "^1.0.0",
+    "cross-env": "^7.0.2",
     "rimraf": "^3.0.0"
   },
   "dependencies": {
+    "@vates/decorate-with": "^0.0.1",
     "@xen-orchestra/log": "^0.2.0",
     "core-js": "^3.6.4",
     "golike-defer": "^0.5.1",
-    "lodash": "^4.17.15",
     "object-hash": "^2.0.1"
   },
   "private": false,

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

@@ -2,9 +2,10 @@
 import 'core-js/features/symbol/async-iterator'
 
 import assert from 'assert'
-import defer from 'golike-defer'
 import hash from 'object-hash'
 import { createLogger } from '@xen-orchestra/log'
+import { decorateWith } from '@vates/decorate-with'
+import { defer } from 'golike-defer'
 
 const log = createLogger('xo:audit-core')
 
@@ -65,7 +66,7 @@ export class AuditCore {
     this._storage = storage
   }
 
-  @defer
+  @decorateWith(defer)
   async add($defer, subject, event, data) {
     const time = Date.now()
     $defer(await this._storage.acquireLock())
@@ -150,7 +151,7 @@ export class AuditCore {
     }
   }
 
-  @defer
+  @decorateWith(defer)
   async deleteRangeAndRewrite($defer, newest, oldest) {
     assert.notStrictEqual(newest, undefined)
     assert.notStrictEqual(oldest, undefined)

@xen-orchestra/babel-config/index.js

@@ -14,25 +14,13 @@ const configs = {
   '@babel/plugin-proposal-pipeline-operator': {
     proposal: 'minimal',
   },
-  '@babel/preset-env'(pkg) {
-    return {
-      debug: !__TEST__,
+  '@babel/preset-env': {
+    debug: !__TEST__,
 
-      // disabled until https://github.com/babel/babel/issues/8323 is resolved
-      // loose: true,
+    // disabled until https://github.com/babel/babel/issues/8323 is resolved
+    // loose: true,
 
-      shippedProposals: true,
-      targets: (() => {
-        let node = (pkg.engines || {}).node
-        if (node !== undefined) {
-          const trimChars = '^=>~'
-          while (trimChars.includes(node[0])) {
-            node = node.slice(1)
-          }
-        }
-        return { browsers: pkg.browserslist, node }
-      })(),
-    }
+    shippedProposals: true,
   },
 }
 
@@ -44,21 +32,21 @@ const getConfig = (key, ...args) => {
 // some plugins must be used in a specific order
 const pluginsOrder = ['@babel/plugin-proposal-decorators', '@babel/plugin-proposal-class-properties']
 
-module.exports = function (pkg, plugins, presets) {
-  plugins === undefined && (plugins = {})
-  presets === undefined && (presets = {})
+module.exports = function (pkg, configs = {}) {
+  const plugins = {}
+  const presets = {}
 
   Object.keys(pkg.devDependencies || {}).forEach(name => {
     if (!(name in presets) && PLUGINS_RE.test(name)) {
-      plugins[name] = getConfig(name, pkg)
+      plugins[name] = { ...getConfig(name, pkg), ...configs[name] }
     } else if (!(name in presets) && PRESETS_RE.test(name)) {
-      presets[name] = getConfig(name, pkg)
+      presets[name] = { ...getConfig(name, pkg), ...configs[name] }
     }
   })
 
   return {
     comments: !__PROD__,
-    ignore: __TEST__ ? undefined : [/\.spec\.js$/],
+    ignore: __PROD__ ? [/\.spec\.js$/] : undefined,
     plugins: Object.keys(plugins)
       .map(plugin => [plugin, plugins[plugin]])
       .sort(([a], [b]) => {
@@ -67,5 +55,15 @@ module.exports = function (pkg, plugins, presets) {
         return oA !== -1 && oB !== -1 ? oA - oB : a < b ? -1 : 1
       }),
     presets: Object.keys(presets).map(preset => [preset, presets[preset]]),
+    targets: (() => {
+      let node = (pkg.engines || {}).node
+      if (node !== undefined) {
+        const trimChars = '^=>~'
+        while (trimChars.includes(node[0])) {
+          node = node.slice(1)
+        }
+      }
+      return { browsers: pkg.browserslist, node }
+    })(),
   }
 }

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

@@ -7,21 +7,16 @@
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "dependencies": {
     "@xen-orchestra/async-map": "^0.1.2",
-    "@xen-orchestra/backups": "^0.9.1",
-    "@xen-orchestra/fs": "^0.14.0",
+    "@xen-orchestra/backups": "^0.11.0",
+    "@xen-orchestra/fs": "^0.17.0",
     "filenamify": "^4.1.0",
     "getopts": "^2.2.5",
     "lodash": "^4.17.15",
-    "promise-toolbox": "^0.18.0",
-    "vhd-lib": "^1.0.0"
+    "promise-toolbox": "^0.19.2"
   },
   "engines": {
     "node": ">=7.10.1"
   },
-  "files": [
-    "commands",
-    "*.js"
-  ],
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/backups-cli",
   "name": "@xen-orchestra/backups-cli",
   "repository": {
@@ -32,7 +27,7 @@
   "scripts": {
     "postversion": "npm publish --access public"
   },
-  "version": "0.5.0",
+  "version": "0.6.0",
   "license": "AGPL-3.0-or-later",
   "author": {
     "name": "Vates SAS",

@xen-orchestra/backups/Backup.js

@@ -1,14 +1,14 @@
 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 Disposable = require('promise-toolbox/Disposable.js')
+const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
 const { compileTemplate } = require('@xen-orchestra/template')
+const { limitConcurrency } = require('limit-concurrency-decorator')
 
-const { extractIdsFromSimplePattern } = require('./_extractIdsFromSimplePattern')
-const { PoolMetadataBackup } = require('./_PoolMetadataBackup')
-const { Task } = require('./Task')
-const { VmBackup } = require('./_VmBackup')
-const { XoMetadataBackup } = require('./_XoMetadataBackup')
+const { extractIdsFromSimplePattern } = require('./_extractIdsFromSimplePattern.js')
+const { PoolMetadataBackup } = require('./_PoolMetadataBackup.js')
+const { Task } = require('./Task.js')
+const { VmBackup } = require('./_VmBackup.js')
+const { XoMetadataBackup } = require('./_XoMetadataBackup.js')
 
 const noop = Function.prototype
 

@xen-orchestra/backups/ImportVmBackup.js

@@ -1,8 +1,9 @@
 const assert = require('assert')
 
-const { formatFilenameDate } = require('./_filenameDate')
-const { importDeltaVm } = require('./_deltaVm')
-const { Task } = require('./Task')
+const { formatFilenameDate } = require('./_filenameDate.js')
+const { importDeltaVm } = require('./_deltaVm.js')
+const { Task } = require('./Task.js')
+const { watchStreamSize } = require('./_watchStreamSize.js')
 
 exports.ImportVmBackup = class ImportVmBackup {
   constructor({ adapter, metadata, srUuid, xapi, settings: { newMacAddresses } = {} }) {
@@ -18,13 +19,17 @@ exports.ImportVmBackup = class ImportVmBackup {
     const metadata = this._metadata
     const isFull = metadata.mode === 'full'
 
+    const sizeContainer = { size: 0 }
+
     let backup
     if (isFull) {
       backup = await adapter.readFullVmBackup(metadata)
+      watchStreamSize(backup, sizeContainer)
     } else {
       assert.strictEqual(metadata.mode, 'delta')
 
       backup = await adapter.readDeltaVmBackup(metadata)
+      Object.values(backup.streams).forEach(stream => watchStreamSize(stream, sizeContainer))
     }
 
     return Task.run(
@@ -52,7 +57,7 @@ exports.ImportVmBackup = class ImportVmBackup {
         ])
 
         return {
-          size: metadata.size,
+          size: sizeContainer.size,
           id: await xapi.getField('VM', vmRef, 'uuid'),
         }
       }

@xen-orchestra/backups/RemoteAdapter.js

@@ -1,23 +1,24 @@
 const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
-const Disposable = require('promise-toolbox/Disposable')
-const fromCallback = require('promise-toolbox/fromCallback')
-const fromEvent = require('promise-toolbox/fromEvent')
-const pDefer = require('promise-toolbox/defer')
+const Disposable = require('promise-toolbox/Disposable.js')
+const fromCallback = require('promise-toolbox/fromCallback.js')
+const fromEvent = require('promise-toolbox/fromEvent.js')
+const pDefer = require('promise-toolbox/defer.js')
 const pump = require('pump')
 const { basename, dirname, join, normalize, resolve } = require('path')
 const { createLogger } = require('@xen-orchestra/log')
 const { createSyntheticStream, mergeVhd, default: Vhd } = require('vhd-lib')
-const { deduped } = require('@vates/disposable/deduped')
+const { deduped } = require('@vates/disposable/deduped.js')
 const { execFile } = require('child_process')
 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')
+const { BACKUP_DIR } = require('./_getVmBackupDir.js')
+const { cleanVm } = require('./_cleanVm.js')
+const { getTmpDir } = require('./_getTmpDir.js')
+const { isMetadataFile, isVhdFile } = require('./_backupType.js')
+const { isValidXva } = require('./_isValidXva.js')
+const { listPartitions, LVM_PARTITION_TYPE } = require('./_listPartitions.js')
+const { lvs, pvs } = require('./_lvm.js')
 
 const DIR_XO_CONFIG_BACKUPS = 'xo-config-backups'
 exports.DIR_XO_CONFIG_BACKUPS = DIR_XO_CONFIG_BACKUPS
@@ -505,21 +506,14 @@ class RemoteAdapter {
   }
 
   async outputStream(path, input, { checksum = true, validator = noop } = {}) {
-    const handler = this._handler
-    input = await input
-    const tmpPath = `${dirname(path)}/.${basename(path)}`
-    const output = await handler.createOutputStream(tmpPath, {
+    await this._handler.outputStream(path, input, {
       checksum,
       dirMode: this._dirMode,
+      async validator() {
+        await input.task
+        return validator.apply(this, arguments)
+      },
     })
-    try {
-      await Promise.all([fromCallback(pump, input, output), output.checksumWritten, input.task])
-      await validator(tmpPath)
-      await handler.rename(tmpPath, path, { checksum })
-    } catch (error) {
-      await handler.unlink(tmpPath, { checksum })
-      throw error
-    }
   }
 
   async readDeltaVmBackup(metadata) {
@@ -528,7 +522,7 @@ class RemoteAdapter {
     const dir = dirname(metadata._filename)
 
     const streams = {}
-    await asyncMapSettled(Object.entries(vdis), async ([id, vdi]) => {
+    await asyncMapSettled(Object.keys(vdis), async id => {
       streams[`${id}.vhd`] = await createSyntheticStream(handler, join(dir, vhds[id]))
     })
 
@@ -549,10 +543,51 @@ class RemoteAdapter {
   async readVmBackupMetadata(path) {
     return Object.defineProperty(JSON.parse(await this._handler.readFile(path)), '_filename', { value: path })
   }
+
+  async writeFullVmBackup({ jobId, mode, scheduleId, timestamp, vm, vmSnapshot, xva }, sizeContainer, stream) {
+    const basename = formatFilenameDate(timestamp)
+
+    const dataBasename = basename + '.xva'
+    const dataFilename = backupDir + '/' + dataBasename
+
+    const metadataFilename = `${backupDir}/${basename}.json`
+    const metadata = {
+      jobId: job.id,
+      mode: job.mode,
+      scheduleId,
+      timestamp,
+      version: '2.0.0',
+      vm,
+      vmSnapshot: this._backup.exportedVm,
+      xva: './' + dataBasename,
+    }
+
+    const { deleteFirst } = settings
+    if (deleteFirst) {
+      await deleteOldBackups()
+    }
+
+    await adapter.outputStream(stream, dataFilename, {
+      validator: tmpPath => {
+        if (handler._getFilePath !== undefined) {
+          return isValidXva(handler._getFilePath('/' + tmpPath))
+        }
+      },
+    })
+    metadata.size = sizeContainer.size
+    await handler.outputFile(metadataFilename, JSON.stringify(metadata))
+  }
 }
 
-RemoteAdapter.prototype.cleanVm = function (vmDir) {
-  return Disposable.use(this._handler.lock(vmDir), () => cleanVm.apply(this, arguments))
-}
+Object.assign(RemoteAdapter.prototype, {
+  cleanVm(vmDir, { lock = true } = {}) {
+    if (lock) {
+      return Disposable.use(this._handler.lock(vmDir), () => cleanVm.apply(this, arguments))
+    } else {
+      return cleanVm.apply(this, arguments)
+    }
+  },
+  isValidXva,
+})
 
 exports.RemoteAdapter = RemoteAdapter

@xen-orchestra/backups/RestoreMetadataBackup.js

@@ -1,5 +1,5 @@
-const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter')
-const { PATH_DB_DUMP } = require('./_PoolMetadataBackup')
+const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter.js')
+const { PATH_DB_DUMP } = require('./_PoolMetadataBackup.js')
 
 exports.RestoreMetadataBackup = class RestoreMetadataBackup {
   constructor({ backupId, handler, xapi }) {

@xen-orchestra/backups/Task.js

@@ -1,11 +1,15 @@
+const CancelToken = require('promise-toolbox/CancelToken.js')
 const Zone = require('node-zone')
 
-const { SyncThenable } = require('./_syncThenable')
-
-const logAfterEnd = () => {
-  throw new Error('task has already ended')
+const logAfterEnd = function (log) {
+  const error = new Error('task has already ended:' + this.id)
+  error.result = log.result
+  error.log = log
+  throw error
 }
 
+const noop = Function.prototype
+
 // Create a serializable object from an error.
 //
 // Otherwise some fields might be non-enumerable and missing from logs.
@@ -19,163 +23,142 @@ const serializeError = error =>
         stack: error.stack,
       }
     : error
-exports.serializeError = serializeError
 
-class TaskLogger {
-  constructor(logFn, parentId) {
-    this._log = logFn
-    this._parentId = parentId
-    this._taskId = undefined
+const $$task = Symbol('@xen-orchestra/backups/Task')
+
+class Task {
+  static get cancelToken() {
+    const task = Zone.current.data[$$task]
+    return task !== undefined ? task.#cancelToken : CancelToken.none
   }
 
-  get taskId() {
-    const taskId = this._taskId
-    if (taskId === undefined) {
-      throw new Error('start the task first')
-    }
-    return taskId
+  static run(opts, fn) {
+    return new this(opts).run(fn, true)
   }
 
-  // create a subtask
-  fork() {
-    return new TaskLogger(this._log, this.taskId)
-  }
-
-  info(message, data) {
-    return this._log({
-      data,
-      event: 'info',
-      message,
-      taskId: this.taskId,
-      timestamp: Date.now(),
-    })
-  }
-
-  run(message, data, fn) {
-    if (arguments.length === 2) {
-      fn = data
-      data = undefined
-    }
-
-    return SyncThenable.tryUnwrap(
-      SyncThenable.fromFunction(() => {
-        if (this._taskId !== undefined) {
-          throw new Error('task has already started')
-        }
-
-        this._taskId = Math.random().toString(36).slice(2)
-
-        return this._log({
-          data,
-          event: 'start',
-          message,
-          parentId: this._parentId,
-          taskId: this.taskId,
-          timestamp: Date.now(),
-        })
-      })
-        .then(fn)
-        .then(
-          result => {
-            const log = this._log
-            this._log = logAfterEnd
-            return SyncThenable.resolve(
-              log({
-                event: 'end',
-                result,
-                status: 'success',
-                taskId: this.taskId,
-                timestamp: Date.now(),
-              })
-            ).then(() => result)
-          },
-          error => {
-            const log = this._log
-            this._log = logAfterEnd
-            return SyncThenable.resolve(
-              log({
-                event: 'end',
-                result: serializeError(error),
-                status: 'failure',
-                taskId: this.taskId,
-                timestamp: Date.now(),
-              })
-            ).then(() => {
-              throw error
-            })
-          }
-        )
-    )
-  }
-
-  warning(message, data) {
-    return this._log({
-      data,
-      event: 'warning',
-      message,
-      taskId: this.taskId,
-      timestamp: Date.now(),
-    })
-  }
-
-  wrapFn(fn, message, data) {
-    const logger = this
-    return function () {
-      const evaluate = v => (typeof v === 'function' ? v.apply(this, arguments) : v)
-
-      return logger.run(evaluate(message), evaluate(data), () => fn.apply(this, arguments))
-    }
-  }
-}
-
-const $$task = Symbol('current task logger')
-
-const getCurrent = () => Zone.current.data[$$task]
-
-const Task = {
-  info(message, data) {
-    const task = getCurrent()
-    if (task !== undefined) {
-      return task.info(message, data)
-    }
-  },
-
-  run({ name, data, onLog }, fn) {
-    let parentId
-    if (onLog === undefined) {
-      const parent = getCurrent()
-      if (parent === undefined) {
-        return fn()
-      }
-      onLog = parent._log
-      parentId = parent.taskId
-    }
-
-    const task = new TaskLogger(onLog, parentId)
-    const zone = Zone.current.fork('task')
-    zone.data[$$task] = task
-    return task.run(name, data, zone.wrap(fn))
-  },
-
-  warning(message, data) {
-    const task = getCurrent()
-    if (task !== undefined) {
-      return task.warning(message, data)
-    }
-  },
-
-  wrapFn(opts, fn) {
+  static 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))
+      return Task.run(typeof opts === 'function' ? opts.apply(this, arguments) : opts, () => fn.apply(this, arguments))
     }
-  },
+  }
+
+  get id() {
+    return this.#id
+  }
+
+  #cancelToken
+  #id = Math.random().toString(36).slice(2)
+  #onLog
+  #zone
+
+  get id() {
+    return this.#id
+  }
+
+  constructor({ name, data, onLog }) {
+    let parentCancelToken, parentId
+    if (onLog === undefined) {
+      const parent = Zone.current.data[$$task]
+      if (parent === undefined) {
+        onLog = noop
+      } else {
+        onLog = log => parent.#onLog(log)
+        parentCancelToken = parent.#cancelToken
+        parentId = parent.#id
+      }
+    }
+
+    const zone = Zone.current.fork('@xen-orchestra/backups/Task')
+    zone.data[$$task] = this
+    this.#zone = zone
+
+    const { cancel, token } = CancelToken.source(parentCancelToken && [parentCancelToken])
+    this.#cancelToken = token
+    this.cancel = cancel
+
+    this.#onLog = onLog
+
+    this.#log('start', {
+      data,
+      message: name,
+      parentId,
+    })
+  }
+
+  failure(error) {
+    this.#end('failure', serializeError(error))
+  }
+
+  info(message, data) {
+    this.#log('info', { data, message })
+  }
+
+  /**
+   * Run a function in the context of this task
+   *
+   * In case of error, the task will be failed.
+   *
+   * @typedef Result
+   * @param {() => Result)} fn
+   * @param {boolean} last - Whether the task should succeed if there is no error
+   * @returns Result
+   */
+  run(fn, last = false) {
+    return this.#zone.run(() => {
+      try {
+        this.#cancelToken.throwIfRequested()
+
+        const result = fn()
+        let then
+        if (result != null && typeof (then = result.then) === 'function') {
+          then.call(result, last && (value => this.success(value)), error => this.failure(error))
+        } else if (last) {
+          this.success(result)
+        }
+        return result
+      } catch (error) {
+        this.failure(error)
+        throw error
+      }
+    })
+  }
+
+  success(value) {
+    this.#end('success', value)
+  }
+
+  warning(message, data) {
+    this.#log('warning', { data, message })
+  }
+
+  wrapFn(fn, last) {
+    const task = this
+    return function () {
+      return task.run(() => fn.apply(this, arguments), last)
+    }
+  }
+
+  #end(status, result) {
+    this.#log('end', { result, status })
+    this.#onLog = logAfterEnd
+  }
+
+  #log(event, props) {
+    this.#onLog({
+      ...props,
+      event,
+      taskId: this.#id,
+      timestamp: Date.now(),
+    })
+  }
 }
 exports.Task = Task
+
+for (const method of ['info', 'warning']) {
+  Task[method] = (...args) => Zone.current.data[$$task]?.[method](...args)
+}

@xen-orchestra/backups/_PoolMetadataBackup.js

@@ -1,9 +1,9 @@
 const { asyncMap } = require('@xen-orchestra/async-map')
 
-const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter')
-const { forkStreamUnpipe } = require('./_forkStreamUnpipe')
-const { formatFilenameDate } = require('./_filenameDate')
-const { Task } = require('./Task')
+const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter.js')
+const { forkStreamUnpipe } = require('./_forkStreamUnpipe.js')
+const { formatFilenameDate } = require('./_filenameDate.js')
+const { Task } = require('./Task.js')
 
 const PATH_DB_DUMP = '/pool/xmldbdump'
 exports.PATH_DB_DUMP = PATH_DB_DUMP

@xen-orchestra/backups/_VmBackup.js

@@ -1,23 +1,32 @@
-const findLast = require('lodash/findLast')
-const ignoreErrors = require('promise-toolbox/ignoreErrors')
-const keyBy = require('lodash/keyBy')
-const mapValues = require('lodash/mapValues')
-const { asyncMap } = require('@xen-orchestra/async-map')
+const assert = require('assert')
+// const asyncFn = require('promise-toolbox/asyncFn')
+const findLast = require('lodash/findLast.js')
+const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
+const keyBy = require('lodash/keyBy.js')
+const mapValues = require('lodash/mapValues.js')
+const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
 const { createLogger } = require('@xen-orchestra/log')
+const { defer } = require('golike-defer')
 const { formatDateTime } = require('@xen-orchestra/xapi')
 
-const { ContinuousReplicationWriter } = require('./_ContinuousReplicationWriter')
-const { DeltaBackupWriter } = require('./_DeltaBackupWriter')
-const { DisasterRecoveryWriter } = require('./_DisasterRecoveryWriter')
-const { exportDeltaVm } = require('./_deltaVm')
-const { forkStreamUnpipe } = require('./_forkStreamUnpipe')
-const { FullBackupWriter } = require('./_FullBackupWriter')
-const { getOldEntries } = require('./_getOldEntries')
-const { Task } = require('./Task')
-const { watchStreamSize } = require('./_watchStreamSize')
+const { DeltaBackupWriter } = require('./writers/DeltaBackupWriter.js')
+const { DeltaReplicationWriter } = require('./writers/DeltaReplicationWriter.js')
+const { exportDeltaVm } = require('./_deltaVm.js')
+const { forkStreamUnpipe } = require('./_forkStreamUnpipe.js')
+const { FullBackupWriter } = require('./writers/FullBackupWriter.js')
+const { FullReplicationWriter } = require('./writers/FullReplicationWriter.js')
+const { getOldEntries } = require('./_getOldEntries.js')
+const { Task } = require('./Task.js')
+const { watchStreamSize } = require('./_watchStreamSize.js')
 
 const { debug, warn } = createLogger('xo:backups:VmBackup')
 
+const asyncEach = async (iterable, fn, thisArg = iterable) => {
+  for (const item of iterable) {
+    await fn.call(thisArg, item)
+  }
+}
+
 const forkDeltaExport = deltaExport =>
   Object.create(deltaExport, {
     streams: {
@@ -62,12 +71,12 @@ exports.VmBackup = class VmBackup {
 
     // Create writers
     {
-      const writers = []
+      const writers = new Set()
       this._writers = writers
 
       const [BackupWriter, ReplicationWriter] = this._isDelta
-        ? [DeltaBackupWriter, ContinuousReplicationWriter]
-        : [FullBackupWriter, DisasterRecoveryWriter]
+        ? [DeltaBackupWriter, DeltaReplicationWriter]
+        : [FullBackupWriter, FullReplicationWriter]
 
       const allSettings = job.settings
 
@@ -77,7 +86,7 @@ exports.VmBackup = class VmBackup {
           ...allSettings[remoteId],
         }
         if (targetSettings.exportRetention !== 0) {
-          writers.push(new BackupWriter(this, remoteId, targetSettings))
+          writers.add(new BackupWriter({ backup: this, remoteId, settings: targetSettings }))
         }
       })
       srs.forEach(sr => {
@@ -86,12 +95,31 @@ exports.VmBackup = class VmBackup {
           ...allSettings[sr.uuid],
         }
         if (targetSettings.copyRetention !== 0) {
-          writers.push(new ReplicationWriter(this, sr, targetSettings))
+          writers.add(new ReplicationWriter({ backup: this, sr, settings: targetSettings }))
         }
       })
     }
   }
 
+  // calls fn for each function, warns of any errors, and throws only if there are no writers left
+  async _callWriters(fn, warnMessage, parallel = true) {
+    const writers = this._writers
+    if (writers.size === 0) {
+      return
+    }
+    await (parallel ? asyncMap : asyncEach)(writers, async function (writer) {
+      try {
+        await fn(writer)
+      } catch (error) {
+        this.delete(writer)
+        warn(warnMessage, { error, writer: writer.constructor.name })
+      }
+    })
+    if (writers.size === 0) {
+      throw new Error('all targets have failed, step: ' + warnMessage)
+    }
+  }
+
   // ensure the VM itself does not have any backup metadata which would be
   // copied on manual snapshots and interfere with the backup jobs
   async _cleanMetadata() {
@@ -114,7 +142,9 @@ exports.VmBackup = class VmBackup {
 
     const settings = this._settings
 
-    const doSnapshot = this._isDelta || vm.power_state === 'Running' || settings.snapshotRetention !== 0
+    const doSnapshot =
+      this._isDelta || (!settings.offlineBackup && vm.power_state === 'Running') || settings.snapshotRetention !== 0
+    console.log({ doSnapshot })
     if (doSnapshot) {
       await Task.run({ name: 'snapshot' }, async () => {
         if (!settings.bypassVdiChainsCheck) {
@@ -146,31 +176,29 @@ exports.VmBackup = class VmBackup {
   async _copyDelta() {
     const { exportedVm } = this
     const baseVm = this._baseVm
+    const fullVdisRequired = this._fullVdisRequired
 
-    await asyncMap(this._writers, writer => writer.prepare && writer.prepare())
+    const isFull = fullVdisRequired === undefined || fullVdisRequired.size !== 0
+
+    await this._callWriters(writer => writer.prepare({ isFull }), 'writer.prepare()')
 
     const deltaExport = await exportDeltaVm(exportedVm, baseVm, {
-      fullVdisRequired: this._fullVdisRequired,
+      cancelToken: Task.cancelToken,
+      fullVdisRequired,
     })
-    const sizeContainers = mapValues(deltaExport.streams, watchStreamSize)
+    const sizeContainers = mapValues(deltaExport.streams, stream => watchStreamSize(stream))
 
     const timestamp = Date.now()
 
-    await asyncMap(this._writers, async writer => {
-      try {
-        await writer.transfer({
+    await this._callWriters(
+      writer =>
+        writer.transfer({
           deltaExport: forkDeltaExport(deltaExport),
           sizeContainers,
           timestamp,
-        })
-      } catch (error) {
-        warn('copy failure', {
-          error,
-          target: writer.target,
-          vm: this.vm,
-        })
-      }
-    })
+        }),
+      'writer.transfer()'
+    )
 
     this._baseVm = exportedVm
 
@@ -195,12 +223,13 @@ exports.VmBackup = class VmBackup {
       size,
     })
 
-    await asyncMap(this._writers, writer => writer.cleanup && writer.cleanup())
+    await this._callWriters(writer => writer.cleanup(), 'writer.cleanup()')
   }
 
   async _copyFull() {
     const { compression } = this.job
     const stream = await this._xapi.VM_export(this.exportedVm.$ref, {
+      cancelToken: Task.cancelToken,
       compress: Boolean(compression) && (compression === 'native' ? 'gzip' : 'zstd'),
       useSnapshot: false,
     })
@@ -208,21 +237,15 @@ exports.VmBackup = class VmBackup {
 
     const timestamp = Date.now()
 
-    await asyncMap(this._writers, async writer => {
-      try {
-        await writer.run({
+    await this._callWriters(
+      writer =>
+        writer.run({
           sizeContainer,
           stream: forkStreamUnpipe(stream),
           timestamp,
-        })
-      } catch (error) {
-        warn('copy failure', {
-          error,
-          target: writer.target,
-          vm: this.vm,
-        })
-      }
-    })
+        }),
+      'writer.run()'
+    )
 
     const { size } = sizeContainer
     const end = Date.now()
@@ -296,14 +319,11 @@ exports.VmBackup = class VmBackup {
     })
 
     const presentBaseVdis = new Map(baseUuidToSrcVdi)
-    const writers = this._writers
-    for (let i = 0, n = writers.length; presentBaseVdis.size !== 0 && i < n; ++i) {
-      await writers[i].checkBaseVdis(presentBaseVdis, baseVm)
-    }
-
-    if (presentBaseVdis.size === 0) {
-      return
-    }
+    await this._callWriters(
+      writer => presentBaseVdis.size !== 0 && writer.checkBaseVdis(presentBaseVdis, baseVm),
+      'writer.checkBaseVdis()',
+      false
+    )
 
     const fullVdisRequired = new Set()
     baseUuidToSrcVdi.forEach((srcVdi, baseUuid) => {
@@ -314,9 +334,33 @@ exports.VmBackup = class VmBackup {
 
     this._baseVm = baseVm
     this._fullVdisRequired = fullVdisRequired
+
+    Task.info('base data', {
+      vm: baseVm.uuid,
+      fullVdisRequired: Array.from(fullVdisRequired),
+    })
   }
 
-  async run() {
+  run = defer(this.run)
+  async run($defer) {
+    this.exportedVm = this.vm
+    this.timestamp = Date.now()
+
+    const doSnapshot = this._isDelta || vm.power_state === 'Running' || settings.snapshotRetention !== 0
+    if (!this._isDelta) {
+    }
+
+    const settings = this._settings
+    assert(
+      !settings.offlineBackup || settings.snapshotRetention === 0,
+      'offlineBackup is not compatible with snapshotRetention'
+    )
+
+    await this._callWriters(async writer => {
+      await writer.beforeBackup()
+      $defer(() => writer.afterBackup())
+    }, 'writer.beforeBackup()')
+
     await this._fetchJobSnapshots()
 
     if (this._isDelta) {
@@ -326,7 +370,7 @@ exports.VmBackup = class VmBackup {
     await this._cleanMetadata()
     await this._removeUnusedSnapshots()
 
-    const { _settings: settings, vm } = this
+    const { vm } = this
     const isRunning = vm.power_state === 'Running'
     const startAfter = isRunning && (settings.offlineBackup ? 'backup' : settings.offlineSnapshot && 'snapshot')
     if (startAfter) {
@@ -339,7 +383,7 @@ exports.VmBackup = class VmBackup {
         ignoreErrors.call(vm.$callAsync('start', false, false))
       }
 
-      if (this._writers.length !== 0) {
+      if (this._writers.size !== 0) {
         await (this._isDelta ? this._copyDelta() : this._copyFull())
       }
     } finally {
@@ -352,3 +396,6 @@ exports.VmBackup = class VmBackup {
     }
   }
 }
+
+// const { prototype } = exports.VmBackup
+// prototype.run = asyncFn.cancelable(prototype.run)

@xen-orchestra/backups/_XoMetadataBackup.js

@@ -1,8 +1,8 @@
 const { asyncMap } = require('@xen-orchestra/async-map')
 
-const { DIR_XO_CONFIG_BACKUPS } = require('./RemoteAdapter')
-const { formatFilenameDate } = require('./_filenameDate')
-const { Task } = require('./Task')
+const { DIR_XO_CONFIG_BACKUPS } = require('./RemoteAdapter.js')
+const { formatFilenameDate } = require('./_filenameDate.js')
+const { Task } = require('./Task.js')
 
 exports.XoMetadataBackup = class XoMetadataBackup {
   constructor({ config, job, remoteAdapters, schedule, settings }) {

@xen-orchestra/backups/_backupWorker.js

@@ -1,15 +1,19 @@
-const Disposable = require('promise-toolbox/Disposable')
-const ignoreErrors = require('promise-toolbox/ignoreErrors')
+require('@xen-orchestra/log/configure.js').catchGlobalErrors(
+  require('@xen-orchestra/log').createLogger('xo:backups:worker')
+)
+
+const Disposable = require('promise-toolbox/Disposable.js')
+const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
 const { compose } = require('@vates/compose')
-const { createDebounceResource } = require('@vates/disposable/debounceResource')
-const { deduped } = require('@vates/disposable/deduped')
+const { createDebounceResource } = require('@vates/disposable/debounceResource.js')
+const { deduped } = require('@vates/disposable/deduped.js')
 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')
+const { Backup } = require('./Backup.js')
+const { RemoteAdapter } = require('./RemoteAdapter.js')
+const { Task } = require('./Task.js')
 
 class BackupWorker {
   #config

@xen-orchestra/backups/_cancelableMap.js

@@ -1,5 +1,5 @@
-const cancelable = require('promise-toolbox/cancelable')
-const CancelToken = require('promise-toolbox/CancelToken')
+const cancelable = require('promise-toolbox/cancelable.js')
+const CancelToken = require('promise-toolbox/CancelToken.js')
 
 // Similar to `Promise.all` + `map` but pass a cancel token to the callback
 //

@xen-orchestra/backups/_cleanVm.js

@@ -1,11 +1,10 @@
 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')
+const { DISK_TYPE_DIFFERENCING } = require('vhd-lib/dist/_constants.js')
+const { isMetadataFile, isVhdFile, isXvaFile, isXvaSumFile } = require('./_backupType.js')
+const { limitConcurrency } = require('limit-concurrency-decorator')
 
 // chain is an array of VHDs from child to parent
 //
@@ -35,6 +34,8 @@ const mergeVhdChain = limitConcurrency(1)(async function mergeVhdChain(chain, {
       child = children[0]
     }
 
+    onLog(`merging ${child} into ${parent}`)
+
     let done, total
     const handle = setInterval(() => {
       if (done !== undefined) {
@@ -59,19 +60,59 @@ const mergeVhdChain = limitConcurrency(1)(async function mergeVhdChain(chain, {
     )
 
     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)
-    }),
-  ])
+    await Promise.all([
+      handler.rename(parent, child),
+      asyncMap(children.slice(0, -1), child => {
+        onLog(`the VHD ${child} is unused`)
+        if (remove) {
+          onLog(`deleting unused VHD ${child}`)
+          return handler.unlink(child)
+        }
+      }),
+    ])
+  }
 })
 
 const noop = Function.prototype
 
+const INTERRUPTED_VHDS_REG = /^(?:(.+)\/)?\.(.+)\.merge.json$/
+const listVhds = async (handler, vmDir) => {
+  const vhds = []
+  const interruptedVhds = new Set()
+
+  await asyncMap(
+    await handler.list(`${vmDir}/vdis`, {
+      ignoreMissing: true,
+      prependDir: true,
+    }),
+    async jobDir =>
+      asyncMap(
+        await handler.list(jobDir, {
+          prependDir: true,
+        }),
+        async vdiDir => {
+          const list = await handler.list(vdiDir, {
+            filter: file => isVhdFile(file) || INTERRUPTED_VHDS_REG.test(file),
+            prependDir: true,
+          })
+
+          list.forEach(file => {
+            const res = INTERRUPTED_VHDS_REG.exec(file)
+            if (res === null) {
+              vhds.push(file)
+            } else {
+              const [, dir, file] = res
+              interruptedVhds.add(`${dir}/${file}`)
+            }
+          })
+        }
+      )
+  )
+
+  return { vhds, interruptedVhds }
+}
+
 exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop }) {
   const handler = this._handler
 
@@ -79,37 +120,34 @@ exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop })
   const vhdParents = { __proto__: null }
   const vhdChildren = { __proto__: null }
 
+  const vhdsList = await listVhds(handler, vmDir)
+
   // 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)
+  await asyncMap(vhdsList.vhds, async path => {
+    try {
+      const vhd = new Vhd(handler, path)
+      await vhd.readHeaderAndFooter(!vhdsList.interruptedVhds.has(path))
+      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}`, { error })
+      if (error?.code === 'ERR_ASSERTION' && remove) {
+        onLog(`deleting broken ${path}`)
+        await handler.unlink(path)
       }
     }
-  )
+  })
 
   // remove VHDs with missing ancestors
   {
@@ -132,6 +170,7 @@ exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop })
 
         onLog(`the parent ${parent} of the VHD ${vhd} is missing`)
         if (remove) {
+          onLog(`deleting orphan VHD ${vhd}`)
           deletions.push(handler.unlink(vhd))
         }
       }
@@ -167,7 +206,7 @@ exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop })
   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))) {
+    if (!(await this.isValidXva(path))) {
       onLog(`the XVA with path ${path} is potentially broken`)
     }
   })
@@ -181,20 +220,21 @@ exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop })
     const metadata = JSON.parse(await handler.readFile(json))
     const { mode } = metadata
     if (mode === 'full') {
-      const linkedXva = resolve(vmDir, metadata.xva)
+      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) {
+          onLog(`deleting incomplete backup ${json}`)
           await handler.unlink(json)
         }
       }
     } else if (mode === 'delta') {
       const linkedVhds = (() => {
         const { vhds } = metadata
-        return Object.keys(vhds).map(key => resolve(vmDir, vhds[key]))
+        return Object.keys(vhds).map(key => resolve('/', vmDir, vhds[key]))
       })()
 
       // FIXME: find better approach by keeping as much of the backup as
@@ -204,6 +244,7 @@ exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop })
       } else {
         onLog(`Some VHDs linked to the metadata ${json} are missing`)
         if (remove) {
+          onLog(`deleting incomplete backup ${json}`)
           await handler.unlink(json)
         }
       }
@@ -244,6 +285,7 @@ exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop })
 
       onLog(`the VHD ${vhd} is unused`)
       if (remove) {
+        onLog(`deleting unused VHD ${vhd}`)
         unusedVhdsDeletion.push(handler.unlink(vhd))
       }
     }
@@ -252,25 +294,38 @@ exports.cleanVm = async function cleanVm(vmDir, { remove, merge, onLog = noop })
       vhdChainsToMerge[vhd] = getUsedChildChainOrDelete(vhd)
     })
 
+    // merge interrupted VHDs
+    if (merge) {
+      vhdsList.interruptedVhds.forEach(parent => {
+        vhdChainsToMerge[parent] = [vhdChildren[parent], parent]
+      })
+    }
+
     Object.keys(vhdChainsToMerge).forEach(key => {
       const chain = vhdChainsToMerge[key]
       if (chain !== undefined) {
-        unusedVhdsDeletion.push(mergeVhdChain(chain, { onLog, remove, merge }))
+        unusedVhdsDeletion.push(mergeVhdChain(chain, { handler, onLog, remove, merge }))
       }
     })
   }
 
   await Promise.all([
-    unusedVhdsDeletion,
+    ...unusedVhdsDeletion,
     asyncMap(unusedXvas, path => {
       onLog(`the XVA ${path} is unused`)
-      return remove && handler.unlink(path)
+      if (remove) {
+        onLog(`deleting unused XVA ${path}`)
+        return 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)
+        if (remove) {
+          onLog(`deleting unused XVA checksum ${path}`)
+          return handler.unlink(path)
+        }
       }
     }),
   ])

@xen-orchestra/backups/_deltaVm.js

@@ -1,14 +1,14 @@
 const compareVersions = require('compare-versions')
-const defer = require('golike-defer').default
-const find = require('lodash/find')
-const groupBy = require('lodash/groupBy')
-const ignoreErrors = require('promise-toolbox/ignoreErrors')
-const omit = require('lodash/omit')
+const find = require('lodash/find.js')
+const groupBy = require('lodash/groupBy.js')
+const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
+const omit = require('lodash/omit.js')
 const { asyncMap } = require('@xen-orchestra/async-map')
 const { CancelToken } = require('promise-toolbox')
 const { createVhdStreamWithLength } = require('vhd-lib')
+const { defer } = require('golike-defer')
 
-const { cancelableMap } = require('./_cancelableMap')
+const { cancelableMap } = require('./_cancelableMap.js')
 
 const TAG_BASE_DELTA = 'xo:base_delta'
 exports.TAG_BASE_DELTA = TAG_BASE_DELTA

@xen-orchestra/backups/_getTmpDir.js

@@ -1,4 +1,4 @@
-const Disposable = require('promise-toolbox/Disposable')
+const Disposable = require('promise-toolbox/Disposable.js')
 const { join } = require('path')
 const { mkdir, rmdir } = require('fs-extra')
 const { tmpdir } = require('os')
@@ -10,7 +10,7 @@ exports.getTmpDir = async function getTmpDir() {
     const path = join(tmpdir(), Math.random().toString(36).slice(2))
     try {
       await mkdir(path)
-      return new Disposable(path, () => rmdir(path))
+      return new Disposable(() => rmdir(path), path)
     } catch (error) {
       if (i === MAX_ATTEMPTS) {
         throw error

@xen-orchestra/backups/_isValidXva.js

@@ -1,10 +1,10 @@
 const assert = require('assert')
-const fs = require('fs-extra')
 
-const isGzipFile = async fd => {
+const isGzipFile = async (handler, fd) => {
   // https://tools.ietf.org/html/rfc1952.html#page-5
   const magicNumber = Buffer.allocUnsafe(2)
-  assert.strictEqual((await fs.read(fd, magicNumber, 0, magicNumber.length, 0)).bytesRead, magicNumber.length)
+
+  assert.strictEqual((await handler.read(fd, magicNumber, 0)).bytesRead, magicNumber.length)
   return magicNumber[0] === 31 && magicNumber[1] === 139
 }
 
@@ -21,32 +21,33 @@ const isGzipFile = async fd => {
 // /^Ref:\d+/\d+\.checksum$/ and then validating the tar structure from it
 //
 // https://github.com/npm/node-tar/issues/234#issuecomment-538190295
-const isValidTar = async (size, fd) => {
+const isValidTar = async (handler, size, fd) => {
   if (size <= 1024 || size % 512 !== 0) {
     return false
   }
 
   const buf = Buffer.allocUnsafe(1024)
-  assert.strictEqual((await fs.read(fd, buf, 0, buf.length, size - buf.length)).bytesRead, buf.length)
+  assert.strictEqual((await handler.read(fd, buf, size - buf.length)).bytesRead, buf.length)
   return buf.every(_ => _ === 0)
 }
 
 // TODO: find an heuristic for compressed files
-const isValidXva = async path => {
+async function isValidXva(path) {
+  const handler = this._handler
   try {
-    const fd = await fs.open(path, 'r')
+    const fd = await handler.openFile(path, 'r')
     try {
-      const { size } = await fs.fstat(fd)
+      const size = await handler.getSize(fd)
       if (size < 20) {
         // neither a valid gzip not tar
         return false
       }
 
-      return (await isGzipFile(fd))
+      return (await isGzipFile(handler, fd))
         ? true // gzip files cannot be validated at this time
-        : await isValidTar(size, fd)
+        : await isValidTar(handler, size, fd)
     } finally {
-      fs.close(fd).catch(noop)
+      handler.closeFile(fd).catch(noop)
     }
   } catch (error) {
     // never throw, log and report as valid to avoid side effects

@xen-orchestra/backups/_listPartitions.js

@@ -1,4 +1,4 @@
-const fromCallback = require('promise-toolbox/fromCallback')
+const fromCallback = require('promise-toolbox/fromCallback.js')
 const { createLogger } = require('@xen-orchestra/log')
 const { createParser } = require('parse-pairs')
 const { execFile } = require('child_process')

@xen-orchestra/backups/_lvm.js

@@ -1,4 +1,4 @@
-const fromCallback = require('promise-toolbox/fromCallback')
+const fromCallback = require('promise-toolbox/fromCallback.js')
 const { createParser } = require('parse-pairs')
 const { execFile } = require('child_process')
 

@xen-orchestra/backups/_syncThenable.js

@@ -1,46 +0,0 @@
-function fulfilledThen(cb) {
-  return typeof cb === 'function' ? SyncThenable.fromFunction(cb, this.value) : this
-}
-
-function rejectedThen(_, cb) {
-  return typeof cb === 'function' ? SyncThenable.fromFunction(cb, this.value) : this
-}
-
-class SyncThenable {
-  static resolve(value) {
-    if (value != null && typeof value.then === 'function') {
-      return value
-    }
-
-    return new this(false, value)
-  }
-
-  static fromFunction(fn, ...arg) {
-    try {
-      return this.resolve(fn(...arg))
-    } catch (error) {
-      return this.reject(error)
-    }
-  }
-
-  static reject(reason) {
-    return new this(true, reason)
-  }
-
-  // unwrap if it's a SyncThenable
-  static tryUnwrap(value) {
-    if (value instanceof this) {
-      if (value.then === rejectedThen) {
-        throw value.value
-      }
-      return value.value
-    }
-    return value
-  }
-
-  constructor(rejected, value) {
-    this.then = rejected ? rejectedThen : fulfilledThen
-    this.value = value
-  }
-}
-exports.SyncThenable = SyncThenable

@xen-orchestra/backups/_watchStreamSize.js

@@ -1,5 +1,4 @@
-exports.watchStreamSize = function watchStreamSize(stream) {
-  const container = { size: 0 }
+exports.watchStreamSize = function watchStreamSize(stream, container = { size: 0 }) {
   stream.on('data', data => {
     container.size += data.length
   })

@xen-orchestra/backups/formatVmBackups.js

@@ -1,4 +1,4 @@
-const mapValues = require('lodash/mapValues')
+const mapValues = require('lodash/mapValues.js')
 const { dirname } = require('path')
 
 function formatVmBackup(backup) {

@xen-orchestra/backups/package.json

@@ -8,9 +8,9 @@
     "type": "git",
     "url": "https://github.com/vatesfr/xen-orchestra.git"
   },
-  "version": "0.9.1",
+  "version": "0.11.0",
   "engines": {
-    "node": ">=14.5"
+    "node": ">=14.6"
   },
   "scripts": {
     "postversion": "npm publish --access public"
@@ -18,28 +18,27 @@
   "dependencies": {
     "@vates/compose": "^2.0.0",
     "@vates/disposable": "^0.1.1",
-    "@vates/multi-key-map": "^0.1.0",
-    "@vates/parse-duration": "^0.1.0",
+    "@vates/parse-duration": "^0.1.1",
     "@xen-orchestra/async-map": "^0.1.2",
-    "@xen-orchestra/fs": "^0.14.0",
+    "@xen-orchestra/fs": "^0.17.0",
     "@xen-orchestra/log": "^0.2.0",
     "@xen-orchestra/template": "^0.1.0",
     "compare-versions": "^3.6.0",
     "d3-time-format": "^3.0.0",
     "end-of-stream": "^1.4.4",
-    "ensure-array": "^1.0.0",
     "fs-extra": "^9.0.0",
     "golike-defer": "^0.5.1",
-    "limit-concurrency-decorator": "^0.4.0",
+    "limit-concurrency-decorator": "^0.5.0",
     "lodash": "^4.17.20",
     "node-zone": "^0.4.0",
     "parse-pairs": "^1.1.0",
-    "promise-toolbox": "^0.18.0",
+    "pump": "^3.0.0",
+    "promise-toolbox": "^0.19.2",
     "vhd-lib": "^1.0.0",
     "yazl": "^2.5.1"
   },
   "peerDependencies": {
-    "@xen-orchestra/xapi": "^0.6.0"
+    "@xen-orchestra/xapi": "^0.6.2"
   },
   "license": "AGPL-3.0-or-later",
   "author": {

@xen-orchestra/backups/parseMetadataBackupId.js

@@ -1,4 +1,4 @@
-const { DIR_XO_CONFIG_BACKUPS, DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter')
+const { DIR_XO_CONFIG_BACKUPS, DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter.js')
 
 exports.parseMetadataBackupId = function parseMetadataBackupId(backupId) {
   const [dir, ...rest] = backupId.split('/')

@xen-orchestra/backups/writers/DeltaBackupWriter.js

@@ -1,42 +1,25 @@
 const assert = require('assert')
-const map = require('lodash/map')
-const mapValues = require('lodash/mapValues')
-const ignoreErrors = require('promise-toolbox/ignoreErrors')
+const map = require('lodash/map.js')
+const mapValues = require('lodash/mapValues.js')
+const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
 const { asyncMap } = require('@xen-orchestra/async-map')
 const { chainVhd, checkVhdChain, default: Vhd } = require('vhd-lib')
 const { createLogger } = require('@xen-orchestra/log')
 const { dirname } = require('path')
 
-const { checkVhd } = require('./_checkVhd')
-const { formatFilenameDate } = require('./_filenameDate')
-const { getOldEntries } = require('./_getOldEntries')
-const { getVmBackupDir } = require('./_getVmBackupDir')
-const { packUuid } = require('./_packUuid')
-const { Task } = require('./Task')
+const { formatFilenameDate } = require('../_filenameDate.js')
+const { getOldEntries } = require('../_getOldEntries.js')
+const { getVmBackupDir } = require('../_getVmBackupDir.js')
+const { Task } = require('../Task.js')
+
+const { MixinBackupWriter } = require('./_MixinBackupWriter.js')
+const { AbstractDeltaWriter } = require('./_AbstractDeltaWriter.js')
+const { checkVhd } = require('./_checkVhd.js')
+const { packUuid } = require('./_packUuid.js')
 
 const { warn } = createLogger('xo:backups:DeltaBackupWriter')
 
-exports.DeltaBackupWriter = class DeltaBackupWriter {
-  constructor(backup, remoteId, settings) {
-    this._adapter = backup.remoteAdapters[remoteId]
-    this._backup = backup
-    this._settings = settings
-
-    this.transfer = Task.wrapFn(
-      {
-        name: 'export',
-        data: ({ deltaExport }) => ({
-          id: remoteId,
-          isFull: Object.values(deltaExport.vdis).some(vdi => vdi.other_config['xo:base_delta'] === undefined),
-          type: 'remote',
-        }),
-      },
-      this.transfer
-    )
-
-    this[settings.deleteFirst ? 'prepare' : 'cleanup'] = this._deleteOldEntries
-  }
-
+exports.DeltaBackupWriter = class DeltaBackupWriter extends MixinBackupWriter(AbstractDeltaWriter) {
   async checkBaseVdis(baseUuidToSrcVdi) {
     const { handler } = this._adapter
     const backup = this._backup
@@ -72,7 +55,28 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
     })
   }
 
-  async prepare() {
+  async beforeBackup() {
+    await super.beforeBackup()
+    return this._cleanVm({ merge: true })
+  }
+
+  prepare({ isFull }) {
+    // create the task related to this export and ensure all methods are called in this context
+    const task = new Task({
+      name: 'export',
+      data: {
+        id: this._remoteId,
+        isFull,
+        type: 'remote',
+      },
+    })
+    this.transfer = task.wrapFn(this.transfer)
+    this.cleanup = task.wrapFn(this.cleanup, true)
+
+    return task.run(() => this._prepare())
+  }
+
+  async _prepare() {
     const adapter = this._adapter
     const settings = this._settings
     const { scheduleId, vm } = this._backup
@@ -99,8 +103,12 @@ exports.DeltaBackupWriter = class DeltaBackupWriter {
 
     if (settings.deleteFirst) {
       await this._deleteOldEntries()
-    } else {
-      this.cleanup = this._deleteOldEntries
+    }
+  }
+
+  async cleanup() {
+    if (!this._settings.deleteFirst) {
+      await this._deleteOldEntries()
     }
   }
 

@xen-orchestra/backups/writers/DeltaReplicationWriter.js

@@ -1,32 +1,17 @@
 const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
-const ignoreErrors = require('promise-toolbox/ignoreErrors')
+const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
 const { formatDateTime } = require('@xen-orchestra/xapi')
 
-const { formatFilenameDate } = require('./_filenameDate')
-const { getOldEntries } = require('./_getOldEntries')
-const { importDeltaVm, TAG_COPY_SRC } = require('./_deltaVm')
-const { listReplicatedVms } = require('./_listReplicatedVms')
-const { Task } = require('./Task')
+const { formatFilenameDate } = require('../_filenameDate.js')
+const { getOldEntries } = require('../_getOldEntries.js')
+const { importDeltaVm, TAG_COPY_SRC } = require('../_deltaVm.js')
+const { Task } = require('../Task.js')
 
-exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
-  constructor(backup, sr, settings) {
-    this._backup = backup
-    this._settings = settings
-    this._sr = sr
-
-    this.transfer = Task.wrapFn(
-      {
-        name: 'export',
-        data: ({ deltaExport }) => ({
-          id: sr.uuid,
-          isFull: Object.values(deltaExport.vdis).some(vdi => vdi.other_config['xo:base_delta'] === undefined),
-          type: 'SR',
-        }),
-      },
-      this.transfer
-    )
-  }
+const { AbstractDeltaWriter } = require('./_AbstractDeltaWriter.js')
+const { MixinReplicationWriter } = require('./_MixinReplicationWriter.js')
+const { listReplicatedVms } = require('./_listReplicatedVms.js')
 
+exports.DeltaReplicationWriter = class DeltaReplicationWriter extends MixinReplicationWriter(AbstractDeltaWriter) {
   async checkBaseVdis(baseUuidToSrcVdi, baseVm) {
     const sr = this._sr
     const replicatedVm = listReplicatedVms(sr.$xapi, this._backup.job.id, sr.uuid, this._backup.vm.uuid).find(
@@ -51,7 +36,23 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
     }
   }
 
-  async prepare() {
+  prepare({ isFull }) {
+    // create the task related to this export and ensure all methods are called in this context
+    const task = new Task({
+      name: 'export',
+      data: {
+        id: this._sr.uuid,
+        isFull,
+        type: 'SR',
+      },
+    })
+    this.transfer = task.wrapFn(this.transfer)
+    this.cleanup = task.wrapFn(this.cleanup, true)
+
+    return task.run(() => this._prepare())
+  }
+
+  async _prepare() {
     const settings = this._settings
     const { uuid: srUuid, $xapi: xapi } = this._sr
     const { scheduleId, vm } = this._backup
@@ -63,8 +64,12 @@ exports.ContinuousReplicationWriter = class ContinuousReplicationWriter {
 
     if (settings.deleteFirst) {
       await this._deleteOldEntries()
-    } else {
-      this.cleanup = this._deleteOldEntries
+    }
+  }
+
+  async cleanup() {
+    if (!this._settings.deleteFirst) {
+      await this._deleteOldEntries()
     }
   }
 

@xen-orchestra/backups/writers/FullBackupWriter.js

@@ -1,20 +1,20 @@
-const { formatFilenameDate } = require('./_filenameDate')
-const { getOldEntries } = require('./_getOldEntries')
-const { getVmBackupDir } = require('./_getVmBackupDir')
-const { isValidXva } = require('./isValidXva')
-const { Task } = require('./Task')
+const { formatFilenameDate } = require('../_filenameDate.js')
+const { getOldEntries } = require('../_getOldEntries.js')
+const { getVmBackupDir } = require('../_getVmBackupDir.js')
+const { Task } = require('../Task.js')
 
-exports.FullBackupWriter = class FullBackupWriter {
-  constructor(backup, remoteId, settings) {
-    this._backup = backup
-    this._remoteId = remoteId
-    this._settings = settings
+const { MixinBackupWriter } = require('./_MixinBackupWriter.js')
+const { AbstractFullWriter } = require('./_AbstractFullWriter.js')
+
+exports.FullBackupWriter = class FullBackupWriter extends MixinBackupWriter(AbstractFullWriter) {
+  constructor(props) {
+    super(props)
 
     this.run = Task.wrapFn(
       {
         name: 'export',
         data: {
-          id: remoteId,
+          id: props.remoteId,
           type: 'remote',
 
           // necessary?
@@ -27,12 +27,11 @@ exports.FullBackupWriter = class FullBackupWriter {
 
   async run({ timestamp, sizeContainer, stream }) {
     const backup = this._backup
-    const remoteId = this._remoteId
     const settings = this._settings
 
     const { job, scheduleId, vm } = backup
 
-    const adapter = backup.remoteAdapters[remoteId]
+    const adapter = this._adapter
     const handler = adapter.handler
     const backupDir = getVmBackupDir(vm.uuid)
 
@@ -68,11 +67,7 @@ exports.FullBackupWriter = class FullBackupWriter {
 
     await Task.run({ name: 'transfer' }, async () => {
       await adapter.outputStream(dataFilename, stream, {
-        validator: tmpPath => {
-          if (handler._getFilePath !== undefined) {
-            return isValidXva(handler._getFilePath('/' + tmpPath))
-          }
-        },
+        validator: tmpPath => adapter.isValidXva(tmpPath),
       })
       return { size: sizeContainer.size }
     })

@xen-orchestra/backups/writers/FullReplicationWriter.js

@@ -1,23 +1,24 @@
-const ignoreErrors = require('promise-toolbox/ignoreErrors')
+const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
 const { asyncMapSettled } = require('@xen-orchestra/async-map')
 const { formatDateTime } = require('@xen-orchestra/xapi')
 
-const { formatFilenameDate } = require('./_filenameDate')
-const { getOldEntries } = require('./_getOldEntries')
-const { listReplicatedVms } = require('./_listReplicatedVms')
-const { Task } = require('./Task')
+const { formatFilenameDate } = require('../_filenameDate.js')
+const { getOldEntries } = require('../_getOldEntries.js')
+const { Task } = require('../Task.js')
 
-exports.DisasterRecoveryWriter = class DisasterRecoveryWriter {
-  constructor(backup, sr, settings) {
-    this._backup = backup
-    this._settings = settings
-    this._sr = sr
+const { AbstractFullWriter } = require('./_AbstractFullWriter.js')
+const { MixinReplicationWriter } = require('./_MixinReplicationWriter.js')
+const { listReplicatedVms } = require('./_listReplicatedVms.js')
+
+exports.FullReplicationWriter = class FullReplicationWriter extends MixinReplicationWriter(AbstractFullWriter) {
+  constructor(props) {
+    super(props)
 
     this.run = Task.wrapFn(
       {
         name: 'export',
         data: {
-          id: sr.uuid,
+          id: props.sr.uuid,
           type: 'SR',
 
           // necessary?

@xen-orchestra/backups/writers/_AbstractDeltaWriter.js

@@ -0,0 +1,19 @@
+const { AbstractWriter } = require('./_AbstractWriter.js')
+
+exports.AbstractDeltaWriter = class AbstractDeltaWriter extends AbstractWriter {
+  checkBaseVdis(baseUuidToSrcVdi, baseVm) {
+    throw new Error('Not implemented')
+  }
+
+  cleanup() {
+    throw new Error('Not implemented')
+  }
+
+  prepare({ isFull }) {
+    throw new Error('Not implemented')
+  }
+
+  transfer({ timestamp, deltaExport, sizeContainers }) {
+    throw new Error('Not implemented')
+  }
+}

@xen-orchestra/backups/writers/_AbstractFullWriter.js

@@ -0,0 +1,7 @@
+const { AbstractWriter } = require('./_AbstractWriter.js')
+
+exports.AbstractFullWriter = class AbstractFullWriter extends AbstractWriter {
+  run({ timestamp, sizeContainer, stream }) {
+    throw new Error('Not implemented')
+  }
+}

@xen-orchestra/backups/writers/_AbstractWriter.js

@@ -0,0 +1,10 @@
+exports.AbstractWriter = class AbstractWriter {
+  constructor({ backup, settings }) {
+    this._backup = backup
+    this._settings = settings
+  }
+
+  beforeBackup() {}
+
+  afterBackup() {}
+}

@xen-orchestra/backups/writers/_MixinBackupWriter.js

@@ -0,0 +1,34 @@
+const { createLogger } = require('@xen-orchestra/log')
+
+const { getVmBackupDir } = require('../_getVmBackupDir.js')
+
+const { warn } = createLogger('xo:backups:MixinBackupWriter')
+
+exports.MixinBackupWriter = (BaseClass = Object) =>
+  class MixinBackupWriter extends BaseClass {
+    constructor({ remoteId, ...rest }) {
+      super(rest)
+
+      this._adapter = rest.backup.remoteAdapters[remoteId]
+      this._remoteId = remoteId
+      this._lock = undefined
+    }
+
+    _cleanVm(options) {
+      return this._adapter
+        .cleanVm(getVmBackupDir(this._backup.vm.uuid), { ...options, onLog: warn, lock: false })
+        .catch(warn)
+    }
+
+    async beforeBackup() {
+      const { handler } = this._adapter
+      const vmBackupDir = getVmBackupDir(this._backup.vm.uuid)
+      await handler.mktree(vmBackupDir)
+      this._lock = await handler.lock(vmBackupDir)
+    }
+
+    async afterBackup() {
+      await this._cleanVm({ remove: true, merge: true })
+      await this._lock.dispose()
+    }
+  }

@xen-orchestra/backups/writers/_MixinReplicationWriter.js

@@ -0,0 +1,8 @@
+exports.MixinReplicationWriter = (BaseClass = Object) =>
+  class MixinReplicationWriter extends BaseClass {
+    constructor({ sr, ...rest }) {
+      super(rest)
+
+      this._sr = sr
+    }
+  }

@xen-orchestra/cr-seed-cli/index.js

@@ -1,7 +1,7 @@
 #!/usr/bin/env node
 
-const defer = require('golike-defer').default
 const { Ref, Xapi } = require('xen-api')
+const { defer } = require('golike-defer')
 
 const pkg = require('./package.json')
 

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

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

@xen-orchestra/cron/.eslintrc.js

@@ -0,0 +1 @@
+../../scripts/babel-eslintrc.js

@xen-orchestra/cron/package.json

@@ -28,9 +28,6 @@
   },
   "preferGlobal": false,
   "main": "dist/",
-  "files": [
-    "dist/"
-  ],
   "browserslist": [
     ">2%"
   ],
@@ -45,7 +42,6 @@
     "@babel/cli": "^7.0.0",
     "@babel/core": "^7.0.0",
     "@babel/preset-env": "^7.0.0",
-    "@babel/preset-flow": "^7.0.0",
     "cross-env": "^7.0.2",
     "rimraf": "^3.0.0"
   },

@xen-orchestra/defined/index.js

@@ -11,7 +11,7 @@
 //   process.env.http_proxy
 // ])
 // ```
-export default function defined() {
+function defined() {
   let args = arguments
   let n = args.length
   if (n === 1) {
@@ -29,6 +29,7 @@ export default function defined() {
     }
   }
 }
+module.exports = exports = defined
 
 // Usage:
 //
@@ -39,7 +40,7 @@ export default function defined() {
 // const getFriendName = _ => _.friends[0].name
 // const friendName = get(getFriendName, props.user)
 // ```
-export const get = (accessor, arg) => {
+function get(accessor, arg) {
   try {
     return accessor(arg)
   } catch (error) {
@@ -49,6 +50,7 @@ export const get = (accessor, arg) => {
     }
   }
 }
+exports.get = get
 
 // Usage:
 //
@@ -58,4 +60,6 @@ export const get = (accessor, arg) => {
 //   _ => new ProxyAgent(_)
 // )
 // ```
-export const ifDef = (value, thenFn) => (value !== undefined ? thenFn(value) : value)
+exports.ifDef = function ifDef(value, thenFn) {
+  return value !== undefined ? thenFn(value) : value
+}

@xen-orchestra/defined/package.json

@@ -16,30 +16,13 @@
     "url": "https://vates.fr"
   },
   "preferGlobal": false,
-  "main": "dist/",
-  "files": [
-    "dist/"
-  ],
   "browserslist": [
     ">2%"
   ],
   "engines": {
     "node": ">=6"
   },
-  "devDependencies": {
-    "@babel/cli": "^7.0.0",
-    "@babel/core": "^7.0.0",
-    "@babel/preset-env": "^7.0.0",
-    "cross-env": "^7.0.2",
-    "rimraf": "^3.0.0"
-  },
   "scripts": {
-    "build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
-    "clean": "rimraf dist/",
-    "dev": "cross-env NODE_ENV=development babel --watch --source-maps --out-dir=dist/ src/",
-    "prebuild": "yarn run clean",
-    "predev": "yarn run prebuild",
-    "prepublishOnly": "yarn run build",
     "postversion": "npm publish"
   }
 }

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

@@ -1,4 +1,4 @@
-export default function emitAsync(event) {
+module.exports = function emitAsync(event) {
   let opts
   let i = 1
 

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

@@ -16,30 +16,13 @@
     "url": "https://vates.fr"
   },
   "preferGlobal": false,
-  "main": "dist/",
-  "files": [
-    "dist/"
-  ],
   "browserslist": [
     ">2%"
   ],
   "engines": {
     "node": ">=6"
   },
-  "devDependencies": {
-    "@babel/cli": "^7.0.0",
-    "@babel/core": "^7.0.0",
-    "@babel/preset-env": "^7.0.0",
-    "cross-env": "^7.0.2",
-    "rimraf": "^3.0.0"
-  },
   "scripts": {
-    "build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
-    "clean": "rimraf dist/",
-    "dev": "cross-env NODE_ENV=development babel --watch --source-maps --out-dir=dist/ src/",
-    "prebuild": "yarn run clean",
-    "predev": "yarn run prebuild",
-    "prepublishOnly": "yarn run build",
     "postversion": "npm publish"
   }
 }

@xen-orchestra/fs/.eslintrc.js

@@ -0,0 +1 @@
+../../scripts/babel-eslintrc.js

@xen-orchestra/fs/package.json

@@ -1,7 +1,7 @@
 {
   "private": false,
   "name": "@xen-orchestra/fs",
-  "version": "0.14.0",
+  "version": "0.17.0",
   "license": "AGPL-3.0-or-later",
   "description": "The File System for Xen Orchestra backups.",
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/fs",
@@ -13,30 +13,27 @@
   },
   "preferGlobal": true,
   "main": "dist/",
-  "files": [
-    "dist/"
-  ],
   "engines": {
-    "node": ">=8.10"
+    "node": ">=14"
   },
   "dependencies": {
     "@marsaud/smb2": "^0.17.2",
     "@sindresorhus/df": "^3.1.1",
     "@sullux/aws-sdk": "^1.0.5",
+    "@vates/coalesce-calls": "^0.1.0",
     "@xen-orchestra/async-map": "^0.1.2",
     "aws-sdk": "^2.686.0",
     "decorator-synchronized": "^0.5.0",
     "execa": "^5.0.0",
     "fs-extra": "^9.0.0",
     "get-stream": "^6.0.0",
-    "limit-concurrency-decorator": "^0.4.0",
+    "limit-concurrency-decorator": "^0.5.0",
     "lodash": "^4.17.4",
-    "promise-toolbox": "^0.18.0",
+    "promise-toolbox": "^0.19.2",
     "proper-lockfile": "^4.1.2",
     "readable-stream": "^3.0.6",
     "through2": "^4.0.2",
-    "tmp": "^0.2.1",
-    "xo-remote-parser": "^0.6.0"
+    "xo-remote-parser": "^0.7.0"
   },
   "devDependencies": {
     "@babel/cli": "^7.0.0",
@@ -45,12 +42,10 @@
     "@babel/plugin-proposal-function-bind": "^7.0.0",
     "@babel/plugin-proposal-nullish-coalescing-operator": "^7.4.4",
     "@babel/preset-env": "^7.0.0",
-    "@babel/preset-flow": "^7.0.0",
     "async-iterator-to-stream": "^1.1.0",
     "babel-plugin-lodash": "^3.3.2",
     "cross-env": "^7.0.2",
     "dotenv": "^8.0.0",
-    "index-modules": "^0.3.0",
     "rimraf": "^3.0.0"
   },
   "scripts": {

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

@@ -1,33 +1,21 @@
-// @flow
-
-// $FlowFixMe
-import getStream from 'get-stream'
-
 import asyncMapSettled from '@xen-orchestra/async-map/legacy'
-import limit from 'limit-concurrency-decorator'
+import getStream from 'get-stream'
 import path, { basename } from 'path'
 import synchronized from 'decorator-synchronized'
+import { coalesceCalls } from '@vates/coalesce-calls'
 import { fromCallback, fromEvent, ignoreErrors, timeout } from 'promise-toolbox'
+import { limitConcurrency } from 'limit-concurrency-decorator'
 import { parse } from 'xo-remote-parser'
+import { pipeline } from 'stream'
 import { randomBytes } from 'crypto'
-import { type Readable, type Writable } from 'stream'
 
 import normalizePath from './_normalizePath'
 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
-type RemoteInfo = { used?: number, size?: number }
-
-type File = FileDescriptor | string
-
 const checksumFile = file => file + '.checksum'
-const computeRate = (hrtime: number[], size: number) => {
+const computeRate = (hrtime, size) => {
   const seconds = hrtime[0] + hrtime[1] / 1e9
   return size / seconds
 }
@@ -41,6 +29,8 @@ const ignoreEnoent = error => {
   }
 }
 
+const noop = Function.prototype
+
 class PrefixWrapper {
   constructor(handler, prefix) {
     this._prefix = prefix
@@ -73,11 +63,7 @@ class PrefixWrapper {
 }
 
 export default class RemoteHandlerAbstract {
-  _highWaterMark: number
-  _remote: Object
-  _timeout: number
-
-  constructor(remote: any, options: Object = {}) {
+  constructor(remote, options = {}) {
     if (remote.url === 'test://') {
       this._remote = remote
     } else {
@@ -88,7 +74,7 @@ export default class RemoteHandlerAbstract {
     }
     ;({ highWaterMark: this._highWaterMark, timeout: this._timeout = DEFAULT_TIMEOUT } = options)
 
-    const sharedLimit = limit(options.maxParallelOperations ?? DEFAULT_MAX_PARALLEL_OPERATIONS)
+    const sharedLimit = limitConcurrency(options.maxParallelOperations ?? DEFAULT_MAX_PARALLEL_OPERATIONS)
     this.closeFile = sharedLimit(this.closeFile)
     this.getInfo = sharedLimit(this.getInfo)
     this.getSize = sharedLimit(this.getSize)
@@ -104,25 +90,28 @@ export default class RemoteHandlerAbstract {
     this.unlink = sharedLimit(this.unlink)
     this.write = sharedLimit(this.write)
     this.writeFile = sharedLimit(this.writeFile)
+
+    this._forget = coalesceCalls(this._forget)
+    this._sync = coalesceCalls(this._sync)
   }
 
   // Public members
 
-  get type(): string {
+  get type() {
     throw new Error('Not implemented')
   }
 
-  addPrefix(prefix: string) {
+  addPrefix(prefix) {
     prefix = normalizePath(prefix)
     return prefix === '/' ? this : new PrefixWrapper(this, prefix)
   }
 
-  async closeFile(fd: FileDescriptor): Promise<void> {
+  async closeFile(fd) {
     await this.__closeFile(fd)
   }
 
   // TODO: remove method
-  async createOutputStream(file: File, { checksum = false, dirMode, ...options }: Object = {}): Promise<LaxWritable> {
+  async createOutputStream(file, { checksum = false, dirMode, ...options } = {}) {
     if (typeof file === 'string') {
       file = normalizePath(file)
     }
@@ -149,7 +138,6 @@ export default class RemoteHandlerAbstract {
     stream.on('error', forwardError)
     checksumStream.pipe(stream)
 
-    // $FlowFixMe
     checksumStream.checksumWritten = checksumStream.checksum
       .then(value => this._outputFile(checksumFile(path), value, { flags: 'wx' }))
       .catch(forwardError)
@@ -157,10 +145,7 @@ export default class RemoteHandlerAbstract {
     return checksumStream
   }
 
-  createReadStream(
-    file: File,
-    { checksum = false, ignoreMissingChecksum = false, ...options }: Object = {}
-  ): Promise<LaxReadable> {
+  createReadStream(file, { checksum = false, ignoreMissingChecksum = false, ...options } = {}) {
     if (typeof file === 'string') {
       file = normalizePath(file)
     }
@@ -197,7 +182,7 @@ export default class RemoteHandlerAbstract {
       checksum =>
         streamP.then(stream => {
           const { length } = stream
-          stream = (validChecksumOfReadStream(stream, String(checksum).trim()): LaxReadable)
+          stream = validChecksumOfReadStream(stream, String(checksum).trim())
           stream.length = length
 
           return stream
@@ -211,16 +196,31 @@ export default class RemoteHandlerAbstract {
     )
   }
 
-  // write a stream to a file using a temporary file
-  async outputStream(
-    path: string,
-    input: Readable | Promise<Readable>,
-    { checksum = true, dirMode }: { checksum?: boolean, dirMode?: number } = {}
-  ): Promise<void> {
-    return this._outputStream(normalizePath(path), await input, {
-      checksum,
+  /**
+   * write a stream to a file using a temporary file
+   *
+   * @param {string} path
+   * @param {ReadableStream} input
+   * @param {object} [options]
+   * @param {boolean} [options.checksum]
+   * @param {number} [options.dirMode]
+   * @param {(this: RemoteHandlerAbstract, path: string) => Promise<undefined>} [options.validator] Function that will be called before the data is commited to the remote, if it fails, file should not exist
+   */
+  async outputStream(path, input, { checksum = true, dirMode, validator } = {}) {
+    path = normalizePath(path)
+    let checksumStream
+    if (checksum) {
+      checksumStream = createChecksumStream()
+      pipeline(input, checksumStream, noop)
+      input = checksumStream
+    }
+    await this._outputStream(path, input, {
       dirMode,
+      validator,
     })
+    if (checksum) {
+      await this._outputFile(checksumFile(path), await checksumStream.checksum, { dirMode, flags: 'wx' })
+    }
   }
 
   // Free the resources possibly dedicated to put the remote at work, when it
@@ -230,73 +230,73 @@ export default class RemoteHandlerAbstract {
   // as mount), forgetting them might breaking other processes using the same
   // remote.
   @synchronized()
-  async forget(): Promise<void> {
+  async forget() {
     await this._forget()
   }
 
-  async getInfo(): Promise<RemoteInfo> {
+  async getInfo() {
     return timeout.call(this._getInfo(), this._timeout)
   }
 
-  async getSize(file: File): Promise<number> {
+  async getSize(file) {
     return timeout.call(this._getSize(typeof file === 'string' ? normalizePath(file) : file), this._timeout)
   }
 
-  async list(
-    dir: string,
-    { filter, prependDir = false }: { filter?: (name: string) => boolean, prependDir?: boolean } = {}
-  ): Promise<string[]> {
-    const virtualDir = normalizePath(dir)
-    dir = normalizePath(dir)
+  async list(dir, { filter, ignoreMissing = false, prependDir = false } = {}) {
+    try {
+      const virtualDir = normalizePath(dir)
+      dir = normalizePath(dir)
 
-    let entries = await timeout.call(this._list(dir), this._timeout)
-    if (filter !== undefined) {
-      entries = entries.filter(filter)
+      let entries = await timeout.call(this._list(dir), this._timeout)
+      if (filter !== undefined) {
+        entries = entries.filter(filter)
+      }
+
+      if (prependDir) {
+        entries.forEach((entry, i) => {
+          entries[i] = virtualDir + '/' + entry
+        })
+      }
+
+      return entries
+    } catch (error) {
+      if (ignoreMissing && error?.code === 'ENOENT') {
+        return []
+      }
+      throw error
     }
-
-    if (prependDir) {
-      entries.forEach((entry, i) => {
-        entries[i] = virtualDir + '/' + entry
-      })
-    }
-
-    return entries
   }
 
-  async lock(path: string): Promise<Disposable> {
+  async lock(path) {
     path = normalizePath(path)
     return { dispose: await this._lock(path) }
   }
 
-  async mkdir(dir: string, { mode }: { mode?: number } = {}): Promise<void> {
+  async mkdir(dir, { mode } = {}) {
     await this.__mkdir(normalizePath(dir), { mode })
   }
 
-  async mktree(dir: string, { mode }: { mode?: number } = {}): Promise<void> {
+  async mktree(dir, { mode } = {}) {
     await this._mktree(normalizePath(dir), { mode })
   }
 
-  openFile(path: string, flags: string): Promise<FileDescriptor> {
+  openFile(path, flags) {
     return this.__openFile(path, flags)
   }
 
-  async outputFile(
-    file: string,
-    data: Data,
-    { dirMode, flags = 'wx' }: { dirMode?: number, flags?: string } = {}
-  ): Promise<void> {
+  async outputFile(file, data, { dirMode, flags = 'wx' } = {}) {
     await this._outputFile(normalizePath(file), data, { dirMode, flags })
   }
 
-  async read(file: File, buffer: Buffer, position?: number): Promise<{| bytesRead: number, buffer: Buffer |}> {
+  async read(file, buffer, position) {
     return this._read(typeof file === 'string' ? normalizePath(file) : file, buffer, position)
   }
 
-  async readFile(file: string, { flags = 'r' }: { flags?: string } = {}): Promise<Buffer> {
+  async readFile(file, { flags = 'r' } = {}) {
     return this._readFile(normalizePath(file), { flags })
   }
 
-  async rename(oldPath: string, newPath: string, { checksum = false }: Object = {}) {
+  async rename(oldPath, newPath, { checksum = false } = {}) {
     oldPath = normalizePath(oldPath)
     newPath = normalizePath(newPath)
 
@@ -307,11 +307,11 @@ export default class RemoteHandlerAbstract {
     return p
   }
 
-  async rmdir(dir: string): Promise<void> {
+  async rmdir(dir) {
     await timeout.call(this._rmdir(normalizePath(dir)).catch(ignoreEnoent), this._timeout)
   }
 
-  async rmtree(dir: string): Promise<void> {
+  async rmtree(dir) {
     await this._rmtree(normalizePath(dir))
   }
 
@@ -320,11 +320,11 @@ export default class RemoteHandlerAbstract {
   //
   // This method MUST ALWAYS be called before using the handler.
   @synchronized()
-  async sync(): Promise<void> {
+  async sync() {
     await this._sync()
   }
 
-  async test(): Promise<Object> {
+  async test() {
     const SIZE = 1024 * 1024 * 10
     const testFileName = normalizePath(`${Date.now()}.test`)
     const data = await fromCallback(randomBytes, SIZE)
@@ -359,11 +359,11 @@ export default class RemoteHandlerAbstract {
     }
   }
 
-  async truncate(file: string, len: number): Promise<void> {
+  async truncate(file, len) {
     await this._truncate(file, len)
   }
 
-  async unlink(file: string, { checksum = true }: Object = {}): Promise<void> {
+  async unlink(file, { checksum = true } = {}) {
     file = normalizePath(file)
 
     if (checksum) {
@@ -373,21 +373,21 @@ export default class RemoteHandlerAbstract {
     await this._unlink(file).catch(ignoreEnoent)
   }
 
-  async write(file: File, buffer: Buffer, position: number): Promise<{| bytesWritten: number, buffer: Buffer |}> {
+  async write(file, buffer, position) {
     await this._write(typeof file === 'string' ? normalizePath(file) : file, buffer, position)
   }
 
-  async writeFile(file: string, data: Data, { flags = 'wx' }: { flags?: string } = {}): Promise<void> {
+  async writeFile(file, data, { flags = 'wx' } = {}) {
     await this._writeFile(normalizePath(file), data, { flags })
   }
 
   // Methods that can be called by private methods to avoid parallel limit on public methods
 
-  async __closeFile(fd: FileDescriptor): Promise<void> {
+  async __closeFile(fd) {
     await timeout.call(this._closeFile(fd.fd), this._timeout)
   }
 
-  async __mkdir(dir: string, { mode }: { mode?: number } = {}): Promise<void> {
+  async __mkdir(dir, { mode } = {}) {
     try {
       await this._mkdir(dir, { mode })
     } catch (error) {
@@ -400,7 +400,7 @@ export default class RemoteHandlerAbstract {
     }
   }
 
-  async __openFile(path: string, flags: string): Promise<FileDescriptor> {
+  async __openFile(path, flags) {
     path = normalizePath(path)
 
     return {
@@ -411,11 +411,11 @@ export default class RemoteHandlerAbstract {
 
   // Methods that can be implemented by inheriting classes
 
-  async _closeFile(fd: mixed): Promise<void> {
+  async _closeFile(fd) {
     throw new Error('Not implemented')
   }
 
-  async _createOutputStream(file: File, { dirMode, ...options }: Object = {}): Promise<LaxWritable> {
+  async _createOutputStream(file, { dirMode, ...options } = {}) {
     try {
       return await this._createWriteStream(file, { ...options, highWaterMark: this._highWaterMark })
     } catch (error) {
@@ -428,40 +428,40 @@ export default class RemoteHandlerAbstract {
     return this._createOutputStream(file, options)
   }
 
-  async _createReadStream(file: File, options?: Object): Promise<LaxReadable> {
+  async _createReadStream(file, options) {
     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> {
+  async _createWriteStream(file, options) {
     throw new Error('Not implemented')
   }
 
   // called to finalize the remote
-  async _forget(): Promise<void> {}
+  async _forget() {}
 
-  async _getInfo(): Promise<Object> {
+  async _getInfo() {
     return {}
   }
 
-  async _lock(path: string): Promise<Function> {
+  async _lock(path) {
     return () => Promise.resolve()
   }
 
-  async _getSize(file: File): Promise<number> {
+  async _getSize(file) {
     throw new Error('Not implemented')
   }
 
-  async _list(dir: string): Promise<string[]> {
+  async _list(dir) {
     throw new Error('Not implemented')
   }
 
-  async _mkdir(dir: string): Promise<void> {
+  async _mkdir(dir) {
     throw new Error('Not implemented')
   }
 
-  async _mktree(dir: string, { mode }: { mode?: number } = {}): Promise<void> {
+  async _mktree(dir, { mode } = {}) {
     try {
       return await this.__mkdir(dir, { mode })
     } catch (error) {
@@ -474,11 +474,11 @@ export default class RemoteHandlerAbstract {
     return this._mktree(dir, { mode })
   }
 
-  async _openFile(path: string, flags: string): Promise<mixed> {
+  async _openFile(path, flags) {
     throw new Error('Not implemented')
   }
 
-  async _outputFile(file: string, data: Data, { dirMode, flags }: { dirMode?: number, flags?: string }): Promise<void> {
+  async _outputFile(file, data, { dirMode, flags }) {
     try {
       return await this._writeFile(file, data, { flags })
     } catch (error) {
@@ -491,42 +491,40 @@ export default class RemoteHandlerAbstract {
     return this._outputFile(file, data, { flags })
   }
 
-  async _outputStream(path: string, input: Readable, { checksum, dirMode }: { checksum?: boolean, dirMode?: number }) {
+  async _outputStream(path, input, { dirMode, validator }) {
     const tmpPath = `${dirname(path)}/.${basename(path)}`
     const output = await this.createOutputStream(tmpPath, {
-      checksum,
       dirMode,
     })
     try {
-      input.pipe(output)
-      await fromEvent(output, 'finish')
-      await output.checksumWritten
-      // $FlowFixMe
-      await input.task
-      await this.rename(tmpPath, path, { checksum })
+      await fromCallback(pipeline, input, output)
+      if (validator !== undefined) {
+        await validator.call(this, tmpPath)
+      }
+      await this.rename(tmpPath, path)
     } catch (error) {
-      await this.unlink(tmpPath, { checksum })
+      await this.unlink(tmpPath)
       throw error
     }
   }
 
-  _read(file: File, buffer: Buffer, position?: number): Promise<{| bytesRead: number, buffer: Buffer |}> {
+  _read(file, buffer, position) {
     throw new Error('Not implemented')
   }
 
-  _readFile(file: string, options?: Object): Promise<Buffer> {
+  _readFile(file, options) {
     return this._createReadStream(file, { ...options, highWaterMark: this._highWaterMark }).then(getStream.buffer)
   }
 
-  async _rename(oldPath: string, newPath: string) {
+  async _rename(oldPath, newPath) {
     throw new Error('Not implemented')
   }
 
-  async _rmdir(dir: string) {
+  async _rmdir(dir) {
     throw new Error('Not implemented')
   }
 
-  async _rmtree(dir: string) {
+  async _rmtree(dir) {
     try {
       return await this._rmdir(dir)
     } catch (error) {
@@ -548,13 +546,13 @@ export default class RemoteHandlerAbstract {
   }
 
   // called to initialize the remote
-  async _sync(): Promise<void> {}
+  async _sync() {}
 
-  async _unlink(file: string): Promise<void> {
+  async _unlink(file) {
     throw new Error('Not implemented')
   }
 
-  async _write(file: File, buffer: Buffer, position: number): Promise<void> {
+  async _write(file, buffer, position) {
     const isPath = typeof file === 'string'
     if (isPath) {
       file = await this.__openFile(file, 'r+')
@@ -568,11 +566,11 @@ export default class RemoteHandlerAbstract {
     }
   }
 
-  async _writeFd(fd: FileDescriptor, buffer: Buffer, position: number): Promise<void> {
+  async _writeFd(fd, buffer, position) {
     throw new Error('Not implemented')
   }
 
-  async _writeFile(file: string, data: Data, options: { flags?: string }): Promise<void> {
+  async _writeFile(file, data, options) {
     throw new Error('Not implemented')
   }
 }

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

@@ -1,10 +1,7 @@
-// @flow
-
 import through2 from 'through2'
 import { createHash } from 'crypto'
 import { defer, fromEvent } from 'promise-toolbox'
 import { invert } from 'lodash'
-import { type Readable, type Transform } from 'stream'
 
 // Format: $<algorithm>$<salt>$<encrypted>
 //
@@ -27,7 +24,7 @@ const ID_TO_ALGORITHM = invert(ALGORITHM_TO_ID)
 //    const checksumStream = source.pipe(createChecksumStream())
 //    checksumStream.resume() // make the data flow without an output
 //    console.log(await checksumStream.checksum)
-export const createChecksumStream = (algorithm: string = 'md5'): Transform & { checksum: Promise<string> } => {
+export const createChecksumStream = (algorithm = 'md5') => {
   const algorithmId = ALGORITHM_TO_ID[algorithm]
 
   if (!algorithmId) {
@@ -54,10 +51,7 @@ export const createChecksumStream = (algorithm: string = 'md5'): Transform & { c
 // Check if the checksum of a readable stream is equals to an expected checksum.
 // The given stream is wrapped in a stream which emits an error event
 // if the computed checksum is not equals to the expected checksum.
-export const validChecksumOfReadStream = (
-  stream: Readable,
-  expectedChecksum: string
-): Readable & { checksumVerified: Promise<void> } => {
+export const validChecksumOfReadStream = (stream, expectedChecksum) => {
   const algorithmId = expectedChecksum.slice(1, expectedChecksum.indexOf('$', 1))
 
   if (!algorithmId) {
@@ -66,7 +60,7 @@ export const validChecksumOfReadStream = (
 
   const hash = createHash(ID_TO_ALGORITHM[algorithmId])
 
-  const wrapper: any = stream.pipe(
+  const wrapper = stream.pipe(
     through2(
       { highWaterMark: 0 },
       (chunk, enc, callback) => {

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

@@ -133,6 +133,14 @@ handlers.forEach(url => {
         await handler.outputFile('dir/file', '')
         expect(await handler.list('dir', { prependDir: true })).toEqual(['/dir/file'])
       })
+
+      it('throws ENOENT if no such directory', async () => {
+        expect((await rejectionOf(handler.list('dir'))).code).toBe('ENOENT')
+      })
+
+      it('can returns empty for missing directory', async () => {
+        expect(await handler.list('dir', { ignoreMissing: true })).toEqual([])
+      })
     })
 
     describe('#mkdir()', () => {

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

@@ -1,16 +1,12 @@
-// @flow
 import execa from 'execa'
+import { parse } from 'xo-remote-parser'
 
-import type RemoteHandler from './abstract'
 import RemoteHandlerLocal from './local'
 import RemoteHandlerNfs from './nfs'
 import RemoteHandlerS3 from './s3'
 import RemoteHandlerSmb from './smb'
 import RemoteHandlerSmbMount from './smb-mount'
 
-export type { default as RemoteHandler } from './abstract'
-export type Remote = { url: string }
-
 const HANDLERS = {
   file: RemoteHandlerLocal,
   nfs: RemoteHandlerNfs,
@@ -24,11 +20,8 @@ try {
   HANDLERS.smb = RemoteHandlerSmb
 }
 
-export const getHandler = (remote: Remote, ...rest: any): RemoteHandler => {
-  // FIXME: should be done in xo-remote-parser.
-  const type = remote.url.split('://')[0]
-
-  const Handler = HANDLERS[type]
+export const getHandler = (remote, ...rest) => {
+  const Handler = HANDLERS[parse(remote.url).type]
   if (!Handler) {
     throw new Error('Unhandled remote type')
   }

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

@@ -84,7 +84,7 @@ export default class LocalHandler extends RemoteHandlerAbstract {
   }
 
   _lock(path) {
-    return lockfile.lock(path)
+    return lockfile.lock(this._getFilePath(path))
   }
 
   _mkdir(dir, { mode }) {

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

@@ -8,7 +8,6 @@ export default class NfsHandler extends MountHandler {
     super(remote, opts, {
       type: 'nfs',
       device: `${host}${port !== undefined ? ':' + port : ''}:${path}`,
-      defaultOptions: 'vers=3',
     })
   }
 

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

@@ -1,9 +1,9 @@
 import aws from '@sullux/aws-sdk'
 import assert from 'assert'
+import http from 'http'
 import { parse } from 'xo-remote-parser'
 
 import RemoteHandlerAbstract from './abstract'
-import { createChecksumStream } from './checksum'
 
 // endpoints https://docs.aws.amazon.com/general/latest/gr/s3.html
 
@@ -16,9 +16,8 @@ const IDEAL_FRAGMENT_SIZE = Math.ceil(MAX_OBJECT_SIZE / MAX_PARTS_COUNT) // the
 export default class S3Handler extends RemoteHandlerAbstract {
   constructor(remote, _opts) {
     super(remote)
-    const { host, path, username, password } = parse(remote.url)
-    // https://www.zenko.io/blog/first-things-first-getting-started-scality-s3-server/
-    this._s3 = aws({
+    const { host, path, username, password, protocol, region } = parse(remote.url)
+    const params = {
       accessKeyId: username,
       apiVersion: '2006-03-01',
       endpoint: host,
@@ -28,7 +27,16 @@ export default class S3Handler extends RemoteHandlerAbstract {
       httpOptions: {
         timeout: 600000,
       },
-    }).s3
+    }
+    if (protocol === 'http') {
+      params.httpOptions.agent = new http.Agent()
+      params.sslEnabled = false
+    }
+    if (region !== undefined) {
+      params.region = region
+    }
+
+    this._s3 = aws(params).s3
 
     const splitPath = path.split('/').filter(s => s.length)
     this._bucket = splitPath.shift()
@@ -43,46 +51,69 @@ export default class S3Handler extends RemoteHandlerAbstract {
     return { Bucket: this._bucket, Key: this._dir + file }
   }
 
-  async _outputStream(path, input, { checksum }) {
-    let inputStream = input
-    if (checksum) {
-      const checksumStream = createChecksumStream()
-      const forwardError = error => {
-        checksumStream.emit('error', error)
+  async _isNotEmptyDir(path) {
+    const result = await this._s3.listObjectsV2({
+      Bucket: this._bucket,
+      MaxKeys: 1,
+      Prefix: this._dir + path + '/',
+    })
+    return result.Contents.length !== 0
+  }
+
+  async _isFile(path) {
+    try {
+      await this._s3.headObject(this._createParams(path))
+      return true
+    } catch (error) {
+      if (error.code === 'NotFound') {
+        return false
       }
-      input.pipe(checksumStream)
-      input.on('error', forwardError)
-      inputStream = checksumStream
+      throw error
     }
+  }
+
+  async _outputStream(path, input, { validator }) {
     await this._s3.upload(
       {
         ...this._createParams(path),
-        Body: inputStream,
+        Body: input,
       },
       { partSize: IDEAL_FRAGMENT_SIZE, queueSize: 1 }
     )
-    if (checksum) {
-      const checksum = await inputStream.checksum
-      const params = {
-        ...this._createParams(path + '.checksum'),
-        Body: checksum,
+    if (validator !== undefined) {
+      try {
+        await validator.call(this, path)
+      } catch (error) {
+        await this.unlink(path)
+        throw error
       }
-      await this._s3.upload(params)
     }
-    await input.task
   }
 
   async _writeFile(file, data, options) {
     return this._s3.putObject({ ...this._createParams(file), Body: data })
   }
 
-  async _createReadStream(file, options) {
+  async _createReadStream(path, options) {
+    if (!(await this._isFile(path))) {
+      const error = new Error(`ENOENT: no such file '${path}'`)
+      error.code = 'ENOENT'
+      error.path = path
+      throw error
+    }
+
     // https://github.com/Sullux/aws-sdk/issues/11
-    return this._s3.getObject.raw(this._createParams(file)).createReadStream()
+    return this._s3.getObject.raw(this._createParams(path)).createReadStream()
   }
 
-  async _unlink(file) {
-    return this._s3.deleteObject(this._createParams(file))
+  async _unlink(path) {
+    await this._s3.deleteObject(this._createParams(path))
+    if (await this._isNotEmptyDir(path)) {
+      const error = new Error(`EISDIR: illegal operation on a directory, unlink '${path}'`)
+      error.code = 'EISDIR'
+      error.path = path
+      throw error
+    }
   }
 
   async _list(dir) {
@@ -106,6 +137,16 @@ export default class S3Handler extends RemoteHandlerAbstract {
     return [...uniq]
   }
 
+  async _mkdir(path) {
+    if (await this._isFile(path)) {
+      const error = new Error(`ENOTDIR: file already exists, mkdir '${path}'`)
+      error.code = 'ENOTDIR'
+      error.path = path
+      throw error
+    }
+    // nothing to do, directories do not exist, they are part of the files' path
+  }
+
   async _rename(oldPath, newPath) {
     const size = await this._getSize(oldPath)
     const multipartParams = await this._s3.createMultipartUpload({ ...this._createParams(newPath) })
@@ -147,6 +188,17 @@ export default class S3Handler extends RemoteHandlerAbstract {
     return { bytesRead: result.Body.length, buffer }
   }
 
+  async _rmdir(path) {
+    if (await this._isNotEmptyDir(path)) {
+      const error = new Error(`ENOTEMPTY: directory not empty, rmdir '${path}`)
+      error.code = 'ENOTEMPTY'
+      error.path = path
+      throw error
+    }
+
+    // nothing to do, directories do not exist, they are part of the files' path
+  }
+
   async _write(file, buffer, position) {
     if (typeof file !== 'string') {
       file = file.fd

@xen-orchestra/log/.babelrc.js

@@ -1 +0,0 @@
-module.exports = require('../../@xen-orchestra/babel-config')(require('./package.json'))

@xen-orchestra/log/configure.js

@@ -1 +1,107 @@
-module.exports = require('./dist/configure')
+const createConsoleTransport = require('./transports/console')
+const { LEVELS, resolve } = require('./levels')
+const { compileGlobPattern } = require('./utils')
+
+// ===================================================================
+
+const createTransport = config => {
+  if (typeof config === 'function') {
+    return config
+  }
+
+  if (Array.isArray(config)) {
+    const transports = config.map(createTransport)
+    const { length } = transports
+    return function () {
+      for (let i = 0; i < length; ++i) {
+        transports[i].apply(this, arguments)
+      }
+    }
+  }
+
+  let { filter } = config
+  let transport = createTransport(config.transport)
+  const level = resolve(config.level)
+
+  if (filter !== undefined) {
+    if (typeof filter === 'string') {
+      const re = compileGlobPattern(filter)
+      filter = log => re.test(log.namespace)
+    }
+
+    const orig = transport
+    transport = function (log) {
+      if ((level !== undefined && log.level >= level) || filter(log)) {
+        return orig.apply(this, arguments)
+      }
+    }
+  } else if (level !== undefined) {
+    const orig = transport
+    transport = function (log) {
+      if (log.level >= level) {
+        return orig.apply(this, arguments)
+      }
+    }
+  }
+
+  return transport
+}
+
+const symbol = typeof Symbol !== 'undefined' ? Symbol.for('@xen-orchestra/log') : '@@@xen-orchestra/log'
+
+const { env } = process
+global[symbol] = createTransport({
+  // display warnings or above, and all that are enabled via DEBUG or
+  // NODE_DEBUG env
+  filter: [env.DEBUG, env.NODE_DEBUG].filter(Boolean).join(','),
+  level: resolve(env.LOG_LEVEL, LEVELS.INFO),
+
+  transport: createConsoleTransport(),
+})
+
+const configure = config => {
+  global[symbol] = createTransport(config)
+}
+exports.configure = configure
+
+// -------------------------------------------------------------------
+
+const catchGlobalErrors = logger => {
+  // patch process
+  const onUncaughtException = error => {
+    logger.error('uncaught exception', { error })
+  }
+  const onUnhandledRejection = error => {
+    logger.warn('possibly unhandled rejection', { error })
+  }
+  const onWarning = error => {
+    logger.warn('Node warning', { error })
+  }
+  process.on('uncaughtException', onUncaughtException)
+  process.on('unhandledRejection', onUnhandledRejection)
+  process.on('warning', onWarning)
+
+  // patch EventEmitter
+  const EventEmitter = require('events')
+  const { prototype } = EventEmitter
+  const { emit } = prototype
+  function patchedEmit(event, error) {
+    if (event === 'error' && this.listenerCount(event) === 0) {
+      logger.error('unhandled error event', { error })
+      return false
+    }
+    return emit.apply(this, arguments)
+  }
+  prototype.emit = patchedEmit
+
+  return () => {
+    process.removeListener('uncaughtException', onUncaughtException)
+    process.removeListener('unhandledRejection', onUnhandledRejection)
+    process.removeListener('warning', onWarning)
+
+    if (prototype.emit === patchedEmit) {
+      prototype.emit = emit
+    }
+  }
+}
+exports.catchGlobalErrors = catchGlobalErrors

@xen-orchestra/log/index.js

@@ -1,5 +1,5 @@
-import createTransport from './transports/console'
-import LEVELS, { resolve } from './levels'
+const createTransport = require('./transports/console')
+const { LEVELS, resolve } = require('./levels')
 
 const symbol = typeof Symbol !== 'undefined' ? Symbol.for('@xen-orchestra/log') : '@@@xen-orchestra/log'
 if (!(symbol in global)) {
@@ -68,5 +68,7 @@ prototype.wrap = function (message, fn) {
   }
 }
 
-export const createLogger = namespace => new Logger(namespace)
-export { createLogger as default }
+const createLogger = namespace => new Logger(namespace)
+
+module.exports = exports = createLogger
+exports.createLogger = createLogger

@xen-orchestra/log/levels.js

@@ -1,5 +1,5 @@
 const LEVELS = Object.create(null)
-export { LEVELS as default }
+exports.LEVELS = LEVELS
 
 // https://github.com/trentm/node-bunyan#levels
 LEVELS.FATAL = 60 // service/app is going down
@@ -8,7 +8,8 @@ LEVELS.WARN = 40 // something went wrong but it's not fatal
 LEVELS.INFO = 30 // detail on unusual but normal operation
 LEVELS.DEBUG = 20
 
-export const NAMES = Object.create(null)
+const NAMES = Object.create(null)
+exports.NAMES = NAMES
 for (const name in LEVELS) {
   NAMES[LEVELS[name]] = name
 }
@@ -16,7 +17,7 @@ for (const name in LEVELS) {
 // resolves to the number representation of a level
 //
 // returns `defaultLevel` if invalid
-export const resolve = (level, defaultLevel) => {
+const resolve = (level, defaultLevel) => {
   const type = typeof level
   if (type === 'number') {
     if (level in NAMES) {
@@ -30,6 +31,7 @@ export const resolve = (level, defaultLevel) => {
   }
   return defaultLevel
 }
+exports.resolve = resolve
 
 Object.freeze(LEVELS)
 Object.freeze(NAMES)

@xen-orchestra/log/levels.spec.js

@@ -1,8 +1,8 @@
 /* eslint-env jest */
 
-import { forEach, isInteger } from 'lodash'
+const { forEach, isInteger } = require('lodash')
 
-import LEVELS, { NAMES, resolve } from './levels'
+const { LEVELS, NAMES, resolve } = require('./levels')
 
 describe('LEVELS', () => {
   it('maps level names to their integer values', () => {

@xen-orchestra/log/package.json

@@ -16,12 +16,6 @@
     "url": "https://vates.fr"
   },
   "preferGlobal": false,
-  "main": "dist/",
-  "files": [
-    "configure.js",
-    "dist/",
-    "transports/"
-  ],
   "browserslist": [
     ">2%"
   ],
@@ -30,24 +24,9 @@
   },
   "dependencies": {
     "lodash": "^4.17.4",
-    "promise-toolbox": "^0.18.0"
-  },
-  "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",
-    "index-modules": "^0.3.0",
-    "rimraf": "^3.0.0"
+    "promise-toolbox": "^0.19.2"
   },
   "scripts": {
-    "build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
-    "clean": "rimraf dist/",
-    "dev": "cross-env NODE_ENV=development babel --watch --source-maps --out-dir=dist/ src/",
-    "prebuild": "yarn run clean",
-    "predev": "yarn run prebuild",
-    "prepublishOnly": "yarn run build",
     "postversion": "npm publish"
   }
 }

@xen-orchestra/log/src/configure.js

@@ -1,105 +0,0 @@
-import createConsoleTransport from './transports/console'
-import LEVELS, { resolve } from './levels'
-import { compileGlobPattern } from './utils'
-
-// ===================================================================
-
-const createTransport = config => {
-  if (typeof config === 'function') {
-    return config
-  }
-
-  if (Array.isArray(config)) {
-    const transports = config.map(createTransport)
-    const { length } = transports
-    return function () {
-      for (let i = 0; i < length; ++i) {
-        transports[i].apply(this, arguments)
-      }
-    }
-  }
-
-  let { filter } = config
-  let transport = createTransport(config.transport)
-  const level = resolve(config.level)
-
-  if (filter !== undefined) {
-    if (typeof filter === 'string') {
-      const re = compileGlobPattern(filter)
-      filter = log => re.test(log.namespace)
-    }
-
-    const orig = transport
-    transport = function (log) {
-      if ((level !== undefined && log.level >= level) || filter(log)) {
-        return orig.apply(this, arguments)
-      }
-    }
-  } else if (level !== undefined) {
-    const orig = transport
-    transport = function (log) {
-      if (log.level >= level) {
-        return orig.apply(this, arguments)
-      }
-    }
-  }
-
-  return transport
-}
-
-const symbol = typeof Symbol !== 'undefined' ? Symbol.for('@xen-orchestra/log') : '@@@xen-orchestra/log'
-
-const { env } = process
-global[symbol] = createTransport({
-  // display warnings or above, and all that are enabled via DEBUG or
-  // NODE_DEBUG env
-  filter: [env.DEBUG, env.NODE_DEBUG].filter(Boolean).join(','),
-  level: resolve(env.LOG_LEVEL, LEVELS.INFO),
-
-  transport: createConsoleTransport(),
-})
-
-export const configure = config => {
-  global[symbol] = createTransport(config)
-}
-
-// -------------------------------------------------------------------
-
-export const catchGlobalErrors = logger => {
-  // patch process
-  const onUncaughtException = error => {
-    logger.error('uncaught exception', { error })
-  }
-  const onUnhandledRejection = error => {
-    logger.warn('possibly unhandled rejection', { error })
-  }
-  const onWarning = error => {
-    logger.warn('Node warning', { error })
-  }
-  process.on('uncaughtException', onUncaughtException)
-  process.on('unhandledRejection', onUnhandledRejection)
-  process.on('warning', onWarning)
-
-  // patch EventEmitter
-  const EventEmitter = require('events')
-  const { prototype } = EventEmitter
-  const { emit } = prototype
-  function patchedEmit(event, error) {
-    if (event === 'error' && this.listenerCount(event) === 0) {
-      logger.error('unhandled error event', { error })
-      return false
-    }
-    return emit.apply(this, arguments)
-  }
-  prototype.emit = patchedEmit
-
-  return () => {
-    process.removeListener('uncaughtException', onUncaughtException)
-    process.removeListener('unhandledRejection', onUnhandledRejection)
-    process.removeListener('warning', onWarning)
-
-    if (prototype.emit === patchedEmit) {
-      prototype.emit = emit
-    }
-  }
-}

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

@@ -1,79 +0,0 @@
-import LEVELS, { NAMES } from '../levels'
-
-const { DEBUG, ERROR, FATAL, INFO, WARN } = LEVELS
-
-let formatLevel, formatNamespace
-if (process.stdout !== undefined && process.stdout.isTTY && process.stderr !== undefined && process.stderr.isTTY) {
-  const ansi = (style, str) => `\x1b[${style}m${str}\x1b[0m`
-
-  const LEVEL_STYLES = {
-    [DEBUG]: '2',
-    [ERROR]: '1;31',
-    [FATAL]: '1;31',
-    [INFO]: '1',
-    [WARN]: '1;33',
-  }
-  formatLevel = level => {
-    const style = LEVEL_STYLES[level]
-    const name = NAMES[level]
-    return style === undefined ? name : ansi(style, name)
-  }
-
-  const NAMESPACE_COLORS = [
-    196,
-    202,
-    208,
-    214,
-    220,
-    226,
-    190,
-    154,
-    118,
-    82,
-    46,
-    47,
-    48,
-    49,
-    50,
-    51,
-    45,
-    39,
-    33,
-    27,
-    21,
-    57,
-    93,
-    129,
-    165,
-    201,
-    200,
-    199,
-    198,
-    197,
-  ]
-  formatNamespace = namespace => {
-    // https://werxltd.com/wp/2010/05/13/javascript-implementation-of-javas-string-hashcode-method/
-    let hash = 0
-    for (let i = 0, n = namespace.length; i < n; ++i) {
-      hash = ((hash << 5) - hash + namespace.charCodeAt(i)) | 0
-    }
-    return ansi(`1;38;5;${NAMESPACE_COLORS[Math.abs(hash) % NAMESPACE_COLORS.length]}`, namespace)
-  }
-} else {
-  formatLevel = str => NAMES[str]
-  formatNamespace = str => str
-}
-
-const consoleTransport = ({ data, level, namespace, message, time }) => {
-  const fn =
-    /* eslint-disable no-console */
-    level < INFO ? console.log : level < WARN ? console.info : level < ERROR ? console.warn : console.error
-  /* eslint-enable no-console */
-
-  const args = [time.toISOString(), formatNamespace(namespace), formatLevel(level), message]
-  if (data != null) {
-    args.push(data)
-  }
-  fn.apply(console, args)
-}
-export default () => consoleTransport

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

@@ -1,64 +0,0 @@
-import fromCallback from 'promise-toolbox/fromCallback'
-import prettyFormat from 'pretty-format' // eslint-disable-line node/no-extraneous-import
-import { createTransport } from 'nodemailer' // eslint-disable-line node/no-extraneous-import
-
-import { evalTemplate, required } from '../utils'
-import { NAMES } from '../levels'
-
-export default ({
-  // transport options (https://nodemailer.com/smtp/)
-  auth,
-  authMethod,
-  host,
-  ignoreTLS,
-  port,
-  proxy,
-  requireTLS,
-  secure,
-  service,
-  tls,
-
-  // message options (https://nodemailer.com/message/)
-  bcc,
-  cc,
-  from = required('from'),
-  to = required('to'),
-  subject = '[{{level}} - {{namespace}}] {{time}} {{message}}',
-}) => {
-  const transporter = createTransport(
-    {
-      auth,
-      authMethod,
-      host,
-      ignoreTLS,
-      port,
-      proxy,
-      requireTLS,
-      secure,
-      service,
-      tls,
-
-      disableFileAccess: true,
-      disableUrlAccess: true,
-    },
-    {
-      bcc,
-      cc,
-      from,
-      to,
-    }
-  )
-
-  return log =>
-    fromCallback(cb =>
-      transporter.sendMail(
-        {
-          subject: evalTemplate(subject, key =>
-            key === 'level' ? NAMES[log.level] : key === 'time' ? log.time.toISOString() : log[key]
-          ),
-          text: prettyFormat(log.data),
-        },
-        cb
-      )
-    )
-}

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

@@ -1,7 +0,0 @@
-export default () => {
-  const memoryLogger = log => {
-    logs.push(log)
-  }
-  const logs = (memoryLogger.logs = [])
-  return memoryLogger
-}

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

@@ -1,41 +0,0 @@
-import fromCallback from 'promise-toolbox/fromCallback'
-import splitHost from 'split-host'
-import { createClient, Facility, Severity, Transport } from 'syslog-client'
-
-import LEVELS from '../levels'
-
-// https://github.com/paulgrove/node-syslog-client#syslogseverity
-const LEVEL_TO_SEVERITY = {
-  [LEVELS.FATAL]: Severity.Critical,
-  [LEVELS.ERROR]: Severity.Error,
-  [LEVELS.WARN]: Severity.Warning,
-  [LEVELS.INFO]: Severity.Informational,
-  [LEVELS.DEBUG]: Severity.Debug,
-}
-
-const facility = Facility.User
-
-export default target => {
-  const opts = {}
-  if (target !== undefined) {
-    if (target.startsWith('tcp://')) {
-      target = target.slice(6)
-      opts.transport = Transport.Tcp
-    } else if (target.startsWith('udp://')) {
-      target = target.slice(6)
-      opts.transport = Transport.Udp
-    }
-
-    ;({ host: target, port: opts.port } = splitHost(target))
-  }
-
-  const client = createClient(target, opts)
-
-  return log =>
-    fromCallback(cb =>
-      client.log(log.message, {
-        facility,
-        severity: LEVEL_TO_SEVERITY[log.level],
-      })
-    )
-}

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

@@ -1 +1,82 @@
-module.exports = require('../dist/transports/console.js')
+const { LEVELS, NAMES } = require('../levels')
+
+const { DEBUG, ERROR, FATAL, INFO, WARN } = LEVELS
+
+let formatLevel, formatNamespace
+if (process.stdout !== undefined && process.stdout.isTTY && process.stderr !== undefined && process.stderr.isTTY) {
+  const ansi = (style, str) => `\x1b[${style}m${str}\x1b[0m`
+
+  const LEVEL_STYLES = {
+    [DEBUG]: '2',
+    [ERROR]: '1;31',
+    [FATAL]: '1;31',
+    [INFO]: '1',
+    [WARN]: '1;33',
+  }
+  formatLevel = level => {
+    const style = LEVEL_STYLES[level]
+    const name = NAMES[level]
+    return style === undefined ? name : ansi(style, name)
+  }
+
+  const NAMESPACE_COLORS = [
+    196,
+    202,
+    208,
+    214,
+    220,
+    226,
+    190,
+    154,
+    118,
+    82,
+    46,
+    47,
+    48,
+    49,
+    50,
+    51,
+    45,
+    39,
+    33,
+    27,
+    21,
+    57,
+    93,
+    129,
+    165,
+    201,
+    200,
+    199,
+    198,
+    197,
+  ]
+  formatNamespace = namespace => {
+    // https://werxltd.com/wp/2010/05/13/javascript-implementation-of-javas-string-hashcode-method/
+    let hash = 0
+    for (let i = 0, n = namespace.length; i < n; ++i) {
+      hash = ((hash << 5) - hash + namespace.charCodeAt(i)) | 0
+    }
+    return ansi(`1;38;5;${NAMESPACE_COLORS[Math.abs(hash) % NAMESPACE_COLORS.length]}`, namespace)
+  }
+} else {
+  formatLevel = str => NAMES[str]
+  formatNamespace = str => str
+}
+
+const consoleTransport = ({ data, level, namespace, message, time }) => {
+  const fn =
+    /* eslint-disable no-console */
+    level < INFO ? console.log : level < WARN ? console.info : level < ERROR ? console.warn : console.error
+  /* eslint-enable no-console */
+
+  const args = [time.toISOString(), formatNamespace(namespace), formatLevel(level), message]
+  if (data != null) {
+    args.push(data)
+  }
+  fn.apply(console, args)
+}
+
+const createTransport = () => consoleTransport
+
+module.exports = exports = createTransport

@xen-orchestra/log/transports/email.js

@@ -1 +1,66 @@
-module.exports = require('../dist/transports/email.js')
+const fromCallback = require('promise-toolbox/fromCallback')
+const nodemailer = require('nodemailer') // eslint-disable-line node/no-extraneous-import
+const prettyFormat = require('pretty-format') // eslint-disable-line node/no-extraneous-import
+
+const { evalTemplate, required } = require('../utils')
+const { NAMES } = require('../levels')
+
+function createTransport({
+  // transport options (https://nodemailer.com/smtp/)
+  auth,
+  authMethod,
+  host,
+  ignoreTLS,
+  port,
+  proxy,
+  requireTLS,
+  secure,
+  service,
+  tls,
+
+  // message options (https://nodemailer.com/message/)
+  bcc,
+  cc,
+  from = required('from'),
+  to = required('to'),
+  subject = '[{{level}} - {{namespace}}] {{time}} {{message}}',
+}) {
+  const transporter = nodemailer.createTransport(
+    {
+      auth,
+      authMethod,
+      host,
+      ignoreTLS,
+      port,
+      proxy,
+      requireTLS,
+      secure,
+      service,
+      tls,
+
+      disableFileAccess: true,
+      disableUrlAccess: true,
+    },
+    {
+      bcc,
+      cc,
+      from,
+      to,
+    }
+  )
+
+  return log =>
+    fromCallback(cb =>
+      transporter.sendMail(
+        {
+          subject: evalTemplate(subject, key =>
+            key === 'level' ? NAMES[log.level] : key === 'time' ? log.time.toISOString() : log[key]
+          ),
+          text: prettyFormat(log.data),
+        },
+        cb
+      )
+    )
+}
+
+module.exports = exports = createTransport

@xen-orchestra/log/transports/memory.js

@@ -1 +1,9 @@
-module.exports = require('../dist/transports/memory.js')
+function createTransport() {
+  const memoryLogger = log => {
+    logs.push(log)
+  }
+  const logs = (memoryLogger.logs = [])
+  return memoryLogger
+}
+
+module.exports = exports = createTransport

@xen-orchestra/log/transports/syslog.js

@@ -1 +1,43 @@
-module.exports = require('../dist/transports/syslog.js')
+const fromCallback = require('promise-toolbox/fromCallback')
+const splitHost = require('split-host')
+const { createClient, Facility, Severity, Transport } = require('syslog-client')
+
+const LEVELS = require('../levels')
+
+// https://github.com/paulgrove/node-syslog-client#syslogseverity
+const LEVEL_TO_SEVERITY = {
+  [LEVELS.FATAL]: Severity.Critical,
+  [LEVELS.ERROR]: Severity.Error,
+  [LEVELS.WARN]: Severity.Warning,
+  [LEVELS.INFO]: Severity.Informational,
+  [LEVELS.DEBUG]: Severity.Debug,
+}
+
+const facility = Facility.User
+
+function createTransport(target) {
+  const opts = {}
+  if (target !== undefined) {
+    if (target.startsWith('tcp://')) {
+      target = target.slice(6)
+      opts.transport = Transport.Tcp
+    } else if (target.startsWith('udp://')) {
+      target = target.slice(6)
+      opts.transport = Transport.Udp
+    }
+
+    ;({ host: target, port: opts.port } = splitHost(target))
+  }
+
+  const client = createClient(target, opts)
+
+  return log =>
+    fromCallback(cb =>
+      client.log(log.message, {
+        facility,
+        severity: LEVEL_TO_SEVERITY[log.level],
+      })
+    )
+}
+
+module.exports = exports = createTransport

@xen-orchestra/log/utils.js

@@ -1,19 +1,20 @@
-import escapeRegExp from 'lodash/escapeRegExp'
+const escapeRegExp = require('lodash/escapeRegExp')
 
 // ===================================================================
 
 const TPL_RE = /\{\{(.+?)\}\}/g
-export const evalTemplate = (tpl, data) => {
+const evalTemplate = (tpl, data) => {
   const getData = typeof data === 'function' ? (_, key) => data(key) : (_, key) => data[key]
 
   return tpl.replace(TPL_RE, getData)
 }
+exports.evalTemplate = evalTemplate
 
 // -------------------------------------------------------------------
 
 const compileGlobPatternFragment = pattern => pattern.split('*').map(escapeRegExp).join('.*')
 
-export const compileGlobPattern = pattern => {
+const compileGlobPattern = pattern => {
   const no = []
   const yes = []
   pattern.split(/[\s,]+/).forEach(pattern => {
@@ -40,19 +41,22 @@ export const compileGlobPattern = pattern => {
 
   return new RegExp(raw.join(''))
 }
+exports.compileGlobPattern = compileGlobPattern
 
 // -------------------------------------------------------------------
 
-export const required = name => {
+const required = name => {
   throw new Error(`missing required arg ${name}`)
 }
+exports.required = required
 
 // -------------------------------------------------------------------
 
-export const serializeError = error => ({
+const serializeError = error => ({
   ...error, // Copy enumerable properties.
   code: error.code,
   message: error.message,
   name: error.name,
   stack: error.stack,
 })
+exports.serializeError = serializeError

@xen-orchestra/log/utils.spec.js

@@ -1,6 +1,6 @@
 /* eslint-env jest */
 
-import { compileGlobPattern } from './utils'
+const { compileGlobPattern } = require('./utils')
 
 describe('compileGlobPattern()', () => {
   it('works', () => {

@xen-orchestra/mixin/.babelrc.js

@@ -1 +0,0 @@
-module.exports = require('../../@xen-orchestra/babel-config')(require('./package.json'))

@xen-orchestra/mixin/index.js

@@ -0,0 +1,39 @@
+const camelCase = require('lodash/camelCase')
+
+const { defineProperties, defineProperty, keys } = Object
+const noop = Function.prototype
+
+const MIXIN_CYCLIC_DESCRIPTOR = {
+  configurable: true,
+  get() {
+    throw new Error('cyclic dependency')
+  },
+}
+
+module.exports = function mixin(object, mixins, args) {
+  // add lazy property for each of the mixin, this allows mixins to depend on
+  // one another without any special ordering
+  const descriptors = {}
+  keys(mixins).forEach(name => {
+    const Mixin = mixins[name]
+    name = camelCase(name)
+
+    descriptors[name] = {
+      configurable: true,
+      get: () => {
+        defineProperty(object, name, MIXIN_CYCLIC_DESCRIPTOR)
+        const instance = new Mixin(object, ...args)
+        defineProperty(object, name, {
+          value: instance,
+        })
+        return instance
+      },
+    }
+  })
+  defineProperties(object, descriptors)
+
+  // access all mixin properties to trigger their creation
+  keys(descriptors).forEach(name => {
+    noop(object[name])
+  })
+}

@xen-orchestra/mixin/legacy.js

@@ -1,4 +1,4 @@
-import { getBoundPropertyDescriptor } from 'bind-property-descriptor'
+const { getBoundPropertyDescriptor } = require('bind-property-descriptor')
 
 // ===================================================================
 
@@ -25,7 +25,7 @@ const ownKeys =
 // -------------------------------------------------------------------
 
 const mixin = Mixins => Class => {
-  if (__DEV__ && !Array.isArray(Mixins)) {
+  if (!Array.isArray(Mixins)) {
     throw new TypeError('Mixins should be an array')
   }
 
@@ -44,7 +44,7 @@ const mixin = Mixins => Class => {
       }
 
       for (const prop of ownKeys(Mixin)) {
-        if (__DEV__ && prop in prototype) {
+        if (prop in prototype) {
           throw new Error(`${name}#${prop} is already defined`)
         }
 
@@ -106,7 +106,7 @@ const mixin = Mixins => Class => {
         return
       }
 
-      if (__DEV__ && prop in descriptors) {
+      if (prop in descriptors) {
         throw new Error(`${name}.${prop} is already defined`)
       }
 
@@ -117,4 +117,4 @@ const mixin = Mixins => Class => {
 
   return DecoratedClass
 }
-export { mixin as default }
+module.exports = mixin

@xen-orchestra/mixin/package.json

@@ -1,7 +1,7 @@
 {
   "private": false,
   "name": "@xen-orchestra/mixin",
-  "version": "0.0.0",
+  "version": "0.1.0",
   "license": "ISC",
   "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/mixin",
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
@@ -15,34 +15,14 @@
     "url": "https://vates.fr"
   },
   "preferGlobal": false,
-  "main": "dist/",
-  "files": [
-    "dist/"
-  ],
-  "browserslist": [
-    ">2%"
-  ],
   "engines": {
     "node": ">=6"
   },
   "dependencies": {
-    "bind-property-descriptor": "^1.0.0"
-  },
-  "devDependencies": {
-    "@babel/cli": "^7.0.0",
-    "@babel/core": "^7.0.0",
-    "@babel/preset-env": "^7.0.0",
-    "babel-plugin-dev": "^1.0.0",
-    "cross-env": "^7.0.2",
-    "rimraf": "^3.0.0"
+    "bind-property-descriptor": "^1.0.0",
+    "lodash": "^4.17.21"
   },
   "scripts": {
-    "build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
-    "clean": "rimraf dist/",
-    "dev": "cross-env NODE_ENV=development babel --watch --source-maps --out-dir=dist/ src/",
-    "prebuild": "yarn run clean",
-    "predev": "yarn run prebuild",
-    "prepublishOnly": "yarn run build",
     "postversion": "npm publish"
   }
 }

@xen-orchestra/mixins/.npmignore

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

@xen-orchestra/mixins/Config.js

@@ -1,12 +1,12 @@
-import get from 'lodash/get'
-import identity from 'lodash/identity'
-import { createLogger } from '@xen-orchestra/log'
-import { parseDuration } from '@vates/parse-duration'
-import { watch } from 'app-conf'
+const get = require('lodash/get')
+const identity = require('lodash/identity')
+const { createLogger } = require('@xen-orchestra/log')
+const { parseDuration } = require('@vates/parse-duration')
+const { watch } = require('app-conf')
 
-const { warn } = createLogger('xo:proxy:config')
+const { warn } = createLogger('xo:mixins:config')
 
-export default class Config {
+module.exports = class Config {
   constructor(app, { appDir, appName, config }) {
     this._config = config
     const watchers = (this._watchers = new Set())

@xen-orchestra/mixins/Hooks.js

@@ -1,9 +1,9 @@
-import assert from 'assert'
-import emitAsync from '@xen-orchestra/emit-async'
-import EventEmitter from 'events'
-import { createLogger } from '@xen-orchestra/log'
+const assert = require('assert')
+const emitAsync = require('@xen-orchestra/emit-async')
+const EventEmitter = require('events')
+const { createLogger } = require('@xen-orchestra/log')
 
-const { debug, warn } = createLogger('xo:proxy:hooks')
+const { debug, warn } = createLogger('xo:mixins:hooks')
 
 const runHook = async (emitter, hook) => {
   debug(`${hook} start…`)
@@ -17,7 +17,7 @@ const runHook = async (emitter, hook) => {
   debug(`${hook} finished`)
 }
 
-export default class Hooks extends EventEmitter {
+module.exports = class Hooks extends EventEmitter {
   // Run *clean* async listeners.
   //
   // They normalize existing data, clear invalid entries, etc.

@xen-orchestra/mixins/README.md

@@ -0,0 +1,30 @@
+<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
+
+# @xen-orchestra/mixins
+
+[![Package Version](https://badgen.net/npm/v/@xen-orchestra/mixins)](https://npmjs.org/package/@xen-orchestra/mixins) ![License](https://badgen.net/npm/license/@xen-orchestra/mixins) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@xen-orchestra/mixins)](https://bundlephobia.com/result?p=@xen-orchestra/mixins) [![Node compatibility](https://badgen.net/npm/node/@xen-orchestra/mixins)](https://npmjs.org/package/@xen-orchestra/mixins)
+
+> Mixins shared between xo-proxy and xo-server
+
+## Install
+
+Installation of the [npm package](https://npmjs.org/package/@xen-orchestra/mixins):
+
+```
+> npm install --save @xen-orchestra/mixins
+```
+
+## Contributions
+
+Contributions are _very_ welcomed, either on the documentation or on
+the code.
+
+You may:
+
+- report any [issue](https://github.com/vatesfr/xen-orchestra/issues)
+  you've encountered;
+- fork and create a pull request.
+
+## License
+
+[AGPL-3.0-or-later](https://spdx.org/licenses/AGPL-3.0-or-later) © [Vates SAS](https://vates.fr)

@xen-orchestra/mixins/package.json

@@ -0,0 +1,31 @@
+{
+  "private": false,
+  "name": "@xen-orchestra/mixins",
+  "description": "Mixins shared between xo-proxy and xo-server",
+  "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@xen-orchestra/mixins",
+  "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
+  "repository": {
+    "directory": "@xen-orchestra/mixins",
+    "type": "git",
+    "url": "https://github.com/vatesfr/xen-orchestra.git"
+  },
+  "author": {
+    "name": "Vates SAS",
+    "url": "https://vates.fr"
+  },
+  "license": "AGPL-3.0-or-later",
+  "version": "0.1.0",
+  "engines": {
+    "node": ">=12"
+  },
+  "dependencies": {
+    "@vates/parse-duration": "^0.1.1",
+    "@xen-orchestra/emit-async": "^0.0.0",
+    "@xen-orchestra/log": "^0.2.0",
+    "app-conf": "^0.9.0",
+    "lodash": "^4.17.21"
+  },
+  "scripts": {
+    "postversion": "npm publish --access public"
+  }
+}

@xen-orchestra/openflow/.eslintrc.js

@@ -0,0 +1 @@
+../../scripts/babel-eslintrc.js

@xen-orchestra/openflow/package.json

@@ -26,7 +26,7 @@
     "@babel/cli": "^7.7.4",
     "@babel/core": "^7.7.4",
     "@babel/preset-env": "^7.7.4",
-    "cross": "^1.0.0",
+    "cross-env": "^7.0.2",
     "rimraf": "^3.0.0"
   },
   "dependencies": {

@xen-orchestra/proxy-cli/.eslintrc.js

@@ -0,0 +1 @@
+../../scripts/babel-eslintrc.js

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

@@ -22,9 +22,6 @@
   "bin": {
     "xo-proxy-cli": "dist/index.js"
   },
-  "files": [
-    "dist/"
-  ],
   "engines": {
     "node": ">=8.10"
   },
@@ -36,9 +33,9 @@
     "content-type": "^1.0.4",
     "cson-parser": "^4.0.7",
     "getopts": "^2.2.3",
-    "http-request-plus": "^0.9.1",
+    "http-request-plus": "^0.10.0",
     "json-rpc-protocol": "^0.13.1",
-    "promise-toolbox": "^0.18.1",
+    "promise-toolbox": "^0.19.2",
     "pump": "^3.0.0",
     "pumpify": "^2.0.1",
     "split2": "^3.1.1"

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

@@ -140,6 +140,16 @@ ${pkg.name} v${pkg.version}`
     }
   }
 
+  const $import = ({ $import: path }) => {
+    const data = fs.readFileSync(path, 'utf8')
+    const ext = extname(path).slice(1).toLowerCase()
+    const parse = FORMATS[ext]
+    if (parse === undefined) {
+      throw new Error(`unsupported file: ${path}`)
+    }
+    return visit(parse(data))
+  }
+
   const seq = async seq => {
     const j = callPath.length
     for (let i = 0, n = seq.length; i < n; ++i) {
@@ -153,17 +163,13 @@ ${pkg.name} v${pkg.version}`
     if (Array.isArray(node)) {
       return seq(node)
     }
-    return call(node)
+    const keys = Object.keys(node)
+    return keys.length === 1 && keys[0] === '$import' ? $import(node) : call(node)
   }
 
+  let node
   if (file !== '') {
-    const data = fs.readFileSync(file, 'utf8')
-    const ext = extname(file).slice(1).toLowerCase()
-    const parse = FORMATS[ext]
-    if (parse === undefined) {
-      throw new Error(`unsupported file: ${file}`)
-    }
-    await visit(parse(data))
+    node = { $import: file }
   } else {
     const method = args[0]
     const params = {}
@@ -176,8 +182,9 @@ ${pkg.name} v${pkg.version}`
       params[param.slice(0, j)] = parseValue(param.slice(j + 1))
     }
 
-    await call({ method, params })
+    node = { method, params }
   }
+  await visit(node)
 }
 main(process.argv.slice(2)).then(
   () => {