Revert "feat: adding ack to block produced websocket 🔌"

cli/lib/nf3.mjs

@@ -898,7 +898,7 @@ class Nf3 {
     };
     connection.onmessage = async message => {
       const msg = JSON.parse(message.data);
-      const { id, type, txDataToSign, block, transactions } = msg;
+      const { type, txDataToSign, block, transactions } = msg;
       logger.debug(`Proposer received websocket message of type ${type}`);
       if (type === 'block') {
         proposerQueue.push(async () => {
@@ -913,8 +913,6 @@ class Nf3 {
             blockProposeEmitter.emit('error', err, block, transactions);
           }
         });
-
-        connection.send(id);
       }
       return null;
     };

nightfall-optimist/src/event-handlers/chain-reorg.mjs

@@ -66,7 +66,7 @@ import {
   deleteTreeByBlockNumberL2,
   deleteNullifiersForBlock,
 } from '../services/database.mjs';
-import { waitForContract } from '../utils/index.mjs';
+import { waitForContract } from './subscribe.mjs';
 
 const { STATE_CONTRACT_NAME } = constants;
 

nightfall-optimist/src/event-handlers/index.mjs

@@ -1,4 +1,10 @@
-import { startEventQueue } from '../utils/index.mjs';
+import {
+  startEventQueue,
+  subscribeToBlockAssembledWebSocketConnection,
+  subscribeToChallengeWebSocketConnection,
+  subscribeToInstantWithDrawalWebSocketConnection,
+  subscribeToProposedBlockWebSocketConnection,
+} from './subscribe.mjs';
 import blockProposedEventHandler from './block-proposed.mjs';
 import newCurrentProposerEventHandler from './new-current-proposer.mjs';
 import transactionSubmittedEventHandler from './transaction-submitted.mjs';
@@ -37,4 +43,11 @@ const eventHandlers = {
   },
 };
 
-export { startEventQueue, eventHandlers };
+export {
+  startEventQueue,
+  subscribeToBlockAssembledWebSocketConnection,
+  subscribeToChallengeWebSocketConnection,
+  subscribeToInstantWithDrawalWebSocketConnection,
+  subscribeToProposedBlockWebSocketConnection,
+  eventHandlers,
+};

nightfall-optimist/src/event-handlers/subscribe.mjs

@@ -0,0 +1,158 @@
+/* eslint-disable no-await-in-loop */
+
+/**
+ * Module to subscribe to blockchain events
+ */
+import WebSocket from 'ws';
+import config from 'config';
+import logger from 'common-files/utils/logger.mjs';
+import { getContractInstance, getContractAddress } from 'common-files/utils/contract.mjs';
+import constants from 'common-files/constants/index.mjs';
+
+const {
+  PROPOSERS_CONTRACT_NAME,
+  SHIELD_CONTRACT_NAME,
+  CHALLENGES_CONTRACT_NAME,
+  STATE_CONTRACT_NAME,
+} = constants;
+const { RETRIES, WEBSOCKET_PORT, WEBSOCKET_PING_TIME } = config;
+const wss = new WebSocket.Server({ port: WEBSOCKET_PORT });
+
+/**
+Function that does some standardised setting up of a websocket's events.
+It logs open, close and error events, sets up a ping and logs the pong. It will
+close the socket on pong failure.  The user is expected to handle the reconnect.
+It does not set up the onmessage event because this tends to be case-specific.
+*/
+function setupWebsocketEvents(ws, socketName) {
+  let timeoutID;
+  // setup a pinger to ping the websocket correspondent
+  const intervalID = setInterval(() => {
+    ws.ping();
+    // set up a timeout - will close the websocket, which will trigger a reconnect
+    timeoutID = setTimeout(() => {
+      logger.warn(`Timed out waiting for ping response from ${socketName}`);
+      ws.terminate();
+    }, 2 * WEBSOCKET_PING_TIME);
+  }, WEBSOCKET_PING_TIME);
+  // check we received a pong in time (clears the timer set by the pinger)
+  ws.on('pong', () => {
+    // logger.debug(`Got pong from ${socketName} websocket`);
+    clearTimeout(timeoutID);
+  });
+  ws.on('error', () => {
+    logger.debug(`ERROR ${socketName}`);
+  });
+  ws.on('open', () => {
+    logger.debug(`OPEN ${socketName}`);
+  });
+  ws.on('close', err => {
+    logger.debug(`CLOSE ${socketName} ${err}`);
+    clearInterval(intervalID);
+  });
+}
+
+/**
+ * Function that tries to get a (named) contract instance and, if it fails, will
+ * retry after 3 seconds.  After RETRIES attempts, it will give up and throw.
+ * This is useful in case nightfall-optimist comes up before the contract
+ * is fully deployed.
+ */
+export async function waitForContract(contractName) {
+  let errorCount = 0;
+  let error;
+  let instance;
+  while (errorCount < RETRIES) {
+    try {
+      error = undefined;
+      const address = await getContractAddress(contractName);
+      if (address === undefined) throw new Error(`${contractName} contract address was undefined`);
+      instance = getContractInstance(contractName, address);
+      return instance;
+    } catch (err) {
+      error = err;
+      errorCount++;
+      logger.warn(`Unable to get a ${contractName} contract instance will try again in 3 seconds`);
+      await new Promise(resolve => setTimeout(() => resolve(), 3000));
+    }
+  }
+  if (error) throw error;
+  return instance;
+}
+
+/**
+ *
+ * @param callback - The function that distributes events to the event-handler function
+ * @param arg - List of arguments to be passed to callback, the first element must be the event-handler functions
+ * @returns = List of emitters from each contract.
+ */
+export async function startEventQueue(callback, ...arg) {
+  const contractNames = [
+    STATE_CONTRACT_NAME,
+    SHIELD_CONTRACT_NAME,
+    CHALLENGES_CONTRACT_NAME,
+    PROPOSERS_CONTRACT_NAME,
+  ];
+  const contracts = await Promise.all(contractNames.map(c => waitForContract(c)));
+  const emitters = contracts.map(e => {
+    const emitterC = e.events.allEvents();
+    emitterC.on('changed', event => callback(event, arg));
+    emitterC.on('data', event => callback(event, arg));
+    return emitterC;
+  });
+  logger.debug('Subscribed to layer 2 state events');
+  return emitters;
+}
+
+export async function subscribeToChallengeWebSocketConnection(callback, ...args) {
+  wss.on('connection', ws => {
+    ws.on('message', message => {
+      if (message === 'challenge') {
+        setupWebsocketEvents(ws, 'challenge');
+        callback(ws, args);
+      }
+    });
+  });
+  logger.debug('Subscribed to Challenge WebSocket connection');
+}
+
+export async function subscribeToBlockAssembledWebSocketConnection(callback, ...args) {
+  wss.on('connection', ws => {
+    ws.on('message', message => {
+      if (message === 'blocks') {
+        setupWebsocketEvents(ws, 'proposer');
+        callback(ws, args);
+      }
+    });
+  });
+  logger.debug('Subscribed to BlockAssembled WebSocket connection');
+}
+
+export async function subscribeToInstantWithDrawalWebSocketConnection(callback, ...args) {
+  wss.on('connection', ws => {
+    ws.on('message', message => {
+      if (message === 'instant') {
+        setupWebsocketEvents(ws, 'liquidity provider');
+        callback(ws, args);
+      }
+    });
+  });
+  logger.debug('Subscribed to InstantWithDrawal WebSocket connection');
+}
+
+export async function subscribeToProposedBlockWebSocketConnection(callback, ...args) {
+  wss.on('connection', ws => {
+    ws.on('message', message => {
+      try {
+        if (JSON.parse(message).type === 'sync') {
+          logger.info(`SUBSCRIBING TO PROPOSEDBLOCK`);
+          setupWebsocketEvents(ws, 'publisher');
+          callback(ws, args);
+        }
+      } catch (error) {
+        logger.debug('Not JSON Message');
+      }
+    });
+  });
+  logger.debug('Subscribed to ProposedBlock WebSocket connection');
+}

