Merge pull request #399 from proto-kit/refactor/busy-pattern

Refactor/busy pattern

Author: rpanic

packages/sequencer/src/helpers/BusyGuard.ts

@@ -0,0 +1,30 @@
+import { log } from "@proto-kit/common";
+/**
+ * Decorator that ensures a function/method is not currently in use.
+ * Mostly useful for production of blocks, batches and tasks.
+ */
+export function ensureNotBusy<T>() {
+  let inProgress = false;
+
+  return function innerFunction(
+    _target: T,
+    methodName: string,
+    descriptor: TypedPropertyDescriptor<(...args: any[]) => Promise<any>>
+  ): void {
+    const originalMethod = descriptor.value!;
+
+    descriptor.value = async function wrapped(this: T, ...args: unknown[]) {
+      if (inProgress) {
+        log.trace(`${methodName} is in use at the moment.`);
+        return undefined;
+      }
+
+      inProgress = true;
+      try {
+        return await originalMethod.apply(this, args);
+      } finally {
+        inProgress = false;
+      }
+    };
+  };
+}

packages/sequencer/src/index.ts

@@ -1,4 +1,5 @@
 export * from "./helpers/utils";
+export * from "./helpers/BusyGuard";
 export * from "./mempool/Mempool";
 export * from "./mempool/PendingTransaction";
 export * from "./mempool/CompressedSignature";

packages/sequencer/src/protocol/production/BatchProducerModule.ts

@@ -17,6 +17,7 @@ import { BlockWithResult } from "../../storage/model/Block";
 import type { Database } from "../../storage/Database";
 import { AsyncLinkedLeafStore } from "../../state/async/AsyncLinkedLeafStore";
 import { CachedLinkedLeafStore } from "../../state/lmt/CachedLinkedLeafStore";
+import { ensureNotBusy } from "../../helpers/BusyGuard";
 
 import { BlockProofSerializer } from "./tasks/serializers/BlockProofSerializer";
 import { BatchTracingService } from "./tracing/BatchTracingService";
