parallel rpc batching

This commit is contained in:
gozzy 2022-11-11 02:01:46 +00:00
parent 85a222d93a
commit 61c2865c46
4 changed files with 85 additions and 52 deletions

View File

@ -5,17 +5,10 @@ import { download } from '@/store/snark'
import networkConfig from '@/networkConfig' import networkConfig from '@/networkConfig'
import InstanceABI from '@/abis/Instance.abi.json' import InstanceABI from '@/abis/Instance.abi.json'
import { CONTRACT_INSTANCES, eventsType } from '@/constants' import { CONTRACT_INSTANCES, eventsType } from '@/constants'
import { sleep, formatEvents, capitalizeFirstLetter } from '@/utils' import { sleep, flattenNArray, formatEvents, capitalizeFirstLetter } from '@/utils'
const supportedCaches = ['1', '56', '100', '137'] const supportedCaches = ['1', '56', '100', '137']
let store
if (process.browser) {
window.onNuxtReady(({ $store }) => {
store = $store
})
}
class EventService { class EventService {
constructor({ netId, amount, currency, factoryMethods }) { constructor({ netId, amount, currency, factoryMethods }) {
this.idb = window.$nuxt.$indexedDB(netId) this.idb = window.$nuxt.$indexedDB(netId)
@ -237,7 +230,7 @@ class EventService {
} }
} }
async getEventsPartFromRpc({ fromBlock, toBlock, type }) { async getEventsPartFromRpc({ fromBlock, toBlock, type }, shouldRetry = false, i = 0) {
try { try {
const { currentBlockNumber } = await this.getBlocksDiff({ fromBlock }) const { currentBlockNumber } = await this.getBlocksDiff({ fromBlock })
@ -248,11 +241,43 @@ class EventService {
} }
} }
const events = await this.contract.getPastEvents(capitalizeFirstLetter(type), { const rpcRequest = new Promise((resolve, reject) => {
fromBlock, const repsonse = this.contract.getPastEvents(capitalizeFirstLetter(type), {
toBlock fromBlock,
toBlock
})
if (repsonse) {
resolve(repsonse)
} else {
reject(new Error())
}
}) })
let events = []
if (shouldRetry) {
i = i + 1
try {
events = await Promise.resolve(rpcRequest)
} catch (e) {
await sleep(200)
events = await this.getEventsPartFromRpc(
{
fromBlock,
toBlock,
type
},
i !== 5,
i
)
}
} else {
events = await Promise.resolve(rpcRequest)
}
if (!events?.length) { if (!events?.length) {
return { return {
events: [], events: [],
@ -268,55 +293,56 @@ class EventService {
} }
} }
createBatchRequest({ batchIndex, batchSize, batchBlocks, blockDenom, type }) {
return new Array(batchSize).fill('').map(
(e, i) =>
new Promise(async (resolve) => {
const toBlock = batchBlocks[batchIndex * batchSize + i]
const fromBlock = toBlock - blockDenom
const batchEvents = await this.getEventsPartFromRpc(
{
fromBlock,
toBlock,
type
},
true
)
resolve(batchEvents.events)
})
)
}
async getBatchEventsFromRpc({ fromBlock, type }) { async getBatchEventsFromRpc({ fromBlock, type }) {
try { try {
const batchSize = 10
const blockRange = 10000 const blockRange = 10000
const { blockDifference, currentBlockNumber } = await this.getBlocksDiff({ fromBlock }) const { blockDifference, currentBlockNumber } = await this.getBlocksDiff({ fromBlock })
let numberParts = blockDifference === 0 ? 1 : Math.ceil(blockDifference / blockRange) const batchDigest = blockDifference === 0 ? 1 : Math.ceil(blockDifference / blockRange)
const part = Math.ceil(blockDifference / numberParts) const blockDenom = Math.ceil(blockDifference / batchDigest)
const batchCount = Math.ceil(batchDigest / batchSize)
const blocks = new Array(batchCount * batchSize).fill('')
const batchBlocks = blocks.map((e, i) => (i + 1) * blockDenom + fromBlock)
let events = [] let events = []
let loadedBlocks = 0
let toBlock = fromBlock + part
if (fromBlock < currentBlockNumber) { if (fromBlock < currentBlockNumber) {
if (toBlock >= currentBlockNumber) { for (let batchIndex = 0; batchIndex < batchCount; batchIndex++) {
toBlock = 'latest' const batch = await Promise.all(
numberParts = 1 this.createBatchRequest({ batchIndex, batchBlocks, blockDenom, batchSize, type })
} )
if (store.state.loading.progress !== 98) {
store.dispatch('loading/updateProgress', { message: 'Fetching the past events', progress: 0 }) events = events.concat(batch)
} }
for (let i = 0; i < numberParts; i++) { events = flattenNArray(events)
try {
await sleep(200)
const partOfEvents = await this.getEventsPartFromRpc({ fromBlock, toBlock, type })
if (partOfEvents) {
events = events.concat(partOfEvents.events)
}
loadedBlocks += toBlock - fromBlock
fromBlock = toBlock
toBlock += part
const progressInt = parseInt((loadedBlocks / blockDifference) * 100) return {
console.log('Progress: ', progressInt) lastBlock: events[events.length - 1].blockNumber,
if (store.state.loading.progress !== 98) { events
store.dispatch('loading/updateProgress', {
message: 'Fetching the past events',
progress: progressInt === 100 ? 98 : progressInt
})
}
} catch {
numberParts = numberParts + 1
}
}
if (events.length) {
return {
events,
lastBlock: toBlock === 'latest' ? currentBlockNumber : toBlock
}
} }
} }
return undefined return undefined
@ -328,6 +354,7 @@ class EventService {
async getEventsFromRpc({ fromBlock, type }) { async getEventsFromRpc({ fromBlock, type }) {
try { try {
const { blockDifference } = await this.getBlocksDiff({ fromBlock }) const { blockDifference } = await this.getBlocksDiff({ fromBlock })
let events let events
if (blockDifference < 10000) { if (blockDifference < 10000) {

View File

@ -22,12 +22,12 @@ class MerkleTreeService {
commitment, commitment,
instanceName, instanceName,
fileFolder: 'trees', fileFolder: 'trees',
fileName: `deposits_${currency}_${amount}_bloom.json.zip` fileName: `deposits_${currency}_${amount}_bloom.json.gz`
}) })
} }
getFileName(partNumber = trees.PARTS_COUNT) { getFileName(partNumber = trees.PARTS_COUNT) {
return `trees/deposits_${this.currency}_${this.amount}_slice${partNumber}.json.zip` return `trees/deposits_${this.currency}_${this.amount}_slice${partNumber}.json.gz`
} }
createTree({ events }) { createTree({ events }) {

View File

@ -358,7 +358,7 @@ const actions = {
try { try {
const module = await download({ const module = await download({
contentType: 'string', contentType: 'string',
name: `events/encrypted_notes_${netId}.json.zip` name: `events/encrypted_notes_${netId}.json.gz`
}) })
if (module) { if (module) {

View File

@ -8,6 +8,12 @@ export * from './stringUtils'
export * from './numberUtils' export * from './numberUtils'
export * from './instanceUtils' export * from './instanceUtils'
export function flattenNArray(arr) {
return arr.reduce((flat, toFlatten) => {
return flat.concat(Array.isArray(toFlatten) ? flattenNArray(toFlatten) : toFlatten)
}, [])
}
export function sleep(ms) { export function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms)) return new Promise((resolve) => setTimeout(resolve, ms))
} }