nits, improvements, cleanup after #1858 (#1863)

Author: ryanio

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

@@ -106,7 +106,7 @@ export class BlockFetcher extends BlockFetcherBase<Block[], Block> {
           blocks[0]?.header.number
         } last=${blocks[blocks.length - 1]?.header.number})`
       )
-      this.config.events.emit(Event.SYNC_FETCHER_FETCHED, blocks.slice(0, num))
+      this.config.events.emit(Event.SYNC_FETCHED_BLOCKS, blocks.slice(0, num))
     } catch (e: any) {
       this.debug(
         `Error storing fetcher results in blockchain (blocks num=${blocks.length} first=${

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

@@ -28,13 +28,12 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
 > {
   protected chain: Chain
   /**
-   * `first` is from where the fetcher pendancy starts apart from the jobs/tasks already
-   *  in the `in` queue.
+   * Where the fetcher starts apart from the tasks already in the `in` queue
    */
   first: BN
   /**
-   * `count` are the number of items in fetcher pendancy starting from (and including)
-   * `first`. `first + count - 1` gives the height fetcher is attempting to reach
+   * Number of items in the fetcher starting from (and including) `first`.
+   * `first + count - 1` gives the height fetcher is attempting to reach
    */
   count: BN
 
@@ -53,10 +52,10 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
   }
 
   /**
-   * Generate list of tasks to fetch, modifies `first`` and `count` to indicate any
-   * remaning pendancy of the fetcher apart from jobs/tasks it pushes in the queue
+   * Generate list of tasks to fetch. Modifies `first` and `count` to indicate
+   * remaining items apart from the tasks it pushes in the queue
    */
-  tasks(first = this.first, count = this.count, maxTasks = Infinity): JobTask[] {
+  tasks(first = this.first, count = this.count, maxTasks = this.config.maxFetcherJobs): JobTask[] {
     const max = this.config.maxPerRequest
     const tasks: JobTask[] = []
     let debugStr = `first=${first}`
@@ -72,15 +71,15 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
       tasks.push({ first: first.clone(), count: count.toNumber() })
       pushedCount.iadd(count)
     }
-    debugStr = `${debugStr} count=${pushedCount}`
+    debugStr += ` count=${pushedCount}`
     this.debug(`Created new tasks num=${tasks.length} ${debugStr}`)
     return tasks
   }
 
   nextTasks(): void {
     if (this.in.length === 0 && this.count.gten(0)) {
-      this.debug(`Fetcher has pendancy with first=${this.first} count=${this.count}`)
-      const tasks = this.tasks(this.first, this.count, this.config.maxFetcherJobs)
+      this.debug(`Fetcher pending with first=${this.first} count=${this.count}`)
+      const tasks = this.tasks(this.first, this.count)
       for (const task of tasks) {
         this.enqueueTask(task)
       }
@@ -95,9 +94,9 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
     let first = this.first
     let last = this.first.add(this.count).subn(1)
 
-    // we have to loop and figure out because the jobs won't always be an monotonically
-    // increasing order, some jobs could have refetch tasks enqueued. So better to find
-    // the first and last examining each job
+    // We have to loop because the jobs won't always be in increasing order.
+    // Some jobs could have refetch tasks enqueued, so it is better to find
+    // the first and last by examining each job.
     while (this.in.length > 0) {
       const job = this.in.remove()
       if (!job) break
@@ -111,7 +110,7 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
     }
     this.first = first
     this.count = last.sub(this.first).addn(1)
-    // already removed jobs from the `in` heap, just pass to super for further cleanup
+    // Already removed jobs from the `in` heap, just pass to super for further cleanup
     super.clear()
   }
 
@@ -134,8 +133,8 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
       this.count.iadd(max.sub(last))
       updateHeightStr = `updated height=${max}`
     }
-    // Re enqueue the numbers which are < `this.first`  to refetch them else they will be
-    // fetched in the future automatically on the jobs created by `nextTasks`
+    // Re-enqueue the numbers which are less than `first` to refetch them,
+    // else they will be fetched in the future with the jobs created by `nextTasks`.
     numberList = numberList.filter((num) => num.lte(this.first))
     const numBlocks = numberList.length
     let bulkRequest = true

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

@@ -188,7 +188,7 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
       state: 'idle',
       peer: null,
     }
-    this.debug(`enqueueTask ${this.jobStr(job)}`)
+    this.debug(`enqueueTask: ${this.jobStr(job)}`)
     this.in.insert(job)
     if (!this.running && autoRestart) {
       void this.fetch()

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

@@ -88,7 +88,7 @@ export class HeaderFetcher extends BlockFetcherBase<BlockHeaderResult, BlockHead
           headers[0]?.number
         } last=${headers[headers.length - 1]?.number})`
       )
-      this.config.events.emit(Event.SYNC_FETCHER_FETCHED, headers.slice(0, num))
+      this.config.events.emit(Event.SYNC_FETCHED_HEADERS, headers.slice(0, num))
     } catch (e: any) {
       this.debug(
         `Error storing fetcher results in blockchain (headers num=${headers.length} first=${

packages/client/lib/sync/fullsync.ts

@@ -6,7 +6,7 @@ import { Event } from '../types'
 import { Synchronizer, SynchronizerOptions } from './sync'
 import { BlockFetcher } from './fetcher'
 import { VMExecution } from '../execution'
-import type { Block, BlockHeader } from '@ethereumjs/block'
+import type { Block } from '@ethereumjs/block'
 import type { TxPool } from '../service/txpool'
 
 interface FullSynchronizerOptions extends SynchronizerOptions {
@@ -39,7 +39,7 @@ export class FullSynchronizer extends Synchronizer {
     this.processBlocks = this.processBlocks.bind(this)
     this.stop = this.stop.bind(this)
 
-    this.config.events.on(Event.SYNC_FETCHER_FETCHED, this.processBlocks)
+    this.config.events.on(Event.SYNC_FETCHED_BLOCKS, this.processBlocks)
     this.config.events.on(Event.SYNC_EXECUTION_VM_ERROR, this.stop)
 
     void this.chain.update()
@@ -113,48 +113,50 @@ export class FullSynchronizer extends Synchronizer {
   }
 
   /**
-   * Sync all blocks and state from peer starting from current height.
+   * Called from `sync()` to sync blocks and state from peer starting from current height.
    * @param peer remote peer to sync with
-   * @return Resolves when sync completed
+   * @returns a boolean if the setup was successful
    */
-  async syncWithPeer(peer?: Peer): Promise<BlockFetcher | null> {
-    let latest
-    if (peer && (latest = await this.latest(peer))) {
-      const height = latest.number
-      if (!this.config.syncTargetHeight || this.config.syncTargetHeight.lt(latest.number)) {
-        this.config.syncTargetHeight = height
-        this.config.logger.info(`New sync target height=${height} hash=${short(latest.hash())}`)
-      }
+  async syncWithPeer(peer?: Peer): Promise<boolean> {
+    const latest = peer ? await this.latest(peer) : undefined
+    if (!latest) return false
+
+    const height = latest.number
+    if (!this.config.syncTargetHeight || this.config.syncTargetHeight.lt(latest.number)) {
+      this.config.syncTargetHeight = height
+      this.config.logger.info(`New sync target height=${height} hash=${short(latest.hash())}`)
+    }
 
-      // Start fetcher from a safe distance behind because if the previous fetcher exited
-      // due to a reorg, it would make sense to step back and refetch.
-      const first = BN.max(
-        this.chain.blocks.height.addn(1).subn(this.config.safeReorgDistance),
-        new BN(1)
-      )
-      const count = height.sub(first).addn(1)
-      if (count.gtn(0)) {
-        if (!this.fetcher || this.fetcher.errored) {
-          return new BlockFetcher({
-            config: this.config,
-            pool: this.pool,
-            chain: this.chain,
-            interval: this.interval,
-            first,
-            count,
-            destroyWhenDone: false,
-          })
-        } else {
-          const fetcherHeight = this.fetcher.first.add(this.fetcher.count).subn(1)
-          if (height.gt(fetcherHeight)) this.fetcher.count.iadd(height.sub(fetcherHeight))
-          this.config.logger.info(`peer=${peer} updated fetcher target to height=${height}`)
-        }
-      }
+    // Start fetcher from a safe distance behind because if the previous fetcher exited
+    // due to a reorg, it would make sense to step back and refetch.
+    const first = BN.max(
+      this.chain.blocks.height.addn(1).subn(this.config.safeReorgDistance),
+      new BN(1)
+    )
+    const count = height.sub(first).addn(1)
+    if (count.lten(0)) return false
+    if (!this.fetcher || this.fetcher.errored) {
+      this.fetcher = new BlockFetcher({
+        config: this.config,
+        pool: this.pool,
+        chain: this.chain,
+        interval: this.interval,
+        first,
+        count,
+        destroyWhenDone: false,
+      })
+    } else {
+      const fetcherHeight = this.fetcher.first.add(this.fetcher.count).subn(1)
+      if (height.gt(fetcherHeight)) this.fetcher.count.iadd(height.sub(fetcherHeight))
+      this.config.logger.info(`Updated fetcher target to height=${height} peer=${peer} `)
     }
-    return null
+    return true
   }
 
-  async processBlocks(blocks: Block[] | BlockHeader[]) {
+  /**
+   * Process blocks fetched from the fetcher.
+   */
+  async processBlocks(blocks: Block[]) {
     if (this.config.chainCommon.gteHardfork(Hardfork.Merge)) {
       if (this.fetcher !== null) {
         // If we are beyond the merge block we should stop the fetcher
@@ -170,7 +172,6 @@ export class FullSynchronizer extends Synchronizer {
       return
     }
 
-    blocks = blocks as Block[]
     const first = blocks[0].header.number
     const last = blocks[blocks.length - 1].header.number
     const hash = short(blocks[0].hash())
@@ -305,38 +306,33 @@ export class FullSynchronizer extends Synchronizer {
     let min = new BN(-1)
     let newSyncHeight: [Buffer, BN] | undefined
     const blockNumberList: BN[] = []
-    data.forEach((value) => {
+    for (const value of data) {
       const blockNumber = value[1]
       blockNumberList.push(blockNumber)
       if (min.eqn(-1) || blockNumber.lt(min)) {
         min = blockNumber
       }
 
       // Check if new sync target height can be set
-      if (
-        (!newSyncHeight &&
-          (!this.config.syncTargetHeight || blockNumber.gt(this.config.syncTargetHeight))) ||
-        (newSyncHeight && blockNumber.gt(newSyncHeight[1]))
-      ) {
-        newSyncHeight = value
-      }
-    })
+      if (newSyncHeight && blockNumber.lte(newSyncHeight[1])) continue
+      if (this.config.syncTargetHeight && blockNumber.lte(this.config.syncTargetHeight)) continue
+      newSyncHeight = value
+    }
 
     if (!newSyncHeight) return
     const [hash, height] = newSyncHeight
     this.config.syncTargetHeight = height
     this.config.logger.info(`New sync target height number=${height} hash=${short(hash)}`)
-    // enqueue if we are close enough to chain head
+    // Enqueue if we are close enough to chain head
     if (min.lt(this.chain.headers.height.addn(3000))) {
       this.fetcher.enqueueByNumberList(blockNumberList, min, height)
     }
   }
 
   async stop(): Promise<boolean> {
-    this.config.events.removeListener(Event.SYNC_FETCHER_FETCHED, this.processBlocks)
+    this.config.events.removeListener(Event.SYNC_FETCHED_BLOCKS, this.processBlocks)
     this.config.events.removeListener(Event.SYNC_EXECUTION_VM_ERROR, this.stop)
-
-    return await super.stop()
+    return super.stop()
   }
 
   /**