hanna/fabrica

feat(mirror): background sync scheduler and post-receive trigger

b63d883 · hanna committed on 2026-07-26

Add the `mirror` crate: sync_one runs one mirror (fetch for pull, push
with a branch-filter refspec for push) and records the outcome;
run_scheduler is the 60s background loop over due mirrors; and
trigger_push_mirrors propagates a received push to sync-on-push mirrors.
The server spawns the scheduler alongside web/ssh, and the post-receive
hook calls the trigger.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Signed by hannaSSH key fingerprint: SHA256:4g9cWhkAAw8gwqhJUcVbSnGEvdGwklW+1Aa/fMUu59k

Cargo.lock +14 −0

Cargo.toml +2 −0

crates/cli/Cargo.toml +1 −0

crates/cli/src/hook_cmd.rs +3 −1

crates/mirror/Cargo.toml +20 −0

crates/mirror/src/lib.rs +161 −0

crates/web/Cargo.toml +1 −0

src/serve.rs +4 −0

  "config",
  "git",
  "mail",
+ "mirror",
  "model",
  "rpassword",
  "serde",
  "cli",
  "config",
  "git",
+ "mirror",
  "ssh",
  "store",
  "tokio",
+]
 [[package]]
+name = "mirror"
+version = "0.1.0"
+dependencies = [
+ "config",
+ "git",
+ "store",
+ "tokio",
+ "tracing",
+]
+[[package]]
 name = "ml-kem"
 version = "0.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
  "mail",
  "maud",
  "mime_guess",
+ "mirror",
  "model",
  "reqwest",
  "rust-embed",

 ssh = { path = "crates/ssh" }
 config = { path = "crates/config" }
 store = { path = "crates/store" }
+mirror = { path = "crates/mirror" }
 # Direct, optional dep only to expose git's `vendored` feature at the binary
 # (feature unification then applies it to the transitive `git` too).
 git = { path = "crates/git", optional = true }
 web = { path = "crates/web" }
 api = { path = "crates/api" }
 cli = { path = "crates/cli" }
+mirror = { path = "crates/mirror" }
 # External dependencies. Versions are centralised here; crates opt in with
 # `<dep> = { workspace = true }` so the whole workspace stays on one version.

 store = { workspace = true }
 auth = { workspace = true }
 git = { workspace = true }
+mirror = { workspace = true }
 mail = { workspace = true }
 model = { workspace = true }
 anyhow = { workspace = true }

     let size = dir_size(&git_dir);
     block_on(async move {
-        let (_config, store) = open_store(config_path.as_deref()).await?;
+        let (config, store) = open_store(config_path.as_deref()).await?;
         store.record_push(&repo_id, size).await?;
+        // Propagate the push to any "sync on push" mirrors (best-effort).
+        mirror::trigger_push_mirrors(&store, &config, &repo_id).await;
         anyhow::Ok(())
     })
+}

+[package]
+name = "mirror"
+description = "Repository mirror sync: periodic push/pull to and from remotes."
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+authors.workspace = true
+repository.workspace = true
+publish.workspace = true
+[dependencies]
+config = { workspace = true }
+store = { workspace = true }
+git = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+[lints]
+workspace = true

