Added retry logic to minaTransactionSender

console-jest.config.js

@@ -1,3 +1,3 @@
-import console from "console";
+import "reflect-metadata";
 
-global.console = console;
+globalThis.console = console;

package.json

@@ -57,7 +57,7 @@
     "ts-jest": "^29.0.5",
     "typedoc": "^0.26.11",
     "typedoc-plugin-frontmatter": "1.0.0",
-    "typedoc-plugin-inline-sources": "1.2.0",
+    "typedoc-plugin-inline-sources": "^1.2.0",
     "typedoc-plugin-markdown": "4.2.10",
     "typescript": "5.1"
   },

packages/indexer/src/api/GeneratedResolverFactoryGraphqlModule.ts

@@ -90,6 +90,7 @@ export class GeneratedResolverFactoryGraphqlModule extends ResolverFactoryGraphq
   public async initializePrismaClient() {
     // setup the prisma client and feed it to the server,
     // since this is necessary for the returned resolvers to work
+
     const prismaClient = new PrismaClient({
       // datasourceUrl: 'postgresql://admin:password@localhost:5433/protokit-indexer?schema=public'
     });

packages/persistance/prisma/migrations/20251222105633_pending_l1_transactions/migration.sql

@@ -0,0 +1,44 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `settlementTransactionHash` on the `Batch` table. All the data in the column will be lost.
+  - The primary key for the `Settlement` table will be changed. If it partially fails, the table could be left without primary key constraint.
+  - You are about to drop the column `transactionHash` on the `Settlement` table. All the data in the column will be lost.
+  - Added the required column `transactionId` to the `Settlement` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- DropForeignKey
+ALTER TABLE "Batch" DROP CONSTRAINT "Batch_settlementTransactionHash_fkey";
+
+-- AlterTable
+ALTER TABLE "Batch" DROP COLUMN "settlementTransactionHash",
+ADD COLUMN     "settlementTransactionId" TEXT;
+
+-- AlterTable
+ALTER TABLE "Settlement" DROP CONSTRAINT "Settlement_pkey",
+DROP COLUMN "transactionHash",
+ADD COLUMN     "transactionId" TEXT NOT NULL,
+ADD CONSTRAINT "Settlement_pkey" PRIMARY KEY ("transactionId");
+
+-- CreateTable
+CREATE TABLE "PendingL1Transaction" (
+    "id" TEXT NOT NULL,
+    "sender" TEXT NOT NULL,
+    "nonce" INTEGER NOT NULL,
+    "attempts" INTEGER NOT NULL,
+    "status" VARCHAR(32) NOT NULL,
+    "transaction" JSON NOT NULL,
+    "lastError" TEXT,
+    "sentAt" TIMESTAMP(3),
+
+    CONSTRAINT "PendingL1Transaction_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "PendingL1Transaction_sender_nonce_key" ON "PendingL1Transaction"("sender", "nonce");
+
+-- AddForeignKey
+ALTER TABLE "Batch" ADD CONSTRAINT "Batch_settlementTransactionId_fkey" FOREIGN KEY ("settlementTransactionId") REFERENCES "Settlement"("transactionId") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "Settlement" ADD CONSTRAINT "Settlement_transactionId_fkey" FOREIGN KEY ("transactionId") REFERENCES "PendingL1Transaction"("id") ON DELETE RESTRICT ON UPDATE CASCADE;

packages/persistance/prisma/schema.prisma

@@ -105,8 +105,8 @@ model Batch {
 
   blocks Block[]
 
-  settlementTransactionHash String?
-  settlement                Settlement? @relation(fields: [settlementTransactionHash], references: [transactionHash])
+  settlementTransactionId String?
+  settlement              Settlement? @relation(fields: [settlementTransactionId], references: [transactionId])
 }
 
 model BlockResult {
@@ -123,11 +123,11 @@ model BlockResult {
 }
 
 model Settlement {
-  // transaction String
-  transactionHash      String @id
+  transactionId        String @id
   promisedMessagesHash String
 
-  batches Batch[]
+  batches              Batch[]
+  pendingL1Transaction PendingL1Transaction? @relation(fields: [transactionId], references: [id])
 }
 
 model IncomingMessageBatchTransaction {
@@ -148,3 +148,21 @@ model IncomingMessageBatch {
 
   messages IncomingMessageBatchTransaction[]
 }
+
+model PendingL1Transaction {
+  id          String    @id @default(cuid())
+  sender      String
+  nonce       Int
+  attempts    Int
+  status      String    @db.VarChar(32)
+  transaction Json      @db.Json
+  hash        String?
+  lastError   String?
+  sentAt      DateTime?
+  queuedAt    DateTime?
+  nextActionAt DateTime?
+
+  settlement Settlement?
+
+  @@unique([sender, nonce])
+}

packages/persistance/src/PrismaDatabaseConnection.ts

@@ -13,6 +13,7 @@ import { PrismaBlockStorage } from "./services/prisma/PrismaBlockStorage";
 import { PrismaSettlementStorage } from "./services/prisma/PrismaSettlementStorage";
 import { PrismaMessageStorage } from "./services/prisma/PrismaMessageStorage";
 import { PrismaTransactionStorage } from "./services/prisma/PrismaTransactionStorage";
+import { PrismaPendingL1TransactionStorage } from "./services/prisma/PrismaPendingL1TransactionStorage";
 
 export interface PrismaDatabaseConfig {
   // Either object-based config or connection string
@@ -82,6 +83,9 @@ export class PrismaDatabaseConnection
       transactionStorage: {
         useClass: PrismaTransactionStorage,
       },
+      pendingL1TransactionStorage: {
+        useClass: PrismaPendingL1TransactionStorage,
+      },
     };
   }
 
@@ -97,6 +101,7 @@ export class PrismaDatabaseConnection
       "IncomingMessageBatch",
       "IncomingMessageBatchTransaction",
       "LinkedLeaf",
+      "PendingL1Transaction",
     ];
 
     await this.prismaClient.$transaction(

packages/persistance/src/index.ts

@@ -7,6 +7,7 @@ export * from "./services/prisma/PrismaBatchStore";
 export * from "./services/prisma/PrismaSettlementStorage";
 export * from "./services/prisma/PrismaMessageStorage";
 export * from "./services/prisma/PrismaTransactionStorage";
+export * from "./services/prisma/PrismaPendingL1TransactionStorage";
 export * from "./services/prisma/mappers/BatchMapper";
 export * from "./services/prisma/mappers/BlockMapper";
 export * from "./services/prisma/mappers/FieldMapper";

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

@@ -0,0 +1,96 @@
+import { inject, injectable } from "tsyringe";
+import {
+  PendingL1TransactionRecord,
+  PendingL1TransactionStatus,
+  PendingL1TransactionStorage,
+} from "@proto-kit/sequencer";
+
+import type { PrismaConnection } from "../../PrismaDatabaseConnection";
+
+import { PendingL1TransactionMapper } from "./mappers/PendingL1TransactionMapper";
+
+@injectable()
+export class PrismaPendingL1TransactionStorage
+  implements PendingL1TransactionStorage
+{
+  private readonly mapper = new PendingL1TransactionMapper();
+
+  public constructor(
+    @inject("Database") private readonly connection: PrismaConnection
+  ) {}
+
+  public async queue(
+    record: Omit<PendingL1TransactionRecord, "status" | "id">
+  ): Promise<string> {
+    const { prismaClient } = this.connection;
+    const status: PendingL1TransactionStatus = "queued";
+    const txnRecord = await prismaClient.pendingL1Transaction.create({
+      data: this.mapper.mapOut({ ...record, status }),
+    });
+    return txnRecord.id;
+  }
+
+  public async update(
+    id: string,
+    updates: Partial<Omit<PendingL1TransactionRecord, "id">>
+  ): Promise<void> {
+    const { prismaClient } = this.connection;
+    await prismaClient.pendingL1Transaction.update({
+      where: { id },
+      data: {
+        ...(updates.attempts !== undefined && { attempts: updates.attempts }),
+        ...(updates.status !== undefined && { status: updates.status }),
+        ...(updates.transaction !== undefined && {
+          transaction: updates.transaction.toJSON(),
+        }),
+        ...(updates.hash !== undefined && { hash: updates.hash }),
+        ...(updates.lastError !== undefined && {
+          lastError: updates.lastError,
+        }),
+        ...(updates.sentAt !== undefined && { sentAt: updates.sentAt }),
+        ...(updates.queuedAt !== undefined && { queuedAt: updates.queuedAt }),
+        ...(updates.nextActionAt !== undefined && {
+          nextActionAt: updates.nextActionAt,
+        }),
+      },
+    });
+  }
+
+  public async delete(id: string): Promise<void> {
+    const { prismaClient } = this.connection;
+    await prismaClient.pendingL1Transaction.delete({
+      where: {
+        id,
+      },
+    });
+  }
+
+  public async findById(
+    id: string
+  ): Promise<PendingL1TransactionRecord | undefined> {
+    const { prismaClient } = this.connection;
+    const record = await prismaClient.pendingL1Transaction.findUnique({
+      where: {
+        id,
+      },
+    });
+    if (!record) {
+      return undefined;
+    }
+    return this.mapper.mapIn(record);
+  }
+
+  public async findByStatuses(
+    statuses: PendingL1TransactionStatus[]
+  ): Promise<PendingL1TransactionRecord[]> {
+    const { prismaClient } = this.connection;
+    const rows = await prismaClient.pendingL1Transaction.findMany({
+      where: {
+        status: {
+          in: statuses,
+        },
+      },
+    });
+    return rows.map((record) => this.mapper.mapIn(record));
+  }
+}

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

@@ -18,7 +18,7 @@ export class PrismaSettlementStorage implements SettlementStorage {
 
     const batch = await prismaClient.batch.findFirst({
       where: {
-        settlementTransactionHash: {
+        settlementTransactionId: {
           not: null,
         },
       },

packages/persistance/src/services/prisma/mappers/BatchMapper.ts

@@ -21,7 +21,7 @@ export class BatchMapper
     const batch: PrismaBatch = {
       proof: input.proof,
       height: input.height,
-      settlementTransactionHash: null,
+      settlementTransactionId: null,
     };
     return [batch, []];
   }

packages/persistance/src/services/prisma/mappers/PendingL1TransactionMapper.ts

@@ -0,0 +1,43 @@
+import { PendingL1Transaction, Prisma } from "@prisma/client";
+import {
+  PendingL1TransactionRecord,
+  PendingL1TransactionStatus,
+} from "@proto-kit/sequencer";
+import { Mina } from "o1js";
+
+export class PendingL1TransactionMapper {
+  public mapIn(input: PendingL1Transaction): PendingL1TransactionRecord {
+    return {
+      id: input.id,
+      sender: input.sender,
+      nonce: input.nonce,
+      attempts: input.attempts,
+      status: input.status as PendingL1TransactionStatus,
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+      transaction: Mina.Transaction.fromJSON(input.transaction as any),
+      hash: input.hash ?? undefined,
+      lastError: input.lastError ?? undefined,
+      sentAt: input.sentAt ?? undefined,
+      queuedAt: input.queuedAt ?? undefined,
+      nextActionAt: input.nextActionAt ?? undefined,
+    };
+  }
+
+  public mapOut(
+    input: Omit<PendingL1TransactionRecord, "id"> & { id?: string }
+  ): Prisma.PendingL1TransactionCreateInput {
+    return {
+      id: input.id ?? undefined,
+      sender: input.sender,
+      nonce: input.nonce,
+      attempts: input.attempts,
+      status: input.status,
+      transaction: input.transaction.toJSON(),
+      hash: input.hash ?? null,
+      lastError: input.lastError ?? null,
+      sentAt: input.sentAt ?? null,
+      queuedAt: input.queuedAt ?? null,
+      nextActionAt: input.nextActionAt ?? null,
+    };
+  }
+}

packages/persistance/src/services/prisma/mappers/SettlementMapper.ts

@@ -12,7 +12,7 @@ export class SettlementMapper
     const [settlement, batches] = input;
     return {
       batches,
-      transactionHash: settlement.transactionHash,
+      transactionId: settlement.transactionId,
       promisedMessagesHash: settlement.promisedMessagesHash,
     };
   }
@@ -21,7 +21,7 @@ export class SettlementMapper
     return [
       {
         promisedMessagesHash: input.promisedMessagesHash,
-        transactionHash: input.transactionHash,
+        transactionId: input.transactionId,
       },
       input.batches,
     ];

packages/sequencer/src/index.ts

@@ -11,7 +11,6 @@ export * from "./sequencer/builder/Closeable";
 export * from "./worker/flow/Flow";
 export * from "./worker/flow/Task";
 export * from "./worker/flow/JSONTaskSerializer";
-// export * from "./worker/queue/BullQueue";
 export * from "./worker/queue/TaskQueue";
 export * from "./worker/queue/LocalTaskQueue";
 export * from "./worker/queue/ListenerList";
@@ -71,13 +70,15 @@ export * from "./storage/repositories/BlockStorage";
 export * from "./storage/repositories/SettlementStorage";
 export * from "./storage/repositories/MessageStorage";
 export * from "./storage/repositories/TransactionStorage";
+export * from "./storage/repositories/PendingL1TransactionStorage";
 export * from "./storage/inmemory/InMemoryDatabase";
 export * from "./storage/inmemory/InMemoryAsyncMerkleTreeStore";
 export * from "./storage/inmemory/InMemoryBlockStorage";
 export * from "./storage/inmemory/InMemoryBatchStorage";
 export * from "./storage/inmemory/InMemorySettlementStorage";
 export * from "./storage/inmemory/InMemoryMessageStorage";
 export * from "./storage/inmemory/InMemoryTransactionStorage";
+export * from "./storage/inmemory/InMemoryPendingL1TransactionStorage";
 export * from "./storage/StorageDependencyFactory";
 export * from "./storage/Database";
 export * from "./storage/DatabasePruneModule";
@@ -108,6 +109,10 @@ export * from "./settlement/permissions/BaseLayerContractPermissions";
 export * from "./settlement/permissions/ProvenSettlementPermissions";
 export * from "./settlement/permissions/SignedSettlementPermissions";
 export * from "./settlement/tasks/SettlementProvingTask";
+export * from "./settlement/transactions/L1TransactionRetryStrategy";
+export * from "./settlement/transactions/DefaultL1TransactionRetryStrategy";
+export * from "./settlement/transactions/L1TransactionDispatcher";
+export * from "./settlement/transactions/TxStatusWaiter";
 export * from "./settlement/transactions/MinaTransactionSender";
 export * from "./settlement/transactions/MinaTransactionSimulator";
 export * from "./settlement/transactions/MinaSimulationService";