feat(xen-api/_watchEvents): detect and fix desynchornizations

`Promise.race()` leads to memory leaks if some promises are never resolved.

See nodejs/node#17469

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

@@ -16,6 +16,6 @@
   },
   "dependencies": {
     "golike-defer": "^0.4.1",
-    "xen-api": "^0.25.0"
+    "xen-api": "^0.25.1"
   }
 }

CHANGELOG.md

@@ -1,5 +1,16 @@
 # ChangeLog
 
+## **5.33.1** (2019-04-04)
+
+### Bug fix
+
+- Fix major memory leak [2563be4](https://github.com/vatesfr/xen-orchestra/commit/2563be472bfd84c6ed867efd21c4aeeb824d387f)
+
+### Released packages
+
+- xen-api v0.25.1
+- xo-server v5.38.2
+
 ## **5.33.0** (2019-03-29)
 
 ### Enhancements

CHANGELOG.unreleased.md

@@ -4,6 +4,7 @@
 
 - [Settings/remotes] Expose mount options field for SMB [#4063](https://github.com/vatesfr/xen-orchestra/issues/4063) (PR [#4067](https://github.com/vatesfr/xen-orchestra/pull/4067))
 - [Backup/Schedule] Add warning regarding DST when you add a schedule [#4042](https://github.com/vatesfr/xen-orchestra/issues/4042) (PR [#4056](https://github.com/vatesfr/xen-orchestra/pull/4056))
+- [Import] Avoid blocking the UI when dropping a big OVA file on the UI (PR [#4018](https://github.com/vatesfr/xen-orchestra/pull/4018))
 
 ### Bug fixes
 
@@ -11,6 +12,7 @@
 
 ### Released packages
 
+- xo-vmdk-to-vhd v0.1.7
 - vhd-lib v0.6.1
 - xo-server v5.39.0
 - xo-web v5.39.0

docs/metadata_backup.md

@@ -1,6 +1,6 @@
 # Metadata backup
 
-> WARNING: Metadata backup is an experimental feature. Restore is not yet available and some unexpected issues may occur.
+> WARNING: Metadata backup is an experimental feature. Unexpected issues are possible, but unlikely.
 
 ## Introduction
 
@@ -11,21 +11,38 @@ In Xen Orchestra, Metadata backup is divided into two different options:
 * Pool metadata backup
 * XO configuration backup
 
-### How to use metadata backup
+### Performing a backup
 
-In the backup job section, when creating a new backup job, you will now have a choice between backing up VMs and backing up Metadata.
-![](https://user-images.githubusercontent.com/21563339/53413921-bd636f00-39cd-11e9-8a3c-d4f893135fa4.png)
+In the backup job section, when creating a new backup job, you will now have a choice between backing up VMs and backing up Metadata:
+![](./assets/metadata-1.png)
 
 When you select Metadata backup, you will have a new backup job screen, letting you choose between a pool metadata backup and an XO configuration backup (or both at the same time):
 
-![](https://user-images.githubusercontent.com/21563339/52416838-d2de2b00-2aea-11e9-8da0-340fcb2767db.png)
+![](./assets/metadata-2.png)
 
 Define the name and retention for the job.
 
-![](https://user-images.githubusercontent.com/21563339/52471527-65390a00-2b91-11e9-8019-600a4d9eeafb.png)
+![](./assets/metadata-3.png)
 
 Once created, the job is displayed with the other classic jobs.
 
-![](https://user-images.githubusercontent.com/21563339/52416802-c0fc8800-2aea-11e9-8ef0-b0c1bd0e48b8.png)
+![](./assets/metadata-4.png)
 
-> Restore for metadata backup jobs should be available in XO 5.33
+
+### Performing a restore
+
+> WARNING: restoring pool metadata completely overwrites the XAPI database of a host. Only perform a metadata restore if it is a new server with nothing running on it (eg replacing a host with new hardware).
+
+If you browse to the Backup NG Restore panel, you will now notice a Metadata filter button:
+
+![](./assets/metadata-5.png)
+
+If you click this button, it will show you Metadata backups available for restore:
+
+![](./assets/metadata-6.png)
+
+You can see both our Xen Orchestra config backup, and our pool metadata backup. To restore one, simply click the blue restore arrow, choose a backup date to restore, and click OK:
+
+![](./assets/metadata-7.png)
+
+That's it!

packages/xapi-explore-sr/package.json

@@ -41,7 +41,7 @@
     "human-format": "^0.10.0",
     "lodash": "^4.17.4",
     "pw": "^0.0.4",
-    "xen-api": "^0.25.0"
+    "xen-api": "^0.25.1"
   },
   "devDependencies": {
     "@babel/cli": "^7.1.5",

packages/xen-api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "xen-api",
-  "version": "0.25.0",
+  "version": "0.25.1",
   "license": "ISC",
   "description": "Connector to the Xen API",
   "keywords": [

packages/xen-api/src/_MultiCounter.js

@@ -0,0 +1,8 @@
+const handler = {
+  get(target, property) {
+    const value = target[property]
+    return value !== undefined ? value : 0
+  },
+}
+
+export const create = () => new Proxy({ __proto__: null }, handler)

packages/xen-api/src/_Signal.js

@@ -1,142 +0,0 @@
-function request() {
-  if (this._requested) {
-    return
-  }
-
-  this._requested = true
-
-  const resolve = this._resolve
-  if (resolve !== undefined) {
-    this._resolve = undefined
-    resolve()
-  }
-
-  const listeners = this._listeners
-  if (listeners !== undefined) {
-    this._listeners = undefined
-    for (let i = 0, n = listeners.length; i < n; ++i) {
-      listeners[i].call(this)
-    }
-  }
-}
-
-const INTERNAL = {}
-
-function Source(signals) {
-  const request_ = (this.request = request.bind(
-    (this.signal = new Signal(INTERNAL))
-  ))
-
-  if (signals === undefined) {
-    return
-  }
-
-  const n = signals.length
-  for (let i = 0; i < n; ++i) {
-    if (signals[i].requested) {
-      request_()
-      return
-    }
-  }
-  for (let i = 0; i < n; ++i) {
-    signals[i].addListener(request_)
-  }
-}
-
-class Subscription {
-  constructor(signal, listener) {
-    this._listener = listener
-    this._signal = signal
-  }
-
-  get closed() {
-    return this._signal === undefined
-  }
-
-  unsubscribe() {
-    const signal = this._signal
-    if (signal !== undefined) {
-      const listener = this._listener
-      this._listener = this._signal = undefined
-
-      const listeners = signal._listeners
-      if (listeners !== undefined) {
-        const i = listeners.indexOf(listener)
-        if (i !== -1) {
-          listeners.splice(i, 1)
-        }
-      }
-    }
-  }
-}
-const closedSubscription = new Subscription()
-
-export default class Signal {
-  static source(signals) {
-    return new Source(signals)
-  }
-
-  constructor(executor) {
-    this._listeners = undefined
-    this._promise = undefined
-    this._requested = false
-    this._resolve = undefined
-
-    if (executor !== INTERNAL) {
-      executor(request.bind(this))
-    }
-  }
-
-  get description() {
-    return this._description
-  }
-
-  get requested() {
-    return this._requested
-  }
-
-  throwIfRequested() {
-    if (this._requested) {
-      throw new Error('this signal has been requested')
-    }
-  }
-
-  // ===========================================================================
-  // Promise like API
-  // ===========================================================================
-
-  then(listener) {
-    if (typeof listener !== 'function') {
-      return this
-    }
-
-    let promise = this._promise
-    if (promise === undefined) {
-      const requested = this._requested
-      promise = this._promise = requested
-        ? Promise.resolve()
-        : new Promise(resolve => {
-            this._resolve = resolve
-          })
-    }
-    return promise.then(listener)
-  }
-
-  // ===========================================================================
-  // Observable like API (but not compatible)
-  // ===========================================================================
-
-  subscribe(listener) {
-    if (this._requested) {
-      listener.call(this)
-      return closedSubscription
-    }
-    const listeners = this._listeners
-    if (listeners === undefined) {
-      this._listeners = [listener]
-    } else {
-      listeners.push(listener)
-    }
-    return new Subscription(this, listener)
-  }
-}

packages/xen-api/src/index.js

@@ -15,6 +15,7 @@ import {
   pTimeout,
 } from 'promise-toolbox'
 
+import * as MultiCounter from './_MultiCounter'
 import autoTransport from './transports/auto'
 import coalesceCalls from './_coalesceCalls'
 import debug from './_debug'
@@ -25,7 +26,6 @@ import isReadOnlyCall from './_isReadOnlyCall'
 import makeCallSetting from './_makeCallSetting'
 import parseUrl from './_parseUrl'
 import replaceSensitiveValues from './_replaceSensitiveValues'
-import Signal from './_Signal'
 import XapiError from './_XapiError'
 
 // ===================================================================
@@ -93,15 +93,20 @@ export class Xapi extends EventEmitter {
     this._allowUnauthorized = opts.allowUnauthorized
     this._setUrl(url)
 
-    this._connected = Signal.source()
-    this._disconnected = Signal.source()
+    this._connected = new Promise(resolve => {
+      this._resolveConnected = resolve
+    })
+    this._disconnected = Promise.resolve()
     this._sessionId = undefined
     this._status = DISCONNECTED
 
+    this._counter = MultiCounter.create()
     this._debounce = opts.debounce ?? 200
     this._objects = new Collection()
     this._objectsByRef = { __proto__: null }
-    this._objectsFetched = Signal.source()
+    this._objectsFetched = new Promise(resolve => {
+      this._resolveObjectsFetched = resolve
+    })
     this._eventWatchers = { __proto__: null }
     this._taskWatchers = { __proto__: null }
     this._watchedTypes = undefined
@@ -127,11 +132,11 @@ export class Xapi extends EventEmitter {
   // ===========================================================================
 
   get connected() {
-    return this._connected.signal
+    return this._connected
   }
 
   get disconnected() {
-    return this._disconnected.signal
+    return this._disconnected
   }
 
   get pool() {
@@ -158,7 +163,9 @@ export class Xapi extends EventEmitter {
     assert(status === DISCONNECTED)
 
     this._status = CONNECTING
-    this._disconnected = Signal.source()
+    this._disconnected = new Promise(resolve => {
+      this._resolveDisconnected = resolve
+    })
 
     try {
       await this._sessionOpen()
@@ -181,7 +188,8 @@ export class Xapi extends EventEmitter {
 
       debug('%s: connected', this._humanId)
       this._status = CONNECTED
-      this._connected.request()
+      this._resolveConnected()
+      this._resolveConnected = undefined
       this.emit(CONNECTED)
     } catch (error) {
       ignoreErrors.call(this.disconnect())
@@ -198,7 +206,9 @@ export class Xapi extends EventEmitter {
     }
 
     if (status === CONNECTED) {
-      this._connected = Signal.source()
+      this._connected = new Promise(resolve => {
+        this._resolveConnected = resolve
+      })
     } else {
       assert(status === CONNECTING)
     }
@@ -212,7 +222,8 @@ export class Xapi extends EventEmitter {
     debug('%s: disconnected', this._humanId)
 
     this._status = DISCONNECTED
-    this._disconnected.request()
+    this._resolveDisconnected()
+    this._resolveDisconnected = undefined
     this.emit(DISCONNECTED)
   }
 
@@ -663,41 +674,47 @@ export class Xapi extends EventEmitter {
   }
 
   _interruptOnDisconnect(promise) {
-    let subscription
+    let listener
     const pWrapper = new Promise((resolve, reject) => {
-      subscription = this._disconnected.signal.subscribe(() => {
-        reject(new Error('disconnected'))
-      })
       promise.then(resolve, reject)
+      this.on(
+        DISCONNECTED,
+        (listener = () => {
+          reject(new Error('disconnected'))
+        })
+      )
     })
     const clean = () => {
-      subscription.unsubscribe()
+      this.removeListener(DISCONNECTED, listener)
     }
     pWrapper.then(clean, clean)
     return pWrapper
   }
 
-  async _sessionCall(method, args, timeout) {
+  _sessionCallRetryOptions = {
+    tries: 2,
+    when: error =>
+      this._status !== DISCONNECTED && error?.code === 'SESSION_INVALID',
+    onRetry: () => this._sessionOpen(),
+  }
+  _sessionCall(method, args, timeout) {
     if (method.startsWith('session.')) {
-      throw new Error('session.*() methods are disabled from this interface')
+      return Promise.reject(
+        new Error('session.*() methods are disabled from this interface')
+      )
     }
 
-    const sessionId = this._sessionId
-    assert.notStrictEqual(sessionId, undefined)
+    return pRetry(() => {
+      const sessionId = this._sessionId
+      assert.notStrictEqual(sessionId, undefined)
 
-    const newArgs = [sessionId]
-    if (args !== undefined) {
-      newArgs.push.apply(newArgs, args)
-    }
-
-    return pRetry(
-      () => this._interruptOnDisconnect(this._call(method, newArgs, timeout)),
-      {
-        tries: 2,
-        when: { code: 'SESSION_INVALID' },
-        onRetry: () => this._sessionOpen(),
+      const newArgs = [sessionId]
+      if (args !== undefined) {
+        newArgs.push.apply(newArgs, args)
       }
-    )
+
+      return this._call(method, newArgs, timeout)
+    }, this._sessionCallRetryOptions)
   }
 
   // FIXME: (probably rare) race condition leading to unnecessary login when:
@@ -758,6 +775,10 @@ export class Xapi extends EventEmitter {
     this._objects.set(object.$id, object)
     objectsByRef[ref] = object
 
+    if (prev === undefined) {
+      ++this._counter[type]
+    }
+
     if (type === 'pool') {
       this._pool = object
 
@@ -770,10 +791,6 @@ export class Xapi extends EventEmitter {
         }
       })
     } else if (type === 'task') {
-      if (prev === undefined) {
-        ++this._nTasks
-      }
-
       const taskWatchers = this._taskWatchers
       const taskWatcher = taskWatchers[ref]
       if (taskWatcher !== undefined) {
@@ -805,6 +822,7 @@ export class Xapi extends EventEmitter {
   }
 
   async _refreshCachedRecords(types) {
+    const counter = this._counter
     const toRemoveByType = { __proto__: null }
     types.forEach(type => {
       toRemoveByType[type] = new Set()
@@ -836,8 +854,15 @@ export class Xapi extends EventEmitter {
             this._removeRecordFromCache(type, ref)
           })
 
-          if (type === 'task') {
-            this._nTasks = refs.length
+          const count = refs.length
+          if (counter[type] !== count) {
+            console.warn(
+              '_refreshCachedRecords(%s): xapi=%d != local=%d',
+              type,
+              count,
+              counter[type]
+            )
+            counter[type] = count
           }
         } catch (error) {
           // there is nothing ideal to do here, do not interrupt event
@@ -858,9 +883,7 @@ export class Xapi extends EventEmitter {
       this._objects.unset(object.$id)
       delete byRefs[ref]
 
-      if (type === 'task') {
-        --this._nTasks
-      }
+      --this._counter[type]
     }
 
     const taskWatchers = this._taskWatchers
@@ -878,8 +901,10 @@ export class Xapi extends EventEmitter {
   async _watchEvents() {
     // eslint-disable-next-line no-labels
     mainLoop: while (true) {
-      if (this._objectsFetched.signal.requested) {
-        this._objectsFetched = Signal.source()
+      if (this._resolveObjectsFetched === undefined) {
+        this._objectsFetched = new Promise(resolve => {
+          this._resolveObjectsFetched = resolve
+        })
       }
 
       await this._connected
@@ -907,7 +932,18 @@ export class Xapi extends EventEmitter {
 
       // initial fetch
       await this._refreshCachedRecords(types)
-      this._objectsFetched.request()
+      this._resolveObjectsFetched()
+      this._resolveObjectsFetched = undefined
+
+      const IGNORED_TYPES = {
+        __proto__: null,
+        message: true,
+        role: true,
+        session: true,
+        user: true,
+        VBD_metrics: true,
+        VIF_metrics: true,
+      }
 
       // event loop
       const debounce = this._debounce
@@ -941,10 +977,25 @@ export class Xapi extends EventEmitter {
         fromToken = result.token
         this._processEvents(result.events)
 
-        // detect and fix disappearing tasks (e.g. when toolstack restarts)
-        if (result.valid_ref_counts.task !== this._nTasks) {
-          await this._refreshCachedRecords(['task'])
-        }
+        // detect and fix desynchronized records
+        const localCounts = this._counter
+        const xapiCounts = result.valid_ref_counts
+        await this._refreshCachedRecords(
+          types.filter(type => {
+            if (type in IGNORED_TYPES) {
+              return false
+            }
+
+            // XAPI uses lowercased types in events, but this may change, so we
+            // handle both
+            let xapiCount = xapiCounts[type]
+            if (xapiCount === undefined) {
+              xapiCount = xapiCounts[type.toLowerCase()]
+            }
+
+            return localCounts[type] !== xapiCount
+          })
+        )
       }
     }
   }
@@ -954,8 +1005,10 @@ export class Xapi extends EventEmitter {
   //
   // It also has to manually get all objects first.
   async _watchEventsLegacy() {
-    if (this._objectsFetched.signal.requested) {
-      this._objectsFetched = Signal.source()
+    if (this._resolveObjectsFetched === undefined) {
+      this._objectsFetched = new Promise(resolve => {
+        this._resolveObjectsFetched = resolve
+      })
     }
 
     await this._connected
@@ -964,7 +1017,8 @@ export class Xapi extends EventEmitter {
 
     // initial fetch
     await this._refreshCachedRecords(types)
-    this._objectsFetched.request()
+    this._resolveObjectsFetched()
+    this._resolveObjectsFetched = undefined
 
     await this._sessionCall('event.register', [types])
 

packages/xo-server/package.json

@@ -1,6 +1,6 @@
 {
   "name": "xo-server",
-  "version": "5.38.1",
+  "version": "5.38.2",
   "license": "AGPL-3.0",
   "description": "Server part of Xen-Orchestra",
   "keywords": [
@@ -122,7 +122,7 @@
     "value-matcher": "^0.2.0",
     "vhd-lib": "^0.6.0",
     "ws": "^6.0.0",
-    "xen-api": "^0.25.0",
+    "xen-api": "^0.25.1",
     "xml2js": "^0.4.19",
     "xo-acl-resolver": "^0.4.1",
     "xo-collection": "^0.4.1",

packages/xo-server/src/xapi/index.js

@@ -869,7 +869,13 @@ export default class Xapi extends XapiBase {
   _assertHealthyVdiChains(vm) {
     const cache = { __proto__: null }
     forEach(vm.$VBDs, ({ $VDI }) => {
-      this._assertHealthyVdiChain($VDI, cache)
+      try {
+        this._assertHealthyVdiChain($VDI, cache)
+      } catch (error) {
+        error.VDI = $VDI
+        error.VM = vm
+        throw error
+      }
     })
   }
 

packages/xo-vmdk-to-vhd/src/vmdk-read-table.js

@@ -12,10 +12,10 @@ const GRAIN_ADDRESS_OFFSET = 56
  */
 export default async function readVmdkGrainTable(fileAccessor) {
   const getLongLong = (buffer, offset, name) => {
-    if (buffer.length < offset + 8) {
+    if (buffer.byteLength < offset + 8) {
       throw new Error(
         `buffer ${name} is too short, expecting ${offset + 8} minimum, got ${
-          buffer.length
+          buffer.byteLength
         }`
       )
     }
@@ -61,11 +61,12 @@ export default async function readVmdkGrainTable(fileAccessor) {
   const grainTablePhysicalSize = numGTEsPerGT * 4
   const grainDirectoryEntries = Math.ceil(grainCount / numGTEsPerGT)
   const grainDirectoryPhysicalSize = grainDirectoryEntries * 4
-  const grainDirBuffer = await fileAccessor(
-    grainDirPosBytes,
-    grainDirPosBytes + grainDirectoryPhysicalSize
+  const grainDir = new Uint32Array(
+    await fileAccessor(
+      grainDirPosBytes,
+      grainDirPosBytes + grainDirectoryPhysicalSize
+    )
   )
-  const grainDir = new Uint32Array(grainDirBuffer)
   const cachedGrainTables = []
   for (let i = 0; i < grainDirectoryEntries; i++) {
     const grainTableAddr = grainDir[i] * SECTOR_SIZE

packages/xo-web/src/common/xo/index.js

@@ -1373,11 +1373,15 @@ export const fetchVmStats = (vm, granularity) =>
 
 export const getVmsHaValues = () => _call('vm.getHaValues')
 
-export const importVm = (file, type = 'xva', data = undefined, sr) => {
+export const importVm = async (file, type = 'xva', data = undefined, sr) => {
   const { name } = file
 
   info(_('startVmImport'), name)
-
+  if (data !== undefined && data.tables !== undefined) {
+    for (const k in data.tables) {
+      data.tables[k] = await data.tables[k]
+    }
+  }
   return _call('vm.import', { type, data, sr: resolveId(sr) }).then(
     ({ $sendTo }) =>
       post($sendTo, file)
@@ -1388,8 +1392,9 @@ export const importVm = (file, type = 'xva', data = undefined, sr) => {
           success(_('vmImportSuccess'), name)
           return res.json().then(body => body.result)
         })
-        .catch(() => {
+        .catch(err => {
           error(_('vmImportFailed'), name)
+          throw err
         })
   )
 }
@@ -1429,9 +1434,11 @@ export const importVdi = async vdi => {
 export const importVms = (vms, sr) =>
   Promise.all(
     map(vms, ({ file, type, data }) =>
-      importVm(file, type, data, sr).catch(noop)
+      importVm(file, type, data, sr).catch(error => {
+        console.warn('importVms', file.name, error)
+      })
     )
-  )
+  ).then(ids => ids.filter(_ => _ !== undefined))
 
 import ExportVmModalBody from './export-vm-modal' // eslint-disable-line import/first
 export const exportVm = vm =>

packages/xo-web/src/xo-app/vm-import/index.js

@@ -237,7 +237,9 @@ const parseFile = async (file, type, func) => {
 }
 
 const getRedirectionUrl = vms =>
-  vms.length === 1
+  vms.length === 0
+    ? undefined // no redirect
+    : vms.length === 1
     ? `/vms/${vms[0]}`
     : `/home?s=${encodeURIComponent(`id:|(${vms.join(' ')})`)}&t=VM`
 

packages/xo-web/src/xo-app/vm-import/ova/index.js

@@ -217,7 +217,8 @@ async function parseTarFile(file) {
           const fileSlice = file.slice(offset, offset + header.fileSize)
           const readFile = async (start, end) =>
             readFileFragment(fileSlice, start, end)
-          data.tables[header.fileName] = await readVmdkGrainTable(readFile)
+          // storing the promise, not the value
+          data.tables[header.fileName] = readVmdkGrainTable(readFile)
         }
       }
       offset += Math.ceil(header.fileSize / 512) * 512
@@ -228,6 +229,4 @@ async function parseTarFile(file) {
   }
 }
 
-const parseOvaFile = async file => parseTarFile(file)
-
-export { parseOvaFile as default }
+export { parseTarFile as default }