topology: refresh oversized mark on every heartbeat (#10829)

* topology: refresh oversized mark on every heartbeat

The oversized flag on a volume location was only set when the volume was
registered (RegisterVolume). A volume that later grew past the size limit
kept its stale "not oversized" mark, so the heartbeat path
(ensureCorrectWritables) kept re-adding it to the writable list while
RecordAssign removed it on every assign - a writable/unwritable flip loop
that let writes continue past the limit and made vacuum race in-flight
writes.

Refresh the mark from each heartbeat's reported size in both heartbeat
paths (ApplyVolumeChanges and SyncDataNodeRegistration), mirroring what
RegisterVolume already did at registration time. A volume that grew past
the limit now stays unwritable, and one that shrank back clears the mark
and can recover.

* topology: order heartbeat writable correction after decay and honor cooldown

Review feedback (Greptile, CodeRabbit) on the oversized-mark refresh:

1. Greptile: clearing the oversized mark before EnsureCorrectWritables let
   the delay-unaware helper re-add a just-compacted volume to writables,
   bypassing capacityRecoveryDelay. ensureCorrectWritables now checks
   fullSince and skips the re-add while the cooldown is pending, so a
   volume removed for capacity only recovers through UpdateVolumeSize's
   heartbeat recovery path.

2. CodeRabbit: in the full-heartbeat path the mark was refreshed after
   the writable correction, so a newly oversized volume stayed writable
   for an extra heartbeat cycle. The standalone changedVolumes loop is
   merged into the volumeInfos loop and EnsureCorrectWritables now runs
   after UpdateOversizedState + UpdateVolumeSize in both heartbeat paths,
   using the freshly refreshed mark.

3. TestHandlingVolumeServerHeartbeat used a size (254320) that is past
   the test's volumeSizeLimit (32768); it only passed because the stale
   mark hid the oversized state. Sized down to 30000 to keep testing the
   add/remove flow, and added TestEnsureCorrectWritablesHonorsRecoveryCooldown
   covering the cooldown window and the recovery after it.

* topology: do not restore a still-crowded volume after the cooldown

Greptile review: after capacityRecoveryDelay elapses, ensureCorrectWritables
could restore a volume whose effective size is still past the crowded
threshold. UpdateVolumeSize refuses the recovery (effectiveSize > crowded
threshold -> setVolumeCrowded + return false), but the cooldown check in
ensureCorrectWritables only looked at fullSince, so once the delay passed
it re-added the volume even though capacity tracking still considers it
crowded.

Check the crowded mark before re-adding: a volume UpdateVolumeSize just
marked crowded must not be restored here, otherwise assignments resume
while the volume is still flagged for growth.

Adds TestEnsureCorrectWritablesDoesNotRestoreCrowdedVolume: effectiveSize
decays to 10500 (past the 9000 crowded threshold) after a report of 8000,
and ensureCorrectWritables keeps the volume unwritable past the cooldown.

* ci: trigger re-run of flaky FUSE jobs

* topology: gate the writable restore on the limit, not on crowded

A crowded volume is above the growth threshold, not full, and is normally
writable. Refusing to restore one locks it out for good: nothing writes to
a volume that is not writable, so its size can never fall back under the
threshold. Gate on the same size the assign path uses to remove it.

* topology: let only the heartbeat refresh set the oversized mark

Registration also set it, from whatever VolumeInfo it was handed. The
incremental path builds that from a short heartbeat message, which carries
no size, so every arrival announcement cleared the mark and handed the
volume back to the writable list until the next full report.

* topology: use the re-resolved layout after a dropped one is replaced

A layout dropped with its collection makes RegisterVolume refuse, and the
full heartbeat then re-registered against a fresh layout but kept applying
the size, oversized and writable updates to the dropped one.

---------

Co-authored-by: hzsunchao <hzsunchao@corp.netease.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/topology/topology.go

@@ -613,19 +613,13 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 		}
 	}
 	// find out the delta volumes
