From 4d1d2d88addd7598b9979f7d215ec7a575dda866 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Thu, 9 Jul 2026 23:57:29 +0200 Subject: [PATCH] fix(ocpp): implement OCPP 1.6 GetDiagnostics supersession via AbortController (#1991) MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Closes #1971 The entry guard added by PR #1987 at handleRequestGetDiagnostics prevented concurrent FTP uploads racing on the same ${chargingStationId}_logs.tar.gz archive by silently dropping the second GetDiagnostics.req with an empty GetDiagnostics.conf. That mitigated the concrete race but was not the OCPP 1.6 supersession semantic: the new request needed to abort the in-flight upload and start a fresh one, with a terminal DiagnosticsStatusNotification emitted for the superseded upload. Refactor the entry guard from skip-second into abort-then-install: - Add `activeDiagnosticsAbortController?: AbortController` to `OCPP16StationState` (alphabetical, above the existing fields). - On supersession, the entry guard calls `stationState.activeDiagnosticsAbortController?.abort()` and installs a fresh controller. `diagnosticsUploadInProgress` stays `true` across the handoff so the §4.4 / §7.24 trigger cross-check never observes a false-idle window. - `basic-ftp` v6 does not natively consume `AbortSignal`; the documented interruption path is `Client.close()`, which invokes `FtpContext.close()` and rejects the pending `uploadFrom` task with `User closed client during task`. The handler wires `signal.addEventListener('abort', () => ftpClient?.close(), { once: true })`. - The existing `catch` at L1390 already emits `DiagnosticsStatusNotification(UploadFailed)` for any thrown error, so a close-triggered rejection flows through unchanged — no second emission needed. OCPP 1.6 `DiagnosticsStatusNotification.req` (§6.17) does not carry a `requestId`; the terminal for the superseded upload is uncorrelated by design, only its temporal position tells the CSMS which upload it belonged to. - The `finally` clause identity-guards its cleanup (`if (stationState.activeDiagnosticsAbortController === abortController)`) so a superseded handler's late unwind cannot clobber the new lifecycle's state. Cleaner than the microtask-yield primitive sketched in the initial design because it removes the ordering dependency between the two handlers' async continuations. - `resetStationState` aborts the in-flight controller before the base template deletes the WeakMap entry, following the cancel-before-delete ordering documented on `OCPP20IncomingRequestService.resetStationState`. Mirrors the OCPP 2.0.1 log-upload supersession pattern (commits 29a8330f, ac6ed218, 64f384fa, be29b4c8) but does not import `AcceptedCanceled` — that response semantic is a 2.0.1 concept absent from OCPP 1.6 `GetDiagnostics.conf` (§6.26). --- .../ocpp/1.6/OCPP16IncomingRequestService.ts | 93 ++++- ...P16IncomingRequestService-Firmware.test.ts | 45 +-- ...omingRequestService-GetDiagnostics.test.ts | 340 ++++++++++++++++++ 3 files changed, 421 insertions(+), 57 deletions(-) create mode 100644 tests/charging-station/ocpp/1.6/OCPP16IncomingRequestService-GetDiagnostics.test.ts diff --git a/src/charging-station/ocpp/1.6/OCPP16IncomingRequestService.ts b/src/charging-station/ocpp/1.6/OCPP16IncomingRequestService.ts index 609bac06..0b6c5340 100644 --- a/src/charging-station/ocpp/1.6/OCPP16IncomingRequestService.ts +++ b/src/charging-station/ocpp/1.6/OCPP16IncomingRequestService.ts @@ -152,6 +152,29 @@ const moduleName = 'OCPP16IncomingRequestService' * rather than requestId-presence sentinels. */ interface OCPP16StationState { + /** + * `AbortController` scoped to the in-flight diagnostics FTP upload + * spawned by the FTP branch of `handleRequestGetDiagnostics`. Set + * synchronously with `diagnosticsUploadInProgress = true` at entry; + * released by the handler's `finally` clause via identity guard + * (comparison against the local variable) so a superseding + * request that has already overwritten the state reference is not + * clobbered by the superseded handler's late cleanup. Aborted from + * two sites: + * 1. `handleRequestGetDiagnostics`'s FTP-branch entry guard when a + * superseding request arrives while a prior lifecycle is still + * in flight. The `abort` listener closes the `basic-ftp` client, + * which rejects the pending `uploadFrom` promise with + * `User closed client during task`; the outer catch then emits + * the terminal `DiagnosticsStatusNotification(UploadFailed)` per + * OCPP 1.6 §4.4 / §7.24 (uncorrelated terminal — §6.17 does not + * carry a `requestId`, so the temporal position tells the CSMS + * which upload it belonged to). + * 2. {@link OCPP16IncomingRequestService.resetStationState} on + * `stop()`, following the cancel-before-delete ordering documented + * on {@link OCPP20IncomingRequestService.resetStationState}. + */ + activeDiagnosticsAbortController?: AbortController /** * `setTimeout` handle for a deferred `updateFirmwareSimulation` * scheduled by the `UPDATE_FIRMWARE` event listener when the incoming @@ -169,13 +192,19 @@ interface OCPP16StationState { * `stationInfo.diagnosticsStatus` left at `Uploading` after an * exception unwind does not leak to the CSMS, per OCPP 1.6 §4.4 / * §7.24 (Idle when not busy uploading diagnostics). - * 2. `handleRequestGetDiagnostics`'s FTP-branch entry guard: - * early-returns the empty `GetDiagnostics.conf` payload when a - * prior lifecycle is still in flight, preventing concurrent FTP - * uploads from racing on the same - * `${chargingStationId}_logs.tar.gz` archive filename and - * emitting duplicate `DiagnosticsStatusNotification` progress - * messages to the CSMS. + * 2. `handleRequestGetDiagnostics`'s FTP-branch entry guard: indicates + * that a prior lifecycle is still in flight and therefore must be + * aborted before the new one claims the state. The abort path + * piggybacks on `activeDiagnosticsAbortController`; the flag stays + * `true` synchronously across the abort-then-install handoff so + * the trigger cross-check above cannot observe a false idle window + * during supersession. Once the superseding handler resolves + * (normally, via early return, or via exception) its + * identity-guarded `finally` clears the flag — the superseded + * handler's terminal `UploadFailed` emission may still be in + * flight after that point, and a trigger arriving during that + * narrow window will observe `Idle` (transport already closed by + * `ftpClient?.close()`, per §4.4 "not busy uploading"). */ diagnosticsUploadInProgress?: boolean /** @@ -679,7 +708,7 @@ export class OCPP16IncomingRequestService extends OCPPIncomingRequestService { - delete stationState.deferredFirmwareUpdateTimer + stationState.deferredFirmwareUpdateTimer = undefined this.updateFirmwareSimulation(chargingStation).catch((error: unknown) => { logger.error( `${chargingStation.logPrefix()} ${moduleName}.constructor: UpdateFirmware simulation error:`, @@ -717,14 +746,19 @@ export class OCPP16IncomingRequestService extends OCPPIncomingRequestService { + ftpClient?.close() + } + abortController.signal.addEventListener('abort', abortListener, { once: true }) + stationState.activeDiagnosticsAbortController = abortController stationState.diagnosticsUploadInProgress = true try { const logConfiguration = Configuration.getConfigurationSection( @@ -1355,7 +1408,11 @@ export class OCPP16IncomingRequestService extends OCPPIncomingRequestService { undefined ) }) - - await it('should skip the diagnostics FTP upload when diagnosticsUploadInProgress is already set (concurrent invocation safety)', async t => { - // Regression guard: two concurrent GetDiagnostics.req arriving via - // the base-class dispatcher's fire-and-forget WebSocket message - // handling must not spawn concurrent FTP uploads racing on the same - // ${chargingStationId}_logs.tar.gz archive filename or emit - // duplicate DiagnosticsStatusNotification progress messages to the - // CSMS. - - // Arrange - const { incomingRequestService, station, testableService } = context - upsertConfigurationKey( - station, - OCPP16StandardParametersKey.SupportedFeatureProfiles, - 'Core,FirmwareManagement' - ) - const plumbing = incomingRequestService as unknown as { - getOrCreateStationState: (cs: ChargingStation) => { diagnosticsUploadInProgress?: boolean } - stationsState: WeakMap - } - plumbing.getOrCreateStationState(station).diagnosticsUploadInProgress = true - const warnSpy = t.mock.method(logger, 'warn') - - // Act - const response = await testableService.handleRequestGetDiagnostics(station, { - location: 'ftp://localhost/diagnostics', - }) - - // Assert - assert.strictEqual(Object.keys(response).length, 0) - assert.strictEqual(plumbing.stationsState.get(station)?.diagnosticsUploadInProgress, true) - const warnMessages = warnSpy.mock.calls.map(call => call.arguments[0] as unknown as string) - assert.ok( - warnMessages.some(m => m.includes('a diagnostics upload lifecycle is already in flight')), - `expected an "already in flight" warning, got: ${JSON.stringify(warnMessages)}` - ) - }) }) // §6.4: UpdateFirmware diff --git a/tests/charging-station/ocpp/1.6/OCPP16IncomingRequestService-GetDiagnostics.test.ts b/tests/charging-station/ocpp/1.6/OCPP16IncomingRequestService-GetDiagnostics.test.ts new file mode 100644 index 00000000..2b948394 --- /dev/null +++ b/tests/charging-station/ocpp/1.6/OCPP16IncomingRequestService-GetDiagnostics.test.ts @@ -0,0 +1,340 @@ +/** + * @file Tests for OCPP16IncomingRequestService GetDiagnostics supersession semantics + * @description Unit tests for OCPP 1.6 GetDiagnostics (§6.1) supersession + * lifecycle backed by `activeDiagnosticsAbortController` on + * `OCPP16StationState`. Exercises the entry guard's abort-then-install + * pattern and the identity-guarded finally cleanup that lets a superseding + * handler claim state fields without the superseded handler's late cleanup + * clobbering them. + * + * Test strategy: `basic-ftp` `Client` is not directly mockable in this + * project — the import is module-scoped, there is no dependency-injection + * seam, and `--experimental-test-module-mocks` is not enabled by the test + * script. The FTP branch is instead exercised through the no-log-file + * early-exit inside its `try` block (mocking + * `Configuration.getConfigurationSection` so `logConfiguration.file` is + * `undefined`). That path still claims state (`abortController` set, + * `diagnosticsUploadInProgress = true`), enters the `try`, returns the empty + * response, and runs `finally` — covering the entire state lifecycle without + * touching the network. Supersession is verified by pre-populating + * `stationState.activeDiagnosticsAbortController` on a spy + * `AbortController` and asserting its `signal.aborted === true` after a + * superseding dispatch returns; the peek-inside-the-try trick captures the + * fresh controller before `finally` clears it. + */ + +import assert from 'node:assert/strict' +import { afterEach, beforeEach, describe, it } from 'node:test' + +import type { ChargingStation } from '../../../../src/charging-station/index.js' +import type { OCPP16IncomingRequestService } from '../../../../src/charging-station/ocpp/1.6/OCPP16IncomingRequestService.js' +import type { GetDiagnosticsRequest } from '../../../../src/types/index.js' + +import { ConfigurationSection, OCPP16StandardParametersKey } from '../../../../src/types/index.js' +import { Configuration } from '../../../../src/utils/index.js' +import { standardCleanup } from '../../../helpers/TestLifecycleHelpers.js' +import { + createOCPP16IncomingRequestTestContext, + type OCPP16IncomingRequestTestContext, + setMockRequestHandler, + upsertConfigurationKey, +} from './OCPP16TestUtils.js' + +interface OCPP16StationStateShape { + activeDiagnosticsAbortController?: AbortController + diagnosticsUploadInProgress?: boolean +} + +interface PlumbingAccess { + getOrCreateStationState: (chargingStation: ChargingStation) => OCPP16StationStateShape + stationsState: WeakMap +} + +const asPlumbing = (service: OCPP16IncomingRequestService): PlumbingAccess => + service as unknown as PlumbingAccess + +const enableFirmwareManagement = (station: ChargingStation): void => { + upsertConfigurationKey( + station, + OCPP16StandardParametersKey.SupportedFeatureProfiles, + 'Core,FirmwareManagement' + ) +} + +const FTP_LOCATION = 'ftp://localhost/diagnostics' +const HTTP_LOCATION = 'http://localhost/diagnostics' + +await describe('OCPP16IncomingRequestService — GetDiagnostics supersession', async () => { + let context: OCPP16IncomingRequestTestContext + + beforeEach(() => { + context = createOCPP16IncomingRequestTestContext() + }) + + afterEach(() => { + standardCleanup() + }) + + await it('should claim activeDiagnosticsAbortController on entry and release both fields in the finally block', async t => { + // Arrange + const { incomingRequestService, station, testableService } = context + enableFirmwareManagement(station) + const plumbing = asPlumbing(incomingRequestService) + + let midHandlerController: AbortController | undefined + let midHandlerInProgress: boolean | undefined + t.mock.method(Configuration, 'getConfigurationSection', (section: ConfigurationSection) => { + if (section === ConfigurationSection.log) { + const midState = plumbing.stationsState.get(station) + midHandlerController = midState?.activeDiagnosticsAbortController + midHandlerInProgress = midState?.diagnosticsUploadInProgress + return { file: undefined } + } + return {} + }) + + const request: GetDiagnosticsRequest = { location: FTP_LOCATION } + + // Act + const response = await testableService.handleRequestGetDiagnostics(station, request) + + // Assert — mid-handler state was claimed with a fresh, non-aborted controller + assert.strictEqual(Object.keys(response).length, 0) + assert.notStrictEqual( + midHandlerController, + undefined, + 'activeDiagnosticsAbortController must be set before the log-configuration read' + ) + assert.ok(midHandlerController instanceof AbortController) + assert.strictEqual(midHandlerController.signal.aborted, false) + assert.strictEqual(midHandlerInProgress, true) + + // Assert — finally released both fields symmetrically + const finalState = plumbing.stationsState.get(station) + assert.strictEqual(finalState?.activeDiagnosticsAbortController, undefined) + assert.strictEqual(finalState?.diagnosticsUploadInProgress, undefined) + }) + + await it('should abort the prior in-flight controller and install a fresh controller when a superseding request arrives', async t => { + // Arrange — pre-populate an in-flight lifecycle so the next dispatch takes + // the supersession branch of the entry guard. + const { incomingRequestService, station, testableService } = context + enableFirmwareManagement(station) + const plumbing = asPlumbing(incomingRequestService) + + const priorController = new AbortController() + const priorState = plumbing.getOrCreateStationState(station) + priorState.activeDiagnosticsAbortController = priorController + priorState.diagnosticsUploadInProgress = true + + let midHandlerController: AbortController | undefined + t.mock.method(Configuration, 'getConfigurationSection', (section: ConfigurationSection) => { + if (section === ConfigurationSection.log) { + midHandlerController = plumbing.stationsState.get(station)?.activeDiagnosticsAbortController + return { file: undefined } + } + return {} + }) + + // Act + const response = await testableService.handleRequestGetDiagnostics(station, { + location: FTP_LOCATION, + }) + + // Assert — prior controller was aborted by the entry guard + assert.strictEqual(Object.keys(response).length, 0) + assert.strictEqual( + priorController.signal.aborted, + true, + 'the prior in-flight AbortController must be signalled on supersession' + ) + + // Assert — a fresh, non-aborted controller replaced the prior one mid-handler + assert.notStrictEqual(midHandlerController, undefined) + assert.notStrictEqual( + midHandlerController, + priorController, + 'the superseding handler must install a fresh AbortController' + ) + assert.strictEqual(midHandlerController?.signal.aborted, false) + + // Assert — finally released both fields (identity guard matched) + const finalState = plumbing.stationsState.get(station) + assert.strictEqual(finalState?.activeDiagnosticsAbortController, undefined) + assert.strictEqual(finalState?.diagnosticsUploadInProgress, undefined) + }) + + await it('should start subsequent dispatches with a fresh, non-aborted controller after a supersession', async t => { + // Arrange — pre-populate to force B to supersede A. Then a follow-up C + // dispatches on the clean state left by B's finally. + const { incomingRequestService, station, testableService } = context + enableFirmwareManagement(station) + const plumbing = asPlumbing(incomingRequestService) + + const priorController = new AbortController() + const priorState = plumbing.getOrCreateStationState(station) + priorState.activeDiagnosticsAbortController = priorController + priorState.diagnosticsUploadInProgress = true + + const observedControllers: (AbortController | undefined)[] = [] + t.mock.method(Configuration, 'getConfigurationSection', (section: ConfigurationSection) => { + if (section === ConfigurationSection.log) { + observedControllers.push( + plumbing.stationsState.get(station)?.activeDiagnosticsAbortController + ) + return { file: undefined } + } + return {} + }) + + // Act — B (supersedes A), then C (fresh) + await testableService.handleRequestGetDiagnostics(station, { location: FTP_LOCATION }) + await testableService.handleRequestGetDiagnostics(station, { location: FTP_LOCATION }) + + // Assert — B and C each installed their own fresh controller, distinct from + // the pre-populated prior controller and from each other. + assert.strictEqual(observedControllers.length, 2) + const [duringB, duringC] = observedControllers + assert.notStrictEqual(duringB, undefined) + assert.notStrictEqual(duringC, undefined) + assert.notStrictEqual(duringB, priorController) + assert.notStrictEqual(duringC, priorController) + assert.notStrictEqual( + duringB, + duringC, + 'each dispatch must install its own fresh AbortController' + ) + assert.strictEqual(duringC?.signal.aborted, false) + + const finalState = plumbing.stationsState.get(station) + assert.strictEqual(finalState?.activeDiagnosticsAbortController, undefined) + assert.strictEqual(finalState?.diagnosticsUploadInProgress, undefined) + }) + + await it('should abort activeDiagnosticsAbortController and delete the WeakMap entry on stop()', () => { + // Arrange — pre-populate an in-flight lifecycle, then stop. + const { incomingRequestService, station } = context + const plumbing = asPlumbing(incomingRequestService) + const inflightController = new AbortController() + const state = plumbing.getOrCreateStationState(station) + state.activeDiagnosticsAbortController = inflightController + state.diagnosticsUploadInProgress = true + + // Act + incomingRequestService.stop(station) + + // Assert — cancel-before-delete ordering: the signal must fire before the + // base template deletes the WeakMap entry, and the entry must be gone. + assert.strictEqual( + inflightController.signal.aborted, + true, + 'stop() must signal the in-flight diagnostics AbortController' + ) + assert.strictEqual(plumbing.stationsState.has(station), false) + }) + + await it('should not install activeDiagnosticsAbortController for non-FTP locations', async () => { + // Arrange — the else branch calls ocppRequestService.requestHandler to send + // DiagnosticsStatusNotification(UploadFailed); stub it out. + const { incomingRequestService, station, testableService } = context + enableFirmwareManagement(station) + setMockRequestHandler(station, async () => Promise.resolve({})) + const plumbing = asPlumbing(incomingRequestService) + + // Act + const response = await testableService.handleRequestGetDiagnostics(station, { + location: HTTP_LOCATION, + }) + + // Assert — the AbortController plumbing lives inside the FTP branch only. + assert.strictEqual(Object.keys(response).length, 0) + const state = plumbing.stationsState.get(station) + assert.strictEqual(state?.activeDiagnosticsAbortController, undefined) + assert.strictEqual(state?.diagnosticsUploadInProgress, undefined) + }) + + await it('should keep diagnosticsUploadInProgress true across the supersession handoff so the trigger cross-check never sees false-idle', async t => { + // Arrange — pre-populate an in-flight lifecycle. During the superseding + // handler's try block (before finally clears state), the flag must remain + // true. The trigger cross-check at handleRequestTriggerMessage's + // DiagnosticsStatusNotification case reads exactly this predicate + // (`stationsState.get(chargingStation)?.diagnosticsUploadInProgress === true`); + // JS single-threaded semantics guarantee the read cannot interleave with a + // partial state mutation. + const { incomingRequestService, station, testableService } = context + enableFirmwareManagement(station) + const plumbing = asPlumbing(incomingRequestService) + + const priorController = new AbortController() + const priorState = plumbing.getOrCreateStationState(station) + priorState.activeDiagnosticsAbortController = priorController + priorState.diagnosticsUploadInProgress = true + + let midHandlerInProgress: boolean | undefined + t.mock.method(Configuration, 'getConfigurationSection', (section: ConfigurationSection) => { + if (section === ConfigurationSection.log) { + midHandlerInProgress = plumbing.stationsState.get(station)?.diagnosticsUploadInProgress + return { file: undefined } + } + return {} + }) + + // Act + await testableService.handleRequestGetDiagnostics(station, { + location: FTP_LOCATION, + }) + + // Assert — flag observed as true mid-handoff, cleared after finally. + assert.strictEqual( + midHandlerInProgress, + true, + 'diagnosticsUploadInProgress must remain true across the supersession handoff' + ) + assert.strictEqual(plumbing.stationsState.get(station)?.diagnosticsUploadInProgress, undefined) + }) + + await it('should isolate supersession state between stations so an abort on A does not touch B', async t => { + // Arrange — one shared service, two stations with independent state entries. + const { incomingRequestService, station: stationA, testableService } = context + const { station: stationB } = createOCPP16IncomingRequestTestContext({ + baseName: 'get-diag-iso-b', + }) + enableFirmwareManagement(stationA) + + const plumbing = asPlumbing(incomingRequestService) + const priorAController = new AbortController() + const stateA = plumbing.getOrCreateStationState(stationA) + stateA.activeDiagnosticsAbortController = priorAController + stateA.diagnosticsUploadInProgress = true + + const priorBController = new AbortController() + const stateB = plumbing.getOrCreateStationState(stationB) + stateB.activeDiagnosticsAbortController = priorBController + stateB.diagnosticsUploadInProgress = true + + t.mock.method(Configuration, 'getConfigurationSection', (section: ConfigurationSection) => { + if (section === ConfigurationSection.log) { + return { file: undefined } + } + return {} + }) + + // Act — supersede A only. + await testableService.handleRequestGetDiagnostics(stationA, { + location: FTP_LOCATION, + }) + + // Assert — A aborted and cleaned. + assert.strictEqual(priorAController.signal.aborted, true) + const finalA = plumbing.stationsState.get(stationA) + assert.strictEqual(finalA?.activeDiagnosticsAbortController, undefined) + assert.strictEqual(finalA?.diagnosticsUploadInProgress, undefined) + + // Assert — B fully untouched (controller reference preserved, not aborted). + assert.strictEqual(priorBController.signal.aborted, false) + const finalB = plumbing.stationsState.get(stationB) + assert.ok(finalB, 'station B state entry must survive station A supersession') + assert.strictEqual(finalB.activeDiagnosticsAbortController, priorBController) + assert.strictEqual(finalB.diagnosticsUploadInProgress, true) + }) +}) -- 2.53.0