async fn start(mut self, timestamper: &Timestamper) -> Result<(), SourceError> {
        // Buffer rows from snapshot to retract and retry, if initial snapshot fails.
        // Postgres sources cannot proceed without a successful snapshot.
        {
            let mut snapshot_tx = timestamper.start_tx().await;
            loop {
                let file =
                    tokio::fs::File::from_std(tempfile::tempfile().map_err(|e| SourceError {
                        source_name: self.source_name.clone(),
                        error: SourceErrorDetails::FileIO(e.to_string()),
                    })?);
                let mut writer = tokio::io::BufWriter::new(file);
                match self.produce_snapshot(&mut snapshot_tx, &mut writer).await {
                    Ok(_) => {
                        log::info!(
                            "replication snapshot for source {} succeeded",
                            &self.source_name
                        );
                        break;
                    }
                    Err(ReplicationError::Recoverable(e)) => {
                        writer.flush().await.map_err(|e| SourceError {
                            source_name: self.source_name.clone(),
                            error: SourceErrorDetails::Initialization(e.to_string()),
                        })?;
                        log::warn!(
                            "replication snapshot for source {} failed, retrying: {}",
                            &self.source_name,
                            e
                        );
                        let reader = BufReader::new(writer.into_inner().into_std().await);
                        self.revert_snapshot(&mut snapshot_tx, reader)
                            .await
                            .map_err(|e| SourceError {
                                source_name: self.source_name.clone(),
                                error: SourceErrorDetails::FileIO(e.to_string()),
                            })?;
                    }
                    Err(ReplicationError::Fatal(e)) => {
                        return Err(SourceError {
                            source_name: self.source_name,
                            error: SourceErrorDetails::Initialization(e.to_string()),
                        })
                    }
                }

                // TODO(petrosagg): implement exponential back-off
                tokio::time::sleep(Duration::from_secs(3)).await;
            }
        }

        loop {
            match self.produce_replication(timestamper).await {
                Err(ReplicationError::Recoverable(e)) => {
                    log::warn!(
                        "replication for source {} interrupted, retrying: {}",
                        &self.source_name,
                        e
                    )
                }
                Err(ReplicationError::Fatal(e)) => {
                    return Err(SourceError {
                        source_name: self.source_name,
                        error: SourceErrorDetails::FileIO(e.to_string()),
                    })
                }
                Ok(_) => unreachable!("replication stream cannot exit without an error"),
            }

            // TODO(petrosagg): implement exponential back-off
            tokio::time::sleep(Duration::from_secs(3)).await;
            log::info!("resuming replication for source {}", &self.source_name);
        }
    }
