servers clean up disconnects properly (#5086)

Co-authored-by: Gheric Speiginer <gheric.speiginer@gmail.com>

packages/engine/src/ecs/classes/World.ts

@@ -41,9 +41,6 @@ export class World {
     this.worldEntity = createEntity(this)
     this.localClientEntity = isClient ? (createEntity(this) as Entity) : (NaN as Entity)
 
-    this.userIdToUserIndex = new Map()
-    this.userIndexToUserId = new Map()
-
     if (!Engine.currentWorld) Engine.currentWorld = this
 
     addComponent(this.worldEntity, PersistTagComponent, {}, this)
@@ -90,10 +87,10 @@ export class World {
   actionHistory = new Set<Action>()
 
   /** Map of numerical user index to user client IDs */
-  userIndexToUserId: Map<number, UserId>
+  userIndexToUserId = new Map<number, UserId>()
 
   /** Map of user client IDs to numerical user index */
-  userIdToUserIndex: Map<UserId, number>
+  userIdToUserIndex = new Map<UserId, number>()
 
   userIndexCount = 0
 

packages/engine/src/ecs/functions/ActionCleanupSystem.ts

@@ -0,0 +1,28 @@
+import { Network } from '../../networking/classes/Network'
+import { NetworkTransport } from '../../networking/interfaces/NetworkTransport'
+import { World } from '../classes/World'
+
+const sendActionsOnTransport = (transport: NetworkTransport) => (world: World) => {
+  const { outgoingActions } = world
+
+  for (const o of outgoingActions) console.log('OUTGOING', o)
+
+  try {
+    transport.sendActions(outgoingActions)
+  } catch (e) {
+    console.error(e)
+  }
+
+  outgoingActions.clear()
+
+  return world
+}
+
+export default async function ActionDispatchSystem(world: World) {
+  const worldTransport = Network.instance.transportHandler.getWorldTransport()
+  const sendActions = sendActionsOnTransport(worldTransport)
+
+  return () => {
+    sendActions(world)
+  }
+}

packages/engine/src/initializeEngine.ts

@@ -139,12 +139,24 @@ export const initializeMediaServerSystems = async () => {
       args: { tickRate: 60 }
     },
     {
-      type: SystemUpdateType.FIXED,
+      type: SystemUpdateType.FIXED_EARLY,
       systemModulePromise: import('./ecs/functions/ActionDispatchSystem')
+    },
+    {
+      type: SystemUpdateType.FIXED_LATE,
+      systemModulePromise: Promise.resolve({
+        // media servers dont need to send actions anywhere
+        default: async function ClearOutgoingActionsSystem(world: World) {
+          return () => {
+            world.outgoingActions.clear()
+          }
+        }
+      })
     }
   )
 
   const world = useWorld()
+
   await initSystems(world, coreSystems)
 
   const executeWorlds = (delta, elapsedTime) => {
@@ -153,6 +165,8 @@ export const initializeMediaServerSystems = async () => {
     }
   }
 
+  NetworkActionReceptors.createNetworkActionReceptor(world)
+
   Engine.engineTimer = Timer(executeWorlds)
   Engine.engineTimer.start()
 
@@ -169,7 +183,7 @@ export const initializeCoreSystems = async (systems: SystemModuleType<any>[] = [
       args: { tickRate: 60 }
     },
     {
-      type: SystemUpdateType.FIXED,
+      type: SystemUpdateType.FIXED_EARLY,
       systemModulePromise: import('./ecs/functions/ActionDispatchSystem')
     },
     {
@@ -183,6 +197,10 @@ export const initializeCoreSystems = async (systems: SystemModuleType<any>[] = [
     {
       type: SystemUpdateType.FIXED_LATE,
       systemModulePromise: import('./scene/systems/SceneObjectSystem')
+    },
+    {
+      type: SystemUpdateType.FIXED_LATE,
+      systemModulePromise: import('./ecs/functions/ActionCleanupSystem')
     }
   )
 

packages/engine/src/networking/functions/dispatchFrom.ts

@@ -18,6 +18,7 @@ export const dispatchFrom = <A extends Action>(userId: UserId, actionCb: () => A
     action.$from = action.$from ?? Engine.userId
     action.$to = action.$to ?? 'all'
     action.$tick = action.$tick ?? Engine.currentWorld.fixedTick + 2
+    Engine.currentWorld.isHosting && Engine.currentWorld.incomingActions.add(action as any)
     Engine.currentWorld.outgoingActions.add(action)
   }
 

packages/engine/src/networking/systems/OutgoingNetworkSystem.ts

@@ -26,22 +26,6 @@ const serializeAndSend = (world: World, serialize: Function, sendData: Function)
   }
 }
 
-const sendActionsOnTransport = (transport: NetworkTransport) => (world: World) => {
-  const { outgoingActions } = world
-
-  for (const o of outgoingActions) console.log('OUTGOING', o)
-
-  try {
-    transport.sendActions(outgoingActions)
-  } catch (e) {
-    console.error(e)
-  }
-
-  outgoingActions.clear()
-
-  return world
-}
-
 const sendDataOnTransport = (transport: NetworkTransport) => (data) => {
   try {
     transport.sendData(data)
@@ -52,17 +36,13 @@ const sendDataOnTransport = (transport: NetworkTransport) => (data) => {
 
 export default async function OutgoingNetworkSystem(world: World) {
   const worldTransport = Network.instance.transportHandler.getWorldTransport()
-  const sendActions = sendActionsOnTransport(worldTransport)
   const sendData = sendDataOnTransport(worldTransport)
 
   const serialize = createDataWriter()
 
   return () => {
     if (!Engine.isInitialized) return
 
-    // side effect - network IO
-    sendActions(world)
-
     serializeAndSend(world, serialize, sendData)
   }
 }

packages/gameserver/src/channels.ts

@@ -22,7 +22,7 @@ import {
   initializeSceneSystems
 } from '@xrengine/engine/src/initializeEngine'
 import { Network } from '@xrengine/engine/src/networking/classes/Network'
-import { UserId } from '@xrengine/common/src/interfaces/UserId'
+import { HostUserId, UserId } from '@xrengine/common/src/interfaces/UserId'
 import { useWorld } from '@xrengine/engine/src/ecs/functions/SystemHooks'
 
 type InstanceMetadata = {
@@ -213,12 +213,21 @@ const loadEngine = async (app: Application, sceneId: string) => {
   if (app.isChannelInstance) {
     Network.instance.transportHandler.mediaTransports.set('media' as UserId, app.transport)
     Engine.publicPath = config.client.url
-    Engine.userId = 'media' as UserId
+    const userId = 'media' as HostUserId
+    Engine.userId = userId
     createEngine()
+    const world = useWorld()
+    world.hostId = userId
     initializeNode()
     await initializeMediaServerSystems()
     const projects = (await app.service('project').find(null!)).data.map((project) => project.name)
     await initializeProjectSystems(projects, [])
+
+    const hostIndex = world.userIndexCount++
+    world.clients.set(userId, { userId, name: 'media', userIndex: hostIndex })
+    world.userIdToUserIndex.set(userId, hostIndex)
+    world.userIndexToUserId.set(hostIndex, userId)
+
     Engine.sceneLoaded = true
     dispatchLocal(EngineActions.sceneLoaded(true) as any)
     dispatchLocal(EngineActions.joinedWorld(true) as any)
@@ -244,6 +253,8 @@ export default (app: Application): void => {
 
   let shutdownTimeout
   app.on('connection', async (connection) => {
+    console.log('connection')
+
     clearTimeout(shutdownTimeout)
     const token = (connection as any).socketQuery?.token
     if (token != null) {
@@ -463,7 +474,7 @@ export default (app: Application): void => {
 
         if (instanceId != null && instance != null) {
           const activeClients = Engine.currentWorld.clients
-          const activeUsers = new Map([...activeClients].filter(([, v]) => v.name !== 'server'))
+          const activeUsers = new Map([...activeClients].filter(([, v]) => v.name !== Engine.userId))
           const activeUsersCount = activeUsers.size
           try {
             await app.service('instance').patch(instanceId, {