wip: block number indexer

Co-authored-by: Michał Podsiadły <[email protected]>

packages/backend/src/Application.ts

@@ -3,6 +3,7 @@ import { Logger } from '@l2beat/backend-tools'
 import { ApiServer } from './api/ApiServer'
 import { Config } from './config'
 import { ApplicationModule } from './modules/ApplicationModule'
+import { createEthereumDiscoveryModule } from './modules/EthereumDiscoveryModule'
 import { createHealthModule } from './modules/HealthModule'
 import { Database } from './peripherals/database/shared/Database'
 
@@ -19,6 +20,7 @@ export class Application {
 
     const modules: (ApplicationModule | undefined)[] = [
       createHealthModule(config),
+      createEthereumDiscoveryModule(database, logger),
     ]
 
     const apiServer = new ApiServer(

packages/backend/src/indexers/BlockNumberIndexer.test.ts

@@ -0,0 +1,195 @@
+import { assert, Logger } from '@l2beat/backend-tools'
+import { Hash256, UnixTime } from '@lz/libs'
+import { expect, mockFn, mockObject } from 'earl'
+
+import {
+  BlockchainClient,
+  BlockFromClient,
+} from '../peripherals/clients/BlockchainClient'
+import {
+  BlockNumberRecord,
+  BlockNumberRepository,
+} from '../peripherals/database/BlockNumberRepository'
+import { IndexerStateRepository } from '../peripherals/database/IndexerStateRepository'
+import { BlockNumberIndexer } from './BlockNumberIndexer'
+import { ClockIndexer } from './ClockIndexer'
+
+describe(BlockNumberIndexer.name, () => {
+  describe(BlockNumberIndexer.prototype.update.name, () => {
+    it('downloads a new block and returns its timestamp without reorg', async () => {
+      const blockNumberIndexer = new BlockNumberIndexer(
+        mockObject<BlockchainClient>({
+          getBlockNumberAtOrBefore,
+          getBlock: async (number) => mockBlock(number),
+        }),
+        mockObject<BlockNumberRepository>({
+          findByNumber: async (number) => mockBlockRecord(number),
+          addMany: async () => [0],
+          add: async () => 0,
+        }),
+        mockObject<IndexerStateRepository>(),
+        0,
+        mockObject<ClockIndexer>({
+          subscribe: () => {},
+        }),
+        Logger.SILENT,
+      )
+
+      // from stays at 0 as we do not check against `from` value
+      expect(await blockNumberIndexer.update(0, 1)).toEqual(0)
+      expect(await blockNumberIndexer.update(0, 2000)).toEqual(1000)
+      expect(await blockNumberIndexer.update(1000, 2000)).toEqual(2000)
+    })
+
+    it('downloads a new block and returns its timestamp with reorg', async () => {
+      const memoryBlockNumberRepository = mockBlockNumberRepository([
+        blockToRecord(BLOCKS[1]!),
+      ])
+
+      const reorgedBlock = {
+        number: 2,
+        hash: Hash256.random().toString(),
+        parentHash: HASH1,
+        timestamp: 1500,
+      }
+      const reorgedBlockRecord: BlockNumberRecord = {
+        blockNumber: reorgedBlock.number,
+        blockHash: Hash256(reorgedBlock.hash),
+        timestamp: new UnixTime(reorgedBlock.timestamp),
+      }
+      const blockNumberIndexer = new BlockNumberIndexer(
+        mockObject<BlockchainClient>({
+          getBlockNumberAtOrBefore,
+          getBlock: mockFn(async (number: number) =>
+            mockBlock(number),
+          ).resolvesToOnce(reorgedBlock),
+        }),
+        memoryBlockNumberRepository,
+        mockObject<IndexerStateRepository>({
+          findById: async () => ({ id: 'BlockNumberIndexer', height: 1 }),
+        }),
+        1,
+        mockObject<ClockIndexer>({
+          subscribe: () => {},
+        }),
+        Logger.DEBUG,
+      )
+
+      await blockNumberIndexer.start()
+
+      // saves block to database
+      expect(await blockNumberIndexer.update(0, 2000)).toEqual(1500)
+      expect(memoryBlockNumberRepository.add).toHaveBeenCalledWith(
+        reorgedBlockRecord,
+      )
+
+      expect(await blockNumberIndexer.update(0, 3000)).toEqual(1000)
+      expect(await blockNumberIndexer.update(1000, 3000)).toEqual(3000)
+      expect(memoryBlockNumberRepository.addMany).toHaveBeenCalledTimes(1)
+      expect(memoryBlockNumberRepository.addMany).toHaveBeenNthCalledWith(
+        1,
+        [BLOCKS[2], BLOCKS[3]].map(blockToRecord),
+      )
+    })
+  })
+})
+
+const HASH0 = Hash256.random()
+const HASH1 = Hash256.random()
+const HASH2 = Hash256.random()
+const HASH3 = Hash256.random()
+const HASH4 = Hash256.random()
+
+const BLOCKS = [
+  {
+    number: 0,
+    hash: HASH0.toString(),
+    parentHash: '',
+    timestamp: 0,
+  },
+  {
+    number: 1,
+    hash: HASH1.toString(),
+    parentHash: HASH0.toString(),
+    timestamp: 1000,
+  },
+  {
+    number: 2,
+    hash: HASH2.toString(),
+    parentHash: HASH1.toString(),
+    timestamp: 2000,
+  },
+  {
+    number: 3,
+    hash: HASH3.toString(),
+    parentHash: HASH2.toString(),
+    timestamp: 3000,
+  },
+  {
+    number: 4,
+    hash: HASH4.toString(),
+    parentHash: HASH3.toString(),
+    timestamp: 4000,
+  },
+] as const
+
+function mockBlockRecord(blockNumber: number): BlockNumberRecord | undefined {
+  if (blockNumber === 0) {
+    return
+  }
+  const block = BLOCKS.find((b) => b.number === blockNumber)
+  assert(block, `Block not found for given number: ${blockNumber}`)
+  return {
+    blockNumber: block.number,
+    blockHash: Hash256(block.hash),
+    timestamp: new UnixTime(block.timestamp),
+  }
+}
+
+function mockBlock(blockId: number | Hash256): BlockFromClient {
+  console.log('mockBlock', blockId)
+  if (typeof blockId === 'number') {
+    const block = BLOCKS.find((b) => b.number === blockId)
+    assert(block, `Block not found for given number: ${blockId}`)
+    return block
+  }
+
+  const block = BLOCKS.find((b) => b.hash === blockId.toString())
+  assert(block, `Block not found for given hash: ${blockId}`)
+  return block
+}
+
+function getBlockNumberAtOrBefore(timestamp: UnixTime): Promise<number> {
+  const block = BLOCKS.filter((b) => b.timestamp <= timestamp.toNumber())
+    .sort((a, b) => b.timestamp - a.timestamp)
+    .shift()
+  assert(block, `Block not found for given timestamp: ${timestamp.toString()}`)
+  console.log('getBlockNumberAtOrBefore', timestamp.toString(), block.number)
+  return Promise.resolve(block.number)
+}
+
+function mockBlockNumberRepository(initialStorage: BlockNumberRecord[]) {
+  const blockNumberStorage: BlockNumberRecord[] = [...initialStorage]
+
+  return mockObject<BlockNumberRepository>({
+    findByNumber: async (number) =>
+      blockNumberStorage.find((bnr) => bnr.blockNumber === number),
+    findLast: async () => blockNumberStorage.at(-1),
+    addMany: async (blocks: BlockNumberRecord[]) => {
+      blockNumberStorage.push(...blocks)
+      return blocks.map((b) => b.blockNumber)
+    },
+    add: async (block: BlockNumberRecord) => {
+      blockNumberStorage.push(block)
+      return block.blockNumber
+    },
+  })
+}
+
+function blockToRecord(blockFromClient: BlockFromClient): BlockNumberRecord {
+  return {
+    blockNumber: blockFromClient.number,
+    blockHash: Hash256(blockFromClient.hash),
+    timestamp: new UnixTime(blockFromClient.timestamp),
+  }
+}

packages/backend/src/indexers/BlockNumberIndexer.ts

@@ -0,0 +1,126 @@
+import { Logger } from '@l2beat/backend-tools'
+import { ChildIndexer } from '@l2beat/uif'
+import { Hash256, UnixTime } from '@lz/libs'
+
+import { BlockchainClient } from '../peripherals/clients/BlockchainClient'
+import {
+  BlockNumberRecord,
+  BlockNumberRepository,
+} from '../peripherals/database/BlockNumberRepository'
+import { IndexerStateRepository } from '../peripherals/database/IndexerStateRepository'
+import { ClockIndexer } from './ClockIndexer'
+
+export class BlockNumberIndexer extends ChildIndexer {
+  private lastKnownNumber = 0
+  private reorgedBlocks = [] as BlockNumberRecord[]
+  private readonly id: string
+
+  constructor(
+    private readonly blockchainClient: BlockchainClient,
+    private readonly blockRepository: BlockNumberRepository,
+    private readonly indexerRepository: IndexerStateRepository,
+    private readonly startBlock: number,
+    clockIndexer: ClockIndexer,
+    logger: Logger,
+  ) {
+    super(logger, [clockIndexer])
+    this.id = 'BlockDownloader' // this should be unique across all indexers
+  }
+
+  override async start(): Promise<void> {
+    await super.start()
+    this.lastKnownNumber =
+      (await this.blockRepository.findLast())?.blockNumber ?? this.startBlock
+  }
+
+  async update(_fromTimestamp: number, toTimestamp: number): Promise<number> {
+    if (this.reorgedBlocks.length > 0) {
+      // we do not need to check if lastKnown < to because we are sure that
+      // those blocks are from the past
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      this.lastKnownNumber = this.reorgedBlocks.at(-1)!.blockNumber
+      await this.blockRepository.addMany(this.reorgedBlocks)
+      this.reorgedBlocks = []
+    }
+
+    const tip = await this.blockchainClient.getBlockNumberAtOrBefore(
+      new UnixTime(toTimestamp),
+    )
+    if (tip <= this.lastKnownNumber) {
+      return toTimestamp
+    }
+
+    return await this.advanceChain(this.lastKnownNumber + 1)
+  }
+
+  async invalidate(to: number): Promise<number> {
+    await this.blockRepository.deleteAfter(to)
+    return to
+  }
+
+  private async advanceChain(blockNumber: number): Promise<number> {
+    let [block, parent] = await Promise.all([
+      this.blockchainClient.getBlock(blockNumber),
+      this.getKnownBlock(blockNumber - 1),
+    ])
+
+    if (Hash256(block.parentHash) !== parent.blockHash) {
+      const changed = [block]
+
+      let current = blockNumber
+      while (Hash256(block.parentHash) !== parent.blockHash) {
+        current--
+        ;[block, parent] = await Promise.all([
+          this.blockchainClient.getBlock(Hash256(block.parentHash)),
+          this.getKnownBlock(current - 1),
+        ])
+        changed.push(block)
+      }
+
+      this.reorgedBlocks = changed.reverse().map((block) => {
+        return {
+          blockNumber: block.number,
+          blockHash: Hash256(block.hash),
+          timestamp: new UnixTime(block.timestamp),
+        }
+      })
+
+      return parent.timestamp.toNumber()
+    }
+
+    const record: BlockNumberRecord = {
+      blockNumber: block.number,
+      blockHash: Hash256(block.hash),
+      timestamp: new UnixTime(block.timestamp),
+    }
+    await this.blockRepository.add(record)
+    this.lastKnownNumber = block.number
+
+    return block.timestamp
+  }
+
+  async setSafeHeight(height: number): Promise<void> {
+    await this.indexerRepository.addOrUpdate({ id: this.id, height })
+  }
+
+  async getSafeHeight(): Promise<number> {
+    const record = await this.indexerRepository.findById(this.id)
+    return record?.height ?? 0
+  }
+
+  private async getKnownBlock(blockNumber: number): Promise<BlockNumberRecord> {
+    const known = await this.blockRepository.findByNumber(blockNumber)
+    if (known) {
+      return known
+    }
+    const downloaded = await this.blockchainClient.getBlock(blockNumber)
+
+    const record: BlockNumberRecord = {
+      blockNumber: downloaded.number,
+      blockHash: Hash256(downloaded.hash),
+      timestamp: new UnixTime(downloaded.timestamp),
+    }
+    await this.blockRepository.add(record)
+    return record
+  }
+}

packages/backend/src/indexers/ClockIndexer.ts

@@ -14,6 +14,10 @@ export class ClockIndexer extends RootIndexer {
   }
 
   tick(): Promise<number> {
-    return Promise.resolve(Date.now())
+    return Promise.resolve(getTimeSeconds())
   }
 }
+
+function getTimeSeconds(): number {
+  return Math.floor(Date.now() / 1000)
+}

packages/backend/src/modules/EthereumDiscoveryModule.ts

@@ -0,0 +1,41 @@
+import { Logger } from '@l2beat/backend-tools'
+import { JsonRpcProvider } from 'ethers'
+
+import { BlockNumberIndexer } from '../indexers/BlockNumberIndexer'
+import { ClockIndexer } from '../indexers/ClockIndexer'
+import { BlockchainClient } from '../peripherals/clients/BlockchainClient'
+import { BlockNumberRepository } from '../peripherals/database/BlockNumberRepository'
+import { IndexerStateRepository } from '../peripherals/database/IndexerStateRepository'
+import { Database } from '../peripherals/database/shared/Database'
+import { ApplicationModule } from './ApplicationModule'
+
+export function createEthereumDiscoveryModule(
+  database: Database,
+  logger: Logger,
+): ApplicationModule {
+  const blockRepository = new BlockNumberRepository(database, logger)
+  const indexerRepository = new IndexerStateRepository(database, logger)
+
+  const provider = new JsonRpcProvider(
+    'https://eth-mainnet.g.alchemy.com/v2/CLeXrqsc9lGb40KK9gRIbhQKGiakgp-S',
+  )
+  const blockchainClient = new BlockchainClient(provider, logger)
+
+  const clockIndexer = new ClockIndexer(logger, 10 * 1000)
+  const blockNumberIndexer = new BlockNumberIndexer(
+    blockchainClient,
+    blockRepository,
+    indexerRepository,
+    18127698,
+    clockIndexer,
+    logger,
+  )
+
+  return {
+    routers: [],
+    start: async () => {
+      await clockIndexer.start()
+      await blockNumberIndexer.start()
+    },
+  }
+}