Diffusion reconciler: every 10 minutes the pin index is checked and mended 0e04dfc20211

-- The state each followed bookmark's pins were indexed at (oxygenhub#210):

-- the reconciler compares it with the bookmark as it is and re-indexes

-- where they differ, so an index write that failed is mended within

-- minutes rather than at the next restart.

CREATE TABLE diffusion_indexed (

    consumer_id INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    bookmark TEXT NOT NULL,

    heads TEXT NOT NULL,

    updated_ms INTEGER NOT NULL,

    PRIMARY KEY (consumer_id, bookmark)

);

    sqlx::query("INSERT OR REPLACE INTO diffusion_indexed (consumer_id, bookmark, heads, updated_ms) VALUES (?, ?, ?, ?)")

        .bind(row.id)

        .bind(bookmark)

        .bind(heads_key(heads))

        .bind(now)

        .execute(&mut *tx)

        .await

        .map_err(|e| format!("noting the state: {e}"))?;

            }

        }

    }

}

/// Heads as one comparable string: sorted, joined with `+`.

fn heads_key(heads: &[ChangeHash]) -> String {

    let mut h: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();

    h.sort();

    h.join("+")

}

/// How often [`reconciler`] looks.

pub const RECONCILE_EVERY: std::time::Duration = std::time::Duration::from_secs(10 * 60);

/// What one reconciling pass mended.

#[derive(Debug, Default, PartialEq, Eq)]

pub struct Repairs {

    /// Followed bookmarks whose index was behind the bookmark.

    pub reindexed: usize,

    /// Consumers behind an upstream's newest passing state with no open

    /// bump, bumped now.

    pub caught_up: usize,

}

/// Every [`RECONCILE_EVERY`], mend the pin index and the bumps it drives

/// (oxygenhub#210).

pub async fn reconciler(app: Shared) {

    loop {

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

        let r = reconcile(&app).await;

        if r != Repairs::default() {

            tracing::warn!("diffusion: reconciled: {} bookmarks re-indexed, {} consumers caught up", r.reindexed, r.caught_up);

        }

    }

}

/// One pass: re-index each followed bookmark whose indexed state is not

/// the state it is at now; then bump each consumer whose pin is not its

/// upstream's newest passing state and that has no open bump for it.

