Merge pull request #417 from proto-kit/feature/independent-trigger-timers

Made TimedBlockTrigger use independent timers

Author: rpanic

CHANGELOG.md

@@ -15,3 +15,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Separated settlement and bridging functionally, so now settlement can be used without bridging [#376](https://github.com/proto-kit/framework/pull/376)
 - Added nightly releases via pkg.pr.new [#384](https://github.com/proto-kit/framework/pull/384)
 - Introduced Changelog [#378](https://github.com/proto-kit/framework/pull/378)
+
+### Removed
+
+- Removed `tick` event on `BlockTrigger` [#417](https://github.com/proto-kit/framework/pull/417)
+

packages/persistance/src/services/prisma/PrismaTransactionStorage.ts

@@ -49,6 +49,24 @@ export class PrismaTransactionStorage implements TransactionStorage {
     return txs.map((tx) => this.transactionMapper.mapIn(tx));
   }
 
+  public async countPendingUserTransactions() {
+    const { prismaClient } = this.connection;
+
+    return await prismaClient.transaction.count({
+      where: {
+        executionResult: {
+          is: null,
+        },
+        isMessage: {
+          equals: false,
+        },
+        inputPaths: {
+          is: null,
+        },
+      },
+    });
+  }
+
   public async removeTx(hashes: string[], type: "included" | "dropped") {
     // In our schema, included txs are simply just linked with blocks, so we only
     // need to delete if we drop a tx

packages/sequencer/src/helpers/BusyGuard.ts

@@ -4,18 +4,20 @@ import { log } from "@proto-kit/common";
  * Mostly useful for production of blocks, batches and tasks.
  */
 export function ensureNotBusy<T>() {
-  let inProgress = false;
-
-  return function innerFunction(
+  return function innerFunction<R>(
     _target: T,
     methodName: string,
-    descriptor: TypedPropertyDescriptor<(...args: any[]) => Promise<any>>
+    descriptor: TypedPropertyDescriptor<
+      (...args: any[]) => Promise<R | undefined>
+    >
   ): void {
+    let inProgress = false;
+
     const originalMethod = descriptor.value!;
 
     descriptor.value = async function wrapped(this: T, ...args: unknown[]) {
       if (inProgress) {
-        log.trace(`${methodName} is in use at the moment.`);
+        log.trace(`${methodName} is in use at the moment, skipping execution.`);
         return undefined;
       }
 

packages/sequencer/src/mempool/Mempool.ts

@@ -22,4 +22,6 @@ export interface Mempool<Events extends MempoolEvents = MempoolEvents>
   getMandatoryTxs: () => Promise<PendingTransaction[]>;
 
   removeTxs: (included: string[], dropped: string[]) => Promise<void>;
+
+  length: () => Promise<number>;
 }

packages/sequencer/src/mempool/private/PrivateMempool.ts

@@ -47,8 +47,12 @@ export class PrivateMempool
   }
 
   public async length(): Promise<number> {
-    const txs = await this.transactionStorage.getPendingUserTransactions(0);
-    return txs.length;
+    const numUserTxs =
+      await this.transactionStorage.countPendingUserTransactions();
+
+    const messages = await this.getMandatoryTxs();
+
+    return numUserTxs + messages.length;
   }
 
   public async add(tx: PendingTransaction): Promise<boolean> {

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

@@ -1,6 +1,5 @@
 import { inject, injectable } from "tsyringe";
 import { log } from "@proto-kit/common";
-import gcd from "compute-gcd";
 
 import { closeable, Closeable } from "../../../sequencer/builder/Closeable";
 import { BatchProducerModule } from "../BatchProducerModule";
@@ -14,37 +13,23 @@ import {
 } from "../../../settlement/BridgingModule";
 import { ensureNotBusy } from "../../../helpers/BusyGuard";
 
-import { BlockEvents, BlockTriggerBase } from "./BlockTrigger";
+import { BlockTriggerBase } from "./BlockTrigger";
 
 export interface TimedBlockTriggerConfig {
-  /**
-   * Interval for the tick event to be fired.
-   * The time x of any block trigger time is always guaranteed to be
-   * tick % x == 0.
-   * Value has to be a divisor of gcd(blockInterval, settlementInterval).
-   * If it doesn't satisfy this requirement, this config will not be respected
-   */
-  tick?: number;
   settlementInterval?: number;
   blockInterval: number;
   produceEmptyBlocks?: boolean;
 
   settlementTokenConfig: SettlementTokenConfig;
 }
 
-export interface TimedBlockTriggerEvent extends BlockEvents {
-  tick: [number];
-}
-
 @injectable()
 @closeable()
 export class TimedBlockTrigger
-  extends BlockTriggerBase<TimedBlockTriggerConfig, TimedBlockTriggerEvent>
+  extends BlockTriggerBase<TimedBlockTriggerConfig>
   implements Closeable
 {
-  // There is no real type for interval ids somehow, so any it is
-
-  private interval?: any;
+  private intervals: NodeJS.Timeout[] = [];
 
   public constructor(
     @inject("BatchProducerModule", { isOptional: true })
@@ -69,26 +54,6 @@ export class TimedBlockTrigger
     );
   }
 
-  private getTimerInterval(): number {
-    const { settlementInterval, blockInterval, tick } = this.config;
-
-    let timerInterval =
-      settlementInterval !== undefined
-        ? gcd(settlementInterval, blockInterval)
-        : blockInterval;
-
-    const definedTick = tick ?? 1000;
-    if (definedTick <= timerInterval) {
-      // Check if tick is a divisor of the calculated interval
-      const div = timerInterval / definedTick;
-      if (Math.floor(div) === div) {
-        timerInterval = definedTick;
-      }
-    }
-
-    return timerInterval;
-  }
-
   public async start(): Promise<void> {
     log.info("Starting timed block trigger");
     const { settlementInterval, blockInterval } = this.config;
@@ -102,43 +67,39 @@ export class TimedBlockTrigger
       );
     }
 
-    const timerInterval = this.getTimerInterval();
-
-    let totalTime = 0;
-    this.interval = setInterval(async () => {
-      totalTime += timerInterval;
-
-      this.events.emit("tick", totalTime);
-
+    const blockIntervalId = setInterval(async () => {
       try {
         // Trigger unproven blocks
-        if (totalTime % blockInterval === 0) {
-          await this.produceUnprovenBlock();
-        }
-
-        // Trigger proven (settlement) blocks
-        // Only produce settlements if a time has been set
-        // otherwise treat as unproven-only
-        if (
-          settlementInterval !== undefined &&
-          totalTime % settlementInterval === 0
-        ) {
-          await this.tryProduceSettlement();
-        }
+        await this.produceUnprovenBlock();
       } catch (error) {
         log.error(error);
       }
-    }, timerInterval);
+    }, blockInterval);
+    this.intervals.push(blockIntervalId);
+
+    if (settlementInterval !== undefined) {
+      const settlementIntervalId = setInterval(async () => {
+        try {
+          // Trigger settlement
+          await this.tryProduceSettlement();
+        } catch (error) {
+          log.error(error);
+        }
+      }, settlementInterval);
+      this.intervals.push(settlementIntervalId);
+    }
 
     await super.start();
   }
 
+  // This is technically not necessary since produceBlock checks business down the line
+  // but we save a bunch of DB checks before
+  @ensureNotBusy()
   private async produceUnprovenBlock() {
-    // TODO Optimize towards mempool.length()
-    const mempoolTxs = await this.mempool.getTxs(0);
+    const mempoolLength = await this.mempool.length();
     // Produce a block if either produceEmptyBlocks is true or we have more
-    // than 1 tx in mempool
-    if (mempoolTxs.length > 0 || (this.config.produceEmptyBlocks ?? true)) {
+    // than 1 tx in mempool or messages
+    if (mempoolLength > 0 || (this.config.produceEmptyBlocks ?? true)) {
       await this.produceBlock();
     }
   }
@@ -152,7 +113,8 @@ export class TimedBlockTrigger
   }
 
   public async close(): Promise<void> {
-    // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
-    clearInterval(this.interval);
+    this.intervals.forEach((interval) => {
+      clearInterval(interval);
+    });
   }
 }

packages/sequencer/src/storage/inmemory/InMemoryTransactionStorage.ts

@@ -66,6 +66,10 @@ export class InMemoryTransactionStorage implements TransactionStorage {
     return this.queue.slice(from, to).map(({ tx }) => tx);
   }
 
+  public async countPendingUserTransactions() {
+    return (await this.getPendingUserTransactions(0)).length;
+  }
+
   public async pushUserTransaction(
     tx: PendingTransaction,
     priority: number

packages/sequencer/src/storage/repositories/TransactionStorage.ts

@@ -11,6 +11,8 @@ export interface TransactionStorage {
     limit?: number
   ) => Promise<PendingTransaction[]>;
 
+  countPendingUserTransactions: () => Promise<number>;
+
   removeTx: (txHashes: string[], type: "included" | "dropped") => Promise<void>;
 
   /**