diff --git a/weed/topology/topology.go b/weed/topology/topology.go index ebaaf5fda..e8b12b357 100644 --- a/weed/topology/topology.go +++ b/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) diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index 8772ecb89..b0894313e 100644 --- a/weed/topology/topology_event_handling.go +++ b/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) } diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 5a96207ee..93d226a7b 100644 --- a/weed/topology/volume_layout.go +++ b/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() diff --git a/weed/topology/volume_layout_capacity_recovery_test.go b/weed/topology/volume_layout_capacity_recovery_test.go index cd617e17e..8edb9eead 100644 --- a/weed/topology/volume_layout_capacity_recovery_test.go +++ b/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) + } +} diff --git a/weed/topology/volume_layout_drain_test.go b/weed/topology/volume_layout_drain_test.go index 6132b7a2c..00490b234 100644 --- a/weed/topology/volume_layout_drain_test.go +++ b/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) } }() diff --git a/weed/topology/volume_layout_oversized_heartbeat_test.go b/weed/topology/volume_layout_oversized_heartbeat_test.go index cae8549cf..efa76b89a 100644 --- a/weed/topology/volume_layout_oversized_heartbeat_test.go +++ b/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) diff --git a/weed/topology/volume_layout_pick_test.go b/weed/topology/volume_layout_pick_test.go index ce13c6c22..78c741fb1 100644 --- a/weed/topology/volume_layout_pick_test.go +++ b/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) }