Function holochain::test_utils::wait_for_integration_with_others

pub async fn wait_for_integration_with_others<Db: ReadAccess<DbKindDht>>(
    db: &Db,
    others: &[&Db],
    expected_count: usize,
    num_attempts: usize,
    delay: Duration,
    start: Option<Instant>
)

Same as wait for integration but can print other states at the same time

Examples found in repository

src/test_utils.rs (line 620)

async fn consistency_dbs_others<AuthorDb, DhtDb>(
    all_cell_dbs: &[(&AgentPubKey, &AuthorDb, &DhtDb)],
    num_attempts: usize,
    delay: Duration,
) where
    AuthorDb: ReadAccess<DbKindAuthored>,
    DhtDb: ReadAccess<DbKindDht>,
{
    let mut expected_count = 0;
    for (author, db) in all_cell_dbs.iter().map(|(author, a, _)| (author, a)) {
        let count = get_published_ops(*db, *author).len();
        expected_count += count;
    }
    let start = Some(std::time::Instant::now());
    for (i, &db) in all_cell_dbs.iter().map(|(_, _, d)| d).enumerate() {
        let mut others: Vec<_> = all_cell_dbs.iter().map(|(_, _, d)| *d).collect();
        others.remove(i);
        wait_for_integration_with_others(db, &others, expected_count, num_attempts, delay, start)
            .await
    }
}

fn get_published_ops<Db: ReadAccess<DbKindAuthored>>(
    db: &Db,
    author: &AgentPubKey,
) -> Vec<DhtOpLight> {
    fresh_reader_test(db.clone(), |txn| {
        txn.prepare(
            "
            SELECT
            DhtOp.type, Action.hash, Action.blob
            FROM DhtOp
            JOIN
            Action ON DhtOp.action_hash = Action.hash
            WHERE
            Action.author = :author
            AND (DhtOp.type != :store_entry OR Action.private_entry = 0)
        ",
        )
        .unwrap()
        .query_and_then(
            named_params! {
                ":store_entry": DhtOpType::StoreEntry,
                ":author": author,
            },
            |row| {
                let op_type: DhtOpType = row.get("type")?;
                let hash: ActionHash = row.get("hash")?;
                let action: SignedAction = from_blob(row.get("blob")?)?;
                Ok(DhtOpLight::from_type(op_type, hash, &action.0)?)
            },
        )
        .unwrap()
        .collect::<StateQueryResult<_>>()
        .unwrap()
    })
}

/// Same as wait_for_integration but with a default wait time of 10 seconds
#[tracing::instrument(skip(db))]
pub async fn wait_for_integration_1m<Db: ReadAccess<DbKindDht>>(db: &Db, expected_count: usize) {
    const NUM_ATTEMPTS: usize = 120;
    const DELAY_PER_ATTEMPT: std::time::Duration = std::time::Duration::from_millis(500);
    wait_for_integration(db, expected_count, NUM_ATTEMPTS, DELAY_PER_ATTEMPT).await
}

/// Exit early if the expected number of ops
/// have been integrated or wait for num_attempts * delay
#[tracing::instrument(skip(db))]
pub async fn wait_for_integration<Db: ReadAccess<DbKindDht>>(
    db: &Db,
    expected_count: usize,
    num_attempts: usize,
    delay: Duration,
) {
    for i in 0..num_attempts {
        let count = display_integration(db).await;
        if count >= expected_count {
            if count > expected_count {
                tracing::warn!("count > expected_count, meaning you may not be accounting for all nodes in this test.
                Consistency may not be complete.")
            }
            return;
        } else {
            let total_time_waited = delay * i as u32;
            tracing::debug!(?count, ?total_time_waited, counts = ?query_integration(db).await);
        }
        tokio::time::sleep(delay).await;
    }
}

/// Same as wait for integration but can print other states at the same time
pub async fn wait_for_integration_with_others_10s<Db: ReadAccess<DbKindDht>>(
    db: &Db,
    others: &[&Db],
    expected_count: usize,
    start: Option<std::time::Instant>,
) {
    const NUM_ATTEMPTS: usize = 100;
    const DELAY_PER_ATTEMPT: std::time::Duration = std::time::Duration::from_millis(100);
    wait_for_integration_with_others(
        db,
        others,
        expected_count,
        NUM_ATTEMPTS,
        DELAY_PER_ATTEMPT,
        start,
    )
    .await
}