alice.pds.demo.boltless.dev
This repository has no description
12 kB
347 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::fs::File;
2use std::io::{self, Write};
3use std::path::{Path, PathBuf};
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::time::{Duration, SystemTime};
6
7const STAGING_INFIX: &str = ".knot-tmp.";
8
9// A staging file belongs to whoever is filling it, and another process filling
10// the same target is normal for a knot sharing a volume with a migrate
11// or maintenance run. Only reclaim one old enough that no live writer could
12// still own it.
13const STAGING_REAP_AFTER: Duration = Duration::from_secs(3600);
14
15#[derive(Debug)]
16pub struct FsError {
17 pub path: PathBuf,
18 pub source: io::Error,
19}
20
21impl FsError {
22 fn at(path: &Path, source: io::Error) -> Self {
23 Self {
24 path: path.to_path_buf(),
25 source,
26 }
27 }
28}
29
30impl std::fmt::Display for FsError {
31 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32 write!(f, "{}: {}", self.path.display(), self.source)
33 }
34}
35
36impl std::error::Error for FsError {
37 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
38 Some(&self.source)
39 }
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub enum FileMode {
44 Inherited,
45 Private,
46}
47
48pub fn staging_nonce() -> String {
49 static COUNTER: AtomicU64 = AtomicU64::new(0);
50 format!(
51 "{}.{}",
52 std::process::id(),
53 COUNTER.fetch_add(1, Ordering::Relaxed)
54 )
55}
56
57fn staging_prefix(name: &str) -> String {
58 format!(".{name}{STAGING_INFIX}")
59}
60
61fn pre_hidden_staging_prefix(name: &str) -> String {
62 format!("{name}{STAGING_INFIX}")
63}
64
65fn remove_matching(
66 dir: &Path,
67 prefixes: &[&str],
68 reclaimable: impl Fn(&std::fs::DirEntry) -> bool,
69) {
70 let Ok(entries) = std::fs::read_dir(dir) else {
71 return;
72 };
73 entries
74 .filter_map(Result::ok)
75 .filter(|entry| {
76 entry
77 .file_name()
78 .to_str()
79 .is_some_and(|name| prefixes.iter().any(|prefix| name.starts_with(prefix)))
80 })
81 .filter(|entry| reclaimable(entry))
82 .for_each(|entry| {
83 let _ = std::fs::remove_file(entry.path());
84 });
85}
86
87pub fn clear_temps(dir: &Path, prefix: &str) {
88 remove_matching(dir, &[prefix], |_| true);
89}
90
91pub fn clear_stale(dir: &Path, prefix: &str) {
92 let now = SystemTime::now();
93 remove_matching(dir, &[prefix], |entry| abandoned(entry, now));
94}
95
96fn abandoned(entry: &std::fs::DirEntry, now: SystemTime) -> bool {
97 entry
98 .metadata()
99 .and_then(|meta| meta.modified())
100 .ok()
101 .and_then(|modified| now.duration_since(modified).ok())
102 .is_some_and(|age| age >= STAGING_REAP_AFTER)
103}
104
105pub fn clear_staging(path: &Path) {
106 let (Some(parent), Some(name)) = (
107 path.parent(),
108 path.file_name().and_then(|name| name.to_str()),
109 ) else {
110 return;
111 };
112 let (hidden, pre_hidden) = (staging_prefix(name), pre_hidden_staging_prefix(name));
113 let now = SystemTime::now();
114 remove_matching(parent, &[&hidden, &pre_hidden], |entry| {
115 abandoned(entry, now)
116 });
117}
118
119pub fn fsync_path(path: &Path) -> Result<(), FsError> {
120 match File::open(path) {
121 Ok(file) => file.sync_all().map_err(|error| FsError::at(path, error)),
122 Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
123 Err(error) => Err(FsError::at(path, error)),
124 }
125}
126
127fn create(path: &Path, mode: FileMode) -> io::Result<File> {
128 let mut options = std::fs::OpenOptions::new();
129 options.write(true).create(true).truncate(true);
130 #[cfg(unix)]
131 if mode == FileMode::Private {
132 use std::os::unix::fs::OpenOptionsExt;
133 options.mode(0o600);
134 }
135 options.open(path)
136}
137
138pub fn atomic_write<E, F>(path: &Path, mode: FileMode, fill: F) -> Result<(), E>
139where
140 F: FnOnce(&mut File) -> Result<(), E>,
141 E: From<FsError>,
142{
143 let (Some(parent), Some(name)) = (
144 path.parent(),
145 path.file_name().and_then(|name| name.to_str()),
146 ) else {
147 return Err(FsError::at(path, io::Error::from(io::ErrorKind::InvalidInput)).into());
148 };
149 clear_staging(path);
150 let staging = parent.join(format!("{}{}", staging_prefix(name), staging_nonce()));
151
152 let outcome = create(&staging, mode)
153 .map_err(|error| E::from(FsError::at(path, error)))
154 .and_then(|mut file| {
155 fill(&mut file)?;
156 file.sync_all()
157 .map_err(|error| E::from(FsError::at(path, error)))
158 })
159 .and_then(|()| {
160 std::fs::rename(&staging, path).map_err(|error| E::from(FsError::at(path, error)))
161 });
162
163 match outcome {
164 Ok(()) => fsync_path(parent).map_err(Into::into),
165 Err(error) => {
166 let _ = std::fs::remove_file(&staging);
167 Err(error)
168 }
169 }
170}
171
172pub fn atomic_write_bytes(path: &Path, contents: &[u8], mode: FileMode) -> Result<(), FsError> {
173 let target = path.to_path_buf();
174 atomic_write(path, mode, move |file| {
175 file.write_all(contents)
176 .map_err(|error| FsError::at(&target, error))
177 })
178}
179
180#[cfg(test)]
181mod tests {
182 use super::*;
183
184 fn names_in(dir: &Path) -> Vec<String> {
185 std::fs::read_dir(dir)
186 .unwrap()
187 .filter_map(Result::ok)
188 .filter_map(|entry| entry.file_name().to_str().map(str::to_string))
189 .collect()
190 }
191
192 fn age(path: &Path, by: Duration) {
193 let when = SystemTime::now() - by;
194 File::options()
195 .write(true)
196 .open(path)
197 .unwrap()
198 .set_times(std::fs::FileTimes::new().set_modified(when))
199 .unwrap();
200 }
201
202 #[test]
203 fn a_write_stages_under_a_hidden_name_and_leaves_only_the_target_behind() {
204 let dir = tempfile::tempdir().unwrap();
205 let target = dir.path().join("keys.sealed");
206 let sibling = dir.path().join("keys.sealed.v2");
207 let staging = std::sync::Mutex::new(Vec::new());
208 atomic_write::<FsError, _>(&target, FileMode::Private, |file| {
209 *staging.lock().unwrap() = names_in(dir.path());
210 file.write_all(b"first")
211 .map_err(|error| FsError::at(&target, error))
212 })
213 .unwrap();
214 atomic_write_bytes(&target, b"second", FileMode::Private).unwrap();
215 atomic_write_bytes(&sibling, b"two", FileMode::Inherited).unwrap();
216
217 let staging = staging.into_inner().unwrap();
218 assert_eq!(
219 staging
220 .iter()
221 .filter(|name| name.starts_with("keys.sealed"))
222 .count(),
223 0,
224 "a tool matching on the target's own prefix mustn't find the half-written staging file"
225 );
226 assert!(
227 staging
228 .iter()
229 .any(|name| name.starts_with(".keys.sealed.knot-tmp.")),
230 "saw {staging:?}"
231 );
232 assert_eq!(std::fs::read(&target).unwrap(), b"second");
233 assert_eq!(
234 std::fs::read(&sibling).unwrap(),
235 b"two",
236 "a name that extends another mustn't share its staging path"
237 );
238 let left = names_in(dir.path());
239 assert_eq!(left.len(), 2, "left staging files behind: {left:?}");
240 #[cfg(unix)]
241 {
242 use std::os::unix::fs::PermissionsExt;
243 let mode = std::fs::metadata(&target).unwrap().permissions().mode();
244 assert_eq!(mode & 0o777, 0o600, "sealed material stays owner-only");
245 }
246 assert!(fsync_path(&dir.path().join("never-written")).is_ok());
247 }
248
249 #[test]
250 fn a_write_that_fails_or_overlaps_another_keeps_what_is_already_stored() {
251 let dir = tempfile::tempdir().unwrap();
252 let target = dir.path().join("packed-refs");
253 atomic_write_bytes(&target, b"kept", FileMode::Inherited).unwrap();
254 let failed: Result<(), FsError> = atomic_write(&target, FileMode::Inherited, |_| {
255 Err(FsError::at(&target, io::Error::other("fill failed")))
256 });
257 assert!(failed.is_err());
258 assert_eq!(
259 std::fs::read(&target).unwrap(),
260 b"kept",
261 "a failed write mustn't destroy what was already stored"
262 );
263 assert_eq!(
264 names_in(dir.path()).len(),
265 1,
266 "a failed write removes its staging file"
267 );
268
269 let overlapping: Result<(), FsError> = atomic_write(&target, FileMode::Inherited, |file| {
270 atomic_write_bytes(
271 &target,
272 b"the second writer's contents",
273 FileMode::Inherited,
274 )?;
275 file.write_all(b"first writer finishes after")
276 .map_err(|error| FsError::at(&target, error))
277 });
278 assert!(
279 overlapping.is_ok(),
280 "an overlapping write mustn't delete the staging file this one is filling: \
281 {overlapping:?}"
282 );
283 assert_eq!(
284 std::fs::read(&target).unwrap(),
285 b"first writer finishes after"
286 );
287 }
288
289 #[test]
290 fn a_write_reclaims_aged_staging_files_of_either_naming_and_spares_a_fresh_one() {
291 let dir = tempfile::tempdir().unwrap();
292 let target = dir.path().join("packed-refs");
293 let crashed = dir.path().join(".packed-refs.knot-tmp.4242.7");
294 let pre_hidden = dir.path().join("packed-refs.knot-tmp.4242.8");
295 let in_flight = dir.path().join(".packed-refs.knot-tmp.4243.0");
296 std::fs::write(&crashed, b"left by a crashed run").unwrap();
297 std::fs::write(&pre_hidden, b"left beside the target by an older build").unwrap();
298 std::fs::write(&in_flight, b"another process is mid-write").unwrap();
299 let aged = STAGING_REAP_AFTER + Duration::from_secs(60);
300 age(&crashed, aged);
301 age(&pre_hidden, aged);
302
303 atomic_write_bytes(&target, b"fresh", FileMode::Inherited).unwrap();
304
305 assert!(
306 !crashed.exists(),
307 "the write reclaims a crashed run's staging file"
308 );
309 assert!(
310 !pre_hidden.exists(),
311 "moving staging under a dot mustn't strand the temps the previous naming left"
312 );
313 assert!(
314 in_flight.exists(),
315 "a second knot on the same volume is a supported deployment \
316 whose live staging file this write mustn't sweep out from under its rename"
317 );
318 }
319
320 #[test]
321 fn a_prefix_sweep_spares_a_live_file_only_where_another_writer_could_own_it() {
322 let dir = tempfile::tempdir().unwrap();
323 let crashed = dir.path().join(".knot-repack.4242.7.pack");
324 let in_flight = dir.path().join(".knot-repack.4243.0.pack");
325 let installed = dir.path().join("multi-pack-index");
326 let bitmap = dir.path().join("multi-pack-index-abc.bitmap");
327 std::fs::write(&crashed, b"left by a crashed run").unwrap();
328 std::fs::write(&in_flight, b"another process is streaming into this").unwrap();
329 std::fs::write(&installed, b"index").unwrap();
330 std::fs::write(&bitmap, b"bitmap").unwrap();
331 age(&crashed, STAGING_REAP_AFTER + Duration::from_secs(60));
332
333 clear_stale(dir.path(), ".knot-repack.");
334 clear_temps(dir.path(), "multi-pack-index");
335
336 assert!(!crashed.exists());
337 assert!(
338 in_flight.exists(),
339 "deleting the staging pack another repack is streaming into fails that run's rename"
340 );
341 assert!(!installed.exists());
342 assert!(
343 !bitmap.exists(),
344 "removing the index without its bitmap would leave a bitmap describing an index that is gone"
345 );
346 }
347}