vatesfr/xen-orchestra
1
0
Fork 0
mirror of https://github.com/vatesfr/xen-orchestra.git synced 2026-10-12 02:24:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(nbd-client): implement a nbd-client for stdin/stdout instead of a tcp connection (#10381)

Browse Source
This commit is contained in:
Florent BEAUCHAMP
2026-09-25 09:57:53 +02:00
committed by GitHub
parent d1029ffb24
commit b5f119f1d4
12 changed files with 1456 additions and 414 deletions
Download Patch File Download Diff File Expand all files Collapse all files
@vates/nbd-client/.USAGE.md
+60 -2
View File
@@ -1,6 +1,12 @@
### `new NdbClient({address, exportname, secure = true, port = 10809})`
This package provides a NBD client, in two flavors sharing the same protocol
implementation (`AbstractNbdClient`): over the network, or over the standard
streams of a NBD server run as a child process.
create a new nbd client
### `new NbdClient({address, exportname, cert, port = 10809})`
Create a new nbd client connecting to a server over TCP.
`NbdClient` is also exported as `NbdTcpClient`.
```js
import NbdClient from '@vates/nbd-client'
@@ -14,3 +20,55 @@ await client.connect()
const block = await client.readBlock(blockIndex, BlockSize)
await client.disconnect()
```
### `new NbdStdioClient({command, args, exportname})`
Create a new nbd client talking to a server through its standard streams, like
`nbdkit --single` or any server reading its queries on stdin and writing its
answers on stdout.
The process is spawned by `connect()` and stopped by `disconnect()`.
```js
import { NbdStdioClient } from '@vates/nbd-client'
const client = new NbdStdioClient({
command: 'nbdkit',
args: ['--single', '--exit-with-parent', '--read-only', 'file', '/path/to/disk.raw'],
exportname: '', // optional, the default export of the server
})
await client.connect()
const block = await client.readBlock(blockIndex, BlockSize)
await client.disconnect()
```
There is no TLS here: the transport is a pair of pipes, there is nothing to
encrypt.
`readBlock()` retries on error, and a retry reconnects: with this client that
means killing the server and spawning a new one, so the command must be
restartable and must serve the same content on each run.
`getMap()` is not implemented, since `nbdinfo` needs either an URI or a server
supporting systemd socket activation: compute the map on the caller side and
pass it explicitly, for example `new NbdDisk(infos, blockSize, { dataMap })`.
The tail of the server stderr is available as `client.stderr` and attached to
the errors raised when the server dies unexpectedly.
### Writing another client
Subclass `AbstractNbdClient` and implement the transport:
- `_openTransport()`: **required**, returns `{ readable, writable }`, called on
every (re)connection.
- `_secureTransport(transport)`: optional, called during the handshake, the only
place where the transport can be upgraded (`NBD_OPT_STARTTLS`).
- `_closeTransport(transport, lastMessage)`: optional, must hand over
`lastMessage` (`NBD_CMD_DISC`) before closing.
- `_destroyTransport(transport)`: optional, forceful cleanup, must not throw.
- `getMap(signal)`: optional, the base implementation rejects with the code
`NBD_MAP_UNSUPPORTED`.
Any extra property set on the transport object is passed back to
`_closeTransport()` and `_destroyTransport()`.
@vates/nbd-client/AbstractNbdClient.mjs
+572
View File
@@ -0,0 +1,572 @@
import assert from 'node:assert'
import { pRetry, pDelay, pTimeout, pFromCallback } from 'promise-toolbox'
import { readChunkStrict } from '@vates/read-chunk'
import { createLogger } from '@xen-orchestra/log'
import {
INIT_PASSWD,
NBD_CMD_DISC,
NBD_CMD_READ,
NBD_DEFAULT_BLOCK_SIZE,
NBD_FLAG_FIXED_NEWSTYLE,
NBD_FLAG_HAS_FLAGS,
NBD_OPT_EXPORT_NAME,
NBD_OPT_REPLY_MAGIC,
NBD_REPLY_ACK,
NBD_REPLY_MAGIC,
NBD_REQUEST_MAGIC,
OPTS_MAGIC,
} from './constants.mjs'
const { debug, warn } = createLogger('vates:nbd-client')
// documentation is here : https://github.com/NetworkBlockDevice/nbd/blob/master/doc/proto.md
/**
* A pair of streams carrying the NBD protocol.
*
* For a TCP client both are the same socket, for a client talking to a server
* through its standard streams they are the two ends of two different pipes.
*
* @typedef {object} NbdTransport
* @property {import('node:stream').Readable} readable - where the server answers are read from
* @property {import('node:stream').Writable} writable - where the client queries are written to
*/
/**
* Transport agnostic NBD client: it implements the handshake and the
* transmission phases, and delegates the opening/closing of the underlying
* streams to its subclasses.
*
* Subclasses MUST implement `_openTransport()` and MAY override
* `_secureTransport()`, `_closeTransport()`, `_destroyTransport()` and `getMap()`.
*
* @abstract
*/
export default class AbstractNbdClient {
#exportName
#exportSize
/** @type {NbdTransport|undefined} */
#transport
// incremented on each connection attempt: an attempt which is not the last
// one anymore has been superseded (its caller timed out on it) and must not
// publish itself
#connectGeneration = 0
#waitBeforeReconnect
#readBlockRetries
#reconnectRetry
#connectTimeout
#messageTimeout
// AFAIK, there is no guaranty the server answers in the same order as the queries
// so we handle a backlog of command waiting for response and handle concurrency manually
#waitingForResponse // there is already a listener waiting for a response
#nextCommandQueryId = BigInt(0)
// map of command waiting for a response queryId => { size/*in byte*/, resolve, reject}
#commandQueryBacklog = new Map()
#connected = false
#reconnectingPromise
/**
* @param {object} settings
* @param {string} [settings.exportname] - the NBD export to open, empty for the default one
* @param {object} [options]
* @param {number} [options.connectTimeout]
* @param {number} [options.messageTimeout]
* @param {number} [options.waitBeforeReconnect]
* @param {number} [options.readBlockRetries]
* @param {number} [options.reconnectRetry]
*/
constructor(
{ exportname = '' },
{
connectTimeout = 6e4,
messageTimeout = 6e4,
waitBeforeReconnect = 1e3,
readBlockRetries = 5,
reconnectRetry = 5,
} = {}
) {
this.#exportName = exportname
this.#waitBeforeReconnect = waitBeforeReconnect
this.#readBlockRetries = readBlockRetries
this.#reconnectRetry = reconnectRetry
this.#connectTimeout = connectTimeout
this.#messageTimeout = messageTimeout
}
get exportSize() {
return this.#exportSize
}
get exportName() {
return this.#exportName
}
get connected() {
return this.#connected
}
/* ---------------------------------------------------------------------------
* transport, to be implemented by the subclasses
* ------------------------------------------------------------------------- */
/**
* Open the streams used to talk to the NBD server.
*
* Called on every (re)connection: it must be able to open a brand new
* transport each time it is called.
*
* @abstract
* @returns {Promise<NbdTransport>}
*/
async _openTransport() {
throw new Error(`${this.constructor.name} must implement _openTransport()`)
}
/**
* Hook called during the handshake, after the client flags have been sent and
* before the export is selected: this is the only moment where the transport
* can be upgraded (NBD_OPT_STARTTLS).
*
* Returning another transport than the one received replaces it, returning
* the same one is a no-op.
*
* @param {NbdTransport} transport
* @returns {Promise<NbdTransport>}
*/
async _secureTransport(transport) {
return transport
}
/**
* Gracefully close the transport.
*
* `lastMessage` is a ready to send NBD_CMD_DISC: implementations must hand it
* over to the server (it is the only clean way to tell it we're done) before
* closing their side.
*
* Failing or timing out here is not fatal: `_destroyTransport()` will be
* called as a fallback.
*
* @param {NbdTransport} transport
* @param {Buffer} lastMessage
* @returns {Promise<void>}
*/
async _closeTransport(transport, lastMessage) {
await pFromCallback(cb => transport.writable.end(lastMessage, cb))
}
/**
* Forcefully release the transport and everything backing it.
*
* Must be idempotent, and must not throw.
*
* @param {NbdTransport} transport
* @returns {void}
*/
_destroyTransport(transport) {
transport.readable.destroy()
if (transport.writable !== transport.readable) {
transport.writable.destroy()
}
}
/**
* Returns the map of the file with holes, zeros and data, useful to handle
* efficiently sparse sources.
*
* The base implementation is unable to compute it: subclasses able to do so
* must override this method.
*
* To implement this here: use structured replies if the server supports them,
* and then ask for BLOCK_STATUS.
*
* @returns {Promise<{ offset: number, length: number, type: number }[]>}
* A promise that resolves to an array where each object represents a segment:
* - `offset` — The byte offset from the start.
* - `length` — The size of the segment in bytes.
* - `type` — A numeric code indicating the segment type ( 0 means data).
*/
async getMap() {
const error = new Error(`${this.constructor.name} does not support getMap()`)
error.code = 'NBD_MAP_UNSUPPORTED'
throw error
}
/* ---------------------------------------------------------------------------
* connection
* ------------------------------------------------------------------------- */
#onTransportError = error => {
// without this listener an error on the transport would be thrown as an
// uncaught exception, and pending reads would only fail on message timeout
//
// an error outside of the connected window is expected (the other end can
// be gone already while we're closing), and always reported to the caller
// through the connect()/readBlock() rejection
const log = this.#connected ? warn : debug
log('error on the nbd transport', { error })
this.#rejectAll(error)
}
#watchTransport(transport) {
transport.readable.on('error', this.#onTransportError)
if (transport.writable !== transport.readable) {
transport.writable.on('error', this.#onTransportError)
}
}
async #connect() {
const generation = ++this.#connectGeneration
// the transport is kept local until the client is connected: connect() does
// not cancel on timeout, so a superseded attempt must not be able to use,
// replace or destroy the transport of the attempt which replaced it
let transport = await this._openTransport()
this.#watchTransport(transport)
try {
// the transport can be replaced during the handshake (TLS upgrade)
transport = await this.#handshake(transport)
if (generation !== this.#connectGeneration) {
const error = new Error('this connection has been superseded by a newer one')
error.code = 'NBD_CONNECT_SUPERSEDED'
throw error
}
} catch (error) {
// don't leak the transport (a socket, a child process, ...): disconnect()
// is a no-op as long as we're not connected, so nobody else will close it
this.#destroyTransport(transport)
throw error
}
this.#transport = transport
this.#connected = true
// reset internal state if we reconnected a nbd client
this.#commandQueryBacklog = new Map()
this.#waitingForResponse = false
}
async connect() {
const promise = this.#connect()
try {
return await pTimeout.call(promise, this.#connectTimeout)
} catch (error) {
// pTimeout does not cancel: #connect() may still be running and end up
// with an opened transport nobody owns anymore
promise.then(
() =>
this.disconnect().catch(cleanupError => warn('error while cleaning up a late connection', { cleanupError })),
() => {} // the failure is already reported through `error`
)
throw error
}
}
async disconnect() {
if (!this.#connected) {
return
}
this.#connected = false
const transport = this.#transport
this.#transport = undefined
if (transport === undefined) {
return
}
const queryId = this.#nextCommandQueryId
this.#nextCommandQueryId++
const buffer = Buffer.alloc(28)
buffer.writeInt32BE(NBD_REQUEST_MAGIC, 0) // it is a nbd request
buffer.writeInt16BE(0, 4) // no command flags for a disconnect
buffer.writeInt16BE(NBD_CMD_DISC, 6) // we want to disconnect from nbd server
buffer.writeBigUInt64BE(queryId, 8)
buffer.writeBigUInt64BE(0n, 16)
buffer.writeInt32BE(0, 24)
try {
await pTimeout.call(this._closeTransport(transport, buffer), this.#messageTimeout)
} catch (error) {
// expected when the other end is already gone, nothing actionable: the
// transport is destroyed instead of being closed gracefully
debug('error while closing the nbd transport, destroying it', { error })
this.#destroyTransport(transport)
}
}
#destroyTransport(transport) {
try {
this._destroyTransport(transport)
} catch (error) {
warn('error while destroying the nbd transport', { error })
}
}
#clearReconnectPromise = () => {
this.#reconnectingPromise = undefined
}
async #reconnect() {
await this.disconnect().catch(() => {})
await pDelay(this.#waitBeforeReconnect) // need to let the xapi clean things on its side
await this.connect()
}
async reconnect() {
// we need to ensure reconnections do not occur in parallel
if (this.#reconnectingPromise === undefined) {
this.#reconnectingPromise = pRetry(() => this.#reconnect(), {
tries: this.#reconnectRetry,
})
this.#reconnectingPromise.then(this.#clearReconnectPromise, this.#clearReconnectPromise)
}
return this.#reconnectingPromise
}
/* ---------------------------------------------------------------------------
* handshake
* ------------------------------------------------------------------------- */
/**
* Send an option to the server and check it acknowledged it.
*
* We can use individual read/write from the transport here since there is no
* concurrency during the handshake.
*
* The transport is passed explicitly since the client is not connected yet:
* it is only published once the handshake succeeded.
*
* protected: subclasses need it to negotiate their own options (NBD_OPT_STARTTLS)
*
* @param {NbdTransport} transport
* @param {number} option
* @param {Buffer} [buffer] - the payload of the option
*/
async _sendOption(transport, option, buffer = Buffer.alloc(0)) {
await this.#write(transport, OPTS_MAGIC)
await this.#writeInt32(transport, option)
await this.#writeInt32(transport, buffer.length)
await this.#write(transport, buffer)
assert.strictEqual(await this.#readInt64(transport), NBD_OPT_REPLY_MAGIC) // magic number everywhere
assert.strictEqual(await this.#readInt32(transport), option) // the option passed
assert.strictEqual(await this.#readInt32(transport), NBD_REPLY_ACK) // ACK
const length = await this.#readInt32(transport)
assert.strictEqual(length, 0) // length
}
// we can use individual read/write from the transport here since there is only one handshake at once, no concurrency
//
// it works on the transport it is given and returns the one to use for the
// transmission phase, which is not necessarily the same (TLS upgrade)
//
/**
* @param {NbdTransport} transport
* @returns {Promise<NbdTransport>}
*/
async #handshake(transport) {
assert((await this.#read(transport, 8)).equals(INIT_PASSWD))
assert((await this.#read(transport, 8)).equals(OPTS_MAGIC))
const flagsBuffer = await this.#read(transport, 2)
const flags = flagsBuffer.readInt16BE(0)
assert.strictEqual(flags & NBD_FLAG_FIXED_NEWSTYLE, NBD_FLAG_FIXED_NEWSTYLE) // only FIXED_NEWSTYLE one is supported from the server options
await this.#writeInt32(transport, NBD_FLAG_FIXED_NEWSTYLE) // client also support NBD_FLAG_C_FIXED_NEWSTYLE
// let the subclass upgrade the transport if it needs to (TLS)
const secured = await this._secureTransport(transport)
if (secured !== transport) {
this.#watchTransport(secured)
transport = secured
}
// send export name we want to access.
// it's implicitly closing the negotiation phase.
await this.#write(transport, OPTS_MAGIC)
await this.#writeInt32(transport, NBD_OPT_EXPORT_NAME)
const exportNameBuffer = Buffer.from(this.#exportName)
await this.#writeInt32(transport, exportNameBuffer.length)
await this.#write(transport, exportNameBuffer)
// 8 (export size ) + 2 (flags) + 124 zero = 134
// must read all to ensure nothing stays in the buffer
const answer = await this.#read(transport, 134)
this.#exportSize = answer.readBigUInt64BE(0)
const transmissionFlags = answer.readInt16BE(8)
assert.strictEqual(transmissionFlags & NBD_FLAG_HAS_FLAGS, NBD_FLAG_HAS_FLAGS, 'NBD_FLAG_HAS_FLAGS') // must always be 1 by the norm
// note : xapi server always send NBD_FLAG_READ_ONLY (3) as a flag
return transport
}
/* ---------------------------------------------------------------------------
* raw read/write
* ------------------------------------------------------------------------- */
#getTransport() {
const transport = this.#transport
if (transport === undefined) {
const error = new Error('the nbd client is not connected')
error.code = 'NBD_NOT_CONNECTED'
throw error
}
return transport
}
#read(transport, length) {
const promise = readChunkStrict(transport.readable, length)
return pTimeout.call(promise, this.#messageTimeout)
}
#write(transport, buffer) {
let timeout
const messageTimeout = this.#messageTimeout
const { writable } = transport
return Promise.race([
new Promise((resolve, reject) => {
timeout = setTimeout(() => {
reject(new Error('timeout'))
}, messageTimeout)
}),
new Promise(resolve =>
writable.write(buffer, () => {
clearTimeout(timeout)
resolve()
})
),
])
}
async #readInt32(transport) {
const buffer = await this.#read(transport, 4)
return buffer.readInt32BE(0)
}
async #readInt64(transport) {
const buffer = await this.#read(transport, 8)
return buffer.readBigUInt64BE(0)
}
#writeInt32(transport, int) {
const buffer = Buffer.alloc(4)
buffer.writeInt32BE(int)
return this.#write(transport, buffer)
}
/* ---------------------------------------------------------------------------
* transmission
* ------------------------------------------------------------------------- */
// when one read fail ,stop everything
// (sync: it must be usable from an event handler without leaving a floating promise)
#rejectAll(error) {
this.#commandQueryBacklog.forEach(({ reject }) => {
reject(error)
})
}
async #readBlockResponse() {
// ensure at most one read occur in parallel
if (this.#waitingForResponse) {
return
}
try {
this.#waitingForResponse = true
const transport = this.#getTransport()
const buffer = await this.#read(transport, 16)
const magic = buffer.readInt32BE(0)
if (magic !== NBD_REPLY_MAGIC) {
throw new Error(`magic number for block answer is wrong : ${magic} ${NBD_REPLY_MAGIC}`)
}
const error = buffer.readInt32BE(4)
if (error !== 0) {
// @todo use error code from constants.mjs
throw new Error(`GOT ERROR CODE : ${error}`)
}
const blockQueryId = buffer.readBigUInt64BE(8)
const query = this.#commandQueryBacklog.get(blockQueryId)
if (!query) {
throw new Error(` no query associated with id ${blockQueryId}`)
}
this.#commandQueryBacklog.delete(blockQueryId)
const data = await this.#read(transport, query.size)
query.resolve(data)
this.#waitingForResponse = false
if (this.#commandQueryBacklog.size > 0) {
// it doesn't throw directly but will throw all relevant promise on failure
this.#readBlockResponse()
}
} catch (error) {
// reject all the promises
// we don't need to call readBlockResponse on failure
// since we will empty the backlog
this.#rejectAll(error)
}
}
async #readBlock(index, size) {
// we don't want to add anything in backlog while reconnecting
if (this.#reconnectingPromise) {
await this.#reconnectingPromise
}
const queryId = this.#nextCommandQueryId
this.#nextCommandQueryId++
// create and send command at once to ensure there is no concurrency issue
const buffer = Buffer.alloc(28)
buffer.writeInt32BE(NBD_REQUEST_MAGIC, 0) // it is a nbd request
buffer.writeInt16BE(0, 4) // no command flags for a simple block read
buffer.writeInt16BE(NBD_CMD_READ, 6) // we want to read a data block
buffer.writeBigUInt64BE(queryId, 8)
// byte offset in the raw disk
const offset = BigInt(index) * BigInt(size)
const remaining = this.#exportSize - offset
if (remaining < BigInt(size)) {
size = Number(remaining)
}
buffer.writeBigUInt64BE(offset, 16)
buffer.writeInt32BE(size, 24)
return new Promise((resolve, reject) => {
function decoratedReject(error) {
error.index = index
error.size = size
reject(error)
}
// this will handle one block response, but it can be another block
// since server does not guaranty to handle query in order
this.#commandQueryBacklog.set(queryId, {
size,
resolve,
reject: decoratedReject,
})
// really send the command to the server
this.#write(this.#getTransport(), buffer).catch(decoratedReject)
// #readBlockResponse never throws directly
// but if it fails it will reject all the promises in the backlog
this.#readBlockResponse()
})
}
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()
},
})
}
}
@vates/nbd-client/AbstractNbdClient.test.mjs
+177
View File
@@ -0,0 +1,177 @@
import assert from 'node:assert/strict'
import { PassThrough } from 'node:stream'
import { setTimeout as pSleep } from 'node:timers/promises'
import { describe, it } from 'node:test'
import AbstractNbdClient from './AbstractNbdClient.mjs'
import { serveNbd } from './tests/fake-nbd-server.mjs'
const BLOCK_SIZE = 512
// non aligned size, to check the truncated last block
const DATA = Buffer.alloc(BLOCK_SIZE * 3 + 100)
for (let i = 0; i < DATA.length; i += 4) {
DATA.writeUInt32BE(i, i)
}
// exercises AbstractNbdClient without any I/O: the fake server runs in the same
// process, on the other end of two in-memory streams
class InMemoryNbdClient extends AbstractNbdClient {
#serverOptions
/** @type {Array<object>} */
transports = []
/** @type {Array<object>} */
destroyedTransports = []
secureTransportCalls = 0
constructor({ exportname, ...serverOptions } = {}, options) {
super({ exportname }, options)
this.#serverOptions = serverOptions
}
async _openTransport() {
const toServer = new PassThrough()
const toClient = new PassThrough()
const transport = { readable: toClient, writable: toServer }
this.transports.push(transport)
// the server runs alongside the client, a failure on its side is reported
// to the client the same way a network error would be
serveNbd({ readable: toServer, writable: toClient, data: DATA, ...this.#serverOptions }).catch(error =>
toClient.destroy(error)
)
return transport
}
async _secureTransport(transport) {
this.secureTransportCalls++
return super._secureTransport(transport)
}
_destroyTransport(transport) {
this.destroyedTransports.push(transport)
super._destroyTransport(transport)
}
}
// the first connection is opened on a transport nobody ever answers on: its
// handshake stalls until the message timeout, well after connect() gave up
class StallingFirstConnectClient extends InMemoryNbdClient {
#openCalls = 0
async _openTransport() {
if (this.#openCalls++ === 0) {
const transport = { readable: new PassThrough(), writable: new PassThrough() }
this.transports.push(transport)
return transport
}
return super._openTransport()
}
}
describe('AbstractNbdClient', () => {
it('handshakes and reads blocks', async () => {
const client = new InMemoryNbdClient()
await client.connect()
try {
assert.equal(client.exportSize, BigInt(DATA.length))
assert.equal(client.connected, true)
const block = await client.readBlock(1, BLOCK_SIZE)
assert.ok(block.equals(DATA.subarray(BLOCK_SIZE, 2 * BLOCK_SIZE)))
} finally {
await client.disconnect()
}
assert.equal(client.connected, false)
})
it('truncates the last block of a non aligned export', async () => {
const client = new InMemoryNbdClient()
await client.connect()
try {
const block = await client.readBlock(3, BLOCK_SIZE)
assert.equal(block.length, DATA.length - 3 * BLOCK_SIZE)
assert.ok(block.equals(DATA.subarray(3 * BLOCK_SIZE)))
} finally {
await client.disconnect()
}
})
it('handles answers coming out of order', async () => {
const client = new InMemoryNbdClient({ answerInReverse: true })
await client.connect()
try {
const [first, second] = await Promise.all([client.readBlock(0, BLOCK_SIZE), client.readBlock(1, BLOCK_SIZE)])
assert.ok(first.equals(DATA.subarray(0, BLOCK_SIZE)))
assert.ok(second.equals(DATA.subarray(BLOCK_SIZE, 2 * BLOCK_SIZE)))
} finally {
await client.disconnect()
}
})
it('calls _secureTransport during the handshake', async () => {
const client = new InMemoryNbdClient()
await client.connect()
assert.equal(client.secureTransportCalls, 1)
await client.disconnect()
})
it('does not leak the transport when the handshake fails', async () => {
// the server expects another export name, and fails the handshake
const client = new InMemoryNbdClient({ exportname: 'wrong', exportName: 'expected' })
await assert.rejects(client.connect())
assert.equal(client.connected, false)
assert.equal(client.transports.length, 1)
assert.deepEqual(client.destroyedTransports, client.transports)
assert.equal(client.transports[0].writable.destroyed, true)
})
it('reports the errors answered by the server, after having retried', async () => {
const client = new InMemoryNbdClient(
{ errorCode: () => 5 /* EIO */ },
{ readBlockRetries: 2, reconnectRetry: 1, waitBeforeReconnect: 0 }
)
await client.connect()
try {
await assert.rejects(client.readBlock(0, BLOCK_SIZE), /ERROR CODE/)
// one transport for the initial connection, one for the retry
assert.equal(client.transports.length, 2)
} finally {
await client.disconnect()
}
})
it('does not let a connection which timed out disturb the next one', async () => {
const client = new StallingFirstConnectClient({}, { connectTimeout: 50, messageTimeout: 150 })
await assert.rejects(client.connect())
// the previous attempt is still running: it must not publish itself, nor
// touch the transport of this one
await client.connect()
try {
// let the stalled handshake reach its message timeout
await pSleep(250)
assert.equal(client.connected, true)
assert.equal(client.transports.length, 2)
// only the stalled transport has been destroyed
assert.deepEqual(client.destroyedTransports, [client.transports[0]])
const block = await client.readBlock(0, BLOCK_SIZE)
assert.ok(block.equals(DATA.subarray(0, BLOCK_SIZE)))
} finally {
await client.disconnect()
}
})
it('does not support getMap()', async () => {
const client = new InMemoryNbdClient()
await assert.rejects(client.getMap(), ({ code }) => code === 'NBD_MAP_UNSUPPORTED')
})
it('can be disconnected without being connected', async () => {
await new InMemoryNbdClient().disconnect()
})
})
@vates/nbd-client/NbdDisk.mjs
+20 -3
View File
@@ -5,12 +5,18 @@ import NbdClient from '@vates/nbd-client'
* @typedef {{ offset: number, length: number }} DataRange
*/
/**
* any NBD client, whatever transport it uses
*
* @typedef {import('@vates/nbd-client').AbstractNbdClient} AbstractNbdClient
*/
/**
* @extends {RandomAccessDisk}
*/
export class NbdDisk extends RandomAccessDisk {
#nbdInfos
/** @type {NbdClient|undefined} */
/** @type {AbstractNbdClient|undefined} */
#nbdClient
/** @type {Array<DataRange> | undefined} */
@@ -25,10 +31,21 @@ export class NbdDisk extends RandomAccessDisk {
/** @type {number | undefined} */
#hasBlockPreviousIndex
constructor(nbdInfos, blockSize, { dataMap } = {}) {
/** @type {typeof AbstractNbdClient} */
#ClientClass
/**
* @param {object} nbdInfos - the settings of the client, depends on `ClientClass`
* @param {number} blockSize
* @param {object} [options]
* @param {Array<{offset: number, length: number, type: number}>} [options.dataMap] - computed through `getMap()` when not provided
* @param {typeof AbstractNbdClient} [options.ClientClass] - to talk to a NBD server through another transport
*/
constructor(nbdInfos, blockSize, { dataMap, ClientClass = NbdClient } = {}) {
super()
this.#blockSize = blockSize
this.#nbdInfos = nbdInfos
this.#ClientClass = ClientClass
this.#dataMap = dataMap && this.#processDatamap(dataMap)
}
@@ -111,7 +128,7 @@ export class NbdDisk extends RandomAccessDisk {
* @returns {Promise<void>}
*/
async init() {
this.#nbdClient = new NbdClient(this.#nbdInfos)
this.#nbdClient = new this.#ClientClass(this.#nbdInfos)
await this.#nbdClient.connect()
if (this.#dataMap === undefined) {
this.#dataMap = this.#processDatamap(await this.#nbdClient.getMap())
@vates/nbd-client/NbdStdioClient.mjs
+157
View File
@@ -0,0 +1,157 @@
import assert from 'node:assert'
import { spawn } from 'node:child_process'
import { pFromCallback } from 'promise-toolbox'
import { createLogger } from '@xen-orchestra/log'
import AbstractNbdClient from './AbstractNbdClient.mjs'
const { warn } = createLogger('vates:nbd-client:stdio')
// only the tail is kept: it's the part describing why the server died
const STDERR_MAX_LENGTH = 4096
/**
* NBD client talking to a server through its standard streams, like
* `nbdkit -s` or `qemu-nbd` in `--fd` mode: the queries are written to the
* server stdin and its answers are read from its stdout.
*
* The process is spawned on `connect()` and killed on `disconnect()`, and
* `reconnect()` (used by `readBlock()` retries) spawns a brand new one: the
* server must be restartable, and must serve the same content on each run.
*
* There is no TLS here: the transport is a pair of pipes, nothing to encrypt.
*
* @extends {AbstractNbdClient}
*/
export default class NbdStdioClient extends AbstractNbdClient {
#args
#command
#cwd
#env
/** @type {{ child: import('node:child_process').ChildProcess, closing: boolean, stderr: string }|undefined} */
#state
/**
* @param {object} settings
* @param {string} settings.command - the NBD server to run
* @param {ReadonlyArray<string>} [settings.args] - its arguments
* @param {string} [settings.exportname] - empty (the default export) for most single connection servers
* @param {NodeJS.ProcessEnv} [settings.env]
* @param {string} [settings.cwd]
* @param {object} [options] - see {@link AbstractNbdClient}
*/
constructor({ command, args = [], exportname, env, cwd }, options) {
super({ exportname }, options)
assert.strictEqual(typeof command, 'string', 'command must be a string')
this.#command = command
this.#args = args
this.#env = env
this.#cwd = cwd
}
/**
* the tail of what the server wrote on stderr since it was spawned, useful to
* report why it failed
*
* @returns {string}
*/
get stderr() {
return this.#state?.stderr ?? ''
}
/**
* the pid of the current server process, undefined if it's not running
*
* @returns {number|undefined}
*/
get pid() {
return this.#state?.child.pid
}
async _openTransport() {
const child = spawn(this.#command, this.#args, {
cwd: this.#cwd,
env: this.#env,
// the protocol goes through stdin/stdout, stderr is only used for diagnostics
stdio: ['pipe', 'pipe', 'pipe'],
})
const state = { child, closing: false, stderr: '' }
this.#state = state
child.stderr.setEncoding('utf8')
child.stderr.on('data', chunk => {
state.stderr = (state.stderr + chunk).slice(-STDERR_MAX_LENGTH)
})
// spawn failures (ENOENT, EACCES, ...) are reported asynchronously
await new Promise((resolve, reject) => {
const onError = error => {
error.command = this.#command
reject(error)
}
child.once('error', onError)
child.once('spawn', () => {
child.removeListener('error', onError)
// from now on, an error on the process itself (a failed kill for
// example) must not be thrown as an uncaught exception
child.on('error', error => warn('error on the nbd server process', { error, command: this.#command }))
resolve()
})
})
child.once('exit', (code, signal) => {
if (state.closing) {
// expected: disconnect() asked it to quit and is waiting for this
return
}
const error = new Error(`the nbd server process exited (code: ${code}, signal: ${signal})`)
error.code = 'NBD_SERVER_EXITED'
error.exitCode = code
error.signal = signal
error.command = this.#command
error.stderr = state.stderr
warn('the nbd server process exited unexpectedly', { error })
// make the pending and future reads fail with this error instead of
// waiting for the message timeout on a stream that will never answer
child.stdout.destroy(error)
child.stdin.destroy()
})
return { readable: child.stdout, writable: child.stdin, state }
}
async _closeTransport(transport, lastMessage) {
const { state } = transport
state.closing = true
// hand over NBD_CMD_DISC, then close stdin: that's how the server knows
// it can quit
await pFromCallback(cb => transport.writable.end(lastMessage, cb))
// don't let the server block on a write while we wait for its exit
transport.readable.resume()
const { child } = state
if (child.exitCode === null && child.signalCode === null) {
await new Promise(resolve => child.once('exit', () => resolve()))
}
}
_destroyTransport(transport) {
const { state } = transport
const { child } = state
// we're the one tearing it down: the exit that follows is expected, not a
// crash, and destroying the streams can be enough to make the server quit
state.closing = true
transport.readable.destroy()
transport.writable.destroy()
if (child.exitCode === null && child.signalCode === null) {
child.kill('SIGKILL')
}
}
// getMap() is not implemented: `nbdinfo` needs an URI or a server supporting
// systemd socket activation, and this one only speaks through its standard
// streams. Compute the map on the caller side and pass it explicitly (
// `new NbdDisk(infos, blockSize, { dataMap })`).
}
@vates/nbd-client/NbdTcpClient.mjs
+106
View File
@@ -0,0 +1,106 @@
import { Socket } from 'node:net'
import { connect } from 'node:tls'
import { spawn } from 'node:child_process'
import AbstractNbdClient from './AbstractNbdClient.mjs'
import { NBD_DEFAULT_PORT, NBD_OPT_STARTTLS } from './constants.mjs'
/**
* NBD client talking to a server over TCP, optionally upgrading the connection
* to TLS during the handshake.
*
* @extends {AbstractNbdClient}
*/
export default class NbdTcpClient extends AbstractNbdClient {
#serverAddress
#serverCert
#serverPort
/**
* @param {object} settings
* @param {string} settings.address
* @param {number} [settings.port]
* @param {string} [settings.exportname]
* @param {string} [settings.cert] - PEM certificate, enables TLS when set
* @param {object} [options] - see {@link AbstractNbdClient}
*/
constructor({ address, port = NBD_DEFAULT_PORT, exportname, cert }, options) {
super({ exportname }, options)
this.#serverAddress = address
this.#serverPort = port
this.#serverCert = cert
}
// mandatory , at least to start the handshake: the connection always starts
// unsecured, and is upgraded to TLS during the handshake
async _openTransport() {
const socket = new Socket()
await new Promise((resolve, reject) => {
socket.connect(this.#serverPort, this.#serverAddress)
socket.once('error', reject)
socket.once('connect', () => {
socket.removeListener('error', reject)
resolve()
})
})
return { readable: socket, writable: socket }
}
async _secureTransport(transport) {
if (this.#serverCert === undefined) {
return transport
}
await this._sendOption(transport, NBD_OPT_STARTTLS)
const secured = await new Promise((resolve, reject) => {
const socket = connect({
socket: transport.writable,
rejectUnauthorized: false,
cert: this.#serverCert,
})
socket.once('error', reject)
socket.once('secureConnect', () => {
socket.removeListener('error', reject)
resolve(socket)
})
})
return { readable: secured, writable: secured }
}
/**
* @param {AbortSignal} [signal]
* @returns {Promise<{ offset: number, length: number, type: number }[]>}
*/
/* async */ getMap(signal) {
return new Promise((resolve, reject) => {
const process = spawn('nbdinfo', [
'--json',
'--map',
`nbd://${this.#serverAddress}:${this.#serverPort}/${encodeURIComponent(this.exportName)}`,
])
let text = ''
let errText = ''
process.stdout.on('data', data => (text += data))
process.stderr.on('data', data => (errText += data))
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}`))
}
try {
const json = JSON.parse(text)
resolve(json)
} catch (error) {
reject(new Error(`${error} ${text}`))
}
})
})
}
}
@vates/nbd-client/README.md
+60 -2
View File
@@ -14,9 +14,15 @@ npm install --save @vates/nbd-client
## Usage
### `new NdbClient({address, exportname, secure = true, port = 10809})`
This package provides a NBD client, in two flavors sharing the same protocol
implementation (`AbstractNbdClient`): over the network, or over the standard
streams of a NBD server run as a child process.
create a new nbd client
### `new NbdClient({address, exportname, cert, port = 10809})`
Create a new nbd client connecting to a server over TCP.
`NbdClient` is also exported as `NbdTcpClient`.
```js
import NbdClient from '@vates/nbd-client'
@@ -31,6 +37,58 @@ const block = await client.readBlock(blockIndex, BlockSize)
await client.disconnect()
```
### `new NbdStdioClient({command, args, exportname})`
Create a new nbd client talking to a server through its standard streams, like
`nbdkit --single` or any server reading its queries on stdin and writing its
answers on stdout.
The process is spawned by `connect()` and stopped by `disconnect()`.
```js
import { NbdStdioClient } from '@vates/nbd-client'
const client = new NbdStdioClient({
command: 'nbdkit',
args: ['--single', '--exit-with-parent', '--read-only', 'file', '/path/to/disk.raw'],
exportname: '', // optional, the default export of the server
})
await client.connect()
const block = await client.readBlock(blockIndex, BlockSize)
await client.disconnect()
```
There is no TLS here: the transport is a pair of pipes, there is nothing to
encrypt.
`readBlock()` retries on error, and a retry reconnects: with this client that
means killing the server and spawning a new one, so the command must be
restartable and must serve the same content on each run.
`getMap()` is not implemented, since `nbdinfo` needs either an URI or a server
supporting systemd socket activation: compute the map on the caller side and
pass it explicitly, for example `new NbdDisk(infos, blockSize, { dataMap })`.
The tail of the server stderr is available as `client.stderr` and attached to
the errors raised when the server dies unexpectedly.
### Writing another client
Subclass `AbstractNbdClient` and implement the transport:
- `_openTransport()`: **required**, returns `{ readable, writable }`, called on
every (re)connection.
- `_secureTransport(transport)`: optional, called during the handshake, the only
place where the transport can be upgraded (`NBD_OPT_STARTTLS`).
- `_closeTransport(transport, lastMessage)`: optional, must hand over
`lastMessage` (`NBD_CMD_DISC`) before closing.
- `_destroyTransport(transport)`: optional, forceful cleanup, must not throw.
- `getMap(signal)`: optional, the base implementation rejects with the code
`NBD_MAP_UNSUPPORTED`.
Any extra property set on the transport object is passed back to
`_closeTransport()` and `_destroyTransport()`.
## Contributions
Contributions are _very_ welcomed, either on the documentation or on
@vates/nbd-client/index.mjs
+5 -407
View File
@@ -1,408 +1,6 @@
import assert from 'node:assert'
import { Socket } from 'node:net'
import { connect } from 'node:tls'
import { pRetry, pDelay, pTimeout, pFromCallback } from 'promise-toolbox'
import { readChunkStrict } from '@vates/read-chunk'
import { createLogger } from '@xen-orchestra/log'
import {
INIT_PASSWD,
NBD_CMD_READ,
NBD_DEFAULT_BLOCK_SIZE,
NBD_DEFAULT_PORT,
NBD_FLAG_FIXED_NEWSTYLE,
NBD_FLAG_HAS_FLAGS,
NBD_OPT_EXPORT_NAME,
NBD_OPT_REPLY_MAGIC,
NBD_OPT_STARTTLS,
NBD_REPLY_ACK,
NBD_REPLY_MAGIC,
NBD_REQUEST_MAGIC,
OPTS_MAGIC,
NBD_CMD_DISC,
} from './constants.mjs'
import { spawn } from 'node:child_process'
// the TCP client is the historical default export of this package
export { default } from './NbdTcpClient.mjs'
const { warn } = createLogger('vates:nbd-client')
// documentation is here : https://github.com/NetworkBlockDevice/nbd/blob/master/doc/proto.md
export default class NbdClient {
#serverAddress
#serverCert
#serverPort
#serverSocket
#exportName
#exportSize
#waitBeforeReconnect
#readBlockRetries
#reconnectRetry
#connectTimeout
#messageTimeout
// AFAIK, there is no guaranty the server answers in the same order as the queries
// so we handle a backlog of command waiting for response and handle concurrency manually
#waitingForResponse // there is already a listener waiting for a response
#nextCommandQueryId = BigInt(0)
#commandQueryBacklog // map of command waiting for a response queryId => { size/*in byte*/, resolve, reject}
#connected = false
#reconnectingPromise
constructor(
{ address, port = NBD_DEFAULT_PORT, exportname, cert },
{
connectTimeout = 6e4,
messageTimeout = 6e4,
waitBeforeReconnect = 1e3,
readBlockRetries = 5,
reconnectRetry = 5,
} = {}
) {
this.#serverAddress = address
this.#serverPort = port
this.#exportName = exportname
this.#serverCert = cert
this.#waitBeforeReconnect = waitBeforeReconnect
this.#readBlockRetries = readBlockRetries
this.#reconnectRetry = reconnectRetry
this.#connectTimeout = connectTimeout
this.#messageTimeout = messageTimeout
}
get exportSize() {
return this.#exportSize
}
async #tlsConnect() {
return new Promise((resolve, reject) => {
this.#serverSocket = connect({
socket: this.#serverSocket,
rejectUnauthorized: false,
cert: this.#serverCert,
})
this.#serverSocket.once('error', reject)
this.#serverSocket.once('secureConnect', () => {
this.#serverSocket.removeListener('error', reject)
resolve()
})
})
}
// mandatory , at least to start the handshake
async #unsecureConnect() {
this.#serverSocket = new Socket()
return new Promise((resolve, reject) => {
this.#serverSocket.connect(this.#serverPort, this.#serverAddress)
this.#serverSocket.once('error', reject)
this.#serverSocket.once('connect', () => {
this.#serverSocket.removeListener('error', reject)
resolve()
})
})
}
async #connect() {
// first we connect to the server without tls, and then we upgrade the connection
// to tls during the handshake
await this.#unsecureConnect()
await this.#handshake()
this.#connected = true
// reset internal state if we reconnected a nbd client
this.#commandQueryBacklog = new Map()
this.#waitingForResponse = false
}
async connect() {
return pTimeout.call(this.#connect(), this.#connectTimeout)
}
async disconnect() {
if (!this.#connected) {
return
}
this.#connected = false
const socket = this.#serverSocket
const queryId = this.#nextCommandQueryId
this.#nextCommandQueryId++
const buffer = Buffer.alloc(28)
buffer.writeInt32BE(NBD_REQUEST_MAGIC, 0) // it is a nbd request
buffer.writeInt16BE(0, 4) // no command flags for a disconnect
buffer.writeInt16BE(NBD_CMD_DISC, 6) // we want to disconnect from nbd server
buffer.writeBigUInt64BE(queryId, 8)
buffer.writeBigUInt64BE(0n, 16)
buffer.writeInt32BE(0, 24)
const promise = pFromCallback(cb => {
socket.end(buffer, 'utf8', cb)
})
try {
await pTimeout.call(promise, this.#messageTimeout)
} catch (error) {
socket.destroy()
}
this.#serverSocket = undefined
this.#connected = false
}
#clearReconnectPromise = () => {
this.#reconnectingPromise = undefined
}
async #reconnect() {
await this.disconnect().catch(() => {})
await pDelay(this.#waitBeforeReconnect) // need to let the xapi clean things on its side
await this.connect()
}
async reconnect() {
// we need to ensure reconnections do not occur in parallel
if (this.#reconnectingPromise === undefined) {
this.#reconnectingPromise = pRetry(() => this.#reconnect(), {
tries: this.#reconnectRetry,
})
this.#reconnectingPromise.then(this.#clearReconnectPromise, this.#clearReconnectPromise)
}
return this.#reconnectingPromise
}
// we can use individual read/write from the socket here since there is no concurrency
async #sendOption(option, buffer = Buffer.alloc(0)) {
await this.#write(OPTS_MAGIC)
await this.#writeInt32(option)
await this.#writeInt32(buffer.length)
await this.#write(buffer)
assert.strictEqual(await this.#readInt64(), NBD_OPT_REPLY_MAGIC) // magic number everywhere
assert.strictEqual(await this.#readInt32(), option) // the option passed
assert.strictEqual(await this.#readInt32(), NBD_REPLY_ACK) // ACK
const length = await this.#readInt32()
assert.strictEqual(length, 0) // length
}
// we can use individual read/write from the socket here since there is only one handshake at once, no concurrency
async #handshake() {
assert((await this.#read(8)).equals(INIT_PASSWD))
assert((await this.#read(8)).equals(OPTS_MAGIC))
const flagsBuffer = await this.#read(2)
const flags = flagsBuffer.readInt16BE(0)
assert.strictEqual(flags & NBD_FLAG_FIXED_NEWSTYLE, NBD_FLAG_FIXED_NEWSTYLE) // only FIXED_NEWSTYLE one is supported from the server options
await this.#writeInt32(NBD_FLAG_FIXED_NEWSTYLE) // client also support NBD_FLAG_C_FIXED_NEWSTYLE
if (this.#serverCert !== undefined) {
// upgrade socket to TLS if needed
await this.#sendOption(NBD_OPT_STARTTLS)
await this.#tlsConnect()
}
// send export name we want to access.
// it's implicitly closing the negotiation phase.
await this.#write(OPTS_MAGIC)
await this.#writeInt32(NBD_OPT_EXPORT_NAME)
const exportNameBuffer = Buffer.from(this.#exportName)
await this.#writeInt32(exportNameBuffer.length)
await this.#write(exportNameBuffer)
// 8 (export size ) + 2 (flags) + 124 zero = 134
// must read all to ensure nothing stays in the buffer
const answer = await this.#read(134)
this.#exportSize = answer.readBigUInt64BE(0)
const transmissionFlags = answer.readInt16BE(8)
assert.strictEqual(transmissionFlags & NBD_FLAG_HAS_FLAGS, NBD_FLAG_HAS_FLAGS, 'NBD_FLAG_HAS_FLAGS') // must always be 1 by the norm
// note : xapi server always send NBD_FLAG_READ_ONLY (3) as a flag
}
#read(length) {
const promise = readChunkStrict(this.#serverSocket, length)
return pTimeout.call(promise, this.#messageTimeout)
}
#write(buffer) {
let timeout
const messageTimeout = this.#messageTimeout
const socket = this.#serverSocket
return Promise.race([
new Promise((resolve, reject) => {
timeout = setTimeout(() => {
reject(new Error('timeout'))
}, messageTimeout)
}),
new Promise(resolve =>
socket.write(buffer, () => {
clearTimeout(timeout)
resolve()
})
),
])
}
async #readInt32() {
const buffer = await this.#read(4)
return buffer.readInt32BE(0)
}
async #readInt64() {
const buffer = await this.#read(8)
return buffer.readBigUInt64BE(0)
}
#writeInt32(int) {
const buffer = Buffer.alloc(4)
buffer.writeInt32BE(int)
return this.#write(buffer)
}
// when one read fail ,stop everything
async #rejectAll(error) {
this.#commandQueryBacklog.forEach(({ reject }) => {
reject(error)
})
}
async #readBlockResponse() {
// ensure at most one read occur in parallel
if (this.#waitingForResponse) {
return
}
try {
this.#waitingForResponse = true
const buffer = await this.#read(16)
const magic = buffer.readInt32BE(0)
if (magic !== NBD_REPLY_MAGIC) {
throw new Error(`magic number for block answer is wrong : ${magic} ${NBD_REPLY_MAGIC}`)
}
const error = buffer.readInt32BE(4)
if (error !== 0) {
// @todo use error code from constants.mjs
throw new Error(`GOT ERROR CODE : ${error}`)
}
const blockQueryId = buffer.readBigUInt64BE(8)
const query = this.#commandQueryBacklog.get(blockQueryId)
if (!query) {
throw new Error(` no query associated with id ${blockQueryId}`)
}
this.#commandQueryBacklog.delete(blockQueryId)
const data = await this.#read(query.size)
query.resolve(data)
this.#waitingForResponse = false
if (this.#commandQueryBacklog.size > 0) {
// it doesn't throw directly but will throw all relevant promise on failure
this.#readBlockResponse()
}
} catch (error) {
// reject all the promises
// we don't need to call readBlockResponse on failure
// since we will empty the backlog
await this.#rejectAll(error)
}
}
async #readBlock(index, size) {
// we don't want to add anything in backlog while reconnecting
if (this.#reconnectingPromise) {
await this.#reconnectingPromise
}
const queryId = this.#nextCommandQueryId
this.#nextCommandQueryId++
// create and send command at once to ensure there is no concurrency issue
const buffer = Buffer.alloc(28)
buffer.writeInt32BE(NBD_REQUEST_MAGIC, 0) // it is a nbd request
buffer.writeInt16BE(0, 4) // no command flags for a simple block read
buffer.writeInt16BE(NBD_CMD_READ, 6) // we want to read a data block
buffer.writeBigUInt64BE(queryId, 8)
// byte offset in the raw disk
const offset = BigInt(index) * BigInt(size)
const remaining = this.#exportSize - offset
if (remaining < BigInt(size)) {
size = Number(remaining)
}
buffer.writeBigUInt64BE(offset, 16)
buffer.writeInt32BE(size, 24)
return new Promise((resolve, reject) => {
function decoratedReject(error) {
error.index = index
error.size = size
reject(error)
}
// this will handle one block response, but it can be another block
// since server does not guaranty to handle query in order
this.#commandQueryBacklog.set(queryId, {
size,
resolve,
reject: decoratedReject,
})
// really send the command to the server
this.#write(buffer).catch(decoratedReject)
// #readBlockResponse never throws directly
// but if it fails it will reject all the promises in the backlog
this.#readBlockResponse()
})
}
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()
},
})
}
/**
* returns the map of the file with holes, zeros and data, useful to handle efficiently sparse source
* to implement this internally: use structure response if the server supports it, and then ask for BLOCK_STATUS *
*
* @returns {Promise<{ offset: number, length: number, type: number }[]>}
* A promise that resolves to an array where each object represents a segment:
* - `offset` — The byte offset from the start.
* - `length` — The size of the segment in bytes.
* - `type` — A numeric code indicating the segment type ( 0 means data).
*/
/* async */ getMap(signal) {
return new Promise((resolve, reject) => {
const process = spawn('nbdinfo', [
'--json',
'--map',
`nbd://${this.#serverAddress}:${this.#serverPort}/${encodeURIComponent(this.#exportName)}`,
])
let text = ''
let errText = ''
process.stdout.on('data', data => (text += data))
process.stderr.on('data', data => (errText += data))
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}`))
}
try {
const json = JSON.parse(text)
resolve(json)
} catch (error) {
reject(new Error(`${error} ${text}`))
}
})
})
}
}
export { default as AbstractNbdClient } from './AbstractNbdClient.mjs'
export { default as NbdStdioClient } from './NbdStdioClient.mjs'
export { default as NbdTcpClient } from './NbdTcpClient.mjs'
@vates/nbd-client/tests/fake-nbd-server.mjs
+132
View File
@@ -0,0 +1,132 @@
import assert from 'node:assert'
import { pFromCallback } from 'promise-toolbox'
import { readChunk, readChunkStrict } from '@vates/read-chunk'
import {
INIT_PASSWD,
NBD_CMD_DISC,
NBD_CMD_READ,
NBD_FLAG_FIXED_NEWSTYLE,
NBD_FLAG_HAS_FLAGS,
NBD_FLAG_READ_ONLY,
NBD_OPT_EXPORT_NAME,
NBD_OPT_REPLY_MAGIC,
NBD_REPLY_ACK,
NBD_REPLY_MAGIC,
NBD_REQUEST_MAGIC,
OPTS_MAGIC,
} from '../constants.mjs'
const NBD_EINVAL = 22
/**
* Minimal read-only newstyle NBD server, used by the tests only.
*
* It is transport agnostic on purpose: the unit tests drive it through two
* in-memory `PassThrough`, the integration tests through the standard streams
* of a real process.
*
* @param {object} options
* @param {import('node:stream').Readable} options.readable - where the client queries are read from
* @param {import('node:stream').Writable} options.writable - where the answers are written to
* @param {Buffer} options.data - the content of the export
* @param {string} [options.exportName] - the export name the client must ask for
* @param {boolean} [options.answerInReverse] - answer by pairs, in reverse order, to check the client handles out of order answers
* @param {(request: {offset: bigint, length: number, index: number}) => number} [options.errorCode] - non zero to answer an error to this read
* @returns {Promise<void>} resolves when the client disconnected or closed the connection
*/
export async function serveNbd({ readable, writable, data, exportName = '', answerInReverse = false, errorCode }) {
const write = buffer => pFromCallback(cb => writable.write(buffer, cb))
// handshake: server flags
await write(INIT_PASSWD)
await write(OPTS_MAGIC)
const serverFlags = Buffer.alloc(2)
serverFlags.writeInt16BE(NBD_FLAG_FIXED_NEWSTYLE)
await write(serverFlags)
// client flags
await readChunkStrict(readable, 4)
// options, until the client selects an export
let selected = false
do {
assert.ok((await readChunkStrict(readable, 8)).equals(OPTS_MAGIC), 'option magic')
const option = (await readChunkStrict(readable, 4)).readInt32BE(0)
const length = (await readChunkStrict(readable, 4)).readInt32BE(0)
const payload = length > 0 ? await readChunkStrict(readable, length) : Buffer.alloc(0)
if (option === NBD_OPT_EXPORT_NAME) {
assert.strictEqual(payload.toString(), exportName, 'export name')
// 8 (size) + 2 (transmission flags) + 124 zeroes
const answer = Buffer.alloc(134)
answer.writeBigUInt64BE(BigInt(data.length), 0)
answer.writeInt16BE(NBD_FLAG_HAS_FLAGS | NBD_FLAG_READ_ONLY, 8)
await write(answer)
selected = true
} else {
// ack everything else
const answer = Buffer.alloc(20)
answer.writeBigUInt64BE(NBD_OPT_REPLY_MAGIC, 0)
answer.writeInt32BE(option, 8)
answer.writeInt32BE(NBD_REPLY_ACK, 12)
answer.writeInt32BE(0, 16)
await write(answer)
}
} while (!selected)
// transmission
let index = 0
let pending = []
const answer = async ({ handle, offset, length, index }, forcedCode) => {
const code = forcedCode ?? errorCode?.({ handle, offset, length, index }) ?? 0
const header = Buffer.alloc(16)
header.writeInt32BE(NBD_REPLY_MAGIC, 0)
header.writeInt32BE(code, 4)
header.writeBigUInt64BE(handle, 8)
await write(header)
if (code === 0) {
await write(data.subarray(Number(offset), Number(offset) + length))
}
}
const flush = async () => {
const requests = pending
pending = []
for (const request of requests.reverse()) {
await answer(request)
}
}
while (true) {
const request = await readChunk(readable, 28)
if (request === null || request.length < 28) {
// the client closed the connection without a NBD_CMD_DISC
break
}
assert.strictEqual(request.readInt32BE(0), NBD_REQUEST_MAGIC, 'request magic')
const type = request.readInt16BE(6)
const handle = request.readBigUInt64BE(8)
if (type === NBD_CMD_DISC) {
await flush()
break
}
if (type !== NBD_CMD_READ) {
await answer({ handle, offset: 0n, length: 0, index: index++ }, NBD_EINVAL)
continue
}
const read = {
handle,
offset: request.readBigUInt64BE(16),
length: request.readInt32BE(24),
index: index++,
}
pending.push(read)
if (!answerInReverse || pending.length === 2) {
await flush()
}
}
}
@vates/nbd-client/tests/nbdstdio.integ.mjs
+114
View File
@@ -0,0 +1,114 @@
import assert from 'node:assert'
import fs from 'node:fs/promises'
import { fileURLToPath } from 'node:url'
import { pFromCallback } from 'promise-toolbox'
import { test } from 'tap'
import tmp from 'tmp'
import NbdStdioClient from '../NbdStdioClient.mjs'
// nbdkit in single connection mode: the real thing this client targets
const NBDKIT = path => ({
command: 'nbdkit',
args: ['--single', '--exit-with-parent', '--read-only', 'file', path],
})
// a tiny NBD server we can ask to die in the middle of the transfer, which
// nbdkit won't do
const FLAKY_SERVER = fileURLToPath(new URL('./stdio-nbd-server.mjs', import.meta.url))
const CHUNK_SIZE = 64 * 1024
const FILE_SIZE = CHUNK_SIZE * 8 + 1232 // non aligned file size
async function createTempFile(size) {
const tmpPath = await pFromCallback(cb => tmp.file(cb))
const data = Buffer.alloc(size)
for (let i = 0; i < size; i += 4) {
data.writeUInt32BE(i, i)
}
await fs.writeFile(tmpPath, data)
return tmpPath
}
function checkBlock(tap, block, index) {
tap.equal(block.length, Math.min(CHUNK_SIZE, FILE_SIZE - index * CHUNK_SIZE), `block ${index} length`)
let ok = true
for (let i = 0; i < block.length; i += 4) {
ok = ok && block.readUInt32BE(i) === index * CHUNK_SIZE + i
}
tap.ok(ok, `block ${index} content`)
}
async function readWholeExport(tap, client) {
for (let index = 0; index < Math.ceil(FILE_SIZE / CHUNK_SIZE); index++) {
checkBlock(tap, await client.readBlock(index, CHUNK_SIZE), index)
}
}
const isAlive = pid => {
try {
process.kill(pid, 0)
return true
} catch (error) {
return false
}
}
test('it reads an export served by nbdkit on stdin/stdout', async tap => {
const path = await createTempFile(FILE_SIZE)
const client = new NbdStdioClient(NBDKIT(path))
await client.connect()
const { pid } = client
tap.equal(client.exportSize, BigInt(FILE_SIZE))
await readWholeExport(tap, client)
await client.disconnect()
tap.notOk(isAlive(pid), 'the server process is stopped after disconnect')
// double disconnection shouldn't pose any problem
await client.disconnect()
await fs.unlink(path)
})
test('it fails when the server does not serve the requested export', async tap => {
const path = await createTempFile(CHUNK_SIZE)
// the file plugin of nbdkit ignores the export name, so use our own server here
const client = new NbdStdioClient(
{
command: process.execPath,
args: [FLAKY_SERVER, path, '--export-name=MY_SECRET_EXPORT'],
exportname: 'NOPE',
},
{ connectTimeout: 5e3 }
)
await assert.rejects(client.connect())
tap.match(client.stderr, /NOPE/, 'the stderr of the server is captured')
await fs.unlink(path)
})
test('it recovers from a server dying in the middle of the transfer', async tap => {
const path = await createTempFile(FILE_SIZE)
const client = new NbdStdioClient(
{ command: process.execPath, args: [FLAKY_SERVER, path, '--die-after=3'] },
{ waitBeforeReconnect: 100 }
)
await client.connect()
await readWholeExport(tap, client)
await client.disconnect()
await fs.unlink(path)
})
test('it fails cleanly when the server cannot be spawned', async () => {
const client = new NbdStdioClient({ command: '/does/not/exist' })
await assert.rejects(client.connect(), ({ code, command }) => code === 'ENOENT' && command === '/does/not/exist')
// nothing to disconnect from
await client.disconnect()
})
test('it does not implement getMap', async () => {
const client = new NbdStdioClient({ command: 'nbdkit' })
await assert.rejects(client.getMap(), ({ code }) => code === 'NBD_MAP_UNSUPPORTED')
})
@vates/nbd-client/tests/stdio-nbd-server.mjs
+52
View File
@@ -0,0 +1,52 @@
#!/usr/bin/env node
// Read-only NBD server serving a file through its standard streams, à la
// `nbdkit -s`. Used by nbdstdio.integ.mjs, since nbdkit is not necessarily
// installed on the test machine.
//
// Usage: stdio-nbd-server.mjs <file> [--export-name=NAME] [--die-after=N]
//
// --die-after=N exits with a failure after having answered N read requests, to
// exercise the reconnection path of the client.
import fs from 'node:fs/promises'
import { serveNbd } from './fake-nbd-server.mjs'
const args = process.argv.slice(2)
const path = args.find(arg => !arg.startsWith('--'))
if (path === undefined) {
console.error('usage: stdio-nbd-server.mjs <file> [--export-name=NAME] [--die-after=N]')
process.exit(64)
}
const getOption = (name, defaultValue) => {
const prefix = `--${name}=`
const arg = args.find(arg => arg.startsWith(prefix))
return arg === undefined ? defaultValue : arg.slice(prefix.length)
}
const exportName = getOption('export-name', '')
const dieAfter = Number(getOption('die-after', Infinity))
const data = await fs.readFile(path)
let answered = 0
try {
await serveNbd({
readable: process.stdin,
writable: process.stdout,
data,
exportName,
errorCode: () => {
if (++answered > dieAfter) {
console.error(`stdio-nbd-server: dying after ${dieAfter} read(s)`)
process.exit(1)
}
return 0
},
})
} catch (error) {
console.error('stdio-nbd-server:', error)
process.exit(1)
}
CHANGELOG.unreleased.md
+1
View File
@@ -105,6 +105,7 @@
<!--packages-start-->
- @vates/iscsi minor
- @vates/nbd-client minor
- @vates/node-vsphere-soap minor
- @vates/types minor
- @xen-orchestra/acl minor