SandboxBuilder and tests for dynamic propose timeout [ECR-2033] (exon…

…um#850)

* Introduce SandboxBuilder

This commit introduces a new way to configure Sandbox for tests.
Currently most of the tests use timestamping_sandbox() function which
creates the default Sandbox environment. However, some tests may have
different needs. SandboxBuilder introduces a consistent way of
configuring the sandbox.

SandboxBuilder in its current form does not introduce any new
features, supporting only current use cases. Existing code is updated
to use the builder.

  - Function sandbox_with_services_uninitialized() has been made
    private and sandbox_with_services() has been removed.

  - Function timestamping_sandbox() remains available as it used
    almost everywhere. It still provides a default sandbox
    configuration.

  - New function timestamping_sandbox_builder() returns a builder
    with the same configuration as for timestamping_sandbox(). This
    may be useful if you only want to tweak the default sandbox.

* Tweakable ConsensusConfig in SandboxBuilder

Currently builder's implementaion is still one giant tangle of code
in sandbox_with_services_uninitialized(). This shows how we can
extract parts of the configuration and make them tweakable from
within the builder.

* Add tests for expedited Proposal injection

Here is main course of this patch set. We add some tests to verify
that Proposals are actually injected with minimal timeout if the
transaction pool goes over the threshold. For this we will need to
tweak the default sandbox configuration so that proposal values
make more sense. However, we must take care to not touch the rest
of tests which rely on the default behavior.

* Reformat code according to cargo +1.27.0 fmt

We're still using 'oldstable' version of Rust on the CI server and
it expects the code to be formatted in accordance with 1.27 which
is different from the new 1.28 format.

exonum/src/sandbox/consensus/recovery.rs

@@ -23,7 +23,7 @@ use messages::{Connect, PeersRequest, Precommit, Prevote};
 use node;
 
 use sandbox::{
-    sandbox::{sandbox_with_services_uninitialized, timestamping_sandbox}, sandbox_tests_helper::*,
+    sandbox::{timestamping_sandbox, SandboxBuilder}, sandbox_tests_helper::*,
 };
 
 #[test]
@@ -432,7 +432,9 @@ fn test_recover_consensus_messages_in_other_round() {
 #[test]
 fn should_restore_peers_after_restart() {
     // create sandbox with nodes not aware about each other
-    let sandbox = sandbox_with_services_uninitialized(vec![]);
+    let sandbox = SandboxBuilder::new()
+        .do_not_initialize_connections()
+        .build();
 
     let (v0, v1) = (ValidatorId(0), ValidatorId(1));
     let (p0, s0, a0) = (sandbox.p(v0), sandbox.s(v0).clone(), sandbox.a(v0));

exonum/src/sandbox/consensus/transactions.rs

@@ -18,18 +18,44 @@ use bit_vec::BitVec;
 
 use std::time::Duration;
 
-use crypto::{gen_keypair, CryptoHash};
-use helpers::{Height, Round, ValidatorId};
+use crypto::{gen_keypair, CryptoHash, Hash};
+use helpers::{Height, Milliseconds, Round, ValidatorId};
 use messages::{
     Message, Precommit, Prevote, PrevotesRequest, ProposeRequest, TransactionsRequest,
     TransactionsResponse,
 };
 use node::state::TRANSACTIONS_REQUEST_TIMEOUT;
 use sandbox::{
-    config_updater::TxConfig, sandbox::timestamping_sandbox, sandbox_tests_helper::*,
-    timestamping::{TimestampingTxGenerator, DATA_SIZE},
+    config_updater::TxConfig,
+    sandbox::{timestamping_sandbox, timestamping_sandbox_builder, Sandbox},
+    sandbox_tests_helper::*, timestamping::{TimestampTx, TimestampingTxGenerator, DATA_SIZE},
 };
 
+const MAX_PROPOSE_TIMEOUT: Milliseconds = 200;
+const MIN_PROPOSE_TIMEOUT: Milliseconds = 10;
+const PROPOSE_THRESHOLD: u32 = 3;
+
+fn timestamping_sandbox_with_threshold() -> Sandbox {
+    let sandbox = timestamping_sandbox_builder()
+        .with_consensus(|config| {
+            config.max_propose_timeout = MAX_PROPOSE_TIMEOUT;
+            config.min_propose_timeout = MIN_PROPOSE_TIMEOUT;
+            config.propose_timeout_threshold = PROPOSE_THRESHOLD;
+        })
+        .build();
+
+    // Wait for us to become the leader.
+    sandbox.add_time(Duration::from_millis(sandbox.round_timeout()));
+    sandbox.add_time(Duration::from_millis(sandbox.round_timeout()));
+    sandbox
+}
+
+fn tx_hashes(transactions: &[TimestampTx]) -> Vec<Hash> {
+    let mut hashes = transactions.iter().map(|tx| tx.hash()).collect::<Vec<_>>();
+    hashes.sort();
+    hashes
+}
+
 /// idea of the test is to verify request transaction scenario: other node requests
 /// transaction from our node
 #[test]
@@ -554,3 +580,51 @@ fn request_txs_when_get_propose_or_prevote() {
 
     sandbox.add_time(Duration::from_millis(0));
 }
+
+#[test]
+fn regular_propose_when_no_transaction_pressure() {
+    let sandbox = timestamping_sandbox_with_threshold();
+
+    // Generate and receive some transactions (fewer than the threshold).
+    let transactions = TimestampingTxGenerator::new(64)
+        .take(PROPOSE_THRESHOLD as usize - 1)
+        .collect::<Vec<_>>();
+
+    for tx in &transactions {
+        sandbox.recv(tx);
+    }
+
+    // Proposal is expected to arrive after maximum timeout as we're still not over the threshold.
+    sandbox.add_time(Duration::from_millis(MAX_PROPOSE_TIMEOUT));
+
+    let propose = ProposeBuilder::new(&sandbox)
+        .with_tx_hashes(&tx_hashes(&transactions))
+        .build();
+
+    sandbox.broadcast(&propose);
+    sandbox.broadcast(&make_prevote_from_propose(&sandbox, &propose));
+}
+
+#[test]
+fn expedited_propose_on_transaction_pressure() {
+    let sandbox = timestamping_sandbox_with_threshold();
+
+    // Generate and receive some transactions (at the threshold).
+    let transactions = TimestampingTxGenerator::new(64)
+        .take(PROPOSE_THRESHOLD as usize)
+        .collect::<Vec<_>>();
+
+    for tx in &transactions {
+        sandbox.recv(tx);
+    }
+
+    // Proposal should be expedited and is expected to arrive after minimum timeout.
+    sandbox.add_time(Duration::from_millis(MIN_PROPOSE_TIMEOUT));
+
+    let propose = ProposeBuilder::new(&sandbox)
+        .with_tx_hashes(&tx_hashes(&transactions))
+        .build();
+
+    sandbox.broadcast(&propose);
+    sandbox.broadcast(&make_prevote_from_propose(&sandbox, &propose));
+}

exonum/src/sandbox/sandbox.rs

@@ -704,22 +704,68 @@ impl ConnectList {
     }
 }
 
+pub struct SandboxBuilder {
+    initialize: bool,
+    services: Vec<Box<dyn Service>>,
+    consensus_config: ConsensusConfig,
+}
+
+impl SandboxBuilder {
+    pub fn new() -> Self {
+        SandboxBuilder {
+            initialize: true,
+            services: Vec::new(),
+            consensus_config: ConsensusConfig {
+                round_timeout: 1000,
+                status_timeout: 600_000,
+                peers_timeout: 600_000,
+                txs_block_limit: 1000,
+                max_message_len: 1024 * 1024,
+                min_propose_timeout: PROPOSE_TIMEOUT,
+                max_propose_timeout: PROPOSE_TIMEOUT,
+                propose_timeout_threshold: std::u32::MAX,
+            },
+        }
+    }
+
+    pub fn do_not_initialize_connections(mut self) -> Self {
+        self.initialize = false;
+        self
+    }
+
+    pub fn with_services(mut self, services: Vec<Box<dyn Service>>) -> Self {
+        self.services = services;
+        self
+    }
+
+    pub fn with_consensus<F: FnOnce(&mut ConsensusConfig)>(mut self, update: F) -> Self {
+        update(&mut self.consensus_config);
+        self
+    }
+
+    pub fn build(self) -> Sandbox {
+        let mut sandbox = sandbox_with_services_uninitialized(self.services, self.consensus_config);
+
+        if self.initialize {
+            let time = sandbox.time();
+            let validators_count = sandbox.validators_map.len();
+            sandbox.initialize(time, 1, validators_count);
+        }
+
+        sandbox
+    }
+}
+
 fn gen_primitive_socket_addr(idx: u8) -> SocketAddr {
     let addr = Ipv4Addr::new(idx, idx, idx, idx);
     SocketAddr::new(IpAddr::V4(addr), u16::from(idx))
 }
 
-/// Constructs an instance of a `Sandbox` and initializes connections.
-pub fn sandbox_with_services(services: Vec<Box<dyn Service>>) -> Sandbox {
-    let mut sandbox = sandbox_with_services_uninitialized(services);
-    let time = sandbox.time();
-    let validators_count = sandbox.validators_map.len();
-    sandbox.initialize(time, 1, validators_count);
-    sandbox
-}
-
 /// Constructs an uninitialized instance of a `Sandbox`.
-pub fn sandbox_with_services_uninitialized(services: Vec<Box<dyn Service>>) -> Sandbox {
+fn sandbox_with_services_uninitialized(
+    services: Vec<Box<dyn Service>>,
+    consensus: ConsensusConfig,
+) -> Sandbox {
     let validators = vec![
         gen_keypair_from_seed(&Seed::new([12; SEED_LENGTH])),
         gen_keypair_from_seed(&Seed::new([13; SEED_LENGTH])),
@@ -745,16 +791,6 @@ pub fn sandbox_with_services_uninitialized(services: Vec<Box<dyn Service>>) -> S
         ApiSender::new(api_channel.0.clone()),
     );
 
-    let consensus = ConsensusConfig {
-        round_timeout: 1000,
-        status_timeout: 600_000,
-        peers_timeout: 600_000,
-        txs_block_limit: 1000,
-        max_message_len: 1024 * 1024,
-        min_propose_timeout: PROPOSE_TIMEOUT,
-        max_propose_timeout: PROPOSE_TIMEOUT,
-        propose_timeout_threshold: std::u32::MAX,
-    };
     let genesis = GenesisConfig::new_with_consensus(
         consensus,
         validators
@@ -840,7 +876,11 @@ pub fn sandbox_with_services_uninitialized(services: Vec<Box<dyn Service>>) -> S
 
 pub fn timestamping_sandbox() -> Sandbox {
     let _ = env_logger::try_init();
-    sandbox_with_services(vec![
+    timestamping_sandbox_builder().build()
+}
+
+pub fn timestamping_sandbox_builder() -> SandboxBuilder {
+    SandboxBuilder::new().with_services(vec![
         Box::new(TimestampingService::new()),
         Box::new(ConfigUpdateService::new()),
     ])
@@ -1086,10 +1126,12 @@ mod tests {
 
     #[test]
     fn test_sandbox_service_after_commit() {
-        let sandbox = sandbox_with_services(vec![
-            Box::new(AfterCommitService),
-            Box::new(TimestampingService::new()),
-        ]);
+        let sandbox = SandboxBuilder::new()
+            .with_services(vec![
+                Box::new(AfterCommitService),
+                Box::new(TimestampingService::new()),
+            ])
+            .build();
         let state = SandboxState::new();
         add_one_height(&sandbox, &state);
         let tx = TxAfterCommit::new_with_height(Height(1));