Merge pull request #1584 from waku-org/fix/lightpush-error-1569

packages/core/src/lib/stream_manager.ts

@@ -6,8 +6,8 @@ import { selectConnection } from "@waku/utils/libp2p";
 import debug from "debug";
 
 export class StreamManager {
-  private streamPool: Map<string, Promise<Stream>>;
-  private log: debug.Debugger;
+  private streamPool: Map<string, Promise<Stream | void>>;
+  private readonly log: debug.Debugger;
 
   constructor(
     public multicodec: string,
@@ -38,7 +38,7 @@ export class StreamManager {
 
     const stream = await streamPromise;
 
-    if (stream.status === "closed") {
+    if (!stream || stream.status === "closed") {
       return this.newStream(peer); // fallback by creating a new stream on the spot
     }
 
@@ -55,7 +55,10 @@ export class StreamManager {
   }
 
   private prepareNewStream(peer: Peer): void {
-    const streamPromise = this.newStream(peer);
+    const streamPromise = this.newStream(peer).catch(() => {
+      // No error thrown as this call is not triggered by the user
+      this.log(`Failed to prepare a new stream for ${peer.id.toString()}`);
+    });
     this.streamPool.set(peer.id.toString(), streamPromise);
   }
 

packages/tests/tests/waku.node.spec.ts

@@ -12,15 +12,25 @@ import {
   createEncoder,
   generateSymmetricKey
 } from "@waku/message-encryption/symmetric";
-import { createLightNode, createRelayNode } from "@waku/sdk";
+import {
+  createLightNode,
+  createEncoder as createPlainEncoder,
+  createRelayNode
+} from "@waku/sdk";
 import { bytesToUtf8, utf8ToBytes } from "@waku/utils/bytes";
 import { expect } from "chai";
 
-import { makeLogFileName, NOISE_KEY_1, NOISE_KEY_2 } from "../src/index.js";
-import { NimGoNode } from "../src/node/node.js";
+import {
+  makeLogFileName,
+  NimGoNode,
+  NOISE_KEY_1,
+  NOISE_KEY_2
+} from "../src/index.js";
 
 const TestContentTopic = "/test/1/waku/utf8";
 
+const TestEncoder = createPlainEncoder({ contentTopic: TestContentTopic });
+
 describe("Waku Dial [node only]", function () {
   describe("Interop: NimGoNode", function () {
     let waku: Waku;
@@ -56,6 +66,35 @@ describe("Waku Dial [node only]", function () {
       const nimPeerId = await nwaku.getPeerId();
       expect(await waku.libp2p.peerStore.has(nimPeerId)).to.be.true;
     });
+
+    it("Does not throw an exception when node disconnects", async function () {
+      this.timeout(20_000);
+
+      process.on("unhandledRejection", (e) =>
+        expect.fail("unhandledRejection", e)
+      );
+      process.on("uncaughtException", (e) =>
+        expect.fail("uncaughtException", e)
+      );
+
+      nwaku = new NimGoNode(makeLogFileName(this));
+      await nwaku.start({
+        filter: true,
+        store: true,
+        lightpush: true
+      });
+      const multiAddrWithId = await nwaku.getMultiaddrWithId();
+
+      waku = await createLightNode({
+        staticNoiseKey: NOISE_KEY_1
+      });
+      await waku.start();
+      await waku.dial(multiAddrWithId);
+      await nwaku.stop();
+      await waku.lightPush?.send(TestEncoder, {
+        payload: utf8ToBytes("hello world")
+      });
+    });
   });
 
   describe("Bootstrap", function () {