draining queues in sequencer.start

Author: Raunaque97

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

@@ -23,6 +23,7 @@ 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";
 
@@ -89,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/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>;
 }