Repository: celestiaorg/celestia-node
Pull request #1948 (2023-03-21 12:58:33)

feat(core/listener): listen for EventDataSignedBlock
## Overview

Resolves https://github.com/celestiaorg/celestia-node/issues/1640




Patch:
diff --git a/core/exchange.go b/core/exchange.go
index 49e5074c..97eaf261 100644
--- a/core/exchange.go
+++ b/core/exchange.go
@@ -91,7 +91,7 @@ func (ce *Exchange) Get(ctx context.Context, hash libhead.Hash) (*header.Extende
 		return nil, err
 	}
 	// construct extended header
-	eh, err := ce.construct(ctx, block, comm, vals, eds)
+	eh, err := ce.construct(ctx, &block.Header, comm, vals, eds)
 	if err != nil {
 		return nil, err
 	}
@@ -113,12 +113,7 @@ func (ce *Exchange) Head(ctx context.Context) (*header.ExtendedHeader, error) {
 }
 
 func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height *int64) (*header.ExtendedHeader, error) {
-	b, err := ce.fetcher.GetBlock(ctx, height)
-	if err != nil {
-		return nil, err
-	}
-
-	comm, vals, err := ce.fetcher.GetBlockInfo(ctx, &b.Height)
+	b, err := ce.fetcher.GetSignedBlock(ctx, height)
 	if err != nil {
 		return nil, err
 	}
@@ -129,7 +124,7 @@ func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height *int64
 		return nil, err
 	}
 	// create extended header
-	eh, err := ce.construct(ctx, b, comm, vals, eds)
+	eh, err := ce.construct(ctx, &b.Header, &b.Commit, &b.ValidatorSet, eds)
 	if err != nil {
 		return nil, err
 	}
diff --git a/core/fetcher.go b/core/fetcher.go
index 758c2328..a23a6da2 100644
--- a/core/fetcher.go
+++ b/core/fetcher.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	logging "github.com/ipfs/go-log/v2"
+	coretypes "github.com/tendermint/tendermint/rpc/core/types"
 	"github.com/tendermint/tendermint/types"
 
 	libhead "github.com/celestiaorg/celestia-node/libs/header"
