feat(xo-server/backups,rest-api): stream changes of the VM backup archives over SSE (#10472)

The cache now announces what a listing finds and what a journal replay
changes, as `add`, `update` and `remove` events carrying the archive and its
previous value, on an emitter the app exposes as `app.vmBackupArchives`. What
was last announced for a repository outlives its entry, so a repository read
again from scratch only announces what really changed. A repository which is
removed or disabled goes through the new `forgetVmBackupRepository()`, which
announces all its archives as removed. A listener which throws is logged and no
longer fails the listing.

The REST API adds `backup-archive` to the SSE collections. It is bound to
`app.vmBackupArchives` the way `task` is, and filtered by `backup-archive:read`
through the same path as every other non-XAPI collection.

Nothing polls the repositories: changes are pushed whenever any client lists a
repository. The first listing after a restart sends every archive it already
holds as `add`.

XO-2992
XO-2993
This commit is contained in:
Killian
2026-10-07 11:58:55 +02:00
committed by GitHub
parent 2ee51a3694
commit 0ff6196e72
14 changed files with 913 additions and 42 deletions

View File

@@ -233,6 +233,12 @@ export type XoApp = {
}
}
/**
* the VM backup archives of the backup repositories, as a collection: `add`, `update` and
* `remove` events carrying the archive and its previous value
*/
vmBackupArchives: EventEmitter
tasks: EventEmitter & {
abort(id: XoTask['id'], reason?: string): Promise<void>
clearLogs(): Promise<void>

View File

@@ -0,0 +1,171 @@
import assert from 'node:assert/strict'
import { describe, it } from 'node:test'
import { EventEmitter } from 'node:events'
import { PassThrough } from 'node:stream'
// this module depends on the IoC container, which cannot be initialized on its own
// because of circular imports between the services: load the generated routes first,
// like `index.mts` does
import '../open-api/routes/routes.js'
import type { AnyPrivilege } from '@xen-orchestra/acl'
import type { XoUser, XoVmBackupArchive } from '@vates/types'
import { Subscriber, SubscriberManager, XoListener } from './event.class.mjs'
import { iocContainer } from '../ioc/ioc.mjs'
import { RestApi } from '../rest-api/rest-api.mjs'
const REPOSITORY_ID = '231264c3-af43-4ec0-a3be-394c5b1fdbfc'
const OTHER_REPOSITORY_ID = '1af95910-01b4-4e87-9c2f-d895cafe0776'
function createArchive(backupRepository: string, timestamp: number, props?: Partial<XoVmBackupArchive>) {
return {
id: `${backupRepository}/xo-vm-backups/6ef7c09e-677b-1e6f-0546-7ab30413c61c/20250801T080832Z.json`,
type: 'xo-vm-backup',
backupRepository,
disks: [],
jobId: 'job-1',
mode: 'delta',
scheduleId: 'schedule-1',
size: 1024,
timestamp,
vm: { uuid: '6ef7c09e-677b-1e6f-0546-7ab30413c61c', name_description: '', name_label: 'web', tags: [] },
withMemory: false,
...props,
} as unknown as XoVmBackupArchive
}
/**
* Binds a `RestApi` serving a single user with the given privileges, which is what
* `Listener#getAclEvent` resolves the subscriber against.
*/
function bindRestApi(user: Partial<XoUser>, userPrivileges: AnyPrivilege[] = []) {
const _user = { id: 'user-1', permission: 'none', ...user } as XoUser
if (iocContainer.isBound(RestApi)) {
iocContainer.unbind(RestApi)
}
iocContainer.bind(RestApi).toConstantValue({
xoApp: {
getUser: async () => _user,
getAclV2UserPrivileges: async () => userPrivileges,
},
} as unknown as RestApi)
return _user
}
/** Collects the SSE frames written on the wire, parsed back into `{ event, data }`. */
function createSubscriber(userId: XoUser['id']) {
const connection = new PassThrough()
const frames: { event: string; data: Record<string, unknown> }[] = []
let buffer = ''
connection.on('data', chunk => {
buffer += String(chunk)
let index: number
while ((index = buffer.indexOf('\n\n')) !== -1) {
const [eventLine, dataLine] = buffer.slice(0, index).split('\n')
buffer = buffer.slice(index + 2)
frames.push({ event: eventLine.slice('event: '.length), data: JSON.parse(dataLine.slice('data: '.length)) })
}
})
const subscriber = new Subscriber(connection, new SubscriberManager(), userId)
// drop the `init` frame the subscriber does not send itself
frames.length = 0
return { frames, subscriber }
}
/** Lets the listener broadcast: it handles the data asynchronously, and the stream is async too. */
async function flush() {
for (let i = 0; i < 20; i++) {
await new Promise(resolve => setImmediate(resolve))
}
}
describe('XoListener on the backup-archive collection', () => {
it('broadcasts the archive of each change, tagged with its subscription', async () => {
const user = bindRestApi({ permission: 'admin' })
const { frames, subscriber } = createSubscriber(user.id)
const emitter = new EventEmitter()
const listener = new XoListener('backup-archive', emitter)
listener.addSubscriber(subscriber)
// a backup has been discovered by a listing
const added = createArchive(REPOSITORY_ID, 1)
emitter.emit('add', added, undefined)
// it has been merged: same archive, bigger
const merged = createArchive(REPOSITORY_ID, 1, { size: 2048 })
emitter.emit('update', merged, added)
// it is gone from the repository
emitter.emit('remove', undefined, merged)
await flush()
assert.deepEqual(
frames.map(({ event }) => event),
['add', 'update', 'remove']
)
// every frame carries the archive itself, tagged with the collection it comes from
frames.forEach(({ data }) => {
assert.equal(data.$subscription, 'backup-archive')
assert.equal(data.id, added.id)
assert.equal(data.backupRepository, REPOSITORY_ID)
})
assert.equal(frames[0].data.size, 1024)
assert.equal(frames[1].data.size, 2048)
// a remove carries the value the archive had, since it no longer has one
assert.equal(frames[2].data.size, 2048)
})
it('sends nothing to a user without the backup-archive:read privilege', async () => {
const user = bindRestApi({ permission: 'none' }, [])
const { frames, subscriber } = createSubscriber(user.id)
const emitter = new EventEmitter()
const listener = new XoListener('backup-archive', emitter)
listener.addSubscriber(subscriber)
const archive = createArchive(REPOSITORY_ID, 1)
emitter.emit('add', archive, undefined)
emitter.emit('update', createArchive(REPOSITORY_ID, 1, { size: 2048 }), archive)
emitter.emit('remove', undefined, archive)
await flush()
assert.deepEqual(frames, [])
})
it('sends the archives a user has the backup-archive:read privilege on, and only those', async () => {
const user = bindRestApi({ permission: 'none' }, [
{
id: 'privilege-1',
resource: 'backup-archive',
action: 'read',
selector: `backupRepository:${REPOSITORY_ID}`,
effect: 'allow',
roleId: 'role-1',
},
] as unknown as AnyPrivilege[])
const { frames, subscriber } = createSubscriber(user.id)
const emitter = new EventEmitter()
const listener = new XoListener('backup-archive', emitter)
listener.addSubscriber(subscriber)
const readable = createArchive(REPOSITORY_ID, 1)
const hidden = createArchive(OTHER_REPOSITORY_ID, 2)
emitter.emit('add', readable, undefined)
emitter.emit('add', hidden, undefined)
await flush()
assert.equal(frames.length, 1)
assert.equal(frames[0].event, 'add')
assert.equal(frames[0].data.backupRepository, REPOSITORY_ID)
})
})

View File

@@ -48,6 +48,9 @@ export class EventService {
let eventEmitter: EventEmitter
if (type === 'task') {
eventEmitter = this.#restApi.xoApp.tasks
} else if (type === 'backup-archive') {
// the backup archives are not a registered collection: the backups cache owns their emitter
eventEmitter = this.#restApi.xoApp.vmBackupArchives
} else if (XAPI_TYPES.includes(isMessage ? 'message' : type)) {
// alarm is purely XO-related; it doesn't exist at the XAPI level.
// alarm is a message with parsed values. So, in the case of an alarm listener, it listens for message collection.

View File

@@ -0,0 +1,114 @@
import assert from 'node:assert/strict'
import { describe, it } from 'node:test'
import { EventEmitter } from 'node:events'
import { PassThrough } from 'node:stream'
import type { Response } from 'express'
// this module depends on the IoC container, which cannot be initialized on its own
// because of circular imports between the services: load the generated routes first,
// like `index.mts` does
import '../open-api/routes/routes.js'
import type { XoUser, XoVmBackupArchive } from '@vates/types'
import { EventService } from './event.service.mjs'
import { iocContainer } from '../ioc/ioc.mjs'
import { RestApi } from '../rest-api/rest-api.mjs'
import { AlarmService } from '../alarms/alarm.service.mjs'
import { BackupJobService } from '../backup-jobs/backup-job.service.mjs'
const REPOSITORY_ID = '231264c3-af43-4ec0-a3be-394c5b1fdbfc'
const user = { id: 'user-1', permission: 'admin' } as XoUser
const archive = {
id: `${REPOSITORY_ID}/xo-vm-backups/6ef7c09e-677b-1e6f-0546-7ab30413c61c/20250801T080832Z.json`,
type: 'xo-vm-backup',
backupRepository: REPOSITORY_ID,
size: 1024,
} as unknown as XoVmBackupArchive
/**
* A `RestApi` whose app exposes the backup archives as a collection, bound both as the dependency of
* the service and in the container, which is where `Listener#getAclEvent` resolves the user from.
*/
function createRestApi(vmBackupArchives: EventEmitter) {
const restApi = {
getCurrentUser: () => user,
ioc: {
get: (service: unknown) => {
assert.ok(service === AlarmService || service === BackupJobService)
return {}
},
},
xoApp: {
config: { get: () => 1 },
getUser: async () => user,
getAclV2UserPrivileges: async () => [],
vmBackupArchives,
},
} as unknown as RestApi
if (iocContainer.isBound(RestApi)) {
iocContainer.unbind(RestApi)
}
iocContainer.bind(RestApi).toConstantValue(restApi)
return restApi
}
/** An express response a subscriber can be piped into, which collects the SSE frames it is sent. */
function createResponse() {
const res = new PassThrough() as unknown as Response & PassThrough
res.setHeaders = () => res
const frames: { event: string; data: Record<string, unknown> }[] = []
let buffer = ''
res.on('data', chunk => {
buffer += String(chunk)
let index: number
while ((index = buffer.indexOf('\n\n')) !== -1) {
const [eventLine, dataLine] = buffer.slice(0, index).split('\n')
buffer = buffer.slice(index + 2)
frames.push({ event: eventLine.slice('event: '.length), data: JSON.parse(dataLine.slice('data: '.length)) })
}
})
return { frames, res }
}
async function flush() {
for (let i = 0; i < 20; i++) {
await new Promise(resolve => setImmediate(resolve))
}
}
describe('EventService', () => {
it('streams the changes of the backup archives of the app to a backup-archive subscriber', async t => {
// the ping listener the subscriber always gets would otherwise keep the test process alive
t.mock.timers.enable({ apis: ['setInterval'] })
const vmBackupArchives = new EventEmitter()
const eventService = new EventService(createRestApi(vmBackupArchives))
const { frames, res } = createResponse()
const subscriberId = eventService.createSseSubscriber(res)
eventService.addListenerFor(subscriberId, { type: 'backup-archive' })
vmBackupArchives.emit('add', archive, undefined)
await flush()
// the `init` frame of the connection, then the archive
assert.deepEqual(
frames.map(({ event }) => event),
['init', 'add']
)
assert.equal(frames[1].data.$subscription, 'backup-archive')
assert.equal(frames[1].data.id, archive.id)
// stops listening to the collection once nobody subscribes to it anymore
eventService.removeListenerFor(subscriberId, 'backup-archive')
assert.equal(vmBackupArchives.listenerCount('add'), 0)
})
})

View File

@@ -15,6 +15,7 @@ export type NonXapiListenerType =
| 'server'
| 'backup-repository'
| 'backup-job'
| 'backup-archive'
| 'schedule'
export type XoListenerType = XapiListenerType | NonXapiListenerType

View File

@@ -15,6 +15,7 @@
- [XO6/Tasks] Add link and object resolution to Tasks Overview panel, better text flow for resolved task names (PR [#10265](https://github.com/vatesfr/xen-orchestra/pull/10265))
- [XO5/Settings/IPs] Show an example of the expected IP format when adding IPs to an IP pool (PR [#10522](https://github.com/vatesfr/xen-orchestra/pull/10522))
- [XO6/BRs] Add backup repository create form (PR [#10271](https://github.com/vatesfr/xen-orchestra/pull/10271))
- [REST API] SSE now supports the `backup-archive` collection: backups appearing, being merged or disappearing from a backup repository are pushed to the subscribers, instead of each client listing the repositories again to spot them. XO does not read a repository on its own, so the changes of a repository are pushed as it is listed — a listing by any client is enough — and the archives it already holds arrive as `add` events the first time it is listed after a restart (PR [#10472](https://github.com/vatesfr/xen-orchestra/pull/10472))
- [Plugins/load balancer] Added VM-to-host affinity to force VMs to run on a given set of hosts if possible (PR [#10207](https://github.com/vatesfr/xen-orchestra/pull/10207))
- [Plugins/load balancer] Improved migration decision making (PR [#10207](https://github.com/vatesfr/xen-orchestra/pull/10207))

View File

@@ -7,11 +7,13 @@ const REMOTE_ID = 'a-remote-id'
// instantiates the mixin with the minimum an `Xo` app provides to it
const createRemotes = () => {
const dropped = []
const purged = []
const forgotten = []
const app = {
config: { get: () => ({}) },
forgetVmBackupRepository: id => dropped.push(id),
hooks: { on() {} },
invalidateVmBackupsListing: id => purged.push(id),
}
@@ -31,37 +33,43 @@ const createRemotes = () => {
forget: async () => forgotten.push(REMOTE_ID),
}
return { forgotten, purged, removed, remotes, stored }
return { dropped, forgotten, purged, removed, remotes, stored }
}
describe('updateRemote', () => {
it('drops the handler and purges the VM backups cache', async () => {
const { forgotten, purged, remotes } = createRemotes()
it('drops the handler and the backups of a remote which is disabled', async () => {
const { dropped, forgotten, purged, remotes } = createRemotes()
await remotes.updateRemote(REMOTE_ID, { enabled: false })
assert.deepEqual(purged, [REMOTE_ID])
// it will not be listed again: its backups leave the collection instead of waiting for a
// listing which will never happen
assert.deepEqual(dropped, [REMOTE_ID])
assert.deepEqual(purged, [])
assert.deepEqual(forgotten, [REMOTE_ID])
assert.equal(remotes._handlers[REMOTE_ID], undefined)
})
it('purges the VM backups cache when the remote is re-pointed', async () => {
const { purged, remotes, stored } = createRemotes()
const { dropped, purged, remotes, stored } = createRemotes()
await remotes.updateRemote(REMOTE_ID, { url: 'file:///media/other' })
// it is going to be read again: only what actually changed is announced
assert.deepEqual(purged, [REMOTE_ID])
assert.deepEqual(dropped, [])
assert.equal(stored.url, 'file:///media/other')
})
})
describe('removeRemote', () => {
it('drops the handler and purges the VM backups cache', async () => {
const { forgotten, purged, removed, remotes } = createRemotes()
it('drops the handler and the backups of the remote', async () => {
const { dropped, forgotten, purged, removed, remotes } = createRemotes()
await remotes.removeRemote(REMOTE_ID)
assert.deepEqual(purged, [REMOTE_ID])
assert.deepEqual(dropped, [REMOTE_ID])
assert.deepEqual(purged, [])
assert.deepEqual(forgotten, [REMOTE_ID])
assert.deepEqual(removed, [REMOTE_ID])
assert.equal(remotes._handlers[REMOTE_ID], undefined)

View File

@@ -1,7 +1,9 @@
// @ts-check
import isEqual from 'lodash/isEqual.js'
import { compareTimestamp } from '@xen-orchestra/backups/RemoteAdapter.mjs'
import { createLogger } from '@xen-orchestra/log'
import { EventEmitter } from 'node:events'
import { journalCursorAt } from '@xen-orchestra/backups/_backupJournal.mjs'
/**
@@ -114,6 +116,51 @@ const removeBackup = (backupsByVm, vmUuid, key) => {
}
}
/**
* What a journal event changed in the backups of a repository, as `#emit()` announces it.
*
* @typedef {object} Change
* @property {'add' | 'update' | 'remove'} event
* @property {FormattedBackup} [backup] current value of the archive, `undefined` on `remove`
* @property {FormattedBackup} [previous] value it had before, `undefined` on `add`
*/
/**
* Applies one journal event to the backups of a repository.
*
* @param {BackupsByVm} backupsByVm backups to bring up to date, mutated in place
* @param {JournalEvent} journalEvent
* @returns {Change | undefined} what it changed, `undefined` when it changed nothing
*/
const applyEvent = (backupsByVm, journalEvent) => {
const { vmUuid, filename } = journalEvent
const previous = backupsByVm[vmUuid]?.[filename]
if (journalEvent.event === 'del') {
removeBackup(backupsByVm, vmUuid, filename)
return previous === undefined ? undefined : { event: 'remove', previous }
}
const { backup } = journalEvent
const backups = (backupsByVm[vmUuid] ??= {})
backups[filename] = backup
if (isEqual(previous, backup)) {
return
}
return { event: previous === undefined ? 'add' : 'update', backup, previous }
}
/**
* The archive a backup is served and announced as: the cache keys the backups of a repository by
* the name of their metadata, which is only unique within that repository.
*
* @param {FormattedBackup} backup
* @param {string} repositoryId
* @returns {FormattedBackup}
*/
const archiveOf = (backup, repositoryId) =>
/** @type {FormattedBackup} */ ({ ...backup, id: `${repositoryId}/${backup.id}` })
/**
* Turns the backups of a repository into the shape expected by the API:
* `{ [vmUuid]: <backups sorted by timestamp> }`, restricted to `vmId` when it is given.
@@ -133,7 +180,7 @@ export function serveVmBackups(backupsByVm, remoteId, vmId) {
? []
: Object.values(backups)
// inject the remote id on the backup which is needed for importVmBackupNg()
.map(backup => /** @type {FormattedBackup} */ ({ ...backup, id: `${remoteId}/${backup.id}` }))
.map(backup => archiveOf(backup, remoteId))
.sort(compareTimestamp)
}
return result
@@ -156,8 +203,27 @@ export function serveVmBackups(backupsByVm, remoteId, vmId) {
* `immutable-backups` daemon lifting the immutability of a backup), when the remote is re-pointed,
* reconfigured or moved to another proxy, which the entry detects by itself from what it was built
* from, and when the source turns out not to be able to replay the repository at all.
*
* What it holds is also served as a collection, `archives`: every change is announced as an `add`,
* `update` or `remove` event carrying the archive and its previous value, with the same signature as
* the other collections of the app.
*/
export class VmBackupsCache {
// repository id → the backups last announced for it, which is the object the entry holds while it
// has one: a replay mutates it in place, and only a listing replaces it
//
// it outlives the entry so that a repository which is read from scratch again only announces what
// really changed in the meantime, instead of removing then re-adding everything it holds
//
// therefore only `remove()` releases it: it is the call which says that a repository is not coming
// back, while `delete()` keeps it for the listing which is going to compare itself against it
/** @type {Map<string, BackupsByVm>} */
#announced = new Map()
// a plain emitter rather than the cache itself, so that its consumers can only listen: they must
// not reach `delete()` or `remove()`, which would bypass the listing state their owner keeps
#archives = new EventEmitter()
// repository id → { backupsByVm, cursor, journalConfirmed, options, proxy, refreshedAt, stale, url }
/** @type {Map<string, Entry>} */
#entries = new Map()
@@ -172,6 +238,16 @@ export class VmBackupsCache {
/** @type {Source} */
#source
/**
* The archives the cache holds, as a collection: `add`, `update` and `remove` events carrying the
* archive and its previous value.
*
* @returns {EventEmitter}
*/
get archives() {
return this.#archives
}
/**
* @param {Source} source
* @param {object} [options]
@@ -179,6 +255,9 @@ export class VmBackupsCache {
* repository, in milliseconds
*/
constructor(source, { minRefreshDelay = 0 } = {}) {
// process-wide collection: the number of consumers subscribing to it is not bounded by 10
this.#archives.setMaxListeners(0)
this.#source = source
this.#minRefreshDelay = minRefreshDelay
}
@@ -193,6 +272,14 @@ export class VmBackupsCache {
* arriving after `delete()` must not be served the outcome of an operation which started before it,
* e.g. against a since-reconfigured repository.
*
* Announces nothing: the repository is going to be read again, and the listing which does it only
* announces what actually changed — the archives last announced for it are kept for that
* comparison, and are only released by `remove()`.
*
* A repository which is not going to be listed again must therefore go through `remove()` instead:
* after this call, its archives stay in the collection, and in memory, for as long as the process
* lives.
*
* @param {Repository['id']} repositoryId
* @returns {void}
*/
@@ -203,6 +290,32 @@ export class VmBackupsCache {
this.#pending.delete(repositoryId)
}
/**
* Announces that a repository is gone: every archive it held is removed from the collection, and
* it is forgotten as `delete()` does.
*
* To call when the repository itself is removed or disabled, i.e. when it will not be listed
* again; a repository which is merely re-read must go through `delete()`.
*
* @param {Repository['id']} repositoryId
* @returns {void}
*/
remove(repositoryId) {
const announced = this.#announced.get(repositoryId)
this.delete(repositoryId)
if (announced === undefined) {
return
}
this.#announced.delete(repositoryId)
for (const backups of Object.values(announced)) {
for (const backup of Object.values(backups)) {
this.#emit('remove', repositoryId, undefined, backup)
}
}
}
/**
* Marks a repository as stale: the next listing will replay its journal instead of waiting for the
* end of the current refresh window.
@@ -308,6 +421,65 @@ export class VmBackupsCache {
return Object.keys(backups).length === 0 ? {} : { [vmUuid]: backups }
}
/**
* @param {'add' | 'update' | 'remove'} event
* @param {string} repositoryId
* @param {FormattedBackup} [backup] current value of the archive, `undefined` on `remove`
* @param {FormattedBackup} [previous] value it had before, `undefined` on `add`
* @returns {void}
*/
#emit(event, repositoryId, backup, previous) {
// the listeners run synchronously inside the listing path: a consumer which throws must not fail
// the listing which announced the change, nor the changes announced after it
try {
this.#archives.emit(
event,
backup === undefined ? undefined : archiveOf(backup, repositoryId),
previous === undefined ? undefined : archiveOf(previous, repositoryId)
)
} catch (error) {
warn('a listener failed', { event, repositoryId, error })
}
}
/**
* Announces what changed between the backups a repository has just been listed with and the ones
* last announced for it, and makes them the announced ones.
*
* @param {string} repositoryId
* @param {BackupsByVm} backupsByVm
* @returns {void}
*/
#announce(repositoryId, backupsByVm) {
const announced = this.#announced.get(repositoryId)
this.#announced.set(repositoryId, backupsByVm)
for (const [vmUuid, backups] of Object.entries(backupsByVm)) {
const announcedBackups = announced?.[vmUuid]
for (const [key, backup] of Object.entries(backups)) {
const previous = announcedBackups?.[key]
if (previous === undefined) {
this.#emit('add', repositoryId, backup)
} else if (!isEqual(previous, backup)) {
this.#emit('update', repositoryId, backup, previous)
}
}
}
if (announced === undefined) {
return
}
for (const [vmUuid, backups] of Object.entries(announced)) {
const currentBackups = backupsByVm[vmUuid]
for (const [key, backup] of Object.entries(backups)) {
if (currentBackups?.[key] === undefined) {
this.#emit('remove', repositoryId, undefined, backup)
}
}
}
}
/**
* @param {Repository} repository
* @returns {Promise<BackupsByVm>}
@@ -339,6 +511,13 @@ export class VmBackupsCache {
debug('entry built', { repositoryId: id, nVms: Object.keys(backupsByVm).length })
entry.backupsByVm = backupsByVm
// the entry may have been dropped while it was being built: it is not resurrected, and what it
// read is not announced either
if (this.#entries.get(id) === entry) {
this.#announce(id, backupsByVm)
}
return backupsByVm
} catch (error) {
// don't leave a half-built entry behind, but don't wipe one a newer build has published
@@ -371,18 +550,24 @@ export class VmBackupsCache {
return false
}
// the repository may have been removed while its journal was being read: the entry is still
// brought up to date for the call which started the replay, but what it reads is only announced
// while these backups are still the announced ones
const announced = this.#announced.get(repository.id) === backupsByVm
// the source reduced the events to the last one of each backup, therefore they are independent
// and the order they are applied in does not matter
for (const journalEvent of read.events) {
const { vmUuid, filename } = journalEvent
if (journalEvent.event === 'del') {
removeBackup(backupsByVm, vmUuid, filename)
} else {
const backups = (backupsByVm[vmUuid] ??= {})
backups[filename] = journalEvent.backup
const change = applyEvent(backupsByVm, journalEvent)
if (announced && change !== undefined) {
this.#emit(change.event, repository.id, change.backup, change.previous)
}
}
// the cursor, not the events, is what says whether the journal moved forward: the entries it
// covers may all have resolved to no event at all, e.g. they are of a kind this version does
// not support, and reading them again on every replay would widen the read a bit more every
// minute, until the next rebuild
if (read.cursor !== undefined && read.cursor !== entry.cursor) {
entry.cursor = /** @type {string} */ (read.cursor)
// the journal has now actually been observed to exist: it disappearing on a later replay is

View File

@@ -1,6 +1,7 @@
import assert from 'node:assert/strict'
import { describe, it } from 'node:test'
import { formatVmBackupAt } from '@xen-orchestra/backups/formatVmBackups.mjs'
import { isKnownJournalEvent } from '@xen-orchestra/backups/_backupJournal.mjs'
import { normalize } from '@xen-orchestra/fs/path'
import { filenameOf, journalEntryPath, metadataOf, OTHER_VM, VM } from './_vmBackupsFixtures.mjs'
@@ -20,6 +21,9 @@ class Repository {
nOneVmListings = 0
nJournalReads = 0
// the cursor of each journal read, in order
cursors = []
failWith
// set to make the source report that it cannot replay this repository
@@ -89,6 +93,7 @@ class Repository {
readJournal: async (repository, cursor, { mustExist = false } = {}) => {
this.#mayFail()
this.nJournalReads++
this.cursors.push(cursor)
if (this.cannotReplay) {
return undefined
@@ -106,15 +111,18 @@ class Repository {
cursor = entries[entries.length - 1]._filename
}
// `VmBackupsSource` hands the cache the current value of each backup an event is about, or
// nothing at all when the backup is gone
const events = entries.map(({ event, filename, vmUuid }) => {
const key = normalize(filename)
const metadata = this.metadataByFilename.get(key)
return metadata === undefined
? { event, vmUuid, filename: key }
: { event, vmUuid, filename: key, backup: this.#format(metadata) }
})
// `VmBackupsSource` ignores the events it does not know how to resolve, and hands the cache
// the current value of each backup the others are about, or nothing at all when the backup
// is gone
const events = entries
.filter(({ event }) => isKnownJournalEvent(event))
.map(({ event, filename, vmUuid }) => {
const key = normalize(filename)
const metadata = this.metadataByFilename.get(key)
return metadata === undefined
? { event, vmUuid, filename: key }
: { event, vmUuid, filename: key, backup: this.#format(metadata) }
})
return { events, cursor }
},
}
@@ -414,6 +422,40 @@ describe('VmBackupsCache', () => {
assert.equal(repository.nListings, 2)
})
it('advances the cursor past the entries which resolved to no event', async t => {
const { tick } = mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const repository = new Repository([metadataOf(VM, '20260811T090000')])
const cache = new VmBackupsCache(repository.source, { minRefreshDelay: 60e3 })
await cache.get(REPOSITORY)
// an event of a kind this version does not support, e.g. written by a newer one: it is consumed
// by the source, which reports no event at all
tick(60e3)
repository.pushEvent('an-unsupported-event', filenameOf(VM, '20260811T100000'), VM, Date.now())
await cache.get(REPOSITORY)
tick(60e3)
await cache.get(REPOSITORY)
// the entry is behind the cursor: the next read does not cover it again
assert.equal(repository.cursors[1], repository.journal[0]._filename)
assert.equal(repository.nListings, 1)
})
it('confirms the journal directory even when its entries resolved to no event', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const repository = new Repository([metadataOf(VM, '20260811T090000')])
const cache = new VmBackupsCache(repository.source, { minRefreshDelay: 0 })
await cache.get(REPOSITORY)
repository.pushEvent('an-unsupported-event', filenameOf(VM, '20260811T100000'), VM, Date.now())
await cache.get(REPOSITORY) // reads the entry, confirming the journal directory exists
repository.journalDirMissing = true
await assert.rejects(cache.get(REPOSITORY), /ENOENT/)
})
describe('getOneVm()', () => {
it('lists a single VM instead of the whole repository on a cold entry', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
@@ -479,6 +521,207 @@ describe('VmBackupsCache', () => {
})
})
// the id the cache announces an archive under, which is the one the API serves it as
const archiveIdOf = (vmUuid, name) => `${REPOSITORY.id}/${filenameOf(vmUuid, name)}`
// records what the cache announces, in order
const recordEvents = cache => {
const events = []
for (const name of ['add', 'update', 'remove']) {
cache.archives.on(name, (archive, previous) => events.push({ event: name, archive, previous }))
}
return events
}
// what an event says, without the whole archive
const summarize = events => events.map(({ event, archive, previous }) => ({ event, id: (archive ?? previous).id }))
describe('VmBackupsCache collection', () => {
it('announces the backups a listing discovered', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const metadata = metadataOf(VM, '20260811T090000')
const repository = new Repository([metadata])
const cache = new VmBackupsCache(repository.source)
const events = recordEvents(cache)
await cache.get(REPOSITORY)
assert.deepEqual(events, [
{
event: 'add',
archive: {
...formatVmBackupAt(metadata, metadata._filename, REPOSITORY.id),
id: archiveIdOf(VM, '20260811T090000'),
},
previous: undefined,
},
])
})
it('announces what the journal replays', async t => {
const { tick } = mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const changed = metadataOf(VM, '20260811T090000')
const removed = metadataOf(VM, '20260811T093000')
const repository = new Repository([changed, removed])
const cache = new VmBackupsCache(repository.source, { minRefreshDelay: 60e3 })
await cache.get(REPOSITORY)
const events = recordEvents(cache)
tick(60e3)
const added = metadataOf(OTHER_VM, '20260811T100000')
repository.add(added, Date.now())
repository.change({ ...changed, size: 42 }, Date.now())
repository.del(removed, Date.now())
await cache.get(REPOSITORY)
assert.deepEqual(summarize(events), [
{ event: 'add', id: archiveIdOf(OTHER_VM, '20260811T100000') },
{ event: 'update', id: archiveIdOf(VM, '20260811T090000') },
{ event: 'remove', id: archiveIdOf(VM, '20260811T093000') },
])
assert.equal(events[1].archive.size, 42)
assert.equal(events[1].previous.size, 1)
assert.equal(events[2].previous.id, archiveIdOf(VM, '20260811T093000'))
})
it('announces nothing when a journal event changed nothing', async t => {
const { tick } = mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const metadata = metadataOf(VM, '20260811T090000')
const repository = new Repository([metadata])
const cache = new VmBackupsCache(repository.source, { minRefreshDelay: 60e3 })
await cache.get(REPOSITORY)
const events = recordEvents(cache)
tick(60e3)
// e.g. a cache file rewritten with the same content
repository.change(metadata, Date.now())
await cache.get(REPOSITORY)
assert.equal(repository.nJournalReads, 1)
assert.deepEqual(events, [])
})
it('announces nothing when a repository is listed again with the same backups', async t => {
const { tick } = mockTime(t, Date.parse('2026-08-11T23:59:00Z'))
const repository = new Repository([metadataOf(VM, '20260811T090000')])
const cache = new VmBackupsCache(repository.source, { minRefreshDelay: 60e3 })
await cache.get(REPOSITORY)
const events = recordEvents(cache)
// crossing a UTC day rebuilds the entry from scratch
tick(120e3)
await cache.get(REPOSITORY)
assert.equal(repository.nListings, 2)
assert.deepEqual(events, [])
})
it('only announces what changed when a repository is read from scratch again', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const kept = metadataOf(VM, '20260811T090000')
const removed = metadataOf(VM, '20260811T093000')
const repository = new Repository([kept, removed])
const cache = new VmBackupsCache(repository.source)
await cache.get(REPOSITORY)
const events = recordEvents(cache)
// as `delete()` is called when the repository is reconfigured, or when the user forces a refresh
cache.delete(REPOSITORY.id)
repository.metadataByFilename.delete(removed._filename)
const added = metadataOf(OTHER_VM, '20260811T100000')
repository.metadataByFilename.set(added._filename, added)
await cache.get(REPOSITORY)
assert.equal(repository.nListings, 2)
assert.deepEqual(summarize(events), [
{ event: 'add', id: archiveIdOf(OTHER_VM, '20260811T100000') },
{ event: 'remove', id: archiveIdOf(VM, '20260811T093000') },
])
})
it('remove() announces the removal of every archive of the repository', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const repository = new Repository([metadataOf(VM, '20260811T090000'), metadataOf(OTHER_VM, '20260811T093000')])
const cache = new VmBackupsCache(repository.source)
await cache.get(REPOSITORY)
const events = recordEvents(cache)
cache.remove(REPOSITORY.id)
assert.deepEqual(summarize(events), [
{ event: 'remove', id: archiveIdOf(VM, '20260811T090000') },
{ event: 'remove', id: archiveIdOf(OTHER_VM, '20260811T093000') },
])
// the repository is forgotten, and announced from scratch if it comes back
await cache.get(REPOSITORY)
assert.equal(repository.nListings, 2)
assert.deepEqual(summarize(events).slice(2), [
{ event: 'add', id: archiveIdOf(VM, '20260811T090000') },
{ event: 'add', id: archiveIdOf(OTHER_VM, '20260811T093000') },
])
})
it('does not fail a listing because a listener threw', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const repository = new Repository([metadataOf(VM, '20260811T090000'), metadataOf(OTHER_VM, '20260811T093000')])
const cache = new VmBackupsCache(repository.source)
const events = recordEvents(cache)
cache.archives.on('add', () => {
throw new Error('a broken consumer')
})
const backupsByVm = await cache.get(REPOSITORY)
// the listing itself succeeded, and the archives after the throwing one were still announced
assert.deepEqual(
filenames(backupsByVm).sort(),
[filenameOf(VM, '20260811T090000'), filenameOf(OTHER_VM, '20260811T093000')].sort()
)
assert.deepEqual(summarize(events), [
{ event: 'add', id: archiveIdOf(VM, '20260811T090000') },
{ event: 'add', id: archiveIdOf(OTHER_VM, '20260811T093000') },
])
})
it('removes every archive of a repository even when a listener throws', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const repository = new Repository([metadataOf(VM, '20260811T090000'), metadataOf(OTHER_VM, '20260811T093000')])
const cache = new VmBackupsCache(repository.source)
await cache.get(REPOSITORY)
const events = recordEvents(cache)
cache.archives.on('remove', () => {
throw new Error('a broken consumer')
})
cache.remove(REPOSITORY.id)
assert.deepEqual(summarize(events), [
{ event: 'remove', id: archiveIdOf(VM, '20260811T090000') },
{ event: 'remove', id: archiveIdOf(OTHER_VM, '20260811T093000') },
])
})
it('announces nothing for a repository which is already gone', async t => {
mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
const repository = new Repository([metadataOf(VM, '20260811T090000')])
const cache = new VmBackupsCache(repository.source)
const events = recordEvents(cache)
cache.remove(REPOSITORY.id)
assert.deepEqual(events, [])
})
})
describe('serveVmBackups', () => {
const cached = {
[VM]: {

View File

@@ -9,6 +9,7 @@ import { formatJournalEvents, formatVmBackupAt } from '@xen-orchestra/backups/fo
import { isKnownJournalEvent } from '@xen-orchestra/backups/_backupJournal.mjs'
import { invalidParameters } from 'xo-common/api-errors.js'
/** @typedef {import('@vates/types').XoProxy['id']} ProxyId */
/** @typedef {import('./_vmBackupsCache.mjs').BackupsByVm} BackupsByVm */
/** @typedef {import('./_vmBackupsCache.mjs').Backups} Backups */
/** @typedef {import('./_vmBackupsCache.mjs').FormattedBackup} FormattedBackup */
@@ -17,11 +18,17 @@ import { invalidParameters } from 'xo-common/api-errors.js'
/** @typedef {import('./_vmBackupsCache.mjs').Repository} Repository */
/** @typedef {import('@xen-orchestra/backups/formatVmBackups.mjs').VmBackupMetadata} VmBackupMetadata */
/**
* A repository whose backups are read through a proxy instead of by this process.
*
* @typedef {Repository & { proxy: ProxyId }} ProxiedRepository
*/
/**
* What the source needs from the `Xo` app.
*
* @typedef {object} App
* @property {(proxyId: string, method: string, params: object) => Promise<any>} callProxyMethod
* @property {(proxyId: ProxyId, method: string, params: object) => Promise<any>} callProxyMethod
* @property {(repository: Repository) => object} getBackupsRemoteAdapter a disposable of the
* `RemoteAdapter` of the repository
*/
@@ -41,6 +48,15 @@ const { warn } = createLogger('xo:xo-mixins:backups-ng:vmBackupsSource')
// `Error`: only its code can be matched on
const METHOD_NOT_FOUND_CODE = -32601
/**
* Whether a repository is read through a proxy, as a guard so that the methods which only handle
* that case can require the proxy id the whole `Repository` type only makes optional.
*
* @param {Repository} repository
* @returns {repository is ProxiedRepository}
*/
const isProxied = repository => repository.proxy !== undefined
/**
* What a proxy needs to reach a repository. The credentials travel in the url, protected by TLS and
* by the authentication token of the proxy.
@@ -87,7 +103,7 @@ export class VmBackupsSource {
//
// it only gates the warning: the method is still called every time, so a proxy which gets
// upgraded starts being replayed at once
/** @type {Set<string>} */
/** @type {Set<ProxyId>} */
#proxiesWithoutJournal = new Set()
/**
@@ -114,7 +130,7 @@ export class VmBackupsSource {
* @returns {Promise<BackupsByVm>}
*/
async listAll(repository) {
if (repository.proxy !== undefined) {
if (isProxied(repository)) {
return mapValues(await this.#listVmBackupsOnProxy(repository), keyBackups)
}
@@ -131,7 +147,7 @@ export class VmBackupsSource {
* @returns {Promise<Backups>}
*/
async listOneVm(repository, vmUuid) {
if (repository.proxy !== undefined) {
if (isProxied(repository)) {
return this.#listOneVmOnProxy(repository, vmUuid)
}
@@ -200,7 +216,7 @@ export class VmBackupsSource {
}
/**
* @param {Repository} repository
* @param {ProxiedRepository} repository
* @param {string} vmUuid
* @returns {Promise<Backups>}
*/
@@ -221,14 +237,14 @@ export class VmBackupsSource {
}
/**
* @param {Repository} repository
* @param {ProxiedRepository} repository
* @param {string} [vmId]
* @returns {Promise<Record<string, FormattedBackup[]>>}
*/
async #listVmBackupsOnProxy(repository, vmId) {
const { id } = repository
// set: this is only called for a repository attached to a proxy
const proxy = /** @type {string} */ (repository.proxy)
const { id, proxy } = repository
/** @type {Record<string, Record<string, FormattedBackup[]>>} */
const { [id]: backupsByVm } = await this.#app.callProxyMethod(proxy, 'backup.listVmBackups', {
remotes: { [id]: remoteOf(repository) },
vmId,

View File

@@ -19,9 +19,13 @@ const createSource = adapter =>
const PROXY = 'a-proxy-id'
const PROXIED_REPOSITORY = { ...REPOSITORY, proxy: PROXY }
const methodNotFound = method =>
// what `callProxyMethod()` throws back: the deserialized payload, not an `Error`
({ code: -32601, message: `method not found: ${method}`, data: method })
// what `callProxyMethod()` throws back: the payload of the JSON-RPC error as it was deserialized,
// which is a plain object and not an `Error`
const methodNotFound = method => ({ code: -32601, message: `method not found: ${method}`, data: method })
// a proxy validates the params of a method against a schema which refuses the properties it does not
// declare, and answers the failures of that schema
const invalidParameters = errors => ({ code: 10, message: 'invalid parameters', data: { errors } })
// records every call and answers them from `handlers`
const createProxiedSource = handlers => {
@@ -201,8 +205,11 @@ describe('on a repository attached to a proxy', () => {
const { calls, source } = createProxiedSource({
'backup.listVmBackups': ({ vmId }) => {
if (vmId !== undefined) {
// what a proxy older than the `vmId` parameter answers
throw Object.assign(new Error('invalid parameters'), { code: 10 })
// what a proxy older than the `vmId` parameter answers: its schema does not declare it,
// so it is rejected as an unknown property
throw invalidParameters([
{ instancePath: '', keyword: 'additionalProperties', params: { additionalProperty: 'vmId' } },
])
}
return { [REPOSITORY.id]: { [VM]: [backup] } }
},

View File

@@ -100,6 +100,16 @@ export default class BackupNg {
return this._runningRestores
}
/**
* the VM backup archives of the backup repositories, as a collection: `add`, `update` and
* `remove` events carrying the archive and its previous value
*
* @returns {import('node:events').EventEmitter}
*/
get vmBackupArchives() {
return this.#vmBackupsCache.archives
}
constructor(app) {
this._app = app
this._runningRestores = new Set()
@@ -411,6 +421,10 @@ export default class BackupNg {
app.registerJobExecutor('backup', executor)
app.registerJobExecutor('mirrorBackup', executor)
})
app.hooks.on('stop', () => {
this.#vmBackupsCache.archives.removeAllListeners()
})
}
async createBackupNgJob(type, props, schedules) {
@@ -865,8 +879,11 @@ export default class BackupNg {
* forgets everything known about a backup repository: its backups are read from scratch on the
* next listing, instead of being brought up to date from its journal
*
* its archives stay in the collection until that listing says what changed, so a repository which
* will not be listed again must go through `forgetVmBackupRepository()` instead, otherwise they
* are never removed from it
*
* public because it is also called by the remotes mixin when a backup repository is updated
* or removed
*
* @param {XoBackupRepository['id']} remoteId
*/
@@ -874,4 +891,17 @@ export default class BackupNg {
this.#vmBackupsCache.delete(remoteId)
this.#resetVmBackupsListingState(remoteId)
}
/**
* forgets a backup repository which will not be listed again — it has been removed or disabled —
* and announces that its archives are gone with it
*
* public because it is called by the remotes mixin
*
* @param {XoBackupRepository['id']} remoteId
*/
forgetVmBackupRepository(remoteId) {
this.#vmBackupsCache.remove(remoteId)
this.#resetVmBackupsListingState(remoteId)
}
}

View File

@@ -469,6 +469,14 @@ describe('deleteVmBackupsNg', () => {
})
describe('on a repository attached to a proxy', () => {
beforeEach(() => {
mock.timers.enable({ apis: ['setTimeout', 'Date'] })
})
afterEach(() => {
mock.timers.reset()
})
it('makes a deletion visible at once, without listing the repository again', async () => {
const repository = new Repository([metadataOf(VM, '20260811T090000'), metadataOf(VM, '20260811T093000')])
const backupNg = createBackupNgWithProxiedRepository(repository)
@@ -525,10 +533,85 @@ describe('invalidateVmBackupsListing() on a cached repository', () => {
await backupNg.listVmBackupsNg([REMOTE_ID])
assert.equal(repository.nListings, 1)
// as the `remotes` mixin does when the repository is gone or has been reconfigured
// as the `remotes` mixin does when the repository has been reconfigured
backupNg.invalidateVmBackupsListing(REMOTE_ID)
await backupNg.listVmBackupsNg([REMOTE_ID])
assert.equal(repository.nListings, 2)
})
})
describe('vmBackupArchives', () => {
beforeEach(() => {
mock.timers.enable({ apis: ['setTimeout', 'Date'] })
})
afterEach(() => {
mock.timers.reset()
})
// records what the mixin announces, in order
const recordEvents = backupNg => {
const events = []
for (const name of ['add', 'update', 'remove']) {
backupNg.vmBackupArchives.on(name, (archive, previous) =>
events.push({ event: name, id: (archive ?? previous).id })
)
}
return events
}
it('announces the archives a listing discovered', async () => {
const repository = new Repository([metadataOf(VM, '20260811T090000')])
const backupNg = createBackupNgWithRepository(repository)
const events = recordEvents(backupNg)
await backupNg.listVmBackupsNg([REMOTE_ID])
assert.deepEqual(events, [{ event: 'add', id: idOf('20260811T090000') }])
})
it('announces a backup deleted through the mixin', async () => {
const repository = new Repository([metadataOf(VM, '20260811T090000'), metadataOf(VM, '20260811T093000')])
const backupNg = createBackupNgWithRepository(repository)
await backupNg.listVmBackupsNg([REMOTE_ID])
const events = recordEvents(backupNg)
await backupNg.deleteVmBackupsNg([idOf('20260811T093000')])
await backupNg.listVmBackupsNg([REMOTE_ID])
assert.deepEqual(events, [{ event: 'remove', id: idOf('20260811T093000') }])
})
it('announces the removal of every archive of a repository which is gone', async () => {
const repository = new Repository([metadataOf(VM, '20260811T090000'), metadataOf(VM, '20260811T093000')])
const backupNg = createBackupNgWithRepository(repository)
await backupNg.listVmBackupsNg([REMOTE_ID])
const events = recordEvents(backupNg)
// as the `remotes` mixin does when the backup repository is removed or disabled
backupNg.forgetVmBackupRepository(REMOTE_ID)
assert.deepEqual(events, [
{ event: 'remove', id: idOf('20260811T090000') },
{ event: 'remove', id: idOf('20260811T093000') },
])
})
it('announces nothing when a repository is merely invalidated and read again', async () => {
const repository = new Repository([metadataOf(VM, '20260811T090000')])
const backupNg = createBackupNgWithRepository(repository)
await backupNg.listVmBackupsNg([REMOTE_ID])
const events = recordEvents(backupNg)
// as `_forceRefresh` does
backupNg.invalidateVmBackupsListing(REMOTE_ID)
await backupNg.listVmBackupsNg([REMOTE_ID])
assert.equal(repository.nListings, 2)
assert.deepEqual(events, [])
})
})

View File

@@ -439,9 +439,12 @@ export default class {
}
this._cancelRemoteInfoRetry(id)
this._app.invalidateVmBackupsListing(id)
if (enabled === false) {
// it will not be listed again: its backups are no longer part of the collection
this._app.forgetVmBackupRepository(id)
delete this._remotesInfo[id]
} else {
this._app.invalidateVmBackupsListing(id)
}
return this._updateRemote(id, {
@@ -473,7 +476,7 @@ export default class {
async removeRemote(id) {
this._cancelRemoteInfoRetry(id)
this._app.invalidateVmBackupsListing(id)
this._app.forgetVmBackupRepository(id)
delete this._remotesInfo[id]
const handlers = this._handlers