heartbeat: keep the master current through collection churn (#10657)

* heartbeat: name departed volumes in delta heartbeats

* master: release the lookup index with a deleted collection

* master: keep a fresh grow safe from the report that raced it

* volume: name the volumes a deleted collection took with it

Deleting a collection left the master to work out what went by omission from
the next full volume list, which it no longer gets: heartbeats carry the whole
list only when the master asks for it. The volumes a bucket's churn creates and
destroys between two of those requests are never named in either direction, so
the master keeps counting their slots as occupied and a cluster that creates
and drops collections quickly runs its free-slot accounting dry -- assigns fail
with no free volumes left while the disk holds a handful of volumes.

The destroy path already knows exactly which volumes it removed, so send them
down the same channel every other deletion uses.

* rust: name the volumes a deleted collection took with it

Mirrors the Go volume server. The notify path derives its deltas by diffing
snapshots, so a collection delete that does not wake it is invisible until the
master next asks for the whole list.

seaweed-volume/src/server/grpc_server.rs

@@ -660,10 +660,17 @@ impl VolumeServer for VolumeGrpcService {
     ) -> Result<Response<volume_server_pb::DeleteCollectionResponse>, Status> {
         self.check_grpc_admin_auth(&request)?;
         let collection = &request.into_inner().collection;
-        let mut store = self.state.store.write().unwrap();
-        store
-            .delete_collection(collection)
-            .map_err(|e| Status::internal(e))?;
+        {
+            let mut store = self.state.store.write().unwrap();
+            store
+                .delete_collection(collection)
+                .map_err(|e| Status::internal(e))?;
+        }
+        // The delta the notify path derives is the only thing that tells the
+        // master these slots came free: a heartbeat carries the whole list only
+        // when the master asks, and a volume grown and destroyed between two of
+        // them was never in one at all.
+        self.state.volume_state_notify.notify_one();
         Ok(Response::new(volume_server_pb::DeleteCollectionResponse {}))
     }
 

weed/server/master_grpc_server_changed_volumes_test.go

@@ -145,3 +145,61 @@ func TestRepairedLookupEntryIsAnnounced(t *testing.T) {
 		})
 	}
 }
+
+// Deleting a collection throws its layouts away wholesale. The lookup bits
+// they held must go with them: leaked bits keep the node's held and servable
+// digests apart forever, and the master then asks for the full volume list on
+// every heartbeat for the rest of the process's life.
+func TestDeletedCollectionReleasesTheLookupIndex(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024), changedTestVolume(2, 1024),
+	}, dn)
+	if !dn.HasConsistentVolumeIndex() {
+		t.Fatal("expected a freshly synced node to be consistent")
+	}
+
+	topo.DeleteCollection("c")
+	// The node still holds the volumes until a heartbeat names their
+	// departure; this is that heartbeat.
+	topo.IncrementalSyncDataNodeRegistration(nil, []*master_pb.VolumeShortInformationMessage{
+		{Id: 1, Collection: "c", Version: 3},
+		{Id: 2, Collection: "c", Version: 3},
+	}, dn)
+
+	if !dn.HasConsistentVolumeIndex() {
+		t.Fatal("the deleted collection's lookup entries were not released")
+	}
+}
+
+// A full list races the growth that runs while it is in flight: collected
+// before the grow finished, it cannot name the volumes the grow registered.
+// Erasing them strands their collection without writable volumes until a
+// later report happens to re-add them.
+func TestStaleFullListDoesNotEraseAFreshGrow(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	full := []*master_pb.VolumeInformationMessage{changedTestVolume(1, 1024)}
+	topo.SyncDataNodeRegistration(full, dn)
+
+	// what volume growth registers, after the list above was collected
+	vi, err := storage.NewVolumeInfo(changedTestVolume(2, 8))
+	if err != nil {
+		t.Fatal(err)
+	}
+	dn.AddProvisionalVolume(vi)
+	topo.RegisterVolumeLayout(vi, dn)
+
+	// the stale list arrives
+	topo.SyncDataNodeRegistration(full, dn)
+	if _, err := dn.GetVolumesById(needle.VolumeId(2)); err != nil {
+		t.Fatal("a stale full list erased a freshly grown volume")
+	}
+
+	// once a report names it, a list without it means it is really gone
+	confirmed := append(append([]*master_pb.VolumeInformationMessage{}, full...), changedTestVolume(2, 8))
+	topo.SyncDataNodeRegistration(confirmed, dn)
+	topo.SyncDataNodeRegistration(full, dn)
+	if _, err := dn.GetVolumesById(needle.VolumeId(2)); err == nil {
+		t.Fatal("a confirmed volume survived a list that dropped it")
+	}
+}

