Handle critical errors

zaino-serve/src/server/grpc.rs

@@ -34,9 +34,8 @@ impl TonicServer {
     pub async fn spawn(
         service_subscriber: IndexerSubscriber<FetchServiceSubscriber>,
         server_config: GrpcConfig,
-        status: AtomicStatus,
     ) -> Result<Self, ServerError> {
-        status.store(StatusType::Spawning as usize);
+        let status = AtomicStatus::new(StatusType::Spawning.into());
 
         let svc = CompactTxStreamerServer::new(GrpcClient {
             service_subscriber: service_subscriber.clone(),
@@ -78,7 +77,7 @@ impl TonicServer {
     }
 
     /// Sets the servers to close gracefully.
-    pub async fn shutdown(&mut self) {
+    pub async fn close(&mut self) {
         self.status.store(StatusType::Closing as usize);
 
         if let Some(handle) = self.server_handle.take() {

zaino-state/src/fetch.rs

@@ -34,7 +34,7 @@ use crate::{
     indexer::{IndexerSubscriber, LightWalletIndexer, ZcashIndexer, ZcashService},
     local_cache::{BlockCache, BlockCacheSubscriber},
     mempool::{Mempool, MempoolSubscriber},
-    status::{AtomicStatus, StatusType},
+    status::StatusType,
     stream::{
         AddressStream, CompactBlockStream, CompactTransactionStream, RawTransactionStream,
         SubtreeRootReplyStream, UtxoReplyStream,
@@ -62,8 +62,6 @@ pub struct FetchService {
     data: ServiceMetadata,
     /// StateService config data.
     config: FetchServiceConfig,
-    /// Thread-safe status indicator.
-    status: AtomicStatus,
 }
 
 #[async_trait]
@@ -72,13 +70,8 @@ impl ZcashService for FetchService {
     type Subscriber = FetchServiceSubscriber;
     type Config = FetchServiceConfig;
     /// Initializes a new StateService instance and starts sync process.
-    async fn spawn(
-        config: FetchServiceConfig,
-        status: AtomicStatus,
-    ) -> Result<Self, FetchServiceError> {
+    async fn spawn(config: FetchServiceConfig) -> Result<Self, FetchServiceError> {
         info!("Launching Chain Fetch Service..");
-        let status = status.clone();
-        status.store(StatusType::Spawning.into());
 
         let fetcher = JsonRpcConnector::new_from_config_parts(
             config.validator_cookie_auth,
@@ -97,20 +90,16 @@ impl ZcashService for FetchService {
             zebra_build_data.subversion,
         );
 
-        let block_cache =
-            BlockCache::spawn(&fetcher, config.clone().into(), status.clone()).await?;
+        let block_cache = BlockCache::spawn(&fetcher, config.clone().into()).await?;
 
         let mempool = Mempool::spawn(&fetcher, None).await?;
 
-        status.store(StatusType::Ready.into());
-
         Ok(Self {
             fetcher,
             block_cache,
             mempool,
             data,
             config,
-            status,
         })
     }
 
@@ -122,18 +111,35 @@ impl ZcashService for FetchService {
             mempool: self.mempool.subscriber(),
             data: self.data.clone(),
             config: self.config.clone(),
-            status: self.status.clone(),
         })
     }
 
     /// Fetches the current status
     fn status(&self) -> StatusType {
-        self.status.load().into()
+        let mempool_status = self.mempool.status();
+        let block_cache_status = self.block_cache.status();
+
+        match (mempool_status, block_cache_status) {
+            // If either is Closing, return Closing.
+            (StatusType::Closing, _) | (_, StatusType::Closing) => StatusType::Closing,
+            // If either is Offline or CriticalError, return CriticalError.
+            (StatusType::Offline, _)
+            | (_, StatusType::Offline)
+            | (StatusType::CriticalError, _)
+            | (_, StatusType::CriticalError) => StatusType::CriticalError,
+            // If either is Spawning, return Spawning.
+            (StatusType::Spawning, _) | (_, StatusType::Spawning) => StatusType::Spawning,
+            // If either is Syncing, return Syncing.
+            (StatusType::Syncing, _) | (_, StatusType::Syncing) => StatusType::Syncing,
+            // Otherwise, return Ready.
+            _ => StatusType::Ready,
+        }
     }
 
     /// Shuts down the StateService.
     fn close(&mut self) {
         self.mempool.close();
+        self.block_cache.close();
     }
 }
 
@@ -158,14 +164,29 @@ pub struct FetchServiceSubscriber {
     data: ServiceMetadata,
     /// StateService config data.
     config: FetchServiceConfig,
-    /// Thread-safe status indicator.
-    status: AtomicStatus,
 }
 
 impl FetchServiceSubscriber {
     /// Fetches the current status
     pub fn status(&self) -> StatusType {
-        self.status.load().into()
+        let mempool_status = self.mempool.status();
+        let block_cache_status = self.block_cache.status();
+
+        match (mempool_status, block_cache_status) {
+            // If either is Closing, return Closing.
+            (StatusType::Closing, _) | (_, StatusType::Closing) => StatusType::Closing,
+            // If either is Offline or CriticalError, return CriticalError.
+            (StatusType::Offline, _)
+            | (_, StatusType::Offline)
+            | (StatusType::CriticalError, _)
+            | (_, StatusType::CriticalError) => StatusType::CriticalError,
+            // If either is Spawning, return Spawning.
+            (StatusType::Spawning, _) | (_, StatusType::Spawning) => StatusType::Spawning,
+            // If either is Syncing, return Syncing.
+            (StatusType::Syncing, _) | (_, StatusType::Syncing) => StatusType::Syncing,
+            // Otherwise, return Ready.
+            _ => StatusType::Ready,
+        }
     }
 }
 
@@ -1746,30 +1767,27 @@ mod tests {
         .await
         .unwrap();
 
-        let fetch_service = FetchService::spawn(
-            FetchServiceConfig::new(
-                test_manager.zebrad_rpc_listen_address,
-                false,
-                None,
-                None,
-                None,
-                None,
-                None,
-                None,
-                None,
-                test_manager
-                    .local_net
-                    .data_dir()
-                    .path()
-                    .to_path_buf()
-                    .join("zaino"),
-                None,
-                Network::new_regtest(Some(1), Some(1)),
-                true,
-                true,
-            ),
-            AtomicStatus::new(StatusType::Spawning.into()),
-        )
+        let fetch_service = FetchService::spawn(FetchServiceConfig::new(
+            test_manager.zebrad_rpc_listen_address,
+            false,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            test_manager
+                .local_net
+                .data_dir()
+                .path()
+                .to_path_buf()
+                .join("zaino"),
+            None,
+            Network::new_regtest(Some(1), Some(1)),
+            true,
+            true,
+        ))
         .await
         .unwrap();
         let subscriber = fetch_service.get_subscriber().inner();

zaino-state/src/indexer.rs

@@ -18,7 +18,7 @@ use zebra_rpc::methods::{
 };
 
 use crate::{
-    status::{AtomicStatus, StatusType},
+    status::StatusType,
     stream::{
         AddressStream, CompactBlockStream, CompactTransactionStream, RawTransactionStream,
         SubtreeRootReplyStream, UtxoReplyStream,
@@ -41,12 +41,9 @@ where
     Service: ZcashService,
 {
     /// Creates a new `IndexerService` using the provided `config`.
-    pub async fn spawn(
-        config: Service::Config,
-        status: AtomicStatus,
-    ) -> Result<Self, Service::Error> {
+    pub async fn spawn(config: Service::Config) -> Result<Self, Service::Error> {
         Ok(IndexerService {
-            service: Service::spawn(config, status).await?,
+            service: Service::spawn(config).await?,
         })
     }
 
@@ -74,7 +71,7 @@ pub trait ZcashService: Sized {
     type Config: Clone;
 
     /// Spawns a [`Service`].
-    async fn spawn(config: Self::Config, status: AtomicStatus) -> Result<Self, Self::Error>;
+    async fn spawn(config: Self::Config) -> Result<Self, Self::Error>;
 
     /// Returns a [`ServiceSubscriber`].
     fn get_subscriber(&self) -> IndexerSubscriber<Self::Subscriber>;

zaino-state/src/local_cache.rs

@@ -1,10 +1,6 @@
 //! Holds Zaino's local compact block cache implementation.
 
-use crate::{
-    config::BlockCacheConfig,
-    error::BlockCacheError,
-    status::{AtomicStatus, StatusType},
-};
+use crate::{config::BlockCacheConfig, error::BlockCacheError, status::StatusType};
 
 pub mod finalised_state;
 pub mod non_finalised_state;
@@ -30,31 +26,32 @@ pub struct BlockCache {
     fetcher: JsonRpcConnector,
     non_finalised_state: NonFinalisedState,
     finalised_state: Option<FinalisedState>,
+    config: BlockCacheConfig,
 }
 
 impl BlockCache {
     /// Spawns a new [`BlockCache`].
     pub async fn spawn(
         fetcher: &JsonRpcConnector,
         config: BlockCacheConfig,
-        status: AtomicStatus,
     ) -> Result<Self, BlockCacheError> {
         info!("Launching Local Block Cache..");
         let (channel_tx, channel_rx) = tokio::sync::mpsc::channel(100);
 
         let finalised_state = if !config.no_db {
-            Some(FinalisedState::spawn(fetcher, channel_rx, config.clone(), status.clone()).await?)
+            Some(FinalisedState::spawn(fetcher, channel_rx, config.clone()).await?)
         } else {
             None
         };
 
         let non_finalised_state =
-            NonFinalisedState::spawn(fetcher, channel_tx, config.clone(), status.clone()).await?;
+            NonFinalisedState::spawn(fetcher, channel_tx, config.clone()).await?;
 
         Ok(BlockCache {
             fetcher: fetcher.clone(),
             non_finalised_state,
             finalised_state,
+            config,
         })
     }
 
@@ -68,25 +65,46 @@ impl BlockCache {
             fetcher: self.fetcher.clone(),
             non_finalised_state: self.non_finalised_state.subscriber(),
             finalised_state: finalised_state_subscriber,
+            config: self.config.clone(),
         }
     }
 
-    /// Returns the status of the block cache as:
-    /// (non_finalised_state_status, finalised_state_status).
-    pub fn status(&self) -> (StatusType, StatusType) {
-        let finalised_state_status = match &self.finalised_state {
-            Some(finalised_state) => finalised_state.status(),
-            None => StatusType::Offline,
+    /// Returns the status of the block cache.
+    pub fn status(&self) -> StatusType {
+        let non_finalised_state_status = self.non_finalised_state.status();
+        let finalised_state_status = match self.config.no_db {
+            true => StatusType::Ready,
+            false => match &self.finalised_state {
+                Some(finalised_state) => finalised_state.status(),
+                None => return StatusType::Offline,
+            },
         };
 
-        (self.non_finalised_state.status(), finalised_state_status)
+        match (non_finalised_state_status, finalised_state_status) {
+            // If either is Closing, return Closing.
+            (StatusType::Closing, _) | (_, StatusType::Closing) => StatusType::Closing,
+            // If either is Offline or CriticalError, return CriticalError.
+            (StatusType::Offline, _)
+            | (_, StatusType::Offline)
+            | (StatusType::CriticalError, _)
+            | (_, StatusType::CriticalError) => StatusType::CriticalError,
+            // If either is Spawning, return Spawning.
+            (StatusType::Spawning, _) | (_, StatusType::Spawning) => StatusType::Spawning,
+            // If either is Syncing, return Syncing.
+            (StatusType::Syncing, _) | (_, StatusType::Syncing) => StatusType::Syncing,
+            // Otherwise, return Ready.
+            _ => StatusType::Ready,
+        }
     }
 
     /// Sets the block cache to close gracefully.
     pub fn close(&mut self) {
         self.non_finalised_state.close();
         if self.finalised_state.is_some() {
-            self.finalised_state.take().unwrap().close();
+            self.finalised_state
+                .take()
+                .expect("error taking Option<(Some)finalised_state> in block_cache::close")
+                .close();
         }
     }
 }
@@ -97,6 +115,7 @@ pub struct BlockCacheSubscriber {
     fetcher: JsonRpcConnector,
     non_finalised_state: NonFinalisedStateSubscriber,
     finalised_state: Option<FinalisedStateSubscriber>,
+    config: BlockCacheConfig,
 }
 
 impl BlockCacheSubscriber {
@@ -187,12 +206,31 @@ impl BlockCacheSubscriber {
     }
 
     /// Returns the status of the [`BlockCache`]..
-    pub fn status(&self) -> (StatusType, StatusType) {
-        let finalised_state_status = match &self.finalised_state {
-            Some(finalised_state) => finalised_state.status(),
-            None => StatusType::Offline,
+    pub fn status(&self) -> StatusType {
+        let non_finalised_state_status = self.non_finalised_state.status();
+        let finalised_state_status = match self.config.no_db {
+            true => StatusType::Ready,
+            false => match &self.finalised_state {
+                Some(finalised_state) => finalised_state.status(),
+                None => return StatusType::Offline,
+            },
         };
-        (self.non_finalised_state.status(), finalised_state_status)
+
+        match (non_finalised_state_status, finalised_state_status) {
+            // If either is Closing, return Closing.
+            (StatusType::Closing, _) | (_, StatusType::Closing) => StatusType::Closing,
+            // If either is Offline or CriticalError, return CriticalError.
+            (StatusType::Offline, _)
+            | (_, StatusType::Offline)
+            | (StatusType::CriticalError, _)
+            | (_, StatusType::CriticalError) => StatusType::CriticalError,
+            // If either is Spawning, return Spawning.
+            (StatusType::Spawning, _) | (_, StatusType::Spawning) => StatusType::Spawning,
+            // If either is Syncing, return Syncing.
+            (StatusType::Syncing, _) | (_, StatusType::Syncing) => StatusType::Syncing,
+            // Otherwise, return Ready.
+            _ => StatusType::Ready,
+        }
     }
 }
 
@@ -317,13 +355,9 @@ mod tests {
             no_db: zaino_no_db,
         };
 
-        let block_cache = BlockCache::spawn(
-            &json_service,
-            block_cache_config,
-            AtomicStatus::new(StatusType::Spawning.into()),
-        )
-        .await
-        .unwrap();
+        let block_cache = BlockCache::spawn(&json_service, block_cache_config)
+            .await
+            .unwrap();
 
         let block_cache_subscriber = block_cache.subscriber();