[voluche.chek.disk] needles older than the cutoff time are not missing yet (#3922)

needles older than the cutoff time are not missing yet

https://github.com/seaweedfs/seaweedfs/issues/3919

weed/shell/command_volume_check_disk.go

@@ -13,6 +13,7 @@ import (
 	"google.golang.org/grpc"
 	"io"
 	"math"
+	"time"
 )
 
 func init() {
@@ -121,6 +122,7 @@ func (c *commandVolumeCheckDisk) checkBoth(a *VolumeReplica, b *VolumeReplica, a
 	}()
 
 	// read index db
+	readIndexDbCutoffFrom := uint64(time.Now().UnixNano())
 	if err = c.readIndexDatabase(aDB, a.info.Collection, a.info.Id, pb.NewServerAddressFromDataNode(a.location.dataNode), verbose, writer); err != nil {
 		return true, true, fmt.Errorf("readIndexDatabase %s volume %d: %v", a.location.dataNode, a.info.Id, err)
 	}
@@ -129,25 +131,37 @@ func (c *commandVolumeCheckDisk) checkBoth(a *VolumeReplica, b *VolumeReplica, a
 	}
 
 	// find and make up the differences
-	if aHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, verbose, writer, applyChanges, nonRepairThreshold); err != nil {
+	if aHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, verbose, writer, applyChanges, nonRepairThreshold, readIndexDbCutoffFrom); err != nil {
 		return true, true, fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", b.location.dataNode.Id, a.location.dataNode.Id, b.info.Id, err)
 	}
-	if bHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, verbose, writer, applyChanges, nonRepairThreshold); err != nil {
+	if bHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, verbose, writer, applyChanges, nonRepairThreshold, readIndexDbCutoffFrom); err != nil {
 		return true, true, fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", a.location.dataNode.Id, b.location.dataNode.Id, a.info.Id, err)
 	}
 	return
 }
 
-func (c *commandVolumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64) (hasChanges bool, err error) {
+func (c *commandVolumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64, cutoffFromAtNs uint64) (hasChanges bool, err error) {
 
 	// find missing keys
 	// hash join, can be more efficient
 	var missingNeedles []needle_map.NeedleValue
 	var counter int
-	minuend.AscendingVisit(func(value needle_map.NeedleValue) error {
+	doCutoffOfLastNeedle := true
+	minuend.DescendingVisit(func(value needle_map.NeedleValue) error {
 		counter++
-		if _, found := subtrahend.Get(value.Key); !found && value.Size.IsValid() {
+		if _, found := subtrahend.Get(value.Key); !found {
+			if doCutoffOfLastNeedle {
+				if needleMeta, err := readNeedleMeta(c.env.option.GrpcDialOption, pb.NewServerAddressFromDataNode(source.location.dataNode), source.info.Id, value); err == nil {
+					// needles older than the cutoff time are not missing yet
+					if needleMeta.AppendAtNs > cutoffFromAtNs {
+						return nil
+					}
+					doCutoffOfLastNeedle = false
+				}
+			}
 			missingNeedles = append(missingNeedles, value)
+		} else if doCutoffOfLastNeedle {
+			doCutoffOfLastNeedle = false
 		}
 		return nil
 	})
@@ -166,7 +180,6 @@ func (c *commandVolumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_m
 	}
 
 	for _, needleValue := range missingNeedles {
-
 		needleBlob, err := readSourceNeedleBlob(c.env.option.GrpcDialOption, pb.NewServerAddressFromDataNode(source.location.dataNode), source.info.Id, needleValue)
 		if err != nil {
 			return hasChanges, err

weed/shell/commands.go

@@ -1,7 +1,11 @@
 package shell
 
 import (
+	"context"
 	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle_map"
 	"io"
 	"net/url"
 	"strconv"
@@ -148,3 +152,20 @@ func findInputDirectory(args []string) (input string) {
 	}
 	return input
 }
+
+func readNeedleMeta(grpcDialOption grpc.DialOption, volumeServer pb.ServerAddress, volumeId uint32, needleValue needle_map.NeedleValue) (resp *volume_server_pb.ReadNeedleMetaResponse, err error) {
+	err = operation.WithVolumeServerClient(false, volumeServer, grpcDialOption,
+		func(client volume_server_pb.VolumeServerClient) error {
+			if resp, err = client.ReadNeedleMeta(context.Background(), &volume_server_pb.ReadNeedleMetaRequest{
+				VolumeId: volumeId,
+				NeedleId: uint64(needleValue.Key),
+				Offset:   needleValue.Offset.ToActualOffset(),
+				Size:     int32(needleValue.Size),
+			}); err != nil {
+				return err
+			}
+			return nil
+		},
+	)
+	return
+}

weed/storage/needle_map/memdb.go

@@ -6,6 +6,7 @@ import (
 	"os"
 
 	"github.com/syndtr/goleveldb/leveldb"
+	"github.com/syndtr/goleveldb/leveldb/iterator"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 	"github.com/syndtr/goleveldb/leveldb/storage"
 
@@ -61,17 +62,47 @@ func (cm *MemDb) Get(key NeedleId) (*NeedleValue, bool) {
 }
 
 // Visit visits all entries or stop if any error when visiting
+func doVisit(iter iterator.Iterator, visit func(NeedleValue) error) (ret error) {
+	key := BytesToNeedleId(iter.Key())
+	data := iter.Value()
+	offset := BytesToOffset(data[0:OffsetSize])
+	size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize])
+
+	needle := NeedleValue{Key: key, Offset: offset, Size: size}
+	ret = visit(needle)
+	if ret != nil {
+		return
+	}
+	return nil
+}
+
 func (cm *MemDb) AscendingVisit(visit func(NeedleValue) error) (ret error) {
 	iter := cm.db.NewIterator(nil, nil)
+	if iter.First() {
+		if ret = doVisit(iter, visit); ret != nil {
+			return
+		}
+	}
 	for iter.Next() {
-		key := BytesToNeedleId(iter.Key())
-		data := iter.Value()
-		offset := BytesToOffset(data[0:OffsetSize])
-		size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize])
+		if ret = doVisit(iter, visit); ret != nil {
+			return
+		}
+	}
+	iter.Release()
+	ret = iter.Error()
 
-		needle := NeedleValue{Key: key, Offset: offset, Size: size}
-		ret = visit(needle)
-		if ret != nil {
+	return
+}
+
+func (cm *MemDb) DescendingVisit(visit func(NeedleValue) error) (ret error) {
+	iter := cm.db.NewIterator(nil, nil)
+	if iter.Last() {
+		if ret = doVisit(iter, visit); ret != nil {
+			return
+		}
+	}
+	for iter.Prev() {
+		if ret = doVisit(iter, visit); ret != nil {
 			return
 		}
 	}