Skip to content
This repository was archived by the owner on Dec 22, 2024. It is now read-only.

Optimize getBlockByNumber queries #78

Open
guilledk wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Optimize getBlockByNumber queries #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e2c8db1
Optimize getBlockByNumber queries
Mar 15, 2024
a3f4fdd
Fix minor issue was using evm block num for index calc on new optimiz…
Mar 15, 2024
1410b7c
Fix bug when processing genesis block on batch requests
Mar 15, 2024
c217ae4
Add configurable docs per index
Mar 18, 2024
50af881
Fix bugs related to minor changes to elastic document handling after …
Apr 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/main.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,10 @@ program
.option('-c, --config [path to config.json]', 'Path to config.json file', 'config.json')
.action(async (options) => {
const config: TelosEvmConfig = JSON.parse(readFileSync(options.config).toString());

if (!config.elasticIndexDocsAmount)
config.elasticIndexDocsAmount = 1e7;

const rpc: TelosEVMRPC = new TelosEVMRPC(config);

console.log("Starting Telos EVM RPC...");
Expand Down
181 changes: 145 additions & 36 deletions src/routes/evm/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,8 +200,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
}
}

async function getVRS(receiptDoc): Promise<any> {
let receipt = receiptDoc["@raw"];
function getVRS(receipt): {v: string, r: string, s: string} {
const v = removeLeftZeros(BigInt(receipt.v).toString(16), true);
const r = removeLeftZeros(receipt.r, true);
const s = removeLeftZeros(receipt.s, true);
Expand Down Expand Up @@ -292,8 +291,8 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
}

function adjustBlockNum(num: number): number {
// convert to native block num and divide over index size 10 million
return Math.floor((num + opts.blockNumberDelta) / 1e7);
// convert to native block num and divide over dos per index
return Math.floor((num + opts.blockNumberDelta) / opts.elasticIndexDocsAmount);
}

function indexSuffixForBlock(blockNumber: number): string {
Expand All @@ -302,7 +301,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
}

async function getDeltaDocFromNumber(blockNumber: number, retry: number = 0) {
const indexSuffix = indexSuffixForBlock(blockNumber);
const indexSuffix = indexSuffixForBlock(blockNumber + opts.blockNumberDelta);
const results = await fastify.elastic.search({
index: `${opts.elasticIndexPrefix}-delta-${opts.elasticIndexVersion}-${indexSuffix}`,
size: 1,
Expand All @@ -325,7 +324,67 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
return blockDelta;
}

async function emptyBlockFromDelta(blockDelta: any) {
function getIndexSuffixesForBlocks(blockNumbers: number[]) {
const blocksPerSuff = new Map<string, number[]>();
for (const blockNum of blockNumbers) {
const indexSuff = indexSuffixForBlock(blockNum + opts.blockNumberDelta);
const blocks = [blockNum];
if (blocksPerSuff.has(indexSuff))
blocksPerSuff.get(indexSuff).push(...blocks);
else
blocksPerSuff.set(indexSuff, blocks);
}
return blocksPerSuff;
}

async function getMultipleReceipts(blockNumbers: number[]) {
const receipts = [];
for (const [suffix, blocks] of getIndexSuffixesForBlocks(blockNumbers).entries()) {
const results = await fastify.elastic.search({
index: `${opts.elasticIndexPrefix}-action-${opts.elasticIndexVersion}-${suffix}`,
size: 2000,
query: {
terms: {'@raw.block': blocks}
}
}).catch(err => {
console.error(err);
throw err;
});
receipts.push(
...results.hits.hits.map(h => h._source));
}
return receipts;
}

async function getMultipleDeltaDocsFromNumbers(blockNumbers: number[]) {
const blocksWithReceipts = [];
const blocks = [];
for (const [suffix, blockNums] of getIndexSuffixesForBlocks(blockNumbers).entries()) {
const searchBlock = await fastify.elastic.search({
index: `${opts.elasticIndexPrefix}-delta-${opts.elasticIndexVersion}-${suffix}`,
size: blockNums.length,
query: {
terms: {"@global.block_num": blockNums}
}
}).catch(error => {
console.error(error);
throw error;
});
for (const hit of searchBlock.hits.hits) {
const doc: any = hit._source;
if (doc.txAmount)
blocksWithReceipts.push(doc['@global'].block_num)
blocks.push(doc);
}
}
let receipts = [];
if (blocksWithReceipts.length > 0)
receipts = await getMultipleReceipts(blocksWithReceipts);

return [blocks, receipts];
}

function emptyBlockFromDelta(blockDelta: any) {
const blockNumberHex = addHexPrefix(blockDelta['@global'].block_num.toString(16));
const timestamp = new Date(blockDelta['@timestamp']).getTime() / 1000;
const parentHash = addHexPrefix(blockDelta['@evmPrevBlockHash']);
Expand All @@ -350,7 +409,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
if (!blockDelta)
return null;

return await emptyBlockFromDelta(blockDelta);
return emptyBlockFromDelta(blockDelta);
} catch (e) {
Logger.error(e);
return null;
Expand All @@ -373,15 +432,15 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
return null;
}

return await emptyBlockFromDelta(blockDelta);
return emptyBlockFromDelta(blockDelta);
} catch (e) {
Logger.error(e);
return null;
}
}


async function reconstructBlockFromReceipts(receipts: any[], full: boolean, client: any) {
function reconstructBlockFromReceipts(block, receipts: any[], full: boolean, client: any) {
try {
let blockHash;
let blockHex: string;
Expand All @@ -390,9 +449,8 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
let bloom = new Bloom();
const trxs = [];
//Logger.debug(`Reconstructing block from receipts: ${JSON.stringify(receipts)}`)
for (const receiptDoc of receipts) {
const {v, r, s} = await getVRS(receiptDoc._source);
const receipt = receiptDoc._source['@raw'];
for (const receipt of receipts) {
const {v, r, s} = getVRS(receipt);

if (!blockHash) {
blockHash = addHexPrefix(receipt['block_hash']);
Expand Down Expand Up @@ -431,11 +489,6 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
}
}

const block = await getDeltaDocFromNumber(blockNum);
if(!block){
Logger.error("Could not find block for receipts");
return null;
}
const timestamp = new Date(block['@timestamp']).getTime() / 1000;
const gasUsedBlock = addHexPrefix(removeLeftZeros(new BN(block['gasUsed']).toString('hex')));
const extraData = addHexPrefix(block['@blockHash']);
Expand Down Expand Up @@ -473,7 +526,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
size: 2000,
query: { bool: { must: [{ term: termStruct }] } }
});
return results?.hits?.hits;
return results?.hits?.hits.map(h => h._source['@raw']);
}

async function getCurrentBlockNumber(indexed: boolean = false, retry: number = 0) {
Expand Down Expand Up @@ -514,7 +567,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
if (indices.length > 1) {
const docsCount = parseInt(index['docs.count']);
const adjustedNum = indexToSuffixNum(index.index);
lastBlockNum = (adjustedNum * 1e7) + docsCount - opts.blockNumberDelta - 1;
lastBlockNum = (adjustedNum * opts.elasticIndexDocsAmount) + docsCount - opts.blockNumberDelta - 1;
} else if (index?.index) {
const results = await fastify.elastic.search({
index: `${index.index}`,
Expand Down Expand Up @@ -1173,7 +1226,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
// lookup raw action
const receiptAction = await searchActionByHash(trxHash, client);
if (!receiptAction) return null;
const {v, r, s} = await getVRS(receiptAction);
const {v, r, s} = getVRS(receiptAction);
const receipt = receiptAction['@raw'];

// lookup receipt delta
Expand Down Expand Up @@ -1210,10 +1263,20 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
_hash = _hash.slice(2);
}
const receipts = await getReceiptsByTerm("@raw.block_hash", _hash);
const block = receipts.length > 0 ? await reconstructBlockFromReceipts(receipts, true, client) : await emptyBlockFromHash(_hash);
if(!block || block.transactions?.length === 0) return null;
let block;
if (receipts.length == 0)
block = await emptyBlockFromHash(_hash);
if (!block) return null;

else {
const blockNum = receipts[0].block;
const delta = await getDeltaDocFromNumber(blockNum);
if (!delta) return null;
block = reconstructBlockFromReceipts(delta, receipts, true, client);
}

const trxIndex = parseInt(trxIndexHex, 16);
if(!block.transactions[trxIndex]) return null;
if(!block.transactions[trxIndex]) return null;
let trx = block.transactions[trxIndex];
trx.type = "0x0"; // TODO: Determine type with 1559
trx.chainId = CHAIN_ID_HEX;
Expand All @@ -1229,7 +1292,6 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
if (blockNumber === 0)
return GENESIS_BLOCK;


try {
const blockDelta = await getDeltaDocFromNumber(blockNumber);
if(!blockDelta){
Expand All @@ -1238,16 +1300,17 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
}
if (blockDelta['@transactionsRoot'] === NULL_TRIE)
return emptyBlockFromDelta(blockDelta);
else {
const receipts = await getReceiptsByTerm("@raw.block", blockNumber);
return receipts.length > 0 ? reconstructBlockFromReceipts(blockDelta, receipts, full, client) : await emptyBlockFromNumber(blockNumber);
}
} catch (e) {
Logger.error(`Could not find block from block number ${blockNumber}: ${e}`);
if(e?.message.startsWith("index_not_found_exception")){
return null;
}
throw(e);
}

const receipts = await getReceiptsByTerm("@raw.block", blockNumber);
return receipts.length > 0 ? await reconstructBlockFromReceipts(receipts, full, client) : await emptyBlockFromNumber(blockNumber);
});

/**
Expand All @@ -1262,7 +1325,16 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
_hash = _hash.slice(2);
}
const receipts = await getReceiptsByTerm("@raw.block_hash", _hash);
return receipts.length > 0 ? await reconstructBlockFromReceipts(receipts, full, client) : await emptyBlockFromHash(_hash);
let block;
if (receipts.length == 0)
block = await emptyBlockFromHash(_hash);

else {
const blockNum = receipts[0].block;
const delta = await getDeltaDocFromNumber(blockNum - opts.blockNumberDelta);
block = reconstructBlockFromReceipts(delta, receipts, true, client);
}
return block;
});

/**
Expand Down Expand Up @@ -1648,8 +1720,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {
throw new Error("trace_replayBlockTransactions only supports the \"trace\" type of trace (not vmTrace or stateDiff");

const blockNumber = parseInt(await toBlockNumber(block), 16);
const receiptHits = await getReceiptsByTerm("@raw.block", blockNumber);
const receipts = receiptHits.map(r => r._source["@raw"]);
const receipts = await getReceiptsByTerm("@raw.block", blockNumber);
const sortedReceipts = receipts.sort((a, b) => {
return a.trx_index - b.trx_index;
})
Expand All @@ -1666,8 +1737,7 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {

methods.set('trace_block', async ([block]) => {
const blockNumber = parseInt(await toBlockNumber(block), 16);
const receiptHits = await getReceiptsByTerm("@raw.block", blockNumber);
const receipts = receiptHits.map(r => r._source["@raw"]);
const receipts = await getReceiptsByTerm("@raw.block", blockNumber);
const sortedReceipts = receipts.sort((a, b) => {
return a.trx_index - b.trx_index;
})
Expand Down Expand Up @@ -1796,12 +1866,51 @@ export default async function (fastify: FastifyInstance, opts: TelosEvmConfig) {

const tRef = process.hrtime.bigint();

let promises = [];
const promises = [];
// gather all getBlockByNumber requests
const requestedBlockNums = [];
// used to order responses after async processing
const respMap = new Map<number, any>();

// generate promises for request resolution
for (let i = 0; i < payload.length; i++) {
let promise = doRpcMethod(payload[i], clientInfo, reply);
promises.push(promise);
let { jsonrpc, id, method, params } = payload[i];
if (method === 'eth_getBlockByNumber') {
// gather block request info for later
requestedBlockNums.push({params, index: i});
} else {
// normal request
promises.push(doRpcMethod(payload[i], clientInfo, reply).then(resp => respMap.set(i, resp)));
}
}

if (requestedBlockNums.length > 0) {
// perform optimized batch get block by number request
promises.push(
getMultipleDeltaDocsFromNumbers(requestedBlockNums.map(req => req.params[0])).then(
blockData => {
const [blocks, allReceipts] = blockData;
for (const delta of blocks) {
const blockNum = delta['@global'].block_num;
const request = requestedBlockNums.find(req => req.params[0] == blockNum);
if (!delta.txAmount) {
respMap.set(request.index, emptyBlockFromDelta(delta));
continue;
}

const receipts = allReceipts.find(r => r["@raw"].block == blockNum);
respMap.set(request.index, reconstructBlockFromReceipts(delta, receipts, request.params[1], clientInfo));
}
}
));
}
let responses = await Promise.all(promises);

await Promise.all(promises);

// gather results in same order user requested
const responses = []
for (let i = 0; i < respMap.size; i++)
responses.push(respMap.get(i));

const duration = ((Number(process.hrtime.bigint()) - Number(tRef)) / 1000).toFixed(3);
Logger.log(`RPCREQUESTBATCH: ${new Date().toISOString()} - ${duration} μs - ${ip} (${usage}/${limit}) - ${origin} - BATCH OF ${responses.length}`);
Expand Down
1 change: 1 addition & 0 deletions src/types/index.d.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ export interface TelosEvmConfig {
elasticPass: string;
elasticIndexPrefix: string;
elasticIndexVersion: string;
elasticIndexDocsAmount: number;
orderNonces: boolean;
orderNonceRetryTimeout: number;
syncingThreshhold: number;
Expand Down