IPC blocking fix (backport #5400) [release/3.7.x] (#5404)

Co-authored-by: swbsi <[email protected]>

common/api/core-backend.api.md

@@ -123,6 +123,7 @@ import { InternetConnectivityStatus } from '@itwin/core-common';
 import { IpcAppNotifications } from '@itwin/core-common';
 import { IpcListener } from '@itwin/core-common';
 import { IpcSocketBackend } from '@itwin/core-common';
+import { IpcWebSocketBackend } from '@itwin/core-common';
 import { JSONSchema } from '@itwin/core-bentley';
 import { JSONSchemaType } from '@itwin/core-bentley';
 import { JSONSchemaTypeName } from '@itwin/core-bentley';
@@ -3636,6 +3637,8 @@ export class LocalhostIpcHost {
     // (undocumented)
     static connect(connection: ws): void;
     // (undocumented)
+    static socket: IpcWebSocketBackend;
+    // (undocumented)
     static startup(opts?: {
         localhostIpcHost?: LocalhostIpcHostOpts;
         iModelHost?: IModelHostOptions;

common/changes/@itwin/core-backend/ipc-fix_2023-04-14-19-56.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@itwin/core-backend",
+      "comment": "Prevent IPC requests from blocking the backend.",
+      "type": "none"
+    }
+  ],
+  "packageName": "@itwin/core-backend"
+}

common/changes/@itwin/core-common/ipc-fix_2023-04-14-19-56.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@itwin/core-common",
+      "comment": "Prevent IPC requests from blocking the backend.",
+      "type": "none"
+    }
+  ],
+  "packageName": "@itwin/core-common"
+}

core/backend/src/LocalhostIpcHost.ts

