diff --git a/packages/@n8n/api-types/src/index.ts b/packages/@n8n/api-types/src/index.ts index 4cfd61dc098..51adcccc525 100644 --- a/packages/@n8n/api-types/src/index.ts +++ b/packages/@n8n/api-types/src/index.ts @@ -3,6 +3,7 @@ export type * from './datetime'; export * from './dto'; export type * from './push'; export type * from './scaling'; +export type * from './process-internals'; export * from './frontend-settings'; export type * from './user'; export type * from './consent'; diff --git a/packages/@n8n/api-types/src/process-internals.ts b/packages/@n8n/api-types/src/process-internals.ts new file mode 100644 index 00000000000..04f64dac0b5 --- /dev/null +++ b/packages/@n8n/api-types/src/process-internals.ts @@ -0,0 +1,30 @@ +/** + * Counts of a process's in-memory state, served by the test-only + * `GET /rest/e2e/internals` route on mains, workers, and webhook processes. + * + * Counts only, never contents. Tests diff two readings to find state that + * grows without bound, for example a map that never evicts entries. + */ +export interface ProcessInternals { + /** Bumped when a field is renamed or removed. Adding a field is not a breaking change. */ + version: 1; + instanceType: 'main' | 'worker' | 'webhook' | 'engine'; + hostId: string; + isLeader: boolean; + /** `process.memoryUsage()`, in bytes. */ + memory: { + rss: number; + heapTotal: number; + heapUsed: number; + external: number; + arrayBuffers: number; + }; + /** Live libuv resources by type, e.g. `{ Timeout: 12, TCPSocketWrap: 4 }`. */ + resources: Record; + /** + * Sizes of known in-memory collections, keyed `.`, + * e.g. `scaling.jobResults`. A key is absent when its owner does not run + * in this process. + */ + collections: Record; +} diff --git a/packages/cli/src/__tests__/active-executions.test.ts b/packages/cli/src/__tests__/active-executions.test.ts index c12772f66fd..dccf0c05f81 100644 --- a/packages/cli/src/__tests__/active-executions.test.ts +++ b/packages/cli/src/__tests__/active-executions.test.ts @@ -118,6 +118,19 @@ describe('ActiveExecutions', () => { expect(executionPersistence.updateExistingExecution).toHaveBeenCalledTimes(0); }); + test('should report execution and response mode counts until execution cleanup', async () => { + expect(activeExecutions.getDiagnosticCounts()).toEqual({ executions: 0, responseModes: 0 }); + + const executionId = await activeExecutions.add(executionData); + activeExecutions.setResponseMode(executionId, 'lastNode'); + expect(activeExecutions.getDiagnosticCounts()).toEqual({ executions: 1, responseModes: 1 }); + + const completed = activeExecutions.getPostExecutePromise(executionId); + activeExecutions.finalizeExecution(executionId); + await completed; + expect(activeExecutions.getDiagnosticCounts()).toEqual({ executions: 0, responseModes: 0 }); + }); + test('Should update execution if add is called with execution ID', async () => { const executionId = await activeExecutions.add(executionData, { executionId: FAKE_SECOND_EXECUTION_ID, diff --git a/packages/cli/src/__tests__/wait-tracker.test.ts b/packages/cli/src/__tests__/wait-tracker.test.ts index 9c2533f9dd0..205afc0cc4f 100644 --- a/packages/cli/src/__tests__/wait-tracker.test.ts +++ b/packages/cli/src/__tests__/wait-tracker.test.ts @@ -89,6 +89,17 @@ describe('WaitTracker', () => { vi.clearAllMocks(); }); + it('reports waiting executions until their timers are stopped', async () => { + expect(waitTracker.getDiagnosticCounts()).toEqual({ waitingExecutions: 0 }); + executionRepository.getWaitingExecutions.mockResolvedValue([execution]); + + await waitTracker.getWaitingExecutions(); + expect(waitTracker.getDiagnosticCounts()).toEqual({ waitingExecutions: 1 }); + + waitTracker.stopExecution(execution.id); + expect(waitTracker.getDiagnosticCounts()).toEqual({ waitingExecutions: 0 }); + }); + describe('init()', () => { it('should query DB for waiting executions if leader', () => { executionRepository.getWaitingExecutions.mockResolvedValue([execution]); diff --git a/packages/cli/src/active-executions.ts b/packages/cli/src/active-executions.ts index 6ba924d8ef4..8006e76f55e 100644 --- a/packages/cli/src/active-executions.ts +++ b/packages/cli/src/active-executions.ts @@ -334,6 +334,14 @@ export class ActiveExecutions { ); } + /** Sizes of the in-memory collections, for diagnostics and tests. */ + getDiagnosticCounts() { + return { + executions: Object.keys(this.activeExecutions).length, + responseModes: this.responseModes.size, + }; + } + /** * @param writeDeadlineMs - How long to wait for the cancelled status to be recorded. * Pass what the caller's own shutdown window can still afford. diff --git a/packages/cli/src/commands/__tests__/worker.test.ts b/packages/cli/src/commands/__tests__/worker.test.ts index f96216d2fb8..60c4c42590b 100644 --- a/packages/cli/src/commands/__tests__/worker.test.ts +++ b/packages/cli/src/commands/__tests__/worker.test.ts @@ -41,6 +41,14 @@ import { Worker } from '../worker'; vi.mock('@/crash-journal'); +const e2eFlags = vi.hoisted(() => ({ inE2ETests: false })); +vi.mock('@/constants', async (importOriginal) => ({ + ...(await importOriginal()), + get inE2ETests() { + return e2eFlags.inE2ETests; + }, +})); + const dbConnection = mockInstance(DbConnection); dbConnection.init.mockResolvedValue(undefined); dbConnection.migrate.mockResolvedValue(undefined); @@ -78,6 +86,7 @@ const systemTaskRunner = mockInstance(SystemTaskRunner); describe('Worker', () => { beforeEach(() => { vi.clearAllMocks(); + e2eFlags.inE2ETests = false; }); /** Worker with the init steps that go beyond `super.init()` stubbed, as in start.test.ts. */ @@ -322,6 +331,19 @@ describe('Worker', () => { expect(mockScalingService.setupWorker).toHaveBeenCalledWith(10); }); + it('should initialize WorkerServer for E2E diagnostics when no other endpoints are enabled', async () => { + e2eFlags.inE2ETests = true; + + await createWorkerForRun().run(); + + expect(mockWorkerServer.init).toHaveBeenCalledWith({ + health: false, + overwrites: false, + metrics: false, + }); + expect(mockWorkerServer.markAsReady).toHaveBeenCalled(); + }); + it('should start the system tasks once the server is up', async () => { Container.get(GlobalConfig).queue.health.active = true; diff --git a/packages/cli/src/commands/worker.ts b/packages/cli/src/commands/worker.ts index e2e83e219e9..81e8daddea1 100644 --- a/packages/cli/src/commands/worker.ts +++ b/packages/cli/src/commands/worker.ts @@ -6,7 +6,7 @@ import { BinaryDataConfig } from 'n8n-core'; import { z } from 'zod'; import { ActiveExecutions } from '@/active-executions'; -import { N8N_VERSION } from '@/constants'; +import { inE2ETests, N8N_VERSION } from '@/constants'; import { CredentialsOverwrites } from '@/credentials-overwrites'; import { DeprecationService } from '@/deprecation/deprecation.service'; import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-message-generic'; @@ -220,7 +220,7 @@ export class Worker extends BaseCommand> { }; let workerServer: WorkerServer | undefined; - if (Object.values(endpointsConfig).some((e) => e)) { + if (inE2ETests || Object.values(endpointsConfig).some((e) => e)) { const { WorkerServer } = await import('@/scaling/worker-server.js'); workerServer = Container.get(WorkerServer); await workerServer.init(endpointsConfig); diff --git a/packages/cli/src/controllers/e2e.controller.ts b/packages/cli/src/controllers/e2e.controller.ts index faa3c40c31b..0b0c19caa4d 100644 --- a/packages/cli/src/controllers/e2e.controller.ts +++ b/packages/cli/src/controllers/e2e.controller.ts @@ -20,8 +20,6 @@ import { Request } from 'express'; import type nodeFs from 'node:fs'; import type { Profiler } from 'node:inspector'; import type * as nodeInspectorPromises from 'node:inspector/promises'; -import type nodePath from 'node:path'; -import type nodeV8 from 'node:v8'; import { v4 as uuid } from 'uuid'; import { ActiveWorkflowManager } from '@/active-workflow-manager'; @@ -34,7 +32,9 @@ import { Push } from '@/push'; import { WorkflowScheduledJobOwner } from '@/scheduling/workflow-scheduled-job-owner'; import { CacheService } from '@n8n/backend-services'; import { FrontendService } from '@/services/frontend.service'; +import { HeapDiagnosticsService } from '@/services/heap-diagnostics.service'; import { PasswordUtility } from '@/services/password.utility'; +import { ProcessInternalsService } from '@/services/process-internals.service'; import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service'; if (!inE2ETests) { @@ -208,6 +208,8 @@ export class E2EController { private readonly workflowScheduledJobOwner: WorkflowScheduledJobOwner, private readonly pollerStateRepository: PollerStateRepository, private readonly workflowStaticDataService: WorkflowStaticDataService, + private readonly processInternalsService: ProcessInternalsService, + private readonly heapDiagnosticsService: HeapDiagnosticsService, ) { license.isLicensed = (feature: BooleanLicenseFeature) => this.enabledFeatures[feature] ?? false; @@ -374,74 +376,25 @@ export class E2EController { */ @Post('/gc', { skipAuth: true }) triggerGarbageCollection() { - if (typeof global.gc === 'function') { - // Call GC twice to allow for more reclaimation - global.gc(); - global.gc(); - return { success: true, message: 'Garbage collection triggered' }; - } - return { - success: false, - message: 'Garbage collection not available. Ensure Node.js is started with --expose-gc flag.', - }; + return this.heapDiagnosticsService.collectGarbage(); } /** - * Write a V8 heap snapshot for memory leak analysis. - * Triggers GC first for a cleaner snapshot. - * Returns the file path inside the container — retrieve via `docker cp` or keepalive mode. + * Write a V8 heap snapshot for memory leak analysis. Triggers GC first. + * Returns the file name to pass to the download route. */ @Post('/heap-snapshot', { skipAuth: true }) takeHeapSnapshot() { - const v8 = require('node:v8') as typeof nodeV8; - const fs = require('node:fs') as typeof nodeFs; - - if (typeof global.gc === 'function') { - global.gc(); - global.gc(); - } - - const filePath = v8.writeHeapSnapshot(); - if (!filePath) { - return { success: false, message: 'Failed to write heap snapshot' }; - } - - const path = require('node:path') as typeof nodePath; - const stats = fs.statSync(filePath); - const filename = path.basename(filePath); - this.heapSnapshotPaths.set(filename, filePath); - - return { - success: true, - filePath: filename, - sizeBytes: stats.size, - sizeMB: Math.round(stats.size / 1024 / 1024), - }; + return this.heapDiagnosticsService.writeHeapSnapshot(); } - private heapSnapshotPaths = new Map(); - /** * Download a heap snapshot file as a stream. * The response-helper pipes Readable streams directly to the response. */ @Get('/heap-snapshot/:filename', { skipAuth: true }) downloadHeapSnapshot(req: Request) { - const fs = require('node:fs') as typeof nodeFs; - const path = require('node:path') as typeof nodePath; - - const filename = path.basename(req.params.filename); - if (!filename.endsWith('.heapsnapshot')) { - throw new Error('Invalid file type'); - } - - // Look up the full path stored during POST, or try cwd - const filePath = this.heapSnapshotPaths.get(filename) ?? path.resolve(filename); - if (!fs.existsSync(filePath)) { - throw new Error(`Snapshot not found: ${filename} (tried ${filePath})`); - } - - return fs.createReadStream(filePath); + return this.heapDiagnosticsService.openHeapSnapshot(String(req.params.filename)); } // --- Per-spec backend V8 coverage (DEVP-370) ----------------------------- @@ -527,6 +480,15 @@ export class E2EController { return { success: true, result: E2EController.deltaCoverage(result, this.coverageBaseline) }; } + /** + * Return counts of this process's in-memory state (collections, libuv + * resources, memory). Tests diff two readings to find unbounded growth. + */ + @Get('/internals', { skipAuth: true }) + getInternals() { + return this.processInternalsService.collect(); + } + /** * Return a parsed breakdown of process RSS from /proc/self/smaps. * Groups memory mappings by pathname and returns both a rollup summary diff --git a/packages/cli/src/push/__tests__/sse.push.test.ts b/packages/cli/src/push/__tests__/sse.push.test.ts index 3637120506f..e7a391d7ef1 100644 --- a/packages/cli/src/push/__tests__/sse.push.test.ts +++ b/packages/cli/src/push/__tests__/sse.push.test.ts @@ -70,6 +70,14 @@ describe('SSEPush', () => { }); }); + it('reports the current connection count after a client disconnects', () => { + expect(ssePush.getDiagnosticCounts()).toEqual({ connections: 2 }); + + req.emit('close'); + + expect(ssePush.getDiagnosticCounts()).toEqual({ connections: 1 }); + }); + describe('does not remove replaced connection when old connection closes', () => { test.each([ ['end', 'req'], diff --git a/packages/cli/src/push/abstract.push.ts b/packages/cli/src/push/abstract.push.ts index e5847a7104c..dbb06be93e0 100644 --- a/packages/cli/src/push/abstract.push.ts +++ b/packages/cli/src/push/abstract.push.ts @@ -127,6 +127,11 @@ export abstract class AbstractPush extends TypedEmitter { }); }); + describe('getDiagnosticCounts', () => { + it('should report stored job results, queue listeners, and running jobs', async () => { + const activeExecutions = mock(); + activeExecutions.has.mockReturnValue(true); + const outcomeTracker = new JobOutcomeTracker(mockLogger(), activeExecutions, mock(), mock()); + scalingService = new ScalingService( + mockLogger(), + mock(), + activeExecutions, + jobProcessor, + globalConfig, + mock(), + mock(), + instanceSettings, + mock(), + webhookResponseRelay, + executionCrashService, + outcomeTracker, + ); + await scalingService.setupQueue(); + queue.eventNames.mockReturnValue(['global:progress', 'global:completed']); + queue.listenerCount.mockImplementation((event) => (event === 'global:completed' ? 2 : 1)); + jobProcessor.getRunningJobIds.mockReturnValue(['job-1']); + + const messageHandler = queue.on.mock.calls.find( + ([event]) => (event as string) === 'global:progress', + )?.[1] as (jobId: JobId, msg: unknown) => void; + messageHandler('job-789', { + kind: 'job-finished', + version: 2, + executionId: 'exec-123', + workerId: 'worker-456', + success: true, + status: 'success', + startedAt: '2026-07-25T11:59:00.000Z', + stoppedAt: '2026-07-25T11:59:30.000Z', + }); + + expect(scalingService.getDiagnosticCounts()).toEqual({ + jobResults: 1, + queueListeners: 3, + runningJobs: 1, + }); + + scalingService.popJobResult('exec-123'); + + expect(scalingService.getDiagnosticCounts().jobResults).toBe(0); + }); + }); + describe('recoverFromQueue', () => { it('should mark running executions as crashed if they are missing from the queue and queue is empty', async () => { await scalingService.setupQueue(); diff --git a/packages/cli/src/scaling/__tests__/worker-server.test.ts b/packages/cli/src/scaling/__tests__/worker-server.test.ts index 69feaacd507..4e6c44fc810 100644 --- a/packages/cli/src/scaling/__tests__/worker-server.test.ts +++ b/packages/cli/src/scaling/__tests__/worker-server.test.ts @@ -17,9 +17,19 @@ import type { RedisClientService } from '@n8n/backend-services'; import { WorkerServer } from '../worker-server'; const app = mock(); +const e2eFlags = vi.hoisted(() => ({ inE2ETests: false })); vi.mock('node:http'); vi.mock('express', () => ({ __esModule: true, default: () => app })); +vi.mock('@/services/e2e-diagnostics.router', () => ({ + createE2EDiagnosticsRouter: vi.fn(() => vi.fn()), +})); +vi.mock('@/constants', async (importOriginal) => ({ + ...(await importOriginal()), + get inE2ETests() { + return e2eFlags.inE2ETests; + }, +})); const addressInUseError = () => { const error: NodeJS.ErrnoException = new Error('Port already in use'); @@ -51,6 +61,7 @@ describe('WorkerServer', () => { ); beforeEach(() => { + e2eFlags.inE2ETests = false; globalConfig = mock({ path: '/', queue: { @@ -190,6 +201,23 @@ describe('WorkerServer', () => { ).rejects.toThrowError(AssertionError); }); + it('should mount E2E diagnostics when no other endpoints are enabled', async () => { + e2eFlags.inE2ETests = true; + globalConfig.endpoints.rest = 'rest'; + const server = mock(); + vi.spyOn(http, 'createServer').mockReturnValue(server); + server.listen.mockImplementation((...args: unknown[]) => { + const callback = args.find((arg) => typeof arg === 'function'); + if (callback) callback(); + return server; + }); + + await newWorkerServer().init({ health: false, overwrites: false, metrics: false }); + + expect(app.use).toHaveBeenCalledWith('/rest/e2e', expect.any(Function)); + expect(server.listen).toHaveBeenCalled(); + }); + it('should call `worker.ready` external hook', async () => { const server = mock(); vi.spyOn(http, 'createServer').mockReturnValue(server); diff --git a/packages/cli/src/scaling/job-outcome-tracker.ts b/packages/cli/src/scaling/job-outcome-tracker.ts index 23ba1f1f22f..0cd13d3c18b 100644 --- a/packages/cli/src/scaling/job-outcome-tracker.ts +++ b/packages/cli/src/scaling/job-outcome-tracker.ts @@ -77,6 +77,10 @@ export class JobOutcomeTracker { this.logger = this.logger.scoped('scaling'); } + getDiagnosticCounts() { + return { jobResults: this.results.size }; + } + /** * Record that the worker reported the job as finished, with the result when * the worker sent one. Bull broadcasts the message to every main and webhook diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index 6652ad8d0d0..3adfb484fc3 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/packages/cli/src/scaling/scaling.service.ts @@ -448,6 +448,25 @@ export class ScalingService { return this.jobProcessor.getRunningJobIds().length; } + /** + * Sizes of the in-memory collections, for diagnostics and tests. + * `queueListeners` counts Bull event listeners across this process's queues. + */ + getDiagnosticCounts() { + let queueListeners = 0; + for (const queue of this.queueByName.values()) { + for (const eventName of queue.eventNames()) { + queueListeners += queue.listenerCount(eventName); + } + } + + return { + ...this.jobOutcomeTracker.getDiagnosticCounts(), + queueListeners, + runningJobs: this.getRunningJobsCount(), + }; + } + // #endregion // #region Listeners diff --git a/packages/cli/src/scaling/worker-server.ts b/packages/cli/src/scaling/worker-server.ts index 41e339e319f..16d6d7202bc 100644 --- a/packages/cli/src/scaling/worker-server.ts +++ b/packages/cli/src/scaling/worker-server.ts @@ -9,6 +9,7 @@ import { strict as assert } from 'node:assert'; import http from 'node:http'; import type { Server } from 'node:http'; +import { inE2ETests } from '@/constants'; import { CredentialsOverwrites } from '@/credentials-overwrites'; import { CredentialsOverwritesAlreadySetError } from '@/errors/credentials-overwrites-already-set.error'; import { NonJsonBodyError } from '@/errors/non-json-body.error'; @@ -89,11 +90,11 @@ export class WorkerServer { } async init(endpointsConfig: WorkerServerEndpointsConfig) { - assert(Object.values(endpointsConfig).some((e) => e)); + assert(inE2ETests || Object.values(endpointsConfig).some((e) => e)); this.endpointsConfig = endpointsConfig; - this.mountEndpoints(); + await this.mountEndpoints(); this.logger.debug('Worker server initialized', { endpoints: Object.keys(this.endpointsConfig), @@ -106,7 +107,7 @@ export class WorkerServer { this.logger.info(`\nn8n worker server listening on port ${this.port}`); } - private mountEndpoints() { + private async mountEndpoints() { const { health, overwrites, metrics } = this.endpointsConfig; if (health) { @@ -139,6 +140,12 @@ export class WorkerServer { if (metrics) { this.prometheusMetricsService.init(this.app); } + + // Workers have no REST server, so the test-only diagnostics routes live here. + if (inE2ETests) { + const { createE2EDiagnosticsRouter } = await import('@/services/e2e-diagnostics.router.js'); + this.app.use(`/${this.globalConfig.endpoints.rest}/e2e`, createE2EDiagnosticsRouter()); + } } private async readiness(_req: express.Request, res: express.Response) { diff --git a/packages/cli/src/services/__tests__/e2e-diagnostics.router.test.ts b/packages/cli/src/services/__tests__/e2e-diagnostics.router.test.ts new file mode 100644 index 00000000000..0fa3ac8a465 --- /dev/null +++ b/packages/cli/src/services/__tests__/e2e-diagnostics.router.test.ts @@ -0,0 +1,74 @@ +import { mockInstance } from '@n8n/backend-test-utils'; +import type { ProcessInternals } from '@n8n/api-types'; +import { NotFoundError } from '@n8n/errors'; +import express from 'express'; +import { Readable } from 'node:stream'; +import request from 'supertest'; + +import { createE2EDiagnosticsRouter } from '@/services/e2e-diagnostics.router'; +import { HeapDiagnosticsService } from '@/services/heap-diagnostics.service'; +import { ProcessInternalsService } from '@/services/process-internals.service'; + +describe('createE2EDiagnosticsRouter', () => { + const heapDiagnostics = mockInstance(HeapDiagnosticsService); + const processInternals = mockInstance(ProcessInternalsService); + + const createApp = () => express().use('/rest/e2e', createE2EDiagnosticsRouter()); + + it('should serve process internals in the data envelope', async () => { + const internals = { version: 1, instanceType: 'worker' } as ProcessInternals; + processInternals.collect.mockReturnValue(internals); + + const response = await request(createApp()).get('/rest/e2e/internals').expect(200); + + expect(response.body).toEqual({ data: internals }); + }); + + it('should run garbage collection', async () => { + heapDiagnostics.collectGarbage.mockReturnValue({ success: true, message: 'ok' }); + + const response = await request(createApp()).post('/rest/e2e/gc').expect(200); + + expect(response.body).toEqual({ data: { success: true, message: 'ok' } }); + }); + + it('should write a heap snapshot and return its file name', async () => { + heapDiagnostics.writeHeapSnapshot.mockReturnValue({ + success: true, + filePath: 'Heap.1.heapsnapshot', + sizeBytes: 10, + sizeMB: 0, + }); + + const response = await request(createApp()).post('/rest/e2e/heap-snapshot').expect(200); + + expect(response.body.data).toMatchObject({ success: true, filePath: 'Heap.1.heapsnapshot' }); + }); + + it('should stream a heap snapshot file', async () => { + heapDiagnostics.openHeapSnapshot.mockReturnValue( + Readable.from(['snapshot-bytes']) as ReturnType, + ); + + const response = await request(createApp()) + .get('/rest/e2e/heap-snapshot/Heap.1.heapsnapshot') + .buffer(true) + .parse((res, done) => { + let body = ''; + res.on('data', (chunk: Buffer) => (body += chunk.toString())); + res.on('end', () => done(null, body)); + }) + .expect(200); + + expect(heapDiagnostics.openHeapSnapshot).toHaveBeenCalledWith('Heap.1.heapsnapshot'); + expect(response.body).toBe('snapshot-bytes'); + }); + + it('should return the status code of a response error', async () => { + heapDiagnostics.openHeapSnapshot.mockImplementation(() => { + throw new NotFoundError('Snapshot not found: missing.heapsnapshot'); + }); + + await request(createApp()).get('/rest/e2e/heap-snapshot/missing.heapsnapshot').expect(404); + }); +}); diff --git a/packages/cli/src/services/__tests__/heap-diagnostics.service.test.ts b/packages/cli/src/services/__tests__/heap-diagnostics.service.test.ts new file mode 100644 index 00000000000..10211ca5581 --- /dev/null +++ b/packages/cli/src/services/__tests__/heap-diagnostics.service.test.ts @@ -0,0 +1,101 @@ +import { BadRequestError, NotFoundError } from '@n8n/errors'; +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { HeapDiagnosticsService } from '@/services/heap-diagnostics.service'; + +const { writeHeapSnapshot } = vi.hoisted(() => ({ writeHeapSnapshot: vi.fn<() => string>() })); +vi.mock('node:v8', async (importOriginal) => ({ + ...(await importOriginal()), + writeHeapSnapshot, +})); + +describe('HeapDiagnosticsService', () => { + const originalGc = global.gc; + let dir: string; + let service: HeapDiagnosticsService; + + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'heap-diagnostics-')); + service = new HeapDiagnosticsService(); + }); + + afterEach(() => { + global.gc = originalGc; + rmSync(dir, { recursive: true, force: true }); + }); + + describe('collectGarbage', () => { + it('should report failure when gc is not exposed', () => { + global.gc = undefined; + + expect(service.collectGarbage()).toMatchObject({ success: false }); + }); + + it('should run gc twice when gc is exposed', () => { + const gc = vi.fn(); + global.gc = gc as unknown as typeof global.gc; + + expect(service.collectGarbage()).toMatchObject({ success: true }); + expect(gc).toHaveBeenCalledTimes(2); + }); + }); + + describe('writeHeapSnapshot', () => { + it('should run gc first and return the snapshot file name and size', () => { + const gc = vi.fn(); + global.gc = gc as unknown as typeof global.gc; + const filePath = join(dir, 'Heap.1.heapsnapshot'); + writeHeapSnapshot.mockImplementation(() => { + writeFileSync(filePath, 'x'.repeat(2048)); + return filePath; + }); + + const result = service.writeHeapSnapshot(); + + expect(gc).toHaveBeenCalled(); + expect(result).toEqual({ + success: true, + filePath: 'Heap.1.heapsnapshot', + sizeBytes: 2048, + sizeMB: 0, + }); + }); + + it('should report failure when V8 returns no file path', () => { + writeHeapSnapshot.mockReturnValue(''); + + expect(service.writeHeapSnapshot()).toMatchObject({ success: false }); + }); + }); + + describe('openHeapSnapshot', () => { + it('should open a snapshot this process wrote', async () => { + const filePath = join(dir, 'Heap.2.heapsnapshot'); + writeHeapSnapshot.mockImplementation(() => { + writeFileSync(filePath, 'snapshot'); + return filePath; + }); + service.writeHeapSnapshot(); + + const chunks: Buffer[] = []; + for await (const chunk of service.openHeapSnapshot('Heap.2.heapsnapshot')) { + chunks.push(chunk as Buffer); + } + + expect(Buffer.concat(chunks).toString()).toBe('snapshot'); + }); + + it('should reject files that are not heap snapshots', () => { + expect(() => service.openHeapSnapshot('config.json')).toThrow(BadRequestError); + }); + + it('should resolve only the base name of the requested file', () => { + const open = () => service.openHeapSnapshot('../../secret/missing.heapsnapshot'); + + expect(open).toThrow(NotFoundError); + expect(open).toThrow('Snapshot not found: missing.heapsnapshot'); + }); + }); +}); diff --git a/packages/cli/src/services/__tests__/process-internals.service.test.ts b/packages/cli/src/services/__tests__/process-internals.service.test.ts new file mode 100644 index 00000000000..dec85a6546f --- /dev/null +++ b/packages/cli/src/services/__tests__/process-internals.service.test.ts @@ -0,0 +1,103 @@ +import { mockInstance } from '@n8n/backend-test-utils'; +import { GlobalConfig } from '@n8n/config'; +import { ActiveWorkflowTriggers, InstanceSettings, ScheduledTaskManager } from 'n8n-core'; +import { mock } from 'vitest-mock-extended'; + +import { ActiveExecutions } from '@/active-executions'; +import { Push } from '@/push'; +import type { WebSocketPush } from '@/push/websocket.push'; +import { ScalingService } from '@/scaling/scaling.service'; +import { ProcessInternalsService } from '@/services/process-internals.service'; +import { WaitTracker } from '@/wait-tracker'; + +describe('ProcessInternalsService', () => { + const scalingService = mockInstance(ScalingService); + const activeExecutions = mockInstance(ActiveExecutions); + const activeWorkflowTriggers = mockInstance(ActiveWorkflowTriggers); + const scheduledTaskManager = mockInstance(ScheduledTaskManager); + const waitTracker = mockInstance(WaitTracker); + const push = mockInstance(Push); + + beforeEach(() => { + vi.clearAllMocks(); + scalingService.getDiagnosticCounts.mockReturnValue({ + jobResults: 3, + queueListeners: 7, + runningJobs: 0, + }); + activeExecutions.getDiagnosticCounts.mockReturnValue({ executions: 2, responseModes: 1 }); + activeWorkflowTriggers.getDiagnosticCounts.mockReturnValue({ workflows: 4 }); + scheduledTaskManager.getDiagnosticCounts.mockReturnValue({ crons: 2 }); + waitTracker.getDiagnosticCounts.mockReturnValue({ waitingExecutions: 6 }); + push.getBackend.mockReturnValue( + mock({ + getDiagnosticCounts: () => ({ connections: 8 }), + }), + ); + }); + + const createService = ( + instanceType: InstanceSettings['instanceType'], + mode: 'regular' | 'queue', + ) => + new ProcessInternalsService( + mock({ instanceType, hostId: `${instanceType}-1`, isLeader: true }), + mock({ executions: { mode } }), + ); + + it('should report every collection a queue-mode main owns', () => { + const internals = createService('main', 'queue').collect(); + + expect(internals).toMatchObject({ + version: 1, + instanceType: 'main', + hostId: 'main-1', + isLeader: true, + }); + expect(internals.collections).toEqual({ + 'scaling.jobResults': 3, + 'scaling.queueListeners': 7, + 'scaling.runningJobs': 0, + 'activeExecutions.executions': 2, + 'activeExecutions.responseModes': 1, + 'triggers.workflows': 4, + 'scheduledTasks.crons': 2, + 'waitTracker.waitingExecutions': 6, + 'push.connections': 8, + }); + expect(internals.memory.heapUsed).toBeGreaterThan(0); + expect(Object.values(internals.resources).every((count) => count > 0)).toBe(true); + }); + + it('should omit scaling collections in regular mode', () => { + const internals = createService('main', 'regular').collect(); + + expect(Object.keys(internals.collections).some((key) => key.startsWith('scaling.'))).toBe( + false, + ); + }); + + it('should report only scaling collections on a worker', () => { + const internals = createService('worker', 'queue').collect(); + + expect(Object.keys(internals.collections)).toEqual([ + 'scaling.jobResults', + 'scaling.queueListeners', + 'scaling.runningJobs', + ]); + expect(activeExecutions.getDiagnosticCounts).not.toHaveBeenCalled(); + expect(waitTracker.getDiagnosticCounts).not.toHaveBeenCalled(); + }); + + it('should report scaling and active executions on a webhook process', () => { + const internals = createService('webhook', 'queue').collect(); + + expect(Object.keys(internals.collections)).toEqual([ + 'scaling.jobResults', + 'scaling.queueListeners', + 'scaling.runningJobs', + 'activeExecutions.executions', + 'activeExecutions.responseModes', + ]); + }); +}); diff --git a/packages/cli/src/services/e2e-diagnostics.router.ts b/packages/cli/src/services/e2e-diagnostics.router.ts new file mode 100644 index 00000000000..99b6edbaae3 --- /dev/null +++ b/packages/cli/src/services/e2e-diagnostics.router.ts @@ -0,0 +1,59 @@ +import { Container } from '@n8n/di'; +import { ensureError } from '@n8n/utils/errors/ensure-error'; +import { Router, type Request, type Response } from 'express'; +import { pipeline } from 'node:stream/promises'; + +import * as ResponseHelper from '@/response-helper'; +import { HeapDiagnosticsService } from '@/services/heap-diagnostics.service'; +import { ProcessInternalsService } from '@/services/process-internals.service'; + +/** + * Test-only diagnostics routes for processes without REST controllers + * (workers and webhook processes). Mirrors the matching routes of the main's + * E2E controller, including the `{ data }` response envelope. + * Mount only when `E2E_TESTS` is set, at `//e2e`. + */ +export function createE2EDiagnosticsRouter(): Router { + const router = Router(); + const heapDiagnostics = Container.get(HeapDiagnosticsService); + + const handle = + (handler: (req: Request, res: Response) => Promise | unknown) => + async (req: Request, res: Response) => { + try { + const data = await handler(req, res); + if (data !== undefined) res.json({ data }); + } catch (error) { + // A failed stream has already sent headers, so only close the connection. + if (res.headersSent) res.destroy(); + else ResponseHelper.sendErrorResponse(res, ensureError(error)); + } + }; + + router.get( + '/internals', + handle(() => Container.get(ProcessInternalsService).collect()), + ); + + router.post( + '/gc', + handle(() => heapDiagnostics.collectGarbage()), + ); + + router.post( + '/heap-snapshot', + handle(() => heapDiagnostics.writeHeapSnapshot()), + ); + + router.get( + '/heap-snapshot/:filename', + handle(async (req, res) => { + const stream = heapDiagnostics.openHeapSnapshot(String(req.params.filename)); + res.setHeader('Content-Type', 'application/octet-stream'); + await pipeline(stream, res); + return undefined; + }), + ); + + return router; +} diff --git a/packages/cli/src/services/heap-diagnostics.service.ts b/packages/cli/src/services/heap-diagnostics.service.ts new file mode 100644 index 00000000000..1d0ec0a0aaf --- /dev/null +++ b/packages/cli/src/services/heap-diagnostics.service.ts @@ -0,0 +1,73 @@ +import { Service } from '@n8n/di'; +import { BadRequestError, NotFoundError } from '@n8n/errors'; +import { createReadStream, existsSync, statSync, type ReadStream } from 'node:fs'; +import { basename, resolve } from 'node:path'; +import { writeHeapSnapshot } from 'node:v8'; + +export type GarbageCollectionResult = { success: boolean; message: string }; + +export type HeapSnapshotResult = + | { success: true; filePath: string; sizeBytes: number; sizeMB: number } + | { success: false; message: string }; + +/** + * Garbage collection and V8 heap snapshots for E2E and soak tests. Shared by + * the main E2E controller and the worker and webhook servers. + */ +@Service() +export class HeapDiagnosticsService { + /** Snapshot file name to its full path, so downloads resolve only files this process wrote. */ + private readonly snapshotPaths = new Map(); + + /** Needs `--expose-gc`. Runs twice so that objects freed by the first pass are also collected. */ + collectGarbage(): GarbageCollectionResult { + if (typeof global.gc !== 'function') { + return { + success: false, + message: + 'Garbage collection not available. Ensure Node.js is started with --expose-gc flag.', + }; + } + global.gc(); + global.gc(); + return { success: true, message: 'Garbage collection triggered' }; + } + + /** + * Runs GC, then writes a heap snapshot to the working directory. Blocks the + * event loop and needs about twice the heap size in memory while it writes. + */ + writeHeapSnapshot(): HeapSnapshotResult { + this.collectGarbage(); + + const filePath = writeHeapSnapshot(); + if (!filePath) return { success: false, message: 'Failed to write heap snapshot' }; + + const fullPath = resolve(filePath); + const filename = basename(fullPath); + const { size } = statSync(fullPath); + this.snapshotPaths.set(filename, fullPath); + + return { + success: true, + filePath: filename, + sizeBytes: size, + sizeMB: Math.round(size / 1024 / 1024), + }; + } + + /** + * Opens a snapshot by file name. Falls back to the working directory so that + * snapshots written by `--heapsnapshot-signal` or `--heapsnapshot-near-heap-limit` + * can also be downloaded. + */ + openHeapSnapshot(requestedFilename: string): ReadStream { + const filename = basename(requestedFilename); + if (!filename.endsWith('.heapsnapshot')) throw new BadRequestError('Invalid file type'); + + const fullPath = this.snapshotPaths.get(filename) ?? resolve(filename); + if (!existsSync(fullPath)) throw new NotFoundError(`Snapshot not found: ${filename}`); + + return createReadStream(fullPath); + } +} diff --git a/packages/cli/src/services/process-internals.service.ts b/packages/cli/src/services/process-internals.service.ts new file mode 100644 index 00000000000..e6cadf9fba4 --- /dev/null +++ b/packages/cli/src/services/process-internals.service.ts @@ -0,0 +1,63 @@ +import type { ProcessInternals } from '@n8n/api-types'; +import { GlobalConfig } from '@n8n/config'; +import { Container, Service } from '@n8n/di'; +import { ActiveWorkflowTriggers, InstanceSettings, ScheduledTaskManager } from 'n8n-core'; + +import { ActiveExecutions } from '@/active-executions'; +import { Push } from '@/push'; +import { ScalingService } from '@/scaling/scaling.service'; +import { WaitTracker } from '@/wait-tracker'; + +type Counts = Record; + +/** + * Collects counts of in-memory state in this process, for E2E and soak tests. + * Counts only, never contents. Reads only the services this process type runs. + */ +@Service() +export class ProcessInternalsService { + constructor( + private readonly instanceSettings: InstanceSettings, + private readonly globalConfig: GlobalConfig, + ) {} + + collect(): ProcessInternals { + const { instanceType, hostId, isLeader } = this.instanceSettings; + const collections: Counts = {}; + const add = (owner: string, counts: Counts) => { + for (const [name, value] of Object.entries(counts)) collections[`${owner}.${name}`] = value; + }; + + if (this.globalConfig.executions.mode === 'queue') { + add('scaling', Container.get(ScalingService).getDiagnosticCounts()); + } + + if (instanceType === 'main' || instanceType === 'webhook') { + add('activeExecutions', Container.get(ActiveExecutions).getDiagnosticCounts()); + } + + if (instanceType === 'main') { + add('triggers', Container.get(ActiveWorkflowTriggers).getDiagnosticCounts()); + add('scheduledTasks', Container.get(ScheduledTaskManager).getDiagnosticCounts()); + add('waitTracker', Container.get(WaitTracker).getDiagnosticCounts()); + add('push', Container.get(Push).getBackend().getDiagnosticCounts()); + } + + const resources: Counts = {}; + for (const type of process.getActiveResourcesInfo()) { + resources[type] = (resources[type] ?? 0) + 1; + } + + const { rss, heapTotal, heapUsed, external, arrayBuffers } = process.memoryUsage(); + + return { + version: 1, + instanceType, + hostId, + isLeader, + memory: { rss, heapTotal, heapUsed, external, arrayBuffers }, + resources, + collections, + }; + } +} diff --git a/packages/cli/src/wait-tracker.ts b/packages/cli/src/wait-tracker.ts index 508b12f91c8..4609437a0aa 100644 --- a/packages/cli/src/wait-tracker.ts +++ b/packages/cli/src/wait-tracker.ts @@ -68,6 +68,11 @@ export class WaitTracker { return this.waitingExecutions[executionId] !== undefined; } + /** Sizes of the in-memory collections, for diagnostics and tests. */ + getDiagnosticCounts() { + return { waitingExecutions: Object.keys(this.waitingExecutions).length }; + } + init() { if (this.instanceSettings.isLeader) this.startTracking(); } diff --git a/packages/cli/src/webhooks/__tests__/webhook-server.test.ts b/packages/cli/src/webhooks/__tests__/webhook-server.test.ts index f45efbb02a5..42e5788474c 100644 --- a/packages/cli/src/webhooks/__tests__/webhook-server.test.ts +++ b/packages/cli/src/webhooks/__tests__/webhook-server.test.ts @@ -11,10 +11,23 @@ import { ChatServer } from '@/chat/chat-server'; import { WebhookServer } from '@/webhooks/webhook-server'; const mockApp = mock(); +const e2eFlags = vi.hoisted(() => ({ inE2ETests: false })); +const diagnosticsRouter = vi.hoisted(() => vi.fn()); vi.mock('express', async () => ({ __esModule: true, default: () => mockApp })); +vi.mock('@/constants', async (importOriginal) => ({ + ...(await importOriginal()), + get inE2ETests() { + return e2eFlags.inE2ETests; + }, +})); +vi.mock('@/services/e2e-diagnostics.router', () => ({ + createE2EDiagnosticsRouter: vi.fn(() => diagnosticsRouter), +})); describe('WebhookServer', () => { - it('should mount the chat WebSocket server', () => { + beforeEach(() => { + vi.clearAllMocks(); + e2eFlags.inE2ETests = false; mockInstance(Logger); Container.set(DbConnection, mock()); Container.set( @@ -41,6 +54,9 @@ describe('WebhookServer', () => { }, }), ); + }); + + it('should mount the chat WebSocket server', () => { const chatServer = mockInstance(ChatServer); const webhookServer = new WebhookServer(); @@ -51,4 +67,15 @@ describe('WebhookServer', () => { expect(chatServer.setup).toHaveBeenCalledWith(httpServer, mockApp); }); + + it('should mount diagnostics only during E2E tests', async () => { + const webhookServer = new WebhookServer(); + + await webhookServer.configure(); + expect(mockApp.use).not.toHaveBeenCalled(); + + e2eFlags.inE2ETests = true; + await webhookServer.configure(); + expect(mockApp.use).toHaveBeenCalledWith('/rest/e2e', diagnosticsRouter); + }); }); diff --git a/packages/cli/src/webhooks/webhook-server.ts b/packages/cli/src/webhooks/webhook-server.ts index 0e09840b1d9..9c1e6ec108b 100644 --- a/packages/cli/src/webhooks/webhook-server.ts +++ b/packages/cli/src/webhooks/webhook-server.ts @@ -2,6 +2,7 @@ import { Container, Service } from '@n8n/di'; import { AbstractServer } from '@/abstract-server'; import { ChatServer } from '@/chat/chat-server'; +import { inE2ETests } from '@/constants'; @Service() export class WebhookServer extends AbstractServer { @@ -11,6 +12,12 @@ export class WebhookServer extends AbstractServer { const { PrometheusMetricsService } = await import('@/metrics/prometheus/index.js'); Container.get(PrometheusMetricsService).init(this.app); } + + // Webhook procs have no REST controllers, so the test-only diagnostics routes live here. + if (inE2ETests) { + const { createE2EDiagnosticsRouter } = await import('@/services/e2e-diagnostics.router.js'); + this.app.use(`/${this.restEndpoint}/e2e`, createE2EDiagnosticsRouter()); + } } /** The chat widget opens its WebSocket against the same origin that served the chat webhook */ diff --git a/packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts b/packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts index ecef977d31d..8acd62f8d3c 100644 --- a/packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts +++ b/packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts @@ -132,6 +132,16 @@ describe('ActiveWorkflowTriggers', () => { ); }; + it('counts a workflow with multiple triggers once and removes it after deactivation', async () => { + expect(activeWorkflowTriggers.getDiagnosticCounts()).toEqual({ workflows: 0 }); + + await addWorkflow({ triggerNodes: [triggerNode], pollNodes: [pollNode] }); + expect(activeWorkflowTriggers.getDiagnosticCounts()).toEqual({ workflows: 1 }); + + await activeWorkflowTriggers.remove(workflowId); + expect(activeWorkflowTriggers.getDiagnosticCounts()).toEqual({ workflows: 0 }); + }); + describe('addAllTriggers()', () => { describe('should activate workflow', () => { it('with trigger function nodes', async () => { diff --git a/packages/core/src/execution-engine/__tests__/scheduled-task-manager.test.ts b/packages/core/src/execution-engine/__tests__/scheduled-task-manager.test.ts index 02898246b5f..9b1a200865a 100644 --- a/packages/core/src/execution-engine/__tests__/scheduled-task-manager.test.ts +++ b/packages/core/src/execution-engine/__tests__/scheduled-task-manager.test.ts @@ -161,6 +161,18 @@ describe('ScheduledTaskManager', () => { expect(scheduledTaskManager.getTargetIds(workflowGroup('unknown-workflow'))).toEqual([]); }); + it('reports the number of registered crons across group types', () => { + scheduledTaskManager.register(cronContext({ targetId: 'node-a' }), onTick); + scheduledTaskManager.register(cronContext({ targetId: 'node-b' }), onTick); + scheduledTaskManager.register(cronContext({ group: agentTaskGroup() }), onTick); + + expect(scheduledTaskManager.getDiagnosticCounts()).toEqual({ crons: 3 }); + + scheduledTaskManager.deregisterGroup(workflowGroup()); + + expect(scheduledTaskManager.getDiagnosticCounts()).toEqual({ crons: 1 }); + }); + it('should deregister CronJobs for a single target, leaving other targets intact', () => { const nodeA = 'node-a'; const nodeB = 'node-b'; diff --git a/packages/core/src/execution-engine/active-workflow-triggers.ts b/packages/core/src/execution-engine/active-workflow-triggers.ts index 2c3f73bd6ca..a7f04cb86c5 100644 --- a/packages/core/src/execution-engine/active-workflow-triggers.ts +++ b/packages/core/src/execution-engine/active-workflow-triggers.ts @@ -90,6 +90,11 @@ export class ActiveWorkflowTriggers { return Array.from(this.activeTriggersByWorkflowId.keys()); } + /** Sizes of the in-memory collections, for diagnostics and tests. */ + getDiagnosticCounts() { + return { workflows: this.activeTriggersByWorkflowId.size }; + } + /** * Returns the workflow data for the given ID if currently active in memory. */ diff --git a/packages/core/src/execution-engine/scheduled-task-manager.ts b/packages/core/src/execution-engine/scheduled-task-manager.ts index b8e83892874..55879039e41 100644 --- a/packages/core/src/execution-engine/scheduled-task-manager.ts +++ b/packages/core/src/execution-engine/scheduled-task-manager.ts @@ -176,6 +176,15 @@ export class ScheduledTaskManager { return Array.from(this.cronsByGroupType.get(groupType)?.keys() ?? []); } + /** Sizes of the in-memory collections, for diagnostics and tests. */ + getDiagnosticCounts() { + let crons = 0; + for (const cronsByGroup of this.cronsByGroupType.values()) { + for (const groupCrons of cronsByGroup.values()) crons += groupCrons.size; + } + return { crons }; + } + getTargetIds(group: ScheduledTaskGroup): string[] { const groupCrons = this.getGroupCrons(group); if (!groupCrons) return []; diff --git a/packages/quality/environments/containers/stack.ts b/packages/quality/environments/containers/stack.ts index c37cc1b19ff..4bee711c53f 100644 --- a/packages/quality/environments/containers/stack.ts +++ b/packages/quality/environments/containers/stack.ts @@ -80,9 +80,28 @@ export interface N8NStack { * which cannot reach the host-mapped ports in `baseUrl`/`mainUrls`. */ internalMainUrls: string[]; + /** + * Direct host-mapped URL of every n8n process (mains, workers, webhook procs), + * for per-process diagnostics such as `/metrics` and `/rest/e2e/internals`. + */ + processUrls: N8NProcessUrl[]; startupDiagnostics: N8NStartupDiagnostics; } +export interface N8NProcessUrl { + role: 'main' | 'worker' | 'webhook'; + /** Container name, e.g. `-n8n-worker-1`. */ + name: string; + url: string; +} + +function n8nProcessRole(containerName: string): N8NProcessUrl['role'] | undefined { + if (containerName.endsWith('-n8n') || containerName.includes('-n8n-main-')) return 'main'; + if (containerName.includes('-n8n-worker-')) return 'worker'; + if (containerName.includes('-n8n-webhook-')) return 'webhook'; + return undefined; +} + function shouldServiceStart(name: ServiceName, service: Service, ctx: StartContext): boolean { // Explicitly requested services always start if (ctx.config.services?.includes(name)) return true; @@ -464,6 +483,15 @@ export async function createN8NStack(config: N8NConfig = {}): Promise } log(`Direct main URLs: ${mainUrls.join(', ')}`); + const processUrls: N8NProcessUrl[] = []; + for (const container of containers) { + const name = container.getName().replace(/^\//, ''); + const role = n8nProcessRole(name); + if (role) { + processUrls.push({ role, name, url: `http://localhost:${container.getMappedPort(5678)}` }); + } + } + // Run verification hooks (e.g. keycloak connectivity check) const n8nContainers = containers.filter((c) => { const name = c.getName(); @@ -617,6 +645,7 @@ export async function createN8NStack(config: N8NConfig = {}): Promise }, mainUrls, internalMainUrls, + processUrls, startupDiagnostics: n8nResult.diagnostics, }; } catch (error) { diff --git a/packages/quality/testing/playwright/fixtures/base.ts b/packages/quality/testing/playwright/fixtures/base.ts index 1bb1e669b9a..05cc5225fb9 100644 --- a/packages/quality/testing/playwright/fixtures/base.ts +++ b/packages/quality/testing/playwright/fixtures/base.ts @@ -2,7 +2,7 @@ import type { CurrentsFixtures, CurrentsWorkerFixtures } from '@currents/playwri import { fixtures as currentsFixtures } from '@currents/playwright'; import { test as base, expect, request } from '@playwright/test'; import type { ServiceHelpers } from 'n8n-containers/services/types'; -import type { N8NConfig, N8NStack } from 'n8n-containers/stack'; +import type { N8NConfig, N8NProcessUrl, N8NStack } from 'n8n-containers/stack'; import { createN8NStack } from 'n8n-containers/stack'; import { a11yFixtures, type A11yTestFixtures } from './a11y'; @@ -42,6 +42,11 @@ type TestFixtures = { * Index 0 = main-1, Index 1 = main-2, etc. */ mainUrls: string[]; + /** + * Direct URL of every n8n process (mains, workers, webhook procs). Outside + * container mode this is the backend URL as a single main. + */ + processUrls: N8NProcessUrl[]; /** * Create an API helper for a specific main instance (bypasses load balancer). * Useful for multi-main testing scenarios. @@ -301,6 +306,10 @@ export const test = base.extend< await use(urls); }, + processUrls: async ({ n8nContainer, backendUrl }, use) => { + await use(n8nContainer?.processUrls ?? [{ role: 'main', name: 'main', url: backendUrl }]); + }, + createApiForMain: async ({ n8nContainer, n8nStackConfig }, use, testInfo) => { const contexts: Array<{ dispose: () => Promise }> = []; diff --git a/packages/quality/testing/playwright/tests/e2e/capabilities/process-internals.spec.ts b/packages/quality/testing/playwright/tests/e2e/capabilities/process-internals.spec.ts new file mode 100644 index 00000000000..fab73af0604 --- /dev/null +++ b/packages/quality/testing/playwright/tests/e2e/capabilities/process-internals.spec.ts @@ -0,0 +1,125 @@ +import { nanoid } from 'nanoid'; +import type { N8NProcessUrl } from 'n8n-containers/stack'; + +import { test, expect } from '../../../fixtures/base'; +import { + collectProcessGarbage, + diffProcessInternals, + probeMissingHeapSnapshot, + readProcessInternals, +} from '../../../utils/process-internals'; + +test.use({ capability: { env: { TEST_ISOLATION: 'process-internals' } } }); + +/** Collections each process role must report, whatever the execution mode. */ +const REQUIRED_COLLECTIONS: Record = { + main: ['activeExecutions.executions', 'triggers.workflows', 'push.connections'], + webhook: ['activeExecutions.executions', 'scaling.jobResults'], + worker: ['scaling.runningJobs'], +}; + +test.describe( + 'Process internals', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('should report in-memory state counts for every n8n process', async ({ processUrls }) => { + const readings = await readProcessInternals(processUrls); + + for (const { role, name, internals } of readings) { + expect(internals.version, name).toBe(1); + expect(internals.instanceType, name).toBe(role); + expect(internals.memory.heapUsed, name).toBeGreaterThan(0); + expect(Object.keys(internals.collections), name).toEqual( + expect.arrayContaining(REQUIRED_COLLECTIONS[role]), + ); + } + }); + + test('should expose garbage collection and heap snapshot routes on every n8n process', async ({ + processUrls, + n8nContainer, + }) => { + // Container stacks start n8n with --expose-gc; a local dev server may not. + test.skip(!n8nContainer, 'container-only: requires --expose-gc'); + + for (const target of processUrls) { + expect(await collectProcessGarbage(target), target.name).toBe(true); + + // A cheap probe that the snapshot routes are mounted: taking a real + // snapshot pauses the process and is left to soak and memory suites. + const missing = await probeMissingHeapSnapshot(target); + expect(missing.status, target.name).toBe(404); + expect(missing.body, target.name).toContain('Snapshot not found'); + } + }); + + test('should return in-memory execution state to baseline after webhook executions finish', async ({ + api, + processUrls, + }) => { + const path = `internals-${nanoid()}`; + const { workflowId, createdWorkflow, webhookPath } = + await api.workflows.createWorkflowFromDefinition( + { + name: `Process internals ${nanoid()}`, + nodes: [ + { + id: 'webhook', + name: 'Webhook', + type: 'n8n-nodes-base.webhook', + typeVersion: 2, + position: [0, 0], + parameters: { httpMethod: 'POST', path, responseMode: 'lastNode', options: {} }, + }, + { + id: 'noop', + name: 'NoOp', + type: 'n8n-nodes-base.noOp', + typeVersion: 1, + position: [200, 0], + parameters: {}, + }, + ], + connections: { Webhook: { main: [[{ node: 'NoOp', type: 'main', index: 0 }]] } }, + }, + { webhookPrefix: 'internals' }, + ); + await api.workflows.activate(workflowId, createdWorkflow.versionId!); + const trigger = async () => + await api.webhooks.trigger(`/webhook/${webhookPath}`, { + method: 'POST', + data: { hello: 'world' }, + maxNotFoundRetries: 20, + }); + // Warm up once so that lazily created state is part of the baseline. + expect((await trigger()).ok()).toBe(true); + + const before = await readProcessInternals(processUrls); + for (let i = 0; i < 5; i++) { + expect((await trigger()).ok()).toBe(true); + } + + // `lastNode` responds when the last node runs, just before cleanup. + await expect + .poll(async () => { + const deltas = diffProcessInternals(before, await readProcessInternals(processUrls)); + return deltas.map(({ name, collections }) => ({ + name, + executions: collections['activeExecutions.executions'] ?? 0, + responseModes: collections['activeExecutions.responseModes'] ?? 0, + jobResults: collections['scaling.jobResults'] ?? 0, + })); + }) + .toEqual( + processUrls.map(({ name }) => ({ + name, + executions: 0, + responseModes: 0, + jobResults: 0, + })), + ); + }); + }, +); diff --git a/packages/quality/testing/playwright/utils/process-internals.ts b/packages/quality/testing/playwright/utils/process-internals.ts new file mode 100644 index 00000000000..503e5a7667a --- /dev/null +++ b/packages/quality/testing/playwright/utils/process-internals.ts @@ -0,0 +1,68 @@ +import type { ProcessInternals } from '@n8n/api-types'; +import type { N8NProcessUrl } from 'n8n-containers/stack'; + +import { TestError } from '../Types'; + +export interface ProcessInternalsReading extends N8NProcessUrl { + internals: ProcessInternals; +} + +export interface ProcessInternalsDelta extends N8NProcessUrl { + /** Collections whose size changed, as `after - before`. */ + collections: Record; +} + +/** Calls a test-only `/rest/e2e` route on one n8n process and unwraps the `{ data }` envelope. */ +async function callE2ERoute(target: N8NProcessUrl, route: string, method = 'GET'): Promise { + const response = await fetch(`${target.url}/rest/e2e/${route}`, { method }); + if (!response.ok) { + throw new TestError( + `${method} /rest/e2e/${route} failed on ${target.name}: ${response.status} ${await response.text()}`, + ); + } + return ((await response.json()) as { data: T }).data; +} + +/** Reads `/rest/e2e/internals` from every given n8n process in parallel. */ +export async function readProcessInternals( + processes: N8NProcessUrl[], +): Promise { + return await Promise.all( + processes.map(async (target) => ({ + ...target, + internals: await callE2ERoute(target, 'internals'), + })), + ); +} + +/** + * Runs garbage collection in one n8n process: main, worker, or webhook process. + * Needs `--expose-gc`, which container stacks set. + */ +export async function collectProcessGarbage(target: N8NProcessUrl): Promise { + return (await callE2ERoute<{ success: boolean }>(target, 'gc', 'POST')).success; +} + +/** Probes the snapshot download route without taking a heap snapshot. */ +export async function probeMissingHeapSnapshot(target: N8NProcessUrl) { + const response = await fetch(`${target.url}/rest/e2e/heap-snapshot/missing.heapsnapshot`); + return { status: response.status, body: await response.text() }; +} + +/** Pairs readings by process name and returns only the collection sizes that changed. */ +export function diffProcessInternals( + before: ProcessInternalsReading[], + after: ProcessInternalsReading[], +): ProcessInternalsDelta[] { + return after.flatMap(({ internals, ...target }) => { + const previous = before.find((reading) => reading.name === target.name)?.internals; + if (!previous) return []; + + const collections: Record = {}; + for (const [key, size] of Object.entries(internals.collections)) { + const change = size - (previous.collections[key] ?? 0); + if (change !== 0) collections[key] = change; + } + return [{ ...target, collections }]; + }); +}