diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3274dae342..eb935ca176 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -35,9 +35,12 @@ jobs: # This would have been done part of npm pretest but it gives better # visibility in the CI if done as a separate step - name: Build wakunode2 - if: steps.cache-nim-waku.outputs.cache-hit != 'true' shell: bash - run: cd nim-waku && make wakunode2 + run: cd nim-waku && ./build/wakunode2 --help || make wakunode2 + + - name: Ensure wakunode2 is ready + shell: bash + run: cd nim-waku && ./build/wakunode2 --help - name: Install bufbuild uses: mu-io/setup-buf@v1beta @@ -68,3 +71,10 @@ jobs: - name: test run: npm run test + + - name: Upload logs on failure + uses: actions/upload-artifact@v2 + if: failure() + with: + name: nim-waku-logs + path: log/ diff --git a/src/lib/waku.ts b/src/lib/waku.ts index c39d3e8039..f9c3e0c9f7 100644 --- a/src/lib/waku.ts +++ b/src/lib/waku.ts @@ -3,8 +3,10 @@ import Mplex from 'libp2p-mplex'; import { bytes } from 'libp2p-noise/dist/src/@types/basic'; import { Noise } from 'libp2p-noise/dist/src/noise'; import TCP from 'libp2p-tcp'; +import Multiaddr from 'multiaddr'; +import PeerId from 'peer-id'; -import { WakuRelay, WakuRelayPubsub } from './waku_relay'; +import { CODEC, WakuRelay, WakuRelayPubsub } from './waku_relay'; export default class Waku { private constructor(public libp2p: Libp2p, public relay: WakuRelay) {} @@ -35,6 +37,11 @@ export default class Waku { return new Waku(libp2p, new WakuRelay(libp2p.pubsub)); } + async dialWithMultiAddr(peerId: PeerId, multiaddr: Multiaddr[]) { + this.libp2p.peerStore.addressBook.set(peerId, multiaddr); + await this.libp2p.dialProtocol(peerId, CODEC); + } + async stop() { await this.libp2p.stop(); } diff --git a/src/lib/waku_relay.spec.ts b/src/lib/waku_relay.spec.ts index 4a9c569fa7..ba1d02a86d 100644 --- a/src/lib/waku_relay.spec.ts +++ b/src/lib/waku_relay.spec.ts @@ -10,134 +10,244 @@ import { Message } from './waku_message'; import { CODEC, TOPIC } from './waku_relay'; describe('Waku Relay', () => { - // TODO: Fix this, see https://github.com/ChainSafe/js-libp2p-gossipsub/issues/151 - it.skip('Publish', async () => { - const message = Message.fromUtf8String('Bird bird bird, bird is the word!'); + afterEach(function () { + if (this.currentTest!.state === 'failed') { + console.log(`Test failed, log file name is ${makeLogFileName(this)}`); + } + }); - const [waku1, waku2] = await Promise.all([ + let waku1: Waku; + let waku2: Waku; + beforeEach(async function () { + [waku1, waku2] = await Promise.all([ Waku.create(NOISE_KEY_1), Waku.create(NOISE_KEY_2), ]); - // Add node's 2 data to the PeerStore - waku1.libp2p.peerStore.addressBook.set( - waku2.libp2p.peerId, - waku2.libp2p.multiaddrs - ); - await waku1.libp2p.dial(waku2.libp2p.peerId); + await waku1.dialWithMultiAddr(waku2.libp2p.peerId, waku2.libp2p.multiaddrs); + await Promise.all([ + new Promise((resolve) => + waku1.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ), + new Promise((resolve) => + waku2.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ), + ]); + + await waku1.relay.subscribe(); await waku2.relay.subscribe(); - await new Promise((resolve) => - waku2.libp2p.pubsub.once('pubsub:subscription-change', (...args) => - resolve(args) - ) - ); - // Setup the promise before publishing to ensure the event is not missed - const promise = waitForNextData(waku1.libp2p.pubsub); - - await waku2.relay.publish(message); - - const node1Received = await promise; - - expect(node1Received.isEqualTo(message)).to.be.true; - - await Promise.all([waku1.stop(), waku2.stop()]); + await Promise.all([ + new Promise((resolve) => + waku1.libp2p.pubsub.once('pubsub:subscription-change', (...args) => + resolve(args) + ) + ), + new Promise((resolve) => + waku2.libp2p.pubsub.once('pubsub:subscription-change', (...args) => + resolve(args) + ) + ), + ]); }); - it('Registers waku relay protocol', async function () { - const waku = await Waku.create(NOISE_KEY_1); + afterEach(async function () { + await waku1.stop(); + await waku2.stop(); + }); - const protocols = Array.from(waku.libp2p.upgrader.protocols.keys()); + it('Subscribe', async function () { + const subscribers1 = waku1.libp2p.pubsub.getSubscribers(TOPIC); + const subscribers2 = waku2.libp2p.pubsub.getSubscribers(TOPIC); + + expect(subscribers1).to.contain(waku2.libp2p.peerId.toB58String()); + expect(subscribers2).to.contain(waku1.libp2p.peerId.toB58String()); + }); + + it('Register correct protocols', async function () { + const protocols = Array.from(waku1.libp2p.upgrader.protocols.keys()); expect(protocols).to.contain(CODEC); - - await waku.stop(); + expect(protocols.findIndex((value) => value.match(/sub/))).to.eq(-1); }); - it('Does not register any sub protocol', async function () { - const waku = await Waku.create(NOISE_KEY_1); + // TODO: Fix this + it.skip('Publish', async function () { + this.timeout(10000); - const protocols = Array.from(waku.libp2p.upgrader.protocols.keys()); - expect(protocols.findIndex((value) => value.match(/sub/))).to.eq(-1); + const message = Message.fromUtf8String('JS to JS communication works'); + // waku.libp2p.pubsub.globalSignaturePolicy = 'StrictSign'; - await waku.stop(); + const receivedPromise = waitForNextData(waku2.libp2p.pubsub); + + await Promise.all([ + new Promise((resolve) => + waku1.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ), + new Promise((resolve) => + waku2.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ), + ]); + + await waku1.relay.publish(message); + + const receivedMsg = await receivedPromise; + + expect(receivedMsg.contentTopic).to.eq(message.contentTopic); + expect(receivedMsg.version).to.eq(message.version); + + const payload = Buffer.from(receivedMsg.payload!); + expect(Buffer.compare(payload, message.payload!)).to.eq(0); }); describe('Interop: Nim', function () { - let waku: Waku; - let nimWaku: NimWaku; + describe('Nim connects to js', function () { + let waku: Waku; + let nimWaku: NimWaku; - beforeEach(async function () { - this.timeout(10_000); - waku = await Waku.create(NOISE_KEY_1); + beforeEach(async function () { + this.timeout(10_000); + waku = await Waku.create(NOISE_KEY_1); - const peerId = waku.libp2p.peerId.toB58String(); - const localMultiaddr = waku.libp2p.multiaddrs.find((addr) => - addr.toString().match(/127\.0\.0\.1/) - ); - const multiAddrWithId = localMultiaddr + '/p2p/' + peerId; + const peerId = waku.libp2p.peerId.toB58String(); + const localMultiaddr = waku.libp2p.multiaddrs.find((addr) => + addr.toString().match(/127\.0\.0\.1/) + ); + const multiAddrWithId = localMultiaddr + '/p2p/' + peerId; - nimWaku = new NimWaku(makeLogFileName(this)); - await nimWaku.start({ staticnode: multiAddrWithId }); + nimWaku = new NimWaku(makeLogFileName(this)); + await nimWaku.start({ staticnode: multiAddrWithId }); - await waku.relay.subscribe(); - await new Promise((resolve) => - waku.libp2p.pubsub.once('gossipsub:heartbeat', resolve) - ); + await waku.relay.subscribe(); + await new Promise((resolve) => + waku.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ); + }); + + afterEach(async function () { + nimWaku ? nimWaku.stop() : null; + waku ? await waku.stop() : null; + }); + + it('nim subscribes to js', async function () { + const nimPeerId = await nimWaku.getPeerId(); + const subscribers = waku.libp2p.pubsub.getSubscribers(TOPIC); + + expect(subscribers).to.contain(nimPeerId.toB58String()); + }); + + it('Js publishes to nim', async function () { + this.timeout(5000); + + const message = Message.fromUtf8String('This is a message'); + + await waku.relay.publish(message); + + await nimWaku.waitForLog('WakuMessage received'); + + const msgs = await nimWaku.messages(); + + expect(msgs[0].contentTopic).to.equal(message.contentTopic); + expect(msgs[0].version).to.equal(message.version); + + const payload = Buffer.from(msgs[0].payload); + expect(Buffer.compare(payload, message.payload!)).to.equal(0); + }); + + it('Nim publishes to js', async function () { + this.timeout(5000); + const message = Message.fromUtf8String('Here is another message.'); + + await waku.relay.subscribe(); + + await new Promise((resolve) => + waku.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ); + + const receivedPromise = waitForNextData(waku.libp2p.pubsub); + + await nimWaku.sendMessage(message); + + const receivedMsg = await receivedPromise; + + expect(receivedMsg.contentTopic).to.eq(message.contentTopic); + expect(receivedMsg.version).to.eq(message.version); + + const payload = Buffer.from(receivedMsg.payload!); + expect(Buffer.compare(payload, message.payload!)).to.eq(0); + }); }); - afterEach(async function () { - nimWaku ? nimWaku.stop() : null; - waku ? await waku.stop() : null; - }); + describe('Js connects to nim', function () { + let waku: Waku; + let nimWaku: NimWaku; - it('nim subscribes to js', async function () { - const nimPeerId = await nimWaku.getPeerId(); - const subscribers = waku.libp2p.pubsub.getSubscribers(TOPIC); + beforeEach(async function () { + this.timeout(10_000); + waku = await Waku.create(NOISE_KEY_1); - expect(subscribers).to.contain(nimPeerId.toB58String()); - }); + nimWaku = new NimWaku(makeLogFileName(this)); + await nimWaku.start(); - it('Js publishes to nim', async function () { - this.timeout(5000); + const nimPeerId = await nimWaku.getPeerId(); - const message = Message.fromUtf8String('This is a message'); + await waku.dialWithMultiAddr(nimPeerId, [nimWaku.multiaddr]); - await waku.relay.publish(message); + await new Promise((resolve) => + waku.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ); - await nimWaku.waitForLog('WakuMessage received'); + await waku.relay.subscribe(); - const msgs = await nimWaku.messages(); + await new Promise((resolve) => + waku.libp2p.pubsub.once('gossipsub:heartbeat', resolve) + ); + }); - expect(msgs[0].contentTopic).to.equal(message.contentTopic); - expect(msgs[0].version).to.equal(message.version); + afterEach(async function () { + nimWaku ? nimWaku.stop() : null; + waku ? await waku.stop() : null; + }); - const payload = Buffer.from(msgs[0].payload); - expect(Buffer.compare(payload, message.payload!)).to.equal(0); - }); + it('nim subscribes to js', async function () { + const subscribers = waku.libp2p.pubsub.getSubscribers(TOPIC); - it('Nim publishes to js', async function () { - this.timeout(5000); - const message = Message.fromUtf8String('Here is another message.'); + const nimPeerId = await nimWaku.getPeerId(); + expect(subscribers).to.contain(nimPeerId.toB58String()); + }); - await waku.relay.subscribe(); + it('Js publishes to nim', async function () { + const message = Message.fromUtf8String('This is a message'); - await new Promise((resolve) => - waku.libp2p.pubsub.once('gossipsub:heartbeat', resolve) - ); + await waku.relay.publish(message); - const receivedPromise = waitForNextData(waku.libp2p.pubsub); + await nimWaku.waitForLog('WakuMessage received'); - await nimWaku.sendMessage(message); + const msgs = await nimWaku.messages(); - const receivedMsg = await receivedPromise; + expect(msgs[0].contentTopic).to.equal(message.contentTopic); + expect(msgs[0].version).to.equal(message.version); - expect(receivedMsg.contentTopic).to.eq(message.contentTopic); - expect(receivedMsg.version).to.eq(message.version); + const payload = Buffer.from(msgs[0].payload); + expect(Buffer.compare(payload, message.payload!)).to.equal(0); + }); - const payload = Buffer.from(receivedMsg.payload!); - expect(Buffer.compare(payload, message.payload!)).to.eq(0); + it('Nim publishes to js', async function () { + const message = Message.fromUtf8String('Here is another message.'); + + const receivedPromise = waitForNextData(waku.libp2p.pubsub); + + await nimWaku.sendMessage(message); + + const receivedMsg = await receivedPromise; + + expect(receivedMsg.contentTopic).to.eq(message.contentTopic); + expect(receivedMsg.version).to.eq(message.version); + + const payload = Buffer.from(receivedMsg.payload!); + expect(Buffer.compare(payload, message.payload!)).to.eq(0); + }); }); }); }); diff --git a/src/test_utils/nim_waku.ts b/src/test_utils/nim_waku.ts index 5d6957b2f6..484e93ce04 100644 --- a/src/test_utils/nim_waku.ts +++ b/src/test_utils/nim_waku.ts @@ -33,6 +33,7 @@ export interface Args { export class NimWaku { private process?: ChildProcess; + private pid?: number; private portsShift: number; private peerId?: PeerId; private logPath: string; @@ -42,7 +43,7 @@ export class NimWaku { this.logPath = `${LOG_DIR}/nim-waku_${logName}.log`; } - async start(args: Args) { + async start(args?: Args) { try { await existsAsync(LOG_DIR); } catch (e) { @@ -70,21 +71,38 @@ export class NimWaku { logFile, // stderr ], }); + this.pid = this.process.pid; + console.log( + `nim-waku ${ + this.process.pid + } started at ${new Date().toLocaleTimeString()}` + ); this.process.on('exit', (signal) => { - if (signal != 0) { - console.log(`nim-waku process exited with ${signal}`); - } + console.log( + `nim-waku ${ + this.process ? this.process.pid : this.pid + } process exited with ${signal} at ${new Date().toLocaleTimeString()}` + ); }); this.process.on('error', (err) => { - console.log(`nim-waku process encountered an error: ${err}`); + console.log( + `nim-waku ${ + this.process ? this.process.pid : this.pid + } process encountered an error: ${err} at ${new Date().toLocaleTimeString()}` + ); }); await this.waitForLog('RPC Server started'); } public stop() { + console.log( + `nim-waku ${ + this.process ? this.process.pid : this.pid + } getting SIGINT at ${new Date().toLocaleTimeString()}` + ); this.process ? this.process.kill('SIGINT') : null; this.process = undefined; }