history_scan.rs 91.2%

279 of 306 lines run, 48 of 85 functions · run #503 on main · run missed

//! The history scan (hub spec §15.4, oxygenhub#277): push protection's
//! rules run over what a repository already holds, so a secret committed
//! before push protection existed, or while it was off, is found too.
//!
//! What is scanned is every change reachable from the repository's
//! bookmarks: the lines it inserts and the text blobs it puts in place of
//! a file's content, exactly what push protection reads in a push. A
//! repository's progress is a row in `secret_scans`: the heads whose
//! history is scanned (`done_heads`), and while a scan is under way the
//! heads it works toward (`target_heads`) and how far it got (`position`
//! in its list of changes: those reachable from the target and not from
//! the done heads, in clock order then by hash, so the same list every
//! pass). A pass scans within a [`Budget`] and saves where it stopped; the
//! next one resumes there. Once the list is done the target becomes the
//! done heads, so after a push only the new changes are scanned. A change
//! of rules (`version` in `push_protection.toml`) or a maintainer's "Scan
//! history now" scans the whole history again; findings are kept one per
//! value and file, so a rescan adds nothing already known and keeps what
//! was resolved.
//!
//! The scan is the worker role's `secrets.history_scan` task (docs/roles.md),
//! one repository at a time, one pass each in turn, on one blocking thread
//! that rests as long as it works: at most half a core, so it never starves
//! requests. It reads repositories with the shared lock only, as pages do.
//! Repositories with push protection off are not scanned.

use crate::bus::{Event, Topic};
use crate::config::Config;
use crate::db::{self, push_protection as pdb, repos::RepoRow};
use crate::push_protection::{self as pp, Finding};
use crate::repos::{RepoError, RepoName};
use crate::Shared;
use oxy_core::hash::{ChangeHash, ObjectType};
use oxy_core::objects::{Blob, Edit};
use oxy_core::protocol::Bundle;
use oxy_core::repo::Repo;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::time::{Duration, Instant};

/// How much one pass over one repository may scan: `bytes` of text (a
/// change counts at least one, however little it adds) and `time`
/// wall-clock (resting included), each file or inserted text at most
/// `file` bytes. At least one change is scanned per pass, so a pass always
/// gets somewhere.
#[derive(Clone, Copy, Debug)]
pub struct Budget {
    pub bytes: u64,
    pub time: Duration,
    pub file: u64,
}

impl Default for Budget {
    fn default() -> Self {
        Budget { bytes: 64 << 20, time: Duration::from_secs(5), file: pp::Budget::default().file }
    }
}

impl Budget {
    /// The hub's budget from `[limits]`; `None` when the scan is off.
    pub fn of(config: &Config) -> Option<Budget> {
        let l = &config.limits;
        (l.history_scan_limit.0 > 0).then(|| Budget {
            bytes: l.history_scan_limit.0,
            time: Duration::from_millis(l.history_scan_time_ms),
            file: l.scan_file_limit.0,
        })
    }
}

/// What one pass did.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Pass {
    /// Changes and bytes scanned in this pass.
    pub changes: usize,
    pub bytes: u64,
    /// Findings new to this repository.
    pub found: usize,
    /// Changes still to scan toward the target.
    pub left: usize,
    /// Nothing to scan: the history up to the bookmarks is scanned.
    pub idle: bool,
}

/// How long the worker waits after starting before its first sweep, so a
/// starting hub serves first.
const START_DELAY: Duration = Duration::from_secs(10);

/// How long it rests between repositories.
const PAUSE: Duration = Duration::from_millis(200);

/// Every repository is looked at again this often, for the pushes the bus
/// did not tell of (it delivers at most once) and bookmarks moved without
/// a push (landings).
const SWEEP_EVERY: Duration = Duration::from_secs(15 * 60);

/// The bus topic a "Scan history now" is announced on, its kind the
/// repository's id.
pub const REQUEST_TOPIC: &str = "secrets:scan";

