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