1use std::path::{Path, PathBuf};
2
3use eoka_protocol::sidecar::{Artifact, ArtifactContainer, ErrorCode, RecordingFailureStage};
4use tokio::io::AsyncWriteExt;
5
6pub const MAX_SAFE_BYTES: u64 = 9_007_199_254_740_991;
7
8#[derive(Debug)]
9pub struct FileFailure {
10 pub stage: RecordingFailureStage,
11 pub message: String,
12 pub partial_path: Option<String>,
13}
14
15impl FileFailure {
16 fn new(
17 stage: RecordingFailureStage,
18 error: impl std::fmt::Display,
19 path: Option<&Path>,
20 ) -> Self {
21 Self {
22 stage,
23 message: error.to_string(),
24 partial_path: path.map(|path| path.to_string_lossy().into_owned()),
25 }
26 }
27
28 pub fn server_error(self) -> eoka_protocol::sidecar::ServerError {
29 let mut error =
30 eoka_protocol::sidecar::ServerError::new(ErrorCode::FileOutput, self.message);
31 error.data.get_or_insert_with(Default::default).partial_path = self.partial_path;
32 error.data.get_or_insert_with(Default::default).detail =
33 Some(eoka_protocol::ErrorDetail::new(
34 format!("file_{:?}", self.stage).to_ascii_lowercase(),
35 error.message.clone(),
36 ));
37 error
38 }
39}
40
41pub struct StagedArtifact {
42 pub output: PathBuf,
43 pub partial: PathBuf,
44 file: Option<tokio::fs::File>,
45 bytes: u64,
46 sealed: bool,
47 identity: (u64, u128),
48}
49
50impl StagedArtifact {
51 pub async fn prepare(output: &str) -> Result<Self, FileFailure> {
52 let output = output.to_owned();
53 tokio::task::spawn_blocking(move || {
54 let path = Path::new(&output);
55 if !path.is_absolute() {
56 return Err(FileFailure::new(
57 RecordingFailureStage::Start,
58 "output path must be absolute",
59 None,
60 ));
61 }
62 let parent = path.parent().ok_or_else(|| {
63 FileFailure::new(
64 RecordingFailureStage::Start,
65 "output needs a parent directory",
66 None,
67 )
68 })?;
69 let name = path.file_name().ok_or_else(|| {
70 FileFailure::new(
71 RecordingFailureStage::Start,
72 "output needs a filename",
73 None,
74 )
75 })?;
76 let parent = std::fs::canonicalize(parent)
77 .map_err(|error| FileFailure::new(RecordingFailureStage::Start, error, None))?;
78 let output = parent.join(name);
79 match std::fs::symlink_metadata(&output) {
80 Ok(_) => {
81 return Err(FileFailure::new(
82 RecordingFailureStage::Start,
83 "output already exists",
84 None,
85 ))
86 }
87 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
88 Err(error) => {
89 return Err(FileFailure::new(RecordingFailureStage::Start, error, None))
90 }
91 }
92 let partial = parent.join(format!(".eoka-capture-{}.partial", uuid::Uuid::new_v4()));
93 let mut options = std::fs::OpenOptions::new();
94 options.read(true).write(true).create_new(true);
95 #[cfg(unix)]
96 {
97 use std::os::unix::fs::OpenOptionsExt;
98 options.mode(0o600);
99 }
100 let file = options
101 .open(&partial)
102 .map_err(|error| FileFailure::new(RecordingFailureStage::Start, error, None))?;
103 let identity = crate::file_publication::identity(&file).map_err(|error| {
104 FileFailure::new(RecordingFailureStage::Start, error, Some(&partial))
105 })?;
106 Ok(Self {
107 output,
108 partial,
109 file: Some(tokio::fs::File::from_std(file)),
110 bytes: 0,
111 sealed: false,
112 identity,
113 })
114 })
115 .await
116 .map_err(|error| FileFailure::new(RecordingFailureStage::Start, error, None))?
117 }
118
119 pub async fn write(&mut self, data: &[u8]) -> Result<(), FileFailure> {
120 let bytes = self
121 .bytes
122 .checked_add(data.len() as u64)
123 .filter(|bytes| *bytes <= MAX_SAFE_BYTES)
124 .ok_or_else(|| {
125 FileFailure::new(
126 RecordingFailureStage::Write,
127 "artifact exceeds the JavaScript-safe byte limit",
128 Some(&self.partial),
129 )
130 })?;
131 let file = self.file.as_mut().ok_or_else(|| {
132 FileFailure::new(
133 RecordingFailureStage::Write,
134 "staging file is closed",
135 Some(&self.partial),
136 )
137 })?;
138 file.write_all(data).await.map_err(|error| {
139 FileFailure::new(RecordingFailureStage::Write, error, Some(&self.partial))
140 })?;
141 self.bytes = bytes;
142 Ok(())
143 }
144
145 pub(crate) fn recording_file(&mut self) -> Result<&mut tokio::fs::File, FileFailure> {
146 self.file.as_mut().ok_or_else(|| {
147 FileFailure::new(
148 RecordingFailureStage::Close,
149 "recording staging file is closed",
150 Some(&self.partial),
151 )
152 })
153 }
154
155 pub async fn seal(&mut self) -> Result<(), FileFailure> {
156 let Some(mut file) = self.file.take() else {
157 return if self.sealed {
158 Ok(())
159 } else {
160 Err(FileFailure::new(
161 RecordingFailureStage::Close,
162 "staging file was abandoned or failed closure",
163 Some(&self.partial),
164 ))
165 };
166 };
167 let flushed = async {
168 file.flush().await?;
169 file.sync_all().await
170 }
171 .await;
172 let closed = crate::file_publication::close_file(file.into_std().await);
173 flushed.map_err(|error| {
174 FileFailure::new(RecordingFailureStage::Flush, error, Some(&self.partial))
175 })?;
176 closed.map_err(|error| {
177 FileFailure::new(RecordingFailureStage::Close, error, Some(&self.partial))
178 })?;
179 self.sealed = true;
180 Ok(())
181 }
182
183 pub async fn abandon(&mut self) -> Result<(), FileFailure> {
184 if let Some(mut file) = self.file.take() {
185 let flushed = file.flush().await;
186 let closed = crate::file_publication::close_file(file.into_std().await);
187 flushed.map_err(|error| {
188 FileFailure::new(RecordingFailureStage::Flush, error, Some(&self.partial))
189 })?;
190 closed.map_err(|error| {
191 FileFailure::new(RecordingFailureStage::Close, error, Some(&self.partial))
192 })?;
193 }
194 Ok(())
195 }
196
197 fn verify(&self, path: &Path) -> Result<(), FileFailure> {
198 let metadata = std::fs::symlink_metadata(path).map_err(|error| {
199 FileFailure::new(RecordingFailureStage::Publish, error, Some(&self.partial))
200 })?;
201 if !metadata.is_file() {
202 return Err(FileFailure::new(
203 RecordingFailureStage::Publish,
204 "artifact is not a regular file",
205 Some(&self.partial),
206 ));
207 }
208 let file = std::fs::File::open(path).map_err(|error| {
209 FileFailure::new(RecordingFailureStage::Publish, error, Some(&self.partial))
210 })?;
211 let metadata = file.metadata();
212 #[cfg(not(unix))]
213 let identity = crate::file_publication::identity(&file);
214 let closed = crate::file_publication::close_file(file);
215 let metadata = metadata.map_err(|error| {
216 FileFailure::new(RecordingFailureStage::Publish, error, Some(&self.partial))
217 })?;
218 #[cfg(unix)]
219 let identity = {
220 use std::os::unix::fs::MetadataExt;
221 (metadata.dev(), u128::from(metadata.ino()))
222 };
223 #[cfg(not(unix))]
224 let identity = identity.map_err(|error| {
225 FileFailure::new(RecordingFailureStage::Publish, error, Some(&self.partial))
226 })?;
227 closed.map_err(|error| {
228 FileFailure::new(RecordingFailureStage::Close, error, Some(&self.partial))
229 })?;
230 if !metadata.is_file() || metadata.len() != self.bytes || identity != self.identity {
231 return Err(FileFailure::new(
232 RecordingFailureStage::Publish,
233 "staged artifact identity or byte length changed",
234 Some(&self.partial),
235 ));
236 }
237 Ok(())
238 }
239
240 pub fn publish(
241 &self,
242 mime_type: &str,
243 container: Option<ArtifactContainer>,
244 ) -> Result<Artifact, FileFailure> {
245 if !self.sealed {
246 return Err(FileFailure::new(
247 RecordingFailureStage::Publish,
248 "staging file must be successfully flushed and closed before publication",
249 Some(&self.partial),
250 ));
251 }
252 self.verify(&self.partial)?;
253 crate::file_publication::publish_no_replace(&self.partial, &self.output).map_err(
254 |error| FileFailure::new(RecordingFailureStage::Publish, error, Some(&self.partial)),
255 )?;
256 if let Err(failure) = self.verify(&self.output) {
257 let rollback = std::fs::remove_file(&self.output);
258 return Err(FileFailure::new(
259 RecordingFailureStage::Publish,
260 format!("{}; publication rollback: {rollback:?}", failure.message),
261 Some(&self.partial),
262 ));
263 }
264 if let Err(error) = std::fs::remove_file(&self.partial) {
265 let rollback = std::fs::remove_file(&self.output);
266 return Err(FileFailure::new(
267 RecordingFailureStage::Publish,
268 format!("staging cleanup failed: {error}; publication rollback: {rollback:?}"),
269 Some(&self.partial),
270 ));
271 }
272 Ok(Artifact {
273 path: self.output.to_string_lossy().into_owned(),
274 bytes: self.bytes,
275 mime_type: mime_type.to_owned(),
276 container,
277 })
278 }
279}
280
281#[cfg(test)]
282mod tests {
283 use super::*;
284
285 #[tokio::test]
286 async fn collision_retains_partial_and_never_overwrites() {
287 let root = tempfile::tempdir().unwrap();
288 let output = root.path().join("capture.png");
289 let mut stage = StagedArtifact::prepare(output.to_str().unwrap())
290 .await
291 .unwrap();
292 stage.write(b"complete image").await.unwrap();
293 assert!(stage.publish("image/png", None).is_err());
294 stage.seal().await.unwrap();
295 std::fs::write(&output, b"previous").unwrap();
296 let failure = stage.publish("image/png", None).unwrap_err();
297 assert_eq!(failure.stage, RecordingFailureStage::Publish);
298 assert_eq!(std::fs::read(&output).unwrap(), b"previous");
299 assert_eq!(std::fs::read(&stage.partial).unwrap(), b"complete image");
300 }
301
302 #[tokio::test]
303 async fn staged_bytes_publish_only_after_close() {
304 let root = tempfile::tempdir().unwrap();
305 let output = root.path().join("capture.webp");
306 let mut stage = StagedArtifact::prepare(output.to_str().unwrap())
307 .await
308 .unwrap();
309 stage.write(b"one").await.unwrap();
310 stage.write(b"two").await.unwrap();
311 assert!(!output.exists());
312 stage.seal().await.unwrap();
313 let artifact = stage.publish("image/webp", None).unwrap();
314 assert_eq!(artifact.bytes, 6);
315 assert_eq!(std::fs::read(output).unwrap(), b"onetwo");
316 assert!(!stage.partial.exists());
317 }
318
319 #[tokio::test]
320 async fn failed_write_and_symlink_destinations_are_not_success() {
321 let root = tempfile::tempdir().unwrap();
322 let output = root.path().join("capture.mp4");
323 let mut stage = StagedArtifact::prepare(output.to_str().unwrap())
324 .await
325 .unwrap();
326 stage.bytes = MAX_SAFE_BYTES;
327 assert_eq!(
328 stage.write(b"x").await.unwrap_err().stage,
329 RecordingFailureStage::Write
330 );
331 stage.abandon().await.unwrap();
332 assert!(!output.exists());
333 assert!(stage.partial.exists());
334 #[cfg(unix)]
335 {
336 std::os::unix::fs::symlink(root.path().join("missing"), &output).unwrap();
337 assert!(StagedArtifact::prepare(output.to_str().unwrap())
338 .await
339 .is_err());
340 }
341 }
342
343 #[tokio::test]
344 async fn borrowed_recording_file_keeps_original_identity_when_path_is_replaced() {
345 use std::io::SeekFrom;
346 use tokio::io::{AsyncReadExt, AsyncSeekExt};
347
348 let root = tempfile::tempdir().unwrap();
349 let output = root.path().join("capture.mp4");
350 let retained = root.path().join("retained");
351 let mut stage = StagedArtifact::prepare(output.to_str().unwrap())
352 .await
353 .unwrap();
354 stage.write(b"original").await.unwrap();
355 stage.recording_file().unwrap().flush().await.unwrap();
356 std::fs::rename(&stage.partial, &retained).unwrap();
357 std::fs::write(&stage.partial, b"replaced").unwrap();
358 let file = stage.recording_file().unwrap();
359 file.seek(SeekFrom::Start(0)).await.unwrap();
360 let mut bytes = [0; 8];
361 file.read_exact(&mut bytes).await.unwrap();
362 assert_eq!(&bytes, b"original");
363 file.seek(SeekFrom::Start(0)).await.unwrap();
364 file.write_all(b"retained").await.unwrap();
365 stage.seal().await.unwrap();
366 assert!(stage.recording_file().is_err());
367 assert!(stage
368 .publish("video/mp4", Some(ArtifactContainer::Mp4))
369 .is_err());
370 assert_eq!(std::fs::read(&retained).unwrap(), b"retained");
371 assert_eq!(std::fs::read(&stage.partial).unwrap(), b"replaced");
372 assert!(!output.exists());
373 }
374
375 #[tokio::test]
376 async fn replaced_staging_files_and_abandoned_files_cannot_publish() {
377 let root = tempfile::tempdir().unwrap();
378 let output = root.path().join("capture.png");
379 let mut stage = StagedArtifact::prepare(output.to_str().unwrap())
380 .await
381 .unwrap();
382 stage.write(b"original").await.unwrap();
383 stage.seal().await.unwrap();
384 std::fs::rename(&stage.partial, root.path().join("original")).unwrap();
385 std::fs::write(&stage.partial, b"replaced").unwrap();
386 assert!(stage.publish("image/png", None).is_err());
387 assert!(!output.exists());
388 let mut abandoned = StagedArtifact::prepare(output.to_str().unwrap())
389 .await
390 .unwrap();
391 abandoned.write(b"partial").await.unwrap();
392 abandoned.abandon().await.unwrap();
393 assert!(abandoned.seal().await.is_err());
394 assert!(abandoned.publish("image/png", None).is_err());
395 assert!(!output.exists());
396 }
397}