forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
9.1 KiB
1//! The mirror sync worker. It polls for due mirrors every minute, and wakes at once when
2//! `AppState::mirror_waker` is notified (a "sync now", a new mirror, or a push). At most two runs
3//! are in flight, and never two on the same repo.
4
5use std::collections::HashSet;
6use std::time::Duration;
7
8use forge_core::time::now_stamp;
9use time::OffsetDateTime;
10use time::format_description::well_known::Rfc3339;
11use tokio::sync::mpsc;
12
13use super::{policy, run_one};
14use crate::state::AppState;
15use crate::webhooks::client::WebhookPolicy;
16
17/// The longest the worker sleeps when nothing wakes it.
18const POLL: Duration = Duration::from_secs(60);
19/// The shortest sleep, so a due time that has already passed does not spin the loop.
20const MIN_SLEEP: Duration = Duration::from_secs(1);
21/// Runs in flight at once.
22const MAX_RUNNING: usize = 2;
23/// Due mirrors read per pass.
24const CLAIM_BATCH: u32 = 20;
25
26/// Starts the worker task. It runs until the process exits.
27pub fn spawn(state: AppState) {
28 tokio::spawn(run(state));
29}
30
31async fn run(state: AppState) {
32 let policy = policy(&state);
33 let (done_tx, mut done_rx) = mpsc::unbounded_channel::<(String, String)>();
34 let mut running_mirrors: HashSet<String> = HashSet::new();
35 let mut running_repos: HashSet<String> = HashSet::new();
36 loop {
37 claim(
38 &state,
39 &policy,
40 &done_tx,
41 &mut running_mirrors,
42 &mut running_repos,
43 )
44 .await;
45 // Wake when the earliest enabled mirror is due, so a push mirror queued with a short
46 // debounce runs on time instead of waiting for the next poll.
47 let next = match state.store.next_mirror_due().await {
48 Ok(next) => next,
49 Err(e) => {
50 tracing::error!(error = %e, "mirrors: could not read the next due time");
51 None
52 }
53 };
54 let nap = sleep_for(next.as_deref(), OffsetDateTime::now_utc());
55 tokio::select! {
56 _ = tokio::time::sleep(nap) => {}
57 _ = state.mirror_waker.notified() => {}
58 Some((mirror_id, repo_id)) = done_rx.recv() => {
59 running_mirrors.remove(&mirror_id);
60 running_repos.remove(&repo_id);
61 }
62 }
63 }
64}
65
66/// How long the worker sleeps before its next pass: until `next_due` (an RFC 3339 stamp, the
67/// earliest enabled mirror), clamped to `MIN_SLEEP..=POLL`. `None` or an unparsable stamp gives
68/// `POLL`.
69pub fn sleep_for(next_due: Option<&str>, now: OffsetDateTime) -> Duration {
70 let Some(due) = next_due.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok()) else {
71 return POLL;
72 };
73 let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO);
74 until.clamp(MIN_SLEEP, POLL)
75}
76
77/// Starts runs for due mirrors until the concurrency limit, skipping mirrors and repos already
78/// in flight.
79async fn claim(
80 state: &AppState,
81 policy: &WebhookPolicy,
82 done_tx: &mpsc::UnboundedSender<(String, String)>,
83 running_mirrors: &mut HashSet<String>,
84 running_repos: &mut HashSet<String>,
85) {
86 let due = match state.store.due_mirrors(&now_stamp(), CLAIM_BATCH).await {
87 Ok(due) => due,
88 Err(e) => {
89 tracing::error!(error = %e, "mirrors: could not list due mirrors");
90 return;
91 }
92 };
93 for mirror in due {
94 if running_mirrors.len() >= MAX_RUNNING {
95 break;
96 }
97 if running_mirrors.contains(&mirror.id) || running_repos.contains(&mirror.repo_id) {
98 continue;
99 }
100 running_mirrors.insert(mirror.id.clone());
101 running_repos.insert(mirror.repo_id.clone());
102 let (mirror_id, repo_id) = (mirror.id.clone(), mirror.repo_id.clone());
103 let state = state.clone();
104 let policy = policy.clone();
105 let done = done_tx.clone();
106 tokio::spawn(async move {
107 run_one(&state, &mirror, &policy).await;
108 let _ = done.send((mirror_id, repo_id));
109 });
110 }
111}
112
113#[cfg(test)]
114mod tests {
115 use std::path::Path;
116
117 use forge_core::secret;
118 use forge_core::webhooks::WebhookEvent;
119
120 use super::super::pull_refs;
121 use crate::test_support::test_state;
122
123 #[test]
124 fn sleep_stops_at_the_next_due_time_and_is_clamped() {
125 use std::time::Duration;
126
127 use time::OffsetDateTime;
128 use time::format_description::well_known::Rfc3339;
129 use time::macros::datetime;
130
131 let now = datetime!(2026-10-15 12:00:00 UTC);
132 let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp");
133 // Nothing enabled: the full poll interval.
134 assert_eq!(super::sleep_for(None, now), super::POLL);
135 // A push queued with a 10 s debounce: sleep until it is due, not the full poll.
136 let in_10s = stamp(now + time::Duration::seconds(10));
137 assert_eq!(
138 super::sleep_for(Some(&in_10s), now),
139 Duration::from_secs(10)
140 );
141 // Due later than the poll interval: wake at the poll interval.
142 let in_1h = stamp(now + time::Duration::hours(1));
143 assert_eq!(super::sleep_for(Some(&in_1h), now), super::POLL);
144 // Already due (or in flight and still due): do not spin, wait the minimum.
145 let past = stamp(now - time::Duration::seconds(5));
146 assert_eq!(super::sleep_for(Some(&past), now), super::MIN_SLEEP);
147 // A stamp that does not parse falls back to the poll interval.
148 assert_eq!(super::sleep_for(Some("soon"), now), super::POLL);
149 }
150
151 /// Plain git against a bare repo, for setting up the fake upstream.
152 fn plumbing(git_dir: &Path, args: &[&str]) -> String {
153 let out = std::process::Command::new("git")
154 .arg(format!("--git-dir={}", git_dir.display()))
155 .args(args)
156 .env("GIT_AUTHOR_NAME", "t")
157 .env("GIT_AUTHOR_EMAIL", "t@example.com")
158 .env("GIT_COMMITTER_NAME", "t")
159 .env("GIT_COMMITTER_EMAIL", "t@example.com")
160 .output()
161 .expect("git runs");
162 assert!(out.status.success(), "git {args:?}: {out:?}");
163 String::from_utf8_lossy(&out.stdout).trim().to_string()
164 }
165
166 /// Makes a new commit in `git_dir` with the given message and points `refname` at it.
167 fn commit_to(git_dir: &Path, refname: &str, message: &str) {
168 let tree = plumbing(git_dir, &["hash-object", "-t", "tree", "-w", "--stdin"]);
169 let commit = plumbing(git_dir, &["commit-tree", &tree, "-m", message]);
170 plumbing(git_dir, &["update-ref", refname, &commit]);
171 }
172
173 #[tokio::test]
174 async fn pull_sync_passes_only_changed_refs_to_post_update() {
175 let dir = tempfile::tempdir().expect("tempdir");
176 let state = test_state(dir.path()).await;
177 let owner = state
178 .store
179 .create_user("alice", None, "unused", false)
180 .await
181 .expect("user");
182 let repo = state
183 .store
184 .create_repo(
185 &owner.id,
186 "mirrored",
187 "",
188 forge_core::Visibility::Public,
189 "main",
190 )
191 .await
192 .expect("repo");
193 let bare = forge_git::repo::repo_path(&state.git.data_dir, &repo.id).expect("repo path");
194 forge_git::repo::init_bare(&state.git, &bare, "main")
195 .await
196 .expect("init bare");
197 // A webhook on push, so a changed ref shows up as a queued delivery.
198 let sealed = secret::seal(&[9u8; 32], b"hook-secret")
199 .expect("seal")
200 .into_bytes();
201 let hook = state
202 .store
203 .create_webhook(
204 &repo.id,
205 &owner.id,
206 "https://example.com/hook",
207 &sealed,
208 &[WebhookEvent::Push],
209 true,
210 )
211 .await
212 .expect("hook");
213
214 // First sync: the upstream has a branch and a tag.
215 let upstream = bare.clone();
216 let first = pull_refs(&state, &repo, async {
217 commit_to(&upstream, "refs/heads/main", "one");
218 commit_to(&upstream, "refs/tags/v1", "tagged");
219 Ok(())
220 })
221 .await
222 .expect("first sync");
223 assert_eq!(first, 2);
224 let deliveries = state
225 .store
226 .list_deliveries(&hook.id, 10)
227 .await
228 .expect("deliveries");
229 assert!(
230 deliveries.iter().any(|d| d.event == WebhookEvent::Push),
231 "a changed branch queues a push event"
232 );
233
234 // A sync with nothing new changes nothing.
235 let idle = pull_refs(&state, &repo, async { Ok(()) })
236 .await
237 .expect("idle sync");
238 assert_eq!(idle, 0);
239
240 // Moving the branch is detected as one change; the tag stays put.
241 let moved = pull_refs(&state, &repo, async {
242 commit_to(&upstream, "refs/heads/main", "two");
243 Ok(())
244 })
245 .await
246 .expect("moved sync");
247 assert_eq!(moved, 1);
248
249 // A failed fetch reports its error and changes nothing.
250 let failed = pull_refs(&state, &repo, async { Err("boom".to_string()) }).await;
251 assert_eq!(failed.unwrap_err(), "boom");
252 }
253}