Improve stability with avail DA (keep-starknet-strange#1257)

Co-authored-by: Leouarz <ghalielouarzazi@hotmail.com>
Co-authored-by: 0xevolve <Artevolve@yahoo.com>
Co-authored-by: lanaivina <31368580+lana-shanghai@users.noreply.github.com>

Author: 4 people

CHANGELOG.md

@@ -12,6 +12,8 @@
 - chore(data-availability): update avail-subxt to version 0.4.0
 - fix(ci): setup should fetch files from local config
 - chore: deprecate `madara-app` and `madara-dev-explorer` modules
+- chore(data-availability-avail): implement fire and forget, and add ws
+  reconnection logic
 
 ## v0.5.0
 

Cargo.lock

@@ -45,7 +45,7 @@ starknet_api = { workspace = true, default-features = true }
 ethers = "2.0.7"
 
 # Avail subxt dependency
-avail-subxt = { git = "https://github.com/availproject/avail", version = "0.4.0", tag = "v1.8.0.0-rc1" }
+avail-subxt = { git = "https://github.com/availproject/avail", version = "0.4.0", tag = "v1.8.0.0" }
 sp-keyring = { workspace = true }
 subxt = "0.29"
 

crates/client/data-availability/Cargo.toml

@@ -1,16 +1,18 @@
 pub mod config;
 
-use anyhow::Result;
+use std::sync::Arc;
+
+use anyhow::{anyhow, Result};
 use async_trait::async_trait;
 use avail_subxt::api::runtime_types::avail_core::AppId;
 use avail_subxt::api::runtime_types::bounded_collections::bounded_vec::BoundedVec;
-use avail_subxt::api::runtime_types::da_control::pallet::Call as DaCall;
-use avail_subxt::avail::{AppUncheckedExtrinsic, Client as AvailSubxtClient};
+use avail_subxt::avail::Client as AvailSubxtClient;
 use avail_subxt::primitives::AvailExtrinsicParams;
-use avail_subxt::{api as AvailApi, build_client, AvailConfig, Call};
+use avail_subxt::{api as AvailApi, build_client, AvailConfig};
 use ethers::types::{I256, U256};
-use sp_core::H256;
+use futures::lock::Mutex;
 use subxt::ext::sp_core::sr25519::Pair;
+use subxt::OnlineClient;
 
 use crate::utils::get_bytes_from_state_diff;
 use crate::{DaClient, DaMode};
@@ -19,22 +21,55 @@ type AvailPairSigner = subxt::tx::PairSigner<AvailConfig, Pair>;
 
 #[derive(Clone)]
 pub struct AvailClient {
-    ws_client: AvailSubxtClient,
+    ws_client: Arc<Mutex<SubxtClient>>,
     app_id: AppId,
     signer: AvailPairSigner,
     mode: DaMode,
 }
 
+pub struct SubxtClient {
+    client: AvailSubxtClient,
+    config: config::AvailConfig,
+}
+
+pub fn try_build_avail_subxt(conf: &config::AvailConfig) -> Result<OnlineClient<AvailConfig>> {
+    let client =
+        futures::executor::block_on(async { build_client(conf.ws_provider.as_str(), conf.validate_codegen).await })
+            .map_err(|e| anyhow::anyhow!("DA Layer error: could not initialize ws endpoint {e}"))?;
+
+    Ok(client)
+}
+
+impl SubxtClient {
+    pub async fn restart(&mut self) -> Result<(), anyhow::Error> {
+        self.client = match build_client(self.config.ws_provider.as_str(), self.config.validate_codegen).await {
+            Ok(i) => i,
+            Err(e) => return Err(anyhow!("DA Layer error: could not restart ws endpoint {e}")),
+        };
+
+        Ok(())
+    }
+
+    pub fn client(&self) -> &OnlineClient<AvailConfig> {
+        &self.client
+    }
+}
+
+impl TryFrom<config::AvailConfig> for SubxtClient {
+    type Error = anyhow::Error;
+
+    fn try_from(conf: config::AvailConfig) -> Result<Self, Self::Error> {
+        Ok(Self { client: try_build_avail_subxt(&conf)?, config: conf })
+    }
+}
+
 #[async_trait]
 impl DaClient for AvailClient {
     async fn publish_state_diff(&self, state_diff: Vec<U256>) -> Result<()> {
         let bytes = get_bytes_from_state_diff(&state_diff);
         let bytes = BoundedVec(bytes);
+        self.publish_data(&bytes).await?;
 
-        let submitted_block_hash = self.publish_data(&bytes).await?;
-
-        // This theoritically do not have to be put here since we wait for finalization before
-        self.verify_bytes_inclusion(submitted_block_hash, &bytes).await?;
         Ok(())
     }
 
@@ -50,38 +85,22 @@ impl DaClient for AvailClient {
 }
 
 impl AvailClient {
-    async fn publish_data(&self, bytes: &BoundedVec<u8>) -> Result<H256> {
+    async fn publish_data(&self, bytes: &BoundedVec<u8>) -> Result<()> {
+        let mut ws_client = self.ws_client.lock().await;
+
         let data_transfer = AvailApi::tx().data_availability().submit_data(bytes.clone());
         let extrinsic_params = AvailExtrinsicParams::new_with_app_id(self.app_id);
-        let events = self
-            .ws_client
-            .tx()
-            .sign_and_submit_then_watch(&data_transfer, &self.signer, extrinsic_params)
-            .await?
-            .wait_for_finalized_success()
-            .await?;
-
-        Ok(events.block_hash())
-    }
 
-    async fn verify_bytes_inclusion(&self, block_hash: H256, bytes: &BoundedVec<u8>) -> Result<()> {
-        let submitted_block = self
-            .ws_client
-            .rpc()
-            .block(Some(block_hash))
-            .await?
-            .ok_or(anyhow::anyhow!("Invalid hash, block not found"))?;
-
-        submitted_block
-            .block
-            .extrinsics
-            .into_iter()
-            .filter_map(|chain_block_ext| AppUncheckedExtrinsic::try_from(chain_block_ext).map(|ext| ext.function).ok())
-            .find(|call| match call {
-                Call::DataAvailability(DaCall::submit_data { data }) => data == bytes,
-                _ => false,
-            })
-            .ok_or(anyhow::anyhow!("Bytes not found in specified block"))?;
+        match ws_client.client().tx().sign_and_submit(&data_transfer, &self.signer, extrinsic_params).await {
+            Ok(i) => i,
+            Err(e) => {
+                if e.to_string().contains("restart required") {
+                    ws_client.restart().await;
+                }
+
+                return Err(anyhow!("DA Layer error : failed due to closed websocket connection {e}"));
+            }
+        };
 
         Ok(())
     }
@@ -95,11 +114,12 @@ impl TryFrom<config::AvailConfig> for AvailClient {
 
         let app_id = AppId(conf.app_id);
 
-        let ws_client =
-            futures::executor::block_on(async { build_client(conf.ws_provider.as_str(), conf.validate_codegen).await })
-                .map_err(|e| anyhow::anyhow!("DA Layer error: could not initialize ws endpoint {e}"))?;
-
-        Ok(Self { ws_client, app_id, signer, mode: conf.mode })
+        Ok(Self {
+            ws_client: Arc::new(Mutex::new(SubxtClient::try_from(conf.clone())?)),
+            app_id,
+            signer,
+            mode: conf.mode,
+        })
     }
 }