reduce scope and simplify

for running task : only compute the sum of running transfer
for finished task: use transferredSize / (end - start)

@vates/task/.USAGE.md

@@ -102,6 +102,69 @@ await Task.run.call(object, 'method')
 await Task.run.call(object, 'method', 'foo', 'bar')
 ```
 
+### `TaskMetrics`
+
+Metrics of a task (bytes read, bytes written, queue depth…), with a bounded history to draw charts.
+
+The metrics are bound to an explicit task, not to the current one, so they can be updated from any async context (e.g. a source read by multiple consumers). The task must be started before the first update.
+
+```js
+import { TaskMetrics } from '@vates/task/metrics'
+
+const metrics = new TaskMetrics(task, {
+  // maximum number of samples in the history, covering the whole duration of the task
+  historySize: 20,
+
+  // minimum duration between two emits, in milliseconds
+  minEmitInterval: 5e3,
+
+  // number of recent samples, one per emit
+  recentSize: 5,
+})
+
+// optional declaration, before the first update
+metrics.define('bytesRead', {
+  // emitted as is, for the clients to format the values
+  unit: 'bytes',
+})
+
+// expected final value, if known
+metrics.setTotal('bytesRead', bytesToRead)
+
+// adds to a cumulative metric
+metrics.add('bytesRead', block.length)
+
+// emits the latest values whatever the throttling, to call before ending the task
+metrics.flush()
+```
+
+The metrics are emitted in the `metrics` property of the task:
+
+```js
+{
+  bytesRead: {
+    // latest value
+    value: 1234567890,
+
+    // expected final value, if known
+    total: 3072327680,
+
+    // if defined
+    unit: 'bytes',
+
+    // latest samples ([timestamp, value]), one per emit: current speed
+    recent: [[1729000000000, 1200000000], …],
+
+    // at most `historySize` samples covering the whole duration of the task: chart
+    history: [[1728999000000, 0], …],
+  },
+}
+```
+
+The history interval starts at the emit interval and is doubled when the history is full: every other sample is dropped, which loses resolution, not volume, as the values are cumulative.
+
+A client watching the task can draw the history as the skeleton of the chart, then append the recent samples it receives to get a finer curve.
+
 ### `combineEvents`
 
 Create a consolidated log from individual events.

@vates/task/README.md

@@ -118,6 +118,69 @@ await Task.run.call(object, 'method')
 await Task.run.call(object, 'method', 'foo', 'bar')
 ```
 
+### `TaskMetrics`
+
+Metrics of a task (bytes read, bytes written, queue depth…), with a bounded history to draw charts.
+
+The metrics are bound to an explicit task, not to the current one, so they can be updated from any async context (e.g. a source read by multiple consumers). The task must be started before the first update.
+
+```js
+import { TaskMetrics } from '@vates/task/metrics'
+
+const metrics = new TaskMetrics(task, {
+  // maximum number of samples in the history, covering the whole duration of the task
+  historySize: 20,
+
+  // minimum duration between two emits, in milliseconds
+  minEmitInterval: 5e3,
+
+  // number of recent samples, one per emit
+  recentSize: 5,
+})
+
+// optional declaration, before the first update
+metrics.define('bytesRead', {
+  // emitted as is, for the clients to format the values
+  unit: 'bytes',
+})
+
+// expected final value, if known
+metrics.setTotal('bytesRead', bytesToRead)
+
+// adds to a cumulative metric
+metrics.add('bytesRead', block.length)
+
+// emits the latest values whatever the throttling, to call before ending the task
+metrics.flush()
+```
+
+The metrics are emitted in the `metrics` property of the task:
+
+```js
+{
+  bytesRead: {
+    // latest value
+    value: 1234567890,
+
+    // expected final value, if known
+    total: 3072327680,
+
+    // if defined
+    unit: 'bytes',
+
+    // latest samples ([timestamp, value]), one per emit: current speed
+    recent: [[1729000000000, 1200000000], …],
+
+    // at most `historySize` samples covering the whole duration of the task: chart
+    history: [[1728999000000, 0], …],
+  },
+}
+```
+
+The history interval starts at the emit interval and is doubled when the history is full: every other sample is dropped, which loses resolution, not volume, as the values are cumulative.
+
+A client watching the task can draw the history as the skeleton of the chart, then append the recent samples it receives to get a finer curve.
+
 ### `combineEvents`
 
 Create a consolidated log from individual events.

@vates/task/metrics.js

@@ -0,0 +1,180 @@
+'use strict'
+
+const assert = require('node:assert').strict
+
+class Metric {
+  constructor({ historyInterval, unit }) {
+    // samples (`[timestamp, value]`)
+    this.history = []
+    // only grows, see TaskMetrics
+    this.historyInterval = historyInterval
+    this.recent = []
+    this.total = undefined
+    this.unit = unit
+    this.value = 0
+  }
+
+  toJSON() {
+    const { history, recent, total, unit, value } = this
+    const json = { value, recent: recent.slice(), history: history.slice() }
+    if (total !== undefined) {
+      json.total = total
+    }
+    if (unit !== undefined) {
+      json.unit = unit
+    }
+    return json
+  }
+}
+
+/**
+ * Cumulative metrics of a task (bytes read, bytes written…), with a bounded history to draw charts
+ *
+ * The metrics are emitted in the `metrics` property of the task, throttled. Each metric is:
+ *
+ * - `value`: the latest value
+ * - `total`: the expected final value, if known
+ * - `unit`: the unit of the values, if defined (`bytes`, `blocks`…)
+ * - `recent`: the latest samples (`[timestamp, value]`), one per emit, to compute the current speed
+ * - `history`: samples covering the whole duration of the task, at most `historySize`
+ *
+ * The history interval starts at the emit interval and is doubled when the history is full: every other sample is
+ * dropped, which loses resolution, not volume, as the values are cumulative.
+ *
+ * The metrics are bound to an explicit task, not to the current one: they can be updated from any async context. The
+ * task must be started before the first update.
+ */
+class TaskMetrics {
+  #historySize
+  #lastEmitDate
+  #metrics = new Map()
+  #minEmitInterval
+  #recentSize
+  #task
+
+  /**
+   * @param {import('./index.js').Task} task
+   * @param {object} [opts]
+   * @param {number} [opts.historySize] maximum number of samples in the history
+   * @param {number} [opts.minEmitInterval] minimum duration between two emits, in milliseconds
+   * @param {number} [opts.recentSize] number of recent samples
+   */
+  constructor(task, { historySize = 20, minEmitInterval = 5e3, recentSize = 5 } = {}) {
+    assert(historySize >= 2, 'historySize must be at least 2')
+    assert(recentSize >= 2, 'recentSize must be at least 2')
+
+    this.#historySize = historySize
+    this.#minEmitInterval = minEmitInterval
+    this.#recentSize = recentSize
+    this.#task = task
+  }
+
+  /**
+   * Declares a metric, before its first update
+   *
+   * Optional: an undeclared metric has no unit
+   *
+   * @param {string} name
+   * @param {object} [opts]
+   * @param {string} [opts.unit]
+   */
+  define(name, { unit } = {}) {
+    assert(!this.#metrics.has(name), `metric ${name} is already defined`)
+
+    this.#metrics.set(name, new Metric({ historyInterval: this.#minEmitInterval, unit }))
+  }
+
+  #getMetric(name) {
+    let metric = this.#metrics.get(name)
+    if (metric === undefined) {
+      metric = new Metric({ historyInterval: this.#minEmitInterval })
+      this.#metrics.set(name, metric)
+    }
+    return metric
+  }
+
+  /**
+   * Adds `delta` to a metric
+   *
+   * @param {string} name
+   * @param {number} delta
+   */
+  add(name, delta) {
+    this.#getMetric(name).value += delta
+    this.#maybeEmit()
+  }
+
+  /**
+   * Sets the expected final value of a metric
+   *
+   * @param {string} name
+   * @param {number | undefined} total
+   */
+  setTotal(name, total) {
+    this.#getMetric(name).total = total
+  }
+
+  /**
+   * Emits the metrics now, whatever the throttling, the latest values are always in the history
+   *
+   * To call once the task is done, before ending it
+   */
+  flush() {
+    this.#emit({ isFinal: true })
+  }
+
+  #maybeEmit() {
+    if (this.#lastEmitDate === undefined || Date.now() - this.#lastEmitDate >= this.#minEmitInterval) {
+      this.#emit({ isFinal: false })
+    }
+  }
+
+  #emit({ isFinal }) {
+    const task = this.#task
+    // ended task: late updates are ignored
+    if (task.status !== 'pending') {
+      return
+    }
+
+    const now = Date.now()
+    this.#lastEmitDate = now
+
+    const metrics = {}
+    for (const [name, metric] of this.#metrics) {
+      this.#sample(metric, now, isFinal)
+      metrics[name] = metric.toJSON()
+    }
+    task.set('metrics', metrics)
+  }
+
+  #sample(metric, now, isFinal) {
+    const sample = [now, metric.value]
+
+    const { recent } = metric
+    recent.push(sample)
+    if (recent.length > this.#recentSize) {
+      recent.shift()
+    }
+
+    const { history } = metric
+    const last = history[history.length - 1]
+    if (last === undefined || now - last[0] >= metric.historyInterval) {
+      history.push(sample)
+    } else if (isFinal) {
+      if (history.length > 1) {
+        // the final value replaces the latest sample to keep the spacing
+        history[history.length - 1] = sample
+      } else {
+        // but never the first one
+        history.push(sample)
+      }
+    }
+
+    if (history.length > this.#historySize) {
+      // keep every other sample, and always the latest one
+      metric.history = history.filter((_, i) => i % 2 === 0 || i === history.length - 1)
+      metric.historyInterval *= 2
+    }
+  }
+}
+exports.TaskMetrics = TaskMetrics

@vates/task/metrics.test.js

