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

.github/workflows/reusable/cached-build/action.yml

@@ -36,5 +36,7 @@ runs:
       shell: bash
     - name: build
       if: steps.cache-build.outputs.cache-hit != 'true'
-      run: npm run build
+      run: npm run build && npm ci --no-audit
       shell: bash
+
+

.github/workflows/validate.yml

@@ -122,7 +122,7 @@ jobs:
       package: broker
       docker-services: cassandra init-keyspace dev-chain-fast deploy-network-subgraphs-fastchain
       run-entry-point: true
-      command: npm run test-integration
+      command: npm run test-sequential && npm run test-integration
   client-browser:
     needs: build
     uses: ./.github/workflows/test-setup.yml

packages/broker/package.json

@@ -22,8 +22,8 @@
     "eslint": "eslint --cache --cache-location=node_modules/.cache/.eslintcache/ '*/**/*.{js,ts}'",
     "test": "npm run test-unit && npm run test-integration && npm run test-sequential",
     "test-unit": "jest test/unit",
-    "test-sequential": "jest --bail --forceExit --maxWorkers=1 test/sequential # always run sequential tests with maxWorkers=1",
-    "test-integration": "jest --bail --forceExit test/integration && npm run test-sequential"
+    "test-sequential": "streamr-keyserver 45456 800 899 & PIDS=$!;  sleep 5 && jest --bail --maxWorkers=1 test/sequential/$npm_config_test && kill $PIDS || (kill $PIDS; return 1)",
+    "test-integration": "streamr-keyserver 45455 500 799 & PID=$!;  sleep 5 && jest --bail test/integration/$npm_config_test && kill $PID || (kill $PID; return1)"
   },
   "author": "Streamr Network AG <[email protected]>",
   "license": "STREAMR NETWORK OPEN SOURCE LICENSE",

packages/broker/src/plugins/websocket/WebsocketServer.ts

@@ -68,8 +68,9 @@ export class WebsocketServer {
                 connection = this.createConnection(connectionUrl)
             } catch (err) {
                 logger.warn('Reject incoming connection', {
-                    requestUrl: request.url,
-                    reason: err?.message
+                    reason: err?.message,
+                    requestUrl: request.url
+
                 })
                 sendHttpError('400 Bad Request', socket)
                 return

packages/broker/test/integration/broker-subscriptions.test.ts

@@ -4,7 +4,7 @@ import { StreamrClient, Stream, StreamPartID, StreamPermission } from 'streamr-c
 import { fastWallet, fetchPrivateKeyWithGas } from '@streamr/test-utils'
 import { wait, waitForCondition } from '@streamr/utils'
 import { Broker } from '../../src/broker'
-import { startBroker, createClient, createTestStream } from '../utils'
+import { startBroker, createClient, createTestStream, KEYSERVER_PORT } from '../utils'
 
 jest.setTimeout(50000)
 
@@ -60,8 +60,8 @@ describe('broker subscriptions', () => {
             }
         })
 
-        client1 = createClient(await fetchPrivateKeyWithGas())
-        client2 = createClient(await fetchPrivateKeyWithGas())
+        client1 = createClient(await fetchPrivateKeyWithGas(KEYSERVER_PORT))
+        client2 = createClient(await fetchPrivateKeyWithGas(KEYSERVER_PORT))
 
         mqttClient1 = await createMqttClient(mqttPort1)
         mqttClient2 = await createMqttClient(mqttPort2)

packages/broker/test/integration/createMessagingPluginTest.ts

@@ -3,7 +3,7 @@ import { MessageMetadata, Stream, StreamrClient } from 'streamr-client'
 import { fetchPrivateKeyWithGas, Queue } from '@streamr/test-utils'
 import { Broker } from '../../src/broker'
 import { Message } from '../../src/helpers/PayloadFormat'
