Fix(chain): fix apply chunks (#4228)

Port #4226 to master.

Test plan
----------
Nayduck http://nayduck.eastus.cloudapp.azure.com:3000/#/run/1508

chain/chain/src/chain.rs

@@ -2712,6 +2712,7 @@ impl<'a> ChainUpdate<'a> {
                 &challenges_result,
                 *block.header().random_value(),
                 true,
+                true,
             )
             .unwrap();
         let partial_state = apply_result.proof.unwrap().nodes;
@@ -2864,6 +2865,7 @@ impl<'a> ChainUpdate<'a> {
                             gas_limit,
                             &block.header().challenges_result(),
                             *block.header().random_value(),
+                            true,
                         )
                         .map_err(|e| ErrorKind::Other(e.to_string()))?;
 
@@ -2918,13 +2920,27 @@ impl<'a> ChainUpdate<'a> {
                             new_extra.gas_limit(),
                             &block.header().challenges_result(),
                             *block.header().random_value(),
+                            false,
                         )
                         .map_err(|e| ErrorKind::Other(e.to_string()))?;
 
                     self.chain_store_update.save_trie_changes(apply_result.trie_changes);
                     *new_extra.state_root_mut() = apply_result.new_root;
 
                     self.chain_store_update.save_chunk_extra(&block.hash(), shard_id, new_extra);
+
+                    if !apply_result.outcomes.is_empty() {
+                        // debug_assert!(false);
+                        // Remove in next release
+                        let (_, outcome_paths) =
+                            ApplyTransactionResult::compute_outcomes_proof(&apply_result.outcomes);
+                        self.chain_store_update.save_outcomes_with_proofs(
+                            &block.hash(),
+                            shard_id,
+                            apply_result.outcomes,
+                            outcome_paths,
+                        );
+                    }
                 }
             }
         }
@@ -3612,6 +3628,7 @@ impl<'a> ChainUpdate<'a> {
             gas_limit,
             &block_header.challenges_result(),
             *block_header.random_value(),
+            true,
         )?;
 
         let (outcome_root, outcome_proofs) =
@@ -3690,6 +3707,7 @@ impl<'a> ChainUpdate<'a> {
             chunk_extra.gas_limit(),
             &block_header.challenges_result(),
             *block_header.random_value(),
+            false,
         )?;
 
         self.chain_store_update.save_trie_changes(apply_result.trie_changes);

chain/chain/src/test_utils.rs

