Merge branch 'develop' into feature/closeable

# Conflicts:
#	packages/sequencer/src/storage/Database.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/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/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: {

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

@@ -51,25 +51,24 @@ export class PrismaMessageStorage implements MessageStorage {
     );
 
     const { prismaClient } = this.connection;
-    await prismaClient.$transaction([
-      prismaClient.transaction.createMany({
-        data: transactions,
-        skipDuplicates: true,
-      }),
 
-      prismaClient.incomingMessageBatch.create({
-        data: {
-          fromMessageHash,
-          toMessageHash,
-          messages: {
-            createMany: {
-              data: transactions.map((transaction) => ({
-                transactionHash: transaction.hash,
-              })),
-            },
+    await prismaClient.transaction.createMany({
+      data: transactions,
+      skipDuplicates: true,
+    });
+
+    await prismaClient.incomingMessageBatch.create({
+      data: {
+        fromMessageHash,
+        toMessageHash,
+        messages: {
+          createMany: {
+            data: transactions.map((transaction) => ({
+              transactionHash: transaction.hash,
+            })),
           },
         },
-      }),
-    ]);
+      },
+    });
   }
 }

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

@@ -36,19 +36,17 @@ export class PrismaStateService implements AsyncStateService {
         mask: this.mask,
       }));
 
-    await prismaClient.$transaction([
-      prismaClient.state.deleteMany({
-        where: {
-          path: {
-            in: this.cache.map((x) => new Decimal(x.key.toString())),
-          },
-          mask: this.mask,
+    await prismaClient.state.deleteMany({
+      where: {
+        path: {
+          in: this.cache.map((x) => new Decimal(x.key.toString())),
         },
-      }),
-      prismaClient.state.createMany({
-        data,
-      }),
-    ]);
+        mask: this.mask,
+      },
+    });
+    await prismaClient.state.createMany({
+      data,
+    });
 
     this.cache = [];
   }

packages/persistance/src/services/redis/RedisMerkleTreeStore.ts

@@ -34,7 +34,7 @@ export class RedisMerkleTreeStore implements AsyncMerkleTreeStore {
     }
 
     try {
-      await this.connection.redisClient.mSet(array.flat(1));
+      this.connection.currentMulti.mSet(array.flat(1));
     } catch (error) {
       log.error(error);
     }
@@ -62,8 +62,8 @@ export class RedisMerkleTreeStore implements AsyncMerkleTreeStore {
   public writeNodes(nodes: MerkleTreeNode[]): void {
     this.cache = this.cache.concat(nodes);
     // TODO Filter distinct
-    // We might not even need this, since the distinctness filter might already
-    // be implicitely done by the layer above (i.e. cachedmtstore)
+    //  We might not even need this, since the distinctness filter might already
+    //  be implicitely done by the layer above (i.e. cachedmtstore)
 
     // Leaving this for now until I get to implementing it
     // const concat = this.cache.concat(nodes);