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

@@ -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;

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

seaweed-volume/src/server/store_ec.rs

@@ -70,6 +70,10 @@ struct Snapshot {
     intervals: Vec<IntervalResult>,
     cached_locations: HashMap<ShardId, Vec<String>>,
     cache_refreshed_at: Option<Instant>,
+    /// 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<ShardId, Vec<String>>,
     data_shards: usize,
     parity_shards: usize,
+    expected_encode_ts_ns: i64,
 ) -> io::Result<Vec<u8>> {
     // 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<Vec<u8>> {
     let mut last_err: Option<io::Error> = 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<Vec<u8>> {
     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<ShardId, Vec<String>>,
     data_shards: usize,
     parity_shards: usize,
+    expected_encode_ts_ns: i64,
 ) -> io::Result<Vec<u8>> {
     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)

seaweed-volume/src/storage/erasure_coding/ec_volume.rs

@@ -57,6 +57,10 @@ pub struct EcVolume {
     pub shard_locations_refresh_time: std::sync::Mutex<Option<std::time::Instant>>,
     /// 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()
         };

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()
         };

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()
         };

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,
                 }
             }),
         }