diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index ff01027a8..14769d76d 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -486,6 +486,7 @@ message VolumeEcShardReadRequest { message VolumeEcShardReadResponse { bytes data = 1; bool is_deleted = 2; + int64 encode_ts_ns = 3; // identity of the shard actually served; client rejects a mismatch (0 = pre-upgrade server) } message VolumeEcBlobDeleteRequest { diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 2cc21a31f..6d82373a4 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -3992,6 +3992,7 @@ type VolumeEcShardReadResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` IsDeleted bool `protobuf:"varint,2,opt,name=is_deleted,json=isDeleted,proto3" json:"is_deleted,omitempty"` + EncodeTsNs int64 `protobuf:"varint,3,opt,name=encode_ts_ns,json=encodeTsNs,proto3" json:"encode_ts_ns,omitempty"` // identity of the shard actually served; client rejects a mismatch (0 = pre-upgrade server) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -4040,6 +4041,13 @@ func (x *VolumeEcShardReadResponse) GetIsDeleted() bool { return false } +func (x *VolumeEcShardReadResponse) GetEncodeTsNs() int64 { + if x != nil { + return x.EncodeTsNs + } + return 0 +} + type VolumeEcBlobDeleteRequest struct { state protoimpl.MessageState `protogen:"open.v1"` VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` @@ -7289,11 +7297,13 @@ const file_volume_server_proto_rawDesc = "" + "\x04size\x18\x04 \x01(\x03R\x04size\x12\x19\n" + "\bfile_key\x18\x05 \x01(\x04R\afileKey\x12 \n" + "\fencode_ts_ns\x18\a \x01(\x03R\n" + - "encodeTsNsJ\x04\b\x06\x10\a\"N\n" + + "encodeTsNsJ\x04\b\x06\x10\a\"p\n" + "\x19VolumeEcShardReadResponse\x12\x12\n" + "\x04data\x18\x01 \x01(\fR\x04data\x12\x1d\n" + "\n" + - "is_deleted\x18\x02 \x01(\bR\tisDeleted\"\x8d\x01\n" + + "is_deleted\x18\x02 \x01(\bR\tisDeleted\x12 \n" + + "\fencode_ts_ns\x18\x03 \x01(\x03R\n" + + "encodeTsNs\"\x8d\x01\n" + "\x19VolumeEcBlobDeleteRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" + "\n" + diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index cd9cdbfa6..87954001c 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -686,7 +686,8 @@ func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardRea _, size, _ := ecVolume.FindNeedleFromEcx(types.Uint64ToNeedleId(req.FileKey)) if size.IsDeleted() { return stream.Send(&volume_server_pb.VolumeEcShardReadResponse{ - IsDeleted: true, + IsDeleted: true, + EncodeTsNs: ecVolume.EncodeTsNs, }) } } @@ -714,7 +715,8 @@ func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardRea bytesread = int(bytesToRead) } err = stream.Send(&volume_server_pb.VolumeEcShardReadResponse{ - Data: buffer[:bytesread], + Data: buffer[:bytesread], + EncodeTsNs: ecVolume.EncodeTsNs, }) if err != nil { // println("sending", bytesread, "bytes err", err.Error()) diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index d4aeaadfb..c67cddf24 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -590,6 +590,12 @@ func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, nee if receiveErr != nil { return fmt.Errorf("receiving ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, receiveErr) } + // Validate the served shard's identity client-side, so the guard holds + // even against a pre-upgrade server that ignored the request field (it + // returns 0). A mismatch fails the read; the caller recovers from parity. + if expectedEncodeTsNs != 0 && resp.EncodeTsNs != expectedEncodeTsNs { + return fmt.Errorf("ec shard %d.%d from %s belongs to a different encode run (want %d, got %d)", vid, shardId, sourceDataNode, expectedEncodeTsNs, resp.EncodeTsNs) + } if resp.IsDeleted { is_deleted = true }