forge
ExploreLog in

Tahinli/forge

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