diff --git a/apps/hotstuff/chainedhotstuff/crypto/crypto.go b/apps/hotstuff/chainedhotstuff/crypto/crypto.go new file mode 100644 index 00000000..16e74cff --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/crypto/crypto.go @@ -0,0 +1,97 @@ +package crypto + +import ( + "bytes" + "encoding/json" + pb "github.com/yu/apps/hotstuff/chainedhotstuff/proto" +) + +type Crypto struct { + PrivKey *ecdsa.PrivateKey + PubKey *ecdsa.PublicKey +} + +func NewCrypto(privKey *ecdsa.PrivateKey, pubKey *ecdsa.PublicKey) { + return &Crypto{PrivKey: PrivKey, PubKey: PubKey} +} + +func SignProposalMsg(msg *pb.ProposalMsg) (*pb.ProposalMsg, error) { + msgDigest, err := MakeProposalMsgDigest(msg) + + if err != nil { + return nil, err + } + + msg.MsgDigest = msgDigest + sig, err = SignECDSA(c.PrivateKey, msgDigest) + if err != nil { + return nil, err + } + + msg.Sig = &pb.QuorumCertSignature{ + Address: GetAddressFromPublicKey(c.PubKey), + PublicKey: GetEcdsaPublicKeyJsonFormat(c.PubKey), + Sig: sig, + } + + return msg, nil +} + +func MakeProposalMsgDigest(msg *pb.ProposalMsg) ([]byte, error) { + msgEncoder, err := encodeProposalMsg(msg) + + if err != nil { + return nil, err + } + + msg.MsgDigest = DoubleSha256(msgEncoder) + return msg.MsgDigest, nil +} + +func encodeProposalMsg(msg *pb.ProposalMsg) ([]byte, error) { + var msgBuf bytes.Buffer + encoder := json.NewEncoder(&msgBuf) + if err := encoder.Encode(msg.ProposalView); err != nil { + return nil, err + } + if err := encoder.Encode(msg.ProposalId); err != nil { + return nil, err + } + if err := encoder.Encode(msg.Timestamp); err != nil { + return nil, err + } + if err := encoder.Encode(msg.JustifyQC); err != nil { + return nil, err + } + return msgBuf.Bytes(), nil +} + +func (c *Crypto) SignVoteMsg(msg []byte) (*pb.QuorumCertSignature, error) { + sig, err := SignECDSA(c.PrivateKey, msg) + if err != nil { + return nil, err + } + + return &pb.QuorumCertSign{ + Address: GetAddressFromPublicKey(c.PubKey), + PublicKey: GetEcdsaPublicKeyJsonFormat(c.PubKey), + Sign: sign, + }, nil +} + +func (c *Crypto) VerifyVoteMsgSign(sig *pb.QuorumCertSignature, msg []byte) (bool, error) { + ak, err = GetEcdsaPublicKeyFromJsonStr(sig.GetPublicKey()) + if err != nil { + return false, err + } + + addr, err := GetAddressFromPublicKey(ak) + if err != nil { + return false, err + } + + if addr != sig.GetAddress() { + return false, errors.New("VerifyVoteMsgSign error, addr not match pk: " + addr) + } + return VerifyECDSA(ak, sig.GetSign(), msg) +} diff --git a/apps/hotstuff/chainedhotstuff/crypto/ecdsa.go b/apps/hotstuff/chainedhotstuff/crypto/ecdsa.go new file mode 100644 index 00000000..f7c1cf79 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/crypto/ecdsa.go @@ -0,0 +1,325 @@ +package ecdsa + +import ( + "crypto/ecdsa" + "encoding/json" + "ioutil" + "math/big" +) + +//将公钥序列化成byte数组 +func MarshalPublicKey(publicKey *ecdsa.PublicKey) []byte { + return elliptic.Marshal(publicKey.Curve, publicKey.X, publicKey.Y) +} + +func MarshalECDSASignature(r, s *big.Int) ([]byte, error) { + return asn1.Marshal(ECDSASignature{r, s}) + +} +func UnmarshalECDSASignature(rawSig []byte) (*big.Int, *big.Int, error) { + sig := new(ECDSASignature) + _, err := asn1.Unmarshal(rawSig, sig) + + if err != nil { + return nil, nil, fmt.Errorf("failed to unmashal the signature [%v] to R & S, and the error is [%s]", rawSig, err) + + } + + if sig.R == nil { + return nil, nil, errors.New("invalid signature, R is nil") + + } + if sig.S == nil { + return nil, nil, errors.New("invalid signature, S is nil") + + } + + if sig.R.Sign() != 1 { + return nil, nil, errors.New("invalid signature, R must be larger than zero") + + } + if sig.S.Sign() != 1 { + return nil, nil, errors.New("invalid signature, S must be larger than zero") + + } + + return sig.R, sig.S, nil + +} + +func SignECDSA(privKey *ecdsa.PrivateKey, hash byte) (sig []byte, err error) { + r, s, err := ecdsa.Sign(rand.Reader, privKey, hash[:]) + if err != nil { + return nil, logrus.Warn("ecdsa: sign failed. %w", err) + } + return MarshalECDSASignature(r, s) +} + +func VerifyECDSA(k *ecdsa.PublicKey, sig, msg []byte) (bool, error) { + r, s, err := UnmarshalECDSASignature(sig) + if err != nil { + return false, fmt.Errorf("Failed to unmarshal the ecdsa signature [%s]", err) + + } + + return ecdsa.Verify(k, msg, r, s), nil +} + +type ECDSAPrivateKey struct { + Curvname string + X, Y, D *big.Int +} + +// 通过这个数据结构来生成公钥的json +type ECDSAPublicKey struct { + Curvname string + X, Y *big.Int +} + +func getNewEcdsaPrivateKey(k *ecdsa.PrivateKey) *ECDSAPrivateKey { + key := new(ECDSAPrivateKey) + key.Curvname = k.Params().Name + key.D = k.D + key.X = k.X + key.Y = k.Y + + return key + +} + +func getNewEcdsaPublicKey(k *ecdsa.PrivateKey) *ECDSAPublicKey { + key := new(ECDSAPublicKey) + key.Curvname = k.Params().Name + key.X = k.X + key.Y = k.Y + + return key + +} + +func getNewEcdsaPublicKeyFromPublicKey(k *ecdsa.PublicKey) *ECDSAPublicKey { + key := new(ECDSAPublicKey) + key.Curvname = k.Params().Name + key.X = k.X + key.Y = k.Y + + return key +} + +// 获得公钥所对应的的json +func GetEcdsaPublicKeyJsonFormat(k *ecdsa.PrivateKey) (string, error) { + // 转换为自定义的数据结构 + key := getNewEcdsaPublicKey(k) + + // 转换json + data, err := json.Marshal(key) + + return string(data), err + +} + +// 获得公钥所对应的的json +func GetEcdsaPublicKeyJsonFormatFromPublicKey(k *ecdsa.PublicKey) (string, error) { + // 转换为自定义的数据结构 + key := getNewEcdsaPublicKeyFromPublicKey(k) + + // 转换json + data, err := json.Marshal(key) + + return string(data), err + +} + +func GetEcdsaPublicKeyJsonFormatStrFromPublicKe(k *ecdsa.PrivateKey) (string, error) { + return GetEcdsaPublicKeyJsonFormatFromPublicKey(k) +} + +func getAddressFromKeyData(pub *ecdsa.PublicKey, data []byte) (string, error) { + outputSha256 := hash.HashUsingSha256(data) + OutputRipemd160 := hash.HashUsingRipemd160(outputSha256) + + //暂时只支持一个字节长度，也就是uint8的密码学标志位 + // 判断是否是nist标准的私钥 + nVersion := config.Nist + + switch pub.Params().Name { + case config.CurveNist: // NIST + case config.CurveGm: // 国密 + nVersion = config.Gm + default: // 不支持的密码学类型 + return "", fmt.Errorf("This cryptography[%v] has not been supported yet.", pub.Params().Name) + + } + + bufVersion := []byte{byte(nVersion)} + + strSlice := make([]byte, len(bufVersion)+len(OutputRipemd160)) + copy(strSlice, bufVersion) + copy(strSlice[len(bufVersion):], OutputRipemd160) + + //using double SHA256 for future risks + checkCode := hash.DoubleSha256(strSlice) + simpleCheckCode := checkCode[:4] + + slice := make([]byte, len(strSlice)+len(simpleCheckCode)) + copy(slice, strSlice) + copy(slice[len(strSlice):], simpleCheckCode) + + //使用base58编码，手写不容易出错。 + //相比Base64，Base58不使用数字"0"，字母大写"O"，字母大写"I"，和字母小写"l"，以及"+"和"/"符号。 + strEnc := base58.Encode(slice) + + return strEnc, nil + +} + +//返回33位长度的地址 +func GetAddressFromPublicKey(pub *ecdsa.PublicKey) (string, error) { + //using SHA256 and Ripemd160 for hash summary + data := elliptic.Marshal(pub.Curve, pub.X, pub.Y) + + address, err := getAddressFromKeyData(pub, data) + + return address, err +} + +func readFileUsingFilename(filename string) ([]byte, error) { + // 从filename指定的文件中读取数据并返回文件的内容 + content, err := ioutil.ReadFile(filename) + if os.IsNotExist(err) { + log.Printf("File [%v] does not exist", filename) + } + if err != nil { + return nil, err + } + return content, err +} + +func GetEcdsaPrivateKeyFromJson(jsonContent []byte) (*ecdsa.PrivateKey, error) { + privateKey := new(ECDSAPrivateKey) + err := json.Unmarshal(jsonContent, privateKey) + if err != nil { + return nil, err + } + if privateKey.Curvname != "P-256" { + log.Printf("curve [%v] is not supported yet.", privateKey.Curvname) + err = fmt.Errorf("curve [%v] is not supported yet.", privateKey.Curvname) + return nil, err + } + ecdsaPrivateKey := &ecdsa.PrivateKey{} + ecdsaPrivateKey.PublicKey.Curve = elliptic.P256() + ecdsaPrivateKey.X = privateKey.X + ecdsaPrivateKey.Y = privateKey.Y + ecdsaPrivateKey.D = privateKey.D + + return ecdsaPrivateKey, nil +} + +func GetEcdsaPrivateKeyFromFile(filename string) (*ecdsa.PrivateKey, error) { + content, err := readFileUsingFilename(filename) + if err != nil { + log.Printf("readFileUsingFilename failed, the err is %v", err) + return nil, err + } + + return GetEcdsaPrivateKeyFromJson(content) +} + +func GetEcdsaPublicKeyFromJson(jsonContent []byte) (*ecdsa.PublicKey, error) { + publicKey := new(ECDSAPublicKey) + err := json.Unmarshal(jsonContent, publicKey) + if err != nil { + return nil, err //json有问题 + } + if publicKey.Curvname != "P-256" { + log.Printf("curve [%v] is not supported yet.", publicKey.Curvname) + err = fmt.Errorf("curve [%v] is not supported yet.", publicKey.Curvname) + return nil, err + } + ecdsaPublicKey := &ecdsa.PublicKey{} + ecdsaPublicKey.Curve = elliptic.P256() + ecdsaPublicKey.X = publicKey.X + ecdsaPublicKey.Y = publicKey.Y + + return ecdsaPublicKey, nil +} + +func GetEcdsaPublicKeyFromFile(filename string) (*ecdsa.PublicKey, error) { + content, err := readFileUsingFilename(filename) + if err != nil { + log.Printf("readFileUsingFilename failed, the err is %v", err) + return nil, err + } + + return GetEcdsaPublicKeyFromJson(content) +} + +type ECDSAPublicKey struct { + Curvname string + X, Y *big.Int +} + +func getNewEcdsaPrivateKey(k *ecdsa.PrivateKey) *ECDSAPrivateKey { + key := new(ECDSAPrivateKey) + key.Curvname = k.Params().Name + key.D = k.D + key.X = k.X + key.Y = k.Y + + return key + +} + +func getNewEcdsaPublicKey(k *ecdsa.PrivateKey) *ECDSAPublicKey { + key := new(ECDSAPublicKey) + key.Curvname = k.Params().Name + key.X = k.X + key.Y = k.Y + + return key + +} + +func getNewEcdsaPublicKeyFromPublicKey(k *ecdsa.PublicKey) *ECDSAPublicKey { + key := new(ECDSAPublicKey) + key.Curvname = k.Params().Name + key.X = k.X + key.Y = k.Y + + return key + +} + +// 获得私钥所对应的的json +func GetEcdsaPrivateKeyJsonFormat(k *ecdsa.PrivateKey) (string, error) { + // 转换为自定义的数据结构 + key := getNewEcdsaPrivateKey(k) + + // 转换json + data, err := json.Marshal(key) + + return string(data), err +} + +// 获得公钥所对应的的json +func GetEcdsaPublicKeyJsonFormat(k *ecdsa.PrivateKey) (string, error) { + // 转换为自定义的数据结构 + key := getNewEcdsaPublicKey(k) + + // 转换json + data, err := json.Marshal(key) + + return string(data), err +} + + +// 获得公钥所对应的的json +func GetEcdsaPublicKeyJsonFormatFromPublicKey(k *ecdsa.PublicKey) (string, error) { + // 转换为自定义的数据结构 + key := getNewEcdsaPublicKeyFromPublicKey(k) + + // 转换json + data, err := json.Marshal(key) + + return string(data), err +} diff --git a/apps/hotstuff/chainedhotstuff/crypto/hash.go b/apps/hotstuff/chainedhotstuff/crypto/hash.go new file mode 100644 index 00000000..4bb5c2aa --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/crypto/hash.go @@ -0,0 +1,27 @@ +import ( + "crypto/sha256" + "crypto/sha512" + "golang.org/x/crypto/ripemd160" +) + +func DoubleSha256(data []byte) []byte { + return HashUsingSha256(HashUsingSha256(data)) +} + +func HashUsingSha256(data []byte) []byte { + h := sha256.New() + h.Write(data) + out := h.Sum(nil) + + return out +} + +// Ripemd160，这种hash算法可以缩短长度 +func HashUsingRipemd160(data []byte) []byte { + h := ripemd160.New() + h.Write(data) + out := h.Sum(nil) + + return out + +} diff --git a/apps/hotstuff/chainedhotstuff/pacemaker.go b/apps/hotstuff/chainedhotstuff/pacemaker.go new file mode 100644 index 00000000..41b38f62 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/pacemaker.go @@ -0,0 +1,42 @@ +package chainedhotstuff + +import ( + "errors" + + "github.com/yu/apps/hotstuff/chainedhotstuff/storage" +) + +// PacemakerInterface is the interface of Pacemaker. It responsible for generating a new round. +// We assume Pacemaker in all correct replicas will have synchronized leadership after GST. +// Safty is entirely decoupled from liveness by any potential instantiation of Packmaker. +// Different consensus have different pacemaker implement +type IPacemaker interface { + // CurrentView return current view of this node. + GetCurrentView() int64 + // 原NextNewProposal，generate new proposal directly. + AdvanceView(qc IQuorumCert) (bool, error) +} + +// DefaultPaceMaker 是一个PacemakerInterface的默认实现，我们与PacemakerInterface放置在一起，方便查看 +// PacemakerInterface的新实现直接直接替代DefaultPaceMaker即可 +// The Pacemaker keeps track of votes and of time. +// TODO: the Pacemaker broadcasts a TimeoutMsg notification. +type DefaultPaceMaker struct { + CurrentView int64 + // timeout int64 +} + +func (p *DefaultPaceMaker) AdvanceView(qc storage.IQuorumCert) (bool, error) { + if qc == nil { + return false, ErrNilQC + } + r := qc.GetProposalView() + if r+1 > p.CurrentView { + p.CurrentView = r + 1 + } + return true, nil +} + +func (p *DefaultPaceMaker) GetCurrentView() int64 { + return p.CurrentView +} diff --git a/apps/hotstuff/chainedhotstuff/proposer_election.go b/apps/hotstuff/chainedhotstuff/proposer_election.go new file mode 100644 index 00000000..93e9a369 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/proposer_election.go @@ -0,0 +1,8 @@ +package chainedhotstuff + +type IProposerElection interface { + // 获取指定round的主节点Address, 注意, 若存在validators变更, 则需要在此处进行addrToIntAddr的更新操作 + GetLeader(round int64) string + // 获取指定round的候选人节点Address + GetValidators(round int64) []string +} diff --git a/apps/hotstuff/chainedhotstuff/proto/build.sh b/apps/hotstuff/chainedhotstuff/proto/build.sh new file mode 100644 index 00000000..df675af1 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/proto/build.sh @@ -0,0 +1 @@ +protoc -I . --go_opt=paths=source_relative --go_out=plugins=grpc:./ chainedbft.proto diff --git a/apps/hotstuff/chainedhotstuff/proto/chainedbft.pb.go b/apps/hotstuff/chainedhotstuff/proto/chainedbft.pb.go new file mode 100644 index 00000000..55dc8829 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/proto/chainedbft.pb.go @@ -0,0 +1,418 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.26.0 +// protoc v3.20.1 +// source: chainedbft.proto + +package __ + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type MsgType int32 + +const ( + MsgType_NEW_VIEW MsgType = 0 + MsgType_NEW_PROPOSAL MsgType = 1 + MsgType_VOTE MsgType = 2 +) + +// Enum value maps for MsgType. +var ( + MsgType_name = map[int32]string{ + 0: "NEW_VIEW", + 1: "NEW_PROPOSAL", + 2: "VOTE", + } + MsgType_value = map[string]int32{ + "NEW_VIEW": 0, + "NEW_PROPOSAL": 1, + "VOTE": 2, + } +) + +func (x MsgType) Enum() *MsgType { + p := new(MsgType) + *p = x + return p +} + +func (x MsgType) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (MsgType) Descriptor() protoreflect.EnumDescriptor { + return file_chainedbft_proto_enumTypes[0].Descriptor() +} + +func (MsgType) Type() protoreflect.EnumType { + return &file_chainedbft_proto_enumTypes[0] +} + +func (x MsgType) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use MsgType.Descriptor instead. +func (MsgType) EnumDescriptor() ([]byte, []int) { + return file_chainedbft_proto_rawDescGZIP(), []int{0} +} + +type QuorumCertSignature struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Address string `protobuf:"bytes,1,opt,name=Address,proto3" json:"Address,omitempty"` + PublicKey string `protobuf:"bytes,2,opt,name=PublicKey,proto3" json:"PublicKey,omitempty"` + Sig []byte `protobuf:"bytes,3,opt,name=Sig,proto3" json:"Sig,omitempty"` +} + +func (x *QuorumCertSignature) Reset() { + *x = QuorumCertSignature{} + if protoimpl.UnsafeEnabled { + mi := &file_chainedbft_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *QuorumCertSignature) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*QuorumCertSignature) ProtoMessage() {} + +func (x *QuorumCertSignature) ProtoReflect() protoreflect.Message { + mi := &file_chainedbft_proto_msgTypes[0] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use QuorumCertSignature.ProtoReflect.Descriptor instead. +func (*QuorumCertSignature) Descriptor() ([]byte, []int) { + return file_chainedbft_proto_rawDescGZIP(), []int{0} +} + +func (x *QuorumCertSignature) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + +func (x *QuorumCertSignature) GetPublicKey() string { + if x != nil { + return x.PublicKey + } + return "" +} + +func (x *QuorumCertSignature) GetSig() []byte { + if x != nil { + return x.Sig + } + return nil +} + +type ProposalMsg struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + ProposalView int64 `protobuf:"varint,1,opt,name=proposalView,proto3" json:"proposalView,omitempty"` + ProposalId []byte `protobuf:"bytes,2,opt,name=proposalId,proto3" json:"proposalId,omitempty"` + Timestamp int64 `protobuf:"varint,3,opt,name=timestamp,proto3" json:"timestamp,omitempty"` + JustifyQC []byte `protobuf:"bytes,4,opt,name=JustifyQC,proto3" json:"JustifyQC,omitempty"` + Sig *QuorumCertSignature `protobuf:"bytes,5,opt,name=Sig,proto3" json:"Sig,omitempty"` + MsgDigest []byte `protobuf:"bytes,6,opt,name=MsgDigest,proto3" json:"MsgDigest,omitempty"` +} + +func (x *ProposalMsg) Reset() { + *x = ProposalMsg{} + if protoimpl.UnsafeEnabled { + mi := &file_chainedbft_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *ProposalMsg) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ProposalMsg) ProtoMessage() {} + +func (x *ProposalMsg) ProtoReflect() protoreflect.Message { + mi := &file_chainedbft_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ProposalMsg.ProtoReflect.Descriptor instead. +func (*ProposalMsg) Descriptor() ([]byte, []int) { + return file_chainedbft_proto_rawDescGZIP(), []int{1} +} + +func (x *ProposalMsg) GetProposalView() int64 { + if x != nil { + return x.ProposalView + } + return 0 +} + +func (x *ProposalMsg) GetProposalId() []byte { + if x != nil { + return x.ProposalId + } + return nil +} + +func (x *ProposalMsg) GetTimestamp() int64 { + if x != nil { + return x.Timestamp + } + return 0 +} + +func (x *ProposalMsg) GetJustifyQC() []byte { + if x != nil { + return x.JustifyQC + } + return nil +} + +func (x *ProposalMsg) GetSig() *QuorumCertSignature { + if x != nil { + return x.Sig + } + return nil +} + +func (x *ProposalMsg) GetMsgDigest() []byte { + if x != nil { + return x.MsgDigest + } + return nil +} + +type VoteMsg struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + VoteInfo []byte `protobuf:"bytes,1,opt,name=VoteInfo,proto3" json:"VoteInfo,omitempty"` + LedgerCommitInfo []byte `protobuf:"bytes,2,opt,name=LedgerCommitInfo,proto3" json:"LedgerCommitInfo,omitempty"` + Sigs []*QuorumCertSignature `protobuf:"bytes,3,rep,name=sigs,proto3" json:"sigs,omitempty"` +} + +func (x *VoteMsg) Reset() { + *x = VoteMsg{} + if protoimpl.UnsafeEnabled { + mi := &file_chainedbft_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *VoteMsg) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*VoteMsg) ProtoMessage() {} + +func (x *VoteMsg) ProtoReflect() protoreflect.Message { + mi := &file_chainedbft_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use VoteMsg.ProtoReflect.Descriptor instead. +func (*VoteMsg) Descriptor() ([]byte, []int) { + return file_chainedbft_proto_rawDescGZIP(), []int{2} +} + +func (x *VoteMsg) GetVoteInfo() []byte { + if x != nil { + return x.VoteInfo + } + return nil +} + +func (x *VoteMsg) GetLedgerCommitInfo() []byte { + if x != nil { + return x.LedgerCommitInfo + } + return nil +} + +func (x *VoteMsg) GetSigs() []*QuorumCertSignature { + if x != nil { + return x.Sigs + } + return nil +} + +var File_chainedbft_proto protoreflect.FileDescriptor + +var file_chainedbft_proto_rawDesc = []byte{ + 0x0a, 0x10, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x65, 0x64, 0x62, 0x66, 0x74, 0x2e, 0x70, 0x72, 0x6f, + 0x74, 0x6f, 0x12, 0x0c, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x65, 0x64, 0x62, 0x66, 0x74, 0x70, 0x62, + 0x22, 0x5f, 0x0a, 0x13, 0x51, 0x75, 0x6f, 0x72, 0x75, 0x6d, 0x43, 0x65, 0x72, 0x74, 0x53, 0x69, + 0x67, 0x6e, 0x61, 0x74, 0x75, 0x72, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x41, 0x64, 0x64, 0x72, 0x65, + 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, + 0x73, 0x12, 0x1c, 0x0a, 0x09, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x4b, 0x65, 0x79, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x4b, 0x65, 0x79, 0x12, + 0x10, 0x0a, 0x03, 0x53, 0x69, 0x67, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x03, 0x53, 0x69, + 0x67, 0x22, 0xe0, 0x01, 0x0a, 0x0b, 0x50, 0x72, 0x6f, 0x70, 0x6f, 0x73, 0x61, 0x6c, 0x4d, 0x73, + 0x67, 0x12, 0x22, 0x0a, 0x0c, 0x70, 0x72, 0x6f, 0x70, 0x6f, 0x73, 0x61, 0x6c, 0x56, 0x69, 0x65, + 0x77, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0c, 0x70, 0x72, 0x6f, 0x70, 0x6f, 0x73, 0x61, + 0x6c, 0x56, 0x69, 0x65, 0x77, 0x12, 0x1e, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x6f, 0x73, 0x61, + 0x6c, 0x49, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x6f, + 0x73, 0x61, 0x6c, 0x49, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, + 0x6d, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, + 0x61, 0x6d, 0x70, 0x12, 0x1c, 0x0a, 0x09, 0x4a, 0x75, 0x73, 0x74, 0x69, 0x66, 0x79, 0x51, 0x43, + 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x4a, 0x75, 0x73, 0x74, 0x69, 0x66, 0x79, 0x51, + 0x43, 0x12, 0x33, 0x0a, 0x03, 0x53, 0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x21, + 0x2e, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x65, 0x64, 0x62, 0x66, 0x74, 0x70, 0x62, 0x2e, 0x51, 0x75, + 0x6f, 0x72, 0x75, 0x6d, 0x43, 0x65, 0x72, 0x74, 0x53, 0x69, 0x67, 0x6e, 0x61, 0x74, 0x75, 0x72, + 0x65, 0x52, 0x03, 0x53, 0x69, 0x67, 0x12, 0x1c, 0x0a, 0x09, 0x4d, 0x73, 0x67, 0x44, 0x69, 0x67, + 0x65, 0x73, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x4d, 0x73, 0x67, 0x44, 0x69, + 0x67, 0x65, 0x73, 0x74, 0x22, 0x88, 0x01, 0x0a, 0x07, 0x56, 0x6f, 0x74, 0x65, 0x4d, 0x73, 0x67, + 0x12, 0x1a, 0x0a, 0x08, 0x56, 0x6f, 0x74, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x0c, 0x52, 0x08, 0x56, 0x6f, 0x74, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x2a, 0x0a, 0x10, + 0x4c, 0x65, 0x64, 0x67, 0x65, 0x72, 0x43, 0x6f, 0x6d, 0x6d, 0x69, 0x74, 0x49, 0x6e, 0x66, 0x6f, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x10, 0x4c, 0x65, 0x64, 0x67, 0x65, 0x72, 0x43, 0x6f, + 0x6d, 0x6d, 0x69, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x35, 0x0a, 0x04, 0x73, 0x69, 0x67, 0x73, + 0x18, 0x03, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x65, 0x64, + 0x62, 0x66, 0x74, 0x70, 0x62, 0x2e, 0x51, 0x75, 0x6f, 0x72, 0x75, 0x6d, 0x43, 0x65, 0x72, 0x74, + 0x53, 0x69, 0x67, 0x6e, 0x61, 0x74, 0x75, 0x72, 0x65, 0x52, 0x04, 0x73, 0x69, 0x67, 0x73, 0x2a, + 0x33, 0x0a, 0x07, 0x4d, 0x73, 0x67, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0c, 0x0a, 0x08, 0x4e, 0x45, + 0x57, 0x5f, 0x56, 0x49, 0x45, 0x57, 0x10, 0x00, 0x12, 0x10, 0x0a, 0x0c, 0x4e, 0x45, 0x57, 0x5f, + 0x50, 0x52, 0x4f, 0x50, 0x4f, 0x53, 0x41, 0x4c, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x56, 0x4f, + 0x54, 0x45, 0x10, 0x02, 0x42, 0x04, 0x5a, 0x02, 0x2e, 0x2f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, + 0x6f, 0x33, +} + +var ( + file_chainedbft_proto_rawDescOnce sync.Once + file_chainedbft_proto_rawDescData = file_chainedbft_proto_rawDesc +) + +func file_chainedbft_proto_rawDescGZIP() []byte { + file_chainedbft_proto_rawDescOnce.Do(func() { + file_chainedbft_proto_rawDescData = protoimpl.X.CompressGZIP(file_chainedbft_proto_rawDescData) + }) + return file_chainedbft_proto_rawDescData +} + +var file_chainedbft_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_chainedbft_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_chainedbft_proto_goTypes = []interface{}{ + (MsgType)(0), // 0: chainedbftpb.MsgType + (*QuorumCertSignature)(nil), // 1: chainedbftpb.QuorumCertSignature + (*ProposalMsg)(nil), // 2: chainedbftpb.ProposalMsg + (*VoteMsg)(nil), // 3: chainedbftpb.VoteMsg +} +var file_chainedbft_proto_depIdxs = []int32{ + 1, // 0: chainedbftpb.ProposalMsg.Sig:type_name -> chainedbftpb.QuorumCertSignature + 1, // 1: chainedbftpb.VoteMsg.sigs:type_name -> chainedbftpb.QuorumCertSignature + 2, // [2:2] is the sub-list for method output_type + 2, // [2:2] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_chainedbft_proto_init() } +func file_chainedbft_proto_init() { + if File_chainedbft_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_chainedbft_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*QuorumCertSignature); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_chainedbft_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*ProposalMsg); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_chainedbft_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*VoteMsg); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_chainedbft_proto_rawDesc, + NumEnums: 1, + NumMessages: 3, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_chainedbft_proto_goTypes, + DependencyIndexes: file_chainedbft_proto_depIdxs, + EnumInfos: file_chainedbft_proto_enumTypes, + MessageInfos: file_chainedbft_proto_msgTypes, + }.Build() + File_chainedbft_proto = out.File + file_chainedbft_proto_rawDesc = nil + file_chainedbft_proto_goTypes = nil + file_chainedbft_proto_depIdxs = nil +} diff --git a/apps/hotstuff/chainedhotstuff/proto/chainedbft.proto b/apps/hotstuff/chainedhotstuff/proto/chainedbft.proto new file mode 100644 index 00000000..8279963a --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/proto/chainedbft.proto @@ -0,0 +1,74 @@ +syntax = "proto3"; + +option go_package = "./"; + +package chainedbft; + +message QuorumCertSignature { + string Address = 1; + string PublicKey = 2; + bytes Sig = 3; +} + +message Message { + enum MsgType { + NEW_VIEW = 0; + NEW_PROPOSAL = 1; + VOTE = 2; + } + + enum ErrorType { + // success + SUCCESS = 0; + NONE = 1; + // common error + UNKNOW_ERROR = 2; + CHECK_SUM_ERROR = 3; + UNMARSHAL_MSG_BODY_ERROR = 4; + CONNECT_REFUSE = 5; + // block error + GET_BLOCKCHAIN_ERROR = 6; + BLOCKCHAIN_NOTEXIST = 7; + GET_BLOCK_ERROR = 8; + CONFIRM_BLOCKCHAINSTATUS_ERROR = 9; + GET_AUTHENTICATION_ERROR = 10; + GET_AUTHENTICATION_NOT_PASS = 11; + + } + + message MsgHeader { + string version = 1; + string logid = 2; + string from = 3; + string bcname = 4; + MessageType type = 5; + uint32 dataCheckSum = 6; + ErrorType errorType = 7; + bool enableCompress = 8; + + } + + message MsgData { + // msgInfo is the message infomation, use protobuf coding style + bytes msgInfo = 3; + + } + + MsgHeader = 1; + MsgData = 2; +} + +message ProposalMsg { + int64 proposalView = 1; + bytes proposalId = 2; + int64 timestamp = 3; + bytes JustifyQC = 4; + QuorumCertSignature Sig = 5; + bytes MsgDigest = 6; +} + +message VoteMsg { + bytes VoteInfo = 1; + bytes LedgerCommitInfo = 2; + repeated QuorumCertSignature sigs = 3; +} diff --git a/apps/hotstuff/chainedhotstuff/saftyrules.go b/apps/hotstuff/chainedhotstuff/saftyrules.go new file mode 100644 index 00000000..299e1c60 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/saftyrules.go @@ -0,0 +1,181 @@ +package chainedhotstuff + +import ( + "errors" + + crypto "github.com/yu-org/yu/apps/hotstuff/chainedhotstuff/crypto" + "github.com/yu-org/yu/hotstuff/chainedhotstuff/storage" + "github.com/sirupsen/logrus" +) + +const ( + StrictInternal = 3 + PermissiveInternal = 6 +) + +var ( + EmptyVoteSignErr = errors.New("No signature in vote.") + InvalidVoteAddr = errors.New("Vote address is not a validator in the target validators.") + InvalidVoteSign = errors.New("Vote sign is invalid compared with its publicKey") + TooLowVoteView = errors.New("Vote received is lower than local lastVoteRound.") + TooLowVParentView = errors.New("Vote's parent received is lower than local preferredRound.") + TooLowProposalView = errors.New("Proposal received is lower than local lastVoteRound.") + EmptyParentQC = errors.New("Parent qc is empty.") + NoEnoughVotes = errors.New("Parent qc doesn't have enough votes.") + EmptyParentNode = errors.New("Parent's node is empty.") + EmptyValidators = errors.New("Justify validators are empty.") +) + +type saftyRulesInterface interface { + UpdatePreferredRound(round int64) bool + VoteProposal(proposalId []byte, proposalRound int64, parentQc storage.QuorumCertInterface) bool + CheckVote(qc storage.QuorumCertInterface, logid string, validators []string) error + CalVotesThreshold(input, sum int) bool + CheckProposal(proposal, parent storage.QuorumCertInterface, justifyValidators []string) error + CheckPacemaker(pending, local int64) bool +} + +type DefaultSaftyRules struct { + // lastVoteRound 存储着本地最近一次投票的轮数 + lastVoteRound int64 + // preferredRound 存储着本地PendingTree + // 即有[两个子孙节点的节点] + // 若本地有相同高度的节点，则自然排序后选出preferredRound + preferredRound int64 + Crypto *crypto.Crypto + QcTree *storage.QCPendingTree +} + +func (s *DefaultSaftyRules) UpdatePreferredRound(round int64) bool { + if round-1 > s.preferredRound { + s.preferredRound = round - 1 + } + // TODO: 检查LedgerInfo是否一致 + return true +} + +// VoteProposal 返回是否需要发送voteMsg给下一个Leader +// DefaultSaftyRules 并没有严格比对proposalRound和parentRound的相邻自增关系 +// 但需要注意的是，在上层bcs的实现中，由于共识操纵了账本回滚。因此实际上safetyrules需要proposalRound和parentRound严格相邻的 +// 因此由于账本的可回滚性，因此lastVoteRound和preferredRound比对时，仅需比对新来的数据是否小于local数据-3即可 +// 此处-3代表数据已经落盘 +func (s *DefaultSaftyRules) VoteProposal(proposalId []byte, proposalRound int64, parentQc storage.QuorumCertInterface) bool { + if proposalRound < s.lastVoteRound-StrictInternal { + return false + } + if parentQc.GetProposalView() < s.preferredRound-StrictInternal { + return false + } + s.increaseLastVoteRound(proposalRound) + return true +} + +// CheckVote 检查logid、voteInfoHash是否正确 +func (s *DefaultSaftyRules) CheckVote(qc storage.QuorumCertInterface, logid string, validators []string) error { + // 检查签名, vote目前为单个签名，因此只需要验证第一个即可，验证的内容为签名信息是否在合法的validators里面 + signs := qc.GetSignsInfo() + if len(signs) == 0 { + return EmptyVoteSignErr + } + // 是否是来自有效的候选人 + if !isInSlice(signs[0].GetAddress(), validators) { + logrus.Error("DefaultSaftyRules::CheckVote error", "validators", validators, "from", signs[0].GetAddress()) + return InvalidVoteAddr + } + // 签名和公钥是否匹配 + if ok, err := s.Crypto.VerifyVoteMsgSign(signs[0], qc.GetProposalId()); !ok { + return err + } + // 检查voteinfo信息, proposalView小于lastVoteRound，parentView不小于preferredRound + if qc.GetProposalView() < s.lastVoteRound-StrictInternal { + return TooLowVoteView + } + if qc.GetParentView() < s.preferredRound-StrictInternal { + return TooLowVParentView + } + // TODO: 检查commit消息 + return nil +} + +func (s *DefaultSaftyRules) increaseLastVoteRound(round int64) { + if round > s.lastVoteRound { + s.lastVoteRound = round + } +} + +func (s *DefaultSaftyRules) CalVotesThreshold(input, sum int) bool { + // 计算最大恶意节点数, input+1表示去除自己的签名 + f := (sum - 1) / 3 + if f < 0 { + return false + } + if f == 0 { + return input+1 >= sum + } + return input+1 >= sum-f +} + +// CheckProposalMsg 原IsQuorumCertValidate 判断justify，即需check的block的parentQC是否合法 +// 需要注意的是，在上层bcs的实现中，由于共识操纵了账本回滚。因此实际上safetyrules需要proposalRound和parentRound严格相邻的 +// 因此在此proposal和parent的QC稍微宽松检查 +func (s *DefaultSaftyRules) CheckProposal(proposal, parent storage.QuorumCertInterface, justifyValidators []string) error { + if proposal.GetProposalView() < s.lastVoteRound-PermissiveInternal { + return TooLowProposalView + } + if justifyValidators == nil { + return EmptyValidators + } + // step2: verify justify's votes + + // verify justify sign number + if parent.GetProposalId() == nil { + return EmptyParentQC + } + + // 新qc至少要在本地qcTree挂上, 那么justify的节点需要在本地 + // 或者新qc目前为孤儿节点，有可能未来切换成HighQC，此时仅需要proposal在[root+1, root+6] + // 是+6不是+3的原因是考虑到重起的时候的情况，重起时，root为tipId-3，而外界状态最多到tipId+3，此处简化处理 + if parentNode := s.QcTree.DFSQueryNode(parent.GetProposalId()); parentNode == nil { + if proposal.GetProposalView() <= s.QcTree.GetRootQC().In.GetParentView() || proposal.GetProposalView() > s.QcTree.GetRootQC().In.GetProposalView()+PermissiveInternal { + return EmptyParentNode + } + } + + // 检查justify的所有vote签名 + justifySigns := parent.GetSignsInfo() + logrus.Debug("DefaultSaftyRules::CheckProposal", "parent", parent, "justifyValidators", justifyValidators) + validCnt := 0 + for _, v := range justifySigns { + if !isInSlice(v.GetAddress(), justifyValidators) { + continue + } + // 签名和公钥是否匹配 + if ok, _ := s.Crypto.VerifyVoteMsgSign(v, parent.GetProposalId()); !ok { + return InvalidVoteSign + } + validCnt++ + } + if !s.CalVotesThreshold(validCnt, len(justifyValidators)) { + return NoEnoughVotes + } + return nil +} + +// CheckPacemaker +// 注意： 由于本smr支持不同节点产生同一round， 因此下述round比较和leader比较与原文(验证Proposal的Round是否和pacemaker的Round相等)并不同。 +// 仅需proposal round不超过范围即可 +func (s *DefaultSaftyRules) CheckPacemaker(pending int64, local int64) bool { + if pending <= local-StrictInternal { + return false + } + return true +} + +func isInSlice(target string, s []string) bool { + for _, v := range s { + if target == v { + return true + } + } + return false +} diff --git a/apps/hotstuff/chainedhotstuff/smr.go b/apps/hotstuff/chainedhotstuff/smr.go new file mode 100644 index 00000000..199a453e --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/smr.go @@ -0,0 +1,882 @@ +package chainedhotstuff + +import ( + "bytes" + "container/list" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + + "github.com/golang/protobuf/proto" + "github.com/golang/snappy" + "github.com/sirupsen/logrus" + pb "github.com/yu-org/yu/hotstuff/chainedhotstuff/proto" + "github.com/yu-org/yu/hotstuff/chainedhotstuff/storage" + + "github.com/sirupsen/logrus" + "github.com/yu-org/yu/hotstuff/chainedhotstuff/crypto" +) + +type IBlock interface { + GetProposer() []byte + GetHeight() int64 + GetBlockid() []byte + GetConsensusStorage() ([]byte, error) + GetTimestamp() int64 + SetItem(item string, value interface{}) error + MakeBlockId() ([]byte, error) + GetPreHash() []byte + GetNextHash() []byte + GetPublicKey() string + GetSign() []byte + GetTxIDs() []string + GetInTrunk() bool +} + +func VerifyChecksum(msg *pb.XuperMessage) bool { + return crc32.ChecksumIEEE(msg.GetData().GetMsgInfo()) == msg.GetHeader().GetDataCheckSum() +} + +func Unmarshal(msg *pb.Message, message proto.Message) error { + if !VerifyChecksum(msg) { + return errors.New("verify checksum error") + } + + data, err := Decompress(msg) + if err != nil { + return errors.New("decompress error") + } + + err = proto.Unmarshal(data, message) + if err != nil { + return errors.New("message unmarshal error") + } + + return nil +} + +type MessageOption func(*pb.Message) + +func GenPseudoUniqId() uint64 { + nano := time.Now().UnixNano() + + randNum1 := rand.Int63() + randNum2 := rand.Int63() + shift1 := rand.Intn(16) + 2 + shift2 := rand.Intn(8) + 1 + + uId := ((randNum1 >> uint(shift1)) + (randNum2 >> uint(shift2)) + (nano >> 1)) & + 0x1FFFFFFFFFFFFF + return uint64(uId) + +} + +func GenLogId() string { + return fmt.Sprintf("%d_%d", time.Now().Unix(), GenPseudoUniqId()) +} + +func NewMessage(ty pb.Message_MsgType, message proto.Message, opts ...MessageOption) pb.Message { + msg := pb.Message{ + Header: &pb.Message_MsgHeader{ + Version: "3.0.0", + Bcname: "yu", + Logid: GenLogId(), + Type: ty, + EnableCompress: false, + ErrorType: pb.Message_NONE, + }, + Data: message.Message_MsgData{}, + } + + if message != nil { + data, _ := proto.Marshal(message) + msg.Data.MsgInfo = data + } + + for _, op := range opts { + op(msg) + } + Compress(msg) + msg.Header.DataCheckSum = Checksum(msg) + return msg +} + +func WithBCName(bcname string) MessageOption { + return func(msg *pb.Message) { + msg.Header.Bcname = bcname + } +} + +type MarkPoint struct { + tag string + delta float64 +} + +type Timer struct { + bornTime int64 + latestTime int64 + points []*MarkPoint +} + +func NewTimer() *Timer { + now := time.Now().UnixNano() + return &Timer{ + bornTime: now, + latestTime: now, + } +} + +var ( + ErrTooLowNewView = errors.New("nextView is lower than local pacemaker's currentView") + ErrP2PInternalErr = errors.New("internal err in p2p module") + ErrTooLowNewProposal = errors.New("proposal is lower than local pacemaker's currentView") + ErrEmptyHighQC = errors.New("no valid highQC in qcTree") + ErrSameProposalNotify = errors.New("same proposal has been made") + ErrJustifyVotesEmpty = errors.New("justify qc's votes are empty") + ErrEmptyTarget = errors.New("target parameter is empty") + ErrRegisterErr = errors.New("register to p2p error") +) + +const ( + // DefaultNetMsgChanSize is the default size of network msg channel + DefaultNetMsgChanSize = 1000 +) + +// smr 组装了三个模块: pacemaker、saftyrules和propose election +// smr有自己的存储即PendingTree +// 原本的ChainedBft(联结smr和本地账本，在preferredVote被确认后, 触发账本commit操作) +// 被替代成smr和上层bcs账本的·组合实现，以减少不必要的代码，考虑到chained-bft暂无扩展性 +// 注意：本smr的round并不是强自增唯一的，不同节点可能产生相同round（考虑到上层账本的块可回滚） +type Smr struct { + bcName string + log loglogrus.er + address string // 包含一个私钥生成的地址 + // smr定义了自己需要的P2P消息类型 + // p2pMsgChan is the msg channel registered to network + //p2pMsgChan chan *xuperp2p.XuperMessage + // p2p interface + p2p *LibP2P + // cBFTCrypto 封装了本BFT需要的加密相关的接口和方法 + cryptoClient *crypto.Crypto + + // quitCh stop channel + quitCh chan bool + + pacemaker PacemakerInterface + saftyrules saftyRulesInterface + election IProposerElection + qcTree *storage.QCPendingTree + // smr本地存储和外界账本存储的唯一关联，该字段标识了账本状态， + // 但此处并不直接使用ledger handler作为变量，旨在结偶smr存储和本地账本存储 + // smr存储应该仅仅是账本区块头存储的很小的子集 + ledgerState int64 + + // map[proposalId]int64 + localProposal *sync.Map + // votes of QC in mem, key: voteId, value: []*QuorumCertSign + qcVoteMsgs *sync.Map + + // 该锁保护状态机处理msg或者bcs层操作过程，防止状态机get/set时由于bcs操作和msg处理并发导致的脏读脏写 + mtx sync.Mutex +} + +func NewSmr(bcName, address string, log loglogrus.er, p2p *LibP2P, cryptoClient *crypto.Crypto, pacemaker PacemakerInterface, + saftyrules saftyRulesInterface, election IProposerElection, qcTree *storage.QCPendingTree) *Smr { + s := &Smr{ + bcName: bcName, + log: log, + address: address, + //p2pMsgChan: make(chan *xuperp2p.XuperMessage, DefaultNetMsgChanSize), + p2p: p2p, + cryptoClient: cryptoClient, + quitCh: make(chan bool, 1), + pacemaker: pacemaker, + saftyrules: saftyrules, + election: election, + qcTree: qcTree, + localProposal: &sync.Map{}, + qcVoteMsgs: &sync.Map{}, + } + // smr初始值装载 + s.localProposal.Store(fmt.Sprintf("%x", qcTree.GetRootQC().In.GetProposalId()), 0) + if qcTree.GetHighQC() != nil { + s.ledgerState = int64(qcTree.GetHighQC().In.GetProposalView()) + } else if qcTree.GetGenericQC() != nil { + s.ledgerState = int64(qcTree.GetGenericQC().In.GetProposalView()) + } else { + s.ledgerState = int64(qcTree.GetRootQC().In.GetProposalView()) + } + return s +} + +func (s *Smr) LoadVotes(proposalId []byte, signs []*pb.QuorumCertSignature) { + if signs != nil { + s.qcVoteMsgs.Store(fmt.Sprintf("%x", proposalId), signs) + } +} + +func MsgTypeToString(msgType pb.MsgType) string { + switch msgType { + case MsgType_NEW_VIEW: + return "MSG_NEW_VIEW" + case MsgType_NEW_PROPOSAL: + return "MSG_NEW_PROPOSAL" + case MsgType_VOTE: + return "MSG_VOTE" + } + + return nil +} + +func (s *Smr) RegisterToNetwork() error { + sub1, err := s.p2p.SubP2P(MsgTypeToString(pb.MsgType_NEW_VIEW)) + if err != nil { + return err + } + + sub2, err := s.p2p.SubP2P(MsgTypeToString(pb.MsgType_NEW_PROPOSAL)) + if err != nil { + return err + } + + sub3, err := s.p2p.SubP2P(MsgTypeToString(pb.MsgType_NEW_PROPOSAL)) + if err != nil { + return err + } +} + +// Start used to start smr instance and process msg +func (s *Smr) Start() { + s.RegisterToNetwork() + go func() { + for { + select { + case msg := <-s.p2pMsgChan: + s.handleReceivedMsg(msg) + case <-s.quitCh: + return + } + } + }() +} + +// stop used to stop smr instance +func (s *Smr) Stop() { + s.quitCh <- true + s.UnRegisterToNetwork() +} + +// GetRootQC 查询状态树的Root节点，Root节点已经被账本commit +func (s *Smr) GetRootQC() storage.QuorumCertInterface { + return s.qcTree.GetRootQC().In +} + +func (s *Smr) GetCurrentView() int64 { + return s.pacemaker.GetCurrentView() +} + +func (s *Smr) GetAddress() string { + return s.address +} + +func (s *Smr) CheckProposal(block IBlock, justify storage.QuorumCertInterface, validators []string) error { + s.mtx.Lock() + defer s.mtx.Unlock() + + pNode := s.blockToProposalNode(block) + return s.saftyrules.CheckProposal(pNode.In, justify, validators) +} + +func (s *Smr) KeepUpWithBlock(block IBlock, justify storage.QuorumCertInterface, validators []string) error { + s.mtx.Lock() + defer s.mtx.Unlock() + + s.updateJustifyQcStatus(justify) + if validators != nil { + err := s.ProcessProposal(block.GetHeight(), block.GetBlockid(), block.GetPreHash(), validators) + if err != nil && err != ErrSameProposalNotify && err != ErrTooLowNewProposal { + return err + } + } + // 在不在候选人节点中，都直接调用smr生成新的qc树，矿工调用避免了proposal消息后于vote消息 + pNode := s.blockToProposalNode(block) + err := s.updateQcStatus(pNode) + if err != nil { + return err + } + s.qcTree.UpdateCommit(block.GetPreHash()) + s.pacemaker.AdvanceView(justify) + logrus.Debug("consensus:smr:KeepUpWithBlock: current parameters: ", "highQC", fmt.Sprintf("%x", s.getHighQC().GetProposalId()), "blockId", fmt.Sprintf("%x", block.GetBlockid()), + "pacemaker view", s.pacemaker.GetCurrentView(), "QCTree Root", fmt.Sprintf("%x", s.qcTree.GetRootQC().In.GetProposalId())) + return nil +} + +func (s *Smr) ResetProposerStatus(tipBlock IBlock, + queryBlockFunc func(blkId []byte) (IBlock, error), + validators []string) (bool, storage.QuorumCertInterface, error) { + s.mtx.Lock() + defer s.mtx.Unlock() + + if bytes.Equal(s.getHighQC().GetProposalId(), tipBlock.GetBlockid()) && + s.validNewHighQC(tipBlock.GetBlockid(), validators) { + // 此处需要获取带签名的完整Justify + return false, s.getCompleteHighQC(), nil + } + + // 从当前TipBlock开始往前追溯，交给smr根据状态进行回滚。 + // 在本地状态树上找到指代TipBlock的QC，若找不到，则在状态树上找和TipBlock同一分支上的最近值 + var qc storage.QuorumCertInterface + targetId := tipBlock.GetBlockid() + for { + block, err := queryBlockFunc(targetId) + if err != nil { + logrus.Error("consensus:smr:ResetProposerStatus: queryBlockFunc error.", "error", err) + return false, nil, ErrEmptyTarget + } + // 至多回滚到root节点 + if block.GetHeight() <= s.GetRootQC().GetProposalView() { + logrus.Warn("consensus:smr:ResetProposerStatus: set root qc.", "root", fmt.Sprintf("%x", s.GetRootQC().GetProposalId()), "root height", s.GetRootQC().GetProposalView(), + "block", fmt.Sprintf("%x", block.GetBlockid()), "block height", block.GetHeight()) + qc = s.GetRootQC() + break + } + // 查找目标Id是否挂在状态树上，若否，则从target网上查找知道状态树里有 + node := s.qcTree.DFSQueryNode(block.GetBlockid()) + if node == nil { + targetId = block.GetPreHash() + continue + } + // node在状态树上找到之后，以此为起点(包括当前点)，继续向上查找，知道找到符合全名数量要求的QC，该QC可强制转化为新的HighQC + wantProposers := s.election.GetValidators(block.GetHeight()) + if wantProposers == nil { + logrus.Error("consensus:smr:ResetProposerStatus: election error.") + return false, nil, ErrEmptyTarget + } + if !s.validNewHighQC(node.In.GetProposalId(), wantProposers) { + logrus.Warn("consensus:smr:ResetProposerStatus: target not ready", "target", fmt.Sprintf("%x", node.In.GetProposalId()), "wantProposers", wantProposers, "height", node.In.GetProposalView()) + targetId = block.GetPreHash() + continue + } + qc = node.In + break + } + if qc == nil { + return false, nil, ErrEmptyHighQC + } + ok, err := s.enforceUpdateHighQC(qc.GetProposalId()) + if err != nil { + logrus.Error("consensus:smr:ResetProposerStatus: EnforceUpdateHighQC error.", "error", err) + return false, nil, err + } + if ok { + logrus.Debug("consensus:smr:ResetProposerStatus: EnforceUpdateHighQC success.", "target", fmt.Sprintf("%x", qc.GetProposalId()), "height", qc.GetProposalView()) + } + // 此处需要获取带签名的完整Justify, 此时HighQC已经更新 + return true, s.getCompleteHighQC(), nil +} + +// handleReceivedMsg used to process msg received from network +func (s *Smr) handleReceivedMsg(msg *pb.Message) error { + // filter msg from other chain + if msg.GetHeader().GetBcname() != s.bcName { + return nil + } + switch msg.GetHeader().GetType() { + case pb.Message_NEW_PROPOSAL_MSG: + s.handleReceivedProposal(msg) + case pb.Message_VOTE_MSG: + s.handleReceivedVoteMsg(msg) + default: + logrus.Error("smr::handleReceivedMsg receive unknow type msg", "type", msg.GetHeader().GetType()) + return nil + } + return nil +} + +// UpdateJustifyQcStatus 用于支持可回滚的账本，生成相同高度的块 +// 为了支持生成相同round的块，需要拿到justify的full votes，因此需要在上层账本收到新块时调用，在CheckMinerMatch后 +// 注意：为了支持回滚操作，必须调用该函数 +func (s *Smr) updateJustifyQcStatus(justify storage.QuorumCertInterface) { + if justify == nil { + return + } + v, ok := s.qcVoteMsgs.Load(fmt.Sprintf("%x", justify.GetProposalId())) + var signs []*pb.QuorumCertSignature + if ok { + signs, _ = v.([]*pb.QuorumCertSignature) + } + justifySigns := justify.GetSignsInfo() + if justifySigns == nil { + return + } + signs = appendSigns(signs, justifySigns) + s.qcVoteMsgs.Store(fmt.Sprintf("%x", justify.GetProposalId()), signs) + // 根据justify check情况更新本地HighQC, 注意：由于CheckMinerMatch已经检查justify签名 + s.qcTree.UpdateHighQC(justify.GetProposalId()) +} + +// UpdateQcStatus 除了更新本地smr的QC之外，还更新了smr的和账本相关的状态，以此区别于smr receive proposal时的updateQcStatus +func (s *Smr) updateQcStatus(node *storage.ProposalNode) error { + if node == nil { + return ErrEmptyTarget + } + // 更新ledgerStatus + if node.In.GetProposalView() > s.ledgerState { + s.ledgerState = node.In.GetProposalView() + } + return s.qcTree.UpdateQcStatus(node) +} + +// ProcessProposal 即Chained-HotStuff的NewView阶段，LibraBFT的process_proposal阶段 +// 对于一个认为自己当前是Leader的节点，它试图生成一个新的提案，即一个新的QC，并广播 +// 本节点产生一个Proposal，该proposal包含一个最新的round, 最新的proposalId，一个parentQC，并将该消息组合成一个ProposalMsg消息给所有节点 +// 全部完成后leader更新本地localProposal +func (s *Smr) ProcessProposal(viewNumber int64, proposalID []byte, parentID []byte, validatesIpInfo []string) error { + // ATTENTION::TODO:: 由于本次设计面向的是viewNumber可能重复的BFT，因此账本回滚后高度会相同，在此用LockedQC高度为标记 + if validatesIpInfo == nil { + return ErrEmptyTarget + } + if s.pacemaker.GetCurrentView() != s.qcTree.GetGenesisQC().In.GetProposalView()+1 && + s.qcTree.GetLockedQC() != nil && s.pacemaker.GetCurrentView() < s.qcTree.GetLockedQC().In.GetProposalView() { + logrus.Error("smr::ProcessProposal error", "error", ErrTooLowNewProposal, "pacemaker view", s.pacemaker.GetCurrentView(), "lockQC view", + s.qcTree.GetLockedQC().In.GetProposalView()) + return ErrTooLowNewProposal + } + if s.getHighQC() == nil { + logrus.Error("smr::ProcessProposal empty HighQC error") + return ErrEmptyHighQC + } + if _, ok := s.localProposal.Load(fmt.Sprintf("%x", proposalID)); ok { + return ErrSameProposalNotify + } + // Libra-BFT中的parentQC为本地HighQC，但由于本系统支持回滚，故HighQC有可能在新QC生成时变更，否则会导致QC序错误 + // 故本系统的parentQC必须提前指定，不能是highQC + parentQuorumCert, err := s.reloadJustifyQC(parentID) + if err != nil { + logrus.Error("smr::ProcessProposal reloadJustifyQC error", "err", err) + return err + } + parentQuorumCertBytes, err := json.Marshal(parentQuorumCert) + if err != nil { + return err + } + proposal := &pb.ProposalMsg{ + ProposalView: viewNumber, + ProposalId: proposalID, + Timestamp: time.Now().UnixNano(), + JustifyQC: parentQuorumCertBytes, + } + propMsg, err := s.cryptoClient.SignProposalMsg(proposal) + if err != nil { + logrus.Error("smr::ProcessProposal SignProposalMsg error", "error", err) + return err + } + netMsg := NewMessage(pb.Message_NEW_PROPOSAL, propMsg, WithBCName(s.bcName)) + // 全部预备之后，再调用该接口 + if netMsg == nil { + logrus.Error("smr::ProcessProposal::NewMessage error") + return ErrP2PInternalErr + } + + //go s.p2p.SubP2P(MsgTypeToString(pb.MsgType_NEW_VIEW)) + go s.p2p.PubP2P(MsgTypeToString(pb.MsgType_NEW_PROPOSAL), proto.Marshal(netMsg)) + logrus.Debug("smr::ProcessProposal::proposal", "localAddress", s.address, "validatesIpInfo", validatesIpInfo, + "ProposalView", proposal.ProposalView, "ProposalId", fmt.Sprintf("%x", proposal.ProposalId), + "Timestamp", proposal.Timestamp, "JustifyQC", proposal.JustifyQC) + + s.localProposal.Store(fmt.Sprintf("%x", proposalID), proposal.Timestamp) + // 若为单候选人情况，则此处需要特殊处理，矿工需要给自己提前签名 + if len(validatesIpInfo) == 1 { + s.voteToSelf(viewNumber, proposalID, parentQuorumCert) + } + logrus.Debug("smr:ProcessProposal::new proposal has been made", "address", s.address, "proposalID", fmt.Sprintf("%x", proposalID), "target", validatesIpInfo) + return nil +} + +func (s *Smr) voteToSelf(viewNumber int64, proposalID []byte, parent storage.QuorumCertInterface) { + selfVote := &storage.VoteInfo{ + ProposalId: proposalID, + ProposalView: viewNumber, + ParentId: parent.GetProposalId(), + } + selfLedgerInfo := &storage.LedgerCommitInfo{ + VoteInfoHash: proposalID, + } + selfQC := storage.NewQuorumCert(selfVote, selfLedgerInfo, nil) + selfSign, err := s.cryptoClient.SignVoteMsg(proposalID) + if err != nil { + logrus.Error("smr::voteProposal::voteToSelf error", "err", err) + return + } + s.qcVoteMsgs.LoadOrStore(fmt.Sprintf("%x", proposalID), []*pb.QuorumCertSign{selfSign}) + selfNode := &storage.ProposalNode{ + In: selfQC, + } + if err := s.qcTree.UpdateQcStatus(selfNode); err != nil { + logrus.Error("smr::voteProposal::updateQcStatus error", "err", err) + return + } + // 更新本地smr状态机 + s.pacemaker.AdvanceView(selfQC) + s.qcTree.UpdateHighQC(proposalID) + logrus.Debug("smr:voteProposal::done local voting", "address", s.address, "proposalID", fmt.Sprintf("%x", proposalID)) +} + +// reloadJustifyQC 与LibraBFT不同，返回一个指定的parentQC +func (s *Smr) reloadJustifyQC(parentID []byte) (storage.QuorumCertInterface, error) { + // 第一次proposal，highQC==rootQC==genesisQC + if bytes.Equal(s.qcTree.GetGenesisQC().In.GetProposalId(), parentID) { + highQC := s.getHighQC() + return highQC, nil + } + // 若当前找不到，可能是qcTree已经更新了，废弃 + qc := s.qcTree.DFSQueryNode(parentID) + if qc == nil { + return nil, ErrEmptyTarget + } + v := &storage.VoteInfo{ + ProposalView: qc.In.GetProposalView(), + ProposalId: qc.In.GetProposalId(), + } + // 查看qcTree是否包含当前可以commit的Id + var commitId []byte + if s.qcTree.GetCommitQC() != nil { + commitId = s.qcTree.GetCommitQC().In.GetProposalId() + } + + // 根据qcTree生成一个parentQC + // 上一个view的votes + value, ok := s.qcVoteMsgs.Load(fmt.Sprintf("%x", v.ProposalId)) + if !ok { + return nil, ErrJustifyVotesEmpty + } + signs, _ := value.([]*pb.QuorumCertSignature) + parentQuorumCert := storage.NewQuorumCert(v, &storage.LedgerCommitInfo{ + CommitStateId: commitId, + }, signs) + return parentQuorumCert, nil +} + +// handleReceivedProposal 该阶段在收到一个ProposalMsg后触发，与LibraBFT的process_proposal阶段类似 +// 该阶段分两个角色，一个是认为自己是currentRound的Leader，一个是Replica +// 0. 查看ProposalMsg消息的合法性 +// 1. 检查新的view是否符合账本状态要求 +// 2. 比较本地pacemaker是否需要AdvanceRound +// 3. 检查qcTree是否需要更新CommitQC +// 4. 查看收到的view是否符合要求 +// 5. 向本地PendingTree插入该QC，即更新QC +// 6. 发送一个vote消息给下一个Leader +// 注意：该过程删除了当前round的leader是否符合计算，将该步骤后置到上层共识CheckMinerMatch，原因：需要支持上层基于时间调度而不是基于round调度，减小耦合 +func (s *Smr) handleReceivedProposal(msg *proto.Message) { + s.mtx.Lock() + defer s.mtx.Unlock() + + newProposalMsg := &pb.ProposalMsg{} + if err := Unmarshal(msg, newProposalMsg); err != nil { + logrus.Error("smr::handleReceivedProposal Unmarshal msg error", "logid", msg.GetHeader().GetLogid(), "error", err) + return + } + + _, ok := s.localProposal.LoadOrStore(fmt.Sprintf("%x", newProposalMsg.GetProposalId()), newProposalMsg.Timestamp) + if ok && newProposalMsg.GetSign().Address != s.address { + return + } + + logrus.Debug("smr::handleReceivedProposal::received a proposal", "logid", msg.GetHeader().GetLogid(), + "newView", newProposalMsg.GetProposalView(), "newProposalId", fmt.Sprintf("%x", newProposalMsg.GetProposalId())) + parentQCBytes := newProposalMsg.GetJustifyQC() + parentQC := &storage.QuorumCert{} + if err := json.Unmarshal(parentQCBytes, parentQC); err != nil { + logrus.Error("smr::handleReceivedProposal Unmarshal parentQC error", "error", err) + return + } + + newVote := &storage.VoteInfo{ + ProposalId: newProposalMsg.GetProposalId(), + ProposalView: newProposalMsg.GetProposalView(), + ParentId: parentQC.GetProposalId(), + ParentView: parentQC.GetProposalView(), + } + isFirstJustify := bytes.Equal(s.qcTree.GetGenesisQC().In.GetProposalId(), parentQC.GetProposalId()) + // 0.若为初始状态，则无需检查justify，否则需要检查qc有效性 + if !isFirstJustify { + proposalQC := storage.NewQuorumCert(newVote, nil, []*pb.QuorumCertSignature{newProposalMsg.GetSign()}) + if err := s.saftyrules.CheckProposal(proposalQC, parentQC, s.election.GetValidators(parentQC.GetProposalView())); err != nil { + logrus.Debug("smr::handleReceivedProposal::CheckProposal error", "error", err, + "parentView", parentQC.GetProposalView(), "parentId", fmt.Sprintf("%x", parentQC.GetProposalId())) + return + } + } + // 1.检查账本状态和收到新round是否符合要求 + if s.ledgerState+StrictInternal < newVote.ProposalView { + logrus.Error("smr::handleReceivedProposal::local IBlockledger hasn't been updated.", "LedgerState", s.ledgerState, "ProposalView", newVote.ProposalView) + return + } + // 2.本地pacemaker试图更新currentView, 并返回一个是否需要将新消息通知该轮Leader, 是该轮不是下轮！主要解决P2PIP端口不能通知Loop的问题 + sendMsg, _ := s.pacemaker.AdvanceView(parentQC) + logrus.Debug("smr::handleReceivedProposal::pacemaker update", "view", s.pacemaker.GetCurrentView()) + // 通知current Leader + if sendMsg { + netMsg := NewMessage(pb.Message_NEW_PROPOSAL, newProposalMsg, WithBCName(s.bcName)) + leader := newProposalMsg.GetSign().GetAddress() + // 此处如果失败，仍会执行下层逻辑，因为是多个节点通知该轮Leader，因此若发不出去仍可继续运行 + if leader != "" && netMsg != nil && leader != s.address { + go s.p2p.PubP2P(MsgTypeToString(pb.MsgType_NEW_PROPOSAL), proto.Marshal(netMsg)) + logrus.Debug("smr::handleReceivedProposal::proposal", "localAddress", s.address, "leader", leader, + "ProposalView", newProposalMsg.ProposalView, "ProposalId", fmt.Sprintf("%x", newProposalMsg.ProposalId), + "Timestamp", newProposalMsg.Timestamp, "JustifyQC", newProposalMsg.JustifyQC) + } + } + + // 3.本地safetyrules更新, 如有可以commit的QC，执行commit操作并更新本地rootQC + if parentQC.LedgerCommitInfo != nil && parentQC.LedgerCommitInfo.CommitStateId != nil && + s.saftyrules.UpdatePreferredRound(parentQC.GetProposalView()) { + s.qcTree.UpdateCommit(parentQC.GetProposalId()) + } + // 4.查看收到的view是否符合要求, 此处接受孤儿节点 + if !s.saftyrules.CheckPacemaker(newProposalMsg.GetProposalView(), s.pacemaker.GetCurrentView()) { + logrus.Error("smr::handleReceivedProposal::error", "error", ErrTooLowNewProposal, "local want", s.pacemaker.GetCurrentView(), + "proposal have", newProposalMsg.GetProposalView()) + return + } + + // 注意：删除此处的验证收到的proposal是否符合local计算，在本账本状态中后置到上层共识CheckMinerMatch + // 根据本地saftyrules返回是否 需要发送voteMsg给下一个Leader + if !s.saftyrules.VoteProposal(newProposalMsg.GetProposalId(), newProposalMsg.GetProposalView(), parentQC) { + logrus.Error("smr::handleReceivedProposal::VoteProposal fail", "view", newProposalMsg.GetProposalView(), "proposalId", newProposalMsg.GetProposalId()) + return + } + + // 这个newVoteId表示的是本地最新一次vote的id，生成voteInfo的hash，标识vote消息 + newLedgerInfo := &storage.LedgerCommitInfo{ + VoteInfoHash: newProposalMsg.GetProposalId(), + } + newNode := &storage.ProposalNode{ + In: storage.NewQuorumCert(newVote, newLedgerInfo, nil), + } + // 5.与proposal.ParentId相比，更新本地qcTree，insert新节点, 包括更新CommitQC等等 + if err := s.qcTree.UpdateQcStatus(newNode); err != nil { + logrus.Error("smr::handleReceivedProposal::updateQcStatus error", "err", err) + return + } + logrus.Debug("smr::handleReceivedProposal::pacemaker changed", "round", s.pacemaker.GetCurrentView()) + // 6.发送一个vote消息给下一个Leader + nextLeader := s.election.GetLeader(s.pacemaker.GetCurrentView() + 1) + if nextLeader == "" { + logrus.Warn("smr::handleReceivedProposal::empty next leader", "next round", s.pacemaker.GetCurrentView()+1) + return + } + s.voteProposal(newProposalMsg.GetProposalId(), newVote, newLedgerInfo, nextLeader) +} + +// voteProposal 当Replica收到一个Proposal并对该Proposal检查之后，该节点会针对该QC投票 +// 节点的vote包含一个本次vote的对象的基本信息，和本地上次vote对象的基本信息，和本地账本的基本信息，和一个签名 +// 只要vote过，就在本地map中更新值 +func (s *Smr) voteProposal(msg []byte, vote *storage.VoteInfo, ledger *storage.LedgerCommitInfo, voteTo string) { + // 若为自己直接先返回 + if voteTo == s.address { + return + } + nextSign, err := s.cryptoClient.SignVoteMsg(msg) + if err != nil { + logrus.Error("smr::voteProposal::SignVoteMsg error", "err", err) + return + } + voteBytes, err := json.Marshal(vote) + if err != nil { + logrus.Error("smr::voteProposal::Marshal vote error", "err", err) + return + } + ledgerBytes, err := json.Marshal(ledger) + if err != nil { + logrus.Error("smr::voteProposal::Marshal commit error", "err", err) + return + } + voteMsg := &pb.VoteMsg{ + VoteInfo: voteBytes, + LedgerCommitInfo: ledgerBytes, + Signature: []*pb.QuorumCertSignature{nextSign}, + } + netMsg := NewMessage(pb.Message_VOTE, voteMsg, WithBCName(s.bcName)) + // 全部预备之后，再调用该接口 + if netMsg == nil { + logrus.Error("smr::ProcessProposal::NewMessage error") + return + } + go s.p2p.PubP2P(MsgTypeToString(pb.MsgType_VOTE), proto.Marshal(netMsg)) + logrus.Debug("smr::voteProposal::vote", "vote to next leader", voteTo, "vote view number", vote.ProposalView) +} + +// handleReceivedVoteMsg 当前Leader在发送一个proposal消息之后，由下一Leader等待周围replica的投票，收集vote消息 +// 当收到2f+1个vote消息之后，本地pacemaker调用AdvanceView，并更新highQC +// 该方法针对Leader而言 +func (s *Smr) handleReceivedVoteMsg(msg *pb.Message) error { + s.mtx.Lock() + defer s.mtx.Unlock() + + newVoteMsg := &pb.VoteMsg{} + if err := Unmarshal(msg, newVoteMsg); err != nil { + logrus.Error("smr::handleReceivedVoteMsg Unmarshal msg error", "logid", msg.GetHeader().GetLogid(), "error", err) + return err + } + voteQC, err := s.voteMsgToQC(newVoteMsg) + if err != nil { + logrus.Error("smr::handleReceivedVoteMsg VoteMsgToQC error", "error", err) + return err + } + // 检查logid、voteInfoHash是否正确 + if err := s.saftyrules.CheckVote(voteQC, msg.GetHeader().GetLogid(), s.election.GetValidators(voteQC.GetProposalView())); err != nil { + logrus.Error("smr::handleReceivedVoteMsg CheckVote error", "error", err, "msg", fmt.Sprintf("%x", voteQC.GetProposalId())) + return err + } + logrus.Debug("smr::handleReceivedVoteMsg::receive vote", "voteId", fmt.Sprintf("%x", voteQC.GetProposalId()), "voteView", voteQC.GetProposalView(), "from", voteQC.GetSignsInfo()[0].Address) + + // 若vote先于proposal到达，则直接丢弃票数 + if _, ok := s.localProposal.Load(fmt.Sprintf("%x", voteQC.GetProposalId())); !ok { + logrus.Debug("smr::handleReceivedVoteMsg::haven't received the related proposal msg, drop it.") + return ErrEmptyTarget + } + if node := s.qcTree.DFSQueryNode(voteQC.GetProposalId()); node == nil { + logrus.Debug("smr::handleReceivedVoteMsg::haven't finish proposal process, drop it.") + return ErrEmptyTarget + } + + // 存入本地voteInfo内存，查看签名数量是否超过2f+1 + var VoteLen int + // 注意隐式，若!ok则证明签名数量为1，此时不可能超过2f+1 + v, ok := s.qcVoteMsgs.LoadOrStore(fmt.Sprintf("%x", voteQC.GetProposalId()), voteQC.GetSignsInfo()) + // 若ok=false，则仅store一个vote签名 + VoteLen = 1 + if ok { + signs, _ := v.([]*pb.QuorumCertSignature) + stored := false + for _, sign := range signs { + // 自己给自己投票将自动忽略 + if sign.Address == voteQC.GetSignsInfo()[0].Address || voteQC.GetSignsInfo()[0].Address == s.address { + stored = true + } + } + if !stored { + signs = append(signs, voteQC.GetSignsInfo()[0]) + s.qcVoteMsgs.Store(fmt.Sprintf("%x", voteQC.GetProposalId()), signs) + } + VoteLen = len(signs) + } + // 查看签名数量是否达到2f+1, 需要获取justify对应的validators + if !s.saftyrules.CalVotesThreshold(VoteLen, len(s.election.GetValidators(voteQC.GetProposalView()))) { + return nil + } + + // 更新本地pacemaker AdvanceRound + s.pacemaker.AdvanceView(voteQC) + logrus.Debug("smr::handleReceivedVoteMsg::FULL VOTES!", "pacemaker view", s.pacemaker.GetCurrentView()) + // 更新HighQC + s.qcTree.UpdateHighQC(voteQC.GetProposalId()) + return nil +} + +// voteMsgToQC 提供一个从VoteMsg转化为quorumCert的方法，注意，两者struct其实相仿 +func (s *Smr) voteMsgToQC(msg *pb.VoteMsg) (storage.QuorumCertInterface, error) { + voteInfo := &storage.VoteInfo{} + if err := json.Unmarshal(msg.VoteInfo, voteInfo); err != nil { + return nil, err + } + ledgerCommitInfo := &storage.LedgerCommitInfo{} + if err := json.Unmarshal(msg.LedgerCommitInfo, ledgerCommitInfo); err != nil { + return nil, err + } + return storage.NewQuorumCert(voteInfo, ledgerCommitInfo, msg.GetSignature()), nil +} + +func (s *Smr) blockToProposalNode(block IBlock) *storage.ProposalNode { + targetId := block.GetBlockid() + if node := s.qcTree.DFSQueryNode(targetId); node != nil { + return node + } + v := &storage.VoteInfo{ + ProposalId: block.GetBlockid(), + ProposalView: block.GetHeight(), + ParentId: block.GetPreHash(), + ParentView: block.GetHeight() - 1, + } + return &storage.ProposalNode{In: storage.NewQuorumCert(v, nil, nil)} +} + +func (s *Smr) getHighQC() storage.QuorumCertInterface { + return s.qcTree.GetHighQC().In +} + +// getCompleteHighQC 本地qcTree不带签名，因此smr需要重新组装完整的QC +func (s *Smr) getCompleteHighQC() storage.QuorumCertInterface { + raw := s.getHighQC() + vote := &storage.VoteInfo{ + ProposalId: raw.GetProposalId(), + ProposalView: raw.GetProposalView(), + ParentId: raw.GetParentProposalId(), + ParentView: raw.GetProposalView(), + } + signInfo, ok := s.qcVoteMsgs.Load(fmt.Sprintf("%x", raw.GetProposalId())) + if !ok { + return storage.NewQuorumCert(vote, nil, nil) + } + signs, _ := signInfo.([]*pb.QuorumCertSignature) + return storage.NewQuorumCert(vote, nil, signs) +} + +func (s *Smr) validNewHighQC(inProposalId []byte, validators []string) bool { + signInfo, ok := s.qcVoteMsgs.Load(fmt.Sprintf("%x", inProposalId)) + if !ok { + return false + } + signs, ok := signInfo.([]*pb.QuorumCertSignature) + if !ok { + return false + } + if len(validators) == 1 { + return len(signs) == len(validators) + } + return s.saftyrules.CalVotesThreshold(len(signs), len(validators)) +} + +func (s *Smr) enforceUpdateHighQC(inProposalId []byte) (bool, error) { + if bytes.Equal(s.getHighQC().GetProposalId(), inProposalId) { + return false, nil + } + return true, s.qcTree.EnforceUpdateHighQC(inProposalId) +} + +func (s *Smr) removeLocalValidator(in []string) []string { + var out []string + for _, addr := range in { + if addr != s.address { + out = append(out, addr) + } + } + return out +} + +// func createNewBCtx() *xctx.BaseCtx { +// log, _ := logs.NewLogger("", "smr") +// return &xctx.BaseCtx{ +// XLog: log, +// Timer: timer.NewXTimer(), +// } +// } + +// appendSigns 将p中不重复的签名append进q中 +func appendSigns(q []*pb.QuorumCertSignature, p []*pb.QuorumCertSignature) []*pb.QuorumCertSignature { + signSet := make(map[string]bool) + for _, sign := range q { + if _, ok := signSet[sign.Address]; !ok { + signSet[sign.Address] = true + } + } + for _, sign := range p { + if _, ok := signSet[sign.Address]; !ok { + q = append(q, sign) + } + } + return q +} diff --git a/apps/hotstuff/chainedhotstuff/storage/qc_tree.go b/apps/hotstuff/chainedhotstuff/storage/qc_tree.go new file mode 100644 index 00000000..bc1cd58a --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/storage/qc_tree.go @@ -0,0 +1,447 @@ +package storage + +import ( + "fmt" + "bytes" + "container/list" + "errors" + "sync" + + cctx "github.com/xuperchain/xupercore/kernel/consensus/context" + "github.com/sirupsen/logrus" +) + +type ProposalNode struct { + In QuorumCertInterface + // Parent QuorumCertInterface + Sons []*ProposalNode + // Parent *ProposalNode +} + +type LedgerRely interface { + GetConsensusConf() ([]byte, error) + QueryBlockHeader(blkId []byte) (ledger.BlockHandle, error) + QueryBlockHeaderByHeight(int64) (ledger.BlockHandle, error) + GetTipBlock() ledger.BlockHandle + GetTipXMSnapshotReader() (ledger.XMSnapshotReader, error) + CreateSnapshot(blkId []byte) (ledger.XMReader, error) + GetTipSnapshot() (ledger.XMReader, error) + QueryTipBlockHeader() ledger.BlockHandle +} + + +func NewTreeNode(ledger LedgerRely, height int64) *ProposalNode { + b, err := ledger.QueryBlockHeaderByHeight(height) + if err != nil { + return nil + } + pre, err := ledger.QueryBlockHeaderByHeight(height - 1) + vote := VoteInfo{ + ProposalId: b.GetBlockid(), + ProposalView: b.GetHeight(), + } + ledgerInfo := LedgerCommitInfo{ + CommitStateId: b.GetBlockid(), + } + if err != nil { + return &ProposalNode{ + In: NewQuorumCert(&vote, &ledgerInfo, nil), + } + } + vote.ParentId = pre.GetBlockid() + vote.ParentView = pre.GetHeight() + return &ProposalNode{ + In: NewQuorumCert(&vote, &ledgerInfo, nil), + } +} + +// PendingTree 是一个内存内的QC状态存储树，仅存放目前未commit(即可能触发账本回滚)的区块信息 +// 当PendingTree中的某个节点有[严格连续的]三代子孙后，将出发针对该节点的账本Commit操作 +// 本数据结构替代原有Chained-BFT的三层QC存储，即proposalQC,generateQC和lockedQC +type QCPendingTree struct { + genesis *ProposalNode // Tree中第一个Node + root *ProposalNode + highQC *ProposalNode // Tree中最高的QC指针 + genericQC *ProposalNode + lockedQC *ProposalNode + commitQC *ProposalNode + + orphanList *list.List // []*ProposalNode孤儿数组 + orphanMap map[string]bool + + mtx sync.RWMutex + + log logs.Logger +} + +func MockTree(genesis *ProposalNode, root *ProposalNode, highQC *ProposalNode, + genericQC *ProposalNode, lockedQC *ProposalNode, commitQC *ProposalNode, + log logs.Logger) *QCPendingTree { + return &QCPendingTree{ + genesis: genesis, + root: root, + highQC: highQC, + genericQC: genericQC, + lockedQC: lockedQC, + commitQC: commitQC, + log: log, + orphanList: list.New(), + orphanMap: make(map[string]bool), + } +} + +// initQCTree 创建了smr需要的QC树存储，该Tree存储了目前待commit的QC信息 +func InitQCTree(startHeight int64, ledger cctx.LedgerRely) *QCPendingTree { + // 初始状态应该是start高度的前一个区块为genesisQC，即tipBlock + g, err := ledger.QueryBlockHeaderByHeight(startHeight - 1) + if err != nil { + logrus.Warn("InitQCTree QueryBlockHeaderByHeight failed", "error", err.Error()) + return nil + } + gQC := NewQuorumCert( + &VoteInfo{ + ProposalId: g.GetBlockid(), + ProposalView: g.GetHeight(), + }, + &LedgerCommitInfo{ + CommitStateId: g.GetBlockid(), + }, + nil) + gNode := &ProposalNode{ + In: gQC, + } + tip := ledger.GetTipBlock() + // 当前为初始状态 + if tip.GetHeight() <= startHeight { + return &QCPendingTree{ + genesis: gNode, + root: gNode, + highQC: gNode, + log: log, + orphanList: list.New(), + orphanMap: make(map[string]bool), + } + } + // 重启状态时将root->tipBlock-3, generic->tipBlock-2, highQC->tipBlock-1 + // 若tipBlock<=2, root->genesisBlock, highQC->tipBlock-1 + tipNode := NewTreeNode(ledger, tip.GetHeight()) + if tip.GetHeight() < 3 { + tree := &QCPendingTree{ + genesis: gNode, + root: NewTreeNode(ledger, 0), + log: log, + orphanList: list.New(), + orphanMap: make(map[string]bool), + } + switch tip.GetHeight() { + case 0: + tree.highQC = tree.root + return tree + case 1: + tree.highQC = tree.root + tree.highQC.Sons = append(tree.highQC.Sons, tipNode) + return tree + case 2: + tree.highQC = NewTreeNode(ledger, 1) + tree.highQC.Sons = append(tree.highQC.Sons, tipNode) + tree.root.Sons = append(tree.root.Sons, tree.highQC) + } + return tree + } + tree := &QCPendingTree{ + genesis: gNode, + root: NewTreeNode(ledger, tip.GetHeight()-3), + genericQC: NewTreeNode(ledger, tip.GetHeight()-2), + highQC: NewTreeNode(ledger, tip.GetHeight()-1), + log: log, + orphanList: list.New(), + orphanMap: make(map[string]bool), + } + // 手动组装Tree结构 + tree.root.Sons = append(tree.root.Sons, tree.genericQC) + tree.genericQC.Sons = append(tree.genericQC.Sons, tree.highQC) + tree.highQC.Sons = append(tree.highQC.Sons, tipNode) + return tree +} + +func (t *QCPendingTree) MockGetOrphan() *list.List { + t.mtx.RLock() + defer t.mtx.RUnlock() + return t.orphanList +} + +func (t *QCPendingTree) GetGenesisQC() *ProposalNode { + t.mtx.RLock() + defer t.mtx.RUnlock() + return t.genesis +} + +func (t *QCPendingTree) GetRootQC() *ProposalNode { + t.mtx.RLock() + defer t.mtx.RUnlock() + return t.root +} + +func (t *QCPendingTree) GetGenericQC() *ProposalNode { + t.mtx.RLock() + defer t.mtx.RUnlock() + return t.genericQC +} + +func (t *QCPendingTree) GetCommitQC() *ProposalNode { + t.mtx.RLock() + defer t.mtx.RUnlock() + return t.commitQC +} + +func (t *QCPendingTree) GetLockedQC() *ProposalNode { + t.mtx.RLock() + defer t.mtx.RUnlock() + return t.lockedQC +} + +func (t *QCPendingTree) GetHighQC() *ProposalNode { + t.mtx.RLock() + defer t.mtx.RUnlock() + return t.highQC +} + +// DFSQueryNode实现的比较简单，从root节点开始寻找，后续有更优方法可优化 +func (t *QCPendingTree) DFSQueryNode(id []byte) *ProposalNode { + t.mtx.RLock() + defer t.mtx.RUnlock() + return dfsQuery(t.root, id) +} + +// updateCommit 此方法向存储接口发送一个ProcessCommit，通知存储落盘，此时的block将不再被回滚 +// 同时此方法将原先的root更改为commit node，因为commit node在本BFT中已确定不会回滚 +func (t *QCPendingTree) UpdateCommit(id []byte) { + t.mtx.Lock() + defer t.mtx.Unlock() + + node := dfsQuery(t.root, id) + if node == nil { + return + } + parent := dfsQuery(t.root, node.In.GetParentProposalId()) + if parent == nil { + return + } + parentParent := dfsQuery(t.root, parent.In.GetParentProposalId()) + if parentParent == nil { + return + } + parentParentParent := dfsQuery(t.root, parentParent.In.GetParentProposalId()) + if parentParentParent == nil { + return + } + parentParentParentParent := dfsQuery(t.root, parentParentParent.In.GetParentProposalId()) + if parentParentParentParent == nil { + return + } + parentParentParentParent.Sons = nil + t.root = parentParentParent + // TODO: commitQC/lockedQC/genericQC/highQC是否有指向原root及以上的Node +} + +// 更新本地qcTree, insert新节点, 将新节点parentQC和本地HighQC对比，如有必要进行更新 +func (t *QCPendingTree) UpdateQcStatus(node *ProposalNode) error { + t.mtx.Lock() + defer t.mtx.Unlock() + + if node.Sons == nil { + node.Sons = make([]*ProposalNode, 0) + } + if dfsQuery(t.root, node.In.GetProposalId()) != nil { + logrus.Debug("QCPendingTree::updateQcStatus::has been inserted", "search", fmt.Sprintf("%x", node.In.GetProposalId())) + return nil + } + if err := t.insert(node); err != nil { + logrus.Error("QCPendingTree::updateQcStatus insert err", "err", err) + return err + } + logrus.Debug("QCPendingTree::updateQcStatus", "insert new", fmt.Sprintf("%x", node.In.GetProposalId()), "height", node.In.GetProposalView(), "highQC", fmt.Sprintf("%x", t.highQC.In.GetProposalId())) + + // HighQCs试图更新成收到node的parentQC + parent := dfsQuery(t.root, node.In.GetParentProposalId()) + if parent == nil { + logrus.Debug("QCPendingTree::updateHighQC::orphan", "id", fmt.Sprintf("%x", node.In.GetParentProposalId())) + return nil + } + // 若新验证过的node和原HighQC高度相同，使用新验证的node + if parent.In.GetProposalView() < t.highQC.In.GetProposalView() { + return nil + } + t.updateQCs(parent) + return nil +} + +// updateHighQC 对比QC树，将本地HighQC和输入id比较，高度更高的更新为HighQC，此时连同GenericQC、LockedQC、CommitQC一起修改 +func (t *QCPendingTree) UpdateHighQC(inProposalId []byte) { + t.mtx.Lock() + defer t.mtx.Unlock() + + node := dfsQuery(t.root, inProposalId) + if node == nil { + logrus.Debug("QCPendingTree::updateHighQC::dfsQuery nil!", "id", fmt.Sprintf("%x", inProposalId)) + return + } + // 若新验证过的node和原HighQC高度相同，使用新验证的node + if node.In.GetProposalView() < t.highQC.In.GetProposalView() { + return + } + t.updateQCs(node) +} + +// enforceUpdateHighQC 强制更改HighQC指针，用于错误时回滚，注意: 本实现没有timeoutQC因此需要此方法 +func (t *QCPendingTree) EnforceUpdateHighQC(inProposalId []byte) error { + t.mtx.Lock() + defer t.mtx.Unlock() + + node := dfsQuery(t.root, inProposalId) + if node == nil { + logrus.Debug("QCPendingTree::enforceUpdateHighQC::dfsQuery nil") + return ErrNoValidQC + } + logrus.Debug("QCPendingTree::enforceUpdateHighQC::start.") + return t.updateQCs(node) +} + +func (t *QCPendingTree) updateQCs(highQCNode *ProposalNode) error { + // 更改HighQC以及一系列的GenericQC、LockedQC和CommitQC + t.highQC = highQCNode + t.genericQC = nil + t.lockedQC = nil + t.commitQC = nil + logrus.Debug("QCPendingTree::updateHighQC", "HighQC height", highQCNode.In.GetProposalView(), "HighQC", fmt.Sprintf("%x", highQCNode.In.GetProposalId())) + parent := dfsQuery(t.root, highQCNode.In.GetParentProposalId()) + if parent == nil { + return nil + } + t.genericQC = parent + logrus.Debug("QCPendingTree::updateHighQC", "GenericQC height", t.genericQC.In.GetProposalView(), "GenericQC", fmt.Sprintf("%x", t.genericQC.In.GetProposalId())) + // 找grand节点，标为LockedQC + parentParent := dfsQuery(t.root, parent.In.GetParentProposalId()) + if parentParent == nil { + return nil + } + t.lockedQC = parentParent + logrus.Debug("QCPendingTree::updateHighQC", "LockedQC height", t.lockedQC.In.GetProposalView(), "LockedQC", fmt.Sprintf("%x", t.lockedQC.In.GetProposalId())) + // 找grandgrand节点，标为CommitQC + parentParentParent := dfsQuery(t.root, parentParent.In.GetParentProposalId()) + if parentParentParent == nil { + return nil + } + t.commitQC = parentParentParent + logrus.Debug("QCPendingTree::updateHighQC", "CommitQC height", t.commitQC.In.GetProposalView(), "CommitQC", fmt.Sprintf("%x", t.commitQC.In.GetProposalId())) + return nil +} + +// insert 向本地QC树Insert一个ProposalNode，如有必要，连同HighQC、GenericQC、LockedQC、CommitQC一起修改 +func (t *QCPendingTree) insert(node *ProposalNode) error { + if node.In == nil { + logrus.Error("QCPendingTree::insert err", "err", ErrNoValidQC) + return ErrNoValidQC + } + if node.In.GetParentProposalId() == nil { + return ErrNoValidParentId + } + parent := dfsQuery(t.root, node.In.GetParentProposalId()) + if parent != nil { + parent.Sons = append(parent.Sons, node) + t.adoptOrphans(node) + return nil + } + // 作为孤儿节点加入 + t.insertOrphan(node) + return nil +} + +// insertOrphan为向孤儿数组插入孤儿节点的逻辑 +// 若该node的父节点不存在在slice中，则查看该node的是否为slice中节点的父节点，若是则代替该节点反转挂上，若否继续查看 +// 若该node的父节点存在在sli中，则直接挂在父节点下，如否则在sli中追加节点 +// [A1, B1, C1, D1 ...] +// ｜ || +// A2 B2 B2' +// | +// A3 +func (t *QCPendingTree) insertOrphan(node *ProposalNode) error { + if _, ok := t.orphanMap[fmt.Sprintf("%x", node.In.GetProposalId())]; ok { + return nil // 重复退出 + } + t.orphanMap[fmt.Sprintf("%x", node.In.GetProposalId())] = true + if t.orphanList.Len() == 0 { + t.orphanList.PushBack(node) + return nil + } + // 遍历整个Sli，查看是否能够挂上 + ptr := t.orphanList.Front() + for ptr != nil { + curPtr := ptr + n, ok := curPtr.Value.(*ProposalNode) + if !ok { + return errors.New("QCPendingTree::insertOrphan::element type invalid") + } + ptr = ptr.Next() + // 查看头节点是否已经时间失效了, 失效的时候所有依赖该高度长得树实际上都没有意义了，需要删除 + if n.In.GetProposalView() <= t.root.In.GetProposalView() { + t.orphanList.Remove(curPtr) + continue + } + // 查看头节点是否是node的儿子, 直接在头部插入 + if bytes.Equal(n.In.GetParentProposalId(), node.In.GetProposalId()) { + node.Sons = append(node.Sons, n) + t.orphanList.Remove(curPtr) + t.orphanList.PushBack(node) + return nil + } + // 否则遍历该树试图挂在子树上面 + parent := dfsQuery(n, node.In.GetParentProposalId()) + if parent != nil { + parent.Sons = append(parent.Sons, node) + return nil + } + } + // 没有可以挂的地方，则直接append + t.orphanList.PushBack(node) + return nil +} + +// adoptOrphans 查看孤儿节点列表是否可以挂在该节点上 +func (t *QCPendingTree) adoptOrphans(node *ProposalNode) error { + if t.orphanList.Len() == 0 { + return nil + } + ptr := t.orphanList.Front() + for ptr != nil { + curPtr := ptr + n, ok := curPtr.Value.(*ProposalNode) + if !ok { + return errors.New("QCPendingTree::insertOrphan::element type invalid") + } + ptr = ptr.Next() + if bytes.Equal(n.In.GetParentProposalId(), node.In.GetProposalId()) { + node.Sons = append(node.Sons, n) + t.orphanList.Remove(curPtr) + } + } + return nil +} + +func dfsQuery(node *ProposalNode, target []byte) *ProposalNode { + if target == nil || node == nil { + return nil + } + if bytes.Equal(node.In.GetProposalId(), target) { + return node + } + if node.Sons == nil { + return nil + } + for _, node := range node.Sons { + if n := dfsQuery(node, target); n != nil { + return n + } + } + return nil +} diff --git a/apps/hotstuff/chainedhotstuff/storage/quorum_cert.go b/apps/hotstuff/chainedhotstuff/storage/quorum_cert.go new file mode 100644 index 00000000..cb0ff856 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/storage/quorum_cert.go @@ -0,0 +1,55 @@ +package storage + +import ( + "errors" + + pb "github.com/yu-org/yu/apps/hotstuff/chainedhotstuff/proto" +) + +var _ QuorumCertInterface = (*QuorumCert)(nil) + +var ( + ErrNoValidQC = errors.New("target qc is empty") + ErrNoValidParentId = errors.New("parentId is empty") +) + +// 本文件定义了chained-bft下有关的数据结构和接口 +// QuorumCertInterface 规定了pacemaker和saftyrules操作的qc接口 +// QuorumCert 为一个QuorumCertInterface的实现 TODO: smr彻底接口化是否可能? +// QCPendingTree 规定了smr内存存储的组织形式，其为一个QC树状结构 + +// IQuorumCert 接口 +type IQuorumCert interface { + GetProposalView() int64 + GetProposalId() []byte + GetParentProposalId() []byte + GetParentView() int64 + GetSignsInfo() []*pb.QuorumCertSignature +} + +// VoteInfo 包含了本次和上次的vote对象 +type VoteInfo struct { + // 本次vote的对象 + ProposalId []byte + ProposalView int64 + // 本地上次vote的对象 + ParentId []byte + ParentView int64 +} + +// ledgerCommitInfo 表示的是本地账本和QC存储的状态，包含一个commitStateId和一个voteInfoHash +// commitStateId 表示本地账本状态，TODO: = 本地账本merkel root +// voteInfoHash 表示本地vote的vote_info的哈希，即本地QC的最新状态 +type LedgerCommitInfo struct { + CommitStateId []byte + VoteInfoHash []byte +} + +func NewQuorumCert(v *VoteInfo, l *LedgerCommitInfo, s []*pb.QuorumCertSignature) QuorumCertInterface { + qc := QuorumCert{ + VoteInfo: v, + LedgerCommitInfo: l, + SignInfos: s, + } + return &qc +} diff --git a/apps/hotstuff/chainedhotstuff/storage/quorum_cert_impl.go b/apps/hotstuff/chainedhotstuff/storage/quorum_cert_impl.go new file mode 100644 index 00000000..8ec85363 --- /dev/null +++ b/apps/hotstuff/chainedhotstuff/storage/quorum_cert_impl.go @@ -0,0 +1,35 @@ +package storage + +import ( + pb "github.com/yu-org/yu/apps/hotstuff/chainedhotstuff/proto" +) + +// quorumCert 是HotStuff的基础结构，它表示了一个节点本地状态以及其余节点对该状态的确认 +type QuorumCert struct { + // 本次qc的vote对象，该对象中嵌入了上次的QCid，因此删除原有的ProposalMsg部分 + VoteInfo *VoteInfo + // 当前本地账本的状态 + LedgerCommitInfo *LedgerCommitInfo + // SignInfos is the signs of the leader gathered from replicas of a specifically certType. + SignInfos []*pb.QuorumCertSignature +} + +func (qc *QuorumCert) GetProposalView() int64 { + return qc.VoteInfo.ProposalView +} + +func (qc *QuorumCert) GetProposalId() []byte { + return qc.VoteInfo.ProposalId +} + +func (qc *QuorumCert) GetParentProposalId() []byte { + return qc.VoteInfo.ParentId +} + +func (qc *QuorumCert) GetParentView() int64 { + return qc.VoteInfo.ParentView +} + +func (qc *QuorumCert) GetSignsInfo() []*pb.QuorumCertSignature { + return qc.SignInfos +}