fix(xen-api): mark long running calls as blocking

Fixes https://xcp-ng.org/forum/post/77066

packages/xen-api/index.mjs

@@ -773,10 +773,10 @@ export class Xapi extends EventEmitter {
   // Private
   // ===========================================================================
 
-  async _call(method, args = [], timeout = this._callTimeout(method, args)) {
+  async _call(method, args = [], { timeout = this._callTimeout(method, args), ...requestOpts } = {}) {
     const startTime = Date.now()
     try {
-      const result = await pTimeout.call(this._addSyncStackTrace(this._transport(method, args)), timeout)
+      const result = await pTimeout.call(this._addSyncStackTrace(this._transport(method, args, requestOpts)), timeout)
       debug('%s: %s(...) [%s] ==> %s', this._humanId, method, ms(Date.now() - startTime), kindOf(result))
       return result
     } catch (error) {
@@ -1209,7 +1209,10 @@ export class Xapi extends EventEmitter {
               fromToken,
               EVENT_TIMEOUT + 0.1, // must be float for XML-RPC transport
             ],
-            EVENT_TIMEOUT * 1e3 * 1.1
+            {
+              blocking: true,
+              timeout: EVENT_TIMEOUT * 1e3 * 1.1,
+            }
           )
           this._lastEventFetchedTimestamp = Date.now()
           this._watchEventsError = undefined
@@ -1270,7 +1273,12 @@ export class Xapi extends EventEmitter {
 
       try {
         await this._connected
-        this._processEvents(await this._roCall('event.next', undefined, EVENT_TIMEOUT * 1e3))
+        this._processEvents(
+          await this._call('event.next', undefined, {
+            blocking: true,
+            timeout: EVENT_TIMEOUT * 1e3,
+          })
+        )
       } catch (error) {
         if (error?.code === 'EVENTS_LOST') {
           await ignoreErrors.call(this._sessionCall('event.unregister', [types]))

packages/xen-api/transports/auto.mjs

@@ -13,14 +13,14 @@ export default opts => {
     const current = factories[i++](opts)
     if (i < length) {
       const currentI = i
-      call = (method, args) =>
-        current(method, args).catch(error => {
+      call = (...args) =>
+        current(...args).catch(error => {
           if (error instanceof UnsupportedTransport) {
             if (currentI === i) {
               // not changed yet
               create()
             }
-            return call(method, args)
+            return call(...args)
           }
 
           throw error
@@ -31,5 +31,5 @@ export default opts => {
   }
   create()
 
-  return (method, args) => call(method, args)
+  return (...args) => call(...args)
 }

packages/xen-api/transports/json-rpc.mjs

@@ -9,8 +9,10 @@ import UnsupportedTransport from './_UnsupportedTransport.mjs'
 export default ({ dispatcher, url }) => {
   url = new URL('./jsonrpc', Object.assign(new URL('http://localhost'), url))
 
-  return async function (method, args) {
+  return async function (method, args, requestOpts) {
     const res = await request(url, {
+      ...requestOpts,
+
       dispatcher,
       body: format.request(0, method, args),
       headers: {

packages/xen-api/transports/xml-rpc.mjs

@@ -27,8 +27,10 @@ const parseResult = result => {
 export default ({ dispatcher, url }) => {
   url = new URL('./xmlrpc', Object.assign(new URL('http://localhost'), url))
 
-  return async function (method, args) {
+  return async function (method, args, requestOpts) {
     const res = await request(url, {
+      ...requestOpts,
+
       dispatcher,
       body: formatXml(
         xmlRpcFormatter.format_methodCall({