volume server: sweep stale EC artifacts before VolumeEcShardsGenerate re-encodes (#11413)

* volume server: sweep stale EC artifacts before VolumeEcShardsGenerate re-encodes

The Rust VolumeEcShardsGenerate went straight into write_ec_files: no unload
of an already-mounted EC volume and no stale-artifact sweep. Only .ec00..ecNN
on the encoding disk were truncated, so a retry could mix two encode runs. A
stale N.ec03 left on a sibling disk survived, reconcile later mounted it
against the new .ecx, and the new .vif made the encode_ts_ns identity guard
pass, so reads served old-run bytes at new-run offsets.

Mirror Go's VolumeEcShardsGenerate (#9880 / #9953): UnloadEcVolume on every
disk, then removeStaleEcArtifacts on every disk location before encoding.
remove_ec_volume_files_full_teardown already has removeStaleEcArtifacts'
semantics (.ec00..ec31, .ecx/.ecj/.ecsum[.vN] in both the data and idx dirs,
.vif only on a shard-only disk; never the source .dat/.idx), so reuse it. Add
Store::unload_ec_volume, which unlike remove_ec_volume does not stop at the
first disk and closes the descriptors so the unlink frees the inodes. The
store write lock covers only unload + sweep, not the encode.

The failure arm now also drops the generation-0 .ecsum, as Go's defer does.

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

* volume server: wake the heartbeat after VolumeEcShardsGenerate unloads shards

The pre-encode unload drops mounted EC shards from memory, but unlike every
other unmount path it did not wake the heartbeat, so the master kept routing
reads to shards this server no longer serves until the next pulse. Notify
once the store lock is released, and before the sweep error propagates: a
failed sweep has unloaded the shards too.

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

* volume server: clean up encode artifacts when the .vif write fails too

Go's shouldCleanup defer covers every error before the .vif commits,
not just a failed encode. A serialize or write failure on the .vif left
the fresh .ecNN/.ecx/.ecsum behind, which the next generate would have
to rely on the new sweep to remove. Extract the cleanup and run it on
the .vif error paths as well.

* volume server: write the EC .vif atomically

Go's SaveVolumeInfo writes a temp file, syncs it, and renames it over
the target, so a failed write leaves the previous metadata intact and a
read-only .vif fails the save. The direct fs::write truncated the file
first, so a write or sync failure could leave an empty .vif even after
cleanup_encode removed the generated shards.

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

seaweed-volume/src/server/grpc_server.rs

@@ -3124,6 +3124,45 @@ impl VolumeServer for VolumeGrpcService {
                 tonic::Status::internal(format!("read ec shard config for volume {}: {}", vid.0, e))
             })?;
 
+        // Wipe any EC artifacts from a prior encode so a retry never mixes two
+        // runs (Go's UnloadEcVolume + removeStaleEcArtifacts). Evict the
+        // in-memory EcVolume first so the unlink frees the inodes instead of
+        // leaving open fds serving the old bytes, and sweep every disk: a
+        // stale shard on a sibling disk would otherwise be mounted against the
+        // new .ecx at reconcile. The source .dat/.idx/.vif are kept.
+        let swept = {
+            let mut store = self.state.store.write().unwrap();
+            store.unload_ec_volume(vid);
+            store.locations.iter().try_for_each(|loc| {
+                loc.remove_ec_volume_files_full_teardown(collection, vid)
+                    .map_err(|e| {
+                        Status::internal(format!(
+                            "wipe stale EC artifacts for volume {} on {}: {}",
+                            vid.0, loc.directory, e
+                        ))
+                    })
+            })
+        };
+        // The unload dropped mounted shards, so wake the heartbeat now rather
+        // than leaving the master to route reads here until the next pulse. A
+        // failed sweep has unloaded them too, hence before the `?`.
+        self.state.volume_state_notify.notify_one();
+        swept?;
+
+        // On any failure before the .vif is committed, remove the freshly
+        // written .ecNN, .ecx and generation-0 .ecsum (matching Go's deferred
+        // cleanup) so a retry never starts from this run's leftovers.
+        let cleanup_encode = || {
+            let base = crate::storage::volume::volume_file_name(&dir, collection, vid);
+            for i in 0..(data_shards + parity_shards) {
+                let _ = std::fs::remove_file(format!("{}.ec{:02}", base, i));
+            }
+            let _ = std::fs::remove_file(format!("{}.ecx", base));
+            let _ = std::fs::remove_file(
+                crate::storage::erasure_coding::ec_bitrot::bitrot_sidecar_path(&base, 0),
+            );
+        };
+
         let block_size = match crate::storage::erasure_coding::ec_encoder::write_ec_files(
             &dir,
             &idx_dir,
@@ -3134,14 +3173,7 @@ impl VolumeServer for VolumeGrpcService {
         ) {
             Ok(block_size) => block_size,
             Err(e) => {
-                // Cleanup partially-created .ecNN and .ecx files on failure (matching Go defer)
-                let base = crate::storage::volume::volume_file_name(&dir, collection, vid);
-                let total_shards = data_shards + parity_shards;
-                for i in 0..total_shards {
-                    let shard_path = format!("{}.ec{:02}", base, i);
-                    let _ = std::fs::remove_file(&shard_path);
-                }
-                let _ = std::fs::remove_file(format!("{}.ecx", base));
+                cleanup_encode();
                 return Err(Status::internal(e.to_string()));
             }
         };
@@ -3167,10 +3199,10 @@ impl VolumeServer for VolumeGrpcService {
                 }),
                 ..Default::default()
             };
