From f69b7b854cd0a5ac973ba4ef1aed0b7c67e9bc31 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 9 Jun 2026 09:42:23 -0700 Subject: [PATCH] feat(ec): mirror the encode-run identity guard + full_teardown into the Rust volume server The Go volume server stamps an encode-run identity (encode_ts_ns) into the .vif and rejects a read served from a shard of a different run; full_teardown wipes a whole generation and acknowledges it. The Rust volume server had none of it. Mirror the shared logic: load encode_ts_ns from the .vif onto the EcVolume, stamp it on every read response, and reject a request/response mismatch on both the server and the distributed-read client (recovering from parity); handle full_teardown by evicting the volume and wiping every EC artifact on each disk, echoing full_teardown_done so the caller can detect a server that ignored it. --- seaweed-volume/proto/volume_server.proto | 6 +++ seaweed-volume/src/server/grpc_server.rs | 48 ++++++++++++++++++- seaweed-volume/src/server/store_ec.rs | 27 +++++++++++ .../src/storage/erasure_coding/ec_volume.rs | 24 ++++++++-- seaweed-volume/src/storage/store_ec_mirror.rs | 1 + .../src/storage/store_ec_reconcile.rs | 2 + seaweed-volume/src/storage/volume.rs | 6 +++ 7 files changed, 109 insertions(+), 5 deletions(-) diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index c63177cb4..7be0ed6ac 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -450,8 +450,10 @@ message VolumeEcShardsDeleteRequest { uint32 volume_id = 1; 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 } 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 } message VolumeEcShardsMountRequest { @@ -476,10 +478,13 @@ message VolumeEcShardReadRequest { int64 offset = 3; int64 size = 4; uint64 file_key = 5; + reserved 6; + int64 encode_ts_ns = 7; // caller's expected encode time; the server rejects a shard from a different encode run } message VolumeEcShardReadResponse { bytes data = 1; bool is_deleted = 2; + int64 encode_ts_ns = 3; // identity of the shard actually served; client rejects a mismatch (0 = pre-upgrade server) } message VolumeEcBlobDeleteRequest { @@ -577,6 +582,7 @@ message VolumeInfo { message EcShardConfig { uint32 data_shards = 1; // Number of data shards (e.g., 10) uint32 parity_shards = 2; // Number of parity shards (e.g., 4) + int64 encode_ts_ns = 3; // encode time (unix nanos); a read served from a shard of a different encode run is rejected } message OldVersionVolumeInfo { repeated RemoteFile files = 1; diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 4e84ddec8..3df3e0a5f 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -2133,6 +2133,12 @@ impl VolumeServer for VolumeGrpcService { ec_shard_config: Some(crate::storage::volume::VifEcShardConfig { data_shards: data_shards, parity_shards: parity_shards, + // This run's identity; the read path rejects a shard from a + // different encode run. + encode_ts_ns: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() as i64, }), ..Default::default() }; @@ -2593,12 +2599,35 @@ impl VolumeServer for VolumeGrpcService { self.state.check_maintenance()?; let req = request.into_inner(); 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. + { + let mut store = self.state.store.write().unwrap(); + let _ = store.remove_ec_volume(vid); + for loc in &store.locations { + loc.remove_ec_volume_files(&req.collection, vid); + } + } + self.state.volume_state_notify.notify_one(); + return Ok(Response::new( + volume_server_pb::VolumeEcShardsDeleteResponse { + full_teardown_done: true, + }, + )); + } + let mut store = self.state.store.write().unwrap(); store.delete_ec_shards(vid, &req.collection, &req.shard_ids); drop(store); self.state.volume_state_notify.notify_one(); Ok(Response::new( - volume_server_pb::VolumeEcShardsDeleteResponse {}, + volume_server_pb::VolumeEcShardsDeleteResponse { + full_teardown_done: false, + }, )) } @@ -2672,6 +2701,21 @@ impl VolumeServer for VolumeGrpcService { )) })?; + // Reject a shard whose identity doesn't match the caller's index; the caller + // then recovers from parity. Lenient only when the caller has no identity + // (pre-upgrade reader): a known caller must not accept an unstamped holder, + // which would serve a stale shard from a different encode run. + if req.encode_ts_ns != 0 && req.encode_ts_ns != ec_vol.encode_ts_ns { + return Err(Status::failed_precondition(format!( + "ec shard {}.{} belongs to a different encode run", + req.volume_id, req.shard_id + ))); + } + + // Identity of the shard actually served, echoed on every response chunk so + // the client can reject a different encode run even from a pre-upgrade server. + let served_encode_ts_ns = ec_vol.encode_ts_ns; + // Check if the requested needle is deleted (via .ecx index, matching Go) if req.file_key > 0 { let needle_id = NeedleId(req.file_key); @@ -2682,6 +2726,7 @@ impl VolumeServer for VolumeGrpcService { if size.is_deleted() { let results = vec![Ok(volume_server_pb::VolumeEcShardReadResponse { is_deleted: true, + encode_ts_ns: served_encode_ts_ns, ..Default::default() })]; return Ok(Response::new(Box::pin(tokio_stream::iter(results)))); @@ -2730,6 +2775,7 @@ impl VolumeServer for VolumeGrpcService { results.push(Ok(volume_server_pb::VolumeEcShardReadResponse { data: buf, is_deleted: false, + encode_ts_ns: served_encode_ts_ns, })); if n < chunk_size { break; // short read means EOF diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 37d957458..a71eb14d9 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -70,6 +70,10 @@ struct Snapshot { intervals: Vec, cached_locations: HashMap>, cache_refreshed_at: Option, + /// This volume's encode identity, carried to peers on remote shard reads so a + /// shard from a different encode run is rejected rather than served at a + /// mismatched offset. 0 for a pre-feature volume (lenient). + encode_ts_ns: i64, } /// Top-level entry point. Returns `Ok(None)` for "not found" (matches @@ -147,6 +151,7 @@ pub async fn read_ec_shard_needle_distributed( &shard_locations, snapshot.data_shards as usize, snapshot.parity_shards as usize, + snapshot.encode_ts_ns, ) .await?; assembled.push(buf); @@ -259,6 +264,7 @@ fn snapshot_under_lock( intervals: interval_results, cached_locations, cache_refreshed_at, + encode_ts_ns: ecv.encode_ts_ns, })) } @@ -383,6 +389,7 @@ async fn fetch_one_interval( shard_locations: &HashMap>, data_shards: usize, parity_shards: usize, + expected_encode_ts_ns: i64, ) -> io::Result> { // Direct peer read against the cached locations for this shard. if let Some(sources) = shard_locations.get(&shard_id) { @@ -395,6 +402,7 @@ async fn fetch_one_interval( shard_id, shard_offset, size, + expected_encode_ts_ns, ) .await { @@ -424,6 +432,7 @@ async fn fetch_one_interval( shard_locations, data_shards, parity_shards, + expected_encode_ts_ns, ) .await } @@ -436,6 +445,7 @@ async fn read_remote_ec_shard_interval( shard_id: ShardId, shard_offset: i64, size: usize, + expected_encode_ts_ns: i64, ) -> io::Result> { let mut last_err: Option = None; for src in sources { @@ -447,6 +457,7 @@ async fn read_remote_ec_shard_interval( shard_id, shard_offset, size, + expected_encode_ts_ns, ) .await { @@ -470,6 +481,7 @@ async fn do_read_remote_ec_shard_interval( shard_id: ShardId, shard_offset: i64, size: usize, + expected_encode_ts_ns: i64, ) -> io::Result> { let grpc_addr = parse_grpc_address(source).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; @@ -505,6 +517,7 @@ async fn do_read_remote_ec_shard_interval( offset: shard_offset, size: size as i64, file_key: needle_id.0, + encode_ts_ns: expected_encode_ts_ns, }; let resp = client .volume_ec_shard_read(Request::new(req)) @@ -523,6 +536,18 @@ async fn do_read_remote_ec_shard_interval( .await .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("recv: {}", e)))? { + // Validate the served shard's identity client-side, so the guard holds even + // against a pre-upgrade server that ignored the request field (returns 0). + // A mismatch fails the read; the caller recovers from parity. + if expected_encode_ts_ns != 0 && msg.encode_ts_ns != expected_encode_ts_ns { + return Err(io::Error::new( + io::ErrorKind::Other, + format!( + "ec shard {}.{} from {} belongs to a different encode run (want {} got {})", + vid.0, shard_id, source, expected_encode_ts_ns, msg.encode_ts_ns + ), + )); + } if !msg.data.is_empty() { out.extend_from_slice(&msg.data); } @@ -554,6 +579,7 @@ async fn recover_one_remote_ec_shard_interval( shard_locations: &HashMap>, data_shards: usize, parity_shards: usize, + expected_encode_ts_ns: i64, ) -> io::Result> { let total_shards = data_shards + parity_shards; let rs = ReedSolomon::new(data_shards, parity_shards).map_err(|e| { @@ -614,6 +640,7 @@ async fn recover_one_remote_ec_shard_interval( sid, shard_offset, size, + expected_encode_ts_ns, ) .await; (sid, res) diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 5d137542a..77d095a4c 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -57,6 +57,10 @@ pub struct EcVolume { pub shard_locations_refresh_time: std::sync::Mutex>, /// EC volume expiration time (unix epoch seconds), set during EC encode from TTL. pub expire_at_sec: u64, + /// Encode-run identity (unix nanos) loaded from the .vif EcShardConfig. A read + /// served from a shard of a different encode run is rejected (server- and + /// client-side); 0 for a pre-feature volume, which is treated leniently. + pub encode_ts_ns: i64, } /// Locate the `.vif` for a (collection, vid) by preferring the data dir @@ -137,7 +141,7 @@ impl EcVolume { // for shard-size math and by the Store-level prune in // `store_ec_reconcile.rs` to verify a sibling-disk .dat is // plausibly the encoding source (#9478). - let (expire_at_sec, vif_version, vif_dat_file_size) = { + let (expire_at_sec, vif_version, vif_dat_file_size, encode_ts_ns) = { let vif_path = locate_vif_path(dir, dir_idx, collection, volume_id); if let Ok(vif_content) = std::fs::read_to_string(&vif_path) { if let Ok(vif_info) = @@ -148,12 +152,21 @@ impl EcVolume { } else { Version::current() }; - (vif_info.expire_at_sec, ver, vif_info.dat_file_size) + let cfg_encode_ts_ns = vif_info + .ec_shard_config + .as_ref() + .map_or(0, |c| c.encode_ts_ns); + ( + vif_info.expire_at_sec, + ver, + vif_info.dat_file_size, + cfg_encode_ts_ns, + ) } else { - (0, Version::current(), 0) + (0, Version::current(), 0, 0) } } else { - (0, Version::current(), 0) + (0, Version::current(), 0, 0) } }; @@ -177,6 +190,7 @@ impl EcVolume { shard_locations: std::sync::RwLock::new(HashMap::new()), shard_locations_refresh_time: std::sync::Mutex::new(None), expire_at_sec, + encode_ts_ns, }; // Open .ecx file (sorted index) in read/write mode for in-place deletion marking. @@ -1243,6 +1257,7 @@ mod tests { ec_shard_config: Some(crate::storage::volume::VifEcShardConfig { data_shards: 6, parity_shards: 3, + ..Default::default() }), ..Default::default() }; @@ -1268,6 +1283,7 @@ mod tests { ec_shard_config: Some(crate::storage::volume::VifEcShardConfig { data_shards: 10, parity_shards: 10, + ..Default::default() }), ..Default::default() }; diff --git a/seaweed-volume/src/storage/store_ec_mirror.rs b/seaweed-volume/src/storage/store_ec_mirror.rs index 3b93b345a..77b3af56c 100644 --- a/seaweed-volume/src/storage/store_ec_mirror.rs +++ b/seaweed-volume/src/storage/store_ec_mirror.rs @@ -327,6 +327,7 @@ mod tests { ec_shard_config: Some(VifEcShardConfig { data_shards, parity_shards, + ..Default::default() }), ..Default::default() }; diff --git a/seaweed-volume/src/storage/store_ec_reconcile.rs b/seaweed-volume/src/storage/store_ec_reconcile.rs index f42934fae..002adbd33 100644 --- a/seaweed-volume/src/storage/store_ec_reconcile.rs +++ b/seaweed-volume/src/storage/store_ec_reconcile.rs @@ -459,6 +459,7 @@ mod tests { ec_shard_config: Some(VifEcShardConfig { data_shards, parity_shards, + ..Default::default() }), ..Default::default() }; @@ -1078,6 +1079,7 @@ mod tests { ec_shard_config: Some(VifEcShardConfig { data_shards: 10, parity_shards: 4, + ..Default::default() }), ..Default::default() }; diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 056db6491..c92c19fdb 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -189,6 +189,10 @@ pub struct VifEcShardConfig { pub data_shards: u32, #[serde(default, rename = "parityShards")] pub parity_shards: u32, + /// Encode time (unix nanos). Shards and .ecx of one encode run share it, + /// so a read served from a different run's shard is rejected. + #[serde(default, rename = "encodeTsNs", with = "string_or_i64")] + pub encode_ts_ns: i64, } /// Serde-compatible representation of OldVersionVolumeInfo for legacy .vif JSON deserialization. @@ -279,6 +283,7 @@ impl VifVolumeInfo { ec_shard_config: pb.ec_shard_config.as_ref().map(|c| VifEcShardConfig { data_shards: c.data_shards, parity_shards: c.parity_shards, + encode_ts_ns: c.encode_ts_ns, }), } } @@ -309,6 +314,7 @@ impl VifVolumeInfo { crate::pb::volume_server_pb::EcShardConfig { data_shards: c.data_shards, parity_shards: c.parity_shards, + encode_ts_ns: c.encode_ts_ns, } }), }