volume: validate volume correctness if last entry is a deletion

weed/storage/volume_checking.go

@@ -28,10 +28,14 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uin
 		return 0, nil
 	}
 	if size < 0 {
-		size = -size
-	}
-	if lastAppendAtNs, e = verifyNeedleIntegrity(v.DataBackend, v.Version(), offset.ToAcutalOffset(), key, size); e != nil {
-		return lastAppendAtNs, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
+		// read the deletion entry
+		if lastAppendAtNs, e = verifyDeletedNeedleIntegrity(v.DataBackend, v.Version(), key); e != nil {
+			return lastAppendAtNs, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
+		}
+	} else {
+		if lastAppendAtNs, e = verifyNeedleIntegrity(v.DataBackend, v.Version(), offset.ToAcutalOffset(), key, size); e != nil {
+			return lastAppendAtNs, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
+		}
 	}
 	return
 }
@@ -65,3 +69,20 @@ func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version,
 	}
 	return n.AppendAtNs, err
 }
+
+func verifyDeletedNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version, key NeedleId) (lastAppendAtNs uint64, err error) {
+	n := new(needle.Needle)
+	size := n.DiskSize(v)
+	var fileSize int64
+	fileSize, _, err = datFile.GetStat()
+	if err != nil {
+		return 0, fmt.Errorf("GetStat: %v", err)
+	}
+	if err = n.ReadData(datFile, fileSize-size, Size(0), v); err != nil {
+		return n.AppendAtNs, fmt.Errorf("read data [%d,%d) : %v", fileSize-size, size, err)
+	}
+	if n.Id != key {
+		return n.AppendAtNs, fmt.Errorf("index key %#x does not match needle's Id %#x", key, n.Id)
+	}
+	return n.AppendAtNs, err
+}