Add publish with retry method

bindings/wasm/Cargo.toml

@@ -18,10 +18,9 @@ crate-type = ["cdylib", "rlib"]
 console_error_panic_hook = { version = "0.1" }
 futures = { version = "0.3" }
 js-sys = { version = "0.3" }
-reqwest = { version = "=0.11.4" } # pin version to 0.11.4: https://github.com/iotaledger/identity.rs/issues/438
+reqwest = { version = "0.11.6" }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", default-features = false }
-tokio = { version = "*", default-features = false, features = ["rt"] } # enable "rt" feature to fix build: https://github.com/iotaledger/identity.rs/issues/403
 wasm-bindgen = { version = "0.2", features = ["serde-serialize"] }
 wasm-bindgen-futures = { version = "0.4", default-features = false }
 
@@ -37,6 +36,7 @@ wasm-bindgen-test = { version = "0.3" }
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 chrono = { version = "0.4", features = ["wasmbind"] }
 getrandom = { version = "0.2", features = ["js"] }
+parking_lot = { version = "0.11", features = ["wasm-bindgen"] }
 
 [package.metadata.wasm-pack.profile.release]
 wasm-opt = true

bindings/wasm/docs/api-reference.md

@@ -48,6 +48,8 @@
 <dd></dd>
 <dt><a href="#KeyType">KeyType</a></dt>
 <dd></dd>
+<dt><a href="#Digest">Digest</a></dt>
+<dd></dd>
 </dl>
 
 ## Functions
