diff --git a/@xen-orchestra/backups/_runners/_createStreamThrottle.mjs b/@xen-orchestra/backups/_runners/_createStreamThrottle.mjs index b7f05d8dff..1b11293d1c 100644 --- a/@xen-orchestra/backups/_runners/_createStreamThrottle.mjs +++ b/@xen-orchestra/backups/_runners/_createStreamThrottle.mjs @@ -10,6 +10,9 @@ export default function createStreamThrottle(rate) { } const group = new ThrottleGroup({ rate }) return function throttleStream(stream) { - return pipeline(stream, group.createThrottle(), noop) + const throttled = pipeline(stream, group.createThrottle(), noop) + throttled.length = stream.length + throttled.maxStreamLength = stream.maxStreamLength + return throttled } } diff --git a/@xen-orchestra/backups/_runners/_createStreamThrottle.test.mjs b/@xen-orchestra/backups/_runners/_createStreamThrottle.test.mjs new file mode 100644 index 0000000000..456d3380bd --- /dev/null +++ b/@xen-orchestra/backups/_runners/_createStreamThrottle.test.mjs @@ -0,0 +1,58 @@ +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' +import { Readable } from 'node:stream' + +import createStreamThrottle from './_createStreamThrottle.mjs' + +const makeStream = (chunks, props = {}) => { + const stream = Readable.from(chunks) + Object.assign(stream, props) + return stream +} + +const readAll = async stream => { + const chunks = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + return Buffer.concat(chunks.map(c => (typeof c === 'string' ? Buffer.from(c) : c))).toString() +} + +describe('createStreamThrottle', () => { + it('rate=0 returns identity, stream data and props intact', async () => { + const throttleStream = createStreamThrottle(0) + const stream = makeStream(['hello world'], { length: 11, maxStreamLength: 20 }) + const throttled = throttleStream(stream) + assert.strictEqual(throttled, stream) + assert.strictEqual(throttled.length, 11) + assert.strictEqual(throttled.maxStreamLength, 20) + assert.strictEqual(await readAll(throttled), 'hello world') + }) + + it('rate>0 passes data through and preserves length and maxStreamLength', async () => { + const throttleStream = createStreamThrottle(1024 * 1024) + const stream = makeStream(['hello world'], { length: 11, maxStreamLength: 20 }) + const throttled = throttleStream(stream) + assert.strictEqual(throttled.length, 11) + assert.strictEqual(throttled.maxStreamLength, 20) + assert.strictEqual(await readAll(throttled), 'hello world') + }) + + it('rate>0 handles undefined length and maxStreamLength', async () => { + const throttleStream = createStreamThrottle(1024 * 1024) + const stream = makeStream(['hello world']) + const throttled = throttleStream(stream) + assert.strictEqual(throttled.length, undefined) + assert.strictEqual(throttled.maxStreamLength, undefined) + assert.strictEqual(await readAll(throttled), 'hello world') + }) + + // regression: maxStreamLength was lost after throttling, causing S3 to ignore it and cap at 50GB + it('rate>0 preserves maxStreamLength so S3 can compute correct partSize', async () => { + const throttleStream = createStreamThrottle(1024 * 1024) + const stream = makeStream(['hello world'], { maxStreamLength: 11 }) + const throttled = throttleStream(stream) + assert.strictEqual(throttled.maxStreamLength, 11) + assert.strictEqual(await readAll(throttled), 'hello world') + }) +}) diff --git a/CHANGELOG.unreleased.md b/CHANGELOG.unreleased.md index 925f109c90..d75dcfda3e 100644 --- a/CHANGELOG.unreleased.md +++ b/CHANGELOG.unreleased.md @@ -16,6 +16,7 @@ > Users must be able to say: "I had this issue, happy to know it's fixed" - [Backups] fix alignment issue in full mirror with disk > 50GB from non encrypted to encrypted remote (PR [#10061](https://github.com/vatesfr/xen-orchestra/pull/10061)) +- [Backups] Add length to throttled stream during transfer size (PR [#10079](https://github.com/vatesfr/xen-orchestra/pull/10079)) ### Packages to release @@ -33,6 +34,7 @@ +- @xen-orchestra/backups patch - @xen-orchestra/fs patch