forks/sorcery

Mirror of https://git.t4t.associates/char/sorcery

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

Charlotte Somdaemon: write catalog pages to disk743f502

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