+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+//! Repository mirror sync.
+//!
+//! [`sync_one`] runs a single mirror (push out or pull in) and records the
+//! outcome. [`run_scheduler`] is the background loop the server spawns; it wakes
+//! periodically and syncs every mirror whose interval has elapsed.
+//! [`trigger_push_mirrors`] is called from the post-receive hook so a received
+//! push propagates immediately to any "sync on push" mirrors.
+use std::sync::Arc;
+use std::time::Duration;
+use config::Config;
+use git::mirror::RemoteCred;
+use store::{Mirror, MirrorDirection, Store};
+/// How often the scheduler wakes to look for due mirrors.
+const TICK: Duration = Duration::from_secs(60);
+/// The push refspecs for a branch filter. Empty/absent mirrors all branches;
+/// otherwise each comma- or whitespace-separated pattern becomes a heads
+/// refspec. Tags are always included.
+fn push_refspecs(branch_filter: Option<&str>) -> Vec<String> {
+    let patterns: Vec<&str> = branch_filter
+        .unwrap_or("")
+        .split([',', ' ', '\t', '\n'])
+        .map(str::trim)
+        .filter(|p| !p.is_empty())
+        .collect();
+    if patterns.is_empty() {
+        return git::mirror::all_refs();
+    }
+    let mut specs: Vec<String> = patterns
+        .iter()
+        .map(|p| format!("+refs/heads/{p}:refs/heads/{p}"))
+        .collect();
+    specs.push("+refs/tags/*:refs/tags/*".to_string());
+    specs
+}
+/// Run one mirror's sync and record its outcome on the mirror row.
+///
+/// # Errors
+///
+/// Returns the git error text (also stored as the mirror's `last_error`), or a
+/// message if the repository or its on-disk path could not be resolved.
+pub async fn sync_one(store: &Store, config: &Config, mirror: &Mirror) -> Result<(), String> {
+    let repo = store
+        .repo_by_id(&mirror.repo_id)
+        .await
+        .map_err(|e| e.to_string())?
+        .ok_or_else(|| "repository no longer exists".to_string())?;
+    let repo_dir = git::repo_path(&config.storage.repo_dir, &repo.id).map_err(|e| e.to_string())?;
+    let home = config.storage.data_dir.join("tmp");
+    let cred = RemoteCred {
+        username: mirror.username.as_deref(),
+        secret: mirror.secret.as_deref(),
+    };
+    let result = match mirror.direction {
+        MirrorDirection::Pull => {
+            git::mirror::fetch(
+                &config.git.binary,
+                &repo_dir,
+                &mirror.remote_url,
+                cred,
+                &home,
+            )
+            .await
+        }
+        MirrorDirection::Push => {
+            let refspecs = push_refspecs(mirror.branch_filter.as_deref());
+            git::mirror::push(
+                &config.git.binary,
+                &repo_dir,
+                &mirror.remote_url,
+                cred,
+                &refspecs,
+                &home,
+            )
+            .await
+        }
+    };
+    if let Err(err) = &result {
+        tracing::warn!(mirror.id = %mirror.id, error = %err, "mirror sync failed");
+    }
+    let _ = store
+        .mirror_synced(&mirror.id, result.as_ref().err().map(String::as_str))
+        .await;
+    result
+}
+/// Sync every push mirror on `repo_id` that is configured to sync after a push.
+/// Best-effort: failures are recorded on the mirror, never propagated.
+pub async fn trigger_push_mirrors(store: &Store, config: &Config, repo_id: &str) {
+    let mirrors = match store.push_mirrors_on_push(repo_id).await {
+        Ok(m) => m,
+        Err(e) => {
+            tracing::warn!(error = %e, "could not load push mirrors");
+            return;
+        }
+    };
+    for mirror in &mirrors {
+        let _ = sync_one(store, config, mirror).await;
+    }
+}
+/// The background scheduler loop: every [`TICK`], sync each mirror whose interval
+/// has elapsed. Runs until the process ends.
+pub async fn run_scheduler(store: Store, config: Arc<Config>) {
+    let mut ticker = tokio::time::interval(TICK);
+    loop {
+        ticker.tick().await;
+        let now = now_ms();
+        let due = match store.due_mirrors(now).await {
+            Ok(d) => d,
+            Err(e) => {
+                tracing::warn!(error = %e, "could not query due mirrors");
+                continue;
+            }
+        };
+        for mirror in &due {
+            let _ = sync_one(&store, &config, mirror).await;
+        }
+    }
+}
+/// The current time in Unix milliseconds.
+fn now_ms() -> i64 {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn push_refspecs_default_is_all_refs() {
+        assert_eq!(push_refspecs(None), git::mirror::all_refs());
+        assert_eq!(push_refspecs(Some("  ")), git::mirror::all_refs());
+    }
+    #[test]
+    fn push_refspecs_honours_a_branch_filter() {
+        let specs = push_refspecs(Some("main, release/*"));
+        assert_eq!(
+            specs,
+            vec![
+                "+refs/heads/main:refs/heads/main".to_string(),
+                "+refs/heads/release/*:refs/heads/release/*".to_string(),
+                "+refs/tags/*:refs/tags/*".to_string(),
+            ]
+        );
+    }
+}

+            }
         });
+        // The mirror scheduler syncs due push/pull mirrors in the background.
+        let mirror_task = tokio::spawn(mirror::run_scheduler(store.clone(), Arc::clone(&config)));
         let state = web::AppState::build(config, store, secret)?;
         // Load admin config overrides from the database before serving.
         state.reload_settings().await;
         let result = web::serve(state, addr).await;
         ssh_task.abort();
+        mirror_task.abort();
         result?;
         Ok(ExitCode::SUCCESS)
     })