feat(spice): handle uncertified validator proposals at epoch boundary

chain/chain/src/spice_core.rs

@@ -6,12 +6,15 @@ use near_primitives::block_body::{SpiceCoreStatement, SpiceCoreStatements};
 use near_primitives::errors::InvalidSpiceCoreStatementsError;
 use near_primitives::gas::Gas;
 use near_primitives::hash::CryptoHash;
+use near_primitives::shard_layout::ShardUId;
 use near_primitives::stateless_validation::spice_chunk_endorsement::{
     SpiceEndorsementCoreStatement, SpiceStoredVerifiedEndorsement,
 };
+use near_primitives::types::chunk_extra::ChunkExtra;
+use near_primitives::types::validator_stake::ValidatorStake;
 use near_primitives::types::{
-    AccountId, BlockExecutionResults, ChunkExecutionResult, ChunkExecutionResultHash, ShardId,
-    SpiceChunkId, SpiceUncertifiedChunkInfo,
+    AccountId, BlockExecutionResults, BlockHeight, ChunkExecutionResult, ChunkExecutionResultHash,
+    ShardId, SpiceChunkId, SpiceUncertifiedChunkInfo,
 };
 use near_primitives::utils::{get_endorsements_key, get_execution_results_key};
 use near_store::adapter::StoreAdapter as _;
@@ -109,6 +112,89 @@ impl SpiceCoreReader {
         get_uncertified_chunks(&self.chain_store, block_hash)
     }
 
