Improve RPC provider request handling

background/package.json

@@ -73,6 +73,7 @@
     "@types/sinon": "^10.0.12",
     "@types/uuid": "^8.3.4",
     "@types/webextension-polyfill": "^0.12.0",
+    "@types/ws": "^8.5.14",
     "@walletconnect/legacy-types": "^2.0.0",
     "@walletconnect/types": "^2.7.7",
     "crypto-browserify": "^3.12.0",

background/services/chain/serial-fallback-provider.ts

@@ -1,6 +1,5 @@
 import {
   EventType,
-  JsonRpcBatchProvider,
   JsonRpcProvider,
   Listener,
   WebSocketProvider,
@@ -28,13 +27,18 @@ import { FeatureFlags, isEnabled } from "../../features"
 import { RpcConfig } from "./db"
 import TahoAlchemyProvider from "./taho-provider"
 import { getErrorType } from "./errors"
+import TahoRPCProvider from "./taho-rpc-provider"
 
 export type ProviderCreator = {
   type: "alchemy" | "custom" | "generic"
   supportedMethods?: string[]
-  creator: () => WebSocketProvider | JsonRpcProvider
+  creator: () => JsonRpcProvider
 }
 
+const isWebSocketProvider = (
+  provider: JsonRpcProvider,
+): provider is WebSocketProvider => provider instanceof WebSocketProvider
+
 /**
  * Method list, to describe which rpc method calls on which networks should
  * prefer alchemy provider over the generic ones.
@@ -111,11 +115,8 @@ function backedOffMs(): number {
 function isClosedOrClosingWebSocketProvider(
   provider: JsonRpcProvider,
 ): boolean {
-  if (provider instanceof WebSocketProvider) {
-    // Digging into the innards of Ethers here because there's no
-    // other way to get access to the WebSocket connection situation.
-    // eslint-disable-next-line no-underscore-dangle
-    const webSocket = provider._websocket as WebSocket
+  if (isWebSocketProvider(provider)) {
+    const webSocket = provider.websocket
 
     return (
       webSocket.readyState === WebSocket.CLOSING ||
@@ -131,11 +132,8 @@ function isClosedOrClosingWebSocketProvider(
  * connecting. Ethers does not provide direct access to this information.
  */
 function isConnectingWebSocketProvider(provider: JsonRpcProvider): boolean {
-  if (provider instanceof WebSocketProvider) {
-    // Digging into the innards of Ethers here because there's no
-    // other way to get access to the WebSocket connection situation.
-    // eslint-disable-next-line no-underscore-dangle
-    const webSocket = provider._websocket as WebSocket
+  if (isWebSocketProvider(provider)) {
+    const webSocket = provider.websocket
     return webSocket.readyState === WebSocket.CONNECTING
   }
 
@@ -195,10 +193,7 @@ function customOrDefaultProvider(
 export default class SerialFallbackProvider extends JsonRpcProvider {
   // Functions that will create and initialize a new provider, in priority
   // order.
-  private providerCreators: [
-    () => WebSocketProvider | JsonRpcProvider,
-    ...(() => JsonRpcProvider)[],
-  ]
+  private providerCreators: (() => JsonRpcProvider)[]
 
   // The currently-used provider, produced by the provider-creator at
   // currentProviderIndex.
@@ -227,9 +222,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
     }
   } = {}
 
-  private alchemyProviderCreator:
-    | (() => WebSocketProvider | JsonRpcProvider)
-    | undefined
+  private alchemyProviderCreator: (() => JsonRpcProvider) | undefined
 
   supportsAlchemy = false
 
@@ -247,21 +240,6 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
   // for reconnects when relevant.
   private currentProviderIndex = 0
 
-  // If nonzero and the underlying provider is a batch provider, forces the
-  // batch size to be no more than this number, holding other requests until
-  // the existing batch has cleared.
-  private forcedBatchMaxSize: number = 0
-
-  // If this promise is set, new RPC calls will await on it before being
-  // processed. When forcedBatchMaxSize is nonzero and that number of RPC calls
-  // are pending, this promise will be set so subsequent requests will wait
-  // until the batch flushes.
-  private forcedBatchMaxPromise: Promise<void> | undefined = undefined
-
-  // During max size update, this value is set so that the value is not
-  // decreased by multiple failed requests.
-  private forcedBatchMaxPreviousSize: number = 0
-
   // TEMPORARY cache for latest account balances to reduce number of rpc calls
   // This is intended as a temporary fix to the burst of account enrichment that
   // happens when the extension is first loaded up as a result of activity emission
@@ -281,7 +259,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
   // reloaded and the property of having code updates quite rarely.
   private latestHasCodeCache: {
     [address: string]: {
-      hasCode: boolean
+      hasCode: string
     }
   } = {}
 
@@ -372,46 +350,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
       delete this.messagesToSend[messageId]
       return cachedResult
     }
-
-    if (this.forcedBatchMaxPromise) {
-      await this.forcedBatchMaxPromise
-    }
-
-    const pendingBatch =
-      "_pendingBatch" in this.currentProvider
-        ? // Accessing ethers internals for forced max batch sizing.
-          // eslint-disable-next-line no-underscore-dangle
-          (this.currentProvider._pendingBatch as { length: number } | undefined)
-        : undefined
-    const pendingBatchSize = pendingBatch?.length
     const existingProviderIndex = this.currentProviderIndex
-
-    if (
-      pendingBatch &&
-      this.forcedBatchMaxSize &&
-      // Accessing ethers internals for forced max batch sizing.
-      // eslint-disable-next-line no-underscore-dangle
-      pendingBatch.length >= this.forcedBatchMaxSize
-    ) {
-      this.forcedBatchMaxPromise = new Promise((resolve) => {
-        const checkInterval = setInterval(() => {
-          const latestPendingBatch =
-            "_pendingBatch" in this.currentProvider
-              ? // Accessing ethers internals for forced max batch sizing.
-                // eslint-disable-next-line no-underscore-dangle
-                (this.currentProvider._pendingBatch as
-                  | { length: number }
-                  | undefined)
-              : undefined
-
-          if ((latestPendingBatch?.length ?? 0) < this.forcedBatchMaxSize) {
-            resolve()
-            clearInterval(checkInterval)
-          }
-        }, 5)
-      })
-    }
-
     try {
       if (isClosedOrClosingWebSocketProvider(this.currentProvider)) {
         // Detect disconnected WebSocket and immediately throw.
@@ -451,44 +390,41 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
       return result
     } catch (error) {
       // Awful, but what can ya do.
+
       const stringifiedError = String(error)
 
       const errorType = getErrorType(stringifiedError, method)
 
       if (
         errorType === "batch-limit-exceeded" &&
-        (pendingBatchSize === undefined || pendingBatchSize === 0)
+        this.currentProvider instanceof TahoRPCProvider
       ) {
-        this.forcedBatchMaxPreviousSize =
-          pendingBatch?.length ?? pendingBatchSize ?? 1
-        this.forcedBatchMaxSize =
-          (pendingBatch?.length ?? pendingBatchSize ?? 2) / 2
+        const requestBatch = this.currentProvider.getBatchFromError(error)
 
-        logger.debug(
-          "Setting a max batch size of",
-          this.forcedBatchMaxSize,
-          "on chain",
-          this.chainID,
-          "and retrying: ",
-          method,
-          params,
-        )
+        // Note that every other request in the batch will set the length to
+        // the same value
+        if (
+          requestBatch.length <=
+          this.currentProvider.getOptions().maxBatchLength
+        ) {
+          const newMaxBatchLen = Math.max(
+            Math.floor(requestBatch.length / 2),
+            1,
+          )
 
-        return this.routeRpcCall(messageId)
-      }
+          this.currentProvider.setOptions({
+            maxBatchLength: newMaxBatchLen,
+          })
 
-      if (errorType === "batch-limit-exceeded") {
-        logger.debug(
-          "Using max batch size of",
-          this.forcedBatchMaxSize,
-          "on chain",
-          this.chainID,
-          "and retrying: ",
-          method,
-          params,
-        )
-
-        return this.routeRpcCall(messageId)
+          logger.debug(
+            "Setting a max batch size of",
+            newMaxBatchLen,
+            "for rpc",
+            this.currentProvider.connection.url,
+          )
+        }
+        // Retry with a new limit on batch length
+        return waitAnd(500, () => this.routeRpcCall(messageId))
       }
 
       if (
@@ -617,7 +553,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
     if (method === "eth_getCode" && (params as string[])[1] === "latest") {
       const address = (params as string[])[0]
       this.latestHasCodeCache[address] = {
-        hasCode: result as boolean,
+        hasCode: result as string,
       }
     }
   }
@@ -756,7 +692,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
   ): Promise<void> {
     const subscription = { tag, param, processFunc }
 
-    if (this.currentProvider instanceof WebSocketProvider) {
+    if (isWebSocketProvider(this.currentProvider)) {
       // eslint-disable-next-line no-underscore-dangle
       await this.currentProvider._subscribe(tag, param, processFunc)
       this.subscriptions.push(subscription)
@@ -827,7 +763,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
   }
 
   /**
-   * Behaves the same as the `JsonRpcProvider` `on` method, but also trakcs the
+   * Behaves the same as the `JsonRpcProvider` `on` method, but also tracks the
    * event subscription so that an underlying provider failure will not prevent
    * it from firing.
    */
@@ -844,7 +780,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
   }
 
   /**
-   * Behaves the same as the `JsonRpcProvider` `once` method, but also trakcs
+   * Behaves the same as the `JsonRpcProvider` `once` method, but also tracks
    * the event subscription so that an underlying provider failure will not
    * prevent it from firing.
    */
@@ -911,12 +847,12 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
   private disconnectCurrentProvider() {
     logger.debug(
       "Disconnecting current provider; websocket: ",
-      this.currentProvider instanceof WebSocketProvider,
+      isWebSocketProvider(this.currentProvider),
       "on chain",
       this.chainID,
       ".",
     )
-    if (this.currentProvider instanceof WebSocketProvider) {
+    if (isWebSocketProvider(this.currentProvider)) {
       this.currentProvider.destroy()
     } else {
       // For non-WebSocket providers, kill all subscriptions so the listeners
@@ -1009,9 +945,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
       return false
     }
 
-    if (provider instanceof WebSocketProvider) {
-      const websocketProvider = provider as WebSocketProvider
-
+    if (isWebSocketProvider(provider)) {
       // Chain promises to serially resubscribe.
       //
       // TODO If anything fails along the way, it should yield the same kind of
@@ -1023,7 +957,7 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
               // Direct subscriptions are internal, but we want to be able to
               // restore them.
               // eslint-disable-next-line no-underscore-dangle
-              websocketProvider._subscribe(tag, param, processFunc),
+              provider._subscribe(tag, param, processFunc),
             ),
           ),
         Promise.resolve(),
@@ -1162,23 +1096,13 @@ export default class SerialFallbackProvider extends JsonRpcProvider {
   }
 }
 
-function getProviderCreator(
-  rpcUrl: string,
-): JsonRpcProvider | WebSocketProvider {
+function getProviderCreator(rpcUrl: string): JsonRpcProvider {
   const url = new URL(rpcUrl)
   if (/^wss?/.test(url.protocol)) {
     return new WebSocketProvider(rpcUrl)
   }
 
-  if (/rpc\.ankr\.com|1rpc\.io|polygon-rpc\.com/.test(url.href)) {
-    return new JsonRpcBatchProvider({
-      url: rpcUrl,
-      throttleLimit: 1,
-      timeout: PROVIDER_REQUEST_TIMEOUT,
-    })
-  }
-
-  return new JsonRpcProvider({
+  return new TahoRPCProvider({
     url: rpcUrl,
     throttleLimit: 1,
     timeout: PROVIDER_REQUEST_TIMEOUT,
@@ -1189,7 +1113,8 @@ export function makeFlashbotsProviderCreator(): ProviderCreator {
   return {
     type: "custom",
     supportedMethods: ["eth_sendRawTransaction"],
-    creator: () => getProviderCreator(FLASHBOTS_RPC_URL),
+    creator: () =>
+      new TahoRPCProvider(FLASHBOTS_RPC_URL, undefined, { maxBatchLength: 1 }),
   }
 }
 
@@ -1202,7 +1127,7 @@ export function makeSerialFallbackProvider(
     return new SerialFallbackProvider(FORK.chainID, [
       {
         type: "generic" as const,
-        creator: () => new JsonRpcProvider(process.env.MAINNET_FORK_URL),
+        creator: () => new TahoRPCProvider(process.env.MAINNET_FORK_URL),
       },
     ])
   }
@@ -1221,7 +1146,7 @@ export function makeSerialFallbackProvider(
     return new SerialFallbackProvider(ARBITRUM_SEPOLIA.chainID, [
       {
         type: "generic" as const,
-        creator: () => new JsonRpcBatchProvider(process.env.ARBITRUM_FORK_RPC),
+        creator: () => new TahoRPCProvider(process.env.ARBITRUM_FORK_RPC),
       },
     ])
   }