1use std::fs::{self, File, OpenOptions};
24use std::io::{Read, Seek, SeekFrom, Write};
25use std::path::{Path, PathBuf};
26use std::sync::atomic::{AtomicU64, Ordering};
27use std::time::{Duration, Instant};
28
29use alpha2_ffmpeg_bridge::{CreateOutputError, OutputFile};
30use alpha2_media::{CancelToken, Container, EncodeError};
31use alpha2_os::{NameLimit, NameUnit, PublishMethod, Published, RetryPolicy};
32
33pub const PARTIAL_OUTPUT_IN_USE: &str = "another export is writing this output";
35
36const TAG_PARTIAL: &str = "a2partial";
37const TAG_KEPT: &str = "a2kept";
38const TAG_PROBE: [&str; 2] = ["a2probe1", "a2probe2"];
40const KEPT_NAMES: u32 = 99;
42
43pub(crate) const PUBLISH_RETRY_BUDGET: Duration = Duration::from_secs(10);
46const PROBE_RETRY_BUDGET: Duration = Duration::from_secs(2);
48const CLAIM_IN_USE_WAIT: Duration = Duration::from_millis(250);
51const CLAIM_IN_USE_POLL: Duration = Duration::from_millis(10);
52const STAGING_COPY_CHUNK: usize = 4 << 20;
53
54fn output_error(error: impl std::fmt::Display) -> EncodeError {
55 EncodeError::Output(error.to_string())
56}
57
58fn output_error_at(path: &Path, error: impl std::fmt::Display) -> EncodeError {
59 EncodeError::Output(format!("{}: {error}", path.display()))
60}
61
62#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
64pub enum ExistingOutput {
65 #[default]
67 Refuse,
68 Replace,
71}
72
73#[derive(Clone, Debug, Eq, PartialEq)]
75pub struct OutputTarget {
76 pub path: PathBuf,
78 pub existing: ExistingOutput,
80}
81
82impl OutputTarget {
83 #[must_use]
85 pub fn replacing(path: impl Into<PathBuf>) -> Self {
86 Self {
87 path: path.into(),
88 existing: ExistingOutput::Replace,
89 }
90 }
91}
92
93impl From<PathBuf> for OutputTarget {
94 fn from(path: PathBuf) -> Self {
95 Self {
96 path,
97 existing: ExistingOutput::Refuse,
98 }
99 }
100}
101
102impl From<&Path> for OutputTarget {
103 fn from(path: &Path) -> Self {
104 path.to_owned().into()
105 }
106}
107
108impl From<&PathBuf> for OutputTarget {
109 fn from(path: &PathBuf) -> Self {
110 path.clone().into()
111 }
112}
113
114impl From<String> for OutputTarget {
115 fn from(path: String) -> Self {
116 PathBuf::from(path).into()
117 }
118}
119
120impl From<&String> for OutputTarget {
121 fn from(path: &String) -> Self {
122 PathBuf::from(path).into()
123 }
124}
125
126impl From<&str> for OutputTarget {
127 fn from(path: &str) -> Self {
128 PathBuf::from(path).into()
129 }
130}
131
132impl From<&OutputTarget> for OutputTarget {
133 fn from(target: &OutputTarget) -> Self {
134 target.clone()
135 }
136}
137
138#[derive(Clone, Copy, Debug, Eq, PartialEq)]
140pub struct PublishReport {
141 pub method: PublishMethod,
143 pub staged_locally: bool,
146 pub retries: u32,
148}
149
150#[derive(Clone, Copy, Debug, Eq, PartialEq)]
152pub(crate) enum SeekNeed {
153 Required,
155 NotNeeded,
156}
157
158const CLAIM_ATTEMPTS: usize = 8;
161
162fn parent_dir(final_path: &Path) -> &Path {
163 final_path
164 .parent()
165 .filter(|parent| !parent.as_os_str().is_empty())
166 .unwrap_or(Path::new("."))
167}
168
169fn fnv1a_64(bytes: &[u8]) -> u64 {
171 let mut hash = 0xcbf2_9ce4_8422_2325_u64;
172 for byte in bytes {
173 hash ^= u64::from(*byte);
174 hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
175 }
176 hash
177}
178
179fn sibling_name(name: &str, stem: &str, extension: &str, tag: &str, limit: NameLimit) -> String {
183 let named = |stem: &str, tag: &str| format!("{stem}.{tag}.{extension}");
184 if limit.length(&named(stem, TAG_PARTIAL)) <= limit.max {
185 return named(stem, tag);
186 }
187 let hash = format!("~{:08x}", fnv1a_64(name.as_bytes()) & 0xffff_ffff);
188 let room = limit
189 .max
190 .saturating_sub(limit.length(&named(&hash, TAG_PARTIAL)));
191 let mut used = 0;
192 let mut prefix = 0;
193 for (index, character) in stem.char_indices() {
194 let units = match limit.unit {
195 NameUnit::Bytes => character.len_utf8(),
196 NameUnit::Utf16 => character.len_utf16(),
197 };
198 if used + units > room {
199 break;
200 }
201 used += units;
202 prefix = index + character.len_utf8();
203 }
204 named(&format!("{}{hash}", &stem[..prefix]), tag)
205}
206
207fn sibling_path(final_path: &Path, tag: &str) -> Result<PathBuf, EncodeError> {
213 let name = final_path
214 .file_name()
215 .and_then(|name| name.to_str())
216 .ok_or_else(|| EncodeError::Output("output path has no UTF-8 file name".into()))?;
217 let extension = final_path
218 .extension()
219 .and_then(|extension| extension.to_str())
220 .unwrap_or("bin");
221 let stem = final_path
222 .file_stem()
223 .and_then(|stem| stem.to_str())
224 .unwrap_or(name);
225 let limit = alpha2_os::name_limit(parent_dir(final_path));
226 Ok(final_path.with_file_name(sibling_name(name, stem, extension, tag, limit)))
227}
228
229pub(crate) fn partial_output_path(final_path: &Path) -> Result<PathBuf, EncodeError> {
234 sibling_path(final_path, TAG_PARTIAL)
235}
236
237fn kept_tag(n: u32) -> String {
238 if n <= 1 {
239 TAG_KEPT.to_owned()
240 } else {
241 format!("{TAG_KEPT}{n}")
242 }
243}
244
245static NEXT_SCRATCH: AtomicU64 = AtomicU64::new(0);
246
247fn discard_path(partial: &Path) -> PathBuf {
249 let n = NEXT_SCRATCH.fetch_add(1, Ordering::Relaxed);
250 partial.with_file_name(format!(".a2del-{}-{n}", std::process::id()))
251}
252
253fn in_use(partial: &Path) -> EncodeError {
254 EncodeError::OutputInUse(format!(
255 "{PARTIAL_OUTPUT_IN_USE}: its partial file {} is held by an export still running",
256 partial.display()
257 ))
258}
259
260type Lock<'a> = &'a mut dyn FnMut(&File) -> std::io::Result<bool>;
262
263fn lock_partial(
272 path: &Path,
273 create: bool,
274 lock: Lock<'_>,
275) -> Result<Option<(File, bool)>, EncodeError> {
276 let at = |error: std::io::Error| output_error_at(path, error);
277 let started = Instant::now();
278 let mut changes = 0;
279 while changes < CLAIM_ATTEMPTS {
280 let Some((file, created)) = open_partial(path, create).map_err(at)? else {
281 if create {
282 changes += 1;
285 continue;
286 }
287 return Ok(None);
288 };
289 let locked = match lock(&file) {
290 Ok(locked) => locked,
291 Err(error) => {
292 if created && alpha2_os::same_file(&file, path).unwrap_or(false) {
293 let _ = fs::remove_file(path);
294 }
295 return Err(at(error));
296 }
297 };
298 if !locked {
299 if started.elapsed() < CLAIM_IN_USE_WAIT {
300 drop(file);
301 std::thread::sleep(CLAIM_IN_USE_POLL);
302 continue;
303 }
304 return Err(in_use(path));
305 }
306 if alpha2_os::same_file(&file, path).map_err(at)? {
308 return Ok(Some((file, created)));
309 }
310 changes += 1;
311 }
312 Err(EncodeError::Output(format!(
313 "the partial file {} kept changing while it was claimed",
314 path.display()
315 )))
316}
317
318fn open_partial(path: &Path, create: bool) -> std::io::Result<Option<(File, bool)>> {
321 let mut options = OpenOptions::new();
322 options.read(true).write(true);
323 if create {
324 match options.clone().create_new(true).open(path) {
325 Ok(file) => return Ok(Some((file, true))),
326 Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
327 Err(error) => return Err(error),
328 }
329 }
330 match options.open(path) {
331 Ok(file) => Ok(Some((file, false))),
332 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
333 Err(error) => Err(error),
334 }
335}
336
337fn other_names(file: &File) -> String {
340 #[cfg(unix)]
341 {
342 use std::os::unix::fs::MetadataExt;
343 match file.metadata().map(|metadata| metadata.nlink()) {
344 Ok(links) if links > 1 => format!(" (it had {} other names)", links - 1),
345 _ => String::new(),
346 }
347 }
348 #[cfg(not(unix))]
349 {
350 let _ = file;
351 String::new()
352 }
353}
354
355#[derive(Clone, Copy, Debug, Eq, PartialEq)]
357enum PartialState {
358 Writing,
360 Finished,
362 Published,
364 Kept,
366 Discarded,
368}
369
370pub(crate) struct PartialOutput {
377 target: OutputTarget,
378 path: PathBuf,
379 lock: Option<File>,
380 staging: Option<PathBuf>,
381 state: PartialState,
382}
383
384impl PartialOutput {
385 pub(crate) fn claim(target: &OutputTarget, seek: SeekNeed) -> Result<Self, EncodeError> {
393 match target.existing {
394 ExistingOutput::Refuse => {
395 if target.path.exists() {
396 return Err(EncodeError::Output(
397 "refusing to overwrite an existing output file".into(),
398 ));
399 }
400 }
401 ExistingOutput::Replace => {
402 replaceable(&target.path).map_err(EncodeError::Output)?;
403 #[cfg(windows)]
404 warn_if_held(&target.path);
405 }
406 }
407 let mut claimed = Self::claim_partial_locking(target, &mut alpha2_os::try_lock_exclusive)?;
408 if !claimed.probe(seek)? && seek == SeekNeed::Required {
409 let staging = create_staging(&target.path)?;
410 alpha2_os::diagnostic!(
411 "alpha2: {} cannot rewrite a file at an earlier offset; this export is written to {} and copied there when it is finished",
412 parent_dir(&target.path).display(),
413 staging.display()
414 );
415 claimed.staging = Some(staging);
416 }
417 Ok(claimed)
418 }
419
420 fn claim_partial_locking(target: &OutputTarget, lock: Lock<'_>) -> Result<Self, EncodeError> {
423 let path = partial_output_path(&target.path)?;
424 for _ in 0..CLAIM_ATTEMPTS {
425 let (file, created) = lock_partial(&path, true, lock)?
426 .ok_or_else(|| EncodeError::Output("the partial file was not created".into()))?;
427 if created {
428 return Ok(Self {
429 target: target.clone(),
430 path,
431 lock: Some(file),
432 staging: None,
433 state: PartialState::Writing,
434 });
435 }
436 alpha2_os::diagnostic!(
437 "alpha2: replacing {}, a partial file no running export holds{}",
438 path.display(),
439 other_names(&file)
440 );
441 match fs::remove_file(&path) {
443 Ok(()) => {}
444 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
445 Err(error) => return Err(output_error_at(&path, error)),
446 }
447 drop(file);
448 }
449 Err(EncodeError::Output(format!(
450 "the partial file {} kept changing while it was claimed",
451 path.display()
452 )))
453 }
454
455 fn probe(&self, seek: SeekNeed) -> Result<bool, EncodeError> {
463 let first = sibling_path(&self.target.path, TAG_PROBE[0])?;
464 let second = sibling_path(&self.target.path, TAG_PROBE[1])?;
465 for probe in [&first, &second] {
466 match fs::remove_file(probe) {
467 Ok(()) => {}
468 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
469 Err(error) => return Err(output_error_at(probe, error)),
470 }
471 }
472 let probed = self.probe_with(&first, &second, seek);
473 let _ = fs::remove_file(&first);
474 let _ = fs::remove_file(&second);
475 probed
476 }
477
478 fn probe_with(&self, first: &Path, second: &Path, seek: SeekNeed) -> Result<bool, EncodeError> {
479 let mut file = OpenOptions::new()
480 .write(true)
481 .create_new(true)
482 .open(first)
483 .map_err(|error| output_error_at(first, error))?;
484 let seekable = match seek {
485 SeekNeed::NotNeeded => true,
486 SeekNeed::Required => match rewrite_at_start(&mut file) {
487 Ok(()) => !hooks::not_seekable(),
488 Err(error) if cannot_seek(&error) => false,
489 Err(error) => return Err(output_error_at(first, error)),
490 },
491 };
492 drop(file);
493 let keep_trying = || true;
494 publish_no_replace(
495 first,
496 second,
497 &RetryPolicy {
498 budget: PROBE_RETRY_BUDGET,
499 on_wait: &keep_trying,
500 },
501 )
502 .map_err(|error| {
503 EncodeError::Output(format!(
504 "{} cannot take a finished file without replacing one: {error}",
505 parent_dir(&self.target.path).display()
506 ))
507 })?;
508 Ok(seekable)
509 }
510
511 pub(crate) fn write_path(&self) -> &Path {
513 self.staging.as_deref().unwrap_or(&self.path)
514 }
515
516 pub(crate) fn open_muxer(&self, container: Container) -> Result<OutputFile, EncodeError> {
522 OutputFile::create(self.write_path(), crate::convert::muxer_name(container)).map_err(
523 |error| match error {
524 CreateOutputError::Path(error) => {
525 EncodeError::Output(format!("output path {error}"))
526 }
527 CreateOutputError::Ffmpeg(error) => output_error(error),
528 },
529 )
530 }
531
532 pub(crate) fn open_file(&self) -> Result<File, EncodeError> {
537 OpenOptions::new()
538 .write(true)
539 .open(self.write_path())
540 .map_err(output_error)
541 }
542
543 pub(crate) fn explain(&self, error: EncodeError) -> EncodeError {
546 match (error, &self.staging) {
547 (EncodeError::Adapter(message), Some(staging)) => EncodeError::Adapter(format!(
548 "{message}; the output folder cannot be written in place, so this export was written to the staging file {} first",
549 staging.display()
550 )),
551 (error, _) => error,
552 }
553 }
554
555 pub(crate) fn finish(&mut self, cancel: &CancelToken) -> Result<PublishReport, EncodeError> {
563 if self.state != PartialState::Writing {
564 return Err(EncodeError::Adapter(
565 "the output was already finished".into(),
566 ));
567 }
568 self.state = PartialState::Finished;
569 let staged_locally = self.staging.is_some();
570 if let Some(staging) = self.staging.clone()
571 && let Err(error) = self.copy_staging(&staging)
572 {
573 return Err(self.keep_staging(staging, &error));
574 }
575 let beat_unless_cancelled = || {
576 crate::encode::beat_encode_call();
577 !cancel.is_cancelled()
578 };
579 let retry = RetryPolicy {
580 budget: PUBLISH_RETRY_BUDGET,
581 on_wait: &beat_unless_cancelled,
582 };
583 match self.sync_and_publish(&retry) {
584 Ok(published) => {
585 self.state = PartialState::Published;
586 drop(self.lock.take());
587 self.remove_staging();
588 #[cfg(unix)]
589 if let Err(error) = sync_directory(parent_dir(&self.target.path)) {
590 alpha2_os::diagnostic!(
591 "alpha2: {} is in place, but syncing its folder failed: {error}",
592 self.target.path.display()
593 );
594 }
595 Ok(PublishReport {
596 method: published.method,
597 staged_locally,
598 retries: published.retries,
599 })
600 }
601 Err(_) if cancel.is_cancelled() => {
602 self.state = PartialState::Writing;
603 self.discard()?;
604 Err(EncodeError::Cancelled)
605 }
606 Err(message) => Err(self.keep(message)),
607 }
608 }
609
610 fn copy_staging(&self, staging: &Path) -> Result<(), String> {
613 let mut source = File::open(staging).map_err(|error| error.to_string())?;
614 let mut lock = self
615 .lock
616 .as_ref()
617 .ok_or_else(|| "the partial file is no longer held".to_owned())?;
618 lock.seek(SeekFrom::Start(0))
619 .map_err(|error| error.to_string())?;
620 let mut chunk = vec![0_u8; STAGING_COPY_CHUNK];
621 loop {
622 crate::encode::beat_encode_call();
623 let read = source
624 .read(&mut chunk)
625 .map_err(|error| format!("reading {}: {error}", staging.display()))?;
626 if read == 0 {
627 return Ok(());
628 }
629 lock.write_all(&chunk[..read])
630 .map_err(|error| format!("writing {}: {error}", self.path.display()))?;
631 }
632 }
633
634 fn sync_and_publish(&self, retry: &RetryPolicy<'_>) -> Result<Published, String> {
636 let final_path = self.target.path.as_path();
637 let lock = self
638 .lock
639 .as_ref()
640 .ok_or_else(|| "the partial file is no longer held".to_owned())?;
641 lock.sync_all()
642 .map_err(|error| format!("syncing {}: {error}", self.path.display()))?;
643 hooks::record("sync");
644 let published = match self.target.existing {
645 ExistingOutput::Refuse => {
646 publish_no_replace(&self.path, final_path, retry).map_err(|error| {
647 if error.kind() == std::io::ErrorKind::AlreadyExists {
648 format!(
649 "a file appeared at {} while this export ran; it was left as it is",
650 final_path.display()
651 )
652 } else {
653 format!("{}: {error}", final_path.display())
654 }
655 })
656 }
657 ExistingOutput::Replace => replaceable(final_path).and_then(|()| {
660 alpha2_os::publish_replace(&self.path, final_path, retry)
661 .map_err(|error| format!("{}: {error}", final_path.display()))
662 }),
663 }?;
664 hooks::record("rename");
665 Ok(published)
666 }
667
668 fn keep(&mut self, message: String) -> EncodeError {
671 let kept = self.move_aside(&self.path);
672 self.state = PartialState::Kept;
673 drop(self.lock.take());
674 self.remove_staging();
675 match kept {
676 Ok(kept) => EncodeError::Publish {
677 message,
678 kept: Some(kept),
679 },
680 Err(error) => EncodeError::Publish {
681 message: format!(
682 "{message}; moving it aside failed too ({error}), so it is left as {}, which the next export to this path removes",
683 self.path.display()
684 ),
685 kept: None,
686 },
687 }
688 }
689
690 fn keep_staging(&mut self, staging: PathBuf, error: &str) -> EncodeError {
693 self.staging = None;
694 self.state = PartialState::Writing;
695 let _ = self.discard();
696 self.state = PartialState::Kept;
697 EncodeError::Publish {
698 message: format!(
699 "{}: copying the finished render into {} failed: {error}",
700 self.target.path.display(),
701 self.path.display()
702 ),
703 kept: Some(staging),
704 }
705 }
706
707 fn move_aside(&self, source: &Path) -> Result<PathBuf, String> {
710 let beat = || {
711 crate::encode::beat_encode_call();
712 true
713 };
714 let retry = RetryPolicy {
715 budget: PUBLISH_RETRY_BUDGET,
716 on_wait: &beat,
717 };
718 for n in 1..=KEPT_NAMES {
719 let kept =
720 sibling_path(&self.target.path, &kept_tag(n)).map_err(|error| error.to_string())?;
721 match publish_no_replace(source, &kept, &retry) {
722 Ok(_) => return Ok(kept),
723 Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
724 Err(error) => return Err(format!("{}: {error}", kept.display())),
725 }
726 }
727 Err(format!(
728 "every kept name from {} to {} is taken",
729 kept_tag(1),
730 kept_tag(KEPT_NAMES)
731 ))
732 }
733
734 fn remove_staging(&mut self) {
735 if let Some(staging) = self.staging.take() {
736 let _ = fs::remove_file(staging);
737 }
738 }
739
740 pub(crate) fn discard(&mut self) -> Result<(), EncodeError> {
751 match self.state {
752 PartialState::Writing => {}
753 PartialState::Finished => {
754 let kept = self.keep("the export stopped before its output was published".into());
755 alpha2_os::diagnostic!("alpha2: {kept}");
756 return Ok(());
757 }
758 PartialState::Published | PartialState::Kept | PartialState::Discarded => {
759 return Ok(());
760 }
761 }
762 self.state = PartialState::Discarded;
763 self.remove_staging();
764 let Some(lock) = self.lock.take() else {
765 return Ok(());
766 };
767 let discard = discard_path(&self.path);
768 let (removed, at) = if publish_no_replace(&self.path, &discard, &RetryPolicy::NONE).is_ok()
769 {
770 close_lock(lock, &discard);
771 let keep_trying = || true;
772 let retry = RetryPolicy {
773 budget: PUBLISH_RETRY_BUDGET,
774 on_wait: &keep_trying,
775 };
776 let removed =
777 alpha2_os::with_publish_retries(&retry, || fs::remove_file(&discard)).map(|_| ());
778 (removed, discard)
779 } else {
780 let removed = fs::remove_file(&self.path);
781 close_lock(lock, &self.path);
782 (removed, self.path.clone())
783 };
784 match removed {
785 Ok(()) => Ok(()),
786 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
787 Err(error) => Err(output_error_at(&at, error)),
788 }
789 }
790}
791
792impl Drop for PartialOutput {
793 fn drop(&mut self) {
794 let _ = self.discard();
795 }
796}
797
798fn close_lock(lock: File, name: &Path) {
800 drop(lock);
801 hooks::lock_closed(name);
802}
803
804fn rewrite_at_start(file: &mut File) -> std::io::Result<()> {
806 file.write_all(&[0, 0])?;
807 file.seek(SeekFrom::Start(0))?;
808 file.write_all(&[1])?;
809 file.sync_all()
810}
811
812fn cannot_seek(error: &std::io::Error) -> bool {
815 matches!(
816 error.kind(),
817 std::io::ErrorKind::NotSeekable
818 | std::io::ErrorKind::InvalidInput
819 | std::io::ErrorKind::Unsupported
820 )
821}
822
823fn create_staging(final_path: &Path) -> Result<PathBuf, EncodeError> {
825 let extension = final_path
826 .extension()
827 .and_then(|extension| extension.to_str())
828 .unwrap_or("bin");
829 for _ in 0..CLAIM_ATTEMPTS {
830 let n = NEXT_SCRATCH.fetch_add(1, Ordering::Relaxed);
831 let staging = std::env::temp_dir().join(format!(
832 "alpha2-staging-{}-{n}.{extension}",
833 std::process::id()
834 ));
835 match OpenOptions::new()
836 .write(true)
837 .create_new(true)
838 .open(&staging)
839 {
840 Ok(_) => return Ok(staging),
841 Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
842 Err(error) => return Err(output_error_at(&staging, error)),
843 }
844 }
845 Err(EncodeError::Output(
846 "no staging file could be created in the temporary directory".into(),
847 ))
848}
849
850#[cfg(windows)]
852fn warn_if_held(final_path: &Path) {
853 use std::os::windows::fs::OpenOptionsExt;
854 const DELETE: u32 = 0x0001_0000;
855 const SHARE_READ_WRITE_DELETE: u32 = 0x1 | 0x2 | 0x4;
856 const ERROR_SHARING_VIOLATION: i32 = 32;
857 if let Err(error) = OpenOptions::new()
858 .access_mode(DELETE)
859 .share_mode(SHARE_READ_WRITE_DELETE)
860 .open(final_path)
861 && error.raw_os_error() == Some(ERROR_SHARING_VIOLATION)
862 {
863 alpha2_os::diagnostic!(
864 "alpha2: {} is open in another program; the export will wait up to {} s for it when it finishes",
865 final_path.display(),
866 PUBLISH_RETRY_BUDGET.as_secs()
867 );
868 }
869}
870
871fn publish_no_replace(
873 from: &Path,
874 to: &Path,
875 retry: &RetryPolicy<'_>,
876) -> std::io::Result<Published> {
877 if let Some(error) = hooks::publish_failure(to) {
878 return Err(error);
879 }
880 alpha2_os::publish_no_replace(from, to, retry)
881}
882
883#[cfg(unix)]
885fn sync_directory(dir: &Path) -> std::io::Result<()> {
886 alpha2_os::sync_directory(dir)?;
887 hooks::record("directory sync");
888 Ok(())
889}
890
891fn replaceable(final_path: &Path) -> Result<(), String> {
894 match fs::symlink_metadata(final_path) {
895 Ok(metadata) if metadata.file_type().is_file() => Ok(()),
896 Ok(_) => Err(format!(
897 "refusing to replace {}: not a regular file",
898 final_path.display()
899 )),
900 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
901 Err(error) => Err(format!("{}: {error}", final_path.display())),
902 }
903}
904
905pub fn write_whole(
914 target: impl Into<OutputTarget>,
915 write: impl FnOnce(&mut File) -> Result<(), String>,
916) -> Result<PublishReport, EncodeError> {
917 let target = target.into();
918 let mut partial = PartialOutput::claim(&target, SeekNeed::NotNeeded)?;
919 let mut file = partial.open_file()?;
920 write(&mut file).map_err(|error| output_error_at(partial.write_path(), error))?;
921 drop(file);
922 partial.finish(&CancelToken::default())
923}
924
925pub fn write_png(
931 target: impl Into<OutputTarget>,
932 width: u32,
933 height: u32,
934 rgba: &[u8],
935) -> Result<PublishReport, EncodeError> {
936 let expected = u64::from(width) * u64::from(height) * 4;
937 if u64::try_from(rgba.len()).ok() != Some(expected) {
938 return Err(EncodeError::Adapter(format!(
939 "a {width}x{height} RGBA image is {expected} bytes, not {}",
940 rgba.len()
941 )));
942 }
943 write_whole(target, |file| {
944 let mut buffered = std::io::BufWriter::new(file);
945 let mut encoder = png::Encoder::new(&mut buffered, width, height);
946 encoder.set_color(png::ColorType::Rgba);
947 encoder.set_depth(png::BitDepth::Eight);
948 encoder.set_source_srgb(png::SrgbRenderingIntent::Perceptual);
950 let mut writer = encoder.write_header().map_err(|error| error.to_string())?;
951 writer
952 .write_image_data(rgba)
953 .map_err(|error| error.to_string())?;
954 writer.finish().map_err(|error| error.to_string())?;
955 buffered
956 .into_inner()
957 .map(|_| ())
958 .map_err(|error| error.error().to_string())
959 })
960}
961
962pub fn remove_stale_partial_output(path: impl AsRef<Path>) -> Result<bool, EncodeError> {
970 let partial = partial_output_path(path.as_ref())?;
971 let Some((file, _)) = lock_partial(&partial, false, &mut alpha2_os::try_lock_exclusive)? else {
972 return Ok(false);
973 };
974 let removed = fs::remove_file(&partial);
975 drop(file);
976 match removed {
977 Ok(()) => Ok(true),
978 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
979 Err(error) => Err(output_error(error)),
980 }
981}
982
983#[cfg(not(test))]
985mod hooks {
986 use std::path::Path;
987
988 pub(super) fn publish_failure(_to: &Path) -> Option<std::io::Error> {
989 None
990 }
991
992 pub(super) fn not_seekable() -> bool {
993 false
994 }
995
996 pub(super) fn record(_event: &'static str) {}
997
998 pub(super) fn lock_closed(_name: &Path) {}
999}
1000
1001#[cfg(test)]
1003pub(crate) mod hooks {
1004 use std::cell::{Cell, RefCell};
1005 use std::path::{Path, PathBuf};
1006
1007 pub(crate) type LockCloseHook = Box<dyn Fn(&Path)>;
1009
1010 thread_local! {
1011 pub(crate) static FAIL_PUBLISH_TO: RefCell<Option<(PathBuf, i32)>> =
1013 const { RefCell::new(None) };
1014 pub(crate) static NOT_SEEKABLE: Cell<bool> = const { Cell::new(false) };
1016 pub(crate) static EVENTS: RefCell<Option<Vec<&'static str>>> =
1018 const { RefCell::new(None) };
1019 pub(crate) static ON_LOCK_CLOSE: RefCell<Option<LockCloseHook>> =
1021 const { RefCell::new(None) };
1022 }
1023
1024 pub(super) fn publish_failure(to: &Path) -> Option<std::io::Error> {
1025 FAIL_PUBLISH_TO.with_borrow(|fail| {
1026 fail.as_ref()
1027 .filter(|(path, _)| path == to)
1028 .map(|(_, code)| std::io::Error::from_raw_os_error(*code))
1029 })
1030 }
1031
1032 pub(super) fn not_seekable() -> bool {
1033 NOT_SEEKABLE.get()
1034 }
1035
1036 pub(super) fn record(event: &'static str) {
1037 EVENTS.with_borrow_mut(|events| {
1038 if let Some(events) = events {
1039 events.push(event);
1040 }
1041 });
1042 }
1043
1044 pub(super) fn lock_closed(name: &Path) {
1045 ON_LOCK_CLOSE.with_borrow(|hook| {
1046 if let Some(hook) = hook {
1047 hook(name);
1048 }
1049 });
1050 }
1051}
1052
1053#[cfg(test)]
1054mod tests {
1055 use std::cell::RefCell;
1056 use std::io::{Read, Write};
1057
1058 use super::*;
1059
1060 fn final_path(label: &str) -> PathBuf {
1062 let path =
1063 std::env::temp_dir().join(format!("alpha2-claim-{label}-{}.mp4", std::process::id()));
1064 let _ = fs::remove_file(&path);
1065 let _ = fs::remove_file(partial_output_path(&path).expect("partial path"));
1066 path
1067 }
1068
1069 fn no_locks(_: &File) -> std::io::Result<bool> {
1071 Err(std::io::Error::other("No locks available"))
1072 }
1073
1074 #[test]
1075 fn a_partial_the_claim_created_and_could_not_lock_is_removed() {
1076 let path = final_path("unlockable");
1077 let partial = partial_output_path(&path).expect("partial path");
1078 let Err(error) =
1079 PartialOutput::claim_partial_locking(&OutputTarget::from(&path), &mut no_locks)
1080 else {
1081 panic!("a claim without a lock succeeded");
1082 };
1083 assert!(
1084 matches!(&error, EncodeError::Output(message) if message.contains(&partial.display().to_string())),
1085 "{error}"
1086 );
1087 assert!(
1088 !partial.exists(),
1089 "the refused claim left the partial it created"
1090 );
1091 }
1092
1093 #[test]
1094 fn a_partial_that_was_there_is_left_when_the_lock_fails() {
1095 let path = final_path("unlockable-existing");
1096 let partial = partial_output_path(&path).expect("partial path");
1097 fs::write(&partial, b"another export's").expect("existing partial");
1098 assert!(
1099 PartialOutput::claim_partial_locking(&OutputTarget::from(&path), &mut no_locks)
1100 .is_err()
1101 );
1102 assert_eq!(
1103 fs::read(&partial).expect("still there"),
1104 b"another export's"
1105 );
1106 fs::remove_file(&partial).expect("clean up");
1107 }
1108
1109 #[test]
1110 fn a_partial_its_holder_removes_while_it_is_claimed_is_claimed_new() {
1111 let _child = crate::tests::child_process_lock();
1112 let path = final_path("removed-while-claimed");
1115 let partial = partial_output_path(&path).expect("partial path");
1116 let target = OutputTarget::from(&path);
1117 let mut holder = Some(PartialOutput::claim(&target, SeekNeed::Required).expect("holder"));
1118 let claimed = PartialOutput::claim_partial_locking(&target, &mut |file| {
1119 drop(holder.take());
1120 alpha2_os::try_lock_exclusive(file)
1121 })
1122 .expect("claimed after the holder finished");
1123 assert!(holder.is_none());
1124 let mut writer = claimed.open_file().expect("the claimed partial opens");
1125 writer.write_all(b"new export").expect("write");
1126 drop(writer);
1127 let mut read = String::new();
1128 File::open(&partial)
1129 .and_then(|mut file| file.read_to_string(&mut read))
1130 .expect("read back");
1131 assert_eq!(read, "new export");
1132 drop(claimed);
1133 assert!(!partial.exists());
1134 }
1135
1136 #[test]
1137 fn a_stale_partial_is_replaced_not_emptied() {
1138 let path = final_path("stale-replaced");
1139 let partial = partial_output_path(&path).expect("partial path");
1140 fs::write(&partial, b"what a killed export left").expect("stale partial");
1141 let stale = File::open(&partial).expect("the stale file");
1142 let claimed =
1143 PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required).expect("claim");
1144 assert!(
1145 !alpha2_os::same_file(&stale, &partial).expect("identity"),
1146 "the stale file was reused"
1147 );
1148 let mut bytes = Vec::new();
1149 (&stale)
1150 .read_to_end(&mut bytes)
1151 .expect("read the stale file");
1152 assert_eq!(
1153 bytes, b"what a killed export left",
1154 "the stale file was emptied"
1155 );
1156 assert_eq!(fs::read(&partial).expect("new partial"), b"");
1157 drop(claimed);
1158 assert!(!partial.exists());
1159 }
1160
1161 #[test]
1162 #[cfg(unix)]
1163 fn a_stale_partial_that_is_a_second_name_of_the_output_leaves_the_output_intact() {
1164 let path = final_path("stale-second-name");
1165 let partial = partial_output_path(&path).expect("partial path");
1166 fs::write(&path, b"the published output").expect("output");
1167 fs::hard_link(&path, &partial).expect("second name");
1168 let claimed = PartialOutput::claim(&OutputTarget::replacing(&path), SeekNeed::Required)
1169 .expect("claim");
1170 assert_eq!(
1171 fs::read(&path).expect("output"),
1172 b"the published output",
1173 "claiming the partial changed the output"
1174 );
1175 assert!(!alpha2_os::same_file_at(&path, &partial).expect("identity"));
1176 drop(claimed);
1177 assert!(!partial.exists());
1178 assert_eq!(fs::read(&path).expect("output"), b"the published output");
1179 fs::remove_file(&path).expect("clean up");
1180 }
1181
1182 #[test]
1183 fn a_held_partial_is_refused_after_a_short_wait() {
1184 let path = final_path("held-refused");
1185 let target = OutputTarget::from(&path);
1186 let held = PartialOutput::claim(&target, SeekNeed::Required).expect("holder");
1187 let started = Instant::now();
1188 let refused = PartialOutput::claim(&target, SeekNeed::Required);
1189 let waited = started.elapsed();
1190 assert!(
1191 matches!(&refused, Err(EncodeError::OutputInUse(message)) if message.starts_with(PARTIAL_OUTPUT_IN_USE)),
1192 "{:?}",
1193 refused.err()
1194 );
1195 assert!(waited >= CLAIM_IN_USE_WAIT, "refused after {waited:?}");
1196 drop(held);
1197 }
1198
1199 #[test]
1200 fn a_partial_released_within_the_wait_is_claimed() {
1201 let path = final_path("held-released");
1202 let partial = partial_output_path(&path).expect("partial path");
1203 let holder = OpenOptions::new()
1204 .read(true)
1205 .write(true)
1206 .create_new(true)
1207 .open(&partial)
1208 .expect("holder's partial");
1209 assert!(alpha2_os::try_lock_exclusive(&holder).expect("holder's lock"));
1210 let released = std::thread::spawn(move || {
1211 std::thread::sleep(Duration::from_millis(50));
1212 drop(holder);
1213 });
1214 let claimed = PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required)
1215 .expect("claimed once the holder let go");
1216 released.join().expect("holder");
1217 drop(claimed);
1218 assert!(!partial.exists());
1219 }
1220
1221 #[test]
1222 fn a_name_near_the_limit_gets_a_shortened_partial_name() {
1223 let directory = std::env::temp_dir();
1224 let limit = alpha2_os::name_limit(&directory);
1225 let syllables = match limit.unit {
1229 alpha2_os::NameUnit::Bytes => 81,
1230 alpha2_os::NameUnit::Utf16 => 243,
1231 };
1232 let name = format!("{}.mp4", "\u{d55c}".repeat(syllables));
1233 assert_eq!(limit.length(&name), 247);
1234 let path = directory.join(&name);
1235 let _ = fs::remove_file(&path);
1236 let partial = partial_output_path(&path).expect("partial path");
1237 assert_eq!(
1238 partial,
1239 partial_output_path(&path).expect("again"),
1240 "deterministic"
1241 );
1242 let siblings = [
1243 partial.clone(),
1244 sibling_path(&path, &kept_tag(1)).expect("kept"),
1245 sibling_path(&path, &kept_tag(KEPT_NAMES)).expect("last kept"),
1246 sibling_path(&path, TAG_PROBE[0]).expect("probe"),
1247 sibling_path(&path, TAG_PROBE[1]).expect("probe"),
1248 ];
1249 for sibling in &siblings {
1250 let sibling_name = sibling
1251 .file_name()
1252 .and_then(|name| name.to_str())
1253 .expect("name");
1254 assert!(
1255 limit.length(sibling_name) <= limit.max.min(255),
1256 "{sibling_name} is {} long",
1257 limit.length(sibling_name)
1258 );
1259 assert!(sibling_name.contains('~'), "{sibling_name}");
1260 }
1261 let partial_name = partial
1262 .file_name()
1263 .and_then(|name| name.to_str())
1264 .expect("name");
1265 assert!(partial_name.starts_with('\u{d55c}'), "{partial_name}");
1266 assert!(partial_name.ends_with(".a2partial.mp4"), "{partial_name}");
1267
1268 let mut claimed =
1269 PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required).expect("claim");
1270 assert!(partial.exists());
1271 drop(claimed.lock.take());
1273 claimed.state = PartialState::Kept;
1274 drop(claimed);
1275 assert!(remove_stale_partial_output(&path).expect("stale partial"));
1276 assert!(!partial.exists());
1277
1278 let short = directory.join("short.mp4");
1279 assert_eq!(
1280 partial_output_path(&short).expect("partial path"),
1281 directory.join("short.a2partial.mp4")
1282 );
1283 }
1284
1285 #[test]
1286 fn a_discarded_partial_leaves_nothing_even_when_its_name_is_recreated_on_close() {
1287 let path = final_path("recreated-on-close");
1289 let partial = partial_output_path(&path).expect("partial path");
1290 let closed_as = std::rc::Rc::new(RefCell::new(None));
1291 let record = std::rc::Rc::clone(&closed_as);
1292 hooks::ON_LOCK_CLOSE.set(Some(Box::new(move |name: &Path| {
1293 fs::write(name, b"written back").expect("write back");
1294 *record.borrow_mut() = Some(name.to_owned());
1295 })));
1296 let claimed =
1297 PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required).expect("claim");
1298 drop(claimed);
1299 hooks::ON_LOCK_CLOSE.set(None);
1300 let closed_as = closed_as.borrow().clone().expect("the lock closed");
1301 assert_ne!(
1302 closed_as, partial,
1303 "the lock closed under the partial's name"
1304 );
1305 assert!(!partial.exists(), "the partial name came back");
1306 assert!(
1307 !closed_as.exists(),
1308 "the written-back discard name was left"
1309 );
1310 }
1311}