From 9b4e1aee3eabacb619ebb127d096cc9e43341add Mon Sep 17 00:00:00 2001 From: Florent BEAUCHAMP Date: Tue, 25 Aug 2026 15:42:12 +0200 Subject: [PATCH] fix(xo-server): task controller leaking memory (#10237) --- .../rest-api/src/tasks/task.controller.mts | 38 +++++-- CHANGELOG.unreleased.md | 2 + packages/xo-server/config.toml | 3 + packages/xo-web/src/common/xo/index.js | 98 +++++++++++++++---- 4 files changed, 114 insertions(+), 27 deletions(-) diff --git a/@xen-orchestra/rest-api/src/tasks/task.controller.mts b/@xen-orchestra/rest-api/src/tasks/task.controller.mts index de0363f402..e3a6ee8679 100644 --- a/@xen-orchestra/rest-api/src/tasks/task.controller.mts +++ b/@xen-orchestra/rest-api/src/tasks/task.controller.mts @@ -39,6 +39,9 @@ import type { CreateActionReturnType } from '../abstract-classes/base-controller import { safeParseComplexMatcher } from '../helpers/utils.helper.mjs' import { RestApi } from '../rest-api/rest-api.mjs' import { AnyPrivilege, hasPrivilegeOn } from '@xen-orchestra/acl' +import { createLogger } from '@xen-orchestra/log' + +const { warn } = createLogger('xo:rest-api:task-controller') @Route('tasks') @Security('*') @@ -94,23 +97,46 @@ export class TaskController extends XoController { } const userFilter = filter === undefined ? undefined : safeParseComplexMatcher(filter).createPredicate() + const mapper = makeObjectMapper(req) const stream = new Transform({ objectMode: true, + // in object mode this is a number of queued events, see `safeWrite()` + highWaterMark: this.restApi.xoApp.config.get('rest-api.maxEventsQueuedPerWatchClient'), transform([event, object], encoding, callback) { - const mapper = makeObjectMapper(req) callback(null, JSON.stringify([event, mapper(object)]) + '\n') }, }) + // The producer is an event emitter shared by the whole application, it + // cannot be slowed down: a client which stops consuming this stream — a + // dropped connection, a sleeping laptop — would otherwise have its events + // queued forever, and each of them retains a whole task log. + // + // Destroy such a client instead, like SSE subscribers are: it reconnects + // and refetches the whole collection, so no state is lost. + const safeWrite = (event: ['update', XoTask] | ['remove', { id: XoTask['id'] }]) => { + if (!stream.write(event)) { + warn('too many events queued for this client, the connection is going to be destroyed', { + queued: stream.writableLength, + }) + req.destroy() + } + } + + const onSigTerm = () => { + req.destroy() + } + stream.on('close', () => { this.restApi.tasks.off('update', update).off('remove', remove) + // this listener would otherwise retain `req` — and through it this + // stream and everything it has buffered — for the whole process life + process.off('SIGTERM', onSigTerm) }) req.on('close', () => { stream.destroy() }) - process.on('SIGTERM', () => { - req.destroy() - }) + process.on('SIGTERM', onSigTerm) const userId = this.restApi.getCurrentUser().id const update = async (task: XoTask) => { @@ -124,7 +150,7 @@ export class TaskController extends XoController { hasPrivilegeOn({ user, userPrivileges, action: 'read', resource: 'task', objects: task }) && (userFilter === undefined || userFilter(task)) ) { - stream.write(['update', task]) + safeWrite(['update', task]) } } const remove = async (task: XoTask) => { @@ -138,7 +164,7 @@ export class TaskController extends XoController { hasPrivilegeOn({ user, userPrivileges, action: 'read', resource: 'task', objects: task }) && (userFilter === undefined || userFilter(task)) ) { - stream.write(['remove', { id: task.id }]) + safeWrite(['remove', { id: task.id }]) } } diff --git a/CHANGELOG.unreleased.md b/CHANGELOG.unreleased.md index 6a232e0562..8cc7b39089 100644 --- a/CHANGELOG.unreleased.md +++ b/CHANGELOG.unreleased.md @@ -42,6 +42,8 @@ - [VIF] Preserve other_config, rate limit, MTU and device when changing a VIF's MAC address (PR [#10284](https://github.com/vatesfr/xen-orchestra/pull/10284)) - **XO 5**: - [VM/Console] Fix the page header and tab navigation disappearing permanently in the console tab (PR [#10007](https://github.com/vatesfr/xen-orchestra/pull/10007)) +- [xo-server] Disconnect user stalling task result and using too much memory (PR [#10237](https://github.com/vatesfr/xen-orchestra/pull/10237)) +- [XO5/tasks] Show a task when a user is disconnected because its task flow is stalled (PR [#10237](https://github.com/vatesfr/xen-orchestra/pull/10237)) ### Packages to release diff --git a/packages/xo-server/config.toml b/packages/xo-server/config.toml index cc56cec804..e6e71c0bd3 100644 --- a/packages/xo-server/config.toml +++ b/packages/xo-server/config.toml @@ -283,6 +283,9 @@ dashboardCacheTimeout = '1 min' # Allow to set a limit of memory per SSE client # before destroying the client. In MiB maxRamAllocatedPerSseClient = 20 +# Allow to set a limit of events queued for a client watching the tasks +# collection (`GET /rest/v0/tasks?watch=true`) before destroying the client +maxEventsQueuedPerWatchClient = 1000 [jsonrpc-api] xvaImportFromUrlTimeout = '6s' diff --git a/packages/xo-web/src/common/xo/index.js b/packages/xo-web/src/common/xo/index.js index 0006e4e810..fe5277811a 100644 --- a/packages/xo-web/src/common/xo/index.js +++ b/packages/xo-web/src/common/xo/index.js @@ -572,6 +572,25 @@ export const subscribeXoTasks = (() => { } }, 100) + // Surfaces a lost tasks stream in the UI: the subscription retries silently, + // which used to leave the list frozen with no indication that it was no + // longer live. This entry is dropped by the `cache.clear()` below as soon as + // the stream is successfully reestablished. + const DISCONNECTED_ID = 'xo:tasks-stream-disconnected' + function notifyDisconnected(error) { + const now = Date.now() + cache.set(DISCONNECTED_ID, { + id: DISCONNECTED_ID, + name: 'XO tasks stream disconnected, retrying…', + properties: { name: 'XO tasks stream disconnected, retrying…' }, + start: now, + end: now, + status: 'failure', + error: error === undefined ? undefined : String(error?.message ?? error), + }) + notify() + } + async function run() { if (abortController !== undefined) { return @@ -584,39 +603,76 @@ export const subscribeXoTasks = (() => { // starts watching collection const resWatch = await fetch(basePath + '&ndjson=true&watch=true', { signal: abortController.signal }) - // fetches existing objects - const response = await fetch(basePath, { signal: abortController.signal }) - const objects = await response.json() - cache.clear() - for (const object of objects) { - cache.set(object.id, object) + const applyEvent = ([event, object]) => { + if (event === 'remove') { + cache.delete(object.id) + } else { + cache.set(object.id, object) + } } - notify() - // handles events - let buf = '' - for await (const chunk of resWatch.body) { - buf += String.fromCharCode(...chunk) + // events received before the existing objects have been fetched cannot + // be applied yet: the fetched collection would override them + let ready = false + const queuedEvents = [] - let i - while ((i = buf.indexOf('\n')) !== -1) { - const line = buf.slice(0, i) - buf = buf.slice(i + 1) - const [event, object] = JSON.parse(line) - if (event === 'remove') { - cache.delete(object.id) - } else { - cache.set(object.id, object) + // this stream must be consumed as soon as possible: events not read are + // buffered by the server, which closes the connection when too many of + // them pile up + const watching = (async () => { + // eslint-disable-next-line n/no-unsupported-features/node-builtins + const decoder = new TextDecoder() + let buf = '' + for await (const chunk of resWatch.body) { + buf += decoder.decode(chunk, { stream: true }) + + let i + while ((i = buf.indexOf('\n')) !== -1) { + const line = buf.slice(0, i) + buf = buf.slice(i + 1) + const event = JSON.parse(line) + if (ready) { + applyEvent(event) + } else { + queuedEvents.push(event) + } + } + if (ready) { + notify() } } + })() + + // fetches existing objects + const fetching = (async () => { + const response = await fetch(basePath, { signal: abortController.signal }) + const objects = await response.json() + cache.clear() + for (const object of objects) { + cache.set(object.id, object) + } + + for (const event of queuedEvents) { + applyEvent(event) + } + queuedEvents.length = 0 + ready = true + notify() - } + })() + + // `Promise.all()` also ensures none of these rejections is unhandled + await Promise.all([fetching, watching]) + + // the iteration ended without an error: the server closed the stream + notifyDisconnected() } catch (error) { if (error === 'abort') { break } console.error('monitor XO tasks', error) + notifyDisconnected(error) } await new Promise(resolve => setTimeout(resolve, 10e3))