The hub garbage-collects every repository nightly 8a3334b7a364

//! Housekeeping that runs on its own (hub spec §17.2): the nightly

//! garbage collection of every repository, one at a time under its

//! write lock, so a push never races a repack. `oxygenhub gc` runs the

//! same pass by hand.

use crate::db;

use crate::repos::RepoName;

use crate::Shared;

/// What one repository's gc did, for the log and `oxygenhub gc`.

#[derive(Clone, Debug)]

pub struct Done {

    pub repo: String,

    pub failed: bool,

    pub summary: String,

}

impl std::fmt::Display for Done {

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        write!(f, "{}: {}", self.repo, self.summary)

    }

}

/// Garbage-collect every repository, in turn. A repository that fails is

/// reported and the rest still run.

pub async fn gc_all(app: &Shared) -> Vec<Done> {

    let rows = match db::repos::list(&app.db).await {

        Ok(rows) => rows,

        Err(e) => {

            tracing::warn!("gc: listing repositories: {e}");

            return Vec::new();

        }

    };

    let mut done = Vec::with_capacity(rows.len());

    for row in rows {

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

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

        if !app.repos.exists(&name) {

            continue;

        }

        let label = slug.clone();

        let started = std::time::Instant::now();

        let result =

            app.repos.write(&name, move |repo, _| Ok(repo.gc_with(false, &mut |phase| tracing::debug!("gc {label}: {phase}"))?)).await;

        let d = match result {

            Ok(r) => Done {

                summary: format!(

                    "{} reachable; pruned {} ({} bytes){}; packed {} into one pack in {:.1} s",

                    r.reachable,

                    r.pruned.objects,

                    r.pruned.bytes,

                    if r.pruned.deferred > 0 { format!(", {} deferred by the grace period", r.pruned.deferred) } else { String::new() },

                    r.packed.objects,

                    started.elapsed().as_secs_f64()

                ),

                repo: slug,

                failed: false,

            },

            Err(e) => Done { summary: format!("failed: {e}"), repo: slug, failed: true },

        };

        if d.failed {

            tracing::warn!("gc {}", d);

        } else {

            tracing::info!("gc {}", d);

        }

        done.push(d);

    }

    done

}

/// Run [`gc_all`] once a day at `retention.gc_hour` UTC.

pub async fn nightly(app: Shared) {

    let hour = app.config.retention.gc_hour;

    if hour >= 24 {

        tracing::info!("gc: the nightly run is off (retention.gc_hour = {hour})");

        return;

    }

    let mut last_day = None;

    loop {

        tokio::time::sleep(std::time::Duration::from_secs(60)).await;

        let now = time::OffsetDateTime::now_utc();

        if u32::from(now.hour()) != hour {

            continue;

        }

        let day = now.date();

        if last_day == Some(day) {

            continue;

        }

        last_day = Some(day);

        let done = gc_all(&app).await;

        tracing::info!("gc: nightly run over {} repositories, {} failed", done.len(), done.iter().filter(|d| d.failed).count());

    }

}

//! The nightly garbage collection (hub spec §17.2), run by hand.

use oxy_core::repo::Repo;

use oxygenhub::config::Config;

use oxygenhub::db;

use oxygenhub::repos::RepoName;

use std::path::Path;

fn hermetic() {

    static ONCE: std::sync::Once = std::sync::Once::new();

    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();

    ONCE.call_once(|| {

        let d = tempfile::tempdir().unwrap();

        std::env::set_var("OXY_HOME", d.path());

        DIR.set(d).ok();

    });

}

fn write(root: &Path, rel: &str, content: &[u8]) {

    let p = root.join(rel);

    std::fs::create_dir_all(p.parent().unwrap()).unwrap();

    std::fs::write(p, content).unwrap();

}

#[tokio::test(flavor = "multi_thread")]