@@ -73,7 +73,7 @@ class RpcHandler extends IpcHandler {
 /** @internal */
 export class LocalhostIpcHost {
   private static _initialized = false;
-  private static _socket: IpcWebSocketBackend;
+  public static socket: IpcWebSocketBackend;
 
   public static connect(connection: ws) {
     (IpcWebSocket.transport as LocalTransport).connect(connection);
@@ -85,11 +85,11 @@ export class LocalhostIpcHost {
     if (!this._initialized) {
       registerHandler = true;
       IpcWebSocket.transport = new LocalTransport(opts?.localhostIpcHost ?? {});
-      this._socket = new IpcWebSocketBackend();
+      this.socket = new IpcWebSocketBackend();
       this._initialized = true;
     }
 
-    await IpcHost.startup({ ipcHost: { socket: this._socket }, iModelHost: opts?.iModelHost });
+    await IpcHost.startup({ ipcHost: { socket: this.socket }, iModelHost: opts?.iModelHost });
 
     if (registerHandler) {
       RpcHandler.register();

core/common/src/ipc/IpcWebSocket.ts

@@ -132,7 +132,6 @@ export class IpcWebSocketFrontend extends IpcWebSocket implements IpcSocketFront
 export class IpcWebSocketBackend extends IpcWebSocket implements IpcSocketBackend {
   private _handlers = new Map<string, (event: Event, methodName: string, ...args: any[]) => Promise<any>>();
   private _processingQueue: IpcWebSocketMessage[] = [];
-  private _processing: IpcWebSocketMessage | undefined;
 
   public constructor() {
     super();
@@ -161,16 +160,14 @@ export class IpcWebSocketBackend extends IpcWebSocket implements IpcSocketBacken
   }
 
   private async processMessages() {
-    if (this._processing || !this._processingQueue.length) {
+    if (!this._processingQueue.length) {
       return;
     }
 
     const message = this._processingQueue.shift();
     if (message && message.method) {
       const handler = this._handlers.get(message.channel);
       if (handler) {
-        this._processing = message;
-
         let args = message.data;
         if (typeof (args) === "undefined")
           args = [];
@@ -184,8 +181,6 @@ export class IpcWebSocketBackend extends IpcWebSocket implements IpcSocketBacken
           data: response,
           sequence: -1,
         });
-
-        this._processing = undefined;
       }
     }
 

full-stack-tests/rpc/src/backend/electron.ts

@@ -6,6 +6,7 @@ import { registerBackendCallback } from "@itwin/certa/lib/utils/CallbackUtils";
 import { BackendTestCallbacks } from "../common/SideChannels";
 import { commonSetup } from "./CommonBackendSetup";
 import { ElectronHost } from "@itwin/core-electron/lib/cjs/ElectronBackend";
+import { setupIpcTestElectron } from "./ipc";
 
 async function init() {
   await commonSetup();
@@ -14,6 +15,7 @@ async function init() {
     ElectronHost.rpcConfig.protocol.transferChunkThreshold = value;
     return true;
   });
+  setupIpcTestElectron();
 }
 
 module.exports = init();

full-stack-tests/rpc/src/backend/ipc.ts

@@ -6,15 +6,35 @@ import { registerBackendCallback } from "@itwin/certa/lib/utils/CallbackUtils";
 import { IpcWebSocketBackend, iTwinChannel } from "@itwin/core-common";
 import { BackendTestCallbacks } from "../common/SideChannels";
 
-export async function setupIpcTest(before = async () => { }) {
+function orderTest(socket: { handle(channel: string, listener: (event: any, ...args: any[]) => Promise<any>): void }) {
+  socket.handle("a", async (_event: Event, methodName: string, ..._args: any[]) => {
+    return [methodName, "a"];
+  });
+
+  socket.handle("b", async (_event: Event, methodName: string, ..._args: any[]) => {
+    return new Promise((resolve) => {
+      setTimeout(() => resolve([methodName, "b"]), 1000);
+    });
+  });
+
+  socket.handle("c", async (_event: Event, methodName: string, ..._args: any[]) => {
+    return [methodName, "c"];
+  });
+}
+
+export function setupIpcTestElectron() {
+  orderTest(require("electron").ipcMain); // eslint-disable-line @typescript-eslint/no-var-requires
+}
+
+export async function setupIpcTest(before = async () => { }, socketOverride?: IpcWebSocketBackend) {
   let socket: IpcWebSocketBackend;
   let ready: () => void;
   const started = new Promise<void>((resolve) => ready = resolve);
 
   registerBackendCallback(BackendTestCallbacks.startIpcTest, () => {
     setTimeout(async () => {
       await before();
-      socket = new IpcWebSocketBackend();
+      socket = socketOverride || (new IpcWebSocketBackend());
 
       socket.addListener("test", (_evt: Event, ...arg: any[]) => {
         if (arg[0] !== 1 || arg[1] !== 2 || arg[2] !== 3) {
@@ -26,6 +46,8 @@ export async function setupIpcTest(before = async () => { }) {
         return [methodName, ...args];
       });
 
+      orderTest(socket);
+
       socket.handle(iTwinChannel("ipc-app"), async (_event: Event, _methodName: string, ..._args: any[]) => {
         return { result: undefined };
       });

full-stack-tests/rpc/src/backend/websocket.ts

@@ -9,6 +9,7 @@ import { WebEditServer } from "@itwin/express-server";
 import { BackendTestCallbacks } from "../common/SideChannels";
 import { AttachedInterface, rpcInterfaces } from "../common/TestRpcInterface";
 import { commonSetup } from "./CommonBackendSetup";
+import { setupIpcTest } from "./ipc";
 import { AttachedInterfaceImpl } from "./TestRpcImpl";
 
 async function init() {
@@ -29,6 +30,7 @@ async function init() {
   console.log(`Web backend for rpc full-stack-tests listening on port ${port}`);
 
   initializeAttachedInterfacesTest(rpcConfig);
+  setupIpcTest(async () => Promise.resolve(), LocalhostIpcHost.socket); // eslint-disable-line @typescript-eslint/no-floating-promises
 
   return () => {
     httpServer.close();

full-stack-tests/rpc/src/frontend/IpcWebSocket.test.ts → full-stack-tests/rpc/src/frontend/Ipc.test.ts

@@ -7,26 +7,45 @@ import { IpcWebSocketFrontend } from "@itwin/core-common";
 import { executeBackendCallback } from "@itwin/certa/lib/utils/CallbackUtils";
 import { assert } from "chai";
 import { BackendTestCallbacks } from "../common/SideChannels";
-import { currentEnvironment } from "./_Setup.test";
 
-if (!ProcessDetector.isElectronAppFrontend) {
+function orderTest(it: Mocha.TestFunction, socketSource: () => { invoke(channel: string, ...args: any[]): Promise<any> }) {
+  async function onResponse(request: Promise<any>, responses: string[]) {
+    const data = await request;
+    responses.push(data[0]);
+  }
+
+  it("should preserve order", async () => {
+    const socket = socketSource();
+
+    const responses: string[] = [];
+
+    const a = socket.invoke("a", "a");
+    const b = socket.invoke("b", "b");
+    const c = socket.invoke("c", "c");
+
+    onResponse(a, responses); // eslint-disable-line @typescript-eslint/no-floating-promises
+    onResponse(b, responses); // eslint-disable-line @typescript-eslint/no-floating-promises
+    onResponse(c, responses); // eslint-disable-line @typescript-eslint/no-floating-promises
+
+    await Promise.all([a, b, c]);
+    assert.deepEqual(responses, ["a", "c", "b"]);
+  });
+}
+
+if (ProcessDetector.isElectronAppFrontend) {
+  describe("ElectronIpc", () => {
+    orderTest(it, () => require("electron").ipcRenderer); // eslint-disable-line @typescript-eslint/no-var-requires
+  });
+} else {
   describe("IpcWebSocket", () => {
     let socket: IpcWebSocketFrontend;
 
     before(async () => {
-      if (currentEnvironment === "websocket") {
-        return;
-      }
-
       assert(await executeBackendCallback(BackendTestCallbacks.startIpcTest));
       socket = new IpcWebSocketFrontend();
     });
 
     it("should support send/receive", async () => {
-      if (currentEnvironment === "websocket") {
-        return;
-      }
-
       return new Promise(async (resolve) => {
         socket.addListener("test", (_evt: Event, ...arg: any[]) => {
           assert.equal(arg[0], 4);
@@ -42,10 +61,6 @@ if (!ProcessDetector.isElectronAppFrontend) {
     });
 
     it("should support invoke", async () => {
-      if (currentEnvironment === "websocket") {
-        return;
-      }
-
       return new Promise(async (resolve) => {
         const invoked = await socket.invoke("testinvoke", "hi", 1, 2, 3);
         assert.equal(invoked[0], "hi");
@@ -55,5 +70,7 @@ if (!ProcessDetector.isElectronAppFrontend) {
         resolve();
       });
     });
+
+    orderTest(it, () => socket);
   });
 }