fix(ui-server): defer HTTP broadcast responses until worker replies aggregate (#2028...

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.

diff --git a/src/charging-station/ui-server/UIHttpServer.ts b/src/charging-station/ui-server/UIHttpServer.ts
index ff6dda0d5da5d206b264124cdf8b2d13a6dbbece..ca119b5f52c856393c983afbd069cdf6bf1a04e9 100644 (file)
--- 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 b528d38bf8afe01a8cfcd3e4b4bb8bad1025ab47..f67a0c9e13c9fd6472a301b69c30e463ce7171b4 100644 (file)
--- 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<string, string> = {}
+): 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<void> => {
+  for (let i = 0; i < 200; i++) {
+    if (service.getBroadcastChannelOutstandingResponseCount(uuid) === expected) {
+      return
+    }
+    await new Promise<void>(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)
+    })
   })
 })