volume: compaction can cause readonly volumes

address https://github.com/chrislusf/seaweedfs/issues/1233

4 files changed, 9 insertions, 5 deletions

diff --git a/weed/storage/store.go b/weed/storage/store.go
index 19dbcb70e..f0dbbdf18 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -145,7 +145,7 @@ func (s *Store) VolumeInfos() []*VolumeInfo {
 				FileCount:        int(v.FileCount()),
 				DeleteCount:      int(v.DeletedCount()),
 				DeletedByteCount: v.DeletedSize(),
-				ReadOnly:         v.noWriteOrDelete || v.noWriteCanDelete,
+				ReadOnly:         v.IsReadOnly(),
 				Ttl:              v.Ttl,
 				CompactRevision:  uint32(v.CompactionRevision),
 			}
@@ -229,7 +229,7 @@ func (s *Store) Close() {
 
 func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (isUnchanged bool, err error) {
 	if v := s.findVolume(i); v != nil {
-		if v.noWriteOrDelete || v.noWriteCanDelete {
+		if v.v.IsReadOnly() {
 			err = fmt.Errorf("volume %d is read only", i)
 			return
 		}
diff --git a/weed/storage/volume.go b/weed/storage/volume.go
index 7da83de7a..755b98b79 100644
--- a/weed/storage/volume.go
+++ b/weed/storage/volume.go
@@ -215,7 +215,7 @@ func (v *Volume) ToVolumeInformationMessage() *master_pb.VolumeInformationMessag
 		FileCount:        v.FileCount(),
 		DeleteCount:      v.DeletedCount(),
 		DeletedByteCount: v.DeletedSize(),
-		ReadOnly:         v.noWriteOrDelete || v.noWriteCanDelete,
+		ReadOnly:         v.IsReadOnly(),
 		ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
 		Version:          uint32(v.Version()),
 		Ttl:              v.Ttl.ToUint32(),
@@ -237,3 +237,7 @@ func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
 	}
 	return v.volumeInfo.GetFiles()[0].BackendName(), v.volumeInfo.GetFiles()[0].GetKey()
 }
+
+func (v *Volume) IsReadOnly() bool {
+	return v.noWriteOrDelete || v.noWriteCanDelete
+}
diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go
index a65c2a3ff..c33f0049a 100644
--- a/weed/storage/volume_checking.go
+++ b/weed/storage/volume_checking.go
@@ -58,7 +58,7 @@ func readIndexEntryAtOffset(indexFile *os.File, offset int64) (bytes []byte, err
 func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version, offset int64, key NeedleId, size uint32) (lastAppendAtNs uint64, err error) {
 	n := new(needle.Needle)
 	if err = n.ReadData(datFile, offset, size, v); err != nil {
-		return n.AppendAtNs, err
+		return n.AppendAtNs, fmt.Errorf("read data [%d,%d) : %v", offset, offset+int64(size), err)
 	}
 	if n.Id != key {
 		return n.AppendAtNs, fmt.Errorf("index key %#x does not match needle's Id %#x", key, n.Id)
diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go
index 6b42fc452..3b0897bca 100644
--- a/weed/storage/volume_loading.go
+++ b/weed/storage/volume_loading.go
@@ -94,7 +94,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 			glog.V(0).Infof("volumeDataIntegrityChecking failed %v", err)
 		}
 
-		if v.noWriteOrDelete || v.noWriteCanDelete {
+		if v.IsReadOnly() {
 			if v.nm, err = NewSortedFileNeedleMap(fileName, indexFile); err != nil {
 				glog.V(0).Infof("loading sorted db %s error: %v", fileName+".sdx", err)
 			}