Merge pull request #363 from locke2833/reattach

Implement transaction reattachment related APIs

Author: jserv

README.md

@@ -156,7 +156,9 @@ For more information for MQTT connectivity of `tangle-accelerator`, you could re
 ### Optional: Enable external database for transaction reattachment
 Transaction reattachment is an optional feature.
 
-You can enable it in the build time by adding option : `--define db=enable`
+You can enable it in the build time by adding option : `--define db=enable`.
+
+When enabling reattachment, every transaction issues from the `tangle-accelerator` API called `Send Transfer Message` will be stored in the specific ScyllaDB host and response a UUID string for each transfer message as the identifier. With a promoting process that monitors the status of storing transactions, persistent pending transactions will be reattached to the Tangle.
 
 Transaction reattachment relies on ScyllDB, you need to install the dependency by following commands.
 

accelerator/BUILD

@@ -54,13 +54,26 @@ cc_binary(
             "-pg",
         ],
         "//conditions:default": ["-DNDEBUG"],
+    }) + select({
+        "//connectivity/mqtt:mqtt_enable": [
+            "-DMQTT_ENABLE",
+        ],
+        "//conditions:default": [],
+    }) + select({
+        "//storage:db_enable": ["-DDB_ENABLE"],
+        "//conditions:default": [],
     }),
     deps = [
         ":http",
         ":ta_config",
         ":ta_errors",
         "@entangled//utils/handles:signal",
-    ],
+    ] + select({
+        "//connectivity/mqtt:mqtt_enable": [
+            "//connectivity/mqtt:mqtt_utils",
+        ],
+        "//conditions:default": [],
+    }),
 )
 
 cc_image(
@@ -81,6 +94,15 @@ cc_library(
     name = "apis",
     srcs = ["apis.c"],
     hdrs = ["apis.h"],
+    copts = select({
+        "//connectivity/mqtt:mqtt_enable": [
+            "-DMQTT_ENABLE",
+        ],
+        "//conditions:default": [],
+    }) + select({
+        "//storage:db_enable": ["-DDB_ENABLE"],
+        "//conditions:default": [],
+    }),
     linkopts = ["-lpthread"],
     visibility = ["//visibility:public"],
     deps = [
@@ -114,6 +136,15 @@ cc_library(
     name = "http",
     srcs = ["http.c"],
     hdrs = ["http.h"],
+    copts = select({
+        "//connectivity/mqtt:mqtt_enable": [
+            "-DMQTT_ENABLE",
+        ],
+        "//conditions:default": [],
+    }) + select({
+        "//storage:db_enable": ["-DDB_ENABLE"],
+        "//conditions:default": [],
+    }),
     visibility = ["//visibility:public"],
     deps = [
         ":apis",
@@ -128,6 +159,10 @@ cc_library(
     name = "common_core",
     srcs = ["common_core.c"],
     hdrs = ["common_core.h"],
+    copts = select({
+        "//storage:db_enable": ["-DDB_ENABLE"],
+        "//conditions:default": [],
+    }),
     visibility = ["//visibility:public"],
     deps = [
         ":ta_config",

accelerator/apis.c

@@ -500,8 +500,7 @@ status_t api_mam_send_message(const iota_config_t* const iconf, const iota_clien
   return ret;
 }
 
-status_t api_send_transfer(const iota_config_t* const iconf, const iota_client_service_t* const service,
-                           const char* const obj, char** json_result) {
+status_t api_send_transfer(const ta_core_t* const core, const char* const obj, char** json_result) {
   status_t ret = SC_OK;
   ta_send_transfer_req_t* req = ta_send_transfer_req_new();
   ta_send_transfer_res_t* res = ta_send_transfer_res_new();
@@ -521,26 +520,31 @@ status_t api_send_transfer(const iota_config_t* const iconf, const iota_client_s
     goto done;
   }
 
-  ret = ta_send_transfer(iconf, service, req, res);
+  ret = ta_send_transfer(&core->iota_conf, &core->iota_service, req, res);
   if (ret) {
     lock_handle_unlock(&cjson_lock);
     goto done;
   }
   lock_handle_unlock(&cjson_lock);
-
   // return transaction object
   hash243_queue_push(&txn_obj_req->hashes, hash243_queue_peek(res->hash));
 
   lock_handle_lock(&cjson_lock);
-  ret = ta_find_transaction_objects(service, txn_obj_req, res_txn_array);
+  ret = ta_find_transaction_objects(&core->iota_service, txn_obj_req, res_txn_array);
   if (ret) {
     lock_handle_unlock(&cjson_lock);
     goto done;
   }
   lock_handle_unlock(&cjson_lock);
-
-  ret = ta_send_transfer_res_serialize(res_txn_array, json_result);
-
+  res->txn_array = res_txn_array;
+#ifdef DB_ENABLE
+  ret = db_insert_tx_into_identity(&core->db_service, res->hash, PENDING_TXN, res->uuid_string);
+  if (ret != SC_OK) {
+    ta_log_error("fail to insert new pending transaction for reattachement\n");
+    goto done;
+  }
+#endif
+  ret = ta_send_transfer_res_serialize(res, json_result);
 done:
   ta_send_transfer_req_free(&req);
   ta_send_transfer_res_free(&res);
@@ -580,3 +584,90 @@ status_t api_send_trytes(const iota_config_t* const iconf, const iota_client_ser
   hash_array_free(trytes);
   return ret;
 }
+
+#ifdef DB_ENABLE
+status_t api_find_transactions_by_id(const iota_client_service_t* const iota_service,
+                                     const db_client_service_t* const db_service, const char* const obj,
+                                     char** json_result) {
+  if (obj == NULL) {
+    ta_log_error("Invalid NULL pointer to uuid string\n");
+    return SC_TA_NULL;
+  }
+  status_t ret = SC_OK;
+  ta_log_info("find transaction by uuid string: %s\n", obj);
+  db_identity_array_t* db_identity_array = db_identity_array_new();
+  ret = db_get_identity_objs_by_uuid_string(db_service, obj, db_identity_array);
+  if (ret != SC_OK) {
+    ta_log_error("fail to find transaction by uuid string\n");
+    goto exit;
+  }
+
+  db_identity_t* itr = (db_identity_t*)utarray_front(db_identity_array);
+  if (itr != NULL) {
+    ret = api_find_transaction_object_single(iota_service, (const char* const)db_ret_identity_hash(itr), json_result);
+  } else {
+    ta_log_error("No corresponding transaction found by uuid string : %s\n", obj);
+    ret = SC_TA_WRONG_REQUEST_OBJ;
+  }
+
+exit:
+  db_identity_array_free(&db_identity_array);
+  return ret;
+}
+
+status_t api_get_identity_info_by_hash(const db_client_service_t* const db_service, const char* const obj,
+                                       char** json_result) {
+  if (obj == NULL) {
+    ta_log_error("Invalid NULL pointer to uuid string\n");
+    return SC_TA_NULL;
+  }
+  status_t ret = SC_OK;
+  ta_log_info("get identity info by hash : %s\n", obj);
+  db_identity_array_t* db_identity_array = db_identity_array_new();
+  ret = db_get_identity_objs_by_hash(db_service, (const cass_byte_t*)obj, db_identity_array);
+  if (ret != SC_OK) {
+    ta_log_error("fail to get identity objs by transaction hash\n");
+    goto exit;
+  }
+
+  db_identity_t* itr = (db_identity_t*)utarray_front(db_identity_array);
+  if (itr != NULL) {
+    ret = db_identity_serialize(json_result, itr);
+  } else {
+    ta_log_error("No corresponding identity info found by hash : %s\n", obj);
+    ret = SC_TA_WRONG_REQUEST_OBJ;
+  }
+exit:
+  db_identity_array_free(&db_identity_array);
+  return ret;
+}
+
+status_t api_get_identity_info_by_id(const db_client_service_t* const db_service, const char* const obj,
+                                     char** json_result) {
+  if (obj == NULL) {
+    ta_log_error("Invalid NULL pointer to uuid string\n");
+    return SC_TA_NULL;
+  }
+
+  status_t ret = SC_OK;
+  ta_log_info("get identity info by uuid string : %s\n", obj);
+  db_identity_array_t* db_identity_array = db_identity_array_new();
+  ret = db_get_identity_objs_by_uuid_string(db_service, obj, db_identity_array);
+  if (ret != SC_OK) {
+    ta_log_error("fail to get identity objs by uuid string\n");
+    goto exit;
+  }
+
+  db_identity_t* itr = (db_identity_t*)utarray_front(db_identity_array);
+  if (itr != NULL) {
+    ret = db_identity_serialize(json_result, itr);
+  } else {
+    ta_log_error("No corresponding identity info found by uuid string : %s\n", obj);
+    ret = SC_TA_WRONG_REQUEST_OBJ;
+  }
+
+exit:
+  db_identity_array_free(&db_identity_array);
+  return ret;
+}
+#endif

accelerator/apis.h

@@ -167,17 +167,15 @@ status_t api_mam_send_message(const iota_config_t* const iconf, const iota_clien
  * fields include address, value, tag, and message. This API would also try to
  * find the transactions after bundle sent.
  *
- * @param[in] iconf IOTA API parameter configurations
- * @param[in] service IRI node end point service
+ * @param core[in] Pointer to Tangle-accelerator core configuration structure
  * @param[in] obj Input data in JSON
  * @param[out] json_result Result containing transaction objects in json format
  *
  * @return
  * - SC_OK on success
  * - non-zero on error
  */
-status_t api_send_transfer(const iota_config_t* const iconf, const iota_client_service_t* const service,
-                           const char* const obj, char** json_result);
+status_t api_send_transfer(const ta_core_t* const core, const char* const obj, char** json_result);
 
 /**
  * @brief Return transaction object with given single transaction hash.
@@ -268,6 +266,54 @@ status_t api_find_transactions_obj_by_tag(const iota_client_service_t* const ser
  */
 status_t api_send_trytes(const iota_config_t* const iconf, const iota_client_service_t* const service,
                          const char* const obj, char** json_result);
+#ifdef DB_ENABLE
+/**
+ * @brief Return transaction object with given single identity number.
+ *
+ * Explore transaction hash information with given single identity number. This would
+ * return whole transaction object details in json format instead of raw trytes.
+ *
+ * @param[in] iota_service IRI node end point service
+ * @param[in] db_service db client service
+ * @param[in] obj identity number
+ * @param[out] json_result Result containing the only one transaction object in json format
+ *
+ * @return
+ * - SC_OK on success
+ * - non-zero on error
+ */
+status_t api_find_transactions_by_id(const iota_client_service_t* const iota_service,
+                                     const db_client_service_t* const db_service, const char* const obj,
+                                     char** json_result);
+
+/**
+ * @brief Return db identity object with given single transaction hash.
+ *
+ * @param[in] db_service db client service
+ * @param[in] obj transaction hash
+ * @param[out] json_result Result containing the only one db identity object in json format
+ *
+ * @return
+ * - SC_OK on success
+ * - non-zero on error
+ */
+status_t api_get_identity_info_by_hash(const db_client_service_t* const db_service, const char* const obj,
+                                       char** json_result);
+
+/**
+ * @brief Return db identity object with given single transaction id.
+ *
+ * @param[in] db_service db client service
+ * @param[in] obj transaction id
+ * @param[out] json_result Result containing the only one db identity object in json format
+ *
+ * @return
+ * - SC_OK on success
+ * - non-zero on error
+ */
+status_t api_get_identity_info_by_id(const db_client_service_t* const db_service, const char* const obj,
+                                     char** json_result);
+#endif
 
 #ifdef __cplusplus
 }

accelerator/config.c

@@ -78,6 +78,7 @@ static status_t cli_core_set(ta_core_t* const core, int key, char* const value)
 #ifdef DB_ENABLE
     // DB configuration
     case DB_HOST_CLI:
+      free(db_service->host);
       db_service->host = strdup(value);
       break;
 #endif
@@ -337,7 +338,7 @@ status_t ta_core_set(ta_core_t* core) {
   cache_init(cache->cache_state, cache->host, cache->port);
 #ifdef DB_ENABLE
   ta_log_info("Initializing db client service\n");
-  if ((ret = db_client_service_init(db_service)) != SC_OK) {
+  if ((ret = db_client_service_init(db_service, DB_USAGE_REATTACH)) != SC_OK) {
     ta_log_error("Initializing DB connection failed\n");
   }
 #endif
@@ -347,20 +348,14 @@ status_t ta_core_set(ta_core_t* core) {
 }
 
 void ta_core_destroy(ta_core_t* const core) {
-  iota_client_service_t* const iota_service = &core->iota_service;
-#ifdef DB_ENABLE
-  db_client_service_t* const db_service = &core->db_service;
-#endif
-
   ta_log_info("Destroying IRI connection\n");
   iota_client_extended_destroy();
-  iota_client_core_destroy(iota_service);
+  iota_client_core_destroy(&core->iota_service);
 #ifdef DB_ENABLE
-  if (db_service->enabled) {
-    ta_log_info("Destroying DB connection\n");
-    db_client_service_free(db_service);
-  }
+  ta_log_info("Destroying DB connection\n");
+  db_client_service_free(&core->db_service);
 #endif
+
   pow_destroy();
   cache_stop();
   logger_helper_release(logger_id);