feat(core): Run instance reporting as a durable system task (no-changelog) (#39924)

This commit is contained in:
Lorent Lempereur
2026-10-01 10:45:56 +00:00
committed by GitHub
parent 58ddcd7bbb
commit 06b27d1461
18 changed files with 1519 additions and 987 deletions
-1
View File
@@ -359,7 +359,6 @@ export default defineConfig(
// tasks. NEVER add to this list — new periodic leader work must be a
// @SystemTask() class. Entries are removed as each migrates on its own ticket.
files: [
'./src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts',
'./src/services/pruning/executions-pruning.service.ts',
'./src/services/workflow-statistics-rollup.service.ts',
],
-1
View File
@@ -364,7 +364,6 @@ export default defineConfig({
// tasks. NEVER add to this list — new periodic leader work must be a
// @SystemTask() class. Entries are removed as each migrates on its own ticket.
files: [
'./src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts',
'./src/services/pruning/executions-pruning.service.ts',
'./src/services/workflow-statistics-rollup.service.ts',
],
@@ -4,7 +4,7 @@ How to prove, on one machine, that a dev n8n instance reports to a local
instance-reporting receiver on `http://127.0.0.1:3456`, and that the two data
points hold the numbers from `insights` and `workflow_statistics`.
The daily timer makes an unassisted end-to-end run take up to 24 hours. The
The daily report time makes an unassisted end-to-end run take up to 24 hours. The
plan removes the wait: the scheduler re-derives "is a report due" from the
database on every tick, so a report time in the past plus a restart forces the
report immediately (the catch-up path). Nothing in the module is changed for
@@ -49,7 +49,7 @@ export N8N_INSIGHTS_COMPACTION_INTERVAL_MINUTES=1
export N8N_INSIGHTS_FLUSH_INTERVAL_SECONDS=5
export N8N_LOG_LEVEL=debug
export N8N_LOG_SCOPES=instance-reporting
export N8N_LOG_SCOPES=instance-reporting,system-tasks
pnpm dev
```
@@ -58,7 +58,7 @@ Expected on the first boot:
- No warning about `N8N_INSTANCE_REPORTING_BASE_URL` being unset, and none
about a missing license certificate.
- The log line `Started the instance reporting timer`.
- A system task routing log with `name: instance-reporting`.
- A `Resolved the instance reporting time` line with a random `HH:mm` at or
after `03:00`.
@@ -186,18 +186,18 @@ Run these after step 4. Each is short.
| # | Scenario | Steps | Expected |
|---|---|---|---|
| 5.1 | No second report the same day | Restart the instance | No new request, no new row. `hasSettledToday` short-circuits the tick |
| 5.2 | Retry resends the same measurement | Make the receiver answer 500. Delete today's delivered row, then restart | Request arrives, `lastError` holds `rejected with status 500`, `deliveredAt` NULL, `attempts` grows. Retries land ~5 minutes apart, 3 attempts in total, then `Giving up on the instance report for today`. Every retry carries the **same** `batchId` and the same values — no re-measurement |
| 5.2 | Retry resends the same measurement | Make the receiver answer 500. Delete today's delivered row, then restart | Request arrives, `lastError` holds `rejected with status 500`, `deliveredAt` NULL, `attempts` grows. Retries land about 15 minutes apart, 3 attempts in total, then `Giving up on the instance report for today`. Every retry carries the **same** `batchId` and the same values — no re-measurement |
| 5.3 | Recovery keeps the pending row | During 5.2, switch the receiver back to 201 before the third attempt | The next attempt reuses the pending row and marks it delivered. No second row for the day |
| 5.4 | Untrusted certificate | Set `N8N_LICENSE_CERT` to a certificate the receiver does not trust (any well-formed one from another CA), clear today's row, restart | The receiver answers 401, delivery fails, `lastError` names the status. Nothing is marked delivered |
| 5.4b | No certificate | Unset `N8N_LICENSE_CERT`, clear today's row, restart | Warning `no license certificate, so no reports will be sent`; no timer, no request, no row |
| 5.4b | No certificate | Unset `N8N_LICENSE_CERT`, clear today's row, restart | Warning `no license certificate, so no reports will be sent`; no reporting task, no request, no row |
| 5.4c | Token set | Set `N8N_INSTANCE_REPORTING_AUTH_TOKEN` to a token the receiver accepts, unset `N8N_LICENSE_CERT`, clear today's row, restart | No certificate warning. The request carries `Authorization: Bearer <token>` and the body has no `licenseCert`. The receiver answers 201 and the row is marked delivered |
| 5.5 | Non-201 success code is a failure | Make the receiver answer 200 | Treated as a rejection: `Instance report was rejected with status 200` |
| 5.6 | Receiver down | Stop the receiver, clear today's row, restart | Delivery fails with a connection error in `lastError`; the instance stays healthy and keeps serving |
| 5.7 | Redirect is not followed | Make the receiver answer 302 to another local port | The report is rejected with status 302. The second port never sees the license certificate |
| 5.8 | Base URL unset | Unset `N8N_INSTANCE_REPORTING_BASE_URL`, restart | Warning `enabled but N8N_INSTANCE_REPORTING_BASE_URL is unset`; no timer, no request |
| 5.8 | Base URL unset | Unset `N8N_INSTANCE_REPORTING_BASE_URL`, restart | Warning `enabled but N8N_INSTANCE_REPORTING_BASE_URL is unset`; no reporting task, no request |
| 5.9 | Insights disabled | `N8N_DISABLED_MODULES=insights`, restart | Startup fails with the `UserError` that names both variables |
| 5.10 | Trailing slash in the base URL | Use `http://127.0.0.1:3456/`, clear today's row, restart | The path is still `/api/v1/instance-reports`, with no double slash |
| 5.11 | Slot still ahead | Set `reportTime` to a time later today, clear today's row, restart | No request. The log shows the timer armed for the remaining interval |
| 5.11 | Slot still ahead | Set `reportTime` to a time later today, clear today's row, restart | No request. The task checks the report time on each 15-minute pass |
`clear today's row` means:
@@ -232,10 +232,9 @@ want your dev data untouched.
A loopback address therefore works without extra configuration.
- **UTC everywhere.** The report time, the slot maths and the reported date are
all UTC. Convert before you read the local clock.
- **Multi-main.** Only the leader holds the timer. A single dev instance is
always the leader, so this plan cannot show handover. To cover it, run two
mains against the same Postgres with `N8N_MULTI_MAIN_SETUP_ENABLED=true` and
confirm only one request arrives per day, and that the follower starts the
timer after the leader stops.
- **Multi-main.** Enable both scheduler flags on two mains. Confirm that a
follower can deliver a claimed report. Let both mains claim the same due
occurrence and confirm that one request arrives. Disable either flag on both
mains and confirm that only the leader's in-memory task runs.
- **Request timeout.** Delivery gives up after 30 seconds. A receiver that
sleeps longer is a way to check the timeout path in 5.6.
@@ -64,6 +64,7 @@ without reading this document:
| `status` | Meaning |
|---|---|
| `pending` | Not delivered yet, and attempts remain |
| `sending` | One main holds the report while it sends a request |
| `delivered` | The receiver answered 201 |
| `skipped_after_max_retries` | The instance stopped delivering that day: after three failed attempts, because the report's own slot passed before it landed, or because the receiver rejected the payload (`400` or `413`) |
@@ -75,7 +76,7 @@ next one.
scheduler holding them in memory. A restart therefore resumes the same report's
three attempts instead of granting three more, and waits out the rest of the
five minutes since the last attempt before trying again — otherwise a crash loop
would spend the whole budget in seconds. `InstanceReportingScheduler` keeps no
would spend the whole budget in seconds. `InstanceReportingTask` keeps no
attempt state of its own.
The delivery retry above and the missed-day backfill are two separate
@@ -84,22 +85,37 @@ and where the logic lives.
## Scheduling
The daily fire is driven by `InstanceReportingScheduler`, a leader-gated
in-process timer (the same pattern as execution pruning and workflow history
compaction) rather than the durable scheduler: that framework has no
first-class support yet for system-owned jobs like this one, only for
workflow-triggered jobs, and this module is meant to move onto it once it does.
`InstanceReportingTask` checks the stored report time every 15 minutes. Changes
to that time apply on the next pass without a restart or a schedule update. A
report can start up to 15 minutes after its UTC slot. A pass sends a new report
for the latest slot at or before its time, unless the day of that slot is
settled. Yesterday's slot stays due for one hour, so a slot late in the UTC day
is sent after midnight. That report is still dated by its slot and ends on the
day before it.
In multi-main, only the leader holds the timer, so a cluster reports once
rather than once per main; leadership handover moves the timer along with it.
In place of the durability a scheduler-backed job would give:
With `N8N_SCHEDULER_ENABLED` and `N8N_SCHEDULER_SYSTEM_TASKS_ENABLED` enabled,
any main can claim the durable task. With either flag disabled, the shared
system task runner uses the leader's in-memory timer. It also runs a catch-up
pass at startup and on leader takeover. The runner starts after the server.
- **Catch-up.** Every tick asks the database whether today's report was
delivered, rather than trusting a timer fired at the right moment — so a
restart, or a leadership handover, that straddles the report time still
reports that day.
- **Bounded retry.** A failed delivery is retried a few times, a few minutes
apart, before the day is left to the next slot.
The report row owns retries in both modes. Each scheduler occurrence has one
attempt. A delivery failure is recorded as a failed occurrence. Later passes
read the report row and wait at least five minutes before retrying. During normal
operation, retries run on the next 15-minute pass. After three
failed deliveries, the row is skipped. At the pending row's next daily slot,
a new report replaces it even if the retry delay has not elapsed. A `sending`
row stays active until its request finishes. A stopped main's claim expires
after two minutes.
The unique `reportDate` key allows one report row per UTC day. Concurrent
creators cannot send different measurements for that day. Retries use the
persisted data and `batchId`. If delivery succeeds before a main stops, a repeat
uses that same batch. The receiver answers `409`, which counts as delivery.
A database claim lets only one main send a pending row at a time. Claim age
uses the database clock. A failure write must match the active claim timestamp.
A late `201` or `409` marks any undelivered row as delivered, even if another
main reclaimed or skipped it. Further results leave a delivered row and its
attempt count unchanged.
## Enabling
@@ -150,7 +166,7 @@ host.
| Env var | Default | Notes |
|---|---|---|
| `N8N_INSTANCE_REPORTING_BASE_URL` | `''` | Base URL of the receiver. The report is POSTed to `<base>/api/v1/instance-reports`. Left unset, the module loads but warns and never sends: it starts no scheduler and claims no report time. |
| `N8N_INSTANCE_REPORTING_BASE_URL` | `''` | Base URL of the receiver. The report is POSTed to `<base>/api/v1/instance-reports`. Left unset, the module loads but warns and never sends: it registers no reporting task and claims no report time. |
| `N8N_INSTANCE_REPORTING_LABEL` | `''` | Sent as `label` in the payload, when set. |
| `N8N_INSTANCE_REPORTING_AUTH_TOKEN` | `''` | Sent as `Authorization: Bearer …`, when set. Replaces the license certificate as the credential; `licenseCert` is then omitted from the body. |
| `N8N_LICENSE_CERT` | `''` | Not owned by this module. Its value, or the persisted certificate of an activated license, is sent as `licenseCert` and is the credential the receiver checks, unless a token is set. |
@@ -2,7 +2,7 @@
The module has two retry mechanisms. The code keeps them separate. Both feed
the same state machine on `instance_monitoring_report`: a row goes from
`pending` to `delivered`, or from `pending` to `skipped_after_max_retries`.
`pending` through `sending` to `delivered`, or from `pending` to `skipped_after_max_retries`.
- **Type 1, delivery retry.** Resend the pending row until it lands, its budget
runs out, or its own next slot passes. It continues across the UTC midnight
@@ -15,32 +15,38 @@ takes over and delivers that row.
## The send decision
`start()` fires the first `tick()` (on `server-started`, or on leader takeover).
Every pass then re-arms the next with `setTimeout`, so the loop drives itself.
Each pass makes one choice, from `findPending()` — the newest row, and only if
that row is still pending:
`InstanceReportingTask.run()` makes one pass every 15 minutes. The shared system
task runner owns scheduling and leadership. Each pass reads the current report
time and the newest report row. Only a pending row can be retried. A `sending`
row waits for its request to finish, including when the next slot passes. A
claim left by a stopped main returns to `pending` on the first pass after two
minutes. Claim age uses the database clock. A row with no claim timestamp can
be released immediately. Failure writes must match the active claim. A late
`201` or `409` still records delivery after a reclaim or skip. Further results
cannot change a delivered row.
```mermaid
flowchart TD
S(["start(): server-started / leader takeover"]) --> T
T["tick()"] --> P{"waiting between retries?<br/>(up to 5 min, but not past its slot)"}
P -- yes --> ARM
P -- no --> L["report = findPending()"]
L --> Q1{"is there a report to resume?"}
Q1 -- yes --> Q1b{"too old to still send?"}
Q1b -- no --> R["resend it — retry, even past midnight"]
Q1b -- yes --> SK["mark it skipped"] --> Q2
Q1 -- "no — delivered, skipped, or none" --> Q2{"reached today's report time?"}
Q2 -- no --> N["do nothing — wait"]
Q2 -- yes --> Q3{"already reported today?"}
T["System task pass"] --> L["Read report time and newest pending row"]
L --> Q1{"Pending report?"}
Q1 -- yes --> Q1b{"Its next slot passed?"}
Q1b -- no --> P{"Five minutes since last attempt?"}
P -- no --> N["Return"]
P -- yes --> R["Resend stored batch, even after midnight"]
Q1b -- yes --> SK["Mark it skipped"] --> Q2
Q1 -- no --> Q2{"Today's slot passed, or yesterday's within one hour?"}
Q2 -- no --> N
Q2 -- yes --> Q3{"Day settled?"}
Q3 -- yes --> N
Q3 -- no --> C["create a new report"]
R --> ARM
C --> ARM
N --> ARM
ARM["scheduleNext() — arms setTimeout(tick)"] -. re-arms .-> T
Q3 -- no --> C["Create and send a report"]
R --> N
C --> N
```
The report row owns delivery retries. Durable occurrences use `maxAttempts: 1`.
A delivery error fails that occurrence. The next scheduled pass checks the
stored retry delay and budget. It does not get a new delivery budget.
## Type 1: delivery retry
Scope: the newest row while it is `pending`, whose last delivery attempt failed.
@@ -51,8 +57,9 @@ next slot has passed, the retry stops: the row is marked
- The row is created once, as `pending`, with its data points already
measured. A retry changes only `attempts`, `lastAttemptAt`, `lastError` and,
in the end, `status`. The data points and the `batchId` stay the same.
- `InstanceReportingScheduler.msUntilRetryAllowed()` reads `lastAttemptAt` from
the row. The scheduler waits `RETRY_DELAY_MS` (5 minutes) between attempts.
- `InstanceReportingTask.run()` reads `lastAttemptAt` from the row. A retry
waits at least `RETRY_DELAY_MS` (5 minutes), then runs on the next 15-minute
pass. The five-minute floor also applies after a restart or takeover.
- The budget is `MAX_ATTEMPTS` (3). The attempt that spends the last one flips
the row to `skipped_after_max_retries` at once. There is no fourth attempt,
and no new row for today.
@@ -124,14 +131,14 @@ gap in delivered rows.
```mermaid
flowchart TD
subgraph T1["Type 1: delivery retry (max 3 attempts, 5 min apart, crosses midnight)"]
subgraph T1["Type 1: delivery retry (max 3 attempts, 15-minute passes, crosses midnight)"]
A["Newest row is pending\n+ measured data points"] --> B["POST attempt"]
B -->|"201 / 409"| C["delivered"]
B -->|"failure"| D["attempts++\nlastAttemptAt, lastError"]
D --> R{"400 / 413?"}
R -- yes --> G
R -- no --> E{"attempts >= 3?"}
E -- no --> F["wait 5 min"] --> B
E -- no --> F["wait for next pass, at least 5 min"] --> B
E -- yes --> G["skipped_after_max_retries"]
end
@@ -0,0 +1,402 @@
import { createFakeOutboundHttp, type Route } from '@n8n/backend-network/testing';
import type { EventService } from '@n8n/backend-services';
import { mockLogger, testDb, testModules } from '@n8n/backend-test-utils';
import { Time } from '@n8n/constants';
import type { LicenseMetricsRepository } from '@n8n/db';
import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import { createScheduler, type Scheduler, type SchedulerPasses } from '@n8n/scheduler';
import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import { Tracing, type InstanceSettings } from 'n8n-core';
import { mock } from 'vitest-mock-extended';
import type { License } from '@/license';
import { InsightsConfig } from '@/modules/insights/insights.config';
import type { InsightsService } from '@/modules/insights/insights.service';
import { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner';
import { buildMaterializerTransaction } from '@/scheduling/durable-scheduler';
import { SystemTaskHandler } from '@/scheduling/system-tasks/system-task-handler';
import { systemTaskProvisionRequest } from '@/scheduling/system-tasks/system-task-job-registrar';
import { SystemTaskScheduledJobOwner } from '@/scheduling/system-tasks/system-task-scheduled-job-owner';
import { retryUntil } from '@test-integration/retry-until';
import { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository';
import type { InstanceReportingSettingsService } from '../instance-reporting-settings.service';
import { InstanceReportingConfig } from '../instance-reporting.config';
import { INSTANCE_REPORTS_PATH } from '../instance-reporting.constants';
import { InstanceReportingService } from '../instance-reporting.service';
import { InstanceReportingTask } from '../instance-reporting.task';
const TASK_TYPE = 'system:instance-reporting';
const REPORT_DAY = new Date(Date.now() - Time.days.toMilliseconds).toISOString().slice(0, 10);
const accepted: Route = { method: 'POST', pathname: INSTANCE_REPORTS_PATH, status: 201 };
describe('instance reporting across mains', () => {
let reports: InstanceMonitoringReportRepository;
let jobs: ScheduledJobRepository;
let occurrences: ScheduledTaskRepository;
let schedulers: Array<Scheduler & SchedulerPasses> = [];
beforeAll(async () => {
await testModules.loadModules(['insights', 'instance-reporting']);
await testDb.init();
reports = Container.get(InstanceMonitoringReportRepository);
jobs = Container.get(ScheduledJobRepository);
occurrences = Container.get(ScheduledTaskRepository);
});
beforeEach(async () => {
await testDb.truncate(['InstanceMonitoringReport', 'ScheduledTask', 'ScheduledJob']);
// Keep the database clock aligned with the simulated report slots in these tests.
vi.spyOn(reports, 'readDbNow').mockImplementation(async () => new Date());
});
afterEach(async () => {
await Promise.all(schedulers.map(async (scheduler) => await scheduler.stop()));
schedulers = [];
vi.useRealTimers();
vi.restoreAllMocks();
});
afterAll(async () => await testDb.terminate());
function buildMain(hostId: string, isLeader: boolean, routes: Route[] = [accepted]) {
const { outboundHttp, httpRequest } = createFakeOutboundHttp(
routes,
vi.fn as unknown as Parameters<typeof createFakeOutboundHttp>[1],
);
const config = new InstanceReportingConfig();
config.instanceReportingBaseUrl = 'https://receiver.test';
const insights = mock<InsightsService>();
insights.getDailyExecutionTotals.mockResolvedValue(new Map([[REPORT_DAY, 5]]));
const metrics = mock<LicenseMetricsRepository>();
metrics.getLicenseRenewalMetrics.mockResolvedValue(
mock<Awaited<ReturnType<LicenseMetricsRepository['getLicenseRenewalMetrics']>>>({
productionRootExecutions: 10,
}),
);
const events = mock<EventService>();
const service = new InstanceReportingService(
config,
reports,
mock<InstanceReportingSettingsService>({ getReportTime: async () => '00:00' }),
insights,
new InsightsConfig(),
mock<InstanceSettings>({ instanceId: 'shared-instance', instanceType: 'main', isLeader }),
metrics,
mock<License>({ loadCertStr: async () => 'license-cert' }),
mockLogger(),
events,
outboundHttp,
);
const task = new InstanceReportingTask(service);
const scheduler = createScheduler({
hostId,
materializerTransaction: buildMaterializerTransaction(
Container.get(DataSource),
jobs,
occurrences,
),
taskStore: occurrences,
executor: { leaseSeconds: 30, lookaheadSeconds: 0, batchSize: 1 },
});
scheduler.registerTaskHandler(
TASK_TYPE,
new SystemTaskHandler(
task,
new AbortController().signal,
mockLogger(),
events,
new Tracing(),
vi.fn(),
),
);
schedulers.push(scheduler);
return { task, scheduler, httpRequest, insights };
}
async function provision(task: InstanceReportingTask) {
await Container.get(DurableJobProvisioner).provision(
systemTaskProvisionRequest(
task,
Container.get(SystemTaskScheduledJobOwner),
'UTC',
new Date(),
),
);
return await jobs.findOneByOrFail({ name: TASK_TYPE });
}
async function expectOccurrenceStatus(status: 'succeeded' | 'failed') {
await retryUntil(
async () => {
expect(await occurrences.countBy({ status })).toBe(1);
},
{ timeoutMs: 5000 },
);
}
it('sends once when two mains claim the same due occurrence', async () => {
const leader = buildMain('main-a', true);
const follower = buildMain('main-b', false);
await provision(leader.task);
await provision(follower.task);
expect(await jobs.count()).toBe(1);
const claims = await Promise.all([leader.scheduler.execute(), follower.scheduler.execute()]);
expect(claims.flat()).toHaveLength(1);
await expectOccurrenceStatus('succeeded');
expect(leader.httpRequest.mock.calls.length + follower.httpRequest.mock.calls.length).toBe(1);
expect(await reports.countBy({ status: 'delivered' })).toBe(1);
await Promise.all([leader.task.run(), follower.task.run()]);
expect(leader.httpRequest.mock.calls.length + follower.httpRequest.mock.calls.length).toBe(1);
});
it('delivers on a follower without a leader check', async () => {
const follower = buildMain('main-b', false);
await provision(follower.task);
await follower.scheduler.execute();
await expectOccurrenceStatus('succeeded');
expect(follower.httpRequest).toHaveBeenCalledTimes(1);
expect(await occurrences.findOneByOrFail({ status: 'succeeded' })).toMatchObject({
claimedBy: 'main-b',
});
});
it('records a failed occurrence once and resumes the report on another main after five minutes', async () => {
const leader = buildMain('main-a', true, [{ ...accepted, status: 503 }]);
const job = await provision(leader.task);
await leader.scheduler.execute();
await expectOccurrenceStatus('failed');
expect(await occurrences.findOneByOrFail({ jobId: job.id })).toMatchObject({
attempts: 1,
maxAttempts: 1,
});
const [pending] = await reports.find();
expect(pending).toMatchObject({ status: 'pending', attempts: 1 });
const follower = buildMain('main-b', false);
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(pending.lastAttemptAt!.getTime() + Time.minutes.toMilliseconds);
await follower.task.run();
expect(follower.httpRequest).not.toHaveBeenCalled();
vi.setSystemTime(pending.lastAttemptAt!.getTime() + 5 * Time.minutes.toMilliseconds);
await follower.task.run();
expect(follower.httpRequest).toHaveBeenCalledTimes(1);
expect(follower.httpRequest.mock.calls[0][0].body).toMatchObject({
batchId: pending.id,
dataPoints: pending.dataPoints,
});
expect(await reports.findOneByOrFail({ id: pending.id })).toMatchObject({
status: 'delivered',
attempts: 2,
});
});
it('keeps one batch when two passes collect different measurements at once', async () => {
const leader = buildMain('main-a', true);
const follower = buildMain('main-b', false);
let arrived = 0;
const barrier = createDeferredPromise();
for (const [index, main] of [leader, follower].entries()) {
main.insights.getDailyExecutionTotals.mockImplementation(async () => {
if (++arrived === 2) barrier.resolve();
await barrier.promise;
return new Map([[REPORT_DAY, index + 1]]);
});
}
await Promise.all([leader.task.run(), follower.task.run()]);
expect(await reports.count()).toBe(1);
const calls = [...leader.httpRequest.mock.calls, ...follower.httpRequest.mock.calls];
expect(calls).toHaveLength(1);
const [report] = await reports.find();
expect(calls[0][0].body).toMatchObject({ batchId: report.id, dataPoints: report.dataPoints });
});
it('resends the same batch after the receiver accepted a report before a main stopped', async () => {
const pending = await reports.createPending(
[{ kind: 'daily', name: 'billableExecutions', value: 5, date: REPORT_DAY }],
new Date(),
);
const follower = buildMain('main-b', false, [{ ...accepted, status: 409 }]);
await provision(follower.task);
await follower.scheduler.execute();
await expectOccurrenceStatus('succeeded');
expect(follower.httpRequest.mock.calls[0][0].body).toMatchObject({
batchId: pending?.id,
dataPoints: pending?.dataPoints,
});
expect(await reports.countBy({ status: 'delivered' })).toBe(1);
});
it('waits for an in-flight send before expiring its report at the next slot', async () => {
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(new Date('2026-03-26T23:59:50.000Z'));
const pending = await reports.createPending(
[{ kind: 'daily', name: 'billableExecutions', value: 5, date: '2026-03-25' }],
new Date(),
);
if (!pending) throw new Error('Could not create the pending report');
await reports.update({ id: pending.id }, { createdAt: new Date() });
const leader = buildMain('main-a', true);
const follower = buildMain('main-b', false);
const requestStarted = createDeferredPromise();
const releaseResponse = createDeferredPromise();
vi.mocked(leader.httpRequest).mockImplementation(async () => {
requestStarted.resolve();
await releaseResponse.promise;
return { statusCode: 201, body: undefined, headers: {} };
});
const firstPass = leader.task.run();
await requestStarted.promise;
vi.setSystemTime(new Date('2026-03-27T00:00:10.000Z'));
await follower.task.run();
expect(await reports.count()).toBe(1);
expect(follower.httpRequest).not.toHaveBeenCalled();
releaseResponse.resolve();
await firstPass;
expect(await reports.findOneByOrFail({ id: pending.id })).toMatchObject({
status: 'delivered',
});
await follower.task.run();
expect(follower.httpRequest).toHaveBeenCalledTimes(1);
expect(await reports.count()).toBe(2);
});
it('resumes a claim left by a stopped main on a later pass', async () => {
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(new Date('2026-03-26T07:43:00.000Z'));
const pending = await reports.createPending(
[{ kind: 'daily', name: 'billableExecutions', value: 5, date: '2026-03-25' }],
new Date(),
);
if (!pending) throw new Error('Could not create the pending report');
await reports.update({ id: pending.id }, { createdAt: new Date() });
await reports.claimForSend(pending.id);
const follower = buildMain('main-b', false);
vi.setSystemTime(new Date('2026-03-26T07:46:00.000Z'));
await follower.task.run();
expect(follower.httpRequest).not.toHaveBeenCalled();
expect(await reports.findOneByOrFail({ id: pending.id })).toMatchObject({ status: 'pending' });
vi.setSystemTime(new Date('2026-03-26T07:48:00.000Z'));
await follower.task.run();
expect(follower.httpRequest).toHaveBeenCalledTimes(1);
expect(follower.httpRequest.mock.calls[0][0].body).toMatchObject({ batchId: pending.id });
expect(await reports.findOneByOrFail({ id: pending.id })).toMatchObject({
status: 'delivered',
});
});
it('records a late 201 after another main reclaims the batch', async () => {
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(new Date('2026-03-26T07:43:00.000Z'));
const pending = await reports.createPending(
[{ kind: 'daily', name: 'billableExecutions', value: 5, date: '2026-03-25' }],
new Date(),
);
if (!pending) throw new Error('Could not create the pending report');
await reports.update({ id: pending.id }, { createdAt: new Date() });
const leader = buildMain('main-a', true);
const follower = buildMain('main-b', false);
const firstRequestStarted = createDeferredPromise();
const releaseFirstResponse = createDeferredPromise();
const secondRequestStarted = createDeferredPromise();
const releaseSecondResponse = createDeferredPromise();
vi.mocked(leader.httpRequest).mockImplementationOnce(async () => {
firstRequestStarted.resolve();
await releaseFirstResponse.promise;
return { statusCode: 201, body: undefined, headers: {} };
});
vi.mocked(follower.httpRequest).mockImplementationOnce(async () => {
secondRequestStarted.resolve();
await releaseSecondResponse.promise;
throw new Error('Reclaimed request failed');
});
const firstPass = leader.task.run();
await firstRequestStarted.promise;
vi.setSystemTime(new Date('2026-03-26T07:58:00.000Z'));
const secondResult = expect(follower.task.run()).rejects.toThrow('Reclaimed request failed');
await secondRequestStarted.promise;
expect(follower.httpRequest.mock.calls[0][0].body).toEqual(
leader.httpRequest.mock.calls[0][0].body,
);
expect(await reports.findOneByOrFail({ id: pending.id })).toMatchObject({
status: 'sending',
lastAttemptAt: new Date(),
});
releaseFirstResponse.resolve();
await firstPass;
const delivered = await reports.findOneByOrFail({ id: pending.id });
releaseSecondResponse.resolve();
await secondResult;
expect(delivered).toMatchObject({ status: 'delivered', attempts: 1 });
expect(await reports.findOneByOrFail({ id: pending.id })).toEqual(delivered);
vi.setSystemTime(new Date('2026-03-26T08:13:00.000Z'));
await follower.task.run();
expect(follower.httpRequest).toHaveBeenCalledTimes(1);
vi.setSystemTime(new Date('2026-03-27T07:43:00.000Z'));
await follower.task.run();
expect(follower.httpRequest).toHaveBeenCalledTimes(2);
expect(follower.httpRequest.mock.calls[1][0].body).toMatchObject({
dataPoints: [
expect.objectContaining({ kind: 'cumulative' }),
expect.objectContaining({ kind: 'daily', date: '2026-03-26' }),
],
});
});
it.each(['success', 'failure'])(
'ignores a late %s after another main delivers the batch',
async (outcome) => {
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(new Date('2026-03-26T07:43:00.000Z'));
const pending = await reports.createPending(
[{ kind: 'daily', name: 'billableExecutions', value: 5, date: '2026-03-25' }],
new Date(),
);
if (!pending) throw new Error('Could not create the pending report');
await reports.update({ id: pending.id }, { createdAt: new Date() });
const leader = buildMain('main-a', true);
const follower = buildMain('main-b', false);
const requestStarted = createDeferredPromise();
const releaseResponse = createDeferredPromise();
vi.mocked(leader.httpRequest).mockImplementation(async () => {
requestStarted.resolve();
await releaseResponse.promise;
if (outcome === 'failure') throw new Error('Late response failed');
return { statusCode: 201, body: undefined, headers: {} };
});
const firstPass = leader.task.run();
const firstResult =
outcome === 'failure'
? expect(firstPass).rejects.toThrow('Late response failed')
: expect(firstPass).resolves.toBeUndefined();
await requestStarted.promise;
vi.setSystemTime(new Date('2026-03-26T07:58:00.000Z'));
await follower.task.run();
const delivered = await reports.findOneByOrFail({ id: pending.id });
expect(delivered).toMatchObject({ status: 'delivered', attempts: 1 });
releaseResponse.resolve();
await firstResult;
expect(await reports.findOneByOrFail({ id: pending.id })).toEqual(delivered);
await follower.task.run();
expect(follower.httpRequest).toHaveBeenCalledTimes(1);
},
);
});
@@ -1,330 +0,0 @@
import type { EventService } from '@n8n/backend-services';
import { mockLogger } from '@n8n/backend-test-utils';
import { Time } from '@n8n/constants';
import type { InstanceSettings } from 'n8n-core';
import type { Mocked } from 'vitest';
import { mock } from 'vitest-mock-extended';
import type { InstanceMonitoringReport } from '../database/entities/instance-monitoring-report';
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';
import type { InstanceReportingService } from '../instance-reporting.service';
const REPORT_TIME = '07:42';
/** Just before the day's slot, so nothing is due yet. */
const BEFORE_SLOT = '2026-03-26T07:41:00.000Z';
/** Just after the day's slot. */
const AFTER_SLOT = '2026-03-26T07:43:00.000Z';
interface Harness {
scheduler: InstanceReportingScheduler;
reportingService: Mocked<InstanceReportingService>;
reportRepository: Mocked<InstanceMonitoringReportRepository>;
settingsService: Mocked<InstanceReportingSettingsService>;
instanceSettings: Mocked<InstanceSettings>;
eventService: Mocked<EventService>;
/** Fire the deferred `server-started` listeners (only needed when autoStart is false). */
startServer: () => void;
}
function makeHarness({
isLeader = true,
// 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 reportingService = mock<InstanceReportingService>();
const reportRepository = mock<InstanceMonitoringReportRepository>();
reportRepository.hasSettledToday.mockResolvedValue(false);
const settingsService = mock<InstanceReportingSettingsService>();
settingsService.getReportTime.mockResolvedValue(REPORT_TIME);
const instanceSettings = mock<InstanceSettings>({
instanceType: 'main',
instanceRole: isLeader ? 'leader' : 'follower',
isLeader,
});
const serverStartedListeners: Array<() => void> = [];
const eventService = mock<EventService>();
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(
reportingService,
reportRepository,
settingsService,
instanceSettings,
eventService,
mockLogger(),
);
return {
scheduler,
reportingService,
reportRepository,
settingsService,
instanceSettings,
eventService,
startServer: () => serverStartedListeners.forEach((listener) => listener()),
};
}
/** Let the detached first tick settle before asserting. */
async function settle() {
await vi.advanceTimersByTimeAsync(0);
}
describe('InstanceReportingScheduler', () => {
beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(new Date(BEFORE_SLOT));
});
afterEach(() => {
vi.clearAllTimers();
vi.useRealTimers();
});
describe('init', () => {
test('does not start a timer on a follower', async () => {
const { scheduler, reportingService } = makeHarness({ isLeader: false });
scheduler.init();
await vi.advanceTimersByTimeAsync(Time.days.toMilliseconds);
expect(reportingService.sendReport).not.toHaveBeenCalled();
});
test('fires at the configured time on the leader', async () => {
const { scheduler, reportingService } = makeHarness();
scheduler.init();
await settle();
expect(reportingService.sendReport).not.toHaveBeenCalled();
// One minute short of the slot, then past it.
await vi.advanceTimersByTimeAsync(Time.minutes.toMilliseconds);
expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
});
test('fires once a day thereafter', async () => {
const { scheduler, reportingService } = makeHarness();
scheduler.init();
await vi.advanceTimersByTimeAsync(3 * Time.days.toMilliseconds);
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 });
scheduler.init();
await settle();
expect(reportingService.sendReport).not.toHaveBeenCalled();
startServer();
await settle();
expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
});
});
describe('catch-up', () => {
test("reports immediately when the day's slot already passed unreported", async () => {
vi.setSystemTime(new Date(AFTER_SLOT));
const { scheduler, reportingService } = makeHarness();
scheduler.init();
await settle();
expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
});
test("stays quiet when the day's report was already delivered", async () => {
vi.setSystemTime(new Date(AFTER_SLOT));
const { scheduler, reportingService, reportRepository } = makeHarness();
reportRepository.hasSettledToday.mockResolvedValue(true);
scheduler.init();
await settle();
expect(reportingService.sendReport).not.toHaveBeenCalled();
});
test('a caught-up day does not fire again later the same day', async () => {
vi.setSystemTime(new Date(AFTER_SLOT));
const { scheduler, reportingService, reportRepository } = makeHarness();
scheduler.init();
await settle();
// The catch-up delivered it, so the rest of the day has nothing to do.
reportRepository.hasSettledToday.mockResolvedValue(true);
await vi.advanceTimersByTimeAsync(Time.hours.toMilliseconds * 12);
expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
});
});
describe('leadership', () => {
test('starts reporting when this main takes over', async () => {
vi.setSystemTime(new Date(AFTER_SLOT));
const { scheduler, reportingService, instanceSettings } = makeHarness({ isLeader: false });
scheduler.init();
await settle();
expect(reportingService.sendReport).not.toHaveBeenCalled();
// Takeover: the now-leader main finds the day unreported and catches up.
// `start` is called directly here rather than through the multi-main event
// bus `@OnLeaderTakeover` wraps, since that plumbing is out of scope for a
// unit test; the decorator's only job is to invoke this same method.
Object.defineProperty(instanceSettings, 'isLeader', { get: () => true, configurable: true });
scheduler.start();
await settle();
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 });
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();
scheduler.init();
await settle();
scheduler.stop();
await vi.advanceTimersByTimeAsync(2 * Time.days.toMilliseconds);
expect(reportingService.sendReport).not.toHaveBeenCalled();
});
test('shutdown stops the timer and blocks a later takeover', async () => {
const { scheduler, reportingService } = makeHarness();
scheduler.init();
await settle();
scheduler.shutdown();
scheduler.start();
await vi.advanceTimersByTimeAsync(2 * Time.days.toMilliseconds);
expect(reportingService.sendReport).not.toHaveBeenCalled();
});
});
describe('retry', () => {
test('re-attempts a failed delivery a few minutes later', async () => {
vi.setSystemTime(new Date(AFTER_SLOT));
const { scheduler, reportingService } = makeHarness();
reportingService.sendReport.mockRejectedValueOnce(new Error('Network error'));
scheduler.init();
await settle();
expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(5 * Time.minutes.toMilliseconds);
expect(reportingService.sendReport).toHaveBeenCalledTimes(2);
});
test('stops retrying once the report has run out of attempts', async () => {
// The row owns the budget, so the service settles the day and this only
// has to notice. Three attempts, then the row reads as settled.
vi.setSystemTime(new Date(AFTER_SLOT));
const { scheduler, reportingService, reportRepository } = makeHarness();
reportingService.sendReport.mockRejectedValue(new Error('Network error'));
reportRepository.hasSettledToday.mockImplementation(
async () => reportingService.sendReport.mock.calls.length >= 3,
);
scheduler.init();
await vi.advanceTimersByTimeAsync(Time.hours.toMilliseconds);
expect(reportingService.sendReport).toHaveBeenCalledTimes(3);
});
test('waits out the remaining delay after a restart mid-retry', async () => {
// A crash between attempts would otherwise let the fresh process attempt at
// once, and a crash loop would spend the whole budget in seconds.
vi.setSystemTime(new Date(AFTER_SLOT));
const { scheduler, reportingService, reportRepository } = makeHarness();
reportRepository.findPending.mockResolvedValue(
mock<InstanceMonitoringReport>({
createdAt: new Date(AFTER_SLOT),
lastAttemptAt: new Date(Date.now() - 3 * Time.minutes.toMilliseconds),
}),
);
scheduler.init();
await settle();
expect(reportingService.sendReport).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(2 * Time.minutes.toMilliseconds);
expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
});
test('retries when the report time cannot even be resolved', async () => {
const { scheduler, settingsService } = makeHarness();
settingsService.getReportTime.mockRejectedValueOnce(new Error('DB down'));
scheduler.init();
await settle();
expect(settingsService.getReportTime).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(5 * Time.minutes.toMilliseconds);
expect(settingsService.getReportTime).toHaveBeenCalledTimes(2);
});
});
describe('clock changes', () => {
test('a fire before the slot reports nothing and re-arms', async () => {
const { scheduler, reportingService } = makeHarness();
scheduler.init();
await settle();
// Clock jumps backwards an hour, well before the slot.
vi.setSystemTime(new Date('2026-03-26T06:41:00.000Z'));
await vi.advanceTimersByTimeAsync(Time.minutes.toMilliseconds);
expect(reportingService.sendReport).not.toHaveBeenCalled();
});
});
});
@@ -12,7 +12,6 @@ import { LicenseMetricsRepository, SettingsRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import { DateTime } from 'luxon';
import { InstanceSettings } from 'n8n-core';
import type { MockInstance } from 'vitest';
import { mock } from 'vitest-mock-extended';
import type { License } from '@/license';
@@ -25,7 +24,7 @@ import type {
InstanceReportDataPoint,
} from '../database/entities/instance-monitoring-report';
import { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository';
import { InstanceReportingScheduler } from '../instance-reporting-scheduler.service';
import { InstanceReportingTask } from '../instance-reporting.task';
import { InstanceReportingSettingsService } from '../instance-reporting-settings.service';
import { InstanceReportingConfig } from '../instance-reporting.config';
import {
@@ -48,9 +47,6 @@ const NEXT_SLOT = '2026-03-27T07:42:00.000Z';
const MAX_ATTEMPTS = 3;
const RETRY_DELAY_MS = 5 * Time.minutes.toMilliseconds;
/** Room for the fake clock's drift while `vi.waitFor` polls. */
const CLOCK_TOLERANCE_MS = 10 * Time.seconds.toMilliseconds;
interface ReportPayload {
batchId: string;
dataPoints: InstanceReportDataPoint[];
@@ -66,24 +62,13 @@ const accepted = (): Route => ({ method: 'POST', pathname: INSTANCE_REPORTS_PATH
const tooLarge = (): Route => ({ method: 'POST', pathname: INSTANCE_REPORTS_PATH, status: 413 });
/** The private arming hook, spied on so a test can tell when a pass has finished. */
type SchedulerInternals = { scheduleNext(delayMs: number): void };
interface Harness {
scheduler: InstanceReportingScheduler;
/** Every resolved request the fake receiver saw. */
task: InstanceReportingTask;
httpRequest: ReturnType<typeof createFakeOutboundHttp>['httpRequest'];
/**
* Called once a pass has finished its last database write, so its call count
* marks "pass over, next timer armed". Pending fake timers cannot serve as
* that signal: the SQLite pool arms a transient timeout for every query.
*/
scheduleNext: MockInstance<SchedulerInternals['scheduleNext']>;
}
describe('instance reporting retries', () => {
let repository: InstanceMonitoringReportRepository;
let schedulers: InstanceReportingScheduler[] = [];
beforeAll(async () => {
await testModules.loadModules(['insights', 'instance-reporting']);
@@ -113,15 +98,12 @@ describe('instance reporting retries', () => {
{},
);
// Only what the scheduler uses. `setImmediate` and `nextTick` stay real so
// the database driver's I/O still completes under the fake clock.
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] });
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(new Date(AFTER_SLOT));
vi.spyOn(repository, 'readDbNow').mockImplementation(async () => new Date());
});
afterEach(() => {
for (const scheduler of schedulers) scheduler.shutdown();
schedulers = [];
vi.useRealTimers();
vi.restoreAllMocks();
});
@@ -141,6 +123,7 @@ describe('instance reporting retries', () => {
const service = new InstanceReportingService(
Container.get(InstanceReportingConfig),
repository,
Container.get(InstanceReportingSettingsService),
Container.get(InsightsService),
Container.get(InsightsConfig),
instanceSettings,
@@ -150,34 +133,9 @@ describe('instance reporting retries', () => {
eventService,
outboundHttp,
);
const scheduler = new InstanceReportingScheduler(
service,
repository,
Container.get(InstanceReportingSettingsService),
instanceSettings,
eventService,
mockLogger(),
);
schedulers.push(scheduler);
const task = new InstanceReportingTask(service);
const scheduleNext = vi.spyOn(scheduler as unknown as SchedulerInternals, 'scheduleNext');
return { scheduler, httpRequest, scheduleNext };
}
/** Wait until the scheduler has completed `passes` passes and armed the next one. */
async function armed({ scheduleNext }: Harness, passes: number) {
await vi.waitFor(() => expect(scheduleNext).toHaveBeenCalledTimes(passes), {
timeout: 5_000,
});
}
/** Assert the delay the scheduler last armed lands on `slot`. */
function expectArmedFor({ scheduleNext }: Harness, slot: string) {
const delay = scheduleNext.mock.lastCall?.[0] ?? Number.NaN;
const target = new Date(slot).getTime();
expect(Date.now() + delay).toBeGreaterThanOrEqual(target - CLOCK_TOLERANCE_MS);
expect(Date.now() + delay).toBeLessThanOrEqual(target + CLOCK_TOLERANCE_MS);
return { task, httpRequest };
}
function sentPayload({ httpRequest }: Harness, index: number): ReportPayload {
@@ -208,6 +166,8 @@ describe('instance reporting retries', () => {
{ kind: 'daily', name: 'billableExecutions', value: 3, date: day },
],
);
const claimedAt = await repository.claimForSend(report.id);
if (!claimedAt) throw new Error('Could not claim the report');
await repository.markDelivered(report.id, new Date());
return report;
}
@@ -270,8 +230,7 @@ describe('instance reporting retries', () => {
test('delivers on the third attempt once the receiver is reachable again', async () => {
const harness = makeHarness([unreachable(), unreachable(), accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await expect(harness.task.run()).rejects.toThrow();
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
await expect(repository.count()).resolves.toBe(1);
@@ -279,20 +238,18 @@ describe('instance reporting retries', () => {
expect(report).toMatchObject({ status: 'pending', attempts: 1 });
expect(report.lastError).toEqual(expect.any(String));
expect(report.lastAttemptAt).toEqual(expect.any(Date));
expect(harness.scheduleNext).toHaveBeenLastCalledWith(RETRY_DELAY_MS);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 2);
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await expect(harness.task.run()).rejects.toThrow();
expect(harness.httpRequest).toHaveBeenCalledTimes(2);
await expect(repository.findOneByOrFail({ id: report.id })).resolves.toMatchObject({
status: 'pending',
attempts: 2,
});
expect(harness.scheduleNext).toHaveBeenLastCalledWith(RETRY_DELAY_MS);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 3);
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(3);
await expect(repository.findOneByOrFail({ id: report.id })).resolves.toMatchObject({
@@ -310,10 +267,9 @@ describe('instance reporting retries', () => {
}
// The day is settled, so the scheduler now waits for tomorrow's slot.
expectArmedFor(harness, NEXT_SLOT);
await vi.advanceTimersByTimeAsync(Time.hours.toMilliseconds);
vi.setSystemTime(Date.now() + Time.hours.toMilliseconds);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(3);
expect(harness.scheduleNext).toHaveBeenCalledTimes(3);
await expect(repository.count()).resolves.toBe(1);
});
@@ -324,12 +280,11 @@ describe('instance reporting retries', () => {
const harness = makeHarness([unreachable(), unreachable(), unreachable(), accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 2);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 3);
await expect(harness.task.run()).rejects.toThrow();
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await expect(harness.task.run()).rejects.toThrow();
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await expect(harness.task.run()).rejects.toThrow();
expect(harness.httpRequest).toHaveBeenCalledTimes(MAX_ATTEMPTS);
const [abandoned] = await repository.find({ where: { status: 'skipped_after_max_retries' } });
@@ -337,13 +292,12 @@ describe('instance reporting retries', () => {
expect(dailyPoints(sentPayload(harness, 0))).toEqual([{ date: '2026-03-25', value: 5 }]);
// The next pass finds the day settled and moves on to tomorrow's slot.
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 4);
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(MAX_ATTEMPTS);
expectArmedFor(harness, NEXT_SLOT);
await vi.advanceTimersByTimeAsync(new Date(NEXT_SLOT).getTime() - Date.now());
await armed(harness, 5);
vi.setSystemTime(new Date(NEXT_SLOT));
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(MAX_ATTEMPTS + 1);
const backfill = sentPayload(harness, MAX_ATTEMPTS);
@@ -367,19 +321,18 @@ describe('instance reporting retries', () => {
const harness = makeHarness([tooLarge(), accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
const [rejected] = await repository.find({ where: { status: 'skipped_after_max_retries' } });
expect(rejected).toMatchObject({ attempts: 1 });
expectArmedFor(harness, NEXT_SLOT);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(new Date(NEXT_SLOT).getTime() - Date.now());
await armed(harness, 2);
vi.setSystemTime(new Date(NEXT_SLOT));
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(2);
const backfill = sentPayload(harness, 1);
@@ -397,8 +350,7 @@ describe('instance reporting retries', () => {
await seedDailyExecutions({ '2026-03-23': 4, '2026-03-25': 9 });
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
const payload = sentPayload(harness, 0);
@@ -421,8 +373,7 @@ describe('instance reporting retries', () => {
await seedDailyExecutions({ '2026-01-10': 4, '2026-02-24': 6, '2026-03-25': 8 });
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
const points = dailyPoints(sentPayload(harness, 0));
@@ -460,8 +411,7 @@ describe('instance reporting retries', () => {
});
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
const points = dailyPoints(sentPayload(harness, 0));
// From 29 days back to yesterday. 02-24 falls in the one day of margin.
@@ -492,18 +442,16 @@ describe('instance reporting retries', () => {
...Array.from({ length: 3 * MAX_ATTEMPTS }, unreachable),
accepted(),
]);
harness.scheduler.start();
let passes = 0;
for (const nextSlot of [
'2026-03-27T07:42:00.000Z',
'2026-03-28T07:42:00.000Z',
'2026-03-29T07:42:00.000Z',
]) {
await armed(harness, ++passes);
await expect(harness.task.run()).rejects.toThrow();
for (let attempt = 1; attempt < MAX_ATTEMPTS; attempt++) {
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, ++passes);
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await expect(harness.task.run()).rejects.toThrow();
}
const [skipped] = await repository.find({
@@ -513,12 +461,11 @@ describe('instance reporting retries', () => {
});
expect(skipped.attempts).toBe(MAX_ATTEMPTS);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, ++passes);
expectArmedFor(harness, nextSlot);
await vi.advanceTimersByTimeAsync(new Date(nextSlot).getTime() - Date.now());
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await harness.task.run();
vi.setSystemTime(new Date(nextSlot));
}
await armed(harness, ++passes);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(3 * MAX_ATTEMPTS + 1);
// Every report started at the first day with data, not at its own yesterday.
@@ -547,8 +494,7 @@ describe('instance reporting retries', () => {
test('resumes the retry budget and the remaining wait after a restart', async () => {
const before = makeHarness([unreachable()]);
before.scheduler.start();
await armed(before, 1);
await expect(before.task.run()).rejects.toThrow();
expect(before.httpRequest).toHaveBeenCalledTimes(1);
await expect(repository.count()).resolves.toBe(1);
@@ -556,23 +502,15 @@ describe('instance reporting retries', () => {
expect(report).toMatchObject({ status: 'pending', attempts: 1 });
// The process dies two minutes into the five-minute wait.
before.scheduler.shutdown();
await vi.advanceTimersByTimeAsync(2 * Time.minutes.toMilliseconds);
vi.setSystemTime(Date.now() + 2 * Time.minutes.toMilliseconds);
const after = makeHarness([accepted()]);
after.scheduler.start();
await armed(after, 1);
await after.task.run();
// Held back by the row's `lastAttemptAt` rather than attempting at once.
expect(after.httpRequest).not.toHaveBeenCalled();
// Three of the five minutes remain, less the fake clock's drift.
const [remainingWait] = after.scheduleNext.mock.calls[0];
const expectedWait = RETRY_DELAY_MS - 2 * Time.minutes.toMilliseconds;
expect(remainingWait).toBeGreaterThanOrEqual(expectedWait - CLOCK_TOLERANCE_MS);
expect(remainingWait).toBeLessThanOrEqual(expectedWait);
await vi.advanceTimersByTimeAsync(remainingWait);
await armed(after, 2);
vi.setSystemTime(Date.now() + 3 * Time.minutes.toMilliseconds);
await after.task.run();
expect(after.httpRequest).toHaveBeenCalledTimes(1);
expect(sentPayload(after, 0).batchId).toBe(report.id);
@@ -594,8 +532,7 @@ describe('instance reporting retries', () => {
});
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
// The retry after midnight resends the same row. It does not make a new row.
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
@@ -608,7 +545,6 @@ describe('instance reporting retries', () => {
attempts: 2,
});
await expect(repository.count()).resolves.toBe(1);
expectArmedFor(harness, '2026-03-27T23:56:00.000Z');
});
test('keeps retrying across the boundary, then backfills the day after it is skipped', async () => {
@@ -624,10 +560,9 @@ describe('instance reporting retries', () => {
});
const harness = makeHarness([unreachable(), unreachable(), accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 2);
await expect(harness.task.run()).rejects.toThrow();
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await expect(harness.task.run()).rejects.toThrow();
// Two more attempts run after midnight. Then the row stops.
expect(harness.httpRequest).toHaveBeenCalledTimes(2);
@@ -637,12 +572,11 @@ describe('instance reporting retries', () => {
});
// The next pass finds the day settled. It waits for the next slot.
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 3);
expectArmedFor(harness, '2026-03-27T23:56:00.000Z');
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await harness.task.run();
await vi.advanceTimersByTimeAsync(new Date('2026-03-27T23:56:00.000Z').getTime() - Date.now());
await armed(harness, 4);
vi.setSystemTime(new Date('2026-03-27T23:56:00.000Z'));
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(3);
const backfill = sentPayload(harness, 2);
@@ -668,8 +602,7 @@ describe('instance reporting retries', () => {
});
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
// The latest row is delivered. So the code sends nothing and leaves the orphan.
expect(harness.httpRequest).not.toHaveBeenCalled();
@@ -704,8 +637,7 @@ describe('instance reporting retries', () => {
vi.setSystemTime(new Date('2026-03-29T20:59:00.000Z'));
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
// Four days is past its own next slot, so it is given up, not resent — even
// though the time of day is still before tonight's slot.
@@ -713,11 +645,10 @@ describe('instance reporting retries', () => {
await expect(repository.findOneByOrFail({ id: stale.id })).resolves.toMatchObject({
status: 'skipped_after_max_retries',
});
expectArmedFor(harness, '2026-03-29T23:58:00.000Z');
// Tonight's slot creates one fresh report for the whole gap.
await vi.advanceTimersByTimeAsync(new Date('2026-03-29T23:58:00.000Z').getTime() - Date.now());
await armed(harness, 2);
vi.setSystemTime(new Date('2026-03-29T23:58:00.000Z'));
await harness.task.run();
const backfill = sentPayload(harness, 0);
expect(backfill.batchId).not.toBe(stale.id);
@@ -734,7 +665,7 @@ describe('instance reporting retries', () => {
expect(new Set(dates).size).toBe(dates.length);
});
test('caps the retry to the slot, so a failure at the end of the day backfills the same night', async () => {
test('expires the pending row at its next slot, even during the retry delay', async () => {
await setReportTime('23:58');
await seedDeliveredReport('2026-03-23');
@@ -752,18 +683,16 @@ describe('instance reporting retries', () => {
vi.setSystemTime(new Date('2026-03-26T23:56:00.000Z'));
const harness = makeHarness([unreachable(), accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await expect(harness.task.run()).rejects.toThrow();
// The resume fails; the next attempt is capped to the slot, two minutes away,
// rather than a full five minutes that would cross midnight.
// The resume fails two minutes before the next slot.
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
expect(sentPayload(harness, 0).batchId).toBe(stale.id);
// Two minutes later, at the slot, the row is given up and a fresh report goes
// out the same night — not deferred to the next day's slot.
await vi.advanceTimersByTimeAsync(2 * Time.minutes.toMilliseconds);
await armed(harness, 2);
vi.setSystemTime(Date.now() + 2 * Time.minutes.toMilliseconds);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(2);
await expect(repository.findOneByOrFail({ id: stale.id })).resolves.toMatchObject({
@@ -775,7 +704,6 @@ describe('instance reporting retries', () => {
{ date: '2026-03-24', value: 4 },
{ date: '2026-03-25', value: 6 },
]);
expectArmedFor(harness, '2026-03-27T23:58:00.000Z');
});
test('skips a stale row and sends a fresh report in the same pass, after the slot', async () => {
@@ -795,8 +723,7 @@ describe('instance reporting retries', () => {
vi.setSystemTime(new Date('2026-03-27T23:59:00.000Z'));
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
// The first pass gives up the stale row and creates the new report at once.
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
@@ -815,7 +742,7 @@ describe('instance reporting retries', () => {
expect(new Set(dates).size).toBe(dates.length);
});
test('retries after the delay and sends the report of a process that created it first and stopped', async () => {
test('resumes the report of a process that created it first and stopped', async () => {
const otherPoints: InstanceReportDataPoint[] = [
{ kind: 'cumulative', name: 'billableExecutions', value: 999 },
];
@@ -830,14 +757,12 @@ describe('instance reporting retries', () => {
});
const harness = makeHarness([accepted()]);
harness.scheduler.start();
await armed(harness, 1);
await harness.task.run();
expect(harness.httpRequest).not.toHaveBeenCalled();
expect(harness.scheduleNext).toHaveBeenLastCalledWith(RETRY_DELAY_MS);
await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS);
await armed(harness, 2);
vi.setSystemTime(Date.now() + RETRY_DELAY_MS);
await harness.task.run();
expect(harness.httpRequest).toHaveBeenCalledTimes(1);
expect(sentPayload(harness, 0)).toMatchObject({ batchId: other?.id, dataPoints: otherPoints });
@@ -11,7 +11,7 @@ import { mock } from 'vitest-mock-extended';
import { License } from '@/license';
import { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository';
import { InstanceReportingScheduler } from '../instance-reporting-scheduler.service';
import { InstanceReportingTask } from '../instance-reporting.task';
import { InstanceReportingSettingsService } from '../instance-reporting-settings.service';
import { InstanceReportingConfig } from '../instance-reporting.config';
import { InstanceReportingModule } from '../instance-reporting.module';
@@ -51,10 +51,7 @@ function setUpContainer({
reportRepository.findLastDeliveryTime.mockResolvedValue(lastDelivery);
Container.set(InstanceMonitoringReportRepository, reportRepository);
const scheduler = mock<InstanceReportingScheduler>();
Container.set(InstanceReportingScheduler, scheduler);
return { settingsService, reportRepository, scheduler };
return { settingsService, reportRepository };
}
describe('InstanceReportingModule', () => {
@@ -98,45 +95,39 @@ describe('InstanceReportingModule', () => {
});
});
describe('init()', () => {
it('starts the scheduler', async () => {
const { scheduler } = setUpContainer();
describe('systemTasks()', () => {
it('returns the reporting task', async () => {
setUpContainer();
await new InstanceReportingModule().init();
expect(scheduler.init).toHaveBeenCalled();
expect(await new InstanceReportingModule().systemTasks()).toEqual([InstanceReportingTask]);
});
it('returns no task when no receiver is configured', async () => {
setUpContainer({ baseUrl: '' });
expect(await new InstanceReportingModule().systemTasks()).toEqual([]);
});
it('returns no task when the instance has no license certificate', async () => {
setUpContainer({ licenseCert: '' });
expect(await new InstanceReportingModule().systemTasks()).toEqual([]);
});
it('returns the task with an auth token but no license certificate', async () => {
setUpContainer({ licenseCert: '', authToken: 'secret-token' });
expect(await new InstanceReportingModule().systemTasks()).toEqual([InstanceReportingTask]);
});
});
describe('init()', () => {
it('fails when the insights module is disabled', async () => {
setUpContainer({ disabledModules: ['insights'] });
await expect(new InstanceReportingModule().init()).rejects.toThrow(UserError);
});
it('leaves the scheduler alone when no receiver is configured', async () => {
const { scheduler } = setUpContainer({ baseUrl: '' });
await new InstanceReportingModule().init();
expect(scheduler.init).not.toHaveBeenCalled();
});
it('leaves the scheduler alone when the instance has no license certificate', async () => {
const { scheduler } = setUpContainer({ licenseCert: '' });
await new InstanceReportingModule().init();
expect(scheduler.init).not.toHaveBeenCalled();
});
it('starts the scheduler with an auth token but no license certificate', async () => {
const { scheduler } = setUpContainer({ licenseCert: '', authToken: 'secret-token' });
await new InstanceReportingModule().init();
expect(scheduler.init).toHaveBeenCalled();
});
// The route belongs to the loaded module, not to the receiver, so a client
// that asks anyway gets an answer instead of a 404.
it('registers the status route even when no receiver is configured', async () => {
@@ -5,6 +5,7 @@ import type {
HttpRequestClientOptions,
OutboundHttp,
} from '@n8n/backend-network';
import { Time } from '@n8n/constants';
import type { LicenseMetricsRepository } from '@n8n/db';
import type { InstanceSettings } from 'n8n-core';
import type { IHttpRequestOptions } from 'n8n-workflow';
@@ -17,11 +18,9 @@ import type { InsightsService } from '@/modules/insights/insights.service';
import type { InstanceMonitoringReport } from '../database/entities/instance-monitoring-report';
import type { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository';
import type { InstanceReportingSettingsService } from '../instance-reporting-settings.service';
import { InstanceReportingConfig } from '../instance-reporting.config';
import {
InstanceReportAlreadyCreatedError,
InstanceReportingService,
} from '../instance-reporting.service';
import { type DueReportWork, InstanceReportingService } from '../instance-reporting.service';
vi.mock('@/constants', async (importOriginal) => ({
...(await importOriginal<typeof import('@/constants')>()),
@@ -82,6 +81,7 @@ function makeReport(overrides: Partial<InstanceMonitoringReport> = {}): Instance
interface Harness {
service: InstanceReportingService;
reportRepository: Mocked<InstanceMonitoringReportRepository>;
settingsService: Mocked<InstanceReportingSettingsService>;
insightsService: Mocked<InsightsService>;
insightsConfig: InsightsConfig;
license: Mocked<License>;
@@ -93,6 +93,11 @@ interface Harness {
function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness {
const reportRepository = mock<InstanceMonitoringReportRepository>();
reportRepository.findPending.mockResolvedValue(null);
reportRepository.claimForSend.mockImplementation(async () => new Date());
reportRepository.readDbNow.mockImplementation(async () => new Date());
reportRepository.markDelivered.mockResolvedValue(true);
reportRepository.recordFailure.mockResolvedValue(true);
reportRepository.markSkipped.mockResolvedValue(true);
// A report already covers the day before the one under test, so the default
// harness reports exactly one day.
reportRepository.findLastCoveredDay.mockResolvedValue('2026-03-24');
@@ -100,6 +105,9 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness {
makeReport({ dataPoints }),
);
const settingsService = mock<InstanceReportingSettingsService>();
settingsService.getReportTime.mockResolvedValue('07:42');
const insightsService = mock<InsightsService>();
// The reported day held 42 executions.
insightsService.getDailyExecutionTotals.mockResolvedValue(totalsByDay({ [REPORT_DATE]: 42 }));
@@ -130,6 +138,7 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness {
const service = new InstanceReportingService(
config,
reportRepository,
settingsService,
insightsService,
insightsConfig,
mock<InstanceSettings>({ instanceId: 'abc123' }),
@@ -143,6 +152,7 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness {
return {
service,
reportRepository,
settingsService,
insightsService,
insightsConfig,
license,
@@ -153,6 +163,283 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness {
}
describe('InstanceReportingService', () => {
describe('findDueWork', () => {
const BEFORE_SLOT = new Date('2026-03-26T07:41:00.000Z');
const AT_SLOT = new Date('2026-03-26T07:42:00.000Z');
const AFTER_SLOT = new Date('2026-03-26T07:43:00.000Z');
beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(AFTER_SLOT);
});
afterEach(() => vi.useRealTimers());
it('waits for the configured UTC slot', async () => {
const { service } = makeHarness();
expect(await service.findDueWork(BEFORE_SLOT)).toEqual({
expiredReport: null,
reportDue: false,
slot: null,
});
expect(await service.findDueWork(AT_SLOT)).toEqual({
expiredReport: null,
reportDue: true,
slot: AT_SLOT,
});
});
it('catches up after the slot and leaves a settled day alone', async () => {
const { service, reportRepository } = makeHarness();
expect((await service.findDueWork(AFTER_SLOT)).reportDue).toBe(true);
reportRepository.hasSettledToday.mockResolvedValue(true);
expect((await service.findDueWork(AFTER_SLOT)).reportDue).toBe(false);
});
it('uses the current report time on every call', async () => {
const { service, settingsService } = makeHarness();
settingsService.getReportTime.mockResolvedValueOnce('08:00');
expect((await service.findDueWork(AFTER_SLOT)).reportDue).toBe(false);
expect((await service.findDueWork(AFTER_SLOT)).reportDue).toBe(true);
settingsService.getReportTime.mockResolvedValue('08:00');
expect((await service.findDueWork(AFTER_SLOT)).reportDue).toBe(false);
});
it('waits out the persisted retry delay of a pending report', async () => {
const { service, reportRepository } = makeHarness();
reportRepository.findPending.mockResolvedValue(
makeReport({
createdAt: AFTER_SLOT,
lastAttemptAt: new Date(AFTER_SLOT.getTime() - 3 * Time.minutes.toMilliseconds),
}),
);
expect(await service.findDueWork(AFTER_SLOT)).toEqual({
expiredReport: null,
reportDue: false,
slot: AT_SLOT,
});
const afterDelay = new Date(AFTER_SLOT.getTime() + 2 * Time.minutes.toMilliseconds);
vi.setSystemTime(afterDelay);
expect(await service.findDueWork(afterDelay)).toEqual({
expiredReport: null,
reportDue: true,
slot: AT_SLOT,
});
expect(reportRepository.hasSettledToday).not.toHaveBeenCalled();
});
it('expires a stale report before applying its retry delay', async () => {
const { service, reportRepository } = makeHarness();
const pending = makeReport({
createdAt: new Date('2026-03-25T07:42:00.000Z'),
lastAttemptAt: new Date(AFTER_SLOT.getTime() - Time.minutes.toMilliseconds),
});
reportRepository.findPending.mockResolvedValue(pending);
expect(await service.findDueWork(AFTER_SLOT)).toEqual({
expiredReport: pending,
reportDue: true,
slot: AT_SLOT,
});
});
it('keeps a claimed report active across its next slot', async () => {
const { service, reportRepository } = makeHarness();
const sending = makeReport({
status: 'sending',
createdAt: new Date('2026-03-25T07:42:00.000Z'),
lastAttemptAt: new Date('2026-03-26T07:41:30.000Z'),
});
reportRepository.findPending.mockResolvedValue(sending);
expect(await service.findDueWork(AFTER_SLOT)).toEqual({
expiredReport: null,
reportDue: false,
slot: null,
});
expect(reportRepository.releaseStaleSend).not.toHaveBeenCalled();
});
it('releases a claim left by a stopped main', async () => {
const { service, reportRepository } = makeHarness();
const startedAt = new Date('2026-03-26T07:40:00.000Z');
const sending = makeReport({
status: 'sending',
createdAt: AFTER_SLOT,
lastAttemptAt: startedAt,
});
reportRepository.findPending
.mockResolvedValueOnce(sending)
.mockResolvedValue(makeReport({ createdAt: AFTER_SLOT, lastAttemptAt: startedAt }));
expect(await service.findDueWork(AFTER_SLOT)).toEqual({
expiredReport: null,
reportDue: false,
slot: AT_SLOT,
});
expect(reportRepository.releaseStaleSend).toHaveBeenCalledWith(BATCH_ID, startedAt);
});
it('releases a sending row with no claim timestamp', async () => {
const { service, reportRepository } = makeHarness();
reportRepository.findPending
.mockResolvedValueOnce(makeReport({ status: 'sending', lastAttemptAt: null }))
.mockResolvedValue(makeReport({ createdAt: AFTER_SLOT }));
expect(await service.findDueWork(AFTER_SLOT)).toEqual({
expiredReport: null,
reportDue: true,
slot: AT_SLOT,
});
expect(reportRepository.releaseStaleSend).toHaveBeenCalledWith(BATCH_ID, null);
});
it('uses the database clock to check claim age', async () => {
const { service, reportRepository } = makeHarness();
reportRepository.findPending.mockResolvedValue(
makeReport({ status: 'sending', lastAttemptAt: AFTER_SLOT }),
);
reportRepository.readDbNow.mockResolvedValue(AFTER_SLOT);
expect(await service.findDueWork(new Date('2026-03-27T07:43:00.000Z'))).toEqual({
expiredReport: null,
reportDue: false,
slot: null,
});
expect(reportRepository.releaseStaleSend).not.toHaveBeenCalled();
});
it('uses the database clock to check the retry delay', async () => {
const { service, reportRepository } = makeHarness();
const oneMinuteAfterAttempt = new Date(AFTER_SLOT.getTime() + Time.minutes.toMilliseconds);
const tenMinutesAfterAttempt = new Date(
AFTER_SLOT.getTime() + 10 * Time.minutes.toMilliseconds,
);
reportRepository.findPending.mockResolvedValue(
makeReport({ createdAt: AFTER_SLOT, lastAttemptAt: AFTER_SLOT }),
);
reportRepository.readDbNow.mockResolvedValue(oneMinuteAfterAttempt);
expect(await service.findDueWork(tenMinutesAfterAttempt)).toEqual({
expiredReport: null,
reportDue: false,
slot: AT_SLOT,
});
reportRepository.readDbNow.mockResolvedValue(tenMinutesAfterAttempt);
expect(await service.findDueWork(oneMinuteAfterAttempt)).toEqual({
expiredReport: null,
reportDue: true,
slot: AT_SLOT,
});
});
});
describe('findDueWork with a slot late in the UTC day', () => {
const LATE_SLOT = new Date('2026-03-26T23:50:00.000Z');
function lateHarness() {
const harness = makeHarness();
harness.settingsService.getReportTime.mockResolvedValue('23:50');
return harness;
}
it('sends the slot on the first pass after midnight', async () => {
const { service, reportRepository } = lateHarness();
expect((await service.findDueWork(new Date('2026-03-26T23:47:00.000Z'))).reportDue).toBe(
false,
);
const work = await service.findDueWork(new Date('2026-03-27T00:02:00.000Z'));
expect(work.reportDue).toBe(true);
expect(work.slot).toEqual(LATE_SLOT);
expect(reportRepository.hasSettledToday).toHaveBeenCalledWith(LATE_SLOT);
});
it('expires a report sent after midnight at the next slot', async () => {
const { service, reportRepository } = lateHarness();
const pending = makeReport({
reportDate: '2026-03-26',
createdAt: new Date('2026-03-27T00:02:00.000Z'),
});
reportRepository.findPending.mockResolvedValue(pending);
expect((await service.findDueWork(new Date('2026-03-27T23:49:00.000Z'))).expiredReport).toBe(
null,
);
expect((await service.findDueWork(new Date('2026-03-27T23:50:00.000Z'))).expiredReport).toBe(
pending,
);
});
it('stops waiting for the slot once its grace has passed', async () => {
const { service } = lateHarness();
expect((await service.findDueWork(new Date('2026-03-27T00:49:00.000Z'))).reportDue).toBe(
true,
);
expect(await service.findDueWork(new Date('2026-03-27T00:50:00.000Z'))).toEqual({
expiredReport: null,
reportDue: false,
slot: null,
});
});
});
describe('sendDueReport', () => {
const NOW = new Date('2026-03-26T07:43:00.000Z');
const SLOT = new Date('2026-03-26T07:42:00.000Z');
function setup(work: Omit<DueReportWork, 'slot'>) {
const harness = makeHarness();
const findDueWork = vi
.spyOn(harness.service, 'findDueWork')
.mockResolvedValue({ ...work, slot: SLOT });
const sendReport = vi.spyOn(harness.service, 'sendReport').mockResolvedValue();
return { ...harness, findDueWork, sendReport };
}
afterEach(() => vi.restoreAllMocks());
it('sends the report when one is due', async () => {
const { service, findDueWork, sendReport } = setup({ expiredReport: null, reportDue: true });
await service.sendDueReport(NOW);
expect(findDueWork).toHaveBeenCalledWith(NOW);
expect(sendReport).toHaveBeenCalledExactlyOnceWith(SLOT);
});
it('sends nothing when no report is due', async () => {
const { service, reportRepository, sendReport } = setup({
expiredReport: null,
reportDue: false,
});
await service.sendDueReport(NOW);
expect(reportRepository.markSkipped).not.toHaveBeenCalled();
expect(sendReport).not.toHaveBeenCalled();
});
it('skips an expired report before it sends', async () => {
const expiredReport = makeReport({ id: 'pending-report', attempts: 1 });
const { service, reportRepository, sendReport } = setup({ expiredReport, reportDue: true });
await service.sendDueReport(NOW);
expect(reportRepository.markSkipped).toHaveBeenCalledWith('pending-report');
expect(reportRepository.markSkipped.mock.invocationCallOrder[0]).toBeLessThan(
sendReport.mock.invocationCallOrder[0],
);
});
it('exposes a delivery failure to the caller', async () => {
const { service, sendReport } = setup({ expiredReport: null, reportDue: true });
sendReport.mockRejectedValue(new Error('Network error'));
await expect(service.sendDueReport(NOW)).rejects.toThrow('Network error');
});
it('recovers from a failed read of the due work on the next pass', async () => {
const { service, findDueWork, sendReport } = setup({ expiredReport: null, reportDue: true });
findDueWork.mockRejectedValueOnce(new Error('DB unavailable'));
await expect(service.sendDueReport(NOW)).rejects.toThrow('DB unavailable');
await service.sendDueReport(NOW);
expect(sendReport).toHaveBeenCalledTimes(1);
});
});
describe('sendReport', () => {
beforeEach(() => {
// Pinned so the previous UTC day the service derives is deterministic.
@@ -185,7 +472,7 @@ describe('InstanceReportingService', () => {
test('posts the report to the receiver endpoint under the configured base URL', async () => {
const { service, http, clientOptions } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(clientOptions?.baseURL).toBe('https://example.com');
expect(sentOptions(http)).toMatchObject({
@@ -218,7 +505,7 @@ describe('InstanceReportingService', () => {
test('does not follow redirects, so the credential reaches only the configured host', async () => {
const { service, http } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(sentOptions(http).disableFollowRedirect).toBe(true);
});
@@ -226,7 +513,7 @@ describe('InstanceReportingService', () => {
test('sends one cumulative and one daily data point, with the row id as batchId', async () => {
const { service, http } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(body(http)).toMatchObject({
instanceId: 'abc123',
@@ -243,7 +530,7 @@ describe('InstanceReportingService', () => {
test('omits label when no label is configured', async () => {
const { service, http } = makeHarness(makeConfig({ instanceReportingLabel: '' }));
await service.sendReport();
await service.sendReport(new Date());
expect(body(http)).not.toHaveProperty('label');
});
@@ -251,7 +538,7 @@ describe('InstanceReportingService', () => {
test('sends the license certificate in the body as the credential', async () => {
const { service, http, clientOptions } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(body(http).licenseCert).toBe(LICENSE_CERT);
// No token: the certificate is the whole credential, and an `undefined`
@@ -262,11 +549,11 @@ describe('InstanceReportingService', () => {
test('reads the certificate fresh for every report, so a renewed license is sent', async () => {
const { service, license, http, reportRepository } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
license.loadCertStr.mockResolvedValue('renewed-license-cert');
reportRepository.findLastCoveredDay.mockResolvedValue('2026-03-23');
await service.sendReport();
await service.sendReport(new Date());
expect(body(http, 1).licenseCert).toBe('renewed-license-cert');
});
@@ -276,7 +563,7 @@ describe('InstanceReportingService', () => {
license.loadCertStr.mockResolvedValue('');
// Resolves rather than throws: this is not a delivery failure to retry.
await expect(service.sendReport()).resolves.toBeUndefined();
await expect(service.sendReport(new Date())).resolves.toBeUndefined();
expect(http.request).not.toHaveBeenCalled();
expect(insightsService.getDailyExecutionTotals).not.toHaveBeenCalled();
@@ -295,7 +582,7 @@ describe('InstanceReportingService', () => {
test('does not send or read the license certificate', async () => {
const { service, http, license } = makeHarness(tokenConfig());
await service.sendReport();
await service.sendReport(new Date());
expect(body(http)).not.toHaveProperty('licenseCert');
expect(license.loadCertStr).not.toHaveBeenCalled();
@@ -306,7 +593,7 @@ describe('InstanceReportingService', () => {
const { service, http, license } = makeHarness(tokenConfig());
license.loadCertStr.mockResolvedValue('');
await service.sendReport();
await service.sendReport(new Date());
expect(http.request).toHaveBeenCalledTimes(1);
});
@@ -315,7 +602,7 @@ describe('InstanceReportingService', () => {
test('reads the daily execution totals for the reported UTC day', async () => {
const { service, insightsService } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(insightsService.getDailyExecutionTotals).toHaveBeenCalledWith({
startDate: new Date('2026-03-25T00:00:00.000Z'),
@@ -326,7 +613,7 @@ describe('InstanceReportingService', () => {
test('records the measurement before sending, then marks the report delivered', async () => {
const { service, reportRepository } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(reportRepository.createPending).toHaveBeenCalledWith(
[
@@ -338,11 +625,27 @@ describe('InstanceReportingService', () => {
expect(reportRepository.markDelivered).toHaveBeenCalledWith(BATCH_ID, expect.any(Date));
});
test('throws without sending when another process already created the report for today', async () => {
test('dates a report sent after midnight by its slot', async () => {
const { service, reportRepository } = makeHarness();
vi.setSystemTime(new Date('2026-03-27T00:02:00.000Z'));
const slot = new Date('2026-03-26T23:50:00.000Z');
await service.sendReport(slot);
expect(reportRepository.createPending).toHaveBeenCalledWith(
[
{ kind: 'cumulative', name: 'billableExecutions', value: 815 },
{ kind: 'daily', name: 'billableExecutions', value: 42, date: REPORT_DATE },
],
slot,
);
});
test('sends nothing when another process already created the report for today', async () => {
const { service, reportRepository, http } = makeHarness();
reportRepository.createPending.mockResolvedValue(null);
await expect(service.sendReport()).rejects.toThrow(InstanceReportAlreadyCreatedError);
await expect(service.sendReport(new Date())).resolves.toBeUndefined();
expect(http.request).not.toHaveBeenCalled();
expect(reportRepository.recordFailure).not.toHaveBeenCalled();
@@ -353,10 +656,11 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
vi.mocked(http.request).mockRejectedValue(new Error('Network error'));
await expect(service.sendReport()).rejects.toThrow('Network error');
await expect(service.sendReport(new Date())).rejects.toThrow('Network error');
expect(reportRepository.recordFailure).toHaveBeenCalledWith(
BATCH_ID,
expect.any(Date),
'Network error',
expect.any(Date),
);
@@ -367,10 +671,11 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
vi.mocked(http.request).mockResolvedValue({ statusCode: 200, body: '', headers: {} });
await expect(service.sendReport()).rejects.toThrow('200');
await expect(service.sendReport(new Date())).rejects.toThrow('200');
expect(reportRepository.recordFailure).toHaveBeenCalledWith(
BATCH_ID,
expect.any(Date),
expect.stringContaining('200'),
expect.any(Date),
);
@@ -381,10 +686,11 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
vi.mocked(http.request).mockResolvedValue({ statusCode: 500, body: '', headers: {} });
await expect(service.sendReport()).rejects.toThrow('500');
await expect(service.sendReport(new Date())).rejects.toThrow('500');
expect(reportRepository.recordFailure).toHaveBeenCalledWith(
BATCH_ID,
expect.any(Date),
expect.stringContaining('500'),
expect.any(Date),
);
@@ -395,7 +701,7 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
vi.mocked(http.request).mockResolvedValue({ statusCode: 301, body: '', headers: {} });
await expect(service.sendReport()).rejects.toThrow('301');
await expect(service.sendReport(new Date())).rejects.toThrow('301');
expect(reportRepository.markDelivered).not.toHaveBeenCalled();
});
@@ -403,17 +709,37 @@ describe('InstanceReportingService', () => {
test('emits a delivered event once the report is delivered', async () => {
const { service, eventService } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(eventService.emit).toHaveBeenCalledWith('instance-report-delivered');
expect(eventService.emit).not.toHaveBeenCalledWith('instance-report-failed');
});
test('does not emit another delivery event for an already delivered report', async () => {
const { service, reportRepository, eventService } = makeHarness();
reportRepository.markDelivered.mockResolvedValue(false);
await service.sendReport(new Date());
expect(eventService.emit).not.toHaveBeenCalled();
});
test('does not settle a rejected report after losing the claim', async () => {
const { service, reportRepository, eventService, http } = makeHarness();
vi.mocked(http.request).mockResolvedValue({ statusCode: 413, body: '', headers: {} });
reportRepository.recordFailure.mockResolvedValue(false);
await service.sendReport(new Date());
expect(reportRepository.markSkipped).not.toHaveBeenCalled();
expect(eventService.emit).not.toHaveBeenCalled();
});
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');
await expect(service.sendReport(new Date())).rejects.toThrow('Network error');
expect(eventService.emit).toHaveBeenCalledWith('instance-report-failed');
expect(eventService.emit).not.toHaveBeenCalledWith('instance-report-delivered');
@@ -424,7 +750,7 @@ describe('InstanceReportingService', () => {
vi.mocked(http.request).mockResolvedValue({ statusCode: 409, body: '', headers: {} });
// Resolving is what stops the retry: the scheduler waits for the next slot.
await expect(service.sendReport()).resolves.toBeUndefined();
await expect(service.sendReport(new Date())).resolves.toBeUndefined();
expect(reportRepository.markDelivered).toHaveBeenCalledWith(BATCH_ID, expect.any(Date));
expect(reportRepository.recordFailure).not.toHaveBeenCalled();
@@ -436,7 +762,7 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
vi.mocked(http.request).mockResolvedValue({ statusCode, body: '', headers: {} });
await expect(service.sendReport()).resolves.toBeUndefined();
await expect(service.sendReport(new Date())).resolves.toBeUndefined();
expect(reportRepository.markSkipped).toHaveBeenCalledWith(BATCH_ID);
},
@@ -446,7 +772,7 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
vi.mocked(http.request).mockResolvedValue({ statusCode, body: '', headers: {} });
await expect(service.sendReport()).rejects.toThrow();
await expect(service.sendReport(new Date())).rejects.toThrow();
expect(reportRepository.markSkipped).not.toHaveBeenCalled();
});
@@ -455,10 +781,10 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
vi.mocked(http.request).mockRejectedValueOnce(new Error('Network error'));
await expect(service.sendReport()).rejects.toThrow();
await expect(service.sendReport(new Date())).rejects.toThrow();
// The retry picks up the still-undelivered row the first attempt created.
reportRepository.findPending.mockResolvedValue(makeReport({ attempts: 1 }));
await service.sendReport();
await service.sendReport(new Date());
expect(body(http, 1).batchId).toBe(BATCH_ID);
});
@@ -472,7 +798,7 @@ describe('InstanceReportingService', () => {
] as InstanceMonitoringReport['dataPoints'];
reportRepository.findPending.mockResolvedValue(makeReport({ dataPoints: measured }));
await service.sendReport();
await service.sendReport(new Date());
// Re-measuring would sample the cumulative total at a different point in
// the day and stretch its interval past 24 hours.
@@ -498,7 +824,7 @@ describe('InstanceReportingService', () => {
vi.mocked(http.request).mockRejectedValue(new Error('Network error'));
reportRepository.findPending.mockResolvedValue(makeReport({ attempts: 1 }));
await expect(service.sendReport()).rejects.toThrow();
await expect(service.sendReport(new Date())).rejects.toThrow();
expect(reportRepository.markSkipped).not.toHaveBeenCalled();
});
@@ -509,7 +835,7 @@ describe('InstanceReportingService', () => {
// Two attempts already recorded on the row, so this one is the last.
reportRepository.findPending.mockResolvedValue(makeReport({ attempts: 2 }));
await expect(service.sendReport()).rejects.toThrow();
await expect(service.sendReport(new Date())).rejects.toThrow();
expect(reportRepository.markSkipped).toHaveBeenCalledWith(BATCH_ID);
});
@@ -519,7 +845,7 @@ describe('InstanceReportingService', () => {
// A crash between the failure record and the skip leaves this row pending.
reportRepository.findPending.mockResolvedValue(makeReport({ attempts: 3 }));
await expect(service.sendReport()).resolves.toBeUndefined();
await expect(service.sendReport(new Date())).resolves.toBeUndefined();
expect(http.request).not.toHaveBeenCalled();
expect(reportRepository.recordFailure).not.toHaveBeenCalled();
@@ -555,7 +881,7 @@ describe('InstanceReportingService', () => {
totalsByDay({ '2026-03-23': 5, '2026-03-24': 7, '2026-03-25': 9 }),
);
await service.sendReport();
await service.sendReport(new Date());
expect(dailyPoints(http)).toEqual([
{ value: 5, date: '2026-03-23' },
@@ -576,7 +902,7 @@ describe('InstanceReportingService', () => {
// Insights returns no row for a day that saw nothing.
insightsService.getDailyExecutionTotals.mockResolvedValue(totalsByDay({ '2026-03-25': 9 }));
await service.sendReport();
await service.sendReport(new Date());
expect(dailyPoints(http)).toEqual([
{ value: 0, date: '2026-03-24' },
@@ -590,7 +916,7 @@ describe('InstanceReportingService', () => {
reportRepository.findLastCoveredDay.mockResolvedValue(null);
insightsService.getEarliestDataDate.mockResolvedValue(new Date('2026-01-01T00:00:00.000Z'));
await service.sendReport();
await service.sendReport(new Date());
const daily = dailyPoints(http);
expect(daily).toHaveLength(84);
@@ -610,7 +936,7 @@ describe('InstanceReportingService', () => {
totalsByDay({ '2026-03-20': 4, [REPORT_DATE]: 42 }),
);
await service.sendReport();
await service.sendReport(new Date());
// Inside the window, a day without data saw no executions.
expect(dailyPoints(http)).toEqual([
@@ -628,7 +954,7 @@ describe('InstanceReportingService', () => {
reportRepository.findLastCoveredDay.mockResolvedValue(null);
insightsService.getDailyExecutionTotals.mockResolvedValue(new Map());
await service.sendReport();
await service.sendReport(new Date());
expect(points(http)).toEqual([
{ kind: 'cumulative', name: 'billableExecutions', value: 815 },
@@ -642,7 +968,7 @@ describe('InstanceReportingService', () => {
insightsService.getEarliestDataDate.mockResolvedValue(new Date('2026-03-26T00:00:00.000Z'));
insightsService.getDailyExecutionTotals.mockResolvedValue(new Map());
await service.sendReport();
await service.sendReport(new Date());
expect(dailyPoints(http)).toEqual([{ value: 0, date: REPORT_DATE }]);
});
@@ -654,7 +980,7 @@ describe('InstanceReportingService', () => {
reportRepository.findLastCoveredDay.mockResolvedValue(null);
insightsService.getEarliestDataDate.mockResolvedValue(new Date('2026-03-22T00:00:00.000Z'));
await service.sendReport();
await service.sendReport(new Date());
expect(dailyPoints(http).map(({ date }) => date)).toEqual([
'2026-03-22',
@@ -676,7 +1002,7 @@ describe('InstanceReportingService', () => {
makeReport({ dataPoints: measured, attempts: 1 }),
);
await service.sendReport();
await service.sendReport(new Date());
expect(points(http)).toEqual(measured);
expect(insightsService.getEarliestDataDate).not.toHaveBeenCalled();
@@ -689,7 +1015,7 @@ describe('InstanceReportingService', () => {
reportRepository.findLastCoveredDay.mockResolvedValue('2026-01-01');
insightsService.getEarliestDataDate.mockResolvedValue(new Date('2025-10-01T00:00:00.000Z'));
await service.sendReport();
await service.sendReport(new Date());
const daily = dailyPoints(http);
expect(daily).toHaveLength(83);
@@ -702,7 +1028,7 @@ describe('InstanceReportingService', () => {
reportRepository.findLastCoveredDay.mockResolvedValue('2025-12-30');
insightsService.getEarliestDataDate.mockResolvedValue(new Date('2026-01-10T00:00:00.000Z'));
await service.sendReport();
await service.sendReport(new Date());
expect(dailyPoints(http).at(0)).toEqual({ value: 0, date: '2026-01-10' });
});
@@ -712,7 +1038,7 @@ describe('InstanceReportingService', () => {
reportRepository.findLastCoveredDay.mockResolvedValue(null);
insightsService.getEarliestDataDate.mockResolvedValue(new Date('2026-01-01T00:00:00.000Z'));
await service.sendReport();
await service.sendReport(new Date());
expect(insightsService.getDailyExecutionTotals).toHaveBeenCalledTimes(1);
expect(insightsService.getDailyExecutionTotals).toHaveBeenCalledWith({
@@ -728,7 +1054,7 @@ describe('InstanceReportingService', () => {
reportRepository.findLastCoveredDay.mockResolvedValue(null);
insightsService.getEarliestDataDate.mockResolvedValue(new Date('2023-01-01T00:00:00.000Z'));
await service.sendReport();
await service.sendReport(new Date());
const daily = dailyPoints(http);
expect(daily).toHaveLength(29);
@@ -739,7 +1065,7 @@ describe('InstanceReportingService', () => {
test('does not read the insights history when only yesterday is owed', async () => {
const { service, insightsService } = makeHarness();
await service.sendReport();
await service.sendReport(new Date());
expect(insightsService.getEarliestDataDate).not.toHaveBeenCalled();
});
@@ -748,7 +1074,7 @@ describe('InstanceReportingService', () => {
const { service, reportRepository, http } = makeHarness();
reportRepository.findLastCoveredDay.mockResolvedValue(REPORT_DATE);
await service.sendReport();
await service.sendReport(new Date());
expect(reportRepository.createPending).not.toHaveBeenCalled();
expect(http.request).not.toHaveBeenCalled();
@@ -0,0 +1,136 @@
import type { EventService } from '@n8n/backend-services';
import { mockLogger } from '@n8n/backend-test-utils';
import type { GlobalConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
import type { ScheduledJobRepository } from '@n8n/db';
import { resolveSystemTaskRunOptions, SystemTaskMetadata } from '@n8n/decorators';
import { Container } from '@n8n/di';
import { Tracing, type ErrorReporter, type InstanceSettings } from 'n8n-core';
import { mock } from 'vitest-mock-extended';
import type { DurableScheduler } from '@/scheduling/durable-scheduler';
import type { SystemTaskJobRegistrar } from '@/scheduling/system-tasks/system-task-job-registrar';
import { SystemTaskRunner } from '@/scheduling/system-tasks/system-task-runner';
import { SystemTaskScheduledJobOwner } from '@/scheduling/system-tasks/system-task-scheduled-job-owner';
import type { InstanceReportingService } from '../instance-reporting.service';
import { InstanceReportingTask } from '../instance-reporting.task';
function setup() {
const service = mock<InstanceReportingService>();
const task = new InstanceReportingTask(service);
return { service, task };
}
describe('InstanceReportingTask', () => {
beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-03-26T07:43:00.000Z'));
});
afterEach(() => vi.useRealTimers());
it('leaves delivery retries to the report row', () => {
const { task } = setup();
expect(resolveSystemTaskRunOptions(task)).toMatchObject({ maxAttempts: 1 });
expect(task.schedule).toEqual({
kind: 'interval',
intervalSeconds: 15 * Time.minutes.toSeconds,
});
});
it('sends the report due at the current time', async () => {
const { task, service } = setup();
await task.run();
expect(service.sendDueReport).toHaveBeenCalledWith(new Date());
});
it('exposes a delivery failure to the runner', async () => {
const { task, service } = setup();
service.sendDueReport.mockRejectedValue(new Error('Network error'));
await expect(task.run()).rejects.toThrow('Network error');
});
describe('runner placement', () => {
function setupRunner(
isLeader: boolean,
schedulerActive: boolean,
enabledForSystemTasks: boolean,
) {
const harness = setup();
Container.set(InstanceReportingTask, harness.task);
const metadata = new SystemTaskMetadata();
metadata.register(InstanceReportingTask);
const scheduler = mock<DurableScheduler>();
scheduler.isActive.mockReturnValue(schedulerActive);
const registrar = mock<SystemTaskJobRegistrar>();
const runner = new SystemTaskRunner(
mockLogger(),
metadata,
scheduler,
registrar,
new SystemTaskScheduledJobOwner(mock<ScheduledJobRepository>()),
mock<GlobalConfig>({ generic: { timezone: 'UTC' }, scheduler: { enabledForSystemTasks } }),
mock<InstanceSettings>({
instanceType: 'main',
instanceRole: isLeader ? 'leader' : 'follower',
isLeader,
}),
mock<ErrorReporter>(),
mock<EventService>(),
new Tracing(),
);
onTestFinished(async () => await runner.shutdown());
return { ...harness, runner, scheduler, registrar };
}
it.each([
[false, false],
[true, false],
[false, true],
])(
'catches up on the leader timer with scheduler=%s and systemTasks=%s',
async (schedulerActive, enabledForSystemTasks) => {
const { runner, service, scheduler } = setupRunner(
true,
schedulerActive,
enabledForSystemTasks,
);
await runner.init();
await vi.advanceTimersByTimeAsync(0);
expect(service.sendDueReport).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(15 * Time.minutes.toMilliseconds);
expect(service.sendDueReport).toHaveBeenCalledTimes(2);
expect(scheduler.registerTaskHandler).not.toHaveBeenCalled();
},
);
it('keeps a follower idle in timer mode and catches up on takeover', async () => {
const { runner, service } = setupRunner(false, false, false);
await runner.init();
await vi.advanceTimersByTimeAsync(15 * Time.minutes.toMilliseconds);
expect(service.sendDueReport).not.toHaveBeenCalled();
runner.startLeaderTimers();
await vi.advanceTimersByTimeAsync(0);
expect(service.sendDueReport).toHaveBeenCalledTimes(1);
await runner.stopLeaderTimers();
await vi.advanceTimersByTimeAsync(15 * Time.minutes.toMilliseconds);
expect(service.sendDueReport).toHaveBeenCalledTimes(1);
});
it.each([true, false])(
'provisions a durable task without starting a timer when isLeader=%s',
async (isLeader) => {
const { runner, service, task, scheduler, registrar } = setupRunner(isLeader, true, true);
await runner.init();
await vi.advanceTimersByTimeAsync(15 * Time.minutes.toMilliseconds);
expect(service.sendDueReport).not.toHaveBeenCalled();
expect(registrar.provision).toHaveBeenCalledWith(task);
expect(scheduler.registerTaskHandler).toHaveBeenCalledWith(
'system:instance-reporting',
expect.anything(),
);
},
);
});
});
@@ -9,11 +9,16 @@ export type InstanceReportDataPoint =
/**
* Where a report stands.
*
* `sending` means a main has claimed the row for one request.
* `skipped_after_max_retries` means the instance stopped trying that day, not
* that the numbers were lost: only a delivered report crosses a day off, so a
* skipped day is measured again and covered by the next report.
*/
export type InstanceReportStatus = 'pending' | 'delivered' | 'skipped_after_max_retries';
export type InstanceReportStatus =
| 'pending'
| 'sending'
| 'delivered'
| 'skipped_after_max_retries';
/**
* One instance report: what was sent to the central monitoring receiver, and
@@ -59,9 +64,9 @@ export class InstanceMonitoringReport extends WithTimestamps {
attempts: number;
/**
* When the last attempt finished, successful or not; `null` before the first.
*
* Persisted so the wait between attempts survives a restart.
* When an active attempt started, or when the last attempt finished.
* Uses the database clock. `null` before the first attempt.
* While sending, this timestamp identifies the claim that can record a failure.
*/
@DateTimeColumn({ nullable: true })
lastAttemptAt: Date | null;
@@ -24,6 +24,11 @@ describe('InstanceMonitoringReportRepository', () => {
await testDb.truncate(['InstanceMonitoringReport']);
});
afterEach(() => {
vi.useRealTimers();
vi.restoreAllMocks();
});
afterAll(async () => {
await testDb.terminate();
});
@@ -35,6 +40,21 @@ describe('InstanceMonitoringReportRepository', () => {
return report;
}
async function claim(id: string) {
const claimedAt = await repository.claimForSend(id);
if (!claimedAt) throw new Error('Could not claim the report');
return claimedAt;
}
async function deliver(id: string, deliveredAt: Date) {
await claim(id);
return await repository.markDelivered(id, deliveredAt);
}
async function fail(id: string, message: string, failedAt: Date) {
return await repository.recordFailure(id, await claim(id), message, failedAt);
}
describe('createPending', () => {
test('records the measurement with a generated id, undelivered', async () => {
const report = await createOn(new Date());
@@ -77,6 +97,75 @@ describe('InstanceMonitoringReportRepository', () => {
});
});
describe('send claims', () => {
test('claims a pending report once and releases only the matching claim', async () => {
const { id } = await createOn(new Date());
const claimedAt = await claim(id);
await expect(repository.claimForSend(id)).resolves.toBeNull();
await expect(repository.findPending()).resolves.toMatchObject({ status: 'sending' });
await expect(
repository.releaseStaleSend(id, new Date(claimedAt.getTime() - 1)),
).resolves.toBe(false);
await expect(repository.releaseStaleSend(id, null)).resolves.toBe(false);
await expect(repository.releaseStaleSend(id, claimedAt)).resolves.toBe(true);
await expect(repository.findPending()).resolves.toMatchObject({ status: 'pending' });
});
test('uses the database clock when the main clock is ahead', async () => {
const { id } = await createOn(new Date());
const databaseNow = await repository.readDbNow();
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(new Date('2099-01-01T00:00:00.000Z'));
const claimedAt = await claim(id);
expect(claimedAt.getTime()).toBeGreaterThanOrEqual(databaseNow.getTime());
expect(claimedAt.getTime() - databaseNow.getTime()).toBeLessThan(10_000);
await expect(repository.findOneByOrFail({ id })).resolves.toMatchObject({
lastAttemptAt: claimedAt,
});
});
test('releases a sending row with no claim timestamp', async () => {
const { id } = await createOn(new Date());
await repository.update({ id }, { status: 'sending', lastAttemptAt: null });
await expect(repository.releaseStaleSend(id, null)).resolves.toBe(true);
await expect(repository.findPending()).resolves.toMatchObject({ status: 'pending' });
});
test('ignores an old failure while a newer claim is active or delivered', async () => {
const { id } = await createOn(new Date());
vi.spyOn(repository, 'readDbNow')
.mockResolvedValueOnce(new Date('2026-03-26T07:42:00.000Z'))
.mockResolvedValueOnce(new Date('2026-03-26T07:57:00.000Z'));
const firstClaim = await claim(id);
await repository.releaseStaleSend(id, firstClaim);
const secondClaim = await claim(id);
const completedAt = new Date('2026-03-26T07:57:01.000Z');
await expect(
repository.recordFailure(id, firstClaim, 'Late failure', completedAt),
).resolves.toBe(false);
await expect(repository.findOneByOrFail({ id })).resolves.toMatchObject({
status: 'sending',
lastAttemptAt: secondClaim,
attempts: 0,
lastError: null,
});
await expect(repository.markDelivered(id, completedAt)).resolves.toBe(true);
const delivered = await repository.findOneByOrFail({ id });
await expect(
repository.recordFailure(id, firstClaim, 'Late failure', completedAt),
).resolves.toBe(false);
await expect(
repository.recordFailure(id, secondClaim, 'Reclaimed request failed', completedAt),
).resolves.toBe(false);
await expect(repository.findOneByOrFail({ id })).resolves.toEqual(delivered);
});
});
describe('findPending', () => {
async function createdOn(date: string) {
const report = await createOn(date);
@@ -101,7 +190,7 @@ describe('InstanceMonitoringReportRepository', () => {
test('returns nothing once the newest report is delivered', async () => {
const created = await createOn(new Date());
await repository.markDelivered(created.id, new Date());
await deliver(created.id, new Date());
await expect(repository.findPending()).resolves.toBeNull();
});
@@ -112,14 +201,14 @@ describe('InstanceMonitoringReportRepository', () => {
// times. Filtering on `status` in the query would return the orphan here.
await createdOn('2026-03-20T07:42:00.000Z');
const newer = await createdOn('2026-03-25T07:42:00.000Z');
await repository.markDelivered(newer.id, new Date());
await deliver(newer.id, new Date());
await expect(repository.findPending()).resolves.toBeNull();
});
test('returns the newest row when it is pending, ignoring an older delivered one', async () => {
const older = await createdOn('2026-03-20T07:42:00.000Z');
await repository.markDelivered(older.id, new Date());
await deliver(older.id, new Date());
const newer = await createdOn('2026-03-25T07:42:00.000Z');
await expect(repository.findPending()).resolves.toMatchObject({ id: newer.id });
@@ -128,7 +217,7 @@ describe('InstanceMonitoringReportRepository', () => {
test('carries the last attempt time, so the wait between attempts survives a restart', async () => {
const created = await createOn(new Date());
const failedAt = new Date('2026-03-26T07:42:00.000Z');
await repository.recordFailure(created.id, 'Network error', failedAt);
await fail(created.id, 'Network error', failedAt);
const pending = await repository.findPending();
@@ -147,9 +236,16 @@ describe('InstanceMonitoringReportRepository', () => {
await expect(repository.hasSettledToday(new Date())).resolves.toBe(false);
});
test("is false while today's report is sending", async () => {
const { id } = await createOn(new Date());
await claim(id);
await expect(repository.hasSettledToday(new Date())).resolves.toBe(false);
});
test("is true once today's report was delivered", async () => {
const created = await createOn(new Date());
await repository.markDelivered(created.id, new Date());
await deliver(created.id, new Date());
await expect(repository.hasSettledToday(new Date())).resolves.toBe(true);
});
@@ -164,7 +260,7 @@ describe('InstanceMonitoringReportRepository', () => {
test("ignores an earlier day's delivered report", async () => {
const delivered = await createOn(new Date());
await repository.markDelivered(delivered.id, new Date());
await deliver(delivered.id, new Date());
const tomorrow = new Date(Date.now() + 24 * 60 * 60 * 1000);
@@ -173,7 +269,7 @@ describe('InstanceMonitoringReportRepository', () => {
test('reads the UTC day the report was created on', async () => {
const delivered = await createOn('2026-03-25T23:59:00.000Z');
await repository.markDelivered(delivered.id, new Date());
await deliver(delivered.id, new Date());
await expect(repository.hasSettledToday(new Date('2026-03-25T00:00:00.000Z'))).resolves.toBe(
true,
@@ -200,7 +296,7 @@ describe('InstanceMonitoringReportRepository', () => {
{ kind: 'daily', name: 'billableExecutions', value: 1, date: '2026-03-24' },
{ kind: 'daily', name: 'billableExecutions', value: 2, date: REPORT_DATE },
]);
await repository.markDelivered(delivered.id, new Date());
await deliver(delivered.id, new Date());
await expect(repository.findLastCoveredDay()).resolves.toBe(REPORT_DATE);
});
@@ -210,10 +306,10 @@ describe('InstanceMonitoringReportRepository', () => {
// nothing was saved and the day is still owed. Reports pile up because the
// scheduler creates a fresh one each day.
const delivered = await createOn('2026-03-23T07:42:00.000Z', daily('2026-03-22', 1));
await repository.markDelivered(delivered.id, new Date());
await deliver(delivered.id, new Date());
const failed = await createOn('2026-03-24T07:42:00.000Z', daily('2026-03-23', 2));
await repository.recordFailure(failed.id, 'Network error', new Date());
await fail(failed.id, 'Network error', new Date());
await createOn('2026-03-25T07:42:00.000Z', daily('2026-03-24', 3));
await expect(repository.findLastCoveredDay()).resolves.toBe('2026-03-22');
@@ -229,11 +325,11 @@ describe('InstanceMonitoringReportRepository', () => {
test('returns the latest delivery time across delivered reports', async () => {
const earlier = await createOn('2026-03-24T07:41:00.000Z');
await repository.markDelivered(earlier.id, new Date('2026-03-24T07:42:00.000Z'));
await deliver(earlier.id, new Date('2026-03-24T07:42:00.000Z'));
const latest = await createOn('2026-03-25T07:42:00.000Z');
const latestDeliveredAt = new Date('2026-03-25T07:42:13.000Z');
await repository.markDelivered(latest.id, latestDeliveredAt);
await deliver(latest.id, latestDeliveredAt);
const lastDelivery = await repository.findLastDeliveryTime();
@@ -243,10 +339,10 @@ describe('InstanceMonitoringReportRepository', () => {
test('ignores reports that never reached the receiver', async () => {
const delivered = await createOn('2026-03-24T07:41:00.000Z');
const deliveredAt = new Date('2026-03-24T07:42:00.000Z');
await repository.markDelivered(delivered.id, deliveredAt);
await deliver(delivered.id, deliveredAt);
const failed = await createOn('2026-03-25T07:42:00.000Z');
await repository.recordFailure(failed.id, 'Network error', new Date());
await fail(failed.id, 'Network error', new Date());
const skipped = await createOn('2026-03-26T07:42:00.000Z');
await repository.markSkipped(skipped.id);
await createOn('2026-03-27T07:42:00.000Z');
@@ -270,7 +366,7 @@ describe('InstanceMonitoringReportRepository', () => {
test('leaves a report delivered when another process delivered it first', async () => {
const { id } = await createOn(new Date());
await repository.markDelivered(id, new Date());
await deliver(id, new Date());
await repository.markSkipped(id);
@@ -281,12 +377,43 @@ describe('InstanceMonitoringReportRepository', () => {
});
describe('markDelivered', () => {
test.each(['pending', 'sending', 'skipped_after_max_retries'] as const)(
'records acceptance from %s and keeps its days covered',
async (status) => {
const { id } = await createOn(new Date());
await repository.update({ id }, { status, attempts: 3, lastError: 'Network error' });
const deliveredAt = new Date('2026-03-26T07:42:00.000Z');
await expect(repository.markDelivered(id, deliveredAt)).resolves.toBe(true);
await expect(repository.findOneByOrFail({ id })).resolves.toMatchObject({
status: 'delivered',
attempts: 4,
deliveredAt,
lastAttemptAt: deliveredAt,
lastError: null,
});
await expect(repository.findLastCoveredDay()).resolves.toBe(REPORT_DATE);
},
);
test('keeps an already delivered report unchanged on another acceptance', async () => {
const { id } = await createOn(new Date());
await deliver(id, new Date('2026-03-26T07:42:00.000Z'));
const delivered = await repository.findOneByOrFail({ id });
await expect(
repository.markDelivered(id, new Date('2026-03-26T07:57:00.000Z')),
).resolves.toBe(false);
await expect(repository.findOneByOrFail({ id })).resolves.toEqual(delivered);
});
test('stamps the delivery time, counts the attempt and clears any earlier error', async () => {
const { id } = await createOn(new Date());
await repository.recordFailure(id, 'Network error', new Date());
await fail(id, 'Network error', new Date());
const deliveredAt = new Date('2026-03-26T07:42:00.000Z');
await repository.markDelivered(id, deliveredAt);
await deliver(id, deliveredAt);
const stored = await repository.findOneByOrFail({ id });
expect(stored.deliveredAt?.toISOString()).toBe(deliveredAt.toISOString());
@@ -299,8 +426,8 @@ describe('InstanceMonitoringReportRepository', () => {
test('counts the attempt and keeps the report undelivered', async () => {
const { id } = await createOn(new Date());
await repository.recordFailure(id, 'Network error', new Date());
await repository.recordFailure(id, 'Still down', new Date());
await fail(id, 'Network error', new Date());
await fail(id, 'Still down', new Date());
await expect(repository.findOneByOrFail({ id })).resolves.toMatchObject({
attempts: 2,
@@ -1,6 +1,6 @@
import { isUniqueConstraintError } from '@n8n/db';
import { dbNowLiteral, isUniqueConstraintError, parseDbTime } from '@n8n/db';
import { Service } from '@n8n/di';
import { DataSource, IsNull, Not, Repository } from '@n8n/typeorm';
import { DataSource, In, IsNull, Not, Repository } from '@n8n/typeorm';
import { v4 as uuid } from 'uuid';
import type { InstanceReportDataPoint } from '../entities/instance-monitoring-report';
@@ -13,8 +13,8 @@ export class InstanceMonitoringReportRepository extends Repository<InstanceMonit
}
/**
* The report a retry may resume, or `null` when there is nothing to resume.
* It is the newest row, and only if that row is still `pending`.
* The active report, or `null` when there is nothing to resume.
* It is the newest row, and only if it is `pending` or `sending`.
*
* At most one report is genuinely pending: the scheduler settles a stale row
* before it creates the next one. But earlier versions left a failed report
@@ -31,7 +31,35 @@ export class InstanceMonitoringReportRepository extends Repository<InstanceMonit
async findPending(): Promise<InstanceMonitoringReport | null> {
const [latest] = await this.find({ order: { createdAt: 'DESC', id: 'DESC' }, take: 1 });
return latest?.status === 'pending' ? latest : null;
return latest?.status === 'pending' || latest?.status === 'sending' ? latest : null;
}
/** Use one clock for claim age and retry delays across mains. */
async readDbNow(): Promise<Date> {
const isPostgres = this.manager.connection.options.type === 'postgres';
const [row]: Array<{ dbNow: Date | string }> = await this.query(
`SELECT ${dbNowLiteral(isPostgres)} AS "dbNow"`,
);
return parseDbTime(row.dbNow);
}
/** Return the claim timestamp. Failure writes must match this claim. */
async claimForSend(id: string): Promise<Date | null> {
const startedAt = await this.readDbNow();
const result = await this.update(
{ id, status: 'pending' },
{ status: 'sending', lastAttemptAt: startedAt },
);
return result.affected === 1 ? startedAt : null;
}
/** Release a claim left by a stopped main. A newer claim stays in place. */
async releaseStaleSend(id: string, startedAt: Date | null): Promise<boolean> {
const result = await this.update(
{ id, status: 'sending', lastAttemptAt: startedAt ?? IsNull() },
{ status: 'pending' },
);
return result.affected === 1;
}
/**
@@ -48,7 +76,10 @@ export class InstanceMonitoringReportRepository extends Repository<InstanceMonit
* lost — only a delivered report crosses a day off.
*/
async hasSettledToday(now: Date): Promise<boolean> {
return await this.existsBy({ reportDate: utcDay(now), status: Not('pending') });
return await this.existsBy({
reportDate: utcDay(now),
status: In(['delivered', 'skipped_after_max_retries']),
});
}
/**
@@ -113,25 +144,49 @@ export class InstanceMonitoringReportRepository extends Repository<InstanceMonit
return report?.deliveredAt ?? null;
}
async markDelivered(id: string, deliveredAt: Date): Promise<void> {
await this.increment({ id }, 'attempts', 1);
await this.update(
{ id },
{ status: 'delivered', deliveredAt, lastAttemptAt: deliveredAt, lastError: null },
/** An acceptance confirms delivery even after another main reclaims the report. */
async markDelivered(id: string, deliveredAt: Date): Promise<boolean> {
const result = await this.update(
{ id, status: Not('delivered') },
{
attempts: () => 'attempts + 1',
status: 'delivered',
deliveredAt,
lastAttemptAt: deliveredAt,
lastError: null,
},
);
return result.affected === 1;
}
async recordFailure(id: string, error: string, failedAt: Date): Promise<void> {
await this.increment({ id }, 'attempts', 1);
await this.update({ id }, { lastAttemptAt: failedAt, lastError: error });
async recordFailure(
id: string,
claimedAt: Date,
error: string,
failedAt: Date,
): Promise<boolean> {
const result = await this.update(
{ id, status: 'sending', lastAttemptAt: claimedAt },
{
attempts: () => 'attempts + 1',
status: 'pending',
lastAttemptAt: failedAt,
lastError: error,
},
);
return result.affected === 1;
}
/**
* Stop trying to deliver this report. Its days are covered by the next one.
* Already delivered report stays delivered.
*/
async markSkipped(id: string): Promise<void> {
await this.update({ id, status: 'pending' }, { status: 'skipped_after_max_retries' });
async markSkipped(id: string): Promise<boolean> {
const result = await this.update(
{ id, status: 'pending' },
{ status: 'skipped_after_max_retries' },
);
return result.affected === 1;
}
}
@@ -1,274 +0,0 @@
import { Logger } from '@n8n/backend-common';
import { EventService } from '@n8n/backend-services';
import { Time } from '@n8n/constants';
import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
import { Service } from '@n8n/di';
import { InstanceSettings } from 'n8n-core';
import { strict } from 'node:assert';
import type { InstanceMonitoringReport } from './database/entities/instance-monitoring-report';
import { InstanceMonitoringReportRepository } from './database/repositories/instance-monitoring-report.repository';
import { InstanceReportingSettingsService } from './instance-reporting-settings.service';
import {
InstanceReportAlreadyCreatedError,
InstanceReportingService,
} from './instance-reporting.service';
const MINUTES_PER_DAY = 24 * 60;
/** How long to wait before re-attempting a delivery that failed. */
const RETRY_DELAY_MS = 5 * Time.minutes.toMilliseconds;
/**
* Fires the daily instance report at this instance's configured report time.
*
* Uses a plain in-process timer, the same way pruning and compaction do,
* because the durable scheduler has no support for system-owned jobs yet. It is
* the intended home for this job once it does, so this class does one thing:
* decide *when*, then call {@link InstanceReportingService.sendReport}.
*
* Durability comes from elsewhere instead:
*
* - **Leader-only.** In multi-main, followers hold no timer, so exactly one
* instance reports. Handover moves the timer with leadership.
* - **Catch-up over precision.** Every tick asks the database whether today's
* report has settled rather than trusting that a timer fired, so a restart or
* handover that straddles the report time still reports that day.
* - **Bounded retry, held in the database.** The report row carries the attempts
* made and when the last one finished, so a restart resumes that budget rather
* than starting a fresh one.
*/
@Service()
export class InstanceReportingScheduler {
private timeout: NodeJS.Timeout | undefined;
private isShuttingDown = false;
private readonly onServerStarted = () => this.start();
constructor(
private readonly reportingService: InstanceReportingService,
private readonly reportRepository: InstanceMonitoringReportRepository,
private readonly settingsService: InstanceReportingSettingsService,
private readonly instanceSettings: InstanceSettings,
private readonly eventService: EventService,
private readonly logger: Logger,
) {
this.logger = this.logger.scoped('instance-reporting');
}
/**
* Start reporting if this instance leads.
*
* Runs from the module entrypoint, which checks that the module has what it
* needs before it loads this class at all, and which the registry calls after
* `initOrchestration` has settled this instance's role, so `isLeader` is
* already meaningful here.
*/
init(): void {
strict(this.instanceSettings.instanceRole !== 'unset', 'Instance role is not set');
// 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 {
return this.instanceSettings.instanceType === 'main' && this.instanceSettings.isLeader;
}
/**
* Begin reporting. The first tick runs immediately: it reports if this day's
* slot has already passed and nothing was delivered for it, then arms the timer
* for the next slot. That is what makes a restart, or taking over from a main
* that died before its slot, still report the day.
*/
@OnLeaderTakeover()
start(): void {
if (!this.isEnabled || this.isShuttingDown || this.timeout !== undefined) return;
this.logger.debug('Started the instance reporting timer');
void this.tick();
}
@OnLeaderStepdown()
stop(): void {
this.eventService.off('server-started', this.onServerStarted);
if (this.timeout === undefined) return;
clearTimeout(this.timeout);
this.timeout = undefined;
this.logger.debug('Stopped the instance reporting timer');
}
@OnShutdown()
shutdown(): void {
this.isShuttingDown = true;
this.stop();
}
/**
* One pass: report if due, then arm the next one. Never throws — a pass that
* fails still re-arms, otherwise one bad day would stop reporting for good.
*/
private async tick(): Promise<void> {
try {
const reportTime = await this.settingsService.getReportTime();
// Wait out the gap between retries, but never past the pending row's own
// next slot — at that slot it is skipped, not retried. A wait that reaches
// the slot collapses to zero, so this pass falls through to reportIfDue,
// which skips the stale row and sends a fresh report.
const waitMs = Math.min(
await this.msUntilRetryAllowed(new Date()),
await this.msUntilNextSlot(reportTime),
);
if (waitMs > 0) {
this.scheduleNext(waitMs);
return;
}
// A failed delivery retries; the row decides when the attempts run out,
// after which the day reads as settled and this waits for the next slot.
if ((await this.reportIfDue(reportTime)) === 'failed') {
this.scheduleNext(Math.min(RETRY_DELAY_MS, await this.msUntilNextSlot(reportTime)));
return;
}
this.scheduleNext(msUntilNext(reportTime, new Date()));
} catch (error) {
// Reaching here means the report time could not even be resolved (e.g. the
// database is briefly unavailable), so retry rather than stall until tomorrow.
this.logger.error('Instance reporting pass failed', { error });
this.scheduleNext(RETRY_DELAY_MS);
}
}
/**
* Send the report when it is due.
*
* A pending report is a retry. Resend it now, also after midnight; it holds
* frozen data, so the slot does not apply to it. But once its own next slot
* has passed, its cycle is over: stop trying and let a new report cover its
* day. A new report waits for the slot, which makes sure the day is complete
* before the code measures it.
*/
private async reportIfDue(reportTime: string): Promise<'sent' | 'skipped' | 'failed'> {
const now = new Date();
const pending = await this.reportRepository.findPending();
if (pending) {
if (pendingIsStale(pending, reportTime, now)) {
await this.reportingService.skip(
pending.id,
pending.attempts,
'slot-passed',
pending.lastError,
);
} else {
return await this.trySend();
}
}
// A new report waits for the slot and runs only when the day is not settled.
if (now.getTime() < slotOn(reportTime, now)) return 'skipped';
if (await this.reportRepository.hasSettledToday(now)) return 'skipped';
return await this.trySend();
}
/**
* Milliseconds until the pending row's own next slot, or `Infinity` when no row
* is pending. Capping a retry sleep with this keeps the timer from sleeping past
* the slot, where the row is skipped and a fresh report takes over — otherwise a
* slot near the end of the UTC day would defer the next report by almost a day.
*/
private async msUntilNextSlot(reportTime: string): Promise<number> {
const pending = await this.reportRepository.findPending();
if (!pending) return Number.POSITIVE_INFINITY;
const nextSlot =
slotOn(reportTime, pending.createdAt) + MINUTES_PER_DAY * Time.minutes.toMilliseconds;
return nextSlot - Date.now();
}
private async trySend(): Promise<'sent' | 'failed'> {
try {
await this.reportingService.sendReport();
return 'sent';
} catch (error) {
if (error instanceof InstanceReportAlreadyCreatedError) {
this.logger.info(
'Unexpected second instance report triggered for today, ignored because there is already a report for today',
);
// In this case, the report is not actually considered "failed" from a business perspective,
// but our code has a bug that led to a report being generated a second time for the same day.
// Returning 'failed' here, will lead to a retry of the report being scheduled after RETRY_DELAY_MS.
// The expectation is that after that single no-op retry, retrying stops
// as InstanceReportingService.sendReport will exit early.
return 'failed';
}
this.logger.warn('Failed to deliver the instance report', { error });
return 'failed';
}
}
private scheduleNext(delayMs: number): void {
if (!this.isEnabled || this.isShuttingDown) return;
this.timeout = setTimeout(async () => await this.tick(), delayMs);
}
/**
* How long the scheduler must wait before attempting today's report again, or
* `0` when it may attempt now.
*
* Derived from the report row, so the wait survives a restart. Without it, a
* crash loop would attempt at once every time and spend the whole budget in
* seconds.
*/
private async msUntilRetryAllowed(now: Date): Promise<number> {
const pending = await this.reportRepository.findPending();
if (!pending?.lastAttemptAt) return 0;
const elapsed = now.getTime() - pending.lastAttemptAt.getTime();
return Math.max(0, RETRY_DELAY_MS - elapsed);
}
}
/**
* Whether a pending row sat unsent past its own next slot. Its cycle is then
* over, so a new report covers its day instead of a late resend.
*/
function pendingIsStale(report: InstanceMonitoringReport, reportTime: string, now: Date): boolean {
const nextSlot =
slotOn(reportTime, report.createdAt) + MINUTES_PER_DAY * Time.minutes.toMilliseconds;
return now.getTime() >= nextSlot;
}
/** Epoch ms of `reportTime` on `now`'s UTC day. */
function slotOn(reportTime: string, now: Date): number {
const [hour, minute] = reportTime.split(':').map(Number);
return Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate(), hour, minute, 0, 0);
}
/**
* Milliseconds until the next `reportTime` strictly after `now` — later today if
* the slot is still ahead, otherwise tomorrow. Recomputed from the wall clock on
* every pass, so a clock correction shifts the next fire instead of accumulating
* drift.
*/
function msUntilNext(reportTime: string, now: Date): number {
const today = slotOn(reportTime, now);
const next =
today > now.getTime() ? today : today + MINUTES_PER_DAY * Time.minutes.toMilliseconds;
return next - now.getTime();
}
@@ -9,8 +9,7 @@ import { UserError } from 'n8n-workflow';
* receiver, once a day.
*
* Opt-in: not a default module, so it runs only when an operator lists it in
* `N8N_ENABLED_MODULES`. Main-only, and within a multi-main deployment only the
* leader holds the timer, so a cluster reports once rather than once per main.
* `N8N_ENABLED_MODULES`. A cluster system task sends reports from a main.
*
* The daily figure comes from the insights module, but the receiver only sees
* data points, so that source is an implementation detail of
@@ -36,21 +35,19 @@ export class InstanceReportingModule implements ModuleInterface {
logger.warn(
'Instance reporting is enabled but N8N_INSTANCE_REPORTING_BASE_URL is unset, so no reports will be sent',
);
return;
}
if (!(await this.hasCredential())) {
} else if (!(await this.hasCredential())) {
logger.warn(
'Instance reporting is enabled but this instance has no license certificate, so no reports will be sent. The receiver accepts reports only from licensed instances. Set N8N_LICENSE_CERT, activate a license, or set N8N_INSTANCE_REPORTING_AUTH_TOKEN.',
);
return;
}
}
const { InstanceReportingScheduler } = await import(
'./instance-reporting-scheduler.service.js'
);
Container.get(InstanceReportingScheduler).init();
async systemTasks() {
if (await this.isConfigured()) {
const { InstanceReportingTask } = await import('./instance-reporting.task.js');
return [InstanceReportingTask];
}
return [];
}
/**
@@ -14,8 +14,12 @@ import { License } from '@/license';
import { InsightsConfig } from '@/modules/insights/insights.config';
import { InsightsService } from '@/modules/insights/insights.service';
import type { InstanceReportDataPoint } from './database/entities/instance-monitoring-report';
import type {
InstanceMonitoringReport,
InstanceReportDataPoint,
} from './database/entities/instance-monitoring-report';
import { InstanceMonitoringReportRepository } from './database/repositories/instance-monitoring-report.repository';
import { InstanceReportingSettingsService } from './instance-reporting-settings.service';
import { InstanceReportingConfig } from './instance-reporting.config';
import { INSTANCE_REPORTS_PATH } from './instance-reporting.constants';
@@ -29,6 +33,18 @@ const REQUEST_TIMEOUT_MS = 30 * Time.seconds.toMilliseconds;
*/
const MAX_ATTEMPTS = 3;
/** How long a pending report waits after a failed attempt before the next one. */
const RETRY_DELAY_MS = 5 * Time.minutes.toMilliseconds;
/** Longer than the HTTP timeout, so a stopped main does not hold a report forever. */
const SEND_CLAIM_TIMEOUT_MS = 2 * Time.minutes.toMilliseconds;
/**
* How long yesterday's slot stays due. Longer than the pass interval, so a slot
* late in the UTC day is still sent after midnight.
*/
const SLOT_GRACE_MS = Time.hours.toMilliseconds;
/** The receiver rejects the payload itself, which a pending report resends unchanged. */
const PAYLOAD_REJECTED_STATUSES = new Set([400, 413]);
@@ -39,14 +55,15 @@ class InstanceReportRejectedError extends OperationalError {
}
}
export class InstanceReportAlreadyCreatedError extends OperationalError {
constructor() {
super('Another process already created the instance report for today');
}
}
type SkipReason = 'max-retries' | 'slot-passed' | 'rejected';
export interface DueReportWork {
expiredReport: InstanceMonitoringReport | null;
reportDue: boolean;
/** The slot a new report is created for, or `null` when no slot is due. */
slot: Date | null;
}
const SKIP_MESSAGES: Record<SkipReason, string> = {
'max-retries': 'Giving up on the instance report after repeated delivery failures',
'slot-passed': 'Giving up on the instance report because its slot has passed',
@@ -54,8 +71,8 @@ const SKIP_MESSAGES: Record<SkipReason, string> = {
};
/**
* Measures and delivers one instance report. *When* that happens is
* {@link InstanceReportingScheduler}'s concern.
* Measures and delivers one instance report. How often a pass runs is
* {@link InstanceReportingTask}'s concern.
*/
@Service()
export class InstanceReportingService {
@@ -64,6 +81,7 @@ export class InstanceReportingService {
constructor(
private readonly config: InstanceReportingConfig,
private readonly reportRepository: InstanceMonitoringReportRepository,
private readonly settingsService: InstanceReportingSettingsService,
private readonly insightsService: InsightsService,
private readonly insightsConfig: InsightsConfig,
private readonly instanceSettings: InstanceSettings,
@@ -90,6 +108,60 @@ export class InstanceReportingService {
});
}
/**
* The work due at `now`, from the report time and the stored report rows.
* A pending report inside its slot is due once its retry delay has elapsed.
* A new slot expires the pending report, even during the retry delay. Then a
* report is due for the latest slot at or before `now`, unless the day of
* that slot is settled. Yesterday's slot counts only within its grace.
*/
async findDueWork(now: Date): Promise<DueReportWork> {
const reportTime = await this.settingsService.getReportTime();
let pending = await this.reportRepository.findPending();
const retryNow = pending ? await this.reportRepository.readDbNow() : now;
if (pending?.status === 'sending') {
if (
pending.lastAttemptAt === null ||
retryNow.getTime() - pending.lastAttemptAt.getTime() >= SEND_CLAIM_TIMEOUT_MS
) {
await this.reportRepository.releaseStaleSend(pending.id, pending.lastAttemptAt);
pending = await this.reportRepository.findPending();
}
}
let work: DueReportWork = { expiredReport: null, reportDue: false, slot: null };
if (pending?.status !== 'sending') {
const current =
pending && now.getTime() < slotOn(reportTime, slotDay(pending)) + Time.days.toMilliseconds
? pending
: null;
const slot = dueSlot(reportTime, now);
const reportDue = current
? !isInRetryDelay(current, retryNow)
: slot !== null && !(await this.reportRepository.hasSettledToday(slot));
work = { expiredReport: current ? null : pending, reportDue, slot };
}
return work;
}
/** Skip the report whose slot passed at `now`, then send the report due at `now`, if any. */
async sendDueReport(now: Date): Promise<void> {
const { expiredReport, reportDue, slot } = await this.findDueWork(now);
if (expiredReport) {
await this.skip(
expiredReport.id,
expiredReport.attempts,
'slot-passed',
expiredReport.lastError,
);
}
if (reportDue) {
await this.sendReport(slot);
}
}
/**
* Report billable executions as a daily data point for every unreported day,
* plus the instance's lifetime total as a cumulative one. Only a finished day
@@ -113,10 +185,13 @@ export class InstanceReportingService {
*
* A 400 or 413 skips the report at once, since a resend carries the same payload.
*
* @throws when delivery fails and a retry may succeed, or another process created
* today's report, so the scheduler retries with backoff.
* A new report is created for the UTC day of `slot` and ends on the day before
* it, also when it is sent after midnight. Without a pending report and a
* `slot`, nothing is sent.
*
* @throws when delivery fails and a retry may succeed, so the scheduler retries with backoff.
*/
async sendReport(): Promise<void> {
async sendReport(slot: Date | null): Promise<void> {
const licenseCert = this.config.instanceReportingAuthToken
? undefined
: await this.license.loadCertStr();
@@ -124,30 +199,40 @@ export class InstanceReportingService {
this.logger.warn(
'Skipping the instance report because this instance has no license certificate.',
);
return;
}
const now = new Date();
let report = await this.reportRepository.findPending();
// A crash between recording a failure and skipping the report leaves an
// exhausted row pending, so the budget is re-checked before sending rather
// than only after. Settling it here also ends the day for the scheduler.
if (report && report.attempts >= MAX_ATTEMPTS) {
await this.skip(report.id, report.attempts, 'max-retries', report.lastError);
return;
}
if (!report) {
const days = await this.missedDays(now);
if (days.length === 0) return;
report = await this.reportRepository.createPending(await this.collectDataPoints(days), now);
if (!report) {
throw new InstanceReportAlreadyCreatedError();
} else {
const report = await this.findOrCreateReport(slot);
if (report?.status === 'pending' && report.attempts >= MAX_ATTEMPTS) {
// Recover a stop between recording the last failure and settling the row.
await this.skip(report.id, report.attempts, 'max-retries', report.lastError);
} else if (report?.status === 'pending') {
const claimedAt = await this.reportRepository.claimForSend(report.id);
if (claimedAt) {
await this.deliverReport(report, claimedAt, licenseCert);
}
}
}
}
private async findOrCreateReport(slot: Date | null): Promise<InstanceMonitoringReport | null> {
let report = await this.reportRepository.findPending();
if (!report && slot) {
const days = await this.missedDays(slot);
if (days.length > 0) {
// `null` when a concurrent pass created the row. That pass sends it.
report = await this.reportRepository.createPending(
await this.collectDataPoints(days),
slot,
);
}
}
return report;
}
private async deliverReport(
report: InstanceMonitoringReport,
claimedAt: Date,
licenseCert: string | undefined,
): Promise<void> {
const payload = {
instanceId: this.instanceSettings.instanceId,
batchId: report.id,
@@ -157,6 +242,7 @@ export class InstanceReportingService {
...(licenseCert ? { licenseCert } : {}),
};
let accepted = false;
try {
const response = await this.http.request<unknown>({
url: INSTANCE_REPORTS_PATH,
@@ -186,41 +272,52 @@ export class InstanceReportingService {
`Instance report was rejected with status ${response.statusCode}`,
);
}
this.eventService.emit('instance-report-delivered');
accepted = true;
} 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.
const attempts = report.attempts + 1;
if (error instanceof InstanceReportRejectedError) {
await this.skip(report.id, attempts, 'rejected', message);
return;
const recorded = await this.reportRepository.recordFailure(
report.id,
claimedAt,
message,
await this.reportRepository.readDbNow(),
);
if (recorded) {
this.eventService.emit('instance-report-failed');
const attempts = report.attempts + 1;
if (error instanceof InstanceReportRejectedError) {
await this.skip(report.id, attempts, 'rejected', message);
} else if (attempts >= MAX_ATTEMPTS) {
await this.skip(report.id, attempts, 'max-retries', message);
}
}
if (attempts >= MAX_ATTEMPTS) {
await this.skip(report.id, attempts, 'max-retries', message);
if (!(error instanceof InstanceReportRejectedError)) {
throw error;
}
throw error;
}
await this.reportRepository.markDelivered(report.id, new Date());
this.logger.debug('Sent instance report', { batchId: report.id });
if (accepted) {
const recorded = await this.reportRepository.markDelivered(
report.id,
await this.reportRepository.readDbNow(),
);
if (recorded) {
this.eventService.emit('instance-report-delivered');
this.logger.debug('Sent instance report', { batchId: report.id });
}
}
}
/** Stop trying to deliver this report; the next one covers its days again. */
async skip(
private async skip(
id: string,
attempts: number,
reason: SkipReason,
lastError: string | null,
): Promise<void> {
await this.reportRepository.markSkipped(id);
this.logger.error(SKIP_MESSAGES[reason], { batchId: id, attempts, lastError });
if (await this.reportRepository.markSkipped(id)) {
this.logger.error(SKIP_MESSAGES[reason], { batchId: id, attempts, lastError });
}
}
/**
@@ -313,6 +410,33 @@ export class InstanceReportingService {
}
}
/** Epoch milliseconds of the configured time on this UTC day. */
function slotOn(reportTime: string, day: Date): number {
const [hour, minute] = reportTime.split(':').map(Number);
return Date.UTC(day.getUTCFullYear(), day.getUTCMonth(), day.getUTCDate(), hour, minute);
}
/** The UTC day of the report's slot. A legacy row without `reportDate` falls back to `createdAt`. */
function slotDay(report: InstanceMonitoringReport): Date {
return report.reportDate ? new Date(`${report.reportDate}T00:00:00.000Z`) : report.createdAt;
}
/** The latest slot at or before `now`, or `null` once yesterday's slot is past its grace. */
function dueSlot(reportTime: string, now: Date): Date | null {
const today = slotOn(reportTime, now);
const yesterday = today - Time.days.toMilliseconds;
if (now.getTime() >= today) {
return new Date(today);
}
return now.getTime() - yesterday < SLOT_GRACE_MS ? new Date(yesterday) : null;
}
function isInRetryDelay(report: InstanceMonitoringReport, now: Date): boolean {
return (
report.lastAttemptAt !== null && now.getTime() - report.lastAttemptAt.getTime() < RETRY_DELAY_MS
);
}
/** The UTC calendar day `count` days before `instant`, as `YYYY-MM-DD`. */
function utcDayBefore(instant: Date, count: number): string {
return new Date(instant.getTime() - count * Time.days.toMilliseconds).toISOString().slice(0, 10);
@@ -0,0 +1,28 @@
import { Time } from '@n8n/constants';
import { intervalFromSeconds, SystemTask } from '@n8n/decorators';
import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators';
import { InstanceReportingService } from './instance-reporting.service';
@SystemTask()
export class InstanceReportingTask implements SystemTask {
readonly name = 'instance-reporting';
readonly schedule: SystemTaskSchedule = intervalFromSeconds(15 * Time.minutes.toSeconds);
readonly effects: SystemTaskEffects = 'idempotent';
readonly placement: SystemTaskPlacement = {
scope: 'cluster',
durable: true,
runOnTakeover: true,
};
readonly maxAttempts = 1;
constructor(private readonly reportingService: InstanceReportingService) {}
async run(): Promise<void> {
await this.reportingService.sendDueReport(new Date());
}
}