Port tangle changes to stardust

bee-api/bee-rest-api/Cargo.toml

@@ -23,7 +23,7 @@ bee-message = { version = "0.2.0", path = "../../bee-message", default-features
 bee-pow = { version = "0.2.0", path = "../../bee-pow", default-features = false }
 bee-protocol = { version = "0.2.2", path = "../../bee-protocol", default-features = false, optional = true }
 bee-runtime = { version = "0.1.1-alpha", path = "../../bee-runtime", default-features = false, optional = true  }
-bee-storage = { version = "0.9.0", path = "../../bee-storage/bee-storage", default-features = false, optional = true }
+bee-storage = { version = "0.12.0", path = "../../bee-storage/bee-storage", default-features = false, optional = true }
 bee-tangle = { version = "0.3.0", path = "../../bee-tangle", default-features = false, optional = true }
 
 async-trait = { version = "0.1.51", default-features = false, optional = true }

bee-api/bee-rest-api/src/endpoints/mod.rs

@@ -35,7 +35,7 @@ pub(crate) type Bech32Hrp = String;
 
 pub(crate) const CONFIRMED_THRESHOLD: u32 = 5;
 
-pub async fn init_full_node<N: Node>(
+pub fn init_full_node<N: Node>(
     rest_api_config: RestApiConfig,
     protocol_config: ProtocolConfig,
     network_id: NetworkId,
@@ -105,7 +105,7 @@ where
                 requested_messages,
                 consensus_worker,
             )
-            .recover(handle_rejection);
+            .recover(|err| async { handle_rejection(err) });
 
             let (_, server) =
                 warp::serve(routes).bind_with_graceful_shutdown(rest_api_config.bind_socket_addr(), async {
@@ -121,7 +121,7 @@ where
     }
 }
 
-async fn handle_rejection(err: Rejection) -> Result<impl Reply, Infallible> {
+fn handle_rejection(err: Rejection) -> Result<impl Reply, Infallible> {
     let (http_code, err_code, reason) = match err.find() {
         // handle custom rejections
         Some(CustomRejection::Forbidden) => (StatusCode::FORBIDDEN, "403", "access forbidden"),
@@ -149,7 +149,7 @@ async fn handle_rejection(err: Rejection) -> Result<impl Reply, Infallible> {
     ))
 }
 
-pub async fn init_entry_node<N: Node>(rest_api_config: RestApiConfig, node_builder: N::Builder) -> N::Builder
+pub fn init_entry_node<N: Node>(rest_api_config: RestApiConfig, node_builder: N::Builder) -> N::Builder
 where
     N::Backend: StorageBackend,
 {
@@ -170,7 +170,9 @@ where
         node.spawn::<Self, _, _>(|shutdown| async move {
             info!("Running.");
 
-            let health = warp::path("health").map(|| StatusCode::OK).recover(handle_rejection);
+            let health = warp::path("health")
+                .map(|| StatusCode::OK)
+                .recover(|err| async { handle_rejection(err) });
 
             let (_, server) = warp::serve(health).bind_with_graceful_shutdown(config.bind_socket_addr(), async {
                 shutdown.await.ok();

bee-api/bee-rest-api/src/endpoints/routes/api/v2/info.rs

@@ -49,11 +49,23 @@ pub(crate) fn filter<B: StorageBackend>(
         .and(with_protocol_config(protocol_config))
         .and(with_node_info(node_info))
         .and(with_peer_manager(peer_manager))
-        .and_then(info)
+        .and_then(
+            |tangle, network_id, bech32_hrp, rest_api_config, protocol_config, node_info, peer_manager| async {
+                info(
+                    tangle,
+                    network_id,
+                    bech32_hrp,
+                    rest_api_config,
+                    protocol_config,
+                    node_info,
+                    peer_manager,
+                )
+            },
+        )
         .boxed()
 }
 
-pub(crate) async fn info<B: StorageBackend>(
+pub(crate) fn info<B: StorageBackend>(
     tangle: ResourceHandle<Tangle<B>>,
     network_id: NetworkId,
     bech32_hrp: Bech32Hrp,
@@ -65,15 +77,14 @@ pub(crate) async fn info<B: StorageBackend>(
     let latest_milestone_index = tangle.get_latest_milestone_index();
     let latest_milestone_timestamp = tangle
         .get_milestone(latest_milestone_index)
-        .await
         .map(|m| m.timestamp())
         .unwrap_or_default();
 
     Ok(warp::reply::json(&InfoResponse {
         name: node_info.name.clone(),
         version: node_info.version.clone(),
         status: StatusResponse {
-            is_healthy: health::is_healthy(&tangle, &peer_manager).await,
+            is_healthy: health::is_healthy(&tangle, &peer_manager),
             latest_milestone_timestamp,
             latest_milestone_index: *latest_milestone_index,
             confirmed_milestone_index: *tangle.get_confirmed_milestone_index(),

bee-api/bee-rest-api/src/endpoints/routes/api/v2/message.rs

@@ -32,15 +32,15 @@ pub(crate) fn filter<B: StorageBackend>(
         .and(warp::get())
         .and(has_permission(ROUTE_MESSAGE, public_routes, allowed_ips))
         .and(with_tangle(tangle))
-        .and_then(message)
+        .and_then(|message_id, tangle| async move { message(message_id, tangle) })
         .boxed()
 }
 
-pub(crate) async fn message<B: StorageBackend>(
+pub(crate) fn message<B: StorageBackend>(
     message_id: MessageId,
     tangle: ResourceHandle<Tangle<B>>,
 ) -> Result<impl Reply, Rejection> {
-    match tangle.get(&message_id).await.map(|m| (*m).clone()) {
+    match tangle.get(&message_id) {
         Some(message) => Ok(warp::reply::json(&MessageResponse(MessageDto::from(&message)))),
         None => Err(reject::custom(CustomRejection::NotFound(
             "can not find message".to_string(),

bee-api/bee-rest-api/src/endpoints/routes/api/v2/message_children.rs

@@ -33,15 +33,15 @@ pub(crate) fn filter<B: StorageBackend>(
         .and(warp::get())
         .and(has_permission(ROUTE_MESSAGE_CHILDREN, public_routes, allowed_ips))
         .and(with_tangle(tangle))
-        .and_then(message_children)
+        .and_then(|message_id, tangle| async move { message_children(message_id, tangle) })
         .boxed()
 }
 
-pub async fn message_children<B: StorageBackend>(
+pub fn message_children<B: StorageBackend>(
     message_id: MessageId,
     tangle: ResourceHandle<Tangle<B>>,
 ) -> Result<impl Reply, Rejection> {
-    let mut children = Vec::from_iter(tangle.get_children(&message_id).await.unwrap_or_default());
+    let mut children = Vec::from_iter(tangle.get_children(&message_id).unwrap_or_default());
     let count = children.len();
     let max_results = 1000;
     children.truncate(max_results);

bee-api/bee-rest-api/src/endpoints/routes/api/v2/message_metadata.rs

@@ -33,11 +33,11 @@ pub(crate) fn filter<B: StorageBackend>(
         .and(warp::get())
         .and(has_permission(ROUTE_MESSAGE_METADATA, public_routes, allowed_ips))
         .and(with_tangle(tangle))
-        .and_then(message_metadata)
+        .and_then(|message_id, tangle| async move { message_metadata(message_id, tangle) })
         .boxed()
 }
 
-pub(crate) async fn message_metadata<B: StorageBackend>(
+pub(crate) fn message_metadata<B: StorageBackend>(
     message_id: MessageId,
     tangle: ResourceHandle<Tangle<B>>,
 ) -> Result<impl Reply, Rejection> {
@@ -47,11 +47,8 @@ pub(crate) async fn message_metadata<B: StorageBackend>(
         )));
     }
 
-    match tangle.get(&message_id).await.map(|m| (*m).clone()) {
-        Some(message) => {
-            // existing message <=> existing metadata, therefore unwrap() is safe
-            let metadata = tangle.get_metadata(&message_id).await.unwrap();
-
+    match tangle.get_message_and_metadata(&message_id) {
+        Some((message, metadata)) => {
             // TODO: access constants from URTS
             let ymrsi_delta = 8;
             let omrsi_delta = 13;
@@ -112,13 +109,17 @@ pub(crate) async fn message_metadata<B: StorageBackend>(
                     conflict_reason = None;
 
                     let cmi = *tangle.get_confirmed_milestone_index();
+
                     // unwrap() of OMRSI/YMRSI is safe since message is solid
-                    if (cmi - *metadata.omrsi().unwrap().index()) > below_max_depth {
+                    let (omrsi, ymrsi) = metadata
+                        .omrsi_and_ymrsi()
+                        .map(|(o, y)| (*o.index(), *y.index()))
+                        .unwrap();
+
+                    if (cmi - omrsi) > below_max_depth {
                         should_promote = Some(false);
                         should_reattach = Some(true);
-                    } else if (cmi - *metadata.ymrsi().unwrap().index()) > ymrsi_delta
-                        || (cmi - *metadata.omrsi().unwrap().index()) > omrsi_delta
-                    {
+                    } else if (cmi - ymrsi) > ymrsi_delta || (cmi - omrsi) > omrsi_delta {
                         should_promote = Some(true);
                         should_reattach = Some(false);
                     } else {

bee-api/bee-rest-api/src/endpoints/routes/api/v2/message_raw.rs

@@ -31,15 +31,15 @@ pub(crate) fn filter<B: StorageBackend>(
         .and(warp::get())
         .and(has_permission(ROUTE_MESSAGE_RAW, public_routes, allowed_ips))
         .and(with_tangle(tangle))
-        .and_then(message_raw)
+        .and_then(|message_id, tangle| async move { message_raw(message_id, tangle) })
         .boxed()
 }
 
-pub async fn message_raw<B: StorageBackend>(
+pub fn message_raw<B: StorageBackend>(
     message_id: MessageId,
     tangle: ResourceHandle<Tangle<B>>,
 ) -> Result<impl Reply, Rejection> {
-    match tangle.get(&message_id).await.map(|m| (*m).clone()) {
+    match tangle.get(&message_id) {
         Some(message) => Ok(Response::builder()
             .header("Content-Type", "application/octet-stream")
             .body(message.pack_to_vec())),

bee-api/bee-rest-api/src/endpoints/routes/api/v2/milestone.rs

@@ -32,27 +32,22 @@ pub(crate) fn filter<B: StorageBackend>(
         .and(warp::get())
         .and(has_permission(ROUTE_MILESTONE, public_routes, allowed_ips))
         .and(with_tangle(tangle))
-        .and_then(milestone)
+        .and_then(|milestone_index, tangle| async move { milestone(milestone_index, tangle) })
         .boxed()
 }
 
-pub(crate) async fn milestone<B: StorageBackend>(
+pub(crate) fn milestone<B: StorageBackend>(
     milestone_index: MilestoneIndex,
     tangle: ResourceHandle<Tangle<B>>,
 ) -> Result<impl Reply, Rejection> {
-    match tangle.get_milestone_message_id(milestone_index).await {
-        Some(message_id) => match tangle.get_metadata(&message_id).await {
-            Some(metadata) => Ok(warp::reply::json(&MilestoneResponse {
-                milestone_index: *milestone_index,
-                message_id: message_id.to_string(),
-                timestamp: metadata.arrival_timestamp(),
-            })),
-            None => Err(reject::custom(CustomRejection::NotFound(
-                "can not find metadata for milestone".to_string(),
-            ))),
-        },
+    match tangle.get_milestone(milestone_index) {
+        Some(milestone) => Ok(warp::reply::json(&MilestoneResponse {
+            milestone_index: *milestone_index,
+            message_id: milestone.message_id().to_string(),
+            timestamp: milestone.timestamp(),
+        })),
         None => Err(reject::custom(CustomRejection::NotFound(
-            "can not find milestone".to_string(),
+            "cannot find milestone".to_string(),
         ))),
     }
 }

bee-api/bee-rest-api/src/endpoints/routes/api/v2/remove_peer.rs

@@ -28,11 +28,11 @@ pub(crate) fn filter(
         .and(warp::delete())
         .and(has_permission(ROUTE_REMOVE_PEER, public_routes, allowed_ips))
         .and(with_network_command_sender(network_command_sender))
-        .and_then(remove_peer)
+        .and_then(|peer_id, network_controller| async move { remove_peer(peer_id, network_controller) })
         .boxed()
 }
 
-pub(crate) async fn remove_peer(
+pub(crate) fn remove_peer(
     peer_id: PeerId,
     network_controller: ResourceHandle<NetworkCommandSender>,
 ) -> Result<impl Reply, Rejection> {

bee-api/bee-rest-api/src/endpoints/routes/api/v2/submit_message.rs

@@ -156,7 +156,7 @@ pub(crate) async fn submit_message<B: StorageBackend>(
         if parsed_nonce == 0 { None } else { Some(parsed_nonce) }
     };
 
-    let message = build_message(parents, payload, nonce, rest_api_config, protocol_config).await?;
+    let message = build_message(parents, payload, nonce, rest_api_config, protocol_config)?;
     let message_id = forward_to_message_submitter(message.pack_to_vec(), message_submitter).await?;
 
     Ok(warp::reply::with_status(
@@ -167,7 +167,7 @@ pub(crate) async fn submit_message<B: StorageBackend>(
     ))
 }
 
-pub(crate) async fn build_message(
+pub(crate) fn build_message(
     parents: Vec<MessageId>,
     payload: Option<Payload>,
     nonce: Option<u64>,

bee-api/bee-rest-api/src/endpoints/routes/api/v2/transaction_included_message.rs

@@ -43,11 +43,13 @@ pub(crate) fn filter<B: StorageBackend>(
         ))
         .and(with_storage(storage))
         .and(with_tangle(tangle))
-        .and_then(transaction_included_message)
+        .and_then(|transaction_id, storage, tangle| async move {
+            transaction_included_message(transaction_id, storage, tangle)
+        })
         .boxed()
 }
 
-pub(crate) async fn transaction_included_message<B: StorageBackend>(
+pub(crate) fn transaction_included_message<B: StorageBackend>(
     transaction_id: TransactionId,
     storage: ResourceHandle<B>,
     tangle: ResourceHandle<Tangle<B>>,
@@ -60,7 +62,7 @@ pub(crate) async fn transaction_included_message<B: StorageBackend>(
             "Can not fetch from storage".to_string(),
         ))
     })? {
-        Some(output) => message::message(*output.message_id(), tangle).await,
+        Some(output) => message::message(*output.message_id(), tangle),
         None => Err(reject::custom(CustomRejection::NotFound(
             "Can not find output".to_string(),
         ))),

bee-api/bee-rest-api/src/endpoints/routes/health.rs

@@ -37,22 +37,22 @@ pub(crate) fn filter<B: StorageBackend>(
         .and(has_permission(ROUTE_HEALTH, public_routes, allowed_ips))
         .and(with_tangle(tangle))
         .and(with_peer_manager(peer_manager))
-        .and_then(health)
+        .and_then(|tangle, peer_manager| async move { health(tangle, peer_manager) })
         .boxed()
 }
 
-pub(crate) async fn health<B: StorageBackend>(
+pub(crate) fn health<B: StorageBackend>(
     tangle: ResourceHandle<Tangle<B>>,
     peer_manager: ResourceHandle<PeerManager>,
 ) -> Result<impl Reply, Infallible> {
-    if is_healthy(&tangle, &peer_manager).await {
+    if is_healthy(&tangle, &peer_manager) {
         Ok(StatusCode::OK)
     } else {
         Ok(StatusCode::SERVICE_UNAVAILABLE)
     }
 }
 
-pub async fn is_healthy<B: StorageBackend>(tangle: &Tangle<B>, peer_manager: &PeerManager) -> bool {
+pub fn is_healthy<B: StorageBackend>(tangle: &Tangle<B>, peer_manager: &PeerManager) -> bool {
     if !tangle.is_confirmed_threshold(HEALTH_CONFIRMED_THRESHOLD) {
         return false;
     }
@@ -61,7 +61,7 @@ pub async fn is_healthy<B: StorageBackend>(tangle: &Tangle<B>, peer_manager: &Pe
         return false;
     }
 
-    match tangle.get_milestone(tangle.get_latest_milestone_index()).await {
+    match tangle.get_milestone(tangle.get_latest_milestone_index()) {
         Some(milestone) => {
             (SystemTime::now()
                 .duration_since(UNIX_EPOCH)