async fn every_repository_is_garbage_collected_in_turn() {

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let mut config = Config::load(dir.path()).unwrap();

    config.auth.push_tokens = vec!["site".into()];

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let port = listener.local_addr().unwrap().port();

    config.server.base_url = format!("http://127.0.0.1:{port}");

    assert_eq!(config.retention.gc_hour, 3, "nightly at 03:00 UTC by default");

    let app = oxygenhub::build(config).await.unwrap();

    let router = oxygenhub::router(app.clone());

    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });

    db::accounts::create_account(&app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    for name in ["one", "two"] {

        app.repos.create(&RepoName::parse(&format!("andy/{name}")).unwrap(), None).unwrap();

        db::repos::create(&app.db, "andy", name, "", hub_api::Visibility::Public).await.unwrap();

    }

    // A gc folds what the pushes stored into one pack.

    let work = tempfile::tempdir().unwrap();

    let mut c = Repo::init(work.path()).unwrap();

    c.config.set("remote.origin.url", &format!("http://127.0.0.1:{port}/andy/one")).unwrap();

    c.config.set("remote.origin.token", "site").unwrap();

    c.save_config().unwrap();

    for i in 0..3 {

        write(work.path(), &format!("f{i}.txt"), format!("{i}\n").as_bytes());

        c.commit(&format!("c{i}"), &[]).unwrap();

    }

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let hosted = app.repos.dir(&RepoName::parse("andy/one").unwrap());

    let before = Repo::open(&hosted).unwrap();

    assert!(before.store().loose_count().unwrap() > 0, "the push left loose objects");

    drop(before);

    let done = oxygenhub::maintenance::gc_all(&app).await;

    assert_eq!(done.len(), 2, "{done:?}");

    assert!(done.iter().all(|d| !d.failed), "{done:?}");

    assert!(done[0].summary.contains("packed"), "{}", done[0]);

    let after = Repo::open(&hosted).unwrap();

    assert_eq!(after.store().loose_count().unwrap(), 0, "everything is in a pack");

    assert_eq!(after.store().pack_count(), 1);

    assert_eq!(after.bookmark_heads().unwrap(), c.bookmark_heads().unwrap(), "and the history is intact");

    // The clone after the gc sees the same repository.

    let again = tempfile::tempdir().unwrap();

    let d = Repo::clone(&format!("http://127.0.0.1:{port}/andy/one"), again.path()).unwrap();

    assert_eq!(std::fs::read(again.path().join("f2.txt")).unwrap(), b"2\n");

    let _ = d;

}

    /// The UTC hour at which every repository is garbage collected

    /// (spec §17.2); 24 or more turns the nightly gc off.

    pub gc_hour: u32,

        Retention { trash_days: 30, webhook_log_days: 30, gc_hour: 3 }

pub mod maintenance;

    tokio::spawn(maintenance::nightly(app.clone()));

    /// Garbage-collect every repository now, as the nightly job does

    Gc,

        Cmd::Gc => {

            let config = Config::load(&cli.data)?;

            let app = oxygenhub::build(config).await?;

            let done = oxygenhub::maintenance::gc_all(&app).await;

            for d in &done {

                println!("{d}");

            }

            if done.iter().any(|d| d.failed) {

                anyhow::bail!("a repository could not be garbage collected");

            }

            Ok(())

        }

    let opts = RemoteOptions { token: Some(token.clone()), ca: None, repo_id: None, cache_dir: None };

    let bad = RemoteOptions { token: Some("flowjob_00".into()), ca: None, repo_id: None, cache_dir: None };

SQLite in WAL mode; migrations embedded and applied at start; `PRAGMA user_version` guards downgrades. Repositories are ordinary Oxygen repositories; `oxy gc` runs per repository as a nightly job, one repository at a time under its write lock, at `retention.gc_hour` UTC (default 3; 24 or more turns it off), and `oxygenhub gc` runs the same pass by hand. Each repository's run is logged with what it pruned and packed.