mirror of
https://github.com/openai/codex.git
synced 2026-10-20 00:18:49 +00:00
Preserve rollout access metadata across compression and restoration (#55252)
## Why Replacing rollout files can lose Unix ownership or macOS ACLs, changing who can read or append to a transcript. An ACL can also permit appending while denying explicit timestamp updates. ## What changed - Preserve Unix ownership, permissions, modification times, and macOS ACLs before publishing compressed or restored rollouts, and verify the replacement can be reopened. - Abort replacement on metadata failures, reject macOS delete-deny ACLs to keep temporary files removable, and detect Unix access metadata changes through `ctime` checks. - Use managed restoration temporary files whose names avoid stale `*.tmp` cleanup. - Allow appending when refreshing the rollout's modification time returns `PermissionDenied`. ## Testing Add regression coverage for macOS ACL preservation, inherited ACL removal, deletion-protected rollouts, append access despite denied timestamp changes, Unix group preservation, and restoration temporary files surviving stale-file cleanup. GitOrigin-RevId: 23f3de6e35c0b2e19f3491170e320159ea3bdb6f
This commit is contained in:
Generated
+1
@@ -4556,6 +4556,7 @@ dependencies = [
|
||||
"codex-utils-path",
|
||||
"codex-utils-process",
|
||||
"codex-utils-string",
|
||||
"libc",
|
||||
"memchr",
|
||||
"pretty_assertions",
|
||||
"regex",
|
||||
|
||||
@@ -51,6 +51,9 @@ tracing = { workspace = true }
|
||||
uuid = { workspace = true }
|
||||
zstd = { workspace = true }
|
||||
|
||||
[target.'cfg(unix)'.dependencies]
|
||||
libc = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
codex-utils-absolute-path = { workspace = true }
|
||||
pretty_assertions = { workspace = true }
|
||||
|
||||
@@ -1,21 +1,16 @@
|
||||
use std::ffi::OsStr;
|
||||
use std::fs::File;
|
||||
use std::fs::Permissions;
|
||||
use std::io;
|
||||
use std::io::Read;
|
||||
use std::io::Write;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::atomic::AtomicU64;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
|
||||
mod access_metadata;
|
||||
mod blocking_reader;
|
||||
mod error_metrics;
|
||||
mod path_metadata;
|
||||
@@ -30,8 +25,6 @@ use read_metrics::ReadMetrics;
|
||||
const COMPRESSED_SUFFIX: &str = ".zst";
|
||||
const MAX_NOT_FOUND_RETRIES: usize = 3;
|
||||
const OPEN_ROLLOUT_LINE_READER_RETRY_DELAY: Duration = Duration::from_millis(50);
|
||||
const TEMP_SUFFIX: &str = ".tmp";
|
||||
static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
/// The entry point that requested a compression pass, not a Statsig cohort.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
@@ -131,40 +124,45 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result
|
||||
}
|
||||
|
||||
let started_at = Instant::now();
|
||||
let temp_path = temp_path_for(plain_path.as_path(), "decompress");
|
||||
let mut stage = "prepare_directory";
|
||||
let result: io::Result<()> = (|| {
|
||||
if let Some(parent) = plain_path.parent() {
|
||||
std::fs::create_dir_all(parent)?;
|
||||
}
|
||||
stage = "read_metadata";
|
||||
let metadata = std::fs::metadata(compressed_path.as_path())?;
|
||||
let permissions = metadata.permissions();
|
||||
stage = "open_source";
|
||||
let input = File::open(compressed_path.as_path())?;
|
||||
stage = "create_temp";
|
||||
let mut output = create_file_with_permissions(temp_path.as_path(), &permissions)?;
|
||||
// Restoring mtime makes this file look stale before publication. Older
|
||||
// workers delete *.tmp without coordinating with materialization.
|
||||
let mut output = tempfile::Builder::new()
|
||||
.prefix(".rollout-materialize-")
|
||||
.make_in(
|
||||
plain_path.parent().unwrap_or_else(|| Path::new(".")),
|
||||
|path| {
|
||||
// Hard-link publication must not retain Windows' temporary-file attribute.
|
||||
let mut options = std::fs::OpenOptions::new();
|
||||
options.write(true).create_new(true);
|
||||
#[cfg(unix)]
|
||||
options.mode(0o600);
|
||||
options.open(path)
|
||||
},
|
||||
)?;
|
||||
{
|
||||
stage = "open_source";
|
||||
let input = File::open(compressed_path.as_path())?;
|
||||
stage = "decode_and_write";
|
||||
let mut decoder = zstd::stream::read::Decoder::new(input)?;
|
||||
let mut decoder = zstd::stream::read::Decoder::new(&input)?;
|
||||
io::copy(&mut decoder, &mut output)?;
|
||||
}
|
||||
stage = "flush";
|
||||
output.flush()?;
|
||||
stage = "sync";
|
||||
output.sync_all()?;
|
||||
stage = "publish";
|
||||
match std::fs::hard_link(temp_path.as_path(), plain_path.as_path()) {
|
||||
Ok(()) => {}
|
||||
Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {}
|
||||
Err(_) => persist_temp_file_noclobber(temp_path.as_path(), plain_path.as_path())?,
|
||||
}
|
||||
stage = "set_metadata";
|
||||
output.set_times(std::fs::FileTimes::new().set_modified(metadata.modified()?))?;
|
||||
access_metadata::copy(&input, output.as_file())?;
|
||||
stage = "verify";
|
||||
File::open(output.path())?;
|
||||
stage = "sync";
|
||||
output.sync_all()?;
|
||||
drop(output);
|
||||
let _ = std::fs::remove_file(temp_path.as_path());
|
||||
output.as_file().sync_all()?;
|
||||
stage = "publish";
|
||||
persist_temp_file_noclobber(output, plain_path.as_path())?;
|
||||
drop(input);
|
||||
stage = "remove_source";
|
||||
match std::fs::remove_file(compressed_path.as_path()) {
|
||||
Ok(()) => {}
|
||||
@@ -174,7 +172,6 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result
|
||||
Ok(())
|
||||
})();
|
||||
if let Err(err) = &result {
|
||||
let _ = std::fs::remove_file(temp_path.as_path());
|
||||
FailureMetric::Materialize.record(stage, err);
|
||||
metrics::materialize_duration("failed", started_at.elapsed());
|
||||
}
|
||||
@@ -184,10 +181,17 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result
|
||||
Ok(plain_path)
|
||||
}
|
||||
|
||||
fn persist_temp_file_noclobber(temp_path: &Path, destination: &Path) -> io::Result<()> {
|
||||
let temp_path = tempfile::TempPath::try_from_path(temp_path)?;
|
||||
match temp_path.persist_noclobber(destination) {
|
||||
Ok(()) => Ok(()),
|
||||
fn persist_temp_file_noclobber(
|
||||
temp_file: tempfile::NamedTempFile,
|
||||
destination: &Path,
|
||||
) -> io::Result<()> {
|
||||
match std::fs::hard_link(temp_file.path(), destination) {
|
||||
Ok(()) => return Ok(()),
|
||||
Err(err) if err.kind() == io::ErrorKind::AlreadyExists => return Ok(()),
|
||||
Err(_) => {}
|
||||
}
|
||||
match temp_file.persist_noclobber(destination) {
|
||||
Ok(_) => Ok(()),
|
||||
Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => Ok(()),
|
||||
Err(err) => Err(err.error),
|
||||
}
|
||||
@@ -333,10 +337,11 @@ type BlockingLineReader = std::io::Lines<std::io::BufReader<Box<dyn Read + Send>
|
||||
mod worker {
|
||||
use std::ffi::OsStr;
|
||||
use std::fs::File;
|
||||
use std::fs::FileTimes;
|
||||
use std::fs::Permissions;
|
||||
use std::io;
|
||||
use std::io::Write;
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
@@ -355,6 +360,7 @@ mod worker {
|
||||
|
||||
use super::RolloutCompressionTrigger;
|
||||
use super::RolloutFile;
|
||||
use super::access_metadata;
|
||||
use super::error_metrics::FailureMetric;
|
||||
use super::metrics;
|
||||
use super::path;
|
||||
@@ -882,14 +888,14 @@ mod worker {
|
||||
.suffix(TEMP_SUFFIX)
|
||||
.tempfile_in(temp_dir)
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("create_temp", err))?;
|
||||
encode_zstd_to_writer(path, temp_file.as_file_mut())
|
||||
let input = File::open(path)
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("open_source", err))?;
|
||||
encode_zstd_to_writer(&input, temp_file.as_file_mut())
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("encode_and_write", err))?;
|
||||
temp_file
|
||||
.as_file_mut()
|
||||
.flush()
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("flush", err))?;
|
||||
verify_zstd(temp_file.path())
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("verify", err))?;
|
||||
if !same_file_state(path, &before)
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("recheck_source", err))?
|
||||
{
|
||||
@@ -899,8 +905,12 @@ mod worker {
|
||||
/*compressed_bytes*/ None,
|
||||
));
|
||||
}
|
||||
set_file_metadata(temp_file.as_file(), before.modified, &before.permissions)
|
||||
access_metadata::copy(&input, temp_file.as_file())
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("set_metadata", err))?;
|
||||
// Reopen after applying final access metadata: verifying the initial
|
||||
// owner-only temporary would miss an inaccessible published file.
|
||||
verify_zstd(temp_file.path())
|
||||
.inspect_err(|err| FailureMetric::File(trigger).record("verify", err))?;
|
||||
temp_file
|
||||
.as_file()
|
||||
.sync_all()
|
||||
@@ -970,6 +980,8 @@ mod worker {
|
||||
len: u64,
|
||||
modified: SystemTime,
|
||||
permissions: Permissions,
|
||||
#[cfg(unix)]
|
||||
change_time: (i64, i64),
|
||||
}
|
||||
|
||||
fn cold_file_state(path: &Path) -> io::Result<ColdFileState> {
|
||||
@@ -988,6 +1000,8 @@ mod worker {
|
||||
len: metadata.len(),
|
||||
modified,
|
||||
permissions: metadata.permissions(),
|
||||
#[cfg(unix)]
|
||||
change_time: (metadata.ctime(), metadata.ctime_nsec()),
|
||||
};
|
||||
let age = SystemTime::now()
|
||||
.duration_since(modified)
|
||||
@@ -1000,16 +1014,23 @@ mod worker {
|
||||
|
||||
fn same_file_state(path: &Path, expected: &FileState) -> io::Result<bool> {
|
||||
match std::fs::metadata(path) {
|
||||
Ok(metadata) => Ok(metadata.len() == expected.len
|
||||
&& metadata.modified()? == expected.modified
|
||||
&& metadata.permissions() == expected.permissions),
|
||||
Ok(metadata) => {
|
||||
// Ownership and ACL edits change ctime even when mode, length
|
||||
// and mtime are unchanged. Do not publish stale access metadata.
|
||||
#[cfg(unix)]
|
||||
if (metadata.ctime(), metadata.ctime_nsec()) != expected.change_time {
|
||||
return Ok(false);
|
||||
}
|
||||
Ok(metadata.len() == expected.len
|
||||
&& metadata.modified()? == expected.modified
|
||||
&& metadata.permissions() == expected.permissions)
|
||||
}
|
||||
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false),
|
||||
Err(err) => Err(err),
|
||||
}
|
||||
}
|
||||
|
||||
fn encode_zstd_to_writer(source: &Path, output: impl Write) -> io::Result<()> {
|
||||
let mut input = File::open(source)?;
|
||||
fn encode_zstd_to_writer(mut input: &File, output: impl Write) -> io::Result<()> {
|
||||
let mut encoder = zstd::stream::write::Encoder::new(output, COMPRESSION_LEVEL)?;
|
||||
// Preserve fast byte-bound checks for paginated history without decoding the whole file.
|
||||
encoder.set_pledged_src_size(Some(input.metadata()?.len()))?;
|
||||
@@ -1026,15 +1047,6 @@ mod worker {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn set_file_metadata(
|
||||
file: &File,
|
||||
modified: SystemTime,
|
||||
permissions: &Permissions,
|
||||
) -> io::Result<()> {
|
||||
file.set_times(FileTimes::new().set_modified(modified))?;
|
||||
file.set_permissions(permissions.clone())
|
||||
}
|
||||
|
||||
async fn cleanup_stale_temps(
|
||||
codex_home: &Path,
|
||||
trigger: RolloutCompressionTrigger,
|
||||
@@ -1395,40 +1407,6 @@ mod reader {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result<File> {
|
||||
let file = std::fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.mode(permissions.mode() & 0o7777)
|
||||
.open(path)?;
|
||||
file.set_permissions(permissions.clone())?;
|
||||
Ok(file)
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result<File> {
|
||||
let file = std::fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(path)?;
|
||||
file.set_permissions(permissions.clone())?;
|
||||
Ok(file)
|
||||
}
|
||||
|
||||
fn temp_path_for(path: &Path, operation: &str) -> PathBuf {
|
||||
let mut file_name = path
|
||||
.file_name()
|
||||
.map(OsStr::to_os_string)
|
||||
.unwrap_or_else(|| OsStr::new("rollout").to_os_string());
|
||||
let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
|
||||
file_name.push(format!(
|
||||
".{operation}.{}.{counter}{TEMP_SUFFIX}",
|
||||
std::process::id()
|
||||
));
|
||||
path.with_file_name(file_name)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "compression_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
//! Preserves access to replacement rollout files before they are published.
|
||||
//! Metadata failures leave the source intact; never silently replace ownership
|
||||
//! or discard a macOS ACL when switching between plain and compressed files.
|
||||
//! Apply ACLs last because they can prohibit subsequent metadata updates.
|
||||
//! Reject delete-deny ACLs before installation so failed replacements remain removable.
|
||||
|
||||
use std::fs::File;
|
||||
use std::fs::FileTimes;
|
||||
use std::io;
|
||||
#[cfg(target_os = "macos")]
|
||||
use std::os::fd::AsRawFd;
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
// libc does not expose the macOS ACL APIs.
|
||||
#[cfg(target_os = "macos")]
|
||||
unsafe extern "C" {
|
||||
fn acl_get_fd(fd: libc::c_int) -> *mut libc::c_void;
|
||||
fn acl_get_entry(
|
||||
acl: *mut libc::c_void,
|
||||
entry_id: libc::c_int,
|
||||
entry: *mut *mut libc::c_void,
|
||||
) -> libc::c_int;
|
||||
fn acl_get_tag_type(entry: *mut libc::c_void, tag: *mut libc::c_uint) -> libc::c_int;
|
||||
fn acl_get_permset_mask_np(entry: *mut libc::c_void, permissions: *mut u64) -> libc::c_int;
|
||||
fn acl_init(count: libc::c_int) -> *mut libc::c_void;
|
||||
fn acl_set_fd(fd: libc::c_int, acl: *mut libc::c_void) -> libc::c_int;
|
||||
fn acl_free(acl: *mut libc::c_void) -> libc::c_int;
|
||||
}
|
||||
|
||||
pub(super) fn copy(source: &File, destination: &File) -> io::Result<()> {
|
||||
let metadata = source.metadata()?;
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let destination_metadata = destination.metadata()?;
|
||||
if (destination_metadata.uid(), destination_metadata.gid())
|
||||
!= (metadata.uid(), metadata.gid())
|
||||
{
|
||||
// Chown can clear mode bits, so restore permissions afterwards.
|
||||
// Failure must abort publication rather than change who owns the file.
|
||||
std::os::unix::fs::fchown(destination, Some(metadata.uid()), Some(metadata.gid()))?;
|
||||
}
|
||||
}
|
||||
destination.set_permissions(metadata.permissions())?;
|
||||
destination.set_times(FileTimes::new().set_modified(metadata.modified()?))?;
|
||||
#[cfg(target_os = "macos")]
|
||||
{
|
||||
// COPYFILE_ACL merges inherited destination entries and drops inherited
|
||||
// source entries. Replace the entire ACL to preserve its order and denies.
|
||||
// SAFETY: source stays open and the returned ACL is freed below.
|
||||
let mut acl = unsafe { acl_get_fd(source.as_raw_fd()) };
|
||||
let source_has_no_acl = acl.is_null();
|
||||
if source_has_no_acl {
|
||||
let error = io::Error::last_os_error();
|
||||
if error.raw_os_error() != Some(libc::ENOENT) {
|
||||
return Err(error);
|
||||
}
|
||||
// No source ACL: also clear any ACL inherited by the new file.
|
||||
// SAFETY: zero is a valid entry count; allocation failure returns null.
|
||||
acl = unsafe {
|
||||
acl_init(/*count*/ 0)
|
||||
};
|
||||
if acl.is_null() {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
}
|
||||
let result = (|| {
|
||||
// These values come from macOS <sys/acl.h>.
|
||||
const ACL_EXTENDED_DENY: libc::c_uint = 2;
|
||||
const ACL_DELETE: u64 = 1 << 4;
|
||||
let mut entry_id = 0;
|
||||
loop {
|
||||
let mut entry = std::ptr::null_mut();
|
||||
// SAFETY: acl is valid and entry points to writable pointer storage.
|
||||
if unsafe { acl_get_entry(acl, entry_id, &mut entry) } != 0 {
|
||||
let error = io::Error::last_os_error();
|
||||
// macOS returns EINVAL when the entry index reaches the end.
|
||||
if error.raw_os_error() == Some(libc::EINVAL) {
|
||||
break;
|
||||
}
|
||||
return Err(error);
|
||||
}
|
||||
let mut tag = 0;
|
||||
let mut permissions = 0;
|
||||
// SAFETY: entry belongs to the live acl; both outputs are writable.
|
||||
if unsafe { acl_get_tag_type(entry, &mut tag) } != 0
|
||||
|| unsafe { acl_get_permset_mask_np(entry, &mut permissions) } != 0
|
||||
{
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
// Conservatively reject any delete deny, regardless of principal or
|
||||
// ordering. Installing it can prevent both publication and cleanup,
|
||||
// and the ACL may also deny restoring permissions after a failure.
|
||||
if tag == ACL_EXTENDED_DENY && permissions & ACL_DELETE != 0 {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::PermissionDenied,
|
||||
"cannot replace a rollout with a delete-deny ACL",
|
||||
));
|
||||
}
|
||||
entry_id += 1;
|
||||
}
|
||||
// SAFETY: destination stays open and acl is a valid owned allocation.
|
||||
if unsafe { acl_set_fd(destination.as_raw_fd(), acl) } == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
let error = io::Error::last_os_error();
|
||||
// Filesystems without ACL support reject even an empty ACL. There
|
||||
// is nothing to preserve only when the source had no ACL.
|
||||
if source_has_no_acl && error.raw_os_error() == Some(libc::ENOTSUP) {
|
||||
return Ok(());
|
||||
}
|
||||
Err(error)
|
||||
})();
|
||||
// SAFETY: acl came from acl_get_fd or acl_init and is no longer used.
|
||||
unsafe { acl_free(acl) };
|
||||
result?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -3,6 +3,7 @@ use std::fs;
|
||||
use std::fs::FileTimes;
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
use std::time::SystemTime;
|
||||
|
||||
@@ -441,10 +442,21 @@ async fn worker_compresses_old_active_and_archived_rollouts() -> anyhow::Result<
|
||||
let stale_temp = active_path.with_file_name("rollout-stale.jsonl.zst.tmp");
|
||||
fs::write(&stale_temp, "stale temp")?;
|
||||
set_old_mtime(&stale_temp)?;
|
||||
// A crash during metadata restoration can leave the data file unreadable.
|
||||
#[cfg(unix)]
|
||||
fs::set_permissions(&stale_temp, fs::Permissions::from_mode(0o000))?;
|
||||
|
||||
let fresh_temp = active_path.with_file_name("rollout-fresh.jsonl.zst.tmp");
|
||||
fs::write(&fresh_temp, "fresh temp")?;
|
||||
|
||||
// Pause materialization after restoring the old metadata, before publication.
|
||||
let mut live_temp = tempfile::Builder::new()
|
||||
.prefix(".rollout-materialize-")
|
||||
.tempfile_in(active_path.parent().unwrap())?;
|
||||
live_temp.write_all(b"live materialization")?;
|
||||
access_metadata::copy(&fs::File::open(&active_path)?, live_temp.as_file())?;
|
||||
let source_modified = fs::metadata(&active_path)?.modified()?;
|
||||
|
||||
worker::run(home.path().to_path_buf(), Startup).await?;
|
||||
|
||||
assert!(!active_path.exists());
|
||||
@@ -455,6 +467,13 @@ async fn worker_compresses_old_active_and_archived_rollouts() -> anyhow::Result<
|
||||
assert!(!compressed_rollout_path(&fresh_path).exists());
|
||||
assert!(!stale_temp.exists());
|
||||
assert!(fresh_temp.exists());
|
||||
let materialized_path = home.path().join("materialized.jsonl");
|
||||
persist_temp_file_noclobber(live_temp, &materialized_path)?;
|
||||
assert_eq!(
|
||||
fs::metadata(&materialized_path)?.modified()?,
|
||||
source_modified
|
||||
);
|
||||
assert_eq!(fs::read(&materialized_path)?, b"live materialization");
|
||||
assert!(
|
||||
home.path()
|
||||
.join(".tmp")
|
||||
@@ -623,6 +642,211 @@ async fn compression_preserves_rollout_permissions() -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
#[tokio::test]
|
||||
async fn worker_preserves_inherited_acl_through_append() -> anyhow::Result<()> {
|
||||
use anyhow::Context;
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
use std::process::Command;
|
||||
|
||||
let home = TempDir::new()?;
|
||||
let uuid = Uuid::from_u128(61);
|
||||
let thread_id = ThreadId::from_string(&uuid.to_string())?;
|
||||
let path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid);
|
||||
write_rollout(&path, thread_id, "ACL-protected transcript")?;
|
||||
set_old_mtime(&path)?;
|
||||
// An archived rollout can retain inherited denies from its former directory.
|
||||
let acl = Command::new("/bin/chmod")
|
||||
.args(["+ai#", "0", "everyone deny writeattr"])
|
||||
.arg(&path)
|
||||
.output()?;
|
||||
assert!(acl.status.success(), "{acl:?}");
|
||||
fs::set_permissions(&path, fs::Permissions::from_mode(0o000))?;
|
||||
let acl = Command::new("/bin/chmod")
|
||||
.args(["+a#", "1", "everyone allow read,write,append,writeattr"])
|
||||
.arg(&path)
|
||||
.output()?;
|
||||
assert!(acl.status.success(), "{acl:?}");
|
||||
let before = fs::metadata(&path)?;
|
||||
let expected = RolloutRecorder::load_rollout_items(&path)
|
||||
.await
|
||||
.context("read the ACL-protected source")?;
|
||||
fs::OpenOptions::new()
|
||||
.append(true)
|
||||
.open(&path)
|
||||
.context("the source ACL must permit appending before compression")?;
|
||||
assert_eq!(
|
||||
fs::File::open(&path)?
|
||||
.set_times(FileTimes::new().set_modified(SystemTime::now()))
|
||||
.unwrap_err()
|
||||
.kind(),
|
||||
io::ErrorKind::PermissionDenied
|
||||
);
|
||||
|
||||
// Exercise discovery, compression, final metadata and publication through
|
||||
// the real worker, not the test-only compress_now helper.
|
||||
worker::run(home.path().to_path_buf(), Startup).await?;
|
||||
let compressed = compressed_rollout_path(&path);
|
||||
assert!(!path.exists());
|
||||
assert_eq!(
|
||||
fs::File::open(&compressed)?
|
||||
.set_times(FileTimes::new().set_modified(SystemTime::now()))
|
||||
.unwrap_err()
|
||||
.kind(),
|
||||
io::ErrorKind::PermissionDenied
|
||||
);
|
||||
let after = fs::metadata(&compressed)?;
|
||||
assert_eq!(after.modified()?, before.modified()?);
|
||||
assert_eq!(
|
||||
(after.uid(), after.gid(), after.permissions().mode()),
|
||||
(before.uid(), before.gid(), before.permissions().mode())
|
||||
);
|
||||
assert_eq!(
|
||||
serde_json::to_value(
|
||||
RolloutRecorder::load_rollout_items(&path)
|
||||
.await
|
||||
.context("read the worker's compressed replacement")?
|
||||
)?,
|
||||
serde_json::to_value(&expected)?
|
||||
);
|
||||
|
||||
let appended = RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
|
||||
message: "append through the preserved ACL".to_string(),
|
||||
..Default::default()
|
||||
}));
|
||||
append_rollout_item_to_path(&path, &appended)
|
||||
.await
|
||||
.context("materialize and append through the preserved ACL")?;
|
||||
assert!(!compressed.exists());
|
||||
assert_eq!(
|
||||
fs::File::open(&path)?
|
||||
.set_times(FileTimes::new().set_modified(SystemTime::now()))
|
||||
.unwrap_err()
|
||||
.kind(),
|
||||
io::ErrorKind::PermissionDenied
|
||||
);
|
||||
let after = fs::metadata(&path)?;
|
||||
assert_eq!(
|
||||
(after.uid(), after.gid(), after.permissions().mode()),
|
||||
(before.uid(), before.gid(), before.permissions().mode())
|
||||
);
|
||||
let mut expected_items = expected.0;
|
||||
expected_items.push(appended);
|
||||
assert_eq!(
|
||||
serde_json::to_value(RolloutRecorder::load_rollout_items(&path).await?)?,
|
||||
serde_json::to_value((expected_items, expected.1, expected.2))?
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn worker_preserves_alternate_group_through_materialization() -> anyhow::Result<()> {
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
use std::process::Command;
|
||||
|
||||
let home = TempDir::new()?;
|
||||
let uuid = Uuid::from_u128(62);
|
||||
let thread_id = ThreadId::from_string(&uuid.to_string())?;
|
||||
let path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid);
|
||||
write_rollout(&path, thread_id, "different group")?;
|
||||
set_old_mtime(&path)?;
|
||||
let original_group = fs::metadata(&path)?.gid();
|
||||
let groups = Command::new("id").arg("-G").output()?;
|
||||
assert!(groups.status.success());
|
||||
let Some(group) = std::str::from_utf8(&groups.stdout)?
|
||||
.split_whitespace()
|
||||
.map(str::parse::<u32>)
|
||||
.collect::<Result<Vec<_>, _>>()?
|
||||
.into_iter()
|
||||
.find(|group| *group != original_group)
|
||||
else {
|
||||
eprintln!("skipping ownership regression: no alternate supplementary group");
|
||||
return Ok(());
|
||||
};
|
||||
std::os::unix::fs::chown(&path, /*uid*/ None, Some(group))?;
|
||||
assert_eq!(fs::metadata(&path)?.gid(), group);
|
||||
|
||||
worker::run(home.path().to_path_buf(), Startup).await?;
|
||||
assert!(!path.exists());
|
||||
assert_eq!(fs::metadata(compressed_rollout_path(&path))?.gid(), group);
|
||||
materialize_rollout_for_append(&path, /*writer_lock*/ None).await?;
|
||||
assert_eq!(fs::metadata(&path)?.gid(), group);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
#[tokio::test]
|
||||
async fn worker_does_not_leak_temps_for_deletion_protected_rollouts() -> anyhow::Result<()> {
|
||||
use std::process::Command;
|
||||
|
||||
let home = TempDir::new()?;
|
||||
let uuid = Uuid::from_u128(63);
|
||||
let thread_id = ThreadId::from_string(&uuid.to_string())?;
|
||||
let path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid);
|
||||
write_rollout(&path, thread_id, "deletion-protected transcript")?;
|
||||
set_old_mtime(&path)?;
|
||||
let acl = Command::new("/bin/chmod")
|
||||
.args(["+a", "everyone deny delete,writesecurity"])
|
||||
.arg(&path)
|
||||
.output()?;
|
||||
assert!(acl.status.success(), "{acl:?}");
|
||||
let contents = fs::read(&path)?;
|
||||
let modified = fs::metadata(&path)?.modified()?;
|
||||
let access = Command::new("/bin/ls").arg("-le").arg(&path).output()?;
|
||||
assert!(access.status.success(), "{access:?}");
|
||||
|
||||
// Collect results before asserting so even a regression's protected temps
|
||||
// have their ACLs cleared before TempDir tries to remove them.
|
||||
let result: anyhow::Result<_> = async {
|
||||
let mut files_after_passes = Vec::new();
|
||||
for _ in 0..2 {
|
||||
worker::run(home.path().to_path_buf(), Startup).await?;
|
||||
files_after_passes.push(
|
||||
fs::read_dir(path.parent().unwrap())?
|
||||
.map(|entry| entry.map(|entry| entry.path()))
|
||||
.collect::<io::Result<Vec<_>>>()?,
|
||||
);
|
||||
// Allow the next pass without waiting for the run marker to expire.
|
||||
fs::remove_file(home.path().join(".tmp/rollout-compression.lock"))?;
|
||||
}
|
||||
let access_after = Command::new("/bin/ls").arg("-le").arg(&path).output()?;
|
||||
Ok((
|
||||
files_after_passes,
|
||||
fs::read(&path)?,
|
||||
fs::metadata(&path)?.modified()?,
|
||||
access_after,
|
||||
fs::remove_file(&path).map_err(|err| err.kind()),
|
||||
))
|
||||
}
|
||||
.await;
|
||||
for entry in fs::read_dir(path.parent().unwrap())? {
|
||||
let cleanup = Command::new("/bin/chmod")
|
||||
.arg("-N")
|
||||
.arg(entry?.path())
|
||||
.output()?;
|
||||
assert!(cleanup.status.success(), "{cleanup:?}");
|
||||
}
|
||||
let (files_after_passes, contents_after, modified_after, access_after, deletion) = result?;
|
||||
assert!(access_after.status.success(), "{access_after:?}");
|
||||
assert_eq!(files_after_passes, vec![vec![path.clone()], vec![path]]);
|
||||
assert_eq!(
|
||||
(
|
||||
contents_after,
|
||||
modified_after,
|
||||
access_after.stdout,
|
||||
deletion
|
||||
),
|
||||
(
|
||||
contents,
|
||||
modified,
|
||||
access.stdout,
|
||||
Err(io::ErrorKind::PermissionDenied)
|
||||
)
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn append_materialization_preserves_compressed_rollout_permissions() -> anyhow::Result<()> {
|
||||
@@ -656,11 +880,12 @@ async fn append_materialization_preserves_compressed_rollout_permissions() -> an
|
||||
#[test]
|
||||
fn persist_temp_file_noclobber_installs_completed_temp() -> anyhow::Result<()> {
|
||||
let home = TempDir::new()?;
|
||||
let temp_path = home.path().join("rollout.jsonl.tmp");
|
||||
let mut temp_file = tempfile::NamedTempFile::new_in(home.path())?;
|
||||
let temp_path = temp_file.path().to_path_buf();
|
||||
let destination = home.path().join("rollout.jsonl");
|
||||
fs::write(&temp_path, "completed rollout")?;
|
||||
temp_file.write_all(b"completed rollout")?;
|
||||
|
||||
persist_temp_file_noclobber(&temp_path, &destination)?;
|
||||
persist_temp_file_noclobber(temp_file, &destination)?;
|
||||
|
||||
assert!(!temp_path.exists());
|
||||
assert_eq!(fs::read_to_string(destination)?, "completed rollout");
|
||||
@@ -670,12 +895,13 @@ fn persist_temp_file_noclobber_installs_completed_temp() -> anyhow::Result<()> {
|
||||
#[test]
|
||||
fn persist_temp_file_noclobber_does_not_replace_existing_destination() -> anyhow::Result<()> {
|
||||
let home = TempDir::new()?;
|
||||
let temp_path = home.path().join("rollout.jsonl.tmp");
|
||||
let mut temp_file = tempfile::NamedTempFile::new_in(home.path())?;
|
||||
let temp_path = temp_file.path().to_path_buf();
|
||||
let destination = home.path().join("rollout.jsonl");
|
||||
fs::write(&temp_path, "candidate rollout")?;
|
||||
temp_file.write_all(b"candidate rollout")?;
|
||||
fs::write(&destination, "existing rollout")?;
|
||||
|
||||
persist_temp_file_noclobber(&temp_path, &destination)?;
|
||||
persist_temp_file_noclobber(temp_file, &destination)?;
|
||||
|
||||
assert!(!temp_path.exists());
|
||||
assert_eq!(fs::read_to_string(destination)?, "existing rollout");
|
||||
@@ -693,6 +919,18 @@ async fn compression_preserves_read_only_rollout_permissions() -> anyhow::Result
|
||||
set_old_mtime(&rollout_path)?;
|
||||
fs::set_permissions(&rollout_path, fs::Permissions::from_mode(0o400))?;
|
||||
let source_modified = fs::metadata(&rollout_path)?.modified()?;
|
||||
#[cfg(target_os = "macos")]
|
||||
{
|
||||
// New files inherit an allow that is absent from the existing source.
|
||||
let acl = std::process::Command::new("/bin/chmod")
|
||||
.args([
|
||||
"+a",
|
||||
"everyone allow write,append,file_inherit,only_inherit",
|
||||
])
|
||||
.arg(rollout_path.parent().unwrap())
|
||||
.output()?;
|
||||
assert!(acl.status.success(), "{acl:?}");
|
||||
}
|
||||
|
||||
worker::run(home.path().to_path_buf(), Startup).await?;
|
||||
|
||||
@@ -701,6 +939,27 @@ async fn compression_preserves_read_only_rollout_permissions() -> anyhow::Result
|
||||
assert!(!rollout_path.exists());
|
||||
assert_eq!(compressed_metadata.permissions().mode() & 0o777, 0o400);
|
||||
assert_eq!(compressed_metadata.modified()?, source_modified);
|
||||
#[cfg(target_os = "macos")]
|
||||
{
|
||||
assert_eq!(
|
||||
fs::OpenOptions::new()
|
||||
.append(true)
|
||||
.open(&compressed_path)
|
||||
.unwrap_err()
|
||||
.kind(),
|
||||
io::ErrorKind::PermissionDenied
|
||||
);
|
||||
crate::materialize_rollout_for_reference(&rollout_path).await?;
|
||||
assert_eq!(fs::metadata(&rollout_path)?.modified()?, source_modified);
|
||||
assert_eq!(
|
||||
fs::OpenOptions::new()
|
||||
.append(true)
|
||||
.open(&rollout_path)
|
||||
.unwrap_err()
|
||||
.kind(),
|
||||
io::ErrorKind::PermissionDenied
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -1776,12 +1776,22 @@ fn open_log_file(path: &Path) -> std::io::Result<File> {
|
||||
.create(true)
|
||||
.open(path)?;
|
||||
if refresh_modified_time {
|
||||
file.set_times(std::fs::FileTimes::new().set_modified(std::time::SystemTime::now()))?;
|
||||
refresh_rollout_modified_time(&file)?;
|
||||
}
|
||||
ensure_rollout_is_newline_terminated(&mut file)?;
|
||||
Ok(file)
|
||||
}
|
||||
|
||||
fn refresh_rollout_modified_time(file: &File) -> std::io::Result<()> {
|
||||
match file.set_times(std::fs::FileTimes::new().set_modified(std::time::SystemTime::now())) {
|
||||
Ok(()) => Ok(()),
|
||||
// A preserved ACL can permit appending while denying explicit timestamp changes.
|
||||
// The append itself will update mtime, so this refresh must not prevent it.
|
||||
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => Ok(()),
|
||||
Err(err) => Err(err),
|
||||
}
|
||||
}
|
||||
|
||||
/// Mutable state owned by the background rollout writer.
|
||||
///
|
||||
/// Items are first appended to `pending_items`; persist/flush/shutdown remove each item from that
|
||||
@@ -2045,7 +2055,7 @@ async fn open_rollout_for_append(
|
||||
.append(true)
|
||||
.open(path_for_open.as_path())?;
|
||||
if refresh_modified_time {
|
||||
file.set_times(std::fs::FileTimes::new().set_modified(std::time::SystemTime::now()))?;
|
||||
refresh_rollout_modified_time(&file)?;
|
||||
}
|
||||
ensure_rollout_is_newline_terminated(&mut file)?;
|
||||
let ordinal_state = ordinal_state_for_rollout(&mut file, path_for_open.as_path())?;
|
||||
|
||||
@@ -174,6 +174,31 @@ async fn opening_existing_rollout_preserves_modified_time() -> std::io::Result<(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
#[test]
|
||||
fn open_log_file_materializes_rollout_when_acl_denies_timestamp_changes() -> anyhow::Result<()> {
|
||||
let home = TempDir::new()?;
|
||||
let rollout_path = home.path().join("rollout.jsonl");
|
||||
let compressed_path = rollout_path.with_extension("jsonl.zst");
|
||||
fs::write(
|
||||
&compressed_path,
|
||||
zstd::stream::encode_all(&b"original\n"[..], 0)?,
|
||||
)?;
|
||||
let acl = std::process::Command::new("/bin/chmod")
|
||||
.args(["+a", "everyone deny writeattr"])
|
||||
.arg(&compressed_path)
|
||||
.output()?;
|
||||
assert!(acl.status.success(), "{acl:?}");
|
||||
|
||||
let mut file = open_log_file(&rollout_path)?;
|
||||
file.write_all(b"appended\n")?;
|
||||
drop(file);
|
||||
|
||||
assert!(!compressed_path.exists());
|
||||
assert_eq!(fs::read(&rollout_path)?, b"original\nappended\n");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> {
|
||||
let home = TempDir::new().expect("temp dir");
|
||||
|
||||
Reference in New Issue
Block a user