From a2ffc7aadfcc35b4eb30b58a220a31e10f280357 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 8 Aug 2026 20:23:10 -0700 Subject: [PATCH] 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 | 15 ++-- ...master_grpc_server_changed_volumes_test.go | 58 +++++++++++++++ weed/storage/disk_location.go | 6 +- weed/storage/store.go | 49 +++++++++++-- weed/storage/store_delete_collection_test.go | 46 ++++++++++++ weed/storage/store_volume_report.go | 42 ++++++++++- weed/storage/store_volume_report_test.go | 72 ++++++++++++++++++- weed/topology/collection.go | 3 + weed/topology/data_node.go | 9 +++ weed/topology/disk.go | 57 +++++++++++++-- weed/topology/topology.go | 9 +++ weed/topology/volume_growth.go | 2 +- weed/topology/volume_index_digest.go | 14 ++++ 13 files changed, 358 insertions(+), 24 deletions(-) create mode 100644 weed/storage/store_delete_collection_test.go diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index a6f46a6a5..2eb70165c 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -660,10 +660,17 @@ impl VolumeServer for VolumeGrpcService { ) -> Result, 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 {})) } diff --git a/weed/server/master_grpc_server_changed_volumes_test.go b/weed/server/master_grpc_server_changed_volumes_test.go index 42fee2ded..3770732ab 100644 --- a/weed/server/master_grpc_server_changed_volumes_test.go +++ b/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") + } +} diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 414603e22..4eadb03f7 100644 --- a/weed/storage/disk_location.go +++ b/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() diff --git a/weed/storage/store.go b/weed/storage/store.go index aa66ce46b..22e333f5e 100644 --- a/weed/storage/store.go +++ b/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, diff --git a/weed/storage/store_delete_collection_test.go b/weed/storage/store_delete_collection_test.go new file mode 100644 index 000000000..303756712 --- /dev/null +++ b/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") + } +} diff --git a/weed/storage/store_volume_report.go b/weed/storage/store_volume_report.go index 11e4dd326..059e78bed 100644 --- a/weed/storage/store_volume_report.go +++ b/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 diff --git a/weed/storage/store_volume_report_test.go b/weed/storage/store_volume_report_test.go index dbbee7a43..7a5697f16 100644 --- a/weed/storage/store_volume_report_test.go +++ b/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) + } +} diff --git a/weed/topology/collection.go b/weed/topology/collection.go index 69538a495..c8c0841ba 100644 --- a/weed/topology/collection.go +++ b/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) } diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index 7226aac8d..847d30cf8 100644 --- a/weed/topology/data_node.go +++ b/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) { diff --git a/weed/topology/disk.go b/weed/topology/disk.go index f6755f706..5dd250fc0 100644 --- a/weed/topology/disk.go +++ b/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) } } diff --git a/weed/topology/topology.go b/weed/topology/topology.go index ce74977d8..c94765a2b 100644 --- a/weed/topology/topology.go +++ b/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) } diff --git a/weed/topology/volume_growth.go b/weed/topology/volume_growth.go index 2ccf85a68..c4aff30fe 100644 --- a/weed/topology/volume_growth.go +++ b/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()) } diff --git a/weed/topology/volume_index_digest.go b/weed/topology/volume_index_digest.go index 2812fd4fa..afb18293d 100644 --- a/weed/topology/volume_index_digest.go +++ b/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.