Merge pull request #486 from celestiaorg/tzdybal/commit_gossiping

feat: support Commit gossiping on P2P level

Author: tzdybal

block/manager.go

@@ -59,6 +59,10 @@ type Manager struct {
 	HeaderOutCh chan *types.Header
 	HeaderInCh  chan *types.Header
 
+	CommitInCh  chan *types.Commit
+	CommitOutCh chan *types.Commit
+	lastCommit  *types.Commit
+
 	syncTarget uint64
 	blockInCh  chan newBlockEvent
 	syncCache  map[uint64]*types.Block
@@ -136,6 +140,8 @@ func NewManager(
 		// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
 		HeaderOutCh: make(chan *types.Header, 100),
 		HeaderInCh:  make(chan *types.Header, 100),
+		CommitInCh:  make(chan *types.Commit, 100),
+		CommitOutCh: make(chan *types.Commit, 100),
 		blockInCh:   make(chan newBlockEvent, 100),
 		retrieveMtx: new(sync.Mutex),
 		syncCache:   make(map[uint64]*types.Block),
@@ -199,6 +205,9 @@ func (m *Manager) SyncLoop(ctx context.Context) {
 				atomic.StoreUint64(&m.syncTarget, newHeight)
 				m.retrieveCond.Signal()
 			}
+		case commit := <-m.CommitInCh:
+			// TODO(tzdybal): check if it's from right aggregator
+			m.lastCommit = commit
 		case blockEvent := <-m.blockInCh:
 			block := blockEvent.block
 			daHeight := blockEvent.daHeight
@@ -433,6 +442,7 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 	m.store.SetHeight(block.Header.Height)
 
 	m.publishHeader(block)
+	m.publishCommit(lastCommit)
 
 	return nil
 }
@@ -471,10 +481,16 @@ func (m *Manager) exponentialBackoff(backoff time.Duration) time.Duration {
 	return backoff
 }
 
+// TODO(tzdybal): consider inlining
 func (m *Manager) publishHeader(block *types.Block) {
 	m.HeaderOutCh <- &block.Header
 }
 
+// TODO(tzdybal): consider inlining
+func (m *Manager) publishCommit(commit *types.Commit) {
+	m.CommitOutCh <- commit
+}
+
 func updateState(s *types.State, res *abci.ResponseInitChain) {
 	// If the app did not return an app hash, we keep the one set from the genesis doc in
 	// the state. We don't set appHash since we don't want the genesis doc app hash

node/node.go

@@ -154,6 +154,7 @@ func NewNode(ctx context.Context, conf config.NodeConfig, p2pKey crypto.PrivKey,
 
 	node.P2P.SetTxValidator(node.newTxValidator())
 	node.P2P.SetHeaderValidator(node.newHeaderValidator())
+	node.P2P.SetCommitValidator(node.newCommitValidator())
 
 	return node, nil
 }
@@ -324,6 +325,27 @@ func (n *Node) newHeaderValidator() p2p.GossipValidator {
 	}
 }
 
+// newCommitValidator returns a pubsub validator that runs basic checks and forwards
+// the deserialized commit for further processing
+func (n *Node) newCommitValidator() p2p.GossipValidator {
+	return func(commitMsg *p2p.GossipMessage) bool {
+		n.Logger.Debug("commit received", "from", commitMsg.From, "bytes", len(commitMsg.Data))
+		var commit types.Commit
+		err := commit.UnmarshalBinary(commitMsg.Data)
+		if err != nil {
+			n.Logger.Error("failed to deserialize commit", "error", err)
+			return false
+		}
+		err = commit.ValidateBasic()
+		if err != nil {
+			n.Logger.Error("failed to validate commit", "error", err)
+			return false
+		}
+		n.blockManager.CommitInCh <- &commit
+		return true
+	}
+}
+
 func createAndStartIndexerService(
 	conf config.NodeConfig,
 	kvStore store.KVStore,

p2p/client.go

@@ -38,6 +38,9 @@ const (
 
 	// headerTopicSuffix is added after namespace to create pubsub topic for block header gossiping.
 	headerTopicSuffix = "-header"
+
+	// commitTopicSuffix is added after namespace to create pubsub topic for block commit gossiping.
+	commitTopicSuffix = "-commit"
 )
 
 // Client is a P2P client, implemented with libp2p.
@@ -60,6 +63,9 @@ type Client struct {
 	headerGossiper  *Gossiper
 	headerValidator GossipValidator
 
+	commitGossiper  *Gossiper
+	commitValidator GossipValidator
+
 	// cancel is used to cancel context passed to libp2p functions
 	// it's required because of discovery.Advertise call
 	cancel context.CancelFunc
@@ -165,6 +171,17 @@ func (c *Client) SetHeaderValidator(validator GossipValidator) {
 	c.headerValidator = validator
 }
 
+// GossipCommit sends the block commit to the P2P network.
+func (c *Client) GossipCommit(ctx context.Context, commitBytes []byte) error {
+	c.logger.Debug("Gossiping block commit", "len", len(commitBytes))
+	return c.commitGossiper.Publish(ctx, commitBytes)
+}
+
+// SetCommitValidator sets the callback function, that will be invoked after block commit is received from P2P network.
+func (c *Client) SetCommitValidator(validator GossipValidator) {
+	c.commitValidator = validator
+}
+
 // Addrs returns listen addresses of Client.
 func (c *Client) Addrs() []multiaddr.Multiaddr {
 	return c.host.Addrs()
@@ -312,13 +329,22 @@ func (c *Client) setupGossiping(ctx context.Context) error {
 	}
 	go c.txGossiper.ProcessMessages(ctx)
 
-	c.headerGossiper, err = NewGossiper(c.host, ps, c.getHeaderTopic(), c.logger,
-		WithValidator(c.headerValidator))
+	c.headerGossiper, err = NewGossiper(c.host, ps, c.getHeaderTopic(), c.logger, WithValidator(c.headerValidator))
 	if err != nil {
 		return err
 	}
 	go c.headerGossiper.ProcessMessages(ctx)
 
+	var opts []GossiperOption
+	if c.commitValidator != nil {
+		opts = append(opts, WithValidator(c.commitValidator))
+	}
+	c.commitGossiper, err = NewGossiper(c.host, ps, c.getCommitTopic(), c.logger, opts...)
+	if err != nil {
+		return err
+	}
+	go c.commitGossiper.ProcessMessages(ctx)
+
 	return nil
 }
 
@@ -359,3 +385,7 @@ func (c *Client) getTxTopic() string {
 func (c *Client) getHeaderTopic() string {
 	return c.getNamespace() + headerTopicSuffix
 }
+
+func (c *Client) getCommitTopic() string {
+	return c.getNamespace() + commitTopicSuffix
+}