Merge pull request #526 from HowJMay/publish_txn

feat(core): Broadcast buffered transaction objects

Author: howjmay

accelerator/config.c

@@ -236,6 +236,8 @@ status_t ta_core_default_init(ta_core_t* const core) {
   cache->host = REDIS_HOST;
   cache->port = REDIS_PORT;
   cache->cache_state = false;
+  cache->buffer_list_name = BUFFER_LIST_NAME;
+  cache->done_list_name = DONE_LIST_NAME;
 
   ta_log_info("Initializing IRI configuration\n");
   iota_conf->milestone_depth = MILESTONE_DEPTH;

accelerator/config.h

@@ -53,6 +53,8 @@ extern "C" {
   "AMRWQP9BUMJALJHBXUCHOD9HFFD9LGTGEAWMJWWXSDVOF9PI9YGJAPBQLQUOMNYEQCZPGCTHGV" \
   "NNAPGHA"
 #define MAM_FILE_PREFIX "/tmp/mam_bin_XXXXXX"
+#define BUFFER_LIST_NAME "txn_buff_list"
+#define DONE_LIST_NAME "done_txn_buff_list"
 #define IRI_HEALTH_TRACK_PERIOD 1800  // Check every half hour in default
 
 /** @name Redis connection config */
@@ -88,9 +90,12 @@ typedef struct iota_config_s {
 
 /** struct type of accelerator cache */
 typedef struct ta_cache_s {
-  char* host;       /**< Binding address of redis server */
-  uint16_t port;    /**< Binding port of redis server */
-  bool cache_state; /**< Set it true to turn on cache server */
+  char* host;             /**< Binding address of redis server */
+  uint64_t timeout;       /**< Timeout for keys in redis */
+  char* buffer_list_name; /**< Name of the list to buffer transactions */
+  char* done_list_name;   /**< Name of the list to store successfully broadcast transactions from buffer */
+  uint16_t port;          /**< Binding port of redis server */
+  bool cache_state;       /**< Set it true to turn on cache server */
 } ta_cache_t;
 
 /** struct type of accelerator core */

accelerator/core/BUILD

@@ -4,7 +4,6 @@ cc_library(
     hdrs = ["apis.h"],
     linkopts = [
         "-lpthread",
-        "-luuid",
     ],
     visibility = ["//visibility:public"],
     deps = [
@@ -35,6 +34,9 @@ cc_library(
     name = "core",
     srcs = ["core.c"],
     hdrs = ["core.h"],
+    linkopts = [
+        "-luuid",
+    ],
     visibility = ["//visibility:public"],
     deps = [
         "//accelerator:ta_config",

accelerator/core/apis.c

@@ -334,7 +334,6 @@ status_t api_recv_mam_message(const iota_config_t* const iconf, const iota_clien
   }
 
   recv_mam_data_id_mam_v1_t* data_id = (recv_mam_data_id_mam_v1_t*)req->data_id;
-
   if (mam_api_init(&mam, (tryte_t*)iconf->seed) != RC_OK) {
     ret = SC_MAM_FAILED_INIT;
     ta_log_error("%s\n", "SC_MAM_FAILED_INIT");
@@ -566,26 +565,10 @@ status_t api_send_transfer(const ta_core_t* const core, const char* const obj, c
     goto done;
   }
 
-  ret = ta_send_transfer(&core->ta_conf, &core->iota_conf, &core->iota_service, req, res);
+  ret = ta_send_transfer(&core->ta_conf, &core->iota_conf, &core->iota_service, &core->cache, req, res);
   if (ret == SC_CCLIENT_FAILED_RESPONSE) {
     lock_handle_unlock(&cjson_lock);
     ta_log_info("%s\n", "Caching transaction");
-    // TODO generate a UUID as redis key
-    uuid_t binuuid;
-    uuid_generate_random(binuuid);
-    uuid_unparse(binuuid, res->uuid);
-    if (!res->uuid[0]) {
-      ta_log_error("%s\n", "Failed to generate UUID");
-      goto done;
-    }
-
-    // Cache the txn in redis
-    // TODO use a timer to reattach these failed transactions
-    ret = cache_set(res->uuid, UUID_STR_LEN - 1, obj, strlen(obj), CACHE_FAILED_TXN_TIMEOUT);
-    if (ret) {
-      ta_log_error("%s\n", ta_error_to_string(ret));
-      goto done;
-    }
 
     // Cache the request and serialize UUID as response directly
     goto serialize;

accelerator/core/apis.h

@@ -15,11 +15,6 @@
 #include "mam/mam/mam_channel_t_set.h"
 #include "serializer/serializer.h"
 
-// TODO The temporary default timeout in cache server is 1 week. We should investigate the performance of redis to
-// design a better data structure and appropriate timeout period. And we should study the methodology to partially
-// release cached data.
-#define CACHE_FAILED_TXN_TIMEOUT (7 * 24 * 60 * 60)
-
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -54,7 +49,7 @@ int apis_logger_release();
  * @param tangle[in] iota configuration variables
  * @param cache[in] redis configuration variables
  * @param service[in] IRI connection configuration variables
- * @param[out] json_result Result containing tangle accelerator information in json format
+ * @param json_result[out] Result containing tangle accelerator information in json format
  *
  * @return
  * - SC_OK on success

accelerator/core/core.c

@@ -31,7 +31,6 @@ status_t ta_attach_to_tangle(const attach_to_tangle_req_t* const req, attach_to_
   iota_transaction_t tx;
   flex_trit_t* elt = NULL;
 
-  // TODO Save fetched transaction object in a set.
   // create bundle
   bundle_transactions_new(&bundle);
   HASH_ARRAY_FOREACH(req->trytes, elt) {
@@ -91,7 +90,9 @@ status_t ta_send_trytes(const ta_config_t* const info, const iota_config_t* cons
   HASH_ARRAY_FOREACH(trytes, elt) { attach_to_tangle_req_trytes_add(attach_req, elt); }
   attach_to_tangle_req_init(attach_req, get_transactions_to_approve_res_trunk(tx_approve_res),
                             get_transactions_to_approve_res_branch(tx_approve_res), iconf->mwm);
-  if (ta_attach_to_tangle(attach_req, attach_res) != SC_OK) {
+  ret = ta_attach_to_tangle(attach_req, attach_res);
+  if (ret != SC_OK) {
+    ta_log_error("%s\n", ta_error_to_string(ret));
     goto done;
   }
 
@@ -163,8 +164,8 @@ status_t ta_generate_address(const iota_config_t* const iconf, const iota_client
 }
 
 status_t ta_send_transfer(const ta_config_t* const info, const iota_config_t* const iconf,
-                          const iota_client_service_t* const service, const ta_send_transfer_req_t* const req,
-                          ta_send_transfer_res_t* res) {
+                          const iota_client_service_t* const service, const ta_cache_t* const cache,
+                          const ta_send_transfer_req_t* const req, ta_send_transfer_res_t* res) {
   if (req == NULL || res == NULL) {
     ta_log_error("%s\n", "SC_TA_NULL");
     return SC_TA_NULL;
@@ -225,6 +226,9 @@ status_t ta_send_transfer(const ta_config_t* const info, const iota_config_t* co
 
   ret = ta_send_trytes(info, iconf, service, raw_tx);
   if (ret) {
+    ta_log_error("Error in ta_send_trytes. Push transaction trytes to buffer.\n");
+    res->uuid = (char*)malloc(sizeof(char) * UUID_STR_LEN);
+    push_txn_to_buffer(cache, raw_tx, res->uuid);
     goto done;
   }
 
@@ -597,3 +601,95 @@ status_t ta_update_iri_conneciton(ta_config_t* const ta_conf, iota_client_servic
 done:
   return ret;
 }
+
+status_t push_txn_to_buffer(const ta_cache_t* const cache, hash8019_array_p raw_txn_flex_trit_array, char* uuid) {
+  status_t ret = SC_OK;
+  if (!uuid) {
+    ret = SC_CORE_NULL;
+    ta_log_error("%s\n", ta_error_to_string(ret));
+    goto done;
+  }
+
+  uuid_t binuuid;
+  uuid_generate_random(binuuid);
+  uuid_unparse(binuuid, uuid);
+  if (!uuid[0]) {
+    ta_log_error("%s\n", "Failed to generate UUID");
+    goto done;
+  }
+
+  // TODO We push only one transaction raw trits into list, we need to solve this in future.
+  ret = cache_set(uuid, UUID_STR_LEN - 1, hash_array_at(raw_txn_flex_trit_array, 0),
+                  NUM_FLEX_TRITS_SERIALIZED_TRANSACTION, cache->timeout);
+  if (ret) {
+    ta_log_error("%s\n", ta_error_to_string(ret));
+    goto done;
+  }
+
+  ret =
+      cache_list_push(cache->buffer_list_name, strlen(cache->buffer_list_name), uuid, UUID_STR_LEN - 1, cache->timeout);
+  if (ret) {
+    ta_log_error("%s\n", ta_error_to_string(ret));
+  }
+
+done:
+  return ret;
+}
+
+status_t broadcast_buffered_txn(const ta_core_t* const core) {
+  status_t ret = SC_OK;
+  int uuid_list_len = 0;
+  hash8019_array_p txn_trytes_array = hash8019_array_new();
+
+  do {
+    char uuid[UUID_STR_LEN];
+    flex_trit_t req_txn_flex_trits[NUM_FLEX_TRITS_SERIALIZED_TRANSACTION + 1];
+
+    ret = cache_list_size(core->cache.buffer_list_name, &uuid_list_len);
+    if (ret) {
+      ta_log_error("%s\n", ta_error_to_string(ret));
+      goto done;
+    }
+
+    ret = cache_list_peek(core->cache.buffer_list_name, UUID_STR_LEN, uuid);
+    if (ret) {
+      ta_log_error("%s\n", ta_error_to_string(ret));
+      goto done;
+    }
+
+    // TODO Now we assume every time we call `cache_get()`, we would get a transaction object. However, in the future,
+    // the returned result may be a bunlde.
+    ret = cache_get(uuid, (char*)req_txn_flex_trits);
+    if (ret) {
+      ta_log_error("%s\n", ta_error_to_string(ret));
+      goto done;
+    }
+
+    hash_array_push(txn_trytes_array, req_txn_flex_trits);
+    ret = ta_send_trytes(&core->ta_conf, &core->iota_conf, &core->iota_service, txn_trytes_array);
+    if (ret) {
+      ta_log_error("%s\n", ta_error_to_string(ret));
+      goto done;
+    }
+    utarray_done(txn_trytes_array);
+
+    // Pop transaction from buffered list
+    ret = cache_list_pop(core->cache.buffer_list_name, (char*)req_txn_flex_trits);
+    if (ret) {
+      ta_log_error("%s\n", ta_error_to_string(ret));
+      goto done;
+    }
+
+    // Transfer the transaction to another list in where we store all the successfully broadcasted transactions.
+    ret = cache_list_push(core->cache.done_list_name, strlen(core->cache.done_list_name), uuid, UUID_STR_LEN - 1,
+                          core->cache.timeout);
+    if (ret) {
+      ta_log_error("%s\n", ta_error_to_string(ret));
+      goto done;
+    }
+  } while (!uuid_list_len);
+
+done:
+  hash_array_free(txn_trytes_array);
+  return ret;
+}

accelerator/core/core.h

@@ -84,6 +84,7 @@ status_t ta_generate_address(const iota_config_t* const iconf, const iota_client
  * @param[in] info Tangle-accelerator configuration variables
  * @param[in] iconf IOTA API parameter configurations
  * @param[in] service IRI node end point service
+ * @param[in] cache redis configuration variables
  * @param[in] req Request containing address value, message, tag in
  *                ta_send_transfer_req_t
  * @param[out] res Result containing transaction hash in ta_send_transfer_res_t
@@ -93,8 +94,8 @@ status_t ta_generate_address(const iota_config_t* const iconf, const iota_client
  * - non-zero on error
  */
 status_t ta_send_transfer(const ta_config_t* const info, const iota_config_t* const iconf,
-                          const iota_client_service_t* const service, const ta_send_transfer_req_t* const req,
-                          ta_send_transfer_res_t* res);
+                          const iota_client_service_t* const service, const ta_cache_t* const cache,
+                          const ta_send_transfer_req_t* const req, ta_send_transfer_res_t* res);
 
 /**
  * @brief Send trytes to tangle.
@@ -254,6 +255,37 @@ status_t ta_get_iri_status(const iota_client_service_t* const service);
  */
 status_t ta_update_iri_conneciton(ta_config_t* const ta_conf, iota_client_service_t* const service);
 
+/**
+ * @brief Push failed transactions in raw trytes into transaction buffer
+ *
+ * Given raw trytes array would be pushed into buffer. An UUID will be returned for client to fetch the information of
+ * their request. The UUIDs are stored in a list, so once reaching the capacity of the buffer, buffered transactions can
+ * be popped from the buffer.
+ *
+ * @param cache[in] Redis configuration variables
+ * @param raw_txn_flex_trit_array[in] Raw transction trytes array in flex_trit_t type
+ * @param uuid[out] Returned UUID for fetching transaction status and information
+ *
+ * @return
+ * - SC_OK on success
+ * - non-zero on error
+ */
+status_t push_txn_to_buffer(const ta_cache_t* const cache, hash8019_array_p raw_txn_flex_trit_array, char* uuid);
+
+/**
+ * @brief Broadcast transactions in transaction buffer
+ *
+ * Failed transactions would be stored in transaciton buffer. Once tangle-accelerator retrieve the connetion with
+ * Tangle, then tangle-accelerator will start to broadcast these failed transaction trytes.
+ *
+ * @param core[in] Pointer to Tangle-accelerator core configuration structure
+ *
+ * @return
+ * - SC_OK on success
+ * - non-zero on error
+ */
+status_t broadcast_buffered_txn(const ta_core_t* const core);
+
 #ifdef __cplusplus
 }
 #endif

accelerator/core/response/ta_send_transfer.c

@@ -11,12 +11,13 @@
 ta_send_transfer_res_t* ta_send_transfer_res_new() {
   ta_send_transfer_res_t* res = (ta_send_transfer_res_t*)malloc(sizeof(ta_send_transfer_res_t));
   res->hash = NULL;
-  memset(res->uuid, 0, UUID_STR_LEN);
+  res->uuid = NULL;
   return res;
 }
 
 void ta_send_transfer_res_free(ta_send_transfer_res_t** res) {
   if ((*res)) {
+    free((*res)->uuid);
     hash243_queue_free(&(*res)->hash);
     free((*res));
     *res = NULL;

accelerator/core/response/ta_send_transfer.h

@@ -30,9 +30,7 @@ typedef struct {
   /** Transaction address is a 243 long flex trits hash queue. */
   hash243_queue_t hash;
   transaction_array_t* txn_array;
-  // If the value of `CASS_UUID_STRING_LENGTH` changes, then we may need to modify array length of uuid_string as well,
-  // since UUID_STR_LEN is defined in `uuid/uuid.h`.
-  char uuid[UUID_STR_LEN];
+  char* uuid;
 #ifdef DB_ENABLE
   char uuid_string[DB_UUID_STRING_LENGTH];
 #endif

accelerator/core/serializer/serializer.c

@@ -822,7 +822,7 @@ status_t ta_send_transfer_res_serialize(ta_send_transfer_res_t* res, char** obj)
     goto done;
   }
 
-  if (res->uuid[0]) {
+  if (res->uuid) {
     cJSON_AddStringToObject(json_root, "uuid", res->uuid);
   } else {
     ret = iota_transaction_to_json_object(transaction_array_at(res->txn_array, 0), &json_root);