feat(xen-api): rewrite from scratch

- never stop event watching no matter what's happening
- never disconnect unless explicit requested
- better handling of critical sections

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

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

@xen-orchestra/fs/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@xen-orchestra/fs",
-  "version": "0.7.1",
+  "version": "0.8.0",
   "license": "AGPL-3.0",
   "description": "The File System for Xen Orchestra backups.",
   "keywords": [],

CHANGELOG.md

@@ -1,6 +1,6 @@
 # ChangeLog
 
-## Next (2019-03-19)
+## **5.33.0** (2019-03-29)
 
 ### Enhancements
 
@@ -13,6 +13,29 @@
 - [Home] Save the current page in url [#3993](https://github.com/vatesfr/xen-orchestra/issues/3993) (PR [#3999](https://github.com/vatesfr/xen-orchestra/pull/3999))
 - [VDI] Ensure suspend VDI is destroyed when destroying a VM  [#4027](https://github.com/vatesfr/xen-orchestra/issues/4027) (PR [#4038](https://github.com/vatesfr/xen-orchestra/pull/4038))
 - [VM/disk]: Warning when 2 VDIs are on 2 different hosts' local SRs [#3911](https://github.com/vatesfr/xen-orchestra/issues/3911) (PR [#3969](https://github.com/vatesfr/xen-orchestra/pull/3969))
+- [Remotes] Benchmarks (read and write rate speed) added when remote is tested [#3991](https://github.com/vatesfr/xen-orchestra/issues/3991) (PR [#4015](https://github.com/vatesfr/xen-orchestra/pull/4015))
+- [Cloud Config] Support both NoCloud and Config Drive 2 datasources for maximum compatibility (PR [#4053](https://github.com/vatesfr/xen-orchestra/pull/4053))
+- [Advanced] Configurable cookie validity (PR [#4059](https://github.com/vatesfr/xen-orchestra/pull/4059))
+- [Plugins] Display number of installed plugins [#4008](https://github.com/vatesfr/xen-orchestra/issues/4008) (PR [#4050](https://github.com/vatesfr/xen-orchestra/pull/4050))
+- [Continuous Replication] Opt-in mode to guess VHD size, should help with XenServer 7.1 CU2 and various `VDI_IO_ERROR` errors (PR [#3726](https://github.com/vatesfr/xen-orchestra/pull/3726))
+- [VM/Snapshots] Always delete broken quiesced snapshots [#4074](https://github.com/vatesfr/xen-orchestra/issues/4074) (PR [#4075](https://github.com/vatesfr/xen-orchestra/pull/4075))
+- [Settings/Servers] Display link to pool [#4041](https://github.com/vatesfr/xen-orchestra/issues/4041) (PR [#4045](https://github.com/vatesfr/xen-orchestra/pull/4045))
+- [Import] Change wording of drop zone (PR [#4020](https://github.com/vatesfr/xen-orchestra/pull/4020))
+- [Backup NG] Ability to set the interval of the full backups [#1783](https://github.com/vatesfr/xen-orchestra/issues/1783) (PR [#4083](https://github.com/vatesfr/xen-orchestra/pull/4083))
+- [Hosts] Display a warning icon if you have XenServer license restrictions [#4091](https://github.com/vatesfr/xen-orchestra/issues/4091) (PR [#4094](https://github.com/vatesfr/xen-orchestra/pull/4094))
+- [Restore] Ability to restore a metadata backup [#4004](https://github.com/vatesfr/xen-orchestra/issues/4004) (PR [#4023](https://github.com/vatesfr/xen-orchestra/pull/4023))
+- Improve connection to XCP-ng/XenServer hosts:
+  - never disconnect by itself even in case of errors
+  - never stop watching events
+
+### Released packages
+
+- xen-api v0.25.0
+- vhd-lib v0.6.0
+- @xen-orchestra/fs v0.8.0
+- xo-server-usage-report v0.7.2
+- xo-server v5.38.1
+- xo-web v5.38.0
 
 ### Bug fixes
 
@@ -22,6 +45,16 @@
 - [Home/VM] Bulk migration: fixed VM VDIs not migrated to the selected SR [#3986](https://github.com/vatesfr/xen-orchestra/issues/3986) (PR [#3987](https://github.com/vatesfr/xen-orchestra/pull/3987))
 - [Stats] Fix cache usage with simultaneous requests [#4017](https://github.com/vatesfr/xen-orchestra/issues/4017) (PR [#4028](https://github.com/vatesfr/xen-orchestra/pull/4028))
 - [Backup NG] Fix compression displayed for the wrong backup mode (PR [#4021](https://github.com/vatesfr/xen-orchestra/pull/4021))
+- [Home] Always sort the items by their names as a secondary sort criteria [#3983](https://github.com/vatesfr/xen-orchestra/issues/3983) (PR [#4047](https://github.com/vatesfr/xen-orchestra/pull/4047))
+- [Remotes] Fixes `spawn mount EMFILE` error during backup
+- Properly redirect to sign in page instead of being stuck in a refresh loop
+- [Backup-ng] No more false positives when list matching VMs on Home page [#4078](https://github.com/vatesfr/xen-orchestra/issues/4078) (PR [#4085](https://github.com/vatesfr/xen-orchestra/pull/4085))
+- [Plugins] Properly remove optional settings when unchecking _Fill information_ (PR [#4076](https://github.com/vatesfr/xen-orchestra/pull/4076))
+- [Patches] (PR [#4077](https://github.com/vatesfr/xen-orchestra/pull/4077))
+  - Add a host to a pool: fixes the auto-patching of the host on XenServer < 7.2 [#3783](https://github.com/vatesfr/xen-orchestra/issues/3783)
+  - Add a host to a pool: homogenizes both the host and **pool**'s patches [#2188](https://github.com/vatesfr/xen-orchestra/issues/2188)
+  - Safely install a subset of patches on a pool [#3777](https://github.com/vatesfr/xen-orchestra/issues/3777)
+  - XCP-ng: no longer requires to run `yum install xcp-ng-updater` when it's already installed [#3934](https://github.com/vatesfr/xen-orchestra/issues/3934)
 
 ## **5.32.2** (2019-02-28)
 

CHANGELOG.unreleased.md

@@ -2,36 +2,9 @@
 
 ### Enhancements
 
-- [Remotes] Benchmarks (read and write rate speed) added when remote is tested [#3991](https://github.com/vatesfr/xen-orchestra/issues/3991) (PR [#4015](https://github.com/vatesfr/xen-orchestra/pull/4015))
-- [Cloud Config] Support both NoCloud and Config Drive 2 datasources for maximum compatibility (PR [#4053](https://github.com/vatesfr/xen-orchestra/pull/4053))
-- [Advanced] Configurable cookie validity (PR [#4059](https://github.com/vatesfr/xen-orchestra/pull/4059))
-- [Plugins] Display number of installed plugins [#4008](https://github.com/vatesfr/xen-orchestra/issues/4008) (PR [#4050](https://github.com/vatesfr/xen-orchestra/pull/4050))
-- [Continuous Replication] Opt-in mode to guess VHD size, should help with XenServer 7.1 CU2 and various `VDI_IO_ERROR` errors (PR [#3726](https://github.com/vatesfr/xen-orchestra/pull/3726))
-- [VM/Snapshots] Always delete broken quiesced snapshots [#4074](https://github.com/vatesfr/xen-orchestra/issues/4074) (PR [#4075](https://github.com/vatesfr/xen-orchestra/pull/4075))
-- [Settings/Servers] Display link to pool [#4041](https://github.com/vatesfr/xen-orchestra/issues/4041) (PR [#4045](https://github.com/vatesfr/xen-orchestra/pull/4045))
-- [Import] Change wording of drop zone (PR [#4020](https://github.com/vatesfr/xen-orchestra/pull/4020))
-- [Backup NG] Ability to set the interval of the full backups [#1783](https://github.com/vatesfr/xen-orchestra/issues/1783) (PR [#4083](https://github.com/vatesfr/xen-orchestra/pull/4083))
-- [Hosts] Display a warning icon if you have XenServer license restrictions [#4091](https://github.com/vatesfr/xen-orchestra/issues/4091) (PR [#4094](https://github.com/vatesfr/xen-orchestra/pull/4094))
-- [Restore] Ability to restore a metadata backup [#4004](https://github.com/vatesfr/xen-orchestra/issues/4004) (PR [#4023](https://github.com/vatesfr/xen-orchestra/pull/4023))
-
 ### Bug fixes
 
-- [Home] Always sort the items by their names as a secondary sort criteria [#3983](https://github.com/vatesfr/xen-orchestra/issues/3983) (PR [#4047](https://github.com/vatesfr/xen-orchestra/pull/4047))
-- [Remotes] Fixes `spawn mount EMFILE` error during backup
-- Properly redirect to sign in page instead of being stuck in a refresh loop
-- [Backup-ng] No more false positives when list matching VMs on Home page [#4078](https://github.com/vatesfr/xen-orchestra/issues/4078) (PR [#4085](https://github.com/vatesfr/xen-orchestra/pull/4085))
-- [Plugins] Properly remove optional settings when unchecking _Fill information_ (PR [#4076](https://github.com/vatesfr/xen-orchestra/pull/4076))
-- [Patches] (PR [#4077](https://github.com/vatesfr/xen-orchestra/pull/4077))
-  - Add a host to a pool: fixes the auto-patching of the host on XenServer < 7.2 [#3783](https://github.com/vatesfr/xen-orchestra/issues/3783)
-  - Add a host to a pool: homogenizes both the host and **pool**'s patches [#2188](https://github.com/vatesfr/xen-orchestra/issues/2188)
-  - Safely install a subset of patches on a pool [#3777](https://github.com/vatesfr/xen-orchestra/issues/3777)
-  - XCP-ng: no longer requires to run `yum install xcp-ng-updater` when it's already installed [#3934](https://github.com/vatesfr/xen-orchestra/issues/3934)
-
 ### Released packages
 
-- xen-api v0.24.6
-- vhd-lib v0.6.0
-- @xen-orchestra/fs v0.8.0
-- xo-server-usage-report v0.7.2
-- xo-server v5.38.0
-- xo-web v5.38.0
+- xo-server v5.39.0
+- xo-web v5.39.0

packages/vhd-cli/package.json

@@ -27,7 +27,7 @@
     "node": ">=6"
   },
   "dependencies": {
-    "@xen-orchestra/fs": "^0.7.1",
+    "@xen-orchestra/fs": "^0.8.0",
     "cli-progress": "^2.0.0",
     "exec-promise": "^0.7.0",
     "getopts": "^2.2.3",

packages/vhd-lib/package.json

@@ -35,7 +35,7 @@
     "@babel/core": "^7.0.0",
     "@babel/preset-env": "^7.0.0",
     "@babel/preset-flow": "^7.0.0",
-    "@xen-orchestra/fs": "^0.7.1",
+    "@xen-orchestra/fs": "^0.8.0",
     "babel-plugin-lodash": "^3.3.2",
     "cross-env": "^5.1.3",
     "execa": "^1.0.0",

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.24.6"
+    "xen-api": "^0.25.0"
   },
   "devDependencies": {
     "@babel/cli": "^7.1.5",

packages/xen-api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "xen-api",
-  "version": "0.24.6",
+  "version": "0.25.0",
   "license": "ISC",
   "description": "Connector to the Xen API",
   "keywords": [
@@ -38,7 +38,6 @@
     "event-to-promise": "^0.8.0",
     "exec-promise": "^0.7.0",
     "http-request-plus": "^0.8.0",
-    "iterable-backoff": "^0.1.0",
     "jest-diff": "^24.0.0",
     "json-rpc-protocol": "^0.13.1",
     "kindof": "^2.0.0",

packages/xen-api/src/index.js

@@ -4,26 +4,15 @@ import kindOf from 'kindof'
 import ms from 'ms'
 import httpRequest from 'http-request-plus'
 import { EventEmitter } from 'events'
-import { fibonacci } from 'iterable-backoff'
-import {
-  forEach,
-  forOwn,
-  isArray,
-  map,
-  noop,
-  omit,
-  reduce,
-  startsWith,
-} from 'lodash'
+import { isArray, map, noop, omit } from 'lodash'
 import {
   cancelable,
   defer,
   fromEvents,
   ignoreErrors,
-  pCatch,
   pDelay,
+  pRetry,
   pTimeout,
-  TimeoutError,
 } from 'promise-toolbox'
 
 import autoTransport from './transports/auto'
@@ -43,51 +32,6 @@ import XapiError from './_XapiError'
 // in seconds!
 const EVENT_TIMEOUT = 60
 
-// http://www.gnu.org/software/libc/manual/html_node/Error-Codes.html
-const NETWORK_ERRORS = {
-  // Connection has been closed outside of our control.
-  ECONNRESET: true,
-
-  // Connection has been aborted locally.
-  ECONNABORTED: true,
-
-  // Host is up but refuses connection (typically: no such service).
-  ECONNREFUSED: true,
-
-  // TODO: ??
-  EINVAL: true,
-
-  // Host is not reachable (does not respond).
-  EHOSTUNREACH: true,
-
-  // network is unreachable
-  ENETUNREACH: true,
-
-  // Connection configured timed out has been reach.
-  ETIMEDOUT: true,
-}
-
-const isNetworkError = ({ code }) => NETWORK_ERRORS[code]
-
-// -------------------------------------------------------------------
-
-const XAPI_NETWORK_ERRORS = {
-  HOST_STILL_BOOTING: true,
-  HOST_HAS_NO_MANAGEMENT_IP: true,
-}
-
-const isXapiNetworkError = ({ code }) => XAPI_NETWORK_ERRORS[code]
-
-// -------------------------------------------------------------------
-
-const areEventsLost = ({ code }) => code === 'EVENTS_LOST'
-
-const isHostSlave = ({ code }) => code === 'HOST_IS_SLAVE'
-
-const isMethodUnknown = ({ code }) => code === 'MESSAGE_METHOD_UNKNOWN'
-
-const isSessionInvalid = ({ code }) => code === 'SESSION_INVALID'
-
 // ===================================================================
 
 const { defineProperties, freeze, keys: getKeys } = Object
@@ -98,10 +42,6 @@ export const NULL_REF = 'OpaqueRef:NULL'
 
 // -------------------------------------------------------------------
 
-const getKey = o => o.$id
-
-// -------------------------------------------------------------------
-
 const RESERVED_FIELDS = {
   id: true,
   pool: true,
@@ -120,17 +60,15 @@ const CONNECTED = 'connected'
 const CONNECTING = 'connecting'
 const DISCONNECTED = 'disconnected'
 
-// timeout of XenAPI HTTP connections
-const HTTP_TIMEOUT = 24 * 3600 * 1e3
-
 // -------------------------------------------------------------------
 
 export class Xapi extends EventEmitter {
   constructor(opts) {
     super()
 
-    this._callTimeout = makeCallSetting(opts.callTimeout, 0)
+    this._callTimeout = makeCallSetting(opts.callTimeout, 60 * 60 * 1e3) // 1 hour but will be reduced in the future
     this._httpInactivityTimeout = opts.httpInactivityTimeout ?? 5 * 60 * 1e3 // 5 mins
+    this._eventPollDelay = opts.eventPollDelay ?? 60 * 1e3 // 1 min
     this._pool = null
     this._readOnly = Boolean(opts.readOnly)
     this._RecordsByType = { __proto__: null }
@@ -154,21 +92,22 @@ export class Xapi extends EventEmitter {
     this._allowUnauthorized = opts.allowUnauthorized
     this._setUrl(url)
 
-    this._connecting = undefined
     this._connected = new Promise(resolve => {
       this._resolveConnected = resolve
     })
     this._disconnected = Promise.resolve()
     this._sessionId = undefined
     this._status = DISCONNECTED
-    ;(this._objects = new Collection()).getKey = getKey
-    this._debounce = opts.debounce == null ? 200 : opts.debounce
+
+    this._debounce = opts.debounce ?? 200
+    this._objects = new Collection()
+    this._objectsByRef = { __proto__: null }
+    this._objectsFetched = new Promise(resolve => {
+      this._resolveObjectsFetched = resolve
+    })
+    this._eventWatchers = { __proto__: null }
+    this._taskWatchers = { __proto__: null }
     this._watchedTypes = undefined
-    this._watching = false
-
-    this.on(DISCONNECTED, this._clearObjects)
-    this._clearObjects()
-
     const { watchEvents } = opts
     if (watchEvents !== false) {
       if (isArray(watchEvents)) {
@@ -203,28 +142,17 @@ export class Xapi extends EventEmitter {
   }
 
   get sessionId() {
-    const id = this._sessionId
-
-    if (id === undefined || id === CONNECTING) {
-      throw new Error('sessionId is only available when connected')
-    }
-
-    return id
+    assert(this._status === CONNECTED)
+    return this._sessionId
   }
 
   get status() {
-    const id = this._sessionId
-
-    return id === undefined
-      ? DISCONNECTED
-      : id === CONNECTING
-      ? CONNECTING
-      : CONNECTED
+    return this._status
   }
 
   connect = coalesceCalls(this.connect)
   async connect() {
-    const { status } = this
+    const status = this._status
 
     if (status === CONNECTED) {
       return
@@ -232,21 +160,18 @@ export class Xapi extends EventEmitter {
 
     assert(status === DISCONNECTED)
 
-    const auth = this._auth
-
-    this._sessionId = CONNECTING
+    this._status = CONNECTING
+    this._disconnected = new Promise(resolve => {
+      this._resolveDisconnected = resolve
+    })
 
     try {
-      const [methods, sessionId] = await Promise.all([
-        this._transportCall('system.listMethods', []),
-        this._transportCall('session.login_with_password', [
-          auth.user,
-          auth.password,
-        ]),
-      ])
+      await this._sessionOpen()
 
       // Uses introspection to list available types.
-      const types = (this._types = methods
+      const types = (this._types = (await this._interruptOnDisconnect(
+        this._call('system.listMethods')
+      ))
         .filter(isGetAllRecordsMethod)
         .map(method => method.slice(0, method.indexOf('.'))))
       this._lcToTypes = { __proto__: null }
@@ -257,13 +182,10 @@ export class Xapi extends EventEmitter {
         }
       })
 
-      this._sessionId = sessionId
       this._pool = (await this.getAllRecords('pool'))[0]
 
       debug('%s: connected', this._humanId)
-      this._disconnected = new Promise(resolve => {
-        this._resolveDisconnected = resolve
-      })
+      this._status = CONNECTED
       this._resolveConnected()
       this._resolveConnected = undefined
       this.emit(CONNECTED)
@@ -275,7 +197,7 @@ export class Xapi extends EventEmitter {
   }
 
   async disconnect() {
-    const { status } = this
+    const status = this._status
 
     if (status === DISCONNECTED) {
       return
@@ -285,9 +207,6 @@ export class Xapi extends EventEmitter {
       this._connected = new Promise(resolve => {
         this._resolveConnected = resolve
       })
-
-      this._resolveDisconnected()
-      this._resolveDisconnected = undefined
     } else {
       assert(status === CONNECTING)
     }
@@ -295,11 +214,14 @@ export class Xapi extends EventEmitter {
     const sessionId = this._sessionId
     if (sessionId !== undefined) {
       this._sessionId = undefined
-      ignoreErrors.call(this._transportCall('session.logout', [sessionId]))
+      ignoreErrors.call(this._call('session.logout', [sessionId]))
     }
 
     debug('%s: disconnected', this._humanId)
 
+    this._status = DISCONNECTED
+    this._resolveDisconnected()
+    this._resolveDisconnected = undefined
     this.emit(DISCONNECTED)
   }
 
@@ -569,6 +491,10 @@ export class Xapi extends EventEmitter {
     return this._objects
   }
 
+  get objectsFetched() {
+    return this._objectsFetched
+  }
+
   // ensure we have received all events up to this call
   //
   // optionally returns the up to date object for the given ref
@@ -652,26 +578,16 @@ export class Xapi extends EventEmitter {
     // Objects ids are already UUIDs if they have one.
     const object = this._objects.all[uuid]
 
-    if (object) return object
+    if (object !== undefined) return object
 
     if (arguments.length > 1) return defaultValue
 
     throw new Error('no object with UUID: ' + uuid)
   }
 
+  // manually run events watching if set to `false` in constructor
   watchEvents() {
-    this._eventWatchers = { __proto__: null }
-
-    this._taskWatchers = { __proto__: null }
-
-    if (this.status === CONNECTED) {
-      this._watchEventsWrapper()
-    }
-
-    this.on('connected', this._watchEventsWrapper)
-    this.on('disconnected', () => {
-      this._objects.clear()
-    })
+    ignoreErrors.call(this._watchEvents())
   }
 
   watchTask(ref) {
@@ -703,13 +619,42 @@ export class Xapi extends EventEmitter {
   // Private
   // ===========================================================================
 
-  _clearObjects() {
-    ;(this._objectsByRef = { __proto__: null })[NULL_REF] = undefined
-    this._nTasks = 0
-    this._objects.clear()
-    this.objectsFetched = new Promise(resolve => {
-      this._resolveObjectsFetched = resolve
-    })
+  async _call(method, args, timeout = this._callTimeout(method, args)) {
+    const startTime = Date.now()
+    try {
+      const result = await pTimeout.call(this._transport(method, args), timeout)
+      debug(
+        '%s: %s(...) [%s] ==> %s',
+        this._humanId,
+        method,
+        ms(Date.now() - startTime),
+        kindOf(result)
+      )
+      return result
+    } catch (e) {
+      const error = e instanceof Error ? e : XapiError.wrap(e)
+
+      // do not log the session ID
+      //
+      // TODO: should log at the session level to avoid logging sensitive
+      // values?
+      const params = args[0] === this._sessionId ? args.slice(1) : args
+
+      error.call = {
+        method,
+        params: replaceSensitiveValues(params, '* obfuscated *'),
+      }
+
+      debug(
+        '%s: %s(...) [%s] =!> %s',
+        this._humanId,
+        method,
+        ms(Date.now() - startTime),
+        error
+      )
+
+      throw error
+    }
   }
 
   // return a promise which resolves to a task ref or undefined
@@ -726,41 +671,67 @@ export class Xapi extends EventEmitter {
     return Promise.resolve(task)
   }
 
-  // Medium level call: handle session errors.
-  _sessionCall(method, args, timeout = this._callTimeout(method, args)) {
-    try {
-      if (startsWith(method, 'session.')) {
-        throw new Error('session.*() methods are disabled from this interface')
-      }
+  _interruptOnDisconnect(promise) {
+    return Promise.race([
+      promise,
+      this._disconnected.then(() => {
+        throw new Error('disconnected')
+      }),
+    ])
+  }
 
-      const newArgs = [this.sessionId]
-      if (args !== undefined) {
-        newArgs.push.apply(newArgs, args)
-      }
-
-      return pTimeout.call(
-        pCatch.call(
-          this._transportCall(method, newArgs),
-          isSessionInvalid,
-          () => {
-            // XAPI is sometimes reinitialized and sessions are lost.
-            // Try to login again.
-            debug('%s: the session has been reinitialized', this._humanId)
-
-            this._sessionId = undefined
-            return this.connect().then(() => this._sessionCall(method, args))
-          }
-        ),
-        timeout
-      )
-    } catch (error) {
-      return Promise.reject(error)
+  async _sessionCall(method, args, timeout) {
+    if (method.startsWith('session.')) {
+      throw new Error('session.*() methods are disabled from this interface')
     }
+
+    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(),
+      }
+    )
+  }
+
+  // FIXME: (probably rare) race condition leading to unnecessary login when:
+  // 1. two calls using an invalid session start
+  // 2. one fails with SESSION_INVALID and renew the session by calling
+  //    `_sessionOpen`
+  // 3. the session is renewed
+  // 4. the second call fails with SESSION_INVALID which leads to a new
+  //    unnecessary renewal
+  _sessionOpen = coalesceCalls(this._sessionOpen)
+  async _sessionOpen() {
+    const { user, password } = this._auth
+    const params = [user, password]
+    this._sessionId = await pRetry(
+      () =>
+        this._interruptOnDisconnect(
+          this._call('session.login_with_password', params)
+        ),
+      {
+        tries: 2,
+        when: { code: 'HOST_IS_SLAVE' },
+        onRetry: error => {
+          this._setUrl({ ...this._url, hostname: error.params[0] })
+        },
+      }
+    )
   }
 
   _setUrl(url) {
     this._humanId = `${this._auth.user}@${url.hostname}`
-    this._call = autoTransport({
+    this._transport = autoTransport({
       allowUnauthorized: this._allowUnauthorized,
       url,
     })
@@ -779,11 +750,15 @@ export class Xapi extends EventEmitter {
     // An object's UUID can change during its life.
     const prev = objectsByRef[ref]
     let prevUuid
-    if (prev && (prevUuid = prev.uuid) && prevUuid !== object.uuid) {
+    if (
+      prev !== undefined &&
+      (prevUuid = prev.uuid) !== undefined &&
+      prevUuid !== object.uuid
+    ) {
       objects.remove(prevUuid)
     }
 
-    this._objects.set(object)
+    this._objects.set(object.$id, object)
     objectsByRef[ref] = object
 
     if (type === 'pool') {
@@ -815,6 +790,7 @@ export class Xapi extends EventEmitter {
   }
 
   _processEvents(events) {
+    const flush = this._objects.bufferEvents()
     events.forEach(event => {
       let type = event.class
       const lcToTypes = this._lcToTypes
@@ -828,6 +804,54 @@ export class Xapi extends EventEmitter {
         this._addRecordToCache(type, ref, event.snapshot)
       }
     })
+    flush()
+  }
+
+  async _refreshCachedRecords(types) {
+    const toRemoveByType = { __proto__: null }
+    types.forEach(type => {
+      toRemoveByType[type] = new Set()
+    })
+    const byRefs = this._objectsByRef
+    getKeys(byRefs).forEach(ref => {
+      const { $type } = byRefs[ref]
+      const toRemove = toRemoveByType[$type]
+      if (toRemove !== undefined) {
+        toRemove.add(ref)
+      }
+    })
+
+    const flush = this._objects.bufferEvents()
+    await Promise.all(
+      types.map(async type => {
+        try {
+          const toRemove = toRemoveByType[type]
+          const records = await this._sessionCall(`${type}.get_all_records`)
+          const refs = getKeys(records)
+          refs.forEach(ref => {
+            toRemove.delete(ref)
+
+            // we can bypass _processEvents here because they are all *add*
+            // event and all objects are of the same type
+            this._addRecordToCache(type, ref, records[ref])
+          })
+          toRemove.forEach(ref => {
+            this._removeRecordFromCache(type, ref)
+          })
+
+          if (type === 'task') {
+            this._nTasks = refs.length
+          }
+        } catch (error) {
+          // there is nothing ideal to do here, do not interrupt event
+          // handling
+          if (error?.code !== 'MESSAGE_REMOVED') {
+            console.warn('_refreshCachedRecords', type, error)
+          }
+        }
+      })
+    )
+    flush()
   }
 
   _removeRecordFromCache(type, ref) {
@@ -853,120 +877,80 @@ export class Xapi extends EventEmitter {
     }
   }
 
-  // - prevent multiple watches
-  // - swallow errors
-  async _watchEventsWrapper() {
-    if (!this._watching) {
-      this._watching = true
-      try {
-        await this._watchEvents()
-      } catch (error) {
-        console.error('_watchEventsWrapper', error)
-      }
-      this._watching = false
-    }
-  }
-
-  // TODO: cancelation
+  _watchEvents = coalesceCalls(this._watchEvents)
   async _watchEvents() {
-    this._clearObjects()
-
-    // compute the initial token for the event loop
-    //
-    // we need to do this before the initial fetch to avoid losing events
-    let fromToken
-    try {
-      fromToken = await this._sessionCall('event.inject', [
-        'pool',
-        this._pool.$ref,
-      ])
-    } catch (error) {
-      if (isMethodUnknown(error)) {
-        return this._watchEventsLegacy()
-      }
-    }
-
-    const types = this._watchedTypes || this._types
-
-    // initial fetch
-    const flush = this.objects.bufferEvents()
-    try {
-      await Promise.all(
-        types.map(async type => {
-          try {
-            // FIXME: use _transportCall to avoid auto-reconnection
-            forOwn(
-              await this._sessionCall(`${type}.get_all_records`),
-              (record, ref) => {
-                // we can bypass _processEvents here because they are all *add*
-                // event and all objects are of the same type
-                this._addRecordToCache(type, ref, record)
-              }
-            )
-          } catch (error) {
-            // there is nothing ideal to do here, do not interrupt event
-            // handling
-            if (error != null && error.code !== 'MESSAGE_REMOVED') {
-              console.warn('_watchEvents', 'initial fetch', type, error)
-            }
-          }
+    // eslint-disable-next-line no-labels
+    mainLoop: while (true) {
+      if (this._resolveObjectsFetched === undefined) {
+        this._objectsFetched = new Promise(resolve => {
+          this._resolveObjectsFetched = resolve
         })
-      )
-    } finally {
-      flush()
-    }
-    this._resolveObjectsFetched()
-
-    // event loop
-    const debounce = this._debounce
-    while (true) {
-      if (debounce != null) {
-        await pDelay(debounce)
       }
 
-      let result
+      await this._connected
+
+      // compute the initial token for the event loop
+      //
+      // we need to do this before the initial fetch to avoid losing events
+      let fromToken
       try {
-        result = await this._sessionCall(
-          'event.from',
-          [
-            types,
-            fromToken,
-            EVENT_TIMEOUT + 0.1, // must be float for XML-RPC transport
-          ],
-          EVENT_TIMEOUT * 1e3 * 1.1
-        )
+        fromToken = await this._sessionCall('event.inject', [
+          'pool',
+          this._pool.$ref,
+        ])
       } catch (error) {
-        if (error instanceof TimeoutError) {
+        if (error?.code === 'MESSAGE_METHOD_UNKNOWN') {
+          return this._watchEventsLegacy()
+        }
+
+        console.warn('_watchEvents', error)
+        await pDelay(this._eventPollDelay)
+        continue
+      }
+
+      const types = this._watchedTypes ?? this._types
+
+      // initial fetch
+      await this._refreshCachedRecords(types)
+      this._resolveObjectsFetched()
+      this._resolveObjectsFetched = undefined
+
+      // event loop
+      const debounce = this._debounce
+      while (true) {
+        await pDelay(debounce)
+
+        await this._connected
+
+        let result
+        try {
+          result = await this._sessionCall(
+            'event.from',
+            [
+              types,
+              fromToken,
+              EVENT_TIMEOUT + 0.1, // must be float for XML-RPC transport
+            ],
+            EVENT_TIMEOUT * 1e3 * 1.1
+          )
+        } catch (error) {
+          if (error?.code === 'EVENTS_LOST') {
+            // eslint-disable-next-line no-labels
+            continue mainLoop
+          }
+
+          console.warn('_watchEvents', error)
+          await pDelay(this._eventPollDelay)
           continue
         }
-        if (areEventsLost(error)) {
-          return this._watchEvents()
+
+        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'])
         }
-        throw error
-      }
-
-      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 ignoreErrors.call(
-          this._sessionCall('task.get_all_records').then(tasks => {
-            const toRemove = new Set()
-            forOwn(this.objects.all, object => {
-              if (object.$type === 'task') {
-                toRemove.add(object.$ref)
-              }
-            })
-            forOwn(tasks, (task, ref) => {
-              toRemove.delete(ref)
-              this._addRecordToCache('task', ref, task)
-            })
-            toRemove.forEach(ref => {
-              this._removeRecordFromCache('task', ref)
-            })
-          })
-        )
       }
     }
   }
@@ -975,55 +959,46 @@ export class Xapi extends EventEmitter {
   // methods.
   //
   // It also has to manually get all objects first.
-  _watchEventsLegacy() {
-    const getAllObjects = async () => {
-      const flush = this.objects.bufferEvents()
+  async _watchEventsLegacy() {
+    if (this._resolveObjectsFetched === undefined) {
+      this._objectsFetched = new Promise(resolve => {
+        this._resolveObjectsFetched = resolve
+      })
+    }
+
+    await this._connected
+
+    const types = this._watchedTypes ?? this._types
+
+    // initial fetch
+    await this._refreshCachedRecords(types)
+    this._resolveObjectsFetched()
+    this._resolveObjectsFetched = undefined
+
+    await this._sessionCall('event.register', [types])
+
+    // event loop
+    const debounce = this._debounce
+    while (true) {
+      await pDelay(debounce)
+
       try {
-        await Promise.all(
-          this._types.map(type =>
-            this._sessionCall(`${type}.get_all_records`).then(
-              objects => {
-                forEach(objects, (object, ref) => {
-                  this._addRecordToCache(type, ref, object)
-                })
-              },
-              error => {
-                if (error.code !== 'MESSAGE_REMOVED') {
-                  throw error
-                }
-              }
-            )
-          )
+        await this._connected
+        this._processEvents(
+          await this._sessionCall('event.next', undefined, EVENT_TIMEOUT * 1e3)
         )
-      } finally {
-        flush()
+      } catch (error) {
+        if (error?.code === 'EVENTS_LOST') {
+          await ignoreErrors.call(
+            this._sessionCall('event.unregister', [types])
+          )
+          return this._watchEventsLegacy()
+        }
+
+        console.warn('_watchEventsLegacy', error)
+        await pDelay(this._eventPollDelay)
       }
-      this._resolveObjectsFetched()
     }
-
-    const watchEvents = () =>
-      this._sessionCall('event.register', [['*']]).then(loop)
-
-    const loop = () =>
-      this.status === CONNECTED &&
-      this._sessionCall('event.next').then(onSuccess, onFailure)
-
-    const onSuccess = events => {
-      this._processEvents(events)
-
-      const debounce = this._debounce
-      return debounce == null ? loop() : pDelay(debounce).then(loop)
-    }
-
-    const onFailure = error => {
-      if (areEventsLost(error)) {
-        return this._sessionCall('event.unregister', [['*']]).then(watchEvents)
-      }
-
-      throw error
-    }
-
-    return getAllObjects().then(watchEvents)
   }
 
   _wrapRecord(type, ref, data) {
@@ -1116,123 +1091,6 @@ export class Xapi extends EventEmitter {
   }
 }
 
-Xapi.prototype._transportCall = reduce(
-  [
-    function(method, args) {
-      return pTimeout
-        .call(this._call(method, args), HTTP_TIMEOUT)
-        .catch(error => {
-          if (!(error instanceof Error)) {
-            error = XapiError.wrap(error)
-          }
-
-          // do not log the session ID
-          //
-          // TODO: should log at the session level to avoid logging sensitive
-          // values?
-          const params = args[0] === this._sessionId ? args.slice(1) : args
-
-          error.call = {
-            method,
-            params: replaceSensitiveValues(params, '* obfuscated *'),
-          }
-          throw error
-        })
-    },
-    call =>
-      function() {
-        let iterator // lazily created
-        const loop = () =>
-          pCatch.call(
-            call.apply(this, arguments),
-            isNetworkError,
-            isXapiNetworkError,
-            error => {
-              if (iterator === undefined) {
-                iterator = fibonacci()
-                  .clamp(undefined, 60)
-                  .take(10)
-                  .toMs()
-              }
-
-              const cursor = iterator.next()
-              if (!cursor.done) {
-                // TODO: ability to cancel the connection
-                // TODO: ability to force immediate reconnection
-
-                const delay = cursor.value
-                debug(
-                  '%s: network error %s, next try in %s ms',
-                  this._humanId,
-                  error.code,
-                  delay
-                )
-                return pDelay(delay).then(loop)
-              }
-
-              debug('%s: network error %s, aborting', this._humanId, error.code)
-
-              // mark as disconnected
-              pCatch.call(this.disconnect(), noop)
-
-              throw error
-            }
-          )
-        return loop()
-      },
-    call =>
-      function loop() {
-        return pCatch.call(
-          call.apply(this, arguments),
-          isHostSlave,
-          ({ params: [master] }) => {
-            debug(
-              '%s: host is slave, attempting to connect at %s',
-              this._humanId,
-              master
-            )
-
-            const newUrl = {
-              ...this._url,
-              hostname: master,
-            }
-            this.emit('redirect', newUrl)
-            this._url = newUrl
-
-            return loop.apply(this, arguments)
-          }
-        )
-      },
-    call =>
-      function(method) {
-        const startTime = Date.now()
-        return call.apply(this, arguments).then(
-          result => {
-            debug(
-              '%s: %s(...) [%s] ==> %s',
-              this._humanId,
-              method,
-              ms(Date.now() - startTime),
-              kindOf(result)
-            )
-            return result
-          },
-          error => {
-            debug(
-              '%s: %s(...) [%s] =!> %s',
-              this._humanId,
-              method,
-              ms(Date.now() - startTime),
-              error
-            )
-            throw error
-          }
-        )
-      },
-  ],
-  (call, decorator) => decorator(call)
-)
-
 // ===================================================================
 
 // The default value is a factory function.

packages/xo-server-usage-report/package.json

@@ -1,6 +1,6 @@
 {
   "name": "xo-server-usage-report",
-  "version": "0.7.1",
+  "version": "0.7.2",
   "license": "AGPL-3.0",
   "description": "",
   "keywords": [

packages/xo-server/package.json

@@ -1,6 +1,6 @@
 {
   "name": "xo-server",
-  "version": "5.37.0",
+  "version": "5.38.1",
   "license": "AGPL-3.0",
   "description": "Server part of Xen-Orchestra",
   "keywords": [
@@ -37,7 +37,7 @@
     "@xen-orchestra/cron": "^1.0.3",
     "@xen-orchestra/defined": "^0.0.0",
     "@xen-orchestra/emit-async": "^0.0.0",
-    "@xen-orchestra/fs": "^0.7.1",
+    "@xen-orchestra/fs": "^0.8.0",
     "@xen-orchestra/log": "^0.1.4",
     "@xen-orchestra/mixin": "^0.0.0",
     "ajv": "^6.1.1",
@@ -122,7 +122,7 @@
     "value-matcher": "^0.2.0",
     "vhd-lib": "^0.6.0",
     "ws": "^6.0.0",
-    "xen-api": "^0.24.6",
+    "xen-api": "^0.25.0",
     "xml2js": "^0.4.19",
     "xo-acl-resolver": "^0.4.1",
     "xo-collection": "^0.4.1",

packages/xo-web/package.json

@@ -1,7 +1,7 @@
 {
   "private": false,
   "name": "xo-web",
-  "version": "5.37.0",
+  "version": "5.38.0",
   "license": "AGPL-3.0",
   "description": "Web interface client for Xen-Orchestra",
   "keywords": [