use std::ffi::{OsStr, OsString, c_void};
use std::fs::{File, OpenOptions};
use std::io;
use std::os::windows::{ffi::OsStrExt, io::AsRawHandle};
use std::path::{Component, Path, Prefix};
use std::time::{Duration, Instant, SystemTime};
const ERROR_ACCESS_DENIED: i32 = 5;
const DENIED_RETRIES: u32 = 8;
const PUBLISHING: Duration = Duration::from_secs(2);
const FRESH: Duration = Duration::from_secs(10);
const AHEAD: Duration = Duration::from_secs(1);
pub(super) fn open_record(open: &OpenOptions, path: &Path) -> io::Result<File> {
retry_open(
|| open.open(path),
|| publishing(path),
Instant::now() + PUBLISHING,
)
}
fn retry_open<T>(
mut open: impl FnMut() -> io::Result<T>,
mut publishing: impl FnMut() -> bool,
deadline: Instant,
) -> io::Result<T> {
let mut denied = 0;
let mut rechecked = false;
let mut opened = None;
let mut missing = None;
crate::backoff::poll(
|| {
loop {
match open() {
Ok(file) => {
opened = Some(file);
return Ok(true);
}
Err(error)
if error.raw_os_error() == Some(ERROR_ACCESS_DENIED)
&& denied < DENIED_RETRIES =>
{
denied += 1;
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
if publishing() {
rechecked = false;
missing = Some(error);
return Ok(false);
}
if rechecked {
return Err(error);
}
rechecked = true;
}
Err(error) => return Err(error),
}
}
},
Some(deadline),
)?;
opened.ok_or_else(|| missing.unwrap_or_else(|| io::ErrorKind::NotFound.into()))
}
fn publishing(path: &Path) -> bool {
let (Some(parent), Some(record)) = (path.parent(), path.file_name()) else {
return false;
};
let parent = if parent.as_os_str().is_empty() {
Path::new(".")
} else {
parent
};
let Ok(entries) = std::fs::read_dir(parent) else {
return false;
};
let now = SystemTime::now();
entries.flatten().any(|entry| {
is_temporary(&entry.file_name(), record)
&& entry
.metadata()
.and_then(|metadata| metadata.created())
.is_ok_and(|created| fresh(created, now))
})
}
fn fresh(created: SystemTime, now: SystemTime) -> bool {
match now.duration_since(created) {
Ok(age) => age < FRESH,
Err(ahead) => ahead.duration() <= AHEAD,
}
}
fn is_temporary(name: &OsStr, record: &OsStr) -> bool {
let name: Vec<u16> = name.encode_wide().collect();
let record: Vec<u16> = record.encode_wide().collect();
let Some(rest) = name.strip_prefix(&[u16::from(b'.')]) else {
return false;
};
let (Some(named), Some(rest)) = (rest.get(..record.len()), rest.get(record.len()..)) else {
return false;
};
let Ok(rest) = String::from_utf16(rest) else {
return false;
};
let Some(numbers) = rest
.strip_prefix('.')
.and_then(|rest| rest.strip_suffix(".tmp"))
else {
return false;
};
let mut parts = numbers.split('.');
let number = |part: Option<&str>| {
part.is_some_and(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit()))
};
number(parts.next())
&& number(parts.next())
&& parts.next().is_none()
&& same_name(named, &record)
}
fn same_name(a: &[u16], b: &[u16]) -> bool {
let (Ok(a_length), Ok(b_length)) = (i32::try_from(a.len()), i32::try_from(b.len())) else {
return false;
};
let compared = unsafe { CompareStringOrdinal(a.as_ptr(), a_length, b.as_ptr(), b_length, 1) };
compared == CSTR_EQUAL
}
const CSTR_EQUAL: i32 = 2;
const GENERIC_WRITE: u32 = 0x4000_0000;
const DELETE: u32 = 0x0001_0000;
pub(super) const ACCESS: u32 = GENERIC_WRITE | DELETE;
const FILE_RENAME_INFO_EX: i32 = 22;
const FILE_RENAME_FLAG_REPLACE_IF_EXISTS: u32 = 0x1;
const FILE_RENAME_FLAG_POSIX_SEMANTICS: u32 = 0x2;
const ERROR_INVALID_FUNCTION: i32 = 1;
const ERROR_NOT_SUPPORTED: i32 = 50;
const ERROR_INVALID_PARAMETER: i32 = 87;
#[repr(C)]
struct FileRenameInfo {
flags: u32,
root_directory: *mut c_void,
file_name_length: u32,
file_name: [u16; 1],
}
#[link(name = "kernel32")]
unsafe extern "system" {
fn CompareStringOrdinal(
first: *const u16,
first_length: i32,
second: *const u16,
second_length: i32,
ignore_case: i32,
) -> i32;
fn SetFileInformationByHandle(
file: *mut c_void,
class: i32,
information: *const c_void,
size: u32,
) -> i32;
}
pub(super) fn replace(file: File, from: &Path, to: &Path, durable: bool) -> io::Result<()> {
replace_with(file, from, to, durable, rename_posix, File::sync_all)
}
fn replace_with(
file: File,
from: &Path,
to: &Path,
durable: bool,
rename: impl FnOnce(&File, &Path) -> io::Result<()>,
flush: impl FnOnce(&File) -> io::Result<()>,
) -> io::Result<()> {
match rename(&file, to) {
Err(error)
if matches!(
error.raw_os_error(),
Some(ERROR_INVALID_FUNCTION | ERROR_NOT_SUPPORTED | ERROR_INVALID_PARAMETER)
) =>
{
drop(file);
std::fs::rename(from, to)?;
if durable {
flush(&open_record(OpenOptions::new().write(true), to)?)?;
}
Ok(())
}
Ok(()) if durable => flush(&file),
result => result,
}
}
fn rename_posix(file: &File, to: &Path) -> io::Result<()> {
let name: Vec<u16> = verbatim(&std::path::absolute(to)?).encode_wide().collect();
let too_long = || io::Error::new(io::ErrorKind::InvalidInput, "record path is too long");
let name_bytes = u32::try_from(name.len() * 2).map_err(|_| too_long())?;
let offset = std::mem::offset_of!(FileRenameInfo, file_name);
let size = offset + name.len() * 2 + 2;
let mut buffer = vec![0usize; size.div_ceil(std::mem::size_of::<usize>())];
let info = buffer.as_mut_ptr().cast::<FileRenameInfo>();
unsafe {
(&raw mut (*info).flags)
.write(FILE_RENAME_FLAG_REPLACE_IF_EXISTS | FILE_RENAME_FLAG_POSIX_SEMANTICS);
(&raw mut (*info).root_directory).write(std::ptr::null_mut());
(&raw mut (*info).file_name_length).write(name_bytes);
(&raw mut (*info).file_name)
.cast::<u16>()
.copy_from_nonoverlapping(name.as_ptr(), name.len());
}
let size = u32::try_from(size).map_err(|_| too_long())?;
let renamed = unsafe {
SetFileInformationByHandle(
file.as_raw_handle(),
FILE_RENAME_INFO_EX,
buffer.as_ptr().cast(),
size,
)
};
if renamed == 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
fn verbatim(path: &Path) -> OsString {
match path.components().next() {
Some(Component::Prefix(prefix)) if matches!(prefix.kind(), Prefix::Disk(_)) => {
let mut verbatim = OsString::from(r"\\?\");
verbatim.push(path.as_os_str());
verbatim
}
_ => path.as_os_str().to_owned(),
}
}
#[cfg(test)]
mod tests {
use std::cell::RefCell;
use std::collections::VecDeque;
use std::fs::File;
use std::io::{self, Read, Write};
use std::os::windows::{fs::OpenOptionsExt, io::AsRawHandle};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant, SystemTime};
fn scripted(
opens: Vec<io::Result<u32>>,
publishing: Vec<bool>,
) -> (io::Result<u32>, usize, usize) {
let mut opens = VecDeque::from(opens);
let mut publishing = VecDeque::from(publishing);
let result = super::retry_open(
|| opens.pop_front().expect("one open too many"),
|| publishing.pop_front().expect("one check too many"),
Instant::now() + Duration::from_secs(10),
);
(result, opens.len(), publishing.len())
}
fn missing() -> io::Result<u32> {
Err(io::ErrorKind::NotFound.into())
}
#[test]
fn a_missing_record_is_waited_for_while_it_is_published() {
let (opened, opens, checks) = scripted(vec![missing(), missing(), Ok(7)], vec![true, true]);
assert_eq!(opened.unwrap(), 7);
assert_eq!((opens, checks), (0, 0));
}
#[test]
fn without_a_publication_a_missing_record_is_opened_once_more() {
let (opened, opens, checks) = scripted(vec![missing(), Ok(7)], vec![false]);
assert_eq!(opened.unwrap(), 7);
assert_eq!((opens, checks), (0, 0));
let (opened, opens, checks) = scripted(vec![missing(), missing()], vec![false, false]);
assert_eq!(opened.unwrap_err().kind(), io::ErrorKind::NotFound);
assert_eq!((opens, checks), (0, 0));
}
#[test]
fn a_record_published_after_a_wait_is_opened_once_more() {
let (opened, opens, checks) = scripted(
vec![missing(), missing(), missing(), Ok(7)],
vec![false, true, false],
);
assert_eq!(opened.unwrap(), 7);
assert_eq!((opens, checks), (0, 0));
}
#[test]
fn a_publication_is_waited_for_only_until_the_deadline() {
let start = Instant::now();
let deadline = start + Duration::from_millis(20);
let opened: io::Result<u32> = super::retry_open(missing, || true, deadline);
assert_eq!(opened.unwrap_err().kind(), io::ErrorKind::NotFound);
assert!(Instant::now() >= deadline);
}
#[test]
fn a_refused_open_is_retried_a_few_times() {
let denied = || Err(io::Error::from_raw_os_error(super::ERROR_ACCESS_DENIED));
let (opened, opens, _) = scripted(vec![denied(), denied(), Ok(7)], vec![]);
assert_eq!(opened.unwrap(), 7);
assert_eq!(opens, 0);
let mut opens: Vec<_> = (0..=super::DENIED_RETRIES).map(|_| denied()).collect();
opens.push(Ok(7));
let (opened, opens, _) = scripted(opens, vec![]);
assert_eq!(
opened.unwrap_err().raw_os_error(),
Some(super::ERROR_ACCESS_DENIED)
);
assert_eq!(opens, 1);
}
#[test]
fn only_a_recent_temporary_marks_a_publication() {
let now = SystemTime::now();
assert!(super::fresh(now, now));
assert!(super::fresh(now - Duration::from_secs(9), now));
assert!(!super::fresh(now - Duration::from_secs(11), now));
assert!(super::fresh(now + Duration::from_millis(500), now));
assert!(!super::fresh(now + Duration::from_secs(3600), now));
}
#[test]
fn a_record_another_handle_holds_open_is_replaced_in_one_step() {
let root = tempfile::tempdir().unwrap();
let record = root.path().join("record");
std::fs::write(&record, b"old").unwrap();
let mut reader = std::fs::File::open(&record).unwrap();
let temporary = root.path().join("temporary");
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.access_mode(super::ACCESS)
.open(&temporary)
.unwrap();
file.write_all(b"new").unwrap();
super::rename_posix(&file, &record).unwrap();
drop(file);
assert_eq!(std::fs::read(&record).unwrap(), b"new");
assert!(!temporary.exists());
let mut old = Vec::new();
reader.read_to_end(&mut old).unwrap();
assert_eq!(old, b"old");
}
#[test]
fn only_this_records_temporaries_mark_a_publication() {
use std::ffi::OsStr;
let record = OsStr::new("discovery.json");
let temporary = |name: &str| super::is_temporary(OsStr::new(name), record);
assert!(temporary(".discovery.json.4321.0.tmp"));
assert!(temporary(".discovery.json.1.25.tmp"));
assert!(temporary(".Discovery.JSON.1.0.tmp"));
assert!(super::is_temporary(
OsStr::new(".ÉTAT.1.0.tmp"),
OsStr::new("état")
));
assert!(!super::is_temporary(
OsStr::new(".etat.1.0.tmp"),
OsStr::new("état")
));
for other in [
"discovery.json",
"discovery.json.1.0.tmp",
".discovery.json.tmp",
".discovery.json.1.tmp",
".discovery.json.1.2.3.tmp",
".discovery.json.1..tmp",
".discovery.json.x.0.tmp",
".discovery.json.1.0.tmp.old",
".discovery.jsonx.1.0.tmp",
".other.1.0.tmp",
] {
assert!(!temporary(other), "{other}");
}
}
#[test]
fn a_lasting_denial_is_returned() {
let root = tempfile::tempdir().unwrap();
let record = root.path().join("record");
std::fs::write(&record, b"old").unwrap();
let holder = std::fs::OpenOptions::new()
.access_mode(super::DELETE)
.open(&record)
.unwrap();
let delete: u8 = 1;
let marked = unsafe {
super::SetFileInformationByHandle(
holder.as_raw_handle(),
FILE_DISPOSITION_INFO,
(&raw const delete).cast(),
1,
)
};
assert_ne!(marked, 0, "{}", std::io::Error::last_os_error());
let mut read = std::fs::OpenOptions::new();
read.read(true);
let start = std::time::Instant::now();
let refused = super::open_record(&read, &record).unwrap_err();
assert_eq!(refused.raw_os_error(), Some(super::ERROR_ACCESS_DENIED));
assert!(start.elapsed() < std::time::Duration::from_secs(1));
drop(holder);
let missing = super::open_record(&read, &record).unwrap_err();
assert_eq!(missing.kind(), std::io::ErrorKind::NotFound);
}
const FILE_DISPOSITION_INFO: i32 = 4;
#[test]
fn a_file_system_without_posix_renames_gets_std_rename() {
let root = tempfile::tempdir().unwrap();
let record = root.path().join("record");
std::fs::write(&record, b"old").unwrap();
let temporary = root.path().join("temporary");
std::fs::write(&temporary, b"new").unwrap();
let file = std::fs::File::open(&temporary).unwrap();
let unsupported = |_: &std::fs::File, _: &std::path::Path| {
Err(std::io::Error::from_raw_os_error(
super::ERROR_NOT_SUPPORTED,
))
};
super::replace_with(file, &temporary, &record, false, unsupported, sync).unwrap();
assert_eq!(std::fs::read(&record).unwrap(), b"new");
assert!(!temporary.exists());
std::fs::write(&temporary, b"durable").unwrap();
let file = std::fs::File::open(&temporary).unwrap();
super::replace_with(file, &temporary, &record, true, unsupported, sync).unwrap();
assert_eq!(std::fs::read(&record).unwrap(), b"durable");
assert!(!temporary.exists());
let denied =
|_: &std::fs::File, _: &std::path::Path| Err(std::io::Error::from_raw_os_error(5));
std::fs::write(&temporary, b"newer").unwrap();
let file = std::fs::File::open(&temporary).unwrap();
assert!(super::replace_with(file, &temporary, &record, true, denied, sync).is_err());
assert_eq!(std::fs::read(&record).unwrap(), b"durable");
}
#[test]
fn a_durable_replacement_does_not_reopen_the_record_by_name() {
let root = tempfile::tempdir().unwrap();
let record = root.path().join("record");
std::fs::write(&record, b"old").unwrap();
let temporary = root.path().join("temporary");
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.access_mode(super::ACCESS)
.open(&temporary)
.unwrap();
file.write_all(b"new").unwrap();
let aside = root.path().join("aside");
let renamed_away = |file: &std::fs::File, to: &std::path::Path| {
super::rename_posix(file, to)?;
std::fs::rename(to, &aside)
};
super::replace_with(file, &temporary, &record, true, renamed_away, sync).unwrap();
assert_eq!(std::fs::read(&aside).unwrap(), b"new");
assert!(!record.exists());
assert!(!temporary.exists());
}
fn sync(file: &File) -> io::Result<()> {
file.sync_all()
}
fn prepared(root: &Path) -> (PathBuf, PathBuf, File) {
let record = root.join("record");
std::fs::write(&record, b"old").unwrap();
let temporary = root.join("temporary");
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.access_mode(super::ACCESS)
.open(&temporary)
.unwrap();
file.write_all(b"new").unwrap();
(record, temporary, file)
}
fn rename(posix: bool) -> impl Fn(&File, &Path) -> io::Result<()> {
move |file: &File, to: &Path| {
if posix {
super::rename_posix(file, to)
} else {
Err(io::Error::from_raw_os_error(super::ERROR_NOT_SUPPORTED))
}
}
}
#[test]
fn a_durable_replacement_flushes_the_publishers_handle_after_the_rename() {
let root = tempfile::tempdir().unwrap();
let (record, temporary, file) = prepared(root.path());
let publisher = file.as_raw_handle() as usize;
let steps = RefCell::new(Vec::new());
let rename = |file: &File, to: &Path| {
steps.borrow_mut().push("rename");
super::rename_posix(file, to)
};
let flush = |file: &File| {
assert_eq!(file.as_raw_handle() as usize, publisher);
steps.borrow_mut().push("flush");
file.sync_all()
};
super::replace_with(file, &temporary, &record, true, rename, flush).unwrap();
assert_eq!(steps.into_inner(), ["rename", "flush"]);
assert_eq!(std::fs::read(&record).unwrap(), b"new");
}
#[test]
fn a_durable_fallback_flushes_the_record_it_reopened_after_the_rename() {
let root = tempfile::tempdir().unwrap();
let (record, temporary, file) = prepared(root.path());
let flushes = RefCell::new(0);
let flush = |file: &File| {
assert!(!temporary.exists());
assert_eq!(std::fs::read(&record).unwrap(), b"new");
*flushes.borrow_mut() += 1;
file.sync_all()
};
super::replace_with(file, &temporary, &record, true, rename(false), flush).unwrap();
assert_eq!(flushes.into_inner(), 1);
}
#[test]
fn a_replacement_that_is_not_durable_is_not_flushed() {
for posix in [true, false] {
let root = tempfile::tempdir().unwrap();
let (record, temporary, file) = prepared(root.path());
let flush = |_: &File| -> io::Result<()> { panic!("flushed (posix: {posix})") };
super::replace_with(file, &temporary, &record, false, rename(posix), flush).unwrap();
assert_eq!(std::fs::read(&record).unwrap(), b"new");
}
}
#[test]
fn a_failed_flush_is_returned() {
for posix in [true, false] {
let root = tempfile::tempdir().unwrap();
let (record, temporary, file) = prepared(root.path());
let flush = |_: &File| -> io::Result<()> { Err(io::Error::other("flush failed")) };
let error = super::replace_with(file, &temporary, &record, true, rename(posix), flush)
.unwrap_err();
assert_eq!(error.to_string(), "flush failed", "posix: {posix}");
assert_eq!(std::fs::read(&record).unwrap(), b"new");
}
}
}