cleanup

This reverts commit 0579da66

api/helpers/cacheManager.ts

@@ -28,19 +28,19 @@ export class CacheManager {
         setInterval(() => {
             try {
                 // remove expired entries
-                let removeCount = 0;
+                // let removeCount = 0;
                 const now = Date.now();
                 this.v1Caches.forEach((pathCacheMap: Map<string, CachedEntry>, pathKey: string) => {
                     pathCacheMap.forEach((cache: CachedEntry, entryKey: string, map: Map<string, CachedEntry>) => {
                         if (cache.exp < now) {
                             map.delete(entryKey);
-                            removeCount++;
+                            // removeCount++;
                         }
                     });
                 });
-                if (removeCount > 0) {
-                    console.log(`${removeCount} expired cache entries removed`);
-                }
+                // if (removeCount > 0) {
+                //     console.log(`${removeCount} expired cache entries removed`);
+                // }
             } catch (e) {
                 console.log(e);
             }

api/helpers/functions.ts

@@ -584,7 +584,7 @@ export function chainApiHandler(fastify: FastifyInstance) {
         // check cache
         const [cachedData, hash, path] = fastify.cacheManager.getCachedData(request);
         if (cachedData) {
-            // console.log('cache hit:', path, hash);
+            console.log('cache hit:', path, hash);
             reply.headers({'hyperion-cached': true}).send(cachedData);
         } else {
             // call actual request

api/routes.ts

@@ -62,7 +62,7 @@ export function registerRoutes(server: FastifyInstance) {
     // chain api redirects
     addRoute(server, 'v1-chain', '/v1/chain');
 
-	// other v1 requests
+    // other v1 requests
     server.route({
         url: '/v1/chain/*',
         method: ["GET", "POST"],
@@ -95,6 +95,25 @@ export function registerRoutes(server: FastifyInstance) {
         }
     });
 
+    // global onRequest hook to collect hourly statistics for each path on redis
+    server.addHook('onResponse', (request: FastifyRequest, reply: FastifyReply, done) => {
+        const path = request.url.split('?')[0].replace(/\/$/, '');
+        const now = new Date();
+        const ts = now.getTime();
+        // normalized time in hours
+        const factor = 1000 * 60 * 60;
+        const ts_hour_unix = Math.floor(ts / factor) * factor;
+        const key = `stats:${server.manager.chain}:H:${ts_hour_unix}`;
+        // console.log(reply.statusCode, key, new Date(ts_hour_unix));
+        server.redis.multi()
+            .incr(`stats:${server.manager.chain}:total:${reply.statusCode}:${path}`)
+            .zincrby(key, 1, `[${reply.statusCode}]${path}`)
+            .expire(key, 60 * 60 * 24 * 7)
+            .exec()
+            .catch(console.log);
+        done();
+    });
+
     server.addHook('onError', (request: FastifyRequest, reply: FastifyReply, error: FastifyError, done) => {
         console.log(`[${request.headers['x-real-ip'] || request.ip}] ${request.method} ${request.url} failed >> ${error.message}`);
         done();

api/routes/v1-chain/abi_bin_to_json/index.ts

@@ -9,11 +9,7 @@ export default function (fastify: FastifyInstance, opts: any, next) {
         {
             "code": {$ref: 'AccountName#'},
             "action": {$ref: 'AccountName#'},
-            "binargs": {
-                "type": "string",
-                "pattern": "^(0x)(([0-9a-f][0-9a-f])+)?$",
-                "title": "Hex"
-            }
+            "binargs": {"type": "string"}
         },
         ["code", "action", "binargs"]
     );

api/routes/v1-chain/abi_json_to_bin/index.ts

@@ -7,13 +7,14 @@ export default function (fastify: FastifyInstance, opts: any, next) {
         getRouteName(__filename),
         'Convert JSON object to binary',
         {
-            "binargs": {
-                "type": "string",
-                "pattern": "^(0x)(([0-9a-f][0-9a-f])+)?$",
-                "title": "Hex"
+            "code": {$ref: 'AccountName#'},
+            "action": {$ref: 'AccountName#'},
+            "args": {
+                type: 'object',
+                additionalProperties: true
             }
         },
-        ["binargs"]
+        ["code", "action", "args"]
     );
     next();
 }

api/routes/v1-chain/get_table_rows/index.ts

@@ -37,7 +37,7 @@ export default function (fastify: FastifyInstance, opts: any, next) {
                 "type": "string"
             }
         },
-        ["code", "table"]
+        ["code", "table", "scope"]
     );
     next();
 }

api/routes/v2-stats/get_api_usage/get_api_usage.ts

@@ -0,0 +1,53 @@
+import {FastifyInstance, FastifyReply, FastifyRequest} from "fastify";
+import {timedQuery} from "../../../helpers/functions";
+import {parseInt} from "lodash";
+
+async function getApiUsage(fastify: FastifyInstance, request: FastifyRequest) {
+    const response = {
+        total: {responses: {}},
+        buckets: []
+    } as any;
+    const data = await fastify.redis.keys(`stats:${fastify.manager.chain}:*`);
+    const buckets = [];
+    for (const key of data) {
+        const parts = key.split(':');
+        if (parts[2] === 'H') {
+            const bucket = {
+                timestamp: new Date(parseInt(parts[3])),
+                responses: {}
+            };
+            const sortedSet = await fastify.redis.zrange(key, 0, -1, 'WITHSCORES');
+            for (let i = 0; i < sortedSet.length; i += 2) {
+                const [status, path] = sortedSet[i].split(']');
+                if (status) {
+                    const statusCode = status.slice(1);
+                    if (!bucket.responses[statusCode]) {
+                        bucket.responses[statusCode] = {};
+                    }
+                    if (statusCode) {
+                        bucket.responses[statusCode][path] = parseInt(sortedSet[i + 1]);
+                    }
+                }
+            }
+            response.buckets.push(bucket);
+        } else {
+            if (parts[2] === 'total' && parts[4]) {
+                const statusCode = parts[3];
+                if (!response.total.responses[statusCode]) {
+                    response.total.responses[statusCode] = {};
+                }
+                const value = await fastify.redis.get(key);
+                response.total.responses[statusCode][parts[4]] = parseInt(value);
+            }
+        }
+    }
+    // reverse sorte by date
+    response.buckets.sort((a, b) => b.timestamp.getTime() - a.timestamp.getTime());
+    return response;
+}
+
+export function getApiUsageHandler(fastify: FastifyInstance, route: string) {
+    return async (request: FastifyRequest, reply: FastifyReply) => {
+        reply.send(await timedQuery(getApiUsage, fastify, request, route));
+    }
+}

api/routes/v2-stats/get_api_usage/index.ts

@@ -0,0 +1,41 @@
+import {FastifyInstance} from "fastify";
+import {addApiRoute, extendResponseSchema, getRouteName} from "../../../helpers/functions";
+import {getApiUsageHandler} from "./get_api_usage";
+
+export default function (fastify: FastifyInstance, opts: any, next) {
+    const schema = {
+        description: 'get hyperion api usage statistics',
+        summary: 'get hyperion api usage statistics',
+        tags: ['stats'],
+        response: extendResponseSchema({
+            total: {
+                type: 'object',
+                properties: {
+                    responses: {
+                        additionalProperties: true
+                    }
+                }
+            },
+            buckets: {
+                type: 'array',
+                items: {
+                    type: 'object',
+                    properties: {
+                        timestamp: {type: 'string'},
+                        responses: {
+                            additionalProperties: true
+                        }
+                    }
+                }
+            }
+        })
+    };
+    addApiRoute(
+        fastify,
+        'GET',
+        getRouteName(__filename),
+        getApiUsageHandler,
+        schema
+    );
+    next();
+}

api/server.ts

@@ -16,7 +16,6 @@ import {io, Socket} from "socket.io-client";
 import {CacheManager} from "./helpers/cacheManager";
 
 import {bootstrap} from 'global-agent';
-import {Api} from "eosjs";
 
 class HyperionApiServer {
 
@@ -227,8 +226,8 @@ class HyperionApiServer {
             hLog('Failed to check elasticsearch version!');
             process.exit();
         });
-        hLog('Elasticsearch validated!');
 
+        hLog('Elasticsearch validated!');
         hLog('Registering plugins...');
 
         registerPlugins(this.fastify, this.pluginParams);

connections/amqp.ts

@@ -17,17 +17,16 @@ export async function createConnection(config): Promise<Connection> {
 }
 
 export function getAmpqUrl(config): string {
-	let frameMaxValue = '0x10000';
-	if (config.frameMax) {
-		frameMaxValue = config.frameMax;
-	}
-	const u = encodeURIComponent(config.user);
-	const p = encodeURIComponent(config.pass);
-	const v = encodeURIComponent(config.vhost)
-	return `amqp://${u}:${p}@${config.host}/${v}?frameMax=${frameMaxValue}`;
+    let frameMaxValue = '0x10000';
+    if (config.frameMax) {
+        frameMaxValue = config.frameMax;
+    }
+    const u = encodeURIComponent(config.user);
+    const p = encodeURIComponent(config.pass);
+    const v = encodeURIComponent(config.vhost);
+    return `amqp://${u}:${p}@${config.host}/${v}?frameMax=${frameMaxValue}`;
 }
 
-
 async function createChannels(connection) {
 	try {
 		const channel = await connection.createChannel();

connections/manager.class.ts

@@ -9,7 +9,6 @@ import {StateHistorySocket} from "./state-history";
 import fetch from 'cross-fetch';
 import {exec} from "child_process";
 import {hLog} from "../helpers/common_functions";
-import {readFileSync} from "fs";
 
 export class ConnectionManager {
 
@@ -20,8 +19,8 @@ export class ConnectionManager {
     last_commit_hash: string;
     current_version: string;
 
-    private readonly esIngestClients: Client[];
-    private esIngestClient: Client;
+    esIngestClients: Client[];
+    esIngestClient: Client;
 
     constructor(private cm: ConfigurationModule) {
         this.config = cm.config;

ecosystem.config.js

@@ -20,4 +20,11 @@ readdirSync(chainsRoot)
         }
     });
 
+apps.push({
+    name: 'hyperion-governor',
+    namespace: 'hyperion',
+    script: 'governor/server/index.js',
+    watch: false,
+});
+
 module.exports = {apps};

modules/parsers/base-parser.ts

@@ -192,7 +192,6 @@ export abstract class BaseParser {
         if (ds_act) {
 
             if (ds_act.account && ds_act.name && ds_act.authorization) {
-                console.log(ds_act);
                 action.act.data = ds_act.data;
             }
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hyperion-history",
-  "version": "3.3.9-6",
+  "version": "3.3.9-8",
   "description": "Scalable Full History API Solution for EOSIO based blockchains",
   "main": "launcher.js",
   "scripts": {
@@ -26,52 +26,52 @@
   },
   "license": "MIT",
   "dependencies": {
-    "@elastic/elasticsearch": "^8.8.0",
-    "@eosrio/node-abieos": "3.2.0",
-    "@fastify/autoload": "5.7.1",
-    "@fastify/cors": "8.3.0",
-    "@fastify/formbody": "^7.4.0",
-    "@fastify/rate-limit": "^8.0.0",
-    "@fastify/redis": "^6.1.1",
-    "@fastify/swagger": "8.5.1",
+    "@elastic/elasticsearch": "^7.17.0",
+    "@eosrio/node-abieos": "2.1.1",
+    "@fastify/autoload": "4.0.1",
+    "@fastify/cors": "7.0.0",
+    "@fastify/formbody": "^6.0.1",
+    "@fastify/rate-limit": "^6.0.1",
+    "@fastify/redis": "^5.0.0",
+    "@fastify/swagger": "6.1.0",
     "@pm2/io": "^5.0.0",
     "amqplib": "^0.10.3",
     "async": "^3.2.4",
     "base-x": "^4.0.0",
     "cli-progress": "^3.12.0",
-    "commander": "^10.0.1",
-    "cross-fetch": "^3.1.6",
+    "commander": "^8.3.0",
+    "cross-fetch": "^3.1.8",
     "eosjs": "^22.1.0",
-    "fast-json-stringify": "5.7.0",
-    "fastify": "^4.17.0",
+    "fast-json-stringify": "2.7.13",
+    "fastify": "3.29.4",
     "fastify-elasticsearch": "^2.0.0",
-    "fastify-plugin": "^4.5.0",
+    "fastify-plugin": "^3.0.1",
     "flatstr": "^1.0.12",
     "global-agent": "^3.0.0",
-    "got": "11.8.6",
-    "ioredis": "^5.3.2",
+    "got": "11.8.5",
+    "ioredis": "^4.28.5",
     "lodash": "^4.17.21",
     "moment": "^2.29.4",
-    "nodemailer": "^6.9.3",
-    "pino-pretty": "^10.0.0",
+    "nodemailer": "^6.9.0",
+    "pino-pretty": "^10.2.0",
     "portfinder": "^1.0.32",
-    "socket.io": "4.6.2",
-    "socket.io-client": "4.6.2",
+    "socket.io": "4.7.2",
+    "socket.io-client": "4.7.2",
     "socket.io-redis": "^6.1.1",
     "telegraf": "^4.12.3-canary.1",
-    "typescript": "^5.1.3",
-    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.30.0",
-    "ws": "^8.12.1"
+    "typescript": "^4.5.2",
+    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.31.0",
+    "ws": "^8.14.2"
   },
   "devDependencies": {
-    "@types/amqplib": "^0.10.1",
+    "@types/amqplib": "^0.10.2",
     "@types/async": "^3.2.18",
     "@types/global-agent": "^2.1.1",
-    "@types/ioredis": "^5.0.0",
-    "@types/lodash": "^4.14.195",
-    "@types/node": "^20.2.5",
-    "@types/nodemailer": "^6.4.8",
-    "@types/ws": "^8.5.4"
+    "@types/ioredis": "^4.28.10",
+    "@types/lodash": "^4.14.197",
+    "@types/node": "^20.5.7",
+    "@types/nodemailer": "^6.4.9",
+    "@types/ws": "^8.5.5"
   },
   "optionalDependencies": {
     "bufferutil": "^4.0.7",

scripts/hyp-repair.ts

@@ -509,43 +509,73 @@ async function repairChain(chain: string, file: string, args: any) {
     await fillMissingBlocksFromFile(args.host, chain, file, args.dry);
 }
 
-async function fillMissingBlocksFromFile(host: string, chain: string, file: string, dryRun: string) {
+async function fillMissingBlocksFromFile(host, chain, file, dryRun) {
     const config = readConnectionConfig();
     const controlPort = config.chains[chain].control_port;
     let hyperionIndexer = `ws://localhost:${controlPort}`;
     if (host) {
-        hyperionIndexer = 'ws://' + host + ':' + controlPort;
+        hyperionIndexer = `ws://${host}:${controlPort}`;
     }
     const controller = new WebSocket(hyperionIndexer + '/local');
+
+    // Function to send Chunk
+    async function sendChunk(chunk): Promise<void> {
+        return new Promise((resolve, reject) => {
+            const payload = {
+                "event": "fill_missing_blocks",
+                "data": chunk
+            };
+            controller.send(JSON.stringify(payload));
+            
+            // Wait repair_completed confirmation
+            controller.once('message', (data) => {
+                const parsed = JSON.parse(data.toString());
+                if (parsed.event === 'repair_completed') {
+                    // console.log(`Hyperion repair completed for chunk!`);
+                    resolve();
+                }
+            });
+        });
+    }
+
     controller.on('open', async () => {
         console.log('Connected to Hyperion Controller');
         const parsedFile = JSON.parse(readFileSync(file).toString());
+        const chunkSize = 20; // Chunk size
+        
         if (!dryRun) {
-            const payload = {
-                "event": "fill_missing_blocks",
-                "data": parsedFile
-            };
-            controller.send(JSON.stringify(payload));
+            const totalChunks = Math.ceil(parsedFile.length / chunkSize);
+            let completedChunks = 0;
+            for (let i = 0; i < parsedFile.length; i += chunkSize) {
+                const progress = (completedChunks / totalChunks) * 100;
+                const progressBar = Array(Math.round(progress / 2)).fill('#').join('');
+                process.stdout.write(`Progress: [${progressBar}] ${progress.toFixed(2)}%     \r`);
+                //  console.log(`Progress: [${progressBar}] ${progress.toFixed(2)}%`);
+
+                const chunk = parsedFile.slice(i, i + chunkSize);
+                await sendChunk(chunk);
+
+                 // Update progess bar
+                 completedChunks++;
+            }
+            console.log();
+            controller.close();
         } else {
             console.log('Dry run, skipping repair');
             controller.close();
         }
     });
-    controller.on('message', (data) => {
-        const parsed = JSON.parse(data.toString());
-        if (parsed.event === 'repair_completed') {
-            console.log(`Hyperion repair completed!`);
-            controller.close();
-        }
-    });
+
     controller.on('close', () => {
         console.log('Disconnected from Hyperion Controller');
     });
+
     controller.on('error', (err) => {
         console.log(err);
     });
 }
 
+
 function viewFile(file: string) {
     const data = readFileSync(file, 'utf8');
     const parsed = JSON.parse(data);

workers/deserializer.ts

@@ -8,6 +8,7 @@ import {Type} from "../addons/eosjs-native/eosjs-serialize";
 import {debugLog, hLog} from "../helpers/common_functions";
 import {createHash} from "crypto";
 import flatstr from 'flatstr';
+import {Options} from "amqplib";
 
 const FJS = require('fast-json-stringify');
 
@@ -124,6 +125,9 @@ export default class MainDSWorker extends HyperionWorker {
 
     allowedDynamicContracts: Set<string> = new Set<string>();
 
+    backpressureQueue: any[] = [];
+    waitToSend = false;
+
     constructor() {
 
         super();
@@ -290,6 +294,16 @@ export default class MainDSWorker extends HyperionWorker {
             this.ch.consume(process.env['worker_queue'], (data) => {
                 this.consumerQueue.push(data).catch(console.log);
             });
+            this.ch.on('drain', args => {
+                this.waitToSend = false;
+                while (this.backpressureQueue.length > 0) {
+                    const msg = this.backpressureQueue.shift();
+                    const status = this.controlledSendToQueue(msg.queue, msg.payload, msg.options);
+                    if (!status) {
+                        break;
+                    }
+                }
+            });
         }
     }
 
@@ -478,6 +492,7 @@ export default class MainDSWorker extends HyperionWorker {
                                 hLog(`${block_num} was filtered with ${inline_count} actions!`);
                             }
                             try {
+                                trace[1].signatures = signatures;
                                 this.routeToPool(trace[1], {
                                     block_num,
                                     block_id,
@@ -485,8 +500,7 @@ export default class MainDSWorker extends HyperionWorker {
                                     ts,
                                     inline_count,
                                     filtered,
-                                    live: process.env['live_mode'],
-                                    signatures
+                                    live: process.env['live_mode']
                                 });
                             } catch (e) {
                                 hLog(e);
@@ -622,15 +636,33 @@ export default class MainDSWorker extends HyperionWorker {
         }
 
         const pool_queue = `${this.chain}:ds_pool:${selected_q}`;
-        if (this.ch_ready) {
-            // console.log('selected_q', pool_queue);
-            this.ch.sendToQueue(pool_queue, bufferFromJson(trace, true), {headers});
-            return true;
+        const payload = bufferFromJson(trace, true);
+
+        if (!this.waitToSend) {
+            if (this.ch_ready) {
+                this.controlledSendToQueue(pool_queue, payload, {headers});
+                return true;
+            } else {
+                return false;
+            }
         } else {
+            this.backpressureQueue.push({
+                queue: pool_queue,
+                payload: payload,
+                options: {headers}
+            });
             return false;
         }
     }
 
+    controlledSendToQueue(pool_queue: string, payload: Buffer, options: Options.Publish): boolean {
+        const enqueueResult = this.ch.sendToQueue(pool_queue, payload, options);
+        if (!enqueueResult) {
+            this.waitToSend = true;
+        }
+        return enqueueResult;
+    }
+
     createSerialBuffer(inputArray) {
         return new Serialize.SerialBuffer({textEncoder: this.txEnc, textDecoder: this.txDec, array: inputArray});
     }
@@ -1160,7 +1192,7 @@ export default class MainDSWorker extends HyperionWorker {
                 let jsonRow = await this.processContractRowNative(payload, block_num);
 
                 if (jsonRow?.value && !jsonRow['_blacklisted']) {
-                    console.log(jsonRow);
+                    debugLog(jsonRow);
                     debugLog('Delta DS failed ->>', jsonRow);
                     jsonRow = await this.processContractRowNative(payload, block_num - 1);
                     debugLog('Retry with previous ABI ->>', jsonRow);
@@ -1495,7 +1527,7 @@ export default class MainDSWorker extends HyperionWorker {
                             } catch (e) {
                                 hLog(`Delta struct [${key}] processing error: ${e.message}`);
                                 hLog(e);
-                                console.log(data[1]);
+                                hLog(data[1]);
                             }
                         }
                     }

workers/ds-pool.ts

@@ -454,8 +454,8 @@ export default class DSPoolWorker extends HyperionWorker {
     }
 
     async processTraces(transaction_trace, extra) {
-        const {cpu_usage_us, net_usage_words} = transaction_trace;
-        const {block_num, block_id, producer, ts, inline_count, filtered, live, signatures} = extra;
+        const {cpu_usage_us, net_usage_words, signatures} = transaction_trace;
+        const {block_num, block_id, producer, ts, inline_count, filtered, live} = extra;
 
         if (transaction_trace.status === 0) {
             let action_count = 0;

workers/state-reader.ts

@@ -375,7 +375,7 @@ export default class StateReader extends HyperionWorker {
                 this.shipInitStatus = result[1];
                 hLog(`\n| SHIP Status Report\n| Init block: ${this.shipInitStatus['chain_state_begin_block']}\n| Head block: ${this.shipInitStatus['chain_state_end_block']}`);
                 const chain_state_begin_block = this.shipInitStatus['chain_state_begin_block'];
-                if (!this.conf.indexer.disable_reading) {
+                if (!this.conf.indexer.disable_reading || process.env['worker_role'] === 'repair_reader') {
 
                     switch (process.env['worker_role']) {