fix(listener): fix memory leak in core.listener (#4733)

Author: walldiss

core/fetcher.go

@@ -145,8 +145,8 @@ func (f *BlockFetcher) ValidatorSet(ctx context.Context, height int64) (*types.V
 
 // SubscribeNewBlockEvent subscribes to new block events from Core, returning
 // a new block event channel.
-func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (chan types.EventDataSignedBlock, error) {
-	signedBlockCh := make(chan types.EventDataSignedBlock, 1)
+func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (chan SignedBlock, error) {
+	signedBlockCh := make(chan SignedBlock, 1)
 
 	go func() {
 		defer close(signedBlockCh)
@@ -178,7 +178,7 @@ func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (chan types.E
 
 func (f *BlockFetcher) receive(
 	ctx context.Context,
-	signedBlockCh chan types.EventDataSignedBlock,
+	signedBlockCh chan SignedBlock,
 	subscription coregrpc.BlockAPI_SubscribeNewHeightsClient,
 ) error {
 	log.Debug("fetcher: started listening for new blocks")
@@ -198,11 +198,11 @@ func (f *BlockFetcher) receive(
 		}
 
 		select {
-		case signedBlockCh <- types.EventDataSignedBlock{
-			Header:       *signedBlock.Header,
-			Commit:       *signedBlock.Commit,
-			ValidatorSet: *signedBlock.ValidatorSet,
-			Data:         *signedBlock.Data,
+		case signedBlockCh <- SignedBlock{
+			Header:       signedBlock.Header,
+			Commit:       signedBlock.Commit,
+			ValidatorSet: signedBlock.ValidatorSet,
+			Data:         signedBlock.Data,
 		}:
 		case <-ctx.Done():
 			return ctx.Err()

core/fetcher_no_race_test.go

@@ -7,7 +7,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cometbft/cometbft/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -43,7 +42,7 @@ func TestBlockFetcherHeaderValues(t *testing.T) {
 	valSet, err := fetcher.ValidatorSet(ctx, h)
 	require.NoError(t, err)
 	// get next block
-	var nextBlock types.EventDataSignedBlock
+	var nextBlock SignedBlock
 	select {
 	case nextBlock = <-newBlockChan:
 	case <-ctx.Done():

core/fetcher_test.go

@@ -32,10 +32,10 @@ func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
 			h := newBlockFromChan.Header.Height
 			block, err := fetcher.GetSignedBlock(ctx, h)
 			require.NoError(t, err)
-			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)
+			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())

core/listener.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/cometbft/cometbft/types"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -126,7 +125,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.EventDataSignedBlock) {
+func (cl *Listener) listen(ctx context.Context, sub <-chan SignedBlock) {
 	defer close(cl.closed)
 	defer log.Info("listener: listening stopped")
 	timeout := time.NewTimer(cl.listenerTimeout)
@@ -166,7 +165,7 @@ func (cl *Listener) listen(ctx context.Context, sub <-chan types.EventDataSigned
 	}
 }
 
-func (cl *Listener) handleNewSignedBlock(ctx context.Context, b types.EventDataSignedBlock) error {
+func (cl *Listener) handleNewSignedBlock(ctx context.Context, b SignedBlock) error {
 	var err error
 
 	ctx, span := tracer.Start(ctx, "listener/handleNewSignedBlock")
@@ -183,7 +182,7 @@ func (cl *Listener) handleNewSignedBlock(ctx context.Context, b types.EventDataS
 	}
 
 	// generate extended header
-	eh, err := cl.construct(&b.Header, &b.Commit, &b.ValidatorSet, eds)
+	eh, err := cl.construct(b.Header, b.Commit, b.ValidatorSet, eds)
 	if err != nil {
 		panic(fmt.Errorf("making extended header: %w", err))
 	}
@@ -212,7 +211,7 @@ func (cl *Listener) handleNewSignedBlock(ctx context.Context, b types.EventDataS
 		if err != nil && !errors.Is(err, context.Canceled) {
 			log.Errorw("listener: broadcasting data hash",
 				"height", b.Header.Height,
-				"hash", b.Header.Hash(), "err", err) // TODO: hash or datahash?
+				"datahash", eh.DAH.String(), "err", err)
 		}
 	}