volume: report EC file and delete counts in incremental shard messages (#12404)

* seaweed-volume: report EC file and delete counts in heartbeat deltas

The full EC heartbeat already carries file_count and delete_count, but the
incremental new/deleted shard messages built in
collect_ec_shard_delta_messages left both at zero. A volume the master first
learns of through a delta then reads 0 files until the next full EC
heartbeat. Fill both from file_and_delete_count, the same source the full
heartbeat uses, computed once per volume rather than once per shard.

Claude-Session: https://claude.ai/code/session_011UpvafcfUuC2mLxNAXLQJi

* volume: report EC file and delete counts in mount notifications

Go parity with the Rust delta change: the NewEcShardsChan messages sent on
MountEcShards and on startup shard load now carry FileCount and DeleteCount
from EcVolume.FileAndDeleteCount, the same values the full EC heartbeat
reports, instead of leaving them at zero.

Claude-Session: https://claude.ai/code/session_011UpvafcfUuC2mLxNAXLQJi

* seaweed-volume: leave count-only EC changes out of the shard delta

The delta messages now carry file_count and delete_count, so the whole-message
comparison re-sent every shard of a volume after any delete, and the master
broadcast each one as a new EC vid. Compare the shard fields only; the full EC
heartbeat already carries the counts.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* topology: mount messages now carry EC counts, only older servers omit them

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* storage: guard the captured EC mount messages in the disk-type tests

The drain goroutine appended to the slice that waitForShardMsg polled with no
lock, which go test -race reports.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* storage: test EC counts on the startup notification and repeated .ecj ids

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: move file_count too in the count-only delta test

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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, &current);
+        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, &current);
+        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);
+    }
 }

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:

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)

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

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

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
 		}