fix: Avail gas relay decoding issues

core/node/da_clients/src/avail/client.rs

@@ -42,15 +42,39 @@ pub struct AvailClient {
 pub struct BridgeAPIResponse {
     blob_root: Option<H256>,
     bridge_root: Option<H256>,
-    data_root_index: Option<u64>,
+    #[serde(deserialize_with = "deserialize_u256_from_integer")]
+    data_root_index: Option<U256>,
     data_root_proof: Option<Vec<H256>>,
     leaf: Option<H256>,
-    leaf_index: Option<u64>,
+    #[serde(deserialize_with = "deserialize_u256_from_integer")]
+    leaf_index: Option<U256>,
     leaf_proof: Option<Vec<H256>>,
     range_hash: Option<H256>,
     error: Option<String>,
 }
 
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum U256Value {
+    Number(u64),
+    String(String),
+}
+
+fn deserialize_u256_from_integer<'de, D>(deserializer: D) -> Result<Option<U256>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    use serde::de::Error;
+
+    match Option::<U256Value>::deserialize(deserializer)? {
+        Some(U256Value::Number(num)) => Ok(Some(U256::from(num))),
+        Some(U256Value::String(s)) => U256::from_str_radix(s.strip_prefix("0x").unwrap_or(&s), 16)
+            .map(Some)
+            .map_err(|e| D::Error::custom(format!("failed to parse hex string: {}", e))),
+        None => Ok(None),
+    }
+}
+
 #[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 struct MerkleProofInput {
