fix: multiple runs with proxy providers (#581)

deps/std/http.ts

@@ -0,0 +1 @@
+export * from "https://deno.land/[email protected]/http/mod.ts"

fluent/rpc_runes.ts

@@ -12,12 +12,12 @@ class RunRpcClient<DV, SED, HED, CED> extends Run<rpc.Client<DV, SED, HED, CED>,
   }
 
   client?: rpc.Client<DV, SED, HED, CED>
-  _evaluate(): Promise<rpc.Client<DV, SED, HED, CED>> {
-    return Promise.resolve(this.client ??= new rpc.Client(this.provider, this.discoveryValue))
+  async _evaluate(): Promise<rpc.Client<DV, SED, HED, CED>> {
+    return this.client ??= new rpc.Client(this.provider, this.discoveryValue)
   }
 
-  override cleanup(): void {
-    this.client?.discard()
+  override async cleanup(): Promise<void> {
+    await this.client?.discard()
     super.cleanup()
   }
 }

main.ts

@@ -1,5 +1,5 @@
 import * as flags from "./deps/std/flags.ts"
-import { serve } from "./deps/std/http/server.ts"
+import { serve } from "./deps/std/http.ts"
 import { PolkadotDevProvider, WssProvider, ZombienetProvider } from "./providers/frame/mod.ts"
 import { handler } from "./server/local.ts"
 import { Env } from "./server/mod.ts"

patterns/ink/InkRune.ts

@@ -1,4 +1,4 @@
-import { equals } from "../../deps/std/bytes/mod.ts"
+import { equals } from "../../deps/std/bytes.ts"
 import { Chain, ClientRune, CodecRune, ExtrinsicRune, state } from "../../fluent/mod.ts"
 import { Event } from "../../primitives/mod.ts"
 import { Rune, RunicArgs, ValueRune } from "../../rune/mod.ts"

providers/frame/FrameProxyProvider.ts

@@ -1,11 +1,12 @@
 import { File } from "../../codegen/frame/mod.ts"
+import { deferred } from "../../deps/std/async.ts"
 import { Client, proxyProvider } from "../../rpc/mod.ts"
 import { PathInfo } from "../../server/mod.ts"
 import { fromPathInfo } from "../../server/PathInfo.ts"
 import { FrameProvider } from "./FrameProvider.ts"
 
 export abstract class FrameProxyProvider extends FrameProvider {
-  override handle(request: Request, pathInfo: PathInfo): Promise<Response> {
+  override async handle(request: Request, pathInfo: PathInfo): Promise<Response> {
     if (
       pathInfo.vRuntime && pathInfo.vRuntime !== "latest" && !pathInfo.filePath
       && request.headers.get("upgrade") === "websocket"
@@ -19,33 +20,30 @@ export abstract class FrameProxyProvider extends FrameProvider {
     const url = await this.dynamicUrl(pathInfo)
     const server = new WebSocket(url)
     const { socket: client, response } = Deno.upgradeWebSocket(request)
-    server.addEventListener("close", () => {
-      try {
-        client.close()
-      } catch {}
-    })
-    server.addEventListener("message", (event) => {
-      try {
-        client.send(event.data)
-      } catch {
-        client.close()
-        server.close()
-      }
-    })
-    client.addEventListener("close", async () => {
-      try {
-        server.close()
-      } catch {}
-    })
-    client.addEventListener("message", async (event) => {
-      try {
-        server.send(event.data)
-      } catch {
-        client.close()
-        server.close()
-      }
-    })
+    setup(client, server)
+    setup(server, client)
     return response
+
+    function setup(a: WebSocket, b: WebSocket) {
+      const ready = deferred()
+      b.addEventListener("open", () => {
+        ready.resolve()
+      })
+      a.addEventListener("close", async () => {
+        try {
+          b.close()
+        } catch {}
+      })
+      a.addEventListener("message", async (event) => {
+        try {
+          await ready
+          b.send(event.data)
+        } catch {
+          a.close()
+          b.close()
+        }
+      })
+    }
   }
 
   abstract dynamicUrl(pathInfo: PathInfo): Promise<string>
@@ -73,11 +71,11 @@ export abstract class FrameProxyProvider extends FrameProvider {
       import * as C from "./capi.ts"
       import type { Chain } from "./mod.ts"
 
-      export const client = C.rpcClient(C.rpc.proxyProvider, "${url}")["_asCodegen"]<Chain>()
+      export const discoveryValue = "${url}"
 
-      export const rawClient = new C.rpc.Client(C.rpc.proxyProvider, "${url}")
+      export const client = C.rpcClient(C.rpc.proxyProvider, discoveryValue)["_asCodegen"]<Chain>()
 
-      export const discoveryValue = "${url}"
+      export const rawClient = new C.rpc.Client(C.rpc.proxyProvider, discoveryValue)
     `)
   }
 }

rpc/provider/errors.ts

@@ -11,7 +11,7 @@ export class ProviderSendError<Data> extends Error {
 }
 
 export class ProviderHandlerError<Data> extends Error {
-  override readonly name = "ProviderInternalError"
+  override readonly name = "ProviderHandlerError"
   constructor(override readonly cause: Data) {
     super()
   }

rpc/provider/proxy.ts

@@ -66,7 +66,9 @@ function connection(
     ws.addEventListener("close", (e) => {
       conn.forEachListener(new ProviderHandlerError(e))
     }, controller)
-    return new ProviderConnection(ws, () => controller.abort())
+    return new ProviderConnection(ws, () => {
+      controller.abort()
+    })
   })
   conn.addListener(listener)
   return conn

server/factories.ts

@@ -1,6 +1,6 @@
 import { escapeHtml } from "../deps/escape.ts"
 import * as shiki from "../deps/shiki.ts"
-import { Status } from "../deps/std/http/http_status.ts"
+import { Status } from "../deps/std/http.ts"
 import { CacheBase } from "../util/cache/base.ts"
 
 shiki.setCDN("https://unpkg.com/shiki/")

server/local.ts

@@ -18,10 +18,12 @@ export function handler(env: Env) {
       }
       const provider = env.providers[generatorId]?.[providerId]
       if (provider) {
-        return await provider.handle(request, pathInfo).catch((e) => {
+        try {
+          return await provider.handle(request, pathInfo)
+        } catch (e) {
           if (e instanceof Response) return e
           return f.serverError(Deno.inspect(e))
-        })
+        }
       }
     }
     for (const dir of staticDirs) {