Merge branch 'develop' into refactor/sequencing

# Conflicts:
#	packages/sequencer/src/protocol/production/sequencing/BlockProducerModule.ts
#	packages/sequencer/src/protocol/production/sequencing/TransactionExecutionService.ts

Author: rpanic

.github/workflows/pull-request-develop.yml

@@ -91,6 +91,8 @@ jobs:
 
       - name: "Test"
         run: npm run test:ci
+        env:
+          IN_CI: true
 
   integration:
     runs-on: ubuntu-latest

packages/api/src/graphql/modules/MempoolResolver.ts

@@ -137,6 +137,7 @@ export class MempoolResolver extends GraphqlModule {
     return decoded.hash().toString();
   }
 
+  // TODO Add retrieval of pending messages somewhere as well
   @Query(() => InclusionStatus, {
     description: "Returns the state of a given transaction",
   })
@@ -146,13 +147,6 @@ export class MempoolResolver extends GraphqlModule {
     })
     hash: string
   ): Promise<InclusionStatus> {
-    const txs = await this.mempool.getTxs();
-    const tx = txs.find((x) => x.hash().toString() === hash);
-
-    if (tx) {
-      return InclusionStatus.PENDING;
-    }
-
     const dbTx = await this.transactionStorage.findTransaction(hash);
 
     if (dbTx !== undefined) {
@@ -162,6 +156,7 @@ export class MempoolResolver extends GraphqlModule {
       if (dbTx.block !== undefined) {
         return InclusionStatus.INCLUDED;
       }
+      return InclusionStatus.PENDING;
     }
 
     return InclusionStatus.UNKNOWN;
@@ -172,7 +167,7 @@ export class MempoolResolver extends GraphqlModule {
       "Returns the hashes of all transactions that are currently inside the mempool",
   })
   public async transactions() {
-    const txs = await this.mempool.getTxs();
+    const txs = await this.transactionStorage.getPendingUserTransactions();
     return txs.map((x) => x.hash().toString());
   }
 }

packages/common/src/index.ts

@@ -22,3 +22,4 @@ export * from "./compiling/AtomicCompileHelper";
 export * from "./compiling/CompileRegistry";
 export * from "./compiling/CompilableModule";
 export * from "./compiling/services/ChildVerificationKeyService";
+export * from "./config/injectAlias";

packages/common/src/log.ts

@@ -25,6 +25,11 @@ function logProvable(
 }
 /* eslint-enable */
 