@@ -122,7 +146,7 @@ impl AvailClient {
                 let seed_phrase = secrets
                     .seed_phrase
                     .ok_or_else(|| anyhow::anyhow!("Seed phrase is missing"))?;
-                // these unwraps are safe because we validate in protobuf config
+
                 let sdk_client = RawAvailClient::new(
                     conf.app_id,
                     seed_phrase.0.expose_secret(),
@@ -222,20 +246,28 @@ impl DataAvailabilityClient for AvailClient {
             .await
             .map_err(to_retriable_da_error)?;
 
-        let attestation_data: MerkleProofInput = MerkleProofInput {
-            data_root_proof: bridge_api_data.data_root_proof.unwrap(),
-            leaf_proof: bridge_api_data.leaf_proof.unwrap(),
-            range_hash: bridge_api_data.range_hash.unwrap(),
-            data_root_index: bridge_api_data.data_root_index.unwrap().into(),
-            blob_root: bridge_api_data.blob_root.unwrap(),
-            bridge_root: bridge_api_data.bridge_root.unwrap(),
-            leaf: bridge_api_data.leaf.unwrap(),
-            leaf_index: bridge_api_data.leaf_index.unwrap().into(),
-        };
+        tracing::info!("Bridge API Response: {:?}", bridge_api_data);
 
-        Ok(Some(InclusionData {
-            data: ethabi::encode(&attestation_data.into_tokens()),
-        }))
+        // Check if there's an error in the response
+        if let Some(err) = bridge_api_data.error {
+            tracing::info!(
+                "Bridge API returned error: {:?}. Data might not be available yet.",
+                err
+            );
+            return Ok(None);
+        }
+
+        match bridge_response_to_merkle_proof_input(bridge_api_data) {
+            Some(attestation_data) => Ok(Some(InclusionData {
+                data: ethabi::encode(&attestation_data.into_tokens()),
+            })),
+            None => {
+                tracing::info!(
+                    "Bridge API response missing required fields. Data might not be available yet."
+                );
+                Ok(None)
+            }
+        }
     }
 
     fn clone_boxed(&self) -> Box<dyn DataAvailabilityClient> {
@@ -269,3 +301,18 @@ impl DataAvailabilityClient for AvailClient {
         }
     }
 }
+
+fn bridge_response_to_merkle_proof_input(
+    bridge_api_response: BridgeAPIResponse,
+) -> Option<MerkleProofInput> {
+    Some(MerkleProofInput {
+        data_root_proof: bridge_api_response.data_root_proof?,
+        leaf_proof: bridge_api_response.leaf_proof?,
+        range_hash: bridge_api_response.range_hash?,
+        data_root_index: bridge_api_response.data_root_index?,
+        blob_root: bridge_api_response.blob_root?,
+        bridge_root: bridge_api_response.bridge_root?,
+        leaf: bridge_api_response.leaf?,
+        leaf_index: bridge_api_response.leaf_index?,
+    })
+}

core/node/da_clients/src/avail/sdk.rs

@@ -3,7 +3,7 @@
 
 use std::{fmt::Debug, sync::Arc, time};
 
-use anyhow::Context;
+use anyhow::{bail, Context};
 use backon::{ConstantBuilder, Retryable};
 use bytes::Bytes;
 use jsonrpsee::{
@@ -450,13 +450,26 @@ impl GasRelayClient {
             .post(&submit_url)
             .body(Bytes::from(data))
             .header("Content-Type", "text/plain")
-            .header("Authorization", &self.api_key)
+            .header("Authorization", format!("Bearer {}", self.api_key))
             .send()
-            .await?;
+            .await
+            .context("Failed to submit data to the gas relay")?;
 
-        let submit_response = submit_response
-            .json::<GasRelayAPISubmissionResponse>()
-            .await?;
+        let response_bytes = submit_response
+            .bytes()
+            .await
+            .context("Failed to read response body")?;
+
+        let submit_response =
+            match serde_json::from_slice::<GasRelayAPISubmissionResponse>(&response_bytes) {
+                Ok(response) => response,
+                Err(_) => {
+                    bail!(
+                        "Unexpected response from gas relay: {:?}",
+                        String::from_utf8_lossy(&response_bytes).as_ref()
+                    )
+                }
+            };
 
         let status_url = format!(
             "{}/user/get_submission_info?submission_id={}",
@@ -467,7 +480,7 @@ impl GasRelayClient {
         let status_response = (|| async {
             self.api_client
                 .get(&status_url)
-                .header("Authorization", &self.api_key)
+                .header("Authorization", format!("Bearer {}", self.api_key))
                 .send()
                 .await
         })
@@ -478,7 +491,22 @@ impl GasRelayClient {
         )
         .await?;
 
-        let status_response = status_response.json::<GasRelayAPIStatusResponse>().await?;
+        let status_response_bytes = status_response
+            .bytes()
+            .await
+            .context("Failed to read response body")?;
+
+        let status_response =
+            match serde_json::from_slice::<GasRelayAPIStatusResponse>(&status_response_bytes) {
+                Ok(response) => response,
+                Err(_) => {
+                    bail!(
+                        "Unexpected status response from gas relay: {:?}",
+                        String::from_utf8_lossy(&status_response_bytes).as_ref()
+                    )
+                }
+            };
+
         let (block_hash, extrinsic_index) = (
             status_response.submission.block_hash.ok_or_else(|| {
                 anyhow::anyhow!("Block hash not found in the response from the gas relay")

core/node/da_dispatcher/src/da_dispatcher.rs

@@ -255,7 +255,12 @@ where
                 retries += 1;
                 let sleep_duration = Duration::from_secs(backoff_secs)
                     .mul_f32(rand::thread_rng().gen_range(0.8..1.2));
-                tracing::warn!(%err, "Failed DA dispatch request {retries}/{max_retries} for batch {batch_number}, retrying in {} milliseconds.", sleep_duration.as_millis());
+                tracing::warn!(
+                    %err,
+                    "Failed DA dispatch request {retries}/{} for batch {batch_number}, retrying in {} milliseconds.",
+                    max_retries+1,
+                    sleep_duration.as_millis()
+                );
                 tokio::time::sleep(sleep_duration).await;
 
                 backoff_secs = (backoff_secs * 2).min(128); // cap the back-off at 128 seconds