Need help setting up Kad DHT using nodejs

[muzahidul-islam-sbu]

I'm trying to set up a Kad DHT for local peers (will move to remote peers later), but am having trouble with the DHT replicating the keys and values.

The goal is that I want to start a peer node in one process, and another peer node in a different process. Then I want to put a key value pair in node 1 (for example key = 'hello', value = 'world), and I then want to retrieve this value in node 2.

Currently, I can put and get to the DHT fine in a single node, and the nodes connect to each other using mdns, but I can't copy the DHT key/values between two nodes. Here is my code:

import { noise } from '@chainsafe/libp2p-noise'
import { yamux } from '@chainsafe/libp2p-yamux'
import { bootstrap } from '@libp2p/bootstrap'
import { identify } from '@libp2p/identify'
import { kadDHT } from '@libp2p/kad-dht'
import { mplex } from '@libp2p/mplex'
import { tcp } from '@libp2p/tcp'
import { createLibp2p } from 'libp2p'
import { mdns } from '@libp2p/mdns'
import readline from 'readline'

const node = await createLibp2p({
    addresses: {
        listen: ['/ip4/0.0.0.0/tcp/0']
    },
    transports: [tcp()],
    streamMuxers: [yamux(), mplex()],
    connectionEncryption: [noise()],
    peerDiscovery: [
        // bootstrap({
        //   list: bootstrappers
        // }),
        mdns()
    ],
    services: {
        kadDHT: kadDHT({
            kBucketSize: 20,
            enabled: true,
            randomWalk: {
                enabled: true,            // Allows to disable discovery (enabled by default)
                interval: 300e3,
                timeout: 10e3
            }
        }),
        identify: identify()
    }
})

node.addEventListener('peer:connect', (evt) => {
    const peerId = evt.detail
    console.log('Connection established to:', peerId.toString()) // Emitted when a peer has been found
})

node.addEventListener('peer:discovery', (evt) => {
    const peerInfo = evt.detail

    console.log('Discovered:', peerInfo.id.toString())
})


async function putKeyValue(node, key, value) {
    const keyUint8Array = new Uint8Array(Buffer.from(key));
    const valueUint8Array = new Uint8Array(Buffer.from(value));
    try {
        let retrievedValue = node.services.kadDHT.put(keyUint8Array, valueUint8Array);
        for await (const queryEvent of retrievedValue) {
            console.log('dht:put', queryEvent)
        }
    } catch (err) {
        throw err;
    }
}

async function getValue(node, key) {
    const keyUint8Array = new Uint8Array(Buffer.from(key));
    for await (const queryEvent of node.services.kadDHT.getClosestPeers(keyUint8Array)) {
        console.log('dht:peerRouting', queryEvent)
    }
    for await (const queryEvent of node.services.kadDHT.get(keyUint8Array)) {
        console.log('dht:get', queryEvent)
    }
}

async function cli(node) {
    const rl = readline.createInterface({
        input: process.stdin,
        output: process.stdout
    });

    rl.setPrompt('> ');

    rl.prompt();

    rl.on('line', async (input) => {
        const [command, ...args] = input.trim().split(' ');
        switch (command) {
            case 'put':
                if (args.length !== 2) {
                    console.error('Usage: put <key> <value>');
                    break;
                }
                await putKeyValue(node, args[0], args[1]);
                await node.services.kadDHT.refreshRoutingTable()
                break;
            case 'get':
                if (args.length !== 1) {
                    console.error('Usage: get <key>');
                    break;
                }
                await node.services.kadDHT.refreshRoutingTable();
                await getValue(node, args[0]);
                break;
            case 'get_mode':
                if (args.length !== 0) {
                    console.error('Usage: get_mode');
                    break;
                }
                console.log(node.services.kadDHT.getMode());
                break;
            case 'set_mode':
                if (args.length !== 1) {
                    console.error('Usage: set_mode <client/server>');
                    break;
                }
                await node.services.kadDHT.setMode(args[0]);
                break;
            default:
                console.error('Unknown command:', command);
                break;
        }
        rl.prompt();
    });

    process.on('SIGTERM', () => {
        rl.close();
        node.stop();
    });

    process.on('SIGINT', () => {
        rl.close();
        node.stop();
    });
    rl.on('close', () => {
        console.log('Exiting...');
        process.exit(0);
    });
}

console.log(node.peerId)
node.services.kadDHT.contentFetching.log.enabled = true;
cli(node)

[achingbrain]

Two nodes isn't really enough to make a DHT, so to make the example work you have to take some shortcuts.

Ordinarily the steps you take when connecting to a DHT are:

1. Connect to peers we know are already in the DHT (e.g. bootstrap peers)
2. Run a query for your own PeerId to discover peers close to you in KAD-space
3. Dial those peers, make & respond to queries

Next is configuring the DHT properly:

import { kadDHT, removePublicAddressesMapper } from '@libp2p/kad-dht'

kadDHT: kadDHT({
  // must be in server mode to accept records from peers
  // - normally this is auto-detected based on having public addresses but here
  // we are local-only so just enable server mode
  clientMode: false,
  // use a custom protocol to ensure our local nodes are actually doing the work
  protocol: '/ipfs/lan/kad/1.0.0',
  // remove any public addresses from peer records because we are LAN-only
  peerInfoMapper: removePublicAddressesMapper,
  // ensure we can publish/resolve records with a custom prefix
  // if validators/selectors for the key prefix are missing the records will be ignored
  validators: {
    'my-prefix': async (key, value) => {
      // do nothing, it's valid
    }
  },
  // select a record when there are multiples resolved for the same key
  selectors: {
    'my-prefix': (key, records) => {
      // take the first record
      return 0
    }
  }
})

Finally put it all together. Because we aren't bootstrapping to any peers to start filling out our routing table, we'll just dial whatever peers are discovered by mDNS.

package.json

{
  "type": "module",
  "dependencies": {
    "@chainsafe/libp2p-noise": "^16.0.0",
    "@chainsafe/libp2p-yamux": "^7.0.1",
    "@libp2p/identify": "^3.0.13",
    "@libp2p/kad-dht": "^14.1.4",
    "@libp2p/mdns": "^11.0.14",
    "@libp2p/ping": "^2.0.13",
    "@libp2p/tcp": "^10.0.14"
  }
}

publisher.js:

import { noise } from '@chainsafe/libp2p-noise'
import { yamux } from '@chainsafe/libp2p-yamux'
import { identify, identifyPush } from '@libp2p/identify'
import { kadDHT, removePublicAddressesMapper } from '@libp2p/kad-dht'
import { tcp } from '@libp2p/tcp'
import { createLibp2p } from 'libp2p'
import { mdns } from '@libp2p/mdns'
import { ping } from '@libp2p/ping'

const prefix = 'my-prefix'

const node = await createLibp2p({
    addresses: {
        listen: ['/ip4/0.0.0.0/tcp/0']
    },
    transports: [tcp()],
    streamMuxers: [yamux()],
    connectionEncrypters: [noise()],
    peerDiscovery: [
      mdns()
    ],
    services: {
        kadDHT: kadDHT({
          clientMode: false,
          protocol: '/ipfs/lan/kad/1.0.0',
          peerInfoMapper: removePublicAddressesMapper,
          validators: {
            [prefix]: async (key, value) => {
              // do nothing, it's valid
            }
          },
          selectors: {
            [prefix]: (key, records) => {
              // take the first record
              return 0
            }
          }
        }),
        identify: identify(),
        identifyPush: identifyPush(),
        ping: ping()
    }
})

node.addEventListener('peer:connect', (evt) => {
    const peerId = evt.detail
    console.log('Connection established to:', peerId.toString()) // Emitted when a peer has been found
})

node.addEventListener('peer:discovery', async (evt) => {
    const peerInfo = evt.detail
    console.log('Discovered:', peerInfo.id.toString())

    // we aren't bootstrapping to an existing network so just dial any peers
    // that are discovered
    try {
      await node.dial(peerInfo.multiaddrs)
    } catch {}

    const key = `/${prefix}/hello`
    const value = 'world'

    const keyUint8Array = Buffer.from(key)
    const valueUint8Array = Buffer.from(value)

    console.info('putting value')

    await node.contentRouting.put(keyUint8Array, valueUint8Array)

    console.info('put value')
})

console.info('Publisher:', node.peerId.toString())

resolver.js

import { noise } from '@chainsafe/libp2p-noise'
import { yamux } from '@chainsafe/libp2p-yamux'
import { identify, identifyPush } from '@libp2p/identify'
import { kadDHT, removePublicAddressesMapper } from '@libp2p/kad-dht'
import { tcp } from '@libp2p/tcp'
import { createLibp2p } from 'libp2p'
import { mdns } from '@libp2p/mdns'
import { ping } from '@libp2p/ping'

const prefix = 'my-prefix'

const node = await createLibp2p({
    addresses: {
        listen: ['/ip4/0.0.0.0/tcp/0']
    },
    transports: [tcp()],
    streamMuxers: [yamux()],
    connectionEncrypters: [noise()],
    peerDiscovery: [
      mdns()
    ],
    services: {
        kadDHT: kadDHT({
          clientMode: false,
          protocol: '/ipfs/lan/kad/1.0.0',
          peerInfoMapper: removePublicAddressesMapper,
          validators: {
            [prefix]: async (key, value) => {
              // do nothing, it's valid
            }
          },
          selectors: {
            [prefix]: (key, records) => {
              // take the first record
              return 0
            }
          }
        }),
        identify: identify(),
        identifyPush: identifyPush(),
        ping: ping()
    }
})


node.addEventListener('peer:connect', (evt) => {
    const peerId = evt.detail
    console.log('Connection established to:', peerId.toString()) // Emitted when a peer has been found
})

node.addEventListener('peer:discovery', async (evt) => {
    const peerInfo = evt.detail
    console.log('Discovered:', peerInfo.id.toString())

    // we aren't bootstrapping to an existing network so just dial any peers
    // that are discovered
    try {
      await node.dial(peerInfo.multiaddrs)
    } catch {}

    const key = `/${prefix}/hello`
    const keyUint8Array = Buffer.from(key)

    console.info('get value')

    const value = await node.contentRouting.get(keyUint8Array)

    console.info(`got value "${value}"`)
})

console.info('Resolver:', node.peerId.toString())

Start the publisher, then start the resolver.

% node ./publisher.js 
Publisher: 12D3KooWGT2Du2GGQ1vtescifyqzg5RxZWiuTEDxcxqK3jdxMCJ5
Discovered: 12D3KooWGSWBiG1am6iwChjoCiK8C8apEe9jjsVoK1aLZU8W3d7W
Connection established to: 12D3KooWGSWBiG1am6iwChjoCiK8C8apEe9jjsVoK1aLZU8W3d7W
putting value
put value

% node resolver.js 
Resolver: 12D3KooWGSWBiG1am6iwChjoCiK8C8apEe9jjsVoK1aLZU8W3d7W
Discovered: 12D3KooWGT2Du2GGQ1vtescifyqzg5RxZWiuTEDxcxqK3jdxMCJ5
Connection established to: 12D3KooWGT2Du2GGQ1vtescifyqzg5RxZWiuTEDxcxqK3jdxMCJ5
get value
got value "world"