feat: Implement notifyConsumedBlock method in HyperionLifecycleManager for pruning checks

Author: igorls

src/indexer/modules/lifecycleManager.ts

@@ -3,6 +3,7 @@ import { HyperionConfig, NodeAttributeRequirement, TieredIndexAllocationSettings
 import { ConnectionManager } from "../connections/manager.class.js";
 import { hLog } from "../helpers/common_functions.js";
 import { HyperionMaster } from "./master.js";
+import { delay } from "lodash";
 
 interface IndexData {
     index: string;
@@ -27,6 +28,8 @@ export class HyperionLifecycleManager {
     autoPrune = false;
     totalDeletedBytes = 0;
 
+    private lastPruningCheckBlockNum = 0;
+
     constructor(master: HyperionMaster) {
         this.master = master;
         this.conf = this.master.conf;
@@ -72,6 +75,40 @@ export class HyperionLifecycleManager {
         }
     }
 
+    async notifyConsumedBlock(blockNum: number) {
+        if (!this.autoPrune) {
+            return;
+        }
+
+        const partitionSize = this.conf.settings.index_partition_size;
+        if (partitionSize <= 0) {
+            return; // Pruning is based on partitions, so this is required.
+        }
+
+        if (this.lastPruningCheckBlockNum === 0) {
+            // First block seen, initialize and return.
+            this.lastPruningCheckBlockNum = blockNum;
+            return;
+        }
+
+        // Calculate partitions based on 1-based block numbers
+        const oldPartition = Math.floor((this.lastPruningCheckBlockNum - 1) / partitionSize);
+        const newPartition = Math.floor((blockNum - 1) / partitionSize);
+
+        if (newPartition > oldPartition) {
+            hLog(`New index partition boundary crossed at block ${blockNum}. Triggering pruning check in 5 seconds...`);
+            setTimeout(() => {
+                this.checkPruning().catch((err) => {
+                    hLog(`Error during pruning check: ${err.message}`);
+                });
+            }, 5000);
+        }
+
+        if (blockNum > this.lastPruningCheckBlockNum) {
+            this.lastPruningCheckBlockNum = blockNum;
+        }
+    }
+
     startAllocationMonitoring() {
         if (!this.allocationEnabled || !this.allocation) {
             return;
@@ -128,7 +165,9 @@ export class HyperionLifecycleManager {
             bytes: 'b',
             index: `${this.master.chain}-block-${this.conf.settings.index_version}*`
         });
-        console.dir(blockIndices, { depth: Infinity });
+
+        // console.dir(blockIndices, { depth: Infinity });
+
         if (blockIndices.length === 1) {
 
             const indexName = blockIndices[0].index;
@@ -137,6 +176,8 @@ export class HyperionLifecycleManager {
                 return;
             }
 
+            hLog(`Found block index: ${indexName}`);
+
             // Get the current head block number
             const chainInfo = await this.master.rpc.v1.chain.get_info();
 
@@ -167,7 +208,7 @@ export class HyperionLifecycleManager {
                 query: { bool: { must: [{ range: { block_num: { lte: finalBlockToKeep } } }] } }
             });
 
-            console.dir(response, { depth: Infinity });
+            // console.dir(response, { depth: Infinity });
 
             const totalHits = response.hits.total as estypes.SearchTotalHits;
             if (totalHits && totalHits.value > 0) {
@@ -189,6 +230,7 @@ export class HyperionLifecycleManager {
 
         } else if (blockIndices.length > 1) {
             // Partitioned block index, we can prune by index
+            hLog(`Multiple block indices found. Pruning by index...`);
             await this.pruneIndices('block');
         }
     }

src/indexer/modules/loader.ts

@@ -1,11 +1,11 @@
-import {ConfigurationModule} from "./config.js";
-import path, {join} from "path";
-import {existsSync, readdirSync, readFileSync} from "fs";
-import {HyperionConnections} from "../../interfaces/hyperionConnections.js";
-import {HyperionConfig} from "../../interfaces/hyperionConfig.js";
-import {BaseParser} from "./parsers/base-parser.js";
-import {hLog} from "../helpers/common_functions.js";
-import {ActionTrace} from "../../interfaces/action-trace.js";
+import { ConfigurationModule } from "./config.js";
+import path, { join } from "path";
+import { existsSync, readdirSync, readFileSync } from "fs";
+import { HyperionConnections } from "../../interfaces/hyperionConnections.js";
+import { HyperionConfig } from "../../interfaces/hyperionConfig.js";
+import { BaseParser } from "./parsers/base-parser.js";
+import { hLog } from "../helpers/common_functions.js";
+import { ActionTrace } from "../../interfaces/action-trace.js";
 import { HyperionPlugin, HyperionStreamHandler } from "@eosrio/hyperion-plugin-core";
 
 export class HyperionModuleLoader {
@@ -134,7 +134,9 @@ export class HyperionModuleLoader {
     // main loader function for plugin modules
     private async loadPlugins() {
         const base = join(import.meta.dirname, '../../../', 'plugins');
-        hLog(`Loading plugins from ${base}`);
+
+        // hLog(`Loading plugins from ${base}`);
+
         if (!existsSync(base)) {
             console.error('Plugin folder not found');
             return;
@@ -155,7 +157,7 @@ export class HyperionModuleLoader {
             return;
         }
 
-        console.log(JSON.stringify(pState, null, 2));
+        // console.log(JSON.stringify(pState, null, 2));
 
         for (const key in this.config.plugins) {
             if (this.config.plugins.hasOwnProperty(key)) {

src/indexer/modules/master.ts

@@ -273,14 +273,21 @@ export class HyperionMaster {
                 }
 
                 if (msg.live === 'false') {
+
                     this.indexerMonitor.consumedBlocks++;
                     if (msg.block_num && msg.block_num > this.lastProcessedBlockNum) {
                         this.lastProcessedBlockNum = msg.block_num;
                     }
+
                 } else {
                     // LIVE READER
                     this.indexerMonitor.liveConsumedBlocks++;
 
+                    // notify the lifecycle manager
+                    this.lifecycleManager.notifyConsumedBlock(msg.block_num).catch(e => {
+                        hLog(`Error from lifecycle manager notification: ${e.message}`);
+                    });
+
                     // cache the last block number for quick api access
                     this.ioRedisClient.set(`${this.chain}:last_idx_block`, `${msg.block_num}@${msg.block_ts}`).catch(console.error);
 

src/indexer/workers/deserializer.ts

@@ -1021,6 +1021,7 @@ export default class MainDSWorker extends HyperionWorker {
             // TODO: check error on stake.libre::temppower
             if (!row.data) {
                 console.log(`Failed to deserialize ${row.code}::${row.table} at block ${block}`);
+                console.log(row);
             }
 
             // const row_sb = this.createSerialBuffer(Serialize.hexToUint8Array(row['value']));

src/indexer/workers/indexer.ts

@@ -73,7 +73,7 @@ export default class IndexerWorker extends HyperionWorker {
             const queueName = process.env.queue;
             if (this.ch && queueName) {
                 this.ch_ready = true;
-                console.log('Consumer on:', queueName);
+                hLog('Starting consumer on:', queueName);
                 this.ch.on('close', () => {
                     hLog('Channel closed for queue:', queueName);
                     this.indexQueue.pause();
@@ -94,7 +94,7 @@ export default class IndexerWorker extends HyperionWorker {
     startMonitoring() {
         setInterval(() => {
             if (this.temp_indexed_count > 0) {
-                process.send?.({event: 'add_index', size: this.temp_indexed_count});
+                process.send?.({ event: 'add_index', size: this.temp_indexed_count });
             }
             this.temp_indexed_count = 0;
         }, 1000);