feat(p2p): block sync protocol

block/block.go

@@ -90,7 +90,6 @@ func (m *Manager) applyBlock(block *types.Block, commit *types.Commit, blockMeta
 	if err != nil {
 		return fmt.Errorf("update state: %w", err)
 	}
-
 	// Prune old heights, if requested by ABCI app.
 	if 0 < retainHeight {
 		err = m.pruneBlocks(uint64(retainHeight))
@@ -132,7 +131,7 @@ func (m *Manager) attemptApplyCachedBlocks() error {
 			return fmt.Errorf("block not valid at height %d, dropping it: err:%w", cachedBlock.Block.Header.Height, err)
 		}
 
-		err := m.applyBlock(cachedBlock.Block, cachedBlock.Commit, types.BlockMetaData{Source: types.GossipedBlock})
+		err := m.applyBlock(cachedBlock.Block, cachedBlock.Commit, types.BlockMetaData{Source: cachedBlock.Source})
 		if err != nil {
 			return fmt.Errorf("apply cached block: expected height: %d: %w", expectedHeight, err)
 		}

block/block_cache.go

@@ -9,8 +9,8 @@ type Cache struct {
 	cache map[uint64]types.CachedBlock
 }
 
-func (m *Cache) AddBlockToCache(h uint64, b *types.Block, c *types.Commit) {
-	m.cache[h] = types.CachedBlock{Block: b, Commit: c}
+func (m *Cache) AddBlockToCache(h uint64, b *types.Block, c *types.Commit, source types.BlockSource) {
+	m.cache[h] = types.CachedBlock{Block: b, Commit: c, Source: source}
 	types.BlockCacheSizeGauge.Set(float64(m.Size()))
 }
 

block/manager.go

@@ -8,7 +8,6 @@
 	"sync"
 	"sync/atomic"
 
-	"code.cloudfoundry.org/go-diodes"
 	"github.com/dymensionxyz/gerr-cosmos/gerrc"
 	"golang.org/x/sync/errgroup"
 
@@ -65,15 +64,15 @@
 	// Protect against processing two blocks at once when there are two routines handling incoming gossiped blocks,
 	// and incoming DA blocks, respectively.
 	retrieverMu sync.Mutex
-	Retriever   da.BatchRetriever
-	// get the next target height to sync local state to
-	targetSyncHeight diodes.Diode
-	// TargetHeight holds the value of the current highest block seen from either p2p (probably higher) or the DA
-	TargetHeight atomic.Uint64
-
+	// Protect against syncing twice from DA in case new batch is posted but it did not finish to sync yet.
+	syncFromDaMu sync.Mutex
+	Retriever    da.BatchRetriever
 	// Cached blocks and commits for applying at future heights. The blocks may not be valid, because
 	// we can only do full validation in sequential order.
 	blockCache *Cache
+
+	// TargetHeight holds the value of the current highest block seen from either p2p (probably higher) or the DA
+	TargetHeight atomic.Uint64
 }
 
 // NewManager creates new block Manager.
@@ -101,18 +100,17 @@
 	}
 
 	m := &Manager{
-		Pubsub:           pubsub,
-		p2pClient:        p2pClient,
-		LocalKey:         localKey,
-		Conf:             conf,
-		Genesis:          genesis,
-		Store:            store,
-		Executor:         exec,
-		DAClient:         dalc,
-		SLClient:         settlementClient,
-		Retriever:        dalc.(da.BatchRetriever),
-		targetSyncHeight: diodes.NewOneToOne(1, nil),
-		logger:           logger,
+		Pubsub:    pubsub,
+		p2pClient: p2pClient,
+		LocalKey:  localKey,
+		Conf:      conf,
+		Genesis:   genesis,
+		Store:     store,
+		Executor:  exec,
+		DAClient:  dalc,
+		SLClient:  settlementClient,
+		Retriever: dalc.(da.BatchRetriever),
+		logger:    logger,
 		blockCache: &Cache{
 			cache: make(map[uint64]types.CachedBlock),
 		},
@@ -147,16 +145,6 @@
 		}
 	}
 
