diff --git a/packages/cli/src/eventbus/event-message-classes/__tests__/event-names.test.ts b/packages/cli/src/eventbus/event-message-classes/__tests__/event-names.test.ts index b775e704253..06d8a3ea85f 100644 --- a/packages/cli/src/eventbus/event-message-classes/__tests__/event-names.test.ts +++ b/packages/cli/src/eventbus/event-message-classes/__tests__/event-names.test.ts @@ -15,6 +15,7 @@ const KNOWN_EVENT_GROUPS = [ 'n8n.runner.', 'n8n.queue.', 'n8n.execution.', + 'n8n.instanceReporting.', ]; describe('log streaming event names', () => { diff --git a/packages/cli/src/eventbus/event-message-classes/index.ts b/packages/cli/src/eventbus/event-message-classes/index.ts index d8bfc9b6765..7e84edb3ef3 100644 --- a/packages/cli/src/eventbus/event-message-classes/index.ts +++ b/packages/cli/src/eventbus/event-message-classes/index.ts @@ -66,6 +66,10 @@ export const eventNamesWorkflow = [ 'n8n.workflow.cancelled', ] as const; export const eventNamesGeneric = ['n8n.worker.started', 'n8n.worker.stopped'] as const; +export const eventNamesInstanceReporting = [ + 'n8n.instanceReporting.success', + 'n8n.instanceReporting.failed', +] as const; export const eventNamesNode = ['n8n.node.started', 'n8n.node.finished'] as const; export const eventNamesExecution = [ 'n8n.execution.throttled', @@ -183,6 +187,7 @@ export type EventNamesAuditType = (typeof eventNamesAudit)[number]; export type EventNamesNodeType = (typeof eventNamesNode)[number]; export type EventNamesExecutionType = (typeof eventNamesExecution)[number]; export type EventNamesGenericType = (typeof eventNamesGeneric)[number]; +export type EventNamesInstanceReportingType = (typeof eventNamesInstanceReporting)[number]; export type EventNamesTypes = | EventNamesAuditType @@ -190,6 +195,7 @@ export type EventNamesTypes = | EventNamesNodeType | EventNamesExecutionType | EventNamesGenericType + | EventNamesInstanceReportingType | EventNamesAiNodesType | EventNamesRunnerType | EventNamesQueueType @@ -201,6 +207,7 @@ export const eventNamesAll = [ ...eventNamesWorkflow, ...eventNamesNode, ...eventNamesGeneric, + ...eventNamesInstanceReporting, ...eventNamesAiNodes, ...eventNamesRunner, ...eventNamesQueue, diff --git a/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts b/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts index 0ce50ab116c..388e4c28706 100644 --- a/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts +++ b/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts @@ -3358,4 +3358,22 @@ describe('LogStreamingEventRelay', () => { }); }); }); + + describe('instance reporting events', () => { + it('should log on instance-report-delivered event', () => { + eventService.emit('instance-report-delivered'); + + expect(eventBus.send).toHaveBeenCalledWith( + expect.objectContaining({ eventName: 'n8n.instanceReporting.success' }), + ); + }); + + it('should log on instance-report-failed event', () => { + eventService.emit('instance-report-failed'); + + expect(eventBus.send).toHaveBeenCalledWith( + expect.objectContaining({ eventName: 'n8n.instanceReporting.failed' }), + ); + }); + }); }); diff --git a/packages/cli/src/events/maps/relay.event-map.ts b/packages/cli/src/events/maps/relay.event-map.ts index a27013937ce..9b85808fa63 100644 --- a/packages/cli/src/events/maps/relay.event-map.ts +++ b/packages/cli/src/events/maps/relay.event-map.ts @@ -93,6 +93,12 @@ export type RelayEventMap = { credentialId?: string; }; + // Delivery outcome of an instance usage report. No payload: the event name is + // the whole signal a log-streaming consumer needs. + 'instance-report-delivered': {}; + + 'instance-report-failed': {}; + // #endregion // #region Workflow diff --git a/packages/cli/src/events/relays/log-streaming.event-relay.ts b/packages/cli/src/events/relays/log-streaming.event-relay.ts index 28937decc92..af5f89a3cc3 100644 --- a/packages/cli/src/events/relays/log-streaming.event-relay.ts +++ b/packages/cli/src/events/relays/log-streaming.event-relay.ts @@ -3,6 +3,7 @@ import { Service } from '@n8n/di'; import { InstanceSettings } from 'n8n-core'; import type { IWorkflowBase, JsonValue } from 'n8n-workflow'; +import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-message-generic'; import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; import { EventService } from '@/events/event.service'; import type { RelayEventMap, UserLike } from '@/events/maps/relay.event-map'; @@ -187,6 +188,8 @@ export class LogStreamingEventRelay extends EventRelay { 'mcp-oauth-completed': (event) => this.mcpOauthCompleted(event), 'mcp-tool-called': (event) => this.mcpToolCalled(event), 'mcp-access-updated': (event) => this.mcpAccessUpdated(event), + 'instance-report-delivered': () => this.instanceReportDelivered(), + 'instance-report-failed': () => this.instanceReportFailed(), }); } @@ -222,6 +225,16 @@ export class LogStreamingEventRelay extends EventRelay { }); } + private instanceReportDelivered() { + void this.eventBus.send( + new EventMessageGeneric({ eventName: 'n8n.instanceReporting.success' }), + ); + } + + private instanceReportFailed() { + void this.eventBus.send(new EventMessageGeneric({ eventName: 'n8n.instanceReporting.failed' })); + } + @Redactable() private packageImportFailed({ user, ...rest }: RelayEventMap['n8n-package-import-failed']) { void this.eventBus.sendAuditEvent({ diff --git a/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting-scheduler.service.test.ts b/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting-scheduler.service.test.ts index ab6a49b559d..579d184ca9f 100644 --- a/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting-scheduler.service.test.ts +++ b/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting-scheduler.service.test.ts @@ -6,6 +6,8 @@ import { UserError } from 'n8n-workflow'; import type { Mocked } from 'vitest'; import { mock } from 'vitest-mock-extended'; +import type { EventService } from '@/events/event.service'; + import type { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository'; import { InstanceReportingScheduler } from '../instance-reporting-scheduler.service'; import type { InstanceReportingSettingsService } from '../instance-reporting-settings.service'; @@ -27,12 +29,19 @@ interface Harness { settingsService: Mocked; instanceSettings: Mocked; modulesConfig: Mocked; + eventService: Mocked; + /** Fire the deferred `server-started` listeners (only needed when autoStart is false). */ + startServer: () => void; } function makeHarness({ baseUrl = 'https://example.com', isLeader = true, disabledModules = [] as ModuleName[], + // The scheduler arms its first tick on `server-started`. By default the harness + // fires that listener the moment it is registered, so `init()` starts reporting + // as the tests expect. Pass false to hold it and fire it later via `startServer`. + autoStart = true, } = {}): Harness { const config = new InstanceReportingConfig(); config.instanceReportingBaseUrl = baseUrl; @@ -54,6 +63,23 @@ function makeHarness({ }); const modulesConfig = mock({ disabledModules }); + const serverStartedListeners: Array<() => void> = []; + const eventService = mock(); + eventService.once.mockImplementation((event, listener) => { + if (event === 'server-started') { + if (autoStart) (listener as () => void)(); + else serverStartedListeners.push(listener as () => void); + } + return eventService; + }); + eventService.off.mockImplementation((event, listener) => { + if (event === 'server-started') { + const index = serverStartedListeners.indexOf(listener as () => void); + if (index !== -1) serverStartedListeners.splice(index, 1); + } + return eventService; + }); + const scheduler = new InstanceReportingScheduler( config, reportingService, @@ -61,6 +87,7 @@ function makeHarness({ settingsService, instanceSettings, modulesConfig, + eventService, mockLogger(), ); @@ -71,6 +98,8 @@ function makeHarness({ settingsService, instanceSettings, modulesConfig, + eventService, + startServer: () => serverStartedListeners.forEach((listener) => listener()), }; } @@ -136,6 +165,22 @@ describe('InstanceReportingScheduler', () => { expect(reportingService.sendReport).toHaveBeenCalledTimes(3); }); + + test('holds the first tick until the server has started', async () => { + // A boot catch-up report must not run before the event bus is up, or its + // success/failure event is emitted with nowhere to land. + vi.setSystemTime(new Date(AFTER_SLOT)); + const { scheduler, reportingService, startServer } = makeHarness({ autoStart: false }); + + await scheduler.init(); + await settle(); + expect(reportingService.sendReport).not.toHaveBeenCalled(); + + startServer(); + await settle(); + + expect(reportingService.sendReport).toHaveBeenCalledTimes(1); + }); }); describe('catch-up', () => { @@ -194,6 +239,21 @@ describe('InstanceReportingScheduler', () => { expect(reportingService.sendReport).toHaveBeenCalledTimes(1); }); + test('drops the deferred first tick when it steps down before the server starts', async () => { + // Leader at boot, but it loses leadership before `server-started`. The + // pending listener must be removed, so it neither leaks nor reports, and a + // later takeover cannot start it a second time. + vi.setSystemTime(new Date(AFTER_SLOT)); + const { scheduler, reportingService, startServer } = makeHarness({ autoStart: false }); + + await scheduler.init(); + scheduler.stop(); + startServer(); + await settle(); + + expect(reportingService.sendReport).not.toHaveBeenCalled(); + }); + test('stops reporting when this main steps down', async () => { const { scheduler, reportingService } = makeHarness(); diff --git a/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting.service.test.ts b/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting.service.test.ts index 339e43cd895..48c333eec57 100644 --- a/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting.service.test.ts +++ b/packages/cli/src/modules/instance-reporting/__tests__/instance-reporting.service.test.ts @@ -10,6 +10,7 @@ import type { IHttpRequestOptions } from 'n8n-workflow'; import type { Mocked } from 'vitest'; import { mock } from 'vitest-mock-extended'; +import type { EventService } from '@/events/event.service'; import type { InsightsService } from '@/modules/insights/insights.service'; import type { OwnershipService } from '@/services/ownership.service'; @@ -83,6 +84,7 @@ interface Harness { reportRepository: Mocked; insightsService: Mocked; http: HttpRequestClient; + eventService: Mocked; clientOptions: HttpRequestClientOptions | undefined; } @@ -116,6 +118,8 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness { }), }); + const eventService = mock(); + const service = new InstanceReportingService( config, reportRepository, @@ -124,10 +128,11 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness { ownershipService, licenseMetricsRepository, mockLogger(), + eventService, outboundHttp, ); - return { service, reportRepository, insightsService, http, clientOptions }; + return { service, reportRepository, insightsService, http, eventService, clientOptions }; } describe('InstanceReportingService', () => { @@ -317,6 +322,25 @@ describe('InstanceReportingService', () => { expect(reportRepository.markDelivered).not.toHaveBeenCalled(); }); + test('emits a delivered event once the report is delivered', async () => { + const { service, eventService } = makeHarness(); + + await service.sendReport(); + + expect(eventService.emit).toHaveBeenCalledWith('instance-report-delivered'); + expect(eventService.emit).not.toHaveBeenCalledWith('instance-report-failed'); + }); + + test('emits a failed event when a delivery attempt fails', async () => { + const { service, eventService, http } = makeHarness(); + vi.mocked(http.request).mockRejectedValue(new Error('Network error')); + + await expect(service.sendReport()).rejects.toThrow('Network error'); + + expect(eventService.emit).toHaveBeenCalledWith('instance-report-failed'); + expect(eventService.emit).not.toHaveBeenCalledWith('instance-report-delivered'); + }); + test('marks a report the receiver already holds as delivered, without retrying', async () => { const { service, reportRepository, http } = makeHarness(); vi.mocked(http.request).mockResolvedValue({ statusCode: 409, body: '', headers: {} }); diff --git a/packages/cli/src/modules/instance-reporting/instance-reporting-scheduler.service.ts b/packages/cli/src/modules/instance-reporting/instance-reporting-scheduler.service.ts index 116a64ea2fd..53f0e4c5752 100644 --- a/packages/cli/src/modules/instance-reporting/instance-reporting-scheduler.service.ts +++ b/packages/cli/src/modules/instance-reporting/instance-reporting-scheduler.service.ts @@ -11,6 +11,8 @@ import { InstanceReportingSettingsService } from './instance-reporting-settings. import { InstanceReportingConfig } from './instance-reporting.config'; import { InstanceReportingService, RETRY_DELAY_MS } from './instance-reporting.service'; +import { EventService } from '@/events/event.service'; + const MINUTES_PER_DAY = 24 * 60; /** @@ -38,6 +40,8 @@ export class InstanceReportingScheduler { private isShuttingDown = false; + private readonly onServerStarted = () => this.start(); + constructor( private readonly config: InstanceReportingConfig, private readonly reportingService: InstanceReportingService, @@ -45,6 +49,7 @@ export class InstanceReportingScheduler { private readonly settingsService: InstanceReportingSettingsService, private readonly instanceSettings: InstanceSettings, private readonly modulesConfig: ModulesConfig, + private readonly eventService: EventService, private readonly logger: Logger, ) { this.logger = this.logger.scoped('instance-reporting'); @@ -74,7 +79,11 @@ export class InstanceReportingScheduler { return; } - if (this.instanceSettings.isLeader) this.start(); + // Defer the first tick until the server has finished starting. A boot + // catch-up report can send right away, but log streaming module was not properly initialized + if (this.instanceSettings.isLeader) { + this.eventService.once('server-started', this.onServerStarted); + } } get isEnabled(): boolean { @@ -101,6 +110,8 @@ export class InstanceReportingScheduler { @OnLeaderStepdown() stop(): void { + this.eventService.off('server-started', this.onServerStarted); + if (this.timeout === undefined) return; clearTimeout(this.timeout); diff --git a/packages/cli/src/modules/instance-reporting/instance-reporting.service.ts b/packages/cli/src/modules/instance-reporting/instance-reporting.service.ts index 74466ece66c..b223d476e5e 100644 --- a/packages/cli/src/modules/instance-reporting/instance-reporting.service.ts +++ b/packages/cli/src/modules/instance-reporting/instance-reporting.service.ts @@ -8,6 +8,7 @@ import { InstanceSettings } from 'n8n-core'; import { OperationalError } from 'n8n-workflow'; import { N8N_VERSION } from '@/constants'; +import { EventService } from '@/events/event.service'; import { InsightsService } from '@/modules/insights/insights.service'; import { OwnershipService } from '@/services/ownership.service'; @@ -55,6 +56,7 @@ export class InstanceReportingService { private readonly ownershipService: OwnershipService, private readonly licenseMetricsRepository: LicenseMetricsRepository, private readonly logger: Logger, + private readonly eventService: EventService, outboundHttp: OutboundHttp, ) { this.logger = this.logger.scoped('instance-reporting'); @@ -153,8 +155,10 @@ export class InstanceReportingService { `Instance report was rejected with status ${response.statusCode}`, ); } + this.eventService.emit('instance-report-delivered'); } catch (error) { const message = error instanceof Error ? error.message : String(error); + this.eventService.emit('instance-report-failed'); await this.reportRepository.recordFailure(report.id, message, new Date()); // `recordFailure` incremented the count, so the in-memory row is one behind. diff --git a/packages/frontend/@n8n/i18n/src/locales/en.json b/packages/frontend/@n8n/i18n/src/locales/en.json index 2e5b70972c5..b26c3823a82 100644 --- a/packages/frontend/@n8n/i18n/src/locales/en.json +++ b/packages/frontend/@n8n/i18n/src/locales/en.json @@ -3840,6 +3840,8 @@ "settings.log-streaming.eventGroup.n8n.runner.info": "Will send an event when a Code node execution is requested from a task runner, and when a response is received from the runner with the result.", "settings.log-streaming.eventGroup.n8n.queue": "Queue events", "settings.log-streaming.eventGroup.n8n.queue.info": "Will send an event when a queue-related event occurs, e.g. enqueuing, dequeueing, completion, failure, or stalling.", + "settings.log-streaming.eventGroup.n8n.instanceReporting": "Instance Reporting", + "settings.log-streaming.eventGroup.n8n.instanceReporting.info": "Will send an event each time an instance usage report to the central monitoring service succeeds or fails.", "settings.log-streaming.eventName.n8n.audit.mcp.oauth.completed": "MCP OAuth completed", "settings.log-streaming.eventName.n8n.audit.mcp.tool.called": "MCP tool called", "settings.log-streaming.eventName.n8n.audit.mcp.access.updated": "MCP access updated", diff --git a/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.test.ts b/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.test.ts index fdf6ebc077c..ba2cfcd327b 100644 --- a/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.test.ts +++ b/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.test.ts @@ -98,4 +98,28 @@ describe('LogStreamingStore', () => { expect(nodeEventGroup!.indeterminate).toBe(false); }); }); + + describe('event group boundaries', () => { + it('should not put a longer group under one that only shares leading characters', () => { + logStreamingStore.addEventName('n8n.foo.created'); + logStreamingStore.addEventName('n8n.foobar.created'); + + logStreamingStore.addDestination({ + id: 'boundaryDestination', + label: 'Boundary Destination', + enabled: true, + subscribedEvents: [], + anonymizeAuditMessages: false, + }); + + const groups = logStreamingStore.items.boundaryDestination.eventGroups; + const fooGroup = groups.find((group) => group.name === 'n8n.foo'); + const foobarGroup = groups.find((group) => group.name === 'n8n.foobar'); + + // `n8n.foobar` starts with `n8n.foo`, but the two are distinct groups on a + // segment boundary: neither must swallow the other's events. + expect(fooGroup!.children.map((c) => c.name)).toEqual(['n8n.foo.created']); + expect(foobarGroup!.children.map((c) => c.name)).toEqual(['n8n.foobar.created']); + }); + }); }); diff --git a/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.ts b/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.ts index bdffa82f2af..092ad64d2a9 100644 --- a/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.ts +++ b/packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.ts @@ -78,7 +78,7 @@ const eventGroupsFromStringList = ( selected: selectionList.has(group), indeterminate: false, }; - const eventsOfGroup = eventNameArray.filter((e) => e.startsWith(group)); + const eventsOfGroup = eventNameArray.filter((e) => e === group || e.startsWith(`${group}.`)); for (const event of eventsOfGroup) { if (!collection.selected && selectionList.has(event)) { collection.indeterminate = true;