fix(ec): group orphan-source completeness by encode generation (topology encode_ts_ns) (#9952)

* feat(ec): carry the encode generation through the topology heartbeat

Add encode_ts_ns (field 14) to VolumeEcShardInformationMessage and
populate it from each EC volume's .vif identity. The volume server emits
it on the full and incremental heartbeats; the master stores it on
EcVolumeInfo and re-emits it via GetTopologyInfo, so the admin/worker
layer can see which encode run produced each shard set. Field 14 avoids
the enterprise fork's reserved 10-13. Mirror the proto field and both
heartbeat emit sites in the Rust volume server.

* fix(ec): group orphan-source shard completeness by encode generation

countExistingEcShardsForVolume ORed EcIndexBits across every disk, so two
interrupted encode runs whose shard sets overlap unioned into a
false-complete set -- triggering the orphaned-source delete while no
single generation was actually complete. Group shards by encode_ts_ns and
return the largest single generation's count, so the trigger fires only
when one run holds the full set. Shards from pre-upgrade servers
(encode_ts_ns==0) form their own bucket.

The heartbeat carries one encode_ts_ns per (volume, disk), so this
separates generations on different disks; same-disk mixing is prevented
upstream by the pre-encode artifact wipe and the cross-run read guard.

* fix(ec): guard against a nil Ec shard info entry in the generation count

Defensive: a manually-constructed or corrupted topology could carry a nil
entry in EcShardInfos. Skip it rather than dereference.

* fix(ec): carry the encode generation on the EC shard unmount delta

The mount delta sets EncodeTsNs; the unmount deletion delta left it 0.
Populate it from the Ec volume before unloading so both incremental
deltas are consistent (the Rust volume server already does this via its
snapshot diff).

seaweed-volume/proto/master.proto

@@ -147,7 +147,11 @@ message VolumeEcShardInformationMessage {
   repeated int64 shard_sizes = 7; // optimized: sizes for shards in order of set bits in ec_index_bits
   uint64 file_count = 8;   // total needles in the .ecx index (live + tombstoned)
   uint64 delete_count = 9; // node-local tombstones in the .ecj deletion journal
-  // fields 10-19 reserved for future upstream open-source additions.
+  // encode-run identity (unix nanos) from the .vif EcShardConfig; lets the admin
+  // group shards by encode generation. Numbered 14 (not 10) to skip the
+  // enterprise fork's reserved 10-13.
+  int64 encode_ts_ns = 14;
+  // fields 15-19 reserved for future upstream open-source additions.
   // fields 20+ are owned by the enterprise fork (e.g. data_shards/parity_shards)
   // and must not be used here without coordination.
 }

seaweed-volume/src/server/heartbeat.rs

@@ -314,6 +314,7 @@ fn collect_ec_shard_delta_messages(
                         disk_type: ec_vol.disk_type.to_string(),
                         expire_at_sec: ec_vol.expire_at_sec,
                         disk_id: disk_id as u32,
+                        encode_ts_ns: ec_vol.encode_ts_ns,
                         ..Default::default()
                     },
                 );

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

@@ -448,6 +448,7 @@ impl EcVolume {
             disk_id,
             file_count,
             delete_count,
+            encode_ts_ns: self.encode_ts_ns,
             ..Default::default()
         }]
     }

weed/pb/master.proto

@@ -147,7 +147,11 @@ message VolumeEcShardInformationMessage {
   repeated int64 shard_sizes = 7; // optimized: sizes for shards in order of set bits in ec_index_bits
   uint64 file_count = 8;   // total needles in the .ecx index (live + tombstoned)
   uint64 delete_count = 9; // node-local tombstones in the .ecj deletion journal
-  // fields 10-19 reserved for future upstream open-source additions.
+  // encode-run identity (unix nanos) from the .vif EcShardConfig; lets the admin
+  // group shards by encode generation. Numbered 14 (not 10) to skip the
+  // enterprise fork's reserved 10-13.
+  int64 encode_ts_ns = 14;
+  // fields 15-19 reserved for future upstream open-source additions.
   // fields 20+ are owned by the enterprise fork (e.g. data_shards/parity_shards)
   // and must not be used here without coordination.
 }

weed/pb/master_pb/master.pb.go