-	var changedVolumes []storage.VolumeInfo
-	newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos)
+	newVolumes, deletedVolumes, _ = dn.UpdateVolumes(volumeInfos)
 	for _, v := range newVolumes {
 		t.RegisterVolumeLayout(v, dn)
 	}
 	for _, v := range deletedVolumes {
 		t.UnRegisterVolumeLayout(v, dn)
 	}
-	for _, v := range changedVolumes {
-		diskType := types.ToDiskType(v.DiskType)
-		vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
-		vl.EnsureCorrectWritables(&v)
-	}
 	// Update effective sizes for all reported volumes (decay pending estimates).
 	// If decay brings a volume eagerly removed by RecordAssign back under the
 	// writable threshold, restore the matching activeVolumeCount.
@@ -643,11 +637,10 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 		// Without this, the volume stays visible in volume.list/admin UI yet
 		// LookupVolume returns "volume id not found".
 		if !vl.HasDataNode(v.Id, dn) {
-			if vl.RegisterVolume(&v, dn) {
-				vl.EnsureCorrectWritables(&v)
-			} else {
-				// Dropped with its collection; re-resolve.
-				t.RegisterVolumeLayout(v, dn)
+			for !vl.RegisterVolume(&v, dn) {
+				// Dropped with its collection; the next lookup creates a fresh
+				// one, which the calls below must use too.
+				vl = t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 			}
 			// Volumes new to the disk map were registered above, so reaching
 			// here means only the lookup index had lost it. Clients were told
@@ -655,9 +648,11 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 			// it is back.
 			newVolumes = append(newVolumes, v)
 		}
+		vl.UpdateOversizedState(&v, dn)
 		if vl.UpdateVolumeSize(v.Id, v.Size, v.CompactRevision) {
 			vl.AdjustActiveVolumeCountAfterRecovery(v.Id)
 		}
+		vl.EnsureCorrectWritables(&v)
 	}
 	return
 }
@@ -731,10 +726,11 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 		if isNew || becameServable {
 			newVolumes = append(newVolumes, vi)
 		}
-		vl.EnsureCorrectWritables(&vi)
+		vl.UpdateOversizedState(&vi, dn)
 		if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision) {
 			vl.AdjustActiveVolumeCountAfterRecovery(vi.Id)
 		}
+		vl.EnsureCorrectWritables(&vi)
 	}
 	return newVolumes
 }

weed/topology/topology_test.go

