Pipelining (near#12015)

This reverts commit 4e93e46.
This reverts commit 86cd45b.

To address the issues seen that led to the functionality being reverted
in the first place I had to add some mechanisms to query The State in a
side-effect-free (pure) manner and to also revert some changes from the
original change-set where we the new code would *not* produce the
side-effects that were expected by the other nodes.

core/store/src/contract.rs

@@ -0,0 +1,52 @@
+use crate::TrieStorage;
+use near_primitives::hash::CryptoHash;
+use near_vm_runner::ContractCode;
+use std::collections::BTreeMap;
+use std::sync::{Arc, Mutex};
+
+/// Reads contract code from the trie by its hash.
+///
+/// Cloning is cheap.
+#[derive(Clone)]
+pub struct ContractStorage {
+    storage: Arc<dyn TrieStorage>,
+
+    /// During an apply of a single chunk contracts may be deployed through the
+    /// `Action::DeployContract`.
+    ///
+    /// Unfortunately `TrieStorage` does not have a way to write to the underlying storage, and the
+    /// `TrieUpdate` will only write the contract once the whole transaction is committed at the
+    /// end of the chunk's apply.
+    ///
+    /// As a temporary work-around while we're still involving `Trie` for `ContractCode` storage,
+    /// we'll keep a list of such deployed contracts here. Once the contracts are no longer part of
+    /// The State this field should be removed, and the `Storage::store` function should be
+    /// adjusted to write out the contract into the relevant part of the database immediately
+    /// (without going through transactional storage operations and such).
+    uncommitted_deploys: Arc<Mutex<BTreeMap<CryptoHash, ContractCode>>>,
+}
+
+impl ContractStorage {
+    pub fn new(storage: Arc<dyn TrieStorage>) -> Self {
+        Self { storage, uncommitted_deploys: Default::default() }
+    }
+
+    pub fn get(&self, code_hash: CryptoHash) -> Option<ContractCode> {
+        {
+            let guard = self.uncommitted_deploys.lock().expect("no panics");
+            if let Some(v) = guard.get(&code_hash) {
+                return Some(ContractCode::new(v.code().to_vec(), Some(code_hash)));
+            }
+        }
+
+        match self.storage.retrieve_raw_bytes(&code_hash) {
+            Ok(raw_code) => Some(ContractCode::new(raw_code.to_vec(), Some(code_hash))),
+            Err(_) => None,
+        }
+    }
+
+    pub fn store(&self, code: ContractCode) {
+        let mut guard = self.uncommitted_deploys.lock().expect("no panics");
+        guard.insert(*code.hash(), code);
+    }
+}

core/store/src/lib.rs

@@ -46,6 +46,7 @@ use strum;
 pub mod cold_storage;
 mod columns;
 pub mod config;
+pub mod contract;
 pub mod db;
 pub mod flat;
 pub mod genesis;
@@ -761,6 +762,22 @@ pub fn get<T: BorshDeserialize>(
     }
 }
 
+/// [`get`] without incurring side effects.
+pub fn get_pure<T: BorshDeserialize>(
+    trie: &dyn TrieAccess,
+    key: &TrieKey,
+) -> Result<Option<T>, StorageError> {
+    match trie.get_no_side_effects(key)? {
+        None => Ok(None),
+        Some(data) => match T::try_from_slice(&data) {
+            Err(_err) => {
+                Err(StorageError::StorageInconsistentState("Failed to deserialize".to_string()))
+            }
+            Ok(value) => Ok(Some(value)),
+        },
+    }
+}
+
 /// Writes an object into Trie.
 pub fn set<T: BorshSerialize>(state_update: &mut TrieUpdate, key: TrieKey, value: &T) {
     let data = borsh::to_vec(&value).expect("Borsh serializer is not expected to ever fail");
@@ -970,15 +987,6 @@ pub fn set_code(state_update: &mut TrieUpdate, account_id: AccountId, code: &Con
     state_update.set(TrieKey::ContractCode { account_id }, code.code().to_vec());
 }
 
-pub fn get_code(
-    trie: &dyn TrieAccess,
-    account_id: &AccountId,
-    code_hash: Option<CryptoHash>,
-) -> Result<Option<ContractCode>, StorageError> {
-    let key = TrieKey::ContractCode { account_id: account_id.clone() };
-    trie.get(&key).map(|opt| opt.map(|code| ContractCode::new(code, code_hash)))
-}
-
 /// Removes account, code and all access keys associated to it.
 pub fn remove_account(
     state_update: &mut TrieUpdate,
@@ -1134,6 +1142,18 @@ impl ContractRuntimeCache for StoreContractRuntimeCache {
     }
 }
 
+/// Get the contract WASM code from The State.
+///
+/// Executing all the usual storage access side-effects.
+pub fn get_code(
+    trie: &dyn TrieAccess,
+    account_id: &AccountId,
+    code_hash: Option<CryptoHash>,
+) -> Result<Option<ContractCode>, StorageError> {
+    let key = TrieKey::ContractCode { account_id: account_id.clone() };
+    trie.get(&key).map(|opt| opt.map(|code| ContractCode::new(code, code_hash)))
+}
+
 #[cfg(test)]
 mod tests {
     use near_primitives::hash::CryptoHash;

core/store/src/trie/mem/loading.rs

@@ -271,7 +271,7 @@ mod tests {
             // Check that the accessed nodes are consistent with those from disk.
             for (node_hash, serialized_node) in nodes_accessed {
                 let expected_serialized_node =
-                    trie.internal_retrieve_trie_node(&node_hash, false).unwrap();
+                    trie.internal_retrieve_trie_node(&node_hash, false, true).unwrap();
                 assert_eq!(expected_serialized_node, serialized_node);
             }
         }