Krolyxon
Commit 55c1225

Branches

Tags

Refs →

ferreniumsrcindexmod.rs

253 lines · 8.1 KB · Rust

Raw Blame History Download
1//! Repository discovery and the in-memory repository index.
2//!
3//! Readers get a consistent snapshot via [`Indexer::load`] without locking.
4//! Writers (full scans and single-repo refreshes) are serialized and swap in
5//! a new snapshot copy-on-write.
6
7pub mod meta;
8pub mod scan;
9pub mod watch;
10
11use std::collections::{BTreeMap, BTreeSet};
12use std::path::{Path, PathBuf};
13use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
14use std::sync::{Arc, Mutex};
15use std::time::{Duration, Instant};
16
17use arc_swap::ArcSwap;
18
19use crate::config::Config;
20use crate::git::ThreadSafeRepository;
21use meta::{Inspected, RepoMeta};
22use scan::Candidate;
23
24pub struct Repo {
25 pub name: String,
26 pub git_dir: PathBuf,
27 pub meta: RepoMeta,
28 pub gix: ThreadSafeRepository,
29 /// Changes whenever this repo's refs or metadata change; part of cache
30 /// keys for ref-dependent data.
31 pub generation: u64,
32}
33
34#[derive(Default, Clone)]
35pub struct RepoIndex {
36 /// Visible repositories by name.
37 pub repos: BTreeMap<String, Arc<Repo>>,
38 /// Every discovered git dir (hidden ones included) → name.
39 pub git_dirs: BTreeMap<PathBuf, String>,
40 pub hidden: BTreeSet<String>,
41 /// Non-repository directories walked during discovery.
42 pub dirs: Vec<PathBuf>,
43 pub warnings: Vec<String>,
44}
45
46#[derive(Debug, Default, PartialEq, Eq)]
47pub struct ScanStats {
48 pub total: usize,
49 pub hidden: usize,
50 pub added: usize,
51 pub removed: usize,
52 pub changed: usize,
53 pub elapsed: Duration,
54}
55
56impl ScanStats {
57 pub fn any_change(&self) -> bool {
58 self.added + self.removed + self.changed > 0
59 }
60}
61
62#[derive(Debug, PartialEq, Eq)]
63pub enum Refresh {
64 Unchanged,
65 Updated(String),
66 Hidden(String),
67 /// The path is not (or no longer) a known repository: do a full scan.
68 NeedsRescan,
69 Failed(String),
70}
71
72pub struct Indexer {
73 current: ArcSwap<RepoIndex>,
74 generation: AtomicU64,
75 write: Mutex<()>,
76}
77
78impl Default for Indexer {
79 fn default() -> Self {
80 Self::new()
81 }
82}
83
84impl Indexer {
85 pub fn new() -> Self {
86 Indexer {
87 current: ArcSwap::from_pointee(RepoIndex::default()),
88 generation: AtomicU64::new(0),
89 write: Mutex::new(()),
90 }
91 }
92
93 pub fn load(&self) -> Arc<RepoIndex> {
94 self.current.load_full()
95 }
96
97 /// Rediscover everything under the scan path and rebuild the index.
98 pub fn full_scan(&self, config: &Config) -> ScanStats {
99 let _guard = self.write.lock().unwrap_or_else(|e| e.into_inner());
100 let start = Instant::now();
101 let old = self.current.load_full();
102 let discovery = scan::discover(
103 &config.repos.scan_path,
104 config.repos.max_depth,
105 config.repos.follow_symlinks,
106 );
107
108 let mut new = RepoIndex {
109 dirs: discovery.dirs,
110 warnings: discovery.warnings,
111 ..RepoIndex::default()
112 };
113 let inspected = inspect_all(config, &discovery.candidates);
114 for (candidate, result) in discovery.candidates.into_iter().zip(inspected) {
115 new.git_dirs
116 .insert(candidate.git_dir.clone(), candidate.name.clone());
117 match result {
118 Err(e) => new.warnings.push(format!("{}: {e:#}", candidate.name)),
119 Ok(inspected) => {
120 new.warnings.extend(inspected.warnings.iter().cloned());
121 if inspected.hidden {
122 new.hidden.insert(candidate.name);
123 } else {
124 let repo = self.make_repo(&old, candidate, inspected);
125 new.repos.insert(repo.name.clone(), repo);
126 }
127 }
128 }
129 }
130 let known: BTreeSet<&String> = new.git_dirs.values().collect();
131 for o in &config.repo_overrides {
132 if !known.contains(&o.name) {
133 new.warnings.push(format!(
134 "[[repo]] {:?} does not match any repository",
135 o.name
136 ));
137 }
138 }
139
140 let mut stats = ScanStats {
141 total: new.repos.len(),
142 hidden: new.hidden.len(),
143 ..ScanStats::default()
144 };
145 for (name, repo) in &new.repos {
146 match old.repos.get(name) {
147 None => stats.added += 1,
148 Some(prev) if !Arc::ptr_eq(prev, repo) => stats.changed += 1,
149 Some(_) => {}
150 }
151 }
152 stats.removed = old
153 .repos
154 .keys()
155 .filter(|name| !new.repos.contains_key(*name))
156 .count();
157 self.current.store(Arc::new(new));
158 stats.elapsed = start.elapsed();
159 stats
160 }
161
162 /// Re-inspect the single repository at `git_dir` (after a ref change).
163 pub fn refresh(&self, config: &Config, git_dir: &Path) -> Refresh {
164 let _guard = self.write.lock().unwrap_or_else(|e| e.into_inner());
165 let old = self.current.load_full();
166 let Some(name) = old.git_dirs.get(git_dir).cloned() else {
167 return Refresh::NeedsRescan;
168 };
169 if scan::repo_git_dir(git_dir).as_deref() != Some(git_dir) {
170 return Refresh::NeedsRescan;
171 }
172 let candidate = Candidate {
173 name: name.clone(),
174 git_dir: git_dir.to_path_buf(),
175 };
176 let inspected = match meta::inspect(config, &candidate) {
177 Ok(inspected) => inspected,
178 // Keep serving the previous state; the next event or poll retries.
179 Err(e) => return Refresh::Failed(format!("{name}: {e:#}")),
180 };
181
182 let mut new = (*old).clone();
183 if inspected.hidden {
184 if new.repos.remove(&name).is_none() {
185 return Refresh::Unchanged;
186 }
187 new.hidden.insert(name.clone());
188 self.current.store(Arc::new(new));
189 return Refresh::Hidden(name);
190 }
191 let repo = self.make_repo(&old, candidate, inspected);
192 if old
193 .repos
194 .get(&name)
195 .is_some_and(|prev| Arc::ptr_eq(prev, &repo))
196 {
197 return Refresh::Unchanged;
198 }
199 new.hidden.remove(&name);
200 new.repos.insert(name.clone(), repo);
201 self.current.store(Arc::new(new));
202 Refresh::Updated(name)
203 }
204
205 /// Reuse the previous entry when nothing changed, so its generation (and
206 /// everything cached under it) stays valid.
207 fn make_repo(&self, old: &RepoIndex, candidate: Candidate, inspected: Inspected) -> Arc<Repo> {
208 if let Some(prev) = old.repos.get(&candidate.name)
209 && prev.git_dir == candidate.git_dir
210 && prev.meta == inspected.meta
211 {
212 return prev.clone();
213 }
214 Arc::new(Repo {
215 name: candidate.name,
216 git_dir: candidate.git_dir,
217 meta: inspected.meta,
218 gix: inspected.gix,
219 generation: self.generation.fetch_add(1, Ordering::Relaxed) + 1,
220 })
221 }
222}
223
224/// Inspect candidates on a few threads; results keep the input order.
225fn inspect_all(config: &Config, candidates: &[Candidate]) -> Vec<anyhow::Result<Inspected>> {
226 let workers = std::thread::available_parallelism()
227 .map_or(4, |n| n.get())
228 .min(candidates.len())
229 .max(1);
230 let next = AtomicUsize::new(0);
231 let mut results: Vec<(usize, anyhow::Result<Inspected>)> = std::thread::scope(|s| {
232 let handles: Vec<_> = (0..workers)
233 .map(|_| {
234 s.spawn(|| {
235 let mut out = Vec::new();
236 loop {
237 let i = next.fetch_add(1, Ordering::Relaxed);
238 let Some(candidate) = candidates.get(i) else {
239 break out;
240 };
241 out.push((i, meta::inspect(config, candidate)));
242 }
243 })
244 })
245 .collect();
246 handles
247 .into_iter()
248 .flat_map(|h| h.join().expect("inspect worker panicked"))
249 .collect()
250 });
251 results.sort_by_key(|(i, _)| *i);
252 results.into_iter().map(|(_, r)| r).collect()
253}