From da243b942390c0dcd1510626b54d963726edde97 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 14 Jun 2026 00:14:12 -0700 Subject: [PATCH] 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 | 6 +- seaweed-volume/src/server/heartbeat.rs | 1 + .../src/storage/erasure_coding/ec_volume.rs | 1 + weed/pb/master.proto | 6 +- weed/pb/master_pb/master.pb.go | 37 +++++--- weed/storage/erasure_coding/ec_volume.go | 1 + weed/storage/erasure_coding/ec_volume_info.go | 3 + weed/storage/store.go | 1 + weed/storage/store_ec.go | 8 ++ weed/topology/topology_ec.go | 3 + weed/worker/tasks/erasure_coding/detection.go | 32 +++++-- .../detection_ec_generation_test.go | 92 +++++++++++++++++++ 12 files changed, 169 insertions(+), 22 deletions(-) create mode 100644 weed/worker/tasks/erasure_coding/detection_ec_generation_test.go diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto index 4d112232c..e9279f911 100644 --- a/seaweed-volume/proto/master.proto +++ b/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. } diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 2f5272c5d..0906cc257 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/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() }, ); diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 460e463d7..d0aaa4775 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/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() }] } diff --git a/weed/pb/master.proto b/weed/pb/master.proto index 57fa1f268..beee73b1e 100644 --- a/weed/pb/master.proto +++ b/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. } diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index eb47afb5d..efdb66380 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/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" + diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index ac919b37a..91388b8a0 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/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 } diff --git a/weed/storage/erasure_coding/ec_volume_info.go b/weed/storage/erasure_coding/ec_volume_info.go index 13d22ee60..0c6d18714 100644 --- a/weed/storage/erasure_coding/ec_volume_info.go +++ b/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, } } diff --git a/weed/storage/store.go b/weed/storage/store.go index fbeee62b5..0f420203c 100644 --- a/weed/storage/store.go +++ b/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 diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index e28094743..690f81c6f 100644 --- a/weed/storage/store_ec.go +++ b/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 diff --git a/weed/topology/topology_ec.go b/weed/topology/topology_ec.go index 6f7189f1a..f121ab4c7 100644 --- a/weed/topology/topology_ec.go +++ b/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) diff --git a/weed/worker/tasks/erasure_coding/detection.go b/weed/worker/tasks/erasure_coding/detection.go index 17331627b..571fa0132 100644 --- a/weed/worker/tasks/erasure_coding/detection.go +++ b/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 } diff --git a/weed/worker/tasks/erasure_coding/detection_ec_generation_test.go b/weed/worker/tasks/erasure_coding/detection_ec_generation_test.go new file mode 100644 index 000000000..bd735d453 --- /dev/null +++ b/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, "")) +}