pub fn sync_messages(&self) -> Instrumented<tokio::task::JoinHandle<color_eyre::Result<()>>> {
        let span = info_span!("MessageContractSync");

        let db = self.db.clone();
        let indexer = self.indexer.clone();
        let indexed_height = self.indexed_height.clone();

        let from_height = self.from_height;
        let chunk_size = self.chunk_size;

        tokio::spawn(async move {
            let mut next_height: u32 = db
                .retrieve_decodable("", MESSAGES_LAST_INSPECTED)
                .expect("db failure")
                .unwrap_or(from_height);
            info!(
                next_height = next_height,
                "resuming indexer from {}", next_height
            );

            loop {
                indexed_height.set(next_height as i64);
                let tip = indexer.get_block_number().await?;
                let candidate = next_height + chunk_size;
                let to = min(tip, candidate);

                info!(
                    next_height = next_height,
                    to = to,
                    "indexing block heights {}...{}",
                    next_height,
                    to
                );

                let messages = indexer.fetch_sorted_messages(next_height, to).await?;

                for message in messages {
                    db.store_raw_committed_message(&message)?;

                    let committed_message: CommittedMessage = message.try_into()?;
                    info!(
                        "Stored new message in db. Leaf index: {}. Origin: {}. Destination: {}. Nonce: {}.",
                        &committed_message.leaf_index,
                        &committed_message.message.origin,
                        &committed_message.message.destination,
                        &committed_message.message.nonce
                    );
                }

                db
                    .store_encodable("", MESSAGES_LAST_INSPECTED, &next_height)?;
                next_height = to;
                // sleep here if we've caught up
                if to == tip {
                    sleep(Duration::from_secs(100)).await;
                }
            }
        })
        .instrument(span)
    }
/// Spawn task that continuously looks for new on-chain messages and stores
/// them in db
Spawn task that continuously looks for new on-chain messages and stores
them in db
Spawn task that continuously looks for new on - chain messages and stores them in db
pub fn sync_messages(&self) -> Instrumented<tokio::task::JoinHandle<color_eyre::Result<()>>> {
        let span = info_span!("MessageContractSync");
        let db = self.db.clone();
        let indexer = self.indexer.clone();
        let indexed_height = self.indexed_height.clone();
        let from_height = self.from_height;
        let chunk_size = self.chunk_size;
        tokio::spawn(async move {
            let mut next_height: u32 = db
                .retrieve_decodable("", MESSAGES_LAST_INSPECTED)
                .expect("db failure")
                .unwrap_or(from_height);
            info!(
                next_height = next_height,
                "resuming indexer from {}", next_height
            );
            loop {
                indexed_height.set(next_height as i64);
                let tip = indexer.get_block_number().await?;
                let candidate = next_height + chunk_size;
                let to = min(tip, candidate);
                info!(
                    next_height = next_height,
                    to = to,
                    "indexing block heights {}...{}",
                    next_height,
                    to
                );
                let messages = indexer.fetch_sorted_messages(next_height, to).await?;
                for message in messages {
                    db.store_raw_committed_message(&message)?;
                    let committed_message: CommittedMessage = message.try_into()?;
                    info!(
                        "Stored new message in db. Leaf index: {}. Origin: {}. Destination: {}. Nonce: {}.",
                        &committed_message.leaf_index,
                        &committed_message.message.origin,
                        &committed_message.message.destination,
                        &committed_message.message.nonce
                    );
                }
                db
                    .store_encodable("", MESSAGES_LAST_INSPECTED, &next_height)?;
                next_height = to;
                // sleep here if we've caught up
                if to == tip {
                    sleep(Duration::from_secs(100)).await;
                }
            }
        })
        .instrument(span)
    }
pub fn sync_messages ( & self ) -> Instrumented < tokio :: task :: JoinHandle < color_eyre :: Result < ( ) > > > { let span = info_span ! ( " " ) ; let db = self . db . clone ( ) ; let indexer = self . indexer . clone ( ) ; let indexed_height = self . indexed_height . clone ( ) ; let from_height = self . from_height ; let chunk_size = self . chunk_size ; tokio :: spawn ( async move { let mut next_height : u32 = db . retrieve_decodable ( " " , MESSAGES_LAST_INSPECTED ) . expect ( " " ) . unwrap_or ( from_height ) ; info ! ( next_height next_height " " next_height ) ; loop { indexed_height . set ( next_height as i64 ) ; let tip = indexer . get_block_number ( ) . await ? ; let candidate = next_height + chunk_size ; let to = min ( tip , candidate ) ; info ! ( next_height next_height to to " " next_height to ) ; let messages = indexer . fetch_sorted_messages ( next_height , to ) . await ? ; for message in messages { db . store_raw_committed_message ( & message ) ? ; let committed_message : CommittedMessage = message . try_into ( ) ? ; info ! ( " " committed_message leaf_index committed_message message origin committed_message message destination committed_message message nonce ) ; } db . store_encodable ( " " , MESSAGES_LAST_INSPECTED , & next_height ) ? ; next_height = to ; // sleep here if we've caught up if to == tip { sleep ( Duration :: from_secs ( 100 ) ) . await ; } } } ) . instrument ( span ) }
Spawn task that continuously looks for new on-chain messages and stores
them in db
Spawn task that continuously looks for new on - chain messages and stores them in db
&self
&self