fix: Fix all hanging tests, reduce flakyness of other tests

packages/dht/protos/DhtRpc.proto

@@ -46,7 +46,7 @@ service WebrtcConnectorRpc {
 service ConnectionLockRpc {
   rpc lockRequest (LockRequest) returns (LockResponse);
   rpc unlockRequest (UnlockRequest) returns (google.protobuf.Empty);
-  rpc gracefulDisconnect (DisconnectNotice) returns (DisconnectNoticeResponse);
+  rpc gracefulDisconnect (DisconnectNotice) returns (google.protobuf.Empty);
 }
 
 service ExternalApiRpc {
@@ -290,9 +290,6 @@ message DisconnectNotice {
   DisconnectMode disconnectMode = 1;
 }
 
-message DisconnectNoticeResponse {
-}
-
 message ExternalFindDataRequest {
   bytes key = 1;
 }

packages/dht/src/connection/ConnectionLockRpcRemote.ts

@@ -54,7 +54,7 @@ export class ConnectionLockRpcRemote extends RpcRemote<IConnectionLockRpcClient>
         const options = this.formDhtRpcOptions({
             doNotConnect: true,
             doNotMindStopped: true,
-            timeout: 2000  // TODO use config option or named constant?
+            timeout: 2001  // TODO use config option or named constant?
         })
         await this.getClient().gracefulDisconnect(request, options)
     }

packages/dht/src/connection/ConnectionManager.ts

@@ -14,7 +14,6 @@ import { protoToString } from '../helpers/protoToString'
 import {
     DisconnectMode,
     DisconnectNotice,
-    DisconnectNoticeResponse,
     LockRequest,
     LockResponse,
     Message,
@@ -147,7 +146,7 @@ export class ConnectionManager extends EventEmitter<TransportEvents> implements
             (req: LockRequest, context: ServerCallContext) => lockRpcLocal.lockRequest(req, context))
         this.rpcCommunicator.registerRpcNotification(UnlockRequest, 'unlockRequest',
             (req: UnlockRequest, context: ServerCallContext) => lockRpcLocal.unlockRequest(req, context))
-        this.rpcCommunicator.registerRpcMethod(DisconnectNotice, DisconnectNoticeResponse, 'gracefulDisconnect',
+        this.rpcCommunicator.registerRpcNotification(DisconnectNotice, 'gracefulDisconnect',
             (req: DisconnectNotice, context: ServerCallContext) => lockRpcLocal.gracefulDisconnect(req, context))
     }
 
@@ -156,7 +155,7 @@ export class ConnectionManager extends EventEmitter<TransportEvents> implements
             return
         }
         const disconnectionCandidates = new SortedContactList<Contact>({
-            referenceId: getNodeIdFromPeerDescriptor(this.getLocalPeerDescriptor()), 
+            referenceId: getNodeIdFromPeerDescriptor(this.getLocalPeerDescriptor()),
             maxSize: 100000,  // TODO use config option or named constant?
             allowToContainReferenceId: false,
             emitEvents: false
@@ -251,7 +250,10 @@ export class ConnectionManager extends EventEmitter<TransportEvents> implements
     }
 
     public async send(message: Message, doNotConnect = false, doNotMindStopped = false): Promise<void> {
-        if (this.state === ConnectionManagerState.STOPPED && !doNotMindStopped) {
+        if (
+            (this.state === ConnectionManagerState.STOPPED
+                || this.state === ConnectionManagerState.STOPPING)
+            && !doNotMindStopped) {
             return
         }
         const peerDescriptor = message.targetDescriptor!

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/ManagedConnection.ts

@@ -253,7 +253,7 @@ export class ManagedConnection extends EventEmitter<Events> {
 
             try {
                 result = await runAndRaceEvents3<OutpuBufferEvents>([() => { this.outputBuffer.push(data) }],
-                    this.outputBufferEmitter, ['bufferSent', 'bufferSendingFailed'], 15000)  // TODO use config option or named constant?
+                    this.outputBufferEmitter, ['bufferSent', 'bufferSendingFailed'], 15002)  // TODO use config option or named constant?
             } catch (e) {
                 logger.debug(`Connection to ${getNodeIdOrUnknownFromPeerDescriptor(this.remotePeerDescriptor)} timed out`, {
                     peerDescriptor: this.remotePeerDescriptor,

packages/dht/src/connection/simulator/SimulatorConnection.ts

@@ -55,6 +55,9 @@ export class SimulatorConnection extends Connection implements IConnection {
             + ', ' + getNodeIdFromPeerDescriptor(this.targetPeerDescriptor) + ' close()')
 
         if (!this.stopped) {
+            this.emit('disconnected', gracefulLeave, undefined, 'close() called')
+            this.removeAllListeners()
+
             logger.trace(getNodeIdFromPeerDescriptor(this.localPeerDescriptor) + ', '
                 + getNodeIdFromPeerDescriptor(this.targetPeerDescriptor) + ' close() not stopped')
             this.stopped = true

packages/dht/src/connection/simulator/SimulatorConnector.ts

@@ -32,6 +32,9 @@ export class SimulatorConnector {
     }
 
     public connect(targetPeerDescriptor: PeerDescriptor): ManagedConnection {
+        if (this.stopped) {
+            console.error('connect() called on stopped SimulatorConnector')
+        }
         logger.trace('connect() ' + getNodeIdFromPeerDescriptor(targetPeerDescriptor))
         const nodeId = getNodeIdFromPeerDescriptor(targetPeerDescriptor)
         const existingConnection = this.connectingConnections.get(nodeId)

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/src/dht/DhtNode.ts

@@ -77,6 +77,7 @@ export interface DhtNodeOptions {
     storageRedundancyFactor?: number
 
     transport?: ITransport
+    stopGivenTransport?: boolean
     peerDescriptor?: PeerDescriptor
     entryPoints?: PeerDescriptor[]
     websocketHost?: string
@@ -139,6 +140,7 @@ export class DhtNode extends EventEmitter<Events> implements ITransport {
     private readonly config: StrictDhtNodeOptions
     private rpcCommunicator?: RoutingRpcCommunicator
     private transport?: ITransport
+    private stopGivenTransport = false
     private localPeerDescriptor?: PeerDescriptor
     public router?: Router
     private storeManager?: StoreManager
@@ -190,6 +192,7 @@ export class DhtNode extends EventEmitter<Events> implements ITransport {
         // If transport is given, do not create a ConnectionManager
         if (this.config.transport) {
             this.transport = this.config.transport
+            this.stopGivenTransport = this.config.stopGivenTransport ?? false
             this.localPeerDescriptor = this.transport.getLocalPeerDescriptor()
             if (this.config.transport instanceof ConnectionManager) {
                 this.connectionManager = this.config.transport
@@ -572,7 +575,7 @@ export class DhtNode extends EventEmitter<Events> implements ITransport {
         this.router!.stop()
         this.recursiveOperationManager!.stop()
         this.peerDiscovery!.stop()
-        if (this.config.transport === undefined) {
+        if (this.config.transport === undefined || this.stopGivenTransport) {
             // if the transport was not given in config, the instance was created in start() and
             // this component is responsible for stopping it
             await this.transport!.stop()

packages/dht/src/dht/recursive-operation/RecursiveOperationManager.ts

@@ -134,7 +134,7 @@ export class RecursiveOperationManager implements IRecursiveOperationManager {
                     [() => this.doRouteRequest(routeMessage, excludedPeer)],
                     [[session, 'completed']],
                     // TODO use config option or named constant?
-                    15000
+                    15003
                 )
             } catch (err) {
                 logger.debug(`doRouteRequest failed with error ${err}`)
@@ -202,7 +202,7 @@ export class RecursiveOperationManager implements IRecursiveOperationManager {
                 .doSendResponse(routingPath, closestNodes, dataEntries, noCloserNodesFound)
         } else {
             // TODO use config option or named constant?
-            const remoteCommunicator = new ListeningRpcCommunicator(serviceId, this.sessionTransport, { rpcRequestTimeout: 15000 })
+            const remoteCommunicator = new ListeningRpcCommunicator(serviceId, this.sessionTransport, { rpcRequestTimeout: 15007 })
             const rpcRemote = new RecursiveOperationSessionRpcRemote(
                 this.localPeerDescriptor,
                 targetPeerDescriptor,

packages/dht/src/dht/recursive-operation/RecursiveOperationSession.ts

@@ -54,7 +54,7 @@ export class RecursiveOperationSession extends EventEmitter<RecursiveOperationSe
         })
         this.operation = config.operation
         this.rpcCommunicator = new ListeningRpcCommunicator(this.serviceId, this.transport, {
-            rpcRequestTimeout: 15000  // TODO use config option or named constant?
+            rpcRequestTimeout: 15001  // TODO use config option or named constant?
         })
         this.registerLocalRpcMethods()
     }

packages/dht/src/proto/google/protobuf/any.ts

@@ -75,7 +75,7 @@ import { MessageType } from "@protobuf-ts/runtime";
  *       foo = any.unpack(Foo.class);
  *     }
  *
- *  Example 3: Pack and unpack a message in Python.
+ * Example 3: Pack and unpack a message in Python.
  *
  *     foo = Foo(...)
  *     any = Any()
@@ -85,13 +85,16 @@ import { MessageType } from "@protobuf-ts/runtime";
  *       any.Unpack(foo)
  *       ...
  *
- *  Example 4: Pack and unpack a message in Go
+ * Example 4: Pack and unpack a message in Go
  *
  *      foo := &pb.Foo{...}
- *      any, err := ptypes.MarshalAny(foo)
+ *      any, err := anypb.New(foo)
+ *      if err != nil {
+ *        ...
+ *      }
  *      ...
  *      foo := &pb.Foo{}
- *      if err := ptypes.UnmarshalAny(any, foo); err != nil {
+ *      if err := any.UnmarshalTo(foo); err != nil {
  *        ...
  *      }
  *
@@ -103,7 +106,7 @@ import { MessageType } from "@protobuf-ts/runtime";
  *
  *
  * JSON
- * ====
+ *
  * The JSON representation of an `Any` value uses the regular
  * representation of the deserialized, embedded message, with an
  * additional field `@type` which contains the type URL. Example:

packages/dht/src/proto/google/protobuf/empty.ts

@@ -50,7 +50,6 @@ import { MessageType } from "@protobuf-ts/runtime";
  *       rpc Bar(google.protobuf.Empty) returns (google.protobuf.Empty);
  *     }
  *
- * The JSON representation for `Empty` is empty JSON object `{}`.
  *
  * @generated from protobuf message google.protobuf.Empty
  */

packages/dht/src/proto/google/protobuf/timestamp.ts

@@ -99,7 +99,16 @@ import { MessageType } from "@protobuf-ts/runtime";
  *         .setNanos((int) ((millis % 1000) * 1000000)).build();
  *
  *
- * Example 5: Compute Timestamp from current time in Python.
+ * Example 5: Compute Timestamp from Java `Instant.now()`.
+ *
+ *     Instant now = Instant.now();
+ *
+ *     Timestamp timestamp =
+ *         Timestamp.newBuilder().setSeconds(now.getEpochSecond())
+ *             .setNanos(now.getNano()).build();
+ *
+ *
+ * Example 6: Compute Timestamp from current time in Python.
  *
  *     timestamp = Timestamp()
  *     timestamp.GetCurrentTime()

packages/dht/src/proto/packages/dht/protos/DhtRpc.client.ts

@@ -7,7 +7,6 @@ import type { ExternalStoreDataRequest } from "./DhtRpc";
 import type { ExternalFindDataResponse } from "./DhtRpc";
 import type { ExternalFindDataRequest } from "./DhtRpc";
 import { ConnectionLockRpc } from "./DhtRpc";
-import type { DisconnectNoticeResponse } from "./DhtRpc";
 import type { DisconnectNotice } from "./DhtRpc";
 import type { UnlockRequest } from "./DhtRpc";
 import type { LockResponse } from "./DhtRpc";
@@ -314,9 +313,9 @@ export interface IConnectionLockRpcClient {
      */
     unlockRequest(input: UnlockRequest, options?: RpcOptions): UnaryCall<UnlockRequest, Empty>;
     /**
-     * @generated from protobuf rpc: gracefulDisconnect(dht.DisconnectNotice) returns (dht.DisconnectNoticeResponse);
+     * @generated from protobuf rpc: gracefulDisconnect(dht.DisconnectNotice) returns (google.protobuf.Empty);
      */
-    gracefulDisconnect(input: DisconnectNotice, options?: RpcOptions): UnaryCall<DisconnectNotice, DisconnectNoticeResponse>;
+    gracefulDisconnect(input: DisconnectNotice, options?: RpcOptions): UnaryCall<DisconnectNotice, Empty>;
 }
 /**
  * @generated from protobuf service dht.ConnectionLockRpc
@@ -342,11 +341,11 @@ export class ConnectionLockRpcClient implements IConnectionLockRpcClient, Servic
         return stackIntercept<UnlockRequest, Empty>("unary", this._transport, method, opt, input);
     }
     /**
-     * @generated from protobuf rpc: gracefulDisconnect(dht.DisconnectNotice) returns (dht.DisconnectNoticeResponse);
+     * @generated from protobuf rpc: gracefulDisconnect(dht.DisconnectNotice) returns (google.protobuf.Empty);
      */
-    gracefulDisconnect(input: DisconnectNotice, options?: RpcOptions): UnaryCall<DisconnectNotice, DisconnectNoticeResponse> {
+    gracefulDisconnect(input: DisconnectNotice, options?: RpcOptions): UnaryCall<DisconnectNotice, Empty> {
         const method = this.methods[2], opt = this._transport.mergeOptions(options);
-        return stackIntercept<DisconnectNotice, DisconnectNoticeResponse>("unary", this._transport, method, opt, input);
+        return stackIntercept<DisconnectNotice, Empty>("unary", this._transport, method, opt, input);
     }
 }
 /**

packages/dht/src/proto/packages/dht/protos/DhtRpc.server.ts

@@ -5,7 +5,6 @@ import { ExternalStoreDataResponse } from "./DhtRpc";
 import { ExternalStoreDataRequest } from "./DhtRpc";
 import { ExternalFindDataResponse } from "./DhtRpc";
 import { ExternalFindDataRequest } from "./DhtRpc";
-import { DisconnectNoticeResponse } from "./DhtRpc";
 import { DisconnectNotice } from "./DhtRpc";
 import { UnlockRequest } from "./DhtRpc";
 import { LockResponse } from "./DhtRpc";
@@ -133,9 +132,9 @@ export interface IConnectionLockRpc<T = ServerCallContext> {
      */
     unlockRequest(request: UnlockRequest, context: T): Promise<Empty>;
     /**
-     * @generated from protobuf rpc: gracefulDisconnect(dht.DisconnectNotice) returns (dht.DisconnectNoticeResponse);
+     * @generated from protobuf rpc: gracefulDisconnect(dht.DisconnectNotice) returns (google.protobuf.Empty);
      */
-    gracefulDisconnect(request: DisconnectNotice, context: T): Promise<DisconnectNoticeResponse>;
+    gracefulDisconnect(request: DisconnectNotice, context: T): Promise<Empty>;
 }
 /**
  * @generated from protobuf service dht.ExternalApiRpc