-
Notifications
You must be signed in to change notification settings - Fork 82
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
fix: Reorder the closing of services, prevent sagas running multiple times and close backend server properly #2499
Changes from 5 commits
e1a97d4
c8718c5
a3951e0
bf0ebb5
f7b370c
cec2c0d
de2c55a
9758408
06dee86
c98b93e
6a0a5a8
05390e6
4483f71
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -223,47 +223,51 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI | |
} | ||
} | ||
|
||
public async closeAllServices(options: { saveTor: boolean } = { saveTor: false }) { | ||
public async closeAllServices( | ||
options: { saveTor: boolean; purgeLocalDb: boolean } = { saveTor: false, purgeLocalDb: false } | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Feels a little clunky to add another parameter here since There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [question] Why do we need to add this parameter? Maybe I'm reading into the name There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I'm not really sure, if we are sure that LevelDB would be totally cleared by There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If we close and then re-open the same LevelDB, is there data in memory that sticks around? Or does removing the LevelDB files clear everything? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. From the link above "DestroyDB operation will be There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I just tested this and it appears to work. |
||
) { | ||
this.logger('Closing services') | ||
|
||
await this.closeSocket() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Close the data server first |
||
|
||
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) { | ||
if (options.purgeLocalDb) { | ||
this.logger('Purging local DB') | ||
await this.localDbService.purge() | ||
} | ||
this.logger('Closing local DB') | ||
await this.localDbService.close() | ||
} | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Re-organizing. Data server should get closed first and then OrbitDB, then LibP2P |
||
|
||
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() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I remove the log because there is a similar log in |
||
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() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Again, I remove the log because there is a similar log in socketService.listen. We can keep both also, just let me know. |
||
this.logger('Attempting to redial peers!') | ||
if (this.peerInfo && (this.peerInfo?.connected.length !== 0 || this.peerInfo?.dialed.length !== 0)) { | ||
|
@@ -286,24 +290,17 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI | |
await this.socketService.init() | ||
} | ||
|
||
public async leaveCommunity(): Promise<boolean> { | ||
public async leaveCommunity() { | ||
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 }) | ||
await this.closeAllServices({ saveTor: true, purgeLocalDb: true }) | ||
|
||
this.logger('Purging data') | ||
await this.purgeData() | ||
|
||
this.logger('Resetting Tor') | ||
this.tor.resetHiddenServices() | ||
|
||
this.logger('Resetting state') | ||
await this.resetState() | ||
|
||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -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' | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [nitpick] unused import |
||
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,72 @@ 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 => { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [bug] This seems like a confusion of the server and the socket io. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. And in my testing never seems to be called There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 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 |
||
sockets.add(conn) | ||
conn.on('close', () => { | ||
sockets.delete(conn) | ||
}) | ||
}) | ||
|
||
return () => sockets.forEach(s => s.destroy()) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [design] This feels very unintuitive to me. |
||
} | ||
|
||
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> => { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [refactor] |
||
return await new Promise(resolve => { | ||
if (this.serverIoProvider.server.listening) resolve() | ||
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) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [question] Wouldn't it be better to just clean up the existing connections and proceed with listening? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 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}`) | ||
this.listening = true | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [design] This seems like reinventing the wheel. I feel like this just introduces chances for desync between the state of the http server and the state of |
||
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.listening) { | ||
this.logger('Data server is not running.') | ||
resolve() | ||
return | ||
} | ||
|
||
this.serverIoProvider.io.close(err => { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [question] Does this suffer from the same issue mentioned in the cleanup function stuck in There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 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') | ||
this.listening = false | ||
resolve() | ||
}) | ||
this.logger('Disconnecting sockets') | ||
this.closeSockets() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [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. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I don't think we need to detach any listeners. They are just attached once and persist between opening and closing of the server. |
||
}) | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This only needs to happen in sequence |
||
} | ||
|
||
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) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This only needs to happen once ( |
||
} | ||
|
||
function* handleSocketLifecycleActions(socket: Socket, socketIOData: WebsocketConnectionPayload): Generator { | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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') | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Just adding additional logging around all of these long-running sagas (anything that runs until cancelled... so anything with a |
||
} | ||
} | ||
|
||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,4 +1,4 @@ | ||
import { select, call, takeLeading, putResolve } from 'typed-redux-saga' | ||
import { select, call, putResolve } 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* putResolve(app.actions.closeServices()) | ||
|
||
yield takeLeading(initActions.canceledRootTask.type, clearReduxStore) | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We never want to run |
||
|
||
export function* clearReduxStore(): Generator { | ||
|
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