async fn sync_headers_reverse(
        &self,
        head: Arc<Tipset>,
        to: &Tipset,
    ) -> Result<Vec<Arc<Tipset>>, Error> {
        info!("Syncing headers from: {:?}", head.key());
        self.state.write().await.set_epoch(to.epoch());

        let mut accepted_blocks: Vec<Cid> = Vec::new();

        let sync_len = head.epoch() - to.epoch();
        if sync_len < 0 {
            return Err(Error::Other(
                "Target tipset must be after heaviest".to_string(),
            ));
        }

        // invariant: never empty, only appended to
        let mut return_set = Vec::with_capacity(sync_len as usize + 1);
        return_set.push(head);

        // Loop until most recent tipset height is less than or equal to tipset height
        'sync: loop {
            let cur_ts = return_set.last().unwrap();

            // Check if parent cids exist in bad block caches
            self.validate_tipset_against_cache(cur_ts.parents(), &accepted_blocks)
                .await?;

            if cur_ts.epoch() <= to.epoch() {
                // Current tipset is less than epoch of tipset syncing toward
                break;
            }

            // Try to load parent tipset from local storage
            if let Ok(ts) = self.chain_store().tipset_from_keys(cur_ts.parents()).await {
                // Add blocks in tipset to accepted chain and push the tipset to return set
                accepted_blocks.extend_from_slice(ts.cids());
                return_set.push(ts);
                continue;
            }

            let epoch_diff = cur_ts.epoch() - to.epoch();
            debug!("ChainExchange from: {} to {}", cur_ts.epoch(), to.epoch());
            // TODO tweak request window when socket frame is tested
            let window = min(epoch_diff, self.req_window);

            // Load blocks from network using chain_exchange
            // TODO consider altering window size before returning error for failed sync.
            let tipsets = self
                .network
                .chain_exchange_headers(None, cur_ts.parents(), window as u64)
                .await?;

            info!(
                "Got tipsets: Height: {}, Len: {}",
                tipsets[0].epoch(),
                tipsets.len()
            );

            // Loop through each tipset received from network
            for ts in tipsets {
                if ts.epoch() < to.epoch() {
                    // Break out of sync loop if epoch lower than to tipset
                    // This should not be hit if response from server is correct
                    break 'sync;
                }
                // Check Cids of blocks against bad block cache
                self.validate_tipset_against_cache(&ts.key(), &accepted_blocks)
                    .await?;

                accepted_blocks.extend_from_slice(ts.cids());
                self.state.write().await.set_epoch(ts.epoch());
                // Add tipset to vector of tipsets to return
                return_set.push(ts);
            }
        }

        let last_ts = return_set.last().unwrap();
        if last_ts.parents() == to.parents() {
            // block received part of same tipset as best block
            // This removes need to sync fork
            return Ok(return_set);
        }

        // Check if local chain was fork
        if last_ts.key() != to.key() {
            info!("Local chain was fork. Syncing fork...");
            // add fork into return set
            let fork = self.sync_fork(&last_ts, &to).await?;
            info!("Fork Synced");
            return_set.extend(fork);
        }
        info!("Sync Header reverse complete");
        Ok(return_set)
    }
/// Syncs chain data and persists it to blockstore
Syncs chain data and persists it to blockstore
Syncs chain data and persists it to blockstore
async fn sync_headers_reverse(
        &self,
        head: Arc<Tipset>,
        to: &Tipset,
    ) -> Result<Vec<Arc<Tipset>>, Error> {
        info!("Syncing headers from: {:?}", head.key());
        self.state.write().await.set_epoch(to.epoch());
        let mut accepted_blocks: Vec<Cid> = Vec::new();
        let sync_len = head.epoch() - to.epoch();
        if sync_len < 0 {
            return Err(Error::Other(
                "Target tipset must be after heaviest".to_string(),
            ));
        }
        // invariant: never empty, only appended to
        let mut return_set = Vec::with_capacity(sync_len as usize + 1);
        return_set.push(head);
        // Loop until most recent tipset height is less than or equal to tipset height
        'sync: loop {
            let cur_ts = return_set.last().unwrap();
            // Check if parent cids exist in bad block caches
            self.validate_tipset_against_cache(cur_ts.parents(), &accepted_blocks)
                .await?;
            if cur_ts.epoch() <= to.epoch() {
                // Current tipset is less than epoch of tipset syncing toward
                break;
            }
            // Try to load parent tipset from local storage
            if let Ok(ts) = self.chain_store().tipset_from_keys(cur_ts.parents()).await {
                // Add blocks in tipset to accepted chain and push the tipset to return set
                accepted_blocks.extend_from_slice(ts.cids());
                return_set.push(ts);
                continue;
            }
            let epoch_diff = cur_ts.epoch() - to.epoch();
            debug!("ChainExchange from: {} to {}", cur_ts.epoch(), to.epoch());
            // TODO tweak request window when socket frame is tested
            let window = min(epoch_diff, self.req_window);
            // Load blocks from network using chain_exchange
            // TODO consider altering window size before returning error for failed sync.
            let tipsets = self
                .network
                .chain_exchange_headers(None, cur_ts.parents(), window as u64)
                .await?;
            info!(
                "Got tipsets: Height: {}, Len: {}",
                tipsets[0].epoch(),
                tipsets.len()
            );
            // Loop through each tipset received from network
            for ts in tipsets {
                if ts.epoch() < to.epoch() {
                    // Break out of sync loop if epoch lower than to tipset
                    // This should not be hit if response from server is correct
                    break 'sync;
                }
                // Check Cids of blocks against bad block cache
                self.validate_tipset_against_cache(&ts.key(), &accepted_blocks)
                    .await?;
                accepted_blocks.extend_from_slice(ts.cids());
                self.state.write().await.set_epoch(ts.epoch());
                // Add tipset to vector of tipsets to return
                return_set.push(ts);
            }
        }
        let last_ts = return_set.last().unwrap();
        if last_ts.parents() == to.parents() {
            // block received part of same tipset as best block
            // This removes need to sync fork
            return Ok(return_set);
        }
        // Check if local chain was fork
        if last_ts.key() != to.key() {
            info!("Local chain was fork. Syncing fork...");
            // add fork into return set
            let fork = self.sync_fork(&last_ts, &to).await?;
            info!("Fork Synced");
            return_set.extend(fork);
        }
        info!("Sync Header reverse complete");
        Ok(return_set)
    }
