feat: add connection retry to proxyProvider

deps/std/async.ts

@@ -1 +1 @@
-export * from "https://deno.land/std@0.127.0/async/mod.ts"
+export * from "https://deno.land/std@0.170.0/async/mod.ts"

examples/derived.ts

@@ -1,9 +1,18 @@
 import * as C from "http://localhost:5646/@local/mod.ts"
 import * as U from "http://localhost:5646/@local/util/mod.ts"
 
-const ids = C.entryRead(C.polkadot)("Paras", "Parachains", [])
-  .access("value")
-  .as<number[]>()
+// TODO: uncomment these lines to run a single effect upon solving `count` in zones
+// or effects/rpc.ts#discardCheck
+// const ids = C.entryRead(C.polkadot)("Paras", "Parachains", [])
+//   .access("value")
+//   .as<number[]>()
+
+const ids = U.throwIfError(
+  await C.entryRead(C.polkadot)("Paras", "Parachains", [])
+    .access("value")
+    .as<number[]>()
+    .run(),
+)
 
 const root = C.Z.each(ids, (id) => {
   return C.entryRead(C.polkadot)("Paras", "Heads", [id])

rpc/provider/base.ts

@@ -71,3 +71,54 @@ export function nextIdFactory() {
   let i = 0
   return () => i++
 }
+
+export class ListenersContainer<
+  DiscoveryValue,
+  SendErrorData,
+  HandlerErrorData,
+> {
+  #listeners = new Map<
+    DiscoveryValue,
+    Map<
+      ProviderListener<SendErrorData, HandlerErrorData>,
+      ProviderListener<SendErrorData, HandlerErrorData>
+    >
+  >()
+
+  set(discoveryValue: DiscoveryValue, listener: ProviderListener<SendErrorData, HandlerErrorData>) {
+    const map = U.getOrInit(this.#listeners, discoveryValue, () =>
+      new Map<
+        ProviderListener<SendErrorData, HandlerErrorData>,
+        ProviderListener<SendErrorData, HandlerErrorData>
+      >())
+    if (map.has(listener)) return
+    map.set(
+      listener,
+      listener.bind({
+        stop: () => map!.delete(listener),
+      }),
+    )
+  }
+
+  delete(
+    discoveryValue: DiscoveryValue,
+    listener: ProviderListener<SendErrorData, HandlerErrorData>,
+  ) {
+    this.#listeners.get(discoveryValue)?.delete(listener)
+  }
+
+  count(discoveryValue: DiscoveryValue) {
+    return this.#listeners.get(discoveryValue)?.size ?? 0
+  }
+
+  forEachListener(
+    discoveryValue: DiscoveryValue,
+    message: Parameters<ProviderListener<SendErrorData, HandlerErrorData>>[0],
+  ) {
+    const map = this.#listeners.get(discoveryValue)
+    if (!map) return
+    for (const listener of map.values()) {
+      listener(message)
+    }
+  }
+}

rpc/provider/proxy.test.ts

@@ -1,15 +1,25 @@
 import * as A from "../../deps/std/testing/asserts.ts"
 import * as T from "../../test_util/mod.ts"
-import { proxyProvider } from "./proxy.ts"
+import { proxyProviderFactory } from "./proxy.ts"
 import { setup } from "./test_util.ts"
 
 Deno.test({
   name: "Proxy Provider",
+  sanitizeResources: false,
+  sanitizeOps: false,
   async fn(t) {
     await t.step({
       name: "send/listen",
+      // TODO: await T.polkadot initializes a 2nd proxyProvider
+      sanitizeResources: false,
+      sanitizeOps: false,
       async fn() {
-        const [ref, message] = await setup(proxyProvider, await T.polkadot.url, "system_health", [])
+        const [ref, message] = await setup(
+          proxyProviderFactory(),
+          await T.polkadot.url,
+          "system_health",
+          [],
+        )
         A.assertNotInstanceOf(message, Error)
         A.assertExists(message.result)
         A.assertNotInstanceOf(await ref.release(), Error)
@@ -20,7 +30,7 @@ Deno.test({
       name: "create WebSocket error",
       async fn() {
         const [ref, message] = await setup(
-          proxyProvider,
+          proxyProviderFactory({ retryOptions: { maxAttempts: 1 } }),
           "invalid-endpoint-url",
           "system_health",
           [],
@@ -37,7 +47,7 @@ Deno.test({
           this.close()
         })
         const [ref, message] = await setup(
-          proxyProvider,
+          proxyProviderFactory(),
           server.url,
           "system_health",
           [],
@@ -53,7 +63,7 @@ Deno.test({
       async fn() {
         const server = createWebSocketServer()
         const [ref, message] = await setup(
-          proxyProvider,
+          proxyProviderFactory(),
           server.url,
           "system_health",
           // make JSON.stringify to throw

rpc/provider/proxy.ts

@@ -1,95 +1,126 @@
-import * as U from "../../util/mod.ts"
+import { retry, RetryOptions } from "../../deps/std/async.ts"
 import * as msg from "../messages.ts"
-import { nextIdFactory, Provider, ProviderConnection, ProviderListener } from "./base.ts"
+import { ListenersContainer, nextIdFactory, Provider, ProviderListener } from "./base.ts"
 import { ProviderCloseError, ProviderHandlerError, ProviderSendError } from "./errors.ts"
 
-/** Global lookup of existing connections */
-const connections = new Map<string, ProxyProviderConnection>()
-type ProxyProviderConnection = ProviderConnection<WebSocket, Event, Event>
+const CUSTOM_WS_CLOSE_CODE = 4000
 
-const nextId = nextIdFactory()
+export interface ProxyProviderFactoryProps {
+  retryOptions?: RetryOptions
+}
 
-export const proxyProvider: Provider<string, Event, Event, Event> = (url, listener) => {
-  return {
-    nextId,
-    send: (message) => {
-      let conn
-      try {
-        conn = connection(url, listener)
-      } catch (error) {
-        listener(new ProviderHandlerError(error as Event))
-        return
-      }
-      ;(async () => {
-        const openError = await ensureWsOpen(conn.inner)
-        if (openError) {
-          conn.forEachListener(new ProviderSendError(openError, message))
-          return
-        }
-        try {
-          conn.inner.send(JSON.stringify(message))
-        } catch (error) {
-          listener(new ProviderSendError(error as Event, message))
+export const proxyProviderFactory = (
+  { retryOptions }: ProxyProviderFactoryProps = {},
+): Provider<string, Event, Event, Event> => {
+  const listenersContainer = new ListenersContainer<string, Event, Event>()
+  const activeWs = new Map<string, WebSocket>()
+  const connectingWs = new Map<string, Promise<WebSocket>>()
+  return (url, listener) => {
+    listenersContainer.set(url, listener)
+    let ws: WebSocket | undefined
+    return {
+      nextId: nextIdFactory(),
+      send: (message) => {
+        ;(async () => {
+          try {
+            ws = await openedWs({
+              url,
+              activeWs,
+              connectingWs,
+              retryOptions,
+              listener: (e) => listenersContainer.forEachListener(url, e),
+            })
+          } catch (error) {
+            return listener(new ProviderHandlerError(error as Event))
+          }
+          try {
+            ws.send(JSON.stringify(message))
+          } catch (error) {
+            listener(new ProviderSendError(error as Event, message))
+          }
+        })()
+      },
+      release: () => {
+        listenersContainer.delete(url, listener)
+        if (!listenersContainer.count(url) && ws) {
+          return closeWs(ws)
         }
-      })()
-    },
-    release: () => {
-      const conn = connections.get(url)
-      if (!conn) {
         return Promise.resolve(undefined)
-      }
-      const { cleanUp, listeners, inner } = conn
-      listeners.delete(listener)
-      if (!listeners.size) {
-        connections.delete(url)
-        cleanUp()
-        return closeWs(inner)
-      }
-      return Promise.resolve(undefined)
-    },
+      },
+    }
   }
 }
 
-function connection(
-  url: string,
-  listener: ProviderListener<Event, Event>,
-): ProxyProviderConnection {
-  const conn = U.getOrInit(connections, url, () => {
-    const controller = new AbortController()
-    const ws = new WebSocket(url)
-    ws.addEventListener("message", (e) => {
-      conn!.forEachListener(msg.parse(e.data))
-    }, controller)
-    ws.addEventListener("error", (e) => {
-      conn!.forEachListener(new ProviderHandlerError(e))
-    }, controller)
-    ws.addEventListener("close", (e) => {
-      conn!.forEachListener(new ProviderHandlerError(e))
-    }, controller)
-    return new ProviderConnection(ws, () => controller.abort())
-  })
-  conn.addListener(listener)
-  return conn
+export const proxyProvider = proxyProviderFactory()
+
+interface OpenedWsProps {
+  url: string
+  activeWs: Map<string, WebSocket>
+  connectingWs: Map<string, Promise<WebSocket>>
+  retryOptions?: RetryOptions
+  listener: ProviderListener<Event, Event>
 }
 
-function ensureWsOpen(ws: WebSocket): Promise<undefined | Event> {
-  if (ws.readyState === WebSocket.OPEN) {
-    return Promise.resolve(undefined)
-  } else if (ws.readyState === WebSocket.CLOSING || ws.readyState === WebSocket.CLOSED) {
-    return Promise.resolve(new Event("error"))
-  } else {
-    return new Promise<undefined | Event>((resolve) => {
-      const controller = new AbortController()
-      ws.addEventListener("open", () => {
-        controller.abort()
-        resolve(undefined)
-      }, controller)
-      ws.addEventListener("error", (e) => {
-        controller.abort()
-        resolve(e)
-      }, controller)
+function openedWs(
+  { url, activeWs, connectingWs, listener, retryOptions }: OpenedWsProps,
+): Promise<WebSocket> {
+  return retry(() => {
+    const activeWsValue = activeWs.get(url)
+    if (activeWsValue) return Promise.resolve(activeWsValue)
+    const connectingWsValue = connectingWs.get(url)
+    if (connectingWsValue) return connectingWsValue
+    const openedWs = new Promise<WebSocket>((resolve, reject) => {
+      const connectingWsController = new AbortController()
+      const ws = new WebSocket(url)
+      ws.addEventListener(
+        "open",
+        () => {
+          connectingWsController.abort()
+          connectingWs.delete(url)
+          activeWs.set(url, ws)
+          const activeWsController = new AbortController()
+          ws.addEventListener(
+            "message",
+            (e) => listener(msg.parse(e.data)),
+            activeWsController,
+          )
+          ws.addEventListener(
+            "error",
+            (e) => {
+              activeWs.delete(url)
+              listener(new ProviderHandlerError(e))
+            },
+            activeWsController,
+          )
+          ws.addEventListener(
+            "close",
+            (e) => {
+              activeWs.delete(url)
+              activeWsController.abort()
+              if (e.code !== CUSTOM_WS_CLOSE_CODE) {
+                listener(new ProviderHandlerError(e))
+              }
+            },
+            activeWsController,
+          )
+          resolve(ws)
+        },
+        connectingWsController,
+      )
+      ws.addEventListener(
+        "close",
+        (e) => {
+          connectingWsController.abort()
+          connectingWs.delete(url)
+          activeWs.delete(url)
+          reject(e)
+        },
+        connectingWsController,
+      )
     })
-  }
+    connectingWs.set(url, openedWs)
+    return openedWs
+  }, retryOptions)
 }
 
 function closeWs(socket: WebSocket): Promise<undefined | ProviderCloseError<Event>> {
@@ -106,6 +137,6 @@ function closeWs(socket: WebSocket): Promise<undefined | ProviderCloseError<Even
       controller.abort()
       resolve(new ProviderCloseError(e))
     }, controller)
-    socket.close()
+    socket.close(CUSTOM_WS_CLOSE_CODE, "Client normal closure")
   })
 }