feat: improve smoldot integration

deps/smoldot.ts

@@ -1 +1 @@
-export * from "https://deno.land/x/[email protected].3/index-deno.js";
+export * from "https://deno.land/x/[email protected].6/index-deno.js";

rpc/client.ts

@@ -37,7 +37,12 @@ export class Client<
   }
 
   #listener: ProviderListener<SendErrorData, HandlerErrorData> = (e) => {
-    if (e instanceof Error) {
+    if (e instanceof ProviderSendError) {
+      const egressMessageId = e.egressMessage.id;
+      const pendingCall = this.pendingCalls[egressMessageId];
+      pendingCall?.resolve(e);
+      delete this.pendingCalls[egressMessageId];
+    } else if (e instanceof Error) {
       for (const id in this.pendingCalls) {
         const pendingCall = this.pendingCalls[id]!;
         pendingCall.resolve(e);
@@ -52,13 +57,7 @@ export class Client<
       }
     } else if (e.id) {
       const pendingCall = this.pendingCalls[e.id];
-      if (!pendingCall) {
-        console.log({ e });
-        // TODO: pipe error to listeners and message the likely cause,
-        //       a duplicate client.
-        throw new Error();
-      }
-      pendingCall.resolve(e);
+      pendingCall?.resolve(e);
       delete this.pendingCalls[e.id];
       if (this.pendingSubscriptions[e.id]) {
         if (e.error) {
@@ -92,7 +91,7 @@ export class Client<
       delete this.activeSubscriptionByMessageId[message.id];
       waiter.resolve(activeSubscriptionId);
     };
-    this.pendingSubscriptions[message.id] = listener.bind({
+    const listenerBound = listener.bind({
       message,
       stop,
       state: <T>(ctor: new() => T) => {
@@ -103,7 +102,14 @@ export class Client<
         );
       },
     }) as ClientSubscribeListener<SendErrorData, HandlerErrorData>;
-    this.call(message);
+    this.pendingSubscriptions[message.id] = listenerBound;
+    this.call(message)
+      .then((maybeError) => {
+        if (maybeError instanceof Error) {
+          listenerBound(maybeError);
+          stop();
+        }
+      });
     return waiter;
   };
 

rpc/provider/errors.ts

@@ -1,6 +1,11 @@
+import { msg } from "../mod.ts";
+
 export class ProviderSendError<Data> extends Error {
   override readonly name = "ProviderSendError";
-  constructor(override readonly cause: Data) {
+  constructor(
+    override readonly cause: Data,
+    readonly egressMessage: msg.EgressMessage,
+  ) {
     super();
   }
 }

rpc/provider/proxy.ts

@@ -17,9 +17,13 @@ export const proxyProvider: Provider<string, Event, Event, Event> = (url, listen
       (async () => {
         const openError = await ensureWsOpen(conn.inner);
         if (openError) {
-          conn.forEachListener(new ProviderSendError(openError));
-        } else {
+          conn.forEachListener(new ProviderSendError(openError, message));
+          return;
+        }
+        try {
           conn.inner.send(JSON.stringify(message));
+        } catch (error) {
+          listener(new ProviderSendError(error as Event, message));
         }
       })();
     },

rpc/provider/smoldot.test.ts

@@ -1,30 +1,143 @@
-// TODO
 import { deferred } from "../../deps/std/async.ts";
 import { assertExists, assertNotInstanceOf } from "../../deps/std/testing/asserts.ts";
+import { ProviderListener } from "./base.ts";
 import { smoldotProvider } from "./smoldot.ts";
 
-const POLKADOT_CHAIN_SPEC_URL =
-  "https://raw.githubusercontent.com/paritytech/substrate-connect/main/packages/connect/src/connector/specs/polkadot.json";
-
 Deno.test({
   name: "Smoldot Provider",
-  ignore: true,
-  async fn() {
-    const stopped = deferred();
-    const polkadotChainSpec = await (await fetch(POLKADOT_CHAIN_SPEC_URL)).text();
-    const provider = smoldotProvider(polkadotChainSpec, (message) => {
-      assertNotInstanceOf(message, Error);
-      assertExists(message.result);
-      stopped.resolve();
+  sanitizeOps: false,
+  sanitizeResources: false,
+  async fn(t) {
+    await t.step({
+      name: "relay chain connection",
+      async fn() {
+        const relayChainSpec = await (
+          await fetch(
+            "https://raw.githubusercontent.com/paritytech/substrate-connect/main/packages/connect/src/connector/specs/polkadot.json",
+          )
+        )
+          .text();
+        const pendingSubscriptionId = deferred<string>();
+        const initialized = deferred();
+        const unsubscribed = deferred();
+        const checks: ProviderListener<any, any>[] = [
+          // check for chainHead_unstable_follow subscription
+          (message) => {
+            assertNotInstanceOf(message, Error);
+            assertExists(message.result);
+            pendingSubscriptionId.resolve(message.result);
+          },
+          // check for chainHead_unstable_follow initialized event
+          (message) => {
+            assertNotInstanceOf(message, Error);
+            assertExists(message.params?.result);
+            if (message.params?.result.event === "initialized") {
+              initialized.resolve();
+            }
+          },
+          // check for chainHead_unstable_unfollow unsubscribe
+          (message) => {
+            assertNotInstanceOf(message, Error);
+            if (message?.result === null) {
+              unsubscribed.resolve();
+            }
+          },
+        ];
+        const provider = smoldotProvider({ relayChainSpec }, (message) => {
+          if (checks.length > 1) {
+            checks.shift()!(message);
+          } else {
+            checks[0]!(message);
+          }
+        });
+        provider.send({
+          jsonrpc: "2.0",
+          id: provider.nextId(),
+          method: "chainHead_unstable_follow",
+          params: [false],
+        });
+        const subscriptionId = await pendingSubscriptionId;
+        await initialized;
+        provider.send({
+          jsonrpc: "2.0",
+          id: provider.nextId(),
+          method: "chainHead_unstable_unfollow",
+          params: [subscriptionId],
+        });
+        await unsubscribed;
+        const providerRelease = await provider.release();
+        assertNotInstanceOf(providerRelease, Error);
+      },
     });
-    provider.send({
-      jsonrpc: "2.0",
-      id: provider.nextId(),
-      method: "system_health",
-      params: [],
+    await t.step({
+      name: "parachain connection",
+      async fn() {
+        const relayChainSpec = await (
+          await fetch(
+            "https://raw.githubusercontent.com/paritytech/substrate-connect/main/packages/connect/src/connector/specs/westend2.json",
+          )
+        )
+          .text();
+        const parachainSpec = await (
+          await fetch(
+            "https://raw.githubusercontent.com/paritytech/substrate-connect/main/projects/demo/src/assets/westend-westmint.json",
+          )
+        )
+          .text();
+        const pendingSubscriptionId = deferred<string>();
+        const initialized = deferred();
+        const unsubscribed = deferred();
+        const checks: ProviderListener<any, any>[] = [
+          // check for chainHead_unstable_follow subscription
+          (message) => {
+            assertNotInstanceOf(message, Error);
+            assertExists(message.result);
+            pendingSubscriptionId.resolve(message.result);
+          },
+          // check for chainHead_unstable_follow initialized event
+          (message) => {
+            assertNotInstanceOf(message, Error);
+            assertExists(message.params?.result);
+            if (message.params?.result.event === "initialized") {
+              initialized.resolve();
+            }
+          },
+          // check for chainHead_unstable_unfollow unsubscribe
+          (message) => {
+            assertNotInstanceOf(message, Error);
+            if (message?.result === null) {
+              unsubscribed.resolve();
+            }
+          },
+        ];
+        const provider = smoldotProvider(
+          { parachainSpec, relayChainSpec },
+          (message) => {
+            if (checks.length > 1) {
+              checks.shift()!(message);
+            } else {
+              checks[0]!(message);
+            }
+          },
+        );
+        provider.send({
+          jsonrpc: "2.0",
+          id: provider.nextId(),
+          method: "chainHead_unstable_follow",
+          params: [false],
+        });
+        const subscriptionId = await pendingSubscriptionId;
+        await initialized;
+        provider.send({
+          jsonrpc: "2.0",
+          id: provider.nextId(),
+          method: "chainHead_unstable_unfollow",
+          params: [subscriptionId],
+        });
+        await unsubscribed;
+        const providerRelease = await provider.release();
+        assertNotInstanceOf(providerRelease, Error);
+      },
     });
-
-    await stopped;
-    await provider.release();
   },
 });