feat(consensus): implement propeller

buf.gen.yaml

@@ -7,4 +7,4 @@ plugins:
 inputs:
   - git_repo: https://github.com/starknet-io/starknet-p2p-specs.git
     branch: bcfa353a169c859e4d5d97757caccbe76f75bc06 # Latest commit as of 2025 May 6th
-    depth: 1
+    depth: 1

consensus/consensus.go

@@ -52,7 +52,9 @@ func Init(
 	}
 	currentHeight := types.Height(chainHeight + 1)
 
-	tendermintDB := consensusDB.NewTendermintDB[starknet.Value, starknet.Hash, starknet.Address](database)
+	tendermintDB := consensusDB.NewTendermintDB[
+		starknet.Value, starknet.Hash, starknet.Address,
+	](database)
 
 	executor := builder.NewExecutor(blockchain, vm, logger, false, false)
 	builder := builder.New(blockchain, executor)

consensus/p2p/buffered/proto_broadcaster.go

@@ -57,7 +57,8 @@ func (b ProtoBroadcaster[M]) Loop(ctx context.Context, topic *pubsub.Topic) {
 			}
 
 			for {
-				if err := topic.Publish(ctx, msgBytes); err != nil && !errors.Is(err, context.Canceled) {
+				err := topic.Publish(ctx, msgBytes)
+				if err != nil && !errors.Is(err, context.Canceled) {
 					b.logger.Error("unable to send message", zap.Error(err))
 					time.Sleep(b.retryInterval)
 					continue
@@ -70,7 +71,8 @@ func (b ProtoBroadcaster[M]) Loop(ctx context.Context, topic *pubsub.Topic) {
 			}
 		case <-rebroadcasted.trigger:
 			for msgBytes := range rebroadcasted.messages {
-				if err := topic.Publish(ctx, msgBytes); err != nil && !errors.Is(err, context.Canceled) {
+				err := topic.Publish(ctx, msgBytes)
+				if err != nil && !errors.Is(err, context.Canceled) {
 					b.logger.Error("unable to rebroadcast message", zap.Error(err))
 				}
 			}

consensus/p2p/validator/proposal_stream.go

@@ -48,7 +48,9 @@ func newSingleProposalStream(
 	}
 }
 
-func (s *proposalStream) start(ctx context.Context, firstMessage *consensus.StreamMessage) (types.Height, error) {
+func (s *proposalStream) start(
+	ctx context.Context, firstMessage *consensus.StreamMessage,
+) (types.Height, error) {
 	content := firstMessage.GetContent()
 	if content == nil {
 		return 0, fmt.Errorf("first message has empty content")

consensus/p2p/vote/vote_broadcasters.go

@@ -34,7 +34,9 @@ func NewVoteBroadcaster[H types.Hash, A types.Addr](
 	}
 }
 
-func (b *voteBroadcaster[H, A]) broadcast(ctx context.Context, message *types.Vote[H, A], voteType consensus.Vote_VoteType) {
+func (b *voteBroadcaster[H, A]) broadcast(
+	ctx context.Context, message *types.Vote[H, A], voteType consensus.Vote_VoteType,
+) {
 	msg, err := b.voteAdapter.FromVote(message, voteType)
 	if err != nil {
 		b.logger.Error("unable to convert vote", zap.Error(err))
@@ -60,6 +62,10 @@ func (b *prevoteBroadcaster[H, A]) Broadcast(ctx context.Context, message *types
 
 type precommitBroadcaster[H types.Hash, A types.Addr] voteBroadcaster[H, A]
 
-func (b *precommitBroadcaster[H, A]) Broadcast(ctx context.Context, message *types.Precommit[H, A]) {
-	(*voteBroadcaster[H, A])(b).broadcast(ctx, (*types.Vote[H, A])(message), consensus.Vote_Precommit)
+func (b *precommitBroadcaster[H, A]) Broadcast(
+	ctx context.Context, message *types.Precommit[H, A],
+) {
+	(*voteBroadcaster[H, A])(b).broadcast(
+		ctx, (*types.Vote[H, A])(message), consensus.Vote_Precommit,
+	)
 }