@@ -70,6 +72,7 @@
         * [.publishDocument(document)](#Client+publishDocument) ⇒ <code>Promise.&lt;any&gt;</code>
         * [.publishDiff(message_id, diff)](#Client+publishDiff) ⇒ <code>Promise.&lt;any&gt;</code>
         * [.publishJSON(index, data)](#Client+publishJSON) ⇒ <code>Promise.&lt;any&gt;</code>
+        * [.publishJsonWithRetry(index, data, interval, max_attempts)](#Client+publishJsonWithRetry) ⇒ <code>Promise.&lt;any&gt;</code>
         * [.resolve(did)](#Client+resolve) ⇒ <code>Promise.&lt;any&gt;</code>
         * [.resolveHistory(did)](#Client+resolveHistory) ⇒ <code>Promise.&lt;any&gt;</code>
         * [.resolveDiffHistory(document)](#Client+resolveDiffHistory) ⇒ <code>Promise.&lt;any&gt;</code>
@@ -125,6 +128,22 @@ Publishes arbitrary JSON data to the specified index on the Tangle.
 | index | <code>string</code> | 
 | data | <code>any</code> | 
 
+<a name="Client+publishJsonWithRetry"></a>
+
+### client.publishJsonWithRetry(index, data, interval, max_attempts) ⇒ <code>Promise.&lt;any&gt;</code>
+Publishes arbitrary JSON data to the specified index on the Tangle.
+Retries (promotes or reattaches) the message until it’s included (referenced by a milestone).
+Default interval is 5 seconds and max attempts is 40.
+
+**Kind**: instance method of [<code>Client</code>](#Client)  
+
+| Param | Type |
+| --- | --- |
+| index | <code>string</code> | 
+| data | <code>any</code> | 
+| interval | <code>number</code> \| <code>undefined</code> | 
+| max_attempts | <code>number</code> \| <code>undefined</code> | 
+
 <a name="Client+resolve"></a>
 
 ### client.resolve(did) ⇒ <code>Promise.&lt;any&gt;</code>
@@ -1781,6 +1800,10 @@ Deserializes a `VerificationMethod` object from a JSON object.
 
 ## KeyType
 **Kind**: global variable  
+<a name="Digest"></a>
+
+## Digest
+**Kind**: global variable  
 <a name="start"></a>
 
 ## start()

bindings/wasm/src/tangle/client.rs

@@ -124,6 +124,37 @@ impl Client {
     Ok(promise)
   }
 
+  /// Publishes arbitrary JSON data to the specified index on the Tangle.
+  /// Retries (promotes or reattaches) the message until it’s included (referenced by a milestone).
+  /// Default interval is 5 seconds and max attempts is 40.
+  #[wasm_bindgen(js_name = publishJsonWithRetry)]
+  pub fn publish_json_with_retry(
+    &self,
+    index: &str,
+    data: &JsValue,
+    interval: Option<u32>,
+    max_attempts: Option<u32>,
+  ) -> Result<Promise> {
+    let client: Rc<IotaClient> = self.client.clone();
+
+    let index = index.to_owned();
+    let value: serde_json::Value = data.into_serde().wasm_result()?;
+    let promise: Promise = future_to_promise(async move {
+      client
+        .publish_json_with_retry(
+          &index,
+          &value,
+          interval.map(|interval| interval as u64),
+          max_attempts.map(|max_attempts| max_attempts as u64),
+        )
+        .await
+        .wasm_result()
+        .and_then(|receipt| JsValue::from_serde(&receipt).wasm_result())
+    });
+
+    Ok(promise)
+  }
+
   #[wasm_bindgen]
   pub fn resolve(&self, did: &str) -> Result<Promise> {
     #[derive(Serialize)]

identity-iota/Cargo.toml

@@ -12,6 +12,7 @@ description = "An IOTA Tangle intergration for the identity-rs library."
 
 [dependencies]
 async-trait = { version = "0.1", default-features = false }
+bee-rest-api = { version = "0.1.3", default-features = false }
 brotli = { version = "3.3", default-features = false, features = ["std"] }
 dashmap = { version = "4.0" }
 form_urlencoded = { version = "1.0" }
@@ -29,7 +30,7 @@ thiserror = { version = "1.0", default-features = false }
 
 [dependencies.iota-client]
 git = "https://github.com/iotaledger/iota.rs"
-rev = "e8c050a749a2e7c13633e97b3372b38388f48c37"
+rev = "c48db779d2162b9a3037ee63c4d1e267d04633eb"
 default-features = false
 
 [dependencies.iota-crypto]

identity-iota/src/error.rs

@@ -57,4 +57,6 @@ pub enum Error {
   CompressionError,
   #[error("invalid message flags")]
   InvalidMessageFlags,
+  #[error("Unable to Check Message Inclusion in the Tangle")]
+  MessageInclusionNotChecked,
 }

identity-iota/src/tangle/client.rs

@@ -1,9 +1,11 @@
 // Copyright 2020-2021 IOTA Stiftung
 // SPDX-License-Identifier: Apache-2.0
 
+use bee_rest_api::types::dtos::LedgerInclusionStateDto;
 use futures::stream::FuturesUnordered;
 use futures::stream::TryStreamExt;
 use iota_client::Client as IotaClient;
+use iota_client::Error as IotaClientError;
 
 use identity_core::convert::ToJson;
 
@@ -77,14 +79,22 @@ impl Client {
   }
 
   /// Publishes an [`IotaDocument`] to the Tangle.
+  /// This method calls `publish_json_with_retry` with its default `interval` and `max_attempts` values for increasing
+  /// the probability that the message will be referenced by a milestone.
   pub async fn publish_document(&self, document: &IotaDocument) -> Result<Receipt> {
-    self.publish_json(document.integration_index(), document).await
+    self
+      .publish_json_with_retry(document.integration_index(), document, None, None)
+      .await
   }
 
-  /// Publishes a [`DocumentDiff`] to the Tangle to form part of the diff chain for the integration
+  /// Publishes a [`DocumentDiff`] to the Tangle to form part of the diff chain for the integration.
   /// chain message specified by the given [`MessageId`].
+  /// This method calls `publish_json_with_retry` with its default `interval` and `max_attempts` values for increasing
+  /// the probability that the message will be referenced by a milestone.
   pub async fn publish_diff(&self, message_id: &MessageId, diff: &DocumentDiff) -> Result<Receipt> {
-    self.publish_json(&IotaDocument::diff_index(message_id)?, diff).await
+    self
+      .publish_json_with_retry(&IotaDocument::diff_index(message_id)?, diff, None, None)
+      .await
   }
 
   /// Compresses and publishes arbitrary JSON data to the specified index on the Tangle.
@@ -101,6 +111,39 @@ impl Client {
       .map(|message| Receipt::new(self.network.clone(), message))
   }
 
+  /// Publishes arbitrary JSON data to the specified index on the Tangle.
+  /// Retries (promotes or reattaches) the message until it’s included (referenced by a milestone).
+  /// Default interval is 5 seconds and max attempts is 40.
+  pub async fn publish_json_with_retry<T: ToJson>(
+    &self,
+    index: &str,
+    data: &T,
+    interval: Option<u64>,
+    max_attempts: Option<u64>,
+  ) -> Result<Receipt> {
+    let receipt: Receipt = self.publish_json(index, data).await?;
+    let reattached_messages: Vec<(MessageId, Message)> = match self
+      .client
+      .retry_until_included(receipt.message_id(), interval, max_attempts)
+      .await
+    {
+      Ok(reattached_messages) => reattached_messages,
+      Err(e) => match e {
+        IotaClientError::TangleInclusionError(_) => {
+          if self.is_message_included(receipt.message_id()).await? {
+            return Ok(receipt);
+          }
+          return Err(Error::from(e));
+        }
+        _ => return Err(Error::from(e)),
+      },
+    };
+    match reattached_messages.into_iter().next() {
+      Some((_, message)) => Ok(Receipt::new(self.network.clone(), message)),
+      None => Err(Error::MessageInclusionNotChecked),
+    }
+  }
+
   /// Fetch the [`IotaDocument`] specified by the given [`IotaDID`].
   pub async fn read_document(&self, did: &IotaDID) -> Result<IotaDocument> {
     self.read_document_chain(did).await.and_then(DocumentChain::fold)
@@ -187,6 +230,22 @@ impl Client {
       .await
       .map_err(Into::into)
   }
+
+  async fn is_message_included(&self, message_id: &MessageId) -> Result<bool> {
+    match self
+      .client
+      .get_message()
+      .metadata(message_id)
+      .await?
+      .ledger_inclusion_state
+    {
+      Some(ledger_inclusion_state) => match ledger_inclusion_state {
+        LedgerInclusionStateDto::Included | LedgerInclusionStateDto::NoTransaction => Ok(true),
+        LedgerInclusionStateDto::Conflicting => Ok(false),
+      },
+      None => Ok(false),
+    }
+  }
 }
 
 #[async_trait::async_trait(?Send)]