[volume.chek.disk] sync of deletions the fix (#3923)

* sync of deletions the fix

* avoid return if only partiallyDeletedNeedles

* refactor sync deletions

2 files changed, 9 insertions, 1 deletions

diff --git a/weed/storage/needle_map/memdb.go b/weed/storage/needle_map/memdb.go
index f2b161792..d3d47b605 100644
--- a/weed/storage/needle_map/memdb.go
+++ b/weed/storage/needle_map/memdb.go
@@ -145,8 +145,12 @@ func (cm *MemDb) LoadFromIdx(idxName string) (ret error) {
 
 func (cm *MemDb) LoadFromReaderAt(readerAt io.ReaderAt) (ret error) {
 
+	return cm.LoadFilterFromReaderAt(readerAt, true, true)
+}
+
+func (cm *MemDb) LoadFilterFromReaderAt(readerAt io.ReaderAt, isFilterOffsetZero bool, isFilterDeleted bool) (ret error) {
 	return idx.WalkIndexFile(readerAt, 0, func(key NeedleId, offset Offset, size Size) error {
-		if offset.IsZero() || size.IsDeleted() {
+		if (isFilterOffsetZero && offset.IsZero()) || (isFilterDeleted && size.IsDeleted()) {
 			return cm.Delete(key)
 		}
 		return cm.Set(key, offset, size)
diff --git a/weed/storage/types/needle_id_type.go b/weed/storage/types/needle_id_type.go
index e65c862f2..1ea9b0a1a 100644
--- a/weed/storage/types/needle_id_type.go
+++ b/weed/storage/types/needle_id_type.go
@@ -34,6 +34,10 @@ func (k NeedleId) String() string {
 	return strconv.FormatUint(uint64(k), 16)
 }
 
+func (k NeedleId) FileId(volumeId uint32) string {
+	return fmt.Sprintf("%d,%s00000000", volumeId, k.String())
+}
+
 func ParseNeedleId(idString string) (NeedleId, error) {
 	key, err := strconv.ParseUint(idString, 16, 64)
 	if err != nil {