mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-16 18:08:21 +00:00
Merge remote-tracking branch 'origin/main' into cxymds/fix-rebalance-multipart-retry
This commit is contained in:
马登山
@@ -37,7 +37,6 @@ hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-filemeta/hotpath-cpu"]
|
||||
hotpath.workspace = true
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
rmp-serde = { workspace = true }
|
||||
async-trait = { workspace = true }
|
||||
rustfs-filemeta = { workspace = true }
|
||||
|
||||
[lib]
|
||||
|
||||
@@ -846,8 +846,15 @@ impl DataUsageEntry {
|
||||
}
|
||||
}
|
||||
|
||||
/// Data usage cache info
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
|
||||
/// Read-only projection of the scanner's `.usage-cache.bin` info block.
|
||||
///
|
||||
/// The canonical wire format is written by the hand-written map-encoded
|
||||
/// `Serialize` on the scanner-side `DataUsageCacheInfo`
|
||||
/// (`crates/scanner/src/data_usage_define.rs`), which carries 16 fields.
|
||||
/// This type decodes only the shared subset and is deliberately not
|
||||
/// `Serialize`: a derived (array) encoding of this 6-field subset would
|
||||
/// corrupt the cache for scanner readers, so no write path may exist here.
|
||||
#[derive(Clone, Debug, Default, Deserialize)]
|
||||
pub struct DataUsageCacheInfo {
|
||||
pub name: String,
|
||||
pub next_cycle: u64,
|
||||
@@ -863,8 +870,12 @@ pub struct DataUsageCacheInfo {
|
||||
pub snapshot_complete: bool,
|
||||
}
|
||||
|
||||
/// Data usage cache
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
|
||||
/// Read-only projection of a scanner-written `.usage-cache.bin` file.
|
||||
///
|
||||
/// The scanner-side `DataUsageCache` (`crates/scanner/src/data_usage_define.rs`)
|
||||
/// owns the persisted format; this type only decodes it (see
|
||||
/// [`DataUsageCacheInfo`]) and must never grow a serialization path.
|
||||
#[derive(Clone, Debug, Default, Deserialize)]
|
||||
pub struct DataUsageCache {
|
||||
pub info: DataUsageCacheInfo,
|
||||
pub cache: HashMap<String, DataUsageEntry>,
|
||||
@@ -1186,31 +1197,10 @@ impl DataUsageCache {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn marshal_msg(&self) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let mut buf = Vec::new();
|
||||
self.serialize(&mut rmp_serde::Serializer::new(&mut buf))?;
|
||||
Ok(buf)
|
||||
}
|
||||
|
||||
pub fn unmarshal(buf: &[u8]) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let t: Self = rmp_serde::from_slice(buf)?;
|
||||
Ok(t)
|
||||
}
|
||||
|
||||
// Note: load and save methods are storage-specific and should be implemented
|
||||
// in the ecstore crate where storage access is available
|
||||
}
|
||||
|
||||
/// Trait for storage-specific operations on DataUsageCache
|
||||
#[async_trait::async_trait]
|
||||
pub trait DataUsageCacheStorage {
|
||||
/// Load data usage cache from backend storage
|
||||
async fn load(store: &dyn std::any::Any, name: &str) -> Result<Self, Box<dyn std::error::Error + Send + Sync>>
|
||||
where
|
||||
Self: Sized;
|
||||
|
||||
/// Save data usage cache to backend storage
|
||||
async fn save(&self, name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
|
||||
}
|
||||
|
||||
// Helper structs and functions for cache operations
|
||||
@@ -1832,6 +1822,82 @@ mod tests {
|
||||
assert!(decoded.all_tier_stats.is_none());
|
||||
}
|
||||
|
||||
/// Scanner-written `.usage-cache.bin` bytes: a 2-element array of the
|
||||
/// canonical 16-field map-encoded info block and one map-encoded entry.
|
||||
/// Captured from the canonical writer's `marshal_msg` — see
|
||||
/// `usage_cache_wire_format_is_pinned` in
|
||||
/// `crates/scanner/src/data_usage_define.rs`, which pins these exact
|
||||
/// bytes and documents regeneration. Hardcoded here because a
|
||||
/// dev-dependency on rustfs-scanner would pull the whole ecstore tree
|
||||
/// into this crate's test build, and a fixture generated at test runtime
|
||||
/// could not detect writer drift anyway.
|
||||
const SCANNER_USAGE_CACHE_WIRE_FIXTURE: &[u8] = &[
|
||||
0x92, 0xde, 0x00, 0x10, 0xa4, 0x6e, 0x61, 0x6d, 0x65, 0xab, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65,
|
||||
0x74, 0xaa, 0x6e, 0x65, 0x78, 0x74, 0x5f, 0x63, 0x79, 0x63, 0x6c, 0x65, 0x07, 0xac, 0x6c, 0x65, 0x61, 0x64, 0x65, 0x72,
|
||||
0x5f, 0x65, 0x70, 0x6f, 0x63, 0x68, 0x09, 0xab, 0x6c, 0x61, 0x73, 0x74, 0x5f, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x92,
|
||||
0xce, 0x65, 0x53, 0xf1, 0x00, 0x00, 0xac, 0x73, 0x6b, 0x69, 0x70, 0x5f, 0x68, 0x65, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0xc3,
|
||||
0xa9, 0x6c, 0x69, 0x66, 0x65, 0x63, 0x79, 0x63, 0x6c, 0x65, 0xc0, 0xab, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74,
|
||||
0x69, 0x6f, 0x6e, 0xc0, 0xae, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x81,
|
||||
0xb0, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x2f, 0x6c, 0x6f, 0x73, 0x74, 0x0b, 0xb1, 0x73,
|
||||
0x63, 0x61, 0x6e, 0x5f, 0x72, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x5f, 0x61, 0x66, 0x74, 0x65, 0x72, 0xb2, 0x77, 0x69, 0x72,
|
||||
0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x2f, 0x72, 0x65, 0x73, 0x75, 0x6d, 0x65, 0xaf, 0x73, 0x63, 0x61, 0x6e,
|
||||
0x5f, 0x63, 0x68, 0x65, 0x63, 0x6b, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0xc0, 0xad, 0x70, 0x65, 0x6e, 0x64, 0x69, 0x6e, 0x67,
|
||||
0x5f, 0x68, 0x65, 0x61, 0x6c, 0x73, 0x91, 0x9a, 0xa6, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0xab, 0x77, 0x69, 0x72, 0x65,
|
||||
0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0xa6, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x6e, 0xc0, 0x01, 0x64, 0xcc, 0xc8, 0x03,
|
||||
0xa8, 0x64, 0x65, 0x66, 0x65, 0x72, 0x72, 0x65, 0x64, 0xa6, 0x62, 0x75, 0x64, 0x67, 0x65, 0x74, 0xab, 0x6f, 0x62, 0x6a,
|
||||
0x65, 0x63, 0x74, 0x5f, 0x6c, 0x6f, 0x63, 0x6b, 0xc0, 0xa6, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x92, 0x01, 0x02, 0xb1,
|
||||
0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x5f, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0xc3, 0xb0, 0x73,
|
||||
0x63, 0x61, 0x6e, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x5f, 0x64, 0x69, 0x67, 0x65, 0x73, 0x74, 0xdc, 0x00, 0x20, 0x03, 0x03,
|
||||
0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03,
|
||||
0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0xb0, 0x63, 0x61, 0x63, 0x68, 0x65, 0x5f, 0x6b, 0x65, 0x79,
|
||||
0x5f, 0x66, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x01, 0x81, 0xab, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65,
|
||||
0x74, 0x8b, 0xa8, 0x63, 0x68, 0x69, 0x6c, 0x64, 0x72, 0x65, 0x6e, 0x90, 0xa4, 0x73, 0x69, 0x7a, 0x65, 0xcd, 0x10, 0x00,
|
||||
0xa7, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x03, 0xa8, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x05, 0xae,
|
||||
0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x5f, 0x6d, 0x61, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x01, 0xa9, 0x6f, 0x62, 0x6a, 0x5f,
|
||||
0x73, 0x69, 0x7a, 0x65, 0x73, 0x9b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xac, 0x6f, 0x62,
|
||||
0x6a, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x97, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xb1, 0x72,
|
||||
0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0xc0, 0xa9, 0x63, 0x6f,
|
||||
0x6d, 0x70, 0x61, 0x63, 0x74, 0x65, 0x64, 0xc3, 0xae, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x6f, 0x62, 0x6a, 0x65,
|
||||
0x63, 0x74, 0x73, 0x02, 0xae, 0x61, 0x6c, 0x6c, 0x5f, 0x74, 0x69, 0x65, 0x72, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0x91,
|
||||
0x81, 0xa4, 0x57, 0x41, 0x52, 0x4d, 0x93, 0xcd, 0x08, 0x00, 0x02, 0x01,
|
||||
];
|
||||
|
||||
#[test]
|
||||
fn thin_usage_cache_decodes_scanner_wire_fixture() {
|
||||
let decoded =
|
||||
DataUsageCache::unmarshal(SCANNER_USAGE_CACHE_WIRE_FIXTURE).expect("thin projection decodes a scanner-written cache");
|
||||
|
||||
// The six fields shared with the scanner's 16-field info block; the
|
||||
// remaining ten (lifecycle, replication, checkpoint, heals, ...) must
|
||||
// be skipped, not error.
|
||||
assert_eq!(decoded.info.name, "wire-bucket");
|
||||
assert_eq!(decoded.info.next_cycle, 7);
|
||||
assert_eq!(
|
||||
decoded.info.last_update,
|
||||
Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000))
|
||||
);
|
||||
assert!(decoded.info.skip_healing);
|
||||
assert_eq!(decoded.info.failed_objects.get("wire-bucket/lost"), Some(&11));
|
||||
assert!(decoded.info.snapshot_complete);
|
||||
|
||||
// Entries use the shared canonical map-encoded type end to end.
|
||||
let entry = decoded.cache.get("wire-bucket").expect("fixture entry decodes");
|
||||
assert_eq!(entry.size, 4096);
|
||||
assert_eq!(entry.objects, 3);
|
||||
assert_eq!(entry.versions, 5);
|
||||
assert_eq!(entry.delete_markers, 1);
|
||||
assert!(entry.compacted);
|
||||
assert_eq!(entry.failed_objects, 2);
|
||||
assert_eq!(
|
||||
entry.all_tier_stats.as_ref().and_then(|tiers| tiers.tiers.get("WARM")),
|
||||
Some(&TierStats {
|
||||
total_size: 2048,
|
||||
num_versions: 2,
|
||||
num_objects: 1,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hash_path_uses_portable_slash_semantics() {
|
||||
for (input, expected) in [
|
||||
|
||||
@@ -62,8 +62,8 @@ pub(crate) struct VersionShardCensus {
|
||||
pub data_dir: Option<String>,
|
||||
pub erasure_index: Option<usize>,
|
||||
pub expected_part_numbers: BTreeSet<usize>,
|
||||
pub present_part_numbers: BTreeSet<usize>,
|
||||
pub present_part_fingerprints: BTreeMap<usize, PartShardFingerprint>,
|
||||
pub inline_data_fingerprint: Option<PartShardFingerprint>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
@@ -74,7 +74,12 @@ pub(crate) struct PartShardFingerprint {
|
||||
|
||||
impl VersionShardCensus {
|
||||
pub(crate) fn is_complete(&self) -> bool {
|
||||
self.has_xl_meta && self.expected_part_numbers == self.present_part_numbers
|
||||
self.has_xl_meta
|
||||
&& self.expected_part_numbers.len() == self.present_part_fingerprints.len()
|
||||
&& self
|
||||
.expected_part_numbers
|
||||
.iter()
|
||||
.all(|part_number| self.present_part_fingerprints.contains_key(part_number))
|
||||
}
|
||||
|
||||
pub(crate) fn matches_manifest(&self, manifest: &Self) -> bool {
|
||||
@@ -85,6 +90,7 @@ impl VersionShardCensus {
|
||||
&& self.erasure_index == manifest.erasure_index
|
||||
&& self.expected_part_numbers == manifest.expected_part_numbers
|
||||
&& self.present_part_fingerprints == manifest.present_part_fingerprints
|
||||
&& self.inline_data_fingerprint == manifest.inline_data_fingerprint
|
||||
}
|
||||
}
|
||||
|
||||
@@ -93,6 +99,13 @@ fn sha256_hex(data: &[u8]) -> String {
|
||||
digest.iter().map(|byte| format!("{byte:02x}")).collect()
|
||||
}
|
||||
|
||||
fn shard_fingerprint(data: &[u8]) -> ChaosResult<PartShardFingerprint> {
|
||||
Ok(PartShardFingerprint {
|
||||
size: u64::try_from(data.len())?,
|
||||
sha256: sha256_hex(data),
|
||||
})
|
||||
}
|
||||
|
||||
/// Single-node RustFS server with `disk_count` local volume directories that
|
||||
/// can be faulted individually while the server is running.
|
||||
pub struct DiskFaultHarness {
|
||||
@@ -301,8 +314,8 @@ pub(crate) fn census_object_version_on_disk(
|
||||
data_dir: None,
|
||||
erasure_index: None,
|
||||
expected_part_numbers: BTreeSet::new(),
|
||||
present_part_numbers: BTreeSet::new(),
|
||||
present_part_fingerprints: BTreeMap::new(),
|
||||
inline_data_fingerprint: None,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -315,10 +328,10 @@ pub(crate) fn census_object_version_on_disk(
|
||||
};
|
||||
let data_dir = file_info.data_dir.map(|id| id.to_string());
|
||||
let erasure_index = Some(file_info.erasure.index);
|
||||
let inline_data_fingerprint = file_info.data.as_deref().map(shard_fingerprint).transpose()?;
|
||||
let part_dir = data_dir.as_ref().map_or_else(|| object_dir.clone(), |id| object_dir.join(id));
|
||||
let (present_part_numbers, present_part_fingerprints) = match std::fs::read_dir(&part_dir) {
|
||||
let present_part_fingerprints = match std::fs::read_dir(&part_dir) {
|
||||
Ok(entries) => {
|
||||
let mut numbers = BTreeSet::new();
|
||||
let mut fingerprints = BTreeMap::new();
|
||||
for entry in entries {
|
||||
let entry = entry?;
|
||||
@@ -333,19 +346,12 @@ pub(crate) fn census_object_version_on_disk(
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
numbers.insert(part_number);
|
||||
let data = std::fs::read(entry.path())?;
|
||||
fingerprints.insert(
|
||||
part_number,
|
||||
PartShardFingerprint {
|
||||
size: u64::try_from(data.len())?,
|
||||
sha256: sha256_hex(&data),
|
||||
},
|
||||
);
|
||||
fingerprints.insert(part_number, shard_fingerprint(&data)?);
|
||||
}
|
||||
(numbers, fingerprints)
|
||||
fingerprints
|
||||
}
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => (BTreeSet::new(), BTreeMap::new()),
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => BTreeMap::new(),
|
||||
Err(error) => return Err(error.into()),
|
||||
};
|
||||
|
||||
@@ -355,8 +361,8 @@ pub(crate) fn census_object_version_on_disk(
|
||||
data_dir,
|
||||
erasure_index,
|
||||
expected_part_numbers,
|
||||
present_part_numbers,
|
||||
present_part_fingerprints,
|
||||
inline_data_fingerprint,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -396,3 +402,43 @@ pub async fn signed_admin_post(url: &str, body: Option<&str>, access_key: &str,
|
||||
|
||||
Ok(body)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn complete_census() -> VersionShardCensus {
|
||||
VersionShardCensus {
|
||||
version_id: Some("version".to_string()),
|
||||
has_xl_meta: true,
|
||||
data_dir: Some("data-dir".to_string()),
|
||||
erasure_index: Some(3),
|
||||
expected_part_numbers: BTreeSet::from([1]),
|
||||
present_part_fingerprints: BTreeMap::from([(1, shard_fingerprint(b"part").unwrap())]),
|
||||
inline_data_fingerprint: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shard_fingerprint_uses_physical_length_and_sha256() {
|
||||
assert_eq!(
|
||||
shard_fingerprint(b"abc").unwrap(),
|
||||
PartShardFingerprint {
|
||||
size: 3,
|
||||
sha256: "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad".to_string(),
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn manifest_requires_matching_inline_payload() {
|
||||
let mut expected = complete_census();
|
||||
expected.expected_part_numbers.clear();
|
||||
expected.present_part_fingerprints.clear();
|
||||
expected.inline_data_fingerprint = Some(shard_fingerprint(b"expected").unwrap());
|
||||
let mut changed = expected.clone();
|
||||
changed.inline_data_fingerprint = Some(shard_fingerprint(b"changed").unwrap());
|
||||
assert!(expected.matches_manifest(&expected));
|
||||
assert!(!changed.matches_manifest(&expected));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -349,11 +349,32 @@ mod tests {
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
let first_inline = client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key("versions/inline.bin")
|
||||
.body(ByteStream::from(payload(8 * 1024, 40)))
|
||||
.send()
|
||||
.await?;
|
||||
let first_inline_version = first_inline
|
||||
.version_id()
|
||||
.ok_or("first inline PUT did not return a version ID")?;
|
||||
let second_inline = client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key("versions/inline.bin")
|
||||
.body(ByteStream::from(payload(8 * 1024, 41)))
|
||||
.send()
|
||||
.await?;
|
||||
let second_inline_version = second_inline
|
||||
.version_id()
|
||||
.ok_or("second inline PUT did not return a version ID")?;
|
||||
|
||||
let first = client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.body(ByteStream::from(payload(256 * 1024, 41)))
|
||||
.body(ByteStream::from(payload(128 * 1024, 41)))
|
||||
.send()
|
||||
.await?;
|
||||
let first_version = first.version_id().ok_or("first PUT did not return a version ID")?;
|
||||
@@ -361,16 +382,36 @@ mod tests {
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.body(ByteStream::from(payload(256 * 1024, 42)))
|
||||
.body(ByteStream::from(payload(3 * 1024 * 1024, 42)))
|
||||
.send()
|
||||
.await?;
|
||||
let second_version = second.version_id().ok_or("second PUT did not return a version ID")?;
|
||||
let delete = client.delete_object().bucket(bucket).key(key).send().await?;
|
||||
let delete_version = delete.version_id().ok_or("delete marker did not return a version ID")?;
|
||||
|
||||
let first_inline_census = harness.census_object_version(0, bucket, "versions/inline.bin", Some(first_inline_version))?;
|
||||
let second_inline_census =
|
||||
harness.census_object_version(0, bucket, "versions/inline.bin", Some(second_inline_version))?;
|
||||
let first_census = harness.census_object_version(0, bucket, key, Some(first_version))?;
|
||||
let first_other_disk_census = harness.census_object_version(1, bucket, key, Some(first_version))?;
|
||||
let second_census = harness.census_object_version(0, bucket, key, Some(second_version))?;
|
||||
let delete_census = harness.census_object_version(0, bucket, key, Some(delete_version))?;
|
||||
assert!(
|
||||
first_inline_census.is_complete() && second_inline_census.is_complete(),
|
||||
"inline version physical census is incomplete: first={first_inline_census:?} second={second_inline_census:?}"
|
||||
);
|
||||
assert!(
|
||||
first_inline_census.present_part_fingerprints.is_empty() && second_inline_census.present_part_fingerprints.is_empty(),
|
||||
"inline versions must not select external shard files: first={first_inline_census:?} second={second_inline_census:?}"
|
||||
);
|
||||
assert!(
|
||||
first_inline_census.inline_data_fingerprint.is_some() && second_inline_census.inline_data_fingerprint.is_some(),
|
||||
"inline versions must fingerprint payload bytes stored in xl.meta"
|
||||
);
|
||||
assert_ne!(
|
||||
first_inline_census.inline_data_fingerprint, second_inline_census.inline_data_fingerprint,
|
||||
"same-size inline versions with different payloads must retain distinct xl.meta fingerprints"
|
||||
);
|
||||
assert!(
|
||||
first_census.is_complete(),
|
||||
"first version physical census is incomplete: {first_census:?}"
|
||||
@@ -379,6 +420,14 @@ mod tests {
|
||||
second_census.is_complete(),
|
||||
"second version physical census is incomplete: {second_census:?}"
|
||||
);
|
||||
assert!(
|
||||
first_other_disk_census.is_complete(),
|
||||
"first version physical census on the second disk is incomplete: {first_other_disk_census:?}"
|
||||
);
|
||||
assert_ne!(
|
||||
first_census.erasure_index, first_other_disk_census.erasure_index,
|
||||
"physical census must preserve each disk's erasure index"
|
||||
);
|
||||
assert_ne!(
|
||||
first_census.data_dir, second_census.data_dir,
|
||||
"distinct object versions must select distinct physical data directories"
|
||||
@@ -387,6 +436,24 @@ mod tests {
|
||||
first_census.expected_part_numbers, second_census.expected_part_numbers,
|
||||
"same single-part shape should expose the same part numbers"
|
||||
);
|
||||
let first_part = first_census
|
||||
.present_part_fingerprints
|
||||
.values()
|
||||
.next()
|
||||
.ok_or("first version did not expose a physical part fingerprint")?;
|
||||
let second_part = second_census
|
||||
.present_part_fingerprints
|
||||
.values()
|
||||
.next()
|
||||
.ok_or("second version did not expose a physical part fingerprint")?;
|
||||
assert_ne!(
|
||||
first_part.size, second_part.size,
|
||||
"different shard lengths must retain their physical sizes"
|
||||
);
|
||||
assert_ne!(
|
||||
first_part.sha256, second_part.sha256,
|
||||
"different shard contents must retain their physical hashes"
|
||||
);
|
||||
assert!(
|
||||
delete_census.is_complete(),
|
||||
"delete marker physical census is incomplete: {delete_census:?}"
|
||||
@@ -396,7 +463,7 @@ mod tests {
|
||||
"delete marker must not declare object shards: {delete_census:?}"
|
||||
);
|
||||
assert!(
|
||||
delete_census.present_part_numbers.is_empty(),
|
||||
delete_census.present_part_fingerprints.is_empty(),
|
||||
"delete marker must not select stale object shards: {delete_census:?}"
|
||||
);
|
||||
Ok(())
|
||||
|
||||
@@ -59,6 +59,13 @@ mod tests {
|
||||
expected: VersionShardCensus,
|
||||
}
|
||||
|
||||
#[derive(Debug, Eq, PartialEq)]
|
||||
enum CompletionSample {
|
||||
Pending,
|
||||
Ready,
|
||||
CompletedWithIncomplete(BTreeSet<String>),
|
||||
}
|
||||
|
||||
struct MountNamespaceGuard {
|
||||
mounts: Vec<PathBuf>,
|
||||
}
|
||||
@@ -75,7 +82,7 @@ mod tests {
|
||||
impl MountNamespaceGuard {
|
||||
fn new() -> Result<Self, Box<dyn Error + Send + Sync>> {
|
||||
verify_isolated_mount_namespace()?;
|
||||
run_command("mount", ["--make-rprivate", "/"])?;
|
||||
run_command("mount", &["--make-rprivate", "/"])?;
|
||||
Ok(Self { mounts: Vec::new() })
|
||||
}
|
||||
|
||||
@@ -108,15 +115,15 @@ mod tests {
|
||||
return Err("losetup --find --show returned an empty loop device".into());
|
||||
}
|
||||
|
||||
run_command_dynamic("mkfs.ext4", &["-F", &loop_device])?;
|
||||
run_command("mkfs.ext4", &["-F", &loop_device])?;
|
||||
let sectors = run_command_stdout("blockdev", &["--getsz", &loop_device])?;
|
||||
let dm_name = format!("rustfs_e2e_{label}_{}", std::process::id());
|
||||
let table = format!("0 {sectors} linear {loop_device} 0");
|
||||
let mapper = format!("/dev/mapper/{dm_name}");
|
||||
run_command_dynamic("dmsetup", &["create", &dm_name, "--table", &table])?;
|
||||
run_command("dmsetup", &["create", &dm_name, "--table", &table])?;
|
||||
|
||||
let target_arg = path_to_string(target, "faultable mount target")?;
|
||||
run_command_dynamic("mount", &[&mapper, &target_arg])?;
|
||||
run_command("mount", &[&mapper, &target_arg])?;
|
||||
|
||||
Ok(Self {
|
||||
target: target.to_path_buf(),
|
||||
@@ -131,17 +138,17 @@ mod tests {
|
||||
fn make_unavailable(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let sectors = run_command_stdout("blockdev", &["--getsz", &self.loop_device])?;
|
||||
let error_table = format!("0 {sectors} error");
|
||||
run_command_dynamic("dmsetup", &["suspend", &self.dm_name])?;
|
||||
run_command_dynamic("dmsetup", &["load", &self.dm_name, "--table", &error_table])?;
|
||||
run_command_dynamic("dmsetup", &["resume", &self.dm_name])
|
||||
run_command("dmsetup", &["suspend", &self.dm_name])?;
|
||||
run_command("dmsetup", &["load", &self.dm_name, "--table", &error_table])?;
|
||||
run_command("dmsetup", &["resume", &self.dm_name])
|
||||
}
|
||||
|
||||
fn restore_available(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let sectors = run_command_stdout("blockdev", &["--getsz", &self.loop_device])?;
|
||||
let linear_table = format!("0 {sectors} linear {} 0", self.loop_device);
|
||||
run_command_dynamic("dmsetup", &["suspend", &self.dm_name])?;
|
||||
run_command_dynamic("dmsetup", &["load", &self.dm_name, "--table", &linear_table])?;
|
||||
run_command_dynamic("dmsetup", &["resume", &self.dm_name])
|
||||
run_command("dmsetup", &["suspend", &self.dm_name])?;
|
||||
run_command("dmsetup", &["load", &self.dm_name, "--table", &linear_table])?;
|
||||
run_command("dmsetup", &["resume", &self.dm_name])
|
||||
}
|
||||
|
||||
fn cleanup(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
@@ -157,14 +164,14 @@ mod tests {
|
||||
}
|
||||
}
|
||||
if self.dm_created {
|
||||
if let Err(error) = run_command_dynamic("dmsetup", &["remove", "-f", &self.dm_name]) {
|
||||
if let Err(error) = run_command("dmsetup", &["remove", "-f", &self.dm_name]) {
|
||||
first_error.get_or_insert(error);
|
||||
} else {
|
||||
self.dm_created = false;
|
||||
}
|
||||
}
|
||||
if !self.loop_device.is_empty() {
|
||||
if let Err(error) = run_command_dynamic("losetup", &["-d", &self.loop_device]) {
|
||||
if let Err(error) = run_command("losetup", &["-d", &self.loop_device]) {
|
||||
first_error.get_or_insert(error);
|
||||
} else {
|
||||
self.loop_device.clear();
|
||||
@@ -188,14 +195,10 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn run_command<const N: usize>(program: &str, args: [&str; N]) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
run_command_dynamic(program, &args)
|
||||
}
|
||||
|
||||
fn run_command_dynamic(program: &str, args: &[&str]) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
fn checked_command_output(program: &str, args: &[&str]) -> Result<std::process::Output, Box<dyn Error + Send + Sync>> {
|
||||
let output = Command::new(program).args(args).output()?;
|
||||
if output.status.success() {
|
||||
return Ok(());
|
||||
return Ok(output);
|
||||
}
|
||||
Err(format!(
|
||||
"{program} {} failed with status {}: stdout={} stderr={}",
|
||||
@@ -207,19 +210,14 @@ mod tests {
|
||||
.into())
|
||||
}
|
||||
|
||||
fn run_command(program: &str, args: &[&str]) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
checked_command_output(program, args).map(drop)
|
||||
}
|
||||
|
||||
fn run_command_stdout(program: &str, args: &[&str]) -> Result<String, Box<dyn Error + Send + Sync>> {
|
||||
let output = Command::new(program).args(args).output()?;
|
||||
if output.status.success() {
|
||||
return Ok(String::from_utf8_lossy(&output.stdout).trim().to_string());
|
||||
}
|
||||
Err(format!(
|
||||
"{program} {} failed with status {}: stdout={} stderr={}",
|
||||
args.join(" "),
|
||||
output.status,
|
||||
String::from_utf8_lossy(&output.stdout),
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
)
|
||||
.into())
|
||||
Ok(String::from_utf8(checked_command_output(program, args)?.stdout)?
|
||||
.trim()
|
||||
.to_string())
|
||||
}
|
||||
|
||||
fn path_to_string(path: &Path, label: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
|
||||
@@ -260,14 +258,14 @@ mod tests {
|
||||
let target = target
|
||||
.to_str()
|
||||
.ok_or_else(|| format!("tmpfs target path is not UTF-8: {target:?}"))?;
|
||||
run_command("mount", ["-t", "tmpfs", "-o", MOUNT_SIZE, label, target])
|
||||
run_command("mount", &["-t", "tmpfs", "-o", MOUNT_SIZE, label, target])
|
||||
}
|
||||
|
||||
fn detach_mount(target: &Path) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let target = target
|
||||
.to_str()
|
||||
.ok_or_else(|| format!("umount target path is not UTF-8: {target:?}"))?;
|
||||
run_command("umount", [target])
|
||||
run_command("umount", &[target])
|
||||
}
|
||||
|
||||
fn privileged_run_enabled() -> Result<bool, Box<dyn Error + Send + Sync>> {
|
||||
@@ -423,6 +421,10 @@ mod tests {
|
||||
.await?;
|
||||
versions.push((versioned_bucket, "history/object.bin", deleted.version_id().map(str::to_owned), None));
|
||||
|
||||
let (version_id, body_sha256) =
|
||||
put_object_version(client, versioned_bucket, "history/inline.bin", payload(8 * 1024, 9)).await?;
|
||||
versions.push((versioned_bucket, "history/inline.bin", version_id, Some(body_sha256)));
|
||||
|
||||
let (version_id, body_sha256) = put_multipart_version(
|
||||
client,
|
||||
versioned_bucket,
|
||||
@@ -436,7 +438,7 @@ mod tests {
|
||||
put_object_version(client, null_bucket, "null/current.bin", payload(512 * 1024, 8)).await?;
|
||||
versions.push((null_bucket, "null/current.bin", version_id, Some(body_sha256)));
|
||||
|
||||
versions
|
||||
let versions = versions
|
||||
.into_iter()
|
||||
.map(|(bucket, key, version_id, body_sha256)| {
|
||||
let expected = census_object_version_on_disk(target_disk, bucket, key, version_id.as_deref())?;
|
||||
@@ -451,7 +453,15 @@ mod tests {
|
||||
expected,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
.collect::<Result<Vec<_>, Box<dyn Error + Send + Sync>>>()?;
|
||||
let inline = versions
|
||||
.iter()
|
||||
.find(|version| version.key == "history/inline.bin")
|
||||
.ok_or("inline replacement baseline was not recorded")?;
|
||||
if inline.expected.inline_data_fingerprint.is_none() || !inline.expected.present_part_fingerprints.is_empty() {
|
||||
return Err(format!("inline replacement baseline lacks xl.meta payload evidence: {:?}", inline.expected).into());
|
||||
}
|
||||
Ok(versions)
|
||||
}
|
||||
|
||||
async fn verify_bodies(client: &Client, versions: &[BaselineVersion]) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
@@ -597,6 +607,33 @@ mod tests {
|
||||
Ok(String::from_utf8_lossy(&log[start..]).into_owned())
|
||||
}
|
||||
|
||||
fn live_disk_loss_scan_completed(log: &str, target_disk: &Path) -> bool {
|
||||
let target = target_disk.to_string_lossy();
|
||||
let mut saw_live_loss = false;
|
||||
for line in log.lines() {
|
||||
if line.contains("Heal auto-scan disk inspection failed")
|
||||
&& line.contains("check_failed")
|
||||
&& line.contains(target.as_ref())
|
||||
{
|
||||
saw_live_loss = true;
|
||||
continue;
|
||||
}
|
||||
if saw_live_loss && (line.contains("Heal auto disk scanner idle") || line.contains("Heal auto-scan cycle completed"))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn live_disk_loss_scan_completed_from_path(
|
||||
log_path: &Path,
|
||||
start_offset: u64,
|
||||
target_disk: &Path,
|
||||
) -> Result<bool, Box<dyn Error + Send + Sync>> {
|
||||
Ok(live_disk_loss_scan_completed(&log_from_offset(log_path, start_offset)?, target_disk))
|
||||
}
|
||||
|
||||
async fn wait_for_live_disk_loss_observation(
|
||||
log_path: &Path,
|
||||
target_disk: &Path,
|
||||
@@ -605,20 +642,12 @@ mod tests {
|
||||
) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let deadline = Instant::now() + Duration::from_secs(timeout_secs);
|
||||
let mut tick = interval(Duration::from_secs(1));
|
||||
let target = target_disk.to_string_lossy();
|
||||
loop {
|
||||
let log = log_from_offset(log_path, start_offset)?;
|
||||
let mut saw_live_loss = false;
|
||||
for line in log.lines() {
|
||||
if line.contains("check_failed") && line.contains(target.as_ref()) {
|
||||
saw_live_loss = true;
|
||||
continue;
|
||||
}
|
||||
if saw_live_loss && line.contains("Heal auto disk scanner idle") {
|
||||
return Ok(());
|
||||
}
|
||||
if live_disk_loss_scan_completed_from_path(log_path, start_offset, target_disk)? {
|
||||
return Ok(());
|
||||
}
|
||||
if Instant::now() >= deadline {
|
||||
let log = log_from_offset(log_path, start_offset)?;
|
||||
return Err(format!(
|
||||
"scanner did not finish a live target-loss scan for {target_disk:?} within {timeout_secs}s; log tail:\n{}",
|
||||
log_tail(&log)
|
||||
@@ -629,14 +658,10 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn require_definitive_replacement_status(
|
||||
status: &serde_json::Value,
|
||||
context: &str,
|
||||
) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
if status["cluster"]["definitive"].as_bool().unwrap_or(false) {
|
||||
return Ok(());
|
||||
}
|
||||
Err(format!("{context} requires a definitive cluster replacement status: {status}").into())
|
||||
fn cluster_status_is_definitive(status: &serde_json::Value) -> Result<bool, Box<dyn Error + Send + Sync>> {
|
||||
status["cluster"]["definitive"]
|
||||
.as_bool()
|
||||
.ok_or_else(|| format!("replacement recovery status omitted cluster.definitive: {status}").into())
|
||||
}
|
||||
|
||||
async fn assert_no_replacement_status_records(
|
||||
@@ -644,8 +669,17 @@ mod tests {
|
||||
target_disk: &Path,
|
||||
) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let status = replacement_status(cluster).await?;
|
||||
require_definitive_replacement_status(&status, "live missing replacement status check")?;
|
||||
let states = target_record_states(&status, target_disk);
|
||||
assert_no_replacement_status_records_in_status(&status, target_disk)
|
||||
}
|
||||
|
||||
fn assert_no_replacement_status_records_in_status(
|
||||
status: &serde_json::Value,
|
||||
target_disk: &Path,
|
||||
) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
if !cluster_status_is_definitive(status)? {
|
||||
return Err(format!("live missing replacement status check requires a definitive cluster status: {status}").into());
|
||||
}
|
||||
let states = target_record_states(status, target_disk);
|
||||
if states.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
@@ -655,11 +689,6 @@ mod tests {
|
||||
.into())
|
||||
}
|
||||
|
||||
fn target_record_has_state(status: &serde_json::Value, target_disk: &Path, states: &[&str]) -> bool {
|
||||
let present_states = target_record_states(status, target_disk);
|
||||
states.iter().any(|state| present_states.contains(*state))
|
||||
}
|
||||
|
||||
fn target_record_states(status: &serde_json::Value, target_disk: &Path) -> BTreeSet<String> {
|
||||
let target = target_disk.to_string_lossy();
|
||||
status["cluster"]["records"]
|
||||
@@ -694,6 +723,57 @@ mod tests {
|
||||
Ok(missing)
|
||||
}
|
||||
|
||||
fn replacement_completion_state(
|
||||
status: &serde_json::Value,
|
||||
target_disk: &Path,
|
||||
missing: BTreeSet<String>,
|
||||
) -> Result<CompletionSample, Box<dyn Error + Send + Sync>> {
|
||||
if !cluster_status_is_definitive(status)? {
|
||||
return Ok(CompletionSample::Pending);
|
||||
}
|
||||
if !target_record_states(status, target_disk).contains("completed") {
|
||||
return Ok(CompletionSample::Pending);
|
||||
}
|
||||
if missing.is_empty() {
|
||||
return Ok(CompletionSample::Ready);
|
||||
}
|
||||
Ok(CompletionSample::CompletedWithIncomplete(missing))
|
||||
}
|
||||
|
||||
async fn sample_replacement_completion<C, S, F>(
|
||||
target_disk: &Path,
|
||||
census: C,
|
||||
status: S,
|
||||
) -> Result<CompletionSample, Box<dyn Error + Send + Sync>>
|
||||
where
|
||||
C: FnOnce() -> Result<BTreeSet<String>, Box<dyn Error + Send + Sync>>,
|
||||
S: FnOnce() -> F,
|
||||
F: std::future::Future<Output = Result<serde_json::Value, Box<dyn Error + Send + Sync>>>,
|
||||
{
|
||||
let missing = census()?;
|
||||
let status = status().await?;
|
||||
replacement_completion_state(&status, target_disk, missing)
|
||||
}
|
||||
|
||||
async fn confirm_replacement_completion<C, S, F>(
|
||||
target_disk: &Path,
|
||||
mut census: C,
|
||||
mut status: S,
|
||||
) -> Result<CompletionSample, Box<dyn Error + Send + Sync>>
|
||||
where
|
||||
C: FnMut() -> Result<BTreeSet<String>, Box<dyn Error + Send + Sync>>,
|
||||
S: FnMut() -> F,
|
||||
F: std::future::Future<Output = Result<serde_json::Value, Box<dyn Error + Send + Sync>>>,
|
||||
{
|
||||
let missing = census()?;
|
||||
let status = status().await?;
|
||||
let first = replacement_completion_state(&status, target_disk, missing)?;
|
||||
if matches!(first, CompletionSample::CompletedWithIncomplete(_)) {
|
||||
return replacement_completion_state(&status, target_disk, census()?);
|
||||
}
|
||||
Ok(first)
|
||||
}
|
||||
|
||||
async fn wait_for_completed_replacement_with_census(
|
||||
cluster: &RustFSTestClusterEnvironment,
|
||||
target_disk: &Path,
|
||||
@@ -703,28 +783,25 @@ mod tests {
|
||||
let deadline = Instant::now() + Duration::from_secs(timeout_secs);
|
||||
let mut tick = interval(Duration::from_secs(1));
|
||||
loop {
|
||||
let status = replacement_status(cluster).await?;
|
||||
let missing = incomplete_versions(target_disk, versions)?;
|
||||
if require_definitive_replacement_status(&status, "replacement completion poll").is_err() {
|
||||
if Instant::now() >= deadline {
|
||||
match confirm_replacement_completion(
|
||||
target_disk,
|
||||
|| incomplete_versions(target_disk, versions),
|
||||
|| replacement_status(cluster),
|
||||
)
|
||||
.await?
|
||||
{
|
||||
CompletionSample::Ready => return Ok(()),
|
||||
CompletionSample::CompletedWithIncomplete(confirmed_missing) => {
|
||||
return Err(format!(
|
||||
"replacement recovery status never became definitive within {timeout_secs}s while waiting for physical census; latest status: {status}; missing: {missing:?}"
|
||||
)
|
||||
.into());
|
||||
"replacement status remained completed across two incomplete physical censuses: {confirmed_missing:?}"
|
||||
)
|
||||
.into());
|
||||
}
|
||||
tick.tick().await;
|
||||
continue;
|
||||
}
|
||||
if target_record_has_state(&status, target_disk, &["completed"]) {
|
||||
if !missing.is_empty() {
|
||||
return Err(format!(
|
||||
"replacement status reached completed before target physical census matched baseline: {missing:?}; status: {status}"
|
||||
)
|
||||
.into());
|
||||
}
|
||||
return Ok(());
|
||||
CompletionSample::Pending => {}
|
||||
}
|
||||
if Instant::now() >= deadline {
|
||||
let missing = incomplete_versions(target_disk, versions)?;
|
||||
let status = replacement_status(cluster).await?;
|
||||
return Err(format!(
|
||||
"replacement target did not reach completed with matching physical census within {timeout_secs}s: missing={missing:?}; status={status}"
|
||||
)
|
||||
@@ -812,6 +889,175 @@ mod tests {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn live_loss_barrier_requires_scanner_failure_after_log_offset() -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let target = Path::new("/mnt/target");
|
||||
assert!(live_disk_loss_scan_completed(
|
||||
"Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto-scan cycle completed",
|
||||
target
|
||||
));
|
||||
assert!(live_disk_loss_scan_completed(
|
||||
"Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle",
|
||||
target
|
||||
));
|
||||
assert!(!live_disk_loss_scan_completed(
|
||||
"Heal auto disk scanner idle\nHeal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed",
|
||||
target
|
||||
));
|
||||
assert!(!live_disk_loss_scan_completed(
|
||||
"event=disk_health_check_failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle",
|
||||
target
|
||||
));
|
||||
assert!(!live_disk_loss_scan_completed(
|
||||
"Heal auto-scan disk inspection failed endpoint=/mnt/other disk_state=check_failed\nHeal auto disk scanner idle",
|
||||
target
|
||||
));
|
||||
let path = std::env::temp_dir().join(format!("rustfs-replacement-scan-{}.log", std::process::id()));
|
||||
let stale =
|
||||
"Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle\n";
|
||||
fs::write(&path, stale)?;
|
||||
let offset = log_len(&path)?;
|
||||
assert!(!live_disk_loss_scan_completed_from_path(&path, offset, target)?);
|
||||
let fresh =
|
||||
"Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle\n";
|
||||
fs::write(&path, format!("{stale}{fresh}"))?;
|
||||
assert!(live_disk_loss_scan_completed_from_path(&path, offset, target)?);
|
||||
fs::remove_file(path)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completion_requires_definitive_status_and_prior_census_match() {
|
||||
let target = Path::new("/mnt/target");
|
||||
let non_definitive = serde_json::json!({
|
||||
"cluster": {"definitive": false, "records": [{"state": "completed", "targetSlots": ["/mnt/target"]}]}
|
||||
});
|
||||
assert_eq!(
|
||||
replacement_completion_state(&non_definitive, target, BTreeSet::new()).unwrap(),
|
||||
CompletionSample::Pending
|
||||
);
|
||||
|
||||
let omitted = serde_json::json!({
|
||||
"cluster": {"records": [{"state": "completed", "targetSlots": ["/mnt/target"]}]}
|
||||
});
|
||||
assert!(replacement_completion_state(&omitted, target, BTreeSet::new()).is_err());
|
||||
|
||||
let definitive = serde_json::json!({
|
||||
"cluster": {"definitive": true, "records": [{"state": "completed", "targetSlots": ["/mnt/target"]}]}
|
||||
});
|
||||
assert_eq!(
|
||||
replacement_completion_state(&definitive, target, BTreeSet::from(["missing".to_string()])).unwrap(),
|
||||
CompletionSample::CompletedWithIncomplete(BTreeSet::from(["missing".to_string()]))
|
||||
);
|
||||
assert_eq!(
|
||||
replacement_completion_state(&definitive, target, BTreeSet::new()).unwrap(),
|
||||
CompletionSample::Ready
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn completion_poll_samples_census_before_status() {
|
||||
let order = std::rc::Rc::new(std::cell::RefCell::new(Vec::new()));
|
||||
let census_order = order.clone();
|
||||
let status_order = order.clone();
|
||||
let sample = sample_replacement_completion(
|
||||
Path::new("/mnt/target"),
|
||||
move || {
|
||||
census_order.borrow_mut().push("census");
|
||||
Ok::<_, Box<dyn Error + Send + Sync>>(BTreeSet::new())
|
||||
},
|
||||
move || async move {
|
||||
status_order.borrow_mut().push("status");
|
||||
Ok::<_, Box<dyn Error + Send + Sync>>(serde_json::json!({
|
||||
"cluster": {"definitive": true, "records": [{"state": "completed", "targetSlots": ["/mnt/target"]}]}
|
||||
}))
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(sample, CompletionSample::Ready);
|
||||
assert_eq!(*order.borrow(), ["census", "status"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn completed_status_confirms_a_stale_incomplete_census() {
|
||||
let samples = std::rc::Rc::new(std::cell::RefCell::new(std::collections::VecDeque::from([
|
||||
BTreeSet::from(["missing".to_string()]),
|
||||
BTreeSet::new(),
|
||||
])));
|
||||
let census_samples = samples.clone();
|
||||
let result = confirm_replacement_completion(
|
||||
Path::new("/mnt/target"),
|
||||
move || {
|
||||
census_samples
|
||||
.borrow_mut()
|
||||
.pop_front()
|
||||
.ok_or_else(|| "missing census sample".into())
|
||||
},
|
||||
|| async {
|
||||
Ok(serde_json::json!({
|
||||
"cluster": {"definitive": true, "records": [{"state": "completed", "targetSlots": ["/mnt/target"]}]}
|
||||
}))
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(result, CompletionSample::Ready);
|
||||
assert!(samples.borrow().is_empty());
|
||||
|
||||
let status_samples = std::rc::Rc::new(std::cell::RefCell::new(std::collections::VecDeque::from([
|
||||
serde_json::json!({
|
||||
"cluster": {"definitive": true, "records": [{"state": "completed", "targetSlots": ["/mnt/target"]}]}
|
||||
}),
|
||||
serde_json::json!({
|
||||
"cluster": {"definitive": false, "records": []}
|
||||
}),
|
||||
])));
|
||||
let persistent = std::rc::Rc::new(std::cell::RefCell::new(std::collections::VecDeque::from([
|
||||
BTreeSet::from(["missing".to_string()]),
|
||||
BTreeSet::from(["still-missing".to_string()]),
|
||||
])));
|
||||
let census_samples = persistent.clone();
|
||||
let statuses = status_samples.clone();
|
||||
let result = confirm_replacement_completion(
|
||||
Path::new("/mnt/target"),
|
||||
move || {
|
||||
census_samples
|
||||
.borrow_mut()
|
||||
.pop_front()
|
||||
.ok_or_else(|| "missing census sample".into())
|
||||
},
|
||||
move || {
|
||||
let statuses = statuses.clone();
|
||||
async move {
|
||||
statuses
|
||||
.borrow_mut()
|
||||
.pop_front()
|
||||
.ok_or_else(|| "missing status sample".into())
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
result,
|
||||
CompletionSample::CompletedWithIncomplete(BTreeSet::from(["still-missing".to_string()]))
|
||||
);
|
||||
assert!(persistent.borrow().is_empty());
|
||||
assert_eq!(status_samples.borrow().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn absent_status_requires_definitive_empty_records() {
|
||||
let target = Path::new("/mnt/target");
|
||||
let non_definitive = serde_json::json!({"cluster": {"definitive": false, "records": []}});
|
||||
assert!(assert_no_replacement_status_records_in_status(&non_definitive, target).is_err());
|
||||
let omitted = serde_json::json!({"cluster": {"records": []}});
|
||||
assert!(assert_no_replacement_status_records_in_status(&omitted, target).is_err());
|
||||
let definitive = serde_json::json!({"cluster": {"definitive": true, "records": []}});
|
||||
assert!(assert_no_replacement_status_records_in_status(&definitive, target).is_ok());
|
||||
}
|
||||
|
||||
/// Linux mount namespaces are per-thread; keep mount setup and process
|
||||
/// spawning on one OS thread so child RustFS nodes inherit the test mounts.
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
|
||||
@@ -2150,30 +2150,6 @@ pub async fn load_data_usage_cache(store: &crate::set_disk::SetDisks, name: &str
|
||||
Ok(d)
|
||||
}
|
||||
|
||||
#[instrument(skip(cache))]
|
||||
pub async fn save_data_usage_cache(cache: &DataUsageCache, name: &str) -> crate::error::Result<()> {
|
||||
use crate::config::com::save_config;
|
||||
use crate::disk::BUCKET_META_PREFIX;
|
||||
use std::path::Path;
|
||||
|
||||
let Some(store) = runtime_sources::object_store_handle() else {
|
||||
return Err(Error::other("errServerNotInitialized"));
|
||||
};
|
||||
let buf = cache.marshal_msg().map_err(Error::other)?;
|
||||
let buf_clone = buf.clone();
|
||||
|
||||
let store_clone = store.clone();
|
||||
|
||||
let name = Path::new(BUCKET_META_PREFIX).join(name).to_string_lossy().to_string();
|
||||
|
||||
let name_clone = name.clone();
|
||||
tokio::spawn(async move {
|
||||
let _ = save_config(store_clone, &format!("{}{}", name_clone, ".bkp"), buf_clone).await;
|
||||
});
|
||||
save_config(store, &name, buf).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Persist the current in-memory compression total to the backend.
|
||||
/// Resets the debounce counter so the next auto-persist won't fire
|
||||
/// immediately after this manual flush (intended for shutdown paths).
|
||||
|
||||
@@ -331,7 +331,14 @@ impl From<std::io::Error> for DiskError {
|
||||
}
|
||||
match e.downcast::<DiskError>() {
|
||||
Ok(disk_error) => disk_error,
|
||||
Err(io_error) => DiskError::Io(io_error),
|
||||
// Mirror `From<io::Error> for StorageError`: a StorageError boxed
|
||||
// through `From<StorageError> for io::Error` must recover its typed
|
||||
// classification instead of degrading to `DiskError::Io`, which
|
||||
// quorum aggregation (`reduce_errs`) would count as a distinct error.
|
||||
Err(io_error) => match io_error.downcast::<crate::error::StorageError>() {
|
||||
Ok(storage_error) => storage_error.into(),
|
||||
Err(io_error) => DiskError::Io(io_error),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -953,6 +960,27 @@ mod tests {
|
||||
assert_eq!(original_disk_error, recovered_disk_error);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_io_error_with_storage_error_inside() {
|
||||
use crate::error::StorageError;
|
||||
|
||||
// An io::Error boxing a disk-representable StorageError (as produced by
|
||||
// `From<StorageError> for io::Error`) must recover the typed DiskError
|
||||
// variant instead of degrading to an opaque DiskError::Io.
|
||||
let io_with_storage_error: std::io::Error = StorageError::FaultyRemoteDisk.into();
|
||||
let recovered: DiskError = io_with_storage_error.into();
|
||||
assert_eq!(recovered, DiskError::FaultyRemoteDisk);
|
||||
|
||||
let io_with_storage_error: std::io::Error = StorageError::FileAccessDenied.into();
|
||||
let recovered: DiskError = io_with_storage_error.into();
|
||||
assert_eq!(recovered, DiskError::FileAccessDenied);
|
||||
|
||||
// A StorageError with no DiskError analog stays an opaque Io error.
|
||||
let io_with_bucket_error: std::io::Error = StorageError::BucketNotFound("bucket".to_string()).into();
|
||||
let recovered: DiskError = io_with_bucket_error.into();
|
||||
assert!(matches!(recovered, DiskError::Io(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_io_error_different_kinds() {
|
||||
use std::io::ErrorKind;
|
||||
|
||||
@@ -2955,11 +2955,31 @@ impl std::fmt::Debug for StdBackend {
|
||||
|
||||
impl StdBackend {
|
||||
pub(crate) fn new(root: PathBuf) -> Self {
|
||||
Self::build(root, true)
|
||||
}
|
||||
|
||||
/// Construct without the descriptor cache.
|
||||
///
|
||||
/// `UringBackend` wraps a `StdBackend` and runs its own `FdCache` over the
|
||||
/// same positioned reads. If the inner `StdBackend` also built a cache, a
|
||||
/// fallback read (`UringBackend::pread_bytes` delegates to the inner backend
|
||||
/// on latch-off / O_DIRECT / buffered errors) would populate a *second*
|
||||
/// cache that `UringBackend`'s invalidation never touches — re-opening the
|
||||
/// stale-inode hazard `FdCache` exists to close (rustfs/backlog#1176/#1801).
|
||||
/// The wrapper therefore owns the only cache for the disk; the inner backend
|
||||
/// opens per read. This also avoids double-counting `FD_CACHE_CAPACITY`
|
||||
/// against `RLIMIT_NOFILE` (rustfs/backlog#1178).
|
||||
#[cfg(target_os = "linux")]
|
||||
pub(crate) fn new_without_fd_cache(root: PathBuf) -> Self {
|
||||
Self::build(root, false)
|
||||
}
|
||||
|
||||
fn build(root: PathBuf, build_fd_cache: bool) -> Self {
|
||||
// Gate the fd cache on RLIMIT_NOFILE headroom (rustfs/backlog#1178):
|
||||
// 512 fds/disk with a low soft limit and several disks would hit EMFILE.
|
||||
// Fall back to open-per-read when the limit is too small.
|
||||
#[cfg(target_os = "linux")]
|
||||
let fd_cache = if is_local_fd_cache_enabled() {
|
||||
let fd_cache = if build_fd_cache && is_local_fd_cache_enabled() {
|
||||
if rlimit_allows_fd_cache() {
|
||||
Some(FdCache::new())
|
||||
} else {
|
||||
@@ -2973,6 +2993,10 @@ impl StdBackend {
|
||||
} else {
|
||||
None
|
||||
};
|
||||
// `build_fd_cache` is only consulted on Linux (for the fd cache); on
|
||||
// other platforms it has no effect and would trip the unused-variable lint.
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
let _ = build_fd_cache;
|
||||
Self {
|
||||
root,
|
||||
#[cfg(target_os = "linux")]
|
||||
@@ -4120,7 +4144,7 @@ impl UringBackend {
|
||||
// struct (rustfs/backlog#1185).
|
||||
let root_label = root.display().to_string();
|
||||
Some(Self {
|
||||
inner: StdBackend::new(root.clone()),
|
||||
inner: StdBackend::new_without_fd_cache(root.clone()),
|
||||
root,
|
||||
root_label,
|
||||
driver: std::mem::ManuallyDrop::new(driver),
|
||||
@@ -19919,6 +19943,23 @@ mod test {
|
||||
assert_eq!(cache.entry_count().await, 0, "prefix invalidation must drop the cached descriptor");
|
||||
}
|
||||
|
||||
/// `StdBackend::new_without_fd_cache` must not build a descriptor cache.
|
||||
/// `UringBackend` wraps a `StdBackend` and owns the only cache for the disk,
|
||||
/// so an inner cache would be populated by fallback reads
|
||||
/// (`UringBackend::pread_bytes` delegates inward) yet never invalidated —
|
||||
/// the stale-inode hazard `FdCache` exists to close (backlog#1176/#1801).
|
||||
/// This pins the contract so a future constructor change cannot regress it.
|
||||
#[cfg(target_os = "linux")]
|
||||
#[test]
|
||||
fn new_without_fd_cache_builds_no_descriptor_cache() {
|
||||
let root_dir = tempfile::tempdir().expect("operation should succeed");
|
||||
let backend = StdBackend::new_without_fd_cache(root_dir.path().to_path_buf());
|
||||
assert!(
|
||||
backend.fd_cache.is_none(),
|
||||
"new_without_fd_cache must not build a descriptor cache — UringBackend owns the only cache for the disk"
|
||||
);
|
||||
}
|
||||
|
||||
/// The mutation paths on `LocalDisk` must actually call
|
||||
/// `invalidate_cached_fds`, not merely have it available (backlog#1145).
|
||||
/// `rename_file` replaces the inode at a path a reader has already cached;
|
||||
|
||||
@@ -358,6 +358,13 @@ impl From<StorageError> for DiskError {
|
||||
StorageError::VolumeNotFound => DiskError::VolumeNotFound,
|
||||
StorageError::VolumeExists => DiskError::VolumeExists,
|
||||
StorageError::FileNameTooLong => DiskError::FileNameTooLong,
|
||||
StorageError::FaultyRemoteDisk => DiskError::FaultyRemoteDisk,
|
||||
StorageError::DiskAccessDenied => DiskError::DiskAccessDenied,
|
||||
StorageError::DriveIsRoot => DiskError::DriveIsRoot,
|
||||
StorageError::IsNotRegular => DiskError::IsNotRegular,
|
||||
StorageError::VolumeNotEmpty => DiskError::VolumeNotEmpty,
|
||||
StorageError::VolumeAccessDenied => DiskError::VolumeAccessDenied,
|
||||
StorageError::FileAccessDenied => DiskError::FileAccessDenied,
|
||||
_ => DiskError::other(val),
|
||||
}
|
||||
}
|
||||
@@ -1492,6 +1499,49 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
// Every DiskError variant must survive DiskError -> StorageError -> DiskError
|
||||
// unchanged. A variant that degrades to `DiskError::Io` on the way back loses
|
||||
// its identity for quorum aggregation (`reduce_errs` classifies by variant
|
||||
// equality), so ignore-list entries such as FaultyRemoteDisk and
|
||||
// DiskAccessDenied would silently stop matching.
|
||||
#[test]
|
||||
fn test_disk_error_storage_error_round_trip_identity_all_variants() {
|
||||
// DiskError codes are contiguous from 0x01, so enumerating via from_u32
|
||||
// covers every variant and picks up newly appended ones automatically.
|
||||
let all_variants: Vec<DiskError> = (1u32..).map_while(DiskError::from_u32).collect();
|
||||
assert!(
|
||||
all_variants.len() >= 42,
|
||||
"DiskError variant enumeration shrank: got {}, expected at least 42",
|
||||
all_variants.len()
|
||||
);
|
||||
|
||||
for original in all_variants {
|
||||
let storage_error: StorageError = original.clone().into();
|
||||
let round_tripped: DiskError = storage_error.into();
|
||||
|
||||
assert_eq!(
|
||||
std::mem::discriminant(&original),
|
||||
std::mem::discriminant(&round_tripped),
|
||||
"round trip changed variant: {original:?} -> {round_tripped:?}"
|
||||
);
|
||||
assert_eq!(original, round_tripped, "round trip not identical for {original:?}");
|
||||
}
|
||||
|
||||
// Io is the only payload-carrying variant: a representative kind and
|
||||
// message must both survive the round trip.
|
||||
let io_original = DiskError::Io(IoError::new(ErrorKind::PermissionDenied, "denied"));
|
||||
let storage_error: StorageError = io_original.clone().into();
|
||||
let io_round_tripped: DiskError = storage_error.into();
|
||||
assert_eq!(io_original, io_round_tripped);
|
||||
match io_round_tripped {
|
||||
DiskError::Io(inner) => {
|
||||
assert_eq!(inner.kind(), ErrorKind::PermissionDenied);
|
||||
assert_eq!(inner.to_string(), "denied");
|
||||
}
|
||||
other => panic!("expected DiskError::Io, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_storage_error_from_io_error() {
|
||||
// Test direct IO error conversion
|
||||
|
||||
@@ -9179,6 +9179,71 @@ mod tests {
|
||||
assert_eq!(body.as_ref(), payload);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn direct_memory_versioned_bucket_uses_inline_data_shards_for_latest() {
|
||||
let tempdir = tempfile::tempdir().expect("tempdir should be created");
|
||||
let endpoint =
|
||||
Endpoint::try_from(tempdir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
|
||||
let disk = new_disk(
|
||||
&endpoint,
|
||||
&DiskOption {
|
||||
cleanup: false,
|
||||
health_check: false,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("disk should be created");
|
||||
|
||||
let payload = vec![b'v'; 64 * 1024];
|
||||
let payload_size = i64::try_from(payload.len()).expect("test payload size should fit i64");
|
||||
let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
|
||||
let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
|
||||
fi.size = payload_size;
|
||||
fi.data = files[0].data.clone();
|
||||
fi.add_object_part(1, String::new(), payload.len(), None, payload_size, None, None);
|
||||
|
||||
let mut object_info = ObjectInfo {
|
||||
size: payload_size,
|
||||
actual_size: payload_size,
|
||||
parts: Arc::new(vec![ObjectPartInfo {
|
||||
number: 1,
|
||||
size: payload.len(),
|
||||
actual_size: payload_size,
|
||||
..Default::default()
|
||||
}]),
|
||||
..Default::default()
|
||||
};
|
||||
object_info.inlined = true;
|
||||
let opts = ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
};
|
||||
let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size);
|
||||
|
||||
assert_eq!(
|
||||
get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &opts, true, 128 * 1024),
|
||||
GetDirectMemoryDecision::Use {
|
||||
object_size: payload.len()
|
||||
}
|
||||
);
|
||||
|
||||
let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
|
||||
"bucket",
|
||||
"object",
|
||||
&fi,
|
||||
&files,
|
||||
&vec![Some(disk); erasure.total_shard_count()],
|
||||
true,
|
||||
GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
|
||||
metrics_size_bucket,
|
||||
)
|
||||
.await
|
||||
.expect("versioned latest direct-memory read should not fail")
|
||||
.expect("versioned latest should use inline data shards");
|
||||
|
||||
assert_eq!(body.as_ref(), payload);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn direct_memory_data_shards_direct_read_reassembles_single_block_payload() {
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -221,9 +221,9 @@ impl SetDisks {
|
||||
let disks = self.disks.read().await.clone();
|
||||
let required_reads = self.default_read_quorum();
|
||||
|
||||
let bucket = bucket.to_string();
|
||||
let object = object.to_string();
|
||||
let version_id = version_id.to_string();
|
||||
let bucket: Arc<str> = Arc::from(bucket);
|
||||
let object: Arc<str> = Arc::from(object);
|
||||
let version_id: Arc<str> = Arc::from(version_id);
|
||||
let opts = *opts;
|
||||
|
||||
let processor = runtime_sources::batch_processors().read_processor();
|
||||
|
||||
@@ -2403,6 +2403,150 @@ mod tests {
|
||||
assert_eq!(decoded.cache.get("bucket").map(|entry| entry.objects), Some(3));
|
||||
}
|
||||
|
||||
/// Deterministic, fully populated cache used to pin the persisted
|
||||
/// `.usage-cache.bin` wire bytes. Every map/set holds at most one element
|
||||
/// so the map-encoded `marshal_msg` output is byte-stable.
|
||||
fn wire_fixture_cache() -> DataUsageCache {
|
||||
let mut entry = DataUsageEntry {
|
||||
size: 4096,
|
||||
objects: 3,
|
||||
versions: 5,
|
||||
delete_markers: 1,
|
||||
compacted: true,
|
||||
failed_objects: 2,
|
||||
..Default::default()
|
||||
};
|
||||
entry.add_tier_sizes(&HashMap::from([(
|
||||
"WARM".to_string(),
|
||||
TierStats {
|
||||
total_size: 2048,
|
||||
num_versions: 2,
|
||||
num_objects: 1,
|
||||
},
|
||||
)]));
|
||||
let mut cache = DataUsageCache {
|
||||
info: DataUsageCacheInfo {
|
||||
name: "wire-bucket".to_string(),
|
||||
next_cycle: 7,
|
||||
leader_epoch: 9,
|
||||
last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000)),
|
||||
skip_healing: true,
|
||||
failed_objects: HashMap::from([("wire-bucket/lost".to_string(), 11)]),
|
||||
scan_resume_after: Some("wire-bucket/resume".to_string()),
|
||||
pending_heals: vec![PendingScannerHeal {
|
||||
kind: PendingScannerHealKind::Object,
|
||||
bucket: "wire-bucket".to_string(),
|
||||
object: Some("broken".to_string()),
|
||||
version_id: None,
|
||||
scan_mode: HealScanMode::Normal,
|
||||
first_seen: 100,
|
||||
last_attempt: 200,
|
||||
attempts: 3,
|
||||
last_admission_result: "deferred".to_string(),
|
||||
last_admission_reason: "budget".to_string(),
|
||||
}],
|
||||
source: Some(DataUsageCacheSource::new(1, 2)),
|
||||
snapshot_complete: true,
|
||||
scan_plan_digest: Some(TEST_PLAN_DIGEST),
|
||||
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
cache.replace("wire-bucket", "", entry);
|
||||
cache
|
||||
}
|
||||
|
||||
/// Persisted `.usage-cache.bin` bytes produced by [`wire_fixture_cache`]
|
||||
/// via `DataUsageCache::marshal_msg`: a 2-element array of the 16-field
|
||||
/// map-encoded info block and the map of map-encoded entries.
|
||||
///
|
||||
/// The thin read-only projection in `crates/data-usage` decodes a copy of
|
||||
/// this fixture (`thin_usage_cache_decodes_scanner_wire_fixture`); when
|
||||
/// the encoding legitimately changes, regenerate both copies from
|
||||
/// `wire_fixture_cache().marshal_msg()` and re-verify old readers.
|
||||
const USAGE_CACHE_WIRE_FIXTURE: &[u8] = &[
|
||||
0x92, 0xde, 0x00, 0x10, 0xa4, 0x6e, 0x61, 0x6d, 0x65, 0xab, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65,
|
||||
0x74, 0xaa, 0x6e, 0x65, 0x78, 0x74, 0x5f, 0x63, 0x79, 0x63, 0x6c, 0x65, 0x07, 0xac, 0x6c, 0x65, 0x61, 0x64, 0x65, 0x72,
|
||||
0x5f, 0x65, 0x70, 0x6f, 0x63, 0x68, 0x09, 0xab, 0x6c, 0x61, 0x73, 0x74, 0x5f, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x92,
|
||||
0xce, 0x65, 0x53, 0xf1, 0x00, 0x00, 0xac, 0x73, 0x6b, 0x69, 0x70, 0x5f, 0x68, 0x65, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0xc3,
|
||||
0xa9, 0x6c, 0x69, 0x66, 0x65, 0x63, 0x79, 0x63, 0x6c, 0x65, 0xc0, 0xab, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74,
|
||||
0x69, 0x6f, 0x6e, 0xc0, 0xae, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x81,
|
||||
0xb0, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x2f, 0x6c, 0x6f, 0x73, 0x74, 0x0b, 0xb1, 0x73,
|
||||
0x63, 0x61, 0x6e, 0x5f, 0x72, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x5f, 0x61, 0x66, 0x74, 0x65, 0x72, 0xb2, 0x77, 0x69, 0x72,
|
||||
0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x2f, 0x72, 0x65, 0x73, 0x75, 0x6d, 0x65, 0xaf, 0x73, 0x63, 0x61, 0x6e,
|
||||
0x5f, 0x63, 0x68, 0x65, 0x63, 0x6b, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0xc0, 0xad, 0x70, 0x65, 0x6e, 0x64, 0x69, 0x6e, 0x67,
|
||||
0x5f, 0x68, 0x65, 0x61, 0x6c, 0x73, 0x91, 0x9a, 0xa6, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0xab, 0x77, 0x69, 0x72, 0x65,
|
||||
0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0xa6, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x6e, 0xc0, 0x01, 0x64, 0xcc, 0xc8, 0x03,
|
||||
0xa8, 0x64, 0x65, 0x66, 0x65, 0x72, 0x72, 0x65, 0x64, 0xa6, 0x62, 0x75, 0x64, 0x67, 0x65, 0x74, 0xab, 0x6f, 0x62, 0x6a,
|
||||
0x65, 0x63, 0x74, 0x5f, 0x6c, 0x6f, 0x63, 0x6b, 0xc0, 0xa6, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x92, 0x01, 0x02, 0xb1,
|
||||
0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x5f, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0xc3, 0xb0, 0x73,
|
||||
0x63, 0x61, 0x6e, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x5f, 0x64, 0x69, 0x67, 0x65, 0x73, 0x74, 0xdc, 0x00, 0x20, 0x03, 0x03,
|
||||
0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03,
|
||||
0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0xb0, 0x63, 0x61, 0x63, 0x68, 0x65, 0x5f, 0x6b, 0x65, 0x79,
|
||||
0x5f, 0x66, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x01, 0x81, 0xab, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63, 0x6b, 0x65,
|
||||
0x74, 0x8b, 0xa8, 0x63, 0x68, 0x69, 0x6c, 0x64, 0x72, 0x65, 0x6e, 0x90, 0xa4, 0x73, 0x69, 0x7a, 0x65, 0xcd, 0x10, 0x00,
|
||||
0xa7, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x03, 0xa8, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x05, 0xae,
|
||||
0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x5f, 0x6d, 0x61, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x01, 0xa9, 0x6f, 0x62, 0x6a, 0x5f,
|
||||
0x73, 0x69, 0x7a, 0x65, 0x73, 0x9b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xac, 0x6f, 0x62,
|
||||
0x6a, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x97, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xb1, 0x72,
|
||||
0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0xc0, 0xa9, 0x63, 0x6f,
|
||||
0x6d, 0x70, 0x61, 0x63, 0x74, 0x65, 0x64, 0xc3, 0xae, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x6f, 0x62, 0x6a, 0x65,
|
||||
0x63, 0x74, 0x73, 0x02, 0xae, 0x61, 0x6c, 0x6c, 0x5f, 0x74, 0x69, 0x65, 0x72, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0x91,
|
||||
0x81, 0xa4, 0x57, 0x41, 0x52, 0x4d, 0x93, 0xcd, 0x08, 0x00, 0x02, 0x01,
|
||||
];
|
||||
|
||||
#[test]
|
||||
fn usage_cache_wire_format_is_pinned() {
|
||||
// Writer: the canonical map-encoded serializer must reproduce the
|
||||
// pinned bytes. Round-trip tests cannot see format drift, so any
|
||||
// encoding change (field rename/reorder, map->array switch) fails
|
||||
// here and forces re-verifying old readers and the thin projection
|
||||
// in crates/data-usage before the fixture is regenerated.
|
||||
let encoded = wire_fixture_cache().marshal_msg().expect("marshal fixture cache");
|
||||
assert_eq!(
|
||||
encoded.as_slice(),
|
||||
USAGE_CACHE_WIRE_FIXTURE,
|
||||
"persisted .usage-cache.bin encoding drifted from the pinned fixture"
|
||||
);
|
||||
|
||||
// Reader: the pinned bytes decode with every field intact.
|
||||
let decoded = DataUsageCache::unmarshal(USAGE_CACHE_WIRE_FIXTURE).expect("decode pinned fixture");
|
||||
assert_eq!(decoded.info.name, "wire-bucket");
|
||||
assert_eq!(decoded.info.next_cycle, 7);
|
||||
assert_eq!(decoded.info.leader_epoch, 9);
|
||||
assert_eq!(
|
||||
decoded.info.last_update,
|
||||
Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000))
|
||||
);
|
||||
assert!(decoded.info.skip_healing);
|
||||
assert_eq!(decoded.info.failed_objects.get("wire-bucket/lost"), Some(&11));
|
||||
assert_eq!(decoded.info.scan_resume_after.as_deref(), Some("wire-bucket/resume"));
|
||||
assert_eq!(decoded.info.pending_heals.len(), 1);
|
||||
assert_eq!(decoded.info.pending_heals[0].kind, PendingScannerHealKind::Object);
|
||||
assert_eq!(decoded.info.pending_heals[0].object.as_deref(), Some("broken"));
|
||||
assert_eq!(decoded.info.source, Some(DataUsageCacheSource::new(1, 2)));
|
||||
assert!(decoded.info.snapshot_complete);
|
||||
assert_eq!(decoded.info.scan_plan_digest, Some(TEST_PLAN_DIGEST));
|
||||
assert_eq!(decoded.info.cache_key_format, DATA_USAGE_CACHE_KEY_FORMAT);
|
||||
|
||||
let entry = decoded.cache.get("wire-bucket").expect("fixture entry decodes");
|
||||
assert_eq!(entry.size, 4096);
|
||||
assert_eq!(entry.objects, 3);
|
||||
assert_eq!(entry.versions, 5);
|
||||
assert_eq!(entry.delete_markers, 1);
|
||||
assert!(entry.compacted);
|
||||
assert_eq!(entry.failed_objects, 2);
|
||||
assert_eq!(
|
||||
entry.all_tier_stats.as_ref().and_then(|tiers| tiers.tiers.get("WARM")),
|
||||
Some(&TierStats {
|
||||
total_size: 2048,
|
||||
num_versions: 2,
|
||||
num_objects: 1,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn data_usage_cache_prepare_for_scan_rejects_unscoped_distributed_cache() {
|
||||
let mut cache = DataUsageCache {
|
||||
|
||||
Reference in New Issue
Block a user