-	if !isSequencer {
-		// Fullnode loop can start before syncing from DA
-		go uevent.MustSubscribe(ctx, m.Pubsub, "applyGossipedBlocksLoop", p2p.EventQueryNewNewGossipedBlock, m.onNewGossipedBlock, m.logger)
-	}
-
-	err = m.syncBlockManager()
-	if err != nil {
-		return fmt.Errorf("sync block manager: %w", err)
-	}
-
 	eg, ctx := errgroup.WithContext(ctx)
 
 	if isSequencer {
@@ -167,26 +155,33 @@
 		go func() {
 			bytesProducedC <- nBytes
 		}()
+		err = m.syncFromSettlement()
+		if err != nil {
+			return fmt.Errorf("sync block manager from settlement: %w", err)
+		}
 		eg.Go(func() error {
 			return m.SubmitLoop(ctx, bytesProducedC)
 		})
 		eg.Go(func() error {
 			return m.ProduceBlockLoop(ctx, bytesProducedC)
 		})
+
 	} else {
-		eg.Go(func() error {
-			return m.RetrieveLoop(ctx)
-		})
-		eg.Go(func() error {
-			return m.SyncToTargetHeightLoop(ctx)
-		})
-	}
+		// Full-nodes can sync from DA but it is not necessary to wait for it, since it can sync from P2P as well in parallel.
+		go func() {
+			err := m.syncFromSettlement()
+			if err != nil {
+				m.logger.Error("sync block manager from settlement", "err", err)
+			}
+			// DA Sync. Subscribe to SL next batch events
+			go uevent.MustSubscribe(ctx, m.Pubsub, "syncTargetLoop", settlement.EventQueryNewSettlementBatchAccepted, m.onNewStateUpdate, m.logger)
+		}()
 
-	go func() {
-		err := eg.Wait()
-		m.logger.Info("Block manager err group finished.", "err", err)
-	}()
+		// P2P Sync. Subscribe to P2P received blocks events
+		go uevent.MustSubscribe(ctx, m.Pubsub, "applyGossipedBlocksLoop", p2p.EventQueryNewGossipedBlock, m.onReceivedBlock, m.logger)
+		go uevent.MustSubscribe(ctx, m.Pubsub, "applyBlockSyncBlocksLoop", p2p.EventQueryNewBlockSyncBlock, m.onReceivedBlock, m.logger)
 
+	}
 	return nil
 }
 
@@ -208,21 +203,23 @@
 	return m.LastSubmittedHeight.Load() + 1
 }
 
-// syncBlockManager enforces the node to be synced on initial run.
-func (m *Manager) syncBlockManager() error {
+// syncFromSettlement enforces the node to be synced on initial run from SL and DA.
+func (m *Manager) syncFromSettlement() error {
 	res, err := m.SLClient.GetLatestBatch()
 	if errors.Is(err, gerrc.ErrNotFound) {
 		// The SL hasn't got any batches for this chain yet.
 		m.logger.Info("No batches for chain found in SL.")
 		m.LastSubmittedHeight.Store(uint64(m.Genesis.InitialHeight - 1))
 		return nil
 	}
+
 	if err != nil {
 		// TODO: separate between fresh rollapp and non-registered rollapp
 		return err
 	}
 	m.LastSubmittedHeight.Store(res.EndHeight)
 	err = m.syncToTargetHeight(res.EndHeight)
+	m.UpdateTargetHeight(res.EndHeight)
 	if err != nil {
 		return err
 	}

block/manager_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ipfs/go-datastore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -55,10 +56,11 @@ func TestInitialState(t *testing.T) {
 	// Init p2p client
 	privKey, _, _ := crypto.GenerateEd25519Key(rand.Reader)
 	p2pClient, err := p2p.NewClient(config.P2PConfig{
-		ListenAddress:      config.DefaultListenAddress,
-		GossipSubCacheSize: 50,
-		BootstrapRetryTime: 30 * time.Second,
-	}, privKey, "TestChain", pubsubServer, logger)
+		ListenAddress:                config.DefaultListenAddress,
+		GossipSubCacheSize:           50,
+		BootstrapRetryTime:           30 * time.Second,
+		BlockSyncRequestIntervalTime: 30 * time.Second,
+	}, privKey, "TestChain", emptyStore, pubsubServer, datastore.NewMapDatastore(), logger)
 	assert.NoError(err)
 	assert.NotNil(p2pClient)
 

block/gossip.go → block/p2p.go

@@ -4,19 +4,40 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/tendermint/tendermint/libs/pubsub"
-
 	"github.com/dymensionxyz/dymint/p2p"
 	"github.com/dymensionxyz/dymint/types"
+	"github.com/tendermint/tendermint/libs/pubsub"
 )
 
