Use consensus mechanism for sending L2 txs

environments/local.toml

@@ -51,22 +51,28 @@
         privateKey = ''
         contractAddress = ''
 
-      [server.p2p.nitro]
-        store = ''
-        privateKey = ''
-        chainPrivateKey = ''
-
-        [server.p2p.nitro.payments]
-          ratesFile = ''
-          requestTimeoutInSecs = 10
-
-          [server.p2p.nitro.payments.cache]
-            maxAccounts = 1000
-            accountTTLInSecs = 1800
-            maxVouchersPerAccount = 1000
-            voucherTTLInSecs = 300
-            maxPaymentChannels = 10000
-            paymentChannelTTLInSecs = 1800
+    [server.p2p.nitro]
+      store = ''
+      privateKey = ''
+      chainPrivateKey = ''
+
+      [server.p2p.nitro.payments]
+        ratesFile = ''
+        requestTimeoutInSecs = 10
+
+        [server.p2p.nitro.payments.cache]
+          maxAccounts = 1000
+          accountTTLInSecs = 1800
+          maxVouchersPerAccount = 1000
+          voucherTTLInSecs = 300
+          maxPaymentChannels = 10000
+          paymentChannelTTLInSecs = 1800
+
+    [server.p2p.consensus]
+      enabled = true
+      publicKey = ''
+      privateKey = ''
+      watcherPartyFile = ''
 
 [metrics]
   host = "127.0.0.1"

package.json

@@ -25,7 +25,8 @@
     "import-state:dev": "DEBUG=vulcanize:* ts-node src/cli/import-state.ts",
     "inspect-cid": "DEBUG=vulcanize:* ts-node src/cli/inspect-cid.ts",
     "index-block": "DEBUG=vulcanize:* ts-node src/cli/index-block.ts",
-    "peer": "DEBUG='vulcanize:*, laconic:*' node --enable-source-maps dist/cli/peer.js"
+    "peer": "DEBUG='vulcanize:*, laconic:*' node --enable-source-maps dist/cli/peer.js",
+    "gen-consensus-keys": "DEBUG=laconic:* node --enable-source-maps dist/cli/gen-consensus-keys.js"
   },
   "repository": {
     "type": "git",
@@ -39,13 +40,12 @@
   "homepage": "https://github.com/cerc-io/mobymask-v2-watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.3.19",
-    "@cerc-io/peer": "^0.2.56",
-    "@cerc-io/cli": "^0.2.56",
-    "@cerc-io/ipld-eth-client": "^0.2.56",
-    "@cerc-io/nitro-node": "^0.1.9",
-    "@cerc-io/nitro-util": "^0.1.9",
-    "@cerc-io/solidity-mapper": "^0.2.56",
-    "@cerc-io/util": "^0.2.56",
+    "@cerc-io/peer": "^0.2.57",
+    "@cerc-io/cli": "^0.2.57",
+    "@cerc-io/ipld-eth-client": "^0.2.57",
+    "@cerc-io/nitro-node": "^0.1.10",
+    "@cerc-io/solidity-mapper": "^0.2.57",
+    "@cerc-io/util": "^0.2.57",
     "@ethersproject/providers": "^5.4.4",
     "apollo-type-bigint": "^0.1.3",
     "debug": "^4.3.1",

src/cli/gen-consensus-keys.ts

@@ -0,0 +1,56 @@
+//
+// Copyright 2023 Vulcanize, Inc.
+//
+
+import crypto from 'crypto';
+import debug from 'debug';
+import fs from 'fs';
+import path from 'path';
+import { hideBin } from 'yargs/helpers';
+import yargs from 'yargs';
+
+const log = debug('laconic:gen-consensus-keys');
+
+interface Arguments {
+  file: string;
+}
+
+async function main (): Promise<void> {
+  const node = crypto.createECDH('secp256k1');
+  node.generateKeys('hex');
+  const obj = {
+    publicKey: node.getPublicKey('hex', 'compressed'),
+    privateKey: node.getPrivateKey('hex')
+  };
+
+  const argv: Arguments = _getArgv();
+  if (argv.file) {
+    const exportFilePath = path.resolve(argv.file);
+    const exportFileDir = path.dirname(exportFilePath);
+
+    if (!fs.existsSync(exportFileDir)) {
+      fs.mkdirSync(exportFileDir, { recursive: true });
+    }
+
+    fs.writeFileSync(exportFilePath, JSON.stringify(obj, null, 2));
+    log(`Key pair exported to file ${exportFilePath}`);
+  } else {
+    log(obj);
+  }
+}
+
+function _getArgv (): any {
+  return yargs(hideBin(process.argv)).parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    file: {
+      type: 'string',
+      alias: 'f',
+      describe: 'Peer Id export file path (json)'
+    }
+  }).argv;
+}
+
+main().catch(err => {
+  log(err);
+});

src/libp2p-utils.ts

