fix: Sagas running multiple times and socket closing

packages/backend/src/backendManager.ts

@@ -109,12 +109,13 @@ export const runBackendMobile = async () => {
     { logger: ['warn', 'error', 'log', 'debug', 'verbose'] }
   )
 
-  rn_bridge.channel.on('close', async () => {
+  rn_bridge.channel.on('close', () => {
     const connectionsManager = app.get<ConnectionsManagerService>(ConnectionsManagerService)
+
     connectionsManager.closeSocket()
   })
 
-  rn_bridge.channel.on('open', async (msg: OpenServices) => {
+  rn_bridge.channel.on('open', (msg: OpenServices) => {
     const connectionsManager = app.get<ConnectionsManagerService>(ConnectionsManagerService)
     const torControl = app.get<TorControl>(TorControl)
     const proxyAgent = app.get<{ proxy: { port: string } }>(SOCKS_PROXY_AGENT)
@@ -123,7 +124,11 @@ export const runBackendMobile = async () => {
     torControl.torControlParams.auth.value = msg.authCookie
     proxyAgent.proxy.port = msg.httpTunnelPort
 
-    await connectionsManager.openSocket()
+    // NOTE: There could be a race condition here. Nothing awaits
+    // these events and so it may be possible for the server to not be
+    // fully closed at this point which would result in `openSocket`
+    // failing to open the server.
+    connectionsManager.openSocket()
   })
 }
 

packages/backend/src/nest/connections-manager/connections-manager.service.ts

@@ -240,10 +240,8 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
     }
   }
 
-  public closeSocket() {
-    this.logger('Closing socket server')
-    // TODO: We should call this.socketService.close() instead
-    this.serverIoProvider.io.close()
+  public async closeSocket() {
+    await this.socketService.close()
   }
 
   // This method is only used on iOS through rn-bridge for reacting on lifecycle changes

packages/backend/src/nest/socket/socket.service.ts

@@ -27,13 +27,17 @@ import { CONFIG_OPTIONS, SERVER_IO_PROVIDER } from '../const'
 import { ConfigOptions, ServerIoProviderTypes } from '../types'
 import { suspendableSocketEvents } from './suspendable.events'
 import Logger from '../common/logger'
+import { sleep } from '../common/sleep'
+import type net from 'node:net'
 
 @Injectable()
 export class SocketService extends EventEmitter implements OnModuleInit {
   private readonly logger = Logger(SocketService.name)
 
   public resolveReadyness: (value: void | PromiseLike<void>) => void
   public readyness: Promise<void>
+  private listening: boolean
+  private closeSockets: () => void
 
   constructor(
     @Inject(SERVER_IO_PROVIDER) public readonly serverIoProvider: ServerIoProviderTypes,
@@ -44,12 +48,14 @@ export class SocketService extends EventEmitter implements OnModuleInit {
     this.readyness = new Promise<void>(resolve => {
       this.resolveReadyness = resolve
     })
+
+    this.listening = false
+    this.closeSockets = this.attachListeners()
   }
 
   async onModuleInit() {
     this.logger('init: Started')
 
-    this.attachListeners()
     await this.init()
 
     this.logger('init: Finished')
@@ -71,7 +77,7 @@ export class SocketService extends EventEmitter implements OnModuleInit {
     this.logger('init: Frontend connected')
   }
 
-  private readonly attachListeners = (): void => {
+  private readonly attachListeners = (): (() => void) => {
     // Attach listeners here
     this.serverIoProvider.io.on(SocketActionTypes.CONNECTION, socket => {
       this.logger('Socket connection')
@@ -195,25 +201,65 @@ export class SocketService extends EventEmitter implements OnModuleInit {
         this.emit(SocketActionTypes.LOAD_MIGRATION_DATA, data)
       })
     })
+
+    // Ensure the underlying connections get closed. See:
+    // https://github.com/socketio/socket.io/issues/1602
+    //
+    // I also tried `this.serverIoProvider.io.disconnectSockets(true)`
+    // which didn't work for me.
+    const sockets = new Set<net.Socket>()
+
+    this.serverIoProvider.server.on('connection', conn => {
+      sockets.add(conn)
+      conn.on('close', () => {
+        sockets.delete(conn)
+      })
+    })
+
+    return () => sockets.forEach(s => s.destroy())
+  }
+
+  public getConnections = (): Promise<number> => {
+    return new Promise(resolve => {
+      this.serverIoProvider.server.getConnections((err, count) => {
+        if (err) throw new Error(err.message)
+        resolve(count)
+      })
+    })
   }
 
   public listen = async (port = this.configOptions.socketIOPort): Promise<void> => {
+    this.logger(`Opening data server on port ${this.configOptions.socketIOPort}`)
+
+    // Sometimes socket.io closes the HTTP server but doesn't close
+    // all underlying connections. So it doesn't appear that
+    // `this.serverIoProvider.server.listening` is sufficient.
+    if (this.listening) {
+      const numConnections = await this.getConnections()
+      this.logger('Failed to listen. Connections still open:', numConnections)
+      return
+    }
+
     return await new Promise(resolve => {
-      if (this.serverIoProvider.server.listening) resolve()
       this.serverIoProvider.server.listen(this.configOptions.socketIOPort, '127.0.0.1', () => {
         this.logger(`Data server running on port ${this.configOptions.socketIOPort}`)
+        this.listening = true
         resolve()
       })
     })
   }
 
-  public close = async (): Promise<void> => {
-    this.logger(`Closing data server on port ${this.configOptions.socketIOPort}`)
-    return await new Promise(resolve => {
-      this.serverIoProvider.server.close(err => {
+  public close = (): Promise<void> => {
+    return new Promise(resolve => {
+      this.logger(`Closing data server on port ${this.configOptions.socketIOPort}`)
+      this.serverIoProvider.io.close(err => {
         if (err) throw new Error(err.message)
+        this.logger('Data server closed')
+        this.listening = false
         resolve()
       })
+      this.logger('Disconnecting sockets')
+      this.closeSockets()
     })
   }
 }

packages/mobile/src/store/init/startConnection/startConnection.saga.ts

@@ -1,5 +1,17 @@
 import { io } from 'socket.io-client'
-import { select, put, call, cancel, fork, takeEvery, FixedTask, delay, apply } from 'typed-redux-saga'
+import {
+  select,
+  put,
+  call,
+  cancel,
+  fork,
+  take,
+  takeLeading,
+  takeEvery,
+  FixedTask,
+  delay,
+  apply,
+} from 'typed-redux-saga'
 import { PayloadAction } from '@reduxjs/toolkit'
 import { socket as stateManager, Socket } from '@quiet/state-manager'
 import { encodeSecret } from '@quiet/common'
@@ -36,17 +48,20 @@ export function* startConnectionSaga(
   })
   yield* fork(handleSocketLifecycleActions, socket, action.payload)
   // Handle opening/restoring connection
-  yield* takeEvery(initActions.setWebsocketConnected, setConnectedSaga, socket)
+  yield* takeLeading(initActions.setWebsocketConnected, setConnectedSaga, socket)
 }
 
 function* setConnectedSaga(socket: Socket): Generator {
+  console.log('Frontend is ready. Forking state-manager sagas and starting backend...')
+
   const task = yield* fork(stateManager.useIO, socket)
-  console.log('WEBSOCKET', 'Forking state-manager sagas', task)
-  // Handle suspending current connection
-  yield* takeEvery(initActions.suspendWebsocketConnection, cancelRootTaskSaga, task)
-  console.log('Frontend is ready. Starting backend...')
+
   // @ts-ignore - Why is this broken?
   yield* apply(socket, socket.emit, [SocketActionTypes.START])
+
+  // Handle suspending current connection
+  const suspendAction = yield* take(initActions.suspendWebsocketConnection)
+  yield* call(cancelRootTaskSaga, task, suspendAction)
 }
 
 function* handleSocketLifecycleActions(socket: Socket, socketIOData: WebsocketConnectionPayload): Generator {

packages/mobile/src/store/nativeServices/events/nativeServicesCallbacks.ts

@@ -1,5 +1,5 @@
 import { eventChannel } from 'redux-saga'
-import { call, put, take } from 'typed-redux-saga'
+import { call, put, take, cancelled } from 'typed-redux-saga'
 import { app, publicChannels, WEBSOCKET_CONNECTION_CHANNEL, INIT_CHECK_CHANNEL, network } from '@quiet/state-manager'
 import { initActions, InitCheckPayload, WebsocketConnectionPayload } from '../../init/init.slice'
 import { ScreenNames } from '../../../const/ScreenNames.enum'
@@ -9,10 +9,18 @@ import { navigationActions } from '../../navigation/navigation.slice'
 import { nativeServicesActions } from '../nativeServices.slice'
 
 export function* nativeServicesCallbacksSaga(): Generator {
-  const channel = yield* call(deviceEvents)
-  while (true) {
-    const action = yield* take(channel)
-    yield put(action)
+  console.log('nativeServicesCallbacksSaga starting')
+  try {
+    const channel = yield* call(deviceEvents)
+    while (true) {
+      const action = yield* take(channel)
+      yield put(action)
+    }
+  } finally {
+    console.log('nativeServicesCallbacksSaga stopping')
+    if (yield cancelled()) {
+      console.log('nativeServicesCallbacksSaga cancelled')
+    }
   }
 }
 

packages/mobile/src/store/nativeServices/leaveCommunity/leaveCommunity.saga.ts

@@ -1,4 +1,4 @@
-import { select, call, put, takeLeading } from 'typed-redux-saga'
+import { select, call, put } from 'typed-redux-saga'
 import { app } from '@quiet/state-manager'
 import { persistor } from '../../store'
 import { nativeServicesActions } from '../nativeServices.slice'
@@ -9,11 +9,8 @@ import { ScreenNames } from '../../../../src/const/ScreenNames.enum'
 
 export function* leaveCommunitySaga(): Generator {
   console.log('Leaving community')
-
   // Restart backend
   yield* put(app.actions.closeServices())
-
-  yield takeLeading(initActions.canceledRootTask.type, clearReduxStore)
 }
 
 export function* clearReduxStore(): Generator {

packages/mobile/src/store/nativeServices/nativeServices.master.saga.ts

@@ -1,13 +1,21 @@
-import { all, fork, takeEvery } from 'typed-redux-saga'
+import { all, fork, takeEvery, cancelled } from 'typed-redux-saga'
 import { nativeServicesCallbacksSaga } from './events/nativeServicesCallbacks'
 import { leaveCommunitySaga } from './leaveCommunity/leaveCommunity.saga'
 import { flushPersistorSaga } from './flushPersistor/flushPersistor.saga'
 import { nativeServicesActions } from './nativeServices.slice'
 
 export function* nativeServicesMasterSaga(): Generator {
-  yield all([
-    fork(nativeServicesCallbacksSaga),
-    takeEvery(nativeServicesActions.leaveCommunity.type, leaveCommunitySaga),
-    takeEvery(nativeServicesActions.flushPersistor.type, flushPersistorSaga),
-  ])
+  console.log('nativeServicesMasterSaga starting')
+  try {
+    yield all([
+      fork(nativeServicesCallbacksSaga),
+      takeEvery(nativeServicesActions.leaveCommunity.type, leaveCommunitySaga),
+      takeEvery(nativeServicesActions.flushPersistor.type, flushPersistorSaga),
+    ])
+  } finally {
+    console.log('nativeServicesMasterSaga stopping')
+    if (yield cancelled()) {
+      console.log('nativeServicesMasterSaga cancelled')
+    }
+  }
 }

packages/mobile/src/store/root.saga.ts

@@ -1,4 +1,4 @@
-import { all, takeEvery, fork } from 'typed-redux-saga'
+import { all, takeEvery, takeLeading, fork, cancelled } from 'typed-redux-saga'
 import { nativeServicesMasterSaga } from './nativeServices/nativeServices.master.saga'
 import { navigationMasterSaga } from './navigation/navigation.master.saga'
 import { initMasterSaga } from './init/init.master.saga'
@@ -7,15 +7,25 @@ import { setupCryptoSaga } from './init/setupCrypto/setupCrypto.saga'
 import { publicChannels } from '@quiet/state-manager'
 import { showNotificationSaga } from './nativeServices/showNotification/showNotification.saga'
 import { restoreConnectionSaga } from './init/startConnection/restoreConnection/restoreConnection.saga'
+import { clearReduxStore } from './nativeServices/leaveCommunity/leaveCommunity.saga'
 
 export function* rootSaga(): Generator {
-  yield all([
-    takeEvery(initActions.setStoreReady.type, setupCryptoSaga),
-    takeEvery(initActions.setStoreReady.type, initMasterSaga),
-    takeEvery(initActions.setStoreReady.type, navigationMasterSaga),
-    takeEvery(initActions.setStoreReady.type, nativeServicesMasterSaga),
-    fork(restoreConnectionSaga),
-    // Below line is reponsible for displaying notifications about messages from channels other than currently viewing one
-    takeEvery(publicChannels.actions.markUnreadChannel.type, showNotificationSaga),
-  ])
+  console.log('rootSaga starting')
+  try {
+    yield all([
+      fork(setupCryptoSaga),
+      fork(initMasterSaga),
+      fork(navigationMasterSaga),
+      fork(nativeServicesMasterSaga),
+      fork(restoreConnectionSaga),
+      // Below line is reponsible for displaying notifications about messages from channels other than currently viewing one
+      takeEvery(publicChannels.actions.markUnreadChannel.type, showNotificationSaga),
+      takeLeading(initActions.canceledRootTask.type, clearReduxStore),
+    ])
+  } finally {
+    console.log('rootSaga stopping')
+    if (yield cancelled()) {
+      console.log('rootSaga cancelled')
+    }
+  }
 }

packages/state-manager/src/sagas/app/app.master.saga.ts

@@ -1,14 +1,22 @@
 import { Socket } from '../../types'
-import { all, takeEvery, takeLeading } from 'typed-redux-saga'
+import { all, takeEvery, takeLeading, cancelled } from 'typed-redux-saga'
 import { appActions } from './app.slice'
 import { closeServicesSaga } from './closeServices.saga'
 import { stopBackendSaga } from './stopBackend/stopBackend.saga'
 import { loadMigrationDataSaga } from './loadMigrationData/loadMigrationData.saga'
 
 export function* appMasterSaga(socket: Socket): Generator {
-  yield* all([
-    takeLeading(appActions.closeServices.type, closeServicesSaga, socket),
-    takeEvery(appActions.stopBackend.type, stopBackendSaga, socket),
-    takeEvery(appActions.loadMigrationData.type, loadMigrationDataSaga, socket),
-  ])
+  console.log('appMasterSaga starting')
+  try {
+    yield* all([
+      takeLeading(appActions.closeServices.type, closeServicesSaga, socket),
+      takeEvery(appActions.stopBackend.type, stopBackendSaga, socket),
+      takeEvery(appActions.loadMigrationData.type, loadMigrationDataSaga, socket),
+    ])
+  } finally {
+    console.log('appMasterSaga stopping')
+    if (yield cancelled()) {
+      console.log('appMasterSaga cancelled')
+    }
+  }
 }

packages/state-manager/src/sagas/appConnection/connection.master.saga.ts

@@ -1,6 +1,14 @@
-import { all, fork } from 'typed-redux-saga'
+import { all, fork, cancelled } from 'typed-redux-saga'
 import { uptimeSaga } from './uptime/uptime.saga'
 
 export function* connectionMasterSaga(): Generator {
-  yield all([fork(uptimeSaga)])
+  console.log('connectionMasterSaga starting')
+  try {
+    yield all([fork(uptimeSaga)])
+  } finally {
+    console.log('connectionMasterSaga stopping')
+    if (yield cancelled()) {
+      console.log('connectionMasterSaga cancelled')
+    }
+  }
 }

packages/state-manager/src/sagas/communities/communities.master.saga.ts

@@ -1,16 +1,24 @@
 import { type Socket } from '../../types'
-import { all, takeEvery } from 'typed-redux-saga'
+import { all, takeEvery, cancelled } from 'typed-redux-saga'
 import { communitiesActions } from './communities.slice'
 import { connectionActions } from '../appConnection/connection.slice'
 import { createCommunitySaga } from './createCommunity/createCommunity.saga'
 import { initCommunities, launchCommunitySaga } from './launchCommunity/launchCommunity.saga'
 import { createNetworkSaga } from './createNetwork/createNetwork.saga'
 
 export function* communitiesMasterSaga(socket: Socket): Generator {
-  yield all([
-    takeEvery(communitiesActions.createNetwork.type, createNetworkSaga, socket),
-    takeEvery(connectionActions.torBootstrapped.type, initCommunities),
-    takeEvery(communitiesActions.createCommunity.type, createCommunitySaga, socket),
-    takeEvery(communitiesActions.launchCommunity.type, launchCommunitySaga, socket),
-  ])
+  console.log('communitiesMasterSaga starting')
+  try {
+    yield all([
+      takeEvery(communitiesActions.createNetwork.type, createNetworkSaga, socket),
+      takeEvery(connectionActions.torBootstrapped.type, initCommunities),
+      takeEvery(communitiesActions.createCommunity.type, createCommunitySaga, socket),
+      takeEvery(communitiesActions.launchCommunity.type, launchCommunitySaga, socket),
+    ])
+  } finally {
+    console.log('communitiesMasterSaga stopping')
+    if (yield cancelled()) {
+      console.log('communitiesMasterSaga cancelled')
+    }
+  }
 }