v3.3.9

- fix readers stopping before having finished

api/routes/v2-history/get_actions/definitions.ts

@@ -1,5 +1,6 @@
 export const terms = [
     "notified",
+    "receipts.receiver",
     "act.authorization.actor"
 ];
 
@@ -15,8 +16,8 @@ export const extendedActions = new Set([
 ]);
 
 export const primaryTerms = [
-    "notified",
     "block_num",
+    "block_id",
     "global_sequence",
     "producer",
     "@timestamp",
@@ -24,5 +25,10 @@ export const primaryTerms = [
     "action_ordinal",
     "cpu_usage_us",
     "net_usage_words",
-    "trx_id"
+    "trx_id",
+    "receipts.receiver",
+    "receipts.global_sequence",
+    "receipts.recv_sequence",
+    "receipts.auth_sequence.account",
+    "receipts.auth_sequence.sequence"
 ];

api/routes/v2-history/get_actions/functions.ts

@@ -1,236 +1,257 @@
 import {primaryTerms, terms} from "./definitions";
 
 export function addSortedBy(query, queryBody, sort_direction) {
-	if (query['sortedBy']) {
-		const opts = query['sortedBy'].split(":");
-		const sortedByObj = {};
-		sortedByObj[opts[0]] = opts[1];
-		queryBody['sort'] = sortedByObj;
-	} else {
-		queryBody['sort'] = {
-			"global_sequence": sort_direction
-		};
-	}
+    if (query['sortedBy']) {
+        const opts = query['sortedBy'].split(":");
+        const sortedByObj = {};
+        sortedByObj[opts[0]] = opts[1];
+        queryBody['sort'] = sortedByObj;
+    } else {
+        queryBody['sort'] = {
+            "global_sequence": sort_direction
+        };
+    }
 }
 
 export function processMultiVars(queryStruct, parts, field) {
-	const must = [];
-	const mustNot = [];
+    const must = [];
+    const mustNot = [];
 
-	parts.forEach(part => {
-		if (part.startsWith("!")) {
-			mustNot.push(part.replace("!", ""));
-		} else {
-			must.push(part);
-		}
-	});
+    parts.forEach(part => {
+        if (part.startsWith("!")) {
+            mustNot.push(part.replace("!", ""));
+        } else {
+            must.push(part);
+        }
+    });
 
-	if (must.length > 1) {
-		queryStruct.bool.must.push({
-			bool: {
-				should: must.map(elem => {
-					const _q = {};
-					_q[field] = elem;
-					return {term: _q}
-				})
-			}
-		});
-	} else if (must.length === 1) {
-		const mustQuery = {};
-		mustQuery[field] = must[0];
-		queryStruct.bool.must.push({term: mustQuery});
-	}
+    if (must.length > 1) {
+        queryStruct.bool.must.push({
+            bool: {
+                should: must.map(elem => {
+                    const _q = {};
+                    _q[field] = elem;
+                    return {term: _q}
+                })
+            }
+        });
+    } else if (must.length === 1) {
+        const mustQuery = {};
+        mustQuery[field] = must[0];
+        queryStruct.bool.must.push({term: mustQuery});
+    }
 
-	if (mustNot.length > 1) {
-		queryStruct.bool.must_not.push({
-			bool: {
-				should: mustNot.map(elem => {
-					const _q = {};
-					_q[field] = elem;
-					return {term: _q}
-				})
-			}
-		});
-	} else if (mustNot.length === 1) {
-		const mustNotQuery = {};
-		mustNotQuery[field] = mustNot[0].replace("!", "");
-		queryStruct.bool.must_not.push({term: mustNotQuery});
-	}
+    if (mustNot.length > 1) {
+        queryStruct.bool.must_not.push({
+            bool: {
+                should: mustNot.map(elem => {
+                    const _q = {};
+                    _q[field] = elem;
+                    return {term: _q}
+                })
+            }
+        });
+    } else if (mustNot.length === 1) {
+        const mustNotQuery = {};
+        mustNotQuery[field] = mustNot[0].replace("!", "");
+        queryStruct.bool.must_not.push({term: mustNotQuery});
+    }
 }
 
 function addRangeQuery(queryStruct, prop, pkey, query) {
-	const _termQuery = {};
-	const parts = query[prop].split("-");
-	_termQuery[pkey] = {
-		"gte": parts[0],
-		"lte": parts[1]
-	};
-	queryStruct.bool.must.push({range: _termQuery});
+    const _termQuery = {};
+    const parts = query[prop].split("-");
+    _termQuery[pkey] = {
+        "gte": parts[0],
+        "lte": parts[1]
+    };
+    queryStruct.bool.must.push({range: _termQuery});
 }
 
 export function applyTimeFilter(query, queryStruct) {
-	if (query['after'] || query['before']) {
-		let _lte = "now";
-		let _gte = "0";
-		if (query['before']) {
-			try {
-				_lte = new Date(query['before']).toISOString();
-			} catch (e) {
-				throw new Error(e.message + ' [before]');
-			}
-		}
-		if (query['after']) {
-			try {
-				_gte = new Date(query['after']).toISOString();
-			} catch (e) {
-				throw new Error(e.message + ' [after]');
-			}
-		}
-		if (!queryStruct.bool['filter']) {
-			queryStruct.bool['filter'] = [];
-		}
-		queryStruct.bool['filter'].push({
-			range: {
-				"@timestamp": {
-					"gte": _gte,
-					"lte": _lte
-				}
-			}
-		});
-	}
+    if (query['after'] || query['before']) {
+        if (query['after']?.includes('T') || query['before']?.includes('T')) {
+            let _lte = "now";
+            let _gte = "0";
+            if (query['before']) {
+                try {
+                    _lte = new Date(query['before']).toISOString();
+                } catch (e) {
+                    throw new Error(e.message + ' [before]');
+                }
+            }
+            if (query['after']) {
+                try {
+                    _gte = new Date(query['after']).toISOString();
+                } catch (e) {
+                    throw new Error(e.message + ' [after]');
+                }
+            }
+            if (!queryStruct.bool['filter']) {
+                queryStruct.bool['filter'] = [];
+            }
+            queryStruct.bool['filter'].push({
+                range: {
+                    "@timestamp": {
+                        "gte": _gte,
+                        "lte": _lte
+                    }
+                }
+            });
+        } else {
+            // search by block number
+            const rangeObj = {
+                range: {
+                    block_num: {}
+                }
+            };
+            if (parseInt(query['after']) > 0) {
+                rangeObj.range.block_num['gte'] = query['after'];
+            }
+            if (parseInt(query['before']) > 0) {
+                rangeObj.range.block_num['lte'] = query['before'];
+            }
+            if (Object.keys(rangeObj.range.block_num).length > 0) {
+                if (!queryStruct.bool['filter']) {
+                    queryStruct.bool['filter'] = [];
+                }
+                queryStruct.bool['filter'].push(rangeObj);
+            }
+        }
+    }
 }
 
 export function applyGenericFilters(query, queryStruct, allowedExtraParams: Set<string>) {
-	for (const prop in query) {
-		if (Object.prototype.hasOwnProperty.call(query, prop)) {
-			const pair = prop.split(".");
-			if (pair.length > 1 || primaryTerms.includes(pair[0])) {
-				let pkey;
-				if (pair.length > 1 && allowedExtraParams) {
-					pkey = allowedExtraParams.has(pair[0]) ? "@" + prop : prop;
-				} else {
-					pkey = prop;
-				}
-				if (query[prop].indexOf("-") !== -1) {
-					addRangeQuery(queryStruct, prop, pkey, query);
-				} else {
-					const _qObj = {};
-					const parts = query[prop].split(",");
-					if (parts.length > 1) {
-						processMultiVars(queryStruct, parts, prop);
-					} else if (parts.length === 1) {
+    for (const prop in query) {
+        if (Object.prototype.hasOwnProperty.call(query, prop)) {
+            const pair = prop.split(".");
+            if (pair.length > 1 || primaryTerms.includes(pair[0])) {
+                let pkey;
+                if (pair.length > 1 && allowedExtraParams) {
+                    pkey = allowedExtraParams.has(pair[0]) ? "@" + prop : prop;
+                } else {
+                    pkey = prop;
+                }
+                if (query[prop].indexOf("-") !== -1) {
+                    addRangeQuery(queryStruct, prop, pkey, query);
+                } else {
+                    const _qObj = {};
+                    const parts = query[prop].split(",");
+                    if (parts.length > 1) {
+                        processMultiVars(queryStruct, parts, prop);
+                    } else if (parts.length === 1) {
 
-						// @transfer.memo special case
-						if (pkey === '@transfer.memo') {
-							_qObj[pkey] = {
-								query: parts[0]
-							};
+                        // @transfer.memo special case
+                        if (pkey === '@transfer.memo') {
+                            _qObj[pkey] = {
+                                query: parts[0]
+                            };
 
-							if (query.match_fuzziness) {
-								_qObj[pkey].fuzziness = query.match_fuzziness;
-							}
+                            if (query.match_fuzziness) {
+                                _qObj[pkey].fuzziness = query.match_fuzziness;
+                            }
 
-							if (query.match_operator) {
-								_qObj[pkey].operator = query.match_operator;
-							}
+                            if (query.match_operator) {
+                                _qObj[pkey].operator = query.match_operator;
+                            }
 
-							queryStruct.bool.must.push({
-								match: _qObj
-							});
-						} else {
-							const andParts = parts[0].split(" ");
-							if (andParts.length > 1) {
-								andParts.forEach(value => {
-									const _q = {};
-									_q[pkey] = value;
-									queryStruct.bool.must.push({term: _q});
-								});
-							} else {
-								if (parts[0].startsWith("!")) {
-									_qObj[pkey] = parts[0].replace("!", "");
-									queryStruct.bool.must_not.push({term: _qObj});
-								} else {
-									_qObj[pkey] = parts[0];
-									queryStruct.bool.must.push({term: _qObj});
-								}
-							}
-						}
-					}
-				}
-			}
-		}
-	}
+                            queryStruct.bool.must.push({
+                                match: _qObj
+                            });
+                        } else {
+                            const andParts = parts[0].split(" ");
+                            if (andParts.length > 1) {
+                                andParts.forEach(value => {
+                                    const _q = {};
+                                    _q[pkey] = value;
+                                    queryStruct.bool.must.push({term: _q});
+                                });
+                            } else {
+                                if (parts[0].startsWith("!")) {
+                                    _qObj[pkey] = parts[0].replace("!", "");
+                                    queryStruct.bool.must_not.push({term: _qObj});
+                                } else {
+                                    _qObj[pkey] = parts[0];
+                                    queryStruct.bool.must.push({term: _qObj});
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
 }
 
 export function makeShouldArray(query) {
-	const should_array = [];
-	for (const entry of terms) {
-		const tObj = {term: {}};
-		tObj.term[entry] = query.account;
-		should_array.push(tObj);
-	}
-	return should_array;
+    const should_array = [];
+    for (const entry of terms) {
+        const tObj = {term: {}};
+        tObj.term[entry] = query.account;
+        should_array.push(tObj);
+    }
+    return should_array;
 }
 
 export function applyCodeActionFilters(query, queryStruct) {
-	let filterObj = [];
-	if (query.filter) {
-		for (const filter of query.filter.split(',')) {
-			if (filter !== '*:*') {
-				const _arr = [];
-				const parts = filter.split(':');
-				if (parts.length === 2) {
-					const [code, method] = parts;
-					if (code && code !== "*") {
-						_arr.push({'term': {'act.account': code}});
-					}
-					if (method && method !== "*") {
-						_arr.push({'term': {'act.name': method}});
-					}
-				}
-				if (_arr.length > 0) {
-					filterObj.push({bool: {must: _arr}});
-				}
-			}
-		}
-		if (filterObj.length > 0) {
-			queryStruct.bool['should'] = filterObj;
-			queryStruct.bool['minimum_should_match'] = 1;
-		}
-	}
+    let filterObj = [];
+    if (query.filter) {
+        for (const filter of query.filter.split(',')) {
+            if (filter !== '*:*') {
+                const _arr = [];
+                const parts = filter.split(':');
+                if (parts.length === 2) {
+                    const [code, method] = parts;
+                    if (code && code !== "*") {
+                        _arr.push({'term': {'act.account': code}});
+                    }
+                    if (method && method !== "*") {
+                        _arr.push({'term': {'act.name': method}});
+                    }
+                }
+                if (_arr.length > 0) {
+                    filterObj.push({bool: {must: _arr}});
+                }
+            }
+        }
+        if (filterObj.length > 0) {
+            queryStruct.bool['should'] = filterObj;
+            queryStruct.bool['minimum_should_match'] = 1;
+        }
+    }
 }
 
 export function getSkipLimit(query, max?: number) {
-	let skip, limit;
-	skip = parseInt(query.skip, 10);
-	if (skip < 0) {
-		throw new Error('invalid skip parameter');
-	}
-	limit = parseInt(query.limit, 10);
-	if (limit < 1) {
-		throw new Error('invalid limit parameter');
-	} else if (limit > max) {
-		throw new Error(`limit too big, maximum: ${max}`);
-	}
-	return {skip, limit};
+    let skip, limit;
+    skip = parseInt(query.skip, 10);
+    if (skip < 0) {
+        throw new Error('invalid skip parameter');
+    }
+    limit = parseInt(query.limit, 10);
+    if (limit < 1) {
+        throw new Error('invalid limit parameter');
+    } else if (limit > max) {
+        throw new Error(`limit too big, maximum: ${max}`);
+    }
+    return {skip, limit};
 }
 
 export function getSortDir(query) {
-	let sort_direction = 'desc';
-	if (query.sort) {
-		if (query.sort === 'asc' || query.sort === '1') {
-			sort_direction = 'asc';
-		} else if (query.sort === 'desc' || query.sort === '-1') {
-			sort_direction = 'desc'
-		} else {
-			throw new Error('invalid sort direction');
-		}
-	}
-	return sort_direction;
+    let sort_direction = 'desc';
+    if (query.sort) {
+        if (query.sort === 'asc' || query.sort === '1') {
+            sort_direction = 'asc';
+        } else if (query.sort === 'desc' || query.sort === '-1') {
+            sort_direction = 'desc'
+        } else {
+            throw new Error('invalid sort direction');
+        }
+    }
+    return sort_direction;
 }
 
 export function applyAccountFilters(query, queryStruct) {
-	if (query.account) {
-		queryStruct.bool.must.push({"bool": {should: makeShouldArray(query)}});
-	}
+    if (query.account) {
+        queryStruct.bool.must.push({"bool": {should: makeShouldArray(query)}});
+    }
 }

api/routes/v2-history/get_actions/get_actions.ts

@@ -22,11 +22,17 @@ async function getActions(fastify: FastifyInstance, request: FastifyRequest) {
     };
 
     const {skip, limit} = getSkipLimit(query, maxActions);
+
     const sort_direction = getSortDir(query);
+
     applyAccountFilters(query, queryStruct);
+
     applyGenericFilters(query, queryStruct, fastify.allowedActionQueryParamSet);
+
     applyTimeFilter(query, queryStruct);
+
     applyCodeActionFilters(query, queryStruct);
+
     // allow precise counting of total hits
     const trackTotalHits = getTrackTotalHits(query);
 

api/routes/v2-history/get_actions/index.ts

@@ -3,130 +3,157 @@ import {getActionsHandler} from "./get_actions";
 import {addApiRoute, extendQueryStringSchema, extendResponseSchema, getRouteName} from "../../../helpers/functions";
 
 export const getActionResponseSchema = {
-	"@timestamp": {type: "string"},
-	"timestamp": {type: "string"},
-	"block_num": {type: "number"},
-	"trx_id": {type: "string"},
-	"act": {
-		type: 'object',
-		properties: {
-			"account": {type: "string"},
-			"name": {type: "string"}
-		},
-		additionalProperties: true
-	},
-	"notified": {
-		type: "array", items: {type: "string"}
-	},
-	"cpu_usage_us": {type: "number"},
-	"net_usage_words": {type: "number"},
-	"account_ram_deltas": {
-		type: "array",
-		items: {
-			type: "object",
-			properties: {
-				"account": {type: "string"},
-				"delta": {type: "number"}
-			},
-			additionalProperties: true
-		}
-	},
-	"global_sequence": {type: "number"},
-	"receiver": {type: 'string'},
-	"producer": {type: "string"},
-	"parent": {type: "number"},
-	"action_ordinal": {type: 'number'},
-	"creator_action_ordinal": {type: 'number'},
-	"signatures": {type: "array", items: {type: 'string'}}
+    "timestamp": {type: "string"},
+    "block_num": {type: "number"},
+    "block_id": {type: "string"},
+    "trx_id": {type: "string"},
+    "act": {
+        type: 'object',
+        properties: {
+            "account": {type: "string"},
+            "name": {type: "string"},
+            "authorization": {
+                type: "array",
+                items: {
+                    type: "object",
+                    properties: {
+                        "actor": {type: "string"},
+                        "permission": {type: "string"},
+                    }
+                }
+            }
+        },
+        additionalProperties: true
+    },
+    "receipts": {
+        type: "array",
+        items: {
+            type: "object",
+            properties: {
+                "receiver": {type: "string"},
+                "global_sequence": {type: "number"},
+                "recv_sequence": {type: "number"},
+                "auth_sequence": {
+                    type: "array",
+                    items: {
+                        type: "object",
+                        properties: {
+                            "account": {type: "string"},
+                            "sequence": {type: "number"},
+                        }
+                    }
+                }
+            }
+        }
+    },
+    "cpu_usage_us": {type: "number"},
+    "net_usage_words": {type: "number"},
+    "account_ram_deltas": {
+        type: "array",
+        items: {
+            type: "object",
+            properties: {
+                "account": {type: "string"},
+                "delta": {type: "number"}
+            },
+            additionalProperties: true
+        }
+    },
+    "global_sequence": {type: "number"},
+    "producer": {type: "string"},
+    "parent": {type: "number"},
+    "action_ordinal": {type: 'number'},
+    "creator_action_ordinal": {type: 'number'},
+    "signatures": {type: "array", items: {type: 'string'}}
 };
 
 export default function (fastify: FastifyInstance, opts: any, next) {
-	const schema: FastifySchema = {
-		description: 'get actions based on notified account. this endpoint also accepts generic filters based on indexed fields' +
-			' (e.g. act.authorization.actor=eosio or act.name=delegatebw), if included they will be combined with a AND operator',
-		summary: 'get root actions',
-		tags: ['history'],
-		querystring: extendQueryStringSchema({
-			"account": {
-				description: 'notified account',
-				type: 'string',
-				minLength: 1,
-				maxLength: 12
-			},
-			"track": {
-				description: 'total results to track (count) [number or true]',
-				type: 'string'
-			},
-			"filter": {
-				description: 'code:name filter',
-				type: 'string',
-				minLength: 3
-			},
-			"sort": {
-				description: 'sort direction',
-				enum: ['desc', 'asc', '1', '-1'],
-				type: 'string'
-			},
-			"after": {
-				description: 'filter after specified date (ISO8601)',
-				type: 'string'
-			},
-			"before": {
-				description: 'filter before specified date (ISO8601)',
-				type: 'string'
-			},
-			"simple": {
-				description: 'simplified output mode',
-				type: 'boolean'
-			},
-			"hot_only": {
-				description: 'search only the latest hot index',
-				type: 'boolean'
-			},
-			"noBinary": {
-				description: "exclude large binary data",
-				type: 'boolean'
-			},
-			"checkLib": {
-				description: "perform reversibility check",
-				type: 'boolean'
-			},
-		}),
-		response: extendResponseSchema({
-			"simple_actions": {
-				type: "array",
-				items: {
-					type: "object",
-					properties: {
-						"block": {type: "number"},
-						"timestamp": {type: "string"},
-						"irreversible": {type: "boolean"},
-						"contract": {type: "string"},
-						"action": {type: "string"},
-						"actors": {type: "string"},
-						"notified": {type: "string"},
-						"transaction_id": {type: "string"},
-						"data": {
-							additionalProperties: true
-						}
-					}
-				}
-			},
-			"actions": {
-				type: "array",
-				items: {
-					type: 'object',
-					properties: getActionResponseSchema
-				}
-			}
-		})
-	};
-	addApiRoute(
-		fastify,
-		'GET',
-		getRouteName(__filename),
-		getActionsHandler,
-		schema
-	);
-	next();
+    const schema: FastifySchema = {
+        description: 'get actions based on notified account. this endpoint also accepts generic filters based on indexed fields' +
+            ' (e.g. act.authorization.actor=eosio or act.name=delegatebw), if included they will be combined with a AND operator',
+        summary: 'get root actions',
+        tags: ['history'],
+        querystring: extendQueryStringSchema({
+            "account": {
+                description: 'notified account',
+                type: 'string',
+                minLength: 1,
+                maxLength: 12
+            },
+            "track": {
+                description: 'total results to track (count) [number or true]',
+                type: 'string'
+            },
+            "filter": {
+                description: 'code:name filter',
+                type: 'string',
+                minLength: 3
+            },
+            "sort": {
+                description: 'sort direction',
+                enum: ['desc', 'asc', '1', '-1'],
+                type: 'string'
+            },
+            "after": {
+                description: 'filter after specified date (ISO8601)',
+                type: 'string'
+            },
+            "before": {
+                description: 'filter before specified date (ISO8601)',
+                type: 'string'
+            },
+            "simple": {
+                description: 'simplified output mode',
+                type: 'boolean'
+            },
+            "hot_only": {
+                description: 'search only the latest hot index',
+                type: 'boolean'
+            },
+            "noBinary": {
+                description: "exclude large binary data",
+                type: 'boolean'
+            },
+            "checkLib": {
+                description: "perform reversibility check",
+                type: 'boolean'
+            },
+        }),
+        response: extendResponseSchema({
+            "simple_actions": {
+                type: "array",
+                items: {
+                    type: "object",
+                    properties: {
+                        "block": {type: "number"},
+                        "timestamp": {type: "string"},
+                        "irreversible": {type: "boolean"},
+                        "contract": {type: "string"},
+                        "action": {type: "string"},
+                        "actors": {type: "string"},
+                        "notified": {type: "string"},
+                        "transaction_id": {type: "string"},
+                        "data": {
+                            additionalProperties: true
+                        }
+                    }
+                }
+            },
+            "actions": {
+                type: "array",
+                items: {
+                    type: 'object',
+                    properties: getActionResponseSchema
+                }
+            }
+        })
+    };
+    addApiRoute(
+        fastify,
+        'GET',
+        getRouteName(__filename),
+        getActionsHandler,
+        schema
+    );
+    next();
 }

api/socketManager.ts

@@ -118,8 +118,9 @@ export class SocketManager {
         });
 
         try {
-            this.uwsApp.listen(1234, () => {
-                hLog('Socket.IO via uWS started!');
+            const port = this.server.manager.config.api.stream_port || 1234;
+            this.uwsApp.listen(port, () => {
+                hLog(`Socket.IO via uWS started on port ${port}`);
             });
         } catch (e) {
             hLog(e.message);

chains/example.config.json

@@ -6,6 +6,7 @@
     "chain_name": "EXAMPLE Chain",
     "server_addr": "127.0.0.1",
     "server_port": 7000,
+    "stream_port": 1234,
     "server_name": "127.0.0.1:7000",
     "provider_name": "Example Provider",
     "provider_url": "https://example.com",
@@ -62,7 +63,7 @@
     "preview": false,
     "chain": "eos",
     "eosio_alias": "eosio",
-    "parser": "2.1",
+    "parser": "3.2",
     "auto_stop": 0,
     "index_version": "v1",
     "debug": false,
@@ -119,7 +120,7 @@
       "voters": true
     },
     "index_deltas": true,
-    "index_transfer_memo": false,
+    "index_transfer_memo": true,
     "index_all_deltas": true,
     "deferred_trx": false,
     "failed_trx": false,

connections/manager.class.ts

@@ -9,170 +9,167 @@ 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 {
 
-	config: HyperionConfig;
-	conn: HyperionConnections;
+    config: HyperionConfig;
+    conn: HyperionConnections;
 
-	chain: string;
-	last_commit_hash: string;
-	current_version: string;
+    chain: string;
+    last_commit_hash: string;
+    current_version: string;
 
-	private readonly esIngestClients: Client[];
-	private esIngestClient: Client;
+    private readonly esIngestClients: Client[];
+    private esIngestClient: Client;
 
-	constructor(private cm: ConfigurationModule) {
-		this.config = cm.config;
-		this.conn = cm.connections;
-		if (!this.conn.amqp.protocol) {
-			this.conn.amqp.protocol = 'http';
-		}
-		this.chain = this.config.settings.chain;
-		this.esIngestClients = [];
-		this.prepareESClient();
-		this.prepareIngestClients();
-	}
+    constructor(private cm: ConfigurationModule) {
+        this.config = cm.config;
+        this.conn = cm.connections;
+        if (!this.conn.amqp.protocol) {
+            this.conn.amqp.protocol = 'http';
+        }
+        this.chain = this.config.settings.chain;
+        this.esIngestClients = [];
+        this.prepareESClient();
+        this.prepareIngestClients();
+    }
 
-	get nodeosJsonRPC() {
-		// @ts-ignore
-		return new JsonRpc(this.conn.chains[this.chain].http, {fetch});
-	}
+    get nodeosJsonRPC() {
+        // @ts-ignore
+        return new JsonRpc(this.conn.chains[this.chain].http, {fetch});
+    }
 
-	async purgeQueues() {
-		hLog(`Purging all ${this.chain} queues!`);
-		const apiUrl = `http://${this.conn.amqp.api}`;
-		const vHost = encodeURIComponent(this.conn.amqp.vhost);
-		const getAllQueuesFromVHost = apiUrl + `/api/queues/${vHost}`;
-		const opts = {
-			username: this.conn.amqp.user,
-			password: this.conn.amqp.pass
-		};
-		let result;
-		try {
-			const data = await got(getAllQueuesFromVHost, opts);
-			if (data) {
-				result = JSON.parse(data.body);
-			}
-		} catch (e) {
-			console.log(e.message);
-			console.error('failed to connect to rabbitmq http api');
-			process.exit(1);
-		}
-		if (result) {
-			for (const queue of result) {
-				if (queue.name.startsWith(this.chain + ":")) {
-					const msg_count = parseInt(queue.messages);
-					if (msg_count > 0) {
-						try {
-							await got.delete(apiUrl + `/api/queues/${vHost}/${queue.name}/contents`, opts);
-							hLog(`${queue.messages} messages deleted on queue ${queue.name}`);
-						} catch (e) {
-							console.log(e.message);
-							console.error('failed to connect to rabbitmq http api');
-							process.exit(1);
-						}
-					}
-				}
-			}
-		}
-	}
+    async purgeQueues() {
+        hLog(`Purging all ${this.chain} queues!`);
+        const apiUrl = `http://${this.conn.amqp.api}`;
+        const vHost = encodeURIComponent(this.conn.amqp.vhost);
+        const getAllQueuesFromVHost = apiUrl + `/api/queues/${vHost}`;
+        const opts = {
+            username: this.conn.amqp.user,
+            password: this.conn.amqp.pass
+        };
+        let result;
+        try {
+            const data = await got(getAllQueuesFromVHost, opts);
+            if (data) {
+                result = JSON.parse(data.body);
+            }
+        } catch (e) {
+            console.log(e.message);
+            console.error('failed to connect to rabbitmq http api');
+            process.exit(1);
+        }
+        if (result) {
+            for (const queue of result) {
+                if (queue.name.startsWith(this.chain + ":")) {
+                    const msg_count = parseInt(queue.messages);
+                    if (msg_count > 0) {
+                        try {
+                            await got.delete(apiUrl + `/api/queues/${vHost}/${queue.name}/contents`, opts);
+                            hLog(`${queue.messages} messages deleted on queue ${queue.name}`);
+                        } catch (e) {
+                            console.log(e.message);
+                            console.error('failed to connect to rabbitmq http api');
+                            process.exit(1);
+                        }
+                    }
+                }
+            }
+        }
+    }
 
-	prepareESClient() {
-		let es_url;
-		const _es = this.conn.elasticsearch;
-		if (!_es.protocol) {
-			_es.protocol = 'http';
-		}
-		if (_es.user !== '') {
-			es_url = `${_es.protocol}://${_es.user}:${_es.pass}@${_es.host}`;
-		} else {
-			es_url = `${_es.protocol}://${_es.host}`
-		}
-		this.esIngestClient = new Client({
-			node: es_url,
-			ssl: {
-				rejectUnauthorized: false
-			}
-		});
-	}
+    prepareESClient() {
+        const _es = this.conn.elasticsearch;
+        if (!_es.protocol) {
+            _es.protocol = 'http';
+        }
+        this.esIngestClient = new Client({
+            node: `${_es.protocol}://${_es.host}`,
+            auth: {
+                username: _es.user,
+                password: _es.pass
+            },
+            ssl: _es.protocol === 'https' ? {
+                rejectUnauthorized: false
+            } : undefined
+        });
+    }
 
-	get elasticsearchClient() {
-		return this.esIngestClient;
-	}
+    get elasticsearchClient() {
+        return this.esIngestClient;
+    }
 
-	prepareIngestClients() {
-		const _es = this.conn.elasticsearch;
-		if (!_es.protocol) {
-			_es.protocol = 'http';
-		}
-		if (_es.ingest_nodes) {
-			if (_es.ingest_nodes.length > 0) {
-				for (const node of _es.ingest_nodes) {
-					let es_url;
-					if (_es.user !== '') {
-						es_url = `${_es.protocol}://${_es.user}:${_es.pass}@${node}`;
-					} else {
-						es_url = `${_es.protocol}://${node}`
-					}
-					this.esIngestClients.push(new Client({
-						node: es_url,
-						pingTimeout: 100,
-						ssl: {
-							rejectUnauthorized: false
-						}
-					}));
-				}
-			}
-		}
-	}
+    prepareIngestClients() {
+        const _es = this.conn.elasticsearch;
+        if (!_es.protocol) {
+            _es.protocol = 'http';
+        }
+        if (_es.ingest_nodes) {
+            if (_es.ingest_nodes.length > 0) {
+                for (const node of _es.ingest_nodes) {
+                    this.esIngestClient = new Client({
+                        node: `${_es.protocol}://${_es.host}`,
+                        auth: {
+                            username: _es.user,
+                            password: _es.pass
+                        },
+                        pingTimeout: 100,
+                        ssl: _es.protocol === 'https' ? {
+                            rejectUnauthorized: false
+                        } : undefined
+                    });
+                }
+            }
+        }
+    }
 
-	get ingestClients() {
-		if (this.esIngestClients.length > 0) {
-			return this.esIngestClients;
-		} else {
-			return [this.esIngestClient];
-		}
-	}
+    get ingestClients() {
+        if (this.esIngestClients.length > 0) {
+            return this.esIngestClients;
+        } else {
+            return [this.esIngestClient];
+        }
+    }
 
-	async createAMQPChannels(onReconnect, onClose) {
-		return await amqpConnect(onReconnect, this.conn.amqp, onClose);
-	}
+    async createAMQPChannels(onReconnect, onClose) {
+        return await amqpConnect(onReconnect, this.conn.amqp, onClose);
+    }
 
-	async checkQueueSize(queue) {
-		return await checkQueueSize(queue, this.conn.amqp);
-	}
+    async checkQueueSize(queue) {
+        return await checkQueueSize(queue, this.conn.amqp);
+    }
 
-	get shipClient(): StateHistorySocket {
-		return new StateHistorySocket(this.conn.chains[this.config.settings.chain]['ship'], this.config.settings.max_ws_payload_mb);
-	}
+    get shipClient(): StateHistorySocket {
+        return new StateHistorySocket(this.conn.chains[this.config.settings.chain]['ship'], this.config.settings.max_ws_payload_mb);
+    }
 
-	get ampqUrl() {
-		return getAmpqUrl(this.conn.amqp);
-	}
+    get ampqUrl() {
+        return getAmpqUrl(this.conn.amqp);
+    }
 
-	calculateServerHash() {
-		exec('git rev-parse HEAD', (err, stdout) => {
-			if (err) {
-				// hLog(`\n ${err.message}\n >>> Failed to check last commit hash. Version hash will be "custom"`);
-				hLog(`Failed to check last commit hash. Version hash will be "custom"`);
-				this.last_commit_hash = 'custom';
-			} else {
-				hLog('Last commit hash on this branch is:', stdout);
-				this.last_commit_hash = stdout.trim();
-			}
-			this.getHyperionVersion();
-		});
-	}
+    calculateServerHash() {
+        exec('git rev-parse HEAD', (err, stdout) => {
+            if (err) {
+                // hLog(`\n ${err.message}\n >>> Failed to check last commit hash. Version hash will be "custom"`);
+                hLog(`Failed to check last commit hash. Version hash will be "custom"`);
+                this.last_commit_hash = 'custom';
+            } else {
+                hLog('Last commit hash on this branch is:', stdout);
+                this.last_commit_hash = stdout.trim();
+            }
+            this.getHyperionVersion();
+        });
+    }
 
-	getServerHash() {
-		return this.last_commit_hash;
-	}
+    getServerHash() {
+        return this.last_commit_hash;
+    }
 
-	getHyperionVersion() {
-		this.current_version = require('../package.json').version;
-		if (this.last_commit_hash === 'custom') {
-			this.current_version = this.current_version + '-dirty';
-		}
-	}
+    getHyperionVersion() {
+        this.current_version = require('../package.json').version;
+        if (this.last_commit_hash === 'custom') {
+            this.current_version = this.current_version + '-dirty';
+        }
+    }
 }

connections/state-history.ts

@@ -6,6 +6,8 @@ export class StateHistorySocket {
 	private ws;
 	private readonly shipUrl;
 	private readonly max_payload_mb;
+	retryOnDisconnect = true;
+	connected = false;
 
 	constructor(ship_url, max_payload_mb) {
 		this.shipUrl = ship_url;
@@ -17,13 +19,13 @@ export class StateHistorySocket {
 	}
 
 	connect(onMessage, onDisconnect, onError, onConnected) {
-
 		debugLog(`Connecting to ${this.shipUrl}...`);
 		this.ws = new WebSocket(this.shipUrl, {
 			perMessageDeflate: false,
 			maxPayload: this.max_payload_mb * 1024 * 1024,
 		});
 		this.ws.on('open', () => {
+			this.connected = true;
 			hLog('Websocket connected!');
 			if (onConnected) {
 				onConnected();
@@ -31,15 +33,21 @@ export class StateHistorySocket {
 		});
 		this.ws.on('message', (data) => onMessage(data));
 		this.ws.on('close', () => {
+			this.connected = false;
 			hLog('Websocket disconnected!');
-			onDisconnect();
+			if(this.retryOnDisconnect) {
+				onDisconnect();
+			}
 		});
 		this.ws.on('error', (err) => {
 			hLog(`${this.shipUrl} :: ${err.message}`);
 		});
 	}
 
-	close() {
+	close(graceful: boolean) {
+		if(graceful) {
+			this.retryOnDisconnect = false;
+		}
 		this.ws.close();
 	}
 

definitions/index-templates.ts

@@ -50,6 +50,16 @@ if (cm.config.settings.hot_warm_policy) {
 	actionSettings["routing"] = {"allocation": {"exclude": {"data": "warm"}}};
 }
 
+const transferProps = {
+	"properties": {
+		"from": {"type": "keyword"},
+		"to": {"type": "keyword"},
+		"amount": {"type": "float"},
+		"symbol": {"type": "keyword"},
+		"memo": {"type": "text"}
+	}
+};
+
 export const action = {
 	order: 0,
 	index_patterns: [
@@ -78,7 +88,6 @@ export const action = {
 			"abi_sequence": {"type": "integer"},
 			"trx_id": {"type": "keyword"},
 			"producer": {"type": "keyword"},
-			"notified": {"type": "keyword"},
 			"signatures": {"enabled": false},
 			"inline_count": {"type": "short"},
 			"max_inline": {"type": "short"},
@@ -116,15 +125,7 @@ export const action = {
 			},
 
 			// *::transfer
-			"@transfer": {
-				"properties": {
-					"from": {"type": "keyword"},
-					"to": {"type": "keyword"},
-					"amount": {"type": "float"},
-					"symbol": {"type": "keyword"},
-					"memo": {"type": "text"}
-				}
-			},
+			"@transfer": transferProps,
 
 			// eosio::unstaketorex
 			"@unstaketorex": {

interfaces/hyperionConfig.ts

@@ -103,6 +103,7 @@ interface ApiConfigs {
     access_log: boolean;
     chain_name: string;
     server_port: number;
+    stream_port: number;
     server_addr: string;
     server_name: string;
     provider_name: string;

modules/action_data/daobet-eosio-delegatebw.js

@@ -1,45 +0,0 @@
-const hyperionModule = {
-    chain: "74d023a9293d9b68c3c52e2f738ee681c1671cc3dc0f263cf2c533cd5523ff95",
-    contract: 'eosio',
-    action: 'delegatebw',
-    parser_version: ['1.7'],
-    defineQueryPrefix: 'delegatebw',
-    mappings: {
-        action: {
-            "@delegatebw": {
-                "properties": {
-                    "from": {"type": "keyword"},
-                    "receiver": {"type": "keyword"},
-                    "stake_cpu_quantity": {"type": "float"},
-                    "stake_net_quantity": {"type": "float"},
-                    "stake_vote_quantity": {"type": "float"},
-                    "transfer": {"type": "boolean"},
-                    "amount": {"type": "float"}
-                }
-            }
-        }
-    },
-    handler: (action) => {
-        const data = action['act']['data'];
-        let cpu_qtd = null;
-        let net_qtd = null;
-        let vote_qtd = null;
-        if (data['stake_net_quantity'] && data['stake_cpu_quantity'] && data['stake_vote_quantity']) {
-            cpu_qtd = parseFloat(data['stake_cpu_quantity'].split(' ')[0]);
-            net_qtd = parseFloat(data['stake_net_quantity'].split(' ')[0]);
-            vote_qtd = parseFloat(data['stake_vote_quantity'].split(' ')[0]);
-        }
-        action['@delegatebw'] = {
-            amount: cpu_qtd + net_qtd + vote_qtd,
-            stake_cpu_quantity: cpu_qtd,
-            stake_net_quantity: net_qtd,
-            stake_vote_quantity: vote_qtd,
-            from: data['from'],
-            receiver: data['receiver'],
-            transfer: data['transfer']
-        };
-        delete action['act']['data'];
-    }
-};
-
-module.exports = {hyperionModule};

modules/action_data/daobet-eosio-undelegatebw.js

@@ -1,43 +0,0 @@
-const hyperionModule = {
-    chain: "74d023a9293d9b68c3c52e2f738ee681c1671cc3dc0f263cf2c533cd5523ff95",
-    contract: 'eosio',
-    action: 'undelegatebw',
-    parser_version: ['1.7'],
-    defineQueryPrefix: 'undelegatebw',
-    mappings: {
-        action: {
-            "@undelegatebw": {
-                "properties": {
-                    "from": {"type": "keyword"},
-                    "receiver": {"type": "keyword"},
-                    "unstake_cpu_quantity": {"type": "float"},
-                    "unstake_net_quantity": {"type": "float"},
-                    "unstake_vote_quantity": {"type": "float"},
-                    "amount": {"type": "float"}
-                }
-            }
-        }
-    },
-    handler: (action) => {
-        const data = action['act']['data'];
-        let cpu_qtd = null;
-        let net_qtd = null;
-        let vote_qtd = null;
-        if (data['unstake_net_quantity'] && data['unstake_cpu_quantity'] && data['unstake_vote_quantity']) {
-            cpu_qtd = parseFloat(data['unstake_cpu_quantity'].split(' ')[0]);
-            net_qtd = parseFloat(data['unstake_net_quantity'].split(' ')[0]);
-            vote_qtd = parseFloat(data['unstake_vote_quantity'].split(' ')[0]);
-        }
-        action['@undelegatebw'] = {
-            amount: cpu_qtd + net_qtd + vote_qtd,
-            unstake_cpu_quantity: cpu_qtd,
-            unstake_net_quantity: net_qtd,
-            unstake_vote_quantity: vote_qtd,
-            from: data['from'],
-            receiver: data['receiver']
-        };
-        delete action['act']['data'];
-    }
-};
-
-module.exports = {hyperionModule};

modules/action_data/eosio-buyram.js

@@ -1,20 +1,20 @@
 const hyperionModule = {
-  chain: '*',
-  contract: 'eosio',
-  action: 'buyram',
-  defineQueryPrefix: 'buyram',
-  parser_version: ['2.1', '1.8', '1.7'],
-  handler: (action) => {
-    const data = action['act']['data'];
-    action['@buyram'] = {
-      payer: data['payer'],
-      receiver: data['receiver'],
-    };
-    if (data['quant']) {
-      action['@buyram']['quant'] = parseFloat(data['quant'].split(' ')[0]);
-    }
-    delete action['act']['data'];
-  },
+    chain: '*',
+    contract: 'eosio',
+    action: 'buyram',
+    defineQueryPrefix: 'buyram',
+    parser_version: ['3.2', '2.1', '1.8', '1.7'],
+    handler: (action) => {
+        const data = action['act']['data'];
+        action['@buyram'] = {
+            payer: data['payer'],
+            receiver: data['receiver'],
+        };
+        if (data['quant']) {
+            action['@buyram']['quant'] = parseFloat(data['quant'].split(' ')[0]);
+        }
+        delete action['act']['data'];
+    },
 };
 
 module.exports = {hyperionModule};

modules/action_data/eosio-buyrambytes.js

@@ -3,7 +3,7 @@ const hyperionModule = {
     contract: 'eosio',
     action: 'buyrambytes',
     defineQueryPrefix: 'buyrambytes',
-    parser_version: ['2.1','1.8','1.7'],
+    parser_version: ['3.2', '2.1','1.8','1.7'],
     handler: (action) => {
         const data = action['act']['data'];
         action['@buyrambytes'] = {

modules/action_data/eosio-buyrex.js

@@ -3,7 +3,7 @@ const hyperionModule = {
     contract: 'eosio',
     action: 'buyrex',
     defineQueryPrefix: 'buyrex',
-    parser_version: ['2.1','1.8','1.7'],
+    parser_version: ['3.2', '2.1','1.8','1.7'],
     handler: (action) => {
         const data = action['act']['data'];
         let qtd = null;

modules/action_data/eosio-delegatebw.js

@@ -2,7 +2,7 @@ const hyperionModule = {
     chain: "*",
     contract: 'eosio',
     action: 'delegatebw',
-    parser_version: ['2.1','1.8','1.7'],
+    parser_version: ['3.2', '2.1','1.8','1.7'],
     defineQueryPrefix: 'delegatebw',
     handler: (action) => {
         const data = action['act']['data'];

modules/action_data/eosio-newaccount.js

@@ -2,7 +2,7 @@ const hyperionModule = {
     chain: "*",
     contract: 'eosio',
     action: 'newaccount',
-    parser_version: ['2.1','1.8','1.7'],
+    parser_version: ['3.2', '2.1','1.8','1.7'],
     defineQueryPrefix: 'newaccount',
     handler: (action) => {
         let name = null;

modules/action_data/eosio-undelegatebw.js

@@ -2,7 +2,7 @@ const hyperionModule = {
     chain: "*",
     contract: 'eosio',
     action: 'undelegatebw',
-    parser_version: ['2.1','1.8','1.7'],
+    parser_version: ['3.2', '2.1','1.8','1.7'],
     defineQueryPrefix: 'undelegatebw',
     handler: (action) => {
         const data = action['act']['data'];

modules/action_data/eosio-unstaketorex.js

@@ -2,7 +2,7 @@ const hyperionModule = {
     chain: "*",
     contract: 'eosio',
     action: 'unstaketorex',
-    parser_version: ['2.1','1.8','1.7'],
+    parser_version: ['3.2', '2.1','1.8','1.7'],
     defineQueryPrefix: 'unstaketorex',
     handler: (action) => {
         const data = action['act']['data'];

modules/action_data/eosio-updateauth.js

@@ -2,7 +2,7 @@ const hyperionModule = {
     chain: "*",
     contract: 'eosio',
     action: 'updateauth',
-    parser_version: ['2.1','1.8','1.7'],
+    parser_version: ['3.2', '2.1','1.8','1.7'],
     defineQueryPrefix: 'updateauth',
     handler: (action) => {
         const data = action['act']['data'];

modules/action_data/eosio-voteproducer.js

@@ -2,7 +2,7 @@ const hyperionModule = {
     chain: "*",
     contract: 'eosio',
     action: 'voteproducer',
-    parser_version: ['2.1','1.8', '1.7'],
+    parser_version: ['3.2', '2.1','1.8', '1.7'],
     defineQueryPrefix: 'voteproducer',
     mappings: {
         action: {

modules/action_data/fio-bprewdupdate.js

@@ -2,7 +2,7 @@ const hyperionModule = {
   chain: '21dcae42c0182200e93f954a074011f9048a7624c6fe81d3c9541a614a88bd1c',
   contract: 'fio.treasury',
   action: 'bprewdupdate',
-  parser_version: ['1.8'],
+  parser_version: ['3.2', '1.8'],
   defineQueryPrefix: 'bprewdupdate',
   mappings: {
     action: {

modules/action_data/fio-fdtnrwdupdat.js

@@ -2,7 +2,7 @@ const hyperionModule = {
   chain: '21dcae42c0182200e93f954a074011f9048a7624c6fe81d3c9541a614a88bd1c',
   contract: 'fio.treasury',
   action: 'fdtnrwdupdat',
-  parser_version: ['1.8'],
+  parser_version: ['3.2', '1.8'],
   defineQueryPrefix: 'fdtnrwdupdat',
   mappings: {
     action: {

modules/action_data/transfer.js

@@ -4,7 +4,7 @@ const hyperionModule = {
   chain: '*',
   contract: '*',
   action: 'transfer',
-  parser_version: ['2.1','1.8', '1.7'],
+  parser_version: ['3.2', '2.1', '1.8', '1.7'],
   defineQueryPrefix: 'transfer',
   handler: (action) => {
     let qtd = null;

modules/master.ts

@@ -234,6 +234,8 @@ export class HyperionMaster {
                         tx: msg.trx_ids
                     });
 
+                    this.lastProducedBlockNum = msg.block_num;
+
                     if (this.conf.settings.bp_monitoring && !this.conf.indexer.abi_scan_mode) {
                         this.liveBlockQueue.push(msg).catch(reason => {
                             hLog('Error handling consumed_block:', reason);
@@ -287,6 +289,16 @@ export class HyperionMaster {
                     this.total_range = this.head - msg.block_num;
                 }
             },
+            'kill_worker': (msg: any) => {
+                for (let workersKey in cluster.workers) {
+                    const w = cluster.workers[workersKey];
+                    if (w.id === parseInt(msg.id)) {
+                        const idx = this.workerMap.findIndex(value => value.worker_id === w.id);
+                        this.workerMap.splice(idx, 1);
+                        w.kill();
+                    }
+                }
+            },
             'completed': (msg: any) => {
                 if (msg.id === this.doctorId.toString()) {
                     hLog('repair worker completed', msg);
@@ -316,17 +328,38 @@ export class HyperionMaster {
                         if (end > this.head) {
                             end = this.head;
                         }
-                        this.lastAssignedBlock += this.maxBatchSize;
                         const def = {
                             first_block: start,
                             last_block: end
                         };
+                        this.lastAssignedBlock = def.last_block;
                         this.activeReadersCount++;
                         messageAllWorkers(cluster, {
                             event: 'new_range',
                             target: msg.id,
                             data: def
                         });
+                    } else {
+                        if (this.lastAssignedBlock >= this.head) {
+                            hLog(`Parallel readers finished the requested range`);
+                            const readers = this.workerMap.filter(value => value.worker_role === 'reader');
+                            this.workerMap = this.workerMap.filter(value => value.worker_role !== 'reader');
+                            readers.forEach(value => value.wref.kill());
+                            // for (let hyperionWorkerDef of this.workerMap) {
+                            //     if (hyperionWorkerDef.worker_role === 'reader') {
+                            //         hyperionWorkerDef.wref.kill();
+                            //     }
+                            // }
+                            // for (let workersKey in cluster.workers) {
+                            //     const w = cluster.workers[workersKey];
+                            //     console.log(w);
+                            //     if (w.id === parseInt(msg.id)) {
+                            //         const idx = this.workerMap.findIndex(value => value.worker_id === w.id);
+                            //         this.workerMap.splice(idx, 1);
+                            //         w.kill();
+                            //     }
+                            // }
+                        }
                     }
                 }
             },
@@ -645,7 +678,7 @@ export class HyperionMaster {
                             index: new_index
                         });
                     } catch (e) {
-                        console.log(e);
+                        hLog(e);
                         process.exit(1);
                     }
 
@@ -657,7 +690,7 @@ export class HyperionMaster {
                             name: `${queue_prefix}-${index.type}`
                         });
                     } catch (e) {
-                        console.log(e);
+                        hLog(e);
                         process.exit(1);
                     }
 
@@ -1451,6 +1484,12 @@ export class HyperionMaster {
                 avg_consume_rate = consume_rate;
             }
             const log_msg = [];
+
+            // print current head for live reading
+            if (this.lastProducedBlockNum > 0) {
+                log_msg.push(`#${this.lastProducedBlockNum}`);
+            }
+
             log_msg.push(`W:${_workers}`);
 
             const _r = (this.pushedBlocks + this.livePushedBlocks) / tScale;
@@ -1474,9 +1513,12 @@ export class HyperionMaster {
             this.metrics.indexingRate?.set(_ir);
 
             if (this.total_blocks < this.total_range && !this.conf.indexer.live_only_mode) {
-                const remaining = this.total_range - this.total_blocks;
-                const estimated_time = Math.round(remaining / avg_consume_rate);
-                const time_string = moment().add(estimated_time, 'seconds').fromNow(false);
+                let time_string = 'waiting for indexer';
+                if (avg_consume_rate > 0) {
+                    const remaining = this.total_range - this.total_blocks;
+                    const estimated_time = Math.round(remaining / avg_consume_rate);
+                    time_string = moment().add(estimated_time, 'seconds').fromNow(false);
+                }
                 const pct_parsed = ((this.total_blocks / this.total_range) * 100).toFixed(1);
                 const pct_read = ((this.total_read / this.total_range) * 100).toFixed(1);
                 log_msg.push(`${this.total_blocks}/${this.total_read}/${this.total_range}`);
@@ -1500,7 +1542,11 @@ export class HyperionMaster {
                 hLog(log_msg.join(' | '));
             }
 
-            if (this.indexedObjects === 0 && this.deserializedActions === 0 && this.consumedBlocks === 0 && !this.mode_transition) {
+            if (this.liveConsumedBlocks > 0 && this.consumedBlocks === 0 && this.conf.indexer.abi_scan_mode) {
+                hLog('Warning: Live reading on ABI SCAN mode')
+            }
+
+            if (this.liveConsumedBlocks + this.indexedObjects + this.deserializedActions + this.consumedBlocks === 0 && !this.mode_transition) {
 
                 // Report completed range (parallel reading)
                 if (this.total_blocks === this.total_range && !this.range_completed) {
@@ -1549,6 +1595,13 @@ export class HyperionMaster {
                             }
                         } else {
                             if (!this.shutdownStarted) {
+                                const readers = this.workerMap.filter(value => {
+                                    return value.worker_role === 'reader' || value.worker_role === 'continuous_reader';
+                                });
+                                if (readers.length === 0) {
+                                    hLog(`No more active workers, stopping now...`);
+                                    process.exit();
+                                }
                                 const idleMsg = 'No blocks are being processed, please check your state-history node!';
                                 if (this.idle_count === 2) {
                                     this.emitAlert('warning', idleMsg);
@@ -1715,7 +1768,16 @@ export class HyperionMaster {
         this.ioRedisClient = new IORedis(this.manager.conn.redis);
 
         // Remove first indexed block from cache (v2/health)
-        await this.ioRedisClient.del(`${this.manager.chain}::fib`)
+        await this.ioRedisClient.del(`${this.manager.chain}::fib`);
+
+        // check nodeos
+        try {
+            const info = await this.rpc.get_info();
+            hLog(`Nodeos version: ${info.server_version_string}`);
+        } catch (e) {
+            hLog(`Chain API Error: ${e.message}`);
+            process.exit();
+        }
 
         // Elasticsearch
         this.client = this.manager.elasticsearchClient;
@@ -1724,6 +1786,7 @@ export class HyperionMaster {
             hLog(`Elasticsearch: ${esInfo.body.version.number} | Lucene: ${esInfo.body.version.lucene_version}`);
             this.emitAlert('info', `Indexer started using ES v${esInfo.body.version.number}`);
         } catch (e) {
+            console.log(e);
             hLog('Failed to check elasticsearch version!');
             process.exit();
         }
@@ -1821,9 +1884,9 @@ export class HyperionMaster {
         // handle worker disconnection events
         cluster.on('disconnect', (worker) => {
             if (!this.mode_transition && !this.shutdownStarted) {
-                hLog(`The worker #${worker.id} has disconnected, attempting to re-launch in 5 seconds...`);
                 const workerReference = this.workerMap.find(value => value.worker_id === worker.id);
                 if (workerReference) {
+                    hLog(`The worker #${worker.id} has disconnected, attempting to re-launch in 5 seconds...`);
                     workerReference.wref = null;
                     workerReference.failures++;
                     hLog(`New worker defined: ${workerReference.worker_role} for ${workerReference.worker_queue}`);
@@ -1839,7 +1902,7 @@ export class HyperionMaster {
                         }, 1000);
                     }, 5000);
                 } else {
-                    console.log(`Worker #${worker.id} not found in map!`);
+                    hLog(`The worker #${worker.id} has disconnected`);
                 }
             }
         });
@@ -2215,9 +2278,7 @@ export class HyperionMaster {
         await this.setupIndexers();
         await this.setupStreaming();
         await this.setupDSPool();
-        this.addWorker({
-            worker_role: "delta_updater"
-        });
+        this.addWorker({worker_role: "delta_updater"});
     }
 
     private async findRange() {
@@ -2243,8 +2304,7 @@ export class HyperionMaster {
         try {
             this.chain_data = await this.rpc.get_info();
         } catch (e) {
-            console.log(e.message);
-            console.error('failed to connect to chain api');
+            hLog('Failed to connect to chain api: ' + e.message);
             process.exit(1);
         }
         this.head = this.chain_data.head_block_num;

modules/parsers/3.2-parser.ts

@@ -1,31 +1,27 @@
+// noinspection JSUnusedGlobalSymbols
+
 import {BaseParser} from "./base-parser";
 import MainDSWorker from "../../workers/deserializer";
 import {Message} from "amqplib";
 import DSPoolWorker from "../../workers/ds-pool";
 import {TrxMetadata} from "../../interfaces/trx-metadata";
 import {ActionTrace} from "../../interfaces/action-trace";
-import {hLog} from "../../helpers/common_functions";
-import {unzip} from "zlib";
-import {omit} from "lodash";
-
-async function unzipAsync(data) {
-    return new Promise((resolve, reject) => {
-        const buf = Buffer.from(data, 'hex');
-        unzip(buf, (err, result) => {
-            if (err) {
-                reject(err);
-            } else {
-                resolve(result);
-            }
-        })
-    });
-}
+import {deserialize, hLog} from "../../helpers/common_functions";
 
 export default class HyperionParser extends BaseParser {
 
-    flatten = true;
+    flatten = false;
 
-    public async parseAction(worker: DSPoolWorker, ts, action: ActionTrace, trx_data: TrxMetadata, _actDataArray, _processedTraces: ActionTrace[], full_trace, usageIncluded): Promise<boolean> {
+    public async parseAction(
+        worker: DSPoolWorker,
+        ts,
+        action: ActionTrace,
+        trx_data: TrxMetadata,
+        _actDataArray,
+        _processedTraces: ActionTrace[],
+        full_trace,
+        usageIncluded
+    ): Promise<boolean> {
 
         // check filters
         if (this.checkBlacklist(action.act)) return false;
@@ -37,35 +33,32 @@ export default class HyperionParser extends BaseParser {
 
         action["@timestamp"] = ts;
         action.block_num = trx_data.block_num;
+        action.block_id = trx_data.block_id;
         action.producer = trx_data.producer;
         action.trx_id = trx_data.trx_id;
 
+        action.account_ram_deltas = action.account_ram_deltas.filter(value => value.delta !== '0')
+
         if (action.account_ram_deltas.length === 0) {
             delete action.account_ram_deltas;
         }
 
         if (action.except === null) {
-
             if (!action.receipt) {
                 console.log(full_trace.status);
                 console.log(action);
             }
-
             action.receipt = action.receipt[1];
             action.global_sequence = parseInt(action.receipt.global_sequence, 10);
-
             delete action.except;
-            delete action.error_code;
-            delete action.elapsed;
-            delete action.context_free;
-            delete action.level;
-
+            if (action.error_code === null) {
+                delete action.error_code;
+            }
+            // add usage data to the first action on the transaction
             if (!usageIncluded.status) {
                 this.extendFirstAction(worker, action, trx_data, full_trace, usageIncluded);
             }
-
             _processedTraces.push(action);
-
         } else {
             hLog(action);
         }
@@ -75,7 +68,8 @@ export default class HyperionParser extends BaseParser {
 
     public async parseMessage(worker: MainDSWorker, messages: Message[]): Promise<void> {
         for (const message of messages) {
-            const ds_msg = worker.deserializeNative('result', message.content);
+            let allowProcessing = true;
+            const ds_msg = deserialize('result', message.content, this.txEnc, this.txDec, worker.types);
             if (!ds_msg) {
                 if (worker.ch_ready) {
                     worker.ch.nack(message);
@@ -83,34 +77,58 @@ export default class HyperionParser extends BaseParser {
                 }
             }
             const res = ds_msg[1];
-            let block, traces = [], deltas = [];
+            let block = null;
+            let traces = [];
+            let deltas = [];
+
             if (res.block && res.block.length) {
+
                 block = worker.deserializeNative('signed_block', res.block);
-                if (!block) {
-                    hLog('null block', res);
+
+                if (block === null) {
+                    hLog('incompatible block');
                     process.exit(1);
                 }
-            }
-            if (res['traces'] && res['traces'].length) {
-                try {
-                    traces = worker.deserializeNative(
-                        'transaction_trace[]',
-                        await unzipAsync(res['traces'])
-                    );
-                } catch (e) {
-                    hLog(e);
+
+                // verify for whitelisted contracts (root actions only)
+                if (worker.conf.whitelists.root_only) {
+                    try {
+                        if (worker.conf.whitelists && (worker.conf.whitelists.actions.length > 0 || worker.conf.whitelists.deltas.length > 0)) {
+                            allowProcessing = false;
+                            for (const transaction of block.transactions) {
+                                if (transaction.status === 0 && transaction.trx[1] && transaction.trx[1].packed_trx) {
+                                    const unpacked_trx = worker.api.deserializeTransaction(Buffer.from(transaction.trx[1].packed_trx, 'hex'));
+                                    for (const act of unpacked_trx.actions) {
+                                        if (this.checkWhitelist(act)) {
+                                            allowProcessing = true;
+                                            break;
+                                        }
+                                    }
+                                    if (allowProcessing) break;
+                                }
+                            }
+                        }
+                    } catch (e) {
+                        console.log(e);
+                        allowProcessing = true;
+                    }
                 }
             }
-            if (res['deltas'] && res['deltas'].length) {
-                try {
-                    deltas = worker.deserializeNative(
-                        'table_delta[]',
-                        await unzipAsync(res['deltas'])
-                    );
-                } catch (e) {
-                    hLog(e);
+
+            if (allowProcessing && res.traces && res.traces.length) {
+                traces = worker.deserializeNative('transaction_trace[]', res.traces);
+                if (!traces) {
+                    hLog(`[WARNING] transaction_trace[] deserialization failed on block ${res['this_block']['block_num']}`);
                 }
             }
+
+            if (allowProcessing && res.deltas && res.deltas.length) {
+                deltas = deserialize('table_delta[]', res.deltas, this.txEnc, this.txDec, worker.types);
+                if (!deltas) {
+                    hLog(`[WARNING] table_delta[] deserialization failed on block ${res['this_block']['block_num']}`);
+                }
+            }
+
             let result;
             try {
                 result = await worker.processBlock(res, block, traces, deltas);
@@ -144,28 +162,9 @@ export default class HyperionParser extends BaseParser {
         }
     }
 
-    async flattenInlineActions(action_traces: any[], level: number, trace_counter: any, parent_index: number): Promise<any[]> {
-        const arr = [];
-        const nextLevel = [];
-        for (const action_trace of action_traces) {
-            const trace = action_trace[1];
-            trace['creator_action_ordinal'] = parent_index;
-            trace_counter.trace_index++;
-            trace['level'] = level;
-            trace['action_ordinal'] = trace_counter.trace_index;
-            arr.push(["action_trace_v0", omit(trace, "inline_traces")]);
-            if (trace.inline_traces && trace.inline_traces.length > 0) {
-                nextLevel.push({
-                    traces: trace.inline_traces,
-                    parent: trace_counter.trace_index
-                });
-            }
-        }
-        for (const data of nextLevel) {
-            const appendedArray = await this.flattenInlineActions(data.traces, level + 1, trace_counter, data.parent);
-            arr.push(...appendedArray);
-        }
-        return Promise.resolve(arr);
+    async flattenInlineActions(action_traces: any[]): Promise<any[]> {
+        hLog(`Calling undefined flatten operation!`);
+        return Promise.resolve(undefined);
     }
 
 }

modules/parsers/base-parser.ts

@@ -70,7 +70,13 @@ export abstract class BaseParser {
         return this.filters.action_whitelist.has(this.codeActionPair(act));
     }
 
-    protected extendFirstAction(worker: DSPoolWorker, action: ActionTrace, trx_data: TrxMetadata, full_trace: any, usageIncluded) {
+    protected extendFirstAction(
+        worker: DSPoolWorker,
+        action: ActionTrace,
+        trx_data: TrxMetadata,
+        full_trace: any,
+        usageIncluded: { status: boolean }
+    ) {
         action.cpu_usage_us = trx_data.cpu_usage_us;
         action.net_usage_words = trx_data.net_usage_words;
         action.signatures = trx_data.signatures;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hyperion-history",
-  "version": "3.3.7",
+  "version": "3.3.9",
   "description": "Scalable Full History API Solution for EOSIO based blockchains",
   "main": "launcher.js",
   "scripts": {
@@ -9,6 +9,7 @@
     "start:indexer": "pm2 start --only Indexer --update-env",
     "tsc": "tsc",
     "build": "tsc",
+    "build:watch": "tsc --watch",
     "postinstall": "npm run build && npm run fix-permissions",
     "fix-permissions": "node scripts/fix-permissions.js"
   },
@@ -29,8 +30,8 @@
     "@eosrio/node-abieos": "^2.1.1",
     "@fastify/autoload": "4.0.1",
     "@fastify/cors": "7.0.0",
-    "@fastify/formbody": "^6.0.0",
-    "@fastify/rate-limit": "^6.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",
@@ -40,8 +41,8 @@
     "commander": "^8.3.0",
     "cross-fetch": "^3.1.5",
     "eosjs": "^22.1.0",
-    "fast-json-stringify": "^2.7.9",
-    "fastify": "^3.29.3",
+    "fast-json-stringify": "2.7.13",
+    "fastify": "3.29.4",
     "fastify-elasticsearch": "^2.0.0",
     "fastify-plugin": "^3.0.1",
     "flatstr": "^1.0.12",
@@ -50,28 +51,29 @@
     "ioredis": "^4.28.5",
     "lodash": "^4.17.21",
     "moment": "^2.29.4",
-    "nodemailer": "^6.7.8",
+    "nodemailer": "^6.9.0",
     "portfinder": "^1.0.32",
-    "socket.io": "4.5.3",
-    "socket.io-client": "4.5.3",
+    "pino-pretty": "^9.1.1",
+    "socket.io": "4.5.4",
+    "socket.io-client": "4.5.4",
     "socket.io-redis": "^6.1.1",
-    "telegraf": "^4.9.1",
+    "telegraf": "^4.11.2",
     "typescript": "^4.5.2",
-    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.15.0",
-    "ws": "^8.11.0"
+    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.19.0",
+    "ws": "^8.12.0"
   },
   "devDependencies": {
     "@types/amqplib": "^0.8.2",
-    "@types/async": "^3.2.10",
+    "@types/async": "^3.2.16",
     "@types/global-agent": "^2.1.1",
-    "@types/ioredis": "^4.28.1",
-    "@types/lodash": "^4.14.177",
-    "@types/node": "^18.11.9",
-    "@types/nodemailer": "^6.4.4",
-    "@types/ws": "^8.5.3"
+    "@types/ioredis": "^4.28.10",
+    "@types/lodash": "^4.14.191",
+    "@types/node": "^18.11.18",
+    "@types/nodemailer": "^6.4.7",
+    "@types/ws": "^8.5.4"
   },
   "optionalDependencies": {
     "bufferutil": "^4.0.7",
-    "utf-8-validate": "^5.0.7"
+    "utf-8-validate": "^5.0.10"
   }
 }

workers/ds-pool.ts

@@ -21,6 +21,48 @@ interface CustomAbiDef {
     endingBlock: number;
 }
 
+function cleanActionTrace(t: any) {
+    try {
+        if (t.return_value === '') {
+            delete t.return_value;
+        }
+        if (t.context_free === false) {
+            delete t.context_free;
+        }
+        if (t.elapsed === '0') {
+            delete t.elapsed;
+        }
+
+        // remove act_digest from grouped receipts since it was written to the action
+        if (t.receipts && t.receipts.length > 0) {
+            t.act_digest = t.receipts[0].act_digest;
+            for (let receipt of t.receipts) {
+                delete receipt.act_digest;
+            }
+        } else {
+            delete t.receipts;
+        }
+
+        delete t.console;
+        delete t.receiver;
+
+        // onblock action case
+        if (t.signatures && t.signatures.length === 0) {
+            delete t.signatures;
+        }
+
+        if (t.inline_count === 0) {
+            delete t.inline_count;
+        }
+
+        if (t.net_usage_words === 0) {
+            delete t.net_usage_words;
+        }
+    } catch (e) {
+        console.log(e);
+    }
+}
+
 export default class DSPoolWorker extends HyperionWorker {
 
     abi;
@@ -256,7 +298,7 @@ export default class DSPoolWorker extends HyperionWorker {
         const [_status, actionType] = await self.verifyLocalType(_action.account, _action.name, block_num, "action");
         if (_status) {
             try {
-                return JSON.parse(AbiEOS.bin_to_json(_action.account, actionType, Buffer.from(_action.data, 'hex')));
+                return AbiEOS.bin_to_json(_action.account, actionType, Buffer.from(_action.data, 'hex'));
             } catch (e) {
                 debugLog(`(abieos) ${_action.account}::${_action.name} @ ${block_num} >>> ${e.message}`);
             }
@@ -464,20 +506,12 @@ export default class DSPoolWorker extends HyperionWorker {
             }
 
             const _finalTraces = [];
+
             if (_processedTraces.length > 1) {
                 const act_digests = {};
 
                 // collect digests & receipts
                 for (const _trace of _processedTraces) {
-
-                    if (this.conf.settings.dsp_parser) {
-                        if (_trace.console !== '') {
-                            await parseDSPEvent(this, _trace);
-                        }
-                    } else {
-                        delete _trace.console;
-                    }
-
                     if (act_digests[_trace.receipt.act_digest]) {
                         act_digests[_trace.receipt.act_digest].push(_trace.receipt);
                     } else {
@@ -488,18 +522,17 @@ export default class DSPoolWorker extends HyperionWorker {
                 // Apply notified accounts to first trace instance
                 for (const _trace of _processedTraces) {
                     if (act_digests[_trace.receipt.act_digest]) {
-                        const notifiedSet = new Set();
+                        // const notifiedSet = new Set();
                         _trace['receipts'] = [];
                         for (const _receipt of act_digests[_trace.receipt.act_digest]) {
-                            notifiedSet.add(_receipt.receiver);
+                            // notifiedSet.add(_receipt.receiver);
                             _trace['code_sequence'] = _receipt['code_sequence'];
                             delete _receipt['code_sequence'];
                             _trace['abi_sequence'] = _receipt['abi_sequence'];
                             delete _receipt['abi_sequence'];
-                            delete _receipt['act_digest'];
                             _trace['receipts'].push(_receipt);
                         }
-                        _trace['notified'] = [...notifiedSet];
+                        // _trace['notified'] = [...notifiedSet];
                         delete act_digests[_trace.receipt.act_digest];
                         delete _trace['receipt'];
                         delete _trace['receiver'];
@@ -514,10 +547,12 @@ export default class DSPoolWorker extends HyperionWorker {
                 _trace['code_sequence'] = _trace['receipt'].code_sequence;
                 _trace['abi_sequence'] = _trace['receipt'].abi_sequence;
                 _trace['act_digest'] = _trace['receipt'].act_digest;
-                _trace['notified'] = [_trace['receipt'].receiver];
+
+                // notified array is not required since receipts.receiver can be indexed directly
+                // _trace['notified'] = [_trace['receipt'].receiver];
+
                 delete _trace['receipt']['code_sequence'];
                 delete _trace['receipt']['abi_sequence'];
-                delete _trace['receipt']['act_digest'];
                 _trace['receipts'] = [_trace['receipt']];
                 delete _trace['receipt'];
                 _finalTraces.push(_trace);
@@ -528,6 +563,7 @@ export default class DSPoolWorker extends HyperionWorker {
             const redisPayload = new Map<string, IORedis.ValueType>();
 
             for (const uniqueAction of _finalTraces) {
+                cleanActionTrace(uniqueAction);
                 const payload = Buffer.from(flatstr(JSON.stringify(uniqueAction)));
                 redisPayload.set(uniqueAction.global_sequence.toString(), payload);
                 this.actionDsCounter++;
@@ -566,20 +602,24 @@ export default class DSPoolWorker extends HyperionWorker {
 
     pushToActionStreamingQueue(payload, uniqueAction) {
         if (this.allowStreaming && this.conf.features['streaming'].traces) {
-            const notifArray = new Set();
-            uniqueAction.act.authorization.forEach(auth => {
-                notifArray.add(auth.actor);
-            });
-            uniqueAction.notified.forEach(acc => {
-                notifArray.add(acc);
-            });
-            const headers = {
-                event: 'trace',
-                account: uniqueAction['act']['account'],
-                name: uniqueAction['act']['name'],
-                notified: [...notifArray].join(",")
-            };
-            this.ch.publish('', this.chain + ':stream', payload, {headers});
+            try {
+                const notifArray = new Set();
+                uniqueAction.act.authorization.forEach(auth => {
+                    notifArray.add(auth.actor);
+                });
+                uniqueAction.receipts.forEach(rec => {
+                    notifArray.add(rec.receiver);
+                });
+                const headers = {
+                    event: 'trace',
+                    account: uniqueAction['act']['account'],
+                    name: uniqueAction['act']['name'],
+                    notified: [...notifArray].join(",")
+                };
+                this.ch.publish('', this.chain + ':stream', payload, {headers});
+            } catch (e) {
+                hLog(e);
+            }
         }
     }
 

workers/state-reader.ts

@@ -40,6 +40,7 @@ export default class StateReader extends HyperionWorker {
     private shipInitStatus: any;
 
     private forkedBlocks = new Map<string, number>();
+    private idle = true;
 
     constructor() {
         super();
@@ -122,7 +123,7 @@ export default class StateReader extends HyperionWorker {
                             console.log('Message nacked!');
                             console.log(err.message);
                         } else {
-                            process.send({event: 'read_block', live: this.isLiveReader});
+                            process.send({event: 'read_block', live: this.isLiveReader, block_num: d.num});
                         }
                     });
 
@@ -231,7 +232,7 @@ export default class StateReader extends HyperionWorker {
             case 'stop': {
                 if (this.isLiveReader) {
                     console.log('[LIVE READER] Closing Websocket');
-                    this.ship.close();
+                    this.ship.close(true);
                     setTimeout(() => {
                         console.log('[LIVE READER] Process killed');
                         process.exit(1);
@@ -262,9 +263,7 @@ export default class StateReader extends HyperionWorker {
                             pending++;
                         }
                     });
-                    if (pending === this.lastPendingCount && pending > 0) {
-                        // console.log(`[${process.env['worker_id']}] Pending blocks: ${pending}`);
-                    } else {
+                    if (!(pending === this.lastPendingCount && pending > 0)) {
                         this.lastPendingCount = pending;
                     }
                 }
@@ -310,56 +309,66 @@ export default class StateReader extends HyperionWorker {
 
         if (!process.env.worker_role) {
             console.log("[FATAL ERROR] undefined role! Exiting now!");
-            this.ship.close();
+            this.ship.close(false);
             process.exit(1);
             return;
         }
 
         // NORMAL OPERATION MODE
         if (!this.recovery) {
+
             const result = deserialize('result', data, this.txEnc, this.txDec, this.types);
+
+            // ship status message
             if (result[0] === 'get_status_result_v0') {
                 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) {
+
                     switch (process.env['worker_role']) {
+
+                        // range reader setup
                         case 'reader': {
                             if (chain_state_begin_block > process.env.first_block) {
+
+                                // skip snapshot block until a solution to parse it is found
+                                const nextBlock = chain_state_begin_block + 2;
                                 hLog(`First saved block is ahead of requested range! - Req: ${process.env.first_block} | First: ${chain_state_begin_block}`);
-                                hLog('Requesting a single block');
-                                if (this.conf.settings.ignore_snapshot) {
-                                    this.local_block_num = chain_state_begin_block;
-                                    process.send({event: 'update_init_block', block_num: chain_state_begin_block + 2,});
-                                    this.newRange({
-                                        first_block: chain_state_begin_block + 1,
-                                        last_block: chain_state_begin_block + this.conf.scaling.batch_size
-                                    });
-                                } else {
-                                    process.send({event: 'update_init_block', block_num: chain_state_begin_block + 1});
-                                    this.newRange({
-                                        first_block: chain_state_begin_block,
-                                        last_block: chain_state_begin_block + 1
-                                    });
-                                }
+                                this.local_block_num = nextBlock - 1;
+                                process.send({
+                                    event: 'update_init_block',
+                                    block_num: nextBlock
+                                });
+                                this.newRange({
+                                    first_block: nextBlock,
+                                    last_block: nextBlock + this.conf.scaling.batch_size
+                                });
                             } else {
                                 this.requestBlocks(0);
                             }
                             break;
                         }
+
+                        // live reader setup
                         case 'continuous_reader': {
                             this.requestBlocks(parseInt(process.env['worker_last_processed_block'], 10));
                             break;
                         }
+
                     }
                 } else {
-                    this.ship.close();
+                    this.ship.close(true);
                     process.exit(1);
                 }
+
             } else {
+
+                // ship block message
                 const res = result[1];
 
                 if (res['this_block']) {
+                    this.idle = false;
                     const blk_num = res['this_block']['block_num'];
                     const lib = res['last_irreversible'];
                     const task_payload = {num: blk_num, content: data};
@@ -440,7 +449,15 @@ export default class StateReader extends HyperionWorker {
                         }
                     }
                 } else {
-                    return 0;
+                    this.idle = true;
+                    // hLog(`Reader is idle! - Head at: ${result[1].head.block_num}`);
+                    // this.ship.close(true);
+                    // const queueSize = [this.stageOneDistQueue.length(), this.blockReadingQueue.length()];
+                    // hLog(queueSize);
+                    // process.send({
+                    //     event: 'kill_worker',
+                    //     id: process.env['worker_id']
+                    // });
                 }
             }
 
@@ -500,8 +517,12 @@ export default class StateReader extends HyperionWorker {
         request.start_block_num = parseInt(first_block > 0 ? first_block.toString() : '1', 10);
         request.end_block_num = parseInt(last_block.toString(), 10);
         const reqType = 'get_blocks_request_' + this.shipRev;
-        debugLog(`Reader ${process.env.worker_id} sending ${reqType} from: ${request.start_block_num} to: ${request.end_block_num}`);
-        this.send([reqType, request]);
+        if (this.ship.connected) {
+            debugLog(`Reader ${process.env.worker_id} sending ${reqType} from: ${request.start_block_num} to: ${request.end_block_num}`);
+            this.send([reqType, request]);
+        } else {
+            hLog('[Warning] Request failed - SHIP is not online!');
+        }
     }
 
     private send(req_data: (string | any)[]) {
@@ -514,8 +535,12 @@ export default class StateReader extends HyperionWorker {
         this.local_block_num = request.start_block_num - 1;
         request.end_block_num = parseInt(last, 10);
         const reqType = 'get_blocks_request_' + this.shipRev;
-        debugLog(`Reader ${process.env.worker_id} sending ${reqType} from: ${request.start_block_num} to: ${request.end_block_num}`);
-        this.send([reqType, request]);
+        if (this.ship.connected) {
+            debugLog(`Reader ${process.env.worker_id} sending ${reqType} from: ${request.start_block_num} to: ${request.end_block_num}`);
+            this.send([reqType, request]);
+        } else {
+            hLog('[Warning] Request failed - SHIP is not online!');
+        }
     }
 
     private async deleteForkedBlock(block_id: string) {
@@ -529,6 +554,7 @@ export default class StateReader extends HyperionWorker {
             refresh: true,
             body: searchBody
         });
+
         if (dbqResultDelta.body && dbqResultDelta.statusCode === 200) {
             hLog(`${dbqResultDelta.body.deleted} deltas removed from ${block_id}`);
         } else {
@@ -541,6 +567,7 @@ export default class StateReader extends HyperionWorker {
             refresh: true,
             body: searchBody
         });
+
         if (dbqResultAction.body && dbqResultAction.statusCode === 200) {
             hLog(`${dbqResultAction.body.deleted} traces removed from ${block_id}`);
         } else {
@@ -561,6 +588,7 @@ export default class StateReader extends HyperionWorker {
                 id: targetBlock
             });
             targetBlock++;
+            console.log(blockData);
             if (blockData.body) {
                 const targetBlockId = blockData.body._source.block_id;
                 this.forkedBlocks.set(targetBlockId, Date.now());
@@ -643,7 +671,7 @@ export default class StateReader extends HyperionWorker {
 
     private handleLostConnection() {
         this.recovery = true;
-        this.ship.close();
+        this.ship.close(false);
         hLog(`Retrying connection in 5 seconds... [attempt: ${this.reconnectCount + 1}]`);
         debugLog(`PENDING REQUESTS:', ${this.pendingRequest}`);
         debugLog(`LOCAL BLOCK:', ${this.local_block_num}`);