Add sequencer id to job payloads

.putout.json

@@ -3,4 +3,4 @@
     "remove-nested-blocks": "off",
     "apply-shorthand-properties": "off"
   }
-}
+}

nx.json

@@ -15,23 +15,16 @@
   },
   "targetDefaults": {
     "prisma:generate": {
-      "inputs": [
-        "{projectRoot}/package.json",
-        "{projectRoot}/prisma/**/*"
-      ]
+      "inputs": ["{projectRoot}/package.json", "{projectRoot}/prisma/**/*"]
     },
     "build": {
-      "dependsOn": [
-        "^build"
-      ],
+      "dependsOn": ["^build"],
       "inputs": [
         "{projectRoot}/src/**/*",
         "{projectRoot}/package.json",
         "{projectRoot}/tsconfig.json"
       ],
-      "outputs": [
-        "{projectRoot}/dist"
-      ]
+      "outputs": ["{projectRoot}/dist"]
     }
   }
 }

packages/deployment/src/queue/BullQueue.ts

@@ -1,5 +1,5 @@
 import { MetricsTime, Queue, QueueEvents, Worker } from "bullmq";
-import { log, noop } from "@proto-kit/common";
+import { log } from "@proto-kit/common";
 import {
   TaskPayload,
   Closeable,
@@ -102,7 +102,8 @@ export class BullQueue
   }
 
   public async start() {
-    noop();
+    // Drain all queues to clear stale tasks from previous sequencer instances
+    await this.drainAllQueues();
   }
 
   public async close() {

packages/deployment/src/queue/InstantiatedBullQueue.ts

@@ -62,6 +62,10 @@ export class InstantiatedBullQueue implements InstantiatedQueue {
     this.listeners.removeListener(listenerId);
   }
 
+  async drain() {
+    await this.queue.drain();
+  }
+
   async close(): Promise<void> {
     await this.events.close();
     await this.queue.drain();

packages/indexer/src/IndexerNotifier.ts

@@ -5,6 +5,7 @@ import {
   SequencerModule,
   TaskPayload,
   TaskQueue,
+  SequencerIdProvider,
 } from "@proto-kit/sequencer";
 import { log } from "@proto-kit/common";
 import { inject } from "tsyringe";
@@ -22,7 +23,8 @@ export class IndexerNotifier extends SequencerModule<Record<never, never>> {
     public sequencer: Sequencer<NotifierMandatorySequencerModules>,
     @inject("TaskQueue")
     public taskQueue: TaskQueue,
-    public indexBlockTask: IndexBlockTask
+    public indexBlockTask: IndexBlockTask,
+    private readonly sequencerIdProvider: SequencerIdProvider
   ) {
     super();
   }
@@ -37,11 +39,13 @@ export class IndexerNotifier extends SequencerModule<Record<never, never>> {
         block.block.height.toBigInt()
       );
       const payload = await inputSerializer.toJSON(block);
+      const sequencerId = this.sequencerIdProvider.getSequencerId();
 
       const task: TaskPayload = {
         name: this.indexBlockTask.name,
         payload,
         flowId: "", // empty for now
+        sequencerId,
       };
 
       await queue.addTask(task);

packages/indexer/test/IndexBlockTask.test.ts

@@ -41,6 +41,7 @@ describe("IndexBlockTask", () => {
       name: indexBlockTask.name,
       payload,
       flowId: "",
+      sequencerId: "test-sequencer",
     };
 
     await queue.addTask(task);

packages/indexer/test/IndexerNotifier.test.ts

@@ -166,6 +166,7 @@ describe.skip("IndexerNotifier", () => {
       addTask: addTaskSpy,
       onCompleted: jest.fn(async () => 5),
       close: jest.fn(async () => {}),
+      drain: jest.fn(async () => {}),
     };
   });
 

packages/sequencer/src/index.ts

@@ -5,6 +5,7 @@ export * from "./mempool/CompressedSignature";
 export * from "./mempool/private/PrivateMempool";
 export * from "./sequencer/executor/Sequencer";
 export * from "./sequencer/executor/Sequenceable";
+export * from "./sequencer/SequencerIdProvider";
 export * from "./sequencer/builder/SequencerModule";
 export * from "./sequencer/builder/Closeable";
 export * from "./worker/flow/Flow";

packages/sequencer/src/sequencer/SequencerIdProvider.ts

@@ -0,0 +1,16 @@
+import { injectable, singleton } from "tsyringe";
+import { Field } from "o1js";
+
+@injectable()
+@singleton()
+export class SequencerIdProvider {
+  private readonly sequencerId: string;
+
+  public constructor() {
+    this.sequencerId = Field.random().toString();
+  }
+
+  public getSequencerId(): string {
+    return this.sequencerId;
+  }
+}

packages/sequencer/src/sequencer/executor/Sequencer.ts

@@ -34,6 +34,10 @@ export class Sequencer<Modules extends SequencerModulesRecord>
   extends ModuleContainer<Modules>
   implements Sequenceable
 {
+  public constructor(definition: Modules) {
+    super(definition);
+  }
+
   /**
    * Alternative constructor for Sequencer
    * @param definition

packages/sequencer/src/worker/flow/Flow.ts

@@ -3,6 +3,7 @@ import { log, mapSequential } from "@proto-kit/common";
 
 import { InstantiatedQueue, TaskQueue } from "../queue/TaskQueue";
 import { Closeable } from "../../sequencer/builder/Closeable";
+import { SequencerIdProvider } from "../../sequencer/SequencerIdProvider";
 
 import { Task, TaskPayload } from "./Task";
 
@@ -42,7 +43,8 @@ export class Flow<State> implements Closeable {
   public constructor(
     private readonly queueImpl: TaskQueue,
     public readonly flowId: string,
-    public state: State
+    public state: State,
+    public readonly sequencerId: string
   ) {}
 
   private async waitForResult(
@@ -82,6 +84,11 @@ export class Flow<State> implements Closeable {
       const resolveFunction = this.resultsPending[response.taskId];
 
       if (!this.erroredOut) {
+        // skip responses from old sequencers
+        if (response.sequencerId !== this.sequencerId) {
+          return;
+        }
+
         if (response.status === "error") {
           this.reject(
             new Error(
@@ -123,6 +130,7 @@ export class Flow<State> implements Closeable {
       taskId,
       flowId: this.flowId,
       payload,
+      sequencerId: this.sequencerId,
     });
 
     this.tasksInProgress += 1;
@@ -177,10 +185,16 @@ export class Flow<State> implements Closeable {
 @injectable()
 export class FlowCreator {
   public constructor(
-    @inject("TaskQueue") private readonly queueImpl: TaskQueue
+    @inject("TaskQueue") private readonly queueImpl: TaskQueue,
+    private readonly sequencerIdProvider: SequencerIdProvider
   ) {}
 
   public createFlow<State>(flowId: string, state: State): Flow<State> {
-    return new Flow(this.queueImpl, flowId, state);
+    return new Flow(
+      this.queueImpl,
+      flowId,
+      state,
+      this.sequencerIdProvider.getSequencerId()
+    );
   }
 }

packages/sequencer/src/worker/flow/Task.ts

@@ -20,4 +20,5 @@ export interface TaskPayload {
   payload: string;
   taskId?: string;
   flowId: string;
+  sequencerId: string;
 }

packages/sequencer/src/worker/queue/AbstractTaskQueue.ts

@@ -22,4 +22,10 @@ export abstract class AbstractTaskQueue<
       Object.values(this.queues).map(async (queue) => await queue.close())
     );
   }
+
+  public async drainAllQueues(): Promise<void> {
+    await Promise.all(
+      Object.values(this.queues).map(async (queue) => await queue.drain())
+    );
+  }
 }

packages/sequencer/src/worker/queue/LocalTaskQueue.ts

@@ -68,6 +68,10 @@ class InMemoryInstantiatedQueue implements InstantiatedQueue {
   async close() {
     noop();
   }
+
+  async drain() {
+    this.taskQueue.queuedTasks[this.name] = [];
+  }
 }
 
 @sequencerModule()

packages/sequencer/src/worker/queue/TaskQueue.ts

@@ -13,6 +13,8 @@ export interface TaskQueue {
     executor: (data: TaskPayload) => Promise<TaskPayload>,
     options?: { concurrency?: number }
   ) => Closeable;
+
+  drainAllQueues: () => Promise<void>;
 }
 /**
  * Object that abstracts a concrete connection to a queue instance.
@@ -36,4 +38,9 @@ export interface InstantiatedQueue extends Closeable {
   ) => Promise<number>;
 
   offCompleted: (listenerId: number) => void;
+
+  /**
+   * Drains the queue to clear stale tasks
+   */
+  drain: () => Promise<void>;
 }

packages/sequencer/src/worker/worker/FlowTaskWorker.ts

@@ -50,6 +50,7 @@ export class FlowTaskWorker<Tasks extends Task<any, any>[]>
           flowId: data.flowId,
           name: data.name,
           payload: await task.resultSerializer().toJSON(output),
+          sequencerId: data.sequencerId,
         };
 
         log.debug(
@@ -70,6 +71,7 @@ export class FlowTaskWorker<Tasks extends Task<any, any>[]>
           flowId: data.flowId,
           name: data.name,
           payload,
+          sequencerId: data.sequencerId,
         };
       }
     });

packages/sequencer/test/worker/LocalTaskQueue.test.ts

@@ -38,6 +38,7 @@ describe("localTaskQueue", () => {
           name: String(index),
           payload: String(input),
           flowId: "0",
+          sequencerId: "0",
         });
       }
 
@@ -48,6 +49,7 @@ describe("localTaskQueue", () => {
           name: String(index),
           payload: String(input),
           flowId: "0",
+          sequencerId: "0",
         });
       }
     }