Skip to content

The Raft consensus loop took the bytes of a committed log entry (or s… #5467

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
neeelkhadwal wants to merge 1 commit into
base: main
Choose a base branch
from
Open

The Raft consensus loop took the bytes of a committed log entry (or s… #5467

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 24 additions & 3 deletions orderer/consensus/etcdraft/chain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,10 @@ func NewChain(
var snapBlkNum uint64
var cc raftpb.ConfState
if s := storage.Snapshot(); !raft.IsEmptySnap(s) {
b := protoutil.UnmarshalBlockOrPanic(s.Data)
b, err := protoutil.UnmarshalBlock(s.Data)
if err != nil {
return nil, errors.Wrap(err, "failed to unmarshal block from raft snapshot")
}
snapBlkNum = b.Header.Number
cc = s.Metadata.ConfState
}
Expand Down Expand Up @@ -1189,7 +1192,17 @@ func (c *Chain) apply(ents []raftpb.Entry) {
break
}

block := protoutil.UnmarshalBlockOrPanic(ents[i].Data)
block, err := protoutil.UnmarshalBlock(ents[i].Data)
if err != nil {
// A committed Raft entry that does not unmarshal as a Block indicates either
// storage corruption or a byzantine proposer. Skipping the entry would advance
// appliedIndex past data that other followers may apply successfully, silently
// forking this node's state. Halt this chain so an operator can investigate;
// other channels' chains in this orderer process continue running.
c.logger.Errorf("Failed to unmarshal committed block at raft index %d: %s; halting chain", ents[i].Index, err)
go c.halt()
return
}
c.writeBlock(block, ents[i].Index)
c.Metrics.CommittedBlockNumber.Set(float64(block.Header.Number))

Expand Down Expand Up @@ -1250,7 +1263,15 @@ func (c *Chain) apply(ents []raftpb.Entry) {
// at position==0, ents[position].Type is ambiguous, it can be either of {raftpb.EntryNormal, raftpb.EntryConfChange}
// take a snapshot only for ents[position].Type == raftpb.EntryNormal
if c.accDataSize >= c.sizeLimit && ents[position].Type == raftpb.EntryNormal && len(ents[position].Data) > 0 {
b := protoutil.UnmarshalBlockOrPanic(ents[position].Data)
b, err := protoutil.UnmarshalBlock(ents[position].Data)
if err != nil {
// We just successfully unmarshalled and applied this same entry above; if we
// fail here, memory has been corrupted between then and now. Halt the chain
// rather than panic the orderer process.
c.logger.Errorf("Failed to unmarshal block at raft index %d while preparing snapshot: %s; halting chain", ents[position].Index, err)
go c.halt()
return
}
select {
case c.gcC <- &gc{index: c.appliedIndex, state: c.confState, data: ents[position].Data}:
c.logger.Infof("Accumulated %d bytes since last snapshot, exceeding size limit (%d bytes), "+
Expand Down