-// onNewGossipedBlock will take a block and apply it
-func (m *Manager) onNewGossipedBlock(event pubsub.Message) {
-	eventData, _ := event.Data().(p2p.GossipedBlock)
+// onReceivedBlock receives a block received event from P2P, saves the block to a cache and tries to apply the blocks from the cache.
+func (m *Manager) onReceivedBlock(event pubsub.Message) {
+	eventData, ok := event.Data().(p2p.P2PBlockEvent)
+	if !ok {
+		m.logger.Error("onReceivedBlock", "err", "wrong event data received")
+		return
+	}
+	var source types.BlockSource
+
+	if len(event.Events()[p2p.EventTypeKey]) != 1 {
+		m.logger.Error("onReceivedBlock", "err", "wrong number of event types received with the event", "received", len(event.Events()[p2p.EventTypeKey]))
+		return
+	}
+
+	switch event.Events()[p2p.EventTypeKey][0] {
+	case p2p.EventNewBlockSyncBlock:
+		source = types.BlockSync
+	case p2p.EventNewGossipedBlock:
+		source = types.Gossiped
+	default:
+		m.logger.Error("onReceivedBlock", "err", "wrong event type received", "type", event.Events()[p2p.EventTypeKey][0])
+		return
+	}
+
 	block := eventData.Block
 	commit := eventData.Commit
-	m.retrieverMu.Lock() // needed to protect blockCache access
 	height := block.Header.Height
+	m.retrieverMu.Lock() // needed to protect blockCache access
+
 	// It is not strictly necessary to return early, for correctness, but doing so helps us avoid mutex pressure and unnecessary repeated attempts to apply cached blocks
 	if m.blockCache.HasBlockInCache(height) {
 		m.retrieverMu.Unlock()
@@ -30,7 +51,7 @@ func (m *Manager) onNewGossipedBlock(event pubsub.Message) {
 
 	nextHeight := m.State.NextHeight()
 	if height >= nextHeight {
-		m.blockCache.AddBlockToCache(height, &block, &commit)
+		m.blockCache.AddBlockToCache(height, &block, &commit, source)
 	}
 	m.retrieverMu.Unlock() // have to give this up as it's locked again in attempt apply, and we're not re-entrant
 
@@ -40,8 +61,10 @@ func (m *Manager) onNewGossipedBlock(event pubsub.Message) {
 	}
 }
 
+// gossipBlock sends created blocks by the sequencer to full-nodes using P2P gossipSub
 func (m *Manager) gossipBlock(ctx context.Context, block types.Block, commit types.Commit) error {
-	gossipedBlock := p2p.GossipedBlock{Block: block, Commit: commit}
+	m.logger.Info("Gossipping block", "height", block.Header.Height)
+	gossipedBlock := p2p.P2PBlockEvent{Block: block, Commit: commit}
 	gossipedBlockBytes, err := gossipedBlock.MarshalBinary()
 	if err != nil {
 		return fmt.Errorf("marshal binary: %w: %w", err, ErrNonRecoverable)
@@ -51,5 +74,6 @@ func (m *Manager) gossipBlock(ctx context.Context, block types.Block, commit typ
 		// could cause that to fail, so we assume recoverable.
 		return fmt.Errorf("p2p gossip block: %w: %w", err, ErrRecoverable)
 	}
+
 	return nil
 }

block/produce.go

@@ -71,6 +71,7 @@ func (m *Manager) ProduceBlockLoop(ctx context.Context, bytesProducedC chan int)
 			}
 
 			bytesProducedN := block.SizeBytes() + commit.SizeBytes()
+			m.logger.Info("New block.", "size", uint64(block.ToProto().Size()))
 			select {
 			case <-ctx.Done():
 				return nil
@@ -181,7 +182,7 @@ func (m *Manager) produceBlock(allowEmpty bool) (*types.Block, *types.Commit, er
 		}
 	}
 
-	if err := m.applyBlock(block, commit, types.BlockMetaData{Source: types.ProducedBlock}); err != nil {
+	if err := m.applyBlock(block, commit, types.BlockMetaData{Source: types.Produced}); err != nil {
 		return nil, nil, fmt.Errorf("apply block: %w: %w", err, ErrNonRecoverable)
 	}
 

block/pruning.go

@@ -1,6 +1,7 @@
 package block
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/dymensionxyz/gerr-cosmos/gerrc"
@@ -14,6 +15,10 @@ func (m *Manager) pruneBlocks(retainHeight uint64) error {
 			gerrc.ErrInvalidArgument)
 	}
 
+	err := m.p2pClient.RemoveBlocks(context.TODO(), m.State.BaseHeight, retainHeight)
+	if err != nil {
+		m.logger.Error("pruning block-sync store", "retain_height", retainHeight, "err", err)
+	}
 	pruned, err := m.Store.PruneBlocks(m.State.BaseHeight, retainHeight)
 	if err != nil {
 		return fmt.Errorf("prune block store: %w", err)

block/retriever.go

@@ -1,41 +1,38 @@
 package block
 
 import (
-	"context"
 	"errors"
 	"fmt"
 
-	"code.cloudfoundry.org/go-diodes"
 	"github.com/dymensionxyz/gerr-cosmos/gerrc"
 
 	"github.com/dymensionxyz/dymint/da"
+	"github.com/dymensionxyz/dymint/settlement"
 	"github.com/dymensionxyz/dymint/types"
+	"github.com/tendermint/tendermint/libs/pubsub"
 )
 
-// RetrieveLoop listens for new target sync heights and then syncs the chain by
-// fetching batches from the settlement layer and then fetching the actual blocks
-// from the DA.
-func (m *Manager) RetrieveLoop(ctx context.Context) (err error) {
-	m.logger.Info("Started retrieve loop.")
-	p := diodes.NewPoller(m.targetSyncHeight, diodes.WithPollingContext(ctx))
-
-	for {
-		targetHeight := p.Next() // We only care about the latest one
-		if targetHeight == nil {
-			return
-		}
-
-		if err = m.syncToTargetHeight(*(*uint64)(targetHeight)); err != nil {
-			err = fmt.Errorf("sync until target: %w", err)
-			return
-		}
+// onNewStateUpdate will try to sync to new height, if not already synced
+func (m *Manager) onNewStateUpdate(event pubsub.Message) {
+	eventData, ok := event.Data().(*settlement.EventDataNewBatchAccepted)
+	if !ok {
+		m.logger.Error("onReceivedBatch", "err", "wrong event data received")
+		return
+	}
+	h := eventData.EndHeight
+	m.UpdateTargetHeight(h)
+	err := m.syncToTargetHeight(h)
+	if err != nil {
+		m.logger.Error("sync until target", "err", err)
 	}
 }
 
 // syncToTargetHeight syncs blocks until the target height is reached.
 // It fetches the batches from the settlement, gets the DA height and gets
 // the actual blocks from the DA.
 func (m *Manager) syncToTargetHeight(targetHeight uint64) error {
+	defer m.syncFromDaMu.Unlock()
+	m.syncFromDaMu.Lock()
 	for currH := m.State.NextHeight(); currH <= targetHeight; currH = m.State.NextHeight() {
 		// if we have the block locally, we don't need to fetch it from the DA
 		err := m.applyLocalBlock(currH)
@@ -100,7 +97,7 @@ func (m *Manager) applyLocalBlock(height uint64) error {
 	}
 
 	m.retrieverMu.Lock()
-	err = m.applyBlock(block, commit, types.BlockMetaData{Source: types.LocalDbBlock})
+	err = m.applyBlock(block, commit, types.BlockMetaData{Source: types.LocalDb})
 	if err != nil {
 		return fmt.Errorf("apply block from local store: height: %d: %w", height, err)
 	}
@@ -113,7 +110,6 @@ func (m *Manager) ProcessNextDABatch(daMetaData *da.DASubmitMetaData) error {
 	m.logger.Debug("trying to retrieve batch from DA", "daHeight", daMetaData.Height)
 	batchResp := m.fetchBatch(daMetaData)
 	if batchResp.Code != da.StatusSuccess {
-		m.logger.Error("fetching batch from DA", batchResp.Message)
 		return batchResp.Error
 	}
 
@@ -132,7 +128,8 @@ func (m *Manager) ProcessNextDABatch(daMetaData *da.DASubmitMetaData) error {
 				m.logger.Error("validate block from DA", "height", block.Header.Height, "err", err)
 				continue
 			}
-			err := m.applyBlock(block, batch.Commits[i], types.BlockMetaData{Source: types.DABlock, DAHeight: daMetaData.Height})
+
+			err := m.applyBlock(block, batch.Commits[i], types.BlockMetaData{Source: types.DA, DAHeight: daMetaData.Height})
 			if err != nil {
 				return fmt.Errorf("apply block: height: %d: %w", block.Header.Height, err)
 			}