fix(epoch-sync): generate EpochSyncInfo on header sync (#10065)

#10029
The fix is 10 lines in `chain/chain/src/chain.rs`. Everything else is
for testing.

I don't like this approach, I am fixing a symptom, not the problem. But
it is good enough for now (and possibly forever).
But, just for context, **the problem** is as follows.
We should write `EpochSyncInfo` every time we write `EpochInfo` of a
full epoch.
It is fine not to write `EpochSyncInfo` for genesis epoch of one block
for two reasons:
- Everyone has full genesis epoch on startup
- Next epoch has the same `epoch_id` and therefore it is incorrect to
have that key in DB before the end of first proper epoch

`EpochInfo` is written in function `record_block_info` that is called
from `add_validator_proposals`.
Both function belong to `EpochManager`, and we cannot add
`EpochSyncInfo` generation to `EpochManager`, as it requires headers
(including the last header of an epoch, that is probably not written
into DB during the execution of anything in `EpochManager`). From an
architecture point of view `EpochSyncInfo` generation is a `Chain`
function.

So, I couldn't enforce that `EpochSyncInfo` is tied to `EpochInfo`
generation. But I traced every usage of `add_validator_proposals`, and
we have only two usages that are
-  in `neard`
- not in tests
- not related to genesis epoch

First usage was covered by the commit that introduced `EpochSyncInfo`
#9440.
Second usage is covered with this PR.

chain/chain/src/chain.rs

@@ -1961,6 +1961,16 @@ impl Chain {
                     .add_validator_proposals(BlockHeaderInfo::new(header, last_finalized_height))?;
                 chain_update.chain_store_update.merge(epoch_manager_update);
                 chain_update.commit()?;
+
+                #[cfg(feature = "new_epoch_sync")]
+                {
+                    // At this point BlockInfo for this header should be in DB and in `epoch_manager`s cache.
+                    let block_info = self.epoch_manager.get_block_info(header.hash())?;
+
+                    let mut chain_update = self.chain_update();
+                    chain_update.save_epoch_sync_info(header.epoch_id(), header, &block_info)?;
+                    chain_update.commit()?;
+                }
             }
         }
 

core/primitives/src/epoch_manager.rs

