fabrica

hanna/fabrica

main
Switch branch
mainView all branches
Clone
HTTPS
SSH
5360 bytes
Raw
1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! Repository mirror sync.
6//!
7//! [`sync_one`] runs a single mirror (push out or pull in) and records the
8//! outcome. [`run_scheduler`] is the background loop the server spawns; it wakes
9//! periodically and syncs every mirror whose interval has elapsed.
10//! [`trigger_push_mirrors`] is called from the post-receive hook so a received
11//! push propagates immediately to any "sync on push" mirrors.
12
13use std::sync::Arc;
14use std::time::Duration;
15
16use config::Config;
17use git::mirror::RemoteCred;
18use store::{Mirror, MirrorDirection, Store};
19
20/// How often the scheduler wakes to look for due mirrors.
21const TICK: Duration = Duration::from_secs(60);
22
23/// The push refspecs for a branch filter. Empty/absent mirrors all branches;
24/// otherwise each comma- or whitespace-separated pattern becomes a heads
25/// refspec. Tags are always included.
26fn push_refspecs(branch_filter: Option<&str>) -> Vec<String> {
27 let patterns: Vec<&str> = branch_filter
28 .unwrap_or("")
29 .split([',', ' ', '\t', '\n'])
30 .map(str::trim)
31 .filter(|p| !p.is_empty())
32 .collect();
33 if patterns.is_empty() {
34 return git::mirror::all_refs();
35 }
36 let mut specs: Vec<String> = patterns
37 .iter()
38 .map(|p| format!("+refs/heads/{p}:refs/heads/{p}"))
39 .collect();
40 specs.push("+refs/tags/*:refs/tags/*".to_string());
41 specs
42}
43
44/// Run one mirror's sync and record its outcome on the mirror row.
45///
46/// # Errors
47///
48/// Returns the git error text (also stored as the mirror's `last_error`), or a
49/// message if the repository or its on-disk path could not be resolved.
50pub async fn sync_one(store: &Store, config: &Config, mirror: &Mirror) -> Result<(), String> {
51 let repo = store
52 .repo_by_id(&mirror.repo_id)
53 .await
54 .map_err(|e| e.to_string())?
55 .ok_or_else(|| "repository no longer exists".to_string())?;
56 let repo_dir = git::repo_path(&config.storage.repo_dir, &repo.id).map_err(|e| e.to_string())?;
57 let home = config.storage.data_dir.join("tmp");
58 let cred = RemoteCred {
59 username: mirror.username.as_deref(),
60 secret: mirror.secret.as_deref(),
61 };
62
63 let result = match mirror.direction {
64 MirrorDirection::Pull => {
65 git::mirror::fetch(
66 &config.git.binary,
67 &repo_dir,
68 &mirror.remote_url,
69 cred,
70 &home,
71 )
72 .await
73 }
74 MirrorDirection::Push => {
75 let refspecs = push_refspecs(mirror.branch_filter.as_deref());
76 git::mirror::push(
77 &config.git.binary,
78 &repo_dir,
79 &mirror.remote_url,
80 cred,
81 &refspecs,
82 &home,
83 )
84 .await
85 }
86 };
87 if let Err(err) = &result {
88 tracing::warn!(mirror.id = %mirror.id, error = %err, "mirror sync failed");
89 }
90 let _ = store
91 .mirror_synced(&mirror.id, result.as_ref().err().map(String::as_str))
92 .await;
93 result
94}
95
96/// Sync every push mirror on `repo_id` that is configured to sync after a push.
97/// Best-effort: failures are recorded on the mirror, never propagated.
98pub async fn trigger_push_mirrors(store: &Store, config: &Config, repo_id: &str) {
99 let mirrors = match store.push_mirrors_on_push(repo_id).await {
100 Ok(m) => m,
101 Err(e) => {
102 tracing::warn!(error = %e, "could not load push mirrors");
103 return;
104 }
105 };
106 for mirror in &mirrors {
107 let _ = sync_one(store, config, mirror).await;
108 }
109}
110
111/// The background scheduler loop: every [`TICK`], sync each mirror whose interval
112/// has elapsed. Runs until the process ends.
113pub async fn run_scheduler(store: Store, config: Arc<Config>) {
114 let mut ticker = tokio::time::interval(TICK);
115 loop {
116 ticker.tick().await;
117 let now = now_ms();
118 let due = match store.due_mirrors(now).await {
119 Ok(d) => d,
120 Err(e) => {
121 tracing::warn!(error = %e, "could not query due mirrors");
122 continue;
123 }
124 };
125 for mirror in &due {
126 let _ = sync_one(&store, &config, mirror).await;
127 }
128 }
129}
130
131/// The current time in Unix milliseconds.
132fn now_ms() -> i64 {
133 use std::time::{SystemTime, UNIX_EPOCH};
134 SystemTime::now()
135 .duration_since(UNIX_EPOCH)
136 .map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
137}
138
139#[cfg(test)]
140mod tests {
141 use super::*;
142
143 #[test]
144 fn push_refspecs_default_is_all_refs() {
145 assert_eq!(push_refspecs(None), git::mirror::all_refs());
146 assert_eq!(push_refspecs(Some(" ")), git::mirror::all_refs());
147 }
148
149 #[test]
150 fn push_refspecs_honours_a_branch_filter() {
151 let specs = push_refspecs(Some("main, release/*"));
152 assert_eq!(
153 specs,
154 vec![
155 "+refs/heads/main:refs/heads/main".to_string(),
156 "+refs/heads/release/*:refs/heads/release/*".to_string(),
157 "+refs/tags/*:refs/tags/*".to_string(),
158 ]
159 );
160 }
161}