reduce the window size between unregistering a volume server and creating volumes on that server

fix https://github.com/seaweedfs/seaweedfs/issues/4467

3 files changed, 13 insertions, 7 deletions

diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go
index ee0d37956..5e48056f7 100644
--- a/weed/topology/data_node.go
+++ b/weed/topology/data_node.go
@@ -14,12 +14,13 @@ import (
 
 type DataNode struct {
 	NodeImpl
-	Ip        string
-	Port      int
-	GrpcPort  int
-	PublicUrl string
-	LastSeen  int64 // unix time in seconds
-	Counter   int   // in race condition, the previous dataNode was not dead
+	Ip            string
+	Port          int
+	GrpcPort      int
+	PublicUrl     string
+	LastSeen      int64 // unix time in seconds
+	Counter       int   // in race condition, the previous dataNode was not dead
+	IsTerminating bool
 }
 
 func NewDataNode(id string) *DataNode {
diff --git a/weed/topology/node.go b/weed/topology/node.go
index 89a6305e8..a42146339 100644
--- a/weed/topology/node.go
+++ b/weed/topology/node.go
@@ -179,7 +179,11 @@ func (n *NodeImpl) ReserveOneVolume(r int64, option *VolumeGrowOption) (assigned
 		} else {
 			if node.IsDataNode() && node.AvailableSpaceFor(option) > 0 {
 				// fmt.Println("vid =", vid, " assigned to node =", node, ", freeSpace =", node.FreeSpace())
-				return node.(*DataNode), nil
+				dn := node.(*DataNode)
+				if dn.IsTerminating {
+					continue
+				}
+				return dn, nil
 			}
 			assignedNode, err = node.ReserveOneVolume(r, option)
 			if err == nil {
diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go
index 0a6c1a6c6..761649ff4 100644
--- a/weed/topology/topology_event_handling.go
+++ b/weed/topology/topology_event_handling.go
@@ -82,6 +82,7 @@ func (t *Topology) SetVolumeCrowded(volumeInfo storage.VolumeInfo) {
 }
 
 func (t *Topology) UnRegisterDataNode(dn *DataNode) {
+	dn.IsTerminating = true
 	for _, v := range dn.GetVolumes() {
 		glog.V(0).Infoln("Removing Volume", v.Id, "from the dead volume server", dn.Id())
 		diskType := types.ToDiskType(v.DiskType)