pmeunier/postrep - Change 6Z6PYXTSU3DYKZKXWSZ5JZXAO2MO4KJADTYNDLUYMD4LROXRREZAC

Testing and debugging

Created by pmeunier on January 7, 2022

6Z6PYXTSU3DYKZKXWSZ5JZXAO2MO4KJADTYNDLUYMD4LROXRREZAC

Dependencies

In channels

main

Change contents

Deletion in src/main.rs at line 43 [3.17]

B:BD[3.1147] → [3.1147:1483]

    let (db, connection) = tokio_postgres::connect("host=localhost user=postgres", NoTls)
        .await
        .unwrap();
    let db = Arc::new(Mutex::new(db));
    let dbt = tokio::spawn(async move {
        if let Err(e) = connection.await {
            eprintln!("connection error: {}", e);
        }
    });
    tokio::pin!(dbt);

Insertion in src/main.rs at line 49 [3.17]

[3.1757]

    let db = Arc::new(Mutex::new(None));
    let db_ = db.clone();
    let lead_ = lead.clone();
    tokio::spawn(async move {
        let mut is_normal = false;
        loop {
            debug!("reconnect");
            if let Ok((db, connection)) =
                tokio_postgres::connect("host=localhost user=postgres", NoTls).await
            {
                if is_normal {
                    is_normal = false;
                }
                *db_.lock().await = Some(db);
                let e = connection.await;
                eprintln!("connection ended: {:?}", e);
                // Ignore "normal" restarts (i.e. promotes).
                let db = db_.lock().await.take();
                if db.is_some() {
                    lead_.send(false).unwrap();
                    *db_.lock().await = None;
                } else {
                    debug!("normal restart, reconnecting");
                    is_normal = true;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
    });

Replacement in src/main.rs at line 87 [3.17]

B:BD[3.1943] → [3.1943:2140]

    let mut race = tokio::spawn(race(
        host.clone(),
        signal.clone(),
        lead.clone(),
        is_leader.clone(),
    ));
    let addr = SocketAddr::from((
        [0, 0, 0, 0],

[3.1943]

[3.2140]

    let mut race = tokio::spawn(race(host.clone(), signal.clone(), is_leader.clone()));
    // let addr = SocketAddr::from((
    //     [0, 0, 0, 0],
    //     matches
    //         .value_of("port")
    //         .and_then(|x| x.parse().ok())
    //         .unwrap_or(8008),
    // ));
    let addr6 = SocketAddr::from((
        [0, 0, 0, 0, 0, 0, 0, 0],

Insertion in src/main.rs at line 106 [3.17]
[3.2469]
[3.2469]

Replacement in src/main.rs at line 109 [3.17]

B:BD[3.2504] → [3.2504:2573]

            _ = Server::bind(&addr).serve(make_service.clone()) => {

[3.2504]

[2.90]

            _ = Server::bind(&addr6).serve(make_service.clone()) => {

Replacement in src/main.rs at line 113 [3.17]

B:BD[3.2658] → [3.2658:3527]

B:BD[3.3527] → [2.139:179]

            _ = &mut dbt => {
                // If this database breaks, release the lock.
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                info!("Restarting database connection");
                let (db_, connection) = if let Ok(x) = tokio_postgres::connect("host=localhost user=postgres", NoTls).await {
                    x
                } else {
                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                    continue
                };
                lead.send(false).unwrap();
                dbt.set(tokio::spawn(async move {
                    if let Err(e) = connection.await {
                        eprintln!("connection error: {}", e);
                    }
                }));
                *db.lock().await = db_
            }
            _ = &mut race => {
                error!("race stopped");

[3.2658]

[3.3567]

            e = &mut race => {
                error!("race stopped: {:?}", e);

Replacement in src/main.rs at line 117 [3.17]

B:BD[3.3603] → [3.3603:3647]

B:BD[3.3647] → [2.180:233]

            _ = &mut election_observer => {
                error!("election observer stopped");

[3.3603]

[3.3700]

            e = &mut election_observer => {
                error!("election observer stopped: {:?}", e);

Deletion in src/main.rs at line 141 [3.17]
B:BD[3.4147] → [3.4147:4176]
```
    lead: Arc<Sender<bool>>,
```

Insertion in src/main.rs at line 143 [3.17]

[3.4244]

    let mut client = Client::connect(["localhost:2379"], None).await.unwrap();
    let lease = client.lease_grant(LEASE_DURATION, None).await?;
    let id = lease.id();
    debug!("id = {:?}", id);
    let (mut keeper, _stream) = client.lease_keep_alive(id).await?;
    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_secs(LEASE_DURATION as u64 / 2)).await;
        loop {
            trace!("sending keep alive");
            if let Err(e) = keeper.keep_alive().await {
                error!("{:?}", e);
                break;
            }
            tokio::time::sleep(std::time::Duration::from_secs(LEASE_DURATION as u64 / 2)).await;
        }
        debug!("done renewing");
    });

Deletion in src/main.rs at line 163 [3.17]

B:BD[3.4255] → [3.4255:4436]

        let mut client = Client::connect(["localhost:2379"], None).await.unwrap();
        let lease = client.lease_grant(LEASE_DURATION, None).await?;
        let id = lease.id();

Deletion in src/main.rs at line 167 [3.17]

B:BD[3.4591] → [3.4591:5242]

        let (mut keeper, _stream) = client.lease_keep_alive(id).await?;
        let is_leader_ = is_leader.clone();
        let t = tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_secs(LEASE_DURATION as u64 / 2)).await;
            while *is_leader_.borrow() {
                debug!("sending keep alive");
                if let Err(e) = keeper.keep_alive().await {
                    error!("{:?}", e);
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_secs(LEASE_DURATION as u64 / 2)).await;
            }
            debug!("done renewing");
        });

Deletion in src/main.rs at line 174 [3.17]
B:BD[3.5391] → [3.5391:5425]
```
        lead.send(true).unwrap();
```

Deletion in src/main.rs at line 180 [3.17]

B:BD[3.5684] → [3.5684:5772]

                // at.store(false, Ordering::SeqCst);
                t.await.unwrap();

Replacement in src/main.rs at line 194 [3.17]

B:BD[3.6086] → [3.6086:6132]

    pool: Arc<Mutex<tokio_postgres::Client>>,

[3.6086]

[3.6132]

    pool: Arc<Mutex<Option<tokio_postgres::Client>>>,

Insertion in src/main.rs at line 201 [3.17]
[3.6418]
[3.6418]
```
        debug!("leader = {:?}", leader);
```
Deletion in src/main.rs at line 208 [3.17]
B:BD[3.6643] → [3.6643:6700]
```
            std::fs::remove_file(&signal).unwrap_or(());
```

Replacement in src/main.rs at line 210 [3.17]

B:BD[3.6748] → [3.6748:6942]

        let pool = pool.lock().await;
        pool.execute(format!("ALTER SYSTEM SET primary_conninfo='host={} port=5432 user=replication password={}'", leader, password).as_str(), &[]).await?;

[3.6748]

[3.6942]

        let pool_ = pool.lock().await.take();
        if let Some(pool_) = pool_ {
            pool_.execute(format!("ALTER SYSTEM SET primary_conninfo='host={} port=5432 user=replication password={}'", leader, password).as_str(), &[]).await?;
            if leader == host {
                info!("promoting");
                pool_.execute("SELECT pg_promote()", &[]).await.unwrap_or(0);
                // Don't replace pool_ into *pool, since the server might restart.
                *pool.lock().await = Some(pool_)
            } else {
                *pool.lock().await = Some(pool_)
            }
        } else {
            lead.send(false)?;
        }