feat: Integrate Sync P2P protocol to consensus

Author: infrmtcs

consensus/consensus.go

@@ -12,11 +12,13 @@ import (
 	"github.com/NethermindEth/juno/consensus/proposal"
 	"github.com/NethermindEth/juno/consensus/proposer"
 	"github.com/NethermindEth/juno/consensus/starknet"
+	consensusSync "github.com/NethermindEth/juno/consensus/sync"
 	"github.com/NethermindEth/juno/consensus/tendermint"
 	"github.com/NethermindEth/juno/consensus/types"
 	"github.com/NethermindEth/juno/consensus/votecounter"
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/db"
+	"github.com/NethermindEth/juno/p2p/sync"
 	"github.com/NethermindEth/juno/utils"
 	"github.com/NethermindEth/juno/vm"
 	"github.com/libp2p/go-libp2p/core/host"
@@ -59,13 +61,22 @@ func Init(
 	p2p := p2p.New(host, logger, &builder, &proposalStore, currentHeight, &config.DefaultBufferSizes, bootstrapPeersFn)
 
 	commitListener := driver.NewCommitListener(logger, &proposalStore, proposer, p2p)
+
+	blockFetcher := sync.NewBlockFetcher(blockchain, host, logger)
+	messageExtractor := consensusSync.New[starknet.Value, starknet.Hash, starknet.Address](
+		toValue,
+		&proposalStore,
+	)
+
 	driver := driver.New(
 		logger,
 		tendermintDB,
 		stateMachine,
 		commitListener,
 		p2p.Broadcasters(),
 		p2p.Listeners(),
+		&blockFetcher,
+		&messageExtractor,
 		timeoutFn,
 	)
 

consensus/driver/commit_listener.go

@@ -15,8 +15,6 @@ type CommitHook[V types.Hashable[H], H types.Hash] interface {
 }
 
 // CommitListener is a component that is used to notify different components that a new committed block is available.
-//
-//go:generate mockgen -destination=../mocks/mock_commit_listener.go -package=mocks github.com/NethermindEth/juno/consensus/driver CommitListener
 type CommitListener[V types.Hashable[H], H types.Hash] interface {
 	CommitHook[V, H]
 	// Listen returns a channel that will receive committed blocks.

consensus/driver/driver.go

@@ -7,26 +7,31 @@ import (
 
 	"github.com/NethermindEth/juno/consensus/db"
 	"github.com/NethermindEth/juno/consensus/p2p"
+	consensusSync "github.com/NethermindEth/juno/consensus/sync"
 	"github.com/NethermindEth/juno/consensus/tendermint"
 	"github.com/NethermindEth/juno/consensus/types"
 	"github.com/NethermindEth/juno/consensus/types/actions"
+	"github.com/NethermindEth/juno/p2p/sync"
 	"github.com/NethermindEth/juno/utils"
 )
 
 type TimeoutFn func(step types.Step, round types.Round) time.Duration
 
 type Driver[V types.Hashable[H], H types.Hash, A types.Addr] struct {
-	log            utils.Logger
-	db             db.TendermintDB[V, H, A]
-	stateMachine   tendermint.StateMachine[V, H, A]
-	commitListener CommitListener[V, H]
-	broadcasters   p2p.Broadcasters[V, H, A]
-	listeners      p2p.Listeners[V, H, A]
+	log              utils.Logger
+	db               db.TendermintDB[V, H, A]
+	stateMachine     tendermint.StateMachine[V, H, A]
+	commitListener   CommitListener[V, H]
+	broadcasters     p2p.Broadcasters[V, H, A]
+	listeners        p2p.Listeners[V, H, A]
+	blockFetcher     *sync.BlockFetcher
+	messageExtractor *consensusSync.MessageExtractor[V, H, A]
 
 	getTimeout TimeoutFn
 
 	scheduledTms map[types.Timeout]*time.Timer
 	timeoutsCh   chan types.Timeout
+	syncListener chan sync.BlockBody
 }
 
 func New[V types.Hashable[H], H types.Hash, A types.Addr](
@@ -36,18 +41,23 @@ func New[V types.Hashable[H], H types.Hash, A types.Addr](
 	commitListener CommitListener[V, H],
 	broadcasters p2p.Broadcasters[V, H, A],
 	listeners p2p.Listeners[V, H, A],
+	blockFetcher *sync.BlockFetcher,
+	messageExtractor *consensusSync.MessageExtractor[V, H, A],
 	getTimeout TimeoutFn,
 ) Driver[V, H, A] {
 	return Driver[V, H, A]{
-		log:            log,
-		db:             db,
-		stateMachine:   stateMachine,
-		commitListener: commitListener,
-		broadcasters:   broadcasters,
-		listeners:      listeners,
-		getTimeout:     getTimeout,
-		scheduledTms:   make(map[types.Timeout]*time.Timer),
-		timeoutsCh:     make(chan types.Timeout),
+		log:              log,
+		db:               db,
+		stateMachine:     stateMachine,
+		commitListener:   commitListener,
+		broadcasters:     broadcasters,
+		listeners:        listeners,
+		blockFetcher:     blockFetcher,
+		messageExtractor: messageExtractor,
+		getTimeout:       getTimeout,
+		scheduledTms:     make(map[types.Timeout]*time.Timer),
+		timeoutsCh:       make(chan types.Timeout),
+		syncListener:     make(chan sync.BlockBody),
 	}
 }
 
@@ -123,6 +133,12 @@ func (d *Driver[V, H, A]) listen(ctx context.Context) error {
 					return nil
 				}
 				actions = d.stateMachine.ProcessPrecommit(p)
+			case p, ok := <-d.syncListener:
+				if !ok {
+					return nil
+				}
+				proposal, precommits := d.messageExtractor.Extract(&p)
+				actions = d.stateMachine.ProcessSync(&proposal, precommits)
 			}
 
 			isCommitted, err = d.execute(ctx, false, actions)
@@ -180,14 +196,28 @@ func (d *Driver[V, H, A]) execute(
 			return true, nil
 
 		case *actions.TriggerSync:
-			d.triggerSync(*action)
+			d.triggerSync(ctx, *action)
 		}
 	}
 	return false, nil
 }
 
-func (d *Driver[V, H, A]) triggerSync(sync actions.TriggerSync) {
-	// TODO: Implement this
+func (d *Driver[V, H, A]) triggerSync(ctx context.Context, sync actions.TriggerSync) {
+	for i := sync.Start; i <= sync.End; i++ {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+
+			if err := d.blockFetcher.ProcessBlock(ctx, uint64(i), d.syncListener); err != nil {
+				d.log.Errorw("failed to process block", "block", i, "err", err)
+			} else {
+				break
+			}
+		}
+	}
 }
 
 func (d *Driver[V, H, A]) setTimeout(ctx context.Context, timeout types.Timeout) {

consensus/driver/driver_test.go

@@ -178,6 +178,8 @@ func TestDriver(t *testing.T) {
 		newMockCommitListener(t, &commitAction),
 		broadcasters,
 		listeners,
+		nil, // TODO: Add tests for trigger sync
+		nil, // TODO: Add tests for trigger sync
 		mockTimeoutFn,
 	)
 

consensus/mock.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/NethermindEth/juno/consensus/driver"
 	"github.com/NethermindEth/juno/consensus/starknet"
+	consensusSync "github.com/NethermindEth/juno/consensus/sync"
 	"github.com/NethermindEth/juno/consensus/types"
 	"github.com/NethermindEth/juno/consensus/votecounter"
 	"github.com/NethermindEth/juno/core/felt"
@@ -47,7 +48,14 @@ func (n mockValidators) TotalVotingPower(height types.Height) types.VotingPower
 	return types.VotingPower(len(n))
 }
 
+// Currently we mock one voting power for all validators, to be removed once
+// we can query the voting power from the staking contracts.
+// The special case is for precommits from sync protocol, to be removed once
+// we can extract precommits from the sync protocol messages.
 func (n mockValidators) ValidatorVotingPower(height types.Height, addr *starknet.Address) types.VotingPower {
+	if addr != nil && *addr == consensusSync.SyncProtocolPrecommitSender {
+		return types.VotingPower(len(n))
+	}
 	return types.VotingPower(1)
 }