async fn sync_messages_check_state(&self, tipsets: Vec<Arc<Tipset>>) -> Result<(), Error> {
        let mut ts_iter = tipsets.into_iter().rev();
        // Currently syncing 1 height at a time, no reason for us to sync more
        const REQUEST_WINDOW: usize = 1;

        while let Some(ts) = ts_iter.next() {
            // check storage first to see if we have full tipset
            match self.chain_store().fill_tipset(&ts) {
                Some(fts) => {
                    // full tipset found in storage; validate and continue
                    let curr_epoch = fts.epoch();
                    self.validate_tipset(fts).await?;
                    self.state.write().await.set_epoch(curr_epoch);
                }
                None => {
                    // no full tipset in storage; request messages via chain_exchange

                    let batch_size = REQUEST_WINDOW;
                    debug!(
                        "ChainExchange message sync tipsets: epoch: {}, len: {}",
                        ts.epoch(),
                        batch_size
                    );

                    // receive tipset bundle from block sync
                    let compacted_messages = self
                        .network
                        .chain_exchange_messages(None, ts.key(), batch_size as u64)
                        .await?;

                    // Chain current tipset with iterator
                    let mut bs_iter = std::iter::once(ts).chain(&mut ts_iter);

                    // since the bundle only has messages, we have to put the headers in them
                    for messages in compacted_messages {
                        let t = bs_iter.next().ok_or_else(|| {
                            Error::Other("Messages returned exceeded tipsets in chain".to_string())
                        })?;

                        let bundle = TipsetBundle {
                            blocks: t.blocks().to_vec(),
                            messages: Some(messages),
                        };
                        // construct full tipsets from fetched messages
                        let fts = FullTipset::try_from(&bundle).map_err(Error::Other)?;

                        // validate tipset and messages
                        let curr_epoch = fts.epoch();
                        self.validate_tipset(fts).await?;
                        self.state.write().await.set_epoch(curr_epoch);

                        // store messages
                        if let Some(m) = bundle.messages {
                            let bs = self.state_manager.blockstore();
                            chain::persist_objects(bs, &m.bls_msgs)?;
                            chain::persist_objects(bs, &m.secp_msgs)?;
                        } else {
                            warn!("Chain Exchange request for messages returned null messages");
                        }
                    }
                }
            }
        }

        Ok(())
    }
/// Syncs messages by first checking state for message existence otherwise fetches messages from
/// chain exchange.
Syncs messages by first checking state for message existence otherwise fetches messages from
chain exchange.
Syncs messages by first checking state for message existence otherwise fetches messages from chain exchange .
async fn sync_messages_check_state(&self, tipsets: Vec<Arc<Tipset>>) -> Result<(), Error> {
        let mut ts_iter = tipsets.into_iter().rev();
        // Currently syncing 1 height at a time, no reason for us to sync more
        const REQUEST_WINDOW: usize = 1;
        while let Some(ts) = ts_iter.next() {
            // check storage first to see if we have full tipset
            match self.chain_store().fill_tipset(&ts) {
                Some(fts) => {
                    // full tipset found in storage; validate and continue
                    let curr_epoch = fts.epoch();
                    self.validate_tipset(fts).await?;
                    self.state.write().await.set_epoch(curr_epoch);
                }
                None => {
                    // no full tipset in storage; request messages via chain_exchange
                    let batch_size = REQUEST_WINDOW;
                    debug!(
                        "ChainExchange message sync tipsets: epoch: {}, len: {}",
                        ts.epoch(),
                        batch_size
                    );
                    // receive tipset bundle from block sync
                    let compacted_messages = self
                        .network
                        .chain_exchange_messages(None, ts.key(), batch_size as u64)
                        .await?;
                    // Chain current tipset with iterator
                    let mut bs_iter = std::iter::once(ts).chain(&mut ts_iter);
                    // since the bundle only has messages, we have to put the headers in them
                    for messages in compacted_messages {
                        let t = bs_iter.next().ok_or_else(|| {
                            Error::Other("Messages returned exceeded tipsets in chain".to_string())
                        })?;
                        let bundle = TipsetBundle {
                            blocks: t.blocks().to_vec(),
                            messages: Some(messages),
                        };
                        // construct full tipsets from fetched messages
                        let fts = FullTipset::try_from(&bundle).map_err(Error::Other)?;
                        // validate tipset and messages
                        let curr_epoch = fts.epoch();
                        self.validate_tipset(fts).await?;
                        self.state.write().await.set_epoch(curr_epoch);
                        // store messages
                        if let Some(m) = bundle.messages {
                            let bs = self.state_manager.blockstore();
                            chain::persist_objects(bs, &m.bls_msgs)?;
                            chain::persist_objects(bs, &m.secp_msgs)?;
                        } else {
                            warn!("Chain Exchange request for messages returned null messages");
                        }
                    }
                }
            }
        }
        Ok(())
    }
async fn sync_messages_check_state ( & self , tipsets : Vec < Arc < Tipset > > ) -> Result < ( ) , Error > { let mut ts_iter = tipsets . into_iter ( ) . rev ( ) ; // Currently syncing 1 height at a time, no reason for us to sync more const REQUEST_WINDOW : usize = 1 ; while let Some ( ts ) = ts_iter . next ( ) { // check storage first to see if we have full tipset match self . chain_store ( ) . fill_tipset ( & ts ) { Some ( fts ) => { // full tipset found in storage; validate and continue let curr_epoch = fts . epoch ( ) ; self . validate_tipset ( fts ) . await ? ; self . state . write ( ) . await . set_epoch ( curr_epoch ) ; } None => { // no full tipset in storage; request messages via chain_exchange let batch_size = REQUEST_WINDOW ; debug ! ( " " ts epoch ( ) batch_size ) ; // receive tipset bundle from block sync let compacted_messages = self . network . chain_exchange_messages ( None , ts . key ( ) , batch_size as u64 ) . await ? ; // Chain current tipset with iterator let mut bs_iter = std :: iter :: once ( ts ) . chain ( & mut ts_iter ) ; // since the bundle only has messages, we have to put the headers in them for messages in compacted_messages { let t = bs_iter . next ( ) . ok_or_else ( | | { Error :: Other ( " " . to_string ( ) ) } ) ? ; let bundle = TipsetBundle { blocks : t . blocks ( ) . to_vec ( ) , messages : Some ( messages ) , } ; // construct full tipsets from fetched messages let fts = FullTipset :: try_from ( & bundle ) . map_err ( Error :: Other ) ? ; // validate tipset and messages let curr_epoch = fts . epoch ( ) ; self . validate_tipset ( fts ) . await ? ; self . state . write ( ) . await . set_epoch ( curr_epoch ) ; // store messages if let Some ( m ) = bundle . messages { let bs = self . state_manager . blockstore ( ) ; chain :: persist_objects ( bs , & m . bls_msgs ) ? ; chain :: persist_objects ( bs , & m . secp_msgs ) ? ; } else { warn ! ( " " ) ; } } } } } Ok ( ( ) ) }
Syncs messages by first checking state for message existence otherwise fetches messages from
chain exchange.
Syncs messages by first checking state for message existence otherwise fetches messages from chain exchange .
&self
tipsets
Vec<Arc<Tipset>>