streamr-1.0 fix-start-and-stop-of-autocertifier-client

packages/dht/src/connection/ConnectorFacade.ts

@@ -99,7 +99,7 @@ export class DefaultConnectorFacade implements ConnectorFacade {
         this.setLocalPeerDescriptor(localPeerDescriptor)
         if (localPeerDescriptor.websocket && !this.config.tlsCertificate && this.config.websocketServerEnableTls) {
             try {
-                await this.websocketConnector.autoCertify()
+                await this.websocketConnector.startAutocertifierClient()
                 const connectivityResponse = await this.websocketConnector.checkConnectivity(false)
                 const autocertifiedLocalPeerDescriptor = this.config.createLocalPeerDescriptor(connectivityResponse)
                 if (autocertifiedLocalPeerDescriptor.websocket !== undefined) {

packages/dht/src/connection/websocket/AutoCertifierClientFacade.ts

@@ -1,14 +1,14 @@
 import {
     AutoCertifierClient,
     HasSessionRequest,
-    HasSessionResponse, 
+    HasSessionResponse,
     CertifiedSubdomain,
     SERVICE_ID as AUTO_CERTIFIER_SERVICE_ID,
     HasSession
 } from '@streamr/autocertifier-client'
 import { ListeningRpcCommunicator } from '../../transport/ListeningRpcCommunicator'
-import { Logger, waitForEvent3 } from '@streamr/utils'
-import { ITransport } from '../../transport/ITransport' 
+import { Logger, withTimeout } from '@streamr/utils'
+import { ITransport } from '../../transport/ITransport'
 
 const START_TIMEOUT = 60 * 1000
 
@@ -20,14 +20,14 @@ const defaultAutoCertifierClientFactory = (
 ) => new AutoCertifierClient(
     configFile,
     wsServerPort,
-    autoCertifierUrl, 
+    autoCertifierUrl,
     (_serviceId: string, rpcMethodName: string, method: HasSession) => {
         autoCertifierRpcCommunicator.registerRpcMethod(
             HasSessionRequest,
             HasSessionResponse,
             rpcMethodName,
             method
-        )                       
+        )
     }
 )
 
@@ -36,7 +36,6 @@ export interface IAutoCertifierClient {
     stop(): void
     on(eventName: string, cb: (subdomain: CertifiedSubdomain) => void): void
 }
-
 interface AutoCertifierClientFacadeConfig {
     url: string
     configFile: string
@@ -57,12 +56,13 @@ export class AutoCertifierClientFacade {
     private readonly rpcCommunicator: ListeningRpcCommunicator
     private readonly setHost: (host: string) => void
     private readonly updateCertificate: (certificate: string, privateKey: string) => void
+    private abortController = new AbortController()
 
     constructor(config: AutoCertifierClientFacadeConfig) {
         this.setHost = config.setHost
         this.updateCertificate = config.updateCertificate
         this.rpcCommunicator = new ListeningRpcCommunicator(AUTO_CERTIFIER_SERVICE_ID, config.transport)
-        this.autoCertifierClient = config.createClientFactory ? config.createClientFactory() 
+        this.autoCertifierClient = config.createClientFactory ? config.createClientFactory()
             : defaultAutoCertifierClientFactory(
                 config.configFile,
                 config.url,
@@ -77,13 +77,18 @@ export class AutoCertifierClientFacade {
             this.updateCertificate(subdomain.certificate, subdomain.privateKey)
             logger.trace(`Updated certificate`)
         })
-        await Promise.all([
-            waitForEvent3(this.autoCertifierClient as any, 'updatedCertificate', START_TIMEOUT),
-            this.autoCertifierClient.start()
-        ])
+
+        // waiting for 'updatedCertificate' event is not needed here because
+        // all paths of this.autoCertifierClient.start() will block until 'updatedCertificate' 
+        // event is emitted or an exception is thrown
+
+        await withTimeout(this.autoCertifierClient.start(), START_TIMEOUT,
+            'AutoCertifierClient.start() timed out', this.abortController.signal)
+
     }
 
     stop(): void {
+        this.abortController.abort()
         this.autoCertifierClient.stop()
         this.rpcCommunicator.destroy()
     }

packages/dht/src/connection/websocket/WebsocketConnector.ts

@@ -135,7 +135,7 @@ export class WebsocketConnector {
         })
     }
 
-    public async autoCertify(): Promise<void> {
+    public async startAutocertifierClient(): Promise<void> {
         this.autoCertifierClient = new AutoCertifierClientFacade({
             configFile: this.autoCertifierConfigFile,
             transport: this.autoCertifierTransport,
@@ -144,7 +144,7 @@ export class WebsocketConnector {
             setHost: (hostName: string) => this.setHost(hostName),
             updateCertificate: (certificate: string, privateKey: string) => this.websocketServer!.updateCertificate(certificate, privateKey)
         })
-        logger.trace(`AutoCertifying subdomain...`)
+        logger.trace(`Starting autocertifier client...`)
         await this.autoCertifierClient.start()
     }
 
@@ -334,10 +334,12 @@ export class WebsocketConnector {
 
     public async destroy(): Promise<void> {
         this.abortController.abort()
-        this.rpcCommunicator.destroy()
 
-        const requests = Array.from(this.ongoingConnectRequests.values())
-        await Promise.allSettled(requests.map((conn) => conn.close(false)))
+        if (this.autoCertifierClient) {
+            await this.autoCertifierClient.stop()
+        }
+
+        this.rpcCommunicator.destroy()
 
         const attempts = Array.from(this.connectingConnections.values())
         await Promise.allSettled(attempts.map((conn) => conn.close(false)))

packages/dht/test/end-to-end/RecoveryFromFailedAutoCertification.test.ts

@@ -37,8 +37,8 @@ describe('Failed autocertification', () => {
 
     afterEach(async () => {
         await failedAutocertificationNode.stop()
-        await entryPoint.stop()
         await node.stop()
+        await entryPoint.stop()
     })
 
     it('failed auto certification should default to no tls', async () => {