@@ -0,0 +1,175 @@
+'use strict'
+
+const assert = require('node:assert').strict
+const { afterEach, beforeEach, describe, it, mock } = require('node:test')
+
+const { Task } = require('./index.js')
+const { TaskMetrics } = require('./metrics.js')
+
+const EMIT_INTERVAL = 5e3
+
+// started task recording its `metrics` property
+function createTask() {
+  const emitted = []
+  const task = new Task({
+    onProgress(event) {
+      if (event.type === 'property' && event.name === 'metrics') {
+        emitted.push(event.value)
+      }
+    },
+  })
+  task.start()
+  task.emitted = emitted
+  return task
+}
+
+describe('TaskMetrics', () => {
+  beforeEach(() => {
+    mock.timers.enable({ apis: ['Date'], now: 0 })
+  })
+  afterEach(() => {
+    mock.timers.reset()
+  })
+
+  it('emits the cumulative values, throttled', () => {
+    const task = createTask()
+    const metrics = new TaskMetrics(task)
+
+    metrics.add('bytesRead', 10)
+    metrics.add('bytesRead', 10) // throttled
+    mock.timers.tick(EMIT_INTERVAL)
+    metrics.add('bytesRead', 10)
+
+    assert.deepEqual(
+      task.emitted.map(_ => _.bytesRead.value),
+      [10, 30]
+    )
+  })
+
+  it('always emits the final value', () => {
+    const task = createTask()
+    const metrics = new TaskMetrics(task)
+
+    metrics.add('bytesRead', 10)
+    mock.timers.tick(1e3)
+    metrics.add('bytesRead', 10) // throttled
+    metrics.flush()
+
+    const last = task.emitted.at(-1).bytesRead
+    assert.equal(last.value, 20)
+    assert.deepEqual(last.history.at(-1), [1e3, 20])
+    // the first sample is kept
+    assert.deepEqual(last.history[0], [0, 10])
+  })
+
+  it('keeps a bounded number of recent samples', () => {
+    const task = createTask()
+    const metrics = new TaskMetrics(task, { recentSize: 3 })
+
+    for (let i = 1; i <= 10; i++) {
+      metrics.add('bytesRead', 1)
+      mock.timers.tick(EMIT_INTERVAL)
+    }
+
+    assert.deepEqual(task.emitted.at(-1).bytesRead.recent, [
+      [7 * EMIT_INTERVAL, 8],
+      [8 * EMIT_INTERVAL, 9],
+      [9 * EMIT_INTERVAL, 10],
+    ])
+  })
+
+  it('keeps a bounded history covering the whole duration, without total', () => {
+    const task = createTask()
+    const historySize = 10
+    const metrics = new TaskMetrics(task, { historySize })
+
+    const nbEmits = 1000
+    for (let i = 1; i <= nbEmits; i++) {
+      metrics.add('bytesRead', 1)
+      mock.timers.tick(EMIT_INTERVAL)
+    }
+    metrics.flush()
+
+    const { history } = task.emitted.at(-1).bytesRead
+    assert.ok(history.length <= historySize, `${history.length} samples`)
+    assert.ok(history.length >= historySize / 2, `${history.length} samples`)
+    // covers the whole duration
+    assert.deepEqual(history[0], [0, 1])
+    assert.deepEqual(history.at(-1), [nbEmits * EMIT_INTERVAL, nbEmits])
+    // sorted, and the values are cumulative
+    for (let i = 1; i < history.length; i++) {
+      assert.ok(history[i][0] > history[i - 1][0])
+      assert.ok(history[i][1] >= history[i - 1][1])
+    }
+  })
+
+  it('handles several metrics independently', () => {
+    const task = createTask()
+    const metrics = new TaskMetrics(task, { historySize: 4 })
+
+    for (let i = 1; i <= 100; i++) {
+      metrics.add('bytesRead', 2)
+      metrics.add('bytesWritten', 1)
+      mock.timers.tick(EMIT_INTERVAL)
+    }
+    metrics.flush()
+
+    const { bytesRead, bytesWritten } = task.emitted.at(-1)
+    assert.equal(bytesRead.value, 200)
+    assert.equal(bytesWritten.value, 100)
+    for (const { history } of [bytesRead, bytesWritten]) {
+      assert.ok(history.length >= 2 && history.length <= 4, `${history.length} samples`)
+      assert.equal(history.at(-1)[0], 100 * EMIT_INTERVAL)
+    }
+  })
+
+  it('emits copies, not the internal state', () => {
+    const task = createTask()
+    const metrics = new TaskMetrics(task)
+
+    metrics.add('bytesRead', 1)
+    const first = task.emitted[0].bytesRead
+    mock.timers.tick(EMIT_INTERVAL)
+    metrics.add('bytesRead', 1)
+
+    assert.equal(first.value, 1)
+    assert.equal(first.recent.length, 1)
+  })
+
+  it('ignores the updates once the task has ended', () => {
+    const task = createTask()
+    const metrics = new TaskMetrics(task)
+
+    metrics.add('bytesRead', 1)
+    task.success()
+    mock.timers.tick(EMIT_INTERVAL)
+    metrics.add('bytesRead', 1)
+    metrics.flush()
+
+    assert.equal(task.emitted.length, 1)
+  })
+
+  it('emits the total and the unit', () => {
+    const task = createTask()
+    const metrics = new TaskMetrics(task)
+    metrics.define('bytesRead', { unit: 'bytes' })
+    metrics.setTotal('bytesRead', 10)
+
+    metrics.add('bytesRead', 1)
+    metrics.add('other', 1)
+    metrics.flush()
+
+    const { bytesRead, other } = task.emitted.at(-1)
+    assert.equal(bytesRead.total, 10)
+    assert.equal(bytesRead.unit, 'bytes')
+    assert.equal(other.total, undefined)
+    assert.equal(other.unit, undefined)
+  })
+
+  it('rejects a metric defined twice', () => {
+    const metrics = new TaskMetrics(createTask())
+
+    metrics.define('a')
+    assert.throws(() => metrics.define('a'), /already defined/)
+  })
+})

@vates/task/package.json

@@ -23,6 +23,7 @@
   },
   "exports": {
     ".": "./index.js",
-    "./combineEvents": "./combineEvents.js"
+    "./combineEvents": "./combineEvents.js",
+    "./metrics": "./metrics.js"
   }
 }

@xen-orchestra/backups/ImportVmBackup.mjs

@@ -3,6 +3,7 @@ import assert from 'node:assert'
 import { formatFilenameDate } from './_filenameDate.mjs'
 import { importIncrementalVm } from './_incrementalVm.mjs'
 import { watchStreamSize } from './_watchStreamSize.mjs'
+import { meterIncrementalExport, SourceReadTask } from './_SourceReadTask.mjs'
 import { decorateClass } from '@vates/decorate-with'
 import { Task } from '@vates/task'
 import { createLogger } from '@xen-orchestra/log'
@@ -410,6 +411,7 @@ export class ImportVmBackup {
         throw new Error('live mounting a disk is only supported when restoring an incremental backup')
       }
       backup = await adapter.readFullVmBackup(metadata)
