fix: rpc race condition (#702)

* test race condition

* linter

* final touches and linter

* fix build

Author: menduz

package-lock.json

@@ -66,7 +66,7 @@
     "@dcl/kernel-interface": "^2.0.0-20210922153939.commit-017905d",
     "@dcl/legacy-ecs": "^6.11.8",
     "@dcl/protocol": "^1.0.0-3373991894.commit-8aa3a49",
-    "@dcl/rpc": "^1.0.4-20221113192916.commit-3ef5187",
+    "@dcl/rpc": "^1.1.1-20221115000939.commit-9a51ad0",
     "@dcl/scene-runtime": "^1.0.0-20221102005705.commit-05d463c",
     "@dcl/schemas": "^5.21.0",
     "@dcl/urn-resolver": "^2.0.2",

package.json

@@ -4,6 +4,6 @@ globalThis.process = {
   browser: true,
   env: {},
   nextTick(fn, ...args) {
-    require('queue-microtask')(() => fn(...args))
+    queueMicrotask(() => fn(...args))
   }
 }

packages/entryPoints/inject.js

@@ -1,19 +1,8 @@
-import { createRpcClient, Transport } from '@dcl/rpc'
-import future, { IFuture } from 'fp-future'
-import { registerCRDTService } from './services/crdtService'
-import { RendererProtocol } from './types'
+import { createRpcClient, RpcClientPort, Transport } from '@dcl/rpc'
 
-export const rendererProtocol: IFuture<RendererProtocol> = future()
-
-export async function createRendererRpcClient(transport: Transport): Promise<RendererProtocol> {
+export async function createRendererRpcClient(transport: Transport): Promise<RpcClientPort> {
   const rpcClient = await createRpcClient(transport)
   const clientPort = await rpcClient.createPort('renderer-protocol')
 
-  const crdtService = registerCRDTService(clientPort)
-
-  rendererProtocol.resolve({
-    crdtService
-  })
-
-  return rendererProtocol
+  return clientPort
 }

packages/renderer-protocol/rpcClient.ts

@@ -90,6 +90,9 @@ export function webSocketTransportAdapter(url: string, options: CommonRendererOp
 
   const transport: Transport = {
     ...events,
+    get isConnected(): boolean {
+      return (socket && socket.readyState === socket.OPEN) || false
+    },
     sendMessage(message: any) {
       send(message)
     },

packages/renderer-protocol/transports/webSocketTransportAdapter.ts

@@ -5,7 +5,7 @@ export type WebTransportOptions = {
   wasmModule: any
 }
 
-export function webTransport(options: WebTransportOptions): Transport {
+export function webTransport(options: WebTransportOptions, unityDclInstance: any) {
   const events = mitt<TransportEvents>()
   const ALLOC_SIZE = 8388608
   let heapPtr: number
@@ -17,15 +17,25 @@ export function webTransport(options: WebTransportOptions): Transport {
   }
 
   let isClosed = false
+  let didConnect = false
 
-  ;(globalThis as any).DCL.BinaryMessageFromEngine = function (data: Uint8Array) {
+  unityDclInstance.BinaryMessageFromEngine = function (data: Uint8Array) {
+    if (!didConnect) {
+      throw new Error('Received data from unity before connection was established')
+    }
     const copiedData = new Uint8Array(data)
     events.emit('message', copiedData)
   }
 
   const transport: Transport = {
     ...events,
+    get isConnected() {
+      return didConnect
+    },
     sendMessage(message) {
+      if (!didConnect) {
+        throw new Error('Tried to send a message before connection was established')
+      }
       if (!!sendMessageToRenderer && !isClosed) {
         options.wasmModule.HEAPU8.set(message, heapPtr)
         sendMessageToRenderer(heapPtr, message.length)
@@ -39,7 +49,12 @@ export function webTransport(options: WebTransportOptions): Transport {
     }
   }
 
-  queueMicrotask(() => events.emit('connect', {}))
+  events.on('connect', () => {
+    didConnect = true
+  })
+
+  // connect the transport
+  events.emit('connect', {})
 
   return transport
 }

packages/renderer-protocol/transports/webTransport.ts

@@ -11,8 +11,7 @@ import {
 
 import { PortContext } from './context'
 import { EntityAction, EntityActionType } from 'shared/types'
-
-import { rendererProtocol } from './../../../renderer-protocol/rpcClient'
+import { registerCRDTService } from 'renderer-protocol/services/crdtService'
 
 function getPayload(payloadType: EAType, payload: Payload): any {
   switch (payloadType) {
@@ -60,60 +59,62 @@ function getPayload(payloadType: EAType, payload: Payload): any {
 }
 
 export function registerEngineApiServiceServerImplementation(port: RpcServerPort<PortContext>) {
-  codegen.registerService(port, EngineApiServiceDefinition, async () => ({
-    async sendBatch(req: ManyEntityAction, ctx) {
-      const actions: EntityAction[] = []
+  codegen.registerService(port, EngineApiServiceDefinition, async (port, ctx) => {
+    const crdtService = registerCRDTService(ctx.rendererPort)
+
+    return {
+      async sendBatch(req: ManyEntityAction, ctx) {
+        const actions: EntityAction[] = []
 
-      for (const action of req.actions) {
-        const actionType = eaTypeToStr(action.type)
-        if (actionType && action.payload) {
-          actions.push({
-            type: actionType,
-            tag: action.tag,
-            payload: getPayload(action.type, action.payload as any)
-          })
+        for (const action of req.actions) {
+          const actionType = eaTypeToStr(action.type)
+          if (actionType && action.payload) {
+            actions.push({
+              type: actionType,
+              tag: action.tag,
+              payload: getPayload(action.type, action.payload as any)
+            })
+          }
         }
-      }
 
-      if (actions.length) {
-        ctx.sendBatch(actions)
-      }
+        if (actions.length) {
+          ctx.sendBatch(actions)
+        }
 
-      const events: EventData[] = ctx.events
+        const events: EventData[] = ctx.events
 
-      if (events.length) {
-        ctx.events = []
-      }
+        if (events.length) {
+          ctx.events = []
+        }
 
-      return { events }
-    },
+        return { events }
+      },
 
-    async subscribe(req, ctx) {
-      ctx.subscribedEvents.add(req.eventId)
-      return {}
-    },
-    async unsubscribe(req, ctx) {
-      ctx.subscribedEvents.delete(req.eventId)
-      return {}
-    },
-    async crdtSendToRenderer(req, ctx) {
-      const protocol = await rendererProtocol
-      return protocol.crdtService.sendCrdt({
-        sceneId: ctx.sceneData.id,
-        payload: req.data,
-        sceneNumber: ctx.sceneData.sceneNumber
-      })
-    },
+      async subscribe(req, ctx) {
+        ctx.subscribedEvents.add(req.eventId)
+        return {}
+      },
+      async unsubscribe(req, ctx) {
+        ctx.subscribedEvents.delete(req.eventId)
+        return {}
+      },
+      async crdtSendToRenderer(req, ctx) {
+        return crdtService.sendCrdt({
+          sceneId: ctx.sceneData.id,
+          payload: req.data,
+          sceneNumber: ctx.sceneData.sceneNumber
+        })
+      },
 
-    async crdtGetMessageFromRenderer(_, ctx) {
-      const protocol = await rendererProtocol
-      const response = await protocol.crdtService.pullCrdt({
-        sceneId: ctx.sceneData.id,
-        sceneNumber: ctx.sceneData.sceneNumber
-      })
-      return { data: [response.payload] }
+      async crdtGetMessageFromRenderer(_, ctx) {
+        const response = await crdtService.pullCrdt({
+          sceneId: ctx.sceneData.id,
+          sceneNumber: ctx.sceneData.sceneNumber
+        })
+        return { data: [response.payload] }
+      }
     }
-  }))
+  })
 }
 function eaTypeToStr(type: EAType): EntityActionType | null {
   switch (type) {

packages/shared/apis/host/EngineAPI.ts

@@ -2,6 +2,7 @@ import { ILogger } from './../../logger'
 import { EntityAction, LoadableScene } from './../../types'
 import { PermissionItem } from '@dcl/protocol/out-ts/decentraland/kernel/apis/permissions.gen'
 import { EventData } from '@dcl/protocol/out-ts/decentraland/kernel/apis/engine_api.gen'
+import { RpcClientPort } from '@dcl/rpc'
 
 type WithRequired<T, K extends keyof T> = T & { [P in K]-?: T[P] }
 
@@ -22,4 +23,7 @@ export type PortContext = {
   sendSceneEvent<K extends keyof IEvents>(id: K, event: IEvents[K]): void
   sendProtoSceneEvent(event: EventData): void
   logger: ILogger
+
+  // port used for this specific scene in the renderer
+  rendererPort: RpcClientPort
 }

packages/shared/apis/host/context.ts

@@ -10,7 +10,7 @@ export class OfflineAdapter implements MinimumCommunicationsAdapter {
   async getVoiceHandler(): Promise<VoiceHandler> {
     return createOpusVoiceHandler()
   }
-  async disconnect(error?: Error | undefined): Promise<void> {}
-  send(data: Uint8Array, hints: SendHints): void {}
+  async disconnect(_error?: Error | undefined): Promise<void> {}
+  send(_data: Uint8Array, _hints: SendHints): void {}
   async connect(): Promise<void> {}
 }

packages/shared/comms/adapters/OfflineAdapter.ts

@@ -46,9 +46,9 @@ export class SimulationRoom implements RoomConnection {
     this.tick = setInterval(this.update.bind(this), 60)
     this.roomConnection = new Rfc4RoomConnection({
       events: mitt<CommsAdapterEvents>(),
-      send(data: Uint8Array, hints: SendHints): void {},
+      send(_data: Uint8Array, _hints: SendHints): void {},
       async connect(): Promise<void> {},
-      async disconnect(error?: Error): Promise<void> {},
+      async disconnect(_error?: Error): Promise<void> {},
       async getVoiceHandler() {
         throw new Error('not implemented')
       }
@@ -190,11 +190,11 @@ export class SimulationRoom implements RoomConnection {
     }
   }
 
-  async disconnect(error?: Error | undefined): Promise<void> {
+  async disconnect(_error?: Error | undefined): Promise<void> {
     clearInterval(this.tick)
   }
 
-  send(data: Uint8Array, hints: SendHints): void {}
+  send(_data: Uint8Array, _hints: SendHints): void {}
 
   async connect(): Promise<void> {
     await Promise.all(new Array(100).fill(0).map(() => this.spawnPeer()))