@@ -88,7 +88,7 @@ func TestHandlingVolumeServerHeartbeat(t *testing.T) {
 		for k := 1; k <= volumeCount; k++ {
 			volumeMessage := &master_pb.VolumeInformationMessage{
 				Id:               uint32(k),
-				Size:             uint64(254320),
+				Size:             uint64(30000),
 				Collection:       "",
 				FileCount:        uint64(2343),
 				DeleteCount:      uint64(345),

weed/topology/volume_layout.go

@@ -131,8 +131,6 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) bool
 		return false
 	}
 
-	defer vl.rememberOversizedVolume(v, dn)
-
 	moveLookupOwnership(v.Id, vl.getOrCreateLocationList(v.Id).Set(dn), dn)
 	if !v.ReadOnly {
 		vl.initSizeTracking(v.Id, v.Size, v.CompactRevision)
@@ -164,6 +162,23 @@ func (vl *VolumeLayout) rememberOversizedVolume(v *storage.VolumeInfo, dn *DataN
 	}
 }
 
+// UpdateOversizedState refreshes the per-replica oversized mark from the size
+// reported by this heartbeat, and is the only writer of that mark. The delta
+// heartbeat path (ApplyVolumeChanges) and the full heartbeat path
+// (SyncDataNodeRegistration) both call it, or a volume that grew past the
+// limit would keep looking writable to ensureCorrectWritables and bounce
+// between writable and unwritable on every assign.
+//
+// Registration deliberately does not touch the mark: the incremental path
+// registers from a short heartbeat message that carries no size, so judging
+// the volume by it would clear the mark on every arrival announcement and
+// hand an oversized volume back to the writable list.
+func (vl *VolumeLayout) UpdateOversizedState(v *storage.VolumeInfo, dn *DataNode) {
+	vl.accessLock.Lock()
+	defer vl.accessLock.Unlock()
+	vl.rememberOversizedVolume(v, dn)
+}
+
 // UpdateVolumeSize is called on every heartbeat for every reported volume.
 // It decays the pending size estimate toward the reported size and updates
 // crowded state. Replicated volumes report from multiple DataNodes; decay
@@ -291,6 +306,22 @@ func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) {
 	isAllWritable := vl.isAllWritable(vid)
 	isOversizedVolume := vl.vid2location[vid].AnyOversized()
 	if isEnoughCopies && isAllWritable && !isOversizedVolume {
+		// A volume removed for capacity (fullSince set) must go through the
+		// heartbeat recovery path in UpdateVolumeSize, which enforces
+		// capacityRecoveryDelay. Re-adding it here would bypass the cooldown
+		// and let a just-compacted volume flip back to writable immediately.
+		if st := vl.sizeTracking[vid]; st != nil && !st.fullSince.IsZero() &&
+			time.Since(st.fullSince) < capacityRecoveryDelay {
+			return
+		}
+		// A volume already at the limit must not be restored here: the next
+		// assign's RecordAssign would remove it again. Crowded is the wrong
+		// test for that -- it starts at 90%, and a crowded volume that lost a
+		// replica would never be writable again once the replica returned,
+		// since nothing writes to it and its size can no longer fall.
+		if st := vl.sizeTracking[vid]; st != nil && st.effectiveSize >= vl.volumeSizeLimit {
+			return
+		}
 		vl.setVolumeWritable(vid)
 		return
 	}

weed/topology/volume_layout_oversized_heartbeat_test.go

@@ -0,0 +1,318 @@
+package topology
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// A volume that grew past the limit keeps bouncing between writable and
+// unwritable when the oversized mark is not refreshed on heartbeat: RegisterVolume
+// sets it once, but the delta heartbeat path (ApplyVolumeChanges) never re-sets
+// it, so ensureCorrectWritables sees a stale "not oversized" mark and re-adds
+// the volume to writables on the next heartbeat, while RecordAssign removes it
+// on the next assign. UpdateOversizedState must keep the mark current.
+func TestUpdateOversizedStateKeepsOversizedVolumeUnwritable(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Grow the volume past the limit, as RecordAssign would see it.
+	if !vl.RecordAssign(1, 9000) {
+		t.Fatalf("RecordAssign should report the volume reached capacity")
+	}
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected 0 writable after RecordAssign, got %d", w)
+	}
+
+	// A heartbeat that reports the now-oversized volume must refresh the mark
+	// so ensureCorrectWritables does not re-add it to writables. Mirrors the
+	// heartbeat order: refresh, then decay, then correct.
+	dn := vl.vid2location[1].list[0]
+	vi, err := dn.GetVolumesById(1)
+	if err != nil {
+		t.Fatalf("GetVolumesById: %v", err)
+	}
+	oversized := vi
+	oversized.Size = 12000
+	vl.UpdateOversizedState(&oversized, dn)
+	vl.UpdateVolumeSize(1, oversized.Size, 0)
+	vl.EnsureCorrectWritables(&oversized)
+
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected volume to stay unwritable after heartbeat refresh, got %d writable", w)
+	}
+	if !vl.vid2location[1].AnyOversized() {
+		t.Fatalf("expected AnyOversized to be true after heartbeat refresh")
+	}
+}
+
+// A volume that shrank back under the limit must have its oversized mark
+// cleared by the heartbeat refresh, or it stays locked out of writables
+// forever.
+func TestUpdateOversizedStateClearsMarkWhenVolumeShrinks(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	dn := vl.vid2location[1].list[0]
+	vi, err := dn.GetVolumesById(1)
+	if err != nil {
+		t.Fatalf("GetVolumesById: %v", err)
+	}
+
+	// Mark oversized via a heartbeat that reports a huge size.
+	big := vi
+	big.Size = 12000
+	vl.UpdateOversizedState(&big, dn)
+	if !vl.vid2location[1].AnyOversized() {
+		t.Fatalf("expected AnyOversized after a huge report")
+	}
+
+	// A later heartbeat with a normal size clears the mark.
+	vl.UpdateOversizedState(&vi, dn)
+	if vl.vid2location[1].AnyOversized() {
+		t.Fatalf("expected AnyOversized cleared after the volume shrank")
+	}
+}
+
+// A volume removed for capacity must not be re-added to writables by
+// ensureCorrectWritables before capacityRecoveryDelay elapses, even after its
+// oversized mark is cleared by a heartbeat that reports a smaller size.
+func TestEnsureCorrectWritablesHonorsRecoveryCooldown(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Remove the volume for capacity, as RecordAssign would.
+	if !vl.RecordAssign(1, 9000) {
+		t.Fatalf("RecordAssign should report the volume reached capacity")
+	}
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected 0 writable after RecordAssign, got %d", w)
+	}
+
+	dn := vl.vid2location[1].list[0]
+	vi, err := dn.GetVolumesById(1)
+	if err != nil {
+		t.Fatalf("GetVolumesById: %v", err)
+	}
+
+	// Heartbeat reports the volume shrank back under the limit: the oversized
+	// mark clears and effectiveSize decays, but the volume is still within
+	// capacityRecoveryDelay.
+	vi.Size = 4000
+	vl.UpdateOversizedState(&vi, dn)
+	vl.UpdateVolumeSize(1, vi.Size, 0)
+	vl.EnsureCorrectWritables(&vi)
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected volume to stay unwritable during the cooldown, got %d writable", w)
+	}
+
+	// After the cooldown, a heartbeat with the shrunken size recovers it.
+	advanceSizeTrackingClock(vl, 1, capacityRecoveryDelay+time.Second)
+	if !vl.UpdateVolumeSize(1, vi.Size, 0) {
+		t.Fatalf("expected volume to recover to writable after the cooldown")
+	}
+	vl.EnsureCorrectWritables(&vi)
+	if w, _ := vl.GetWritableVolumeCount(); w != 1 {
+		t.Fatalf("expected volume writable after the cooldown, got %d", w)
+	}
+}
+
+// After the cooldown elapses, a volume whose effective size is still at the
+// limit must not be restored by ensureCorrectWritables: the next assign would
+// remove it again.
+func TestEnsureCorrectWritablesDoesNotRestoreVolumeStillAtLimit(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":8000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Remove the volume for capacity, as RecordAssign would. effectiveSize
+	// lands at 13000, past the limit (10000); after the decay against a
+	// reported size of 8000 it stays at 10500, still past the limit.
+	if !vl.RecordAssign(1, 5000) {
+		t.Fatalf("RecordAssign should report the volume reached capacity")
+	}
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected 0 writable after RecordAssign, got %d", w)
+	}
+
+	dn := vl.vid2location[1].list[0]
+	vi, err := dn.GetVolumesById(1)
+	if err != nil {
+		t.Fatalf("GetVolumesById: %v", err)
+	}
+
+	// Heartbeat reports a size below the limit: the oversized mark clears,
+	// but the decay only halves the pending estimate, so effectiveSize stays
+	// past the limit and UpdateVolumeSize refuses recovery.
+	vi.Size = 8000
+	vl.UpdateOversizedState(&vi, dn)
+	vl.UpdateVolumeSize(1, vi.Size, 0)
+	if vl.vid2location[1].AnyOversized() {
+		t.Fatalf("expected oversized mark cleared after the shrink report")
+	}
+
+	// After the cooldown, the volume is still at the limit:
+	// ensureCorrectWritables must not override UpdateVolumeSize's refusal.
+	advanceSizeTrackingClock(vl, 1, capacityRecoveryDelay+time.Second)
+	vl.EnsureCorrectWritables(&vi)
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected volume at the limit to stay unwritable, got %d writable", w)
+	}
+}
+
+// A crowded volume is not a full one: it is above the growth threshold but
+// still has room. One that drops out of writables while a replica is away must
+// come back when the replica returns -- nothing writes to a volume that is not
+// writable, so its size can never fall on its own and the lockout would be
+// permanent.
+func TestEnsureCorrectWritablesRestoresCrowdedVolumeAfterReplicaReturns(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "ip":"10.0.0.1",
+        "volumes":[
+          {"id":1, "size":9500, "replication":"001"}
+        ],
+        "limit":10
+      },
+      "server2":{
+        "ip":"10.0.0.2",
+        "volumes":[
+          {"id":1, "size":9500, "replication":"001"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo := setupWithLimit(t, layout, 10000)
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+	vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType)
+
+	// 9500 is past the growth threshold (9000) but under the limit (10000).
+	vl.UpdateVolumeSize(1, 9500, 0)
+	if _, crowded := vl.crowded[1]; !crowded {
+		t.Fatalf("expected the volume to be crowded")
+	}
+	if w, _ := vl.GetWritableVolumeCount(); w != 1 {
+		t.Fatalf("a crowded volume is still writable, got %d writable", w)
+	}
+
+	dn := vl.vid2location[1].list[1]
+	vi, err := dn.GetVolumesById(1)
+	if err != nil {
+		t.Fatalf("GetVolumesById: %v", err)
+	}
+	vl.UnRegisterVolume(&vi, dn)
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected 0 writable while a replica is missing, got %d", w)
+	}
+
+	// The replica comes back on the next heartbeat.
+	topo.RegisterVolumeLayout(vi, dn)
+	advanceSizeTrackingClock(vl, 1, 5*time.Second)
+	vl.UpdateOversizedState(&vi, dn)
+	vl.UpdateVolumeSize(1, vi.Size, 0)
+	vl.EnsureCorrectWritables(&vi)
+	if w, _ := vl.GetWritableVolumeCount(); w != 1 {
+		t.Fatalf("expected the volume writable again, got %d", w)
+	}
+}
+
+// An incremental heartbeat announces an arrival with a short message that
+// carries no size. Registering from it must not clear the oversized mark the
+// full heartbeat set, or the volume is handed back to the writable list until
+// the next full report.
+func TestIncrementalRegistrationKeepsOversizedMark(t *testing.T) {
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	dc := topo.GetOrCreateDataCenter("dc1")
+	rack := dc.GetOrCreateRack("rack1")
+	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 25})
+
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType)
+
+	// A full heartbeat reports the volume past the 32 KB limit.
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{
+		Id:               1,
+		Size:             uint64(64 * 1024),
+		ReplicaPlacement: uint32(0),
+		Version:          uint32(needle.GetCurrentVersion()),
+	}}, dn)
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected the oversized volume out of writables, got %d", w)
+	}
+
+	// The same volume is announced again as an arrival.
+	topo.IncrementalSyncDataNodeRegistration([]*master_pb.VolumeShortInformationMessage{{
+		Id:               1,
+		ReplicaPlacement: uint32(0),
+		Version:          uint32(needle.GetCurrentVersion()),
+	}}, nil, dn)
+
+	if !vl.vid2location[1].AnyOversized() {
+		t.Fatalf("expected the oversized mark to survive the arrival announcement")
+	}
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected the oversized volume to stay out of writables, got %d", w)
+	}
+}