client: make fetcher more typesafe

Author: jochem-brouwer

packages/client/lib/net/protocol/ethprotocol.ts

@@ -2,6 +2,7 @@ import { BN, bufferToInt } from 'ethereumjs-util'
 import { Block, BlockHeader, BlockHeaderBuffer } from '@ethereumjs/block'
 import { Chain } from './../../blockchain'
 import { Message, Protocol, ProtocolOptions } from './protocol'
+import { BlockBodyBuffer } from '../../../../block/dist'
 
 interface EthProtocolOptions extends ProtocolOptions {
   /* Blockchain */
@@ -24,7 +25,7 @@ type GetBlockHeadersOpts = {
  */
 export interface EthProtocolMethods {
   getBlockHeaders: (opts: GetBlockHeadersOpts) => Promise<BlockHeader[]>
-  getBlockBodies: (hashes: Buffer[]) => Promise<Block[]>
+  getBlockBodies: (hashes: Buffer[]) => Promise<BlockBodyBuffer[]>
 }
 
 const messages: Message[] = [

packages/client/lib/sync/fetcher/blockfetcher.ts

@@ -4,48 +4,31 @@ import { BN } from 'ethereumjs-util'
 import { Peer } from '../../net/peer'
 import { EthProtocolMethods } from '../../net/protocol'
 import { Chain } from '../../blockchain'
+import { Job } from '../../types'
+import { BlockFetcherBase, JobTask, BlockFetcherOptions } from './blockfetcherbase'
 
-export interface BlockFetcherOptions extends FetcherOptions {
-  /* Blockchain */
-  chain: Chain
-
-  /* Block number to start fetching from */
-  first: BN
-
-  /* How many blocks to fetch */
-  count: BN
-}
 
 /**
  * Implements an eth/62 based block fetcher
  * @memberof module:sync/fetcher
  */
-export class BlockFetcher extends Fetcher {
-  protected chain: Chain
-  protected first: BN
-  protected count: BN
-
+export class BlockFetcher extends BlockFetcherBase<Block>{
   /**
    * Create new block fetcher
    * @param {BlockFetcherOptions}
    */
   constructor(options: BlockFetcherOptions) {
     super(options)
-
-    this.chain = options.chain
-    this.maxPerRequest = options.maxPerRequest ?? 128
-    this.first = options.first
-    this.count = options.count
   }
 
   /**
    * Generate list of tasks to fetch
    * @return {Object[]} tasks
    */
-  tasks(): object[] {
+  tasks(): JobTask[] {
     const { first, count } = this
     const max = this.maxPerRequest
-    const tasks = []
+    const tasks: JobTask[] = []
     while (count.gten(max)) {
       tasks.push({ first: first.clone(), count: max })
       first.iaddn(max)
@@ -61,18 +44,18 @@ export class BlockFetcher extends Fetcher {
    * Requests blocks associated with this job
    * @param job
    */
-  async request(job: any): Promise<any> {
+  async request(job: Job<JobTask, Block>): Promise<Block[]> {
     const { task, peer } = job
     const { first, count } = task
-    const headers = await (peer.eth as EthProtocolMethods).getBlockHeaders({
+    const headers = await (peer!.eth as EthProtocolMethods).getBlockHeaders({
       block: first,
       max: count,
     })
-    const bodies = await peer.eth.getBlockBodies(headers.map((h) => h.hash()))
-    const blocks = bodies.map(([txsData, unclesData]: BlockBodyBuffer, i: number) =>
+    const bodies: BlockBodyBuffer[] = <BlockBodyBuffer[]>await peer!.eth!.getBlockBodies(headers.map((h) => h.hash()))
+    const blocks: Block[] = bodies.map(([txsData, unclesData]: BlockBodyBuffer, i: number) =>
       Block.fromValuesArray([headers[i].raw(), txsData, unclesData], { common: this.config.common })
     )
-    return { blocks }
+    return blocks
   }
 
   /**
@@ -81,18 +64,19 @@ export class BlockFetcher extends Fetcher {
    * @param  result fetch result
    * @return {*} results of processing job or undefined if job not finished
    */
-  process(job: any, result: any) {
-    if (result.blocks && result.blocks.length === job.task.count) {
-      return result.blocks
+  process(job: Job<JobTask, Block>, result: Block[]): Block[] | null {
+    if (result && result.length === job.task.count) {
+      return result
     }
+    return null
   }
 
   /**
    * Store fetch result. Resolves once store operation is complete.
    * @param {Block[]} blocks fetch result
    * @return {Promise}
    */
-  async store(blocks: Array<any>) {
+  async store(blocks: Block[]) {
     await this.chain.putBlocks(blocks)
   }
 
@@ -102,7 +86,7 @@ export class BlockFetcher extends Fetcher {
    * @return {Peer}
    */
   // TODO: find out what _job is supposed to be doing here...
-  peer(_job: any): Peer {
+  peer(): Peer {
     return this.pool.idle((p: any) => p.eth)
   }
 }

packages/client/lib/sync/fetcher/blockfetcherbase.ts

@@ -0,0 +1,59 @@
+import { Fetcher, FetcherOptions } from './fetcher'
+import { BN } from 'ethereumjs-util'
+import { Chain } from '../../blockchain'
+
+
+export interface BlockFetcherOptions extends FetcherOptions {
+  /* Blockchain */
+  chain: Chain
+
+  /* Block number to start fetching from */
+  first: BN
+
+  /* How many blocks to fetch */
+  count: BN
+}
+
+export type JobTask = {
+    first: BN 
+    count: number
+}
+  
+
+export abstract class BlockFetcherBase<JobResult> extends Fetcher<JobTask, JobResult>{
+    protected chain: Chain
+    protected first: BN
+    protected count: BN
+  
+    /**
+     * Create new block fetcher
+     * @param {BlockFetcherOptions}
+     */
+    constructor(options: BlockFetcherOptions) {
+      super(options)
+  
+      this.chain = options.chain
+      this.maxPerRequest = options.maxPerRequest ?? 128
+      this.first = options.first
+      this.count = options.count
+    }
+  
+    /**
+     * Generate list of tasks to fetch
+     * @return {Object[]} tasks
+     */
+    tasks(): JobTask[] {
+      const { first, count } = this
+      const max = this.maxPerRequest
+      const tasks: JobTask[] = []
+      while (count.gten(max)) {
+        tasks.push({ first: first.clone(), count: max })
+        first.iaddn(max)
+        count.isubn(max)
+      }
+      if (count.gtn(0)) {
+        tasks.push({ first: first.clone(), count: count.toNumber() })
+      }
+      return tasks
+    }
+}