fix: Reorder the closing of services, prevent sagas running multiple times and close backend server properly #2499
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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,14 +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() | ||
|
There was a problem hiding this comment. [FYI] attachListeners() moved from There was a problem hiding this comment. Yea it does trigger before |
||
| } | ||
|
|
||
| async onModuleInit() { | ||
| this.logger('init: Started') | ||
| await this.init() | ||
| this.logger('init: Finished') | ||
| } | ||
|
|
||
|
|
@@ -71,7 +74,9 @@ export class SocketService extends EventEmitter implements OnModuleInit { | |
| this.logger('init: Frontend connected') | ||
| } | ||
|
|
||
| private readonly attachListeners = () => { | ||
|
There was a problem hiding this comment. [nitpick] lost the return type There was a problem hiding this comment. I think not specifying void is generally the standard TS convention |
||
| this.logger('Attaching listeners') | ||
|
|
||
| // 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, (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 | ||
| this.serverIoProvider.server.on('connection', conn => { | ||
|
There was a problem hiding this comment. [bug] This seems like a confusion of the server and the socket io. There was a problem hiding this comment. And in my testing never seems to be called There was a problem hiding this comment. This gets called for me. There are two different sockets, the socket.io Socket (https://socket.io/docs/v4/server-api/#socket) and the Node |
||
| 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 | ||
| // | ||
| // I also tried `this.serverIoProvider.io.disconnectSockets(true)` | ||
| // which didn't work for me, but we still call it. | ||
| public closeSockets = () => { | ||
| this.logger('Disconnecting sockets') | ||
| this.serverIoProvider.io.disconnectSockets(true) | ||
| this.sockets.forEach(s => s.destroy()) | ||
| } | ||
|
|
||
| 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) | ||
|
There was a problem hiding this comment. [question] Wouldn't it be better to just clean up the existing connections and proceed with listening? There was a problem hiding this comment. I'm not sure. I thought about that, but there might be some trickiness with timing due to the listen and close being async. I'll have to look into it more, so just went with the simplest option for now. |
||
| 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 = (): 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 => { | ||
|
There was a problem hiding this comment. [question] Does this suffer from the same issue mentioned in the cleanup function stuck in There was a problem hiding this comment. Yea, this doesn't appear to close all connections properly. I don't think we need to remove the listeners ever. So those just get added once in the constructor. |
||
| if (err) throw new Error(err.message) | ||
| this.logger('Data server closed') | ||
| resolve() | ||
| }) | ||
|
|
||
| this.closeSockets() | ||
|
There was a problem hiding this comment. [question] Shouldn't we also detach listeners when closing sockets? I mentioned it before, but I'd rather see a dedicated There was a problem hiding this comment. I don't think we need to detach any listeners. They are just attached once and persist between opening and closing of the server. |
||
| }) | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No need for async here since nothing awaits the event handler