+// eslint-disable-next-line @typescript-eslint/strict-boolean-expressions
+if (process.env?.IN_CI ?? false) {
+  loglevel.setLevel("ERROR");
+}
+
 const timeMap: Record<string, number> = {};
 
 function time(label = "time") {
@@ -111,7 +116,10 @@ export const log = {
   },
 
   setLevel: (level: LogLevelDesc) => {
-    loglevel.setLevel(level);
+    // eslint-disable-next-line @typescript-eslint/strict-boolean-expressions
+    if (!(process.env?.IN_CI ?? false)) {
+      loglevel.setLevel(level);
+    }
   },
 
   get levels() {

packages/common/src/utils.ts

@@ -19,6 +19,10 @@ export function requireTrue(
   }
 }
 
+/**
+ * Utility function to split an array of type T into a record <K, T[]> based on a
+ * function T => K that determines the key of each record
+ */
 export function splitArray<T, K extends string | number>(
   arr: T[],
   split: (t: T) => K

packages/deployment/src/queue/BullQueue.ts

@@ -22,10 +22,6 @@ export interface BullQueueConfig {
   retryAttempts?: number;
 }
 
-interface BullWorker extends Closeable {
-  get worker(): Worker;
-}
-
 /**
  * TaskQueue implementation for BullMQ
  */
@@ -36,10 +32,6 @@ export class BullQueue
 {
   private activePromise?: Promise<void>;
 
-  private activeWorkers: Record<string, BullWorker> = {};
-
-  private activeJobs = 0;
-
   public createWorker(
     name: string,
     executor: (data: TaskPayload) => Promise<TaskPayload>,
@@ -52,7 +44,6 @@ export class BullQueue
         // This is by far not optimal - since it still picks up 1 task per queue but waits until
         // computing them, so that leads to bad performance over multiple workers.
         // For that we need to restructure tasks to be flowing through a single queue however
-        this.activeJobs += 1;
 
         // TODO Use worker.pause()
         while (this.activePromise !== undefined) {
@@ -65,27 +56,10 @@ export class BullQueue
         });
         this.activePromise = promise;
 
-        // Pause all other workers
-        const workersToPause = Object.entries(this.activeWorkers).filter(
-          ([key]) => key !== name
-        );
-        await Promise.all(
-          workersToPause.map(([, workerToPause]) =>
-            workerToPause.worker.pause(true)
-          )
-        );
-
         const result = await executor(job.data);
         this.activePromise = undefined;
         void resOutside();
 
-        this.activeJobs -= 1;
-        if (this.activeJobs === 0) {
-          Object.entries(this.activeWorkers).forEach(([, resumingWorker]) =>
-            resumingWorker.worker.resume()
-          );
-        }
-
         return result;
       },
       {
@@ -104,16 +78,11 @@ export class BullQueue
       log.error(error);
     });
 
-    const instantiatedWorker = {
+    return {
       async close() {
         await worker.close();
       },
-      get worker() {
-        return worker;
-      },
     };
-    this.activeWorkers[name] = instantiatedWorker;
-    return instantiatedWorker;
   }
 
   public async getQueue(queueName: string): Promise<InstantiatedQueue> {

packages/persistance/src/PrismaDatabaseConnection.ts

@@ -137,4 +137,8 @@ export class PrismaDatabaseConnection
   public async close() {
     await this.prismaClient.$disconnect();
   }
+
+  public async executeInTransaction(f: () => Promise<void>) {
+    await this.prismaClient.$transaction(f);
+  }
 }

packages/persistance/src/PrismaRedisDatabase.ts

@@ -18,6 +18,7 @@ import {
   RedisConnection,
   RedisConnectionConfig,
   RedisConnectionModule,
+  RedisTransaction,
 } from "./RedisConnection";
 
 export interface PrismaRedisCombinedConfig {
@@ -49,6 +50,10 @@ export class PrismaRedisDatabase
     return this.redis.redisClient;
   }
 
+  public get currentMulti(): RedisTransaction {
+    return this.redis.currentMulti;
+  }
+
   public create(childContainerProvider: ChildContainerProvider) {
     super.create(childContainerProvider);
     this.prisma.create(childContainerProvider);
@@ -79,4 +84,12 @@ export class PrismaRedisDatabase
     await this.prisma.pruneDatabase();
     await this.redis.pruneDatabase();
   }
+
+  public async executeInTransaction(f: () => Promise<void>) {
+    // TODO Long-term we want to somehow make sure we can rollback one data source
+    //  if commiting the other one's transaction fails
+    await this.prisma.executeInTransaction(async () => {
+      await this.redis.executeInTransaction(f);
+    });
+  }
 }

packages/persistance/src/RedisConnection.ts

@@ -14,8 +14,11 @@ export interface RedisConnectionConfig {
   username?: string;
 }
 
+export type RedisTransaction = ReturnType<RedisClientType["multi"]>;
+
 export interface RedisConnection {
   get redisClient(): RedisClientType;
+  get currentMulti(): RedisTransaction;
 }
 
 export class RedisConnectionModule
@@ -82,4 +85,20 @@ export class RedisConnectionModule
   public async pruneDatabase() {
     await this.redisClient.flushDb();
   }
+
+  private multi?: RedisTransaction;
+
+  public get currentMulti() {
+    if (this.multi === undefined) {
+      throw new Error("Redis multi was access outside of a transaction");
+    }
+    return this.multi;
+  }
+
+  public async executeInTransaction(f: () => Promise<void>) {
+    this.multi = this.redisClient.multi();
+    await f();
+    await this.multi.exec();
+    this.multi = undefined;
+  }
 }

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

