client: fix fetcher type problems

Author: jochem-brouwer

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

@@ -1,5 +1,5 @@
 import { BN, bufferToInt } from 'ethereumjs-util'
-import { Block, BlockHeader, BlockHeaderBuffer } from '@ethereumjs/block'
+import { BlockHeader, BlockHeaderBuffer } from '@ethereumjs/block'
 import { Chain } from './../../blockchain'
 import { Message, Protocol, ProtocolOptions } from './protocol'
 import { BlockBodyBuffer } from '../../../../block/dist'

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

@@ -1,18 +1,14 @@
-import { Fetcher, FetcherOptions } from './fetcher'
 import { Block, BlockBodyBuffer } from '@ethereumjs/block'
-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'
 
-
 /**
  * Implements an eth/62 based block fetcher
  * @memberof module:sync/fetcher
  */
-export class BlockFetcher extends BlockFetcherBase<Block>{
+export class BlockFetcher extends BlockFetcherBase<Block[], Block> {
   /**
    * Create new block fetcher
    * @param {BlockFetcherOptions}
@@ -44,14 +40,16 @@ export class BlockFetcher extends BlockFetcherBase<Block>{
    * Requests blocks associated with this job
    * @param job
    */
-  async request(job: Job<JobTask, Block>): Promise<Block[]> {
+  async request(job: Job<JobTask, Block[]>): Promise<Block[]> {
     const { task, peer } = job
     const { first, count } = task
     const headers = await (peer!.eth as EthProtocolMethods).getBlockHeaders({
       block: first,
       max: count,
     })
-    const bodies: BlockBodyBuffer[] = <BlockBodyBuffer[]>await peer!.eth!.getBlockBodies(headers.map((h) => h.hash()))
+    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 })
     )
@@ -64,7 +62,8 @@ export class BlockFetcher extends BlockFetcherBase<Block>{
    * @param  result fetch result
    * @return {*} results of processing job or undefined if job not finished
    */
-  process(job: Job<JobTask, Block>, result: Block[]): Block[] | null {
+  process(job: Job<JobTask, Block[]>, result: Block[]): Block[] | null {
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (result && result.length === job.task.count) {
       return result
     }

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

@@ -2,7 +2,6 @@ import { Fetcher, FetcherOptions } from './fetcher'
 import { BN } from 'ethereumjs-util'
 import { Chain } from '../../blockchain'
 
-
 export interface BlockFetcherOptions extends FetcherOptions {
   /* Blockchain */
   chain: Chain
@@ -15,45 +14,48 @@ export interface BlockFetcherOptions extends FetcherOptions {
 }
 
 export type JobTask = {
-    first: BN 
-    count: number
+  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
+
+export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
+  JobTask,
+  JobResult,
+  StorageItem
+> {
+  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)
     }
-  
-    /**
-     * 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
+    if (count.gtn(0)) {
+      tasks.push({ first: first.clone(), count: count.toNumber() })
     }
-}
+    return tasks
+  }
+}

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

@@ -4,7 +4,7 @@ const Heap = require('qheap')
 import { PeerPool } from '../../net/peerpool'
 import { Config } from '../../config'
 
-import {QHeap, Job} from '../../types'
+import { QHeap, Job } from '../../types'
 import { Peer } from '../../net/peer'
 
 export interface FetcherOptions {
@@ -37,7 +37,7 @@ export interface FetcherOptions {
  * inorder.
  * @memberof module:sync/fetcher
  */
-export abstract class Fetcher<JobTask, JobResult> extends Readable {
+export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable {
   public config: Config
 
   protected pool: PeerPool
@@ -53,7 +53,8 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
   protected finished: number // number of tasks which are both processed and also finished writing
   protected running: boolean
   protected reading: boolean
-  private _readableState?: { // This property is inherited from Readable. We only need `length`.
+  private _readableState?: {
+    // This property is inherited from Readable. We only need `length`.
     length: number
   }
 
@@ -72,8 +73,12 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
     this.maxQueue = options.maxQueue ?? 16
     this.maxPerRequest = options.maxPerRequest ?? 128
 
-    this.in = new Heap({ comparBefore: (a: Job<JobTask, JobResult>, b: Job<JobTask, JobResult>) => a.index < b.index })
-    this.out = new Heap({ comparBefore: (a: Job<JobTask, JobResult>, b: Job<JobTask, JobResult>) => a.index < b.index })
+    this.in = new Heap({
+      comparBefore: (a: Job<JobTask, JobResult>, b: Job<JobTask, JobResult>) => a.index < b.index,
+    })
+    this.out = new Heap({
+      comparBefore: (a: Job<JobTask, JobResult>, b: Job<JobTask, JobResult>) => a.index < b.index,
+    })
     this.total = 0
     this.processed = 0
     this.finished = 0
@@ -131,7 +136,7 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
    * @param  job successful job
    * @param  result job result
    */
-  success(job: Job<JobTask, JobResult>, result?: JobResult[]) {
+  success(job: Job<JobTask, JobResult>, result?: JobResult) {
     if (job.state !== 'active') return
     if (result === undefined) {
       this.enqueue(job)
@@ -194,7 +199,7 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
         this.expire(job)
       }, this.timeout)
       this.request(job, peer)
-        .then((result: JobResult[]) => this.success(job, result))
+        .then((result?: JobResult) => this.success(job, result))
         .catch((error: Error) => this.failure(job, error))
         .finally(() => clearTimeout(timeout))
       return job
@@ -217,7 +222,7 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
    * to support backpressure from storing results.
    */
   write() {
-    const _write = async (result: JobResult[], encoding: string | null, cb: Function) => {
+    const _write = async (result: StorageItem[], encoding: string | null, cb: Function) => {
       try {
         await this.store(result)
         this.finished++
@@ -230,8 +235,14 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
     const writer = new Writable({
       objectMode: true,
       write: _write,
-      writev: (many: { chunk: JobResult, encoding: string}[], cb: Function) =>
-        _write((<JobResult[]>[]).concat(...many.map((x: { chunk: JobResult, encoding: string}) => x.chunk)), null, cb),
+      writev: (many: { chunk: StorageItem; encoding: string }[], cb: Function) =>
+        _write(
+          (<StorageItem[]>[]).concat(
+            ...many.map((x: { chunk: StorageItem; encoding: string }) => x.chunk)
+          ),
+          null,
+          cb
+        ),
     })
     this.on('close', () => {
       this.running = false
@@ -296,14 +307,14 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
    * @param  peer
    * @return {Promise}
    */
-  abstract request(_job?: Job<JobTask, JobResult>, _peer?: Peer): Promise<JobResult[]> 
+  abstract request(_job?: Job<JobTask, JobResult>, _peer?: Peer): Promise<JobResult | undefined>
 
   /**
    * Process the reply for the given job
    * @param  job fetch job
    * @param  result result data
    */
-  abstract process(_job?: Job<JobTask, JobResult>, _result?: JobResult[]): JobResult[] | null
+  abstract process(_job?: Job<JobTask, JobResult>, _result?: JobResult): JobResult | null
 
   /**
    * Expire job that has timed out and ban associated peer. Timed out tasks will
@@ -329,7 +340,7 @@ export abstract class Fetcher<JobTask, JobResult> extends Readable {
    * @param result fetch result
    * @return {Promise}
    */
-  abstract async store(_result?: JobResult[]): Promise<void>
+  abstract async store(_result?: StorageItem[]): Promise<void>
 
   async wait(delay?: number) {
     await new Promise((resolve) => setTimeout(resolve, delay || this.interval))

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

@@ -16,7 +16,7 @@ type BlockHeaderResult = { reqId: BN; bv: BN; headers: BlockHeader[] }
  * Implements an les/1 based header fetcher
  * @memberof module:sync/fetcher
  */
-export class HeaderFetcher extends BlockFetcherBase<BlockHeaderResult> {
+export class HeaderFetcher extends BlockFetcherBase<BlockHeaderResult, BlockHeader> {
   private flow: FlowControl
 
   /**
@@ -39,9 +39,13 @@ export class HeaderFetcher extends BlockFetcherBase<BlockHeaderResult> {
       // we reached our request limit. try with a different peer.
       return undefined
     }
-    const response = await (peer!.les as LesProtocolMethods).getBlockHeaders({ block: task.first, max: task.count })
+    const response = await (peer!.les as LesProtocolMethods).getBlockHeaders({
+      block: task.first,
+      max: task.count,
+    })
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (response) {
-      return response.headers
+      return response
     }
     return undefined
   }

packages/client/lib/types.ts

@@ -1,4 +1,4 @@
-import { Peer } from "./net/peer"
+import { Peer } from './net/peer'
 
 export type Key = Buffer
 export type KeyLike = string | Key
@@ -15,33 +15,31 @@ export type BootnodeLike = string | string[] | Bootnode | Bootnode[]
 
 // QHeap types. @types/qheap does not exist, so we create a custom interface here.
 type QHeapOptions = {
-  comparBefore(a: any, b:any): boolean 
-  compar(a: any, b:any): number
-  freeSpace: number 
-  size: number 
+  comparBefore(a: any, b: any): boolean
+  compar(a: any, b: any): number
+  freeSpace: number
+  size: number
 }
 
-export interface QHeap<T> { 
-  constructor(opts: QHeapOptions): QHeap<T>
+export interface QHeap<T> {
+  // eslint-disable-next-line @typescript-eslint/no-misused-new
+  new (opts: QHeapOptions): QHeap<T>
   insert(item: T): void
   push(item: T): void
-  enqueue(item: T): void 
-  remove(): T | undefined 
+  enqueue(item: T): void
+  remove(): T | undefined
   shift(): T | undefined
   dequeue(): T | undefined
-  peek(): T | undefined 
+  peek(): T | undefined
   length: number
-  gc(opts: {
-    minLength: number
-    maxLength: number 
-  }): void  
+  gc(opts: { minLength: number; maxLength: number }): void
 }
 
 export type Job<JobTask, JobResult> = {
   task: JobTask
   time: number
   index: number
-  result: JobResult[] | null
+  result: JobResult | null
   state: 'idle' | 'expired' | 'active'
   peer: Peer | null
-}
+}