Feat/job config

migrations/20240419100359_config_job.ts

@@ -0,0 +1,16 @@
+import { Knex } from 'knex';
+import { ConfigJob } from '../src/models';
+
+export async function up(knex: Knex): Promise<void> {
+  await knex.schema.createTable(ConfigJob.tableName, (table) => {
+    table.increments();
+    table.string('job_name').index().notNullable();
+    table.integer('expected_row').notNullable();
+    table.integer('block_range').notNullable();
+    table.integer('acceptance_error').notNullable();
+    table.integer('block_balance').notNullable();
+    table.boolean('is_sync').defaultTo(false).notNullable();
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {}

src/models/block_checkpoint.ts

@@ -33,7 +33,7 @@ export class BlockCheckpoint extends BaseModel {
     jobName: string,
     lastHeightJobNames: string[],
     configName?: string
-  ): Promise<[number, number, BlockCheckpoint]> {
+  ): Promise<[number, number, BlockCheckpoint, BlockCheckpoint | undefined]> {
     const [jobCheckpoint, lastHeightCheckpoint] = await Promise.all([
       BlockCheckpoint.query().select('*').where('job_name', jobName).first(),
       BlockCheckpoint.query()
@@ -64,6 +64,11 @@ export class BlockCheckpoint extends BaseModel {
           )
         : lastHeightCheckpoint.height;
 
-    return [startHeight, endHeight, updateBlockCheckpoint];
+    return [
+      startHeight,
+      endHeight,
+      updateBlockCheckpoint,
+      lastHeightCheckpoint,
+    ];
   }
 }

src/models/config_job.ts

@@ -0,0 +1,115 @@
+import BaseModel from './base';
+import { BlockCheckpoint } from './block_checkpoint';
+
+export class ConfigJob extends BaseModel {
+  id!: number;
+
+  job_name!: string;
+
+  // Expected row get for handle when each time job run
+  expected_row!: number;
+
+  // Range query by block
+  block_range!: number;
+
+  // ABS(rows retrieve - expected_row) should less than acceptance_error, if not do the block_balance below
+  acceptance_error!: number;
+
+  // When get rows by block range not fit with expected_row, then increase/decrease block_range for fit with
+  block_balance!: number;
+
+  // Keep job run normal or use config job and sync
+  is_sync!: boolean;
+
+  static get tableName() {
+    return 'config_jobs';
+  }
+
+  /**
+   * @description:
+   * @param jobName Job name want to query to get config
+   * job crawl block complete on that block we want to crawl
+   */
+  static async getConfigJob(jobName: string): Promise<ConfigJob | undefined> {
+    return ConfigJob.query().select('*').where('job_name', jobName).first();
+  }
+
+  // Range job will from currentStartBlock to bestEndBlock
+  static determineBestRangeBlockForRunJob(
+    currentStartBlock: number,
+    currentEndBlock: number,
+    dependingCheckPointJob: BlockCheckpoint | undefined,
+    configJob: ConfigJob | undefined
+  ): number {
+    let bestEndBlock = currentEndBlock;
+
+    // No config job, then use default config from block checkpoint
+    if (!configJob || !configJob.is_sync) return bestEndBlock;
+
+    if (dependingCheckPointJob) {
+      if (
+        currentStartBlock + configJob.block_range <=
+        dependingCheckPointJob.height
+      ) {
+        bestEndBlock = currentStartBlock + configJob.block_range;
+      } else {
+        bestEndBlock = dependingCheckPointJob.height;
+      }
+    } else {
+      bestEndBlock = currentStartBlock + configJob.block_range;
+    }
+
+    return bestEndBlock;
+  }
+
+  // Balance config job
+  static prepareBalanceJob(
+    bestEndBlock: number,
+    dependingCheckPointJob: BlockCheckpoint | undefined,
+    configJob: ConfigJob | undefined,
+    currentTotalRowFetch: number
+  ): ConfigJob | null {
+    // No job config so keep every thing normal
+    if (!configJob || !configJob.is_sync) return null;
+
+    // Job depending on another job, so keep current config job
+    if (bestEndBlock === dependingCheckPointJob?.height) return null;
+
+    const errorRow = configJob.expected_row - currentTotalRowFetch;
+    // error acceptance
+    if (Math.abs(errorRow) <= configJob.acceptance_error) return null;
+
+    if (errorRow > 0) {
+      // increase range block if total row fetch too low with expected row
+      // eslint-disable-next-line no-param-reassign
+      configJob.block_range += configJob.block_balance;
+    } else {
+      // decrease range block if total row fetch too low with expected row
+      // eslint-disable-next-line no-param-reassign
+      configJob.block_range -= configJob.block_balance;
+    }
+
+    return configJob;
+  }
+
+  static get jsonSchema() {
+    return {
+      type: 'object',
+      required: [
+        'job_name',
+        'expected_row',
+        'block_range',
+        'acceptance_error',
+        'block_balance',
+      ],
+      properties: {
+        id: { type: 'number' },
+        job_name: { type: 'string' },
+        expected_row: { type: 'number' },
+        block_range: { type: 'number' },
+        acceptance_error: { type: 'number' },
+        block_balance: { type: 'number' },
+      },
+    };
+  }
+}

src/models/index.ts

@@ -41,3 +41,4 @@ export * from './erc20_activity';
 export * from './evm_proxy_history';
 export * from './evm_internal_transaction';
 export * from './account_balance';
+export * from './config_job';

src/services/crawl-tx/crawl_tx.service.ts

@@ -22,6 +22,7 @@ import {
 import {
   Block,
   BlockCheckpoint,
+  ConfigJob,
   Event,
   Transaction,
   TransactionMessage,
@@ -85,7 +86,7 @@ export default class CrawlTxService extends BullableService {
     jobName: BULL_JOB_NAME.HANDLE_TRANSACTION,
   })
   public async jobHandlerCrawlTx(): Promise<void> {
-    const [startBlock, endBlock, blockCheckpoint] =
+    const [startBlock, endBlock, blockCheckpoint, dependBlockCheckPoint] =
       await BlockCheckpoint.getCheckpoint(
         BULL_JOB_NAME.HANDLE_TRANSACTION,
         [BULL_JOB_NAME.CRAWL_TRANSACTION],
@@ -99,11 +100,30 @@ export default class CrawlTxService extends BullableService {
     if (startBlock >= endBlock) {
       return;
     }
+
+    const configJob = await ConfigJob.getConfigJob(
+      BULL_JOB_NAME.HANDLE_TRANSACTION
+    );
+    const bestEndBlock = ConfigJob.determineBestRangeBlockForRunJob(
+      startBlock,
+      endBlock,
+      dependBlockCheckPoint,
+      configJob
+    );
+
     const listTxRaw = await Transaction.query()
       .where('height', '>', startBlock)
-      .andWhere('height', '<=', endBlock)
+      .andWhere('height', '<=', bestEndBlock)
       .orderBy('height', 'asc')
       .orderBy('index', 'asc');
+
+    const balanceConfigJob = ConfigJob.prepareBalanceJob(
+      bestEndBlock,
+      dependBlockCheckPoint,
+      configJob,
+      listTxRaw.length
+    );
+
     await knex.transaction(async (trx) => {
       await this.insertRelatedTx(listTxRaw, trx);
       if (blockCheckpoint) {
@@ -115,6 +135,14 @@ export default class CrawlTxService extends BullableService {
           .returning('id')
           .transacting(trx);
       }
+      if (balanceConfigJob) {
+        await ConfigJob.query()
+          .insert(balanceConfigJob)
+          .onConflict('job_name')
+          .merge()
+          .returning('id')
+          .transacting(trx);
+      }
     });
   }