Merge pull request #38 from idky137/add_queue_worker

Server 2.0

README.md

@@ -12,22 +12,16 @@ If you believe you have discovered a security issue, please contact us at:
 [email protected]
 
 # Zingo-RPC
-Will eventually hold the rust implementations of the LightWallet Service and Darkside RPCs, along with the wallet-side and server-side Nym Service implementations.
+Will eventually hold the rust implementations of the LightWallet Service and Darkside RPCs, along with the wallet-side and server-side Nym-powered implementations.
 
 # Zingo-IndexerD
-Currently a lightweight gRPC server for testing and development. Zingo-IndexerD also has a basic nym server, currently only receives send_transaction and get_lightd_info commands send over the mixnet. 
-This should not be used to run mainnet nodes in its current form as it lacks the queueing and error checking logic necessary.
+A gRPC server capable of servicing clients both over http and over the nym mixnet. currently only send_transaction and get_lightd_info have been implemented over nym.
 
-Under the "nym_poc" feature flag Zingo-IndexerD can also act as a Nym powered indexer between zcash wallets and Zingo-IndexerD, capable of sending zcash transactions over the Nym Mixnet. 
+Under the "nym_poc" feature flag Zingo-IndexerD can also act as a Nym powered proxy, running between zcash wallets and Zingo-IndexerD, capable of sending zcash transactions over the Nym Mixnet. 
 Note: The wallet-side nym service RPC implementations are moving to CompactTxStreamerClient for easier consumption by wallets. Functionality under the "nym_poc" feature flag will be removed once a working example has been implemented directly in zingolib.
 
 This is the POC and initial work on enabling zcash infrastructure to use the nym mixnet.
 
-[Nym_POC](./docs/nym_poc.pdf) shows the current state of this work ands our vision for the future. 
-
-Our plan is to first enable wallets to send and receive transactions via a nym powered indexer between wallets and a lightwalletd/zebrad before looking at the wider zcash ecosystem.
-
-
 # Dependencies
 1) zebrad <https://github.com/ZcashFoundation/zebra.git>
 2) lightwalletd <https://github.com/zcash/lightwalletd.git> [require for testing]
@@ -41,7 +35,7 @@ Our plan is to first enable wallets to send and receive transactions via a nym p
 3) Run `$ cargo nextest run` or `$ cargo test`
 
 # zingoindexerd
-- To run zingo-cli through zingo-indexer, connecting to lightwalletd/zebrad locally:
+- To run zingo-cli through zingo-indexer, connecting to lightwalletd/zebrad locally [in seperate terminals]:
 1) Run `$ zebrad --config #PATH_TO_ZINGO_PROXY/zebrad.toml start`
 3) Run `$ cargo run`
 
@@ -50,7 +44,7 @@ From zingolib:
 
 # Nym POC
 The walletside Nym implementations are moving to ease wallet integration but the POC walletside nym server is still available under the "nym_poc" feature flag.
-- To run the POC:
+- To run the POC [in seperate terminals]:
 1) Run `$ zebrad --config #PATH_TO_ZINGO_PROXY/zebrad.toml start`
 3) Run `$ cargo run`
 4) Copy nym address displayed

integration-tests/tests/integrations.rs

