ADHOC: volume fsck using append at ns (#3906)

* ADHOC: volume fsck using append at ns

* nit

* nit

Co-authored-by: root <root@HQ-10MSTD3EY.roblox.local>

weed/pb/filer_pb/filer.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
-// 	protoc        v3.17.3
+// 	protoc-gen-go v1.28.1
+// 	protoc        v3.21.4
 // source: filer.proto
 
 package filer_pb
@@ -3239,9 +3239,9 @@ func (x *LocateBrokerResponse) GetResources() []*LocateBrokerResponse_Resource {
 	return nil
 }
 
-/////////////////////////
+// ///////////////////////
 // Key-Value operations
-/////////////////////////
+// ///////////////////////
 type KvGetRequest struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -3446,9 +3446,9 @@ func (x *KvPutResponse) GetError() string {
 	return ""
 }
 
-/////////////////////////
+// ///////////////////////
 // path-based configurations
-/////////////////////////
+// ///////////////////////
 type FilerConf struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -3504,9 +3504,9 @@ func (x *FilerConf) GetLocations() []*FilerConf_PathConf {
 	return nil
 }
 
-/////////////////////////
+// ///////////////////////
 // Remote Storage related
-/////////////////////////
+// ///////////////////////
 type CacheRemoteObjectToLocalClusterRequest struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache

weed/pb/filer_pb/filer_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.21.4
+// source: filer.proto
 
 package filer_pb
 

weed/pb/iam_pb/iam.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
-// 	protoc        v3.17.3
+// 	protoc-gen-go v1.28.1
+// 	protoc        v3.21.4
 // source: iam.proto
 
 package iam_pb

weed/pb/iam_pb/iam_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.21.4
+// source: iam.proto
 
 package iam_pb
 

weed/pb/master_pb/master.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
-// 	protoc        v3.17.3
+// 	protoc-gen-go v1.28.1
+// 	protoc        v3.21.4
 // source: master.proto
 
 package master_pb
@@ -1611,9 +1611,7 @@ func (x *StatisticsResponse) GetFileCount() uint64 {
 	return 0
 }
 
-//
 // collection related
-//
 type Collection struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -1848,9 +1846,7 @@ func (*CollectionDeleteResponse) Descriptor() ([]byte, []int) {
 	return file_master_proto_rawDescGZIP(), []int{23}
 }
 
-//
 // volume related
-//
 type DiskInfo struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache

weed/pb/master_pb/master_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.21.4
+// source: master.proto
 
 package master_pb
 

weed/pb/mount_pb/mount.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
-// 	protoc        v3.17.3
+// 	protoc-gen-go v1.28.1
+// 	protoc        v3.21.4
 // source: mount.proto
 
 package mount_pb

weed/pb/mount_pb/mount_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.21.4
+// source: mount.proto
 
 package mount_pb
 

weed/pb/mq_pb/mq.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
-// 	protoc        v3.17.3
+// 	protoc-gen-go v1.28.1
+// 	protoc        v3.21.4
 // source: mq.proto
 
 package mq_pb
@@ -20,7 +20,7 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-//////////////////////////////////////////////////
+// ////////////////////////////////////////////////
 type SegmentInfo struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -617,7 +617,7 @@ func (x *CheckBrokerLoadResponse) GetBytesCount() int64 {
 	return 0
 }
 
-//////////////////////////////////////////////////
+// ////////////////////////////////////////////////
 type PublishRequest struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache

weed/pb/mq_pb/mq_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.21.4
+// source: mq.proto
 
 package mq_pb
 

weed/pb/remote_pb/remote.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
-// 	protoc        v3.17.3
+// 	protoc-gen-go v1.28.1
+// 	protoc        v3.21.4
 // source: remote.proto
 
 package remote_pb
@@ -20,9 +20,9 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-/////////////////////////
+// ///////////////////////
 // Remote Storage related
-/////////////////////////
+// ///////////////////////
 type RemoteConf struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache

weed/pb/s3_pb/s3.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
-// 	protoc        v3.17.3
+// 	protoc-gen-go v1.28.1
+// 	protoc        v3.21.4
 // source: s3.proto
 
 package s3_pb

weed/pb/s3_pb/s3_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.21.4
+// source: s3.proto
 
 package s3_pb
 

weed/pb/volume_server.proto

@@ -297,6 +297,7 @@ message ReadNeedleMetaResponse {
     uint64 last_modified = 2;
     uint32 crc = 3;
     string ttl = 4;
+    uint64 append_at_ns = 5;
 }
 
 message WriteNeedleBlobRequest {

weed/pb/volume_server_pb/volume_server_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.21.4
+// source: volume_server.proto
 
 package volume_server_pb
 
@@ -18,7 +22,7 @@ const _ = grpc.SupportPackageIsVersion7
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type VolumeServerClient interface {
-	//Experts only: takes multiple fid parameters. This function does not propagate deletes to replicas.
+	// Experts only: takes multiple fid parameters. This function does not propagate deletes to replicas.
 	BatchDelete(ctx context.Context, in *BatchDeleteRequest, opts ...grpc.CallOption) (*BatchDeleteResponse, error)
 	VacuumVolumeCheck(ctx context.Context, in *VacuumVolumeCheckRequest, opts ...grpc.CallOption) (*VacuumVolumeCheckResponse, error)
 	VacuumVolumeCompact(ctx context.Context, in *VacuumVolumeCompactRequest, opts ...grpc.CallOption) (VolumeServer_VacuumVolumeCompactClient, error)
@@ -688,7 +692,7 @@ func (c *volumeServerClient) Ping(ctx context.Context, in *PingRequest, opts ...
 // All implementations must embed UnimplementedVolumeServerServer
 // for forward compatibility
 type VolumeServerServer interface {
-	//Experts only: takes multiple fid parameters. This function does not propagate deletes to replicas.
+	// Experts only: takes multiple fid parameters. This function does not propagate deletes to replicas.
 	BatchDelete(context.Context, *BatchDeleteRequest) (*BatchDeleteResponse, error)
 	VacuumVolumeCheck(context.Context, *VacuumVolumeCheckRequest) (*VacuumVolumeCheckResponse, error)
 	VacuumVolumeCompact(*VacuumVolumeCompactRequest, VolumeServer_VacuumVolumeCompactServer) error

weed/server/volume_grpc_read_write.go

@@ -49,6 +49,7 @@ func (vs *VolumeServer) ReadNeedleMeta(ctx context.Context, req *volume_server_p
 	if n.HasTtl() {
 		resp.Ttl = n.Ttl.String()
 	}
+	resp.AppendAtNs = n.AppendAtNs
 	return resp, nil
 }
 

weed/shell/command_volume_fsck.go

@@ -396,9 +396,8 @@ func (c *commandVolumeFsck) collectOneVolumeFileIds(dataNodeId string, volumeId
 				}
 				buf.Write(resp.FileContent)
 			}
-			fileredBuf := filterDeletedNeedleFromIdx(buf.Bytes())
 			if vinfo.isReadOnly == false {
-				index, err := idx.FirstInvalidIndex(fileredBuf.Bytes(),
+				index, err := idx.FirstInvalidIndex(buf.Bytes(),
 					func(key types.NeedleId, offset types.Offset, size types.Size) (bool, error) {
 						resp, err := volumeServerClient.ReadNeedleMeta(context.Background(), &volume_server_pb.ReadNeedleMetaRequest{
 							VolumeId: volumeId,
@@ -409,16 +408,16 @@ func (c *commandVolumeFsck) collectOneVolumeFileIds(dataNodeId string, volumeId
 						if err != nil {
 							return false, fmt.Errorf("to read needle meta with id %d from volume %d with error %v", key, volumeId, err)
 						}
-						return resp.LastModified <= cutoffFrom, nil
+						return resp.AppendAtNs <= cutoffFrom, nil
 					})
 				if err != nil {
 					fmt.Fprintf(c.writer, "Failed to search for last valid index on volume %d with error %v", volumeId, err)
 				} else {
-					fileredBuf.Truncate(index * types.NeedleMapEntrySize)
+					buf.Truncate(index * types.NeedleMapEntrySize)
 				}
 			}
 			idxFilename := getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)
-			err = writeToFile(fileredBuf.Bytes(), idxFilename)
+			err = writeToFile(buf.Bytes(), idxFilename)
 			if err != nil {
 				return fmt.Errorf("failed to copy %d%s from %s: %v", volumeId, ext, vinfo.server, err)
 			}
@@ -704,14 +703,3 @@ func writeToFile(bytes []byte, fileName string) error {
 	dst.Write(bytes)
 	return nil
 }
-
-func filterDeletedNeedleFromIdx(arr []byte) bytes.Buffer {
-	var filteredBuf bytes.Buffer
-	for i := 0; i < len(arr); i += types.NeedleMapEntrySize {
-		size := types.BytesToSize(arr[i+types.NeedleIdSize+types.OffsetSize : i+types.NeedleIdSize+types.OffsetSize+types.SizeSize])
-		if size > 0 {
-			filteredBuf.Write(arr[i : i+types.NeedleIdSize+types.OffsetSize+types.SizeSize])
-		}
-	}
-	return filteredBuf
-}

weed/storage/needle/needle_read_page.go

@@ -49,12 +49,11 @@ func (n *Needle) ReadNeedleMeta(r backend.BackendStorageFile, offset int64, size
 			return ErrorSizeMismatch
 		}
 	}
-	if !n.Size.IsValid() {
-		return ErrorSizeInvalid
-	}
 	n.DataSize = util.BytesToUint32(bytes[NeedleHeaderSize : NeedleHeaderSize+DataSizeSize])
-
-	startOffset := offset + NeedleHeaderSize + DataSizeSize + int64(n.DataSize)
+	startOffset := offset + NeedleHeaderSize
+	if size.IsValid() {
+		startOffset = offset + NeedleHeaderSize + DataSizeSize + int64(n.DataSize)
+	}
 	dataSize := GetActualSize(size, version)
 	stopOffset := offset + dataSize
 	metaSize := stopOffset - startOffset
@@ -69,12 +68,10 @@ func (n *Needle) ReadNeedleMeta(r backend.BackendStorageFile, offset int64, size
 	}
 	var index int
 	index, err = n.readNeedleDataVersion2NonData(metaSlice)
-
 	n.Checksum = CRC(util.BytesToUint32(metaSlice[index : index+NeedleChecksumSize]))
 	if version == Version3 {
 		n.AppendAtNs = util.BytesToUint64(metaSlice[index+NeedleChecksumSize : index+NeedleChecksumSize+TimestampSize])
 	}
-
 	return err
 
 }

weed/storage/volume_read.go

@@ -84,6 +84,10 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize
 func (v *Volume) readNeedleMetaAt(n *needle.Needle, offset int64, size int32) (err error) {
 	v.dataFileAccessLock.RLock()
 	defer v.dataFileAccessLock.RUnlock()
+	// read deleted needle meta data
+	if size < 0 {
+		size = 0
+	}
 	err = n.ReadNeedleMeta(v.DataBackend, offset, Size(size), v.Version())
 	if err == needle.ErrorSizeMismatch && OffsetSize == 4 {
 		err = n.ReadNeedleMeta(v.DataBackend, offset+int64(MaxPossibleVolumeSize), Size(size), v.Version())