Merge pull request #62 from eosrio/es-8-alpha

Es 8 alpha

Author: igorls

api/routes/v1-history/get_transaction/get_transaction.ts

@@ -2,28 +2,18 @@ import {FastifyInstance, FastifyReply, FastifyRequest} from "fastify";
 import {mergeActionMeta, timedQuery} from "../../../helpers/functions";
 import {createHash} from "crypto";
 import flatstr from 'flatstr';
+import {GetInfoResult} from "eosjs/dist/eosjs-rpc-interfaces";
 
 async function getTransaction(fastify: FastifyInstance, request: FastifyRequest) {
 	if (typeof request.body === 'string') {
 		request.body = JSON.parse(request.body)
 	}
 	const body: any = request.body;
-	const pResults = await Promise.all([fastify.eosjs.rpc.get_info(), fastify.elastic['search']({
-		"index": fastify.manager.chain + '-action-*',
-		"body": {
-			"query": {
-				"bool": {
-					must: [
-						{term: {"trx_id": body.id.toLowerCase()}}
-					]
-				}
-			},
-			"sort": {
-				"global_sequence": "asc"
-			}
-		}
-	})]);
-	const results = pResults[1];
+	const redis = fastify.redis;
+	const trxId = body.id.toLowerCase();
+	const conf = fastify.manager.config;
+	const cachedData = await redis.hgetall('trx_' + trxId);
+
 	const response: any = {
 		"id": body.id,
 		"trx": {
@@ -42,11 +32,89 @@ async function getTransaction(fastify: FastifyInstance, request: FastifyRequest)
 		},
 		"block_num": 0,
 		"block_time": "",
-		"last_irreversible_block": pResults[0].last_irreversible_block_num,
+		"last_irreversible_block": undefined,
 		"traces": []
 	};
 
-	const hits = results['body']['hits']['hits'];
+	let hits;
+
+	// build get_info request with caching
+	const $getInfo = new Promise<GetInfoResult>(resolve => {
+		const key = `${fastify.manager.chain}_get_info`;
+		fastify.redis.get(key).then(value => {
+			if (value) {
+				resolve(JSON.parse(value));
+			} else {
+				fastify.eosjs.rpc.get_info().then(value1 => {
+					fastify.redis.set(key, JSON.stringify(value1), 'EX', 6);
+					resolve(value1);
+				}).catch((reason) => {
+					console.log(reason);
+					response.error = 'failed to get last_irreversible_block_num'
+					resolve(null);
+				});
+			}
+		});
+	});
+
+	// reconstruct hits from cached data
+	if (cachedData && Object.keys(cachedData).length > 0) {
+		const gsArr = [];
+		for (let cachedDataKey in cachedData) {
+			gsArr.push(cachedData[cachedDataKey]);
+		}
+		gsArr.sort((a, b) => {
+			return a.global_sequence - b.global_sequence;
+		});
+		hits = gsArr.map(value => {
+			return {
+				_source: JSON.parse(value)
+			};
+		});
+		const promiseResults = await Promise.all([
+			redis.ttl('trx_' + trxId),
+			$getInfo
+		]);
+		response.cache_expires_in = promiseResults[0];
+		response.last_irreversible_block = promiseResults[1].last_irreversible_block_num;
+	}
+
+	// search on ES if cache is not present
+	if (!hits) {
+		const _size = conf.api.limits.get_trx_actions || 100;
+		const blockHint = parseInt(body.block_num_hint, 10);
+		let indexPattern = '';
+		if (blockHint) {
+			const idxPart = Math.ceil(blockHint / conf.settings.index_partition_size).toString().padStart(6, '0');
+			indexPattern = fastify.manager.chain + `-action-${conf.settings.index_version}-${idxPart}`;
+		} else {
+			indexPattern = fastify.manager.chain + '-action-*';
+		}
+		let pResults;
+		try {
+
+			// build search request
+			const $search = fastify.elastic.search({
+				index: indexPattern,
+				size: _size,
+				body: {
+					query: {bool: {must: [{term: {trx_id: trxId}}]}},
+					sort: {global_sequence: "asc"}
+				}
+			});
+
+			// execute in parallel
+			pResults = await Promise.all([$getInfo, $search]);
+		} catch (e) {
+			console.log(e.message);
+			if (e.meta.statusCode === 404) {
+				return response;
+			}
+		}
+		hits = pResults[1]['body']['hits']['hits'];
+		response.last_irreversible_block = pResults[0].last_irreversible_block_num;
+	}
+
 
 	if (hits.length > 0) {
 		const actions = hits;

api/routes/v1-history/get_transaction/index.ts

@@ -9,7 +9,10 @@ export default function (fastify: FastifyInstance, opts: any, next) {
         tags: ['history'],
         body: {
             type: ['object', 'string'],
-            properties: {id: {description: 'transaction id', type: 'string'}},
+            properties: {
+                id: {description: 'transaction id', type: 'string'},
+                block_num_hint: {description: 'block number hint', type: 'integer'},
+            },
             required: ["id"]
         }
     });

helpers/elastic-routes.ts

@@ -370,9 +370,11 @@ export class ElasticRoutes {
             });
         }
         this.ingestNodeCounters[minIdx].docs += bulkData.body.length;
-        if (this.ingestNodeCounters[minIdx].docs > 1000) {
+        if (this.ingestNodeCounters[minIdx].docs > 10000) {
             this.resetCounters();
         }
+        // print full request
+        // console.log(JSON.stringify(bulkData));
         return this.cm.ingestClients[minIdx]['bulk'](bulkData);
     }
 

package-lock.json

@@ -26,7 +26,7 @@
   },
   "license": "MIT",
   "dependencies": {
-    "@elastic/elasticsearch": "^7.13.0",
+    "@elastic/elasticsearch": "^7.14.0",
     "@eosrio/node-abieos": "^2.1.1",
     "@pm2/io": "^5.0.0",
     "amqplib": "^0.8.0",
@@ -35,7 +35,7 @@
     "cross-fetch": "^3.1.4",
     "eosjs": "^22.1.0",
     "fast-json-stringify": "^2.7.8",
-    "fastify": "^3.20.1",
+    "fastify": "^3.20.2",
     "fastify-autoload": "^3.8.0",
     "fastify-compress": "^3.6.0",
     "fastify-cors": "^6.0.2",
@@ -58,15 +58,15 @@
     "socket.io-redis": "^6.1.1",
     "telegraf": "^4.4.1",
     "typescript": "^4.3.5",
-    "ws": "^8.0.0",
-    "yargs": "^17.1.0"
+    "ws": "^8.1.0",
+    "yargs": "^17.1.1"
   },
   "devDependencies": {
     "@types/amqplib": "^0.8.1",
     "@types/async": "^3.2.7",
-    "@types/ioredis": "^4.26.6",
+    "@types/ioredis": "^4.26.7",
     "@types/lodash": "^4.14.172",
-    "@types/node": "^16.4.12",
+    "@types/node": "^16.6.1",
     "@types/nodemailer": "^6.4.4",
     "@types/ws": "^7.4.7"
   },