steid
Sign in

@jamesgill /

steid

steid/src/application/reconcile.rs
Rust13.3 KBCode·Blame·History·Raw
1//! Reconciling the data directory with the rows that describe it.
2//!
3//! A repository is two writes that cannot share a transaction — a row and a directory —
4//! so a process that dies between them leaves one without the other. A row without a
5//! directory is visible: every page it serves fails loudly. **A directory without a row
6//! is invisible**, and worse than invisible: it silently blocks re-creating a repository
7//! by that name, because `init_bare` refuses to adopt what is already there and a rename
8//! refuses to move onto it.
9//!
10//! There are three ways to get one, and by wave 2 all three had been written down as
11//! open — see `plans/architecture.md` on DB-plus-filesystem writes:
12//!
13//! - a **create** that died between `init_bare` and the row insert;
14//! - a **rename** that died between the directory move and the row update, which strands
15//! the *new* name;
16//! - an **import** whose repository was deleted while the clone was still running, which
17//! `run_import` tries to clean up and cannot if it is the process that died.
18//!
19//! The sweep runs on boot, beside the interrupted-import and reflog sweeps, and for the
20//! same reason: nothing has been spawned yet, so a directory with no row is genuinely
21//! finished rather than half-written by a task still running.
22
23use std::{
24 collections::HashSet,
25 path::PathBuf,
26 time::{SystemTime, UNIX_EPOCH},
27};
28
29use crate::domain::{
30 OrgName, RepoName,
31 repository::{OrgRepository, RepoRepository},
32};
33
34use super::{
35 error::Result,
36 port::{GitStorage, RepoDir},
37};
38
39/// Which of the directories found on disk have no repository row.
40///
41/// Pure, and the whole of the decision — so what counts as an orphan is testable without
42/// a disk, a database, or a clock. `known` is every `{handle}/{name}` pair the rows
43/// describe, however it was gathered.
44///
45/// Comparison is on the pair, not on the path: a directory is an orphan when nothing
46/// claims that *name*, which is exactly the condition that blocks re-using it. A
47/// directory whose handle or name would not parse can never be claimed by a row, so it
48/// falls out as an orphan without needing a rule of its own.
49pub fn orphan_dirs(found: &[RepoDir], known: &HashSet<(String, String)>) -> Vec<RepoDir> {
50 found
51 .iter()
52 .filter(|dir| !known.contains(&(dir.handle.clone(), dir.name.clone())))
53 .cloned()
54 .collect()
55}
56
57/// Where an orphan is moved to, under the quarantine directory.
58///
59/// `<seconds since the epoch>-<name>`: sortable, unique enough for a sweep that runs once
60/// per boot, and it keeps the original name visible so an operator can see what they are
61/// looking at. Seconds rather than a formatted date because Steid has no date formatter
62/// and a filename is the last place to acquire a timezone.
63///
64/// The name is *not* sanitised — it came off the filesystem, so it is already a single
65/// legal component, and re-deriving it through [`RepoName`] would refuse exactly the
66/// weird ones this exists to preserve.
67pub fn quarantine_label(name: &str, now: SystemTime) -> String {
68 let seconds = now
69 .duration_since(UNIX_EPOCH)
70 .map(|elapsed| elapsed.as_secs())
71 .unwrap_or(0);
72
73 format!("{seconds}-{name}")
74}
75
76/// Moves every repository directory with no row aside, and says where each went.
77///
78/// **Moved, never deleted.** Every way of producing an orphan is a write that was
79/// interrupted, and an interrupted write can still have left the only copy of somebody's
80/// objects. Moving frees the name — which is the actual complaint — and keeps what was
81/// there for whoever wants to look.
82///
83/// Takes no [`Actor`](crate::domain::Actor), like the other boot sweeps: it runs before
84/// the first request, so there is nobody to be.
85///
86/// The rows are read one handle at a time rather than all at once, because the handles
87/// worth asking about are exactly the ones that have a directory. A handle with a
88/// directory and no organisation contributes nothing to `known`, so everything under it
89/// is an orphan — which is right: no row can ever claim those names.
90pub async fn sweep_orphan_directories(
91 now: SystemTime,
92 orgs: &impl OrgRepository,
93 repos: &impl RepoRepository,
94 storage: &impl GitStorage,
95) -> Result<Vec<PathBuf>> {
96 let found = storage.list_repo_dirs().await?;
97
98 let mut known = HashSet::new();
99
100 for handle in found
101 .iter()
102 .map(|dir| dir.handle.clone())
103 .collect::<HashSet<_>>()
104 {
105 // A directory name that is not a legal handle cannot have an organisation, so
106 // there is nothing to ask for.
107 let Ok(name) = OrgName::new(&handle) else {
108 continue;
109 };
110
111 let Some(org) = orgs.find_by_name(&name).await? else {
112 continue;
113 };
114
115 for repo in repos.list_by_org(&org.id).await? {
116 known.insert((handle.clone(), repo.name.as_str().to_owned()));
117 }
118 }
119
120 let mut moved = Vec::new();
121
122 for orphan in orphan_dirs(&found, &known) {
123 let label = quarantine_label(&orphan.name, now);
124
125 match storage.quarantine(&orphan, &label).await {
126 Ok(destination) => {
127 // Logged with its path, because this is the only trace the operation
128 // leaves and the whole point is that an orphan was invisible before.
129 println!(
130 "steid: {} has no repository row; moved to {}",
131 orphan.path.display(),
132 destination.display()
133 );
134 moved.push(destination);
135 }
136 // Not fatal, and not worth abandoning the rest of the sweep for: a directory
137 // that would not move is one whose name stays blocked, which is where it
138 // already was.
139 Err(error) => eprintln!(
140 "steid: {} has no repository row and could not be moved aside: {error}",
141 orphan.path.display()
142 ),
143 }
144 }
145
146 Ok(moved)
147}
148
149/// The pair a [`RepoDir`] is matched on, for a caller building `known` by hand.
150pub fn claimed(handle: &OrgName, name: &RepoName) -> (String, String) {
151 (handle.as_str().to_owned(), name.as_str().to_owned())
152}
153
154#[cfg(test)]
155mod tests {
156 use super::*;
157 use crate::{
158 domain::{OrgId, Organization, RepoId, Repository, Visibility},
159 infrastructure::{
160 git::InMemoryGitStorage,
161 repository::{InMemoryOrgRepo, InMemoryRepoRepo},
162 },
163 };
164
165 fn dir(handle: &str, name: &str) -> RepoDir {
166 RepoDir {
167 handle: handle.to_owned(),
168 name: name.to_owned(),
169 path: PathBuf::from(format!("/data/{handle}/{name}.git")),
170 }
171 }
172
173 fn known(pairs: &[(&str, &str)]) -> HashSet<(String, String)> {
174 pairs
175 .iter()
176 .map(|(handle, name)| ((*handle).to_owned(), (*name).to_owned()))
177 .collect()
178 }
179
180 // --- the decision -------------------------------------------------------------
181
182 #[test]
183 fn a_directory_with_a_row_is_left_alone() {
184 let found = [dir("acme", "steid"), dir("acme", "notes")];
185
186 assert!(orphan_dirs(&found, &known(&[("acme", "steid"), ("acme", "notes")])).is_empty());
187 }
188
189 #[test]
190 fn a_directory_with_no_row_is_an_orphan() {
191 let found = [dir("acme", "steid"), dir("acme", "half-made")];
192
193 assert_eq!(
194 orphan_dirs(&found, &known(&[("acme", "steid")])),
195 vec![dir("acme", "half-made")]
196 );
197 }
198
199 #[test]
200 fn the_same_name_under_a_different_handle_does_not_claim_it() {
201 // The pair is what is matched on. A row for `other/steid` says nothing about
202 // whether `acme/steid` may be re-created.
203 let found = [dir("acme", "steid")];
204
205 assert_eq!(
206 orphan_dirs(&found, &known(&[("other", "steid")])),
207 vec![dir("acme", "steid")]
208 );
209 }
210
211 #[test]
212 fn a_rename_that_died_strands_the_new_name_and_not_the_old_one() {
213 // The move landed and the row did not, so the row still says `before` and the
214 // directory is called `after`.
215 let found = [dir("acme", "after")];
216
217 assert_eq!(
218 orphan_dirs(&found, &known(&[("acme", "before")])),
219 vec![dir("acme", "after")]
220 );
221 }
222
223 #[test]
224 fn a_directory_no_row_could_ever_claim_falls_out_as_an_orphan() {
225 // `Not A Handle` is not a legal `OrgName`, so nothing can be recorded against it
226 // and every name under it is unreachable.
227 let found = [dir("Not A Handle", "thing")];
228
229 assert_eq!(
230 orphan_dirs(&found, &known(&[("acme", "thing")])),
231 vec![dir("Not A Handle", "thing")]
232 );
233 }
234
235 #[test]
236 fn nothing_on_disk_is_nothing_to_do() {
237 assert!(orphan_dirs(&[], &known(&[("acme", "steid")])).is_empty());
238 }
239
240 #[test]
241 fn a_label_is_sortable_and_keeps_the_name_it_came_from() {
242 let at = UNIX_EPOCH + std::time::Duration::from_secs(1_757_000_000);
243
244 assert_eq!(quarantine_label("steid", at), "1757000000-steid");
245 // Whatever the filesystem had, kept: this is not a `RepoName` being re-derived.
246 assert_eq!(quarantine_label("Odd Name", at), "1757000000-Odd Name");
247 }
248
249 // --- against a storage adapter --------------------------------------------------
250
251 async fn fixture() -> (InMemoryOrgRepo, InMemoryRepoRepo, InMemoryGitStorage, OrgId) {
252 let orgs = InMemoryOrgRepo::new();
253 let repos = InMemoryRepoRepo::new();
254 let storage = InMemoryGitStorage::new();
255
256 let org = Organization::new(OrgId::generate(), "acme", None).expect("valid org");
257 orgs.save(&org).await.expect("save org");
258
259 (orgs, repos, storage, org.id)
260 }
261
262 async fn make(
263 repos: &InMemoryRepoRepo,
264 storage: &InMemoryGitStorage,
265 org_id: &OrgId,
266 handle: &OrgName,
267 name: &str,
268 ) -> RepoName {
269 let repo = Repository::new(
270 RepoId::generate(),
271 org_id.clone(),
272 name,
273 None,
274 Visibility::Public,
275 SystemTime::UNIX_EPOCH,
276 )
277 .expect("valid repo");
278
279 repos.save(&repo).await.expect("save repo");
280 storage
281 .init_bare(handle, &repo.name)
282 .await
283 .expect("init bare");
284
285 repo.name
286 }
287
288 #[tokio::test]
289 async fn the_sweep_moves_the_orphan_and_leaves_the_rest() {
290 let (orgs, repos, storage, org_id) = fixture().await;
291 let handle = OrgName::new("acme").expect("valid handle");
292
293 let kept = make(&repos, &storage, &org_id, &handle, "steid").await;
294
295 // A create that died: the directory landed, the row never did.
296 let stranded = RepoName::new("half-made").expect("valid name");
297 storage
298 .init_bare(&handle, &stranded)
299 .await
300 .expect("init bare");
301
302 let moved = sweep_orphan_directories(
303 UNIX_EPOCH + std::time::Duration::from_secs(1_757_000_000),
304 &orgs,
305 &repos,
306 &storage,
307 )
308 .await
309 .expect("should sweep");
310
311 assert_eq!(moved.len(), 1);
312 assert!(moved[0].ends_with("1757000000-half-made"), "{:?}", moved[0]);
313
314 assert!(storage.contains(&handle, &kept), "the row's own directory");
315 assert!(
316 !storage.contains(&handle, &stranded),
317 "the name is free to re-create"
318 );
319 }
320
321 #[tokio::test]
322 async fn a_reconciled_directory_can_have_its_name_taken_again() {
323 // The actual complaint an orphan causes: `init_bare` refuses to adopt what is
324 // already there, so the name is unusable until something moves it.
325 let (orgs, repos, storage, org_id) = fixture().await;
326 let handle = OrgName::new("acme").expect("valid handle");
327 let name = RepoName::new("thing").expect("valid name");
328
329 storage.init_bare(&handle, &name).await.expect("init bare");
330 assert!(storage.init_bare(&handle, &name).await.is_err());
331
332 sweep_orphan_directories(SystemTime::UNIX_EPOCH, &orgs, &repos, &storage)
333 .await
334 .expect("should sweep");
335
336 make(&repos, &storage, &org_id, &handle, "thing").await;
337 assert!(storage.contains(&handle, &name));
338 }
339
340 #[tokio::test]
341 async fn a_directory_under_a_handle_with_no_organisation_is_an_orphan() {
342 let (orgs, repos, storage, _org_id) = fixture().await;
343 let nobody = OrgName::new("nobody").expect("valid handle");
344 let name = RepoName::new("thing").expect("valid name");
345
346 storage.init_bare(&nobody, &name).await.expect("init bare");
347
348 assert_eq!(
349 sweep_orphan_directories(SystemTime::UNIX_EPOCH, &orgs, &repos, &storage)
350 .await
351 .expect("should sweep")
352 .len(),
353 1
354 );
355 assert!(!storage.contains(&nobody, &name));
356 }
357
358 #[tokio::test]
359 async fn a_data_directory_that_matches_its_rows_is_swept_and_nothing_moves() {
360 let (orgs, repos, storage, org_id) = fixture().await;
361 let handle = OrgName::new("acme").expect("valid handle");
362
363 make(&repos, &storage, &org_id, &handle, "steid").await;
364 make(&repos, &storage, &org_id, &handle, "notes").await;
365
366 assert!(
367 sweep_orphan_directories(SystemTime::UNIX_EPOCH, &orgs, &repos, &storage)
368 .await
369 .expect("should sweep")
370 .is_empty()
371 );
372 assert_eq!(storage.len(), 2);
373 }
374}