async function main() {
  const context = await new Context().init(config, errorCounter)

  let processedToBlock = await getLastBlock(context.config)
  logger.info(`Resuming processing from ${processedToBlock}`)

  if (context.config.concurrency > 1) {
    logger.warn(`Backfill mode: concurrency=${context.config.concurrency}`)
  }

  // Helper function to wait at most tickIntervalSeconds.
  const tickIntervalSeconds = 5
  let start

  async function nextTick() {
    const elapsed = new Date() - start
    const delay = Math.max(tickIntervalSeconds * 1000 - elapsed, 1)
    return new Promise(resolve => setTimeout(() => resolve(true), delay))
  }

  do {
    start = new Date()

    // Compute the range of blocks to process,
    // while respecting trailing block configuration.
    const currentBlock = await context.web3.eth.getBlockNumber()
    const toBlock = Math.max(currentBlock - context.config.trailBlocks, 0)

    // Listener is up to date. Nothing to do.
    if (toBlock <= processedToBlock) {
      logger.debug('No new blocks to process')
      continue
    }
    logger.info(`Querying events from ${processedToBlock} up to ${toBlock}`)

    // Update the event caches to set their max block number.
    contractsContext.marketplace.eventCache.updateBlock(toBlock)
    contractsContext.identityEvents.eventCache.updateBlock(toBlock)

    // Retrieve all events for the relevant contracts
    const eventArrays = await Promise.all([
      contractsContext.marketplace.eventCache.allEvents(),
      contractsContext.identityEvents.eventCache.allEvents()
    ])

    // Flatten array of arrays filtering out anything undefined
    const events = [].concat(...eventArrays.filter(x => x))
    // Filter to only new events
    let newEvents = events.filter(event => event.blockNumber > processedToBlock)
    logger.debug(`Got ${newEvents.length} new events`)

    if (context.config.concurrency > 1) {
      // Concurrency greater than 1 -> Backfill mode.
      const limit = pLimit(context.config.concurrency)
      const promises = []
      newEvents.forEach(newEvent => {
        promises.push(limit(() => handleEvent(newEvent, context)))
      })
      await Promise.all(promises)
    } else {
      // Concurrency set to 1 -> Normal operation mode.
      // Sort events by blockNumber and logIndex to process them in order.
      // In normal operation (not backfill), ordering does matter for handlers.
      // For example assume a buyer updates their identity to add their email
      // then makes a purchase. If the handler would process the purchase
      // event before the identity update, we may not be able to send the buyer
      // a confirmation email about their offer.
      newEvents = sortBy(newEvents, ['blockNumber', 'logIndex'], ['asc', 'asc'])

      // Have the handler process each event.
      for (const event of newEvents) {
        // Note: we purposely do not set the exitOnError option of withRetrys to false.
        // In case all retries fails, it indicates something is wrong at the system
        // level and a process restart may fix it.
        await withRetrys(async () => {
          handleEvent(event, context)
        })
      }
    }

    // Record state of processing
    logger.debug(`Updating last processed block to ${toBlock}`)
    await setLastBlock(context.config, toBlock)
    processedToBlock = toBlock
    if (context.config.enableMetrics) {
      blockGauge.set(toBlock)
    }
  } while (await nextTick())
}
/**
 * Creates runtime context and starts the tracking loop
 * @return {Promise<void>}
 */