+    /// Returns the most recent validator proposals from uncertified chunks for
+    /// a given shard. These are proposals that are not yet certified on-chain
+    /// at the given block hash, and need to be accounted for in
+    /// `last_proposals` at epoch boundaries.
+    ///
+    /// Proposals are sorted ascending by block height. If multiple uncertified
+    /// chunks contain proposals for the same account, the most recent one (last
+    /// in iteration order) should be kept by the caller's fold/insert logic.
+    pub fn get_uncertified_validator_proposals(
+        &self,
+        block_hash: &CryptoHash,
+        shard_id: ShardId,
+    ) -> Result<Vec<ValidatorStake>, Error> {
+        let uncertified_chunks = self.get_uncertified_chunks(block_hash)?;
+        let shard_uncertified_chunks: Vec<_> = uncertified_chunks
+            .into_iter()
+            .filter(|info| info.chunk_id.shard_id == shard_id)
+            .collect();
+
+        // Resolve ChunkExtra for each uncertified chunk. Try ChunkExtra column
+        // first (available on tracking nodes). For any remaining chunks, perform
+        // a backward scan from the current chain head toward `block_hash` to find
+        // on-chain certified execution results in blocks' core statements.
+        let epoch_id = self.epoch_manager.get_epoch_id(block_hash)?;
+        let shard_layout = self.epoch_manager.get_shard_layout(&epoch_id)?;
+        let shard_uid = ShardUId::from_shard_id_and_layout(shard_id, &shard_layout);
+        let mut chunk_extras: HashMap<SpiceChunkId, Arc<ChunkExtra>> = HashMap::new();
+        for info in &shard_uncertified_chunks {
+            if let Ok(extra) = self
+                .chain_store
+                .chunk_store()
+                .get_chunk_extra(&info.chunk_id.block_hash, &shard_uid)
+            {
+                chunk_extras.insert(info.chunk_id.clone(), extra);
+            }
+        }
+        // Walk backward from head, collecting certified execution results
+        // from blocks' core statements for any chunks not resolved above.
+        let mut current_hash = self.chain_store.head()?.last_block_hash;
+        while chunk_extras.len() < shard_uncertified_chunks.len() && current_hash != *block_hash {
+            let block = self.chain_store.get_block(&current_hash)?;
+            for (cid, result) in block.spice_core_statements().iter_execution_results() {
+                if cid.shard_id == shard_id && !chunk_extras.contains_key(cid) {
+                    chunk_extras.insert(cid.clone(), Arc::new(result.chunk_extra.clone()));
+                }
+            }
+            current_hash = *block.header().prev_hash();
+        }
+
+        let mut height_proposals: Vec<(BlockHeight, Vec<ValidatorStake>)> = Vec::new();
+        for info in &shard_uncertified_chunks {
+            let Some(chunk_extra) = chunk_extras.get(&info.chunk_id) else {
+                return Err(Error::Other(format!("no chunk extra for {:?}", info.chunk_id)));
+            };
+            let proposals: Vec<ValidatorStake> = chunk_extra.validator_proposals().collect();
+            if !proposals.is_empty() {
+                let height = self.chain_store.get_block_height(&info.chunk_id.block_hash)?;
+                height_proposals.push((height, proposals));
+            }
+        }
+        height_proposals.sort_by_key(|(h, _)| *h);
+        debug_assert!(
+            height_proposals.windows(2).all(|w| w[0].0 != w[1].0),
+            "multiple uncertified chunks at the same height for shard {shard_id}"
+        );
+        Ok(height_proposals.into_iter().flat_map(|(_, proposals)| proposals).collect())
+    }
+
+    /// Returns validator proposals to use as `prev_validator_proposals` when
+    /// constructing `NewChunkData`. At epoch boundaries, returns proposals from
+    /// uncertified chunks; otherwise returns an empty vec.
+    pub fn prev_validator_proposals(
+        &self,
+        prev_block_hash: &CryptoHash,
+        shard_id: ShardId,
+    ) -> Result<Vec<ValidatorStake>, Error> {
+        if self.epoch_manager.is_next_block_epoch_start(prev_block_hash)? {
+            self.get_uncertified_validator_proposals(prev_block_hash, shard_id)
+        } else {
+            Ok(vec![])
+        }
+    }
+
     pub fn get_execution_results_by_shard_id(
         &self,
         block_header: &BlockHeader,

chain/chain/src/stateless_validation/spice_chunk_validation.rs

@@ -22,6 +22,7 @@ use tracing::Span;
 use crate::chain::{NewChunkData, NewChunkResult, ShardContext, StorageContext, apply_new_chunk};
 use crate::sharding::{get_receipts_shuffle_salt, shuffle_receipt_proofs};
 use crate::spice_chunk_application::build_spice_apply_chunk_block_context;
+use crate::spice_core::SpiceCoreReader;
 use crate::store::filter_incoming_receipts_for_shard;
 use crate::types::{RuntimeAdapter, StorageDataSource};
 use crate::{Chain, ChainStore};
@@ -37,6 +38,7 @@ pub fn spice_pre_validate_chunk_state_witness(
     prev_execution_results: &BlockExecutionResults,
     epoch_manager: &dyn EpochManagerAdapter,
     store: &ChainStore,
+    core_reader: &SpiceCoreReader,
 ) -> Result<SpicePreValidationOutput, Error> {
     assert_eq!(block.hash(), &state_witness.chunk_id().block_hash);
     let epoch_id = epoch_manager.get_epoch_id(block.header().hash())?;
@@ -140,10 +142,12 @@ pub fn spice_pre_validate_chunk_state_witness(
             prev_execution_results,
             epoch_manager,
         )?;
+        let prev_validator_proposals =
+            core_reader.prev_validator_proposals(prev_block.hash(), shard_id)?;
         NewChunkData {
             gas_limit: prev_chunk_chunk_extra.gas_limit(),
             prev_state_root: *prev_chunk_chunk_extra.state_root(),
-            prev_validator_proposals: prev_chunk_chunk_extra.validator_proposals().collect(),
+            prev_validator_proposals,
             chunk_hash: if chunk_header.is_new_chunk(block.header().height()) {
                 Some(chunk_header.chunk_hash().clone())
             } else {
@@ -585,6 +589,7 @@ mod tests {
             &BlockExecutionResults(HashMap::new()),
             test_chain.chain.epoch_manager.as_ref(),
             test_chain.chain.chain_store(),
+            &test_chain.chain.spice_core_reader,
         );
 
         let error_message = unwrap_error_message(result);
@@ -628,6 +633,7 @@ mod tests {
             &BlockExecutionResults(HashMap::new()),
             test_chain.chain.epoch_manager.as_ref(),
             test_chain.chain.chain_store(),
+            &test_chain.chain.spice_core_reader,
         );
 
         let error_message = unwrap_error_message(result);
@@ -1080,6 +1086,7 @@ mod tests {
                 &prev_execution_results,
                 self.chain.epoch_manager.as_ref(),
                 self.chain.chain_store(),
+                &self.chain.spice_core_reader,
             )
             .unwrap();
 
@@ -1105,6 +1112,7 @@ mod tests {
                 &prev_execution_results,
                 self.chain.epoch_manager.as_ref(),
                 self.chain.chain_store(),
+                &self.chain.spice_core_reader,
             )
         }
 
@@ -1140,11 +1148,17 @@ mod tests {
 
             let prev_execution_result = prev_execution_results.0.get(&self.shard_id()).unwrap();
             let prev_chunk_chunk_extra = &prev_execution_result.chunk_extra;
+            let prev_block_hash = block.header().prev_hash();
+            let prev_validator_proposals = self
+                .chain
+                .spice_core_reader
+                .prev_validator_proposals(prev_block_hash, self.shard_id())
+                .unwrap();
             let txs_validity = std::iter::repeat_n(true, transactions.len()).collect_vec();
             let new_chunk_data = NewChunkData {
                 gas_limit: prev_chunk_chunk_extra.gas_limit(),
                 prev_state_root: *prev_chunk_chunk_extra.state_root(),
-                prev_validator_proposals: prev_chunk_chunk_extra.validator_proposals().collect(),
+                prev_validator_proposals,
                 chunk_hash: None,
                 transactions: SignedValidPeriodTransactions::new(transactions, txs_validity),
                 receipts,