@@ -1235,14 +1235,14 @@ pub mod epoch_sync {
     use crate::types::validator_stake::ValidatorStake;
     use borsh::{BorshDeserialize, BorshSerialize};
 
-    #[derive(BorshSerialize, BorshDeserialize)]
+    #[derive(BorshSerialize, BorshDeserialize, PartialEq, Debug)]
     pub struct BlockHeaderPair {
         pub header: BlockHeader,
         pub last_finalised_header: BlockHeader,
     }
 
     /// Struct to keep all the info that is transferred for one epoch during Epoch Sync.
-    #[derive(BorshSerialize, BorshDeserialize)]
+    #[derive(BorshSerialize, BorshDeserialize, PartialEq, Debug)]
     pub struct EpochSyncInfo {
         /// None is only used for corner case of the first epoch
         pub first: BlockHeaderPair,

integration-tests/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod genesis_helpers;
+pub mod nearcore_utils;
 pub mod node;
 pub mod runtime_utils;
 pub mod test_helpers;

integration-tests/src/nearcore_utils.rs

@@ -0,0 +1,130 @@
+use crate::genesis_helpers::genesis_block;
+use actix::Addr;
+use near_chain::Block;
+use near_chain_configs::Genesis;
+use near_client::{BlockResponse, ClientActor};
+use near_network::tcp;
+use near_network::test_utils::convert_boot_nodes;
+use near_network::types::PeerInfo;
+use near_o11y::WithSpanContextExt;
+use near_primitives::block::Approval;
+use near_primitives::hash::CryptoHash;
+use near_primitives::merkle::PartialMerkleTree;
+use near_primitives::num_rational::{Ratio, Rational32};
+use near_primitives::types::validator_stake::ValidatorStake;
+use near_primitives::types::{BlockHeightDelta, EpochId};
+use near_primitives::validator_signer::ValidatorSigner;
+use near_primitives::version::PROTOCOL_VERSION;
+use nearcore::config::{GenesisExt, TESTING_INIT_STAKE};
+use nearcore::{load_test_config, NearConfig};
+
+// This assumes that there is no height skipped. Otherwise epoch hash calculation will be wrong.
+pub fn add_blocks(
+    mut blocks: Vec<Block>,
+    client: Addr<ClientActor>,
+    num: usize,
+    epoch_length: BlockHeightDelta,
+    signer: &dyn ValidatorSigner,
+) -> Vec<Block> {
+    let mut prev = &blocks[blocks.len() - 1];
+    let mut block_merkle_tree = PartialMerkleTree::default();
+    for block in blocks.iter() {
+        block_merkle_tree.insert(*block.hash());
+    }
+    for _ in 0..num {
+        let prev_height = prev.header().height();
+        let prev_epoch_height = prev_height / epoch_length;
+        let prev_epoch_last_block_height = prev_epoch_height * epoch_length;
+
+        let height = prev_height + 1;
+        let epoch_id = if height <= epoch_length {
+            EpochId::default()
+        } else {
+            let prev_prev_epoch_height = prev_epoch_height - 1;
+            let prev_prev_epoch_last_block_height = prev_prev_epoch_height * epoch_length;
+            EpochId(*blocks[prev_prev_epoch_last_block_height as usize].hash())
+        };
+
+        let next_epoch_id = EpochId(*blocks[prev_epoch_last_block_height as usize].hash());
+
+        let next_bp_hash = CryptoHash::hash_borsh_iter([ValidatorStake::new(
+            "other".parse().unwrap(),
+            signer.public_key(),
+            TESTING_INIT_STAKE,
+        )]);
+        let block = Block::produce(
+            PROTOCOL_VERSION,
+            PROTOCOL_VERSION,
+            prev.header(),
+            prev.header().height() + 1,
+            prev.header().block_ordinal() + 1,
+            blocks[0].chunks().iter().cloned().collect(),
+            epoch_id,
+            next_epoch_id,
+            None,
+            vec![Some(Box::new(
+                Approval::new(
+                    *prev.hash(),
+                    prev.header().height(),
+                    prev.header().height() + 1,
+                    signer,
+                )
+                .signature,
+            ))],
+            Ratio::from_integer(0),
+            0,
+            1000,
+            Some(0),
+            vec![],
+            vec![],
+            signer,
+            next_bp_hash,
+            block_merkle_tree.root(),
+            None,
+        );
+        block_merkle_tree.insert(*block.hash());
+        let _ = client.do_send(
+            BlockResponse {
+                block: block.clone(),
+                peer_id: PeerInfo::random().id,
+                was_requested: false,
+            }
+            .with_span_context(),
+        );
+        blocks.push(block);
+        prev = &blocks[blocks.len() - 1];
+    }
+    blocks
+}
+
+pub fn setup_configs_with_epoch_length(
+    epoch_length: u64,
+) -> (Genesis, Block, NearConfig, NearConfig) {
+    let mut genesis = Genesis::test(vec!["other".parse().unwrap()], 1);
+    genesis.config.epoch_length = epoch_length;
+    // Avoid InvalidGasPrice error. Blocks must contain accurate `total_supply` value.
+    // Accounting for the inflation in tests is hard.
+    // Disabling inflation in tests is much simpler.
+    genesis.config.max_inflation_rate = Rational32::from_integer(0);
+    let genesis_block = genesis_block(&genesis);
+
+    let (port1, port2) =
+        (tcp::ListenerAddr::reserve_for_test(), tcp::ListenerAddr::reserve_for_test());
+    let mut near1 = load_test_config("test1", port1, genesis.clone());
+    near1.network_config.peer_store.boot_nodes = convert_boot_nodes(vec![("test2", *port2)]);
+    near1.client_config.min_num_peers = 1;
+    near1.client_config.epoch_sync_enabled = false;
+    near1.client_config.state_sync_enabled = true;
+
+    let mut near2 = load_test_config("test2", port2, genesis.clone());
+    near2.network_config.peer_store.boot_nodes = convert_boot_nodes(vec![("test1", *port1)]);
+    near2.client_config.min_num_peers = 1;
+    near2.client_config.epoch_sync_enabled = false;
+    near2.client_config.state_sync_enabled = true;
+
+    (genesis, genesis_block, near1, near2)
+}
+
+pub fn setup_configs() -> (Genesis, Block, NearConfig, NearConfig) {
+    setup_configs_with_epoch_length(5)
+}

integration-tests/src/tests/client/epoch_sync.rs

@@ -1,16 +1,33 @@
+use crate::nearcore_utils::{add_blocks, setup_configs_with_epoch_length};
+use crate::test_helpers::heavy_test;
+use actix::Actor;
+use actix_rt::System;
+use futures::{future, FutureExt};
+use near_actix_test_utils::run_actix;
 use near_chain::{ChainGenesis, Provenance};
 use near_chain_configs::Genesis;
 use near_client::test_utils::TestEnv;
 use near_client::ProcessTxResponse;
+use near_client_primitives::types::GetBlock;
 use near_crypto::{InMemorySigner, KeyType};
-use near_o11y::testonly::init_test_logger;
+use near_network::test_utils::WaitOrTimeoutActor;
+use near_o11y::testonly::{init_integration_logger, init_test_logger};
+use near_o11y::WithSpanContextExt;
+use near_primitives::epoch_manager::epoch_sync::EpochSyncInfo;
+use near_primitives::test_utils::create_test_signer;
 use near_primitives::transaction::{
     Action, DeployContractAction, FunctionCallAction, SignedTransaction,
 };
 use near_primitives_core::hash::CryptoHash;
 use near_primitives_core::types::BlockHeight;
+use near_store::Mode::ReadOnly;
+use near_store::{DBCol, NodeStorage};
 use nearcore::config::GenesisExt;
 use nearcore::test_utils::TestEnvNightshadeSetupExt;
+use nearcore::{start_with_config, NearConfig};
+use std::collections::HashSet;
+use std::path::Path;
+use std::sync::{Arc, RwLock};
 
 fn generate_transactions(last_hash: &CryptoHash, h: BlockHeight) -> Vec<SignedTransaction> {
     let mut txs = vec![];
@@ -96,3 +113,112 @@ fn test_continuous_epoch_sync_info_population() {
         }
     }
 }
+
+/// Produce 4 epochs + 10 blocks.
+/// Start second node without epoch sync, but with state sync.
+/// Sync second node to first node (at least headers).
+/// Check that it has all EpochSyncInfo records and all of them are correct.
+///
+/// The header sync part is based on `integration-tests::nearcore::sync_nodes::sync_nodes`.
+#[test]
+fn test_continuous_epoch_sync_info_population_on_header_sync() {
+    heavy_test(|| {
+        init_integration_logger();
+
+        let (genesis, genesis_block, mut near1_base, mut near2_base) =
+            setup_configs_with_epoch_length(50);
+
+        let dir1_base =
+            tempfile::Builder::new().prefix("epoch_sync_info_in_header_sync_1").tempdir().unwrap();
+        let dir2_base =
+            tempfile::Builder::new().prefix("epoch_sync_info_in_header_sync_2").tempdir().unwrap();
+        let epoch_ids_base = Arc::new(RwLock::new(HashSet::new()));
+
+        let near1 = near1_base.clone();
+        let near2 = near2_base.clone();
+        let dir1_path = dir1_base.path();
+        let dir2_path = dir2_base.path();
+        let epoch_ids = epoch_ids_base.clone();
+
+        run_actix(async move {
+            // Start first node
+            let nearcore::NearNode { client: client1, .. } =
+                start_with_config(dir1_path, near1).expect("start_with_config");
+
+            // Generate 4 epochs + 10 blocks
+            let signer = create_test_signer("other");
+            let blocks =
+                add_blocks(vec![genesis_block], client1, 210, genesis.config.epoch_length, &signer);
+
+            // Save all finished epoch_ids
+            let mut epoch_ids = epoch_ids.write().unwrap();
+            for block in blocks[0..200].iter() {
+                epoch_ids.insert(block.header().epoch_id().clone());
+            }
+
+            // Start second node
+            let nearcore::NearNode { view_client: view_client2, .. } =
+                start_with_config(dir2_path, near2).expect("start_with_config");
+
+            // Wait for second node's headers to sync.
+            // Timeout here means that header sync is not working.
+            // Header sync is better debugged through other tests.
+            // For example, run `integration-tests::nearcore::sync_nodes::sync_nodes` test,
+            // on which this test's setup is based.
+            WaitOrTimeoutActor::new(
+                Box::new(move |_ctx| {
+                    let actor = view_client2.send(GetBlock::latest().with_span_context());
+                    let actor = actor.then(|res| {
+                        match &res {
+                            Ok(Ok(b)) if b.header.height == 210 => System::current().stop(),
+                            Err(_) => return future::ready(()),
+                            _ => {}
+                        };
+                        future::ready(())
+                    });
+                    actix::spawn(actor);
+                }),
+                100,
+                120000,
+            )
+            .start();
+        });
+
+        // Open storages of both nodes
+        let open_read_only_storage = |home_dir: &Path, near_config: &NearConfig| -> NodeStorage {
+            let opener = NodeStorage::opener(home_dir, false, &near_config.config.store, None);
+            opener.open_in_mode(ReadOnly).unwrap()
+        };
+
+        let store1 = open_read_only_storage(dir1_base.path(), &mut near1_base).get_hot_store();
+        let store2 = open_read_only_storage(dir2_base.path(), &mut near2_base).get_hot_store();
+
+        // Check that for every epoch second store has EpochSyncInfo.
+        // And that values in both stores are the same.
+        let epoch_ids = epoch_ids_base.read().unwrap();
+        for epoch_id in epoch_ids.iter() {
+            // Check that we have a value for EpochSyncInfo in the synced node
+            assert!(
+                store2
+                    .get_ser::<EpochSyncInfo>(DBCol::EpochSyncInfo, epoch_id.as_ref())
+                    .unwrap()
+                    .is_some(),
+                "{:?}",
+                epoch_id
+            );
+            // Check that it matches value in full node exactly
+            assert_eq!(
+                store1
+                    .get_ser::<EpochSyncInfo>(DBCol::EpochSyncInfo, epoch_id.as_ref())
+                    .unwrap()
+                    .unwrap(),
+                store2
+                    .get_ser::<EpochSyncInfo>(DBCol::EpochSyncInfo, epoch_id.as_ref())
+                    .unwrap()
+                    .unwrap(),
+                "{:?}",
+                epoch_id
+            );
+        }
+    });
+}