Keep bookmark briefs across restarts, and log slow requests 39442b000e6b

    web::router(app).layer(axum::middleware::from_fn(slow_requests))

/// Requests slower than this are logged, with their path (#145).

const SLOW: std::time::Duration = std::time::Duration::from_secs(2);

/// Log a slow request's method, path and time, so a slow page can be

/// found afterwards. Never the query, which may hold a token.

async fn slow_requests(req: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response {

    let (method, path) = (req.method().clone(), req.uri().path().to_string());

    let start = std::time::Instant::now();

    let res = next.run(req).await;

    let took = start.elapsed();

    if took >= SLOW {

        tracing::warn!("slow request: {method} {path} took {:.1} s ({})", took.as_secs_f64(), res.status().as_u16());

    }

    res

}

                // The moved bookmarks' briefs, worked out now so no page

                // waits for them (#145).

                if let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) {

                    let all: Vec<Vec<oxy_core::hash::ChangeHash>> = moved.iter().map(|(_, _, h)| h.clone()).collect();

                    let warm = app.clone();

                    tokio::spawn(async move {

                        warm.repos.read(&name, move |r, cache| Ok(cache.briefs(r, &all).map(|_| ())?)).await.ok();

                    });

                }

use std::path::{Path, PathBuf};

impl Summary {

    pub fn brief(&self) -> Brief {

        Brief { changes: self.changes.len(), latest_ms: self.latest_ms, conflicted: self.conflicted }

    }

}

/// What a bookmark list and the overview show of a state (#145): how many

/// changes, the latest, and whether it is conflicted. A pure function of

/// the heads, so it is kept on disk as well and a restart costs nothing:

/// metalgear's 162 bookmarks took ~31 s to summarise again after one.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub struct Brief {

    pub changes: usize,

    pub latest_ms: u64,

    pub conflicted: bool,

}

/// Most summaries computed at once when a page needs many.

const PARALLEL: usize = 4;

    /// Where briefs are kept on disk, when anywhere.

    briefs: Option<PathBuf>,

        StateCache {

            capacity: capacity.max(1),

            budget,

            inner: Mutex::new((HashMap::new(), 0)),

            summaries: Mutex::new((HashMap::new(), 0)),

            briefs: None,

        }

    }

    /// Keep briefs on disk under `dir`, one small file per state.

    pub fn with_briefs(mut self, dir: PathBuf) -> StateCache {

        self.briefs = Some(dir);

        self

    }

    fn brief_path(&self, key: &Key) -> Option<PathBuf> {

        let dir = self.briefs.as_ref()?;

        let mut h = blake3::Hasher::new();

        for c in &key.heads {

            h.update(&c.0);

        }

        Some(dir.join(hex::encode(key.repo)).join(h.finalize().to_hex().as_str()))

    }

    fn known_brief(&self, key: &Key) -> Option<Brief> {

        if let Some((s, _)) = self.summaries.lock().unwrap().0.get(key) {

            return Some(s.brief());

        }

        read_brief(&self.brief_path(key)?)

    }

    /// The brief of the state with these heads: from memory, from disk, or

    /// worked out and kept.

    pub fn brief(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Brief, oxy_core::repo::RepoError> {

        Ok(self.briefs(repo, &[heads.to_vec()])?[0])

    }

    /// The briefs of many states, those not yet known worked out a few at

    /// a time in parallel.

    pub fn briefs(&self, repo: &Repo, all: &[Vec<ChangeHash>]) -> Result<Vec<Brief>, oxy_core::repo::RepoError> {

        let id = repo.id()?;

        let mut out: Vec<Option<Brief>> = all.iter().map(|h| self.known_brief(&Key { repo: id, heads: sorted(h) })).collect();

        let missing: Vec<usize> = (0..all.len()).filter(|&i| out[i].is_none()).collect();

        if missing.is_empty() {

            return Ok(out.into_iter().map(|b| b.expect("known")).collect());

        }

        let next = Mutex::new(missing.iter());

        let found: Mutex<Vec<(usize, Result<Brief, oxy_core::repo::RepoError>)>> = Mutex::new(Vec::new());

        let threads = PARALLEL.min(missing.len()).min(std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1));

        // A `Repo` stays on its thread, so each worker opens its own.

        let root = repo.root().to_path_buf();

        std::thread::scope(|s| {

            for _ in 0..threads {

                s.spawn(|| {

                    let mine = Repo::open(&root);

                    loop {

                        let Some(&i) = next.lock().unwrap().next() else { break };

                        let b = match &mine {

                            Ok(r) => self.summary(r, &all[i]).map(|s| s.brief()),

                            Err(e) => Err(oxy_core::repo::RepoError::Msg(format!("open {}: {e}", root.display()))),

                        };

                        found.lock().unwrap().push((i, b));

                    }

                });

            }

        });

        for (i, b) in found.into_inner().unwrap() {

            out[i] = Some(b?);

        }

        Ok(out.into_iter().map(|b| b.expect("worked out")).collect())

        let sorted = sorted(heads);

        if let Some(path) = self.brief_path(&key) {

            write_brief(&path, summary.brief());

        }

    }

}

fn sorted(heads: &[ChangeHash]) -> Vec<ChangeHash> {

    let mut v = heads.to_vec();

    v.sort();

    v.dedup();

    v

}

