Add support for gossipsub (#410)

* Replace floodsub with gossipsub

* Add a pubsub flag to support both floodsub and gossipsub

* Fix deps

* Handle empty string values

* Upgrade @cerc-io/libp2p dependency

* Upgrade package versions

lerna.json

@@ -2,7 +2,7 @@
   "packages": [
     "packages/*"
   ],
-  "version": "0.2.55",
+  "version": "0.2.56",
   "npmClient": "yarn",
   "useWorkspaces": true,
   "command": {

packages/cache/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cache",
-  "version": "0.2.55",
+  "version": "0.2.56",
   "description": "Generic object cache",
   "main": "dist/index.js",
   "scripts": {

packages/cli/README.md

@@ -51,6 +51,6 @@ A basic CLI to pass messages between peers using `stdin`/`stdout`
   * `dial-timeout`: timeout for dial to peers (ms)
   * `max-relay-connections`: max number of relay node connections for this peer
   * `peer-id-file (f)`: file path for peer id to be used (json)
-  * `enable-debug-info`: Whether to broadcast node's info over floodsub on request
+  * `enable-debug-info`: Whether to broadcast node's info over pubsub on request
 
 * The process starts reading from `stdin` and outputs messages from others peers over the `/chat/1.0.0` protocol to `stdout`.

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cli",
-  "version": "0.2.55",
+  "version": "0.2.56",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "scripts": {
@@ -12,11 +12,11 @@
   },
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.55",
-    "@cerc-io/ipld-eth-client": "^0.2.55",
-    "@cerc-io/peer": "^0.2.55",
-    "@cerc-io/rpc-eth-client": "^0.2.55",
-    "@cerc-io/util": "^0.2.55",
+    "@cerc-io/cache": "^0.2.56",
+    "@cerc-io/ipld-eth-client": "^0.2.56",
+    "@cerc-io/peer": "^0.2.56",
+    "@cerc-io/rpc-eth-client": "^0.2.56",
+    "@cerc-io/util": "^0.2.56",
     "@ethersproject/providers": "^5.4.4",
     "@graphql-tools/utils": "^9.1.1",
     "@ipld/dag-cbor": "^8.0.0",

packages/cli/src/peer.ts

@@ -14,6 +14,7 @@ import {
   PeerIdObj
   // @ts-expect-error https://github.com/microsoft/TypeScript/issues/49721#issuecomment-1319854183
 } from '@cerc-io/peer';
+import { PubsubType } from '@cerc-io/util';
 
 import { readPeerId } from './utils';
 
@@ -26,6 +27,7 @@ interface Arguments {
   dialTimeout: number;
   maxRelayConnections: number;
   peerIdFile: string;
+  pubsub?: PubsubType;
   enableDebugInfo: boolean;
 }
 
@@ -60,6 +62,7 @@ export class PeerCmd {
       maxConnections: argv.maxConnections,
       dialTimeout: argv.dialTimeout,
       maxRelayConnections: argv.maxRelayConnections,
+      pubsub: argv.pubsub,
       enableDebugInfo: argv.enableDebugInfo
     };
 
@@ -112,9 +115,13 @@ function _getArgv (): any {
       alias: 'f',
       describe: 'Peer id file path (json)'
     },
+    pubsub: {
+      type: 'string',
+      describe: "Pubsub to use ('floodsub' | 'gossipsub')"
+    },
     enableDebugInfo: {
       type: 'boolean',
-      describe: 'Whether to participate in exchanging debug info over floodsub',
+      describe: 'Whether to participate in exchanging debug info over pubsub',
       default: false
     }
   }).argv;

packages/cli/src/server.ts

@@ -187,6 +187,7 @@ export class ServerCmd {
         redialInterval: relayConfig.redialInterval ?? RELAY_REDIAL_INTERVAL,
         maxDialRetry: relayConfig.maxDialRetry ?? RELAY_DEFAULT_MAX_DIAL_RETRY,
         peerIdObj,
+        pubsub: relayConfig.pubsub,
         enableDebugInfo: relayConfig.enableDebugInfo
       };
       await createRelayNode(relayNodeInit);
@@ -212,6 +213,7 @@ export class ServerCmd {
         relayRedialInterval: peerConfig.relayRedialInterval,
         maxConnections: peerConfig.maxConnections,
         dialTimeout: peerConfig.dialTimeout,
+        pubsub: peerConfig.pubsub,
         enableDebugInfo: peerConfig.enableDebugInfo
       };
       await this._peer.init(peerNodeInit, peerIdObj);

packages/codegen/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/codegen",
-  "version": "0.2.55",
+  "version": "0.2.56",
   "description": "Code generator",
   "private": true,
   "main": "index.js",
@@ -20,7 +20,7 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/util": "^0.2.55",
+    "@cerc-io/util": "^0.2.56",
     "@graphql-tools/load-files": "^6.5.2",
     "@poanet/solidity-flattener": "https://github.com/vulcanize/solidity-flattener.git",
     "@solidity-parser/parser": "^0.13.2",

packages/codegen/src/templates/package-template.handlebars

@@ -41,12 +41,12 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.3.19",
-    "@cerc-io/cli": "^0.2.55",
-    "@cerc-io/ipld-eth-client": "^0.2.55",
-    "@cerc-io/solidity-mapper": "^0.2.55",
-    "@cerc-io/util": "^0.2.55",
+    "@cerc-io/cli": "^0.2.56",
+    "@cerc-io/ipld-eth-client": "^0.2.56",
+    "@cerc-io/solidity-mapper": "^0.2.56",
+    "@cerc-io/util": "^0.2.56",
     {{#if (subgraphPath)}}
-    "@cerc-io/graph-node": "^0.2.55",
+    "@cerc-io/graph-node": "^0.2.56",
     {{/if}}
     "@ethersproject/providers": "^5.4.4",
     "apollo-type-bigint": "^0.1.3",

packages/graph-node/package.json

@@ -1,10 +1,10 @@
 {
   "name": "@cerc-io/graph-node",
-  "version": "0.2.55",
+  "version": "0.2.56",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {
-    "@cerc-io/solidity-mapper": "^0.2.55",
+    "@cerc-io/solidity-mapper": "^0.2.56",
     "@ethersproject/providers": "^5.4.4",
     "@graphprotocol/graph-ts": "^0.22.0",
     "@nomiclabs/hardhat-ethers": "^2.0.2",
@@ -51,9 +51,9 @@
   "dependencies": {
     "@apollo/client": "^3.3.19",
     "@cerc-io/assemblyscript": "0.19.10-watcher-ts-0.1.2",
-    "@cerc-io/cache": "^0.2.55",
-    "@cerc-io/ipld-eth-client": "^0.2.55",
-    "@cerc-io/util": "^0.2.55",
+    "@cerc-io/cache": "^0.2.56",
+    "@cerc-io/ipld-eth-client": "^0.2.56",
+    "@cerc-io/util": "^0.2.56",
     "@types/json-diff": "^0.5.2",
     "@types/yargs": "^17.0.0",
     "bn.js": "^4.11.9",

packages/ipld-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/ipld-eth-client",
-  "version": "0.2.55",
+  "version": "0.2.56",
   "description": "IPLD ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -20,7 +20,7 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.55",
+    "@cerc-io/cache": "^0.2.56",
     "cross-fetch": "^3.1.4",
     "debug": "^4.3.1",
     "ethers": "^5.4.4",

packages/peer/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/peer",
-  "version": "0.2.55",
+  "version": "0.2.56",
   "description": "libp2p module",
   "main": "dist/index.js",
   "exports": "./dist/index.js",
@@ -25,11 +25,13 @@
     "dev": "node dist/index.js",
     "create-peer": "node dist/cli/create-peer.js",
     "relay-node": "DEBUG='laconic:*' node dist/cli/relay.js",
-    "test": "mocha dist/peer.test.js"
+    "test": "mocha dist/peer.test.js --bail"
   },
   "dependencies": {
-    "@cerc-io/libp2p": "0.42.2-laconic-0.1.3",
+    "@cerc-io/libp2p": "0.42.2-laconic-0.1.4",
     "@cerc-io/prometheus-metrics": "1.1.4",
+    "@cerc-io/util": "^0.2.56",
+    "@chainsafe/libp2p-gossipsub": "^6.0.0",
     "@chainsafe/libp2p-noise": "^11.0.0",
     "@libp2p/floodsub": "^6.0.0",
     "@libp2p/mplex": "^7.1.1",

packages/peer/src/cli/relay.ts

@@ -3,6 +3,8 @@ import { hideBin } from 'yargs/helpers';
 import fs from 'fs';
 import path from 'path';
 
+import { PubsubType } from '@cerc-io/util';
+
 import { RelayNodeInitConfig, createRelayNode } from '../relay.js';
 import { PeerIdObj } from '../peer.js';
 import {
@@ -25,6 +27,7 @@ interface Arguments {
   pingInterval: number;
   redialInterval: number;
   maxDialRetry: number;
+  pubsub?: PubsubType;
   enableDebugInfo?: boolean;
 }
 
@@ -81,12 +84,13 @@ async function main (): Promise<void> {
     pingInterval: argv.pingInterval,
     redialInterval: argv.redialInterval,
     maxDialRetry: argv.maxDialRetry,
+    pubsub: argv.pubsub,
     enableDebugInfo: argv.enableDebugInfo
   };
   await createRelayNode(relayNodeInit);
 }
 
-function _getArgv (): Arguments {
+function _getArgv (): any {
   return yargs(hideBin(process.argv)).parserConfiguration({
     'parse-numbers': false
   }).options({
@@ -141,6 +145,10 @@ function _getArgv (): Arguments {
       describe: 'Maximum number of dial retries to be attempted to a relay peer',
       default: RELAY_DEFAULT_MAX_DIAL_RETRY
     },
+    pubsub: {
+      type: 'string',
+      describe: "Pubsub to use ('floodsub' | 'gossipsub')"
+    },
     enableDebugInfo: {
       type: 'boolean',
       describe: "Whether to broadcast node's info over floodsub on request"

packages/peer/src/constants.ts

@@ -7,6 +7,8 @@ export const P2P_CIRCUIT_ID = 'p2p-circuit';
 export const CHAT_PROTOCOL = '/chat/1.0.0';
 export const DEBUG_INFO_TOPIC = 'debug-info';
 
+export const DEFAULT_PUBSUB_TYPE = 'floodsub';
+
 // How often a peer should broadcast it's peer data over pubsub discovery topic
 // (interval at which other peers get corresponding discovery event)
 export const PUBSUB_DISCOVERY_INTERVAL = 10000; // 10 seconds

packages/peer/src/peer.test.ts

@@ -61,6 +61,12 @@ describe('basic p2p tests', () => {
       const otherPeersId = peers[1 - index].node?.peerId.toString();
 
       return new Promise<void>((resolve, reject) => {
+        // Resolve if already connected to the other peer
+        const alreadyConnected = peer.node?.getPeers().some(peerId => peerId.toString() === otherPeersId);
+        if (alreadyConnected) {
+          resolve();
+        }
+
         peer.node?.addEventListener('peer:connect', async (event) => {
           const connection: Connection = event.detail;
 

packages/peer/src/peer.ts

@@ -12,6 +12,7 @@ import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string';
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string';
 import debug from 'debug';
 
+import { PubsubType } from '@cerc-io/util';
 import { createLibp2p, Libp2p, Libp2pInit } from '@cerc-io/libp2p';
 import { webSockets } from '@libp2p/websockets';
 import { noise } from '@chainsafe/libp2p-noise';
@@ -22,7 +23,6 @@ import type { Message } from '@libp2p/interface-pubsub';
 import type { PeerId } from '@libp2p/interface-peer-id';
 import { createFromJSON, createEd25519PeerId } from '@libp2p/peer-id-factory';
 import { multiaddr, Multiaddr } from '@multiformats/multiaddr';
-import { floodsub } from '@libp2p/floodsub';
 import { pubsubPeerDiscovery } from '@libp2p/pubsub-peer-discovery';
 import { PrometheusMetrics } from '@cerc-io/prometheus-metrics';
 
@@ -32,7 +32,6 @@ import {
   MIN_CONNECTIONS,
   DIAL_TIMEOUT,
   PUBSUB_DISCOVERY_INTERVAL,
-  PUBSUB_SIGNATURE_POLICY,
   RELAY_TAG,
   RELAY_REDIAL_INTERVAL,
   DEFAULT_MAX_RELAY_CONNECTIONS,
@@ -43,7 +42,7 @@ import {
   P2P_WEBRTC_STAR_ID
 } from './constants.js';
 import { PeerHearbeatChecker } from './peer-heartbeat-checker.js';
-import { debugInfoRequestHandler, dialWithRetry, getConnectionsInfo, getPseudonymForPeerId, getSelfInfo, isMultiaddrBlacklisted, wsPeerFilter } from './utils/index.js';
+import { debugInfoRequestHandler, dialWithRetry, getConnectionsInfo, getPseudonymForPeerId, getSelfInfo, initPubsub, isMultiaddrBlacklisted, wsPeerFilter } from './utils/index.js';
 import { ConnectionType, DebugPeerInfo, DebugRequest, PeerConnectionInfo, PeerSelfInfo } from './types/debug-info.js';
 
 const log = debug('laconic:peer');
@@ -66,6 +65,7 @@ export interface PeerInitConfig {
   maxConnections?: number;
   minConnections?: number;
   dialTimeout?: number;
+  pubsub?: PubsubType;
   enableDebugInfo?: boolean;
   transports?: Libp2pInit['transports'];
   listenMultiaddrs?: string[];
@@ -158,7 +158,7 @@ export class Peer {
         },
         connectionEncryption: [noise()],
         streamMuxers: [mplex()],
-        pubsub: floodsub({ globalSignaturePolicy: PUBSUB_SIGNATURE_POLICY }),
+        pubsub: initPubsub(initOptions.pubsub),
         peerDiscovery: [
           // Use pubsub based discovery; relay server acts as a peer discovery source
           pubsubPeerDiscovery({

packages/peer/src/relay.ts

@@ -2,15 +2,15 @@
 // Copyright 2022 Vulcanize, Inc.
 //
 
-import { Libp2p, createLibp2p } from '@cerc-io/libp2p';
 import debug from 'debug';
 import assert from 'assert';
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string';
 
+import { PubsubType } from '@cerc-io/util';
+import { Libp2p, createLibp2p } from '@cerc-io/libp2p';
 import { noise } from '@chainsafe/libp2p-noise';
 import { mplex } from '@libp2p/mplex';
 import { webSockets } from '@libp2p/websockets';
-import { floodsub } from '@libp2p/floodsub';
 import { pubsubPeerDiscovery } from '@libp2p/pubsub-peer-discovery';
 import type { Message } from '@libp2p/interface-pubsub';
 import type { Connection } from '@libp2p/interface-connection';
@@ -23,12 +23,11 @@ import {
   HOP_TIMEOUT,
   DEFAULT_PING_TIMEOUT,
   PUBSUB_DISCOVERY_INTERVAL,
-  PUBSUB_SIGNATURE_POLICY,
   MAX_CONCURRENT_DIALS_PER_PEER,
   DEBUG_INFO_TOPIC
 } from './constants.js';
 import { PeerHearbeatChecker } from './peer-heartbeat-checker.js';
-import { debugInfoRequestHandler, dialWithRetry, getConnectionsInfo, getPseudonymForPeerId, getSelfInfo, isMultiaddrBlacklisted } from './utils/index.js';
+import { debugInfoRequestHandler, dialWithRetry, getConnectionsInfo, getPseudonymForPeerId, getSelfInfo, initPubsub, isMultiaddrBlacklisted } from './utils/index.js';
 import { PeerIdObj } from './peer.js';
 import { SelfInfo, ConnectionInfo } from './types/debug-info.js';
 
@@ -46,6 +45,7 @@ export interface RelayNodeInitConfig {
   pingTimeout?: number;
   redialInterval: number;
   maxDialRetry: number;
+  pubsub?: PubsubType;
   enableDebugInfo?: boolean;
 }
 
@@ -75,7 +75,7 @@ export async function createRelayNode (init: RelayNodeInitConfig): Promise<Libp2
     transports: [webSockets()],
     connectionEncryption: [noise()],
     streamMuxers: [mplex()],
-    pubsub: floodsub({ globalSignaturePolicy: PUBSUB_SIGNATURE_POLICY }),
+    pubsub: initPubsub(init.pubsub),
     peerDiscovery: [
       pubsubPeerDiscovery({
         interval: PUBSUB_DISCOVERY_INTERVAL