fix(topology): recover heartbeat-fulled volumes once they shrink (#9742)

A volume removed from writables by the write-assign path stamps fullSince
in RecordAssign, which UpdateVolumeSize's recovery branch needs to re-add
it once it decays back under the limit. A volume removed by the heartbeat
capacity path (SetVolumeCapacityFull) never stamped it, so after the
reported size dropped — vacuum, TTL expiry, bulk deletes — the volume
stayed out of writables forever, even though every heartbeat carried the
smaller size.

Stamp fullSince when the capacity path actually removes a volume from
writables, so the existing recovery branch fires. Gating on the removal
keeps it paired with the caller's activeVolumeCount decrement, matching
RecordAssign. Oversized volumes still stay out, as before.

weed/topology/volume_layout.go

@@ -108,11 +108,11 @@ type volumeSizeTracking struct {
 	reportedSize    uint64    // last heartbeat-reported size (dedup replicas)
 	compactRevision uint32    // detect compaction to reset instead of decay
 	lastUpdateTime  time.Time // dedup replicas within the same heartbeat cycle
-	fullSince       time.Time // non-zero while the volume is eagerly marked full by RecordAssign
+	fullSince       time.Time // non-zero while the volume is marked full (RecordAssign or capacity heartbeat)
 }
 
 // capacityRecoveryDelay is the minimum time a volume must stay out of the
-// writable list after being eagerly removed by RecordAssign before it can
+// writable list after being removed for capacity before it can
 // be considered for re-addition by heartbeat-driven decay. Combined with
 // the effectiveSize hysteresis band, this avoids bouncing the volume in
 // and out of writable within a single burst of assigns.
@@ -821,6 +821,13 @@ func (vl *VolumeLayout) SetVolumeCapacityFull(vid needle.VolumeId) bool {
 
 	wasWritable := vl.removeFromWritable(vid)
 	if wasWritable {
+		// Stamp fullSince so UpdateVolumeSize's recovery branch can re-add the
+		// volume once it shrinks; RecordAssign does this for the write path.
+		// Only on actual removal, to stay paired with the activeVolumeCount
+		// decrement the caller does for the same bool.
+		if st := vl.sizeTracking[vid]; st != nil && st.fullSince.IsZero() {
+			st.fullSince = time.Now()
+		}
 		glog.V(0).Infof("Volume %d reaches full capacity.", vid)
 	}
 	return wasWritable

weed/topology/volume_layout_capacity_recovery_test.go

@@ -0,0 +1,69 @@
+package topology
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// A volume removed by the heartbeat capacity path (not RecordAssign) must
+// still recover once it shrinks — only works if SetVolumeCapacityFull stamps
+// fullSince.
+func TestSetVolumeCapacityFullStampsFullSinceAndRecovers(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo, vl := setupPickTest(t, layout, 10000)
+
+	initialActive := topo.diskUsages.usages[types.HardDriveType].activeVolumeCount
+	initialWritables, _ := vl.GetWritableVolumeCount()
+	if initialWritables != 1 {
+		t.Fatalf("expected 1 writable volume initially, got %d", initialWritables)
+	}
+
+	if !vl.SetVolumeCapacityFull(1) {
+		t.Fatalf("SetVolumeCapacityFull should report the volume was writable")
+	}
+	vl.AdjustActiveVolumeCountForFull(1)
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected 0 writable after capacity full, got %d", w)
+	}
+	if vl.sizeTracking[1].fullSince.IsZero() {
+		t.Fatalf("expected SetVolumeCapacityFull to stamp fullSince")
+	}
+
+	// No recovery before capacityRecoveryDelay.
+	advanceSizeTrackingClock(vl, 1, 3*time.Second)
+	if vl.UpdateVolumeSize(1, 4000, 0) {
+		t.Fatalf("recovery should not fire before capacityRecoveryDelay")
+	}
+
+	// After the delay, a smaller size restores it.
+	advanceSizeTrackingClock(vl, 1, capacityRecoveryDelay)
+	if !vl.UpdateVolumeSize(1, 4000, 0) {
+		t.Fatalf("expected volume to recover to writable after shrinking")
+	}
+	vl.AdjustActiveVolumeCountAfterRecovery(1)
+
+	if w, _ := vl.GetWritableVolumeCount(); w != initialWritables {
+		t.Fatalf("expected %d writable after recovery, got %d", initialWritables, w)
+	}
+	if got := topo.diskUsages.usages[types.HardDriveType].activeVolumeCount; got != initialActive {
+		t.Fatalf("expected activeVolumeCount restored to %d, got %d", initialActive, got)
+	}
+	if !vl.sizeTracking[1].fullSince.IsZero() {
+		t.Fatalf("expected fullSince cleared after recovery")
+	}
+}