Add publish_json_with_retry method (#455)

* add publish with retry method

* replace publish_json with publish_json_with_retry

* add wasm bindings for retry_until_included and publish_with_retry

* run cargo fmt and cargo clippy

* replace u64 for u32 in wasm interface

* run cargo fmt

* improve comments, remove method

* unpin reqwest dependency, update comments

* add parking_lot with wasm-bindgen feature as a dependency

* remove tokio dependency

* add type annotation to new variables, remove unnecessary clone

* remove unnecessary clone

* add check for message inclusion in retry_until_included

* add bee_rest_api dependency

* solve doc conflicts

* restructure match statement

* add updated docs

* add minimum supported version for node

* bump node version in github actions, increase timeout for tests

.github/workflows/build-and-test.yml

@@ -165,7 +165,7 @@ jobs:
       - name: Set up Node.js
         uses: actions/setup-node@v1
         with:
-          node-version: 15.x
+          node-version: 16.x
 
       - name: Install wasm-pack
         run: npm install -g wasm-pack

bindings/wasm/README.md

@@ -36,6 +36,10 @@ or for the `web` with
 npm run build:web
 ```
 
+## Minimum Requirements
+
+The minimum supported version for node is: `v16.0.0`
+
 ## NodeJS Usage
 <!-- 
 Test this example using https://github.com/anko/txm: `txm README.md`

bindings/wasm/cypress/integration/browser_test.js

@@ -6,7 +6,7 @@ import {
 describe(
     "Test browser examples",
     {
-        defaultCommandTimeout: 180000, // 3 minutes to account for spurious network delays
+        defaultCommandTimeout: 300000, // 5 minutes to account for spurious network delays
     },
     () => {
         beforeEach(async () => {

bindings/wasm/docs/api-reference.md

@@ -44,12 +44,12 @@
 ## Members
 
 <dl>
+<dt><a href="#KeyType">KeyType</a></dt>
+<dd></dd>
 <dt><a href="#DIDMessageEncoding">DIDMessageEncoding</a></dt>
 <dd></dd>
 <dt><a href="#Digest">Digest</a></dt>
 <dd></dd>
-<dt><a href="#KeyType">KeyType</a></dt>
-<dd></dd>
 </dl>
 
 ## Functions
@@ -72,6 +72,7 @@
         * [.publishDocument(document)](#Client+publishDocument) ⇒ [<code>Promise.&lt;Receipt&gt;</code>](#Receipt)
         * [.publishDiff(message_id, diff)](#Client+publishDiff) ⇒ [<code>Promise.&lt;Receipt&gt;</code>](#Receipt)
         * [.publishJSON(index, data)](#Client+publishJSON) ⇒ [<code>Promise.&lt;Receipt&gt;</code>](#Receipt)
+        * [.publishJsonWithRetry(index, data, interval, max_attempts)](#Client+publishJsonWithRetry) ⇒ <code>Promise.&lt;any&gt;</code>
         * [.resolve(did)](#Client+resolve) ⇒ [<code>Promise.&lt;Document&gt;</code>](#Document)
         * [.resolveHistory(did)](#Client+resolveHistory) ⇒ [<code>Promise.&lt;DocumentHistory&gt;</code>](#DocumentHistory)
         * [.resolveDiffHistory(document)](#Client+resolveDiffHistory) ⇒ [<code>Promise.&lt;DiffChainHistory&gt;</code>](#DiffChainHistory)
@@ -127,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;Document&gt;</code>](#Document)
@@ -1827,6 +1844,10 @@ Deserializes a `VerificationMethod` object from a JSON object.
 | --- | --- |
 | value | <code>any</code> | 
 
+<a name="KeyType"></a>
+
+## KeyType
+**Kind**: global variable  
 <a name="DIDMessageEncoding"></a>
 
 ## DIDMessageEncoding
@@ -1835,10 +1856,6 @@ Deserializes a `VerificationMethod` object from a JSON object.
 
 ## Digest
 **Kind**: global variable  
-<a name="KeyType"></a>
-
-## KeyType
-**Kind**: global variable  
 <a name="start"></a>
 
 ## start()

bindings/wasm/examples/src/test.js

@@ -10,7 +10,7 @@ import { resolveHistory } from "./resolve_history";
 import { CLIENT_CONFIG } from "./config";
 import { createIdentityPrivateTangle } from "./private_tangle";
 
-jest.setTimeout(180000); // 3 minutes to account for spurious network delays, most tests pass in a few seconds
+jest.setTimeout(300000); // 5 minutes to account for spurious network delays, most tests pass in a few seconds
 
 // Run all Node.js examples as jest tests in parallel.
 // If a function throws an exception, it will run again to make the tests more consistent (less prone to network issues). 

bindings/wasm/package.json

@@ -61,5 +61,8 @@
   },
   "dependencies": {
     "node-fetch": "^2.6.0"
+  },
+  "engines": {
+    "node": ">=16"
   }
 }

bindings/wasm/src/tangle/client.rs

@@ -136,6 +136,37 @@ impl Client {
     Ok(promise.unchecked_into::<PromiseReceipt>())
   }
 
+  /// 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<PromiseDocument> {
     let client: Rc<IotaClient> = self.client.clone();

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" }
@@ -28,7 +29,8 @@ strum = { version = "0.21", features = ["derive"] }
 thiserror = { version = "1.0", default-features = false }
 
 [dependencies.iota-client]
-version = "1.1.0"
+git = "https://github.com/iotaledger/iota.rs"
+rev = "c48db779d2162b9a3037ee63c4d1e267d04633eb"
 default-features = false
 
 [dependencies.iota-crypto]

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,42 @@ 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 retry_result: Result<Vec<(MessageId, Message)>, IotaClientError> = self
+      .client
+      .retry_until_included(receipt.message_id(), interval, max_attempts)
+      .await;
+    let reattached_messages: Vec<(MessageId, Message)> = match retry_result {
+      Ok(reattached_messages) => reattached_messages,
+      Err(inclusion_error @ IotaClientError::TangleInclusionError(_)) => {
+        if self.is_message_included(receipt.message_id()).await? {
+          return Ok(receipt);
+        } else {
+          return Err(Error::from(inclusion_error));
+        }
+      }
+      Err(error) => {
+        return Err(Error::from(error));
+      }
+    };
+    match reattached_messages.into_iter().next() {
+      Some((_, message)) => Ok(Receipt::new(self.network.clone(), message)),
+      None => Err(Error::from(IotaClientError::TangleInclusionError(
+        receipt.message_id().to_string(),
+      ))),
+    }
+  }
+
   /// 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 +233,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)]