fix: Reorder the closing of services, prevent sagas running multiple times and close backend server properly

packages/backend/ipfs-pubsub-peer-monitor.patch

@@ -0,0 +1,11 @@
+--- node_modules/ipfs-pubsub-peer-monitor/src/ipfs-pubsub-peer-monitor.js	2024-05-08 12:44:48
++++ node_modules/ipfs-pubsub-peer-monitor/src/ipfs-pubsub-peer-monitor.backup.js	2024-05-08 12:44:25
+@@ -55,7 +55,7 @@
+   async _pollPeers () {
+     try {
+       const peers = await this._pubsub.peers(this._topic)
+-      IpfsPubsubPeerMonitor._emitJoinsAndLeaves(new Set(this._peers), new Set(peers), this)
++      IpfsPubsubPeerMonitor._emitJoinsAndLeaves(new Set(this._peers.map(p => p.toString())), new Set(peers.map(p => p.toString())), this)
+       this._peers = peers
+     } catch (err) {
+       clearInterval(this._interval)

packages/backend/package.json

@@ -12,7 +12,7 @@
     "build": "tsc -p tsconfig.build.json",
     "webpack": "webpack --env mode=development && cp ./lib/bundle.cjs ../backend-bundle/bundle.cjs",
     "webpack:prod": "webpack --env mode=production && cp ./lib/bundle.cjs ../backend-bundle/bundle.cjs",
-    "applyPatches": "patch -f -p0 < ./electron-fetch.patch || true && patch -f -p0 --forward --binary < ./parse-duration.patch || true && patch -f -p0 --forward --binary < ./parse-duration-esm.patch || true",
+    "applyPatches": "patch -f -p0 < ./electron-fetch.patch || true && patch -f -p0 --forward --binary < ./parse-duration.patch || true && patch -f -p0 --forward --binary < ./parse-duration-esm.patch || true && patch -f -p0 < ./ipfs-pubsub-peer-monitor.patch",
     "prepare": "npm run applyPatches && npm run webpack",
     "version": "git add -A src",
     "lint:no-fix": "eslint --ext .jsx,.js,.ts,.tsx ./src/",

packages/backend/src/backendManager.ts

@@ -109,12 +109,12 @@ 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)
-    await connectionsManager.pause()
+    connectionsManager.pause()
   })
 
-  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 +123,7 @@ export const runBackendMobile = async () => {
     torControl.torControlParams.auth.value = msg.authCookie
     proxyAgent.proxy.port = msg.httpTunnelPort
 
-    await connectionsManager.resume()
+    connectionsManager.resume()
   })
 }
 

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

@@ -224,46 +224,44 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
   }
 
   public async closeAllServices(options: { saveTor: boolean } = { saveTor: false }) {
+    this.logger('Closing services')
+
+    await this.closeSocket()
+
     if (this.tor && !options.saveTor) {
       this.logger('Killing tor')
       await this.tor.kill()
     } else if (options.saveTor) {
       this.logger('Saving tor')
     }
     if (this.storageService) {
-      this.logger('Stopping orbitdb')
+      this.logger('Stopping OrbitDB')
       await this.storageService?.stopOrbitDb()
     }
-    if (this.serverIoProvider?.io) {
-      this.logger('Closing socket server')
-      this.serverIoProvider.io.close()
-    }
-    if (this.localDbService) {
-      this.logger('Closing local storage')
-      await this.localDbService.close()
-    }
     if (this.libp2pService) {
       this.logger('Stopping libp2p')
       await this.libp2pService.close()
     }
+    if (this.localDbService) {
+      this.logger('Closing local DB')
+      await this.localDbService.close()
+    }
   }
 
