ADHOC: Volume fsck use a time cutoff param (#3626)

* ADHOC: cut off volumn fsck

* more

* fix typo

* add test

* modify name

* fix comment

* fix comments

* nit

* fix typo

* Update weed/shell/command_volume_fsck.go

Co-authored-by: root <root@HQ-10MSTD3EY.roblox.local>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

weed/storage/idx/binary_search.go

@@ -0,0 +1,29 @@
+package idx
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// firstInvalidIndex find the first index the failed lessThanOrEqualToFn function's requirement.
+func FirstInvalidIndex(bytes []byte, lessThanOrEqualToFn func(key types.NeedleId, offset types.Offset, size types.Size) (bool, error)) (int, error) {
+	left, right := 0, len(bytes)/types.NeedleMapEntrySize-1
+	index := right + 1
+	for left <= right {
+		mid := left + (right-left)>>1
+		loc := mid * types.NeedleMapEntrySize
+		key := types.BytesToNeedleId(bytes[loc : loc+types.NeedleIdSize])
+		offset := types.BytesToOffset(bytes[loc+types.NeedleIdSize : loc+types.NeedleIdSize+types.OffsetSize])
+		size := types.BytesToSize(bytes[loc+types.NeedleIdSize+types.OffsetSize : loc+types.NeedleIdSize+types.OffsetSize+types.SizeSize])
+		res, err := lessThanOrEqualToFn(key, offset, size)
+		if err != nil {
+			return -1, err
+		}
+		if res {
+			left = mid + 1
+		} else {
+			index = mid
+			right = mid - 1
+		}
+	}
+	return index, nil
+}

weed/storage/idx_binary_search_test.go

@@ -0,0 +1,57 @@
+package storage
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/storage/idx"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/stretchr/testify/assert"
+	"os"
+	"testing"
+)
+
+func TestFirstInvalidIndex(t *testing.T) {
+	dir := t.TempDir()
+
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	type WriteInfo struct {
+		offset int64
+		size   int32
+	}
+	// initialize 20 needles then update first 10 needles
+	for i := 1; i <= 30; i++ {
+		n := newRandomNeedle(uint64(i))
+		n.Flags = 0x08
+		_, _, _, err := v.writeNeedle2(n, true, false)
+		if err != nil {
+			t.Fatalf("write needle %d: %v", i, err)
+		}
+	}
+	b, err := os.ReadFile(v.IndexFileName() + ".idx")
+	// base case every record is valid -> nothing is filtered
+	index, err := idx.FirstInvalidIndex(b, func(key types.NeedleId, offset types.Offset, size types.Size) (bool, error) {
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("failed to complete binary search %v", err)
+	}
+	assert.Equal(t, 30, index, "when every record is valid nothing should be filtered from binary search")
+	index, err = idx.FirstInvalidIndex(b, func(key types.NeedleId, offset types.Offset, size types.Size) (bool, error) {
+		return false, nil
+	})
+	assert.Equal(t, 0, index, "when every record is invalid everything should be filtered from binary search")
+	index, err = idx.FirstInvalidIndex(b, func(key types.NeedleId, offset types.Offset, size types.Size) (bool, error) {
+		return key < 20, nil
+	})
+	// needle key range from 1 to 30 so < 20 means 19 keys are valid and cutoff the bytes at 19 * 16 = 304
+	assert.Equal(t, 19, index, "when every record is invalid everything should be filtered from binary search")
+
+	index, err = idx.FirstInvalidIndex(b, func(key types.NeedleId, offset types.Offset, size types.Size) (bool, error) {
+		return key <= 1, nil
+	})
+	// needle key range from 1 to 30 so <=1 1 means 1 key is valid and cutoff the bytes at 1 * 16 = 16
+	assert.Equal(t, 1, index, "when every record is invalid everything should be filtered from binary search")
+}