Creates runtime context and starts the tracking loop
@return {Promise}
Creates runtime context and starts the tracking loop @return { Promise }
async function main() {
  const context = await new Context().init(config, errorCounter)
  let processedToBlock = await getLastBlock(context.config)
  logger.info(`Resuming processing from ${processedToBlock}`)
  if (context.config.concurrency > 1) {
    logger.warn(`Backfill mode: concurrency=${context.config.concurrency}`)
  }
  const tickIntervalSeconds = 5
  let start
  async function nextTick() {
    const elapsed = new Date() - start
    const delay = Math.max(tickIntervalSeconds * 1000 - elapsed, 1)
    return new Promise(resolve => setTimeout(() => resolve(true), delay))
  }
  do {
    start = new Date()
    const currentBlock = await context.web3.eth.getBlockNumber()
    const toBlock = Math.max(currentBlock - context.config.trailBlocks, 0)
    if (toBlock <= processedToBlock) {
      logger.debug('No new blocks to process')
      continue
    }
    logger.info(`Querying events from ${processedToBlock} up to ${toBlock}`)
    contractsContext.marketplace.eventCache.updateBlock(toBlock)
    contractsContext.identityEvents.eventCache.updateBlock(toBlock)
    const eventArrays = await Promise.all([
      contractsContext.marketplace.eventCache.allEvents(),
      contractsContext.identityEvents.eventCache.allEvents()
    ])
    const events = [].concat(...eventArrays.filter(x => x))
    let newEvents = events.filter(event => event.blockNumber > processedToBlock)
    logger.debug(`Got ${newEvents.length} new events`)
    if (context.config.concurrency > 1) {
      const limit = pLimit(context.config.concurrency)
      const promises = []
      newEvents.forEach(newEvent => {
        promises.push(limit(() => handleEvent(newEvent, context)))
      })
      await Promise.all(promises)
    } else {
      newEvents = sortBy(newEvents, ['blockNumber', 'logIndex'], ['asc', 'asc'])
      for (const event of newEvents) {
        await withRetrys(async () => {
          handleEvent(event, context)
        })
      }
    }
    logger.debug(`Updating last processed block to ${toBlock}`)
    await setLastBlock(context.config, toBlock)
    processedToBlock = toBlock
    if (context.config.enableMetrics) {
      blockGauge.set(toBlock)
    }
  } while (await nextTick())
}
async function main ( ) { const context = await new Context ( ) . init ( config , errorCounter ) let processedToBlock = await getLastBlock ( context . config ) logger . info ( ` ${ processedToBlock } ` ) if ( context . config . concurrency > 1 ) { logger . warn ( ` ${ context . config . concurrency } ` ) } const tickIntervalSeconds = 5 let start async function nextTick ( ) { const elapsed = new Date ( ) - start const delay = Math . max ( tickIntervalSeconds * 1000 - elapsed , 1 ) return new Promise ( resolve => setTimeout ( ( ) => resolve ( true ) , delay ) ) } do { start = new Date ( ) const currentBlock = await context . web3 . eth . getBlockNumber ( ) const toBlock = Math . max ( currentBlock - context . config . trailBlocks , 0 ) if ( toBlock <= processedToBlock ) { logger . debug ( 'No new blocks to process' ) continue } logger . info ( ` ${ processedToBlock } ${ toBlock } ` ) contractsContext . marketplace . eventCache . updateBlock ( toBlock ) contractsContext . identityEvents . eventCache . updateBlock ( toBlock ) const eventArrays = await Promise . all ( [ contractsContext . marketplace . eventCache . allEvents ( ) , contractsContext . identityEvents . eventCache . allEvents ( ) ] ) const events = [ ] . concat ( ... eventArrays . filter ( x => x ) ) let newEvents = events . filter ( event => event . blockNumber > processedToBlock ) logger . debug ( ` ${ newEvents . length } ` ) if ( context . config . concurrency > 1 ) { const limit = pLimit ( context . config . concurrency ) const promises = [ ] newEvents . forEach ( newEvent => { promises . push ( limit ( ( ) => handleEvent ( newEvent , context ) ) ) } ) await Promise . all ( promises ) } else { newEvents = sortBy ( newEvents , [ 'blockNumber' , 'logIndex' ] , [ 'asc' , 'asc' ] ) for ( const event of newEvents ) { await withRetrys ( async ( ) => { handleEvent ( event , context ) } ) } } logger . debug ( ` ${ toBlock } ` ) await setLastBlock ( context . config , toBlock ) processedToBlock = toBlock if ( context . config . enableMetrics ) { blockGauge . set ( toBlock ) } } while ( await nextTick ( ) ) }
Creates runtime context and starts the tracking loop
 @return {Promise<void>}
Creates runtime context and starts the tracking loop @return { Promise<void > }
// Helper function to wait at most tickIntervalSeconds. // Compute the range of blocks to process, // while respecting trailing block configuration. // Listener is up to date. Nothing to do. // Update the event caches to set their max block number. // Retrieve all events for the relevant contracts // Flatten array of arrays filtering out anything undefined // Filter to only new events // Concurrency greater than 1 -> Backfill mode. // Concurrency set to 1 -> Normal operation mode. // Sort events by blockNumber and logIndex to process them in order. // In normal operation (not backfill), ordering does matter for handlers. // For example assume a buyer updates their identity to add their email // then makes a purchase. If the handler would process the purchase // event before the identity update, we may not be able to send the buyer // a confirmation email about their offer. // Have the handler process each event. // Note: we purposely do not set the exitOnError option of withRetrys to false. // In case all retries fails, it indicates something is wrong at the system // level and a process restart may fix it. // Record state of processing