diff --git a/weed/topology/maintenance_mode_test.go b/weed/topology/maintenance_mode_test.go index 596ece558f..72fe3d66b8 100644 --- a/weed/topology/maintenance_mode_test.go +++ b/weed/topology/maintenance_mode_test.go @@ -231,7 +231,7 @@ func TestMaintenanceModeHoldsThroughVacuumAndMarkWritable(t *testing.T) { vl.DrainAndRemoveFromWritable(1) topo.SetDataNodeMaintenanceMode(dn1, true) for _, dn := range []*DataNode{dn1, dn2} { - if vl.SetVolumeAvailable(dn, 1, false, false, false) { + if vl.SetVolumeAvailable(dn, 1, 100, false, false, false) { t.Errorf("vacuum commit on %s made the volume writable with a replica in maintenance", dn.Id()) } } diff --git a/weed/topology/oversized_reregister_test.go b/weed/topology/oversized_reregister_test.go new file mode 100644 index 0000000000..b1942586de --- /dev/null +++ b/weed/topology/oversized_reregister_test.go @@ -0,0 +1,228 @@ +package topology + +import ( + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// The oversized verdict must outlive the location list: when every replica +// leaves, vid2location drops the entry and its per-replica marks with it, and +// a replica re-registering from a size-less report must not put the volume +// back on the writable list until a heartbeat re-measures it. +func TestOversizedVolumeStaysUnwritableAcrossLocationReset(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("010") + + dn1 := NewDataNode("server1") + dn1.Ip, dn1.Port = "10.0.0.1", 8080 + dn2 := NewDataNode("server2") + dn2.Ip, dn2.Port = "10.0.0.2", 8080 + + v := storage.VolumeInfo{Id: 35, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Size: 300} + dn1.UpdateVolumes([]storage.VolumeInfo{v}) + dn2.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn1) + topo.RegisterVolumeLayout(v, dn2) + + vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType) + vl.UpdateOversizedState(&v, dn1) + vl.UpdateOversizedState(&v, dn2) + vl.EnsureCorrectWritables(&v) + if isWritable(t, topo, v) { + t.Fatal("an oversized volume should not be writable") + } + + // Every replica leaves: the location list and its per-replica marks go. + vl.SetVolumeUnavailable(dn1, v.Id) + vl.SetVolumeUnavailable(dn2, v.Id) + + // Both replicas re-register from a report that carries no size. + small := v + small.Size = 0 + topo.RegisterVolumeLayout(small, dn1) + topo.RegisterVolumeLayout(small, dn2) + + if isWritable(t, topo, v) { + t.Fatal("an oversized volume was re-admitted to the writable list before its size was re-measured") + } + + // A report under the limit clears the remembered verdict. + shrunk := v + shrunk.Size = 10 + vl.UpdateOversizedState(&shrunk, dn1) + vl.UpdateOversizedState(&shrunk, dn2) + vl.EnsureCorrectWritables(&shrunk) + if !isWritable(t, topo, v) { + t.Fatal("a shrunk volume should be writable again") + } +} + +// The writable admission gate itself must refuse a known-oversized volume: +// paths like quarantine release and mark-writable do not consult the marks. +func TestOversizedVolumeRejectsExplicitWritableMark(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("010") + + dn1 := NewDataNode("server1") + dn1.Ip, dn1.Port = "10.0.0.1", 8080 + dn2 := NewDataNode("server2") + dn2.Ip, dn2.Port = "10.0.0.2", 8080 + + v := storage.VolumeInfo{Id: 35, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Size: 300} + dn1.UpdateVolumes([]storage.VolumeInfo{v}) + dn2.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn1) + topo.RegisterVolumeLayout(v, dn2) + + vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType) + vl.UpdateOversizedState(&v, dn1) + vl.UpdateOversizedState(&v, dn2) + vl.EnsureCorrectWritables(&v) + + if vl.SetVolumeWritable(dn1, v.Id) || isWritable(t, topo, v) { + t.Fatal("mark-writable re-admitted an oversized volume") + } + + // A vacuum commit asks the node whether the volume fits; its fresh + // answer supersedes that replica's remembered verdict. Both replicas + // commit at the new shrunken size while the cached reports on the nodes + // still carry the pre-compaction size. + shrunk := v + shrunk.Size = 10 + if vl.SetVolumeAvailable(dn1, v.Id, uint64(shrunk.Size), false, false, false) { + t.Fatal("the other replica's oversized verdict still stands") + } + if !vl.SetVolumeAvailable(dn2, v.Id, uint64(shrunk.Size), false, false, false) || !isWritable(t, topo, v) { + t.Fatal("a committed volume under the limit should be writable") + } +} + +// One replica shrinking clears only its own remembered verdict: a second +// replica last measured oversized keeps the volume out of the writable list +// even after its location entry is gone, until it too reports under the +// limit. +func TestOversizedVerdictClearsOnlyForReportingReplica(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("010") + + dn1 := NewDataNode("server1") + dn1.Ip, dn1.Port = "10.0.0.1", 8080 + dn2 := NewDataNode("server2") + dn2.Ip, dn2.Port = "10.0.0.2", 8080 + + v := storage.VolumeInfo{Id: 35, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Size: 300} + dn1.UpdateVolumes([]storage.VolumeInfo{v}) + dn2.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn1) + topo.RegisterVolumeLayout(v, dn2) + + vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType) + vl.UpdateOversizedState(&v, dn1) + vl.UpdateOversizedState(&v, dn2) + vl.EnsureCorrectWritables(&v) + + // dn2 leaves, taking its live mark with it; its verdict is remembered. + vl.SetVolumeUnavailable(dn2, v.Id) + + // dn1 shrinks under the limit: its verdict clears, dn2's stands. The + // compaction bump makes the size drop real rather than pending decay. + shrunk := v + shrunk.Size = 10 + shrunk.CompactRevision = v.CompactRevision + 1 + vl.UpdateOversizedState(&shrunk, dn1) + vl.UpdateVolumeSize(v.Id, uint64(shrunk.Size), shrunk.CompactRevision, true) + vl.EnsureCorrectWritables(&shrunk) + if !vl.isMarkedOversized(v.Id) { + t.Fatal("one replica's under-limit report cleared the absent replica's verdict") + } + + // dn2 returns from a report that carries no size; the verdict must + // still hold it out until it is re-measured. + small := v + small.Size = 0 + topo.RegisterVolumeLayout(small, dn2) + if isWritable(t, topo, v) { + t.Fatal("an oversized replica was re-admitted before its size was re-measured") + } + + // dn2's own under-limit report clears the last verdict. + vl.UpdateOversizedState(&shrunk, dn2) + vl.UpdateVolumeSize(v.Id, uint64(shrunk.Size), shrunk.CompactRevision, true) + vl.EnsureCorrectWritables(&shrunk) + if !isWritable(t, topo, v) { + t.Fatal("every replica under the limit should leave the volume writable") + } +} + +// The verdict TTL forgives a replica that never comes back; it must not +// forgive one that did. A remount announces the volume without a size, so an +// expired verdict still has to hold the volume out of the writable list +// until the remounted replica's first size-bearing report. +func TestExpiredOversizedVerdictHoldsForRemountedReplica(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("010") + + dn1 := NewDataNode("server1") + dn1.Ip, dn1.Port = "10.0.0.1", 8080 + dn2 := NewDataNode("server2") + dn2.Ip, dn2.Port = "10.0.0.2", 8080 + + v := storage.VolumeInfo{Id: 35, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Size: 300} + dn1.UpdateVolumes([]storage.VolumeInfo{v}) + dn2.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn1) + topo.RegisterVolumeLayout(v, dn2) + + vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType) + vl.UpdateOversizedState(&v, dn1) + vl.UpdateOversizedState(&v, dn2) + + // Every replica leaves, then the remembered verdicts age past the TTL. + vl.SetVolumeUnavailable(dn1, v.Id) + vl.SetVolumeUnavailable(dn2, v.Id) + vl.accessLock.Lock() + for nodeId := range vl.oversizedVolumes[v.Id] { + vl.oversizedVolumes[v.Id][nodeId] = time.Now().Add(-2 * oversizedVerdictTTL) + } + vl.accessLock.Unlock() + + // Both replicas remount on a report that carries no size. + sizeless := v + sizeless.Size = 0 + topo.RegisterVolumeLayout(sizeless, dn1) + topo.RegisterVolumeLayout(sizeless, dn2) + + if isWritable(t, topo, v) { + t.Fatal("a remounted oversized volume was re-admitted before its size was re-measured") + } + + // A replica that stayed gone loses its verdict to expiry: dn3 never + // returns, so its entry must be swept once the TTL passes. + dn3 := NewDataNode("server3") + dn3.Ip, dn3.Port = "10.0.0.3", 8080 + vl.accessLock.Lock() + vl.oversizedVolumes[v.Id][dn3.Id()] = time.Now().Add(-2 * oversizedVerdictTTL) + vl.accessLock.Unlock() + vl.isMarkedOversized(v.Id) + vl.accessLock.RLock() + _, kept := vl.oversizedVolumes[v.Id][dn3.Id()] + vl.accessLock.RUnlock() + if kept { + t.Fatal("an absent replica's expired verdict should have been dropped") + } + + // The next size-bearing report clears the stale verdict. + shrunk := v + shrunk.Size = 10 + vl.UpdateOversizedState(&shrunk, dn1) + vl.UpdateOversizedState(&shrunk, dn2) + vl.EnsureCorrectWritables(&shrunk) + if !isWritable(t, topo, v) { + t.Fatal("a shrunk volume should be writable again") + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 7165eba990..8811ba6043 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -869,6 +869,14 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) { }) t.quarantineLock.Unlock() if registered { + // RegisterVolume rebuilds only the read-only flags, so a + // location list recreated flagless would pass the writable + // check before the next heartbeat re-measures the volume. + // This report carries the size; mark from it first. A report + // that carries none leaves the remembered verdict in place. + if v.Size > 0 { + vl.UpdateOversizedState(&v, dn) + } vl.EnsureCorrectWritables(&v) t.flagDuplicateVolumeIdAsync(v, dn) return diff --git a/weed/topology/topology_test.go b/weed/topology/topology_test.go index e761ede3c5..5bf0202c25 100644 --- a/weed/topology/topology_test.go +++ b/weed/topology/topology_test.go @@ -851,7 +851,7 @@ func TestSetVolumeAvailableRepairsMissingVolume(t *testing.T) { } // The vacuum commit re-marks the volume available; it must re-register it. - vl.SetVolumeAvailable(dn, vid, false, false, false) + vl.SetVolumeAvailable(dn, vid, 100, false, false, false) if got := topo.Lookup("", vid); len(got) != 1 { t.Fatalf("after SetVolumeAvailable: lookup %d got %v, want 1 location", vid, got) } diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index ea6de330f4..83144cea1d 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -156,6 +156,7 @@ func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *V isCommitSuccess := true isReadOnly := false isFullCapacity := false + freshSizes := make(map[*DataNode]uint64) for _, dn := range vacuumLocationList.list { glog.V(0).Infoln("Start Committing vacuum", vid, "on", dn.Url()) err := operation.WithVolumeServerClient(false, dn.ServerAddress(), grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { @@ -165,10 +166,11 @@ func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *V VolumeId: uint32(vid), }) if resp != nil { + freshSizes[dn] = resp.VolumeSize if resp.IsReadOnly { isReadOnly = true } - if resp.VolumeSize > t.volumeSizeLimit { + if resp.VolumeSize >= t.volumeSizeLimit { isFullCapacity = true } } @@ -203,7 +205,7 @@ func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *V if resp.IsReadOnly { isReadOnly = true } - if resp.VolumeSize > t.volumeSizeLimit { + if resp.VolumeSize >= t.volumeSizeLimit { isFullCapacity = true } } @@ -227,7 +229,7 @@ func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *V for _, dn := range vacuumLocationList.list { t.quarantineLock.Lock() - vl.SetVolumeAvailable(dn, vid, isReadOnly, isFullCapacity, t.isQuarantined(dn, vid)) + vl.SetVolumeAvailable(dn, vid, freshSizes[dn], isReadOnly, isFullCapacity, t.isQuarantined(dn, vid)) t.quarantineLock.Unlock() } } diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index c74fefccbc..c05598620e 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -51,6 +51,12 @@ type volumeSizeTracking struct { // and out of writable within a single burst of assigns. const capacityRecoveryDelay = 30 * time.Second +// oversizedVerdictTTL bounds how long a remembered oversized verdict holds +// after the replica that earned it is gone. Long enough to outlive ordinary +// re-registration churn, short enough that a permanently lost replica cannot +// pin its volume out of the writable list forever. +const oversizedVerdictTTL = 24 * time.Hour + // mapping from volume to its locations, inverted from server to volume type VolumeLayout struct { growRequest atomic.Bool @@ -62,8 +68,18 @@ type VolumeLayout struct { writables []needle.VolumeId // transient array of writable volume id // writableMembers mirrors writables for membership tests, which run for // every volume on every heartbeat through ensureCorrectWritables. - writableMembers map[needle.VolumeId]struct{} - crowded map[needle.VolumeId]struct{} + writableMembers map[needle.VolumeId]struct{} + crowded map[needle.VolumeId]struct{} + // oversizedVolumes remembers, per replica, when a report last measured + // the volume over the size limit. The per-replica marks in the location + // list die with the list when its last location leaves, and a replica + // re-registering from a size-less report would make the volume look + // writable again until the next full heartbeat re-measures it. A + // replica's entry leaves only when that replica reports under the + // limit, or when the verdict ages out — a replica gone that long either + // lost the volume or returns to a master that re-measures it on the + // next heartbeat. + oversizedVolumes map[needle.VolumeId]map[NodeId]time.Time vacuumedVolumes map[needle.VolumeId]time.Time deletingVolumes map[needle.VolumeId]struct{} volumeSizeLimit uint64 @@ -95,6 +111,7 @@ func NewVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType writables: *new([]needle.VolumeId), writableMembers: make(map[needle.VolumeId]struct{}), crowded: make(map[needle.VolumeId]struct{}), + oversizedVolumes: make(map[needle.VolumeId]map[NodeId]time.Time), vacuumedVolumes: make(map[needle.VolumeId]time.Time), deletingVolumes: make(map[needle.VolumeId]struct{}), volumeSizeLimit: volumeSizeLimit, @@ -170,22 +187,76 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode, isQu } func (vl *VolumeLayout) rememberOversizedVolume(v *storage.VolumeInfo, dn *DataNode) { + oversized := vl.isOversized(v) + if oversized { + replicas, ok := vl.oversizedVolumes[v.Id] + if !ok { + replicas = make(map[NodeId]time.Time) + vl.oversizedVolumes[v.Id] = replicas + } + replicas[dn.Id()] = time.Now() + } else if replicas, ok := vl.oversizedVolumes[v.Id]; ok { + delete(replicas, dn.Id()) + if len(replicas) == 0 { + delete(vl.oversizedVolumes, v.Id) + } + } if location, ok := vl.vid2location[v.Id]; ok { - location.SetOversized(dn, vl.isOversized(v)) + location.SetOversized(dn, oversized) } } +// isMarkedOversized reports whether vid is known oversized: a remembered +// report from some replica says so, or a location in the live list is +// marked. A verdict past oversizedVerdictTTL forgives a replica that never +// came back, but still counts while its replica sits in the live list: a +// remount announces the volume without a size, so expiring then would +// re-admit an oversized volume until the next size-bearing heartbeat. +func (vl *VolumeLayout) isMarkedOversized(vid needle.VolumeId) bool { + if replicas, ok := vl.oversizedVolumes[vid]; ok { + now := time.Now() + location := vl.vid2location[vid] + marked := false + for nodeId, markedAt := range replicas { + present := false + if location != nil { + for _, dnl := range location.list { + if dnl.Id() == nodeId { + present = true + break + } + } + } + if now.Sub(markedAt) < oversizedVerdictTTL || present { + marked = true + continue + } + delete(replicas, nodeId) + } + if len(replicas) == 0 { + delete(vl.oversizedVolumes, vid) + } + if marked { + return true + } + } + return vl.vid2location[vid].AnyOversized() +} + // 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. +// reported by this heartbeat, and is the only writer of that mark and of this +// replica's remembered verdict. 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. +// Registration marks only through the size its report happens to carry: the +// incremental path registers from a short heartbeat message that holds none, +// and judging the volume by that absence would clear the mark on every +// arrival announcement and hand an oversized volume back to the writable +// list — the remembered verdict is what those size-less arrivals check +// against. func (vl *VolumeLayout) UpdateOversizedState(v *storage.VolumeInfo, dn *DataNode) { vl.accessLock.Lock() defer vl.accessLock.Unlock() @@ -289,7 +360,7 @@ func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint6 if reportedSize >= vl.volumeSizeLimit { return false // actual on-disk size still over limit; stay out } - if vl.vid2location[vid].AnyOversized() { + if vl.isMarkedOversized(vid) { return false } if !vl.enoughCopies(vid) || !vl.isAllWritable(vid) { @@ -375,7 +446,7 @@ func (vl *VolumeLayout) EnsureCorrectWritables(v *storage.VolumeInfo) { func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) { isEnoughCopies := vl.enoughCopies(vid) isAllWritable := vl.isAllWritable(vid) - isOversizedVolume := vl.vid2location[vid].AnyOversized() + isOversizedVolume := vl.isMarkedOversized(vid) if isEnoughCopies && isAllWritable && !isOversizedVolume { // A volume removed for capacity (fullSince set) must go through the // heartbeat recovery path in UpdateVolumeSize, which enforces @@ -583,7 +654,7 @@ func (vl *VolumeLayout) UnmarkDeleting(vid needle.VolumeId) { // Re-run the standard writable gate so a surviving volume regains // assignment immediately; digest heartbeats only re-report changed // volumes, so waiting on them could strand it unwritable indefinitely. - if !vl.vid2location[vid].AnyOversized() && vl.enoughCopies(vid) && vl.isAllWritable(vid) { + if !vl.isMarkedOversized(vid) && vl.enoughCopies(vid) && vl.isAllWritable(vid) { vl.setVolumeWritable(vid) } vl.accessLock.Unlock() @@ -1057,6 +1128,13 @@ func (vl *VolumeLayout) setVolumeWritable(vid needle.VolumeId) bool { if _, ok := vl.deletingVolumes[vid]; ok { return false } + // Every admission to the writable list funnels here, including callers + // that do not consult the oversized marks themselves (quarantine + // release, mark-writable, vacuum commit). Holding the gate here keeps a + // known-oversized volume out until a report measures it under the limit. + if vl.isMarkedOversized(vid) { + return false + } if _, ok := vl.writableMembers[vid]; ok { return false } @@ -1150,7 +1228,7 @@ func (vl *VolumeLayout) SetVolumeUnavailable(dn *DataNode, vid needle.VolumeId) } return false } -func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, isReadOnly, isFullCapacity, quarantined bool) (becameWritable bool) { +func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, size uint64, isReadOnly, isFullCapacity, quarantined bool) (becameWritable bool) { restoreActiveVolumeCount := false vl.accessLock.Lock() defer func() { @@ -1174,6 +1252,12 @@ func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, is location := vl.getOrCreateLocationList(vid) moveLookupOwnership(vid, location.Set(dn), dn) + // The caller just asked the node whether the volume fits; record that + // fresh measurement on this replica's mark and remembered verdict rather + // than trusting the cached report, which predates the compaction. + vInfo.Size = size + vl.rememberOversizedVolume(&vInfo, dn) + if quarantined { location.SetReadOnly(dn, true) return false @@ -1181,7 +1265,7 @@ func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, is if vInfo.ReadOnly || isReadOnly || isFullCapacity { return false } - vl.initSizeTracking(vid, vInfo.Size, vInfo.CompactRevision) + vl.initSizeTracking(vid, size, vInfo.CompactRevision) if vl.enoughCopies(vid) && vl.isAllWritable(vid) { becameWritable = vl.setVolumeWritable(vid) diff --git a/weed/topology/volume_layout_capacity_recovery_test.go b/weed/topology/volume_layout_capacity_recovery_test.go index 686c67e4ba..aca60f8ae1 100644 --- a/weed/topology/volume_layout_capacity_recovery_test.go +++ b/weed/topology/volume_layout_capacity_recovery_test.go @@ -98,7 +98,7 @@ func TestSetVolumeAvailableRestoresActiveCountForCapacityFullVolume(t *testing.T dn := vl.vid2location[1].list[0] vl.accessLock.RUnlock() - if !vl.SetVolumeAvailable(dn, 1, false, false, false) { + if !vl.SetVolumeAvailable(dn, 1, 4000, false, false, false) { t.Fatalf("SetVolumeAvailable should report the volume became writable") } if got := topo.diskUsages.usages[types.HardDriveType].activeVolumeCount; got != initialActive {