feat: implement shutdown method for helios-ts (#778)

- Implement a proper cleanup so Helios instance can be safely garbage collected
- Bonus: return Helios reference from on() and removeListener() methods as per EIP-1193 spec

Author: karen-sarkisyan

Cargo.lock

@@ -14,6 +14,7 @@ wasm-bindgen-futures = "0.4.33"
 wasm-bindgen-test = "0.3.0"
 serde-wasm-bindgen = "0.6.5"
 console_error_panic_hook = "0.1.7"
+futures-util = "0.3"
 
 eyre.workspace = true
 alloy.workspace = true

helios-ts/Cargo.toml

@@ -127,6 +127,19 @@ const blockNumber = await client.getBlockNumber();
 console.log('Latest block number:', blockNumber);
 ```
 
+## Cleanup
+
+If Helios is used as a transport for `ethers` or `viem`, calling `EthersProvider.destroy()` will NOT destroy underlying Helios instance nor will stop the background tasks.
+
+For proper cleanup, call `shutdown()` (or its alias `destroy()`) on your Helios provider instance to properly release resources:
+
+```typescript
+// Clean up when done
+await heliosProvider.shutdown();
+```
+
+This unsubscribes all active subscriptions, aborts background tasks, removes event listeners, and frees WASM memory. The method is idempotent—calling it multiple times is safe. After dereferencing Helios provider instance will be garbage collected.
+
 ## Documentation
 
 See [helios github repo](https://github.com/a16z/helios/) for more details.

helios-ts/README.md

@@ -81,6 +81,8 @@ export class HeliosProvider {
   #client;
   #chainId;
   #eventEmitter;
+  #closed = false;
+  #subscriptionIds: Set<string> = new Set();
 
   private constructor(config: Config, kind: NetworkKind) {
     const executionRpc = config.executionRpc;
@@ -116,8 +118,10 @@ export class HeliosProvider {
   }
 
   #setHeliosEvents() {
+    // Capture only the emitter reference, not `this`
+    const emitter = this.#eventEmitter;
     this.#client.set_helios_events((event: string, data: any) => {
-      this.#eventEmitter.emit(event, data);
+      emitter.emit(event, data);
     });
   }
 
@@ -175,6 +179,9 @@ export class HeliosProvider {
    * ```
    */
   async request(req: Request): Promise<any> {
+    if (this.#closed) {
+      throw new Error("Provider has been shut down");
+    }
     try {
       return await this.#req(req);
     } catch (err) {
@@ -292,7 +299,9 @@ export class HeliosProvider {
         return this.#handleSubscribe(req);
       }
       case "eth_unsubscribe": {
-        return this.#client.unsubscribe(req.params[0]);
+        const id = req.params[0];
+        this.#subscriptionIds.delete(id);
+        return this.#client.unsubscribe(id);
       }
       case "helios_getCurrentCheckpoint": {
         return this.#client.get_current_checkpoint();
@@ -305,18 +314,21 @@ export class HeliosProvider {
 
   async #handleSubscribe(req: Request) {
     try {
-      let id = uuidv4();
-      await this.#client.subscribe(req.params[0], id, (data: any, id: string) => {
-        let result = data instanceof Map ? mapToObj(data) : data;
-        let payload = {
+      const id = uuidv4();
+      // Capture only the emitter reference, not `this`
+      const emitter = this.#eventEmitter;
+      await this.#client.subscribe(req.params[0], id, (data: any, subId: string) => {
+        const result = data instanceof Map ? mapToObj(data) : data;
+        const payload = {
           type: 'eth_subscription',
           data: {
-            subscription: id,
+            subscription: subId,
             result,
           },
         };
-        this.#eventEmitter.emit("message", payload);
+        emitter.emit("message", payload);
       });
+      this.#subscriptionIds.add(id);
       return id;
     } catch (err) {
       throw new Error(err.toString());
@@ -353,8 +365,9 @@ export class HeliosProvider {
   on(
     eventName: string,
     handler: (data: any) => void
-  ): void {
+  ): this {
     this.#eventEmitter.on(eventName, handler);
+    return this;
   }
 
   /**
@@ -381,8 +394,59 @@ export class HeliosProvider {
   removeListener(
     eventName: string,
     handler: (data: any) => void
-  ): void {
+  ): this {
     this.#eventEmitter.off(eventName, handler);
+    return this;
+  }
+
+  /**
+   * Shuts down the provider and releases all resources.
+   * 
+   * @returns A promise that resolves when the provider has been shut down
+   * 
+   * @remarks
+   * After shutdown:
+   * - All future `request()` calls will reject with an error
+   * - All active subscriptions are unsubscribed
+   * - All event listeners are removed
+   * - Background tasks are stopped
+   * 
+   * The provider instance will be garbage collected after the user drops all references.
+   * 
+   * @example
+   * ```typescript
+   * const provider = await createHeliosProvider(config, "ethereum");
+   * 
+   * // ... use the provider ...
+   * 
+   * // Clean up when done
+   * await provider.shutdown();
+   * ```
+   */
+  async shutdown(): Promise<void> {
+    if (this.#closed) {
+      return;
+    }
+    this.#closed = true;
+
+    for (const id of this.#subscriptionIds) {
+      try {
+        this.#client.unsubscribe(id);
+      } catch {
+        // Ignore errors during cleanup
+      }
+    }
+    this.#subscriptionIds.clear();
+    await this.#client.shutdown();
+    this.#eventEmitter.removeAllListeners();
+    this.#client.free();
+  }
+  
+  /**
+   * This method is equivalent to `shutdown()`
+   */
+  async destroy(): Promise<void> {
+    await this.shutdown();
   }
 }
 

helios-ts/lib.ts

@@ -9,6 +9,7 @@ use alloy::hex::{self, FromHex};
 use alloy::primitives::{Address, B256, U256};
 use alloy::rpc::types::{state::StateOverride, Filter, TransactionRequest};
 use eyre::Result;
+use futures_util::future::{AbortHandle, Abortable};
 use url::Url;
 use wasm_bindgen::prelude::*;
 use web_sys::js_sys::Function;
@@ -67,6 +68,7 @@ pub struct EthereumClient {
     chain_id: u64,
     active_subscriptions: HashMap<String, Subscription<Ethereum>>,
     event_handler: Option<Function>,
+    events_abort_handle: Option<AbortHandle>,
 }
 
 #[wasm_bindgen]
@@ -141,6 +143,7 @@ impl EthereumClient {
             chain_id,
             active_subscriptions: HashMap::new(),
             event_handler: None,
+            events_abort_handle: None,
         })
     }
 
@@ -154,9 +157,7 @@ impl EthereumClient {
         let sub_type: SubscriptionType = serde_wasm_bindgen::from_value(sub_type)?;
         let rx = map_err(self.inner.subscribe(sub_type).await)?;
 
-        let subscription = Subscription::<Ethereum>::new(id.clone());
-
-        subscription.listen(rx, callback).await;
+        let subscription = Subscription::<Ethereum>::spawn_listener(id.clone(), rx, callback);
         self.active_subscriptions.insert(id, subscription);
 
         Ok(true)
@@ -453,7 +454,10 @@ impl EthereumClient {
 
         let mut rx = map_err(self.inner.new_checkpoints_recv())?;
 
-        wasm_bindgen_futures::spawn_local(async move {
+        let (abort_handle, abort_registration) = AbortHandle::new_pair();
+        self.events_abort_handle = Some(abort_handle);
+
+        let future = async move {
             loop {
                 if rx.changed().await.is_err() {
                     break;
@@ -470,6 +474,10 @@ impl EthereumClient {
                     }
                 }
             }
+        };
+
+        wasm_bindgen_futures::spawn_local(async move {
+            let _ = Abortable::new(future, abort_registration).await;
         });
 
         Ok(())
@@ -482,4 +490,19 @@ impl EthereumClient {
             &checkpoint.map(|c| format!("0x{}", hex::encode(c))),
         )?)
     }
+
+    #[wasm_bindgen]
+    pub async fn shutdown(&mut self) {
+        if let Some(abort_handle) = self.events_abort_handle.take() {
+            abort_handle.abort();
+        }
+
+        self.event_handler = None;
+
+        for (_, subscription) in self.active_subscriptions.drain() {
+            subscription.abort();
+        }
+
+        self.inner.shutdown().await;
+    }
 }

helios-ts/src/ethereum.rs

@@ -76,9 +76,7 @@ impl LineaClient {
         let sub_type: SubscriptionType = serde_wasm_bindgen::from_value(sub_type)?;
         let rx = map_err(self.inner.subscribe(sub_type).await)?;
 
-        let subscription = Subscription::<Linea>::new(id.clone());
-
-        subscription.listen(rx, callback).await;
+        let subscription = Subscription::<Linea>::spawn_listener(id.clone(), rx, callback);
         self.active_subscriptions.insert(id, subscription);
 
         Ok(true)
@@ -378,4 +376,13 @@ impl LineaClient {
         // Linea does not support checkpoints
         Err(JsError::new("Linea does not support checkpoints"))
     }
+
+    #[wasm_bindgen]
+    pub async fn shutdown(&mut self) {
+        for (_, subscription) in self.active_subscriptions.drain() {
+            subscription.abort();
+        }
+
+        self.inner.shutdown().await;
+    }
 }

helios-ts/src/linea.rs

@@ -94,9 +94,7 @@ impl OpStackClient {
         let sub_type: SubscriptionType = serde_wasm_bindgen::from_value(sub_type)?;
         let rx = map_err(self.inner.subscribe(sub_type).await)?;
 
-        let subscription = Subscription::<OpStack>::new(id.clone());
-
-        subscription.listen(rx, callback).await;
+        let subscription = Subscription::<OpStack>::spawn_listener(id.clone(), rx, callback);
         self.active_subscriptions.insert(id, subscription);
 
         Ok(true)
@@ -396,4 +394,13 @@ impl OpStackClient {
         // OP Stack does not support checkpoints
         Err(JsError::new("OP Stack does not support checkpoints"))
     }
+
+    #[wasm_bindgen]
+    pub async fn shutdown(&mut self) {
+        for (_, subscription) in self.active_subscriptions.drain() {
+            subscription.abort();
+        }
+
+        self.inner.shutdown().await;
+    }
 }

helios-ts/src/opstack.rs

@@ -1,47 +1,51 @@
-use std::cell::Cell;
 use std::marker::PhantomData;
-use std::rc::Rc;
 use wasm_bindgen::prelude::*;
 use web_sys::js_sys::Function;
 
+use futures_util::future::{AbortHandle, Abortable};
+
 use helios_common::network_spec::NetworkSpec;
 use helios_common::types::SubEventRx;
 
 pub struct Subscription<N: NetworkSpec> {
-    id: String,
-    active: Rc<Cell<bool>>,
+    abort_handle: AbortHandle,
     _phantom: PhantomData<N>,
 }
 
 impl<N: NetworkSpec> Subscription<N> {
-    pub fn new(id: String) -> Self {
+    pub fn new(abort_handle: AbortHandle) -> Self {
         Self {
-            id,
-            active: Rc::new(Cell::new(true)),
+            abort_handle,
             _phantom: PhantomData,
         }
     }
 
-    pub async fn listen(&self, mut rx: SubEventRx<N>, callback: Function) {
-        let id = self.id.clone();
-        let active = self.active.clone();
+    pub fn spawn_listener(id: String, mut rx: SubEventRx<N>, callback: Function) -> Self {
+        let (abort_handle, abort_registration) = AbortHandle::new_pair();
 
-        wasm_bindgen_futures::spawn_local(async move {
+        let listener_id = id.clone();
+        let future = async move {
             while let Ok(msg) = rx.recv().await {
-                if !active.get() {
-                    break;
-                }
-
                 if let Ok(data) = serde_wasm_bindgen::to_value(&msg) {
-                    let _ = callback.call2(&JsValue::NULL, &data, &JsValue::from_str(&id));
+                    let _ = callback.call2(&JsValue::NULL, &data, &JsValue::from_str(&listener_id));
                 }
             }
+        };
+
+        wasm_bindgen_futures::spawn_local(async move {
+            let _ = Abortable::new(future, abort_registration).await;
         });
+
+        Self::new(abort_handle)
+    }
+
+    pub fn abort(&self) {
+        self.abort_handle.abort();
     }
 }
 
 impl<N: NetworkSpec> Drop for Subscription<N> {
     fn drop(&mut self) {
-        self.active.set(false);
+        self.abort_handle.abort();
     }
 }