alice.pds.demo.boltless.dev
This repository has no description
11 kB
337 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::path::{Path, PathBuf};
2use std::sync::atomic::{AtomicUsize, Ordering};
3
4use knot_git::{GitError, Layout, Repo};
5use knot_types::{ObjectFormat, RepoDid};
6
7use crate::mapping::AdoptRepo;
8use crate::source::SourceRepoDid;
9
10#[derive(Debug, thiserror::Error)]
11pub enum AdoptError {
12 #[error("layout path for {repo}: {source}")]
13 Layout { repo: RepoDid, source: GitError },
14 #[error("repo {repo} resolves to the reserved knot meta-repo path")]
15 ReservesMeta { repo: RepoDid },
16 #[error("place {path}: {source}")]
17 Place {
18 path: PathBuf,
19 source: std::io::Error,
20 },
21 #[error("sync {path}: {source}")]
22 Sync {
23 path: PathBuf,
24 source: std::io::Error,
25 },
26 #[error("adopted repo {repo} doesn't open as a git repository: {source}")]
27 Unopenable { repo: RepoDid, source: GitError },
28 #[error("source directory for {repo} vanished between mapping and adoption")]
29 Vanished { repo: RepoDid },
30 #[error("consuming the source needs {scan_path} and {target} on one filesystem")]
31 CrossDeviceConsume { scan_path: PathBuf, target: PathBuf },
32}
33
34#[derive(Debug, PartialEq, Eq)]
35pub struct AdoptOutcome {
36 pub transfer: Transfer,
37 pub adopted: u64,
38 pub already_present: u64,
39 pub sha1: u64,
40 pub sha256: u64,
41}
42
43impl AdoptOutcome {
44 fn empty(transfer: Transfer) -> Self {
45 Self {
46 transfer,
47 adopted: 0,
48 already_present: 0,
49 sha1: 0,
50 sha256: 0,
51 }
52 }
53
54 fn merge(self, other: Self) -> Self {
55 Self {
56 transfer: self.transfer,
57 adopted: self.adopted + other.adopted,
58 already_present: self.already_present + other.already_present,
59 sha1: self.sha1 + other.sha1,
60 sha256: self.sha256 + other.sha256,
61 }
62 }
63}
64
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum SourcePolicy {
67 Preserve,
68 Consume,
69}
70
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72pub enum Transfer {
73 Rename,
74 Copy,
75}
76
77impl std::fmt::Display for Transfer {
78 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79 match self {
80 Self::Rename => f.write_str("rename"),
81 Self::Copy => f.write_str("copy"),
82 }
83 }
84}
85
86enum Placement {
87 AlreadyPresent,
88 Staged(PathBuf),
89 Moved,
90}
91
92struct Staged<'repo> {
93 did: &'repo RepoDid,
94 destination: PathBuf,
95 placement: Placement,
96}
97
98pub fn source_dir(source_root: &Path, repo_did: &SourceRepoDid) -> PathBuf {
99 source_root.join(repo_did.as_str())
100}
101
102pub fn source_is_repo(source_root: &Path, repo_did: &SourceRepoDid) -> bool {
103 source_dir(source_root, repo_did).join("HEAD").is_file()
104}
105
106pub fn adopt_all(
107 layout: &Layout,
108 source_root: &Path,
109 repos: &[AdoptRepo],
110 policy: SourcePolicy,
111) -> Result<AdoptOutcome, AdoptError> {
112 let root = layout.scratch_dir();
113 std::fs::create_dir_all(root).map_err(|source| AdoptError::Place {
114 path: root.to_path_buf(),
115 source,
116 })?;
117 let transfer = transfer_mode(source_root, root, policy)?;
118 let staged = in_lanes(repos, |repo| stage_one(layout, source_root, repo, transfer))?;
119 sync_filesystem(root)?;
120 staged.iter().try_for_each(commit_one)?;
121 sync_filesystem(root)?;
122 in_lanes(&staged, |staged| count_one(staged, transfer)).map(|counted| {
123 counted
124 .into_iter()
125 .fold(AdoptOutcome::empty(transfer), AdoptOutcome::merge)
126 })
127}
128
129fn transfer_mode(
130 source_root: &Path,
131 target_root: &Path,
132 policy: SourcePolicy,
133) -> Result<Transfer, AdoptError> {
134 use std::os::unix::fs::MetadataExt;
135 let device = |path: &Path| {
136 std::fs::metadata(path)
137 .map(|meta| meta.dev())
138 .map_err(|source| AdoptError::Place {
139 path: path.to_path_buf(),
140 source,
141 })
142 };
143 let one_filesystem = device(source_root)? == device(target_root)?;
144 match (policy, one_filesystem) {
145 (SourcePolicy::Consume, true) => Ok(Transfer::Rename),
146 (SourcePolicy::Consume, false) => Err(AdoptError::CrossDeviceConsume {
147 scan_path: source_root.to_path_buf(),
148 target: target_root.to_path_buf(),
149 }),
150 (SourcePolicy::Preserve, _) => Ok(Transfer::Copy),
151 }
152}
153
154fn in_lanes<'items, T, R, F>(items: &'items [T], work: F) -> Result<Vec<R>, AdoptError>
155where
156 T: Sync,
157 R: Send,
158 F: Fn(&'items T) -> Result<R, AdoptError> + Sync,
159{
160 let cursor = AtomicUsize::new(0);
161 std::thread::scope(|scope| {
162 (0..lane_count(items.len()))
163 .map(|_| {
164 scope.spawn(|| {
165 std::iter::from_fn(|| items.get(cursor.fetch_add(1, Ordering::Relaxed)))
166 .map(&work)
167 .collect::<Result<Vec<R>, AdoptError>>()
168 })
169 })
170 .collect::<Vec<_>>()
171 .into_iter()
172 .map(|lane| lane.join().expect("adoption lane doesn't panic"))
173 .collect::<Result<Vec<Vec<R>>, AdoptError>>()
174 .map(|lanes| lanes.into_iter().flatten().collect())
175 })
176}
177
178fn lane_count(items: usize) -> usize {
179 std::thread::available_parallelism()
180 .map(std::num::NonZeroUsize::get)
181 .unwrap_or(1)
182 .min(items.max(1))
183}
184
185fn stage_one<'repo>(
186 layout: &Layout,
187 source_root: &Path,
188 repo: &'repo AdoptRepo,
189 transfer: Transfer,
190) -> Result<Staged<'repo>, AdoptError> {
191 let destination = layout
192 .guarded_path(&repo.did)
193 .map_err(|source| match source {
194 GitError::ReservedDid(_) => AdoptError::ReservesMeta {
195 repo: repo.did.clone(),
196 },
197 source => AdoptError::Layout {
198 repo: repo.did.clone(),
199 source,
200 },
201 })?;
202 match destination.exists() {
203 true => Ok(Staged {
204 did: &repo.did,
205 destination,
206 placement: Placement::AlreadyPresent,
207 }),
208 false => {
209 let source = source_dir(source_root, &repo.source_did);
210 match source.is_dir() {
211 false => Err(AdoptError::Vanished {
212 repo: repo.did.clone(),
213 }),
214 true => place(&source, &destination, transfer).map(|placement| Staged {
215 did: &repo.did,
216 destination,
217 placement,
218 }),
219 }
220 }
221 }
222}
223
224fn place(source: &Path, destination: &Path, transfer: Transfer) -> Result<Placement, AdoptError> {
225 match transfer {
226 Transfer::Rename => {
227 make_parent(destination)?;
228 std::fs::rename(source, destination)
229 .map(|()| Placement::Moved)
230 .map_err(|error| AdoptError::Place {
231 path: destination.to_path_buf(),
232 source: error,
233 })
234 }
235 Transfer::Copy => stage_tree(source, destination).map(Placement::Staged),
236 }
237}
238
239fn commit_one(staged: &Staged<'_>) -> Result<(), AdoptError> {
240 match &staged.placement {
241 Placement::AlreadyPresent | Placement::Moved => Ok(()),
242 Placement::Staged(staging) => {
243 std::fs::rename(staging, &staged.destination).map_err(|source| AdoptError::Place {
244 path: staged.destination.clone(),
245 source,
246 })
247 }
248 }
249}
250
251fn count_one(staged: &Staged<'_>, transfer: Transfer) -> Result<AdoptOutcome, AdoptError> {
252 let opened = Repo::open(&staged.destination).map_err(|source| AdoptError::Unopenable {
253 repo: staged.did.clone(),
254 source,
255 })?;
256 let fresh = u64::from(!matches!(staged.placement, Placement::AlreadyPresent));
257 let counted = AdoptOutcome {
258 adopted: fresh,
259 already_present: 1 - fresh,
260 ..AdoptOutcome::empty(transfer)
261 };
262 Ok(match opened.object_format() {
263 ObjectFormat::SHA1 => AdoptOutcome { sha1: 1, ..counted },
264 _ => AdoptOutcome {
265 sha256: 1,
266 ..counted
267 },
268 })
269}
270
271fn make_parent(destination: &Path) -> Result<&Path, AdoptError> {
272 let parent = destination
273 .parent()
274 .expect("layout repo paths always have a parent");
275 std::fs::create_dir_all(parent)
276 .map(|()| parent)
277 .map_err(|source| AdoptError::Place {
278 path: parent.to_path_buf(),
279 source,
280 })
281}
282
283fn stage_tree(source: &Path, destination: &Path) -> Result<PathBuf, AdoptError> {
284 let io = |path: &Path| {
285 let path = path.to_path_buf();
286 move |source: std::io::Error| AdoptError::Place { path, source }
287 };
288 let parent = make_parent(destination)?;
289 let staging = parent.join(format!(
290 ".migrate-staging.{}",
291 destination
292 .file_name()
293 .expect("layout repo paths always have a file name")
294 .to_string_lossy()
295 ));
296 if staging.exists() {
297 std::fs::remove_dir_all(&staging).map_err(io(&staging))?;
298 }
299 place_tree(source, &staging).map(|()| staging)
300}
301
302fn place_tree(source: &Path, destination: &Path) -> Result<(), AdoptError> {
303 walkdir::WalkDir::new(source)
304 .into_iter()
305 .try_for_each(|entry| {
306 let entry = entry.map_err(|error| AdoptError::Place {
307 path: source.to_path_buf(),
308 source: error.into(),
309 })?;
310 let relative = entry
311 .path()
312 .strip_prefix(source)
313 .expect("walkdir yields paths under its root");
314 let target = destination.join(relative);
315 let io = |source: std::io::Error| AdoptError::Place {
316 path: entry.path().to_path_buf(),
317 source,
318 };
319 match entry.file_type() {
320 kind if kind.is_dir() => std::fs::create_dir_all(&target).map_err(io),
321 #[cfg(unix)]
322 kind if kind.is_symlink() => std::fs::read_link(entry.path())
323 .and_then(|link| std::os::unix::fs::symlink(link, &target))
324 .map_err(io),
325 _ => std::fs::copy(entry.path(), &target).map(|_| ()).map_err(io),
326 }
327 })
328}
329
330fn sync_filesystem(root: &Path) -> Result<(), AdoptError> {
331 std::fs::File::open(root)
332 .and_then(|anchor| rustix::fs::syncfs(&anchor).map_err(std::io::Error::from))
333 .map_err(|source| AdoptError::Sync {
334 path: root.to_path_buf(),
335 source,
336 })
337}