/// A brief on disk: `changes latest_ms conflicted`.

fn read_brief(path: &Path) -> Option<Brief> {

    let text = std::fs::read_to_string(path).ok()?;

    let mut f = text.split_whitespace();

    Some(Brief { changes: f.next()?.parse().ok()?, latest_ms: f.next()?.parse().ok()?, conflicted: f.next()? == "1" })

}

/// Kept when it can be; a brief that is not kept is only worked out again.

fn write_brief(path: &Path, b: Brief) {

    let Some(dir) = path.parent() else { return };

    if std::fs::create_dir_all(dir).is_err() {

        return;

    }

    let tmp = path.with_extension(format!("tmp{}", std::process::id()));

    if std::fs::write(&tmp, format!("{} {} {}", b.changes, b.latest_ms, u8::from(b.conflicted))).is_ok() {

        std::fs::rename(&tmp, path).ok();

    }

    #[test]

    fn briefs_are_worked_out_in_parallel_and_survive_a_restart() {

        let dir = tempfile::tempdir().unwrap();

        let repo = Repo::init(dir.path()).unwrap();

        std::fs::write(dir.path().join("a.txt"), "a\n").unwrap();

        let one = repo.commit("one", &[]).unwrap();

        std::fs::write(dir.path().join("b.txt"), "b\n").unwrap();

        let two = repo.commit("two", &[]).unwrap();

        let kept = tempfile::tempdir().unwrap();

        let cache = StateCache::new(8).with_briefs(kept.path().to_path_buf());

        let got = cache.briefs(&repo, &[vec![one], vec![one, two], vec![]]).unwrap();

        assert_eq!(got.iter().map(|b| b.changes).collect::<Vec<_>>(), vec![1, 2, 0]);

        assert!(got[1].latest_ms >= got[0].latest_ms && !got[1].conflicted);

        // A new cache (a restarted hub) reads them back without loading a

        // state: a repository that no longer has the changes still answers.

        let gone = tempfile::tempdir().unwrap();

        let empty = Repo::init(gone.path()).unwrap();

        let empty_id = empty.id().unwrap();

        for f in std::fs::read_dir(kept.path()).unwrap() {

            let f = f.unwrap().path();

            std::fs::rename(&f, kept.path().join(hex::encode(empty_id))).unwrap();

        }

        let restarted = StateCache::new(8).with_briefs(kept.path().to_path_buf());

        assert_eq!(restarted.brief(&empty, &[two, one]).unwrap(), got[1], "from disk, whatever the heads' order");

        assert!(StateCache::new(8).brief(&empty, &[one, two]).is_err(), "without the disk it would have to load");

        // Unknown heads are an error, not a guess.

        assert!(cache.briefs(&repo, &[vec![one], vec![oxy_core::hash::ObjectRef([9; 32])]]).is_err());

        // A bare repository, as the hub keeps them, works across threads too.

        let bare_dir = tempfile::tempdir().unwrap();

        let bare = Repo::init_bare(bare_dir.path()).unwrap();

        assert_eq!(cache.briefs(&bare, &[vec![], vec![]]).unwrap().len(), 2);

        // States are bounded by their estimated bytes (`cache::budget`); the

        // count only stops many tiny states piling up. At 64 it was always

        // full and evicting long before the bytes were (#145).

        let briefs = root.parent().unwrap_or(&root).join("cache").join("summaries").join(root.file_name().unwrap_or_default());

        let cache = Arc::new(StateCache::new(256).with_briefs(briefs));

        Repos { root, locks: Mutex::new(HashMap::new()), cache, searches: search::SearchCache::default() }

    let list = repo.bookmarks()?;

    let briefs = cache.briefs(repo, &list.iter().map(|b| b.heads.clone()).collect::<Vec<_>>())?;

    for (b, s) in list.iter().zip(briefs) {

            changes: s.changes,

    let list = repo.bookmarks()?;

    let briefs = cache.briefs(repo, &list.iter().map(|b| b.heads.clone()).collect::<Vec<_>>())?;

    for (b, brief) in list.iter().zip(briefs) {

        if !brief.conflicted {

Keyed by state hash, so they never go stale: loaded `State` objects (LRU by memory, up to 256 and `OXYHUB_STATE_CACHE_MB`, default 3 GiB), a state's brief (its change count, latest change and whether it is conflicted, #145: kept in memory and on disk under `<data>/cache/summaries/`, worked out four at a time when a page needs many and as soon as a push moves a bookmark, so a restart never makes a bookmark list or an overview load every bookmark's state again), tree listings, rendered file views, blame, history graph layouts per bookmark head set, projected states per proposal, archives, code search results. Keyed by content: rendered Markdown (BLAKE3 of the text and its two link bases, 32 MiB) and syntax highlighted files (BLAKE3 of the file name, which picks the syntax, and the text, 64 MiB), each an LRU bounded by the bytes it holds, rendering outside the lock; an entry larger than a quarter of its budget is not kept. Keyed by repository: bookmark and tag lists, invalidated by the push path. `ETag` and `Cache-Control: private, max-age=0, must-revalidate` on pages; `immutable` on content-addressed responses (raw files by hash, archives by state). A request that takes two seconds or more is logged at warn with its method, path (never the query) and status.