mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-12 06:48:45 +01:00
* weed/server: verify data shards against bitrot sidecar before ec decode (#12422) * ec.decode: verify data shards against the mounted volume's bitrot protection The decode looked up only the generation-0 .ecsum beside the data base and each location's data directory, so it missed a sidecar kept with the index and ignored the generation the volume actually serves. Take the protection the volume's per-disk runtimes resolved instead (On, else Invalid, else Off), and test a corrupt block past the first. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec.decode: check for live entries before verifying the shards A volume whose needles are all deleted gets its shards purged on the no-live-entries status; a failing checksum must not keep it around. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: verify data shards against the bitrot protection before a decode Mirrors the Go volume server: the decode takes the protection the volume's per-disk runtimes resolved and refuses a data shard that fails its .ecsum checksum, after the no-live-entries check. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test: a decode refuses a data shard that fails its .ecsum checksum Runs against both volume server implementations: the refused decode keeps every shard and the sidecar, and the needle stays readable through EC. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec.decode: reload the bitrot sidecar before verifying the shards A rebuild's backfill writes the sidecar without reloading the mounted runtimes, which then still report no protection; resolve it again across the EC metadata directories, as a shard mount does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: reload the bitrot sidecar before verifying the shards Mirrors the Go volume server: the decode job resolves the sidecar again on every runtime of the volume before choosing its protection. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test: a decode finds a sidecar written after the shards were mounted Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test: read through the EC shards after the refused decode The source volume stayed mounted, so the final read never touched the shards it is meant to check. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec.decode: refuse when a sidecar reload reports another layout ReloadBitrotSidecar logged and dropped the layout-mismatch error, so the runtime stayed Off and the decode skipped the data-shard check. It now returns that error and the decode refuses on it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: refuse a decode when a sidecar reload reports another layout Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test: a decode refuses a late sidecar that records another layout Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec.decode: refuse when an unmount left no runtime to resolve the sidecar Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: refuse a decode when an unmount left no runtime to resolve the sidecar Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
8 files changed
+535
-12
No files matched your search
@@ -4353,7 +4353,7 @@ impl VolumeServer for VolumeGrpcService {
|
||||
// however often it reloaded. One delivered copy, reachable from all.
|
||||
let ec_metadata_dirs = store.ec_metadata_dirs();
|
||||
for ec_vol in store.find_all_ec_volumes_mut(vid) {
|
||||
ec_vol.reload_bitrot_sidecar(&ec_metadata_dirs);
|
||||
let _ = ec_vol.reload_bitrot_sidecar(&ec_metadata_dirs);
|
||||
}
|
||||
drop(store);
|
||||
self.state.volume_state_notify.notify_one();
|
||||
@@ -7766,6 +7766,45 @@ impl EcDecodeJob {
|
||||
)));
|
||||
}
|
||||
|
||||
let shard_paths: Vec<String> = shard_dirs
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, dir)| {
|
||||
crate::storage::erasure_coding::ec_shard::EcVolumeShard::new(
|
||||
dir,
|
||||
&collection,
|
||||
vid,
|
||||
i as ShardId,
|
||||
)
|
||||
.file_name()
|
||||
})
|
||||
.collect();
|
||||
// A rebuild's backfill writes the sidecar without reloading the
|
||||
// mounted runtimes, so resolve it again the way a shard mount does.
|
||||
// Any runtime left finds it on every disk; with none left (an unmount
|
||||
// raced the decode) the protection would read as Off.
|
||||
let (bitrot, bitrot_status) = {
|
||||
let mut store = state.store.write().unwrap();
|
||||
let ec_metadata_dirs = store.ec_metadata_dirs();
|
||||
let runtimes = store.find_all_ec_volumes_mut(vid);
|
||||
if runtimes.is_empty() {
|
||||
return Err(Status::internal(format!(
|
||||
"ec volume {} was unmounted during the decode",
|
||||
vid
|
||||
)));
|
||||
}
|
||||
for ec_vol in runtimes {
|
||||
ec_vol
|
||||
.reload_bitrot_sidecar(&ec_metadata_dirs)
|
||||
.map_err(|e| Status::internal(format!("ec volume {}: {}", vid, e)))?;
|
||||
}
|
||||
crate::storage::erasure_coding::ec_volume::EcVolume::select_bitrot_protection(
|
||||
&store.find_all_ec_volumes(vid),
|
||||
)
|
||||
};
|
||||
ec_bitrot::verify_ec_volume_data_shards(bitrot.as_ref(), bitrot_status, &shard_paths)
|
||||
.map_err(|e| Status::internal(format!("ec volume {}: {}", vid, e)))?;
|
||||
|
||||
// Calculate .dat file size from .ecx entries (.ec00 lives on
|
||||
// its own disk, .ecx on the index disk).
|
||||
let dat_file_size = ec_decoder::find_dat_file_size_with_dirs(
|
||||
@@ -14805,6 +14844,149 @@ mod tests {
|
||||
assert!(std::path::Path::new(&idx_sidecar).exists());
|
||||
}
|
||||
|
||||
/// A data shard that fails its .ecsum checksum must fail the decode and
|
||||
/// leave every shard and the sidecar in place for a repair.
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn test_volume_ec_shards_to_volume_rejects_a_corrupt_data_shard() {
|
||||
let (service, _tmp, data, _idx, _) = make_split_idx_ec_decode_service(&[], false).await;
|
||||
assert_corrupt_data_shard_rejected(&service, &data).await;
|
||||
}
|
||||
|
||||
/// A rebuild's backfill writes the sidecar after the shards are mounted,
|
||||
/// without reloading the runtimes: the decode must still find it.
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn test_volume_ec_shards_to_volume_rejects_a_corrupt_shard_with_a_late_sidecar() {
|
||||
let (service, _tmp, data, _idx, _) = make_split_idx_ec_decode_service(&[], false).await;
|
||||
let sidecar = format!("{data}/1.ecsum");
|
||||
let hidden = format!("{data}/1.ecsum.hidden");
|
||||
std::fs::rename(&sidecar, &hidden).unwrap();
|
||||
{
|
||||
let mut store = service.state.store.write().unwrap();
|
||||
for ec_vol in store.find_all_ec_volumes_mut(VolumeId(1)) {
|
||||
ec_vol.reload_bitrot_sidecar(&[]).unwrap();
|
||||
assert_eq!(
|
||||
ec_vol.bitrot_protection().1,
|
||||
crate::storage::erasure_coding::ec_bitrot::BitrotStatus::Off,
|
||||
"precondition: the mounted runtime holds no protection"
|
||||
);
|
||||
}
|
||||
}
|
||||
std::fs::rename(&hidden, &sidecar).unwrap();
|
||||
assert_corrupt_data_shard_rejected(&service, &data).await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn test_volume_ec_shards_to_volume_refuses_a_late_sidecar_with_another_layout() {
|
||||
use crate::storage::erasure_coding::ec_bitrot;
|
||||
let (service, _tmp, data, _idx, _) = make_split_idx_ec_decode_service(&[], false).await;
|
||||
let sidecar = format!("{data}/1.ecsum");
|
||||
let hidden = format!("{data}/1.ecsum.hidden");
|
||||
std::fs::rename(&sidecar, &hidden).unwrap();
|
||||
{
|
||||
let mut store = service.state.store.write().unwrap();
|
||||
for ec_vol in store.find_all_ec_volumes_mut(VolumeId(1)) {
|
||||
ec_vol.reload_bitrot_sidecar(&[]).unwrap();
|
||||
}
|
||||
}
|
||||
let mut prot = ec_bitrot::load_bitrot_sidecar(&hidden).unwrap();
|
||||
prot.ec_shard_config.as_mut().unwrap().parity_shards = 2;
|
||||
ec_bitrot::save_bitrot_sidecar(&sidecar, &prot).unwrap();
|
||||
std::fs::remove_file(&hidden).unwrap();
|
||||
|
||||
let err = service
|
||||
.volume_ec_shards_to_volume(ec_shards_to_volume_request())
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(
|
||||
err.message().contains("refusing to serve"),
|
||||
"unexpected error: {}",
|
||||
err.message()
|
||||
);
|
||||
assert!(!std::path::Path::new(&format!("{data}/1.dat")).exists());
|
||||
for shard_id in 0..14 {
|
||||
assert!(std::path::Path::new(&format!("{data}/1.ec{shard_id:02}")).exists());
|
||||
}
|
||||
}
|
||||
|
||||
/// An unmount between the job snapshot and the verification leaves no
|
||||
/// runtime to resolve the sidecar: the job must refuse, not verify as Off.
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn test_ec_decode_job_refuses_when_every_runtime_was_unmounted() {
|
||||
let (service, _tmp, data, _idx, _) = make_split_idx_ec_decode_service(&[], false).await;
|
||||
let job = {
|
||||
let store = service.state.store.read().unwrap();
|
||||
let (ec_vol, shard_dirs) = store
|
||||
.collect_ec_shard_dirs(
|
||||
VolumeId(1),
|
||||
crate::storage::erasure_coding::ec_shard::MAX_SHARD_COUNT,
|
||||
)
|
||||
.unwrap();
|
||||
let data_shards = ec_vol.data_shards as usize;
|
||||
EcDecodeJob {
|
||||
vid: VolumeId(1),
|
||||
dat_dir: ec_vol.dir.clone(),
|
||||
ecx_dir: ec_vol.ecx_actual_dir().to_string(),
|
||||
idx_dir: ec_vol.dir_idx.clone(),
|
||||
collection: ec_vol.collection.clone(),
|
||||
vif_dat_file_size: ec_vol.dat_file_size,
|
||||
large_block_size: ec_vol.large_block_size() as usize,
|
||||
small_block_size: ec_vol.small_block_size() as usize,
|
||||
shard_dirs: shard_dirs[..data_shards]
|
||||
.iter()
|
||||
.map(|d| d.clone().unwrap())
|
||||
.collect(),
|
||||
needle_map_kind: store.needle_map_kind,
|
||||
}
|
||||
};
|
||||
service
|
||||
.volume_ec_shards_unmount(Request::new(
|
||||
volume_server_pb::VolumeEcShardsUnmountRequest {
|
||||
volume_id: 1,
|
||||
shard_ids: (0..14).collect(),
|
||||
encode_ts_ns: 0,
|
||||
},
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let shard = format!("{data}/1.ec03");
|
||||
let mut bytes = std::fs::read(&shard).unwrap();
|
||||
bytes[0] ^= 0xff;
|
||||
std::fs::write(&shard, bytes).unwrap();
|
||||
|
||||
let err = job.run(&service.state).unwrap_err();
|
||||
|
||||
assert!(
|
||||
err.message().contains("unmounted during the decode"),
|
||||
"unexpected error: {}",
|
||||
err.message()
|
||||
);
|
||||
assert!(!std::path::Path::new(&format!("{data}/1.dat")).exists());
|
||||
}
|
||||
|
||||
async fn assert_corrupt_data_shard_rejected(service: &VolumeGrpcService, data: &str) {
|
||||
let shard = format!("{data}/1.ec03");
|
||||
let mut bytes = std::fs::read(&shard).unwrap();
|
||||
bytes[0] ^= 0xff;
|
||||
std::fs::write(&shard, bytes).unwrap();
|
||||
|
||||
let err = service
|
||||
.volume_ec_shards_to_volume(ec_shards_to_volume_request())
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(
|
||||
err.message().contains("data shard 3"),
|
||||
"unexpected error: {}",
|
||||
err.message()
|
||||
);
|
||||
assert!(!std::path::Path::new(&format!("{data}/1.dat")).exists());
|
||||
assert!(std::path::Path::new(&format!("{data}/1.ecsum")).exists());
|
||||
for shard_id in 0..14 {
|
||||
assert!(std::path::Path::new(&format!("{data}/1.ec{shard_id:02}")).exists());
|
||||
}
|
||||
}
|
||||
|
||||
/// Replaces the idx-dir .ecj with a FIFO: the decode's open of it blocks
|
||||
/// until [`release_ecj_fifo`] opens the write end.
|
||||
#[cfg(unix)]
|
||||
|
||||
Reference in new issue
Block a user