@@ -1,11 +1,11 @@
-//! Integration tests for zingo-Proxy.
+//! Integration tests for zingo-Indexer.
 //! Currently uses ZCashD as ZebraD has not yet implemented Regtest Mode.
 
 #![forbid(unsafe_code)]
 
 use std::sync::{atomic::AtomicBool, Arc};
 use zingoindexer_testutils::{
-    drop_test_manager, get_zingo_address, start_zingo_mempool_monitor, ProxyPool, TestManager,
+    drop_test_manager, get_zingo_address, start_zingo_mempool_monitor, Pool, TestManager,
 };
 
 mod wallet_basic {
@@ -203,7 +203,7 @@ mod wallet_basic {
         assert_eq!(balance.sapling_balance.unwrap(), 250_000);
 
         zingo_client
-            .do_shield(&[ProxyPool::Sapling.into()], None)
+            .do_shield(&[Pool::Sapling.into()], None)
             .await
             .unwrap();
         test_manager.regtest_manager.generate_n_blocks(1).unwrap();
@@ -247,7 +247,7 @@ mod wallet_basic {
         assert_eq!(balance.transparent_balance.unwrap(), 250_000);
 
         zingo_client
-            .do_shield(&[ProxyPool::Transparent.into()], None)
+            .do_shield(&[Pool::Transparent.into()], None)
             .await
             .unwrap();
         test_manager.regtest_manager.generate_n_blocks(1).unwrap();
@@ -300,10 +300,7 @@ mod wallet_basic {
         assert_eq!(balance.transparent_balance.unwrap(), 250_000);
 
         zingo_client
-            .do_shield(
-                &[ProxyPool::Sapling.into(), ProxyPool::Transparent.into()],
-                None,
-            )
+            .do_shield(&[Pool::Sapling.into(), Pool::Transparent.into()], None)
             .await
             .unwrap();
         test_manager.regtest_manager.generate_n_blocks(1).unwrap();

zingo-indexerd/Cargo.toml

@@ -39,6 +39,9 @@ nym-sphinx-anonymous-replies = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tonic = { workspace = true }
 http = { workspace = true }
+thiserror = { workspace = true }
 
 # Miscellaneous Crate
+tower = { version = "0.4.13" }
+hyper = { version = "0.14.28", features = ["full"] } # { version = "1.4.1", features = ["full"] }
 ctrlc = "3.2.1" # "3.4.4"

zingo-indexerd/src/bin/zingoindexerd.rs

@@ -1,54 +1,8 @@
 //! Zingo-Indexer daemon
 
-use std::{
-    process,
-    sync::{
-        atomic::{AtomicBool, Ordering},
-        Arc,
-    },
-};
-use zingoindexerlib::indexer::spawn_indexer;
+use zingoindexerlib::{config::IndexerConfig, indexer::Indexer};
 
 #[tokio::main]
 async fn main() {
-    let online = Arc::new(AtomicBool::new(true));
-    let online_ctrlc = online.clone();
-    ctrlc::set_handler(move || {
-        println!("@zingoindexerd: Received Ctrl+C, exiting.");
-        online_ctrlc.store(false, Ordering::SeqCst);
-        process::exit(0);
-    })
-    .expect("Error setting Ctrl-C handler");
-
-    nym_bin_common::logging::setup_logging();
-
-    #[allow(unused_mut)]
-    let mut indexer_port: u16 = 8080;
-    #[cfg(feature = "nym_poc")]
-    {
-        indexer_port = 8088;
-    }
-
-    #[allow(unused_mut)]
-    let mut lwd_port: u16 = 9067;
-    #[cfg(feature = "nym_poc")]
-    {
-        lwd_port = 8080;
-    }
-
-    let zcashd_port: u16 = 18232;
-
-    let (_handles, _nym_address) = spawn_indexer(
-        &indexer_port,
-        &lwd_port,
-        &zcashd_port,
-        "/tmp/nym_server",
-        online.clone(),
-    )
-    .await;
-
-    let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(500));
-    while online.load(Ordering::SeqCst) {
-        interval.tick().await;
-    }
+    Indexer::start(IndexerConfig::default()).await.unwrap();
 }

zingo-indexerd/src/config.rs

@@ -0,0 +1,105 @@
+//! Zingo-Indexer config.
+
+use std::path::Path;
+
+use crate::error::IndexerError;
+
+/// Config information required for Zingo-Indexer.
+#[derive(Debug, Clone)]
+pub struct IndexerConfig {
+    /// Sets the TcpIngestor's status.
+    pub tcp_active: bool,
+    /// TcpIngestors listen port
+    pub listen_port: Option<u16>,
+    /// Sets the NymIngestor's and NymDispatchers status.
+    pub nym_active: bool,
+    /// Nym conf path used for micnet client conf.
+    pub nym_conf_path: Option<String>,
+    /// LightWalletD listen port [DEPRECATED].
+    /// Used by nym_poc and zingo-testutils.
+    pub lightwalletd_port: u16,
+    /// Full node / validator listen port.
+    pub zebrad_port: u16,
+    /// Full node Username.
+    pub node_user: Option<String>,
+    /// full node Password.
+    pub node_password: Option<String>,
+    /// Maximum requests allowed in the request queue.
+    pub max_queue_size: u16,
+    /// Maximum workers allowed in the worker pool
+    pub max_worker_pool_size: u16,
+    /// Minimum number of workers held in the workerpool when idle.
+    pub idle_worker_pool_size: u16,
+}
+
+impl IndexerConfig {
+    /// Performs checks on config data.
+    ///
+    /// - Checks that at least 1 of nym or tpc is active.
+    /// - Checks listen port is given is tcp is active.
+    /// - Checks nym_conf_path is given if nym is active and holds a valid utf8 string.
+    pub fn check_config(&self) -> Result<(), IndexerError> {
+        if (!self.tcp_active) && (!self.nym_active) {
+            return Err(IndexerError::ConfigError(
+                "Cannot start server with no ingestors selected, at least one of either nym or tcp must be set to active in conf.".to_string(),
+            ));
+        }
+        if self.tcp_active && self.listen_port.is_none() {
+            return Err(IndexerError::ConfigError(
+                "TCP is active but no address provided.".to_string(),
+            ));
+        }
+        if let Some(path_str) = self.nym_conf_path.clone() {
+            if Path::new(&path_str).to_str().is_none() {
+                return Err(IndexerError::ConfigError(
+                    "Invalid nym conf path syntax or non-UTF-8 characters in path.".to_string(),
+                ));
+            }
+        } else {
+            if self.nym_active {
+                return Err(IndexerError::ConfigError(
+                    "NYM is active but no conf path provided.".to_string(),
+                ));
+            }
+        }
+        Ok(())
+    }
+}
+
+#[cfg(not(feature = "nym_poc"))]
+impl Default for IndexerConfig {
+    fn default() -> Self {
+        Self {
+            tcp_active: true,
+            listen_port: Some(8080),
+            nym_active: true,
+            nym_conf_path: Some("/tmp/indexer/nym".to_string()),
+            lightwalletd_port: 9067,
+            zebrad_port: 18232,
+            node_user: Some("xxxxxx".to_string()),
+            node_password: Some("xxxxxx".to_string()),
+            max_queue_size: 1024,
+            max_worker_pool_size: 32,
+            idle_worker_pool_size: 4,
+        }
+    }
+}
+
+#[cfg(feature = "nym_poc")]
+impl Default for IndexerConfig {
+    fn default() -> Self {
+        Self {
+            tcp_active: true,
+            listen_port: Some(8088),
+            nym_active: false,
+            nym_conf_path: None,
+            lightwalletd_port: 8080,
+            zebrad_port: 18232,
+            node_user: Some("xxxxxx".to_string()),
+            node_password: Some("xxxxxx".to_string()),
+            max_queue_size: 1024,
+            max_worker_pool_size: 32,
+            idle_worker_pool_size: 4,
+        }
+    }
+}

zingo-indexerd/src/error.rs

@@ -0,0 +1,26 @@
+//! Hold error types for the Indexer and related functionality.
+
+use zingo_rpc::{jsonrpc::error::JsonRpcConnectorError, server::error::ServerError};
+
+/// Zingo-Indexer errors.
+#[derive(Debug, thiserror::Error)]
+pub enum IndexerError {
+    /// Server based errors.
+    #[error("Server error: {0}")]
+    ServerError(#[from] ServerError),
+    /// Configuration errors.
+    #[error("Configuration error: {0}")]
+    ConfigError(String),
+    /// JSON RPC connector errors.
+    #[error("JSON RPC connector error: {0}")]
+    JsonRpcConnectorError(#[from] JsonRpcConnectorError),
+    /// HTTP related errors due to invalid URI.
+    #[error("HTTP error: Invalid URI {0}")]
+    HttpError(#[from] http::Error),
+    /// Returned from tokio joinhandles..
+    #[error("Join handle error: Invalid URI {0}")]
+    TokioJoinError(#[from] tokio::task::JoinError),
+    /// Custom indexor errors.
+    #[error("Misc indexer error: {0}")]
+    MiscIndexerError(String),
+}