async fn sync_headers_reverse ( & self , head : Arc < Tipset > , to : & Tipset , ) -> Result < Vec < Arc < Tipset > > , Error > { info ! ( " " head key ( ) ) ; self . state . write ( ) . await . set_epoch ( to . epoch ( ) ) ; let mut accepted_blocks : Vec < Cid > = Vec :: new ( ) ; let sync_len = head . epoch ( ) - to . epoch ( ) ; if sync_len < 0 { return Err ( Error :: Other ( " " . to_string ( ) , ) ) ; } // invariant: never empty, only appended to let mut return_set = Vec :: with_capacity ( sync_len as usize + 1 ) ; return_set . push ( head ) ; // Loop until most recent tipset height is less than or equal to tipset height ' sync : loop { let cur_ts = return_set . last ( ) . unwrap ( ) ; // Check if parent cids exist in bad block caches self . validate_tipset_against_cache ( cur_ts . parents ( ) , & accepted_blocks ) . await ? ; if cur_ts . epoch ( ) <= to . epoch ( ) { // Current tipset is less than epoch of tipset syncing toward break ; } // Try to load parent tipset from local storage if let Ok ( ts ) = self . chain_store ( ) . tipset_from_keys ( cur_ts . parents ( ) ) . await { // Add blocks in tipset to accepted chain and push the tipset to return set accepted_blocks . extend_from_slice ( ts . cids ( ) ) ; return_set . push ( ts ) ; continue ; } let epoch_diff = cur_ts . epoch ( ) - to . epoch ( ) ; debug ! ( " " cur_ts epoch ( ) to epoch ( ) ) ; // TODO tweak request window when socket frame is tested let window = min ( epoch_diff , self . req_window ) ; // Load blocks from network using chain_exchange // TODO consider altering window size before returning error for failed sync. let tipsets = self . network . chain_exchange_headers ( None , cur_ts . parents ( ) , window as u64 ) . await ? ; info ! ( " " tipsets [ 0 ] epoch ( ) tipsets len ( ) ) ; // Loop through each tipset received from network for ts in tipsets { if ts . epoch ( ) < to . epoch ( ) { // Break out of sync loop if epoch lower than to tipset // This should not be hit if response from server is correct break ' sync ; } // Check Cids of blocks against bad block cache self . validate_tipset_against_cache ( & ts . key ( ) , & accepted_blocks ) . await ? ; accepted_blocks . extend_from_slice ( ts . cids ( ) ) ; self . state . write ( ) . await . set_epoch ( ts . epoch ( ) ) ; // Add tipset to vector of tipsets to return return_set . push ( ts ) ; } } let last_ts = return_set . last ( ) . unwrap ( ) ; if last_ts . parents ( ) == to . parents ( ) { // block received part of same tipset as best block // This removes need to sync fork return Ok ( return_set ) ; } // Check if local chain was fork if last_ts . key ( ) != to . key ( ) { info ! ( " " ) ; // add fork into return set let fork = self . sync_fork ( & last_ts , & to ) . await ? ; info ! ( " " ) ; return_set . extend ( fork ) ; } info ! ( " " ) ; Ok ( return_set ) }
Syncs chain data and persists it to blockstore
Syncs chain data and persists it to blockstore
&self
head
Arc<Tipset>
to
&Tipset