check existing volumes for writable status

weed/topology/volume_layout.go

@@ -139,7 +139,7 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
-	defer vl.ensureCorrectWritables(v)
+	defer vl.ensureCorrectWritables(v.Id)
 	defer vl.rememberOversizedVolume(v, dn)
 
 	if _, ok := vl.vid2location[v.Id]; !ok {
@@ -189,7 +189,7 @@ func (vl *VolumeLayout) UnRegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 
 		vl.readonlyVolumes.Remove(v.Id, dn)
 		vl.oversizedVolumes.Remove(v.Id, dn)
-		vl.ensureCorrectWritables(v)
+		vl.ensureCorrectWritables(v.Id)
 
 		if location.Length() == 0 {
 			delete(vl.vid2location, v.Id)
@@ -202,19 +202,30 @@ func (vl *VolumeLayout) EnsureCorrectWritables(v *storage.VolumeInfo) {
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
-	vl.ensureCorrectWritables(v)
+	vl.ensureCorrectWritables(v.Id)
 }
 
-func (vl *VolumeLayout) ensureCorrectWritables(v *storage.VolumeInfo) {
-	if vl.enoughCopies(v.Id) && vl.isWritable(v) {
-		if !vl.oversizedVolumes.IsTrue(v.Id) {
-			vl.setVolumeWritable(v.Id)
+func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) {
+	if vl.enoughCopies(vid) && vl.isAllWritable(vid) {
+		if !vl.oversizedVolumes.IsTrue(vid) {
+			vl.setVolumeWritable(vid)
 		}
 	} else {
-		vl.removeFromWritable(v.Id)
+		vl.removeFromWritable(vid)
 	}
 }
 
+func (vl *VolumeLayout) isAllWritable(vid needle.VolumeId) bool {
+	for _, dn := range vl.vid2location[vid].list {
+		if v, found := dn.volumes[vid]; found {
+			if v.ReadOnly {
+				return false
+			}
+		}
+	}
+	return true
+}
+
 func (vl *VolumeLayout) isOversized(v *storage.VolumeInfo) bool {
 	return uint64(v.Size) >= vl.volumeSizeLimit
 }