/// The worker role's task: every repository first, then those a push, a
/// request or a sweep says have something new, each a pass at a time.
pub async fn worker(app: Shared) {
    let Some(budget) = Budget::of(&app.config) else {
        // Off in `[limits]`: the task holds its lock and does nothing.
        return std::future::pending().await;
    };
    let mut events = app.bus.subscribe(Topic::Live);
    tokio::time::sleep(START_DELAY).await;
    let mut sweep = true;
    let mut dirty: BTreeSet<i64> = BTreeSet::new();
    loop {
        let mut order = pdb::requested(&app.db).await.unwrap_or_default();
        if sweep {
            order.extend(db::repos::list_with_drops(&app.db).await.unwrap_or_default().into_iter().map(|r| r.id));
            sweep = false;
        }
        order.extend(std::mem::take(&mut dirty));
        let mut seen = HashSet::new();
        order.retain(|id| seen.insert(*id));
        for id in order {
            sweep |= hear(&mut events, &mut dirty);
            let Ok(Some(row)) = db::repos::get_by_id(&app.db, id).await else { continue };
            if !row.push_protection {
                continue;
            }
            match pass(&app, &row, budget).await {
                Ok(p) => {
                    if p.left > 0 {
                        dirty.insert(id);
                    }
                    if p.found > 0 {
                        tracing::warn!(
                            "history scan: {}/{}: {} possible secret(s) found in history; see Settings › Push protection",
                            row.owner,
                            row.name,
                            p.found
                        );
                    }
                }
                Err(e) => tracing::warn!("history scan: {}/{}: {e:#}", row.owner, row.name),
            }
            tokio::time::sleep(PAUSE).await;
        }
        sweep |= hear(&mut events, &mut dirty);
        if sweep || !dirty.is_empty() {
            continue;
        }
        // Nothing left: wait for a push, a request or the next sweep.
        let wake = tokio::time::sleep(SWEEP_EVERY);
        tokio::pin!(wake);
        loop {
            tokio::select! {
                ev = events.recv() => match ev {
                    Ok(ev) => if let Some(id) = wakes(&ev) { dirty.insert(id); break; },
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { sweep = true; break; }
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => { (&mut wake).await; sweep = true; break; }
                },
                _ = &mut wake => { sweep = true; break; }
            }
        }
        if !sweep {
            // Let a burst of pushes settle into one pass each.
            tokio::time::sleep(Duration::from_secs(1)).await;
            hear(&mut events, &mut dirty);
        }
    }
}

/// Take the events waiting: the repositories they name go in `dirty`.
/// Whether events were lost, so everything should be looked at.
fn hear(events: &mut tokio::sync::broadcast::Receiver<Event>, dirty: &mut BTreeSet<i64>) -> bool {
    use tokio::sync::broadcast::error::TryRecvError;
    loop {
        match events.try_recv() {
            Ok(ev) => {
                if let Some(id) = wakes(&ev) {
                    dirty.insert(id);
                }
            }
            Err(TryRecvError::Lagged(_)) => return true,
            Err(_) => return false,
        }
    }
}

/// The repository an event says has something to scan: a push to it
/// (`repo:N`, `push`), or a maintainer's request.
fn wakes(ev: &Event) -> Option<i64> {
    let Event::Live { topic, kind } = ev else { return None };
    if topic == REQUEST_TOPIC {
        return kind.parse().ok();
    }
    if kind != "push" {
        return None;
    }
    topic.strip_prefix("repo:")?.parse().ok()
}

/// Ask for repository `repo_id`'s whole history to be scanned again now
/// ("Scan history now", maintain role).
pub async fn request(app: &Shared, repo_id: i64) -> sqlx::Result<()> {
    pdb::request_scan(&app.db, repo_id).await?;
    wake(app, repo_id);
    Ok(())
}

/// Tell the worker that repository `repo_id` may have history to scan (its
/// push protection was turned back on).
pub fn wake(app: &Shared, repo_id: i64) {
    app.bus.live(REQUEST_TOPIC, &repo_id.to_string());
}

fn heads_text(heads: &[ChangeHash]) -> String {
    heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>().join(",")
}

fn heads_of(text: &str) -> Vec<ChangeHash> {
    text.split(',').filter_map(ChangeHash::from_hex).collect()
}

/// One pass over `row`'s history within `budget`: resume the scan under
/// way, or start one toward the bookmarks if they moved past what was
/// scanned; record what it finds and where it stopped.
pub async fn pass(app: &Shared, row: &RepoRow, budget: Budget) -> anyhow::Result<Pass> {
    let name = RepoName::new(&row.owner, &row.name).ok_or_else(|| anyhow::anyhow!("not a repository name"))?;
    let loaded = pdb::scan_of(&app.db, row.id).await?.unwrap_or(pdb::ScanRow { repo_id: row.id, ..Default::default() });
    let mut s = loaded.clone();
    let version = pp::rules().version as i64;
    if s.rules_version != version {
        // New rules: the whole history again.
        s.rules_version = version;
        s.done_heads.clear();
        s.target_heads.clear();
        s.position = 0;
    }
    let from = Progress { done: heads_of(&s.done_heads), target: heads_of(&s.target_heads), position: s.position.max(0) as usize };
    let default = row.default_bookmark.clone();
    let started = db::now_ms();
    let work = app.repos.read(&name, move |repo, _| scan_in(repo, from, &default, budget)).await;
    let work = match work {
        Ok(w) => w,
        Err(e) => {
            s.error = e.to_string();
            s.pass_ms = Some(started);
            save(app, &loaded, &s).await?;
            return Err(e.into());
        }
    };
    if work.idle && s == loaded {
        // Up to date, as the row already says: nothing to write.
        return Ok(Pass { idle: true, ..Default::default() });
    }
    let records: Vec<pdb::NewHistoryFinding> = work
        .findings
        .iter()
        .map(|(f, bookmark)| pdb::NewHistoryFinding {
            finding: pdb::NewFinding {
                rule: &f.rule,
                description: &f.description,
                path: &f.path,
                line: f.line as i64,
                fingerprint: &f.fingerprint,
                excerpt: &f.excerpt,
            },
            change_hash: f.change_hash.map(|h| h.to_hex()).unwrap_or_default(),
            bookmark: bookmark.clone(),
        })
        .collect();
    let found = pdb::record_history(&app.db, row.id, &records).await?;
    let left = work.pending - work.position;
    if work.idle || left == 0 {
        s.done_heads = heads_text(&work.target);
        s.target_heads.clear();
        s.position = 0;
        s.pending = 0;
        s.requested_ms = None;
        if !work.idle {
            s.finished_ms = Some(db::now_ms());
        }
    } else {
        s.target_heads = heads_text(&work.target);
        s.position = work.position as i64;
        s.pending = work.pending as i64;
    }
    s.changes += work.changes as i64;
    s.bytes += work.bytes as i64;
    s.pass_ms = Some(started);
    s.error.clear();
    save(app, &loaded, &s).await?;
    Ok(Pass { changes: work.changes, bytes: work.bytes, found, left, idle: work.idle })
}

