Add a Queue for Ogmios Requests

package.json

@@ -53,6 +53,7 @@
     "lucid-cardano": "^0.10.7",
     "mysql2": "^3.6.0",
     "node-cache": "^5.1.2",
+    "p-queue": "^8.0.1",
     "pg": "^8.11.5",
     "queue-promise": "^2.2.1",
     "reflect-metadata": "^0.1.13",

src/IndexerApplication.ts

@@ -29,6 +29,7 @@ import { VyFiAnalyzer } from './dex/VyFiAnalyzer';
 import { ChainSynchronization } from '@cardano-ogmios/client';
 import { MinswapV2Analyzer } from './dex/MinswapV2Analyzer';
 import { SundaeSwapV3Analyzer } from './dex/SundaeSwapV3Analyzer';
+import { QueueProcessor } from './utils';
 
 export class IndexerApplication {
 
@@ -130,6 +131,8 @@ export class IndexerApplication {
         });
     }
 
+    blockQueue = new QueueProcessor(5000, (block) => this.rollForward(block), (block) => this.rollBackward(block));
+
     /**
      * Boot Ogmios connection.
      */
@@ -162,8 +165,8 @@ export class IndexerApplication {
         this._chainSyncClient = await createChainSynchronizationClient(
             this._ogmiosContext,
             {
-                rollForward: this.rollForward.bind(this),
-                rollBackward: this.rollBackward.bind(this),
+                rollForward: (response, nextBlock) => this.blockQueue.enqueue(response, nextBlock),
+                rollBackward: (response, nextBlock) => this.blockQueue.enqueue(response, nextBlock),
             },
             {
                 sequential: true,
@@ -198,7 +201,7 @@ export class IndexerApplication {
      * @param update - New block update.
      * @param requestNext - Callback to request next block.
      */
-    private async rollForward(update: { block: Block, tip: TipOrOrigin }, requestNext: () => void): Promise<void> {
+    private async rollForward(update: { block: Block, tip: TipOrOrigin }): Promise<void> {
         if (update.block.type === 'praos') {
             const block: BlockPraos = update.block;
 
@@ -216,16 +219,14 @@ export class IndexerApplication {
 
             logInfo(`====== Finished with block at slot ${block.slot} ======`);
         }
-
-        requestNext();
     }
 
     /**
      * Handler for Ogmios rollback. Send to all indexers.
      * @param update - Point in which to revert to.
      * @param requestNext - Callback to request next block.
      */
-    private async rollBackward(update: { point: PointOrOrigin }, requestNext: () => void): Promise<void> {
+    private async rollBackward(update: { point: PointOrOrigin }): Promise<void> {
         if (typeof update.point === 'object' && 'slot' in update.point) {
             logInfo(`Rollback occurred to slot ${update.point.slot}`);
 
@@ -235,8 +236,6 @@ export class IndexerApplication {
                 this._indexers.map((indexer: BaseIndexer) => indexer.onRollBackward(point.id, point.slot)),
             );
         }
-
-        requestNext();
     }
 
 }

src/utils.ts

@@ -11,11 +11,17 @@ import { Lucid, Utils } from 'lucid-cardano';
 import { Asset, Token } from './db/entities/Asset';
 import { LiquidityPool } from './db/entities/LiquidityPool';
 import {
+    Block,
     BlockPraos,
     Transaction as OgmiosTransaction,
+    Origin,
+    PointOrOrigin,
+    Tip,
     TransactionOutput,
     TransactionOutputReference
 } from '@cardano-ogmios/schema';
+import PQueue from 'p-queue';
+import { logError, logInfo } from './logger';
 
 export const lucidUtils: Utils = new Utils(new Lucid());
 
@@ -149,3 +155,52 @@ export function formatTransaction(block: BlockPraos | null, transaction: OgmiosT
         scriptHashes: Object.keys(transaction.scripts ?? {}),
     };
 }
+
+type ForwardBlock = { block: Block, tip: Tip | Origin};
+type BackwardBlock = { point: PointOrOrigin };
+
+export class QueueProcessor {
+    private queue: PQueue; // Use p-queue for queue management
+    private maxSize: number;
+
+    constructor(
+        maxSize: number,
+        private rollForward: (update: ForwardBlock) => Promise<void>,
+        private rollBackward: (update: BackwardBlock) => Promise<void>
+    ) {
+        this.maxSize = maxSize;
+        this.queue = new PQueue({
+            concurrency: 1, // Ensure sequential processing
+            autoStart: true, // Automatically start processing tasks
+        });
+    }
+
+    async enqueue(response: ForwardBlock | BackwardBlock, requestNext: () => void): Promise<void> {
+        try {
+            // Add the task to the queue
+            this.queue.add(async () => {
+                if ('block' in response) {
+                    await this.rollForward(response);
+                } else {
+                    await this.rollBackward(response);
+                }
+            });
+
+            // Call next block when there's room
+            await this.queue.onSizeLessThan(this.maxSize)
+            requestNext();
+        } catch (error) {
+            logError(`BlockQueue.enqueue error: ${error}`);
+        }
+    }
+
+    async stopProcessing(): Promise<void> {
+        logInfo('Stopping Queue Processor...');
+        this.queue.pause(); // Pause the queue
+        await this.queue.onIdle(); // Wait until all tasks are finished
+    }
+
+    queueSize(): number {
+        return this.queue.size;
+    }
+}