mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-29 11:15:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

volume server: VolumeEcShardsDelete full teardown unloads every disk and keeps the shard gauge honest (#11446)

Browse Source 
* volume server: VolumeEcShardsDelete full teardown unloads every disk and keeps the shard gauge honest

Go's VolumeEcShardsDelete full teardown calls vs.store.UnloadEcVolume in
the blanket path (weed/server/volume_grpc_erasure_coding.go:488) and
location.UnloadEcVolume in the generation-fenced path (:511): each disk
that had the volume registered drops it, closes its shard descriptors
and gives back its ec_shards gauge before the artifacts are unlinked.

The Rust handler used Store::remove_ec_volume / DiskLocation::remove_ec_volume
instead, which only remove the map entry. Store::remove_ec_volume also
stops at the FIRST disk holding the vid, so on a split-disk volume
(shards on several disks) the blanket teardown left the sibling disks'
EcVolume registered with open fds while the unlink loop deleted their
files underneath it: the heartbeat kept advertising shards whose files
were gone, the inodes stayed pinned by the open descriptors, and the
VOLUME_GAUGE{collection,"ec_shards"} never came back down. The fenced
path leaked the gauge and the descriptors the same way on the one disk
it wiped.

Both paths now use the unload_ec_volume helpers from #11413 (every disk
for the blanket teardown, the strictly-older disk for the fenced one),
and the two Status::internal messages name the disk directory like Go's
"... on %s: %w".

