forks/sorcery

git clone https://git.gae.moe/forks/sorcery

Charlotte Somperf: prioritize push-triggered builds over background scans2688dec

typst_support
branchesdarwin-flakegrace-patchesmaintypst_supportall refs →
13.4 KiB413 linesraw
1use std::collections::{HashMap, HashSet};
2use std::fs;
3use std::path::{Path, PathBuf};
4use std::sync::Arc;
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::time::Instant;
7
8use anyhow::{Context, Result};
9use axum::body::Bytes;
10use tokio::sync::{Mutex, watch};
11
12use crate::catalog::{self, Repository};
13use crate::generate::{self, Config as GenerateConfig, STATE_FILE};
14use crate::render::{catalog_index, user_index};
15
16use super::{AppState, Inner};
17
18#[derive(Default)]
19pub(super) struct Snapshot {
20    pub(super) catalog_html: Bytes,
21    pub(super) user_html: HashMap<String, Bytes>,
22    pub(super) repos: HashMap<String, RepoEntry>,
23}
24
25#[derive(Clone)]
26pub(super) struct RepoEntry {
27    pub(super) repository: Repository,
28    build_gate: Arc<Mutex<()>>,
29}
30
31#[derive(Clone, Copy, Default)]
32struct BuildQueueState {
33    running: bool,
34    priority_waiting: usize,
35}
36
37/// Serializes site builds: each already fans out over every core with
38/// rayon, so concurrent builds only thrash. Human-triggered builds claim
39/// the slot ahead of background rebuilds, and cancel the background build
40/// currently holding it; the scanner re-queues whatever it abandons.
41pub(super) struct BuildQueue {
42    state: watch::Sender<BuildQueueState>,
43    background: std::sync::Mutex<Option<Arc<AtomicBool>>>,
44}
45
46pub(super) struct BuildPermit<'a> {
47    queue: &'a BuildQueue,
48}
49
50impl BuildQueue {
51    pub(super) fn new() -> Self {
52        Self {
53            state: watch::channel(BuildQueueState::default()).0,
54            background: std::sync::Mutex::new(None),
55        }
56    }
57
58    fn register_background(&self) -> Arc<AtomicBool> {
59        let flag = Arc::new(AtomicBool::new(false));
60        *self.background.lock().unwrap_or_else(|e| e.into_inner()) = Some(flag.clone());
61        flag
62    }
63
64    fn clear_background(&self, flag: &Arc<AtomicBool>) {
65        let mut background = self.background.lock().unwrap_or_else(|e| e.into_inner());
66        if background.as_ref().is_some_and(|active| Arc::ptr_eq(active, flag)) {
67            *background = None;
68        }
69    }
70
71    async fn acquire(&self, priority: bool) -> BuildPermit<'_> {
72        if priority {
73            self.state
74                .send_modify(|state| state.priority_waiting += 1);
75            let background = self.background.lock().unwrap_or_else(|e| e.into_inner());
76            if let Some(flag) = background.as_ref() {
77                flag.store(true, Ordering::Relaxed);
78            }
79        }
80        let mut seen = self.state.subscribe();
81        loop {
82            let mut claimed = false;
83            self.state.send_modify(|state| {
84                if !state.running && (priority || state.priority_waiting == 0) {
85                    state.running = true;
86                    if priority {
87                        state.priority_waiting -= 1;
88                    }
89                    claimed = true;
90                }
91            });
92            if claimed {
93                return BuildPermit { queue: self };
94            }
95            seen.borrow_and_update();
96            let _ = seen.changed().await;
97        }
98    }
99}
100
101impl Drop for BuildPermit<'_> {
102    fn drop(&mut self) {
103        self.queue
104            .state
105            .send_modify(|state| state.running = false);
106    }
107}
108
109pub(super) async fn scan_periodically(state: AppState) {
110    let mut tick = tokio::time::interval(state.config.check_interval);
111    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
112    tick.tick().await; // the first tick is immediate; run() already scanned
113    loop {
114        tokio::select! {
115            _ = tick.tick() => {},
116            _ = state.rescan_requested.notified() => {},
117        }
118        if let Err(error) = state.rescan().await {
119            eprintln!("sorceryd: rescan: {error}");
120        }
121    }
122}
123
124impl Inner {
125    /// Discover repositories and publish a fresh immutable catalog snapshot.
126    pub(super) async fn rescan(&self) -> Result<()> {
127        let _gate = self.rescan_gate.lock().await;
128        let config = self.config.clone();
129        let (repositories, freshness) = tokio::task::spawn_blocking(move || -> Result<_> {
130            let repositories = catalog::discover(&config.repositories)?;
131            let freshness = repositories
132                .iter()
133                .map(|repository| is_current(&generate_config(&config, repository)))
134                .collect::<Vec<_>>();
135            Ok((repositories, freshness))
136        })
137        .await??;
138
139        let old = self.snapshot.load_full();
140        let mut repos = HashMap::with_capacity(repositories.len());
141        let mut stale = Vec::new();
142        for (repository, current) in repositories.iter().zip(freshness) {
143            let current = match current {
144                Ok(current) => current,
145                Err(error) => {
146                    eprintln!(
147                        "sorceryd: skipping {}/{}: {error}",
148                        repository.user, repository.name
149                    );
150                    continue;
151                }
152            };
153            let key = format!("{}/{}", repository.user, repository.name);
154            let build_gate = old
155                .repos
156                .get(&key)
157                .map(|entry| entry.build_gate.clone())
158                .unwrap_or_else(|| Arc::new(Mutex::new(())));
159            let entry = RepoEntry {
160                repository: repository.clone(),
161                build_gate,
162            };
163            if !current {
164                stale.push(entry.clone());
165            }
166            repos.insert(key, entry);
167        }
168
169        let instance_name = &self.config.instance_name;
170        let user_html = repositories
171            .chunk_by(|a, b| a.user == b.user)
172            .map(|repos| {
173                let user = &repos[0].user;
174                (user.clone(), user_index(instance_name, user, repos).into())
175            })
176            .collect();
177        self.snapshot.store(Arc::new(Snapshot {
178            catalog_html: catalog_index(instance_name, &repositories).into(),
179            user_html,
180            repos,
181        }));
182        self.background.send_replace(stale);
183        Ok(())
184    }
185
186    pub(super) async fn build(&self, entry: &RepoEntry, priority: bool) -> Result<bool> {
187        let background = (!priority).then(|| self.build_queue.register_background());
188        let permit = self.build_queue.acquire(priority).await;
189        let _gate = entry.build_gate.lock().await;
190        let _cache = self.cache_lock.read().await;
191        let config = generate_config(&self.config, &entry.repository);
192        let highlights = self.highlights.clone();
193        let cancel = background.clone();
194        let result =
195            tokio::task::spawn_blocking(move || ensure_current(&config, highlights, cancel)).await;
196        if let Some(flag) = &background {
197            self.build_queue.clear_background(flag);
198        }
199        drop(permit);
200        result?
201    }
202}
203
204pub(super) async fn background_worker(state: AppState, mut stale: watch::Receiver<Vec<RepoEntry>>) {
205    while stale.changed().await.is_ok() {
206        let entries = stale.borrow_and_update().clone();
207        for entry in entries {
208            let started = Instant::now();
209            let name = format!("{}/{}", entry.repository.user, entry.repository.name);
210            match state.build(&entry, false).await {
211                Ok(true) => eprintln!(
212                    "sorceryd: rebuilt {name} in {:.2}s",
213                    started.elapsed().as_secs_f64(),
214                ),
215                Ok(false) => {}
216                Err(error) => eprintln!("sorceryd: background rebuild of {name}: {error}"),
217            }
218        }
219        match garbage_collect(&state).await {
220            Ok(None | Some(0)) => {}
221            Ok(Some(count)) => {
222                eprintln!("sorceryd: removed {count} stale cache directories")
223            }
224            Err(error) => eprintln!("sorceryd: cache garbage collection: {error}"),
225        }
226    }
227}
228
229async fn garbage_collect(state: &Inner) -> Result<Option<usize>> {
230    let Ok(_cache) = state.cache_lock.try_write() else {
231        return Ok(None);
232    };
233    let cache = state.config.cache.clone();
234    let live = state
235        .snapshot
236        .load()
237        .repos
238        .values()
239        .map(|entry| state.config.repo_cache(&entry.repository))
240        .collect::<HashSet<_>>();
241    tokio::task::spawn_blocking(move || garbage_collect_cache(&cache, &live))
242        .await?
243        .map(Some)
244}
245
246fn garbage_collect_cache(cache: &Path, live: &HashSet<PathBuf>) -> Result<usize> {
247    let mut removed = 0;
248    for user in fs::read_dir(cache)? {
249        let user = user?;
250        if !user.file_type()?.is_dir() {
251            continue;
252        }
253        for repo in fs::read_dir(user.path())? {
254            let repo = repo?;
255            if !repo.file_type()?.is_dir() {
256                continue;
257            }
258            let path = repo.path();
259            if path.join(STATE_FILE).is_file() && !live.contains(&path) {
260                fs::remove_dir_all(path)?;
261                removed += 1;
262            }
263        }
264    }
265    Ok(removed)
266}
267
268fn generate_config(config: &super::Config, repository: &Repository) -> GenerateConfig {
269    GenerateConfig {
270        repo: repository.path.clone(),
271        out: config.repo_cache(repository),
272        instance_name: config.instance_name.clone(),
273        name: Some(format!("{}/{}", repository.user, repository.name)),
274        clone_url: config.site_url(repository),
275    }
276}
277
278fn is_current(config: &GenerateConfig) -> Result<bool> {
279    let expected = generate::expected_state(config)?;
280    Ok(fs::read_to_string(config.out.join(STATE_FILE))
281        .ok()
282        .as_deref()
283        == Some(&expected))
284}
285
286/// Regenerate the cache directory if its recorded state differs from the
287/// repository's, building into a staging sibling and swapping atomically.
288fn ensure_current(
289    config: &GenerateConfig,
290    highlights: Arc<crate::highlight::Cache>,
291    cancel: Option<Arc<AtomicBool>>,
292) -> Result<bool> {
293    if is_current(config)? {
294        return Ok(false);
295    }
296
297    let parent = config
298        .out
299        .parent()
300        .context("cache repository has no parent")?;
301    let name = config
302        .out
303        .file_name()
304        .context("cache repository has no name")?;
305    fs::create_dir_all(parent)?;
306    let staging = parent.join(format!(
307        ".{}.new-{}",
308        name.to_string_lossy(),
309        std::process::id()
310    ));
311    if staging.exists() {
312        fs::remove_dir_all(&staging)?;
313    }
314    let mut staging_config = config.clone();
315    staging_config.out = staging.clone();
316    if let Err(error) = generate::full_cancellable(
317        &staging_config,
318        Some(&config.out),
319        highlights,
320        cancel.clone(),
321    ) {
322        // A human-triggered build wanted the machine; the scanner will
323        // re-queue this repository for a later background attempt.
324        if cancel
325            .as_ref()
326            .is_some_and(|flag| flag.load(Ordering::Relaxed))
327        {
328            return Ok(false);
329        }
330        return Err(error);
331    }
332    // keep info/refs + objects/info/packs current for static git-dir access
333    let status = std::process::Command::new("git")
334        .arg("-C")
335        .arg(&config.repo)
336        .arg("update-server-info")
337        .status()?;
338    if !status.success() {
339        eprintln!(
340            "sorceryd: update-server-info failed for {}",
341            config.repo.display()
342        );
343    }
344    generate::swap_dir(&staging, &config.out)?;
345    Ok(true)
346}
347
348#[cfg(test)]
349mod tests {
350    use std::collections::HashSet;
351    use std::fs;
352    use std::sync::Arc;
353    use std::sync::atomic::Ordering;
354
355    use anyhow::Result;
356
357    use crate::generate::STATE_FILE;
358    use crate::testutil::TempDir;
359
360    use super::{BuildQueue, garbage_collect_cache};
361
362    #[tokio::test]
363    async fn priority_builds_claim_the_slot_ahead_of_background_builds() {
364        let queue = Arc::new(BuildQueue::new());
365        let first = queue.acquire(false).await;
366        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
367        for (priority, label) in [(false, "background"), (true, "priority")] {
368            let queue = queue.clone();
369            let tx = tx.clone();
370            tokio::spawn(async move {
371                let _permit = queue.acquire(priority).await;
372                tx.send(label).unwrap();
373            });
374            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
375        }
376        drop(first);
377        assert_eq!(rx.recv().await, Some("priority"));
378        assert_eq!(rx.recv().await, Some("background"));
379    }
380
381    #[tokio::test]
382    async fn priority_builds_cancel_the_active_background_build() {
383        let queue = BuildQueue::new();
384        let flag = queue.register_background();
385        let _permit = queue.acquire(true).await;
386        assert!(flag.load(Ordering::Relaxed));
387        queue.clear_background(&flag);
388    }
389
390    #[test]
391    fn garbage_collects_only_inactive_site_caches() -> Result<()> {
392        let root = TempDir::new("cache-gc");
393        let live = root.join("alice/live");
394        let stale = root.join("alice/stale");
395        let staging = root.join("alice/.live.new-1");
396        let asset = root.join("fonts/inter");
397        for path in [&live, &stale, &staging] {
398            fs::create_dir_all(path)?;
399            fs::write(path.join(STATE_FILE), "state")?;
400        }
401        fs::create_dir_all(&asset)?;
402
403        assert_eq!(
404            garbage_collect_cache(&root, &HashSet::from([live.clone()]))?,
405            2
406        );
407        assert!(live.is_dir());
408        assert!(!stale.exists());
409        assert!(!staging.exists());
410        assert!(asset.is_dir());
411        Ok(())
412    }
413}