fix(DHT): NET-1325 remove ScaleDownDht.test.ts, fix WS server graceful leave bug

packages/dht/src/connection/ConnectionLockRpcLocal.ts

@@ -18,7 +18,7 @@ import { DhtAddress, areEqualPeerDescriptors, getNodeIdFromPeerDescriptor } from
 interface ConnectionLockRpcLocalOptions {
     addRemoteLocked: (id: DhtAddress, lockId: LockID) => void
     removeRemoteLocked: (id: DhtAddress, lockId: LockID) => void
-    closeConnection: (peerDescriptor: PeerDescriptor, gracefulLeave: boolean, reason?: string) => void
+    closeConnection: (peerDescriptor: PeerDescriptor, gracefulLeave: boolean, reason?: string) => Promise<void>
     getLocalPeerDescriptor: () => PeerDescriptor
 }
 
@@ -60,9 +60,9 @@ export class ConnectionLockRpcLocal implements IConnectionLockRpc {
         logger.trace(getNodeIdOrUnknownFromPeerDescriptor(senderPeerDescriptor) + ' received gracefulDisconnect notice')
 
         if (disconnectNotice.disconnectMode === DisconnectMode.LEAVING) {
-            this.options.closeConnection(senderPeerDescriptor, true, 'graceful leave notified')
+            await this.options.closeConnection(senderPeerDescriptor, true, 'graceful leave notified')
         } else {
-            this.options.closeConnection(senderPeerDescriptor, false, 'graceful disconnect notified')
+            await this.options.closeConnection(senderPeerDescriptor, false, 'graceful disconnect notified')
         }
         return {}
     }

packages/dht/src/connection/ConnectionManager.ts

@@ -152,10 +152,8 @@ export class ConnectionManager extends EventEmitter<TransportEvents> implements
         const lockRpcLocal = new ConnectionLockRpcLocal({
             addRemoteLocked: (id: DhtAddress, lockId: LockID) => this.locks.addRemoteLocked(id, lockId),
             removeRemoteLocked: (id: DhtAddress, lockId: LockID) => this.locks.removeRemoteLocked(id, lockId),
-            closeConnection: (peerDescriptor: PeerDescriptor, gracefulLeave: boolean, reason?: string) => {
-                // TODO should we have some handling for this floating promise?
-                this.closeConnection(peerDescriptor, gracefulLeave, reason)
-            },
+            closeConnection: (peerDescriptor: PeerDescriptor, gracefulLeave: boolean, reason?: string) => 
+                this.closeConnection(peerDescriptor, gracefulLeave, reason),
             getLocalPeerDescriptor: () => this.getLocalPeerDescriptor()
         })
         this.rpcCommunicator.registerRpcMethod(LockRequest, LockResponse, 'lockRequest',
@@ -224,8 +222,6 @@ export class ConnectionManager extends EventEmitter<TransportEvents> implements
         if (this.disconnectorIntervalRef) {
             clearInterval(this.disconnectorIntervalRef)
         }
-        await this.connectorFacade.stop()
-
         await Promise.all(Array.from(this.endpoints.values()).map(async (endpoint) => {
             if (endpoint.connected) {
                 try {
@@ -249,7 +245,7 @@ export class ConnectionManager extends EventEmitter<TransportEvents> implements
                 }
             }
         }))
-
+        await this.connectorFacade.stop()
         this.state = ConnectionManagerState.STOPPED
         this.rpcCommunicator!.stop()
         this.duplicateMessageDetector.clear()

packages/dht/test/integration/ConnectionManager.test.ts

@@ -416,4 +416,57 @@ describe('ConnectionManager', () => {
         expect(connectionManager1.getLocalPeerDescriptor().websocket!.tls).toEqual(false)
         await connectionManager1.stop()
     })
+
+    it('Stopping ConnectionManager is cleaned up from peers', async () => {
+        const connectionManager1 = createConnectionManager({
+            transport: mockTransport,
+            websocketHost: '127.0.0.1',
+            websocketServerEnableTls: false,
+            websocketPortRange: { min: 10005, max: 10005 }
+        })
+
+        await connectionManager1.start()
+
+        const connectionManager2 = createConnectionManager({
+            transport: mockTransport,
+            websocketHost: '127.0.0.1',
+            websocketServerEnableTls: false,
+            websocketPortRange: { min: 10006, max: 10006 }
+        })
+
+        await connectionManager2.start()
+
+        const msg: Message = {
+            serviceId: SERVICE_ID,
+            messageId: '1',
+            body: {
+                oneofKind: 'rpcMessage',
+                rpcMessage: RpcMessage.create()
+            },
+            targetDescriptor: connectionManager1.getLocalPeerDescriptor()
+        }
+
+        const connectedPromise1 = new Promise<void>((resolve, _reject) => {
+            connectionManager1.on('connected', () => {
+                resolve()
+            })
+        })
+
+        const connectedPromise2 = new Promise<void>((resolve, _reject) => {
+            connectionManager2.on('connected', () => {
+                resolve()
+            })
+        })
+        await Promise.all([connectedPromise1, connectedPromise2, connectionManager2.send(msg)])
+
+        expect(connectionManager1.getConnections().length).toEqual(1)
+        expect(connectionManager2.getConnections().length).toEqual(1)
+
+        await connectionManager1.stop()
+
+        expect(connectionManager2.getConnections().length).toEqual(0)
+
+        await connectionManager2.stop()
+    })
+
 })