@@ -603,6 +603,7 @@ impl RuntimeAdapter for KeyValueRuntime {
         _challenges: &ChallengesResult,
         _random_seed: CryptoHash,
         generate_storage_proof: bool,
+        _is_new_chunk: bool,
     ) -> Result<ApplyTransactionResult, Error> {
         assert!(!generate_storage_proof);
         let mut tx_results = vec![];
@@ -781,6 +782,7 @@ impl RuntimeAdapter for KeyValueRuntime {
         _gas_limit: Gas,
         _challenges: &ChallengesResult,
         _random_value: CryptoHash,
+        _is_new_chunk: bool,
     ) -> Result<ApplyTransactionResult, Error> {
         unimplemented!();
     }

chain/chain/src/types.rs

@@ -515,6 +515,7 @@ pub trait RuntimeAdapter: Send + Sync {
         gas_limit: Gas,
         challenges_result: &ChallengesResult,
         random_seed: CryptoHash,
+        is_new_chunk: bool,
     ) -> Result<ApplyTransactionResult, Error> {
         self.apply_transactions_with_optional_storage_proof(
             shard_id,
@@ -531,6 +532,7 @@ pub trait RuntimeAdapter: Send + Sync {
             challenges_result,
             random_seed,
             false,
+            is_new_chunk,
         )
     }
 
@@ -550,6 +552,7 @@ pub trait RuntimeAdapter: Send + Sync {
         challenges_result: &ChallengesResult,
         random_seed: CryptoHash,
         generate_storage_proof: bool,
+        is_new_chunk: bool,
     ) -> Result<ApplyTransactionResult, Error>;
 
     fn check_state_transition(
@@ -568,6 +571,7 @@ pub trait RuntimeAdapter: Send + Sync {
         gas_limit: Gas,
         challenges_result: &ChallengesResult,
         random_value: CryptoHash,
+        is_new_chunk: bool,
     ) -> Result<ApplyTransactionResult, Error>;
 
     /// Query runtime with given `path` and `data`.

chain/chain/src/validate.rs

@@ -340,6 +340,8 @@ fn validate_chunk_state_challenge(
             prev_chunk_header.gas_limit(),
             &ChallengesResult::default(),
             *block_header.random_value(),
+            // TODO: set it properly when challenges are enabled
+            true,
         )
         .map_err(|_| Error::from(ErrorKind::MaliciousChallenge))?;
     let outcome_root = ApplyTransactionResult::compute_outcomes_proof(&result.outcomes).0;

chain/client/tests/process_blocks.rs

@@ -1,4 +1,4 @@
-use std::collections::HashSet;
+use std::collections::{HashSet, VecDeque};
 use std::iter::FromIterator;
 use std::path::Path;
 use std::str::FromStr;
@@ -43,16 +43,22 @@ use near_primitives::sharding::{EncodedShardChunk, ReedSolomonWrapper, ShardChun
 #[cfg(feature = "protocol_feature_block_header_v3")]
 use near_primitives::sharding::{ShardChunkHeaderInner, ShardChunkHeaderV3};
 
+use near_primitives::receipt::DelayedReceiptIndices;
 use near_primitives::syncing::{get_num_state_parts, ShardStateSyncResponseHeader};
 use near_primitives::transaction::{
-    Action, DeployContractAction, FunctionCallAction, SignedTransaction, Transaction,
+    Action, DeployContractAction, ExecutionStatus, FunctionCallAction, SignedTransaction,
+    Transaction,
 };
+use near_primitives::trie_key::TrieKey;
 use near_primitives::types::validator_stake::ValidatorStake;
 use near_primitives::types::{AccountId, BlockHeight, EpochId, NumBlocks};
 use near_primitives::utils::to_timestamp;
 use near_primitives::validator_signer::{InMemoryValidatorSigner, ValidatorSigner};
 use near_primitives::version::PROTOCOL_VERSION;
-use near_primitives::views::{BlockHeaderView, QueryRequest, QueryResponseKind};
+use near_primitives::views::{
+    BlockHeaderView, FinalExecutionStatus, QueryRequest, QueryResponseKind,
+};
+use near_store::get;
 use near_store::test_utils::create_test_store;
 use neard::config::{GenesisExt, TESTING_INIT_BALANCE, TESTING_INIT_STAKE};
 use neard::NEAR_BASE;
@@ -2618,6 +2624,126 @@ fn test_block_ordinal() {
     assert_eq!(next_block.header().block_ordinal(), ordinal);
 }
 
+#[test]
+fn test_congestion_receipt_execution() {
+    init_test_logger();
+    let epoch_length = 100;
+    let mut genesis = Genesis::test(vec!["test0", "test1"], 1);
+    genesis.config.epoch_length = epoch_length;
+    genesis.config.gas_limit = 10000000000000;
+    let chain_genesis = ChainGenesis::from(&genesis);
+    let mut env =
+        TestEnv::new_with_runtime(chain_genesis, 1, 1, create_nightshade_runtimes(&genesis, 1));
+    let genesis_block = env.clients[0].chain.get_block_by_height(0).unwrap().clone();
+    let signer = InMemorySigner::from_seed("test0", KeyType::ED25519, "test0");
+    let validator_signer = InMemoryValidatorSigner::from_seed("test0", KeyType::ED25519, "test0");
+
+    // step 0: deploy contract to test0
+    let tx = SignedTransaction::from_actions(
+        1,
+        "test0".to_string(),
+        "test0".to_string(),
+        &signer,
+        vec![Action::DeployContract(DeployContractAction {
+            code: near_test_contracts::rs_contract().to_vec(),
+        })],
+        *genesis_block.hash(),
+    );
+    env.clients[0].process_tx(tx, false, false);
+    for i in 1..3 {
+        env.produce_block(0, i);
+    }
+
+    // step 1: create function call transactions that generate promises
+    let gas_1 = 9_000_000_000_000;
+    let gas_2 = gas_1 / 3;
+    let mut tx_hashes = vec![];
+
+    for i in 0..3 {
+        let data = serde_json::json!([
+            {"create": {
+            "account_id": "test0",
+            "method_name": "call_promise",
+            "arguments": [],
+            "amount": "0",
+            "gas": gas_2,
+            }, "id": 0 }
+        ]);
+
+        let signed_transaction = SignedTransaction::from_actions(
+            i + 10,
+            "test0".to_string(),
+            "test0".to_string(),
+            &signer,
+            vec![Action::FunctionCall(FunctionCallAction {
+                method_name: "call_promise".to_string(),
+                args: serde_json::to_vec(&data).unwrap(),
+                gas: gas_1,
+                deposit: 0,
+            })],
+            *genesis_block.hash(),
+        );
+        tx_hashes.push(signed_transaction.get_hash());
+        env.clients[0].process_tx(signed_transaction, false, false);
+    }
+
+    // step 2: produce block with no new chunk
+    env.produce_block(0, 3);
+    let height = 4;
+    env.produce_block(0, height);
+    let prev_block = env.clients[0].chain.get_block_by_height(height).unwrap().clone();
+    let chunk_extra = env.clients[0].chain.get_chunk_extra(prev_block.hash(), 0).unwrap().clone();
+    assert!(chunk_extra.gas_used() >= chunk_extra.gas_limit());
+    let state_update =
+        env.clients[0].runtime_adapter.get_tries().new_trie_update(0, *chunk_extra.state_root());
+    let delayed_indices =
+        get::<DelayedReceiptIndices>(&state_update, &TrieKey::DelayedReceiptIndices)
+            .unwrap()
+            .unwrap();
+    assert!(delayed_indices.next_available_index > 0);
+    let mut block = env.clients[0].produce_block(height + 1).unwrap().unwrap();
+    let chunk_headers = vec![prev_block.chunks()[0].clone()];
+    block.set_chunks(chunk_headers.clone());
+    block.mut_header().get_mut().inner_rest.chunk_headers_root =
+        Block::compute_chunk_headers_root(&chunk_headers).0;
+    block.mut_header().get_mut().inner_rest.chunk_tx_root =
+        Block::compute_chunk_tx_root(&chunk_headers);
+    block.mut_header().get_mut().inner_rest.chunk_receipts_root =
+        Block::compute_chunk_receipts_root(&chunk_headers);
+    block.mut_header().get_mut().inner_lite.prev_state_root =
+        Block::compute_state_root(&chunk_headers);
+    block.mut_header().get_mut().inner_rest.chunk_mask = vec![false];
+    block.mut_header().get_mut().inner_rest.gas_price = prev_block.header().gas_price();
+    block.mut_header().resign(&validator_signer);
+    env.process_block(0, block.clone(), Provenance::NONE);
+
+    // let all receipts finish
+    for i in height + 2..height + 7 {
+        env.produce_block(0, i);
+    }
+
+    for tx_hash in &tx_hashes {
+        let final_outcome = env.clients[0].chain.get_final_transaction_result(&tx_hash).unwrap();
+        assert!(matches!(final_outcome.status, FinalExecutionStatus::SuccessValue(_)));
+
+        // Check that all receipt ids have corresponding execution outcomes. This means that all receipts generated are executed.
+        let transaction_outcome = env.clients[0].chain.get_execution_outcome(tx_hash).unwrap();
+        let mut receipt_ids: VecDeque<_> =
+            transaction_outcome.outcome_with_id.outcome.receipt_ids.into();
+        while !receipt_ids.is_empty() {
+            let receipt_id = receipt_ids.pop_front().unwrap();
+            let receipt_outcome = env.clients[0].chain.get_execution_outcome(&receipt_id).unwrap();
+            match receipt_outcome.outcome_with_id.outcome.status {
+                ExecutionStatus::SuccessValue(_) | ExecutionStatus::SuccessReceiptId(_) => {}
+                ExecutionStatus::Failure(_) | ExecutionStatus::Unknown => {
+                    panic!("unexpected receipt execution outcome")
+                }
+            }
+            receipt_ids.extend(receipt_outcome.outcome_with_id.outcome.receipt_ids);
+        }
+    }
+}
+
 #[cfg(feature = "protocol_feature_access_key_nonce_range")]
 #[cfg(test)]
 mod access_key_nonce_range_tests {

core/primitives/src/runtime/apply_state.rs

@@ -34,6 +34,8 @@ pub struct ApplyState {
     pub config: Arc<RuntimeConfig>,
     /// Cache for compiled contracts.
     pub cache: Option<Arc<dyn CompiledContractCache>>,
+    /// Whether the chunk being applied is new.
+    pub is_new_chunk: bool,
     /// Ethereum chain id.
     #[cfg(feature = "protocol_feature_evm")]
     pub evm_chain_id: u64,

core/primitives/src/version.rs

@@ -13,7 +13,7 @@ pub struct Version {
 pub type DbVersion = u32;
 
 /// Current version of the database.
-pub const DB_VERSION: DbVersion = 19;
+pub const DB_VERSION: DbVersion = 20;
 
 /// Protocol version type.
 pub use near_primitives_core::types::ProtocolVersion;
@@ -102,11 +102,12 @@ pub enum ProtocolFeature {
     TransactionSizeLimit,
     #[cfg(feature = "protocol_feature_allow_create_account_on_delete")]
     AllowCreateAccountOnDelete,
+    FixApplyChunks,
 }
 
 /// Current latest stable version of the protocol.
 #[cfg(not(feature = "nightly_protocol"))]
-pub const PROTOCOL_VERSION: ProtocolVersion = 43;
+pub const PROTOCOL_VERSION: ProtocolVersion = 44;
 
 /// Current latest nightly version of the protocol.
 #[cfg(feature = "nightly_protocol")]
@@ -119,6 +120,7 @@ impl ProtocolFeature {
             #[cfg(feature = "protocol_feature_lower_storage_cost")]
             ProtocolFeature::LowerStorageCost => 42,
             ProtocolFeature::DeleteActionRestriction => 43,
+            ProtocolFeature::FixApplyChunks => 44,
 
             // Nightly features
             #[cfg(feature = "protocol_feature_forward_chunk_parts")]