Merge pull request #142 from eosrio/large-tx-fix

Large tx-sig fix
eosrio · Sep 25, 2023 · b94f99d · b94f99d
2 parents c59005b + 76807c6
commit b94f99d
Show file tree

Hide file tree

Showing 7 changed files with 77 additions and 48 deletions.
diff --git a/connections/amqp.ts b/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();

diff --git a/ecosystem.config.js b/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};
diff --git a/modules/parsers/base-parser.ts b/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;
             }
 

diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "hyperion-history",
-  "version": "3.3.9-7",
+  "version": "3.3.9-8",
   "description": "Scalable Full History API Solution for EOSIO based blockchains",
   "main": "launcher.js",
   "scripts": {
@@ -60,11 +60,11 @@
     "socket.io-redis": "^6.1.1",
     "telegraf": "^4.12.3-canary.1",
     "typescript": "^4.5.2",
-    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.24.0",
-    "ws": "^8.12.1"
+    "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": "^4.28.10",

diff --git a/workers/deserializer.ts b/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,15 +492,15 @@ 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,
                                     producer,
                                     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]);
                             }
                         }
                     }

diff --git a/workers/ds-pool.ts b/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;