pub async fn reconcile(app: &Shared) -> Repairs {

    let mut repairs = Repairs::default();

    let Ok(rows) = db::repos::list(&app.db).await else { return repairs };

    for row in &rows {

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

        for bookmark in followed_bookmarks(&app.db, row).await {

            let b = bookmark.clone();

            let Ok(heads) = app.repos.read(&name, move |r, _| heads_of(r, &b)).await else { continue };

            let indexed: Option<String> = sqlx::query_scalar("SELECT heads FROM diffusion_indexed WHERE consumer_id = ? AND bookmark = ?")

                .bind(row.id)

                .bind(&bookmark)

                .fetch_optional(&app.db)

                .await

                .ok()

                .flatten();

            if indexed.as_deref() != Some(heads_key(&heads).as_str()) {

                tracing::info!("diffusion: {}/{} {bookmark}: the index was not at the bookmark; re-indexing", row.owner, row.name);

                record(app, row, &bookmark, &heads).await;

                repairs.reindexed += 1;

        }

    }

    // Consumers behind with no open bump.

    let pins: Vec<(i64, String, String, String)> =

        sqlx::query_as("SELECT consumer_id, path, upstream, heads FROM diffusion_pins WHERE upstream_repo_id IS NOT NULL")

            .fetch_all(&app.db)

            .await

            .unwrap_or_default();

    let mut caught = std::collections::HashSet::new();

    for (consumer_id, path, upstream, pinned) in pins {

        if caught.contains(&consumer_id) {

            continue;

        }

        let Some(up) = repo_by_oxy_id(app, &upstream).await else { continue };

        let Some(green) = last_green(app, &up).await else { continue };

        if heads_key(&green) == heads_key(&pinned.split('+').filter_map(ChangeHash::from_hex).collect::<Vec<_>>()) {

            continue;

        }

        let Ok(Some(consumer)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };

        let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { continue };

        let learned = target(&app.db, &consumer).await;

        let modules = oxymodules(app, &cname, &learned).await;

        if rule(&app.config.diffusion.follow, &consumer, modules.as_deref(), &path).follow == Follow::Off {

            continue;

        }

        // The newest bump at this path: open or a draft is a bump already,

        // and one a person closed is a "no" the reconciler respects (the

        // upstream's next move proposes again as usual).

        let bump_bm = bump_bookmark(&path);

        let newest = || async {

            db::proposals::list(&app.db, consumer_id, &[])

                .await

                .unwrap_or_default()

                .into_iter()

                .filter(|p| p.source_bookmark == bump_bm)

                .max_by_key(|p| p.number)

                .map(|p| p.state)

        };

        let has_bump = || async { matches!(newest().await.as_deref(), Some("open" | "draft")) };

        if matches!(newest().await.as_deref(), Some("open" | "draft" | "closed")) {

            continue;

        }

        caught.insert(consumer_id);

        catch_up(app, &consumer).await;

        if has_bump().await {

            tracing::info!(

                "diffusion: {}/{} {path} was behind {}/{} with no bump; bumped",

                consumer.owner,

                consumer.name,

                up.owner,

                up.name

            );

            repairs.caught_up += 1;

    repairs

    // Mend the pin index and the bumps it drives (oxygenhub#210).

    tokio::spawn(diffusion::reconciler(app.clone()));

    // The reconciler (#210): an index left behind its bookmark (as when

    // both writes failed) is mended from the bookmark itself, and a pass

    // with nothing to mend changes nothing.

    let pins_of = || async {

        sqlx::query_scalar::<_, String>("SELECT heads FROM diffusion_pins WHERE consumer_id = ?")

            .bind(game.id)

            .fetch_one(&h.app.db)

            .await

            .unwrap()

    };

    let right = pins_of().await;

    sqlx::query("UPDATE diffusion_pins SET heads = 'stale' WHERE consumer_id = ?").bind(game.id).execute(&h.app.db).await.unwrap();

    sqlx::query("DELETE FROM diffusion_indexed WHERE consumer_id = ?").bind(game.id).execute(&h.app.db).await.unwrap();

    let r = oxygenhub::diffusion::reconcile(&h.app).await;

    assert!(r.reindexed >= 1, "{r:?}");

    assert_eq!(pins_of().await, right, "the pin read again from the game's bookmark");

    assert_eq!(oxygenhub::diffusion::reconcile(&h.app).await.reindexed, 0, "nothing left to mend");

    // A bump someone closed is a "no": the reconciler does not open it again.

    sqlx::query("UPDATE proposals SET state = 'closed' WHERE id = ?").bind(caught.id).execute(&h.app.db).await.unwrap();

    let r = oxygenhub::diffusion::reconcile(&h.app).await;

    assert_eq!(r.caught_up, 0, "{r:?}");

    assert!(db::proposals::list(&h.app.db, g2_row.id, &["open", "draft"]).await.unwrap().is_empty(), "not reopened");

Keeping it right (oxygenhub#210). Each write records the state it indexed

(`diffusion_indexed`); a failed write is logged with the step that failed

and tried once more, and a state whose pins cannot be read leaves the

index as it was. Every ten minutes a reconciler compares each followed

bookmark with the state it was indexed at and re-indexes where they

differ; in the same pass, a consumer whose pin is not its upstream's

newest passing state, with no bump open or in draft, is caught up as if it

had just turned Diffusion on. A bump a person closed counts as a "no": it

is not opened again until the upstream moves. Repairs are logged

(`diffusion: reconciled: …`).