master: stop hintless small-file assigns from marking volumes full (#10944)

* master: estimate a hintless assign's size from the volume's average file size

An assign that carries no dataSize hint charged a flat 1MB per file id
against the volume's effective size. A small-file workload overpays by
orders of magnitude: bulk-writing 4KB files marks volumes holding a few
hundred MB of real data as crowded and then full, so the master grows
unnecessary volumes and, once every volume is spuriously full, fails all
assigns. Estimate from the volume's own average file size instead, and
keep the 1MB fallback only for volumes with no history.

* master: decay pending assign sizes for volumes gone quiet

The decay that corrects pending assign estimates runs only when a
heartbeat reports the volume, and a heartbeat only reports a volume
whose content changed. A volume held out of the writable list takes no
writes, so once inflated estimates mark every volume full, nothing is
ever reported again, nothing decays, and the cluster refuses all writes
until a restart. Run the decay from the master's periodic loop for
volumes no heartbeat has reported within two pulses, feeding the last
reported size back through the same path an unchanged heartbeat would
take.

* master: trim the comments on the assign size estimate

* master: keep the periodic decay out of the replica-dedup window

UpdateVolumeSize ignores a report arriving within two seconds of the last
one, so replicas of the same volume do not each halve the pending
estimate. The periodic decay went through the same path and stamped that
window, so a real heartbeat landing right behind it was dropped along
with its reported size and compact revision. Only a volume whose content
changed is reported at all, so nothing would send that size again and
the master kept a stale one. Let the dedup window belong to volume
server reports alone.

* master: let the decay read the size record under the lock it mutates

The periodic decay picked its volumes under a read lock and replayed
them under a write one, carrying the size it had read across the gap. A
heartbeat landing in between was rolled back: the replay wrote the older
size and compact revision over the fresh ones, and a compaction report
lost that way is never resent, since only a volume whose content changed
is reported. The decay has no size of its own to contribute, so it now
reads the record under the same lock it mutates.

* master: let a heartbeat that beat the decay stand for the cycle

The decay chooses its volumes under a read lock and applies them under a
write one. A heartbeat landing in that gap already did the halving the
cycle owed, so applying the decay on top of it halved twice and forgot
pending bytes the volume has not written yet - the double-halving the
replica-dedup window exists to prevent. Both callers now give way to a
report already handled for this cycle; only a real report still advances
lastUpdateTime, so a quiet volume keeps decaying every pulse.

* master: keep genuinely full volumes out of the decay pass

A volume the disk really did fill keeps its fullSince set for good, so it
was selected every pulse for a decay that cannot help it: UpdateVolumeSize
refuses to recover a volume whose reported size is at the limit, and
replaying a size that cannot move leaves the record as it found it. Full
and quiet is the ordinary resting state of a cluster, so this was most of
the pass, taking the layout write lock away from the heartbeats to do
nothing. On a million tracked volumes with a hundredth of them phantom-full
it costs ten thousand write locks a pulse instead of a million.

* master: put the stale-replay test back on the path it guards

Giving the decay the dedup window left this test short-circuiting there,
so it no longer reached the locked read it was written for and passed
with that read removed. Age the record past the window, which is the only
case where reading it under the lock is what saves the report.

weed/topology/topology.go

@@ -403,11 +403,16 @@ func (t *Topology) PickForWrite(requestedCount uint64, option *VolumeGrowOption,
 	if volumeLocationList == nil || volumeLocationList.Length() == 0 {
 		return "", 0, nil, shouldGrow, fmt.Errorf("%s available for collection:%s replication:%s ttl:%s", NoWritableVolumes, option.Collection, option.ReplicaPlacement.String(), option.Ttl.String())
 	}
-	// Track estimated assigned bytes to spread load between heartbeats.
-	// Use the client hint if provided, otherwise fall back to 1MB estimate.
+	// Track estimated assigned bytes to spread load between heartbeats. A flat
+	// fallback overcharges a small-file workload enough to mark near-empty
+	// volumes full, so prefer the volume's own average.
 	sizePerFile := DefaultNeedleSizeEstimate
 	if expectedDataSize > 0 {
 		sizePerFile = expectedDataSize
+	} else if vi, infoErr := volumeLocationList.Head().GetVolumesById(vid); infoErr == nil && vi.FileCount > 0 {
+		if avg := vi.Size / uint64(vi.FileCount); avg > 0 {
+			sizePerFile = avg
+		}
 	}
 	pendingBytes := min(uint64(count)*sizePerFile, uint64(math.MaxInt64))
 	if volumeLayout.RecordAssign(vid, int64(pendingBytes)) {
@@ -424,6 +429,18 @@ func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.Replic
 	}).(*Collection).GetOrCreateVolumeLayout(rp, ttl, diskType)
 }
 
