test(xo-server/backups): cover the immediate push of the VM backup archives changes

A deletion and the export of a VM announced without waiting for a listing, a repository not in the
collection left alone, and a `refresh()` during a replay or a listing reading again, as many times
as it is called, unless reading again means listing the repository in full.

Author: KAN-vates

packages/xo-server/src/xo-mixins/backups-ng/_vmBackupsCache.test.mjs

@@ -682,6 +682,119 @@ describe('VmBackupsCache collection', () => {
     assert.equal(events[1].payload[0].size, 42)
   })
 
+  it('reads the journal again when refresh() is called while it is being replayed', async t => {
+    const { tick } = mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
+    const repository = new Repository([metadataOf(VM, '20260811T090000')])
+    const source = repository.source
+    const cache = new VmBackupsCache(source, { minRefreshDelay: 60e3 })
+
+    await cache.get(REPOSITORY)
+    const events = recordEvents(cache)
+    tick(60e3)
+
+    const held = holdNextRead(source, 'readJournal')
+    const pending = cache.get(REPOSITORY)
+    await held.read
+
+    // written after the journal has been read, e.g. by the next VM of a backup job
+    const added = metadataOf(OTHER_VM, '20260811T100000')
+    repository.add(added, Date.now())
+    cache.refresh(REPOSITORY.id)
+    held.release()
+
+    assert.deepEqual(filenames(await pending).sort(), [filenameOf(VM, '20260811T090000'), added._filename].sort())
+    assert.equal(repository.nJournalReads, 2)
+    assert.deepEqual(summarize(events), [{ event: 'add', ids: [archiveIdOf(OTHER_VM, '20260811T100000')] }])
+  })
+
+  it('reads the journal again for every refresh() called while it is being read again', async t => {
+    const { tick } = mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
+    const repository = new Repository([metadataOf(VM, '20260811T090000')])
+    const source = repository.source
+    const cache = new VmBackupsCache(source, { minRefreshDelay: 60e3 })
+
+    await cache.get(REPOSITORY)
+    tick(60e3)
+
+    const firstRead = holdNextRead(source, 'readJournal')
+    const pending = cache.get(REPOSITORY)
+    await firstRead.read
+
+    const added = metadataOf(OTHER_VM, '20260811T100000')
+    repository.add(added, Date.now())
+    cache.refresh(REPOSITORY.id)
+    const secondRead = holdNextRead(source, 'readJournal')
+    firstRead.release()
+    await secondRead.read
+
+    // e.g. the VM after it in the same backup job
+    const addedLater = metadataOf(VM, '20260811T100000')
+    repository.add(addedLater, Date.now())
+    cache.refresh(REPOSITORY.id)
+    secondRead.release()
+
+    assert.deepEqual(
+      filenames(await pending).sort(),
+      [filenameOf(VM, '20260811T090000'), added._filename, addedLater._filename].sort()
+    )
+    assert.equal(repository.nJournalReads, 3)
+  })
+
+  it('reads the journal when refresh() is called while the repository is being listed', async t => {
+    mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
+    const repository = new Repository([metadataOf(VM, '20260811T090000')])
+    const source = repository.source
+    const cache = new VmBackupsCache(source, { minRefreshDelay: 60e3 })
+    const events = recordEvents(cache)
+
+    const held = holdNextRead(source, 'listAll')
+    const pending = cache.get(REPOSITORY)
+    await held.read
+
+    const added = metadataOf(OTHER_VM, '20260811T100000')
+    repository.add(added, Date.now())
+    cache.refresh(REPOSITORY.id)
+    held.release()
+
+    assert.deepEqual(filenames(await pending).sort(), [filenameOf(VM, '20260811T090000'), added._filename].sort())
+    assert.equal(repository.nListings, 1)
+    assert.equal(repository.nJournalReads, 1)
+    assert.deepEqual(summarize(events), [
+      { event: 'add', ids: [archiveIdOf(VM, '20260811T090000')] },
+      { event: 'add', ids: [archiveIdOf(OTHER_VM, '20260811T100000')] },
+    ])
+  })
+
+  it('leaves a refresh() called while a repository is listed in full to the next read', async t => {
+    const { tick } = mockTime(t, Date.parse('2026-08-11T10:00:00Z'))
+    const repository = new Repository([metadataOf(VM, '20260811T090000')])
+    repository.cannotReplay = true
+    const source = repository.source
+    const cache = new VmBackupsCache(source, { minRefreshDelay: 60e3 })
+
+    // the cache only finds out on the first replay
+    await cache.get(REPOSITORY)
+    tick(60e3)
+    await cache.get(REPOSITORY)
+    tick(60e3)
+
+    const held = holdNextRead(source, 'listAll')
+    const pending = cache.get(REPOSITORY)
+    await held.read
+
+    const added = metadataOf(OTHER_VM, '20260811T100000')
+    repository.add(added, Date.now())
+    cache.refresh(REPOSITORY.id)
+    held.release()
+
+    assert.deepEqual(filenames(await pending), [filenameOf(VM, '20260811T090000')])
+    assert.equal(repository.nListings, 3)
+
+    // still stale: listed again without waiting for the end of the refresh window
+    assert.ok(filenames(await cache.get(REPOSITORY)).includes(added._filename))
+    assert.equal(repository.nListings, 4)
+  })
+
   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')

