diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 7be0ed6ac..ab5d2e5af 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -451,6 +451,7 @@ message VolumeEcShardsDeleteRequest { string collection = 2; repeated uint32 shard_ids = 3; bool full_teardown = 4; // pre-encode cleanup: wipe every EC artifact + generation for this volume, not just shard_ids + int64 encode_ts_ns = 5; // full_teardown generation fence: delete only a disk whose .vif generation is strictly OLDER than this; preserve same-or-newer, generation 0, and an unreadable .vif. 0 => wipe-all (shell pre-encode / pre-upgrade) } message VolumeEcShardsDeleteResponse { bool full_teardown_done = 1; // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op @@ -468,6 +469,7 @@ message VolumeEcShardsMountResponse { message VolumeEcShardsUnmountRequest { uint32 volume_id = 1; repeated uint32 shard_ids = 3; + int64 encode_ts_ns = 4; // generation fence: skip a disk whose mounted EC volume is this generation or newer (0 = unfenced, unmount all) } message VolumeEcShardsUnmountResponse { } diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 02e0e808e..6a0f80801 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -2662,15 +2662,54 @@ impl VolumeServer for VolumeGrpcService { let vid = VolumeId(req.volume_id); if req.full_teardown { - // Pre-encode cleanup: evict the volume and wipe every EC artifact for it - // on every disk, not just the listed shards, so a remote node retains no - // stale generation that a fresh gen-0 copy would collide with. Echo the - // acknowledgement so the caller can tell a pre-upgrade server apart. - { + if req.encode_ts_ns == 0 { + // Blanket teardown (shell pre-encode cleanup / pre-upgrade caller): evict + // the volume and wipe every EC artifact for it on every disk, not just the + // 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. + { + let mut store = self.state.store.write().unwrap(); + let _ = store.remove_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 + )) + })?; + } + } + } else { + // Generation-fenced teardown (stale-worker pre-distribute cleanup): wipe + // only a disk whose .vif generation is strictly OLDER than the request; + // preserve same-or-newer, generation 0 (recovered/pre-upgrade live volume), + // and an unreadable .vif, so a stale run never wipes a newer run's live + // shards. Unload and remove only the strictly-older disks, never node-wide. let mut store = self.state.store.write().unwrap(); - let _ = store.remove_ec_volume(vid); - for loc in &store.locations { - loc.remove_ec_volume_files_full_teardown(&req.collection, vid); + for disk_id in 0..store.locations.len() { + let disk_gen = store.locations[disk_id].ec_generation_ts_ns(&req.collection, vid); + let older = matches!(disk_gen, Some(g) if g > 0 && g < req.encode_ts_ns); + if !older { + tracing::info!( + volume_id = vid.0, + disk_id, + ?disk_gen, + req = req.encode_ts_ns, + "ec full teardown preserved: generation not older than request" + ); + continue; + } + store.locations[disk_id].remove_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 + )) + })?; } } self.state.volume_state_notify.notify_one(); @@ -2728,9 +2767,11 @@ impl VolumeServer for VolumeGrpcService { // Matches Go: for _, shardId := range req.ShardIds { err = vs.store.UnmountEcShards(...) } let mut store = self.state.store.write().unwrap(); for &shard_id in &req.shard_ids { - store.unmount_ec_shard(vid, shard_id).map_err(|e| { - Status::internal(format!("unmount {}.{}: {}", req.volume_id, shard_id, e)) - })?; + store + .unmount_ec_shard(vid, shard_id, req.encode_ts_ns) + .map_err(|e| { + Status::internal(format!("unmount {}.{}: {}", req.volume_id, shard_id, e)) + })?; } drop(store); self.state.volume_state_notify.notify_one(); diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 4ba3e9f68..13c71cbed 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -157,7 +157,7 @@ impl DiskLocation { volume_id = vid.0, "EC volume validation failed, removing incomplete EC files" ); - self.remove_ec_volume_files(&collection, vid); + let _ = self.remove_ec_volume_files(&collection, vid); // Fall through to load .dat file } } @@ -377,24 +377,25 @@ impl DiskLocation { /// `store_ec_reconcile.rs` can call it to scrub partial EC artefacts /// when a healthy `.dat` for the same vid lives on a sibling disk /// (seaweedfs/seaweedfs#9478). - pub(crate) fn remove_ec_volume_files(&self, collection: &str, vid: VolumeId) { + pub(crate) fn remove_ec_volume_files(&self, collection: &str, vid: VolumeId) -> io::Result<()> { let base = volume_file_name(&self.directory, collection, vid); let idx_base = volume_file_name(&self.idx_directory, collection, vid); const MAX_SHARD_COUNT: usize = 32; // Remove index files from idx directory (.ecx, .ecj) - let _ = fs::remove_file(format!("{}.ecx", idx_base)); - let _ = fs::remove_file(format!("{}.ecj", idx_base)); + rm_if_present(format!("{}.ecx", idx_base))?; + rm_if_present(format!("{}.ecj", idx_base))?; // Also try data directory in case .ecx/.ecj were created before -dir.idx was configured if self.idx_directory != self.directory { - let _ = fs::remove_file(format!("{}.ecx", base)); - let _ = fs::remove_file(format!("{}.ecj", base)); + rm_if_present(format!("{}.ecx", base))?; + rm_if_present(format!("{}.ecj", base))?; } // Remove all EC shard files (.ec00 ~ .ec31) for i in 0..MAX_SHARD_COUNT { - let _ = fs::remove_file(format!("{}.ec{:02}", base, i)); + rm_if_present(format!("{}.ec{:02}", base, i))?; } + Ok(()) } /// Full-teardown variant: everything remove_ec_volume_files clears, PLUS the @@ -405,8 +406,12 @@ impl DiskLocation { /// .vif. Reconcile/load-fallback call remove_ec_volume_files directly and /// intentionally preserve it, mirroring Go's removeEcVolumeFiles (reconcile) vs /// removeStaleEcArtifacts (teardown) split. - pub(crate) fn remove_ec_volume_files_full_teardown(&self, collection: &str, vid: VolumeId) { - self.remove_ec_volume_files(collection, vid); + pub(crate) fn remove_ec_volume_files_full_teardown( + &self, + collection: &str, + vid: VolumeId, + ) -> io::Result<()> { + self.remove_ec_volume_files(collection, vid)?; let base = volume_file_name(&self.directory, collection, vid); let idx_base = volume_file_name(&self.idx_directory, collection, vid); // try_exists, not exists(): a stat error on a PRESENT .idx must not read as @@ -416,16 +421,36 @@ impl DiskLocation { std::path::Path::new(&format!("{}.idx", idx_base)).try_exists(), Ok(false) ) { - let _ = fs::remove_file(format!("{}.vif", idx_base)); + rm_if_present(format!("{}.vif", idx_base))?; if self.idx_directory != self.directory && matches!( std::path::Path::new(&format!("{}.idx", base)).try_exists(), Ok(false) ) { - let _ = fs::remove_file(format!("{}.vif", base)); + rm_if_present(format!("{}.vif", base))?; } } + Ok(()) + } + + /// EC encode generation recorded in this disk's .vif (data dir first, then idx + /// dir for the split-disk layout). None if no .vif was readable, so the fenced + /// teardown preserves it (fail-safe). A present .vif with no ec_shard_config + /// yields Some(0) (a recovered or pre-upgrade live volume), also preserved. + pub(crate) fn ec_generation_ts_ns(&self, collection: &str, vid: VolumeId) -> Option { + for dir in [&self.directory, &self.idx_directory] { + let vif = format!("{}.vif", volume_file_name(dir, collection, vid)); + if let Ok(s) = fs::read_to_string(&vif) { + if let Ok(vi) = serde_json::from_str::(&s) { + return Some(vi.ec_shard_config.map(|c| c.encode_ts_ns).unwrap_or(0)); + } + } + if self.directory == self.idx_directory { + break; + } + } + None } /// Find a volume by ID. @@ -924,7 +949,7 @@ impl DiskLocation { volume_id = vid.0, "Incomplete or invalid EC volume: .dat exists but validation failed, cleaning up EC files", ); - self.remove_ec_volume_files(collection, vid); + let _ = self.remove_ec_volume_files(collection, vid); return; } @@ -966,7 +991,7 @@ impl DiskLocation { "Found {} EC shards without .ecx file (incomplete encoding interrupted before .ecx), cleaning up", shards.len(), ); - self.remove_ec_volume_files(collection, vid); + let _ = self.remove_ec_volume_files(collection, vid); return true; } false @@ -1046,6 +1071,16 @@ fn calculate_expected_shard_size(dat_file_size: i64, data_shards: usize) -> i64 /// Resolve the EC data-shard count from the volume's own `.vif` (the volume /// server never holds the cluster EC config in memory), checking the data dir /// then the idx dir. Falls back to the default ratio when no EC `.vif` is found. +/// Remove a file, treating a missing file as success but propagating real errors +/// (permissions, disk full). Mirrors Go's removeFileIfExists so a teardown failure +/// surfaces instead of silently leaving stale artifacts. +fn rm_if_present(path: String) -> io::Result<()> { + match fs::remove_file(&path) { + Err(e) if e.kind() != io::ErrorKind::NotFound => Err(e), + _ => Ok(()), + } +} + fn ec_data_shards_from_vif(directory: &str, idx_directory: &str, collection: &str, vid: VolumeId) -> usize { for dir in [directory, idx_directory] { let vif = format!("{}.vif", volume_file_name(dir, collection, vid)); @@ -1278,6 +1313,61 @@ mod tests { ); } + // The per-disk generation reader behind the fenced EC teardown: a present .vif + // yields its generation (or 0 with no EC config), a missing .vif is unreadable + // (preserved, fail-safe), and the idx dir is a fallback for the split-disk layout. + #[test] + fn test_ec_generation_ts_ns_reads_and_fences() { + let tmp = TempDir::new().unwrap(); + let data = tmp.path().join("data"); + let idx = tmp.path().join("idx"); + std::fs::create_dir_all(&data).unwrap(); + std::fs::create_dir_all(&idx).unwrap(); + let loc = DiskLocation::new( + data.to_str().unwrap(), + idx.to_str().unwrap(), + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + let vid = VolumeId(88); + let dbase = volume_file_name(data.to_str().unwrap(), "", vid); + let ibase = volume_file_name(idx.to_str().unwrap(), "", vid); + + // No .vif anywhere -> unreadable (the fenced teardown preserves on None). + assert_eq!(loc.ec_generation_ts_ns("", vid), None); + + // .vif in the data dir carries its generation. + let with_gen = VifVolumeInfo { + version: 3, + ec_shard_config: Some(crate::storage::volume::VifEcShardConfig { + data_shards: 10, + parity_shards: 4, + encode_ts_ns: 4242, + ..Default::default() + }), + ..Default::default() + }; + std::fs::write(format!("{}.vif", dbase), serde_json::to_string(&with_gen).unwrap()).unwrap(); + assert_eq!(loc.ec_generation_ts_ns("", vid), Some(4242)); + + // A .vif with no EC config reads as generation 0 (recovered/pre-upgrade live volume). + std::fs::remove_file(format!("{}.vif", dbase)).unwrap(); + let no_cfg = VifVolumeInfo { + version: 3, + ..Default::default() + }; + std::fs::write(format!("{}.vif", dbase), serde_json::to_string(&no_cfg).unwrap()).unwrap(); + assert_eq!(loc.ec_generation_ts_ns("", vid), Some(0)); + + // idx-dir fallback: only the idx dir holds the .vif. + std::fs::remove_file(format!("{}.vif", dbase)).unwrap(); + std::fs::write(format!("{}.vif", ibase), serde_json::to_string(&with_gen).unwrap()).unwrap(); + assert_eq!(loc.ec_generation_ts_ns("", vid), Some(4242)); + } + #[test] fn test_parse_volume_filename() { assert_eq!( diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 0ad57e2bf..da5aa177e 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -716,17 +716,37 @@ impl Store { /// Unmount a single EC shard, searching all locations. /// Matches Go's Store.UnmountEcShards which unmounts one shard at a time. - pub fn unmount_ec_shard(&mut self, vid: VolumeId, shard_id: u32) -> Result<(), VolumeError> { + pub fn unmount_ec_shard( + &mut self, + vid: VolumeId, + shard_id: u32, + req_encode_ts_ns: i64, + ) -> Result<(), VolumeError> { // Walk all locations rather than stopping at the first with the // vid — split-disk reconciled volumes can have the same vid on // multiple disks, with the target shard on any of them. for disk_id in 0..self.locations.len() { - let has_shard = self.locations[disk_id] - .find_ec_volume(vid) - .is_some_and(|ec_vol| ec_vol.has_shard(shard_id as u8)); + let ec_vol = self.locations[disk_id].find_ec_volume(vid); + let has_shard = ec_vol.is_some_and(|ec_vol| ec_vol.has_shard(shard_id as u8)); if !has_shard { continue; } + // Generation fence: when the caller carries a generation (stale-worker + // cleanup), only unmount a strictly-older generation; preserve a disk + // whose generation is same-or-newer, 0, or unknown, so a stale run cannot + // unmount a newer run's live shards. req 0 (legacy/shell) unmounts all. + let disk_gen = ec_vol.map_or(0, |v| v.encode_ts_ns); + if req_encode_ts_ns > 0 && !(disk_gen > 0 && disk_gen < req_encode_ts_ns) { + tracing::info!( + volume_id = vid.0, + shard_id, + disk_id, + disk_gen, + req = req_encode_ts_ns, + "UnmountEcShards skipped: generation not older than request" + ); + continue; + } tracing::info!( volume_id = vid.0, shard_id, diff --git a/seaweed-volume/src/storage/store_ec_reconcile.rs b/seaweed-volume/src/storage/store_ec_reconcile.rs index afeef7547..9c494f29a 100644 --- a/seaweed-volume/src/storage/store_ec_reconcile.rs +++ b/seaweed-volume/src/storage/store_ec_reconcile.rs @@ -332,7 +332,7 @@ impl Store { // Also sweep any unmounted shard files (.ec00 .. .ec31) // that the per-disk loader skipped — destroy() only walks // the in-memory shards, but the disk may still hold others. - loc.remove_ec_volume_files(&v.collection, v.vid); + let _ = loc.remove_ec_volume_files(&v.collection, v.vid); } } @@ -1044,7 +1044,7 @@ mod tests { let (mut store, _tmp) = build_split_disk_store(7003); let vid = VolumeId(7003); - store.unmount_ec_shard(vid, 1).unwrap(); + store.unmount_ec_shard(vid, 1, 0).unwrap(); assert_eq!(store.find_ec_shard_location(vid, 1), None); // Other shards untouched. assert_eq!(store.find_ec_shard_location(vid, 0), Some(0)); diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 14769d76d..5825445d3 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -453,6 +453,7 @@ message VolumeEcShardsDeleteRequest { string collection = 2; repeated uint32 shard_ids = 3; bool full_teardown = 4; // pre-encode cleanup: wipe every EC artifact + generation for this volume, not just shard_ids + int64 encode_ts_ns = 5; // full_teardown generation fence: delete only a disk whose .vif generation is strictly OLDER than this; preserve same-or-newer, generation 0, and an unreadable .vif. 0 => wipe-all (shell pre-encode / pre-upgrade) } message VolumeEcShardsDeleteResponse { bool full_teardown_done = 1; // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op @@ -470,6 +471,7 @@ message VolumeEcShardsMountResponse { message VolumeEcShardsUnmountRequest { uint32 volume_id = 1; repeated uint32 shard_ids = 3; + int64 encode_ts_ns = 4; // generation fence: skip a disk whose mounted EC volume is this generation or newer (0 = unfenced, unmount all) } message VolumeEcShardsUnmountResponse { } diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 6d82373a4..ff0f4786d 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -3606,6 +3606,7 @@ type VolumeEcShardsDeleteRequest struct { Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"` ShardIds []uint32 `protobuf:"varint,3,rep,packed,name=shard_ids,json=shardIds,proto3" json:"shard_ids,omitempty"` FullTeardown bool `protobuf:"varint,4,opt,name=full_teardown,json=fullTeardown,proto3" json:"full_teardown,omitempty"` // pre-encode cleanup: wipe every EC artifact + generation for this volume, not just shard_ids + EncodeTsNs int64 `protobuf:"varint,5,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"` // full_teardown generation fence: delete only a disk whose .vif generation is strictly OLDER than this; preserve same-or-newer, generation 0, and an unreadable .vif. 0 => wipe-all (shell pre-encode / pre-upgrade) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3668,6 +3669,13 @@ func (x *VolumeEcShardsDeleteRequest) GetFullTeardown() bool { return false } +func (x *VolumeEcShardsDeleteRequest) GetEncodeTsNs() int64 { + if x != nil { + return x.EncodeTsNs + } + return 0 +} + type VolumeEcShardsDeleteResponse struct { state protoimpl.MessageState `protogen:"open.v1"` FullTeardownDone bool `protobuf:"varint,1,opt,name=full_teardown_done,json=fullTeardownDone,proto3" json:"full_teardown_done,omitempty"` // set by a new server that performed full_teardown; absent from an old server lets the caller detect the silent no-op @@ -3820,6 +3828,7 @@ type VolumeEcShardsUnmountRequest struct { state protoimpl.MessageState `protogen:"open.v1"` VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` ShardIds []uint32 `protobuf:"varint,3,rep,packed,name=shard_ids,json=shardIds,proto3" json:"shard_ids,omitempty"` + EncodeTsNs int64 `protobuf:"varint,4,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"` // generation fence: skip a disk whose mounted EC volume is this generation or newer (0 = unfenced, unmount all) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3868,6 +3877,13 @@ func (x *VolumeEcShardsUnmountRequest) GetShardIds() []uint32 { return nil } +func (x *VolumeEcShardsUnmountRequest) GetEncodeTsNs() int64 { + if x != nil { + return x.EncodeTsNs + } + return 0 +} + type VolumeEcShardsUnmountResponse struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -7268,14 +7284,16 @@ const file_volume_server_proto_rawDesc = "" + "\rcopy_vif_file\x18\a \x01(\bR\vcopyVifFile\x12\x17\n" + "\adisk_id\x18\b \x01(\rR\x06diskId\x12&\n" + "\x0fcopy_ecsum_file\x18\t \x01(\bR\rcopyEcsumFile\"\x1c\n" + - "\x1aVolumeEcShardsCopyResponse\"\x9c\x01\n" + + "\x1aVolumeEcShardsCopyResponse\"\xbe\x01\n" + "\x1bVolumeEcShardsDeleteRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" + "\n" + "collection\x18\x02 \x01(\tR\n" + "collection\x12\x1b\n" + "\tshard_ids\x18\x03 \x03(\rR\bshardIds\x12#\n" + - "\rfull_teardown\x18\x04 \x01(\bR\ffullTeardown\"L\n" + + "\rfull_teardown\x18\x04 \x01(\bR\ffullTeardown\x12 \n" + + "\fencode_ts_ns\x18\x05 \x01(\x03R\n" + + "encodeTsNs\"L\n" + "\x1cVolumeEcShardsDeleteResponse\x12,\n" + "\x12full_teardown_done\x18\x01 \x01(\bR\x10fullTeardownDone\"\xa0\x01\n" + "\x1aVolumeEcShardsMountRequest\x12\x1b\n" + @@ -7285,10 +7303,12 @@ const file_volume_server_proto_rawDesc = "" + "collection\x12\x1b\n" + "\tshard_ids\x18\x03 \x03(\rR\bshardIds\x12(\n" + "\x10source_disk_type\x18\x04 \x01(\tR\x0esourceDiskType\"\x1d\n" + - "\x1bVolumeEcShardsMountResponse\"X\n" + + "\x1bVolumeEcShardsMountResponse\"z\n" + "\x1cVolumeEcShardsUnmountRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1b\n" + - "\tshard_ids\x18\x03 \x03(\rR\bshardIds\"\x1f\n" + + "\tshard_ids\x18\x03 \x03(\rR\bshardIds\x12 \n" + + "\fencode_ts_ns\x18\x04 \x01(\x03R\n" + + "encodeTsNs\"\x1f\n" + "\x1dVolumeEcShardsUnmountResponse\"\xc1\x01\n" + "\x18VolumeEcShardReadRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x19\n" + diff --git a/weed/pb/worker.proto b/weed/pb/worker.proto index 51a84c007..00d4cb288 100644 --- a/weed/pb/worker.proto +++ b/weed/pb/worker.proto @@ -178,6 +178,7 @@ message ErasureCodingTaskParams { string master_client = 5; // Master server address bool cleanup_source = 6; // Whether to cleanup source volume after EC string source_disk_type = 8; // Source volume's disk type, passed to VolumeEcShardsMount so shards report under it (#9423) + int64 encode_ts_ns = 9; // admin-issued encode generation (unix nanos); fences a stale worker's shard cleanup against a newer run. 0 => unfenced (legacy/shell), falls back to a blanket teardown. } // TaskSource represents a unified source location for any task type diff --git a/weed/pb/worker_pb/worker.pb.go b/weed/pb/worker_pb/worker.pb.go index d3f0b3628..d3305a474 100644 --- a/weed/pb/worker_pb/worker.pb.go +++ b/weed/pb/worker_pb/worker.pb.go @@ -1323,6 +1323,7 @@ type ErasureCodingTaskParams struct { MasterClient string `protobuf:"bytes,5,opt,name=master_client,json=masterClient,proto3" json:"master_client,omitempty"` // Master server address CleanupSource bool `protobuf:"varint,6,opt,name=cleanup_source,json=cleanupSource,proto3" json:"cleanup_source,omitempty"` // Whether to cleanup source volume after EC SourceDiskType string `protobuf:"bytes,8,opt,name=source_disk_type,json=sourceDiskType,proto3" json:"source_disk_type,omitempty"` // Source volume's disk type, passed to VolumeEcShardsMount so shards report under it (#9423) + EncodeTsNs int64 `protobuf:"varint,9,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"` // admin-issued encode generation (unix nanos); fences a stale worker's shard cleanup against a newer run. 0 => unfenced (legacy/shell), falls back to a blanket teardown. unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1406,6 +1407,13 @@ func (x *ErasureCodingTaskParams) GetSourceDiskType() string { return "" } +func (x *ErasureCodingTaskParams) GetEncodeTsNs() int64 { + if x != nil { + return x.EncodeTsNs + } + return 0 +} + // TaskSource represents a unified source location for any task type type TaskSource struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -4056,7 +4064,7 @@ const file_worker_proto_rawDesc = "" + "batch_size\x18\x03 \x01(\x05R\tbatchSize\x12\x1f\n" + "\vworking_dir\x18\x04 \x01(\tR\n" + "workingDir\x12'\n" + - "\x0fverify_checksum\x18\x05 \x01(\bR\x0everifyChecksum\"\xae\x02\n" + + "\x0fverify_checksum\x18\x05 \x01(\bR\x0everifyChecksum\"\xd0\x02\n" + "\x17ErasureCodingTaskParams\x120\n" + "\x14estimated_shard_size\x18\x01 \x01(\x04R\x12estimatedShardSize\x12\x1f\n" + "\vdata_shards\x18\x02 \x01(\x05R\n" + @@ -4066,7 +4074,9 @@ const file_worker_proto_rawDesc = "" + "workingDir\x12#\n" + "\rmaster_client\x18\x05 \x01(\tR\fmasterClient\x12%\n" + "\x0ecleanup_source\x18\x06 \x01(\bR\rcleanupSource\x12(\n" + - "\x10source_disk_type\x18\b \x01(\tR\x0esourceDiskTypeJ\x04\b\a\x10\b\"\xcf\x01\n" + + "\x10source_disk_type\x18\b \x01(\tR\x0esourceDiskType\x12 \n" + + "\fencode_ts_ns\x18\t \x01(\x03R\n" + + "encodeTsNsJ\x04\b\a\x10\b\"\xcf\x01\n" + "\n" + "TaskSource\x12\x12\n" + "\x04node\x18\x01 \x01(\tR\x04node\x12\x17\n" + diff --git a/weed/server/volume_grpc_ec_generation_fence_test.go b/weed/server/volume_grpc_ec_generation_fence_test.go new file mode 100644 index 000000000..2b6c8b1e9 --- /dev/null +++ b/weed/server/volume_grpc_ec_generation_fence_test.go @@ -0,0 +1,185 @@ +package weed_server + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/require" +) + +// buildEcStoreWithGeneration creates a single-disk store holding one EC volume +// whose .vif records the given encode generation, with the given shards mounted. +func buildEcStoreWithGeneration(t *testing.T, dir, collection string, vid needle.VolumeId, encodeTsNs int64, shardIds []erasure_coding.ShardId) *storage.Store { + t.Helper() + require.NoError(t, os.MkdirAll(dir, 0o755)) + store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "", + storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, stats.DefaultDiskIOProbeConfig()) + done := make(chan struct{}) + go func() { + for { + select { + case <-store.NewEcShardsChan: + case <-store.NewVolumesChan: + case <-store.DeletedVolumesChan: + case <-store.DeletedEcShardsChan: + case <-store.StateUpdateChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + store.Close() + close(done) + }) + + base := erasure_coding.EcShardFileName(collection, dir, int(vid)) + require.NoError(t, os.WriteFile(base+".ecx", make([]byte, 16), 0o644)) + require.NoError(t, os.WriteFile(base+".ecj", nil, 0o644)) + require.NoError(t, volume_info.SaveVolumeInfo(base+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + DatFileSize: 10 * 1024 * 1024, + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: 10, + ParityShards: 4, + EncodeTsNs: encodeTsNs, + }, + })) + for _, sid := range shardIds { + f, err := os.Create(base + erasure_coding.ToExt(int(sid))) + require.NoError(t, err) + require.NoError(t, f.Truncate(1)) + require.NoError(t, f.Close()) + } + for _, sid := range shardIds { + require.NoError(t, store.MountEcShards(collection, vid, sid, "")) + } + return store +} + +func mountedEcShardIds(t *testing.T, vs *VolumeServer, vid needle.VolumeId) map[int]bool { + t.Helper() + resp, err := vs.VolumeEcShardsInfo(context.Background(), &volume_server_pb.VolumeEcShardsInfoRequest{VolumeId: uint32(vid)}) + require.NoError(t, err) + ids := make(map[int]bool) + for _, info := range resp.GetEcShardInfos() { + ids[int(info.GetShardId())] = true + } + return ids +} + +// TestFullTeardownFencedByGeneration pins the finding-27 data-safety rule: a +// generation-fenced FullTeardown deletes a disk whose .vif generation is strictly +// older than the request, but preserves a same-or-newer generation, a generation-0 +// (recovered/pre-upgrade) volume, and falls back to a blanket wipe for request 0. +func TestFullTeardownFencedByGeneration(t *testing.T) { + const collection = "ec-fence" + vid := needle.VolumeId(55) + shardIds := []erasure_coding.ShardId{0, 1} + shardExists := func(dir string) bool { + base := erasure_coding.EcShardFileName(collection, dir, int(vid)) + return util.FileExists(base + erasure_coding.ToExt(0)) + } + teardown := func(vs *VolumeServer, reqGen int64) { + _, err := vs.VolumeEcShardsDelete(context.Background(), &volume_server_pb.VolumeEcShardsDeleteRequest{ + VolumeId: uint32(vid), + Collection: collection, + FullTeardown: true, + EncodeTsNs: reqGen, + }) + require.NoError(t, err) + } + + t.Run("older_disk_wiped", func(t *testing.T) { + dir := t.TempDir() + vs := &VolumeServer{store: buildEcStoreWithGeneration(t, dir, collection, vid, 100, shardIds)} + teardown(vs, 200) // request newer than the disk's generation 100 + require.False(t, shardExists(dir), "a strictly-older generation must be wiped") + }) + + t.Run("newer_disk_preserved", func(t *testing.T) { + dir := t.TempDir() + vs := &VolumeServer{store: buildEcStoreWithGeneration(t, dir, collection, vid, 200, shardIds)} + teardown(vs, 100) // request older than the disk's generation 200 + require.True(t, shardExists(dir), "a newer generation (a live newer run) must be preserved") + }) + + t.Run("zero_gen_preserved", func(t *testing.T) { + dir := t.TempDir() + vs := &VolumeServer{store: buildEcStoreWithGeneration(t, dir, collection, vid, 0, shardIds)} + teardown(vs, 200) // a recovered/pre-upgrade live volume reports generation 0 + require.True(t, shardExists(dir), "a generation-0 volume must be preserved under a fenced teardown") + }) + + t.Run("zero_request_blanket_wipe", func(t *testing.T) { + dir := t.TempDir() + vs := &VolumeServer{store: buildEcStoreWithGeneration(t, dir, collection, vid, 200, shardIds)} + teardown(vs, 0) // shell pre-encode / pre-upgrade caller wipes everything + require.False(t, shardExists(dir), "request generation 0 must blanket-wipe") + }) +} + +// TestUnmountEcShardsFencedByGeneration pins that the gen-aware unmount (issued +// before the teardown) preserves a same-or-newer mounted generation, so a stale +// worker cannot unmount a newer run's live shards out from under it. +func TestUnmountEcShardsFencedByGeneration(t *testing.T) { + const collection = "ec-unmount-fence" + vid := needle.VolumeId(56) + dir := t.TempDir() + store := buildEcStoreWithGeneration(t, dir, collection, vid, 200, []erasure_coding.ShardId{0, 1}) + vs := &VolumeServer{store: store} + + // Request older than the disk generation 200: preserve (skip unmount). + require.NoError(t, store.UnmountEcShards(vid, 0, 100)) + require.True(t, mountedEcShardIds(t, vs, vid)[0], "a same-or-newer generation shard must stay mounted") + + // Request newer than the disk generation 200: a genuinely-older leftover is unmounted. + require.NoError(t, store.UnmountEcShards(vid, 1, 300)) + require.False(t, mountedEcShardIds(t, vs, vid)[1], "a strictly-older generation shard must be unmounted") +} + +// TestReadEcGenerationTsNs covers the per-disk .vif generation read used by the +// fenced teardown: a present .vif yields its generation (or 0 when it has no EC +// config), and a missing .vif is reported unreadable (preserved, fail-safe). +func TestReadEcGenerationTsNs(t *testing.T) { + dir := t.TempDir() + base := filepath.Join(dir, "9") + + if _, readable := readEcGenerationTsNs(base, base); readable { + t.Fatalf("a missing .vif must be reported unreadable") + } + + require.NoError(t, volume_info.SaveVolumeInfo(base+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + EcShardConfig: &volume_server_pb.EcShardConfig{DataShards: 10, ParityShards: 4, EncodeTsNs: 12345}, + })) + gen, readable := readEcGenerationTsNs(base, base) + require.True(t, readable) + require.Equal(t, int64(12345), gen) + + // A .vif with no EC config (recovered / live source volume) reads as generation 0. + noCfg := filepath.Join(dir, "10") + require.NoError(t, volume_info.SaveVolumeInfo(noCfg+".vif", &volume_server_pb.VolumeInfo{Version: uint32(needle.Version3)})) + gen0, readable0 := readEcGenerationTsNs(noCfg, noCfg) + require.True(t, readable0) + require.Equal(t, int64(0), gen0) + + // A present-but-unparseable .vif is reported as generation 0 (present), which the + // fenced teardown still preserves — never wiping on a parse error. + corrupt := filepath.Join(dir, "11") + require.NoError(t, os.WriteFile(corrupt+".vif", []byte("not-a-valid-vif"), 0o644)) + genC, readableC := readEcGenerationTsNs(corrupt, corrupt) + require.True(t, readableC) + require.Equal(t, int64(0), genC) +} diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index c19c21088..ac4d91ec6 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -418,17 +418,38 @@ func (vs *VolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_se bName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId)) if req.FullTeardown { - // Pre-encode cleanup: evict the volume and wipe every EC artifact for it on - // every disk (the same teardown the generator does locally), not just the - // listed shards, so a remote node retains no stale generation that a fresh - // gen-0 copy would later collide with. - glog.V(0).Infof("ec volume %s full teardown", bName) - vs.store.UnloadEcVolume(needle.VolumeId(req.VolumeId)) + if req.EncodeTsNs == 0 { + // Blanket teardown (shell pre-encode cleanup / pre-upgrade caller): evict the + // volume and wipe every EC artifact for it on every disk, not just the listed + // shards, so a remote node retains no stale generation a fresh copy collides with. + glog.V(0).Infof("ec volume %s full teardown", bName) + vs.store.UnloadEcVolume(needle.VolumeId(req.VolumeId)) + for _, location := range vs.store.Locations { + dataBase := storage.VolumeFileName(location.Directory, req.Collection, int(req.VolumeId)) + idxBase := storage.VolumeFileName(location.IdxDirectory, req.Collection, int(req.VolumeId)) + if err := removeStaleEcArtifacts(dataBase, idxBase, erasure_coding.MaxShardCount); err != nil { + return nil, fmt.Errorf("full teardown of ec volume %d on %s: %w", req.VolumeId, location.Directory, err) + } + } + return &volume_server_pb.VolumeEcShardsDeleteResponse{FullTeardownDone: true}, nil + } + // Generation-fenced teardown (stale-worker pre-distribute cleanup): wipe only a + // disk whose .vif generation is strictly OLDER than the request; preserve + // same-or-newer, generation 0 (recovered/pre-upgrade live volume), and an + // unreadable .vif, so a stale run can never wipe a newer run's live shards. + // Unload and remove only the strictly-older disks, never node-wide. + glog.V(0).Infof("ec volume %s full teardown fenced at generation %d", bName, req.EncodeTsNs) for _, location := range vs.store.Locations { dataBase := storage.VolumeFileName(location.Directory, req.Collection, int(req.VolumeId)) idxBase := storage.VolumeFileName(location.IdxDirectory, req.Collection, int(req.VolumeId)) + diskGen, readable := readEcGenerationTsNs(dataBase, idxBase) + if !readable || diskGen == 0 || diskGen >= req.EncodeTsNs { + glog.V(1).Infof("ec volume %d on %s preserved: disk generation %d (readable=%v) not older than request %d", req.VolumeId, location.Directory, diskGen, readable, req.EncodeTsNs) + continue + } + location.UnloadEcVolume(needle.VolumeId(req.VolumeId)) if err := removeStaleEcArtifacts(dataBase, idxBase, erasure_coding.MaxShardCount); err != nil { - return nil, fmt.Errorf("full teardown of ec volume %d on %s: %w", req.VolumeId, location.Directory, err) + return nil, fmt.Errorf("fenced teardown of ec volume %d on %s: %w", req.VolumeId, location.Directory, err) } } return &volume_server_pb.VolumeEcShardsDeleteResponse{FullTeardownDone: true}, nil @@ -534,6 +555,24 @@ func removeFileIfExists(path string) error { return nil } +// readEcGenerationTsNs returns the EC encode generation recorded in a disk's .vif +// (data dir first, then idx dir for the split-disk layout) and whether a .vif file +// was present. A present .vif with no EcShardConfig — or one that failed to parse — +// yields (0, true): generation 0, which the fenced teardown preserves anyway (a +// recovered or pre-upgrade live volume). (0, false) means no .vif file was found. +// Both 0 and a missing .vif are preserved, so the read is fail-safe in every case. +func readEcGenerationTsNs(dataBaseFileName, indexBaseFileName string) (int64, bool) { + for _, base := range []string{dataBaseFileName, indexBaseFileName} { + if vi, _, found, _ := volume_info.MaybeLoadVolumeInfo(base + ".vif"); found { + return vi.GetEcShardConfig().GetEncodeTsNs(), true + } + if dataBaseFileName == indexBaseFileName { + break + } + } + return 0, false +} + // removeStaleEcArtifacts deletes the shard, index, journal, and bitrot sidecar // files of a prior encode so a fresh encode never mixes runs. total is the // shard-id range to scan (pass the cap for custom ratios). Returns the first @@ -663,7 +702,7 @@ func (vs *VolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_s glog.V(0).Infof("VolumeEcShardsUnmount: %v", req) for _, shardId := range req.ShardIds { - err := vs.store.UnmountEcShards(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(shardId)) + err := vs.store.UnmountEcShards(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(shardId), req.EncodeTsNs) if err != nil { glog.Errorf("ec shard unmount %v: %v", req, err) diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index 39b22f0e3..e676c1f11 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -45,6 +45,13 @@ func (l *DiskLocation) DestroyEcVolume(vid needle.VolumeId) { } } +// UnloadEcVolume drops the in-memory EcVolume for vid from this one disk without +// deleting files. Exported for the generation-fenced teardown, which unloads only +// the strictly-older disks rather than node-wide. +func (l *DiskLocation) UnloadEcVolume(vid needle.VolumeId) { + l.unloadEcVolume(vid) +} + // unloadEcVolume removes an EC volume from memory without deleting its files on disk. // This is useful for distributed EC volumes where shards may be on other servers. func (l *DiskLocation) unloadEcVolume(vid needle.VolumeId) { diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 690f81c6f..ac1c5d678 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -252,7 +252,7 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er return fmt.Errorf("MountEcShards %d.%d load failures: %s", vid, shardId, string(b)) } -func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.ShardId) error { +func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.ShardId, reqEncodeTsNs int64) error { // Walk every disk: a split-disk reconciled volume can mount the same vid on // more than one disk, so a first-match unmount would leave a sibling copy // mounted and heartbeating. Emit one deletion delta per disk. @@ -269,6 +269,14 @@ func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.Shar if ecVolume, ok := location.FindEcVolume(vid); ok { encodeTsNs = ecVolume.EncodeTsNs } + // Generation fence: when the caller carries a generation (stale-worker + // cleanup), only unmount a strictly-older generation; preserve a disk whose + // generation is same-or-newer, 0, or unknown, so a stale run cannot unmount + // a newer run's live shards. reqEncodeTsNs==0 (legacy/shell) unmounts all. + if reqEncodeTsNs > 0 && !(encodeTsNs > 0 && encodeTsNs < reqEncodeTsNs) { + glog.V(1).Infof("UnmountEcShards %d.%d disk_id:%d skipped: disk gen %d not older than request gen %d", vid, shardId, diskId, encodeTsNs, reqEncodeTsNs) + continue + } if deleted := location.UnloadEcShard(vid, shardId); deleted { si := erasure_coding.NewShardsInfo() si.Set(erasure_coding.NewShardInfo(shardId, 0)) diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index ad7b64279..a50d5182e 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -41,6 +41,7 @@ type ErasureCodingTask struct { dataShards int32 parityShards int32 sourceDiskType string // source volume's disk type, forwarded to Mount RPC (#9423) + encodeTsNs int64 // admin-issued encode generation; stamps the .vif and fences the stale-shard cleanup. 0 => unfenced (legacy/shell) targets []*worker_pb.TaskTarget // Unified targets for EC shards sources []*worker_pb.TaskSource // Unified sources for cleanup shardAssignment map[string][]string // destination -> assigned shard types @@ -79,6 +80,7 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP t.dataShards = ecParams.DataShards t.parityShards = ecParams.ParityShards t.sourceDiskType = ecParams.SourceDiskType + t.encodeTsNs = ecParams.EncodeTsNs t.workDir = ecParams.WorkingDir t.targets = params.Targets // Get unified targets t.sources = params.Sources // Get unified sources @@ -625,10 +627,17 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string // not pass to the encoder). vifFile := baseName + ".vif" defaultCtx := erasure_coding.NewDefaultECContext("", 0) + // Use the admin-issued generation when present so the distributed .vif carries + // the same generation the stale-shard cleanup fences on; fall back to a local + // timestamp only for the unfenced legacy/shell path (keeps the read guard on). + encodeTsNs := t.encodeTsNs + if encodeTsNs == 0 { + encodeTsNs = time.Now().UnixNano() + } ecShardConfig := &volume_server_pb.EcShardConfig{ DataShards: uint32(defaultCtx.DataShards), ParityShards: uint32(defaultCtx.ParityShards), - EncodeTsNs: time.Now().UnixNano(), + EncodeTsNs: encodeTsNs, } if ecBitrot != nil && ecBitrot.EcShardConfig != nil { ecShardConfig.DataShards = ecBitrot.EcShardConfig.DataShards @@ -938,7 +947,7 @@ func (t *ErasureCodingTask) cleanupStaleEcShards(ctx context.Context) error { "shard_ids": allShards, }).Info("Clearing stale EC shards on destination before re-distribute") - if err := unmountAndDeleteEcShards(ctx, t.grpcDialOption, node, t.volumeID, t.collection, allShards); err != nil { + if err := unmountAndDeleteEcShards(ctx, t.grpcDialOption, node, t.volumeID, t.collection, allShards, t.encodeTsNs); err != nil { cleanupErrors = append(cleanupErrors, fmt.Sprintf("%s: %v", node, err)) t.GetLogger().WithFields(map[string]interface{}{ "volume_id": t.volumeID, @@ -983,12 +992,17 @@ func unmountAndDeleteEcShards( volumeID uint32, collection string, shardIds []uint32, + encodeTsNs int64, ) error { return operation.WithVolumeServerClient(false, pb.ServerAddress(destination), dialOption, func(client volume_server_pb.VolumeServerClient) error { + // encodeTsNs fences both RPCs against a newer run on a shared node: the + // server skips a disk whose mounted/on-disk generation is same-or-newer. + // 0 (legacy/shell) leaves the unconditional unmount + blanket teardown. if _, err := client.VolumeEcShardsUnmount(ctx, &volume_server_pb.VolumeEcShardsUnmountRequest{ - VolumeId: volumeID, - ShardIds: shardIds, + VolumeId: volumeID, + ShardIds: shardIds, + EncodeTsNs: encodeTsNs, }); err != nil { return fmt.Errorf("unmount: %w", err) } @@ -997,6 +1011,7 @@ func unmountAndDeleteEcShards( Collection: collection, ShardIds: shardIds, FullTeardown: true, + EncodeTsNs: encodeTsNs, }) if err != nil { return fmt.Errorf("delete: %w", err) diff --git a/weed/worker/tasks/erasure_coding/ec_task_test.go b/weed/worker/tasks/erasure_coding/ec_task_test.go index a8e809d46..56dbc9d20 100644 --- a/weed/worker/tasks/erasure_coding/ec_task_test.go +++ b/weed/worker/tasks/erasure_coding/ec_task_test.go @@ -170,6 +170,54 @@ func TestGenerateEcShardsLocallyStampsEncodeIdentity(t *testing.T) { require.NotZero(t, vi.EcShardConfig.EncodeTsNs, "worker-encoded .vif must carry an encode identity") } +// A fenced run must stamp the admin-issued generation (t.encodeTsNs) verbatim into +// the distributed .vif, so the generation the worker distributes equals the one it +// later sends on the teardown RPCs and the volume-server fence compares like-for-like. +func TestGenerateEcShardsLocallyUsesAdminGeneration(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + const volumeID = uint32(957) + framework.AllocateVolume(t, grpcClient, volumeID, "") + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 2002, 0x5555EEEE) + resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("payload-for-admin-generation")) + _ = framework.ReadAllAndClose(t, resp) + require.Equal(t, http.StatusCreated, resp.StatusCode) + + task := NewErasureCodingTask( + "ec-admin-generation", + clusterHarness.VolumeServerAddress(), + volumeID, + "", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + const adminGeneration int64 = 1_700_000_000_000_000_321 + task.encodeTsNs = adminGeneration + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + require.NoError(t, task.markReplicasReadonly(ctx)) + localFiles, err := task.copyVolumeFilesToWorker(ctx, t.TempDir()) + require.NoError(t, err) + + shardFiles, err := task.generateEcShardsLocally(localFiles, t.TempDir()) + require.NoError(t, err) + + vi, _, found, err := volume_info.MaybeLoadVolumeInfo(shardFiles["vif"]) + require.NoError(t, err) + require.True(t, found) + require.NotNil(t, vi.EcShardConfig) + require.Equal(t, adminGeneration, vi.EcShardConfig.EncodeTsNs, "fenced run must stamp the admin generation verbatim, not a local timestamp") +} + func compactVolumeOnce(t *testing.T, grpcClient volume_server_pb.VolumeServerClient, volumeID uint32) { t.Helper() diff --git a/weed/worker/tasks/erasure_coding/plugin_handler.go b/weed/worker/tasks/erasure_coding/plugin_handler.go index e0fed1fa4..f62fea820 100644 --- a/weed/worker/tasks/erasure_coding/plugin_handler.go +++ b/weed/worker/tasks/erasure_coding/plugin_handler.go @@ -241,6 +241,18 @@ func (h *ErasureCodingHandler) Detect( if err != nil { return err } + // Stamp the admin-issued encode generation onto every EC proposal. DetectionSequence + // is minted once per cycle on the single admin clock, so generations are globally + // ordered even though detection runs on a rotating worker; this lets a stale worker's + // shard cleanup fence against a newer run instead of wiping it. + for _, result := range results { + if result == nil || result.TypedParams == nil { + continue + } + if ecp := result.TypedParams.GetErasureCodingParams(); ecp != nil { + ecp.EncodeTsNs = request.DetectionSequence + } + } if traceErr := emitErasureCodingDetectionDecisionTrace(sender, metrics, workerConfig.TaskConfig, results, maxResults, hasMore); traceErr != nil { glog.Warningf("Plugin worker failed to emit erasure_coding detection trace: %v", traceErr) }