Replace readable-stream with web stream api

Author: am1r021

packages/statemanager/src/stateManager.ts

@@ -962,21 +962,14 @@ export class DefaultStateManager implements EVMStateManagerInterface {
       throw new Error(`dumpStorage f() can only be called for an existing account`)
     }
     const trie = this._getStorageTrie(address, account)
+    const storage: StorageDump = {}
+    const stream = trie.createReadStream()
 
-    return new Promise((resolve, reject) => {
-      const storage: StorageDump = {}
-      const stream = trie.createReadStream()
+    for await (const chunk of stream) {
+      if (chunk !== null) storage[bytesToHex(chunk.key)] = bytesToHex(chunk.value)
+    }
 
-      stream.on('data', (val: any) => {
-        storage[bytesToHex(val.key)] = bytesToHex(val.value)
-      })
-      stream.on('end', () => {
-        resolve(storage)
-      })
-      stream.on('error', (e) => {
-        reject(e)
-      })
-    })
+    return storage
   }
 
   /**
@@ -999,44 +992,33 @@ export class DefaultStateManager implements EVMStateManagerInterface {
       throw new Error(`Account does not exist.`)
     }
     const trie = this._getStorageTrie(address, account)
-
-    return new Promise((resolve, reject) => {
-      let inRange = false
-      let i = 0
-
-      /** Object conforming to {@link StorageRange.storage}. */
-      const storageMap: StorageRange['storage'] = {}
-      const stream = trie.createReadStream()
-
-      stream.on('data', (val: any) => {
+    let inRange = false
+    let i = 0
+
+    /** Object conforming to {@link StorageRange.storage}. */
+    const storageMap: StorageRange['storage'] = {}
+    const stream = trie.createReadStream()
+    for await (const chunk of stream) {
+      if (chunk !== null) {
         if (!inRange) {
           // Check if the key is already in the correct range.
-          if (bytesToBigInt(val.key) >= startKey) {
+          if (bytesToBigInt(chunk.key) >= startKey) {
             inRange = true
           } else {
-            return
+            continue
           }
         }
-
         if (i < limit) {
-          storageMap[bytesToHex(val.key)] = { key: null, value: bytesToHex(val.value) }
+          storageMap[bytesToHex(chunk.key)] = { key: null, value: bytesToHex(chunk.value) }
           i++
         } else if (i === limit) {
-          resolve({
+          return {
             storage: storageMap,
-            nextKey: bytesToHex(val.key),
-          })
+            nextKey: bytesToHex(chunk.key),
+          }
         }
-      })
-
-      stream.on('end', () => {
-        resolve({
-          storage: storageMap,
-          nextKey: null,
-        })
-      })
-      stream.on('error', (e) => reject(e))
-    })
+      }
+    }
   }
 
   /**

packages/trie/src/trie.ts

@@ -43,6 +43,8 @@ import type {
 import type { OnFound } from './util/asyncWalk.js'
 import type { BatchDBOp, DB, PutBatch } from '@ethereumjs/util'
 import type { Debugger } from 'debug'
+// eslint-disable-next-line implicit-dependencies/no-implicit
+import type { ReadableStream } from 'node:stream/web'
 
 interface Path {
   node: TrieNode | null
@@ -1080,8 +1082,8 @@ export class Trie {
    * The `data` event is given an `Object` that has two properties; the `key` and the `value`. Both should be Uint8Arrays.
    * @return Returns a [stream](https://nodejs.org/dist/latest-v12.x/docs/api/stream.html#stream_class_stream_readable) of the contents of the `trie`
    */
-  createReadStream(): ReadStream {
-    return new ReadStream(this)
+  createReadStream(): ReadableStream {
+    return ReadStream(this)
   }
 
   /**

packages/trie/src/util/readStream.ts

@@ -1,6 +1,5 @@
 // eslint-disable-next-line implicit-dependencies/no-implicit
 import { ReadableStream } from 'node:stream/web'
-import { Readable } from 'readable-stream'
 
 import { BranchNode, LeafNode } from '../node/index.js'
 
@@ -9,53 +8,10 @@ import { nibblestoBytes } from './nibbles.js'
 import type { Trie } from '../trie.js'
 import type { FoundNodeFunction } from '../types.js'
 
-export class TrieReadStream extends Readable {
-  private trie: Trie
-  private _started: boolean
-
-  constructor(trie: Trie) {
-    super({ objectMode: true })
-
-    const s = new ReadableStream()
-    void s.cancel()
-    this.trie = trie
-    this._started = false
-  }
-
-  async _read() {
-    if (this._started) {
-      return
-    }
-    this._started = true
-    try {
-      await this._findValueNodes(async (_, node, key, walkController) => {
-        if (node !== null) {
-          this.push({
-            key: nibblestoBytes(key),
-            value: node.value(),
-          })
-          walkController.allChildren(node, key)
-        }
-      })
-    } catch (error: any) {
-      if (error.message === 'Missing node in DB') {
-        // pass
-      } else {
-        throw error
-      }
-    }
-    this.push(null)
-  }
-
-  /**
-   * Finds all nodes that store k,v values
-   * called by {@link TrieReadStream}
-   * @private
-   */
-  async _findValueNodes(onFound: FoundNodeFunction): Promise<void> {
+export function TrieReadStream(trie: Trie) {
+  const _findValueNodes = async (onFound: FoundNodeFunction): Promise<void> => {
     const outerOnFound: FoundNodeFunction = async (nodeRef, node, key, walkController) => {
       let fullKey = key
-
       if (node instanceof LeafNode) {
         fullKey = key.concat(node.key())
         // found leaf node!
@@ -70,6 +26,29 @@ export class TrieReadStream extends Readable {
         }
       }
     }
-    await this.trie.walkTrie(this.trie.root(), outerOnFound)
+    await trie.walkTrie(trie.root(), outerOnFound)
   }
+
+  return new ReadableStream({
+    async start(controller) {
+      try {
+        await _findValueNodes(async (_, node, key, walkController) => {
+          if (node !== null) {
+            controller.enqueue({
+              key: nibblestoBytes(key),
+              value: node.value(),
+            })
+            walkController.allChildren(node, key)
+          }
+        })
+      } catch (error: any) {
+        if (error.message === 'Missing node in DB') {
+          // pass
+        } else {
+          throw error
+        }
+      }
+      controller.close()
+    },
+  })
 }