fix: recreate index include deleted files (#5579)

* fix: recreate index include deleted files
https://github.com/seaweedfs/seaweedfs/issues/5508

* fix: counting the number of files

* fix: log

3 files changed, 18 insertions, 19 deletions

diff --git a/weed/storage/needle_map_memory.go b/weed/storage/needle_map_memory.go
index a2beb6c33..c75514a31 100644
--- a/weed/storage/needle_map_memory.go
+++ b/weed/storage/needle_map_memory.go
@@ -36,8 +36,8 @@ func LoadCompactNeedleMap(file *os.File) (*NeedleMap, error) {
 func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) {
 	e := idx.WalkIndexFile(file, 0, func(key NeedleId, offset Offset, size Size) error {
 		nm.MaybeSetMaxFileKey(key)
-		nm.FileCounter++
 		if !offset.IsZero() && size.IsValid() {
+			nm.FileCounter++
 			nm.FileByteCounter = nm.FileByteCounter + uint64(size)
 			oldOffset, oldSize := nm.m.Set(NeedleId(key), offset, size)
 			if !oldOffset.IsZero() && oldSize.IsValid() {
@@ -51,7 +51,7 @@ func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) {
 		}
 		return nil
 	})
-	glog.V(1).Infof("max file key: %d for file: %s", nm.MaxFileKey(), file.Name())
+	glog.V(1).Infof("max file key: %v count: %d deleted: %d for file: %s", nm.MaxFileKey(), nm.FileCount(), nm.DeletedCount(), file.Name())
 	return nm, e
 }
 
diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go
index 0c5f154e8..f5ceffcce 100644
--- a/weed/storage/volume_checking.go
+++ b/weed/storage/volume_checking.go
@@ -109,9 +109,6 @@ func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version,
 			return 0, fmt.Errorf("verifyNeedleIntegrity check %s entry offset %d size %d: %v", datFile.Name(), offset, size, err)
 		}
 		n.AppendAtNs = util.BytesToUint64(bytes)
-		if n.HasTtl() {
-			return n.AppendAtNs, nil
-		}
 		fileTailOffset := offset + needle.GetActualSize(size, v)
 		fileSize, _, err := datFile.GetStat()
 		if err != nil {
@@ -130,7 +127,7 @@ func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version,
 		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)
+		return n.AppendAtNs, fmt.Errorf("index key %v does not match needle's Id %v", key, n.Id)
 	}
 	return n.AppendAtNs, err
 }
@@ -147,7 +144,7 @@ func verifyDeletedNeedleIntegrity(datFile backend.BackendStorageFile, v needle.V
 		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, fmt.Errorf("index key %v does not match needle's Id %v", key, n.Id)
 	}
 	return n.AppendAtNs, err
 }
diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go
index c8098493d..6bbbde71d 100644
--- a/weed/storage/volume_vacuum.go
+++ b/weed/storage/volume_vacuum.go
@@ -487,19 +487,21 @@ func (v *Volume) copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, da
 	if err != nil {
 		return err
 	}
-	dstDatSize, _, err := dstDatBackend.GetStat()
-	if err != nil {
-		return err
-	}
-	if v.nm.ContentSize() > v.nm.DeletedSize() {
-		expectedContentSize := v.nm.ContentSize() - v.nm.DeletedSize()
-		if expectedContentSize > uint64(dstDatSize) {
-			return fmt.Errorf("volume %s unexpected new data size: %d does not match size of content minus deleted: %d",
-				v.Id.String(), dstDatSize, expectedContentSize)
+	if v.Ttl.String() == "" {
+		dstDatSize, _, err := dstDatBackend.GetStat()
+		if err != nil {
+			return err
+		}
+		if v.nm.ContentSize() > v.nm.DeletedSize() {
+			expectedContentSize := v.nm.ContentSize() - v.nm.DeletedSize()
+			if expectedContentSize > uint64(dstDatSize) {
+				return fmt.Errorf("volume %s unexpected new data size: %d does not match size of content minus deleted: %d",
+					v.Id.String(), dstDatSize, expectedContentSize)
+			}
+		} else {
+			glog.Warningf("volume %s content size: %d less deleted size: %d, new size: %d",
+				v.Id.String(), v.nm.ContentSize(), v.nm.DeletedSize(), dstDatSize)
 		}
-	} else {
-		glog.Warningf("volume %s content size: %d less deleted size: %d, new size: %d",
-			v.Id.String(), v.nm.ContentSize(), v.nm.DeletedSize(), dstDatSize)
 	}
 	err = newNm.SaveToIdx(datIdxName)
 	if err != nil {