+// DecayQuietVolumeSizes decays pending assign estimates across every layout.
+// A volume that changed reports within a pulse, so two quiet pulses mean the
+// size on record is the size there is.
+func (t *Topology) DecayQuietVolumeSizes() {
+	quietCutoff := time.Duration(2*t.pulse) * time.Second
+	for _, c := range t.collectionMap.Items() {
+		for _, vl := range c.(*Collection).GetAllVolumeLayouts() {
+			vl.DecayQuietVolumeSizes(quietCutoff)
+		}
+	}
+}
+
 // CollectionVolumeStats aggregates stats across all volume layouts and EC
 // volumes of one collection, or across every collection when collectionName is
 // empty.
@@ -642,7 +659,7 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 			newVolumes = append(newVolumes, v)
 		}
 		vl.UpdateOversizedState(&v, dn)
-		if vl.UpdateVolumeSize(v.Id, v.Size, v.CompactRevision) {
+		if vl.UpdateVolumeSize(v.Id, v.Size, v.CompactRevision, true) {
 			vl.AdjustActiveVolumeCountAfterRecovery(v.Id)
 		}
 		vl.EnsureCorrectWritables(&v)
@@ -720,7 +737,7 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 			newVolumes = append(newVolumes, vi)
 		}
 		vl.UpdateOversizedState(&vi, dn)
-		if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision) {
+		if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision, true) {
 			vl.AdjustActiveVolumeCountAfterRecovery(vi.Id)
 		}
 		vl.EnsureCorrectWritables(&vi)

weed/topology/topology_event_handling.go

@@ -19,6 +19,7 @@ func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, g
 			if t.IsLeader() {
 				freshThreshHold := time.Now().Unix() - 3*t.pulse //3 times of sleep interval
 				t.CollectDeadNodeAndFullVolumes(freshThreshHold, t.volumeSizeLimit, growThreshold)
+				t.DecayQuietVolumeSizes()
 			}
 			time.Sleep(time.Duration(float32(t.pulse*1e3)*(1+rand.Float32())) * time.Millisecond)
 		}

weed/topology/volume_layout.go