-  public closeSocket() {
-    this.serverIoProvider.io.close()
+  public async closeSocket() {
+    await this.socketService.close()
   }
 
   public async pause() {
     this.logger('Pausing!')
-    this.logger('Closing socket!')
-    this.closeSocket()
+    await this.closeSocket()
     this.logger('Pausing libp2pService!')
     this.peerInfo = await this.libp2pService?.pause()
     this.logger('Found the following peer info on pause: ', this.peerInfo)
   }
 
   public async resume() {
     this.logger('Resuming!')
-    this.logger('Reopening socket!')
     await this.openSocket()
     this.logger('Attempting to redial peers!')
     if (this.peerInfo && (this.peerInfo?.connected.length !== 0 || this.peerInfo?.dialed.length !== 0)) {
@@ -289,21 +287,14 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI
   public async leaveCommunity(): Promise<boolean> {
     this.logger('Running leaveCommunity')
 
-    this.logger('Resetting tor')
-    this.tor.resetHiddenServices()
-
-    this.logger('Closing the socket')
-    this.closeSocket()
-
-    this.logger('Purging local DB')
-    await this.localDbService.purge()
-
-    this.logger('Closing services')
     await this.closeAllServices({ saveTor: true })
 
     this.logger('Purging data')
     await this.purgeData()
 
+    this.logger('Resetting Tor')
+    this.tor.resetHiddenServices()
+
     this.logger('Resetting state')
     await this.resetState()
 

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

@@ -51,7 +51,7 @@ export class RegistrationService extends EventEmitter implements OnModuleInit {
       // Get the next event.
       const event = this.registrationEvents.shift()
       if (event) {
-        this.logger('Processing registration event', event)
+        this.logger('Processing registration event')
         // Event processing in progress
         this.registrationEventInProgress = true
 

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

@@ -27,13 +27,15 @@ import { CONFIG_OPTIONS, SERVER_IO_PROVIDER } from '../const'
 import { ConfigOptions, ServerIoProviderTypes } from '../types'
 import { suspendableSocketEvents } from './suspendable.events'
 import Logger from '../common/logger'
+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 sockets: Set<net.Socket>
 
   constructor(
     @Inject(SERVER_IO_PROVIDER) public readonly serverIoProvider: ServerIoProviderTypes,
@@ -44,12 +46,15 @@ export class SocketService extends EventEmitter implements OnModuleInit {
     this.readyness = new Promise<void>(resolve => {
       this.resolveReadyness = resolve
     })
+
+    this.sockets = new Set<net.Socket>()
+
+    this.attachListeners()
   }
 
   async onModuleInit() {
     this.logger('init: Started')
 
-    this.attachListeners()
     await this.init()
 
     this.logger('init: Finished')
@@ -71,7 +76,7 @@ export class SocketService extends EventEmitter implements OnModuleInit {
     this.logger('init: Frontend connected')
   }
 
-  private readonly attachListeners = (): void => {
+  private readonly attachListeners = () => {
     // Attach listeners here
     this.serverIoProvider.io.on(SocketActionTypes.CONNECTION, socket => {
       this.logger('Socket connection')
@@ -173,7 +178,7 @@ export class SocketService extends EventEmitter implements OnModuleInit {
         }
       )
 
-      socket.on(SocketActionTypes.LEAVE_COMMUNITY, async (callback: (closed: boolean) => void) => {
+      socket.on(SocketActionTypes.LEAVE_COMMUNITY, (callback: (closed: boolean) => void) => {
         this.logger('Leaving community')
         this.emit(SocketActionTypes.LEAVE_COMMUNITY, callback)
       })
@@ -195,25 +200,77 @@ 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.
+    this.serverIoProvider.server.on('connection', conn => {
+      this.sockets.add(conn)
+      conn.on('close', () => {
+        this.sockets.delete(conn)
+      })
+    })
+  }
+
+  public getConnections = (): Promise<number> => {
+    return new Promise(resolve => {
+      this.serverIoProvider.server.getConnections((err, count) => {
+        if (err) throw new Error(err.message)
+        resolve(count)
+      })
+    })
+  }
+
+  // Ensure the underlying connections get closed. See:
+  // https://github.com/socketio/socket.io/issues/1602
+  public closeSockets = () => {
+    this.logger('Disconnecting sockets')
+    this.sockets.forEach(s => s.destroy())
   }
 
-  public listen = async (port = this.configOptions.socketIOPort): Promise<void> => {
-    return await new Promise(resolve => {
-      if (this.serverIoProvider.server.listening) resolve()
+  public listen = async (): Promise<void> => {
+    this.logger(`Opening data server on port ${this.configOptions.socketIOPort}`)
+
+    if (this.serverIoProvider.server.listening) {
+      this.logger('Failed to listen. Server already listening.')
+      return
+    }
+
+    const numConnections = await this.getConnections()
+
+    if (numConnections > 0) {
+      this.logger('Failed to listen. Connections still open:', numConnections)
+      return
+    }
+
+    return new Promise(resolve => {
       this.serverIoProvider.server.listen(this.configOptions.socketIOPort, '127.0.0.1', () => {
         this.logger(`Data server running on port ${this.configOptions.socketIOPort}`)
         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}`)
+
+      if (!this.serverIoProvider.server.listening) {
+        this.logger('Data server is not running.')
+        resolve()
+        return
+      }
+
+      this.serverIoProvider.io.close(err => {
         if (err) throw new Error(err.message)
+        this.logger('Data server closed')
         resolve()
       })
+
+      this.serverIoProvider.io.disconnectSockets(true)
+      this.closeSockets()
     })
   }
 }

packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.ts

@@ -29,6 +29,7 @@ export class CertificatesRequestsStore extends EventEmitter {
         write: ['*'],
       },
     })
+    await this.store.load()
 
     this.store.events.on('write', async (_address, entry) => {
       this.logger('Added CSR to database')
@@ -40,8 +41,9 @@ export class CertificatesRequestsStore extends EventEmitter {
       this.loadedCertificateRequests()
     })
 
-    // @ts-ignore
-    await this.store.load({ fetchEntryTimeout: 15000 })
+    // // @ts-ignore
+    // await this.store.load({ fetchEntryTimeout: 15000 })
+
     this.logger('Initialized')
   }
 
@@ -52,9 +54,9 @@ export class CertificatesRequestsStore extends EventEmitter {
   }
 
   public async close() {
-    this.logger('Closing...')
+    this.logger('Closing certificate requests DB')
     await this.store?.close()
-    this.logger('Closed')
+    this.logger('Closed certificate requests DB')
   }
 
   public getAddress() {

packages/backend/src/nest/storage/certificates/certificates.store.ts

@@ -72,7 +72,9 @@ export class CertificatesStore extends EventEmitter {
   }
 
   public async close() {
+    this.logger('Closing certificates DB')
     await this.store?.close()
+    this.logger('Closed certificates DB')
   }
 
   public getAddress() {

packages/backend/src/nest/storage/communityMetadata/communityMetadata.store.ts

@@ -90,7 +90,9 @@ export class CommunityMetadataStore extends EventEmitter {
   }
 
   public async close() {
+    logger('Closing community metadata DB')
     await this.store?.close()
+    logger('Closed community metadata DB')
   }
 
   public async updateCommunityMetadata(newMeta: CommunityMetadata): Promise<CommunityMetadata | undefined> {

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

@@ -251,7 +251,9 @@ export class StorageService extends EventEmitter {
 
   public async stopOrbitDb() {
     try {
+      this.logger('Closing channels DB')
       await this.channels?.close()
+      this.logger('Closed channels DB')
     } catch (e) {
       this.logger.error('Error closing channels db', e)
     }

packages/backend/src/nest/storage/userProfile/userProfile.store.ts

@@ -86,7 +86,9 @@ export class UserProfileStore extends EventEmitter {
   }
 
   public async close() {
+    logger('Closing user profile DB')
     await this.store?.close()
+    logger('Closed user profile DB')
   }
 
   public async addUserProfile(userProfile: UserProfile) {

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

@@ -1,5 +1,18 @@
 import { io } from 'socket.io-client'
-import { select, put, call, cancel, fork, takeEvery, FixedTask, delay, apply, putResolve } from 'typed-redux-saga'
+import {
+  select,
+  put,
+  putResolve,
+  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'
@@ -49,17 +62,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 {