Address comments

subquery · Dec 3, 2024 · 33221f7 · 33221f7
1 parent a858912
commit 33221f7
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/packages/node-core/src/indexer/blockDispatcher/block-dispatcher.ts b/packages/node-core/src/indexer/blockDispatcher/block-dispatcher.ts
@@ -16,7 +16,6 @@ import {StoreService} from '../store.service';
 import {IStoreModelProvider} from '../storeModelProvider';
 import {IProjectService, ISubqueryProject} from '../types';
 import {BaseBlockDispatcher, ProcessBlockResponse} from './base-block-dispatcher';
-// import { RampQueue } from '@subql/node-core/utils/rampQueue';
 
 const logger = getLogger('BlockDispatcherService');
 

diff --git a/packages/node-core/src/utils/queues/rampQueue.ts b/packages/node-core/src/utils/queues/rampQueue.ts
@@ -23,6 +23,7 @@ const logger = getLogger('RampQueue');
 export class RampQueue<T> extends AutoQueue<T> {
   #maxConcurrency: number;
   #sizes: number[] = [];
+  #totalItems = 0;
 
   constructor(
     private getSize: (data: T) => number,
@@ -72,9 +73,9 @@ export class RampQueue<T> extends AutoQueue<T> {
 
       if (size > m * 2) {
         // Inverse of the size compared to the median. E.g if a block is 5x as big as the median then the batch size should be 1/5 of the max
-        const multiplier = 1 / (size / m);
+        const multiplier = m / size;
         this.setConcurrency(this.#maxConcurrency * multiplier);
-      } else if (this.#sizes.length % MIN_SIZES === 0) {
+      } else if (this.#totalItems % MIN_SIZES === 0) {
         // Increase by 10% of max
         this.setConcurrency(this.concurrency + Math.floor(this.#maxConcurrency / 10));
       }
@@ -87,6 +88,7 @@ export class RampQueue<T> extends AutoQueue<T> {
     if (this.#sizes.length >= MAX_SIZES) {
       this.#sizes.shift();
     }
+    this.#totalItems++;
     this.#sizes.push(size);
   }
 }