/// Save `s` unless a maintainer asked for a new scan while the pass ran:
/// that request's fresh start wins.
async fn save(app: &Shared, loaded: &pdb::ScanRow, s: &pdb::ScanRow) -> sqlx::Result<()> {
    let now = pdb::scan_of(&app.db, s.repo_id).await?;
    if now.as_ref().and_then(|n| n.requested_ms) != loaded.requested_ms {
        return Ok(());
    }
    pdb::save_scan(&app.db, s).await
}

/// Where a scan stands, as read from its row.
struct Progress {
    done: Vec<ChangeHash>,
    target: Vec<ChangeHash>,
    position: usize,
}

/// What [`scan_in`] did.
struct Work {
    /// The heads it scanned toward, and how far it got of how many.
    target: Vec<ChangeHash>,
    position: usize,
    pending: usize,
    idle: bool,
    changes: usize,
    bytes: u64,
    /// Each finding, placed, with the bookmark it was found on.
    findings: Vec<(Finding, String)>,
}

/// The pass itself, on a blocking thread with the repository open.
fn scan_in(repo: &Repo, from: Progress, default: &str, budget: Budget) -> Result<Work, RepoError> {
    let started = Instant::now();
    let view = repo.view()?;
    let mut bookmarks: Vec<(String, Vec<ChangeHash>)> = Vec::new();
    for name in view.bookmarks.keys() {
        bookmarks.push((name.clone(), repo.heads_of(&view, name)?));
    }
    // The default bookmark first: a finding is named by the first bookmark
    // that reaches it.
    bookmarks.sort_by_key(|(n, _)| (n != default, n.clone()));
    let mut current: Vec<ChangeHash> = bookmarks.iter().flat_map(|(_, h)| h.iter().copied()).collect();
    current.sort();
    current.dedup();
    let mut done = from.done;
    done.sort();
    let (mut target, mut position) = (from.target, from.position);
    if target.is_empty() {
        if current == done {
            return Ok(Work { target: done, position: 0, pending: 0, idle: true, changes: 0, bytes: 0, findings: vec![] });
        }
        target = current.clone();
        position = 0;
    }
    let state = match repo.load_state_history(&target) {
        Ok(s) => s,
        // What the scan worked toward is gone (a bookmark rewritten, then
        // collected): start again toward the bookmarks as they are.
        Err(_) if target != current => {
            target = current.clone();
            position = 0;
            repo.load_state_history(&target)?
        }
        Err(e) => return Err(e.into()),
    };
    // What is scanned already: the done heads' history.
    let scanned: HashSet<ChangeHash> = if done.iter().all(|h| state.contains(h)) {
        closure(&done, |h| state.get(h).map(|m| m.deps.clone()))
    } else {
        match repo.load_state_history(&done) {
            Ok(old) => old.changes().map(|(h, _)| *h).collect(),
            Err(_) => HashSet::new(),
        }
    };
    let mut list: Vec<(oxy_core::objects::Hlc, ChangeHash)> =
        state.changes().filter(|(h, _)| !scanned.contains(*h)).map(|(h, m)| (m.hlc, *h)).collect();
    list.sort();
    let pending = list.len();
    if position > pending {
        position = 0;
    }
    let file = pp::Budget { bytes: u64::MAX, file: budget.file, time: Duration::MAX };
    let mut work = Work { target, position, pending, idle: false, changes: 0, bytes: 0, findings: vec![] };
    let mut reach: HashMap<String, HashSet<ChangeHash>> = HashMap::new();
    let mut busy = Duration::ZERO;
    let mut spent: u64 = 0;
    while work.position < pending {
        if work.changes > 0 && (spent >= budget.bytes || started.elapsed() >= budget.time) {
            break;
        }
        let t = Instant::now();
        let h = list[work.position].1;
        let bundle = bundle_of(repo, &h, budget.file)?;
        let scan = pp::scan(&bundle, file);
        work.bytes += scan.scanned;
        spent += scan.scanned.max(1);
        if !scan.findings.is_empty() {
            let mut findings = scan.findings;
            pp::locate(repo, &bundle, &mut findings);
            let on = bookmarks
                .iter()
                .find(|(name, heads)| {
                    reach.entry(name.clone()).or_insert_with(|| closure(heads, |c| state.get(c).map(|m| m.deps.clone()))).contains(&h)
                })
                .map(|(name, _)| name.clone())
                .unwrap_or_default();
            work.findings.extend(findings.into_iter().map(|f| (f, on.clone())));
        }
        work.position += 1;
        work.changes += 1;
        // Rest as long as it worked, so the scan takes at most half a core.
        busy += t.elapsed();
        if busy >= Duration::from_millis(20) {
            std::thread::sleep(busy);
            busy = Duration::ZERO;
        }
    }
    Ok(work)
}

