feat(artifacts): add bounded reconciliation

This commit is contained in:
2026-08-27 07:04:22 +03:00
parent 38ba898b72
commit 889b1bdb57
5 changed files with 985 additions and 9 deletions
+489 -2
View File
@@ -1,13 +1,66 @@
use std::{
fmt,
os::fd::AsRawFd,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use crate::{
ArtifactError, ArtifactStore,
store::{RootLock, ensure_root_unchanged, fsync_fd, open_existing_dir, unlinkat},
ArtifactError, ArtifactStore, ReconciliationMutation, ReconciliationNamespace,
ReconciliationScan,
store::{
QUARANTINE_DIR, RootLock, check_file, checkpointed, ensure_root_unchanged, fsync_fd,
open_existing_dir, open_or_create_dir, rename_no_replace_at, stat_fd, unlinkat,
},
};
const FINAL_NAMESPACE: u8 = 0;
const QUARANTINE_NAMESPACE: u8 = 1;
/// Opaque continuation for a bounded reconciliation traversal. It is valid
/// only for the unchanged pinned root from which it was returned.
pub struct ReconciliationCursor {
root_dev: u64,
root_ino: u64,
namespace: u8,
shard: u8,
position: i64,
}
impl fmt::Debug for ReconciliationCursor {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("ReconciliationCursor(..)")
}
}
/// Opaque, inode-bound evidence for one valid final or quarantined artifact.
/// It intentionally exposes neither its digest nor filesystem path.
#[derive(Clone)]
pub struct ReconciliationCandidate {
root_dev: u64,
root_ino: u64,
namespace: ReconciliationNamespace,
shard: String,
shard_dev: u64,
shard_ino: u64,
name: String,
dev: u64,
ino: u64,
modified_seconds: i64,
modified_nanoseconds: i64,
}
impl fmt::Debug for ReconciliationCandidate {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("ReconciliationCandidate(..)")
}
}
impl ReconciliationCandidate {
pub fn namespace(&self) -> ReconciliationNamespace {
self.namespace
}
}
/// Opaque, single-use evidence that a particular stale temporary inode was
/// observed under this store's pinned root. It is not a blob reference.
#[derive(Debug)]
@@ -30,6 +83,259 @@ pub struct TempScan {
}
impl ArtifactStore {
/// Streams valid final and quarantine entries with an opaque continuation.
/// Neither the report nor the returned capabilities disclose paths/digests.
pub fn scan_reconciliation(
&self,
continuation: Option<ReconciliationCursor>,
scan_budget: usize,
result_limit: usize,
) -> Result<(ReconciliationScan, Vec<ReconciliationCandidate>), ArtifactError> {
let root = self.root()?;
let _lock = RootLock::shared(root)?;
ensure_root_unchanged(root)?;
let mut cursor = match continuation {
Some(cursor) => {
if cursor.root_dev != root.dev
|| cursor.root_ino != root.ino
|| cursor.namespace > QUARANTINE_NAMESPACE
{
return Err(ArtifactError::UnsafeRoot);
}
cursor
}
None => reconciliation_cursor(root.dev, root.ino, FINAL_NAMESPACE, 0, 0),
};
let mut report = ReconciliationScan::empty(None);
if scan_budget == 0 || result_limit == 0 {
report.continuation = Some(cursor);
return Ok((report, Vec::new()));
}
let mut remaining = scan_budget;
let mut entries = Vec::new();
while cursor.namespace <= QUARANTINE_NAMESPACE {
if remaining == 0 || entries.len() == result_limit {
report.continuation = Some(cursor);
return Ok((report, entries));
}
let namespace_name = namespace_name(cursor.namespace);
let namespace = match open_existing_dir(root.fd.as_raw_fd(), namespace_name) {
Ok(fd) => fd,
Err(ArtifactError::NotFound) => {
advance_namespace(&mut cursor);
continue;
}
Err(error) => return Err(error),
};
loop {
if remaining == 0 {
report.continuation = Some(cursor);
return Ok((report, entries));
}
let shard_name = format!("{:02x}", cursor.shard);
let shard = match open_existing_dir(namespace.as_raw_fd(), shard_name.as_bytes()) {
Ok(fd) => fd,
Err(ArtifactError::NotFound) => {
consume_scan_work(&mut report, &mut remaining);
if cursor.shard == u8::MAX {
break;
}
advance_shard(&mut cursor);
continue;
}
Err(error) => return Err(error),
};
let shard_stat = stat_fd(shard.as_raw_fd())?;
let mut directory = DirectoryStream::open(shard.as_raw_fd(), cursor.position)?;
loop {
if remaining == 0 || entries.len() == result_limit {
report.continuation = Some(cursor);
return Ok((report, entries));
}
let Some((name, after)) = directory.next()? else {
// An empty directory still consumes one bounded
// traversal unit, so a sparse namespace cannot make
// us probe all 256 shards in one page.
consume_scan_work(&mut report, &mut remaining);
break;
};
cursor.position = after;
consume_scan_work(&mut report, &mut remaining);
let Ok(name) = String::from_utf8(name) else {
report.malformed += 1;
continue;
};
if !valid_final_name(&name, &shard_name) {
report.malformed += 1;
continue;
}
#[cfg(debug_assertions)]
let _ = crate::test_support::checkpoint("reconciliation_before_stat");
let stat = match checkpointed("reconciliation_stat", || {
nofollow_stat(shard.as_raw_fd(), &name)
}) {
Ok(stat) => stat,
Err(ArtifactError::NotFound) => {
// The entry was concurrently removed. It is no
// longer retryable, so the cursor may advance.
continue;
}
Err(ArtifactError::UnsafeRoot) => {
report.unsafe_entries += 1;
continue;
}
Err(error) => return Err(error),
};
if check_file(&stat).is_err() {
report.unsafe_entries += 1;
continue;
}
let namespace = if cursor.namespace == FINAL_NAMESPACE {
report.final_entries += 1;
ReconciliationNamespace::Final
} else {
report.quarantined_entries += 1;
ReconciliationNamespace::Quarantine
};
entries.push(ReconciliationCandidate {
root_dev: root.dev,
root_ino: root.ino,
namespace,
shard: shard_name.clone(),
shard_dev: shard_stat.st_dev,
shard_ino: shard_stat.st_ino,
name,
dev: stat.st_dev,
ino: stat.st_ino,
modified_seconds: stat.st_mtime,
modified_nanoseconds: stat.st_mtime_nsec,
});
// The continuation is deliberately kept at `after`,
// immediately before the next unissued item.
}
if cursor.shard == u8::MAX {
break;
}
advance_shard(&mut cursor);
}
advance_namespace(&mut cursor);
}
Ok((report, entries))
}
/// Moves a revalidated final inode into the private quarantine namespace
/// without replacing any existing quarantine inode.
pub fn quarantine_reconciliation(
&self,
candidate: ReconciliationCandidate,
) -> Result<ReconciliationMutation, ArtifactError> {
if candidate.namespace != ReconciliationNamespace::Final {
return Err(ArtifactError::UnsafeRoot);
}
let root = self.root()?;
let _lock = RootLock::exclusive(root)?;
ensure_reconciliation_root(root, &candidate)?;
let final_root = open_existing_dir(root.fd.as_raw_fd(), namespace_name(FINAL_NAMESPACE))?;
let final_shard = open_existing_dir(final_root.as_raw_fd(), candidate.shard.as_bytes())?;
match revalidate_candidate(&candidate, final_shard.as_raw_fd()) {
Ok(()) => {}
Err(ArtifactError::NotFound) => {
return self.recover_quarantine(root, &candidate);
}
Err(ArtifactError::UnsafeRoot) => {
return match self.recover_quarantine(root, &candidate) {
Ok(outcome) => Ok(outcome),
Err(ArtifactError::NotFound) => Err(ArtifactError::UnsafeRoot),
Err(error) => Err(error),
};
}
Err(error) => return Err(error),
}
let quarantine_root = open_or_create_dir(root.fd.as_raw_fd(), QUARANTINE_DIR)?;
let quarantine_shard =
open_or_create_dir(quarantine_root.as_raw_fd(), candidate.shard.as_bytes())?;
match checkpointed("reconciliation_quarantine_rename", || {
rename_no_replace_at(
final_shard.as_raw_fd(),
candidate.name.as_bytes(),
quarantine_shard.as_raw_fd(),
candidate.name.as_bytes(),
)
})? {
true => match fsync_quarantine_dirs(&final_shard, &quarantine_shard) {
Ok(()) => Ok(ReconciliationMutation::Quarantined),
Err(_) => Ok(ReconciliationMutation::Retryable),
},
false => self.recover_quarantine(root, &candidate),
}
}
/// Idempotently removes a revalidated quarantined inode and fsyncs its
/// directory. An fsync ambiguity is explicitly retryable.
pub fn delete_quarantined_reconciliation(
&self,
candidate: ReconciliationCandidate,
) -> Result<ReconciliationMutation, ArtifactError> {
if candidate.namespace != ReconciliationNamespace::Quarantine {
return Err(ArtifactError::UnsafeRoot);
}
let root = self.root()?;
let _lock = RootLock::exclusive(root)?;
ensure_reconciliation_root(root, &candidate)?;
let quarantine_root = match open_existing_dir(root.fd.as_raw_fd(), QUARANTINE_DIR) {
Ok(fd) => fd,
Err(ArtifactError::NotFound) => return Ok(ReconciliationMutation::AlreadyAbsent),
Err(error) => return Err(error),
};
let shard = match open_existing_dir(quarantine_root.as_raw_fd(), candidate.shard.as_bytes())
{
Ok(fd) => fd,
Err(ArtifactError::NotFound) => return Ok(ReconciliationMutation::AlreadyAbsent),
Err(error) => return Err(error),
};
match revalidate_candidate(&candidate, shard.as_raw_fd()) {
Ok(()) => {}
Err(ArtifactError::NotFound) => {
return fsync_mutation(&shard, ReconciliationMutation::AlreadyAbsent);
}
Err(error) => return Err(error),
}
match checkpointed("reconciliation_delete_unlink", || {
unlinkat(shard.as_raw_fd(), candidate.name.as_bytes())
}) {
Ok(()) => fsync_mutation(&shard, ReconciliationMutation::Deleted),
Err(ArtifactError::NotFound) => {
fsync_mutation(&shard, ReconciliationMutation::AlreadyAbsent)
}
Err(error) => Err(error),
}
}
fn recover_quarantine(
&self,
root: &crate::store::Root,
candidate: &ReconciliationCandidate,
) -> Result<ReconciliationMutation, ArtifactError> {
let final_root = open_existing_dir(root.fd.as_raw_fd(), namespace_name(FINAL_NAMESPACE))?;
let final_shard = open_existing_dir(final_root.as_raw_fd(), candidate.shard.as_bytes())?;
let quarantine_root = match open_existing_dir(root.fd.as_raw_fd(), QUARANTINE_DIR) {
Ok(fd) => fd,
Err(ArtifactError::NotFound) => return Err(ArtifactError::NotFound),
Err(error) => return Err(error),
};
let shard = match open_existing_dir(quarantine_root.as_raw_fd(), candidate.shard.as_bytes())
{
Ok(fd) => fd,
Err(ArtifactError::NotFound) => return Err(ArtifactError::NotFound),
Err(error) => return Err(error),
};
revalidate_moved_candidate(candidate, shard.as_raw_fd())?;
match fsync_quarantine_dirs(&final_shard, &shard) {
Ok(()) => Ok(ReconciliationMutation::AlreadyQuarantined),
Err(_) => Ok(ReconciliationMutation::Retryable),
}
}
/// Streams temporary entries, reading at most `scan_budget` directory
/// entries and returning at most `result_limit` stale capabilities.
pub fn scan_stale_temps(
@@ -147,6 +453,187 @@ impl ArtifactStore {
}
}
fn reconciliation_cursor(
root_dev: u64,
root_ino: u64,
namespace: u8,
shard: u8,
position: i64,
) -> ReconciliationCursor {
ReconciliationCursor {
root_dev,
root_ino,
namespace,
shard,
position,
}
}
fn namespace_name(namespace: u8) -> &'static [u8] {
match namespace {
FINAL_NAMESPACE => b"sha256",
QUARANTINE_NAMESPACE => QUARANTINE_DIR,
_ => unreachable!("validated reconciliation namespace"),
}
}
fn advance_shard(cursor: &mut ReconciliationCursor) {
debug_assert_ne!(cursor.shard, u8::MAX);
cursor.shard += 1;
cursor.position = 0;
}
fn advance_namespace(cursor: &mut ReconciliationCursor) {
cursor.namespace = cursor.namespace.saturating_add(1);
cursor.shard = 0;
cursor.position = 0;
}
fn consume_scan_work(report: &mut ReconciliationScan, remaining: &mut usize) {
debug_assert!(*remaining > 0);
*remaining -= 1;
report.scanned += 1;
}
fn ensure_reconciliation_root(
root: &crate::store::Root,
candidate: &ReconciliationCandidate,
) -> Result<(), ArtifactError> {
ensure_root_unchanged(root)?;
if candidate.root_dev != root.dev
|| candidate.root_ino != root.ino
|| candidate.shard.len() != 2
|| !candidate.shard.bytes().all(is_lower_hex)
|| !valid_final_name(&candidate.name, &candidate.shard)
{
return Err(ArtifactError::UnsafeRoot);
}
Ok(())
}
fn revalidate_candidate(
candidate: &ReconciliationCandidate,
shard: i32,
) -> Result<(), ArtifactError> {
let shard_stat = stat_fd(shard)?;
if shard_stat.st_dev != candidate.shard_dev || shard_stat.st_ino != candidate.shard_ino {
return Err(ArtifactError::UnsafeRoot);
}
let stat = nofollow_stat(shard, &candidate.name)?;
if stat.st_dev != candidate.dev
|| stat.st_ino != candidate.ino
|| stat.st_mtime != candidate.modified_seconds
|| stat.st_mtime_nsec != candidate.modified_nanoseconds
{
return Err(ArtifactError::UnsafeRoot);
}
check_file(&stat).map_err(|_| ArtifactError::UnsafeRoot)
}
/// Recovery follows a successful cross-directory rename. The candidate remains
/// bound to the original final shard, so the quarantine shard identity cannot
/// equal its recorded shard identity; only the canonical name and inode are
/// revalidated here.
fn revalidate_moved_candidate(
candidate: &ReconciliationCandidate,
shard: i32,
) -> Result<(), ArtifactError> {
let stat = nofollow_stat(shard, &candidate.name)?;
if stat.st_dev != candidate.dev
|| stat.st_ino != candidate.ino
|| stat.st_mtime != candidate.modified_seconds
|| stat.st_mtime_nsec != candidate.modified_nanoseconds
{
return Err(ArtifactError::UnsafeRoot);
}
check_file(&stat).map_err(|_| ArtifactError::UnsafeRoot)
}
fn fsync_mutation(
shard: &std::os::fd::OwnedFd,
success: ReconciliationMutation,
) -> Result<ReconciliationMutation, ArtifactError> {
match checkpointed("reconciliation_delete_fsync", || {
fsync_fd(shard.as_raw_fd())
}) {
Ok(()) => Ok(success),
Err(_) => Ok(ReconciliationMutation::Retryable),
}
}
/// Persist both directory entry changes even if the first fsync reports an
/// error. Either failure remains an ambiguity, but skipping the second fsync
/// would needlessly enlarge the crash window.
fn fsync_quarantine_dirs(
final_shard: &std::os::fd::OwnedFd,
quarantine_shard: &std::os::fd::OwnedFd,
) -> Result<(), ArtifactError> {
let source = checkpointed("reconciliation_quarantine_source_fsync", || {
fsync_fd(final_shard.as_raw_fd())
});
let destination = checkpointed("reconciliation_quarantine_destination_fsync", || {
fsync_fd(quarantine_shard.as_raw_fd())
});
source.and(destination)
}
fn valid_final_name(name: &str, shard: &str) -> bool {
name.len() == 64 && name.starts_with(shard) && name.bytes().all(is_lower_hex)
}
struct DirectoryStream {
directory: *mut libc::DIR,
}
impl DirectoryStream {
fn open(parent: i32, position: i64) -> Result<Self, ArtifactError> {
let duplicate = crate::store::duplicate_fd(parent)?;
let raw = std::os::fd::IntoRawFd::into_raw_fd(duplicate);
let directory = unsafe { libc::fdopendir(raw) };
if directory.is_null() {
unsafe {
libc::close(raw);
}
return Err(ArtifactError::Storage);
}
if position != 0 {
unsafe {
libc::seekdir(directory, position as libc::c_long);
}
}
Ok(Self { directory })
}
fn next(&mut self) -> Result<Option<(Vec<u8>, i64)>, ArtifactError> {
loop {
unsafe {
*libc::__errno_location() = 0;
}
let entry = unsafe { libc::readdir(self.directory) };
if entry.is_null() {
if unsafe { *libc::__errno_location() } != 0 {
return Err(ArtifactError::Storage);
}
return Ok(None);
}
let after = unsafe { libc::telldir(self.directory) } as i64;
let name = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes();
if name == b"." || name == b".." {
continue;
}
return Ok(Some((name.to_vec(), after)));
}
}
}
impl Drop for DirectoryStream {
fn drop(&mut self) {
unsafe {
libc::closedir(self.directory);
}
}
}
fn list_names(
parent: i32,
remaining: &mut usize,
+3 -2
View File
@@ -18,8 +18,9 @@ mod store;
pub mod test_support;
pub use error::ArtifactError;
pub use housekeeping::{StaleTemp, TempScan};
pub use housekeeping::{ReconciliationCandidate, ReconciliationCursor, StaleTemp, TempScan};
pub use model::{
ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, RegisteredArtifact, StoredArtifact,
ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, ReconciliationMutation,
ReconciliationNamespace, ReconciliationScan, RegisteredArtifact, StoredArtifact,
};
pub use store::ArtifactStore;
+49
View File
@@ -77,6 +77,55 @@ pub struct RegisteredArtifact {
pub(crate) size_bytes: usize,
}
/// Namespace in which reconciliation observed an artifact inode. It does not
/// disclose the artifact's digest or filesystem location.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReconciliationNamespace {
Final,
Quarantine,
}
/// Bounded, redacted progress report from a reconciliation page.
///
/// `continuation` is opaque evidence for resuming an unchanged namespace. It
/// intentionally cannot be created from a caller-supplied path or digest.
#[derive(Debug)]
pub struct ReconciliationScan {
pub scanned: usize,
pub malformed: usize,
pub unsafe_entries: usize,
pub final_entries: usize,
pub quarantined_entries: usize,
pub continuation: Option<crate::ReconciliationCursor>,
}
impl ReconciliationScan {
pub(crate) fn empty(continuation: Option<crate::ReconciliationCursor>) -> Self {
Self {
scanned: 0,
malformed: 0,
unsafe_entries: 0,
final_entries: 0,
quarantined_entries: 0,
continuation,
}
}
}
/// The durable/retryable result of a reconciliation mutation.
///
/// A successful rename or unlink followed by an fsync failure is intentionally
/// reported as `Retryable`: a subsequent bounded scan determines the durable
/// state without treating the operation as a failed no-op.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReconciliationMutation {
Quarantined,
AlreadyQuarantined,
Deleted,
AlreadyAbsent,
Retryable,
}
impl RegisteredArtifact {
pub fn artifact_ref(&self) -> &ArtifactRef {
&self.artifact_ref
+14 -5
View File
@@ -16,6 +16,7 @@ use sha2::{Digest, Sha256};
use crate::{ArtifactError, ArtifactRef, MAX_ARTIFACT_BYTES, RegisteredArtifact, StoredArtifact};
const SHA_DIR: &[u8] = b"sha256";
pub(crate) const QUARANTINE_DIR: &[u8] = b"quarantine";
const TEMP_PREFIX: &str = ".crank-artifact-tmp-v1-";
const RESOLVE_NO_SYMLINKS: u64 = 0x04;
static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);
@@ -382,7 +383,7 @@ fn check_private_dir(stat: &libc::stat) -> Result<(), ArtifactError> {
}
Ok(())
}
fn stat_fd(fd: i32) -> Result<libc::stat, ArtifactError> {
pub(crate) fn stat_fd(fd: i32) -> Result<libc::stat, ArtifactError> {
let mut stat = unsafe { std::mem::zeroed() };
if unsafe { libc::fstat(fd, &mut stat) } != 0 {
Err(ArtifactError::Storage)
@@ -390,7 +391,7 @@ fn stat_fd(fd: i32) -> Result<libc::stat, ArtifactError> {
Ok(stat)
}
}
fn check_file(stat: &libc::stat) -> Result<(), ArtifactError> {
pub(crate) fn check_file(stat: &libc::stat) -> Result<(), ArtifactError> {
if (stat.st_mode & libc::S_IFMT) != libc::S_IFREG
|| stat.st_uid != unsafe { libc::geteuid() }
|| stat.st_nlink != 1
@@ -508,7 +509,7 @@ fn write_all(fd: &OwnedFd, bytes: &[u8]) -> Result<(), ArtifactError> {
/// Debug/test-only fault seam coupled to one syscall wrapper. `Fail` models a
/// syscall that did not run; crash/hold actions happen only after success.
fn checkpointed<T>(
pub(crate) fn checkpointed<T>(
stage: &str,
operation: impl FnOnce() -> Result<T, ArtifactError>,
) -> Result<T, ArtifactError> {
@@ -608,14 +609,22 @@ fn chmod_read_only(fd: i32) -> Result<(), ArtifactError> {
}
}
fn rename_no_replace(parent: i32, old: &[u8], new: &[u8]) -> Result<bool, ArtifactError> {
rename_no_replace_at(parent, old, parent, new)
}
pub(crate) fn rename_no_replace_at(
old_parent: i32,
old: &[u8],
new_parent: i32,
new: &[u8],
) -> Result<bool, ArtifactError> {
let old = c_name(old)?;
let new = c_name(new)?;
let result = unsafe {
libc::syscall(
libc::SYS_renameat2,
parent,
old_parent,
old.as_ptr(),
parent,
new_parent,
new.as_ptr(),
libc::RENAME_NOREPLACE,
)