packages/xo-server/src/xo-mixins/backups-ng/index.test.mjs

@@ -4,6 +4,7 @@ import EventEmitter from 'node:events'
 import test from 'node:test'
 import TimeoutError from 'promise-toolbox/TimeoutError'
 import { formatJournalEvents, formatVmBackups } from '@xen-orchestra/backups/formatVmBackups.mjs'
+import { makeOnProgress } from '@vates/task/combineEvents'
 
 import BackupNg, { backupsListingRetryDelay } from './index.mjs'
 import { filenameOf, journalEntryPath, metadataOf, VM } from './_vmBackupsFixtures.mjs'
@@ -430,10 +431,12 @@ const createBackupNgWithRepository = repository => {
 const PROXY_ID = 'a-proxy-id'
 
 // serves the same in-memory repository through the RPCs a proxy exposes
-const createBackupNgWithProxiedRepository = (repository, { journal = true, failsToList = false } = {}) => {
+//
+// `backupLogs` is what the proxy streams when it runs a backup job: an async iterable of task events
+const createBackupNgWithProxiedRepository = (repository, { journal = true, failsToList = false, backupLogs } = {}) => {
   const adapter = repository.adapter
   const app = {
-    config: { getDuration: () => LISTING_DEBOUNCE },
+    config: { get: () => ({}), getDuration: () => LISTING_DEBOUNCE },
     getRemoteWithCredentials: async id => {
       assert.equal(id, REMOTE_ID)
       return { id: REMOTE_ID, url: 'file:///media/backup', proxy: PROXY_ID }
@@ -458,10 +461,18 @@ const createBackupNgWithProxiedRepository = (repository, { journal = true, fails
         return adapter.deleteVmBackups(params.filenames)
       }
 
+      if (method === 'backup.run') {
+        return backupLogs()
+      }
+
       throw new Error(`unexpected proxy method ${method}`)
     },
     hooks: new EventEmitter(),
-    registerJobExecutor() {},
+    jobExecutors: {},
+    registerJobExecutor(type, executor) {
+      this.jobExecutors[type] = executor
+    },
+    tasks: { createExternalProgressHandler: makeOnProgress },
   }
   return new BackupNg(app)
 }
@@ -618,17 +629,72 @@ describe('the backup-archive collection', () => {
     assert.deepEqual(events, [{ event: 'add', ids: [idOf('20260811T090000')] }])
   })
 
-  it('announces a backup deleted through the mixin', async () => {
+  it('announces a backup deleted through the mixin, without waiting for a listing', 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])
+    await tick(0)
 
     assert.deepEqual(events, [{ event: 'remove', ids: [idOf('20260811T093000')] }])
+    assert.equal(repository.nListings, 1)
+  })
+
+  it('does not read a repository whose archives are not in the collection after a deletion', async () => {
+    const repository = new Repository([metadataOf(VM, '20260811T090000'), metadataOf(VM, '20260811T093000')])
+    const backupNg = createBackupNgWithRepository(repository)
+    const events = recordEvents(backupNg)
+
+    await backupNg.deleteVmBackupsNg([idOf('20260811T093000')])
+    await tick(0)
+
+    assert.equal(repository.nListings, 0)
+    assert.deepEqual(events, [])
+  })
+
+  it('announces the backup of a VM once it is exported, before the end of the job', async () => {
+    const repository = new Repository([metadataOf(VM, '20260811T090000')])
+    let eventsWhenExported
+    const backupNg = createBackupNgWithProxiedRepository(repository, {
+      // a single VM exported to a single repository
+      async *backupLogs() {
+        yield { type: 'start', id: 'run', timestamp: 0, properties: { name: 'backup run' } }
+        yield {
+          type: 'start',
+          id: 'export',
+          parentId: 'run',
+          timestamp: 0,
+          properties: { id: REMOTE_ID, name: 'export', type: 'remote' },
+        }
+        repository.add(metadataOf(VM, '20260811T100000'))
+        yield { type: 'end', id: 'export', timestamp: 0, status: 'success' }
+
+        await tick(0)
+        eventsWhenExported = events.slice()
+
+        yield { type: 'end', id: 'run', timestamp: 0, status: 'success' }
+      },
+    })
+    const { hooks, jobExecutors } = backupNg._app
+
+    await backupNg.listVmBackupsNg([REMOTE_ID])
+    const events = recordEvents(backupNg)
+    hooks.emit('start')
+
+    await jobExecutors.mirrorBackup({
+      job: { id: 'job', name: 'job', proxy: PROXY_ID, remotes: { id: REMOTE_ID }, settings: { schedule: {} } },
+      jobData: {},
+      jobUpdateFct: async () => {},
+      schedule: { id: 'schedule' },
+    })
+
+    assert.deepEqual(eventsWhenExported, [{ event: 'add', ids: [idOf('20260811T100000')] }])
+    assert.equal(repository.nListings, 1)
+
+    hooks.emit('stop')
   })
 
   it('reads the repositories of the collection on a timer, for the changes made by other processes', async () => {