/// `heads` and every change they depend on, by `deps`.
fn closure(heads: &[ChangeHash], deps: impl Fn(&ChangeHash) -> Option<Vec<ChangeHash>>) -> HashSet<ChangeHash> {
    let mut seen: HashSet<ChangeHash> = HashSet::new();
    let mut stack: Vec<ChangeHash> = heads.to_vec();
    while let Some(h) = stack.pop() {
        if !seen.insert(h) {
            continue;
        }
        if let Some(d) = deps(&h) {
            stack.extend(d.into_iter().filter(|x| !seen.contains(x)));
        }
    }
    seen
}

/// One stored change as push protection reads a push: the change, and the
/// blobs (with their chunks, unless over the file budget) its edits put in
/// place of a file's content. A blob the store does not hold (lazy) is
/// left out, so it is not scanned.
fn bundle_of(repo: &Repo, h: &ChangeHash, file: u64) -> Result<Bundle, RepoError> {
    let store = repo.store();
    let change = store.get_change(h)?;
    let mut objects = Vec::new();
    for e in &change.edits {
        let Edit::Replace { new: Some(r), .. } = e else { continue };
        let Ok(body) = store.get_typed(r, ObjectType::Blob) else { continue };
        let Ok(blob) = Blob::decode(&body) else { continue };
        if blob.length <= file {
            for c in &blob.chunks {
                if let Ok(chunk) = store.get_typed(c, ObjectType::Chunk) {
                    objects.push((ObjectType::Chunk, chunk));
                }
            }
        }
        objects.push((ObjectType::Blob, body));
    }
    Ok(Bundle { changes: vec![(change, vec![])], hashes: vec![*h], objects, ..Default::default() })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn pushes_and_requests_wake_the_scan() {
        let live = |topic: &str, kind: &str| Event::Live { topic: topic.into(), kind: kind.into() };
        assert_eq!(wakes(&live("repo:7", "push")), Some(7));
        assert_eq!(wakes(&live("repo:7", "issue")), None);
        assert_eq!(wakes(&live("repo:7/proposal:2", "push")), None);
        assert_eq!(wakes(&live(REQUEST_TOPIC, "9")), Some(9));
        assert_eq!(wakes(&Event::DeliveryWaiting), None);
    }

    #[test]
    fn heads_round_trip() {
        let a = ChangeHash::from_hex(&"ab".repeat(32)).unwrap();
        let b = ChangeHash::from_hex(&"cd".repeat(32)).unwrap();
        assert_eq!(heads_of(&heads_text(&[a, b])), [a, b]);
        assert!(heads_of("").is_empty());
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! The history scan (hub spec §15.4, oxygenhub#277): push protection's
2		//! rules run over what a repository already holds, so a secret committed
3		//! before push protection existed, or while it was off, is found too.
4		//!
5		//! What is scanned is every change reachable from the repository's
6		//! bookmarks: the lines it inserts and the text blobs it puts in place of
7		//! a file's content, exactly what push protection reads in a push. A
8		//! repository's progress is a row in `secret_scans`: the heads whose
9		//! history is scanned (`done_heads`), and while a scan is under way the
10		//! heads it works toward (`target_heads`) and how far it got (`position`
11		//! in its list of changes: those reachable from the target and not from
12		//! the done heads, in clock order then by hash, so the same list every
13		//! pass). A pass scans within a [`Budget`] and saves where it stopped; the
14		//! next one resumes there. Once the list is done the target becomes the
15		//! done heads, so after a push only the new changes are scanned. A change
16		//! of rules (`version` in `push_protection.toml`) or a maintainer's "Scan
17		//! history now" scans the whole history again; findings are kept one per
18		//! value and file, so a rescan adds nothing already known and keeps what
19		//! was resolved.
20		//!
21		//! The scan is the worker role's `secrets.history_scan` task (docs/roles.md),
22		//! one repository at a time, one pass each in turn, on one blocking thread
23		//! that rests as long as it works: at most half a core, so it never starves
24		//! requests. It reads repositories with the shared lock only, as pages do.
25		//! Repositories with push protection off are not scanned.
26
27		use crate::bus::{Event, Topic};
28		use crate::config::Config;
29		use crate::db::{self, push_protection as pdb, repos::RepoRow};
30		use crate::push_protection::{self as pp, Finding};
31		use crate::repos::{RepoError, RepoName};
32		use crate::Shared;
33		use oxy_core::hash::{ChangeHash, ObjectType};
34		use oxy_core::objects::{Blob, Edit};
35		use oxy_core::protocol::Bundle;
36		use oxy_core::repo::Repo;
37		use std::collections::{BTreeSet, HashMap, HashSet};
38		use std::time::{Duration, Instant};
39
40		/// How much one pass over one repository may scan: `bytes` of text (a
41		/// change counts at least one, however little it adds) and `time`
42		/// wall-clock (resting included), each file or inserted text at most
43		/// `file` bytes. At least one change is scanned per pass, so a pass always
44		/// gets somewhere.
45		#[derive(Clone, Copy, Debug)]
46		pub struct Budget {
47		pub bytes: u64,
48		pub time: Duration,
49		pub file: u64,
50		}
51
52		impl Default for Budget {
53	202×	fn default() -> Self {
54	202×	Budget { bytes: 64 << 20, time: Duration::from_secs(5), file: pp::Budget::default().file }
55	202×	}
56		}
57
58		impl Budget {
59		/// The hub's budget from `[limits]`; `None` when the scan is off.
60	325×	pub fn of(config: &Config) -> Option<Budget> {
61	325×	let l = &config.limits;
62	325×	(l.history_scan_limit.0 > 0).then(\|\| Budget {
63	314×	bytes: l.history_scan_limit.0,
64	314×	time: Duration::from_millis(l.history_scan_time_ms),
65	314×	file: l.scan_file_limit.0,
66	314×	})
67	325×	}
68		}
69
70		/// What one pass did.
71		#[derive(Clone, Debug, Default, PartialEq, Eq)]
72		pub struct Pass {
73		/// Changes and bytes scanned in this pass.
74		pub changes: usize,
75		pub bytes: u64,
76		/// Findings new to this repository.
77		pub found: usize,
78		/// Changes still to scan toward the target.
79		pub left: usize,
80		/// Nothing to scan: the history up to the bookmarks is scanned.
81		pub idle: bool,
82		}
83
84		/// How long the worker waits after starting before its first sweep, so a
85		/// starting hub serves first.
86		const START_DELAY: Duration = Duration::from_secs(10);
87
88		/// How long it rests between repositories.
89		const PAUSE: Duration = Duration::from_millis(200);
90
91		/// Every repository is looked at again this often, for the pushes the bus
92		/// did not tell of (it delivers at most once) and bookmarks moved without
93		/// a push (landings).
94		const SWEEP_EVERY: Duration = Duration::from_secs(15 * 60);
95
96		/// The bus topic a "Scan history now" is announced on, its kind the
97		/// repository's id.
98		pub const REQUEST_TOPIC: &str = "secrets:scan";
99
100		/// The worker role's task: every repository first, then those a push, a
101		/// request or a sweep says have something new, each a pass at a time.
102	192×	pub async fn worker(app: Shared) {
103	192×	let Some(budget) = Budget::of(&app.config) else {
104		// Off in `[limits]`: the task holds its lock and does nothing.
105	2×	return std::future::pending().await;
106		};
107	190×	let mut events = app.bus.subscribe(Topic::Live);
108	190×	tokio::time::sleep(START_DELAY).await;
109	6×	let mut sweep = true;
110	6×	let mut dirty: BTreeSet<i64> = BTreeSet::new();
111		loop {
112	14×	let mut order = pdb::requested(&app.db).await.unwrap_or_default();
113	14×	if sweep {
114	6×	order.extend(db::repos::list_with_drops(&app.db).await.unwrap_or_default().into_iter().map(\|r\| r.id));
115	6×	sweep = false;
116	8×	}
117	14×	order.extend(std::mem::take(&mut dirty));
118	14×	let mut seen = HashSet::new();
119	14×	order.retain(\|id\| seen.insert(*id));
120	14×	for id in order {
121	12×	sweep \|= hear(&mut events, &mut dirty);
122	12×	let Ok(Some(row)) = db::repos::get_by_id(&app.db, id).await else { continue };
123	12×	if !row.push_protection {
124	1×	continue;
125	11×	}
126	11×	match pass(&app, &row, budget).await {
127	11×	Ok(p) => {
128	11×	if p.left > 0 {
129	0×	dirty.insert(id);
130	11×	}
131	11×	if p.found > 0 {
132	2×	tracing::warn!(
133		"history scan: {}/{}: {} possible secret(s) found in history; see Settings › Push protection",
134		row.owner,
135		row.name,
136		p.found
137		);
138	9×	}
139		}
140	0×	Err(e) => tracing::warn!("history scan: {}/{}: {e:#}", row.owner, row.name),
141		}
142	11×	tokio::time::sleep(PAUSE).await;
143		}
144	13×	sweep \|= hear(&mut events, &mut dirty);
145	13×	if sweep \|\| !dirty.is_empty() {
146	5×	continue;
147	8×	}
148		// Nothing left: wait for a push, a request or the next sweep.
149	8×	let wake = tokio::time::sleep(SWEEP_EVERY);
150	8×	tokio::pin!(wake);
151		loop {
152	16×	tokio::select! {
153	16×	ev = events.recv() => match ev {
154	12×	Ok(ev) => if let Some(id) = wakes(&ev) { dirty.insert(id); break; },
155	0×	Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { sweep = true; break; }
156	0×	Err(tokio::sync::broadcast::error::RecvError::Closed) => { (&mut wake).await; sweep = true; break; }
157		},
158	16×	_ = &mut wake => { sweep = true; break; }
159		}
160		}
161	4×	if !sweep {
162		// Let a burst of pushes settle into one pass each.
163	4×	tokio::time::sleep(Duration::from_secs(1)).await;
164	3×	hear(&mut events, &mut dirty);
165	0×	}
166		}
167	0×	}
168
169		/// Take the events waiting: the repositories they name go in `dirty`.
170		/// Whether events were lost, so everything should be looked at.
171	28×	fn hear(events: &mut tokio::sync::broadcast::Receiver<Event>, dirty: &mut BTreeSet<i64>) -> bool {
172		use tokio::sync::broadcast::error::TryRecvError;
173		loop {
174	84×	match events.try_recv() {
175	56×	Ok(ev) => {
176	56×	if let Some(id) = wakes(&ev) {
177	24×	dirty.insert(id);
178	32×	}
179		}
180	0×	Err(TryRecvError::Lagged(_)) => return true,
181	28×	Err(_) => return false,
182		}
183		}
184	28×	}
185
186		/// The repository an event says has something to scan: a push to it
187		/// (`repo:N`, `push`), or a maintainer's request.
188	73×	fn wakes(ev: &Event) -> Option<i64> {
189	73×	let Event::Live { topic, kind } = ev else { return None };
190	72×	if topic == REQUEST_TOPIC {
191	1×	return kind.parse().ok();
192	71×	}
193	71×	if kind != "push" {
194	41×	return None;
195	30×	}
196	30×	topic.strip_prefix("repo:")?.parse().ok()
197	73×	}
198
199		/// Ask for repository `repo_id`'s whole history to be scanned again now
200		/// ("Scan history now", maintain role).
201	1×	pub async fn request(app: &Shared, repo_id: i64) -> sqlx::Result<()> {
202	1×	pdb::request_scan(&app.db, repo_id).await?;
203	1×	wake(app, repo_id);
204	1×	Ok(())
205	1×	}
206
207		/// Tell the worker that repository `repo_id` may have history to scan (its
208		/// push protection was turned back on).
209	3×	pub fn wake(app: &Shared, repo_id: i64) {
210	3×	app.bus.live(REQUEST_TOPIC, &repo_id.to_string());
211	3×	}
212
213	20×	fn heads_text(heads: &[ChangeHash]) -> String {
214	117×	heads.iter().map(\|h\| h.to_hex()).collect::<Vec<_>>().join(",")
215	20×	}
216
217	50×	fn heads_of(text: &str) -> Vec<ChangeHash> {
218	50×	text.split(',').filter_map(ChangeHash::from_hex).collect()
219	50×	}
220
221		/// One pass over `row`'s history within `budget`: resume the scan under
222		/// way, or start one toward the bookmarks if they moved past what was
223		/// scanned; record what it finds and where it stopped.
224	24×	pub async fn pass(app: &Shared, row: &RepoRow, budget: Budget) -> anyhow::Result<Pass> {
225	24×	let name = RepoName::new(&row.owner, &row.name).ok_or_else(\|\| anyhow::anyhow!("not a repository name"))?;
226	24×	let loaded = pdb::scan_of(&app.db, row.id).await?.unwrap_or(pdb::ScanRow { repo_id: row.id, ..Default::default() });
227	24×	let mut s = loaded.clone();
228	24×	let version = pp::rules().version as i64;
229	24×	if s.rules_version != version {
230	6×	// New rules: the whole history again.
231	6×	s.rules_version = version;
232	6×	s.done_heads.clear();
233	6×	s.target_heads.clear();
234	6×	s.position = 0;
235	18×	}
236	24×	let from = Progress { done: heads_of(&s.done_heads), target: heads_of(&s.target_heads), position: s.position.max(0) as usize };
237	24×	let default = row.default_bookmark.clone();
238	24×	let started = db::now_ms();
239	24×	let work = app.repos.read(&name, move \|repo, _\| scan_in(repo, from, &default, budget)).await;
240	24×	let work = match work {
241	24×	Ok(w) => w,
242	0×	Err(e) => {
243	0×	s.error = e.to_string();
244	0×	s.pass_ms = Some(started);
245	0×	save(app, &loaded, &s).await?;
246	0×	return Err(e.into());
247		}
248		};
249	24×	if work.idle && s == loaded {
250		// Up to date, as the row already says: nothing to write.
251	5×	return Ok(Pass { idle: true, ..Default::default() });
252	19×	}
253	19×	let records: Vec<pdb::NewHistoryFinding> = work
254	19×	.findings
255	19×	.iter()
256	19×	.map(\|(f, bookmark)\| pdb::NewHistoryFinding {
257	7×	finding: pdb::NewFinding {
258	7×	rule: &f.rule,
259	7×	description: &f.description,
260	7×	path: &f.path,
261	7×	line: f.line as i64,
262	7×	fingerprint: &f.fingerprint,
263	7×	excerpt: &f.excerpt,
264	7×	},
265	7×	change_hash: f.change_hash.map(\|h\| h.to_hex()).unwrap_or_default(),
266	7×	bookmark: bookmark.clone(),
267	7×	})
268	19×	.collect();
269	19×	let found = pdb::record_history(&app.db, row.id, &records).await?;
270	19×	let left = work.pending - work.position;
271	19×	if work.idle \|\| left == 0 {
272	14×	s.done_heads = heads_text(&work.target);
273	14×	s.target_heads.clear();
274	14×	s.position = 0;
275	14×	s.pending = 0;
276	14×	s.requested_ms = None;
277	14×	if !work.idle {
278	14×	s.finished_ms = Some(db::now_ms());
279	14×	}
280	5×	} else {
281	5×	s.target_heads = heads_text(&work.target);
282	5×	s.position = work.position as i64;
283	5×	s.pending = work.pending as i64;
284	5×	}
285	19×	s.changes += work.changes as i64;
286	19×	s.bytes += work.bytes as i64;
287	19×	s.pass_ms = Some(started);
288	19×	s.error.clear();
289	19×	save(app, &loaded, &s).await?;
290	19×	Ok(Pass { changes: work.changes, bytes: work.bytes, found, left, idle: work.idle })
291	24×	}
292
293		/// Save `s` unless a maintainer asked for a new scan while the pass ran:
294		/// that request's fresh start wins.
295	19×	async fn save(app: &Shared, loaded: &pdb::ScanRow, s: &pdb::ScanRow) -> sqlx::Result<()> {
296	19×	let now = pdb::scan_of(&app.db, s.repo_id).await?;
297	19×	if now.as_ref().and_then(\|n\| n.requested_ms) != loaded.requested_ms {
298	0×	return Ok(());
299	19×	}
300	19×	pdb::save_scan(&app.db, s).await
301	19×	}
302
303		/// Where a scan stands, as read from its row.
304		struct Progress {
305		done: Vec<ChangeHash>,
306		target: Vec<ChangeHash>,
307		position: usize,
308		}
309
310		/// What [`scan_in`] did.
311		struct Work {
312		/// The heads it scanned toward, and how far it got of how many.
313		target: Vec<ChangeHash>,
314		position: usize,
315		pending: usize,
316		idle: bool,
317		changes: usize,
318		bytes: u64,
319		/// Each finding, placed, with the bookmark it was found on.
320		findings: Vec<(Finding, String)>,
321		}
322
323		/// The pass itself, on a blocking thread with the repository open.
324	24×	fn scan_in(repo: &Repo, from: Progress, default: &str, budget: Budget) -> Result<Work, RepoError> {
325	24×	let started = Instant::now();
326	24×	let view = repo.view()?;
327	24×	let mut bookmarks: Vec<(String, Vec<ChangeHash>)> = Vec::new();
328	59×	for name in view.bookmarks.keys() {
329	59×	bookmarks.push((name.clone(), repo.heads_of(&view, name)?));
330		}
331		// The default bookmark first: a finding is named by the first bookmark
332		// that reaches it.
333	100×	bookmarks.sort_by_key(\|(n, _)\| (n != default, n.clone()));
334	59×	let mut current: Vec<ChangeHash> = bookmarks.iter().flat_map(\|(_, h)\| h.iter().copied()).collect();
335	24×	current.sort();
336	24×	current.dedup();
337	24×	let mut done = from.done;
338	24×	done.sort();
339	24×	let (mut target, mut position) = (from.target, from.position);
340	24×	if target.is_empty() {
341	19×	if current == done {
342	5×	return Ok(Work { target: done, position: 0, pending: 0, idle: true, changes: 0, bytes: 0, findings: vec![] });
343	14×	}
344	14×	target = current.clone();
345	14×	position = 0;
346	5×	}
347	19×	let state = match repo.load_state_history(&target) {
348	19×	Ok(s) => s,
349		// What the scan worked toward is gone (a bookmark rewritten, then
350		// collected): start again toward the bookmarks as they are.
351	0×	Err(_) if target != current => {
352	0×	target = current.clone();
353	0×	position = 0;
354	0×	repo.load_state_history(&target)?
355		}
356	0×	Err(e) => return Err(e.into()),
357		};
358		// What is scanned already: the done heads' history.
359	50×	let scanned: HashSet<ChangeHash> = if done.iter().all(\|h\| state.contains(h)) {
360	58×	closure(&done, \|h\| state.get(h).map(\|m\| m.deps.clone()))
361		} else {
362	0×	match repo.load_state_history(&done) {
363	0×	Ok(old) => old.changes().map(\|(h, _)\| *h).collect(),
364	0×	Err(_) => HashSet::new(),
365		}
366		};
367	19×	let mut list: Vec<(oxy_core::objects::Hlc, ChangeHash)> =
368	129×	state.changes().filter(\|(h, _)\| !scanned.contains(h)).map(\|(h, m)\| (m.hlc, h)).collect();
369	19×	list.sort();
370	19×	let pending = list.len();
371	19×	if position > pending {
372	0×	position = 0;
373	19×	}
374	19×	let file = pp::Budget { bytes: u64::MAX, file: budget.file, time: Duration::MAX };
375	19×	let mut work = Work { target, position, pending, idle: false, changes: 0, bytes: 0, findings: vec![] };
376	19×	let mut reach: HashMap<String, HashSet<ChangeHash>> = HashMap::new();
377	19×	let mut busy = Duration::ZERO;
378	19×	let mut spent: u64 = 0;
379	60×	while work.position < pending {
380	46×	if work.changes > 0 && (spent >= budget.bytes \|\| started.elapsed() >= budget.time) {
381	5×	break;
382	41×	}
383	41×	let t = Instant::now();
384	41×	let h = list[work.position].1;
385	41×	let bundle = bundle_of(repo, &h, budget.file)?;
386	41×	let scan = pp::scan(&bundle, file);
387	41×	work.bytes += scan.scanned;
388	41×	spent += scan.scanned.max(1);
389	41×	if !scan.findings.is_empty() {
390	7×	let mut findings = scan.findings;
391	7×	pp::locate(repo, &bundle, &mut findings);
392	7×	let on = bookmarks
393	7×	.iter()
394	7×	.find(\|(name, heads)\| {
395	23×	reach.entry(name.clone()).or_insert_with(\|\| closure(heads, \|c\| state.get(c).map(\|m\| m.deps.clone()))).contains(&h)
396	7×	})
397	7×	.map(\|(name, _)\| name.clone())
398	7×	.unwrap_or_default();
399	7×	work.findings.extend(findings.into_iter().map(\|f\| (f, on.clone())));
400	34×	}
401	41×	work.position += 1;
402	41×	work.changes += 1;
403		// Rest as long as it worked, so the scan takes at most half a core.
404	41×	busy += t.elapsed();
405	41×	if busy >= Duration::from_millis(20) {
406	0×	std::thread::sleep(busy);
407	0×	busy = Duration::ZERO;
408	41×	}
409		}
410	19×	Ok(work)
411	24×	}
412
413		/// `heads` and every change they depend on, by `deps`.
414	25×	fn closure(heads: &[ChangeHash], deps: impl Fn(&ChangeHash) -> Option<Vec<ChangeHash>>) -> HashSet<ChangeHash> {
415	25×	let mut seen: HashSet<ChangeHash> = HashSet::new();
416	25×	let mut stack: Vec<ChangeHash> = heads.to_vec();
417	123×	while let Some(h) = stack.pop() {
418	98×	if !seen.insert(h) {
419	17×	continue;
420	81×	}
421	81×	if let Some(d) = deps(&h) {
422	80×	stack.extend(d.into_iter().filter(\|x\| !seen.contains(x)));
423	1×	}
424		}
425	25×	seen
426	25×	}
427
428		/// One stored change as push protection reads a push: the change, and the
429		/// blobs (with their chunks, unless over the file budget) its edits put in
430		/// place of a file's content. A blob the store does not hold (lazy) is
431		/// left out, so it is not scanned.
432	41×	fn bundle_of(repo: &Repo, h: &ChangeHash, file: u64) -> Result<Bundle, RepoError> {
433	41×	let store = repo.store();
434	41×	let change = store.get_change(h)?;
435	41×	let mut objects = Vec::new();
436	46×	for e in &change.edits {
437	41×	let Edit::Replace { new: Some(r), .. } = e else { continue };
438	5×	let Ok(body) = store.get_typed(r, ObjectType::Blob) else { continue };
439	5×	let Ok(blob) = Blob::decode(&body) else { continue };
440	5×	if blob.length <= file {
441	5×	for c in &blob.chunks {
442	5×	if let Ok(chunk) = store.get_typed(c, ObjectType::Chunk) {
443	5×	objects.push((ObjectType::Chunk, chunk));
444	5×	}
445		}
446	0×	}
447	5×	objects.push((ObjectType::Blob, body));
448		}
449	41×	Ok(Bundle { changes: vec![(change, vec![])], hashes: vec![*h], objects, ..Default::default() })
450	41×	}
451
452		#[cfg(test)]
453		mod tests {
454		use super::*;
455
456		#[test]
457	1×	fn pushes_and_requests_wake_the_scan() {
458	4×	let live = \|topic: &str, kind: &str\| Event::Live { topic: topic.into(), kind: kind.into() };
459	1×	assert_eq!(wakes(&live("repo:7", "push")), Some(7));
460	1×	assert_eq!(wakes(&live("repo:7", "issue")), None);
461	1×	assert_eq!(wakes(&live("repo:7/proposal:2", "push")), None);
462	1×	assert_eq!(wakes(&live(REQUEST_TOPIC, "9")), Some(9));
463	1×	assert_eq!(wakes(&Event::DeliveryWaiting), None);
464	1×	}
465
466		#[test]
467	1×	fn heads_round_trip() {
468	1×	let a = ChangeHash::from_hex(&"ab".repeat(32)).unwrap();
469	1×	let b = ChangeHash::from_hex(&"cd".repeat(32)).unwrap();
470	1×	assert_eq!(heads_of(&heads_text(&[a, b])), [a, b]);
471	1×	assert!(heads_of("").is_empty());
472	1×	}
473		}

Coverage / crates/hub/src/history_scan.rs 91.2%

Keyboard