POS-141: Minor code cleanup (#13)

* consensus.go: Log messages formatting fix,
state.go: copying seal fix, variables renaming
go.mod: tidy dependencies

* Remove pushMessage internal function

* Remove FIXME comment

consensus.go

@@ -248,7 +248,7 @@ func (p *Pbft) runAcceptState(ctx context.Context) { // start new round
 
  if !p.state.validators.Includes(p.validator.NodeID()) {
  // we are not a validator anymore, move back to sync state
- p.logger.Printf("[INFO] we are not a validator anymore")
+ p.logger.Print("[INFO] we are not a validator anymore")
  p.setState(SyncState)
  return
  }
@@ -329,7 +329,7 @@ func (p *Pbft) runAcceptState(ctx context.Context) { // start new round
 
  // retrieve the proposal
  if err := p.backend.Validate(msg.Proposal); err != nil {
- p.logger.Printf("[ERROR] failed to validate proposal: %v", err)
+ p.logger.Printf("[ERROR] failed to validate proposal. Error message: %v", err)
  p.setState(RoundChangeState)
  return
  }
@@ -482,7 +482,7 @@ func (p *Pbft) runCommitState(ctx context.Context) {
  if err := p.backend.Insert(pp); err != nil {
  // start a new round with the state unlocked since we need to
  // be able to propose/validate a different proposal
- p.logger.Print("[ERROR] failed to insert proposal", "err", err)
+ p.logger.Printf("[ERROR] failed to insert proposal. Error message: %v", err)
  p.handleStateErr(errFailedToInsertProposal)
  } else {
  p.setSequence(p.state.view.Sequence + 1)
@@ -543,13 +543,13 @@ func (p *Pbft) runRoundChangeState(ctx context.Context) {
  // if the round was triggered due to an error, we send our own
  // next round change
  if err := p.state.getErr(); err != nil {
- p.logger.Print("[DEBUG] round change handle err", "err", err)
+ p.logger.Printf("[DEBUG] round change handle error. Error message: %v", err)
  sendNextRoundChange()
  } else {
  // otherwise, it is due to a timeout in any stage
  // First, we try to sync up with any max round already available
  if maxRound, ok := p.state.maxRound(); ok {
- p.logger.Print("[DEBUG] round change set max round", "round", maxRound)
+ p.logger.Printf("[DEBUG] round change, max round=%d", maxRound)
  sendRoundChange(maxRound)
  } else {
  // otherwise, do your best to sync up
@@ -569,7 +569,7 @@ func (p *Pbft) runRoundChangeState(ctx context.Context) {
  return
  }
  if msg == nil {
- p.logger.Printf("[DEBUG] round change timeout")
+ p.logger.Print("[DEBUG] round change timeout")
  checkTimeout()
  // update the timeout duration
  timeout = p.exponentialTimeout()
@@ -637,7 +637,7 @@ func (p *Pbft) gossip(typ MsgType) {
 
  seal, err := p.validator.Sign(hash)
  if err != nil {
- p.logger.Print("[ERROR] failed to commit seal", "err", err)
+ p.logger.Printf("[ERROR] failed to commit seal. Error message: %v", err)
  return
  }
  msg.Seal = seal
@@ -647,10 +647,10 @@ func (p *Pbft) gossip(typ MsgType) {
  // send a copy to ourselves so that we can process this message as well
  msg2 := msg.Copy()
  msg2.From = p.validator.NodeID()
- p.pushMessage(msg2)
+ p.PushMessage(msg2)
  }
  if err := p.transport.Gossip(msg); err != nil {
- p.logger.Print("[ERROR] failed to gossip", "err", err)
+ p.logger.Printf("[ERROR] failed to gossip. Error message: %v", err)
  }
 }
 
@@ -732,12 +732,8 @@ func (p *Pbft) getNextMessage(span trace.Span, timeout time.Duration) (*MessageR
  }
 }
 
+// PushMessage pushes a new message to the message queue
 func (p *Pbft) PushMessage(msg *MessageReq) {
- p.pushMessage(msg)
-}
-
-// pushMessage pushes a new message to the message queue
-func (p *Pbft) pushMessage(msg *MessageReq) {
  p.msgQueue.pushMessage(msg)
 
  select {

consensus_test.go

@@ -460,7 +460,7 @@ func (m *mockPbft) emitMsg(msg *MessageReq) {
  // from := m.pool.get(string(msg.From)).Address()
  // msg.From = from
 
- m.Pbft.pushMessage(msg)
+ m.Pbft.PushMessage(msg)
 }
 
 func (m *mockPbft) addMessage(msg *MessageReq) {

go.mod

@@ -5,11 +5,11 @@ go 1.17
 require (
  github.com/stretchr/testify v1.7.0
  go.opentelemetry.io/otel v1.1.0
+ go.opentelemetry.io/otel/trace v1.1.0
 )
 
 require (
  github.com/davecgh/go-spew v1.1.0 // indirect
  github.com/pmezard/go-difflib v1.0.0 // indirect
- go.opentelemetry.io/otel/trace v1.1.0 // indirect
  gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )

state.go

@@ -50,8 +50,8 @@ type MessageReq struct {
  Proposal []byte
 }
 
-func (m *MessageReq) SetProposal(b []byte) {
- m.Proposal = append([]byte{}, b...)
+func (m *MessageReq) SetProposal(proposal []byte) {
+ m.Proposal = append([]byte{}, proposal...)
 }
 
 func (m *MessageReq) Copy() *MessageReq {
@@ -61,7 +61,10 @@ func (m *MessageReq) Copy() *MessageReq {
  mm.View = m.View.Copy()
  }
  if m.Proposal != nil {
- mm.Proposal = append([]byte{}, m.Proposal...)
+ mm.SetProposal(m.Proposal)
+ }
+ if m.Seal != nil {
+ mm.Seal = append([]byte{}, m.Seal...)
  }
  return mm
 }
@@ -228,12 +231,12 @@ func (c *currentState) getErr() error {
 func (c *currentState) maxRound() (maxRound uint64, found bool) {
  num := c.MaxFaultyNodes() + 1
 
- for k, round := range c.roundMessages {
- if len(round) < num {
+ for currentRound, messages := range c.roundMessages {
+ if len(messages) < num {
  continue
  }
- if maxRound < k {
- maxRound = k
+ if maxRound < currentRound {
+ maxRound = currentRound
  found = true
  }
  }
@@ -308,11 +311,12 @@ func (c *currentState) addMessage(msg *MessageReq) {
  c.prepared[addr] = msg
  } else if msg.Type == MessageReq_RoundChange {
  view := msg.View
- if _, ok := c.roundMessages[view.Round]; !ok {
- c.roundMessages[view.Round] = map[NodeID]*MessageReq{}
+ roundMessages, exists := c.roundMessages[view.Round]
+ if !exists {
+ roundMessages = map[NodeID]*MessageReq{}
+ c.roundMessages[view.Round] = roundMessages
  }
-
- c.roundMessages[view.Round][addr] = msg
+ roundMessages[addr] = msg
  }
 }