From d8fb33b23ffb5bd59f5e4e1e33e4bdb678303d57 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 19 Jul 2026 21:27:45 +0200 Subject: [PATCH] fix(ui-server): defer HTTP broadcast responses until worker replies aggregate (#2028) (#2037) The deprecated HTTP UI transport emitted a synthetic `success` the instant the request handler resolved `undefined` (the broadcast-pending signal), without awaiting worker replies. PR #2020's Set-based aggregation therefore covered the WebSocket and MCP transports but not HTTP: a fan-out command reported `200`/`success` before any worker acted, and the aggregated `hashIdsFailed`/`responsesFailed` never reached the client. Align HTTP with the WebSocket and MCP transports (issue #2028 option 1): only synchronous, non-broadcast procedures respond inline. A broadcast keeps its already-registered response handler open so the later aggregated `sendResponse` writes the real status and `hashIdsSucceeded`/`hashIdsFailed`, reusing the existing 60s safety-net timeout (`UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS`) so a never-arriving aggregation yields a bounded failure with no hung socket. Client-disconnect cleanup and server-side uuid minting are unchanged; no new timeout or aggregation path is introduced. This is a UI SRPC transport-layer fix only; no OCPP PDU/message format changes, and no change to WebSocket/MCP behavior or the shared aggregation semantics. --- .../ui-server/UIHttpServer.ts | 34 +- .../ui-server/UIHttpServer.test.ts | 339 +++++++++++++++++- 2 files changed, 364 insertions(+), 9 deletions(-) diff --git a/src/charging-station/ui-server/UIHttpServer.ts b/src/charging-station/ui-server/UIHttpServer.ts index ff6dda0d..ca119b5f 100644 --- a/src/charging-station/ui-server/UIHttpServer.ts +++ b/src/charging-station/ui-server/UIHttpServer.ts @@ -1,7 +1,7 @@ import type { IncomingMessage, ServerResponse } from 'node:http' import { getReasonPhrase, StatusCodes } from 'http-status-codes' -import { createGzip } from 'node:zlib' +import { createGzip, type Gzip } from 'node:zlib' import type { IBootstrap } from '../IBootstrap.js' @@ -32,6 +32,10 @@ import { HttpMethod, isProtocolAndVersionSupported } from './UIServerUtils.js' const moduleName = 'UIHttpServer' /** + * Broadcast/fan-out commands defer their HTTP response until the aggregated + * per-station outcome arrives, aligning with the WebSocket and MCP + * transports, instead of synthesizing an immediate success; only + * synchronous, non-broadcast procedures respond inline. * @deprecated Use UIMCPServer (ApplicationProtocol.MCP) instead. Will be removed in a future major version. */ export class UIHttpServer extends AbstractUIServer { @@ -71,7 +75,18 @@ export class UIHttpServer extends AbstractUIServer { 'Content-Type': 'application/json', Vary: 'Accept-Encoding', }) - const gzip = createGzip() + const gzip = this.createGzipStream() + gzip.on('error', (error: Error) => { + logger.error( + `${this.logPrefix(moduleName, 'sendResponse')} Error at compressing response id '${uuid}':`, + error + ) + // pipe() does not forward the gzip (source) 'error' to res, so this + // handler is required: an unhandled compression error would otherwise + // crash the process. Destroy without an error argument because the + // headers are already sent, so nothing can be signaled to the client. + res.destroy() + }) gzip.pipe(res) gzip.end(body) } else { @@ -105,6 +120,15 @@ export class UIHttpServer extends AbstractUIServer { this.httpServer.on('request', this.requestListener.bind(this)) } + /** + * Overridable gzip transform factory; a test seam for exercising the + * compression error branch. + * @returns The gzip transform used to compress the response body. + */ + protected createGzipStream (): Gzip { + return createGzip() + } + private async handleRequestBody ( req: IncomingMessage, res: ServerResponse, @@ -134,10 +158,12 @@ export class UIHttpServer extends AbstractUIServer { const protocolResponse = await service.requestHandler( this.buildProtocolRequest(uuid, procedureName, requestPayload) ) + // A non-null response is a synchronous, non-broadcast procedure: respond + // now. A null response is a deferred broadcast: keep the request open so + // the later aggregated sendResponse writes the real per-station outcome, + // bounded by the safety-net timeout, instead of synthesizing success. if (protocolResponse != null) { this.sendResponse(protocolResponse) - } else { - this.sendResponse(this.buildProtocolResponse(uuid, { status: ResponseStatus.SUCCESS })) } } diff --git a/tests/charging-station/ui-server/UIHttpServer.test.ts b/tests/charging-station/ui-server/UIHttpServer.test.ts index b528d38b..f67a0c9e 100644 --- a/tests/charging-station/ui-server/UIHttpServer.test.ts +++ b/tests/charging-station/ui-server/UIHttpServer.test.ts @@ -6,26 +6,38 @@ import type { IncomingMessage } from 'node:http' import assert from 'node:assert/strict' +import { Readable } from 'node:stream' import { afterEach, beforeEach, describe, it } from 'node:test' -import { gunzipSync } from 'node:zlib' +import { gunzipSync, type Gzip } from 'node:zlib' -import type { UIServerConfiguration, UUIDv4 } from '../../../src/types/index.js' +import type { AbstractUIService } from '../../../src/charging-station/ui-server/ui-services/AbstractUIService.js' +import type { ResponsePayload, UIServerConfiguration, UUIDv4 } from '../../../src/types/index.js' import { UIHttpServer } from '../../../src/charging-station/ui-server/UIHttpServer.js' import { DEFAULT_COMPRESSION_THRESHOLD_BYTES } from '../../../src/charging-station/ui-server/UIServerSecurity.js' -import { ApplicationProtocol, ResponseStatus } from '../../../src/types/index.js' -import { standardCleanup } from '../../helpers/TestLifecycleHelpers.js' -import { TEST_UUID } from './UIServerTestConstants.js' +import { ApplicationProtocol, ProtocolVersion, ResponseStatus } from '../../../src/types/index.js' +import { Constants, logger } from '../../../src/utils/index.js' +import { + createLoggerMocks, + flushMicrotasks, + standardCleanup, + withMockTimers, +} from '../../helpers/TestLifecycleHelpers.js' +import { TEST_HASH_ID, TEST_HASH_ID_2, TEST_UUID } from './UIServerTestConstants.js' import { awaitFinish, createMockBootstrap, + createMockChargingStationData, createMockIncomingMessage, createMockUIServerConfiguration, + emitWorkerResponse, MockServerResponse, } from './UIServerTestUtils.js' // eslint-disable-next-line @typescript-eslint/no-deprecated class TestableUIHttpServer extends UIHttpServer { + public lastGzipStream?: Gzip + public constructor (config: UIServerConfiguration) { // eslint-disable-next-line @typescript-eslint/no-deprecated super(config, createMockBootstrap()) @@ -35,6 +47,10 @@ class TestableUIHttpServer extends UIHttpServer { this.responseHandlers.set(uuid, res as never) } + public addStation (hashId: string): void { + this.setChargingStationData(hashId, createMockChargingStationData(hashId)) + } + public emitRequest (req: IncomingMessage, res: MockServerResponse): void { const httpServer = Reflect.get(this, 'httpServer') as { emit: (eventName: string, req: IncomingMessage, res: MockServerResponse) => boolean @@ -50,9 +66,30 @@ class TestableUIHttpServer extends UIHttpServer { return this.responseHandlers.size } + public getResponseHandlerUuids (): UUIDv4[] { + return [...this.responseHandlers.keys()] + } + + public getUIService (version: ProtocolVersion): AbstractUIService | undefined { + return this.uiServices.get(version) + } + + public mockListen (): void { + const httpServer = Reflect.get(this, 'httpServer') as { + listen: (...args: unknown[]) => unknown + } + httpServer.listen = () => httpServer + } + public setAcceptsGzip (uuid: UUIDv4, value: boolean): void { this.getAcceptsGzip().set(uuid, value) } + + protected override createGzipStream (): Gzip { + // eslint-disable-next-line @typescript-eslint/no-deprecated + this.lastGzipStream = super.createGzipStream() + return this.lastGzipStream + } } const createHttpServerConfig = () => @@ -63,6 +100,59 @@ const createLargePayload = (status: ResponseStatus = ResponseStatus.SUCCESS) => status, }) +const buildProcedureRequest = ( + procedureName: string, + payload: object, + headers: Record = {} +): IncomingMessage => { + const req = Readable.from([Buffer.from(JSON.stringify(payload))]) as unknown as IncomingMessage + Object.assign(req, { + complete: true, + headers: { host: 'localhost', ...headers }, + headersDistinct: {}, + method: 'POST', + rawHeaders: [], + socket: { encrypted: false, remoteAddress: '127.0.0.1' }, + url: `/ui/${ProtocolVersion['0.0.1']}/${procedureName}`, + }) + return req +} + +const createBroadcastServer = (): TestableUIHttpServer => { + const server = new TestableUIHttpServer( + createMockUIServerConfiguration({ + options: { host: '127.0.0.1', port: 0 }, + type: ApplicationProtocol.HTTP, + }) + ) + server.mockListen() + server.start() + return server +} + +// The HTTP handler resolves undefined for a broadcast and dispatches the +// fan-out asynchronously; poll until the safety-net tracking is armed. +const waitForOutstanding = async ( + service: AbstractUIService, + uuid: UUIDv4, + expected: number +): Promise => { + for (let i = 0; i < 200; i++) { + if (service.getBroadcastChannelOutstandingResponseCount(uuid) === expected) { + return + } + await new Promise(resolve => { + setImmediate(resolve) + }) + } + throw new Error( + `waitForOutstanding: expected ${expected.toString()} outstanding response(s) for ${uuid}` + ) +} + +const parseHttpResponsePayload = (res: MockServerResponse): ResponsePayload => + JSON.parse(res.body ?? '{}') as ResponsePayload + await describe('UIHttpServer', async () => { let server: TestableUIHttpServer @@ -287,6 +377,223 @@ await describe('UIHttpServer', async () => { assert.strictEqual(rateLimiterCalls[0], '203.0.113.10') }) + await describe('deferred broadcast aggregation (issue #2028)', async () => { + await it('should not report success for a fan-out with a failing target, returning the aggregated payload', async () => { + const broadcastServer = createBroadcastServer() + try { + broadcastServer.addStation(TEST_HASH_ID) + broadcastServer.addStation(TEST_HASH_ID_2) + const res = new MockServerResponse() + + broadcastServer.emitRequest( + buildProcedureRequest('stopChargingStation', { + hashIds: [TEST_HASH_ID, TEST_HASH_ID_2], + }), + res + ) + + assert.strictEqual(broadcastServer.getResponseHandlerUuids().length, 1) + const uuid = broadcastServer.getResponseHandlerUuids()[0] + const service = broadcastServer.getUIService(ProtocolVersion['0.0.1']) + if (service == null) { + assert.fail('Expected UI service to be registered') + } + await waitForOutstanding(service, uuid, 2) + assert.strictEqual(res.ended, false) + + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }, uuid) + emitWorkerResponse( + service, + { hashId: TEST_HASH_ID_2, status: ResponseStatus.FAILURE }, + uuid + ) + + assert.strictEqual(res.ended, true) + assert.strictEqual(res.statusCode, 400) + const payload = parseHttpResponsePayload(res) + assert.strictEqual(payload.status, ResponseStatus.FAILURE) + assert.deepStrictEqual(payload.hashIdsSucceeded, [TEST_HASH_ID]) + assert.deepStrictEqual(payload.hashIdsFailed, [TEST_HASH_ID_2]) + assert.strictEqual(broadcastServer.getResponseHandlersSize(), 0) + } finally { + broadcastServer.stop() + } + }) + + await it('should defer a fan-out until every target succeeds, returning the aggregated succeeded hashIds', async () => { + const broadcastServer = createBroadcastServer() + try { + broadcastServer.addStation(TEST_HASH_ID) + broadcastServer.addStation(TEST_HASH_ID_2) + const res = new MockServerResponse() + + broadcastServer.emitRequest( + buildProcedureRequest('stopChargingStation', { + hashIds: [TEST_HASH_ID, TEST_HASH_ID_2], + }), + res + ) + + assert.strictEqual(broadcastServer.getResponseHandlerUuids().length, 1) + const uuid = broadcastServer.getResponseHandlerUuids()[0] + const service = broadcastServer.getUIService(ProtocolVersion['0.0.1']) + if (service == null) { + assert.fail('Expected UI service to be registered') + } + await waitForOutstanding(service, uuid, 2) + assert.strictEqual(res.ended, false) + + emitWorkerResponse(service, { hashId: TEST_HASH_ID, status: ResponseStatus.SUCCESS }, uuid) + assert.strictEqual(res.ended, false) + emitWorkerResponse( + service, + { hashId: TEST_HASH_ID_2, status: ResponseStatus.SUCCESS }, + uuid + ) + + assert.strictEqual(res.ended, true) + assert.strictEqual(res.statusCode, 200) + const payload = parseHttpResponsePayload(res) + assert.strictEqual(payload.status, ResponseStatus.SUCCESS) + assert.deepStrictEqual(payload.hashIdsSucceeded, [TEST_HASH_ID, TEST_HASH_ID_2]) + assert.strictEqual(payload.hashIdsFailed, undefined) + assert.strictEqual(broadcastServer.getResponseHandlersSize(), 0) + } finally { + broadcastServer.stop() + } + }) + + await it('should gzip a deferred aggregated failure payload above the compression threshold', async () => { + const broadcastServer = createBroadcastServer() + try { + broadcastServer.addStation(TEST_HASH_ID) + const res = new MockServerResponse() + + broadcastServer.emitRequest( + buildProcedureRequest( + 'stopChargingStation', + { hashIds: [TEST_HASH_ID] }, + { 'accept-encoding': 'gzip' } + ), + res + ) + + const uuid = broadcastServer.getResponseHandlerUuids()[0] + const service = broadcastServer.getUIService(ProtocolVersion['0.0.1']) + if (service == null) { + assert.fail('Expected UI service to be registered') + } + await waitForOutstanding(service, uuid, 1) + assert.strictEqual(res.ended, false) + assert.strictEqual(broadcastServer.getAcceptsGzip().get(uuid), true) + + const errorMessage = 'e'.repeat(DEFAULT_COMPRESSION_THRESHOLD_BYTES + 100) + emitWorkerResponse( + service, + { errorMessage, hashId: TEST_HASH_ID, status: ResponseStatus.FAILURE }, + uuid + ) + + await awaitFinish(res) + + assert.strictEqual(res.statusCode, 400) + assert.strictEqual(res.headers['Content-Encoding'], 'gzip') + assert.strictEqual(res.headers['Content-Type'], 'application/json') + assert.strictEqual(res.headers.Vary, 'Accept-Encoding') + if (res.bodyBuffer == null) { + throw new Error('Expected bodyBuffer to be defined') + } + const decompressed = gunzipSync(res.bodyBuffer).toString('utf8') + const payload = JSON.parse(decompressed) as ResponsePayload + assert.strictEqual(payload.status, ResponseStatus.FAILURE) + assert.deepStrictEqual(payload.hashIdsFailed, [TEST_HASH_ID]) + assert.strictEqual(payload.responsesFailed?.[0]?.errorMessage, errorMessage) + assert.strictEqual(broadcastServer.getResponseHandlersSize(), 0) + } finally { + broadcastServer.stop() + } + }) + + await it('should return a bounded failure when the aggregation never arrives', async t => { + await withMockTimers(t, ['setTimeout'], async () => { + const broadcastServer = createBroadcastServer() + try { + broadcastServer.addStation(TEST_HASH_ID) + const res = new MockServerResponse() + + broadcastServer.emitRequest( + buildProcedureRequest('stopChargingStation', { hashIds: [TEST_HASH_ID] }), + res + ) + + const uuid = broadcastServer.getResponseHandlerUuids()[0] + const service = broadcastServer.getUIService(ProtocolVersion['0.0.1']) + if (service == null) { + assert.fail('Expected UI service to be registered') + } + await waitForOutstanding(service, uuid, 1) + assert.strictEqual(res.ended, false) + + t.mock.timers.tick(Constants.UI_SERVER_BROADCAST_CHANNEL_REQUEST_TIMEOUT_MS) + + assert.strictEqual(res.ended, true) + assert.strictEqual(res.statusCode, 400) + const payload = parseHttpResponsePayload(res) + assert.strictEqual(payload.status, ResponseStatus.FAILURE) + assert.strictEqual(broadcastServer.getResponseHandlersSize(), 0) + } finally { + broadcastServer.stop() + } + }) + }) + + await it('should respond immediately to a synchronous non-broadcast procedure', async () => { + const broadcastServer = createBroadcastServer() + try { + const res = new MockServerResponse() + + broadcastServer.emitRequest(buildProcedureRequest('listChargingStations', {}), res) + await awaitFinish(res) + + assert.strictEqual(res.statusCode, 200) + const payload = parseHttpResponsePayload(res) + assert.strictEqual(payload.status, ResponseStatus.SUCCESS) + assert.strictEqual(broadcastServer.getResponseHandlersSize(), 0) + } finally { + broadcastServer.stop() + } + }) + + await it('should clean up the deferred handler entry when the client disconnects', async () => { + const broadcastServer = createBroadcastServer() + try { + broadcastServer.addStation(TEST_HASH_ID) + const res = new MockServerResponse() + + broadcastServer.emitRequest( + buildProcedureRequest('stopChargingStation', { hashIds: [TEST_HASH_ID] }), + res + ) + + const uuid = broadcastServer.getResponseHandlerUuids()[0] + const service = broadcastServer.getUIService(ProtocolVersion['0.0.1']) + if (service == null) { + assert.fail('Expected UI service to be registered') + } + await waitForOutstanding(service, uuid, 1) + assert.strictEqual(broadcastServer.getResponseHandlersSize(), 1) + assert.strictEqual(broadcastServer.getAcceptsGzip().has(uuid), true) + + res.emit('close') + + assert.strictEqual(broadcastServer.getResponseHandlersSize(), 0) + assert.strictEqual(broadcastServer.getAcceptsGzip().has(uuid), false) + } finally { + broadcastServer.stop() + } + }) + }) + await describe('Gzip compression', async () => { let gzipServer: TestableUIHttpServer @@ -387,5 +694,27 @@ await describe('UIHttpServer', async () => { assert.strictEqual(gzipServer.getAcceptsGzip().has(TEST_UUID), false) }) + + await it('should destroy the response and log once when gzip compression fails', async t => { + const res = new MockServerResponse() + const { errorMock } = createLoggerMocks(t, logger) + + gzipServer.addResponseHandler(TEST_UUID, res) + gzipServer.setAcceptsGzip(TEST_UUID, true) + gzipServer.sendResponse([TEST_UUID, createLargePayload()]) + + const gzip = gzipServer.lastGzipStream + if (gzip == null) { + throw new Error('Expected the gzip stream to be captured') + } + // A genuine mid-compression failure discards the queued end(body) flush, + // so the body is never written and only the error handler acts on res. + gzip.destroy(new Error('gzip compression failure')) + await flushMicrotasks() + + assert.strictEqual(res.destroyed, true) + assert.strictEqual(res.ended, false) + assert.strictEqual(errorMock.mock.calls.length, 1) + }) }) }) -- 2.53.0