fix(ec): fence stale-worker EC shard cleanup by encode generation (#9953)

* feat(ec): add encode_ts_ns to the EC task params, shard-unmount, and shard-delete RPCs

The generation fence for stale EC-worker cleanup needs the encode
generation on three messages: ErasureCodingTaskParams (admin issues it),
VolumeEcShardsUnmountRequest, and VolumeEcShardsDeleteRequest (the worker
carries it to the volume server). Additive fields only; 0 preserves the
existing unfenced behavior. Mirror the two volume-server fields in the
Rust volume server's proto copy.

* feat(ec): issue the EC encode generation from the admin and carry it on the worker

Stamp each EC proposal's encode_ts_ns from the admin's per-cycle
DetectionSequence (a single-clock value) so generations are globally
ordered even though detection runs on a rotating worker. The worker
writes that generation into the distributed .vif and passes it on its
shard unmount/delete RPCs; it falls back to a local timestamp for the
.vif only on the unfenced legacy/shell path (keeping the read guard on).

* fix(ec): fence the stale-worker EC shard unmount and teardown by generation

A reaped-but-still-running EC worker's cleanupStaleEcShards issued a
generation-blind unmount + full teardown that could unmount and then
overwrite a newer run's live shards on a shared node. Both RPCs now
carry the encode generation: the volume server unmounts/deletes a disk
only when its .vif generation is strictly older than the request, and
preserves a same-or-newer generation, a generation-0 (recovered or
pre-upgrade) volume, and an unreadable .vif. Unload is per-disk, never
node-wide. Request generation 0 keeps the blanket teardown for the shell
pre-encode cleanup and pre-upgrade callers. Mirrored in the Rust volume
server.

* test(ec): cover the generation-fenced teardown and unmount

End-to-end volume-server tests: a fenced FullTeardown wipes a strictly-
older generation, preserves a newer one, preserves a generation-0 volume,
and blanket-wipes on request generation 0; the gen-aware unmount preserves
a same-or-newer mounted generation; and the .vif generation reader handles
present/absent/no-config cases.

* test(ec): pin the fenced .vif==teardown generation and the unreadable-.vif preserve

A fenced run must stamp the admin generation verbatim into the .vif so it
matches the generation sent on the teardown RPCs; add a regression test
that sets the task generation and asserts the .vif carries it exactly.
Also cover the present-but-unparseable .vif case (reads as generation 0,
preserved) and correct the readEcGenerationTsNs docstring accordingly.

* fix(ec): surface EC full-teardown filesystem errors in the Rust volume server

remove_ec_volume_files(_full_teardown) discarded every fs::remove_file
error, so a teardown that failed on permissions or a full disk still
returned full_teardown_done=true and left stale artifacts to collide with
the next encode. Return io::Result, ignore NotFound, propagate the first
real error, and have the teardown RPC surface it -- matching the Go
contract. The best-effort reconcile/load-cleanup callers keep ignoring it.

* refactor(ec): reuse the EC volume lookup on unmount and short-circuit the gen read

Address review: the Rust unmount fence reuses the ec_vol it already
fetched instead of a second find_ec_volume; the Go .vif generation reader
breaks out of the data/idx loop early when the two dirs are the same.

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)

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

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