From 14e4c7d62cb8819ba739a33231101a9f799b19c2 Mon Sep 17 00:00:00 2001 From: Florent BEAUCHAMP Date: Fri, 28 Aug 2026 10:57:17 +0200 Subject: [PATCH] fix(backups): timeout NBD transfer omnibus fixes (#10306) * fix(backups): timeout was not used correctly for NBD transfers * fix(nbd-client): evict a dead client in multi instead actual code was retrying 5 times, each time with a 60s timeout multi also have its own retry on conneciton since the client used is determnistic for a block, any stalled client will be hammerred for a lot of time before giving up this PR evict a nbd client that is failing to read a block (with its retry logic) * fix(nbd-client/multi): better spread read accross client the actual code derivate the clientId from the block index we have no information on the block distribution, only hopes that it's random enough this PR introduce a mechanism to use a round robin to select the next client * fix: correct fall back to full with /without nbd on qcow2 --- @vates/nbd-client/multi.mjs | 28 ++++- @vates/nbd-client/multi.test.mjs | 130 ++++++++++++++++++++ @vates/nbd-client/package.json | 3 +- @vates/nbd-client/tests/nbdclient.integ.mjs | 8 +- @xen-orchestra/xapi/disks/Xapi.mjs | 98 +++++++++------ @xen-orchestra/xapi/disks/XapiStreamNbd.mjs | 8 +- @xen-orchestra/xapi/disks/XapiVhdCbt.mjs | 1 + @xen-orchestra/xapi/disks/utils.mjs | 1 + CHANGELOG.unreleased.md | 3 + 9 files changed, 238 insertions(+), 42 deletions(-) create mode 100644 @vates/nbd-client/multi.test.mjs diff --git a/@vates/nbd-client/multi.mjs b/@vates/nbd-client/multi.mjs index 956870cd9d..38089237bf 100644 --- a/@vates/nbd-client/multi.mjs +++ b/@vates/nbd-client/multi.mjs @@ -7,6 +7,7 @@ const { warn } = createLogger('vates:nbd-client:multi') export default class MultiNbdClient { #clients = [] #nbdConcurrency + #nextClient = 0 #options #readAhead #settings @@ -97,8 +98,31 @@ export default class MultiNbdClient { * @returns {Promise} */ async readBlock(index, size = NBD_DEFAULT_BLOCK_SIZE) { - const clientId = index % this.#clients.length - return this.#clients[clientId].readBlock(index, size) + const clientId = this.#nextClient++ % this.#clients.length + const client = this.#clients[clientId] + try { + return await client.readBlock(index, size) + } catch (err) { + // client.readBlock() already exhausted its own retries/reconnects: this connection is dead. + // Evict it so future reads stop being routed to it, and retry this read on a surviving + // client, since the data is usually still reachable through the others. + this.#evict(client) + if (this.#clients.length === 0) { + throw err + } + warn(`evicted a dead nbd client, retrying block ${index} on a remaining client`, { err }) + return this.readBlock(index, size) + } + } + + // no-op if `client` was already evicted by a concurrent failed read on the same client + #evict(client) { + const i = this.#clients.indexOf(client) + if (i === -1) { + return + } + this.#clients.splice(i, 1) + client.disconnect().catch(() => {}) } /** diff --git a/@vates/nbd-client/multi.test.mjs b/@vates/nbd-client/multi.test.mjs new file mode 100644 index 0000000000..dec4ebe2bd --- /dev/null +++ b/@vates/nbd-client/multi.test.mjs @@ -0,0 +1,130 @@ +import { describe, it, mock } from 'node:test' +import assert from 'node:assert' + +// MultiNbdClient instantiates NbdClient itself (`import NbdClient from './index.mjs'`), so the +// only way to control individual connections from a unit test is to replace that module with a +// fake before importing multi.mjs. +const created = [] +class FakeNbdClient { + constructor(nbdInfo) { + this.nbdInfo = nbdInfo + this.disconnected = false + // overridden per-test after connect() to script this instance's behavior + this.readBlock = async () => { + throw new Error('readBlock not configured for this fake client') + } + created.push(this) + } + + async connect() {} + + async disconnect() { + this.disconnected = true + } + + async getMap() { + return [] + } +} + +mock.module('./index.mjs', { + defaultExport: FakeNbdClient, +}) + +const { default: MultiNbdClient } = await import('./multi.mjs') + +// a single settings entry is enough: connect() reuses candidates across concurrent slots, so +// nbdConcurrency still yields that many distinct FakeNbdClient instances +async function connectWithFakes(nbdConcurrency) { + created.length = 0 + const client = new MultiNbdClient({ address: 'fake-host' }, { nbdConcurrency }) + await client.connect() + return { client, fakes: created.slice() } +} + +describe('MultiNbdClient.readBlock eviction', () => { + it('evicts a client whose readBlock rejects and retries on a surviving client', async () => { + const { client, fakes } = await connectWithFakes(2) + const [dead, alive] = fakes + dead.readBlock = async () => { + throw new Error('connection is dead') + } + alive.readBlock = async index => Buffer.from([index]) + + // index 0 initially routes to `dead` (0 % 2 === 0) + const data = await client.readBlock(0) + + assert.deepStrictEqual(data, Buffer.from([0])) + assert.strictEqual(dead.disconnected, true, 'the dead client should have been disconnected') + + // future reads must no longer be routed to the evicted client: with only `alive` left, + // every index should now succeed through it + alive.readBlock = async index => Buffer.from([index]) + const data2 = await client.readBlock(1) + assert.deepStrictEqual(data2, Buffer.from([1])) + }) + + it('rejects once every client has been evicted, instead of retrying forever', async () => { + const { client, fakes } = await connectWithFakes(2) + const boom = new Error('connection is dead') + for (const fake of fakes) { + fake.readBlock = async () => { + throw boom + } + } + + await assert.rejects(() => client.readBlock(0), boom) + assert.ok( + fakes.every(fake => fake.disconnected), + 'every client should have been evicted (and disconnected) before giving up' + ) + }) + + it('evicting the same dead client from concurrent failed reads is idempotent', async () => { + const { client, fakes } = await connectWithFakes(2) + const [dead, alive] = fakes + dead.readBlock = async () => { + throw new Error('connection is dead') + } + alive.readBlock = async index => Buffer.from([index]) + + let disconnectCalls = 0 + dead.disconnect = async () => { + disconnectCalls++ + dead.disconnected = true + } + + // routing is round-robin by call order (0, 1, 0) over a 2-client pool: the 1st and 3rd + // calls both land on `dead` before either failure has been handled, so they race to evict + // the same client. The exact interleaving isn't guaranteed, only that eviction is safe + // either way and every read still completes. + const [data0, data1, data2] = await Promise.all([client.readBlock(10), client.readBlock(11), client.readBlock(12)]) + + assert.deepStrictEqual(data0, Buffer.from([10])) + assert.deepStrictEqual(data1, Buffer.from([11])) + assert.deepStrictEqual(data2, Buffer.from([12])) + assert.strictEqual(disconnectCalls, 1, 'the dead client must only be evicted/disconnected once') + }) +}) + +describe('MultiNbdClient.readBlock routing', () => { + it('distributes reads round-robin by call order, not by index residue', async () => { + const { client, fakes } = await connectWithFakes(4) + const readsPerClient = fakes.map(() => 0) + fakes.forEach((fake, i) => { + fake.readBlock = async index => { + readsPerClient[i]++ + return Buffer.from([index]) + } + }) + + // a CBT-style changed-block set striding by exactly `nbdConcurrency`: under the old + // `index % clients.length` routing this would collapse entirely onto a single client + const stridedIndexes = Array.from({ length: 20 }, (_, i) => i * 4) + for (const index of stridedIndexes) { + await client.readBlock(index) + } + + assert.deepStrictEqual(readsPerClient, [5, 5, 5, 5]) + }) +}) diff --git a/@vates/nbd-client/package.json b/@vates/nbd-client/package.json index 5c79e93d3c..4dc344bc3a 100644 --- a/@vates/nbd-client/package.json +++ b/@vates/nbd-client/package.json @@ -33,6 +33,7 @@ }, "scripts": { "postversion": "npm publish --access public", - "test-integration": "tap --allow-incomplete-coverage" + "test": "node --experimental-test-module-mocks --test", + "test-integration": "tap --allow-incomplete-coverage tests/*.integ.mjs" } } diff --git a/@vates/nbd-client/tests/nbdclient.integ.mjs b/@vates/nbd-client/tests/nbdclient.integ.mjs index c8db3773d5..23f3648fef 100644 --- a/@vates/nbd-client/tests/nbdclient.integ.mjs +++ b/@vates/nbd-client/tests/nbdclient.integ.mjs @@ -153,7 +153,13 @@ CYu1Xn/FVPx1HoRgWc7E8wFhDcA/P3SJtfIQWHB9FzSaBflKGR4t8WCE2eE8+cTB nbdServer = await spawnNbdKit(path) } } - assert.rejects(() => client.readBlock(100, CHUNK_SIZE)) + // reading past the end of the export must reject: a single client so the failure isn't + // multiplied by nbdConcurrency evicting and retrying on every other (equally healthy, equally + // unable to satisfy this out-of-range request) client + const singleClient = new MultiNbdClient(connectionSettings, { nbdConcurrency: 1, readAhead: 2 }) + await singleClient.connect() + await assert.rejects(() => singleClient.readBlock(100, CHUNK_SIZE)) + await singleClient.disconnect() await client.disconnect() // double disconnection shouldn't pose any problem diff --git a/@xen-orchestra/xapi/disks/Xapi.mjs b/@xen-orchestra/xapi/disks/Xapi.mjs index d744c4cb6f..e57ab28ce2 100644 --- a/@xen-orchestra/xapi/disks/Xapi.mjs +++ b/@xen-orchestra/xapi/disks/Xapi.mjs @@ -1,11 +1,16 @@ // @ts-check /** * @typedef {import('@xen-orchestra/disk-transform').DiskBlock} DiskBlock - * @typedef {import('@xen-orchestra/disk-transform').RandomAccessDisk} RandomAccessDisk * @typedef {import('@xen-orchestra/disk-transform').Disk} Disk */ -import { DiskLargerBlock, DiskPassthrough, ReadAhead, TimeoutDisk } from '@xen-orchestra/disk-transform' +import { + DiskLargerBlock, + DiskPassthrough, + RandomAccessDisk, + ReadAhead, + TimeoutDisk, +} from '@xen-orchestra/disk-transform' import { createLogger } from '@xen-orchestra/log' import { Task } from '@vates/task' import { XapiVhdCbtSource } from './XapiVhdCbt.mjs' @@ -82,55 +87,45 @@ export class XapiDiskSource extends DiskPassthrough { * Create a disk source using stream export + NBD. * On failure, fall back to a full export. * - * @returns {Promise} + * @returns {Promise} */ async #openNbdStream() { const xapi = this.#xapi - const baseRef = this.#baseRef const vdiRef = this.#vdiRef /** - * @type {Disk} + * @type {XapiStreamNbdSource|undefined} */ let source let streamSource try { - streamSource = await this.#openExportStream() + streamSource = await this.#openExportStream({ onlyListChangedBlocks: true }) if (streamSource === undefined) { throw new Error(`Can't open stream source`) } source = new XapiStreamNbdSource(streamSource, { vdiRef, - baseRef, xapi, nbdConcurrency: this.#nbdConcurrency, onlyListChangedBlocks: this.#onlyListChangedBlocks, }) await source.init() - if (source.getBlockSize() < this.#blockSize) { - source = new DiskLargerBlock(source, this.#blockSize) - } + this.#useNbd = true + + return await this.#formatSourceDisk(source, 'NBT') } catch (err) { - if (err.code === 'NO_NBD_AVAILABLE') { + // init probaby failed, so nothing to close , but better safe than sorry + await source?.close().catch(warn) + + if (/** @type {NodeJS.ErrnoException} */ (err).code === 'NO_NBD_AVAILABLE') { const warningMessage = `can't connect through NBD, fall back to stream export` // @ts-ignore Task.warning is a static alias set up dynamically, not visible to TS Task.warning(warningMessage) warn(warningMessage, err) - if (streamSource === undefined) { - throw new Error(`Can't open stream source`) - } - return streamSource + // reopen the stream with the block data + return this.#openExportStream() } - // init probaby failed, so nothing to close , but better safe than sorry - await source?.close().catch(warn) throw err } - this.#useNbd = true - const readAhead = new ReadAhead(source) - source = new TimeoutDisk(source, this.#timeout) - const label = await xapi.getField('VDI', vdiRef, 'name_label') - // manually create an export task for NBD since xapi xan't do it automatically - readAhead.addProgressHandler(new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through NBD`)) - return readAhead } async #getPreferedExportFormat() { @@ -168,7 +163,7 @@ export class XapiDiskSource extends DiskPassthrough { * * @returns {Promise} */ - async #openExportStream() { + async #openExportStream({ onlyListChangedBlocks = false } = {}) { const xapi = this.#xapi const baseRef = this.#baseRef const vdiRef = this.#vdiRef @@ -185,6 +180,17 @@ export class XapiDiskSource extends DiskPassthrough { } await source.init() if (source.getBlockSize() < this.#blockSize) { + if (!onlyListChangedBlocks && baseRef !== undefined) { + // enlarging blocks needs to fill gaps from the parent chain (see DiskLargerBlock's + // isDifferencing branch), but XapiQcow2StreamSource doesn't implement instantiateParent() + // yet: safe when onlyListChangedBlocks (readBlock() is never called, only used for + // hasBlock/getBlockIndexes metadata by XapiStreamNbdSource), unsafe otherwise. + throw new Error(`Can't change the block size of a differencing disk through xapi export`) + } + if (!(source instanceof RandomAccessDisk)) { + // XapiQcow2StreamSource is a RandomAccessDisk but only when reading forward + throw new Error(`can't adapt ${source.constructor.name}'s block size: not random access`) + } source = new DiskLargerBlock(source, this.#blockSize) } source = new TimeoutDisk(source, this.#timeout) @@ -205,6 +211,34 @@ export class XapiDiskSource extends DiskPassthrough { return source } + /** + * + * @param {RandomAccessDisk} source + * @param {string} exportMethod + * @returns {Promise} + */ + async #formatSourceDisk(source, exportMethod) { + const xapi = this.#xapi + const vdiRef = this.#vdiRef + let formattedSource = source + if (formattedSource.getBlockSize() < this.#blockSize) { + formattedSource = new DiskLargerBlock(source, this.#blockSize) + } else if (formattedSource.getBlockSize() > this.#blockSize) { + throw new Error( + `Smaller block size is not implemented, source is ${formattedSource.getBlockSize()}, asked for ${this.#blockSize}` + ) + } + const readAhead = new ReadAhead(formattedSource) + const label = await xapi.getField('VDI', vdiRef, 'name_label') + // manually create an export task for NBD since xapi xan't do it automatically + readAhead.addProgressHandler( + new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through ${exportMethod}`) + ) + // wraps the ReadAhead's block generator, not the raw NBD source: see #openNbdStream for why + // TimeoutDisk must sit outside ReadAhead. + return new TimeoutDisk(readAhead, this.#timeout) + } + /** * Create a disk source using NBD and CBT. * On failure, fall back to stream + NBD. @@ -218,7 +252,7 @@ export class XapiDiskSource extends DiskPassthrough { /** * @type {RandomAccessDisk} */ - let source = new XapiVhdCbtSource({ + const source = new XapiVhdCbtSource({ vdiRef, baseRef, xapi, @@ -229,17 +263,9 @@ export class XapiDiskSource extends DiskPassthrough { await source.init() this.#useNbd = true this.#useCbt = true - const readAhead = new ReadAhead(source) - source = new TimeoutDisk(source, this.#timeout) - if (source.getBlockSize() < this.#blockSize) { - source = new DiskLargerBlock(source, this.#blockSize) - } - const label = await xapi.getField('VDI', vdiRef, 'name_label') - // manually create an export task for NBD since xapi xan't do it automatically - readAhead.addProgressHandler(new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through NBD+CBT`)) - return readAhead + return await this.#formatSourceDisk(source, 'NBT+CBT') } catch (error) { - if (error.code !== 'CBT_DISABLED') { + if (/** @type {NodeJS.ErrnoException} */ (error).code !== 'CBT_DISABLED') { info('Error in openNbdCBT', error) } // init probaby failed, so nothing to close , but better safe than sorry diff --git a/@xen-orchestra/xapi/disks/XapiStreamNbd.mjs b/@xen-orchestra/xapi/disks/XapiStreamNbd.mjs index 7c4b67a7f1..802fa00abe 100644 --- a/@xen-orchestra/xapi/disks/XapiStreamNbd.mjs +++ b/@xen-orchestra/xapi/disks/XapiStreamNbd.mjs @@ -6,13 +6,13 @@ * @typedef {import('@vates/nbd-client/multi.mjs').default} MultiNbdClient */ -import { DiskPassthrough } from '@xen-orchestra/disk-transform' +import { RandomDiskPassthrough } from '@xen-orchestra/disk-transform' import { createLogger } from '@xen-orchestra/log' import { connectNbdClientIfPossible } from './utils.mjs' const { warn } = createLogger('xo:xapi:XapiStreamNbd') -export class XapiStreamNbdSource extends DiskPassthrough { +export class XapiStreamNbdSource extends RandomDiskPassthrough { /** @type {MultiNbdClient|undefined} */ #nbdClient /** @type {number } */ @@ -42,6 +42,10 @@ export class XapiStreamNbdSource extends DiskPassthrough { if (streamSourceDisk === undefined) { throw new Error(`A stream source must be given`) } + // streamSourceDisk is only ever used here for its plain-Disk metadata (getBlockSize, + // getBlockIndexes, getVirtualSize, close) — readBlock is fully overridden below to go + // through NBD instead, so it never needs streamSourceDisk to be a RandomAccessDisk + // @ts-ignore super(streamSourceDisk) this.#nbdConcurrency = nbdConcurrency this.#vdiRef = vdiRef diff --git a/@xen-orchestra/xapi/disks/XapiVhdCbt.mjs b/@xen-orchestra/xapi/disks/XapiVhdCbt.mjs index 335ac20c44..0db4341682 100644 --- a/@xen-orchestra/xapi/disks/XapiVhdCbt.mjs +++ b/@xen-orchestra/xapi/disks/XapiVhdCbt.mjs @@ -68,6 +68,7 @@ export class XapiVhdCbtSource extends RandomAccessDisk { xapi.getField('VDI', ref, 'virtual_size'), ]) if (!cbt_enabled) { + /** @type {NodeJS.ErrnoException} */ const error = new Error(`CBT is disabled`) error.code = 'CBT_DISABLED' throw error diff --git a/@xen-orchestra/xapi/disks/utils.mjs b/@xen-orchestra/xapi/disks/utils.mjs index 300ae8c937..3eab9ac01d 100644 --- a/@xen-orchestra/xapi/disks/utils.mjs +++ b/@xen-orchestra/xapi/disks/utils.mjs @@ -20,6 +20,7 @@ export async function connectNbdClientIfPossible(xapi, vdiRef, nbdConcurrency) { } if (nbdInfos.length === 0) { + /** @type {NodeJS.ErrnoException} */ const error = new Error(`can't connect to any nbd client`) error.code = 'NO_NBD_AVAILABLE' throw error diff --git a/CHANGELOG.unreleased.md b/CHANGELOG.unreleased.md index 15d771476a..3507bd228d 100644 --- a/CHANGELOG.unreleased.md +++ b/CHANGELOG.unreleased.md @@ -20,6 +20,7 @@ > Users must be able to say: "I had this issue, happy to know it's fixed" - [sdn-controller] Fix `update_traffic_rule` keeping the previous port: `newRule` is a partial update and a field sent as `null` is now removed from the rule (PR [#10307](https://github.com/vatesfr/xen-orchestra/pull/10307)) +- [Backup/NBD] Better behaviour with a stalled NBD client, and respect global export timeout (PR [#10306](https://github.com/vatesfr/xen-orchestra/pull/10306)) ### Packages to release @@ -37,10 +38,12 @@ +- @vates/nbd-client patch - @vates/types major - @xen-orchestra/rest-api minor - @xen-orchestra/web minor - @xen-orchestra/web-core minor +- @xen-orchestra/xapi patch - xo-server-sdn-controller minor