Regression tests build a two-disk store with the same vid mounted on
each disk (the SplitDiskEcFixture, which gains a collection knob so the
gauge read is isolated from parallel tests mounting under "") and assert
that a blanket teardown leaves no EcVolume registered on any disk and
returns the gauge to its pre-mount value, and that a fenced teardown
decrements the gauge for the older disk's shard while preserving the
newer disk. Both fail against the previous handler.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* volume server: trim comments on the EC full-teardown unload path

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Eliah Rusin
2026-09-26 12:00:23 +08:00
committed by GitHub
co-authored by Claude Fable 5.1 Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Chris Lu Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
parent 3e9fc9e75b
commit 4299fdf578
1 changed files with 192 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+192 -17
View File
@@ -3760,15 +3760,18 @@ impl VolumeServer for VolumeGrpcService {
// listed shards, so a remote node retains no stale generation a fresh copy
// collides with. Echo the acknowledgement so the caller can tell a
// pre-upgrade server apart.
//
// Unload every disk holding the vid (Go's Store.UnloadEcVolume):
// each registration returns its descriptors and ec_shards gauge.
{
let mut store = self.state.store.write().unwrap();
let _ = store.remove_ec_volume(vid);
store.unload_ec_volume(vid);
for loc in &store.locations {
loc.remove_ec_volume_files_full_teardown(&req.collection, vid)
.map_err(|e| {
Status::internal(format!(
"full teardown of ec volume {}: {}",
req.volume_id, e
"full teardown of ec volume {} on {}: {}",
req.volume_id, loc.directory, e
))
})?;
}
@@ -3794,13 +3797,13 @@ impl VolumeServer for VolumeGrpcService {
);
continue;
}
store.locations[disk_id].remove_ec_volume(vid);
store.locations[disk_id].unload_ec_volume(vid);
store.locations[disk_id]
.remove_ec_volume_files_full_teardown(&req.collection, vid)
.map_err(|e| {
Status::internal(format!(
"fenced teardown of ec volume {}: {}",
req.volume_id, e
"fenced teardown of ec volume {} on {}: {}",
req.volume_id, store.locations[disk_id].directory, e
))
})?;
}
@@ -7990,6 +7993,166 @@ mod tests {
);
}
/// Blanket full teardown (`encode_ts_ns == 0`) on a split-disk volume:
/// like Go's Store.UnloadEcVolume, every disk unregisters its EcVolume
/// and returns its ec_shards gauge before the files are unlinked.
#[tokio::test]
async fn test_volume_ec_shards_delete_full_teardown_unloads_every_disk() {
let collection = "ecdel-blanket";
let vid = VolumeId(7060);
let gauge = crate::metrics::VOLUME_GAUGE.with_label_values(&[collection, "ec_shards"]);
let gauge_before_mount = gauge.get();
let (service, tmp) = SplitDiskEcFixture {
collection,
..SplitDiskEcFixture::new(vid.0)
}
.build();
{
let store = service.state.store.read().unwrap();
assert_eq!(store.locations.len(), 2);
for (disk_id, loc) in store.locations.iter().enumerate() {
assert!(
loc.has_ec_volume(vid),
"fixture must mount the vid on disk {disk_id} for this to be meaningful"
);
}
}
assert_eq!(
gauge.get(),
gauge_before_mount + 2.0,
"fixture must account one shard per disk"
);
let resp = service
.volume_ec_shards_delete(Request::new(
volume_server_pb::VolumeEcShardsDeleteRequest {
volume_id: vid.0,
collection: collection.to_string(),
shard_ids: Vec::new(),
full_teardown: true,
encode_ts_ns: 0,
},
))
.await
.unwrap()
.into_inner();
assert!(resp.full_teardown_done);
{
let store = service.state.store.read().unwrap();
for (disk_id, loc) in store.locations.iter().enumerate() {
assert!(
!loc.has_ec_volume(vid),
"blanket full teardown must unload the EcVolume on disk {disk_id}, \
not just the first disk holding the vid"
);
}
}
assert_eq!(
gauge.get(),
gauge_before_mount,
"the ec_shards gauge must return to its pre-mount value"
);
for (dir, ext) in [("data0", ".ec00"), ("data1", ".ec01"), ("data1", ".ecx")] {
let path = tmp
.path()
.join(dir)
.join(format!("{}_{}{}", collection, vid.0, ext));
assert!(!path.exists(), "{} must be gone", path.display());
}
}
/// Fenced teardown (`encode_ts_ns != 0`) unloads only a disk whose
/// generation is strictly older than the request, as Go's
/// location.UnloadEcVolume does, returning the ec_shards gauge it held.
#[tokio::test]
async fn test_volume_ec_shards_delete_fenced_teardown_gives_back_gauge_for_older_disk() {
let collection = "ecdel-fenced";
let vid = VolumeId(7061);
let gauge = crate::metrics::VOLUME_GAUGE.with_label_values(&[collection, "ec_shards"]);
let gauge_before_mount = gauge.get();
// dir0 carries generation 100, dir1 generation 200: a fence at 150
// wipes dir0 alone.
let (service, tmp) = SplitDiskEcFixture {
collection,
dir0_encode_ts_ns: Some(100),
dir1_encode_ts_ns: 200,
..SplitDiskEcFixture::new(vid.0)
}
.build();
{
let store = service.state.store.read().unwrap();
assert_eq!(
store.locations[0].ec_generation_ts_ns(collection, vid),
Some(100)
);
assert_eq!(
store.locations[1].ec_generation_ts_ns(collection, vid),
Some(200)
);
assert!(store.locations[0].has_ec_volume(vid));
assert!(store.locations[1].has_ec_volume(vid));
}
assert_eq!(
gauge.get(),
gauge_before_mount + 2.0,
"fixture must account one shard per disk"
);
let resp = service
.volume_ec_shards_delete(Request::new(
volume_server_pb::VolumeEcShardsDeleteRequest {
volume_id: vid.0,
collection: collection.to_string(),
shard_ids: Vec::new(),
full_teardown: true,
encode_ts_ns: 150,
},
))
.await
.unwrap()
.into_inner();
assert!(resp.full_teardown_done);
{
let store = service.state.store.read().unwrap();
assert!(
!store.locations[0].has_ec_volume(vid),
"the strictly-older disk must be unloaded"
);
assert!(
store.locations[1].has_ec_volume(vid),
"the newer disk must be preserved"
);
}
assert_eq!(
gauge.get(),
gauge_before_mount + 1.0,
"fenced teardown must give back the ec_shards gauge for the older \
disk's shard and leave the preserved disk's shard counted"
);
let older_shard = tmp
.path()
.join("data0")
.join(format!("{}_{}.ec00", collection, vid.0));
assert!(
!older_shard.exists(),
"{} must be gone",
older_shard.display()
);
let newer_shard = tmp
.path()
.join("data1")
.join(format!("{}_{}.ec01", collection, vid.0));
assert!(
newer_shard.exists(),
"{} must survive",
newer_shard.display()
);
}
/// REGRESSION: a node-wide scrub must survive a volume that legitimately
/// disappears while it runs.
///
@@ -8334,6 +8497,9 @@ mod tests {
/// field is a deliberate departure from it.
struct SplitDiskEcFixture {
vid_raw: u32,
/// Collection the files and mounted volume carry; the delete tests use
/// unique names so parallel tests under `""` share no gauge labels.
collection: &'static str,
/// Which shard id each disk gets. Which disk holds the shard a needle
/// actually spans is the whole difference between reaching one disk and
/// reaching both, so the LOCAL/CHECKSUM tests move shard 0 to disk 1.
@@ -8377,6 +8543,7 @@ mod tests {
fn new(vid_raw: u32) -> Self {
SplitDiskEcFixture {
vid_raw,
collection: "",
dir0_shard_id: 0,
dir1_shard_id: 1,
dir0_encode_ts_ns: None,
@@ -8396,6 +8563,7 @@ mod tests {
) -> (VolumeGrpcService, TempDir) {
let SplitDiskEcFixture {
vid_raw,
collection,
dir0_shard_id,
dir1_shard_id,
dir0_encode_ts_ns,
@@ -8409,6 +8577,17 @@ mod tests {
let dir1 = tmp.path().join("data1");
std::fs::create_dir_all(&dir0).unwrap();
std::fs::create_dir_all(&dir1).unwrap();
let file = |dir: &std::path::Path, ext: &str| {
format!(
"{}{}",
crate::storage::volume::volume_file_name(
dir.to_str().unwrap(),
collection,
VolumeId(vid_raw),
),
ext
)
};
let ec_vif = |encode_ts_ns: i64| crate::storage::volume::VifVolumeInfo {
version: 3,
@@ -8423,20 +8602,20 @@ mod tests {
// dir0: one shard, no .ecx.
std::fs::write(
dir0.join(format!("{}.ec{:02}", vid_raw, dir0_shard_id)),
file(&dir0, &format!(".ec{:02}", dir0_shard_id)),
b"shard data nonempty",
)
.unwrap();
if let Some(ts) = dir0_encode_ts_ns {
std::fs::write(
dir0.join(format!("{}.vif", vid_raw)),
file(&dir0, ".vif"),
serde_json::to_string(&ec_vif(ts)).unwrap(),
)
.unwrap();
}
// dir1: one shard plus the index files.
std::fs::write(
dir1.join(format!("{}.ec{:02}", vid_raw, dir1_shard_id)),
file(&dir1, &format!(".ec{:02}", dir1_shard_id)),
b"shard data nonempty",
)
.unwrap();
@@ -8453,10 +8632,10 @@ mod tests {
} else {
vec![0u8; 20]
};
std::fs::write(dir1.join(format!("{}.ecx", vid_raw)), ecx).unwrap();
std::fs::write(dir1.join(format!("{}.ecj", vid_raw)), b"").unwrap();
std::fs::write(file(&dir1, ".ecx"), ecx).unwrap();
std::fs::write(file(&dir1, ".ecj"), b"").unwrap();
std::fs::write(
dir1.join(format!("{}.vif", vid_raw)),
file(&dir1, ".vif"),
serde_json::to_string(&ec_vif(dir1_encode_ts_ns)).unwrap(),
)
.unwrap();
@@ -8489,11 +8668,7 @@ mod tests {
SidecarPlacement::Dir1Only => vec![dir1.as_path()],
};
for d in sidecar_dirs {
let base = crate::storage::volume::volume_file_name(
d.to_str().unwrap(),
"",
VolumeId(vid_raw),
);
let base = file(d, "");
ec_bitrot::save_bitrot_sidecar(&ec_bitrot::bitrot_sidecar_path(&base, 0), &prot)
.unwrap();
}