Always record storage when stateless validation is enabled

For stateless validation we need to record all trie reads
performed when applying a chunk in order to prepare a `PartialState`
that will be included in `ChunkStateWitness`

Initially it was only necessary to record trie reads when producing
a chunk. Validators could read all the values from the provided `PartialState`
without recording anything.

A recent change (near#10703) introduced a soft
limit on the size of `PartialState`. When applying a chunk we watch how much
state was recorded, and once the amount of state exceeds the soft limit we stop
applying the receipts in this chunk.
This needs to be done on both the chunk producer and chunk validator - if the
chunk validator doesn't record reads and enforce the limit, it will produce
a different result of chunk application, which would lead to validation failure.

This means that state should be recorded in all cases when a statelessly validated
chunk is applied. Let's remove the configuration option that controls whether
trie reads should be recorded (`record_storage`) and just record the reads on
every chunk application (when `statelessnet_protocol` feature is enabled).

Refs: [zulip conversation](https://near.zulipchat.com/#narrow/stream/407237-core.2Fstateless-validation/topic/State.20witness.20size.20limit/near/428313518)

chain/chain/src/chain.rs

@@ -3323,12 +3323,8 @@ impl Chain {
             // only for a single shard. This so far has been enough.
             let state_patch = state_patch.take();
 
-            let storage_context = StorageContext {
-                storage_data_source: StorageDataSource::Db,
-                state_patch,
-                record_storage: self
-                    .should_produce_state_witness_for_this_or_next_epoch(me, block.header())?,
-            };
+            let storage_context =
+                StorageContext { storage_data_source: StorageDataSource::Db, state_patch };
             let stateful_job = self.get_update_shard_job(
                 me,
                 block,

chain/chain/src/runtime/mod.rs

@@ -704,7 +704,7 @@ impl RuntimeAdapter for NightshadeRuntime {
                 storage_config.use_flat_storage,
             ),
         };
-        if storage_config.record_storage {
+        if cfg!(feature = "statelessnet_protocol") {
             trie = trie.recording_reads();
         }
         let mut state_update = TrieUpdate::new(trie);
@@ -865,7 +865,7 @@ impl RuntimeAdapter for NightshadeRuntime {
                 storage_config.use_flat_storage,
             ),
         };
-        if storage_config.record_storage {
+        if cfg!(feature = "statelessnet_protocol") {
             trie = trie.recording_reads();
         }
 

chain/chain/src/runtime/tests.rs

@@ -1609,7 +1609,6 @@ fn test_prepare_transactions_storage_proof() {
         use_flat_storage: true,
         source: StorageDataSource::Db,
         state_patch: Default::default(),
-        record_storage: true,
     };
 
     let proposed_transactions = prepare_transactions(
@@ -1630,7 +1629,6 @@ fn test_prepare_transactions_storage_proof() {
             nodes: proposed_transactions.storage_proof.unwrap(),
         }),
         state_patch: Default::default(),
-        record_storage: false,
     };
 
     let validated_transactions = prepare_transactions(
@@ -1655,7 +1653,6 @@ fn test_prepare_transactions_empty_storage_proof() {
         use_flat_storage: true,
         source: StorageDataSource::Db,
         state_patch: Default::default(),
-        record_storage: true,
     };
 
     let proposed_transactions = prepare_transactions(
@@ -1676,7 +1673,6 @@ fn test_prepare_transactions_empty_storage_proof() {
             nodes: PartialState::default(), // We use empty storage proof here.
         }),
         state_patch: Default::default(),
-        record_storage: false,
     };
 
     let validation_result = prepare_transactions(

chain/chain/src/test_utils/kv_runtime.rs

@@ -1083,7 +1083,7 @@ impl RuntimeAdapter for KeyValueRuntime {
 
     fn prepare_transactions(
         &self,
-        storage: RuntimeStorageConfig,
+        _storage: RuntimeStorageConfig,
         _chunk: PrepareTransactionsChunkContext,
         _prev_block: PrepareTransactionsBlockContext,
         transaction_groups: &mut dyn TransactionGroupIterator,
@@ -1094,11 +1094,9 @@ impl RuntimeAdapter for KeyValueRuntime {
         while let Some(iter) = transaction_groups.next() {
             res.push(iter.next().unwrap());
         }
-        Ok(PreparedTransactions {
-            transactions: res,
-            limited_by: None,
-            storage_proof: if storage.record_storage { Some(Default::default()) } else { None },
-        })
+        let storage_proof =
+            if cfg!(feature = "statelessnet_protocol") { Some(Default::default()) } else { None };
+        Ok(PreparedTransactions { transactions: res, limited_by: None, storage_proof })
     }
 
     fn apply_chunk(
@@ -1242,7 +1240,8 @@ impl RuntimeAdapter for KeyValueRuntime {
         let state_root = hash(&data);
         self.state.write().unwrap().insert(state_root, state);
         self.state_size.write().unwrap().insert(state_root, state_size);
-
+        let storage_proof =
+            if cfg!(feature = "statelessnet_protocol") { Some(Default::default()) } else { None };
         Ok(ApplyChunkResult {
             trie_changes: WrappedTrieChanges::new(
                 self.get_tries(),
@@ -1258,7 +1257,7 @@ impl RuntimeAdapter for KeyValueRuntime {
             validator_proposals: vec![],
             total_gas_burnt: 0,
             total_balance_burnt: 0,
-            proof: if storage_config.record_storage { Some(Default::default()) } else { None },
+            proof: storage_proof,
             processed_delayed_receipts: vec![],
             applied_receipts_hash: hash(&borsh::to_vec(receipts).unwrap()),
         })

chain/chain/src/types.rs

@@ -263,7 +263,6 @@ pub struct RuntimeStorageConfig {
     pub use_flat_storage: bool,
     pub source: StorageDataSource,
     pub state_patch: SandboxStatePatch,
-    pub record_storage: bool,
 }
 
 impl RuntimeStorageConfig {
@@ -273,7 +272,6 @@ impl RuntimeStorageConfig {
             use_flat_storage,
             source: StorageDataSource::Db,
             state_patch: Default::default(),
-            record_storage: false,
         }
     }
 }

chain/chain/src/update_shard.rs

@@ -115,7 +115,6 @@ pub struct StorageContext {
     /// Data source used for processing shard update.
     pub storage_data_source: StorageDataSource,
     pub state_patch: SandboxStatePatch,
-    pub record_storage: bool,
 }
 
 /// Processes shard update with given block and shard.
@@ -185,7 +184,6 @@ pub fn apply_new_chunk(
         use_flat_storage: true,
         source: storage_context.storage_data_source,
         state_patch: storage_context.state_patch,
-        record_storage: storage_context.record_storage,
     };
     match runtime.apply_chunk(
         storage_config,
@@ -247,7 +245,6 @@ pub fn apply_old_chunk(
         use_flat_storage: true,
         source: storage_context.storage_data_source,
         state_patch: storage_context.state_patch,
-        record_storage: storage_context.record_storage,
     };
     match runtime.apply_chunk(
         storage_config,

chain/client/src/client.rs

@@ -999,18 +999,11 @@ impl Client {
         let prepared_transactions = if let Some(mut iter) =
             sharded_tx_pool.get_pool_iterator(shard_uid)
         {
-            let me = self
-                .validator_signer
-                .as_ref()
-                .map(|validator_signer| validator_signer.validator_id().clone());
-            let record_storage = chain
-                .should_produce_state_witness_for_this_or_next_epoch(&me, &prev_block_header)?;
             let storage_config = RuntimeStorageConfig {
                 state_root: *chunk_extra.state_root(),
                 use_flat_storage: true,
                 source: StorageDataSource::Db,
                 state_patch: Default::default(),
-                record_storage,
             };
             runtime.prepare_transactions(
                 storage_config,

chain/client/src/stateless_validation/chunk_validator/mod.rs

@@ -262,7 +262,6 @@ pub(crate) fn pre_validate_chunk_state_witness(
                 nodes: state_witness.new_transactions_validation_state.clone(),
             }),
             state_patch: Default::default(),
-            record_storage: false,
         };
 
         match validate_prepared_transactions(
@@ -314,7 +313,6 @@ pub(crate) fn pre_validate_chunk_state_witness(
                     nodes: state_witness.main_state_transition.base_state.clone(),
                 }),
                 state_patch: Default::default(),
-                record_storage: false,
             },
         })
     };
@@ -529,7 +527,6 @@ pub(crate) fn validate_chunk_state_witness(
                     nodes: transition.base_state,
                 }),
                 state_patch: Default::default(),
-                record_storage: false,
             },
         };
         let OldChunkResult { apply_result, .. } = apply_old_chunk(

chain/client/src/stateless_validation/shadow_validate.rs

@@ -57,7 +57,6 @@ impl Client {
             use_flat_storage: true,
             source: StorageDataSource::Db,
             state_patch: Default::default(),
-            record_storage: true,
         };
 
         // We call `validate_prepared_transactions()` here because we need storage proof for transactions validation.