feat(vmware-explorer): make V2V import cancellable (#9552)

@vates/nbd-client/index.mjs

@@ -353,6 +353,7 @@ export default class NbdClient {
   async readBlock(index, size = NBD_DEFAULT_BLOCK_SIZE) {
     return pRetry(() => this.#readBlock(index, size), {
       tries: this.#readBlockRetries,
+      when: error => error.code !== 'ERR_ABORTED',
       onRetry: async err => {
         warn('will retry reading block ', index, err)
         await this.reconnect()
@@ -370,7 +371,7 @@ export default class NbdClient {
    * - `length` — The size of the segment in bytes.
    * - `type` — A numeric code indicating the segment type ( 0 means data).
    */
-  /* async */ getMap() {
+  /* async */ getMap(signal) {
     return new Promise((resolve, reject) => {
       const process = spawn('nbdinfo', [
         '--json',
@@ -381,8 +382,17 @@ export default class NbdClient {
       let errText = ''
       process.stdout.on('data', data => (text += data))
       process.stderr.on('data', data => (errText += data))
-      process.on('error', reject)
+      const onAbort = () => {
+        process.kill()
+        reject(signal.reason)
+      }
+      signal?.addEventListener('abort', onAbort, { once: true })
+      process.on('error', err => {
+        signal?.removeEventListener('abort', onAbort)
+        reject(err)
+      })
       process.on('close', code => {
+        signal?.removeEventListener('abort', onAbort)
         if (code !== 0) {
           return reject(new Error(`Error during getMap (code: ${code}): ${errText}`))
         }

@vates/nbd-client/multi.mjs

@@ -138,8 +138,8 @@ export default class MultiNbdClient {
    * - `length` — The size of the segment in bytes.
    * - `type` — A numeric code indicating the segment type (0 means no data).
    */
-  async getMap() {
+  async getMap(signal) {
     // ask the map from one of the connected client
-    return this.#clients[Math.floor(this.#clients.length * Math.random())].getMap()
+    return this.#clients[Math.floor(this.#clients.length * Math.random())].getMap(signal)
   }
 }

@xen-orchestra/disk-transform/package.json

@@ -34,7 +34,6 @@
     "node": ">=20.17"
   },
   "dependencies": {
-    "vhd-lib": "^4.14.7",
     "@vates/generator-toolbox": "^1.1.0"
   }
 }

@xen-orchestra/qcow2/src/consumer/ConsumerQcowStream.integ.mts

@@ -192,6 +192,59 @@ describe('QCOW2 Stream Generation', () => {
     }
   })
 
+  it('should abort stream when signal is aborted before start', async () => {
+    const disk = new MockDisk(
+      128,
+      Array.from({ length: 128 }, (_, i) => i)
+    )
+    const controller = new AbortController()
+    controller.abort()
+
+    const stream = toQcow2Stream(disk, { signal: controller.signal })
+    await assert.rejects(
+      async () => {
+        for await (const _chunk of stream) {
+          assert.fail(`should not have any chunk on a aborted stream , got ${_chunk?.length} bytes`)
+        }
+      },
+      (err: unknown) => {
+        assert(err instanceof Error)
+        assert.strictEqual((err as NodeJS.ErrnoException).name, 'AbortError')
+        return true
+      }
+    )
+  })
+
+  it('should abort stream mid-way when signal is aborted', async () => {
+    const NB_BLOCKS = 1024
+    const disk = new MockDisk(
+      NB_BLOCKS,
+      Array.from({ length: NB_BLOCKS }, (_, i) => i)
+    )
+    const controller = new AbortController()
+
+    const stream = toQcow2Stream(disk, { signal: controller.signal })
+    let chunksReceived = 0
+
+    await assert.rejects(
+      async () => {
+        for await (const _chunk of stream) {
+          chunksReceived++
+          if (chunksReceived === 1) {
+            controller.abort()
+          }
+        }
+      },
+      (err: unknown) => {
+        assert(err instanceof Error)
+        assert.strictEqual((err as NodeJS.ErrnoException).name, 'AbortError')
+        return true
+      }
+    )
+
+    assert(chunksReceived >= 1, 'should have received at least one chunk before abort')
+  })
+
   it('should handle unaligned disks', async () => {
     const NB_BLOCKS = 5
     const RATIO = 3

@xen-orchestra/qcow2/src/consumer/ConsumerQcowStream.mts

@@ -236,9 +236,10 @@ export class QcowStreamGenerator {
 
   /**
    * Creates a Readable stream from the generator
+   * @param signal Optional AbortSignal to cancel the stream
    * @returns Readable stream with optional length property
    */
-  stream(): WithLength<Readable> {
+  stream(signal?: AbortSignal): WithLength<Readable> {
     const disk = this.#disk
     const nbAllocatedBlocks = disk.getBlockIndexes().length
     const nbTotalBlock = Math.ceil(disk.getVirtualSize() / disk.getBlockSize())
@@ -269,6 +270,7 @@ export class QcowStreamGenerator {
     const self = this
     async function* generator(): AsyncGenerator<Buffer, void, unknown> {
       // Yield all parts in order
+      signal?.throwIfAborted()
       yield* self.#trackAndYield(header)
       assert.strictEqual(self.#offset, CLUSTER_SIZE, 'header aligned')
       yield* self.#yieldRefCounts(expectedStreamLength / CLUSTER_SIZE)
@@ -286,6 +288,7 @@ export class QcowStreamGenerator {
 
       let truncatedBlock: DiskBlock | null = null
       for await (const { index, data } of disk.diskBlocks()) {
+        signal?.throwIfAborted()
         if (index < previous) {
           throw new Error('Qcow can only be generated from sorted disk')
         }
@@ -322,9 +325,11 @@ export class QcowStreamGenerator {
 /**
  * Creates a QCOW2 stream from a RandomAccessDisk
  * @param disk The disk to convert
+ * @param options Optional options
+ * @param options.signal Optional AbortSignal to cancel the stream
  * @returns Readable stream of QCOW2 data
  */
-export function toQcow2Stream(disk: Disk): Readable {
+export function toQcow2Stream(disk: Disk, { signal }: { signal?: AbortSignal } = {}): Readable {
   const generator = new QcowStreamGenerator(disk)
-  return generator.stream()
+  return generator.stream(signal)
 }

@xen-orchestra/vmware-explorer/esxi.mjs

@@ -108,7 +108,7 @@ export default class Esxi extends EventEmitter {
     })
   }
 
-  async #fetch(url, headers = {}) {
+  async #fetch(url, headers = {}, signal) {
     if (this.#cookies) {
       headers.cookie = this.#cookies
     } else {
@@ -119,6 +119,7 @@ export default class Esxi extends EventEmitter {
       method: 'GET',
       headers,
       highWaterMark: 10 * 1024 * 1024,
+      signal,
     })
     if (res.status < 200 || res.status >= 300) {
       const error = new Error(res.status + ' ' + res.statusText + ' ' + url)
@@ -131,7 +132,7 @@ export default class Esxi extends EventEmitter {
     return res
   }
 
-  async #download(dataStore, path, range) {
+  async #download(dataStore, path, range, signal) {
     strictEqual(this.#ready, true)
     const url = new URL('https://localhost')
     url.host = this.#host
@@ -143,17 +144,18 @@ export default class Esxi extends EventEmitter {
       headers['content-type'] = 'multipart/byteranges'
       headers.Range = 'bytes=' + range
     }
-    return this.#fetch(url, headers)
+    return this.#fetch(url, headers, signal)
   }
 
-  async download(dataStore, path, range) {
+  async download(dataStore, path, range, signal) {
     let tries = 5
     let lastError
     while (tries > 0) {
       try {
-        const res = await this.#download(dataStore, path, range)
+        const res = await this.#download(dataStore, path, range, signal)
         return res
       } catch (error) {
+        signal?.throwIfAborted()
         warn('got error , will retry in 2 seconds', { error })
         lastError = error
       }
@@ -616,7 +618,7 @@ export default class Esxi extends EventEmitter {
     }
   }
 
-  async #getDataMapFromVddk(vmId, datastoreName, diskPath) {
+  async #getDataMapFromVddk(vmId, datastoreName, diskPath, signal) {
     let nbdClient
     try {
       const start = Date.now()
@@ -626,6 +628,7 @@ export default class Esxi extends EventEmitter {
       })
 
       info(`nbd server for data map spawned`)
+      signal?.throwIfAborted()
 
       nbdClient = new NbdClient(nbdInfoSpawn.nbdInfos)
 
@@ -633,7 +636,7 @@ export default class Esxi extends EventEmitter {
 
       info(`nbd client for data map connected`)
 
-      const dataMap = await nbdClient.getMap()
+      const dataMap = await nbdClient.getMap(signal)
 
       info(
         `got the data map of the single disk in ${Math.round((Date.now() - start) / 1000)} seconds ,${dataMap.length} blocks`
@@ -648,8 +651,8 @@ export default class Esxi extends EventEmitter {
     }
   }
 
-  async #getDataMapFromCowd(datastoreName, diskPath) {
-    const descriptorResponse = await this.download(datastoreName, diskPath, '0-512')
+  async #getDataMapFromCowd(datastoreName, diskPath, signal) {
+    const descriptorResponse = await this.download(datastoreName, diskPath, '0-512', signal)
     const descriptorBlob = await new Response(descriptorResponse.body).blob()
     const descriptorBytes = new Uint8Array(await descriptorBlob.arrayBuffer()).slice(0, 512)
 
@@ -658,7 +661,7 @@ export default class Esxi extends EventEmitter {
     const diskPathArray = diskPath.split('/')
     const extentPath = diskPathArray.slice(0, -1).join('/') + '/' + parsedDescriptor.fileName
 
-    const extentHeaderResponse = await this.download(datastoreName, extentPath, `0-2048`)
+    const extentHeaderResponse = await this.download(datastoreName, extentPath, `0-2048`, signal)
     const extentHeaderBlob = await new Response(extentHeaderResponse.body).blob()
     const extentHeaderBuffer = Buffer.from(await extentHeaderBlob.arrayBuffer())
 
@@ -666,7 +669,12 @@ export default class Esxi extends EventEmitter {
 
     const extentNumGdEntries = extentHeaderBuffer.readUInt32LE(24)
 
-    const extentGDResponse = await this.download(datastoreName, extentPath, `2048-${2048 + extentNumGdEntries * 4}`)
+    const extentGDResponse = await this.download(
+      datastoreName,
+      extentPath,
+      `2048-${2048 + extentNumGdEntries * 4}`,
+      signal
+    )
     const extentGDBlob = await new Response(extentGDResponse.body).blob()
     const extentGDBuffer = Buffer.from(await extentGDBlob.arrayBuffer())
 
@@ -689,13 +697,14 @@ export default class Esxi extends EventEmitter {
     return dataMap
   }
 
-  async getDataMap(vmId, datastoreName, diskPath) {
+  async getDataMap(vmId, datastoreName, diskPath, signal) {
     try {
       // We await the result of getDataMapFromVddk so we can catch errors and fallback to the direct metadata reading.
-      return await this.#getDataMapFromVddk(vmId, datastoreName, diskPath)
+      return await this.#getDataMapFromVddk(vmId, datastoreName, diskPath, signal)
     } catch (error) {
+      signal?.throwIfAborted()
       warn('error while getting datamap from vddk, fall back to a direct metadata reading', error)
-      return this.#getDataMapFromCowd(datastoreName, diskPath)
+      return this.#getDataMapFromCowd(datastoreName, diskPath, signal)
     }
   }
 }

CHANGELOG.unreleased.md

@@ -11,6 +11,8 @@
 
 > Users must be able to say: "Nice enhancement, I'm eager to test it"
 
+- [V2V] Make import cancellable (PR [#9552](https://github.com/vatesfr/xen-orchestra/pull/9552))
+
 - [MCP] Support token authentication via `XO_TOKEN` environment variable as an alternative to username/password (PR [#9577](https://github.com/vatesfr/xen-orchestra/pull/9577))
 - [MCP] Add `list_vdis` tool to list virtual disks (PR [#9559](https://github.com/vatesfr/xen-orchestra/pull/9559))
 - [Replication] Reuse the same VM as an incremental replication target (PR [#9524](https://github.com/vatesfr/xen-orchestra/pull/9524))
@@ -46,13 +48,17 @@
 <!--packages-start-->
 
 - @vates/generator-toolbox patch
+- @vates/nbd-client minor
 - @vates/types minor
 - @xen-orchestra/backups minor
 - @xen-orchestra/disk-transform patch
 - @xen-orchestra/fs minor
 - @xen-orchestra/mcp minor
+- @xen-orchestra/qcow2 minor
 - @xen-orchestra/rest-api minor
+- @xen-orchestra/vmware-explorer minor
 - @xen-orchestra/web patch
+- vhd-lib minor
 - xo-server minor
 - xo-server-openmetrics minor
 

packages/vhd-lib/disk-consumer/DiskConsumerVhdDirectory.mjs

@@ -40,14 +40,16 @@ export class DiskConsumerVhdDirectory extends BaseVhd {
   }
 
   /**
+   * @param {AbortSignal} [signal]
    * @returns {Promise<void>}
    */
-  async write() {
+  async write(signal) {
     const { handler, path, compression, flags, validator, concurrency } = this.#target
     const dataPath = `${dirname(path)}/data/${uuidv4()}.vhd`
     const uid = 'to stream ' + Math.random()
     let generator
     try {
+      signal?.throwIfAborted()
       generator = this.source.diskBlocks(uid)
       await handler.mktree(dataPath)
       const vhd = new VhdDirectory(handler, dataPath, { flags, compression })
@@ -61,6 +63,7 @@ export class DiskConsumerVhdDirectory extends BaseVhd {
       await asyncEach(
         generator,
         async ({ index, data }) => {
+          signal?.throwIfAborted()
           if (truncatedBlock !== null) {
             throw new Error(
               `Expecting a ${DEFAULT_BLOCK_SIZE} bytes block, got a ${truncatedBlock.data.length}, for index ${truncatedBlock.index}`

packages/vhd-lib/disk-consumer/DiskConsumerVhdStream.integ.mjs

@@ -0,0 +1,171 @@
+import { describe, it } from 'node:test'
+import assert from 'node:assert/strict'
+import fs from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { pipeline } from 'node:stream/promises'
+import execa from 'execa'
+import { RandomAccessDisk } from '@xen-orchestra/disk-transform'
+import computeGeometryForSize from '../_computeGeometryForSize.js'
+
+import { toVhdStream } from './index.mjs'
+
+const DEFAULT_BLOCK_SIZE = 0x00200000 // 2MB, from vhd-lib spec
+
+class MockDisk extends RandomAccessDisk {
+  #size
+  #blockIndexes
+
+  constructor(nbBlocks, blockIndexes) {
+    super()
+    // VHD virtual size must match the CHS-geometry-aligned actual size so that
+    // qemu-img derives the same maxTableEntries from both footer and geometry fields
+    const { actualSize } = computeGeometryForSize(nbBlocks * DEFAULT_BLOCK_SIZE)
+    this.#size = actualSize
+    this.#blockIndexes = blockIndexes
+  }
+
+  async readBlock(index) {
+    const remaining = this.#size - DEFAULT_BLOCK_SIZE * index
+    return {
+      index,
+      data: Buffer.alloc(Math.min(remaining, DEFAULT_BLOCK_SIZE), index % 256),
+    }
+  }
+
+  getVirtualSize() {
+    return this.#size
+  }
+
+  getBlockSize() {
+    return DEFAULT_BLOCK_SIZE
+  }
+
+  isDifferencing() {
+    return false
+  }
+
+  getBlockIndexes() {
+    return this.#blockIndexes
+  }
+
+  hasBlock(index) {
+    return this.#blockIndexes.includes(index)
+  }
+
+  async init() {}
+  async close() {}
+}
+
+async function writeStreamToFile(stream, filePath) {
+  const fileHandle = await fs.open(filePath, 'w')
+  try {
+    await pipeline(stream, fileHandle.createWriteStream())
+  } finally {
+    await fileHandle.close()
+  }
+}
+
+describe('DiskConsumerVhdStream', () => {
+  it('should generate valid VHD files', async () => {
+    const testCases = [
+      { name: 'empty disk', nbBlocks: 8, blockIndexes: [] },
+      { name: 'sparse disk', nbBlocks: 8, blockIndexes: [0, 3, 7] },
+      { name: 'fully allocated disk', nbBlocks: 4, blockIndexes: Array.from({ length: 4 }, (_, i) => i) },
+    ]
+
+    for (const { name, nbBlocks, blockIndexes } of testCases) {
+      const disk = new MockDisk(nbBlocks, blockIndexes)
+      const tmpVhd = join(tmpdir(), `test-vhd-${Date.now()}-${name.replace(/\s+/g, '-')}.vhd`)
+      const tmpRaw = `${tmpVhd}.raw`
+      try {
+        const stream = await toVhdStream(disk)
+        await writeStreamToFile(stream, tmpVhd)
+
+        // qemu-img check doesn't support vpc format; convert to raw and verify
+        // the output size matches the declared virtual size
+        await execa('qemu-img', ['convert', '-f', 'vpc', '-O', 'raw', tmpVhd, tmpRaw])
+        const { size } = await fs.stat(tmpRaw)
+        assert.strictEqual(size, disk.getVirtualSize(), `Raw size mismatch for ${name}`)
+      } finally {
+        await fs.unlink(tmpVhd).catch(() => {})
+        await fs.unlink(tmpRaw).catch(() => {})
+      }
+    }
+  })
+
+  it('should generate files with correct metadata', async () => {
+    const NB_BLOCKS = 4
+    const disk = new MockDisk(NB_BLOCKS, [0, 1])
+    const tmpVhd = join(tmpdir(), `test-vhd-metadata-${Date.now()}.vhd`)
+    const tmpRaw = `${tmpVhd}.raw`
+    try {
+      const stream = await toVhdStream(disk)
+      await writeStreamToFile(stream, tmpVhd)
+
+      const { stdout } = await execa('qemu-img', ['info', tmpVhd])
+      assert.match(stdout, /file format: vpc/, 'File format should be vpc (VHD)')
+
+      await execa('qemu-img', ['convert', '-f', 'vpc', '-O', 'raw', tmpVhd, tmpRaw])
+      const { size } = await fs.stat(tmpRaw)
+      assert.strictEqual(size, disk.getVirtualSize(), 'Raw size should match declared virtual size')
+    } finally {
+      await fs.unlink(tmpVhd).catch(() => {})
+      await fs.unlink(tmpRaw).catch(() => {})
+    }
+  })
+
+  it('should abort stream when signal is aborted before start', async () => {
+    const disk = new MockDisk(
+      8,
+      Array.from({ length: 8 }, (_, i) => i)
+    )
+    const controller = new AbortController()
+    controller.abort()
+
+    const stream = await toVhdStream(disk, { signal: controller.signal })
+    await assert.rejects(
+      async () => {
+        for await (const _chunk of stream) {
+          assert.fail(`should not have any chunk on a aborted stream , got ${_chunk?.length} bytes`)
+        }
+      },
+      err => {
+        assert(err instanceof Error)
+        assert.strictEqual(err.name, 'AbortError')
+        return true
+      }
+    )
+  })
+
+  it('should abort stream mid-way when signal is aborted', async () => {
+    const NB_BLOCKS = 16
+    const disk = new MockDisk(
+      NB_BLOCKS,
+      Array.from({ length: NB_BLOCKS }, (_, i) => i)
+    )
+    const controller = new AbortController()
+
+    const stream = await toVhdStream(disk, { signal: controller.signal })
+    let chunksReceived = 0
+
+    await assert.rejects(
+      async () => {
+        // eslint-disable-next-line no-unused-vars
+        for await (const _chunk of stream) {
+          chunksReceived++
+          if (chunksReceived === 1) {
+            controller.abort()
+          }
+        }
+      },
+      err => {
+        assert(err instanceof Error)
+        assert.strictEqual(err.name, 'AbortError')
+        return true
+      }
+    )
+
+    assert(chunksReceived >= 1, 'should have received at least one chunk before abort')
+  })
+})

packages/vhd-lib/disk-consumer/DiskConsumerVhdStream.mjs

@@ -11,9 +11,10 @@ import { DEFAULT_BLOCK_SIZE } from '../_constants.js'
  */
 export class DiskConsumerVhdStream extends BaseVhd {
   /**
+   * @param {AbortSignal} [signal]
    * @returns {VhdStream}
    */
-  async toStream() {
+  async toStream(signal) {
     const footer = this.computeVhdFooter()
     const header = this.computeVhdHeader()
     const { bat, fileSize } = this.computeVhdBatAndFileSize() // the bat contains the calculated position of the futures blocks
@@ -21,11 +22,13 @@ export class DiskConsumerVhdStream extends BaseVhd {
     const blockGenerator = this.source.diskBlocks(uid)
     const EXPECTED_FULL_BUFFER_SIZE = DEFAULT_BLOCK_SIZE + FULL_BLOCK_BITMAP.length
     async function* generator() {
+      signal?.throwIfAborted()
       yield footer
       yield header
       yield bat
       let truncatedBlock = null
       for await (const { data, index } of blockGenerator) {
+        signal?.throwIfAborted()
         // only the last block can be truncated
         // but the stream expect a full block
         if (truncatedBlock !== null) {

packages/vhd-lib/disk-consumer/index.mjs

@@ -20,11 +20,12 @@ import { DiskConsumerVhdDirectory } from './DiskConsumerVhdDirectory.mjs'
 /**
  *
  * @param {Disk} disk
+ * @param {{ signal?: AbortSignal }} [options]
  * @returns {Promise<Readable>}
  */
-export async function toVhdStream(disk) {
+export async function toVhdStream(disk, { signal } = {}) {
   const consumer = new DiskConsumerVhdStream(disk)
-  return consumer.toStream()
+  return consumer.toStream(signal)
 }
 
 /**
@@ -32,8 +33,9 @@ export async function toVhdStream(disk) {
  * @param {object} param
  * @param {Disk} param.disk
  * @param {VhdRemoteTarget} param.target
+ * @param {AbortSignal} [param.signal]
  */
-export async function writeToVhdDirectory({ disk, target }) {
+export async function writeToVhdDirectory({ disk, target, signal }) {
   const consumer = new DiskConsumerVhdDirectory(disk, target)
-  await consumer.write()
+  await consumer.write(signal)
 }

packages/vhd-lib/package.json

@@ -33,6 +33,7 @@
     "uuid": "^9.0.0"
   },
   "devDependencies": {
+    "@xen-orchestra/disk-transform": "^1.2.1",
     "@xen-orchestra/fs": "^4.6.7",
     "execa": "^5.0.0",
     "get-stream": "^6.0.0",
@@ -41,7 +42,7 @@
   },
   "scripts": {
     "postversion": "npm publish",
-    "test-integration": "node --test ./*.integ.js tests/*.integ.js Vhd/*.integ.js"
+    "test-integration": "node --test ./*.integ.js tests/*.integ.js Vhd/*.integ.js disk-consumer/*.integ.mjs"
   },
   "author": {
     "name": "Vates SAS",

packages/xo-server/src/xo-mixins/vmware/importDisksfromDatastore.mjs

@@ -9,25 +9,31 @@ import { toQcow2Stream } from '@xen-orchestra/qcow2'
 const { warn } = createLogger('xo:importdiskfromdatastore')
 
 export async function importStream({ esxi, dataMap, disk, vmId, format }, consumerCallback) {
+  const signal = Task.abortSignal
   const { datastore: datastoreName, diskPath } = disk
 
   let vmdk
+  let stream
   try {
     // we read the data from the full chain to ensure we don't have partial blocks ( blocks with 0 when clusters are in parent only)
     const { nbdInfos } = await esxi.spanwNbdKitProcess(vmId, `[${datastoreName}] ${diskPath}`)
+    signal?.throwIfAborted()
     vmdk = new NbdDisk(nbdInfos, READ_BLOCK_SIZE, { dataMap })
 
     await vmdk.init()
+    signal?.throwIfAborted()
     vmdk = new ReadAhead(vmdk)
-    let stream
+
     if (format === VDI_FORMAT_QCOW2) {
-      stream = await toQcow2Stream(vmdk)
+      stream = await toQcow2Stream(vmdk, { signal })
     } else {
-      stream = await toVhdStream(vmdk)
+      stream = await toVhdStream(vmdk, { signal })
     }
+    Task.info(`got source stream for ${diskPath}`)
     await consumerCallback(stream)
     return Math.round((vmdk.getNbGeneratedBlock() * vmdk.getBlockSize()) / 1024 / 1024)
   } catch (err) {
+    stream?.destroy(err)
     Task.warning(err)
     throw err
   } finally {
@@ -86,7 +92,7 @@ async function importDiskChain({ esxi, sr, vm, chainByNode, userdevice, vmId })
     existingVdi = diskIsAlreadyImported(existingVdis, existingDisk)
     Task.info(`found a previous import`, { vdiRef: existingVdi.$ref })
 
-    dataMap = await esxi.getDataMap(vmId, datastoreName, diskPath)
+    dataMap = await esxi.getDataMap(vmId, datastoreName, diskPath, Task.abortSignal)
   } else {
     Task.info(`no reference disk found, fall back a full import`)
   }

packages/xo-server/src/xo-mixins/vmware/index.mjs

@@ -209,9 +209,12 @@ export default class MigrateVm {
         )
       }
     }
+    Task.abortSignal?.throwIfAborted()
     const vm = await this.#createVmAndNetworks($defer, { metadata, networkId, template, xapi })
+    Task.abortSignal?.throwIfAborted()
     await xapi.barrier()
     await this.#importDisks($defer, { esxi, metadata, stopSource, vm, sr, vmId })
+    Task.abortSignal?.throwIfAborted()
     await Task.run({ properties: { name: 'Finishing transfer' } }, async () => {
       // remove the importing in label
       await vm.set_name_label(metadata.name_label)