feat(core): Add test-only process internals and heap diagnostics routes (no-changelog) (#39881)

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

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<string, number>;
+	/**
+	 * Sizes of known in-memory collections, keyed `<owner>.<collection>`,
+	 * e.g. `scaling.jobResults`. A key is absent when its owner does not run
+	 * in this process.
+	 */
+	collections: Record<string, number>;
+}

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,

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]);

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.

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<typeof import('@/constants')>()),
+	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;
 

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<z.infer<typeof flagsSchema>> {
 		};
 
 		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);

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<string, string>();
-
 	/**
 	 * 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

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'],

packages/cli/src/push/abstract.push.ts

@@ -127,6 +127,11 @@ export abstract class AbstractPush<Connection> extends TypedEmitter<AbstractPush
 		this.sendTo(pushMsg, userPushRefs);
 	}
 
+	/** Sizes of the in-memory collections, for diagnostics and tests. */
+	getDiagnosticCounts() {
+		return { connections: Object.keys(this.connections).length };
+	}
+
 	closeAllConnections() {
 		for (const pushRef in this.connections) {
 			// Signal the connection that we want to close it.

packages/cli/src/scaling/__tests__/scaling.service.test.ts

@@ -988,6 +988,56 @@ describe('ScalingService', () => {
 		});
 	});
 
+	describe('getDiagnosticCounts', () => {
+		it('should report stored job results, queue listeners, and running jobs', async () => {
+			const activeExecutions = mock<ActiveExecutions>();
+			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();

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<express.Application>();
+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<typeof import('@/constants')>()),
+	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<GlobalConfig>({
 			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<http.Server>();
+			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<http.Server>();
 			vi.spyOn(http, 'createServer').mockReturnValue(server);

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

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

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

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<HeapDiagnosticsService['openHeapSnapshot']>,
+		);
+
+		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);
+	});
+});

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<typeof import('node:v8')>()),
+	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');
+		});
+	});
+});

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<WebSocketPush>({
+				getDiagnosticCounts: () => ({ connections: 8 }),
+			}),
+		);
+	});
+
+	const createService = (
+		instanceType: InstanceSettings['instanceType'],
+		mode: 'regular' | 'queue',
+	) =>
+		new ProcessInternalsService(
+			mock<InstanceSettings>({ instanceType, hostId: `${instanceType}-1`, isLeader: true }),
+			mock<GlobalConfig>({ 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',
+		]);
+	});
+});

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 `/<rest endpoint>/e2e`.
+ */
+export function createE2EDiagnosticsRouter(): Router {
+	const router = Router();
+	const heapDiagnostics = Container.get(HeapDiagnosticsService);
+
+	const handle =
+		(handler: (req: Request, res: Response) => Promise<unknown> | 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;
+}

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<string, string>();
+
+	/** 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);
+	}
+}

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<string, number>;
+
+/**
+ * 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,
+		};
+	}
+}

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();
 	}

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<express.Application>();
+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<typeof import('@/constants')>()),
+	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<DbConnection>());
 		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);
+	});
 });

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 */

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 () => {

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

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.
 	 */

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 [];

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. `<project>-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<N8NStack>
 		}
 		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<N8NStack>
 			},
 			mainUrls,
 			internalMainUrls,
+			processUrls,
 			startupDiagnostics: n8nResult.diagnostics,
 		};
 	} catch (error) {

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<void> }> = [];
 

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<N8NProcessUrl['role'], string[]> = {
+	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,
+					})),
+				);
+		});
+	},
+);

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<string, number>;
+}
+
+/** Calls a test-only `/rest/e2e` route on one n8n process and unwraps the `{ data }` envelope. */
+async function callE2ERoute<T>(target: N8NProcessUrl, route: string, method = 'GET'): Promise<T> {
+	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<ProcessInternalsReading[]> {
+	return await Promise.all(
+		processes.map(async (target) => ({
+			...target,
+			internals: await callE2ERoute<ProcessInternals>(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<boolean> {
+	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<string, number> = {};
+		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 }];
+	});
+}