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/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

@@ -37,11 +37,13 @@ export class IndexerNotifier extends SequencerModule<Record<never, never>> {
         block.block.height.toBigInt()
       );
       const payload = await inputSerializer.toJSON(block);
+      const sequencerId = this.sequencer.id;
 
       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/sequencer/executor/Sequencer.ts

@@ -17,11 +17,13 @@ import {
   ProtocolModulesRecord,
 } from "@proto-kit/protocol";
 import { injectable } from "tsyringe";
+import { Field } from "o1js";
 
 import { SequencerModule } from "../builder/SequencerModule";
 import { Closeable } from "../builder/Closeable";
 import { ConsoleTracingFactory } from "../../logging/ConsoleTracingFactory";
 import { StartableModule } from "../builder/StartableModule";
+import { TaskQueue } from "../../worker/queue/TaskQueue";
 
 import { Sequenceable } from "./Sequenceable";
 
@@ -34,6 +36,13 @@ export class Sequencer<Modules extends SequencerModulesRecord>
   extends ModuleContainer<Modules>
   implements Sequenceable
 {
+  public readonly id: string;
+
+  public constructor(definition: Modules) {
+    super(definition);
+    this.id = Field.random().toString();
+  }
+
   /**
    * Alternative constructor for Sequencer
    * @param definition
@@ -81,6 +90,12 @@ export class Sequencer<Modules extends SequencerModulesRecord>
 
     this.useDependencyFactory(MethodIdFactory);
 
+    // Drain all task queues to clear stale tasks from previous sequencer instances
+    if (this.container.isRegistered("TaskQueue")) {
+      const taskQueue = this.container.resolve<TaskQueue>("TaskQueue");
+      await taskQueue.drainAllQueues();
+    }
+
     // Log startup info
     const moduleClassNames = Object.values(this.definition).map(
       (clazz) => clazz.name

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 type { Sequencer } from "../../sequencer/executor/Sequencer";
 
 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,11 @@ export class Flow<State> implements Closeable {
 @injectable()
 export class FlowCreator {
   public constructor(
-    @inject("TaskQueue") private readonly queueImpl: TaskQueue
+    @inject("TaskQueue") private readonly queueImpl: TaskQueue,
+    @inject("Sequencer") private readonly sequencer: Sequencer<any>
   ) {}
 
   public createFlow<State>(flowId: string, state: State): Flow<State> {
-    return new Flow(this.queueImpl, flowId, state);
+    return new Flow(this.queueImpl, flowId, state, this.sequencer.id);
   }
 }

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/protocol/production/flow/ReductionTaskFlow.test.ts

@@ -90,6 +90,7 @@ describe("ReductionTaskFlow", () => {
     di.register("TaskQueue", {
       useValue: queue,
     });
+    di.register("Sequencer", { useValue: { id: "1" } });
 
     const worker = new FlowTaskWorker(di.resolve("TaskQueue"), [
       di.resolve(NumberIdentityTask),

packages/sequencer/test/worker/Flow.test.ts

@@ -226,6 +226,7 @@ describe("flow", () => {
       await worker.start();
 
       container.register("TaskQueue", { useValue: queue });
+      container.register("Sequencer", { useValue: { id: "1" } });
       const flowCreator = container.resolve(FlowCreator);
 
       const flow = flowCreator.createFlow("1", {

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",
         });
       }
     }