weed/storage/disk_location.go

@@ -429,7 +429,10 @@ func (l *DiskLocation) reconcileCompactStates() {
 	}
 }
 
-func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (e error) {
+// DeleteCollectionFromDiskLocation destroys the collection's volumes and ec
+// shards, and returns the volumes it destroyed so the caller can tell the
+// master they are gone.
+func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (deleted []*Volume, e error) {
 
 	l.volumesLock.Lock()
 	delVolsMap := l.unmountVolumeByCollection(collection)
@@ -450,6 +453,7 @@ func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (e er
 				l.volumesLock.Lock()
 				delete(l.volumes, k)
 				l.volumesLock.Unlock()
+				deleted = append(deleted, v)
 			}
 		}
 		wg.Done()

weed/storage/store.go

@@ -243,12 +243,26 @@ func (s *Store) AddVolume(volumeId needle.VolumeId, collection string, needleMap
 
 func (s *Store) DeleteCollection(collection string) (e error) {
 	for _, location := range s.Locations {
-		e = location.DeleteCollectionFromDiskLocation(collection)
-		if e != nil {
-			return
+		deleted, err := location.DeleteCollectionFromDiskLocation(collection)
+		// Name every volume destroyed. Waiting for the next heartbeat to say so
+		// by omission only works while heartbeats carry the whole list, and a
+		// volume grown and destroyed between two of them was never reported at
+		// all, so nothing else would ever tell the master its slot came free.
+		for _, v := range deleted {
+			s.DeletedVolumesChan <- &master_pb.VolumeShortInformationMessage{
+				Id:               uint32(v.Id),
+				Collection:       v.Collection,
+				ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
+				Version:          uint32(v.Version()),
+				Ttl:              v.Ttl.ToUint32(),
+				DiskType:         string(location.DiskType),
+				DiskId:           v.diskId,
+			}
+		}
+		if err != nil {
+			return err
 		}
 		stats.DeleteCollectionMetrics(collection)
-		// let the heartbeat send the list of volumes, instead of sending the deleted volume ids to DeletedVolumesChan
 	}
 	return
 }
@@ -420,7 +434,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	// Volumes skipped below -- quarantined, phantom, expired -- are in neither.
 	var volumeDigest uint64
 	sendFullList, reportGeneration := s.volumeReport.begin()
-	reportedHashes := make(map[volumeReportKey]uint64)
+	reported := make(map[volumeReportKey]reportedVolume)
 	maxVolumeCounts := make(map[string]uint32)
 	// Per-disk effective max for DiskTag, captured alongside the per-type sum.
 	diskMaxByID := make(map[int]int32)
@@ -495,7 +509,18 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 			if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
 				reportHash := reportHashOf(volumeMessage)
 				volumeDigest ^= reportHash
-				reportedHashes[volumeReportKey{diskId: volumeMessage.DiskId, volumeId: volumeMessage.Id}] = reportHash
+				reported[volumeReportKey{diskId: volumeMessage.DiskId, volumeId: volumeMessage.Id}] = reportedVolume{
+					hash: reportHash,
+					short: &master_pb.VolumeShortInformationMessage{
+						Id:               volumeMessage.Id,
+						Collection:       volumeMessage.Collection,
+						ReplicaPlacement: volumeMessage.ReplicaPlacement,
+						Version:          volumeMessage.Version,
+						Ttl:              volumeMessage.Ttl,
+						DiskType:         volumeMessage.DiskType,
+						DiskId:           volumeMessage.DiskId,
+					},
+				}
 				if sendFullList || s.volumeReport.changed(volumeMessage, reportHash) {
 					volumeMessages = append(volumeMessages, volumeMessage)
 				}
@@ -593,7 +618,16 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		}
 	}
 
-	s.volumeReport.commit(reportedHashes, reportGeneration)
+	// A delta says nothing through silence, so volumes gone since the last
+	// report -- a deleted collection, an expired ttl -- must be named, or the
+	// master counts them until a digest mismatch buys it a full list. A full
+	// list needs no such naming: it is already the whole truth.
+	var departedVolumes []*master_pb.VolumeShortInformationMessage
+	if !sendFullList {
+		departedVolumes = s.volumeReport.departed(reported)
+	}
+
+	s.volumeReport.commit(reported, reportGeneration)
 
 	// has_no_volumes says the server holds nothing, so it may only be derived
 	// from a full list. Deriving it from a changed-only heartbeat would make a
@@ -619,6 +653,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		Rack:            s.rack,
 		Volumes:         heartbeatVolumes,
 		ChangedVolumes:  changedVolumes,
+		DeletedVolumes:  departedVolumes,
 		VolumeDigest:    &volumeDigest,
 		DeletedEcShards: deletedEcVolumes,
 		HasNoVolumes:    hasNoVolumes,

weed/storage/store_delete_collection_test.go

@@ -0,0 +1,46 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
+func mountCollectionVolume(t *testing.T, loc *DiskLocation, vid needle.VolumeId, collection string) {
+	t.Helper()
+	v, err := NewVolume(loc.Directory, loc.IdxDirectory, collection, vid, NeedleMapInMemory,
+		&super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	loc.SetVolume(vid, v)
+}
+
+// A collection deleted between two heartbeats is the deletion no report would
+// ever name: its volumes can be grown and destroyed without a single list
+// mentioning them, so nothing but this would tell the master their slots came
+// free.
+func TestDeleteCollectionNamesTheVolumesItDestroyed(t *testing.T) {
+	store := newTestStore(t, 1)
+	mountCollectionVolume(t, store.Locations[0], 1, "books")
+	mountCollectionVolume(t, store.Locations[0], 2, "books")
+	mountCollectionVolume(t, store.Locations[0], 3, "movies")
+
+	if err := store.DeleteCollection("books"); err != nil {
+		t.Fatal(err)
+	}
+
+	named := make(map[uint32]string)
+	for len(store.DeletedVolumesChan) > 0 {
+		m := <-store.DeletedVolumesChan
+		named[m.Id] = m.Collection
+	}
+	if len(named) != 2 || named[1] != "books" || named[2] != "books" {
+		t.Fatalf("delete named %v, want volumes 1 and 2 of books", named)
+	}
+
+	if _, found := store.Locations[0].FindVolume(3); !found {
+		t.Error("another collection's volume was destroyed")
+	}
+}

weed/storage/store_volume_report.go

@@ -14,6 +14,13 @@ type volumeReportKey struct {
 	volumeId uint32
 }
 
+// reportedVolume is what the master was told about one volume copy: the hash
+// that detects change, and enough identity to name the volume if it departs.
+type reportedVolume struct {
+	hash  uint64
+	short *master_pb.VolumeShortInformationMessage
+}
+
 // volumeReportState remembers what the master was last told about each volume,
 // so a heartbeat can carry only what moved since.
 //
@@ -30,7 +37,7 @@ type volumeReportState struct {
 	// fullListGeneration counts requests for the whole list, so one arriving
 	// while a heartbeat is being built is not marked satisfied by it.
 	fullListGeneration uint64
-	lastReported       map[volumeReportKey]uint64
+	lastReported       map[volumeReportKey]reportedVolume
 }
 
 // reset drops everything known about the master's view.
@@ -70,12 +77,41 @@ func (s *volumeReportState) changed(m *master_pb.VolumeInformationMessage, hash
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	previous, known := s.lastReported[volumeReportKey{diskId: m.DiskId, volumeId: m.Id}]
-	return !known || previous != hash
+	return !known || previous.hash != hash
+}
+
+// departed returns the volumes the master was told about that the current
+// report no longer holds on any disk. A delta heartbeat says nothing through
+// silence, so these must be named or the master keeps counting them until a
+// digest mismatch buys it a full list — long enough for a busy cluster to run
+// its free-slot accounting dry. A volume that moved disks is still held, so it
+// is not a departure.
+func (s *volumeReportState) departed(current map[volumeReportKey]reportedVolume) []*master_pb.VolumeShortInformationMessage {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.lastReported) == 0 {
+		return nil
+	}
+	liveIds := make(map[uint32]bool, len(current))
+	for key := range current {
+		liveIds[key.volumeId] = true
+	}
+	var gone []*master_pb.VolumeShortInformationMessage
+	for key, prior := range s.lastReported {
+		if _, still := current[key]; still {
+			continue
+		}
+		if liveIds[key.volumeId] {
+			continue
+		}
+		gone = append(gone, prior.short)
+	}
+	return gone
 }
 
 // commit records what this heartbeat told the master. Volumes absent from
 // reported are forgotten, so one that comes back is reported again.
-func (s *volumeReportState) commit(reported map[volumeReportKey]uint64, generation uint64) {
+func (s *volumeReportState) commit(reported map[volumeReportKey]reportedVolume, generation uint64) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.lastReported = reported

weed/storage/store_volume_report_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 )
 
 func reportingStore(t *testing.T, vids ...needle.VolumeId) *Store {
@@ -147,9 +148,78 @@ func TestFullListRequestDuringCollectionSurvives(t *testing.T) {
 		t.Fatal("expected to be past the first full list")
 	}
 	store.RequestFullVolumeList()
-	store.volumeReport.commit(map[volumeReportKey]uint64{}, generation)
+	store.volumeReport.commit(map[volumeReportKey]reportedVolume{}, generation)
 
 	if heartbeat := store.CollectHeartbeat(); len(heartbeat.Volumes) != 2 {
 		t.Errorf("a resend request made during collection was lost: %d volumes sent", len(heartbeat.Volumes))
 	}
 }
+
+// A delta says nothing through silence, so a volume that disappears must be
+// named or the master keeps counting it until a digest mismatch buys a full
+// list. This is what keeps a collection delete from leaving phantom volumes
+// in the master's free-slot accounting.
+func TestDepartedVolumeIsNamedInTheDelta(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	store.Locations[0].UnloadVolume(needle.VolumeId(2))
+	heartbeat := store.CollectHeartbeat()
+	if len(heartbeat.DeletedVolumes) != 1 || heartbeat.DeletedVolumes[0].Id != 2 {
+		t.Fatalf("expected volume 2 to be named as departed, got %v", heartbeat.DeletedVolumes)
+	}
+
+	// Named once: the next quiet heartbeat has nothing left to say about it.
+	if next := store.CollectHeartbeat(); len(next.DeletedVolumes) != 0 {
+		t.Errorf("a departure was reported twice: %v", next.DeletedVolumes)
+	}
+}
+
+// A full list is already the whole truth; naming departures beside it would
+// tell the master to remove what the list already excludes.
+func TestFullListCarriesNoDepartures(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	store.Locations[0].UnloadVolume(needle.VolumeId(2))
+	store.RequestFullVolumeList()
+	heartbeat := store.CollectHeartbeat()
+	if len(heartbeat.Volumes) != 1 {
+		t.Fatalf("full list carried %d volumes, want 1", len(heartbeat.Volumes))
+	}
+	if len(heartbeat.DeletedVolumes) != 0 {
+		t.Errorf("a full list named departures: %v", heartbeat.DeletedVolumes)
+	}
+}
+
+// A volume that moved disks is still held; naming it as departed would have
+// the master unregister a volume the same heartbeat re-adds.
+func TestMovedVolumeIsNotADeparture(t *testing.T) {
+	store := newTestStore(t, 2)
+	mountTestVolume(t, store.Locations[0], 1)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	store.Locations[0].UnloadVolume(needle.VolumeId(1))
+	// mountTestVolume leaves diskId at zero; the real mount path stamps the
+	// destination disk, which is what makes the copy a different report key.
+	moved, err := NewVolume(store.Locations[1].Directory, store.Locations[1].IdxDirectory, "", 1,
+		NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	moved.diskId = 1
+	store.Locations[1].SetVolume(needle.VolumeId(1), moved)
+	heartbeat := store.CollectHeartbeat()
+	if len(heartbeat.DeletedVolumes) != 0 {
+		t.Errorf("a moved volume was named as departed: %v", heartbeat.DeletedVolumes)
+	}
+	if len(heartbeat.ChangedVolumes) != 1 || heartbeat.ChangedVolumes[0].Id != 1 {
+		t.Errorf("the moved volume was not reported as changed: %v", heartbeat.ChangedVolumes)
+	}
+}

weed/topology/collection.go

@@ -75,6 +75,9 @@ func (c *Collection) DeleteVolumeLayout(rp *super_block.ReplicaPlacement, ttl *n
 	if diskType != types.HardDriveType {
 		keyString += string(diskType)
 	}
+	if vl, found := c.GetVolumeLayout(rp, ttl, diskType); found {
+		vl.releaseLookupOwnership()
+	}
 	c.storageType2VolumeLayout.Delete(keyString)
 }
 

weed/topology/data_node.go

@@ -79,6 +79,15 @@ func (dn *DataNode) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged
 	return disk.AddOrUpdateVolume(v)
 }
 
+// AddProvisionalVolume records a volume the master registered on its own,
+// ahead of any server report naming it. See Disk.AddProvisionalVolume.
+func (dn *DataNode) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
+	dn.Lock()
+	defer dn.Unlock()
+	disk := dn.getOrCreateDisk(v.DiskType)
+	return disk.AddProvisionalVolume(v)
+}
+
 // UpdateVolumes detects new/deleted/changed volumes on a volume server
 // used in master to notify master clients of these changes.
 func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolumes, deletedVolumes, changedVolumes []storage.VolumeInfo) {

weed/topology/disk.go

@@ -5,6 +5,7 @@ import (
 	"slices"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -35,8 +36,20 @@ type Disk struct {
 	// reads from. The two indexes are maintained separately and have been seen
 	// to drift.
 	volumeIdDigest uint64
+	// volumeAddedAt remembers when each volume reached this view of the disk
+	// without a server report having confirmed it yet. Registration by the
+	// master itself -- volume growth -- races the heartbeat in flight, which
+	// cannot name a volume created after it was collected.
+	volumeAddedAt map[needle.VolumeId]time.Time
 }
 
+// volumeRemovalGracePeriod is how long an unconfirmed volume survives a report
+// that does not name it. Removing a just-grown volume strands its collection
+// without writable volumes, so the report that raced the grow does not get to
+// erase it; the cap keeps a registration that never materializes server-side
+// from lingering forever.
+const volumeRemovalGracePeriod = 10 * time.Second
+
 // ecShardSlots returns the number of volume slots consumed by the given
 // number of EC shards, rounded up to whole-volume equivalents.
 func ecShardSlots(ecShardCount int64) int64 {
@@ -49,6 +62,7 @@ func NewDisk(diskType string) *Disk {
 	s.nodeType = "Disk"
 	s.diskUsages = newDiskUsages()
 	s.volumes = make(map[needle.VolumeId]storage.VolumeInfo, 2)
+	s.volumeAddedAt = make(map[needle.VolumeId]time.Time, 2)
 	s.ecShards = make(map[needle.VolumeId]map[types.DiskId]*erasure_coding.EcVolumeInfo, 2)
 	s.NodeImpl.value = s
 	return s
@@ -165,13 +179,25 @@ func (d *Disk) String() string {
 func (d *Disk) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
 	d.Lock()
 	defer d.Unlock()
-	return d.doAddOrUpdateVolume(v)
+	return d.doAddOrUpdateVolume(v, true)
 }
 
-func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
+// AddProvisionalVolume records a volume the master registered on its own --
+// volume growth -- before any server report has named it. Until one does, the
+// volume is protected from removal by a report that raced its creation.
+func (d *Disk) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged bool) {
+	d.Lock()
+	defer d.Unlock()
+	return d.doAddOrUpdateVolume(v, false)
+}
+
+func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo, fromReport bool) (isNew, isChanged bool) {
 	deltaDiskUsage := &DiskUsageCounts{}
 	if oldV, ok := d.volumes[v.Id]; !ok {
 		d.volumes[v.Id] = v
+		if !fromReport {
+			d.volumeAddedAt[v.Id] = time.Now()
+		}
 		d.volumeDigest ^= v.ReportHash()
 		d.volumeIdDigest ^= VolumeIdDigestHash(v.Id)
 		deltaDiskUsage.volumeCount = 1
@@ -195,6 +221,9 @@ func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool)
 			d.UpAdjustDiskUsageDelta(types.ToDiskType(v.DiskType), deltaDiskUsage)
 		}
 		d.volumeDigest ^= oldV.ReportHash() ^ v.ReportHash()
+		if fromReport {
+			delete(d.volumeAddedAt, v.Id)
+		}
 		isChanged = d.volumes[v.Id].ReadOnly != v.ReadOnly
 		if isChanged {
 			// Adjust active volume count when ReadOnly status changes
@@ -242,13 +271,26 @@ func (d *Disk) RemoveVolumesNotIn(reported *reportedVolumes) (removed []storage.
 	diskTypeIndex := reported.diskTypeIndex(string(d.Id()))
 	d.Lock()
 	defer d.Unlock()
+	now := time.Now()
 	for vid, v := range d.volumes {
-		if !reported.namedOn(vid, diskTypeIndex) {
-			removed = append(removed, v)
-			delete(d.volumes, vid)
-			d.volumeDigest ^= v.ReportHash()
-			d.volumeIdDigest ^= VolumeIdDigestHash(vid)
+		if reported.namedOn(vid, diskTypeIndex) {
+			// The server confirmed this volume; from here on its absence from
+			// a report is meaningful.
+			delete(d.volumeAddedAt, vid)
+			continue
 		}
+		// A volume the master registered itself and no report has confirmed
+		// yet is likely racing the list being applied, which was collected
+		// before the grow finished. Explicitly reported deletions still
+		// remove immediately through DeleteVolumeById.
+		if addedAt, unconfirmed := d.volumeAddedAt[vid]; unconfirmed && now.Sub(addedAt) < volumeRemovalGracePeriod {
+			continue
+		}
+		removed = append(removed, v)
+		delete(d.volumes, vid)
+		delete(d.volumeAddedAt, vid)
+		d.volumeDigest ^= v.ReportHash()
+		d.volumeIdDigest ^= VolumeIdDigestHash(vid)
 	}
 	return removed
 }
@@ -271,6 +313,7 @@ func (d *Disk) DeleteVolumeById(id needle.VolumeId) {
 		d.volumeDigest ^= v.ReportHash()
 		d.volumeIdDigest ^= VolumeIdDigestHash(id)
 		delete(d.volumes, id)
+		delete(d.volumeAddedAt, id)
 	}
 }
 

weed/topology/topology.go

@@ -499,6 +499,15 @@ func (t *Topology) FindCollection(collectionName string) (*Collection, bool) {
 }
 
 func (t *Topology) DeleteCollection(collectionName string) {
+	// The layouts vanish with the collection, but every location they served
+	// holds a bit in its node's lookup digest. Left in place, those bits keep
+	// the node's held and servable digests apart forever, and the master asks
+	// for the full volume list on every heartbeat from then on.
+	if collection, found := t.FindCollection(collectionName); found {
+		for _, vl := range collection.GetAllVolumeLayouts() {
+			vl.releaseLookupOwnership()
+		}
+	}
 	t.collectionMap.Delete(collectionName)
 }
 

weed/topology/volume_growth.go

@@ -392,7 +392,7 @@ func (vg *VolumeGrowth) grow(grpcDialOption grpc.DialOption, topo *Topology, vid
 	if growErr == nil {
 		for i, vi := range createdVolumes {
 			server := servers[i]
-			server.AddOrUpdateVolume(vi)
+			server.AddProvisionalVolume(vi)
 			topo.RegisterVolumeLayout(vi, server)
 			glog.V(0).Infof("Registered Volume %d on %s", vid, server.NodeImpl.String())
 		}

weed/topology/volume_index_digest.go

@@ -39,6 +39,20 @@ func (dn *DataNode) HasConsistentVolumeIndex() bool {
 	return held == servable
 }
 
+// releaseLookupOwnership clears the lookup digest bit of every location this
+// layout serves. Used when the layout is dropped wholesale with its collection
+// rather than volume by volume.
+func (vl *VolumeLayout) releaseLookupOwnership() {
+	vl.accessLock.Lock()
+	defer vl.accessLock.Unlock()
+	for vid, location := range vl.vid2location {
+		for _, dn := range location.list {
+			moveLookupOwnership(vid, dn, nil)
+		}
+	}
+	vl.vid2location = make(map[needle.VolumeId]*VolumeLocationList)
+}
+
 // moveLookupOwnership transfers the digest bit for vid from the node a lookup
 // entry used to name to the node it now names. Either may be nil, for an entry
 // being created or dropped.