@@ -191,7 +191,14 @@ func (vl *VolumeLayout) UpdateOversizedState(v *storage.VolumeInfo, dn *DataNode
 // previously eagerly removed by RecordAssign back under the writable
 // threshold and this call re-added it to the writable list. The caller
 // should mirror the activeVolumeCount bookkeeping.
-func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint64, compactRevision uint32) (recoveredToWritable bool) {
+//
+// fromHeartbeat marks a volume server's report. The periodic decay stands in
+// for the report a quiet volume never sends, so it passes no size of its own —
+// it reads the record under this lock, and never advances lastUpdateTime, or
+// it would swallow the next real report and strand the master on a stale size
+// no one will send again. Both callers give way to a report already handled
+// for this cycle.
+func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint64, compactRevision uint32, fromHeartbeat bool) (recoveredToWritable bool) {
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
@@ -211,6 +218,9 @@ func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint6
 	now := time.Now()
 	st := vl.sizeTracking[vid]
 	if st == nil {
+		if !fromHeartbeat {
+			return false // the entry went while the decay was picking its work
+		}
 		st = &volumeSizeTracking{
 			effectiveSize:   reportedSize,
 			reportedSize:    reportedSize,
@@ -219,10 +229,21 @@ func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint6
 		}
 		vl.sizeTracking[vid] = st
 	} else if now.Sub(st.lastUpdateTime) < 2*time.Second {
-		return false // duplicate replica in the same heartbeat cycle
+		// Something already decayed this volume for this cycle: another replica
+		// of the same report, or for the decay a heartbeat that arrived after
+		// the pass chose the volume. Halving twice would forget pending bytes
+		// the volume has not written yet.
+		return false
 	} else {
-		st.lastUpdateTime = now
-		st.reportedSize = reportedSize
+		if fromHeartbeat {
+			st.lastUpdateTime = now
+			st.reportedSize = reportedSize
+		} else {
+			// Take the record as it stands under this lock, not as the decay
+			// pass saw it: a heartbeat may have landed since, and replaying
+			// the older size would roll its report back.
+			reportedSize, compactRevision = st.reportedSize, st.compactRevision
+		}
 		if compactRevision != st.compactRevision {
 			// Compaction happened — size drop is real, not pending. Reset.
 			st.compactRevision = compactRevision
@@ -270,6 +291,43 @@ func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint6
 	return true
 }
 
+// DecayQuietVolumeSizes re-runs the size decay for volumes whose heartbeat
+// reports have gone quiet. Only a volume whose content changed is reported,
+// and a volume held out of the writable list takes no writes, so without this
+// an inflated estimate is never decayed and the volume never returns.
+//
+// A volume the disk really did fill is skipped: UpdateVolumeSize refuses to
+// recover one whose reported size is at the limit, so walking it every pulse
+// only takes the write lock away from the heartbeats. Nothing is lost by
+// waiting — shrinking it means compaction, which changes content and is
+// therefore reported.
+func (vl *VolumeLayout) DecayQuietVolumeSizes(quietCutoff time.Duration) {
+	for _, vid := range vl.quietDecayCandidates(quietCutoff) {
+		if vl.UpdateVolumeSize(vid, 0, 0, false) {
+			vl.AdjustActiveVolumeCountAfterRecovery(vid)
+		}
+	}
+}
+
+// quietDecayCandidates names the volumes the decay has something to do for.
+// Skipping the rest is what keeps the pass off the write lock: the work is
+// what costs, not the outcome, since replaying a size that cannot move leaves
+// the record exactly as it found it.
+func (vl *VolumeLayout) quietDecayCandidates(quietCutoff time.Duration) (quiets []needle.VolumeId) {
+	now := time.Now()
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+	for vid, st := range vl.sizeTracking {
+		if now.Sub(st.lastUpdateTime) < quietCutoff {
+			continue
+		}
+		if st.effectiveSize > st.reportedSize || (!st.fullSince.IsZero() && st.reportedSize < vl.volumeSizeLimit) {
+			quiets = append(quiets, vid)
+		}
+	}
+	return quiets
+}
+
 func (vl *VolumeLayout) UnRegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()

weed/topology/volume_layout_capacity_recovery_test.go

@@ -46,13 +46,13 @@ func TestSetVolumeCapacityFullStampsFullSinceAndRecovers(t *testing.T) {
 
 	// No recovery before capacityRecoveryDelay.
 	advanceSizeTrackingClock(vl, 1, 3*time.Second)
-	if vl.UpdateVolumeSize(1, 4000, 0) {
+	if vl.UpdateVolumeSize(1, 4000, 0, true) {
 		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) {
+	if !vl.UpdateVolumeSize(1, 4000, 0, true) {
 		t.Fatalf("expected volume to recover to writable after shrinking")
 	}
 	vl.AdjustActiveVolumeCountAfterRecovery(1)
@@ -109,10 +109,224 @@ func TestSetVolumeAvailableRestoresActiveCountForCapacityFullVolume(t *testing.T
 	}
 
 	advanceSizeTrackingClock(vl, 1, capacityRecoveryDelay+time.Second)
-	if vl.UpdateVolumeSize(1, 4000, 0) {
+	if vl.UpdateVolumeSize(1, 4000, 0, true) {
 		t.Fatalf("heartbeat recovery should not fire after SetVolumeAvailable already restored the volume")
 	}
 	if got := topo.diskUsages.usages[types.HardDriveType].activeVolumeCount; got != initialActive {
 		t.Fatalf("expected activeVolumeCount to remain %d, got %d", initialActive, got)
 	}
 }
+
+// A volume marked full by pending estimates takes no writes, so no heartbeat
+// reports it again and only the periodic decay can bring it back.
+func TestDecayQuietVolumeSizesRecoversPhantomFullVolume(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo, vl := setupPickTest(t, layout, 10000)
+	VolumeGrowStrategy.Threshold = 0.9
+
+	initialActive := topo.diskUsages.usages[types.HardDriveType].activeVolumeCount
+	if !vl.RecordAssign(1, 20000) {
+		t.Fatalf("expected RecordAssign to remove the volume from writable")
+	}
+	vl.AdjustActiveVolumeCountForFull(1)
+	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
+		t.Fatalf("expected 0 writable after the pending estimate filled the volume, got %d", w)
+	}
+
+	recovered := false
+	for i := 0; i < 10 && !recovered; i++ {
+		advanceSizeTrackingClock(vl, 1, capacityRecoveryDelay+time.Second)
+		topo.DecayQuietVolumeSizes()
+		w, _ := vl.GetWritableVolumeCount()
+		recovered = w == 1
+	}
+	if !recovered {
+		t.Fatalf("the periodic decay never returned the volume to the writable list")
+	}
+	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")
+	}
+}
+
+// The decay must not consume the replica-dedup window: a real report landing
+// right behind it carries a size no volume server will send again, since only
+// a volume whose content changed is reported at all.
+func TestDecayQuietVolumeSizesKeepsTheNextHeartbeat(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo, vl := setupPickTest(t, layout, 10000)
+
+	vl.RecordAssign(1, 2000)
+	advanceSizeTrackingClock(vl, 1, 30*time.Second)
+	topo.DecayQuietVolumeSizes()
+
+	vl.UpdateVolumeSize(1, 6000, 7, true)
+
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+	if got := vl.sizeTracking[1].reportedSize; got != 6000 {
+		t.Errorf("the heartbeat after a decay left reportedSize at %d, want the reported 6000", got)
+	}
+	if got := vl.sizeTracking[1].compactRevision; got != 7 {
+		t.Errorf("the heartbeat after a decay left compactRevision at %d, want the reported 7", got)
+	}
+}
+
+// The decay picks its volumes under a read lock and replays them under a write
+// one. A compaction heartbeat landing in that gap must survive, so the replay
+// takes the record as it stands rather than the size the pass set out with —
+// the arguments below stand in for that older snapshot.
+func TestDecayQuietVolumeSizesDoesNotRollBackAHeartbeat(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":9000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	vl.RecordAssign(1, 500)
+	// The heartbeat that wins the race: a compaction shrank the volume.
+	vl.UpdateVolumeSize(1, 2000, 3, true)
+	// Far enough behind that the dedup window no longer covers for the replay,
+	// which is the only case where reading the record under the lock is what
+	// saves the report.
+	advanceSizeTrackingClock(vl, 1, 3*time.Second)
+
+	vl.UpdateVolumeSize(1, 9000, 0, false)
+
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+	if got := vl.sizeTracking[1].reportedSize; got != 2000 {
+		t.Errorf("the decay rolled reportedSize back to %d, want the compacted 2000", got)
+	}
+	if got := vl.sizeTracking[1].compactRevision; got != 3 {
+		t.Errorf("the decay rolled compactRevision back to %d, want the reported 3", got)
+	}
+	if got := vl.sizeTracking[1].effectiveSize; got > 2000 {
+		t.Errorf("effectiveSize %d outgrew the compacted size the heartbeat reported", got)
+	}
+}
+
+// The decay stands in for a report that never came, so a heartbeat arriving
+// after the pass chose the volume takes its place rather than adding to it:
+// halving twice in one cycle forgets pending bytes the volume has yet to write.
+func TestDecayQuietVolumeSizesYieldsToAHeartbeatItRaced(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	vl.RecordAssign(1, 4000)
+	advanceSizeTrackingClock(vl, 1, 30*time.Second)
+
+	// The heartbeat wins the race and does the one decay this cycle owes.
+	vl.UpdateVolumeSize(1, 4000, 0, true)
+	vl.accessLock.RLock()
+	afterHeartbeat := vl.sizeTracking[1].effectiveSize
+	vl.accessLock.RUnlock()
+	if afterHeartbeat != 6000 {
+		t.Fatalf("the heartbeat left effectiveSize at %d, want 6000", afterHeartbeat)
+	}
+
+	vl.UpdateVolumeSize(1, 0, 0, false)
+
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+	if got := vl.sizeTracking[1].effectiveSize; got != afterHeartbeat {
+		t.Errorf("the decay halved again to %d, want the heartbeat's %d left alone", got, afterHeartbeat)
+	}
+}
+
+// A volume the disk really did fill can never recover through the decay, so it
+// must leave the candidate set instead of costing a write lock every pulse for
+// the rest of its life.
+func TestDecayQuietVolumeSizesSkipsAGenuinelyFullVolume(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":10000, "replication":"000"},
+          {"id":2, "size":4000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Volume 1 is full on disk; volume 2 only looks full to the estimate.
+	vl.SetVolumeCapacityFull(1)
+	vl.RecordAssign(2, 6000)
+	advanceSizeTrackingClock(vl, 1, 30*time.Second)
+	advanceSizeTrackingClock(vl, 2, 30*time.Second)
+
+	candidates := vl.quietDecayCandidates(10 * time.Second)
+	for _, vid := range candidates {
+		if vid == 1 {
+			t.Errorf("the decay keeps taking the write lock for volume 1, which the disk really did fill")
+		}
+	}
+	if len(candidates) != 1 || candidates[0] != 2 {
+		t.Errorf("candidates %v, want only the phantom-full volume 2", candidates)
+	}
+
+	vl.DecayQuietVolumeSizes(10 * time.Second)
+
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+	if got := vl.sizeTracking[2].effectiveSize; got != 7000 {
+		t.Errorf("the phantom-full volume was left at %d, want it decayed to 7000", got)
+	}
+}

weed/topology/volume_layout_drain_test.go

@@ -38,7 +38,7 @@ func TestGetPendingSize(t *testing.T) {
 	}
 
 	// UpdateVolumeSize (heartbeat) decays pending
-	vl.UpdateVolumeSize(1, 3000, 0)
+	vl.UpdateVolumeSize(1, 3000, 0, true)
 	// effective was 6000, reported 3000 → decay to 3000 + (6000-3000)/2 = 4500
 	// pending = 4500 - 3000 = 1500
 	if p := vl.GetPendingSize(1); p != 1500 {
@@ -72,7 +72,7 @@ func TestGetPendingSize_CompactionResets(t *testing.T) {
 	// Compaction happens — size drops from 5000 to 2000, revision changes.
 	// Without compaction awareness, decay would give: 2000 + (9000-2000)/2 = 5500.
 	// With compaction awareness, vid2size resets to 2000 (the real size).
-	vl.UpdateVolumeSize(1, 2000, 1) // revision 0 → 1
+	vl.UpdateVolumeSize(1, 2000, 1, true) // revision 0 → 1
 
 	if p := vl.GetPendingSize(1); p != 0 {
 		t.Errorf("expected 0 pending after compaction reset, got %d", p)
@@ -198,7 +198,7 @@ func TestDrainAndRemoveFromWritable_DecaysViaConcurrentHeartbeat(t *testing.T) {
 				st.lastUpdateTime = time.Time{} // reset to allow update
 			}
 			vl.accessLock.Unlock()
-			vl.UpdateVolumeSize(1, 1000+uint64(i+1)*1000, 0)
+			vl.UpdateVolumeSize(1, 1000+uint64(i+1)*1000, 0, true)
 		}
 	}()
 

weed/topology/volume_layout_oversized_heartbeat_test.go

@@ -53,7 +53,7 @@ func TestUpdateOversizedStateKeepsOversizedVolumeUnwritable(t *testing.T) {
 	oversized := vi
 	oversized.Size = 12000
 	vl.UpdateOversizedState(&oversized, dn)
-	vl.UpdateVolumeSize(1, oversized.Size, 0)
+	vl.UpdateVolumeSize(1, oversized.Size, 0, true)
 	vl.EnsureCorrectWritables(&oversized)
 
 	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
@@ -144,7 +144,7 @@ func TestEnsureCorrectWritablesHonorsRecoveryCooldown(t *testing.T) {
 	// capacityRecoveryDelay.
 	vi.Size = 4000
 	vl.UpdateOversizedState(&vi, dn)
-	vl.UpdateVolumeSize(1, vi.Size, 0)
+	vl.UpdateVolumeSize(1, vi.Size, 0, true)
 	vl.EnsureCorrectWritables(&vi)
 	if w, _ := vl.GetWritableVolumeCount(); w != 0 {
 		t.Fatalf("expected volume to stay unwritable during the cooldown, got %d writable", w)
@@ -152,7 +152,7 @@ func TestEnsureCorrectWritablesHonorsRecoveryCooldown(t *testing.T) {
 
 	// After the cooldown, a heartbeat with the shrunken size recovers it.
 	advanceSizeTrackingClock(vl, 1, capacityRecoveryDelay+time.Second)
-	if !vl.UpdateVolumeSize(1, vi.Size, 0) {
+	if !vl.UpdateVolumeSize(1, vi.Size, 0, true) {
 		t.Fatalf("expected volume to recover to writable after the cooldown")
 	}
 	vl.EnsureCorrectWritables(&vi)
@@ -202,7 +202,7 @@ func TestEnsureCorrectWritablesDoesNotRestoreVolumeStillAtLimit(t *testing.T) {
 	// past the limit and UpdateVolumeSize refuses recovery.
 	vi.Size = 8000
 	vl.UpdateOversizedState(&vi, dn)
-	vl.UpdateVolumeSize(1, vi.Size, 0)
+	vl.UpdateVolumeSize(1, vi.Size, 0, true)
 	if vl.vid2location[1].AnyOversized() {
 		t.Fatalf("expected oversized mark cleared after the shrink report")
 	}
@@ -249,7 +249,7 @@ func TestEnsureCorrectWritablesRestoresCrowdedVolumeAfterReplicaReturns(t *testi
 	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)
+	vl.UpdateVolumeSize(1, 9500, 0, true)
 	if _, crowded := vl.crowded[1]; !crowded {
 		t.Fatalf("expected the volume to be crowded")
 	}
@@ -271,7 +271,7 @@ func TestEnsureCorrectWritablesRestoresCrowdedVolumeAfterReplicaReturns(t *testi
 	topo.RegisterVolumeLayout(vi, dn)
 	advanceSizeTrackingClock(vl, 1, 5*time.Second)
 	vl.UpdateOversizedState(&vi, dn)
-	vl.UpdateVolumeSize(1, vi.Size, 0)
+	vl.UpdateVolumeSize(1, vi.Size, 0, true)
 	vl.EnsureCorrectWritables(&vi)
 	if w, _ := vl.GetWritableVolumeCount(); w != 1 {
 		t.Fatalf("expected the volume writable again, got %d", w)

weed/topology/volume_layout_pick_test.go

@@ -49,6 +49,9 @@ func setupWithLimit(t testing.TB, topologyLayout string, volumeSizeLimit uint64)
 						Size:    uint64(m["size"].(float64)),
 						Version: needle.GetCurrentVersion(),
 					}
+					if mVal, ok := m["fileCount"]; ok {
+						vi.FileCount = uint32(mVal.(float64))
+					}
 					if mVal, ok := m["collection"]; ok {
 						vi.Collection = mVal.(string)
 					}
@@ -176,6 +179,80 @@ func TestPickForWriteWithPendingSize(t *testing.T) {
 	}
 }
 
+// A flat 1MB charge per hintless file id overcharges a small-file workload by
+// orders of magnitude, marking volumes full while they hold a fraction of the
+// limit.
+func TestPickForWriteEstimatesPendingSizeFromVolumeAverage(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":409600, "fileCount":100, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo, vl := setupPickTest(t, layout, 30*1024*1024)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	option := &VolumeGrowOption{ReplicaPlacement: rp}
+
+	pendingAfterAssign := func(expectedDataSize uint64) uint64 {
+		vl.accessLock.RLock()
+		before := vl.sizeTracking[1].effectiveSize
+		vl.accessLock.RUnlock()
+		if _, _, _, _, err := topo.PickForWrite(1, option, vl, expectedDataSize); err != nil {
+			t.Fatalf("PickForWrite: %v", err)
+		}
+		vl.accessLock.RLock()
+		defer vl.accessLock.RUnlock()
+		return vl.sizeTracking[1].effectiveSize - before
+	}
+
+	if got := pendingAfterAssign(0); got != 4096 {
+		t.Errorf("a hintless assign charged %d, want the volume's 4096-byte average", got)
+	}
+	if got := pendingAfterAssign(8192); got != 8192 {
+		t.Errorf("a hinted assign charged %d, want the 8192-byte hint to win over the average", got)
+	}
+}
+
+// A volume with no files yet has no average to draw on, so the 1MB fallback
+// still applies.
+func TestPickForWriteEstimateFallsBackWithoutHistory(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":0, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo, vl := setupPickTest(t, layout, 30*1024*1024)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	option := &VolumeGrowOption{ReplicaPlacement: rp}
+
+	if _, _, _, _, err := topo.PickForWrite(1, option, vl, 0); err != nil {
+		t.Fatalf("PickForWrite: %v", err)
+	}
+	vl.accessLock.RLock()
+	pending := vl.sizeTracking[1].effectiveSize - vl.sizeTracking[1].reportedSize
+	vl.accessLock.RUnlock()
+	if pending != DefaultNeedleSizeEstimate {
+		t.Errorf("a hintless assign on an empty volume charged %d, want the %d fallback", pending, DefaultNeedleSizeEstimate)
+	}
+}
+
 func TestPickForWriteSingleWritable(t *testing.T) {
 	layout := `
 {
@@ -425,7 +502,7 @@ func TestUpdateVolumeSizeRecoversEagerlyRemovedVolume(t *testing.T) {
 	// *not* re-add the volume (even though effectiveSize would now be
 	// under the threshold).
 	advanceSizeTrackingClock(vl, 1, 3*time.Second) // past the 2s dedup window, but before 30s delay
-	if vl.UpdateVolumeSize(1, 4000, 0) {
+	if vl.UpdateVolumeSize(1, 4000, 0, true) {
 		t.Fatalf("recovery should not fire before capacityRecoveryDelay")
 	}
 	w, _ = vl.GetWritableVolumeCount()
@@ -437,7 +514,7 @@ func TestUpdateVolumeSizeRecoversEagerlyRemovedVolume(t *testing.T) {
 	// effectiveSize drops below the crowded threshold (9000).
 	for i := 0; i < 6; i++ {
 		advanceSizeTrackingClock(vl, 1, 10*time.Second)
-		recovered := vl.UpdateVolumeSize(1, 4000, 0)
+		recovered := vl.UpdateVolumeSize(1, 4000, 0, true)
 		if recovered {
 			vl.AdjustActiveVolumeCountAfterRecovery(1)
 			break
@@ -455,7 +532,7 @@ func TestUpdateVolumeSizeRecoversEagerlyRemovedVolume(t *testing.T) {
 	// fullSince should have been cleared so a subsequent heartbeat doesn't
 	// try to recover again.
 	advanceSizeTrackingClock(vl, 1, 60*time.Second)
-	if vl.UpdateVolumeSize(1, 4000, 0) {
+	if vl.UpdateVolumeSize(1, 4000, 0, true) {
 		t.Errorf("recovery should not re-fire after the volume is already writable")
 	}
 }
@@ -488,7 +565,7 @@ func TestUpdateVolumeSizeNoRecoveryWhenDiskStillOversized(t *testing.T) {
 	// Plenty of time elapsed — but reported stays at 10500 (over limit).
 	for i := 0; i < 5; i++ {
 		advanceSizeTrackingClock(vl, 1, 10*time.Second)
-		if vl.UpdateVolumeSize(1, 10500, 0) {
+		if vl.UpdateVolumeSize(1, 10500, 0, true) {
 			t.Fatalf("recovery must not fire when reported >= limit")
 		}
 	}
@@ -535,7 +612,7 @@ func TestHeartbeatDecaysPendingSize(t *testing.T) {
 	// Heartbeat: volume server reports size=3000 (some writes landed).
 	// Old effective=9000, new reported=3000 → excess=6000 → decayed to 3000.
 	// So vid2size should become 3000 + 6000/2 = 6000, not just 3000.
-	vl.UpdateVolumeSize(1, 3000, 0)
+	vl.UpdateVolumeSize(1, 3000, 0, true)
 
 	vl.accessLock.RLock()
 	if vl.sizeTracking[1].effectiveSize != 6000 {
@@ -546,7 +623,7 @@ func TestHeartbeatDecaysPendingSize(t *testing.T) {
 	// Second heartbeat: size=5000. Old effective=6000 → excess=1000 → decay to 500.
 	// vid2size should become 5000 + 1000/2 = 5500.
 	advanceCycle()
-	vl.UpdateVolumeSize(1, 5000, 0)
+	vl.UpdateVolumeSize(1, 5000, 0, true)
 
 	vl.accessLock.RLock()
 	if vl.sizeTracking[1].effectiveSize != 5500 {
@@ -557,7 +634,7 @@ func TestHeartbeatDecaysPendingSize(t *testing.T) {
 	// Third heartbeat: size=5500. Old effective=5500 → no excess.
 	// vid2size should be exactly 5500.
 	advanceCycle()
-	vl.UpdateVolumeSize(1, 5500, 0)
+	vl.UpdateVolumeSize(1, 5500, 0, true)
 
 	vl.accessLock.RLock()
 	if vl.sizeTracking[1].effectiveSize != 5500 {
@@ -621,8 +698,8 @@ func TestHeartbeatDecayDedupReplicas(t *testing.T) {
 
 	// Both replicas report size=3000. Decay should happen once: 3000 + (9000-3000)/2 = 6000.
 	// Calling UpdateVolumeSize twice simulates two replicas reporting in the same cycle.
-	vl.UpdateVolumeSize(1, 3000, 0)
-	vl.UpdateVolumeSize(1, 3000, 0) // second replica, same size — should be a no-op
+	vl.UpdateVolumeSize(1, 3000, 0, true)
+	vl.UpdateVolumeSize(1, 3000, 0, true) // second replica, same size — should be a no-op
 
 	vl.accessLock.RLock()
 	got := vl.sizeTracking[1].effectiveSize
@@ -659,7 +736,7 @@ func TestUpdateVolumeSize_DecaysEvenWhenReportedSizeUnchanged(t *testing.T) {
 
 	// First heartbeat: reported size unchanged at 1000 (writes haven't landed).
 	// Decay should still run: 1000 + (9000-1000)/2 = 5000.
-	vl.UpdateVolumeSize(1, 1000, 0)
+	vl.UpdateVolumeSize(1, 1000, 0, true)
 	if p := vl.GetPendingSize(1); p != 4000 {
 		t.Errorf("expected 4000 pending after first decay, got %d", p)
 	}
@@ -671,7 +748,7 @@ func TestUpdateVolumeSize_DecaysEvenWhenReportedSizeUnchanged(t *testing.T) {
 	vl.accessLock.Unlock()
 
 	// Second heartbeat: still 1000. Decay again: 1000 + (5000-1000)/2 = 3000.
-	vl.UpdateVolumeSize(1, 1000, 0)
+	vl.UpdateVolumeSize(1, 1000, 0, true)
 	if p := vl.GetPendingSize(1); p != 2000 {
 		t.Errorf("expected 2000 pending after second decay, got %d", p)
 	}