use std::fs::{self, File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use alpha2_ffmpeg_bridge::{CreateOutputError, OutputFile};
use alpha2_media::{CancelToken, Container, EncodeError};
use alpha2_os::{NameLimit, NameUnit, PublishMethod, Published, RetryPolicy};
pub const PARTIAL_OUTPUT_IN_USE: &str = "another export is writing this output";
const TAG_PARTIAL: &str = "a2partial";
const TAG_KEPT: &str = "a2kept";
const TAG_PROBE: [&str; 2] = ["a2probe1", "a2probe2"];
const KEPT_NAMES: u32 = 99;
pub(crate) const PUBLISH_RETRY_BUDGET: Duration = Duration::from_secs(10);
const PROBE_RETRY_BUDGET: Duration = Duration::from_secs(2);
const CLAIM_IN_USE_WAIT: Duration = Duration::from_millis(250);
const CLAIM_IN_USE_POLL: Duration = Duration::from_millis(10);
const STAGING_COPY_CHUNK: usize = 4 << 20;
fn output_error(error: impl std::fmt::Display) -> EncodeError {
EncodeError::Output(error.to_string())
}
fn output_error_at(path: &Path, error: impl std::fmt::Display) -> EncodeError {
EncodeError::Output(format!("{}: {error}", path.display()))
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum ExistingOutput {
#[default]
Refuse,
Replace,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OutputTarget {
pub path: PathBuf,
pub existing: ExistingOutput,
}
impl OutputTarget {
#[must_use]
pub fn replacing(path: impl Into<PathBuf>) -> Self {
Self {
path: path.into(),
existing: ExistingOutput::Replace,
}
}
}
impl From<PathBuf> for OutputTarget {
fn from(path: PathBuf) -> Self {
Self {
path,
existing: ExistingOutput::Refuse,
}
}
}
impl From<&Path> for OutputTarget {
fn from(path: &Path) -> Self {
path.to_owned().into()
}
}
impl From<&PathBuf> for OutputTarget {
fn from(path: &PathBuf) -> Self {
path.clone().into()
}
}
impl From<String> for OutputTarget {
fn from(path: String) -> Self {
PathBuf::from(path).into()
}
}
impl From<&String> for OutputTarget {
fn from(path: &String) -> Self {
PathBuf::from(path).into()
}
}
impl From<&str> for OutputTarget {
fn from(path: &str) -> Self {
PathBuf::from(path).into()
}
}
impl From<&OutputTarget> for OutputTarget {
fn from(target: &OutputTarget) -> Self {
target.clone()
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PublishReport {
pub method: PublishMethod,
pub staged_locally: bool,
pub retries: u32,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum SeekNeed {
Required,
NotNeeded,
}
const CLAIM_ATTEMPTS: usize = 8;
fn parent_dir(final_path: &Path) -> &Path {
final_path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or(Path::new("."))
}
fn fnv1a_64(bytes: &[u8]) -> u64 {
let mut hash = 0xcbf2_9ce4_8422_2325_u64;
for byte in bytes {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
hash
}
fn sibling_name(name: &str, stem: &str, extension: &str, tag: &str, limit: NameLimit) -> String {
let named = |stem: &str, tag: &str| format!("{stem}.{tag}.{extension}");
if limit.length(&named(stem, TAG_PARTIAL)) <= limit.max {
return named(stem, tag);
}
let hash = format!("~{:08x}", fnv1a_64(name.as_bytes()) & 0xffff_ffff);
let room = limit
.max
.saturating_sub(limit.length(&named(&hash, TAG_PARTIAL)));
let mut used = 0;
let mut prefix = 0;
for (index, character) in stem.char_indices() {
let units = match limit.unit {
NameUnit::Bytes => character.len_utf8(),
NameUnit::Utf16 => character.len_utf16(),
};
if used + units > room {
break;
}
used += units;
prefix = index + character.len_utf8();
}
named(&format!("{}{hash}", &stem[..prefix]), tag)
}
fn sibling_path(final_path: &Path, tag: &str) -> Result<PathBuf, EncodeError> {
let name = final_path
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| EncodeError::Output("output path has no UTF-8 file name".into()))?;
let extension = final_path
.extension()
.and_then(|extension| extension.to_str())
.unwrap_or("bin");
let stem = final_path
.file_stem()
.and_then(|stem| stem.to_str())
.unwrap_or(name);
let limit = alpha2_os::name_limit(parent_dir(final_path));
Ok(final_path.with_file_name(sibling_name(name, stem, extension, tag, limit)))
}
pub(crate) fn partial_output_path(final_path: &Path) -> Result<PathBuf, EncodeError> {
sibling_path(final_path, TAG_PARTIAL)
}
fn kept_tag(n: u32) -> String {
if n <= 1 {
TAG_KEPT.to_owned()
} else {
format!("{TAG_KEPT}{n}")
}
}
static NEXT_SCRATCH: AtomicU64 = AtomicU64::new(0);
fn discard_path(partial: &Path) -> PathBuf {
let n = NEXT_SCRATCH.fetch_add(1, Ordering::Relaxed);
partial.with_file_name(format!(".a2del-{}-{n}", std::process::id()))
}
fn in_use(partial: &Path) -> EncodeError {
EncodeError::OutputInUse(format!(
"{PARTIAL_OUTPUT_IN_USE}: its partial file {} is held by an export still running",
partial.display()
))
}
type Lock<'a> = &'a mut dyn FnMut(&File) -> std::io::Result<bool>;
fn lock_partial(
path: &Path,
create: bool,
lock: Lock<'_>,
) -> Result<Option<(File, bool)>, EncodeError> {
let at = |error: std::io::Error| output_error_at(path, error);
let started = Instant::now();
let mut changes = 0;
while changes < CLAIM_ATTEMPTS {
let Some((file, created)) = open_partial(path, create).map_err(at)? else {
if create {
changes += 1;
continue;
}
return Ok(None);
};
let locked = match lock(&file) {
Ok(locked) => locked,
Err(error) => {
if created && alpha2_os::same_file(&file, path).unwrap_or(false) {
let _ = fs::remove_file(path);
}
return Err(at(error));
}
};
if !locked {
if started.elapsed() < CLAIM_IN_USE_WAIT {
drop(file);
std::thread::sleep(CLAIM_IN_USE_POLL);
continue;
}
return Err(in_use(path));
}
if alpha2_os::same_file(&file, path).map_err(at)? {
return Ok(Some((file, created)));
}
changes += 1;
}
Err(EncodeError::Output(format!(
"the partial file {} kept changing while it was claimed",
path.display()
)))
}
fn open_partial(path: &Path, create: bool) -> std::io::Result<Option<(File, bool)>> {
let mut options = OpenOptions::new();
options.read(true).write(true);
if create {
match options.clone().create_new(true).open(path) {
Ok(file) => return Ok(Some((file, true))),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => return Err(error),
}
}
match options.open(path) {
Ok(file) => Ok(Some((file, false))),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(error),
}
}
fn other_names(file: &File) -> String {
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
match file.metadata().map(|metadata| metadata.nlink()) {
Ok(links) if links > 1 => format!(" (it had {} other names)", links - 1),
_ => String::new(),
}
}
#[cfg(not(unix))]
{
let _ = file;
String::new()
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum PartialState {
Writing,
Finished,
Published,
Kept,
Discarded,
}
pub(crate) struct PartialOutput {
target: OutputTarget,
path: PathBuf,
lock: Option<File>,
staging: Option<PathBuf>,
state: PartialState,
}
impl PartialOutput {
pub(crate) fn claim(target: &OutputTarget, seek: SeekNeed) -> Result<Self, EncodeError> {
match target.existing {
ExistingOutput::Refuse => {
if target.path.exists() {
return Err(EncodeError::Output(
"refusing to overwrite an existing output file".into(),
));
}
}
ExistingOutput::Replace => {
replaceable(&target.path).map_err(EncodeError::Output)?;
#[cfg(windows)]
warn_if_held(&target.path);
}
}
let mut claimed = Self::claim_partial_locking(target, &mut alpha2_os::try_lock_exclusive)?;
if !claimed.probe(seek)? && seek == SeekNeed::Required {
let staging = create_staging(&target.path)?;
alpha2_os::diagnostic!(
"alpha2: {} cannot rewrite a file at an earlier offset; this export is written to {} and copied there when it is finished",
parent_dir(&target.path).display(),
staging.display()
);
claimed.staging = Some(staging);
}
Ok(claimed)
}
fn claim_partial_locking(target: &OutputTarget, lock: Lock<'_>) -> Result<Self, EncodeError> {
let path = partial_output_path(&target.path)?;
for _ in 0..CLAIM_ATTEMPTS {
let (file, created) = lock_partial(&path, true, lock)?
.ok_or_else(|| EncodeError::Output("the partial file was not created".into()))?;
if created {
return Ok(Self {
target: target.clone(),
path,
lock: Some(file),
staging: None,
state: PartialState::Writing,
});
}
alpha2_os::diagnostic!(
"alpha2: replacing {}, a partial file no running export holds{}",
path.display(),
other_names(&file)
);
match fs::remove_file(&path) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(output_error_at(&path, error)),
}
drop(file);
}
Err(EncodeError::Output(format!(
"the partial file {} kept changing while it was claimed",
path.display()
)))
}
fn probe(&self, seek: SeekNeed) -> Result<bool, EncodeError> {
let first = sibling_path(&self.target.path, TAG_PROBE[0])?;
let second = sibling_path(&self.target.path, TAG_PROBE[1])?;
for probe in [&first, &second] {
match fs::remove_file(probe) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(output_error_at(probe, error)),
}
}
let probed = self.probe_with(&first, &second, seek);
let _ = fs::remove_file(&first);
let _ = fs::remove_file(&second);
probed
}
fn probe_with(&self, first: &Path, second: &Path, seek: SeekNeed) -> Result<bool, EncodeError> {
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(first)
.map_err(|error| output_error_at(first, error))?;
let seekable = match seek {
SeekNeed::NotNeeded => true,
SeekNeed::Required => match rewrite_at_start(&mut file) {
Ok(()) => !hooks::not_seekable(),
Err(error) if cannot_seek(&error) => false,
Err(error) => return Err(output_error_at(first, error)),
},
};
drop(file);
let keep_trying = || true;
publish_no_replace(
first,
second,
&RetryPolicy {
budget: PROBE_RETRY_BUDGET,
on_wait: &keep_trying,
},
)
.map_err(|error| {
EncodeError::Output(format!(
"{} cannot take a finished file without replacing one: {error}",
parent_dir(&self.target.path).display()
))
})?;
Ok(seekable)
}
pub(crate) fn write_path(&self) -> &Path {
self.staging.as_deref().unwrap_or(&self.path)
}
pub(crate) fn open_muxer(&self, container: Container) -> Result<OutputFile, EncodeError> {
OutputFile::create(self.write_path(), crate::convert::muxer_name(container)).map_err(
|error| match error {
CreateOutputError::Path(error) => {
EncodeError::Output(format!("output path {error}"))
}
CreateOutputError::Ffmpeg(error) => output_error(error),
},
)
}
pub(crate) fn open_file(&self) -> Result<File, EncodeError> {
OpenOptions::new()
.write(true)
.open(self.write_path())
.map_err(output_error)
}
pub(crate) fn explain(&self, error: EncodeError) -> EncodeError {
match (error, &self.staging) {
(EncodeError::Adapter(message), Some(staging)) => EncodeError::Adapter(format!(
"{message}; the output folder cannot be written in place, so this export was written to the staging file {} first",
staging.display()
)),
(error, _) => error,
}
}
pub(crate) fn finish(&mut self, cancel: &CancelToken) -> Result<PublishReport, EncodeError> {
if self.state != PartialState::Writing {
return Err(EncodeError::Adapter(
"the output was already finished".into(),
));
}
self.state = PartialState::Finished;
let staged_locally = self.staging.is_some();
if let Some(staging) = self.staging.clone()
&& let Err(error) = self.copy_staging(&staging)
{
return Err(self.keep_staging(staging, &error));
}
let beat_unless_cancelled = || {
crate::encode::beat_encode_call();
!cancel.is_cancelled()
};
let retry = RetryPolicy {
budget: PUBLISH_RETRY_BUDGET,
on_wait: &beat_unless_cancelled,
};
match self.sync_and_publish(&retry) {
Ok(published) => {
self.state = PartialState::Published;
drop(self.lock.take());
self.remove_staging();
#[cfg(unix)]
if let Err(error) = sync_directory(parent_dir(&self.target.path)) {
alpha2_os::diagnostic!(
"alpha2: {} is in place, but syncing its folder failed: {error}",
self.target.path.display()
);
}
Ok(PublishReport {
method: published.method,
staged_locally,
retries: published.retries,
})
}
Err(_) if cancel.is_cancelled() => {
self.state = PartialState::Writing;
self.discard()?;
Err(EncodeError::Cancelled)
}
Err(message) => Err(self.keep(message)),
}
}
fn copy_staging(&self, staging: &Path) -> Result<(), String> {
let mut source = File::open(staging).map_err(|error| error.to_string())?;
let mut lock = self
.lock
.as_ref()
.ok_or_else(|| "the partial file is no longer held".to_owned())?;
lock.seek(SeekFrom::Start(0))
.map_err(|error| error.to_string())?;
let mut chunk = vec![0_u8; STAGING_COPY_CHUNK];
loop {
crate::encode::beat_encode_call();
let read = source
.read(&mut chunk)
.map_err(|error| format!("reading {}: {error}", staging.display()))?;
if read == 0 {
return Ok(());
}
lock.write_all(&chunk[..read])
.map_err(|error| format!("writing {}: {error}", self.path.display()))?;
}
}
fn sync_and_publish(&self, retry: &RetryPolicy<'_>) -> Result<Published, String> {
let final_path = self.target.path.as_path();
let lock = self
.lock
.as_ref()
.ok_or_else(|| "the partial file is no longer held".to_owned())?;
lock.sync_all()
.map_err(|error| format!("syncing {}: {error}", self.path.display()))?;
hooks::record("sync");
let published = match self.target.existing {
ExistingOutput::Refuse => {
publish_no_replace(&self.path, final_path, retry).map_err(|error| {
if error.kind() == std::io::ErrorKind::AlreadyExists {
format!(
"a file appeared at {} while this export ran; it was left as it is",
final_path.display()
)
} else {
format!("{}: {error}", final_path.display())
}
})
}
ExistingOutput::Replace => replaceable(final_path).and_then(|()| {
alpha2_os::publish_replace(&self.path, final_path, retry)
.map_err(|error| format!("{}: {error}", final_path.display()))
}),
}?;
hooks::record("rename");
Ok(published)
}
fn keep(&mut self, message: String) -> EncodeError {
let kept = self.move_aside(&self.path);
self.state = PartialState::Kept;
drop(self.lock.take());
self.remove_staging();
match kept {
Ok(kept) => EncodeError::Publish {
message,
kept: Some(kept),
},
Err(error) => EncodeError::Publish {
message: format!(
"{message}; moving it aside failed too ({error}), so it is left as {}, which the next export to this path removes",
self.path.display()
),
kept: None,
},
}
}
fn keep_staging(&mut self, staging: PathBuf, error: &str) -> EncodeError {
self.staging = None;
self.state = PartialState::Writing;
let _ = self.discard();
self.state = PartialState::Kept;
EncodeError::Publish {
message: format!(
"{}: copying the finished render into {} failed: {error}",
self.target.path.display(),
self.path.display()
),
kept: Some(staging),
}
}
fn move_aside(&self, source: &Path) -> Result<PathBuf, String> {
let beat = || {
crate::encode::beat_encode_call();
true
};
let retry = RetryPolicy {
budget: PUBLISH_RETRY_BUDGET,
on_wait: &beat,
};
for n in 1..=KEPT_NAMES {
let kept =
sibling_path(&self.target.path, &kept_tag(n)).map_err(|error| error.to_string())?;
match publish_no_replace(source, &kept, &retry) {
Ok(_) => return Ok(kept),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => return Err(format!("{}: {error}", kept.display())),
}
}
Err(format!(
"every kept name from {} to {} is taken",
kept_tag(1),
kept_tag(KEPT_NAMES)
))
}
fn remove_staging(&mut self) {
if let Some(staging) = self.staging.take() {
let _ = fs::remove_file(staging);
}
}
pub(crate) fn discard(&mut self) -> Result<(), EncodeError> {
match self.state {
PartialState::Writing => {}
PartialState::Finished => {
let kept = self.keep("the export stopped before its output was published".into());
alpha2_os::diagnostic!("alpha2: {kept}");
return Ok(());
}
PartialState::Published | PartialState::Kept | PartialState::Discarded => {
return Ok(());
}
}
self.state = PartialState::Discarded;
self.remove_staging();
let Some(lock) = self.lock.take() else {
return Ok(());
};
let discard = discard_path(&self.path);
let (removed, at) = if publish_no_replace(&self.path, &discard, &RetryPolicy::NONE).is_ok()
{
close_lock(lock, &discard);
let keep_trying = || true;
let retry = RetryPolicy {
budget: PUBLISH_RETRY_BUDGET,
on_wait: &keep_trying,
};
let removed =
alpha2_os::with_publish_retries(&retry, || fs::remove_file(&discard)).map(|_| ());
(removed, discard)
} else {
let removed = fs::remove_file(&self.path);
close_lock(lock, &self.path);
(removed, self.path.clone())
};
match removed {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(output_error_at(&at, error)),
}
}
}
impl Drop for PartialOutput {
fn drop(&mut self) {
let _ = self.discard();
}
}
fn close_lock(lock: File, name: &Path) {
drop(lock);
hooks::lock_closed(name);
}
fn rewrite_at_start(file: &mut File) -> std::io::Result<()> {
file.write_all(&[0, 0])?;
file.seek(SeekFrom::Start(0))?;
file.write_all(&[1])?;
file.sync_all()
}
fn cannot_seek(error: &std::io::Error) -> bool {
matches!(
error.kind(),
std::io::ErrorKind::NotSeekable
| std::io::ErrorKind::InvalidInput
| std::io::ErrorKind::Unsupported
)
}
fn create_staging(final_path: &Path) -> Result<PathBuf, EncodeError> {
let extension = final_path
.extension()
.and_then(|extension| extension.to_str())
.unwrap_or("bin");
for _ in 0..CLAIM_ATTEMPTS {
let n = NEXT_SCRATCH.fetch_add(1, Ordering::Relaxed);
let staging = std::env::temp_dir().join(format!(
"alpha2-staging-{}-{n}.{extension}",
std::process::id()
));
match OpenOptions::new()
.write(true)
.create_new(true)
.open(&staging)
{
Ok(_) => return Ok(staging),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => return Err(output_error_at(&staging, error)),
}
}
Err(EncodeError::Output(
"no staging file could be created in the temporary directory".into(),
))
}
#[cfg(windows)]
fn warn_if_held(final_path: &Path) {
use std::os::windows::fs::OpenOptionsExt;
const DELETE: u32 = 0x0001_0000;
const SHARE_READ_WRITE_DELETE: u32 = 0x1 | 0x2 | 0x4;
const ERROR_SHARING_VIOLATION: i32 = 32;
if let Err(error) = OpenOptions::new()
.access_mode(DELETE)
.share_mode(SHARE_READ_WRITE_DELETE)
.open(final_path)
&& error.raw_os_error() == Some(ERROR_SHARING_VIOLATION)
{
alpha2_os::diagnostic!(
"alpha2: {} is open in another program; the export will wait up to {} s for it when it finishes",
final_path.display(),
PUBLISH_RETRY_BUDGET.as_secs()
);
}
}
fn publish_no_replace(
from: &Path,
to: &Path,
retry: &RetryPolicy<'_>,
) -> std::io::Result<Published> {
if let Some(error) = hooks::publish_failure(to) {
return Err(error);
}
alpha2_os::publish_no_replace(from, to, retry)
}
#[cfg(unix)]
fn sync_directory(dir: &Path) -> std::io::Result<()> {
alpha2_os::sync_directory(dir)?;
hooks::record("directory sync");
Ok(())
}
fn replaceable(final_path: &Path) -> Result<(), String> {
match fs::symlink_metadata(final_path) {
Ok(metadata) if metadata.file_type().is_file() => Ok(()),
Ok(_) => Err(format!(
"refusing to replace {}: not a regular file",
final_path.display()
)),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(format!("{}: {error}", final_path.display())),
}
}
pub fn write_whole(
target: impl Into<OutputTarget>,
write: impl FnOnce(&mut File) -> Result<(), String>,
) -> Result<PublishReport, EncodeError> {
let target = target.into();
let mut partial = PartialOutput::claim(&target, SeekNeed::NotNeeded)?;
let mut file = partial.open_file()?;
write(&mut file).map_err(|error| output_error_at(partial.write_path(), error))?;
drop(file);
partial.finish(&CancelToken::default())
}
pub fn write_png(
target: impl Into<OutputTarget>,
width: u32,
height: u32,
rgba: &[u8],
) -> Result<PublishReport, EncodeError> {
let expected = u64::from(width) * u64::from(height) * 4;
if u64::try_from(rgba.len()).ok() != Some(expected) {
return Err(EncodeError::Adapter(format!(
"a {width}x{height} RGBA image is {expected} bytes, not {}",
rgba.len()
)));
}
write_whole(target, |file| {
let mut buffered = std::io::BufWriter::new(file);
let mut encoder = png::Encoder::new(&mut buffered, width, height);
encoder.set_color(png::ColorType::Rgba);
encoder.set_depth(png::BitDepth::Eight);
encoder.set_source_srgb(png::SrgbRenderingIntent::Perceptual);
let mut writer = encoder.write_header().map_err(|error| error.to_string())?;
writer
.write_image_data(rgba)
.map_err(|error| error.to_string())?;
writer.finish().map_err(|error| error.to_string())?;
buffered
.into_inner()
.map(|_| ())
.map_err(|error| error.error().to_string())
})
}
pub fn remove_stale_partial_output(path: impl AsRef<Path>) -> Result<bool, EncodeError> {
let partial = partial_output_path(path.as_ref())?;
let Some((file, _)) = lock_partial(&partial, false, &mut alpha2_os::try_lock_exclusive)? else {
return Ok(false);
};
let removed = fs::remove_file(&partial);
drop(file);
match removed {
Ok(()) => Ok(true),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(error) => Err(output_error(error)),
}
}
#[cfg(not(test))]
mod hooks {
use std::path::Path;
pub(super) fn publish_failure(_to: &Path) -> Option<std::io::Error> {
None
}
pub(super) fn not_seekable() -> bool {
false
}
pub(super) fn record(_event: &'static str) {}
pub(super) fn lock_closed(_name: &Path) {}
}
#[cfg(test)]
pub(crate) mod hooks {
use std::cell::{Cell, RefCell};
use std::path::{Path, PathBuf};
pub(crate) type LockCloseHook = Box<dyn Fn(&Path)>;
thread_local! {
pub(crate) static FAIL_PUBLISH_TO: RefCell<Option<(PathBuf, i32)>> =
const { RefCell::new(None) };
pub(crate) static NOT_SEEKABLE: Cell<bool> = const { Cell::new(false) };
pub(crate) static EVENTS: RefCell<Option<Vec<&'static str>>> =
const { RefCell::new(None) };
pub(crate) static ON_LOCK_CLOSE: RefCell<Option<LockCloseHook>> =
const { RefCell::new(None) };
}
pub(super) fn publish_failure(to: &Path) -> Option<std::io::Error> {
FAIL_PUBLISH_TO.with_borrow(|fail| {
fail.as_ref()
.filter(|(path, _)| path == to)
.map(|(_, code)| std::io::Error::from_raw_os_error(*code))
})
}
pub(super) fn not_seekable() -> bool {
NOT_SEEKABLE.get()
}
pub(super) fn record(event: &'static str) {
EVENTS.with_borrow_mut(|events| {
if let Some(events) = events {
events.push(event);
}
});
}
pub(super) fn lock_closed(name: &Path) {
ON_LOCK_CLOSE.with_borrow(|hook| {
if let Some(hook) = hook {
hook(name);
}
});
}
}
#[cfg(test)]
mod tests {
use std::cell::RefCell;
use std::io::{Read, Write};
use super::*;
fn final_path(label: &str) -> PathBuf {
let path =
std::env::temp_dir().join(format!("alpha2-claim-{label}-{}.mp4", std::process::id()));
let _ = fs::remove_file(&path);
let _ = fs::remove_file(partial_output_path(&path).expect("partial path"));
path
}
fn no_locks(_: &File) -> std::io::Result<bool> {
Err(std::io::Error::other("No locks available"))
}
#[test]
fn a_partial_the_claim_created_and_could_not_lock_is_removed() {
let path = final_path("unlockable");
let partial = partial_output_path(&path).expect("partial path");
let Err(error) =
PartialOutput::claim_partial_locking(&OutputTarget::from(&path), &mut no_locks)
else {
panic!("a claim without a lock succeeded");
};
assert!(
matches!(&error, EncodeError::Output(message) if message.contains(&partial.display().to_string())),
"{error}"
);
assert!(
!partial.exists(),
"the refused claim left the partial it created"
);
}
#[test]
fn a_partial_that_was_there_is_left_when_the_lock_fails() {
let path = final_path("unlockable-existing");
let partial = partial_output_path(&path).expect("partial path");
fs::write(&partial, b"another export's").expect("existing partial");
assert!(
PartialOutput::claim_partial_locking(&OutputTarget::from(&path), &mut no_locks)
.is_err()
);
assert_eq!(
fs::read(&partial).expect("still there"),
b"another export's"
);
fs::remove_file(&partial).expect("clean up");
}
#[test]
fn a_partial_its_holder_removes_while_it_is_claimed_is_claimed_new() {
let _child = crate::tests::child_process_lock();
let path = final_path("removed-while-claimed");
let partial = partial_output_path(&path).expect("partial path");
let target = OutputTarget::from(&path);
let mut holder = Some(PartialOutput::claim(&target, SeekNeed::Required).expect("holder"));
let claimed = PartialOutput::claim_partial_locking(&target, &mut |file| {
drop(holder.take());
alpha2_os::try_lock_exclusive(file)
})
.expect("claimed after the holder finished");
assert!(holder.is_none());
let mut writer = claimed.open_file().expect("the claimed partial opens");
writer.write_all(b"new export").expect("write");
drop(writer);
let mut read = String::new();
File::open(&partial)
.and_then(|mut file| file.read_to_string(&mut read))
.expect("read back");
assert_eq!(read, "new export");
drop(claimed);
assert!(!partial.exists());
}
#[test]
fn a_stale_partial_is_replaced_not_emptied() {
let path = final_path("stale-replaced");
let partial = partial_output_path(&path).expect("partial path");
fs::write(&partial, b"what a killed export left").expect("stale partial");
let stale = File::open(&partial).expect("the stale file");
let claimed =
PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required).expect("claim");
assert!(
!alpha2_os::same_file(&stale, &partial).expect("identity"),
"the stale file was reused"
);
let mut bytes = Vec::new();
(&stale)
.read_to_end(&mut bytes)
.expect("read the stale file");
assert_eq!(
bytes, b"what a killed export left",
"the stale file was emptied"
);
assert_eq!(fs::read(&partial).expect("new partial"), b"");
drop(claimed);
assert!(!partial.exists());
}
#[test]
#[cfg(unix)]
fn a_stale_partial_that_is_a_second_name_of_the_output_leaves_the_output_intact() {
let path = final_path("stale-second-name");
let partial = partial_output_path(&path).expect("partial path");
fs::write(&path, b"the published output").expect("output");
fs::hard_link(&path, &partial).expect("second name");
let claimed = PartialOutput::claim(&OutputTarget::replacing(&path), SeekNeed::Required)
.expect("claim");
assert_eq!(
fs::read(&path).expect("output"),
b"the published output",
"claiming the partial changed the output"
);
assert!(!alpha2_os::same_file_at(&path, &partial).expect("identity"));
drop(claimed);
assert!(!partial.exists());
assert_eq!(fs::read(&path).expect("output"), b"the published output");
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_held_partial_is_refused_after_a_short_wait() {
let path = final_path("held-refused");
let target = OutputTarget::from(&path);
let held = PartialOutput::claim(&target, SeekNeed::Required).expect("holder");
let started = Instant::now();
let refused = PartialOutput::claim(&target, SeekNeed::Required);
let waited = started.elapsed();
assert!(
matches!(&refused, Err(EncodeError::OutputInUse(message)) if message.starts_with(PARTIAL_OUTPUT_IN_USE)),
"{:?}",
refused.err()
);
assert!(waited >= CLAIM_IN_USE_WAIT, "refused after {waited:?}");
drop(held);
}
#[test]
fn a_partial_released_within_the_wait_is_claimed() {
let path = final_path("held-released");
let partial = partial_output_path(&path).expect("partial path");
let holder = OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&partial)
.expect("holder's partial");
assert!(alpha2_os::try_lock_exclusive(&holder).expect("holder's lock"));
let released = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(50));
drop(holder);
});
let claimed = PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required)
.expect("claimed once the holder let go");
released.join().expect("holder");
drop(claimed);
assert!(!partial.exists());
}
#[test]
fn a_name_near_the_limit_gets_a_shortened_partial_name() {
let directory = std::env::temp_dir();
let limit = alpha2_os::name_limit(&directory);
let syllables = match limit.unit {
alpha2_os::NameUnit::Bytes => 81,
alpha2_os::NameUnit::Utf16 => 243,
};
let name = format!("{}.mp4", "\u{d55c}".repeat(syllables));
assert_eq!(limit.length(&name), 247);
let path = directory.join(&name);
let _ = fs::remove_file(&path);
let partial = partial_output_path(&path).expect("partial path");
assert_eq!(
partial,
partial_output_path(&path).expect("again"),
"deterministic"
);
let siblings = [
partial.clone(),
sibling_path(&path, &kept_tag(1)).expect("kept"),
sibling_path(&path, &kept_tag(KEPT_NAMES)).expect("last kept"),
sibling_path(&path, TAG_PROBE[0]).expect("probe"),
sibling_path(&path, TAG_PROBE[1]).expect("probe"),
];
for sibling in &siblings {
let sibling_name = sibling
.file_name()
.and_then(|name| name.to_str())
.expect("name");
assert!(
limit.length(sibling_name) <= limit.max.min(255),
"{sibling_name} is {} long",
limit.length(sibling_name)
);
assert!(sibling_name.contains('~'), "{sibling_name}");
}
let partial_name = partial
.file_name()
.and_then(|name| name.to_str())
.expect("name");
assert!(partial_name.starts_with('\u{d55c}'), "{partial_name}");
assert!(partial_name.ends_with(".a2partial.mp4"), "{partial_name}");
let mut claimed =
PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required).expect("claim");
assert!(partial.exists());
drop(claimed.lock.take());
claimed.state = PartialState::Kept;
drop(claimed);
assert!(remove_stale_partial_output(&path).expect("stale partial"));
assert!(!partial.exists());
let short = directory.join("short.mp4");
assert_eq!(
partial_output_path(&short).expect("partial path"),
directory.join("short.a2partial.mp4")
);
}
#[test]
fn a_discarded_partial_leaves_nothing_even_when_its_name_is_recreated_on_close() {
let path = final_path("recreated-on-close");
let partial = partial_output_path(&path).expect("partial path");
let closed_as = std::rc::Rc::new(RefCell::new(None));
let record = std::rc::Rc::clone(&closed_as);
hooks::ON_LOCK_CLOSE.set(Some(Box::new(move |name: &Path| {
fs::write(name, b"written back").expect("write back");
*record.borrow_mut() = Some(name.to_owned());
})));
let claimed =
PartialOutput::claim(&OutputTarget::from(&path), SeekNeed::Required).expect("claim");
drop(claimed);
hooks::ON_LOCK_CLOSE.set(None);
let closed_as = closed_as.borrow().clone().expect("the lock closed");
assert_ne!(
closed_as, partial,
"the lock closed under the partial's name"
);
assert!(!partial.exists(), "the partial name came back");
assert!(
!closed_as.exists(),
"the written-back discard name was left"
);
}
}