Create FVR integration tests

fog/view/server/Cargo.toml

@@ -73,7 +73,7 @@ mc-util-serial = { path = "../../../util/serial" }
 mc-util-test-helper = { path = "../../../util/test-helper" }
 mc-util-uri = { path = "../../../util/uri" }
 pem = "1.0"
-portpicker = "0.1.1"
 rand = "0.8"
 rand_core = "0.6"
+portpicker = "0.1.1"
 tempdir = "0.3"

fog/view/server/test-utils/src/lib.rs

@@ -9,9 +9,13 @@ use mc_common::{logger::Logger, time::SystemTimeProvider, ResponderId};
 use mc_fog_api::view_grpc::FogViewStoreApiClient;
 use mc_fog_sql_recovery_db::{test_utils::SqlRecoveryDbTestContext, SqlRecoveryDb};
 use mc_fog_test_infra::get_enclave_path;
-use mc_fog_types::common::BlockRange;
+use mc_fog_types::{
+    common::BlockRange,
+    view::{QueryResponse, TxOutSearchResult, TxOutSearchResultCode},
+    ETxOutRecord,
+};
 use mc_fog_uri::{FogViewRouterAdminUri, FogViewRouterUri, FogViewStoreUri};
-use mc_fog_view_connection::fog_view_router_client::FogViewRouterGrpcClient;
+use mc_fog_view_connection::fog_view_router_client::{Error, FogViewRouterGrpcClient};
 use mc_fog_view_enclave::SgxViewEnclave;
 use mc_fog_view_server::{
     config::{
@@ -25,9 +29,11 @@ use mc_fog_view_server::{
 use mc_util_grpc::ConnectionUriGrpcioChannel;
 use mc_util_uri::ConnectionUri;
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     str::FromStr,
     sync::{Arc, RwLock},
+    thread::sleep,
+    time::Duration,
 };
 
 /// Contains the core structs used by router integration tests and manages their
@@ -217,3 +223,60 @@ impl Drop for RouterTestEnvironment {
         self.db_test_context = None;
     }
 }
+
+/// Ensure that all provided ETxOutRecords are in the enclave, and that
+/// non-existing ones aren't.
+pub async fn assert_e_tx_out_records(
+    client: &mut FogViewRouterGrpcClient,
+    records: &[ETxOutRecord],
+) -> Result<QueryResponse, Error> {
+    // Construct an array of expected results that includes both records we expect
+    // to find and records we expect not to find.
+    let mut expected_results = HashSet::new();
+    for record in records {
+        expected_results.insert(TxOutSearchResult {
+            search_key: record.search_key.clone(),
+            result_code: TxOutSearchResultCode::Found as u32,
+            ciphertext: record.payload.clone(),
+        });
+    }
+
+    let search_keys: Vec<_> = expected_results
+        .iter()
+        .map(|result| result.search_key.clone())
+        .collect();
+
+    let mut allowed_tries = 60usize;
+    loop {
+        let result = client.query(0, 0, search_keys.clone()).await.unwrap();
+
+        let actual_tx_out_search_results: HashSet<TxOutSearchResult> =
+            interpret_tx_out_search_results(&result.tx_out_search_results);
+        if actual_tx_out_search_results == expected_results {
+            return Ok(result);
+        }
+        if allowed_tries == 0 {
+            panic!("Server did not catch up to database!");
+        }
+        allowed_tries -= 1;
+        sleep(Duration::from_millis(1000));
+    }
+}
+
+/// Takes the TxOutSearchResults and interprets the ciphertext as follows:
+/// (1) The first byte is the delta between the max payload length and the
+/// length of this payload.
+/// (2) The rest of the bytes for that length are part of the ciphertext.
+fn interpret_tx_out_search_results(results: &[TxOutSearchResult]) -> HashSet<TxOutSearchResult> {
+    results
+        .iter()
+        .map(|result| {
+            let payload_length = result.ciphertext.len() - (result.ciphertext[0] as usize);
+            TxOutSearchResult {
+                search_key: result.search_key[..].to_vec(),
+                result_code: result.result_code,
+                ciphertext: result.ciphertext[1..payload_length + 1].to_owned(),
+            }
+        })
+        .collect()
+}

fog/view/server/tests/router_smoke_tests.rs

@@ -0,0 +1,124 @@
+// Copyright (c) 2018-2022 The MobileCoin Foundation
+
+use futures::executor::block_on;
+use mc_common::logger::{log, Logger};
+use mc_crypto_keys::{CompressedRistrettoPublic, RistrettoPublic};
+use mc_fog_kex_rng::KexRngPubkey;
+use mc_fog_recovery_db_iface::{RecoveryDb, ReportData, ReportDb};
+use mc_fog_test_infra::db_tests::random_block;
+use mc_fog_view_server_test_utils::RouterTestEnvironment;
+use mc_util_from_random::FromRandom;
+use rand::{rngs::StdRng, SeedableRng};
+use std::{thread::sleep, time::Duration};
+
+async fn test_router_integration(test_environment: &mut RouterTestEnvironment, logger: Logger) {
+    let store_count = 5;
+    let mut rng: StdRng = SeedableRng::from_seed([123u8; 32]);
+    let db = test_environment
+        .db_test_context
+        .as_ref()
+        .unwrap()
+        .get_db_instance();
+
+    let ingress_key = CompressedRistrettoPublic::from(RistrettoPublic::from_random(&mut rng));
+    let accepted_block_1 = db.new_ingress_key(&ingress_key, 0).unwrap();
+    assert_eq!(accepted_block_1, 0);
+
+    db.set_report(
+        &ingress_key,
+        "",
+        &ReportData {
+            pubkey_expiry: 6,
+            ingest_invocation_id: None,
+            report: Default::default(),
+        },
+    )
+    .unwrap();
+
+    let total_block_count = store_count;
+    let egress_public_key = KexRngPubkey {
+        public_key: [1; 32].to_vec(),
+        version: 0,
+    };
+
+    let invoc_id1 = db
+        .new_ingest_invocation(None, &ingress_key, &egress_public_key, 0)
+        .unwrap();
+
+    let mut expected_records = Vec::new();
+    for i in 0..total_block_count {
+        let (block, records) = random_block(&mut rng, i as u64, 2);
+        db.add_block_data(&invoc_id1, &block, 0, &records).unwrap();
+        expected_records.extend(records);
+    }
+
+    // Wait until first server has added stuff to ORAM. Since all view servers
+    // should load ORAM at the same time, we could choose to wait for any view
+    // server.
+    let mut allowed_tries = 1000usize;
+    loop {
+        let db_num_blocks = db
+            .get_highest_known_block_index()
+            .unwrap()
+            .map(|v| v + 1) // convert index to count
+            .unwrap_or(0);
+        let server_num_blocks = test_environment
+            .store_servers
+            .as_ref()
+            .unwrap()
+            .iter()
+            .map(|server| server.highest_processed_block_count())
+            .max()
+            .unwrap_or_default();
+        if server_num_blocks > db_num_blocks {
+            panic!(
+                "Server num blocks should never be larger than db num blocks: {} > {}",
+                server_num_blocks, db_num_blocks
+            );
+        }
+        if server_num_blocks == db_num_blocks {
+            log::info!(logger, "Stopping, block {}", server_num_blocks);
+            break;
+        }
+        log::info!(
+            logger,
+            "Waiting for server to catch up to db... {} < {}",
+            server_num_blocks,
+            db_num_blocks
+        );
+        if allowed_tries == 0 {
+            panic!("Server did not catch up to database!");
+        }
+        allowed_tries -= 1;
+        sleep(Duration::from_secs(1));
+    }
+
+    let router_client = test_environment.router_client.as_mut().unwrap();
+    let result =
+        mc_fog_view_server_test_utils::assert_e_tx_out_records(router_client, &expected_records)
+            .await;
+
+    assert!(result.is_ok());
+
+    let result = result.unwrap();
+    // TODO: see what we should do about collating these fields... Right now they
+    // are in sync, but that won't always be the case...
+    assert_eq!(result.highest_processed_block_count, 5);
+    assert_eq!(result.next_start_from_user_event_id, 1);
+    assert_eq!(result.rng_records.len(), 1);
+    assert_eq!(result.rng_records[0].pubkey, egress_public_key);
+    assert_eq!(result.missed_block_ranges.len(), 0);
+    assert_eq!(result.last_known_block_count, 5);
+}
+
+#[test]
+fn test_1000() {
+    let (logger, _global_logger_guard) =
+        mc_common::logger::create_app_logger(mc_common::logger::o!());
+    let mut test_environment = RouterTestEnvironment::new(1000, 5, logger.clone());
+
+    block_on(test_router_integration(
+        &mut test_environment,
+        logger.clone(),
+    ))
+}