-import { createClient, startBroker, createTestStream } from '../utils'
+import { createClient, startBroker, createTestStream, KEYSERVER_PORT } from '../utils'
 import { wait, merge } from '@streamr/utils'
 
 interface MessagingPluginApi<T> {
@@ -58,7 +58,7 @@ export const createMessagingPluginTest = <T>(
         let messageQueue: Queue<Message>
 
         beforeAll(async () => {
-            brokerUser = new Wallet(await fetchPrivateKeyWithGas())
+            brokerUser = new Wallet(await fetchPrivateKeyWithGas(KEYSERVER_PORT))
             broker = await startBroker({
                 privateKey: brokerUser.privateKey,
                 apiAuthentication: {

packages/broker/test/integration/multiple-publisher-plugins.test.ts

@@ -4,7 +4,7 @@ import fetch from 'node-fetch'
 import { StreamPermission } from 'streamr-client'
 import { fetchPrivateKeyWithGas, Queue } from '@streamr/test-utils'
 import { Broker } from '../../src/broker'
-import { startBroker, createClient, createTestStream } from '../utils'
+import { startBroker, createClient, createTestStream, KEYSERVER_PORT } from '../utils'
 import { fastPrivateKey } from '@streamr/test-utils'
 import { wait, waitForEvent, waitForCondition } from '@streamr/utils'
 import sample from 'lodash/sample'
@@ -103,7 +103,7 @@ describe('multiple publisher plugins', () => {
     let streamId: string
 
     beforeAll(async () => {
-        privateKey = await fetchPrivateKeyWithGas()
+        privateKey = await fetchPrivateKeyWithGas(KEYSERVER_PORT)
         const client = createClient(privateKey)
         const stream = await createTestStream(client, module)
         streamId = stream.id

packages/broker/test/integration/plugins/mqtt/Bridge.test.ts

@@ -1,10 +1,11 @@
 import { Stream, StreamrClient } from 'streamr-client'
 import mqtt from 'async-mqtt'
-import { fetchPrivateKeyWithGas, Queue } from '@streamr/test-utils'
+import { fetchPrivateKeyWithGas } from '@streamr/test-utils'
 import { Broker } from '../../../../src/broker'
-import { createClient, startBroker, createTestStream } from '../../../utils'
-import { wait } from '@streamr/utils'
+import { createClient, startBroker, createTestStream, KEYSERVER_PORT } from '../../../utils'
+import { waitForEvent3 } from '@streamr/utils'
 import { Wallet } from '@ethersproject/wallet'
+import EventEmitter from 'eventemitter3'
 
 const MQTT_PLUGIN_PORT = 12470
 
@@ -14,21 +15,43 @@ const createMqttClient = () => {
     return mqtt.connectAsync(`mqtt://127.0.0.1:${MQTT_PLUGIN_PORT}`)
 }
 
+export interface SubscriberEvents {
+    messageObject: (messageObject: object) => void
+}
+class Subscriber extends EventEmitter<SubscriberEvents> {
+    private mqttClient?: mqtt.AsyncMqttClient
+    public readonly receivedMessageObjects: object[] = []
+
+    async subscribe(topic: string): Promise<void> {
+        this.mqttClient = await createMqttClient()
+        this.mqttClient.on('message', this.onMessage)
+        await this.mqttClient.subscribe(topic)     
+    }
+
+    async unsubscribe(topic: string): Promise<void> {
+        return this.mqttClient!.unsubscribe(topic)
+    }
+
+    async stop(): Promise<void> {
+        this.mqttClient!.off('message', this.onMessage)
+        await this.mqttClient!.end(true)
+    }
+
+    private onMessage = (_topic: string, message: Buffer): void => {
+        const messageObject = JSON.parse(message.toString())
+        this.receivedMessageObjects.push(messageObject)
+        this.emit('messageObject', messageObject)
+    }
+}
+
 describe('MQTT Bridge', () => {
     let stream: Stream
     let streamrClient: StreamrClient
     let broker: Broker
     let brokerUser: Wallet
 
-    const createSubscriber = async (messageQueue: Queue<any>) => {
-        const subscriber = await createMqttClient()
-        subscriber.on('message', (_topic, message) => messageQueue.push(JSON.parse(message.toString())))
-        subscriber.subscribe(stream.id)
-        return subscriber
-    }
-
     beforeAll(async () => {
-        brokerUser = new Wallet(await fetchPrivateKeyWithGas())
+        brokerUser = new Wallet(await fetchPrivateKeyWithGas(KEYSERVER_PORT))
         broker = await startBroker({
             privateKey: brokerUser.privateKey,
             extraPlugins: {
@@ -58,15 +81,20 @@ describe('MQTT Bridge', () => {
         const expected = {
             foo: Date.now()
         }
-        const messageQueue = new Queue<any>()
-        const subscriber = await createSubscriber(messageQueue)
+
+        const subscriber = new Subscriber()
+        const promise = waitForEvent3<SubscriberEvents>(subscriber, 'messageObject', 20000)
+
+        await subscriber.subscribe(stream.id)
         const publisher = await createMqttClient()
+
         publisher.publish(stream.id, JSON.stringify(expected))
+        await promise
 
-        expect(await messageQueue.pop()).toEqual(expected)
+        expect(subscriber.receivedMessageObjects[0]).toEqual(expected)
 
         await Promise.allSettled([
-            subscriber.end(true),
+            subscriber.stop(),
             publisher.end(true)
         ])
     })
@@ -75,70 +103,91 @@ describe('MQTT Bridge', () => {
         const expected = {
             foo: Date.now()
         }
-        const messageQueue = new Queue<any>()
-        const subscriber = await createSubscriber(messageQueue)
-        await streamrClient.publish(stream.id, expected)
+
+        const subscriber = new Subscriber()
+        const promise = waitForEvent3<SubscriberEvents>(subscriber, 'messageObject', 20000)
+        await subscriber.subscribe(stream.id)
 
-        expect(await messageQueue.pop()).toEqual(expected)
+        await streamrClient.publish(stream.id, expected)
+        await promise
+
+        expect(await subscriber.receivedMessageObjects[0]).toEqual(expected)
 
-        await subscriber.end(true)
+        await subscriber.stop()
     })
 
     test('message should be delivered once per client if subscribed by multiple clients', async () => {
         const expected = {
             foo: Date.now()
         }
-        const messageQueue1 = new Queue<any>()
-        const messageQueue2 = new Queue<any>()
-        const subscriber1 = await createSubscriber(messageQueue1)
-        const subscriber2 = await createSubscriber(messageQueue2)
+        const subscriber1 = new Subscriber()
+        const promise1 = waitForEvent3<SubscriberEvents>(subscriber1, 'messageObject', 20000)
+        await subscriber1.subscribe(stream.id)
+
+        const subscriber2 = new Subscriber()
+        const promise2 = waitForEvent3<SubscriberEvents>(subscriber2, 'messageObject', 20000)
+        await subscriber2.subscribe(stream.id)
+
         await streamrClient.publish(stream.id, expected)
 
-        expect(await messageQueue1.pop()).toEqual(expected)
-        expect(await messageQueue2.pop()).toEqual(expected)
+        await Promise.all([promise1, promise2])
+
+        expect(subscriber1.receivedMessageObjects[0]).toEqual(expected)
+        expect(subscriber2.receivedMessageObjects[0]).toEqual(expected)
 
         await Promise.allSettled([
-            subscriber1.end(true),
-            subscriber2.end(true)
+            subscriber1.stop(),
+            subscriber2.stop()
         ])
     })
 
     it('subscription should not be unsubscribed if it was not subscribed by that client', async () => {
         const expected = {
             foo: Date.now()
         }
-        const messageQueue = new Queue<any>()
-        const subscriber1 = await createSubscriber(messageQueue)
-        const subscriber2 = await createMqttClient()
-        subscriber2.unsubscribe(stream.id)
+
+        const subscriber1 = new Subscriber()
+        const promise1 = waitForEvent3<SubscriberEvents>(subscriber1, 'messageObject', 20000)
+        await subscriber1.subscribe(stream.id)
+
+        const subscriber2 = new Subscriber()
+        await subscriber2.subscribe(stream.id)
+
+        await subscriber2.unsubscribe(stream.id)
+
         await streamrClient.publish(stream.id, expected)
 
-        expect(await messageQueue.pop()).toEqual(expected)
+        await promise1
+        expect(subscriber1.receivedMessageObjects[0]).toEqual(expected)
 
         await Promise.allSettled([
-            subscriber1.end(true),
-            subscriber2.end(true)
+            subscriber1.stop(),
+            subscriber2.stop()
         ])
     })
 
     test('message should be delivered to remaining subscribers if one subscriber unsubscribes', async () => {
         const expected = {
             foo: Date.now()
         }
-        const messageQueue1 = new Queue<any>()
-        const messageQueue2 = new Queue<any>()
-        const subscriber1 = await createSubscriber(messageQueue1)
-        const subscriber2 = await createSubscriber(messageQueue2)
-        subscriber2.unsubscribe(stream.id)
+        const subscriber1 = new Subscriber()
+        const promise1 = waitForEvent3<SubscriberEvents>(subscriber1, 'messageObject', 20000)
+        await subscriber1.subscribe(stream.id)
+
+        const subscriber2 = new Subscriber()
+        await subscriber2.subscribe(stream.id)
+
+        await subscriber2.unsubscribe(stream.id)
         await streamrClient.publish(stream.id, expected)
 
-        expect(await messageQueue1.pop()).toEqual(expected)
-        await wait(100)  // wait for a while so that the message would have been delivered also to messageQueue2
-        expect(messageQueue2.values()).toEqual([])
+        await promise1
+        expect(subscriber1.receivedMessageObjects[0]).toEqual(expected)
+
+        expect(subscriber2.receivedMessageObjects).toEqual([])
 
         await Promise.allSettled([
-            subscriber1.end(true),
-            subscriber2.end(true)
+            subscriber1.stop(),
+            subscriber2.stop()
         ])
     })
 })

packages/broker/test/integration/plugins/mqtt/MqttPlugin.test.ts

@@ -24,7 +24,7 @@ createMessagingPluginTest('mqtt',
             await client.publish(streamId, JSON.stringify(msg))
         },
         subscribe: async (messageQueue: Queue<Message>, streamId: string, client: AsyncMqttClient): Promise<void> => {
-            client.once('message', (topic: string, message: Buffer) => {
+            client.on('message', (topic: string, message: Buffer) => {
                 if (topic === streamId) {
                     messageQueue.push(JSON.parse(message.toString()))
                 }

packages/broker/test/integration/plugins/operator/ContractFacade.test.ts

@@ -13,11 +13,11 @@ import {
     sponsor,
     stake
 } from './contractUtils'
-import { createClient, createTestStream } from '../../../utils'
+import { createClient, createTestStream, KEYSERVER_PORT } from '../../../utils'
 import { fetchPrivateKeyWithGas } from '@streamr/test-utils'
 
 async function createStream(): Promise<string> {
-    const client = createClient(await fetchPrivateKeyWithGas())
+    const client = createClient(await fetchPrivateKeyWithGas(KEYSERVER_PORT))
     const streamId = (await createTestStream(client, module)).id
     await client.destroy()
     return streamId

packages/broker/test/integration/plugins/operator/MaintainTopologyService.test.ts

@@ -3,7 +3,7 @@ import { fastPrivateKey, fetchPrivateKeyWithGas } from '@streamr/test-utils'
 import { toEthereumAddress, waitForCondition } from '@streamr/utils'
 import { Stream, StreamrClient } from 'streamr-client'
 import { OperatorFleetState } from '../../../../src/plugins/operator/OperatorFleetState'
-import { createClient, createTestStream } from '../../../utils'
+import { createClient, createTestStream, KEYSERVER_PORT } from '../../../utils'
 import {
     TEST_CHAIN_CONFIG,
     delegate,
@@ -19,7 +19,7 @@ import { MaintainTopologyService } from '../../../../src/plugins/operator/Mainta
 import { ContractFacade } from '../../../../src/plugins/operator/ContractFacade'
 
 async function setUpStreams(): Promise<[Stream, Stream]> {
-    const privateKey = await fetchPrivateKeyWithGas()
+    const privateKey = await fetchPrivateKeyWithGas(KEYSERVER_PORT)
     const client = createClient(privateKey)
     const s1 = await createTestStream(client, module, { partitions: 1 })
     const s2 = await createTestStream(client, module, { partitions: 3 })

packages/broker/test/integration/plugins/operator/OperatorPlugin.test.ts

@@ -4,7 +4,7 @@ import { collect, waitForCondition } from '@streamr/utils'
 import { Wallet } from 'ethers'
 import { ProxyDirection, StreamPermission } from 'streamr-client'
 import { Broker, createBroker } from '../../../../src/broker'
-import { createClient, createTestStream, formConfig, startBroker } from '../../../utils'
+import { createClient, createTestStream, formConfig, startBroker, KEYSERVER_PORT } from '../../../utils'
 import { delegate, deploySponsorshipContract, generateWalletWithGasAndTokens, setupOperatorContract, sponsor, stake } from './contractUtils'
 
 describe('OperatorPlugin', () => {
@@ -28,7 +28,7 @@ describe('OperatorPlugin', () => {
     })
 
     it('accepts proxy connections', async () => {
-        const subscriber = createClient(await fetchPrivateKeyWithGas())
+        const subscriber = createClient(await fetchPrivateKeyWithGas(KEYSERVER_PORT))
         const stream = await createTestStream(subscriber, module)
 
         const sponsorer = await generateWalletWithGasAndTokens()

packages/broker/test/integration/plugins/operator/maintainOperatorValue.test.ts

@@ -1,6 +1,6 @@
 import { fetchPrivateKeyWithGas } from '@streamr/test-utils'
 import { Logger, waitForCondition } from '@streamr/utils'
-import { createClient, createTestStream } from '../../../utils'
+import { createClient, createTestStream, KEYSERVER_PORT } from '../../../utils'
 import { delegate, deploySponsorshipContract, generateWalletWithGasAndTokens, setupOperatorContract, sponsor, stake } from './contractUtils'
 import { maintainOperatorValue } from '../../../../src/plugins/operator/maintainOperatorValue'
 import { multiply } from '../../../../src/helpers/multiply'
@@ -17,7 +17,7 @@ describe('maintainOperatorValue', () => {
 
     beforeAll(async () => {
         logger.debug('Creating stream for the test')
-        const client = createClient(await fetchPrivateKeyWithGas())
+        const client = createClient(await fetchPrivateKeyWithGas(KEYSERVER_PORT))
         streamId = (await createTestStream(client, module)).id
         await client.destroy()
     }, 60 * 1000)