gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-08 21:25:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): publish committed MRF runtime checkpoints (#7490)

Browse Source 
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-08 20:32:22 +08:00
committed by GitHub
parent 4d68c32b75
commit ee5f76c180
3 changed files with 341 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/mrf_queue.rs
+143 -7
View File
@@ -521,6 +521,8 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
struct MrfRuntime {
queue: MrfQueue,
config: MrfConsumerConfig,
checkpoint_owner: Uuid,
next_checkpoint_sequence: u64,
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
@@ -537,6 +539,12 @@ struct MrfRuntime {
/// Partial-write responsibilities accepted from replay and waiting for an
/// exact storage-owned proof before the startup journal can be deleted.
durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
/// Startup replay source to remove after the retained replay
/// responsibilities are discharged. `None` means the runtime only needs
/// the legacy journal cleanup path for snapshots it wrote itself.
replay_cleanup: Option<ReplayCleanup>,
/// Last committed checkpoint published by this runtime flush path.
runtime_checkpoint: Option<(Uuid, u64)>,
/// Earliest instant a full-admission retry may proceed.
backoff_until: Option<tokio::time::Instant>,
}
@@ -560,6 +568,34 @@ impl MrfRuntime {
async fn flush(&mut self) {
let (authoritative, legacy) = self.snapshot();
let (committed_persisted, committed_on_disk) = if authoritative.is_empty() {
(true, false)
} else {
match snapshot::publish_committed_snapshot(
&journal_disks().await,
self.checkpoint_owner,
self.next_checkpoint_sequence,
&authoritative,
self.config.journal_max_bytes,
)
.await
{
Ok(publication) => {
self.runtime_checkpoint = Some((publication.owner, publication.sequence));
self.next_checkpoint_sequence = publication.sequence.saturating_add(1);
(true, true)
}
Err(err) => {
tracing::warn!(
target: "rustfs::heal::mrf",
error = %err,
sequence = self.next_checkpoint_sequence,
"MRF committed checkpoint publish failed; retaining previous replay anchor"
);
(false, false)
}
}
};
let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await;
if !authoritative.is_empty() {
counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
@@ -570,10 +606,10 @@ impl MrfRuntime {
// old reader from observing a newer epoch that a new reader cannot
// see when the canonical write is unavailable.
let legacy_persisted = authoritative_persisted && write_journal(MRF_JOURNAL_PATH, &legacy).await;
// Keep dirty until both the authoritative snapshot and its
// compatibility mirror have been accepted; otherwise a one-sided
// failure would never retry the missing file.
let persisted = authoritative_persisted && legacy_persisted;
// Keep dirty until the committed checkpoint, authoritative snapshot,
// and compatibility mirror have all been accepted; otherwise a
// one-sided failure would never retry the missing recovery anchor.
let persisted = committed_persisted && authoritative_persisted && legacy_persisted;
self.new_since_flush = 0;
// Keep the dirty flag when every disk write failed: a clean backlog
// would otherwise never rewrite, losing the periodic persist retry a
@@ -581,7 +617,7 @@ impl MrfRuntime {
if persisted {
self.dirty = false;
}
self.journal_on_disk |= authoritative_persisted || legacy_persisted;
self.journal_on_disk |= committed_on_disk || authoritative_persisted || legacy_persisted;
}
/// Drain pending intents into the heal manager until it is full, the
@@ -639,6 +675,45 @@ impl MrfRuntime {
self.retain_replay_journal || !self.durable_replay_anchors.is_empty()
}
fn replay_cleanup_to_delete(&self) -> Option<ReplayCleanup> {
if self.journal_on_disk && !self.retained_replay_journal() {
Some(self.replay_cleanup.unwrap_or(ReplayCleanup::Legacy))
} else {
None
}
}
async fn delete_idle_recovery_anchors(&mut self) -> bool {
let runtime_deleted = match self.runtime_checkpoint {
Some((owner, sequence)) => {
match snapshot::delete_committed_snapshots_through(owner, sequence, self.config.journal_max_bytes).await {
Ok(deleted) => deleted,
Err(err) => {
tracing::warn!(
target: "rustfs::heal::mrf",
error = %err,
sequence,
"MRF runtime checkpoint cleanup failed"
);
false
}
}
}
None => true,
};
let replay_deleted = match self.replay_cleanup_to_delete() {
Some(cleanup) => delete_replay_source(cleanup, self.config.journal_max_bytes).await,
None => true,
};
if runtime_deleted && replay_deleted {
self.runtime_checkpoint = None;
self.replay_cleanup = None;
true
} else {
false
}
}
fn discharge_durable_replay_anchors(&mut self) {
if self.durable_replay_anchors.is_empty() {
return;
@@ -708,6 +783,8 @@ struct ReplayOutcome {
journal_on_disk: bool,
retain_journal_for_replay: bool,
durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
cleanup: Option<ReplayCleanup>,
next_checkpoint_sequence: u64,
}
fn replay_must_retain_journal(
@@ -719,7 +796,7 @@ fn replay_must_retain_journal(
rearm_incomplete || pending_depth > 0 || accepted_without_durable_anchor || durable_replay_anchors > 0
}
#[derive(Clone, Copy)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ReplayCleanup {
Legacy,
Committed { owner: Uuid, sequence: u64 },
@@ -793,6 +870,8 @@ async fn replay_into(
journal_on_disk: false,
retain_journal_for_replay: false,
durable_replay_anchors: Vec::new(),
cleanup: None,
next_checkpoint_sequence: 1,
};
}
Err(err) => {
@@ -806,10 +885,16 @@ async fn replay_into(
journal_on_disk: true,
retain_journal_for_replay: true,
durable_replay_anchors: Vec::new(),
cleanup: None,
next_checkpoint_sequence: 1,
};
}
};
let cleanup = source.cleanup;
let next_checkpoint_sequence = match cleanup {
ReplayCleanup::Legacy => 1,
ReplayCleanup::Committed { sequence, .. } => sequence.saturating_add(1),
};
let data = source.data;
let (decoded, truncated) = decode_journal(&data);
let replayed = decoded.len();
@@ -913,6 +998,8 @@ async fn replay_into(
journal_on_disk,
retain_journal_for_replay,
durable_replay_anchors,
cleanup: journal_on_disk.then_some(cleanup),
next_checkpoint_sequence,
}
}
@@ -923,11 +1010,15 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
let mut runtime = MrfRuntime {
queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
config: config.clone(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
new_since_flush: 0,
dirty: false,
journal_on_disk: false,
retain_replay_journal: false,
durable_replay_anchors: Vec::new(),
replay_cleanup: None,
runtime_checkpoint: None,
backoff_until: None,
};
@@ -937,6 +1028,8 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
runtime.journal_on_disk = replay.journal_on_disk;
runtime.retain_replay_journal = replay.retain_journal_for_replay;
runtime.durable_replay_anchors = replay.durable_replay_anchors;
runtime.replay_cleanup = replay.cleanup;
runtime.next_checkpoint_sequence = replay.next_checkpoint_sequence;
// 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
// startup anchor.
@@ -1001,7 +1094,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
TickAction::DeleteJournal => {
// All replayed intents have either been accepted,
// merged, or replaced by a pending successor snapshot.
if delete_journals().await {
if runtime.delete_idle_recovery_anchors().await {
runtime.journal_on_disk = false;
gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
}
@@ -1117,14 +1210,23 @@ mod tests {
let bucket_incarnation_id = uuid::Uuid::new_v4();
let anchor = rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&intent, bucket_incarnation_id)
.expect("fresh replay lease and bucket incarnation build a durable anchor");
let cleanup_owner = uuid::Uuid::new_v4();
let cleanup = ReplayCleanup::Committed {
owner: cleanup_owner,
sequence: 17,
};
let mut runtime = MrfRuntime {
queue: MrfQueue::new(2, usize::MAX),
config: MrfConsumerConfig::default(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
new_since_flush: 0,
dirty: false,
journal_on_disk: true,
retain_replay_journal: false,
durable_replay_anchors: vec![anchor],
replay_cleanup: Some(cleanup),
runtime_checkpoint: None,
backoff_until: None,
};
rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
@@ -1142,14 +1244,48 @@ mod tests {
runtime.retained_replay_journal(),
"anchor must retain the startup journal before proof is consumed"
);
assert_eq!(
runtime.replay_cleanup_to_delete(),
None,
"the committed replay source must not be reclaimed before the exact proof"
);
runtime.discharge_durable_replay_anchors();
assert!(
!runtime.retained_replay_journal(),
"matching verified proof discharges the durable replay anchor"
);
assert_eq!(
runtime.replay_cleanup_to_delete(),
Some(cleanup),
"proof discharge must preserve the committed owner/sequence cleanup target"
);
rustfs_common::mrf_channel::release_mrf_intent(&intent);
}
#[test]
fn runtime_cleanup_defaults_to_legacy_for_runtime_written_journals() {
let runtime = MrfRuntime {
queue: MrfQueue::new(2, usize::MAX),
config: MrfConsumerConfig::default(),
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: 1,
new_since_flush: 0,
dirty: false,
journal_on_disk: true,
retain_replay_journal: false,
durable_replay_anchors: Vec::new(),
replay_cleanup: None,
runtime_checkpoint: None,
backoff_until: None,
};
assert_eq!(
runtime.replay_cleanup_to_delete(),
Some(ReplayCleanup::Legacy),
"journals written by the runtime still use the legacy cleanup path"
);
}
#[test]
fn durable_replay_acquires_a_fresh_lease_before_manager_admission() {
let unique = uuid::Uuid::new_v4();
crates/heal/src/heal/mrf_queue/snapshot.rs
+154 -28
View File
@@ -22,8 +22,9 @@
//! responsibility through a newer durable snapshot or a verified repair proof.
//! An unreadable commit path cannot prove that only legacy data exists. This
//! explicit inspection API fails closed and never mutates recovery anchors.
//! It is wired into the replay reader before writer activation, but the writer
//! remains gated on ownership-aware handoff.
//! The live consumer writes committed checkpoints alongside the scoped and
//! legacy journal mirrors; cleanup remains gated by replay ownership and exact
//! verified repair proof handoff.
//! One surviving committed replica supports process restart recovery only;
//! this reader does not establish a replication quorum or a power-loss policy.
@@ -356,8 +357,8 @@ fn validate_reusable_manifest_slot(existing: Option<&[u8]>, sequence: u64, paylo
/// The writer is a narrow production primitive for the ownership-aware MRF
/// handoff: it validates the whole journal payload, preserves the previous
/// committed slot, and publishes the manifest only after the successor payload
/// reaches the same disk. It does not delete legacy journals, tombstone older
/// anchors, or activate the live consumer.
/// reaches the same disk. It does not delete legacy journals or tombstone older
/// anchors by itself; the consumer decides cleanup after replay handoff.
pub async fn publish_committed_snapshot(
disks: &[EcstoreDiskStore],
owner: Uuid,
@@ -371,7 +372,10 @@ pub async fn publish_committed_snapshot(
if owner.is_nil() || sequence == 0 || sequence == u64::MAX {
return Err(SnapshotError::Corrupt);
}
if payload.len() > limit || decode_journal(payload).1 != 0 {
if payload.len() > limit {
return Err(SnapshotError::TooLarge);
}
if decode_journal(payload).1 != 0 {
return Err(SnapshotError::Corrupt);
}
let current = read_committed(disks, limit).await?;
@@ -555,13 +559,13 @@ pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option
read_committed(&super::journal_disks().await, max_bytes).await
}
/// Remove committed manifests from `owner` whose sequence is no newer than
/// Remove committed checkpoints from `owner` whose sequence is no newer than
/// `committed_through`.
///
/// Payload files are intentionally left as orphans after their manifest is
/// removed. Readers cannot discover a payload without its matching manifest,
/// and deleting manifests first prevents an older retained slot from becoming
/// visible again after the newest replay has been fully discharged.
/// Cleanup is manifest-first so readers cannot rediscover an older payload
/// after the newest replay has been fully discharged. The payload is removed
/// only after the manifest and body were revalidated as one complete committed
/// checkpoint; damaged, future, mismatched, or foreign-owner slots are retained.
pub async fn delete_committed_snapshots_through(
owner: Uuid,
committed_through: u64,
@@ -580,11 +584,10 @@ async fn delete_committed_snapshots_through_on(
if disks.is_empty() {
return Err(SnapshotError::NoWritableReplica);
}
let mut any_changed = false;
let mut first_error = None;
for disk in disks {
for path in MANIFEST_PATHS {
let existing = match read_bounded(disk, path, MANIFEST_LEN).await {
for (manifest_path, payload_path) in MANIFEST_PATHS.into_iter().zip(PAYLOAD_PATHS) {
let manifest_bytes = match read_bounded(disk, manifest_path, MANIFEST_LEN).await {
Ok(Some(existing)) => existing,
Ok(None) => continue,
Err(error) => {
@@ -594,7 +597,7 @@ async fn delete_committed_snapshots_through_on(
continue;
}
};
let manifest = match Manifest::decode(&existing, max_bytes) {
let manifest = match Manifest::decode(&manifest_bytes, max_bytes) {
Ok(manifest) => manifest,
Err(error) => {
if first_error.is_none() {
@@ -606,16 +609,58 @@ async fn delete_committed_snapshots_through_on(
if manifest.owner != owner || manifest.sequence > committed_through {
continue;
}
let payload_bytes = match read_bounded(disk, payload_path, manifest.payload_len).await {
Ok(Some(payload)) => payload,
Ok(None) => {
if first_error.is_none() {
first_error = Some(SnapshotError::Corrupt);
}
continue;
}
Err(error) => {
if first_error.is_none() {
first_error = Some(error);
}
continue;
}
};
if let Err(error) = CommittedSnapshot::decode(0, &manifest_bytes, payload_bytes.clone(), max_bytes) {
if first_error.is_none() {
first_error = Some(error);
}
continue;
}
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
path,
Some(EcstoreDiskBytes::from(existing)),
manifest_path,
Some(EcstoreDiskBytes::copy_from_slice(&manifest_bytes)),
None,
)
.await
{
Ok(EcstoreConditionalFileUpdate::Updated) => any_changed = true,
Ok(EcstoreConditionalFileUpdate::Updated) => {
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
payload_path,
Some(EcstoreDiskBytes::copy_from_slice(&payload_bytes)),
None,
)
.await
{
Ok(
EcstoreConditionalFileUpdate::Updated
| EcstoreConditionalFileUpdate::Missing
| EcstoreConditionalFileUpdate::Mismatch,
) => {}
Err(error) => {
if first_error.is_none() {
first_error = Some(SnapshotError::Disk(error));
}
}
}
}
Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {}
Err(error) => {
if first_error.is_none() {
@@ -625,13 +670,7 @@ async fn delete_committed_snapshots_through_on(
}
}
}
if any_changed {
Ok(true)
} else if let Some(error) = first_error {
Err(error)
} else {
Ok(true)
}
if let Some(error) = first_error { Err(error) } else { Ok(true) }
}
async fn read_recovery_snapshot(disks: &[EcstoreDiskStore], limit: usize) -> Result<Option<RecoverySnapshot>, SnapshotError> {
@@ -1101,6 +1140,45 @@ mod tests {
);
}
#[tokio::test]
async fn committed_snapshot_writer_capacity_failure_preserves_previous_anchor() {
let root = TempDir::new().expect("test directory");
let store = disk(&root, "disk").await;
let owner = Uuid::new_v4();
let old = payload("old");
let next = payload("next");
commit(&store, 0, owner, 1, &old).await;
let result = publish_committed_snapshot(std::slice::from_ref(&store), owner, 2, &next, next.len() - 1).await;
assert!(
matches!(result, Err(SnapshotError::TooLarge)),
"capacity failure must be reported separately from corruption: {result:?}"
);
let reopened = disk(&root, "disk").await;
let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
.await
.expect("read previous committed snapshot")
.expect("old anchor remains committed");
assert_eq!(recovered.sequence(), 1);
assert_eq!(recovered.slot(), 0);
assert_eq!(recovered.payload(), old.as_slice());
assert_eq!(
EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
.await
.expect("old manifest retained")
.as_ref(),
manifest(owner, 1, &old).as_slice()
);
assert!(
matches!(
EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1]).await,
Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
),
"oversized successor payload must not be staged"
);
}
#[tokio::test]
async fn committed_snapshot_writer_does_not_overwrite_damaged_inactive_manifest() {
let root = TempDir::new().expect("test directory");
@@ -1355,7 +1433,7 @@ mod tests {
}
#[tokio::test]
async fn committed_cleanup_removes_only_manifests_at_or_below_sequence() {
async fn committed_cleanup_removes_complete_checkpoint_at_or_below_sequence() {
let root = TempDir::new().expect("test directory");
let disk = disk(&root, "disk").await;
let owner = Uuid::new_v4();
@@ -1377,12 +1455,19 @@ mod tests {
),
"old manifest is gone, so the old payload cannot become visible again"
);
assert!(
matches!(
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
),
"old payload should be reclaimed after its manifest is removed"
);
assert_eq!(
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
.await
.expect("old payload orphan may remain")
.expect("newer payload retained")
.as_ref(),
older
newer
);
let recovered = read_committed(std::slice::from_ref(&disk), 4096)
.await
@@ -1392,6 +1477,40 @@ mod tests {
assert_eq!(recovered.payload(), newer);
}
#[tokio::test]
async fn committed_cleanup_retains_manifest_when_payload_identity_mismatches() {
let root = TempDir::new().expect("test directory");
let disk = disk(&root, "disk").await;
let owner = Uuid::new_v4();
let declared = payload("declared");
let actual = payload("actual");
install(&disk, MANIFEST_PATHS[0], &manifest(owner, 3, &declared)).await;
install(&disk, PAYLOAD_PATHS[0], &actual).await;
assert!(
matches!(
delete_committed_snapshots_through_on(std::slice::from_ref(&disk), owner, 3, 4096).await,
Err(SnapshotError::Corrupt)
),
"cleanup must fail closed when the committed body no longer matches its manifest"
);
assert_eq!(
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
.await
.expect("mismatched manifest retained")
.as_ref(),
manifest(owner, 3, &declared)
);
assert_eq!(
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
.await
.expect("mismatched payload retained")
.as_ref(),
actual
);
}
#[tokio::test]
async fn committed_cleanup_preserves_other_owner_manifests_within_sequence_window() {
let root = TempDir::new().expect("test directory");
@@ -1417,6 +1536,13 @@ mod tests {
),
"the replay owner's manifest is reclaimed"
);
assert!(
matches!(
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
),
"the replay owner's payload is reclaimed"
);
assert_eq!(
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
.await
crates/heal/tests/mrf_pipeline_test.rs
+44 -2
View File
@@ -51,6 +51,8 @@ const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin";
const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin";
const COMMITTED_PAYLOAD_REL: &str = ".heal-mrf-snapshot.0.bin";
const COMMITTED_MANIFEST_REL: &str = ".heal-mrf-commit.0.bin";
const COMMITTED_PAYLOAD_RELS: [&str; 2] = [".heal-mrf-snapshot.0.bin", ".heal-mrf-snapshot.1.bin"];
const COMMITTED_MANIFEST_RELS: [&str; 2] = [".heal-mrf-commit.0.bin", ".heal-mrf-commit.1.bin"];
const COMMITTED_MAGIC: &[u8; 8] = b"RFMRFC01";
const COMMITTED_MANIFEST_LEN: usize = 8 + 1 + 16 + 8 + 8 + 32 + 32;
@@ -235,6 +237,44 @@ fn journal_matches_on_all_disks(disk_paths: &[PathBuf], relative_path: &str, exp
.all(|path| std::fs::read(path.join(META_BUCKET).join(relative_path)).is_ok_and(|actual| actual == expected))
}
fn committed_checkpoint_matches_on_all_disks(disk_paths: &[PathBuf], sequence: u64, expected_payload: &[u8]) -> bool {
disk_paths.iter().all(|path| {
let root = path.join(META_BUCKET);
COMMITTED_PAYLOAD_RELS
.into_iter()
.zip(COMMITTED_MANIFEST_RELS)
.any(|(payload_rel, manifest_rel)| {
let Ok(payload) = std::fs::read(root.join(payload_rel)) else {
return false;
};
if payload != expected_payload {
return false;
}
let Ok(manifest) = std::fs::read(root.join(manifest_rel)) else {
return false;
};
if manifest.len() != COMMITTED_MANIFEST_LEN || &manifest[..8] != COMMITTED_MAGIC || manifest[8] != 1 {
return false;
}
let Ok(recorded_sequence) = <[u8; 8]>::try_from(&manifest[25..33]).map(u64::from_le_bytes) else {
return false;
};
let Ok(recorded_len) = <[u8; 8]>::try_from(&manifest[33..41]).map(u64::from_le_bytes) else {
return false;
};
let Ok(expected_len) = u64::try_from(expected_payload.len()) else {
return false;
};
if recorded_sequence != sequence || recorded_len != expected_len {
return false;
}
let payload_digest: [u8; 32] = Sha256::digest(expected_payload).into();
let manifest_digest: [u8; 32] = Sha256::digest(&manifest[..COMMITTED_MANIFEST_LEN - 32]).into();
payload_digest.as_slice() == &manifest[41..73] && manifest_digest.as_slice() == &manifest[73..]
})
})
}
async fn wait_until<F, Fut>(deadline: Duration, mut probe: F) -> bool
where
F: FnMut() -> Fut,
@@ -629,13 +669,14 @@ fn mrf_successor_flush_child_process_fixture() {
let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
let flushed = wait_until(Duration::from_secs(10), || async {
manager.operations_snapshot().await.queued_by_source.mrf == 1
&& committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
&& journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
&& journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
})
.await;
assert!(
flushed,
"child process must publish the pending successor snapshot before the delete phase"
"child process must publish the committed pending successor before the delete phase"
);
});
std::process::exit(78);
@@ -676,13 +717,14 @@ fn mrf_successor_flush_waiting_child_process_fixture() {
let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
let flushed = wait_until(Duration::from_secs(10), || async {
manager.operations_snapshot().await.queued_by_source.mrf == 1
&& committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
&& journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
&& journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
})
.await;
assert!(
flushed,
"child process must publish the pending successor snapshot before it can be killed"
"child process must publish the committed pending successor before it can be killed"
);
std::fs::write(&ready_path, b"ready").expect("write ready marker");
loop {