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<string, Responses>
+ private readonly responses: Map<UUIDv4, BroadcastChannelResponsePayload[]>
private readonly uiService: AbstractUIService
constructor (uiService: AbstractUIService) {
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<string, Responses>()
+ this.responses = new Map<UUIDv4, BroadcastChannelResponsePayload[]>()
+ }
+
+ /**
+ * 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)
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
...(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
...(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
}
}
+ 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:`,
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)
}
}
}
} from '../../../types/index.js'
import {
Configuration,
+ Constants,
ensureError,
getErrorMessage,
isNotEmptyArray,
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
}
interface BroadcastChannelRequestContext {
- readonly expectedResponses: number
readonly origin: UIRequestOrigin
+ readonly outstandingHashIds: Set<string>
readonly procedureName: BroadcastChannelProcedureName
+ readonly timeout: ReturnType<typeof setTimeout>
}
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 }
public stop (): void {
this.stopped = true
+ for (const requestContext of this.broadcastChannelRequests.values()) {
+ clearTimeout(requestContext.timeout)
+ }
this.broadcastChannelRequests.clear()
this.uiServiceWorkerBroadcastChannel.close()
}
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<string>(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
}
}
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,
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,
)
}
+const registerTransportStopRequestFor = async (
+ service: AbstractUIService,
+ uuid: UUIDv4,
+ hashIds: string[]
+): Promise<void> => {
+ await service.requestHandler(
+ createProtocolRequest(uuid, ProcedureName.STOP_CHARGING_STATION, { hashIds })
+ )
+}
+
+const registerTransportDeleteRequest = async (
+ service: AbstractUIService,
+ hashIds: string[]
+): Promise<void> => {
+ 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()
assert.notStrictEqual(service, undefined)
if (service != null) {
- assert.strictEqual(service.getBroadcastChannelExpectedResponses(TEST_UUID), 0)
+ assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0)
service.stop()
}
})
assert.notStrictEqual(service, undefined)
if (service != null) {
service.stop()
- assert.strictEqual(service.getBroadcastChannelExpectedResponses(TEST_UUID), 0)
+ assert.strictEqual(service.getBroadcastChannelOutstandingResponseCount(TEST_UUID), 0)
}
})
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()
}
}
})
+ 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)
repositories / e-mobility-charging-stations-simulator.git / commitdiff
