From 6119452049739c05506351a9589d753e84c59e9f Mon Sep 17 00:00:00 2001 From: Daniel <7558512+DerGenaue@users.noreply.github.com> Date: Sat, 18 Jul 2026 00:17:54 +0200 Subject: [PATCH] fix(ui-server): make UI broadcast-channel commands complete reliably (#2018) (#2020) MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * fix(ui-server): time out broadcast-channel requests so UI commands cannot hang A UI control command (start/stop/delete a station) is dispatched over the worker broadcast channel and the UI service waits for a fixed number of worker responses, sampled at send time. If a targeted worker never replies -- e.g. the station is deleted while the command is in flight -- that count is never reached, so the request is never completed or released and the client waits forever. Read commands keep working, which masks the wedge. Arm a per-request safety-net timeout when a broadcast-channel request is sent. On expiry the request is completed with a failure (reporting the charging stations that did reply successfully) and both the request and response aggregation state are released, so the caller gets an answer instead of hanging. The timeout is cleared on normal completion, on a dispatch failure, and on service stop. Closes #2018. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Daniel <7558512+DerGenaue@users.noreply.github.com> * fix(ui-server): complete broadcast requests by target set and reconcile on station deletion (#2018) Broadcast-channel request completion was gated on a frozen integer count sampled at send time, blind to which stations were targeted. When the set of stations changed mid-flight the aggregator could not reconcile the drift, producing failure modes on top of the hang the safety-net timeout already backstops: - Hang until timeout when a targeted station is deleted mid-flight (the count is never reached). - Late double-completion: after release the count reads 0, so a late reply re-enters and 1 >= 0 re-completes the request. - False success: an empty explicit hashIds array degraded to a broadcast-to-all instead of failing. Track the outstanding responders as a Set of target hashIds and make the Set the single source of truth for completion: 1. Snapshot the resolved targets (validated explicit hashIds, or the live station set for a broadcast) into the request context; complete when the Set empties. AbstractUIServer.deleteChargingStationData fans out AbstractUIService.reconcileDeletedStation, which drops the departed hashId from every in-flight request and completes any that empty with the truthful aggregated payload. A DELETE_CHARGING_STATIONS request is left untouched by reconciliation: its targets self-delete yet still post their command reply on a separate transport, so that reply (not the racing `deleted` event) is the completion source; a worker that dies mid-delete is covered by the timeout. 2. Drop untracked responses (unknown/released request, or a hashId that is not an outstanding responder) in the response handler, closing the late double-completion re-entry. A reply without a hashId for a still-tracked request is dropped and logged distinctly so the resulting timeout is diagnosable. 3. Reject unknown and empty explicit targets instead of degrading to a broadcast-to-all. The 60 s safety-net timeout stays as a pure backstop for genuine worker crash/deadlock; normal and reconciled completion clear its timer. The completion accessor is named getBroadcastChannelOutstandingResponseCount to reflect that it returns the responses still outstanding, not a frozen total. Duplicate-identity false success and orphan-worker termination are not addressed here (a hashId-keyed set cannot disambiguate two workers sharing an identity) and are deferred to follow-up changes. The deprecated HTTP transport bypasses aggregation and is likewise out of scope. Tests cover reconcile-on-delete, the untracked-response guard, unknown and empty target rejection, the DELETE self-target ordering (reconcile-first completes via the reply; no reply falls back to the timeout), reconcile that does not empty the set, two concurrent requests reconciled by one deletion, a hashId-less reply, and the previously untested timeout behaviors (partial-success payload, dispatch-failure and stop() timer clearing, and the completion-race guard). Closes #2018. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Jérôme Benoit --------- Signed-off-by: Daniel <7558512+DerGenaue@users.noreply.github.com> Signed-off-by: Jérôme Benoit Co-authored-by: Claude Opus 4.8 (1M context) Co-authored-by: Jérôme Benoit --- .../UIServiceWorkerBroadcastChannel.ts | 135 +++-- .../ui-server/AbstractUIServer.ts | 15 +- .../ui-services/AbstractUIService.ts | 137 ++++- src/utils/Constants.ts | 7 + .../ui-server/UIServerTestConstants.ts | 2 +- .../ui-services/AbstractUIService.test.ts | 552 +++++++++++++++++- 6 files changed, 766 insertions(+), 82 deletions(-) diff --git a/src/charging-station/broadcast-channel/UIServiceWorkerBroadcastChannel.ts b/src/charging-station/broadcast-channel/UIServiceWorkerBroadcastChannel.ts index 5460434c..f0d26d1b 100644 --- a/src/charging-station/broadcast-channel/UIServiceWorkerBroadcastChannel.ts +++ b/src/charging-station/broadcast-channel/UIServiceWorkerBroadcastChannel.ts @@ -9,20 +9,15 @@ import { type MessageEvent, type ResponsePayload, ResponseStatus, + type UUIDv4, } from '../../types/index.js' import { isNotEmptyArray, logger } from '../../utils/index.js' import { WorkerBroadcastChannel } from './WorkerBroadcastChannel.js' const moduleName = 'UIServiceWorkerBroadcastChannel' -interface Responses { - responses: BroadcastChannelResponsePayload[] - responsesExpected: number - responsesReceived: number -} - export class UIServiceWorkerBroadcastChannel extends WorkerBroadcastChannel { - private readonly responses: Map + private readonly responses: Map private readonly uiService: AbstractUIService constructor (uiService: AbstractUIService) { @@ -30,11 +25,42 @@ export class UIServiceWorkerBroadcastChannel extends WorkerBroadcastChannel { this.uiService = uiService this.onmessage = this.responseHandler.bind(this) as (message: unknown) => void this.onmessageerror = this.messageErrorHandler.bind(this) as (message: unknown) => void - this.responses = new Map() + this.responses = new Map() + } + + /** + * Completes a broadcast request whose safety-net timeout fired before all + * expected worker responses arrived (e.g. a targeted station was deleted + * mid-flight), so the caller is not left waiting forever, then releases the + * aggregation state. + * @param uuid - Request identifier. + */ + public completeExpiredRequest (uuid: UUIDv4): void { + if (this.uiService.getBroadcastChannelOutstandingResponseCount(uuid) === 0) { + // Already completed and released by the normal response path. + return + } + try { + this.uiService.sendResponse(uuid, this.buildTimeoutResponsePayload(uuid)) + } finally { + this.responses.delete(uuid) + this.uiService.deleteBroadcastChannelRequest(uuid) + } + } + + /** + * Completes a broadcast request whose last outstanding responder was a + * station deleted mid-flight: the surviving stations have all replied, so + * the request is completed with the truthful aggregated payload (the deleted + * station simply absent) instead of hanging. + * @param uuid - Request identifier. + */ + public completeReconciledRequest (uuid: UUIDv4): void { + this.completeRequest(uuid) } - private buildResponsePayload (uuid: string): ResponsePayload { - const responsesArray = this.responses.get(uuid)?.responses ?? [] + private buildResponsePayload (uuid: UUIDv4): ResponsePayload { + const responsesArray = this.responses.get(uuid) ?? [] const responsesStatus = isNotEmptyArray(responsesArray) && responsesArray.every(response => response.status === ResponseStatus.SUCCESS) @@ -43,8 +69,7 @@ export class UIServiceWorkerBroadcastChannel extends WorkerBroadcastChannel { return { hashIdsSucceeded: responsesArray .map(response => { - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (response?.hashId != null && response?.status === ResponseStatus.SUCCESS) { + if (response.hashId != null && response.status === ResponseStatus.SUCCESS) { return response.hashId } return undefined @@ -54,8 +79,7 @@ export class UIServiceWorkerBroadcastChannel extends WorkerBroadcastChannel { ...(responsesStatus === ResponseStatus.FAILURE && { hashIdsFailed: responsesArray .map(response => { - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (response?.hashId != null && response?.status === ResponseStatus.FAILURE) { + if (response.hashId != null && response.status === ResponseStatus.FAILURE) { return response.hashId } return undefined @@ -65,8 +89,7 @@ export class UIServiceWorkerBroadcastChannel extends WorkerBroadcastChannel { ...(responsesStatus === ResponseStatus.FAILURE && { responsesFailed: responsesArray .map(response => { - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (response?.status === ResponseStatus.FAILURE) { + if (response.status === ResponseStatus.FAILURE) { return response } return undefined @@ -76,6 +99,31 @@ export class UIServiceWorkerBroadcastChannel extends WorkerBroadcastChannel { } } + private buildTimeoutResponsePayload (uuid: UUIDv4): ResponsePayload { + const responsesArray = this.responses.get(uuid) ?? [] + const responsesReceived = responsesArray.length + const responsesExpected = + responsesReceived + this.uiService.getBroadcastChannelOutstandingResponseCount(uuid) + return { + errorMessage: `Timed out waiting for charging station responses (received ${responsesReceived.toString()} of ${responsesExpected.toString()})`, + hashIdsSucceeded: responsesArray + .filter(response => response.status === ResponseStatus.SUCCESS) + .map(response => response.hashId) + .filter((hashId): hashId is string => hashId != null), + status: ResponseStatus.FAILURE, + } + } + + private completeRequest (uuid: UUIDv4): void { + // Always release aggregation state, even if downstream sendResponse throws. + try { + this.uiService.sendResponse(uuid, this.buildResponsePayload(uuid)) + } finally { + this.responses.delete(uuid) + this.uiService.deleteBroadcastChannelRequest(uuid) + } + } + private messageErrorHandler (messageEvent: MessageEvent): void { logger.error( `${this.uiService.logPrefix(moduleName, 'messageErrorHandler')} Error at handling message:`, @@ -92,35 +140,38 @@ export class UIServiceWorkerBroadcastChannel extends WorkerBroadcastChannel { return } const [uuid, responsePayload] = validatedMessageEvent.data as BroadcastChannelResponse - if (!this.responses.has(uuid)) { - this.responses.set(uuid, { - responses: [responsePayload], - responsesExpected: this.uiService.getBroadcastChannelExpectedResponses(uuid), - responsesReceived: 1, - }) - } else { - const responses = this.responses.get(uuid) - if (responses != null) { - if (responses.responsesReceived < responses.responsesExpected) { - ++responses.responsesReceived - responses.responses.push(responsePayload) - } else { - logger.debug( - `${this.uiService.logPrefix(moduleName, 'responseHandler')} Received response after all expected responses:`, - { responsePayload, uuid } - ) - } + const outcome = this.uiService.recordBroadcastChannelResponse(uuid, responsePayload.hashId) + if (outcome === 'untracked') { + if ( + responsePayload.hashId == null && + this.uiService.getBroadcastChannelOutstandingResponseCount(uuid) > 0 + ) { + // Reply for a still-tracked request but without a hashId, so it cannot + // be matched to an outstanding responder; completion is left to the + // remaining replies and the safety-net timeout. Logged distinctly from a + // late/duplicate drop to make such a stall diagnosable. + logger.debug( + `${this.uiService.logPrefix(moduleName, 'responseHandler')} Dropping broadcast response without hashId for a tracked request:`, + { responsePayload, uuid } + ) + return } + // Late, duplicate or reconciled-away reply for an already-released or + // unexpected station: drop it so it cannot re-complete the request. + logger.debug( + `${this.uiService.logPrefix(moduleName, 'responseHandler')} Dropping untracked broadcast response:`, + { responsePayload, uuid } + ) + return } const responses = this.responses.get(uuid) - if (responses != null && responses.responsesReceived >= responses.responsesExpected) { - // Always release aggregation state, even if downstream sendResponse throws. - try { - this.uiService.sendResponse(uuid, this.buildResponsePayload(uuid)) - } finally { - this.responses.delete(uuid) - this.uiService.deleteBroadcastChannelRequest(uuid) - } + if (responses == null) { + this.responses.set(uuid, [responsePayload]) + } else { + responses.push(responsePayload) + } + if (outcome === 'completed') { + this.completeRequest(uuid) } } } diff --git a/src/charging-station/ui-server/AbstractUIServer.ts b/src/charging-station/ui-server/AbstractUIServer.ts index ea44d4b6..5b917236 100644 --- a/src/charging-station/ui-server/AbstractUIServer.ts +++ b/src/charging-station/ui-server/AbstractUIServer.ts @@ -214,8 +214,7 @@ export abstract class AbstractUIServer { break default: throw new BaseError( - // eslint-disable-next-line @typescript-eslint/restrict-template-expressions - `Unsupported application protocol version ${this.uiServerConfiguration.version} in '${ConfigurationSection.uiServer}' configuration section` + `Unsupported application protocol version ${String(this.uiServerConfiguration.version)} in '${ConfigurationSection.uiServer}' configuration section` ) } if ('requestTimeout' in this.httpServer) { @@ -249,7 +248,13 @@ export abstract class AbstractUIServer { } public deleteChargingStationData (hashId: string): boolean { - return this.chargingStations.delete(hashId) + const deleted = this.chargingStations.delete(hashId) + // Reconcile in-flight broadcast requests so any that were waiting on the + // departed station complete truthfully instead of hanging. + for (const uiService of this.uiServices.values()) { + uiService.reconcileDeletedStation(hashId) + } + return deleted } public getBootstrap (): IBootstrap { @@ -260,6 +265,10 @@ export abstract class AbstractUIServer { return this.chargingStations.get(hashId) } + public getChargingStationHashIds (): string[] { + return [...this.chargingStations.keys()] + } + public getChargingStationsCount (): number { return this.chargingStations.size } diff --git a/src/charging-station/ui-server/ui-services/AbstractUIService.ts b/src/charging-station/ui-server/ui-services/AbstractUIService.ts index de1f0b9d..b09f6a54 100644 --- a/src/charging-station/ui-server/ui-services/AbstractUIService.ts +++ b/src/charging-station/ui-server/ui-services/AbstractUIService.ts @@ -24,6 +24,7 @@ import { } from '../../../types/index.js' import { Configuration, + Constants, ensureError, getErrorMessage, isNotEmptyArray, @@ -50,6 +51,17 @@ export interface BroadcastChannelResponseLogContext { readonly uuid: UUIDv4 } +/** + * Outcome of recording a worker response against an in-flight broadcast + * request (see {@link AbstractUIService.recordBroadcastChannelResponse}): + * - `completed`: the responding station was the last outstanding one. + * - `outstanding`: recorded, but other stations are still expected. + * - `untracked`: the request is unknown/released or the station is not an + * outstanding responder (late, duplicate or reconciled-away reply) and must + * be dropped without aggregation. + */ +export type BroadcastChannelResponseOutcome = 'completed' | 'outstanding' | 'untracked' + interface AddChargingStationsRequestPayload extends RequestPayload { numberOfStations: number options?: ChargingStationOptions @@ -57,9 +69,10 @@ interface AddChargingStationsRequestPayload extends RequestPayload { } interface BroadcastChannelRequestContext { - readonly expectedResponses: number readonly origin: UIRequestOrigin + readonly outstandingHashIds: Set readonly procedureName: BroadcastChannelProcedureName + readonly timeout: ReturnType } export abstract class AbstractUIService { @@ -146,17 +159,62 @@ export abstract class AbstractUIService { } public deleteBroadcastChannelRequest (uuid: UUIDv4): void { + const requestContext = this.broadcastChannelRequests.get(uuid) + if (requestContext != null) { + clearTimeout(requestContext.timeout) + } this.broadcastChannelRequests.delete(uuid) } - public getBroadcastChannelExpectedResponses (uuid: UUIDv4): number { - return this.broadcastChannelRequests.get(uuid)?.expectedResponses ?? 0 + public getBroadcastChannelOutstandingResponseCount (uuid: UUIDv4): number { + return this.broadcastChannelRequests.get(uuid)?.outstandingHashIds.size ?? 0 } public logPrefix = (modName: string, methodName: string): string => { return this.uiServer.logPrefix(modName, methodName, this.version) } + public reconcileDeletedStation (hashId: string): void { + for (const [uuid, requestContext] of this.broadcastChannelRequests) { + // A DELETE_CHARGING_STATIONS request deletes its own targets: each target + // still posts its command reply, which is the reliable completion source. + // The `deleted` worker event that drives this reconciliation is posted on + // a different transport and can arrive before that reply, so reconciling + // here would complete the request as a failure against an empty response + // set and then drop the genuine reply as untracked. A DELETE target + // removed by an external cause mid-flight (so it never sends its own + // reply), and a worker that dies mid-delete, are both completed only via + // the safety-net timeout: reconcile cannot tell an external deletion from + // a self-delete, and treating either as a synthetic success would move + // completion off the authoritative reply. This is a bounded latency cost, + // never a hang or a false result. + if (requestContext.procedureName === BroadcastChannelProcedureName.DELETE_CHARGING_STATIONS) { + continue + } + if ( + requestContext.outstandingHashIds.delete(hashId) && + requestContext.outstandingHashIds.size === 0 + ) { + this.uiServiceWorkerBroadcastChannel.completeReconciledRequest(uuid) + } + } + } + + public recordBroadcastChannelResponse ( + uuid: UUIDv4, + hashId: string | undefined + ): BroadcastChannelResponseOutcome { + const requestContext = this.broadcastChannelRequests.get(uuid) + if ( + requestContext == null || + hashId == null || + !requestContext.outstandingHashIds.delete(hashId) + ) { + return 'untracked' + } + return requestContext.outstandingHashIds.size === 0 ? 'completed' : 'outstanding' + } + public async requestHandler ( request: ProtocolRequest, context: UIRequestContext = { origin: UIRequestOrigin.TRANSPORT } @@ -226,6 +284,9 @@ export abstract class AbstractUIService { public stop (): void { this.stopped = true + for (const requestContext of this.broadcastChannelRequests.values()) { + clearTimeout(requestContext.timeout) + } this.broadcastChannelRequests.clear() this.uiServiceWorkerBroadcastChannel.close() } @@ -456,42 +517,58 @@ export abstract class AbstractUIService { payload: BroadcastChannelRequestPayload, context: UIRequestContext ): void { - if (isNotEmptyArray(payload.hashIds)) { - payload.hashIds = payload.hashIds - .map(hashId => { - if (this.uiServer.hasChargingStationData(hashId)) { - return hashId - } - logger.warn( - `${this.logPrefix( - moduleName, - 'sendBroadcastChannelRequest' - )} Charging station with hashId '${hashId}' not found` - ) - return undefined - }) - .filter((hashId): hashId is string => hashId != null) - } else { - delete payload.hashIds + // An explicitly supplied hashIds array (even empty) means the caller targets + // specific stations: it must resolve to at least one live station, otherwise + // the request fails rather than silently broadcasting to every station. + if (Array.isArray(payload.hashIds)) { + payload.hashIds = payload.hashIds.filter(hashId => { + if (this.uiServer.hasChargingStationData(hashId)) { + return true + } + logger.warn( + `${this.logPrefix( + moduleName, + 'sendBroadcastChannelRequest' + )} Charging station with hashId '${hashId}' not found` + ) + return false + }) + if (!isNotEmptyArray(payload.hashIds)) { + throw new BaseError( + 'hashIds array in the request payload does not contain any valid charging station hashId' + ) + } } - const expectedNumberOfResponses = Array.isArray(payload.hashIds) - ? payload.hashIds.length - : this.uiServer.getChargingStationsCount() - if (expectedNumberOfResponses === 0) { - throw new BaseError( - 'hashIds array in the request payload does not contain any valid charging station hashId' - ) + // Snapshot the set of stations expected to respond: the validated explicit + // targets, or the live station set for a broadcast. Completion tracks this + // set rather than a frozen count, so a station deleted mid-flight can be + // reconciled instead of hanging the request. + const outstandingHashIds = new Set( + isNotEmptyArray(payload.hashIds) + ? payload.hashIds + : this.uiServer.getChargingStationHashIds() + ) + if (outstandingHashIds.size === 0) { + throw new BaseError('No charging station is available to handle the broadcast request') } + // Safety-net timeout: if not all expected worker responses arrive (e.g. a + // targeted station is deleted while the request is in flight), complete the + // request with a failure instead of leaving the caller waiting forever. + const timeout = setTimeout(() => { + this.uiServiceWorkerBroadcastChannel.completeExpiredRequest(uuid) + }, Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + timeout.unref() this.broadcastChannelRequests.set(uuid, { - expectedResponses: expectedNumberOfResponses, origin: context.origin, + outstandingHashIds, procedureName, + timeout, }) - // Rollback expected-response accounting if dispatch to the worker channel throws. + // Rollback the outstanding-response tracking if dispatch to the worker channel throws. try { this.uiServiceWorkerBroadcastChannel.sendRequest([uuid, procedureName, payload]) } catch (error) { - this.broadcastChannelRequests.delete(uuid) + this.deleteBroadcastChannelRequest(uuid) throw error } } diff --git a/src/utils/Constants.ts b/src/utils/Constants.ts index 2de368ee..d2f05726 100644 --- a/src/utils/Constants.ts +++ b/src/utils/Constants.ts @@ -155,6 +155,13 @@ export class Constants { static readonly STOP_CHARGING_STATIONS_TIMEOUT_MS = 60_000 static readonly STOP_MESSAGE_SEQUENCE_TIMEOUT_MS = 30_000 + /** + * Safety-net timeout for a UI broadcast-channel request: if the expected + * worker responses have not all arrived within this window (e.g. a targeted + * station was deleted mid-flight), the request is completed with a failure + * instead of hanging forever. + */ + static readonly UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS = 60_000 /** Divider between base units (A) and centi units (cA). */ static readonly UNIT_DIVIDER_CENTI = 100 diff --git a/tests/charging-station/ui-server/UIServerTestConstants.ts b/tests/charging-station/ui-server/UIServerTestConstants.ts index d469e248..90bec33d 100644 --- a/tests/charging-station/ui-server/UIServerTestConstants.ts +++ b/tests/charging-station/ui-server/UIServerTestConstants.ts @@ -6,7 +6,7 @@ import type { UUIDv4 } from '../../../src/types/index.js' export const TEST_UUID = '550e8400-e29b-41d4-a716-446655440000' as UUIDv4 -export const TEST_UUID_2 = '6ba7b810-9dad-11d1-80b4-00c04fd430c8' as UUIDv4 +export const TEST_UUID_2 = '6ba7b810-9dad-41d1-80b4-00c04fd430c8' as UUIDv4 export const TEST_PROCEDURES = { AUTHORIZE: 'Authorize', diff --git a/tests/charging-station/ui-server/ui-services/AbstractUIService.test.ts b/tests/charging-station/ui-server/ui-services/AbstractUIService.test.ts index a1f99660..b11987fd 100644 --- a/tests/charging-station/ui-server/ui-services/AbstractUIService.test.ts +++ b/tests/charging-station/ui-server/ui-services/AbstractUIService.test.ts @@ -8,6 +8,11 @@ import { afterEach, describe, it } from 'node:test' import type { UIServiceWorkerBroadcastChannel } from '../../../../src/charging-station/broadcast-channel/UIServiceWorkerBroadcastChannel.js' import type { AbstractUIService } from '../../../../src/charging-station/ui-server/ui-services/AbstractUIService.js' +import type { + BroadcastChannelResponse, + BroadcastChannelResponsePayload, + UUIDv4, +} from '../../../../src/types/index.js' import { BroadcastChannelProcedureName, @@ -16,9 +21,9 @@ import { ResponseStatus, UIRequestOrigin, } from '../../../../src/types/index.js' -import { logger } from '../../../../src/utils/index.js' -import { standardCleanup } from '../../../helpers/TestLifecycleHelpers.js' -import { TEST_HASH_ID, TEST_UUID } from '../UIServerTestConstants.js' +import { Constants, logger } from '../../../../src/utils/index.js' +import { standardCleanup, withMockTimers } from '../../../helpers/TestLifecycleHelpers.js' +import { TEST_HASH_ID, TEST_HASH_ID_2, TEST_UUID, TEST_UUID_2 } from '../UIServerTestConstants.js' import { createMockChargingStationData, createMockUIServerConfiguration, @@ -54,6 +59,43 @@ const registerTransportStopRequest = async (service: AbstractUIService): Promise ) } +const registerTransportStopRequestFor = async ( + service: AbstractUIService, + uuid: UUIDv4, + hashIds: string[] +): Promise => { + await service.requestHandler( + createProtocolRequest(uuid, ProcedureName.STOP_CHARGING_STATION, { hashIds }) + ) +} + +const registerTransportDeleteRequest = async ( + service: AbstractUIService, + hashIds: string[] +): Promise => { + await service.requestHandler( + createProtocolRequest(TEST_UUID, ProcedureName.DELETE_CHARGING_STATIONS, { hashIds }) + ) +} + +/** + * Deliver a worker broadcast-channel response into the service's aggregation + * path, mirroring what a charging station worker posts back on the channel. + * @param service - UI service whose worker broadcast channel receives the reply. + * @param responsePayload - The per-station response payload to deliver. + * @param uuid - Request identifier the reply belongs to (defaults to TEST_UUID). + */ +const emitWorkerResponse = ( + service: AbstractUIService, + responsePayload: BroadcastChannelResponsePayload, + uuid: UUIDv4 = TEST_UUID +): void => { + const channel = Reflect.get(service, 'uiServiceWorkerBroadcastChannel') as { + onmessage: (message: { data: BroadcastChannelResponse }) => void + } + channel.onmessage({ data: [uuid, responsePayload] }) +} + await describe('AbstractUIService', async () => { afterEach(() => { standardCleanup() @@ -166,7 +208,7 @@ await describe('AbstractUIService', async () => { assert.notStrictEqual(service, undefined) if (service != null) { - assert.strictEqual(service.getBroadcastChannelExpectedResponses(TEST_UUID), 0) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) service.stop() } }) @@ -182,7 +224,7 @@ await describe('AbstractUIService', async () => { assert.notStrictEqual(service, undefined) if (service != null) { service.stop() - assert.strictEqual(service.getBroadcastChannelExpectedResponses(TEST_UUID), 0) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) } }) @@ -363,7 +405,7 @@ await describe('AbstractUIService', async () => { if (response != null) { assert.strictEqual(response[1].status, ResponseStatus.FAILURE) } - assert.strictEqual(service.getBroadcastChannelExpectedResponses(TEST_UUID), 0) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) } finally { service.stop() } @@ -406,6 +448,504 @@ await describe('AbstractUIService', async () => { } }) + await it('should complete a broadcast request with a failure when responses time out', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { service } = createServiceContext() + + await withMockTimers(t, ['setTimeout'], async () => { + try { + // Broadcast to the single known station; no worker ever replies. + await registerTransportStopRequest(service) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + // The request is released instead of hanging, and a failure is emitted. + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + expectSingleLog( + mocks, + 'warn', + /Failed broadcast response completed without response handler/, + { + hashIdsSucceeded: [], + origin: UIRequestOrigin.TRANSPORT, + procedureName: BroadcastChannelProcedureName.STOP_CHARGING_STATION, + status: ResponseStatus.FAILURE, + uuid: TEST_UUID, + } + ) + } finally { + service.stop() + } + }) + }) + + await it('should report partial successes when a broadcast request times out', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { server, service } = createServiceContext() + server.setChargingStationData(TEST_HASH_ID_2, createMockChargingStationData(TEST_HASH_ID_2)) + + await withMockTimers(t, ['setTimeout'], async () => { + try { + // Broadcast to two stations; only the first ever replies. + await registerTransportStopRequest(service) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 2) + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + // The timeout payload reports the station that did reply successfully. + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + expectSingleLog( + mocks, + 'warn', + /Failed broadcast response completed without response handler/, + { + hashIdsSucceeded: [TEST_HASH_ID], + origin: UIRequestOrigin.TRANSPORT, + procedureName: BroadcastChannelProcedureName.STOP_CHARGING_STATION, + status: ResponseStatus.FAILURE, + uuid: TEST_UUID, + } + ) + } finally { + service.stop() + } + }) + }) + + await it('should not fire the request timeout once the request is released', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { service } = createServiceContext() + + await withMockTimers(t, ['setTimeout'], async () => { + try { + await registerTransportStopRequest(service) + // Simulate the normal completion path releasing the request. + service.deleteBroadcastChannelRequest(TEST_UUID) + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + // Timer was cleared: no timeout response is emitted after release. + assert.strictEqual(mocks.warn.mock.calls.length, 0) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + } finally { + service.stop() + } + }) + }) + + await it('should clear the safety-net timeout when broadcast dispatch throws', async t => { + const { service } = createServiceContext() + const channel = Reflect.get( + service, + 'uiServiceWorkerBroadcastChannel' + ) as UIServiceWorkerBroadcastChannel + t.mock.method(channel, 'sendRequest', () => { + throw new Error('dispatch failed') + }) + const completeExpiredSpy = t.mock.method(channel, 'completeExpiredRequest') + + await withMockTimers(t, ['setTimeout'], async () => { + try { + const response = await service.requestHandler( + createProtocolRequest(TEST_UUID, ProcedureName.STOP_CHARGING_STATION, {}) + ) + assert.strictEqual(response?.[1].status, ResponseStatus.FAILURE) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + // The rollback cleared the timer, so its callback never runs. + assert.strictEqual(completeExpiredSpy.mock.calls.length, 0) + } finally { + service.stop() + } + }) + }) + + await it('should clear pending safety-net timeouts on service stop', async t => { + const { service } = createServiceContext() + const channel = Reflect.get( + service, + 'uiServiceWorkerBroadcastChannel' + ) as UIServiceWorkerBroadcastChannel + const completeExpiredSpy = t.mock.method(channel, 'completeExpiredRequest') + + await withMockTimers(t, ['setTimeout'], async () => { + await registerTransportStopRequest(service) + service.stop() + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + // Stop cleared the timer, so its callback never runs. + assert.strictEqual(completeExpiredSpy.mock.calls.length, 0) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + }) + }) + + await it('should complete a broadcast request truthfully when an outstanding station is deleted', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { server, service } = createServiceContext() + server.setChargingStationData(TEST_HASH_ID_2, createMockChargingStationData(TEST_HASH_ID_2)) + const channel = Reflect.get( + service, + 'uiServiceWorkerBroadcastChannel' + ) as UIServiceWorkerBroadcastChannel + const completeExpiredSpy = t.mock.method(channel, 'completeExpiredRequest') + + await withMockTimers(t, ['setTimeout'], async () => { + try { + // Broadcast to two stations; only the first replies. + await registerTransportStopRequest(service) + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + + // Deleting the other targeted station reconciles the request: the + // surviving reply is reported, the deleted station is dropped. + server.deleteChargingStationData(TEST_HASH_ID_2) + + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + expectSingleLog(mocks, 'debug', /Broadcast response completed without response handler/, { + hashIdsSucceeded: [TEST_HASH_ID], + origin: UIRequestOrigin.TRANSPORT, + procedureName: BroadcastChannelProcedureName.STOP_CHARGING_STATION, + status: ResponseStatus.SUCCESS, + uuid: TEST_UUID, + }) + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + // Reconciliation cleared the safety-net timer. + assert.strictEqual(completeExpiredSpy.mock.calls.length, 0) + } finally { + service.stop() + } + }) + }) + + await it('should drop a late broadcast response instead of re-completing the request', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { service } = createServiceContext() + + try { + await registerTransportStopRequest(service) + // The single targeted station replies and the request completes. + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + const debugCallsAfterCompletion = mocks.debug.mock.calls.length + assert.strictEqual(debugCallsAfterCompletion, 1) + + // A late duplicate reply for the released request must be dropped. + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }) + + assert.strictEqual(mocks.debug.mock.calls.length, debugCallsAfterCompletion + 1) + const [lastMessage] = mocks.debug.mock.calls.at(-1)?.arguments ?? [] + if (typeof lastMessage !== 'string') { + assert.fail('Expected debug log message to be a string') + } + assert.match(lastMessage, /Dropping untracked broadcast response/) + assert.strictEqual(mocks.warn.mock.calls.length, 0) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + } finally { + service.stop() + } + }) + + await it('should not re-complete an already-released request when its timeout fires', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { service } = createServiceContext() + const channel = Reflect.get( + service, + 'uiServiceWorkerBroadcastChannel' + ) as UIServiceWorkerBroadcastChannel + + try { + await registerTransportStopRequest(service) + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }) + const debugCallsAfterCompletion = mocks.debug.mock.calls.length + + // A timeout callback racing after normal completion must be a no-op. + channel.completeExpiredRequest(TEST_UUID) + + assert.strictEqual(mocks.warn.mock.calls.length, 0) + assert.strictEqual(mocks.debug.mock.calls.length, debugCallsAfterCompletion) + } finally { + service.stop() + } + }) + + await it('should reject a broadcast request that targets only unknown stations', async () => { + const { service } = createServiceContext() + + try { + const response = await service.requestHandler( + createProtocolRequest(TEST_UUID, ProcedureName.STOP_CHARGING_STATION, { + hashIds: ['unknown-station'], + }) + ) + + assert.notStrictEqual(response, undefined) + if (response != null) { + assert.strictEqual(response[1].status, ResponseStatus.FAILURE) + assert.match(response[1].errorMessage as string, /does not contain any valid/) + } + // The request was never tracked: nothing to complete or time out. + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + } finally { + service.stop() + } + }) + + await it('should reject a broadcast request with an empty hashIds array instead of broadcasting', async () => { + const { server, service } = createServiceContext() + server.setChargingStationData(TEST_HASH_ID_2, createMockChargingStationData(TEST_HASH_ID_2)) + + try { + const response = await service.requestHandler( + createProtocolRequest(TEST_UUID, ProcedureName.STOP_CHARGING_STATION, { hashIds: [] }) + ) + + assert.notStrictEqual(response, undefined) + if (response != null) { + assert.strictEqual(response[1].status, ResponseStatus.FAILURE) + assert.match(response[1].errorMessage as string, /does not contain any valid/) + } + // An empty explicit target does not degrade to broadcasting to every station. + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + } finally { + service.stop() + } + }) + + await it('should complete a self-target DELETE via its worker reply when deleted first', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { server, service } = createServiceContext() + + try { + await registerTransportDeleteRequest(service, [TEST_HASH_ID]) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + + // The `deleted` worker event arrives before the DELETE reply: reconcile + // must not complete the request, leaving it to the station's own reply. + server.deleteChargingStationData(TEST_HASH_ID) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + assert.strictEqual(mocks.debug.mock.calls.length, 0) + assert.strictEqual(mocks.warn.mock.calls.length, 0) + + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }) + + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + expectSingleLog(mocks, 'debug', /Broadcast response completed without response handler/, { + hashIdsSucceeded: [TEST_HASH_ID], + origin: UIRequestOrigin.TRANSPORT, + procedureName: BroadcastChannelProcedureName.DELETE_CHARGING_STATIONS, + status: ResponseStatus.SUCCESS, + uuid: TEST_UUID, + }) + } finally { + service.stop() + } + }) + + await it('should fail a self-target DELETE via the timeout when the reply never arrives', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { server, service } = createServiceContext() + + await withMockTimers(t, ['setTimeout'], async () => { + try { + await registerTransportDeleteRequest(service, [TEST_HASH_ID]) + server.deleteChargingStationData(TEST_HASH_ID) + // Reconcile skipped the DELETE; the safety-net timeout is the backstop. + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + expectSingleLog( + mocks, + 'warn', + /Failed broadcast response completed without response handler/, + { + hashIdsSucceeded: [], + origin: UIRequestOrigin.TRANSPORT, + procedureName: BroadcastChannelProcedureName.DELETE_CHARGING_STATIONS, + status: ResponseStatus.FAILURE, + uuid: TEST_UUID, + } + ) + } finally { + service.stop() + } + }) + }) + + await it('should keep a broadcast request in-flight when reconcile does not empty the set', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { server, service } = createServiceContext() + server.setChargingStationData(TEST_HASH_ID_2, createMockChargingStationData(TEST_HASH_ID_2)) + + try { + // Broadcast STOP to both stations, then delete one before any reply. + await registerTransportStopRequest(service) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 2) + + server.deleteChargingStationData(TEST_HASH_ID) + // The deleted station is dropped, but the request stays in-flight. + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + assert.strictEqual(mocks.debug.mock.calls.length, 0) + assert.strictEqual(mocks.warn.mock.calls.length, 0) + + emitWorkerResponse(service, { hashId: TEST_HASH_ID_2, status: ResponseStatus.SUCCESS }) + + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + expectSingleLog(mocks, 'debug', /Broadcast response completed without response handler/, { + hashIdsSucceeded: [TEST_HASH_ID_2], + origin: UIRequestOrigin.TRANSPORT, + procedureName: BroadcastChannelProcedureName.STOP_CHARGING_STATION, + status: ResponseStatus.SUCCESS, + uuid: TEST_UUID, + }) + } finally { + service.stop() + } + }) + + await it('should reconcile two concurrent in-flight requests on a single station deletion', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { server, service } = createServiceContext() + server.setChargingStationData(TEST_HASH_ID_2, createMockChargingStationData(TEST_HASH_ID_2)) + + try { + await registerTransportStopRequestFor(service, TEST_UUID, [TEST_HASH_ID, TEST_HASH_ID_2]) + await registerTransportStopRequestFor(service, TEST_UUID_2, [TEST_HASH_ID, TEST_HASH_ID_2]) + // Each request receives the second station's reply, leaving TEST_HASH_ID outstanding on both. + emitWorkerResponse( + service, + { hashId: TEST_HASH_ID_2, status: ResponseStatus.SUCCESS }, + TEST_UUID + ) + emitWorkerResponse( + service, + { hashId: TEST_HASH_ID_2, status: ResponseStatus.SUCCESS }, + TEST_UUID_2 + ) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID_2), 1) + + // Deleting the shared station reconciles BOTH requests in one iteration + // (the Map is mutated as each completes). + server.deleteChargingStationData(TEST_HASH_ID) + + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID_2), 0) + assert.strictEqual(mocks.debug.mock.calls.length, 2) + assert.strictEqual(mocks.warn.mock.calls.length, 0) + } finally { + service.stop() + } + }) + + await it('should skip a DELETE while reconciling a coexisting non-DELETE on the same deleted station', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { server, service } = createServiceContext() + server.setChargingStationData(TEST_HASH_ID_2, createMockChargingStationData(TEST_HASH_ID_2)) + + try { + // A DELETE (TEST_UUID) and a STOP (TEST_UUID_2) both target the two stations. + await registerTransportDeleteRequest(service, [TEST_HASH_ID, TEST_HASH_ID_2]) + await registerTransportStopRequestFor(service, TEST_UUID_2, [TEST_HASH_ID, TEST_HASH_ID_2]) + // Each receives the second station's reply, leaving TEST_HASH_ID outstanding on both. + emitWorkerResponse(service, { hashId: TEST_HASH_ID_2, status: ResponseStatus.SUCCESS }) + emitWorkerResponse( + service, + { hashId: TEST_HASH_ID_2, status: ResponseStatus.SUCCESS }, + TEST_UUID_2 + ) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID_2), 1) + + // One reconcile pass on the shared station: the DELETE is skipped, the STOP completes. + server.deleteChargingStationData(TEST_HASH_ID) + + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID_2), 0) + expectSingleLog(mocks, 'debug', /Broadcast response completed without response handler/, { + hashIdsSucceeded: [TEST_HASH_ID_2], + origin: UIRequestOrigin.TRANSPORT, + procedureName: BroadcastChannelProcedureName.STOP_CHARGING_STATION, + status: ResponseStatus.SUCCESS, + uuid: TEST_UUID_2, + }) + assert.strictEqual(mocks.warn.mock.calls.length, 0) + + // The DELETE still completes truthfully via its own reply. + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }) + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0) + } finally { + service.stop() + } + }) + + await it('should drop a hashId-less reply for a tracked request and log it distinctly', async t => { + const mocks = { + debug: t.mock.method(logger, 'debug', () => undefined), + warn: t.mock.method(logger, 'warn', () => undefined), + } + const { service } = createServiceContext() + + try { + await registerTransportStopRequest(service) + // A reply without a hashId cannot be matched to an outstanding responder. + emitWorkerResponse(service, { hashId: undefined, status: ResponseStatus.SUCCESS }) + + // The request stays in-flight (completion deferred to the timeout), and + // the drop is logged distinctly from an untracked/late drop. + assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 1) + expectSingleLog( + mocks, + 'debug', + /Dropping broadcast response without hashId for a tracked request/ + ) + } finally { + service.stop() + } + }) + await it('should prevent duplicate service registrations', () => { const config = createMockUIServerConfiguration() const server = new TestableUIWebSocketServer(config) -- 2.53.0