feat: gossipsub as default pubsub (#2298)

BREAKING CHANGE: The default pubsub implementation has changed from floodsub to [gossipsub](https://github.com/ChainSafe/gossipsub-js). Additionally, to enable pubsub programmatically set `pubsub.enabled: true` instead of `EXPERIMENTAL.pubsub: true` or via the CLI pass `--enable-pubsub` instead of `--enable-pubsub-experiment` to `jsipfs daemon`.

Author: vasco-santos

README.md

@@ -324,7 +324,7 @@ Configure remote preload nodes. The remote will preload content added on this no
 
 | Type | Default |
 |------|---------|
-| object | `{ pubsub: false, ipnsPubsub: false, sharding: false }` |
+| object | `{ ipnsPubsub: false, sharding: false }` |
 
 Enable and configure experimental features.
 
@@ -495,6 +495,8 @@ You can see the bundle in action in the [custom libp2p example](examples/custom-
 - `modules` (object):
     - `transport` (Array<[libp2p.Transport](https://github.com/libp2p/interface-transport)>): An array of Libp2p transport classes/instances to use _instead_ of the defaults. See [libp2p/interface-transport](https://github.com/libp2p/interface-transport) for details.
     - `peerDiscovery` (Array<[libp2p.PeerDiscovery](https://github.com/libp2p/interface-peer-discovery)>): An array of Libp2p peer discovery classes/instances to use _instead_ of the defaults. See [libp2p/peer-discovery](https://github.com/libp2p/interface-peer-discovery) for details. If passing a class, configuration can be passed using the config section below under the key corresponding to you module's unique `tag` (a static property on the class)
+    - `dht` (object): a DHT implementation that enables PeerRouting and ContentRouting. Example [libp2p/js-libp2p-kad-dht](https://github.com/libp2p/js-libp2p-kad-dht)
+    - `pubsub` (object): a Pubsub implementation on top of [libp2p/js-libp2p-pubsub](https://github.com/libp2p/js-libp2p-pubsub)
 - `config` (object):
     - `peerDiscovery` (object):
         - `autoDial` (boolean): Dial to discovered peers when under the Connection Manager min peer count watermark. (default `true`)
@@ -506,6 +508,11 @@ You can see the bundle in action in the [custom libp2p example](examples/custom-
         - `kBucketSize` (number): bucket size (default `20`)
         - `randomWalk` (object): configuration for random walk
             - `enabled` (boolean): whether random DHT walking is enabled (default `false`)
+    - `pubsub` (object): Configuration options for Pubsub
+        - `enabled` (boolean): if pubbsub subsystem should be enabled (default: `false`)
+        - `emitSelf` (boolean): whether the node should emit to self on publish, in the event of the topic being subscribed (default: `true`)
+        - `signMessages` (boolean): if messages should be signed (default: `true`)
+        - `strictSigning` (boolean): if message signing should be required (default: `true`)
 
 ##### `options.connectionManager`
 

doc/config.md

@@ -19,6 +19,8 @@ The js-ipfs config file is a JSON document located in the root directory of the
   - [`PeerID`](#peerid)
   - [`PrivKey`](#privkey)
 - [`Keychain`](#keychain)
+- [`Pubsub`](#pubsub)
+  - [`Router`](#router)
 - [`Swarm`](#swarm)
   - [`ConnMgr`](#connmgr)
 
@@ -168,6 +170,16 @@ Default:
 
 You can check the [parameter choice for pbkdf2](https://cryptosense.com/parameter-choice-for-pbkdf2/) for more information.
 
+## `Pubsub`
+
+Options for configuring the pubsub subsystem. It is important pointing out that this is not supported in the browser. If you want to configure a different pubsub router in the browser you must configure `libp2p.modules.pubsub` options instead.
+
+### `Router`
+
+A string value for specifying which pubsub routing protocol to use. You can either use `gossipsub` in order to use the [ChainSafe/gossipsub-js](https://github.com/ChainSafe/gossipsub-js) implementation, or `floodsub` to use the [libp2p/js-libp2p-floodsub](https://github.com/libp2p/js-libp2p-floodsub) implementation. You can read more about these implementations on the [libp2p/specs/pubsub](https://github.com/libp2p/specs/tree/master/pubsub) document.
+
+Default: `gossipsub`
+
 ## `Swarm`
 
 Options for configuring the swarm.

examples/circuit-relaying/src/app.js

@@ -36,8 +36,8 @@ document.addEventListener('DOMContentLoaded', async () => {
         enabled: true // make this node a relay (HOP)
       }
     },
-    EXPERIMENTAL: {
-      pubsub: true // enable pubsub
+    pubsub: {
+      enabled: true
     },
     config: {
       Bootstrap: []

examples/custom-libp2p/index.js

@@ -96,8 +96,8 @@ const libp2pBundle = (opts) => {
           timeout: 2e3 // End the query quickly since we're running so frequently
         }
       },
-      EXPERIMENTAL: {
-        pubsub: true
+      pubsub: {
+        enabled: true
       }
     }
   })

examples/custom-libp2p/package.json

@@ -10,7 +10,7 @@
   "license": "MIT",
   "dependencies": {
     "ipfs": "file:../../",
-    "libp2p": "~0.25.0",
+    "libp2p": "~0.26.1",
     "libp2p-bootstrap": "~0.9.7",
     "libp2p-kad-dht": "~0.15.0",
     "libp2p-mdns": "~0.12.2",

examples/exchange-files-in-browser/public/app.js

@@ -44,8 +44,8 @@ let info
 async function start () {
   if (!node) {
     const options = {
-      EXPERIMENTAL: {
-        pubsub: true
+      pubsub: {
+        enabled: true
       },
       repo: 'ipfs-' + Math.random(),
       config: {

package.json

@@ -21,6 +21,7 @@
     "./src/core/runtime/dns-nodejs.js": "./src/core/runtime/dns-browser.js",
     "./src/core/runtime/fetch-nodejs.js": "./src/core/runtime/fetch-browser.js",
     "./src/core/runtime/libp2p-nodejs.js": "./src/core/runtime/libp2p-browser.js",
+    "./src/core/runtime/libp2p-pubsub-routers-nodejs.js": "./src/core/runtime/libp2p-pubsub-routers-browser.js",
     "./src/core/runtime/preload-nodejs.js": "./src/core/runtime/preload-browser.js",
     "./src/core/runtime/repo-nodejs.js": "./src/core/runtime/repo-browser.js",
     "./src/core/runtime/ipld-nodejs.js": "./src/core/runtime/ipld-browser.js",
@@ -96,7 +97,7 @@
     "ipfs-bitswap": "~0.25.1",
     "ipfs-block": "~0.8.1",
     "ipfs-block-service": "~0.15.2",
-    "ipfs-http-client": "^33.1.1",
+    "ipfs-http-client": "^34.0.0",
     "ipfs-http-response": "~0.3.1",
     "ipfs-mfs": "~0.12.0",
     "ipfs-multipart": "~0.1.1",
@@ -121,11 +122,13 @@
     "iso-url": "~0.4.6",
     "just-safe-set": "^2.1.0",
     "kind-of": "^6.0.2",
-    "libp2p": "~0.25.4",
+    "libp2p": "~0.26.1",
     "libp2p-bootstrap": "~0.9.3",
     "libp2p-crypto": "~0.16.0",
     "libp2p-delegated-content-routing": "^0.2.4",
     "libp2p-delegated-peer-routing": "^0.2.4",
+    "libp2p-floodsub": "^0.17.0",
+    "libp2p-gossipsub": "~0.0.4",
     "libp2p-kad-dht": "~0.15.3",
     "libp2p-keychain": "~0.4.2",
     "libp2p-mdns": "~0.12.0",
@@ -191,8 +194,8 @@
     "execa": "^2.0.4",
     "form-data": "^2.5.1",
     "hat": "0.0.3",
-    "interface-ipfs-core": "^0.110.0",
-    "ipfsd-ctl": "^0.44.1",
+    "interface-ipfs-core": "^0.111.0",
+    "ipfsd-ctl": "~0.45.0",
     "libp2p-websocket-star": "~0.10.2",
     "ncp": "^2.0.0",
     "p-event": "^4.1.0",

src/cli/commands/daemon.js

@@ -16,7 +16,8 @@ module.exports = {
         type: 'boolean',
         default: false
       })
-      .option('enable-pubsub-experiment', {
+      .option('enable-pubsub', {
+        alias: 'enable-pubsub-experiment',
         type: 'boolean',
         default: false
       })
@@ -53,8 +54,8 @@ module.exports = {
         offline: argv.offline,
         pass: argv.pass,
         preload: { enabled: argv.enablePreload },
+        pubsub: { enabled: argv.enablePubsub },
         EXPERIMENTAL: {
-          pubsub: argv.enablePubsubExperiment,
           ipnsPubsub: argv.enableNamesysPubsub,
           dht: argv.enableDhtExperiment,
           sharding: argv.enableShardingExperiment

src/cli/daemon.js

@@ -30,7 +30,7 @@ class Daemon {
   async start () {
     this._log('starting')
 
-    const libp2p = { modules: {} }
+    const libp2p = { modules: {}, config: {} }
 
     // Attempt to use any of the WebRTC versions available globally
     let electronWebRTC

src/cli/utils.js

@@ -50,8 +50,8 @@ exports.getIPFS = (argv, callback) => {
     init: false,
     start: false,
     pass: argv.pass,
-    EXPERIMENTAL: {
-      pubsub: true
+    pubsub: {
+      enabled: true
     }
   })
 

src/core/components/libp2p.js

@@ -2,10 +2,12 @@
 
 const get = require('dlv')
 const mergeOptions = require('merge-options')
+const errCode = require('err-code')
 const ipnsUtils = require('../ipns/routing/utils')
 const multiaddr = require('multiaddr')
 const DelegatedPeerRouter = require('libp2p-delegated-peer-routing')
 const DelegatedContentRouter = require('libp2p-delegated-content-routing')
+const PubsubRouters = require('../runtime/libp2p-pubsub-routers-nodejs')
 
 module.exports = function libp2p (self, config) {
   const options = self._options || {}
@@ -58,13 +60,24 @@ function defaultBundle ({ datastore, peerInfo, peerBook, options, config }) {
     peerRouting = [new DelegatedPeerRouter(delegatedApiOptions)]
   }
 
+  const getPubsubRouter = () => {
+    const router = get(config, 'Pubsub.Router', 'gossipsub')
+
+    if (!PubsubRouters[router]) {
+      throw errCode(new Error(`Router unavailable. Configure libp2p.modules.pubsub to use the ${router} router.`), 'ERR_NOT_SUPPORTED')
+    }
+
+    return PubsubRouters[router]
+  }
+
   const libp2pDefaults = {
     datastore,
     peerInfo,
     peerBook,
     modules: {
       contentRouting,
-      peerRouting
+      peerRouting,
+      pubsub: getPubsubRouter()
     },
     config: {
       peerDiscovery: {
@@ -105,8 +118,8 @@ function defaultBundle ({ datastore, peerInfo, peerBook, options, config }) {
           ipns: ipnsUtils.selector
         }
       },
-      EXPERIMENTAL: {
-        pubsub: get(options, 'EXPERIMENTAL.pubsub', false)
+      pubsub: {
+        enabled: get(options, 'pubsub.enabled', false)
       }
     },
     connectionManager: get(options, 'connectionManager',

src/core/components/pubsub.js

@@ -16,65 +16,56 @@ module.exports = function pubsub (self) {
         options = {}
       }
 
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!self.libp2p.pubsub) {
         return callback
           ? setImmediate(() => callback(errPubsubDisabled()))
           : Promise.reject(errPubsubDisabled())
       }
 
       if (!callback) {
-        return new Promise((resolve, reject) => {
-          self.libp2p.pubsub.subscribe(topic, options, handler, (err) => {
-            if (err) {
-              return reject(err)
-            }
-            resolve()
-          })
-        })
+        return self.libp2p.pubsub.subscribe(topic, handler, options)
       }
 
-      self.libp2p.pubsub.subscribe(topic, options, handler, callback)
+      self.libp2p.pubsub.subscribe(topic, handler, options, callback)
     },
 
     unsubscribe: (topic, handler, callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!self.libp2p.pubsub) {
         return callback
           ? setImmediate(() => callback(errPubsubDisabled()))
           : Promise.reject(errPubsubDisabled())
       }
 
-      self.libp2p.pubsub.unsubscribe(topic, handler)
-
       if (!callback) {
-        return Promise.resolve()
+        return self.libp2p.pubsub.unsubscribe(topic, handler)
       }
 
-      setImmediate(() => callback())
+      self.libp2p.pubsub.unsubscribe(topic, handler, callback)
     },
 
     publish: promisify((topic, data, callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!self.libp2p.pubsub) {
         return setImmediate(() => callback(errPubsubDisabled()))
       }
       self.libp2p.pubsub.publish(topic, data, callback)
     }),
 
     ls: promisify((callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!self.libp2p.pubsub) {
         return setImmediate(() => callback(errPubsubDisabled()))
       }
       self.libp2p.pubsub.ls(callback)
     }),
 
     peers: promisify((topic, callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!self.libp2p.pubsub) {
         return setImmediate(() => callback(errPubsubDisabled()))
       }
       self.libp2p.pubsub.peers(topic, callback)
     }),
 
     setMaxListeners (n) {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!self.libp2p.pubsub) {
         throw errPubsubDisabled()
       }
       self.libp2p.pubsub.setMaxListeners(n)

src/core/config.js

@@ -32,6 +32,9 @@ const configSchema = s({
     addresses: optional(s(['multiaddr'])),
     interval: 'number?'
   }, { enabled: true, interval: 30 * 1000 }),
+  pubsub: optional(s({
+    enabled: 'boolean?'
+  })),
   init: optional(union(['boolean', s({
     bits: 'number?',
     emptyRepo: 'boolean?',
@@ -68,6 +71,9 @@ const configSchema = s({
       }))
     })),
     Bootstrap: optional(s(['multiaddr-ipfs'])),
+    Pubsub: optional(s({
+      Router: 'string?'
+    })),
     Swarm: optional(s({
       ConnMgr: optional(s({
         LowWater: 'number?',

src/core/index.js

@@ -15,6 +15,7 @@ const multihashing = require('multihashing-async')
 const CID = require('cids')
 const debug = require('debug')
 const mergeOptions = require('merge-options')
+const get = require('dlv')
 const EventEmitter = require('events')
 
 const config = require('./config')
@@ -46,6 +47,9 @@ class IPFS extends EventEmitter {
       init: true,
       start: true,
       EXPERIMENTAL: {},
+      pubsub: {
+        enabled: false
+      },
       preload: {
         enabled: true,
         addresses: [
@@ -131,13 +135,14 @@ class IPFS extends EventEmitter {
     this.stats = components.stats(this)
     this.resolve = components.resolve(this)
 
-    if (this._options.EXPERIMENTAL.pubsub) {
-      this.log('EXPERIMENTAL pubsub is enabled')
+    if (this._options.pubsub.enabled) {
+      this.log('pubsub is enabled')
     }
     if (this._options.EXPERIMENTAL.ipnsPubsub) {
-      if (!this._options.EXPERIMENTAL.pubsub) {
-        this.log('EXPERIMENTAL pubsub is enabled to use IPNS pubsub')
-        this._options.EXPERIMENTAL.pubsub = true
+      // if (!this._options.pubsub.enabled) {
+      if (!get(this._options, 'pubsub.enabled', false)) {
+        this.log('pubsub is enabled to use EXPERIMENTAL IPNS pubsub')
+        this._options.pubsub.enabled = true
       }
 
       this.log('EXPERIMENTAL IPNS pubsub is enabled')

src/core/runtime/config-nodejs.js

@@ -40,6 +40,9 @@ module.exports = () => ({
     '/dns4/node0.preload.ipfs.io/tcp/443/wss/ipfs/QmZMxNdpMkewiVZLMRxaNxUeZpDUb34pWjZ1kZvsd16Zic',
     '/dns4/node1.preload.ipfs.io/tcp/443/wss/ipfs/Qmbut9Ywz9YEDrz8ySBSgWyJk41Uvm2QJPhwDJzJyGFsD6'
   ],
+  Pubsub: {
+    Router: 'gossipsub'
+  },
   Swarm: {
     ConnMgr: {
       LowWater: 200,