+      sizeContainer.readTask = new SourceReadTask({ name: 'read' })
       watchStreamSize(backup, sizeContainer)
     } else {
       assert.strictEqual(metadata.mode, 'delta')
@@ -418,6 +420,7 @@ export class ImportVmBackup {
       // releases them, a successful one hands their ids back to the caller
       $defer.onFailure(() => this.#releaseLiveMounts())
       backup = await this._decorateIncrementalVmMetadata()
+      meterIncrementalExport(backup)
     }
 
     return Task.run(

@xen-orchestra/backups/_SourceReadTask.mjs

@@ -0,0 +1,116 @@
+import { Task } from '@vates/task'
+import { TaskMetrics } from '@vates/task/metrics'
+import { MeteredDisk } from '@xen-orchestra/disk-transform'
+
+const MAX_DURATION_BETWEEN_EMIT = 5e3
+
+/**
+ * Reports the reading of a source (a disk, a XVA) in its own task, child of the current task
+ *
+ * A source is read once, even when it is consumed by multiple writers, but from the async context of the writer
+ * pulling the next block: the task is held here instead of relying on the current task
+ *
+ * Its lifecycle follows the source: started before the first read, ended when the source is exhausted, fails or is
+ * closed early. A source never read (a step failing before the transfer, a disk without blocks) leaves no task
+ *
+ * The bytes read are reported in the `bytesRead` metric (`metrics` property, see `@vates/task/metrics`), with the
+ * number of bytes to read as total, and the `progress` property when this number is known
+ *
+ * Implements `ReadMeter` from `@xen-orchestra/disk-transform`
+ */
+export class SourceReadTask {
+  #isStarted = false
+  #lastProgressDate = 0
+  #bytesToRead
+  #bytesRead = 0
+  #metrics
+  #task
+
+  /**
+   * @param {Record<string, unknown>} properties of the task, must not contain `type` or `id`: xo-web would display it
+   *   as a VM/SR/remote task
+   */
+  constructor(properties) {
+    // created here to be a child of the current task, only started on the first read
+    this.#task = new Task({ properties })
+    this.#metrics = new TaskMetrics(this.#task, { minEmitInterval: MAX_DURATION_BETWEEN_EMIT })
+    this.#metrics.define('bytesRead', { unit: 'bytes' })
+  }
+
+  get bytesRead() {
+    return this.#bytesRead
+  }
+
+  #setProgress() {
+    this.#lastProgressDate = Date.now()
+    if (this.#bytesToRead > 0) {
+      this.#task.set('progress', Math.min(100, Math.round((this.#bytesRead * 100) / this.#bytesToRead)))
+    }
+  }
+
+  /**
+   * No-op once started
+   *
+   * @param {object} [opts]
+   * @param {number} [opts.bytesToRead] number of bytes to read, if known
+   */
+  onStart({ bytesToRead } = {}) {
+    if (this.#isStarted) {
+      return
+    }
+    this.#isStarted = true
+    this.#bytesToRead = bytesToRead
+    this.#task.start()
+    this.#metrics.setTotal('bytesRead', bytesToRead)
+  }
+
+  /**
+   * @param {number} length number of bytes read
+   */
+  onRead(length) {
+    this.onStart()
+    if (this.#task.status !== 'pending') {
+      // already ended, late reads are ignored
+      return
+    }
+    this.#bytesRead += length
+    this.#metrics.add('bytesRead', length)
+    if (Date.now() - this.#lastProgressDate >= MAX_DURATION_BETWEEN_EMIT) {
+      this.#setProgress()
+    }
+  }
+
+  /**
+   * A source closed early is not a failure: its consumers stopped, they report their own errors
+   *
+   * @param {{ isComplete: boolean, error?: unknown }} result
+   */
+  onEnd({ isComplete, error }) {
+    const task = this.#task
+    if (!this.#isStarted || task.status !== 'pending') {
+      return
+    }
+    this.#metrics.flush()
+    this.#setProgress()
+    if (error !== undefined) {
+      task.failure(error)
+    } else {
+      task.success({ isComplete })
+    }
+  }
+}
+
+/**
+ * Reports the reading of each disk of an incremental export in its own task, child of the current task
+ *
+ * Each disk is replaced by a `MeteredDisk`, which only exposes the generic `Disk` interface: source specific methods
+ * (`useNbd()` for example) must be called before
+ *
+ * @param {{ disks: Record<string, import('@xen-orchestra/disk-transform').Disk>, vdis: Record<string, { name_label: string, uuid: string }> }} incrementalExport
+ */
+export function meterIncrementalExport({ disks, vdis }) {
+  for (const [key, disk] of Object.entries(disks)) {
+    const vdi = vdis[key]
+    disks[key] = new MeteredDisk(disk, new SourceReadTask({ name: `read disk ${vdi.name_label}`, vdiUuid: vdi.uuid }))
+  }
+}

@xen-orchestra/backups/_SourceReadTask.test.mjs

@@ -0,0 +1,243 @@
+import assert from 'node:assert/strict'
+import { Readable, Writable } from 'node:stream'
+import { pipeline } from 'node:stream/promises'
+import { describe, it } from 'node:test'
+import { setTimeout as delay } from 'node:timers/promises'
+import { Task } from '@vates/task'
+import { MeteredDisk, RandomAccessDisk, SynchronizedDisk } from '@xen-orchestra/disk-transform'
+
+import { forkStreamUnpipe } from './_runners/_forkStreamUnpipe.mjs'
+import { meterIncrementalExport, SourceReadTask } from './_SourceReadTask.mjs'
+import { watchStreamSize } from './_watchStreamSize.mjs'
+
+const BLOCK_SIZE = 1024
+const NB_BLOCKS = 20
+
+class MemoryDisk extends RandomAccessDisk {
+  getVirtualSize() {
+    return NB_BLOCKS * BLOCK_SIZE
+  }
+  getBlockSize() {
+    return BLOCK_SIZE
+  }
+  async init() {}
+  async close() {}
+  isDifferencing() {
+    return false
+  }
+  getBlockIndexes() {
+    return Array.from({ length: NB_BLOCKS }, (_, index) => index)
+  }
+  hasBlock() {
+    return true
+  }
+  async readBlock(index) {
+    return { index, data: Buffer.alloc(BLOCK_SIZE) }
+  }
+}
+
+/**
+ * Runs `fn` in a root task, and returns the tasks created inside, by name
+ */
+async function collectTasks(fn) {
+  const tasksById = new Map()
+  const root = new Task({
+    properties: { name: 'root' },
+    onProgress(event) {
+      if (event.type === 'start') {
+        const { name, ...properties } = event.properties
+        tasksById.set(event.id, { id: event.id, name, parentId: event.parentId, properties })
+        return
+      }
+      const task = tasksById.get(event.id)
+      if (task === undefined) {
+        return
+      }
+      if (event.type === 'property') {
+        task.properties[event.name] = event.value
+      } else if (event.type === 'end') {
+        task.status = event.status
+        task.result = event.result
+      }
+    },
+  })
+  await root.run(fn).catch(() => {})
+  const tasksByName = {}
+  for (const task of tasksById.values()) {
+    task.isRootChild = task.parentId === root.id
+    tasksByName[task.name] = task
+  }
+  return tasksByName
+}
+
+// each writer consumes its fork at its own pace, in its own task
+function runWriters(consumers) {
+  return Promise.all(Object.entries(consumers).map(([name, consume]) => Task.run({ properties: { name } }, consume)))
+}
+
+function createMeteredDisk(diskProperties = { name: 'read disk', vdiUuid: 'vdi-uuid' }) {
+  return new MeteredDisk(new MemoryDisk(), new SourceReadTask(diskProperties))
+}
+
+describe('SourceReadTask', () => {
+  it('reports a disk read by multiple writers only in its own task', async () => {
+    const tasks = await collectTasks(async () => {
+      const synchronizedDisk = new SynchronizedDisk(createMeteredDisk())
+      const forks = { backup: synchronizedDisk.fork('backup'), replication: synchronizedDisk.fork('replication') }
+
+      await runWriters({
+        async backup() {
+          for await (const block of forks.backup.diskBlocks()) {
+            assert.equal(block.data.length, BLOCK_SIZE)
+            await delay(1)
+          }
+        },
+        async replication() {
+          for await (const block of forks.replication.diskBlocks()) {
+            assert.equal(block.data.length, BLOCK_SIZE)
+            await delay(Math.random() * 3)
+          }
+        },
+      })
+    })
+
+    const readTask = tasks['read disk']
+    assert.equal(readTask.isRootChild, true)
+    assert.equal(readTask.status, 'success')
+    assert.deepEqual(readTask.result, { isComplete: true })
+    assert.deepEqual(Object.keys(readTask.properties).sort(), ['metrics', 'progress', 'vdiUuid'])
+    assert.equal(readTask.properties.vdiUuid, 'vdi-uuid')
+    assert.equal(readTask.properties.progress, 100)
+    const { bytesRead } = readTask.properties.metrics
+    assert.equal(bytesRead.value, NB_BLOCKS * BLOCK_SIZE)
+    assert.equal(bytesRead.total, NB_BLOCKS * BLOCK_SIZE)
+    // the final value is always in the history
+    assert.equal(bytesRead.history.at(-1)[1], NB_BLOCKS * BLOCK_SIZE)
+    assert.deepEqual(tasks.backup.properties, {})
+    assert.deepEqual(tasks.replication.properties, {})
+  })
+
+  it('reports a forked stream of unknown length only in its own task', async () => {
+    const chunks = Array.from({ length: NB_BLOCKS }, () => Buffer.alloc(BLOCK_SIZE))
+    const tasks = await collectTasks(async () => {
+      const stream = Readable.from(chunks)
+      watchStreamSize(stream, { size: 0, readTask: new SourceReadTask({ name: 'read' }) })
+      const forks = [forkStreamUnpipe(stream), forkStreamUnpipe(stream)]
+      const sink = () =>
+        new Writable({
+          write(_chunk, _encoding, callback) {
+            setTimeout(callback, 1)
+          },
+        })
+
+      await runWriters({
+        remote: () => pipeline(forks[0], sink()),
+        sr: () => pipeline(forks[1], sink()),
+      })
+    })
+
+    assert.equal(tasks.read.isRootChild, true)
+    assert.equal(tasks.read.status, 'success')
+    assert.deepEqual(tasks.read.result, { isComplete: true })
+    assert.equal(tasks.read.properties.metrics.bytesRead.value, NB_BLOCKS * BLOCK_SIZE)
+    assert.equal(tasks.read.properties.metrics.bytesRead.total, undefined)
+    assert.equal(tasks.read.properties.progress, undefined)
+  })
+
+  it('leaves no task when the source is never read', async () => {
+    const tasks = await collectTasks(async () => {
+      createMeteredDisk()
+    })
+
+    assert.deepEqual(Object.keys(tasks), ['root'])
+  })
+
+  it('succeeds as incomplete when its consumers stop early', async () => {
+    const tasks = await collectTasks(async () => {
+      let nbRead = 0
+      for await (const block of createMeteredDisk().diskBlocks()) {
+        assert.equal(block.data.length, BLOCK_SIZE)
+        if (++nbRead === 5) {
+          break
+        }
+      }
+    })
+
+    assert.equal(tasks['read disk'].status, 'success')
+    assert.deepEqual(tasks['read disk'].result, { isComplete: false })
+    assert.equal(tasks['read disk'].properties.metrics.bytesRead.value, 5 * BLOCK_SIZE)
+    assert.equal(tasks['read disk'].properties.progress, 25)
+  })
+
+  it('fails when the source fails', async () => {
+    const tasks = await collectTasks(async () => {
+      const disk = new MemoryDisk()
+      disk.readBlock = async index => {
+        if (index === 2) {
+          throw new Error('NBD connection lost')
+        }
+        return { index, data: Buffer.alloc(BLOCK_SIZE) }
+      }
+      const meteredDisk = new MeteredDisk(disk, new SourceReadTask({ name: 'read disk' }))
+      await assert.rejects(async () => {
+        for await (const block of meteredDisk.diskBlocks()) {
+          assert.equal(block.data.length, BLOCK_SIZE)
+        }
+      })
+    })
+
+    assert.equal(tasks['read disk'].status, 'failure')
+    assert.equal(tasks['read disk'].result.message, 'NBD connection lost')
+    assert.equal(tasks['read disk'].properties.metrics.bytesRead.value, 2 * BLOCK_SIZE)
+  })
+
+  it('reports each disk of an incremental export in its own task', async () => {
+    const tasks = await collectTasks(async () => {
+      const incrementalExport = {
+        disks: { 'OpaqueRef:1': new MemoryDisk(), 'OpaqueRef:2': new MemoryDisk() },
+        vdis: {
+          'OpaqueRef:1': { name_label: 'system', uuid: 'vdi-1' },
+          'OpaqueRef:2': { name_label: 'data', uuid: 'vdi-2' },
+        },
+      }
+      meterIncrementalExport(incrementalExport)
+      for (const disk of Object.values(incrementalExport.disks)) {
+        assert.ok(disk instanceof MeteredDisk)
+        for await (const block of disk.diskBlocks()) {
+          assert.equal(block.data.length, BLOCK_SIZE)
+        }
+      }
+    })
+
+    for (const [name, vdiUuid] of [
+      ['read disk system', 'vdi-1'],
+      ['read disk data', 'vdi-2'],
+    ]) {
+      assert.equal(tasks[name].isRootChild, true)
+      assert.equal(tasks[name].status, 'success')
+      assert.equal(tasks[name].properties.vdiUuid, vdiUuid)
+      assert.equal(tasks[name].properties.metrics.bytesRead.value, NB_BLOCKS * BLOCK_SIZE)
+    }
+  })
+
+  it('groups the reads of a restore point in the task it is metered in', async () => {
+    const tasks = await collectTasks(async () => {
+      const incrementalExport = {
+        disks: { 'OpaqueRef:1': new MemoryDisk() },
+        vdis: { 'OpaqueRef:1': { name_label: 'system', uuid: 'vdi-1' } },
+      }
+      const readTask = new Task({ properties: { name: 'read', timestamp: 1 } })
+      await readTask.runInside(() => meterIncrementalExport(incrementalExport))
+      for await (const block of incrementalExport.disks['OpaqueRef:1'].diskBlocks()) {
+        assert.equal(block.data.length, BLOCK_SIZE)
+      }
+      readTask.success()
+    })
+
+    assert.equal(tasks.read.isRootChild, true)
+    assert.equal(tasks.read.status, 'success')
+    assert.equal(tasks['read disk system'].isRootChild, false)
+    assert.equal(tasks['read disk system'].parentId, tasks.read.id)
+    assert.equal(tasks['read disk system'].status, 'success')
+  })
+})

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

@@ -4,6 +4,7 @@ import { AbstractRemote } from './_AbstractRemote.mjs'
 import { FullRemoteWriter } from '../_writers/FullRemoteWriter.mjs'
 import { forkStreamUnpipe } from '../_forkStreamUnpipe.mjs'
 import { watchStreamSize } from '../../_watchStreamSize.mjs'
+import { SourceReadTask } from '../../_SourceReadTask.mjs'
 import { AggregatedFullRemoteWriter } from '../_writers/AggregatedFullRemoteWriter.mjs'
 import { Task } from '@vates/task'
 
@@ -24,7 +25,8 @@ export const FullRemote = class FullRemoteVmBackupRunner extends AbstractRemote
     let nbTransferredVms = 0
     for (const metadata of transferList) {
       const stream = this._throttleStream(await this._sourceRemoteAdapter.readFullVmBackup(metadata))
-      const sizeContainer = watchStreamSize(stream)
+      // the backup is read once, whatever the number of writers
+      const sizeContainer = watchStreamSize(stream, { size: 0, readTask: new SourceReadTask({ name: 'read' }) })
       let checksum
       try {
         // the unencrypted destinations store the same data as the source: no need to compute their checksum again

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

@@ -4,6 +4,7 @@ import { forkStreamUnpipe } from '../_forkStreamUnpipe.mjs'
 import { FullRemoteWriter } from '../_writers/FullRemoteWriter.mjs'
 import { FullXapiWriter } from '../_writers/FullXapiWriter.mjs'
 import { watchStreamSize } from '../../_watchStreamSize.mjs'
+import { SourceReadTask } from '../../_SourceReadTask.mjs'
 import { AbstractXapi } from './_AbstractXapi.mjs'
 import { AggregatedFullRemoteWriter } from '../_writers/AggregatedFullRemoteWriter.mjs'
 import { AggregatedFullXapiWriter } from '../_writers/AggregatedFullXapiWriter.mjs'
@@ -78,7 +79,8 @@ export const FullXapi = class FullXapiVmBackupRunner extends AbstractXapi {
       maxStreamLength += XVA_BLOCK_SIZE * nbBlocks + headersSize
     }
 
-    const sizeContainer = watchStreamSize(stream)
+    // the export is read once, whatever the number of writers
+    const sizeContainer = watchStreamSize(stream, { size: 0, readTask: new SourceReadTask({ name: 'read' }) })
 
     const transferStart = Date.now()
     await this._callWriters(

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

@@ -13,6 +13,7 @@ import { SynchronizedDisk, ThrottledDisk } from '@xen-orchestra/disk-transform'
 import { AggregatedIncrementalRemoteWriter } from '../_writers/AggregatedIncrementalRemoteWriter.mjs'
 import { Task } from '@vates/task'
 import { TaskProgressHandler } from './_TaskProgressHandler.mjs'
+import { meterIncrementalExport } from '../../_SourceReadTask.mjs'
 
 const { warn } = createLogger('xo:backups:Incrementalremote')
 class IncrementalRemoteVmBackupRunner extends AbstractRemote {
@@ -77,11 +78,7 @@ class IncrementalRemoteVmBackupRunner extends AbstractRemote {
       const isVhdDifferencing = {}
 
       for (const key in incrementalExport.disks) {
-        let disk = incrementalExport.disks[key]
-        isVhdDifferencing[key] = disk.isDifferencing()
-        disk.addProgressHandler(new TaskProgressHandler())
-        disk = new ThrottledDisk(disk, this._throttleGenerator)
-        incrementalExport.disks[key] = new SynchronizedDisk(disk)
+        isVhdDifferencing[key] = incrementalExport.disks[key].isDifferencing()
       }
 
       const hasDifferencingDisk = Object.values(isVhdDifferencing).includes(true)
@@ -97,18 +94,37 @@ class IncrementalRemoteVmBackupRunner extends AbstractRemote {
       await this._selectBaseVm(metadata)
       await this._callWriters(writer => writer.prepare({ isBase: metadata.isBase }), 'writer.prepare()')
 
-      await this._callWriters(
-        writer =>
-          writer.transfer({
-            deltaExport: forkDeltaExport(incrementalExport, writer.constructor.name),
-            includeNonNbdQcow2Fix: metadata.includeNonNbdQcow2Fix,
-            isVhdDifferencing,
-            timestamp: metadata.timestamp,
-            vm: metadata.vm,
-            vmSnapshot: metadata.vmSnapshot,
-          }),
-        'writer.transfer()'
-      )
+      // one task per restore point, holding the reading of its disks from the source remote: each disk is read once,
+      // whatever the number of writers
+      // not `id` nor `type`: xo-web would display it as a remote task
+      const readTask = new Task({
+        properties: { name: 'read', remoteId: this.job.sourceRemote, timestamp: metadata.timestamp },
+      })
+      await readTask.runInside(() => meterIncrementalExport(incrementalExport))
+      for (const key in incrementalExport.disks) {
+        const disk = incrementalExport.disks[key]
+        disk.addProgressHandler(new TaskProgressHandler())
+        incrementalExport.disks[key] = new SynchronizedDisk(new ThrottledDisk(disk, this._throttleGenerator))
+      }
+
+      try {
+        await this._callWriters(
+          writer =>
+            writer.transfer({
+              deltaExport: forkDeltaExport(incrementalExport, writer.constructor.name),
+              includeNonNbdQcow2Fix: metadata.includeNonNbdQcow2Fix,
+              isVhdDifferencing,
+              timestamp: metadata.timestamp,
+              vm: metadata.vm,
+              vmSnapshot: metadata.vmSnapshot,
+            }),
+          'writer.transfer()'
+        )
+      } catch (error) {
+        readTask.failure(error)
+        throw error
+      }
+      readTask.success()
 
       await this._callWriters(writer => writer.cleanup(), 'writer.cleanup()')
       // for healthcheck

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

@@ -18,6 +18,7 @@ import { ThrottledDisk, SynchronizedDisk } from '@xen-orchestra/disk-transform'
 import { AggregatedIncrementalRemoteWriter } from '../_writers/AggregatedIncrementalRemoteWriter.mjs'
 import { AggregatedIncrementalXapiWriter } from '../_writers/AggregatedIncrementalXapiWriter.mjs'
 import { TaskProgressHandler } from './_TaskProgressHandler.mjs'
+import { meterIncrementalExport } from '../../_SourceReadTask.mjs'
 
 const { debug } = createLogger('xo:backups:IncrementalXapiVmBackup')
 
@@ -48,8 +49,12 @@ export const IncrementalXapi = class IncrementalXapiVmBackupRunner extends Abstr
       preferNbd: this._settings.preferNbd,
     })
 
+    // specific to the XAPI disks: read before they are wrapped
+    const useNbd = Object.values(deltaExport.disks).some(disk => disk.useNbd())
+    // each disk is read once, whatever the number of writers
+    meterIncrementalExport(deltaExport)
+
     const isVhdDifferencing = {}
-    let useNbd = false
     for (const key in deltaExport.disks) {
       let disk = deltaExport.disks[key]
       disk.addProgressHandler(new TaskProgressHandler())
@@ -57,7 +62,6 @@ export const IncrementalXapi = class IncrementalXapiVmBackupRunner extends Abstr
       if (!isFull && !isVhdDifferencing[key] && key !== exportedVm.$suspend_VDI?.$ref) {
         Task.warning('Backup fell back to a full')
       }
-      useNbd = useNbd || disk.useNbd()
       disk = new ThrottledDisk(disk, this._throttleGenerator)
       deltaExport.disks[key] = new SynchronizedDisk(disk)
     }

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

@@ -13,7 +13,9 @@ export class TaskProgressHandler {
     if (progress < 0 || progress > 1) {
       return
     }
+    // the end is always emitted: throttled, a last step under the threshold would leave the task at 99%
     if (
+      progress !== 1 &&
       this.#lastProgressDate !== undefined &&
       this.#lastProgressValue !== undefined &&
       Date.now() - this.#lastProgressDate < MAX_DURATION_BETWEEN_PROGRESS_EMIT &&

@xen-orchestra/backups/_runners/_vmRunners/_TaskProgressHandler.test.mjs

@@ -0,0 +1,41 @@
+import assert from 'node:assert/strict'
+import { describe, it } from 'node:test'
+import { Task } from '@vates/task'
+
+import { TaskProgressHandler } from './_TaskProgressHandler.mjs'
+
+async function collectProgress(fn) {
+  const progress = []
+  const task = new Task({
+    properties: { name: 'transfer' },
+    onProgress(event) {
+      if (event.type === 'property' && event.name === 'progress') {
+        progress.push(event.value)
+      }
+    },
+  })
+  await task.run(fn)
+  return progress
+}
+
+describe('TaskProgressHandler', () => {
+  it('throttles the intermediate steps', async () => {
+    const progress = await collectProgress(async () => {
+      const progressHandler = new TaskProgressHandler()
+      for (let i = 1; i < 1000; i++) {
+        await progressHandler.setProgress(i / 1000)
+      }
+    })
+    // initial value + at most one emit per percent
+    assert.ok(progress.length <= 101, `${progress.length} emits`)
+  })
+
+  it('always emits the end, even when the last step is under the threshold', async () => {
+    const progress = await collectProgress(async () => {
+      const progressHandler = new TaskProgressHandler()
+      await progressHandler.setProgress(0.991)
+      await progressHandler.setProgress(1)
+    })
+    assert.equal(progress.at(-1), 100)
+  })
+})

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

@@ -23,6 +23,7 @@ import { diskHasData } from './_diskHasData.mjs'
 const { warn } = createLogger('xo:backups:DeltaBackupWriter')
 
 export class IncrementalRemoteWriter extends MixinRemoteWriter(AbstractIncrementalWriter) {
+  #exportTask
   #parentVdiPaths
   #parentUuids
   async checkBaseVdis(baseUuidToSrcVdi) {
@@ -141,20 +142,29 @@ export class IncrementalRemoteWriter extends MixinRemoteWriter(AbstractIncrement
   }
 
   prepare({ isFull }) {
+    // the mirror calls prepare() once per restore point: the previous export is over, unless it has already failed
+    if (this.#exportTask?.status === 'pending') {
+      this.#exportTask.success()
+    }
+
     // create the task related to this export and ensure all methods are called in this context
-    const task = new Task({
+    const task = (this.#exportTask = new Task({
       properties: {
         id: this._remoteId,
         isFull,
         name: 'export',
         type: 'remote',
       },
-    })
-    this._prepare = task.wrapInside(this._prepare)
-    this.transfer = task.wrapInside(this.transfer)
-    this.healthCheck = task.wrapInside(this.healthCheck)
-    this.cleanup = task.wrapInside(this.cleanup)
-    this.afterBackup = task.wrap(this.afterBackup)
+    }))
+    // wrap the methods of the class, not the ones wrapped by a previous call, which would run in all the previous
+    // exports
+    const methods = Object.getPrototypeOf(this)
+    this._prepare = task.wrapInside(methods._prepare)
+    this.transfer = task.wrapInside(methods.transfer)
+    this.healthCheck = task.wrapInside(methods.healthCheck)
+    this.cleanup = task.wrapInside(methods.cleanup)
+    // ends the last export
+    this.afterBackup = task.wrap(methods.afterBackup)
 
     return this._prepare()
   }

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

@@ -0,0 +1,93 @@
+import assert from 'node:assert/strict'
+import { describe, it } from 'node:test'
+import { Task } from '@vates/task'
+
+import { IncrementalRemoteWriter } from './IncrementalRemoteWriter.mjs'
+
+// only the task handling is tested: the remote operations are stubbed
+class TestWriter extends IncrementalRemoteWriter {
+  constructor({ failingTimestamp } = {}) {
+    super({ remoteId: 'remote-id', adapter: {}, vmUuid: 'vm-uuid', job: {}, settings: {}, config: {}, schedule: {} })
+    this.failingTimestamp = failingTimestamp
+  }
+
+  async _prepare() {}
+
+  async transfer({ timestamp }) {
+    await Task.run({ properties: { name: 'transfer', timestamp } }, async () => {
+      if (timestamp === this.failingTimestamp) {
+        throw new Error('transfer failed')
+      }
+    })
+  }
+
+  async cleanup() {}
+
+  async healthCheck() {}
+
+  async afterBackup() {}
+}
+
+/**
+ * Runs `fn` in a root task, and returns the tasks created inside
+ */
+async function collectTasks(fn) {
+  const tasksById = new Map()
+  const root = new Task({
+    properties: { name: 'root' },
+    onProgress(event) {
+      if (event.type === 'start') {
+        tasksById.set(event.id, { id: event.id, parentId: event.parentId, properties: event.properties })
+      } else if (event.type === 'end') {
+        tasksById.get(event.id).status = event.status
+      }
+    },
+  })
+  await root.run(fn).catch(() => {})
+  return [...tasksById.values()]
+}
+
+// what the mirror runner does for each restore point to transfer
+async function transferRestorePoint(writer, timestamp) {
+  await writer.prepare({ isFull: false })
+  await writer.transfer({ timestamp })
+  await writer.cleanup()
+}
+
+describe('IncrementalRemoteWriter.prepare()', () => {
+  it('creates one export per restore point, each holding its own transfer', async () => {
+    const tasks = await collectTasks(async () => {
+      const writer = new TestWriter()
+      for (const timestamp of [1, 2, 3]) {
+        await transferRestorePoint(writer, timestamp)
+      }
+      await writer.healthCheck()
+      await writer.afterBackup()
+    })
+
+    const exports = tasks.filter(task => task.properties.name === 'export')
+    assert.equal(exports.length, 3)
+    for (const [i, exportTask] of exports.entries()) {
+      assert.equal(exportTask.status, 'success')
+      const transfers = tasks.filter(task => task.parentId === exportTask.id)
+      assert.deepEqual(
+        transfers.map(task => task.properties.timestamp),
+        [i + 1]
+      )
+    }
+  })
+
+  it('only fails the export of the failing restore point', async () => {
+    const tasks = await collectTasks(async () => {
+      const writer = new TestWriter({ failingTimestamp: 2 })
+      await transferRestorePoint(writer, 1)
+      await transferRestorePoint(writer, 2)
+    })
+
+    const exports = tasks.filter(task => task.properties.name === 'export')
+    assert.deepEqual(
+      exports.map(task => task.status),
+      ['success', 'failure']
+    )
+  })
+})

@xen-orchestra/backups/_watchStreamSize.mjs

@@ -1,17 +1,41 @@
 import { TaskProgressHandler } from './_runners/_vmRunners/_TaskProgressHandler.mjs'
+
+/**
+ * @param {import('node:stream').Readable} stream
+ * @param {{ size: number, progressHandler?: TaskProgressHandler, readTask?: import('./_SourceReadTask.mjs').SourceReadTask }} [container]
+ *   `readTask`: reports the reading of the stream in its own task
+ */
 export function watchStreamSize(stream, container = { size: 0 }) {
-  const progressHandler = container.progressHandler ?? new TaskProgressHandler()
+  const { progressHandler = new TaskProgressHandler(), readTask } = container
   stream.on('data', data => {
+    if (container.size === 0) {
+      // empty for xva exported by xapi
+      readTask?.onStart({ bytesToRead: stream.length || undefined })
+    }
     container.size += data.length
     if (stream.length) {
       // empty for xva exported by xapi
       progressHandler.setProgress(container.size / stream.length)
     }
+    readTask?.onRead(data.length)
   })
 
-  stream.on('finish', () => {
-    progressHandler.setProgress(1)
-  })
+  // a readable only emits 'end', a duplex (throttled stream) emits both
+  let isComplete = false
+  const complete = () => {
+    if (!isComplete) {
+      isComplete = true
+      progressHandler.setProgress(1)
+      readTask?.onEnd({ isComplete })
+    }
+  }
+  stream.on('end', complete)
+  stream.on('finish', complete)
+  if (readTask !== undefined) {
+    // no-op once the read task has ended
+    stream.on('error', error => readTask.onEnd({ isComplete, error }))
+    stream.on('close', () => readTask.onEnd({ isComplete }))
+  }
   stream.pause()
   return container
 }

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

@@ -1,4 +1,5 @@
 import { Disk, DiskBlock, RandomAccessDisk } from './Disk.mjs'
+import type { ProgressHandler } from './ProgressHandler.mjs'
 
 export class DiskPassthrough extends Disk {
   #source: Disk | undefined
@@ -51,6 +52,10 @@ export class DiskPassthrough extends Disk {
   async buildDiskBlockGenerator(): Promise<AsyncGenerator<DiskBlock>> {
     return this.source.buildDiskBlockGenerator()
   }
+  // the blocks are generated by the source, which is the one reporting the progress
+  addProgressHandler(progressHandler: ProgressHandler) {
+    this.source.addProgressHandler(progressHandler)
+  }
 }
 
 export abstract class RandomDiskPassthrough extends RandomAccessDisk {

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

@@ -0,0 +1,55 @@
+import { Disk, DiskBlock } from './Disk.mjs'
+import { DiskPassthrough } from './DiskPassthrough.mjs'
+
+export type ReadMeter = {
+  /** called once, before the first block, with the number of bytes the source will generate */
+  onStart(result: { bytesToRead: number }): void
+  /** called for each block read from the source */
+  onRead(length: number): void
+  /** called once, when the source is exhausted (`isComplete`), closed early by its consumer or failed (`error`) */
+  onEnd(result: { isComplete: boolean; error?: unknown }): void
+}
+
+/**
+ * Reports the bytes read from the source, and the end of its reading
+ *
+ * The meter is called from the async context of the consumer pulling the next block, which may be any of the
+ * forks of a SynchronizedDisk: it must not rely on this context
+ */
+export class MeteredDisk extends DiskPassthrough {
+  #meter: ReadMeter
+
+  constructor(source: Disk, meter: ReadMeter) {
+    super(source)
+    this.#meter = meter
+  }
+
+  async buildDiskBlockGenerator(): Promise<AsyncGenerator<DiskBlock>> {
+    const generator = await this.source.buildDiskBlockGenerator()
+    const meter = this.#meter
+    // the last block may be truncated to the virtual size
+    const bytesToRead = Math.min(this.getBlockIndexesCount() * this.getBlockSize(), this.getVirtualSize())
+    async function* meteredGenerator(): AsyncGenerator<DiskBlock> {
+      let isComplete = false
+      let isStarted = false
+      let error: unknown
+      try {
+        for await (const block of generator) {
+          if (!isStarted) {
+            isStarted = true
+            meter.onStart({ bytesToRead })
+          }
+          meter.onRead(block.data.length)
+          yield block
+        }
+        isComplete = true
+      } catch (err) {
+        error = err
+        throw err
+      } finally {
+        meter.onEnd({ isComplete, error })
+      }
+    }
+    return meteredGenerator()
+  }
+}

@xen-orchestra/disk-transform/src/MeteredDisk.test.mts

@@ -0,0 +1,120 @@
+import { test } from 'node:test'
+import assert from 'node:assert'
+import { DebugDisk } from './utils/DebugDisk.mjs'
+import { MeteredDisk, ReadMeter } from './MeteredDisk.mjs'
+import { SynchronizedDisk } from './SynchronizedDisk.mjs'
+import { DiskBlock } from './Disk.mjs'
+
+const blockSize = 1024
+const nbBlocks = 20
+
+function createMeter() {
+  const meter = {
+    readSize: 0,
+    bytesToRead: [] as number[],
+    ends: [] as Array<{ isComplete: boolean; error?: unknown }>,
+    onStart({ bytesToRead }: { bytesToRead: number }) {
+      meter.bytesToRead.push(bytesToRead)
+    },
+    onRead(length: number) {
+      meter.readSize += length
+    },
+    onEnd(result: { isComplete: boolean; error?: unknown }) {
+      meter.ends.push(result)
+    },
+  } satisfies ReadMeter & Record<string, unknown>
+  return meter
+}
+
+async function createDisk() {
+  const disk = new DebugDisk({ nbBlocks, blockSize, fillRate: 100 })
+  await disk.init()
+  return disk
+}
+
+test('MeteredDisk', async t => {
+  await t.test('reports the bytes read and the completion', async () => {
+    const disk = await createDisk()
+    const meter = createMeter()
+    const metered = new MeteredDisk(disk, meter)
+    for await (const block of metered.diskBlocks()) {
+      assert.strictEqual(block.data.length, blockSize)
+    }
+    assert.deepStrictEqual(meter.bytesToRead, [nbBlocks * blockSize])
+    assert.strictEqual(meter.readSize, nbBlocks * blockSize)
+    assert.deepStrictEqual(meter.ends, [{ isComplete: true, error: undefined }])
+    assert.strictEqual(disk.closeDone, true)
+  })
+
+  await t.test('reports an early stop of the consumer as incomplete', async () => {
+    const meter = createMeter()
+    const metered = new MeteredDisk(await createDisk(), meter)
+    let nbRead = 0
+    for await (const block of metered.diskBlocks()) {
+      assert.strictEqual(block.data.length, blockSize)
+      if (++nbRead === 3) {
+        break
+      }
+    }
+    assert.strictEqual(meter.readSize, 3 * blockSize)
+    assert.deepStrictEqual(meter.ends, [{ isComplete: false, error: undefined }])
+  })
+
+  await t.test('reports the error of the source', async () => {
+    const disk = await createDisk()
+    const error = new Error('read failed')
+    disk.readBlock = async (index: number): Promise<DiskBlock> => {
+      if (index === 5) {
+        throw error
+      }
+      return { index, data: Buffer.alloc(blockSize) }
+    }
+    const meter = createMeter()
+    const metered = new MeteredDisk(disk, meter)
+    await assert.rejects(async () => {
+      for await (const block of metered.diskBlocks()) {
+        assert.strictEqual(block.data.length, blockSize)
+      }
+    }, error)
+    assert.strictEqual(meter.readSize, 5 * blockSize)
+    assert.deepStrictEqual(meter.ends, [{ isComplete: false, error }])
+  })
+
+  await t.test('counts a source read once by multiple forks', async () => {
+    const meter = createMeter()
+    const synchronized = new SynchronizedDisk(new MeteredDisk(await createDisk(), meter))
+    const forks = [synchronized.fork('first'), synchronized.fork('second')]
+    await Promise.all(
+      forks.map(async fork => {
+        for await (const block of fork.diskBlocks()) {
+          assert.strictEqual(block.data.length, blockSize)
+          await new Promise(resolve => setTimeout(resolve, Math.random() * 3))
+        }
+      })
+    )
+    assert.strictEqual(meter.readSize, nbBlocks * blockSize)
+    assert.deepStrictEqual(meter.ends, [{ isComplete: true, error: undefined }])
+  })
+
+  await t.test('reports the size of the blocks present in the source', async () => {
+    const disk = new DebugDisk({ nbBlocks, blockSize, fillRate: 50 })
+    await disk.init()
+    const meter = createMeter()
+    for await (const block of new MeteredDisk(disk, meter).diskBlocks()) {
+      assert.strictEqual(block.data.length, blockSize)
+    }
+    assert.deepStrictEqual(meter.bytesToRead, [disk.getBlockIndexesCount() * blockSize])
+    assert.strictEqual(meter.readSize, disk.getBlockIndexesCount() * blockSize)
+  })
+
+  await t.test('does not start on a source without blocks', async () => {
+    const disk = new DebugDisk({ nbBlocks, blockSize, fillRate: 0 })
+    await disk.init()
+    const meter = createMeter()
+    for await (const block of new MeteredDisk(disk, meter).diskBlocks()) {
+      assert.strictEqual(block.data.length, blockSize)
+    }
+    assert.deepStrictEqual(meter.bytesToRead, [])
+    assert.deepStrictEqual(meter.ends, [{ isComplete: true, error: undefined }])
+  })
+})

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

@@ -9,5 +9,7 @@ export { RawDisk } from './RawDisk.mjs'
 export { ReadAhead } from './ReadAhead.mjs'
 export { SynchronizedDisk } from './SynchronizedDisk.mjs'
 export { TimeoutDisk } from './Timeout.mjs'
+export { MeteredDisk } from './MeteredDisk.mjs'
+export type { ReadMeter } from './MeteredDisk.mjs'
 export { ThrottledDisk } from './Throttled.mjs'
 export { ProgressHandler } from './ProgressHandler.mjs'

@xen-orchestra/web-core/lib/locales/en.json

@@ -274,6 +274,8 @@
   "bucket-name": "Bucket name",
   "build-number": "Build number",
   "by": "By",
+  "bytes-read": "Data read",
+  "bytes-to-read": "Data to read",
   "bytes:gi": "GiB",
   "bytes:kbs": "kB/s",
   "bytes:ki": "KiB",

@xen-orchestra/web-core/lib/locales/fr.json

@@ -274,6 +274,8 @@
   "bucket-name": "Nom du bucket",
   "build-number": "Numéro de build",
   "by": "Par",
+  "bytes-read": "Données lues",
+  "bytes-to-read": "Données à lire",
   "bytes:gi": "Gio",
   "bytes:kbs": "kB/s",
   "bytes:ki": "Kio",

@xen-orchestra/web/src/modules/task/components/list/panel/TaskSidePanel.vue

@@ -2,6 +2,7 @@
   <VtsSidePanel :has-selection="!!task" @close="emit('close')">
     <template v-if="task">
       <TaskQuickInfosCard :task />
+      <TaskReadSpeedCard v-if="readSamples.length > 1" :samples="readSamples" />
       <TaskInfosCard v-if="task.infos" :task />
       <TaskWarningsCard v-if="task.warnings" :task />
       <TaskErrorsCard v-if="isError" :task />
@@ -17,9 +18,11 @@ import TaskInfosCard from '@/modules/task/components/list/panel/cards/TaskInfosC
 import TaskObjectsCard from '@/modules/task/components/list/panel/cards/TaskObjectsCard.vue'
 import TaskPropertiesCard from '@/modules/task/components/list/panel/cards/TaskPropertiesCard.vue'
 import TaskQuickInfosCard from '@/modules/task/components/list/panel/cards/TaskQuickInfosCard.vue'
+import TaskReadSpeedCard from '@/modules/task/components/list/panel/cards/TaskReadSpeedCard.vue'
 import TaskWarningsCard from '@/modules/task/components/list/panel/cards/TaskWarningsCard.vue'
 import { useXoTaskPropertiesUtils } from '@/modules/task/composables/xo-task-properties-utils.composable.ts'
 import type { FrontXoTask } from '@/modules/task/remote-resources/use-xo-task-collection.ts'
+import { getTaskMetric } from '@/modules/task/utils/task-metric.util.ts'
 import VtsSidePanel from '@core/components/panel/VtsSidePanel.vue'
 import { computed } from 'vue'
 
@@ -33,5 +36,8 @@ const emit = defineEmits<{
 
 const { properties } = useXoTaskPropertiesUtils(() => task)
 
+// bytes read from the sources of this task and its subtasks
+const readSamples = computed(() => (task === undefined ? [] : (getTaskMetric(task, 'bytesRead')?.samples ?? [])))
+
 const isError = computed(() => task && task.result && (task.status === 'failure' || task.status === 'interrupted'))
 </script>

@xen-orchestra/web/src/modules/task/components/list/panel/cards/TaskQuickInfosCard.vue

@@ -49,6 +49,18 @@
           {{ formattedEndDate }}
         </template>
       </VtsCardRowKeyValue>
+      <VtsCardRowKeyValue v-if="formattedBytesToRead !== undefined">
+        <template #key>{{ t('bytes-to-read') }}</template>
+        <template #value>{{ formattedBytesToRead }}</template>
+      </VtsCardRowKeyValue>
+      <VtsCardRowKeyValue v-if="formattedBytesRead !== undefined">
+        <template #key>{{ t('bytes-read') }}</template>
+        <template #value>{{ formattedBytesRead }}</template>
+      </VtsCardRowKeyValue>
+      <VtsCardRowKeyValue v-if="formattedSpeed !== undefined">
+        <template #key>{{ t('reading-speed') }}</template>
+        <template #value>{{ formattedSpeed }}</template>
+      </VtsCardRowKeyValue>
 
       <VtsCardRowKeyValue>
         <template #key>{{ t('status') }}</template>
@@ -70,6 +82,7 @@
 <script lang="ts" setup>
 import { useXoTaskNameResolver } from '@/modules/task/composables/xo-task-name-resolver.composable.ts'
 import type { FrontXoTask } from '@/modules/task/remote-resources/use-xo-task-collection.ts'
+import { getTaskMetric } from '@/modules/task/utils/task-metric.util.ts'
 import { getTaskAccents } from '@/modules/task/utils/xo-task.util.ts'
 import { useXoUserResource } from '@/modules/user/remote-resources/use-xo-user.ts'
 import VtsCardRowKeyValue from '@core/components/card/VtsCardRowKeyValue.vue'
@@ -82,6 +95,8 @@ import UiPanelCardTitle from '@core/components/ui/panel-card-title/UiPanelCardTi
 import UiTag from '@core/components/ui/tag/UiTag.vue'
 import UiTagsList from '@core/components/ui/tag/UiTagsList.vue'
 import UiUserLogo from '@core/components/ui/user-logo/UiUserLogo.vue'
+import { formatSize } from '@core/utils/size.util.ts'
+import { formatSpeed } from '@core/utils/speed.util.ts'
 import { computed } from 'vue'
 import { useI18n } from 'vue-i18n'
 
@@ -129,6 +144,29 @@ const formattedEndDate = computed(() => {
 
   return d(endTimestamp, 'datetime_short')
 })
+
+// bytes read from the sources (disks, XVA) of this task and its subtasks, each source counted once whatever the number
+// of targets: this is not what is written
+const read = computed(() => getTaskMetric(task, 'bytesRead'))
+
+const formattedBytesToRead = computed(() => {
+  const bytesToRead = read.value?.total
+
+  return bytesToRead !== undefined ? formatSize(bytesToRead, 2) : undefined
+})
+
+const formattedBytesRead = computed(() => {
+  const bytesRead = read.value?.value
+
+  return bytesRead !== undefined ? formatSize(bytesRead, 2) : undefined
+})
+
+// average speed once ended, otherwise the current speed of the reads still running
+const formattedSpeed = computed(() => {
+  const bytesPerSecond = read.value?.speed
+
+  return bytesPerSecond ? formatSpeed(bytesPerSecond) : undefined
+})
 </script>
 
 <style scoped lang="postcss">

@xen-orchestra/web/src/modules/task/components/list/panel/cards/TaskReadSpeedCard.vue

@@ -0,0 +1,36 @@
+<template>
+  <UiPanelCard class="task-read-speed-card">
+    <UiPanelCardTitle size="medium" :label="t('reading-speed')" />
+    <VtsLinearChart :data="chartData" :max-value :value-formatter="formatChartSpeed" />
+  </UiPanelCard>
+</template>
+
+<script lang="ts" setup>
+import { getSpeedSeries, type MetricSample } from '@/modules/task/utils/task-metric.util.ts'
+import type { LinearChartData } from '@core/types/chart.ts'
+import UiPanelCard from '@core/components/ui/panel-card/UiPanelCard.vue'
+import UiPanelCardTitle from '@core/components/ui/panel-card-title/UiPanelCardTitle.vue'
+import { formatSpeed } from '@core/utils/speed.util.ts'
+import { computed, defineAsyncComponent } from 'vue'
+import { useI18n } from 'vue-i18n'
+
+// cumulative bytes read, consolidated over the task and its subtasks
+const { samples } = defineProps<{
+  samples: MetricSample[]
+}>()
+
+const VtsLinearChart = defineAsyncComponent(() => import('@core/components/linear-chart/VtsLinearChart.vue'))
+
+const { t } = useI18n()
+
+const speedSeries = computed(() => getSpeedSeries(samples))
+
+const chartData = computed<LinearChartData>(() => [{ label: t('reading-speed'), data: speedSeries.value }])
+
+// some headroom above the highest speed
+const maxValue = computed(() => Math.max(1, ...speedSeries.value.map(({ value }) => value)) * 1.1)
+
+function formatChartSpeed(value: number | null): string {
+  return value === null ? '' : formatSpeed(value)
+}
+</script>

@xen-orchestra/web/src/modules/task/utils/task-metric.util.ts

@@ -0,0 +1,261 @@
+import type { XoTask } from '@vates/types'
+
+/** `[timestamp, cumulative value]` */
+export type MetricSample = [timestamp: number, value: number]
+
+export type TaskTree = Pick<XoTask, 'start' | 'end' | 'properties'> & { tasks?: TaskTree[] }
+
+export type TaskMetric = {
+  /** unit of the values, unknown if the sources don't report the same one */
+  unit: string | undefined
+  value: number
+  /** expected final value, unknown if one of the sources doesn't report it */
+  total: number | undefined
+  /** per second: the current speed of the sources still running, or the average speed once the task has ended */
+  speed: number
+  /** cumulative samples, sorted, to draw a chart */
+  samples: MetricSample[]
+}
+
+// a metric as emitted by `TaskMetrics` from `@vates/task/metrics`
+type RawMetric = {
+  unit?: string
+  value: number
+  total?: number
+  recent?: MetricSample[]
+  history?: MetricSample[]
+}
+
+type ConsolidatedMetric = TaskMetric & { isPending: boolean }
+
+// without a new sample for this long, the current speed of a source decreases
+const STALE_DELAY = 10e3
+
+const DEFAULT_MAX_SAMPLES = 60
+
+function isSample(sample: unknown): sample is MetricSample {
+  return Array.isArray(sample) && typeof sample[0] === 'number' && typeof sample[1] === 'number'
+}
+
+function getRawMetric(task: TaskTree, name: string): RawMetric | undefined {
+  const metrics = task.properties.metrics as Record<string, unknown> | undefined
+  const metric = metrics?.[name] as Partial<RawMetric> | undefined
+
+  if (metric === undefined || typeof metric.value !== 'number') {
+    return undefined
+  }
+
+  return {
+    unit: typeof metric.unit === 'string' ? metric.unit : undefined,
+    value: metric.value,
+    total: typeof metric.total === 'number' ? metric.total : undefined,
+    recent: Array.isArray(metric.recent) ? metric.recent.filter(isSample) : [],
+    history: Array.isArray(metric.history) ? metric.history.filter(isSample) : [],
+  }
+}
+
+// the history is the skeleton of the curve, the recent samples refine its end
+function getOwnSamples(task: TaskTree, { history = [], recent = [] }: RawMetric): MetricSample[] {
+  const valueByTime = new Map<number, number>()
+  for (const [time, value] of [...history, ...recent]) {
+    valueByTime.set(time, value)
+  }
+  const samples = [...valueByTime].sort(([a], [b]) => a - b)
+
+  // the source started from 0 when its task started
+  const first = samples[0]
+  if (first !== undefined && first[1] > 0 && task.start < first[0]) {
+    samples.unshift([task.start, 0])
+  }
+
+  return samples
+}
+
+function getAverageSpeed(task: TaskTree, value: number, now: number) {
+  const elapsedMs = (task.end ?? now) - task.start
+
+  return elapsedMs > 0 ? (value * 1e3) / elapsedMs : 0
+}
+
+// slope of the recent samples, decreasing if no sample has been received for a while
+function getCurrentSpeed(task: TaskTree, { value, recent = [] }: RawMetric, now: number) {
+  const first = recent[0]
+  const last = recent.at(-1)
+
+  if (first === undefined || last === undefined || recent.length < 2) {
+    return getAverageSpeed(task, value, now)
+  }
+
+  const end = Math.max(last[0], now - STALE_DELAY)
+
+  return end > first[0] ? ((last[1] - first[1]) * 1e3) / (end - first[0]) : 0
+}
+
+/**
+ * Value of a cumulative series at `time`: 0 before its first sample, linear between two samples, the last value after
+ */
+export function getValueAt(samples: MetricSample[], time: number): number {
+  const first = samples[0]
+  const last = samples.at(-1)
+
+  if (first === undefined || last === undefined || time < first[0]) {
+    return 0
+  }
+  if (time >= last[0]) {
+    return last[1]
+  }
+
+  // samples are sorted: find the first one after `time`
+  let i = 1
+  while (samples[i][0] <= time) {
+    i++
+  }
+  const [t0, v0] = samples[i - 1]
+  const [t1, v1] = samples[i]
+
+  return v0 + ((v1 - v0) * (time - t0)) / (t1 - t0)
+}
+
+/**
+ * Sums the cumulative series of the children, whose samples are not aligned
+ *
+ * Each child is evaluated at the times of all the samples, interpolating between its own ones.
+ *
+ * A child still running only knows its value up to its latest sample: the sum stops at the earliest latest sample of
+ * the running children, otherwise a child not reported yet would make the curve flatten then jump.
+ */
+export function sumSamples(
+  children: { samples: MetricSample[]; isPending: boolean }[],
+  maxSamples = DEFAULT_MAX_SAMPLES
+): MetricSample[] {
+  let watermark = Infinity
+  for (const { samples, isPending } of children) {
+    const last = samples.at(-1)
+    if (isPending && last !== undefined) {
+      watermark = Math.min(watermark, last[0])
+    }
+  }
+
+  let times = [...new Set(children.flatMap(({ samples }) => samples.map(([time]) => time)))]
+    .filter(time => time <= watermark)
+    .sort((a, b) => a - b)
+
+  // many children: resample on a regular grid
+  if (times.length > maxSamples) {
+    const start = times[0]
+    const step = (times[times.length - 1] - start) / (maxSamples - 1)
+    times = Array.from({ length: maxSamples }, (_, i) => start + i * step)
+  }
+
+  return times.map(time => [time, children.reduce((sum, { samples }) => sum + getValueAt(samples, time), 0)])
+}
+
+// the value shared by all the items, undefined if they differ
+function getCommonValue(values: (string | undefined)[]): string | undefined {
+  return values.every(value => value === values[0]) ? values[0] : undefined
+}
+
+function consolidate(task: TaskTree, name: string, now: number, maxSamples: number): ConsolidatedMetric | undefined {
+  const isPending = task.end === undefined
+  const raw = getRawMetric(task, name)
+
+  // a task reporting the metric is a source: its subtasks are not visited, so the same values are never counted twice
+  if (raw !== undefined) {
+    return {
+      unit: raw.unit,
+      value: raw.value,
+      total: raw.total,
+      speed: isPending ? getCurrentSpeed(task, raw, now) : 0,
+      samples: getOwnSamples(task, raw),
+      isPending,
+    }
+  }
+
+  const children = (task.tasks ?? [])
+    .map(subtask => consolidate(subtask, name, now, maxSamples))
+    .filter(child => child !== undefined)
+
+  if (children.length === 0) {
+    return undefined
+  }
+
+  return {
+    unit: getCommonValue(children.map(child => child.unit)),
+    value: children.reduce((sum, child) => sum + child.value, 0),
+    total: children.every(child => child.total !== undefined)
+      ? children.reduce((sum, child) => sum + child.total!, 0)
+      : undefined,
+    // only the children still running have a speed
+    speed: children.reduce((sum, child) => sum + child.speed, 0),
+    samples: sumSamples(children, maxSamples),
+    isPending,
+  }
+}
+
+/**
+ * Consolidates a metric (`bytesRead`…) reported in the `metrics` property of a task and its subtasks
+ *
+ * - value and total: sum of the sources
+ * - speed: while running, sum of the current speeds of the sources still running; once ended, value / duration
+ * - samples: sum of the sources' series, to draw a chart
+ * - unit: the one of the sources, if they all share it
+ */
+export function getTaskMetric(
+  task: TaskTree,
+  name: string,
+  { now = Date.now(), maxSamples = DEFAULT_MAX_SAMPLES } = {}
+): TaskMetric | undefined {
+  const metric = consolidate(task, name, now, maxSamples)
+
+  if (metric === undefined) {
+    return undefined
+  }
+
+  const { isPending, ...taskMetric } = metric
+  if (!isPending) {
+    taskMetric.speed = getAverageSpeed(task, taskMetric.value, now)
+  }
+
+  return taskMetric
+}
+
+/**
+ * Speed (per second) of a cumulative series, averaged over a sliding window ending at each sample
+ *
+ * Consecutive samples can be very close (the first read blocks, already prefetched, right after the start, or the
+ * samples of different sources in a consolidated series): the speed between them is meaningless. The window lasts at
+ * least `minWindow`, and at least `1 / maxPoints` of the whole series, so the chart has the same smoothness whatever
+ * the duration.
+ *
+ * @param opts.minWindow in milliseconds
+ */
+export function getSpeedSeries(
+  samples: MetricSample[],
+  { minWindow = 10e3, maxPoints = 30 } = {}
+): { timestamp: number; value: number }[] {
+  const first = samples[0]
+  const last = samples.at(-1)
+  if (first === undefined || last === undefined || last[0] <= first[0]) {
+    return []
+  }
+
+  // a series shorter than the window: a single average
+  const window = Math.min(last[0] - first[0], Math.max(minWindow, (last[0] - first[0]) / maxPoints))
+
+  const series: { timestamp: number; value: number }[] = []
+  // latest sample at least `window` before the current one
+  let start = 0
+  for (let i = 1; i < samples.length; i++) {
+    const [time, value] = samples[i]
+    while (start + 1 < i && time - samples[start + 1][0] >= window) {
+      start++
+    }
+
+    const [startTime, startValue] = samples[start]
+    if (time - startTime >= window) {
+      series.push({ timestamp: time, value: ((value - startValue) * 1e3) / (time - startTime) })
+    }
+  }
+
+  return series
+}

@xen-orchestra/web/src/modules/task/utils/task-metric.util.unit.test.ts

@@ -0,0 +1,308 @@
+import {
+  getSpeedSeries,
+  getTaskMetric,
+  getValueAt,
+  sumSamples,
+  type MetricSample,
+  type TaskTree,
+} from '@/modules/task/utils/task-metric.util.ts'
+import { describe, expect, it } from 'vitest'
+
+const NOW = 100_000
+
+function createTask({ start = 0, end, properties = {}, tasks }: Partial<TaskTree> = {}): TaskTree {
+  return { start, end, properties, tasks }
+}
+
+function createSource({
+  start = 0,
+  end,
+  value,
+  total,
+  unit = 'bytes',
+  recent = [],
+  history = [],
+}: {
+  start?: number
+  end?: number
+  value: number
+  total?: number
+  unit?: string
+  recent?: MetricSample[]
+  history?: MetricSample[]
+}): TaskTree {
+  return createTask({
+    start,
+    end,
+    properties: { metrics: { bytesRead: { unit, value, total, recent, history } } },
+  })
+}
+
+describe('getValueAt', () => {
+  const samples: MetricSample[] = [
+    [10, 100],
+    [20, 300],
+  ]
+
+  it('is 0 before the first sample', () => {
+    expect(getValueAt(samples, 5)).toBe(0)
+  })
+
+  it('interpolates between two samples', () => {
+    expect(getValueAt(samples, 15)).toBe(200)
+  })
+
+  it('keeps the last value after the last sample', () => {
+    expect(getValueAt(samples, 50)).toBe(300)
+  })
+})
+
+describe('sumSamples', () => {
+  it('sums children whose samples are not aligned', () => {
+    const a = {
+      isPending: false,
+      samples: [
+        [0, 0],
+        [10, 100],
+        [20, 200],
+      ] as MetricSample[],
+    }
+    const b = {
+      isPending: false,
+      samples: [
+        [5, 0],
+        [15, 50],
+      ] as MetricSample[],
+    }
+
+    expect(sumSamples([a, b])).toEqual([
+      [0, 0],
+      [5, 50],
+      [10, 125],
+      [15, 200],
+      [20, 250],
+    ])
+  })
+
+  it('stops at the earliest latest sample of the running children', () => {
+    const running = {
+      isPending: true,
+      samples: [
+        [0, 0],
+        [10, 100],
+      ] as MetricSample[],
+    }
+    const ahead = {
+      isPending: true,
+      samples: [
+        [0, 0],
+        [12, 120],
+      ] as MetricSample[],
+    }
+    const ended = {
+      isPending: false,
+      samples: [
+        [0, 0],
+        [5, 50],
+      ] as MetricSample[],
+    }
+
+    const samples = sumSamples([running, ahead, ended])
+
+    expect(samples.at(-1)).toEqual([10, 100 + 100 + 50])
+  })
+
+  it('resamples on a regular grid when there are too many samples', () => {
+    const children = Array.from({ length: 10 }, (_, i) => ({
+      isPending: false,
+      samples: [
+        [i, 0],
+        [100 + i, 10],
+      ] as MetricSample[],
+    }))
+
+    const samples = sumSamples(children, 5)
+
+    expect(samples.map(([time]) => time)).toEqual([0, 27.25, 54.5, 81.75, 109])
+    expect(samples.at(-1)).toEqual([109, 100])
+  })
+})
+
+describe('getTaskMetric', () => {
+  it('returns undefined when the metric is not reported', () => {
+    expect(getTaskMetric(createTask({ tasks: [createTask()] }), 'bytesRead', { now: NOW })).toBeUndefined()
+  })
+
+  it('uses the slope of the recent samples as the current speed of a running source', () => {
+    const source = createSource({
+      value: 3_000,
+      total: 10_000,
+      recent: [
+        [90_000, 1_000],
+        [95_000, 2_000],
+        [NOW, 3_000],
+      ],
+    })
+
+    expect(getTaskMetric(source, 'bytesRead', { now: NOW })).toMatchObject({
+      value: 3_000,
+      total: 10_000,
+      speed: 200,
+    })
+  })
+
+  it('decreases the current speed of a source without new sample', () => {
+    const source = createSource({
+      value: 2_000,
+      recent: [
+        [50_000, 1_000],
+        [60_000, 2_000],
+      ],
+    })
+
+    // the latest sample is 40 s old: the speed is computed until NOW - 10 s
+    expect(getTaskMetric(source, 'bytesRead', { now: NOW })?.speed).toBe(25)
+  })
+
+  it('uses the average speed of an ended source', () => {
+    const source = createSource({ end: 50_000, value: 10_000 })
+
+    expect(getTaskMetric(source, 'bytesRead', { now: NOW })?.speed).toBe(200)
+  })
+
+  it('sums the values, and only the speeds of the running sources', () => {
+    const running = createSource({
+      value: 1_000,
+      total: 2_000,
+      recent: [
+        [90_000, 0],
+        [NOW, 1_000],
+      ],
+    })
+    const ended = createSource({ end: 50_000, value: 5_000, total: 5_000 })
+    // backup VM > read > disks
+    const root = createTask({ tasks: [createTask({ tasks: [running, ended] })] })
+
+    expect(getTaskMetric(root, 'bytesRead', { now: NOW })).toMatchObject({
+      value: 6_000,
+      total: 7_000,
+      speed: 100,
+    })
+  })
+
+  it('uses the average speed over its own duration once the parent has ended', () => {
+    const root = createTask({
+      start: 0,
+      end: 40_000,
+      tasks: [createSource({ start: 10_000, end: 20_000, value: 4_000 })],
+    })
+
+    expect(getTaskMetric(root, 'bytesRead', { now: NOW })?.speed).toBe(100)
+  })
+
+  it('does not count twice the subtasks of a task reporting the metric', () => {
+    const source = createSource({ end: 10_000, value: 1_000 })
+    source.tasks = [createSource({ end: 10_000, value: 1_000 })]
+
+    expect(getTaskMetric(source, 'bytesRead', { now: NOW })?.value).toBe(1_000)
+  })
+
+  it('does not know the total if a source does not report it', () => {
+    const root = createTask({
+      tasks: [createSource({ value: 1, total: 10 }), createSource({ value: 1 })],
+    })
+
+    expect(getTaskMetric(root, 'bytesRead', { now: NOW })?.total).toBeUndefined()
+  })
+
+  it('merges the history and the recent samples, from 0 at the start of the source', () => {
+    const source = createSource({
+      start: 1_000,
+      end: 30_000,
+      value: 300,
+      history: [
+        [2_000, 50],
+        [20_000, 200],
+      ],
+      recent: [
+        [20_000, 200],
+        [30_000, 300],
+      ],
+    })
+
+    expect(getTaskMetric(source, 'bytesRead', { now: NOW })?.samples).toEqual([
+      [1_000, 0],
+      [2_000, 50],
+      [20_000, 200],
+      [30_000, 300],
+    ])
+  })
+})
+
+describe('getSpeedSeries', () => {
+  it('computes the speed between consecutive samples with a small window', () => {
+    expect(
+      getSpeedSeries(
+        [
+          [0, 0],
+          [2_000, 1_000],
+          [4_000, 1_000],
+          [5_000, 2_000],
+        ],
+        { minWindow: 1, maxPoints: Infinity }
+      )
+    ).toEqual([
+      { timestamp: 2_000, value: 500 },
+      { timestamp: 4_000, value: 0 },
+      { timestamp: 5_000, value: 1_000 },
+    ])
+  })
+
+  it('smooths a burst right after the start', () => {
+    // 100 MB prefetched in 10 ms, then 10 MB/s for 60 s
+    const samples: MetricSample[] = [
+      [0, 0],
+      [10, 100e6],
+      ...Array.from({ length: 12 }, (_, i): MetricSample => [(i + 1) * 5_000, 100e6 + (i + 1) * 50e6]),
+    ]
+
+    const speeds = getSpeedSeries(samples).map(({ value }) => value)
+
+    // no 10 GB/s spike: the burst is averaged over at least 10 s
+    expect(Math.max(...speeds)).toBeLessThan(25e6)
+    expect(speeds.at(-1)).toBe(10e6)
+  })
+
+  it('averages the whole series when it is shorter than the window', () => {
+    expect(
+      getSpeedSeries([
+        [0, 0],
+        [1_000, 500],
+        [2_000, 2_000],
+      ])
+    ).toEqual([{ timestamp: 2_000, value: 1_000 }])
+  })
+
+  it('ignores a series without duration', () => {
+    expect(
+      getSpeedSeries([
+        [0, 0],
+        [0, 10],
+      ])
+    ).toEqual([])
+  })
+})
+
+describe('getTaskMetric unit', () => {
+  it('keeps the unit shared by the sources', () => {
+    const root = createTask({ tasks: [createSource({ value: 1 }), createSource({ value: 2 })] })
+
+    expect(getTaskMetric(root, 'bytesRead', { now: NOW })).toMatchObject({ unit: 'bytes' })
+  })
+
+  it('does not know the unit if the sources differ', () => {
+    const root = createTask({ tasks: [createSource({ value: 1 }), createSource({ value: 2, unit: 'blocks' })] })
+
+    expect(getTaskMetric(root, 'bytesRead', { now: NOW })?.unit).toBeUndefined()
+  })
+})

packages/xo-server/src/xo-mixins/vmware/importDisksfromDatastore.mjs

@@ -1,12 +1,13 @@
 import { Task } from '@xen-orchestra/mixins/Tasks.mjs'
 import { QCOW2_CLUSTER_SIZE, VDI_FORMAT_QCOW2, VDI_FORMAT_VHD, VHD_BLOCK_SIZE, VHD_MAX_SIZE } from '@xen-orchestra/xapi'
-import { ReadAhead } from '@xen-orchestra/disk-transform'
+import { MeteredDisk, ReadAhead } from '@xen-orchestra/disk-transform'
 import { toVhdStream } from 'vhd-lib/disk-consumer/index.mjs'
 import { NbdDisk } from '@vates/nbd-client/NbdDisk.mjs'
 import { NbdStdioClient } from '@vates/nbd-client'
 import { createLogger } from '@xen-orchestra/log'
 import { toQcow2Stream } from '@xen-orchestra/qcow2'
 import { TaskProgressHandler } from '@xen-orchestra/backups/_runners/_vmRunners/_TaskProgressHandler.mjs'
+import { SourceReadTask } from '@xen-orchestra/backups/_SourceReadTask.mjs'
 import {
   diskIsAlreadyImported,
   findPreviouslyImportedIndex,
@@ -46,6 +47,8 @@ export async function importStream(
     vmdk = new ReadAhead(vmdk)
 
     vmdk.addProgressHandler(new TaskProgressHandler())
+    // the bytes read from vectura, with their history, in a `read disk` subtask: the same metric as the backups
+    vmdk = new MeteredDisk(vmdk, new SourceReadTask({ name: `read disk ${diskPath}` }))
 
     if (format === VDI_FORMAT_QCOW2) {
       stream = await toQcow2Stream(vmdk, { signal })