@@ -44,8 +45,6 @@ const errors = {
  */
 @sequencerModule()
 export class BatchProducerModule extends SequencerModule {
-  private productionInProgress = false;
-
   public constructor(
     @inject("AsyncLinkedLeafStore")
     private readonly merkleStore: AsyncLinkedLeafStore,
@@ -62,47 +61,11 @@ export class BatchProducerModule extends SequencerModule {
   /**
    * Main function to call when wanting to create a new block based on the
    * transactions that are present in the mempool. This function should also
-   * be the one called by BlockTriggerss
+   * be the one called by BlockTriggers.
    */
+  @ensureNotBusy()
   public async createBatch(
     blocks: BlockWithResult[]
-  ): Promise<SettleableBatch | undefined> {
-    if (!this.productionInProgress) {
-      try {
-        this.productionInProgress = true;
-
-        const batch = await this.tryProduceBatch(blocks);
-
-        this.productionInProgress = false;
-
-        return batch;
-      } catch (error: unknown) {
-        this.productionInProgress = false;
-        // TODO Check if that still makes sense
-        if (error instanceof Error) {
-          if (
-            !error.message.includes(
-              "Can't create a block with zero transactions"
-            )
-          ) {
-            log.error(error);
-          }
-
-          throw error;
-        } else {
-          log.error(error);
-        }
-      }
-    } else {
-      log.debug(
-        "Skipping new block production because production is still in progress"
-      );
-    }
-    return undefined;
-  }
-
-  private async tryProduceBatch(
-    blocks: BlockWithResult[]
   ): Promise<SettleableBatch | undefined> {
     log.info("Producing batch...");
 

packages/sequencer/src/protocol/production/sequencing/BlockProducerModule.ts

@@ -27,6 +27,7 @@ import { IncomingMessagesService } from "../../../settlement/messages/IncomingMe
 import { Tracer } from "../../../logging/Tracer";
 import { trace } from "../../../logging/trace";
 import { AsyncLinkedLeafStore } from "../../../state/async/AsyncLinkedLeafStore";
+import { ensureNotBusy } from "../../../helpers/BusyGuard";
 
 import { BlockProductionService } from "./BlockProductionService";
 import { BlockResultService } from "./BlockResultService";
@@ -38,8 +39,6 @@ export interface BlockConfig {
 
 @sequencerModule()
 export class BlockProducerModule extends SequencerModule<BlockConfig> {
-  private productionInProgress = false;
-
   public constructor(
     @inject("Mempool") private readonly mempool: Mempool,
     @inject("IncomingMessagesService", { isOptional: true })
@@ -140,37 +139,25 @@ export class BlockProducerModule extends SequencerModule<BlockConfig> {
     return result;
   }
 
+  @ensureNotBusy()
   public async tryProduceBlock(): Promise<Block | undefined> {
-    if (!this.productionInProgress) {
-      try {
-        const block = await this.produceBlock();
-
-        if (block === undefined) {
-          if (!this.allowEmptyBlock()) {
-            log.info("No transactions in mempool, skipping production");
-          } else {
-            log.error("Something wrong happened, skipping block");
-          }
-          return undefined;
-        }
+    const block = await this.produceBlock();
 
-        log.info(
-          `Produced block #${block.height.toBigInt()} (${block.transactions.length} txs)`
-        );
-        this.prettyPrintBlockContents(block);
-
-        return block;
-      } catch (error: unknown) {
-        if (error instanceof Error) {
-          throw error;
-        } else {
-          log.error(error);
-        }
-      } finally {
-        this.productionInProgress = false;
+    if (block === undefined) {
+      if (!this.allowEmptyBlock()) {
+        log.info("No transactions in mempool, skipping production");
+      } else {
+        log.error("Something wrong happened, skipping block");
       }
+      return undefined;
     }
-    return undefined;
+
+    log.info(
+      `Produced block #${block.height.toBigInt()} (${block.transactions.length} txs)`
+    );
+    this.prettyPrintBlockContents(block);
+
+    return block;
   }
 
   // TODO Move to different service, to remove dependency on mempool and messagequeue
@@ -220,8 +207,6 @@ export class BlockProducerModule extends SequencerModule<BlockConfig> {
 
   @trace("block")
   private async produceBlock(): Promise<Block | undefined> {
-    this.productionInProgress = true;
-
     const { txs, metadata } = await this.collectProductionData();
 
     // Skip production if no transactions are available for now
@@ -263,8 +248,6 @@ export class BlockProducerModule extends SequencerModule<BlockConfig> {
       );
     }
 
-    this.productionInProgress = false;
-
     return blockResult?.block;
   }
 

packages/sequencer/src/protocol/production/trigger/TimedBlockTrigger.ts

@@ -12,6 +12,7 @@ import {
   BridgingModule,
   SettlementTokenConfig,
 } from "../../../settlement/BridgingModule";
+import { ensureNotBusy } from "../../../helpers/BusyGuard";
 
 import { BlockEvents, BlockTriggerBase } from "./BlockTrigger";
 
@@ -45,9 +46,6 @@ export class TimedBlockTrigger
 
   private interval?: any;
 
-  // TODO Move that logic to somewhere proper
-  private settlementInProgress = false;
-
   public constructor(
     @inject("BatchProducerModule", { isOptional: true })
     batchProducerModule: BatchProducerModule | undefined,
@@ -123,15 +121,9 @@ export class TimedBlockTrigger
         // otherwise treat as unproven-only
         if (
           settlementInterval !== undefined &&
-          totalTime % settlementInterval === 0 &&
-          !this.settlementInProgress
+          totalTime % settlementInterval === 0
         ) {
-          this.settlementInProgress = true;
-          const batch = await this.produceBatch();
-          if (batch !== undefined) {
-            await this.settle(batch, this.config.settlementTokenConfig);
-          }
-          this.settlementInProgress = false;
+          await this.tryProduceSettlement();
         }
       } catch (error) {
         log.error(error);
@@ -151,6 +143,14 @@ export class TimedBlockTrigger
     }
   }
 
+  @ensureNotBusy()
+  private async tryProduceSettlement(): Promise<void> {
+    const batch = await this.produceBatch();
+    if (batch !== undefined) {
+      await this.settle(batch, this.config.settlementTokenConfig);
+    }
+  }
+
   public async close(): Promise<void> {
     // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
     clearInterval(this.interval);