Diffusion index writes log every failure and try once more 5d54cf746e00

    // A failure is logged and tried once more after a moment (oxygenhub#210):

    // an index left behind shows a consumer behind with no bump.

    let repo = format!("{}/{}", row.owner, row.name);

    for attempt in 1..=2 {

        match try_record(app, row, bookmark, heads).await {

            Ok(()) => {

                if attempt > 1 {

                    tracing::info!("diffusion: indexed {repo} {bookmark} on the second try");

                }

                return;

            }

            Err(why) if attempt == 1 => {

                tracing::warn!("diffusion: indexing {repo} {bookmark} failed ({why}); trying again");

                tokio::time::sleep(RETRY_AFTER).await;

            }

            Err(why) => {

                FAILURES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                tracing::error!("diffusion: indexing {repo} {bookmark} failed again ({why}); its pins stay as they were");

            }

        }

    }

}

/// How long [`record`] waits before its second try.

const RETRY_AFTER: std::time::Duration = std::time::Duration::from_millis(250);

/// Indexing failures that the retry did not mend, since the hub started.

pub static FAILURES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Failures to force, per repository (tests only: [`fail_next_records`]).

static FORCED: std::sync::Mutex<Option<std::collections::HashMap<String, u32>>> = std::sync::Mutex::new(None);

/// Make the next `n` index writes for a repository fail, as a busy

/// database would (for tests of the retry).

#[doc(hidden)]

pub fn fail_next_records(app: &Shared, repo_id: i64, n: u32) {

    FORCED.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id), n);

}

/// Forced failures not yet used (tests only).

#[doc(hidden)]

pub fn forced_left(app: &Shared, repo_id: i64) -> u32 {

    FORCED.lock().unwrap_or_else(|p| p.into_inner()).as_ref().and_then(|m| m.get(&key(app, repo_id)).copied()).unwrap_or(0)

}

fn forced(app: &Shared, repo_id: i64) -> bool {

    let mut forced = FORCED.lock().unwrap_or_else(|p| p.into_inner());

    match forced.as_mut().and_then(|m| m.get_mut(&key(app, repo_id))) {

        Some(n) if *n > 0 => {

            *n -= 1;

            true

        }

        _ => false,

    }

}

/// One try at replacing a bookmark's indexed pins; which step failed and

/// why. A state that cannot be read leaves the index as it was rather than

/// emptying it.

async fn try_record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Result<(), String> {

    let Some(name) = RepoName::new(&row.owner, &row.name) else { return Ok(()) };

    let pins = if heads.is_empty() {

        Vec::new()

    } else {

        crate::flow::pins_in(app, name, heads.to_vec()).await.map_err(|e| format!("reading the pins: {e}"))?

    };

    if forced(app, row.id) {

        return Err("begin: forced for a test".into());

    }

    let mut tx = app.db.begin().await.map_err(|e| format!("begin: {e}"))?;

    sqlx::query("DELETE FROM diffusion_pins WHERE consumer_id = ? AND bookmark = ?")

        .map_err(|e| format!("delete: {e}"))?;

        sqlx::query(

        .map_err(|e| format!("insert {path}: {e}"))?;

    tx.commit().await.map_err(|e| format!("commit: {e}"))?;

    for u in upstreams.iter().flatten() {

        mark(app, row.id);

        mark(app, *u);

    Ok(())

) -> Result<Vec<(String, [u8; 16], Vec<oxy_core::hash::ChangeHash>)>, String> {

        .map_err(|e| e.to_string())

        for (path, id, pinned) in pins_in(app, name, heads).await.unwrap_or_default() {

    // the index follows, so it is no longer behind; even when the first

    // write fails, as a busy database's would (#210: it is tried again).

    let game = oxygenhub::db::repos::get(&h.app.db, "andy", "game").await.unwrap().unwrap();

    oxygenhub::diffusion::fail_next_records(&h.app, game.id, 1);

    assert_eq!(oxygenhub::diffusion::forced_left(&h.app, game.id), 0, "the first write did fail");

    // Up to 15 s: under a full parallel test run the bump can take a while.

    for _ in 0..300 {