@@ -4,9 +4,10 @@
 
 import debug from 'debug';
 import { ethers, Signer } from 'ethers';
+import assert from 'assert';
 
 import { TransactionReceipt, TransactionResponse } from '@ethersproject/providers';
-import { PaymentsManager } from '@cerc-io/util';
+import { PaymentsManager, Consensus } from '@cerc-io/util';
 import { utils as nitroUtils } from '@cerc-io/nitro-node';
 
 import { abi as PhisherRegistryABI } from './artifacts/PhisherRegistry.json';
@@ -28,62 +29,74 @@ export function createMessageToL2Handler (
     contractAddress: string,
     gasLimit?: number
   },
-  paymentsManager: PaymentsManager
+  paymentsManager: PaymentsManager,
+  consensus?: Consensus
 ) {
-  return (peerId: string, data: any): void => {
+  return async (peerId: string, data: any): Promise<void> => {
     log(`[${getCurrentTime()}] Received a message on mobymask P2P network from peer:`, peerId);
-    sendMessageToL2(signer, { contractAddress, gasLimit }, data, paymentsManager);
-  };
-}
 
-export async function sendMessageToL2 (
-  signer: Signer,
-  { contractAddress, gasLimit = DEFAULT_GAS_LIMIT }: {
-    contractAddress: string,
-    gasLimit?: number
-  },
-  data: any,
-  paymentsManager: PaymentsManager
-): Promise<void> {
-  // Message envelope includes the payload as well as a payment (to, vhash, vsig)
-  const {
-    payload: { kind, message },
-    payment
-  } = data;
+    // Message envelope includes the payload as well as a payment (vhash, vsig)
+    const { payload, payment } = data;
 
-  if (!paymentsManager.clientAddress) {
-    log('Ignoring payload, payments manager not subscribed to vouchers yet');
-    return;
-  }
+    // TODO: Check payment status before sending tx to l2
+    await handlePayment(paymentsManager, payment, payload.kind);
 
-  // Ignore if the payload is not meant for us
-  if (payment.to === paymentsManager.clientAddress) {
-    log('Ignoring payload not meant for this client');
-    return;
-  }
+    sendMessageToL2(signer, { contractAddress, gasLimit }, payload, consensus);
+  };
+}
+
+export async function handlePayment (
+  paymentsManager: PaymentsManager,
+  payment: { vhash: string, vsig: string },
+  requestKind: string
+): Promise<boolean> {
+  assert(paymentsManager.clientAddress);
 
   // Retrieve sender address
   const signerAddress = nitroUtils.getSignerAddress(payment.vhash, payment.vsig);
 
   // Get the configured mutation cost
   const mutationRates = paymentsManager.mutationRates;
-  if (kind in mutationRates) {
-    const configuredMutationCost = BigInt(mutationRates[kind as string]);
+  if (requestKind in mutationRates) {
+    const configuredMutationCost = BigInt(mutationRates[requestKind as string]);
 
     // Check for payment voucher received from the sender Nitro account
     const [paymentVoucherReceived, paymentError] = await paymentsManager.authenticatePayment(payment.vhash, signerAddress, configuredMutationCost);
 
     if (!paymentVoucherReceived) {
-      log(`Rejecting a mutation request from ${signerAddress}: ${paymentError}`);
-      return;
+      // log(`Rejecting a mutation request from ${signerAddress}: ${paymentError}`);
+      log(paymentError);
+      return false;
     }
 
-    log(`Serving a paid mutation request for ${signerAddress}`);
+    // log(`Serving a paid mutation request for ${signerAddress}`);
+    log(`Payment received for a mutation request from ${signerAddress}`);
+    return true;
   } else {
     // Serve a mutation request for free if rate is not configured
-    log(`Mutation rate not configured for "${kind}", serving a free mutation request to ${signerAddress}`);
+    // log(`Mutation rate not configured for "${requestKind}", serving a free mutation request to ${signerAddress}`);
+    log(`Mutation rate not configured for "${requestKind}"`);
+    return true;
+  }
+}
+
+export async function sendMessageToL2 (
+  signer: Signer,
+  { contractAddress, gasLimit = DEFAULT_GAS_LIMIT }: {
+    contractAddress: string,
+    gasLimit?: number
+  },
+  data: any,
+  consensus?: Consensus
+): Promise<void> {
+  // If consensus is setup, send tx to L2 only if we are the leader
+  if (consensus && !consensus.isLeader()) {
+    log('Not a leader, skipped sending L2 tx');
+    return;
   }
 
+  const { kind, message } = data;
+
   const contract = new ethers.Contract(contractAddress, PhisherRegistryABI, signer);
   let receipt: TransactionReceipt | undefined;
 

src/server.ts

@@ -11,15 +11,6 @@ import { ethers } from 'ethers';
 
 import { ServerCmd } from '@cerc-io/cli';
 
-import {
-  P2PMessageService,
-  Node,
-  EthChainService,
-  PermissivePolicy,
-  DurableStore,
-  utils
-} from '@cerc-io/nitro-node';
-
 import { createResolvers } from './resolvers';
 import { Indexer } from './indexer';
 import { Database } from './database';
@@ -29,8 +20,7 @@ import {
   virtualPaymentAppAddress,
   consensusAppAddress
 } from './nitro-addresses.json';
-import { Config, PaymentsManager, getConfig } from '@cerc-io/util';
-import { Peer } from '@cerc-io/peer';
+import { PaymentsManager, getConfig } from '@cerc-io/util';
 
 import { RatesConfig } from './config';
 
@@ -41,74 +31,50 @@ export const main = async (): Promise<any> => {
   await serverCmd.init(Database);
   await serverCmd.initIndexer(Indexer);
 
-  let nitroPaymentsManager: PaymentsManager | undefined;
-  let p2pMessageHandler = parseLibp2pMessage;
+  // Initialize / start the p2p nodes
+  const { peer } = await serverCmd.initP2P();
+
+  // Initialize / start the Nitro node
+  const nitro = await serverCmd.initNitro({
+    nitroAdjudicatorAddress,
+    consensusAppAddress,
+    virtualPaymentAppAddress
+  });
 
-  const { enablePeer, peer: { enableL2Txs, l2TxsConfig }, nitro: { payments } } = serverCmd.config.server.p2p;
+  // Initialize / start the consensus engine
+  const consensus = await serverCmd.initConsensus();
+
+  let nitroPaymentsManager: PaymentsManager | undefined;
+  const { enablePeer, peer: { enableL2Txs, l2TxsConfig, pubSubTopic }, nitro: { payments } } = serverCmd.config.server.p2p;
 
   if (enablePeer) {
+    assert(peer);
+    assert(nitro);
+
+    // Setup the payments manager if peer is enabled
     const ratesConfig: RatesConfig = await getConfig(payments.ratesFile);
     nitroPaymentsManager = new PaymentsManager(payments, ratesConfig);
 
+    // Start subscription for payment vouchers received by the client
+    nitroPaymentsManager.subscribeToVouchers(nitro);
+
+    // Register the pubsub topic handler
+    let p2pMessageHandler = parseLibp2pMessage;
+
+    // Send L2 txs for messages if enabled
     if (enableL2Txs) {
       assert(l2TxsConfig);
       const wallet = new ethers.Wallet(l2TxsConfig.privateKey, serverCmd.ethProvider);
-      p2pMessageHandler = createMessageToL2Handler(wallet, l2TxsConfig, nitroPaymentsManager);
+      p2pMessageHandler = createMessageToL2Handler(wallet, l2TxsConfig, nitroPaymentsManager, consensus);
     }
-  }
 
-  const typeDefs = fs.readFileSync(path.join(__dirname, 'schema.gql')).toString();
-  await serverCmd.exec(createResolvers, typeDefs, p2pMessageHandler, nitroPaymentsManager);
-
-  if (enablePeer) {
-    assert(serverCmd.peer);
-    assert(nitroPaymentsManager);
-
-    const client = await setupNitro(serverCmd.config, serverCmd.peer);
-    log(`Nitro client started with address: ${client.address}`);
-
-    // Start subscription for payment vouchers received by the client
-    nitroPaymentsManager.subscribeToVouchers(client);
+    peer.subscribeTopic(pubSubTopic, (peerId, data) => {
+      p2pMessageHandler(peerId.toString(), data);
+    });
   }
-};
-
-const setupNitro = async (config: Config, peer: Peer): Promise<Node> => {
-  // TODO: Use Nitro class from ts-nitro
-  const {
-    server: {
-      p2p: {
-        nitro
-      }
-    },
-    upstream: {
-      ethServer: {
-        rpcProviderEndpoint
-      }
-    }
-  } = config;
-
-  const signer = new utils.KeySigner(nitro.privateKey);
-  await signer.init();
 
-  // TODO: Use serverCmd.peer private key for nitro-client?
-  const store = await DurableStore.newDurableStore(signer, path.resolve(nitro.store));
-  const msgService = await P2PMessageService.newMessageService(store.getAddress(), peer);
-
-  const chainService = await EthChainService.newEthChainService(
-    rpcProviderEndpoint,
-    nitro.chainPrivateKey,
-    nitroAdjudicatorAddress,
-    consensusAppAddress,
-    virtualPaymentAppAddress
-  );
-
-  return Node.new(
-    msgService,
-    chainService,
-    store,
-    undefined,
-    new PermissivePolicy()
-  );
+  const typeDefs = fs.readFileSync(path.join(__dirname, 'schema.gql')).toString();
+  await serverCmd.exec(createResolvers, typeDefs, nitroPaymentsManager);
 };
 
 main().then(() => {