/// The top-level control of the state machine and retry logic
The top-level control of the state machine and retry logic
The top - level control of the state machine and retry logic
async fn start(mut self, timestamper: &Timestamper) -> Result<(), SourceError> {
        // Buffer rows from snapshot to retract and retry, if initial snapshot fails.
        // Postgres sources cannot proceed without a successful snapshot.
        {
            let mut snapshot_tx = timestamper.start_tx().await;
            loop {
                let file =
                    tokio::fs::File::from_std(tempfile::tempfile().map_err(|e| SourceError {
                        source_name: self.source_name.clone(),
                        error: SourceErrorDetails::FileIO(e.to_string()),
                    })?);
                let mut writer = tokio::io::BufWriter::new(file);
                match self.produce_snapshot(&mut snapshot_tx, &mut writer).await {
                    Ok(_) => {
                        log::info!(
                            "replication snapshot for source {} succeeded",
                            &self.source_name
                        );
                        break;
                    }
                    Err(ReplicationError::Recoverable(e)) => {
                        writer.flush().await.map_err(|e| SourceError {
                            source_name: self.source_name.clone(),
                            error: SourceErrorDetails::Initialization(e.to_string()),
                        })?;
                        log::warn!(
                            "replication snapshot for source {} failed, retrying: {}",
                            &self.source_name,
                            e
                        );
                        let reader = BufReader::new(writer.into_inner().into_std().await);
                        self.revert_snapshot(&mut snapshot_tx, reader)
                            .await
                            .map_err(|e| SourceError {
                                source_name: self.source_name.clone(),
                                error: SourceErrorDetails::FileIO(e.to_string()),
                            })?;
                    }
                    Err(ReplicationError::Fatal(e)) => {
                        return Err(SourceError {
                            source_name: self.source_name,
                            error: SourceErrorDetails::Initialization(e.to_string()),
                        })
                    }
                }
                // TODO(petrosagg): implement exponential back-off
                tokio::time::sleep(Duration::from_secs(3)).await;
            }
        }
        loop {
            match self.produce_replication(timestamper).await {
                Err(ReplicationError::Recoverable(e)) => {
                    log::warn!(
                        "replication for source {} interrupted, retrying: {}",
                        &self.source_name,
                        e
                    )
                }
                Err(ReplicationError::Fatal(e)) => {
                    return Err(SourceError {
                        source_name: self.source_name,
                        error: SourceErrorDetails::FileIO(e.to_string()),
                    })
                }
                Ok(_) => unreachable!("replication stream cannot exit without an error"),
            }
            // TODO(petrosagg): implement exponential back-off
            tokio::time::sleep(Duration::from_secs(3)).await;
            log::info!("resuming replication for source {}", &self.source_name);
        }
    }
async fn start ( mut self , timestamper : & Timestamper ) -> Result < ( ) , SourceError > { // Buffer rows from snapshot to retract and retry, if initial snapshot fails. // Postgres sources cannot proceed without a successful snapshot. { let mut snapshot_tx = timestamper . start_tx ( ) . await ; loop { let file = tokio :: fs :: File :: from_std ( tempfile :: tempfile ( ) . map_err ( | e | SourceError { source_name : self . source_name . clone ( ) , error : SourceErrorDetails :: FileIO ( e . to_string ( ) ) , } ) ? ) ; let mut writer = tokio :: io :: BufWriter :: new ( file ) ; match self . produce_snapshot ( & mut snapshot_tx , & mut writer ) . await { Ok ( _ ) => { log :: info ! ( " " self source_name ) ; break ; } Err ( ReplicationError :: Recoverable ( e ) ) => { writer . flush ( ) . await . map_err ( | e | SourceError { source_name : self . source_name . clone ( ) , error : SourceErrorDetails :: Initialization ( e . to_string ( ) ) , } ) ? ; log :: warn ! ( " " self source_name e ) ; let reader = BufReader :: new ( writer . into_inner ( ) . into_std ( ) . await ) ; self . revert_snapshot ( & mut snapshot_tx , reader ) . await . map_err ( | e | SourceError { source_name : self . source_name . clone ( ) , error : SourceErrorDetails :: FileIO ( e . to_string ( ) ) , } ) ? ; } Err ( ReplicationError :: Fatal ( e ) ) => { return Err ( SourceError { source_name : self . source_name , error : SourceErrorDetails :: Initialization ( e . to_string ( ) ) , } ) } } // TODO(petrosagg): implement exponential back-off tokio :: time :: sleep ( Duration :: from_secs ( 3 ) ) . await ; } } loop { match self . produce_replication ( timestamper ) . await { Err ( ReplicationError :: Recoverable ( e ) ) => { log :: warn ! ( " " self source_name e ) } Err ( ReplicationError :: Fatal ( e ) ) => { return Err ( SourceError { source_name : self . source_name , error : SourceErrorDetails :: FileIO ( e . to_string ( ) ) , } ) } Ok ( _ ) => unreachable ! ( " " ) , } // TODO(petrosagg): implement exponential back-off tokio :: time :: sleep ( Duration :: from_secs ( 3 ) ) . await ; log :: info ! ( " " self source_name ) ; } }
The top-level control of the state machine and retry logic
The top - level control of the state machine and retry logic
mut self
timestamper
&Timestamper
mut self
timestamper
&Timestamper