@@ -13,15 +14,15 @@ import (
 const newBlockSubscriber = "NewBlock/Events"
 
 var (
-	log                = logging.Logger("core")
-	newBlockEventQuery = types.QueryForEvent(types.EventNewBlock).String()
+	log                     = logging.Logger("core")
+	newDataSignedBlockQuery = types.QueryForEvent(types.EventSignedBlock).String()
 )
 
 type BlockFetcher struct {
 	client Client
 
-	newBlockCh chan *types.Block
-	doneCh     chan struct{}
+	signedBlockCh chan types.EventDataSignedBlock
+	doneCh        chan struct{}
 }
 
 // NewBlockFetcher returns a new `BlockFetcher`.
@@ -78,6 +79,11 @@ func (f *BlockFetcher) GetBlockByHash(ctx context.Context, hash libhead.Hash) (*
 	return res.Block, nil
 }
 
+// GetSignedBlock queries Core for a `Block` at the given height.
+func (f *BlockFetcher) GetSignedBlock(ctx context.Context, height *int64) (*coretypes.ResultSignedBlock, error) {
+	return f.client.SignedBlock(ctx, height)
+}
+
 // Commit queries Core for a `Commit` from the block at
 // the given height.
 func (f *BlockFetcher) Commit(ctx context.Context, height *int64) (*types.Commit, error) {
@@ -118,22 +124,22 @@ func (f *BlockFetcher) ValidatorSet(ctx context.Context, height *int64) (*types.
 
 // SubscribeNewBlockEvent subscribes to new block events from Core, returning
 // a new block event channel on success.
-func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (<-chan *types.Block, error) {
+func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (<-chan types.EventDataSignedBlock, error) {
 	// start the client if not started yet
 	if !f.client.IsRunning() {
 		return nil, fmt.Errorf("client not running")
 	}
-	eventChan, err := f.client.Subscribe(ctx, newBlockSubscriber, newBlockEventQuery)
+	eventChan, err := f.client.Subscribe(ctx, newBlockSubscriber, newDataSignedBlockQuery)
 	if err != nil {
 		return nil, err
 	}
 
 	// create a wrapper channel for translating ResultEvent to "raw" block
-	if f.newBlockCh != nil {
+	if f.signedBlockCh != nil {
 		return nil, fmt.Errorf("new block event channel exists")
 	}
 
-	f.newBlockCh = make(chan *types.Block)
+	f.signedBlockCh = make(chan types.EventDataSignedBlock)
 	f.doneCh = make(chan struct{})
 
 	go func() {
@@ -145,13 +151,13 @@ func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (<-chan *type
 				if !ok {
 					return
 				}
-				newBlock, ok := newEvent.Data.(types.EventDataNewBlock)
+				signedBlock, ok := newEvent.Data.(types.EventDataSignedBlock)
 				if !ok {
 					log.Warnf("unexpected event: %v", newEvent)
 					continue
 				}
 				select {
-				case f.newBlockCh <- newBlock.Block:
+				case f.signedBlockCh <- signedBlock:
 				case <-f.doneCh:
 					return
 				}
@@ -159,12 +165,12 @@ func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (<-chan *type
 		}
 	}()
 
-	return f.newBlockCh, nil
+	return f.signedBlockCh, nil
 }
 
 // UnsubscribeNewBlockEvent stops the subscription to new block events from Core.
 func (f *BlockFetcher) UnsubscribeNewBlockEvent(ctx context.Context) error {
-	if f.newBlockCh == nil {
+	if f.signedBlockCh == nil {
 		return fmt.Errorf("no new block event channel found")
 	}
 	if f.doneCh == nil {
@@ -174,13 +180,13 @@ func (f *BlockFetcher) UnsubscribeNewBlockEvent(ctx context.Context) error {
 		// send stop signal
 		f.doneCh <- struct{}{}
 		// close out fetcher channels
-		close(f.newBlockCh)
+		close(f.signedBlockCh)
 		close(f.doneCh)
-		f.newBlockCh = nil
+		f.signedBlockCh = nil
 		f.doneCh = nil
 	}()
 
-	return f.client.Unsubscribe(ctx, newBlockSubscriber, newBlockEventQuery)
+	return f.client.Unsubscribe(ctx, newBlockSubscriber, newDataSignedBlockQuery)
 }
 
 // IsSyncing returns the sync status of the Core connection: true for
diff --git a/core/listener.go b/core/listener.go
index b359dda9..78a33f25 100644
--- a/core/listener.go
+++ b/core/listener.go
@@ -75,7 +75,7 @@ func (cl *Listener) Stop(ctx context.Context) error {
 // listen kicks off a loop, listening for new block events from Core,
 // generating ExtendedHeaders and broadcasting them to the header-sub
 // gossipsub network.
-func (cl *Listener) listen(ctx context.Context, sub <-chan *types.Block) {
+func (cl *Listener) listen(ctx context.Context, sub <-chan types.EventDataSignedBlock) {
 	defer log.Info("listener: listening stopped")
 	for {
 		select {
@@ -83,7 +83,7 @@ func (cl *Listener) listen(ctx context.Context, sub <-chan *types.Block) {
 			if !ok {
 				return
 			}
-			log.Debugw("listener: new block from core", "height", b.Height)
+			log.Debugw("listener: new block from core", "height", b.Header.Height)
 
 			syncing, err := cl.fetcher.IsSyncing(ctx)
 			if err != nil {
@@ -91,12 +91,6 @@ func (cl *Listener) listen(ctx context.Context, sub <-chan *types.Block) {
 				return
 			}
 
-			comm, vals, err := cl.fetcher.GetBlockInfo(ctx, &b.Height)
-			if err != nil {
-				log.Errorw("listener: getting block info", "err", err)
-				return
-			}
-
 			// extend block data
 			eds, err := extendBlock(b.Data)
 			if err != nil {
@@ -104,13 +98,14 @@ func (cl *Listener) listen(ctx context.Context, sub <-chan *types.Block) {
 				return
 			}
 			// generate extended header
-			eh, err := cl.construct(ctx, b, comm, vals, eds)
+			eh, err := cl.construct(ctx, &b.Header, &b.Commit, &b.ValidatorSet, eds)
 			if err != nil {
 				log.Errorw("listener: making extended header", "err", err)
 				return
 			}
+
 			// attempt to store block data if not empty
-			err = storeEDS(ctx, eh.DAH.Hash(), eds, cl.store)
+			err = storeEDS(ctx, b.Header.DataHash.Bytes(), eds, cl.store)
 			if err != nil {
 				log.Errorw("listener: storing EDS", "err", err)
 				return
@@ -118,17 +113,19 @@ func (cl *Listener) listen(ctx context.Context, sub <-chan *types.Block) {
 
 			// notify network of new EDS hash only if core is already synced
 			if !syncing {
-				err = cl.hashBroadcaster(ctx, eh.DataHash.Bytes())
+				err = cl.hashBroadcaster(ctx, b.Header.DataHash.Bytes())
 				if err != nil {
-					log.Errorw("listener: broadcasting data hash", "height", eh.Height(),
-						"hash", eh.Hash(), "err", err)
+					log.Errorw("listener: broadcasting data hash",
+						"height", b.Header.Height,
+						"hash", b.Header.Hash(), "err", err) //TODO: hash or datahash?
 				}
 			}
 
 			// broadcast new ExtendedHeader, but if core is still syncing, notify only local subscribers
 			err = cl.headerBroadcaster.Broadcast(ctx, eh, pubsub.WithLocalPublication(syncing))
 			if err != nil {
-				log.Errorw("listener: broadcasting next header", "height", eh.Height(),
+				log.Errorw("listener: broadcasting next header",
+					"height", b.Header.Height,
 					"err", err)
 			}
 		case <-ctx.Done():
diff --git a/header/header.go b/header/header.go
index 28da7fc0..1b1a0fd2 100644
--- a/header/header.go
+++ b/header/header.go
@@ -19,7 +19,7 @@ import (
 // ConstructFn aliases a function that creates an ExtendedHeader.
 type ConstructFn = func(
 	context.Context,
-	*core.Block,
+	*core.Header,
 	*core.Commit,
 	*core.ValidatorSet,
 	*rsmt2d.ExtendedDataSquare,
@@ -70,7 +70,7 @@ var _ libhead.Header = &ExtendedHeader{}
 // MakeExtendedHeader assembles new ExtendedHeader.
 func MakeExtendedHeader(
 	ctx context.Context,
-	b *core.Block,
+	h *core.Header,
 	comm *core.Commit,
 	vals *core.ValidatorSet,
 	eds *rsmt2d.ExtendedDataSquare,
@@ -84,7 +84,7 @@ func MakeExtendedHeader(
 	}
 
 	eh := &ExtendedHeader{
-		RawHeader:    b.Header,
+		RawHeader:    *h,
 		DAH:          &dah,
 		Commit:       comm,
 		ValidatorSet: vals,
diff --git a/nodebuilder/fraud/mocks/api.go b/nodebuilder/fraud/mocks/api.go
index a57a688e..0561602f 100644
--- a/nodebuilder/fraud/mocks/api.go
+++ b/nodebuilder/fraud/mocks/api.go
@@ -10,8 +10,8 @@ import (
 
 	gomock "github.com/golang/mock/gomock"
 
-	fraud0 "github.com/celestiaorg/celestia-node/nodebuilder/fraud"
 	"github.com/celestiaorg/celestia-node/libs/fraud"
+	fraud0 "github.com/celestiaorg/celestia-node/nodebuilder/fraud"
 )
 
 // MockModule is a mock of Module interface.


Test patch:
diff --git a/core/client_test.go b/core/client_test.go
index e197dc58..7467b3c3 100644
--- a/core/client_test.go
+++ b/core/client_test.go
@@ -24,13 +24,13 @@ func TestRemoteClient_StartBlockSubscription_And_GetBlock(t *testing.T) {
 	t.Cleanup(cancel)
 
 	client := StartTestNode(t).Client
-	eventChan, err := client.Subscribe(ctx, newBlockSubscriber, newBlockEventQuery)
+	eventChan, err := client.Subscribe(ctx, newBlockSubscriber, newDataSignedBlockQuery)
 	require.NoError(t, err)
 
 	for i := 1; i <= 3; i++ {
 		select {
 		case evt := <-eventChan:
-			h := evt.Data.(types.EventDataNewBlock).Block.Height
+			h := evt.Data.(types.EventDataSignedBlock).Header.Height
 			block, err := client.Block(ctx, &h)
 			require.NoError(t, err)
 			require.GreaterOrEqual(t, block.Block.Height, int64(i))
@@ -39,5 +39,5 @@ func TestRemoteClient_StartBlockSubscription_And_GetBlock(t *testing.T) {
 		}
 	}
 	// unsubscribe to event channel
-	require.NoError(t, client.Unsubscribe(ctx, newBlockSubscriber, newBlockEventQuery))
+	require.NoError(t, client.Unsubscribe(ctx, newBlockSubscriber, newDataSignedBlockQuery))
 }
diff --git a/core/fetcher_test.go b/core/fetcher_test.go
index b71f7568..de7d1ba0 100644
--- a/core/fetcher_test.go
+++ b/core/fetcher_test.go
@@ -24,11 +24,14 @@ func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
 	for i := 1; i < 3; i++ {
 		select {
 		case newBlockFromChan := <-newBlockChan:
-			h := newBlockFromChan.Height
-			block, err := fetcher.GetBlock(ctx, &h)
+			h := newBlockFromChan.Header.Height
+			block, err := fetcher.GetSignedBlock(ctx, &h)
 			require.NoError(t, err)
-			assert.Equal(t, newBlockFromChan, block)
-			require.GreaterOrEqual(t, block.Height, int64(i))
+			assert.Equal(t, newBlockFromChan.Data, block.Data)
+			assert.Equal(t, newBlockFromChan.Header, block.Header)
+			assert.Equal(t, newBlockFromChan.Commit, block.Commit)
+			assert.Equal(t, newBlockFromChan.ValidatorSet, block.ValidatorSet)
+			require.GreaterOrEqual(t, newBlockFromChan.Header.Height, int64(i))
 		case <-ctx.Done():
 			require.NoError(t, ctx.Err())
 		}
@@ -63,18 +66,17 @@ func TestBlockFetcherHeaderValues(t *testing.T) {
 	valSet, err := fetcher.ValidatorSet(ctx, &h)
 	require.NoError(t, err)
 	// get next block
-	var nextBlock *types.Block
+	var nextBlock types.EventDataSignedBlock
 	select {
 	case nextBlock = <-newBlockChan:
 	case <-ctx.Done():
 		require.NoError(t, ctx.Err())
 	}
 	// compare LastCommit from next block to Commit from first block height
-	assert.Equal(t, nextBlock.LastCommit.Hash(), commit.Hash())
-	assert.Equal(t, nextBlock.LastCommit.Height, commit.Height)
-	assert.Equal(t, nextBlock.LastCommit.Signatures, commit.Signatures)
+	assert.Equal(t, nextBlock.Header.LastCommitHash, commit.Hash())
+	assert.Equal(t, nextBlock.Header.Height, commit.Height+1)
 	// compare ValidatorSet hash to the ValidatorsHash from first block height
 	hexBytes := valSet.Hash()
-	assert.Equal(t, nextBlock.ValidatorsHash.Bytes(), hexBytes)
+	assert.Equal(t, nextBlock.ValidatorSet.Hash(), hexBytes)
 	require.NoError(t, fetcher.UnsubscribeNewBlockEvent(ctx))
 }
diff --git a/core/header_test.go b/core/header_test.go
index 524d9979..57f53d36 100644
--- a/core/header_test.go
+++ b/core/header_test.go
@@ -33,7 +33,7 @@ func TestMakeExtendedHeaderForEmptyBlock(t *testing.T) {
 	eds, err := extendBlock(b.Data)
 	require.NoError(t, err)
 
-	headerExt, err := header.MakeExtendedHeader(ctx, b, comm, val, eds)
+	headerExt, err := header.MakeExtendedHeader(ctx, &b.Header, comm, val, eds)
 	require.NoError(t, err)
 
 	assert.Equal(t, header.EmptyDAH(), *headerExt.DAH)
diff --git a/header/headertest/testing.go b/header/headertest/testing.go
index 71ce0727..f446122e 100644
--- a/header/headertest/testing.go
+++ b/header/headertest/testing.go
@@ -297,14 +297,14 @@ func RandBlockID(t *testing.T) types.BlockID {
 func FraudMaker(t *testing.T, faultHeight int64, bServ blockservice.BlockService) header.ConstructFn {
 	log.Warn("Corrupting block...", "height", faultHeight)
 	return func(ctx context.Context,
-		b *types.Block,
+		h *types.Header,
 		comm *types.Commit,
 		vals *types.ValidatorSet,
 		eds *rsmt2d.ExtendedDataSquare,
 	) (*header.ExtendedHeader, error) {
-		if b.Height == faultHeight {
+		if h.Height == faultHeight {
 			eh := &header.ExtendedHeader{
-				RawHeader:    b.Header,
+				RawHeader:    *h,
 				Commit:       comm,
 				ValidatorSet: vals,
 			}
@@ -315,7 +315,7 @@ func FraudMaker(t *testing.T, faultHeight int64, bServ blockservice.BlockService
 			}
 			return eh, nil
 		}
-		return header.MakeExtendedHeader(ctx, b, comm, vals, eds)
+		return header.MakeExtendedHeader(ctx, h, comm, vals, eds)
 	}
 }