@@ -8,6 +8,7 @@ import {
   BlockStorage,
   BlockWithResult,
   BlockWithPreviousResult,
+  BlockWithMaybeResult,
 } from "@proto-kit/sequencer";
 import { filterNonNull, log } from "@proto-kit/common";
 import {
@@ -39,7 +40,7 @@ export class PrismaBlockStorage
 
   private async getBlockByQuery(
     where: { height: number } | { hash: string }
-  ): Promise<BlockWithResult | undefined> {
+  ): Promise<BlockWithMaybeResult | undefined> {
     const dbResult = await this.connection.prismaClient.block.findFirst({
       where,
       include: {
@@ -57,18 +58,15 @@ export class PrismaBlockStorage
     const transactions = dbResult.transactions.map<TransactionExecutionResult>(
       (txresult) => this.transactionResultMapper.mapIn([txresult, txresult.tx])
     );
-    if (dbResult.result === undefined || dbResult.result === null) {
-      throw new Error(
-        `No Metadata has been set for block ${JSON.stringify(where)} yet`
-      );
-    }
 
     return {
       block: {
         ...this.blockMapper.mapIn(dbResult),
         transactions,
       },
-      result: this.blockResultMapper.mapIn(dbResult.result),
+      result: dbResult.result
+        ? this.blockResultMapper.mapIn(dbResult.result)
+        : undefined,
     };
   }
 
@@ -100,45 +98,42 @@ export class PrismaBlockStorage
 
     const { prismaClient } = this.connection;
 
-    await prismaClient.$transaction([
-      prismaClient.transaction.createMany({
-        data: block.transactions.map((txr) =>
-          this.transactionMapper.mapOut(txr.tx)
-        ),
-        skipDuplicates: true,
-      }),
-
-      prismaClient.block.create({
-        data: {
-          ...encodedBlock,
-          beforeNetworkState:
-            encodedBlock.beforeNetworkState as Prisma.InputJsonObject,
-          duringNetworkState:
-            encodedBlock.duringNetworkState as Prisma.InputJsonObject,
-
-          transactions: {
-            createMany: {
-              data: transactions.map((tx) => {
-                return {
-                  status: tx.status,
-                  statusMessage: tx.statusMessage,
-                  txHash: tx.txHash,
-
-                  stateTransitions:
-                    tx.stateTransitions as Prisma.InputJsonArray,
-                  protocolTransitions:
-                    tx.protocolTransitions as Prisma.InputJsonArray,
-                  events: tx.events as Prisma.InputJsonArray,
-                };
-              }),
-              skipDuplicates: true,
-            },
-          },
+    await prismaClient.transaction.createMany({
+      data: block.transactions.map((txr) =>
+        this.transactionMapper.mapOut(txr.tx)
+      ),
+      skipDuplicates: true,
+    });
+
+    await prismaClient.block.create({
+      data: {
+        ...encodedBlock,
+        beforeNetworkState:
+          encodedBlock.beforeNetworkState as Prisma.InputJsonObject,
+        duringNetworkState:
+          encodedBlock.duringNetworkState as Prisma.InputJsonObject,
 
-          batchHeight: undefined,
+        transactions: {
+          createMany: {
+            data: transactions.map((tx) => {
+              return {
+                status: tx.status,
+                statusMessage: tx.statusMessage,
+                txHash: tx.txHash,
+
+                stateTransitions: tx.stateTransitions as Prisma.InputJsonArray,
+                protocolTransitions:
+                  tx.protocolTransitions as Prisma.InputJsonArray,
+                events: tx.events as Prisma.InputJsonArray,
+              };
+            }),
+            skipDuplicates: true,
+          },
         },
-      }),
-    ]);
+
+        batchHeight: undefined,
+      },
+    });
   }
 
   public async pushResult(result: BlockResult): Promise<void> {
@@ -169,7 +164,9 @@ export class PrismaBlockStorage
     return (result?._max.height ?? -1) + 1;
   }
 
-  public async getLatestBlock(): Promise<BlockWithResult | undefined> {
+  public async getLatestBlockAndResult(): Promise<
+    BlockWithMaybeResult | undefined
+  > {
     const latestBlock = await this.connection.prismaClient.$queryRaw<
       { hash: string }[]
     >`SELECT b1."hash" FROM "Block" b1 
@@ -185,6 +182,22 @@ export class PrismaBlockStorage
     });
   }
 
+  public async getLatestBlock(): Promise<BlockWithResult | undefined> {
+    const result = await this.getLatestBlockAndResult();
+    if (result !== undefined) {
+      if (result.result === undefined) {
+        throw new Error(
+          `Block result for block ${result.block.height.toString()} not found`
+        );
+      }
+      return {
+        block: result.block,
+        result: result.result,
+      };
+    }
+    return result;
+  }
+
   public async getNewBlocks(): Promise<BlockWithPreviousResult[]> {
     const blocks = await this.connection.prismaClient.block.findMany({
       where: {