Administrator/Xen-Orchestra-xen-orchestra-MIRROR
1
0
Fork 1
mirror of https://github.com/vatesfr/xen-orchestra.git synced 2026-09-10 22:14:48 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix(xo-server): task controller leaking memory (#10237)

Browse Source
This commit is contained in:
Florent BEAUCHAMP
2026-08-25 15:42:12 +02:00
committed by GitHub
parent ab7b2fc7a9
commit 9b4e1aee3e
4 changed files with 114 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
@xen-orchestra/rest-api/src/tasks/task.controller.mts
View File

@@ -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<XoTask> {
}
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<number>('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<XoTask> {
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<XoTask> {
hasPrivilegeOn({ user, userPrivileges, action: 'read', resource: 'task', objects: task }) &&
(userFilter === undefined || userFilter(task))
) {
stream.write(['remove', { id: task.id }])
safeWrite(['remove', { id: task.id }])
}
}

2
CHANGELOG.unreleased.md
View File

@@ -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

3
packages/xo-server/config.toml
View File

@@ -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'

98
packages/xo-web/src/common/xo/index.js
View File

@@ -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))