diff --git a/packages/cli/eslint.config.mjs b/packages/cli/eslint.config.mjs index 2a477cb73e7..3187f144768 100644 --- a/packages/cli/eslint.config.mjs +++ b/packages/cli/eslint.config.mjs @@ -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', ], diff --git a/packages/cli/oxlint.config.mts b/packages/cli/oxlint.config.mts index a8292c2c53e..50f4356755d 100644 --- a/packages/cli/oxlint.config.mts +++ b/packages/cli/oxlint.config.mts @@ -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', ], diff --git a/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md b/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md index 13464d8b83d..09c05421415 100644 --- a/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md +++ b/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md @@ -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 ` 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. diff --git a/packages/cli/src/modules/instance-reporting.ee/README.md b/packages/cli/src/modules/instance-reporting.ee/README.md index 9d9677f749a..466a15a2e0c 100644 --- a/packages/cli/src/modules/instance-reporting.ee/README.md +++ b/packages/cli/src/modules/instance-reporting.ee/README.md @@ -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 `/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 `/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. | diff --git a/packages/cli/src/modules/instance-reporting.ee/RETRIES.md b/packages/cli/src/modules/instance-reporting.ee/RETRIES.md index c16cf007d23..f3783c101b4 100644 --- a/packages/cli/src/modules/instance-reporting.ee/RETRIES.md +++ b/packages/cli/src/modules/instance-reporting.ee/RETRIES.md @@ -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?
(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 diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-multi-main.integration.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-multi-main.integration.test.ts new file mode 100644 index 00000000000..dc30a7f0b1d --- /dev/null +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-multi-main.integration.test.ts @@ -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 = []; + + 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[1], + ); + const config = new InstanceReportingConfig(); + config.instanceReportingBaseUrl = 'https://receiver.test'; + const insights = mock(); + insights.getDailyExecutionTotals.mockResolvedValue(new Map([[REPORT_DAY, 5]])); + const metrics = mock(); + metrics.getLicenseRenewalMetrics.mockResolvedValue( + mock>>({ + productionRootExecutions: 10, + }), + ); + const events = mock(); + const service = new InstanceReportingService( + config, + reports, + mock({ getReportTime: async () => '00:00' }), + insights, + new InsightsConfig(), + mock({ instanceId: 'shared-instance', instanceType: 'main', isLeader }), + metrics, + mock({ 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); + }, + ); +}); diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-scheduler.service.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-scheduler.service.test.ts deleted file mode 100644 index cf972f6439b..00000000000 --- a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-scheduler.service.test.ts +++ /dev/null @@ -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; - reportRepository: Mocked; - settingsService: Mocked; - instanceSettings: Mocked; - eventService: Mocked; - /** 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(); - - const reportRepository = mock(); - reportRepository.hasSettledToday.mockResolvedValue(false); - - const settingsService = mock(); - settingsService.getReportTime.mockResolvedValue(REPORT_TIME); - - const instanceSettings = mock({ - instanceType: 'main', - instanceRole: isLeader ? 'leader' : 'follower', - isLeader, - }); - - const serverStartedListeners: Array<() => void> = []; - const eventService = mock(); - eventService.once.mockImplementation((event, listener) => { - if (event === 'server-started') { - if (autoStart) (listener as () => void)(); - else serverStartedListeners.push(listener as () => void); - } - return eventService; - }); - eventService.off.mockImplementation((event, listener) => { - if (event === 'server-started') { - const index = serverStartedListeners.indexOf(listener as () => void); - if (index !== -1) serverStartedListeners.splice(index, 1); - } - return eventService; - }); - - const scheduler = new InstanceReportingScheduler( - 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({ - 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(); - }); - }); -}); diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts index ae7cf5b6a02..aa689f75657 100644 --- a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts @@ -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['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; } 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 }); diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.module.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.module.test.ts index 35d70afbdab..36642e8b7ef 100644 --- a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.module.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.module.test.ts @@ -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(); - 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 () => { diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts index 698b054c0d3..0a141e16316 100644 --- a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts @@ -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()), @@ -82,6 +81,7 @@ function makeReport(overrides: Partial = {}): Instance interface Harness { service: InstanceReportingService; reportRepository: Mocked; + settingsService: Mocked; insightsService: Mocked; insightsConfig: InsightsConfig; license: Mocked; @@ -93,6 +93,11 @@ interface Harness { function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness { const reportRepository = mock(); 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(); + settingsService.getReportTime.mockResolvedValue('07:42'); + const insightsService = mock(); // 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({ 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) { + 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(); diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.task.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.task.test.ts new file mode 100644 index 00000000000..36aa8914e53 --- /dev/null +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.task.test.ts @@ -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(); + 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(); + scheduler.isActive.mockReturnValue(schedulerActive); + const registrar = mock(); + const runner = new SystemTaskRunner( + mockLogger(), + metadata, + scheduler, + registrar, + new SystemTaskScheduledJobOwner(mock()), + mock({ generic: { timezone: 'UTC' }, scheduler: { enabledForSystemTasks } }), + mock({ + instanceType: 'main', + instanceRole: isLeader ? 'leader' : 'follower', + isLeader, + }), + mock(), + mock(), + 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(), + ); + }, + ); + }); +}); diff --git a/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts b/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts index 13dfe68985a..8a1ee17bde3 100644 --- a/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts +++ b/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts @@ -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; diff --git a/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts b/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts index 43be3fcc102..a8a55758af8 100644 --- a/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts @@ -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, diff --git a/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts b/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts index 258f8cf46f2..aa923b0969f 100644 --- a/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts +++ b/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts @@ -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 { 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 { + 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 { + 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 { + 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 { - 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 { - 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 { + 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 { - 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 { + 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 { - await this.update({ id, status: 'pending' }, { status: 'skipped_after_max_retries' }); + async markSkipped(id: string): Promise { + const result = await this.update( + { id, status: 'pending' }, + { status: 'skipped_after_max_retries' }, + ); + return result.affected === 1; } } diff --git a/packages/cli/src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts b/packages/cli/src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts deleted file mode 100644 index 34cee517665..00000000000 --- a/packages/cli/src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts +++ /dev/null @@ -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 { - 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 { - 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 { - 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(); -} diff --git a/packages/cli/src/modules/instance-reporting.ee/instance-reporting.module.ts b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.module.ts index ebb3ae14634..1f2cbc0a84f 100644 --- a/packages/cli/src/modules/instance-reporting.ee/instance-reporting.module.ts +++ b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.module.ts @@ -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 []; } /** diff --git a/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts index 45878bcd9fc..abe8874508f 100644 --- a/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts +++ b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts @@ -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 = { '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 = { }; /** - * 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 { + 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 { + 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 { + async sendReport(slot: Date | null): Promise { 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 { + 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 { 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({ 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 { - 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); diff --git a/packages/cli/src/modules/instance-reporting.ee/instance-reporting.task.ts b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.task.ts new file mode 100644 index 00000000000..a79fca0721d --- /dev/null +++ b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.task.ts @@ -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 { + await this.reportingService.sendDueReport(new Date()); + } +}