nightfall-optimist/src/index.mjs

@@ -1,9 +1,14 @@
 import logger from 'common-files/utils/logger.mjs';
 import { queueManager, queues, enqueueEvent } from 'common-files/utils/event-queue.mjs';
-import { NFWebsocket } from 'common-files/utils/websocket.mjs';
-import config from 'config';
 import app from './app.mjs';
-import { startEventQueue, eventHandlers } from './event-handlers/index.mjs';
+import {
+  startEventQueue,
+  subscribeToBlockAssembledWebSocketConnection,
+  subscribeToChallengeWebSocketConnection,
+  subscribeToInstantWithDrawalWebSocketConnection,
+  subscribeToProposedBlockWebSocketConnection,
+  eventHandlers,
+} from './event-handlers/index.mjs';
 import Proposer from './classes/proposer.mjs';
 import {
   setBlockAssembledWebSocketConnection,
@@ -15,27 +20,15 @@ import { setInstantWithdrawalWebSocketConnection } from './services/instant-with
 import { setProposer } from './routes/proposer.mjs';
 import { setBlockProposedWebSocketConnection } from './event-handlers/block-proposed.mjs';
 
-const { WEBSOCKET_PORT, WEBSOCKET_PING_TIME } = config;
-
 const main = async () => {
   try {
     const proposer = new Proposer();
     setProposer(proposer); // passes the proposer instance int the proposer routes
     // subscribe to WebSocket events first
-
-    const ws = new NFWebsocket({ port: WEBSOCKET_PORT, pingTime: WEBSOCKET_PING_TIME });
-
-    ws.subscribe({ topic: 'challenge', socketName: 'challenge' }, setChallengeWebSocketConnection);
-    ws.subscribe({ topic: 'blocks', socketName: 'proposer' }, setBlockAssembledWebSocketConnection);
-    ws.subscribe(
-      { topic: 'instant', socketName: 'liquidity provider' },
-      setInstantWithdrawalWebSocketConnection,
-    );
-    ws.subscribe(
-      { topic: 'sync', socketName: 'publisher', filter: 'type' },
-      setBlockProposedWebSocketConnection,
-    );
-
+    await subscribeToBlockAssembledWebSocketConnection(setBlockAssembledWebSocketConnection);
+    await subscribeToChallengeWebSocketConnection(setChallengeWebSocketConnection);
+    await subscribeToInstantWithDrawalWebSocketConnection(setInstantWithdrawalWebSocketConnection);
+    await subscribeToProposedBlockWebSocketConnection(setBlockProposedWebSocketConnection);
     // start the event queue
     await startEventQueue(queueManager, eventHandlers, proposer);
     // enqueue the block-assembler every time the queue becomes empty

nightfall-optimist/src/routes/proposer.mjs

@@ -20,7 +20,7 @@ import {
   getLatestTree,
   getLatestBlockInfo,
 } from '../services/database.mjs';
-import { waitForContract } from '../utils/index.mjs';
+import { waitForContract } from '../event-handlers/subscribe.mjs';
 import transactionSubmittedEventHandler from '../event-handlers/transaction-submitted.mjs';
 import getProposers from '../services/proposer.mjs';