feat(backup/incremental mirror): don't recompress block if the compression is the same on source and destination (#10581)

* feat(backup/incremental mirror): don't recompress block if the compression is the same on source and destination

packages/vhd-lib/Vhd/VhdDirectory.js

@@ -196,13 +196,17 @@ exports.VhdDirectory = class VhdDirectory extends VhdAbstract {
       `Can't write a chunk ${partName} in ${this._path} with read permission`
     )
 
-    // in case of VhdDirectory, we want to create the file if it does not exists
-    const flags = this._opts?.flags === 'r+' ? 'w' : this._opts?.flags
     const compressed = await this.#compressor.compress(
       // the handler can write an array without concatenating it, most compressors can't
       Array.isArray(buffer) && !this.#compressor.acceptsBufferArray ? Buffer.concat(buffer) : buffer
     )
-    return this._handler.outputFile(this.#getChunkPath(partName), compressed, { flags })
+    return this.#outputChunk(partName, compressed)
+  }
+
+  #outputChunk(partName, buffer) {
+    // in case of VhdDirectory, we want to create the file if it does not exists
+    const flags = this._opts?.flags === 'r+' ? 'w' : this._opts?.flags
+    return this._handler.outputFile(this.#getChunkPath(partName), buffer, { flags })
   }
 
   // put block in subdirectories to limit impact when doing directory listing
@@ -255,6 +259,15 @@ exports.VhdDirectory = class VhdDirectory extends VhdAbstract {
       buffer,
     }
   }
+
+  /**
+   * read a block as stored (bitmap and data compressed together), to be written as is with writeCompressedBlock in a
+   * VhdDirectory using the same compression
+   */
+  async readCompressedBlock(blockId) {
+    return this._handler.readFile(this.getFullBlockPath(blockId))
+  }
+
   ensureBatSize() {
     // nothing to do in directory mode
   }
@@ -353,6 +366,16 @@ exports.VhdDirectory = class VhdDirectory extends VhdAbstract {
     setBitmap(this.#blockTable, blockId)
   }
 
+  /**
+   * write a block read with readCompressedBlock, the caller must ensure it is compressed with the same compression as
+   * this VHD
+   */
+  async writeCompressedBlock(blockId, buffer) {
+    assert.notStrictEqual(this._opts?.flags, 'r', `Can't write block ${blockId} in ${this._path} with read permission`)
+    await this.#outputChunk(this.#getBlockPath(blockId), buffer)
+    setBitmap(this.#blockTable, blockId)
+  }
+
   async _readParentLocatorData(id) {
     return (await this._readChunk('parentLocatorEntry' + id)).buffer
   }

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

@@ -12,6 +12,7 @@ import { VhdDirectory, VhdAbstract } from 'vhd-lib'
 import { unpackFooter, unpackHeader } from 'vhd-lib/Vhd/_utils.js'
 import { DEFAULT_BLOCK_SIZE } from '../_constants.js'
 import assert from 'node:assert'
+import { isCompressedBlock } from './compressedBlock.mjs'
 
 /**
  * @typedef {Object} VhdRemoteTarget
@@ -76,8 +77,20 @@ export class DiskConsumerVhdDirectory extends BaseVhd {
       let truncatedBlock = null
       await asyncEach(
         generator,
-        async ({ index, data }) => {
+        async block => {
           signal?.throwIfAborted()
+          const { index, data } = block
+          if (isCompressedBlock(block)) {
+            // writing it with another compression would produce an unreadable block
+            assert.strictEqual(
+              block.compression,
+              compression,
+              `can't write a ${block.compression} compressed block ${index} in ${dataPath}`
+            )
+            // the size of a compressed block is not related to the block size
+            await vhd.writeCompressedBlock(index, data)
+            return
+          }
           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

@@ -168,4 +168,22 @@ describe('DiskConsumerVhdStream', () => {
 
     assert(chunksReceived >= 1, 'should have received at least one chunk before abort')
   })
+
+  it('should refuse the compressed blocks of a VHD directory', async () => {
+    // a block as read from a VHD directory, see compressedBlock.mjs
+    class MockCompressedDisk extends MockDisk {
+      async readBlock(index) {
+        return { ...(await super.readBlock(index)), compression: 'brotli' }
+      }
+    }
+    const stream = await toVhdStream(new MockCompressedDisk(1, [0]))
+    await assert.rejects(
+      pipeline(stream, async function* (source) {
+        for await (const chunk of source) {
+          yield chunk
+        }
+      }),
+      /can't write a compressed block 0 in a VHD file/
+    )
+  })
 })

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

@@ -3,6 +3,8 @@ import { BaseVhd, FULL_BLOCK_BITMAP } from './BaseVhd.mjs'
 import { DEFAULT_BLOCK_SIZE } from '../_constants.js'
 import { unpackFooter, unpackHeader } from 'vhd-lib/Vhd/_utils.js'
 import { fuFooter, fuHeader, checksumStruct } from 'vhd-lib/_structs.js'
+import assert from 'node:assert'
+import { isCompressedBlock } from './compressedBlock.mjs'
 
 /**
  * @typedef {Readable & { length: number }} VhdStream
@@ -66,8 +68,10 @@ export class DiskConsumerVhdStream extends BaseVhd {
       yield header
       yield bat
       let truncatedBlock = null
-      for await (const { data, index } of blockGenerator) {
+      for await (const block of blockGenerator) {
         signal?.throwIfAborted()
+        const { data, index } = block
+        assert.ok(!isCompressedBlock(block), `can't write a compressed block ${index} in a VHD file`)
         // only the last block can be truncated
         // but the stream expect a full block
         if (truncatedBlock !== null) {

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

@@ -0,0 +1,19 @@
+// @ts-check
+
+/**
+ * @typedef {import('@xen-orchestra/disk-transform').DiskBlock} DiskBlock
+ */
+
+/**
+ * a block read as stored in a VHD directory (bitmap and data compressed together, see
+ * VhdDirectory#readCompressedBlock), it can only be written as is in a VHD directory using the same compression
+ * @typedef {DiskBlock & { compression: string }} CompressedDiskBlock
+ */
+
+/**
+ * @param {DiskBlock | CompressedDiskBlock} block
+ * @returns {block is CompressedDiskBlock}
+ */
+export function isCompressedBlock(block) {
+  return 'compression' in block && block.compression !== undefined
+}