Merge branch 'master' into analyze-delayed-receipts

.github/workflows/master_fuzzer_binaries.yml

@@ -50,7 +50,6 @@ jobs:
       - name: "Compile fuzzers and upload to GCS"
         run: |
           NAME="nearcore-${{ github.ref_name }}-$(env TZ=Etc/UTC  date +"%Y%m%d%H%M%S")"
-          sed -i 's/warnings = "deny"/warnings = "warn"/' Cargo.toml
           # Our Clusterfuzz setup currently (2024-02) runs on Cascade Lake CPUs
           RUSTFLAGS="--cfg fuzz -C target-cpu=cascadelake" cargo +nightly bolero build-clusterfuzz --all-features --profile fuzz
           gsutil cp -Z target/fuzz/clusterfuzz.tar "gs://fuzzer_targets/${{ github.ref_name }}/$NAME.tar.gz"

.github/workflows/near_crates_publish.yml

@@ -31,7 +31,7 @@ jobs:
         run: |
           set -x
           cargo install --git https://github.com/miraclx/cargo-workspaces --tag v0.3.0 cargo-workspaces
-          RUSTFLAGS="-A warnings" cargo ws publish --yes --allow-dirty --force '*' \
+          cargo ws publish --yes --allow-dirty --force '*' \
               --no-git-commit --no-git-push --no-individual-tags --tag-prefix 'crates-' \
               --tag-msg $$'crates.io snapshot\n---%{\n- %n - https://crates.io/crates/%n/%v}'
 

.github/workflows/ondemand_fuzzer_binaries.yml

@@ -86,6 +86,5 @@ jobs:
       - name: "Compile fuzzers and upload to GCS"
         run: |
           NAME="nearcore-$branch_type-$(env TZ=Etc/UTC  date +"%Y%m%d%H%M%S")"
-          sed -i 's/warnings = "deny"/warnings = "warn"/' Cargo.toml
           RUSTFLAGS="--cfg fuzz" cargo +nightly bolero build-clusterfuzz --all-features --profile fuzz
           gsutil cp -Z target/fuzz/clusterfuzz.tar "gs://fuzzer_targets/$branch_type/$NAME.tar.gz"

Cargo.toml

@@ -95,9 +95,6 @@ members = [
     "utils/stdx",
 ]
 
-[workspace.lints.rust]
-warnings = "deny"
-
 [workspace.lints.clippy]
 all = { level = "allow", priority = -100 }
 correctness = { level = "deny", priority = -50 }
@@ -149,7 +146,6 @@ cpu-time = "1.0"
 criterion = { version = "0.5.1", default_features = false, features = ["html_reports", "cargo_bench_support"] }
 crossbeam = "0.8"
 crossbeam-channel = "0.5.8"
-crossbeam-queue = "0.3.8"
 csv = "1.2.1"
 curve25519-dalek = { version = "4.1.1", default-features = false, features = ["alloc", "precomputed-tables", "rand_core"] }
 derive-enum-from-into = "0.1.1"

Justfile

@@ -39,6 +39,7 @@ nextest TYPE *FLAGS: (nextest-unit TYPE FLAGS) (nextest-integration TYPE FLAGS)
 
 # cargo unit tests, TYPE is "stable" or "nightly"
 nextest-unit TYPE *FLAGS:
+    RUSTFLAGS="-D warnings" \
     cargo nextest run \
         --locked \
         --workspace \
@@ -55,6 +56,7 @@ nextest-unit TYPE *FLAGS:
 # cargo integration tests, TYPE is "stable" or "nightly"
 [linux]
 nextest-integration TYPE *FLAGS:
+    RUSTFLAGS="-D warnings" \
     cargo nextest run \
         --locked \
         --package integration-tests \
@@ -73,6 +75,7 @@ nextest-integration TYPE *FLAGS:
 # check various build configurations compile as anticipated
 check-non-default:
     # Ensure that near-vm-runner always builds without default features enabled
+    RUSTFLAGS="-D warnings" \
     cargo check -p near-vm-runner --no-default-features
 
 # check rust formatting
@@ -81,7 +84,9 @@ check-cargo-fmt:
 
 # check clippy lints
 check-cargo-clippy:
-    env CARGO_TARGET_DIR="target/clippy" cargo clippy --all-features --all-targets --locked
+    CARGO_TARGET_DIR="target/clippy" \
+    RUSTFLAGS="-D warnings" \
+    cargo clippy --all-features --all-targets --locked
 
 # check cargo deny lints
 check-cargo-deny:

chain/chain/src/chain.rs

@@ -13,6 +13,7 @@ use crate::state_request_tracker::StateRequestTracker;
 use crate::state_snapshot_actor::SnapshotCallbacks;
 use crate::store::{ChainStore, ChainStoreAccess, ChainStoreUpdate};
 
+use crate::rayon_spawner::RayonAsyncComputationSpawner;
 use crate::types::{
     AcceptedBlock, ApplyChunkBlockContext, BlockEconomicsConfig, ChainConfig, RuntimeAdapter,
     StorageDataSource,
@@ -37,6 +38,7 @@ use borsh::BorshDeserialize;
 use crossbeam_channel::{unbounded, Receiver, Sender};
 use itertools::Itertools;
 use lru::LruCache;
+use near_async::futures::{AsyncComputationSpawner, AsyncComputationSpawnerExt};
 use near_async::time::{Clock, Duration, Instant};
 use near_chain_configs::{MutableConfigValue, ReshardingConfig, ReshardingHandle};
 #[cfg(feature = "new_epoch_sync")]
@@ -217,7 +219,7 @@ type BlockApplyChunksResult = (CryptoHash, Vec<(ShardId, Result<ShardUpdateResul
 /// Provides current view on the state according to the chain state.
 pub struct Chain {
     pub(crate) clock: Clock,
-    chain_store: ChainStore,
+    pub chain_store: ChainStore,
     pub epoch_manager: Arc<dyn EpochManagerAdapter>,
     pub shard_tracker: ShardTracker,
     pub runtime_adapter: Arc<dyn RuntimeAdapter>,
@@ -239,6 +241,8 @@ pub struct Chain {
     apply_chunks_sender: Sender<BlockApplyChunksResult>,
     /// Used to receive apply chunks results
     apply_chunks_receiver: Receiver<BlockApplyChunksResult>,
+    /// Used to spawn the apply chunks jobs.
+    apply_chunks_spawner: Arc<dyn AsyncComputationSpawner>,
     /// Time when head was updated most recently.
     last_time_head_updated: Instant,
     /// Prevents re-application of known-to-be-invalid blocks, so that in case of a
@@ -362,6 +366,7 @@ impl Chain {
             blocks_delay_tracker: BlocksDelayTracker::new(clock.clone()),
             apply_chunks_sender: sc,
             apply_chunks_receiver: rc,
+            apply_chunks_spawner: Arc::new(RayonAsyncComputationSpawner),
             last_time_head_updated: clock.now(),
             invalid_blocks: LruCache::new(INVALID_CHUNKS_POOL_SIZE),
             pending_state_patch: Default::default(),
@@ -384,6 +389,7 @@ impl Chain {
         doomslug_threshold_mode: DoomslugThresholdMode,
         chain_config: ChainConfig,
         snapshot_callbacks: Option<SnapshotCallbacks>,
+        apply_chunks_spawner: Arc<dyn AsyncComputationSpawner>,
     ) -> Result<Chain, Error> {
         // Get runtime initial state and create genesis block out of it.
         let state_roots = get_genesis_state_roots(runtime_adapter.store())?
@@ -538,6 +544,7 @@ impl Chain {
             blocks_delay_tracker: BlocksDelayTracker::new(clock.clone()),
             apply_chunks_sender: sc,
             apply_chunks_receiver: rc,
+            apply_chunks_spawner,
             last_time_head_updated: clock.now(),
             pending_state_patch: Default::default(),
             requested_state_parts: StateRequestTracker::new(),
@@ -1707,7 +1714,7 @@ impl Chain {
         apply_chunks_done_callback: DoneApplyChunkCallback,
     ) {
         let sc = self.apply_chunks_sender.clone();
-        spawn(move || {
+        self.apply_chunks_spawner.spawn("apply_chunks", move || {
             // do_apply_chunks runs `work` in parallel, but still waits for all of them to finish
             let res = do_apply_chunks(block_hash, block_height, work);
             // If we encounter error here, that means the receiver is deallocated and the client
@@ -1719,13 +1726,6 @@ impl Chain {
             }
             apply_chunks_done_callback(block_hash);
         });
-
-        /// `rayon::spawn` decorated to propagate `tracing` context across
-        /// threads.
-        fn spawn(f: impl FnOnce() + Send + 'static) {
-            let dispatcher = tracing::dispatcher::get_default(|it| it.clone());
-            rayon::spawn(move || tracing::dispatcher::with_default(&dispatcher, f))
-        }
     }
 
     fn postprocess_block_only(

chain/chain/src/chain_update.rs

@@ -269,7 +269,7 @@ impl<'a> ChainUpdate<'a> {
                         result.shard_uid,
                         result.trie_changes.state_changes(),
                     )?;
-                    flat_storage_manager.update_flat_storage_for_shard(*shard_uid, block)?;
+                    flat_storage_manager.update_flat_storage_for_shard(result.shard_uid, block)?;
                     self.chain_store_update.merge(store_update);
 
                     self.chain_store_update.save_chunk_extra(

chain/chain/src/lib.rs

@@ -34,6 +34,7 @@ pub mod test_utils;
 pub mod types;
 pub mod validate;
 
+pub mod rayon_spawner;
 pub mod sharding;
 #[cfg(test)]
 mod tests;

chain/chain/src/rayon_spawner.rs

@@ -0,0 +1,10 @@
+use near_async::futures::AsyncComputationSpawner;
+
+pub struct RayonAsyncComputationSpawner;
+
+impl AsyncComputationSpawner for RayonAsyncComputationSpawner {
+    fn spawn_boxed(&self, _name: &str, f: Box<dyn FnOnce() + Send>) {
+        let dispatcher = tracing::dispatcher::get_default(|it| it.clone());
+        rayon::spawn(move || tracing::dispatcher::with_default(&dispatcher, f))
+    }
+}

chain/chain/src/resharding.rs

@@ -145,13 +145,15 @@ fn apply_delayed_receipts<'a>(
     state_roots: HashMap<ShardUId, StateRoot>,
     account_id_to_shard_uid: &(dyn Fn(&AccountId) -> ShardUId + 'a),
 ) -> Result<HashMap<ShardUId, StateRoot>, Error> {
+    let mut total_count = 0;
     let orig_trie_update = tries.new_trie_update_view(orig_shard_uid, orig_state_root);
 
     let mut start_index = None;
     let mut new_state_roots = state_roots;
     while let Some((next_index, receipts)) =
         get_delayed_receipts(&orig_trie_update, start_index, config.batch_size)?
     {
+        total_count += receipts.len() as u64;
         let (store_update, updated_state_roots) = tries.apply_delayed_receipts_to_children_states(
             &new_state_roots,
             &receipts,
@@ -162,6 +164,7 @@ fn apply_delayed_receipts<'a>(
         store_update.commit()?;
     }
 
+    tracing::debug!(target: "resharding", ?orig_shard_uid, ?total_count, "Applied delayed receipts");
     Ok(new_state_roots)
 }
 

chain/chain/src/runtime/tests.rs

@@ -37,6 +37,7 @@ use near_store::{get_genesis_state_roots, NodeStorage, PartialStorage};
 
 use super::*;
 
+use crate::rayon_spawner::RayonAsyncComputationSpawner;
 use near_async::time::Clock;
 use near_primitives::account::id::AccountIdRef;
 use near_primitives::trie_key::TrieKey;
@@ -1541,6 +1542,7 @@ fn get_test_env_with_chain_and_pool() -> (TestEnv, Chain, TransactionPool) {
         DoomslugThresholdMode::NoApprovals,
         ChainConfig::test(),
         None,
+        Arc::new(RayonAsyncComputationSpawner),
     )
     .unwrap();
 

chain/chain/src/store.rs

@@ -607,6 +607,7 @@ impl ChainStore {
     ) -> Result<(), Error> {
         tracing::trace!(target: "resharding", ?protocol_version, shard_id, receipts_shard_id, "reassign_outgoing_receipts_for_resharding");
         // If simple nightshade v2 is enabled and stable use that.
+        // Same reassignment of outgoing receipts works for simple nightshade v3
         if checked_feature!("stable", SimpleNightshadeV2, protocol_version) {
             Self::reassign_outgoing_receipts_for_resharding_v2(
                 receipts,

chain/chain/src/store_validator.rs

@@ -387,6 +387,7 @@ mod tests {
     use near_store::genesis::initialize_genesis_state;
     use near_store::test_utils::create_test_store;
 
+    use crate::rayon_spawner::RayonAsyncComputationSpawner;
     use crate::runtime::NightshadeRuntime;
     use crate::types::ChainConfig;
     use crate::{Chain, ChainGenesis, ChainStoreAccess, DoomslugThresholdMode};
@@ -416,6 +417,7 @@ mod tests {
             DoomslugThresholdMode::NoApprovals,
             ChainConfig::test(),
             None,
+            Arc::new(RayonAsyncComputationSpawner),
         )
         .unwrap();
         (

chain/chain/src/test_utils.rs

@@ -6,6 +6,7 @@ use std::sync::Arc;
 
 use crate::block_processing_utils::BlockNotInPoolError;
 use crate::chain::Chain;
+use crate::rayon_spawner::RayonAsyncComputationSpawner;
 use crate::runtime::NightshadeRuntime;
 use crate::store::ChainStoreAccess;
 use crate::types::{AcceptedBlock, ChainConfig, ChainGenesis};
@@ -73,6 +74,7 @@ pub fn get_chain_with_epoch_length_and_num_shards(
         DoomslugThresholdMode::NoApprovals,
         ChainConfig::test(),
         None,
+        Arc::new(RayonAsyncComputationSpawner),
     )
     .unwrap()
 }
@@ -155,6 +157,7 @@ fn setup_with_tx_validity_period(
         DoomslugThresholdMode::NoApprovals,
         ChainConfig::test(),
         None,
+        Arc::new(RayonAsyncComputationSpawner),
     )
     .unwrap();
 
@@ -194,6 +197,7 @@ pub fn setup_with_validators(
         DoomslugThresholdMode::NoApprovals,
         ChainConfig::test(),
         None,
+        Arc::new(RayonAsyncComputationSpawner),
     )
     .unwrap();
     (chain, epoch_manager, runtime, signers)
@@ -231,6 +235,7 @@ pub fn setup_with_validators_and_start_time(
         DoomslugThresholdMode::NoApprovals,
         ChainConfig::test(),
         None,
+        Arc::new(RayonAsyncComputationSpawner),
     )
     .unwrap();
     (chain, epoch_manager, runtime, signers)

chain/chunks/README.md

@@ -2,7 +2,7 @@
 
 This crate cotains functions to handle chunks. In NEAR - the block consists of multiple chunks - at most one per shard.
 
-When a chunk is created, the creator encodes its contents using Reed Solomon encoding (ErasureCoding) and adds cross-shard receipts - creating PartialEncodedChunks that are later sent to all the validators (each validator gets a subset of them). This is done for data availability reasons (so that we need only a part of the validators to reconstruct the whole chunk). You can read more about it in [the Nightshade paper](https://near.org/papers/nightshade).
+When a chunk is created, the creator encodes its contents using Reed Solomon encoding (ErasureCoding) and adds cross-shard receipts - creating PartialEncodedChunks that are later sent to all the validators (each validator gets a subset of them). This is done for data availability reasons (so that we need only a part of the validators to reconstruct the whole chunk). You can read more about it in [the Nightshade paper](https://near.org/files/nightshade.pdf).
 
 
 A honest validator will only approve a block if it receives its assigned parts for all chunks in the block - which means that for each chunk, it has `has_all_parts()` returning true.

chain/chunks/src/lib.rs

@@ -90,8 +90,10 @@ use metrics::{
     PARTIAL_ENCODED_CHUNK_FORWARD_CACHED_WITHOUT_HEADER,
     PARTIAL_ENCODED_CHUNK_FORWARD_CACHED_WITHOUT_PREV_BLOCK, PARTIAL_ENCODED_CHUNK_RESPONSE_DELAY,
 };
+use near_async::futures::{DelayedActionRunner, DelayedActionRunnerExt};
 use near_async::messaging::Sender;
 use near_async::time;
+use near_async::time::Duration;
 use near_chain::byzantine_assert;
 use near_chain::chunks_store::ReadOnlyChunksStore;
 use near_chain::near_chain_primitives::error::Error::DBNotFoundErr;
@@ -305,6 +307,21 @@ impl ShardsManager {
         }
     }
 
+    pub fn periodically_resend_chunk_requests(
+        &mut self,
+        delayed_action_runner: &mut dyn DelayedActionRunner<Self>,
+        interval: Duration,
+    ) {
+        delayed_action_runner.run_later(
+            "resend_chunk_requests",
+            interval,
+            move |this, delayed_action_runner| {
+                this.resend_chunk_requests();
+                this.periodically_resend_chunk_requests(delayed_action_runner, interval);
+            },
+        )
+    }
+
     pub fn update_chain_heads(&mut self, head: Tip, header_head: Tip) {
         self.encoded_chunks.update_largest_seen_height(
             head.height,