-            let content = serde_json::to_string_pretty(&vif)
-                .map_err(|e| Status::internal(format!("serialize vif: {}", e)))?;
-            std::fs::write(&vif_path, content)
-                .map_err(|e| Status::internal(format!("write vif: {}", e)))?;
+            if let Err(e) = crate::storage::store::save_vif_volume_info(&vif_path, &vif) {
+                cleanup_encode();
+                return Err(Status::internal(format!("write vif: {}", e)));
+            }
         }
 
         Ok(Response::new(
@@ -7803,6 +7835,124 @@ mod tests {
         assert!(vif.expire_at_sec <= before + ttl.to_seconds() + 5);
     }
 
+    /// A re-encode must not mix artifacts from a previous run. Generate used to
+    /// go straight into the encode, truncating only the `.ecNN` files on the
+    /// encoding disk; a stale shard on a sibling disk survived and was later
+    /// mounted against the new `.ecx`. Go evicts the EcVolume and sweeps every
+    /// disk first (UnloadEcVolume + removeStaleEcArtifacts).
+    #[tokio::test]
+    async fn test_volume_ec_shards_generate_sweeps_stale_artifacts_on_every_disk() {
+        let (service, tmp) = make_local_service_with_volume("", None);
+        let sibling = TempDir::new().unwrap();
+        let sibling_dir = sibling.path().to_str().unwrap();
+        service
+            .state
+            .store
+            .write()
+            .unwrap()
+            .add_location(
+                sibling_dir,
+                sibling_dir,
+                10,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+
+        let generate = || {
+            service.volume_ec_shards_generate(Request::new(
+                volume_server_pb::VolumeEcShardsGenerateRequest {
+                    volume_id: 1,
+                    collection: String::new(),
+                },
+            ))
+        };
+        generate().await.unwrap();
+        service
+            .volume_ec_shards_mount(Request::new(volume_server_pb::VolumeEcShardsMountRequest {
+                volume_id: 1,
+                collection: String::new(),
+                shard_ids: (0..14).collect(),
+                source_disk_type: String::new(),
+                recover_missing_index: false,
+            }))
+            .await
+            .unwrap();
+        assert!(
+            service
+                .state
+                .store
+                .read()
+                .unwrap()
+                .has_ec_volume(VolumeId(1))
+        );
+
+        // A prior run's leftovers on the sibling disk. `.ec20` is outside the
+        // 10+4 ratio: the sweep scans the shard-id cap, not the current total.
+        let stale = [
+            ".ec03",
+            ".ec20",
+            ".ecx",
+            ".ecj",
+            ".ecsum",
+            ".ecsum.v2",
+            ".vif",
+        ];
+        for ext in stale {
+            std::fs::write(sibling.path().join(format!("1{}", ext)), b"stale-run").unwrap();
+        }
+
+        // Drain the permit the mount above left, so the wake-up asserted below
+        // can only come from the re-generate's unload.
+        let notify = &service.state.volume_state_notify;
+        let _ = tokio::time::timeout(std::time::Duration::ZERO, notify.notified()).await;
+
+        generate().await.unwrap();
+
+        assert!(
+            tokio::time::timeout(std::time::Duration::ZERO, notify.notified())
+                .await
+                .is_ok(),
+            "unloading the mounted shards must wake the heartbeat"
+        );
+
+        for ext in stale {
+            assert!(
+                !sibling.path().join(format!("1{}", ext)).exists(),
+                "stale 1{} on the sibling disk must not survive a re-generate",
+                ext
+            );
+        }
+        assert!(
+            !service
+                .state
+                .store
+                .read()
+                .unwrap()
+                .has_ec_volume(VolumeId(1)),
+            "the mounted EC volume must be unloaded before its files are swept"
+        );
+        // The source volume and the fresh run stay on the encoding disk.
+        for ext in [".dat", ".idx", ".vif", ".ecx", ".ecsum", ".ec00", ".ec13"] {
+            assert!(
+                tmp.path().join(format!("1{}", ext)).exists(),
+                "1{} must exist on the encoding disk after a re-generate",
+                ext
+            );
+        }
+        assert!(
+            service
+                .state
+                .store
+                .read()
+                .unwrap()
+                .find_volume(VolumeId(1))
+                .is_some(),
+            "the source volume stays loaded"
+        );
+    }
+
     /// REGRESSION: a node-wide scrub must survive a volume that legitimately
     /// disappears while it runs.
     ///

seaweed-volume/src/storage/disk_location.rs

@@ -801,6 +801,20 @@ impl DiskLocation {
         self.ec_volumes.remove(&vid)
     }
 
+    /// Drop the in-memory EC volume for vid and close its descriptors without
+    /// deleting files, so a following unlink frees the inodes instead of
+    /// leaving open fds serving the old bytes. Mirrors Go's unloadEcVolume.
+    pub fn unload_ec_volume(&mut self, vid: VolumeId) {
+        if let Some(mut ec_vol) = self.ec_volumes.remove(&vid) {
+            for _ in 0..ec_vol.shard_count() {
+                crate::metrics::VOLUME_GAUGE
+                    .with_label_values(&[&ec_vol.collection, "ec_shards"])
+                    .dec();
+            }
+            ec_vol.close();
+        }
+    }
+
     /// Mount EC shards for a volume on this location.
     ///
     /// `source_disk_type` is the source volume's disk type carried on the

seaweed-volume/src/storage/store.rs

@@ -1268,6 +1268,16 @@ impl Store {
         None
     }
 
+    /// Drop any in-memory EC volume for vid from EVERY disk and close its
+    /// descriptors without deleting files. Unlike remove_ec_volume this does not
+    /// stop at the first disk: a split-disk volume is registered on each disk
+    /// holding a shard. Mirrors Go's Store.UnloadEcVolume.
+    pub fn unload_ec_volume(&mut self, vid: VolumeId) {
+        for loc in &mut self.locations {
+            loc.unload_ec_volume(vid);
+        }
+    }
+
     /// Find the location index containing EC files for a volume.
     pub fn find_ec_location(&self, vid: VolumeId, collection: &str) -> Option<usize> {
         for (i, loc) in self.locations.iter().enumerate() {
@@ -1544,10 +1554,35 @@ fn load_vif_volume_info(path: &str) -> Result<VifVolumeInfo, VolumeError> {
     )))
 }
 
-fn save_vif_volume_info(path: &str, info: &VifVolumeInfo) -> Result<(), VolumeError> {
+/// Mirrors Go's SaveVolumeInfo: a read-only .vif fails the save, and the
+/// file is replaced atomically so a failed write keeps the previous
+/// metadata intact.
+pub(crate) fn save_vif_volume_info(path: &str, info: &VifVolumeInfo) -> Result<(), VolumeError> {
+    if std::fs::metadata(path).is_ok_and(|m| m.permissions().readonly()) {
+        return Err(VolumeError::Io(io::Error::new(
+            io::ErrorKind::PermissionDenied,
+            format!("failed to check {} not writable", path),
+        )));
+    }
     let content = serde_json::to_string_pretty(info)
         .map_err(|e| VolumeError::Io(io::Error::other(e.to_string())))?;
-    std::fs::write(path, content)?;
+    static TMP_SEQ: AtomicU64 = AtomicU64::new(0);
+    let tmp = format!(
+        "{}.tmp.{}.{}",
+        path,
+        std::process::id(),
+        TMP_SEQ.fetch_add(1, Ordering::Relaxed)
+    );
+    let write = (|| -> io::Result<()> {
+        use std::io::Write;
+        let mut f = std::fs::File::create(&tmp)?;
+        f.write_all(content.as_bytes())?;
+        f.sync_all()
+    })();
+    if let Err(e) = write.and_then(|()| std::fs::rename(&tmp, path)) {
+        let _ = std::fs::remove_file(&tmp);
+        return Err(e.into());
+    }
     Ok(())
 }