Key Image Router Service (#2898)

* Key Image Router Service

* Remove unneeded dead_code annotation

* Update fog/ledger/server/src/error.rs

Co-authored-by: Nick Santana <nick@mobilecoin.com>

* Update fog/ledger/server/src/key_image_router_service.rs

Co-authored-by: Nick Santana <nick@mobilecoin.com>

* Clean up commented-out code

Co-authored-by: Nick Santana <nick@mobilecoin.com>

* Fix misnamed type in a comment

Co-authored-by: Nick Santana <nick@mobilecoin.com>

* Address PR feedback around logging and comments.

* Address error in loop termination logic.

* Parameterize allowed number of retries for query loop

* Update based on changes from previous PRs

* Don't create 'groups' in `mod` or `use` declarations.

Co-authored-by: NotGyro <gyrocoder@gmail.com>
Co-authored-by: Nick Santana <nick@mobilecoin.com>

fog/ledger/server/src/error.rs

@@ -0,0 +1,81 @@
+// Copyright (c) 2018-2022 The MobileCoin Foundation
+
+use displaydoc::Display;
+use grpcio::RpcStatus;
+use mc_common::logger::Logger;
+use mc_fog_ledger_enclave_api::Error as LedgerEnclaveError;
+use mc_sgx_report_cache_untrusted::Error as ReportCacheError;
+use mc_util_grpc::{rpc_internal_error, rpc_permissions_error};
+
+#[derive(Debug, Display)]
+pub enum RouterServerError {
+    /// Error related to contacting Fog Ledger Store: {0}
+    LedgerStoreError(String),
+    /// Ledger Enclave error: {0}
+    Enclave(LedgerEnclaveError),
+}
+
+impl From<grpcio::Error> for RouterServerError {
+    fn from(src: grpcio::Error) -> Self {
+        RouterServerError::LedgerStoreError(format!("{}", src))
+    }
+}
+
+impl From<mc_common::ResponderIdParseError> for RouterServerError {
+    fn from(src: mc_common::ResponderIdParseError) -> Self {
+        RouterServerError::LedgerStoreError(format!("{}", src))
+    }
+}
+
+impl From<mc_util_uri::UriParseError> for RouterServerError {
+    fn from(src: mc_util_uri::UriParseError) -> Self {
+        RouterServerError::LedgerStoreError(format!("{}", src))
+    }
+}
+
+impl From<mc_util_uri::UriConversionError> for RouterServerError {
+    fn from(src: mc_util_uri::UriConversionError) -> Self {
+        RouterServerError::LedgerStoreError(format!("{}", src))
+    }
+}
+
+pub fn router_server_err_to_rpc_status(
+    context: &str,
+    src: RouterServerError,
+    logger: Logger,
+) -> RpcStatus {
+    match src {
+        RouterServerError::LedgerStoreError(_) => {
+            rpc_internal_error(context, format!("{}", src), &logger)
+        }
+        RouterServerError::Enclave(_) => {
+            rpc_permissions_error(context, format!("{}", src), &logger)
+        }
+    }
+}
+
+impl From<LedgerEnclaveError> for RouterServerError {
+    fn from(src: LedgerEnclaveError) -> Self {
+        RouterServerError::Enclave(src)
+    }
+}
+
+#[derive(Display)]
+pub enum LedgerServerError {
+    /// Ledger Enclave error: {0}
+    Enclave(LedgerEnclaveError),
+    /// Report cache error: {0}
+    ReportCache(ReportCacheError),
+}
+
+impl From<LedgerEnclaveError> for LedgerServerError {
+    fn from(src: LedgerEnclaveError) -> Self {
+        LedgerServerError::Enclave(src)
+    }
+}
+
+impl From<ReportCacheError> for LedgerServerError {
+    fn from(src: ReportCacheError) -> Self {
+        Self::ReportCache(src)
+    }
+}

fog/ledger/server/src/key_image_router_service.rs

@@ -0,0 +1,84 @@
+// Copyright (c) 2018-2022 The MobileCoin Foundation
+
+use crate::router_handlers;
+use futures::{FutureExt, TryFutureExt};
+use grpcio::{DuplexSink, RequestStream, RpcContext};
+use mc_common::logger::{log, Logger};
+use mc_fog_api::{
+    ledger::{LedgerRequest, LedgerResponse},
+    ledger_grpc::{self, LedgerApi},
+};
+use mc_fog_ledger_enclave::LedgerEnclaveProxy;
+use mc_fog_uri::KeyImageStoreUri;
+use mc_util_grpc::rpc_logger;
+use mc_util_metrics::SVC_COUNTERS;
+use std::{
+    collections::HashMap,
+    sync::{Arc, RwLock},
+};
+
+#[derive(Clone)]
+pub struct KeyImageRouterService<E>
+where
+    E: LedgerEnclaveProxy,
+{
+    enclave: E,
+    shards: Arc<RwLock<HashMap<KeyImageStoreUri, Arc<ledger_grpc::KeyImageStoreApiClient>>>>,
+    query_retries: usize,
+    logger: Logger,
+}
+
+impl<E: LedgerEnclaveProxy> KeyImageRouterService<E> {
+    /// Creates a new LedgerRouterService that can be used by a gRPC server to
+    /// fulfill gRPC requests.
+    #[allow(dead_code)] // FIXME
+    pub fn new(
+        enclave: E,
+        shards: Arc<RwLock<HashMap<KeyImageStoreUri, Arc<ledger_grpc::KeyImageStoreApiClient>>>>,
+        query_retries: usize,
+        logger: Logger,
+    ) -> Self {
+        Self {
+            enclave,
+            shards,
+            query_retries,
+            logger,
+        }
+    }
+}
+
+impl<E> LedgerApi for KeyImageRouterService<E>
+where
+    E: LedgerEnclaveProxy,
+{
+    fn request(
+        &mut self,
+        ctx: RpcContext,
+        requests: RequestStream<LedgerRequest>,
+        responses: DuplexSink<LedgerResponse>,
+    ) {
+        let _timer = SVC_COUNTERS.req(&ctx);
+        mc_common::logger::scoped_global_logger(&rpc_logger(&ctx, &self.logger), |logger| {
+            log::warn!(
+                self.logger,
+                "Streaming GRPC Ledger API only partially implemented."
+            );
+            let logger = logger.clone();
+
+            let shards = self.shards.read().expect("RwLock poisoned");
+            let future = router_handlers::handle_requests(
+                shards.values().cloned().collect(),
+                self.enclave.clone(),
+                requests,
+                responses,
+                self.query_retries,
+                logger.clone(),
+            )
+            .map_err(move |err| log::error!(&logger, "failed to reply: {}", err))
+            // TODO: Do more with the error than just push it to the log.
+            .map(|_| ());
+
+            ctx.spawn(future)
+        });
+    }
+}

fog/ledger/server/src/lib.rs

@@ -4,8 +4,11 @@ mod block_service;
 mod config;
 mod counters;
 mod db_fetcher;
+mod error;
+mod key_image_router_service;
 mod key_image_service;
 mod merkle_proof_service;
+mod router_handlers;
 mod server;
 mod untrusted_tx_out_service;