Merge #2194: Add is_synced flag to BaseNodeServiceResponse

This PR adds a Boolean flag to all BaseNodeServiceResponse’s so that the node
querying can know if the queried node is synced to the chain tip. The wallet is
then updated to abort a UTXO validation operation if this flag is false. This
is because if a node is not synced it could incorrectly declare UTXO invalid
that it is not yet aware of.

To do this the Base Node Service needed to know whether the Base Node State
machine had synced at least once and was currently in the Listening state. To
allow the Base Node Service to speak to the Base Node State Machine the state
machine had to be refactored into a Service with a Handle that could be
registered in the Service Stack.

Additional the base node was updated to not process new blocks or transactions
until it had synced at least once. This determination was done using the flag
made available via the new BaseNodeStateMachineHandle.

applications/tari_base_node/src/miner.rs

@@ -23,7 +23,7 @@
 
 use tari_broadcast_channel::Subscriber;
 use tari_core::{
-    base_node::{states::StateEvent, LocalNodeCommsInterface},
+    base_node::{state_machine_service::states::StateEvent, LocalNodeCommsInterface},
     consensus::ConsensusManager,
     mempool::MempoolStateEvent,
     mining::Miner,

applications/tari_base_node/src/parser.rs

@@ -22,14 +22,14 @@
 
 use super::LOG_TARGET;
 use crate::{
-    builder::NodeContainer,
+    builder::BaseNodeContext,
     table::Table,
     utils,
     utils::{format_duration_basic, format_naive_datetime},
 };
 use chrono::Utc;
 use chrono_english::{parse_date_string, Dialect};
-use futures::{future::Either, StreamExt};
+use futures::future::Either;
 use log::*;
 use qrcode::{render::unicode, QrCode};
 use regex::Regex;
@@ -53,7 +53,6 @@ use std::{
 };
 use strum::IntoEnumIterator;
 use strum_macros::{Display, EnumIter, EnumString};
-use tari_broadcast_channel::Subscriber;
 use tari_common::GlobalConfig;
 use tari_comms::{
     connection_manager::ConnectionManagerRequester,
@@ -64,7 +63,7 @@ use tari_comms::{
 };
 use tari_comms_dht::{envelope::NodeDestination, DhtDiscoveryRequester};
 use tari_core::{
-    base_node::{states::StatusInfo, LocalNodeCommsInterface},
+    base_node::{state_machine_service::states::StatusInfo, LocalNodeCommsInterface},
     blocks::BlockHeader,
     mempool::service::LocalMempoolService,
     mining::MinerInstruction,
@@ -82,7 +81,11 @@ use tari_wallet::{
     transaction_service::{error::TransactionServiceError, handle::TransactionServiceHandle},
     util::emoji::EmojiId,
 };
-use tokio::{runtime, sync::broadcast::Sender as syncSender, time};
+use tokio::{
+    runtime,
+    sync::{broadcast::Sender as syncSender, watch},
+    time,
+};
 
 /// Enum representing commands used by the basenode
 #[derive(Clone, PartialEq, Debug, Display, EnumIter, EnumString)]
@@ -145,7 +148,7 @@ pub struct Parser {
     miner_hashrate: Arc<AtomicU64>,
     miner_instructions: syncSender<MinerInstruction>,
     miner_thread_count: u64,
-    state_machine_info: Subscriber<StatusInfo>,
+    state_machine_info: watch::Receiver<StatusInfo>,
 }
 
 // Import the auto-generated const values from the Manifest and Git
@@ -184,7 +187,7 @@ impl Hinter for Parser {
 
 impl Parser {
     /// creates a new parser struct
-    pub fn new(executor: runtime::Handle, ctx: &NodeContainer, config: &GlobalConfig) -> Self {
+    pub fn new(executor: runtime::Handle, ctx: &BaseNodeContext, config: &GlobalConfig) -> Self {
         Parser {
             executor,
             wallet_node_identity: ctx.wallet_node_identity(),
@@ -511,12 +514,9 @@ impl Parser {
 
     /// Function to process the get-state-info command
     fn process_state_info(&mut self) {
-        // the channel only holds events of 1 as the channel is created bounded(1)
         let mut channel = self.state_machine_info.clone();
-        // We clone the channel so that allows as to always start to read from the beginning. Hence the channel never
-        // empties.
         self.executor.spawn(async move {
-            match channel.next().await {
+            match channel.recv().await {
                 None => {
                     info!(
                         target: LOG_TARGET,

base_layer/core/src/base_node/chain_metadata_service/error.rs

@@ -24,6 +24,7 @@ use crate::base_node::comms_interface::CommsInterfaceError;
 use prost::DecodeError;
 use tari_comms::message::MessageError;
 use tari_p2p::services::liveness::error::LivenessError;
+use tari_service_framework::reply_channel::TransportChannelError;
 use thiserror::Error;
 
 #[derive(Debug, Error)]

base_layer/core/src/base_node/chain_metadata_service/service.rs

@@ -29,10 +29,10 @@ use crate::{
     },
     chain_storage::{BlockAddResult, ChainMetadata},
 };
-use chrono::{NaiveDateTime, Utc};
 use futures::{stream::StreamExt, SinkExt};
 use log::*;
 use prost::Message;
+use std::time::Instant;
 use tari_broadcast_channel::Publisher;
 use tari_common::log_if_error;
 use tari_comms::{message::MessageExt, peer_manager::NodeId};
@@ -42,7 +42,7 @@ pub(super) struct ChainMetadataService {
     liveness: LivenessHandle,
     base_node: LocalNodeCommsInterface,
     peer_chain_metadata: Vec<PeerChainMetadata>,
-    last_chainstate_flushed_at: NaiveDateTime,
+    last_chainstate_flushed_at: Option<Instant>,
     event_publisher: Publisher<ChainMetadataEvent>,
 }
 
@@ -62,7 +62,7 @@ impl ChainMetadataService {
             liveness,
             base_node,
             peer_chain_metadata: Vec::new(),
-            last_chainstate_flushed_at: Utc::now().naive_utc(),
+            last_chainstate_flushed_at: None,
             event_publisher,
         }
     }
@@ -186,7 +186,7 @@ impl ChainMetadataService {
             .await
             .map_err(|_| ChainMetadataSyncError::EventPublishFailed)?;
 
-        self.last_chainstate_flushed_at = Utc::now().naive_utc();
+        self.last_chainstate_flushed_at = Some(Instant::now());
 
         Ok(())
     }

base_layer/core/src/base_node/mod.rs

@@ -41,9 +41,7 @@ pub mod consts;
 #[cfg(feature = "base_node")]
 pub mod service;
 #[cfg(feature = "base_node")]
-mod state_machine;
-#[cfg(feature = "base_node")]
-pub mod states;
+pub mod state_machine_service;
 #[cfg(feature = "base_node")]
 mod validators;
 #[cfg(feature = "base_node")]
@@ -53,7 +51,7 @@ pub use validators::{ChainBalanceValidator, HeaderValidator, SyncValidators};
 #[cfg(feature = "base_node")]
 pub use comms_interface::{LocalNodeCommsInterface, OutboundNodeCommsInterface};
 #[cfg(feature = "base_node")]
-pub use state_machine::{BaseNodeStateMachine, BaseNodeStateMachineConfig};
+pub use state_machine_service::{BaseNodeStateMachine, BaseNodeStateMachineConfig, StateMachineHandle};
 
 #[cfg(any(feature = "base_node", feature = "base_node_proto"))]
 pub mod proto;

base_layer/core/src/base_node/proto/response.proto

@@ -33,6 +33,7 @@ message BaseNodeServiceResponse {
         // Indicates a MmrNodes response
         MmrNodes MmrNodes = 12;
     }
+    bool is_synced = 13;
 }
 
 message BlockHeaders {

base_layer/core/src/base_node/service/initializer.rs

@@ -25,6 +25,7 @@ use crate::{
         comms_interface::{InboundNodeCommsHandlers, LocalNodeCommsInterface, OutboundNodeCommsInterface},
         proto,
         service::service::{BaseNodeService, BaseNodeServiceConfig, BaseNodeStreams},
+        StateMachineHandle,
     },
     blocks::NewBlock,
     chain_storage::{BlockchainBackend, BlockchainDatabase},
@@ -189,6 +190,10 @@ where T: BlockchainBackend + 'static
                 .get_handle::<OutboundMessageRequester>()
                 .expect("OutboundMessageRequester handle required for BaseNodeService");
 
+            let state_machine = handles
+                .get_handle::<StateMachineHandle>()
+                .expect("StateMachineHandle required to initialize MempoolService");
+
             let streams = BaseNodeStreams::new(
                 outbound_request_stream,
                 outbound_block_stream,
@@ -198,7 +203,8 @@ where T: BlockchainBackend + 'static
                 local_request_stream,
                 local_block_stream,
             );
-            let service = BaseNodeService::new(outbound_message_service, inbound_nch, config).start(streams);
+            let service =
+                BaseNodeService::new(outbound_message_service, inbound_nch, config, state_machine).start(streams);
             futures::pin_mut!(service);
             future::select(service, shutdown).await;
             info!(target: LOG_TARGET, "Base Node Service shutdown");

base_layer/core/src/base_node/service/service.rs

@@ -33,7 +33,9 @@ use crate::{
         generate_request_key,
         proto,
         service::error::BaseNodeServiceError,
+        state_machine_service::states::StatusInfo,
         RequestKey,
+        StateMachineHandle,
         WaitingRequests,
     },
     blocks::{Block, NewBlock},
@@ -148,6 +150,7 @@ pub struct BaseNodeService<B> {
     timeout_sender: Sender<RequestKey>,
     timeout_receiver_stream: Option<Receiver<RequestKey>>,
     config: BaseNodeServiceConfig,
+    state_machine_handle: StateMachineHandle,
 }
 
 impl<B> BaseNodeService<B>
@@ -157,6 +160,7 @@ where B: BlockchainBackend + 'static
         outbound_message_service: OutboundMessageRequester,
         inbound_nch: InboundNodeCommsHandlers<B>,
         config: BaseNodeServiceConfig,
+        state_machine_handle: StateMachineHandle,
     ) -> Self
     {
         let (timeout_sender, timeout_receiver) = channel(100);
@@ -167,6 +171,7 @@ where B: BlockchainBackend + 'static
             timeout_sender,
             timeout_receiver_stream: Some(timeout_receiver),
             config,
+            state_machine_handle,
         }
     }
 
@@ -233,7 +238,7 @@ where B: BlockchainBackend + 'static
 
                 // Incoming block messages from the Comms layer
                 block_msg = inbound_block_stream.select_next_some() => {
-                    self.spawn_handle_incoming_block(block_msg);
+                    self.spawn_handle_incoming_block(block_msg).await;
                 }
 
                 // Incoming local request messages from the LocalNodeCommsInterface and other local services
@@ -304,8 +309,9 @@ where B: BlockchainBackend + 'static
     fn spawn_handle_incoming_request(&self, domain_msg: DomainMessage<proto::base_node::BaseNodeServiceRequest>) {
         let inbound_nch = self.inbound_nch.clone();
         let outbound_message_service = self.outbound_message_service.clone();
+        let state_machine_handle = self.state_machine_handle.clone();
         task::spawn(async move {
-            let _ = handle_incoming_request(inbound_nch, outbound_message_service, domain_msg)
+            let _ = handle_incoming_request(inbound_nch, outbound_message_service, state_machine_handle, domain_msg)
                 .await
                 .or_else(|err| {
                     error!(
@@ -344,7 +350,27 @@ where B: BlockchainBackend + 'static
         });
     }
 
-    fn spawn_handle_incoming_block(&self, new_block: DomainMessage<NewBlock>) {
+    async fn spawn_handle_incoming_block(&self, new_block: DomainMessage<NewBlock>) {
+        // Determine if we are bootstrapped
+        let mut status_watch = self.state_machine_handle.get_status_info_watch();
+
+        let bootstrapped = match status_watch.recv().await {
+            None => false,
+            Some(s) => match s {
+                StatusInfo::Listening(li) => li.is_bootstrapped(),
+                _ => false,
+            },
+        };
+
+        if !bootstrapped {
+            debug!(
+                target: LOG_TARGET,
+                "Propagated block `{}` from peer `{}` not processed while busy with initial sync.",
+                new_block.inner.block_hash.to_hex(),
+                new_block.source_peer.node_id.short_str(),
+            );
+            return;
+        }
         let inbound_nch = self.inbound_nch.clone();
         task::spawn(async move {
             let _ = handle_incoming_block(inbound_nch, new_block).await.or_else(|err| {
@@ -398,6 +424,7 @@ where B: BlockchainBackend + 'static
 async fn handle_incoming_request<B: BlockchainBackend + 'static>(
     inbound_nch: InboundNodeCommsHandlers<B>,
     mut outbound_message_service: OutboundMessageRequester,
+    state_machine_handle: StateMachineHandle,
     domain_request_msg: DomainMessage<proto::BaseNodeServiceRequest>,
 ) -> Result<(), BaseNodeServiceError>
 {
@@ -412,16 +439,28 @@ async fn handle_incoming_request<B: BlockchainBackend + 'static>(
         .handle_request(&request.try_into().map_err(BaseNodeServiceError::InvalidRequest)?)
         .await?;
 
+    // Determine if we are synced
+    let mut status_watch = state_machine_handle.get_status_info_watch();
+    let is_synced = match status_watch.recv().await {
+        None => false,
+        Some(s) => match s {
+            StatusInfo::Listening(li) => li.is_synced(),
+            _ => false,
+        },
+    };
+
     let message = proto::BaseNodeServiceResponse {
         request_key: inner_msg.request_key,
         response: Some(response.into()),
+        is_synced,
     };
 
     trace!(
         target: LOG_TARGET,
         "Attempting outbound message in response to inbound request ({})",
         inner_msg.request_key
     );
+
     let send_message_response = outbound_message_service
         .send_direct(
             origin_public_key,
@@ -472,18 +511,23 @@ async fn handle_incoming_response(
     incoming_response: proto::BaseNodeServiceResponse,
 ) -> Result<(), BaseNodeServiceError>
 {
-    let proto::BaseNodeServiceResponse { request_key, response } = incoming_response;
+    let proto::BaseNodeServiceResponse {
+        request_key,
+        response,
+        is_synced,
+    } = incoming_response;
     let response: NodeCommsResponse = response
         .and_then(|r| r.try_into().ok())
         .ok_or_else(|| BaseNodeServiceError::InvalidResponse("Received an invalid base node response".to_string()))?;
 
     if let Some((reply_tx, started)) = waiting_requests.remove(request_key).await {
         trace!(
             target: LOG_TARGET,
-            "Response for {} (request key: {}) received after {}ms",
+            "Response for {} (request key: {}) received after {}ms and is_synced: {}",
             response,
             &request_key,
-            started.elapsed().as_millis()
+            started.elapsed().as_millis(),
+            is_synced
         );
         let _ = reply_tx.send(Ok(response).or_else(|resp| {
             warn!(

base_layer/core/src/base_node/service/service_response.rs

@@ -28,4 +28,5 @@ use serde::{Deserialize, Serialize};
 pub struct BaseNodeServiceResponse {
     pub request_key: RequestKey,
     pub response: NodeCommsResponse,
+    pub is_synced: bool,
 }