gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 06:18:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: integrate main disk dispatch with physical admission

Browse Source
This commit is contained in:
overtrue
2026-10-25 15:35:27 +08:00
parent 5158b16917 124cbfc00a
commit 136296c76b
31 changed files with 3066 additions and 726 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/concurrency/src/adaptive/permit_set.rs
+9 -9
View File
@@ -611,7 +611,7 @@ mod tests {
let node = node(1);
let topology = AdaptivePermitSetOwner::new(maximum(1), vec![a.clone(), b.clone()]).expect("topology");
let parent = topology
.select(&[a.clone()])
.select(std::slice::from_ref(&a))
.expect("object device")
.try_acquire_with_node(&node)
.expect("object admission");
@@ -643,7 +643,7 @@ mod tests {
let node = node(1);
let topology = AdaptivePermitSetOwner::new(maximum(1), vec![a.clone(), b.clone(), c.clone()]).expect("topology");
let parent = topology
.select(&[a.clone()])
.select(std::slice::from_ref(&a))
.expect("object device")
.try_acquire_with_node(&node)
.expect("object admission");
@@ -677,15 +677,15 @@ mod tests {
let node = node(1);
let topology = AdaptivePermitSetOwner::new(maximum(1), vec![a.clone(), b.clone()]).expect("topology");
let parent = topology
.select(&[a.clone()])
.select(std::slice::from_ref(&a))
.expect("object device")
.try_acquire_with_node(&node)
.expect("object admission");
let native = AdaptivePermitSetOwner::new(maximum(1), vec![b.clone()]).expect("auxiliary disk");
let replacement = AdaptivePermitSetOwner::new(maximum(1), vec![a.clone(), b.clone()]).expect("new topology");
for attachment in [
replacement.select(&[b.clone()]).expect("new attachment"),
topology.select(&[a.clone()]).expect("wrong leaves"),
replacement.select(std::slice::from_ref(&b)).expect("new attachment"),
topology.select(std::slice::from_ref(&a)).expect("wrong leaves"),
] {
assert_eq!(
native
@@ -695,7 +695,7 @@ mod tests {
);
assert_eq!([a.snapshot().active, b.snapshot().active], [1, 0]);
}
let attachment = topology.select(&[b.clone()]).expect("correct attachment");
let attachment = topology.select(std::slice::from_ref(&b)).expect("correct attachment");
let acquired = native
.try_capture_or_acquire_from(parent.resources(), &attachment)
.expect("new native unit");
@@ -719,7 +719,7 @@ mod tests {
let node = node(1);
let topology = AdaptivePermitSetOwner::new(maximum(2), vec![a.clone(), b.clone()]).expect("topology");
let parent = topology
.select(&[a.clone()])
.select(std::slice::from_ref(&a))
.expect("object device")
.try_acquire_with_node(&node)
.expect("object admission");
@@ -741,12 +741,12 @@ mod tests {
let node = node(1);
let topology = AdaptivePermitSetOwner::new(maximum(1), vec![a.clone(), b.clone()]).expect("topology");
let parent = topology
.select(&[a.clone()])
.select(std::slice::from_ref(&a))
.expect("object device")
.try_acquire_with_node(&node)
.expect("object admission");
let native = AdaptivePermitSetOwner::new(maximum(8), vec![b.clone()]).expect("auxiliary carrier");
let attachment = topology.select(&[b.clone()]).expect("attachment");
let attachment = topology.select(std::slice::from_ref(&b)).expect("attachment");
topology.extending[parent.resources().slot].store(true, Ordering::Release);
assert_eq!(
native
crates/concurrency/src/workers.rs
+55 -2
View File
@@ -49,12 +49,28 @@ impl Workers {
/// Acquire a worker slot, waiting until one becomes available.
pub async fn take(&self) {
struct AcquireCancelGuard<'a> {
workers: &'a Workers,
completed: bool,
}
impl Drop for AcquireCancelGuard<'_> {
fn drop(&mut self) {
if !self.completed {
self.workers.in_use.send_modify(|_| {});
}
}
}
let mut guard = AcquireCancelGuard {
workers: self,
completed: false,
};
let permit = match self.semaphore.acquire().await {
Ok(permit) => permit,
// The semaphore is owned by `self` and never closed.
Err(_) => unreachable!("worker semaphore is never closed"),
};
permit.forget(); // returned manually via give()
guard.completed = true;
self.in_use.send_modify(|held| *held += 1);
trace!(
event = "worker_slot.acquire",
@@ -82,6 +98,7 @@ impl Workers {
});
if released {
self.semaphore.add_permits(1);
self.in_use.send_modify(|_| {});
}
trace!(
event = "worker_slot.release",
@@ -99,7 +116,9 @@ impl Workers {
pub async fn wait(&self) {
let mut rx = self.in_use.subscribe();
// The sender is owned by `self`, so it outlives this borrow.
let _ = rx.wait_for(|&held| held == 0).await;
let _ = rx
.wait_for(|&held| held == 0 && self.semaphore.available_permits() == self.limit)
.await;
debug!(
event = "worker_slot.wait",
component = "concurrency",
@@ -114,7 +133,9 @@ impl Workers {
/// Return the current number of available worker slots.
pub async fn available(&self) -> usize {
self.limit.saturating_sub(*self.in_use.borrow())
self.limit
.saturating_sub(*self.in_use.borrow())
.min(self.semaphore.available_permits())
}
}
@@ -217,4 +238,36 @@ mod tests {
}
assert_eq!(workers.available().await, 0);
}
#[tokio::test]
async fn test_wait_does_not_complete_while_permit_is_in_transit_to_queued_taker() {
use std::future::Future;
use std::pin::pin;
use std::task::{Context, Poll, Waker};
let workers = Workers::new(1).unwrap();
workers.take().await;
let mut taker = pin!(workers.take());
let mut cx = Context::from_waker(Waker::noop());
assert!(matches!(taker.as_mut().poll(&mut cx), Poll::Pending));
let mut waiter = pin!(workers.wait());
assert!(matches!(waiter.as_mut().poll(&mut cx), Poll::Pending));
// Hand off the slot to the queued taker before `taker` is polled again.
workers.give().await;
assert_eq!(workers.available().await, 0);
assert!(
matches!(waiter.as_mut().poll(&mut cx), Poll::Pending),
"wait() must not complete while a permit is in transit to a woken taker"
);
assert!(matches!(taker.as_mut().poll(&mut cx), Poll::Ready(())));
assert!(matches!(waiter.as_mut().poll(&mut cx), Poll::Pending));
workers.give().await;
assert!(matches!(waiter.as_mut().poll(&mut cx), Poll::Ready(())));
assert_eq!(workers.available().await, 1);
}
}
crates/e2e_test/src/distributed/multipart_fence_acceptance_test.rs
+50 -5
View File
@@ -162,7 +162,12 @@ where
|| error.as_service_error()?.meta().code() != Some("ServiceUnavailable")
|| !matches!(
response.headers().get("x-rustfs-error-cause"),
Some("namespace_lock_timeout" | "namespace_lock_unavailable" | "storage_write_quorum")
Some(
"namespace_lock_timeout"
| "namespace_lock_unavailable"
| "storage_write_quorum"
| "storage_read_quorum"
)
)
{
return None;
@@ -377,6 +382,42 @@ async fn wait_for_lock_peer_recovery(proxies: &[RpcFaultProxy], client: &Client,
.await
}
async fn wait_for_storage_peer_recovery(
proxies: &[RpcFaultProxy],
client: &Client,
bucket: &str,
upload: &str,
drives_per_node: usize,
) -> TestResult {
let scope = upload_scope(upload)?;
tokio::time::timeout(
DEADLINE,
wait_until(
DEADLINE,
|| async {
let observers = proxies[1..]
.iter()
.map(|proxy| proxy.pause(&["ReadVersion"], usize::MAX, Some(&scope)))
.collect::<Result<Vec<_>, _>>()?;
for observer in &observers {
observer.release();
}
// A fresh absent key forces a full metadata read without mutating
// this upload. Count distinct configured disk routes, not retries
// from the same healthy disk or just lock-RPC reachability.
let key = format!("storage-ready-{}", uuid::Uuid::new_v4());
let _ = client.list_parts().bucket(bucket).key(key).upload_id(upload).send().await;
Ok(observers
.iter()
.all(|observer| observer.forwarded_routes() >= drives_per_node))
},
"all storage disk routes recover after fault retirement",
),
)
.await
.map_err(|_| "storage route recovery deadline exceeded".to_string())?
}
async fn phase_quorum_matrix(
proxies: &[RpcFaultProxy],
clients: &[Client],
@@ -417,7 +458,9 @@ async fn phase_quorum_matrix(
let request = clients[0].create_multipart_upload().bucket(bucket).key(&key);
let result = with_paused_peers(proxies, methods, &limits, bucket, async move { request.send().await }).await?;
if let Some(upload) = quorum_outcome(result, unavailable == unavailable_at_quorum, locks)? {
abort_after_fault(&clients[0], bucket, &key, upload.upload_id().ok_or("missing upload ID")?, false).await?;
let upload_id = upload.upload_id().ok_or("missing upload ID")?;
wait_for_storage_peer_recovery(proxies, &clients[0], bucket, upload_id, drives_per_node).await?;
abort_after_fault(&clients[0], bucket, &key, upload_id, false).await?;
}
} else {
let upload = clients[0]
@@ -520,9 +563,11 @@ async fn phase_quorum_matrix(
}
_ => unreachable!(),
}
// Cleanup must recover even after a below-quorum mutation;
// only an already retired session may return NoSuchUpload.
abort_after_fault(&clients[0], bucket, &key, upload_id, true).await?;
wait_for_storage_peer_recovery(proxies, &clients[0], bucket, upload_id, drives_per_node).await?;
// A rejected Complete must retain authority for ordinary
// Abort. NoSuchUpload must not hide partial staging loss.
let allow_missing = phase != "complete" || unavailable == unavailable_at_quorum;
abort_after_fault(&clients[0], bucket, &key, upload_id, allow_missing).await?;
recover_after_fault(&format!("DeleteObject {key} after {phase}"), || {
clients[0].delete_object().bucket(bucket).key(&key).send()
})
crates/e2e_test/src/rpc_fault_proxy.rs
+60 -18
View File
@@ -23,8 +23,9 @@ use hyper_util::{
server::conn::auto,
};
use prost::Message as _;
use rustfs_protos::proto_gen::node_service::DeleteRequest;
use rustfs_protos::proto_gen::node_service::{DeleteRequest, ReadVersionRequest};
use std::{
collections::HashSet,
io,
net::SocketAddr,
sync::{
@@ -56,6 +57,7 @@ struct PauseState {
matched: AtomicUsize,
blocked: AtomicUsize,
forwarded: AtomicUsize,
forwarded_routes: Mutex<HashSet<String>>,
reached: Notify,
release: watch::Sender<Disposition>,
}
@@ -77,6 +79,10 @@ impl RpcPause {
self.state.forwarded.load(Ordering::SeqCst)
}
pub(crate) fn forwarded_routes(&self) -> usize {
self.state.forwarded_routes.lock().expect("RPC route observations").len()
}
pub(crate) async fn wait_for_hits(&self, count: usize) -> io::Result<()> {
tokio::time::timeout(Duration::from_secs(20), async {
loop {
@@ -186,6 +192,7 @@ impl RpcFaultProxy {
matched: AtomicUsize::new(0),
blocked: AtomicUsize::new(0),
forwarded: AtomicUsize::new(0),
forwarded_routes: Mutex::new(HashSet::new()),
reached: Notify::new(),
release,
});
@@ -197,24 +204,23 @@ impl RpcFaultProxy {
}
}
fn scope_matches(method: &str, body: &[u8], scope: &[u8]) -> io::Result<bool> {
if method != "Delete" {
return Ok(body.windows(scope.len()).any(|window| window == scope));
}
// Abort's recursive root delete must not be confused with transaction or
// part cleanup below that root. This is a unary, uncompressed gRPC frame.
fn unary_payload(body: &[u8]) -> io::Result<&[u8]> {
if body.len() < 5 || body[0] != 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"scoped Delete needs an uncompressed unary frame",
));
return Err(io::Error::new(io::ErrorKind::InvalidData, "scoped RPC needs an uncompressed unary frame"));
}
let mut encoded_length = [0; 4];
encoded_length.copy_from_slice(&body[1..5]);
if u32::try_from(body.len() - 5).ok() != Some(u32::from_be_bytes(encoded_length)) {
return Err(io::Error::new(io::ErrorKind::InvalidData, "scoped Delete frame length mismatch"));
return Err(io::Error::new(io::ErrorKind::InvalidData, "scoped RPC frame length mismatch"));
}
let request = DeleteRequest::decode(&body[5..]).map_err(io::Error::other)?;
Ok(&body[5..])
}
fn scope_matches(method: &str, body: &[u8], scope: &[u8]) -> io::Result<bool> {
if method != "Delete" {
return Ok(body.windows(scope.len()).any(|window| window == scope));
}
let request = DeleteRequest::decode(unary_payload(body)?).map_err(io::Error::other)?;
let options: serde_json::Value = serde_json::from_str(&request.options).map_err(io::Error::other)?;
Ok(request.volume == ".rustfs.sys/multipart"
&& request.path.trim_end_matches('/').as_bytes() == scope
@@ -233,13 +239,22 @@ async fn forward_request(
if let Some(pause) = pause {
let method = parts.uri.path().strip_prefix("/node_service.NodeService/");
if let Some(method) = method.filter(|name| pause.methods.contains(name)) {
let mut route = None;
let matches = if let Some(scope) = pause.scope.as_deref() {
// Inspect bounded routing data while retaining the exact
// original bytes covered by the RPC signature.
// Inspect bounded control routing while preserving the exact
// signed bytes and trailers forwarded to the backend.
let collected = Limited::new(body, 1024 * 1024).collect().await.map_err(io::Error::other)?;
let trailers = collected.trailers().cloned();
let bytes = collected.to_bytes();
let matches = scope_matches(method, &bytes, scope)?;
if matches && method == "ReadVersion" {
let disk = ReadVersionRequest::decode(unary_payload(&bytes)?)
.map_err(io::Error::other)?
.disk;
if !disk.is_empty() {
route = Some(disk);
}
}
body = Full::new(bytes)
.map_err(|never| match never {})
.with_trailers(async move { trailers.map(Ok::<_, io::Error>) })
@@ -264,7 +279,7 @@ async fn forward_request(
if disposition == Disposition::Discard {
return Err(io::Error::new(io::ErrorKind::ConnectionAborted, "injected RPC packet loss"));
}
observed = Some(pause);
observed = Some((pause, route));
}
}
}
@@ -286,11 +301,14 @@ async fn forward_request(
.request(Request::from_parts(parts, body))
.await
.map_err(io::Error::other)?;
if let Some(pause) = observed
if let Some((pause, route)) = observed
&& response.status().is_success()
&& response.headers().get("grpc-status").is_none_or(|status| status == "0")
{
pause.forwarded.fetch_add(1, Ordering::SeqCst);
if let Some(route) = route {
pause.forwarded_routes.lock().expect("RPC route observations").insert(route);
}
}
Ok(response)
}
@@ -487,6 +505,30 @@ mod tests {
.expect("observer forwards without pausing");
another.into_body().collect().await.expect("second body");
assert_eq!(observer.forwarded(), 2);
drop(observer);
let routes = proxy
.pause(&["ReadVersion"], usize::MAX, Some("recovery-route"))
.expect("disk-route observer");
routes.release();
for (disk, expected) in [("drive-a", 1), ("drive-a", 1), ("drive-b", 2)] {
let request = ReadVersionRequest {
disk: disk.to_owned(),
path: "recovery-route".to_owned(),
..Default::default()
}
.encode_to_vec();
let mut frame = vec![0];
frame.extend_from_slice(&u32::try_from(request.len()).expect("small frame").to_be_bytes());
frame.extend_from_slice(&request);
call_body(&client, address, "ReadVersion", Bytes::from(frame))
.await
.expect("routed read")
.into_body()
.collect()
.await
.expect("routed response");
assert_eq!(routes.forwarded_routes(), expected, "repeated traffic from one disk is only one voter");
}
})
.await
.expect("passive observer deadline");
@@ -549,7 +591,7 @@ mod tests {
let root = b"hash/upload";
assert!(scope_matches("Delete", &frame(".rustfs.sys/multipart", "hash/upload"), root).expect("root frame"));
assert!(scope_matches("Delete", &frame(".rustfs.sys/multipart", "hash/upload/"), root).expect("directory frame"));
for path in ["hash/upload/part.1", "hash/upload-other"] {
for path in ["hash/upload/part.1", "hash/upload/.abort-authority", "hash/upload-other"] {
assert!(!scope_matches("Delete", &frame(".rustfs.sys/multipart", path), root).expect("other path"));
}
assert!(!scope_matches("Delete", &frame(".rustfs.sys/tmp", "hash/upload"), root).expect("other volume"));
crates/ecstore/src/core/pools.rs
+40 -10
View File
@@ -12431,6 +12431,22 @@ fn resolve_data_movement_lifecycle_expiry_result(action: IlmAction, apply_action
)))
}
const DATA_MOVEMENT_DELETE_ALL_DEFERRED_REASON: &str = "delete_all_needs_bucket_write_lock";
fn log_data_movement_lifecycle_expiry_deferred(bucket: &str, object: &str, action: IlmAction, reason: &str) {
warn!(
event = EVENT_DECOMMISSION_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_POOLS,
state = "lifecycle_expiry_deferred",
bucket = %bucket,
object = %object,
action = ?action,
reason,
"Data movement deferred a version's lifecycle expiry; moving the version"
);
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn should_skip_lifecycle_for_data_movement(
store: Arc<ECStore>,
@@ -12490,6 +12506,21 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement(
}
let applied = !apply_actions
|| match lock_lost_signal {
// A delete-all is a prefix delete that takes the bucket
// lifecycle write lock, which the lifecycle read locks held
// by the inline apply and by the rebalance entry's bucket
// incarnation fence would block until timeout. The version
// is still live, so it moves like any other version and
// lifecycle expires it on the target.
Some(_) if action.delete_all() => {
log_data_movement_lifecycle_expiry_deferred(
bucket,
&version.name,
action,
DATA_MOVEMENT_DELETE_ALL_DEFERRED_REASON,
);
return Ok(false);
}
Some(signal) => {
apply_expiry_rule_for_data_movement(store, &event, event_source, &object_info, Some(signal)).await
}
@@ -12515,16 +12546,15 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement(
// The version is still live, so it moves like any
// other version and lifecycle expires it on the
// target.
warn!(
event = EVENT_DECOMMISSION_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_POOLS,
state = "lifecycle_expiry_deferred",
bucket = %bucket,
object = %version.name,
action = ?action,
reason = if delete_all { "delete_all_needs_bucket_write_lock" } else { "not_applied" },
"Data movement deferred a transitioned version's lifecycle expiry; moving the version"
log_data_movement_lifecycle_expiry_deferred(
bucket,
&version.name,
action,
if delete_all {
DATA_MOVEMENT_DELETE_ALL_DEFERRED_REASON
} else {
"not_applied"
},
);
return Ok(false);
}
crates/ecstore/src/disk/disk_store.rs
+2 -2
View File
@@ -4039,7 +4039,7 @@ mod tests {
);
let parent = if auxiliary {
shared
.select(&[first.clone()])
.select(std::slice::from_ref(&first))
.expect("initial object device")
.try_acquire_with_node(&node)
} else {
@@ -4281,7 +4281,7 @@ mod tests {
}),
);
let parent = shared
.select(&[a.clone()])
.select(std::slice::from_ref(&a))
.expect("object on A")
.try_acquire_with_node(&node)
.expect("object admission");
crates/ecstore/src/disk/io_admission/physical_sampler.rs
+37 -1
View File
@@ -308,6 +308,42 @@ impl Drop for PhysicalDiskSampler {
}
impl PhysicalDiskSample {
/// Supply one synthetic device for real-storage admission tests on any OS.
/// Publication, shared ownership and lease expiry still use the real sampler.
#[cfg(any(test, feature = "test-util"))]
pub fn with_test_single_device(mut self) -> Self {
assert!(!self.targets.is_empty(), "the fixture must contain local disks");
let now = Instant::now();
let identity = PhysicalDeviceIdentity {
major: 259,
minor: 0,
generation: 1,
};
self.resolved = std::mem::take(&mut self.targets)
.into_iter()
.map(|target| Resolved {
binding: target.binding,
identities: Some(vec![identity]),
began: now,
ended: now,
})
.collect();
self.devices = BTreeMap::from([(
identity,
Some(Observation {
began: now,
ended: now,
ios: 0,
ticks: 0,
queued_ticks: 0,
in_flight: 0,
maximum: 1,
}),
)]);
self.finished = Some(now);
self
}
/// Snapshot current attachments before blocking sysfs reads. A busy set
/// cannot hold the sampler indefinitely or renew stale capacity.
pub fn collect(store: &ECStore) -> Self {
@@ -763,7 +799,7 @@ mod tests {
sample.resolved[0].identities = None;
sample.heal_sets = vec![HealSetSample {
binding: first.clone(),
disks: Some(vec![a.clone()]),
disks: Some(vec![a]),
}];
sampler.publish(sample, ResourcePressure::Stable);
assert!(first.plan().is_none(), "unknown local disks never become empty selections");
crates/ecstore/src/disk/local.rs
+330 -357
View File
@@ -324,151 +324,6 @@ fn read_metadata_only_std(opened: std::io::Result<std::fs::File>) -> Result<(Vec
Ok((data, modtime))
}
const MAX_METADATA_READ_WORKERS_PER_DISK: usize = 2;
type ReadAllDataResult = core::result::Result<(Vec<u8>, Option<OffsetDateTime>), ReadAllError>;
type ReadMetadataOnlyResult = Result<(Vec<u8>, Option<OffsetDateTime>)>;
enum MetadataReadOp {
ReadAll {
tx: tokio::sync::oneshot::Sender<ReadAllDataResult>,
},
ReadMetadataOnly {
tx: tokio::sync::oneshot::Sender<ReadMetadataOnlyResult>,
},
#[cfg(test)]
TestCustom {
is_closed: Box<dyn Fn() -> bool + Send>,
run: Box<dyn FnOnce() + Send>,
},
}
struct MetadataReadJob {
resource_unit: Option<ResourceIoUnit>,
path: PathBuf,
io_permit: Option<DiskIoPermit>,
op: MetadataReadOp,
}
impl MetadataReadJob {
#[inline]
fn is_cancelled(&self) -> bool {
match &self.op {
MetadataReadOp::ReadAll { tx } => tx.is_closed(),
MetadataReadOp::ReadMetadataOnly { tx } => tx.is_closed(),
#[cfg(test)]
MetadataReadOp::TestCustom { is_closed, .. } => is_closed(),
}
}
#[inline]
fn execute(self, root: &os::PublicationRoot) {
if self.is_cancelled() {
return;
}
let _resource_unit = self.resource_unit;
let _io_permit = self.io_permit;
match self.op {
MetadataReadOp::ReadAll { tx } => {
let _ = tx.send(read_all_data_std(root.open_readonly(&self.path)));
}
MetadataReadOp::ReadMetadataOnly { tx } => {
let _ = tx.send(read_metadata_only_std(root.open_readonly(&self.path)));
}
#[cfg(test)]
MetadataReadOp::TestCustom { run, .. } => run(),
}
}
}
#[derive(Default)]
struct MetadataReadQueueState {
jobs: std::collections::VecDeque<MetadataReadJob>,
active_workers: usize,
}
struct MetadataReadQueue {
root: os::PublicationRoot,
state: ParkingLotMutex<MetadataReadQueueState>,
}
struct MetadataReadWorkerGuard {
queue: Arc<MetadataReadQueue>,
disarmed: bool,
}
impl Drop for MetadataReadWorkerGuard {
fn drop(&mut self) {
if self.disarmed {
return;
}
let should_respawn = {
let mut state = self.queue.state.lock();
state.active_workers = state.active_workers.saturating_sub(1);
if !state.jobs.is_empty() && state.active_workers < MAX_METADATA_READ_WORKERS_PER_DISK {
state.active_workers += 1;
true
} else {
false
}
};
if should_respawn {
MetadataReadQueue::spawn_worker(Arc::clone(&self.queue));
}
}
}
impl MetadataReadQueue {
fn new(root: os::PublicationRoot) -> Arc<Self> {
Arc::new(Self {
root,
state: ParkingLotMutex::new(MetadataReadQueueState::default()),
})
}
fn enqueue(self: &Arc<Self>, mut job: MetadataReadJob) {
job.resource_unit = current_resource_unit();
let should_spawn = {
let mut state = self.state.lock();
state.jobs.push_back(job);
if state.active_workers < MAX_METADATA_READ_WORKERS_PER_DISK {
state.active_workers += 1;
true
} else {
false
}
};
if should_spawn {
Self::spawn_worker(Arc::clone(self));
}
}
fn spawn_worker(queue: Arc<Self>) {
drop(tokio::task::spawn_blocking(move || {
let mut guard = MetadataReadWorkerGuard {
queue: Arc::clone(&queue),
disarmed: false,
};
loop {
let Some(job) = ({
let mut state = queue.state.lock();
match state.jobs.pop_front() {
Some(job) => Some(job),
None => {
state.active_workers = state.active_workers.saturating_sub(1);
guard.disarmed = true;
None
}
}
}) else {
return;
};
job.execute(&queue.root);
}
}));
}
}
/// Reads the `xl.meta` header/metadata envelope and 5-byte CRC for directory
/// walks (matching `read_xl_meta_no_data_sync`) without reading trailing
/// `InlineData` object bodies. If the header or metadata envelope is corrupt or
@@ -692,48 +547,71 @@ async fn write_delete_rollback_file(
name: &'static str,
data: &[u8],
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
write_delete_rollback_files(io_root, volume, object_dir, rollback_dir, &[(name, data)], namespace_owner).await
}
async fn write_delete_rollback_files(
io_root: &Path,
volume: &str,
object_dir: &Path,
rollback_dir: Uuid,
files: &[(&'static str, &[u8])],
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
let backup_dir = object_dir.join(rollback_dir.to_string());
let path = backup_dir.join(name);
let lease = os::acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
let data = data.to_vec();
let mut entries = Vec::with_capacity(files.len());
// Publication and cleanup both acquire FINISHED before the quorum receipt.
for &(name, data) in files {
let path = backup_dir.join(name);
let lease = os::acquire_namespace_mutation_lease_with_owner(&path, namespace_owner.clone()).await;
entries.push((name, path, data.to_vec(), lease));
}
let Some(first_lease) = entries.first().map(|(_, _, _, lease)| Arc::clone(lease)) else {
return Ok(());
};
let sync = effective_durability(volume).syncs_commit_metadata();
let staging_root = io_root.join(RUSTFS_META_TMP_BUCKET);
let io_root = io_root.to_path_buf();
os::run_blocking_namespace_operation(lease, move || {
reject_local_disk_symlink_components(&io_root, &path).map_err(std::io::Error::from)?;
os::run_blocking_namespace_operation_with_sync(first_lease, sync, move || {
for (_, path, _, _) in &entries {
reject_local_disk_symlink_components(&io_root, path).map_err(std::io::Error::from)?;
}
reject_local_disk_symlink_components(&io_root, &staging_root).map_err(std::io::Error::from)?;
std::fs::create_dir_all(&backup_dir)?;
std::fs::create_dir_all(&staging_root)?;
#[cfg(test)]
run_owned_file_before_open(&path);
// tempfile's Windows `persist` hands both the staged and the final path
// straight to `MoveFileExW`, bypassing std's long-path handling, so a
// deep object key failed every delete at MAX_PATH (rustfs#8545).
let staging_root = rustfs_utils::extended_length_path(&staging_root)?;
let destination = rustfs_utils::extended_length_path(&path)?;
// Private directory staging is reclaimed by existing tmp cleanup after
// a crash and cannot leave a regular artifact in the user's bucket.
let staging_dir = tempfile::Builder::new()
.prefix(".delete-rollback-")
.tempdir_in(&staging_root)?;
let staged_path = staging_dir.path().join(name);
let mut staged_options = std::fs::OpenOptions::new();
staged_options.write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
staged_options.mode(0o600);
for (name, path, data, lease) in entries {
let _lease = lease;
#[cfg(test)]
run_owned_file_before_open(&path);
let destination = rustfs_utils::extended_length_path(&path)?;
let staged_path = staging_dir.path().join(name);
let mut staged_options = std::fs::OpenOptions::new();
staged_options.write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
staged_options.mode(0o600);
}
let mut staged = staged_options.open(&staged_path)?;
std::io::Write::write_all(&mut staged, &data)?;
if sync {
staged.sync_all()?;
}
// Close the writer before publishing via std::fs::rename, which applies
// verbatim long-path normalization and POSIX-semantics replacement on Windows.
drop(staged);
std::fs::rename(&staged_path, &destination)?;
}
let mut staged = staged_options.open(&staged_path)?;
std::io::Write::write_all(&mut staged, &data)?;
if sync {
staged.sync_all()?;
}
// Close the writer before publishing via std::fs::rename, which applies
// verbatim long-path normalization and POSIX-semantics replacement on Windows.
drop(staged);
std::fs::rename(&staged_path, &destination)?;
if sync {
os::fsync_dir_std(&backup_dir)?;
if let Some(parent) = backup_dir.parent() {
@@ -3100,6 +2978,7 @@ fn run_owned_file_before_open(path: &Path) {
.expect("test file open hook lock should not be poisoned")
.remove(path);
if let Some(hook) = hook {
let _pause = os::enter_blocking_test_hook();
hook();
}
}
@@ -6240,8 +6119,7 @@ pub struct LocalDisk {
exit_signal: Option<tokio::sync::broadcast::Sender<()>>,
io_backend: Arc<dyn LocalIoBackend>,
file_sync_permits: Arc<Semaphore>,
_blocking_queue: Arc<os::DiskBlockingQueue>,
metadata_read_queue: Arc<MetadataReadQueue>,
blocking_queue: Arc<os::DiskBlockingQueue>,
snapshot_leases: Arc<ParkingLotMutex<SnapshotLeaseRegistry>>,
}
@@ -6704,7 +6582,6 @@ impl LocalDisk {
// TODO(backlog): add O_DIRECT I/O support for performance-critical paths
// TODO(backlog): populate DiskInfo in constructor
let metadata_read_queue = MetadataReadQueue::new(publication_root.clone());
let mut disk = Self {
root: root.clone(),
publication_root,
@@ -6740,8 +6617,7 @@ impl LocalDisk {
)
.await,
file_sync_permits: os::disk_file_sync_limiter(&root),
_blocking_queue: os::register_disk_blocking_queue(&root, &io_root),
metadata_read_queue,
blocking_queue: os::register_disk_blocking_queue(&root, &io_root),
snapshot_leases: Arc::new(ParkingLotMutex::new(SnapshotLeaseRegistry::default())),
};
let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| {
@@ -7485,6 +7361,14 @@ impl LocalDisk {
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
crate::hp_guard!("LocalDisk::write_metadata");
if volume == super::RUSTFS_META_MULTIPART_BUCKET {
let recovery = self.io_get_object_path(volume, &format!("{path}/{}", super::MULTIPART_RECOVERY_DIR))?;
match fs::symlink_metadata(recovery).await {
Ok(_) => return Err(DiskError::FileNotFound),
Err(error) if error.kind() == ErrorKind::NotFound => {}
Err(error) => return Err(to_file_error(error).into()),
}
}
fi.validate_for_metadata_read()?;
let p = self.io_get_object_path(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str())?;
@@ -8332,7 +8216,7 @@ impl LocalDisk {
let receipt_path = transaction_path.join(DELETE_QUORUM_COMMITTED_FILE);
let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, namespace_owner.clone()).await;
let sync = effective_durability(volume).syncs_commit_metadata();
os::run_blocking_namespace_operation(lease, move || {
os::run_blocking_namespace_operation_with_sync(lease, sync, move || {
let receipt = match std::fs::OpenOptions::new().write(true).create_new(true).open(&receipt_path) {
Ok(receipt) => receipt,
Err(err) if err.kind() == ErrorKind::AlreadyExists => {
@@ -8434,7 +8318,7 @@ impl LocalDisk {
let metadata_path = self.io_get_object_path(volume, &format!("{object}/{STORAGE_FORMAT_FILE}"))?;
let lease = os::acquire_namespace_mutation_lease_with_owner(&file_path, Some(owner)).await;
let sync = effective_durability(volume).syncs_commit_metadata();
let result = os::run_blocking_namespace_operation(lease, move || {
let result = os::run_blocking_namespace_operation_with_sync(lease, sync, move || {
let bytes = std::fs::read(metadata_path)?;
let metadata = FileMeta::load(&bytes).map_err(|_| std::io::Error::from(ErrorKind::InvalidData))?;
if metadata
@@ -8492,7 +8376,7 @@ impl LocalDisk {
let data_path = file_path.clone();
let sync = effective_durability(volume).syncs_commit_metadata();
let keep_rollback_witnesses = opt.recoverable_delete;
os::run_blocking_namespace_operation(lease, move || {
os::run_blocking_namespace_operation_with_sync(lease, sync, move || {
let entries = match std::fs::read_dir(&data_path) {
Ok(entries) => entries,
Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()),
@@ -8761,19 +8645,32 @@ impl LocalDisk {
// - is_dir -> Error::FileNotFound (NOT to_file_error(EISDIR))
// - metadata failure -> to_file_error
// - parse failure -> propagated verbatim from read_xl_meta_no_data_sync (`?`)
let path = file_path.as_ref().to_path_buf();
let (tx, rx) = tokio::sync::oneshot::channel();
self.metadata_read_queue.enqueue(MetadataReadJob {
resource_unit: None,
path,
io_permit,
op: MetadataReadOp::ReadMetadataOnly { tx },
});
let (data, modtime) = rx.await.map_err(|_| DiskError::other("metadata read worker cancelled"))??;
let (data, modtime) = self
.run_metadata_read(io_permit, file_path.as_ref().to_path_buf(), read_metadata_only_std)
.await??;
Ok((data, modtime))
}
async fn run_metadata_read<T: Send + 'static>(
&self,
io_permit: Option<DiskIoPermit>,
path: PathBuf,
read_fn: impl FnOnce(std::io::Result<std::fs::File>) -> T + Send + 'static,
) -> Result<T> {
let root = self.publication_root.clone();
self.blocking_queue
.run_cancellable(move |is_cancelled| {
if is_cancelled() {
return Err(std::io::Error::other("metadata read worker cancelled"));
}
let _io_permit = io_permit;
Ok(read_fn(root.open_readonly(&path)))
})
.await
.map_err(|err| DiskError::other(err.to_string()))
}
#[hotpath::measure(impl_type = "LocalDisk")]
async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
// TODO(backlog): add configurable timeout for read_all_data operations
@@ -8939,15 +8836,9 @@ impl LocalDisk {
// succeeds the fd is valid, so fstat/read never return ENOENT. Hence
// gating the volume fallback on the open error alone is equivalent to
// the original code, where the fallback lived solely in the open match arm.
let path = file_path.as_ref().to_path_buf();
let (tx, rx) = tokio::sync::oneshot::channel();
self.metadata_read_queue.enqueue(MetadataReadJob {
resource_unit: None,
path,
io_permit,
op: MetadataReadOp::ReadAll { tx },
});
let res = rx.await.map_err(|_| DiskError::other("metadata read worker cancelled"))?;
let res = self
.run_metadata_read(io_permit, file_path.as_ref().to_path_buf(), read_all_data_std)
.await?;
let (bytes, modtime) = self.resolve_read_all_result(volume, volume_dir.as_ref(), res).await?;
@@ -9761,30 +9652,37 @@ impl LocalDisk {
}
let root = self.publication_root.clone();
tokio::task::spawn_blocking(move || {
let operation = move |is_cancelled: &dyn Fn() -> bool| {
if is_cancelled() {
return Err(std::io::Error::other("blocking namespace operation cancelled"));
}
#[cfg(test)]
run_owned_file_before_open(&path);
let (mut file, parent_handle) = if sync == SyncMode::FileAndDir {
root.create_truncate_with_parent(&path).map_err(to_file_error)?
root.create_truncate_with_parent(&path)?
} else {
(root.create_truncate(&path).map_err(to_file_error)?, None)
(root.create_truncate(&path)?, None)
};
std::io::Write::write_all(&mut file, buf.as_ref()).map_err(to_file_error)?;
std::io::Write::write_all(&mut file, buf.as_ref())?;
if sync != SyncMode::None {
file.sync_data().map_err(to_file_error)?;
file.sync_data()?;
// FileOnly callers rename the file away, so the tmp
// directory entry never needs to become durable.
if sync == SyncMode::FileAndDir
&& let Some(parent) = path.parent()
{
os::fsync_directory_handle_std(parent, parent_handle.as_ref()).map_err(to_file_error)?;
os::fsync_directory_handle_std(parent, parent_handle.as_ref())?;
}
}
Ok::<_, std::io::Error>(())
})
.await
.map_err(DiskError::from)??;
Ok(())
};
let result = if sync == SyncMode::None {
self.blocking_queue.run_cancellable(operation).await
} else {
self.blocking_queue.run_fsync_cancellable(operation).await
};
result.map_err(to_file_error)?;
}
}
@@ -9807,9 +9705,14 @@ impl LocalDisk {
let path = path.as_ref().to_path_buf();
let root = self.publication_root.clone();
File::from_std(
tokio::task::spawn_blocking(move || root.create_truncate(&path))
self.blocking_queue
.run_cancellable(move |is_cancelled| {
if is_cancelled() {
return Err(std::io::Error::other("blocking namespace operation cancelled"));
}
root.create_truncate(&path)
})
.await
.map_err(DiskError::from)?
.map_err(to_file_error)?,
)
} else {
@@ -11096,7 +10999,9 @@ impl LocalDisk {
if namespace_owner.is_some() {
let lease = os::acquire_namespace_mutation_lease_with_owner(&marker_path, namespace_owner.clone()).await;
let volume = volume.to_string();
let sync = os::run_blocking_namespace_operation(lease, move || {
// This operation resolves durability inside the worker, so reserve
// the sync lane even if the current bucket override is relaxed.
let sync = os::run_blocking_namespace_operation_with_sync(lease, true, move || {
#[cfg(test)]
run_owned_file_before_open(&marker_path);
let marker = std::fs::File::create(marker_path)?;
@@ -11137,35 +11042,63 @@ impl LocalDisk {
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
let object_path = self.io_get_object_path(volume, object)?;
let mut entries = match fs::read_dir(object_path).await {
Ok(entries) => entries,
Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()),
Err(err) => return Err(to_file_error(err).into()),
};
let prepared_path = self.io_get_object_path(volume, &format!("{object}/{rollback_dir}/{}", delete_recovery::PREPARED))?;
let reserved_name = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{rollback_dir}");
let recovery_owned =
fs::try_exists(self.io_get_object_path(volume, &format!("{object}/{rollback_dir}/{}", delete_recovery::PREPARED))?)
.await
.map_err(to_file_error)?;
let scan_object_path = object_path.clone();
let scan_reserved_name = reserved_name.clone();
let (matching_dirs, recovery_owned) = self
.blocking_queue
.run_cancellable(move |is_cancelled| {
let entries = match std::fs::read_dir(&scan_object_path) {
Ok(entries) => entries,
Err(err) if err.kind() == ErrorKind::NotFound => return Ok((Vec::new(), false)),
Err(err) => return Err(err),
};
let mut matching = Vec::new();
for (idx, entry) in entries.enumerate() {
if idx & 63 == 0 && is_cancelled() {
return Err(std::io::Error::other("blocking namespace operation cancelled"));
}
let entry = entry?;
if !entry.file_type()?.is_dir()
|| !entry.file_name().to_str().is_some_and(|name| Uuid::parse_str(name).is_ok())
{
continue;
}
let path = entry.path();
match std::fs::symlink_metadata(path.join(&scan_reserved_name)) {
Ok(_) => matching.push(path),
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(err),
}
}
let recovery_owned = if matching.is_empty() {
false
} else {
std::fs::exists(&prepared_path)?
};
Ok((matching, recovery_owned))
})
.await
.map_err(to_file_error)?;
if matching_dirs.is_empty() {
return Ok(());
}
let prefix = if recovery_owned {
RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX
} else {
DELETE_DATA_DIR_MARKER_PREFIX
};
let committed_name = format!("{prefix}{rollback_dir}");
while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? {
if !entry.file_type().await.map_err(to_file_error)?.is_dir()
|| !entry.file_name().to_str().is_some_and(|name| Uuid::parse_str(name).is_ok())
{
continue;
}
let reserved_path = entry.path().join(&reserved_name);
match os::rename_with_namespace_owner(&reserved_path, &entry.path().join(&committed_name), namespace_owner.clone())
let sync = effective_durability(volume).syncs_commit_metadata();
for entry_path in matching_dirs {
let reserved_path = entry_path.join(&reserved_name);
match os::rename_with_namespace_owner(&reserved_path, &entry_path.join(&committed_name), namespace_owner.clone())
.await
{
Ok(()) => {
if effective_durability(volume).syncs_commit_metadata() {
os::fsync_dir_with_owner(&entry.path(), namespace_owner.clone())
if sync {
os::fsync_dir_with_owner(&entry_path, namespace_owner.clone())
.await
.map_err(to_file_error)?;
}
@@ -11408,36 +11341,53 @@ impl LocalDisk {
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<(bool, bool)> {
let object_path = self.io_get_object_path(volume, object)?;
let mut entries = match fs::read_dir(object_path).await {
Ok(entries) => entries,
Err(err) if err.kind() == ErrorKind::NotFound => return Ok((false, true)),
Err(err) => return Err(to_file_error(err).into()),
let prepared_path = if preserve_prefix_children {
Some(self.io_get_object_path(volume, &format!("{object}/{rollback_dir}/{}", delete_recovery::PREPARED))?)
} else {
None
};
let recovery_owned = preserve_prefix_children
&& fs::try_exists(
self.io_get_object_path(volume, &format!("{object}/{rollback_dir}/{}", delete_recovery::PREPARED))?,
)
let matching_data_dirs = self
.blocking_queue
.run_cancellable(move |is_cancelled| {
let entries = match std::fs::read_dir(&object_path) {
Ok(entries) => entries,
Err(err) if err.kind() == ErrorKind::NotFound => return Ok(Vec::new()),
Err(err) => return Err(err),
};
let recovery_owned = match prepared_path {
Some(ref path) => std::fs::exists(path)?,
None => false,
};
let prefix = if recovery_owned {
RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX
} else {
DELETE_DATA_DIR_MARKER_PREFIX
};
let marker_name = format!("{prefix}{rollback_dir}");
let mut matched = Vec::new();
for (idx, entry) in entries.enumerate() {
if idx & 63 == 0 && is_cancelled() {
return Err(std::io::Error::other("blocking namespace operation cancelled"));
}
let entry = entry?;
let Some(data_dir) = entry.file_name().to_str().and_then(|data_dir| Uuid::parse_str(data_dir).ok()) else {
continue;
};
match std::fs::metadata(entry.path().join(&marker_name)) {
Ok(metadata) if metadata.is_file() => matched.push(data_dir),
Ok(_) => continue,
Err(err) if err.kind() == ErrorKind::NotFound => continue,
Err(err) => return Err(err),
}
}
Ok(matched)
})
.await
.map_err(to_file_error)?;
let prefix = if recovery_owned {
RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX
} else {
DELETE_DATA_DIR_MARKER_PREFIX
};
let marker_name = format!("{prefix}{rollback_dir}");
let mut first_err = None;
let mut found = false;
let found = !matching_data_dirs.is_empty();
let mut deferred = false;
while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? {
let Some(data_dir) = entry.file_name().to_str().and_then(|data_dir| Uuid::parse_str(data_dir).ok()) else {
continue;
};
match fs::metadata(entry.path().join(&marker_name)).await {
Ok(metadata) if metadata.is_file() => found = true,
Ok(_) => continue,
Err(err) if err.kind() == ErrorKind::NotFound => continue,
Err(err) => return Err(to_file_error(err).into()),
}
for data_dir in matching_data_dirs {
match self
.delete_data_dir_with_namespace_owner(
volume,
@@ -26497,41 +26447,93 @@ mod test {
}
}
#[tokio::test]
async fn durable_owned_write_leaves_metadata_reads_on_foreground_lane() {
let dir = tempfile::tempdir().expect("disk root");
let endpoint = Endpoint::try_from(dir.path().to_string_lossy().as_ref()).expect("endpoint");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk");
let metadata = dir.path().join("existing-metadata");
fs::write(&metadata, b"existing").await.expect("metadata fixture");
for sync in [SyncMode::None, SyncMode::FileOnly, SyncMode::FileAndDir] {
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
let queue = Arc::clone(&disk.blocking_queue);
let blocker = tokio::spawn(async move {
queue
.run_fsync_cancellable(move |_| {
let _ = started_tx.send(());
let _ = release_rx.blocking_recv();
Ok(())
})
.await
});
started_rx.await.expect("fsync lane blocked");
let path = dir.path().join("new-metadata");
let mut write =
std::pin::pin!(disk.write_all_internal(&path, InternalBuf::Owned(Bytes::from_static(b"new")), sync, dir.path(),));
assert!(futures::FutureExt::now_or_never(&mut write).is_none());
let read = tokio::time::timeout(
Duration::from_secs(5),
disk.run_metadata_read(None, metadata.clone(), |opened| {
let mut file = opened?;
let mut bytes = Vec::new();
std::io::Read::read_to_end(&mut file, &mut bytes)?;
Ok::<_, std::io::Error>(bytes)
}),
)
.await;
let published_before_sync = path.exists();
release_tx.send(()).expect("release fsync lane");
blocker.await.expect("blocker joined").expect("blocker completed");
write.await.expect("write completed");
assert_eq!(
read.expect("metadata read must progress")
.expect("queued read")
.expect("read bytes"),
b"existing"
);
assert_eq!(
published_before_sync,
sync == SyncMode::None,
"durable writes wait on fsync, relaxed writes stay on foreground"
);
assert_eq!(fs::read(&path).await.expect("new metadata"), b"new");
fs::remove_file(&path).await.expect("remove fixture");
}
}
#[tokio::test]
async fn metadata_read_job_retains_resource_unit_until_native_completion() {
use crate::disk::io_admission::{test_resource_units, with_optional_resource_unit};
for physical_count in [1, 2] {
let dir = tempfile::tempdir().expect("metadata root");
let queue = MetadataReadQueue::new(os::PublicationRoot::new(dir.path()).expect("publication root"));
let endpoint = Endpoint::try_from(dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
let path = dir.path().join("metadata");
fs::write(&path, b"metadata").await.expect("write metadata fixture");
let (unit, gates) = test_resource_units(physical_count);
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
with_optional_resource_unit(Some(unit), async {
queue.enqueue(MetadataReadJob {
resource_unit: None,
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| false),
run: Box::new(move || {
started_tx.send(()).expect("native metadata worker started");
release_rx.blocking_recv().expect("release native metadata work");
}),
},
});
let mut read = std::pin::pin!(disk.run_metadata_read(None, path, move |opened| {
let _file = opened.expect("metadata file opened");
started_tx.send(()).expect("native metadata worker started");
release_rx.blocking_recv().expect("release native metadata work");
}));
assert!(futures::FutureExt::now_or_never(&mut read).is_none());
started_rx.await.expect("worker owns queued job");
})
.await;
let held: Vec<_> = gates.iter().map(|gate| gate.snapshot().active).collect();
release_tx.send(()).expect("release native job after caller scope ends");
tokio::time::timeout(Duration::from_secs(5), async {
while gates.iter().any(|gate| gate.snapshot().active != 0) || queue.state.lock().active_workers != 0 {
while gates.iter().any(|gate| gate.snapshot().active != 0) {
tokio::task::yield_now().await;
}
})
.await
.expect("native metadata job and worker retire");
.expect("native metadata job returns its resource unit");
assert_eq!(
held,
vec![1; physical_count],
@@ -26544,10 +26546,9 @@ mod test {
#[tokio::test]
async fn metadata_read_queue_bounds_concurrency_skips_cancelled_and_recovers_from_panic() {
let dir = tempfile::tempdir().expect("tempdir should be created");
let root = os::PublicationRoot::new(dir.path()).expect("publication root should be created");
let queue = MetadataReadQueue::new(root);
let endpoint = Endpoint::try_from(dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
// 1. Concurrency bound + cancelled job skip:
let active = Arc::new(AtomicUsize::new(0));
let max_active = Arc::new(AtomicUsize::new(0));
let executed_cancelled = Arc::new(AtomicBool::new(false));
@@ -26555,104 +26556,76 @@ mod test {
let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
let release_rx = Arc::new(ParkingLotMutex::new(release_rx));
for _ in 0..MAX_METADATA_READ_WORKERS_PER_DISK {
let active = Arc::clone(&active);
let max_active = Arc::clone(&max_active);
let entered_tx = entered_tx.clone();
let release_rx = Arc::clone(&release_rx);
queue.enqueue(MetadataReadJob {
resource_unit: None,
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| false),
run: Box::new(move || {
let cur = active.fetch_add(1, Ordering::AcqRel) + 1;
max_active.fetch_max(cur, Ordering::AcqRel);
let active_worker = Arc::clone(&active);
let max_active_worker = Arc::clone(&max_active);
let first_job = tokio::spawn({
let queue = Arc::clone(&disk.blocking_queue);
async move {
queue
.run_cancellable(move |_| {
let cur = active_worker.fetch_add(1, Ordering::AcqRel) + 1;
max_active_worker.fetch_max(cur, Ordering::AcqRel);
let _ = entered_tx.send(());
let _ = release_rx.lock().recv();
active.fetch_sub(1, Ordering::AcqRel);
}),
},
});
}
for _ in 0..MAX_METADATA_READ_WORKERS_PER_DISK {
tokio::time::timeout(Duration::from_secs(5), entered_rx.recv())
.await
.expect("initial workers should enter")
.expect("channel should stay open");
}
assert_eq!(queue.state.lock().active_workers, MAX_METADATA_READ_WORKERS_PER_DISK);
// Enqueue a cancelled production ReadMetadataOnly job and a cancelled TestCustom job while workers are busy.
let (cancelled_tx, cancelled_rx) = tokio::sync::oneshot::channel();
drop(cancelled_rx);
queue.enqueue(MetadataReadJob {
resource_unit: None,
path: dir.path().join("cancelled-never-opened"),
io_permit: None,
op: MetadataReadOp::ReadMetadataOnly { tx: cancelled_tx },
active_worker.fetch_sub(1, Ordering::AcqRel);
Ok(())
})
.await
}
});
let executed_cancelled_clone = Arc::clone(&executed_cancelled);
queue.enqueue(MetadataReadJob {
resource_unit: None,
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| true),
run: Box::new(move || {
tokio::time::timeout(Duration::from_secs(5), entered_rx.recv())
.await
.expect("initial worker should enter")
.expect("channel should stay open");
// Poll both cancelled futures once inside an inner scope while the worker is
// blocked so they are enqueued and their receivers are dropped before release.
let saw_cancelled = Arc::new(AtomicBool::new(false));
{
let mut cancelled_read =
std::pin::pin!(disk.run_metadata_read(None, dir.path().join("cancelled-never-opened"), read_metadata_only_std,));
assert!(futures::FutureExt::now_or_never(&mut cancelled_read).is_none());
let saw_cancelled_clone = Arc::clone(&saw_cancelled);
let executed_cancelled_clone = Arc::clone(&executed_cancelled);
let mut cancelled_custom = std::pin::pin!(disk.blocking_queue.run_cancellable(move |is_cancelled| {
if is_cancelled() {
saw_cancelled_clone.store(true, Ordering::Release);
} else {
executed_cancelled_clone.store(true, Ordering::Release);
}),
},
});
}
Ok(())
}));
assert!(futures::FutureExt::now_or_never(&mut cancelled_custom).is_none());
}
let mut panic_job = std::pin::pin!(
disk.blocking_queue
.run_cancellable::<()>(|_| panic!("synthetic worker panic"))
);
assert!(futures::FutureExt::now_or_never(&mut panic_job).is_none());
// Enqueue a panicking job followed by a normal completion job to verify panic respawn.
queue.enqueue(MetadataReadJob {
resource_unit: None,
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| false),
run: Box::new(|| panic!("synthetic worker panic")),
},
});
let (done_tx, done_rx) = tokio::sync::oneshot::channel();
let active_trailing = Arc::clone(&active);
let max_active_trailing = Arc::clone(&max_active);
queue.enqueue(MetadataReadJob {
resource_unit: None,
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| false),
run: Box::new(move || {
let cur = active_trailing.fetch_add(1, Ordering::AcqRel) + 1;
max_active_trailing.fetch_max(cur, Ordering::AcqRel);
active_trailing.fetch_sub(1, Ordering::AcqRel);
let _ = done_tx.send(());
}),
},
});
let mut trailing_job = std::pin::pin!(disk.blocking_queue.run_cancellable(move |_| {
let cur = active_trailing.fetch_add(1, Ordering::AcqRel) + 1;
max_active_trailing.fetch_max(cur, Ordering::AcqRel);
active_trailing.fetch_sub(1, Ordering::AcqRel);
Ok(())
}));
assert!(futures::FutureExt::now_or_never(&mut trailing_job).is_none());
assert_eq!(
queue.state.lock().active_workers,
MAX_METADATA_READ_WORKERS_PER_DISK,
"extra queued jobs must not exceed per-disk worker cap"
);
// Release only one blocked worker so the panicking job must trigger a respawn
// to execute the trailing completion job while the other worker stays blocked.
release_tx.send(()).expect("release one blocked worker");
tokio::time::timeout(Duration::from_secs(5), done_rx)
release_tx.send(()).expect("release blocked worker");
first_job.await.expect("first job join").expect("first job ok");
assert!(panic_job.await.is_err());
tokio::time::timeout(Duration::from_secs(5), trailing_job)
.await
.expect("trailing job should finish after worker panic recovery")
.expect("trailing job sender should complete");
for _ in 1..MAX_METADATA_READ_WORKERS_PER_DISK {
release_tx.send(()).expect("release remaining blocked worker");
}
.expect("trailing job ok");
assert!(saw_cancelled.load(Ordering::Acquire), "queued job must observe cancellation");
assert!(!executed_cancelled.load(Ordering::Acquire), "cancelled job must be skipped");
assert!(max_active.load(Ordering::Acquire) <= MAX_METADATA_READ_WORKERS_PER_DISK);
assert_eq!(max_active.load(Ordering::Acquire), 1);
}
#[tokio::test]
crates/ecstore/src/disk/local/commit.rs
+49 -27
View File
@@ -524,8 +524,15 @@ impl LocalDisk {
return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
}
// xl.meta path
let src_file_path = self.io_get_object_path(src_volume, format!("{}/{}", src_path, STORAGE_FORMAT_FILE).as_str())?;
// Keep recovery authority separate from active upload metadata so a
// committed upload cannot reopen when set-level cleanup fails.
let preserve_upload_metadata = src_volume == super::super::RUSTFS_META_MULTIPART_BUCKET;
let source_metadata_name = if preserve_upload_metadata {
super::super::MULTIPART_COMMIT_METADATA_FILE
} else {
STORAGE_FORMAT_FILE
};
let src_file_path = self.io_get_object_path(src_volume, format!("{src_path}/{source_metadata_name}").as_str())?;
let dst_file_path = self.io_get_object_open_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?;
// data_dir path (only used by the non-inline commit branch)
@@ -590,6 +597,39 @@ impl LocalDisk {
data_paths: has_data_dir_path.as_ref(),
};
if preserve_upload_metadata {
let source_object_path = self.get_object_path(src_volume, src_path)?;
let authority_owner: Arc<dyn Send + Sync> = if source_object_path == root_destination_object_path {
commit.mutation_lease.clone()
} else {
os::acquire_metadata_mutation_lease(&source_object_path, Some(commit.mutation_lease.clone())).await
};
let authority = self.io_get_object_path(src_volume, &format!("{src_path}/{STORAGE_FORMAT_FILE}"))?;
let recovery = self.io_get_object_path(
src_volume,
&format!("{src_path}/{}/{STORAGE_FORMAT_FILE}", super::super::MULTIPART_RECOVERY_DIR),
)?;
check_path_length(recovery.to_string_lossy().as_ref())?;
os::rename_all_with_owner(
&authority,
&recovery,
&src_volume_dir,
commit.publication_root,
Some(authority_owner.clone()),
)
.await?;
// This is system metadata authority, so use its existing strict
// durability policy even when the object payload is relaxed.
if effective_durability(src_volume).syncs_commit_metadata() {
os::fsync_dir_with_owner(recovery.parent().ok_or(DiskError::InvalidPath)?, Some(authority_owner.clone()))
.await
.map_err(to_file_error)?;
os::fsync_dir_with_owner(src_file_parent, Some(authority_owner.clone()))
.await
.map_err(to_file_error)?;
}
}
if no_inline {
// Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta
let mut xlmeta = FileMeta::new();
@@ -977,14 +1017,8 @@ impl LocalDisk {
#[cfg(windows)]
drop(commit.directory_guard);
if let Some(src_file_path_parent) = src_file_path.parent() {
if src_volume != super::super::RUSTFS_META_MULTIPART_BUCKET {
let _ = std::fs::remove_dir(src_file_path_parent);
} else {
let _ = self
.delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false)
.await;
}
if !preserve_upload_metadata && let Some(parent) = src_file_path.parent() {
let _ = std::fs::remove_dir(parent);
}
// Heal reuses a version's `data_dir` and lands the rebuilt shard on
@@ -1017,18 +1051,12 @@ impl LocalDisk {
// allowing a retry to reuse the deterministic staging path too early.
let src = src_file_path.clone();
let dst = dst_file_path.clone();
let cleanup_path = if src_volume == super::super::RUSTFS_META_MULTIPART_BUCKET {
src_file_path.parent().map(|p| p.to_path_buf())
} else {
None
};
let dst_path_for_failpoint = dst_path.to_string();
let dst_volume_for_source = dst_volume.to_string();
let dst_path_for_source = dst_path.to_string();
let dst_volume_dir_for_commit = dst_volume_dir.clone();
let publication_root_for_commit = self.publication_root.clone();
#[cfg(not(windows))]
let cleanup_src_parent = cleanup_path.is_none();
let cleanup_src_parent = !preserve_upload_metadata;
#[cfg(windows)]
let source_parent = src_file_parent.to_path_buf();
let rename_commit_guard_for_preparation = commit.directory_guard.clone();
@@ -1228,13 +1256,9 @@ impl LocalDisk {
drop(file_sync_admission.take());
#[cfg(windows)]
drop(commit.directory_guard);
if let Some(ref cleanup) = cleanup_path {
let _ = self.delete_file(&dst_volume_dir, cleanup, true, false).await;
} else {
#[cfg(windows)]
if let Some(parent) = src_file_path.parent() {
let _ = std::fs::remove_dir(parent);
}
#[cfg(windows)]
if cleanup_src_parent && let Some(parent) = src_file_path.parent() {
let _ = std::fs::remove_dir(parent);
}
for part_path in &invalidate_part_paths {
self.io_backend.invalidate_cached_fd(dst_volume, part_path).await;
@@ -1452,9 +1476,7 @@ impl LocalDisk {
}
// Cleanup
if let Some(ref cleanup) = cleanup_path {
let _ = self.delete_file(&dst_volume_dir, cleanup, true, false).await;
} else if let Some(parent) = src_file_path.parent() {
if cleanup_src_parent && let Some(parent) = src_file_path.parent() {
let _ = std::fs::remove_dir(parent);
}
crates/ecstore/src/disk/local/delete_recovery.rs
+36 -44
View File
@@ -1219,7 +1219,7 @@ impl LocalDisk {
let journal = journal_bytes(journal)?;
self.wait_for_startup_cleanup().await;
let temporary = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
let root = self.io_get_object_path(volume, object)?;
let (bucket, root) = self.io_get_bucket_and_object_path(volume, object)?;
let destination = root.join(transaction.to_string());
let original = original.map(<[u8]>::to_vec);
let sync = effective_durability(volume).syncs_commit_metadata();
@@ -1227,7 +1227,7 @@ impl LocalDisk {
let record_destination = destination.clone();
let staging_root = temporary.clone();
let lease = os::acquire_namespace_mutation_lease_with_owner(&destination, owner.clone()).await;
let (staged, target) = os::run_blocking_namespace_operation(lease, move || {
let (staged, target) = os::run_blocking_namespace_operation_with_sync(lease, sync, move || {
std::fs::create_dir_all(&staging_root)?;
let staged = Arc::new(
tempfile::Builder::new()
@@ -1280,25 +1280,27 @@ impl LocalDisk {
// The directory owner accompanies the blocking rename, so cancelling
// its waiter cannot delete staged files while activation is running.
let publication_owner: Option<Arc<dyn Send + Sync>> = Some(Arc::new((owner.clone(), staged.clone())));
os::rename_all_with_owner(
staged.path(),
&destination,
&self.io_get_bucket_path(volume)?,
&self.publication_root,
publication_owner,
)
.await?;
os::rename_all_with_owner(staged.path(), &destination, &bucket, &self.publication_root, publication_owner).await?;
if sync {
let bucket = self.io_get_bucket_path(volume)?;
let mut dirs_to_sync = Vec::new();
let mut path = root.as_path();
loop {
os::fsync_dir_with_owner(path, owner.clone()).await.map_err(to_file_error)?;
dirs_to_sync.push(path.to_path_buf());
if path == bucket {
break;
}
path = path.parent().ok_or(DiskError::InvalidPath)?;
}
os::fsync_dir_with_owner(&temporary, owner.clone())
dirs_to_sync.push(temporary);
let sync_owner = owner.clone();
self.blocking_queue
.run_fsync_cancellable(move |_| {
let _owner = sync_owner;
for dir in dirs_to_sync {
os::fsync_dir_std(&dir)?;
}
Ok(())
})
.await
.map_err(to_file_error)?;
}
@@ -1733,18 +1735,15 @@ impl LocalDisk {
}
self.commit_reserved_version_delete_with_namespace_owner(volume, object, transaction, owner.clone())
.await?;
for name in [STORAGE_FORMAT_FILE_BACKUP, DELETE_MARKER_ROLLBACK_FILE] {
match os::remove_file_with_owner(rollback.join(name), owner.clone()).await {
Ok(()) => {}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(to_file_error(err).into()),
}
}
if effective_durability(volume).syncs_commit_metadata() {
os::fsync_dir_with_owner(&rollback, owner.clone())
.await
.map_err(to_file_error)?;
}
os::remove_files_and_fsync_with_owner(
&rollback,
&[STORAGE_FORMAT_FILE_BACKUP, DELETE_MARKER_ROLLBACK_FILE],
effective_durability(volume).syncs_commit_metadata(),
None,
owner.clone(),
)
.await
.map_err(to_file_error)?;
let (_, complete) = self
.finish_version_delete(volume, object, transaction, true, owner.clone())
.await?;
@@ -1798,18 +1797,17 @@ impl LocalDisk {
Err(err) => return Err(to_file_error(err).into()),
}
}
write_delete_rollback_file(self.io_root(), volume, &root, transaction, FINISHED, &[], owner.clone()).await?;
let receipt = CommitReceipt {
journal: journal.clone(),
finished: true,
};
write_delete_rollback_file(
let receipt_bytes = serde_json::to_vec(&receipt)?;
write_delete_rollback_files(
self.io_root(),
volume,
&root,
transaction,
DELETE_QUORUM_COMMITTED_FILE,
&serde_json::to_vec(&receipt)?,
&[(FINISHED, &[]), (DELETE_QUORUM_COMMITTED_FILE, &receipt_bytes)],
owner.clone(),
)
.await?;
@@ -1818,21 +1816,15 @@ impl LocalDisk {
if observed.state != DeleteRecoveryState::CommitFinished {
return Err(DiskError::DiskOngoingReq);
}
for name in [TARGET, PREPARED, FINISHED] {
match os::remove_file_with_owner(rollback.join(name), owner.clone()).await {
Ok(()) => {}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(to_file_error(err).into()),
}
}
if effective_durability(volume).syncs_commit_metadata() {
os::fsync_dir_with_owner(&rollback, owner.clone())
.await
.map_err(to_file_error)?;
}
os::remove_file_with_owner(rollback.join(DELETE_QUORUM_COMMITTED_FILE), owner.clone())
.await
.map_err(to_file_error)?;
os::remove_files_and_fsync_with_owner(
&rollback,
&[TARGET, PREPARED, FINISHED],
effective_durability(volume).syncs_commit_metadata(),
Some(DELETE_QUORUM_COMMITTED_FILE),
owner.clone(),
)
.await
.map_err(to_file_error)?;
self.delete_unleased_with_namespace_owner(
volume,
&format!("{object}/{transaction}"),
crates/ecstore/src/disk/mod.rs
+3
View File
@@ -52,6 +52,9 @@ pub const FORMAT_CONFIG_FILE: &str = "format.json";
pub const HEALING_MARKER_PATH: &str = "healing.bin";
pub const STORAGE_FORMAT_FILE: &str = "xl.meta";
pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp";
/// Original upload authority after it stops admitting active part mutations.
pub(crate) const MULTIPART_RECOVERY_DIR: &str = ".abort-authority";
pub(crate) const MULTIPART_COMMIT_METADATA_FILE: &str = "xl.meta.complete";
pub const PART_TRANSACTION_NEW_META: &str = "new.meta";
pub const PART_TRANSACTION_OLD_META: &str = "old.meta";
pub const PART_TRANSACTION_ROLLBACK: &str = "rollback";
crates/ecstore/src/disk/os.rs
+681 -102
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/services/rebalance/entry.rs
+135
View File
@@ -1782,6 +1782,141 @@ mod tests {
fixture.assert_moved().await;
}
/// An `ExpiredObjectAllVersions` expiry is a prefix delete that takes the
/// bucket lifecycle write lock, while the entry holds that lock for read
/// through its bucket incarnation fence. Applied inline, it waited out the
/// lock timeout, stalled every lifecycle read in the bucket behind it, and
/// failed the entry. The version moves instead and lifecycle expires it on
/// the target (rustfs/backlog#2873).
#[test]
#[serial_test::serial]
fn real_rebalance_moves_a_version_expired_by_a_delete_all_rule() {
run_large_stack_async_test("rebalance-delete-all-expiry", run_delete_all_expiry_case);
}
async fn run_delete_all_expiry_case() {
const REBALANCE_ID: &str = "rebalance-delete-all-expiry";
let (_temp_dirs, store, _peer) = crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(None).await;
let bucket = "rebalance-delete-all";
let object = "expired.bin";
// An active run blocks the data-usage publication bucket creation needs.
Box::pin(store.make_bucket(
bucket,
&MakeBucketOptions {
versioning_enabled: true,
..Default::default()
},
))
.await
.expect("create bucket");
let incarnation = store.bucket_incarnation_id(bucket).await.expect("bucket incarnation");
let rebalance_meta = active_rebalance_meta(REBALANCE_ID);
rebalance_meta
.save(Arc::clone(&store.pools[0]))
.await
.expect("persist the active rebalance run");
*store.rebalance_meta.write().await = Some(rebalance_meta);
let source_set = store.pools[0].get_disks_by_key(object);
let payload = b"expired by an all-versions rule".to_vec();
let written = Box::pin(source_set.put_object(
bucket,
object,
&mut PutObjReader::from_vec(payload.clone()),
&ObjectOptions {
versioned: true,
mod_time: Some(OffsetDateTime::now_utc() - time::Duration::days(10)),
expected_bucket_incarnation_id: Some(incarnation),
..Default::default()
},
))
.await
.expect("seed the expired source version");
let lifecycle = BucketLifecycleConfiguration {
expiry_updated_at: None,
rules: vec![LifecycleRule {
status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
expiration: Some(LifecycleExpiration {
days: Some(1),
expired_object_all_versions: Some(true),
..Default::default()
}),
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
filter: None,
id: Some("delete-all".to_string()),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
};
let entry = metacache_entry_from_source(&source_set, bucket, object).await;
let lock_timeout = crate::set_disk::get_lock_acquire_timeout();
let started = std::time::Instant::now();
let outcome = tokio::time::timeout(
lock_timeout,
Box::pin(Arc::clone(&store).rebalance_entry(
RebalanceEntryTarget {
bucket: bucket.to_string(),
pool_index: 0,
},
entry,
Arc::clone(&source_set),
Arc::new(RebalanceBucketConfigs {
bucket_incarnation_id: Some(incarnation),
lifecycle_config: Some(lifecycle),
..Default::default()
}),
Arc::from(REBALANCE_ID),
CancellationToken::new(),
)),
)
.await
.expect("the entry must not wait out the bucket lifecycle lock timeout")
.expect("rebalance entry");
let elapsed = started.elapsed();
assert_eq!(outcome, RebalanceEntryOutcome::Completed);
assert!(
elapsed < lock_timeout / 2,
"the entry must finish well under the {lock_timeout:?} lock timeout, took {elapsed:?}"
);
assert!(
source_set
.load_file_info_versions_exact(bucket, object)
.await
.expect("read source versions")
.is_none(),
"the source entry must be removed once the version moved"
);
let target = store.pools[1]
.get_disks_by_key(object)
.load_file_info_versions_exact(bucket, object)
.await
.expect("read target versions")
.expect("the version must reach the target pool");
assert_eq!(target.versions.len(), 1, "{:?}", target.versions);
assert_eq!(target.versions[0].version_id, written.version_id);
let mut reader = store.pools[1]
.get_object_reader(
bucket,
object,
None,
HeaderMap::new(),
&ObjectOptions {
versioned: true,
version_id: written.version_id.map(|version| version.to_string()),
..Default::default()
},
)
.await
.expect("the moved version should be readable");
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("drain the moved version");
assert_eq!(body, payload);
}
/// Overwriting after versioning was enabled leaves a newer version above
/// the null version. The old `None` read selected that newer version, the
/// target accepted it as an equivalent copy, and cleanup then deleted the
crates/ecstore/src/set_disk/mod.rs
+3 -1
View File
@@ -12755,7 +12755,7 @@ mod tests {
);
}
#[tokio::test(start_paused = true)]
#[tokio::test]
async fn test_get_disks_info_probes_drives_concurrently() {
use crate::disk::disk_store::DISK_INFO_PROBE_DELAY_FOR_TEST;
@@ -12770,6 +12770,8 @@ mod tests {
disks.push(Some(disk));
}
// Complete real filesystem setup before measuring virtual probe delays.
tokio::time::pause();
let probe_delay = std::time::Duration::from_secs(2);
let started = tokio::time::Instant::now();
let info = DISK_INFO_PROBE_DELAY_FOR_TEST
crates/ecstore/src/set_disk/ops/multipart.rs
+813 -86
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/store/heal.rs
+4 -4
View File
@@ -602,10 +602,10 @@ impl ECStore {
/// Snapshot only the object sets participating in this request. The sampler
/// resolves their physical leaves before Heal enters namespace/cache locks.
pub fn physical_heal_plan(&self, object: &str, opts: &HealOpts) -> Result<Option<crate::disk::PhysicalResourcePlan>> {
if let Some(index) = opts.pool {
if index >= self.pools.len() {
return Err(invalid_heal_pool_index(index, self.pools.len()));
}
if let Some(index) = opts.pool
&& index >= self.pools.len()
{
return Err(invalid_heal_pool_index(index, self.pools.len()));
}
let object = rustfs_utils::path::encode_dir_object_ref(object);
let mut selected: Option<crate::disk::PhysicalResourcePlan> = None;
crates/heal/Cargo.toml
+1
View File
@@ -100,6 +100,7 @@ crc-fast = { workspace = true }
sha2 = { workspace = true }
[dev-dependencies]
rustfs-ecstore = { workspace = true, features = ["test-util"] }
rustfs-filemeta = { workspace = true }
serde_json = { workspace = true, features = ["raw_value"] }
rustfs-test-utils = { workspace = true }
crates/heal/src/heal/erasure_healer/resume_loop_tests/admin_outcome.rs
+10 -3
View File
@@ -318,7 +318,9 @@ async fn admin_erasure_recreated_bucket_cannot_rebind_checkpoint() {
assert_eq!(restarted.get_outcome().await.counters.healed, 1);
}
#[tokio::test(start_paused = true)]
// Durable checkpoints use an independent fsync runtime, so this path needs real
// time. The in-memory admin_object tests cover retry timing with a paused clock.
#[tokio::test]
async fn admin_erasure_dispositions_do_not_invent_success_and_retries_count_once() {
let (_temp, storage) = fixture().await;
objects(&storage, "a", &["healthy", "absent", "unproven", "gone", "offline"]);
@@ -330,12 +332,17 @@ async fn admin_erasure_dispositions_do_not_invent_success_and_retries_count_once
storage.set_outcome("gone", Some("null"), HealOutcome::VersionNotFound);
storage.set_outcome("offline", Some("null"), HealOutcome::Transient);
let task = HealTask::from_request(request(), storage.clone());
task.execute()
let error = task
.execute()
.await
.expect_err("unhealed objects must retain the failed task status");
let outcome = task.get_outcome().await;
let c = &outcome.counters;
assert_eq!((c.processed, c.healed, c.unchanged, c.failed, c.skipped, c.unknown), (5, 0, 2, 0, 3, 1));
assert_eq!(
(c.processed, c.healed, c.unchanged, c.failed, c.skipped, c.unknown),
(5, 0, 2, 0, 3, 1),
"unexpected task error: {error:?}"
);
assert_eq!(c.attempt_failures, 8);
assert_eq!(storage.calls().iter().filter(|(name, _)| name == "gone").count(), 4);
assert_eq!(storage.calls().iter().filter(|(name, _)| name == "offline").count(), 4);
crates/heal/src/heal/manager/tests.rs
+17
View File
@@ -6096,6 +6096,7 @@ fn admin_bucket_discovery_retries_keep_the_existing_budget() {
async fn check_node_admission_task_retry(source: HealRequestSource) {
use crate::heal::outcome::HealFailureClass;
use crate::heal::storage::ECStoreHealStorage;
use crate::heal::storage_api::owner::{PhysicalDiskSample, PhysicalDiskSampler};
use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
use rustfs_heal_contracts::heal_channel::HealScanMode;
use std::num::NonZeroU16;
@@ -6108,6 +6109,18 @@ async fn check_node_admission_task_retry(source: HealRequestSource) {
env.make_bucket("admission-retry", false).await;
env.put_object_bytes("admission-retry", "existing", b"repair me".to_vec())
.await;
let mut physical = PhysicalDiskSampler::default();
physical.publish(
PhysicalDiskSample::collect(&env.ecstore).with_test_single_device(),
ResourcePressure::Unknown,
);
assert!(
env.ecstore
.physical_heal_plan("existing", &Default::default())
.expect("fixture placement")
.is_some(),
"physical discovery must be ready before testing node capacity"
);
let gate = NodeAdmission::new(NonZeroU16::new(1).expect("one node unit"));
let storage: Arc<dyn HealStorageAPI> =
Arc::new(ECStoreHealStorage::new(env.ecstore.clone()).with_admission(Some(gate.clone())));
@@ -6148,6 +6161,10 @@ async fn check_node_admission_task_retry(source: HealRequestSource) {
completed_units: None,
}),
);
physical.publish(
PhysicalDiskSample::collect(&env.ecstore).with_test_single_device(),
ResourcePressure::Unknown,
);
let retried = HealTask::from_request(retry, storage);
retried
.execute()
crates/heal/src/heal/manager/tests/auto_scan.rs
+10 -3
View File
@@ -235,6 +235,9 @@ impl ScannerFixture {
}
async fn queued_request(&self) -> HealRequest {
// Let the immediate interval tick cross the timer's millisecond boundary
// before the ready loop keeps virtual time fixed during filesystem I/O.
tokio::time::sleep(Duration::ZERO).await;
// Keep the paused clock from advancing while filesystem workers
// finish. A wall-clock deadline also bounds a broken admission path.
let deadline = std::time::Instant::now() + Duration::from_secs(5);
@@ -291,10 +294,12 @@ async fn finish_without_advancing_clock<T>(future: impl Future<Output = T>) -> T
}
}
#[tokio::test(start_paused = true)]
#[tokio::test]
#[serial_test::serial(replacement_auto_scan)]
async fn scanner_tick_recovers_remote_survivor_after_all_local_targets_are_formatted() {
let fixture = ScannerFixture::new(false).await;
// Durable fixture writes must finish before virtual scanner time begins.
tokio::time::pause();
fixture
.manager
.start_auto_disk_scanner()
@@ -303,11 +308,12 @@ async fn scanner_tick_recovers_remote_survivor_after_all_local_targets_are_forma
fixture.assert_original_owner_is_queued().await;
}
#[tokio::test(start_paused = true)]
#[tokio::test]
#[serial_test::serial(replacement_auto_scan)]
async fn scanner_tick_finishes_existing_owner_before_admitting_new_unformatted_targets() {
temp_env::async_with_vars([(rustfs_config::ENV_UNSAFE_BYPASS_DISK_CHECK, Some("true"))], async {
let fixture = ScannerFixture::new(true).await;
tokio::time::pause();
fixture
.manager
.start_auto_disk_scanner()
@@ -381,7 +387,7 @@ async fn recovery_disk_index_keeps_first_snapshot_and_sorted_unique_endpoints()
);
}
#[tokio::test(start_paused = true)]
#[tokio::test]
#[serial_test::serial(replacement_auto_scan)]
async fn scanner_tick_blocks_only_the_conflicting_generation_of_a_set() {
// backlog#2902 step 3: two survivors hold the same revision of a sibling
@@ -427,6 +433,7 @@ async fn scanner_tick_blocks_only_the_conflicting_generation_of_a_set() {
..HealConfig::default()
}),
);
tokio::time::pause();
fixture
.manager
.start_auto_disk_scanner()
crates/heal/src/heal/mrf_queue.rs
+59 -14
View File
@@ -212,6 +212,7 @@ pub(crate) const MRF_SCOPED_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal-scop
const MRF_LIFECYCLE_PATHS: [&str; 2] = [".heal-mrf-lifecycle.0.bin", ".heal-mrf-lifecycle.1.bin"];
const MRF_LIFECYCLE_MAGIC: &[u8; 8] = b"RFMFLC01";
const MRF_LIFECYCLE_FORMAT_VERSION: u8 = 1;
const MRF_LIFECYCLE_FORMAT_VERSION_V2: u8 = 2;
/// Record format tag.
const MRF_JOURNAL_FORMAT: u8 = 1;
@@ -632,15 +633,21 @@ enum MrfReplayError {
}
fn encode_mrf_lifecycle_checkpoint(
format_version: u8,
owner: Uuid,
sequence: u64,
records: Vec<ResponsibilityCheckpoint>,
byte_limit: usize,
) -> Option<Vec<u8>> {
if owner.is_nil()
if !matches!(format_version, MRF_LIFECYCLE_FORMAT_VERSION | MRF_LIFECYCLE_FORMAT_VERSION_V2)
|| owner.is_nil()
|| sequence == 0
|| sequence == u64::MAX
|| records.iter().any(|record| !record.is_valid())
|| (format_version == MRF_LIFECYCLE_FORMAT_VERSION
&& records.iter().any(|record| {
matches!(record.state, partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. })
}))
|| records
.iter()
.map(|record| record.responsibility_id)
@@ -657,7 +664,7 @@ fn encode_mrf_lifecycle_checkpoint(
return None;
}
let payload = MrfLifecyclePayload {
format_version: MRF_LIFECYCLE_FORMAT_VERSION,
format_version,
checkpoint_owner: owner,
checkpoint_sequence: sequence,
records,
@@ -669,7 +676,7 @@ fn encode_mrf_lifecycle_checkpoint(
let digest: [u8; 32] = sha2::Sha256::digest(&payload).into();
let mut encoded = Vec::with_capacity(8 + 1 + 8 + 32 + payload.len());
encoded.extend_from_slice(MRF_LIFECYCLE_MAGIC);
encoded.push(MRF_LIFECYCLE_FORMAT_VERSION);
encoded.push(format_version);
encoded.extend_from_slice(&u64::try_from(payload.len()).ok()?.to_le_bytes());
encoded.extend_from_slice(&digest);
encoded.extend_from_slice(&payload);
@@ -681,7 +688,7 @@ fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Result<Mrf
if data.len() < HEADER_LEN || data.len() > byte_limit || &data[..8] != MRF_LIFECYCLE_MAGIC {
return Err(MrfReplayError::InvalidLifecycle);
}
if data[8] != MRF_LIFECYCLE_FORMAT_VERSION {
if !matches!(data[8], MRF_LIFECYCLE_FORMAT_VERSION | MRF_LIFECYCLE_FORMAT_VERSION_V2) {
return Err(snapshot::SnapshotError::Unsupported.into());
}
let payload_len = usize::try_from(u64::from_le_bytes(data[9..17].try_into().map_err(|_| MrfReplayError::InvalidLifecycle)?))
@@ -695,11 +702,15 @@ fn decode_mrf_lifecycle_checkpoint(data: &[u8], byte_limit: usize) -> Result<Mrf
return Err(MrfReplayError::InvalidLifecycle);
}
let decoded: MrfLifecyclePayload = serde_json::from_slice(payload).map_err(MrfReplayError::Lifecycle)?;
if decoded.format_version != MRF_LIFECYCLE_FORMAT_VERSION
if decoded.format_version != data[8]
|| decoded.checkpoint_owner.is_nil()
|| decoded.checkpoint_sequence == 0
|| decoded.checkpoint_sequence == u64::MAX
|| decoded.records.iter().any(|record| !record.is_valid())
|| (decoded.format_version == MRF_LIFECYCLE_FORMAT_VERSION
&& decoded.records.iter().any(|record| {
matches!(record.state, partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. })
}))
|| decoded
.records
.iter()
@@ -771,7 +782,7 @@ async fn read_mrf_lifecycle_checkpoint(
sequence: u64,
checkpoint_max_bytes: usize,
byte_limit: usize,
) -> Result<Option<Vec<ResponsibilityCheckpoint>>, MrfReplayError> {
) -> Result<Option<MrfLifecyclePayload>, MrfReplayError> {
let Some(bytes) = snapshot::read_committed_companion(
&journal_disks().await,
owner,
@@ -788,7 +799,7 @@ async fn read_mrf_lifecycle_checkpoint(
if decoded.checkpoint_owner != owner || decoded.checkpoint_sequence != sequence {
return Err(MrfReplayError::InvalidLifecycle);
}
Ok(Some(decoded.records))
Ok(Some(decoded))
}
/// Write the snapshot to every local disk; returns true when at least one
@@ -953,6 +964,9 @@ struct MrfRuntime {
config: MrfConsumerConfig,
checkpoint_owner: Uuid,
next_checkpoint_sequence: u64,
// Preserve an observed format across checkpoint ownership changes.
// Without a V2 checkpoint, keep writing V1 until writer rollout is qualified.
lifecycle_format_version: u8,
new_since_flush: usize,
/// True while the in-memory pending set has changed since the last
/// journal flush (push, pop, or an attempts bump that alters the encoded
@@ -1072,6 +1086,9 @@ impl MrfRuntime {
.partial_writes
.intent_for_responsibility(responsibility_id)
.ok_or(MrfLifecycleControlError::StaleGeneration)?;
if matches!(state, partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }) {
return Err(MrfLifecycleControlError::BucketIdentityUnresolved);
}
if state.bucket_incarnation_id() != Some(expected_bucket_incarnation_id) {
return Err(MrfLifecycleControlError::IncarnationChanged);
}
@@ -1139,7 +1156,12 @@ impl MrfRuntime {
return Err(MrfLifecycleControlError::StaleGeneration);
}
if expected_bucket_incarnation_id.is_none()
&& (source_incarnation.is_some() || !matches!(state, partial_write::ResponsibilityState::Active {}))
&& (source_incarnation.is_some()
|| !matches!(
state,
partial_write::ResponsibilityState::Active {}
| partial_write::ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }
))
{
return Err(MrfLifecycleControlError::StaleGeneration);
}
@@ -1247,6 +1269,7 @@ impl MrfRuntime {
(records.len() == self.partial_writes.depth())
.then(|| {
encode_mrf_lifecycle_checkpoint(
self.lifecycle_format_version,
self.checkpoint_owner,
self.next_checkpoint_sequence,
records,
@@ -1576,6 +1599,7 @@ pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
lifecycle_format_version: replay.lifecycle_format_version,
new_since_flush: 0,
dirty: false,
journal_on_disk: replay.journal_on_disk,
@@ -1601,6 +1625,7 @@ struct ReplayOutcome {
partial_writes: Vec<(MrfIntent, Option<ResponsibilityCheckpoint>)>,
cleanup: Option<ReplayCleanup>,
next_checkpoint_sequence: u64,
lifecycle_format_version: u8,
}
fn replay_must_retain_journal(
@@ -1623,6 +1648,7 @@ struct ReplaySource {
truncated_bytes: usize,
cleanup: ReplayCleanup,
lifecycle: HashMap<[u8; 32], VecDeque<ResponsibilityCheckpoint>>,
lifecycle_format_version: u8,
}
fn take_lifecycle_record(
@@ -1642,8 +1668,8 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
let owner = committed.owner();
let sequence = committed.sequence();
let (intents, truncated_bytes) = decode_journal(committed.payload());
let records = read_mrf_lifecycle_checkpoint(owner, sequence, max_bytes, lifecycle_max_bytes).await?;
if let Some(records) = records.as_ref() {
let checkpoint = read_mrf_lifecycle_checkpoint(owner, sequence, max_bytes, lifecycle_max_bytes).await?;
if let Some(checkpoint) = checkpoint.as_ref() {
// Validate the complete owner before rearming or submitting any
// intent. Different source generations may share an intent digest.
let mut counts = HashMap::new();
@@ -1651,7 +1677,7 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
let digest = intent_digest(intent).ok_or(MrfReplayError::InvalidLifecycle)?;
*counts.entry(digest).or_insert(0usize) += 1;
}
for record in records {
for record in &checkpoint.records {
match counts.get_mut(&record.intent_digest) {
Some(count) if *count > 0 => *count -= 1,
_ => return Err(MrfReplayError::InvalidLifecycle),
@@ -1661,7 +1687,10 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
return Err(MrfReplayError::InvalidLifecycle);
}
}
let records = records.unwrap_or_default();
let (lifecycle_format_version, records) = match checkpoint {
Some(checkpoint) => (checkpoint.format_version, checkpoint.records),
None => (MRF_LIFECYCLE_FORMAT_VERSION, Vec::new()),
};
let mut lifecycle: HashMap<[u8; 32], VecDeque<ResponsibilityCheckpoint>> = HashMap::with_capacity(records.len());
for record in records {
lifecycle.entry(record.intent_digest).or_default().push_back(record);
@@ -1671,6 +1700,7 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
truncated_bytes,
cleanup: ReplayCleanup::Committed { owner, sequence },
lifecycle,
lifecycle_format_version,
}));
}
// The scoped file is a complete authoritative legacy snapshot. Fall back
@@ -1689,6 +1719,7 @@ async fn read_replay_source(max_bytes: usize, lifecycle_max_bytes: usize) -> Res
truncated_bytes,
cleanup: ReplayCleanup::Legacy,
lifecycle: HashMap::new(),
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
}))
}
@@ -1733,6 +1764,7 @@ async fn replay_into(
partial_writes: Vec::new(),
cleanup: None,
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
};
}
Err(err) => {
@@ -1750,6 +1782,7 @@ async fn replay_into(
partial_writes: Vec::new(),
cleanup: None,
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
};
}
};
@@ -1896,6 +1929,7 @@ async fn replay_into(
partial_writes,
cleanup: journal_on_disk.then_some(cleanup),
next_checkpoint_sequence,
lifecycle_format_version: source.lifecycle_format_version,
}
}
@@ -1914,6 +1948,7 @@ async fn run_mrf_consumer(
config: config.clone(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: false,
@@ -1934,6 +1969,7 @@ async fn run_mrf_consumer(
runtime.durable_replay_anchors = replay.durable_replay_anchors;
runtime.replay_cleanup = replay.cleanup;
runtime.next_checkpoint_sequence = replay.next_checkpoint_sequence;
runtime.lifecycle_format_version = replay.lifecycle_format_version;
runtime.adopt_replayed_partial_writes(replay.partial_writes);
// Anything still pending (e.g. the manager was full and backoff armed)
// must be re-persisted by the next flush before replay can delete the
@@ -2148,6 +2184,7 @@ mod tests {
mod empty_events_tests;
mod lifecycle_replay_tests;
mod lifecycle_v2_reader_tests;
#[tokio::test(start_paused = true)]
async fn journal_disk_io_bounds_fanout_and_collects_every_result() {
@@ -2307,7 +2344,7 @@ mod tests {
},
};
let records = vec![record, accepted];
let encoded = encode_mrf_lifecycle_checkpoint(owner, 42, records.clone(), 8192)
let encoded = encode_mrf_lifecycle_checkpoint(MRF_LIFECYCLE_FORMAT_VERSION, owner, 42, records.clone(), 8192)
.expect("lifecycle checkpoint should fit the configured bound");
let decoded = decode_mrf_lifecycle_checkpoint(&encoded, 8192).expect("valid lifecycle checkpoint");
assert_eq!(decoded.checkpoint_owner, owner);
@@ -2318,7 +2355,7 @@ mod tests {
*torn.last_mut().expect("non-empty checkpoint") ^= 0xff;
assert!(decode_mrf_lifecycle_checkpoint(&torn, 8192).is_err());
assert!(decode_mrf_lifecycle_checkpoint(&encoded, encoded.len() - 1).is_err());
assert!(encode_mrf_lifecycle_checkpoint(Uuid::nil(), 42, Vec::new(), 8192).is_none());
assert!(encode_mrf_lifecycle_checkpoint(MRF_LIFECYCLE_FORMAT_VERSION, Uuid::nil(), 42, Vec::new(), 8192).is_none());
}
fn w13_timestamp() -> String {
@@ -2442,6 +2479,7 @@ mod tests {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
lifecycle_format_version: replay.lifecycle_format_version,
new_since_flush: 0,
dirty: false,
journal_on_disk: replay.journal_on_disk,
@@ -2582,6 +2620,7 @@ mod tests {
config: MrfConsumerConfig::default(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: true,
journal_on_disk: false,
@@ -3175,6 +3214,7 @@ mod tests {
config: MrfConsumerConfig::default(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: true,
@@ -3227,6 +3267,7 @@ mod tests {
config: MrfConsumerConfig::default(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: true,
@@ -3276,6 +3317,7 @@ mod tests {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 9,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: true,
@@ -3388,6 +3430,7 @@ mod tests {
.expect("the fixture bucket must have a persisted incarnation");
let lifecycle_limit = config.journal_max_bytes.saturating_mul(4);
let lifecycle = encode_mrf_lifecycle_checkpoint(
MRF_LIFECYCLE_FORMAT_VERSION,
replay_owner,
11,
vec![ResponsibilityCheckpoint {
@@ -3451,6 +3494,7 @@ mod tests {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
lifecycle_format_version: replay.lifecycle_format_version,
new_since_flush: 0,
dirty: false,
journal_on_disk: replay.journal_on_disk,
@@ -3699,6 +3743,7 @@ mod tests {
config: MrfConsumerConfig::default(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: true,
journal_on_disk: false,
crates/heal/src/heal/mrf_queue/partial_write.rs
+77 -10
View File
@@ -69,6 +69,11 @@ impl PartialWriteKey {
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub(super) enum ResponsibilityState {
Active {},
SourceAndCurrentGenerationUnknown {
since_ms: u64,
next_probe_at_ms: u64,
probe_attempts: u32,
},
HeldUnverifiedLegacy {
bucket_incarnation_id: Uuid,
since_ms: u64,
@@ -102,6 +107,7 @@ impl ResponsibilityState {
fn estimated_bytes(&self) -> usize {
match self {
Self::Active {} => 0,
Self::SourceAndCurrentGenerationUnknown { .. } => 2 * std::mem::size_of::<u64>() + std::mem::size_of::<u32>(),
Self::HeldUnverifiedLegacy { .. } => std::mem::size_of::<Uuid>() + std::mem::size_of::<u64>(),
Self::LegacyGenerationUnknown { .. } => std::mem::size_of::<Uuid>() + std::mem::size_of::<u64>(),
Self::BucketIncarnationChanged { .. } => 2 * std::mem::size_of::<Uuid>() + std::mem::size_of::<u64>(),
@@ -116,7 +122,7 @@ impl ResponsibilityState {
pub(super) fn bucket_incarnation_id(&self) -> Option<Uuid> {
match self {
Self::Active {} => None,
Self::Active {} | Self::SourceAndCurrentGenerationUnknown { .. } => None,
Self::HeldUnverifiedLegacy {
bucket_incarnation_id, ..
}
@@ -158,6 +164,7 @@ impl ResponsibilityCheckpoint {
}
match &self.state {
ResponsibilityState::Active {} => true,
ResponsibilityState::SourceAndCurrentGenerationUnknown { .. } => self.source_bucket_incarnation_id.is_none(),
ResponsibilityState::HeldUnverifiedLegacy {
bucket_incarnation_id, ..
} => !bucket_incarnation_id.is_nil() && self.source_bucket_incarnation_id == Some(*bucket_incarnation_id),
@@ -214,7 +221,11 @@ struct Responsibility {
impl Responsibility {
fn generation_unobserved(&self) -> bool {
self.source_bucket_incarnation_id.is_none() && matches!(self.state, ResponsibilityState::Active {})
self.source_bucket_incarnation_id.is_none()
&& matches!(
self.state,
ResponsibilityState::Active {} | ResponsibilityState::SourceAndCurrentGenerationUnknown { .. }
)
}
fn is_parked(&self) -> bool {
@@ -479,7 +490,8 @@ impl PartialWrites {
.map_or(*since_ms, |oldest| oldest.min(*since_ms)),
);
}
ResponsibilityState::LegacyGenerationUnknown { .. } => summary.legacy_generation_unknown_count += 1,
ResponsibilityState::LegacyGenerationUnknown { .. }
| ResponsibilityState::SourceAndCurrentGenerationUnknown { .. } => summary.legacy_generation_unknown_count += 1,
ResponsibilityState::BucketIncarnationChanged { .. } => summary.bucket_incarnation_changed_count += 1,
ResponsibilityState::OperatorAcceptedUnverified { accepted_at_ms, .. } => {
summary.operator_accepted_count += 1;
@@ -525,6 +537,17 @@ impl PartialWrites {
entry.source_bucket_incarnation_id = record.source_bucket_incarnation_id;
entry.last_operator_acceptance = record.last_operator_acceptance.clone();
entry.state = record.state.clone();
if let ResponsibilityState::SourceAndCurrentGenerationUnknown {
next_probe_at_ms,
probe_attempts,
..
} = &entry.state
{
let delay = Duration::from_millis(next_probe_at_ms.saturating_sub(unix_now_ms()))
.clamp(MRF_RETIREMENT_PROOF_RECHECK_INTERVAL, MAX_UNVERIFIED_RETRY_BACKOFF);
entry.next_attempt = Instant::now() + delay;
entry.consecutive_unverified_attempts = *probe_attempts;
}
self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry));
if entry.is_parked() {
entry.retry_queued = false;
@@ -616,6 +639,9 @@ impl PartialWrites {
return Err("a different operator disposition already exists");
}
ResponsibilityState::Active {} => return Err("responsibility is not held as unverified legacy"),
ResponsibilityState::SourceAndCurrentGenerationUnknown { .. } => {
return Err("current bucket incarnation is unresolved");
}
};
if current_incarnation != expected_bucket_incarnation_id {
return Err("bucket incarnation changed; refresh the responsibility listing");
@@ -743,6 +769,9 @@ impl PartialWrites {
let (status, bucket_incarnation_id, held_since_ms) = match &entry.state {
ResponsibilityState::Active {} if entry.generation_unobserved() => ("legacy_generation_unobserved", None, None),
ResponsibilityState::Active {} => continue,
ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms, .. } => {
("source_and_current_generation_unknown", None, Some(*since_ms))
}
ResponsibilityState::HeldUnverifiedLegacy {
bucket_incarnation_id,
since_ms,
@@ -851,9 +880,17 @@ impl PartialWrites {
entry.retry_queued = false;
} else if source_bucket_incarnation_id.is_none() {
let old_cost = Self::entry_cost(entry);
let detected_at_ms = match entry.state {
ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms, .. }
| ResponsibilityState::LegacyGenerationUnknown {
detected_at_ms: since_ms,
..
} => since_ms,
_ => unix_now_ms(),
};
entry.state = ResponsibilityState::LegacyGenerationUnknown {
observed_bucket_incarnation_id,
detected_at_ms: unix_now_ms(),
detected_at_ms,
};
self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry));
entry.retry_queued = false;
@@ -957,8 +994,27 @@ impl PartialWrites {
continue;
}
// Advance before awaiting so cancellation cannot cause a tight
// retry loop. Restart begins with the same conservative delay.
entry.defer_unverified_attempt(now, config.admission_backoff.max(MRF_RETIREMENT_PROOF_RECHECK_INTERVAL));
// retry loop. V2 checkpoints also retain this pacing on restart.
entry.defer_unverified_attempt(
now,
config
.admission_backoff
.clamp(MRF_RETIREMENT_PROOF_RECHECK_INTERVAL, MAX_UNVERIFIED_RETRY_BACKOFF),
);
let (lifecycle_changed, detected_at_ms) = match &mut entry.state {
ResponsibilityState::SourceAndCurrentGenerationUnknown {
since_ms,
next_probe_at_ms,
probe_attempts,
} => {
*next_probe_at_ms = unix_now_ms().saturating_add(
u64::try_from(entry.next_attempt.saturating_duration_since(now).as_millis()).unwrap_or(u64::MAX),
);
*probe_attempts = entry.consecutive_unverified_attempts;
(true, *since_ms)
}
_ => (false, unix_now_ms()),
};
let result = tokio::time::timeout(
MRF_IDENTITY_PROBE_TIMEOUT,
manager.mrf_bucket_incarnation_id(entry.intent.bucket.as_ref()),
@@ -970,7 +1026,7 @@ impl PartialWrites {
let old_cost = Self::entry_cost(entry);
entry.state = ResponsibilityState::LegacyGenerationUnknown {
observed_bucket_incarnation_id,
detected_at_ms: unix_now_ms(),
detected_at_ms,
};
entry.last_identity_probe_result = Some("known");
self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry));
@@ -979,7 +1035,7 @@ impl PartialWrites {
Ok(Ok(None)) => entry.last_identity_probe_result = Some("unresolved"),
Ok(Err(_)) | Err(_) => entry.last_identity_probe_result = Some("unavailable"),
}
return false;
return lifecycle_changed;
}
false
}
@@ -2127,8 +2183,14 @@ mod tests {
let records = writes.checkpoint_records(super::super::intent_digest);
assert_eq!(records.len(), 2);
assert_eq!(records[0].intent_digest, records[1].intent_digest);
let lifecycle = super::super::encode_mrf_lifecycle_checkpoint(owner, sequence, records, lifecycle_limit)
.expect("same-key generations need distinct lifecycle records");
let lifecycle = super::super::encode_mrf_lifecycle_checkpoint(
crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
owner,
sequence,
records,
lifecycle_limit,
)
.expect("same-key generations need distinct lifecycle records");
super::super::snapshot::publish_committed_snapshot_with_companion(
&disks,
owner,
@@ -2152,6 +2214,7 @@ mod tests {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
lifecycle_format_version: replay.lifecycle_format_version,
new_since_flush: 0,
dirty: false,
journal_on_disk: replay.journal_on_disk,
@@ -2288,6 +2351,7 @@ mod tests {
let lifecycle_limit = config.journal_max_bytes.saturating_mul(4);
let owner = Uuid::new_v4();
let lifecycle = super::super::encode_mrf_lifecycle_checkpoint(
crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
owner,
7,
writes.checkpoint_records(super::super::intent_digest),
@@ -2314,6 +2378,7 @@ mod tests {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
lifecycle_format_version: replay.lifecycle_format_version,
new_since_flush: 0,
dirty: false,
journal_on_disk: replay.journal_on_disk,
@@ -2424,6 +2489,7 @@ mod tests {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: true,
@@ -2501,6 +2567,7 @@ mod tests {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: true,
crates/heal/src/heal/mrf_queue/partial_write/tests/unknown_source_tests.rs
+167
View File
@@ -99,6 +99,7 @@ fn runtime() -> MrfRuntime {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: crate::heal::mrf_queue::MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: false,
@@ -368,3 +369,169 @@ async fn check_unknown_source_deleted_bucket_replay() {
}
assert_eq!(manager.get_queue_length().await, 0, "identity probes must not enqueue object healing");
}
fn restored_v2_runtime(next_probe_at_ms: u64, probe_attempts: u32) -> (MrfRuntime, PartialWriteKey) {
let mut runtime = runtime();
runtime.lifecycle_format_version = 2;
let original = intent("v2-unknown-generation");
let key = PartialWriteKey::new(&original, None);
runtime
.admit_partial_write(original.clone(), None)
.expect("admit retained owner");
let record = ResponsibilityCheckpoint {
intent_digest: super::super::super::intent_digest(&original).expect("intent identity"),
responsibility_id: Uuid::new_v4(),
source_bucket_incarnation_id: None,
last_operator_acceptance: None,
state: ResponsibilityState::SourceAndCurrentGenerationUnknown {
since_ms: 7,
next_probe_at_ms,
probe_attempts,
},
};
assert!(runtime.partial_writes.restore_state(&original, &record));
runtime.partial_writes.mark_persisted();
(runtime, key)
}
#[tokio::test(start_paused = true)]
async fn lifecycle_v2_reader_restores_bounded_probe_schedule_and_attempt_count() {
for (deadline, attempts, expected_delay) in [
(0, 0, MRF_RETIREMENT_PROOF_RECHECK_INTERVAL),
(u64::MAX, u32::MAX, MAX_UNVERIFIED_RETRY_BACKOFF),
] {
let now = Instant::now();
let (runtime, key) = restored_v2_runtime(deadline, attempts);
let entry = runtime.partial_writes.entries.get(&key).expect("retained owner");
assert_eq!(entry.next_attempt, now + expected_delay);
assert_eq!(entry.consecutive_unverified_attempts, attempts);
assert!(entry.anchor.is_none());
assert!(!entry.retry_queued);
assert!(entry.is_parked());
assert_eq!(entry.source_bucket_incarnation_id, None);
assert!(
matches!(entry.state, ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms: 7, next_probe_at_ms, probe_attempts } if next_probe_at_ms == deadline && probe_attempts == attempts)
);
assert_eq!(runtime.partial_writes.bytes(), PartialWrites::entry_cost(entry));
assert_eq!(runtime.partial_writes.lifecycle_summary().legacy_generation_unknown_count, 1);
}
}
#[tokio::test(start_paused = true)]
async fn lifecycle_v2_reader_persists_failed_probe_pacing_without_object_dispatch() {
let storage = Arc::new(ProbeStorage {
lookup: Mutex::new(Lookup::Identity(None)),
calls: AtomicUsize::new(0),
});
let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None);
let (mut runtime, key) = restored_v2_runtime(0, u32::MAX);
let id = runtime.partial_writes.entries[&key].responsibility_id;
let bytes = runtime.partial_writes.bytes();
for outcome in [
Lookup::Identity(None),
Lookup::PermissionDenied,
Lookup::Quorum,
Lookup::Transport,
Lookup::Identity(Some(Uuid::nil())),
Lookup::Pending,
] {
*storage.lookup.lock().expect("set outcome") = outcome;
runtime.partial_writes.entries.get_mut(&key).expect("owner").next_attempt = Instant::now();
let calls = storage.calls.load(Ordering::Relaxed);
runtime.dirty = false;
runtime.dispatch(&manager).await;
assert!(runtime.dirty, "failed V2 probes still change the durable retry schedule");
assert_eq!(storage.calls.load(Ordering::Relaxed), calls + 1);
let entry = &runtime.partial_writes.entries[&key];
assert_eq!(entry.responsibility_id, id);
assert_eq!(entry.source_bucket_incarnation_id, None);
assert!(entry.anchor.is_none());
assert!(!entry.retry_queued);
assert!(entry.next_attempt > Instant::now());
assert!(entry.next_attempt <= Instant::now() + MAX_UNVERIFIED_RETRY_BACKOFF);
assert!(
matches!(entry.state, ResponsibilityState::SourceAndCurrentGenerationUnknown { since_ms: 7, probe_attempts: u32::MAX, next_probe_at_ms } if next_probe_at_ms > unix_now_ms() && next_probe_at_ms <= unix_now_ms() + 60_000)
);
assert_eq!(runtime.partial_writes.bytes(), bytes);
runtime.dirty = false;
runtime.dispatch(&manager).await;
assert!(!runtime.dirty);
assert_eq!(storage.calls.load(Ordering::Relaxed), calls + 1);
}
let observed = Uuid::new_v4();
*storage.lookup.lock().expect("known outcome") = Lookup::Identity(Some(observed));
runtime.partial_writes.entries.get_mut(&key).expect("owner").next_attempt = Instant::now();
runtime.dispatch(&manager).await;
assert!(runtime.dirty);
assert_eq!(runtime.partial_writes.entries[&key].source_bucket_incarnation_id, None);
assert!(
matches!(runtime.partial_writes.entries[&key].state, ResponsibilityState::LegacyGenerationUnknown { observed_bucket_incarnation_id, detected_at_ms: 7 } if observed_bucket_incarnation_id == observed)
);
assert_eq!(runtime.partial_writes.bytes(), bytes);
assert_eq!(manager.get_queue_length().await, 0);
}
#[tokio::test(start_paused = true)]
async fn lifecycle_v2_reader_rejects_risk_acceptance_and_recheck_without_current_identity() {
let storage = Arc::new(ProbeStorage {
lookup: Mutex::new(Lookup::Identity(Some(Uuid::new_v4()))),
calls: AtomicUsize::new(0),
});
let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None);
let (mut runtime, key) = restored_v2_runtime(0, 3);
let id = runtime.partial_writes.entries[&key].responsibility_id;
let before = runtime.partial_writes.checkpoint_records(super::super::super::intent_digest);
let expected = Uuid::new_v4();
assert!(matches!(
runtime.recheck_legacy_generation(&manager, id, expected).await,
Err(MrfLifecycleControlError::BucketIdentityUnresolved)
));
assert!(matches!(
runtime
.accept_legacy_risk(
&manager,
risk_acceptance(id, expected, true, true, "Reviewed the retained owner", "INC-1234")
)
.await,
Err(MrfLifecycleControlError::BucketIdentityUnresolved)
));
assert_eq!(runtime.partial_writes.checkpoint_records(super::super::super::intent_digest), before);
assert_eq!(
storage.calls.load(Ordering::Relaxed),
0,
"a guessed identity cannot bypass explicit refresh"
);
}
#[test]
fn lifecycle_v2_reader_keeps_first_hold_across_repeated_incarnation_refresh() {
let (mut runtime, key) = restored_v2_runtime(0, 3);
let id = runtime.partial_writes.entries[&key].responsibility_id;
let first = Uuid::new_v4();
let second = Uuid::new_v4();
runtime
.partial_writes
.replace_observed_incarnation(id, None, first, None)
.expect("first observation");
assert!(matches!(
runtime.partial_writes.entries[&key].state,
ResponsibilityState::LegacyGenerationUnknown { detected_at_ms: 7, .. }
));
runtime
.partial_writes
.replace_observed_incarnation(id, Some(first), second, None)
.expect("second observation of the same retained responsibility");
let row = runtime
.list_unverified_legacy(None, 1)
.expect("list the retained responsibility")
.responsibilities
.remove(0);
assert_eq!(row.responsibility_id, id);
assert_eq!(row.source_bucket_incarnation_id, None);
assert_eq!(row.bucket_incarnation_id, Some(second));
assert_eq!(
row.held_since_ms,
Some(7),
"changing the observed bucket cannot reset when this responsibility was first retained"
);
}
crates/heal/src/heal/mrf_queue/tests/empty_events_tests.rs
+1
View File
@@ -48,6 +48,7 @@ fn runtime() -> MrfRuntime {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
lifecycle_format_version: MRF_LIFECYCLE_FORMAT_VERSION,
new_since_flush: 0,
dirty: false,
journal_on_disk: false,
crates/heal/src/heal/mrf_queue/tests/lifecycle_replay_tests.rs
+2 -1
View File
@@ -48,7 +48,7 @@ fn unknown_source_deleted_bucket_lifecycle_rejects_unknown_state_version_and_fie
let valid = lifecycle_frame(&payload);
assert!(decode_mrf_lifecycle_checkpoint(&valid, 8192).is_ok());
let mut future_version = valid;
future_version[8] += 1;
future_version[8] = MRF_LIFECYCLE_FORMAT_VERSION_V2 + 1;
assert!(matches!(
decode_mrf_lifecycle_checkpoint(&future_version, 8192),
Err(MrfReplayError::Snapshot(snapshot::SnapshotError::Unsupported))
@@ -176,6 +176,7 @@ async fn assert_unreadable_lifecycle_is_retained(corrupt: fn(&mut Value)) {
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
lifecycle_format_version: replay.lifecycle_format_version,
new_since_flush: 0,
dirty: true,
journal_on_disk: replay.journal_on_disk,
crates/heal/src/heal/mrf_queue/tests/lifecycle_v2_reader_tests.rs
+359
View File
@@ -0,0 +1,359 @@
// Copyright 2026 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
fn lifecycle_reader_frame(payload: &Value, envelope_version: u8) -> Vec<u8> {
let payload = serde_json::to_vec(payload).expect("encode reader fixture");
let mut encoded = Vec::from(MRF_LIFECYCLE_MAGIC.as_slice());
encoded.push(envelope_version);
encoded.extend_from_slice(&u64::try_from(payload.len()).expect("fixture size").to_le_bytes());
encoded.extend_from_slice(&sha2::Sha256::digest(&payload));
encoded.extend_from_slice(&payload);
encoded
}
fn unknown_generation_payload() -> Value {
let intent_digest = [7_u8; 32];
json!({
"format_version": 2,
"checkpoint_owner": Uuid::new_v4(),
"checkpoint_sequence": 1,
"records": [{
"intent_digest": intent_digest,
"responsibility_id": Uuid::new_v4(),
"source_bucket_incarnation_id": null,
"last_operator_acceptance": null,
"state": {
"state": "source_and_current_generation_unknown",
"since_ms": 7,
"next_probe_at_ms": 60_007,
"probe_attempts": 3
}
}]
})
}
#[test]
fn lifecycle_v2_reader_accepts_retained_unknown_generation_without_a_source_identity() {
let payload = unknown_generation_payload();
let decoded = decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&payload, 2), 8192)
.expect("the reader must understand a valid retained V2 responsibility");
assert_eq!(decoded.format_version, 2);
assert_eq!(decoded.records.len(), 1);
assert!(decoded.records[0].is_valid());
assert_eq!(decoded.records[0].source_bucket_incarnation_id, None);
assert_eq!(
serde_json::to_value(&decoded.records[0].state).expect("serialize restored state"),
payload["records"][0]["state"],
"reading must preserve the first hold and bounded retry information"
);
}
#[test]
fn lifecycle_v2_reader_rejects_mislabeled_and_unsupported_formats() {
let payload = unknown_generation_payload();
for (payload_version, envelope_version) in [(1, 1), (1, 2), (2, 1), (3, 3)] {
let mut invalid = payload.clone();
invalid["format_version"] = json!(payload_version);
assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, envelope_version), 8192).is_err());
}
let decoded = decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&payload, 2), 8192).expect("valid V2 fixture");
for version in [0, 1, 3] {
assert!(
encode_mrf_lifecycle_checkpoint(version, decoded.checkpoint_owner, 1, decoded.records.clone(), 8192).is_none(),
"V2 state cannot be written under format {version}"
);
}
let encoded = encode_mrf_lifecycle_checkpoint(2, decoded.checkpoint_owner, 1, decoded.records.clone(), 8192)
.expect("preserve an observed V2 owner");
assert_eq!(encoded[8], 2);
assert_eq!(
decode_mrf_lifecycle_checkpoint(&encoded, 8192)
.expect("read V2 round trip")
.records,
decoded.records
);
}
#[test]
fn lifecycle_v2_reader_rejects_invalid_identity_retry_fields_and_torn_frames() {
let payload = unknown_generation_payload();
for source in [Uuid::nil(), Uuid::new_v4()] {
let mut invalid = payload.clone();
invalid["records"][0]["source_bucket_incarnation_id"] = json!(source);
assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err());
}
for field in ["since_ms", "next_probe_at_ms", "probe_attempts"] {
for replacement in [None, Some(json!(-1)), Some(json!("unknown"))] {
let mut invalid = payload.clone();
let state = invalid["records"][0]["state"].as_object_mut().expect("state object");
match replacement {
Some(value) => {
state.insert(field.to_string(), value);
}
None => {
state.remove(field);
}
}
assert!(
decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err(),
"reject {field}"
);
}
}
let mut invalid = payload.clone();
invalid["records"][0]["state"]["probe_attempts"] = json!(u64::MAX);
assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err());
invalid = payload.clone();
invalid["records"][0]["state"]["current_incarnation"] = json!(Uuid::new_v4());
assert!(decode_mrf_lifecycle_checkpoint(&lifecycle_reader_frame(&invalid, 2), 8192).is_err());
let encoded = lifecycle_reader_frame(&payload, 2);
for length in [0, 48, encoded.len() - 1] {
assert!(decode_mrf_lifecycle_checkpoint(&encoded[..length], 8192).is_err());
}
assert!(decode_mrf_lifecycle_checkpoint(&encoded, encoded.len() - 1).is_err());
let mut corrupt = encoded;
*corrupt.last_mut().expect("frame body") ^= 1;
assert!(decode_mrf_lifecycle_checkpoint(&corrupt, 8192).is_err());
}
#[test]
#[serial]
fn lifecycle_v2_reader_replays_real_files_without_promoting_v1_or_downgrading_v2() {
const STACK_SIZE: usize = 8 * 1024 * 1024;
std::thread::Builder::new()
.name("mrf-lifecycle-reader".to_owned())
.stack_size(STACK_SIZE)
.spawn(|| {
tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.thread_stack_size(STACK_SIZE)
.enable_all()
.build()
.expect("reader fixture runtime")
.block_on(async {
for version in [1, 2] {
check_reader_restart(version).await;
}
});
})
.expect("reader fixture thread")
.join()
.expect("reader fixture must finish");
}
async fn check_reader_restart(version: u8) {
let env = rustfs_test_utils::TestECStoreEnv::builder()
.prefix("rustfs_mrf_lifecycle_reader")
.build()
.await;
let config = MrfConsumerConfig::default();
let owner = Uuid::new_v4();
let bucket = "reader-unknown-bucket";
let state = if version == 2 {
unknown_generation_payload()["records"][0]["state"].clone()
} else {
json!({"state": "active"})
};
let mut payload = Vec::new();
let mut records = Vec::new();
for index in 0..3 {
let mut original = intent(bucket, &format!("retained-{index}"), 0);
original.kind = MrfKind::PartialWrite;
original.scope = Some(rustfs_common::mrf_channel::MrfScope {
pool_index: 0,
set_index: 0,
});
payload.extend(encoded_payload(&original));
records.push(json!({
"intent_digest": intent_digest(&original).expect("fixture identity"),
"responsibility_id": Uuid::new_v4(),
"source_bucket_incarnation_id": null,
"last_operator_acceptance": null,
"state": state
}));
}
let lifecycle = lifecycle_reader_frame(
&json!({
"format_version": version,
"checkpoint_owner": owner,
"checkpoint_sequence": 1,
"records": records
}),
version,
);
snapshot::publish_committed_snapshot_with_companion(
&journal_disks().await,
owner,
1,
&payload,
config.journal_max_bytes,
Some((&MRF_LIFECYCLE_PATHS, &lifecycle, config.journal_max_bytes * 4)),
)
.await
.expect("publish the reader fixture on real disks");
let storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
let manager = Arc::new(HealManager::new_without_root_recovery_for_test(storage, None));
let ids: HashSet<_> = records
.iter()
.map(|record| serde_json::from_value::<Uuid>(record["responsibility_id"].clone()).expect("fixture id"))
.collect();
let (incoming, receiver) = mpsc::channel(4);
let (_durable, durable_receiver) = mpsc::channel(4);
let (commands, lifecycle_receiver) = mpsc::channel(4);
let consumer = tokio::spawn(run_mrf_consumer(manager.clone(), receiver, durable_receiver, lifecycle_receiver));
let (response, listed) = tokio::sync::oneshot::channel();
commands
.send(MrfLifecycleCommand::List {
after: None,
limit: 3,
response,
})
.await
.expect("query the real consumer after its startup replay");
let listing = tokio::time::timeout(Duration::from_secs(5), listed)
.await
.expect("consumer listing deadline")
.expect("consumer listing response")
.expect("V1 and V2 owners must be readable");
let successor_owner = listing.checkpoint_owner.expect("the real consumer must publish a successor");
let successor_sequence = listing.checkpoint_sequence.expect("successor sequence");
assert_ne!(successor_owner, owner);
assert!(successor_sequence > 1);
assert_eq!(
listing
.responsibilities
.iter()
.map(|row| row.responsibility_id)
.collect::<HashSet<_>>(),
ids
);
drop(incoming);
tokio::time::timeout(Duration::from_secs(5), consumer)
.await
.expect("consumer shutdown deadline")
.expect("consumer shutdown");
let published = read_mrf_lifecycle_checkpoint(
successor_owner,
successor_sequence,
config.journal_max_bytes,
config.journal_max_bytes * 4,
)
.await
.expect("read the actual consumer's successor")
.expect("retained lifecycle companion");
assert_eq!(published.format_version, version);
assert_eq!(published.records.len(), 3);
let mut last_records = None;
for restart in 0..3 {
let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
let replay = replay_into(&manager, &mut queue, &mut None).await;
assert!(replay.source_readable);
assert_eq!(replay.partial_writes.len(), 3);
assert_eq!(replay.lifecycle_format_version, version);
let mut runtime = MrfRuntime {
queue,
partial_writes: PartialWrites::default(),
config: config.clone(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
lifecycle_format_version: replay.lifecycle_format_version,
new_since_flush: 0,
dirty: true,
journal_on_disk: replay.journal_on_disk,
retain_replay_journal: replay.retain_journal_for_replay,
replay_readable: replay.source_readable,
durable_replay_anchors: replay.durable_replay_anchors,
replay_cleanup: replay.cleanup,
runtime_checkpoint: None,
backoff_until: None,
};
runtime.adopt_replayed_partial_writes(replay.partial_writes);
if let Some(previous) = &last_records {
let current: HashMap<_, _> = runtime
.partial_writes
.checkpoint_records(intent_digest)
.into_iter()
.map(|record| (record.responsibility_id, record))
.collect();
assert_eq!(&current, previous, "replay must preserve every state and audit field");
}
assert!(runtime.flush().await);
runtime.dispatch(&manager).await;
let mut cursor = None;
let mut visible = HashSet::new();
loop {
let page = runtime.list_unverified_legacy(cursor, 1).expect("list retained records");
assert_eq!(page.legacy_generation_unknown_count, 3);
assert_eq!(page.responsibilities.len(), 1);
let row = &page.responsibilities[0];
assert!(visible.insert(row.responsibility_id));
assert_eq!(row.source_bucket_incarnation_id, None);
assert_eq!((row.pool_index, row.set_index), (Some(0), Some(0)));
assert_eq!(row.version_id, Some(Uuid::from_bytes([7; 16]).to_string()));
if restart < 2 {
assert_eq!(row.bucket_incarnation_id, None);
assert_eq!(
row.status,
if version == 2 {
"source_and_current_generation_unknown"
} else {
"legacy_generation_unobserved"
}
);
assert_eq!(row.held_since_ms, (version == 2).then_some(7));
assert!(row.next_identity_probe_at_ms.is_some());
} else {
assert_eq!(row.status, "legacy_generation_unknown");
assert!(row.bucket_incarnation_id.is_some());
if version == 2 {
assert_eq!(row.held_since_ms, Some(7));
}
}
cursor = page.next_cursor;
if cursor.is_none() {
break;
}
}
assert_eq!(visible, ids);
assert_eq!(manager.get_queue_length().await, 0, "unknown source never authorizes object healing");
if restart == 1 {
env.make_bucket(bucket, false).await;
for id in &ids {
runtime
.refresh_legacy_generation(&manager, *id, None)
.await
.expect("record current identity without rebinding source");
}
}
let (checkpoint_owner, sequence) = runtime.runtime_checkpoint.expect("successor checkpoint");
let stored =
read_mrf_lifecycle_checkpoint(checkpoint_owner, sequence, config.journal_max_bytes, config.journal_max_bytes * 4)
.await
.expect("read actual successor files")
.expect("retained companion");
assert_eq!(
stored.format_version, version,
"owner transfer must preserve the format even after observing a bucket"
);
last_records = Some(
stored
.records
.into_iter()
.map(|record| (record.responsibility_id, record))
.collect::<HashMap<_, _>>(),
);
}
}
crates/heal/src/heal/storage_api.rs
+2 -2
View File
@@ -41,10 +41,10 @@ pub(crate) use rustfs_ecstore::api::storage::{
use rustfs_storage_api as storage_contracts;
pub(crate) mod owner {
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::disk::PhysicalDiskBinding;
pub(crate) use rustfs_ecstore::api::disk::PhysicalResourcePlan;
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::disk::{PhysicalDiskBinding, PhysicalDiskSample, PhysicalDiskSampler};
#[cfg(test)]
pub(crate) use rustfs_ecstore::spawn_blocking_with_resource_ownership as ecstore_spawn_blocking_with_resource_ownership;
pub(crate) use rustfs_ecstore::with_background_resource_set as ecstore_with_background_resource_set;
docs/architecture/erasure-coding.md
+2
View File
@@ -302,6 +302,8 @@ Small objects store their payload inline after the container CRC ([filemeta_inli
- Encode-time gates: writable disks `< write_quorum` ⇒ `ErasureWriteQuorum`; committed shards `< write_quorum` after encode ⇒ error ([set_disk/ops/object.rs](../../crates/ecstore/src/set_disk/ops/object.rs)).
- **INVARIANT — atomic commit with best-effort rollback.** Commit is `rename_data` (per-disk temp → final) fanned across all disks ([core/io_primitives.rs](../../crates/ecstore/src/set_disk/core/io_primitives.rs)). If write quorum is not met (`reduce_write_quorum_errs`), every successful disk is undone (`delete_version{undo_write:true}`) and the original quorum error is returned. **Baseline:** the rollback is **best-effort** — undo failures are counted and `warn!`-logged, never propagated or retried — so a write that both misses quorum *and* whose rollback partially fails can leave shards on some disks; that partial residue is reconciled later by heal/scanner, not by the commit path. The guarantee the commit path enforces is "never *reports* success below quorum", not "never leaves any bytes behind".
- **INVARIANT — separate active uploads from recovery authority.** Before multipart publication, `LocalDisk::rename_data_inner` moves the original upload `xl.meta` into `.abort-authority/xl.meta`, then publishes object metadata through `xl.meta.complete`. The recovery move holds the source metadata mutation lease and is synced under the existing system-metadata durability policy before payload publication. Late upload metadata writes refuse while recovery authority exists. Decoded upload identifiers must remain a single safe path component, so recovery children cannot be addressed as active uploads. Original metadata bytes and payload shards are not copied; shards still move by rename. Successful per-disk publication no longer leaves active upload metadata, so a completed upload cannot accept more parts when set-level cleanup fails. `SetDisks::complete_multipart_upload` owns root cleanup after success; ordinary Abort combines primary and recovery metadata with at most one vote per disk, then retains its full deletion quorum and existing incarnation, access and lifecycle fences. Stale/internal cleanup can read recovery authority; orphan cleanup must prove both authority locations and any prepared metadata absent. Empty upload metadata retains its existing half-set identity quorum, independently of the payload's data-shard count. Older peers and historical residue without enough authority remain a refusal, not an inferred cleanup success.
- **INVARIANT — completion replay needs write authority.** `SetDisks::replay_completed_multipart_upload` requires a matching upload ID and part list, then `confirm_completed_multipart_write_quorum` corroborates the same version, data directory, modification time and payload layout at its full write quorum. A readable minority from an uncertain publication is insufficient, including empty objects. Recovery metadata alone never proves completion and cannot reopen an active upload. When no committed object can be proved but recovery authority remains, Complete refuses rather than inventing NoSuchUpload. Healthy first completions add no destination metadata probe. Unversioned/current-version replay uses two ReadVersion fan-outs; historical-version lookup can additionally scan version metadata. A previously committed object below write quorum can remain readable while completion replay refuses until write quorum returns.
- On success the newly committed dir is `fi.data_dir`. Separately, `reduce_common_data_dir` votes over each disk's **`old_data_dir`** (the *superseded* dir being dereferenced) and returns it when it reaches write_quorum, so the old dir can be reclaimed (`commit_rename_data_dir`) — it is a GC input, **not** the new `data_dir`. `classify_rename_convergence` classifies the commit (`PartialCommit` / `SignatureDivergent`), but **only the multipart-complete path consumes it** (`convergence.needs_heal()` → `send_heal_request`); the regular `put_object` path discards the convergence result and relies on the old-data-dir cleanup / `add_partial` heal enqueue instead.
- The write layout (per-pool parity, storage class, `max_parity`) is resolved once by `resolve_write_layout` into a `WriteLayout` ([set_disk/mod.rs](../../crates/ecstore/src/set_disk/mod.rs)) and consumed by the object and multipart write paths ([set_disk/ops/object.rs](../../crates/ecstore/src/set_disk/ops/object.rs), [set_disk/ops/multipart.rs](../../crates/ecstore/src/set_disk/ops/multipart.rs)); resolution is per pool (§2.2).
docs/operations/mrf-responsibilities.md
+7 -3
View File
@@ -12,7 +12,9 @@
These entries stay outside object-heal dispatch. After an initial 60-second delay, the consumer rotates through a bounded probe queue, attempts at most one identity lookup per dispatch round, and limits each lookup to one second. Failed or unresolved attempts use the existing bounded backoff, with a five-second minimum and a 60-second ceiling. The queue examination budget is `RUSTFS_HEAL_MRF_REPLAY_BATCH`.
The listing exposes `lastIdentityProbeResult` (`unresolved`, `unavailable`, or `known`), `lastIdentityProbeAtMs`, and `nextIdentityProbeAtMs` when available. These are process-local observations. Restart resets the probe delay; these timestamps do not establish when a responsibility was first retained.
The listing exposes `lastIdentityProbeResult` (`unresolved`, `unavailable`, or `known`), `lastIdentityProbeAtMs`, and `nextIdentityProbeAtMs` when available. The result and last-probe timestamp are process-local observations. For lifecycle V1, restart resets the probe delay; these timestamps do not establish when a responsibility was first retained.
A lifecycle V2 checkpoint can contain `source_and_current_generation_unknown`. It retains the first-held timestamp, next-probe timestamp, and attempt count. Both incarnation fields remain null, and the listing exposes the first-held timestamp as `heldSinceMs`. Restart restores the remaining delay within five to 60 seconds, including after clock changes or an expired deadline. Identity probes retain their schedule after unsuccessful lookups. Risk acceptance and targeted recheck require a known current incarnation; a guessed identity cannot bypass refresh.
When a same-name bucket becomes visible, its identity is recorded as the observed current incarnation in `legacy_generation_unknown`. The source stays null. This transition does not enqueue object repair or create a repair proof. Targeted recheck remains rejected because the original generation cannot be established.
@@ -28,6 +30,8 @@ Do not delete a companion, journal, or commit manifest to clear this condition.
## Format and rollout boundary
The writer continues to emit lifecycle format version 1. A source-less `Active` record still means a parked legacy responsibility, and observing a current incarnation uses the existing `LegacyGenerationUnknown` state. No new persistent state is activated by identity probing.
Owners that have not read a V2 checkpoint continue to emit lifecycle format version 1. A source-less `Active` record still means a parked legacy responsibility, and observing a current incarnation uses the existing `LegacyGenerationUnknown` state. No new persistent state is activated by admission or identity probing.
A durable state for unknown source and current generations, including the first-retained time and probe schedule across restarts, needs a separate reader-first rollout. The existing heal-control and cross-pool protocol capabilities do not prove support for a new lifecycle state. New writers require evidence for every reader and the minimum supported rollback reader; do not downgrade past that reader or silently convert an unknown state to `Active`.
The reader accepts lifecycle formats 1 and 2. Replaying a V2 owner preserves that format in successor checkpoints, even after the current bucket becomes visible; it never writes the new state inside a V1 envelope. A valid V2 unknown-generation state cannot contain a source incarnation. Unknown fields, missing retry fields, unsupported versions, and mismatched envelope/payload versions remain unreadable.
Creating the new state requires a separate writer rollout. The existing heal-control and cross-pool protocol capabilities do not prove support for it. New writers require pinned read/restart evidence for every supported reader and an explicit minimum rollback reader. Reader support alone does not qualify writer activation; do not downgrade past that reader or silently convert an unknown state to `Active`.
docs/testing/distributed-e2e.md
+2
View File
@@ -110,6 +110,8 @@ Membership is pinned by `.config/e2e-distributed-selection.txt`. Update the Linu
The `distributed::multipart_fence_acceptance_test` cases separate remote storage RPC delays from lock RPC delays through a method-selective proxy. They require the selected RPC to reach its pause before checking quorum, recovery, or cancellation. The proxy forwards opaque bodies and gRPC trailers and is confined to the test process. It does not change server timeouts, locking, retry policy, or the commit path.
After a quorum-minus-one Complete is rejected and faults are retired, ordinary Abort must succeed. `NoSuchUpload` is not accepted as recovery of that failed completion: it can hide staging metadata consumed by partial publication. Already completed or aborted sessions retain their existing missing-upload cleanup semantics. The ECStore multipart regressions assert unchanged per-disk authority bytes across primary/recovery locations, no duplicated disk votes, rejection of completed-upload resends after cleanup failure, orphan-cleanup refusal while authority remains, preservation of the previous object, and write-quorum completion replay. Scoped Abort faults decode the exact multipart volume, root path and recursive option without changing signed bytes. Post-fault read probes can wait within the existing recovery deadline for classified 503 responses, honoring Retry-After; injected writes and successful streaming bodies remain single-attempt.
Lock-layer regressions also cover every four-node read-quorum ordering, an eight-node write quorum behind silent voters, and the initial fan-out bound for prompt replies. Distributed acquisition probes unstarted voters within its existing attempt budget so silent early peers cannot hide a reachable quorum. Vote thresholds, owner identities, heartbeat checks, and late-grant cleanup retain their existing contracts.
Faults match the test bucket or upload suffix in bounded control-packet bodies, preserving their signed bytes and trailers. Abort injection decodes only the unary Delete routing fields and matches the exact multipart root, excluding child transaction cleanup. Readiness requires an actual matching RPC, without waiting for voters that adaptive fan-out does not contact. Recovery cases release waiting packets while the request is active; quorum cases discard expired packets before restoring the peer. Replaying a packet after its caller has timed out is a separate late-delivery fault model. The injected S3 operation has SDK retries disabled. After the fault is removed, ordinary cleanup must recover within 45 seconds, allowing the existing lease and drive-health recovery intervals; only classified ServiceUnavailable responses are polled, and the server's Retry-After advice is honored. The fault-free eight-node workload keeps single-attempt APIs throughout. Failed Complete uses the best-effort rollback contract in [erasure-coding.md](../architecture/erasure-coding.md#7-write-path-and-write-quorum). A below-quorum Abort can also leave partially deleted staging. Each matrix reuses one cluster and bucket, requiring fresh responses from every lock peer before the next row so circuit-breaker history cannot reduce its available votes. After a partial mutation, ordinary Abort must succeed or report an already retired session, ListParts must return NoSuchUpload, and every node must list no residual uploads. This verifies the Abort recovery merged in #8904 without weakening the delete quorum; it does not claim residue-free rollback of a failed Complete. Bucket-generation recreation is covered by the cancellation case.
rustfs/tests/connect_inventory.rs
+43 -22
View File
@@ -16,7 +16,7 @@
use std::collections::VecDeque;
use std::fs;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
@@ -79,7 +79,7 @@ impl TestPki {
server_params.not_after = now + time::Duration::days(2);
server_params
.subject_alt_names
.push(SanType::DnsName("localhost".try_into().expect("valid DNS name")));
.push(SanType::IpAddress(std::net::Ipv4Addr::LOCALHOST.into()));
server_params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth];
let server = server_params
.signed_by(&server_key, &Issuer::from_params(&root_params, &root_key))
@@ -182,10 +182,27 @@ impl Reply {
struct TestServer {
endpoint: String,
seen: Arc<Mutex<Vec<Value>>>,
seen: watch::Sender<Vec<Value>>,
reject_next_connection: Arc<AtomicBool>,
task: tokio::task::JoinHandle<()>,
}
impl TestServer {
async fn first_request(&self) -> Value {
let mut seen = self.seen.subscribe();
tokio::time::timeout(INVENTORY_STATUS_WAIT_TIMEOUT, async {
seen.wait_for(|requests| !requests.is_empty())
.await
.expect("inventory request observer")
.first()
.expect("observed inventory request")
.clone()
})
.await
.expect("server must receive an inventory request")
}
}
impl Drop for TestServer {
fn drop(&mut self) {
self.task.abort();
@@ -197,10 +214,15 @@ async fn server(pki: &TestPki, replies: Vec<Reply>) -> TestServer {
let address = listener.local_addr().expect("server address");
let acceptor = TlsAcceptor::from(Arc::new(pki.server_config()));
let replies = Arc::new(Mutex::new(VecDeque::from(replies)));
let seen = Arc::new(Mutex::new(Vec::new()));
let (seen, _) = watch::channel(Vec::new());
let captured = seen.clone();
let reject_next_connection = Arc::new(AtomicBool::new(false));
let reject = reject_next_connection.clone();
let task = tokio::spawn(async move {
while let Ok((stream, _)) = listener.accept().await {
if reject.swap(false, Ordering::Relaxed) {
continue;
}
let acceptor = acceptor.clone();
let replies = replies.clone();
let seen = captured.clone();
@@ -212,9 +234,8 @@ async fn server(pki: &TestPki, replies: Vec<Reply>) -> TestServer {
async move {
assert_eq!(request.uri().path(), format!("/agent/clusters/{CLUSTER_UID}/inventorySnapshots"));
let body = request.into_body().collect().await.expect("request body").to_bytes();
seen.lock()
.expect("seen lock")
.push(serde_json::from_slice(&body).expect("request JSON"));
let request = serde_json::from_slice(&body).expect("request JSON");
seen.send_modify(|requests| requests.push(request));
let reply = replies
.lock()
.expect("reply lock")
@@ -240,8 +261,9 @@ async fn server(pki: &TestPki, replies: Vec<Reply>) -> TestServer {
}
});
TestServer {
endpoint: format!("https://localhost:{}/agent/", address.port()),
endpoint: format!("https://{address}/agent/"),
seen,
reject_next_connection,
task,
}
}
@@ -470,7 +492,7 @@ async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_o
assert!(!temp.path().join("private-config-secret/inventory/state.json").exists());
runtime.shutdown().await;
}
assert!(server.seen.lock().expect("seen lock").is_empty());
assert!(server.seen.borrow().is_empty());
}
#[tokio::test]
@@ -478,6 +500,8 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
let pki = TestPki::new();
let content_hash = snapshot().content_hash().expect("content hash");
let first_server = server(&pki, vec![Reply::error(StatusCode::SERVICE_UNAVAILABLE, "UNAVAILABLE")]).await;
// A transport retry can happen before the server receives any HTTP request.
first_server.reject_next_connection.store(true, Ordering::Relaxed);
let temp = safe_tempdir();
let shutdown = CancellationToken::new();
let samples = Arc::new(AtomicUsize::new(0));
@@ -488,14 +512,11 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
})
.expect("start inventory")
.expect("configured inventory");
let mut status = runtime.status();
assert!(matches!(
wait_for(&mut status, |status| matches!(status, InventoryStatus::BackingOff { .. })).await,
InventoryStatus::BackingOff { delay } if delay == Duration::from_millis(20)
));
// BackingOff also describes DNS, connect, and TLS failures. Observe the
// actual POST before shutting down with its persisted request still pending.
let original = first_server.first_request().await;
assert_eq!(samples.load(Ordering::Relaxed), 1);
let original = first_server.seen.lock().expect("seen lock")[0].clone();
let latest: Value = serde_json::from_slice(
&fs::read(temp.path().join("private-config-secret/inventory/latest.json")).expect("latest inventory"),
)
@@ -557,7 +578,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
.expect("restored latest envelope");
assert_eq!(restored_latest["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON"));
assert_eq!(restored_latest["capturedAt"], legacy_persisted_at);
let delivered = restart_server.seen.lock().expect("seen lock").clone();
let delivered = restart_server.seen.borrow().clone();
assert_eq!(delivered, vec![original.clone(), original.clone()]);
assert_eq!(original["sequence"], 0);
let encoded = serde_json::to_string(&original).expect("request JSON");
@@ -597,7 +618,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
InventoryStatus::Unchanged { content_hash: unchanged } if unchanged == content_hash
));
assert_eq!(unchanged_samples.load(Ordering::Relaxed), 1);
assert_eq!(restart_server.seen.lock().expect("seen lock").len(), 2);
assert_eq!(restart_server.seen.borrow().len(), 2);
#[cfg(target_os = "linux")]
{
use std::os::unix::fs::MetadataExt as _;
@@ -675,7 +696,7 @@ async fn connect_inventory_retries_an_incomplete_sample_before_delivery() {
InventoryStatus::Online { content_hash: accepted, .. } if accepted == content_hash
));
assert_eq!(samples.load(Ordering::Relaxed), 2);
assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
assert_eq!(server.seen.borrow().len(), 1);
runtime.shutdown().await;
}
@@ -746,7 +767,7 @@ async fn connect_inventory_unchanged_sample_resets_incomplete_backoff() {
InventoryStatus::BackingOff { delay } if delay == Duration::from_millis(20)
));
assert_eq!(samples.load(Ordering::Relaxed), 4);
assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
assert_eq!(server.seen.borrow().len(), 1);
runtime.shutdown().await;
}
@@ -770,7 +791,7 @@ async fn connect_inventory_revoked_device_stops_without_retrying() {
reason: None
}
));
assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
assert_eq!(server.seen.borrow().len(), 1);
runtime.shutdown().await;
}
@@ -805,7 +826,7 @@ async fn connect_inventory_sequence_overflow_fails_before_sampling_or_network_de
InventoryStatus::Failed { reason } if reason == "connect_inventory_sequence_exhausted"
));
assert_eq!(samples.load(Ordering::Relaxed), 0);
assert!(server.seen.lock().expect("seen lock").is_empty());
assert!(server.seen.borrow().is_empty());
runtime.shutdown().await;
}
@@ -873,7 +894,7 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv
runtime.shutdown().await;
}
assert!(server.seen.lock().expect("seen lock").is_empty());
assert!(server.seen.borrow().is_empty());
}
#[cfg(target_os = "linux")]