@@ -631,16 +631,20 @@ func (x *VolumeShortInformationMessage) GetDiskId() uint32 {
 }
 
 type VolumeEcShardInformationMessage struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	Id            uint32                 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
-	Collection    string                 `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
-	EcIndexBits   uint32                 `protobuf:"varint,3,opt,name=ec_index_bits,json=ecIndexBits,proto3" json:"ec_index_bits,omitempty"`
-	DiskType      string                 `protobuf:"bytes,4,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
-	ExpireAtSec   uint64                 `protobuf:"varint,5,opt,name=expire_at_sec,json=expireAtSec,proto3" json:"expire_at_sec,omitempty"` // used to record the destruction time of ec volume
-	DiskId        uint32                 `protobuf:"varint,6,opt,name=disk_id,json=diskId,proto3" json:"disk_id,omitempty"`
-	ShardSizes    []int64                `protobuf:"varint,7,rep,packed,name=shard_sizes,json=shardSizes,proto3" json:"shard_sizes,omitempty"` // optimized: sizes for shards in order of set bits in ec_index_bits
-	FileCount     uint64                 `protobuf:"varint,8,opt,name=file_count,json=fileCount,proto3" json:"file_count,omitempty"`           // total needles in the .ecx index (live + tombstoned)
-	DeleteCount   uint64                 `protobuf:"varint,9,opt,name=delete_count,json=deleteCount,proto3" json:"delete_count,omitempty"`     // node-local tombstones in the .ecj deletion journal
+	state       protoimpl.MessageState `protogen:"open.v1"`
+	Id          uint32                 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
+	Collection  string                 `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
+	EcIndexBits uint32                 `protobuf:"varint,3,opt,name=ec_index_bits,json=ecIndexBits,proto3" json:"ec_index_bits,omitempty"`
+	DiskType    string                 `protobuf:"bytes,4,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
+	ExpireAtSec uint64                 `protobuf:"varint,5,opt,name=expire_at_sec,json=expireAtSec,proto3" json:"expire_at_sec,omitempty"` // used to record the destruction time of ec volume
+	DiskId      uint32                 `protobuf:"varint,6,opt,name=disk_id,json=diskId,proto3" json:"disk_id,omitempty"`
+	ShardSizes  []int64                `protobuf:"varint,7,rep,packed,name=shard_sizes,json=shardSizes,proto3" json:"shard_sizes,omitempty"` // optimized: sizes for shards in order of set bits in ec_index_bits
+	FileCount   uint64                 `protobuf:"varint,8,opt,name=file_count,json=fileCount,proto3" json:"file_count,omitempty"`           // total needles in the .ecx index (live + tombstoned)
+	DeleteCount uint64                 `protobuf:"varint,9,opt,name=delete_count,json=deleteCount,proto3" json:"delete_count,omitempty"`     // node-local tombstones in the .ecj deletion journal
+	// encode-run identity (unix nanos) from the .vif EcShardConfig; lets the admin
+	// group shards by encode generation. Numbered 14 (not 10) to skip the
+	// enterprise fork's reserved 10-13.
+	EncodeTsNs    int64 `protobuf:"varint,14,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -738,6 +742,13 @@ func (x *VolumeEcShardInformationMessage) GetDeleteCount() uint64 {
 	return 0
 }
 
+func (x *VolumeEcShardInformationMessage) GetEncodeTsNs() int64 {
+	if x != nil {
+		return x.EncodeTsNs
+	}
+	return 0
+}
+
 type StorageBackend struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Type          string                 `protobuf:"bytes,1,opt,name=type,proto3" json:"type,omitempty"`
@@ -4439,7 +4450,7 @@ const file_master_proto_rawDesc = "" +
 	"\x03ttl\x18\n" +
 	" \x01(\rR\x03ttl\x12\x1b\n" +
 	"\tdisk_type\x18\x0f \x01(\tR\bdiskType\x12\x17\n" +
-	"\adisk_id\x18\x10 \x01(\rR\x06diskId\"\xb2\x02\n" +
+	"\adisk_id\x18\x10 \x01(\rR\x06diskId\"\xd4\x02\n" +
 	"\x1fVolumeEcShardInformationMessage\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\rR\x02id\x12\x1e\n" +
 	"\n" +
@@ -4453,7 +4464,9 @@ const file_master_proto_rawDesc = "" +
 	"shardSizes\x12\x1d\n" +
 	"\n" +
 	"file_count\x18\b \x01(\x04R\tfileCount\x12!\n" +
-	"\fdelete_count\x18\t \x01(\x04R\vdeleteCount\"\xbe\x01\n" +
+	"\fdelete_count\x18\t \x01(\x04R\vdeleteCount\x12 \n" +
+	"\fencode_ts_ns\x18\x0e \x01(\x03R\n" +
+	"encodeTsNs\"\xbe\x01\n" +
 	"\x0eStorageBackend\x12\x12\n" +
 	"\x04type\x18\x01 \x01(\tR\x04type\x12\x0e\n" +
 	"\x02id\x18\x02 \x01(\tR\x02id\x12I\n" +

weed/storage/erasure_coding/ec_volume.go

@@ -375,6 +375,7 @@ func (ev *EcVolume) ToVolumeEcShardInformationMessage(diskId uint32) (messages [
 				DiskId:      diskId,
 				FileCount:   fileCount,
 				DeleteCount: deleteCount,
+				EncodeTsNs:  ev.EncodeTsNs,
 			}
 			ecInfoPerVolume[s.VolumeId] = m
 		}

weed/storage/erasure_coding/ec_volume_info.go

@@ -15,6 +15,7 @@ type EcVolumeInfo struct {
 	ShardsInfo  *ShardsInfo
 	FileCount   uint64 // live needle count for this EC volume (same on every node holding shards)
 	DeleteCount uint64 // tombstoned needle count for this EC volume
+	EncodeTsNs  int64  // encode-run identity (unix nanos); one value per (volume, disk)
 }
 
 func (ecInfo *EcVolumeInfo) Minus(other *EcVolumeInfo) *EcVolumeInfo {
@@ -27,6 +28,7 @@ func (ecInfo *EcVolumeInfo) Minus(other *EcVolumeInfo) *EcVolumeInfo {
 		ExpireAtSec: ecInfo.ExpireAtSec,
 		FileCount:   ecInfo.FileCount,
 		DeleteCount: ecInfo.DeleteCount,
+		EncodeTsNs:  ecInfo.EncodeTsNs,
 	}
 }
 
@@ -41,5 +43,6 @@ func (evi *EcVolumeInfo) ToVolumeEcShardInformationMessage() (ret *master_pb.Vol
 		DiskId:      evi.DiskId,
 		FileCount:   evi.FileCount,
 		DeleteCount: evi.DeleteCount,
+		EncodeTsNs:  evi.EncodeTsNs,
 	}
 }

weed/storage/store.go

@@ -140,6 +140,7 @@ func NewStore(
 				DiskType:    string(location.DiskType),
 				ExpireAtSec: ecVolume.ExpireAtSec,
 				DiskId:      diskId,
+				EncodeTsNs:  ecVolume.EncodeTsNs,
 			}:
 			default:
 				// Channel full during startup - this is OK, heartbeat will report EC shards later

weed/storage/store_ec.go

@@ -218,6 +218,7 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er
 				DiskType:    string(ecVolume.DiskType()),
 				ExpireAtSec: ecVolume.ExpireAtSec,
 				DiskId:      uint32(diskId),
+				EncodeTsNs:  ecVolume.EncodeTsNs,
 			}
 			return nil
 		}
@@ -262,6 +263,12 @@ func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.Shar
 		if !found {
 			continue
 		}
+		// Capture the encode generation before unloading so the deletion delta
+		// carries it like the mount delta does.
+		var encodeTsNs int64
+		if ecVolume, ok := location.FindEcVolume(vid); ok {
+			encodeTsNs = ecVolume.EncodeTsNs
+		}
 		if deleted := location.UnloadEcShard(vid, shardId); deleted {
 			si := erasure_coding.NewShardsInfo()
 			si.Set(erasure_coding.NewShardInfo(shardId, 0))
@@ -272,6 +279,7 @@ func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.Shar
 				ShardSizes:  si.SizesInt64(),
 				DiskType:    string(ecShard.DiskType),
 				DiskId:      uint32(diskId),
+				EncodeTsNs:  encodeTsNs,
 			}
 			glog.V(0).Infof("UnmountEcShards %d.%d disk_id:%d", vid, shardId, diskId)
 			unmountedAny = true

weed/topology/topology_ec.go

@@ -28,6 +28,7 @@ func (t *Topology) SyncDataNodeEcShards(shardInfos []*master_pb.VolumeEcShardInf
 			ExpireAtSec: shardInfo.ExpireAtSec,
 			FileCount:   shardInfo.FileCount,
 			DeleteCount: shardInfo.DeleteCount,
+			EncodeTsNs:  shardInfo.EncodeTsNs,
 		}
 
 		shards = append(shards, ecVolumeInfo)
@@ -57,6 +58,7 @@ func (t *Topology) IncrementalSyncDataNodeEcShards(newEcShards, deletedEcShards
 			ExpireAtSec: shardInfo.ExpireAtSec,
 			FileCount:   shardInfo.FileCount,
 			DeleteCount: shardInfo.DeleteCount,
+			EncodeTsNs:  shardInfo.EncodeTsNs,
 		}
 
 		newShards = append(newShards, ecVolumeInfo)
@@ -72,6 +74,7 @@ func (t *Topology) IncrementalSyncDataNodeEcShards(newEcShards, deletedEcShards
 			ExpireAtSec: shardInfo.ExpireAtSec,
 			FileCount:   shardInfo.FileCount,
 			DeleteCount: shardInfo.DeleteCount,
+			EncodeTsNs:  shardInfo.EncodeTsNs,
 		}
 
 		deletedShards = append(deletedShards, ecVolumeInfo)

weed/worker/tasks/erasure_coding/detection.go

@@ -758,11 +758,18 @@ func cleanupOrphanSourceReplicas(ctx context.Context, clusterInfo *types.Cluster
 }
 
 // countExistingEcShardsForVolume returns the number of distinct EC shard IDs
-// for (volumeID, collection) present in the topology. Walks every disk's
-// EcIndexBits bitmap rather than trusting len(EcShardInfos), because a single
-// info entry can carry multiple shards. Used by the #9448 guard to decide
-// whether the EC shard set is complete enough that the orphaned regular
-// replica is safe to delete.
+// for (volumeID, collection) present in the topology, counting only the single
+// largest encode generation. Shards are grouped by encode_ts_ns (the per-encode
+// identity from .vif), so two interrupted encode runs whose shard sets overlap
+// are never unioned into a false-complete set that would wrongly trigger the
+// orphaned-source delete. Walks every disk's EcIndexBits bitmap rather than
+// trusting len(EcShardInfos), because a single info entry can carry multiple
+// shards. Shards reporting encode_ts_ns==0 (pre-upgrade servers) form their own
+// generation bucket.
+//
+// Limitation: the heartbeat carries one encode_ts_ns per (volume, disk), so this
+// separates generations living on different disks; same-disk mixing is prevented
+// upstream by the pre-encode artifact wipe and the cross-run read guard.
 func countExistingEcShardsForVolume(activeTopology *topology.ActiveTopology, volumeID uint32, collection string) int {
 	if activeTopology == nil {
 		return 0
@@ -771,20 +778,29 @@ func countExistingEcShardsForVolume(activeTopology *topology.ActiveTopology, vol
 	if topologyInfo == nil {
 		return 0
 	}
-	var seen erasure_coding.ShardBits
+	perGeneration := make(map[int64]erasure_coding.ShardBits)
 	for _, dc := range topologyInfo.DataCenterInfos {
 		for _, rack := range dc.RackInfos {
 			for _, node := range rack.DataNodeInfos {
 				for _, diskInfo := range node.DiskInfos {
 					for _, ecShardInfo := range diskInfo.EcShardInfos {
+						if ecShardInfo == nil {
+							continue
+						}
 						if ecShardInfo.Id != volumeID || ecShardInfo.Collection != collection {
 							continue
 						}
-						seen |= erasure_coding.ShardBits(ecShardInfo.EcIndexBits)
+						perGeneration[ecShardInfo.EncodeTsNs] |= erasure_coding.ShardBits(ecShardInfo.EcIndexBits)
 					}
 				}
 			}
 		}
 	}
-	return seen.Count()
+	best := 0
+	for _, bits := range perGeneration {
+		if c := bits.Count(); c > best {
+			best = c
+		}
+	}
+	return best
 }

weed/worker/tasks/erasure_coding/detection_ec_generation_test.go

@@ -0,0 +1,92 @@
+package erasure_coding
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/stretchr/testify/require"
+)
+
+type ecShardSet struct {
+	bits       uint32
+	encodeTsNs int64
+}
+
+func ecBits(shardIds ...int) uint32 {
+	var b uint32
+	for _, id := range shardIds {
+		b |= uint32(1) << uint(id)
+	}
+	return b
+}
+
+// buildEcTopology builds an ActiveTopology where each shard set lives on its own
+// node, so countExistingEcShardsForVolume can be exercised across encode
+// generations spread over different disks.
+func buildEcTopology(t *testing.T, vid uint32, collection string, sets []ecShardSet) *topology.ActiveTopology {
+	t.Helper()
+	at := topology.NewActiveTopology(10)
+	nodes := make([]*master_pb.DataNodeInfo, 0, len(sets))
+	for i, set := range sets {
+		nodes = append(nodes, &master_pb.DataNodeInfo{
+			Id: fmt.Sprintf("10.0.0.%d:8080", i+1),
+			DiskInfos: map[string]*master_pb.DiskInfo{
+				"hdd": {
+					DiskId:         0,
+					MaxVolumeCount: 100,
+					EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
+						{Id: vid, Collection: collection, EcIndexBits: set.bits, EncodeTsNs: set.encodeTsNs},
+					},
+				},
+			},
+		})
+	}
+	topologyInfo := &master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{{
+			Id:        "dc1",
+			RackInfos: []*master_pb.RackInfo{{Id: "rack1", DataNodeInfos: nodes}},
+		}},
+	}
+	require.NoError(t, at.UpdateTopology(topologyInfo))
+	return at
+}
+
+// Two interrupted encode runs whose shard sets overlap in count must NOT be
+// unioned into a false-complete set. countExistingEcShardsForVolume returns the
+// largest single generation's shard count, not the cross-run union.
+func TestCountExistingEcShards_DoesNotUnionAcrossGenerations(t *testing.T) {
+	const vid = uint32(42)
+
+	// run A holds shards 0-6 (7 shards, encode_ts 1); run B holds shards 7-13
+	// (7 shards, encode_ts 2). The union would be all 14; the per-generation max
+	// is 7, so the orphan-source delete trigger (>= 14) must not fire.
+	at := buildEcTopology(t, vid, "", []ecShardSet{
+		{bits: ecBits(0, 1, 2, 3, 4, 5, 6), encodeTsNs: 1},
+		{bits: ecBits(7, 8, 9, 10, 11, 12, 13), encodeTsNs: 2},
+	})
+	require.Equal(t, 7, countExistingEcShardsForVolume(at, vid, ""))
+}
+
+// A single complete generation still counts as complete, so genuine stuck-source
+// cleanup keeps working.
+func TestCountExistingEcShards_SingleCompleteGeneration(t *testing.T) {
+	const vid = uint32(43)
+	at := buildEcTopology(t, vid, "", []ecShardSet{
+		{bits: ecBits(0, 1, 2, 3, 4, 5, 6), encodeTsNs: 5},
+		{bits: ecBits(7, 8, 9, 10, 11, 12, 13), encodeTsNs: 5},
+	})
+	require.Equal(t, 14, countExistingEcShardsForVolume(at, vid, ""))
+}
+
+// Shards from pre-upgrade servers report encode_ts_ns==0; a complete 0-generation
+// set is still recognized as complete (its own bucket).
+func TestCountExistingEcShards_PreUpgradeZeroGeneration(t *testing.T) {
+	const vid = uint32(44)
+	at := buildEcTopology(t, vid, "", []ecShardSet{
+		{bits: ecBits(0, 1, 2, 3, 4, 5, 6), encodeTsNs: 0},
+		{bits: ecBits(7, 8, 9, 10, 11, 12, 13), encodeTsNs: 0},
+	})
+	require.Equal(t, 14, countExistingEcShardsForVolume(at, vid, ""))
+}