P2P

package.json

@@ -14,26 +14,35 @@
         "build": "tsc -p ."
     },
     "dependencies": {
+        "@chainsafe/libp2p-noise": "^5.0.3",
+        "@fluxprotocol/oracle-vm": "^2.0.0",
         "big.js": "^6.1.1",
         "bn.js": "^5.2.0",
         "dotenv": "^15.0.0",
         "ethers": "^5.5.4",
+        "libp2p": "^0.36.2",
+        "libp2p-mplex": "^0.10.7",
+        "libp2p-tcp": "^0.17.2",
         "express": "^4.17.3",
         "jsonpath-plus": "^6.0.1",
         "lodash.topath": "^4.5.2",
+        "multiaddr": "^10.0.1",
         "near-api-js": "^0.44.2",
+        "uint8arrays": "^3.0.0",
         "web3": "^1.7.0",
         "winston-daily-rotate-file": "^4.6.0",
         "winston-transport-sentry-node": "^2.3.0"
     },
     "devDependencies": {
         "@types/big.js": "^6.1.2",
+        "@types/bl": "^5.0.2",
+        "@types/debug": "^4.1.7",
         "@types/express": "^4.17.13",
         "@types/jest": "^27.4.0",
         "@types/lodash.topath": "^4.5.6",
-        "@types/node": "^17.0.14",
+        "@types/node": "^17.0.21",
         "jest": "^27.4.7",
         "ts-jest": "^27.1.3",
-        "typescript": "^4.5.5"
+        "typescript": "^4.6.2"
     }
 }

src/main.ts

@@ -1,6 +1,7 @@
 import logger from './services/LoggerService';
 import { parseAppConfig } from './services/AppConfigService';
 import { PROJECT_NAME, PROJECT_VERSION } from './config';
