let mut err: Option<String> = None;

            debug!("reconnect");

                    *db_.lock().await = Some(db);

                    err = None;
                    debug!("reconnected");
                    {
                        *db_.lock().await = Some(db);
                    }
                    debug!("waiting for connection to end");

                    {
                        *db_.lock().await = None;
                    }

                    error!("{:?}", e);

                    let ee = format!("{}", e);
                    if let Some(ee_) = err.take() {
                        if ee_ != ee {
                            error!("{}", e);
                            err = Some(ee)
                        } else {
                            err = Some(ee_)
                        }
                    } else {
                        error!("{}", e);
                        err = Some(ee)
                    }

    let mut race = tokio::spawn(race(host.clone(), is_leader.clone()));

    let mut race = tokio::spawn(race(
        host.clone(),
        path.clone(),
        is_leader.clone(),
        db.clone(),
    ));

    trace!("handle: {:?} {:?}", is_leader.borrow(), name);

async fn race(host: String, mut leader: Receiver<String>) -> Result<(), anyhow::Error> {

async fn race(
    host: String,
    path: String,
    mut leader: Receiver<String>,
    pool: Arc<Mutex<Option<tokio_postgres::Client>>>,
) -> Result<(), anyhow::Error> {

        let has_db = pool.lock().await.is_some();
        debug!("has_db = {:?}", has_db);
        if !has_db {
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            continue;
        }

        debug!("lock = {:?}", lock);

        info!(
            "Lock acquired: {:?} {:?}",
            std::str::from_utf8(leader_key.name()),
            std::str::from_utf8(leader_key.key())
        );

        debug!("Lock acquired {:?}", leader_key);
        promote(&path).await;

    let mut was_leader = None;

    let mut was_leader = {
        if let Some(pool_) = pool.lock().await.take() {
            let row = pool_.query_one("SELECT pg_is_in_recovery()", &[]).await?;
            let is_in_recovery: bool = row.get(0);
            *pool.lock().await = Some(pool_);
            Some(!is_in_recovery)
        } else {
            None
        }
    };
    debug!("was_leader = {:?}", was_leader);

        debug!("obs: {:?}", m);

        debug!("leader = {:?}", leader);

        debug!("leader = {:?}, host = {:?}", leader, host);

        if is_leader {
            info!("The leader is {:?}", leader);
        } else {
            info!("I'm the leader");
        }

        if let Some(pool_) = pool.lock().await.take() {

        let pool_ = { pool.lock().await.take() };
        if let Some(pool_) = pool_ {

        debug!("is_leader = {:?}, was_leader = {:?}", is_leader, was_leader);

        } else if !is_leader && was_leader == Some(true) {
            rewind(&path, port, &db_name, &leader).await

        } else if !is_leader && was_leader != Some(false) {
            let pool_ = { pool.lock().await.take() };
            let recov: bool = if let Some(pool_) = pool_ {
                let row = pool_.query_one("SELECT pg_is_in_recovery()", &[]).await?;
                *pool.lock().await = Some(pool_);
                row.get(0)
            } else {
                false
            };
            debug!("recov = {:?}", recov);
            if !recov {
                rewind(&path, port, &db_name, leader).await;
            }

    debug!("promoting");

    Command::new("pg_ctl")

    let out = Command::new("pg_ctl")

    use std::io::Write;
    std::io::stdout().write_all(&out.stdout).unwrap();
    std::io::stderr().write_all(&out.stderr).unwrap();

    debug!("rewinding");

    Command::new("systemctl")

    let out = Command::new("systemctl")

    Command::new("pg_rewind")
        .args([
            "-D",
            &path,
            "-R",
            "--source-server",
            &format!(
                "port={} user=postgres dbname={} host={}",
                port, db_name, leader
            ),
        ])
        .output()
        .await
        .expect("failed to execute process");
    Command::new("systemctl")

    use std::io::Write;
    std::io::stdout().write_all(&out.stdout).unwrap();
    std::io::stderr().write_all(&out.stderr).unwrap();
    let mut success = false;
    while !success {
        // Waiting for source to promote.
        let out = Command::new("pg_rewind")
            .args([
                "-D",
                &path,
                "-R",
                "--source-server",
                &format!(
                    "port={} user=postgres dbname={} host={}",
                    port, db_name, leader
                ),
            ])
            .output()
            .await
            .expect("failed to execute process");
        std::io::stdout().write_all(&out.stdout).unwrap();
        std::io::stderr().write_all(&out.stderr).unwrap();
        success = out.status.success();
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }
    let out = Command::new("systemctl")

    std::io::stdout().write_all(&out.stdout).unwrap();
    std::io::stderr().write_all(&out.stderr).unwrap();

Postrep uses a fixed replication strategy (streaming replication), a fixed leader election tool (Etcd 3), to achieve proper failover when the cluster leader fails.

Postrep uses a leader election tool (Etcd 3), to achieve proper failover when the cluster leader fails, in the context of PostgreSQL's streaming replication feature.

PostgreSQL has multiple ways to operate in a "leader-replicae" mode. However, failover isn't automatic, and requires another protocol to decide which of the replicae becomes the leader when the previous leader fails.

This crate solves this problem, in the particular configuration where one has an Etcd v3 server running, for example because of requirements from another server running on the same service.

Failover between the leader and replica mode of PostgreSQL isn't automatic, and requires another protocol to decide which of the replicae becomes the leader when the previous leader fails.