1. 1//! The one way a small durable file is rewritten in place: a read-modify-
2. 2//! write held under one process-wide lock from the read to the rename, and
3. 3//! published by renaming a temporary file no other write shares.
4. 4
5. 5use std::path::{Path, PathBuf};
6. 6use std::sync::Mutex;
7. 7use std::sync::atomic::{AtomicU64, Ordering};
8. 8
9. 9/// Held by every [`update_file`], from the read to the rename. One lock for
10. 10/// every path rather than one per path: a file can be reachable under more
11. 11/// than one spelling, and a per-path lock would first have to agree which
12. 12/// spellings name one file. These writes are rare and small.
13. 13static UPDATE_LOCK: Mutex<()> = Mutex::new(());
14. 14
15. 15#[derive(Debug, thiserror::Error)]
16. 16pub enum UpdateError<E> {
17. 17 #[error("{path}: {source}")]
18. 18 Io {
19. 19 path: PathBuf,
20. 20 #[source]
21. 21 source: std::io::Error,
22. 22 },
23. 23 #[error(transparent)]
24. 24 Edit(E),
25. 25}
26. 26
27. 27/// Read `path` (`None` when it does not exist), let `edit` decide the new
28. 28/// contents, and atomically replace the file with them. `edit` returns
29. 29/// `None` to leave the file untouched.
30. 30///
31. 31/// Holding the lock from the read to the rename is what stops two edits made
32. 32/// at the same moment from each rewriting the file from a read taken before
33. 33/// the other landed. `edit` must preserve whatever it does not own, and must
34. 34/// not call `update_file` itself: the lock is not reentrant.
35. 35pub fn update_file<T, E>(
36. 36 path: &Path,
37. 37 edit: impl FnOnce(Option<&str>) -> Result<(Option<String>, T), E>,
38. 38) -> Result<T, UpdateError<E>> {
39. 39 let _guard = UPDATE_LOCK
40. 40 .lock()
41. 41 .unwrap_or_else(std::sync::PoisonError::into_inner);
42. 42 let current = match std::fs::read_to_string(path) {
43. 43 Ok(text) => Some(text),
44. 44 Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
45. 45 Err(source) => {
46. 46 return Err(UpdateError::Io {
47. 47 path: path.to_path_buf(),
48. 48 source,
49. 49 });
50. 50 }
51. 51 };
52. 52 let (next, outcome) = edit(current.as_deref()).map_err(UpdateError::Edit)?;
53. 53 if let Some(next) = next
54. 54 && current.as_deref() != Some(next.as_str())
55. 55 {
56. 56 replace_file(path, &next).map_err(|source| UpdateError::Io {
57. 57 path: path.to_path_buf(),
58. 58 source,
59. 59 })?;
60. 60 }
61. 61 Ok(outcome)
62. 62}
63. 63
64. 64/// Write `contents` to a new sibling of `path` and rename it over `path`, so
65. 65/// a reader sees the whole old document or the whole new one. The temporary
66. 66/// name carries the process id and a per-process sequence number and is
67. 67/// created exclusively, so no two writes ever share one; it is removed if
68. 68/// the write fails.
69. 69pub fn replace_file(path: &Path, contents: &str) -> std::io::Result<()> {
70. 70 static SEQUENCE: AtomicU64 = AtomicU64::new(0);
71. 71 let (Some(parent), Some(name)) = (path.parent(), path.file_name()) else {
72. 72 return Err(std::io::Error::other("path has no parent directory"));
73. 73 };
74. 74 std::fs::create_dir_all(parent)?;
75. 75 let mut attempts = 0;
76. 76 let (temp, mut file) = loop {
77. 77 let temp = parent.join(format!(
78. 78 ".{}.{}.{}.tmp",
79. 79 name.to_string_lossy(),
80. 80 std::process::id(),
81. 81 SEQUENCE.fetch_add(1, Ordering::Relaxed)
82. 82 ));
83. 83 match std::fs::OpenOptions::new()
84. 84 .write(true)
85. 85 .create_new(true)
86. 86 .open(&temp)
87. 87 {
88. 88 Ok(file) => break (temp, file),
89. 89 // Only an earlier process that had this pid can hold a fresh
90. 90 // name; step past its file rather than write into it.
91. 91 Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists && attempts < 8 => {
92. 92 attempts += 1;
93. 93 }
94. 94 Err(error) => return Err(error),
95. 95 }
96. 96 };
97. 97 let written =
98. 98 std::io::Write::write_all(&mut file, contents.as_bytes()).and_then(|()| file.sync_all());
99. 99 drop(file);
100. 100 written
101. 101 .and_then(|()| std::fs::rename(&temp, path))
102. 102 .inspect_err(|_| {
103. 103 let _ = std::fs::remove_file(&temp);
104. 104 })
105. 105}
106. 106
107. 107#[cfg(test)]
108. 108#[allow(clippy::unwrap_used, clippy::expect_used)]
109. 109mod tests {
110. 110 use super::*;
111. 111
112. 112 #[test]
113. 113 fn concurrent_appends_all_land_and_leave_no_temporaries() {
114. 114 let dir = tempfile::tempdir().unwrap();
115. 115 let path = dir.path().join("list.txt");
116. 116 std::thread::scope(|scope| {
117. 117 for writer in 0..16 {
118. 118 let path = &path;
119. 119 scope.spawn(move || {
120. 120 for line in 0..25 {
121. 121 update_file(path, |current| {
122. 122 let mut next = current.unwrap_or_default().to_owned();
123. 123 next.push_str(&format!("{writer}-{line}\n"));
124. 124 Ok::<_, std::convert::Infallible>((Some(next), ()))
125. 125 })
126. 126 .unwrap();
127. 127 }
128. 128 });
129. 129 }
130. 130 });
131. 131 let text = std::fs::read_to_string(&path).unwrap();
132. 132 assert_eq!(text.lines().count(), 16 * 25);
133. 133 let names: Vec<_> = std::fs::read_dir(dir.path())
134. 134 .unwrap()
135. 135 .map(|entry| entry.unwrap().file_name())
136. 136 .collect();
137. 137 assert_eq!(names, vec![std::ffi::OsString::from("list.txt")]);
138. 138 }
139. 139
140. 140 #[test]
141. 141 fn declined_or_failed_edit_writes_nothing() {
142. 142 let dir = tempfile::tempdir().unwrap();
143. 143 let path = dir.path().join("kept.txt");
144. 144 std::fs::write(&path, "original").unwrap();
145. 145 update_file(&path, |_| Ok::<_, &str>((None, ()))).unwrap();
146. 146 assert!(matches!(
147. 147 update_file(&path, |_| Err::<(Option<String>, ()), _>("no")),
148. 148 Err(UpdateError::Edit("no"))
149. 149 ));
150. 150 assert_eq!(std::fs::read_to_string(&path).unwrap(), "original");
151. 151 }
152. 152}
153. 153