Merge branch 'master' into mq-subscribe

3 files changed, 14 insertions, 4 deletions

diff --git a/weed/storage/needle/needle_read.go b/weed/storage/needle/needle_read.go
index 19b801bd4..1041e924d 100644
--- a/weed/storage/needle/needle_read.go
+++ b/weed/storage/needle/needle_read.go
@@ -93,7 +93,17 @@ func (n *Needle) ReadData(r backend.BackendStorageFile, offset int64, size Size,
 	if err != nil {
 		return err
 	}
-	return n.ReadBytes(bytes, offset, size, version)
+
+	err = n.ReadBytes(bytes, offset, size, version)
+	if err == ErrorSizeMismatch && OffsetSize == 4 {
+		offset = offset + int64(MaxPossibleVolumeSize)
+		bytes, err = ReadNeedleBlob(r, offset, size, version)
+		if err != nil {
+			return err
+		}
+		err = n.ReadBytes(bytes, offset, size, version)
+	}
+	return err
 }
 
 func (n *Needle) ParseNeedleHeader(bytes []byte) {
diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go
index e10fedf9b..0f92a2e93 100644
--- a/weed/storage/volume_checking.go
+++ b/weed/storage/volume_checking.go
@@ -60,6 +60,9 @@ func doCheckAndFixVolumeData(v *Volume, indexFile *os.File, indexOffset int64) (
 		}
 	} else {
 		if lastAppendAtNs, err = verifyNeedleIntegrity(v.DataBackend, v.Version(), offset.ToActualOffset(), key, size); err != nil {
+			if err == ErrorSizeMismatch {
+				return verifyNeedleIntegrity(v.DataBackend, v.Version(), offset.ToActualOffset()+int64(MaxPossibleVolumeSize), key, size)
+			}
 			return lastAppendAtNs, err
 		}
 	}
diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go
index 8e9717689..09fbdd701 100644
--- a/weed/storage/volume_read.go
+++ b/weed/storage/volume_read.go
@@ -55,9 +55,6 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize
 	}
 	if readOption == nil || !readOption.IsMetaOnly {
 		err = n.ReadData(v.DataBackend, nv.Offset.ToActualOffset(), readSize, v.Version())
-		if err == needle.ErrorSizeMismatch && OffsetSize == 4 {
-			err = n.ReadData(v.DataBackend, nv.Offset.ToActualOffset()+int64(MaxPossibleVolumeSize), readSize, v.Version())
-		}
 		v.checkReadWriteError(err)
 		if err != nil {
 			return 0, err