fix: rework optionnal depency

See https://xcp-ng.org/forum/topic/6334/coalesce-not-showing-anymore/3
Introduced by a9c1239149

.eslintrc.js

@@ -1,5 +1,7 @@
+'use strict'
+
 module.exports = {
-  extends: ['plugin:eslint-comments/recommended', 'standard', 'standard-jsx', 'prettier'],
+  extends: ['plugin:eslint-comments/recommended', 'plugin:n/recommended', 'standard', 'standard-jsx', 'prettier'],
   globals: {
     __DEV__: true,
     $Dict: true,
@@ -15,11 +17,40 @@ module.exports = {
     {
       files: ['cli.{,c,m}js', '*-cli.{,c,m}js', '**/*cli*/**/*.{,c,m}js'],
       rules: {
+        'n/no-process-exit': 'off',
         'no-console': 'off',
       },
     },
+    {
+      files: ['*.mjs'],
+      parserOptions: {
+        sourceType: 'module',
+      },
+    },
+    {
+      files: ['*.spec.{,c,m}js'],
+      rules: {
+        'n/no-unsupported-features/node-builtins': [
+          'error',
+          {
+            version: '>=16',
+          },
+        ],
+        'n/no-unsupported-features/es-syntax': [
+          'error',
+          {
+            version: '>=16',
+          },
+        ],
+      },
+    },
   ],
 
+  parserOptions: {
+    ecmaVersion: 13,
+    sourceType: 'script',
+  },
+
   rules: {
     // disabled because XAPI objects are using camel case
     camelcase: ['off'],
@@ -34,5 +65,7 @@ module.exports = {
     'lines-between-class-members': 'off',
 
     'no-console': ['error', { allow: ['warn', 'error'] }],
+
+    strict: 'error',
   },
 }

.flowconfig

@@ -1,16 +0,0 @@
-[ignore]
-<PROJECT_ROOT>/node_modules/.*
-
-[include]
-
-[libs]
-
-[lints]
-
-[options]
-esproposal.decorators=ignore
-esproposal.optional_chaining=enable
-include_warnings=true
-module.use_strict=true
-
-[strict]

.github/ISSUE_TEMPLATE/bug_report.md

@@ -6,6 +6,18 @@ labels: 'status: triaging :triangular_flag_on_post:, type: bug :bug:'
 assignees: ''
 ---
 
+**XOA or XO from the sources?**
+
+If XOA:
+
+- which release channel? (`stable` vs `latest`)
+- please consider creating a support ticket in [your dedicated support area](https://xen-orchestra.com/#!/member/support)
+
+If XO from the sources:
+
+- Don't forget to [read this first](https://xen-orchestra.com/docs/community.html)
+- As well as follow [this guide](https://xen-orchestra.com/docs/community.html#report-a-bug)
+
 **Describe the bug**
 A clear and concise description of what the bug is.
 
@@ -23,7 +35,7 @@ A clear and concise description of what you expected to happen.
 **Screenshots**
 If applicable, add screenshots to help explain your problem.
 
-**Desktop (please complete the following information):**
+**Environment (please provide the following information):**
 
 - Node: [e.g. 16.12.1]
 - xo-server: [e.g. 5.82.3]

.github/ISSUE_TEMPLATE/feature_request.md

@@ -4,7 +4,6 @@ about: Suggest an idea for this project
 title: ''
 labels: ''
 assignees: ''
-
 ---
 
 **Is your feature request related to a problem? Please describe.**

.github/workflows/push.yml

@@ -0,0 +1,13 @@
+name: CI
+on: [push]
+jobs:
+  build:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: satackey/action-docker-layer-caching@v0.0.11
+      # Ignore the failure of a step and avoid terminating the job.
+        continue-on-error: true
+      - run: docker-compose  -f docker/docker-compose.dev.yml build
+      - run: docker-compose  -f docker/docker-compose.dev.yml up

.gitignore

@@ -1,5 +1,4 @@
 /_book/
-/coverage/
 /node_modules/
 /lerna-debug.log
 /lerna-debug.log.*
@@ -11,10 +10,6 @@
 /packages/*/dist/
 /packages/*/node_modules/
 
-/@xen-orchestra/proxy/src/app/mixins/index.mjs
-
-/packages/vhd-cli/src/commands/index.js
-
 /packages/xen-api/examples/node_modules/
 /packages/xen-api/plot.dat
 
@@ -35,3 +30,7 @@ pnpm-debug.log.*
 yarn-error.log
 yarn-error.log.*
 .env
+
+# code coverage
+.nyc_output/
+coverage/

.travis.yml

@@ -1,23 +0,0 @@
-language: node_js
-node_js:
-  - 14
-
-# Use containers.
-# http://docs.travis-ci.com/user/workers/container-based-infrastructure/
-sudo: false
-addons:
-  apt:
-    packages:
-      - qemu-utils
-      - blktap-utils
-      - vmdk-stream-converter
-
-before_install:
-  - curl -o- -L https://yarnpkg.com/install.sh | bash
-  - export PATH="$HOME/.yarn/bin:$PATH"
-
-cache:
-  yarn: true
-
-script:
-  - yarn run travis-tests

@vates/async-each/.USAGE.md

@@ -14,7 +14,7 @@ Returns a promise wich rejects as soon as a call to `iteratee` throws or a promi
 
 `opts` is an object that can contains the following options:
 
-- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `1`
+- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `10`. The value `0` means no concurrency limit.
 - `signal`: an abort signal to stop the iteration
 - `stopOnError`: wether to stop iteration of first error, or wait for all calls to finish and throw an `AggregateError`, defaults to `true`
 

@vates/async-each/README.md

@@ -32,7 +32,7 @@ Returns a promise wich rejects as soon as a call to `iteratee` throws or a promi
 
 `opts` is an object that can contains the following options:
 
-- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `1`
+- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `10`. The value `0` means no concurrency limit.
 - `signal`: an abort signal to stop the iteration
 - `stopOnError`: wether to stop iteration of first error, or wait for all calls to finish and throw an `AggregateError`, defaults to `true`
 

@vates/async-each/index.js

@@ -9,7 +9,16 @@ class AggregateError extends Error {
   }
 }
 
-exports.asyncEach = function asyncEach(iterable, iteratee, { concurrency = 1, signal, stopOnError = true } = {}) {
+/**
+ * @template Item
+ * @param {Iterable<Item>} iterable
+ * @param {(item: Item, index: number, iterable: Iterable<Item>) => Promise<void>} iteratee
+ * @returns {Promise<void>}
+ */
+exports.asyncEach = function asyncEach(iterable, iteratee, { concurrency = 10, signal, stopOnError = true } = {}) {
+  if (concurrency === 0) {
+    concurrency = Infinity
+  }
   return new Promise((resolve, reject) => {
     const it = (iterable[Symbol.iterator] || iterable[Symbol.asyncIterator]).call(iterable)
     const errors = []

@vates/async-each/index.spec.js

@@ -36,7 +36,7 @@ describe('asyncEach', () => {
       it('works', async () => {
         const iteratee = jest.fn(async () => {})
 
-        await asyncEach.call(thisArg, iterable, iteratee)
+        await asyncEach.call(thisArg, iterable, iteratee, { concurrency: 1 })
 
         expect(iteratee.mock.instances).toEqual(Array.from(values, () => thisArg))
         expect(iteratee.mock.calls).toEqual(Array.from(values, (value, index) => [value, index, iterable]))
@@ -66,7 +66,7 @@ describe('asyncEach', () => {
           }
         })
 
-        expect(await rejectionOf(asyncEach(iterable, iteratee, { stopOnError: true }))).toBe(error)
+        expect(await rejectionOf(asyncEach(iterable, iteratee, { concurrency: 1, stopOnError: true }))).toBe(error)
         expect(iteratee).toHaveBeenCalledTimes(2)
       })
 
@@ -91,7 +91,9 @@ describe('asyncEach', () => {
           }
         })
 
-        await expect(asyncEach(iterable, iteratee, { signal: ac.signal })).rejects.toThrow('asyncEach aborted')
+        await expect(asyncEach(iterable, iteratee, { concurrency: 1, signal: ac.signal })).rejects.toThrow(
+          'asyncEach aborted'
+        )
         expect(iteratee).toHaveBeenCalledTimes(2)
       })
     })

@vates/async-each/package.json

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

@vates/cached-dns.lookup/.USAGE.md

@@ -0,0 +1,30 @@
+Node does not cache queries to `dns.lookup`, which can lead application doing a lot of connections to have perf issues and to saturate Node threads pool.
+
+This library attempts to mitigate these problems by providing a version of this function with a version short cache, applied on both errors and results.
+
+> Limitation: `verbatim: false` option is not supported.
+
+It has exactly the same API as the native method and can be used directly:
+
+```js
+import { createCachedLookup } from '@vates/cached-dns.lookup'
+
+const lookup = createCachedLookup()
+
+lookup('example.net', { all: true, family: 0 }, (error, result) => {
+  if (error != null) {
+    return console.warn(error)
+  }
+  console.log(result)
+})
+```
+
+Or it can be used to replace the native implementation and speed up the whole app:
+
+```js
+// assign our cached implementation to dns.lookup
+const restore = createCachedLookup().patchGlobal()
+
+// to restore the previous implementation
+restore()
+```

@vates/cached-dns.lookup/.npmignore

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

@vates/cached-dns.lookup/README.md

@@ -0,0 +1,63 @@
+<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
+
+# @vates/cached-dns.lookup
+
+[![Package Version](https://badgen.net/npm/v/@vates/cached-dns.lookup)](https://npmjs.org/package/@vates/cached-dns.lookup) ![License](https://badgen.net/npm/license/@vates/cached-dns.lookup) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@vates/cached-dns.lookup)](https://bundlephobia.com/result?p=@vates/cached-dns.lookup) [![Node compatibility](https://badgen.net/npm/node/@vates/cached-dns.lookup)](https://npmjs.org/package/@vates/cached-dns.lookup)
+
+> Cached implementation of dns.lookup
+
+## Install
+
+Installation of the [npm package](https://npmjs.org/package/@vates/cached-dns.lookup):
+
+```
+> npm install --save @vates/cached-dns.lookup
+```
+
+## Usage
+
+Node does not cache queries to `dns.lookup`, which can lead application doing a lot of connections to have perf issues and to saturate Node threads pool.
+
+This library attempts to mitigate these problems by providing a version of this function with a version short cache, applied on both errors and results.
+
+> Limitation: `verbatim: false` option is not supported.
+
+It has exactly the same API as the native method and can be used directly:
+
+```js
+import { createCachedLookup } from '@vates/cached-dns.lookup'
+
+const lookup = createCachedLookup()
+
+lookup('example.net', { all: true, family: 0 }, (error, result) => {
+  if (error != null) {
+    return console.warn(error)
+  }
+  console.log(result)
+})
+```
+
+Or it can be used to replace the native implementation and speed up the whole app:
+
+```js
+// assign our cached implementation to dns.lookup
+const restore = createCachedLookup().patchGlobal()
+
+// to restore the previous implementation
+restore()
+```
+
+## 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
+
+[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

@vates/cached-dns.lookup/index.js

@@ -0,0 +1,72 @@
+'use strict'
+
+const assert = require('assert')
+const dns = require('dns')
+const LRU = require('lru-cache')
+
+function reportResults(all, results, callback) {
+  if (all) {
+    callback(null, results)
+  } else {
+    const first = results[0]
+    callback(null, first.address, first.family)
+  }
+}
+
+exports.createCachedLookup = function createCachedLookup({ lookup = dns.lookup } = {}) {
+  const cache = new LRU({
+    max: 500,
+
+    // 1 minute: long enough to be effective, short enough so there is no need to bother with DNS TTLs
+    ttl: 60e3,
+  })
+
+  function cachedLookup(hostname, options, callback) {
+    let all = false
+    let family = 0
+    if (typeof options === 'function') {
+      callback = options
+    } else if (typeof options === 'number') {
+      family = options
+    } else if (options != null) {
+      assert.notStrictEqual(options.verbatim, false, 'not supported by this implementation')
+      ;({ all = all, family = family } = options)
+    }
+
+    // cache by family option because there will be an error if there is no
+    // entries for the requestion family so we cannot easily cache all families
+    // and filter on reporting back
+    const key = hostname + '/' + family
+
+    const result = cache.get(key)
+    if (result !== undefined) {
+      setImmediate(reportResults, all, result, callback)
+    } else {
+      lookup(hostname, { all: true, family, verbatim: true }, function onLookup(error, results) {
+        // errors are not cached because this will delay recovery after DNS/network issues
+        //
+        // there are no reliable way to detect if the error is real or simply
+        // that there are no results for the requested hostname
+        //
+        // there should be much fewer errors than success, therefore it should
+        // not be a big deal to not cache them
+        if (error != null) {
+          return callback(error)
+        }
+
+        cache.set(key, results)
+        reportResults(all, results, callback)
+      })
+    }
+  }
+  cachedLookup.patchGlobal = function patchGlobal() {
+    const previous = dns.lookup
+    dns.lookup = cachedLookup
+    return function restoreGlobal() {
+      assert.strictEqual(dns.lookup, cachedLookup)
+      dns.lookup = previous
+    }
+  }
+
+  return cachedLookup
+}

@vates/cached-dns.lookup/package.json

@@ -0,0 +1,32 @@
+{
+  "engines": {
+    "node": ">=8"
+  },
+  "dependencies": {
+    "lru-cache": "^7.0.4"
+  },
+  "private": false,
+  "name": "@vates/cached-dns.lookup",
+  "description": "Cached implementation of dns.lookup",
+  "keywords": [
+    "cache",
+    "dns",
+    "lookup"
+  ],
+  "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/cached-dns.lookup",
+  "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
+  "repository": {
+    "directory": "@vates/cached-dns.lookup",
+    "type": "git",
+    "url": "https://github.com/vatesfr/xen-orchestra.git"
+  },
+  "author": {
+    "name": "Vates SAS",
+    "url": "https://vates.fr"
+  },
+  "license": "ISC",
+  "version": "1.0.0",
+  "scripts": {
+    "postversion": "npm publish --access public"
+  }
+}

@vates/coalesce-calls/index.js

@@ -1,3 +1,5 @@
+'use strict'
+
 exports.coalesceCalls = function (fn) {
   let promise
   const clean = () => {

@vates/coalesce-calls/index.spec.js

@@ -1,3 +1,5 @@
+'use strict'
+
 /* eslint-env jest */
 
 const { coalesceCalls } = require('./')

@vates/compose/index.spec.js

@@ -1,3 +1,5 @@
+'use strict'
+
 /* eslint-env jest */
 
 const { compose } = require('./')

@vates/decorate-with/.USAGE.md

@@ -13,15 +13,19 @@ class Foo {
 }
 ```
 
-### `decorateMethodsWith(class, map)`
+### `decorateClass(class, map)`
 
-Decorates a number of methods directly, without using the decorator syntax:
+Decorates a number of accessors and methods directly, without using the decorator syntax:
 
 ```js
-import { decorateMethodsWith } from '@vates/decorate-with'
+import { decorateClass } from '@vates/decorate-with'
 
 class Foo {
-  bar() {
+  get bar() {
+    // body
+  }
+
+  set bar(value) {
     // body
   }
 
@@ -30,22 +34,28 @@ class Foo {
   }
 }
 
-decorateMethodsWith(Foo, {
-  // without arguments
-  bar: lodash.curry,
+decorateClass(Foo, {
+  // getter and/or setter
+  bar: {
+    // without arguments
+    get: lodash.memoize,
 
-  // with arguments
-  baz: [lodash.debounce, 150],
+    // with arguments
+    set: [lodash.debounce, 150],
+  },
+
+  // method (with or without arguments)
+  baz: lodash.curry,
 })
 ```
 
 The decorated class is returned, so you can export it directly.
 
-To apply multiple transforms to a method, you can either call `decorateMethodsWith` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
+To apply multiple transforms to an accessor/method, you can either call `decorateClass` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
 
 ```js
-decorateMethodsWith(Foo, {
-  bar: compose([
+decorateClass(Foo, {
+  baz: compose([
     [lodash.debounce, 150]
     lodash.curry,
   ])
@@ -69,4 +79,8 @@ class Foo {
 }
 ```
 
-Because it's a normal function, it can also be used with `decorateMethodsWith`, with `compose` or even by itself.
+Because it's a normal function, it can also be used with `decorateClass`, with `compose` or even by itself.
+
+### `decorateMethodsWith(class, map)`
+
+> Deprecated alias for [`decorateClass(class, map)`](#decorateclassclass-map).

@vates/decorate-with/README.md

@@ -31,15 +31,19 @@ class Foo {
 }
 ```
 
-### `decorateMethodsWith(class, map)`
+### `decorateClass(class, map)`
 
-Decorates a number of methods directly, without using the decorator syntax:
+Decorates a number of accessors and methods directly, without using the decorator syntax:
 
 ```js
-import { decorateMethodsWith } from '@vates/decorate-with'
+import { decorateClass } from '@vates/decorate-with'
 
 class Foo {
-  bar() {
+  get bar() {
+    // body
+  }
+
+  set bar(value) {
     // body
   }
 
@@ -48,22 +52,28 @@ class Foo {
   }
 }
 
-decorateMethodsWith(Foo, {
-  // without arguments
-  bar: lodash.curry,
+decorateClass(Foo, {
+  // getter and/or setter
+  bar: {
+    // without arguments
+    get: lodash.memoize,
 
-  // with arguments
-  baz: [lodash.debounce, 150],
+    // with arguments
+    set: [lodash.debounce, 150],
+  },
+
+  // method (with or without arguments)
+  baz: lodash.curry,
 })
 ```
 
 The decorated class is returned, so you can export it directly.
 
-To apply multiple transforms to a method, you can either call `decorateMethodsWith` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
+To apply multiple transforms to an accessor/method, you can either call `decorateClass` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
 
 ```js
-decorateMethodsWith(Foo, {
-  bar: compose([
+decorateClass(Foo, {
+  baz: compose([
     [lodash.debounce, 150]
     lodash.curry,
   ])
@@ -87,7 +97,11 @@ class Foo {
 }
 ```
 
-Because it's a normal function, it can also be used with `decorateMethodsWith`, with `compose` or even by itself.
+Because it's a normal function, it can also be used with `decorateClass`, with `compose` or even by itself.
+
+### `decorateMethodsWith(class, map)`
+
+> Deprecated alias for [`decorateClass(class, map)`](#decorateclassclass-map).
 
 ## Contributions
 

@vates/decorate-with/index.js

@@ -1,3 +1,5 @@
+'use strict'
+
 exports.decorateWith = function decorateWith(fn, ...args) {
   return (target, name, descriptor) => ({
     ...descriptor,
@@ -7,14 +9,27 @@ exports.decorateWith = function decorateWith(fn, ...args) {
 
 const { getOwnPropertyDescriptor, defineProperty } = Object
 
-exports.decorateMethodsWith = function decorateMethodsWith(klass, map) {
+function applyDecorator(decorator, value) {
+  return typeof decorator === 'function' ? decorator(value) : decorator[0](value, ...decorator.slice(1))
+}
+
+exports.decorateClass = exports.decorateMethodsWith = function decorateClass(klass, map) {
   const { prototype } = klass
   for (const name of Object.keys(map)) {
-    const descriptor = getOwnPropertyDescriptor(prototype, name)
-    const { value } = descriptor
-
     const decorator = map[name]
-    descriptor.value = typeof decorator === 'function' ? decorator(value) : decorator[0](value, ...decorator.slice(1))
+    const descriptor = getOwnPropertyDescriptor(prototype, name)
+    if (typeof decorator === 'function' || Array.isArray(decorator)) {
+      descriptor.value = applyDecorator(decorator, descriptor.value)
+    } else {
+      const { get, set } = decorator
+      if (get !== undefined) {
+        descriptor.get = applyDecorator(get, descriptor.get)
+      }
+      if (set !== undefined) {
+        descriptor.set = applyDecorator(set, descriptor.set)
+      }
+    }
+
     defineProperty(prototype, name, descriptor)
   }
   return klass

@vates/decorate-with/index.spec.js

@@ -0,0 +1,152 @@
+'use strict'
+
+const assert = require('assert')
+const { describe, it } = require('tap').mocha
+
+const { decorateClass, decorateWith, decorateMethodsWith, perInstance } = require('./')
+
+const identity = _ => _
+
+describe('decorateWith', () => {
+  it('works', () => {
+    const expectedArgs = [Math.random(), Math.random()]
+    const expectedFn = Function.prototype
+    const newFn = () => {}
+
+    const decorator = decorateWith(function wrapper(fn, ...args) {
+      assert.deepStrictEqual(fn, expectedFn)
+      assert.deepStrictEqual(args, expectedArgs)
+
+      return newFn
+    }, ...expectedArgs)
+
+    const descriptor = {
+      configurable: true,
+      enumerable: false,
+      value: expectedFn,
+      writable: true,
+    }
+    assert.deepStrictEqual(decorator({}, 'foo', descriptor), {
+      ...descriptor,
+      value: newFn,
+    })
+  })
+})
+
+describe('decorateClass', () => {
+  it('works', () => {
+    class C {
+      foo() {}
+      bar() {}
+      get baz() {}
+      // eslint-disable-next-line accessor-pairs
+      set qux(_) {}
+    }
+
+    const expectedArgs = [Math.random(), Math.random()]
+    const P = C.prototype
+
+    const descriptors = Object.getOwnPropertyDescriptors(P)
+
+    const newFoo = () => {}
+    const newBar = () => {}
+    const newGetBaz = () => {}
+    const newSetQux = _ => {}
+
+    decorateClass(C, {
+      foo(fn) {
+        assert.strictEqual(arguments.length, 1)
+        assert.strictEqual(fn, P.foo)
+        return newFoo
+      },
+      bar: [
+        function (fn, ...args) {
+          assert.strictEqual(fn, P.bar)
+          assert.deepStrictEqual(args, expectedArgs)
+          return newBar
+        },
+        ...expectedArgs,
+      ],
+      baz: {
+        get(fn) {
+          assert.strictEqual(arguments.length, 1)
+          assert.strictEqual(fn, descriptors.baz.get)
+          return newGetBaz
+        },
+      },
+      qux: {
+        set: [
+          function (fn, ...args) {
+            assert.strictEqual(fn, descriptors.qux.set)
+            assert.deepStrictEqual(args, expectedArgs)
+            return newSetQux
+          },
+          ...expectedArgs,
+        ],
+      },
+    })
+
+    const newDescriptors = Object.getOwnPropertyDescriptors(P)
+    assert.deepStrictEqual(newDescriptors.foo, { ...descriptors.foo, value: newFoo })
+    assert.deepStrictEqual(newDescriptors.bar, { ...descriptors.bar, value: newBar })
+    assert.deepStrictEqual(newDescriptors.baz, { ...descriptors.baz, get: newGetBaz })
+    assert.deepStrictEqual(newDescriptors.qux, { ...descriptors.qux, set: newSetQux })
+  })
+
+  it('throws if using an accessor decorator for a method', function () {
+    assert.throws(() =>
+      decorateClass(
+        class {
+          foo() {}
+        },
+        { foo: { get: identity, set: identity } }
+      )
+    )
+  })
+
+  it('throws if using a method decorator for an accessor', function () {
+    assert.throws(() =>
+      decorateClass(
+        class {
+          get foo() {}
+        },
+        { foo: identity }
+      )
+    )
+  })
+})
+
+it('decorateMethodsWith is an alias of decorateClass', function () {
+  assert.strictEqual(decorateMethodsWith, decorateClass)
+})
+
+describe('perInstance', () => {
+  it('works', () => {
+    let calls = 0
+
+    const expectedArgs = [Math.random(), Math.random()]
+    const expectedFn = Function.prototype
+    function wrapper(fn, ...args) {
+      assert.strictEqual(fn, expectedFn)
+      assert.deepStrictEqual(args, expectedArgs)
+      const i = ++calls
+      return () => i
+    }
+
+    const wrapped = perInstance(expectedFn, wrapper, ...expectedArgs)
+
+    // decorator is not called before decorated called
+    assert.strictEqual(calls, 0)
+
+    const o1 = {}
+    const o2 = {}
+
+    assert.strictEqual(wrapped.call(o1), 1)
+
+    // the same decorated function is returned for the same instance
+    assert.strictEqual(wrapped.call(o1), 1)
+
+    // a new decorated function is returned for another instance
+    assert.strictEqual(wrapped.call(o2), 2)
+  })
+})

@vates/decorate-with/package.json

@@ -20,11 +20,15 @@
     "url": "https://vates.fr"
   },
   "license": "ISC",
-  "version": "1.0.0",
+  "version": "2.0.0",
   "engines": {
     "node": ">=8.10"
   },
   "scripts": {
-    "postversion": "npm publish --access public"
+    "postversion": "npm publish --access public",
+    "test": "tap"
+  },
+  "devDependencies": {
+    "tap": "^16.0.1"
   }
 }

@vates/disposable/debounceResource.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const { asyncMap } = require('@xen-orchestra/async-map')
 const { createLogger } = require('@xen-orchestra/log')
 

@vates/disposable/debounceResource.spec.js

@@ -1,3 +1,5 @@
+'use strict'
+
 /* eslint-env jest */
 
 const { createDebounceResource } = require('./debounceResource')

@vates/disposable/deduped.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const ensureArray = require('ensure-array')
 const { MultiKeyMap } = require('@vates/multi-key-map')
 

@vates/disposable/deduped.spec.js

@@ -1,3 +1,5 @@
+'use strict'
+
 /* eslint-env jest */
 
 const { deduped } = require('./deduped')

@vates/event-listeners-manager/.USAGE.md

@@ -0,0 +1,50 @@
+> This library is compatible with Node's `EventEmitter` and web browsers' `EventTarget` APIs.
+
+### API
+
+```js
+import { EventListenersManager } from '@vates/event-listeners-manager'
+
+const events = new EventListenersManager(emitter)
+
+// adding listeners
+events.add('foo', onFoo).add('bar', onBar).on('baz', onBaz)
+
+// removing a specific listener
+events.remove('foo', onFoo)
+
+// removing all listeners for a specific event
+events.removeAll('foo')
+
+// removing all listeners
+events.removeAll()
+```
+
+### Typical use case
+
+> Removing all listeners when no longer necessary.
+
+Manually:
+
+```js
+const onFoo = () => {}
+const onBar = () => {}
+const onBaz = () => {}
+emitter.on('foo', onFoo).on('bar', onBar).on('baz', onBaz)
+
+// CODE LOGIC
+
+emitter.off('foo', onFoo).off('bar', onBar).off('baz', onBaz)
+```
+
+With this library:
+
+```js
+const events = new EventListenersManager(emitter)
+
+events.add('foo', () => {})).add('bar', () => {})).add('baz', () => {}))
+
+// CODE LOGIC
+
+events.removeAll()
+```

@vates/event-listeners-manager/.npmignore

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

@vates/event-listeners-manager/README.md

@@ -0,0 +1,81 @@
+<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
+
+# @vates/event-listeners-manager
+
+[![Package Version](https://badgen.net/npm/v/@vates/event-listeners-manager)](https://npmjs.org/package/@vates/event-listeners-manager) ![License](https://badgen.net/npm/license/@vates/event-listeners-manager) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@vates/event-listeners-manager)](https://bundlephobia.com/result?p=@vates/event-listeners-manager) [![Node compatibility](https://badgen.net/npm/node/@vates/event-listeners-manager)](https://npmjs.org/package/@vates/event-listeners-manager)
+
+## Install
+
+Installation of the [npm package](https://npmjs.org/package/@vates/event-listeners-manager):
+
+```
+> npm install --save @vates/event-listeners-manager
+```
+
+## Usage
+
+> This library is compatible with Node's `EventEmitter` and web browsers' `EventTarget` APIs.
+
+### API
+
+```js
+import { EventListenersManager } from '@vates/event-listeners-manager'
+
+const events = new EventListenersManager(emitter)
+
+// adding listeners
+events.add('foo', onFoo).add('bar', onBar).on('baz', onBaz)
+
+// removing a specific listener
+events.remove('foo', onFoo)
+
+// removing all listeners for a specific event
+events.removeAll('foo')
+
+// removing all listeners
+events.removeAll()
+```
+
+### Typical use case
+
+> Removing all listeners when no longer necessary.
+
+Manually:
+
+```js
+const onFoo = () => {}
+const onBar = () => {}
+const onBaz = () => {}
+emitter.on('foo', onFoo).on('bar', onBar).on('baz', onBaz)
+
+// CODE LOGIC
+
+emitter.off('foo', onFoo).off('bar', onBar).off('baz', onBaz)
+```
+
+With this library:
+
+```js
+const events = new EventListenersManager(emitter)
+
+events.add('foo', () => {})).add('bar', () => {})).add('baz', () => {}))
+
+// CODE LOGIC
+
+events.removeAll()
+```
+
+## 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
+
+[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

@vates/event-listeners-manager/index.js

@@ -0,0 +1,56 @@
+'use strict'
+
+exports.EventListenersManager = class EventListenersManager {
+  constructor(emitter) {
+    this._listeners = new Map()
+
+    this._add = (emitter.addListener || emitter.addEventListener).bind(emitter)
+    this._remove = (emitter.removeListener || emitter.removeEventListener).bind(emitter)
+  }
+
+  add(type, listener) {
+    let listeners = this._listeners.get(type)
+    if (listeners === undefined) {
+      listeners = new Set()
+      this._listeners.set(type, listeners)
+    }
+
+    // don't add the same listener multiple times (allowed on Node.js)
+    if (!listeners.has(listener)) {
+      listeners.add(listener)
+      this._add(type, listener)
+    }
+
+    return this
+  }
+
+  remove(type, listener) {
+    const allListeners = this._listeners
+    const listeners = allListeners.get(type)
+    if (listeners !== undefined && listeners.delete(listener)) {
+      this._remove(type, listener)
+      if (listeners.size === 0) {
+        allListeners.delete(type)
+      }
+    }
+
+    return this
+  }
+
+  removeAll(type) {
+    const allListeners = this._listeners
+    const remove = this._remove
+    const types = type !== undefined ? [type] : allListeners.keys()
+    for (const type of types) {
+      const listeners = allListeners.get(type)
+      if (listeners !== undefined) {
+        allListeners.delete(type)
+        for (const listener of listeners) {
+          remove(type, listener)
+        }
+      }
+    }
+
+    return this
+  }
+}

@vates/event-listeners-manager/index.spec.js

@@ -0,0 +1,67 @@
+'use strict'
+
+const t = require('tap')
+const { EventEmitter } = require('events')
+
+const { EventListenersManager } = require('./')
+
+const noop = Function.prototype
+
+// function spy (impl = Function.prototype) {
+//   function spy() {
+//     spy.calls.push([Array.from(arguments), this])
+//   }
+//   spy.calls = []
+//   return spy
+// }
+
+function assertListeners(t, event, listeners) {
+  t.strictSame(t.context.ee.listeners(event), listeners)
+}
+
+t.beforeEach(function (t) {
+  t.context.ee = new EventEmitter()
+  t.context.em = new EventListenersManager(t.context.ee)
+})
+
+t.test('.add adds a listener', function (t) {
+  t.context.em.add('foo', noop)
+
+  assertListeners(t, 'foo', [noop])
+
+  t.end()
+})
+
+t.test('.add does not add a duplicate listener', function (t) {
+  t.context.em.add('foo', noop).add('foo', noop)
+
+  assertListeners(t, 'foo', [noop])
+
+  t.end()
+})
+
+t.test('.remove removes a listener', function (t) {
+  t.context.em.add('foo', noop).remove('foo', noop)
+
+  assertListeners(t, 'foo', [])
+
+  t.end()
+})
+
+t.test('.removeAll removes all listeners of a given type', function (t) {
+  t.context.em.add('foo', noop).add('bar', noop).removeAll('foo')
+
+  assertListeners(t, 'foo', [])
+  assertListeners(t, 'bar', [noop])
+
+  t.end()
+})
+
+t.test('.removeAll removes all listeners', function (t) {
+  t.context.em.add('foo', noop).add('bar', noop).removeAll()
+
+  assertListeners(t, 'foo', [])
+  assertListeners(t, 'bar', [])
+
+  t.end()
+})

@vates/event-listeners-manager/package.json

@@ -0,0 +1,46 @@
+{
+  "engines": {
+    "node": ">=6"
+  },
+  "private": false,
+  "name": "@vates/event-listeners-manager",
+  "descriptions": "Easy way to clean up event listeners",
+  "keywords": [
+    "add",
+    "addEventListener",
+    "addListener",
+    "browser",
+    "clear",
+    "DOM",
+    "emitter",
+    "event",
+    "EventEmitter",
+    "EventTarget",
+    "management",
+    "manager",
+    "node",
+    "remove",
+    "removeEventListener",
+    "removeListener"
+  ],
+  "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/event-listeners-manager",
+  "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
+  "repository": {
+    "directory": "@vates/event-listeners-manager",
+    "type": "git",
+    "url": "https://github.com/vatesfr/xen-orchestra.git"
+  },
+  "author": {
+    "name": "Vates SAS",
+    "url": "https://vates.fr"
+  },
+  "license": "ISC",
+  "version": "1.0.1",
+  "scripts": {
+    "postversion": "npm publish --access public",
+    "test": "tap --branches=72"
+  },
+  "devDependencies": {
+    "tap": "^16.2.0"
+  }
+}

@vates/fuse-vhd/.npmignore

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

@vates/fuse-vhd/index.js

@@ -0,0 +1,66 @@
+'use strict'
+
+const LRU = require('lru-cache')
+const { VhdSynthetic } = require('vhd-lib')
+const { Disposable, fromCallback } = require('promise-toolbox')
+
+// build a s stat object from https://github.com/fuse-friends/fuse-native/blob/master/test/fixtures/stat.js
+const stat = st => ({
+  mtime: st.mtime || new Date(),
+  atime: st.atime || new Date(),
+  ctime: st.ctime || new Date(),
+  size: st.size !== undefined ? st.size : 0,
+  mode: st.mode === 'dir' ? 16877 : st.mode === 'file' ? 33188 : st.mode === 'link' ? 41453 : st.mode,
+  uid: st.uid !== undefined ? st.uid : process.getuid(),
+  gid: st.gid !== undefined ? st.gid : process.getgid(),
+})
+
+exports.mount = Disposable.factory(async function* mount(handler, diskPath, mountDir) {
+  const Fuse = require('fuse-native')
+  const vhd = yield VhdSynthetic.fromVhdChain(handler, diskPath)
+
+  const cache = new LRU({
+    max: 16, // each cached block is 2MB in size
+  })
+  await vhd.readBlockAllocationTable()
+  const fuse = new Fuse(mountDir, {
+    async readdir(path, cb) {
+      if (path === '/') {
+        return cb(null, ['vhd0'])
+      }
+      cb(Fuse.ENOENT)
+    },
+    async getattr(path, cb) {
+      if (path === '/') {
+        return cb(
+          null,
+          stat({
+            mode: 'dir',
+            size: 4096,
+          })
+        )
+      }
+      if (path === '/vhd0') {
+        return cb(
+          null,
+          stat({
+            mode: 'file',
+            size: vhd.footer.currentSize,
+          })
+        )
+      }
+
+      cb(Fuse.ENOENT)
+    },
+    read(path, fd, buf, len, pos, cb) {
+      if (path === '/vhd0') {
+        return vhd.readRawData(pos, len, cache, buf).then(cb)
+      }
+      throw new Error(`read file ${path} not exists`)
+    },
+  })
+  return new Disposable(
+    () => fromCallback(() => fuse.unmount()),
+    fromCallback(() => fuse.mount())
+  )
+})

@vates/fuse-vhd/package.json

@@ -0,0 +1,32 @@
+{
+  "name": "@vates/fuse-vhd",
+  "version": "0.0.1",
+  "license": "ISC",
+  "private": false,
+  "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/fuse-vhd",
+  "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
+  "repository": {
+    "directory": "@vates/fuse-vhd",
+    "type": "git",
+    "url": "https://github.com/vatesfr/xen-orchestra.git"
+  },
+  "author": {
+    "name": "Vates SAS",
+    "url": "https://vates.fr"
+  },
+  "engines": {
+    "node": ">=10.0"
+  },
+  "dependencies": {
+    "@xen-orchestra/log": "^0.3.0",
+    "lru-cache": "^7.14.0",
+    "promise-toolbox": "^0.21.0",
+    "vhd-lib": "^4.0.1"
+  },
+  "optionalDependencies": {
+    "fuse-native": "^2.2.6"
+  },
+  "scripts": {
+    "postversion": "npm publish --access public"
+  }
+}

@vates/multi-key-map/index.js

@@ -1,3 +1,5 @@
+'use strict'
+
 class Node {
   constructor(value) {
     this.children = new Map()

@vates/multi-key-map/index.spec.js

@@ -1,3 +1,5 @@
+'use strict'
+
 /* eslint-env jest */
 
 const { MultiKeyMap } = require('./')

@vates/parse-duration/index.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const ms = require('ms')
 
 exports.parseDuration = value => {

@vates/predicates/.USAGE.md

@@ -0,0 +1,57 @@
+`undefined` predicates are ignored and `undefined` is returned if all predicates are `undefined`, this permits the most efficient composition:
+
+```js
+const compositePredicate = every(undefined, some(predicate2, undefined))
+
+// ends up as
+
+const compositePredicate = predicate2
+```
+
+Predicates can also be passed wrapped in an array:
+
+```js
+const compositePredicate = every([predicate1, some([predicate2, predicate3])])
+```
+
+`this` and all arguments are passed to the nested predicates.
+
+### `every(predicates)`
+
+> Returns a predicate that returns `true` iff every predicate returns `true`.
+
+```js
+const isBetween3And7 = every(
+  n => n >= 3,
+  n => n <= 7
+)
+
+isBetween3And10(0)
+// → false
+
+isBetween3And10(5)
+// → true
+
+isBetween3And10(10)
+// → false
+```
+
+### `some(predicates)`
+
+> Returns a predicate that returns `true` iff some predicate returns `true`.
+
+```js
+const isAliceOrBob = some(
+  name => name === 'Alice',
+  name => name === 'Bob'
+)
+
+isAliceOrBob('Alice')
+// → true
+
+isAliceOrBob('Bob')
+// → true
+
+isAliceOrBob('Oscar')
+// → false
+```

@vates/predicates/.npmignore

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

@vates/predicates/README.md

@@ -0,0 +1,90 @@
+<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
+
+# @vates/predicates
+
+[![Package Version](https://badgen.net/npm/v/@vates/predicates)](https://npmjs.org/package/@vates/predicates) ![License](https://badgen.net/npm/license/@vates/predicates) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@vates/predicates)](https://bundlephobia.com/result?p=@vates/predicates) [![Node compatibility](https://badgen.net/npm/node/@vates/predicates)](https://npmjs.org/package/@vates/predicates)
+
+> Utilities to compose predicates
+
+## Install
+
+Installation of the [npm package](https://npmjs.org/package/@vates/predicates):
+
+```
+> npm install --save @vates/predicates
+```
+
+## Usage
+
+`undefined` predicates are ignored and `undefined` is returned if all predicates are `undefined`, this permits the most efficient composition:
+
+```js
+const compositePredicate = every(undefined, some(predicate2, undefined))
+
+// ends up as
+
+const compositePredicate = predicate2
+```
+
+Predicates can also be passed wrapped in an array:
+
+```js
+const compositePredicate = every([predicate1, some([predicate2, predicate3])])
+```
+
+`this` and all arguments are passed to the nested predicates.
+
+### `every(predicates)`
+
+> Returns a predicate that returns `true` iff every predicate returns `true`.
+
+```js
+const isBetween3And7 = every(
+  n => n >= 3,
+  n => n <= 7
+)
+
+isBetween3And10(0)
+// → false
+
+isBetween3And10(5)
+// → true
+
+isBetween3And10(10)
+// → false
+```
+
+### `some(predicates)`
+
+> Returns a predicate that returns `true` iff some predicate returns `true`.
+
+```js
+const isAliceOrBob = some(
+  name => name === 'Alice',
+  name => name === 'Bob'
+)
+
+isAliceOrBob('Alice')
+// → true
+
+isAliceOrBob('Bob')
+// → true
+
+isAliceOrBob('Oscar')
+// → false
+```
+
+## 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
+
+[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

@vates/predicates/index.js

@@ -0,0 +1,71 @@
+'use strict'
+
+const {
+  isArray,
+  prototype: { filter },
+} = Array
+
+class InvalidPredicate extends TypeError {
+  constructor(value) {
+    super('not a valid predicate')
+    this.value = value
+  }
+}
+
+function isDefinedPredicate(value) {
+  if (value === undefined) {
+    return false
+  }
+
+  if (typeof value !== 'function') {
+    throw new InvalidPredicate(value)
+  }
+
+  return true
+}
+
+function handleArgs() {
+  let predicates
+  if (!(arguments.length === 1 && isArray((predicates = arguments[0])))) {
+    predicates = arguments
+  }
+  return filter.call(predicates, isDefinedPredicate)
+}
+
+exports.every = function every() {
+  const predicates = handleArgs.apply(this, arguments)
+  const n = predicates.length
+  if (n === 0) {
+    return
+  }
+  if (n === 1) {
+    return predicates[0]
+  }
+  return function everyPredicate() {
+    for (let i = 0; i < n; ++i) {
+      if (!predicates[i].apply(this, arguments)) {
+        return false
+      }
+    }
+    return true
+  }
+}
+
+exports.some = function some() {
+  const predicates = handleArgs.apply(this, arguments)
+  const n = predicates.length
+  if (n === 0) {
+    return
+  }
+  if (n === 1) {
+    return predicates[0]
+  }
+  return function somePredicate() {
+    for (let i = 0; i < n; ++i) {
+      if (predicates[i].apply(this, arguments)) {
+        return true
+      }
+    }
+    return false
+  }
+}

@vates/predicates/index.spec.js

@@ -0,0 +1,65 @@
+'use strict'
+
+const assert = require('assert/strict')
+const { describe, it } = require('tap').mocha
+
+const { every, some } = require('./')
+
+const T = () => true
+const F = () => false
+
+const testArgsHandling = fn => {
+  it('returns undefined if all predicates are undefined', () => {
+    assert.equal(fn(undefined), undefined)
+    assert.equal(fn([undefined]), undefined)
+  })
+
+  it('returns the predicate if only a single one is passed', () => {
+    assert.equal(fn(undefined, T), T)
+    assert.equal(fn([undefined, T]), T)
+  })
+
+  it('throws if it receives a non-predicate', () => {
+    const error = new TypeError('not a valid predicate')
+    error.value = 3
+    assert.throws(() => fn(3), error)
+  })
+
+  it('forwards this and arguments to predicates', () => {
+    const thisArg = 'qux'
+    const args = ['foo', 'bar', 'baz']
+    const predicate = function () {
+      assert.equal(this, thisArg)
+      assert.deepEqual(Array.from(arguments), args)
+    }
+    fn(predicate, predicate).apply(thisArg, args)
+  })
+}
+
+const runTests = (fn, truthTable) =>
+  it('works', () => {
+    truthTable.forEach(([result, ...predicates]) => {
+      assert.equal(fn(...predicates)(), result)
+      assert.equal(fn(predicates)(), result)
+    })
+  })
+
+describe('every', () => {
+  testArgsHandling(every)
+  runTests(every, [
+    [true, T, T],
+    [false, T, F],
+    [false, F, T],
+    [false, F, F],
+  ])
+})
+
+describe('some', () => {
+  testArgsHandling(some)
+  runTests(some, [
+    [true, T, T],
+    [true, T, F],
+    [true, F, T],
+    [false, F, F],
+  ])
+})

@vates/predicates/package.json

@@ -0,0 +1,40 @@
+{
+  "private": false,
+  "name": "@vates/predicates",
+  "description": "Utilities to compose predicates",
+  "keywords": [
+    "and",
+    "combine",
+    "compose",
+    "every",
+    "function",
+    "functions",
+    "or",
+    "predicate",
+    "predicates",
+    "some"
+  ],
+  "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/predicates",
+  "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
+  "repository": {
+    "directory": "@vates/predicates",
+    "type": "git",
+    "url": "https://github.com/vatesfr/xen-orchestra.git"
+  },
+  "author": {
+    "name": "Vates SAS",
+    "url": "https://vates.fr"
+  },
+  "license": "ISC",
+  "version": "1.0.0",
+  "engines": {
+    "node": ">=6"
+  },
+  "scripts": {
+    "postversion": "npm publish --access public",
+    "test": "tap"
+  },
+  "devDependencies": {
+    "tap": "^16.0.1"
+  }
+}

@vates/read-chunk/.USAGE.md

@@ -0,0 +1,26 @@
+### `readChunk(stream, [size])`
+
+- returns the next available chunk of data
+- like `stream.read()`, a number of bytes can be specified
+- returns with less data than expected if stream has ended
+- returns `null` if the stream has ended and no data has been read
+
+```js
+import { readChunk } from '@vates/read-chunk'
+;(async () => {
+  let chunk
+  while ((chunk = await readChunk(stream, 1024)) !== null) {
+    // do something with chunk
+  }
+})()
+```
+
+### `readChunkStrict(stream, [size])`
+
+Similar behavior to `readChunk` but throws if the stream ended before the requested data could be read.
+
+```js
+import { readChunkStrict } from '@vates/read-chunk'
+
+const chunk = await readChunkStrict(stream, 1024)
+```

@vates/read-chunk/README.md

@@ -16,9 +16,12 @@ Installation of the [npm package](https://npmjs.org/package/@vates/read-chunk):
 
 ## Usage
 
+### `readChunk(stream, [size])`
+
 - returns the next available chunk of data
 - like `stream.read()`, a number of bytes can be specified
-- returns `null` if the stream has ended
+- returns with less data than expected if stream has ended
+- returns `null` if the stream has ended and no data has been read
 
 ```js
 import { readChunk } from '@vates/read-chunk'
@@ -30,6 +33,16 @@ import { readChunk } from '@vates/read-chunk'
 })()
 ```
 
+### `readChunkStrict(stream, [size])`
+
+Similar behavior to `readChunk` but throws if the stream ended before the requested data could be read.
+
+```js
+import { readChunkStrict } from '@vates/read-chunk'
+
+const chunk = await readChunkStrict(stream, 1024)
+```
+
 ## Contributions
 
 Contributions are _very_ welcomed, either on the documentation or on

@vates/read-chunk/USAGE.md

@@ -1,13 +0,0 @@
-- returns the next available chunk of data
-- like `stream.read()`, a number of bytes can be specified
-- returns `null` if the stream has ended
-
-```js
-import { readChunk } from '@vates/read-chunk'
-;(async () => {
-  let chunk
-  while ((chunk = await readChunk(stream, 1024)) !== null) {
-    // do something with chunk
-  }
-})()
-```

@vates/read-chunk/index.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const readChunk = (stream, size) =>
   size === 0
     ? Promise.resolve(Buffer.alloc(0))
@@ -28,3 +30,22 @@ const readChunk = (stream, size) =>
         onReadable()
       })
 exports.readChunk = readChunk
+
+exports.readChunkStrict = async function readChunkStrict(stream, size) {
+  const chunk = await readChunk(stream, size)
+  if (chunk === null) {
+    throw new Error('stream has ended without data')
+  }
+
+  if (size !== undefined && chunk.length !== size) {
+    const error = new Error('stream has ended with not enough data')
+    Object.defineProperties(error, {
+      chunk: {
+        value: chunk,
+      },
+    })
+    throw error
+  }
+
+  return chunk
+}

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

@@ -1,8 +1,10 @@
+'use strict'
+
 /* eslint-env jest */
 
 const { Readable } = require('stream')
 
-const { readChunk } = require('./')
+const { readChunk, readChunkStrict } = require('./')
 
 const makeStream = it => Readable.from(it, { objectMode: false })
 makeStream.obj = Readable.from
@@ -41,3 +43,27 @@ describe('readChunk', () => {
     })
   })
 })
+
+const rejectionOf = promise =>
+  promise.then(
+    value => {
+      throw value
+    },
+    error => error
+  )
+
+describe('readChunkStrict', function () {
+  it('throws if stream is empty', async () => {
+    const error = await rejectionOf(readChunkStrict(makeStream([])))
+    expect(error).toBeInstanceOf(Error)
+    expect(error.message).toBe('stream has ended without data')
+    expect(error.chunk).toEqual(undefined)
+  })
+
+  it('throws if stream ends with not enough data', async () => {
+    const error = await rejectionOf(readChunkStrict(makeStream(['foo', 'bar']), 10))
+    expect(error).toBeInstanceOf(Error)
+    expect(error.message).toBe('stream has ended with not enough data')
+    expect(error.chunk).toEqual(Buffer.from('foobar'))
+  })
+})

@vates/read-chunk/package.json

@@ -19,7 +19,7 @@
     "type": "git",
     "url": "https://github.com/vatesfr/xen-orchestra.git"
   },
-  "version": "0.1.2",
+  "version": "1.0.0",
   "engines": {
     "node": ">=8.10"
   },

@vates/toggle-scripts/index.js

@@ -1,5 +1,7 @@
 #!/usr/bin/env node
 
+'use strict'
+
 const fs = require('fs')
 
 const mapKeys = (object, iteratee) => {

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

@@ -1,3 +1,5 @@
+'use strict'
+
 const wrapCall = (fn, arg, thisArg) => {
   try {
     return Promise.resolve(fn.call(thisArg, arg))

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

@@ -1,3 +1,5 @@
+'use strict'
+
 /* eslint-env jest */
 
 const { asyncMapSettled } = require('./')

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

@@ -1,3 +1,5 @@
+'use strict'
+
 // type MaybePromise<T> = Promise<T> | T
 //
 // declare export function asyncMap<T1, T2>(

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

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

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

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

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

@@ -1,12 +1,14 @@
-import assert from 'assert'
-import hash from 'object-hash'
-import { createLogger } from '@xen-orchestra/log'
-import { decorateWith } from '@vates/decorate-with'
-import { defer } from 'golike-defer'
+'use strict'
+
+const assert = require('assert')
+const hash = require('object-hash')
+const { createLogger } = require('@xen-orchestra/log')
+const { decorateClass } = require('@vates/decorate-with')
+const { defer } = require('golike-defer')
 
 const log = createLogger('xo:audit-core')
 
-export class Storage {
+exports.Storage = class Storage {
   constructor() {
     this._lock = Promise.resolve()
   }
@@ -29,7 +31,7 @@ const ID_TO_ALGORITHM = {
   5: 'sha256',
 }
 
-export class AlteredRecordError extends Error {
+class AlteredRecordError extends Error {
   constructor(id, nValid, record) {
     super('altered record')
 
@@ -38,8 +40,9 @@ export class AlteredRecordError extends Error {
     this.record = record
   }
 }
+exports.AlteredRecordError = AlteredRecordError
 
-export class MissingRecordError extends Error {
+class MissingRecordError extends Error {
   constructor(id, nValid) {
     super('missing record')
 
@@ -47,8 +50,10 @@ export class MissingRecordError extends Error {
     this.nValid = nValid
   }
 }
+exports.MissingRecordError = MissingRecordError
 
-export const NULL_ID = 'nullId'
+const NULL_ID = 'nullId'
+exports.NULL_ID = NULL_ID
 
 const HASH_ALGORITHM_ID = '5'
 const createHash = (data, algorithmId = HASH_ALGORITHM_ID) =>
@@ -57,13 +62,12 @@ const createHash = (data, algorithmId = HASH_ALGORITHM_ID) =>
     excludeKeys: key => key === 'id',
   })}`
 
-export class AuditCore {
+class AuditCore {
   constructor(storage) {
     assert.notStrictEqual(storage, undefined)
     this._storage = storage
   }
 
-  @decorateWith(defer)
   async add($defer, subject, event, data) {
     const time = Date.now()
     $defer(await this._storage.acquireLock())
@@ -148,7 +152,6 @@ export class AuditCore {
     }
   }
 
-  @decorateWith(defer)
   async deleteRangeAndRewrite($defer, newest, oldest) {
     assert.notStrictEqual(newest, undefined)
     assert.notStrictEqual(oldest, undefined)
@@ -189,3 +192,9 @@ export class AuditCore {
     }
   }
 }
+exports.AuditCore = AuditCore
+
+decorateClass(AuditCore, {
+  add: defer,
+  deleteRangeAndRewrite: defer,
+})

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

@@ -1,6 +1,9 @@
-/* eslint-env jest */
+'use strict'
 
-import { AlteredRecordError, AuditCore, MissingRecordError, NULL_ID, Storage } from '.'
+const assert = require('assert/strict')
+const { afterEach, describe, it } = require('tap').mocha
+
+const { AlteredRecordError, AuditCore, MissingRecordError, NULL_ID, Storage } = require('.')
 
 const asyncIteratorToArray = async asyncIterator => {
   const array = []
@@ -72,7 +75,7 @@ const auditCore = new AuditCore(db)
 const storeAuditRecords = async () => {
   await Promise.all(DATA.map(data => auditCore.add(...data)))
   const records = await asyncIteratorToArray(auditCore.getFrom())
-  expect(records.length).toBe(DATA.length)
+  assert.equal(records.length, DATA.length)
   return records
 }
 
@@ -83,10 +86,11 @@ describe('auditCore', () => {
     const [newestRecord, deletedRecord] = await storeAuditRecords()
 
     const nValidRecords = await auditCore.checkIntegrity(NULL_ID, newestRecord.id)
-    expect(nValidRecords).toBe(DATA.length)
+    assert.equal(nValidRecords, DATA.length)
 
     await db.del(deletedRecord.id)
-    await expect(auditCore.checkIntegrity(NULL_ID, newestRecord.id)).rejects.toEqual(
+    await assert.rejects(
+      auditCore.checkIntegrity(NULL_ID, newestRecord.id),
       new MissingRecordError(deletedRecord.id, 1)
     )
   })
@@ -97,7 +101,8 @@ describe('auditCore', () => {
     alteredRecord.event = ''
     await db.put(alteredRecord)
 
-    await expect(auditCore.checkIntegrity(NULL_ID, newestRecord.id)).rejects.toEqual(
+    await assert.rejects(
+      auditCore.checkIntegrity(NULL_ID, newestRecord.id),
       new AlteredRecordError(alteredRecord.id, 1, alteredRecord)
     )
   })
@@ -107,8 +112,8 @@ describe('auditCore', () => {
 
     await auditCore.deleteFrom(secondRecord.id)
 
-    expect(await db.get(firstRecord.id)).toBe(undefined)
-    expect(await db.get(secondRecord.id)).toBe(undefined)
+    assert.equal(await db.get(firstRecord.id), undefined)
+    assert.equal(await db.get(secondRecord.id), undefined)
 
     await auditCore.checkIntegrity(secondRecord.id, thirdRecord.id)
   })

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

@@ -9,28 +9,14 @@
   },
   "version": "0.2.0",
   "engines": {
-    "node": ">=10"
+    "node": ">=14"
   },
-  "main": "dist/",
   "scripts": {
-    "build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
-    "dev": "cross-env NODE_ENV=development babel --watch --source-maps --out-dir=dist/ src/",
     "postversion": "npm publish --access public",
-    "prebuild": "rimraf dist/",
-    "predev": "yarn run prebuild",
-    "prepublishOnly": "yarn run build"
-  },
-  "devDependencies": {
-    "@babel/cli": "^7.7.4",
-    "@babel/core": "^7.7.4",
-    "@babel/plugin-proposal-decorators": "^7.8.0",
-    "@babel/plugin-proposal-nullish-coalescing-operator": "^7.8.0",
-    "@babel/preset-env": "^7.7.4",
-    "cross-env": "^7.0.2",
-    "rimraf": "^3.0.0"
+    "test": "tap --lines 67 --functions 92 --branches 52 --statements 67"
   },
   "dependencies": {
-    "@vates/decorate-with": "^1.0.0",
+    "@vates/decorate-with": "^2.0.0",
     "@xen-orchestra/log": "^0.3.0",
     "golike-defer": "^0.5.1",
     "object-hash": "^2.0.1"
@@ -40,5 +26,8 @@
   "author": {
     "name": "Vates SAS",
     "url": "https://vates.fr"
+  },
+  "devDependencies": {
+    "tap": "^16.0.1"
   }
 }

@xen-orchestra/babel-config/package.json

@@ -10,7 +10,7 @@
     "url": "https://github.com/vatesfr/xen-orchestra.git"
   },
   "engines": {
-    "node": ">=6"
+    "node": ">=8.3"
   },
   "license": "AGPL-3.0-or-later",
   "author": {

@xen-orchestra/backups-cli/_composeCommands.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const getopts = require('getopts')
 
 const { version } = require('./package.json')

@xen-orchestra/backups-cli/_fs.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const { dirname } = require('path')
 
 const fs = require('promise-toolbox/promisifyAll')(require('fs'))

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

@@ -1,4 +1,4 @@
-#!/usr/bin/env node
+'use strict'
 
 // -----------------------------------------------------------------------------
 
@@ -26,7 +26,13 @@ module.exports = async function main(args) {
   await asyncMap(_, async vmDir => {
     vmDir = resolve(vmDir)
     try {
-      await adapter.cleanVm(vmDir, { fixMetadata: fix, remove, merge, onLog: (...args) => console.warn(...args) })
+      await adapter.cleanVm(vmDir, {
+        fixMetadata: fix,
+        remove,
+        merge,
+        logInfo: (...args) => console.log(...args),
+        logWarn: (...args) => console.warn(...args),
+      })
     } catch (error) {
       console.error('adapter.cleanVm', vmDir, error)
     }

@xen-orchestra/backups-cli/commands/create-symlink-index.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const filenamify = require('filenamify')
 const get = require('lodash/get')
 const { asyncMap } = require('@xen-orchestra/async-map')

@xen-orchestra/backups-cli/commands/info.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const groupBy = require('lodash/groupBy')
 const { asyncMap } = require('@xen-orchestra/async-map')
 const { createHash } = require('crypto')

@xen-orchestra/backups-cli/index.js

@@ -1,5 +1,7 @@
 #!/usr/bin/env node
 
+'use strict'
+
 require('./_composeCommands')({
   'clean-vms': {
     get main() {

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

@@ -7,12 +7,12 @@
   "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
   "dependencies": {
     "@xen-orchestra/async-map": "^0.1.2",
-    "@xen-orchestra/backups": "^0.18.3",
-    "@xen-orchestra/fs": "^0.19.3",
+    "@xen-orchestra/backups": "^0.27.4",
+    "@xen-orchestra/fs": "^3.1.0",
     "filenamify": "^4.1.0",
     "getopts": "^2.2.5",
     "lodash": "^4.17.15",
-    "promise-toolbox": "^0.20.0"
+    "promise-toolbox": "^0.21.0"
   },
   "engines": {
     "node": ">=7.10.1"
@@ -27,7 +27,7 @@
   "scripts": {
     "postversion": "npm publish --access public"
   },
-  "version": "0.6.1",
+  "version": "0.7.7",
   "license": "AGPL-3.0-or-later",
   "author": {
     "name": "Vates SAS",

@xen-orchestra/backups/Backup.js

@@ -1,10 +1,12 @@
+'use strict'
+
 const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
-const Disposable = require('promise-toolbox/Disposable.js')
-const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
+const Disposable = require('promise-toolbox/Disposable')
+const ignoreErrors = require('promise-toolbox/ignoreErrors')
 const { compileTemplate } = require('@xen-orchestra/template')
 const { limitConcurrency } = require('limit-concurrency-decorator')
 
-const { extractIdsFromSimplePattern } = require('./_extractIdsFromSimplePattern.js')
+const { extractIdsFromSimplePattern } = require('./extractIdsFromSimplePattern.js')
 const { PoolMetadataBackup } = require('./_PoolMetadataBackup.js')
 const { Task } = require('./Task.js')
 const { VmBackup } = require('./_VmBackup.js')
@@ -22,6 +24,34 @@ const getAdaptersByRemote = adapters => {
 
 const runTask = (...args) => Task.run(...args).catch(noop) // errors are handled by logs
 
+const DEFAULT_SETTINGS = {
+  reportWhen: 'failure',
+}
+
+const DEFAULT_VM_SETTINGS = {
+  bypassVdiChainsCheck: false,
+  checkpointSnapshot: false,
+  concurrency: 2,
+  copyRetention: 0,
+  deleteFirst: false,
+  exportRetention: 0,
+  fullInterval: 0,
+  healthCheckSr: undefined,
+  healthCheckVmsWithTags: [],
+  maxMergedDeltasPerRun: 2,
+  offlineBackup: false,
+  offlineSnapshot: false,
+  snapshotRetention: 0,
+  timeout: 0,
+  unconditionalSnapshot: false,
+  vmTimeout: 0,
+}
+
+const DEFAULT_METADATA_SETTINGS = {
+  retentionPoolMetadata: 0,
+  retentionXoMetadata: 0,
+}
+
 exports.Backup = class Backup {
   constructor({ config, getAdapter, getConnectedRecord, job, schedule }) {
     this._config = config
@@ -40,17 +70,22 @@ exports.Backup = class Backup {
       '{job.name}': job.name,
       '{vm.name_label}': vm => vm.name_label,
     })
-  }
 
-  run() {
-    const type = this._job.type
+    const { type } = job
+    const baseSettings = { ...DEFAULT_SETTINGS }
     if (type === 'backup') {
-      return this._runVmBackup()
+      Object.assign(baseSettings, DEFAULT_VM_SETTINGS, config.defaultSettings, config.vm?.defaultSettings)
+      this.run = this._runVmBackup
     } else if (type === 'metadataBackup') {
-      return this._runMetadataBackup()
+      Object.assign(baseSettings, DEFAULT_METADATA_SETTINGS, config.defaultSettings, config.metadata?.defaultSettings)
+      this.run = this._runMetadataBackup
     } else {
       throw new Error(`No runner for the backup type ${type}`)
     }
+    Object.assign(baseSettings, job.settings[''])
+
+    this._baseSettings = baseSettings
+    this._settings = { ...baseSettings, ...job.settings[schedule.id] }
   }
 
   async _runMetadataBackup() {
@@ -62,13 +97,6 @@ exports.Backup = class Backup {
     }
 
     const config = this._config
-    const settings = {
-      ...config.defaultSettings,
-      ...config.metadata.defaultSettings,
-      ...job.settings[''],
-      ...job.settings[schedule.id],
-    }
-
     const poolIds = extractIdsFromSimplePattern(job.pools)
     const isEmptyPools = poolIds.length === 0
     const isXoMetadata = job.xoMetadata !== undefined
@@ -76,6 +104,8 @@ exports.Backup = class Backup {
       throw new Error('no metadata mode found')
     }
 
+    const settings = this._settings
+
     const { retentionPoolMetadata, retentionXoMetadata } = settings
 
     if (
@@ -187,14 +217,7 @@ exports.Backup = class Backup {
     const schedule = this._schedule
 
     const config = this._config
-    const { settings } = job
-    const scheduleSettings = {
-      ...config.defaultSettings,
-      ...config.vm.defaultSettings,
-      ...settings[''],
-      ...settings[schedule.id],
-    }
-
+    const settings = this._settings
     await Disposable.use(
       Disposable.all(
         extractIdsFromSimplePattern(job.srs).map(id =>
@@ -222,14 +245,15 @@ exports.Backup = class Backup {
           })
         )
       ),
-      async (srs, remoteAdapters) => {
+      () => (settings.healthCheckSr !== undefined ? this._getRecord('SR', settings.healthCheckSr) : undefined),
+      async (srs, remoteAdapters, healthCheckSr) => {
         // remove adapters that failed (already handled)
         remoteAdapters = remoteAdapters.filter(_ => _ !== undefined)
 
         // remove srs that failed (already handled)
         srs = srs.filter(_ => _ !== undefined)
 
-        if (remoteAdapters.length === 0 && srs.length === 0 && scheduleSettings.snapshotRetention === 0) {
+        if (remoteAdapters.length === 0 && srs.length === 0 && settings.snapshotRetention === 0) {
           return
         }
 
@@ -239,23 +263,27 @@ exports.Backup = class Backup {
 
         remoteAdapters = getAdaptersByRemote(remoteAdapters)
 
+        const allSettings = this._job.settings
+        const baseSettings = this._baseSettings
+
         const handleVm = vmUuid =>
           runTask({ name: 'backup VM', data: { type: 'VM', id: vmUuid } }, () =>
             Disposable.use(this._getRecord('VM', vmUuid), vm =>
               new VmBackup({
+                baseSettings,
                 config,
                 getSnapshotNameLabel,
+                healthCheckSr,
                 job,
-                // remotes,
                 remoteAdapters,
                 schedule,
-                settings: { ...scheduleSettings, ...settings[vmUuid] },
+                settings: { ...settings, ...allSettings[vm.uuid] },
                 srs,
                 vm,
               }).run()
             )
           )
-        const { concurrency } = scheduleSettings
+        const { concurrency } = settings
         await asyncMapSettled(vmIds, concurrency === 0 ? handleVm : limitConcurrency(concurrency)(handleVm))
       }
     )

@xen-orchestra/backups/DurablePartition.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const { asyncMap } = require('@xen-orchestra/async-map')
 
 exports.DurablePartition = class DurablePartition {

@xen-orchestra/backups/HealthCheckVmBackup.js

@@ -0,0 +1,64 @@
+'use strict'
+
+const { Task } = require('./Task')
+
+exports.HealthCheckVmBackup = class HealthCheckVmBackup {
+  #xapi
+  #restoredVm
+
+  constructor({ restoredVm, xapi }) {
+    this.#restoredVm = restoredVm
+    this.#xapi = xapi
+  }
+
+  async run() {
+    return Task.run(
+      {
+        name: 'vmstart',
+      },
+      async () => {
+        let restoredVm = this.#restoredVm
+        const xapi = this.#xapi
+        const restoredId = restoredVm.uuid
+
+        // remove vifs
+        await Promise.all(restoredVm.$VIFs.map(vif => xapi.callAsync('VIF.destroy', vif.$ref)))
+
+        const start = new Date()
+        // start Vm
+
+        await xapi.callAsync(
+          'VM.start',
+          restoredVm.$ref,
+          false, // Start paused?
+          false // Skip pre-boot checks?
+        )
+        const started = new Date()
+        const timeout = 10 * 60 * 1000
+        const startDuration = started - start
+
+        let remainingTimeout = timeout - startDuration
+
+        if (remainingTimeout < 0) {
+          throw new Error(`VM ${restoredId} not started after ${timeout / 1000} second`)
+        }
+
+        // wait for the 'Running' event to be really stored in local xapi object cache
+        restoredVm = await xapi.waitObjectState(restoredVm.$ref, vm => vm.power_state === 'Running', {
+          timeout: remainingTimeout,
+        })
+
+        const running = new Date()
+        remainingTimeout -= running - started
+
+        if (remainingTimeout < 0) {
+          throw new Error(`local xapi  did not get Runnig state for VM ${restoredId} after ${timeout / 1000} second`)
+        }
+        // wait for the guest tool version to be defined
+        await xapi.waitObjectState(restoredVm.guest_metrics, gm => gm?.PV_drivers_version?.major !== undefined, {
+          timeout: remainingTimeout,
+        })
+      }
+    )
+  }
+}

@xen-orchestra/backups/ImportVmBackup.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const assert = require('assert')
 
 const { formatFilenameDate } = require('./_filenameDate.js')
@@ -6,9 +8,9 @@ const { Task } = require('./Task.js')
 const { watchStreamSize } = require('./_watchStreamSize.js')
 
 exports.ImportVmBackup = class ImportVmBackup {
-  constructor({ adapter, metadata, srUuid, xapi, settings: { newMacAddresses } = {} }) {
+  constructor({ adapter, metadata, srUuid, xapi, settings: { newMacAddresses, mapVdisSrs = {} } = {} }) {
     this._adapter = adapter
-    this._importDeltaVmSettings = { newMacAddresses }
+    this._importDeltaVmSettings = { newMacAddresses, mapVdisSrs }
     this._metadata = metadata
     this._srUuid = srUuid
     this._xapi = xapi
@@ -28,7 +30,12 @@ exports.ImportVmBackup = class ImportVmBackup {
     } else {
       assert.strictEqual(metadata.mode, 'delta')
 
-      backup = await adapter.readDeltaVmBackup(metadata)
+      const ignoredVdis = new Set(
+        Object.entries(this._importDeltaVmSettings.mapVdisSrs)
+          .filter(([_, srUuid]) => srUuid === null)
+          .map(([vdiUuid]) => vdiUuid)
+      )
+      backup = await adapter.readDeltaVmBackup(metadata, ignoredVdis)
       Object.values(backup.streams).forEach(stream => watchStreamSize(stream, sizeContainer))
     }
 

@xen-orchestra/backups/RemoteAdapter.js

@@ -1,33 +1,42 @@
+'use strict'
+
 const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
-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 { synchronized } = require('decorator-synchronized')
+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 groupBy = require('lodash/groupBy.js')
+const pickBy = require('lodash/pickBy.js')
 const { dirname, join, normalize, resolve } = require('path')
 const { createLogger } = require('@xen-orchestra/log')
-const { Constants, createVhdDirectoryFromStream, openVhd, VhdAbstract, VhdDirectory, VhdSynthetic } = require('vhd-lib')
+const { createVhdDirectoryFromStream, openVhd, VhdAbstract, VhdDirectory, VhdSynthetic } = require('vhd-lib')
 const { deduped } = require('@vates/disposable/deduped.js')
+const { decorateMethodsWith } = require('@vates/decorate-with')
+const { compose } = require('@vates/compose')
 const { execFile } = require('child_process')
-const { readdir, stat } = require('fs-extra')
+const { readdir, lstat } = require('fs-extra')
 const { v4: uuidv4 } = require('uuid')
 const { ZipFile } = require('yazl')
+const zlib = require('zlib')
 
 const { BACKUP_DIR } = require('./_getVmBackupDir.js')
 const { cleanVm } = require('./_cleanVm.js')
+const { formatFilenameDate } = require('./_filenameDate.js')
 const { getTmpDir } = require('./_getTmpDir.js')
 const { isMetadataFile } = require('./_backupType.js')
 const { isValidXva } = require('./_isValidXva.js')
 const { listPartitions, LVM_PARTITION_TYPE } = require('./_listPartitions.js')
 const { lvs, pvs } = require('./_lvm.js')
-
+const { asyncEach } = require('@vates/async-each')
+const { mount } = require('@vates/fuse-vhd')
 const DIR_XO_CONFIG_BACKUPS = 'xo-config-backups'
 exports.DIR_XO_CONFIG_BACKUPS = DIR_XO_CONFIG_BACKUPS
 
 const DIR_XO_POOL_METADATA_BACKUPS = 'xo-pool-metadata-backups'
 exports.DIR_XO_POOL_METADATA_BACKUPS = DIR_XO_POOL_METADATA_BACKUPS
 
-const { warn } = createLogger('xo:backups:RemoteAdapter')
+const { debug, warn } = createLogger('xo:backups:RemoteAdapter')
 
 const compareTimestamp = (a, b) => a.timestamp - b.timestamp
 
@@ -37,16 +46,13 @@ const resolveRelativeFromFile = (file, path) => resolve('/', dirname(file), path
 
 const resolveSubpath = (root, path) => resolve(root, `.${resolve('/', path)}`)
 
-const RE_VHDI = /^vhdi(\d+)$/
-
 async function addDirectory(files, realPath, metadataPath) {
-  try {
-    const subFiles = await readdir(realPath)
-    await asyncMap(subFiles, file => addDirectory(files, realPath + '/' + file, metadataPath + '/' + file))
-  } catch (error) {
-    if (error == null || error.code !== 'ENOTDIR') {
-      throw error
-    }
+  const stats = await lstat(realPath)
+  if (stats.isDirectory()) {
+    await asyncMap(await readdir(realPath), file =>
+      addDirectory(files, realPath + '/' + file, metadataPath + '/' + file)
+    )
+  } else if (stats.isFile()) {
     files.push({
       realPath,
       metadataPath,
@@ -68,11 +74,16 @@ const debounceResourceFactory = factory =>
   }
 
 class RemoteAdapter {
-  constructor(handler, { debounceResource = res => res, dirMode, vhdDirectoryCompression } = {}) {
+  constructor(
+    handler,
+    { debounceResource = res => res, dirMode, vhdDirectoryCompression, useGetDiskLegacy = false } = {}
+  ) {
     this._debounceResource = debounceResource
     this._dirMode = dirMode
     this._handler = handler
     this._vhdDirectoryCompression = vhdDirectoryCompression
+    this._readCacheListVmBackups = synchronized.withKey()(this._readCacheListVmBackups)
+    this._useGetDiskLegacy = useGetDiskLegacy
   }
 
   get handler() {
@@ -88,9 +99,6 @@ class RemoteAdapter {
     return partition
   }
 
-  _getLvmLogicalVolumes = Disposable.factory(this._getLvmLogicalVolumes)
-  _getLvmLogicalVolumes = deduped(this._getLvmLogicalVolumes, (devicePath, pvId, vgName) => [devicePath, pvId, vgName])
-  _getLvmLogicalVolumes = debounceResourceFactory(this._getLvmLogicalVolumes)
   async *_getLvmLogicalVolumes(devicePath, pvId, vgName) {
     yield this._getLvmPhysicalVolume(devicePath, pvId && (await this._findPartition(devicePath, pvId)))
 
@@ -102,9 +110,6 @@ class RemoteAdapter {
     }
   }
 
-  _getLvmPhysicalVolume = Disposable.factory(this._getLvmPhysicalVolume)
-  _getLvmPhysicalVolume = deduped(this._getLvmPhysicalVolume, (devicePath, partition) => [devicePath, partition?.id])
-  _getLvmPhysicalVolume = debounceResourceFactory(this._getLvmPhysicalVolume)
   async *_getLvmPhysicalVolume(devicePath, partition) {
     const args = []
     if (partition !== undefined) {
@@ -125,11 +130,10 @@ class RemoteAdapter {
     }
   }
 
-  _getPartition = Disposable.factory(this._getPartition)
-  _getPartition = deduped(this._getPartition, (devicePath, partition) => [devicePath, partition?.id])
-  _getPartition = debounceResourceFactory(this._getPartition)
   async *_getPartition(devicePath, partition) {
-    const options = ['loop', 'ro']
+    // the norecovery option is necessary because if the partition is dirty,
+    // mount will try to fix it which is impossible if because the device is read-only
+    const options = ['loop', 'ro', 'norecovery']
 
     if (partition !== undefined) {
       const { size, start } = partition
@@ -180,7 +184,6 @@ class RemoteAdapter {
     })
   }
 
-  _usePartitionFiles = Disposable.factory(this._usePartitionFiles)
   async *_usePartitionFiles(diskId, partitionId, paths) {
     const path = yield this.getPartition(diskId, partitionId)
 
@@ -227,11 +230,30 @@ class RemoteAdapter {
     return promise
   }
 
+  #removeVmBackupsFromCache(backups) {
+    for (const [dir, filenames] of Object.entries(
+      groupBy(
+        backups.map(_ => _._filename),
+        dirname
+      )
+    )) {
+      // detached async action, will not reject
+      this._updateCache(dir + '/cache.json.gz', backups => {
+        for (const filename of filenames) {
+          debug('removing cache entry', { entry: filename })
+          delete backups[filename]
+        }
+      })
+    }
+  }
+
   async deleteDeltaVmBackups(backups) {
     const handler = this._handler
 
-    // unused VHDs will be detected by `cleanVm`
-    await asyncMapSettled(backups, ({ _filename }) => VhdAbstract.unlink(handler, _filename))
+    // this will delete the json, unused VHDs will be detected by `cleanVm`
+    await asyncMapSettled(backups, ({ _filename }) => handler.unlink(_filename))
+
+    this.#removeVmBackupsFromCache(backups)
   }
 
   async deleteMetadataBackup(backupId) {
@@ -259,6 +281,8 @@ class RemoteAdapter {
     await asyncMapSettled(backups, ({ _filename, xva }) =>
       Promise.all([handler.unlink(_filename), handler.unlink(resolveRelativeFromFile(_filename, xva))])
     )
+
+    this.#removeVmBackupsFromCache(backups)
   }
 
   deleteVmBackup(file) {
@@ -266,7 +290,8 @@ class RemoteAdapter {
   }
 
   async deleteVmBackups(files) {
-    const { delta, full, ...others } = groupBy(await asyncMap(files, file => this.readVmBackupMetadata(file)), 'mode')
+    const metadatas = await asyncMap(files, file => this.readVmBackupMetadata(file))
+    const { delta, full, ...others } = groupBy(metadatas, 'mode')
 
     const unsupportedModes = Object.keys(others)
     if (unsupportedModes.length !== 0) {
@@ -277,6 +302,14 @@ class RemoteAdapter {
       delta !== undefined && this.deleteDeltaVmBackups(delta),
       full !== undefined && this.deleteFullVmBackups(full),
     ])
+
+    await asyncMap(new Set(files.map(file => dirname(file))), dir =>
+      // - don't merge in main process, unused VHDs will be merged in the next backup run
+      // - don't error in case this fails:
+      //   - if lock is already being held, a backup is running and cleanVm will be ran at the end
+      //   - otherwise, there is nothing more we can do, orphan file will be cleaned in the future
+      this.cleanVm(dir, { remove: true, logWarn: warn }).catch(noop)
+    )
   }
 
   #getCompressionType() {
@@ -284,17 +317,15 @@ class RemoteAdapter {
   }
 
   #useVhdDirectory() {
-    return this.handler.type === 's3'
+    return this.handler.useVhdDirectory()
   }
 
   #useAlias() {
     return this.#useVhdDirectory()
   }
 
-  getDisk = Disposable.factory(this.getDisk)
-  getDisk = deduped(this.getDisk, diskId => [diskId])
-  getDisk = debounceResourceFactory(this.getDisk)
-  async *getDisk(diskId) {
+  async *#getDiskLegacy(diskId) {
+    const RE_VHDI = /^vhdi(\d+)$/
     const handler = this._handler
 
     const diskPath = handler._getFilePath('/' + diskId)
@@ -324,13 +355,36 @@ class RemoteAdapter {
     }
   }
 
+  async *getDisk(diskId) {
+    if (this._useGetDiskLegacy) {
+      yield* this.#getDiskLegacy(diskId)
+      return
+    }
+    const handler = this._handler
+    // this is a disposable
+    const mountDir = yield getTmpDir()
+
+    try {
+      // this is also a disposable
+      yield mount(handler, diskId, mountDir)
+    } catch (error) {
+      // fallback in case of missing dependency
+      if (error.code === 'MODULE_NOT_FOUND') {
+        yield* this.#getDiskLegacy(diskId)
+      } else {
+        throw error
+      }
+    }
+    // this will yield disk path to caller
+    yield `${mountDir}/vhd0`
+  }
+
   // partitionId values:
   //
   // - undefined: raw disk
   // - `<partitionId>`: partitioned disk
   // - `<pvId>/<vgName>/<lvName>`: LVM on a partitioned disk
   // - `/<vgName>/lvName>`: LVM on a raw disk
-  getPartition = Disposable.factory(this.getPartition)
   async *getPartition(diskId, partitionId) {
     const devicePath = yield this.getDisk(diskId)
     if (partitionId === undefined) {
@@ -359,9 +413,14 @@ class RemoteAdapter {
     const handler = this._handler
 
     const backups = { __proto__: null }
-    await asyncMap(await handler.list(BACKUP_DIR), async vmUuid => {
-      const vmBackups = await this.listVmBackups(vmUuid)
-      backups[vmUuid] = vmBackups
+    await asyncMap(await handler.list(BACKUP_DIR), async entry => {
+      // ignore hidden and lock files
+      if (entry[0] !== '.' && !entry.endsWith('.lock')) {
+        const vmBackups = await this.listVmBackups(entry)
+        if (vmBackups.length !== 0) {
+          backups[entry] = vmBackups
+        }
+      }
     })
 
     return backups
@@ -370,18 +429,25 @@ class RemoteAdapter {
   listPartitionFiles(diskId, partitionId, path) {
     return Disposable.use(this.getPartition(diskId, partitionId), async rootPath => {
       path = resolveSubpath(rootPath, path)
-
       const entriesMap = {}
-      await asyncMap(await readdir(path), async name => {
-        try {
-          const stats = await stat(`${path}/${name}`)
-          entriesMap[stats.isDirectory() ? `${name}/` : name] = {}
-        } catch (error) {
-          if (error == null || error.code !== 'ENOENT') {
-            throw error
+      await asyncEach(
+        await readdir(path),
+        async name => {
+          try {
+            const stats = await lstat(`${path}/${name}`)
+            if (stats.isDirectory()) {
+              entriesMap[name + '/'] = {}
+            } else if (stats.isFile()) {
+              entriesMap[name] = {}
+            }
+          } catch (error) {
+            if (error == null || error.code !== 'ENOENT') {
+              throw error
+            }
           }
-        }
-      })
+        },
+        { concurrency: 1 }
+      )
 
       return entriesMap
     })
@@ -446,34 +512,114 @@ class RemoteAdapter {
     return backupsByPool
   }
 
-  async listVmBackups(vmUuid, predicate) {
+  #getVmBackupsCache(vmUuid) {
+    return `${BACKUP_DIR}/${vmUuid}/cache.json.gz`
+  }
+
+  async #readCache(path) {
+    try {
+      return JSON.parse(await fromCallback(zlib.gunzip, await this.handler.readFile(path)))
+    } catch (error) {
+      if (error.code !== 'ENOENT') {
+        warn('#readCache', { error, path })
+      }
+    }
+  }
+
+  _updateCache = synchronized.withKey()(this._updateCache)
+  // eslint-disable-next-line no-dupe-class-members
+  async _updateCache(path, fn) {
+    const cache = await this.#readCache(path)
+    if (cache !== undefined) {
+      fn(cache)
+
+      await this.#writeCache(path, cache)
+    }
+  }
+
+  async #writeCache(path, data) {
+    try {
+      await this.handler.writeFile(path, await fromCallback(zlib.gzip, JSON.stringify(data)), { flags: 'w' })
+    } catch (error) {
+      warn('#writeCache', { error, path })
+    }
+  }
+
+  async invalidateVmBackupListCache(vmUuid) {
+    await this.handler.unlink(this.#getVmBackupsCache(vmUuid))
+  }
+
+  async #getCachabledDataListVmBackups(dir) {
+    debug('generating cache', { path: dir })
+
     const handler = this._handler
-    const backups = []
+    const backups = {}
 
     try {
-      const files = await handler.list(`${BACKUP_DIR}/${vmUuid}`, {
+      const files = await handler.list(dir, {
         filter: isMetadataFile,
         prependDir: true,
       })
       await asyncMap(files, async file => {
         try {
           const metadata = await this.readVmBackupMetadata(file)
-          if (predicate === undefined || predicate(metadata)) {
-            // inject an id usable by importVmBackupNg()
-            metadata.id = metadata._filename
-
-            backups.push(metadata)
-          }
+          // inject an id usable by importVmBackupNg()
+          metadata.id = metadata._filename
+          backups[file] = metadata
         } catch (error) {
-          warn(`listVmBackups ${file}`, { error })
+          warn(`can't read vm backup metadata`, { error, file, dir })
         }
       })
+      return backups
     } catch (error) {
       let code
       if (error == null || ((code = error.code) !== 'ENOENT' && code !== 'ENOTDIR')) {
         throw error
       }
     }
+  }
+
+  // use _ to mark this method as private by convention
+  // since we decorate it with synchronized.withKey in the constructor
+  // and # function are not writeable.
+  //
+  // read the list of backup of a Vm from cache
+  // if cache is missing  or broken  => regenerate it and return
+
+  async _readCacheListVmBackups(vmUuid) {
+    const path = this.#getVmBackupsCache(vmUuid)
+
+    const cache = await this.#readCache(path)
+    if (cache !== undefined) {
+      debug('found VM backups cache, using it', { path })
+      return cache
+    }
+
+    // nothing cached, or cache unreadable => regenerate it
+    const backups = await this.#getCachabledDataListVmBackups(`${BACKUP_DIR}/${vmUuid}`)
+    if (backups === undefined) {
+      return
+    }
+
+    // detached async action, will not reject
+    this.#writeCache(path, backups)
+
+    return backups
+  }
+
+  async listVmBackups(vmUuid, predicate) {
+    const backups = []
+    const cached = await this._readCacheListVmBackups(vmUuid)
+
+    if (cached === undefined) {
+      return []
+    }
+
+    Object.values(cached).forEach(metadata => {
+      if (predicate === undefined || predicate(metadata)) {
+        backups.push(metadata)
+      }
+    })
 
     return backups.sort(compareTimestamp)
   }
@@ -499,13 +645,35 @@ class RemoteAdapter {
     return backups.sort(compareTimestamp)
   }
 
-  async writeVhd(path, input, { checksum = true, validator = noop } = {}) {
+  async writeVmBackupMetadata(vmUuid, metadata) {
+    const path = `/${BACKUP_DIR}/${vmUuid}/${formatFilenameDate(metadata.timestamp)}.json`
+
+    await this.handler.outputFile(path, JSON.stringify(metadata), {
+      dirMode: this._dirMode,
+    })
+
+    // will not throw
+    this._updateCache(this.#getVmBackupsCache(vmUuid), backups => {
+      debug('adding cache entry', { entry: path })
+      backups[path] = {
+        ...metadata,
+
+        // these values are required in the cache
+        _filename: path,
+        id: path,
+      }
+    })
+
+    return path
+  }
+
+  async writeVhd(path, input, { checksum = true, validator = noop, writeBlockConcurrency } = {}) {
     const handler = this._handler
 
     if (this.#useVhdDirectory()) {
       const dataPath = `${dirname(path)}/data/${uuidv4()}.vhd`
       await createVhdDirectoryFromStream(handler, dataPath, input, {
-        concurrency: 16,
+        concurrency: writeBlockConcurrency,
         compression: this.#getCompressionType(),
         async validator() {
           await input.task
@@ -529,60 +697,42 @@ class RemoteAdapter {
     })
   }
 
-  async _createSyntheticStream(handler, paths) {
-    let disposableVhds = []
-
-    // if it's a path : open all hierarchy of parent
-    if (typeof paths === 'string') {
-      let vhd,
-        vhdPath = paths
-      do {
-        const disposable = await openVhd(handler, vhdPath)
-        vhd = disposable.value
-        disposableVhds.push(disposable)
-        vhdPath = resolveRelativeFromFile(vhdPath, vhd.header.parentUnicodeName)
-      } while (vhd.footer.diskType !== Constants.DISK_TYPES.DYNAMIC)
-    } else {
-      // only open the list of path given
-      disposableVhds = paths.map(path => openVhd(handler, path))
-    }
-
+  // open the  hierarchy of ancestors until we find a full one
+  async _createSyntheticStream(handler, path) {
+    const disposableSynthetic = await VhdSynthetic.fromVhdChain(handler, path)
     // I don't want the vhds to be disposed on return
     // but only when the stream is done ( or failed )
-    const disposables = await Disposable.all(disposableVhds)
-    const vhds = disposables.value
 
     let disposed = false
     const disposeOnce = async () => {
       if (!disposed) {
         disposed = true
-
         try {
-          await disposables.dispose()
+          await disposableSynthetic.dispose()
         } catch (error) {
-          warn('_createSyntheticStream: failed to dispose VHDs', { error })
+          warn('openVhd: failed to dispose VHDs', { error })
         }
       }
     }
-
-    const synthetic = new VhdSynthetic(vhds)
-    await synthetic.readHeaderAndFooter()
+    const synthetic = disposableSynthetic.value
     await synthetic.readBlockAllocationTable()
     const stream = await synthetic.stream()
+
     stream.on('end', disposeOnce)
     stream.on('close', disposeOnce)
     stream.on('error', disposeOnce)
     return stream
   }
 
-  async readDeltaVmBackup(metadata) {
+  async readDeltaVmBackup(metadata, ignoredVdis) {
     const handler = this._handler
-    const { vbds, vdis, vhds, vifs, vm } = metadata
+    const { vbds, vhds, vifs, vm } = metadata
     const dir = dirname(metadata._filename)
+    const vdis = ignoredVdis === undefined ? metadata.vdis : pickBy(metadata.vdis, vdi => !ignoredVdis.has(vdi.uuid))
 
     const streams = {}
-    await asyncMapSettled(Object.keys(vdis), async id => {
-      streams[`${id}.vhd`] = await this._createSyntheticStream(handler, join(dir, vhds[id]))
+    await asyncMapSettled(Object.keys(vdis), async ref => {
+      streams[`${ref}.vhd`] = await this._createSyntheticStream(handler, join(dir, vhds[ref]))
     })
 
     return {
@@ -600,7 +750,10 @@ class RemoteAdapter {
   }
 
   async readVmBackupMetadata(path) {
-    return Object.defineProperty(JSON.parse(await this._handler.readFile(path)), '_filename', { value: path })
+    // _filename is a private field used to compute the backup id
+    //
+    // it's enumerable to make it cacheable
+    return { ...JSON.parse(await this._handler.readFile(path)), _filename: path }
   }
 }
 
@@ -615,4 +768,30 @@ Object.assign(RemoteAdapter.prototype, {
   isValidXva,
 })
 
+decorateMethodsWith(RemoteAdapter, {
+  _getLvmLogicalVolumes: compose([
+    Disposable.factory,
+    [deduped, (devicePath, pvId, vgName) => [devicePath, pvId, vgName]],
+    debounceResourceFactory,
+  ]),
+
+  _getLvmPhysicalVolume: compose([
+    Disposable.factory,
+    [deduped, (devicePath, partition) => [devicePath, partition?.id]],
+    debounceResourceFactory,
+  ]),
+
+  _getPartition: compose([
+    Disposable.factory,
+    [deduped, (devicePath, partition) => [devicePath, partition?.id]],
+    debounceResourceFactory,
+  ]),
+
+  _usePartitionFiles: Disposable.factory,
+
+  getDisk: compose([Disposable.factory, [deduped, diskId => [diskId]], debounceResourceFactory]),
+
+  getPartition: Disposable.factory,
+})
+
 exports.RemoteAdapter = RemoteAdapter

@xen-orchestra/backups/RestoreMetadataBackup.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter.js')
 const { PATH_DB_DUMP } = require('./_PoolMetadataBackup.js')
 

@xen-orchestra/backups/Task.js

@@ -1,12 +1,18 @@
-const CancelToken = require('promise-toolbox/CancelToken.js')
+'use strict'
+
+const CancelToken = require('promise-toolbox/CancelToken')
 const Zone = require('node-zone')
 
-const logAfterEnd = () => {
-  throw new Error('task has already ended')
+const logAfterEnd = log => {
+  const error = new Error('task has already ended')
+  error.log = log
+  throw error
 }
 
 const noop = Function.prototype
 
+const serializeErrors = errors => (Array.isArray(errors) ? errors.map(serializeError) : errors)
+
 // Create a serializable object from an error.
 //
 // Otherwise some fields might be non-enumerable and missing from logs.
@@ -15,6 +21,7 @@ const serializeError = error =>
     ? {
         ...error, // Copy enumerable properties.
         code: error.code,
+        errors: serializeErrors(error.errors), // supports AggregateError
         message: error.message,
         name: error.name,
         stack: error.stack,

@xen-orchestra/backups/_PoolMetadataBackup.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const { asyncMap } = require('@xen-orchestra/async-map')
 
 const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter.js')

@xen-orchestra/backups/_VmBackup.js

@@ -1,11 +1,14 @@
+'use strict'
+
 const assert = require('assert')
 const findLast = require('lodash/findLast.js')
 const groupBy = require('lodash/groupBy.js')
-const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
+const ignoreErrors = require('promise-toolbox/ignoreErrors')
 const keyBy = require('lodash/keyBy.js')
 const mapValues = require('lodash/mapValues.js')
 const { asyncMap } = require('@xen-orchestra/async-map')
 const { createLogger } = require('@xen-orchestra/log')
+const { decorateMethodsWith } = require('@vates/decorate-with')
 const { defer } = require('golike-defer')
 const { formatDateTime } = require('@xen-orchestra/xapi')
 
@@ -21,6 +24,13 @@ const { watchStreamSize } = require('./_watchStreamSize.js')
 
 const { debug, warn } = createLogger('xo:backups:VmBackup')
 
+class AggregateError extends Error {
+  constructor(errors, message) {
+    super(message)
+    this.errors = errors
+  }
+}
+
 const asyncEach = async (iterable, fn, thisArg = iterable) => {
   for (const item of iterable) {
     await fn.call(thisArg, item)
@@ -34,17 +44,28 @@ const forkDeltaExport = deltaExport =>
     },
   })
 
-exports.VmBackup = class VmBackup {
-  constructor({ config, getSnapshotNameLabel, job, remoteAdapters, remotes, schedule, settings, srs, vm }) {
-    if (vm.other_config['xo:backup:job'] === job.id) {
-      // otherwise replicated VMs would be matched and replicated again and again
+class VmBackup {
+  constructor({
+    config,
+    getSnapshotNameLabel,
+    healthCheckSr,
+    job,
+    remoteAdapters,
+    remotes,
+    schedule,
+    settings,
+    srs,
+    vm,
+  }) {
+    if (vm.other_config['xo:backup:job'] === job.id && 'start' in vm.blocked_operations) {
+      // don't match replicated VMs created by this very job otherwise they
+      // will be replicated again and again
       throw new Error('cannot backup a VM created by this very job')
     }
 
     this.config = config
     this.job = job
     this.remoteAdapters = remoteAdapters
-    this.remotes = remotes
     this.scheduleId = schedule.id
     this.timestamp = undefined
 
@@ -58,6 +79,7 @@ exports.VmBackup = class VmBackup {
     this._fullVdisRequired = undefined
     this._getSnapshotNameLabel = getSnapshotNameLabel
     this._isDelta = job.mode === 'delta'
+    this._healthCheckSr = healthCheckSr
     this._jobId = job.id
     this._jobSnapshots = undefined
     this._xapi = vm.$xapi
@@ -84,7 +106,6 @@ exports.VmBackup = class VmBackup {
         : [FullBackupWriter, FullReplicationWriter]
 
       const allSettings = job.settings
-
       Object.keys(remoteAdapters).forEach(remoteId => {
         const targetSettings = {
           ...settings,
@@ -107,33 +128,49 @@ exports.VmBackup = class VmBackup {
   }
 
   // calls fn for each function, warns of any errors, and throws only if there are no writers left
-  async _callWriters(fn, warnMessage, parallel = true) {
+  async _callWriters(fn, step, parallel = true) {
     const writers = this._writers
     const n = writers.size
     if (n === 0) {
       return
     }
-    if (n === 1) {
-      const [writer] = writers
+
+    async function callWriter(writer) {
+      const { name } = writer.constructor
       try {
+        debug('writer step starting', { step, writer: name })
         await fn(writer)
+        debug('writer step succeeded', { duration: step, writer: name })
       } catch (error) {
         writers.delete(writer)
+
+        warn('writer step failed', { error, step, writer: name })
+
+        // these two steps are the only one that are not already in their own sub tasks
+        if (step === 'writer.checkBaseVdis()' || step === 'writer.beforeBackup()') {
+          Task.warning(
+            `the writer ${name} has failed the step ${step} with error ${error.message}. It won't be used anymore in this job execution.`
+          )
+        }
+
         throw error
       }
-      return
+    }
+    if (n === 1) {
+      const [writer] = writers
+      return callWriter(writer)
     }
 
+    const errors = []
     await (parallel ? asyncMap : asyncEach)(writers, async function (writer) {
       try {
-        await fn(writer)
+        await callWriter(writer)
       } catch (error) {
-        this.delete(writer)
-        warn(warnMessage, { error, writer: writer.constructor.name })
+        errors.push(error)
       }
     })
     if (writers.size === 0) {
-      throw new Error('all targets have failed, step: ' + warnMessage)
+      throw new AggregateError(errors, 'all targets have failed, step: ' + step)
     }
   }
 
@@ -160,7 +197,10 @@ exports.VmBackup = class VmBackup {
     const settings = this._settings
 
     const doSnapshot =
-      this._isDelta || (!settings.offlineBackup && vm.power_state === 'Running') || settings.snapshotRetention !== 0
+      settings.unconditionalSnapshot ||
+      this._isDelta ||
+      (!settings.offlineBackup && vm.power_state === 'Running') ||
+      settings.snapshotRetention !== 0
     if (doSnapshot) {
       await Task.run({ name: 'snapshot' }, async () => {
         if (!settings.bypassVdiChainsCheck) {
@@ -168,7 +208,9 @@ exports.VmBackup = class VmBackup {
         }
 
         const snapshotRef = await vm[settings.checkpointSnapshot ? '$checkpoint' : '$snapshot']({
+          ignoreNobakVdis: true,
           name_label: this._getSnapshotNameLabel(vm),
+          unplugVusbs: true,
         })
         this.timestamp = Date.now()
 
@@ -290,22 +332,17 @@ exports.VmBackup = class VmBackup {
   }
 
   async _removeUnusedSnapshots() {
-    const jobSettings = this.job.settings
+    const allSettings = this.job.settings
+    const baseSettings = this._baseSettings
     const baseVmRef = this._baseVm?.$ref
-    const { config } = this
-    const baseSettings = {
-      ...config.defaultSettings,
-      ...config.metadata.defaultSettings,
-      ...jobSettings[''],
-    }
 
     const snapshotsPerSchedule = groupBy(this._jobSnapshots, _ => _.other_config['xo:backup:schedule'])
     const xapi = this._xapi
     await asyncMap(Object.entries(snapshotsPerSchedule), ([scheduleId, snapshots]) => {
       const settings = {
         ...baseSettings,
-        ...jobSettings[scheduleId],
-        ...jobSettings[this.vm.uuid],
+        ...allSettings[scheduleId],
+        ...allSettings[this.vm.uuid],
       }
       return asyncMap(getOldEntries(settings.snapshotRetention, snapshots), ({ $ref }) => {
         if ($ref !== baseVmRef) {
@@ -384,7 +421,24 @@ exports.VmBackup = class VmBackup {
     this._fullVdisRequired = fullVdisRequired
   }
 
-  run = defer(this.run)
+  async _healthCheck() {
+    const settings = this._settings
+
+    if (this._healthCheckSr === undefined) {
+      return
+    }
+
+    // check if current VM has tags
+    const { tags } = this.vm
+    const intersect = settings.healthCheckVmsWithTags.some(t => tags.includes(t))
+
+    if (settings.healthCheckVmsWithTags.length !== 0 && !intersect) {
+      return
+    }
+
+    await this._callWriters(writer => writer.healthCheck(this._healthCheckSr), 'writer.healthCheck()')
+  }
+
   async run($defer) {
     const settings = this._settings
     assert(
@@ -394,7 +448,9 @@ exports.VmBackup = class VmBackup {
 
     await this._callWriters(async writer => {
       await writer.beforeBackup()
-      $defer(() => writer.afterBackup())
+      $defer(async () => {
+        await writer.afterBackup()
+      })
     }, 'writer.beforeBackup()')
 
     await this._fetchJobSnapshots()
@@ -430,5 +486,11 @@ exports.VmBackup = class VmBackup {
       await this._fetchJobSnapshots()
       await this._removeUnusedSnapshots()
     }
+    await this._healthCheck()
   }
 }
+exports.VmBackup = VmBackup
+
+decorateMethodsWith(VmBackup, {
+  run: defer,
+})

@xen-orchestra/backups/_XoMetadataBackup.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const { asyncMap } = require('@xen-orchestra/async-map')
 
 const { DIR_XO_CONFIG_BACKUPS } = require('./RemoteAdapter.js')

@xen-orchestra/backups/_backupType.js

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

@xen-orchestra/backups/_backupWorker.js

@@ -1,11 +1,16 @@
+'use strict'
+
 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')
+require('@vates/cached-dns.lookup').createCachedLookup().patchGlobal()
+
+const Disposable = require('promise-toolbox/Disposable')
+const ignoreErrors = require('promise-toolbox/ignoreErrors')
 const { compose } = require('@vates/compose')
 const { createDebounceResource } = require('@vates/disposable/debounceResource.js')
+const { decorateMethodsWith } = require('@vates/decorate-with')
 const { deduped } = require('@vates/disposable/deduped.js')
 const { getHandler } = require('@xen-orchestra/fs')
 const { parseDuration } = require('@vates/parse-duration')
@@ -58,11 +63,6 @@ class BackupWorker {
     }).run()
   }
 
-  getAdapter = Disposable.factory(this.getAdapter)
-  getAdapter = deduped(this.getAdapter, remote => [remote.url])
-  getAdapter = compose(this.getAdapter, function (resource) {
-    return this.debounceResource(resource)
-  })
   async *getAdapter(remote) {
     const handler = getHandler(remote, this.#remoteOptions)
     await handler.sync()
@@ -77,11 +77,6 @@ class BackupWorker {
     }
   }
 
-  getXapi = Disposable.factory(this.getXapi)
-  getXapi = deduped(this.getXapi, ({ url }) => [url])
-  getXapi = compose(this.getXapi, function (resource) {
-    return this.debounceResource(resource)
-  })
   async *getXapi({ credentials: { username: user, password }, ...opts }) {
     const xapi = new Xapi({
       ...this.#xapiOptions,
@@ -103,6 +98,30 @@ class BackupWorker {
   }
 }
 
+decorateMethodsWith(BackupWorker, {
+  getAdapter: compose([
+    Disposable.factory,
+    [deduped, remote => [remote.url]],
+    [
+      compose,
+      function (resource) {
+        return this.debounceResource(resource)
+      },
+    ],
+  ]),
+
+  getXapi: compose([
+    Disposable.factory,
+    [deduped, xapi => [xapi.url]],
+    [
+      compose,
+      function (resource) {
+        return this.debounceResource(resource)
+      },
+    ],
+  ]),
+})
+
 // Received message:
 //
 // Message {

@xen-orchestra/backups/_cancelableMap.js

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

@xen-orchestra/backups/_cleanVm.integ.spec.js

@@ -1,14 +1,18 @@
+'use strict'
+
 /* eslint-env jest */
 
 const rimraf = require('rimraf')
 const tmp = require('tmp')
 const fs = require('fs-extra')
+const uuid = require('uuid')
 const { getHandler } = require('@xen-orchestra/fs')
 const { pFromCallback } = require('promise-toolbox')
-const crypto = require('crypto')
 const { RemoteAdapter } = require('./RemoteAdapter')
 const { VHDFOOTER, VHDHEADER } = require('./tests.fixtures.js')
 const { VhdFile, Constants, VhdDirectory, VhdAbstract } = require('vhd-lib')
+const { checkAliases } = require('./_cleanVm')
+const { dirname, basename } = require('path')
 
 let tempDir, adapter, handler, jobId, vdiId, basePath
 
@@ -30,12 +34,17 @@ afterEach(async () => {
   await handler.forget()
 })
 
-const uniqueId = () => crypto.randomBytes(16).toString('hex')
+const uniqueId = () => uuid.v1()
+const uniqueIdBuffer = () => uuid.v1({}, Buffer.alloc(16))
 
 async function generateVhd(path, opts = {}) {
   let vhd
 
-  const dataPath = opts.useAlias ? path + '.data' : path
+  let dataPath = path
+  if (opts.useAlias) {
+    await handler.mkdir(dirname(path) + '/data/')
+    dataPath = dirname(path) + '/data/' + basename(path)
+  }
   if (opts.mode === 'directory') {
     await handler.mkdir(dataPath)
     vhd = new VhdDirectory(handler, dataPath)
@@ -45,10 +54,9 @@ async function generateVhd(path, opts = {}) {
   }
 
   vhd.header = { ...VHDHEADER, ...opts.header }
-  vhd.footer = { ...VHDFOOTER, ...opts.footer }
-  vhd.footer.uuid = Buffer.from(crypto.randomBytes(16))
+  vhd.footer = { ...VHDFOOTER, ...opts.footer, uuid: uniqueIdBuffer() }
 
-  if (vhd.header.parentUnicodeName) {
+  if (vhd.header.parentUuid) {
     vhd.footer.diskType = Constants.DISK_TYPES.DIFFERENCING
   } else {
     vhd.footer.diskType = Constants.DISK_TYPES.DYNAMIC
@@ -70,48 +78,53 @@ test('It remove broken vhd', async () => {
   await handler.writeFile(`${basePath}/notReallyAVhd.vhd`, 'I AM NOT A VHD')
   expect((await handler.list(basePath)).length).toEqual(1)
   let loggued = ''
-  const onLog = message => {
+  const logInfo = message => {
     loggued += message
   }
-  await adapter.cleanVm('/', { remove: false, onLog })
-  expect(loggued).toEqual(`error while checking the VHD with path /${basePath}/notReallyAVhd.vhd`)
+  await adapter.cleanVm('/', { remove: false, logInfo, logWarn: logInfo, lock: false })
+  expect(loggued).toEqual(`VHD check error`)
   // not removed
   expect((await handler.list(basePath)).length).toEqual(1)
   // really remove it
-  await adapter.cleanVm('/', { remove: true, onLog })
+  await adapter.cleanVm('/', { remove: true, logInfo, logWarn: () => {}, lock: false })
   expect((await handler.list(basePath)).length).toEqual(0)
 })
 
 test('it remove vhd with missing or multiple ancestors', async () => {
-  // one with a broken parent
+  // one with a broken parent, should be deleted
   await generateVhd(`${basePath}/abandonned.vhd`, {
     header: {
       parentUnicodeName: 'gone.vhd',
-      parentUid: Buffer.from(crypto.randomBytes(16)),
+      parentUuid: uniqueIdBuffer(),
     },
   })
 
-  // one orphan, which is a full vhd, no parent
+  // one orphan, which is a full vhd, no parent : should stay
   const orphan = await generateVhd(`${basePath}/orphan.vhd`)
-  // a child to the orphan
+  // a child to the orphan in the metadata : should stay
   await generateVhd(`${basePath}/child.vhd`, {
     header: {
       parentUnicodeName: 'orphan.vhd',
-      parentUid: orphan.footer.uuid,
+      parentUuid: orphan.footer.uuid,
     },
   })
-
+  await handler.writeFile(
+    `metadata.json`,
+    JSON.stringify({
+      mode: 'delta',
+      vhds: [`${basePath}/child.vhd`, `${basePath}/abandonned.vhd`],
+    }),
+    { flags: 'w' }
+  )
   // clean
   let loggued = ''
-  const onLog = message => {
+  const logInfo = message => {
     loggued += message + '\n'
   }
-  await adapter.cleanVm('/', { remove: true, onLog })
+  await adapter.cleanVm('/', { remove: true, logInfo, logWarn: logInfo, lock: false })
 
   const deletedOrphanVhd = loggued.match(/deleting orphan VHD/g) || []
   expect(deletedOrphanVhd.length).toEqual(1) // only one vhd should have been deleted
-  const deletedAbandonnedVhd = loggued.match(/abandonned.vhd is missing/g) || []
-  expect(deletedAbandonnedVhd.length).toEqual(1) // and it must be abandonned.vhd
 
   // we don't test the filew on disk, since they will all be marker as unused and deleted without a metadata.json file
 })
@@ -139,19 +152,17 @@ test('it remove backup meta data referencing a missing vhd in delta backup', asy
   await generateVhd(`${basePath}/child.vhd`, {
     header: {
       parentUnicodeName: 'orphan.vhd',
-      parentUid: orphan.footer.uuid,
+      parentUuid: orphan.footer.uuid,
     },
   })
 
   let loggued = ''
-  const onLog = message => {
+  const logInfo = message => {
     loggued += message + '\n'
   }
-  await adapter.cleanVm('/', { remove: true, onLog })
-  let matched = loggued.match(/deleting unused VHD /g) || []
+  await adapter.cleanVm('/', { remove: true, logInfo, logWarn: logInfo, lock: false })
+  let matched = loggued.match(/deleting unused VHD/g) || []
   expect(matched.length).toEqual(1) // only one vhd should have been deleted
-  matched = loggued.match(/abandonned.vhd is unused/g) || []
-  expect(matched.length).toEqual(1) // and it must be abandonned.vhd
 
   // a missing vhd cause clean to remove all vhds
   await handler.writeFile(
@@ -162,14 +173,14 @@ test('it remove backup meta data referencing a missing vhd in delta backup', asy
         `${basePath}/deleted.vhd`, // in metadata but not in vhds
         `${basePath}/orphan.vhd`,
         `${basePath}/child.vhd`,
-        // abandonned.json is not here
+        // abandonned.vhd is not here anymore
       ],
     }),
     { flags: 'w' }
   )
   loggued = ''
-  await adapter.cleanVm('/', { remove: true, onLog })
-  matched = loggued.match(/deleting unused VHD /g) || []
+  await adapter.cleanVm('/', { remove: true, logInfo, logWarn: () => {}, lock: false })
+  matched = loggued.match(/deleting unused VHD/g) || []
   expect(matched.length).toEqual(2) // all vhds (orphan and  child  ) should have been deleted
 })
 
@@ -193,30 +204,28 @@ test('it merges delta of non destroyed chain', async () => {
   const child = await generateVhd(`${basePath}/child.vhd`, {
     header: {
       parentUnicodeName: 'orphan.vhd',
-      parentUid: orphan.footer.uuid,
+      parentUuid: orphan.footer.uuid,
     },
   })
   // a grand child
   await generateVhd(`${basePath}/grandchild.vhd`, {
     header: {
       parentUnicodeName: 'child.vhd',
-      parentUid: child.footer.uuid,
+      parentUuid: child.footer.uuid,
     },
   })
 
   let loggued = []
-  const onLog = message => {
+  const logInfo = message => {
     loggued.push(message)
   }
-  await adapter.cleanVm('/', { remove: true, onLog })
-  expect(loggued[0]).toEqual(`the parent /${basePath}/orphan.vhd of the child /${basePath}/child.vhd is unused`)
-  expect(loggued[1]).toEqual(`incorrect size in metadata: 12000 instead of 209920`)
+  await adapter.cleanVm('/', { remove: true, logInfo, logWarn: logInfo, lock: false })
+  expect(loggued[0]).toEqual(`incorrect backup size in metadata`)
 
   loggued = []
-  await adapter.cleanVm('/', { remove: true, merge: true, onLog })
-  const [unused, merging] = loggued
-  expect(unused).toEqual(`the parent /${basePath}/orphan.vhd of the child /${basePath}/child.vhd is unused`)
-  expect(merging).toEqual(`merging /${basePath}/child.vhd into /${basePath}/orphan.vhd`)
+  await adapter.cleanVm('/', { remove: true, merge: true, logInfo, logWarn: () => {}, lock: false })
+  const [merging] = loggued
+  expect(merging).toEqual(`merging VHD chain`)
 
   const metadata = JSON.parse(await handler.readFile(`metadata.json`))
   // size should be the size of children + grand children after the merge
@@ -235,12 +244,8 @@ test('it finish unterminated merge ', async () => {
     `metadata.json`,
     JSON.stringify({
       mode: 'delta',
-      size: undefined,
-      vhds: [
-        `${basePath}/orphan.vhd`, // grand child should not be merged
-        `${basePath}/child.vhd`,
-        // orphan is not here, he should be merged in child
-      ],
+      size: 209920,
+      vhds: [`${basePath}/orphan.vhd`, `${basePath}/child.vhd`],
     })
   )
 
@@ -250,7 +255,7 @@ test('it finish unterminated merge ', async () => {
   const child = await generateVhd(`${basePath}/child.vhd`, {
     header: {
       parentUnicodeName: 'orphan.vhd',
-      parentUid: orphan.footer.uuid,
+      parentUuid: orphan.footer.uuid,
     },
   })
   // a merge in progress file
@@ -266,8 +271,7 @@ test('it finish unterminated merge ', async () => {
     })
   )
 
-  // a unfinished merging
-  await adapter.cleanVm('/', { remove: true, merge: true })
+  await adapter.cleanVm('/', { remove: true, merge: true, logWarn: () => {}, lock: false })
   // merging is already tested in vhd-lib, don't retest it here (and theses vhd are as empty as my stomach at 12h12)
 
   // only check deletion
@@ -279,12 +283,17 @@ test('it finish unterminated merge ', async () => {
 // each of the vhd can be a file, a directory, an alias to a file or an alias to a directory
 // the message an resulting files should be identical to the output with vhd files which is tested independantly
 
-describe('tests mulitple combination ', () => {
+describe('tests multiple combination ', () => {
   for (const useAlias of [true, false]) {
     for (const vhdMode of ['file', 'directory']) {
       test(`alias : ${useAlias}, mode: ${vhdMode}`, async () => {
         // a broken VHD
-        const brokenVhdDataPath = basePath + useAlias ? 'broken.data' : 'broken.vhd'
+        if (useAlias) {
+          await handler.mkdir(basePath + '/data')
+        }
+
+        const brokenVhdDataPath = basePath + (useAlias ? '/data/broken.vhd' : '/broken.vhd')
+
         if (vhdMode === 'directory') {
           await handler.mkdir(brokenVhdDataPath)
         } else {
@@ -302,9 +311,10 @@ describe('tests mulitple combination ', () => {
           mode: vhdMode,
           header: {
             parentUnicodeName: 'gone.vhd',
-            parentUid: crypto.randomBytes(16),
+            parentUuid: uniqueIdBuffer(),
           },
         })
+
         // an ancestor of a vhd present in metadata
         const ancestor = await generateVhd(`${basePath}/ancestor.vhd`, {
           useAlias,
@@ -315,7 +325,7 @@ describe('tests mulitple combination ', () => {
           mode: vhdMode,
           header: {
             parentUnicodeName: 'ancestor.vhd' + (useAlias ? '.alias.vhd' : ''),
-            parentUid: ancestor.footer.uuid,
+            parentUuid: ancestor.footer.uuid,
           },
         })
         // a grand child  vhd in metadata
@@ -324,7 +334,7 @@ describe('tests mulitple combination ', () => {
           mode: vhdMode,
           header: {
             parentUnicodeName: 'child.vhd' + (useAlias ? '.alias.vhd' : ''),
-            parentUid: child.footer.uuid,
+            parentUuid: child.footer.uuid,
           },
         })
 
@@ -339,7 +349,7 @@ describe('tests mulitple combination ', () => {
           mode: vhdMode,
           header: {
             parentUnicodeName: 'cleanAncestor.vhd' + (useAlias ? '.alias.vhd' : ''),
-            parentUid: cleanAncestor.footer.uuid,
+            parentUuid: cleanAncestor.footer.uuid,
           },
         })
 
@@ -367,7 +377,8 @@ describe('tests mulitple combination ', () => {
             ],
           })
         )
-        await adapter.cleanVm('/', { remove: true, merge: true })
+
+        await adapter.cleanVm('/', { remove: true, merge: true, logWarn: () => {}, lock: false })
 
         const metadata = JSON.parse(await handler.readFile(`metadata.json`))
         // size should be the size of children + grand children + clean after the merge
@@ -379,14 +390,16 @@ describe('tests mulitple combination ', () => {
         const survivors = await handler.list(basePath)
         // console.log(survivors)
         if (useAlias) {
+          const dataSurvivors = await handler.list(basePath + '/data')
           // the goal of the alias : do not move a full folder
-          expect(survivors).toContain('ancestor.vhd.data')
-          expect(survivors).toContain('grandchild.vhd.data')
-          expect(survivors).toContain('cleanAncestor.vhd.data')
+          expect(dataSurvivors).toContain('ancestor.vhd')
+          expect(dataSurvivors).toContain('grandchild.vhd')
+          expect(dataSurvivors).toContain('cleanAncestor.vhd')
           expect(survivors).toContain('clean.vhd.alias.vhd')
           expect(survivors).toContain('child.vhd.alias.vhd')
           expect(survivors).toContain('grandchild.vhd.alias.vhd')
-          expect(survivors.length).toEqual(6)
+          expect(survivors.length).toEqual(4) // the 3 ok + data
+          expect(dataSurvivors.length).toEqual(3) // the 3 ok + data
         } else {
           expect(survivors).toContain('clean.vhd')
           expect(survivors).toContain('child.vhd')
@@ -401,7 +414,31 @@ describe('tests mulitple combination ', () => {
 test('it cleans orphan merge states ', async () => {
   await handler.writeFile(`${basePath}/.orphan.vhd.merge.json`, '')
 
-  await adapter.cleanVm('/', { remove: true })
+  await adapter.cleanVm('/', { remove: true, logWarn: () => {}, lock: false })
 
   expect(await handler.list(basePath)).toEqual([])
 })
+
+test('check Aliases should work alone', async () => {
+  await handler.mkdir('vhds')
+  await handler.mkdir('vhds/data')
+  await generateVhd(`vhds/data/ok.vhd`)
+  await VhdAbstract.createAlias(handler, 'vhds/ok.alias.vhd', 'vhds/data/ok.vhd')
+
+  await VhdAbstract.createAlias(handler, 'vhds/missingData.alias.vhd', 'vhds/data/nonexistent.vhd')
+
+  await generateVhd(`vhds/data/missingalias.vhd`)
+
+  await checkAliases(['vhds/missingData.alias.vhd', 'vhds/ok.alias.vhd'], 'vhds/data', {
+    remove: true,
+    handler,
+    logWarn: () => {},
+  })
+
+  // only ok have suvived
+  const alias = (await handler.list('vhds')).filter(f => f.endsWith('.vhd'))
+  expect(alias.length).toEqual(1)
+
+  const data = await handler.list('vhds/data')
+  expect(data.length).toEqual(1)
+})

@xen-orchestra/backups/_cleanVm.js

@@ -1,19 +1,27 @@
-const assert = require('assert')
+'use strict'
+
 const sum = require('lodash/sum')
+const UUID = require('uuid')
 const { asyncMap } = require('@xen-orchestra/async-map')
-const { Constants, mergeVhd, openVhd, VhdAbstract, VhdFile } = require('vhd-lib')
+const { Constants, openVhd, VhdAbstract, VhdFile } = require('vhd-lib')
+const { isVhdAlias, resolveVhdAlias } = require('vhd-lib/aliases')
 const { dirname, resolve } = require('path')
 const { DISK_TYPES } = Constants
 const { isMetadataFile, isVhdFile, isXvaFile, isXvaSumFile } = require('./_backupType.js')
 const { limitConcurrency } = require('limit-concurrency-decorator')
+const { mergeVhdChain } = require('vhd-lib/merge')
 
 const { Task } = require('./Task.js')
 const { Disposable } = require('promise-toolbox')
+const handlerPath = require('@xen-orchestra/fs/path')
 
 // checking the size of a vhd directory is costly
 // 1 Http Query per 1000 blocks
 // we only check size of all the vhd are VhdFiles
-function shouldComputeVhdsSize(vhds) {
+function shouldComputeVhdsSize(handler, vhds) {
+  if (handler.isEncrypted) {
+    return false
+  }
   return vhds.every(vhd => vhd instanceof VhdFile)
 }
 
@@ -21,88 +29,51 @@ const computeVhdsSize = (handler, vhdPaths) =>
   Disposable.use(
     vhdPaths.map(vhdPath => openVhd(handler, vhdPath)),
     async vhds => {
-      if (shouldComputeVhdsSize(vhds)) {
+      if (shouldComputeVhdsSize(handler, vhds)) {
         const sizes = await asyncMap(vhds, vhd => vhd.getSize())
         return sum(sizes)
       }
     }
   )
 
-// chain is an array of VHDs from child to parent
-//
-// the whole chain will be merged into parent, parent will be renamed to child
-// and all the others will deleted
-async function mergeVhdChain(chain, { handler, onLog, remove, merge }) {
-  assert(chain.length >= 2)
-
-  let child = chain[0]
-  const parent = chain[chain.length - 1]
-  const children = chain.slice(0, -1).reverse()
-
-  chain
-    .slice(1)
-    .reverse()
-    .forEach(parent => {
-      onLog(`the parent ${parent} of the child ${child} is unused`)
-    })
-
+// chain is [ ancestor, child_1, ..., child_n ]
+async function _mergeVhdChain(handler, chain, { logInfo, remove, merge, mergeBlockConcurrency }) {
   if (merge) {
-    // `mergeVhd` does not work with a stream, either
-    // - make it accept a stream
-    // - or create synthetic VHD which is not a stream
-    if (children.length !== 1) {
-      // TODO: implement merging multiple children
-      children.length = 1
-      child = children[0]
-    }
-
-    onLog(`merging ${child} into ${parent}`)
+    logInfo(`merging VHD chain`, { chain })
 
     let done, total
     const handle = setInterval(() => {
       if (done !== undefined) {
-        onLog(`merging ${child}: ${done}/${total}`)
+        logInfo('merge in progress', {
+          done,
+          parent: chain[0],
+          progress: Math.round((100 * done) / total),
+          total,
+        })
       }
     }, 10e3)
-
-    const mergedSize = await mergeVhd(
-      handler,
-      parent,
-      handler,
-      child,
-      // children.length === 1
-      //   ? child
-      //   : await createSyntheticStream(handler, children),
-      {
+    try {
+      return await mergeVhdChain(handler, chain, {
+        logInfo,
+        mergeBlockConcurrency,
         onProgress({ done: d, total: t }) {
           done = d
           total = t
         },
-      }
-    )
-
-    clearInterval(handle)
-
-    await Promise.all([
-      VhdAbstract.rename(handler, parent, child),
-      asyncMap(children.slice(0, -1), child => {
-        onLog(`the VHD ${child} is unused`)
-        if (remove) {
-          onLog(`deleting unused VHD ${child}`)
-          return VhdAbstract.unlink(handler, child)
-        }
-      }),
-    ])
-
-    return mergedSize
+        removeUnused: remove,
+      })
+    } finally {
+      clearInterval(handle)
+    }
   }
 }
 
 const noop = Function.prototype
 
 const INTERRUPTED_VHDS_REG = /^\.(.+)\.merge.json$/
-const listVhds = async (handler, vmDir) => {
+const listVhds = async (handler, vmDir, logWarn) => {
   const vhds = new Set()
+  const aliases = {}
   const interruptedVhds = new Map()
 
   await asyncMap(
@@ -119,37 +90,118 @@ const listVhds = async (handler, vmDir) => {
           const list = await handler.list(vdiDir, {
             filter: file => isVhdFile(file) || INTERRUPTED_VHDS_REG.test(file),
           })
+          aliases[vdiDir] = list.filter(vhd => isVhdAlias(vhd)).map(file => `${vdiDir}/${file}`)
 
-          list.forEach(file => {
+          await asyncMap(list, async file => {
             const res = INTERRUPTED_VHDS_REG.exec(file)
             if (res === null) {
               vhds.add(`${vdiDir}/${file}`)
             } else {
-              interruptedVhds.set(`${vdiDir}/${res[1]}`, `${vdiDir}/${file}`)
+              try {
+                const mergeState = JSON.parse(await handler.readFile(`${vdiDir}/${file}`))
+                interruptedVhds.set(`${vdiDir}/${res[1]}`, {
+                  statePath: `${vdiDir}/${file}`,
+                  chain: mergeState.chain,
+                })
+              } catch (error) {
+                // fall back to a non resuming merge
+                vhds.add(`${vdiDir}/${file}`)
+                logWarn('failed to read existing merge state', { path: file, error })
+              }
             }
           })
         }
       )
   )
 
-  return { vhds, interruptedVhds }
+  return { vhds, interruptedVhds, aliases }
 }
 
+async function checkAliases(
+  aliasPaths,
+  targetDataRepository,
+  { handler, logInfo = noop, logWarn = console.warn, remove = false }
+) {
+  const aliasFound = []
+  for (const alias of aliasPaths) {
+    const target = await resolveVhdAlias(handler, alias)
+
+    if (!isVhdFile(target)) {
+      logWarn('alias references non VHD target', { alias, target })
+      if (remove) {
+        logInfo('removing alias and non VHD target', { alias, target })
+        await handler.unlink(target)
+        await handler.unlink(alias)
+      }
+      continue
+    }
+
+    try {
+      const { dispose } = await openVhd(handler, target)
+      try {
+        await dispose()
+      } catch (e) {
+        // error during dispose should not trigger a deletion
+      }
+    } catch (error) {
+      logWarn('missing or broken alias target', { alias, target, error })
+      if (remove) {
+        try {
+          await VhdAbstract.unlink(handler, alias)
+        } catch (error) {
+          if (error.code !== 'ENOENT') {
+            logWarn('error deleting alias target', { alias, target, error })
+          }
+        }
+      }
+      continue
+    }
+
+    aliasFound.push(resolve('/', target))
+  }
+
+  const vhds = await handler.list(targetDataRepository, {
+    ignoreMissing: true,
+    prependDir: true,
+  })
+
+  await asyncMap(vhds, async path => {
+    if (!aliasFound.includes(path)) {
+      logWarn('no alias references VHD', { path })
+      if (remove) {
+        logInfo('deleting unused VHD', { path })
+        await VhdAbstract.unlink(handler, path)
+      }
+    }
+  })
+}
+
+exports.checkAliases = checkAliases
+
 const defaultMergeLimiter = limitConcurrency(1)
 
 exports.cleanVm = async function cleanVm(
   vmDir,
-  { fixMetadata, remove, merge, mergeLimiter = defaultMergeLimiter, onLog = noop }
+  {
+    fixMetadata,
+    remove,
+    merge,
+    mergeBlockConcurrency,
+    mergeLimiter = defaultMergeLimiter,
+    logInfo = noop,
+    logWarn = console.warn,
+  }
 ) {
-  const limitedMergeVhdChain = mergeLimiter(mergeVhdChain)
+  const limitedMergeVhdChain = mergeLimiter(_mergeVhdChain)
 
   const handler = this._handler
 
   const vhdsToJSons = new Set()
+  const vhdById = new Map()
   const vhdParents = { __proto__: null }
   const vhdChildren = { __proto__: null }
 
-  const { vhds, interruptedVhds } = await listVhds(handler, vmDir)
+  const { vhds, interruptedVhds, aliases } = await listVhds(handler, vmDir, logWarn)
 
   // remove broken VHDs
   await asyncMap(vhds, async path => {
@@ -167,12 +219,31 @@ exports.cleanVm = async function cleanVm(
           }
           vhdChildren[parent] = path
         }
+        // Detect VHDs with the same UUIDs
+        //
+        // Due to a bug introduced in a1bcd35e2
+        const duplicate = vhdById.get(UUID.stringify(vhd.footer.uuid))
+        let vhdKept = vhd
+        if (duplicate !== undefined) {
+          logWarn('uuid is duplicated', { uuid: UUID.stringify(vhd.footer.uuid) })
+          if (duplicate.containsAllDataOf(vhd)) {
+            logWarn(`should delete ${path}`)
+            vhdKept = duplicate
+            vhds.delete(path)
+          } else if (vhd.containsAllDataOf(duplicate)) {
+            logWarn(`should delete ${duplicate._path}`)
+            vhds.delete(duplicate._path)
+          } else {
+            logWarn('same ids but different content')
+          }
+        }
+        vhdById.set(UUID.stringify(vhdKept.footer.uuid), vhdKept)
       })
     } catch (error) {
       vhds.delete(path)
-      onLog(`error while checking the VHD with path ${path}`, { error })
+      logWarn('VHD check error', { path, error })
       if (error?.code === 'ERR_ASSERTION' && remove) {
-        onLog(`deleting broken ${path}`)
+        logInfo('deleting broken VHD', { path })
         return VhdAbstract.unlink(handler, path)
       }
     }
@@ -181,21 +252,25 @@ exports.cleanVm = async function cleanVm(
   // remove interrupted merge states for missing VHDs
   for (const interruptedVhd of interruptedVhds.keys()) {
     if (!vhds.has(interruptedVhd)) {
-      const statePath = interruptedVhds.get(interruptedVhd)
+      const { statePath } = interruptedVhds.get(interruptedVhd)
       interruptedVhds.delete(interruptedVhd)
 
-      onLog('orphan merge state', {
+      logWarn('orphan merge state', {
         mergeStatePath: statePath,
         missingVhdPath: interruptedVhd,
       })
       if (remove) {
-        onLog(`deleting orphan merge state ${statePath}`)
+        logInfo('deleting orphan merge state', { statePath })
         await handler.unlink(statePath)
       }
     }
   }
 
-  // @todo : add check for data folder of alias not referenced in a valid alias
+  // check if alias are correct
+  // check if all vhd in data subfolder have a corresponding alias
+  await asyncMap(Object.keys(aliases), async dir => {
+    await checkAliases(aliases[dir], `${dir}/data`, { handler, logInfo, logWarn, remove })
+  })
 
   // remove VHDs with missing ancestors
   {
@@ -216,9 +291,9 @@ exports.cleanVm = async function cleanVm(
       if (!vhds.has(parent)) {
         vhds.delete(vhdPath)
 
-        onLog(`the parent ${parent} of the VHD ${vhdPath} is missing`)
+        logWarn('parent VHD is missing', { parent, child: vhdPath })
         if (remove) {
-          onLog(`deleting orphan VHD ${vhdPath}`)
+          logInfo('deleting orphan VHD', { path: vhdPath })
           deletions.push(VhdAbstract.unlink(handler, vhdPath))
         }
       }
@@ -236,6 +311,7 @@ exports.cleanVm = async function cleanVm(
   }
 
   const jsons = new Set()
+  let mustInvalidateCache = false
   const xvas = new Set()
   const xvaSums = []
   const entries = await handler.list(vmDir, {
@@ -255,7 +331,7 @@ exports.cleanVm = async function cleanVm(
     // check is not good enough to delete the file, the best we can do is report
     // it
     if (!(await this.isValidXva(path))) {
-      onLog(`the XVA with path ${path} is potentially broken`)
+      logWarn('XVA might be broken', { path })
     }
   })
 
@@ -269,7 +345,7 @@ exports.cleanVm = async function cleanVm(
     try {
       metadata = JSON.parse(await handler.readFile(json))
     } catch (error) {
-      onLog(`failed to read metadata file ${json}`, { error })
+      logWarn('failed to read backup metadata', { path: json, error })
       jsons.delete(json)
       return
     }
@@ -280,10 +356,11 @@ exports.cleanVm = async function cleanVm(
       if (xvas.has(linkedXva)) {
         unusedXvas.delete(linkedXva)
       } else {
-        onLog(`the XVA linked to the metadata ${json} is missing`)
+        logWarn('the XVA linked to the backup is missing', { backup: json, xva: linkedXva })
         if (remove) {
-          onLog(`deleting incomplete backup ${json}`)
+          logInfo('deleting incomplete backup', { path: json })
           jsons.delete(json)
+          mustInvalidateCache = true
           await handler.unlink(json)
         }
       }
@@ -303,9 +380,10 @@ exports.cleanVm = async function cleanVm(
           vhdsToJSons[path] = json
         })
       } else {
-        onLog(`Some VHDs linked to the metadata ${json} are missing`, { missingVhds })
+        logWarn('some VHDs linked to the backup are missing', { backup: json, missingVhds })
         if (remove) {
-          onLog(`deleting incomplete backup ${json}`)
+          logInfo('deleting incomplete backup', { path: json })
+          mustInvalidateCache = true
           jsons.delete(json)
           await handler.unlink(json)
         }
@@ -317,7 +395,7 @@ exports.cleanVm = async function cleanVm(
   const unusedVhdsDeletion = []
   const toMerge = []
   {
-    // VHD chains (as list from child to ancestor) to merge indexed by last
+    // VHD chains (as list from oldest to most recent) to merge indexed by most recent
     // ancestor
     const vhdChainsToMerge = { __proto__: null }
 
@@ -341,14 +419,14 @@ exports.cleanVm = async function cleanVm(
       if (child !== undefined) {
         const chain = getUsedChildChainOrDelete(child)
         if (chain !== undefined) {
-          chain.push(vhd)
+          chain.unshift(vhd)
           return chain
         }
       }
 
-      onLog(`the VHD ${vhd} is unused`)
+      logWarn('unused VHD', { path: vhd })
       if (remove) {
-        onLog(`deleting unused VHD ${vhd}`)
+        logInfo('deleting unused VHD', { path: vhd })
         unusedVhdsDeletion.push(VhdAbstract.unlink(handler, vhd))
       }
     }
@@ -359,7 +437,13 @@ exports.cleanVm = async function cleanVm(
 
     // merge interrupted VHDs
     for (const parent of interruptedVhds.keys()) {
-      vhdChainsToMerge[parent] = [vhdChildren[parent], parent]
+      // before #6349 the chain wasn't in the mergeState
+      const { chain, statePath } = interruptedVhds.get(parent)
+      if (chain === undefined) {
+        vhdChainsToMerge[parent] = [parent, vhdChildren[parent]]
+      } else {
+        vhdChainsToMerge[parent] = chain.map(vhdPath => handlerPath.resolveFromFile(statePath, vhdPath))
+      }
     }
 
     Object.values(vhdChainsToMerge).forEach(chain => {
@@ -372,9 +456,15 @@ exports.cleanVm = async function cleanVm(
   const metadataWithMergedVhd = {}
   const doMerge = async () => {
     await asyncMap(toMerge, async chain => {
-      const merged = await limitedMergeVhdChain(chain, { handler, onLog, remove, merge })
+      const merged = await limitedMergeVhdChain(handler, chain, {
+        logInfo,
+        logWarn,
+        remove,
+        merge,
+        mergeBlockConcurrency,
+      })
       if (merged !== undefined) {
-        const metadataPath = vhdsToJSons[chain[0]] // all the chain should have the same metada file
+        const metadataPath = vhdsToJSons[chain[chain.length - 1]] // all the chain should have the same metada file
         metadataWithMergedVhd[metadataPath] = true
       }
     })
@@ -384,18 +474,18 @@ exports.cleanVm = async function cleanVm(
     ...unusedVhdsDeletion,
     toMerge.length !== 0 && (merge ? Task.run({ name: 'merge' }, doMerge) : doMerge()),
     asyncMap(unusedXvas, path => {
-      onLog(`the XVA ${path} is unused`)
+      logWarn('unused XVA', { path })
       if (remove) {
-        onLog(`deleting unused XVA ${path}`)
+        logInfo('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`)
+        logInfo('unused XVA checksum', { path })
         if (remove) {
-          onLog(`deleting unused XVA checksum ${path}`)
+          logInfo('deleting unused XVA checksum', { path })
           return handler.unlink(path)
         }
       }
@@ -417,7 +507,11 @@ exports.cleanVm = async function cleanVm(
       if (mode === 'full') {
         // a full backup : check size
         const linkedXva = resolve('/', vmDir, xva)
-        fileSystemSize = await handler.getSize(linkedXva)
+        try {
+          fileSystemSize = await handler.getSize(linkedXva)
+        } catch (error) {
+          // can fail with encrypted remote
+        }
       } else if (mode === 'delta') {
         const linkedVhds = Object.keys(vhds).map(key => resolve('/', vmDir, vhds[key]))
         fileSystemSize = await computeVhdsSize(handler, linkedVhds)
@@ -429,11 +523,15 @@ exports.cleanVm = async function cleanVm(
 
         // don't warn if the size has changed after a merge
         if (!merged && fileSystemSize !== size) {
-          onLog(`incorrect size in metadata: ${size ?? 'none'} instead of ${fileSystemSize}`)
+          logWarn('incorrect backup size in metadata', {
+            path: metadataPath,
+            actual: size ?? 'none',
+            expected: fileSystemSize,
+          })
         }
       }
     } catch (error) {
-      onLog(`failed to get size of ${metadataPath}`, { error })
+      logWarn('failed to get backup size', { backup: metadataPath, error })
       return
     }
 
@@ -443,11 +541,16 @@ exports.cleanVm = async function cleanVm(
       try {
         await handler.writeFile(metadataPath, JSON.stringify(metadata), { flags: 'w' })
       } catch (error) {
-        onLog(`failed to update size in backup metadata ${metadataPath} after merge`, { error })
+        logWarn('failed to update backup size in metadata', { path: metadataPath, error })
       }
     }
   })
 
+  // purge cache if a metadata file has been deleted
+  if (mustInvalidateCache) {
+    await handler.unlink(vmDir + '/cache.json.gz')
+  }
+
   return {
     // boolean whether some VHDs were merged (or should be merged)
     merge: toMerge.length !== 0,

@xen-orchestra/backups/_deltaVm.js

@@ -1,14 +1,18 @@
-const compareVersions = require('compare-versions')
+'use strict'
+
 const find = require('lodash/find.js')
 const groupBy = require('lodash/groupBy.js')
-const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
+const ignoreErrors = require('promise-toolbox/ignoreErrors')
 const omit = require('lodash/omit.js')
 const { asyncMap } = require('@xen-orchestra/async-map')
 const { CancelToken } = require('promise-toolbox')
+const { compareVersions } = require('compare-versions')
 const { createVhdStreamWithLength } = require('vhd-lib')
 const { defer } = require('golike-defer')
 
 const { cancelableMap } = require('./_cancelableMap.js')
+const { Task } = require('./Task.js')
+const { pick } = require('lodash')
 
 const TAG_BASE_DELTA = 'xo:base_delta'
 exports.TAG_BASE_DELTA = TAG_BASE_DELTA
@@ -17,6 +21,17 @@ const TAG_COPY_SRC = 'xo:copy_of'
 exports.TAG_COPY_SRC = TAG_COPY_SRC
 
 const ensureArray = value => (value === undefined ? [] : Array.isArray(value) ? value : [value])
+const resolveUuid = async (xapi, cache, uuid, type) => {
+  if (uuid == null) {
+    return uuid
+  }
+  let ref = cache.get(uuid)
+  if (ref === undefined) {
+    ref = await xapi.call(`${type}.get_by_uuid`, uuid)
+    cache.set(uuid, ref)
+  }
+  return ref
+}
 
 exports.exportDeltaVm = async function exportDeltaVm(
   vm,
@@ -50,17 +65,6 @@ exports.exportDeltaVm = async function exportDeltaVm(
       return
     }
 
-    // If the VDI name start with `[NOBAK]`, do not export it.
-    if (vdi.name_label.startsWith('[NOBAK]')) {
-      // FIXME: find a way to not create the VDI snapshot in the
-      // first time.
-      //
-      // The snapshot must not exist otherwise it could break the
-      // next export.
-      ignoreErrors.call(vdi.$destroy())
-      return
-    }
-
     vbds[vbd.$ref] = vbd
 
     const vdiRef = vdi.$ref
@@ -165,6 +169,12 @@ exports.importDeltaVm = defer(async function importDeltaVm(
     }
   }
 
+  const cache = new Map()
+  const mapVdisSrRefs = {}
+  for (const [vdiUuid, srUuid] of Object.entries(mapVdisSrs)) {
+    mapVdisSrRefs[vdiUuid] = await resolveUuid(xapi, cache, srUuid, 'SR')
+  }
+
   const baseVdis = {}
   baseVm &&
     baseVm.$VBDs.forEach(vbd => {
@@ -179,19 +189,25 @@ exports.importDeltaVm = defer(async function importDeltaVm(
   let suspendVdi
   if (vmRecord.power_state === 'Suspended') {
     const vdi = vdiRecords[vmRecord.suspend_VDI]
-    suspendVdi = await xapi.getRecord(
-      'VDI',
-      await xapi.VDI_create({
-        ...vdi,
-        other_config: {
-          ...vdi.other_config,
-          [TAG_BASE_DELTA]: undefined,
-          [TAG_COPY_SRC]: vdi.uuid,
-        },
-        sr: mapVdisSrs[vdi.uuid] ?? sr.$ref,
+    if (vdi === undefined) {
+      Task.warning('Suspend VDI not available for this suspended VM', {
+        vm: pick(vmRecord, 'uuid', 'name_label'),
       })
-    )
-    $defer.onFailure(() => suspendVdi.$destroy())
+    } else {
+      suspendVdi = await xapi.getRecord(
+        'VDI',
+        await xapi.VDI_create({
+          ...vdi,
+          other_config: {
+            ...vdi.other_config,
+            [TAG_BASE_DELTA]: undefined,
+            [TAG_COPY_SRC]: vdi.uuid,
+          },
+          sr: mapVdisSrRefs[vdi.uuid] ?? sr.$ref,
+        })
+      )
+      $defer.onFailure(() => suspendVdi.$destroy())
+    }
   }
 
   // 1. Create the VM.
@@ -255,7 +271,7 @@ exports.importDeltaVm = defer(async function importDeltaVm(
             [TAG_BASE_DELTA]: undefined,
             [TAG_COPY_SRC]: vdi.uuid,
           },
-          SR: mapVdisSrs[vdi.uuid] ?? sr.$ref,
+          SR: mapVdisSrRefs[vdi.uuid] ?? sr.$ref,
         })
       )
       $defer.onFailure(() => newVdi.$destroy())

@xen-orchestra/backups/_filenameDate.js

@@ -1,3 +1,5 @@
+'use strict'
+
 const { utcFormat, utcParse } = require('d3-time-format')
 
 // Format a date in ISO 8601 in a safe way to be used in filenames

@xen-orchestra/backups/_forkStreamUnpipe.js

@@ -1,6 +1,10 @@
+'use strict'
+
 const eos = require('end-of-stream')
 const { PassThrough } = require('stream')
 
+const { debug } = require('@xen-orchestra/log').createLogger('xo:backups:forkStreamUnpipe')
+
 // create a new readable stream from an existing one which may be piped later
 //
 // in case of error in the new readable stream, it will simply be unpiped
@@ -9,18 +13,23 @@ exports.forkStreamUnpipe = function forkStreamUnpipe(stream) {
   const { forks = 0 } = stream
   stream.forks = forks + 1
 
+  debug('forking', { forks: stream.forks })
+
   const proxy = new PassThrough()
   stream.pipe(proxy)
   eos(stream, error => {
     if (error !== undefined) {
+      debug('error on original stream, destroying fork', { error })
       proxy.destroy(error)
     }
   })
-  eos(proxy, _ => {
-    stream.forks--
+  eos(proxy, error => {
+    debug('end of stream, unpiping', { error, forks: --stream.forks })
+
     stream.unpipe(proxy)
 
     if (stream.forks === 0) {
+      debug('no more forks, destroying original stream')
       stream.destroy(new Error('no more consumers for this stream'))
     }
   })