diff options
Diffstat (limited to 'packages/pipeline/src/data_sources/contract-wrappers')
3 files changed, 137 insertions, 51 deletions
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts new file mode 100644 index 000000000..e0098122f --- /dev/null +++ b/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts @@ -0,0 +1,45 @@ +import { + ContractWrappers, + ERC20TokenApprovalEventArgs, + ERC20TokenEvents, + ERC20TokenWrapper, +} from '@0x/contract-wrappers'; +import { Web3ProviderEngine } from '@0x/subproviders'; +import { LogWithDecodedArgs } from 'ethereum-types'; + +import { GetEventsFunc, getEventsWithPaginationAsync } from './utils'; + +export class ERC20EventsSource { + private readonly _erc20Wrapper: ERC20TokenWrapper; + private readonly _tokenAddress: string; + constructor(provider: Web3ProviderEngine, networkId: number, tokenAddress: string) { + const contractWrappers = new ContractWrappers(provider, { networkId }); + this._erc20Wrapper = contractWrappers.erc20Token; + this._tokenAddress = tokenAddress; + } + + public async getApprovalEventsAsync( + startBlock: number, + endBlock: number, + ): Promise<Array<LogWithDecodedArgs<ERC20TokenApprovalEventArgs>>> { + return getEventsWithPaginationAsync( + this._getApprovalEventsForRangeAsync.bind(this) as GetEventsFunc<ERC20TokenApprovalEventArgs>, + startBlock, + endBlock, + ); + } + + // Gets all approval events of for a specific sub-range. This getter + // function will be called during each step of pagination. + private async _getApprovalEventsForRangeAsync( + fromBlock: number, + toBlock: number, + ): Promise<Array<LogWithDecodedArgs<ERC20TokenApprovalEventArgs>>> { + return this._erc20Wrapper.getLogsAsync<ERC20TokenApprovalEventArgs>( + this._tokenAddress, + ERC20TokenEvents.Approval, + { fromBlock, toBlock }, + {}, + ); + } +} diff --git a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts index 1717eb8b3..58691e2ab 100644 --- a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts +++ b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts @@ -8,78 +8,52 @@ import { ExchangeWrapper, } from '@0x/contract-wrappers'; import { Web3ProviderEngine } from '@0x/subproviders'; -import { Web3Wrapper } from '@0x/web3-wrapper'; import { LogWithDecodedArgs } from 'ethereum-types'; -import { EXCHANGE_START_BLOCK } from '../../utils'; - -const BLOCK_FINALITY_THRESHOLD = 10; // When to consider blocks as final. Used to compute default toBlock. -const NUM_BLOCKS_PER_QUERY = 20000; // Number of blocks to query for events at a time. +import { GetEventsFunc, getEventsWithPaginationAsync } from './utils'; export class ExchangeEventsSource { private readonly _exchangeWrapper: ExchangeWrapper; - private readonly _web3Wrapper: Web3Wrapper; constructor(provider: Web3ProviderEngine, networkId: number) { - this._web3Wrapper = new Web3Wrapper(provider); const contractWrappers = new ContractWrappers(provider, { networkId }); this._exchangeWrapper = contractWrappers.exchange; } public async getFillEventsAsync( - fromBlock?: number, - toBlock?: number, + startBlock: number, + endBlock: number, ): Promise<Array<LogWithDecodedArgs<ExchangeFillEventArgs>>> { - return this._getEventsAsync<ExchangeFillEventArgs>(ExchangeEvents.Fill, fromBlock, toBlock); + const getFillEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeFillEventArgs>(ExchangeEvents.Fill); + return getEventsWithPaginationAsync(getFillEventsForRangeAsync, startBlock, endBlock); } public async getCancelEventsAsync( - fromBlock?: number, - toBlock?: number, + startBlock: number, + endBlock: number, ): Promise<Array<LogWithDecodedArgs<ExchangeCancelEventArgs>>> { - return this._getEventsAsync<ExchangeCancelEventArgs>(ExchangeEvents.Cancel, fromBlock, toBlock); + const getCancelEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeCancelEventArgs>( + ExchangeEvents.Cancel, + ); + return getEventsWithPaginationAsync(getCancelEventsForRangeAsync, startBlock, endBlock); } public async getCancelUpToEventsAsync( - fromBlock?: number, - toBlock?: number, + startBlock: number, + endBlock: number, ): Promise<Array<LogWithDecodedArgs<ExchangeCancelUpToEventArgs>>> { - return this._getEventsAsync<ExchangeCancelUpToEventArgs>(ExchangeEvents.CancelUpTo, fromBlock, toBlock); - } - - private async _getEventsAsync<ArgsType extends ExchangeEventArgs>( - eventName: ExchangeEvents, - fromBlock: number = EXCHANGE_START_BLOCK, - toBlock?: number, - ): Promise<Array<LogWithDecodedArgs<ArgsType>>> { - const calculatedToBlock = - toBlock === undefined - ? (await this._web3Wrapper.getBlockNumberAsync()) - BLOCK_FINALITY_THRESHOLD - : toBlock; - let events: Array<LogWithDecodedArgs<ArgsType>> = []; - for (let currFromBlock = fromBlock; currFromBlock <= calculatedToBlock; currFromBlock += NUM_BLOCKS_PER_QUERY) { - events = events.concat( - await this._getEventsForRangeAsync<ArgsType>( - eventName, - currFromBlock, - Math.min(currFromBlock + NUM_BLOCKS_PER_QUERY - 1, calculatedToBlock), - ), - ); - } - return events; + const getCancelUpToEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeCancelUpToEventArgs>( + ExchangeEvents.CancelUpTo, + ); + return getEventsWithPaginationAsync(getCancelUpToEventsForRangeAsync, startBlock, endBlock); } - private async _getEventsForRangeAsync<ArgsType extends ExchangeEventArgs>( - eventName: ExchangeEvents, - fromBlock: number, - toBlock: number, - ): Promise<Array<LogWithDecodedArgs<ArgsType>>> { - return this._exchangeWrapper.getLogsAsync<ArgsType>( - eventName, - { - fromBlock, - toBlock, - }, - {}, - ); + // Returns a getter function which gets all events of a specific type for a + // specific sub-range. This getter function will be called during each step + // of pagination. + private _makeGetterFuncForEventType<ArgsType extends ExchangeEventArgs>( + eventType: ExchangeEvents, + ): GetEventsFunc<ArgsType> { + return async (fromBlock: number, toBlock: number) => + this._exchangeWrapper.getLogsAsync<ArgsType>(eventType, { fromBlock, toBlock }, {}); } } diff --git a/packages/pipeline/src/data_sources/contract-wrappers/utils.ts b/packages/pipeline/src/data_sources/contract-wrappers/utils.ts new file mode 100644 index 000000000..67660a37e --- /dev/null +++ b/packages/pipeline/src/data_sources/contract-wrappers/utils.ts @@ -0,0 +1,67 @@ +import { DecodedLogArgs, LogWithDecodedArgs } from 'ethereum-types'; + +const NUM_BLOCKS_PER_QUERY = 10000; // Number of blocks to query for events at a time. +const NUM_RETRIES = 3; // Number of retries if a request fails or times out. + +export type GetEventsFunc<ArgsType extends DecodedLogArgs> = ( + fromBlock: number, + toBlock: number, +) => Promise<Array<LogWithDecodedArgs<ArgsType>>>; + +/** + * Gets all events between the given startBlock and endBlock by querying for + * NUM_BLOCKS_PER_QUERY at a time. Accepts a getter function in order to + * maximize code re-use and allow for getting different types of events for + * different contracts. If the getter function throws with a retryable error, + * it will automatically be retried up to NUM_RETRIES times. + * @param getEventsAsync A getter function which will be called for each step during pagination. + * @param startBlock The start of the entire block range to get events for. + * @param endBlock The end of the entire block range to get events for. + */ +export async function getEventsWithPaginationAsync<ArgsType extends DecodedLogArgs>( + getEventsAsync: GetEventsFunc<ArgsType>, + startBlock: number, + endBlock: number, +): Promise<Array<LogWithDecodedArgs<ArgsType>>> { + let events: Array<LogWithDecodedArgs<ArgsType>> = []; + for (let fromBlock = startBlock; fromBlock <= endBlock; fromBlock += NUM_BLOCKS_PER_QUERY) { + const toBlock = Math.min(fromBlock + NUM_BLOCKS_PER_QUERY - 1, endBlock); + const eventsInRange = await _getEventsWithRetriesAsync(getEventsAsync, NUM_RETRIES, fromBlock, toBlock); + events = events.concat(eventsInRange); + } + return events; +} + +/** + * Calls the getEventsAsync function and retries up to numRetries times if it + * throws with an error that is considered retryable. + * @param getEventsAsync a function that will be called on each iteration. + * @param numRetries the maximum number times to retry getEventsAsync if it fails with a retryable error. + * @param fromBlock the start of the sub-range of blocks we are getting events for. + * @param toBlock the end of the sub-range of blocks we are getting events for. + */ +export async function _getEventsWithRetriesAsync<ArgsType extends DecodedLogArgs>( + getEventsAsync: GetEventsFunc<ArgsType>, + numRetries: number, + fromBlock: number, + toBlock: number, +): Promise<Array<LogWithDecodedArgs<ArgsType>>> { + let eventsInRange: Array<LogWithDecodedArgs<ArgsType>> = []; + for (let i = 0; i <= numRetries; i++) { + try { + eventsInRange = await getEventsAsync(fromBlock, toBlock); + } catch (err) { + if (isErrorRetryable(err) && i < numRetries) { + continue; + } else { + throw err; + } + } + break; + } + return eventsInRange; +} + +function isErrorRetryable(err: Error): boolean { + return err.message.includes('network timeout'); +} |