Merge pull request #243 from status-im/185-idle-stream-abort

.cspell.json

@@ -35,6 +35,7 @@
     "ihave",
     "ihaves",
     "ineed",
+    "ipfs",
     "iwant",
     "jdev",
     "keccak",

CHANGELOG.md

@@ -25,6 +25,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 - **Breaking**: Renamed `WakuRelay.(add|delete)PrivateDecryptionKey` to `WakuRelay.(add|delete)DecryptionKey` to make it clearer that it accepts both symmetric keys and asymmetric private keys.
 - Upgrade libp2p to 0.32.0.
+- **Breaking**: Rename `keepAlive` option to `pingKeepAlive`.
+- Introduced new `relayKeepAlive` option on `Waku` with a default to 59s to send ping messages over relay to ensure the relay stream stays open.
+  This is a workaround until [js-libp2p#744](https://github.com/libp2p/js-libp2p/issues/744) is done as there are issues when TCP(?) timeouts and the stream gets closed
+  ([#185](https://github.com/status-im/js-waku/issues/185), [js-libp2p#939](https://github.com/libp2p/js-libp2p/issues/939))
 
 ### Fixed
 - Align `WakuMessage` readme example with actual code behaviour.

src/lib/waku.ts

@@ -16,11 +16,16 @@ import { Multiaddr, multiaddr } from 'multiaddr';
 import PeerId from 'peer-id';
 
 import { WakuLightPush } from './waku_light_push';
+import { WakuMessage } from './waku_message';
 import { RelayCodecs, WakuRelay } from './waku_relay';
+import { RelayPingContentTopic } from './waku_relay/constants';
 import { StoreCodec, WakuStore } from './waku_store';
 
 const websocketsTransportKey = Websockets.prototype[Symbol.toStringTag];
 
+const DefaultPingKeepAliveValueSecs = 0;
+const DefaultRelayKeepAliveValueSecs = 5 * 60;
+
 export interface CreateOptions {
   /**
    * The PubSub Topic to use. Defaults to {@link DefaultPubsubTopic}.
@@ -37,12 +42,19 @@ export interface CreateOptions {
    */
   pubsubTopic?: string;
   /**
-   * Set keep alive frequency in seconds: Waku will send a ping request to each peer
+   * Set keep alive frequency in seconds: Waku will send a `/ipfs/ping/1.0.0`
-   * after the set number of seconds. Set to 0 to disable the keep alive feature
+   * request to each peer after the set number of seconds. Set to 0 to disable.
    *
-   * @default 0
+   * @default {@link DefaultPingKeepAliveValueSecs}
    */
-  keepAlive?: number;
+  pingKeepAlive?: number;
+  /**
+   * Set keep alive frequency in seconds: Waku will send a ping message over
+   * relay to each peer after the set number of seconds. Set to 0 to disable.
+   *
+   * @default {@link DefaultRelayKeepAliveValueSecs}
+   */
+  relayKeepAlive?: number;
   /**
    * You can pass options to the `Libp2p` instance used by {@link Waku} using the {@link CreateOptions.libp2p} property.
    * This property is the same type than the one passed to [`Libp2p.create`](https://github.com/libp2p/js-libp2p/blob/master/doc/API.md#create)
@@ -67,7 +79,10 @@ export class Waku {
   public store: WakuStore;
   public lightPush: WakuLightPush;
 
-  private keepAliveTimers: {
+  private pingKeepAliveTimers: {
+    [peer: string]: ReturnType<typeof setInterval>;
+  };
+  private relayKeepAliveTimers: {
     [peer: string]: ReturnType<typeof setInterval>;
   };
 
@@ -81,22 +96,21 @@ export class Waku {
     this.relay = libp2p.pubsub as unknown as WakuRelay;
     this.store = store;
     this.lightPush = lightPush;
-    this.keepAliveTimers = {};
+    this.pingKeepAliveTimers = {};
+    this.relayKeepAliveTimers = {};
 
-    const keepAlive = options.keepAlive || 0;
+    const pingKeepAlive =
+      options.pingKeepAlive || DefaultPingKeepAliveValueSecs;
+    const relayKeepAlive =
+      options.relayKeepAlive || DefaultRelayKeepAliveValueSecs;
 
-    if (keepAlive !== 0) {
     libp2p.connectionManager.on('peer:connect', (connection: Connection) => {
-        this.startKeepAlive(connection.remotePeer, keepAlive);
+      this.startKeepAlive(connection.remotePeer, pingKeepAlive, relayKeepAlive);
     });
 
-      libp2p.connectionManager.on(
+    libp2p.connectionManager.on('peer:disconnect', (connection: Connection) => {
-        'peer:disconnect',
-        (connection: Connection) => {
       this.stopKeepAlive(connection.remotePeer);
-        }
+    });
-      );
-    }
   }
 
   /**
@@ -214,21 +228,42 @@ export class Waku {
     return localMultiaddr + '/p2p/' + this.libp2p.peerId.toB58String();
   }
 
-  private startKeepAlive(peerId: PeerId, periodSecs: number): void {
+  private startKeepAlive(
+    peerId: PeerId,
+    pingPeriodSecs: number,
+    relayPeriodSecs: number
+  ): void {
     // Just in case a timer already exist for this peer
     this.stopKeepAlive(peerId);
 
     const peerIdStr = peerId.toB58String();
-    this.keepAliveTimers[peerIdStr] = setInterval(() => {
+
+    if (pingPeriodSecs !== 0) {
+      this.pingKeepAliveTimers[peerIdStr] = setInterval(() => {
         Ping(this.libp2p, peerId);
-    }, periodSecs * 1000);
+      }, pingPeriodSecs * 1000);
+    }
+
+    if (relayPeriodSecs !== 0) {
+      this.relayKeepAliveTimers[peerIdStr] = setInterval(() => {
+        WakuMessage.fromBytes(new Uint8Array(), {
+          contentTopic: RelayPingContentTopic,
+        }).then((wakuMsg) => this.relay.send(wakuMsg));
+      }, relayPeriodSecs * 1000);
+    }
   }
 
   private stopKeepAlive(peerId: PeerId): void {
     const peerIdStr = peerId.toB58String();
-    if (this.keepAliveTimers[peerIdStr]) {
+
-      clearInterval(this.keepAliveTimers[peerIdStr]);
+    if (this.pingKeepAliveTimers[peerIdStr]) {
-      delete this.keepAliveTimers[peerIdStr];
+      clearInterval(this.pingKeepAliveTimers[peerIdStr]);
+      delete this.pingKeepAliveTimers[peerIdStr];
+    }
+
+    if (this.relayKeepAliveTimers[peerIdStr]) {
+      clearInterval(this.relayKeepAliveTimers[peerIdStr]);
+      delete this.relayKeepAliveTimers[peerIdStr];
     }
   }
 }

src/lib/waku_relay/constants.ts

@@ -14,6 +14,8 @@ export const RelayCodecs = [
  */
 export const DefaultPubsubTopic = '/waku/2/default-waku/proto';
 
+export const RelayPingContentTopic = '/relay-ping/1/ping/null';
+
 /**
  * RelayGossipFactor affects how many peers we will emit gossip to at each heartbeat.
  * We will send gossip to RelayGossipFactor * (total number of non-mesh peers), or