+import { aggregate, start_p2p } from './p2p/aggregator';
 
 async function main() {
     logger.info(`🧙 Starting ${PROJECT_NAME} v${PROJECT_VERSION}`);
@@ -20,6 +21,9 @@ async function main() {
 
         await appConfig.healthcheck.start();
 
+        const p2p = await start_p2p(appConfig.p2p_node, appConfig.peers_file);
+        await aggregate(p2p, 'foo');
+
         logger.info(`🚀 Booted`);
     } catch (error) {
         logger.error(`${error}`);

src/models/AppConfig.ts

@@ -1,3 +1,5 @@
+import { CreateOptions } from "libp2p";
+
 import { Healthcheck } from '../healthCheck/Healthcheck';
 import { IModule } from './IModule';
 import { ModuleConfig } from './IModule';
@@ -9,9 +11,13 @@ export interface AppConfig {
     networks: INetwork[];
     modules: IModule[];
     jobs: IJob[];
+    p2p_node: CreateOptions;
+    peers_file: string;
 }
 
 export interface UnparsedAppConfig {
     networks?: Partial<NetworkConfig>[];
     modules?: Partial<ModuleConfig>[];
+    p2p_node: CreateOptions,
+    peers_file: string,
 }

src/p2p/aggregator.ts

@@ -0,0 +1,75 @@
+import { CreateOptions } from "libp2p";
+import Communicator from './communication';
+import TCP from "libp2p-tcp";
+const Mplex = require("libp2p-mplex"); // no ts support yet :/
+import { NOISE } from "@chainsafe/libp2p-noise";
+import BufferList from "bl/BufferList";
+import { fromString } from 'uint8arrays/from-string';
+
+export async function start_p2p(addresses: CreateOptions, peers_file: string): Promise<Communicator> {
+	const p2p = new Communicator(
+		{
+			...addresses,
+			modules: {
+				transport: [TCP],
+				streamMuxer: [Mplex],
+				connEncryption: [NOISE],
+			}
+		},
+		peers_file
+	);
+	await p2p.init();
+
+	return p2p;
+}
+
+export async function elect_leader(p2p: Communicator): Promise<string> {
+	// method should maybe be moved to Communicator?
+	let peers = Array.from(p2p._peers);
+	peers.push(p2p._node_addr);
+	peers = peers.sort();
+	return peers[0];
+}
+
+export function median(list: number[]): number {
+	list = list.sort();
+
+	let mid_point = list.length / 2 ;
+
+	if (mid_point % 2) {
+		return (list[mid_point - 1] + list[mid_point]) / 2;
+	} else {
+		return list[mid_point];
+	}
+	return 0;
+}
+
+export async function aggregate(p2p: Communicator, data_to_send: string) {
+	await p2p.start();
+	let received: number[] = [];
+	let med: number = 0;
+	let leader: string = '';
+
+	p2p.handle_incoming('/elected/leader', async (source: AsyncIterable<Uint8Array | BufferList>) => {
+		for await (const msg of source) {
+			let elected = msg.toString();
+			// check if this node is the leader.
+			if (elected === leader) {
+				// send data to contract.
+			}
+		}
+	});
+
+	p2p.handle_incoming('/send/data', async (source: AsyncIterable<Uint8Array | BufferList>) => {
+		for await (const msg of source) {
+			received.push(parseInt(msg.toString()));
+			if (received.length = p2p._peers.size) {
+				med = median(received);
+				leader = await elect_leader(p2p);
+				p2p.send('/elected/leader', [fromString(leader)]);
+			}
+		}
+	});
+
+	p2p.send('/send/data', [fromString(data_to_send)]);
+}

src/p2p/communication.ts

@@ -0,0 +1,209 @@
+import Libp2p, { Connection, create, CreateOptions, Libp2pOptions } from "libp2p";
+import TCP from "libp2p-tcp";
+const Mplex = require("libp2p-mplex"); // no ts support yet :/
+import { NOISE } from "@chainsafe/libp2p-noise";
+import { Multiaddr } from "multiaddr";
+import PeerId from "peer-id";
+import BufferList from "bl/BufferList";
+import { existsSync, readFileSync, writeFileSync } from "fs";
+
+import logger from "../services/LoggerService";
+
+async function* createAsyncIterable(syncIterable: Uint8Array[]) {
+	for (const elem of syncIterable) {
+		yield elem;
+	}
+}
+
+async function attempt<R>(comm: Communicator, def: R, attempt: (node: Libp2p) => Promise<R>): Promise<R> {
+	if (comm._node === undefined) {
+		logger.error('The communication node has not been started.');
+		return def;
+	} else {
+		return attempt(comm._node);
+	}
+}
+
+export default class Communicator {
+	_connections: Set<Connection> = new Set();
+	_options: Libp2pOptions & CreateOptions;
+	_node?: Libp2p;
+	_node_addr: string;
+	_peers: Set<string>;
+	_peers_file: string;
+	_retry: Set<string> = new Set();
+
+	constructor(config?: Libp2pOptions & CreateOptions, peers_file?: string) {
+		if (config === undefined) {
+			this._options = {
+				addresses: {
+					listen: ['/ip4/0.0.0.0/tcp/0']
+				},
+				modules: {
+					transport: [TCP],
+					streamMuxer: [Mplex],
+					connEncryption: [NOISE],
+				}
+			};
+		} else {
+			this._options = config;
+		}
+
+		if (peers_file === undefined) {
+			this._peers_file = 'peers.json';
+			this._peers = new Set();
+		} else {
+			this._peers_file = peers_file;
+			this._peers = this.load_peers();
+		}
+
+		this._node_addr = '';
+	}
+
+	async init(): Promise<void> {
+		this._node = await create(this._options);
+	}
+
+	async retry(): Promise<void> {
+		attempt(this, null, async () => {
+			for (const peer of this._retry) {
+				if (await this.connect(new Multiaddr(peer))) {
+					this._retry.delete(peer);
+				}
+			}
+		});
+	}
+
+	async start(): Promise<[Multiaddr, PeerId] | void> {
+		return attempt(this, undefined, async (node: Libp2p) => {
+			await node.start();
+
+			for (const peer of this._peers) {
+				if (!await this.connect(new Multiaddr(peer))) {
+					this._retry.add(peer);
+				}
+			}
+
+			this._node_addr = node.multiaddrs[0].toString();
+
+			return [
+				node.multiaddrs[0],
+				node.peerId,
+			];
+		})
+	}
+
+	async stop(): Promise<void> {
+		attempt(this, false, async (node: Libp2p) => {
+			try {
+				for (const peer of this._peers) {
+					node.hangUp(peer);
+				}
+
+				this.save_peers();
+
+				await node.stop();
+
+				return true;
+			} catch (error) {
+				logger.error(`Node failed to stop with error '${error}'.`);
+				return false;
+			}
+		});
+	}
+
+	async connect(ma: Multiaddr): Promise<boolean> {
+		return attempt(this, false, async (node: Libp2p) => {
+			try {
+				this._peers.add(ma.toString());
+				const conn = await node.dial(ma);
+				if (conn !== undefined) {
+					this._connections.add(conn);
+					return true;
+				}
+
+				return false;
+			} catch (error) {
+				logger.error(`Node failed to connect to ${ma} with error '${error}'.`);
+				return false;
+			}
+		});
+
+	}
+
+	async connect_from_details(ip: string, address: string, transport: string, port: string, peerID: string): Promise<boolean> {
+		return attempt(this, false, async (node: Libp2p) => {
+			const mas = `/${ip}/${address}/${transport}/${port}/p2p/${peerID}`;
+			const ma = new Multiaddr(`/${ip}/${address}/${transport}/${port}/p2p/${peerID}`);
+
+			try {
+				if (!this._peers.has(mas)) {
+					this._peers.add(mas);
+				}
+
+				const conn = await node.dial(ma);
+				if (conn !== undefined) {
+					this._connections.add(conn);
+					return true;
+				}
+
+				return false;
+			} catch (error) {
+				this._peers.delete(mas);
+				logger.error(`Node failed to connect to ${ma} with error '${error}'.`);
+				return false;
+			}
+		});
+	}
+
+	async handle_incoming(protocol: string, callback: (source: AsyncIterable<Uint8Array | BufferList>) => Promise<void>): Promise<void> {
+		await attempt(this, null, async (node: Libp2p) => {
+			node.handle(protocol, async ({ stream }) => {
+				await callback(stream.source);
+			});
+		});
+	}
+
+	async send(protocol: string, data: Uint8Array[]): Promise<void> {
+		attempt(this, null, async () => {
+			for (const connection of this._connections) {
+				const { stream } = await connection.newStream(protocol);
+				await stream.sink(createAsyncIterable(data));
+			}
+		});
+	}
+
+	save_peers(): void {
+		let peers = Array.from(this._peers).sort();
+
+		const json = JSON.stringify(peers, null, 4);
+		try {
+			writeFileSync(this._peers_file, json);
+		} catch (error) {
+			logger.error(`Node failed to save peers with error '${error}'.`);
+		}
+
+	}
+
+	load_peers(): Set<string> {
+		if (!existsSync(this._peers_file)) {
+			return new Set();
+		}
+
+		try {
+			const json = readFileSync(this._peers_file, 'utf-8');
+			const peers_list: string[] = JSON.parse(json);
+
+			let peers: Set<string> = new Set();
+			for (const peer of peers_list) {
+				peers.add(peer);
+			}
+
+			return peers;
+		} catch (error) {
+			logger.error(`Node failed to load peers with error '${error}'.`);
+			return new Set();
+		}
+
+	}
+}

src/services/AppConfigService.ts

@@ -22,6 +22,8 @@ export async function parseAppConfig(): Promise<AppConfig> {
         networks: [],
         modules: [],
         jobs: [],
+        p2p_node: config.p2p_node,
+        peers_file: config.peers_file,
     };
 
     if (!config.networks || !Array.isArray(config.networks)) throw new Error(`"networks" is required and must be an array`);