feat: add consensus flow skeleton (#4)

* Update proto location

* Add initial consensus flow

* Detail consensus flow

* Add unit tests for the propose timeout

Makefile

@@ -16,4 +16,4 @@ fixalign:
 protoc:
 	# Make sure the following prerequisites are installed before running these commands:
 	# https://grpc.io/docs/languages/go/quickstart/#prerequisites
-	protoc --go_out=./ ./messages/proto/*.proto
+	protoc --go_out=./ ./messages/types/proto/*.proto

core/broadcast.go

@@ -0,0 +1,108 @@
+package core
+
+import (
+	"github.com/gnolang/go-tendermint/messages/types"
+)
+
+// buildProposalMessage builds a proposal message using the given proposal
+func (t *Tendermint) buildProposalMessage(proposal []byte) *types.ProposalMessage {
+	// TODO make thread safe
+	var (
+		height     = t.state.view.Height
+		round      = t.state.view.Round
+		validRound = t.state.validRound
+	)
+
+	// Build the proposal message (assumes the node will sign it)
+	return &types.ProposalMessage{
+		View: &types.View{
+			Height: height,
+			Round:  round,
+		},
+		From:          t.node.ID(),
+		Proposal:      proposal,
+		ProposalRound: validRound,
+	}
+}
+
+// buildPrevoteMessage builds a prevote message using the given proposal identifier
+func (t *Tendermint) buildPrevoteMessage(id []byte) *types.PrevoteMessage {
+	// TODO make thread safe
+	var (
+		height = t.state.view.Height
+		round  = t.state.view.Round
+
+		processID = t.node.ID()
+	)
+
+	return &types.PrevoteMessage{
+		View: &types.View{
+			Height: height,
+			Round:  round,
+		},
+		From:       processID,
+		Identifier: id,
+	}
+}
+
+// buildPrecommitMessage builds a precommit message using the given precommit identifier
+func (t *Tendermint) buildPrecommitMessage(id []byte) *types.PrecommitMessage {
+	// TODO make thread safe
+	var (
+		height = t.state.view.Height
+		round  = t.state.view.Round
+
+		processID = t.node.ID()
+	)
+
+	return &types.PrecommitMessage{
+		View: &types.View{
+			Height: height,
+			Round:  round,
+		},
+		From:       processID,
+		Identifier: id,
+	}
+}
+
+// broadcastProposal signs and broadcasts the given proposal message
+func (t *Tendermint) broadcastProposal(proposal *types.ProposalMessage) {
+	message := &types.Message{
+		Type:      types.MessageType_PROPOSAL,
+		Signature: t.signer.Sign(proposal.Marshal()),
+		Payload: &types.Message_ProposalMessage{
+			ProposalMessage: proposal,
+		},
+	}
+
+	// Broadcast the proposal message
+	t.broadcast.Broadcast(message)
+}
+
+// broadcastPrevote signs and broadcasts the given prevote message
+func (t *Tendermint) broadcastPrevote(prevote *types.PrevoteMessage) {
+	message := &types.Message{
+		Type:      types.MessageType_PREVOTE,
+		Signature: t.signer.Sign(prevote.Marshal()),
+		Payload: &types.Message_PrevoteMessage{
+			PrevoteMessage: prevote,
+		},
+	}
+
+	// Broadcast the prevote message
+	t.broadcast.Broadcast(message)
+}
+
+// broadcastPrecommit signs and broadcasts the given precommit message
+func (t *Tendermint) broadcastPrecommit(precommit *types.PrecommitMessage) {
+	message := &types.Message{
+		Type:      types.MessageType_PRECOMMIT,
+		Signature: t.signer.Sign(precommit.Marshal()),
+		Payload: &types.Message_PrecommitMessage{
+			PrecommitMessage: precommit,
+		},
+	}
+
+	// Broadcast the precommit message
+	t.broadcast.Broadcast(message)
+}

core/mocks_test.go

@@ -0,0 +1,65 @@
+package core
+
+import "github.com/gnolang/go-tendermint/messages/types"
+
+type broadcastDelegate func(*types.Message)
+
+type mockBroadcast struct {
+	broadcastFn broadcastDelegate
+}
+
+func (m *mockBroadcast) Broadcast(message *types.Message) {
+	if m.broadcastFn != nil {
+		m.broadcastFn(message)
+	}
+}
+
+type (
+	idDelegate            func() []byte
+	hashDelegate          func([]byte) []byte
+	buildProposalDelegate func(uint64) []byte
+)
+
+type mockNode struct {
+	idFn            idDelegate
+	hashFn          hashDelegate
+	buildProposalFn buildProposalDelegate
+}
+
+func (m *mockNode) ID() []byte {
+	if m.idFn != nil {
+		return m.idFn()
+	}
+
+	return nil
+}
+
+func (m *mockNode) Hash(proposal []byte) []byte {
+	if m.hashFn != nil {
+		return m.hashFn(proposal)
+	}
+
+	return nil
+}
+
+func (m *mockNode) BuildProposal(height uint64) []byte {
+	if m.buildProposalFn != nil {
+		return m.buildProposalFn(height)
+	}
+
+	return nil
+}
+
+type signDelegate func([]byte) []byte
+
+type mockSigner struct {
+	signFn signDelegate
+}
+
+func (m *mockSigner) Sign(data []byte) []byte {
+	if m.signFn != nil {
+		return m.signFn(data)
+	}
+
+	return nil
+}

core/state.go

@@ -0,0 +1,55 @@
+package core
+
+import "github.com/gnolang/go-tendermint/messages/types"
+
+// step is the current state step
+type step uint8
+
+const (
+	propose step = iota
+	prevote
+	precommit
+)
+
+func (n step) String() string {
+	switch n {
+	case propose:
+		return "propose"
+	case prevote:
+		return "prevote"
+	case precommit:
+		return "precommit"
+	}
+
+	return ""
+}
+
+// state holds information about the current consensus state
+// TODO make thread safe
+type state struct {
+	view *types.View
+	step step
+
+	acceptedProposal   *types.ProposalMessage
+	acceptedProposalID []byte
+
+	lockedValue []byte
+	lockedRound int64
+
+	validValue []byte
+	validRound int64
+}
+
+// newState creates a fresh state using the given view
+func newState(view *types.View) *state {
+	return &state{
+		view:               view,
+		step:               propose,
+		acceptedProposal:   nil,
+		acceptedProposalID: nil,
+		lockedValue:        nil,
+		lockedRound:        -1,
+		validValue:         nil,
+		validRound:         -1,
+	}
+}

core/store.go

@@ -0,0 +1,76 @@
+package core
+
+import (
+	"github.com/gnolang/go-tendermint/messages"
+	"github.com/gnolang/go-tendermint/messages/types"
+)
+
+// store is the message store
+type store struct {
+	proposeMessages   *messages.Collector[types.ProposalMessage]
+	prevoteMessages   *messages.Collector[types.PrevoteMessage]
+	precommitMessages *messages.Collector[types.PrecommitMessage]
+}
+
+// newStore creates a new message store
+func newStore() *store {
+	return &store{
+		proposeMessages:   messages.NewCollector[types.ProposalMessage](),
+		prevoteMessages:   messages.NewCollector[types.PrevoteMessage](),
+		precommitMessages: messages.NewCollector[types.PrecommitMessage](),
+	}
+}
+
+// AddMessage adds a new message to the store
+func (s *store) AddMessage(message *types.Message) {
+	switch message.Type {
+	case types.MessageType_PROPOSAL:
+		// Parse the propose message
+		wrappedMessage, ok := message.Payload.(*types.Message_ProposalMessage)
+		if !ok {
+			return
+		}
+
+		// Get the proposal
+		proposal := wrappedMessage.ProposalMessage
+
+		s.proposeMessages.AddMessage(proposal.View, proposal.From, proposal)
+	case types.MessageType_PREVOTE:
+		// Parse the prevote message
+		wrappedMessage, ok := message.Payload.(*types.Message_PrevoteMessage)
+		if !ok {
+			return
+		}
+
+		// Get the prevote
+		prevote := wrappedMessage.PrevoteMessage
+
+		s.prevoteMessages.AddMessage(prevote.View, prevote.From, prevote)
+	case types.MessageType_PRECOMMIT:
+		// Parse the precommit message
+		wrappedMessage, ok := message.Payload.(*types.Message_PrecommitMessage)
+		if !ok {
+			return
+		}
+
+		// Get the precommit
+		precommit := wrappedMessage.PrecommitMessage
+
+		s.precommitMessages.AddMessage(precommit.View, precommit.From, precommit)
+	}
+}
+
+// SubscribeToPropose subscribes to incoming PROPOSE messages
+func (s *store) SubscribeToPropose() (<-chan func() []*types.ProposalMessage, func()) {
+	return s.proposeMessages.Subscribe()
+}
+
+// SubscribeToPrevote subscribes to incoming PREVOTE messages
+func (s *store) SubscribeToPrevote() (<-chan func() []*types.PrevoteMessage, func()) {
+	return s.prevoteMessages.Subscribe()
+}
+
+// SubscribeToPrecommit subscribes to incoming PRECOMMIT messages
+func (s *store) SubscribeToPrecommit() (<-chan func() []*types.PrecommitMessage, func()) {
+	return s.precommitMessages.Subscribe()
+}