fix(disk-transform): backup omnibus fix (#8635)

@xen-orchestra/backups/RemoteAdapter.mjs

@@ -34,6 +34,7 @@ import { watchStreamSize } from './_watchStreamSize.mjs'
 import { RemoteVhd } from './disks/RemoteVhd.mjs'
 import { openDiskChain } from './disks/openDiskChain.mjs'
 import { toVhdStream, writeToVhdDirectory } from 'vhd-lib/disk-consumer/index.mjs'
+import { ReadAhead } from '@xen-orchestra/disk-transform'
 
 export const DIR_XO_CONFIG_BACKUPS = 'xo-config-backups'
 
@@ -741,6 +742,7 @@ export class RemoteAdapter {
       disk = new RemoteVhd({ handler, path })
       await disk.init()
     }
+    disk = new ReadAhead(disk)
     return disk
   }
 

@xen-orchestra/backups/_runners/_vmRunners/IncrementalRemote.mjs

@@ -9,6 +9,7 @@ import { IncrementalRemoteWriter } from '../_writers/IncrementalRemoteWriter.mjs
 import { Disposable } from 'promise-toolbox'
 import { openVhd } from 'vhd-lib'
 import { getVmBackupDir } from '../../_getVmBackupDir.mjs'
+import { SynchronizedDisk } from '@xen-orchestra/disk-transform'
 
 const { warn } = createLogger('xo:backups:Incrementalremote')
 class IncrementalRemoteVmBackupRunner extends AbstractRemote {
@@ -68,9 +69,12 @@ class IncrementalRemoteVmBackupRunner extends AbstractRemote {
       // recompute if disks are differencing or not
       const isVhdDifferencing = {}
 
-      Object.entries(incrementalExport.disks).forEach(([key, disk]) => {
+      for (const key in incrementalExport.disks) {
+        const disk = incrementalExport.disks[key]
         isVhdDifferencing[key] = disk.isDifferencing()
-      })
+        incrementalExport.disks[key] = new SynchronizedDisk(disk)
+      }
+
       const hasDifferencingDisk = Object.values(isVhdDifferencing).includes(true)
       if (metadata.isBase === hasDifferencingDisk) {
         warn(`Metadata isBase and real disk value are different`, {
@@ -84,10 +88,18 @@ class IncrementalRemoteVmBackupRunner extends AbstractRemote {
       await this._selectBaseVm(metadata)
       await this._callWriters(writer => writer.prepare({ isBase: metadata.isBase }), 'writer.prepare()')
 
+      function fork(incrementalExport, label) {
+        const { disks, ...forked } = incrementalExport
+        forked.disks = {}
+        for (const key in disks) {
+          forked.disks[key] = disks[key].fork(label)
+        }
+        return forked
+      }
       await this._callWriters(
         writer =>
           writer.transfer({
-            deltaExport: incrementalExport,
+            deltaExport: fork(incrementalExport, writer.constructor.name + ' ' + Math.random()),
             isVhdDifferencing,
             timestamp: metadata.timestamp,
             vm: metadata.vm,

@xen-orchestra/backups/_runners/_writers/IncrementalRemoteWriter.mjs

@@ -202,8 +202,8 @@ export class IncrementalRemoteWriter extends MixinRemoteWriter(AbstractIncrement
 
     let metadataContent = await this._isAlreadyTransferred(timestamp)
     if (metadataContent !== undefined) {
-      // skip backup while being vigilant to not stuck the forked stream
-      /** @todo destroy fork  */
+      // skip backup while being vigilant to not stuck the forked disk
+      await Promise.all(Object.values(deltaExport.disks).map(async disk => disk.close()))
       return { size: 0 }
     }
 

@xen-orchestra/disk-transform/src/Disk.mts

@@ -105,7 +105,6 @@ export abstract class RandomAccessDisk extends Disk {
       await this.progressHandler?.setProgress(1)
     } finally {
       await this.progressHandler?.done()
-      await this.close()
     }
   }
 }

CHANGELOG.unreleased.md

@@ -44,6 +44,8 @@
 <!--packages-start-->
 
 - @vates/types minor
+- @xen-orchestra/backups minor
+- @xen-orchestra/disk-transform patch
 - @xen-orchestra/rest-api minor
 - @xen-orchestra/web patch
 - @xen-orchestra/web-core patch