feat: add stopQueue to halt block production whiel waiting for rollback

common-files/utils/event-queue.mjs

@@ -26,7 +26,8 @@ import logger from 'common-files/utils/logger.mjs';
 const { MAX_QUEUE } = config;
 const fastQueue = new Queue({ autostart: true, concurrency: 1 });
 const slowQueue = new Queue({ autostart: true, concurrency: 1 });
-export const queues = [fastQueue, slowQueue];
+const stopQueue = new Queue({ autostart: false, concurrency: 1 });
+export const queues = [fastQueue, slowQueue, stopQueue];
 
 /**
 This function will return a promise that resolves to true when the next highest
@@ -68,6 +69,10 @@ async function enqueueEvent(callback, priority, args) {
   });
 }
 
+async function dequeueEvent(priority) {
+  queues[priority].shift();
+}
+
 async function queueManager(eventObject, eventArgs) {
   // First element of eventArgs must be the eventHandlers object
   const [eventHandlers, ...args] = eventArgs;
@@ -105,4 +110,4 @@ async function queueManager(eventObject, eventArgs) {
 }
 
 /* ignore unused exports */
-export { flushQueue, enqueueEvent, queueManager };
+export { flushQueue, enqueueEvent, dequeueEvent, queueManager };

nightfall-optimist/src/event-handlers/block-proposed.mjs

@@ -1,6 +1,7 @@
 import logger from 'common-files/utils/logger.mjs';
 import Timber from 'common-files/classes/timber.mjs';
 import config from 'config';
+import { enqueueEvent } from 'common-files/utils/event-queue.mjs';
 import checkBlock from '../services/check-block.mjs';
 import BlockError from '../classes/block-error.mjs';
 import { createChallenge } from '../services/challenges.mjs';
@@ -33,13 +34,7 @@ async function blockProposedEventHandler(data) {
     // asociated with a failed block, and we can't do that if we haven't
     // associated them with a blockHash.
     await stampNullifiers(
-      transactions
-        .map(tx =>
-          tx.nullifiers.filter(
-            nulls => nulls !== '0x0000000000000000000000000000000000000000000000000000000000000000',
-          ),
-        )
-        .flat(Infinity),
+      transactions.map(tx => tx.nullifiers.filter(nulls => nulls !== ZERO)).flat(Infinity),
       block.blockHash,
     );
     // mark transactions so that they are out of the mempool,
@@ -59,6 +54,11 @@ async function blockProposedEventHandler(data) {
   } catch (err) {
     if (err instanceof BlockError) {
       logger.warn(`Block Checker - Block invalid, with code ${err.code}! ${err.message}`);
+      logger.info(`Block is invalid, stopping any block production`);
+      // We enqueue an event onto the stopQueue to halt block production.
+      // This message will not be printed because event dequeuing does not run the job.
+      // This is fine as we are just using it to stop running.
+      await enqueueEvent(() => logger.info('Stop Until Rollback'), 2);
       await createChallenge(block, transactions, err);
     } else {
       logger.error(err.stack);

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

@@ -6,6 +6,7 @@ leafCount values in the Block class
 */
 import logger from 'common-files/utils/logger.mjs';
 import config from 'config';
+import { dequeueEvent, enqueueEvent } from 'common-files/utils/event-queue.mjs';
 import {
   addTransactionsToMemPool,
   deleteBlock,
@@ -163,6 +164,10 @@ async function rollbackEventHandler(data) {
     un => !validTransactionNullifiers.includes(un) && !nullifierArray.includes(un),
   );
   await deleteNullifiers(deletedNullifiers);
+  await dequeueEvent(2); // Remove an event from the stopQueue.
+  // A Rollback triggers a NewCurrentProposer event which shoudl trigger queue[0].end()
+  // But to be safe we enqueue a helper event to guarantee queue[0].end() runs.
+  await enqueueEvent(() => logger.info(`Rollback Completed`), 0);
 }
 
 export default rollbackEventHandler;

nightfall-optimist/src/index.mjs

@@ -30,8 +30,11 @@ const main = async () => {
       await startEventQueue(queueManager, eventHandlers, proposer);
       queues[0].on('end', () => {
         // We do the proposer isMe check here to fail fast instead of re-enqueing.
-        logger.info('Queue has emptied. Queueing block assembler.');
-        if (proposer.isMe) return enqueueEvent(conditionalMakeBlock, 0, proposer);
+        // We check if the queue[2] is empty, this is safe it is manually enqueued/dequeued.
+        if (proposer.isMe && queues[2].length === 0) {
+          logger.info('Queue has emptied. Queueing block assembler.');
+          return enqueueEvent(conditionalMakeBlock, 0, proposer);
+        }
         // eslint-disable-next-line no-void, no-useless-return
         return void false; // This is here to satisfy consistent return rules, we do nothing.
       });