diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index fc95c6964b..785df66dd0 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -465,6 +465,9 @@ fn collect_ec_shard_delta_messages( if quarantined { continue; } + // Same numbers the full EC heartbeat carries, read off the active + // generation's index and journal; one call covers every shard. + let (file_count, delete_count) = ec_vol.file_and_delete_count(); for shard in ec_vol.shards.iter().flatten() { messages.insert( ( @@ -481,6 +484,8 @@ 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, + file_count, + delete_count, encode_ts_ns: ec_vol.encode_ts_ns, ..Default::default() }, @@ -503,7 +508,17 @@ fn diff_ec_shard_delta_messages( let mut deleted_ec_shards = Vec::new(); for (key, message) in current { - if previous.get(key) != Some(message) { + // Counts move with every delete; the full EC heartbeat carries those, + // so only a change to the shard itself is news here. + let unchanged = previous.get(key).is_some_and(|prev| { + *message + == master_pb::VolumeEcShardInformationMessage { + file_count: message.file_count, + delete_count: message.delete_count, + ..prev.clone() + } + }); + if !unchanged { new_ec_shards.push(message.clone()); } } @@ -3157,4 +3172,88 @@ mod tests { assert_eq!(deleted_after_delete[0].ec_index_bits, 1); assert_eq!(deleted_after_delete[0].shard_sizes, vec![0]); } + + #[test] + fn test_diff_ec_shard_delta_messages_ignores_count_changes() { + let key = (81, "ec_delta_case".to_string(), 0, 0); + let previous = HashMap::from([( + key.clone(), + master_pb::VolumeEcShardInformationMessage { + id: 81, + collection: "ec_delta_case".to_string(), + ec_index_bits: 1, + shard_sizes: vec![5], + file_count: 3, + ..Default::default() + }, + )]); + + let mut current = previous.clone(); + current.get_mut(&key).unwrap().file_count = 4; + current.get_mut(&key).unwrap().delete_count = 1; + let (new_ec_shards, deleted_ec_shards) = diff_ec_shard_delta_messages(&previous, ¤t); + assert!(new_ec_shards.is_empty()); + assert!(deleted_ec_shards.is_empty()); + + current.get_mut(&key).unwrap().shard_sizes = vec![6]; + let (new_ec_shards, _) = diff_ec_shard_delta_messages(&previous, ¤t); + assert_eq!(new_ec_shards.len(), 1); + assert_eq!(new_ec_shards[0].delete_count, 1); + } + + /// Both EC emission paths report the counts the Go server's + /// `EcVolume.FileAndDeleteCount` defines: file_count is every entry in the + /// active .ecx (live and tombstoned alike), delete_count the needles this + /// holder deleted. + #[test] + fn test_ec_shard_messages_carry_file_and_delete_counts() { + use crate::storage::types::{NEEDLE_MAP_ENTRY_SIZE, Offset, Size, idx_entry_to_bytes}; + + let temp_dir = tempfile::tempdir().unwrap(); + let dir = temp_dir.path().to_str().unwrap(); + + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + dir, + dir, + 8, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + + std::fs::write(format!("{}/ec_counts_case_91.ec00", dir), b"shard").unwrap(); + let mut ecx = Vec::new(); + for (key, offset) in [(1u64, 8u32), (5, 200), (10, 500)] { + let mut entry = [0u8; NEEDLE_MAP_ENTRY_SIZE]; + idx_entry_to_bytes( + &mut entry, + NeedleId(key), + Offset::from_actual_offset(offset as i64), + Size(100), + ); + ecx.extend_from_slice(&entry); + } + std::fs::write(format!("{}/ec_counts_case_91.ecx", dir), &ecx).unwrap(); + store.locations[0] + .mount_ec_shards(VolumeId(91), "ec_counts_case", &[0], "") + .unwrap(); + + let ec_vol = store.locations[0].find_ec_volume_mut(VolumeId(91)).unwrap(); + ec_vol.journal_delete(NeedleId(1)).unwrap(); + + let delta = collect_ec_shard_delta_messages(&store); + let message = delta + .get(&(91, "ec_counts_case".to_string(), 0, 0)) + .expect("delta message for the mounted shard"); + assert_eq!(message.file_count, 3); + assert_eq!(message.delete_count, 1); + + let heartbeat = collect_ec_heartbeat(&test_config(), &store); + assert_eq!(heartbeat.ec_shards.len(), 1); + assert_eq!(heartbeat.ec_shards[0].file_count, 3); + assert_eq!(heartbeat.ec_shards[0].delete_count, 1); + } } diff --git a/weed/storage/store.go b/weed/storage/store.go index 21ac971ed7..f77dae808c 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -172,6 +172,7 @@ func NewStore( s.UpdateMaxVolumeIdSeen(vid) si := erasure_coding.NewShardsInfo() si.Set(erasure_coding.NewShardInfo(shardId, erasure_coding.ShardSize(ecVolume.ShardSize()))) + fileCount, deleteCount := ecVolume.FileAndDeleteCount() // Use non-blocking send during startup to avoid deadlock // The channel reader only starts after connecting to master, but we're loading during startup @@ -184,6 +185,8 @@ func NewStore( DiskType: string(location.DiskType), ExpireAtSec: ecVolume.ExpireAtSec, DiskId: diskId, + FileCount: fileCount, + DeleteCount: deleteCount, EncodeTsNs: ecVolume.EncodeTsNs, }: default: diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index d3250c45fc..ff8fbf224c 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -268,6 +268,9 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er si := erasure_coding.NewShardsInfo() si.Set(erasure_coding.NewShardInfo(shardId, erasure_coding.ShardSize(ecVolume.ShardSize()))) + // Same counts the full EC heartbeat carries, so a volume the master first + // learns of here does not read 0 files until the next full heartbeat. + fileCount, deleteCount := ecVolume.FileAndDeleteCount() s.NewEcShardsChan <- &master_pb.VolumeEcShardInformationMessage{ Id: uint32(vid), Collection: collection, @@ -276,6 +279,8 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er DiskType: string(ecVolume.DiskType()), ExpireAtSec: ecVolume.ExpireAtSec, DiskId: uint32(diskId), + FileCount: fileCount, + DeleteCount: deleteCount, EncodeTsNs: ecVolume.EncodeTsNs, } s.UpdateMaxVolumeIdSeen(vid) diff --git a/weed/storage/store_ec_disk_type_test.go b/weed/storage/store_ec_disk_type_test.go index 921db2c33a..aac442e1d9 100644 --- a/weed/storage/store_ec_disk_type_test.go +++ b/weed/storage/store_ec_disk_type_test.go @@ -3,6 +3,7 @@ package storage import ( "os" "path/filepath" + "sync" "testing" "time" @@ -23,10 +24,9 @@ import ( // locations and then invoke MountEcShards — this mirrors the real // VolumeEcShardsCopy → VolumeEcShardsMount sequence and avoids racing // with the startup-scan publish (which uses the location's disk type -// and would otherwise pollute the captured slice). The returned slice -// is appended to only by the goroutine; tests should not race against -// it directly — use waitForShardMsg instead. -func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[]*master_pb.VolumeEcShardInformationMessage, vid needle.VolumeId, collection string, plant func()) { +// and would otherwise pollute the captured messages). Read them with +// waitForShardMsg. +func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *shardMsgLog, vid needle.VolumeId, collection string, plant func()) { t.Helper() tempDir := t.TempDir() hddDir := filepath.Join(tempDir, "hdd") @@ -56,14 +56,15 @@ func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[ diskIOProbeConfig, ) - captured := []*master_pb.VolumeEcShardInformationMessage{} - drainedShardMsgs = &captured + drainedShardMsgs = &shardMsgLog{} done := make(chan struct{}) go func() { for { select { case msg := <-store.NewEcShardsChan: - captured = append(captured, msg) + drainedShardMsgs.mu.Lock() + drainedShardMsgs.msgs = append(drainedShardMsgs.msgs, msg) + drainedShardMsgs.mu.Unlock() case <-store.NewVolumesChan: case <-store.DeletedVolumesChan: case <-store.DeletedEcShardsChan: @@ -114,16 +115,31 @@ func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[ return store, drainedShardMsgs, vid, collection, plant } +// shardMsgLog collects the NewEcShardsChan messages the drain goroutine reads. +type shardMsgLog struct { + mu sync.Mutex + msgs []*master_pb.VolumeEcShardInformationMessage +} + +func (l *shardMsgLog) find(vid needle.VolumeId) *master_pb.VolumeEcShardInformationMessage { + l.mu.Lock() + defer l.mu.Unlock() + for _, msg := range l.msgs { + if msg.Id == uint32(vid) { + return msg + } + } + return nil +} + // waitForShardMsg returns the first captured mount message for vid, polling // briefly because the goroutine reads asynchronously from MountEcShards. -func waitForShardMsg(t *testing.T, captured *[]*master_pb.VolumeEcShardInformationMessage, vid needle.VolumeId) *master_pb.VolumeEcShardInformationMessage { +func waitForShardMsg(t *testing.T, captured *shardMsgLog, vid needle.VolumeId) *master_pb.VolumeEcShardInformationMessage { t.Helper() deadline := time.Now().Add(2 * time.Second) for time.Now().Before(deadline) { - for _, msg := range *captured { - if msg.Id == uint32(vid) { - return msg - } + if msg := captured.find(vid); msg != nil { + return msg } time.Sleep(10 * time.Millisecond) } @@ -213,3 +229,79 @@ func TestMountEcShards_FallsBackToLocationDiskTypeWhenEmpty(t *testing.T) { t.Errorf("heartbeat disk_type = %q, want %q", heartbeatMsg.DiskType, string(types.HardDriveType)) } } + +// writeEcCounts gives the planted volume three .ecx entries and one deleted +// needle, journaled twice so a count of .ecj records would read 2. +func writeEcCounts(t *testing.T, store *Store, collection string, vid needle.VolumeId) { + t.Helper() + base := erasure_coding.EcShardFileName(collection, store.Locations[0].Directory, int(vid)) + if err := os.WriteFile(base+".ecx", make([]byte, 3*types.NeedleMapEntrySize), 0o644); err != nil { + t.Fatalf("write .ecx: %v", err) + } + ecj := make([]byte, 2*types.NeedleIdSize) + types.NeedleIdToBytes(ecj[:types.NeedleIdSize], types.NeedleId(7)) + types.NeedleIdToBytes(ecj[types.NeedleIdSize:], types.NeedleId(7)) + if err := os.WriteFile(base+".ecj", ecj, 0o644); err != nil { + t.Fatalf("write .ecj: %v", err) + } +} + +// TestMountEcShards_ReportsFileAndDeleteCounts pins the mount notification to +// the counts the full EC heartbeat carries (EcVolume.FileAndDeleteCount): +// every .ecx entry, and the distinct deletes journaled in .ecj. +func TestMountEcShards_ReportsFileAndDeleteCounts(t *testing.T) { + store, captured, vid, collection, plant := setupECStoreWithMixedDisks(t) + plant() + writeEcCounts(t, store, collection, vid) + + if err := store.MountEcShards(collection, vid, 0, ""); err != nil { + t.Fatalf("MountEcShards: %v", err) + } + + mountMsg := waitForShardMsg(t, captured, vid) + if mountMsg.FileCount != 3 || mountMsg.DeleteCount != 1 { + t.Errorf("NewEcShardsChan counts = (%d, %d), want (3, 1)", mountMsg.FileCount, mountMsg.DeleteCount) + } + + hb := store.CollectErasureCodingHeartbeat() + heartbeatMsg := findHeartbeatShard(t, hb, vid) + if heartbeatMsg.FileCount != 3 || heartbeatMsg.DeleteCount != 1 { + t.Errorf("heartbeat counts = (%d, %d), want (3, 1)", heartbeatMsg.FileCount, heartbeatMsg.DeleteCount) + } +} + +// TestNewStore_StartupEcMessageReportsFileAndDeleteCounts covers the message +// the startup scan publishes for a shard it loads. +func TestNewStore_StartupEcMessageReportsFileAndDeleteCounts(t *testing.T) { + store, _, vid, collection, plant := setupECStoreWithMixedDisks(t) + plant() + writeEcCounts(t, store, collection, vid) + + restarted := NewStore(nil, "localhost", 8081, 18081, "http://localhost:8081", "store-id-restarted", + []string{store.Locations[0].Directory, store.Locations[1].Directory}, + []int32{100, 100}, + []util.MinFreeSpace{{}, {}}, + "", + NeedleMapInMemory, + []types.DiskType{types.HardDriveType, types.SsdType}, + nil, + 3, + stats.DefaultDiskIOProbeConfig(), + ) + t.Cleanup(restarted.Close) + + for { + select { + case msg := <-restarted.NewEcShardsChan: + if msg.Id != uint32(vid) { + continue + } + if msg.FileCount != 3 || msg.DeleteCount != 1 { + t.Errorf("startup NewEcShardsChan counts = (%d, %d), want (3, 1)", msg.FileCount, msg.DeleteCount) + } + return + default: + t.Fatalf("no startup NewEcShardsChan message for volume %d", vid) + } + } +} diff --git a/weed/topology/data_node_ec.go b/weed/topology/data_node_ec.go index 095cc30212..621045bf0e 100644 --- a/weed/topology/data_node_ec.go +++ b/weed/topology/data_node_ec.go @@ -120,9 +120,9 @@ func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) } // ecCountsChanged reports whether a heartbeat moves the stored FileCount or -// DeleteCount. A message reporting 0/0 did not count (older volume servers and -// the incremental mount message omit the fields), so it keeps the stored values -// and carries them onto the incoming record in case the map is rewritten. +// DeleteCount. A message reporting 0/0 did not count (older volume servers omit +// the fields), so it keeps the stored values and carries them onto the incoming +// record in case the map is rewritten. func ecCountsChanged(stored, actual *erasure_coding.EcVolumeInfo) bool { if actual.FileCount == 0 && actual.DeleteCount == 0 { actual.FileCount, actual.DeleteCount = stored.FileCount, stored.DeleteCount diff --git a/weed/topology/disk_ec.go b/weed/topology/disk_ec.go index 4aca6b2824..8c1d0cc838 100644 --- a/weed/topology/disk_ec.go +++ b/weed/topology/disk_ec.go @@ -49,7 +49,7 @@ func (d *Disk) AddOrUpdateEcShard(s *erasure_coding.EcVolumeInfo) { updated := *existing updated.ShardsInfo = existing.ShardsInfo.Plus(s.ShardsInfo) delta = updated.ShardsInfo.Count() - existing.ShardsInfo.Count() - // 0/0 means the message did not count (the mount message omits the fields) + // 0/0 means the message did not count (older volume servers omit the fields) if s.FileCount != 0 || s.DeleteCount != 0 { updated.FileCount, updated.DeleteCount = s.FileCount, s.DeleteCount }